github
/
prometheus
mirror of https://github.com/prometheus/prometheus
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Address outstanding PR comments.

pull/84/head
Julius Volz 2013-03-11 14:21:25 -07:00 committed by Matt T. Proud
parent 62b5d7ce20
commit 1a1cba1bb2
3 changed files with 24 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
storage/metric/leveldb.go
View File

@ -188,7 +188,7 @@ func NewLevelDBMetricPersistence(baseDirectory string) (persistence *LevelDBMetr
func (l *LevelDBMetricPersistence) AppendSample(sample model.Sample) (err error) { func (l *LevelDBMetricPersistence) AppendSample(sample model.Sample) (err error) {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: appendSample, result: success}, map[string]string{operation: appendSample, result: failure}) recordOutcome(duration, err, map[string]string{operation: appendSample, result: success}, map[string]string{operation: appendSample, result: failure})
}() }()
@ -201,7 +201,7 @@ func (l *LevelDBMetricPersistence) AppendSample(sample model.Sample) (err error)
func (l *LevelDBMetricPersistence) AppendSamples(samples model.Samples) (err error) { func (l *LevelDBMetricPersistence) AppendSamples(samples model.Samples) (err error) {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: appendSamples, result: success}, map[string]string{operation: appendSamples, result: failure}) recordOutcome(duration, err, map[string]string{operation: appendSamples, result: success}, map[string]string{operation: appendSamples, result: failure})
}() }()
@ -303,7 +303,6 @@ func (l *LevelDBMetricPersistence) AppendSamples(samples model.Samples) (err err
indexHas, err := l.hasIndexMetric(metricDTO) indexHas, err := l.hasIndexMetric(metricDTO)
if err != nil { if err != nil {
panic(err) panic(err)
continue
} }
if !indexHas { if !indexHas {
absentFingerprints[fingerprint] = samples absentFingerprints[fingerprint] = samples
@ -570,7 +569,7 @@ func (l *LevelDBMetricPersistence) hasIndexMetric(dto *dto.Metric) (value bool,
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: hasIndexMetric, result: success}, map[string]string{operation: hasIndexMetric, result: failure}) recordOutcome(duration, err, map[string]string{operation: hasIndexMetric, result: success}, map[string]string{operation: hasIndexMetric, result: failure})
}() }()
@ -585,7 +584,7 @@ func (l *LevelDBMetricPersistence) HasLabelPair(dto *dto.LabelPair) (value bool,
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: hasLabelPair, result: success}, map[string]string{operation: hasLabelPair, result: failure}) recordOutcome(duration, err, map[string]string{operation: hasLabelPair, result: success}, map[string]string{operation: hasLabelPair, result: failure})
}() }()
@ -600,7 +599,7 @@ func (l *LevelDBMetricPersistence) HasLabelName(dto *dto.LabelName) (value bool,
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: hasLabelName, result: success}, map[string]string{operation: hasLabelName, result: failure}) recordOutcome(duration, err, map[string]string{operation: hasLabelName, result: success}, map[string]string{operation: hasLabelName, result: failure})
}() }()
@ -615,7 +614,7 @@ func (l *LevelDBMetricPersistence) GetFingerprintsForLabelSet(labelSet model.Lab
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: getFingerprintsForLabelSet, result: success}, map[string]string{operation: getFingerprintsForLabelSet, result: failure}) recordOutcome(duration, err, map[string]string{operation: getFingerprintsForLabelSet, result: success}, map[string]string{operation: getFingerprintsForLabelSet, result: failure})
}() }()
@ -665,7 +664,7 @@ func (l *LevelDBMetricPersistence) GetFingerprintsForLabelName(labelName model.L
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: getFingerprintsForLabelName, result: success}, map[string]string{operation: getFingerprintsForLabelName, result: failure}) recordOutcome(duration, err, map[string]string{operation: getFingerprintsForLabelName, result: success}, map[string]string{operation: getFingerprintsForLabelName, result: failure})
}() }()
@ -694,7 +693,7 @@ func (l *LevelDBMetricPersistence) GetMetricForFingerprint(f model.Fingerprint)
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: getMetricForFingerprint, result: success}, map[string]string{operation: getMetricForFingerprint, result: failure}) recordOutcome(duration, err, map[string]string{operation: getMetricForFingerprint, result: success}, map[string]string{operation: getMetricForFingerprint, result: failure})
}() }()
@ -727,7 +726,7 @@ func (l *LevelDBMetricPersistence) GetBoundaryValues(m model.Metric, i model.Int
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: getBoundaryValues, result: success}, map[string]string{operation: getBoundaryValues, result: failure}) recordOutcome(duration, err, map[string]string{operation: getBoundaryValues, result: success}, map[string]string{operation: getBoundaryValues, result: failure})
}() }()
@ -776,7 +775,7 @@ func (l *LevelDBMetricPersistence) GetValueAtTime(m model.Metric, t time.Time, s
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: getValueAtTime, result: success}, map[string]string{operation: getValueAtTime, result: failure}) recordOutcome(duration, err, map[string]string{operation: getValueAtTime, result: success}, map[string]string{operation: getValueAtTime, result: failure})
}() }()
@ -992,7 +991,7 @@ func (l *LevelDBMetricPersistence) GetRangeValues(m model.Metric, i model.Interv
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: getRangeValues, result: success}, map[string]string{operation: getRangeValues, result: failure}) recordOutcome(duration, err, map[string]string{operation: getRangeValues, result: success}, map[string]string{operation: getRangeValues, result: failure})
}() }()

4
storage/metric/memory.go
View File

@ -69,11 +69,11 @@ func (s stream) forEach(decoder storage.RecordDecoder, filter storage.RecordFilt
for !(iterator.Key() == nil || iterator.Value() == nil) { for !(iterator.Key() == nil || iterator.Value() == nil) {
decodedKey, decodeErr := decoder.DecodeKey(iterator.Key()) decodedKey, decodeErr := decoder.DecodeKey(iterator.Key())
if decodeErr != nil { if decodeErr != nil {
continue panic(decodeErr)
} }
decodedValue, decodeErr := decoder.DecodeValue(iterator.Value()) decodedValue, decodeErr := decoder.DecodeValue(iterator.Value())
if decodeErr != nil { if decodeErr != nil {
continue panic(decodeErr)
} }
switch filter.Filter(decodedKey, decodedValue) { switch filter.Filter(decodedKey, decodedValue) {

16
storage/metric/tiered.go
View File

@ -60,6 +60,7 @@ type Storage interface {
// Stops the storage subsystem, flushing all pending operations. // Stops the storage subsystem, flushing all pending operations.
Drain() Drain()
Flush() Flush()
Close()
} }
func NewTieredStorage(appendToMemoryQueueDepth, appendToDiskQueueDepth, viewQueueDepth uint, flushMemoryInterval, writeMemoryInterval, memoryTTL time.Duration, root string) Storage { func NewTieredStorage(appendToMemoryQueueDepth, appendToDiskQueueDepth, viewQueueDepth uint, flushMemoryInterval, writeMemoryInterval, memoryTTL time.Duration, root string) Storage {
@ -126,7 +127,7 @@ func (t *tieredStorage) MakeView(builder ViewRequestBuilder, deadline time.Durat
func (t *tieredStorage) rebuildDiskFrontier() (err error) { func (t *tieredStorage) rebuildDiskFrontier() (err error) {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: appendSample, result: success}, map[string]string{operation: rebuildDiskFrontier, result: failure}) recordOutcome(duration, err, map[string]string{operation: appendSample, result: success}, map[string]string{operation: rebuildDiskFrontier, result: failure})
}() }()
@ -181,7 +182,7 @@ func (t *tieredStorage) reportQueues() {
func (t *tieredStorage) writeMemory() { func (t *tieredStorage) writeMemory() {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, nil, map[string]string{operation: appendSample, result: success}, map[string]string{operation: writeMemory, result: failure}) recordOutcome(duration, nil, map[string]string{operation: appendSample, result: success}, map[string]string{operation: writeMemory, result: failure})
}() }()
@ -200,9 +201,14 @@ func (t *tieredStorage) Flush() {
t.flush() t.flush()
} }
func (t *tieredStorage) Close() {
t.Drain()
t.diskStorage.Close()
}
// Write all pending appends. // Write all pending appends.
func (t *tieredStorage) flush() (err error) { func (t *tieredStorage) flush() (err error) {
// Trim and old values to reduce iterative write costs. // Trim any old values to reduce iterative write costs.
t.flushMemory() t.flushMemory()
t.writeMemory() t.writeMemory()
t.flushMemory() t.flushMemory()
@ -299,7 +305,7 @@ func (f memoryToDiskFlusher) Close() {
func (t *tieredStorage) flushMemory() { func (t *tieredStorage) flushMemory() {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, nil, map[string]string{operation: appendSample, result: success}, map[string]string{operation: flushMemory, result: failure}) recordOutcome(duration, nil, map[string]string{operation: appendSample, result: success}, map[string]string{operation: flushMemory, result: failure})
}() }()
@ -322,7 +328,7 @@ func (t *tieredStorage) flushMemory() {
func (t *tieredStorage) renderView(viewJob viewJob) (err error) { func (t *tieredStorage) renderView(viewJob viewJob) (err error) {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Now().Sub(begin) duration := time.Since(begin)
recordOutcome(duration, err, map[string]string{operation: appendSample, result: success}, map[string]string{operation: renderView, result: failure}) recordOutcome(duration, err, map[string]string{operation: appendSample, result: success}, map[string]string{operation: renderView, result: failure})
}() }()