Add Persistent Queue metrics with number of dispatched batches

Author: pmm-sumo

exporter/exporterhelper/factory.go

@@ -20,6 +20,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenterror"
 	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 )
 
 // FactoryOption apply changes to ExporterOptions.
@@ -78,6 +79,7 @@ func NewFactory(
 	for _, opt := range options {
 		opt(f)
 	}
+	internal.RegisterMetrics()
 	return f
 }
 

exporter/exporterhelper/internal/observability_experimental.go

@@ -0,0 +1,89 @@
+// 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.
+
+//go:build enable_unstable
+// +build enable_unstable
+
+package internal
+
+import (
+	"context"
+	"sync"
+
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+)
+
+var (
+	currentlyDispatchedBatches = stats.Int64(
+		"/currently_dispatched_batches",
+		"Number of batches that are currently being sent",
+		stats.UnitDimensionless)
+
+	totalDispatchedBatches = stats.Int64(
+		"/total_dispatched_batches",
+		"Total number of batches which were processed",
+		stats.UnitDimensionless)
+
+	queueNameKey = tag.MustNewKey("queue_name")
+)
+
+func recordCurrentlyDispatchedBatches(ctx context.Context, dispatchedBatches int, queueName string) {
+	ctx, err := tag.New(ctx, tag.Insert(queueNameKey, queueName))
+	if err != nil {
+		return
+	}
+
+	stats.Record(ctx, currentlyDispatchedBatches.M(int64(dispatchedBatches)))
+}
+
+func recordBatchDispatched(ctx context.Context, queueName string) {
+	ctx, err := tag.New(ctx, tag.Insert(queueNameKey, queueName))
+	if err != nil {
+		return
+	}
+
+	stats.Record(ctx, totalDispatchedBatches.M(int64(1)))
+}
+
+// ExporterHelperInternalViews return the metrics views according to given telemetry level.
+func ExporterHelperInternalViews() []*view.View {
+
+	return []*view.View{
+		{
+			Name:        currentlyDispatchedBatches.Name(),
+			Description: currentlyDispatchedBatches.Description(),
+			Measure:     currentlyDispatchedBatches,
+			Aggregation: view.Count(),
+			TagKeys:     []tag.Key{queueNameKey},
+		},
+		{
+			Name:        totalDispatchedBatches.Name(),
+			Description: totalDispatchedBatches.Description(),
+			Measure:     totalDispatchedBatches,
+			Aggregation: view.Sum(),
+			TagKeys:     []tag.Key{queueNameKey},
+		},
+	}
+}
+
+var onceMetrics sync.Once
+
+// RegisterMetrics registers a set of metric views used by the internal package
+func RegisterMetrics() {
+	onceMetrics.Do(func() {
+		_ = view.Register(ExporterHelperInternalViews()...)
+	})
+}

exporter/exporterhelper/internal/observability_stable.go

@@ -0,0 +1,20 @@
+// 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.
+
+//go:build !enable_unstable
+// +build !enable_unstable
+
+package internal
+
+func RegisterMetrics() {}

exporter/exporterhelper/internal/persistent_storage.go

@@ -360,6 +360,8 @@ func (pcs *persistentContiguousStorage) itemDispatchingStart(ctx context.Context
 		pcs.logger.Debug("Failed updating currently dispatched items",
 			zap.String(zapQueueNameKey, pcs.queueName), zap.Error(err))
 	}
+
+	recordCurrentlyDispatchedBatches(ctx, len(pcs.currentlyDispatchedItems), pcs.queueName)
 }
 
 // itemDispatchingFinish removes the item from the list of currently dispatched items and deletes it from the persistent queue
@@ -380,6 +382,8 @@ func (pcs *persistentContiguousStorage) itemDispatchingFinish(ctx context.Contex
 		pcs.logger.Debug("Failed updating currently dispatched items",
 			zap.String(zapQueueNameKey, pcs.queueName), zap.Error(err))
 	}
+
+	recordBatchDispatched(ctx, pcs.queueName)
 }
 
 func (pcs *persistentContiguousStorage) updateReadIndex(ctx context.Context) {

exporter/exporterhelper/internal/persistent_storage_test.go

@@ -29,6 +29,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
@@ -169,6 +171,37 @@ func TestPersistentStorage_CurrentlyProcessedItems(t *testing.T) {
 	}
 }
 
+func TestPersistentStorage_MetricsReported(t *testing.T) {
+	path := createTemporaryDirectory()
+	defer os.RemoveAll(path)
+
+	traces := newTraces(5, 10)
+	req := newFakeTracesRequest(traces)
+
+	ext := createStorageExtension(path)
+	client := createTestClient(ext)
+	ps := createTestPersistentStorage(client)
+
+	RegisterMetrics()
+
+	for i := 0; i < 5; i++ {
+		err := ps.put(req)
+		require.NoError(t, err)
+	}
+
+	_ = getItemFromChannel(t, ps)
+	requireCurrentlyDispatchedItemsEqual(t, ps, []itemIndex{0, 1})
+
+	dd, err := view.RetrieveData("/currently_dispatched_batches")
+	require.NoError(t, err)
+	require.Equal(t, 1, len(dd))
+	require.Equal(t, 1, len(dd[0].Tags))
+	require.Equal(t, tag.Tag{Key: queueNameKey, Value: "foo"}, dd[0].Tags[0])
+	require.Equal(t, int64(2), dd[0].Data.(*view.CountData).Value)
+
+	ps.stop()
+}
+
 func TestPersistentStorage_RepeatPutCloseReadClose(t *testing.T) {
 	path := createTemporaryDirectory()
 	defer os.RemoveAll(path)