Merge 305a864 into c3e5646

cmd/agent/subcommands/run/command.go

@@ -214,7 +214,7 @@ func run(log log.Component,
 	collector collector.Component,
 ) error {
 	defer func() {
-		stopAgent(cliParams, server, agentAPI)
+		stopAgent(cliParams, agentAPI)
 	}()
 
 	// prepare go runtime
@@ -332,7 +332,7 @@ func getSharedFxOption() fx.Option {
 		statusimpl.Module(),
 		authtokenimpl.Module(),
 		apiimpl.Module(),
-
+		demultiplexerimpl.Module(),
 		dogstatsd.Bundle(),
 		otelcol.Bundle(),
 		rctelemetryreporterimpl.Module(),
@@ -368,7 +368,6 @@ func getSharedFxOption() fx.Option {
 			params.EnableNoAggregationPipeline = config.GetBool("dogstatsd_no_aggregation_pipeline")
 			return params
 		}),
-		demultiplexerimpl.Module(),
 		orchestratorForwarderImpl.Module(),
 		fx.Supply(orchestratorForwarderImpl.NewDefaultParams()),
 		eventplatformimpl.Module(),
@@ -596,16 +595,6 @@ func startAgent(
 
 	demultiplexer.AddAgentStartupTelemetry(version.AgentVersion)
 
-	// start dogstatsd
-	if pkgconfig.Datadog.GetBool("use_dogstatsd") {
-		err := server.Start(demultiplexer)
-		if err != nil {
-			log.Errorf("Could not start dogstatsd: %s", err)
-		} else {
-			log.Debugf("dogstatsd started")
-		}
-	}
-
 	// load and run all configs in AD
 	common.AC.LoadAndRun(ctx)
 
@@ -634,12 +623,12 @@ func startAgent(
 }
 
 // StopAgentWithDefaults is a temporary way for other packages to use stopAgent.
-func StopAgentWithDefaults(server dogstatsdServer.Component, agentAPI internalAPI.Component) {
-	stopAgent(&cliParams{GlobalParams: &command.GlobalParams{}}, server, agentAPI)
+func StopAgentWithDefaults(agentAPI internalAPI.Component) {
+	stopAgent(&cliParams{GlobalParams: &command.GlobalParams{}}, agentAPI)
 }
 
 // stopAgent Tears down the agent process
-func stopAgent(cliParams *cliParams, server dogstatsdServer.Component, agentAPI internalAPI.Component) {
+func stopAgent(cliParams *cliParams, agentAPI internalAPI.Component) {
 	// retrieve the agent health before stopping the components
 	// GetReadyNonBlocking has a 100ms timeout to avoid blocking
 	health, err := health.GetReadyNonBlocking()
@@ -654,7 +643,6 @@ func stopAgent(cliParams *cliParams, server dogstatsdServer.Component, agentAPI
 			pkglog.Errorf("Error shutting down expvar server: %v", err)
 		}
 	}
-	server.Stop()
 	if common.AC != nil {
 		common.AC.Stop()
 	}

cmd/agent/subcommands/run/command_windows.go

@@ -112,7 +112,7 @@ func StartAgentWithDefaults(ctxChan <-chan context.Context) (<-chan error, error
 			collector collector.Component,
 		) error {
 
-			defer StopAgentWithDefaults(server, agentAPI)
+			defer StopAgentWithDefaults(agentAPI)
 
 			err := startAgent(
 				&cliParams{GlobalParams: &command.GlobalParams{}},

cmd/agent/subcommands/run/internal/settings/runtime_settings_test.go

@@ -9,7 +9,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 	"go.uber.org/fx"
 
 	"github.com/DataDog/datadog-agent/comp/aggregator/demultiplexer"
@@ -41,16 +40,11 @@ func TestDogstatsdMetricsStats(t *testing.T) {
 		fx.Supply(server.Params{
 			Serverless: false,
 		}),
+		demultiplexerimpl.MockModule(),
 		dogstatsd.Bundle(),
 		defaultforwarder.MockModule(),
-		demultiplexerimpl.MockModule(),
 	))
 
-	demux := deps.Demultiplexer
-	deps.Server.Start(demux)
-
-	require.Nil(t, err)
-
 	s := DsdStatsRuntimeSetting{
 		ServerDebug: deps.Debug,
 	}

cmd/dogstatsd/subcommands/start/command.go

@@ -276,11 +276,6 @@ func RunDogstatsd(ctx context.Context, cliParams *CLIParams, config config.Compo
 
 	demultiplexer.AddAgentStartupTelemetry(version.AgentVersion)
 
-	err = components.DogstatsdServer.Start(demultiplexer)
-	if err != nil {
-		log.Criticalf("Unable to start dogstatsd: %s", err)
-		return
-	}
 	return
 }
 
@@ -325,8 +320,6 @@ func StopAgent(cancel context.CancelFunc, components *DogstatsdComponents) {
 		}
 	}
 
-	components.DogstatsdServer.Stop()
-
 	pkglog.Info("See ya!")
 	pkglog.Flush()
 }

comp/aggregator/demultiplexer/demultiplexerimpl/demultiplexer.go

@@ -59,6 +59,7 @@ type provides struct {
 	DiagnosticSenderManager diagnosesendermanager.Component
 	SenderManager           sender.SenderManager
 	StatusProvider          status.InformationProvider
+	AggregatorDemultiplexer aggregator.Demultiplexer
 }
 
 func newDemultiplexer(deps dependencies) (provides, error) {
@@ -94,6 +95,7 @@ func newDemultiplexer(deps dependencies) (provides, error) {
 		StatusProvider: status.NewInformationProvider(demultiplexerStatus{
 			Log: deps.Log,
 		}),
+		AggregatorDemultiplexer: demultiplexer,
 	}, nil
 }
 

comp/aggregator/demultiplexer/demultiplexerimpl/demultiplexer_fake_sampler_mock.go

@@ -22,7 +22,11 @@ import (
 // FakeSamplerMockModule defines the fx options for FakeSamplerMock.
 func FakeSamplerMockModule() fxutil.Module {
 	return fxutil.Component(
-		fx.Provide(newFakeSamplerMock))
+		fx.Provide(newFakeSamplerMock),
+		fx.Provide(func(demux demultiplexerComp.FakeSamplerMock) aggregator.Demultiplexer {
+			return demux
+		}),
+	)
 }
 
 type fakeSamplerMockDependencies struct {

comp/aggregator/demultiplexer/demultiplexerimpl/demultiplexer_mock.go

@@ -22,7 +22,11 @@ import (
 // MockModule defines the fx options for this component.
 func MockModule() fxutil.Module {
 	return fxutil.Component(
-		fx.Provide(newMock))
+		fx.Provide(newMock),
+		fx.Provide(func(demux demultiplexerComp.Component) aggregator.Demultiplexer {
+			return demux
+		}),
+	)
 }
 
 type mock struct {

comp/dogstatsd/server/component.go

@@ -9,7 +9,6 @@ package server
 import (
 	"time"
 
-	"github.com/DataDog/datadog-agent/pkg/aggregator"
 	"github.com/DataDog/datadog-agent/pkg/util/fxutil"
 	"go.uber.org/fx"
 )
@@ -18,13 +17,6 @@ import (
 
 // Component is the component type.
 type Component interface {
-
-	// Start starts the dogstatsd server
-	Start(demultiplexer aggregator.Demultiplexer) error
-
-	// Stop stops the dogstatsd server
-	Stop()
-
 	// IsRunning returns true if the server is running
 	IsRunning() bool
 

comp/dogstatsd/server/server.go

@@ -19,13 +19,11 @@ import (
 
 	configComponent "github.com/DataDog/datadog-agent/comp/core/config"
 	logComponent "github.com/DataDog/datadog-agent/comp/core/log"
-	logComponentImpl "github.com/DataDog/datadog-agent/comp/core/log/logimpl"
 	"github.com/DataDog/datadog-agent/comp/dogstatsd/listeners"
 	"github.com/DataDog/datadog-agent/comp/dogstatsd/mapper"
 	"github.com/DataDog/datadog-agent/comp/dogstatsd/packets"
 	"github.com/DataDog/datadog-agent/comp/dogstatsd/replay"
 	serverdebug "github.com/DataDog/datadog-agent/comp/dogstatsd/serverDebug"
-	"github.com/DataDog/datadog-agent/comp/dogstatsd/serverDebug/serverdebugimpl"
 	"github.com/DataDog/datadog-agent/pkg/aggregator"
 	"github.com/DataDog/datadog-agent/pkg/config"
 	"github.com/DataDog/datadog-agent/pkg/metrics"
@@ -67,6 +65,10 @@ var (
 type dependencies struct {
 	fx.In
 
+	Lc fx.Lifecycle
+
+	Demultiplexer aggregator.Demultiplexer
+
 	Log    logComponent.Component
 	Config configComponent.Component
 	Debug  serverdebug.Component
@@ -187,19 +189,22 @@ func initTelemetry(cfg config.Reader, logger logComponent.Component) {
 	packets.InitTelemetry(get("telemetry.dogstatsd.listeners_channel_latency_buckets"))
 }
 
-// TODO: (components) - remove once serverless is an FX app
-//
-//nolint:revive // TODO(AML) Fix revive linter
-func NewServerlessServer() Component {
-	return newServerCompat(config.Datadog, logComponentImpl.NewTemporaryLoggerWithoutInit(), replay.NewServerlessTrafficCapture(), serverdebugimpl.NewServerlessServerDebug(), true)
-}
-
 // TODO: (components) - merge with newServerCompat once NewServerlessServer is removed
 func newServer(deps dependencies) Component {
-	return newServerCompat(deps.Config, deps.Log, deps.Replay, deps.Debug, deps.Params.Serverless)
+	s := newServerCompat(deps.Config, deps.Log, deps.Replay, deps.Debug, deps.Params.Serverless, deps.Demultiplexer)
+
+	if config.Datadog.GetBool("use_dogstatsd") {
+		deps.Lc.Append(fx.Hook{
+			OnStart: s.startHook,
+			OnStop:  s.stop,
+		})
+	}
+
+	return s
+
 }
 
-func newServerCompat(cfg config.Reader, log logComponent.Component, capture replay.Component, debug serverdebug.Component, serverless bool) Component {
+func newServerCompat(cfg config.Reader, log logComponent.Component, capture replay.Component, debug serverdebug.Component, serverless bool, demux aggregator.Demultiplexer) *server {
 	// This needs to be done after the configuration is loaded
 	once.Do(func() { initTelemetry(cfg, log) })
 
@@ -272,7 +277,7 @@ func newServerCompat(cfg config.Reader, log logComponent.Component, capture repl
 		sharedPacketPool:        nil,
 		sharedPacketPoolManager: nil,
 		sharedFloat64List:       newFloat64ListPool(),
-		demultiplexer:           nil,
+		demultiplexer:           demux,
 		listeners:               nil,
 		stopChan:                make(chan bool),
 		serverlessFlushChan:     make(chan bool),
@@ -301,13 +306,22 @@ func newServerCompat(cfg config.Reader, log logComponent.Component, capture repl
 			originOptOutEnabled:       cfg.GetBool("dogstatsd_origin_optout_enabled"),
 		},
 	}
+
 	return s
 }
 
-func (s *server) Start(demultiplexer aggregator.Demultiplexer) error {
+func (s *server) startHook(context context.Context) error {
 
-	// TODO: (components) - DI this into Server when Demultiplexer is made into a component
-	s.demultiplexer = demultiplexer
+	err := s.start(context)
+	if err != nil {
+		s.log.Errorf("Could not start dogstatsd: %s", err)
+	} else {
+		s.log.Debug("dogstatsd started")
+	}
+	return nil
+}
+
+func (s *server) start(context.Context) error {
 
 	packetsChannel := make(chan packets.Packets, s.config.GetInt("dogstatsd_queue_size"))
 	tmpListeners := make([]listeners.StatsdListener, 0, 2)
@@ -444,9 +458,9 @@ func (s *server) Start(demultiplexer aggregator.Demultiplexer) error {
 	return nil
 }
 
-func (s *server) Stop() {
+func (s *server) stop(context.Context) error {
 	if !s.IsRunning() {
-		return
+		return nil
 	}
 	close(s.stopChan)
 	for _, l := range s.listeners {
@@ -460,6 +474,8 @@ func (s *server) Stop() {
 	}
 	s.health.Deregister() //nolint:errcheck
 	s.Started = false
+
+	return nil
 }
 
 func (s *server) IsRunning() bool {

comp/dogstatsd/server/server_bench_test.go

@@ -37,8 +37,7 @@ func benchParsePackets(b *testing.B, rawPacket []byte) {
 	pkgconfig.SetupLogger("", "off", "", "", false, true, false)
 
 	demux := deps.Demultiplexer
-	_ = s.Start(demux)
-	defer s.Stop()
+	defer demux.Stop(false)
 
 	done := make(chan struct{})
 	go func() {
@@ -85,8 +84,6 @@ func BenchmarkPbarseMetricMessage(b *testing.B) {
 	pkgconfig.SetupLogger("", "off", "", "", false, true, false)
 
 	demux := deps.Demultiplexer
-	_ = s.Start(demux)
-	defer s.Stop()
 
 	done := make(chan struct{})
 	go func() {
@@ -138,8 +135,6 @@ func benchmarkMapperControl(b *testing.B, yaml string) {
 	pkgconfig.SetupLogger("", "off", "", "", false, true, false)
 
 	demux := deps.Demultiplexer
-	_ = s.Start(demux)
-	defer s.Stop()
 
 	done := make(chan struct{})
 	go func() {