@@ -16,7 +16,7 @@ import (
1616 "go.opentelemetry.io/collector/pdata/pmetric"
1717 "go.opentelemetry.io/collector/receiver"
1818 "go.opentelemetry.io/collector/scraper"
19- "go.uber.org/multierr "
19+ "go.opentelemetry.io/collector/scraper/scrapererror "
2020
2121 "github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka"
2222 "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkametricsreceiver/internal/metadata"
@@ -70,108 +70,69 @@ func (s *consumerScraperFranz) scrape(ctx context.Context) (pmetric.Metrics, err
7070 return pmetric.Metrics {}, err
7171 }
7272
73- var scrapeErr error
74-
75- // 1) list & filter groups
76- lgs , err := s .adm .ListGroups (ctx )
73+ lgs , err := s .adm .ListGroupsByType (ctx , []string {"classic" , "consumer" })
7774 if err != nil {
78- return pmetric.Metrics {}, fmt .Errorf ("franz-go: ListGroups failed: %w" , err )
75+ return pmetric.Metrics {}, fmt .Errorf ("franz-go: ListGroupsByType failed: %w" , err )
7976 }
77+
8078 var matchedGrpIDs []string
81- for _ , g := range lgs . Sorted () {
79+ for _ , g := range lgs {
8280 if s .groupFilter .MatchString (g .Group ) {
8381 matchedGrpIDs = append (matchedGrpIDs , g .Group )
8482 }
8583 }
8684
87- // 2) list & filter topics
88- td , err := s .adm .ListTopics (ctx ) // non-internal only, same as sarama default
89- if err != nil {
90- return pmetric.Metrics {}, fmt .Errorf ("franz-go: ListTopics failed: %w" , err )
91- }
92- var matchedTopics []string
93- for t := range td {
94- if s .topicFilter .MatchString (t ) {
95- matchedTopics = append (matchedTopics , t )
96- }
97- }
98-
99- // 3) compute partition list + end offsets for matched topics
100- endOffsets , err := s .adm .ListEndOffsets (ctx , matchedTopics ... )
101- if err != nil {
102- return pmetric.Metrics {}, fmt .Errorf ("franz-go: ListEndOffsets failed: %w" , err )
103- }
104- // Build helpers equivalent to Sarama path
105- topicPartitions := make (map [string ][]int32 , len (matchedTopics ))
106- topicPartitionOffset := make (map [string ]map [int32 ]int64 , len (matchedTopics ))
107- endOffsets .Each (func (lo kadm.ListedOffset ) {
108- // lo.Topic, lo.Partition, lo.Offset
109- if _ , ok := topicPartitions [lo .Topic ]; ! ok {
110- topicPartitions [lo .Topic ] = []int32 {}
111- }
112- if _ , ok := topicPartitionOffset [lo .Topic ]; ! ok {
113- topicPartitionOffset [lo .Topic ] = map [int32 ]int64 {}
114- }
115- topicPartitions [lo .Topic ] = append (topicPartitions [lo .Topic ], lo .Partition )
116- topicPartitionOffset [lo.Topic ][lo.Partition ] = lo .Offset
117- })
118-
119- // 4) describe groups for member counts
120- dgs , err := s .adm .DescribeGroups (ctx , matchedGrpIDs ... )
85+ dgls , err := s .adm .Lag (ctx , matchedGrpIDs ... )
12186 if err != nil {
122- return pmetric.Metrics {}, fmt .Errorf ("franz-go: DescribeGroups failed: %w" , err )
87+ return pmetric.Metrics {}, fmt .Errorf ("franz-go: Lag failed: %w" , err )
12388 }
12489
90+ scrapeErrs := scrapererror.ScrapeErrors {}
12591 now := pcommon .NewTimestampFromTime (time .Now ())
126-
127- // 5) per group: fetch committed offsets for matched topics and compute metrics
128- gs := dgs .Sorted ()
129- for i := range gs {
130- g := & gs [i ]
131- grpID := g .Group
132- s .mb .RecordKafkaConsumerGroupMembersDataPoint (now , int64 (len (g .Members )), grpID )
133-
134- offsets , ferr := s .adm .FetchOffsetsForTopics (ctx , grpID , matchedTopics ... )
135- if ferr != nil {
136- scrapeErr = multierr .Append (scrapeErr , fmt .Errorf ("franz-go: FetchOffsetsForTopics(%s) failed: %w" , grpID , ferr ))
92+ for group := range dgls {
93+ dgl := dgls [group ]
94+ if dgl .DescribeErr != nil {
95+ scrapeErrs .AddPartial (1 , fmt .Errorf ("franz-go: returned error from describing the group. group=%s, error=%w" , group , dgl .DescribeErr ))
13796 continue
13897 }
139-
140- for topic , parts := range topicPartitions {
141- isConsumed := false
142- var lagSum int64
98+ s .mb .RecordKafkaConsumerGroupMembersDataPoint (now , int64 (len (dgl .Members )), group )
99+ if dgl .FetchErr != nil {
100+ scrapeErrs .AddPartial (1 , fmt .Errorf ("franz-go: returned error from fetching offsets. group=%s, error=%w" , group , dgl .FetchErr ))
101+ continue
102+ }
103+ for topic := range dgl .Lag {
104+ if ! s .topicFilter .MatchString (topic ) {
105+ continue
106+ }
107+ gmls := dgl .Lag [topic ]
108+ var isConsumed bool
143109 var offsetSum int64
144-
145- for _ , p := range parts {
146- consumerOffset := int64 (- 1 )
147- if or , ok := offsets .Lookup (topic , p ); ok && or .Err == nil {
148- consumerOffset = or .At
110+ var lagSum int64
111+ for partition := range gmls {
112+ gml := gmls [partition ]
113+ if gml .Err != nil {
114+ scrapeErrs .AddPartial (1 , fmt .Errorf ("franz-go: returned either the commit error, or the list end offsets error. group=%s, topic=%s, partition=%d, error=%w" , group , topic , partition , gml .Err ))
115+ continue
149116 }
150- offsetSum += consumerOffset
151- s .mb .RecordKafkaConsumerGroupOffsetDataPoint (now , consumerOffset , grpID , topic , int64 (p ))
152-
153- var consumerLag int64 = - 1
154- if consumerOffset != - 1 {
117+ if gml .Commit .At != - 1 {
155118 isConsumed = true
156- if end , ok := topicPartitionOffset [ topic ][ p ]; ok {
157- consumerLag = end - consumerOffset
158- lagSum += consumerLag
159- }
119+ offsetSum += gml . Commit . At
120+ lagSum += gml . Lag // franz-go clamps Lag to >= 0 and only returns Lag == -1 when gml.Err != nil
121+ s . mb . RecordKafkaConsumerGroupOffsetDataPoint ( now , gml . Commit . At , group , topic , int64 ( partition ))
122+ s . mb . RecordKafkaConsumerGroupLagDataPoint ( now , gml . Lag , group , topic , int64 ( partition ))
160123 }
161- s .mb .RecordKafkaConsumerGroupLagDataPoint (now , consumerLag , grpID , topic , int64 (p ))
162124 }
163-
164125 if isConsumed {
165- s .mb .RecordKafkaConsumerGroupOffsetSumDataPoint (now , offsetSum , grpID , topic )
166- s .mb .RecordKafkaConsumerGroupLagSumDataPoint (now , lagSum , grpID , topic )
126+ s .mb .RecordKafkaConsumerGroupOffsetSumDataPoint (now , offsetSum , group , topic )
127+ s .mb .RecordKafkaConsumerGroupLagSumDataPoint (now , lagSum , group , topic )
167128 }
168129 }
169130 }
170131
171132 rb := s .mb .NewResourceBuilder ()
172133 rb .SetKafkaClusterAlias (s .config .ClusterAlias )
173134
174- return s .mb .Emit (metadata .WithResource (rb .Emit ())), scrapeErr
135+ return s .mb .Emit (metadata .WithResource (rb .Emit ())), scrapeErrs . Combine ()
175136}
176137
177138// Factory helper for franz-go path (selected under the feature gate later).
0 commit comments