chore: add Reporting#Stop

app/apphandlers/embeddedAppHandler.go

@@ -93,6 +93,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
  a.log.Infof("Configured deployment type: %q", deploymentType)
 
  reporting := a.app.Features().Reporting.Setup(ctx, backendconfig.DefaultBackendConfig)
+ defer reporting.Stop()
  syncer := reporting.DatabaseSyncer(types.SyncerConfig{ConnInfo: misc.GetConnectionString(config)})
  g.Go(func() error {
  syncer()

app/apphandlers/processorAppHandler.go

@@ -107,6 +107,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
  a.log.Infof("Configured deployment type: %q", deploymentType)
 
  reporting := a.app.Features().Reporting.Setup(ctx, backendconfig.DefaultBackendConfig)
+ defer reporting.Stop()
  syncer := reporting.DatabaseSyncer(types.SyncerConfig{ConnInfo: misc.GetConnectionString(config.Default)})
  g.Go(misc.WithBugsnag(func() error {
  syncer()

enterprise/reporting/error_index_reporting.go

@@ -117,10 +117,11 @@ func (eir *ErrorIndexReporter) Report(metrics []*types.PUReportedMetric, tx *Tx)
  return nil
 }
 
-// DatabaseSyncer returns a syncer that syncs the errorIndex jobsDB. Once the context is done, it stops the errorIndex jobsDB
-func (eir *ErrorIndexReporter) DatabaseSyncer(
- types.SyncerConfig,
-) types.ReportingSyncer {
+func (eir *ErrorIndexReporter) DatabaseSyncer(types.SyncerConfig) types.ReportingSyncer {
  return func() {
  }
 }
+
+func (edr *ErrorIndexReporter) Stop() {
+ // No op
+}

enterprise/reporting/error_reporting.go

@@ -54,6 +54,8 @@ var ErrorDetailReportsColumns = []string{
 
 type ErrorDetailReporter struct {
  ctx context.Context
+ cancel context.CancelFunc
+ g *errgroup.Group
  configSubscriber *configSubscriber
  reportingServiceURL string
  syncersMu sync.RWMutex
@@ -98,9 +100,12 @@ func NewErrorDetailReporter(
 
  log := logger.NewLogger().Child("enterprise").Child("error-detail-reporting")
  extractor := NewErrorDetailExtractor(log)
-
+ ctx, cancel := context.WithCancel(ctx)
+ g, ctx := errgroup.WithContext(ctx)
  return &ErrorDetailReporter{
  ctx: ctx,
+ cancel: cancel,
+ g: g,
  reportingServiceURL: reportingServiceURL,
  log: log,
  sleepInterval: sleepInterval,
@@ -140,7 +145,10 @@ func (edr *ErrorDetailReporter) DatabaseSyncer(c types.SyncerConfig) types.Repor
  }
 
  return func() {
- edr.mainLoop(edr.ctx, c)
+ edr.g.Go(func() error {
+ edr.mainLoop(edr.ctx, c)
+ return nil
+ })
  }
 }
 
@@ -572,4 +580,7 @@ func (edr *ErrorDetailReporter) sendMetric(ctx context.Context, label string, me
  return err
 }
 
-// Sending metrics to Reporting service --- ENDS
+func (edr *ErrorDetailReporter) Stop() {
+ edr.cancel()
+ _ = edr.g.Wait()
+}

enterprise/reporting/mediator.go

@@ -20,14 +20,22 @@ type Mediator struct {
 
  g *errgroup.Group
  ctx context.Context
+ cancel context.CancelFunc
  reporters []types.Reporting
 }
 
 func NewReportingMediator(ctx context.Context, log logger.Logger, enterpriseToken string, backendConfig backendconfig.BackendConfig) *Mediator {
+ ctx, cancel := context.WithCancel(ctx)
+ g, ctx := errgroup.WithContext(ctx)
+
  rm := &Mediator{
- log: log,
+ log: log,
+ g: g,
+ ctx: ctx,
+ cancel: cancel,
  }
- rm.g, rm.ctx = errgroup.WithContext(ctx)
+ rm.ctx, rm.cancel = context.WithCancel(ctx)
+ rm.g, rm.ctx = errgroup.WithContext(rm.ctx)
 
  reportingEnabled := config.GetBool("Reporting.enabled", types.DefaultReportingEnabled)
  if enterpriseToken == "" || !reportingEnabled {
@@ -108,3 +116,8 @@ func (rm *Mediator) DatabaseSyncer(c types.SyncerConfig) types.ReportingSyncer {
  _ = rm.g.Wait()
  }
 }
+
+func (rm *Mediator) Stop() {
+ rm.cancel()
+ _ = rm.g.Wait()
+}

enterprise/reporting/noop.go

@@ -15,3 +15,5 @@ func (*NOOP) Report(_ []*types.PUReportedMetric, _ *Tx) error {
 func (*NOOP) DatabaseSyncer(c types.SyncerConfig) types.ReportingSyncer {
  return func() {}
 }
+
+func (*NOOP) Stop() {}

enterprise/reporting/reporting.go

@@ -46,6 +46,8 @@ const (
 
 type DefaultReporter struct {
  ctx context.Context
+ cancel context.CancelFunc
+ g *errgroup.Group
  configSubscriber *configSubscriber
  syncersMu sync.RWMutex
  syncers map[string]*types.SyncSource
@@ -85,8 +87,12 @@ func NewDefaultReporter(ctx context.Context, log logger.Logger, configSubscriber
  if whActionsOnly {
  log.Info("REPORTING_WH_ACTIONS_ONLY enabled.only sending reports relevant to wh actions.")
  }
+ ctx, cancel := context.WithCancel(ctx)
+ g, ctx := errgroup.WithContext(ctx)
  return &DefaultReporter{
  ctx: ctx,
+ cancel: cancel,
+ g: g,
  log: log,
  configSubscriber: configSubscriber,
  syncers: make(map[string]*types.SyncSource),
@@ -136,7 +142,10 @@ func (r *DefaultReporter) DatabaseSyncer(c types.SyncerConfig) types.ReportingSy
  return func() {}
  }
  return func() {
- r.mainLoop(r.ctx, c)
+ r.g.Go(func() error {
+ r.mainLoop(r.ctx, c)
+ return nil
+ })
  }
 }
 
@@ -602,3 +611,8 @@ func (r *DefaultReporter) getTags(label string) stats.Tags {
  "clientName": label,
  }
 }
+
+func (r *DefaultReporter) Stop() {
+ r.cancel()
+ _ = r.g.Wait()
+}

enterprise/reporting/setup_test.go

@@ -121,12 +121,8 @@ func TestSetupForDelegates(t *testing.T) {
  EnterpriseToken: "dummy-token",
  }
  }
- ctx, cancel := context.WithCancel(context.Background())
- med := NewReportingMediator(ctx, logger.NOP, f.EnterpriseToken, &backendconfig.NOOP{})
- defer func() {
- cancel()
- _ = med.g.Wait()
- }()
+ med := NewReportingMediator(context.Background(), logger.NOP, f.EnterpriseToken, &backendconfig.NOOP{})
+ defer med.Stop()
  require.Len(t, med.reporters, tc.expectedDelegates)
  })
 

utils/types/types.go

@@ -65,6 +65,9 @@ type Reporting interface {
 
  // DatabaseSyncer creates reporting tables in the database and returns a function to periodically sync the data
  DatabaseSyncer(c SyncerConfig) ReportingSyncer
+
+ // Stop the reporting service
+ Stop()
 }
 
 type ReportingSyncer func()

warehouse/app.go

@@ -341,6 +341,7 @@ func (a *App) Run(ctx context.Context) error {
 
  if !mode.IsStandAloneSlave(a.config.mode) {
  a.reporting = a.app.Features().Reporting.Setup(gCtx, a.bcConfig)
+ defer a.reporting.Stop()
  syncer := a.reporting.DatabaseSyncer(types.SyncerConfig{ConnInfo: a.connectionString(), Label: types.WarehouseReportingLabel})
  g.Go(misc.WithBugsnagForWarehouse(func() error {
  syncer()