exporter_test.go

// Copyright The OpenTelemetry Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package opencensus

import (
	"context"
	"errors"
	"fmt"
	"testing"
	"time"

	"go.opentelemetry.io/otel"

	"go.opencensus.io/metric/metricdata"
	ocresource "go.opencensus.io/resource"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/metric/metrictest"
	"go.opentelemetry.io/otel/metric/number"
	"go.opentelemetry.io/otel/metric/sdkapi"
	"go.opentelemetry.io/otel/metric/unit"
	export "go.opentelemetry.io/otel/sdk/export/metric"
	exportmetric "go.opentelemetry.io/otel/sdk/export/metric"
	"go.opentelemetry.io/otel/sdk/export/metric/aggregation"
	"go.opentelemetry.io/otel/sdk/instrumentation"
	"go.opentelemetry.io/otel/sdk/metric/controller/controllertest"
	"go.opentelemetry.io/otel/sdk/resource"
)

type fakeExporter struct {
	export.Exporter
	records  []export.Record
	resource *resource.Resource
	err      error
}

func (f *fakeExporter) Export(ctx context.Context, res *resource.Resource, ilr exportmetric.InstrumentationLibraryReader) error {
	return controllertest.ReadAll(ilr, export.StatelessExportKindSelector(),
		func(_ instrumentation.Library, record exportmetric.Record) error {
			f.resource = res
			f.records = append(f.records, record)
			return f.err
		})
}

type fakeErrorHandler struct {
	err error
}

func (f *fakeErrorHandler) Handle(err error) {
	f.err = err
}

func (f *fakeErrorHandler) matches(err error) error {
	// make sure err is cleared for the next test
	defer func() { f.err = nil }()
	if !errors.Is(f.err, err) {
		return fmt.Errorf("err(%v), want err(%v)", f.err, err)
	}
	return nil
}

func TestExportMetrics(t *testing.T) {
	now := time.Now()
	basicDesc := metrictest.NewDescriptor(
		"",
		sdkapi.GaugeObserverInstrumentKind,
		number.Int64Kind,
	)
	fakeErrorHandler := &fakeErrorHandler{}
	otel.SetErrorHandler(fakeErrorHandler)
	for _, tc := range []struct {
		desc                 string
		input                []*metricdata.Metric
		exportErr            error
		expected             []export.Record
		expectedResource     *resource.Resource
		expectedHandledError error
	}{
		{
			desc: "no metrics",
		},
		{
			desc: "metric without points is dropped",
			input: []*metricdata.Metric{
				{
					TimeSeries: []*metricdata.TimeSeries{
						{},
					},
				},
			},
		},
		{
			desc: "descriptor conversion error",
			input: []*metricdata.Metric{
				// TypeGaugeDistribution isn't supported
				{Descriptor: metricdata.Descriptor{Type: metricdata.TypeGaugeDistribution}},
			},
			expectedHandledError: errConversion,
		},
		{
			desc: "labels conversion error",
			input: []*metricdata.Metric{
				{
					// No descriptor with label keys.
					TimeSeries: []*metricdata.TimeSeries{
						// 1 label value, which doens't exist in keys.
						{
							LabelValues: []metricdata.LabelValue{{Value: "foo", Present: true}},
							Points: []metricdata.Point{
								{},
							},
						},
					},
				},
			},
			expectedHandledError: errConversion,
		},
		{
			desc: "unsupported summary point type",
			input: []*metricdata.Metric{
				{
					TimeSeries: []*metricdata.TimeSeries{
						{
							Points: []metricdata.Point{
								{Value: &metricdata.Summary{}},
							},
						},
					},
				},
			},
			expectedHandledError: errIncompatibleType,
		},
		{
			desc: "success",
			input: []*metricdata.Metric{
				{
					Resource: &ocresource.Resource{
						Labels: map[string]string{
							"R1": "V1",
							"R2": "V2",
						},
					},
					TimeSeries: []*metricdata.TimeSeries{
						{
							StartTime: now,
							Points: []metricdata.Point{
								{Value: int64(123), Time: now},
							},
						},
					},
				},
			},
			expectedResource: resource.NewSchemaless(
				attribute.String("R1", "V1"),
				attribute.String("R2", "V2"),
			),
			expected: []export.Record{
				export.NewRecord(
					&basicDesc,
					attribute.EmptySet(),
					&ocExactAggregator{
						points: []aggregation.Point{
							{
								Number: number.NewInt64Number(123),
								Time:   now,
							},
						},
					},
					now,
					now,
				),
			},
		},
		{
			desc: "export error after success",
			input: []*metricdata.Metric{
				{
					TimeSeries: []*metricdata.TimeSeries{
						{
							StartTime: now,
							Points: []metricdata.Point{
								{Value: int64(123), Time: now},
							},
						},
					},
				},
			},
			expected: []export.Record{
				export.NewRecord(
					&basicDesc,
					attribute.EmptySet(),
					&ocExactAggregator{
						points: []aggregation.Point{
							{
								Number: number.NewInt64Number(123),
								Time:   now,
							},
						},
					},
					now,
					now,
				),
			},
			exportErr: errors.New("failed to export"),
		},
		{
			desc: "partial success sends correct metrics and drops incorrect metrics with handled err",
			input: []*metricdata.Metric{
				{
					TimeSeries: []*metricdata.TimeSeries{
						{
							StartTime: now,
							Points: []metricdata.Point{
								{Value: int64(123), Time: now},
							},
						},
					},
				},
				// TypeGaugeDistribution isn't supported
				{Descriptor: metricdata.Descriptor{Type: metricdata.TypeGaugeDistribution}},
			},
			expected: []export.Record{
				export.NewRecord(
					&basicDesc,
					attribute.EmptySet(),
					&ocExactAggregator{
						points: []aggregation.Point{
							{
								Number: number.NewInt64Number(123),
								Time:   now,
							},
						},
					},
					now,
					now,
				),
			},
			expectedHandledError: errConversion,
		},
	} {
		t.Run(tc.desc, func(t *testing.T) {
			fakeExporter := &fakeExporter{err: tc.exportErr}
			err := NewMetricExporter(fakeExporter).ExportMetrics(context.Background(), tc.input)
			if !errors.Is(err, tc.exportErr) {
				t.Errorf("NewMetricExporter(%+v) = err(%v), want err(%v)", tc.input, err, tc.exportErr)
			}
			// Check the global error handler, since we don't return errors
			// which occur during conversion.
			err = fakeErrorHandler.matches(tc.expectedHandledError)
			if err != nil {
				t.Fatalf("ExportMetrics(%+v) = %v", tc.input, err)
			}
			output := fakeExporter.records
			if len(tc.expected) != len(output) {
				t.Fatalf("ExportMetrics(%+v) = %d records, want %d records", tc.input, len(output), len(tc.expected))
			}
			if fakeExporter.resource.String() != tc.expectedResource.String() {
				t.Errorf("ExportMetrics(%+v)[i].Resource() = %+v, want %+v", tc.input, fakeExporter.resource.String(), tc.expectedResource.String())
			}
			for i, expected := range tc.expected {
				if output[i].StartTime() != expected.StartTime() {
					t.Errorf("ExportMetrics(%+v)[i].StartTime() = %+v, want %+v", tc.input, output[i].StartTime(), expected.StartTime())
				}
				if output[i].EndTime() != expected.EndTime() {
					t.Errorf("ExportMetrics(%+v)[i].EndTime() = %+v, want %+v", tc.input, output[i].EndTime(), expected.EndTime())
				}
				if output[i].Descriptor().Name() != expected.Descriptor().Name() {
					t.Errorf("ExportMetrics(%+v)[i].Descriptor() = %+v, want %+v", tc.input, output[i].Descriptor().Name(), expected.Descriptor().Name())
				}
				// Don't bother with a complete check of the descriptor.
				// That is checked as part of descriptor conversion tests below.
				if !output[i].Labels().Equals(expected.Labels()) {
					t.Errorf("ExportMetrics(%+v)[i].Labels() = %+v, want %+v", tc.input, output[i].Labels(), expected.Labels())
				}
				if output[i].Aggregation().Kind() != expected.Aggregation().Kind() {
					t.Errorf("ExportMetrics(%+v)[i].Aggregation() = %+v, want %+v", tc.input, output[i].Aggregation().Kind(), expected.Aggregation().Kind())
				}
				// Don't bother checking the contents of the points aggregation.
				// Those tests are done with the aggregations themselves
			}
		})
	}
}

func TestConvertLabels(t *testing.T) {
	setWithMultipleKeys := attribute.NewSet(
		attribute.KeyValue{Key: attribute.Key("first"), Value: attribute.StringValue("1")},
		attribute.KeyValue{Key: attribute.Key("second"), Value: attribute.StringValue("2")},
	)
	for _, tc := range []struct {
		desc        string
		inputKeys   []metricdata.LabelKey
		inputValues []metricdata.LabelValue
		expected    *attribute.Set
		expectedErr error
	}{
		{
			desc:     "no labels",
			expected: attribute.EmptySet(),
		},
		{
			desc:        "different numbers of keys and values",
			inputKeys:   []metricdata.LabelKey{{Key: "foo"}},
			expected:    attribute.EmptySet(),
			expectedErr: errConversion,
		},
		{
			desc:      "multiple keys and values",
			inputKeys: []metricdata.LabelKey{{Key: "first"}, {Key: "second"}},
			inputValues: []metricdata.LabelValue{
				{Value: "1", Present: true},
				{Value: "2", Present: true},
			},
			expected: &setWithMultipleKeys,
		},
		{
			desc:      "multiple keys and values with some not present",
			inputKeys: []metricdata.LabelKey{{Key: "first"}, {Key: "second"}, {Key: "third"}},
			inputValues: []metricdata.LabelValue{
				{Value: "1", Present: true},
				{Value: "2", Present: true},
				{Present: false},
			},
			expected: &setWithMultipleKeys,
		},
	} {
		t.Run(tc.desc, func(t *testing.T) {
			output, err := convertLabels(tc.inputKeys, tc.inputValues)
			if !errors.Is(err, tc.expectedErr) {
				t.Errorf("convertLabels(keys: %v, values: %v) = err(%v), want err(%v)", tc.inputKeys, tc.inputValues, err, tc.expectedErr)
			}
			if !output.Equals(tc.expected) {
				t.Errorf("convertLabels(keys: %v, values: %v) = %+v, want %+v", tc.inputKeys, tc.inputValues, output.ToSlice(), tc.expected.ToSlice())
			}
		})
	}
}
func TestConvertResource(t *testing.T) {
	for _, tc := range []struct {
		desc     string
		input    *ocresource.Resource
		expected *resource.Resource
	}{
		{
			desc: "nil resource",
		},
		{
			desc: "empty resource",
			input: &ocresource.Resource{
				Labels: map[string]string{},
			},
			expected: resource.NewSchemaless(),
		},
		{
			desc: "resource with labels",
			input: &ocresource.Resource{
				Labels: map[string]string{
					"foo":  "bar",
					"tick": "tock",
				},
			},
			expected: resource.NewSchemaless(
				attribute.KeyValue{Key: attribute.Key("foo"), Value: attribute.StringValue("bar")},
				attribute.KeyValue{Key: attribute.Key("tick"), Value: attribute.StringValue("tock")},
			),
		},
	} {
		t.Run(tc.desc, func(t *testing.T) {
			output := convertResource(tc.input)
			if !output.Equal(tc.expected) {
				t.Errorf("convertResource(%v) = %+v, want %+v", tc.input, output, tc.expected)
			}
		})
	}
}
func TestConvertDescriptor(t *testing.T) {
	for _, tc := range []struct {
		desc        string
		input       metricdata.Descriptor
		expected    metric.Descriptor
		expectedErr error
	}{
		{
			desc: "empty descriptor",
			expected: metrictest.NewDescriptor(
				"",
				sdkapi.GaugeObserverInstrumentKind,
				number.Int64Kind,
			),
		},
		{
			desc: "gauge int64 bytes",
			input: metricdata.Descriptor{
				Name:        "foo",
				Description: "bar",
				Unit:        metricdata.UnitBytes,
				Type:        metricdata.TypeGaugeInt64,
			},
			expected: metrictest.NewDescriptor(
				"foo",
				sdkapi.GaugeObserverInstrumentKind,
				number.Int64Kind,
				metric.WithDescription("bar"),
				metric.WithUnit(unit.Bytes),
			),
		},
		{
			desc: "gauge float64 ms",
			input: metricdata.Descriptor{
				Name:        "foo",
				Description: "bar",
				Unit:        metricdata.UnitMilliseconds,
				Type:        metricdata.TypeGaugeFloat64,
			},
			expected: metrictest.NewDescriptor(
				"foo",
				sdkapi.GaugeObserverInstrumentKind,
				number.Float64Kind,
				metric.WithDescription("bar"),
				metric.WithUnit(unit.Milliseconds),
			),
		},
		{
			desc: "cumulative int64 dimensionless",
			input: metricdata.Descriptor{
				Name:        "foo",
				Description: "bar",
				Unit:        metricdata.UnitDimensionless,
				Type:        metricdata.TypeCumulativeInt64,
			},
			expected: metrictest.NewDescriptor(
				"foo",
				sdkapi.CounterObserverInstrumentKind,
				number.Int64Kind,
				metric.WithDescription("bar"),
				metric.WithUnit(unit.Dimensionless),
			),
		},
		{
			desc: "cumulative float64 dimensionless",
			input: metricdata.Descriptor{
				Name:        "foo",
				Description: "bar",
				Unit:        metricdata.UnitDimensionless,
				Type:        metricdata.TypeCumulativeFloat64,
			},
			expected: metrictest.NewDescriptor(
				"foo",
				sdkapi.CounterObserverInstrumentKind,
				number.Float64Kind,
				metric.WithDescription("bar"),
				metric.WithUnit(unit.Dimensionless),
			),
		},
		{
			desc: "incompatible TypeCumulativeDistribution",
			input: metricdata.Descriptor{
				Name:        "foo",
				Description: "bar",
				Type:        metricdata.TypeCumulativeDistribution,
			},
			expectedErr: errConversion,
		},
	} {
		t.Run(tc.desc, func(t *testing.T) {
			output, err := convertDescriptor(tc.input)
			if !errors.Is(err, tc.expectedErr) {
				t.Errorf("convertDescriptor(%v) = err(%v), want err(%v)", tc.input, err, tc.expectedErr)
			}
			if output != tc.expected {
				t.Errorf("convertDescriptor(%v) = %+v, want %+v", tc.input, output, tc.expected)
			}
		})
	}
}