Add ephemeral storage to ingester

cmd/mimir/config-descriptor.json

@@ -2794,6 +2794,17 @@
           "fieldFlag": "ingester.ignore-series-limit-for-metric-names",
           "fieldType": "string",
           "fieldCategory": "advanced"
+        },
+        {
+          "kind": "field",
+          "name": "ephemeral_series_retention_period",
+          "required": false,
+          "desc": "Retention of ephemeral series.",
+          "fieldValue": null,
+          "fieldDefaultValue": 600000000000,
+          "fieldFlag": "ingester.ephemeral-series-retention-period",
+          "fieldType": "duration",
+          "fieldCategory": "advanced"
         }
       ],
       "fieldValue": null,

cmd/mimir/help-all.txt.tmpl

@@ -1003,6 +1003,8 @@ Usage of ./cmd/mimir/mimir:
     	Override the default minimum TLS version. Allowed values: VersionTLS10, VersionTLS11, VersionTLS12, VersionTLS13
   -ingester.client.tls-server-name string
     	Override the expected name on the server certificate.
+  -ingester.ephemeral-series-retention-period duration
+    	Retention of ephemeral series. (default 10m0s)
   -ingester.ignore-series-limit-for-metric-names string
     	Comma-separated list of metric names, for which the -ingester.max-global-series-per-metric limit will be ignored. Does not affect the -ingester.max-global-series-per-user limit.
   -ingester.instance-limits.max-inflight-push-requests int

docs/sources/mimir/reference-configuration-parameters/index.md

@@ -856,6 +856,10 @@ instance_limits:
 # the -ingester.max-global-series-per-user limit.
 # CLI flag: -ingester.ignore-series-limit-for-metric-names
 [ignore_series_limit_for_metric_names: <string> | default = ""]
+
+# (advanced) Retention of ephemeral series.
+# CLI flag: -ingester.ephemeral-series-retention-period
+[ephemeral_series_retention_period: <duration> | default = 10m]
 ```
 
 ### querier

pkg/ingester/ingester.go

@@ -103,6 +103,9 @@ const (
 	tenantsWithOutOfOrderEnabledStatName   = "ingester_ooo_enabled_tenants"
 	minOutOfOrderTimeWindowSecondsStatName = "ingester_ooo_min_window"
 	maxOutOfOrderTimeWindowSecondsStatName = "ingester_ooo_max_window"
+
+	// Prefix used in Prometheus registry for ephemeral storage.
+	ephemeralPrometheusMetricsPrefix = "ephemeral_"
 )
 
 // BlocksUploader interface is used to have an easy way to mock it in tests.
@@ -148,6 +151,8 @@ type Config struct {
 	InstanceLimitsFn func() *InstanceLimits `yaml:"-"`
 
 	IgnoreSeriesLimitForMetricNames string `yaml:"ignore_series_limit_for_metric_names" category:"advanced"`
+
+	EphemeralSeriesRetentionPeriod time.Duration `yaml:"ephemeral_series_retention_period" category:"advanced"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
@@ -167,6 +172,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet, logger log.Logger) {
 	cfg.DefaultLimits.RegisterFlags(f)
 
 	f.StringVar(&cfg.IgnoreSeriesLimitForMetricNames, "ingester.ignore-series-limit-for-metric-names", "", "Comma-separated list of metric names, for which the -ingester.max-global-series-per-metric limit will be ignored. Does not affect the -ingester.max-global-series-per-user limit.")
+	f.DurationVar(&cfg.EphemeralSeriesRetentionPeriod, "ingester.ephemeral-series-retention-period", 10*time.Minute, "Retention of ephemeral series.")
 }
 
 func (cfg *Config) getIgnoreSeriesLimitForMetricNamesMap() map[string]struct{} {
@@ -306,6 +312,11 @@ func New(cfg Config, limits *validation.Overrides, activeGroupsCleanupService *u
 			Help: "The current number of series in memory.",
 		}, i.getMemorySeriesMetric)
 
+		promauto.With(registerer).NewGaugeFunc(prometheus.GaugeOpts{
+			Name: "cortex_ingester_ephemeral_series",
+			Help: "The current number of ephemeral series in memory.",
+		}, i.getEphemeralSeriesMetric)
+
 		promauto.With(registerer).NewGaugeFunc(prometheus.GaugeOpts{
 			Name: "cortex_ingester_oldest_unshipped_block_timestamp_seconds",
 			Help: "Unix timestamp of the oldest TSDB block not shipped to the storage yet. 0 if ingester has no blocks or all blocks have been shipped.",
@@ -678,7 +689,7 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, pushReq *push.Request) (
 	}
 
 	// Early exit if no timeseries in request - don't create a TSDB or an appender.
-	if len(req.Timeseries) == 0 {
+	if len(req.Timeseries) == 0 && len(req.EphemeralTimeseries) == 0 {
 		return &mimirpb.WriteResponse{}, nil
 	}
 
@@ -701,7 +712,7 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, pushReq *push.Request) (
 
 		// Keep track of some stats which are tracked only if the samples will be
 		// successfully committed
-		stats pushStats
+		stats, ephemeralStats pushStats
 
 		minAppendTime, minAppendTimeAvailable = db.Head().AppendableMinValidTime()
 
@@ -714,16 +725,54 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, pushReq *push.Request) (
 	)
 
 	// Walk the samples, appending them to the users database
-	app := db.Appender(ctx).(extendedAppender)
-	level.Debug(spanlog).Log("event", "got appender", "numSeries", len(req.Timeseries))
+	var persistentApp, ephemeralApp extendedAppender
 
-	var activeSeries *activeseries.ActiveSeries
-	if i.cfg.ActiveSeriesMetricsEnabled {
-		activeSeries = db.activeSeries
+	rollback := func() {
+		if persistentApp != nil {
+			if err := persistentApp.Rollback(); err != nil {
+				level.Warn(i.logger).Log("msg", "failed to rollback on error", "user", userID, "err", err)
+			}
+		}
+		if ephemeralApp != nil {
+			if err := ephemeralApp.Rollback(); err != nil {
+				level.Warn(i.logger).Log("msg", "failed to rollback ephemeral appender on error", "user", userID, "err", err)
+			}
+		}
 	}
-	err = i.pushSamplesToAppender(userID, req.Timeseries, app, startAppend, &stats, updateFirstPartial, activeSeries, i.limits.OutOfOrderTimeWindow(userID), minAppendTimeAvailable, minAppendTime)
-	if err != nil {
-		return nil, err
+
+	if len(req.Timeseries) > 0 {
+		persistentApp = db.Appender(ctx).(extendedAppender)
+
+		level.Debug(spanlog).Log("event", "got appender", "series", len(req.Timeseries))
+
+		var activeSeries *activeseries.ActiveSeries
+		if i.cfg.ActiveSeriesMetricsEnabled {
+			activeSeries = db.activeSeries
+		}
+		err = i.pushSamplesToAppender(userID, req.Timeseries, persistentApp, startAppend, &stats, updateFirstPartial, activeSeries, i.limits.OutOfOrderTimeWindow(userID), minAppendTimeAvailable, minAppendTime, true)
+		if err != nil {
+			rollback()
+			return nil, err
+		}
+	}
+
+	if len(req.EphemeralTimeseries) > 0 {
+		a, err := db.EphemeralAppender(ctx)
+		if err != nil {
+			// TODO: handle error caused by limit (ephemeral storage disabled), and report it via firstPartialErr instead.
+			rollback()
+			return nil, err
+		}
+
+		ephemeralApp = a.(extendedAppender)
+
+		level.Debug(spanlog).Log("event", "got appender for ephemeral series", "ephemeralSeries", len(req.EphemeralTimeseries))
+
+		err = i.pushSamplesToAppender(userID, req.EphemeralTimeseries, ephemeralApp, startAppend, &ephemeralStats, updateFirstPartial, nil, 0, minAppendTimeAvailable, minAppendTime, false)
+		if err != nil {
+			rollback()
+			return nil, err
+		}
 	}
 
 	// At this point all samples have been added to the appender, so we can track the time it took.
@@ -735,19 +784,37 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, pushReq *push.Request) (
 		"failedSamplesCount", stats.failedSamplesCount,
 		"succeededExemplarsCount", stats.succeededExemplarsCount,
 		"failedExemplarsCount", stats.failedExemplarsCount,
+
+		"ephemeralSucceededSamplesCount", ephemeralStats.succeededSamplesCount,
+		"ephemeralFailedSamplesCount", ephemeralStats.failedSamplesCount,
+		"ephemeralSucceededExemplarsCount", ephemeralStats.succeededExemplarsCount,
+		"ephemeralFailedExemplarsCount", ephemeralStats.failedExemplarsCount,
 	)
 
 	startCommit := time.Now()
-	if err := app.Commit(); err != nil {
-		return nil, wrapWithUser(err, userID)
+	if persistentApp != nil {
+		if err := persistentApp.Commit(); err != nil {
+			rollback()
+			return nil, wrapWithUser(err, userID)
+		}
+
+		persistentApp = nil // Disable rollback for this appender.
+	}
+	if ephemeralApp != nil {
+		if err := ephemeralApp.Commit(); err != nil {
+			rollback()
+			return nil, wrapWithUser(err, userID)
+		}
+
+		ephemeralApp = nil // Disable rollback for this appender.
 	}
 
 	commitDuration := time.Since(startCommit)
 	i.metrics.appenderCommitDuration.Observe(commitDuration.Seconds())
 	level.Debug(spanlog).Log("event", "complete commit", "commitDuration", commitDuration.String())
 
 	// If only invalid samples are pushed, don't change "last update", as TSDB was not modified.
-	if stats.succeededSamplesCount > 0 {
+	if stats.succeededSamplesCount > 0 || ephemeralStats.succeededSamplesCount > 0 {
 		db.setLastUpdate(time.Now())
 	}
 
@@ -761,8 +828,31 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, pushReq *push.Request) (
 	i.appendedSamplesStats.Inc(int64(stats.succeededSamplesCount))
 	i.appendedExemplarsStats.Inc(int64(stats.succeededExemplarsCount))
 
+	i.metrics.ephemeralIngestedSamples.WithLabelValues(userID).Add(float64(ephemeralStats.succeededSamplesCount))
+	i.metrics.ephemeralIngestedSamplesFail.WithLabelValues(userID).Add(float64(ephemeralStats.failedSamplesCount))
+	i.appendedSamplesStats.Inc(int64(ephemeralStats.succeededSamplesCount))
+
 	group := i.activeGroups.UpdateActiveGroupTimestamp(userID, validation.GroupLabel(i.limits, userID, req.Timeseries), startAppend)
 
+	i.updateMetricsFromPushStats(userID, group, &stats)
+	i.updateMetricsFromPushStats(userID, group, &ephemeralStats)
+
+	db.updatedRatesFromStats(stats.succeededSamplesCount, req.Source)
+	db.updatedRatesFromStats(ephemeralStats.succeededSamplesCount, req.Source)
+
+	if firstPartialErr != nil {
+		code := http.StatusBadRequest
+		var ve *validationError
+		if errors.As(firstPartialErr, &ve) {
+			code = ve.code
+		}
+		return &mimirpb.WriteResponse{}, httpgrpc.Errorf(code, wrapWithUser(firstPartialErr, userID).Error())
+	}
+
+	return &mimirpb.WriteResponse{}, nil
+}
+
+func (i *Ingester) updateMetricsFromPushStats(userID string, group string, stats *pushStats) {
 	if stats.sampleOutOfBoundsCount > 0 {
 		i.metrics.discardedSamplesSampleOutOfBounds.WithLabelValues(userID, group).Add(float64(stats.sampleOutOfBoundsCount))
 	}
@@ -783,34 +873,14 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, pushReq *push.Request) (
 	}
 	if stats.succeededSamplesCount > 0 {
 		i.ingestionRate.Add(int64(stats.succeededSamplesCount))
-
-		switch req.Source {
-		case mimirpb.RULE:
-			db.ingestedRuleSamples.Add(int64(stats.succeededSamplesCount))
-		case mimirpb.API:
-			fallthrough
-		default:
-			db.ingestedAPISamples.Add(int64(stats.succeededSamplesCount))
-		}
-	}
-
-	if firstPartialErr != nil {
-		code := http.StatusBadRequest
-		var ve *validationError
-		if errors.As(firstPartialErr, &ve) {
-			code = ve.code
-		}
-		return &mimirpb.WriteResponse{}, httpgrpc.Errorf(code, wrapWithUser(firstPartialErr, userID).Error())
 	}
-
-	return &mimirpb.WriteResponse{}, nil
 }
 
 // pushSamplesToAppender appends samples and exemplars to the appender. Most errors are handled via updateFirstPartial function,
 // but in case of unhandled errors, appender is rolled back and such error is returned.
 func (i *Ingester) pushSamplesToAppender(userID string, timeseries []mimirpb.PreallocTimeseries, app extendedAppender, startAppend time.Time,
 	stats *pushStats, updateFirstPartial func(errFn func() error), activeSeries *activeseries.ActiveSeries,
-	outOfOrderWindow model.Duration, minAppendTimeAvailable bool, minAppendTime int64) error {
+	outOfOrderWindow model.Duration, minAppendTimeAvailable bool, minAppendTime int64, appendExemplars bool) error {
 	for _, ts := range timeseries {
 		// The labels must be sorted (in our case, it's guaranteed a write request
 		// has sorted labels once hit the ingester).
@@ -901,11 +971,6 @@ func (i *Ingester) pushSamplesToAppender(userID string, timeseries []mimirpb.Pre
 				continue
 			}
 
-			// The error looks an issue on our side, so we should rollback
-			if rollbackErr := app.Rollback(); rollbackErr != nil {
-				level.Warn(i.logger).Log("msg", "failed to rollback on error", "user", userID, "err", rollbackErr)
-			}
-
 			return wrapWithUser(err, userID)
 		}
 
@@ -916,7 +981,7 @@ func (i *Ingester) pushSamplesToAppender(userID string, timeseries []mimirpb.Pre
 			})
 		}
 
-		if len(ts.Exemplars) > 0 && i.limits.MaxGlobalExemplarsPerUser(userID) > 0 {
+		if appendExemplars && len(ts.Exemplars) > 0 && i.limits.MaxGlobalExemplarsPerUser(userID) > 0 {
 			// app.AppendExemplar currently doesn't create the series, it must
 			// already exist.  If it does not then drop.
 			if ref == 0 {
@@ -1673,6 +1738,36 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	}
 
 	i.tsdbMetrics.setRegistryForUser(userID, tsdbPromReg)
+
+	userDB.ephemeralFactory = func() (*tsdb.Head, error) {
+		// TODO: check user limit for ephemeral series. If it's 0, don't create head and return error.
+
+		headOptions := &tsdb.HeadOptions{
+			ChunkRange:                     i.cfg.EphemeralSeriesRetentionPeriod.Milliseconds(),
+			ChunkDirRoot:                   filepath.Join(udir, "ephemeral_chunks"),
+			ChunkWriteBufferSize:           i.cfg.BlocksStorageConfig.TSDB.HeadChunksWriteBufferSize,
+			ChunkEndTimeVariance:           0,
+			ChunkWriteQueueSize:            i.cfg.BlocksStorageConfig.TSDB.HeadChunksWriteQueueSize,
+			StripeSize:                     i.cfg.BlocksStorageConfig.TSDB.StripeSize,
+			SeriesCallback:                 nil, // TODO: handle limits.
+			EnableExemplarStorage:          false,
+			EnableMemorySnapshotOnShutdown: false,
+			IsolationDisabled:              true,
+		}
+
+		// We need to set this, despite OOO time window being 0.
+		headOptions.OutOfOrderCapMax.Store(int64(i.cfg.BlocksStorageConfig.TSDB.OutOfOrderCapacityMax))
+
+		h, err := tsdb.NewHead(prometheus.WrapRegistererWithPrefix(ephemeralPrometheusMetricsPrefix, tsdbPromReg), log.With(userLogger, "ephemeral", "true"), nil, nil, headOptions, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		// Don't allow ingestion of old samples into ephemeral storage.
+		h.SetMinValidTime(time.Now().Add(-i.cfg.EphemeralSeriesRetentionPeriod).UnixMilli())
+		return h, nil
+	}
+
 	return userDB, nil
 }
 
@@ -1832,6 +1927,26 @@ func (i *Ingester) getMemorySeriesMetric() float64 {
 	return float64(count)
 }
 
+// getEphemeralSeriesMetric returns the total number of in-memory series in ephemeral storage across all tenants.
+func (i *Ingester) getEphemeralSeriesMetric() float64 {
+	if err := i.checkRunning(); err != nil {
+		return 0
+	}
+
+	i.tsdbsMtx.RLock()
+	defer i.tsdbsMtx.RUnlock()
+
+	count := uint64(0)
+	for _, db := range i.tsdbs {
+		eph := db.getEphemeralStorage()
+		if eph != nil {
+			count += eph.NumSeries()
+		}
+	}
+
+	return float64(count)
+}
+
 // getOldestUnshippedBlockMetric returns the unix timestamp of the oldest unshipped block or
 // 0 if all blocks have been shipped.
 func (i *Ingester) getOldestUnshippedBlockMetric() float64 {
@@ -2011,7 +2126,7 @@ func (i *Ingester) compactBlocks(ctx context.Context, force bool, allowed *util.
 
 		default:
 			reason = "regular"
-			err = userDB.Compact()
+			err = userDB.Compact(time.Now().Add(-i.cfg.EphemeralSeriesRetentionPeriod))
 		}
 
 		if err != nil {