[receiver/awss3receiver] Add obsreport usage for receiver metrics (#3…

…3110)

**Description:** Add metrics using obsreport to the S3 receiver

**Link to tracking Issue:** #30750 

**Testing:** Full stack test, confirmed
`otelcol_receiver_accepted_spans` and `otelcol_receiver_refused_spans`
was being exported for the `awss3` receiver.

**Documentation:** None

.chloggen/awss3receiver_obsreport.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awss3receiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Uses obsreport to report metrics for the AWS S3 Receiver.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30750]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

receiver/awss3receiver/factory.go

@@ -22,5 +22,5 @@ func NewFactory() receiver.Factory {
 }
 
 func createTracesReceiver(ctx context.Context, settings receiver.CreateSettings, cc component.Config, consumer consumer.Traces) (receiver.Traces, error) {
-	return newAWSS3TraceReceiver(ctx, cc.(*Config), consumer, settings.Logger)
+	return newAWSS3TraceReceiver(ctx, cc.(*Config), consumer, settings)
 }

receiver/awss3receiver/go.mod

@@ -59,6 +59,7 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.53.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
+	go.opentelemetry.io/collector v0.101.0 // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.101.0 // indirect
 	go.opentelemetry.io/otel v1.26.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.48.0 // indirect

receiver/awss3receiver/receiver.go

@@ -13,6 +13,8 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/receiver"
+	"go.opentelemetry.io/collector/receiver/receiverhelper"
 	"go.uber.org/zap"
 )
 
@@ -21,18 +23,28 @@ type awss3TraceReceiver struct {
 	consumer consumer.Traces
 	logger   *zap.Logger
 	cancel   context.CancelFunc
+	obsrecv  *receiverhelper.ObsReport
 }
 
-func newAWSS3TraceReceiver(ctx context.Context, cfg *Config, traces consumer.Traces, logger *zap.Logger) (*awss3TraceReceiver, error) {
+func newAWSS3TraceReceiver(ctx context.Context, cfg *Config, traces consumer.Traces, settings receiver.CreateSettings) (*awss3TraceReceiver, error) {
 	reader, err := newS3Reader(ctx, cfg)
 	if err != nil {
 		return nil, err
 	}
+	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
+		ReceiverID:             settings.ID,
+		Transport:              "s3",
+		ReceiverCreateSettings: settings,
+	})
+	if err != nil {
+		return nil, err
+	}
 	return &awss3TraceReceiver{
 		s3Reader: reader,
 		consumer: traces,
-		logger:   logger,
+		logger:   settings.Logger,
 		cancel:   nil,
+		obsrecv:  obsrecv,
 	}, nil
 }
 
@@ -70,11 +82,14 @@ func (r *awss3TraceReceiver) receiveBytes(ctx context.Context, key string, data
 	}
 
 	var unmarshaler ptrace.Unmarshaler
+	var format string
 	if strings.HasSuffix(key, ".json") {
 		unmarshaler = &ptrace.JSONUnmarshaler{}
+		format = "otlp_json"
 	}
 	if strings.HasSuffix(key, ".binpb") {
 		unmarshaler = &ptrace.ProtoUnmarshaler{}
+		format = "otlp_proto"
 	}
 	if unmarshaler == nil {
 		r.logger.Warn("Unsupported file format", zap.String("key", key))
@@ -84,5 +99,8 @@ func (r *awss3TraceReceiver) receiveBytes(ctx context.Context, key string, data
 	if err != nil {
 		return err
 	}
-	return r.consumer.ConsumeTraces(ctx, traces)
+	obsCtx := r.obsrecv.StartTracesOp(ctx)
+	err = r.consumer.ConsumeTraces(ctx, traces)
+	r.obsrecv.EndTracesOp(obsCtx, format, traces.SpanCount(), err)
+	return err
 }

receiver/awss3receiver/receiver_test.go

@@ -12,6 +12,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/receiver/receiverhelper"
+	"go.opentelemetry.io/collector/receiver/receivertest"
 	conventions "go.opentelemetry.io/collector/semconv/v1.22.0"
 	"go.uber.org/zap"
 )
@@ -121,9 +123,12 @@ func Test_receiveBytes(t *testing.T) {
 				}
 				return nil
 			})
+			obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{ReceiverCreateSettings: receivertest.NewNopCreateSettings()})
+			require.NoError(t, err)
 			r := &awss3TraceReceiver{
 				consumer: tracesConsumer,
 				logger:   zap.NewNop(),
+				obsrecv:  obsrecv,
 			}
 			if err := r.receiveBytes(context.Background(), tt.args.key, tt.args.data); (err != nil) != tt.wantErr {
 				t.Errorf("receiveBytes() error = %v, wantErr %v", err, tt.wantErr)