fix disabling consumergroup or topic collection

Author: weeco

docs/reference-config.yaml

@@ -90,6 +90,8 @@ minion:
     # take precedence over allowed groups.
     ignoredGroups: [ ]
   topics:
+    # Enabled can be set to false in order to disable collecting any topic metrics.
+    enabled: true
     # Granularity can be per topic or per partition. If you want to reduce the number of exported metric series and
     # you aren't interested in per partition metrics you could choose "topic".
     granularity: partition

go.sum

@@ -366,9 +366,8 @@ golang.org/x/net v0.0.0-20220725212005-46097bf591d3/go.mod h1:AaygXjzTFtRAg2ttMY
 golang.org/x/net v0.0.0-20220812174116-3211cb980234/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
-golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
-golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=

minion/config_topic_config.go

@@ -10,6 +10,9 @@ const (
 )
 
 type TopicConfig struct {
+	// Enabled can be set to false in order to not collect any topic metrics at all.
+	Enabled bool `koanf:"enabled"`
+
 	// Granularity can be per topic or per partition. If you want to reduce the number of exported metric series and
 	// you aren't interested in per partition metrics you could choose "topic".
 	Granularity string `koanf:"granularity"`
@@ -60,6 +63,7 @@ func (c *TopicConfig) Validate() error {
 
 // SetDefaults for topic config
 func (c *TopicConfig) SetDefaults() {
+	c.Enabled = true
 	c.Granularity = TopicGranularityPartition
 	c.AllowedTopics = []string{"/.*/"}
 	c.InfoMetric = InfoMetricConfig{ConfigKeys: []string{"cleanup.policy"}}

minion/service.go

@@ -10,12 +10,13 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cloudhut/kminion/v2/kafka"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/kmsg"
 	"github.com/twmb/franz-go/pkg/kversion"
 	"go.uber.org/zap"
 	"golang.org/x/sync/singleflight"
+
+	"github.com/cloudhut/kminion/v2/kafka"
 )
 
 type Service struct {
@@ -47,7 +48,7 @@ func NewService(cfg Config, logger *zap.Logger, kafkaSvc *kafka.Service, metrics
 	kgoOpts := []kgo.Opt{
 		kgo.WithHooks(minionHooks),
 	}
-	if cfg.ConsumerGroups.ScrapeMode == ConsumerGroupScrapeModeOffsetsTopic {
+	if cfg.ConsumerGroups.Enabled && cfg.ConsumerGroups.ScrapeMode == ConsumerGroupScrapeModeOffsetsTopic {
 		kgoOpts = append(kgoOpts,
 			kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
 			kgo.ConsumeTopics("__consumer_offsets"))
@@ -94,7 +95,7 @@ func (s *Service) Start(ctx context.Context) error {
 		return fmt.Errorf("failed to check feature compatibility against Kafka: %w", err)
 	}
 
-	if s.Cfg.ConsumerGroups.ScrapeMode == ConsumerGroupScrapeModeOffsetsTopic {
+	if s.Cfg.ConsumerGroups.Enabled && s.Cfg.ConsumerGroups.ScrapeMode == ConsumerGroupScrapeModeOffsetsTopic {
 		go s.startConsumingOffsets(ctx)
 	}
 

prometheus/collect_consumer_group_lags.go

@@ -2,13 +2,15 @@ package prometheus
 
 import (
 	"context"
-	"github.com/cloudhut/kminion/v2/minion"
+	"math"
+	"strconv"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kmsg"
 	"go.uber.org/zap"
-	"math"
-	"strconv"
+
+	"github.com/cloudhut/kminion/v2/minion"
 )
 
 type waterMark struct {
@@ -19,6 +21,10 @@ type waterMark struct {
 }
 
 func (e *Exporter) collectConsumerGroupLags(ctx context.Context, ch chan<- prometheus.Metric) bool {
+	if !e.minionSvc.Cfg.ConsumerGroups.Enabled {
+		return true
+	}
+
 	// Low Watermarks (at the moment they are not needed at all, they could be used to calculate the lag on partitions
 	// that don't have any active offsets)
 	lowWaterMarks, err := e.minionSvc.ListOffsetsCached(ctx, -2)

prometheus/collect_topic_info.go

@@ -10,6 +10,10 @@ import (
 )
 
 func (e *Exporter) collectTopicInfo(ctx context.Context, ch chan<- prometheus.Metric) bool {
+	if !e.minionSvc.Cfg.Topics.Enabled {
+		return true
+	}
+
 	metadata, err := e.minionSvc.GetMetadataCached(ctx)
 	if err != nil {
 		e.logger.Error("failed to get metadata", zap.Error(err))

prometheus/collect_topic_partition_offsets.go

@@ -2,14 +2,20 @@ package prometheus
 
 import (
 	"context"
-	"github.com/cloudhut/kminion/v2/minion"
+	"strconv"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/twmb/franz-go/pkg/kerr"
 	"go.uber.org/zap"
-	"strconv"
+
+	"github.com/cloudhut/kminion/v2/minion"
 )
 
 func (e *Exporter) collectTopicPartitionOffsets(ctx context.Context, ch chan<- prometheus.Metric) bool {
+	if !e.minionSvc.Cfg.Topics.Enabled {
+		return true
+	}
+
 	isOk := true
 
 	// Low Watermarks