Repository: cortexproject/cortex
Pull request #2166 (2020-02-20 11:29:12)

Convert Cortex components to services
This PR is a supplement to [proposal document](https://docs.google.com/document/d/1GMpNK4EuQ1IQQOLZuub5vTWXCUMZdSMNgKrAEeq1Cy0/edit?usp=sharing) to revamp service model used by Cortex.

The document and this PR are based on the concept of "Service", as used in [Google Guava library](https://github.com/google/guava/wiki/ServiceExplained): Service is an object with operational state, with methods to start/stop the service, well-defined and observable state, methods for waiting for state transitions. Google Guava is a Java library, here we use reimplementation of [Services concept in Go](https://github.com/pstibrany/services). **Update: now part of Cortex, as of PR #2188.**

In this PR, Cortex components like ingester, ruler, ring or lifecycler are converted to such services.
During startup Cortex collects all the services from modules that are supposed to run (based on target module), and then starts them in parallel. Module can either contribute a service directly, or use so-called "wrapped" service – wrapped services will make sure that services are still started and stopped in a correct order based on module dependencies.

**Important change** in module initialization code is that services are returned in "New" state – they are not supposed to start any background tasks until started. That means that initialization is faster.

Main Cortex Run method now simply starts all services, and then waits until "Server" service stops or any service fails, and then initiates shutdown. "Server" service is special because it reacts on SIGTERM signal (this is how Cortex always worked). It also runs HTTP and gRPC servers.

Not all benefits of services model are implemented, but some are: for example Blocks-based StoreQueryable now fetches indexes in the background, while other services start (most importantly "Server" itself with HTTP server, for collecting metrics). Until StoreQueryable is done, Cortex doesn't transition to "Started" phase, but since HTTP server already runs, we can collect its metrics.

WAL replays in ingester are not done in background in this PR... while it would be trivial to do this change, ingester would need more changes to check if it's running already or not, and it seemed like out of scope for this PR.

Another possible area for cleanups is Lifecycler -- now it has a way to return errors, instead of just doing `os.Exit`. We should use that.

Please familiarize yourself with [services concept](https://github.com/google/guava/wiki/ServiceExplained), [Go implementation (Service interface, BasicService implementation and ServiceManager](https://github.com/pstibrany/services). This PR touches many places, but "conversion" to services was pretty straightforward.

This is an alternative to PR #2119.


pstibrany: Thank you for your review Sandeep!

> While the `services` repo looks solid I was thinking while we are at it maybe we should offload dependency management to the new package like starting and stopping the dependencies in order. Cortex or any other project using `services` repo would register a bunch of services as ordered dependencies with each service and that code should take care of it. What do you think?

In general I'd like to keep the package as small as possible and close to Guava, to ease adoption for people who know it already. Especially Service interface, service lifecycle, ServicesManager and listeners are basically set to stone. But this can be a useful extension to include, if implemented well.

In Cortex (with this PR), modules that don't share dependencies already start concurrently, so basically modules on each line start at the same time (assuming they all would start):

- Server, RuntimeConfig and MemberlistKV
- Ring, Overrides, TableManager, Configs, AlertManager, Compactor
- Distributor, Store, QueryFrontend
- StoreQueryable
- Ingester, Ruler

This may be tricky to express nicely in a general way, to be included in general package, but perhaps.

pstibrany: > Good job @pstibrany! I can't say this PR is easy to review: it's pretty large and unfortunately a good candidate for rebasing pain in the future and took me about 3h to review. As a retrospective, we could have tried to work on a smaller initial step (ie. progressively adding `Service` to components first). Let's see if we can get this merged in a reasonable time.

I have tried that. And this is the result :-( I realize it's not small at all, but doing it for single module only didn't make much sense to me.

I will split this PR into two, one to introduce services package to Cortex, and one with the rest of the work.

> Question: what's about the /ready probe? Is it mandatory to have one configured for each Cortex component once this PR will be merged, otherwise we may have some endpoints hit ahead of time (ie. querier)?

Querier already has /ready probe, but it doesn't react on services status, it just always returns 200. This is definitely something to add! I would probably go as far as checking the state of entire Cortex service manager, and report Ready only if all services are Running. In case of microservices, it would reflect the main service, in case of single binary, it's the state of entire Cortex binary. What do you think?
pracucci: > Querier already has /ready probe, but it doesn't react on services status, it just always returns 200. This is definitely something to add! I would probably go as far as checking the state of entire Cortex service manager, and report Ready only if all services are Running. In case of microservices, it would reflect the main service, in case of single binary, it's the state of entire Cortex binary. What do you think?

I don't see any other way to do it. In single binary mode you can't have a per-internal-service readiness probe, so `/ready` should return 200 only after all services are ready.
pstibrany: I think this is now ready for another review.

Here is list of follow-up work when this PR is merged

- /services should be nicer, and documented in docs/apis.md
- /services should include sub-services in the output, see status.go (I'm thinking via new interface which would give access to sub-services in a generic way, but services then need to make sure they return correct results in race-free way)
- `querier.New` should implement a service, so that other components can wait for it to be ready. Also `storeQueryable` should be a service, and querier should then reflect on it.
- ruler should wait until querier is running, before doing anything
- Ingester: move more stuff into Starting (WAL replays). Requires that Ingester checks its running state on pushes and queries.
- Ingester v2: move opening of TSDBs into Starting
- Ingester WAL - convert to service, manage via subservices


pracucci: Fixes #784
pstibrany: > * mention configuring /ready probe for ALL services

Done

> * mention breaking changes when vendoring

Done

> * check querier changes in current PR and see if we return correct errors

I've modified code so that it returns 500 instead of 422 (indicates client error) from API when server is not yet ready:

```
$ curl -v http://localhost:8000/api/prom/api/v1/query?query=up 
> GET /api/prom/api/v1/query?query=up HTTP/1.1
> Host: localhost:8000
> User-Agent: curl/7.64.1
> Accept: */*
> 
< HTTP/1.1 500 Internal Server Error
< Content-Type: application/json
< Date: Fri, 28 Feb 2020 18:25:15 GMT
< Content-Length: 91
< 

{"status":"error","errorType":"internal","error":"BlockQueryable is not running: Starting"}
```

We could also wrap API handler with a check if server is ready and return 503 if it isn't. We may want to do that in a separate PR.

> 
> Could you also rebase `master`, please? I would like to see the new querier integration tests running too.

Done.

Thanks for your review!

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 62fa5ec7a9..d31881e8ce 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,6 +27,8 @@
 * [CHANGE] Experimental TSDB: TSDB head compaction interval and concurrency is now configurable (defaults to 1 min interval and 5 concurrent head compactions). New options: `-experimental.tsdb.head-compaction-interval` and `-experimental.tsdb.head-compaction-concurrency`. #2172
 * [CHANGE] Remove fluentd-based billing infrastructure and flags such as `-distributor.enable-billing`. #1491
 * [CHANGE] Experimental TSDB: the querier in-memory index cache used by the experimental blocks storage shifted from per-tenant to per-querier. The `-experimental.tsdb.bucket-store.index-cache-size-bytes` now configures the per-querier index cache max size instead of a per-tenant cache and its default has been increased to 1GB. #2189
+* [CHANGE] If you are vendoring Cortex and use its components in your project, be aware that many Cortex components no longer start automatically when they are created. You may want to review PR and attached document. #2166
+* [CHANGE] Cortex now has /ready probe for all services, not just ingester and querier as before. In single-binary mode, /ready reports 204 only if all components are running properly. #2166
 * [FEATURE] Added a read-only local alertmanager config store using files named corresponding to their tenant id. #2125
 * [FEATURE] Added user sub rings to distribute users to a subset of ingesters. #1947
   * `--experimental.distributor.user-subring-size`
diff --git a/cmd/cortex/main.go b/cmd/cortex/main.go
index 24e53d8759..6153cb023d 100644
--- a/cmd/cortex/main.go
+++ b/cmd/cortex/main.go
@@ -110,8 +110,6 @@ func main() {
 	}
 
 	runtime.KeepAlive(ballast)
-	err = t.Stop()
-	util.CheckFatal("initializing cortex", err)
 }
 
 // Parse -config.file and -config.expand-env option via separate flag set, to avoid polluting default one and calling flag.Parse on it twice.
diff --git a/pkg/alertmanager/multitenant.go b/pkg/alertmanager/multitenant.go
index 3e2d3aecd9..535594733f 100644
--- a/pkg/alertmanager/multitenant.go
+++ b/pkg/alertmanager/multitenant.go
@@ -24,6 +24,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/alertmanager/alerts"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 var backoffConfig = util.BackoffConfig{
@@ -128,6 +129,8 @@ func (cfg *MultitenantAlertmanagerConfig) RegisterFlags(f *flag.FlagSet) {
 // A MultitenantAlertmanager manages Alertmanager instances for multiple
 // organizations.
 type MultitenantAlertmanager struct {
+	services.Service
+
 	cfg *MultitenantAlertmanagerConfig
 
 	store AlertStore
@@ -147,9 +150,6 @@ type MultitenantAlertmanager struct {
 	metrics *alertmanagerMetrics
 
 	peer *cluster.Peer
-
-	stop chan struct{}
-	done chan struct{}
 }
 
 // NewMultitenantAlertmanager creates a new MultitenantAlertmanager.
@@ -201,6 +201,10 @@ func NewMultitenantAlertmanager(cfg *MultitenantAlertmanagerConfig, logger log.L
 		return nil, err
 	}
 
+	return createMultitenantAlertmanager(cfg, fallbackConfig, peer, store, logger, registerer), nil
+}
+
+func createMultitenantAlertmanager(cfg *MultitenantAlertmanagerConfig, fallbackConfig []byte, peer *cluster.Peer, store AlertStore, logger log.Logger, registerer prometheus.Registerer) *MultitenantAlertmanager {
 	am := &MultitenantAlertmanager{
 		cfg:            cfg,
 		fallbackConfig: string(fallbackConfig),
@@ -210,42 +214,34 @@ func NewMultitenantAlertmanager(cfg *MultitenantAlertmanagerConfig, logger log.L
 		peer:           peer,
 		store:          store,
 		logger:         log.With(logger, "component", "MultiTenantAlertmanager"),
-		stop:           make(chan struct{}),
-		done:           make(chan struct{}),
 	}
 
 	if registerer != nil {
 		registerer.MustRegister(am.metrics)
 	}
 
-	return am, nil
+	am.Service = services.NewTimerService(am.cfg.PollInterval, am.starting, am.iteration, am.stopping)
+	return am
 }
 
-// Run the MultitenantAlertmanager.
-func (am *MultitenantAlertmanager) Run() {
-	defer close(am.done)
-
+func (am *MultitenantAlertmanager) starting(ctx context.Context) error {
 	// Load initial set of all configurations before polling for new ones.
 	am.syncConfigs(am.loadAllConfigs())
-	ticker := time.NewTicker(am.cfg.PollInterval)
-	for {
-		select {
-		case <-ticker.C:
-			err := am.updateConfigs()
-			if err != nil {
-				level.Warn(am.logger).Log("msg", "error updating configs", "err", err)
-			}
-		case <-am.stop:
-			ticker.Stop()
-			return
-		}
+	return nil
+}
+
+func (am *MultitenantAlertmanager) iteration(ctx context.Context) error {
+	err := am.updateConfigs()
+	if err != nil {
+		level.Warn(am.logger).Log("msg", "error updating configs", "err", err)
 	}
+	// Returning error here would stop "MultitenantAlertmanager" service completely,
+	// so we return nil to keep service running.
+	return nil
 }
 
-// Stop stops the MultitenantAlertmanager.
-func (am *MultitenantAlertmanager) Stop() {
-	close(am.stop)
-	<-am.done
+// stopping runs when MultitenantAlertmanager transitions to Stopping state.
+func (am *MultitenantAlertmanager) stopping() error {
 	am.alertmanagersMtx.Lock()
 	for _, am := range am.alertmanagers {
 		am.Stop()
@@ -256,6 +252,7 @@ func (am *MultitenantAlertmanager) Stop() {
 		level.Warn(am.logger).Log("msg", "failed to leave the cluster", "err", err)
 	}
 	level.Debug(am.logger).Log("msg", "stopping")
+	return nil
 }
 
 // Load the full set of configurations from the alert store, retrying with backoff
diff --git a/pkg/chunk/purger/purger.go b/pkg/chunk/purger/purger.go
index 838bd13ab9..f63dfc2a6a 100644
--- a/pkg/chunk/purger/purger.go
+++ b/pkg/chunk/purger/purger.go
@@ -19,6 +19,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/chunk"
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 const millisecondPerDay = int64(24 * time.Hour / time.Millisecond)
@@ -51,6 +52,8 @@ type workerJob struct {
 
 // DataPurger does the purging of data which is requested to be deleted
 type DataPurger struct {
+	services.Service
+
 	cfg          Config
 	deleteStore  *chunk.DeleteStore
 	chunkStore   chunk.Store
@@ -67,8 +70,7 @@ type DataPurger struct {
 	pendingPlansCount    map[string]int // per request pending plan count
 	pendingPlansCountMtx sync.Mutex
 
-	quit chan struct{}
-	wg   sync.WaitGroup
+	wg sync.WaitGroup
 }
 
 // NewDataPurger creates a new DataPurger
@@ -82,45 +84,39 @@ func NewDataPurger(cfg Config, deleteStore *chunk.DeleteStore, chunkStore chunk.
 		workerJobChan:       make(chan workerJob, 50),
 		inProcessRequestIDs: map[string]string{},
 		pendingPlansCount:   map[string]int{},
-		quit:                make(chan struct{}),
 	}
 
+	dataPurger.Service = services.NewTimerService(time.Hour, dataPurger.init, dataPurger.runOneIteration, dataPurger.stop)
 	return &dataPurger, nil
 }
 
 // Run keeps pulling delete requests for planning after initializing necessary things
-func (dp *DataPurger) Run() {
-	dp.wg.Add(1)
-	defer dp.wg.Done()
-
-	pullDeleteRequestsToPlanDeletesTicker := time.NewTicker(time.Hour)
-	defer pullDeleteRequestsToPlanDeletesTicker.Stop()
-
-	for {
-		select {
-		case <-pullDeleteRequestsToPlanDeletesTicker.C:
-			err := dp.pullDeleteRequestsToPlanDeletes()
-			if err != nil {
-				level.Error(util.Logger).Log("msg", "error pulling delete requests for building plans", "err", err)
-			}
-		case <-dp.quit:
-			return
-		}
+func (dp *DataPurger) runOneIteration(ctx context.Context) error {
+	err := dp.pullDeleteRequestsToPlanDeletes()
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "error pulling delete requests for building plans", "err", err)
 	}
+	// Don't return error here, or Timer service will stop.
+	return nil
 }
 
-// Init starts workers, scheduler and then loads in process delete requests
-func (dp *DataPurger) Init() error {
-	dp.runWorkers()
-	go dp.jobScheduler()
+// init starts workers, scheduler and then loads in process delete requests
+func (dp *DataPurger) init(ctx context.Context) error {
+	for i := 0; i < dp.cfg.NumWorkers; i++ {
+		dp.wg.Add(1)
+		go dp.worker()
+	}
+
+	dp.wg.Add(1)
+	go dp.jobScheduler(ctx)
 
 	return dp.loadInprocessDeleteRequests()
 }
 
-// Stop stops all background workers/loops
-func (dp *DataPurger) Stop() {
-	close(dp.quit)
+// Stop waits until all background tasks stop.
+func (dp *DataPurger) stop() error {
 	dp.wg.Wait()
+	return nil
 }
 
 func (dp *DataPurger) workerJobCleanup(job workerJob) {
@@ -154,8 +150,7 @@ func (dp *DataPurger) workerJobCleanup(job workerJob) {
 }
 
 // we send all the delete plans to workerJobChan
-func (dp *DataPurger) jobScheduler() {
-	dp.wg.Add(1)
+func (dp *DataPurger) jobScheduler(ctx context.Context) {
 	defer dp.wg.Done()
 
 	for {
@@ -172,20 +167,13 @@ func (dp *DataPurger) jobScheduler() {
 				dp.workerJobChan <- workerJob{planNo: i, userID: req.UserID,
 					deleteRequestID: req.RequestID, logger: req.logger}
 			}
-		case <-dp.quit:
+		case <-ctx.Done():
 			close(dp.workerJobChan)
 			return
 		}
 	}
 }
 
-func (dp *DataPurger) runWorkers() {
-	for i := 0; i < dp.cfg.NumWorkers; i++ {
-		dp.wg.Add(1)
-		go dp.worker()
-	}
-}
-
 func (dp *DataPurger) worker() {
 	defer dp.wg.Done()
 
diff --git a/pkg/chunk/table_manager.go b/pkg/chunk/table_manager.go
index 1be5fd1c20..362581ee4c 100644
--- a/pkg/chunk/table_manager.go
+++ b/pkg/chunk/table_manager.go
@@ -8,7 +8,6 @@ import (
 	"math/rand"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/go-kit/kit/log/level"
@@ -18,6 +17,7 @@ import (
 	"github.com/weaveworks/common/mtime"
 
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 const (
@@ -116,13 +116,15 @@ func (cfg *ProvisionConfig) RegisterFlags(argPrefix string, f *flag.FlagSet) {
 
 // TableManager creates and manages the provisioned throughput on DynamoDB tables
 type TableManager struct {
+	services.Service
+
 	client       TableClient
 	cfg          TableManagerConfig
 	schemaCfg    SchemaConfig
 	maxChunkAge  time.Duration
-	done         chan struct{}
-	wait         sync.WaitGroup
 	bucketClient BucketClient
+
+	bucketRetentionLoop services.Service
 }
 
 // NewTableManager makes a new TableManager
@@ -137,36 +139,36 @@ func NewTableManager(cfg TableManagerConfig, schemaCfg SchemaConfig, maxChunkAge
 		}
 	}
 
-	return &TableManager{
+	tm := &TableManager{
 		cfg:          cfg,
 		schemaCfg:    schemaCfg,
 		maxChunkAge:  maxChunkAge,
 		client:       tableClient,
-		done:         make(chan struct{}),
 		bucketClient: objectClient,
-	}, nil
+	}
+
+	tm.Service = services.NewBasicService(tm.starting, tm.loop, tm.stopping)
+	return tm, nil
 }
 
 // Start the TableManager
-func (m *TableManager) Start() {
-	m.wait.Add(1)
-	go m.loop()
-
+func (m *TableManager) starting(ctx context.Context) error {
 	if m.bucketClient != nil && m.cfg.RetentionPeriod != 0 && m.cfg.RetentionDeletesEnabled {
-		m.wait.Add(1)
-		go m.bucketRetentionLoop()
+		m.bucketRetentionLoop = services.NewTimerService(bucketRetentionEnforcementInterval, nil, m.bucketRetentionIteration, nil)
+		return services.StartAndAwaitRunning(ctx, m.bucketRetentionLoop)
 	}
+	return nil
 }
 
 // Stop the TableManager
-func (m *TableManager) Stop() {
-	close(m.done)
-	m.wait.Wait()
+func (m *TableManager) stopping() error {
+	if m.bucketRetentionLoop != nil {
+		return services.StopAndAwaitTerminated(context.Background(), m.bucketRetentionLoop)
+	}
+	return nil
 }
 
-func (m *TableManager) loop() {
-	defer m.wait.Done()
-
+func (m *TableManager) loop(ctx context.Context) error {
 	ticker := time.NewTicker(m.cfg.DynamoDBPollInterval)
 	defer ticker.Stop()
 
@@ -179,8 +181,8 @@ func (m *TableManager) loop() {
 	// Sleep for a bit to spread the sync load across different times if the tablemanagers are all started at once.
 	select {
 	case <-time.After(time.Duration(rand.Int63n(int64(m.cfg.DynamoDBPollInterval)))):
-	case <-m.done:
-		return
+	case <-ctx.Done():
+		return nil
 	}
 
 	for {
@@ -191,30 +193,22 @@ func (m *TableManager) loop() {
 			}); err != nil {
 				level.Error(util.Logger).Log("msg", "error syncing tables", "err", err)
 			}
-		case <-m.done:
-			return
+		case <-ctx.Done():
+			return nil
 		}
 	}
 }
 
-func (m *TableManager) bucketRetentionLoop() {
-	defer m.wait.Done()
-
-	ticker := time.NewTicker(bucketRetentionEnforcementInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			err := m.bucketClient.DeleteChunksBefore(context.Background(), mtime.Now().Add(-m.cfg.RetentionPeriod))
+// single iteration of bucket retention loop
+func (m *TableManager) bucketRetentionIteration(ctx context.Context) error {
+	err := m.bucketClient.DeleteChunksBefore(ctx, mtime.Now().Add(-m.cfg.RetentionPeriod))
 
-			if err != nil {
-				level.Error(util.Logger).Log("msg", "error enforcing filesystem retention", "err", err)
-			}
-		case <-m.done:
-			return
-		}
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "error enforcing filesystem retention", "err", err)
 	}
+
+	// don't return error, otherwise timer service would stop.
+	return nil
 }
 
 // SyncTables will calculate the tables expected to exist, create those that do
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index 278b41f7ab..a4805bd9c7 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -7,7 +7,6 @@ import (
 	"hash/fnv"
 	"path"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/go-kit/kit/log"
@@ -23,6 +22,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 // Config holds the Compactor config.
@@ -66,28 +66,29 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 
 // Compactor is a multi-tenant TSDB blocks compactor based on Thanos.
 type Compactor struct {
+	services.Service
+
 	compactorCfg Config
 	storageCfg   cortex_tsdb.Config
 	logger       log.Logger
 
+	// function that creates bucket client and TSDB compactor using the context.
+	// Useful for injecting mock objects from tests.
+	createBucketClientAndTsdbCompactor func(ctx context.Context) (objstore.Bucket, tsdb.Compactor, error)
+
 	// Underlying compactor used to compact TSDB blocks.
 	tsdbCompactor tsdb.Compactor
 
 	// Client used to run operations on the bucket storing blocks.
 	bucketClient objstore.Bucket
 
-	// Wait group used to wait until the internal go routine completes.
-	runner sync.WaitGroup
-
-	// Context used to run compaction and its cancel function to
-	// safely interrupt it on shutdown.
-	ctx       context.Context
-	cancelCtx context.CancelFunc
-
 	// Ring used for sharding compactions.
 	ringLifecycler *ring.Lifecycler
 	ring           *ring.Ring
 
+	// Subservices manager (ring, lifecycler)
+	subservices *services.Manager
+
 	// Metrics.
 	compactionRunsStarted   prometheus.Counter
 	compactionRunsCompleted prometheus.Counter
@@ -99,27 +100,22 @@ type Compactor struct {
 
 // NewCompactor makes a new Compactor.
 func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.Config, logger log.Logger, registerer prometheus.Registerer) (*Compactor, error) {
-	ctx, cancelCtx := context.WithCancel(context.Background())
-
-	bucketClient, err := cortex_tsdb.NewBucketClient(ctx, storageCfg, "compactor", logger)
-	if err != nil {
-		cancelCtx()
-		return nil, errors.Wrap(err, "failed to create the bucket client")
-	}
+	createBucketClientAndTsdbCompactor := func(ctx context.Context) (objstore.Bucket, tsdb.Compactor, error) {
+		bucketClient, err := cortex_tsdb.NewBucketClient(ctx, storageCfg, "compactor", logger)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to create the bucket client")
+		}
 
-	if registerer != nil {
-		bucketClient = objstore.BucketWithMetrics( /* bucket label value */ "", bucketClient, prometheus.WrapRegistererWithPrefix("cortex_compactor_", registerer))
-	}
+		if registerer != nil {
+			bucketClient = objstore.BucketWithMetrics( /* bucket label value */ "", bucketClient, prometheus.WrapRegistererWithPrefix("cortex_compactor_", registerer))
+		}
 
-	tsdbCompactor, err := tsdb.NewLeveledCompactor(ctx, registerer, logger, compactorCfg.BlockRanges.ToMilliseconds(), downsample.NewPool())
-	if err != nil {
-		cancelCtx()
-		return nil, errors.Wrap(err, "failed to create TSDB compactor")
+		compactor, err := tsdb.NewLeveledCompactor(ctx, registerer, logger, compactorCfg.BlockRanges.ToMilliseconds(), downsample.NewPool())
+		return bucketClient, compactor, err
 	}
 
-	cortexCompactor, err := newCompactor(ctx, cancelCtx, compactorCfg, storageCfg, bucketClient, tsdbCompactor, logger, registerer)
+	cortexCompactor, err := newCompactor(compactorCfg, storageCfg, logger, registerer, createBucketClientAndTsdbCompactor)
 	if err != nil {
-		cancelCtx()
 		return nil, errors.Wrap(err, "failed to create Cortex blocks compactor")
 	}
 
@@ -127,23 +123,19 @@ func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.Config, logger log
 }
 
 func newCompactor(
-	ctx context.Context,
-	cancelCtx context.CancelFunc,
 	compactorCfg Config,
 	storageCfg cortex_tsdb.Config,
-	bucketClient objstore.Bucket,
-	tsdbCompactor tsdb.Compactor,
 	logger log.Logger,
 	registerer prometheus.Registerer,
+	createBucketClientAndTsdbCompactor func(ctx context.Context) (objstore.Bucket, tsdb.Compactor, error),
 ) (*Compactor, error) {
 	c := &Compactor{
-		compactorCfg:  compactorCfg,
-		storageCfg:    storageCfg,
-		logger:        logger,
-		bucketClient:  bucketClient,
-		tsdbCompactor: tsdbCompactor,
-		ctx:           ctx,
-		cancelCtx:     cancelCtx,
+		compactorCfg:                       compactorCfg,
+		storageCfg:                         storageCfg,
+		logger:                             logger,
+		syncerMetrics:                      newSyncerMetrics(registerer),
+		createBucketClientAndTsdbCompactor: createBucketClientAndTsdbCompactor,
+
 		compactionRunsStarted: prometheus.NewCounter(prometheus.CounterOpts{
 			Name: "cortex_compactor_runs_started_total",
 			Help: "Total number of compaction runs started.",
@@ -158,72 +150,74 @@ func newCompactor(
 		}),
 	}
 
+	// Register metrics.
+	if registerer != nil {
+		registerer.MustRegister(c.compactionRunsStarted, c.compactionRunsCompleted, c.compactionRunsFailed)
+	}
+
+	c.Service = services.NewBasicService(c.starting, c.running, c.stopping)
+
+	return c, nil
+}
+
+// Start the compactor.
+func (c *Compactor) starting(ctx context.Context) error {
 	// Initialize the compactors ring if sharding is enabled.
-	if compactorCfg.ShardingEnabled {
-		lifecyclerCfg := compactorCfg.ShardingRing.ToLifecyclerConfig()
+	if c.compactorCfg.ShardingEnabled {
+		lifecyclerCfg := c.compactorCfg.ShardingRing.ToLifecyclerConfig()
 		lifecycler, err := ring.NewLifecycler(lifecyclerCfg, ring.NewNoopFlushTransferer(), "compactor", ring.CompactorRingKey, false)
 		if err != nil {
-			return nil, errors.Wrap(err, "unable to initialize compactor ring lifecycler")
+			return errors.Wrap(err, "unable to initialize compactor ring lifecycler")
 		}
 
-		lifecycler.Start()
 		c.ringLifecycler = lifecycler
 
 		ring, err := ring.New(lifecyclerCfg.RingConfig, "compactor", ring.CompactorRingKey)
 		if err != nil {
-			return nil, errors.Wrap(err, "unable to initialize compactor ring")
+			return errors.Wrap(err, "unable to initialize compactor ring")
 		}
 
 		c.ring = ring
-	}
 
-	// Register metrics.
-	if registerer != nil {
-		registerer.MustRegister(c.compactionRunsStarted, c.compactionRunsCompleted, c.compactionRunsFailed)
-		c.syncerMetrics = newSyncerMetrics(registerer)
+		c.subservices, err = services.NewManager(c.ringLifecycler, c.ring)
+		if err == nil {
+			err = services.StartManagerAndAwaitHealthy(ctx, c.subservices)
+		}
+
+		if err != nil {
+			return errors.Wrap(err, "unable to start compactor dependencies")
+		}
 	}
 
-	return c, nil
-}
+	var err error
+	c.bucketClient, c.tsdbCompactor, err = c.createBucketClientAndTsdbCompactor(ctx)
+	if err != nil && c.subservices != nil {
+		c.subservices.StopAsync()
+	}
 
-// Start the compactor.
-func (c *Compactor) Start() {
-	// Start the compactor loop.
-	c.runner.Add(1)
-	go c.run()
+	return errors.Wrap(err, "failed to initialize compactor objects")
 }
 
-// Stop the compactor and waits until done. This may take some time
-// if there's a on-going compaction.
-func (c *Compactor) Stop() {
-	c.cancelCtx()
-	c.runner.Wait()
-
-	// Shutdown the ring lifecycler (if any)
-	if c.ringLifecycler != nil {
-		c.ringLifecycler.Shutdown()
-	}
-
-	if c.ring != nil {
-		c.ring.Stop()
+func (c *Compactor) stopping() error {
+	if c.subservices != nil {
+		return services.StopManagerAndAwaitStopped(context.Background(), c.subservices)
 	}
+	return nil
 }
 
-func (c *Compactor) run() {
-	defer c.runner.Done()
-
+func (c *Compactor) running(ctx context.Context) error {
 	// If sharding is enabled we should wait until this instance is
 	// ACTIVE within the ring.
 	if c.compactorCfg.ShardingEnabled {
 		level.Info(c.logger).Log("msg", "waiting until compactor is ACTIVE in the ring")
-		if err := c.waitRingActive(); err != nil {
-			return
+		if err := c.waitRingActive(ctx); err != nil {
+			return err
 		}
 		level.Info(c.logger).Log("msg", "compactor is ACTIVE in the ring")
 	}
 
 	// Run an initial compaction before starting the interval.
-	c.compactUsersWithRetries(c.ctx)
+	c.compactUsersWithRetries(ctx)
 
 	ticker := time.NewTicker(c.compactorCfg.CompactionInterval)
 	defer ticker.Stop()
@@ -231,9 +225,9 @@ func (c *Compactor) run() {
 	for {
 		select {
 		case <-ticker.C:
-			c.compactUsersWithRetries(c.ctx)
-		case <-c.ctx.Done():
-			return
+			c.compactUsersWithRetries(ctx)
+		case <-ctx.Done():
+			return nil
 		}
 	}
 }
@@ -383,7 +377,7 @@ func (c *Compactor) ownUser(userID string) (bool, error) {
 	return rs.Ingesters[0].Addr == c.ringLifecycler.Addr, nil
 }
 
-func (c *Compactor) waitRingActive() error {
+func (c *Compactor) waitRingActive(ctx context.Context) error {
 	for {
 		// Check if the ingester is ACTIVE in the ring and our ring client
 		// has detected it.
@@ -398,8 +392,8 @@ func (c *Compactor) waitRingActive() error {
 		select {
 		case <-time.After(time.Second):
 			// Nothing to do
-		case <-c.ctx.Done():
-			return c.ctx.Err()
+		case <-ctx.Done():
+			return ctx.Err()
 		}
 	}
 }
diff --git a/pkg/compactor/compactor_http.go b/pkg/compactor/compactor_http.go
index a61295a93e..2c0608ea05 100644
--- a/pkg/compactor/compactor_http.go
+++ b/pkg/compactor/compactor_http.go
@@ -1,13 +1,17 @@
 package compactor
 
 import (
+	"html/template"
 	"net/http"
 
 	"github.com/go-kit/kit/log/level"
+
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
-const (
-	shardingDisabledPage = `
+var (
+	compactorStatusPageTemplate = template.Must(template.New("main").Parse(`
 	<!DOCTYPE html>
 	<html>
 		<head>
@@ -16,20 +20,34 @@ const (
 		</head>
 		<body>
 			<h1>Cortex Compactor Ring</h1>
-			<p>Compactor has no ring because sharding is disabled.</p>
+			<p>{{ .Message }}</p>
 		</body>
-	</html>
-	`
+	</html>`))
 )
 
+func writeMessage(w http.ResponseWriter, message string) {
+	w.WriteHeader(http.StatusOK)
+	err := compactorStatusPageTemplate.Execute(w, struct {
+		Message string
+	}{Message: message})
+
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "unable to serve compactor ring page", "err", err)
+	}
+}
+
 func (c *Compactor) RingHandler(w http.ResponseWriter, req *http.Request) {
-	if c.compactorCfg.ShardingEnabled {
-		c.ring.ServeHTTP(w, req)
+	if !c.compactorCfg.ShardingEnabled {
+		writeMessage(w, "Compactor has no ring because sharding is disabled.")
 		return
 	}
 
-	w.WriteHeader(http.StatusOK)
-	if _, err := w.Write([]byte(shardingDisabledPage)); err != nil {
-		level.Error(c.logger).Log("msg", "unable to serve compactor ring page", "err", err)
+	if c.State() != services.Running {
+		// we cannot read the ring before Compactor is in Running state,
+		// because that would lead to race condition.
+		writeMessage(w, "Compactor is not running yet.")
+		return
 	}
+
+	c.ring.ServeHTTP(w, req)
 }
diff --git a/pkg/cortex/cortex.go b/pkg/cortex/cortex.go
index 8abac8c6a1..277e8af744 100644
--- a/pkg/cortex/cortex.go
+++ b/pkg/cortex/cortex.go
@@ -1,8 +1,11 @@
 package cortex
 
 import (
+	"bytes"
+	"context"
 	"flag"
 	"fmt"
+	"net/http"
 	"os"
 
 	"github.com/go-kit/kit/log"
@@ -36,6 +39,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/runtimeconfig"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
@@ -163,6 +167,9 @@ type Cortex struct {
 	target             moduleName
 	httpAuthMiddleware middleware.Interface
 
+	// set during initialization
+	serviceMap map[moduleName]services.Service
+
 	server        *server.Server
 	ring          *ring.Ring
 	overrides     *validation.Overrides
@@ -203,10 +210,13 @@ func New(cfg Config) (*Cortex, error) {
 
 	cortex.setupAuthMiddleware(&cfg)
 
-	if err := cortex.init(&cfg, cfg.Target); err != nil {
+	serviceMap, err := cortex.initModuleServices(&cfg, cfg.Target)
+	if err != nil {
 		return nil, err
 	}
 
+	cortex.serviceMap = serviceMap
+	cortex.server.HTTP.Handle("/services", http.HandlerFunc(cortex.servicesHandler))
 	return cortex, nil
 }
 
@@ -242,49 +252,128 @@ func (t *Cortex) setupAuthMiddleware(cfg *Config) {
 	}
 }
 
-func (t *Cortex) init(cfg *Config, m moduleName) error {
+func (t *Cortex) initModuleServices(cfg *Config, target moduleName) (map[moduleName]services.Service, error) {
+	servicesMap := map[moduleName]services.Service{}
+
 	// initialize all of our dependencies first
-	for _, dep := range orderedDeps(m) {
-		if err := t.initModule(cfg, dep); err != nil {
-			return err
+	deps := orderedDeps(target)
+	deps = append(deps, target) // lastly, initialize the requested module
+
+	for ix, n := range deps {
+		mod := modules[n]
+
+		var serv services.Service
+
+		if mod.service != nil {
+			s, err := mod.service(t, cfg)
+			if err != nil {
+				return nil, errors.Wrap(err, fmt.Sprintf("error initialising module: %s", n))
+			}
+			serv = s
+		} else if mod.wrappedService != nil {
+			s, err := mod.wrappedService(t, cfg)
+			if err != nil {
+				return nil, errors.Wrap(err, fmt.Sprintf("error initialising module: %s", n))
+			}
+			if s != nil {
+				// We pass servicesMap, which isn't yet finished. By the time service starts,
+				// it will be fully built, so there is no need for extra synchronization.
+				serv = newModuleServiceWrapper(servicesMap, n, s, mod.deps, findInverseDependencies(n, deps[ix+1:]))
+			}
 		}
-	}
-	// lastly, initialize the requested module
-	return t.initModule(cfg, m)
-}
 
-func (t *Cortex) initModule(cfg *Config, m moduleName) error {
-	level.Info(util.Logger).Log("msg", "initialising", "module", m)
-	if modules[m].init != nil {
-		if err := modules[m].init(t, cfg); err != nil {
-			return errors.Wrap(err, fmt.Sprintf("error initialising module: %s", m))
+		if serv != nil {
+			servicesMap[n] = serv
 		}
 	}
-	return nil
+
+	return servicesMap, nil
 }
 
-// Run starts Cortex running, and blocks until a signal is received.
+// Run starts Cortex running, and blocks until a Cortex stops.
 func (t *Cortex) Run() error {
-	return t.server.Run()
-}
+	// get all services, create service manager and tell it to start
+	servs := []services.Service(nil)
+	for _, s := range t.serviceMap {
+		servs = append(servs, s)
+	}
 
-// Stop gracefully stops a Cortex.
-func (t *Cortex) Stop() error {
-	t.stopModule(t.target)
-	deps := orderedDeps(t.target)
-	// iterate over our deps in reverse order and call stopModule
-	for i := len(deps) - 1; i >= 0; i-- {
-		t.stopModule(deps[i])
+	sm, err := services.NewManager(servs...)
+	if err != nil {
+		return err
 	}
-	return nil
+
+	// before starting servers, register /ready handler. It should reflect entire Cortex.
+	t.server.HTTP.Path("/ready").Handler(t.readyHandler(sm))
+
+	// Let's listen for events from this manager, and log them.
+	healthy := func() { level.Info(util.Logger).Log("msg", "Cortex started") }
+	stopped := func() { level.Info(util.Logger).Log("msg", "Cortex stopped") }
+	serviceFailed := func(service services.Service) {
+		// if any service fails, stop entire Cortex
+		sm.StopAsync()
+
+		// let's find out which module failed
+		for m, s := range t.serviceMap {
+			if s == service {
+				level.Error(util.Logger).Log("msg", "module failed", "module", m, "error", service.FailureCase())
+				return
+			}
+		}
+
+		level.Error(util.Logger).Log("msg", "module failed", "module", "unknown", "error", service.FailureCase())
+	}
+
+	sm.AddListener(services.NewManagerListener(healthy, stopped, serviceFailed))
+
+	// Currently it's the Server that reacts on signal handler,
+	// so get Server service, and wait until it gets to Stopping state.
+	// It will also be stopped via service manager if any service fails (see attached service listener)
+	// Attach listener before starting services, or we may miss the notification.
+	serverStopping := make(chan struct{})
+	t.serviceMap[Server].AddListener(services.NewListener(nil, nil, func(from services.State) {
+		close(serverStopping)
+	}, nil, nil))
+
+	// Start all services. This can really only fail if some service is already
+	// in other state than New, which should not be the case.
+	err = sm.StartAsync(context.Background())
+	if err == nil {
+		// no error starting the services, now let's just wait until Server module
+		// transitions to Stopping (after SIGTERM or when some service fails),
+		// and then initiate shutdown
+		<-serverStopping
+	}
+
+	// Stop all the services, and wait until they are all done.
+	// We don't care about this error, as it cannot really fail. `err` has error from startup, which is more important.
+	_ = services.StopManagerAndAwaitStopped(context.Background(), sm)
+	return err
 }
 
-func (t *Cortex) stopModule(m moduleName) {
-	level.Info(util.Logger).Log("msg", "stopping", "module", m)
-	if modules[m].stop != nil {
-		if err := modules[m].stop(t); err != nil {
-			level.Error(util.Logger).Log("msg", "error stopping", "module", m, "err", err)
+func (t *Cortex) readyHandler(sm *services.Manager) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if !sm.IsHealthy() {
+			msg := bytes.Buffer{}
+			msg.WriteString("Some services are not Running:\n")
+
+			byState := sm.ServicesByState()
+			for st, ls := range byState {
+				msg.WriteString(fmt.Sprintf("%v: %d\n", st, len(ls)))
+			}
+
+			http.Error(w, msg.String(), http.StatusServiceUnavailable)
+			return
+		}
+
+		// Ingester has a special check that makes sure that it was able to register into the ring,
+		// and that all other ring entries are OK too.
+		if t.ingester != nil {
+			if err := t.ingester.CheckReady(r.Context()); err != nil {
+				http.Error(w, "Ingester not ready: "+err.Error(), http.StatusServiceUnavailable)
+			}
 		}
+		w.WriteHeader(http.StatusNoContent)
 	}
 }
 
@@ -333,3 +422,19 @@ func orderedDeps(m moduleName) []moduleName {
 	}
 	return result
 }
+
+// find modules in the supplied list, that depend on mod
+func findInverseDependencies(mod moduleName, mods []moduleName) []moduleName {
+	result := []moduleName(nil)
+
+	for _, n := range mods {
+		for _, d := range modules[n].deps {
+			if d == mod {
+				result = append(result, n)
+				break
+			}
+		}
+	}
+
+	return result
+}
diff --git a/pkg/cortex/module_service_wrapper.go b/pkg/cortex/module_service_wrapper.go
new file mode 100644
index 0000000000..c3f060aab3
--- /dev/null
+++ b/pkg/cortex/module_service_wrapper.go
@@ -0,0 +1,96 @@
+package cortex
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/go-kit/kit/log/level"
+	"github.com/pkg/errors"
+
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
+)
+
+// This service wraps module service, and adds waiting for dependencies to start before starting,
+// and dependant modules to stop before stopping this module service.
+type moduleServiceWrapper struct {
+	services.Service
+
+	serviceMap map[moduleName]services.Service
+	module     moduleName
+	service    services.Service
+	startDeps  []moduleName
+	stopDeps   []moduleName
+}
+
+func newModuleServiceWrapper(serviceMap map[moduleName]services.Service, mod moduleName, modServ services.Service, startDeps []moduleName, stopDeps []moduleName) *moduleServiceWrapper {
+	w := &moduleServiceWrapper{
+		serviceMap: serviceMap,
+		module:     mod,
+		service:    modServ,
+		startDeps:  startDeps,
+		stopDeps:   stopDeps,
+	}
+
+	w.Service = services.NewBasicService(w.start, w.run, w.stop)
+	return w
+}
+
+func (w *moduleServiceWrapper) start(serviceContext context.Context) error {
+	// wait until all startDeps are running
+	for _, m := range w.startDeps {
+		s := w.serviceMap[m]
+		if s == nil {
+			continue
+		}
+
+		level.Debug(util.Logger).Log("msg", "module waiting for initialization", "module", w.module, "waiting_for", m)
+
+		err := s.AwaitRunning(serviceContext)
+		if err != nil {
+			return fmt.Errorf("failed to start %v, because it depends on module %v, which has failed: %w", w.module, m, err)
+		}
+	}
+
+	// we don't want to let service to stop until we stop all dependant services,
+	// so we use new context
+	level.Info(util.Logger).Log("msg", "initialising", "module", w.module)
+	err := w.service.StartAsync(context.Background())
+	if err != nil {
+		return errors.Wrapf(err, "error starting module: %s", w.module)
+	}
+
+	return w.service.AwaitRunning(serviceContext)
+}
+
+func (w *moduleServiceWrapper) run(serviceContext context.Context) error {
+	// wait until service stops, or context is canceled, whatever happens first.
+	// We don't care about exact error here
+	_ = w.service.AwaitTerminated(serviceContext)
+	return w.service.FailureCase()
+}
+
+func (w *moduleServiceWrapper) stop() error {
+	// wait until all stopDeps have stopped
+	for _, m := range w.stopDeps {
+		s := w.serviceMap[m]
+		if s == nil {
+			continue
+		}
+
+		// Passed context isn't canceled, so we can only get error here, if service
+		// fails. But we don't care *how* service stops, as long as it is done.
+		_ = s.AwaitTerminated(context.Background())
+	}
+
+	level.Debug(util.Logger).Log("msg", "stopping", "module", w.module)
+
+	w.service.StopAsync()
+	err := w.service.AwaitTerminated(context.Background())
+	if err != nil {
+		level.Warn(util.Logger).Log("msg", "error stopping module", "module", w.module, "err", err)
+	} else {
+		level.Info(util.Logger).Log("msg", "module stopped", "module", w.module)
+	}
+	return err
+}
diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go
index 66a7bf7cc3..578a51b83c 100644
--- a/pkg/cortex/modules.go
+++ b/pkg/cortex/modules.go
@@ -39,6 +39,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/push"
 	"github.com/cortexproject/cortex/pkg/util/runtimeconfig"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
@@ -176,13 +177,14 @@ func (m *moduleName) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	return m.Set(s)
 }
 
-func (t *Cortex) initServer(cfg *Config) error {
-	var err error
-	t.server, err = server.New(cfg.Server)
+func (t *Cortex) initServer(cfg *Config) (services.Service, error) {
+	serv, err := server.New(cfg.Server)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
+	t.server = serv
+
 	t.server.HTTP.HandleFunc("/config", func(w http.ResponseWriter, _ *http.Request) {
 		out, err := yaml.Marshal(cfg)
 		if err != nil {
@@ -197,27 +199,32 @@ func (t *Cortex) initServer(cfg *Config) error {
 		}
 	})
 
-	return nil
-}
-
-func (t *Cortex) stopServer() (err error) {
-	t.server.Shutdown()
-	return
+	servicesToWaitFor := func() []services.Service {
+		svs := []services.Service(nil)
+		for m, s := range t.serviceMap {
+			// Server should not wait for itself.
+			if m != Server {
+				svs = append(svs, s)
+			}
+		}
+		return svs
+	}
+	return NewServerService(t.server, servicesToWaitFor), nil
 }
 
-func (t *Cortex) initRing(cfg *Config) (err error) {
+func (t *Cortex) initRing(cfg *Config) (serv services.Service, err error) {
 	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.Multi.ConfigProvider = multiClientRuntimeConfigChannel(t.runtimeConfig)
 	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.MemberlistKV = t.memberlistKVState.getMemberlistKV
 	t.ring, err = ring.New(cfg.Ingester.LifecyclerConfig.RingConfig, "ingester", ring.IngesterRingKey)
 	if err != nil {
-		return
+		return nil, err
 	}
 	prometheus.MustRegister(t.ring)
 	t.server.HTTP.Handle("/ring", t.ring)
-	return
+	return t.ring, nil
 }
 
-func (t *Cortex) initRuntimeConfig(cfg *Config) (err error) {
+func (t *Cortex) initRuntimeConfig(cfg *Config) (services.Service, error) {
 	if cfg.RuntimeConfig.LoadPath == "" {
 		cfg.RuntimeConfig.LoadPath = cfg.LimitsConfig.PerTenantOverrideConfig
 		cfg.RuntimeConfig.ReloadPeriod = cfg.LimitsConfig.PerTenantOverridePeriod
@@ -227,21 +234,19 @@ func (t *Cortex) initRuntimeConfig(cfg *Config) (err error) {
 	// make sure to set default limits before we start loading configuration into memory
 	validation.SetDefaultLimitsForYAMLUnmarshalling(cfg.LimitsConfig)
 
-	t.runtimeConfig, err = runtimeconfig.NewRuntimeConfigManager(cfg.RuntimeConfig, prometheus.DefaultRegisterer)
-	return err
+	serv, err := runtimeconfig.NewRuntimeConfigManager(cfg.RuntimeConfig, prometheus.DefaultRegisterer)
+	t.runtimeConfig = serv
+	return serv, err
 }
 
-func (t *Cortex) stopRuntimeConfig() (err error) {
-	t.runtimeConfig.Stop()
-	return nil
-}
-
-func (t *Cortex) initOverrides(cfg *Config) (err error) {
+func (t *Cortex) initOverrides(cfg *Config) (serv services.Service, err error) {
 	t.overrides, err = validation.NewOverrides(cfg.LimitsConfig, tenantLimitsFromRuntimeConfig(t.runtimeConfig))
-	return err
+	// overrides don't have operational state, nor do they need to do anything more in starting/stopping phase,
+	// so there is no need to return any service.
+	return nil, err
 }
 
-func (t *Cortex) initDistributor(cfg *Config) (err error) {
+func (t *Cortex) initDistributor(cfg *Config) (serv services.Service, err error) {
 	cfg.Distributor.DistributorRing.ListenPort = cfg.Server.GRPCListenPort
 	cfg.Distributor.DistributorRing.KVStore.MemberlistKV = t.memberlistKVState.getMemberlistKV
 
@@ -258,15 +263,10 @@ func (t *Cortex) initDistributor(cfg *Config) (err error) {
 	t.server.HTTP.HandleFunc("/all_user_stats", t.distributor.AllUserStatsHandler)
 	t.server.HTTP.Handle("/api/prom/push", t.httpAuthMiddleware.Wrap(push.Handler(cfg.Distributor, t.distributor.Push)))
 	t.server.HTTP.Handle("/ha-tracker", t.distributor.Replicas)
-	return
-}
-
-func (t *Cortex) stopDistributor() (err error) {
-	t.distributor.Stop()
-	return nil
+	return t.distributor, nil
 }
 
-func (t *Cortex) initQuerier(cfg *Config) (err error) {
+func (t *Cortex) initQuerier(cfg *Config) (serv services.Service, err error) {
 	queryable, engine := querier.New(cfg.Querier, t.distributor, t.storeQueryable)
 	api := v1.NewAPI(
 		engine,
@@ -301,13 +301,16 @@ func (t *Cortex) initQuerier(cfg *Config) (err error) {
 		return
 	}
 
-	// Once the execution reaches this point, all synchronous initialization has been
-	// done and the querier is ready to serve queries, so we're just returning a 202.
-	t.server.HTTP.Path("/ready").Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.WriteHeader(http.StatusNoContent)
-	}))
+	// TODO: If queryable returned from querier.New was a service, it could actually wait for storeQueryable
+	// (if it also implemented Service) to finish starting... and return error if it's not in Running state.
+	// This requires extra work, which is out of scope for this proof-of-concept...
+	// 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
+	return services.NewIdleService(nil, func() error {
+		t.worker.Stop()
+		return nil
+	}), nil
 }
 
 // Latest Prometheus requires r.RemoteAddr to be set to addr:port, otherwise it reject the request.
@@ -324,34 +327,25 @@ func fakeRemoteAddr(handler http.Handler) http.Handler {
 	})
 }
 
-func (t *Cortex) stopQuerier() error {
-	t.worker.Stop()
-	return nil
-}
-
-func (t *Cortex) initStoreQueryable(cfg *Config) error {
+func (t *Cortex) initStoreQueryable(cfg *Config) (services.Service, error) {
 	if cfg.Storage.Engine == storage.StorageEngineChunks {
 		t.storeQueryable = querier.NewChunkStoreQueryable(cfg.Querier, t.store)
-		return nil
+		return nil, nil
 	}
 
 	if cfg.Storage.Engine == storage.StorageEngineTSDB {
 		storeQueryable, err := querier.NewBlockQueryable(cfg.TSDB, cfg.Server.LogLevel, prometheus.DefaultRegisterer)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		t.storeQueryable = storeQueryable
-		return nil
+		return storeQueryable, nil
 	}
 
-	return fmt.Errorf("unknown storage engine '%s'", cfg.Storage.Engine)
-}
-
-func (t *Cortex) stopStoreQueryable() error {
-	return nil
+	return nil, fmt.Errorf("unknown storage engine '%s'", cfg.Storage.Engine)
 }
 
-func (t *Cortex) initIngester(cfg *Config) (err error) {
+func (t *Cortex) initIngester(cfg *Config) (serv services.Service, err error) {
 	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.Multi.ConfigProvider = multiClientRuntimeConfigChannel(t.runtimeConfig)
 	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.MemberlistKV = t.memberlistKVState.getMemberlistKV
 	cfg.Ingester.LifecyclerConfig.ListenPort = &cfg.Server.GRPCListenPort
@@ -366,21 +360,15 @@ func (t *Cortex) initIngester(cfg *Config) (err error) {
 
 	client.RegisterIngesterServer(t.server.GRPC, t.ingester)
 	grpc_health_v1.RegisterHealthServer(t.server.GRPC, t.ingester)
-	t.server.HTTP.Path("/ready").Handler(http.HandlerFunc(t.ingester.ReadinessHandler))
 	t.server.HTTP.Path("/flush").Handler(http.HandlerFunc(t.ingester.FlushHandler))
 	t.server.HTTP.Path("/shutdown").Handler(http.HandlerFunc(t.ingester.ShutdownHandler))
 	t.server.HTTP.Handle("/push", t.httpAuthMiddleware.Wrap(push.Handler(cfg.Distributor, t.ingester.Push)))
-	return
-}
-
-func (t *Cortex) stopIngester() error {
-	t.ingester.Shutdown()
-	return nil
+	return t.ingester, nil
 }
 
-func (t *Cortex) initStore(cfg *Config) (err error) {
+func (t *Cortex) initStore(cfg *Config) (serv services.Service, err error) {
 	if cfg.Storage.Engine == storage.StorageEngineTSDB {
-		return nil
+		return nil, nil
 	}
 	err = cfg.Schema.Load()
 	if err != nil {
@@ -392,17 +380,13 @@ func (t *Cortex) initStore(cfg *Config) (err error) {
 		return
 	}
 
-	return
-}
-
-func (t *Cortex) stopStore() error {
-	if t.store != nil {
+	return services.NewIdleService(nil, func() error {
 		t.store.Stop()
-	}
-	return nil
+		return nil
+	}), nil
 }
 
-func (t *Cortex) initQueryFrontend(cfg *Config) (err error) {
+func (t *Cortex) initQueryFrontend(cfg *Config) (serv services.Service, err error) {
 	err = cfg.Schema.Load()
 	if err != nil {
 		return
@@ -429,7 +413,7 @@ func (t *Cortex) initQueryFrontend(cfg *Config) (err error) {
 	)
 
 	if err != nil {
-		return err
+		return nil, err
 	}
 	t.cache = cache
 	t.frontend.Wrap(tripperware)
@@ -440,26 +424,24 @@ func (t *Cortex) initQueryFrontend(cfg *Config) (err error) {
 			t.frontend.Handler(),
 		),
 	)
-	return
-}
-
-func (t *Cortex) stopQueryFrontend() (err error) {
-	t.frontend.Close()
-	if t.cache != nil {
-		t.cache.Stop()
-		t.cache = nil
-	}
-	return
+	return services.NewIdleService(nil, func() error {
+		t.frontend.Close()
+		if t.cache != nil {
+			t.cache.Stop()
+			t.cache = nil
+		}
+		return nil
+	}), nil
 }
 
-func (t *Cortex) initTableManager(cfg *Config) error {
+func (t *Cortex) initTableManager(cfg *Config) (services.Service, error) {
 	if cfg.Storage.Engine == storage.StorageEngineTSDB {
-		return nil // table manager isn't used in v2
+		return nil, nil // table manager isn't used in v2
 	}
 
 	err := cfg.Schema.Load()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// Assume the newest config is the one to use
@@ -480,36 +462,24 @@ func (t *Cortex) initTableManager(cfg *Config) error {
 
 	tableClient, err := storage.NewTableClient(lastConfig.IndexType, cfg.Storage)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	bucketClient, err := storage.NewBucketClient(cfg.Storage)
 	util.CheckFatal("initializing bucket client", err)
 
 	t.tableManager, err = chunk.NewTableManager(cfg.TableManager, cfg.Schema, cfg.Ingester.MaxChunkAge, tableClient, bucketClient)
-	if err != nil {
-		return err
-	}
-	t.tableManager.Start()
-	return nil
+	return t.tableManager, err
 }
 
-func (t *Cortex) stopTableManager() error {
-	if t.tableManager != nil {
-		t.tableManager.Stop()
-	}
-
-	return nil
-}
-
-func (t *Cortex) initRuler(cfg *Config) (err error) {
+func (t *Cortex) initRuler(cfg *Config) (serv services.Service, err error) {
 	cfg.Ruler.Ring.ListenPort = cfg.Server.GRPCListenPort
 	cfg.Ruler.Ring.KVStore.MemberlistKV = t.memberlistKVState.getMemberlistKV
 	queryable, engine := querier.New(cfg.Querier, t.distributor, t.storeQueryable)
 
 	t.ruler, err = ruler.NewRuler(cfg.Ruler, engine, queryable, t.distributor, prometheus.DefaultRegisterer, util.Logger)
 	if err != nil {
-		return err
+		return
 	}
 
 	if cfg.Ruler.EnableAPI {
@@ -519,15 +489,10 @@ func (t *Cortex) initRuler(cfg *Config) (err error) {
 	}
 
 	t.server.HTTP.Handle("/ruler_ring", t.ruler)
-	return
-}
-
-func (t *Cortex) stopRuler() error {
-	t.ruler.Stop()
-	return nil
+	return t.ruler, nil
 }
 
-func (t *Cortex) initConfigs(cfg *Config) (err error) {
+func (t *Cortex) initConfig(cfg *Config) (serv services.Service, err error) {
 	t.configDB, err = db.New(cfg.ConfigDB)
 	if err != nil {
 		return
@@ -535,112 +500,86 @@ func (t *Cortex) initConfigs(cfg *Config) (err error) {
 
 	t.configAPI = api.New(t.configDB)
 	t.configAPI.RegisterRoutes(t.server.HTTP)
-	return
-}
-
-func (t *Cortex) stopConfigs() error {
-	t.configDB.Close()
-	return nil
+	return services.NewIdleService(nil, func() error {
+		t.configDB.Close()
+		return nil
+	}), nil
 }
 
-func (t *Cortex) initAlertmanager(cfg *Config) (err error) {
+func (t *Cortex) initAlertManager(cfg *Config) (serv services.Service, err error) {
 	t.alertmanager, err = alertmanager.NewMultitenantAlertmanager(&cfg.Alertmanager, util.Logger, prometheus.DefaultRegisterer)
 	if err != nil {
-		return err
+		return
 	}
-	go t.alertmanager.Run()
-
 	t.server.HTTP.PathPrefix("/status").Handler(t.alertmanager.GetStatusHandler())
 
 	// TODO this clashed with the queirer and the distributor, so we cannot
 	// run them in the same process.
 	t.server.HTTP.PathPrefix("/api/prom").Handler(middleware.AuthenticateUser.Wrap(t.alertmanager))
-	return
-}
-
-func (t *Cortex) stopAlertmanager() error {
-	t.alertmanager.Stop()
-	return nil
+	return t.alertmanager, nil
 }
 
-func (t *Cortex) initCompactor(cfg *Config) (err error) {
+func (t *Cortex) initCompactor(cfg *Config) (serv services.Service, err error) {
 	cfg.Compactor.ShardingRing.ListenPort = cfg.Server.GRPCListenPort
 	cfg.Compactor.ShardingRing.KVStore.MemberlistKV = t.memberlistKVState.getMemberlistKV
 
 	t.compactor, err = compactor.NewCompactor(cfg.Compactor, cfg.TSDB, util.Logger, prometheus.DefaultRegisterer)
 	if err != nil {
-		return err
+		return
 	}
 
-	t.compactor.Start()
-
 	// Expose HTTP endpoints.
 	t.server.HTTP.HandleFunc("/compactor_ring", t.compactor.RingHandler)
 
-	return nil
-}
-
-func (t *Cortex) stopCompactor() error {
-	t.compactor.Stop()
-	return nil
+	return t.compactor, nil
 }
 
-func (t *Cortex) initMemberlistKV(cfg *Config) (err error) {
+func (t *Cortex) initMemberlistKV(cfg *Config) (services.Service, error) {
 	cfg.MemberlistKV.MetricsRegisterer = prometheus.DefaultRegisterer
 	cfg.MemberlistKV.Codecs = []codec.Codec{
 		ring.GetCodec(),
 	}
 	t.memberlistKVState = newMemberlistKVState(&cfg.MemberlistKV)
-	return nil
-}
 
-func (t *Cortex) stopMemberlistKV() (err error) {
-	kv := t.memberlistKVState.kv
-	if kv != nil {
-		kv.Stop()
-	}
-	return nil
+	return services.NewIdleService(nil, func() error {
+		kv := t.memberlistKVState.kv
+		if kv != nil {
+			kv.Stop()
+		}
+		return nil
+	}), nil
 }
 
-func (t *Cortex) initDataPurger(cfg *Config) (err error) {
+func (t *Cortex) initDataPurger(cfg *Config) (services.Service, error) {
 	if !cfg.DataPurgerConfig.Enable {
-		return nil
+		return nil, nil
 	}
 
 	var indexClient chunk.IndexClient
-	indexClient, err = storage.NewIndexClient(cfg.Storage.DeleteStoreConfig.Store, cfg.Storage, cfg.Schema)
+	indexClient, err := storage.NewIndexClient(cfg.Storage.DeleteStoreConfig.Store, cfg.Storage, cfg.Schema)
 	if err != nil {
-		return
+		return nil, err
 	}
 
-	var deleteStore *chunk.DeleteStore
-	deleteStore, err = chunk.NewDeleteStore(cfg.Storage.DeleteStoreConfig, indexClient)
+	deleteStore, err := chunk.NewDeleteStore(cfg.Storage.DeleteStoreConfig, indexClient)
 	if err != nil {
-		return
+		return nil, err
 	}
 
-	var storageClient chunk.ObjectClient
-	storageClient, err = storage.NewObjectClient(cfg.DataPurgerConfig.ObjectStoreType, cfg.Storage)
+	storageClient, err := storage.NewObjectClient(cfg.DataPurgerConfig.ObjectStoreType, cfg.Storage)
 	if err != nil {
-		return
+		return nil, err
 	}
 
 	t.dataPurger, err = purger.NewDataPurger(cfg.DataPurgerConfig, deleteStore, t.store, storageClient)
 	if err != nil {
-		return
+		return nil, err
 	}
 
-	err = t.dataPurger.Init()
-	if err != nil {
-		return
-	}
-
-	go t.dataPurger.Run()
-
 	var deleteRequestHandler *purger.DeleteRequestHandler
 	deleteRequestHandler, err = purger.NewDeleteRequestHandler(deleteStore)
 	if err != nil {
-		return
+		return nil, err
 	}
 
 	adminRouter := t.server.HTTP.PathPrefix(cfg.HTTPPrefix + "/api/v1/admin/tsdb").Subrouter()
@@ -648,118 +587,104 @@ func (t *Cortex) initDataPurger(cfg *Config) (err error) {
 	adminRouter.Path("/delete_series").Methods("PUT", "POST").Handler(t.httpAuthMiddleware.Wrap(http.HandlerFunc(deleteRequestHandler.AddDeleteRequestHandler)))
 	adminRouter.Path("/delete_series").Methods("GET").Handler(t.httpAuthMiddleware.Wrap(http.HandlerFunc(deleteRequestHandler.GetAllDeleteRequestsHandler)))
 
-	return
-}
-
-func (t *Cortex) stopDataPurger() error {
-	if t.dataPurger != nil {
-		t.dataPurger.Stop()
-	}
-	return nil
+	return t.dataPurger, nil
 }
 
 type module struct {
 	deps []moduleName
-	init func(t *Cortex, cfg *Config) error
-	stop func(t *Cortex) error
+
+	// service for this module (can return nil)
+	service func(t *Cortex, cfg *Config) (services.Service, error)
+
+	// service that will be wrapped into moduleServiceWrapper, to wait for dependencies to start / end
+	// (can return nil)
+	wrappedService func(t *Cortex, cfg *Config) (services.Service, error)
 }
 
 var modules = map[moduleName]module{
 	Server: {
-		init: (*Cortex).initServer,
-		stop: (*Cortex).stopServer,
+		// we cannot use 'wrappedService', as stopped Server service is currently a signal to Cortex
+		// that it should shutdown. If we used wrappedService, it wouldn't stop until
+		// all services that depend on it stopped first... but there is nothing that would make them stop.
+		service: (*Cortex).initServer,
 	},
 
 	RuntimeConfig: {
-		init: (*Cortex).initRuntimeConfig,
-		stop: (*Cortex).stopRuntimeConfig,
+		wrappedService: (*Cortex).initRuntimeConfig,
 	},
 
 	MemberlistKV: {
-		init: (*Cortex).initMemberlistKV,
-		stop: (*Cortex).stopMemberlistKV,
+		wrappedService: (*Cortex).initMemberlistKV,
 	},
 
 	Ring: {
-		deps: []moduleName{Server, RuntimeConfig, MemberlistKV},
-		init: (*Cortex).initRing,
+		deps:           []moduleName{Server, RuntimeConfig, MemberlistKV},
+		wrappedService: (*Cortex).initRing,
 	},
 
 	Overrides: {
-		deps: []moduleName{RuntimeConfig},
-		init: (*Cortex).initOverrides,
+		deps:           []moduleName{RuntimeConfig},
+		wrappedService: (*Cortex).initOverrides,
 	},
 
 	Distributor: {
-		deps: []moduleName{Ring, Server, Overrides},
-		init: (*Cortex).initDistributor,
-		stop: (*Cortex).stopDistributor,
+		deps:           []moduleName{Ring, Server, Overrides},
+		wrappedService: (*Cortex).initDistributor,
 	},
 
 	Store: {
-		deps: []moduleName{Overrides},
-		init: (*Cortex).initStore,
-		stop: (*Cortex).stopStore,
+		deps:           []moduleName{Overrides},
+		wrappedService: (*Cortex).initStore,
 	},
 
 	Ingester: {
-		deps: []moduleName{Overrides, Store, Server, RuntimeConfig, MemberlistKV},
-		init: (*Cortex).initIngester,
-		stop: (*Cortex).stopIngester,
+		deps:           []moduleName{Overrides, Store, Server, RuntimeConfig, MemberlistKV},
+		wrappedService: (*Cortex).initIngester,
 	},
 
 	Querier: {
-		deps: []moduleName{Distributor, Store, Ring, Server, StoreQueryable},
-		init: (*Cortex).initQuerier,
-		stop: (*Cortex).stopQuerier,
+		deps:           []moduleName{Distributor, Store, Ring, Server, StoreQueryable},
+		wrappedService: (*Cortex).initQuerier,
 	},
 
 	StoreQueryable: {
-		deps: []moduleName{Store},
-		init: (*Cortex).initStoreQueryable,
-		stop: (*Cortex).stopStoreQueryable,
+		deps:           []moduleName{Store},
+		wrappedService: (*Cortex).initStoreQueryable,
 	},
 
 	QueryFrontend: {
-		deps: []moduleName{Server, Overrides},
-		init: (*Cortex).initQueryFrontend,
-		stop: (*Cortex).stopQueryFrontend,
+		deps:           []moduleName{Server, Overrides},
+		wrappedService: (*Cortex).initQueryFrontend,
 	},
 
 	TableManager: {
-		deps: []moduleName{Server},
-		init: (*Cortex).initTableManager,
-		stop: (*Cortex).stopTableManager,
+		deps:           []moduleName{Server},
+		wrappedService: (*Cortex).initTableManager,
 	},
 
 	Ruler: {
-		deps: []moduleName{Distributor, Store, StoreQueryable},
-		init: (*Cortex).initRuler,
-		stop: (*Cortex).stopRuler,
+		deps:           []moduleName{Distributor, Store, StoreQueryable},
+		wrappedService: (*Cortex).initRuler,
 	},
 
 	Configs: {
-		deps: []moduleName{Server},
-		init: (*Cortex).initConfigs,
-		stop: (*Cortex).stopConfigs,
+		deps:           []moduleName{Server},
+		wrappedService: (*Cortex).initConfig,
 	},
 
 	AlertManager: {
-		deps: []moduleName{Server},
-		init: (*Cortex).initAlertmanager,
-		stop: (*Cortex).stopAlertmanager,
+		deps:           []moduleName{Server},
+		wrappedService: (*Cortex).initAlertManager,
 	},
 
 	Compactor: {
-		deps: []moduleName{Server},
-		init: (*Cortex).initCompactor,
-		stop: (*Cortex).stopCompactor,
+		deps:           []moduleName{Server},
+		wrappedService: (*Cortex).initCompactor,
 	},
 
 	DataPurger: {
-		deps: []moduleName{Store, Server},
-		init: (*Cortex).initDataPurger,
-		stop: (*Cortex).stopDataPurger,
+		deps:           []moduleName{Store, Server},
+		wrappedService: (*Cortex).initDataPurger,
 	},
 
 	All: {
diff --git a/pkg/cortex/server_service.go b/pkg/cortex/server_service.go
new file mode 100644
index 0000000000..8bac072472
--- /dev/null
+++ b/pkg/cortex/server_service.go
@@ -0,0 +1,52 @@
+package cortex
+
+import (
+	"context"
+
+	"github.com/go-kit/kit/log/level"
+	"github.com/weaveworks/common/server"
+
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
+)
+
+// NewServerService constructs service from Server component.
+// servicesToWaitFor is called when server is stopping, and should return all
+// services that need to terminate before server actually stops.
+func NewServerService(serv *server.Server, servicesToWaitFor func() []services.Service) services.Service {
+	serverDone := make(chan error, 1)
+
+	runFn := func(ctx context.Context) error {
+		go func() {
+			defer close(serverDone)
+			serverDone <- serv.Run()
+		}()
+
+		select {
+		case <-ctx.Done():
+			return nil
+		case err := <-serverDone:
+			if err != nil {
+				level.Error(util.Logger).Log("msg", "server failed", "err", err)
+			}
+			return err
+		}
+	}
+
+	stoppingFn := func() error {
+		// wait until all modules are done, and then shutdown server.
+		for _, s := range servicesToWaitFor() {
+			_ = s.AwaitTerminated(context.Background())
+		}
+
+		// shutdown HTTP and gRPC servers (this also unblocks Run)
+		serv.Shutdown()
+
+		// if not closed yet, wait until server stops.
+		<-serverDone
+		level.Info(util.Logger).Log("msg", "server stopped")
+		return nil
+	}
+
+	return services.NewBasicService(nil, runFn, stoppingFn)
+}
diff --git a/pkg/cortex/status.go b/pkg/cortex/status.go
new file mode 100644
index 0000000000..f314f462cb
--- /dev/null
+++ b/pkg/cortex/status.go
@@ -0,0 +1,19 @@
+package cortex
+
+import (
+	"fmt"
+	"net/http"
+)
+
+func (t *Cortex) servicesHandler(w http.ResponseWriter, r *http.Request) {
+	w.WriteHeader(200)
+	w.Header().Set("Content-Type", "text/plain")
+
+	// TODO: this could be extended to also print sub-services, if given service has any
+	for mod := moduleName(0); mod < All; mod++ {
+		s := t.serviceMap[mod]
+		if s != nil {
+			fmt.Fprintf(w, "%v => %v\n", mod, s.State())
+		}
+	}
+}
diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go
index a35e242b4d..e75940c09a 100644
--- a/pkg/distributor/distributor.go
+++ b/pkg/distributor/distributor.go
@@ -8,8 +8,6 @@ import (
 	"strings"
 	"time"
 
-	"google.golang.org/grpc/health/grpc_health_v1"
-
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -18,6 +16,7 @@ import (
 	"github.com/weaveworks/common/httpgrpc"
 	"github.com/weaveworks/common/instrument"
 	"github.com/weaveworks/common/user"
+	"google.golang.org/grpc/health/grpc_health_v1"
 
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	ingester_client "github.com/cortexproject/cortex/pkg/ingester/client"
@@ -27,6 +26,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/util/extract"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/limiter"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
@@ -94,6 +94,8 @@ var (
 // Distributor is a storage.SampleAppender and a client.Querier which
 // forwards appends and queries to individual ingesters.
 type Distributor struct {
+	services.Service
+
 	cfg           Config
 	ingestersRing ring.ReadRing
 	ingesterPool  *ingester_client.Pool
@@ -108,6 +110,9 @@ type Distributor struct {
 
 	// Per-user rate limiter.
 	ingestionRateLimiter *limiter.RateLimiter
+
+	// Manager for subservices (HA Tracker, distributor ring and client pool)
+	subservices *services.Manager
 }
 
 // Config contains the configuration require to
@@ -164,6 +169,9 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 		return nil, err
 	}
 
+	subservices := []services.Service(nil)
+	subservices = append(subservices, replicas)
+
 	// Create the configured ingestion rate limit strategy (local or global). In case
 	// it's an internal dependency and can't join the distributors ring, we skip rate
 	// limiting.
@@ -178,7 +186,7 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 			return nil, err
 		}
 
-		distributorsRing.Start()
+		subservices = append(subservices, distributorsRing)
 
 		ingestionRateStrategy = newGlobalIngestionRateStrategy(limits, distributorsRing)
 	} else {
@@ -195,17 +203,24 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 		Replicas:             replicas,
 	}
 
+	subservices = append(subservices, d.ingesterPool)
+	d.subservices, err = services.NewManager(subservices...)
+	if err != nil {
+		return nil, err
+	}
+
+	d.Service = services.NewIdleService(d.starting, d.stopping)
 	return d, nil
 }
 
-// Stop stops the distributor's maintenance loop.
-func (d *Distributor) Stop() {
-	d.ingesterPool.Stop()
-	d.Replicas.stop()
+func (d *Distributor) starting(ctx context.Context) error {
+	// Only report success if all sub-services start properly
+	return services.StartManagerAndAwaitHealthy(ctx, d.subservices)
+}
 
-	if d.distributorsRing != nil {
-		d.distributorsRing.Shutdown()
-	}
+// Called after distributor is asked to stop via StopAsync.
+func (d *Distributor) stopping() error {
+	return services.StopManagerAndAwaitStopped(context.Background(), d.subservices)
 }
 
 func (d *Distributor) tokenForLabels(userID string, labels []client.LabelAdapter) (uint32, error) {
diff --git a/pkg/distributor/ha_tracker.go b/pkg/distributor/ha_tracker.go
index f90ab4e0d4..363f16f8da 100644
--- a/pkg/distributor/ha_tracker.go
+++ b/pkg/distributor/ha_tracker.go
@@ -24,6 +24,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring/kv"
 	"github.com/cortexproject/cortex/pkg/ring/kv/codec"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 var (
@@ -66,6 +67,8 @@ func NewReplicaDesc() *ReplicaDesc {
 // Track the replica we're accepting samples from
 // for each HA cluster we know about.
 type haTracker struct {
+	services.Service
+
 	logger              log.Logger
 	cfg                 HATrackerConfig
 	client              kv.Client
@@ -74,8 +77,6 @@ type haTracker struct {
 	// Replicas we are accepting samples from.
 	electedLock sync.RWMutex
 	elected     map[string]ReplicaDesc
-	done        chan struct{}
-	cancel      context.CancelFunc
 }
 
 // HATrackerConfig contains the configuration require to
@@ -141,39 +142,40 @@ func GetReplicaDescCodec() codec.Proto {
 }
 
 // NewClusterTracker returns a new HA cluster tracker using either Consul
-// or in-memory KV store.
+// or in-memory KV store. Tracker must be started via StartAsync().
 func newClusterTracker(cfg HATrackerConfig) (*haTracker, error) {
-	codec := GetReplicaDescCodec()
-
 	var jitter time.Duration
 	if cfg.UpdateTimeoutJitterMax > 0 {
 		jitter = time.Duration(rand.Int63n(int64(2*cfg.UpdateTimeoutJitterMax))) - cfg.UpdateTimeoutJitterMax
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	t := haTracker{
+	t := &haTracker{
 		logger:              util.Logger,
 		cfg:                 cfg,
 		updateTimeoutJitter: jitter,
-		done:                make(chan struct{}),
 		elected:             map[string]ReplicaDesc{},
-		cancel:              cancel,
 	}
 
 	if cfg.EnableHATracker {
-		client, err := kv.NewClient(cfg.KVStore, codec)
+		client, err := kv.NewClient(cfg.KVStore, GetReplicaDescCodec())
 		if err != nil {
 			return nil, err
 		}
 		t.client = client
-		go t.loop(ctx)
 	}
-	return &t, nil
+
+	t.Service = services.NewBasicService(nil, t.loop, nil)
+	return t, nil
 }
 
 // Follows pattern used by ring for WatchKey.
-func (c *haTracker) loop(ctx context.Context) {
-	defer close(c.done)
+func (c *haTracker) loop(ctx context.Context) error {
+	if !c.cfg.EnableHATracker {
+		// don't do anything, but wait until asked to stop.
+		<-ctx.Done()
+		return nil
+	}
+
 	// The KVStore config we gave when creating c should have contained a prefix,
 	// which would have given us a prefixed KVStore client. So, we can pass empty string here.
 	c.client.WatchPrefix(ctx, "", func(key string, value interface{}) bool {
@@ -196,14 +198,8 @@ func (c *haTracker) loop(ctx context.Context) {
 		electedReplicaPropagationTime.Observe(time.Since(timestamp.Time(replica.ReceivedAt)).Seconds())
 		return true
 	})
-}
 
-// Stop ends calls the trackers cancel function, which will end the loop for WatchPrefix.
-func (c *haTracker) stop() {
-	if c.cfg.EnableHATracker {
-		c.cancel()
-		<-c.done
-	}
+	return nil
 }
 
 // CheckReplica checks the cluster and replica against the backing KVStore and local cache in the
diff --git a/pkg/ingester/client/pool.go b/pkg/ingester/client/pool.go
index e71b895e74..70d55f09d2 100644
--- a/pkg/ingester/client/pool.go
+++ b/pkg/ingester/client/pool.go
@@ -17,6 +17,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 var clients = promauto.NewGauge(prometheus.GaugeOpts{
@@ -43,14 +44,13 @@ func (cfg *PoolConfig) RegisterFlags(f *flag.FlagSet) {
 
 // Pool holds a cache of grpc_health_v1 clients.
 type Pool struct {
+	services.Service
+
 	cfg     PoolConfig
 	ring    ring.ReadRing
 	factory Factory
 	logger  log.Logger
 
-	quit chan struct{}
-	done sync.WaitGroup
-
 	sync.RWMutex
 	clients map[string]grpc_health_v1.HealthClient
 }
@@ -62,39 +62,20 @@ func NewPool(cfg PoolConfig, ring ring.ReadRing, factory Factory, logger log.Log
 		ring:    ring,
 		factory: factory,
 		logger:  logger,
-		quit:    make(chan struct{}),
 
 		clients: map[string]grpc_health_v1.HealthClient{},
 	}
 
-	p.done.Add(1)
-	go p.loop()
+	p.Service = services.NewTimerService(cfg.ClientCleanupPeriod, nil, p.iteration, nil)
 	return p
 }
 
-func (p *Pool) loop() {
-	defer p.done.Done()
-
-	cleanupClients := time.NewTicker(p.cfg.ClientCleanupPeriod)
-	defer cleanupClients.Stop()
-
-	for {
-		select {
-		case <-cleanupClients.C:
-			p.removeStaleClients()
-			if p.cfg.HealthCheckIngesters {
-				p.cleanUnhealthy()
-			}
-		case <-p.quit:
-			return
-		}
+func (p *Pool) iteration(ctx context.Context) error {
+	p.removeStaleClients()
+	if p.cfg.HealthCheckIngesters {
+		p.cleanUnhealthy()
 	}
-}
-
-// Stop the pool's background cleanup goroutine.
-func (p *Pool) Stop() {
-	close(p.quit)
-	p.done.Wait()
+	return nil
 }
 
 func (p *Pool) fromCache(addr string) (grpc_health_v1.HealthClient, bool) {
diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go
index bff7ebab16..5a234f6cd7 100644
--- a/pkg/ingester/ingester.go
+++ b/pkg/ingester/ingester.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-kit/kit/log/level"
 	"github.com/gogo/status"
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
@@ -23,6 +24,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
@@ -91,6 +93,8 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 // Ingester deals with "in flight" chunks.  Based on Prometheus 1.x
 // MemorySeriesStorage.
 type Ingester struct {
+	services.Service
+
 	cfg          Config
 	clientConfig client.Config
 
@@ -101,9 +105,6 @@ type Ingester struct {
 	limits     *validation.Overrides
 	limiter    *SeriesLimiter
 
-	quit chan struct{}
-	done sync.WaitGroup
-
 	userStatesMtx sync.RWMutex // protects userStates and stopped
 	userStates    *userStates
 	stopped       bool // protected by userStatesMtx
@@ -157,7 +158,6 @@ func New(cfg Config, clientConfig client.Config, limits *validation.Overrides, c
 		metrics:      newIngesterMetrics(registerer, true),
 		limits:       limits,
 		chunkStore:   chunkStore,
-		quit:         make(chan struct{}),
 		flushQueues:  make([]*util.PriorityQueue, cfg.ConcurrentFlushes),
 	}
 
@@ -193,24 +193,31 @@ func New(cfg Config, clientConfig client.Config, limits *validation.Overrides, c
 		return nil, err
 	}
 
-	// Now that user states have been created, we can start the lifecycler
-	i.lifecycler.Start()
+	// TODO: lot more stuff can be put into startingFn (esp. WAL replay), but for now keep it in New
+	i.Service = services.NewBasicService(i.starting, i.loop, i.stopping)
+	return i, nil
+}
+
+func (i *Ingester) starting(ctx context.Context) error {
+	// Now that user states have been created, we can start the lifecycler.
+	// Important: we want to keep lifecycler running until we ask it to stop, so we need to give it independent context
+	if err := i.lifecycler.StartAsync(context.Background()); err != nil {
+		return errors.Wrap(err, "failed to start lifecycler")
+	}
+	if err := i.lifecycler.AwaitRunning(ctx); err != nil {
+		return errors.Wrap(err, "failed to start lifecycler")
+	}
 
-	i.flushQueuesDone.Add(cfg.ConcurrentFlushes)
-	for j := 0; j < cfg.ConcurrentFlushes; j++ {
+	i.flushQueuesDone.Add(i.cfg.ConcurrentFlushes)
+	for j := 0; j < i.cfg.ConcurrentFlushes; j++ {
 		i.flushQueues[j] = util.NewPriorityQueue(i.metrics.flushQueueLength)
 		go i.flushLoop(j)
 	}
 
-	i.done.Add(1)
-	go i.loop()
-
-	return i, nil
+	return nil
 }
 
-func (i *Ingester) loop() {
-	defer i.done.Done()
-
+func (i *Ingester) loop(ctx context.Context) error {
 	flushTicker := time.NewTicker(i.cfg.FlushCheckPeriod)
 	defer flushTicker.Stop()
 
@@ -225,28 +232,18 @@ func (i *Ingester) loop() {
 		case <-rateUpdateTicker.C:
 			i.userStates.updateRates()
 
-		case <-i.quit:
-			return
+		case <-ctx.Done():
+			return nil
 		}
 	}
 }
 
-// Shutdown beings the process to stop this ingester.
-func (i *Ingester) Shutdown() {
-	select {
-	case <-i.quit:
-		// Ingester was already shutdown.
-		return
-	default:
-		// First wait for our flush loop to stop.
-		close(i.quit)
-		i.done.Wait()
-
-		i.wal.Stop()
+// stopping is run when ingester is asked to stop
+func (i *Ingester) stopping() error {
+	i.wal.Stop()
 
-		// Next initiate our graceful exit from the ring.
-		i.lifecycler.Shutdown()
-	}
+	// Next initiate our graceful exit from the ring.
+	return services.StopAndAwaitTerminated(context.Background(), i.lifecycler)
 }
 
 // ShutdownHandler triggers the following set of operations in order:
@@ -256,7 +253,7 @@ func (i *Ingester) ShutdownHandler(w http.ResponseWriter, r *http.Request) {
 	originalState := i.lifecycler.FlushOnShutdown()
 	// We want to flush the chunks if transfer fails irrespective of original flag.
 	i.lifecycler.SetFlushOnShutdown(true)
-	i.Shutdown()
+	_ = services.StopAndAwaitTerminated(context.Background(), i)
 	i.lifecycler.SetFlushOnShutdown(originalState)
 	w.WriteHeader(http.StatusNoContent)
 }
@@ -703,10 +700,9 @@ func (i *Ingester) Watch(in *grpc_health_v1.HealthCheckRequest, stream grpc_heal
 // ReadinessHandler is used to indicate to k8s when the ingesters are ready for
 // the addition removal of another ingester. Returns 204 when the ingester is
 // ready, 500 otherwise.
-func (i *Ingester) ReadinessHandler(w http.ResponseWriter, r *http.Request) {
-	if err := i.lifecycler.CheckReady(r.Context()); err == nil {
-		w.WriteHeader(http.StatusNoContent)
-	} else {
-		http.Error(w, "Not ready: "+err.Error(), http.StatusServiceUnavailable)
+func (i *Ingester) CheckReady(ctx context.Context) error {
+	if s := i.State(); s != services.Running {
+		return fmt.Errorf("service not Running: %v", s)
 	}
+	return i.lifecycler.CheckReady(ctx)
 }
diff --git a/pkg/ingester/ingester_v2.go b/pkg/ingester/ingester_v2.go
index 6b9c823e1c..3f0088efbc 100644
--- a/pkg/ingester/ingester_v2.go
+++ b/pkg/ingester/ingester_v2.go
@@ -26,6 +26,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
@@ -64,6 +65,8 @@ type TSDBState struct {
 	// transferring to a joining ingester
 	transferOnce sync.Once
 
+	subservices *services.Manager
+
 	tsdbMetrics *tsdbMetrics
 
 	// Head compactions metrics.
@@ -88,7 +91,6 @@ func NewV2(cfg Config, clientConfig client.Config, limits *validation.Overrides,
 		metrics:      newIngesterMetrics(registerer, false),
 		limits:       limits,
 		chunkStore:   nil,
-		quit:         make(chan struct{}),
 		wal:          &noopWAL{},
 		TSDBState: TSDBState{
 			dbs:         make(map[string]*userTSDB),
@@ -128,32 +130,58 @@ func NewV2(cfg Config, clientConfig client.Config, limits *validation.Overrides,
 	i.limiter = NewSeriesLimiter(limits, i.lifecycler, cfg.LifecyclerConfig.RingConfig.ReplicationFactor, cfg.ShardByAllLabels)
 	i.userStates = newUserStates(i.limiter, cfg, i.metrics)
 
-	// Scan and open TSDB's that already exist on disk
+	// Scan and open TSDB's that already exist on disk // TODO: move to starting
 	if err := i.openExistingTSDB(context.Background()); err != nil {
 		return nil, err
 	}
 
-	// Now that user states have been created, we can start the lifecycler
-	i.lifecycler.Start()
+	i.Service = services.NewBasicService(i.startingV2, i.updateLoop, i.stoppingV2)
+	return i, nil
+}
 
-	// Run the blocks shipping in a dedicated go routine.
-	if i.cfg.TSDBConfig.ShipInterval > 0 {
-		i.done.Add(1)
-		go i.shipBlocksLoop()
+func (i *Ingester) startingV2(ctx context.Context) error {
+	// Important: we want to keep lifecycler running until we ask it to stop, so we need to give it independent context
+	if err := i.lifecycler.StartAsync(context.Background()); err != nil {
+		return errors.Wrap(err, "failed to start lifecycler")
+	}
+	if err := i.lifecycler.AwaitRunning(ctx); err != nil {
+		return errors.Wrap(err, "failed to start lifecycler")
 	}
 
-	i.done.Add(1)
-	go i.updateLoop()
+	// let's start the rest of subservices via manager
+	servs := []services.Service(nil)
 
-	i.done.Add(1)
-	go i.compactionLoop()
+	compactionService := services.NewBasicService(nil, i.compactionLoop, nil)
+	servs = append(servs, compactionService)
 
-	return i, nil
+	if i.cfg.TSDBConfig.ShipInterval > 0 {
+		shippingService := services.NewBasicService(nil, i.shipBlocksLoop, nil)
+		servs = append(servs, shippingService)
+	}
+
+	var err error
+	i.TSDBState.subservices, err = services.NewManager(servs...)
+	if err == nil {
+		err = services.StartManagerAndAwaitHealthy(ctx, i.TSDBState.subservices)
+	}
+	return errors.Wrap(err, "failed to start ingester components")
 }
 
-func (i *Ingester) updateLoop() {
-	defer i.done.Done()
+// runs when V2 ingester is stopping
+func (i *Ingester) stoppingV2() 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.
+
+	if err := services.StopManagerAndAwaitStopped(context.Background(), i.TSDBState.subservices); err != nil {
+		level.Warn(util.Logger).Log("msg", "stopping ingester subservices", "err", err)
+	}
+
+	// Next initiate our graceful exit from the ring.
+	return services.StopAndAwaitTerminated(context.Background(), i.lifecycler)
+}
 
+func (i *Ingester) updateLoop(ctx context.Context) error {
 	rateUpdateTicker := time.NewTicker(i.cfg.RateUpdatePeriod)
 	defer rateUpdateTicker.Stop()
 
@@ -178,8 +206,8 @@ func (i *Ingester) updateLoop() {
 					userDB.refCache.Purge(time.Now().Add(-cortex_tsdb.DefaultRefCacheTTL))
 				}
 			}
-		case <-i.quit:
-			return
+		case <-ctx.Done():
+			return nil
 		}
 	}
 }
@@ -765,17 +793,13 @@ func (i *Ingester) numSeriesInTSDB() float64 {
 	return float64(count)
 }
 
-func (i *Ingester) shipBlocksLoop() {
-	// It's important to add the shipper loop to the "done" wait group,
-	// because the blocks transfer should start only once it's guaranteed
-	// there's no shipping on-going.
-	defer i.done.Done()
-
+func (i *Ingester) shipBlocksLoop(ctx context.Context) error {
 	// Start a goroutine that will cancel all shipper contexts on ingester
 	// shutdown, so that if there's any shipper sync in progress it will be
 	// quickly canceled.
+	// TODO: this could be a "stoppingFn" for shipper service, but let's keep that for later.
 	go func() {
-		<-i.quit
+		<-ctx.Done()
 
 		for _, userID := range i.getTSDBUsers() {
 			if userDB := i.getTSDB(userID); userDB != nil && userDB.shipperCancel != nil {
@@ -790,15 +814,15 @@ func (i *Ingester) shipBlocksLoop() {
 	for {
 		select {
 		case <-shipTicker.C:
-			i.shipBlocks()
+			i.shipBlocks(ctx)
 
-		case <-i.quit:
-			return
+		case <-ctx.Done():
+			return nil
 		}
 	}
 }
 
-func (i *Ingester) shipBlocks() {
+func (i *Ingester) shipBlocks(ctx context.Context) {
 	// Do not ship blocks if the ingester is PENDING or JOINING. It's
 	// particularly important for the JOINING state because there could
 	// be a blocks transfer in progress (from another ingester) and if we
@@ -810,7 +834,7 @@ func (i *Ingester) shipBlocks() {
 
 	// Number of concurrent workers is limited in order to avoid to concurrently sync a lot
 	// of tenants in a large cluster.
-	i.runConcurrentUserWorkers(i.cfg.TSDBConfig.ShipConcurrency, func(userID string) {
+	i.runConcurrentUserWorkers(ctx, i.cfg.TSDBConfig.ShipConcurrency, func(userID string) {
 		// Get the user's DB. If the user doesn't exist, we skip it.
 		userDB := i.getTSDB(userID)
 		if userDB == nil || userDB.shipper == nil {
@@ -831,31 +855,29 @@ func (i *Ingester) shipBlocks() {
 	})
 }
 
-func (i *Ingester) compactionLoop() {
-	defer i.done.Done()
-
+func (i *Ingester) compactionLoop(ctx context.Context) error {
 	ticker := time.NewTicker(i.cfg.TSDBConfig.HeadCompactionInterval)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			i.compactBlocks()
+			i.compactBlocks(ctx)
 
-		case <-i.quit:
-			return
+		case <-ctx.Done():
+			return nil
 		}
 	}
 }
 
-func (i *Ingester) compactBlocks() {
+func (i *Ingester) compactBlocks(ctx context.Context) {
 	// Don't compact TSDB blocks while JOINING or LEAVING, as there may be ongoing blocks transfers.
 	if ingesterState := i.lifecycler.GetState(); ingesterState == ring.JOINING || ingesterState == ring.LEAVING {
 		level.Info(util.Logger).Log("msg", "TSDB blocks compaction has been skipped because of the current ingester state", "state", ingesterState)
 		return
 	}
 
-	i.runConcurrentUserWorkers(i.cfg.TSDBConfig.HeadCompactionConcurrency, func(userID string) {
+	i.runConcurrentUserWorkers(ctx, i.cfg.TSDBConfig.HeadCompactionConcurrency, func(userID string) {
 		userDB := i.getTSDB(userID)
 		if userDB == nil {
 			return
@@ -872,7 +894,7 @@ func (i *Ingester) compactBlocks() {
 	})
 }
 
-func (i *Ingester) runConcurrentUserWorkers(concurrency int, userFunc func(userID string)) {
+func (i *Ingester) runConcurrentUserWorkers(ctx context.Context, concurrency int, userFunc func(userID string)) {
 	wg := sync.WaitGroup{}
 	ch := make(chan string)
 
@@ -892,7 +914,7 @@ sendLoop:
 		select {
 		case ch <- userID:
 			// ok
-		case <-i.quit:
+		case <-ctx.Done():
 			// don't start new tasks.
 			break sendLoop
 		}
diff --git a/pkg/querier/block.go b/pkg/querier/block.go
index 1f687f9742..1dc3a74265 100644
--- a/pkg/querier/block.go
+++ b/pkg/querier/block.go
@@ -21,11 +21,14 @@ import (
 	"github.com/cortexproject/cortex/pkg/querier/series"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
 
 // BlockQueryable is a storage.Queryable implementation for blocks storage
 type BlockQueryable struct {
+	services.Service
+
 	us *UserStore
 }
 
@@ -46,12 +49,25 @@ func NewBlockQueryable(cfg tsdb.Config, logLevel logging.Level, registerer prome
 	}
 
 	b := &BlockQueryable{us: us}
+	b.Service = services.NewIdleService(b.starting, b.stopping)
 
 	return b, nil
 }
 
+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 {
+	return errors.Wrap(services.StopAndAwaitTerminated(context.Background(), b.us), "stopping UserStore")
+}
+
 // Querier returns a new Querier on the storage.
 func (b *BlockQueryable) Querier(ctx context.Context, mint, maxt int64) (storage.Querier, error) {
+	if s := b.State(); s != services.Running {
+		return nil, promql.ErrStorage{Err: errors.Errorf("BlockQueryable is not running: %v", s)}
+	}
+
 	userID, err := user.ExtractOrgID(ctx)
 	if err != nil {
 		return nil, promql.ErrStorage{Err: err}
diff --git a/pkg/querier/block_store.go b/pkg/querier/block_store.go
index 1083bc26d4..c0981e12c3 100644
--- a/pkg/querier/block_store.go
+++ b/pkg/querier/block_store.go
@@ -27,11 +27,14 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
 
 // UserStore is a multi-tenant version of Thanos BucketStore
 type UserStore struct {
+	services.Service
+
 	logger             log.Logger
 	cfg                tsdb.Config
 	bucket             objstore.Bucket
@@ -50,9 +53,7 @@ type UserStore struct {
 	storesMu sync.RWMutex
 	stores   map[string]*store.BucketStore
 
-	// Used to cancel workers and wait until done.
-	workers       sync.WaitGroup
-	workersCancel context.CancelFunc
+	serv *grpc.Server
 
 	// Metrics.
 	syncTimes prometheus.Histogram
@@ -60,9 +61,6 @@ type UserStore struct {
 
 // NewUserStore returns a new UserStore
 func NewUserStore(cfg tsdb.Config, bucketClient objstore.Bucket, logLevel logging.Level, logger log.Logger, registerer prometheus.Registerer) (*UserStore, error) {
-	var err error
-
-	workersCtx, workersCancel := context.WithCancel(context.Background())
 	indexCacheRegistry := prometheus.NewRegistry()
 
 	u := &UserStore{
@@ -73,7 +71,6 @@ func NewUserStore(cfg tsdb.Config, bucketClient objstore.Bucket, logLevel loggin
 		logLevel:           logLevel,
 		bucketStoreMetrics: newTSDBBucketStoreMetrics(),
 		indexCacheMetrics:  newTSDBIndexCacheMetrics(indexCacheRegistry),
-		workersCancel:      workersCancel,
 		syncTimes: prometheus.NewHistogram(prometheus.HistogramOpts{
 			Name:    "cortex_querier_blocks_sync_seconds",
 			Help:    "The total time it takes to perform a sync stores",
@@ -82,6 +79,7 @@ func NewUserStore(cfg tsdb.Config, bucketClient objstore.Bucket, logLevel loggin
 	}
 
 	// Configure the time range to sync all blocks.
+	var err error
 	if err = u.syncMint.Set("0000-01-01T00:00:00Z"); err != nil {
 		return nil, err
 	}
@@ -98,41 +96,44 @@ func NewUserStore(cfg tsdb.Config, bucketClient objstore.Bucket, logLevel loggin
 		registerer.MustRegister(u.syncTimes, u.bucketStoreMetrics, u.indexCacheMetrics)
 	}
 
-	serv := grpc.NewServer()
-	storepb.RegisterStoreServer(serv, u)
+	u.Service = services.NewBasicService(u.starting, u.syncStoresLoop, u.stopping)
+	return u, nil
+}
+
+func (u *UserStore) starting(ctx context.Context) error {
+	u.serv = grpc.NewServer()
+	storepb.RegisterStoreServer(u.serv, u)
 	l, err := net.Listen("tcp", "")
 	if err != nil {
-		return nil, err
+		return err
 	}
-	go serv.Serve(l) //nolint:errcheck
+	go func() {
+		err := u.serv.Serve(l)
+		if err != nil {
+			level.Error(u.logger).Log("msg", "block store grpc server failed", "err", err)
+		}
+	}()
 
 	cc, err := grpc.Dial(l.Addr().String(), grpc.WithInsecure())
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	u.client = storepb.NewStoreClient(cc)
 
-	// If the sync is disabled we never sync blocks, which means the bucket store
-	// will be empty and no series will be returned once queried.
 	if u.cfg.BucketStore.SyncInterval > 0 {
 		// Run an initial blocks sync, required in order to be able to serve queries.
-		if err := u.initialSync(workersCtx); err != nil {
-			return nil, err
+		if err := u.initialSync(ctx); err != nil {
+			return err
 		}
-
-		// Periodically sync the blocks.
-		u.workers.Add(1)
-		go u.syncStoresLoop(workersCtx)
 	}
 
-	return u, nil
+	return nil
 }
 
-// Stop the blocks sync and waits until done.
-func (u *UserStore) Stop() {
-	u.workersCancel()
-	u.workers.Wait()
+func (u *UserStore) stopping() error {
+	u.serv.Stop()
+	return nil
 }
 
 // initialSync iterates over the storage bucket creating user bucket stores, and calling initialSync on each of them
@@ -151,8 +152,13 @@ func (u *UserStore) initialSync(ctx context.Context) error {
 }
 
 // syncStoresLoop periodically calls syncStores() to synchronize the blocks for all tenants.
-func (u *UserStore) syncStoresLoop(ctx context.Context) {
-	defer u.workers.Done()
+func (u *UserStore) syncStoresLoop(ctx context.Context) error {
+	// If the sync is disabled we never sync blocks, which means the bucket store
+	// will be empty and no series will be returned once queried.
+	if u.cfg.BucketStore.SyncInterval <= 0 {
+		<-ctx.Done()
+		return nil
+	}
 
 	syncInterval := u.cfg.BucketStore.SyncInterval
 
@@ -160,7 +166,7 @@ func (u *UserStore) syncStoresLoop(ctx context.Context) {
 	// sync interval before resynching.
 	select {
 	case <-ctx.Done():
-		return
+		return nil
 	case <-time.After(syncInterval):
 	}
 
@@ -175,12 +181,10 @@ func (u *UserStore) syncStoresLoop(ctx context.Context) {
 		return nil
 	})
 
-	if err != nil {
-		// This should never occur because the rununtil.Repeat() returns error
-		// only if the callback function returns error (which doesn't), but since
-		// we have to handle the error because of the linter, it's better to log it.
-		level.Error(u.logger).Log("msg", "blocks synchronization has been halted due to an unexpected error", "err", err)
-	}
+	// This should never occur because the rununtil.Repeat() returns error
+	// only if the callback function returns error (which doesn't), but since
+	// we have to handle the error because of the linter, it's better to log it.
+	return errors.Wrap(err, "blocks synchronization has been halted due to an unexpected error")
 }
 
 // syncStores iterates over the storage bucket creating user bucket stores
diff --git a/pkg/ring/lifecycler.go b/pkg/ring/lifecycler.go
index 74796e84d3..90b32d5898 100644
--- a/pkg/ring/lifecycler.go
+++ b/pkg/ring/lifecycler.go
@@ -2,6 +2,7 @@ package ring
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"os"
@@ -16,6 +17,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring/kv"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 var (
@@ -104,13 +106,12 @@ func (cfg *LifecyclerConfig) RegisterFlagsWithPrefix(prefix string, f *flag.Flag
 
 // Lifecycler is responsible for managing the lifecycle of entries in the ring.
 type Lifecycler struct {
+	*services.BasicService
+
 	cfg             LifecyclerConfig
 	flushTransferer FlushTransferer
 	KVStore         kv.Client
 
-	// Controls the lifecycle of the ingester
-	quit      chan struct{}
-	done      sync.WaitGroup
 	actorChan chan func()
 
 	// These values are initialised at startup, and never change
@@ -138,9 +139,8 @@ type Lifecycler struct {
 	healthyInstancesCount int
 }
 
-// NewLifecycler makes and starts a new Lifecycler.
+// NewLifecycler creates new Lifecycler. It must be started via StartAsync.
 func NewLifecycler(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringName, ringKey string, flushOnShutdown bool) (*Lifecycler, error) {
-
 	addr := cfg.Addr
 	if addr == "" {
 		var err error
@@ -176,7 +176,6 @@ func NewLifecycler(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringNa
 		RingKey:         ringKey,
 		flushOnShutdown: flushOnShutdown,
 
-		quit:      make(chan struct{}),
 		actorChan: make(chan func()),
 
 		state:     PENDING,
@@ -185,13 +184,9 @@ func NewLifecycler(cfg LifecyclerConfig, flushTransferer FlushTransferer, ringNa
 
 	tokensToOwn.WithLabelValues(l.RingName).Set(float64(cfg.NumTokens))
 
-	return l, nil
-}
+	l.BasicService = services.NewBasicService(nil, l.loop, l.stopping)
 
-// Start the lifecycler
-func (i *Lifecycler) Start() {
-	i.done.Add(1)
-	go i.loop()
+	return l, nil
 }
 
 // CheckReady is used to rate limit the number of ingesters that can be coming or
@@ -247,13 +242,31 @@ func (i *Lifecycler) setState(state IngesterState) {
 	i.state = state
 }
 
+func (i *Lifecycler) sendToLifecyclerLoop(fn func()) error {
+	sc := i.ServiceContext()
+	if sc == nil {
+		return errors.New("lifecycler not running")
+	}
+
+	select {
+	case <-sc.Done():
+		return errors.New("lifecycler not running")
+	case i.actorChan <- fn:
+		return nil
+	}
+}
+
 // ChangeState of the ingester, for use off of the loop() goroutine.
 func (i *Lifecycler) ChangeState(ctx context.Context, state IngesterState) error {
-	err := make(chan error)
-	i.actorChan <- func() {
-		err <- i.changeState(ctx, state)
+	errCh := make(chan error)
+	fn := func() {
+		errCh <- i.changeState(ctx, state)
 	}
-	return <-err
+
+	if err := i.sendToLifecyclerLoop(fn); err != nil {
+		return err
+	}
+	return <-errCh
 }
 
 func (i *Lifecycler) getTokens() Tokens {
@@ -283,9 +296,9 @@ func (i *Lifecycler) setTokens(tokens Tokens) {
 // assign token to the wrong ingester. While we could check for that state here, when this method is called,
 // transfers have already finished -- it's better to check for this *before* transfers start.
 func (i *Lifecycler) ClaimTokensFor(ctx context.Context, ingesterID string) error {
-	err := make(chan error)
+	errCh := make(chan error)
 
-	i.actorChan <- func() {
+	fn := func() {
 		var tokens Tokens
 
 		claimTokens := func(in interface{}) (out interface{}, retry bool, err error) {
@@ -307,10 +320,13 @@ func (i *Lifecycler) ClaimTokensFor(ctx context.Context, ingesterID string) erro
 		}
 
 		i.setTokens(tokens)
-		err <- nil
+		errCh <- nil
 	}
 
-	return <-err
+	if err := i.sendToLifecyclerLoop(fn); err != nil {
+		return err
+	}
+	return <-errCh
 }
 
 // HealthyInstancesCount returns the number of healthy instances in the ring, updated
@@ -322,25 +338,9 @@ func (i *Lifecycler) HealthyInstancesCount() int {
 	return i.healthyInstancesCount
 }
 
-// Shutdown the lifecycle.  It will:
-// - send chunks to another ingester, if it can.
-// - otherwise, flush chunks to the chunk store.
-// - remove config from Consul.
-// - block until we've successfully shutdown.
-func (i *Lifecycler) Shutdown() {
-	// This will prevent us accepting any more samples
-	i.flushTransferer.StopIncomingRequests()
-
-	// closing i.quit triggers loop() to exit, which in turn will trigger
-	// the removal of our tokens etc
-	close(i.quit)
-	i.done.Wait()
-}
-
-func (i *Lifecycler) loop() {
+func (i *Lifecycler) loop(ctx context.Context) error {
 	defer func() {
 		level.Info(util.Logger).Log("msg", "member.loop() exited gracefully", "ring", i.RingName)
-		i.done.Done()
 	}()
 
 	// First, see if we exist in the cluster, update our state to match if we do,
@@ -357,7 +357,6 @@ func (i *Lifecycler) loop() {
 	heartbeatTicker := time.NewTicker(i.cfg.HeartbeatPeriod)
 	defer heartbeatTicker.Stop()
 
-loop:
 	for {
 		select {
 		case <-autoJoinAfter:
@@ -416,10 +415,22 @@ loop:
 		case f := <-i.actorChan:
 			f()
 
-		case <-i.quit:
-			break loop
+		case <-ctx.Done():
+			return nil
 		}
 	}
+}
+
+// Shutdown the lifecycle.  It will:
+// - send chunks to another ingester, if it can.
+// - otherwise, flush chunks to the chunk store.
+// - remove config from Consul.
+func (i *Lifecycler) stopping() error {
+	// This will prevent us accepting any more samples
+	i.flushTransferer.StopIncomingRequests()
+
+	heartbeatTicker := time.NewTicker(i.cfg.HeartbeatPeriod)
+	defer heartbeatTicker.Stop()
 
 	// Mark ourselved as Leaving so no more samples are send to us.
 	err := i.changeState(context.Background(), LEAVING)
@@ -456,6 +467,8 @@ heartbeatLoop:
 		}
 		level.Info(util.Logger).Log("msg", "instance removed from the KV store", "ring", i.RingName)
 	}
+
+	return nil
 }
 
 // initRing is the first thing we do when we start. It:
diff --git a/pkg/ring/ring.go b/pkg/ring/ring.go
index 0748ddf9f5..9005f4ad85 100644
--- a/pkg/ring/ring.go
+++ b/pkg/ring/ring.go
@@ -17,6 +17,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ring/kv"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 const (
@@ -84,12 +85,12 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 
 // Ring holds the information about the members of the consistent hash ring.
 type Ring struct {
+	services.Service
+
 	name     string
 	key      string
 	cfg      Config
 	KVClient kv.Client
-	done     chan struct{}
-	quit     context.CancelFunc
 
 	mtx        sync.RWMutex
 	ringDesc   *Desc
@@ -102,7 +103,7 @@ type Ring struct {
 	oldestTimestampDesc *prometheus.Desc
 }
 
-// New creates a new Ring
+// New creates a new Ring. Being a service, Ring needs to be started to do anything.
 func New(cfg Config, name, key string) (*Ring, error) {
 	if cfg.ReplicationFactor <= 0 {
 		return nil, fmt.Errorf("ReplicationFactor must be greater than zero: %d", cfg.ReplicationFactor)
@@ -118,7 +119,6 @@ func New(cfg Config, name, key string) (*Ring, error) {
 		key:      key,
 		cfg:      cfg,
 		KVClient: store,
-		done:     make(chan struct{}),
 		ringDesc: &Desc{},
 		memberOwnershipDesc: prometheus.NewDesc(
 			"cortex_ring_member_ownership_percent",
@@ -146,20 +146,12 @@ func New(cfg Config, name, key string) (*Ring, error) {
 			[]string{"state", "name"}, nil,
 		),
 	}
-	var ctx context.Context
-	ctx, r.quit = context.WithCancel(context.Background())
-	go r.loop(ctx)
-	return r, nil
-}
 
-// Stop the distributor.
-func (r *Ring) Stop() {
-	r.quit()
-	<-r.done
+	r.Service = services.NewBasicService(nil, r.loop, nil)
+	return r, nil
 }
 
-func (r *Ring) loop(ctx context.Context) {
-	defer close(r.done)
+func (r *Ring) loop(ctx context.Context) error {
 	r.KVClient.WatchKey(ctx, r.key, func(value interface{}) bool {
 		if value == nil {
 			level.Info(util.Logger).Log("msg", "ring doesn't exist in consul yet")
@@ -175,6 +167,7 @@ func (r *Ring) loop(ctx context.Context) {
 		r.ringTokens = ringTokens
 		return true
 	})
+	return nil
 }
 
 // Get returns n (or more) ingesters which form the replicas for the given key.
diff --git a/pkg/ruler/ruler.go b/pkg/ruler/ruler.go
index 9ff6dd2ced..60d2179c96 100644
--- a/pkg/ruler/ruler.go
+++ b/pkg/ruler/ruler.go
@@ -34,6 +34,7 @@ import (
 	store "github.com/cortexproject/cortex/pkg/ruler/rules"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 var (
@@ -104,6 +105,8 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 
 // Ruler evaluates rules.
 type Ruler struct {
+	services.Service
+
 	cfg         Config
 	engine      *promql.Engine
 	queryable   promStorage.Queryable
@@ -111,8 +114,9 @@ type Ruler struct {
 	alertURL    *url.URL
 	notifierCfg *config.Config
 
-	lifecycler *ring.Lifecycler
-	ring       *ring.Ring
+	lifecycler  *ring.Lifecycler
+	ring        *ring.Ring
+	subservices *services.Manager
 
 	store          rules.RuleStore
 	mapper         *mapper
@@ -123,9 +127,6 @@ type Ruler struct {
 	notifiersMtx sync.Mutex
 	notifiers    map[string]*rulerNotifier
 
-	done       chan struct{}
-	terminated chan struct{}
-
 	registry prometheus.Registerer
 	logger   log.Logger
 }
@@ -153,52 +154,53 @@ func NewRuler(cfg Config, engine *promql.Engine, queryable promStorage.Queryable
 		pusher:       pusher,
 		mapper:       newMapper(cfg.RulePath, logger),
 		userManagers: map[string]*promRules.Manager{},
-		done:         make(chan struct{}),
-		terminated:   make(chan struct{}),
 		registry:     reg,
 		logger:       logger,
 	}
 
+	ruler.Service = services.NewBasicService(ruler.starting, ruler.run, ruler.stopping)
+	return ruler, nil
+}
+
+func (r *Ruler) starting(ctx context.Context) error {
 	// If sharding is enabled, create/join a ring to distribute tokens to
 	// the ruler
-	if cfg.EnableSharding {
-		lifecyclerCfg := cfg.Ring.ToLifecyclerConfig()
-		ruler.lifecycler, err = ring.NewLifecycler(lifecyclerCfg, ruler, "ruler", ring.RulerRingKey, true)
+	if r.cfg.EnableSharding {
+		lifecyclerCfg := r.cfg.Ring.ToLifecyclerConfig()
+		var err error
+		r.lifecycler, err = ring.NewLifecycler(lifecyclerCfg, r, "ruler", ring.RulerRingKey, true)
 		if err != nil {
-			return nil, err
+			return errors.Wrap(err, "failed to initialize ruler's lifecycler")
 		}
 
-		ruler.lifecycler.Start()
-
-		ruler.ring, err = ring.New(lifecyclerCfg.RingConfig, "ruler", ring.RulerRingKey)
+		r.ring, err = ring.New(lifecyclerCfg.RingConfig, "ruler", ring.RulerRingKey)
 		if err != nil {
-			return nil, err
+			return errors.Wrap(err, "failed to initialize ruler's ring")
 		}
-	}
 
-	go ruler.run()
-	level.Info(logger).Log("msg", "ruler up and running")
+		r.subservices, err = services.NewManager(r.lifecycler, r.ring)
+		if err == nil {
+			err = services.StartManagerAndAwaitHealthy(ctx, r.subservices)
+		}
+		return errors.Wrap(err, "failed to start ruler's services")
+	}
 
-	return ruler, nil
+	// TODO: ideally, ruler would wait until its queryable is finished starting.
+	return nil
 }
 
 // Stop stops the Ruler.
 // Each function of the ruler is terminated before leaving the ring
-func (r *Ruler) Stop() {
-	close(r.done)
-	<-r.terminated
-
+func (r *Ruler) stopping() error {
 	r.notifiersMtx.Lock()
 	for _, n := range r.notifiers {
 		n.stop()
 	}
 	r.notifiersMtx.Unlock()
 
-	if r.cfg.EnableSharding {
-		level.Info(r.logger).Log("msg", "attempting shutdown lifecycle")
-		r.lifecycler.Shutdown()
-		level.Info(r.logger).Log("msg", "shutting down the ring")
-		r.ring.Stop()
+	if r.subservices != nil {
+		// subservices manages ring and lifecycler, if sharding was enabled.
+		_ = services.StopManagerAndAwaitStopped(context.Background(), r.subservices)
 	}
 
 	level.Info(r.logger).Log("msg", "stopping user managers")
@@ -216,6 +218,7 @@ func (r *Ruler) Stop() {
 	wg.Wait()
 	r.userManagerMtx.Unlock()
 	level.Info(r.logger).Log("msg", "all user managers stopped")
+	return nil
 }
 
 // sendAlerts implements a rules.NotifyFunc for a Notifier.
@@ -327,19 +330,19 @@ func (r *Ruler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	}
 }
 
-func (r *Ruler) run() {
-	defer close(r.terminated)
+func (r *Ruler) run(ctx context.Context) error {
+	level.Info(r.logger).Log("msg", "ruler up and running")
 
 	tick := time.NewTicker(r.cfg.PollInterval)
 	defer tick.Stop()
 
-	r.loadRules(context.Background())
+	r.loadRules(ctx)
 	for {
 		select {
-		case <-r.done:
-			return
+		case <-ctx.Done():
+			return nil
 		case <-tick.C:
-			r.loadRules(context.Background())
+			r.loadRules(ctx)
 		}
 	}
 }
diff --git a/pkg/util/runtimeconfig/manager.go b/pkg/util/runtimeconfig/manager.go
index 87d777037c..af1fc34abb 100644
--- a/pkg/util/runtimeconfig/manager.go
+++ b/pkg/util/runtimeconfig/manager.go
@@ -1,6 +1,7 @@
 package runtimeconfig
 
 import (
+	"context"
 	"flag"
 	"sync"
 	"time"
@@ -9,6 +10,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 // Loader loads the configuration from file.
@@ -31,8 +33,9 @@ func (mc *ManagerConfig) RegisterFlags(f *flag.FlagSet) {
 // Manager periodically reloads the configuration from a file, and keeps this
 // configuration available for clients.
 type Manager struct {
-	cfg  ManagerConfig
-	quit chan struct{}
+	services.Service
+
+	cfg ManagerConfig
 
 	listenersMtx sync.Mutex
 	listeners    []chan interface{}
@@ -46,8 +49,7 @@ type Manager struct {
 // NewRuntimeConfigManager creates an instance of Manager and starts reload config loop based on config
 func NewRuntimeConfigManager(cfg ManagerConfig, registerer prometheus.Registerer) (*Manager, error) {
 	mgr := Manager{
-		cfg:  cfg,
-		quit: make(chan struct{}),
+		cfg: cfg,
 		configLoadSuccess: prometheus.NewGauge(prometheus.GaugeOpts{
 			Name: "cortex_overrides_last_reload_successful",
 			Help: "Whether the last config reload attempt was successful.",
@@ -58,17 +60,18 @@ func NewRuntimeConfigManager(cfg ManagerConfig, registerer prometheus.Registerer
 		registerer.MustRegister(mgr.configLoadSuccess)
 	}
 
-	if cfg.LoadPath != "" {
-		if err := mgr.loadConfig(); err != nil {
+	mgr.Service = services.NewBasicService(mgr.start, mgr.loop, mgr.stop)
+	return &mgr, nil
+}
+
+func (om *Manager) start(_ context.Context) error {
+	if om.cfg.LoadPath != "" {
+		if err := om.loadConfig(); err != nil {
 			// Log but don't stop on error - we don't want to halt all ingesters because of a typo
 			level.Error(util.Logger).Log("msg", "failed to load config", "err", err)
 		}
-		go mgr.loop()
-	} else {
-		level.Info(util.Logger).Log("msg", "runtime config disabled: file not specified")
 	}
-
-	return &mgr, nil
+	return nil
 }
 
 // CreateListenerChannel creates new channel that can be used to receive new config values.
@@ -101,7 +104,13 @@ func (om *Manager) CloseListenerChannel(listener <-chan interface{}) {
 	}
 }
 
-func (om *Manager) loop() {
+func (om *Manager) loop(ctx context.Context) error {
+	if om.cfg.LoadPath == "" {
+		level.Info(util.Logger).Log("msg", "runtime config disabled: file not specified")
+		<-ctx.Done()
+		return nil
+	}
+
 	ticker := time.NewTicker(om.cfg.ReloadPeriod)
 	defer ticker.Stop()
 
@@ -113,8 +122,8 @@ func (om *Manager) loop() {
 				// Log but don't stop on error - we don't want to halt all ingesters because of a typo
 				level.Error(util.Logger).Log("msg", "failed to load config", "err", err)
 			}
-		case <-om.quit:
-			return
+		case <-ctx.Done():
+			return nil
 		}
 	}
 }
@@ -156,9 +165,7 @@ func (om *Manager) callListeners(newValue interface{}) {
 }
 
 // Stop stops the Manager
-func (om *Manager) Stop() {
-	close(om.quit)
-
+func (om *Manager) stop() error {
 	om.listenersMtx.Lock()
 	defer om.listenersMtx.Unlock()
 
@@ -166,6 +173,7 @@ func (om *Manager) Stop() {
 		close(ch)
 	}
 	om.listeners = nil
+	return nil
 }
 
 // GetConfig returns last loaded config value, possibly nil.
diff --git a/pkg/util/services/basic_service.go b/pkg/util/services/basic_service.go
index 2a288991b6..f92daafce6 100644
--- a/pkg/util/services/basic_service.go
+++ b/pkg/util/services/basic_service.go
@@ -101,6 +101,8 @@ func (b *BasicService) StartAsync(parentContext context.Context) error {
 	return nil
 }
 
+// Returns true, if state switch succeeds, false if it fails. Returned state is the state before switch.
+// if state switching succeeds, stateFn runs with lock held.
 func (b *BasicService) switchState(from, to State, stateFn func()) (bool, State) {
 	b.stateMu.Lock()
 	defer b.stateMu.Unlock()
diff --git a/pkg/util/services/manager.go b/pkg/util/services/manager.go
index 765acaaf5c..2a2670e28c 100644
--- a/pkg/util/services/manager.go
+++ b/pkg/util/services/manager.go
@@ -306,3 +306,24 @@ func (f funcBasedManagerListener) Failure(service Service) {
 		f.failure(service)
 	}
 }
+
+// StartManagerAndAwaitHealthy starts the manager (which in turns starts all services managed by it), and then waits
+// until it reaches Running state. All services that this manager manages must be in New state, otherwise starting
+// will fail.
+//
+// Notice that context passed to the manager for starting its services is the same as context used for waiting!
+func StartManagerAndAwaitHealthy(ctx context.Context, manager *Manager) error {
+	err := manager.StartAsync(ctx)
+	if err != nil {
+		return err
+	}
+
+	return manager.AwaitHealthy(ctx)
+}
+
+// StopManagerAndAwaitStopped asks manager to stop its services, and then waits
+// until manager reaches the stopped state or context is stopped.
+func StopManagerAndAwaitStopped(ctx context.Context, manager *Manager) error {
+	manager.StopAsync()
+	return manager.AwaitStopped(ctx)
+}
diff --git a/pkg/util/services/services.go b/pkg/util/services/services.go
index b054f4106a..3978695552 100644
--- a/pkg/util/services/services.go
+++ b/pkg/util/services/services.go
@@ -92,3 +92,41 @@ func (f funcBasedListener) Failed(from State, failure error) {
 		f.failedFn(from, failure)
 	}
 }
+
+// StartAndAwaitRunning starts the service, and then waits until it reaches Running state.
+// If service fails to start, its failure case is returned.
+// Service must be in New state when this function is called.
+//
+// Notice that context passed to the service for starting is the same as context used for waiting!
+// If you need these contexts to be different, please use StartAsync and AwaitRunning directly.
+func StartAndAwaitRunning(ctx context.Context, service Service) error {
+	err := service.StartAsync(ctx)
+	if err != nil {
+		return err
+	}
+
+	err = service.AwaitRunning(ctx)
+	if e := service.FailureCase(); e != nil {
+		return e
+	}
+
+	return err
+}
+
+// StopAndAwaitTerminated asks service to stop, and then waits until service reaches Terminated
+// or Failed state. If service ends in Terminated state, this function returns error. On Failed state,
+// it returns the failure case. Other errors are possible too (eg. if context stops before service does).
+func StopAndAwaitTerminated(ctx context.Context, service Service) error {
+	service.StopAsync()
+	err := service.AwaitTerminated(ctx)
+	if err == nil {
+		return nil
+	}
+
+	if e := service.FailureCase(); e != nil {
+		return e
+	}
+
+	// can happen e.g. if context was canceled
+	return err
+}


Test patch:
diff --git a/integration/backward_compatibility_test.go b/integration/backward_compatibility_test.go
index 935da62dd3..3642875afe 100644
--- a/integration/backward_compatibility_test.go
+++ b/integration/backward_compatibility_test.go
@@ -34,8 +34,12 @@ func TestBackwardCompatibilityWithChunksStorage(t *testing.T) {
 	// Start Cortex components (ingester running on previous version).
 	require.NoError(t, writeFileToSharedDir(s, cortexSchemaConfigFile, []byte(cortexSchemaConfigYaml)))
 	tableManager := e2ecortex.NewTableManager("table-manager", ChunksStorageFlags, previousVersionImage)
+	// Old table-manager doesn't expose a readiness probe, so we just check if the / returns 404
+	tableManager.SetReadinessProbe(e2e.NewReadinessProbe(tableManager.HTTPPort(), "/", 404))
 	ingester1 := e2ecortex.NewIngester("ingester-1", consul.NetworkHTTPEndpoint(), ChunksStorageFlags, "")
 	distributor := e2ecortex.NewDistributor("distributor", consul.NetworkHTTPEndpoint(), ChunksStorageFlags, "")
+	// Old ring didn't have /ready probe, use /ring instead.
+	distributor.SetReadinessProbe(e2e.NewReadinessProbe(distributor.HTTPPort(), "/ring", 200))
 	require.NoError(t, s.StartAndWaitReady(distributor, ingester1, tableManager))
 
 	// Wait until the first table-manager sync has completed, so that we're
diff --git a/integration/e2e/service.go b/integration/e2e/service.go
index 3fa1361e87..3ef6b4cbb4 100644
--- a/integration/e2e/service.go
+++ b/integration/e2e/service.go
@@ -218,6 +218,10 @@ func (s *ConcreteService) NetworkEndpointFor(networkName string, port int) strin
 	return fmt.Sprintf("%s:%d", containerName(networkName, s.name), port)
 }
 
+func (s *ConcreteService) SetReadinessProbe(probe *ReadinessProbe) {
+	s.readiness = probe
+}
+
 func (s *ConcreteService) Ready() error {
 	if !s.isExpectedRunning() {
 		return fmt.Errorf("service %s is stopped", s.Name())
@@ -425,6 +429,10 @@ func (s *HTTPService) metrics() (_ string, err error) {
 	return string(body), err
 }
 
+func (s *HTTPService) HTTPPort() int {
+	return s.httpPort
+}
+
 func (s *HTTPService) HTTPEndpoint() string {
 	return s.Endpoint(s.httpPort)
 }
diff --git a/integration/e2ecortex/services.go b/integration/e2ecortex/services.go
index d91ff2df0f..a882db0d89 100644
--- a/integration/e2ecortex/services.go
+++ b/integration/e2ecortex/services.go
@@ -48,7 +48,7 @@ func NewDistributorWithConfigFile(name, consulAddress, configFile string, flags
 			"-ring.store":      "consul",
 			"-consul.hostname": consulAddress,
 		}, flags))...),
-		e2e.NewReadinessProbe(httpPort, "/ring", 200),
+		e2e.NewReadinessProbe(httpPort, "/ready", 204),
 		httpPort,
 		grpcPort,
 	)
@@ -143,8 +143,7 @@ func NewTableManagerWithConfigFile(name, configFile string, flags map[string]str
 			"-target":    "table-manager",
 			"-log.level": "warn",
 		}, flags))...),
-		// The table-manager doesn't expose a readiness probe, so we just check if the / returns 404
-		e2e.NewReadinessProbe(httpPort, "/", 404),
+		e2e.NewReadinessProbe(httpPort, "/ready", 204),
 		httpPort,
 		grpcPort,
 	)
@@ -170,8 +169,7 @@ func NewQueryFrontendWithConfigFile(name, configFile string, flags map[string]st
 			"-target":    "query-frontend",
 			"-log.level": "warn",
 		}, flags))...),
-		// The query-frontend doesn't expose a readiness probe, so we just check if the / returns 404
-		e2e.NewReadinessProbe(httpPort, "/", 404),
+		e2e.NewReadinessProbe(httpPort, "/ready", 204),
 		httpPort,
 		grpcPort,
 	)
@@ -207,8 +205,7 @@ func NewAlertmanager(name string, flags map[string]string, image string) *Cortex
 			"-target":    "alertmanager",
 			"-log.level": "warn",
 		}, flags))...),
-		// The alertmanager doesn't expose a readiness probe, so we just check if the / returns 404
-		e2e.NewReadinessProbe(httpPort, "/", 404),
+		e2e.NewReadinessProbe(httpPort, "/ready", 204),
 		httpPort,
 		grpcPort,
 	)
diff --git a/pkg/alertmanager/multitenant_test.go b/pkg/alertmanager/multitenant_test.go
index f38c35beaa..158d4a3fc0 100644
--- a/pkg/alertmanager/multitenant_test.go
+++ b/pkg/alertmanager/multitenant_test.go
@@ -6,7 +6,6 @@ import (
 	"context"
 	"io/ioutil"
 	"os"
-	"sync"
 	"testing"
 
 	"github.com/go-kit/kit/log"
@@ -65,20 +64,10 @@ func TestLoadAllConfigs(t *testing.T) {
 	require.NoError(t, err)
 	defer os.RemoveAll(tempDir)
 
-	am := &MultitenantAlertmanager{
-		cfg: &MultitenantAlertmanagerConfig{
-			ExternalURL: externalURL,
-			DataDir:     tempDir,
-		},
-		store:            mockStore,
-		cfgs:             map[string]alerts.AlertConfigDesc{},
-		alertmanagersMtx: sync.Mutex{},
-		alertmanagers:    map[string]*Alertmanager{},
-		logger:           log.NewNopLogger(),
-		stop:             make(chan struct{}),
-		done:             make(chan struct{}),
-		metrics:          newAlertmanagerMetrics(),
-	}
+	am := createMultitenantAlertmanager(&MultitenantAlertmanagerConfig{
+		ExternalURL: externalURL,
+		DataDir:     tempDir,
+	}, nil, nil, mockStore, log.NewNopLogger(), nil)
 
 	// Ensure the configs are synced correctly
 	require.NoError(t, am.updateConfigs())
@@ -123,7 +112,7 @@ func TestLoadAllConfigs(t *testing.T) {
 
 	userAM, exists := am.alertmanagers["user3"]
 	require.True(t, exists)
-	require.False(t, userAM.isActive())
+	require.False(t, userAM.IsActive())
 
 	// Ensure when a 3rd config is re-added, it is synced correctly
 	mockStore.configs["user3"] = alerts.AlertConfigDesc{
@@ -140,5 +129,5 @@ func TestLoadAllConfigs(t *testing.T) {
 
 	userAM, exists = am.alertmanagers["user3"]
 	require.True(t, exists)
-	require.True(t, userAM.isActive())
+	require.True(t, userAM.IsActive())
 }
diff --git a/pkg/chunk/purger/purger_test.go b/pkg/chunk/purger/purger_test.go
index f1eb9e8fd8..3f86d9e9fa 100644
--- a/pkg/chunk/purger/purger_test.go
+++ b/pkg/chunk/purger/purger_test.go
@@ -15,6 +15,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/chunk/testutils"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 const (
@@ -140,7 +141,7 @@ func TestDataPurger_BuildPlan(t *testing.T) {
 			t.Run(fmt.Sprintf("%s/batch-size=%d", tc.name, batchSize), func(t *testing.T) {
 				deleteStore, chunkStore, storageClient, dataPurger := setupStoresAndPurger(t)
 				defer func() {
-					dataPurger.Stop()
+					dataPurger.StopAsync()
 					chunkStore.Stop()
 				}()
 
@@ -221,7 +222,7 @@ func TestDataPurger_ExecutePlan(t *testing.T) {
 			t.Run(fmt.Sprintf("%s/batch-size=%d", tc.name, batchSize), func(t *testing.T) {
 				deleteStore, chunkStore, _, dataPurger := setupStoresAndPurger(t)
 				defer func() {
-					dataPurger.Stop()
+					dataPurger.StopAsync()
 					chunkStore.Stop()
 				}()
 
@@ -305,7 +306,7 @@ func TestDataPurger_Restarts(t *testing.T) {
 	require.NoError(t, err)
 
 	// stop the existing purger
-	dataPurger.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), dataPurger))
 
 	// create a new purger to check whether it picks up in process delete requests
 	var cfg Config
@@ -314,11 +315,9 @@ func TestDataPurger_Restarts(t *testing.T) {
 	require.NoError(t, err)
 
 	// load in process delete requests by calling Run
-	require.NoError(t, newPurger.Init())
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), newPurger))
 
-	defer func() {
-		newPurger.Stop()
-	}()
+	defer newPurger.StopAsync()
 
 	// lets wait till purger finishes execution of in process delete requests
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 5c8ac29e89..8d4c2a5931 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -21,12 +21,14 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"gopkg.in/yaml.v2"
 
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	cortex_testutil "github.com/cortexproject/cortex/pkg/util/test"
 )
 
@@ -81,14 +83,14 @@ func TestCompactor_ShouldDoNothingOnNoUserBlocks(t *testing.T) {
 
 	c, _, logs, registry, cleanup := prepare(t, prepareConfig(), bucketClient)
 	defer cleanup()
-	c.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
 
 	// Wait until a run has completed.
 	cortex_testutil.Poll(t, time.Second, 1.0, func() interface{} {
 		return prom_testutil.ToFloat64(c.compactionRunsCompleted)
 	})
 
-	c.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), c))
 
 	assert.Equal(t, []string{
 		`level=info msg="discovering users from bucket"`,
@@ -171,14 +173,14 @@ func TestCompactor_ShouldRetryOnFailureWhileDiscoveringUsersFromBucket(t *testin
 
 	c, _, logs, registry, cleanup := prepare(t, prepareConfig(), bucketClient)
 	defer cleanup()
-	c.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
 
 	// Wait until all retry attempts have completed.
 	cortex_testutil.Poll(t, time.Second, 1.0, func() interface{} {
 		return prom_testutil.ToFloat64(c.compactionRunsFailed)
 	})
 
-	c.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), c))
 
 	// Ensure the bucket iteration has been retried the configured number of times.
 	bucketClient.AssertNumberOfCalls(t, "Iter", 3)
@@ -272,7 +274,7 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 
 	c, tsdbCompactor, logs, registry, cleanup := prepare(t, prepareConfig(), bucketClient)
 	defer cleanup()
-	c.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
 
 	// Mock the compactor as if there's no compaction to do,
 	// in order to simplify tests (all in all, we just want to
@@ -285,7 +287,7 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 		return prom_testutil.ToFloat64(c.compactionRunsCompleted)
 	})
 
-	c.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), c))
 
 	// Ensure a plan has been executed for the blocks of each user.
 	tsdbCompactor.AssertNumberOfCalls(t, "Plan", 2)
@@ -344,7 +346,7 @@ func TestCompactor_ShouldCompactAllUsersOnShardingEnabledButOnlyOneInstanceRunni
 
 	c, tsdbCompactor, logs, _, cleanup := prepare(t, cfg, bucketClient)
 	defer cleanup()
-	c.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
 
 	// Mock the compactor as if there's no compaction to do,
 	// in order to simplify tests (all in all, we just want to
@@ -357,7 +359,7 @@ func TestCompactor_ShouldCompactAllUsersOnShardingEnabledButOnlyOneInstanceRunni
 		return prom_testutil.ToFloat64(c.compactionRunsCompleted)
 	})
 
-	c.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), c))
 
 	// Ensure a plan has been executed for the blocks of each user.
 	tsdbCompactor.AssertNumberOfCalls(t, "Plan", 2)
@@ -416,7 +418,7 @@ func TestCompactor_ShouldCompactOnlyUsersOwnedByTheInstanceOnShardingEnabledAndM
 		cfg.ShardingRing.KVStore.Mock = kvstore
 
 		c, tsdbCompactor, l, _, cleanup := prepare(t, cfg, bucketClient)
-		defer c.Stop()
+		defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 		defer cleanup()
 
 		compactors = append(compactors, c)
@@ -429,9 +431,15 @@ func TestCompactor_ShouldCompactOnlyUsersOwnedByTheInstanceOnShardingEnabledAndM
 		tsdbCompactor.On("Plan", mock.Anything).Return([]string{}, nil)
 	}
 
+	// Start all compactors
+	for _, c := range compactors {
+		require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+	}
+
 	// Wait until each compactor sees all ACTIVE compactors in the ring
 	for _, c := range compactors {
 		cortex_testutil.Poll(t, 10*time.Second, len(compactors), func() interface{} {
+			// it is safe to access c.ring here, since we know that all compactors are Running now
 			rs, err := c.ring.GetAll()
 			if err != nil {
 				return 0
@@ -448,11 +456,6 @@ func TestCompactor_ShouldCompactOnlyUsersOwnedByTheInstanceOnShardingEnabledAndM
 		})
 	}
 
-	// Start all compactors
-	for _, c := range compactors {
-		c.Start()
-	}
-
 	// Wait until a run has been completed on each compactor
 	for _, c := range compactors {
 		cortex_testutil.Poll(t, 10*time.Second, 1.0, func() interface{} {
@@ -536,8 +539,9 @@ func prepare(t *testing.T, compactorCfg Config, bucketClient *cortex_tsdb.Bucket
 	logger := log.NewLogfmtLogger(logs)
 	registry := prometheus.NewRegistry()
 
-	ctx, cancelCtx := context.WithCancel(context.Background())
-	c, err := newCompactor(ctx, cancelCtx, compactorCfg, storageCfg, bucketClient, tsdbCompactor, logger, registry)
+	c, err := newCompactor(compactorCfg, storageCfg, logger, registry, func(ctx context.Context) (objstore.Bucket, tsdb.Compactor, error) {
+		return bucketClient, tsdbCompactor, nil
+	})
 	require.NoError(t, err)
 
 	return c, tsdbCompactor, logs, registry, cleanup
diff --git a/pkg/cortex/cortex_test.go b/pkg/cortex/cortex_test.go
new file mode 100644
index 0000000000..9eebf952fe
--- /dev/null
+++ b/pkg/cortex/cortex_test.go
@@ -0,0 +1,70 @@
+package cortex
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/pkg/chunk/storage"
+	"github.com/cortexproject/cortex/pkg/ingester"
+	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/ring/kv"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb/backend/s3"
+	"github.com/cortexproject/cortex/pkg/util/services"
+)
+
+func TestCortex(t *testing.T) {
+	cfg := Config{
+		Storage: storage.Config{
+			Engine: storage.StorageEngineTSDB, // makes config easier
+		},
+		Ingester: ingester.Config{
+			TSDBConfig: tsdb.Config{
+				Backend: tsdb.BackendS3,
+				S3: s3.Config{
+					Endpoint: "localhost",
+				},
+			},
+			LifecyclerConfig: ring.LifecyclerConfig{
+				RingConfig: ring.Config{
+					KVStore: kv.Config{
+						Store: "inmemory",
+					},
+					ReplicationFactor: 3,
+				},
+				InfNames: []string{"en0", "eth0", "lo0"},
+			},
+		},
+		TSDB: tsdb.Config{
+			Backend: tsdb.BackendS3,
+			S3: s3.Config{
+				Endpoint: "localhost",
+			},
+		},
+		Target: All,
+	}
+
+	c, err := New(cfg)
+	require.NoError(t, err)
+	require.NotNil(t, c.serviceMap)
+
+	for m, s := range c.serviceMap {
+		// make sure each service is still New
+		require.Equal(t, services.New, s.State(), "module: %s", m)
+	}
+
+	// check random modules that we expect to be configured when using Target=All
+	require.NotNil(t, c.serviceMap[Server])
+	require.NotNil(t, c.serviceMap[Ingester])
+	require.NotNil(t, c.serviceMap[Ring])
+	require.NotNil(t, c.serviceMap[Distributor])
+
+	r, ok := c.serviceMap[Ring].(*moduleServiceWrapper)
+	require.True(t, ok)
+
+	require.ElementsMatch(t, []moduleName{Server, RuntimeConfig, MemberlistKV}, r.startDeps)
+
+	// querier and distributor depend on Ring
+	require.ElementsMatch(t, []moduleName{Distributor, Querier}, r.stopDeps)
+}
diff --git a/pkg/cortex/server_service_test.go b/pkg/cortex/server_service_test.go
new file mode 100644
index 0000000000..8d5249bbf1
--- /dev/null
+++ b/pkg/cortex/server_service_test.go
@@ -0,0 +1,73 @@
+package cortex
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"github.com/weaveworks/common/server"
+
+	"github.com/cortexproject/cortex/pkg/util/services"
+)
+
+func TestServerStopViaContext(t *testing.T) {
+	// server registers some metrics to default registry
+	savedRegistry := prometheus.DefaultRegisterer
+	prometheus.DefaultRegisterer = prometheus.NewRegistry()
+	defer func() {
+		prometheus.DefaultRegisterer = savedRegistry
+	}()
+
+	serv, err := server.New(server.Config{})
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	s := NewServerService(serv, func() []services.Service { return nil })
+	require.NoError(t, s.StartAsync(ctx))
+
+	// should terminate soon, since context has short timeout
+	require.NoError(t, s.AwaitTerminated(context.Background()))
+}
+
+func TestServerStopViaShutdown(t *testing.T) {
+	// server registers some metrics to default registry
+	savedRegistry := prometheus.DefaultRegisterer
+	prometheus.DefaultRegisterer = prometheus.NewRegistry()
+	defer func() {
+		prometheus.DefaultRegisterer = savedRegistry
+	}()
+
+	serv, err := server.New(server.Config{})
+	require.NoError(t, err)
+
+	s := NewServerService(serv, func() []services.Service { return nil })
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), s))
+
+	// we stop HTTP/gRPC Servers here... that should make server stop.
+	serv.Shutdown()
+
+	require.NoError(t, s.AwaitTerminated(context.Background()))
+}
+
+func TestServerStopViaStop(t *testing.T) {
+	// server registers some metrics to default registry
+	savedRegistry := prometheus.DefaultRegisterer
+	prometheus.DefaultRegisterer = prometheus.NewRegistry()
+	defer func() {
+		prometheus.DefaultRegisterer = savedRegistry
+	}()
+
+	serv, err := server.New(server.Config{})
+	require.NoError(t, err)
+
+	s := NewServerService(serv, func() []services.Service { return nil })
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), s))
+
+	serv.Stop()
+
+	require.NoError(t, s.AwaitTerminated(context.Background()))
+}
diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go
index ba116b1513..bc96bdb238 100644
--- a/pkg/distributor/distributor_test.go
+++ b/pkg/distributor/distributor_test.go
@@ -34,6 +34,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/chunkcompat"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/test"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
@@ -96,7 +97,7 @@ func TestDistributor_Push(t *testing.T) {
 				limits.IngestionBurstSize = 20
 
 				d, _ := prepare(t, tc.numIngesters, tc.happyIngesters, 0, shardByAllLabels, limits, nil)
-				defer d.Stop()
+				defer services.StopAndAwaitTerminated(context.Background(), d) //nolint:errcheck
 
 				request := makeWriteRequest(tc.samples)
 				response, err := d.Push(ctx, request)
@@ -175,7 +176,7 @@ func TestDistributor_PushIngestionRateLimiter(t *testing.T) {
 			distributors := make([]*Distributor, testData.distributors)
 			for i := 0; i < testData.distributors; i++ {
 				distributors[i], _ = prepare(t, 1, 1, 0, true, limits, kvStore)
-				defer distributors[i].Stop()
+				defer services.StopAndAwaitTerminated(context.Background(), distributors[i]) //nolint:errcheck
 			}
 
 			// If the distributors ring is setup, wait until the first distributor
@@ -259,7 +260,8 @@ func TestDistributor_PushHAInstances(t *testing.T) {
 						UpdateTimeout:   100 * time.Millisecond,
 						FailoverTimeout: time.Second,
 					})
-					assert.NoError(t, err)
+					require.NoError(t, err)
+					require.NoError(t, services.StartAndAwaitRunning(context.Background(), r))
 					d.Replicas = r
 				}
 
@@ -371,7 +373,7 @@ func TestDistributor_PushQuery(t *testing.T) {
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
 			d, _ := prepare(t, tc.numIngesters, tc.happyIngesters, 0, tc.shardByAllLabels, nil, nil)
-			defer d.Stop()
+			defer services.StopAndAwaitTerminated(context.Background(), d) //nolint:errcheck
 
 			request := makeWriteRequest(tc.samples)
 			writeResponse, err := d.Push(ctx, request)
@@ -454,7 +456,7 @@ func TestDistributor_Push_LabelRemoval(t *testing.T) {
 		limits.AcceptHASamples = tc.removeReplica
 
 		d, ingesters := prepare(t, 1, 1, 0, true, &limits, nil)
-		defer d.Stop()
+		defer services.StopAndAwaitTerminated(context.Background(), d) //nolint:errcheck
 
 		// Push the series to the distributor
 		req := mockWriteRequest(tc.inputSeries, 1, 1)
@@ -556,7 +558,7 @@ func TestDistributor_Push_ShouldGuaranteeShardingTokenConsistencyOverTheTime(t *
 	for testName, testData := range tests {
 		t.Run(testName, func(t *testing.T) {
 			d, ingesters := prepare(t, 1, 1, 0, true, &limits, nil)
-			defer d.Stop()
+			defer services.StopAndAwaitTerminated(context.Background(), d) //nolint:errcheck
 
 			// Push the series to the distributor
 			req := mockWriteRequest(testData.inputSeries, 1, 1)
@@ -593,7 +595,7 @@ func TestSlowQueries(t *testing.T) {
 				expectedErr = promql.ErrStorage{Err: errFail}
 			}
 			d, _ := prepare(t, nIngesters, happy, 100*time.Millisecond, shardByAllLabels, nil, nil)
-			defer d.Stop()
+			defer services.StopAndAwaitTerminated(context.Background(), d) //nolint:errcheck
 
 			_, err := d.Query(ctx, 0, 10, nameMatcher)
 			assert.Equal(t, expectedErr, err)
@@ -659,7 +661,7 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 
 	// Create distributor
 	d, _ := prepare(t, 3, 3, time.Duration(0), true, nil, nil)
-	defer d.Stop()
+	defer services.StopAndAwaitTerminated(context.Background(), d) //nolint:errcheck
 
 	// Push fixtures
 	ctx := user.InjectOrgID(context.Background(), "test")
@@ -761,6 +763,7 @@ func prepare(t *testing.T, numIngesters, happyIngesters int, queryDelay time.Dur
 
 	d, err := New(cfg, clientConfig, overrides, ingestersRing, true)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), d))
 
 	return d, ingesters
 }
@@ -1135,7 +1138,7 @@ func TestDistributorValidation(t *testing.T) {
 			limits.MaxLabelNamesPerSeries = 2
 
 			d, _ := prepare(t, 3, 3, 0, true, &limits, nil)
-			defer d.Stop()
+			defer services.StopAndAwaitTerminated(context.Background(), d) //nolint:errcheck
 
 			_, err := d.Push(ctx, client.ToWriteRequest(tc.labels, tc.samples, client.API))
 			require.Equal(t, tc.err, err)
diff --git a/pkg/distributor/ha_tracker_test.go b/pkg/distributor/ha_tracker_test.go
index e76348df0e..4767650a56 100644
--- a/pkg/distributor/ha_tracker_test.go
+++ b/pkg/distributor/ha_tracker_test.go
@@ -17,6 +17,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring/kv"
 	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 var (
@@ -135,7 +136,9 @@ func TestWatchPrefixAssignment(t *testing.T) {
 		UpdateTimeoutJitterMax: 0,
 		FailoverTimeout:        time.Millisecond * 2,
 	})
-	assert.NoError(t, err)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+	defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 
 	// Write the first time.
 	mtime.NowForce(start)
@@ -163,7 +166,9 @@ func TestCheckReplicaOverwriteTimeout(t *testing.T) {
 		UpdateTimeoutJitterMax: 0,
 		FailoverTimeout:        time.Second,
 	})
-	assert.NoError(t, err)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+	defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 
 	// Write the first time.
 	err = c.checkReplica(context.Background(), "user", "test", replica1)
@@ -196,7 +201,9 @@ func TestCheckReplicaMultiCluster(t *testing.T) {
 		UpdateTimeoutJitterMax: 0,
 		FailoverTimeout:        time.Second,
 	})
-	assert.NoError(t, err)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+	defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 
 	// Write the first time.
 	err = c.checkReplica(context.Background(), "user", "c1", replica1)
@@ -229,7 +236,9 @@ func TestCheckReplicaMultiClusterTimeout(t *testing.T) {
 		UpdateTimeoutJitterMax: 0,
 		FailoverTimeout:        time.Second,
 	})
-	assert.NoError(t, err)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+	defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 
 	// Write the first time.
 	err = c.checkReplica(context.Background(), "user", "c1", replica1)
@@ -278,7 +287,9 @@ func TestCheckReplicaUpdateTimeout(t *testing.T) {
 		UpdateTimeoutJitterMax: 0,
 		FailoverTimeout:        time.Second,
 	})
-	assert.NoError(t, err)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+	defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 
 	// Write the first time.
 	mtime.NowForce(startTime)
@@ -337,7 +348,9 @@ func TestCheckReplicaMultiUser(t *testing.T) {
 		UpdateTimeoutJitterMax: 0,
 		FailoverTimeout:        time.Second,
 	})
-	assert.NoError(t, err)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+	defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 
 	// Write the first time for user 1.
 	mtime.NowForce(start)
@@ -418,7 +431,9 @@ func TestCheckReplicaUpdateTimeoutJitter(t *testing.T) {
 				UpdateTimeoutJitterMax: 0,
 				FailoverTimeout:        time.Second,
 			})
-			assert.NoError(t, err)
+			require.NoError(t, err)
+			require.NoError(t, services.StartAndAwaitRunning(context.Background(), c))
+			defer services.StopAndAwaitTerminated(context.Background(), c) //nolint:errcheck
 
 			// Init context used by the test
 			ctx, cancel := context.WithTimeout(ctxUser1, time.Second)
diff --git a/pkg/ingester/client/pool_test.go b/pkg/ingester/client/pool_test.go
index 704e62ade9..19d2c92107 100644
--- a/pkg/ingester/client/pool_test.go
+++ b/pkg/ingester/client/pool_test.go
@@ -14,6 +14,7 @@ import (
 	"google.golang.org/grpc/health/grpc_health_v1"
 
 	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 type mockIngester struct {
@@ -80,7 +81,8 @@ func TestIngesterCache(t *testing.T) {
 		RemoteTimeout:       50 * time.Millisecond,
 		ClientCleanupPeriod: 10 * time.Second,
 	}, mockReadRing{}, factory, log.NewNopLogger())
-	defer pool.Stop()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), pool))
+	defer services.StopAndAwaitTerminated(context.Background(), pool) //nolint:errcheck
 
 	_, err := pool.GetClientFor("1")
 	require.NoError(t, err)
diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go
index cc690dcecd..b9e9740057 100644
--- a/pkg/ingester/ingester_test.go
+++ b/pkg/ingester/ingester_test.go
@@ -26,6 +26,7 @@ import (
 	promchunk "github.com/cortexproject/cortex/pkg/chunk/encoding"
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/util/chunkcompat"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
@@ -44,6 +45,7 @@ func newTestStore(t require.TestingT, cfg Config, clientConfig client.Config, li
 
 	ing, err := New(cfg, clientConfig, overrides, store, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))
 
 	return store, ing
 }
@@ -201,7 +203,7 @@ func TestIngesterAppend(t *testing.T) {
 	retrieveTestSamples(t, ing, userIDs, testData)
 
 	// Read samples back via chunk store.
-	ing.Shutdown()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
 	store.checkData(t, userIDs, testData)
 }
 
@@ -227,7 +229,7 @@ func TestIngesterSendsOnlySeriesWithData(t *testing.T) {
 	}
 
 	// Read samples back via chunk store.
-	ing.Shutdown()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
 }
 
 func TestIngesterIdleFlush(t *testing.T) {
@@ -302,7 +304,7 @@ func (s *stream) Send(response *client.QueryStreamResponse) error {
 
 func TestIngesterAppendOutOfOrderAndDuplicate(t *testing.T) {
 	_, ing := newDefaultTestStore(t)
-	defer ing.Shutdown()
+	defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
 
 	m := labelPairs{
 		{Name: model.MetricNameLabel, Value: "testmetric"},
@@ -333,7 +335,7 @@ func TestIngesterAppendOutOfOrderAndDuplicate(t *testing.T) {
 // Test that blank labels are removed by the ingester
 func TestIngesterAppendBlankLabel(t *testing.T) {
 	_, ing := newDefaultTestStore(t)
-	defer ing.Shutdown()
+	defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
 
 	lp := labelPairs{
 		{Name: model.MetricNameLabel, Value: "testmetric"},
@@ -364,7 +366,7 @@ func TestIngesterUserSeriesLimitExceeded(t *testing.T) {
 	limits.MaxLocalSeriesPerUser = 1
 
 	_, ing := newTestStore(t, defaultIngesterTestConfig(), defaultClientTestConfig(), limits)
-	defer ing.Shutdown()
+	defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
 
 	userID := "1"
 	labels1 := labels.Labels{{Name: labels.MetricName, Value: "testmetric"}, {Name: "foo", Value: "bar"}}
@@ -421,7 +423,7 @@ func TestIngesterMetricSeriesLimitExceeded(t *testing.T) {
 	limits.MaxLocalSeriesPerMetric = 1
 
 	_, ing := newTestStore(t, defaultIngesterTestConfig(), defaultClientTestConfig(), limits)
-	defer ing.Shutdown()
+	defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
 
 	userID := "1"
 	labels1 := labels.Labels{{Name: labels.MetricName, Value: "testmetric"}, {Name: "foo", Value: "bar"}}
@@ -485,7 +487,7 @@ func BenchmarkIngesterSeriesCreationLocking(b *testing.B) {
 
 func benchmarkIngesterSeriesCreationLocking(b *testing.B, parallelism int) {
 	_, ing := newDefaultTestStore(b)
-	defer ing.Shutdown()
+	defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
 
 	var (
 		wg     sync.WaitGroup
@@ -582,7 +584,7 @@ func benchmarkIngesterPush(b *testing.B, limits validation.Limits, errorsExpecte
 						require.NoError(b, err)
 					}
 				}
-				ing.Shutdown()
+				_ = services.StopAndAwaitTerminated(context.Background(), ing)
 			}
 		})
 	}
diff --git a/pkg/ingester/ingester_v2_test.go b/pkg/ingester/ingester_v2_test.go
index feabef6403..d0dced283b 100644
--- a/pkg/ingester/ingester_v2_test.go
+++ b/pkg/ingester/ingester_v2_test.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/test"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
@@ -205,7 +206,8 @@ func TestIngester_v2Push(t *testing.T) {
 
 			i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, registry)
 			require.NoError(t, err)
-			defer i.Shutdown()
+			require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+			defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 			defer cleanup()
 
 			ctx := user.InjectOrgID(context.Background(), userID)
@@ -256,7 +258,8 @@ func TestIngester_v2Push_ShouldHandleTheCaseTheCachedReferenceIsInvalid(t *testi
 
 	i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	ctx := user.InjectOrgID(context.Background(), userID)
@@ -321,7 +324,8 @@ func TestIngester_v2Push_ShouldCorrectlyTrackMetricsInMultiTenantScenario(t *tes
 
 	i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, registry)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until the ingester is ACTIVE
@@ -392,7 +396,8 @@ func Test_Ingester_v2LabelNames(t *testing.T) {
 	// Create ingester
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until it's ACTIVE
@@ -436,7 +441,8 @@ func Test_Ingester_v2LabelValues(t *testing.T) {
 	// Create ingester
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until it's ACTIVE
@@ -555,7 +561,8 @@ func Test_Ingester_v2Query(t *testing.T) {
 	// Create ingester
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until it's ACTIVE
@@ -590,7 +597,8 @@ func Test_Ingester_v2Query(t *testing.T) {
 func TestIngester_v2Query_ShouldNotCreateTSDBIfDoesNotExists(t *testing.T) {
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Mock request
@@ -610,7 +618,8 @@ func TestIngester_v2Query_ShouldNotCreateTSDBIfDoesNotExists(t *testing.T) {
 func TestIngester_v2LabelValues_ShouldNotCreateTSDBIfDoesNotExists(t *testing.T) {
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Mock request
@@ -630,7 +639,8 @@ func TestIngester_v2LabelValues_ShouldNotCreateTSDBIfDoesNotExists(t *testing.T)
 func TestIngester_v2LabelNames_ShouldNotCreateTSDBIfDoesNotExists(t *testing.T) {
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Mock request
@@ -655,7 +665,8 @@ func TestIngester_v2Push_ShouldNotCreateTSDBIfNotInActiveState(t *testing.T) {
 
 	i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 	require.Equal(t, ring.PENDING, i.lifecycler.GetState())
 
@@ -703,7 +714,8 @@ func TestIngester_getOrCreateTSDB_ShouldNotAllowToCreateTSDBIfIngesterStateIsNot
 
 			i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, nil)
 			require.NoError(t, err)
-			defer i.Shutdown()
+			require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+			defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 			defer cleanup()
 
 			// Switch ingester state to the expected one in the test
@@ -848,7 +860,8 @@ func Test_Ingester_v2MetricsForLabelMatchers(t *testing.T) {
 	// Create ingester
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until it's ACTIVE
@@ -1008,8 +1021,9 @@ func TestIngester_v2LoadTSDBOnStartup(t *testing.T) {
 
 			ingester, err := NewV2(ingesterCfg, clientCfg, overrides, nil)
 			require.NoError(t, err)
+			require.NoError(t, services.StartAndAwaitRunning(context.Background(), ingester))
 
-			defer ingester.Shutdown()
+			defer services.StopAndAwaitTerminated(context.Background(), ingester) //nolint:errcheck
 
 			testData.check(t, ingester)
 		})
@@ -1024,7 +1038,8 @@ func TestIngester_shipBlocks(t *testing.T) {
 	// Create ingester
 	i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until it's ACTIVE
@@ -1047,7 +1062,7 @@ func TestIngester_shipBlocks(t *testing.T) {
 	}
 
 	// Ship blocks and assert on the mocked shipper
-	i.shipBlocks()
+	i.shipBlocks(context.Background())
 
 	for _, m := range mocks {
 		m.AssertNumberOfCalls(t, "Sync", 1)
@@ -1078,7 +1093,8 @@ func Test_Ingester_v2UserStats(t *testing.T) {
 	// Create ingester
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until it's ACTIVE
@@ -1126,7 +1142,8 @@ func Test_Ingester_v2AllUserStats(t *testing.T) {
 	// Create ingester
 	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
 	require.NoError(t, err)
-	defer i.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
 	defer cleanup()
 
 	// Wait until it's ACTIVE
diff --git a/pkg/ingester/lifecycle_test.go b/pkg/ingester/lifecycle_test.go
index a794ad3eaa..343dc97500 100644
--- a/pkg/ingester/lifecycle_test.go
+++ b/pkg/ingester/lifecycle_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring/testutils"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/backend/s3"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/test"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
@@ -72,7 +73,8 @@ func TestIngesterRestart(t *testing.T) {
 	{
 		_, ingester := newTestStore(t, config, clientConfig, limits)
 		time.Sleep(100 * time.Millisecond)
-		ingester.Shutdown() // doesn't actually unregister due to skipUnregister: true
+		// doesn't actually unregister due to skipUnregister: true
+		require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ingester))
 	}
 
 	test.Poll(t, 100*time.Millisecond, 1, func() interface{} {
@@ -82,7 +84,8 @@ func TestIngesterRestart(t *testing.T) {
 	{
 		_, ingester := newTestStore(t, config, clientConfig, limits)
 		time.Sleep(100 * time.Millisecond)
-		ingester.Shutdown() // doesn't actually unregister due to skipUnregister: true
+		// doesn't actually unregister due to skipUnregister: true
+		require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ingester))
 	}
 
 	time.Sleep(200 * time.Millisecond)
@@ -104,6 +107,7 @@ func TestIngesterTransfer(t *testing.T) {
 	cfg1.MaxTransferRetries = 10
 	ing1, err := New(cfg1, defaultClientTestConfig(), limits, nil, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing1))
 
 	test.Poll(t, 100*time.Millisecond, ring.ACTIVE, func() interface{} {
 		return ing1.lifecycler.GetState()
@@ -123,6 +127,7 @@ func TestIngesterTransfer(t *testing.T) {
 	cfg2.LifecyclerConfig.JoinAfter = 100 * time.Second
 	ing2, err := New(cfg2, defaultClientTestConfig(), limits, nil, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing2))
 
 	// Let ing2 send chunks to ing1
 	ing1.cfg.ingesterClientFactory = func(addr string, _ client.Config) (client.HealthAndIngesterClient, error) {
@@ -132,7 +137,8 @@ func TestIngesterTransfer(t *testing.T) {
 	}
 
 	// Now stop the first ingester, and wait for the second ingester to become ACTIVE.
-	ing1.Shutdown()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing1))
+
 	test.Poll(t, 10*time.Second, ring.ACTIVE, func() interface{} {
 		return ing2.lifecycler.GetState()
 	})
@@ -168,6 +174,7 @@ func TestIngesterBadTransfer(t *testing.T) {
 	cfg.LifecyclerConfig.JoinAfter = 100 * time.Second
 	ing, err := New(cfg, defaultClientTestConfig(), limits, nil, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))
 
 	test.Poll(t, 100*time.Millisecond, ring.PENDING, func() interface{} {
 		return ing.lifecycler.GetState()
@@ -371,7 +378,7 @@ func TestIngesterFlush(t *testing.T) {
 	// Now stop the ingester.  Don't call shutdown, as it waits for all goroutines
 	// to exit.  We just want to check that by the time the token is removed from
 	// the ring, the data is in the chunk store.
-	ing.lifecycler.Shutdown()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing.lifecycler))
 	test.Poll(t, 200*time.Millisecond, 0, func() interface{} {
 		r, err := ing.lifecycler.KVStore.Get(context.Background(), ring.IngesterRingKey)
 		if err != nil {
@@ -439,6 +446,7 @@ func TestV2IngesterTransfer(t *testing.T) {
 			cfg1.MaxTransferRetries = 10
 			ing1, err := New(cfg1, defaultClientTestConfig(), limits, nil, nil)
 			require.NoError(t, err)
+			require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing1))
 
 			test.Poll(t, 100*time.Millisecond, ring.ACTIVE, func() interface{} {
 				return ing1.lifecycler.GetState()
@@ -466,6 +474,7 @@ func TestV2IngesterTransfer(t *testing.T) {
 			cfg2.LifecyclerConfig.JoinAfter = 100 * time.Second
 			ing2, err := New(cfg2, defaultClientTestConfig(), limits, nil, nil)
 			require.NoError(t, err)
+			require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing2))
 
 			// Let ing1 send blocks/wal to ing2
 			ingesterClientFactoryCount := 0
@@ -481,7 +490,7 @@ func TestV2IngesterTransfer(t *testing.T) {
 			}
 
 			// Now stop the first ingester, and wait for the second ingester to become ACTIVE.
-			ing1.Shutdown()
+			require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing1))
 			test.Poll(t, 10*time.Second, ring.ACTIVE, func() interface{} {
 				return ing2.lifecycler.GetState()
 			})
diff --git a/pkg/ingester/wal_test.go b/pkg/ingester/wal_test.go
index a48c5f816a..756504dddb 100644
--- a/pkg/ingester/wal_test.go
+++ b/pkg/ingester/wal_test.go
@@ -1,11 +1,14 @@
 package ingester
 
 import (
+	"context"
 	"io/ioutil"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 func TestWAL(t *testing.T) {
@@ -26,7 +29,7 @@ func TestWAL(t *testing.T) {
 	// Build an ingester, add some samples, then shut it down.
 	_, ing := newTestStore(t, cfg, defaultClientTestConfig(), defaultLimitsTestConfig())
 	userIDs, testData := pushTestSamples(t, ing, numSeries, numSamplesPerSeriesPerPush, 0)
-	ing.Shutdown()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
 
 	for r := 0; r < numRestarts; r++ {
 		if r == numRestarts-1 {
@@ -46,6 +49,6 @@ func TestWAL(t *testing.T) {
 			userIDs, testData = pushTestSamples(t, ing, numSeries, numSamplesPerSeriesPerPush, (r+1)*numSamplesPerSeriesPerPush)
 		}
 
-		ing.Shutdown()
+		require.NoError(t, services.StopAndAwaitTerminated(context.Background(), ing))
 	}
 }
diff --git a/pkg/querier/block_store_test.go b/pkg/querier/block_store_test.go
index e65d9e87f9..c9266e2ab5 100644
--- a/pkg/querier/block_store_test.go
+++ b/pkg/querier/block_store_test.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 func TestUserStore_InitialSync(t *testing.T) {
@@ -64,11 +65,12 @@ func TestUserStore_InitialSync(t *testing.T) {
 			testData.setup(bucketClient)
 
 			us, err := NewUserStore(cfg, bucketClient, mockLoggingLevel(), log.NewNopLogger(), nil)
-			if us != nil {
-				defer us.Stop()
+			if err == nil {
+				err = services.StartAndAwaitRunning(context.Background(), us)
+				defer services.StopAndAwaitTerminated(context.Background(), us) //nolint:errcheck
 			}
 
-			require.Equal(t, err, testData.expectedErr)
+			require.Equal(t, testData.expectedErr, err)
 			bucketClient.AssertNumberOfCalls(t, "Iter", testData.expectedIter)
 		})
 	}
@@ -87,7 +89,8 @@ func TestUserStore_syncUserStores(t *testing.T) {
 
 	us, err := NewUserStore(cfg, bucketClient, mockLoggingLevel(), log.NewNopLogger(), nil)
 	require.NoError(t, err)
-	defer us.Stop()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), us))
+	defer services.StopAndAwaitTerminated(context.Background(), us) //nolint:errcheck
 
 	// Sync user stores and count the number of times the callback is called.
 	storesCount := int32(0)
diff --git a/pkg/ring/lifecycler_test.go b/pkg/ring/lifecycler_test.go
index 3bfcae4520..49824f122f 100644
--- a/pkg/ring/lifecycler_test.go
+++ b/pkg/ring/lifecycler_test.go
@@ -13,6 +13,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/test"
 )
 
@@ -58,7 +59,8 @@ func TestLifecycler_HealthyInstancesCount(t *testing.T) {
 
 	r, err := New(ringConfig, "ingester", IngesterRingKey)
 	require.NoError(t, err)
-	defer r.Stop()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), r))
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 	// Add the first ingester to the ring
 	lifecyclerConfig1 := testLifecyclerConfig(ringConfig, "ing1")
@@ -69,7 +71,7 @@ func TestLifecycler_HealthyInstancesCount(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, 0, lifecycler1.HealthyInstancesCount())
 
-	lifecycler1.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), lifecycler1))
 
 	// Assert the first ingester joined the ring
 	test.Poll(t, 1000*time.Millisecond, true, func() interface{} {
@@ -85,7 +87,7 @@ func TestLifecycler_HealthyInstancesCount(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, 0, lifecycler2.HealthyInstancesCount())
 
-	lifecycler2.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), lifecycler2))
 
 	// Assert the second ingester joined the ring
 	test.Poll(t, 1000*time.Millisecond, true, func() interface{} {
@@ -107,14 +109,15 @@ func TestLifecycler_NilFlushTransferer(t *testing.T) {
 	// Create a lifecycler with nil FlushTransferer to make sure it operates correctly
 	lifecycler, err := NewLifecycler(lifecyclerConfig, nil, "ingester", IngesterRingKey, true)
 	require.NoError(t, err)
-	lifecycler.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), lifecycler))
 
 	// Ensure the lifecycler joined the ring
 	test.Poll(t, time.Second, 1, func() interface{} {
 		return lifecycler.HealthyInstancesCount()
 	})
 
-	lifecycler.Shutdown()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), lifecycler))
+
 	assert.Equal(t, 0, lifecycler.HealthyInstancesCount())
 }
 
@@ -130,13 +133,13 @@ func TestLifecycler_TwoRingsWithDifferentKeysOnTheSameKVStore(t *testing.T) {
 
 	lifecycler1, err := NewLifecycler(lifecyclerConfig1, nil, "service-1", "ring-1", true)
 	require.NoError(t, err)
-	lifecycler1.Start()
-	defer lifecycler1.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), lifecycler1))
+	defer services.StopAndAwaitTerminated(context.Background(), lifecycler1) //nolint:errcheck
 
 	lifecycler2, err := NewLifecycler(lifecyclerConfig2, nil, "service-2", "ring-2", true)
 	require.NoError(t, err)
-	lifecycler2.Start()
-	defer lifecycler2.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), lifecycler2))
+	defer services.StopAndAwaitTerminated(context.Background(), lifecycler2) //nolint:errcheck
 
 	// Ensure each lifecycler reports 1 healthy instance, because they're
 	// in a different ring
@@ -165,13 +168,14 @@ func TestRingRestart(t *testing.T) {
 
 	r, err := New(ringConfig, "ingester", IngesterRingKey)
 	require.NoError(t, err)
-	defer r.Stop()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), r))
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 	// Add an 'ingester' with normalised tokens.
 	lifecyclerConfig1 := testLifecyclerConfig(ringConfig, "ing1")
 	l1, err := NewLifecycler(lifecyclerConfig1, &nopFlushTransferer{}, "ingester", IngesterRingKey, true)
 	require.NoError(t, err)
-	l1.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), l1))
 
 	// Check this ingester joined, is active, and has one token.
 	test.Poll(t, 1000*time.Millisecond, true, func() interface{} {
@@ -185,7 +189,7 @@ func TestRingRestart(t *testing.T) {
 	// Add a second ingester with the same settings, so it will think it has restarted
 	l2, err := NewLifecycler(lifecyclerConfig1, &nopFlushTransferer{}, "ingester", IngesterRingKey, true)
 	require.NoError(t, err)
-	l2.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), l2))
 
 	// Check the new ingester picked up the same token
 	test.Poll(t, 1000*time.Millisecond, true, func() interface{} {
@@ -241,12 +245,17 @@ func TestCheckReady(t *testing.T) {
 
 	r, err := New(ringConfig, "ingester", IngesterRingKey)
 	require.NoError(t, err)
-	defer r.Stop()
+	require.NoError(t, r.StartAsync(context.Background()))
+	// This is very atypical, but if we used AwaitRunning, that would fail, because of how quickly service terminates ...
+	// by the time we check for Running state, it is already terminated, because mock ring has no WatchFunc, so it
+	// will just exit.
+	require.NoError(t, r.AwaitTerminated(context.Background()))
+
 	cfg := testLifecyclerConfig(ringConfig, "ring1")
 	cfg.MinReadyDuration = 1 * time.Nanosecond
 	l1, err := NewLifecycler(cfg, &nopFlushTransferer{}, "ingester", IngesterRingKey, true)
-	l1.Start()
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), l1))
 
 	l1.setTokens(Tokens([]uint32{1}))
 
@@ -269,7 +278,8 @@ func TestTokensOnDisk(t *testing.T) {
 
 	r, err := New(ringConfig, "ingester", IngesterRingKey)
 	require.NoError(t, err)
-	defer r.Stop()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), r))
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 	tokenDir, err := ioutil.TempDir(os.TempDir(), "tokens_on_disk")
 	require.NoError(t, err)
@@ -284,7 +294,8 @@ func TestTokensOnDisk(t *testing.T) {
 	// Start first ingester.
 	l1, err := NewLifecycler(lifecyclerConfig, &noopFlushTransferer{}, "ingester", IngesterRingKey, true)
 	require.NoError(t, err)
-	l1.Start()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), l1))
+
 	// Check this ingester joined, is active, and has 512 token.
 	var expTokens []uint32
 	test.Poll(t, 1000*time.Millisecond, true, func() interface{} {
@@ -300,14 +311,14 @@ func TestTokensOnDisk(t *testing.T) {
 			len(desc.Ingesters["ing1"].Tokens) == 512
 	})
 
-	l1.Shutdown()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), l1))
 
 	// Start new ingester at same token directory.
 	lifecyclerConfig.ID = "ing2"
 	l2, err := NewLifecycler(lifecyclerConfig, &noopFlushTransferer{}, "ingester", IngesterRingKey, true)
 	require.NoError(t, err)
-	l2.Start()
-	defer l2.Shutdown()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), l2))
+	defer services.StopAndAwaitTerminated(context.Background(), l2) //nolint:errcheck
 
 	// Check this ingester joined, is active, and has 512 token.
 	var actTokens []uint32
@@ -341,7 +352,8 @@ func TestJoinInLeavingState(t *testing.T) {
 
 	r, err := New(ringConfig, "ingester", IngesterRingKey)
 	require.NoError(t, err)
-	defer r.Stop()
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), r))
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 	cfg := testLifecyclerConfig(ringConfig, "ing1")
 	cfg.NumTokens = 2
@@ -366,8 +378,8 @@ func TestJoinInLeavingState(t *testing.T) {
 	require.NoError(t, err)
 
 	l1, err := NewLifecycler(cfg, &nopFlushTransferer{}, "ingester", IngesterRingKey, true)
-	l1.Start()
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), l1))
 
 	// Check that the lifecycler was able to join after coming up in LEAVING
 	test.Poll(t, 1000*time.Millisecond, true, func() interface{} {
diff --git a/pkg/ring/replication_strategy_test.go b/pkg/ring/replication_strategy_test.go
index 0b972b8c03..bf23d8e4dd 100644
--- a/pkg/ring/replication_strategy_test.go
+++ b/pkg/ring/replication_strategy_test.go
@@ -1,6 +1,7 @@
 package ring
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -10,6 +11,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ring/kv"
 	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 func TestReplicationStrategy(t *testing.T) {
@@ -93,6 +95,8 @@ func TestReplicationStrategy(t *testing.T) {
 			ReplicationFactor: tc.RF,
 		}, "ingester", IngesterRingKey)
 		require.NoError(t, err)
+		require.NoError(t, services.StartAndAwaitRunning(context.Background(), r))
+		defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 		t.Run(fmt.Sprintf("[%d]", i), func(t *testing.T) {
 			liveIngesters, maxFailure, err := r.replicationStrategy(ingesters, tc.op)
diff --git a/pkg/ruler/api_test.go b/pkg/ruler/api_test.go
index 91ecdec595..924faaf2ba 100644
--- a/pkg/ruler/api_test.go
+++ b/pkg/ruler/api_test.go
@@ -1,6 +1,7 @@
 package ruler
 
 import (
+	"context"
 	"encoding/json"
 	"io/ioutil"
 	"net/http"
@@ -9,6 +10,8 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/user"
+
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 func TestRuler_rules(t *testing.T) {
@@ -17,7 +20,7 @@ func TestRuler_rules(t *testing.T) {
 
 	r, rcleanup := newTestRuler(t, cfg)
 	defer rcleanup()
-	defer r.Stop()
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 	req := httptest.NewRequest("GET", "https://localhost:8080/api/prom/api/v1/rules", nil)
 	req.Header.Add(user.OrgIDHeaderName, "user1")
@@ -73,7 +76,7 @@ func TestRuler_alerts(t *testing.T) {
 
 	r, rcleanup := newTestRuler(t, cfg)
 	defer rcleanup()
-	defer r.Stop()
+	defer r.StopAsync()
 
 	req := httptest.NewRequest("GET", "https://localhost:8080/api/prom/api/v1/alerts", nil)
 	req.Header.Add(user.OrgIDHeaderName, "user1")
diff --git a/pkg/ruler/lifecycle_test.go b/pkg/ruler/lifecycle_test.go
index 62abbbb4d7..5ba3922145 100644
--- a/pkg/ruler/lifecycle_test.go
+++ b/pkg/ruler/lifecycle_test.go
@@ -1,13 +1,16 @@
 package ruler
 
 import (
+	"context"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/ring/testutils"
+	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/test"
 )
 
@@ -26,7 +29,7 @@ func TestRulerShutdown(t *testing.T) {
 		return testutils.NumTokens(config.Ring.KVStore.Mock, "localhost", ring.RulerRingKey)
 	})
 
-	r.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), r))
 
 	// Wait until the tokens are unregistered from the ring
 	test.Poll(t, 100*time.Millisecond, 0, func() interface{} {
@@ -50,7 +53,8 @@ func TestRulerRestart(t *testing.T) {
 	})
 
 	// Stop the ruler. Doesn't actually unregister due to skipUnregister: true
-	r.Stop()
+	r.StopAsync()
+	require.NoError(t, r.AwaitTerminated(context.Background()))
 
 	// We expect the tokens are preserved in the ring.
 	assert.Equal(t, config.Ring.NumTokens, testutils.NumTokens(config.Ring.KVStore.Mock, "localhost", ring.RulerRingKey))
@@ -58,7 +62,7 @@ func TestRulerRestart(t *testing.T) {
 	// Create a new ruler which is expected to pick up tokens from the ring.
 	r, rcleanup = newTestRuler(t, config)
 	defer rcleanup()
-	defer r.Stop()
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 	// Wait until the ruler is ACTIVE in the ring.
 	test.Poll(t, 100*time.Millisecond, ring.ACTIVE, func() interface{} {
diff --git a/pkg/ruler/ruler_test.go b/pkg/ruler/ruler_test.go
index a679a885b0..1edcd232d3 100644
--- a/pkg/ruler/ruler_test.go
+++ b/pkg/ruler/ruler_test.go
@@ -28,6 +28,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ruler/rules"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 func defaultRulerConfig(store rules.RuleStore) (Config, func()) {
@@ -81,6 +82,7 @@ func newTestRuler(t *testing.T, cfg Config) (*Ruler, func()) {
 	l = level.NewFilter(l, level.AllowInfo())
 	ruler, err := NewRuler(cfg, engine, noopQueryable, pusher, prometheus.NewRegistry(), l)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), ruler))
 
 	// Ensure all rules are loaded before usage
 	ruler.loadRules(context.Background())
@@ -111,7 +113,8 @@ func TestNotifierSendsUserIDHeader(t *testing.T) {
 
 	r, rcleanup := newTestRuler(t, cfg)
 	defer rcleanup()
-	defer r.Stop()
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
+
 	n, err := r.getOrCreateNotifier("1")
 	require.NoError(t, err)
 
@@ -135,7 +138,7 @@ func TestRuler_Rules(t *testing.T) {
 
 	r, rcleanup := newTestRuler(t, cfg)
 	defer rcleanup()
-	defer r.Stop()
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
 
 	// test user1
 	ctx := user.InjectOrgID(context.Background(), "user1")
diff --git a/pkg/util/runtimeconfig/manager_test.go b/pkg/util/runtimeconfig/manager_test.go
index e72d734b23..62041562e6 100644
--- a/pkg/util/runtimeconfig/manager_test.go
+++ b/pkg/util/runtimeconfig/manager_test.go
@@ -1,6 +1,7 @@
 package runtimeconfig
 
 import (
+	"context"
 	"io/ioutil"
 	"os"
 	"testing"
@@ -9,6 +10,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
 	"gopkg.in/yaml.v2"
+
+	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
 type TestLimits struct {
@@ -76,9 +79,10 @@ func TestNewOverridesManager(t *testing.T) {
 
 	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), overridesManager))
 
 	// Cleaning up
-	overridesManager.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), overridesManager))
 
 	// Make sure test limits were loaded.
 	require.NotNil(t, overridesManager.GetConfig())
@@ -110,6 +114,7 @@ func TestOverridesManager_ListenerWithDefaultLimits(t *testing.T) {
 
 	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), overridesManager))
 
 	// need to use buffer, otherwise loadConfig will throw away update
 	ch := overridesManager.CreateListenerChannel(1)
@@ -137,7 +142,7 @@ func TestOverridesManager_ListenerWithDefaultLimits(t *testing.T) {
 	require.Equal(t, 100, to.Overrides["user2"].Limit1) // from defaults
 
 	// Cleaning up
-	overridesManager.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), overridesManager))
 
 	// Make sure test limits were loaded.
 	require.NotNil(t, overridesManager.GetConfig())
@@ -158,6 +163,7 @@ func TestOverridesManager_ListenerChannel(t *testing.T) {
 
 	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), overridesManager))
 
 	// need to use buffer, otherwise loadConfig will throw away update
 	ch := overridesManager.CreateListenerChannel(1)
@@ -207,11 +213,12 @@ func TestOverridesManager_StopClosesListenerChannels(t *testing.T) {
 
 	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), overridesManager))
 
 	// need to use buffer, otherwise loadConfig will throw away update
 	ch := overridesManager.CreateListenerChannel(0)
 
-	overridesManager.Stop()
+	require.NoError(t, services.StopAndAwaitTerminated(context.Background(), overridesManager))
 
 	select {
 	case _, ok := <-ch:
diff --git a/pkg/util/services/services_test.go b/pkg/util/services/services_test.go
index eb061cd4d4..6706ced8a3 100644
--- a/pkg/util/services/services_test.go
+++ b/pkg/util/services/services_test.go
@@ -2,6 +2,7 @@ package services
 
 import (
 	"context"
+	"errors"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -10,6 +11,8 @@ import (
 )
 
 func TestIdleService(t *testing.T) {
+	t.Parallel()
+
 	started := false
 	stopped := false
 
@@ -36,6 +39,8 @@ func TestIdleService(t *testing.T) {
 }
 
 func TestTimerService(t *testing.T) {
+	t.Parallel()
+
 	iterations := int64(0)
 
 	s := NewTimerService(100*time.Millisecond, nil, func(ctx context.Context) error {
@@ -59,3 +64,67 @@ func TestTimerService(t *testing.T) {
 	require.NoError(t, s.AwaitTerminated(context.Background()))
 	require.Equal(t, Terminated, s.State())
 }
+
+func TestHelperFunctionsNoError(t *testing.T) {
+	t.Parallel()
+
+	s := NewIdleService(nil, nil)
+	require.NoError(t, StartAndAwaitRunning(context.Background(), s))
+	require.NoError(t, StopAndAwaitTerminated(context.Background(), s))
+}
+
+func TestHelperFunctionsStartError(t *testing.T) {
+	t.Parallel()
+
+	e := errors.New("some error")
+	s := NewIdleService(func(serviceContext context.Context) error { return e }, nil)
+
+	require.Equal(t, e, StartAndAwaitRunning(context.Background(), s))
+	require.Equal(t, e, StopAndAwaitTerminated(context.Background(), s))
+}
+
+func TestHelperFunctionsStartTooSlow(t *testing.T) {
+	t.Parallel()
+
+	s := NewIdleService(func(serviceContext context.Context) error {
+		// ignores context
+		time.Sleep(1 * time.Second)
+		return nil
+	}, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go func() {
+		// cancel context early
+		time.Sleep(100 * time.Millisecond)
+		cancel()
+	}()
+
+	require.Equal(t, context.Canceled, StartAndAwaitRunning(ctx, s))
+	require.NoError(t, StopAndAwaitTerminated(context.Background(), s))
+}
+
+func TestHelperFunctionsStopError(t *testing.T) {
+	t.Parallel()
+
+	e := errors.New("some error")
+	s := NewIdleService(nil, func() error { return e })
+
+	require.NoError(t, StartAndAwaitRunning(context.Background(), s))
+	require.Equal(t, e, StopAndAwaitTerminated(context.Background(), s))
+}
+
+func TestHelperFunctionsStopTooSlow(t *testing.T) {
+	t.Parallel()
+
+	s := NewIdleService(nil, func() error { time.Sleep(1 * time.Second); return nil })
+
+	require.NoError(t, StartAndAwaitRunning(context.Background(), s))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	require.Equal(t, context.DeadlineExceeded, StopAndAwaitTerminated(ctx, s))
+	require.NoError(t, StopAndAwaitTerminated(context.Background(), s))
+}