Merge remote-tracking branch 'upstream/master' into tokens-file

Author: Ganesh Vernekar

CHANGELOG.md

@@ -5,6 +5,7 @@
 * [CHANGE] The frontend component has been refactored to be easier to re-use. When upgrading the frontend, cache entries will be discarded and re-created with the new protobuf schema. #1734
 * [CHANGE] Remove direct DB/API access from the ruler
 * [CHANGE] Removed `Delta` encoding. Any old chunks with `Delta` encoding cannot be read anymore. If `ingester.chunk-encoding` is set to `Delta` the ingester will fail to start. #1706
+* [CHANGE] Setting `-ingester.max-transfer-retries` to 0 now disables hand-over when ingester is shutting down. Previously, zero meant infinite number of attempts. #1771
 * [FEATURE] Global limit on the max series per user and metric #1760
   * `-ingester.max-global-series-per-user`
   * `-ingester.max-global-series-per-metric`

docs/arguments.md

@@ -228,9 +228,9 @@ It also talks to a KVStore and has it's own copies of the same flags used by the
 
    How long to wait in PENDING state during the [hand-over process](ingester-handover.md). (default 0s)
 
-- `-ingester.ingester.max-transfer-retries`
+- `-ingester.max-transfer-retries`
 
-   How many times a LEAVING ingester tries to find a PENDING ingester during the [hand-over process](ingester-handover.md). Each attempt takes a second or so. (default 10)
+   How many times a LEAVING ingester tries to find a PENDING ingester during the [hand-over process](ingester-handover.md). Each attempt takes a second or so. Negative value or zero disables hand-over process completely. (default 10)
 
 - `-ingester.normalise-tokens`
 

pkg/chunk/chunk.go

@@ -288,7 +288,9 @@ func (c *Chunk) Decode(decodeContext *DecodeContext, input []byte) error {
 	if err != nil {
 		return errors.Wrap(err, "when decoding chunk metadata")
 	}
-	if len(input)-r.Len() != int(metadataLen) {
+	metadataRead := len(input) - r.Len()
+	// Older versions of Cortex included the initial length word; newer versions do not.
+	if !(metadataRead == int(metadataLen) || metadataRead == int(metadataLen)+4) {
 		return ErrMetadataLength
 	}
 

pkg/chunk/schema.go

@@ -13,6 +13,7 @@ import (
 )
 
 var (
+	chunkTimeRangeKeyV1a = []byte{1}
 	chunkTimeRangeKeyV1  = []byte{'1'}
 	chunkTimeRangeKeyV2  = []byte{'2'}
 	chunkTimeRangeKeyV3  = []byte{'3'}

pkg/chunk/schema_util.go

@@ -187,6 +187,9 @@ func parseChunkTimeRangeValue(rangeValue []byte, value []byte) (
 
 	// v3 schema had four components - label name, label value, chunk ID and version.
 	// "version" is 1 and label value is base64 encoded.
+	// (older code wrote "version" as 1, not '1')
+	case bytes.Equal(components[3], chunkTimeRangeKeyV1a):
+		fallthrough
 	case bytes.Equal(components[3], chunkTimeRangeKeyV1):
 		chunkID = string(components[2])
 		labelValue, err = decodeBase64Value(components[1])

pkg/cortex/modules.go

@@ -12,7 +12,6 @@ import (
 	"github.com/prometheus/common/route"
 	"github.com/prometheus/prometheus/config"
 	v1 "github.com/prometheus/prometheus/web/api/v1"
-	"github.com/thanos-io/thanos/pkg/objstore/s3"
 	httpgrpc_server "github.com/weaveworks/common/httpgrpc/server"
 	"github.com/weaveworks/common/middleware"
 	"github.com/weaveworks/common/server"
@@ -198,14 +197,7 @@ func (t *Cortex) initQuerier(cfg *Config) (err error) {
 	var store querier.ChunkStore
 
 	if cfg.Storage.Engine == storage.StorageEngineTSDB {
-		s3cfg := s3.Config{
-			Bucket:    cfg.TSDB.S3.BucketName,
-			Endpoint:  cfg.TSDB.S3.Endpoint,
-			AccessKey: cfg.TSDB.S3.AccessKeyID,
-			SecretKey: cfg.TSDB.S3.SecretAccessKey,
-			Insecure:  cfg.TSDB.S3.Insecure,
-		}
-		store, err = querier.NewBlockQuerier(s3cfg, cfg.TSDB.SyncDir, prometheus.DefaultRegisterer)
+		store, err = querier.NewBlockQuerier(cfg.TSDB, prometheus.DefaultRegisterer)
 		if err != nil {
 			return err
 		}

pkg/ingester/flush.go

@@ -67,6 +67,10 @@ var (
 		Name: "cortex_ingester_dropped_chunks_total",
 		Help: "Total number of chunks dropped from flushing because they have too few samples.",
 	})
+	oldestUnflushedChunkTimestamp = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "cortex_oldest_unflushed_chunk_timestamp_seconds",
+		Help: "Unix timestamp of the oldest unflushed chunk in the memory",
+	})
 )
 
 // Flush triggers a flush of all the chunks and closes the flush queues.
@@ -110,14 +114,23 @@ func (i *Ingester) sweepUsers(immediate bool) {
 		return
 	}
 
+	oldest := model.Time(0)
+
 	for id, state := range i.userStates.cp() {
 		for pair := range state.fpToSeries.iter() {
 			state.fpLocker.Lock(pair.fp)
 			i.sweepSeries(id, pair.fp, pair.series, immediate)
 			i.removeFlushedChunks(state, pair.fp, pair.series)
+			first := pair.series.firstUnflushedChunkTime()
 			state.fpLocker.Unlock(pair.fp)
+
+			if first > 0 && (oldest == 0 || first < oldest) {
+				oldest = first
+			}
 		}
 	}
+
+	oldestUnflushedChunkTimestamp.Set(float64(oldest.Unix()))
 }
 
 type flushReason int

pkg/ingester/ingester.go

@@ -102,7 +102,7 @@ func newIngesterMetrics(r prometheus.Registerer) *ingesterMetrics {
 type Config struct {
 	LifecyclerConfig ring.LifecyclerConfig `yaml:"lifecycler,omitempty"`
 
-	// Config for transferring chunks.
+	// Config for transferring chunks. Zero or negative = no retries.
 	MaxTransferRetries int `yaml:"max_transfer_retries,omitempty"`
 
 	// Config for chunk flushing.
@@ -134,12 +134,12 @@ type Config struct {
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.LifecyclerConfig.RegisterFlags(f)
 
-	f.IntVar(&cfg.MaxTransferRetries, "ingester.max-transfer-retries", 10, "Number of times to try and transfer chunks before falling back to flushing.")
+	f.IntVar(&cfg.MaxTransferRetries, "ingester.max-transfer-retries", 10, "Number of times to try and transfer chunks before falling back to flushing. Negative value or zero disables hand-over.")
 	f.DurationVar(&cfg.FlushCheckPeriod, "ingester.flush-period", 1*time.Minute, "Period with which to attempt to flush chunks.")
 	f.DurationVar(&cfg.RetainPeriod, "ingester.retain-period", 5*time.Minute, "Period chunks will remain in memory after flushing.")
 	f.DurationVar(&cfg.FlushOpTimeout, "ingester.flush-op-timeout", 1*time.Minute, "Timeout for individual flush operations.")
 	f.DurationVar(&cfg.MaxChunkIdle, "ingester.max-chunk-idle", 5*time.Minute, "Maximum chunk idle time before flushing.")
-	f.DurationVar(&cfg.MaxChunkIdle, "ingester.max-stale-chunk-idle", 0, "Maximum chunk idle time for chunks terminating in stale markers before flushing. 0 disables it and a stale series is not flushed until the max-chunk-idle timeout is reached.")
+	f.DurationVar(&cfg.MaxStaleChunkIdle, "ingester.max-stale-chunk-idle", 0, "Maximum chunk idle time for chunks terminating in stale markers before flushing. 0 disables it and a stale series is not flushed until the max-chunk-idle timeout is reached.")
 	f.DurationVar(&cfg.MaxChunkAge, "ingester.max-chunk-age", 12*time.Hour, "Maximum chunk age before flushing.")
 	f.DurationVar(&cfg.ChunkAgeJitter, "ingester.chunk-age-jitter", 20*time.Minute, "Range of time to subtract from MaxChunkAge to spread out flushes")
 	f.BoolVar(&cfg.SpreadFlushes, "ingester.spread-flushes", false, "If true, spread series flushes across the whole period of MaxChunkAge")

pkg/ingester/ingester_v2.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"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/validation"
 	"github.com/go-kit/kit/log/level"
@@ -16,7 +17,6 @@ import (
 	lbls "github.com/prometheus/prometheus/tsdb/labels"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/objstore"
-	"github.com/thanos-io/thanos/pkg/objstore/s3"
 	"github.com/thanos-io/thanos/pkg/runutil"
 	"github.com/thanos-io/thanos/pkg/shipper"
 	"github.com/weaveworks/common/user"
@@ -32,16 +32,7 @@ type TSDBState struct {
 
 // NewV2 returns a new Ingester that uses prometheus block storage instead of chunk storage
 func NewV2(cfg Config, clientConfig client.Config, limits *validation.Overrides, chunkStore ChunkStore, registerer prometheus.Registerer) (*Ingester, error) {
-	var bkt *s3.Bucket
-	s3Cfg := s3.Config{
-		Bucket:    cfg.TSDBConfig.S3.BucketName,
-		Endpoint:  cfg.TSDBConfig.S3.Endpoint,
-		AccessKey: cfg.TSDBConfig.S3.AccessKeyID,
-		SecretKey: cfg.TSDBConfig.S3.SecretAccessKey,
-		Insecure:  cfg.TSDBConfig.S3.Insecure,
-	}
-	var err error
-	bkt, err = s3.NewBucketWithConfig(util.Logger, s3Cfg, "cortex")
+	bucketClient, err := cortex_tsdb.NewBucketClient(context.Background(), cfg.TSDBConfig, "cortex", util.Logger)
 	if err != nil {
 		return nil, err
 	}
@@ -56,7 +47,7 @@ func NewV2(cfg Config, clientConfig client.Config, limits *validation.Overrides,
 
 		TSDBState: TSDBState{
 			dbs:    make(map[string]*tsdb.DB),
-			bucket: bkt,
+			bucket: bucketClient,
 		},
 	}
 

pkg/ingester/lifecycle_test.go

@@ -45,7 +45,7 @@ func defaultIngesterTestConfig() Config {
 	cfg.LifecyclerConfig.Addr = "localhost"
 	cfg.LifecyclerConfig.ID = "localhost"
 	cfg.LifecyclerConfig.FinalSleep = 0
-	cfg.MaxTransferRetries = -1
+	cfg.MaxTransferRetries = 0
 	return cfg
 }