agentcomponents.go

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

// github.com/DataDog/datadog-agent/comp/core/config is not suppported on AIX
//go:build !aix

package agentcomponents // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/datadog/agentcomponents"

import (
	"runtime"
	"strings"

	coreconfig "github.com/DataDog/datadog-agent/comp/core/config"
	corelog "github.com/DataDog/datadog-agent/comp/core/log/def"
	"github.com/DataDog/datadog-agent/comp/forwarder/defaultforwarder"
	logsconfig "github.com/DataDog/datadog-agent/comp/logs/agent/config"
	pkgconfigcreate "github.com/DataDog/datadog-agent/pkg/config/create"
	pkgconfigmodel "github.com/DataDog/datadog-agent/pkg/config/model"
	pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
	pkgconfigutils "github.com/DataDog/datadog-agent/pkg/config/utils"
	"github.com/DataDog/datadog-agent/pkg/metrics"
	"github.com/DataDog/datadog-agent/pkg/serializer"
	zlib "github.com/DataDog/datadog-agent/pkg/util/compression/impl-zlib"
	"go.opentelemetry.io/collector/component"
	"golang.org/x/net/http/httpproxy"

	datadogconfig "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/datadog/config"
)

// ConfigOption is a function that configures the Datadog agent config component.
// This allows for flexible configuration by different modules.
type ConfigOption func(pkgconfigmodel.Config)

// SerializerWithForwarder is an interface that extends the MetricSerializer interface
// with ability to interact directly with the underlying forwarder's lifecycle methods.
type SerializerWithForwarder interface {
	serializer.MetricSerializer
	Start() error
	State() uint32
	Stop()
	// SendSeriesWithMetadata sends a metrics series to Datadog
	SendSeriesWithMetadata(series metrics.Series) error
}

// forwarderWithLifecycle extends the defaultforwarder.Forwarder interface
// with lifecycle management methods
type forwarderWithLifecycle interface {
	defaultforwarder.Forwarder
	Start() error
	State() uint32
	Stop()
}

// Compile-time check to ensure DefaultForwarder implements ForwarderWithLifecycle
var _ forwarderWithLifecycle = (*defaultforwarder.DefaultForwarder)(nil)

// Compile-time check to ensure datadogSerializer implements SerializerWithForwarder
var _ SerializerWithForwarder = (*datadogSerializer)(nil)

// datadogSerializer is a concrete implementation of SerializerWithForwarder that wraps
// a MetricSerializer and provides access to the underlying forwarder's lifecycle methods
type datadogSerializer struct {
	serializer.MetricSerializer
	forwarder forwarderWithLifecycle
}

// Start delegates to the underlying forwarder's Start method
func (ds *datadogSerializer) Start() error {
	return ds.forwarder.Start()
}

// State delegates to the underlying forwarder's State method
func (ds *datadogSerializer) State() uint32 {
	return ds.forwarder.State()
}

// Stop delegates to the underlying forwarder's Stop method
func (ds *datadogSerializer) Stop() {
	ds.forwarder.Stop()
}

// seriesSource wraps a metrics.Series to implement the metrics.SerieSource interface
type seriesSource struct {
	series metrics.Series
	index  int
}

// MoveNext moves to the next serie in the collection
func (s *seriesSource) MoveNext() bool {
	s.index++
	return s.index < len(s.series)
}

// Current returns the current serie
func (s *seriesSource) Current() *metrics.Serie {
	if s.index < 0 || s.index >= len(s.series) {
		return nil
	}
	return s.series[s.index]
}

// Count returns the total number of series
func (s *seriesSource) Count() uint64 {
	return uint64(len(s.series))
}

// SendSeriesWithMetadata sends a metrics series to Datadog
// This method wraps the series in a SerieSource and uses the serializer's SendIterableSeries
func (ds *datadogSerializer) SendSeriesWithMetadata(series metrics.Series) error {
	if len(series) == 0 {
		return nil
	}
	// Wrap the series in a SerieSource
	source := &seriesSource{
		series: series,
		index:  -1, // Start before the first element
	}
	return ds.SendIterableSeries(source)
}

// NewLogComponent creates a new log component for collector that uses the provided telemetry settings.
func NewLogComponent(set component.TelemetrySettings) corelog.Component {
	zlog := &ZapLogger{
		Logger: set.Logger,
	}
	return zlog
}

// NewSerializerComponent creates a new serializer that serializes and compresses payloads prior to being forwarded
func NewSerializerComponent(cfg coreconfig.Component, logger corelog.Component, hostname string) SerializerWithForwarder {
	forwarder := newForwarderComponent(cfg, logger)
	compressor := zlib.New()
	metricSerializer := serializer.NewSerializer(forwarder, nil, compressor, cfg, logger, hostname)

	return &datadogSerializer{
		MetricSerializer: metricSerializer,
		forwarder:        forwarder,
	}
}

// NewConfigComponent creates a new Datadog agent config component with the given options.
// This function uses the options pattern to allow different modules to configure
// the component with their specific needs.
func NewConfigComponent(options ...ConfigOption) coreconfig.Component {
	pkgconfig := pkgconfigcreate.NewConfig("DD")

	// Register all standard agent config keys so that subsequent Set calls work
	// correctly with the nodetreemodel config backend.
	pkgconfigsetup.InitConfig(pkgconfig)

	// Apply all configuration options
	for _, opt := range options {
		opt(pkgconfig)
	}

	// Mark the config as ready for use. Without this, nodetreemodel silently
	// returns zero values for all Get* calls (reads are blocked before BuildSchema).
	pkgconfig.BuildSchema()

	return pkgconfig
}

// WithAPIConfig configures API-related settings
func WithAPIConfig(cfg *datadogconfig.Config) ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("api_key", string(cfg.API.Key), pkgconfigmodel.SourceFile)
		pkgconfig.Set("site", cfg.API.Site, pkgconfigmodel.SourceFile)
	}
}

// WithForwarderConfig configures forwarder-related settings
func WithForwarderConfig() ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("forwarder_apikey_validation_interval", 60, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("forwarder_num_workers", 1, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("forwarder_backoff_factor", 2, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("forwarder_backoff_base", 2, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("forwarder_backoff_max", 64, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("forwarder_recovery_interval", 2, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("forwarder_http_protocol", "auto", pkgconfigmodel.SourceDefault)
		pkgconfig.Set("forwarder_max_concurrent_requests", 1, pkgconfigmodel.SourceDefault)
	}
}

// WithCustomConfig sets an arbitrary key in the Datadog agent config.
func WithCustomConfig(key string, value any, source pkgconfigmodel.Source) ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set(key, value, source)
	}
}

// WithLoggingConfig sets logging_frequency to avoid a divide-by-zero panic in the agent logger.
func WithLoggingConfig() ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("logging_frequency", 1, pkgconfigmodel.SourceDefault)
	}
}

// WithLogsEnabled enables logs for agent config
func WithLogsEnabled() ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("logs_enabled", true, pkgconfigmodel.SourceAgentRuntime)
	}
}

// WithLogsConfig configures logs-related settings (requires WithLogsEnabled)
func WithLogsConfig(cfg *datadogconfig.Config) ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("logs_config.batch_wait", cfg.Logs.BatchWait, pkgconfigmodel.SourceFile)
		pkgconfig.Set("logs_config.use_compression", cfg.Logs.UseCompression, pkgconfigmodel.SourceFile)
		pkgconfig.Set("logs_config.compression_level", cfg.Logs.CompressionLevel, pkgconfigmodel.SourceFile)
		pkgconfig.Set("logs_config.logs_dd_url", cfg.Logs.Endpoint, pkgconfigmodel.SourceFile)
	}
}

// WithLogsDefaults configures logs default settings (requires WithLogsEnabled)
func WithLogsDefaults() ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("logs_config.auditor_ttl", pkgconfigsetup.DefaultAuditorTTL, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.batch_max_content_size", pkgconfigsetup.DefaultBatchMaxContentSize, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.batch_max_size", pkgconfigsetup.DefaultBatchMaxSize, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.compression_kind", logsconfig.GzipCompressionKind, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.force_use_http", true, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.input_chan_size", pkgconfigsetup.DefaultInputChanSize, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.max_message_size_bytes", pkgconfigsetup.DefaultMaxMessageSizeBytes, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.run_path", "/opt/datadog-agent/run", pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.sender_backoff_factor", pkgconfigsetup.DefaultLogsSenderBackoffFactor, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.sender_backoff_base", pkgconfigsetup.DefaultLogsSenderBackoffBase, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.sender_backoff_max", pkgconfigsetup.DefaultLogsSenderBackoffMax, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.sender_recovery_interval", pkgconfigsetup.DefaultForwarderRecoveryInterval, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.stop_grace_period", 30, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("logs_config.use_v2_api", true, pkgconfigmodel.SourceDefault)
		// add logs config pipelines config value, see https://github.com/DataDog/datadog-agent/pull/31190
		logsPipelines := min(4, runtime.GOMAXPROCS(0))
		pkgconfig.Set("logs_config.pipelines", logsPipelines, pkgconfigmodel.SourceDefault)
	}
}

// WithLogLevel configures log level settings (requires WithLogsEnabled)
func WithLogLevel(set component.TelemetrySettings) ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("log_level", set.Logger.Level().String(), pkgconfigmodel.SourceFile)
	}
}

// WithPayloadsConfig configures payload settings
func WithPayloadsConfig() ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		pkgconfig.Set("enable_payloads.events", true, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("enable_payloads.json_to_v1_intake", true, pkgconfigmodel.SourceDefault)
		pkgconfig.Set("enable_sketch_stream_payload_serialization", true, pkgconfigmodel.SourceDefault)
	}
}

// WithProxy configures proxy settings from config or environment variables
func WithProxy(cfg *datadogconfig.Config) ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		setProxy(cfg, pkgconfig)
	}
}

// WithTLSSetting propagates tls.insecure_skip_verify into the DD-agent forwarder's TLS config
func WithTLSSetting(cfg *datadogconfig.Config) ConfigOption {
	return func(pkgconfig pkgconfigmodel.Config) {
		setTLSSetting(cfg, pkgconfig)
	}
}

func setProxy(cfg *datadogconfig.Config, pkgconfig pkgconfigmodel.Config) {
	proxyConfig := httpproxy.FromEnvironment()
	if proxyConfig.HTTPProxy != "" {
		pkgconfig.Set("proxy.http", proxyConfig.HTTPProxy, pkgconfigmodel.SourceDefault)
	}
	if proxyConfig.HTTPSProxy != "" {
		pkgconfig.Set("proxy.https", proxyConfig.HTTPSProxy, pkgconfigmodel.SourceDefault)
	}

	// proxy_url takes precedence over proxy environment variables if set
	if cfg.ClientConfig.ProxyURL != "" {
		pkgconfig.Set("proxy.http", cfg.ClientConfig.ProxyURL, pkgconfigmodel.SourceFile)
		pkgconfig.Set("proxy.https", cfg.ClientConfig.ProxyURL, pkgconfigmodel.SourceFile)
	}

	// If this is set to an empty []string, viper will have a type conflict when merging
	// this config during secrets resolution. It unmarshals empty yaml lists to type
	// []any, which will then conflict with type []string and fail to merge.
	var noProxy []any
	for v := range strings.SplitSeq(proxyConfig.NoProxy, ",") {
		noProxy = append(noProxy, v)
	}
	// Use SourceAgentRuntime because nodetreemodel's BuildSchema rebuilds the SourceEnvVar layer
	// from actual DD_* env vars, which would erase a manually-set SourceEnvVar value.
	pkgconfig.Set("proxy.no_proxy", noProxy, pkgconfigmodel.SourceAgentRuntime)
}

func setTLSSetting(cfg *datadogconfig.Config, pkgconfig pkgconfigmodel.Config) {
	if cfg.ClientConfig.TLS.InsecureSkipVerify {
		pkgconfig.Set("skip_ssl_validation", cfg.ClientConfig.TLS.InsecureSkipVerify, pkgconfigmodel.SourceFile)
	}
}

// newForwarderComponent creates a new forwarder that sends payloads to Datadog backend
func newForwarderComponent(cfg coreconfig.Component, log corelog.Component) forwarderWithLifecycle {
	keysPerDomain := map[string][]pkgconfigutils.APIKeys{
		"https://api." + cfg.GetString("site"): {pkgconfigutils.NewAPIKeys("api_key", cfg.GetString("api_key"))},
	}
	forwarderOptions, _ := defaultforwarder.NewOptions(cfg, log, keysPerDomain)
	forwarderOptions.DisableAPIKeyChecking = true
	return defaultforwarder.NewDefaultForwarder(cfg, log, forwarderOptions)
}