Update v10.0.0

.github/workflows/ci.yml

@@ -1,7 +1,7 @@
 name: CI
 on: [push, pull_request]
 env:
-  go-version: "1.22.x"
+  go-version: "1.23.x"
 jobs:
   test:
     name: Test

CHANGELOG.md

@@ -1,3 +1,31 @@
+v10.0.0 (2025-01-07)
+-------------------------
+ * Update README.md
+
+v9.3.4 (2024-12-16)
+-------------------------
+ * Update to latest gocommon and tweak default cloudwatch namespace
+
+v9.3.3 (2024-12-13)
+-------------------------
+ * Clean up old analytics use
+
+v9.3.2 (2024-12-13)
+-------------------------
+ * Update to latest gocommon
+
+v9.3.1 (2024-12-13)
+-------------------------
+ * Remove librato, use latest gocommon
+
+v9.3.0 (2024-12-13)
+-------------------------
+ * Send metrics to cloudwatch
+
+v9.2.1 (2024-10-08)
+-------------------------
+ * Don't include status groups in contact indexing
+
 v9.2.0 (2024-07-17)
 -------------------------
  * Test against postgresql 15

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.22
+FROM golang:1.23
 
 WORKDIR /usr/src/app
 

README.md

@@ -37,16 +37,19 @@ We recommend running it with no changes to the configuration and no parameters,
 environment variables to configure it. You can use `% rp-indexer --help` to see a list of the
 environment variables and parameters and for more details on each option.
 
-### RapidPro
-
-For use with RapidPro, you will want to configure these settings:
-
  * `INDEXER_DB`: a URL connection string for your RapidPro database or read replica
  * `INDEXER_ELASTIC_URL`: the URL for your ElasticSearch endpoint
+
+### AWS services:
+
+ * `INDEXER_AWS_ACCESS_KEY_ID`: AWS access key id used to authenticate to AWS
+ * `INDEXER_AWS_SECRET_ACCESS_KEY` AWS secret access key used to authenticate to AWS
+ * `INDEXER_AWS_REGION`: AWS region (ex: `eu-west-1`)
 
-Recommended settings for error reporting:
+### Logging and error reporting:
 
  * `INDEXER_SENTRY_DSN`: DSN to use when logging errors to Sentry
+ * `INDEXER_LOG_LEVEL`: logging level to use (default is `info`)
 
 ## Development
 

WENI-CHANGELOG.md

@@ -1,3 +1,7 @@
+1.2.2-indexer-10.0.0
+----------
+ * Update to version v10.0.0
+
 1.2.2-indexer-9.2.0
 ----------
  * Update to version v9.2.0

cmd/rp-indexer/main.go

@@ -12,8 +12,10 @@ import (
 	"github.com/getsentry/sentry-go"
 	_ "github.com/lib/pq"
 	"github.com/nyaruka/ezconf"
+	"github.com/nyaruka/gocommon/aws/cwatch"
 	indexer "github.com/nyaruka/rp-indexer/v9"
 	"github.com/nyaruka/rp-indexer/v9/indexers"
+	"github.com/nyaruka/rp-indexer/v9/runtime"
 	slogmulti "github.com/samber/slog-multi"
 	slogsentry "github.com/samber/slog-sentry"
 )
@@ -25,7 +27,7 @@ var (
 )
 
 func main() {
-	cfg := indexer.NewDefaultConfig()
+	cfg := runtime.NewDefaultConfig()
 	loader := ezconf.NewLoader(cfg, "indexer", "Indexes RapidPro contacts to ElasticSearch", []string{"indexer.toml"})
 	loader.MustLoad()
 
@@ -36,15 +38,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	rt := &runtime.Runtime{Config: cfg}
+
 	// configure our logger
 	logHandler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: level})
 	slog.SetDefault(slog.New(logHandler))
 
-	logger := slog.With("comp", "main")
-	logger.Info("starting indexer", "version", version, "released", date)
-
 	// if we have a DSN entry, try to initialize it
-	if cfg.SentryDSN != "" {
+	if rt.Config.SentryDSN != "" {
 		err := sentry.Init(sentry.ClientOptions{
 			Dsn:           cfg.SentryDSN,
 			EnableTracing: false,
@@ -55,7 +56,8 @@ func main() {
 		}
 
 		defer sentry.Flush(2 * time.Second)
-		logger = slog.New(
+
+		logger := slog.New(
 			slogmulti.Fanout(
 				logHandler,
 				slogsentry.Option{Level: slog.LevelError}.NewSentryHandler(),
@@ -65,24 +67,32 @@ func main() {
 		slog.SetDefault(logger)
 	}
 
-	db, err := sql.Open("postgres", cfg.DB)
+	log := slog.With("comp", "main")
+	log.Info("starting indexer", "version", version, "released", date)
+
+	rt.DB, err = sql.Open("postgres", cfg.DB)
+	if err != nil {
+		log.Error("unable to connect to database", "error", err)
+	}
+
+	rt.CW, err = cwatch.NewService(rt.Config.AWSAccessKeyID, rt.Config.AWSSecretAccessKey, rt.Config.AWSRegion, rt.Config.CloudwatchNamespace, rt.Config.DeploymentID)
 	if err != nil {
-		logger.Error("unable to connect to database")
+		log.Error("unable to create cloudwatch service", "error", err)
 	}
 
 	idxrs := []indexers.Indexer{
-		indexers.NewContactIndexer(cfg.ElasticURL, cfg.ContactsIndex, cfg.ContactsShards, cfg.ContactsReplicas, 500),
+		indexers.NewContactIndexer(rt.Config.ElasticURL, rt.Config.ContactsIndex, rt.Config.ContactsShards, rt.Config.ContactsReplicas, 500),
 	}
 
-	if cfg.Rebuild {
+	if rt.Config.Rebuild {
 		// if rebuilding, just do a complete index and quit. In future when we support multiple indexers,
 		// the rebuild argument can be become the name of the index to rebuild, e.g. --rebuild=contacts
 		idxr := idxrs[0]
-		if _, err := idxr.Index(db, true, cfg.Cleanup); err != nil {
-			logger.Error("error during rebuilding", "error", err, "indexer", idxr.Name())
+		if _, err := idxr.Index(rt, true, rt.Config.Cleanup); err != nil {
+			log.Error("error during rebuilding", "error", err, "indexer", idxr.Name())
 		}
 	} else {
-		d := indexer.NewDaemon(cfg, db, idxrs, time.Duration(cfg.Poll)*time.Second)
+		d := indexer.NewDaemon(rt, idxrs)
 		d.Start()
 
 		handleSignals(d)

daemon.go

@@ -2,19 +2,19 @@ package indexer
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 	"log/slog"
 	"sync"
 	"time"
 
-	"github.com/nyaruka/gocommon/analytics"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatch/types"
+	"github.com/nyaruka/gocommon/aws/cwatch"
 	"github.com/nyaruka/rp-indexer/v9/indexers"
+	"github.com/nyaruka/rp-indexer/v9/runtime"
 )
 
 type Daemon struct {
-	cfg      *Config
-	db       *sql.DB
+	rt       *runtime.Runtime
 	wg       *sync.WaitGroup
 	quit     chan bool
 	indexers []indexers.Indexer
@@ -24,27 +24,19 @@ type Daemon struct {
 }
 
 // NewDaemon creates a new daemon to run the given indexers
-func NewDaemon(cfg *Config, db *sql.DB, ixs []indexers.Indexer, poll time.Duration) *Daemon {
+func NewDaemon(rt *runtime.Runtime, ixs []indexers.Indexer) *Daemon {
 	return &Daemon{
-		cfg:       cfg,
-		db:        db,
+		rt:        rt,
 		wg:        &sync.WaitGroup{},
 		quit:      make(chan bool),
 		indexers:  ixs,
-		poll:      poll,
+		poll:      time.Duration(rt.Config.Poll) * time.Second,
 		prevStats: make(map[indexers.Indexer]indexers.Stats, len(ixs)),
 	}
 }
 
 // Start starts this daemon
 func (d *Daemon) Start() {
-	// if we have a librato token, configure it
-	if d.cfg.LibratoToken != "" {
-		analytics.RegisterBackend(analytics.NewLibrato(d.cfg.LibratoUsername, d.cfg.LibratoToken, d.cfg.InstanceName, time.Second, d.wg))
-	}
-
-	analytics.Start()
-
 	for _, i := range d.indexers {
 		d.startIndexer(i)
 	}
@@ -68,7 +60,7 @@ func (d *Daemon) startIndexer(indexer indexers.Indexer) {
 			case <-d.quit:
 				return
 			case <-time.After(d.poll):
-				_, err := indexer.Index(d.db, d.cfg.Rebuild, d.cfg.Cleanup)
+				_, err := indexer.Index(d.rt, d.rt.Config.Rebuild, d.rt.Config.Cleanup)
 				if err != nil {
 					log.Error("error during indexing", "error", err)
 				}
@@ -85,7 +77,7 @@ func (d *Daemon) startStatsReporter(interval time.Duration) {
 
 	go func() {
 		defer func() {
-			slog.Info("analytics exiting")
+			slog.Info("metrics reporter exiting")
 			d.wg.Done()
 		}()
 
@@ -107,7 +99,7 @@ func (d *Daemon) reportStats(includeLag bool) {
 	defer cancel()
 
 	log := slog.New(slog.Default().Handler())
-	metrics := make(map[string]float64, len(d.indexers)*2)
+	metrics := make([]types.MetricDatum, 0, len(d.indexers)*3)
 
 	for _, ix := range d.indexers {
 		stats := ix.Stats()
@@ -121,9 +113,13 @@ func (d *Daemon) reportStats(includeLag bool) {
 			rateInPeriod = float64(indexedInPeriod) / (float64(elapsedInPeriod) / float64(time.Second))
 		}
 
-		metrics[ix.Name()+"_indexed"] = float64(indexedInPeriod)
-		metrics[ix.Name()+"_deleted"] = float64(deletedInPeriod)
-		metrics[ix.Name()+"_rate"] = rateInPeriod
+		idxDim := cwatch.Dimension("Index", ix.Name())
+
+		metrics = append(metrics,
+			cwatch.Datum("RecordsIndexed", float64(indexedInPeriod), types.StandardUnitCount, idxDim),
+			cwatch.Datum("RecordsDeleted", float64(deletedInPeriod), types.StandardUnitCount, idxDim),
+			cwatch.Datum("IndexingRate", rateInPeriod, types.StandardUnitCountSecond, idxDim),
+		)
 
 		d.prevStats[ix] = stats
 
@@ -132,14 +128,13 @@ func (d *Daemon) reportStats(includeLag bool) {
 			if err != nil {
 				log.Error("error getting db last modified", "index", ix.Name(), "error", err)
 			} else {
-				metrics[ix.Name()+"_lag"] = lag.Seconds()
+				metrics = append(metrics, cwatch.Datum("IndexingLag", lag.Seconds(), types.StandardUnitSeconds, idxDim))
 			}
 		}
 	}
 
-	for k, v := range metrics {
-		analytics.Gauge("indexer."+k, v)
-		log = log.With(k, v)
+	if err := d.rt.CW.Send(ctx, metrics...); err != nil {
+		log.Error("error putting metrics", "error", err)
 	}
 
 	log.Info("stats reported")
@@ -151,7 +146,7 @@ func (d *Daemon) calculateLag(ctx context.Context, ix indexers.Indexer) (time.Du
 		return 0, fmt.Errorf("error getting ES last modified: %w", err)
 	}
 
-	dbLastModified, err := ix.GetDBLastModified(ctx, d.db)
+	dbLastModified, err := ix.GetDBLastModified(ctx, d.rt.DB)
 	if err != nil {
 		return 0, fmt.Errorf("error getting DB last modified: %w", err)
 	}
@@ -162,7 +157,6 @@ func (d *Daemon) calculateLag(ctx context.Context, ix indexers.Indexer) (time.Du
 // Stop stops this daemon
 func (d *Daemon) Stop() {
 	slog.Info("daemon stopping")
-	analytics.Stop()
 
 	close(d.quit)
 	d.wg.Wait()