o11y-foundry/internal/molding/ingestermolding/ingester.go

package ingestermolding

import (
	"bytes"
	"context"
	"fmt"
	"log/slog"
	"strings"

	"github.com/hanzoai/o11y-foundry/api/v1alpha1"
	"github.com/hanzoai/o11y-foundry/api/v1alpha1/installation"
	foundryerrors "github.com/hanzoai/o11y-foundry/internal/errors"
	"github.com/hanzoai/o11y-foundry/internal/molding"
)

var _ molding.Molding = (*ingester)(nil)

type ingester struct {
	logger *slog.Logger
}

func New(logger *slog.Logger) *ingester {
	return &ingester{
		logger: logger,
	}
}

func (molding *ingester) Kind() v1alpha1.MoldingKind {
	return v1alpha1.MoldingKindIngester
}

func (molding *ingester) MoldV1Alpha1(ctx context.Context, config *installation.Casting) error {
	// render the template for config.yaml
	data, err := molding.getData(config)
	if err != nil {
		molding.logger.ErrorContext(ctx, "failed to get data", foundryerrors.LogAttr(err))
		return err
	}

	configBuf := bytes.NewBuffer(nil)
	if err := ConfigV0129xTemplate.Execute(configBuf, data); err != nil {
		return err
	}

	opampBuf := bytes.NewBuffer(nil)
	if err := OpampV0129xTemplate.Execute(opampBuf, data); err != nil {
		return err
	}

	config.Spec.Ingester.Status.Config.Data = map[string]string{
		"ingester.yaml": configBuf.String(),
		"opamp.yaml":    opampBuf.String(),
	}

	if config.Spec.Ingester.Status.Env == nil {
		config.Spec.Ingester.Status.Env = make(map[string]string)
	}
	config.Spec.Ingester.Status.Env["SIGNOZ_OTEL_COLLECTOR_TIMEOUT"] = "10m"

	return nil
}

func (molding *ingester) getData(config *installation.Casting) (Data, error) {
	if len(config.Spec.O11y.Status.Addresses.Opamp) == 0 {
		return Data{}, fmt.Errorf("o11y address is not set")
	}

	o11yAddress := config.Spec.O11y.Status.Addresses.Opamp[0]

	if len(config.Spec.TelemetryStore.Status.Addresses.TCP) == 0 {
		return Data{}, foundryerrors.Newf(foundryerrors.TypeInternal, "telemetry store address is not set")
	}

	telemetryStoreAddresses := config.Spec.TelemetryStore.Status.Addresses.TCP
	var telemetryStoreTracesAddresses []string
	for _, address := range telemetryStoreAddresses {
		telemetryStoreTracesAddresses = append(telemetryStoreTracesAddresses, address+"/o11y_traces")
	}

	var telemetryStoreMetricsAddresses []string
	for _, address := range telemetryStoreAddresses {
		telemetryStoreMetricsAddresses = append(telemetryStoreMetricsAddresses, address+"/o11y_metrics")
	}

	var telemetryStoreLogsAddresses []string
	for _, address := range telemetryStoreAddresses {
		telemetryStoreLogsAddresses = append(telemetryStoreLogsAddresses, address+"/o11y_logs")
	}

	var telemetryStoreMeterAddresses []string
	for _, address := range telemetryStoreAddresses {
		telemetryStoreMeterAddresses = append(telemetryStoreMeterAddresses, address+"/o11y_meter")
	}

	return Data{
		O11yOpampAddress:             o11yAddress,
		TelemetryStoreTracesAddress:  strings.Join(telemetryStoreTracesAddresses, ","),
		TelemetryStoreMetricsAddress: strings.Join(telemetryStoreMetricsAddresses, ","),
		TelemetryStoreLogsAddress:    strings.Join(telemetryStoreLogsAddresses, ","),
		TelemetryStoreMeterAddress:   strings.Join(telemetryStoreMeterAddresses, ","),
	}, nil
}