Syslog drain app error messages in app log stream

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,27 @@ 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 signed by the loggregator CA"
+  agent.key:
+    description: "TLS private key for syslog 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}]

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,14 @@
       "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")}",
+
+      "BLACKLISTED_SYSLOG_RANGES" => "#{blacklisted_ips}",
     }
   }
   bpm = {"processes" => [process] }

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/egress/applog"
 	"code.cloudfoundry.org/tlsconfig"
 
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding"
@@ -56,19 +57,9 @@ func NewSyslogAgent(
 	cfg Config,
 	m Metrics,
 	l *log.Logger,
+	appLogEmitterFactory applog.LogEmitterFactory,
 ) *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,
-	)
-
 	ingressTLSConfig, err := loggregator.NewIngressTLSConfig(
 		cfg.GRPC.CAFile,
 		cfg.GRPC.CertFile,
@@ -81,17 +72,24 @@ func NewSyslogAgent(
 	logClient, err := loggregator.NewIngressClient(
 		ingressTLSConfig,
 		loggregator.WithLogger(log.New(os.Stderr, "", log.LstdFlags)),
+		loggregator.WithAddr(fmt.Sprintf("localhost:%d", cfg.GRPC.Port)),
 	)
 	if err != nil {
 		l.Panicf("failed to create log client for syslog connector: %q", err)
 	}
 
+	writerFactory := syslog.NewWriterFactory(internalTlsConfig, externalTlsConfig, syslog.NetworkTimeoutConfig{
+		Keepalive:    10 * time.Second,
+		DialTimeout:  10 * time.Second,
+		WriteTimeout: 10 * time.Second,
+	}, m)
+
 	connector := syslog.NewSyslogConnector(
 		cfg.DrainSkipCertVerify,
 		timeoutwaitgroup.New(time.Minute),
 		writerFactory,
 		m,
-		syslog.WithLogClient(logClient, "syslog_agent"),
+		syslog.WithLogEmitter(appLogEmitterFactory.NewLogEmitter(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/egress/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.NewAppLogEmitterFactory()
+
+		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr, &factory)
 		go agent.Run()
 	})
 

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

@@ -13,6 +13,8 @@ import (
 	"strings"
 	"time"
 
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/applog"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
@@ -48,6 +50,8 @@ var _ = Describe("SyslogAgent", func() {
 		agentMetrics *metricsHelpers.SpyMetricsRegistry
 		agentLogr    *log.Logger
 		agent        *app.SyslogAgent
+
+		factory applog.LogEmitterFactory
 	)
 
 	BeforeEach(func() {
@@ -134,7 +138,9 @@ var _ = Describe("SyslogAgent", func() {
 			agentCfg.Cache.PollingInterval = 10 * time.Millisecond
 		}
 
-		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr)
+		factory := applog.NewAppLogEmitterFactory()
+
+		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 appLogEmitter", func() {
+		spyFactory := testhelper.SpyAppLogEmitterFactory{}
+		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
@@ -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/egress/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.NewAppLogEmitterFactory()
+
+	app.NewSyslogAgent(cfg, m, logger, &factory).Run()
 }

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

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/config"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/bindings"
 
 	envstruct "code.cloudfoundry.org/go-envstruct"
 )
@@ -31,14 +32,29 @@ type Config struct {
 	CachePort int `env:"CACHE_PORT, required, report"`
 
 	MetricsServer config.MetricsServer
+
+	ForwarderAgentAddress string `env:"FORWARDER_AGENT_ADDR"`
+	GRPC                  GRPC
+	Blacklist             bindings.BlacklistRanges `env:"BLACKLISTED_SYSLOG_RANGES, 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
 // environment. If loading the config fails for any reason this function will
 // panic.
 func LoadConfig() Config {
 	cfg := Config{
-		APIPollingInterval: 15 * time.Second,
+		APIPollingInterval:    15 * time.Second,
+		ForwarderAgentAddress: "localhost: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,12 +4,16 @@ 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/loggregator-agent-release/src/pkg/egress/applog"
 	"code.cloudfoundry.org/tlsconfig"
 
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding"
@@ -19,13 +23,20 @@ import (
 	"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
+	logger      *log.Logger
 	metrics     Metrics
 	mu          sync.Mutex
+	emitter     applog.LogEmitter
+	checker     IPChecker
 }
 
 type Metrics interface {
@@ -34,11 +45,33 @@ 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(config.ForwarderAgentAddress),
+	)
+	if err != nil {
+		logger.Panicf("failed to create logger client for syslog connector: %q", err)
+	}
+	factory := applog.NewAppLogEmitterFactory()
+	emitter := factory.NewLogEmitter(logClient, "syslog_binding_cache")
+
 	return &SyslogBindingCache{
 		config:  config,
-		log:     log,
+		logger:  logger,
 		metrics: metrics,
+		emitter: emitter,
+		checker: &config.Blacklist,
 	}
 }
 
@@ -50,11 +83,11 @@ func (sbc *SyslogBindingCache) Run() {
 			Handler:           http.DefaultServeMux,
 			ReadHeaderTimeout: 2 * time.Second,
 		}
-		go func() { sbc.log.Println("PPROF SERVER STOPPED " + sbc.pprofServer.ListenAndServe().Error()) }()
+		go func() { sbc.logger.Println("PPROF SERVER STOPPED " + sbc.pprofServer.ListenAndServe().Error()) }()
 	}
 	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.logger, sbc.emitter, &sbc.config.Blacklist)
 
 	go poller.Poll()
 
@@ -103,7 +136,7 @@ func (sbc *SyslogBindingCache) startServer(router chi.Router) {
 	sbc.mu.Unlock()
 	err := sbc.server.ListenAndServeTLS("", "")
 	if err != http.ErrServerClosed {
-		sbc.log.Panicf("error creating listener: %s", err)
+		sbc.logger.Panicf("error creating listener: %s", err)
 	}
 }
 
@@ -115,7 +148,7 @@ func (sbc *SyslogBindingCache) tlsConfig() *tls.Config {
 		tlsconfig.WithClientAuthenticationFromFile(sbc.config.CacheCAFile),
 	)
 	if err != nil {
-		sbc.log.Panicf("failed to load server TLS config: %s", err)
+		sbc.logger.Panicf("failed to load server TLS config: %s", err)
 	}
 
 	if len(sbc.config.CipherSuites) > 0 {

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

@@ -107,6 +107,7 @@ var _ = Describe("App", func() {
 		err = aggDrainFile.Close()
 		Expect(err).ToNot(HaveOccurred())
 		sbcCerts = testhelper.GenerateCerts("binding-cache-ca")
+		grpcPort := 30000 + GinkgoParallelProcess()
 		sbcCfg = app.Config{
 			APIURL:              capi.URL,
 			APIPollingInterval:  10 * time.Millisecond,
@@ -128,6 +129,12 @@ var _ = Describe("App", func() {
 				KeyFile:   sbcCerts.Key("metron"),
 				PprofPort: uint16(pprofPort),
 			},
+			GRPC: app.GRPC{
+				Port:     grpcPort,
+				CAFile:   sbcCerts.CA(),
+				CertFile: sbcCerts.Cert("metron"),
+				KeyFile:  sbcCerts.Key("metron"),
+			},
 		}
 		sbcMetrics = metricsHelpers.NewMetricsRegistry()
 		sbcLogr = log.New(GinkgoWriter, "", log.LstdFlags)