opentelemetry-go/sdk/metric/batcher/defaultkeys/defaultkeys_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 defaultkeys_test

import (
	"context"
	"testing"

	"github.com/stretchr/testify/require"

	"go.opentelemetry.io/otel/api/core"
	export "go.opentelemetry.io/otel/sdk/export/metric"
	"go.opentelemetry.io/otel/sdk/metric/batcher/defaultkeys"
	"go.opentelemetry.io/otel/sdk/metric/batcher/test"
)

func TestGroupingStateless(t *testing.T) {
	ctx := context.Background()
	b := defaultkeys.New(test.NewAggregationSelector(), test.GroupEncoder, false)

	_ = b.Process(ctx, test.NewLastValueRecord(&test.LastValueADesc, test.Labels1, 10))
	_ = b.Process(ctx, test.NewLastValueRecord(&test.LastValueADesc, test.Labels2, 20))
	_ = b.Process(ctx, test.NewLastValueRecord(&test.LastValueADesc, test.Labels3, 30))

	_ = b.Process(ctx, test.NewLastValueRecord(&test.LastValueBDesc, test.Labels1, 10))
	_ = b.Process(ctx, test.NewLastValueRecord(&test.LastValueBDesc, test.Labels2, 20))
	_ = b.Process(ctx, test.NewLastValueRecord(&test.LastValueBDesc, test.Labels3, 30))

	_ = b.Process(ctx, test.NewCounterRecord(&test.CounterADesc, test.Labels1, 10))
	_ = b.Process(ctx, test.NewCounterRecord(&test.CounterADesc, test.Labels2, 20))
	_ = b.Process(ctx, test.NewCounterRecord(&test.CounterADesc, test.Labels3, 40))

	_ = b.Process(ctx, test.NewCounterRecord(&test.CounterBDesc, test.Labels1, 10))
	_ = b.Process(ctx, test.NewCounterRecord(&test.CounterBDesc, test.Labels2, 20))
	_ = b.Process(ctx, test.NewCounterRecord(&test.CounterBDesc, test.Labels3, 40))

	checkpointSet := b.CheckpointSet()
	b.FinishedCollection()

	records := test.Output{}
	err := checkpointSet.ForEach(records.AddTo)
	require.NoError(t, err)

	// Repeat for {counter,lastvalue}.{1,2}.
	// Output lastvalue should have only the "G=H" and "G=" keys.
	// Output counter should have only the "C=D" and "C=" keys.
	require.EqualValues(t, map[string]float64{
		"sum.a/C=D":       30, // labels1 + labels2
		"sum.a/C=":        40, // labels3
		"sum.b/C=D":       30, // labels1 + labels2
		"sum.b/C=":        40, // labels3
		"lastvalue.a/G=H": 10, // labels1
		"lastvalue.a/G=":  30, // labels3 = last value
		"lastvalue.b/G=H": 10, // labels1
		"lastvalue.b/G=":  30, // labels3 = last value
	}, records)

	// Verify that state is reset by FinishedCollection()
	checkpointSet = b.CheckpointSet()
	b.FinishedCollection()
	_ = checkpointSet.ForEach(func(rec export.Record) error {
		t.Fatal("Unexpected call")
		return nil
	})
}

func TestGroupingStateful(t *testing.T) {
	ctx := context.Background()
	b := defaultkeys.New(test.NewAggregationSelector(), test.GroupEncoder, true)

	counterA := test.NewCounterRecord(&test.CounterADesc, test.Labels1, 10)
	caggA := counterA.Aggregator()
	_ = b.Process(ctx, counterA)

	counterB := test.NewCounterRecord(&test.CounterBDesc, test.Labels1, 10)
	caggB := counterB.Aggregator()
	_ = b.Process(ctx, counterB)

	checkpointSet := b.CheckpointSet()
	b.FinishedCollection()

	records1 := test.Output{}
	err := checkpointSet.ForEach(records1.AddTo)
	require.NoError(t, err)

	require.EqualValues(t, map[string]float64{
		"sum.a/C=D": 10, // labels1
		"sum.b/C=D": 10, // labels1
	}, records1)

	// Test that state was NOT reset
	checkpointSet = b.CheckpointSet()
	b.FinishedCollection()

	records2 := test.Output{}
	err = checkpointSet.ForEach(records2.AddTo)
	require.NoError(t, err)

	require.EqualValues(t, records1, records2)

	// Update and re-checkpoint the original record.
	_ = caggA.Update(ctx, core.NewInt64Number(20), &test.CounterADesc)
	_ = caggB.Update(ctx, core.NewInt64Number(20), &test.CounterBDesc)
	caggA.Checkpoint(ctx, &test.CounterADesc)
	caggB.Checkpoint(ctx, &test.CounterBDesc)

	// As yet cagg has not been passed to Batcher.Process.  Should
	// not see an update.
	checkpointSet = b.CheckpointSet()
	b.FinishedCollection()

	records3 := test.Output{}
	err = checkpointSet.ForEach(records3.AddTo)
	require.NoError(t, err)

	require.EqualValues(t, records1, records3)

	// Now process the second update
	_ = b.Process(ctx, export.NewRecord(&test.CounterADesc, test.Labels1, caggA))
	_ = b.Process(ctx, export.NewRecord(&test.CounterBDesc, test.Labels1, caggB))

	checkpointSet = b.CheckpointSet()
	b.FinishedCollection()

	records4 := test.Output{}
	err = checkpointSet.ForEach(records4.AddTo)
	require.NoError(t, err)

	require.EqualValues(t, map[string]float64{
		"sum.a/C=D": 30,
		"sum.b/C=D": 30,
	}, records4)
}