diff --git a/jobs/loggr-syslog-agent/templates/bpm.yml.erb b/jobs/loggr-syslog-agent/templates/bpm.yml.erb index e56828fe6..4a1b5bb0f 100644 --- a/jobs/loggr-syslog-agent/templates/bpm.yml.erb +++ b/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 | diff --git a/jobs/loggr-syslog-binding-cache/spec b/jobs/loggr-syslog-binding-cache/spec index d09c781b8..cac1ab5b4 100644 --- a/jobs/loggr-syslog-binding-cache/spec +++ b/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 diff --git a/jobs/loggr-syslog-binding-cache/templates/bpm.yml.erb b/jobs/loggr-syslog-binding-cache/templates/bpm.yml.erb index a91c6fe2b..5be5f258a 100644 --- a/jobs/loggr-syslog-binding-cache/templates/bpm.yml.erb +++ b/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] } diff --git a/jobs/loggr-syslog-binding-cache/templates/loggregator_agent.crt.erb b/jobs/loggr-syslog-binding-cache/templates/loggregator_agent.crt.erb new file mode 100644 index 000000000..c9afb95a8 --- /dev/null +++ b/jobs/loggr-syslog-binding-cache/templates/loggregator_agent.crt.erb @@ -0,0 +1 @@ +<%= p("loggregator.tls.cert") %> diff --git a/jobs/loggr-syslog-binding-cache/templates/loggregator_agent.key.erb b/jobs/loggr-syslog-binding-cache/templates/loggregator_agent.key.erb new file mode 100644 index 000000000..980c311bb --- /dev/null +++ b/jobs/loggr-syslog-binding-cache/templates/loggregator_agent.key.erb @@ -0,0 +1 @@ +<%= p("loggregator.tls.key") %> diff --git a/jobs/loggr-syslog-binding-cache/templates/loggregator_agent_ca.crt.erb b/jobs/loggr-syslog-binding-cache/templates/loggregator_agent_ca.crt.erb new file mode 100644 index 000000000..edd8ae982 --- /dev/null +++ b/jobs/loggr-syslog-binding-cache/templates/loggregator_agent_ca.crt.erb @@ -0,0 +1 @@ +<%= p("loggregator.tls.ca_cert") %> diff --git a/src/cmd/syslog-agent/app/config.go b/src/cmd/syslog-agent/app/config.go index 47231d3ea..a452f5fd9 100644 --- a/src/cmd/syslog-agent/app/config.go +++ b/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 diff --git a/src/cmd/syslog-agent/app/syslog_agent.go b/src/cmd/syslog-agent/app/syslog_agent.go index 277bbdeb1..d200ea359 100644 --- a/src/cmd/syslog-agent/app/syslog_agent.go +++ b/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, ) diff --git a/src/cmd/syslog-agent/app/syslog_agent_test.go b/src/cmd/syslog-agent/app/syslog_agent_test.go index 46a6d2c24..54278c428 100644 --- a/src/cmd/syslog-agent/app/syslog_agent_test.go +++ b/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(), diff --git a/src/cmd/syslog-agent/main b/src/cmd/syslog-agent/main new file mode 100755 index 000000000..492bcc2af Binary files /dev/null and b/src/cmd/syslog-agent/main differ diff --git a/src/cmd/syslog-binding-cache/app/config.go b/src/cmd/syslog-binding-cache/app/config.go index 0a7f00d76..e5862a83f 100644 --- a/src/cmd/syslog-binding-cache/app/config.go +++ b/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,6 +31,22 @@ 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 @@ -38,7 +54,15 @@ type Config struct { // 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) diff --git a/src/cmd/syslog-binding-cache/app/syslog_binding_cache.go b/src/cmd/syslog-binding-cache/app/syslog_binding_cache.go index ad2dba59d..c27f167cd 100644 --- a/src/cmd/syslog-binding-cache/app/syslog_binding_cache.go +++ b/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() diff --git a/src/cmd/syslog-binding-cache/app/syslog_binding_cache_test.go b/src/cmd/syslog-binding-cache/app/syslog_binding_cache_test.go index d02e7d904..c0f007618 100644 --- a/src/cmd/syslog-binding-cache/app/syslog_binding_cache_test.go +++ b/src/cmd/syslog-binding-cache/app/syslog_binding_cache_test.go @@ -47,34 +47,34 @@ var _ = Describe("App", func() { BeforeEach(func() { r := results{ { - Url: "syslog://drain-a", + Url: "syslog://localhost:1000", Credentials: []binding.Credentials{ { - Cert: "cert", Key: "key", Apps: []binding.App{{Hostname: "org.space.app-name-1", AppID: "app-id-1"}}, + Apps: []binding.App{{Hostname: "org.space.app-name-1", AppID: "app-id-1"}}, }, }, }, { - Url: "syslog://drain-b", + Url: "syslog://localhost:1001", Credentials: []binding.Credentials{ { - Cert: "cert", Key: "key", Apps: []binding.App{{Hostname: "org.space.app-name-1", AppID: "app-id-1"}}, + Apps: []binding.App{{Hostname: "org.space.app-name-1", AppID: "app-id-1"}}, }, }, }, { - Url: "syslog://drain-c", + Url: "syslog://localhost:1002", Credentials: []binding.Credentials{ { - Cert: "cert", Key: "key", Apps: []binding.App{{Hostname: "org.space.app-name-2", AppID: "app-id-2"}}, + Apps: []binding.App{{Hostname: "org.space.app-name-2", AppID: "app-id-2"}}, }, }, }, { - Url: "syslog://drain-d", + Url: "syslog://localhost:1003", Credentials: []binding.Credentials{ { - Cert: "cert", Key: "key", Apps: []binding.App{{Hostname: "org.space.app-name-2", AppID: "app-id-2"}}, + Apps: []binding.App{{Hostname: "org.space.app-name-2", AppID: "app-id-2"}}, }, }, }, @@ -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) @@ -189,13 +196,13 @@ var _ = Describe("App", func() { Expect(results).To(HaveLen(4)) b := findBindings(results, "app-id-1") - Expect(b[0].Url).To(Equal("syslog://drain-a")) - Expect(b[1].Url).To(Equal("syslog://drain-b")) + Expect(b[0].Url).To(Equal("syslog://localhost:1000")) + Expect(b[1].Url).To(Equal("syslog://localhost:1001")) Expect(b[0].Credentials[0].Apps[0].Hostname).To(Equal("org.space.app-name-1")) b = findBindings(results, "app-id-2") - Expect(b[0].Url).To(Equal("syslog://drain-c")) - Expect(b[1].Url).To(Equal("syslog://drain-d")) + Expect(b[0].Url).To(Equal("syslog://localhost:1002")) + Expect(b[1].Url).To(Equal("syslog://localhost:1003")) Expect(b[0].Credentials[0].Apps[0].Hostname).To(Equal("org.space.app-name-2")) }) diff --git a/src/internal/testhelper/spy_log_client.go b/src/internal/testhelper/spy_log_client.go new file mode 100644 index 000000000..cec80c9ca --- /dev/null +++ b/src/internal/testhelper/spy_log_client.go @@ -0,0 +1,67 @@ +package testhelper + +import ( + "sync" + + "code.cloudfoundry.org/go-loggregator/v10" + v2 "code.cloudfoundry.org/go-loggregator/v10/rpc/loggregator_v2" +) + +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{} +} + +func NewSpyLogClient() *SpyLogClient { + return &SpyLogClient{ + _sourceType: 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{}{} +} + +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 original does not escape the mutex and induce a race. + m := make(map[string]struct{}) + for k := range s._sourceType { + m[k] = struct{}{} + } + + return m +} diff --git a/src/pkg/ingress/bindings/bindingsfakes/fake_ipchecker.go b/src/pkg/binding/bindingfakes/fake_ipchecker.go similarity index 97% rename from src/pkg/ingress/bindings/bindingsfakes/fake_ipchecker.go rename to src/pkg/binding/bindingfakes/fake_ipchecker.go index 251bb4f7d..4e6e77bf0 100644 --- a/src/pkg/ingress/bindings/bindingsfakes/fake_ipchecker.go +++ b/src/pkg/binding/bindingfakes/fake_ipchecker.go @@ -1,11 +1,11 @@ // Code generated by counterfeiter. DO NOT EDIT. -package bindingsfakes +package bindingfakes import ( "net" "sync" - "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/bindings" + "code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding" ) type FakeIPChecker struct { @@ -188,4 +188,4 @@ func (fake *FakeIPChecker) recordInvocation(key string, args []interface{}) { fake.invocations[key] = append(fake.invocations[key], args) } -var _ bindings.IPChecker = new(FakeIPChecker) +var _ binding.IPChecker = new(FakeIPChecker) diff --git a/src/pkg/ingress/bindings/blacklist_ranges.go b/src/pkg/binding/blacklist/blacklist_ranges.go similarity index 99% rename from src/pkg/ingress/bindings/blacklist_ranges.go rename to src/pkg/binding/blacklist/blacklist_ranges.go index fe208264f..ec77c124d 100644 --- a/src/pkg/ingress/bindings/blacklist_ranges.go +++ b/src/pkg/binding/blacklist/blacklist_ranges.go @@ -1,4 +1,4 @@ -package bindings +package blacklist import ( "bytes" diff --git a/src/pkg/ingress/bindings/blacklist_ranges_test.go b/src/pkg/binding/blacklist/blacklist_ranges_test.go similarity index 65% rename from src/pkg/ingress/bindings/blacklist_ranges_test.go rename to src/pkg/binding/blacklist/blacklist_ranges_test.go index f1be90b56..693662265 100644 --- a/src/pkg/ingress/bindings/blacklist_ranges_test.go +++ b/src/pkg/binding/blacklist/blacklist_ranges_test.go @@ -1,9 +1,9 @@ -package bindings_test +package blacklist_test import ( "net" - "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/bindings" + "code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding/blacklist" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) @@ -11,44 +11,44 @@ import ( var _ = Describe("BlacklistRanges", func() { Describe("validates", func() { It("accepts valid IP address range", func() { - _, err := bindings.NewBlacklistRanges( - bindings.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4"}, + _, err := blacklist.NewBlacklistRanges( + blacklist.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4"}, ) Expect(err).ToNot(HaveOccurred()) }) It("returns an error with an invalid start address", func() { - _, err := bindings.NewBlacklistRanges( - bindings.BlacklistRange{Start: "127.0.2.2.1", End: "127.0.2.4"}, + _, err := blacklist.NewBlacklistRanges( + blacklist.BlacklistRange{Start: "127.0.2.2.1", End: "127.0.2.4"}, ) Expect(err).To(MatchError("invalid IP Address for Blacklist IP Range: 127.0.2.2.1")) }) It("returns an error with an invalid end address", func() { - _, err := bindings.NewBlacklistRanges( - bindings.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4.3"}, + _, err := blacklist.NewBlacklistRanges( + blacklist.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4.3"}, ) Expect(err).To(HaveOccurred()) }) It("validates multiple blacklist ranges", func() { - _, err := bindings.NewBlacklistRanges( - bindings.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4"}, - bindings.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4.5"}, + _, err := blacklist.NewBlacklistRanges( + blacklist.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4"}, + blacklist.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.4.5"}, ) Expect(err).To(HaveOccurred()) }) It("validates start IP is before end IP", func() { - _, err := bindings.NewBlacklistRanges( - bindings.BlacklistRange{Start: "10.10.10.10", End: "10.8.10.12"}, + _, err := blacklist.NewBlacklistRanges( + blacklist.BlacklistRange{Start: "10.10.10.10", End: "10.8.10.12"}, ) Expect(err).To(MatchError("invalid Blacklist IP Range: Start 10.10.10.10 has to be before End 10.8.10.12")) }) It("accepts start and end as the same", func() { - _, err := bindings.NewBlacklistRanges( - bindings.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.2"}, + _, err := blacklist.NewBlacklistRanges( + blacklist.BlacklistRange{Start: "127.0.2.2", End: "127.0.2.2"}, ) Expect(err).ToNot(HaveOccurred()) }) @@ -56,15 +56,15 @@ var _ = Describe("BlacklistRanges", func() { Describe("CheckBlacklist()", func() { It("allows all urls for empty blacklist range", func() { - ranges, _ := bindings.NewBlacklistRanges() + ranges, _ := blacklist.NewBlacklistRanges() err := ranges.CheckBlacklist(net.ParseIP("127.0.0.1")) Expect(err).ToNot(HaveOccurred()) }) It("returns an error when the IP is in the blacklist range", func() { - ranges, err := bindings.NewBlacklistRanges( - bindings.BlacklistRange{Start: "127.0.1.2", End: "127.0.3.4"}, + ranges, err := blacklist.NewBlacklistRanges( + blacklist.BlacklistRange{Start: "127.0.1.2", End: "127.0.3.4"}, ) Expect(err).ToNot(HaveOccurred()) @@ -75,7 +75,7 @@ var _ = Describe("BlacklistRanges", func() { Describe("ResolveAddr()", func() { It("does not return an error when able to resolve", func() { - ranges, _ := bindings.NewBlacklistRanges() + ranges, _ := blacklist.NewBlacklistRanges() ip, err := ranges.ResolveAddr("localhost") Expect(err).ToNot(HaveOccurred()) @@ -83,7 +83,7 @@ var _ = Describe("BlacklistRanges", func() { }) It("can resolve addresses with port", func() { - ranges, _ := bindings.NewBlacklistRanges() + ranges, _ := blacklist.NewBlacklistRanges() ip, err := ranges.ResolveAddr("localhost:8080") Expect(err).ToNot(HaveOccurred()) @@ -91,7 +91,7 @@ var _ = Describe("BlacklistRanges", func() { }) It("returns an error when it fails to resolve", func() { - ranges, _ := bindings.NewBlacklistRanges() + ranges, _ := blacklist.NewBlacklistRanges() _, err := ranges.ResolveAddr("vcap.me.junky-garbage") Expect(err).To(HaveOccurred()) @@ -100,24 +100,24 @@ var _ = Describe("BlacklistRanges", func() { Describe("UnmarshalEnv", func() { It("returns an error for non-valid input", func() { - bl := &bindings.BlacklistRanges{} + bl := &blacklist.BlacklistRanges{} Expect(bl.UnmarshalEnv("invalid")).ToNot(Succeed()) Expect(bl.UnmarshalEnv("10.244.0.32-10")).ToNot(Succeed()) }) It("parses the given IP ranges", func() { - bl := &bindings.BlacklistRanges{} + bl := &blacklist.BlacklistRanges{} Expect(bl.UnmarshalEnv("10.0.0.4-10.0.0.8,123.4.5.6-123.4.5.7")).To(Succeed()) - Expect(bl.Ranges).To(Equal([]bindings.BlacklistRange{ + Expect(bl.Ranges).To(Equal([]blacklist.BlacklistRange{ {Start: "10.0.0.4", End: "10.0.0.8"}, {Start: "123.4.5.6", End: "123.4.5.7"}, })) }) It("does not return an error for an empty list", func() { - bl := &bindings.BlacklistRanges{} + bl := &blacklist.BlacklistRanges{} Expect(bl.UnmarshalEnv("")).To(Succeed()) }) }) diff --git a/src/pkg/binding/blacklist/blacklist_suite_test.go b/src/pkg/binding/blacklist/blacklist_suite_test.go new file mode 100644 index 000000000..04b1b6e9a --- /dev/null +++ b/src/pkg/binding/blacklist/blacklist_suite_test.go @@ -0,0 +1,13 @@ +package blacklist + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestBlacklist(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Blacklist Suite") +} diff --git a/src/pkg/binding/poller.go b/src/pkg/binding/poller.go index 188f9b18d..b9f13ec13 100644 --- a/src/pkg/binding/poller.go +++ b/src/pkg/binding/poller.go @@ -1,12 +1,19 @@ package binding import ( + "crypto/tls" + "crypto/x509" "encoding/json" + "fmt" "log" + "net" "net/http" + "net/url" "time" metrics "code.cloudfoundry.org/go-metric-registry" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" + "code.cloudfoundry.org/loggregator-agent-release/src/pkg/simplecache" ) type Poller struct { @@ -17,6 +24,18 @@ type Poller struct { logger *log.Logger bindingRefreshErrorCounter metrics.Counter lastBindingCount metrics.Gauge + invalidDrains metrics.Gauge + blacklistedDrains metrics.Gauge + appLogClient v2.LogClient + checker IPChecker + failedHostsCache *simplecache.SimpleCache[string, bool] + warn bool +} + +//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . IPChecker +type IPChecker interface { + ResolveAddr(host string) (net.IP, error) + CheckBlacklist(ip net.IP) error } type client interface { @@ -51,7 +70,20 @@ type Setter interface { Set(bindings []Binding, bindingCount int) } -func NewPoller(ac client, pi time.Duration, s Setter, m Metrics, logger *log.Logger) *Poller { +var allowedSchemes = []string{"syslog", "syslog-tls", "https", "https-batch"} + +func NewPoller( + ac client, + pi time.Duration, + s Setter, + m Metrics, + logger *log.Logger, + appLogClient v2.LogClient, + checker IPChecker, + warn bool, +) *Poller { + opt := metrics.WithMetricLabels(map[string]string{"unit": "total"}) + p := &Poller{ apiClient: ac, pollingInterval: pi, @@ -65,6 +97,20 @@ func NewPoller(ac client, pi time.Duration, s Setter, m Metrics, logger *log.Log "last_binding_refresh_count", "Current number of bindings received from binding provider during last refresh.", ), + invalidDrains: m.NewGauge( + "invalid_drains", + "Count of invalid drains encountered in last binding fetch. Includes blacklisted drains.", + opt, + ), + blacklistedDrains: m.NewGauge( + "blacklisted_drains", + "Count of blacklisted drains encountered in last binding fetch.", + opt, + ), + appLogClient: appLogClient, + checker: checker, + failedHostsCache: simplecache.New[string, bool](120 * time.Second), + warn: warn, } p.poll() return p @@ -88,7 +134,10 @@ func (p *Poller) poll() { return } if resp.StatusCode != http.StatusOK { - p.logger.Printf("unexpected response from internal bindings endpoint. status code: %d", resp.StatusCode) + p.logger.Printf( + "unexpected response from internal bindings endpoint. status code: %d", + resp.StatusCode, + ) return } @@ -107,9 +156,156 @@ func (p *Poller) poll() { } } - bindingCount := CalculateBindingCount(bindings) + bc := &bindingChecker{ + appLogClient: p.appLogClient, + logger: p.logger, + checker: p.checker, + failedHostsCache: p.failedHostsCache, + warn: p.warn, + } + filteredBindings := bc.checkBindings(bindings) + p.blacklistedDrains.Set(bc.blacklistedDrains) + p.invalidDrains.Set(bc.invalidDrains) + + bindingCount := CalculateBindingCount(filteredBindings) p.lastBindingCount.Set(float64(bindingCount)) - p.store.Set(bindings, bindingCount) + p.store.Set(filteredBindings, bindingCount) +} + +// bindingChecker validates bindings and keeps track of invalid +// and blacklisted drain counts. It also sends warning logs when bindings are rejected. +type bindingChecker struct { + appLogClient v2.LogClient + logger *log.Logger + checker IPChecker + failedHostsCache *simplecache.SimpleCache[string, bool] + warn bool + invalidDrains float64 + blacklistedDrains float64 +} + +// rejectBinding increments the appropriate drain counters and sends a warning log if necessary +func (bc *bindingChecker) rejectBinding(creds []Credentials, msg string, invalid bool, blacklisted ...bool) { + if invalid { + bc.invalidDrains++ + } + if blacklisted != nil { + bc.blacklistedDrains++ + } + if bc.warn { + warnApps(msg, creds, bc.appLogClient, bc.logger) + } +} + +func warnApps(msg string, credentials []Credentials, appLogClient v2.LogClient, logger *log.Logger) { + for _, cred := range credentials { + sendAppLogMessage(msg, cred.Apps, appLogClient, logger) + } +} + +// checkBindings checks the bindings and returns a filtered list of valid bindings. +// It also updates the invalid and blacklisted drains gauge metrics. +func (bc *bindingChecker) checkBindings(bindings []Binding) []Binding { + var filteredBindings []Binding + bc.invalidDrains = 0 + bc.blacklistedDrains = 0 + + for _, b := range bindings { + if len(b.Credentials) == 0 { + bc.logger.Println("No credentials - which include appIDs - for a binding. Check the bindings in the cloud controller.") + continue + } + + u, err := url.Parse(b.Url) + if err != nil { + bc.rejectBinding(b.Credentials, "Cannot parse syslog drain URL.", true) + continue + } + + anonymousUrl := *u + anonymousUrl.User = nil + anonymousUrl.RawQuery = "" + + if invalidScheme(u.Scheme) { + bc.rejectBinding(b.Credentials, fmt.Sprintf("Invalid Scheme %s for syslog drain url %s", u.Scheme, anonymousUrl.String()), true) + continue + } + + if len(u.Host) == 0 { + bc.rejectBinding(b.Credentials, fmt.Sprintf("No hostname found in syslog drain url %s", anonymousUrl.String()), true) + continue + } + + _, exists := bc.failedHostsCache.Get(u.Host) + if exists { + bc.rejectBinding(b.Credentials, fmt.Sprintf("Skipped resolve ip address for syslog drain with url %s due to prior failure", anonymousUrl.String()), false) + continue + } + + ip, err := bc.checker.ResolveAddr(u.Host) + if err != nil { + bc.failedHostsCache.Set(u.Host, true) + bc.rejectBinding(b.Credentials, fmt.Sprintf("Cannot resolve ip address for syslog drain with url %s", anonymousUrl.String()), true) + continue + } + + err = bc.checker.CheckBlacklist(ip) + if err != nil { + bc.rejectBinding(b.Credentials, fmt.Sprintf("Resolved ip address for syslog drain with url %s is blacklisted", anonymousUrl.String()), true, true) + continue + } + + var validCredentials []Credentials + for _, cred := range b.Credentials { + if len(cred.Cert) > 0 && len(cred.Key) > 0 { + _, err := tls.X509KeyPair([]byte(cred.Cert), []byte(cred.Key)) + if err != nil { + bc.rejectBinding([]Credentials{cred}, fmt.Sprintf("failed to load certificate for %s", anonymousUrl.String()), true) + continue + } + } + + if len(cred.CA) > 0 { + certPool := x509.NewCertPool() + ok := certPool.AppendCertsFromPEM([]byte(cred.CA)) + if !ok { + bc.rejectBinding([]Credentials{cred}, fmt.Sprintf("failed to load root CA for %s", anonymousUrl.String()), true) + continue + } + } + + validCredentials = append(validCredentials, cred) + } + + if len(validCredentials) > 0 { + filteredBindings = append(filteredBindings, Binding{ + Url: b.Url, + Credentials: validCredentials, + }) + } + } + return filteredBindings +} + +func sendAppLogMessage(msg string, apps []App, appLogClient v2.LogClient, logger *log.Logger) { + for _, app := range apps { + appId := app.AppID + if appId == "" { + continue + } + v2.EmitAppLog(appLogClient, msg, appId) + logger.Printf("%s for app %s", msg, appId) + } +} + +func invalidScheme(scheme string) bool { + for _, s := range allowedSchemes { + if s == scheme { + return false + } + } + + return true } func CalculateBindingCount(bindings []Binding) int { diff --git a/src/pkg/binding/poller_test.go b/src/pkg/binding/poller_test.go index bc6212e37..d60d1c654 100644 --- a/src/pkg/binding/poller_test.go +++ b/src/pkg/binding/poller_test.go @@ -1,4 +1,4 @@ -package binding_test +package binding import ( "bytes" @@ -6,34 +6,43 @@ import ( "errors" "io" "log" + "net" "net/http" "sync/atomic" "time" metricsHelpers "code.cloudfoundry.org/go-metric-registry/testhelpers" + "code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" + "code.cloudfoundry.org/loggregator-agent-release/src/pkg/simplecache" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - - "code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding" + "github.com/onsi/gomega/gbytes" ) var _ = Describe("Poller", func() { var ( + logBuffer *gbytes.Buffer + logger *log.Logger + apiClient *fakeAPIClient store *fakeStore metrics *metricsHelpers.SpyMetricsRegistry - logger = log.New(GinkgoWriter, "", 0) + logClient v2.LogClient ) BeforeEach(func() { + logBuffer = gbytes.NewBuffer() + logger = log.New(logBuffer, "", 0) apiClient = newFakeAPIClient() store = newFakeStore() metrics = metricsHelpers.NewMetricsRegistry() + logClient = testhelper.NewSpyLogClient() }) It("polls for bindings on an interval", func() { - p := binding.NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger) + p := NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger, logClient, &dummyIPChecker{}, false) go p.Poll() Eventually(apiClient.called).Should(BeNumerically(">=", 2)) @@ -41,15 +50,15 @@ var _ = Describe("Poller", func() { It("calls the api client and stores the result", func() { apiClient.bindings <- response{ - Results: []binding.Binding{ + Results: []Binding{ { - Url: "drain-0", - Credentials: []binding.Credentials{ + Url: "syslog://drain-0", + Credentials: []Credentials{ { - Cert: "cert0", Key: "key0", CA: "ca0", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, { - Cert: "cert1", Key: "key1", CA: "ca1", Apps: []binding.App{ + Apps: []App{ {Hostname: "app-hostname1", AppID: "app-id-1"}, {Hostname: "app-hostname2", AppID: "app-id-2"}, }, @@ -57,30 +66,30 @@ var _ = Describe("Poller", func() { }, }, { - Url: "drain-1", - Credentials: []binding.Credentials{ + Url: "syslog://drain-1", + Credentials: []Credentials{ { - Cert: "cert2", Key: "key2", CA: "ca2", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, }, } - p := binding.NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger) + p := NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger, logClient, &dummyIPChecker{}, false) go p.Poll() - var expectedBindings []binding.Binding + var expectedBindings []Binding Eventually(store.bindings).Should(Receive(&expectedBindings)) - Expect(expectedBindings).To(ConsistOf([]binding.Binding{ + Expect(expectedBindings).To(ConsistOf([]Binding{ { - Url: "drain-0", - Credentials: []binding.Credentials{ + Url: "syslog://drain-0", + Credentials: []Credentials{ { - Cert: "cert0", Key: "key0", CA: "ca0", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, { - Cert: "cert1", Key: "key1", CA: "ca1", Apps: []binding.App{ + Apps: []App{ {Hostname: "app-hostname1", AppID: "app-id-1"}, {Hostname: "app-hostname2", AppID: "app-id-2"}, }, @@ -88,10 +97,10 @@ var _ = Describe("Poller", func() { }, }, { - Url: "drain-1", - Credentials: []binding.Credentials{ + Url: "syslog://drain-1", + Credentials: []Credentials{ { - Cert: "cert2", Key: "key2", CA: "ca2", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, @@ -101,20 +110,20 @@ var _ = Describe("Poller", func() { It("fetches the next page of bindings and stores the result", func() { apiClient.bindings <- response{ NextID: 2, - Results: []binding.Binding{ + Results: []Binding{ { - Url: "drain-0", - Credentials: []binding.Credentials{ + Url: "syslog://drain-0", + Credentials: []Credentials{ { - Cert: "cert0", Key: "key0", CA: "ca0", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, { - Url: "drain-1", - Credentials: []binding.Credentials{ + Url: "syslog://drain-1", + Credentials: []Credentials{ { - Cert: "cert1", Key: "key1", CA: "ca1", Apps: []binding.App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, + Apps: []App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, }, }, }, @@ -122,62 +131,62 @@ var _ = Describe("Poller", func() { } apiClient.bindings <- response{ - Results: []binding.Binding{ + Results: []Binding{ { - Url: "drain-2", - Credentials: []binding.Credentials{ + Url: "syslog://drain-2", + Credentials: []Credentials{ { - Cert: "cert2", Key: "key2", CA: "ca2", Apps: []binding.App{{Hostname: "app-hostname2", AppID: "app-id-2"}}, + Apps: []App{{Hostname: "app-hostname2", AppID: "app-id-2"}}, }, }, }, { - Url: "drain-3", - Credentials: []binding.Credentials{ + Url: "syslog://drain-3", + Credentials: []Credentials{ { - Cert: "cert3", Key: "key3", CA: "ca3", Apps: []binding.App{{Hostname: "app-hostname3", AppID: "app-id-3"}}, + Apps: []App{{Hostname: "app-hostname3", AppID: "app-id-3"}}, }, }, }, }, } - p := binding.NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger) + p := NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger, logClient, &dummyIPChecker{}, false) go p.Poll() - var expectedBindings []binding.Binding + var expectedBindings []Binding Eventually(store.bindings).Should(Receive(&expectedBindings)) Expect(expectedBindings).To(ConsistOf( - []binding.Binding{ + []Binding{ { - Url: "drain-0", - Credentials: []binding.Credentials{ + Url: "syslog://drain-0", + Credentials: []Credentials{ { - Cert: "cert0", Key: "key0", CA: "ca0", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, { - Url: "drain-1", - Credentials: []binding.Credentials{ + Url: "syslog://drain-1", + Credentials: []Credentials{ { - Cert: "cert1", Key: "key1", CA: "ca1", Apps: []binding.App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, + Apps: []App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, }, }, }, { - Url: "drain-2", - Credentials: []binding.Credentials{ + Url: "syslog://drain-2", + Credentials: []Credentials{ { - Cert: "cert2", Key: "key2", CA: "ca2", Apps: []binding.App{{Hostname: "app-hostname2", AppID: "app-id-2"}}, + Apps: []App{{Hostname: "app-hostname2", AppID: "app-id-2"}}, }, }, }, { - Url: "drain-3", - Credentials: []binding.Credentials{ + Url: "syslog://drain-3", + Credentials: []Credentials{ { - Cert: "cert3", Key: "key3", CA: "ca3", Apps: []binding.App{{Hostname: "app-hostname3", AppID: "app-id-3"}}, + Apps: []App{{Hostname: "app-hostname3", AppID: "app-id-3"}}, }, }, }, @@ -188,7 +197,7 @@ var _ = Describe("Poller", func() { }) It("tracks the number of API errors", func() { - p := binding.NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger) + p := NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger, logClient, &dummyIPChecker{}, false) go p.Poll() apiClient.errors <- errors.New("expected") @@ -201,7 +210,7 @@ var _ = Describe("Poller", func() { It("does not update the stores if the response code is bad", func() { apiClient.statusCode <- 404 - p := binding.NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger) + p := NewPoller(apiClient, 10*time.Millisecond, store, metrics, logger, logClient, &dummyIPChecker{}, false) go p.Poll() Eventually(store.bindings).Should(BeEmpty()) @@ -209,76 +218,352 @@ var _ = Describe("Poller", func() { It("tracks the number of bindings returned from CAPI", func() { apiClient.bindings <- response{ - Results: []binding.Binding{ + Results: []Binding{ + { + Url: "syslog://drain-0.example.com", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + { + Url: "syslog://drain-1.example.com", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, + }, + }, + }, + }, + } + NewPoller(apiClient, time.Hour, store, metrics, logger, logClient, &dummyIPChecker{}, true) + + Expect(metrics.GetMetric("last_binding_refresh_count", nil).Value()). + To(BeNumerically("==", 2)) + }) + + It("filters invalid bindings", func() { + apiClient.bindings <- response{ + Results: []Binding{ { Url: "drain-0", - Credentials: []binding.Credentials{ + Credentials: []Credentials{ { - Cert: "cert0", Key: "key0", CA: "ca0", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Cert: "cert0", Key: "key0", CA: "ca0", Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, { Url: "drain-1", - Credentials: []binding.Credentials{ + Credentials: []Credentials{ + { + Cert: "cert1", Key: "key1", CA: "ca1", Apps: []App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, + }, + }, + }, + { + Url: "syslog://drain-2.example.com", + Credentials: []Credentials{ { - Cert: "cert1", Key: "key1", CA: "ca1", Apps: []binding.App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, + Apps: []App{{Hostname: "app-hostname2", AppID: "app-id-2"}}, + }, + }, + }, + { + Url: "syslog://blacklisted_domain", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, }, } - binding.NewPoller(apiClient, time.Hour, store, metrics, logger) + NewPoller(apiClient, time.Hour, store, metrics, logger, logClient, &dummyIPChecker{}, false) Expect(metrics.GetMetric("last_binding_refresh_count", nil).Value()). - To(BeNumerically("==", 2)) + To(BeNumerically("==", 1)) + tags := map[string]string{"unit": "total"} + Expect(metrics.GetMetric("invalid_drains", tags).Value()). + To(BeNumerically("==", 3)) + Expect(metrics.GetMetric("blacklisted_drains", tags).Value()). + To(BeNumerically("==", 1)) }) It("tracks the isolated CalculateBindingsCount call", func() { - noBinding := []binding.Binding{} - singleBinding := []binding.Binding{ + noBinding := []Binding{} + singleBinding := []Binding{ { Url: "drain-0", - Credentials: []binding.Credentials{ + Credentials: []Credentials{ { - Cert: "cert0", Key: "key0", CA: "ca0", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Cert: "cert0", Key: "key0", CA: "ca0", Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, { Url: "drain-1", - Credentials: []binding.Credentials{ + Credentials: []Credentials{ { - Cert: "cert1", Key: "key1", CA: "ca1", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Cert: "cert1", Key: "key1", CA: "ca1", Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, } - multipleBindings := []binding.Binding{ + multipleBindings := []Binding{ { Url: "drain-0", - Credentials: []binding.Credentials{ + Credentials: []Credentials{ { - Cert: "cert0", Key: "key0", CA: "ca0", Apps: []binding.App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + Cert: "cert0", Key: "key0", CA: "ca0", Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, }, }, }, { Url: "drain-1", - Credentials: []binding.Credentials{ + Credentials: []Credentials{ { - Cert: "cert1", Key: "key1", CA: "ca1", Apps: []binding.App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, + Cert: "cert1", Key: "key1", CA: "ca1", Apps: []App{{Hostname: "app-hostname1", AppID: "app-id-1"}}, }, }, }, } - Expect(binding.CalculateBindingCount(noBinding)). + Expect(CalculateBindingCount(noBinding)). To(BeNumerically("==", 0)) - Expect(binding.CalculateBindingCount(singleBinding)). + Expect(CalculateBindingCount(singleBinding)). To(BeNumerically("==", 1)) - Expect(binding.CalculateBindingCount(multipleBindings)). + Expect(CalculateBindingCount(multipleBindings)). To(BeNumerically("==", 2)) }) + + Describe("checkBindings", func() { + + var bndChecker *bindingChecker + var logClient *testhelper.SpyLogClient + + BeforeEach(func() { + logBuffer.Clear() //nolint:errcheck + logClient = testhelper.NewSpyLogClient() + + bndChecker = &bindingChecker{ + appLogClient: logClient, + logger: logger, + checker: &dummyIPChecker{}, + failedHostsCache: simplecache.New[string, bool](120 * time.Second), + warn: true, + } + }) + + It("returns no binding and writes an error if no credentials are set for a binding", func() { + + bindings := []Binding{ + { + Url: "syslog://my-syslog-servers.com", + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say("No credentials - which include appIDs - for a binding. Check the bindings in the cloud controller.")) + }) + + It("returns no binding and writes an error if the binding url cannot be parsed", func() { + + bindings := []Binding{ + { + Url: "http://example.com/\x7f", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say("Cannot parse syslog drain URL. for app app-id-0")) + Expect(logClient.Message()).To(ContainElement(Equal("Cannot parse syslog drain URL."))) + Expect(len(logClient.Message())).To(Equal(1)) + Expect(bndChecker.invalidDrains).To(Equal(float64(1))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + + It("returns no binding and writes an error if the binding url has invalid scheme", func() { + + bindings := []Binding{ + { + Url: "syslog-ssl://drain-0.com?user=trlala&password=123213", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say("Invalid Scheme syslog-ssl for syslog drain url syslog-ssl://drain-0.com for app app-id-0")) + Expect(logClient.Message()).To(ContainElement(Equal("Invalid Scheme syslog-ssl for syslog drain url syslog-ssl://drain-0.com"))) + Expect(bndChecker.invalidDrains).To(Equal(float64(1))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + + It("returns no binding if a host cannot be parsed from the given url", func() { + bindings := []Binding{ + { + Url: "syslog:/invalid-url-a-slash-is-missing", + Credentials: []Credentials{ + { + Cert: "cert0", Key: "key0", CA: "ca0", Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say(("No hostname found in syslog drain url syslog:/invalid-url-a-slash-is-missing for app app-id-0"))) + Expect(logClient.Message()).To(ContainElement(Equal("No hostname found in syslog drain url syslog:/invalid-url-a-slash-is-missing"))) + Expect(bndChecker.invalidDrains).To(Equal(float64(1))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + + It("returns no binding when there is a prior IP checking failure for a URL", func() { + bindings := []Binding{ + { + Url: "syslog://syslog-drain-test-37c4f6db-12e2-4206-8bb2-c8d6f440d4d2.example.com", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + cache := simplecache.New[string, bool](120 * time.Second) + cache.Set("syslog-drain-test-37c4f6db-12e2-4206-8bb2-c8d6f440d4d2.example.com", true) + bndChecker.failedHostsCache = cache + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say(("Skipped resolve ip address for syslog drain with url syslog://syslog-drain-test-37c4f6db-12e2-4206-8bb2-c8d6f440d4d2.example.com due to prior failure for app app-id-0"))) + Expect(logClient.Message()).To(ContainElement(Equal("Skipped resolve ip address for syslog drain with url syslog://syslog-drain-test-37c4f6db-12e2-4206-8bb2-c8d6f440d4d2.example.com due to prior failure"))) + Expect(bndChecker.invalidDrains).To(Equal(float64(0))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + + It("returns no binding when the URL cannot be resolved", func() { + bindings := []Binding{ + { + Url: "syslog://fail_to_resolve_ip", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say(("Cannot resolve ip address for syslog drain with url syslog://fail_to_resolve_ip for app app-id-0"))) + Expect(logClient.Message()).To(ContainElement(Equal("Cannot resolve ip address for syslog drain with url syslog://fail_to_resolve_ip"))) + Expect(bndChecker.invalidDrains).To(Equal(float64(1))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + + It("returns no binding which has a blacklisted IP", func() { + bindings := []Binding{ + { + Url: "syslog://blacklisted_domain", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say(("Resolved ip address for syslog drain with url syslog://blacklisted_domain is blacklisted for app app-id-0"))) + Expect(logClient.Message()).To(ContainElement(Equal("Resolved ip address for syslog drain with url syslog://blacklisted_domain is blacklisted"))) + Expect(bndChecker.invalidDrains).To(Equal(float64(1))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(1))) + }) + + It("returns no binding when key pair cannot be loaded", func() { + bindings := []Binding{ + { + Url: "syslog-tls://drain-0", + Credentials: []Credentials{ + { + Cert: "cert0", Key: "key0", Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say(("failed to load certificate for syslog-tls://drain-0 for app app-id-0"))) + Expect(logClient.Message()).To(ContainElement(Equal("failed to load certificate for syslog-tls://drain-0"))) + Expect(bndChecker.invalidDrains).To(Equal(float64(1))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + + It("returns no binding when CA cannot be loaded", func() { + bindings := []Binding{ + { + Url: "syslog-tls://drain-0", + Credentials: []Credentials{ + { + CA: "ca0", Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(BeEmpty()) + Expect(logBuffer).Should(gbytes.Say(("failed to load root CA for syslog-tls://drain-0 for app app-id-0"))) + Expect(logClient.Message()).To(ContainElement(Equal("failed to load root CA for syslog-tls://drain-0"))) + Expect(bndChecker.invalidDrains).To(Equal(float64(1))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + + It("returns bindings when there are no certificates set", func() { + bindings := []Binding{ + { + Url: "syslog-tls://drain-0", + Credentials: []Credentials{ + { + Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}}, + }, + }, + }, + } + + filteredBindings := bndChecker.checkBindings(bindings) + + Expect(filteredBindings).To(HaveLen(1)) + Expect(logBuffer.Contents()).Should(BeEmpty()) + Expect(logClient.Message()).To(BeEmpty()) + Expect(bndChecker.invalidDrains).To(Equal(float64(0))) + Expect(bndChecker.blacklistedDrains).To(Equal(float64(0))) + }) + }) }) type fakeAPIClient struct { @@ -329,20 +614,40 @@ func (c *fakeAPIClient) called() int64 { } type fakeStore struct { - bindings chan []binding.Binding + bindings chan []Binding } func newFakeStore() *fakeStore { return &fakeStore{ - bindings: make(chan []binding.Binding, 100), + bindings: make(chan []Binding, 100), } } -func (c *fakeStore) Set(b []binding.Binding, bindingCount int) { +func (c *fakeStore) Set(b []Binding, bindingCount int) { c.bindings <- b } type response struct { - Results []binding.Binding + Results []Binding NextID int `json:"next_id"` } + +type dummyIPChecker struct{} + +func (d *dummyIPChecker) ResolveAddr(host string) (net.IP, error) { + if host == "fail_to_resolve_ip" { + return net.IPv4(127, 0, 0, 1), errors.New(host) + } + if host == "blacklisted_domain" { + return net.IPv4(192, 168, 188, 15), nil + } + + return net.IPv4(127, 0, 0, 1), nil +} + +func (*dummyIPChecker) CheckBlacklist(ip net.IP) error { + if ip.String() == "192.168.188.15" { + return errors.New(ip.String()) + } + return nil +} diff --git a/src/pkg/egress/syslog/retry_writer_test.go b/src/pkg/egress/syslog/retry_writer_test.go index e44bae451..45b348394 100644 --- a/src/pkg/egress/syslog/retry_writer_test.go +++ b/src/pkg/egress/syslog/retry_writer_test.go @@ -3,11 +3,9 @@ package syslog_test import ( "errors" "net/url" - "sync" "sync/atomic" "time" - "code.cloudfoundry.org/go-loggregator/v10" v2 "code.cloudfoundry.org/go-loggregator/v10/rpc/loggregator_v2" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog" @@ -165,81 +163,6 @@ func (s *spyWriteCloser) WriteAttempts() int { return int(atomic.LoadInt64(&s.writeAttempts)) } -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 -} - func buildDelay(multiplier time.Duration) func(int) time.Duration { return func(attempt int) time.Duration { return time.Duration(attempt) * multiplier diff --git a/src/pkg/egress/syslog/syslog_connector.go b/src/pkg/egress/syslog/syslog_connector.go index a13321c60..66134f24c 100644 --- a/src/pkg/egress/syslog/syslog_connector.go +++ b/src/pkg/egress/syslog/syslog_connector.go @@ -6,11 +6,10 @@ import ( "context" - metrics "code.cloudfoundry.org/go-metric-registry" - "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" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" ) type Binding struct { @@ -33,32 +32,20 @@ type Credentials struct { CA string `json:"ca"` } -// LogClient is used to emit logs. -type LogClient interface { - EmitLog(message string, opts ...loggregator.EmitLogOption) -} - -// nullLogClient ensures that the LogClient is in fact optional. -type nullLogClient struct{} - -// EmitLog drops all messages into /dev/null. -func (nullLogClient) EmitLog(message string, opts ...loggregator.EmitLogOption) { -} - type writerFactory interface { - NewWriter(*URLBinding) (egress.WriteCloser, error) + NewWriter(*URLBinding, v2.LogClient) (egress.WriteCloser, error) } // SyslogConnector creates the various egress syslog writers. type SyslogConnector struct { skipCertVerify bool - logClient LogClient wg egress.WaitGroup - sourceIndex string writerFactory writerFactory + metricClient metricClient - metricClient metricClient droppedMetric metrics.Counter + + logClient v2.LogClient } // NewSyslogConnector configures and returns a new SyslogConnector. @@ -78,7 +65,6 @@ func NewSyslogConnector( sc := &SyslogConnector{ skipCertVerify: skipCertVerify, wg: wg, - logClient: nullLogClient{}, writerFactory: f, metricClient: m, @@ -95,10 +81,9 @@ type ConnectorOption func(*SyslogConnector) // WithLogClient returns a ConnectorOption that will set up logging for any // information about a binding. -func WithLogClient(logClient LogClient, sourceIndex string) ConnectorOption { +func WithLogClient(logClient v2.LogClient) ConnectorOption { return func(sc *SyslogConnector) { sc.logClient = logClient - sc.sourceIndex = sourceIndex } } @@ -110,7 +95,7 @@ func (w *SyslogConnector) Connect(ctx context.Context, b Binding) (egress.Writer return nil, err } - writer, err := w.writerFactory.NewWriter(urlBinding) + writer, err := w.writerFactory.NewWriter(urlBinding, w.logClient) if err != nil { return nil, err } @@ -138,8 +123,8 @@ func (w *SyslogConnector) Connect(ctx context.Context, b Binding) (egress.Writer w.droppedMetric.Add(float64(missed)) drainDroppedMetric.Add(float64(missed)) - w.emitLoggregatorErrorLog(b.AppId, fmt.Sprintf("%d messages lost for application %s in user provided syslog drain with url %s", missed, b.AppId, anonymousUrl.String())) w.emitStandardOutErrorLog(b.AppId, urlBinding.Scheme(), anonymousUrl.String(), missed) + v2.EmitAppLog(w.logClient, fmt.Sprintf("%d messages lost for application %s in user provided syslog drain with url %s", missed, b.AppId, anonymousUrl.String()), b.AppId) }), w.wg) filteredWriter, err := NewFilteringDrainWriter(b, dw) @@ -151,20 +136,6 @@ func (w *SyslogConnector) Connect(ctx context.Context, b Binding) (egress.Writer return filteredWriter, nil } -func (w *SyslogConnector) emitLoggregatorErrorLog(appID, message string) { - if appID == "" { - return - } - option := loggregator.WithAppInfo(appID, "LGR", "") - w.logClient.EmitLog(message, option) - - option = loggregator.WithAppInfo( - appID, - "SYS", - w.sourceIndex, - ) - w.logClient.EmitLog(message, option) -} func (w *SyslogConnector) emitStandardOutErrorLog(appID, scheme, url string, missed int) { errorAppOrAggregate := fmt.Sprintf("for %s's app drain", appID) if appID == "" { diff --git a/src/pkg/egress/syslog/syslog_connector_test.go b/src/pkg/egress/syslog/syslog_connector_test.go index 37dbef5b6..8609e6383 100644 --- a/src/pkg/egress/syslog/syslog_connector_test.go +++ b/src/pkg/egress/syslog/syslog_connector_test.go @@ -1,6 +1,8 @@ package syslog_test import ( + "code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper" + "errors" "fmt" "sync/atomic" @@ -12,6 +14,7 @@ import ( metricsHelpers "code.cloudfoundry.org/go-metric-registry/testhelpers" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -175,14 +178,14 @@ var _ = Describe("SyslogConnector", func() { }).Value).Should(BeNumerically(">=", 10000)) }) - It("emits a LGR and SYS log to the log client about logs that have been dropped", func() { - logClient := newSpyLogClient() + It("emits a LGR log to the log client about logs that have been dropped", func() { + logClient := testhelper.NewSpyLogClient() connector := syslog.NewSyslogConnector( true, spyWaitGroup, writerFactory, sm, - syslog.WithLogClient(logClient, "3"), + syslog.WithLogClient(logClient), ) binding := syslog.Binding{AppId: "app-id", @@ -205,26 +208,22 @@ var _ = Describe("SyslogConnector", func() { } }(writer) - Eventually(logClient.message).Should(ContainElement(MatchRegexp("\\d messages lost for application (.*) in user provided syslog drain with url"))) - Eventually(logClient.appID).Should(ContainElement("app-id")) + Eventually(logClient.Message).Should(ContainElement(MatchRegexp("\\d messages lost for application (.*) in user provided syslog drain with url"))) + Eventually(logClient.AppID).Should(ContainElement("app-id")) - Eventually(logClient.sourceType).Should(HaveLen(2)) - Eventually(logClient.sourceType).Should(HaveKey("LGR")) - Eventually(logClient.sourceType).Should(HaveKey("SYS")) + Eventually(logClient.SourceType).Should(HaveLen(1)) + Eventually(logClient.SourceType).Should(HaveKey("LGR")) - Eventually(logClient.sourceInstance).Should(HaveLen(2)) - Eventually(logClient.sourceInstance).Should(HaveKey("")) - Eventually(logClient.sourceInstance).Should(HaveKey("3")) }) It("doesn't emit LGR and SYS log to the log client about aggregate drains drops", func() { - logClient := newSpyLogClient() + logClient := testhelper.NewSpyLogClient() connector := syslog.NewSyslogConnector( true, spyWaitGroup, writerFactory, sm, - syslog.WithLogClient(logClient, "3"), + syslog.WithLogClient(logClient), ) binding := syslog.Binding{Drain: syslog.Drain{Url: "dropping://"}} @@ -243,7 +242,7 @@ var _ = Describe("SyslogConnector", func() { } }(writer) - Consistently(logClient.message).ShouldNot(ContainElement(MatchRegexp("\\d messages lost for application (.*) in user provided syslog drain with url"))) + Consistently(logClient.Message()).ShouldNot(ContainElement(MatchRegexp("\\d messages lost for application (.*) in user provided syslog drain with url"))) }) It("does not panic on unknown dropped metrics", func() { @@ -280,6 +279,7 @@ type stubWriterFactory struct { func (f *stubWriterFactory) NewWriter( urlBinding *syslog.URLBinding, + appLogClient v2.LogClient, ) (egress.WriteCloser, error) { f.called = true return f.writer, f.err diff --git a/src/pkg/egress/syslog/tcp.go b/src/pkg/egress/syslog/tcp.go index 9e4edd46b..2509ca4dc 100644 --- a/src/pkg/egress/syslog/tcp.go +++ b/src/pkg/egress/syslog/tcp.go @@ -13,6 +13,7 @@ import ( "code.cloudfoundry.org/go-loggregator/v10/rpc/loggregator_v2" metrics "code.cloudfoundry.org/go-metric-registry" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" ) // DialFunc represents a method for creating a connection, either TCP or TLS. @@ -32,6 +33,8 @@ type TCPWriter struct { syslogConverter *Converter egressMetric metrics.Counter + + appLogClient v2.LogClient } // NewTCPWriter creates a new TCP syslog writer. @@ -40,6 +43,7 @@ func NewTCPWriter( netConf NetworkTimeoutConfig, egressMetric metrics.Counter, c *Converter, + appLogClient v2.LogClient, ) egress.WriteCloser { dialer := &net.Dialer{ Timeout: netConf.DialTimeout, @@ -58,6 +62,7 @@ func NewTCPWriter( scheme: "syslog", egressMetric: egressMetric, syslogConverter: c, + appLogClient: appLogClient, } return w @@ -104,6 +109,10 @@ func (w *TCPWriter) connection() (net.Conn, error) { func (w *TCPWriter) connect() (net.Conn, error) { conn, err := w.dialFunc(w.url.Host) if err != nil { + appLogMessage := fmt.Sprintf("Failed to connect to %s", w.url.String()) + v2.EmitAppLog(w.appLogClient, appLogMessage, w.appID) + platformLogMessage := fmt.Sprintf("%s for app %s", appLogMessage, w.appID) + log.Print(platformLogMessage) return nil, err } w.conn = conn diff --git a/src/pkg/egress/syslog/tcp_test.go b/src/pkg/egress/syslog/tcp_test.go index b40d5c334..2a5808a84 100644 --- a/src/pkg/egress/syslog/tcp_test.go +++ b/src/pkg/egress/syslog/tcp_test.go @@ -2,6 +2,9 @@ package syslog_test import ( "bufio" + + "code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper" + "fmt" "io" "net" @@ -51,12 +54,14 @@ var _ = Describe("TCPWriter", func() { BeforeEach(func() { var err error egressCounter = &metricsHelpers.SpyMetric{} + logClient := testhelper.NewSpyLogClient() writer = syslog.NewTCPWriter( binding, netConf, egressCounter, syslog.NewConverter(), + logClient, ) Expect(err).ToNot(HaveOccurred()) }) @@ -183,12 +188,14 @@ var _ = Describe("TCPWriter", func() { It("write returns an error", func() { env := buildLogEnvelope("APP", "2", "just a test", loggregator_v2.Log_OUT) binding.URL, _ = url.Parse("syslog://localhost-garbage:9999") + logClient := testhelper.NewSpyLogClient() writer := syslog.NewTCPWriter( binding, netConf, &metricsHelpers.SpyMetric{}, syslog.NewConverter(), + logClient, ) errs := make(chan error, 1) @@ -208,11 +215,13 @@ var _ = Describe("TCPWriter", func() { Context("with a happy dialer", func() { BeforeEach(func() { var err error + logClient := testhelper.NewSpyLogClient() writer = syslog.NewTCPWriter( binding, netConf, &metricsHelpers.SpyMetric{}, syslog.NewConverter(), + logClient, ) Expect(err).ToNot(HaveOccurred()) diff --git a/src/pkg/egress/syslog/tls.go b/src/pkg/egress/syslog/tls.go index 1610efc4f..0946437c7 100644 --- a/src/pkg/egress/syslog/tls.go +++ b/src/pkg/egress/syslog/tls.go @@ -7,6 +7,7 @@ import ( metrics "code.cloudfoundry.org/go-metric-registry" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" ) // TLSWriter represents a syslog writer that connects over unencrypted TCP. @@ -27,6 +28,7 @@ func NewTLSWriter( tlsConf *tls.Config, egressMetric metrics.Counter, syslogConverter *Converter, + appLogClient v2.LogClient, ) egress.WriteCloser { dialer := &net.Dialer{ @@ -48,6 +50,7 @@ func NewTLSWriter( scheme: "syslog-tls", egressMetric: egressMetric, syslogConverter: syslogConverter, + appLogClient: appLogClient, }, } diff --git a/src/pkg/egress/syslog/tls_test.go b/src/pkg/egress/syslog/tls_test.go index 0f3adf4c4..b0cf19593 100644 --- a/src/pkg/egress/syslog/tls_test.go +++ b/src/pkg/egress/syslog/tls_test.go @@ -55,6 +55,7 @@ var _ = Describe("TLSWriter", func() { Hostname: "test-hostname", URL: url, } + logClient := testhelper.NewSpyLogClient() writer := syslog.NewTLSWriter( binding, netConf, @@ -63,6 +64,7 @@ var _ = Describe("TLSWriter", func() { }, egressCounter, syslog.NewConverter(), + logClient, ) defer writer.Close() diff --git a/src/pkg/egress/syslog/writer_factory.go b/src/pkg/egress/syslog/writer_factory.go index a5d871c6f..5ca68d53f 100644 --- a/src/pkg/egress/syslog/writer_factory.go +++ b/src/pkg/egress/syslog/writer_factory.go @@ -7,6 +7,7 @@ import ( metrics "code.cloudfoundry.org/go-metric-registry" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" ) type metricClient interface { @@ -52,7 +53,7 @@ func NewWriterFactory(internalTlsConfig *tls.Config, externalTlsConfig *tls.Conf } } -func (f WriterFactory) NewWriter(ub *URLBinding) (egress.WriteCloser, error) { +func (f WriterFactory) NewWriter(ub *URLBinding, appLogClient v2.LogClient) (egress.WriteCloser, error) { tlsCfg := f.externalTlsConfig.Clone() if ub.InternalTls { tlsCfg = f.internalTlsConfig.Clone() @@ -60,7 +61,8 @@ func (f WriterFactory) NewWriter(ub *URLBinding) (egress.WriteCloser, error) { if len(ub.Certificate) > 0 && len(ub.PrivateKey) > 0 { cert, err := tls.X509KeyPair(ub.Certificate, ub.PrivateKey) if err != nil { - err = NewWriterFactoryErrorf(ub.URL, "failed to load certificate: %s", err.Error()) + errorMessage := err.Error() + err = NewWriterFactoryErrorf(ub.URL, "failed to load certificate: %s", errorMessage) return nil, err } tlsCfg.Certificates = []tls.Certificate{cert} @@ -120,6 +122,7 @@ func (f WriterFactory) NewWriter(ub *URLBinding) (egress.WriteCloser, error) { f.netConf, egressMetric, converter, + appLogClient, ) case "syslog-tls": w = NewTLSWriter( @@ -128,6 +131,7 @@ func (f WriterFactory) NewWriter(ub *URLBinding) (egress.WriteCloser, error) { tlsCfg, egressMetric, converter, + appLogClient, ) } diff --git a/src/pkg/egress/syslog/writer_factory_test.go b/src/pkg/egress/syslog/writer_factory_test.go index 47e761efa..38e8df787 100644 --- a/src/pkg/egress/syslog/writer_factory_test.go +++ b/src/pkg/egress/syslog/writer_factory_test.go @@ -4,6 +4,7 @@ import ( "crypto/tls" "net/url" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -13,8 +14,9 @@ import ( var _ = Describe("EgressFactory", func() { var ( - f syslog.WriterFactory - sm *metricsHelpers.SpyMetricsRegistry + f syslog.WriterFactory + sm *metricsHelpers.SpyMetricsRegistry + logClient v2.LogClient ) BeforeEach(func() { @@ -30,7 +32,7 @@ var _ = Describe("EgressFactory", func() { URL: url, } - writer, err := f.NewWriter(urlBinding) + writer, err := f.NewWriter(urlBinding, logClient) Expect(err).ToNot(HaveOccurred()) retryWriter, ok := writer.(*syslog.RetryWriter) @@ -49,7 +51,7 @@ var _ = Describe("EgressFactory", func() { URL: url, } - writer, err := f.NewWriter(urlBinding) + writer, err := f.NewWriter(urlBinding, logClient) Expect(err).ToNot(HaveOccurred()) _, ok := writer.(*syslog.HTTPSBatchWriter) @@ -65,7 +67,7 @@ var _ = Describe("EgressFactory", func() { URL: url, } - writer, err := f.NewWriter(urlBinding) + writer, err := f.NewWriter(urlBinding, logClient) Expect(err).ToNot(HaveOccurred()) retryWriter, ok := writer.(*syslog.RetryWriter) @@ -84,7 +86,7 @@ var _ = Describe("EgressFactory", func() { URL: url, } - writer, err := f.NewWriter(urlBinding) + writer, err := f.NewWriter(urlBinding, logClient) Expect(err).ToNot(HaveOccurred()) retryWriter, ok := writer.(*syslog.RetryWriter) @@ -103,7 +105,7 @@ var _ = Describe("EgressFactory", func() { Certificate: []byte("invalid-certificate"), } - _, err = f.NewWriter(urlBinding) + _, err = f.NewWriter(urlBinding, logClient) Expect(err).ToNot(HaveOccurred()) }) }) @@ -117,7 +119,7 @@ var _ = Describe("EgressFactory", func() { PrivateKey: []byte("invalid-private-key"), } - _, err = f.NewWriter(urlBinding) + _, err = f.NewWriter(urlBinding, logClient) Expect(err).ToNot(HaveOccurred()) }) }) @@ -140,7 +142,7 @@ var _ = Describe("EgressFactory", func() { urlBinding.CA = []byte("invalid-ca") } - _, err = f.NewWriter(urlBinding) + _, err = f.NewWriter(urlBinding, logClient) Expect(err).To(MatchError(expectedErr)) }, @@ -166,7 +168,7 @@ var _ = Describe("EgressFactory", func() { AppID: appID, } - _, err = f.NewWriter(urlBinding) + _, err = f.NewWriter(urlBinding, logClient) Expect(err).ToNot(HaveOccurred()) metric := sm.GetMetric("egress", tags) diff --git a/src/pkg/ingress/bindings/filtered_binding_fetcher.go b/src/pkg/ingress/bindings/filtered_binding_fetcher.go index c496ba901..c9a08f0a3 100644 --- a/src/pkg/ingress/bindings/filtered_binding_fetcher.go +++ b/src/pkg/ingress/bindings/filtered_binding_fetcher.go @@ -2,11 +2,9 @@ package bindings import ( "log" - "net" "net/url" "time" - metrics "code.cloudfoundry.org/go-metric-registry" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/simplecache" @@ -14,48 +12,21 @@ import ( var allowedSchemes = []string{"syslog", "syslog-tls", "https", "https-batch"} -//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 . IPChecker -type IPChecker interface { - ResolveAddr(host string) (net.IP, error) - CheckBlacklist(ip net.IP) error -} - -// Metrics is the client used to expose gauge and counter metricsClient. -type metricsClient interface { - NewGauge(name, helpText string, opts ...metrics.MetricOption) metrics.Gauge -} - type FilteredBindingFetcher struct { - ipChecker IPChecker - br binding.Fetcher - warn bool - logger *log.Logger - invalidDrains metrics.Gauge - blacklistedDrains metrics.Gauge - failedHostsCache *simplecache.SimpleCache[string, bool] + ipChecker binding.IPChecker + br binding.Fetcher + warn bool + logger *log.Logger + failedHostsCache *simplecache.SimpleCache[string, bool] } -func NewFilteredBindingFetcher(c IPChecker, b binding.Fetcher, m metricsClient, warn bool, lc *log.Logger) *FilteredBindingFetcher { - opt := metrics.WithMetricLabels(map[string]string{"unit": "total"}) - - invalidDrains := m.NewGauge( - "invalid_drains", - "Count of invalid drains encountered in last binding fetch. Includes blacklisted drains.", - opt, - ) - blacklistedDrains := m.NewGauge( - "blacklisted_drains", - "Count of blacklisted drains encountered in last binding fetch.", - opt, - ) +func NewFilteredBindingFetcher(c binding.IPChecker, b binding.Fetcher, warn bool, lc *log.Logger) *FilteredBindingFetcher { return &FilteredBindingFetcher{ - ipChecker: c, - br: b, - warn: warn, - logger: lc, - invalidDrains: invalidDrains, - blacklistedDrains: blacklistedDrains, - failedHostsCache: simplecache.New[string, bool](120 * time.Second), + ipChecker: c, + br: b, + warn: warn, + logger: lc, + failedHostsCache: simplecache.New[string, bool](120 * time.Second), } } @@ -63,6 +34,12 @@ func (f FilteredBindingFetcher) DrainLimit() int { return f.br.DrainLimit() } +// Fetches bindings from the Syslog Binding Cache +// The validation of the binding URL is moved to the poller in the Syslog Binding Cache. +// The validation here is solely to protect the Syslog Agent from tempering of the +// Syslog Binding Cache's API. The Syslog Agent should always get valid bindings from the +// Syslog Binding Cache, but if it happens that a binding is invalid it will write an error +// message to inform the operators that the Syslog Binding Cache API should be checked func (f *FilteredBindingFetcher) FetchBindings() ([]syslog.Binding, error) { sourceBindings, err := f.br.FetchBindings() if err != nil { @@ -71,12 +48,10 @@ func (f *FilteredBindingFetcher) FetchBindings() ([]syslog.Binding, error) { newBindings := []syslog.Binding{} var invalidDrains float64 - var blacklistedDrains float64 for _, b := range sourceBindings { u, err := url.Parse(b.Drain.Url) if err != nil { invalidDrains += 1 - f.printWarning("Cannot parse syslog drain url for application %s", b.AppId) continue } @@ -85,20 +60,18 @@ func (f *FilteredBindingFetcher) FetchBindings() ([]syslog.Binding, error) { anonymousUrl.RawQuery = "" if invalidScheme(u.Scheme) { - f.printWarning("Invalid scheme %s in syslog drain url %s for application %s", u.Scheme, anonymousUrl.String(), b.AppId) + invalidDrains += 1 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) 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) continue } @@ -106,24 +79,24 @@ func (f *FilteredBindingFetcher) FetchBindings() ([]syslog.Binding, error) { if err != nil { 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) continue } err = f.ipChecker.CheckBlacklist(ip) if err != nil { invalidDrains += 1 - blacklistedDrains += 1 - f.printWarning("Resolved ip address for syslog drain with url %s for application %s is blacklisted", anonymousUrl.String(), b.AppId) continue } newBindings = append(newBindings, b) } - f.blacklistedDrains.Set(blacklistedDrains) - f.invalidDrains.Set(invalidDrains) + if invalidDrains > 0 { + f.printWarning("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API") + } + return newBindings, nil + } func (f FilteredBindingFetcher) printWarning(format string, v ...any) { diff --git a/src/pkg/ingress/bindings/filtered_binding_fetcher_test.go b/src/pkg/ingress/bindings/filtered_binding_fetcher_test.go index a17d25fdb..767b9fdc9 100644 --- a/src/pkg/ingress/bindings/filtered_binding_fetcher_test.go +++ b/src/pkg/ingress/bindings/filtered_binding_fetcher_test.go @@ -6,26 +6,19 @@ import ( "log" "net" - metricsHelpers "code.cloudfoundry.org/go-metric-registry/testhelpers" + "code.cloudfoundry.org/loggregator-agent-release/src/pkg/binding/bindingfakes" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog" "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/bindings" - "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/bindings/bindingsfakes" - . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) var _ = Describe("FilteredBindingFetcher", func() { var ( - log = log.New(GinkgoWriter, "", log.LstdFlags) - filter *bindings.FilteredBindingFetcher - metrics *metricsHelpers.SpyMetricsRegistry + log = log.New(GinkgoWriter, "", log.LstdFlags) + filter *bindings.FilteredBindingFetcher ) - BeforeEach(func() { - metrics = metricsHelpers.NewMetricsRegistry() - }) - It("returns valid bindings", func() { input := []syslog.Binding{ {AppId: "app-id-with-multiple-drains", Hostname: "we.dont.care", Drain: syslog.Drain{Url: "syslog://10.10.10.10"}}, @@ -34,7 +27,7 @@ var _ = Describe("FilteredBindingFetcher", func() { } bindingReader := &SpyBindingReader{bindings: input} - filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{}, bindingReader, metrics, true, log) + filter = bindings.NewFilteredBindingFetcher(&spyIPChecker{}, bindingReader, true, log) actual, err := filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) @@ -44,7 +37,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, true, log) actual, err := filter.FetchBindings() Expect(err).To(HaveOccurred()) @@ -68,7 +61,6 @@ var _ = Describe("FilteredBindingFetcher", func() { filter = bindings.NewFilteredBindingFetcher( &spyIPChecker{}, &SpyBindingReader{bindings: input}, - metrics, warn, log, ) @@ -79,8 +71,7 @@ var _ = Describe("FilteredBindingFetcher", func() { Expect(err).ToNot(HaveOccurred()) Expect(actual).To(Equal([]syslog.Binding{})) - Expect(logBuffer.String()).Should(MatchRegexp("Cannot parse syslog drain url for application")) - Expect(metrics.GetMetric("invalid_drains", map[string]string{"unit": "total"}).Value()).To(Equal(1.0)) + Expect(logBuffer.String()).Should(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) Context("when configured not to warn", func() { @@ -90,7 +81,7 @@ var _ = Describe("FilteredBindingFetcher", func() { It("doesn't log the warning", func() { _, err := filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) - Expect(logBuffer.String()).ToNot(MatchRegexp("Cannot parse syslog drain url for application")) + Expect(logBuffer.String()).ToNot(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) }) }) @@ -112,7 +103,6 @@ var _ = Describe("FilteredBindingFetcher", func() { filter = bindings.NewFilteredBindingFetcher( &spyIPChecker{}, &SpyBindingReader{bindings: input}, - metrics, warn, log, ) @@ -123,8 +113,7 @@ var _ = Describe("FilteredBindingFetcher", func() { Expect(err).ToNot(HaveOccurred()) Expect(actual).To(Equal([]syslog.Binding{})) - Expect(logBuffer.String()).Should(MatchRegexp("No hostname found in syslog drain url (.*) for application")) - Expect(metrics.GetMetric("invalid_drains", map[string]string{"unit": "total"}).Value()).To(Equal(1.0)) + Expect(logBuffer.String()).Should(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) Context("when configured not to warn", func() { @@ -134,7 +123,7 @@ var _ = Describe("FilteredBindingFetcher", func() { It("doesn't log the warning", func() { _, err := filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) - Expect(logBuffer.String()).ToNot(MatchRegexp("No hostname found")) + Expect(logBuffer.String()).ToNot(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) }) }) @@ -160,14 +149,12 @@ var _ = Describe("FilteredBindingFetcher", func() { log.SetOutput(&logBuffer) warn = true - metrics = metricsHelpers.NewMetricsRegistry() }) JustBeforeEach(func() { filter = bindings.NewFilteredBindingFetcher( &spyIPChecker{}, &SpyBindingReader{bindings: input}, - metrics, warn, log, ) @@ -178,8 +165,7 @@ var _ = Describe("FilteredBindingFetcher", func() { Expect(err).ToNot(HaveOccurred()) Expect(actual).To(Equal(input[:3])) - Expect(logBuffer.String()).Should(MatchRegexp("Invalid scheme (.*) in syslog drain url (.*) for application")) - Expect(metrics.GetMetric("invalid_drains", map[string]string{"unit": "total"}).Value()).To(Equal(0.0)) + Expect(logBuffer.String()).Should(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) Context("when configured not to warn", func() { BeforeEach(func() { @@ -188,7 +174,7 @@ var _ = Describe("FilteredBindingFetcher", func() { It("doesn't log the warning", func() { _, err := filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) - Expect(logBuffer.String()).ToNot(MatchRegexp("Invalid scheme")) + Expect(logBuffer.String()).ToNot(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) }) }) @@ -196,13 +182,13 @@ var _ = Describe("FilteredBindingFetcher", func() { Context("when the drain host fails to resolve", func() { var logBuffer bytes.Buffer var warn bool - var mockic *bindingsfakes.FakeIPChecker + var mockic *bindingfakes.FakeIPChecker BeforeEach(func() { logBuffer = bytes.Buffer{} log.SetOutput(&logBuffer) warn = true - mockic = &bindingsfakes.FakeIPChecker{} + mockic = &bindingfakes.FakeIPChecker{} mockic.ResolveAddrReturns(net.IP{}, errors.New("oof ouch ip not resolved")) }) @@ -213,7 +199,6 @@ var _ = Describe("FilteredBindingFetcher", func() { filter = bindings.NewFilteredBindingFetcher( mockic, &SpyBindingReader{bindings: input}, - metrics, warn, log, ) @@ -224,8 +209,7 @@ var _ = Describe("FilteredBindingFetcher", func() { Expect(err).ToNot(HaveOccurred()) Expect(actual).To(Equal([]syslog.Binding{})) - Expect(logBuffer.String()).Should(MatchRegexp("Cannot resolve ip address for syslog drain with url")) - Expect(metrics.GetMetric("invalid_drains", map[string]string{"unit": "total"}).Value()).To(Equal(1.0)) + Expect(logBuffer.String()).Should(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) It("caches bindings that failed to resolve", func() { @@ -233,26 +217,24 @@ var _ = Describe("FilteredBindingFetcher", func() { Expect(err).ToNot(HaveOccurred()) Expect(actual).To(Equal([]syslog.Binding{})) Eventually(func() int { return mockic.ResolveAddrCallCount() }).Should(BeNumerically(">", 0)) - Expect(logBuffer.String()).Should(MatchRegexp("Cannot resolve ip address for syslog drain with url")) - Expect(logBuffer.String()).ToNot(MatchRegexp("Skipped resolve ip address for syslog drain with url")) - Expect(metrics.GetMetric("invalid_drains", map[string]string{"unit": "total"}).Value()).To(Equal(1.0)) + Expect(logBuffer.String()).Should(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) actual, err = filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) Expect(actual).To(BeEmpty()) Consistently(func() int { return mockic.ResolveAddrCallCount() }).Should(Equal(1)) - Expect(logBuffer.String()).Should(MatchRegexp("Skipped resolve ip address for syslog drain with url")) - Expect(metrics.GetMetric("invalid_drains", map[string]string{"unit": "total"}).Value()).To(Equal(1.0)) + Expect(logBuffer.String()).Should(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) Context("when configured not to warn", func() { BeforeEach(func() { warn = false + logBuffer.Truncate(0) }) It("doesn't log the warning", func() { _, err := filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) - Expect(logBuffer.String()).ToNot(MatchRegexp("Cannot resolve ip address for syslog drain with url")) + Expect(logBuffer.String()).ToNot(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) }) }) @@ -278,7 +260,6 @@ var _ = Describe("FilteredBindingFetcher", func() { resolvedIP: net.ParseIP("127.0.0.1"), }, &SpyBindingReader{bindings: input}, - metrics, warn, log, ) @@ -288,21 +269,20 @@ var _ = Describe("FilteredBindingFetcher", func() { actual, err := filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) - Expect(logBuffer.String()).Should(MatchRegexp("Resolved ip address for syslog drain with url (.*) for application (.*) is blacklisted")) + Expect(logBuffer.String()).Should(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) Expect(actual).To(Equal([]syslog.Binding{})) - Expect(metrics.GetMetric("invalid_drains", map[string]string{"unit": "total"}).Value()).To(Equal(1.0)) - Expect(metrics.GetMetric("blacklisted_drains", map[string]string{"unit": "total"}).Value()).To(Equal(1.0)) }) Context("when configured not to warn", func() { BeforeEach(func() { warn = false + logBuffer.Truncate(0) }) It("doesn't log the warning", func() { _, err := filter.FetchBindings() Expect(err).ToNot(HaveOccurred()) - Expect(logBuffer.String()).ToNot(MatchRegexp("Resolved ip address for syslog drain with url (.*) for application (.*) is blacklisted")) + Expect(logBuffer.String()).ToNot(MatchRegexp("Invalid drains detected in the Syslog Agent. This should not happen. Check your Syslog Binding Cache and its API")) }) }) }) diff --git a/src/pkg/ingress/v2/log_client.go b/src/pkg/ingress/v2/log_client.go new file mode 100644 index 000000000..7cb9598ec --- /dev/null +++ b/src/pkg/ingress/v2/log_client.go @@ -0,0 +1,20 @@ +package v2 + +import ( + "code.cloudfoundry.org/go-loggregator/v10" +) + +// LogClient is used to emit logs - i.e. Ingress Client. +type LogClient interface { + EmitLog(message string, opts ...loggregator.EmitLogOption) +} + +// EmitAppLog writes a message in the application log stream using a LogClient. +func EmitAppLog(logClient LogClient, message string, appID string) { + if logClient == nil || appID == "" { + return + } + + logclientOption := loggregator.WithAppInfo(appID, "LGR", "0") + logClient.EmitLog(message, logclientOption) +} diff --git a/src/pkg/ingress/v2/log_client_test.go b/src/pkg/ingress/v2/log_client_test.go new file mode 100644 index 000000000..0b193d7ec --- /dev/null +++ b/src/pkg/ingress/v2/log_client_test.go @@ -0,0 +1,39 @@ +package v2_test + +import ( + "code.cloudfoundry.org/loggregator-agent-release/src/internal/testhelper" + v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("App Log Client", func() { + Describe("Log Stream", func() { + It("should emit a log message for an application with LGR source type with the provided app-ID", func() { + logClient := testhelper.NewSpyLogClient() + + v2.EmitAppLog(logClient, "some-message", "app-id") + + messages := logClient.Message() + appIDs := logClient.AppID() + sourceTypes := logClient.SourceType() + Expect(messages).To(HaveLen(1)) + Expect(messages[0]).To(Equal("some-message")) + Expect(appIDs[0]).To(Equal("app-id")) + Expect(sourceTypes).To(HaveKey("LGR")) + }) + + It("should not emit a log message when the logClient is empty", func() { + logClient := testhelper.NewSpyLogClient() + v2.EmitAppLog(nil, "some-message", "app-id") + Expect(logClient.Message()).To(BeEmpty()) + }) + + It("should not emit a log message when the app-ID is empty", func() { + logClient := testhelper.NewSpyLogClient() + v2.EmitAppLog(nil, "some-message", "") + Expect(logClient.Message()).To(BeEmpty()) + }) + }) + +})