diff --git a/common/ip_resolver.go b/common/ip_resolver.go index a09e05a..19701ef 100644 --- a/common/ip_resolver.go +++ b/common/ip_resolver.go @@ -2,8 +2,11 @@ package common import ( "context" + "errors" "fmt" "log" + "regexp" + "strings" "sync" "time" @@ -804,6 +807,20 @@ func getControllerOwnerRef(refs []metav1.OwnerReference) *metav1.OwnerReference return nil } +// errOwnerNotCached means the owner object exists in the cluster but is not in +// our informer snapshot yet (informers still syncing), or was garbage-collected +// while its pods linger. This is TRANSIENT: the same lookup may succeed later, +// so callers must not memoize an identity derived from this failure. +// +// errUnsupportedOwnerKind means the owner is a kind we do not track (e.g. a +// custom controller like Argo Rollouts). This is TERMINAL: retrying will never +// help, so the identity we already have is the best available and is safe to +// cache. +var ( + errOwnerNotCached = errors.New("owner not in informer cache") + errUnsupportedOwnerKind = errors.New("unsupported owner kind") +) + func (resolver *K8sIPResolver) getControllerOfOwner(owner *metav1.OwnerReference) (*metav1.OwnerReference, error) { var m *sync.Map switch owner.Kind { @@ -820,16 +837,42 @@ func (resolver *K8sIPResolver) getControllerOfOwner(owner *metav1.OwnerReference case "CronJob": m = &resolver.snapshot.CronJobs default: - return nil, fmt.Errorf("unsupported kind: %s", owner.Kind) + return nil, fmt.Errorf("%w: %s", errUnsupportedOwnerKind, owner.Kind) } val, ok := m.Load(owner.UID) if !ok { - return nil, fmt.Errorf("missing %s for UID %s", owner.Kind, owner.UID) + return nil, fmt.Errorf("%w: %s %s", errOwnerNotCached, owner.Kind, owner.UID) } info := val.(MinimalOwnerInfo) return getControllerOwnerRef(info.OwnerReferences), nil } +// podTemplateHashRe matches the suffix the Deployment controller appends when +// it names a ReplicaSet: the pod-template-hash. Kubernetes encodes that hash +// with an alphanumeric alphabet that excludes vowels (to avoid generating +// words), giving 6-11 character segments in practice. +var podTemplateHashRe = regexp.MustCompile(`^[bcdfghjklmnpqrstvwxz2456789]{6,11}$`) + +// stripPodTemplateHash turns a Deployment-generated ReplicaSet name +// ("llm-server-779f867dc9") into its Deployment name ("llm-server"). +// +// It returns ok=false when the name does not carry a pod-template-hash suffix +// — a ReplicaSet created directly rather than by a Deployment ("my-app") — so +// that a bare ReplicaSet is never silently renamed to a Deployment that does +// not exist. A Deployment whose own name ends in a hash-shaped segment is +// unaffected: only the final segment is removed, e.g. +// "foo-2456789-7d9f8bcdfg" -> "foo-2456789". +func stripPodTemplateHash(name string) (string, bool) { + i := strings.LastIndex(name, "-") + if i <= 0 || i == len(name)-1 { + return name, false + } + if !podTemplateHashRe.MatchString(name[i+1:]) { + return name, false + } + return name[:i], true +} + // workloadIdentityLabels defines the priority order for finding a stable // workload identity from pod labels when no higher-level controller exists. var workloadIdentityLabels = []string{ @@ -859,32 +902,63 @@ func (resolver *K8sIPResolver) resolvePodDescriptor(pod *MinimalPod) Workload { return result } } - var err error name := pod.Name namespace := pod.Namespace kind := "pod" + // cacheable stays true only while every step of the ownership climb either + // succeeded or failed terminally. A TRANSIENT failure (owner not yet in the + // informer cache) must not be memoized: doing so pins the pod to an + // intermediate identity — typically the ReplicaSet — for its entire + // lifetime, which is what produced duplicate `` / `-` + // series with kind=Deployment and kind=ReplicaSet for the same workload. + cacheable := true + // Resolve owner hierarchy using OwnerReferences from MinimalPod if len(pod.OwnerReferences) > 0 { // Find the controller owner reference - for _, owner := range pod.OwnerReferences { - if owner.Controller != nil && *owner.Controller { - name = owner.Name - kind = owner.Kind - // Try to climb up the ownership hierarchy - if owner, err := resolver.getControllerOfOwner(&owner); err == nil && owner != nil { - for owner != nil { - name = owner.Name - kind = owner.Kind - owner, err = resolver.getControllerOfOwner(owner) - if err != nil { - klog.V(5).Infof("couldn't retrieve owner of %v - %v", name, err) - break - } + for _, ownerRef := range pod.OwnerReferences { + if ownerRef.Controller == nil || !*ownerRef.Controller { + continue + } + name = ownerRef.Name + kind = ownerRef.Kind + // Climb to the top of the ownership chain (e.g. Pod -> ReplicaSet -> + // Deployment). NOTE: `current` and `err` are deliberately declared in + // this scope. The previous implementation used `:=` inside an `if` + // statement, which shadowed the outer error and left it permanently + // nil, so a failed climb was still treated as a success and cached. + current := ownerRef.DeepCopy() + for { + next, err := resolver.getControllerOfOwner(current) + if err != nil { + if errors.Is(err, errOwnerNotCached) { + cacheable = false } + klog.V(5).Infof("couldn't retrieve owner of %s/%s (%s): %v", namespace, name, kind, err) + break + } + if next == nil { + // Reached the top of the chain — this is the real workload. + break } - break + name = next.Name + kind = next.Kind + current = next } + break + } + } + + // Fallback for an unresolvable ReplicaSet: its owning Deployment was not in + // the cache, so derive the Deployment name from the ReplicaSet's pod-template + // hash suffix rather than publishing a name that churns on every rollout. + // Only applied to the `-` form that Deployments generate; a bare + // ReplicaSet (created directly, no hash suffix) is left untouched. + if kind == "ReplicaSet" { + if stripped, ok := stripPodTemplateHash(name); ok { + name = stripped + kind = "Deployment" } } @@ -922,7 +996,10 @@ func (resolver *K8sIPResolver) resolvePodDescriptor(pod *MinimalPod) Workload { Zone: zone, Instance: pod.NodeName, } - if err == nil { + // Only memoize a fully-resolved identity. If an owner was missing from the + // informer cache we return the best-effort value for this scrape but leave + // the descriptor uncached, so the next call retries once informers sync. + if cacheable { resolver.snapshot.PodDescriptors.Store(pod.UID, result) } return result @@ -942,10 +1019,17 @@ func (resolver *K8sIPResolver) ResolvePodOwner(podName string, podNamespace stri return Workload{ Name: podName, Namespace: podNamespace, - Kind: "Pod", - Region: "", - Zone: "", - Instance: "", + // Lowercase "pod" to match resolvePodDescriptor's default. This path + // previously returned "Pod", so an unowned pod got a different kind + // depending on which code path resolved it, splitting one workload + // across two series. The lowercase spelling is the one already + // present in the TSDB, and it keeps the convention that non-k8s-owner + // sentinels ("pod", "node", "external") are lowercase while real + // owner Kinds ("Deployment", "StatefulSet") keep their k8s casing. + Kind: "pod", + Region: "", + Zone: "", + Instance: "", } } minPod := &MinimalPod{ diff --git a/common/ip_resolver_workload_identity_test.go b/common/ip_resolver_workload_identity_test.go new file mode 100644 index 0000000..09c618c --- /dev/null +++ b/common/ip_resolver_workload_identity_test.go @@ -0,0 +1,200 @@ +package common + +import ( + "testing" + + "github.com/coroot/coroot-node-agent/flags" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +// disableEphemeralAggregation pins AggregateEphemeralWorkloads off so these +// tests exercise the owner-resolution path only. +func disableEphemeralAggregation(t *testing.T) { + t.Helper() + orig := flags.AggregateEphemeralWorkloads + disabled := false + flags.AggregateEphemeralWorkloads = &disabled + t.Cleanup(func() { flags.AggregateEphemeralWorkloads = orig }) +} + +func controllerRef(name string, uid types.UID, kind string) metav1.OwnerReference { + yes := true + return metav1.OwnerReference{Name: name, UID: uid, Kind: kind, Controller: &yes} +} + +func TestStripPodTemplateHash(t *testing.T) { + tests := []struct { + name string + in string + wantName string + wantOk bool + }{ + // Deployment-generated ReplicaSet names — the case that caused the + // duplicate llm-server / llm-server-779f867dc9 series. + {"real replicaset name", "llm-server-779f867dc9", "llm-server", true}, + {"second revision", "llm-server-78c94f7dfd", "llm-server", true}, + {"shorter hash", "services-server-6f476f6f79", "services-server", true}, + {"hyphenated deployment", "my-long-app-name-5b8d9fcc7", "my-long-app-name", true}, + + // Must NOT be rewritten: a bare ReplicaSet has no pod-template-hash, so + // stripping would invent a Deployment that does not exist. + {"bare replicaset, no suffix", "replicaset1", "replicaset1", false}, + {"no hyphen at all", "myapp", "myapp", false}, + {"suffix contains vowels", "my-app-frontend", "my-app-frontend", false}, + {"suffix too short", "my-app-abc", "my-app-abc", false}, + {"suffix too long", "my-app-bcdfghjklmnpqrstvwxz", "my-app-bcdfghjklmnpqrstvwxz", false}, + {"trailing hyphen", "my-app-", "my-app-", false}, + {"leading hyphen only", "-bcdfgh", "-bcdfgh", false}, + {"uppercase not a k8s hash", "my-app-ABCDEF", "my-app-ABCDEF", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := stripPodTemplateHash(tt.in) + assert.Equal(t, tt.wantOk, ok) + assert.Equal(t, tt.wantName, got) + }) + } +} + +// A pod whose ReplicaSet is already cached must resolve all the way to the +// Deployment, and that result is safe to memoize. +func TestResolvePodDescriptor_FullChainIsCached(t *testing.T) { + disableEphemeralAggregation(t) + + rsUID := types.UID(uuid.NewString()) + deployUID := types.UID(uuid.NewString()) + podUID := types.UID(uuid.NewString()) + + r := &K8sIPResolver{} + r.snapshot.ReplicaSets.Store(rsUID, MinimalOwnerInfo{ + OwnerReferences: []metav1.OwnerReference{controllerRef("llm-server", deployUID, "Deployment")}, + }) + r.snapshot.Deployments.Store(deployUID, MinimalOwnerInfo{}) + + pod := &MinimalPod{ + UID: podUID, + Name: "llm-server-779f867dc9-abcde", + Namespace: "nudgebee", + OwnerReferences: []metav1.OwnerReference{controllerRef("llm-server-779f867dc9", rsUID, "ReplicaSet")}, + } + + got := r.resolvePodDescriptor(pod) + assert.Equal(t, "llm-server", got.Name) + assert.Equal(t, "Deployment", got.Kind) + assert.Equal(t, "nudgebee", got.Namespace) + + _, cached := r.snapshot.PodDescriptors.Load(podUID) + assert.True(t, cached, "a fully-resolved chain should be memoized") +} + +// Regression for the primary bug: when the owning ReplicaSet is missing from +// the informer snapshot the climb cannot complete, so the result must NOT be +// memoized. Previously a shadowed `err` left the outer error nil, the partial +// (ReplicaSet) identity was cached, and the pod kept that identity forever. +func TestResolvePodDescriptor_TransientMissIsNotCached(t *testing.T) { + disableEphemeralAggregation(t) + + rsUID := types.UID(uuid.NewString()) + podUID := types.UID(uuid.NewString()) + + r := &K8sIPResolver{} // ReplicaSet deliberately absent from the snapshot + + pod := &MinimalPod{ + UID: podUID, + Name: "llm-server-779f867dc9-abcde", + Namespace: "nudgebee", + OwnerReferences: []metav1.OwnerReference{controllerRef("llm-server-779f867dc9", rsUID, "ReplicaSet")}, + } + + r.resolvePodDescriptor(pod) + + _, cached := r.snapshot.PodDescriptors.Load(podUID) + assert.False(t, cached, "an unresolved owner chain must not be memoized") +} + +// The whole point of not caching: once informers sync, the very next call must +// return the Deployment rather than a stale ReplicaSet identity. +func TestResolvePodDescriptor_RecoversAfterInformerSync(t *testing.T) { + disableEphemeralAggregation(t) + + rsUID := types.UID(uuid.NewString()) + deployUID := types.UID(uuid.NewString()) + podUID := types.UID(uuid.NewString()) + + r := &K8sIPResolver{} + pod := &MinimalPod{ + UID: podUID, + Name: "llm-server-779f867dc9-abcde", + Namespace: "nudgebee", + OwnerReferences: []metav1.OwnerReference{controllerRef("llm-server-779f867dc9", rsUID, "ReplicaSet")}, + } + + // First call: ReplicaSet not cached yet. Falls back to the hash-stripped + // Deployment name and must not poison the cache. + first := r.resolvePodDescriptor(pod) + assert.Equal(t, "llm-server", first.Name, "fallback should strip the pod-template-hash") + + // Informers catch up. + r.snapshot.ReplicaSets.Store(rsUID, MinimalOwnerInfo{ + OwnerReferences: []metav1.OwnerReference{controllerRef("llm-server", deployUID, "Deployment")}, + }) + r.snapshot.Deployments.Store(deployUID, MinimalOwnerInfo{}) + + second := r.resolvePodDescriptor(pod) + assert.Equal(t, "llm-server", second.Name) + assert.Equal(t, "Deployment", second.Kind, "must resolve to Deployment once the RS is cached") + + _, cached := r.snapshot.PodDescriptors.Load(podUID) + assert.True(t, cached, "resolved chain should now be memoized") +} + +// An owner kind we do not track (e.g. Argo Rollouts) is a terminal condition, +// not a transient one: retrying can never help, so the identity we already have +// must still be cached to avoid re-walking the chain on every scrape. +func TestResolvePodDescriptor_UnsupportedOwnerKindIsCached(t *testing.T) { + disableEphemeralAggregation(t) + + podUID := types.UID(uuid.NewString()) + r := &K8sIPResolver{} + + pod := &MinimalPod{ + UID: podUID, + Name: "rollout-pod-1", + Namespace: "nudgebee", + OwnerReferences: []metav1.OwnerReference{controllerRef("my-rollout", types.UID(uuid.NewString()), "Rollout")}, + } + + got := r.resolvePodDescriptor(pod) + assert.Equal(t, "my-rollout", got.Name) + assert.Equal(t, "Rollout", got.Kind) + + _, cached := r.snapshot.PodDescriptors.Load(podUID) + assert.True(t, cached, "terminal (unsupported kind) resolution should be memoized") +} + +// A bare ReplicaSet with no pod-template-hash must keep its own name; inventing +// a Deployment for it would be wrong. +func TestResolvePodDescriptor_BareReplicaSetKeepsItsName(t *testing.T) { + disableEphemeralAggregation(t) + + rsUID := types.UID(uuid.NewString()) + podUID := types.UID(uuid.NewString()) + + r := &K8sIPResolver{} + // Present in cache, but owned by nothing — the chain terminates here. + r.snapshot.ReplicaSets.Store(rsUID, MinimalOwnerInfo{}) + + pod := &MinimalPod{ + UID: podUID, + Name: "standalone-rs-xyz", + Namespace: "nudgebee", + OwnerReferences: []metav1.OwnerReference{controllerRef("standalone-rs", rsUID, "ReplicaSet")}, + } + + got := r.resolvePodDescriptor(pod) + assert.Equal(t, "standalone-rs", got.Name) + assert.Equal(t, "ReplicaSet", got.Kind) +} diff --git a/common/net.go b/common/net.go index f62d716..8f406ac 100644 --- a/common/net.go +++ b/common/net.go @@ -281,10 +281,31 @@ func NewDomain(fqdn string, ips []netaddr.IP) *Domain { return d } +// isKubernetesResolved reports whether a Workload carries a real in-cluster +// identity rather than the "external" placeholder the resolvers return for IPs +// they could not map to a pod, service or node. Empty Kind means unresolved. +func isKubernetesResolved(w Workload) bool { + return w.Kind != "" && w.Kind != "external" +} + func NewDestinationKey(dst, actualDst netaddr.IPPort, domain *Domain, dstWorkload Workload, actualDestWorkload Workload) DestinationKey { if IsIpExternal(actualDst.IP()) && domain != nil && !domain.SpecifyIP { - dstWorkload.Name = domain.FQDN - actualDestWorkload.Name = domain.FQDN + // Substitute the FQDN for the workload name ONLY when we have no better, + // k8s-resolved identity. A cluster-internal Service reached via a route + // whose actual destination looks external still resolves to a real + // workload (kind=Deployment/StatefulSet, real namespace); overwriting its + // name with the FQDN produced mixed-provenance labels such as + // destination_workload_name = "temporal-frontend.nudgebee.svc.cluster.local" + // destination_workload_namespace = "nudgebee" + // destination_workload_kind = "Deployment" + // which splits one workload into an extra series that no query matches. + // The FQDN is still carried by the `destination` label, so nothing is lost. + if !isKubernetesResolved(dstWorkload) { + dstWorkload.Name = domain.FQDN + } + if !isKubernetesResolved(actualDestWorkload) { + actualDestWorkload.Name = domain.FQDN + } return DestinationKey{ destination: HostPortWithEmptyIP(domain.FQDN, dst.Port()), actualDestination: HostPortFromIPPort(actualDst), diff --git a/common/net_workload_identity_test.go b/common/net_workload_identity_test.go new file mode 100644 index 0000000..a65641e --- /dev/null +++ b/common/net_workload_identity_test.go @@ -0,0 +1,87 @@ +package common + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "inet.af/netaddr" +) + +// A genuinely external destination has no in-cluster identity, so the FQDN is +// the best name available and must still be substituted. This is the behaviour +// the external-service discovery queries depend on: they select on +// actual_destination_workload_kind="external" and read +// actual_destination_workload_name, expecting the FQDN. +func TestNewDestinationKey_ExternalKeepsFQDNName(t *testing.T) { + d := netaddr.IPPortFrom(netaddr.MustParseIP("1.1.1.1"), 5432) + ad := netaddr.IPPortFrom(netaddr.MustParseIP("2.2.2.2"), 5432) + domain := &Domain{FQDN: "db.example.rds.amazonaws.com", SpecifyIP: false} + + external := Workload{Name: "2.2.2.2", Namespace: "external", Kind: "external"} + + key := NewDestinationKey(d, ad, domain, external, external) + + assert.Equal(t, "db.example.rds.amazonaws.com", key.destinationWorkload.Name) + assert.Equal(t, "db.example.rds.amazonaws.com", key.actualDestinationWorkload.Name) + assert.Equal(t, "external", key.actualDestinationWorkload.Kind) +} + +// An unresolved workload (zero value, empty Kind) also has nothing better than +// the FQDN. +func TestNewDestinationKey_UnresolvedKeepsFQDNName(t *testing.T) { + d := netaddr.IPPortFrom(netaddr.MustParseIP("1.1.1.1"), 443) + ad := netaddr.IPPortFrom(netaddr.MustParseIP("2.2.2.2"), 443) + domain := &Domain{FQDN: "aa.bb.s3.amazonaws.com", SpecifyIP: false} + + key := NewDestinationKey(d, ad, domain, Workload{}, Workload{}) + + assert.Equal(t, "aa.bb.s3.amazonaws.com", key.destinationWorkload.Name) +} + +// Regression: a cluster-internal Service whose actual destination looks external +// still resolves to a real workload. Its name must NOT be replaced by the FQDN, +// which previously produced mixed-provenance labels (FQDN name + real namespace +// + real kind) and split the workload into an extra series. +func TestNewDestinationKey_ResolvedWorkloadKeepsItsName(t *testing.T) { + d := netaddr.IPPortFrom(netaddr.MustParseIP("1.1.1.1"), 7233) + ad := netaddr.IPPortFrom(netaddr.MustParseIP("2.2.2.2"), 7233) + domain := &Domain{FQDN: "temporal-frontend.nudgebee.svc.cluster.local", SpecifyIP: false} + + resolved := Workload{Name: "temporal-frontend", Namespace: "nudgebee", Kind: "Deployment"} + + key := NewDestinationKey(d, ad, domain, resolved, resolved) + + assert.Equal(t, "temporal-frontend", key.destinationWorkload.Name, + "a k8s-resolved workload must keep its own name") + assert.Equal(t, "nudgebee", key.destinationWorkload.Namespace) + assert.Equal(t, "Deployment", key.destinationWorkload.Kind) + + // The FQDN is still available via the destination label, so nothing is lost. + assert.Equal(t, "temporal-frontend.nudgebee.svc.cluster.local:7233", key.Destination().String()) +} + +// Mixed case: destination resolved in-cluster, actual destination genuinely +// external. Each side is decided independently. +func TestNewDestinationKey_MixedResolution(t *testing.T) { + d := netaddr.IPPortFrom(netaddr.MustParseIP("1.1.1.1"), 6379) + ad := netaddr.IPPortFrom(netaddr.MustParseIP("2.2.2.2"), 6379) + domain := &Domain{FQDN: "redis-master.redis.svc.cluster.local", SpecifyIP: false} + + resolved := Workload{Name: "redis-master", Namespace: "redis", Kind: "StatefulSet"} + external := Workload{Name: "2.2.2.2", Namespace: "external", Kind: "external"} + + key := NewDestinationKey(d, ad, domain, resolved, external) + + assert.Equal(t, "redis-master", key.destinationWorkload.Name) + assert.Equal(t, "redis-master.redis.svc.cluster.local", key.actualDestinationWorkload.Name) +} + +func TestIsKubernetesResolved(t *testing.T) { + assert.False(t, isKubernetesResolved(Workload{}), "empty Kind is unresolved") + assert.False(t, isKubernetesResolved(Workload{Kind: "external"})) + assert.True(t, isKubernetesResolved(Workload{Kind: "Deployment"})) + assert.True(t, isKubernetesResolved(Workload{Kind: "StatefulSet"})) + assert.True(t, isKubernetesResolved(Workload{Kind: "Service"})) + assert.True(t, isKubernetesResolved(Workload{Kind: "pod"})) + assert.True(t, isKubernetesResolved(Workload{Kind: "node"})) +}