add error messages from syslog drains to the app log stream

add error messages from filtered binding fetcher to the app log stream

rename LoggregatorEmitter to AppLogEmitter

change format of NewFilteredBindingFetcher calls

rename references to AppLogEmitter

move spyLogClient to testhelper package

use grpc port for ingress client from config

add comment

rearrange imports

introduce AppLogEmitterFactory

remove AppLogEmitter interface and refactor to struct

rename factory

rename NewAppLogEmitterFactory method

Author: corporatemax

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

@@ -56,19 +56,8 @@ func NewSyslogAgent(
 	cfg Config,
 	m Metrics,
 	l *log.Logger,
+	appLogEmitterFactory syslog.AppLogEmitterFactory,
 ) *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 +70,25 @@ func NewSyslogAgent(
 	logClient, err := loggregator.NewIngressClient(
 		ingressTLSConfig,
 		loggregator.WithLogger(log.New(os.Stderr, "", log.LstdFlags)),
+		loggregator.WithAddr(fmt.Sprintf("127.0.0.1:%d", cfg.GRPC.Port)),
 	)
 	if err != nil {
 		l.Panicf("failed to create log client for syslog connector: %q", err)
 	}
 
+	internalTlsConfig, externalTlsConfig := drainTLSConfig(cfg)
+	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.WithAppLogEmitter(appLogEmitterFactory.NewAppLogEmitter(logClient, "syslog_agent")),
 	)
 
 	var cacheClient *cache.CacheClient
@@ -112,6 +109,7 @@ func NewSyslogAgent(
 			m,
 			cfg.WarnOnInvalidDrains,
 			l,
+			appLogEmitterFactory.NewAppLogEmitter(logClient, "syslog_agent"),
 		)
 		cupsFetcher = bindings.NewDrainParamParser(cupsFetcher, cfg.DefaultDrainMetadata)
 	}

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

@@ -1,6 +1,7 @@
 package app_test
 
 import (
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -154,7 +155,9 @@ var _ = Describe("SyslogAgent with mTLS", func() {
 			agentCfg.Cache.PollingInterval = 10 * time.Millisecond
 		}
 
-		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr)
+		factory := syslog.NewAppLogEmitterFactory()
+
+		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr, &factory)
 		go agent.Run()
 	})
 
@@ -216,6 +219,7 @@ var _ = Describe("SyslogAgent with mTLS", func() {
 		})
 
 		It("will not be able to connect with those drains", func() {
+			//todo check if pr check runs
 			ctx, cancel := context.WithCancel(context.Background())
 			emitLogs(ctx, appIDs, grpcPort, agentCerts)
 			defer cancel()

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

@@ -1,6 +1,7 @@
 package app_test
 
 import (
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog"
 	"context"
 	"crypto/tls"
 	"fmt"
@@ -48,6 +49,8 @@ var _ = Describe("SyslogAgent", func() {
 		agentMetrics *metricsHelpers.SpyMetricsRegistry
 		agentLogr    *log.Logger
 		agent        *app.SyslogAgent
+
+		factory syslog.AppLogEmitterFactory
 	)
 
 	BeforeEach(func() {
@@ -134,7 +137,9 @@ var _ = Describe("SyslogAgent", func() {
 			agentCfg.Cache.PollingInterval = 10 * time.Millisecond
 		}
 
-		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr)
+		factory := syslog.NewAppLogEmitterFactory()
+
+		agent = app.NewSyslogAgent(agentCfg, agentMetrics, agentLogr, &factory)
 		go agent.Run()
 	})
 
@@ -238,6 +243,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 +436,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

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog"
 	"log"
 	_ "net/http/pprof" //nolint:gosec
 	"os"
@@ -33,5 +34,7 @@ func main() {
 		),
 	)
 
-	app.NewSyslogAgent(cfg, m, logger).Run()
+	factory := syslog.NewAppLogEmitterFactory()
+
+	app.NewSyslogAgent(cfg, m, logger, &factory).Run()
 }

src/internal/testhelper/spy_app_log_emitter.go

@@ -0,0 +1,23 @@
+package testhelper
+
+import "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog"
+
+type SpyAppLogEmitterFactory struct {
+	logClient   syslog.LogClient
+	sourceIndex string
+}
+
+func (factory *SpyAppLogEmitterFactory) LogClient() syslog.LogClient {
+	return factory.logClient
+}
+
+func (factory *SpyAppLogEmitterFactory) SourceIndex() string {
+	return factory.sourceIndex
+}
+
+func (factory *SpyAppLogEmitterFactory) NewAppLogEmitter(logClient syslog.LogClient, sourceIndex string) syslog.AppLogEmitter {
+	factory.logClient = logClient
+	factory.sourceIndex = sourceIndex
+	emitterFactory := syslog.NewAppLogEmitterFactory()
+	return emitterFactory.NewAppLogEmitter(logClient, sourceIndex)
+}

src/internal/testhelper/spy_log_client.go

@@ -0,0 +1,82 @@
+package testhelper
+
+import (
+	"code.cloudfoundry.org/go-loggregator/v10"
+	v2 "code.cloudfoundry.org/go-loggregator/v10/rpc/loggregator_v2"
+	"sync"
+)
+
+type spyLogClient struct {
+	mu       sync.Mutex
+	_message []string
+	_appID   []string
+
+	// We use maps to ensure that we can query the keys
+	_sourceType     map[string]struct{}
+	_sourceInstance map[string]struct{}
+}
+
+func NewSpyLogClient() *spyLogClient {
+	return &spyLogClient{
+		_sourceType:     make(map[string]struct{}),
+		_sourceInstance: make(map[string]struct{}),
+	}
+}
+
+func (s *spyLogClient) EmitLog(message string, opts ...loggregator.EmitLogOption) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	env := &v2.Envelope{
+		Tags: make(map[string]string),
+	}
+
+	for _, o := range opts {
+		o(env)
+	}
+
+	s._message = append(s._message, message)
+	s._appID = append(s._appID, env.SourceId)
+	s._sourceType[env.GetTags()["source_type"]] = struct{}{}
+	s._sourceInstance[env.GetInstanceId()] = struct{}{}
+}
+
+func (s *spyLogClient) Message() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s._message
+}
+
+func (s *spyLogClient) AppID() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s._appID
+}
+
+func (s *spyLogClient) SourceType() map[string]struct{} {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Copy map so the orig does not escape the mutex and induce a race.
+	m := make(map[string]struct{})
+	for k := range s._sourceType {
+		m[k] = struct{}{}
+	}
+
+	return m
+}
+
+func (s *spyLogClient) SourceInstance() map[string]struct{} {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Copy map so the orig does not escape the mutex and induce a race.
+	m := make(map[string]struct{})
+	for k := range s._sourceInstance {
+		m[k] = struct{}{}
+	}
+
+	return m
+}

src/pkg/egress/syslog/app_log_emitter.go

@@ -0,0 +1,54 @@
+package syslog
+
+import (
+	"code.cloudfoundry.org/go-loggregator/v10"
+)
+
+// LogClient is used to emit logs.
+type LogClient interface {
+	EmitLog(message string, opts ...loggregator.EmitLogOption)
+}
+
+// AppLogEmitter abstracts the sending of a log to the application log stream.
+type AppLogEmitter struct {
+	logClient   LogClient
+	sourceIndex string
+}
+
+// EmitLog writes a message in the application log stream using a LogClient.
+func (appLogEmitter *AppLogEmitter) EmitLog(appID string, message string) {
+	if appLogEmitter.logClient == nil || appID == "" {
+		return
+	}
+
+	option := loggregator.WithAppInfo(appID, "LGR", "")
+	appLogEmitter.logClient.EmitLog(message, option)
+
+	option = loggregator.WithAppInfo(
+		appID,
+		"SYS",
+		appLogEmitter.sourceIndex,
+	)
+	appLogEmitter.logClient.EmitLog(message, option)
+}
+
+// AppLogEmitterFactory is used to create new instances of AppLogEmitter
+type AppLogEmitterFactory interface {
+	NewAppLogEmitter(logClient LogClient, sourceIndex string) AppLogEmitter
+}
+
+// DefaultAppLogEmitterFactory implementation of AppLogEmitterFactory to produce DefaultAppLogEmitter.
+type DefaultAppLogEmitterFactory struct {
+}
+
+// NewAppLogEmitter creates a new AppLogEmitter.
+func (factory *DefaultAppLogEmitterFactory) NewAppLogEmitter(logClient LogClient, sourceIndex string) AppLogEmitter {
+	return AppLogEmitter{
+		logClient:   logClient,
+		sourceIndex: sourceIndex,
+	}
+}
+
+func NewAppLogEmitterFactory() DefaultAppLogEmitterFactory {
+	return DefaultAppLogEmitterFactory{}
+}

src/pkg/egress/syslog/app_log_emitter_test.go

@@ -0,0 +1,72 @@
+package syslog_test
+
+import (
+	"code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Loggregator Emitter", func() {
+	Describe("DefaultAppLogEmitter", func() {
+		It("emits a log message", func() {
+			logClient := testhelper.NewSpyLogClient()
+			factory := syslog.NewAppLogEmitterFactory()
+			emitter := factory.NewAppLogEmitter(logClient, "0")
+
+			emitter.EmitLog("app-id", "some-message")
+
+			messages := logClient.Message()
+			appIDs := logClient.AppID()
+			sourceTypes := logClient.SourceType()
+			Expect(messages).To(HaveLen(2))
+			Expect(messages[0]).To(Equal("some-message"))
+			Expect(messages[1]).To(Equal("some-message"))
+			Expect(appIDs[0]).To(Equal("app-id"))
+			Expect(appIDs[1]).To(Equal("app-id"))
+			Expect(sourceTypes).To(HaveKey("LGR"))
+			Expect(sourceTypes).To(HaveKey("SYS"))
+		})
+
+		It("does not emit a log message if the appID is empty", func() {
+			logClient := testhelper.NewSpyLogClient()
+			factory := syslog.NewAppLogEmitterFactory()
+			emitter := factory.NewAppLogEmitter(logClient, "0")
+
+			emitter.EmitLog("", "some-message")
+
+			messages := logClient.Message()
+			appIDs := logClient.AppID()
+			sourceTypes := logClient.SourceType()
+			Expect(messages).To(HaveLen(0))
+			Expect(appIDs).To(HaveLen(0))
+			Expect(sourceTypes).ToNot(HaveKey("LGR"))
+			Expect(sourceTypes).ToNot(HaveKey("SYS"))
+		})
+	})
+
+	Describe("DefaultAppLogEmitterFactory", func() {
+		It("produces a AppLogEmitter", func() {
+			factory := syslog.NewAppLogEmitterFactory()
+			logClient := testhelper.NewSpyLogClient()
+			sourceIndex := "test-index"
+
+			emitter := factory.NewAppLogEmitter(logClient, sourceIndex)
+			emitter.EmitLog("app-id", "some-message")
+
+			messages := logClient.Message()
+			appIDs := logClient.AppID()
+			sourceTypes := logClient.SourceType()
+			sourceInstance := logClient.SourceInstance()
+			Expect(messages).To(HaveLen(2))
+			Expect(messages[0]).To(Equal("some-message"))
+			Expect(messages[1]).To(Equal("some-message"))
+			Expect(appIDs[0]).To(Equal("app-id"))
+			Expect(appIDs[1]).To(Equal("app-id"))
+			Expect(sourceTypes).To(HaveKey("LGR"))
+			Expect(sourceTypes).To(HaveKey("SYS"))
+			Expect(sourceInstance).To(HaveKey(""))
+			Expect(sourceInstance).To(HaveKey("test-index"))
+		})
+	})
+})