2014-06-06 09:55:53 +00:00
|
|
|
package storage_ng
|
|
|
|
|
|
|
|
import (
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/golang/glog"
|
|
|
|
|
|
|
|
clientmodel "github.com/prometheus/client_golang/model"
|
|
|
|
"github.com/prometheus/prometheus/storage/metric"
|
|
|
|
)
|
|
|
|
|
|
|
|
const persistQueueCap = 1024
|
|
|
|
|
|
|
|
type storageState uint
|
|
|
|
|
|
|
|
const (
|
|
|
|
storageStarting storageState = iota
|
|
|
|
storageServing
|
|
|
|
storageStopping
|
|
|
|
)
|
|
|
|
|
|
|
|
type memorySeriesStorage struct {
|
|
|
|
mtx sync.RWMutex
|
|
|
|
|
|
|
|
state storageState
|
|
|
|
persistDone chan bool
|
|
|
|
stopServing chan chan<- bool
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
fingerprintToSeries SeriesMap
|
2014-06-06 09:55:53 +00:00
|
|
|
|
|
|
|
memoryEvictionInterval time.Duration
|
|
|
|
memoryRetentionPeriod time.Duration
|
|
|
|
|
|
|
|
persistencePurgeInterval time.Duration
|
|
|
|
persistenceRetentionPeriod time.Duration
|
|
|
|
|
|
|
|
persistQueue chan *persistRequest
|
|
|
|
persistence Persistence
|
|
|
|
}
|
|
|
|
|
|
|
|
type MemorySeriesStorageOptions struct {
|
|
|
|
Persistence Persistence
|
|
|
|
MemoryEvictionInterval time.Duration
|
|
|
|
MemoryRetentionPeriod time.Duration
|
|
|
|
PersistencePurgeInterval time.Duration
|
|
|
|
PersistenceRetentionPeriod time.Duration
|
|
|
|
}
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) (Storage, error) {
|
|
|
|
glog.Info("Loading series map and head chunks...")
|
|
|
|
fingerprintToSeries, err := o.Persistence.LoadSeriesMapAndHeads()
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
numSeries.Set(float64(len(fingerprintToSeries)))
|
2014-06-06 09:55:53 +00:00
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
return &memorySeriesStorage{
|
|
|
|
fingerprintToSeries: fingerprintToSeries,
|
|
|
|
persistDone: make(chan bool),
|
|
|
|
stopServing: make(chan chan<- bool),
|
2014-06-06 09:55:53 +00:00
|
|
|
|
|
|
|
memoryEvictionInterval: o.MemoryEvictionInterval,
|
|
|
|
memoryRetentionPeriod: o.MemoryRetentionPeriod,
|
|
|
|
|
|
|
|
persistencePurgeInterval: o.PersistencePurgeInterval,
|
|
|
|
persistenceRetentionPeriod: o.PersistenceRetentionPeriod,
|
|
|
|
|
|
|
|
persistQueue: make(chan *persistRequest, persistQueueCap),
|
|
|
|
persistence: o.Persistence,
|
|
|
|
}, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
type persistRequest struct {
|
|
|
|
fingerprint clientmodel.Fingerprint
|
|
|
|
chunkDesc *chunkDesc
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) AppendSamples(samples clientmodel.Samples) {
|
|
|
|
/*
|
|
|
|
s.mtx.Lock()
|
|
|
|
defer s.mtx.Unlock()
|
|
|
|
if s.state != storageServing {
|
|
|
|
panic("storage is not serving")
|
|
|
|
}
|
|
|
|
s.mtx.Unlock()
|
|
|
|
*/
|
|
|
|
|
|
|
|
for _, sample := range samples {
|
|
|
|
s.appendSample(sample)
|
|
|
|
}
|
|
|
|
|
|
|
|
numSamples.Add(float64(len(samples)))
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) appendSample(sample *clientmodel.Sample) {
|
|
|
|
s.mtx.Lock()
|
|
|
|
defer s.mtx.Unlock()
|
|
|
|
|
|
|
|
series := s.getOrCreateSeries(sample.Metric)
|
|
|
|
series.add(&metric.SamplePair{
|
|
|
|
Value: sample.Value,
|
|
|
|
Timestamp: sample.Timestamp,
|
|
|
|
}, s.persistQueue)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) getOrCreateSeries(m clientmodel.Metric) *memorySeries {
|
|
|
|
fp := m.Fingerprint()
|
|
|
|
series, ok := s.fingerprintToSeries[fp]
|
|
|
|
|
|
|
|
if !ok {
|
|
|
|
series = newMemorySeries(m)
|
|
|
|
s.fingerprintToSeries[fp] = series
|
|
|
|
numSeries.Set(float64(len(s.fingerprintToSeries)))
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
unarchived, err := s.persistence.UnarchiveMetric(fp)
|
|
|
|
if err != nil {
|
|
|
|
glog.Errorf("Error unarchiving fingerprint %v: %v", fp, err)
|
|
|
|
}
|
2014-06-06 09:55:53 +00:00
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
if unarchived {
|
|
|
|
// The series existed before, had been archived at some
|
|
|
|
// point, and has now been unarchived, i.e. it has
|
|
|
|
// chunks on disk. Set chunkDescsLoaded accordingly so
|
|
|
|
// that they will be looked at later.
|
|
|
|
series.chunkDescsLoaded = false
|
|
|
|
} else {
|
|
|
|
// This was a genuinely new series, so index the metric.
|
|
|
|
if err := s.persistence.IndexMetric(m); err != nil {
|
|
|
|
glog.Errorf("Error indexing metric %v: %v", m, err)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return series
|
|
|
|
}
|
|
|
|
|
|
|
|
/*
|
|
|
|
func (s *memorySeriesStorage) preloadChunksAtTime(fp clientmodel.Fingerprint, ts clientmodel.Timestamp) (chunkDescs, error) {
|
|
|
|
series, ok := s.fingerprintToSeries[fp]
|
|
|
|
if !ok {
|
|
|
|
panic("requested preload for non-existent series")
|
|
|
|
}
|
|
|
|
return series.preloadChunksAtTime(ts, s.persistence)
|
|
|
|
}
|
|
|
|
*/
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) preloadChunksForRange(fp clientmodel.Fingerprint, from clientmodel.Timestamp, through clientmodel.Timestamp) (chunkDescs, error) {
|
|
|
|
s.mtx.RLock()
|
|
|
|
series, ok := s.fingerprintToSeries[fp]
|
|
|
|
s.mtx.RUnlock()
|
|
|
|
|
|
|
|
if !ok {
|
|
|
|
panic("requested preload for non-existent series")
|
|
|
|
}
|
|
|
|
return series.preloadChunksForRange(from, through, s.persistence)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) NewIterator(fp clientmodel.Fingerprint) SeriesIterator {
|
|
|
|
s.mtx.RLock()
|
|
|
|
series, ok := s.fingerprintToSeries[fp]
|
|
|
|
s.mtx.RUnlock()
|
|
|
|
|
|
|
|
if !ok {
|
|
|
|
panic("requested iterator for non-existent series")
|
|
|
|
}
|
|
|
|
return series.newIterator()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) evictMemoryChunks(ttl time.Duration) {
|
|
|
|
s.mtx.RLock()
|
|
|
|
defer s.mtx.RUnlock()
|
|
|
|
|
|
|
|
for _, series := range s.fingerprintToSeries {
|
|
|
|
series.evictOlderThan(clientmodel.TimestampFromTime(time.Now()).Add(-1 * ttl))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func recordPersist(start time.Time, err error) {
|
|
|
|
outcome := success
|
|
|
|
if err != nil {
|
|
|
|
outcome = failure
|
|
|
|
}
|
|
|
|
persistLatencies.WithLabelValues(outcome).Observe(float64(time.Since(start) / time.Millisecond))
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) handlePersistQueue() {
|
2014-09-10 16:41:52 +00:00
|
|
|
// TODO: Perhaps move this into Persistence?
|
2014-06-06 09:55:53 +00:00
|
|
|
for req := range s.persistQueue {
|
|
|
|
// TODO: Make this thread-safe?
|
|
|
|
persistQueueLength.Set(float64(len(s.persistQueue)))
|
|
|
|
|
|
|
|
//glog.Info("Persist request: ", *req.fingerprint)
|
|
|
|
start := time.Now()
|
|
|
|
err := s.persistence.PersistChunk(req.fingerprint, req.chunkDesc.chunk)
|
|
|
|
recordPersist(start, err)
|
|
|
|
if err != nil {
|
|
|
|
glog.Error("Error persisting chunk, requeuing: ", err)
|
|
|
|
s.persistQueue <- req
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
req.chunkDesc.unpin()
|
|
|
|
}
|
|
|
|
s.persistDone <- true
|
|
|
|
}
|
|
|
|
|
|
|
|
// Close stops serving, flushes all pending operations, and frees all resources.
|
|
|
|
func (s *memorySeriesStorage) Close() error {
|
|
|
|
s.mtx.Lock()
|
|
|
|
defer s.mtx.Unlock()
|
|
|
|
|
|
|
|
if s.state == storageStopping {
|
|
|
|
panic("Illegal State: Attempted to restop memorySeriesStorage.")
|
|
|
|
}
|
|
|
|
|
|
|
|
stopped := make(chan bool)
|
|
|
|
glog.Info("Waiting for storage to stop serving...")
|
|
|
|
s.stopServing <- (stopped)
|
|
|
|
glog.Info("Serving stopped.")
|
|
|
|
<-stopped
|
|
|
|
|
|
|
|
glog.Info("Stopping persist loop...")
|
|
|
|
close(s.persistQueue)
|
|
|
|
<-s.persistDone
|
|
|
|
glog.Info("Persist loop stopped.")
|
|
|
|
|
|
|
|
glog.Info("Persisting head chunks...")
|
2014-09-10 16:41:52 +00:00
|
|
|
if err := s.persistence.PersistSeriesMapAndHeads(s.fingerprintToSeries); err != nil {
|
2014-06-06 09:55:53 +00:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
glog.Info("Done persisting head chunks.")
|
|
|
|
|
|
|
|
s.fingerprintToSeries = nil
|
2014-09-10 16:41:52 +00:00
|
|
|
if err := s.persistence.Close(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2014-06-06 09:55:53 +00:00
|
|
|
|
|
|
|
s.state = storageStopping
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) purgePeriodically(stop <-chan bool) {
|
|
|
|
purgeTicker := time.NewTicker(s.persistencePurgeInterval)
|
|
|
|
defer purgeTicker.Stop()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-stop:
|
|
|
|
return
|
|
|
|
case <-purgeTicker.C:
|
|
|
|
glog.Info("Purging old series data...")
|
|
|
|
s.mtx.RLock()
|
|
|
|
fps := make([]clientmodel.Fingerprint, 0, len(s.fingerprintToSeries))
|
2014-08-12 15:46:46 +00:00
|
|
|
for fp := range s.fingerprintToSeries {
|
2014-06-06 09:55:53 +00:00
|
|
|
fps = append(fps, fp)
|
|
|
|
}
|
|
|
|
s.mtx.RUnlock()
|
|
|
|
|
|
|
|
for _, fp := range fps {
|
|
|
|
select {
|
|
|
|
case <-stop:
|
|
|
|
glog.Info("Interrupted running series purge.")
|
|
|
|
return
|
|
|
|
default:
|
2014-09-10 16:41:52 +00:00
|
|
|
s.purgeSeries(fp)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
glog.Info("Done purging old series data.")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
// purgeSeries purges chunks older than persistenceRetentionPeriod from a
|
|
|
|
// series. If the series contains no chunks after the purge, it is dropped
|
|
|
|
// entirely.
|
|
|
|
func (s *memorySeriesStorage) purgeSeries(fp clientmodel.Fingerprint) {
|
|
|
|
ts := clientmodel.TimestampFromTime(time.Now()).Add(-1 * s.persistenceRetentionPeriod)
|
2014-06-06 09:55:53 +00:00
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
s.mtx.Lock()
|
|
|
|
// TODO: This is a lock FAR to coarse! However, we cannot lock using the
|
|
|
|
// memorySeries since we might have none (for series that are on disk
|
|
|
|
// only). And we really don't want to un-archive a series from disk
|
|
|
|
// while we are at the same time purging it. A locking per fingerprint
|
|
|
|
// would be nice. Or something... Have to think about it... Careful,
|
|
|
|
// more race conditions lurk below. Also unsolved: If there are chunks
|
|
|
|
// in the persist queue. persistence.DropChunks and
|
|
|
|
// persistence.PersistChunck needs to be locked on fp level, or
|
|
|
|
// something. And even then, what happens if everything is dropped, but
|
|
|
|
// there are still chunks hung in the persist queue? They would later
|
|
|
|
// re-create a file for a series that doesn't exist anymore...
|
|
|
|
defer s.mtx.Unlock()
|
|
|
|
|
|
|
|
// First purge persisted chunks. We need to do that anyway.
|
|
|
|
allDropped, err := s.persistence.DropChunks(fp, ts)
|
2014-06-06 09:55:53 +00:00
|
|
|
if err != nil {
|
2014-09-10 16:41:52 +00:00
|
|
|
glog.Error("Error purging persisted chunks: ", err)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
// Purge chunks from memory accordingly.
|
|
|
|
if series, ok := s.fingerprintToSeries[fp]; ok {
|
|
|
|
if series.purgeOlderThan(ts) {
|
|
|
|
delete(s.fingerprintToSeries, fp)
|
|
|
|
if err := s.persistence.UnindexMetric(series.metric); err != nil {
|
|
|
|
glog.Errorf("Error unindexing metric %v: %v", series.metric, err)
|
|
|
|
}
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
return
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
// If nothing was in memory, the metric must have been archived. Drop
|
|
|
|
// the archived metric if there are no persisted chunks left.
|
|
|
|
if !allDropped {
|
2014-06-06 09:55:53 +00:00
|
|
|
return
|
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
if err := s.persistence.DropArchivedMetric(fp); err != nil {
|
|
|
|
glog.Errorf("Error dropping archived metric for fingerprint %v: %v", fp, err)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) Serve(started chan<- bool) {
|
|
|
|
s.mtx.Lock()
|
|
|
|
if s.state != storageStarting {
|
|
|
|
panic("Illegal State: Attempted to restart memorySeriesStorage.")
|
|
|
|
}
|
|
|
|
s.state = storageServing
|
|
|
|
s.mtx.Unlock()
|
|
|
|
|
|
|
|
evictMemoryTicker := time.NewTicker(s.memoryEvictionInterval)
|
|
|
|
defer evictMemoryTicker.Stop()
|
|
|
|
|
|
|
|
go s.handlePersistQueue()
|
|
|
|
|
|
|
|
stopPurge := make(chan bool)
|
|
|
|
go s.purgePeriodically(stopPurge)
|
|
|
|
|
|
|
|
started <- true
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-evictMemoryTicker.C:
|
|
|
|
s.evictMemoryChunks(s.memoryRetentionPeriod)
|
|
|
|
case stopped := <-s.stopServing:
|
|
|
|
stopPurge <- true
|
|
|
|
stopped <- true
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) NewPreloader() Preloader {
|
|
|
|
return &memorySeriesPreloader{
|
|
|
|
storage: s,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) GetFingerprintsForLabelMatchers(labelMatchers metric.LabelMatchers) clientmodel.Fingerprints {
|
|
|
|
s.mtx.RLock()
|
|
|
|
defer s.mtx.RUnlock()
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
var result map[clientmodel.Fingerprint]struct{}
|
2014-06-06 09:55:53 +00:00
|
|
|
for _, matcher := range labelMatchers {
|
2014-09-10 16:41:52 +00:00
|
|
|
intersection := map[clientmodel.Fingerprint]struct{}{}
|
2014-06-06 09:55:53 +00:00
|
|
|
switch matcher.Type {
|
|
|
|
case metric.Equal:
|
2014-09-10 16:41:52 +00:00
|
|
|
fps, err := s.persistence.GetFingerprintsForLabelPair(
|
|
|
|
metric.LabelPair{
|
|
|
|
Name: matcher.Name,
|
|
|
|
Value: matcher.Value,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
glog.Error("Error getting fingerprints for label pair: ", err)
|
|
|
|
}
|
|
|
|
if len(fps) == 0 {
|
2014-06-06 09:55:53 +00:00
|
|
|
return nil
|
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
for _, fp := range fps {
|
|
|
|
if _, ok := result[fp]; ok || result == nil {
|
|
|
|
intersection[fp] = struct{}{}
|
|
|
|
}
|
|
|
|
}
|
2014-06-06 09:55:53 +00:00
|
|
|
default:
|
2014-09-10 16:41:52 +00:00
|
|
|
values, err := s.persistence.GetLabelValuesForLabelName(matcher.Name)
|
|
|
|
if err != nil {
|
|
|
|
glog.Errorf("Error getting label values for label name %q: %v", matcher.Name, err)
|
|
|
|
}
|
2014-06-06 09:55:53 +00:00
|
|
|
matches := matcher.Filter(values)
|
|
|
|
if len(matches) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
for _, v := range matches {
|
2014-09-10 16:41:52 +00:00
|
|
|
fps, err := s.persistence.GetFingerprintsForLabelPair(
|
|
|
|
metric.LabelPair{
|
|
|
|
Name: matcher.Name,
|
|
|
|
Value: v,
|
|
|
|
},
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
glog.Error("Error getting fingerprints for label pair: ", err)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
for _, fp := range fps {
|
|
|
|
if _, ok := result[fp]; ok || result == nil {
|
|
|
|
intersection[fp] = struct{}{}
|
|
|
|
}
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
if len(intersection) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
result = intersection
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
fps := make(clientmodel.Fingerprints, 0, len(result))
|
|
|
|
for fp := range result {
|
|
|
|
fps = append(fps, fp)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
return fps
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) GetLabelValuesForLabelName(labelName clientmodel.LabelName) clientmodel.LabelValues {
|
|
|
|
s.mtx.RLock()
|
|
|
|
defer s.mtx.RUnlock()
|
|
|
|
|
2014-09-10 16:41:52 +00:00
|
|
|
lvs, err := s.persistence.GetLabelValuesForLabelName(labelName)
|
|
|
|
if err != nil {
|
|
|
|
glog.Errorf("Error getting label values for label name %q: %v", labelName, err)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
return lvs
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) GetMetricForFingerprint(fp clientmodel.Fingerprint) clientmodel.Metric {
|
|
|
|
s.mtx.RLock()
|
|
|
|
defer s.mtx.RUnlock()
|
|
|
|
|
|
|
|
series, ok := s.fingerprintToSeries[fp]
|
2014-09-10 16:41:52 +00:00
|
|
|
if ok {
|
|
|
|
// TODO: Does this have to be a copy? Ask Julius!
|
|
|
|
return series.metric
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
2014-09-10 16:41:52 +00:00
|
|
|
metric, err := s.persistence.GetArchivedMetric(fp)
|
|
|
|
if err != nil {
|
|
|
|
glog.Errorf("Error retrieving archived metric for fingerprint %v: %v", fp, err)
|
2014-06-06 09:55:53 +00:00
|
|
|
}
|
|
|
|
return metric
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *memorySeriesStorage) GetAllValuesForLabel(labelName clientmodel.LabelName) clientmodel.LabelValues {
|
|
|
|
s.mtx.RLock()
|
|
|
|
defer s.mtx.RUnlock()
|
|
|
|
|
|
|
|
var values clientmodel.LabelValues
|
|
|
|
valueSet := map[clientmodel.LabelValue]struct{}{}
|
|
|
|
for _, series := range s.fingerprintToSeries {
|
|
|
|
if value, ok := series.metric[labelName]; ok {
|
|
|
|
if _, ok := valueSet[value]; !ok {
|
|
|
|
values = append(values, value)
|
|
|
|
valueSet[value] = struct{}{}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return values
|
|
|
|
}
|