chore: Define an E2E ingestion processing time metric (#20054)

Author: benclive

pkg/dataobj/consumer/partition_processor.go

@@ -74,6 +74,9 @@ type partitionProcessor struct {
 	// The initial value is zero and must be reset to zero after each flush.
 	lastModified time.Time
 
+	// earliestRecordTime tracks the earliest timestamp all the records Appended to the builder for each object.
+	earliestRecordTime time.Time
+
 	// Metrics
 	metrics *partitionOffsetMetrics
 
@@ -216,8 +219,9 @@ func (p *partitionProcessor) initBuilder() error {
 
 func (p *partitionProcessor) emitObjectWrittenEvent(ctx context.Context, objectPath string) error {
 	event := &metastore.ObjectWrittenEvent{
-		ObjectPath: objectPath,
-		WriteTime:  p.clock.Now().Format(time.RFC3339),
+		ObjectPath:         objectPath,
+		WriteTime:          p.clock.Now().Format(time.RFC3339),
+		EarliestRecordTime: p.earliestRecordTime.Format(time.RFC3339),
 	}
 
 	eventBytes, err := event.Marshal()
@@ -240,6 +244,10 @@ func (p *partitionProcessor) processRecord(ctx context.Context, record partition
 	// Update offset metric at the end of processing
 	defer p.metrics.updateOffset(record.Offset)
 
+	if record.Timestamp.Before(p.earliestRecordTime) || p.earliestRecordTime.IsZero() {
+		p.earliestRecordTime = record.Timestamp
+	}
+
 	// Observe processing delay
 	p.metrics.observeProcessingDelay(record.Timestamp)
 
@@ -325,6 +333,7 @@ func (p *partitionProcessor) flush(ctx context.Context) error {
 
 	p.lastModified = time.Time{}
 	p.lastFlushed = p.clock.Now()
+	p.earliestRecordTime = time.Time{}
 
 	return nil
 }

pkg/dataobj/index/indexer.go

@@ -325,7 +325,7 @@ func (si *serialIndexer) processBuildRequest(req buildRequest) buildResult {
 
 	// Update metrics
 	buildTime := time.Since(start)
-	si.updateMetrics(buildTime)
+	si.updateMetrics(buildTime, getEarliestIndexedRecord(si.logger, events[:processed]))
 
 	if err != nil {
 		level.Error(si.logger).Log("msg", "failed to build index",
@@ -350,6 +350,21 @@ func (si *serialIndexer) processBuildRequest(req buildRequest) buildResult {
 	}
 }
 
+func getEarliestIndexedRecord(logger log.Logger, events []metastore.ObjectWrittenEvent) time.Time {
+	var earliestIndexedRecordTime time.Time
+	for _, ev := range events {
+		ts, err := time.Parse(time.RFC3339, ev.EarliestRecordTime)
+		if err != nil {
+			level.Error(logger).Log("msg", "failed to parse earliest record time", "err", err)
+			continue
+		}
+		if ts.Before(earliestIndexedRecordTime) || earliestIndexedRecordTime.IsZero() {
+			earliestIndexedRecordTime = ts
+		}
+	}
+	return earliestIndexedRecordTime
+}
+
 // buildIndex is writing all metastore events to a single index object. It
 // returns the index path and the number of events processed or an error if the index object is not created.
 // The number of events processed can be less than the number of events if the builder becomes full
@@ -501,10 +516,13 @@ func (si *serialIndexer) flushIndex(ctx context.Context, partition int32) (strin
 }
 
 // updateMetrics updates internal build metrics
-func (si *serialIndexer) updateMetrics(buildTime time.Duration) {
+func (si *serialIndexer) updateMetrics(buildTime time.Duration, earliestIndexedRecord time.Time) {
 	si.indexerMetrics.incBuilds()
 	si.indexerMetrics.setBuildTime(buildTime)
 	si.indexerMetrics.setQueueDepth(len(si.buildRequestChan))
+	if !earliestIndexedRecord.IsZero() {
+		si.indexerMetrics.setEndToEndProcessingTime(time.Since(earliestIndexedRecord))
+	}
 }
 
 // ObjectKey generates the object key for storing an index object in object storage.

pkg/dataobj/index/metrics.go

@@ -90,6 +90,9 @@ type indexerMetrics struct {
 
 	// Queue metrics
 	queueDepth prometheus.Gauge
+
+	// End-to-end processing time metric
+	endToEndProcessingTime prometheus.Gauge
 }
 
 func newIndexerMetrics() *indexerMetrics {
@@ -110,6 +113,10 @@ func newIndexerMetrics() *indexerMetrics {
 			Name: "loki_index_builder_queue_depth",
 			Help: "Current depth of the build request queue",
 		}),
+		endToEndProcessingTime: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "loki_ingest_end_to_end_processing_time_seconds",
+			Help: "Time between a log line being written to kafka by the distributors and the index-builder making it available for querying in seconds",
+		}),
 	}
 
 	return m
@@ -121,6 +128,7 @@ func (m *indexerMetrics) register(reg prometheus.Registerer) error {
 		m.totalBuilds,
 		m.buildTimeSeconds,
 		m.queueDepth,
+		m.endToEndProcessingTime,
 	}
 
 	for _, collector := range collectors {
@@ -139,6 +147,7 @@ func (m *indexerMetrics) unregister(reg prometheus.Registerer) {
 		m.totalBuilds,
 		m.buildTimeSeconds,
 		m.queueDepth,
+		m.endToEndProcessingTime,
 	}
 
 	for _, collector := range collectors {
@@ -161,3 +170,7 @@ func (m *indexerMetrics) setBuildTime(duration time.Duration) {
 func (m *indexerMetrics) setQueueDepth(depth int) {
 	m.queueDepth.Set(float64(depth))
 }
+
+func (m *indexerMetrics) setEndToEndProcessingTime(duration time.Duration) {
+	m.endToEndProcessingTime.Set(duration.Seconds())
+}

pkg/dataobj/metastore/metastore.pb.go

@@ -5,9 +5,10 @@ package dataobj.metastore;
 
 option go_package = "github.com/grafana/loki/v3/pkg/dataobj/metastore";
 
-// ColumnInfo describes an individual column within a data set.
+// ObjectWrittenEvent describes a new logs object being written to object storage.
 message ObjectWrittenEvent {
   string objectPath = 2;
   string writeTime = 3;
+  string earliestRecordTime = 4;
   reserved 1; // tenant
 }