[INTERNAL] Add a new metrics to count all groups included those ignored

This is useful to capture the total number of groups in kafka

Author: amuraru

minion/consumer_group_offsets.go

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

minion/describe_consumer_groups.go

@@ -16,13 +16,18 @@ type DescribeConsumerGroupsResponse struct {
 	Groups         *kmsg.DescribeGroupsResponse
 }
 
-func (s *Service) listConsumerGroupsCached(ctx context.Context) (*kmsg.ListGroupsResponse, error) {
-	key := "list-consumer-groups"
+type GroupsInfo struct {
+	AllowedGroups  *kmsg.ListGroupsResponse
+	AllGroupsCount int
+}
+
+func (s *Service) listConsumerGroupsCached(ctx context.Context) (*GroupsInfo, error) {
+	keyAllowedGroups := "list-consumer-groups"
 
-	if cachedRes, exists := s.getCachedItem(key); exists {
-		return cachedRes.(*kmsg.ListGroupsResponse), nil
+	if cachedRes, exists := s.getCachedItem(keyAllowedGroups); exists {
+		return cachedRes.(*GroupsInfo), nil
 	}
-	res, err, _ := s.requestGroup.Do(key, func() (interface{}, error) {
+	groups, err, _ := s.requestGroup.Do(keyAllowedGroups, func() (interface{}, error) {
 		res, err := s.listConsumerGroups(ctx)
 		if err != nil {
 			return nil, err
@@ -34,15 +39,19 @@ func (s *Service) listConsumerGroupsCached(ctx context.Context) (*kmsg.ListGroup
 			}
 		}
 		res.Groups = allowedGroups
-		s.setCachedItem(key, res, 120*time.Second)
+		groups := &GroupsInfo{
+			AllGroupsCount: len(res.Groups),
+			AllowedGroups:  res,
+		}
+		s.setCachedItem(keyAllowedGroups, groups, 120*time.Second)
 
-		return res, nil
+		return groups, nil
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	return res.(*kmsg.ListGroupsResponse), nil
+	return groups.(*GroupsInfo), nil
 }
 
 func (s *Service) listConsumerGroups(ctx context.Context) (*kmsg.ListGroupsResponse, error) {
@@ -59,14 +68,14 @@ func (s *Service) listConsumerGroups(ctx context.Context) (*kmsg.ListGroupsRespo
 	return res, nil
 }
 
-func (s *Service) DescribeConsumerGroups(ctx context.Context) ([]DescribeConsumerGroupsResponse, error) {
+func (s *Service) DescribeConsumerGroups(ctx context.Context) ([]DescribeConsumerGroupsResponse, int, error) {
 	listRes, err := s.listConsumerGroupsCached(ctx)
 	if err != nil {
-		return nil, err
+		return nil, -1, err
 	}
 
-	groupIDs := make([]string, len(listRes.Groups))
-	for i, group := range listRes.Groups {
+	groupIDs := make([]string, len(listRes.AllowedGroups.Groups))
+	for i, group := range listRes.AllowedGroups.Groups {
 		groupIDs[i] = group.Group
 	}
 
@@ -90,6 +99,5 @@ func (s *Service) DescribeConsumerGroups(ctx context.Context) ([]DescribeConsume
 			Groups:         res,
 		})
 	}
-
-	return describedGroups, nil
+	return describedGroups, listRes.AllGroupsCount, nil
 }

prometheus/collect_consumer_groups.go

@@ -15,7 +15,7 @@ func (e *Exporter) collectConsumerGroups(ctx context.Context, ch chan<- promethe
 	if !e.minionSvc.Cfg.ConsumerGroups.Enabled {
 		return true
 	}
-	groups, err := e.minionSvc.DescribeConsumerGroups(ctx)
+	groups, allGroups, err := e.minionSvc.DescribeConsumerGroups(ctx)
 	if err != nil {
 		e.logger.Error("failed to collect consumer groups, because Kafka request failed", zap.Error(err))
 		return false
@@ -143,6 +143,11 @@ func (e *Exporter) collectConsumerGroups(ctx context.Context, ch chan<- promethe
 			}
 		}
 	}
+	ch <- prometheus.MustNewConstMetric(
+		e.consumerGroupInfoAllGroups,
+		prometheus.GaugeValue,
+		float64(allGroups),
+	)
 	ch <- prometheus.MustNewConstMetric(
 		e.consumerGroupInfoEmptyGroups,
 		prometheus.GaugeValue,

prometheus/exporter.go

@@ -44,6 +44,7 @@ type Exporter struct {
 
 	// Consumer Groups
 	consumerGroupInfo                    *prometheus.Desc
+	consumerGroupInfoAllGroups           *prometheus.Desc
 	consumerGroupInfoEmptyGroups         *prometheus.Desc
 	consumerGroupMembers                 *prometheus.Desc
 	consumerGroupMembersEmpty            *prometheus.Desc
@@ -180,6 +181,13 @@ func (e *Exporter) InitializeMetrics() {
 		[]string{"group_id", "protocol", "protocol_type", "state", "coordinator_id"},
 		nil,
 	)
+	// Group Info - All Groups
+	e.consumerGroupInfoAllGroups = prometheus.NewDesc(
+		prometheus.BuildFQName(e.cfg.Namespace, "kafka", "consumer_group_info_count"),
+		"Consumer Group info metrics. It will report the number of all groups in cluster.",
+		nil,
+		nil,
+	)
 	// Group Info - Empty Groups
 	e.consumerGroupInfoEmptyGroups = prometheus.NewDesc(
 		prometheus.BuildFQName(e.cfg.Namespace, "kafka", "consumer_group_info_empty_groups"),