chore: add debug metrics

Author: grobinson-grafana

docs/sources/shared/configuration.md

@@ -3188,6 +3188,14 @@ dataobj_tee:
   # Maximum number of bytes to buffer.
   # CLI flag: -distributor.dataobj-tee.max-buffered-bytes
   [max_buffered_bytes: <int> | default = 104857600]
+
+  # The per-tenant partition rate (bytes/sec).
+  # CLI flag: -distributor.dataobj-tee.per-partition-rate-bytes
+  [per_partition_rate_bytes: <int> | default = 1048576]
+
+  # Enables optional debug metrics.
+  # CLI flag: -distributor.dataobj-tee.debug-metrics-enabled
+  [debug_metrics_enabled: <boolean> | default = false]
 ```
 
 ### etcd

pkg/distributor/dataobj_tee.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"flag"
+	"strconv"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -19,13 +20,15 @@ type DataObjTeeConfig struct {
 	Topic                 string `yaml:"topic"`
 	MaxBufferedBytes      int    `yaml:"max_buffered_bytes"`
 	PerPartitionRateBytes int    `yaml:"per_partition_rate_bytes"`
+	DebugMetricsEnabled   bool   `yaml:"debug_metrics_enabled"`
 }
 
 func (c *DataObjTeeConfig) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&c.Enabled, "distributor.dataobj-tee.enabled", false, "Enable data object tee.")
 	f.StringVar(&c.Topic, "distributor.dataobj-tee.topic", "", "Topic for data object tee.")
 	f.IntVar(&c.MaxBufferedBytes, "distributor.dataobj-tee.max-buffered-bytes", 100<<20, "Maximum number of bytes to buffer.")
 	f.IntVar(&c.PerPartitionRateBytes, "distributor.dataobj-tee.per-partition-rate-bytes", 1024*1024, "The per-tenant partition rate (bytes/sec).")
+	f.BoolVar(&c.DebugMetricsEnabled, "distributor.dataobj-tee.debug-metrics-enabled", false, "Enables optional debug metrics.")
 }
 
 func (c *DataObjTeeConfig) Validate() error {
@@ -56,6 +59,10 @@ type DataObjTee struct {
 	// Metrics.
 	failures prometheus.Counter
 	total    prometheus.Counter
+
+	// High cardinality metrics which are only emitted when debug metrics
+	// are enabled.
+	produced *prometheus.CounterVec
 }
 
 // NewDataObjTee returns a new DataObjTee.
@@ -81,6 +88,10 @@ func NewDataObjTee(
 			Name: "loki_distributor_dataobj_tee_duplicate_streams_total",
 			Help: "Total number of streams duplicated.",
 		}),
+		produced: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "loki_distributor_dataobj_tee_produced_bytes_total",
+			Help: "Total number of bytes produced to each partition.",
+		}, []string{"tenant", "partition", "segmentation_key"}),
 	}, nil
 }
 
@@ -139,4 +150,11 @@ func (t *DataObjTee) duplicate(ctx context.Context, tenant string, stream Segmen
 		level.Error(t.logger).Log("msg", "failed to produce records", "err", err)
 		t.failures.Inc()
 	}
+	if t.cfg.DebugMetricsEnabled {
+		t.produced.WithLabelValues(
+			tenant,
+			strconv.FormatInt(int64(partition), 10),
+			string(stream.SegmentationKey),
+		).Add(float64(stream.Stream.Size()))
+	}
 }

pkg/distributor/ingest_limits_test.go

@@ -312,36 +312,30 @@ func TestIngestLimits_UpdateRates(t *testing.T) {
 		name:   "error should be returned if rates cannot be updated",
 		tenant: "test",
 		streams: []SegmentedStream{{
-			HashKeyNoShard: 1,
+			SegmentationKey: "test",
 		}},
-		expectedRequest: &proto.UpdateRatesRequest{
-			Tenant: "test",
-			Streams: []*proto.StreamMetadata{{
-				StreamHash: 1,
-			}},
-		},
 		responseErr: errors.New("failed to update rates"),
 		expectedErr: "failed to update rates",
 	}, {
 		name:   "updates rates",
 		tenant: "test",
 		streams: []SegmentedStream{{
-			HashKeyNoShard: 1,
+			SegmentationKey: "test",
 		}},
 		expectedRequest: &proto.UpdateRatesRequest{
 			Tenant: "test",
 			Streams: []*proto.StreamMetadata{{
-				StreamHash: 1,
+				StreamHash: 0xb5fb79e24c92922f,
 			}},
 		},
 		response: &proto.UpdateRatesResponse{
 			Results: []*proto.UpdateRatesResult{{
-				StreamHash: 1,
+				StreamHash: 0xb5fb79e24c92922f,
 				Rate:       1024,
 			}},
 		},
 		expectedResult: []*proto.UpdateRatesResult{{
-			StreamHash: 1,
+			StreamHash: 0xb5fb79e24c92922f,
 			Rate:       1024,
 		}},
 	}}

pkg/distributor/segment.go

@@ -18,20 +18,12 @@ import (
 // distribute load while preserving stream locality for tenants.
 type SegmentationKey string
 
-<<<<<<< HEAD
 // Sum64 returns a 64 bit, non-cryptographic hash of the key.
-=======
-// Sum64 returns a 64 bit, non-crytogrpahic hash of the key.
->>>>>>> 5ef092522a (feat: add segmentation keys and resolver)
 func (key SegmentationKey) Sum64() uint64 {
 	h := fnv.New64a()
 	// Use a reserved word here to avoid any possible hash conflicts with
-	// streams.
-<<<<<<< HEAD
+	// streams
 	h.Write([]byte("__loki_segmentation_key__"))
-=======
-	h.Write([]byte("__loki__segmentation_key__"))
->>>>>>> 5ef092522a (feat: add segmentation keys and resolver)
 	h.Write([]byte(key))
 	return h.Sum64()
 }
@@ -52,19 +44,12 @@ func GetSegmentationKey(stream KeyedStream) (SegmentationKey, error) {
 type SegmentationPartitionResolver struct {
 	perPartitionRateBytes uint64
 	ringReader            ring.PartitionRingReader
-<<<<<<< HEAD
 	logger                log.Logger
 
 	// Metrics.
 	failed          prometheus.Counter
 	randomlySharded prometheus.Counter
 	total           prometheus.Counter
-=======
-	fallback              prometheus.Counter
-	failed                prometheus.Counter
-	total                 prometheus.Counter
-	logger                log.Logger
->>>>>>> 5ef092522a (feat: add segmentation keys and resolver)
 }
 
 // NewSegmentationPartitionResolver returns a new SegmentationPartitionResolver.