Make downstream otlp host configurable

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. 

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")}",

jobs/loggr-forwarder-agent-windows/spec

@@ -23,11 +23,14 @@ 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 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)"

jobs/loggr-forwarder-agent/spec

@@ -23,11 +23,14 @@ 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 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)"

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"),

src/cmd/forwarder-agent/app/app_suite_test.go

@@ -224,15 +224,15 @@ 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),
 		tc.CA(),
 	)
 	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,15 +286,15 @@ 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),
 		tc.CA(),
 	)
 	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,15 +348,15 @@ 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),
 		tc.CA(),
 	)
 	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, "")

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,

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"))
+	})
+})

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)
 		}
 	}

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