beatconverter.go

// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
// or more contributor license agreements. Licensed under the Elastic License;
// you may not use this file except in compliance with the Elastic License.

package beatconverter

import (
	"context"
	"fmt"

	"go.opentelemetry.io/collector/confmap"

	"github.com/elastic/beats/v7/libbeat/cloudid"
	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
	elasticsearchtranslate "github.com/elastic/beats/v7/x-pack/otel/oteltranslate/outputs/elasticsearch"
	logstashstranslate "github.com/elastic/beats/v7/x-pack/otel/oteltranslate/outputs/logstash"
	"github.com/elastic/elastic-agent-libs/config"
	"github.com/elastic/elastic-agent-libs/logp"
)

// list of supported beatreceivers
var (
	supportedReceivers = []string{"filebeatreceiver", "metricbeatreceiver"} // Add more beat receivers to this list when we add support
	beatsAuthName      = "beatsauth"
)

type Converter struct{}

// NewFactory returns a factory for a  confmap.Converter,
func NewFactory() confmap.ConverterFactory {
	return confmap.NewConverterFactory(newConverter)
}

func newConverter(set confmap.ConverterSettings) confmap.Converter {
	return Converter{}
}

// Convert converts [beatreceiver].output to OTel config here
func (c Converter) Convert(_ context.Context, conf *confmap.Conf) error {

	for _, beatreceiver := range supportedReceivers {
		var out map[string]any

		var beatReceiverConfigKey = "receivers::" + beatreceiver
		// check if supported beat receiver is configured. Skip translation logic if not
		if v := conf.Get(beatReceiverConfigKey); v == nil {
			continue
		}

		// handle cloud id if set
		if conf.IsSet(beatReceiverConfigKey + "::cloud") {
			if err := handleCloudId(beatreceiver, conf); err != nil {
				return fmt.Errorf("error handling cloud id %w", err)
			}
		}

		receiverCfg, _ := conf.Sub(beatReceiverConfigKey)
		output, _ := receiverCfg.Sub("output")

		if len(output.ToStringMap()) > 1 {
			return fmt.Errorf("multiple outputs are not supported")
		}

		for key, output := range output.ToStringMap() {
			switch key {
			case "logstash":
				lsOutputConfig := config.MustNewConfigFrom(output)

				// ignore logstash output if it is not enabled
				if !lsOutputConfig.Enabled() {
					continue
				}

				// when output.queue is set by user, promote it to global level
				if ok := lsOutputConfig.HasField("queue"); ok {
					if err := promoteOutputQueueSettings(beatreceiver, lsOutputConfig, conf); err != nil {
						return err
					}
					// remove queue from logstash output config
					if _, err := lsOutputConfig.Remove("queue", -1); err != nil {
						return err
					}
				}

				lsConfigMap, err := logstashstranslate.ToMap(lsOutputConfig)
				if err != nil {
					return err
				}

				out = map[string]any{
					"service::pipelines::logs::exporters": []string{"logstash"},
					"exporters": map[string]any{
						"logstash": lsConfigMap,
					},
				}

				if err := conf.Merge(confmap.NewFromStringMap(out)); err != nil {
					return err
				}

			case "elasticsearch":
				esConfig := config.MustNewConfigFrom(output)

				// ignore elasticsearch output if it is not enabled
				if !esConfig.Enabled() {
					continue
				}

				// we use development logger here as this method is part of dev-only otel command
				logger, _ := logp.NewDevelopmentLogger("")
				esOTelConfig, err := elasticsearchtranslate.ToOTelConfig(esConfig, logger)
				if err != nil {
					return fmt.Errorf("cannot convert elasticsearch config: %w", err)
				}

				// when output.queue is set by user or it comes from "preset" config, promote it to global level
				if ok := esConfig.HasField("queue"); ok {
					if err := promoteOutputQueueSettings(beatreceiver, esConfig, conf); err != nil {
						return err
					}
				}

				// get beatsauth config
				authConfig, err := getBeatsAuthExtensionConfig(esConfig)
				if err != nil {
					return fmt.Errorf("cannot translate http settings on beatsauth extension: %w", err)
				}

				// set authenticator name on ES exporter
				esOTelConfig["auth"] = map[string]any{
					"authenticator": beatsAuthName,
				}

				out = map[string]any{
					"service::pipelines::logs::exporters": []string{"elasticsearch"},
					"service::extensions":                 []interface{}{beatsAuthName},
					"exporters": map[string]any{
						"elasticsearch": esOTelConfig,
					},
					"extensions": map[string]any{
						beatsAuthName: authConfig,
					},
				}
				err = conf.Merge(confmap.NewFromStringMap(out))
				if err != nil {
					return err
				}
			// noop, it will get replaced by otelconsumer below
			case "otelconsumer":
			default:
				return fmt.Errorf("output type %q is unsupported in OTel mode", key)
			}
		}

		// Replace output.[configured-output] with output.otelconsumer
		out = map[string]any{
			beatReceiverConfigKey + "::output": nil,
		}
		err := conf.Merge(confmap.NewFromStringMap(out))
		if err != nil {
			return err
		}
		out = map[string]any{
			beatReceiverConfigKey + "::output::otelconsumer": nil,
		}

		// inject log level
		receiverConfig, err := config.NewConfigFrom(receiverCfg.ToStringMap())
		if err != nil {
			return fmt.Errorf("error getting receiver config: %w", err)
		}

		if level, _ := receiverConfig.String("logging.level", -1); level != "" {
			out["service::telemetry::logs::level"], err = getOTelLogLevel(level)
			if err != nil {
				return fmt.Errorf("error injecting log level: %w", err)

			}
		}
		// We're disabling the metrics telemetry for beat receivers for now because the default agent also uses the same port.
		// If needed we can make this configurable later.
		out["service::telemetry::metrics::level"] = "none"

		err = conf.Merge(confmap.NewFromStringMap(out))
		if err != nil {
			return err
		}
	}

	return nil
}

func handleCloudId(beatReceiverConfigKey string, conf *confmap.Conf) error {

	receiverCfg, _ := conf.Sub("receivers::" + beatReceiverConfigKey)
	beatCfg := config.MustNewConfigFrom(receiverCfg.ToStringMap())

	// Handle cloud.id the same way Beats does, this will also handle
	// extracting the Kibana URL
	if err := cloudid.OverwriteSettings(beatCfg); err != nil {
		return fmt.Errorf("cannot read cloudid: %w", err)
	}

	var beatOutput map[string]any
	err := beatCfg.Unpack(&beatOutput)
	if err != nil {
		return err
	}

	out := map[string]any{
		"receivers::" + beatReceiverConfigKey: beatOutput,
	}
	err = conf.Merge(confmap.NewFromStringMap(out))
	if err != nil {
		return err
	}

	// we set this to nil to ensure cloudid check does not throw error when output is next set to otelconsumer
	out = map[string]any{
		"receivers::" + beatReceiverConfigKey + "::cloud": nil,
	}
	err = conf.Merge(confmap.NewFromStringMap(out))
	if err != nil {
		return err
	}

	return nil
}

// promoteOutputQueueSettings promotes output.queue settings to global level
func promoteOutputQueueSettings(beatReceiverConfigKey string, outputConfig *config.C, conf *confmap.Conf) error {

	var queueOutput map[string]any
	err := outputConfig.Unpack(&queueOutput)
	if err != nil {
		return err
	}
	out := map[string]any{
		"receivers::" + beatReceiverConfigKey + "::queue": queueOutput["queue"],
	}
	err = conf.Merge(confmap.NewFromStringMap(out))
	if err != nil {
		return err
	}

	return nil
}

// getBeatsAuthExtensionConfig sets http transport settings on beatsauth
// currently this is only supported for elasticsearch output
func getBeatsAuthExtensionConfig(cfg *config.C) (map[string]any, error) {
	defaultTransportSettings := elasticsearch.ESDefaultTransportSettings()
	err := cfg.Unpack(&defaultTransportSettings)
	if err != nil {
		return nil, err
	}

	newConfig, err := config.NewConfigFrom(defaultTransportSettings)
	if err != nil {
		return nil, err
	}

	// proxy_url on newConfig is of type *url.URL which is not understood by beatsauth extension
	// this logic here converts it into string type similar to what a user would set on filebeat config
	if defaultTransportSettings.Proxy.URL != nil {
		proxyURL, err := config.NewConfigFrom(map[string]any{
			"proxy_url": defaultTransportSettings.Proxy.URL.String(),
		})
		if err != nil {
			return nil, fmt.Errorf("error translating proxy_url: %w", err)
		}
		err = newConfig.Merge(proxyURL)
		if err != nil {
			return nil, fmt.Errorf("error merging proxy_url: %w", err)
		}
	}

	var newMap map[string]any
	err = newConfig.Unpack(&newMap)
	if err != nil {
		return nil, err
	}

	return newMap, nil
}