Add some APM instrumentation to storage indexer

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+* Add some APM instrumentation to storage indexer. [#939](https://github.com/elastic/package-registry/pull/939)
+
 ### Deprecated
 
 ### Known Issues

main.go

@@ -110,10 +110,13 @@ func main() {
 	logger := util.Logger()
 	defer logger.Sync()
 
+	apmTracer := initAPMTracer(logger)
+	defer apmTracer.Close()
+
 	config := mustLoadConfig(logger)
 	if dryRun {
 		logger.Info("Running dry-run mode")
-		_ = initIndexer(context.Background(), logger, config)
+		_ = initIndexer(context.Background(), logger, apmTracer, config)
 		os.Exit(0)
 	}
 
@@ -122,7 +125,7 @@ func main() {
 
 	initHttpProf(logger)
 
-	server := initServer(logger, config)
+	server := initServer(logger, apmTracer, config)
 	go func() {
 		err := runServer(server)
 		if err != nil && err != http.ErrServerClosed {
@@ -184,7 +187,7 @@ func initMetricsServer(logger *zap.Logger) {
 	}()
 }
 
-func initIndexer(ctx context.Context, logger *zap.Logger, config *Config) Indexer {
+func initIndexer(ctx context.Context, logger *zap.Logger, apmTracer *apm.Tracer, config *Config) Indexer {
 	packagesBasePaths := getPackagesBasePaths(config)
 
 	var combined CombinedIndexer
@@ -195,6 +198,7 @@ func initIndexer(ctx context.Context, logger *zap.Logger, config *Config) Indexe
 			logger.Fatal("can't initialize storage client", zap.Error(err))
 		}
 		combined = append(combined, storage.NewIndexer(storageClient, storage.IndexerOptions{
+			APMTracer:                    apmTracer,
 			PackageStorageBucketInternal: storageIndexerBucketInternal,
 			PackageStorageEndpoint:       storageEndpoint,
 			WatchInterval:                storageIndexerWatchInterval,
@@ -209,14 +213,13 @@ func initIndexer(ctx context.Context, logger *zap.Logger, config *Config) Indexe
 	return combined
 }
 
-func initServer(logger *zap.Logger, config *Config) *http.Server {
-	apmTracer := initAPMTracer(logger)
+func initServer(logger *zap.Logger, apmTracer *apm.Tracer, config *Config) *http.Server {
 	tx := apmTracer.StartTransaction("initServer", "backend.init")
 	defer tx.End()
 
 	ctx := apm.ContextWithTransaction(context.TODO(), tx)
 
-	indexer := initIndexer(ctx, logger, config)
+	indexer := initIndexer(ctx, logger, apmTracer, config)
 
 	router := mustLoadRouter(logger, config, indexer)
 	apmgorilla.Instrument(router, apmgorilla.WithTracer(apmTracer))

storage/cursor.go

@@ -10,6 +10,7 @@ import (
 
 	"cloud.google.com/go/storage"
 	"github.com/pkg/errors"
+	"go.elastic.co/apm"
 	"go.uber.org/zap"
 
 	"github.com/elastic/package-registry/util"
@@ -28,6 +29,9 @@ func (c *cursor) String() string {
 }
 
 func loadCursor(ctx context.Context, storageClient *storage.Client, bucketName, rootStoragePath string) (*cursor, error) {
+	span, ctx := apm.StartSpan(ctx, "LoadCursor", "app")
+	defer span.End()
+
 	logger := util.Logger()
 	logger.Debug("load cursor file")
 

storage/index.go

@@ -10,6 +10,7 @@ import (
 
 	"cloud.google.com/go/storage"
 	"github.com/pkg/errors"
+	"go.elastic.co/apm"
 	"go.uber.org/zap"
 
 	"github.com/elastic/package-registry/packages"
@@ -25,6 +26,9 @@ type packageIndex struct {
 }
 
 func loadSearchIndexAll(ctx context.Context, storageClient *storage.Client, bucketName, rootStoragePath string, aCursor cursor) (*searchIndexAll, error) {
+	span, ctx := apm.StartSpan(ctx, "LoadSearchIndexAll", "app")
+	defer span.End()
+
 	indexFile := searchIndexAllFile
 
 	logger := util.Logger()

storage/indexer.go

@@ -15,6 +15,7 @@ import (
 	"cloud.google.com/go/storage"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"go.elastic.co/apm"
 	"go.uber.org/zap"
 
 	"github.com/elastic/package-registry/metrics"
@@ -37,12 +38,16 @@ type Indexer struct {
 }
 
 type IndexerOptions struct {
+	APMTracer                    *apm.Tracer
 	PackageStorageBucketInternal string
 	PackageStorageEndpoint       string
 	WatchInterval                time.Duration
 }
 
 func NewIndexer(storageClient *storage.Client, options IndexerOptions) *Indexer {
+	if options.APMTracer == nil {
+		options.APMTracer = apm.DefaultTracer
+	}
 	return &Indexer{
 		storageClient: storageClient,
 		options:       options,
@@ -114,10 +119,15 @@ func (i *Indexer) watchIndices(ctx context.Context) {
 	for {
 		logger.Debug("watchIndices: start")
 
-		err = i.updateIndex(ctx)
-		if err != nil {
-			logger.Error("can't update index file", zap.Error(err))
-		}
+		func() {
+			tx := i.options.APMTracer.StartTransaction("updateIndex", "backend.watcher")
+			defer tx.End()
+
+			err = i.updateIndex(ctx)
+			if err != nil {
+				logger.Error("can't update index file", zap.Error(err))
+			}
+		}()
 
 		logger.Debug("watchIndices: finished")
 		select {
@@ -130,6 +140,9 @@ func (i *Indexer) watchIndices(ctx context.Context) {
 }
 
 func (i *Indexer) updateIndex(ctx context.Context) error {
+	span, ctx := apm.StartSpan(ctx, "UpdateIndex", "app")
+	defer span.End()
+
 	logger := util.Logger()
 	logger.Debug("Update indices")
 	start := time.Now()