feat(LOG-9387): Introduce Ingester metrics to better understand LOG-9387

Author: JoaoBraveCoding

pkg/ingester/instance.go

@@ -229,13 +229,18 @@ func (i *instance) consumeChunk(ctx context.Context, ls labels.Labels, chunk *lo
 // happened to *the last stream in the request*. Ex: if three streams are part of the PushRequest
 // and all three failed, the returned error only describes what happened to the last processed stream.
 func (i *instance) Push(ctx context.Context, req *logproto.PushRequest) error {
+	i.metrics.pushInflight.Inc()
+	defer i.metrics.pushInflight.Dec()
+
 	record := recordPool.GetRecord()
 	record.UserID = i.instanceID
 	defer recordPool.PutRecord(record)
 	rateLimitWholeStream := i.limiter.limits.ShardStreams(i.instanceID).Enabled
 
 	var appendErr error
+	var chunkLockTotal time.Duration
 	for _, reqStream := range req.Streams {
+		lockStart := time.Now()
 
 		s, _, err := i.streams.LoadOrStoreNew(reqStream.Labels,
 			func() (*stream, error) {
@@ -259,9 +264,12 @@ func (i *instance) Push(ctx context.Context, req *logproto.PushRequest) error {
 
 		_, appendErr = s.Push(ctx, reqStream.Entries, record, 0, false, rateLimitWholeStream, i.customStreamsTracker, req.Format)
 		s.chunkMtx.Unlock()
+		chunkLockTotal += time.Since(lockStart)
 	}
+	i.metrics.pushChunkMemoryDuration.Observe(chunkLockTotal.Seconds())
 
 	if !record.IsEmpty() {
+		walStart := time.Now()
 		if err := i.wal.Log(record); err != nil {
 			if e, ok := err.(*os.PathError); ok && e.Err == syscall.ENOSPC {
 				i.metrics.walDiskFullFailures.Inc()
@@ -272,9 +280,15 @@ func (i *instance) Push(ctx context.Context, req *logproto.PushRequest) error {
 					)
 				})
 			} else {
+				i.metrics.pushWALDuration.Observe(time.Since(walStart).Seconds())
 				return err
 			}
 		}
+		i.metrics.pushWALDuration.Observe(time.Since(walStart).Seconds())
+	}
+
+	if ctx.Err() != nil {
+		i.metrics.pushContextAlreadyDone.Inc()
 	}
 
 	return appendErr

pkg/ingester/metrics.go

@@ -71,6 +71,12 @@ type ingesterMetrics struct {
 	flushQueueLength       prometheus.Gauge
 	duplicateLogBytesTotal *prometheus.CounterVec
 	streamsOwnershipCheck  prometheus.Histogram
+
+	// Push path instrumentation (LOG-9387 investigation)
+	pushChunkMemoryDuration prometheus.Histogram
+	pushWALDuration        prometheus.Histogram
+	pushInflight           prometheus.Gauge
+	pushContextAlreadyDone prometheus.Counter
 }
 
 // setRecoveryBytesInUse bounds the bytes reports to >= 0.
@@ -334,5 +340,24 @@ func newIngesterMetrics(r prometheus.Registerer, metricsNamespace string) *inges
 			Name:      "duplicate_log_bytes_total",
 			Help:      "The total number of bytes that were discarded for duplicate log lines.",
 		}, []string{"tenant"}),
+
+		pushChunkMemoryDuration: promauto.With(r).NewHistogram(prometheus.HistogramOpts{
+			Name:    "loki_ingester_push_chunk_memory_duration_seconds",
+			Help:    "Time spent reading entries into chunk memory per push request.",
+			Buckets: prometheus.ExponentialBuckets(0.001, 4, 9),
+		}),
+		pushWALDuration: promauto.With(r).NewHistogram(prometheus.HistogramOpts{
+			Name:    "loki_ingester_push_wal_duration_seconds",
+			Help:    "Time spent writing to the WAL per push request.",
+			Buckets: prometheus.ExponentialBuckets(0.001, 4, 9),
+		}),
+		pushInflight: promauto.With(r).NewGauge(prometheus.GaugeOpts{
+			Name: "loki_ingester_push_inflight",
+			Help: "Number of push requests currently being processed.",
+		}),
+		pushContextAlreadyDone: promauto.With(r).NewCounter(prometheus.CounterOpts{
+			Name: "loki_ingester_push_context_already_done_total",
+			Help: "Total push requests that completed after their context was already canceled.",
+		}),
 	}
 }