Merge pull request #1242 from aduffeck/metrics

Metrics

services/postprocessing/README.md

@@ -119,3 +119,17 @@ Depending if you want to restart/resume all or defined failed uploads, different
       opencloud postprocessing resume -s "finished"  # Equivalent to the above
       opencloud postprocessing resume -s "virusscan" # Resume all uploads currently in virusscan step
       ```
+
+## Metrics
+
+The postprocessing service exposes the following prometheus metrics at `<debug_endpoint>/metrics` (as configured using the `POSTPROCESSING_DEBUG_ADDR` env var):
+
+| Metric Name | Type | Description | Labels |
+| --- | --- | --- | --- |
+| `opencloud_postprocessing_build_info` | Gauge | Build information | `version` |
+| `opencloud_postprocessing_events_outstanding_acks` | Gauge | Number of outstanding acks for events | |
+| `opencloud_postprocessing_events_unprocessed` | Gauge | Number of unprocessed events | |
+| `opencloud_postprocessing_events_redelivered` | Gauge | Number of redelivered events | |
+| `opencloud_postprocessing_in_progress` | Gauge | Number of postprocessing events in progress | |
+| `opencloud_postprocessing_finished` | Counter | Number of finished postprocessing events | `status` |
+| `opencloud_postprocessing_duration_seconds` | Histogram | Duration of postprocessing operations in seconds | `status` |

services/postprocessing/pkg/metrics/metrics.go

@@ -0,0 +1,77 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	// Namespace defines the namespace for the defines metrics.
+	Namespace = "opencloud"
+
+	// Subsystem defines the subsystem for the defines metrics.
+	Subsystem = "postprocessing"
+)
+
+// Metrics defines the available metrics of this service.
+type Metrics struct {
+	// Counter  *prometheus.CounterVec
+	BuildInfo             *prometheus.GaugeVec
+	EventsOutstandingAcks prometheus.Gauge
+	EventsUnprocessed     prometheus.Gauge
+	EventsRedelivered     prometheus.Gauge
+	InProgress            prometheus.Gauge
+	Finished              *prometheus.CounterVec
+	Duration              *prometheus.HistogramVec
+}
+
+// New initializes the available metrics.
+func New() *Metrics {
+	m := &Metrics{
+		BuildInfo: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "build_info",
+			Help:      "Build information",
+		}, []string{"version"}),
+		EventsOutstandingAcks: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_outstanding_acks",
+			Help:      "Number of outstanding acks for events",
+		}),
+		EventsUnprocessed: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_unprocessed",
+			Help:      "Number of unprocessed events",
+		}),
+		EventsRedelivered: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_redelivered",
+			Help:      "Number of redelivered events",
+		}),
+		InProgress: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "in_progress",
+			Help:      "Number of postprocessing events in progress",
+		}),
+		Finished: promauto.NewCounterVec(prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "finished",
+			Help:      "Number of finished postprocessing events",
+		}, []string{"status"}),
+		Duration: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "duration_seconds",
+			Help:      "Duration of postprocessing operations in seconds",
+			Buckets:   []float64{0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 1200},
+		}, []string{"status"}),
+	}
+
+	return m
+}

services/postprocessing/pkg/postprocessing/postprocessing.go

@@ -24,6 +24,7 @@ type Postprocessing struct {
 	Failures          int
 	InitiatorID       string
 	Finished          bool
+	StartTime         time.Time
 
 	config config.Postprocessing
 }

services/postprocessing/pkg/service/service.go

@@ -9,7 +9,9 @@ import (
 	"time"
 
 	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/pkg/version"
 	"github.com/opencloud-eu/opencloud/services/postprocessing/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/postprocessing/pkg/metrics"
 	"github.com/opencloud-eu/opencloud/services/postprocessing/pkg/postprocessing"
 	ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx"
 	"github.com/opencloud-eu/reva/v2/pkg/events"
@@ -22,14 +24,15 @@ import (
 
 // PostprocessingService is an instance of the service handling postprocessing of files
 type PostprocessingService struct {
-	ctx    context.Context
-	log    log.Logger
-	events <-chan raw.Event
-	pub    events.Publisher
-	steps  []events.Postprocessingstep
-	store  store.Store
-	c      config.Postprocessing
-	tp     trace.TracerProvider
+	ctx     context.Context
+	log     log.Logger
+	events  <-chan raw.Event
+	pub     events.Publisher
+	steps   []events.Postprocessingstep
+	store   store.Store
+	c       config.Postprocessing
+	tp      trace.TracerProvider
+	metrics *metrics.Metrics
 }
 
 var (
@@ -78,15 +81,20 @@ func NewPostprocessingService(ctx context.Context, logger log.Logger, sto store.
 		return nil, err
 	}
 
+	m := metrics.New()
+	m.BuildInfo.WithLabelValues(version.GetString()).Set(1)
+	monitorMetrics(raw, "postprocessing-pull", m, logger)
+
 	return &PostprocessingService{
-		ctx:    ctx,
-		log:    logger,
-		events: evs,
-		pub:    pub,
-		steps:  getSteps(cfg.Postprocessing),
-		store:  sto,
-		c:      cfg.Postprocessing,
-		tp:     tp,
+		ctx:     ctx,
+		log:     logger,
+		events:  evs,
+		pub:     pub,
+		steps:   getSteps(cfg.Postprocessing),
+		store:   sto,
+		c:       cfg.Postprocessing,
+		tp:      tp,
+		metrics: m,
 	}, nil
 }
 
@@ -149,7 +157,9 @@ func (pps *PostprocessingService) processEvent(e raw.Event) error {
 			Steps:             pps.steps,
 			InitiatorID:       e.InitiatorID,
 			ImpersonatingUser: ev.ImpersonatingUser,
+			StartTime:         time.Now(),
 		}
+		pps.metrics.InProgress.Inc()
 		next = pp.Init(ev)
 	case events.PostprocessingStepFinished:
 		if ev.UploadID == "" {
@@ -200,17 +210,27 @@ func (pps *PostprocessingService) processEvent(e raw.Event) error {
 			}
 		})
 	case events.UploadReady:
+		pps.metrics.InProgress.Dec()
+		// the upload failed - let's keep it around for a while - but mark it as finished
+		pp, err = pps.getPP(pps.store, ev.UploadID)
+		if err != nil {
+			pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot get upload")
+			return fmt.Errorf("%w: cannot get upload", ErrEvent)
+		}
+
 		if ev.Failed {
-			// the upload failed - let's keep it around for a while - but mark it as finished
-			pp, err = pps.getPP(pps.store, ev.UploadID)
-			if err != nil {
-				pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot get upload")
-				return fmt.Errorf("%w: cannot get upload", ErrEvent)
+			pps.metrics.Finished.WithLabelValues("failed").Inc()
+			if !pp.StartTime.IsZero() {
+				pps.metrics.Duration.WithLabelValues("failed").Observe(time.Since(pp.StartTime).Seconds())
 			}
 			pp.Finished = true
 			return storePP(pps.store, pp)
 		}
 
+		pps.metrics.Finished.WithLabelValues("succeeded").Inc()
+		if !pp.StartTime.IsZero() {
+			pps.metrics.Duration.WithLabelValues("succeeded").Observe(time.Since(pp.StartTime).Seconds())
+		}
 		// the storage provider thinks the upload is done - so no need to keep it any more
 		if err := pps.store.Delete(ev.UploadID); err != nil {
 			pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot delete upload")
@@ -360,3 +380,25 @@ func (pps *PostprocessingService) findUploadsByStep(step events.Postprocessingst
 
 	return ids
 }
+
+func monitorMetrics(stream raw.Stream, name string, m *metrics.Metrics, logger log.Logger) {
+	ctx := context.Background()
+	consumer, err := stream.JetStream().Consumer(ctx, name)
+	if err != nil {
+		logger.Error().Err(err).Msg("failed to get consumer")
+	}
+	ticker := time.NewTicker(5 * time.Second)
+	go func() {
+		for range ticker.C {
+			info, err := consumer.Info(ctx)
+			if err != nil {
+				logger.Error().Err(err).Msg("failed to get consumer")
+			}
+
+			m.EventsOutstandingAcks.Set(float64(info.NumAckPending))
+			m.EventsUnprocessed.Set(float64(info.NumPending))
+			m.EventsRedelivered.Set(float64(info.NumRedelivered))
+			logger.Trace().Msg("updated postprocessing event metrics")
+		}
+	}()
+}

services/search/.mockery.yaml

@@ -15,4 +15,4 @@ packages:
       Retriever: {}
   github.com/opencloud-eu/opencloud/services/search/pkg/search:
     interfaces:
-      Searcher: {}
+      Searcher: {}

services/search/README.md

@@ -162,3 +162,16 @@ The indexing process tries to be self-healing in some situations.
 In the following example, let's assume a file tree `foo/bar/baz` exists.
 If the folder `bar` gets renamed to `new-bar`, the path to `baz` is no longer `foo/bar/baz` but `foo/new-bar/baz`.
 The search service checks the change and either just updates the path in the index or creates a new index for all items affected if none was present.
+
+## Metrics
+
+The search service exposes the following prometheus metrics at `<debug_endpoint>/metrics` (as configured using the `SEARCH_DEBUG_ADDR` env var):
+
+| Metric Name | Type | Description | Labels |
+| --- | --- | --- | --- |
+| `opencloud_search_build_info` | Gauge | Build information | `version` |
+| `opencloud_search_events_outstanding_acks` | Gauge | Number of outstanding acks for events | |
+| `opencloud_search_events_unprocessed` | Gauge | Number of unprocessed events | |
+| `opencloud_search_events_redelivered` | Gauge | Number of redelivered events | |
+| `opencloud_search_search_duration_seconds` | Histogram | Duration of search operations in seconds | `status` |
+| `opencloud_search_index_duration_seconds` | Histogram | Duration of indexing operations in seconds | `status` |

services/search/pkg/metrics/metrics.go

@@ -1,6 +1,9 @@
 package metrics
 
-import "github.com/prometheus/client_golang/prometheus"
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
 
 var (
 	// Namespace defines the namespace for the defines metrics.
@@ -13,21 +16,56 @@ var (
 // Metrics defines the available metrics of this service.
 type Metrics struct {
 	// Counter  *prometheus.CounterVec
-	BuildInfo *prometheus.GaugeVec
+	BuildInfo             *prometheus.GaugeVec
+	EventsOutstandingAcks prometheus.Gauge
+	EventsUnprocessed     prometheus.Gauge
+	EventsRedelivered     prometheus.Gauge
+	SearchDuration        *prometheus.HistogramVec
+	IndexDuration         *prometheus.HistogramVec
 }
 
 // New initializes the available metrics.
 func New() *Metrics {
 	m := &Metrics{
-		BuildInfo: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		BuildInfo: promauto.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: Namespace,
 			Subsystem: Subsystem,
 			Name:      "build_info",
 			Help:      "Build information",
 		}, []string{"version"}),
+		EventsOutstandingAcks: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_outstanding_acks",
+			Help:      "Number of outstanding acks for events",
+		}),
+		EventsUnprocessed: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_unprocessed",
+			Help:      "Number of unprocessed events",
+		}),
+		EventsRedelivered: promauto.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_redelivered",
+			Help:      "Number of redelivered events",
+		}),
+		SearchDuration: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "search_duration_seconds",
+			Help:      "Duration of search operations in seconds",
+			Buckets:   []float64{0.1, 0.5, 1, 2.5, 5, 10, 30, 60},
+		}, []string{"status"}),
+		IndexDuration: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "index_duration_seconds",
+			Help:      "Duration of indexing operations in seconds",
+			Buckets:   []float64{0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 1200},
+		}, []string{"status"}),
 	}
 
-	_ = prometheus.Register(m.BuildInfo)
-	// TODO: implement metrics
 	return m
 }

services/search/pkg/search/events.go

@@ -1,11 +1,13 @@
 package search
 
 import (
+	"context"
 	"time"
 
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/metrics"
 	"github.com/opencloud-eu/reva/v2/pkg/events"
 	"github.com/opencloud-eu/reva/v2/pkg/events/raw"
 	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
@@ -13,7 +15,7 @@ import (
 
 // HandleEvents listens to the needed events,
 // it handles the whole resource indexing livecycle.
-func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, logger log.Logger) error {
+func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, m *metrics.Metrics, logger log.Logger) error {
 	evts := []events.Unmarshaller{
 		events.ItemTrashed{},
 		events.ItemRestored{},
@@ -37,6 +39,10 @@ func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, logger log.
 		return err
 	}
 
+	if m != nil {
+		monitorMetrics(stream, "search-pull", m, logger)
+	}
+
 	if cfg.Events.NumConsumers == 0 {
 		cfg.Events.NumConsumers = 1
 	}
@@ -103,3 +109,25 @@ func HandleEvents(s Searcher, stream raw.Stream, cfg *config.Config, logger log.
 
 	return nil
 }
+
+func monitorMetrics(stream raw.Stream, name string, m *metrics.Metrics, logger log.Logger) {
+	ctx := context.Background()
+	consumer, err := stream.JetStream().Consumer(ctx, name)
+	if err != nil {
+		logger.Error().Err(err).Msg("failed to get consumer")
+	}
+	ticker := time.NewTicker(5 * time.Second)
+	go func() {
+		for range ticker.C {
+			info, err := consumer.Info(ctx)
+			if err != nil {
+				logger.Error().Err(err).Msg("failed to get consumer")
+			}
+
+			m.EventsOutstandingAcks.Set(float64(info.NumAckPending))
+			m.EventsUnprocessed.Set(float64(info.NumPending))
+			m.EventsRedelivered.Set(float64(info.NumRedelivered))
+			logger.Trace().Msg("updated search event metrics")
+		}
+	}()
+}

services/search/pkg/search/events_test.go

@@ -31,7 +31,7 @@ var _ = DescribeTable("events",
 			Events: config.Events{
 				AsyncUploads: asyncUploads,
 			},
-		}, log.NewLogger())
+		}, nil, log.NewLogger())
 
 		for _, mck := range mcks {
 			s.On(mck, mock.Anything, mock.Anything).Return(nil).Run(func(args mock.Arguments) {

services/search/pkg/search/service.go

@@ -14,6 +14,7 @@ import (
 	rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
 	collaborationv1beta1 "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	libregraph "github.com/opencloud-eu/libre-graph-api-go"
 	revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
 	"github.com/opencloud-eu/reva/v2/pkg/errtypes"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
@@ -21,7 +22,6 @@ import (
 	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/walker"
 	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
 	"github.com/opencloud-eu/reva/v2/pkg/utils"
-	libregraph "github.com/opencloud-eu/libre-graph-api-go"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/types/known/fieldmaskpb"
 
@@ -31,6 +31,7 @@ import (
 	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/engine"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/metrics"
 )
 
 const (
@@ -59,6 +60,7 @@ type Service struct {
 	gatewaySelector pool.Selectable[gateway.GatewayAPIClient]
 	engine          engine.Engine
 	extractor       content.Extractor
+	metrics         *metrics.Metrics
 
 	serviceAccountID     string
 	serviceAccountSecret string
@@ -67,12 +69,13 @@ type Service struct {
 var errSkipSpace error
 
 // NewService creates a new Provider instance.
-func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng engine.Engine, extractor content.Extractor, logger log.Logger, cfg *config.Config) *Service {
+func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng engine.Engine, extractor content.Extractor, metrics *metrics.Metrics, logger log.Logger, cfg *config.Config) *Service {
 	var s = &Service{
 		gatewaySelector: gatewaySelector,
 		engine:          eng,
 		logger:          logger,
 		extractor:       extractor,
+		metrics:         metrics,
 
 		serviceAccountID:     cfg.ServiceAccount.ServiceAccountID,
 		serviceAccountSecret: cfg.ServiceAccount.ServiceAccountSecret,
@@ -85,6 +88,21 @@ func NewService(gatewaySelector pool.Selectable[gateway.GatewayAPIClient], eng e
 func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*searchsvc.SearchResponse, error) {
 	s.logger.Debug().Str("query", req.Query).Msg("performing a search")
 
+	// collect metrics
+	startTime := time.Now()
+	success := false
+	defer func() {
+		if s.metrics == nil {
+			return
+		}
+
+		status := "success"
+		if !success {
+			status = "error"
+		}
+		s.metrics.SearchDuration.WithLabelValues(status).Observe(time.Since(startTime).Seconds())
+	}()
+
 	gatewayClient, err := s.gatewaySelector.Next()
 	if err != nil {
 		return nil, err
@@ -255,6 +273,7 @@ func (s *Service) Search(ctx context.Context, req *searchsvc.SearchRequest) (*se
 		matches = matches[0:limit]
 	}
 
+	success = true
 	return &searchsvc.SearchResponse{
 		Matches:      matches,
 		TotalMatches: total,
@@ -425,6 +444,20 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId) error {
 	}
 	rootID.OpaqueId = rootID.SpaceId
 
+	// Collect metrics
+	startTime := time.Now()
+	success := false
+	defer func() {
+		if s.metrics == nil {
+			return
+		}
+		status := "success"
+		if !success {
+			status = "error"
+		}
+		s.metrics.IndexDuration.WithLabelValues(status).Observe(time.Since(startTime).Seconds())
+	}()
+
 	w := walker.NewWalker(s.gatewaySelector)
 	err = w.Walk(ownerCtx, &rootID, func(wd string, info *provider.ResourceInfo, err error) error {
 		if err != nil {
@@ -465,6 +498,7 @@ func (s *Service) IndexSpace(spaceID *provider.StorageSpaceId) error {
 	}
 
 	logDocCount(s.engine, s.logger)
+	success = true
 
 	return nil
 }

services/search/pkg/search/service_test.go

@@ -90,7 +90,7 @@ var _ = Describe("Searchprovider", func() {
 		indexClient = &engineMocks.Engine{}
 		extractor = &contentMocks.Extractor{}
 
-		s = search.NewService(gatewaySelector, indexClient, extractor, logger, &config.Config{})
+		s = search.NewService(gatewaySelector, indexClient, extractor, nil, logger, &config.Config{})
 
 		gatewayClient.On("Authenticate", mock.Anything, mock.Anything).Return(&gateway.AuthenticateResponse{
 			Status: status.NewOK(ctx),
@@ -110,7 +110,7 @@ var _ = Describe("Searchprovider", func() {
 
 	Describe("New", func() {
 		It("returns a new instance", func() {
-			s := search.NewService(gatewaySelector, indexClient, extractor, logger, &config.Config{})
+			s := search.NewService(gatewaySelector, indexClient, extractor, nil, logger, &config.Config{})
 			Expect(s).ToNot(BeNil())
 		})
 	})

services/search/pkg/server/grpc/server.go

@@ -36,6 +36,7 @@ func Server(opts ...Option) (grpc.Service, func(), error) {
 		svc.Logger(options.Logger),
 		svc.JWTSecret(options.JWTSecret),
 		svc.TracerProvider(options.TraceProvider),
+		svc.Metrics(options.Metrics),
 	)
 	if err != nil {
 		options.Logger.Error().

services/search/pkg/service/grpc/v0/option.go

@@ -3,6 +3,7 @@ package service
 import (
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/metrics"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -15,6 +16,7 @@ type Options struct {
 	Config         *config.Config
 	JWTSecret      string
 	TracerProvider trace.TracerProvider
+	Metrics        *metrics.Metrics
 }
 
 func newOptions(opts ...Option) Options {
@@ -54,3 +56,12 @@ func TracerProvider(val trace.TracerProvider) Option {
 		o.TracerProvider = val
 	}
 }
+
+// Metrics provides a function to set the Metrics option.
+func Metrics(val *metrics.Metrics) Option {
+	return func(o *Options) {
+		if val != nil {
+			o.Metrics = val
+		}
+	}
+}

services/search/pkg/service/grpc/v0/service.go

@@ -79,7 +79,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 		return nil, teardown, fmt.Errorf("unknown search extractor: %s", cfg.Extractor.Type)
 	}
 
-	ss := search.NewService(selector, eng, extractor, logger, cfg)
+	ss := search.NewService(selector, eng, extractor, options.Metrics, logger, cfg)
 
 	// setup event handling
 
@@ -98,7 +98,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 		return nil, teardown, err
 	}
 
-	if err := search.HandleEvents(ss, stream, cfg, logger); err != nil {
+	if err := search.HandleEvents(ss, stream, cfg, options.Metrics, logger); err != nil {
 		return nil, teardown, err
 	}