[receiver/statsd] Add self-telemetry

receiver/statsdreceiver/internal/transport/mock_reporter.go

@@ -25,6 +25,10 @@ func NewMockReporter(expectedOnMetricsProcessedCalls int) *MockReporter {
 func (m *MockReporter) OnDebugf(_ string, _ ...any) {
 }
 
+func (m *MockReporter) RecordReceivedMetric(_ error) {}
+
+func (m *MockReporter) RecordFlushedMetrics(_ int64, _ error) {}
+
 // WaitAllOnMetricsProcessedCalls blocks until the number of expected calls
 // specified at creation of the reporter is completed.
 func (m *MockReporter) WaitAllOnMetricsProcessedCalls() {

receiver/statsdreceiver/internal/transport/server.go

@@ -41,4 +41,7 @@ type Reporter interface {
 	OnDebugf(
 		template string,
 		args ...any)
+
+	RecordReceivedMetric(err error)
+	RecordFlushedMetrics(count int64, err error)
 }

receiver/statsdreceiver/receiver.go

@@ -111,12 +111,16 @@ func (r *statsdReceiver) Start(ctx context.Context, _ component.Host) error {
 				for _, batch := range batchMetrics {
 					batchCtx := client.NewContext(ctx, batch.Info)
 
-					if err := r.Flush(batchCtx, batch.Metrics, r.nextConsumer); err != nil {
+					err := r.Flush(batchCtx, batch.Metrics, r.nextConsumer)
+					r.reporter.RecordFlushedMetrics(int64(batch.Metrics.DataPointCount()), err)
+					if err != nil {
 						r.reporter.OnDebugf("Error flushing metrics", zap.Error(err))
 					}
 				}
 			case metric := <-transferChan:
-				if err := r.parser.Aggregate(metric.Raw, metric.Addr); err != nil {
+				err := r.parser.Aggregate(metric.Raw, metric.Addr)
+				r.reporter.RecordReceivedMetric(err)
+				if err != nil {
 					r.reporter.OnDebugf("Error aggregating metric", zap.Error(err))
 				}
 			case <-ctx.Done():

receiver/statsdreceiver/reporter.go

@@ -4,19 +4,29 @@
 package statsdreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver"
 
 import (
+	"context"
+
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver/internal/metadata"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver/internal/transport"
 )
 
 // reporter struct implements the transport.Reporter interface to give consistent
 // observability per Collector metric observability package.
 type reporter struct {
-	logger        *zap.Logger
-	sugaredLogger *zap.SugaredLogger // Used for generic debug logging
-	obsrecv       *receiverhelper.ObsReport
+	logger               *zap.Logger
+	sugaredLogger        *zap.SugaredLogger // Used for generic debug logging
+	obsrecv              *receiverhelper.ObsReport
+	staticAttrs          []attribute.KeyValue
+	acceptedMetricPoints metric.Int64Counter
+	refusedMetricPoints  metric.Int64Counter
+	flushedMetricPoints  metric.Int64Counter
+	flushCount           metric.Int64Counter
 }
 
 var _ transport.Reporter = (*reporter)(nil)
@@ -30,15 +40,89 @@ func newReporter(set receiver.CreateSettings) (transport.Reporter, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &reporter{
+
+	r := &reporter{
 		logger:        set.Logger,
 		sugaredLogger: set.Logger.Sugar(),
 		obsrecv:       obsrecv,
-	}, nil
+		staticAttrs: []attribute.KeyValue{
+			attribute.String("receiver", set.ID.String()),
+		},
+	}
+
+	// See https://github.com/open-telemetry/opentelemetry-collector/blob/241334609fc47927b4a8533dfca28e0f65dad9fe/receiver/receiverhelper/obsreport.go#L104
+	// for the metric naming conventions
+
+	r.acceptedMetricPoints, err = metadata.Meter(set.TelemetrySettings).Int64Counter(
+		"receiver/accepted_metric_points",
+		metric.WithDescription("Number of metric data points accepted"),
+		metric.WithUnit("1"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	r.refusedMetricPoints, err = metadata.Meter(set.TelemetrySettings).Int64Counter(
+		"receiver/refused_metric_points",
+		metric.WithDescription("Number of metric data points refused"),
+		metric.WithUnit("1"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	r.flushedMetricPoints, err = metadata.Meter(set.TelemetrySettings).Int64Counter(
+		"receiver/statsd.flushed_metric_points",
+		metric.WithDescription("Number of metric data points flushed"),
+		metric.WithUnit("1"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	r.flushCount, err = metadata.Meter(set.TelemetrySettings).Int64Counter(
+		"receiver/statsd.flushes",
+		metric.WithDescription("Number of flush operations"),
+		metric.WithUnit("1"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return r, nil
 }
 
 func (r *reporter) OnDebugf(template string, args ...any) {
 	if r.logger.Check(zap.DebugLevel, "debug") != nil {
 		r.sugaredLogger.Debugf(template, args...)
 	}
 }
+
+func (r *reporter) RecordReceivedMetric(err error) {
+	if err == nil {
+		r.acceptedMetricPoints.Add(context.Background(), 1, metric.WithAttributes(r.staticAttrs...))
+	} else {
+		r.refusedMetricPoints.Add(context.Background(), 1, metric.WithAttributes(r.staticAttrs...))
+	}
+}
+
+func (r *reporter) RecordFlushedMetrics(count int64, err error) {
+	var status string
+	if err == nil {
+		status = "success"
+	} else {
+		status = "failure"
+	}
+	r.flushedMetricPoints.Add(
+		context.Background(),
+		count,
+		metric.WithAttributes(r.staticAttrs...),
+		metric.WithAttributes(attribute.String("status", status)),
+	)
+	r.flushCount.Add(
+		context.Background(),
+		1,
+		metric.WithAttributes(r.staticAttrs...),
+		metric.WithAttributes(attribute.String("status", status)),
+	)
+}