[receiver/kafkametricsreceiver] Add support for using franz-go client under a feature gate

.chloggen/feat_41480.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: "receiver/kafkametricsreceiver"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add support for using franz-go client under a feature gate"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [41480]
+
+# (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: [user]

exporter/kafkaexporter/go.mod

@@ -97,6 +97,7 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect
+	github.com/twmb/franz-go/pkg/kadm v1.16.1 // indirect
 	github.com/twmb/franz-go/pkg/kmsg v1.11.2 // indirect
 	github.com/twmb/franz-go/pkg/sasl/kerberos v1.1.0 // indirect
 	github.com/twmb/franz-go/plugin/kzap v1.1.2 // indirect

extension/observer/kafkatopicsobserver/go.mod

@@ -68,6 +68,7 @@ require (
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/twmb/franz-go v1.19.5 // indirect
+	github.com/twmb/franz-go/pkg/kadm v1.16.1 // indirect
 	github.com/twmb/franz-go/pkg/kmsg v1.11.2 // indirect
 	github.com/twmb/franz-go/pkg/sasl/kerberos v1.1.0 // indirect
 	github.com/twmb/franz-go/plugin/kzap v1.1.2 // indirect

internal/kafka/franz_client.go

@@ -14,6 +14,7 @@ import (
 	krb5client "github.com/jcmturner/gokrb5/v8/client"
 	krb5config "github.com/jcmturner/gokrb5/v8/config"
 	"github.com/jcmturner/gokrb5/v8/keytab"
+	"github.com/twmb/franz-go/pkg/kadm"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/kmsg"
 	"github.com/twmb/franz-go/pkg/kversion"
@@ -179,6 +180,39 @@ func NewFranzConsumerGroup(ctx context.Context, clientCfg configkafka.ClientConf
 	return kgo.NewClient(opts...)
 }
 
+// NewFranzClient creates a franz-go client using the same commonOpts used for producer/consumer.
+func NewFranzClient(
+	ctx context.Context,
+	clientCfg configkafka.ClientConfig,
+	logger *zap.Logger,
+	opts ...kgo.Opt,
+) (*kgo.Client, error) {
+	opts, err := commonOpts(ctx, clientCfg, logger, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return kgo.NewClient(opts...)
+}
+
+// NewFranzClusterAdminClient creates a kadm admin client from a freshly created franz client.
+func NewFranzClusterAdminClient(
+	ctx context.Context,
+	clientCfg configkafka.ClientConfig,
+	logger *zap.Logger,
+	opts ...kgo.Opt,
+) (*kadm.Client, *kgo.Client, error) {
+	cl, err := NewFranzClient(ctx, clientCfg, logger, opts...)
+	if err != nil {
+		return nil, nil, err
+	}
+	return kadm.NewClient(cl), cl, nil
+}
+
+// NewFranzAdminFromClient returns a kadm admin bound to an existing kgo client.
+func NewFranzAdminFromClient(cl *kgo.Client) *kadm.Client {
+	return kadm.NewClient(cl)
+}
+
 func commonOpts(ctx context.Context, clientCfg configkafka.ClientConfig,
 	logger *zap.Logger,
 	opts ...kgo.Opt,

internal/kafka/franz_client_test.go

@@ -594,3 +594,25 @@ func TestFranzClient_ProtocolVersion(t *testing.T) {
 		})
 	}
 }
+
+func TestNewFranzClient_And_Admin(t *testing.T) {
+	_, clientCfg := kafkatest.NewCluster(t, kfake.SeedTopics(1, "meta-topic"))
+	tl := zaptest.NewLogger(t, zaptest.Level(zap.WarnLevel))
+
+	// Plain client
+	cl, err := NewFranzClient(t.Context(), clientCfg, tl)
+	require.NoError(t, err)
+	t.Cleanup(cl.Close)
+
+	// Admin from fresh client
+	ad, cl2, err := NewFranzClusterAdminClient(t.Context(), clientCfg, tl)
+	require.NoError(t, err)
+	t.Cleanup(func() { ad.Close(); cl2.Close() })
+
+	// Metadata via admin should return brokers & topic
+	md, err := ad.Metadata(t.Context(), "meta-topic")
+	require.NoError(t, err)
+	assert.NotEmpty(t, md.Brokers)
+	_, ok := md.Topics["meta-topic"]
+	assert.True(t, ok)
+}

internal/kafka/go.mod

@@ -58,7 +58,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/rogpeppe/go-internal v1.13.1 // indirect
 	github.com/twmb/franz-go v1.19.5
-	github.com/twmb/franz-go/pkg/kadm v1.16.1 // indirect
+	github.com/twmb/franz-go/pkg/kadm v1.16.1
 	github.com/twmb/franz-go/pkg/kmsg v1.11.2
 	github.com/twmb/franz-go/plugin/kzap v1.1.2
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect

receiver/kafkametricsreceiver/README.md

@@ -24,13 +24,18 @@ This receiver supports Kafka versions:
 
 ## Getting Started
 
+> [!NOTE]
+> You can opt-in to use [`franz-go`](https://github.com/twmb/franz-go) client by enabling the feature gate
+> `receiver.kafkametricsreceiver.UseFranzGo` when you run the OpenTelemetry Collector. See the following page
+> for more details: [Feature Gates](https://github.com/open-telemetry/opentelemetry-collector/tree/main/featuregate#controlling-gates)
+
 Required settings (no defaults):
 
 - `scrapers`: any combination of the following scrapers can be enabled.
     - `topics`
     - `consumers`
     - `brokers`
-    
+
 Metrics collected by the associated scraper are listed in [metadata.yaml](metadata.yaml)
 
 Optional Settings (with defaults):

receiver/kafkametricsreceiver/broker_scraper_franz.go

@@ -0,0 +1,140 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kafkametricsreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkametricsreceiver"
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/twmb/franz-go/pkg/kadm"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/receiver"
+	"go.opentelemetry.io/collector/scraper"
+	"go.opentelemetry.io/collector/scraper/scrapererror"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkametricsreceiver/internal/metadata"
+)
+
+type brokerScraperFranz struct {
+	// franz-go handles (lazy created on first scrape)
+	adm *kadm.Client
+	cl  *kgo.Client
+
+	settings receiver.Settings
+	config   Config
+	mb       *metadata.MetricsBuilder
+}
+
+func (s *brokerScraperFranz) start(_ context.Context, _ component.Host) error {
+	s.mb = metadata.NewMetricsBuilder(s.config.MetricsBuilderConfig, s.settings)
+	return nil
+}
+
+func (s *brokerScraperFranz) shutdown(context.Context) error {
+	if s.adm != nil {
+		s.adm.Close()
+		s.adm = nil
+	}
+	if s.cl != nil {
+		s.cl.Close()
+		s.cl = nil
+	}
+	return nil
+}
+
+func (s *brokerScraperFranz) ensureClients(ctx context.Context) error {
+	if s.cl != nil && s.adm != nil {
+		return nil
+	}
+	adm, cl, err := kafka.NewFranzClusterAdminClient(ctx, s.config.ClientConfig, s.settings.Logger)
+	if err != nil {
+		return fmt.Errorf("failed to create franz-go admin client: %w", err)
+	}
+	s.adm = adm
+	s.cl = cl
+	return nil
+}
+
+func (s *brokerScraperFranz) scrape(ctx context.Context) (pmetric.Metrics, error) {
+	scrapeErrs := scrapererror.ScrapeErrors{}
+
+	if err := s.ensureClients(ctx); err != nil {
+		return pmetric.Metrics{}, err
+	}
+
+	now := pcommon.NewTimestampFromTime(time.Now())
+	rb := s.mb.NewResourceBuilder()
+	rb.SetKafkaClusterAlias(s.config.ClusterAlias)
+
+	// ---- brokers count ----
+	bdetails, err := s.adm.ListBrokers(ctx)
+	if err != nil {
+		// If we cannot list brokers, emit what we have (resource attrs) and return the error
+		scrapeErrs.Add(err)
+		return s.mb.Emit(metadata.WithResource(rb.Emit())), scrapeErrs.Combine()
+	}
+	brokerIDs := bdetails.NodeIDs()
+	s.mb.RecordKafkaBrokersDataPoint(now, int64(len(brokerIDs)))
+
+	// If log retention metric is disabled, we are done.
+	if !s.config.Metrics.KafkaBrokerLogRetentionPeriod.Enabled {
+		return s.mb.Emit(metadata.WithResource(rb.Emit())), scrapeErrs.Combine()
+	}
+
+	res, err := s.adm.DescribeBrokerConfigs(ctx, brokerIDs...)
+	if err != nil {
+		s.settings.Logger.Warn("franz-go: DescribeBrokerConfigs failed", zap.Error(err))
+		scrapeErrs.AddPartial(len(brokerIDs), fmt.Errorf("DescribeBrokerConfigs: %w", err))
+	}
+
+	// Iterate the result and record the metric for each broker entry we can parse.
+	for _, bid := range brokerIDs {
+		bidStr := strconv.Itoa(int(bid))
+
+		// Look up this broker's config set by resource name (broker id as string).
+		cfg, _ := res.On(bidStr, nil) // fn can be nil to just return the entry
+		if cfg.Err != nil {
+			scrapeErrs.AddPartial(1, fmt.Errorf("broker %s: %w", bidStr, cfg.Err))
+			continue
+		}
+
+		for _, kv := range cfg.Configs {
+			// kadm.Config has Key and MaybeValue() for the string value.
+			// We only care about log.retention.hours here.
+			if kv.Key != logRetentionHours {
+				continue
+			}
+			raw := kv.MaybeValue()
+			hrs, convErr := strconv.Atoi(raw)
+			if convErr != nil {
+				scrapeErrs.AddPartial(1, fmt.Errorf("broker %s: cannot parse %s=%q: %w", bidStr, logRetentionHours, raw, convErr))
+				continue
+			}
+			sec := int64(hrs) * 3600
+			s.mb.RecordKafkaBrokerLogRetentionPeriodDataPoint(now, sec, bidStr)
+		}
+	}
+
+	return s.mb.Emit(metadata.WithResource(rb.Emit())), scrapeErrs.Combine()
+}
+
+// factory for franz-go scraper (internal; selected via gate at the call site later)
+func createBrokerScraperFranz(_ context.Context, cfg Config, settings receiver.Settings) (scraper.Metrics, error) {
+	s := &brokerScraperFranz{
+		settings: settings,
+		config:   cfg,
+	}
+	return scraper.NewMetrics(
+		s.scrape,
+		scraper.WithStart(s.start),
+		scraper.WithShutdown(s.shutdown),
+	)
+}