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,9 +23,16 @@ 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))
+	groupStatesMap := s.Cfg.ConsumerGroups.GetAllowedConsumerGroupStates()
+
 	for i, group := range groupsRes.AllowedGroups.Groups {
-		if group.GroupState != "Dead" {
+		if len(groupStatesMap) == 0 {
 			groupIDs[i] = group.Group
+		} else {
+			// only add group if it's state is allowed
+			if _, ok := groupStatesMap[group.GroupState]; ok {
+				groupIDs[i] = group.Group
+			}
 		}
 	}
 

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

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