[receiver/kafkametrics] Implement client reset for recoverable errors in Sarama calls (#41363)

…

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
This change implements client reset functionality to address recoverable
errors in Sarama calls, such as connection resets and EOF errors. When a
recoverable error is encountered, the client is reset, enabling the
scraper to reconnect and resume metric collection seamlessly.
<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
none
<!--Describe what testing was performed and which tests were added.-->
#### Testing
Existing tests were modified accordingly.
<!--Describe the documentation added.-->
#### Documentation
none
<!--Please delete paragraphs that you did not use before submitting.-->

Author: yurytaranau

.chloggen/kafkametricsreceiver-retryable-errors.yaml

@@ -0,0 +1,28 @@
+# 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: kafkametricsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Implement client reset for recoverable errors in Sarama calls
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [41363]
+
+# (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: |
+  This change implements client reset functionality to address recoverable errors in Sarama calls, such as connection resets and EOF errors. When a recoverable error is encountered, the client is reset, enabling the scraper to reconnect and resume metric collection seamlessly.
+
+# 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/kafkametricsreceiver/broker_scraper.go

@@ -48,7 +48,7 @@ func (s *brokerScraper) shutdown(context.Context) error {
 func (s *brokerScraper) scrape(context.Context) (pmetric.Metrics, error) {
 	scrapeErrors := scrapererror.ScrapeErrors{}
 
-	if s.client == nil {
+	if s.client == nil || s.client.Closed() {
 		client, err := newSaramaClient(context.Background(), s.config.ClientConfig)
 		if err != nil {
 			return pmetric.Metrics{}, fmt.Errorf("failed to create client in brokers scraper: %w", err)

receiver/kafkametricsreceiver/broker_scraper_test.go

@@ -88,6 +88,7 @@ func TestBrokerScraper_shutdown_handles_nil_client(t *testing.T) {
 func TestBrokerScraper_empty_resource_attribute(t *testing.T) {
 	client := newMockClient()
 	client.Mock.On("Brokers").Return(testBrokers)
+	client.Mock.On("Closed").Return(false)
 	bs := brokerScraper{
 		client:   client,
 		settings: receivertest.NewNopSettings(metadata.Type),
@@ -108,6 +109,7 @@ func TestBrokerScraper_empty_resource_attribute(t *testing.T) {
 func TestBrokerScraper_scrape(t *testing.T) {
 	client := newMockClient()
 	client.Mock.On("Brokers").Return(testBrokers)
+	client.Mock.On("Closed").Return(false)
 	bs := brokerScraper{
 		client:   client,
 		settings: receivertest.NewNopSettings(metadata.Type),

receiver/kafkametricsreceiver/consumer_scraper.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"sync"
 	"time"
 
 	"github.com/IBM/sarama"
@@ -28,6 +29,7 @@ type consumerScraper struct {
 	clusterAdmin sarama.ClusterAdmin
 	config       Config
 	mb           *metadata.MetricsBuilder
+	mu           sync.Mutex
 }
 
 func (s *consumerScraper) start(_ context.Context, _ component.Host) error {
@@ -43,7 +45,7 @@ func (s *consumerScraper) shutdown(_ context.Context) error {
 }
 
 func (s *consumerScraper) scrape(context.Context) (pmetric.Metrics, error) {
-	if s.client == nil {
+	if s.client == nil || s.client.Closed() {
 		client, err := newSaramaClient(context.Background(), s.config.ClientConfig)
 		if err != nil {
 			return pmetric.Metrics{}, fmt.Errorf("failed to create client in consumer scraper: %w", err)
@@ -64,7 +66,7 @@ func (s *consumerScraper) scrape(context.Context) (pmetric.Metrics, error) {
 
 	cgs, listErr := s.clusterAdmin.ListConsumerGroups()
 	if listErr != nil {
-		return pmetric.Metrics{}, listErr
+		return pmetric.Metrics{}, s.resetClientOnError(listErr)
 	}
 
 	var matchedGrpIDs []string
@@ -76,7 +78,7 @@ func (s *consumerScraper) scrape(context.Context) (pmetric.Metrics, error) {
 
 	allTopics, listErr := s.clusterAdmin.ListTopics()
 	if listErr != nil {
-		return pmetric.Metrics{}, listErr
+		return pmetric.Metrics{}, s.resetClientOnError(listErr)
 	}
 
 	matchedTopics := map[string]sarama.TopicDetail{}
@@ -110,7 +112,7 @@ func (s *consumerScraper) scrape(context.Context) (pmetric.Metrics, error) {
 	}
 	consumerGroups, listErr := s.clusterAdmin.DescribeConsumerGroups(matchedGrpIDs)
 	if listErr != nil {
-		return pmetric.Metrics{}, listErr
+		return pmetric.Metrics{}, s.resetClientOnError(listErr)
 	}
 
 	now := pcommon.NewTimestampFromTime(time.Now())
@@ -186,3 +188,15 @@ func createConsumerScraper(_ context.Context, cfg Config, settings receiver.Sett
 		scraper.WithShutdown(s.shutdown),
 	)
 }
+
+func (s *consumerScraper) resetClientOnError(err error) error {
+	if isRecoverableError(err) {
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		s.clusterAdmin.Close()
+		s.clusterAdmin = nil
+		return fmt.Errorf("closing client because of reconnection error %w", err)
+	}
+
+	return err
+}

receiver/kafkametricsreceiver/consumer_scraper_test.go

@@ -123,14 +123,16 @@ func TestConsumerScraper_createScraper_handles_invalid_group_match(t *testing.T)
 }
 
 func TestConsumerScraper_scrape(t *testing.T) {
+	client := newMockClient()
 	filter := regexp.MustCompile(defaultGroupMatch)
 	cs := consumerScraper{
-		client:       newMockClient(),
+		client:       client,
 		settings:     receivertest.NewNopSettings(metadata.Type),
 		clusterAdmin: newMockClusterAdmin(),
 		topicFilter:  filter,
 		groupFilter:  filter,
 	}
+	client.Mock.On("Closed").Return(false)
 	require.NoError(t, cs.start(t.Context(), componenttest.NewNopHost()))
 	md, err := cs.scrape(t.Context())
 	assert.NoError(t, err)
@@ -149,36 +151,41 @@ func TestConsumerScraper_scrape_handlesListTopicError(t *testing.T) {
 		topicFilter:  filter,
 		groupFilter:  filter,
 	}
+	client.Mock.On("Closed").Return(false)
 	_, err := cs.scrape(t.Context())
 	assert.Error(t, err)
 }
 
 func TestConsumerScraper_scrape_handlesListConsumerGroupError(t *testing.T) {
+	client := newMockClient()
 	filter := regexp.MustCompile(defaultGroupMatch)
 	clusterAdmin := newMockClusterAdmin()
 	clusterAdmin.consumerGroups = nil
 	cs := consumerScraper{
-		client:       newMockClient(),
+		client:       client,
 		settings:     receivertest.NewNopSettings(metadata.Type),
 		clusterAdmin: clusterAdmin,
 		topicFilter:  filter,
 		groupFilter:  filter,
 	}
+	client.Mock.On("Closed").Return(false)
 	_, err := cs.scrape(t.Context())
 	assert.Error(t, err)
 }
 
 func TestConsumerScraper_scrape_handlesDescribeConsumerError(t *testing.T) {
+	client := newMockClient()
 	filter := regexp.MustCompile(defaultGroupMatch)
 	clusterAdmin := newMockClusterAdmin()
 	clusterAdmin.consumerGroupDescriptions = nil
 	cs := consumerScraper{
-		client:       newMockClient(),
+		client:       client,
 		settings:     receivertest.NewNopSettings(metadata.Type),
 		clusterAdmin: clusterAdmin,
 		topicFilter:  filter,
 		groupFilter:  filter,
 	}
+	client.Mock.On("Closed").Return(false)
 	_, err := cs.scrape(t.Context())
 	assert.Error(t, err)
 }
@@ -196,6 +203,7 @@ func TestConsumerScraper_scrape_handlesOffsetPartialError(t *testing.T) {
 		topicFilter:  filter,
 		clusterAdmin: clusterAdmin,
 	}
+	client.Mock.On("Closed").Return(false)
 	require.NoError(t, cs.start(t.Context(), componenttest.NewNopHost()))
 	_, err := cs.scrape(t.Context())
 	assert.Error(t, err)
@@ -214,6 +222,7 @@ func TestConsumerScraper_scrape_handlesPartitionPartialError(t *testing.T) {
 		topicFilter:  filter,
 		clusterAdmin: clusterAdmin,
 	}
+	client.Mock.On("Closed").Return(false)
 	require.NoError(t, cs.start(t.Context(), componenttest.NewNopHost()))
 	_, err := cs.scrape(t.Context())
 	assert.Error(t, err)

receiver/kafkametricsreceiver/receiver.go

@@ -5,7 +5,12 @@ package kafkametricsreceiver // import "github.com/open-telemetry/opentelemetry-
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
+	"net"
+	"os"
+	"syscall"
 
 	"github.com/IBM/sarama"
 	"go.opentelemetry.io/collector/component"
@@ -60,3 +65,38 @@ var newMetricsReceiver = func(
 		scraperControllerOptions...,
 	)
 }
+
+// isRecoverableError checks if the error can be resolved by re-establishing connection
+func isRecoverableError(err error) bool {
+	if errors.Is(err, sarama.ErrOutOfBrokers) {
+		return true
+	}
+
+	if errors.Is(err, sarama.ErrClosedClient) {
+		return true
+	}
+
+	if errors.Is(err, os.ErrDeadlineExceeded) {
+		// Error example: read tcp 10.2.3.4:62523->4.3.2.1:9093: i/o timeout
+		return true
+	}
+
+	if errors.Is(err, syscall.EPIPE) {
+		return true
+	}
+
+	if errors.Is(err, net.ErrClosed) {
+		return true
+	}
+
+	if errors.Is(err, syscall.ECONNRESET) {
+		// Error example: write tcp 1.2.3.4:56532->4.3.2.1:9093: write: connection reset by peer
+		return true
+	}
+
+	if errors.Is(err, io.EOF) {
+		return true
+	}
+
+	return false
+}

receiver/kafkametricsreceiver/topic_scraper.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"regexp"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/IBM/sarama"
@@ -29,6 +30,7 @@ type topicScraper struct {
 	topicFilter  *regexp.Regexp
 	config       Config
 	mb           *metadata.MetricsBuilder
+	mu           sync.Mutex
 }
 
 const (
@@ -50,7 +52,7 @@ func (s *topicScraper) start(_ context.Context, _ component.Host) error {
 }
 
 func (s *topicScraper) scrape(context.Context) (pmetric.Metrics, error) {
-	if s.client == nil {
+	if s.client == nil || s.client.Closed() {
 		client, err := newSaramaClient(context.Background(), s.config.ClientConfig)
 		if err != nil {
 			return pmetric.Metrics{}, fmt.Errorf("failed to create client in topics scraper: %w", err)
@@ -61,7 +63,7 @@ func (s *topicScraper) scrape(context.Context) (pmetric.Metrics, error) {
 	topics, err := s.client.Topics()
 	if err != nil {
 		s.settings.Logger.Error("Error fetching cluster topics ", zap.Error(err))
-		return pmetric.Metrics{}, err
+		return pmetric.Metrics{}, s.resetClientOnError(err)
 	}
 
 	scrapeErrors := scrapererror.ScrapeErrors{}
@@ -184,3 +186,15 @@ func createTopicsScraper(_ context.Context, cfg Config, settings receiver.Settin
 		scraper.WithShutdown(s.shutdown),
 	)
 }
+
+func (s *topicScraper) resetClientOnError(err error) error {
+	if isRecoverableError(err) {
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		s.client.Close()
+		s.client = nil
+		return fmt.Errorf("closing client because of reconnection error %w", err)
+	}
+
+	return err
+}

receiver/kafkametricsreceiver/topic_scraper_test.go

@@ -109,6 +109,7 @@ func TestTopicScraper_scrapes(t *testing.T) {
 		config:       *config,
 		topicFilter:  match,
 	}
+	client.Mock.On("Closed").Return(false)
 	require.NoError(t, scraper.start(t.Context(), componenttest.NewNopHost()))
 	md, err := scraper.scrape(t.Context())
 	assert.NoError(t, err)
@@ -153,6 +154,7 @@ func TestTopicScraper_scrape_handlesTopicError(t *testing.T) {
 		settings:    receivertest.NewNopSettings(metadata.Type),
 		topicFilter: match,
 	}
+	client.Mock.On("Closed").Return(false)
 	_, err := scraper.scrape(t.Context())
 	assert.Error(t, err)
 }
@@ -168,6 +170,7 @@ func TestTopicScraper_scrape_handlesPartitionError(t *testing.T) {
 		settings:     receivertest.NewNopSettings(metadata.Type),
 		topicFilter:  match,
 	}
+	client.Mock.On("Closed").Return(false)
 	require.NoError(t, scraper.start(t.Context(), componenttest.NewNopHost()))
 	_, err := scraper.scrape(t.Context())
 	assert.Error(t, err)
@@ -187,6 +190,7 @@ func TestTopicScraper_scrape_handlesPartialScrapeErrors(t *testing.T) {
 		settings:     receivertest.NewNopSettings(metadata.Type),
 		topicFilter:  match,
 	}
+	client.Mock.On("Closed").Return(false)
 	require.NoError(t, scraper.start(t.Context(), componenttest.NewNopHost()))
 	_, err := scraper.scrape(t.Context())
 	assert.Error(t, err)