services: add failure case to Stopping function (#2231)

pstibrany · web-flow · commit 3374e33bde0e · 2020-03-09T11:41:06.000+01:00
Stopping function now takes failureCase error as parameter.

This allows stopping function to react differently based on
whether running function has failed or not.

In this commit, all stopping functions simply ignore the passed
parameter.

Signed-off-by: Peter Štibraný &lt;peter.stibrany@grafana.com&gt;
diff --git a/pkg/alertmanager/multitenant.go b/pkg/alertmanager/multitenant.go
@@ -241,7 +241,7 @@ func (am *MultitenantAlertmanager) iteration(ctx context.Context) error {
 }
 
 // stopping runs when MultitenantAlertmanager transitions to Stopping state.
-func (am *MultitenantAlertmanager) stopping() error {
+func (am *MultitenantAlertmanager) stopping(_ error) error {
 	am.alertmanagersMtx.Lock()
 	for _, am := range am.alertmanagers {
 		am.Stop()
diff --git a/pkg/chunk/purger/purger.go b/pkg/chunk/purger/purger.go
@@ -114,7 +114,7 @@ func (dp *DataPurger) init(ctx context.Context) error {
 }
 
 // Stop waits until all background tasks stop.
-func (dp *DataPurger) stop() error {
+func (dp *DataPurger) stop(_ error) error {
 	dp.wg.Wait()
 	return nil
 }
diff --git a/pkg/chunk/table_manager.go b/pkg/chunk/table_manager.go
@@ -190,7 +190,7 @@ func (m *TableManager) starting(ctx context.Context) error {
 }
 
 // Stop the TableManager
-func (m *TableManager) stopping() error {
+func (m *TableManager) stopping(_ error) error {
 	if m.bucketRetentionLoop != nil {
 		return services.StopAndAwaitTerminated(context.Background(), m.bucketRetentionLoop)
 	}
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
@@ -198,7 +198,7 @@ func (c *Compactor) starting(ctx context.Context) error {
 	return errors.Wrap(err, "failed to initialize compactor objects")
 }
 
-func (c *Compactor) stopping() error {
+func (c *Compactor) stopping(_ error) error {
 	if c.subservices != nil {
 		return services.StopManagerAndAwaitStopped(context.Background(), c.subservices)
 	}
diff --git a/pkg/cortex/module_service_wrapper.go b/pkg/cortex/module_service_wrapper.go
@@ -70,7 +70,7 @@ func (w *moduleServiceWrapper) run(serviceContext context.Context) error {
 	return w.service.FailureCase()
 }
 
-func (w *moduleServiceWrapper) stop() error {
+func (w *moduleServiceWrapper) stop(_ error) error {
 	// wait until all stopDeps have stopped
 	for _, m := range w.stopDeps {
 		s := w.serviceMap[m]
diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go
@@ -297,7 +297,7 @@ func (t *Cortex) initQuerier(cfg *Config) (serv services.Service, err error) {
 	// BUT this extra functionality is ONE OF THE REASONS to introduce entire "Services" concept into Cortex.
 	// For now, only return service that stops the worker, and Querier will be used even before storeQueryable has finished starting.
 
-	return services.NewIdleService(nil, func() error {
+	return services.NewIdleService(nil, func(_ error) error {
 		t.worker.Stop()
 		return nil
 	}), nil
@@ -370,7 +370,7 @@ func (t *Cortex) initStore(cfg *Config) (serv services.Service, err error) {
 		return
 	}
 
-	return services.NewIdleService(nil, func() error {
+	return services.NewIdleService(nil, func(_ error) error {
 		t.store.Stop()
 		return nil
 	}), nil
@@ -415,7 +415,7 @@ func (t *Cortex) initQueryFrontend(cfg *Config) (serv services.Service, err erro
 			t.frontend.Handler(),
 		),
 	)
-	return services.NewIdleService(nil, func() error {
+	return services.NewIdleService(nil, func(_ error) error {
 		t.frontend.Close()
 		if t.cache != nil {
 			t.cache.Stop()
@@ -491,7 +491,7 @@ func (t *Cortex) initConfig(cfg *Config) (serv services.Service, err error) {
 
 	t.configAPI = api.New(t.configDB, cfg.Configs.API)
 	t.configAPI.RegisterRoutes(t.server.HTTP)
-	return services.NewIdleService(nil, func() error {
+	return services.NewIdleService(nil, func(_ error) error {
 		t.configDB.Close()
 		return nil
 	}), nil
@@ -532,7 +532,7 @@ func (t *Cortex) initMemberlistKV(cfg *Config) (services.Service, error) {
 	}
 	t.memberlistKVState = newMemberlistKVState(&cfg.MemberlistKV)
 
-	return services.NewIdleService(nil, func() error {
+	return services.NewIdleService(nil, func(_ error) error {
 		kv := t.memberlistKVState.kv
 		if kv != nil {
 			kv.Stop()
diff --git a/pkg/cortex/server_service.go b/pkg/cortex/server_service.go
@@ -41,7 +41,7 @@ func NewServerService(cfg *Config, serv *server.Server, servicesToWaitFor func()
 		}
 	}
 
-	stoppingFn := func() error {
+	stoppingFn := func(_ error) error {
 		// wait until all modules are done, and then shutdown server.
 		for _, s := range servicesToWaitFor() {
 			_ = s.AwaitTerminated(context.Background())
diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go
@@ -219,7 +219,7 @@ func (d *Distributor) starting(ctx context.Context) error {
 }
 
 // Called after distributor is asked to stop via StopAsync.
-func (d *Distributor) stopping() error {
+func (d *Distributor) stopping(_ error) error {
 	return services.StopManagerAndAwaitStopped(context.Background(), d.subservices)
 }
 
diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go
@@ -239,7 +239,7 @@ func (i *Ingester) loop(ctx context.Context) error {
 }
 
 // stopping is run when ingester is asked to stop
-func (i *Ingester) stopping() error {
+func (i *Ingester) stopping(_ error) error {
 	i.wal.Stop()
 
 	// This will prevent us accepting any more samples
diff --git a/pkg/ingester/ingester_v2.go b/pkg/ingester/ingester_v2.go
@@ -169,7 +169,7 @@ func (i *Ingester) startingV2(ctx context.Context) error {
 }
 
 // runs when V2 ingester is stopping
-func (i *Ingester) stoppingV2() error {
+func (i *Ingester) stoppingV2(_ error) error {
 	// It's important to wait until shipper is finished,
 	// because the blocks transfer should start only once it's guaranteed
 	// there's no shipping on-going.
diff --git a/pkg/querier/block.go b/pkg/querier/block.go
@@ -58,7 +58,7 @@ func (b *BlockQueryable) starting(ctx context.Context) error {
 	return errors.Wrap(services.StartAndAwaitRunning(ctx, b.us), "failed to start UserStore")
 }
 
-func (b *BlockQueryable) stopping() error {
+func (b *BlockQueryable) stopping(_ error) error {
 	return errors.Wrap(services.StopAndAwaitTerminated(context.Background(), b.us), "stopping UserStore")
 }
 
diff --git a/pkg/querier/block_store.go b/pkg/querier/block_store.go
@@ -120,7 +120,7 @@ func (u *UserStore) starting(ctx context.Context) error {
 	return nil
 }
 
-func (u *UserStore) stopping() error {
+func (u *UserStore) stopping(_ error) error {
 	u.serv.Stop()
 	return nil
 }
diff --git a/pkg/ring/lifecycler.go b/pkg/ring/lifecycler.go
@@ -425,7 +425,7 @@ func (i *Lifecycler) loop(ctx context.Context) error {
 // - send chunks to another ingester, if it can.
 // - otherwise, flush chunks to the chunk store.
 // - remove config from Consul.
-func (i *Lifecycler) stopping() error {
+func (i *Lifecycler) stopping(_ error) error {
 	heartbeatTicker := time.NewTicker(i.cfg.HeartbeatPeriod)
 	defer heartbeatTicker.Stop()
 
diff --git a/pkg/ruler/ruler.go b/pkg/ruler/ruler.go
@@ -191,7 +191,7 @@ func (r *Ruler) starting(ctx context.Context) error {
 
 // Stop stops the Ruler.
 // Each function of the ruler is terminated before leaving the ring
-func (r *Ruler) stopping() error {
+func (r *Ruler) stopping(_ error) error {
 	r.notifiersMtx.Lock()
 	for _, n := range r.notifiers {
 		n.stop()
diff --git a/pkg/util/runtimeconfig/manager.go b/pkg/util/runtimeconfig/manager.go
@@ -165,7 +165,7 @@ func (om *Manager) callListeners(newValue interface{}) {
 }
 
 // Stop stops the Manager
-func (om *Manager) stop() error {
+func (om *Manager) stop(_ error) error {
 	om.listenersMtx.Lock()
 	defer om.listenersMtx.Unlock()
 
diff --git a/pkg/util/services/basic_service.go b/pkg/util/services/basic_service.go
@@ -22,7 +22,8 @@ type RunningFn func(serviceContext context.Context) error
 // StoppingFn function is called when service enters Stopping state. When it returns, service moves to Terminated or Failed state,
 // depending on whether there was any error returned from previous RunningFn (if it was called) and this StoppingFn function. If both return error,
 // RunningFn's error will be saved as failure case for Failed state.
-type StoppingFn func() error
+// Parameter is error from Running function, or nil if there was no error.
+type StoppingFn func(failureCase error) error
 
 // BasicService implements contract of Service interface, using three supplied functions: StartingFn, RunningFn and StoppingFn.
 // When service is started, these three functions are called as service transitions to Starting, Running and Stopping state.
@@ -180,7 +181,7 @@ stop:
 	b.serviceCancel()
 
 	if b.stoppingFn != nil {
-		err = b.stoppingFn()
+		err = b.stoppingFn(failure)
 		if failure == nil {
 			failure = err
 		}
diff --git a/pkg/util/services/basic_service_test.go b/pkg/util/services/basic_service_test.go
@@ -60,7 +60,7 @@ func (s *serv) run(ctx context.Context) error {
 	return s.conf.runRetVal
 }
 
-func (s *serv) shutDown() error {
+func (s *serv) shutDown(_ error) error {
 	return s.conf.stopRetVal
 }
 
@@ -293,3 +293,23 @@ func (sl *serviceListener) Terminated(from State) {
 	sl.ch <- fmt.Sprintf("terminated: %v", from)
 	close(sl.ch)
 }
+
+func TestFailureCaseFromRunningIsPassedToStopping(t *testing.T) {
+	err := errors.New("test error")
+
+	s := NewBasicService(nil, func(_ context.Context) error {
+		return err
+	}, func(failureCase error) error {
+		if failureCase != err {
+			t.Log("invalid failureCase")
+			t.Fail()
+		}
+		return nil
+	})
+
+	require.NoError(t, s.StartAsync(context.Background()))
+	require.Error(t, s.AwaitTerminated(context.Background())) // service fails, so we expect Failed state
+
+	fc := s.FailureCase()
+	require.Equal(t, err, fc)
+}
diff --git a/pkg/util/services/services_test.go b/pkg/util/services/services_test.go
@@ -19,7 +19,7 @@ func TestIdleService(t *testing.T) {
 	s := NewIdleService(func(ctx context.Context) error {
 		started = true
 		return nil
-	}, func() error {
+	}, func(_ error) error {
 		stopped = true
 		return nil
 	})
@@ -109,7 +109,7 @@ func TestHelperFunctionsStopError(t *testing.T) {
 	t.Parallel()
 
 	e := errors.New("some error")
-	s := NewIdleService(nil, func() error { return e })
+	s := NewIdleService(nil, func(_ error) error { return e })
 
 	require.NoError(t, StartAndAwaitRunning(context.Background(), s))
 	require.Equal(t, e, StopAndAwaitTerminated(context.Background(), s))
@@ -118,7 +118,7 @@ func TestHelperFunctionsStopError(t *testing.T) {
 func TestHelperFunctionsStopTooSlow(t *testing.T) {
 	t.Parallel()
 
-	s := NewIdleService(nil, func() error { time.Sleep(1 * time.Second); return nil })
+	s := NewIdleService(nil, func(_ error) error { time.Sleep(1 * time.Second); return nil })
 
 	require.NoError(t, StartAndAwaitRunning(context.Background(), s))
 

Original file line number	Diff line number	Diff line change
`@@ -241,7 +241,7 @@ func (am *MultitenantAlertmanager) iteration(ctx context.Context) error {`
`241`	`241`	`}`
`242`	`242`
`243`	`243`	`// stopping runs when MultitenantAlertmanager transitions to Stopping state.`
`244`		`-func (am *MultitenantAlertmanager) stopping() error {`
	`244`	`+func (am *MultitenantAlertmanager) stopping(_ error) error {`
`245`	`245`	`am.alertmanagersMtx.Lock()`
`246`	`246`	`for _, am := range am.alertmanagers {`
`247`	`247`	`am.Stop()`
Original file line number	Diff line number	Diff line change
`@@ -114,7 +114,7 @@ func (dp *DataPurger) init(ctx context.Context) error {`
`114`	`114`	`}`
`115`	`115`
`116`	`116`	`// Stop waits until all background tasks stop.`
`117`		`-func (dp *DataPurger) stop() error {`
	`117`	`+func (dp *DataPurger) stop(_ error) error {`
`118`	`118`	`dp.wg.Wait()`
`119`	`119`	`return nil`
`120`	`120`	`}`
Original file line number	Diff line number	Diff line change
`@@ -190,7 +190,7 @@ func (m *TableManager) starting(ctx context.Context) error {`
`190`	`190`	`}`
`191`	`191`
`192`	`192`	`// Stop the TableManager`
`193`		`-func (m *TableManager) stopping() error {`
	`193`	`+func (m *TableManager) stopping(_ error) error {`
`194`	`194`	`if m.bucketRetentionLoop != nil {`
`195`	`195`	`return services.StopAndAwaitTerminated(context.Background(), m.bucketRetentionLoop)`
`196`	`196`	`}`
Original file line number	Diff line number	Diff line change
`@@ -198,7 +198,7 @@ func (c *Compactor) starting(ctx context.Context) error {`
`198`	`198`	`return errors.Wrap(err, "failed to initialize compactor objects")`
`199`	`199`	`}`
`200`	`200`
`201`		`-func (c *Compactor) stopping() error {`
	`201`	`+func (c *Compactor) stopping(_ error) error {`
`202`	`202`	`if c.subservices != nil {`
`203`	`203`	`return services.StopManagerAndAwaitStopped(context.Background(), c.subservices)`
`204`	`204`	`}`
Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ func (w *moduleServiceWrapper) run(serviceContext context.Context) error {`
`70`	`70`	`return w.service.FailureCase()`
`71`	`71`	`}`
`72`	`72`
`73`		`-func (w *moduleServiceWrapper) stop() error {`
	`73`	`+func (w *moduleServiceWrapper) stop(_ error) error {`
`74`	`74`	`// wait until all stopDeps have stopped`
`75`	`75`	`for _, m := range w.stopDeps {`
`76`	`76`	`s := w.serviceMap[m]`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@ func NewServerService(cfg Config, serv server.Server, servicesToWaitFor func()`
`41`	`41`	`}`
`42`	`42`	`}`
`43`	`43`
`44`		`- stoppingFn := func() error {`
	`44`	`+ stoppingFn := func(_ error) error {`
`45`	`45`	`// wait until all modules are done, and then shutdown server.`
`46`	`46`	`for _, s := range servicesToWaitFor() {`
`47`	`47`	`_ = s.AwaitTerminated(context.Background())`
Original file line number	Diff line number	Diff line change
`@@ -219,7 +219,7 @@ func (d *Distributor) starting(ctx context.Context) error {`
`219`	`219`	`}`
`220`	`220`
`221`	`221`	`// Called after distributor is asked to stop via StopAsync.`
`222`		`-func (d *Distributor) stopping() error {`
	`222`	`+func (d *Distributor) stopping(_ error) error {`
`223`	`223`	`return services.StopManagerAndAwaitStopped(context.Background(), d.subservices)`
`224`	`224`	`}`
`225`	`225`
Original file line number	Diff line number	Diff line change
`@@ -239,7 +239,7 @@ func (i *Ingester) loop(ctx context.Context) error {`
`239`	`239`	`}`
`240`	`240`
`241`	`241`	`// stopping is run when ingester is asked to stop`
`242`		`-func (i *Ingester) stopping() error {`
	`242`	`+func (i *Ingester) stopping(_ error) error {`
`243`	`243`	`i.wal.Stop()`
`244`	`244`
`245`	`245`	`// This will prevent us accepting any more samples`
Original file line number	Diff line number	Diff line change
`@@ -169,7 +169,7 @@ func (i *Ingester) startingV2(ctx context.Context) error {`
`169`	`169`	`}`
`170`	`170`
`171`	`171`	`// runs when V2 ingester is stopping`
`172`		`-func (i *Ingester) stoppingV2() error {`
	`172`	`+func (i *Ingester) stoppingV2(_ error) error {`
`173`	`173`	`// It's important to wait until shipper is finished,`
`174`	`174`	`// because the blocks transfer should start only once it's guaranteed`
`175`	`175`	`// there's no shipping on-going.`