@@ -101,7 +101,6 @@ const (
101101type userTSDB struct {
102102 db * tsdb.DB
103103 userID string
104- refCache * cortex_tsdb.RefCache
105104 activeSeries * ActiveSeries
106105 seriesInMetric * metricCounter
107106 limiter * Limiter
@@ -185,7 +184,8 @@ func (u *userTSDB) compactHead(blockDuration int64) error {
185184
186185 defer u .casState (forceCompacting , active )
187186
188- // Ingestion of samples in parallel with forced compaction can lead to overlapping blocks.
187+ // Ingestion of samples in parallel with forced compaction can lead to overlapping blocks,
188+ // and possible invalidation of the references returned from Appender.GetRef().
189189 // So we wait for existing in-flight requests to finish. Future push requests would fail until compaction is over.
190190 u .pushesInFlight .Wait ()
191191
@@ -383,7 +383,6 @@ type TSDBState struct {
383383 walReplayTime prometheus.Histogram
384384 appenderAddDuration prometheus.Histogram
385385 appenderCommitDuration prometheus.Histogram
386- refCachePurgeDuration prometheus.Histogram
387386 idleTsdbChecks * prometheus.CounterVec
388387}
389388
@@ -435,11 +434,6 @@ func newTSDBState(bucketClient objstore.Bucket, registerer prometheus.Registerer
435434 Help : "The total time it takes for a push request to commit samples appended to TSDB." ,
436435 Buckets : []float64 {.001 , .005 , .01 , .025 , .05 , .1 , .25 , .5 , 1 , 2.5 , 5 , 10 },
437436 }),
438- refCachePurgeDuration : promauto .With (registerer ).NewHistogram (prometheus.HistogramOpts {
439- Name : "cortex_ingester_tsdb_refcache_purge_duration_seconds" ,
440- Help : "The total time it takes to purge the TSDB series reference cache for a single tenant." ,
441- Buckets : prometheus .DefBuckets ,
442- }),
443437
444438 idleTsdbChecks : idleTsdbChecks ,
445439 }
@@ -619,11 +613,6 @@ func (i *Ingester) updateLoop(ctx context.Context) error {
619613 rateUpdateTicker := time .NewTicker (i .cfg .RateUpdatePeriod )
620614 defer rateUpdateTicker .Stop ()
621615
622- // We use an hardcoded value for this ticker because there should be no
623- // real value in customizing it.
624- refCachePurgeTicker := time .NewTicker (5 * time .Minute )
625- defer refCachePurgeTicker .Stop ()
626-
627616 var activeSeriesTickerChan <- chan time.Time
628617 if i .cfg .ActiveSeriesMetricsEnabled {
629618 t := time .NewTicker (i .cfg .ActiveSeriesMetricsUpdatePeriod )
@@ -646,17 +635,6 @@ func (i *Ingester) updateLoop(ctx context.Context) error {
646635 db .ingestedRuleSamples .tick ()
647636 }
648637 i .userStatesMtx .RUnlock ()
649- case <- refCachePurgeTicker .C :
650- for _ , userID := range i .getTSDBUsers () {
651- userDB := i .getTSDB (userID )
652- if userDB == nil {
653- continue
654- }
655-
656- startTime := time .Now ()
657- userDB .refCache .Purge (startTime .Add (- cortex_tsdb .DefaultRefCacheTTL ))
658- i .TSDBState .refCachePurgeDuration .Observe (time .Since (startTime ).Seconds ())
659- }
660638
661639 case <- activeSeriesTickerChan :
662640 i .v2UpdateActiveSeries ()
@@ -683,6 +661,12 @@ func (i *Ingester) v2UpdateActiveSeries() {
683661 }
684662}
685663
664+ // GetRef() is an extra method added to TSDB to let Cortex check before calling Add()
665+ type extendedAppender interface {
666+ storage.Appender
667+ storage.GetRef
668+ }
669+
686670// v2Push adds metrics to a block
687671func (i * Ingester ) v2Push (ctx context.Context , req * cortexpb.WriteRequest ) (* cortexpb.WriteResponse , error ) {
688672 var firstPartialErr error
@@ -738,13 +722,17 @@ func (i *Ingester) v2Push(ctx context.Context, req *cortexpb.WriteRequest) (*cor
738722 )
739723
740724 // Walk the samples, appending them to the users database
741- app := db .Appender (ctx )
725+ app := db .Appender (ctx ).( extendedAppender )
742726 for _ , ts := range req .Timeseries {
743- // Check if we already have a cached reference for this series. Be aware
744- // that even if we have a reference it's not guaranteed to be still valid.
727+ // Keeps a reference to labels copy, if it was needed. This is to avoid making a copy twice,
728+ // once for TSDB, and second time for activeSeries map.
729+ var copiedLabels []labels.Label
730+
745731 // The labels must be sorted (in our case, it's guaranteed a write request
746732 // has sorted labels once hit the ingester).
747- cachedRef , copiedLabels , cachedRefExists := db .refCache .Ref (startAppend , cortexpb .FromLabelAdaptersToLabels (ts .Labels ))
733+
734+ // Look up a reference for this series. Holding the appendLock ensures that no compaction will happen while we use it.
735+ ref := app .GetRef (cortexpb .FromLabelAdaptersToLabels (ts .Labels ))
748736
749737 // To find out if any sample was added to this series, we keep old value.
750738 oldSucceededSamplesCount := succeededSamplesCount
@@ -753,30 +741,19 @@ func (i *Ingester) v2Push(ctx context.Context, req *cortexpb.WriteRequest) (*cor
753741 var err error
754742
755743 // If the cached reference exists, we try to use it.
756- if cachedRefExists {
757- var ref uint64
758- if ref , err = app .Append (cachedRef , copiedLabels , s .TimestampMs , s .Value ); err == nil {
744+ if ref != 0 {
745+ labels := cortexpb . FromLabelAdaptersToLabels ( ts . Labels )
746+ if _ , err = app .Append (ref , labels , s .TimestampMs , s .Value ); err == nil {
759747 succeededSamplesCount ++
760- // This means the reference changes which means we need to update our cache.
761- if ref != cachedRef {
762- db .refCache .SetRef (startAppend , copiedLabels , ref )
763- }
764748 continue
765749 }
766750
767751 } else {
768- var ref uint64
769-
770752 // Copy the label set because both TSDB and the cache may retain it.
771753 copiedLabels = cortexpb .FromLabelAdaptersToLabelsWithCopy (ts .Labels )
772754
755+ // Retain the reference in case there are multiple samples for the series.
773756 if ref , err = app .Append (0 , copiedLabels , s .TimestampMs , s .Value ); err == nil {
774- db .refCache .SetRef (startAppend , copiedLabels , ref )
775-
776- // Set these in case there are multiple samples for the series.
777- cachedRef = ref
778- cachedRefExists = true
779-
780757 succeededSamplesCount ++
781758 continue
782759 }
@@ -812,6 +789,9 @@ func (i *Ingester) v2Push(ctx context.Context, req *cortexpb.WriteRequest) (*cor
812789 case errMaxSeriesPerMetricLimitExceeded :
813790 perMetricSeriesLimitCount ++
814791 updateFirstPartial (func () error {
792+ if copiedLabels == nil {
793+ copiedLabels = cortexpb .FromLabelAdaptersToLabelsWithCopy (ts .Labels )
794+ }
815795 return makeMetricLimitError (perMetricSeriesLimit , copiedLabels , i .limiter .FormatError (userID , cause ))
816796 })
817797 continue
@@ -1435,7 +1415,6 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
14351415
14361416 userDB := & userTSDB {
14371417 userID : userID ,
1438- refCache : cortex_tsdb .NewRefCache (),
14391418 activeSeries : NewActiveSeries (),
14401419 seriesInMetric : newMetricCounter (i .limiter ),
14411420 ingestedAPISamples : newEWMARate (0.2 , i .cfg .RateUpdatePeriod ),
0 commit comments