From b23ee5e445755bf9cb0e694375c92b239ecd2877 Mon Sep 17 00:00:00 2001 From: dttung2905 Date: Sat, 5 Apr 2025 23:04:06 +0100 Subject: [PATCH 1/2] Add ACL related metrics Signed-off-by: dttung2905 --- minion/acls.go | 24 ++++++++++++ prometheus/collect_acl_info.go | 69 ++++++++++++++++++++++++++++++++++ prometheus/exporter.go | 23 ++++++++++-- 3 files changed, 113 insertions(+), 3 deletions(-) create mode 100644 minion/acls.go create mode 100644 prometheus/collect_acl_info.go diff --git a/minion/acls.go b/minion/acls.go new file mode 100644 index 00000000..3c26c604 --- /dev/null +++ b/minion/acls.go @@ -0,0 +1,24 @@ +package minion + +import ( + "context" + "github.com/twmb/franz-go/pkg/kmsg" +) + +func (s *Service) ListAllACLs(ctx context.Context) (*kmsg.DescribeACLsResponse, error) { + req := kmsg.NewDescribeACLsRequest() + req.ResourceType = kmsg.ACLResourceTypeAny + req.ResourcePatternType = kmsg.ACLResourcePatternTypeAny + req.ResourceName = nil + req.Principal = nil + req.Host = nil + req.Operation = kmsg.ACLOperationAny + req.PermissionType = kmsg.ACLPermissionTypeAny + + res, err := req.RequestWith(ctx, s.client) + if err != nil { + return nil, err + } + + return res, nil +} diff --git a/prometheus/collect_acl_info.go b/prometheus/collect_acl_info.go new file mode 100644 index 00000000..24f39226 --- /dev/null +++ b/prometheus/collect_acl_info.go @@ -0,0 +1,69 @@ +package prometheus + +import ( + "context" + + "github.com/prometheus/client_golang/prometheus" + "github.com/twmb/franz-go/pkg/kmsg" + "go.uber.org/zap" +) + +func (e *Exporter) collectACLInfo(ctx context.Context, ch chan<- prometheus.Metric) bool { + ACLRes, err := e.minionSvc.ListAllACLs(ctx) + if err != nil { + e.logger.Error("failed to fetch ACLs", zap.Error(err)) + return false + } + + ACLsByType := getResourceTypeName(ACLRes) + totalACLs := 0 + for _, count := range ACLsByType { + totalACLs += count + } + + ch <- prometheus.MustNewConstMetric( + e.aclCount, + prometheus.GaugeValue, + float64(totalACLs), + ) + + for resourceType, count := range ACLsByType { + ch <- prometheus.MustNewConstMetric( + e.aclCountByType, + prometheus.GaugeValue, + float64(count), + resourceType, + ) + } + + return true +} + +func getResourceTypeName(ACLResponse *kmsg.DescribeACLsResponse) map[string]int { + ACLsByType := make(map[string]int) + for _, resource := range ACLResponse.Resources { + resourceType := "unknown" + switch resource.ResourceType { + case 0: + resourceType = "unknown" + case 1: + resourceType = "any" + case 2: + resourceType = "topic" + case 3: + resourceType = "group" + case 4: + resourceType = "cluster" + case 5: + resourceType = "transactional_id" + case 6: + resourceType = "delegation_token" + case 7: + resourceType = "user" + } + + ACLsByType[resourceType] += len(resource.ACLs) + } + + return ACLsByType +} diff --git a/prometheus/exporter.go b/prometheus/exporter.go index d717bcfa..5d6f15cc 100644 --- a/prometheus/exporter.go +++ b/prometheus/exporter.go @@ -48,6 +48,10 @@ type Exporter struct { consumerGroupTopicPartitionLag *prometheus.Desc consumerGroupTopicLag *prometheus.Desc offsetCommits *prometheus.Desc + + // ACLs + aclCount *prometheus.Desc + aclCountByType *prometheus.Desc } func NewExporter(cfg Config, logger *zap.Logger, minionSvc *minion.Service) (*Exporter, error) { @@ -158,7 +162,7 @@ func (e *Exporter) InitializeMetrics() { []string{"group_id"}, nil, ) - // Group Empty Memmbers + // Group Empty Members e.consumerGroupMembersEmpty = prometheus.NewDesc( prometheus.BuildFQName(e.cfg.Namespace, "kafka", "consumer_group_empty_members"), "It will report the number of members in the consumer group with no partition assigned", @@ -207,7 +211,19 @@ func (e *Exporter) InitializeMetrics() { []string{"group_id"}, nil, ) - + // ACLs + e.aclCount = prometheus.NewDesc( + prometheus.BuildFQName(e.cfg.Namespace, "kafka", "acls_total"), + "The total number of ACLs in the cluster", + []string{}, + nil, + ) + e.aclCountByType = prometheus.NewDesc( + prometheus.BuildFQName(e.cfg.Namespace, "kafka", "acls_by_type"), + "The number of ACLs by resource type", + []string{"resource_type"}, + nil, + ) } // Describe implements the prometheus.Collector interface. It sends the @@ -224,7 +240,7 @@ func (e *Exporter) Collect(ch chan<- prometheus.Metric) { ctx, cancel := context.WithTimeout(context.Background(), time.Second*60) defer cancel() - // Attach a unique id which will be used for caching (and and it's invalidation) of the kafka requests + // Attach a unique id which will be used for caching (and it's invalidation) of the kafka requests uuid := uuid2.New() ctx = context.WithValue(ctx, "requestId", uuid.String()) @@ -236,6 +252,7 @@ func (e *Exporter) Collect(ch chan<- prometheus.Metric) { ok = e.collectTopicPartitionOffsets(ctx, ch) && ok ok = e.collectConsumerGroupLags(ctx, ch) && ok ok = e.collectTopicInfo(ctx, ch) && ok + ok = e.collectACLInfo(ctx, ch) && ok if ok { ch <- prometheus.MustNewConstMetric(e.exporterUp, prometheus.GaugeValue, 1.0) From d5911bfc3c79241ef402db8928fcf9775ff55708 Mon Sep 17 00:00:00 2001 From: dttung2905 Date: Mon, 14 Apr 2025 15:26:05 +0100 Subject: [PATCH 2/2] Make acls configurable and address code review Signed-off-by: dttung2905 --- docs/reference-config.yaml | 14 ++++++++++---- minion/acl_config.go | 13 +++++++++++++ minion/config.go | 7 +++++++ prometheus/collect_acl_info.go | 25 +++++-------------------- 4 files changed, 35 insertions(+), 24 deletions(-) create mode 100644 minion/acl_config.go diff --git a/docs/reference-config.yaml b/docs/reference-config.yaml index f8570b28..dbd429eb 100644 --- a/docs/reference-config.yaml +++ b/docs/reference-config.yaml @@ -7,7 +7,7 @@ # file, specify the path to the config file by setting the env variable # CONFIG_FILEPATH. # -# The env variable name is auto generated by upper casing everything and adding +# The env variable name is auto generated by upper-casing everything and adding # an underscore for each indentation/level. Some examples: # kafka.rackId => KAFKA_RACKID # kafka.tls.caFilepath => KAFKA_TLS_CAFILEPATH @@ -40,7 +40,7 @@ kafka: insecureSkipTlsVerify: false sasl: - # Whether or not SASL authentication will be used for authentication + # Whether SASL authentication will be used for authentication enabled: false # Username to use for PLAIN or SCRAM mechanism username: "" @@ -111,6 +111,12 @@ minion: # to version 1.0.0 as describing log dirs was not supported back then. enabled: true + # ACL Metrics + acls: + # Enabled specifies whether ACL information shall be scraped and exported as metrics. + # If disabled, no ACL metrics will be collected. + enabled: false + # EndToEnd Metrics # When enabled, kminion creates a topic which it produces to and consumes from, to measure various advanced metrics. See docs for more info endToEnd: @@ -135,7 +141,7 @@ minion: replicationFactor: 1 # Rarely makes sense to change this, but maybe if you want some sort of cheap load test? - # By default (1) every broker gets one partition + # By default, (1) every broker gets one partition partitionsPerBroker: 1 producer: @@ -153,7 +159,7 @@ minion: groupIdPrefix: kminion-end-to-end # Whether KMinion should try to delete empty consumer groups with the same prefix. This can be used if you want - # KMinion to cleanup it's old consumer groups. It should only be used if you use a unique prefix for KMinion. + # KMinion to clean up it's old consumer groups. It should only be used if you use a unique prefix for KMinion. deleteStaleConsumerGroups: false # This defines: diff --git a/minion/acl_config.go b/minion/acl_config.go new file mode 100644 index 00000000..4d24a7f2 --- /dev/null +++ b/minion/acl_config.go @@ -0,0 +1,13 @@ +package minion + +type ACLsConfig struct { + Enabled bool `koanf:"enabled"` +} + +func (c *ACLsConfig) Validate() error { + return nil +} + +func (c *ACLsConfig) SetDefaults() { + c.Enabled = false +} diff --git a/minion/config.go b/minion/config.go index 3b26a760..a92696a7 100644 --- a/minion/config.go +++ b/minion/config.go @@ -11,6 +11,7 @@ type Config struct { Topics TopicConfig `koanf:"topics"` LogDirs LogDirsConfig `koanf:"logDirs"` EndToEnd e2e.Config `koanf:"endToEnd"` + ACLs ACLsConfig `koanf:"acls"` } func (c *Config) SetDefaults() { @@ -18,6 +19,7 @@ func (c *Config) SetDefaults() { c.Topics.SetDefaults() c.LogDirs.SetDefaults() c.EndToEnd.SetDefaults() + c.ACLs.SetDefaults() } func (c *Config) Validate() error { @@ -41,5 +43,10 @@ func (c *Config) Validate() error { return fmt.Errorf("failed to validate endToEnd config: %w", err) } + err = c.ACLs.Validate() + if err != nil { + return fmt.Errorf("failed to validate ACLs config: %w", err) + } + return nil } diff --git a/prometheus/collect_acl_info.go b/prometheus/collect_acl_info.go index 24f39226..b667dc62 100644 --- a/prometheus/collect_acl_info.go +++ b/prometheus/collect_acl_info.go @@ -9,6 +9,10 @@ import ( ) func (e *Exporter) collectACLInfo(ctx context.Context, ch chan<- prometheus.Metric) bool { + if !e.minionSvc.Cfg.ACLs.Enabled { + return true + } + ACLRes, err := e.minionSvc.ListAllACLs(ctx) if err != nil { e.logger.Error("failed to fetch ACLs", zap.Error(err)) @@ -42,26 +46,7 @@ func (e *Exporter) collectACLInfo(ctx context.Context, ch chan<- prometheus.Metr func getResourceTypeName(ACLResponse *kmsg.DescribeACLsResponse) map[string]int { ACLsByType := make(map[string]int) for _, resource := range ACLResponse.Resources { - resourceType := "unknown" - switch resource.ResourceType { - case 0: - resourceType = "unknown" - case 1: - resourceType = "any" - case 2: - resourceType = "topic" - case 3: - resourceType = "group" - case 4: - resourceType = "cluster" - case 5: - resourceType = "transactional_id" - case 6: - resourceType = "delegation_token" - case 7: - resourceType = "user" - } - + resourceType := resource.ResourceType.String() ACLsByType[resourceType] += len(resource.ACLs) }