Skip to content

[CONTP-1569] KSM pod .total aggregation through a clc cluster_aggregates_only check #53042

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gh-worker-dd-mergequeue-cf854d merged 1 commit into from
Jul 8, 2026
Merged

[CONTP-1569] KSM pod .total aggregation through a clc cluster_aggregates_only check #53042

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 116 additions & 4 deletions pkg/collector/corechecks/cluster/ksm/kubernetes_state.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,25 @@ const (
// enabled on the node agents, because unassigned pods cannot be collected
// from node agents.
clusterUnassignedPodCollection podCollectionMode = "cluster_unassigned"

// clusterAggregatesOnlyPodCollection is the authoritative source for the
// cluster-aggregate `.total` metric family. It emits only the `.total`
// aggregators (no per-pod metrics). Meant to run on the cluster-agent or
// a cluster check runner (same isRunningOnNodeAgent guard as
// clusterUnassignedPodCollection) — never on a node agent. When a cluster
// check runner pool exists (a prerequisite for node_kubelet's
// clusterUnassignedPodCollection companion mode), the check's normal
// cluster-check dispatch (cluster_check: true in its instance config)
// assigns this to exactly one runner in that pool, same as any other
// cluster check; it only falls back to running directly on the
// cluster-agent if no runner pool exists. Other modes suppress these
// aggregators so the cluster sees exactly one authoritative source.
//
// Pod data is sourced via the same direct API-server reflector mechanism
// the default/cluster_unassigned modes use — not workloadmeta — to avoid
// depending on workloadmeta's pod store lifecycle (whether it's running,
// and whether it's already fully populated) for correctness.
clusterAggregatesOnlyPodCollection podCollectionMode = "cluster_aggregates_only"
)

// KSMConfig contains the check config parameters
Expand Down Expand Up @@ -231,8 +250,16 @@ type KSMConfig struct {
UseAPIServerCache bool `yaml:"use_apiserver_cache"`

// PodCollectionMode defines how pods are collected.
// Accepted values are: "default", "node_kubelet", and "cluster_unassigned".
// Accepted values are: "default", "node_kubelet", "cluster_unassigned",
// and "cluster_aggregates_only".
PodCollectionMode podCollectionMode `yaml:"pod_collection_mode"`

// ClusterAggregatesEnabled tells a node_kubelet or cluster_unassigned
// instance to suppress its .total accumulators, because a dedicated
// cluster_aggregates_only instance is deployed elsewhere in the cluster as
// the sole authoritative source. Off by default (legacy behavior). Set by
// the operator/helm on the suppressing instances alongside that deployment.
ClusterAggregatesEnabled bool `yaml:"cluster_aggregates_enabled"`
}

// KSMCheck wraps the config and the metric stores needed to run the check
Expand Down Expand Up @@ -328,7 +355,11 @@ func (k *KSMCheck) Configure(senderManager sender.SenderManager, integrationConf
// Prepare labels mapper
k.mergeLabelsMapper(defaultLabelsMapper())

if k.instance.usesCustomResourceMetrics() && k.instance.PodCollectionMode != nodeKubeletPodCollection {
// Start custom resource discovery if not running in a mode that only
// consumes pods from workloadmeta (no API discovery needed in those modes).
if k.instance.usesCustomResourceMetrics() &&
k.instance.PodCollectionMode != nodeKubeletPodCollection &&
k.instance.PodCollectionMode != clusterAggregatesOnlyPodCollection {
k.crMu.Lock()
if k.cancelCR == nil {
ctx, cancel := context.WithCancel(context.Background())
Expand Down Expand Up @@ -380,10 +411,26 @@ func (k *KSMCheck) buildStores() error {

switch k.instance.PodCollectionMode {
case nodeKubeletPodCollection:
// In this case we don't need to set up anything related to the API
// server.
// Pods come from the kubelet, nothing API-server related to set up.
collectors = []string{"pods"}
k.setupLabelsAndAnnotationsAsTagsFunc()
case clusterAggregatesOnlyPodCollection:
collectors = []string{"pods"}
k.setupLabelsAndAnnotationsAsTagsFunc()

// Watch pods directly via the same API-server client mechanism as
// defaultPodCollection, rather than through workloadmeta.
apiServerClient, err = apiserver.GetAPIClient()
if err != nil {
return err
}

err = apiserver.InitializeGlobalResourceTypeCache(apiServerClient.Cl.Discovery())
if err != nil {
return err
}

builder.WithKubeClient(apiServerClient.InformerCl)
case defaultPodCollection, clusterUnassignedPodCollection:
// We can try to get the API Client directly because this code will be retried if it fails
apiServerClient, err = apiserver.GetAPIClient()
Expand Down Expand Up @@ -579,6 +626,25 @@ func (k *KSMCheck) discoverCustomResources(c *apiserver.APIClient, collectors []
}
}

if k.instance.PodCollectionMode == clusterAggregatesOnlyPodCollection {
// Enable ONLY the extended pod store — it produces the 4
// owner-tagged source metrics for the .total family. Deliberately
// omit the plain "pods" collector so the standard pod store (~54
// per-pod generators, all discarded by processMetrics in this mode)
// is never built or watched. extendedCollectors["pods"] is
// "core/v1, Resource=pods_extended", the exact key
// WithCustomResourceStoreFactories registers NewExtendedPodFactory
// under (util.GVRFromType("pods_extended", *v1.Pod)), so BuildStores
// finds and builds it. Returning nil collectors instead would build
// neither store and make .total disappear.
return customResources{
collectors: []string{extendedCollectors["pods"]},
factories: []customresource.RegistryFactory{
customresources.NewExtendedPodFactory(c),
},
}
}

// extended resource collectors always have a factory registered
factories := []customresource.RegistryFactory{
customresources.NewExtendedJobFactory(c),
Expand Down Expand Up @@ -791,8 +857,38 @@ func (k *KSMCheck) Cancel() {

// processMetrics attaches tags and forwards metrics to the aggregator
func (k *KSMCheck) processMetrics(sender sender.Sender, metrics map[string][]ksmstore.DDMetricsFam, labelJoiner *labelJoiner, now time.Time) {
// Suppress .total accumulation on the per-node instances that would otherwise
// collide, but only when the instance sets cluster_aggregates_enabled: true:
// - node_kubelet (per-node agents — same reduced-tag gauge collides across nodes)
// - cluster_unassigned (CLC runner — partial view, would emit an incomplete .total)
// default mode is intentionally excluded: a single full-pod check is already the
// authoritative source for .total (no collision), so it must stay unaffected (design goal #5).
//
// Turning this on also starts the DCA's cluster_aggregates_only reflector
// warmup in the same window these instances start suppressing — there is no
// clean "DCA serving first, then nodes go quiet" handoff. Expect a brief
// (seconds-scale) window at flag-flip where .total is absent, then correct.
// The flag is a config signal, not a live DCA health check. While it is off
// (the default), nodes fall back to the pre-fix under-reported emission
// rather than going silent.
suppressClusterAggregates := (k.instance.PodCollectionMode == nodeKubeletPodCollection ||
k.instance.PodCollectionMode == clusterUnassignedPodCollection) &&
k.instance.ClusterAggregatesEnabled
// In cluster_aggregates_only mode, the check emits only the .total family
// (via aggregator flush). Skip per-pod transformer/mapper dispatch to avoid
// double-emission of per-pod metrics already produced by node-agents.
aggregatesOnly := k.instance.PodCollectionMode == clusterAggregatesOnlyPodCollection
for _, metricsList := range metrics {
for _, metricFamily := range metricsList {
if suppressClusterAggregates && isClusterAggregateSourceMetric(metricFamily.Name) {
continue
}
// In cluster_aggregates_only mode accumulate only the four .total source
// metrics. Accumulating other aggregators (e.g. pod.count) would double-count
// with node-agents which already emit those metrics individually.
if aggregatesOnly && !isClusterAggregateSourceMetric(metricFamily.Name) {
continue
}
// First check for aggregator, because the check use _labels metrics to aggregate values.
if aggregator, found := k.metricAggregators[metricFamily.Name]; found {
for _, m := range metricFamily.ListMetrics {
Expand All @@ -801,6 +897,9 @@ func (k *KSMCheck) processMetrics(sender sender.Sender, metrics map[string][]ksm
// Some metrics can be aggregated and consumed as-is or by a transformer.
// So, let’s continue the processing.
}
if aggregatesOnly {
continue
}
if transform, found := k.metricTransformers[metricFamily.Name]; found {
lMapperOverride := labelsMapperOverride(metricFamily.Name)
for _, m := range metricFamily.ListMetrics {
Expand Down Expand Up @@ -1121,6 +1220,18 @@ func (k *KSMCheck) configurePodCollection(builder *kubestatemetrics.Builder, col
} else {
builder.WithUnassignedPodsCollection()
}
case clusterAggregatesOnlyPodCollection:
if k.isRunningOnNodeAgent {
log.Warnf("cluster_aggregates_only is enabled but KSM is running in a node agent, so the option will be ignored " +
"(it is only supported on the cluster agent or a cluster check runner)")
k.instance.PodCollectionMode = defaultPodCollection
}
// Else: leave the builder unconfigured here. buildStores() takes the
// same API-server-client branch it uses for defaultPodCollection, and
// the default (non-WLM) reflector path runs as-is — no field selector
// is set, so it watches all pods in the configured namespaces (all
// namespaces by default, same as this mode needs; still subject to
// instance.Namespaces if the user has restricted it).
default:
log.Warnf("invalid pod collection mode %q, falling back to the default mode", k.instance.PodCollectionMode)
k.instance.PodCollectionMode = defaultPodCollection
Expand Down Expand Up @@ -1209,6 +1320,7 @@ func KubeStateMetricsFactoryWithParam(labelsMapper map[string]string, labelJoins
func newKSMCheck(base core.CheckBase, instance *KSMConfig, tagger tagger.Component, wmeta workloadmeta.Component) *KSMCheck {
k := &KSMCheck{
CheckBase: base,
agentConfig: pkgconfigsetup.Datadog(),
instance: instance,
telemetry: newTelemetryCache(),
tagger: tagger,
Expand Down
17 changes: 17 additions & 0 deletions pkg/collector/corechecks/cluster/ksm/kubernetes_state_aggregators.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,23 @@ func (a *lastCronJobAggregator) flush(sender sender.Sender, k *KSMCheck, labelJo
a.accumulator = make(map[cronJob]cronJobState)
}

// clusterAggregateSourceMetrics lists the KSM source metrics that feed the
// cluster-aggregate `.total` family with a reduced tag set (no host/pod/node).
// These must be accumulated only by the single authoritative instance
// (cluster_aggregates_only mode); any other instance emitting them causes
// gauge collapse at ingestion.
var clusterAggregateSourceMetrics = map[string]struct{}{
"kube_pod_container_resource_with_owner_tag_requests": {},
"kube_pod_container_resource_with_owner_tag_limits": {},
"kube_pod_init_container_resource_with_owner_tag_requests": {},
"kube_pod_init_container_resource_with_owner_tag_limits": {},
}

func isClusterAggregateSourceMetric(name string) bool {
_, ok := clusterAggregateSourceMetrics[name]
return ok
}

func defaultMetricAggregators() map[string]metricAggregator {
cronJobAggregator := newLastCronJobAggregator()

Expand Down
Loading
Loading