[stefreceiver] Correctly handle Shutdown request (open-telemetry#40082)

#### Description

Previously the receiver would get stuck in Shutdown if there were open
connections.

Now we terminate open connections if Shutdown is requested and make sure
Shutdown proceeds.

#### Testing

Added unit test and tested manually.

Author: tigrannajaryan

.chloggen/tigran_stefreceivershutdown.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: stefreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Correctly handle Shutdown request
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40082]
+
+# (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: The receiver now correctly Shutdown even if there are active connections
+
+# 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: [user]

receiver/stefreceiver/e2e_test.go

@@ -12,50 +12,87 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/receiver/receivertest"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/stefexporter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/stefreceiver/internal/metadata"
 )
 
-func TestRoundtrip(t *testing.T) {
+func genMetrics() pmetric.Metrics {
+	data := pmetric.NewMetrics()
+	metricPoint := data.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty()
+	metricPoint.SetName("foo")
+	gauge := metricPoint.SetEmptyGauge()
+	dp := gauge.DataPoints().AppendEmpty()
+	dp.SetIntValue(1)
+	dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Now()))
+	return data
+}
+
+func createReceiver(t *testing.T, endpoint string) (receiver.Metrics, *consumertest.MetricsSink) {
 	sink := &consumertest.MetricsSink{}
 	settings := receivertest.NewNopSettings(metadata.Type)
 	settings.Logger, _ = zap.NewDevelopment()
-	m, err := NewFactory().CreateMetrics(context.Background(), settings, createDefaultConfig(), sink)
-	t.Cleanup(func() {
-		err = m.Shutdown(context.Background())
-		require.NoError(t, err)
-	})
+	rcvCfg := createDefaultConfig()
+	rcvCfg.(*Config).NetAddr.Endpoint = endpoint
+	m, err := NewFactory().CreateMetrics(context.Background(), settings, rcvCfg, sink)
 	require.NoError(t, m.Start(context.Background(), componenttest.NewNopHost()))
 	require.NoError(t, err)
+	return m, sink
+}
+
+func createExporter(t *testing.T, endpoint string) exporter.Metrics {
 	cfg := stefexporter.NewFactory().CreateDefaultConfig().(*stefexporter.Config)
-	cfg.Endpoint = "localhost:4320"
+	cfg.Endpoint = endpoint
 	cfg.TLSSetting.Insecure = true
-	exporterSettings := exportertest.NewNopSettings(metadata.Type)
-	exporterSettings.Logger, _ = zap.NewDevelopment()
-	exporter, err := stefexporter.NewFactory().CreateMetrics(context.Background(), exporterSettings, cfg)
+	settings := exportertest.NewNopSettings(metadata.Type)
+	settings.Logger, _ = zap.NewDevelopment()
+	exp, err := stefexporter.NewFactory().CreateMetrics(context.Background(), settings, cfg)
 	require.NoError(t, err)
-	t.Cleanup(func() {
-		err = exporter.Shutdown(context.Background())
-		require.NoError(t, err)
-	})
+	require.NoError(t, exp.Start(context.Background(), componenttest.NewNopHost()))
+	return exp
+}
 
-	require.NoError(t, exporter.Start(context.Background(), componenttest.NewNopHost()))
-	data := pmetric.NewMetrics()
-	metricPoint := data.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty()
-	metricPoint.SetName("foo")
-	gauge := metricPoint.SetEmptyGauge()
-	dp := gauge.DataPoints().AppendEmpty()
-	dp.SetIntValue(1)
-	dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Now()))
-	err = exporter.ConsumeMetrics(context.Background(), data)
+func TestRoundtrip(t *testing.T) {
+	endpoint := testutil.GetAvailableLocalAddress(t)
+	m, sink := createReceiver(t, endpoint)
+	t.Cleanup(func() { require.NoError(t, m.Shutdown(context.Background())) })
+
+	exporter := createExporter(t, endpoint)
+	t.Cleanup(func() { require.NoError(t, exporter.Shutdown(context.Background())) })
+
+	err := exporter.ConsumeMetrics(context.Background(), genMetrics())
 	require.NoError(t, err)
 	assert.EventuallyWithT(t, func(tt *assert.CollectT) {
 		assert.Len(tt, sink.AllMetrics(), 1)
 	}, 1*time.Minute, 10*time.Millisecond)
 }
+
+func TestShutdownWhenConnected(t *testing.T) {
+	endpoint := testutil.GetAvailableLocalAddress(t)
+	receiver, sink := createReceiver(t, endpoint)
+	exporter := createExporter(t, endpoint)
+
+	require.NoError(t, exporter.ConsumeMetrics(context.Background(), genMetrics()))
+
+	assert.EventuallyWithT(
+		t, func(tt *assert.CollectT) {
+			assert.Len(tt, sink.AllMetrics(), 1)
+		}, 1*time.Minute, 10*time.Millisecond,
+	)
+
+	// Try shutdown receiver before shutting down exporter.
+	// This means there is an active connection at the receiver.
+	// Previously we had a bug causing the receiver Shutdown to hang forever
+	// in this situation.
+	require.NoError(t, receiver.Shutdown(context.Background()))
+
+	require.NoError(t, exporter.Shutdown(context.Background()))
+}

receiver/stefreceiver/go.mod

@@ -4,6 +4,7 @@ go 1.23.0
 
 require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/stefexporter v0.126.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.126.0
 	github.com/splunk/stef/go/grpc v0.0.6
 	github.com/splunk/stef/go/otel v0.0.6
 	github.com/splunk/stef/go/pdata v0.0.6
@@ -18,6 +19,7 @@ require (
 	go.opentelemetry.io/collector/consumer v1.32.0
 	go.opentelemetry.io/collector/consumer/consumererror v0.126.0
 	go.opentelemetry.io/collector/consumer/consumertest v0.126.0
+	go.opentelemetry.io/collector/exporter v0.126.0
 	go.opentelemetry.io/collector/exporter/exportertest v0.126.0
 	go.opentelemetry.io/collector/pdata v1.32.0
 	go.opentelemetry.io/collector/receiver v1.32.0
@@ -63,7 +65,6 @@ require (
 	go.opentelemetry.io/collector/config/configopaque v1.32.0 // indirect
 	go.opentelemetry.io/collector/config/configretry v1.32.0 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.126.0 // indirect
-	go.opentelemetry.io/collector/exporter v0.126.0 // indirect
 	go.opentelemetry.io/collector/exporter/xexporter v0.126.0 // indirect
 	go.opentelemetry.io/collector/extension v1.32.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionauth v1.32.0 // indirect

receiver/stefreceiver/stef.go

@@ -6,6 +6,8 @@ import (
 	"context"
 	"errors"
 	"net"
+	"sync/atomic"
+	"time"
 
 	stefgrpc "github.com/splunk/stef/go/grpc"
 	"github.com/splunk/stef/go/grpc/stef_proto"
@@ -32,11 +34,14 @@ type stefReceiver struct {
 	nextMetricsConsumer consumer.Metrics
 	settings            receiver.Settings
 
-	eg errgroup.Group
+	stopping atomic.Bool
+	eg       errgroup.Group
 }
 
 // Start runs the STEF gRPC receiver.
 func (r *stefReceiver) Start(ctx context.Context, host component.Host) error {
+	r.stopping.Store(false)
+
 	var err error
 	if r.serverGRPC, err = r.cfg.ToServer(ctx, host, r.settings.TelemetrySettings); err != nil {
 		return err
@@ -75,8 +80,22 @@ func (r *stefReceiver) Start(ctx context.Context, host component.Host) error {
 
 // Shutdown is a method to turn off receiving.
 func (r *stefReceiver) Shutdown(_ context.Context) error {
+	r.stopping.Store(true)
+
 	if r.serverGRPC != nil {
+		r.settings.Logger.Info("Stopping STEF/gRPC server", zap.String("endpoint", r.cfg.NetAddr.Endpoint))
+
+		// Give graceful stop a second to finish.
+		timer := time.AfterFunc(
+			1*time.Second, func() {
+				r.settings.Logger.Info("STEF/gRPC server couldn't stop gracefully in time. Doing force stop.")
+				r.serverGRPC.Stop()
+			},
+		)
+		defer timer.Stop()
+
 		r.serverGRPC.GracefulStop()
+		r.settings.Logger.Debug("STEF/gRPC server stopped.")
 	}
 
 	return r.eg.Wait()
@@ -100,6 +119,12 @@ func (r *stefReceiver) onStream(grpcReader stefgrpc.GrpcReader, stream stefgrpc.
 
 	// Read, decode, convert the incoming data and push it to the next consumer.
 	for {
+		if r.stopping.Load() {
+			// The receiver is shutting down. Close the connection.
+			r.settings.Logger.Debug("Shutdown requested. Closing STEF/gRPC connection.")
+			return nil
+		}
+
 		respError := resp.LastError()
 		if respError != nil {
 			// We had problem sending responses. Can't continue using this connection since