telemetry/exporter/exporterhelper/internal/base_exporter_test.go

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

package internal

import (
	"context"
	"errors"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/sdk/metric/metricdata"
	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"

	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/component/componenttest"
	"go.opentelemetry.io/collector/config/configoptional"
	"go.opentelemetry.io/collector/config/configretry"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/metadatatest"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
	"go.opentelemetry.io/collector/exporter/exportertest"
	"go.opentelemetry.io/collector/pipeline"
)

func TestBaseExporter(t *testing.T) {
	be, err := NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport)
	require.NoError(t, err)
	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
	require.NoError(t, be.Shutdown(context.Background()))
}

func TestBaseExporterWithOptions(t *testing.T) {
	want := errors.New("my error")
	be, err := NewBaseExporter(
		exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
		WithStart(func(context.Context, component.Host) error { return want }),
		WithShutdown(func(context.Context) error { return want }),
		WithTimeout(NewDefaultTimeoutConfig()),
	)
	require.NoError(t, err)
	require.Equal(t, want, be.Start(context.Background(), componenttest.NewNopHost()))
	require.Equal(t, want, be.Shutdown(context.Background()))
}

func TestQueueOptionsWithRequestExporter(t *testing.T) {
	bs, err := NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
		WithRetry(configretry.NewDefaultBackOffConfig()))
	require.NoError(t, err)
	require.Nil(t, bs.queueBatchSettings.Encoding)
	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
		WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(configoptional.Some(NewDefaultQueueConfig())))
	require.Error(t, err)

	qCfg := NewDefaultQueueConfig()
	storageID := component.NewID(component.MustNewType("test"))
	qCfg.StorageID = &storageID
	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
		WithQueueBatchSettings(newFakeQueueBatch()),
		WithRetry(configretry.NewDefaultBackOffConfig()),
		WithQueueBatch(configoptional.Some(qCfg), queuebatch.Settings[request.Request]{}))
	require.Error(t, err)
}

func TestBaseExporterLogging(t *testing.T) {
	set := exportertest.NewNopSettings(exportertest.NopType)
	logger, observed := observer.New(zap.DebugLevel)
	set.Logger = zap.New(logger)
	rCfg := configretry.NewDefaultBackOffConfig()
	rCfg.Enabled = false
	qCfg := NewDefaultQueueConfig()
	qCfg.WaitForResult = true
	bs, err := NewBaseExporter(set, pipeline.SignalMetrics, errExport,
		WithQueueBatchSettings(newFakeQueueBatch()),
		WithQueue(configoptional.Some(qCfg)),
		WithRetry(rCfg))
	require.NoError(t, err)
	require.NoError(t, bs.Start(context.Background(), componenttest.NewNopHost()))
	sendErr := bs.Send(context.Background(), &requesttest.FakeRequest{Items: 2})
	require.Error(t, sendErr)

	errorLogs := observed.FilterLevelExact(zap.ErrorLevel).All()
	require.Len(t, errorLogs, 2)
	assert.Contains(t, errorLogs[0].Message, "Exporting failed. Dropping data.")
	assert.Equal(t, "my error", errorLogs[0].ContextMap()["error"])
	assert.Contains(t, errorLogs[1].Message, "Exporting failed. Rejecting data.")
	assert.Equal(t, "my error", errorLogs[1].ContextMap()["error"])
	require.NoError(t, bs.Shutdown(context.Background()))
}

func TestBaseExporterExtraAttrs(t *testing.T) {
	set := exportertest.NewNopSettings(exportertest.NopType)
	tt := componenttest.NewTelemetry()
	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
	set.MeterProvider = tt.NewTelemetrySettings().MeterProvider

	rCfg := configretry.NewDefaultBackOffConfig()
	rCfg.Enabled = false
	qCfg := NewDefaultQueueConfig()
	qCfg.WaitForResult = true

	bs, err := NewBaseExporter(set, pipeline.SignalMetrics, errExport,
		WithQueueBatchSettings(newFakeQueueBatch()),
		WithQueue(configoptional.Some(qCfg)),
		WithRetry(rCfg),
		WithAttributes(attribute.String("test", "value"), attribute.Bool("other", true)),
	)
	require.NoError(t, err)
	require.NoError(t, bs.Start(context.Background(), componenttest.NewNopHost()))
	t.Cleanup(func() {
		require.NoError(t, bs.Shutdown(context.Background()))
	})
	sendErr := bs.Send(context.Background(), &requesttest.FakeRequest{Items: 2})
	require.Error(t, sendErr)
	metadatatest.AssertEqualExporterSentMetricPoints(t, tt, []metricdata.DataPoint[int64]{
		{
			Attributes: attribute.NewSet(
				attribute.String("exporter", set.ID.String()),
				attribute.String("test", "value"),
				attribute.Bool("other", true),
			),
			Value: int64(0),
		},
	}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
}

func TestWithQueue_MetadataKeys(t *testing.T) {
	t.Run("with MetadataKeys - configures partitioner and merge function", func(t *testing.T) {
		qCfg := NewDefaultQueueConfig()
		qCfg.Batch.GetOrInsertDefault().Partition.MetadataKeys = []string{"key1", "key2"}

		be, err := NewBaseExporter(
			exportertest.NewNopSettings(exportertest.NopType),
			pipeline.SignalMetrics,
			noopExport,
			WithQueueBatchSettings(newFakeQueueBatch()),
			WithQueue(configoptional.Some(qCfg)),
		)
		require.NoError(t, err)
		assert.NotNil(t, be)

		// Verify partitioner and merge function are configured
		assert.NotNil(t, be.queueBatchSettings.Partitioner, "Partitioner should be set when MetadataKeys is provided")
		assert.NotNil(t, be.queueBatchSettings.MergeCtx, "MergeCtx should be set when MetadataKeys is provided")
	})

	t.Run("without MetadataKeys - does not configure partitioner", func(t *testing.T) {
		tests := []struct {
			name         string
			metadataKeys []string
		}{
			{"empty slice", []string{}},
			{"nil", nil},
		}

		for _, tt := range tests {
			t.Run(tt.name, func(t *testing.T) {
				qCfg := NewDefaultQueueConfig()
				qCfg.Batch.GetOrInsertDefault().Partition.MetadataKeys = tt.metadataKeys

				be, err := NewBaseExporter(
					exportertest.NewNopSettings(exportertest.NopType),
					pipeline.SignalMetrics,
					noopExport,
					WithQueueBatchSettings(newFakeQueueBatch()),
					WithQueue(configoptional.Some(qCfg)),
				)
				require.NoError(t, err)
				assert.NotNil(t, be)

				// Verify partitioner and merge function are NOT configured
				assert.Nil(t, be.queueBatchSettings.Partitioner, "Partitioner should not be set when MetadataKeys is %s", tt.name)
				assert.Nil(t, be.queueBatchSettings.MergeCtx, "MergeCtx should not be set when MetadataKeys is %s", tt.name)
			})
		}
	})

	t.Run("error when custom partitioner already set and metadata_keys used", func(t *testing.T) {
		qCfg := NewDefaultQueueConfig()
		qCfg.Batch.GetOrInsertDefault().Partition.MetadataKeys = []string{"key1", "key2"}

		// Set up queue batch settings with a custom partitioner already configured
		customSettings := newFakeQueueBatch()
		customPartitioner := queuebatch.NewPartitioner(
			func(context.Context, request.Request) string {
				return "custom"
			},
		)
		customSettings.Partitioner = customPartitioner

		_, err := NewBaseExporter(
			exportertest.NewNopSettings(exportertest.NopType),
			pipeline.SignalMetrics,
			noopExport,
			WithQueueBatchSettings(customSettings),
			WithQueue(configoptional.Some(qCfg)),
		)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "cannot use metadata_keys when a custom partitioner is already configured")
	})

	t.Run("error when custom merge function already set and metadata_keys used", func(t *testing.T) {
		qCfg := NewDefaultQueueConfig()
		qCfg.Batch.GetOrInsertDefault().Partition.MetadataKeys = []string{"key1", "key2"}

		// Set up queue batch settings with a custom merge function already configured
		customSettings := newFakeQueueBatch()
		customSettings.MergeCtx = func(context.Context, context.Context) context.Context {
			return context.Background()
		}

		_, err := NewBaseExporter(
			exportertest.NewNopSettings(exportertest.NopType),
			pipeline.SignalMetrics,
			noopExport,
			WithQueueBatchSettings(customSettings),
			WithQueue(configoptional.Some(qCfg)),
		)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "cannot use metadata_keys when a custom merge function is already configured")
	})
}

func TestQueueRetryWithDisabledQueue(t *testing.T) {
	tests := []struct {
		name         string
		queueOptions []Option
	}{
		{
			name: "WithQueue",
			queueOptions: []Option{
				WithQueueBatchSettings(newFakeQueueBatch()),
				func() Option {
					return WithQueue(configoptional.None[queuebatch.Config]())
				}(),
			},
		},
		{
			name: "WithRequestQueue",
			queueOptions: []Option{
				func() Option {
					return WithQueueBatch(configoptional.None[queuebatch.Config](), newFakeQueueBatch())
				}(),
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			set := exportertest.NewNopSettings(exportertest.NopType)
			logger, observed := observer.New(zap.ErrorLevel)
			set.Logger = zap.New(logger)
			be, err := NewBaseExporter(set, pipeline.SignalLogs, errExport, tt.queueOptions...)
			require.NoError(t, err)
			require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
			mockR := &requesttest.FakeRequest{Items: 2}
			require.Error(t, be.Send(context.Background(), mockR))
			assert.Len(t, observed.All(), 1)
			assert.Equal(t, "Exporting failed. Rejecting data. Try enabling sending_queue to survive temporary failures.", observed.All()[0].Message)
			require.NoError(t, be.Shutdown(context.Background()))
		})
	}
}

func errExport(context.Context, request.Request) error {
	return errors.New("my error")
}

func noopExport(context.Context, request.Request) error {
	return nil
}

func newFakeQueueBatch() queuebatch.Settings[request.Request] {
	return queuebatch.Settings[request.Request]{
		Encoding: fakeEncoding{},
	}
}

type fakeEncoding struct{}

func (f fakeEncoding) Marshal(context.Context, request.Request) ([]byte, error) {
	return []byte("mockRequest"), nil
}

func (f fakeEncoding) Unmarshal([]byte) (context.Context, request.Request, error) {
	return context.Background(), &requesttest.FakeRequest{}, nil
}