Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -1272,7 +1272,7 @@ replace github.com/iceber/iouring-go => github.com/lebauce/iouring-go v0.0.0-202
// Fork to remove some text/template usage, https://github.com/DataDog/opa/tree/lightweight-1.7.1
replace github.com/open-policy-agent/opa => github.com/DataDog/opa v0.0.0-20251126100856-d2e1e78e0816

replace github.com/DataDog/agent-payload/v5 => github.com/DataDog/agent-payload/v5 v5.0.196
replace github.com/DataDog/agent-payload/v5 => github.com/DataDog/agent-payload/v5 v5.0.198-0.20260521091023-1b5b2888eeb2

// This section was automatically added by 'dda inv modules.add-all-replace' command, do not edit manually

Expand Down
4 changes: 2 additions & 2 deletions go.sum

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,12 @@ func (p *ClusterProcessor) Process(ctx processors.ProcessorContext, list interfa
clusterModel.CreationTimestamp = kubeSystemCreationTimestamp.Unix()
}

if info, err := k8sTransformers.FetchClusterInfo(context.TODO(), pctx.APIClient.Cl.CoreV1()); err != nil {
log.Debugc(fmt.Sprintf("failed to fetch dd-cluster-info ConfigMap: %s", err), orchestrator.ExtraLogContext...)
} else {
k8sTransformers.ApplyClusterInfo(clusterModel, nodesInfo, info, pctx.Cfg.KubeClusterName)
}

if err := fillClusterResourceVersion(clusterModel); err != nil {
return processResult, 0, fmt.Errorf("failed to compute resource version: %s", err.Error())
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,147 @@ func TestClusterProcessor_Process_Success(t *testing.T) {
assert.Equal(t, *cluster, clusterFromManifest)
}

func TestClusterProcessor_Process_NoClusterInfo(t *testing.T) {
node := createTestClusterNode("node-1", "v1.20.0", "1000m", "2Gi", "110")
kubeSystemNS := &corev1.Namespace{
ObjectMeta: metav1.ObjectMeta{
Name: "kube-system",
CreationTimestamp: metav1.NewTime(time.Date(2021, time.April, 16, 14, 30, 0, 0, time.UTC)),
},
}
client := fake.NewClientset(node, kubeSystemNS)
fakeDiscoveryClient := client.Discovery().(*fakediscovery.FakeDiscovery)
fakeDiscoveryClient.FakedServerVersion = &version.Info{Major: "1", Minor: "20", GitVersion: "v1.20.0"}

cfg := orchestratorconfig.NewDefaultOrchestratorConfig(nil)
cfg.KubeClusterName = "test-cluster"
ctx := &processors.K8sProcessorContext{
BaseProcessorContext: processors.BaseProcessorContext{
Cfg: cfg,
Clock: clock.New(),
ClusterID: "test-cluster-id",
MsgGroupID: 1,
ManifestProducer: true,
},
APIClient: &apiserver.APIClient{Cl: client},
HostName: "test-host",
}

processor := NewClusterProcessor()
result, _, err := processor.Process(ctx, []*corev1.Node{node})
require.NoError(t, err)

cluster := result.MetadataMessages[0].(*model.CollectorCluster).Cluster
assert.Nil(t, cluster.Autoscaling)
assert.Equal(t, int64(0), cluster.ClusterInfoGeneratedAtUnixNano)
assert.Nil(t, cluster.CloudResourceId)
for _, n := range cluster.NodesInfo {
assert.Empty(t, n.NodeManager, "node %q should have no nodeManager when ConfigMap is absent", n.Name)
}
}

func TestClusterProcessor_Process_WithClusterInfo(t *testing.T) {
node1 := createTestClusterNode("node-1", "v1.20.0", "1000m", "2Gi", "110")
node2 := createTestClusterNode("node-2", "v1.20.0", "2000m", "4Gi", "220")
kubeSystemNS := &corev1.Namespace{
ObjectMeta: metav1.ObjectMeta{
Name: "kube-system",
CreationTimestamp: metav1.NewTime(time.Date(2021, time.April, 16, 14, 30, 0, 0, time.UTC)),
},
}
clusterInfoCM := &corev1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
Name: "dd-cluster-info",
Namespace: "dd-karpenter",
Labels: map[string]string{
"app.kubernetes.io/managed-by": "kubectl-datadog",
},
},
Data: map[string]string{
"cluster-info": "apiVersion: v1\n" +
"clusterName: test-cluster\n" +
"clusterArn: arn:aws:eks:eu-west-3:000000000000:cluster/test-cluster\n" +
"region: eu-west-3\n" +
"generatedAt: 2026-05-13T22:42:45Z\n" +
"nodeManagement:\n" +
" karpenter:\n" +
" pool-a:\n" +
" nodes: [node-1]\n" +
" managedByDatadog: true\n" +
" asg:\n" +
" group-z:\n" +
" nodes: [node-2]\n" +
"autoscaling:\n" +
" clusterAutoscaler:\n" +
" present: false\n" +
" karpenter:\n" +
" present: true\n" +
" namespace: dd-karpenter\n" +
" name: karpenter\n" +
" version: 1.12.1\n" +
" managedByDatadog: true\n" +
" installerVersion: v0.7.0\n" +
" eksAutoMode:\n" +
" enabled: false\n",
},
}

client := fake.NewClientset(node1, node2, kubeSystemNS, clusterInfoCM)
fakeDiscoveryClient := client.Discovery().(*fakediscovery.FakeDiscovery)
fakeDiscoveryClient.FakedServerVersion = &version.Info{Major: "1", Minor: "20", GitVersion: "v1.20.0"}

cfg := orchestratorconfig.NewDefaultOrchestratorConfig(nil)
cfg.KubeClusterName = "test-cluster"
ctx := &processors.K8sProcessorContext{
BaseProcessorContext: processors.BaseProcessorContext{
Cfg: cfg,
Clock: clock.New(),
ClusterID: "test-cluster-id",
MsgGroupID: 1,
ManifestProducer: true,
},
APIClient: &apiserver.APIClient{Cl: client},
HostName: "test-host",
}

processor := NewClusterProcessor()
result, _, err := processor.Process(ctx, []*corev1.Node{node1, node2})
require.NoError(t, err)

cluster := result.MetadataMessages[0].(*model.CollectorCluster).Cluster
require.NotNil(t, cluster.Autoscaling)
require.NotNil(t, cluster.Autoscaling.Karpenter)
assert.True(t, cluster.Autoscaling.Karpenter.Present)
assert.Equal(t, "1.12.1", cluster.Autoscaling.Karpenter.Version)
assert.NotZero(t, cluster.ClusterInfoGeneratedAtUnixNano)

arnWrapper, ok := cluster.CloudResourceId.(*model.Cluster_Arn)
require.True(t, ok)
assert.Equal(t, "arn:aws:eks:eu-west-3:000000000000:cluster/test-cluster", arnWrapper.Arn)

nodesByName := make(map[string]*model.ClusterNodeInfo)
for _, n := range cluster.NodesInfo {
nodesByName[n.Name] = n
}
require.Contains(t, nodesByName, "node-1")
assert.Equal(t, "karpenter", nodesByName["node-1"].NodeManager)
assert.Equal(t, "pool-a", nodesByName["node-1"].NodeManagerName)
assert.True(t, nodesByName["node-1"].NodeManagerManagedByDatadog)

require.Contains(t, nodesByName, "node-2")
assert.Equal(t, "asg", nodesByName["node-2"].NodeManager)
assert.Equal(t, "group-z", nodesByName["node-2"].NodeManagerName)
assert.False(t, nodesByName["node-2"].NodeManagerManagedByDatadog)

assert.NotEmpty(t, cluster.ResourceVersion, "cluster info should be hashed into ResourceVersion")

manifestContent := string(result.ManifestMessages[0].(*model.CollectorManifest).Manifests[0].Content)
assert.Contains(t, manifestContent, `"autoscaling":`)
assert.Contains(t, manifestContent, `"karpenter":`)
assert.Contains(t, manifestContent, `"nodeManager":"karpenter"`)
assert.Contains(t, manifestContent, `"arn":"arn:aws:eks:eu-west-3:000000000000:cluster/test-cluster"`)
}

func TestClusterProcessor_ExtendedResources(t *testing.T) {
// Create node with extended resources
node := createTestClusterNode("node-1", "v1.20.0", "1000m", "2Gi", "110")
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,211 @@
// Unless explicitly stated otherwise all files in this repository are licensed
// under the Apache License Version 2.0.
// This product includes software developed at Datadog (https://www.datadoghq.com/).
// Copyright 2016-present Datadog, Inc.

//go:build orchestrator

package k8s

import (
"context"
"fmt"
"time"

"github.com/samber/lo"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
corev1Client "k8s.io/client-go/kubernetes/typed/core/v1"
"sigs.k8s.io/yaml"

model "github.com/DataDog/agent-payload/v5/process"
"github.com/DataDog/datadog-agent/pkg/orchestrator"
"github.com/DataDog/datadog-agent/pkg/util/log"
)

const (
clusterInfoConfigMapName = "dd-cluster-info"
clusterInfoConfigMapDataKey = "cluster-info"
clusterInfoManagedByLabel = "app.kubernetes.io/managed-by"
clusterInfoManagedByValue = "kubectl-datadog"
clusterInfoAPIVersion = "v1"
)

// ClusterInfo mirrors the YAML payload of the dd-cluster-info ConfigMap
// produced by `kubectl datadog autoscaling cluster install`. The Go schema
// lives in datadog-operator; we replicate the YAML tags here so the agent
// does not depend on that repository.
type ClusterInfo struct {
APIVersion string `json:"apiVersion"`
ClusterName string `json:"clusterName"`
ClusterARN string `json:"clusterArn,omitempty"`
Region string `json:"region,omitempty"`
GeneratedAt time.Time `json:"generatedAt"`
NodeManagement map[string]map[string]nodeManagerEntry `json:"nodeManagement"`
Autoscaling autoscalingYAML `json:"autoscaling"`
}

type nodeManagerEntry struct {
Nodes []string `json:"nodes"`
ManagedByDatadog bool `json:"managedByDatadog,omitempty"`
}

type autoscalingYAML struct {
ClusterAutoscaler clusterAutoscalerYAML `json:"clusterAutoscaler"`
Karpenter karpenterYAML `json:"karpenter"`
EKSAutoMode eksAutoModeYAML `json:"eksAutoMode"`
}

type clusterAutoscalerYAML struct {
Present bool `json:"present"`
Namespace string `json:"namespace,omitempty"`
Name string `json:"name,omitempty"`
Version string `json:"version,omitempty"`
}

type karpenterYAML struct {
Present bool `json:"present"`
Namespace string `json:"namespace,omitempty"`
Name string `json:"name,omitempty"`
Version string `json:"version,omitempty"`
ManagedByDatadog bool `json:"managedByDatadog,omitempty"`
InstallerVersion string `json:"installerVersion,omitempty"`
}

type eksAutoModeYAML struct {
Enabled bool `json:"enabled"`
}

// FetchClusterInfo locates the dd-cluster-info ConfigMap anywhere in the
// cluster and returns its parsed payload. The ConfigMap is identified by
// its fixed name and the kubectl-datadog managed-by label. Returns
// (nil, nil) when no matching ConfigMap exists, when its schema version
// is not recognised, or when the payload key is missing — the absence of
// cluster-info is not an error, the snapshot is informational.
func FetchClusterInfo(ctx context.Context, coreClient corev1Client.CoreV1Interface) (*ClusterInfo, error) {
cms, err := coreClient.ConfigMaps(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
LabelSelector: clusterInfoManagedByLabel + "=" + clusterInfoManagedByValue,
FieldSelector: "metadata.name=" + clusterInfoConfigMapName,
})
if err != nil {
return nil, fmt.Errorf("listing %s ConfigMaps: %w", clusterInfoConfigMapName, err)
}
if len(cms.Items) == 0 {
return nil, nil
}

if len(cms.Items) > 1 {
namespaces := lo.Map(cms.Items, func(cm corev1.ConfigMap, _ int) string { return cm.Namespace })
log.Warnc(fmt.Sprintf("found multiple %s ConfigMaps in namespaces %v, using %q",
clusterInfoConfigMapName, namespaces, cms.Items[0].Namespace), orchestrator.ExtraLogContext...)
Comment thread
L3n41c marked this conversation as resolved.
}
chosen := &cms.Items[0]
Comment thread
L3n41c marked this conversation as resolved.

payload, ok := chosen.Data[clusterInfoConfigMapDataKey]
Comment thread
L3n41c marked this conversation as resolved.
Comment thread
L3n41c marked this conversation as resolved.
if !ok {
log.Warnc(fmt.Sprintf("%s ConfigMap in namespace %q is missing the %q data key",
clusterInfoConfigMapName, chosen.Namespace, clusterInfoConfigMapDataKey),
orchestrator.ExtraLogContext...)
return nil, nil
}

var info ClusterInfo
if err := yaml.Unmarshal([]byte(payload), &info); err != nil {
return nil, fmt.Errorf("parsing %s ConfigMap payload: %w", clusterInfoConfigMapName, err)
}
if info.APIVersion != clusterInfoAPIVersion {
log.Warnc(fmt.Sprintf("%s ConfigMap apiVersion %q is not supported (expected %q), ignoring",
clusterInfoConfigMapName, info.APIVersion, clusterInfoAPIVersion),
orchestrator.ExtraLogContext...)
return nil, nil
}

return &info, nil
}

// ApplyClusterInfo enriches a cluster model and its node summaries with
// data sourced from the dd-cluster-info ConfigMap.
func ApplyClusterInfo(cluster *model.Cluster, nodesInfo []*model.ClusterNodeInfo, info *ClusterInfo, agentClusterName string) {
if info == nil {
return
}

if info.ClusterName != "" && agentClusterName != "" && info.ClusterName != agentClusterName {
log.Warnc(fmt.Sprintf("dd-cluster-info clusterName %q does not match the agent cluster name %q",
info.ClusterName, agentClusterName), orchestrator.ExtraLogContext...)
}
if info.Region != "" {
for _, n := range nodesInfo {
if n.Region != "" && n.Region != info.Region {
log.Warnc(fmt.Sprintf("dd-cluster-info region %q does not match node %q region %q",
info.Region, n.Name, n.Region), orchestrator.ExtraLogContext...)
break
}
}
}

cluster.Autoscaling = toAutoscalingInfo(info.Autoscaling)
if !info.GeneratedAt.IsZero() {
cluster.ClusterInfoGeneratedAtUnixNano = info.GeneratedAt.UnixNano()
}
if info.ClusterARN != "" {
cluster.CloudResourceId = &model.Cluster_Arn{Arn: info.ClusterARN}
}

nodeIndex := buildNodeManagerIndex(info.NodeManagement)
for _, n := range nodesInfo {
if entry, ok := nodeIndex[n.Name]; ok {
n.NodeManager = entry.manager
n.NodeManagerName = entry.name
n.NodeManagerManagedByDatadog = entry.managedByDatadog
}
}
}

type nodeManagerAssignment struct {
manager string
name string
managedByDatadog bool
}

// buildNodeManagerIndex inverts the dd-cluster-info nodeManagement
// structure (manager → entity name → entry) into a lookup by node name.
// A given node belongs to exactly one management entity, so the inner
// writes never collide.
func buildNodeManagerIndex(nodeManagement map[string]map[string]nodeManagerEntry) map[string]nodeManagerAssignment {
out := make(map[string]nodeManagerAssignment)
for manager, entries := range nodeManagement {
for name, entry := range entries {
for _, node := range entry.Nodes {
out[node] = nodeManagerAssignment{
manager: manager,
name: name,
managedByDatadog: entry.ManagedByDatadog,
}
}
}
}
return out
}

func toAutoscalingInfo(a autoscalingYAML) *model.AutoscalingInfo {
return &model.AutoscalingInfo{
ClusterAutoscaler: &model.ClusterAutoscalerInfo{
Present: a.ClusterAutoscaler.Present,
Namespace: a.ClusterAutoscaler.Namespace,
Name: a.ClusterAutoscaler.Name,
Version: a.ClusterAutoscaler.Version,
},
Karpenter: &model.KarpenterInfo{
Present: a.Karpenter.Present,
Namespace: a.Karpenter.Namespace,
Name: a.Karpenter.Name,
Version: a.Karpenter.Version,
ManagedByDatadog: a.Karpenter.ManagedByDatadog,
InstallerVersion: a.Karpenter.InstallerVersion,
},
EksAutoMode: &model.EKSAutoModeInfo{
Enabled: a.EKSAutoMode.Enabled,
},
}
}
Loading
Loading