add error message in applog-stream for syslog-drain validation

jobs/loggr-syslog-agent/templates/bpm.yml.erb

@@ -37,6 +37,7 @@
       "PPROF_PORT" => "#{p("metrics.pprof_port")}",
       "USE_RFC3339" => "#{p("logging.format.timestamp") == "rfc3339"}",
       "WARN_ON_INVALID_DRAINS" => "#{p("warn_on_invalid_drains")}",
+      "LOGGREGATOR_AGENT_ADDR" => "localhost:#{p('port')}",
     }
   }
   if_p("drain_cipher_suites") do | ciphers |

jobs/loggr-syslog-binding-cache/spec

@@ -7,6 +7,9 @@ templates:
   api_client.crt.erb: config/certs/api_client.crt
   api_client.key.erb: config/certs/api_client.key
   loggregator_ca.crt.erb: config/certs/loggregator_ca.crt
+  loggregator_agent_ca.crt.erb: config/certs/loggregator_agent_ca.crt
+  loggregator_agent.crt.erb: config/certs/loggregator_agent.crt
+  loggregator_agent.key.erb: config/certs/loggregator_agent.key
   binding_cache.crt.erb: config/certs/binding_cache.crt
   binding_cache.key.erb: config/certs/binding_cache.key
   metrics_ca.crt.erb: config/certs/metrics_ca.crt
@@ -134,3 +137,32 @@ properties:
   logging.format.timestamp:
     description: "Format for timestamp in component logs. Valid values are 'deprecated' and 'rfc3339'."
     default: "deprecated"
+
+  loggregator.host:
+    description: "The gRPC host used to connect to loggregator agent"
+    default: "127.0.0.1"
+  loggregator.port:
+    description: "The gRPC port used to connect to loggregator agent"
+    default: 3458
+  loggregator.tls.ca_cert:
+    description: "The CA cert used to communicate with loggregator agent"
+  loggregator.tls.cert:
+    description: "The cert used to communicate with loggregator agent"
+  loggregator.tls.key:
+    description: "The key used to communicate with loggregator agent"
+  loggregator.tls.cipher_suites:
+    description: |
+         An ordered list of supported SSL cipher suites. Allowed cipher suites are
+         TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 and TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384.
+    default: "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256:TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
+
+  blacklisted_syslog_ranges:
+    description: |
+      A list of IP address ranges that are not allowed to be specified in
+      syslog drain binding URLs.
+    default: []
+    example: [{start: 10.10.10.1, end: 10.10.10.10}]
+
+  warn_on_invalid_drains:
+    description: "Whether to output log warnings on invalid drains"
+    default: true

jobs/loggr-syslog-binding-cache/templates/bpm.yml.erb

@@ -1,4 +1,9 @@
 <%
+  blacklisted_ranges = p("blacklisted_syslog_ranges")
+  blacklisted_ips = blacklisted_ranges.map do |range|
+    "#{range['start']}-#{range['end']}"
+  end.join(",")
+
   certs_dir =  "/var/vcap/jobs/loggr-syslog-binding-cache/config/certs"
   api_url = link("cloud_controller").address
   if_p("api.override_url") {
@@ -32,6 +37,16 @@
       "DEBUG_METRICS" => "#{p("metrics.debug")}",
       "PPROF_PORT" => "#{p("metrics.pprof_port")}",
       "USE_RFC3339" => "#{p("logging.format.timestamp") == "rfc3339"}",
+
+      "LOGGREGATOR_AGENT_HOST" => "#{p("loggregator.host")}",
+      "LOGGREGATOR_AGENT_PORT" => "#{p("loggregator.port")}",
+      "LOGGREGATOR_AGENT_CA_FILE_PATH" => "#{certs_dir}/loggregator_agent_ca.crt",
+      "LOGGREGATOR_AGENT_CERT_FILE_PATH" => "#{certs_dir}/loggregator_agent.crt",
+      "LOGGREGATOR_AGENT_KEY_FILE_PATH" => "#{certs_dir}/loggregator_agent.key",
+      "LOGGREGATOR_AGENT_CIPHER_SUITES" => "#{p("loggregator.tls.cipher_suites").split(":").join(",")}",
+
+      "BLACKLISTED_SYSLOG_RANGES" => "#{blacklisted_ips}",
+      "WARN_ON_INVALID_DRAINS" => "#{p("warn_on_invalid_drains")}",
     }
   }
   bpm = {"processes" => [process] }

jobs/loggr-syslog-binding-cache/templates/loggregator_agent.crt.erb

@@ -0,0 +1 @@
+<%= p("loggregator.tls.cert") %>

jobs/loggr-syslog-binding-cache/templates/loggregator_agent.key.erb

@@ -0,0 +1 @@
+<%= p("loggregator.tls.key") %>

jobs/loggr-syslog-binding-cache/templates/loggregator_agent_ca.crt.erb

@@ -0,0 +1 @@
+<%= p("loggregator.tls.ca_cert") %>

src/cmd/syslog-agent/app/config.go

@@ -5,10 +5,9 @@ import (
 	"strings"
 	"time"
 
-	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/config"
-	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/bindings"
-
 	"code.cloudfoundry.org/go-envstruct"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding/blacklist"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/config"
 )
 
 // GRPC stores the configuration for the router as a server using a PORT
@@ -22,25 +21,26 @@ type GRPC struct {
 }
 
 type Cache struct {
-	URL             string                   `env:"CACHE_URL,                 report"`
-	CAFile          string                   `env:"CACHE_CA_FILE_PATH,        report"`
-	CertFile        string                   `env:"CACHE_CERT_FILE_PATH,      report"`
-	KeyFile         string                   `env:"CACHE_KEY_FILE_PATH,       report"`
-	CommonName      string                   `env:"CACHE_COMMON_NAME,         report"`
-	PollingInterval time.Duration            `env:"CACHE_POLLING_INTERVAL,    report"`
-	Blacklist       bindings.BlacklistRanges `env:"BLACKLISTED_SYSLOG_RANGES, report"`
+	URL             string                    `env:"CACHE_URL,                 report"`
+	CAFile          string                    `env:"CACHE_CA_FILE_PATH,        report"`
+	CertFile        string                    `env:"CACHE_CERT_FILE_PATH,      report"`
+	KeyFile         string                    `env:"CACHE_KEY_FILE_PATH,       report"`
+	CommonName      string                    `env:"CACHE_COMMON_NAME,         report"`
+	PollingInterval time.Duration             `env:"CACHE_POLLING_INTERVAL,    report"`
+	Blacklist       blacklist.BlacklistRanges `env:"BLACKLISTED_SYSLOG_RANGES, report"`
 }
 
 // Config holds the configuration for the syslog agent
 type Config struct {
-	UseRFC3339           bool          `env:"USE_RFC3339"`
-	BindingsPerAppLimit  int           `env:"BINDING_PER_APP_LIMIT,  report"`
-	DrainSkipCertVerify  bool          `env:"DRAIN_SKIP_CERT_VERIFY, report"`
-	DrainCipherSuites    string        `env:"DRAIN_CIPHER_SUITES,    report"`
-	DrainTrustedCAFile   string        `env:"DRAIN_TRUSTED_CA_FILE,  report"`
-	DefaultDrainMetadata bool          `env:"DEFAULT_DRAIN_METADATA, report"`
-	IdleDrainTimeout     time.Duration `env:"IDLE_DRAIN_TIMEOUT, report"`
-	WarnOnInvalidDrains  bool          `env:"WARN_ON_INVALID_DRAINS,    report"`
+	UseRFC3339             bool          `env:"USE_RFC3339"`
+	BindingsPerAppLimit    int           `env:"BINDING_PER_APP_LIMIT,  report"`
+	DrainSkipCertVerify    bool          `env:"DRAIN_SKIP_CERT_VERIFY, report"`
+	DrainCipherSuites      string        `env:"DRAIN_CIPHER_SUITES,    report"`
+	DrainTrustedCAFile     string        `env:"DRAIN_TRUSTED_CA_FILE,  report"`
+	DefaultDrainMetadata   bool          `env:"DEFAULT_DRAIN_METADATA, report"`
+	IdleDrainTimeout       time.Duration `env:"IDLE_DRAIN_TIMEOUT,     report"`
+	WarnOnInvalidDrains    bool          `env:"WARN_ON_INVALID_DRAINS, report"`
+	LoggregatorIngressAddr string        `env:"LOGGREGATOR_AGENT_ADDR, report, required"`
 
 	GRPC          GRPC
 	Cache         Cache

src/cmd/syslog-agent/app/syslog_agent.go

@@ -58,17 +58,11 @@ func NewSyslogAgent(
 	l *log.Logger,
 ) *SyslogAgent {
 	internalTlsConfig, externalTlsConfig := drainTLSConfig(cfg)
-	writerFactory := syslog.NewWriterFactory(
-		internalTlsConfig,
-		externalTlsConfig,
-		syslog.NetworkTimeoutConfig{
-			Keepalive:    10 * time.Second,
-			DialTimeout:  10 * time.Second,
-			WriteTimeout: 10 * time.Second,
-		},
-		m,
-	)
-
+	writerFactory := syslog.NewWriterFactory(internalTlsConfig, externalTlsConfig, syslog.NetworkTimeoutConfig{
+		Keepalive:    10 * time.Second,
+		DialTimeout:  10 * time.Second,
+		WriteTimeout: 10 * time.Second,
+	}, m)
 	ingressTLSConfig, err := loggregator.NewIngressTLSConfig(
 		cfg.GRPC.CAFile,
 		cfg.GRPC.CertFile,
@@ -81,6 +75,7 @@ func NewSyslogAgent(
 	logClient, err := loggregator.NewIngressClient(
 		ingressTLSConfig,
 		loggregator.WithLogger(log.New(os.Stderr, "", log.LstdFlags)),
+		loggregator.WithAddr(cfg.LoggregatorIngressAddr),
 	)
 	if err != nil {
 		l.Panicf("failed to create log client for syslog connector: %q", err)
@@ -91,7 +86,7 @@ func NewSyslogAgent(
 		timeoutwaitgroup.New(time.Minute),
 		writerFactory,
 		m,
-		syslog.WithLogClient(logClient, "syslog_agent"),
+		syslog.WithLogClient(logClient),
 	)
 
 	var cacheClient *cache.CacheClient
@@ -109,7 +104,6 @@ func NewSyslogAgent(
 		cupsFetcher = bindings.NewFilteredBindingFetcher(
 			&cfg.Cache.Blacklist,
 			bindings.NewBindingFetcher(cfg.BindingsPerAppLimit, cacheClient, m, l),
-			m,
 			cfg.WarnOnInvalidDrains,
 			l,
 		)

src/cmd/syslog-agent/app/syslog_agent_test.go

@@ -13,6 +13,8 @@ import (
 	"strings"
 	"time"
 
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding/blacklist"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
@@ -24,7 +26,6 @@ import (
 	"code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper"
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding"
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/config"
-	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/bindings"
 	"code.cloudfoundry.org/tlsconfig"
 )
 
@@ -272,8 +273,8 @@ var _ = Describe("SyslogAgent", func() {
 		BeforeEach(func() {
 			url, err := url.Parse(appHTTPSDrain.server.URL)
 			Expect(err).NotTo(HaveOccurred())
-			agentCfg.Cache.Blacklist = bindings.BlacklistRanges{
-				Ranges: []bindings.BlacklistRange{
+			agentCfg.Cache.Blacklist = blacklist.BlacklistRanges{
+				Ranges: []blacklist.BlacklistRange{
 					{
 						Start: url.Hostname(),
 						End:   url.Hostname(),

src/cmd/syslog-binding-cache/app/config.go

@@ -4,14 +4,14 @@ import (
 	"log"
 	"time"
 
-	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/config"
-
 	envstruct "code.cloudfoundry.org/go-envstruct"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding/blacklist"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/config"
 )
 
 // Config holds the configuration for the syslog binding cache
 type Config struct {
-	UseRFC3339           bool          `env:"USE_RFC3339"`
+	UseRFC3339           bool          `env:"USE_RFC3339, report"`
 	APIURL               string        `env:"API_URL,              required, report"`
 	APICAFile            string        `env:"API_CA_FILE_PATH,     required, report"`
 	APICertFile          string        `env:"API_CERT_FILE_PATH,   required, report"`
@@ -31,14 +31,38 @@ type Config struct {
 	CachePort int `env:"CACHE_PORT, required, report"`
 
 	MetricsServer config.MetricsServer
+
+	GRPC      GRPC
+	Blacklist blacklist.BlacklistRanges `env:"BLACKLISTED_SYSLOG_RANGES, report"`
+
+	WarnOnInvalidDrains bool `env:"WARN_ON_INVALID_DRAINS,    report"`
+}
+
+// GRPC stores the configuration for the forwarder agent using a PORT
+// with mTLS certs and as a client.
+type GRPC struct {
+	Host         string   `env:"LOGGREGATOR_AGENT_HOST,                     report"`
+	Port         int      `env:"LOGGREGATOR_AGENT_PORT,                     report"`
+	CAFile       string   `env:"LOGGREGATOR_AGENT_CA_FILE_PATH,   required, report"`
+	CertFile     string   `env:"LOGGREGATOR_AGENT_CERT_FILE_PATH, required, report"`
+	KeyFile      string   `env:"LOGGREGATOR_AGENT_KEY_FILE_PATH,  required, report"`
+	CipherSuites []string `env:"LOGGREGATOR_AGENT_CIPHER_SUITES,            report"`
 }
 
 // LoadConfig will load the configuration for the syslog binding cache from the
 // environment. If loading the config fails for any reason this function will
 // panic.
 func LoadConfig() Config {
 	cfg := Config{
-		APIPollingInterval: 15 * time.Second,
+		UseRFC3339:           false,
+		APIPollingInterval:   15 * time.Second,
+		APIBatchSize:         1000,
+		APIDisableKeepAlives: true,
+		WarnOnInvalidDrains:  true,
+		GRPC: GRPC{
+			Host: "127.0.0.1",
+			Port: 3458,
+		},
 	}
 	if err := envstruct.Load(&cfg); err != nil {
 		log.Panicf("Failed to load config from environment: %s", err)

src/cmd/syslog-binding-cache/app/syslog_binding_cache.go

@@ -4,28 +4,39 @@ import (
 	"crypto/tls"
 	"fmt"
 	"log"
+	"net"
 	"net/http"
 	_ "net/http/pprof" //nolint:gosec
+	"os"
 	"sync"
 	"time"
 
+	"code.cloudfoundry.org/go-loggregator/v10"
 	metrics "code.cloudfoundry.org/go-metric-registry"
 	"code.cloudfoundry.org/tlsconfig"
 
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding"
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/cache"
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/api"
+	v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2"
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/plumbing"
 	"github.com/go-chi/chi/v5"
 )
 
+type IPChecker interface {
+	ResolveAddr(host string) (net.IP, error)
+	CheckBlacklist(ip net.IP) error
+}
+
 type SyslogBindingCache struct {
-	config      Config
-	pprofServer *http.Server
-	server      *http.Server
-	log         *log.Logger
-	metrics     Metrics
-	mu          sync.Mutex
+	config       Config
+	pprofServer  *http.Server
+	server       *http.Server
+	log          *log.Logger
+	metrics      Metrics
+	mu           sync.Mutex
+	appLogClient v2.LogClient
+	checker      IPChecker
 }
 
 type Metrics interface {
@@ -34,11 +45,31 @@ type Metrics interface {
 	RegisterDebugMetrics()
 }
 
-func NewSyslogBindingCache(config Config, metrics Metrics, log *log.Logger) *SyslogBindingCache {
+func NewSyslogBindingCache(config Config, metrics Metrics, logger *log.Logger) *SyslogBindingCache {
+	ingressTLSConfig, err := loggregator.NewIngressTLSConfig(
+		config.GRPC.CAFile,
+		config.GRPC.CertFile,
+		config.GRPC.KeyFile,
+	)
+	if err != nil {
+		logger.Panicf("failed to configure client TLS: %q", err)
+	}
+
+	logClient, err := loggregator.NewIngressClient(
+		ingressTLSConfig,
+		loggregator.WithLogger(log.New(os.Stderr, "", log.LstdFlags)),
+		loggregator.WithAddr(fmt.Sprintf("%s:%d", config.GRPC.Host, config.GRPC.Port)),
+	)
+	if err != nil {
+		logger.Panicf("failed to create logger client for syslog binding cache: %q", err)
+	}
+
 	return &SyslogBindingCache{
-		config:  config,
-		log:     log,
-		metrics: metrics,
+		config:       config,
+		log:          logger,
+		metrics:      metrics,
+		appLogClient: logClient,
+		checker:      &config.Blacklist,
 	}
 }
 
@@ -54,7 +85,16 @@ func (sbc *SyslogBindingCache) Run() {
 	}
 	store := binding.NewStore(sbc.metrics)
 	aggregateStore := binding.NewAggregateStore(sbc.config.AggregateDrainsFile)
-	poller := binding.NewPoller(sbc.apiClient(), sbc.config.APIPollingInterval, store, sbc.metrics, sbc.log)
+	poller := binding.NewPoller(
+		sbc.apiClient(),
+		sbc.config.APIPollingInterval,
+		store,
+		sbc.metrics,
+		sbc.log,
+		sbc.appLogClient,
+		&sbc.config.Blacklist,
+		sbc.config.WarnOnInvalidDrains,
+	)
 
 	go poller.Poll()