temp fix

Author: amuraru

e2e/bin/integration-test.sh

@@ -100,9 +100,14 @@ validate_builtin_metrics() {
     local core_metrics=(
         "kminion_exporter_up"
         "kminion_exporter_offset_consumer_records_consumed_total"
+        "kminion_kafka_received_bytes"
+        "kminion_kafka_requests_received_total"
+        "kminion_kafka_requests_sent_total"
+        "kminion_kafka_sent_bytes"
+        "kminion_log_messages_total"
     )
 
-    # Kafka cluster metrics
+    # Kafka cluster/broker metrics
     local kafka_metrics=(
         "kminion_kafka_cluster_info"
         "kminion_kafka_broker_info"
@@ -111,17 +116,29 @@ validate_builtin_metrics() {
     # Topic metrics
     local topic_metrics=(
         "kminion_kafka_topic_info"
-        "kminion_kafka_topic_partition_high_water_mark"
-        "kminion_kafka_topic_high_water_mark_sum"
-        "kminion_kafka_topic_partition_low_water_mark"
+        "kminion_kafka_topic_info_min_insync_replicas"
+        "kminion_kafka_topic_info_partitions_count"
+        "kminion_kafka_topic_info_replication_factor"
+        "kminion_kafka_topic_info_retention_ms"
         "kminion_kafka_topic_low_water_mark_sum"
+        "kminion_kafka_topic_partition_low_water_mark"
+        "kminion_kafka_topic_high_water_mark_sum"
+        "kminion_kafka_topic_partition_high_water_mark"
+        "kminion_kafka_topic_max_timestamp"
+        "kminion_kafka_topic_partition_max_timestamp"
     )
 
     # Consumer group metrics
     local consumer_group_metrics=(
         "kminion_kafka_consumer_group_info"
+        "kminion_kafka_consumer_group_info_count"
+        "kminion_kafka_consumer_group_info_empty_groups"
         "kminion_kafka_consumer_group_members"
+        "kminion_kafka_consumer_group_topic_assigned_partitions"
         "kminion_kafka_consumer_group_topic_lag"
+        "kminion_kafka_consumer_group_topic_members"
+        "kminion_kafka_consumer_group_topic_offset_sum"
+        "kminion_kafka_consumer_group_topic_partition_lag"
     )
 
     # Log dir metrics

minion/list_offsets.go

@@ -64,49 +64,19 @@ func (s *Service) listOffsetsCached(ctx context.Context, offsetType string) (kad
 
 // ListEndOffsets fetches the high water mark for all topic partitions.
 func (s *Service) ListEndOffsets(ctx context.Context) (kadm.ListedOffsets, error) {
-	return s.listOffsetsInternal(ctx, func(ctx context.Context, topics ...string) (kadm.ListedOffsets, error) {
-		if len(topics) == 0 {
-			// Get all topics if none specified
-			allTopics, err := s.getAllowedTopics(ctx)
-			if err != nil {
-				return nil, fmt.Errorf("failed to get topics for listing end offsets: %w", err)
-			}
-			topics = allTopics
-		}
-		return s.admClient.ListEndOffsets(ctx, topics...)
-	}, "end")
+	return s.listOffsetsInternal(ctx, s.admClient.ListEndOffsets, "end")
 }
 
 // ListStartOffsets fetches the low water mark for all topic partitions.
 func (s *Service) ListStartOffsets(ctx context.Context) (kadm.ListedOffsets, error) {
-	return s.listOffsetsInternal(ctx, func(ctx context.Context, topics ...string) (kadm.ListedOffsets, error) {
-		if len(topics) == 0 {
-			// Get all topics if none specified
-			allTopics, err := s.getAllowedTopics(ctx)
-			if err != nil {
-				return nil, fmt.Errorf("failed to get topics for listing start offsets: %w", err)
-			}
-			topics = allTopics
-		}
-		return s.admClient.ListStartOffsets(ctx, topics...)
-	}, "start")
+	return s.listOffsetsInternal(ctx, s.admClient.ListStartOffsets, "start")
 }
 
 // ListMaxTimestampOffsets fetches the offsets for the maximum timestamp for all topic partitions.
 // This requires Kafka 3.0+ (see https://issues.apache.org/jira/browse/KAFKA-12541)
 // Uses the dedicated ListMaxTimestampOffsets method from franz-go.
 func (s *Service) ListMaxTimestampOffsets(ctx context.Context) (kadm.ListedOffsets, error) {
-	return s.listOffsetsInternal(ctx, func(ctx context.Context, topics ...string) (kadm.ListedOffsets, error) {
-		if len(topics) == 0 {
-			// Get all topics if none specified
-			allTopics, err := s.getAllowedTopics(ctx)
-			if err != nil {
-				return nil, fmt.Errorf("failed to get topics for listing max timestamp offsets: %w", err)
-			}
-			topics = allTopics
-		}
-		return s.admClient.ListMaxTimestampOffsets(ctx, topics...)
-	}, "maxTimestamp")
+	return s.listOffsetsInternal(ctx, s.admClient.ListMaxTimestampOffsets, "maxTimestamp")
 }
 
 type listOffsetsFunc func(context.Context, ...string) (kadm.ListedOffsets, error)
@@ -159,20 +129,3 @@ func (s *Service) listOffsetsInternal(ctx context.Context, listFunc listOffsetsF
 
 	return listedOffsets, nil
 }
-
-// getAllowedTopics gets all topics from metadata and filters them according to the configuration
-func (s *Service) getAllowedTopics(ctx context.Context) ([]string, error) {
-	metadata, err := s.GetMetadataCached(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get metadata: %w", err)
-	}
-
-	var allowedTopics []string
-	for _, topic := range metadata.Topics {
-		if topic.Topic != nil && s.IsTopicAllowed(*topic.Topic) {
-			allowedTopics = append(allowedTopics, *topic.Topic)
-		}
-	}
-
-	return allowedTopics, nil
-}