@ -18,7 +18,7 @@ import (
"container/list"
"errors"
"fmt"
" math "
" runtime "
"sort"
"sync"
"sync/atomic"
@ -41,8 +41,11 @@ const (
fpMaxSweepTime = 6 * time . Hour
fpMaxWaitDuration = 10 * time . Second
// See waitForNextFP.
maxEvictInterval = time . Minute
// See handleEvictList. This should be clearly shorter than the usual CG
// interval. On the other hand, each evict check calls ReadMemStats,
// which involves stopping the world (at least up to Go1.8). Hence,
// don't just set this to a very short interval.
evictInterval = time . Second
// Constants to control the hysteresis of entering and leaving "rushed
// mode". In rushed mode, the dirty series count is ignored for
@ -76,24 +79,6 @@ const (
fpOtherMatchThreshold = 10000
)
var (
numChunksToPersistDesc = prometheus . NewDesc (
prometheus . BuildFQName ( namespace , subsystem , "chunks_to_persist" ) ,
"The current number of chunks waiting for persistence." ,
nil , nil ,
)
maxChunksToPersistDesc = prometheus . NewDesc (
prometheus . BuildFQName ( namespace , subsystem , "max_chunks_to_persist" ) ,
"The maximum number of chunks that can be waiting for persistence before sample ingestion will stop." ,
nil , nil ,
)
maxMemChunksDesc = prometheus . NewDesc (
prometheus . BuildFQName ( namespace , subsystem , "max_memory_chunks" ) ,
"The configured maximum number of chunks that can be held in memory" ,
nil , nil ,
)
)
type quarantineRequest struct {
fp model . Fingerprint
metric model . Metric
@ -147,12 +132,13 @@ type syncStrategy func() bool
// interfacing with a persistence layer to make time series data persistent
// across restarts and evictable from memory.
type MemorySeriesStorage struct {
// archiveHighWatermark and numChunksToPersist have to be aligned for atomic operations.
// archiveHighWatermark , chunksToPersist, persistUrgency have to be aligned for atomic operations.
archiveHighWatermark model . Time // No archived series has samples after this time.
numChunksToPersist int64 // The number of chunks waiting for persistence.
maxChunksToPersist int // If numChunksToPersist reaches this threshold, ingestion will be throttled .
persistUrgency int32 // Persistence urgency score * 1000, int32 allows atomic operations .
rushed bool // Whether the storage is in rushed mode.
rushedMtx sync . Mutex // Protects entering and exiting rushed mode.
rushedMtx sync . Mutex // Protects rushed.
lastNumGC uint32 // To detect if a GC cycle has run.
throttled chan struct { } // This chan is sent to whenever NeedsThrottling() returns true (for logging).
fpLocker * fingerprintLocker
@ -162,7 +148,7 @@ type MemorySeriesStorage struct {
loopStopping , loopStopped chan struct { }
logThrottlingStopped chan struct { }
maxMemoryChunks int
targetHeapSize uint64
dropAfter time . Duration
headChunkTimeout time . Duration
checkpointInterval time . Duration
@ -178,26 +164,28 @@ type MemorySeriesStorage struct {
quarantineRequests chan quarantineRequest
quarantineStopping , quarantineStopped chan struct { }
persistErrors prometheus . Counter
queuedChunksToPersist prometheus . Counter
numSeries prometheus . Gauge
numHeadChunks prometheus . Gauge
dirtySeries prometheus . Gauge
seriesOps * prometheus . CounterVec
ingestedSamplesCount prometheus . Counter
discardedSamplesCount * prometheus . CounterVec
nonExistentSeriesMatchesCount prometheus . Counter
maintainSeriesDuration * prometheus . SummaryVec
persistenceUrgencyScore prometheus . Gauge
rushedMode prometheus . Gauge
persistErrors prometheus . Counter
queuedChunksToPersist prometheus . Counter
chunksToPersist prometheus . GaugeFunc
memorySeries prometheus . Gauge
headChunks prometheus . Gauge
dirtySeries prometheus . Gauge
seriesOps * prometheus . CounterVec
ingestedSamples prometheus . Counter
discardedSamples * prometheus . CounterVec
nonExistentSeriesMatches prometheus . Counter
memChunks prometheus . GaugeFunc
maintainSeriesDuration * prometheus . SummaryVec
persistenceUrgencyScore prometheus . GaugeFunc
rushedMode prometheus . GaugeFunc
targetHeapSizeBytes prometheus . GaugeFunc
}
// MemorySeriesStorageOptions contains options needed by
// NewMemorySeriesStorage. It is not safe to leave any of those at their zero
// values.
type MemorySeriesStorageOptions struct {
MemoryChunks int // How many chunks to keep in memory.
MaxChunksToPersist int // Max number of chunks waiting to be persisted.
TargetHeapSize uint64 // Desired maximum heap size.
PersistenceStoragePath string // Location of persistence files.
PersistenceRetentionPeriod time . Duration // Chunks at least that old are dropped.
HeadChunkTimeout time . Duration // Head chunks idle for at least that long may be closed.
@ -222,15 +210,13 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) *MemorySeriesStorage
loopStopped : make ( chan struct { } ) ,
logThrottlingStopped : make ( chan struct { } ) ,
throttled : make ( chan struct { } , 1 ) ,
maxMemoryChunks: o . MemoryChunks ,
targetHeapSize: o . TargetHeapSize ,
dropAfter : o . PersistenceRetentionPeriod ,
headChunkTimeout : o . HeadChunkTimeout ,
checkpointInterval : o . CheckpointInterval ,
checkpointDirtySeriesLimit : o . CheckpointDirtySeriesLimit ,
archiveHighWatermark : model . Now ( ) . Add ( - o . HeadChunkTimeout ) ,
maxChunksToPersist : o . MaxChunksToPersist ,
evictList : list . New ( ) ,
evictRequests : make ( chan chunk . EvictRequest , evictRequestsCap ) ,
evictStopping : make ( chan struct { } ) ,
@ -252,13 +238,13 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) *MemorySeriesStorage
Name : "queued_chunks_to_persist_total" ,
Help : "The total number of chunks queued for persistence." ,
} ) ,
num Series: prometheus . NewGauge ( prometheus . GaugeOpts {
memory Series: prometheus . NewGauge ( prometheus . GaugeOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "memory_series" ,
Help : "The current number of series in memory." ,
} ) ,
numH eadChunks: prometheus . NewGauge ( prometheus . GaugeOpts {
h eadChunks: prometheus . NewGauge ( prometheus . GaugeOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "open_head_chunks" ,
@ -279,13 +265,13 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) *MemorySeriesStorage
} ,
[ ] string { opTypeLabel } ,
) ,
ingestedSamples Count : prometheus . NewCounter ( prometheus . CounterOpts {
ingestedSamples : prometheus . NewCounter ( prometheus . CounterOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "ingested_samples_total" ,
Help : "The total number of samples ingested." ,
} ) ,
discardedSamples Count : prometheus . NewCounterVec (
discardedSamples : prometheus . NewCounterVec (
prometheus . CounterOpts {
Namespace : namespace ,
Subsystem : subsystem ,
@ -294,12 +280,21 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) *MemorySeriesStorage
} ,
[ ] string { discardReasonLabel } ,
) ,
nonExistentSeriesMatches Count : prometheus . NewCounter ( prometheus . CounterOpts {
nonExistentSeriesMatches : prometheus . NewCounter ( prometheus . CounterOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "non_existent_series_matches_total" ,
Help : "How often a non-existent series was referred to during label matching or chunk preloading. This is an indication of outdated label indexes." ,
} ) ,
memChunks : prometheus . NewGaugeFunc (
prometheus . GaugeOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "memory_chunks" ,
Help : "The current number of chunks in memory. The number does not include cloned chunks (i.e. chunks without a descriptor)." ,
} ,
func ( ) float64 { return float64 ( atomic . LoadInt64 ( & chunk . NumMemChunks ) ) } ,
) ,
maintainSeriesDuration : prometheus . NewSummaryVec (
prometheus . SummaryOpts {
Namespace : namespace ,
@ -309,24 +304,63 @@ func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) *MemorySeriesStorage
} ,
[ ] string { seriesLocationLabel } ,
) ,
persistenceUrgencyScore : prometheus . NewGauge ( prometheus . GaugeOpts {
}
s . chunksToPersist = prometheus . NewGaugeFunc (
prometheus . GaugeOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "chunks_to_persist" ,
Help : "The current number of chunks waiting for persistence." ,
} ,
func ( ) float64 {
return float64 ( s . getNumChunksToPersist ( ) )
} ,
)
s . rushedMode = prometheus . NewGaugeFunc (
prometheus . GaugeOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "rushed_mode" ,
Help : "1 if the storage is in rushed mode, 0 otherwise." ,
} ,
func ( ) float64 {
s . rushedMtx . Lock ( )
defer s . rushedMtx . Unlock ( )
if s . rushed {
return 1
}
return 0
} ,
)
s . persistenceUrgencyScore = prometheus . NewGaugeFunc (
prometheus . GaugeOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "persistence_urgency_score" ,
Help : "A score of urgency to persist chunks, 0 is least urgent, 1 most." ,
} ) ,
rushedMode : prometheus . NewGauge ( prometheus . GaugeOpts {
} ,
func ( ) float64 {
score , _ := s . getPersistenceUrgencyScore ( )
return score
} ,
)
s . targetHeapSizeBytes = prometheus . NewGaugeFunc (
prometheus . GaugeOpts {
Namespace : namespace ,
Subsystem : subsystem ,
Name : "rushed_mode" ,
Help : "1 if the storage is in rushed mode, 0 otherwise. In rushed mode, the system behaves as if the persistence_urgency_score is 1." ,
} ) ,
}
Name : "target_heap_size_bytes" ,
Help : "The configured target heap size in bytes." ,
} ,
func ( ) float64 {
return float64 ( s . targetHeapSize )
} ,
)
// Initialize metric vectors.
// TODO(beorn7): Rework once we have a utility function for it in client_golang.
s . discardedSamplesCount . WithLabelValues ( outOfOrderTimestamp )
s . discardedSamplesCount . WithLabelValues ( duplicateSample )
s . discardedSamples . WithLabelValues ( outOfOrderTimestamp )
s . discardedSamples . WithLabelValues ( duplicateSample )
s . maintainSeriesDuration . WithLabelValues ( maintainInMemory )
s . maintainSeriesDuration . WithLabelValues ( maintainArchived )
s . seriesOps . WithLabelValues ( create )
@ -353,7 +387,10 @@ func (s *MemorySeriesStorage) Start() (err error) {
case Always :
syncStrategy = func ( ) bool { return true }
case Adaptive :
syncStrategy = func ( ) bool { return s . calculatePersistenceUrgencyScore ( ) < 1 }
syncStrategy = func ( ) bool {
_ , rushed := s . getPersistenceUrgencyScore ( )
return ! rushed
}
default :
panic ( "unknown sync strategy" )
}
@ -385,7 +422,7 @@ func (s *MemorySeriesStorage) Start() (err error) {
for fp := range s . fpToSeries . fpIter ( ) {
if series , ok := s . fpToSeries . get ( fp ) ; ok {
if ! series . headChunkClosed {
s . numH eadChunks. Inc ( )
s . h eadChunks. Inc ( )
}
}
}
@ -394,7 +431,7 @@ func (s *MemorySeriesStorage) Start() (err error) {
return err
}
log . Infof ( "%d series loaded." , s . fpToSeries . length ( ) )
s . num Series. Set ( float64 ( s . fpToSeries . length ( ) ) )
s . memory Series. Set ( float64 ( s . fpToSeries . length ( ) ) )
s . mapper , err = newFPMapper ( s . fpToSeries , p )
if err != nil {
@ -805,7 +842,7 @@ func (s *MemorySeriesStorage) metricForRange(
// we have a chance the archived metric is not in the range.
has , first , last := s . persistence . hasArchivedMetric ( fp )
if ! has {
s . nonExistentSeriesMatches Count . Inc ( )
s . nonExistentSeriesMatches . Inc ( )
return nil , nil , false
}
if first . After ( through ) || last . Before ( from ) {
@ -882,11 +919,11 @@ func (s *MemorySeriesStorage) Append(sample *model.Sample) error {
sample . Value . Equal ( series . lastSampleValue ) {
return nil
}
s . discardedSamples Count . WithLabelValues ( duplicateSample ) . Inc ( )
s . discardedSamples . WithLabelValues ( duplicateSample ) . Inc ( )
return ErrDuplicateSampleForTimestamp // Caused by the caller.
}
if sample . Timestamp < series . lastTime {
s . discardedSamples Count . WithLabelValues ( outOfOrderTimestamp ) . Inc ( )
s . discardedSamples . WithLabelValues ( outOfOrderTimestamp ) . Inc ( )
return ErrOutOfOrderSample // Caused by the caller.
}
completedChunksCount , err := series . add ( model . SamplePair {
@ -897,7 +934,7 @@ func (s *MemorySeriesStorage) Append(sample *model.Sample) error {
s . quarantineSeries ( fp , sample . Metric , err )
return err
}
s . ingestedSamples Count . Inc ( )
s . ingestedSamples . Inc ( )
s . incNumChunksToPersist ( completedChunksCount )
return nil
@ -905,8 +942,7 @@ func (s *MemorySeriesStorage) Append(sample *model.Sample) error {
// NeedsThrottling implements Storage.
func ( s * MemorySeriesStorage ) NeedsThrottling ( ) bool {
if s . getNumChunksToPersist ( ) > s . maxChunksToPersist ||
float64 ( atomic . LoadInt64 ( & chunk . NumMemChunks ) ) > float64 ( s . maxMemoryChunks ) * toleranceFactorMemChunks {
if score , _ := s . getPersistenceUrgencyScore ( ) ; score >= 1 {
select {
case s . throttled <- struct { } { } :
default : // Do nothing, signal already pending.
@ -937,19 +973,19 @@ func (s *MemorySeriesStorage) logThrottling() {
select {
case <- s . throttled :
if ! timer . Reset ( time . Minute ) {
score , _ := s . getPersistenceUrgencyScore ( )
log .
With ( "urgencyScore" , score ) .
With ( "chunksToPersist" , s . getNumChunksToPersist ( ) ) .
With ( "maxChunksToPersist" , s . maxChunksToPersist ) .
With ( "memoryChunks" , atomic . LoadInt64 ( & chunk . NumMemChunks ) ) .
With ( "maxToleratedMemChunks" , int ( float64 ( s . maxMemoryChunks ) * toleranceFactorMemChunks ) ) .
Error ( "Storage needs throttling. Scrapes and rule evaluations will be skipped." )
}
case <- timer . C :
score , _ := s . getPersistenceUrgencyScore ( )
log .
With ( "urgencyScore" , score ) .
With ( "chunksToPersist" , s . getNumChunksToPersist ( ) ) .
With ( "maxChunksToPersist" , s . maxChunksToPersist ) .
With ( "memoryChunks" , atomic . LoadInt64 ( & chunk . NumMemChunks ) ) .
With ( "maxToleratedMemChunks" , int ( float64 ( s . maxMemoryChunks ) * toleranceFactorMemChunks ) ) .
Info ( "Storage does not need throttling anymore." )
case <- s . loopStopping :
return
@ -994,9 +1030,9 @@ func (s *MemorySeriesStorage) getOrCreateSeries(fp model.Fingerprint, m model.Me
return nil , err
}
s . fpToSeries . put ( fp , series )
s . num Series. Inc ( )
s . memory Series. Inc ( )
if ! series . headChunkClosed {
s . numH eadChunks. Inc ( )
s . h eadChunks. Inc ( )
}
}
return series , nil
@ -1061,23 +1097,16 @@ func (s *MemorySeriesStorage) preloadChunksForInstant(
}
func ( s * MemorySeriesStorage ) handleEvictList ( ) {
ticker := time . NewTicker ( maxEvictInterval )
count := 0
// This ticker is supposed to tick at least once per GC cyle. Ideally,
// we would handle the evict list after each finished GC cycle, but I
// don't know of a way to "subscribe" to that kind of event.
ticker := time . NewTicker ( evictInterval )
for {
// To batch up evictions a bit, this tries evictions at least
// once per evict interval, but earlier if the number of evict
// requests with evict==true that have happened since the last
// evict run is more than maxMemoryChunks/1000.
select {
case req := <- s . evictRequests :
if req . Evict {
req . Desc . EvictListElement = s . evictList . PushBack ( req . Desc )
count ++
if count > s . maxMemoryChunks / 1000 {
s . maybeEvict ( )
count = 0
}
} else {
if req . Desc . EvictListElement != nil {
s . evictList . Remove ( req . Desc . EvictListElement )
@ -1085,9 +1114,7 @@ func (s *MemorySeriesStorage) handleEvictList() {
}
}
case <- ticker . C :
if s . evictList . Len ( ) > 0 {
s . maybeEvict ( )
}
s . maybeEvict ( )
case <- s . evictStopping :
// Drain evictRequests forever in a goroutine to not let
// requesters hang.
@ -1106,10 +1133,14 @@ func (s *MemorySeriesStorage) handleEvictList() {
// maybeEvict is a local helper method. Must only be called by handleEvictList.
func ( s * MemorySeriesStorage ) maybeEvict ( ) {
numChunksToEvict := int ( atomic . LoadInt64 ( & chunk . NumMemChunks ) ) - s . maxMemoryChunks
ms := runtime . MemStats { }
runtime . ReadMemStats ( & ms )
numChunksToEvict := s . calculatePersistUrgency ( & ms )
if numChunksToEvict <= 0 {
return
}
chunkDescsToEvict := make ( [ ] * chunk . Desc , numChunksToEvict )
for i := range chunkDescsToEvict {
e := s . evictList . Front ( )
@ -1143,6 +1174,118 @@ func (s *MemorySeriesStorage) maybeEvict() {
} ( )
}
// calculatePersistUrgency calculates and sets s.persistUrgency. Based on the
// calculation, it returns the number of chunks to evict. The runtime.MemStats
// are passed in here for testability.
//
// The persist urgency is calculated by the following formula:
//
// n(toPersist) MAX( h(nextGC), h(current) )
// p = MIN( 1, --------------------------- * ---------------------------- )
// n(toPersist) + n(evictable) h(target)
//
// where:
//
// n(toPersist): Number of chunks waiting for persistence.
// n(evictable): Number of evictable chunks.
// h(nextGC): Heap size at which the next GC will kick in (ms.NextGC).
// h(current): Current heap size (ms.HeapAlloc).
// h(target): Configured target heap size.
//
// Note that the actual value stored in s.persistUrgency is 1000 times the value
// calculated as above to allow using an int32, which supports atomic
// operations.
//
// If no GC has run after the last call of this method, it will always return 0
// (no reason to try to evict any more chunks before we have seen the effect of
// the previous eviction). It will also not decrease the persist urgency in this
// case (but it will increase the persist urgency if a higher value was calculated).
//
// If a GC has run after the last call of this method, the following cases apply:
//
// - If MAX( h(nextGC), h(current) ) < h(target), simply return 0. Nothing to
// evict if the heap is still small enough.
//
// - Otherwise, if n(evictable) is 0, also return 0, but set the urgency score
// to 1 to signal that we want to evict chunk but have no evictable chunks
// available.
//
// - Otherwise, calulate the number of chunks to evict and return it:
//
// MAX( h(nextGC), h(current) ) - h(target)
// n(toEvict) = MIN( n(evictable), ---------------------------------------- )
// c
//
// where c is the size of a chunk.
//
// - In the latter case, the persist urgency might be increased. The final value
// is the following:
//
// n(toEvict)
// MAX( p, ------------ )
// n(evictable)
//
// Broadly speaking, the persist urgency is based on the ratio of the number of
// chunks we want to evict and the number of chunks that are actually
// evictable. However, in particular for the case where we don't need to evict
// chunks yet, it also takes into account how close the heap has already grown
// to the configured target size, and how big the pool of chunks to persist is
// compared to the number of chunks already evictable.
//
// This is a helper method only to be called by MemorySeriesStorage.maybeEvict.
func ( s * MemorySeriesStorage ) calculatePersistUrgency ( ms * runtime . MemStats ) int {
var (
oldUrgency = atomic . LoadInt32 ( & s . persistUrgency )
newUrgency int32
numChunksToPersist = s . getNumChunksToPersist ( )
)
defer func ( ) {
if newUrgency > 1000 {
newUrgency = 1000
}
atomic . StoreInt32 ( & s . persistUrgency , newUrgency )
} ( )
// Take the NextGC as the relevant heap size because the heap will grow
// to that size before GC kicks in. However, at times the current heap
// is already larger than NextGC, in which case we take that worse case.
heapSize := ms . NextGC
if ms . HeapAlloc > ms . NextGC {
heapSize = ms . HeapAlloc
}
if numChunksToPersist > 0 {
newUrgency = int32 ( 1000 * uint64 ( numChunksToPersist ) / uint64 ( numChunksToPersist + s . evictList . Len ( ) ) * heapSize / s . targetHeapSize )
}
// Only continue if a GC has happened since we were here last time.
if ms . NumGC == s . lastNumGC {
if oldUrgency > newUrgency {
// Never reduce urgency without a GC run.
newUrgency = oldUrgency
}
return 0
}
s . lastNumGC = ms . NumGC
if heapSize <= s . targetHeapSize {
return 0 // Heap still small enough, don't evict.
}
if s . evictList . Len ( ) == 0 {
// We want to reduce heap size but there is nothing to evict.
newUrgency = 1000
return 0
}
numChunksToEvict := int ( ( heapSize - s . targetHeapSize ) / chunk . ChunkLen )
if numChunksToEvict > s . evictList . Len ( ) {
numChunksToEvict = s . evictList . Len ( )
}
if u := int32 ( numChunksToEvict * 1000 / s . evictList . Len ( ) ) ; u > newUrgency {
newUrgency = u
}
return numChunksToEvict
}
// waitForNextFP waits an estimated duration, after which we want to process
// another fingerprint so that we will process all fingerprints in a tenth of
// s.dropAfter assuming that the system is doing nothing else, e.g. if we want
@ -1213,7 +1356,11 @@ func (s *MemorySeriesStorage) cycleThroughMemoryFingerprints() chan model.Finger
return
}
// Reduce the wait time according to the urgency score.
s . waitForNextFP ( s . fpToSeries . length ( ) , 1 - s . calculatePersistenceUrgencyScore ( ) )
score , rushed := s . getPersistenceUrgencyScore ( )
if rushed {
score = 1
}
s . waitForNextFP ( s . fpToSeries . length ( ) , 1 - score )
count ++
}
if count > 0 {
@ -1329,8 +1476,8 @@ loop:
// while in a situation like that, where we are clearly lacking speed of disk
// maintenance, the best we can do for crash recovery is to persist chunks as
// quickly as possible. So only checkpoint if the urgency score is < 1.
if dirty >= int64 ( s . checkpointDirtySeriesLimit ) &&
s. calculatePersistenceUrgencyScore ( ) < 1 {
if _, rushed := s . getPersistenceUrgencyScore ( ) ; ! rushed &&
dirty >= int64 ( s . checkpointDirtySeriesLimit ) {
checkpointTimer . Reset ( 0 )
}
}
@ -1404,7 +1551,7 @@ func (s *MemorySeriesStorage) maintainMemorySeries(
}
if closed {
s . incNumChunksToPersist ( 1 )
s . numH eadChunks. Dec ( )
s . h eadChunks. Dec ( )
}
seriesWasDirty := series . dirty
@ -1426,7 +1573,7 @@ func (s *MemorySeriesStorage) maintainMemorySeries(
// an age of at least headChunkTimeout (which is very likely anyway).
if iOldestNotEvicted == - 1 && model . Now ( ) . Sub ( series . lastTime ) > s . headChunkTimeout {
s . fpToSeries . del ( fp )
s . num Series. Dec ( )
s . memory Series. Dec ( )
s . persistence . archiveMetric ( fp , series . metric , series . firstTime ( ) , series . lastTime )
s . seriesOps . WithLabelValues ( archive ) . Inc ( )
oldWatermark := atomic . LoadInt64 ( ( * int64 ) ( & s . archiveHighWatermark ) )
@ -1523,7 +1670,7 @@ func (s *MemorySeriesStorage) writeMemorySeries(
if len ( series . chunkDescs ) == 0 && allDroppedFromPersistence {
// All chunks dropped from both memory and persistence. Delete the series for good.
s . fpToSeries . del ( fp )
s . num Series. Dec ( )
s . memory Series. Dec ( )
s . seriesOps . WithLabelValues ( memoryPurge ) . Inc ( )
s . persistence . unindexMetric ( fp , series . metric )
return true
@ -1586,12 +1733,12 @@ func (s *MemorySeriesStorage) loadChunkDescs(fp model.Fingerprint, offsetFromEnd
return s . persistence . loadChunkDescs ( fp , offsetFromEnd )
}
// getNumChunksToPersist returns numC hunksToPersist in a goroutine-safe way.
// getNumChunksToPersist returns c hunksToPersist in a goroutine-safe way.
func ( s * MemorySeriesStorage ) getNumChunksToPersist ( ) int {
return int ( atomic . LoadInt64 ( & s . numChunksToPersist ) )
}
// incNumChunksToPersist increments numC hunksToPersist in a goroutine-safe way. Use a
// incNumChunksToPersist increments c hunksToPersist in a goroutine-safe way. Use a
// negative 'by' to decrement.
func ( s * MemorySeriesStorage ) incNumChunksToPersist ( by int ) {
atomic . AddInt64 ( & s . numChunksToPersist , int64 ( by ) )
@ -1600,89 +1747,57 @@ func (s *MemorySeriesStorage) incNumChunksToPersist(by int) {
}
}
// calculatePersistenceUrgencyScore calculates and returns an urgency score for
// the speed of persisting chunks. The score is between 0 and 1, where 0 means
// no urgency at all and 1 means highest urgency.
// getPersistenceUrgencyScore returns an urgency score for the speed of
// persisting chunks. The score is between 0 and 1, where 0 means no urgency at
// all and 1 means highest urgency. It also returns if the storage is in
// "rushed mode".
//
// The score is the maximum of the two following sub-scores:
// The storage enters "rushed mode" if the score exceeds
// persintenceUrgencyScoreForEnteringRushedMode at the time this method is
// called. It will leave "rushed mode" if, at a later time this method is
// called, the score is below persintenceUrgencyScoreForLeavingRushedMode.
// "Rushed mode" plays a role for the adaptive series-sync-strategy. It also
// switches off early checkpointing (due to dirty series), and it makes series
// maintenance happen as quickly as possible.
//
// (1) The first sub-score is the number of chunks waiting for persistence
// divided by the maximum number of chunks allowed to be waiting for
// persistence.
// A score of 1 will trigger throttling of sample ingestion.
//
// (2) If there are more chunks in memory than allowed AND there are more chunks
// waiting for persistence than factorMinChunksToPersist times
// -storage.local.max-chunks-to-persist, then the second sub-score is the
// fraction the number of memory chunks has reached between
// -storage.local.memory-chunks and toleranceFactorForMemChunks times
// -storage.local.memory-chunks.
//
// Should the score ever hit persintenceUrgencyScoreForEnteringRushedMode, the
// storage locks into "rushed mode", in which the returned score is always
// bumped up to 1 until the non-bumped score is below
// persintenceUrgencyScoreForLeavingRushedMode.
//
// This method is not goroutine-safe, but it is only ever called by the single
// goroutine that is in charge of series maintenance. According to the returned
// score, series maintenance should be sped up. If a score of 1 is returned,
// checkpointing based on dirty-series count should be disabled, and series
// files should not by synced anymore provided the user has specified the
// adaptive sync strategy.
func ( s * MemorySeriesStorage ) calculatePersistenceUrgencyScore ( ) float64 {
// It is safe to call this method concurrently.
func ( s * MemorySeriesStorage ) getPersistenceUrgencyScore ( ) ( float64 , bool ) {
s . rushedMtx . Lock ( )
defer s . rushedMtx . Unlock ( )
var (
chunksToPersist = float64 ( s . getNumChunksToPersist ( ) )
maxChunksToPersist = float64 ( s . maxChunksToPersist )
memChunks = float64 ( atomic . LoadInt64 ( & chunk . NumMemChunks ) )
maxMemChunks = float64 ( s . maxMemoryChunks )
)
score := chunksToPersist / maxChunksToPersist
if chunksToPersist > maxChunksToPersist * factorMinChunksToPersist {
score = math . Max (
score ,
( memChunks / maxMemChunks - 1 ) / ( toleranceFactorMemChunks - 1 ) ,
)
}
score := float64 ( atomic . LoadInt32 ( & s . persistUrgency ) ) / 1000
if score > 1 {
score = 1
}
s . persistenceUrgencyScore . Set ( score )
if s . rushed {
// We are already in rushed mode. If the score is still above
// persintenceUrgencyScoreForLeavingRushedMode, return 1 and
// leave things as they are.
// persintenceUrgencyScoreForLeavingRushedMode, return the score
// and leave things as they are.
if score > persintenceUrgencyScoreForLeavingRushedMode {
return 1
return score , true
}
// We are out of rushed mode!
s . rushed = false
s . rushedMode . Set ( 0 )
log .
With ( "urgencyScore" , score ) .
With ( "chunksToPersist" , int ( chunksToPersist ) ) .
With ( "maxChunksToPersist" , int ( maxChunksToPersist ) ) .
With ( "memoryChunks" , int ( memChunks ) ) .
With ( "maxMemoryChunks" , int ( maxMemChunks ) ) .
With ( "chunksToPersist" , s . getNumChunksToPersist ( ) ) .
With ( "memoryChunks" , atomic . LoadInt64 ( & chunk . NumMemChunks ) ) .
Info ( "Storage has left rushed mode." )
return score
return score , false
}
if score > persintenceUrgencyScoreForEnteringRushedMode {
// Enter rushed mode.
s . rushed = true
s . rushedMode . Set ( 1 )
log .
With ( "urgencyScore" , score ) .
With ( "chunksToPersist" , int ( chunksToPersist ) ) .
With ( "maxChunksToPersist" , int ( maxChunksToPersist ) ) .
With ( "memoryChunks" , int ( memChunks ) ) .
With ( "maxMemoryChunks" , int ( maxMemChunks ) ) .
With ( "chunksToPersist" , s . getNumChunksToPersist ( ) ) .
With ( "memoryChunks" , atomic . LoadInt64 ( & chunk . NumMemChunks ) ) .
Warn ( "Storage has entered rushed mode." )
return 1
}
return score
return score , s . rushed
}
// quarantineSeries registers the provided fingerprint for quarantining. It
@ -1742,10 +1857,10 @@ func (s *MemorySeriesStorage) purgeSeries(fp model.Fingerprint, m model.Metric,
if series , ok = s . fpToSeries . get ( fp ) ; ok {
s . fpToSeries . del ( fp )
s . num Series. Dec ( )
s . memory Series. Dec ( )
m = series . metric
// Adjust s. numC hunksToPersist and chunk.NumMemChunks down by
// Adjust s. c hunksToPersist and chunk.NumMemChunks down by
// the number of chunks in this series that are not
// persisted yet. Persisted chunks will be deducted from
// chunk.NumMemChunks upon eviction.
@ -1802,19 +1917,19 @@ func (s *MemorySeriesStorage) Describe(ch chan<- *prometheus.Desc) {
ch <- s . persistErrors . Desc ( )
ch <- s . queuedChunksToPersist . Desc ( )
ch <- maxChunksToPersistDesc
ch <- numChunksToPersistDesc
ch <- s . numSeries . Desc ( )
ch <- s . numHeadChunks . Desc ( )
ch <- s . chunksToPersist . Desc ( )
ch <- s . memorySeries . Desc ( )
ch <- s . headChunks . Desc ( )
ch <- s . dirtySeries . Desc ( )
s . seriesOps . Describe ( ch )
ch <- s . ingestedSamples Count . Desc ( )
s . discardedSamples Count . Describe ( ch )
ch <- s . nonExistentSeriesMatches Count . Desc ( )
ch <- chunk. NumMemChunksDesc
ch <- s . ingestedSamples . Desc ( )
s . discardedSamples . Describe ( ch )
ch <- s . nonExistentSeriesMatches . Desc ( )
ch <- s. memChunks . Desc ( )
s . maintainSeriesDuration . Describe ( ch )
ch <- s . persistenceUrgencyScore . Desc ( )
ch <- s . rushedMode . Desc ( )
ch <- s . targetHeapSizeBytes . Desc ( )
}
// Collect implements prometheus.Collector.
@ -1824,34 +1939,17 @@ func (s *MemorySeriesStorage) Collect(ch chan<- prometheus.Metric) {
ch <- s . persistErrors
ch <- s . queuedChunksToPersist
ch <- prometheus . MustNewConstMetric (
maxChunksToPersistDesc ,
prometheus . GaugeValue ,
float64 ( s . maxChunksToPersist ) ,
)
ch <- prometheus . MustNewConstMetric (
numChunksToPersistDesc ,
prometheus . GaugeValue ,
float64 ( s . getNumChunksToPersist ( ) ) ,
)
ch <- s . numSeries
ch <- s . numHeadChunks
ch <- s . chunksToPersist
ch <- s . memorySeries
ch <- s . headChunks
ch <- s . dirtySeries
s . seriesOps . Collect ( ch )
ch <- s . ingestedSamplesCount
s . discardedSamplesCount . Collect ( ch )
ch <- s . nonExistentSeriesMatchesCount
ch <- prometheus . MustNewConstMetric (
maxMemChunksDesc ,
prometheus . GaugeValue ,
float64 ( s . maxMemoryChunks ) ,
)
ch <- prometheus . MustNewConstMetric (
chunk . NumMemChunksDesc ,
prometheus . GaugeValue ,
float64 ( atomic . LoadInt64 ( & chunk . NumMemChunks ) ) ,
)
ch <- s . ingestedSamples
s . discardedSamples . Collect ( ch )
ch <- s . nonExistentSeriesMatches
ch <- s . memChunks
s . maintainSeriesDuration . Collect ( ch )
ch <- s . persistenceUrgencyScore
ch <- s . rushedMode
ch <- s . targetHeapSizeBytes
}
Björn Rabenstein
8 years ago
committed by