Collect metrics for a set of CG states (#7)

* Filter empty consumer groups

* Collect metrics for a set of CG states

* Update minion/config_consumer_group.go

Co-authored-by: Adrian Muraru <[email protected]>

* Implement review

* Implement review

* Update the CG state values

---------

Co-authored-by: Adrian Muraru <[email protected]>

Author: dobrerazvan

minion/config_consumer_group.go

@@ -31,6 +31,11 @@ type ConsumerGroupConfig struct {
 	// IgnoredGroups are regex strings of group ids that shall be ignored/skipped when exporting metrics. Ignored groups
 	// take precedence over allowed groups.
 	IgnoredGroupIDs []string `koanf:"ignoredGroups"`
+
+	// Monitor consumer group states. Empty list means all consumer groups are monitoring regardless of its state
+	// Allowed values are: Dead, Empty, Stable, PreparingRebalance, CompletingRebalance
+	// Source: https://github.com/apache/kafka/blob/3.4/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
+	AllowedConsumerGroupStates []string `koanf:"allowedConsumerGroupStates"`
 }
 
 func (c *ConsumerGroupConfig) SetDefaults() {
@@ -76,3 +81,13 @@ func (c *ConsumerGroupConfig) Validate() error {
 
 	return nil
 }
+
+// Returns a map of allowed group states for faster lookup
+func (c *ConsumerGroupConfig) GetAllowedConsumerGroupStates() map[string]string {
+	// create a map for faster lookup
+	groupStatesMap := make(map[string]string, len(c.AllowedConsumerGroupStates))
+	for _, state := range c.AllowedConsumerGroupStates {
+		groupStatesMap[state] = state
+	}
+	return groupStatesMap
+}

minion/consumer_group_offsets.go

@@ -23,10 +23,9 @@ func (s *Service) ListAllConsumerGroupOffsetsAdminAPI(ctx context.Context) (map[
 		return nil, fmt.Errorf("failed to list groupsRes: %w", err)
 	}
 	groupIDs := make([]string, len(groupsRes.AllowedGroups.Groups))
+
 	for i, group := range groupsRes.AllowedGroups.Groups {
-		if group.GroupState != "Dead" {
-			groupIDs[i] = group.Group
-		}
+		groupIDs[i] = group.Group
 	}
 
 	return s.listConsumerGroupOffsetsBulk(ctx, groupIDs)

minion/describe_consumer_groups.go

@@ -33,8 +33,9 @@ func (s *Service) listConsumerGroupsCached(ctx context.Context) (*GroupsInfo, er
 			return nil, err
 		}
 		allowedGroups := make([]kmsg.ListGroupsResponseGroup, 0)
+
 		for i := range res.Groups {
-			if s.IsGroupAllowed(res.Groups[i].Group) {
+			if s.IsGroupAllowed(res.Groups[i].Group, res.Groups[i].GroupState) {
 				allowedGroups = append(allowedGroups, res.Groups[i])
 			}
 		}

minion/storage.go

@@ -108,7 +108,7 @@ func (s *Storage) getNumberOfConsumedRecords() float64 {
 	return s.consumedRecords.Load()
 }
 
-func (s *Storage) getGroupOffsets(isAllowed func(groupName string) bool) map[string]map[string]map[int32]OffsetCommit {
+func (s *Storage) getGroupOffsets(isAllowed func(groupName string, groupState string) bool) map[string]map[string]map[int32]OffsetCommit {
 	// Offsets by group, topic, partition
 	offsetsByGroup := make(map[string]map[string]map[int32]OffsetCommit)
 
@@ -121,7 +121,7 @@ func (s *Storage) getGroupOffsets(isAllowed func(groupName string) bool) map[str
 	for _, offset := range offsets {
 		val := offset.(OffsetCommit)
 
-		if !isAllowed(val.Key.Group) {
+		if !isAllowed(val.Key.Group, "") {
 			continue
 		}
 

minion/utils.go

@@ -6,7 +6,7 @@ import (
 	"strings"
 )
 
-func (s *Service) IsGroupAllowed(groupName string) bool {
+func (s *Service) IsGroupAllowed(groupName string, groupState string) bool {
 	isAllowed := false
 	for _, regex := range s.AllowedGroupIDsExpr {
 		if regex.MatchString(groupName) {
@@ -21,6 +21,15 @@ func (s *Service) IsGroupAllowed(groupName string) bool {
 			break
 		}
 	}
+
+	if isAllowed && groupState != "" {
+		groupStatesMap := s.Cfg.ConsumerGroups.GetAllowedConsumerGroupStates()
+		if len(groupStatesMap) > 0 {
+			if _, ok := groupStatesMap[groupState]; !ok {
+				isAllowed = false
+			}
+		}
+	}
 	return isAllowed
 }
 

prometheus/collect_broker_info.go

@@ -2,9 +2,10 @@ package prometheus
 
 import (
 	"context"
+	"strconv"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
-	"strconv"
 )
 
 func (e *Exporter) collectBrokerInfo(ctx context.Context, ch chan<- prometheus.Metric) bool {

prometheus/collect_cluster_info.go

@@ -2,9 +2,10 @@ package prometheus
 
 import (
 	"context"
+	"strconv"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
-	"strconv"
 )
 
 func (e *Exporter) collectClusterInfo(ctx context.Context, ch chan<- prometheus.Metric) bool {

prometheus/collect_exporter_metrics.go

@@ -2,6 +2,7 @@ package prometheus
 
 import (
 	"context"
+
 	"github.com/prometheus/client_golang/prometheus"
 )
 

prometheus/collect_log_dirs.go

@@ -2,11 +2,12 @@ package prometheus
 
 import (
 	"context"
+	"strconv"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"go.uber.org/zap"
-	"strconv"
 )
 
 func (e *Exporter) collectLogDirs(ctx context.Context, ch chan<- prometheus.Metric) bool {

prometheus/collect_topic_partition_offsets.go

@@ -4,10 +4,9 @@ import (
 	"context"
 	"strconv"
 
+	"github.com/cloudhut/kminion/v2/minion"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
-
-	"github.com/cloudhut/kminion/v2/minion"
 )
 
 func (e *Exporter) collectTopicPartitionOffsets(ctx context.Context, ch chan<- prometheus.Metric) bool {