Syslog drain app error messages in app log stream

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

@@ -14,6 +14,9 @@ templates:
   metrics.key.erb: config/certs/metrics.key
   aggregate_drains.yml.erb: config/aggregate_drains.yml
   prom_scraper_config.yml.erb: config/prom_scraper_config.yml
+  agent.crt.erb: config/certs/agent.crt
+  agent.key.erb: config/certs/agent.key
+  agent_ca.crt.erb: config/certs/agent_ca.crt
 
 packages:
 - binding-cache
@@ -134,3 +137,31 @@ properties:
   logging.format.timestamp:
     description: "Format for timestamp in component logs. Valid values are 'deprecated' and 'rfc3339'."
     default: "deprecated"
+
+  agent.port:
+    description: "Port the agent is serving gRPC via mTLS"
+    default: 3458
+  agent.ca_cert:
+    description: |
+        TLS loggregator root CA certificate. It is required for key/cert
+         verification.
+  agent.cert:
+    description: "TLS certificate for Syslog Binding Cache signed by the loggregator CA"
+  agent.key:
+    description: "TLS private key for Syslog Binding Cache signed by the loggregator CA"
+  agent.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/agent.crt.erb

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

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

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

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

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

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"}",
+
+      "AGENT_CA_FILE_PATH" => "#{certs_dir}/agent_ca.crt",
+      "AGENT_CERT_FILE_PATH" => "#{certs_dir}/agent.crt",
+      "AGENT_KEY_FILE_PATH" => "#{certs_dir}/agent.key",
+      "AGENT_CIPHER_SUITES" => "#{p("agent.cipher_suites").split(":").join(",")}",
+      "AGENT_PORT" => "#{p("agent.port")}",
+      "AGENT_ADDR" => "localhost:#{p("agent.port")}",
+
+      "BLACKLISTED_SYSLOG_RANGES" => "#{blacklisted_ips}",
+      "WARN_ON_INVALID_DRAINS" => "#{p("warn_on_invalid_drains")}",
     }
   }
   bpm = {"processes" => [process] }

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

@@ -13,6 +13,7 @@ import (
 	gendiodes "code.cloudfoundry.org/go-diodes"
 	"code.cloudfoundry.org/go-loggregator/v10"
 	metrics "code.cloudfoundry.org/go-metric-registry"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/applog"
 	"code.cloudfoundry.org/tlsconfig"
 
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding"
@@ -56,19 +57,14 @@ func NewSyslogAgent(
 	cfg Config,
 	m Metrics,
 	l *log.Logger,
+	appLogStreamFactory applog.AppLogStreamFactory,
 ) *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 +77,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 +88,7 @@ func NewSyslogAgent(
 		timeoutwaitgroup.New(time.Minute),
 		writerFactory,
 		m,
-		syslog.WithLogClient(logClient, "syslog_agent"),
+		syslog.WithAppLogStream(appLogStreamFactory.NewAppLogStream(logClient, "syslog_agent")),
 	)
 
 	var cacheClient *cache.CacheClient

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

@@ -10,6 +10,8 @@ import (
 	"os"
 	"time"
 
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/applog"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
@@ -154,7 +156,9 @@ var _ = Describe("SyslogAgent with mTLS", func() {
 			agentCfg.Cache.PollingInterval = 10 * time.Millisecond
 		}
 
-		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr)
+		factory := applog.NewAppLogStreamFactory()
+
+		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr, &factory)
 		go agent.Run()
 	})
 

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

@@ -13,6 +13,9 @@ import (
 	"strings"
 	"time"
 
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding/blacklist"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/applog"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
@@ -24,7 +27,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"
 )
 
@@ -48,6 +50,8 @@ var _ = Describe("SyslogAgent", func() {
 		agentMetrics *metricsHelpers.SpyMetricsRegistry
 		agentLogr    *log.Logger
 		agent        *app.SyslogAgent
+
+		factory applog.AppLogStreamFactory
 	)
 
 	BeforeEach(func() {
@@ -134,7 +138,9 @@ var _ = Describe("SyslogAgent", func() {
 			agentCfg.Cache.PollingInterval = 10 * time.Millisecond
 		}
 
-		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr)
+		factory := applog.NewAppLogStreamFactory()
+
+		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr, &factory)
 		go agent.Run()
 	})
 
@@ -238,6 +244,14 @@ var _ = Describe("SyslogAgent", func() {
 		Eventually(agentMetrics.GetDebugMetricsEnabled).Should(BeFalse())
 	})
 
+	It("configures appLogStream", func() {
+		spyFactory := testhelper.SpyAppLogStreamFactory{}
+		app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr, &spyFactory)
+
+		Expect(spyFactory.SourceIndex()).Should(Equal("syslog_agent"))
+		Expect(spyFactory.LogClient()).ShouldNot(BeNil())
+	})
+
 	Context("when debug configuration is enabled", func() {
 		BeforeEach(func() {
 			agentCfg.MetricsServer.DebugMetrics = true
@@ -272,8 +286,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(),
@@ -423,7 +437,7 @@ var _ = Describe("SyslogAgent", func() {
 			cfgCopy.GRPC.KeyFile = "invalid"
 
 			msg := `failed to configure client TLS: "failed to load keypair: open invalid: no such file or directory"`
-			Expect(func() { app.NewSyslogAgent(cfgCopy, agentMetrics, agentLogr) }).To(PanicWith(msg))
+			Expect(func() { app.NewSyslogAgent(cfgCopy, agentMetrics, agentLogr, factory) }).To(PanicWith(msg))
 		})
 	})
 })

src/cmd/syslog-agent/main.go

@@ -5,6 +5,8 @@ import (
 	_ "net/http/pprof" //nolint:gosec
 	"os"
 
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/applog"
+
 	metrics "code.cloudfoundry.org/go-metric-registry"
 
 	"code.cloudfoundry.org/loggregator-agent-release/src/cmd/syslog-agent/app"
@@ -33,5 +35,7 @@ func main() {
 		),
 	)
 
-	app.NewSyslogAgent(cfg, m, logger).Run()
+	factory := applog.NewAppLogStreamFactory()
+
+	app.NewSyslogAgent(cfg, m, logger, &factory).Run()
 }

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

@@ -4,9 +4,9 @@ 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
@@ -31,6 +31,22 @@ type Config struct {
 	CachePort int `env:"CACHE_PORT, required, report"`
 
 	MetricsServer config.MetricsServer
+
+	AgentAddress string `env:"AGENT_ADDR, required, report"`
+	GRPC         GRPC
+	Blacklist    blacklist.BlacklistRanges `env:"BLACKLISTED_SYSLOG_RANGES, report"`
+
+	WarnOnInvalidDrains bool `env:"WARN_ON_INVALID_DRAINS,    report"`
+}
+
+// GRPC stores the configuration for the router as a server using a PORT
+// with mTLS certs and as a client.
+type GRPC struct {
+	Port         int      `env:"AGENT_PORT,                     report"`
+	CAFile       string   `env:"AGENT_CA_FILE_PATH,   required, report"`
+	CertFile     string   `env:"AGENT_CERT_FILE_PATH, required, report"`
+	KeyFile      string   `env:"AGENT_KEY_FILE_PATH,  required, report"`
+	CipherSuites []string `env:"AGENT_CIPHER_SUITES,            report"`
 }
 
 // LoadConfig will load the configuration for the syslog binding cache from the