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

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

@@ -113,7 +113,7 @@ func NewSyslogAgent(
 			m,
 			cfg.WarnOnInvalidDrains,
 			l,
-		)
+			syslog.NewLoggregatorEmitter(logClient, "syslog_agent"))
 		cupsFetcher = bindings.NewDrainParamParser(cupsFetcher, cfg.DefaultDrainMetadata)
 	}
 

src/pkg/ingress/bindings/filtered_binding_fetcher.go

@@ -1,6 +1,7 @@
 package bindings
 
 import (
+	"fmt"
 	"log"
 	"net"
 	"net/url"
@@ -34,9 +35,10 @@ type FilteredBindingFetcher struct {
 	invalidDrains     metrics.Gauge
 	blacklistedDrains metrics.Gauge
 	failedHostsCache  *simplecache.SimpleCache[string, bool]
+	emitter           syslog.LoggregatorEmitter
 }
 
-func NewFilteredBindingFetcher(c IPChecker, b binding.Fetcher, m metricsClient, warn bool, lc *log.Logger) *FilteredBindingFetcher {
+func NewFilteredBindingFetcher(c IPChecker, b binding.Fetcher, m metricsClient, warn bool, lc *log.Logger, emitter syslog.LoggregatorEmitter) *FilteredBindingFetcher {
 	opt := metrics.WithMetricLabels(map[string]string{"unit": "total"})
 
 	invalidDrains := m.NewGauge(
@@ -57,6 +59,7 @@ func NewFilteredBindingFetcher(c IPChecker, b binding.Fetcher, m metricsClient,
 		invalidDrains:     invalidDrains,
 		blacklistedDrains: blacklistedDrains,
 		failedHostsCache:  simplecache.New[string, bool](120 * time.Second),
+		emitter:           emitter,
 	}
 }
 
@@ -87,19 +90,22 @@ func (f *FilteredBindingFetcher) FetchBindings() ([]syslog.Binding, error) {
 
 		if invalidScheme(u.Scheme) {
 			f.printWarning("Invalid scheme %s in syslog drain url %s for application %s", u.Scheme, anonymousUrl.String(), b.AppId)
+			f.emitter.WriteLog(b.AppId, fmt.Sprintf("Invalid scheme %s in syslog drain url %s", u.Scheme, anonymousUrl.String()))
 			continue
 		}
 
 		if len(u.Host) == 0 {
 			invalidDrains += 1
 			f.printWarning("No hostname found in syslog drain url %s for application %s", anonymousUrl.String(), b.AppId)
+			f.emitter.WriteLog(b.AppId, fmt.Sprintf("No hostname found in syslog drain url %s", anonymousUrl.String()))
 			continue
 		}
 
 		_, exists := f.failedHostsCache.Get(u.Host)
 		if exists {
 			invalidDrains += 1
 			f.printWarning("Skipped resolve ip address for syslog drain with url %s for application %s due to prior failure", anonymousUrl.String(), b.AppId)
+			f.emitter.WriteLog(b.AppId, fmt.Sprintf("Skipped resolve ip address for syslog drain with url %s due to prior failure", anonymousUrl.String()))
 			continue
 		}
 
@@ -108,6 +114,7 @@ func (f *FilteredBindingFetcher) FetchBindings() ([]syslog.Binding, error) {
 			invalidDrains += 1
 			f.failedHostsCache.Set(u.Host, true)
 			f.printWarning("Cannot resolve ip address for syslog drain with url %s for application %s", anonymousUrl.String(), b.AppId)
+			f.emitter.WriteLog(b.AppId, fmt.Sprintf("Cannot resolve ip address for syslog drain with url %s", anonymousUrl.String()))
 			continue
 		}
 
@@ -116,6 +123,7 @@ func (f *FilteredBindingFetcher) FetchBindings() ([]syslog.Binding, error) {
 			invalidDrains += 1
 			blacklistedDrains += 1
 			f.printWarning("Resolved ip address for syslog drain with url %s for application %s is blacklisted", anonymousUrl.String(), b.AppId)
+			f.emitter.WriteLog(b.AppId, fmt.Sprintf("Resolved ip address for syslog drain with url %s is blacklisted", anonymousUrl.String()))
 			continue
 		}
 

src/pkg/ingress/bindings/filtered_binding_fetcher_test.go

@@ -33,7 +33,7 @@ var _ = Describe("FilteredBindingFetcher", func() {
 		}
 		bindingReader := &SpyBindingReader{bindings: input}
 
-		filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{}, bindingReader, metrics, true, log)
+		filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{}, bindingReader, metrics, true, log, syslog.LoggregatorEmitter{})
 		actual, err := filter.FetchBindings()
 
 		Expect(err).ToNot(HaveOccurred())
@@ -43,7 +43,7 @@ var _ = Describe("FilteredBindingFetcher", func() {
 	It("returns an error if the binding reader cannot fetch bindings", func() {
 		bindingReader := &SpyBindingReader{nil, errors.New("Woops")}
 
-		filter := bindings.NewFilteredBindingFetcher(&spyIPChecker{}, bindingReader, metrics, true, log)
+		filter := bindings.NewFilteredBindingFetcher(&spyIPChecker{}, bindingReader, metrics, true, log, syslog.LoggregatorEmitter{})
 		actual, err := filter.FetchBindings()
 
 		Expect(err).To(HaveOccurred())
@@ -64,13 +64,7 @@ var _ = Describe("FilteredBindingFetcher", func() {
 			input := []syslog.Binding{
 				{AppId: "app-id", Hostname: "we.dont.care", Drain: syslog.Drain{Url: "://"}},
 			}
-			filter = bindings.NewFilteredBindingFetcher(
-				&spyIPChecker{},
-				&SpyBindingReader{bindings: input},
-				metrics,
-				warn,
-				log,
-			)
+			filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{}, &SpyBindingReader{bindings: input}, metrics, warn, log, syslog.LoggregatorEmitter{})
 		})
 
 		It("removes the binding", func() {
@@ -108,13 +102,7 @@ var _ = Describe("FilteredBindingFetcher", func() {
 			input := []syslog.Binding{
 				{AppId: "app-id", Hostname: "we.dont.care", Drain: syslog.Drain{Url: "https:///path"}},
 			}
-			filter = bindings.NewFilteredBindingFetcher(
-				&spyIPChecker{},
-				&SpyBindingReader{bindings: input},
-				metrics,
-				warn,
-				log,
-			)
+			filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{}, &SpyBindingReader{bindings: input}, metrics, warn, log, syslog.LoggregatorEmitter{})
 		})
 
 		It("removes the binding", func() {
@@ -163,13 +151,7 @@ var _ = Describe("FilteredBindingFetcher", func() {
 		})
 
 		JustBeforeEach(func() {
-			filter = bindings.NewFilteredBindingFetcher(
-				&spyIPChecker{},
-				&SpyBindingReader{bindings: input},
-				metrics,
-				warn,
-				log,
-			)
+			filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{}, &SpyBindingReader{bindings: input}, metrics, warn, log, syslog.LoggregatorEmitter{})
 		})
 
 		It("ignores the bindings", func() {
@@ -210,13 +192,7 @@ var _ = Describe("FilteredBindingFetcher", func() {
 			input := []syslog.Binding{
 				{AppId: "app-id", Hostname: "we.dont.care", Drain: syslog.Drain{Url: "syslog://some.invalid.host"}},
 			}
-			filter = bindings.NewFilteredBindingFetcher(
-				mockic,
-				&SpyBindingReader{bindings: input},
-				metrics,
-				warn,
-				log,
-			)
+			filter = bindings.NewFilteredBindingFetcher(mockic, &SpyBindingReader{bindings: input}, metrics, warn, log, syslog.LoggregatorEmitter{})
 		})
 
 		It("removes bindings that failed to resolve", func() {
@@ -272,16 +248,10 @@ var _ = Describe("FilteredBindingFetcher", func() {
 				{AppId: "app-id", Hostname: "we.dont.care", Drain: syslog.Drain{Url: "syslog://some.invalid.host"}},
 			}
 
-			filter = bindings.NewFilteredBindingFetcher(
-				&spyIPChecker{
-					checkBlacklistError: errors.New("blacklist error"),
-					resolvedIP:          net.ParseIP("127.0.0.1"),
-				},
-				&SpyBindingReader{bindings: input},
-				metrics,
-				warn,
-				log,
-			)
+			filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{
+				checkBlacklistError: errors.New("blacklist error"),
+				resolvedIP:          net.ParseIP("127.0.0.1"),
+			}, &SpyBindingReader{bindings: input}, metrics, warn, log, syslog.LoggregatorEmitter{})
 		})
 
 		It("removes the binding", func() {