diff --git a/common/net.go b/common/net.go index 5a8a946..f62d716 100644 --- a/common/net.go +++ b/common/net.go @@ -196,11 +196,69 @@ func (dk DestinationKey) ActualDestinationIfKnown() HostPort { } func (dk DestinationKey) DestinationLabelValue() string { - return dk.destination.String() + return destinationLabelValue(dk.destination, dk.destinationWorkload) } func (dk DestinationKey) ActualDestinationLabelValue() string { - return dk.actualDestination.String() + return destinationLabelValue(dk.actualDestination, dk.actualDestinationWorkload) +} + +// destinationLabelValue produces the destination/actual_destination label value. +// +// External destinations resolved to an FQDN (host set, no IP) are returned as-is +// — they are already bounded and meaningful. +// +// Internal destinations carry a raw IP:port. Pod IPs churn and recycle +// constantly, so emitting IP:port creates a brand-new series on every reconnect, +// which is the dominant driver of TSDB churn/cardinality. When +// CollapseInternalDestinations is enabled (default), we substitute the resolved +// workload identity (namespace/name), which stays stable across pod-IP changes. +// The *_workload_* labels already carry this identity, and backend queries key on +// them rather than the raw destination, so this is transparent to consumers. +// When the workload is resolved to a real name (not just the IP echoed back), +// we use its identity. Otherwise we fall back per destination type: private +// (internal) IPs drop the churning port dimension; external IPs keep IP:port +// since their cardinality is low and the port is useful. +func destinationLabelValue(hp HostPort, wl Workload) string { + // FQDN destinations (external, resolved by DNS) have no IP set — keep them. + if hp.ip.IsZero() { + return hp.String() + } + if flags.CollapseInternalDestinations == nil || !*flags.CollapseInternalDestinations { + return hp.String() + } + // Resolved workload identity (guard against the IP-echoed-back fallback that + // ResolveIP returns for unresolved endpoints). + if wl.Name != "" && wl.Name != hp.ip.String() { + if wl.Namespace != "" { + return wl.Namespace + "/" + wl.Name + } + return wl.Name + } + if IsIpPrivate(hp.ip) { + // Unresolved internal endpoint: drop the churning port dimension. + return hp.ip.String() + } + // Unresolved external endpoint: low cardinality, keep the port. + return hp.String() +} + +// DestinationIPLabelValue collapses a raw destination IP to the resolved workload +// identity for the container_net_latency_seconds destination_ip label, mirroring +// destinationLabelValue. The pinger works with bare IPs (no port), so this variant +// takes a netaddr.IP. Internal pod IPs churn/recycle; keying the RTT series on +// workload identity keeps it stable. External and unresolved IPs are kept as-is. +func DestinationIPLabelValue(ip netaddr.IP, wl Workload) string { + if flags.CollapseInternalDestinations == nil || !*flags.CollapseInternalDestinations { + return ip.String() + } + if IsIpPrivate(ip) && wl.Name != "" && wl.Name != ip.String() { + if wl.Namespace != "" { + return wl.Namespace + "/" + wl.Name + } + return wl.Name + } + return ip.String() } func (dk DestinationKey) String() string { diff --git a/common/net_test.go b/common/net_test.go index 7db18bd..cf1bb25 100644 --- a/common/net_test.go +++ b/common/net_test.go @@ -3,10 +3,23 @@ package common import ( "testing" + "github.com/coroot/coroot-node-agent/flags" "github.com/stretchr/testify/assert" "inet.af/netaddr" ) +// setCollapse sets the CollapseInternalDestinations flag for the duration of a +// test and restores the previous value afterwards. +func setCollapse(t *testing.T, v bool) { + prev := *flags.CollapseInternalDestinations + *flags.CollapseInternalDestinations = v + t.Cleanup(func() { *flags.CollapseInternalDestinations = prev }) +} + +func internalHP(ip string, port uint16) HostPort { + return HostPortFromIPPort(netaddr.IPPortFrom(netaddr.MustParseIP(ip), port)) +} + func TestConnectionFilter(t *testing.T) { f := connectionFilter{whitelist: map[string]netaddr.IPPrefix{}} assert.False(t, f.ShouldBeSkipped(netaddr.MustParseIP("127.0.0.1"), netaddr.MustParseIP("127.0.0.1"))) @@ -73,6 +86,63 @@ func TestNormalizeFQDN(t *testing.T) { assert.Equal(t, "example.io.search_path_suffix", NormalizeFQDN("example.io.svc.default.cluster.local", "TypeA")) } +func TestDestinationLabelValue(t *testing.T) { + fqdn := HostPortWithEmptyIP("api.openai.com", 443) + internal := internalHP("10.64.3.17", 8080) + external := internalHP("1.1.1.1", 443) + resolved := Workload{Name: "api-server", Namespace: "nudgebee", Kind: "Deployment"} + noNamespace := Workload{Name: "kube-dns"} + unresolved := Workload{} // ResolveIP fell through, no name + ipEcho := Workload{Name: "10.64.3.17", Namespace: "external", Kind: "external"} // ResolveIP echoed the IP back + + t.Run("collapse on", func(t *testing.T) { + setCollapse(t, true) + // External FQDN destinations are kept as-is regardless of workload. + assert.Equal(t, "api.openai.com:443", destinationLabelValue(fqdn, unresolved)) + // Internal resolved -> stable workload identity (no churning IP:port). + assert.Equal(t, "nudgebee/api-server", destinationLabelValue(internal, resolved)) + // Internal resolved without namespace -> bare name. + assert.Equal(t, "kube-dns", destinationLabelValue(internal, noNamespace)) + // Internal unresolved (no name) -> bare IP, port dimension dropped. + assert.Equal(t, "10.64.3.17", destinationLabelValue(internal, unresolved)) + // Internal unresolved where ResolveIP echoed the IP back as the name -> + // treated as unresolved (not "external/10.64.3.17"), port dropped. + assert.Equal(t, "10.64.3.17", destinationLabelValue(internal, ipEcho)) + // External unresolved -> low cardinality, keep the port. + assert.Equal(t, "1.1.1.1:443", destinationLabelValue(external, unresolved)) + }) + + t.Run("collapse off", func(t *testing.T) { + setCollapse(t, false) + // Legacy behaviour: raw IP:port for internal, FQDN untouched. + assert.Equal(t, "10.64.3.17:8080", destinationLabelValue(internal, resolved)) + assert.Equal(t, "api.openai.com:443", destinationLabelValue(fqdn, resolved)) + }) +} + +func TestDestinationIPLabelValue(t *testing.T) { + private := netaddr.MustParseIP("10.64.3.17") + public := netaddr.MustParseIP("1.1.1.1") + resolved := Workload{Name: "api-server", Namespace: "nudgebee", Kind: "Deployment"} + + t.Run("collapse on", func(t *testing.T) { + setCollapse(t, true) + // Private + resolved -> workload identity. + assert.Equal(t, "nudgebee/api-server", DestinationIPLabelValue(private, resolved)) + // Private but unresolved (workload name == the IP) -> keep the IP. + assert.Equal(t, "10.64.3.17", DestinationIPLabelValue(private, Workload{Name: "10.64.3.17"})) + // Private with empty workload -> keep the IP. + assert.Equal(t, "10.64.3.17", DestinationIPLabelValue(private, Workload{})) + // Public IP -> never collapsed. + assert.Equal(t, "1.1.1.1", DestinationIPLabelValue(public, Workload{Name: "api.openai.com", Namespace: "external"})) + }) + + t.Run("collapse off", func(t *testing.T) { + setCollapse(t, false) + assert.Equal(t, "10.64.3.17", DestinationIPLabelValue(private, resolved)) + }) +} + func BenchmarkNormalizeFQDN(b *testing.B) { for i := 0; i < b.N; i++ { NormalizeFQDN("ip-172-1-2-3.ec2.internal", "TypeA") diff --git a/containers/container.go b/containers/container.go index 14aa858..024deec 100644 --- a/containers/container.go +++ b/containers/container.go @@ -505,7 +505,8 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { if !*flags.DisablePinger { for ip, rtt := range c.ping() { destination_workload := c.ip_resolver.ResolveIP(ip.String()) - ch <- c.gauge(metrics.NetLatency, rtt, ip.String(), destination_workload.Name, destination_workload.Namespace, destination_workload.Kind) + destIP := common.DestinationIPLabelValue(ip, destination_workload) + ch <- c.gauge(metrics.NetLatency, rtt, destIP, destination_workload.Name, destination_workload.Namespace, destination_workload.Kind) } } diff --git a/containers/metrics.go b/containers/metrics.go index 5739bd2..88715ae 100644 --- a/containers/metrics.go +++ b/containers/metrics.go @@ -37,15 +37,11 @@ var metrics = struct { DiskWriteOps *prometheus.Desc DiskWriteBytes *prometheus.Desc - NetListenInfo *prometheus.Desc - NetConnectionsSuccessful *prometheus.Desc - NetConnectionsTotalTime *prometheus.Desc - NetConnectionsFailed *prometheus.Desc - NetConnectionsActive *prometheus.Desc - NetRetransmits *prometheus.Desc - NetLatency *prometheus.Desc - NetBytesSent *prometheus.Desc - NetBytesReceived *prometheus.Desc + NetListenInfo *prometheus.Desc + NetLatency *prometheus.Desc + // container_net_tcp_* connection/byte metrics are emitted directly from the + // pre-registered CounterVecs in tcp_metrics.go (TCPMetrics.collect); the Desc + // forms previously declared here were dead code and have been removed. LogMessages *prometheus.Desc SensitiveLogMessages *prometheus.Desc @@ -94,15 +90,8 @@ var metrics = struct { DiskWriteOps: metric("container_resources_disk_writes_total", "Total number of writes completed successfully by the container", "mount_point", "device", "volume"), DiskWriteBytes: metric("container_resources_disk_written_bytes_total", "Total number of bytes written to the disk by the container", "mount_point", "device", "volume"), - NetListenInfo: metric("container_net_tcp_listen_info", "Listen address of the container", "listen_addr", "proxy"), - NetConnectionsSuccessful: metric("container_net_tcp_successful_connects_total", "Total number of successful TCP connects", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), - NetConnectionsTotalTime: metric("container_net_tcp_connection_time_seconds_total", "Time spent on TCP connections", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), - NetConnectionsFailed: metric("container_net_tcp_failed_connects_total", "Total number of failed TCP connects", "destination", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), - NetConnectionsActive: metric("container_net_tcp_active_connections", "Number of active outbound connections used by the container", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), - NetRetransmits: metric("container_net_tcp_retransmits_total", "Total number of retransmitted TCP segments", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), - NetLatency: metric("container_net_latency_seconds", "Round-trip time between the container and a remote IP", "destination_ip", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind"), - NetBytesSent: metric("container_net_tcp_bytes_sent_total", "Total number of bytes sent to the peer", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), - NetBytesReceived: metric("container_net_tcp_bytes_received_total", "Total number of bytes received from the peer", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), + NetListenInfo: metric("container_net_tcp_listen_info", "Listen address of the container", "listen_addr", "proxy"), + NetLatency: metric("container_net_latency_seconds", "Round-trip time between the container and a remote IP", "destination_ip", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind"), LogMessages: metric("container_log_messages_total", "Number of messages grouped by the automatically extracted repeated pattern", "source", "level", "pattern_hash", "sample"), SensitiveLogMessages: metric("container_sensitive_log_messages_total", "Number of messages that contain sensitive information", "source", "pattern", "sample", "regex", "name", "pattern_hash"), diff --git a/containers/tcp_metrics.go b/containers/tcp_metrics.go index fb955fd..c62ac96 100644 --- a/containers/tcp_metrics.go +++ b/containers/tcp_metrics.go @@ -166,11 +166,22 @@ func (t *TCPMetrics) ObserveTraffic(key common.DestinationKey, src common.Worklo // resetAndSetActive replaces all active connection gauge values. // Called from the event handler goroutine periodically. +// +// Uses Add (not Set) after Reset so that multiple entries which collapse to the +// same label set — e.g. connections to several pods of one workload once +// destination labels are collapsed to workload identity (CollapseInternalDestinations) +// — sum instead of overwriting each other. With unique labels Add-from-zero is +// equivalent to Set. func (t *TCPMetrics) resetAndSetActive(entries []activeEntry) { t.ensureInitialized() + // Hold the write lock across Reset + rebuild so a concurrent collect() + // (which holds RLock) never observes a partially-rebuilt or empty gauge, + // which would show up as transient dips in container_net_tcp_active_connections. + t.mu.Lock() + defer t.mu.Unlock() t.active.Reset() for _, e := range entries { - t.active.WithLabelValues(e.labels...).Set(float64(e.count)) + t.active.WithLabelValues(e.labels...).Add(float64(e.count)) } } diff --git a/flags/flags.go b/flags/flags.go index 897c840..9533154 100644 --- a/flags/flags.go +++ b/flags/flags.go @@ -71,6 +71,16 @@ var ( AggregateEphemeralWorkloads = kingpin.Flag("aggregate-ephemeral-workloads", "Aggregate metrics for bare pods and standalone Jobs using standard labels to reduce series cardinality").Default("true").Envar("AGGREGATE_EPHEMERAL_WORKLOADS").Bool() + // CollapseInternalDestinations replaces the raw IP:port value of the + // destination/actual_destination labels with the resolved workload identity + // (namespace/name) for internal (non-FQDN) destinations. Pod IPs churn and + // recycle constantly (especially on spot nodes), so raw IP:port creates a new + // series on every reconnect — the dominant source of TSDB churn. Collapsing to + // workload identity keeps series stable across pod-IP changes. External FQDN + // destinations are unaffected. Backend queries key on the *_workload_* labels, + // not the raw destination, so this is transparent to consumers. + CollapseInternalDestinations = kingpin.Flag("collapse-internal-destinations", "Use workload identity instead of raw IP:port for internal destination/actual_destination labels to reduce series cardinality").Default("true").Envar("COLLAPSE_INTERNAL_DESTINATIONS").Bool() + agentVersion = kingpin.Flag("version", "Print version and exit").Default("false").Bool() Version = "unknown" )