From 2bf19ee9dceb1ce7050d0510ba9e5434561bcf1c Mon Sep 17 00:00:00 2001 From: Joris Baum Date: Tue, 13 Oct 2026 10:11:50 +0200 Subject: [PATCH 1/3] Make downstream otlp host configurable In K8s otel-collector and forwarder agent likely do not run on the same machine. --- jobs/loggr-forwarder-agent-windows/monit | 1 + jobs/loggr-forwarder-agent-windows/spec | 3 ++ jobs/loggr-forwarder-agent/spec | 3 ++ .../templates/bpm.yml.erb | 1 + src/cmd/forwarder-agent/app/app_suite_test.go | 24 +++++++++----- src/cmd/forwarder-agent/app/config.go | 2 ++ src/cmd/forwarder-agent/app/config_test.go | 33 +++++++++++++++++++ .../forwarder-agent/app/forwarder_agent.go | 13 ++++++-- .../app/forwarder_agent_test.go | 8 +++-- 9 files changed, 73 insertions(+), 15 deletions(-) create mode 100644 src/cmd/forwarder-agent/app/config_test.go diff --git a/jobs/loggr-forwarder-agent-windows/monit b/jobs/loggr-forwarder-agent-windows/monit index 60c362071..c45e5aab3 100644 --- a/jobs/loggr-forwarder-agent-windows/monit +++ b/jobs/loggr-forwarder-agent-windows/monit @@ -25,6 +25,7 @@ "AGENT_CIPHER_SUITES" => p("tls.cipher_suites").split(":").join(","), "AGENT_TAGS" => tags.map { |k, v| "#{k}:#{v}" }.join(","), "DOWNSTREAM_INGRESS_PORT_GLOB" => p("downstream_ingress_port_glob"), + "OTEL_COLLECTOR_HOST" => p("otel_collector.host"), "EMIT_OTEL_TRACES" => "#{p("emit_otel_traces")}", "EMIT_OTEL_METRICS" => "#{p("emit_otel_metrics")}", "EMIT_OTEL_LOGS" => "#{p("emit_otel_logs")}", diff --git a/jobs/loggr-forwarder-agent-windows/spec b/jobs/loggr-forwarder-agent-windows/spec index 245cf66eb..e24c106e2 100644 --- a/jobs/loggr-forwarder-agent-windows/spec +++ b/jobs/loggr-forwarder-agent-windows/spec @@ -28,6 +28,9 @@ properties: mTLS configuration. The forwarder assumes the downstream server is serving Loggregator's V2 IngressService. See code.cloudfoundry.org/loggregator-api. default: /var/vcap/jobs/*/config/ingress_port.yml + otel_collector.host: + description: "Host used for downstream OpenTelemetry Collector destinations" + default: "127.0.0.1" deployment: description: "Name of deployment (added as tag on all outgoing v1 envelopes)" diff --git a/jobs/loggr-forwarder-agent/spec b/jobs/loggr-forwarder-agent/spec index af2c3737d..0ef69adfb 100644 --- a/jobs/loggr-forwarder-agent/spec +++ b/jobs/loggr-forwarder-agent/spec @@ -28,6 +28,9 @@ properties: mTLS configuration. The forwarder assumes the downstream server is serving Loggregator's V2 IngressService. See code.cloudfoundry.org/loggregator-api. default: /var/vcap/jobs/*/config/ingress_port.yml + otel_collector.host: + description: "Host used for downstream OpenTelemetry Collector destinations" + default: "127.0.0.1" deployment: description: "Name of deployment (added as tag on all outgoing v1 envelopes)" diff --git a/jobs/loggr-forwarder-agent/templates/bpm.yml.erb b/jobs/loggr-forwarder-agent/templates/bpm.yml.erb index 9ce12267c..0db682a0f 100644 --- a/jobs/loggr-forwarder-agent/templates/bpm.yml.erb +++ b/jobs/loggr-forwarder-agent/templates/bpm.yml.erb @@ -30,6 +30,7 @@ "AGENT_TAGS" => tags.map { |k, v| "#{k}:#{v}" }.join(","), "DOWNSTREAM_INGRESS_PORT_GLOB" => p("downstream_ingress_port_glob"), + "OTEL_COLLECTOR_HOST" => p("otel_collector.host"), "EMIT_OTEL_TRACES" => p("emit_otel_traces"), "EMIT_OTEL_METRICS" => p("emit_otel_metrics"), "EMIT_OTEL_LOGS" => p("emit_otel_logs"), diff --git a/src/cmd/forwarder-agent/app/app_suite_test.go b/src/cmd/forwarder-agent/app/app_suite_test.go index acb8c2596..416b0282f 100644 --- a/src/cmd/forwarder-agent/app/app_suite_test.go +++ b/src/cmd/forwarder-agent/app/app_suite_test.go @@ -224,7 +224,7 @@ type spyOtelColMetricServer struct { // registers it as a gRPC service, and writes out a temp file for the forwarder // agent to recognize it as a destination. The cfgPath it accepts is the folder // under which to write the temp file. -func startSpyOtelColMetricServer(cfgPath string, tc *testhelper.TestCerts, commonName string) *spyOtelColMetricServer { +func startSpyOtelColMetricServer(cfgPath string, tc *testhelper.TestCerts, commonName, host string) *spyOtelColMetricServer { serverCreds, err := plumbing.NewServerCredentials( tc.Cert(commonName), tc.Key(commonName), @@ -232,7 +232,7 @@ func startSpyOtelColMetricServer(cfgPath string, tc *testhelper.TestCerts, commo ) ExpectWithOffset(1, err).NotTo(HaveOccurred()) - lis, err := net.Listen("tcp", "127.0.0.1:") + lis, err := net.Listen("tcp", net.JoinHostPort(host, "0")) ExpectWithOffset(1, err).NotTo(HaveOccurred()) s := &spyOtelColMetricServer{ @@ -244,7 +244,9 @@ func startSpyOtelColMetricServer(cfgPath string, tc *testhelper.TestCerts, commo colmetricspb.RegisterMetricsServiceServer(s.srv, s) go s.srv.Serve(lis) //nolint:errcheck - port, err := strconv.Atoi(strings.Split(s.addr, ":")[1]) + _, portString, err := net.SplitHostPort(s.addr) + ExpectWithOffset(1, err).NotTo(HaveOccurred()) + port, err := strconv.Atoi(portString) ExpectWithOffset(1, err).NotTo(HaveOccurred()) dir, err := os.MkdirTemp(cfgPath, "") @@ -284,7 +286,7 @@ type spyOtelColTraceServer struct { // registers it as a gRPC service, and writes out a temp file for the forwarder // agent to recognize it as a destination. The cfgPath it accepts is the folder // under which to write the temp file. -func startSpyOtelColTraceServer(cfgPath string, tc *testhelper.TestCerts, commonName string) *spyOtelColTraceServer { +func startSpyOtelColTraceServer(cfgPath string, tc *testhelper.TestCerts, commonName, host string) *spyOtelColTraceServer { serverCreds, err := plumbing.NewServerCredentials( tc.Cert(commonName), tc.Key(commonName), @@ -292,7 +294,7 @@ func startSpyOtelColTraceServer(cfgPath string, tc *testhelper.TestCerts, common ) ExpectWithOffset(1, err).NotTo(HaveOccurred()) - lis, err := net.Listen("tcp", "127.0.0.1:") + lis, err := net.Listen("tcp", net.JoinHostPort(host, "0")) ExpectWithOffset(1, err).NotTo(HaveOccurred()) s := &spyOtelColTraceServer{ @@ -304,7 +306,9 @@ func startSpyOtelColTraceServer(cfgPath string, tc *testhelper.TestCerts, common coltracepb.RegisterTraceServiceServer(s.srv, s) go s.srv.Serve(lis) //nolint:errcheck - port, err := strconv.Atoi(strings.Split(s.addr, ":")[1]) + _, portString, err := net.SplitHostPort(s.addr) + ExpectWithOffset(1, err).NotTo(HaveOccurred()) + port, err := strconv.Atoi(portString) ExpectWithOffset(1, err).NotTo(HaveOccurred()) dir, err := os.MkdirTemp(cfgPath, "") @@ -344,7 +348,7 @@ type spyOtelColLogServer struct { // registers it as a gRPC service, and writes out a temp file for the forwarder // agent to recognize it as a destination. The cfgPath it accepts is the folder // under which to write the temp file. -func startSpyOtelColLogServer(cfgPath string, tc *testhelper.TestCerts, commonName string) *spyOtelColLogServer { +func startSpyOtelColLogServer(cfgPath string, tc *testhelper.TestCerts, commonName, host string) *spyOtelColLogServer { serverCreds, err := plumbing.NewServerCredentials( tc.Cert(commonName), tc.Key(commonName), @@ -352,7 +356,7 @@ func startSpyOtelColLogServer(cfgPath string, tc *testhelper.TestCerts, commonNa ) ExpectWithOffset(1, err).NotTo(HaveOccurred()) - lis, err := net.Listen("tcp", "127.0.0.1:") + lis, err := net.Listen("tcp", net.JoinHostPort(host, "0")) ExpectWithOffset(1, err).NotTo(HaveOccurred()) s := &spyOtelColLogServer{ @@ -364,7 +368,9 @@ func startSpyOtelColLogServer(cfgPath string, tc *testhelper.TestCerts, commonNa collogspb.RegisterLogsServiceServer(s.srv, s) go s.srv.Serve(lis) //nolint:errcheck - port, err := strconv.Atoi(strings.Split(s.addr, ":")[1]) + _, portString, err := net.SplitHostPort(s.addr) + ExpectWithOffset(1, err).NotTo(HaveOccurred()) + port, err := strconv.Atoi(portString) ExpectWithOffset(1, err).NotTo(HaveOccurred()) dir, err := os.MkdirTemp(cfgPath, "") diff --git a/src/cmd/forwarder-agent/app/config.go b/src/cmd/forwarder-agent/app/config.go index 1b96c767a..d3ffd0dcb 100644 --- a/src/cmd/forwarder-agent/app/config.go +++ b/src/cmd/forwarder-agent/app/config.go @@ -32,6 +32,7 @@ type Config struct { // receive each envelope. It is assumed to adhere to the Loggregator Ingress // Service and use the provided TLS configuration. DownstreamIngressPortCfg string `env:"DOWNSTREAM_INGRESS_PORT_GLOB, report"` + OTelCollectorHost string `env:"OTEL_COLLECTOR_HOST, report"` GRPC GRPC MetricsServer config.MetricsServer Tags map[string]string `env:"AGENT_TAGS"` @@ -51,6 +52,7 @@ func LoadConfig() Config { Host: "127.0.0.1", Port: 3458, }, + OTelCollectorHost: "127.0.0.1", OtelRetry: OtelRetry{ MaxRetries: 7, RetryQueueSize: 1024, diff --git a/src/cmd/forwarder-agent/app/config_test.go b/src/cmd/forwarder-agent/app/config_test.go new file mode 100644 index 000000000..35fa73abd --- /dev/null +++ b/src/cmd/forwarder-agent/app/config_test.go @@ -0,0 +1,33 @@ +package app_test + +import ( + "os" + + "code.cloudfoundry.org/loggregator-agent-release/src/cmd/forwarder-agent/app" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Config", func() { + BeforeEach(func() { + os.Setenv("AGENT_CA_FILE_PATH", "ca.crt") + os.Setenv("AGENT_CERT_FILE_PATH", "agent.crt") + os.Setenv("AGENT_KEY_FILE_PATH", "agent.key") + os.Setenv("METRICS_CA_FILE_PATH", "metrics-ca.crt") + os.Setenv("METRICS_CERT_FILE_PATH", "metrics.crt") + os.Setenv("METRICS_KEY_FILE_PATH", "metrics.key") + }) + + It("defaults the OTel Collector host to localhost", func() { + os.Setenv("OTEL_COLLECTOR_HOST", "") + + Expect(app.LoadConfig().OTelCollectorHost).To(Equal("127.0.0.1")) + }) + + It("configures the OTel Collector host", func() { + os.Setenv("OTEL_COLLECTOR_HOST", "otel-collector") + + Expect(app.LoadConfig().OTelCollectorHost).To(Equal("otel-collector")) + }) +}) diff --git a/src/cmd/forwarder-agent/app/forwarder_agent.go b/src/cmd/forwarder-agent/app/forwarder_agent.go index db3315cb1..85cfcf3f8 100644 --- a/src/cmd/forwarder-agent/app/forwarder_agent.go +++ b/src/cmd/forwarder-agent/app/forwarder_agent.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "net" "os" "path/filepath" "time" @@ -38,6 +39,7 @@ type ForwarderAgent struct { grpc GRPC v2srv *v2.Server downstreamFilePattern string + otelCollectorHost string log *log.Logger tags map[string]string debugMetrics bool @@ -72,6 +74,7 @@ func NewForwarderAgent( grpc: cfg.GRPC, m: m, downstreamFilePattern: cfg.DownstreamIngressPortCfg, + otelCollectorHost: cfg.OTelCollectorHost, log: log, tags: cfg.Tags, debugMetrics: cfg.MetricsServer.DebugMetrics, @@ -101,7 +104,7 @@ func (s *ForwarderAgent) Run() { ingressDropped.Add(float64(missed)) })) - dests := downstreamDestinations(s.downstreamFilePattern, s.log) + dests := downstreamDestinations(s.downstreamFilePattern, s.otelCollectorHost, s.log) writers := downstreamWriters(dests, s.grpc, s.m, s.emitOTelTraces, s.emitOTelMetrics, s.emitOTelLogs, s.otelRetry, s.log) tagger := egress_v2.NewTagger(s.tags) ew := egress_v2.NewEnvelopeWriter( @@ -185,7 +188,7 @@ type destination struct { Protocol string `yaml:"protocol"` } -func downstreamDestinations(pattern string, l *log.Logger) []destination { +func downstreamDestinations(pattern, otelCollectorHost string, l *log.Logger) []destination { files, err := filepath.Glob(pattern) if err != nil { l.Fatal("Unable to read downstream port location") @@ -207,7 +210,11 @@ func downstreamDestinations(pattern string, l *log.Logger) []destination { if d.Ingress == "" { l.Printf("No ingress port defined in %s. Ignoring this destination.", f) } else { - d.Ingress = fmt.Sprintf("127.0.0.1:%s", d.Ingress) + if d.Protocol == "otelcol" { + d.Ingress = net.JoinHostPort(otelCollectorHost, d.Ingress) + } else { + d.Ingress = net.JoinHostPort("127.0.0.1", d.Ingress) + } dests = append(dests, d) } } diff --git a/src/cmd/forwarder-agent/app/forwarder_agent_test.go b/src/cmd/forwarder-agent/app/forwarder_agent_test.go index de7f483f3..26bab20e0 100644 --- a/src/cmd/forwarder-agent/app/forwarder_agent_test.go +++ b/src/cmd/forwarder-agent/app/forwarder_agent_test.go @@ -70,6 +70,7 @@ var _ = Describe("App", func() { KeyFile: agentCerts.Key(agentCN), }, DownstreamIngressPortCfg: fmt.Sprintf("%s/*/ingress_port.yml", ingressCfgPath), + OTelCollectorHost: "127.0.0.1", MetricsServer: config.MetricsServer{ Port: uint16(metricsPort), CAFile: agentCerts.CA(), @@ -358,9 +359,10 @@ var _ = Describe("App", func() { ) BeforeEach(func() { - otelMetricsServer = startSpyOtelColMetricServer(ingressCfgPath, agentCerts, "otel-collector") - otelTraceServer = startSpyOtelColTraceServer(ingressCfgPath, agentCerts, "otel-collector") - otelLogsServer = startSpyOtelColLogServer(ingressCfgPath, agentCerts, "otel-collector") + agentCfg.OTelCollectorHost = "::1" + otelMetricsServer = startSpyOtelColMetricServer(ingressCfgPath, agentCerts, "otel-collector", agentCfg.OTelCollectorHost) + otelTraceServer = startSpyOtelColTraceServer(ingressCfgPath, agentCerts, "otel-collector", agentCfg.OTelCollectorHost) + otelLogsServer = startSpyOtelColLogServer(ingressCfgPath, agentCerts, "otel-collector", agentCfg.OTelCollectorHost) agentCfg.EmitOTelTraces = true agentCfg.EmitOTelMetrics = true agentCfg.EmitOTelLogs = true From 7dc42fca6409af6ed4b53ae748dab344e28afdb9 Mon Sep 17 00:00:00 2001 From: Joris Baum Date: Tue, 13 Oct 2026 10:59:55 +0200 Subject: [PATCH 2/3] Adapt documentation --- README.md | 6 +++--- jobs/loggr-forwarder-agent-windows/spec | 10 +++++----- jobs/loggr-forwarder-agent/spec | 10 +++++----- 3 files changed, 13 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 8292ff8fd..640f5f0eb 100644 --- a/README.md +++ b/README.md @@ -13,9 +13,9 @@ Accepts the Loggregator v1 api over UDP and forwards it to a downstream Loggrega This is available for backwards compatibility between the v1 API and the Forwarder, Syslog, or Loggregator agents. ### Forwarder Agent -A Loggregator v2 api multiplexer. This Agent Accepts Loggregator v2 and forwards to other agents colocated on the same VM -that accept Loggregator v2. To configure downstream agents, simply place a file named `ingress_port.yml` in the job's config -directory. [example][ingress-port] +A Loggregator v2 API multiplexer. Downstream registration files supply a port and protocol; destinations default to localhost. +`protocol: otelcol` destinations use `otel_collector.host` (default `127.0.0.1`), allowing an OTel Collector on another host. +To configure downstream agents, place an `ingress_port.yml` file in the job's config directory. [example][ingress-port] ### Syslog Agent An agent that forwards app logs to a syslog drain. Drains are registered by binding User Provided Services to apps. diff --git a/jobs/loggr-forwarder-agent-windows/spec b/jobs/loggr-forwarder-agent-windows/spec index e24c106e2..5389aeefe 100644 --- a/jobs/loggr-forwarder-agent-windows/spec +++ b/jobs/loggr-forwarder-agent-windows/spec @@ -23,13 +23,13 @@ properties: default: 3458 downstream_ingress_port_glob: description: | - Files matching the glob are expected to contain ports of downstream - consumers that will be bound to 127.0.0.1:{port} with the provided - mTLS configuration. The forwarder assumes the downstream server is - serving Loggregator's V2 IngressService. See code.cloudfoundry.org/loggregator-api. + Files matching the glob are expected to contain a port and protocol for + downstream consumers. Loggregator V2 destinations use 127.0.0.1:{port}. + OpenTelemetry Collector destinations (protocol: otelcol) use + otel_collector.host:{port}. See code.cloudfoundry.org/loggregator-api. default: /var/vcap/jobs/*/config/ingress_port.yml otel_collector.host: - description: "Host used for downstream OpenTelemetry Collector destinations" + description: "Host used with registration file ports for downstream OpenTelemetry Collector destinations" default: "127.0.0.1" deployment: diff --git a/jobs/loggr-forwarder-agent/spec b/jobs/loggr-forwarder-agent/spec index 0ef69adfb..fa4dbccec 100644 --- a/jobs/loggr-forwarder-agent/spec +++ b/jobs/loggr-forwarder-agent/spec @@ -23,13 +23,13 @@ properties: default: 3458 downstream_ingress_port_glob: description: | - Files matching the glob are expected to contain ports of downstream - consumers that will be bound to 127.0.0.1:{port} with the provided - mTLS configuration. The forwarder assumes the downstream server is - serving Loggregator's V2 IngressService. See code.cloudfoundry.org/loggregator-api. + Files matching the glob are expected to contain a port and protocol for + downstream consumers. Loggregator V2 destinations use 127.0.0.1:{port}. + OpenTelemetry Collector destinations (protocol: otelcol) use + otel_collector.host:{port}. See code.cloudfoundry.org/loggregator-api. default: /var/vcap/jobs/*/config/ingress_port.yml otel_collector.host: - description: "Host used for downstream OpenTelemetry Collector destinations" + description: "Host used with registration file ports for downstream OpenTelemetry Collector destinations" default: "127.0.0.1" deployment: From ff84161dd0b6887b78a41779fe38412c5820f28d Mon Sep 17 00:00:00 2001 From: Joris Baum Date: Mon, 19 Oct 2026 11:17:10 +0200 Subject: [PATCH 3/3] Make host configurable for all agents --- README.md | 4 +-- jobs/loggr-forwarder-agent-windows/monit | 1 - jobs/loggr-forwarder-agent-windows/spec | 10 ++---- jobs/loggr-forwarder-agent/spec | 10 ++---- .../templates/bpm.yml.erb | 1 - src/cmd/forwarder-agent/app/app_suite_test.go | 25 ++++++++++---- src/cmd/forwarder-agent/app/config.go | 7 ++-- src/cmd/forwarder-agent/app/config_test.go | 33 ------------------- .../forwarder-agent/app/forwarder_agent.go | 14 ++++---- .../app/forwarder_agent_test.go | 27 ++++++++++----- 10 files changed, 53 insertions(+), 79 deletions(-) delete mode 100644 src/cmd/forwarder-agent/app/config_test.go diff --git a/README.md b/README.md index 640f5f0eb..420872b50 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,8 @@ Accepts the Loggregator v1 api over UDP and forwards it to a downstream Loggrega This is available for backwards compatibility between the v1 API and the Forwarder, Syslog, or Loggregator agents. ### Forwarder Agent -A Loggregator v2 API multiplexer. Downstream registration files supply a port and protocol; destinations default to localhost. -`protocol: otelcol` destinations use `otel_collector.host` (default `127.0.0.1`), allowing an OTel Collector on another host. +A Loggregator v2 API multiplexer. Downstream registration files supply an ingress port and may specify a host and protocol. +The host defaults to `127.0.0.1`; set it in the registration file for remote or IPv6 destinations. To configure downstream agents, place an `ingress_port.yml` file in the job's config directory. [example][ingress-port] ### Syslog Agent diff --git a/jobs/loggr-forwarder-agent-windows/monit b/jobs/loggr-forwarder-agent-windows/monit index c45e5aab3..60c362071 100644 --- a/jobs/loggr-forwarder-agent-windows/monit +++ b/jobs/loggr-forwarder-agent-windows/monit @@ -25,7 +25,6 @@ "AGENT_CIPHER_SUITES" => p("tls.cipher_suites").split(":").join(","), "AGENT_TAGS" => tags.map { |k, v| "#{k}:#{v}" }.join(","), "DOWNSTREAM_INGRESS_PORT_GLOB" => p("downstream_ingress_port_glob"), - "OTEL_COLLECTOR_HOST" => p("otel_collector.host"), "EMIT_OTEL_TRACES" => "#{p("emit_otel_traces")}", "EMIT_OTEL_METRICS" => "#{p("emit_otel_metrics")}", "EMIT_OTEL_LOGS" => "#{p("emit_otel_logs")}", diff --git a/jobs/loggr-forwarder-agent-windows/spec b/jobs/loggr-forwarder-agent-windows/spec index 5389aeefe..235d6b50e 100644 --- a/jobs/loggr-forwarder-agent-windows/spec +++ b/jobs/loggr-forwarder-agent-windows/spec @@ -23,14 +23,10 @@ properties: default: 3458 downstream_ingress_port_glob: description: | - Files matching the glob are expected to contain a port and protocol for - downstream consumers. Loggregator V2 destinations use 127.0.0.1:{port}. - OpenTelemetry Collector destinations (protocol: otelcol) use - otel_collector.host:{port}. See code.cloudfoundry.org/loggregator-api. + Files matching the glob are expected to contain an ingress port and may + contain a host and protocol for downstream consumers. The host defaults + to 127.0.0.1. See code.cloudfoundry.org/loggregator-api. default: /var/vcap/jobs/*/config/ingress_port.yml - otel_collector.host: - description: "Host used with registration file ports for downstream OpenTelemetry Collector destinations" - default: "127.0.0.1" deployment: description: "Name of deployment (added as tag on all outgoing v1 envelopes)" diff --git a/jobs/loggr-forwarder-agent/spec b/jobs/loggr-forwarder-agent/spec index fa4dbccec..3b63b5c7d 100644 --- a/jobs/loggr-forwarder-agent/spec +++ b/jobs/loggr-forwarder-agent/spec @@ -23,14 +23,10 @@ properties: default: 3458 downstream_ingress_port_glob: description: | - Files matching the glob are expected to contain a port and protocol for - downstream consumers. Loggregator V2 destinations use 127.0.0.1:{port}. - OpenTelemetry Collector destinations (protocol: otelcol) use - otel_collector.host:{port}. See code.cloudfoundry.org/loggregator-api. + Files matching the glob are expected to contain an ingress port and may + contain a host and protocol for downstream consumers. The host defaults + to 127.0.0.1. See code.cloudfoundry.org/loggregator-api. default: /var/vcap/jobs/*/config/ingress_port.yml - otel_collector.host: - description: "Host used with registration file ports for downstream OpenTelemetry Collector destinations" - default: "127.0.0.1" deployment: description: "Name of deployment (added as tag on all outgoing v1 envelopes)" diff --git a/jobs/loggr-forwarder-agent/templates/bpm.yml.erb b/jobs/loggr-forwarder-agent/templates/bpm.yml.erb index 0db682a0f..9ce12267c 100644 --- a/jobs/loggr-forwarder-agent/templates/bpm.yml.erb +++ b/jobs/loggr-forwarder-agent/templates/bpm.yml.erb @@ -30,7 +30,6 @@ "AGENT_TAGS" => tags.map { |k, v| "#{k}:#{v}" }.join(","), "DOWNSTREAM_INGRESS_PORT_GLOB" => p("downstream_ingress_port_glob"), - "OTEL_COLLECTOR_HOST" => p("otel_collector.host"), "EMIT_OTEL_TRACES" => p("emit_otel_traces"), "EMIT_OTEL_METRICS" => p("emit_otel_metrics"), "EMIT_OTEL_LOGS" => p("emit_otel_logs"), diff --git a/src/cmd/forwarder-agent/app/app_suite_test.go b/src/cmd/forwarder-agent/app/app_suite_test.go index 416b0282f..75f253baf 100644 --- a/src/cmd/forwarder-agent/app/app_suite_test.go +++ b/src/cmd/forwarder-agent/app/app_suite_test.go @@ -95,7 +95,7 @@ type spyLoggregatorV2Ingress struct { envelopes chan *loggregator_v2.Envelope } -func startSpyLoggregatorV2Ingress(testCerts *testhelper.TestCerts, commonName string, cfgPath string) *spyLoggregatorV2Ingress { +func startSpyLoggregatorV2Ingress(testCerts *testhelper.TestCerts, commonName, cfgPath, host string) *spyLoggregatorV2Ingress { s := &spyLoggregatorV2Ingress{ envelopes: make(chan *loggregator_v2.Envelope, 10000), } @@ -107,7 +107,11 @@ func startSpyLoggregatorV2Ingress(testCerts *testhelper.TestCerts, commonName st ) ExpectWithOffset(1, err).ToNot(HaveOccurred()) - lis, err := net.Listen("tcp", "127.0.0.1:") + listenHost := host + if listenHost == "" { + listenHost = "127.0.0.1" + } + lis, err := net.Listen("tcp", net.JoinHostPort(listenHost, "0")) ExpectWithOffset(1, err).ToNot(HaveOccurred()) s.srv = grpc.NewServer(grpc.Creds(serverCreds), grpc.MaxRecvMsgSize(10*1024*1024)) @@ -119,13 +123,17 @@ func startSpyLoggregatorV2Ingress(testCerts *testhelper.TestCerts, commonName st } s.addr = lis.Addr().String() - port := strings.Split(s.addr, ":") + _, port, err := net.SplitHostPort(s.addr) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) dir, err := os.MkdirTemp(cfgPath, "") ExpectWithOffset(1, err).ToNot(HaveOccurred()) tmpfn := filepath.Join(dir, "ingress_port.yml") - contents := fmt.Sprintf(configTempl, port[len(port)-1]) + contents := fmt.Sprintf(configTempl, port) + if host != "" { + contents += fmt.Sprintf("host: %q\n", host) + } err = os.WriteFile(tmpfn, []byte(contents), 0600) ExpectWithOffset(1, err).ToNot(HaveOccurred()) @@ -256,7 +264,8 @@ func startSpyOtelColMetricServer(cfgPath string, tc *testhelper.TestCerts, commo contents := fmt.Sprintf(`--- ingress: %d protocol: otelcol -`, port) +host: %q +`, port, host) err = os.WriteFile(tmpfn, []byte(contents), 0600) ExpectWithOffset(1, err).ToNot(HaveOccurred()) @@ -318,7 +327,8 @@ func startSpyOtelColTraceServer(cfgPath string, tc *testhelper.TestCerts, common contents := fmt.Sprintf(`--- ingress: %d protocol: otelcol -`, port) +host: %q +`, port, host) err = os.WriteFile(tmpfn, []byte(contents), 0600) ExpectWithOffset(1, err).ToNot(HaveOccurred()) @@ -380,7 +390,8 @@ func startSpyOtelColLogServer(cfgPath string, tc *testhelper.TestCerts, commonNa contents := fmt.Sprintf(`--- ingress: %d protocol: otelcol -`, port) +host: %q +`, port, host) err = os.WriteFile(tmpfn, []byte(contents), 0600) ExpectWithOffset(1, err).ToNot(HaveOccurred()) diff --git a/src/cmd/forwarder-agent/app/config.go b/src/cmd/forwarder-agent/app/config.go index d3ffd0dcb..848abebcf 100644 --- a/src/cmd/forwarder-agent/app/config.go +++ b/src/cmd/forwarder-agent/app/config.go @@ -28,11 +28,9 @@ type OtelRetry struct { // Config holds the configuration for the forwarder agent type Config struct { UseRFC3339 bool `env:"USE_RFC3339"` - // DownstreamIngressPortCfg will define consumers on localhost that will - // receive each envelope. It is assumed to adhere to the Loggregator Ingress - // Service and use the provided TLS configuration. + // DownstreamIngressPortCfg defines consumers that receive each envelope. + // They use the provided TLS configuration. DownstreamIngressPortCfg string `env:"DOWNSTREAM_INGRESS_PORT_GLOB, report"` - OTelCollectorHost string `env:"OTEL_COLLECTOR_HOST, report"` GRPC GRPC MetricsServer config.MetricsServer Tags map[string]string `env:"AGENT_TAGS"` @@ -52,7 +50,6 @@ func LoadConfig() Config { Host: "127.0.0.1", Port: 3458, }, - OTelCollectorHost: "127.0.0.1", OtelRetry: OtelRetry{ MaxRetries: 7, RetryQueueSize: 1024, diff --git a/src/cmd/forwarder-agent/app/config_test.go b/src/cmd/forwarder-agent/app/config_test.go deleted file mode 100644 index 35fa73abd..000000000 --- a/src/cmd/forwarder-agent/app/config_test.go +++ /dev/null @@ -1,33 +0,0 @@ -package app_test - -import ( - "os" - - "code.cloudfoundry.org/loggregator-agent-release/src/cmd/forwarder-agent/app" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" -) - -var _ = Describe("Config", func() { - BeforeEach(func() { - os.Setenv("AGENT_CA_FILE_PATH", "ca.crt") - os.Setenv("AGENT_CERT_FILE_PATH", "agent.crt") - os.Setenv("AGENT_KEY_FILE_PATH", "agent.key") - os.Setenv("METRICS_CA_FILE_PATH", "metrics-ca.crt") - os.Setenv("METRICS_CERT_FILE_PATH", "metrics.crt") - os.Setenv("METRICS_KEY_FILE_PATH", "metrics.key") - }) - - It("defaults the OTel Collector host to localhost", func() { - os.Setenv("OTEL_COLLECTOR_HOST", "") - - Expect(app.LoadConfig().OTelCollectorHost).To(Equal("127.0.0.1")) - }) - - It("configures the OTel Collector host", func() { - os.Setenv("OTEL_COLLECTOR_HOST", "otel-collector") - - Expect(app.LoadConfig().OTelCollectorHost).To(Equal("otel-collector")) - }) -}) diff --git a/src/cmd/forwarder-agent/app/forwarder_agent.go b/src/cmd/forwarder-agent/app/forwarder_agent.go index 85cfcf3f8..24f012e6e 100644 --- a/src/cmd/forwarder-agent/app/forwarder_agent.go +++ b/src/cmd/forwarder-agent/app/forwarder_agent.go @@ -39,7 +39,6 @@ type ForwarderAgent struct { grpc GRPC v2srv *v2.Server downstreamFilePattern string - otelCollectorHost string log *log.Logger tags map[string]string debugMetrics bool @@ -74,7 +73,6 @@ func NewForwarderAgent( grpc: cfg.GRPC, m: m, downstreamFilePattern: cfg.DownstreamIngressPortCfg, - otelCollectorHost: cfg.OTelCollectorHost, log: log, tags: cfg.Tags, debugMetrics: cfg.MetricsServer.DebugMetrics, @@ -104,7 +102,7 @@ func (s *ForwarderAgent) Run() { ingressDropped.Add(float64(missed)) })) - dests := downstreamDestinations(s.downstreamFilePattern, s.otelCollectorHost, s.log) + dests := downstreamDestinations(s.downstreamFilePattern, s.log) writers := downstreamWriters(dests, s.grpc, s.m, s.emitOTelTraces, s.emitOTelMetrics, s.emitOTelLogs, s.otelRetry, s.log) tagger := egress_v2.NewTagger(s.tags) ew := egress_v2.NewEnvelopeWriter( @@ -186,9 +184,10 @@ func (mw multiWriter) Write(e *loggregator_v2.Envelope) error { type destination struct { Ingress string `yaml:"ingress"` Protocol string `yaml:"protocol"` + Host string `yaml:"host"` } -func downstreamDestinations(pattern, otelCollectorHost string, l *log.Logger) []destination { +func downstreamDestinations(pattern string, l *log.Logger) []destination { files, err := filepath.Glob(pattern) if err != nil { l.Fatal("Unable to read downstream port location") @@ -210,11 +209,10 @@ func downstreamDestinations(pattern, otelCollectorHost string, l *log.Logger) [] if d.Ingress == "" { l.Printf("No ingress port defined in %s. Ignoring this destination.", f) } else { - if d.Protocol == "otelcol" { - d.Ingress = net.JoinHostPort(otelCollectorHost, d.Ingress) - } else { - d.Ingress = net.JoinHostPort("127.0.0.1", d.Ingress) + if d.Host == "" { + d.Host = "127.0.0.1" } + d.Ingress = net.JoinHostPort(d.Host, d.Ingress) dests = append(dests, d) } } diff --git a/src/cmd/forwarder-agent/app/forwarder_agent_test.go b/src/cmd/forwarder-agent/app/forwarder_agent_test.go index 26bab20e0..c72596cde 100644 --- a/src/cmd/forwarder-agent/app/forwarder_agent_test.go +++ b/src/cmd/forwarder-agent/app/forwarder_agent_test.go @@ -57,9 +57,9 @@ var _ = Describe("App", func() { ingressCfgPath = GinkgoT().TempDir() ingressClient = newIngressClient(grpcPort, agentCerts, 1) - ingressServer1 = startSpyLoggregatorV2Ingress(agentCerts, agentCN, ingressCfgPath) - ingressServer2 = startSpyLoggregatorV2Ingress(agentCerts, agentCN, ingressCfgPath) - ingressServer3 = startSpyLoggregatorV2Ingress(agentCerts, agentCN, ingressCfgPath) + ingressServer1 = startSpyLoggregatorV2Ingress(agentCerts, agentCN, ingressCfgPath, "") + ingressServer2 = startSpyLoggregatorV2Ingress(agentCerts, agentCN, ingressCfgPath, "") + ingressServer3 = startSpyLoggregatorV2Ingress(agentCerts, agentCN, ingressCfgPath, "::1") ingressServer3.blocking = true agentCfg = app.Config{ @@ -70,7 +70,6 @@ var _ = Describe("App", func() { KeyFile: agentCerts.Key(agentCN), }, DownstreamIngressPortCfg: fmt.Sprintf("%s/*/ingress_port.yml", ingressCfgPath), - OTelCollectorHost: "127.0.0.1", MetricsServer: config.MetricsServer{ Port: uint16(metricsPort), CAFile: agentCerts.CA(), @@ -201,6 +200,18 @@ var _ = Describe("App", func() { Eventually(ingressServer2.envelopes, 5).Should(Receive(protoEqual(sampleEnvelope))) }) + It("defaults a missing destination host to localhost", func() { + ingressClient.Emit(sampleEnvelope) + + Eventually(ingressServer1.envelopes, 5).Should(Receive(protoEqual(sampleEnvelope))) + }) + + It("forwards to an IPv6 host from a registration file", func() { + ingressClient.Emit(sampleEnvelope) + + Eventually(ingressServer3.envelopes, 5).Should(Receive(protoEqual(sampleEnvelope))) + }) + It("can send a batch of 100, max-size (for Diego) messages downstream", func() { ctx, cancel := context.WithCancel(context.Background()) var wg sync.WaitGroup @@ -359,10 +370,10 @@ var _ = Describe("App", func() { ) BeforeEach(func() { - agentCfg.OTelCollectorHost = "::1" - otelMetricsServer = startSpyOtelColMetricServer(ingressCfgPath, agentCerts, "otel-collector", agentCfg.OTelCollectorHost) - otelTraceServer = startSpyOtelColTraceServer(ingressCfgPath, agentCerts, "otel-collector", agentCfg.OTelCollectorHost) - otelLogsServer = startSpyOtelColLogServer(ingressCfgPath, agentCerts, "otel-collector", agentCfg.OTelCollectorHost) + host := "::1" + otelMetricsServer = startSpyOtelColMetricServer(ingressCfgPath, agentCerts, "otel-collector", host) + otelTraceServer = startSpyOtelColTraceServer(ingressCfgPath, agentCerts, "otel-collector", host) + otelLogsServer = startSpyOtelColLogServer(ingressCfgPath, agentCerts, "otel-collector", host) agentCfg.EmitOTelTraces = true agentCfg.EmitOTelMetrics = true agentCfg.EmitOTelLogs = true