Add reporting of lag as a metric

daemon.go

@@ -1,13 +1,15 @@
 package indexer
 
 import (
+	"context"
 	"database/sql"
 	"log/slog"
 	"sync"
 	"time"
 
 	"github.com/nyaruka/gocommon/analytics"
 	"github.com/nyaruka/rp-indexer/v9/indexers"
+	"github.com/pkg/errors"
 )
 
 type Daemon struct {
@@ -78,6 +80,9 @@ func (d *Daemon) startIndexer(indexer indexers.Indexer) {
 func (d *Daemon) startStatsReporter(interval time.Duration) {
 	d.wg.Add(1) // add ourselves to the wait group
 
+	// calculating lag is more expensive than reading indexer stats so we only do it every 5th iteration
+	var iterations int64
+
 	go func() {
 		defer func() {
 			slog.Info("analytics exiting")
@@ -89,13 +94,19 @@ func (d *Daemon) startStatsReporter(interval time.Duration) {
 			case <-d.quit:
 				return
 			case <-time.After(interval):
-				d.reportStats()
+				d.reportStats(iterations%5 == 0)
 			}
+
+			iterations++
 		}
 	}()
 }
 
-func (d *Daemon) reportStats() {
+func (d *Daemon) reportStats(includeLag bool) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	log := slog.New(slog.Default().Handler())
 	metrics := make(map[string]float64, len(d.indexers)*2)
 
 	for _, ix := range d.indexers {
@@ -115,9 +126,16 @@ func (d *Daemon) reportStats() {
 		metrics[ix.Name()+"_rate"] = rateInPeriod
 
 		d.prevStats[ix] = stats
-	}
 
-	log := slog.New(slog.Default().Handler())
+		if includeLag {
+			lag, err := d.calculateLag(ctx, ix)
+			if err != nil {
+				log.Error("error getting db last modified", "index", ix.Name(), "error", err)
+			} else {
+				metrics[ix.Name()+"_lag"] = lag.Seconds()
+			}
+		}
+	}
 
 	for k, v := range metrics {
 		analytics.Gauge("indexer."+k, v)
@@ -127,6 +145,20 @@ func (d *Daemon) reportStats() {
 	log.Info("stats reported")
 }
 
+func (d *Daemon) calculateLag(ctx context.Context, ix indexers.Indexer) (time.Duration, error) {
+	esLastModified, err := ix.GetLastModified(ix.Name())
+	if err != nil {
+		return 0, errors.Wrap(err, "error getting ES last modified")
+	}
+
+	dbLastModified, err := ix.GetDBLastModified(ctx, d.db)
+	if err != nil {
+		return 0, errors.Wrap(err, "error getting DB last modified")
+	}
+
+	return dbLastModified.Sub(esLastModified), nil
+}
+
 // Stop stops this daemon
 func (d *Daemon) Stop() {
 	slog.Info("daemon stopping")

indexers/base.go

@@ -1,6 +1,7 @@
 package indexers
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"fmt"
@@ -31,6 +32,9 @@ type Indexer interface {
 	Name() string
 	Index(db *sql.DB, rebuild, cleanup bool) (string, error)
 	Stats() Stats
+
+	GetLastModified(index string) (time.Time, error)
+	GetDBLastModified(ctx context.Context, db *sql.DB) (time.Time, error)
 }
 
 // IndexDefinition is what we pass to elastic to create an index,
@@ -318,8 +322,13 @@ func (i *baseIndexer) GetLastModified(index string) (time.Time, error) {
 	lastModified := time.Time{}
 
 	// get the newest document on our index
-	queryResponse := queryResponse{}
-	_, err := utils.MakeJSONRequest(http.MethodPost, fmt.Sprintf("%s/%s/_search", i.elasticURL, index), []byte(`{ "sort": [{ "modified_on_mu": "desc" }]}`), &queryResponse)
+	queryResponse := &queryResponse{}
+	_, err := utils.MakeJSONRequest(
+		http.MethodPost,
+		fmt.Sprintf("%s/%s/_search", i.elasticURL, index),
+		[]byte(`{ "sort": [{ "modified_on_mu": "desc" }], "_source": {"includes": ["modified_on", "id"]}, "size": 1}`),
+		queryResponse,
+	)
 	if err != nil {
 		return lastModified, err
 	}

indexers/contacts.go

@@ -274,3 +274,27 @@ func (i *ContactIndexer) indexModified(ctx context.Context, db *sql.DB, index st
 
 	return totalCreated, totalDeleted, nil
 }
+
+func (i *ContactIndexer) GetDBLastModified(ctx context.Context, db *sql.DB) (time.Time, error) {
+	lastModified := time.Time{}
+
+	if err := db.QueryRowContext(ctx, "SELECT MAX(modified_on) FROM contacts_contact").Scan(&lastModified); err != nil {
+		return lastModified, err
+	}
+
+	return lastModified, nil
+}
+
+func (i *ContactIndexer) Lag(ctx context.Context, db *sql.DB) (time.Duration, error) {
+	esLastModified, err := i.GetLastModified(i.name)
+	if err != nil {
+		return 0, errors.Wrap(err, "error getting ES last modified")
+	}
+
+	var dbLastModified time.Time
+	if err := db.QueryRowContext(ctx, "SELECT MAX(modified_on) FROM contacts_contact").Scan(&dbLastModified); err != nil {
+		return 0, errors.Wrap(err, "error getting ES last modified")
+	}
+
+	return dbLastModified.Sub(esLastModified), nil
+}