add metadata forwarder (DataDog#1999)

* add metadata forwarder

* refactor version and platform info, update client to reader

Author: celenechang

cmd/main.go

@@ -20,13 +20,18 @@ import (
 	"go.uber.org/zap/zapcore"
 	"gopkg.in/DataDog/dd-trace-go.v1/profiler"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	apimversion "k8s.io/apimachinery/pkg/version"
+	"k8s.io/client-go/discovery"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/klog/v2"
 	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	ctrlzap "sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
@@ -37,9 +42,13 @@ import (
 	"github.com/DataDog/datadog-operator/internal/controller"
 	"github.com/DataDog/datadog-operator/internal/controller/metrics"
 	"github.com/DataDog/datadog-operator/pkg/config"
+	"github.com/DataDog/datadog-operator/pkg/constants"
 	"github.com/DataDog/datadog-operator/pkg/controller/debug"
+	"github.com/DataDog/datadog-operator/pkg/controller/utils/metadata"
+	"github.com/DataDog/datadog-operator/pkg/kubernetes"
 	"github.com/DataDog/datadog-operator/pkg/remoteconfig"
 	"github.com/DataDog/datadog-operator/pkg/secrets"
+	"github.com/DataDog/datadog-operator/pkg/utils"
 	"github.com/DataDog/datadog-operator/pkg/version"
 
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
@@ -52,8 +61,9 @@ const (
 )
 
 var (
-	scheme   = runtime.NewScheme()
-	setupLog = ctrl.Log.WithName("setup")
+	scheme      = runtime.NewScheme()
+	setupLog    = ctrl.Log.WithName("setup")
+	metadataLog = ctrl.Log.WithName("metadata")
 )
 
 func init() {
@@ -344,10 +354,25 @@ func run(opts *options) error {
 		DatadogGenericResourceEnabled: opts.datadogGenericResourceEnabled,
 	}
 
-	if err = controller.SetupControllers(setupLog, mgr, options); err != nil {
+	versionInfo, platformInfo, err := getVersionAndPlatformInfo(rest.CopyConfig(mgr.GetConfig()))
+	if err != nil {
+		return err
+	}
+
+	if versionInfo != nil {
+		gitVersion := versionInfo.GitVersion
+		if !utils.IsAboveMinVersion(gitVersion, "1.16-0") {
+			setupLog.Error(nil, "Detected Kubernetes version <1.16 which requires CRD version apiextensions.k8s.io/v1beta1. "+
+				"CRDs of this version were removed in v1.10.0.")
+		}
+	}
+
+	if err = controller.SetupControllers(setupLog, mgr, platformInfo, options); err != nil {
 		return setupErrorf(setupLog, err, "Unable to start controllers")
 	}
 
+	setupAndStartMetadataForwarder(metadataLog, mgr.GetAPIReader(), versionInfo.String(), opts)
+
 	// +kubebuilder:scaffold:builder
 
 	setupLog.Info("starting manager")
@@ -358,6 +383,39 @@ func run(opts *options) error {
 	return nil
 }
 
+func getVersionAndPlatformInfo(configCopy *rest.Config) (*apimversion.Info, kubernetes.PlatformInfo, error) {
+	// Never use original mgr.GetConfig(), always copy as clients might modify the configuration
+	discoveryClient, err := discovery.NewDiscoveryClientForConfig(configCopy)
+	if err != nil {
+		return nil, kubernetes.PlatformInfo{}, fmt.Errorf("unable to get discovery client: %w", err)
+	}
+
+	versionInfo, err := discoveryClient.ServerVersion()
+	if err != nil {
+		return nil, kubernetes.PlatformInfo{}, fmt.Errorf("unable to get APIServer version: %w", err)
+	}
+
+	groups, resources, err := getServerGroupsAndResources(setupLog, discoveryClient)
+	if err != nil {
+		return nil, kubernetes.PlatformInfo{}, fmt.Errorf("unable to get API resource versions: %w", err)
+	}
+	platformInfo := kubernetes.NewPlatformInfo(versionInfo, groups, resources)
+
+	return versionInfo, platformInfo, nil
+
+}
+
+func getServerGroupsAndResources(log logr.Logger, discoveryClient *discovery.DiscoveryClient) ([]*v1.APIGroup, []*v1.APIResourceList, error) {
+	groups, resources, err := discoveryClient.ServerGroupsAndResources()
+	if err != nil {
+		if !discovery.IsGroupDiscoveryFailedError(err) {
+			log.Info("GetServerGroupsAndResources ERROR", "err", err)
+			return nil, nil, err
+		}
+	}
+	return groups, resources, nil
+}
+
 func customSetupLogging(logLevel zapcore.Level, logEncoder string) error {
 	encoderConfig := zap.NewProductionEncoderConfig()
 	encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder
@@ -419,3 +477,28 @@ func setupErrorf(logger logr.Logger, err error, msg string, keysAndValues ...any
 	setupLog.Error(err, msg, keysAndValues...)
 	return fmt.Errorf("%s, err:%w", msg, err)
 }
+
+func setupAndStartMetadataForwarder(logger logr.Logger, client client.Reader, kubernetesVersion string, options *options) {
+	mdf := metadata.NewMetadataForwarder(logger, client)
+	mdf.OperatorMetadata = metadata.OperatorMetadata{
+		OperatorVersion:               version.GetVersion(),
+		KubernetesVersion:             kubernetesVersion,
+		InstallMethodTool:             "datadog-operator",
+		InstallMethodToolVersion:      version.GetVersion(),
+		IsLeader:                      true,
+		DatadogAgentEnabled:           options.datadogAgentEnabled,
+		DatadogMonitorEnabled:         options.datadogMonitorEnabled,
+		DatadogDashboardEnabled:       options.datadogDashboardEnabled,
+		DatadogSLOEnabled:             options.datadogSLOEnabled,
+		DatadogGenericResourceEnabled: options.datadogGenericResourceEnabled,
+		DatadogAgentProfileEnabled:    options.datadogAgentProfileEnabled,
+		LeaderElectionEnabled:         options.enableLeaderElection,
+		ExtendedDaemonSetEnabled:      options.supportExtendedDaemonset,
+		RemoteConfigEnabled:           options.remoteConfigEnabled,
+		IntrospectionEnabled:          options.introspectionEnabled,
+		ConfigDDURL:                   os.Getenv(constants.DDURL),
+		ConfigDDSite:                  os.Getenv(constants.DDSite),
+	}
+
+	mdf.Start()
+}

config/manager/manager.yaml

@@ -43,6 +43,10 @@ spec:
         imagePullPolicy: IfNotPresent
         name: manager
         env:
+        - name: DD_HOSTNAME
+          valueFrom:
+            fieldRef:
+              fieldPath: spec.nodeName
         - name: POD_NAME
           valueFrom:
             fieldRef:

internal/controller/datadogagent/component/clusterchecksrunner/default.go

@@ -259,7 +259,7 @@ func defaultEnvVars(dda metav1.Object) []corev1.EnvVar {
 			Value: "false",
 		},
 		{
-			Name: DDHostname,
+			Name: constants.DDHostName,
 			ValueFrom: &corev1.EnvVarSource{
 				FieldRef: &corev1.ObjectFieldSelector{
 					FieldPath: common.FieldPathSpecNodeName,

internal/controller/datadogagent/component/clusterchecksrunner/envvar.go

@@ -10,5 +10,4 @@ const (
 	DDClcRunnerHost            = "DD_CLC_RUNNER_HOST"
 	DDClcRunnerID              = "DD_CLC_RUNNER_ID"
 	DDEnableMetadataCollection = "DD_ENABLE_METADATA_COLLECTION"
-	DDHostname                 = "DD_HOSTNAME"
 )

internal/controller/setup.go

@@ -15,7 +15,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -29,7 +28,6 @@ import (
 	"github.com/DataDog/datadog-operator/pkg/datadogclient"
 	"github.com/DataDog/datadog-operator/pkg/kubernetes"
 	"github.com/DataDog/datadog-operator/pkg/kubernetes/rbac"
-	"github.com/DataDog/datadog-operator/pkg/utils"
 )
 
 const (
@@ -89,33 +87,7 @@ var controllerStarters = map[string]starterFunc{
 }
 
 // SetupControllers starts all controllers (also used by e2e tests)
-func SetupControllers(logger logr.Logger, mgr manager.Manager, options SetupOptions) error {
-	// Get some information about Kubernetes version
-	// Never use original mgr.GetConfig(), always copy as clients might modify the configuration
-	discoveryConfig := rest.CopyConfig(mgr.GetConfig())
-	discoveryClient, err := discovery.NewDiscoveryClientForConfig(discoveryConfig)
-	if err != nil {
-		return fmt.Errorf("unable to get discovery client: %w", err)
-	}
-
-	versionInfo, err := discoveryClient.ServerVersion()
-	if err != nil {
-		return fmt.Errorf("unable to get APIServer version: %w", err)
-	}
-
-	if versionInfo != nil {
-		gitVersion := versionInfo.GitVersion
-		if !utils.IsAboveMinVersion(gitVersion, "1.16-0") {
-			logger.Error(nil, "Detected Kubernetes version <1.16 which requires CRD version apiextensions.k8s.io/v1beta1. "+
-				"CRDs of this version will be deprecated and will not be updated starting with Operator v1.8.0 and will be removed in v1.10.0.")
-		}
-	}
-
-	groups, resources, err := getServerGroupsAndResources(logger, discoveryClient)
-	if err != nil {
-		return fmt.Errorf("unable to get API resource versions: %w", err)
-	}
-	platformInfo := kubernetes.NewPlatformInfo(versionInfo, groups, resources)
+func SetupControllers(logger logr.Logger, mgr manager.Manager, platformInfo kubernetes.PlatformInfo, options SetupOptions) error {
 
 	var metricForwardersMgr datadog.MetricsForwardersManager
 	if options.OperatorMetricsEnabled {
@@ -131,17 +103,6 @@ func SetupControllers(logger logr.Logger, mgr manager.Manager, options SetupOpti
 	return nil
 }
 
-func getServerGroupsAndResources(log logr.Logger, discoveryClient *discovery.DiscoveryClient) ([]*metav1.APIGroup, []*metav1.APIResourceList, error) {
-	groups, resources, err := discoveryClient.ServerGroupsAndResources()
-	if err != nil {
-		if !discovery.IsGroupDiscoveryFailedError(err) {
-			log.Info("GetServerGroupsAndResources ERROR", "err", err)
-			return nil, nil, err
-		}
-	}
-	return groups, resources, nil
-}
-
 func startDatadogAgent(logger logr.Logger, mgr manager.Manager, pInfo kubernetes.PlatformInfo, options SetupOptions, metricForwardersMgr datadog.MetricsForwardersManager) error {
 	if !options.DatadogAgentEnabled {
 		logger.Info("Feature disabled, not starting the controller", "controller", agentControllerName)

internal/controller/suite_v2_test.go

@@ -36,6 +36,7 @@ import (
 
 	datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1"
 	"github.com/DataDog/datadog-operator/internal/controller/testutils"
+	"github.com/DataDog/datadog-operator/pkg/kubernetes"
 
 	"github.com/DataDog/datadog-operator/pkg/config"
 	// +kubebuilder:scaffold:imports
@@ -112,7 +113,9 @@ var _ = BeforeSuite(func() {
 		V2APIEnabled:               true,
 	}
 
-	err = SetupControllers(logger, mgr, options)
+	dummyPlatformInfo := kubernetes.PlatformInfo{}
+
+	err = SetupControllers(logger, mgr, dummyPlatformInfo, options)
 	Expect(err).ToNot(HaveOccurred())
 
 	var mgrCtx context.Context

pkg/constants/envvar.go

@@ -13,4 +13,5 @@ const (
 	DDSite        = "DD_SITE"
 	DDClusterName = "DD_CLUSTER_NAME"
 	DDLogLevel    = "DD_LOG_LEVEL"
+	DDHostName    = "DD_HOSTNAME"
 )