fix(metrics): collapse per-IP destination labels to workload identity to cut TSDB cardinality

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 {

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

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

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

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

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