[receiver/googlecloudpubsub] Turn noisy warn in reset metric (open-telemetry#37571) (open-telemetry#37583)

#### Description
The receiver uses the Google Cloud Pub/Sub StreamingPull API and keeps a
open connection. The Pub/Sub servers
recurrently close the connection after a time period to avoid a
long-running sticky connection. Before the
receiver logged `warn` log lines everytime this happened. These log
lines are moved to debug so that fleets with
lots of collectors with the receiver don't span logs at warn level. To
keep track of the resets, whenever a
connection reset happens a
`otelcol_receiver_googlecloudpubsub_stream_restarts` metric is increased
by one.

#### Link to tracking issue
Fixes 37571

#### Documentation
Documentation is auto generated for the metric

Author: alexvanboxel

.chloggen/pubsubreceiver-turn-warn-into-reset-metric.yaml

@@ -0,0 +1,17 @@
+change_type: enhancement
+
+component: googlecloudpubsubreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Turn noisy `warn` log about Pub/Sub servers into `debug`, and turn the reset count into a metric
+
+issues: [37571]
+
+subtext: |
+  The receiver uses the Google Cloud Pub/Sub StreamingPull API and keeps a open connection. The Pub/Sub servers
+  recurrently close the connection after a time period to avoid a long-running sticky connection. Before the
+  receiver logged `warn` log lines everytime this happened. These log lines are moved to debug so that fleets with
+  lots of collectors with the receiver don't span logs at warn level. To keep track of the resets, whenever a
+  connection reset happens a `otelcol_receiver_googlecloudpubsub_stream_restarts` metric is increased by one.
+
+change_logs: [user]

receiver/googlecloudpubsubreceiver/documentation.md

@@ -0,0 +1,20 @@
+[comment]: <> (Code generated by mdatagen. DO NOT EDIT.)
+
+# googlecloudpubsub
+
+## Internal Telemetry
+
+The following telemetry is emitted by this component.
+
+### otelcol_receiver.googlecloudpubsub.stream_restarts
+
+Number of times the stream (re)starts due to a Pub/Sub servers connection close
+
+The receiver uses the Google Cloud Pub/Sub StreamingPull API and keeps a open connection. The Pub/Sub servers
+recurrently close the connection after a time period to avoid a long-running sticky connection. This metric
+counts the number of the resets that occurred during the lifetime of the container.
+
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| 1 | Sum | Int | true |

receiver/googlecloudpubsubreceiver/factory.go

@@ -41,24 +41,24 @@ func (factory *pubsubReceiverFactory) CreateDefaultConfig() component.Config {
 	return &Config{}
 }
 
-func (factory *pubsubReceiverFactory) ensureReceiver(params receiver.Settings, config component.Config) (*pubsubReceiver, error) {
+func (factory *pubsubReceiverFactory) ensureReceiver(settings receiver.Settings, config component.Config) (*pubsubReceiver, error) {
 	receiver := factory.receivers[config.(*Config)]
 	if receiver != nil {
 		return receiver, nil
 	}
 	rconfig := config.(*Config)
 	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
-		ReceiverID:             params.ID,
+		ReceiverID:             settings.ID,
 		Transport:              reportTransport,
-		ReceiverCreateSettings: params,
+		ReceiverCreateSettings: settings,
 	})
 	if err != nil {
 		return nil, err
 	}
 	receiver = &pubsubReceiver{
-		logger:    params.Logger,
+		settings:  settings,
 		obsrecv:   obsrecv,
-		userAgent: strings.ReplaceAll(rconfig.UserAgent, "{{version}}", params.BuildInfo.Version),
+		userAgent: strings.ReplaceAll(rconfig.UserAgent, "{{version}}", settings.BuildInfo.Version),
 		config:    rconfig,
 	}
 	factory.receivers[config.(*Config)] = receiver

receiver/googlecloudpubsubreceiver/go.mod

@@ -20,6 +20,10 @@ require (
 	go.opentelemetry.io/collector/pdata v1.25.0
 	go.opentelemetry.io/collector/receiver v0.119.0
 	go.opentelemetry.io/collector/receiver/receivertest v0.119.0
+	go.opentelemetry.io/otel v1.34.0
+	go.opentelemetry.io/otel/metric v1.34.0
+	go.opentelemetry.io/otel/sdk/metric v1.34.0
+	go.opentelemetry.io/otel/trace v1.34.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
@@ -72,11 +76,7 @@ require (
 	go.opentelemetry.io/collector/receiver/xreceiver v0.119.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.54.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.54.0 // indirect
-	go.opentelemetry.io/otel v1.34.0 // indirect
-	go.opentelemetry.io/otel/metric v1.34.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.34.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.34.0 // indirect
-	go.opentelemetry.io/otel/trace v1.34.0 // indirect
 	golang.org/x/crypto v0.32.0 // indirect
 	golang.org/x/net v0.34.0 // indirect
 	golang.org/x/oauth2 v0.25.0 // indirect

receiver/googlecloudpubsubreceiver/internal/handler.go

@@ -13,9 +13,14 @@ import (
 	"time"
 
 	"cloud.google.com/go/pubsub/apiv1/pubsubpb"
+	"go.opentelemetry.io/collector/receiver"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudpubsubreceiver/internal/metadata"
 )
 
 // Time to wait before restarting, when the stream stopped
@@ -36,7 +41,8 @@ type StreamHandler struct {
 	streamWaitGroup sync.WaitGroup
 	// wait group for the handler
 	handlerWaitGroup sync.WaitGroup
-	logger           *zap.Logger
+	settings         receiver.Settings
+	telemetryBuilder *metadata.TelemetryBuilder
 	// time that acknowledge loop waits before acknowledging messages
 	ackBatchWait time.Duration
 
@@ -51,19 +57,21 @@ func (handler *StreamHandler) ack(ackID string) {
 
 func NewHandler(
 	ctx context.Context,
-	logger *zap.Logger,
+	settings receiver.Settings,
+	telemetryBuilder *metadata.TelemetryBuilder,
 	client SubscriberClient,
 	clientID string,
 	subscription string,
 	callback func(ctx context.Context, message *pubsubpb.ReceivedMessage) error,
 ) (*StreamHandler, error) {
 	handler := StreamHandler{
-		logger:       logger,
-		client:       client,
-		clientID:     clientID,
-		subscription: subscription,
-		pushMessage:  callback,
-		ackBatchWait: 10 * time.Second,
+		settings:         settings,
+		telemetryBuilder: telemetryBuilder,
+		client:           client,
+		clientID:         clientID,
+		subscription:     subscription,
+		pushMessage:      callback,
+		ackBatchWait:     10 * time.Second,
 	}
 	return &handler, handler.initStream(ctx)
 }
@@ -85,6 +93,11 @@ func (handler *StreamHandler) initStream(ctx context.Context) error {
 		_ = handler.stream.CloseSend()
 		return err
 	}
+	handler.telemetryBuilder.ReceiverGooglecloudpubsubStreamRestarts.Add(ctx, 1,
+		metric.WithAttributes(
+			attribute.String("otelcol.component.kind", "receiver"),
+			attribute.String("otelcol.component.id", handler.settings.ID.String()),
+		))
 	return nil
 }
 
@@ -102,7 +115,7 @@ func (handler *StreamHandler) recoverableStream(ctx context.Context) {
 		var loopCtx context.Context
 		loopCtx, cancel := context.WithCancel(ctx)
 
-		handler.logger.Info("Starting Streaming Pull")
+		handler.settings.Logger.Debug("Starting Streaming Pull")
 		handler.streamWaitGroup.Add(2)
 		go handler.requestStream(loopCtx, cancel)
 		go handler.responseStream(loopCtx, cancel)
@@ -117,13 +130,13 @@ func (handler *StreamHandler) recoverableStream(ctx context.Context) {
 		if handler.isRunning.Load() {
 			err := handler.initStream(ctx)
 			if err != nil {
-				handler.logger.Error("Failed to recovery stream.")
+				handler.settings.Logger.Error("Failed to recovery stream.")
 			}
 		}
-		handler.logger.Warn("End of recovery loop, restarting.")
+		handler.settings.Logger.Debug("End of recovery loop, restarting.")
 		time.Sleep(streamRecoveryBackoffPeriod)
 	}
-	handler.logger.Warn("Shutting down recovery loop.")
+	handler.settings.Logger.Warn("Shutting down recovery loop.")
 	handler.handlerWaitGroup.Done()
 }
 
@@ -157,15 +170,15 @@ func (handler *StreamHandler) requestStream(ctx context.Context, cancel context.
 	for {
 		if err := handler.acknowledgeMessages(); err != nil {
 			if errors.Is(err, io.EOF) {
-				handler.logger.Warn("EOF reached")
+				handler.settings.Logger.Warn("EOF reached")
 				break
 			}
-			handler.logger.Error(fmt.Sprintf("Failed in acknowledge messages with error %v", err))
+			handler.settings.Logger.Error(fmt.Sprintf("Failed in acknowledge messages with error %v", err))
 			break
 		}
 		select {
 		case <-ctx.Done():
-			handler.logger.Warn("requestStream <-ctx.Done()")
+			handler.settings.Logger.Debug("requestStream <-ctx.Done()")
 		case <-timer.C:
 			timer.Reset(handler.ackBatchWait)
 		}
@@ -176,7 +189,7 @@ func (handler *StreamHandler) requestStream(ctx context.Context, cancel context.
 		}
 	}
 	cancel()
-	handler.logger.Warn("Request Stream loop ended.")
+	handler.settings.Logger.Debug("Request Stream loop ended.")
 	_ = handler.stream.CloseSend()
 	handler.streamWaitGroup.Done()
 }
@@ -202,30 +215,30 @@ func (handler *StreamHandler) responseStream(ctx context.Context, cancel context
 			case errors.Is(err, io.EOF):
 				activeStreaming = false
 			case !grpcStatus:
-				handler.logger.Warn("response stream breaking on error",
+				handler.settings.Logger.Warn("response stream breaking on error",
 					zap.Error(err))
 				activeStreaming = false
 			case s.Code() == codes.Unavailable:
-				handler.logger.Info("response stream breaking on gRPC s 'Unavailable'")
+				handler.settings.Logger.Debug("response stream breaking on gRPC s 'Unavailable'")
 				activeStreaming = false
 			case s.Code() == codes.NotFound:
-				handler.logger.Error("resource doesn't exist, wait 60 seconds, and restarting stream")
+				handler.settings.Logger.Error("resource doesn't exist, wait 60 seconds, and restarting stream")
 				time.Sleep(time.Second * 60)
 				activeStreaming = false
 			default:
-				handler.logger.Warn("response stream breaking on gRPC s "+s.Message(),
+				handler.settings.Logger.Warn("response stream breaking on gRPC s "+s.Message(),
 					zap.String("s", s.Message()),
 					zap.Error(err))
 				activeStreaming = false
 			}
 		}
 		if errors.Is(ctx.Err(), context.Canceled) {
 			// Canceling the loop, collector is probably stopping
-			handler.logger.Warn("response stream ctx.Err() == context.Canceled")
+			handler.settings.Logger.Warn("response stream ctx.Err() == context.Canceled")
 			break
 		}
 	}
 	cancel()
-	handler.logger.Warn("Response Stream loop ended.")
+	handler.settings.Logger.Debug("Response Stream loop ended.")
 	handler.streamWaitGroup.Done()
 }

receiver/googlecloudpubsubreceiver/internal/handler_test.go

@@ -12,10 +12,12 @@ import (
 	"cloud.google.com/go/pubsub/apiv1/pubsubpb"
 	"cloud.google.com/go/pubsub/pstest"
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/zap/zaptest"
+	"go.opentelemetry.io/collector/receiver/receivertest"
 	"google.golang.org/api/option"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudpubsubreceiver/internal/metadata"
 )
 
 func TestCancelStream(t *testing.T) {
@@ -41,10 +43,13 @@ func TestCancelStream(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
+	settings := receivertest.NewNopSettings()
+	telemetryBuilder, _ := metadata.NewTelemetryBuilder(settings.TelemetrySettings)
+
 	client, err := pubsub.NewSubscriberClient(ctx, copts...)
 	assert.NoError(t, err)
 
-	handler, err := NewHandler(ctx, zaptest.NewLogger(t), client, "client-id", "projects/my-project/subscriptions/otlp",
+	handler, err := NewHandler(ctx, settings, telemetryBuilder, client, "client-id", "projects/my-project/subscriptions/otlp",
 		func(context.Context, *pubsubpb.ReceivedMessage) error {
 			return nil
 		})