Convert Cortex components to services

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`

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.

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

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)
 }

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,
 	)

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

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())
 }

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()