downsample: Add liveness and readiness probe

CHANGELOG.md

@@ -12,6 +12,7 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 ## Unreleased
 
 ## Added
+- [#1540](https://github.com/thanos-io/thanos/pull/1540) Added `/-/ready` and `/-/healthy` endpoints to Thanos Downsample.
 - [#1538](https://github.com/thanos-io/thanos/pull/1538) Added `/-/ready` and `/-/healthy` endpoints to Thanos Rule.
 - [#1537](https://github.com/thanos-io/thanos/pull/1537) Added `/-/ready` and `/-/healthy` endpoints to Thanos Receive.
 - [#1534](https://github.com/thanos-io/thanos/pull/1534) Added `/-/ready` endpoint to Thanos Query.

cmd/thanos/downsample.go

@@ -22,12 +22,14 @@ import (
 	"github.com/thanos-io/thanos/pkg/component"
 	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/objstore/client"
+	"github.com/thanos-io/thanos/pkg/prober"
 	"github.com/thanos-io/thanos/pkg/runutil"
 	kingpin "gopkg.in/alecthomas/kingpin.v2"
 )
 
-func registerDownsample(m map[string]setupFunc, app *kingpin.Application, name string) {
-	cmd := app.Command(name, "continuously downsamples blocks in an object store bucket")
+func registerDownsample(m map[string]setupFunc, app *kingpin.Application) {
+	comp := component.Downsample
+	cmd := app.Command(comp.String(), "continuously downsamples blocks in an object store bucket")
 
 	httpAddr := regHTTPAddrFlag(cmd)
 
@@ -36,8 +38,8 @@ func registerDownsample(m map[string]setupFunc, app *kingpin.Application, name s
 
 	objStoreConfig := regCommonObjStoreFlags(cmd, "", true)
 
-	m[name] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ bool) error {
-		return runDownsample(g, logger, reg, *httpAddr, *dataDir, objStoreConfig)
+	m[comp.String()] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ bool) error {
+		return runDownsample(g, logger, reg, *httpAddr, *dataDir, objStoreConfig, comp)
 	}
 }
 
@@ -71,6 +73,7 @@ func runDownsample(
 	httpBindAddr string,
 	dataDir string,
 	objStoreConfig *pathOrContent,
+	comp component.Component,
 ) error {
 	confContentYaml, err := objStoreConfig.Content()
 	if err != nil {
@@ -90,13 +93,14 @@ func runDownsample(
 	}()
 
 	metrics := newDownsampleMetrics(reg)
-
+	statusProber := prober.NewProber(comp, logger, prometheus.WrapRegistererWithPrefix("thanos_", reg))
 	// Start cycle of syncing blocks from the bucket and garbage collecting the bucket.
 	{
 		ctx, cancel := context.WithCancel(context.Background())
 
 		g.Add(func() error {
 			defer runutil.CloseWithLogOnErr(logger, bkt, "bucket client")
+			statusProber.SetReady()
 
 			level.Info(logger).Log("msg", "start first pass of downsampling")
 
@@ -116,8 +120,9 @@ func runDownsample(
 		})
 	}
 
-	if err := metricHTTPListenGroup(g, logger, reg, httpBindAddr); err != nil {
-		return err
+	// Initiate HTTP listener providing metrics endpoint and readiness/liveness probes.
+	if err := scheduleHTTPServer(g, logger, reg, statusProber, httpBindAddr, nil, comp); err != nil {
+		return errors.Wrap(err, "schedule HTTP server with probe")
 	}
 
 	level.Info(logger).Log("msg", "starting downsample node")

cmd/thanos/main.go

@@ -79,7 +79,7 @@ func main() {
 	registerRule(cmds, app)
 	registerCompact(cmds, app)
 	registerBucket(cmds, app, "bucket")
-	registerDownsample(cmds, app, "downsample")
+	registerDownsample(cmds, app)
 	registerReceive(cmds, app)
 	registerChecks(cmds, app, "check")