feat compact: added readiness Prober

CHANGELOG.md

@@ -18,6 +18,7 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 ### Changed
 
 - [#1338](https://github.com/thanos-io/thanos/pull/1338) Querier still warns on store API duplicate, but allows a single one from duplicated set. This is gracefully warn about the problematic logic and not disrupt immediately.
+- [#1297](https://github.com/improbable-eng/thanos/pull/1297) Added `/-/ready` and `/-/healthy` endpoints to Thanos compact.
 
 ### Fixed
 

cmd/thanos/compact.go

@@ -15,18 +15,20 @@ import (
  "github.com/go-kit/kit/log"
  "github.com/go-kit/kit/log/level"
  "github.com/oklog/run"
- opentracing "github.com/opentracing/opentracing-go"
+ "github.com/opentracing/opentracing-go"
  "github.com/pkg/errors"
  "github.com/prometheus/client_golang/prometheus"
  "github.com/prometheus/tsdb"
  "github.com/thanos-io/thanos/pkg/block"
  "github.com/thanos-io/thanos/pkg/block/metadata"
  "github.com/thanos-io/thanos/pkg/compact"
  "github.com/thanos-io/thanos/pkg/compact/downsample"
+ "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"
+ "gopkg.in/alecthomas/kingpin.v2"
 )
 
 var (
@@ -49,7 +51,7 @@ func (cs compactionSet) String() string {
  return strings.Join(result, ", ")
 }
 
-// levels returns set of compaction levels not higher than specified max compaction level
+// levels returns set of compaction levels not higher than specified max compaction level.
 func (cs compactionSet) levels(maxLevel int) ([]int64, error) {
  if maxLevel >= len(cs) {
  return nil, errors.Errorf("level is bigger then default set of %d", len(cs))
@@ -62,13 +64,14 @@ func (cs compactionSet) levels(maxLevel int) ([]int64, error) {
  return levels, nil
 }
 
-// maxLevel returns max available compaction level
+// maxLevel returns max available compaction level.
 func (cs compactionSet) maxLevel() int {
  return len(cs) - 1
 }
 
-func registerCompact(m map[string]setupFunc, app *kingpin.Application, name string) {
- cmd := app.Command(name, "continuously compacts blocks in an object store bucket")
+func registerCompact(m map[string]setupFunc, app *kingpin.Application) {
+ comp := component.Compact
+ cmd := app.Command(comp.String(), "continuously compacts blocks in an object store bucket")
 
  haltOnError := cmd.Flag("debug.halt-on-error", "Halt the process if a critical compaction error is detected.").
  Hidden().Default("true").Bool()
@@ -110,7 +113,7 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application, name stri
  compactionConcurrency := cmd.Flag("compact.concurrency", "Number of goroutines to use when compacting groups.").
  Default("1").Int()
 
- m[name] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ bool) error {
+ m[comp.String()] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ bool) error {
  return runCompact(g, logger, reg,
  *httpAddr,
  *dataDir,
@@ -125,7 +128,7 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application, name stri
  compact.ResolutionLevel5m: time.Duration(*retention5m),
  compact.ResolutionLevel1h: time.Duration(*retention1h),
  },
- name,
+ comp,
  *disableDownsampling,
  *maxCompactionLevel,
  *blockSyncConcurrency,
@@ -147,7 +150,7 @@ func runCompact(
  wait bool,
  generateMissingIndexCacheFiles bool,
  retentionByResolution map[compact.ResolutionLevel]time.Duration,
- component string,
+ component component.Component,
  disableDownsampling bool,
  maxCompactionLevel int,
  blockSyncConcurrency int,
@@ -168,12 +171,18 @@ func runCompact(
 
  downsampleMetrics := newDownsampleMetrics(reg)
 
+ readinessProber := prober.NewProber(component, logger, prometheus.WrapRegistererWithPrefix("thanos_", reg))
+ // Initiate default HTTP listener providing metrics endpoint and readiness/liveness probes.
+ if err := defaultHTTPListener(g, logger, reg, httpBindAddr, readinessProber); err != nil {
+ return errors.Wrap(err, "create readiness prober")
+ }
+
  confContentYaml, err := objStoreConfig.Content()
  if err != nil {
  return err
  }
 
- bkt, err := client.NewBucket(logger, confContentYaml, reg, component)
+ bkt, err := client.NewBucket(logger, confContentYaml, reg, component.String())
  if err != nil {
  return err
  }
@@ -318,11 +327,8 @@ func runCompact(
  cancel()
  })
 
- if err := metricHTTPListenGroup(g, logger, reg, httpBindAddr); err != nil {
- return err
- }
-
  level.Info(logger).Log("msg", "starting compact node")
+ readinessProber.SetReady()
  return nil
 }
 

cmd/thanos/main.go

@@ -31,6 +31,7 @@ import (
  "github.com/prometheus/client_golang/prometheus"
  "github.com/prometheus/client_golang/prometheus/promhttp"
  "github.com/prometheus/common/version"
+ "github.com/thanos-io/thanos/pkg/prober"
  "github.com/thanos-io/thanos/pkg/runutil"
  "github.com/thanos-io/thanos/pkg/tracing"
  "github.com/thanos-io/thanos/pkg/tracing/client"
@@ -73,7 +74,7 @@ func main() {
  registerStore(cmds, app, "store")
  registerQuery(cmds, app, "query")
  registerRule(cmds, app, "rule")
- registerCompact(cmds, app, "compact")
+ registerCompact(cmds, app)
  registerBucket(cmds, app, "bucket")
  registerDownsample(cmds, app, "downsample")
  registerReceive(cmds, app, "receive")
@@ -122,7 +123,7 @@ func main() {
  )
 
  prometheus.DefaultRegisterer = metrics
- // Memberlist uses go-metrics
+ // Memberlist uses go-metrics.
  sink, err := gprom.NewPrometheusSink()
  if err != nil {
  fmt.Fprintln(os.Stderr, errors.Wrapf(err, "%s command failed", cmd))
@@ -311,6 +312,7 @@ func defaultGRPCServerOpts(logger log.Logger, reg *prometheus.Registry, tracer o
  return append(opts, grpc.Creds(credentials.NewTLS(tlsCfg))), nil
 }
 
+// TODO Remove once all components are migrated to the new defaultHTTPListener.
 // metricHTTPListenGroup is a run.Group that servers HTTP endpoint with only Prometheus metrics.
 func metricHTTPListenGroup(g *run.Group, logger log.Logger, reg *prometheus.Registry, httpBindAddr string) error {
  mux := http.NewServeMux()
@@ -330,3 +332,27 @@ func metricHTTPListenGroup(g *run.Group, logger log.Logger, reg *prometheus.Regi
  })
  return nil
 }
+
+// defaultHTTPListener starts a run.Group that servers HTTP endpoint with default endpoints providing Prometheus metrics,
+// profiling and liveness/readiness probes.
+func defaultHTTPListener(g *run.Group, logger log.Logger, reg *prometheus.Registry, httpBindAddr string, readinessProber *prober.Prober) error {
+ mux := http.NewServeMux()
+ registerMetrics(mux, reg)
+ registerProfile(mux)
+ readinessProber.RegisterInMux(mux)
+
+ l, err := net.Listen("tcp", httpBindAddr)
+ if err != nil {
+ return errors.Wrap(err, "listen metrics address")
+ }
+
+ g.Add(func() error {
+ level.Info(logger).Log("msg", "listening for metrics", "address", httpBindAddr)
+ readinessProber.SetHealthy()
+ return errors.Wrap(http.Serve(l, mux), "serve metrics")
+ }, func(err error) {
+ readinessProber.SetNotHealthy(err)
+ runutil.CloseWithLogOnErr(logger, l, "metric listener")
+ })
+ return nil
+}

tutorials/kubernetes-demo/manifests/thanos-compactor.yaml

@@ -35,6 +35,14 @@ spec:
  ports:
  - name: http
  containerPort: 10902
+ livenessProbe:
+ httpGet:
+ port: 10902
+ path: /-/healthy
+ readinessProbe:
+ httpGet:
+ port: 10902
+ path: /-/ready
  resources:
  limits:
  cpu: "1"