[ASCII-2510] Add remote-autodiscovery scheduler streaming endpoint

comp/api/api/apiimpl/grpc.go

@@ -10,30 +10,30 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/DataDog/datadog-agent/comp/remote-config/rcservice"
-	"github.com/DataDog/datadog-agent/comp/remote-config/rcservicemrf"
-	"github.com/DataDog/datadog-agent/pkg/util/optional"
-
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
 
+	"github.com/DataDog/datadog-agent/comp/core/autodiscovery"
+	autodiscoverystream "github.com/DataDog/datadog-agent/comp/core/autodiscovery/stream"
 	remoteagentregistry "github.com/DataDog/datadog-agent/comp/core/remoteagentregistry/def"
 	rarproto "github.com/DataDog/datadog-agent/comp/core/remoteagentregistry/proto"
-	workloadmetaServer "github.com/DataDog/datadog-agent/comp/core/workloadmeta/server"
-
 	tagger "github.com/DataDog/datadog-agent/comp/core/tagger/def"
 	taggerProto "github.com/DataDog/datadog-agent/comp/core/tagger/proto"
 	taggerserver "github.com/DataDog/datadog-agent/comp/core/tagger/server"
 	taggerTypes "github.com/DataDog/datadog-agent/comp/core/tagger/types"
+	workloadmetaServer "github.com/DataDog/datadog-agent/comp/core/workloadmeta/server"
 	"github.com/DataDog/datadog-agent/comp/dogstatsd/pidmap"
 	dsdReplay "github.com/DataDog/datadog-agent/comp/dogstatsd/replay/def"
 	dogstatsdServer "github.com/DataDog/datadog-agent/comp/dogstatsd/server"
+	"github.com/DataDog/datadog-agent/comp/remote-config/rcservice"
+	"github.com/DataDog/datadog-agent/comp/remote-config/rcservicemrf"
 	pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/core"
 	"github.com/DataDog/datadog-agent/pkg/util/grpc"
 	"github.com/DataDog/datadog-agent/pkg/util/hostname"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
+	"github.com/DataDog/datadog-agent/pkg/util/optional"
 )
 
 type grpcServer struct {
@@ -51,6 +51,7 @@ type serverSecure struct {
 	capture             dsdReplay.Component
 	pidMap              pidmap.Component
 	remoteAgentRegistry remoteagentregistry.Component
+	autodiscovery       autodiscovery.Component
 }
 
 func (s *grpcServer) GetHostname(ctx context.Context, _ *pb.HostnameRequest) (*pb.HostnameReply, error) {
@@ -202,6 +203,10 @@ func (s *serverSecure) RegisterRemoteAgent(_ context.Context, in *pb.RegisterRem
 	}, nil
 }
 
+func (s *serverSecure) AutodiscoveryStreamConfig(_ *emptypb.Empty, out pb.AgentSecure_AutodiscoveryStreamConfigServer) error {
+	return autodiscoverystream.Config(s.autodiscovery, out)
+}
+
 func init() {
 	grpclog.SetLoggerV2(grpc.NewLogger())
 }

comp/api/api/apiimpl/server_cmd.go

@@ -76,6 +76,7 @@ func (server *apiServer) startCMDServer(
 		capture:             server.capture,
 		pidMap:              server.pidMap,
 		remoteAgentRegistry: server.remoteAgentRegistry,
+		autodiscovery:       server.autoConfig,
 	})
 
 	dcreds := credentials.NewTLS(&tls.Config{

comp/core/autodiscovery/proto/proto.go

@@ -0,0 +1,104 @@
+// 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.
+
+// Package proto provides autodiscovery proto util functions
+package proto
+
+import (
+	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/integration"
+	"github.com/DataDog/datadog-agent/pkg/proto/pbgo/core"
+)
+
+// ProtobufConfigFromAutodiscoveryConfig converts an autodiscovery config to a protobuf config
+func ProtobufConfigFromAutodiscoveryConfig(config *integration.Config) *core.Config {
+	if config == nil {
+		return nil
+	}
+
+	instances := [][]byte{}
+
+	for _, instance := range config.Instances {
+		instances = append(instances, []byte(instance))
+	}
+
+	advancedAdIdentifiers := make([]*core.AdvancedADIdentifier, 0, len(config.AdvancedADIdentifiers))
+	for _, advancedAdIdentifier := range config.AdvancedADIdentifiers {
+		advancedAdIdentifiers = append(advancedAdIdentifiers, &core.AdvancedADIdentifier{
+			KubeService: &core.KubeNamespacedName{
+				Name:      advancedAdIdentifier.KubeService.Name,
+				Namespace: advancedAdIdentifier.KubeService.Namespace,
+			},
+			KubeEndpoints: &core.KubeNamespacedName{
+				Name:      advancedAdIdentifier.KubeEndpoints.Name,
+				Namespace: advancedAdIdentifier.KubeEndpoints.Namespace,
+			},
+		})
+	}
+
+	return &core.Config{
+		Name:                    config.Name,
+		Instances:               instances,
+		InitConfig:              config.InitConfig,
+		MetricConfig:            config.MetricConfig,
+		LogsConfig:              config.LogsConfig,
+		AdIdentifiers:           config.ADIdentifiers,
+		AdvancedAdIdentifiers:   advancedAdIdentifiers,
+		Provider:                config.Provider,
+		ServiceId:               config.ServiceID,
+		TaggerEntity:            config.TaggerEntity,
+		ClusterCheck:            config.ClusterCheck,
+		NodeName:                config.NodeName,
+		Source:                  config.Source,
+		IgnoreAutodiscoveryTags: config.IgnoreAutodiscoveryTags,
+		MetricsExcluded:         config.MetricsExcluded,
+		LogsExcluded:            config.LogsExcluded,
+	}
+}
+
+// AutodiscoveryConfigFromProtobufConfig converts a protobuf config to an autodiscovery config
+func AutodiscoveryConfigFromProtobufConfig(config *core.Config) integration.Config {
+	if config == nil {
+		return integration.Config{}
+	}
+
+	instances := []integration.Data{}
+
+	for _, instance := range config.Instances {
+		instances = append(instances, integration.Data(instance))
+	}
+
+	advancedAdIdentifiers := make([]integration.AdvancedADIdentifier, 0, len(config.AdvancedAdIdentifiers))
+	for _, advancedAdIdentifier := range config.AdvancedAdIdentifiers {
+		advancedAdIdentifiers = append(advancedAdIdentifiers, integration.AdvancedADIdentifier{
+			KubeService: integration.KubeNamespacedName{
+				Name:      advancedAdIdentifier.KubeService.Name,
+				Namespace: advancedAdIdentifier.KubeService.Namespace,
+			},
+			KubeEndpoints: integration.KubeNamespacedName{
+				Name:      advancedAdIdentifier.KubeEndpoints.Name,
+				Namespace: advancedAdIdentifier.KubeEndpoints.Namespace,
+			},
+		})
+	}
+
+	return integration.Config{
+		Name:                    config.Name,
+		Instances:               instances,
+		InitConfig:              config.InitConfig,
+		MetricConfig:            config.MetricConfig,
+		LogsConfig:              config.LogsConfig,
+		ADIdentifiers:           config.AdIdentifiers,
+		AdvancedADIdentifiers:   advancedAdIdentifiers,
+		Provider:                config.Provider,
+		ServiceID:               config.ServiceId,
+		TaggerEntity:            config.TaggerEntity,
+		ClusterCheck:            config.ClusterCheck,
+		NodeName:                config.NodeName,
+		Source:                  config.Source,
+		IgnoreAutodiscoveryTags: config.IgnoreAutodiscoveryTags,
+		MetricsExcluded:         config.MetricsExcluded,
+		LogsExcluded:            config.LogsExcluded,
+	}
+}