Compactor: remove malformed blocks after delay

cmd/thanos/compact.go

@@ -82,7 +82,7 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application, name stri
 
  objStoreConfig := regCommonObjStoreFlags(cmd, "", true)
 
- syncDelay := modelDuration(cmd.Flag("sync-delay", "Minimum age of fresh (non-compacted) blocks before they are being processed.").
+ syncDelay := modelDuration(cmd.Flag("sync-delay", "Minimum age of fresh (non-compacted) blocks before they are being processed. Blocks older than this which are malformed will be removed.").
  Default("30m"))
 
  retentionRaw := modelDuration(cmd.Flag("retention.resolution-raw", "How long to retain raw samples in bucket. 0d - disables this retention").Default("0d"))

docs/components/compact.md

@@ -61,7 +61,8 @@ Flags:
  Alternative to 'objstore.config-file' flag.
  Object store configuration in YAML.
  --sync-delay=30m Minimum age of fresh (non-compacted) blocks
- before they are being processed.
+ before they are being processed. Blocks older
+ than this which are malformed will be removed.
  --retention.resolution-raw=0d
  How long to retain raw samples in bucket. 0d -
  disables this retention

pkg/compact/compact.go

@@ -5,6 +5,7 @@ import (
  "fmt"
  "io/ioutil"
  "os"
+ "path"
  "path/filepath"
  "sort"
  "strings"
@@ -41,7 +42,7 @@ type Syncer struct {
  logger log.Logger
  reg prometheus.Registerer
  bkt objstore.Bucket
- syncDelay  time.Duration
+ consistencyDelay time.Duration
  mtx sync.Mutex
  blocks map[ulid.ULID]*metadata.Meta
  blocksMtx sync.Mutex
@@ -132,14 +133,14 @@ func newSyncerMetrics(reg prometheus.Registerer) *syncerMetrics {
 
 // NewSyncer returns a new Syncer for the given Bucket and directory.
 // Blocks must be at least as old as the sync delay for being considered.
-func NewSyncer(logger log.Logger, reg prometheus.Registerer, bkt objstore.Bucket, syncDelay time.Duration, blockSyncConcurrency int, acceptMalformedIndex bool) (*Syncer, error) {
+func NewSyncer(logger log.Logger, reg prometheus.Registerer, bkt objstore.Bucket, consistencyDelay time.Duration, blockSyncConcurrency int, acceptMalformedIndex bool) (*Syncer, error) {
  if logger == nil {
  logger = log.NewNopLogger()
  }
  return &Syncer{
  logger: logger,
  reg: reg,
- syncDelay:  syncDelay,
+ consistencyDelay: consistencyDelay,
  blocks: map[ulid.ULID]*metadata.Meta{},
  bkt: bkt,
  metrics: newSyncerMetrics(reg),
@@ -149,7 +150,8 @@ func NewSyncer(logger log.Logger, reg prometheus.Registerer, bkt objstore.Bucket
 }
 
 // SyncMetas synchronizes all meta files from blocks in the bucket into
-// the memory.
+// the memory. It removes any partial blocks older than consistencyDelay
+// from the bucket.
 func (c *Syncer) SyncMetas(ctx context.Context) error {
  c.mtx.Lock()
  defer c.mtx.Unlock()
@@ -194,6 +196,9 @@ func (c *Syncer) syncMetas(ctx context.Context) error {
  continue
  }
  if err != nil {
+ if removed := c.removeIfMetaMalformed(workCtx, id); removed {
+ continue
+ }
  errChan <- err
  return
  }
@@ -250,6 +255,10 @@ func (c *Syncer) downloadMeta(ctx context.Context, id ulid.ULID) (*metadata.Meta
 
  meta, err := block.DownloadMeta(ctx, c.logger, c.bkt, id)
  if err != nil {
+ if ulid.Now()-id.Time() < uint64(c.consistencyDelay/time.Millisecond) {
+ level.Debug(c.logger).Log("msg", "block is too fresh for now", "block", id)
+ return nil, blockTooFreshSentinelError
+ }
  return nil, errors.Wrapf(err, "downloading meta.json for %s", id)
  }
 
@@ -259,7 +268,7 @@ func (c *Syncer) downloadMeta(ctx context.Context, id ulid.ULID) (*metadata.Meta
  // - compactor created blocks
  // NOTE: It is not safe to miss "old" block (even that it is newly created) in sync step. Compactor needs to aware of ALL old blocks.
  // TODO(bplotka): https://github.com/improbable-eng/thanos/issues/377
- if ulid.Now()-id.Time() < uint64(c.syncDelay/time.Millisecond) &&
+ if ulid.Now()-id.Time() < uint64(c.consistencyDelay/time.Millisecond) &&
  meta.Thanos.Source != metadata.BucketRepairSource &&
  meta.Thanos.Source != metadata.CompactorSource &&
  meta.Thanos.Source != metadata.CompactorRepairSource {
@@ -271,6 +280,32 @@ func (c *Syncer) downloadMeta(ctx context.Context, id ulid.ULID) (*metadata.Meta
  return &meta, nil
 }
 
+// removeIfMalformed removes a block from the bucket if that block does not have a meta file.
+// It is the responsibility of the caller to ensure that enough time has passed for the block to become consistent.
+func (c *Syncer) removeIfMetaMalformed(ctx context.Context, id ulid.ULID) bool {
+ metaExists, err := c.bkt.Exists(ctx, path.Join(id.String(), block.MetaFilename))
+ if err != nil {
+ level.Warn(c.logger).Log("msg", "failed to check meta exists for block", "block", id, "err", err)
+ return false
+ }
+ if err != nil {
+ level.Warn(c.logger).Log("msg", "failed to check if block is malformed", "block", id, "err", err)
+ return false
+ }
+ if metaExists {
+ // Meta exists, block is not malformed.
+ return false
+ }
+
+ if err := block.Delete(ctx, c.bkt, id); err != nil {
+ level.Warn(c.logger).Log("msg", "failed to delete malformed block", "block", id, "err", err)
+ return false
+ }
+ level.Info(c.logger).Log("msg", "deleted malformed block", "block", id)
+
+ return true
+}
+
 // GroupKey returns a unique identifier for the group the block belongs to. It considers
 // the downsampling resolution and the block's labels.
 func GroupKey(meta metadata.Meta) string {