add upload delay metrics (#2713)

pkg/chunk/cached_store.go

@@ -837,7 +837,7 @@ func (store *cachedStore) uploadStagingFile(key string, stagingPath string) {
 	}()
 
 	store.pendingMutex.Lock()
-	_, ok := store.pendingKeys[key]
+	addTime, ok := store.pendingKeys[key]
 	store.pendingMutex.Unlock()
 	if !ok {
 		logger.Debugf("Key %s is not needed, drop it", key)
@@ -865,6 +865,9 @@ func (store *cachedStore) uploadStagingFile(key string, stagingPath string) {
 		return
 	}
 
+	if m, ok := store.bcache.(*cacheManager); ok {
+		m.stageBlockDelay.Add(time.Since(addTime).Seconds())
+	}
 	if err = store.upload(key, block, nil); err == nil {
 		store.bcache.uploaded(key, blen)
 		store.removePending(key)

pkg/chunk/disk_cache.go

@@ -590,6 +590,7 @@ type cacheManager struct {
 	cacheWriteHist  prometheus.Histogram
 	stageBlocks     prometheus.Gauge
 	stageBlockBytes prometheus.Gauge
+	stageBlockDelay prometheus.Counter
 }
 
 func keyHash(s string) uint32 {
@@ -704,6 +705,10 @@ func newCacheManager(config *Config, reg prometheus.Registerer, uploader func(ke
 			Name: "staging_block_bytes",
 			Help: "Total bytes of blocks in the staging path.",
 		}),
+		stageBlockDelay: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "staging_block_delay_seconds",
+			Help: "Total seconds of delay for staging blocks",
+		}),
 	}
 	if reg != nil {
 		reg.MustRegister(m.cacheWrites)
@@ -713,6 +718,7 @@ func newCacheManager(config *Config, reg prometheus.Registerer, uploader func(ke
 		reg.MustRegister(m.cacheWriteHist)
 		reg.MustRegister(m.stageBlocks)
 		reg.MustRegister(m.stageBlockBytes)
+		reg.MustRegister(m.stageBlockDelay)
 	}
 
 	// 20% of buffer could be used for pending pages