prometheus/storage/metric/memory.go

334 lines
8.7 KiB
Go
Raw Normal View History

2013-02-08 17:03:26 +00:00
// Copyright 2013 Prometheus Team
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package metric
import (
"github.com/prometheus/prometheus/model"
"github.com/prometheus/prometheus/storage"
"github.com/prometheus/prometheus/utility"
"github.com/ryszard/goskiplist/skiplist"
"time"
)
// Models a given sample entry stored in the in-memory arena.
type value interface {
// Gets the given value.
get() model.SampleValue
}
// Models a single sample value. It presumes that there is either no subsequent
// value seen or that any subsequent values are of a different value.
type singletonValue model.SampleValue
func (v singletonValue) get() model.SampleValue {
return model.SampleValue(v)
}
type skipListTime time.Time
func (t skipListTime) LessThan(o skiplist.Ordered) bool {
return time.Time(o.(skipListTime)).Before(time.Time(t))
}
type stream struct {
metric model.Metric
values *skiplist.SkipList
}
func (s *stream) add(timestamp time.Time, value model.SampleValue) {
2013-03-07 01:16:39 +00:00
s.values.Set(skipListTime(timestamp), singletonValue(value))
2013-02-08 17:03:26 +00:00
}
func (s *stream) forEach(decoder storage.RecordDecoder, filter storage.RecordFilter, operator storage.RecordOperator) (scannedEntireCorpus bool, err error) {
2013-03-07 01:16:39 +00:00
if s.values.Len() == 0 {
return false, nil
2013-02-08 17:03:26 +00:00
}
2013-03-07 01:16:39 +00:00
iterator := s.values.SeekToLast()
2013-02-08 17:03:26 +00:00
defer iterator.Close()
2013-03-07 01:16:39 +00:00
for !(iterator.Key() == nil || iterator.Value() == nil) {
2013-02-08 17:03:26 +00:00
decodedKey, decodeErr := decoder.DecodeKey(iterator.Key())
if decodeErr != nil {
2013-03-11 21:21:25 +00:00
panic(decodeErr)
2013-02-08 17:03:26 +00:00
}
decodedValue, decodeErr := decoder.DecodeValue(iterator.Value())
if decodeErr != nil {
2013-03-11 21:21:25 +00:00
panic(decodeErr)
2013-02-08 17:03:26 +00:00
}
switch filter.Filter(decodedKey, decodedValue) {
case storage.STOP:
return false, nil
2013-02-08 17:03:26 +00:00
case storage.SKIP:
continue
case storage.ACCEPT:
opErr := operator.Operate(decodedKey, decodedValue)
if opErr != nil {
if opErr.Continuable {
continue
}
break
}
}
2013-03-07 01:16:39 +00:00
if !iterator.Previous() {
break
}
2013-02-08 17:03:26 +00:00
}
return true, nil
2013-02-08 17:03:26 +00:00
}
func newStream(metric model.Metric) *stream {
return &stream{
2013-02-08 17:03:26 +00:00
values: skiplist.New(),
metric: metric,
}
}
type memorySeriesStorage struct {
fingerprintToSeries map[model.Fingerprint]*stream
labelPairToFingerprints map[model.LabelPair]model.Fingerprints
2013-02-08 17:03:26 +00:00
labelNameToFingerprints map[model.LabelName]model.Fingerprints
}
2013-05-16 14:02:07 +00:00
func (s *memorySeriesStorage) AppendSamples(samples model.Samples) error {
2013-02-08 17:03:26 +00:00
for _, sample := range samples {
s.AppendSample(sample)
}
2013-05-16 14:02:07 +00:00
return nil
2013-02-08 17:03:26 +00:00
}
2013-05-16 14:02:07 +00:00
func (s *memorySeriesStorage) AppendSample(sample model.Sample) error {
metric := sample.Metric
fingerprint := model.NewFingerprintFromMetric(metric)
2013-05-17 10:58:15 +00:00
series, ok := s.fingerprintToSeries[*fingerprint]
2013-02-08 17:03:26 +00:00
if !ok {
series = newStream(metric)
2013-05-17 10:58:15 +00:00
s.fingerprintToSeries[*fingerprint] = series
2013-02-08 17:03:26 +00:00
for k, v := range metric {
labelPair := model.LabelPair{
Name: k,
Value: v,
}
2013-02-08 17:03:26 +00:00
labelPairValues := s.labelPairToFingerprints[labelPair]
labelPairValues = append(labelPairValues, fingerprint)
s.labelPairToFingerprints[labelPair] = labelPairValues
labelNameValues := s.labelNameToFingerprints[k]
labelNameValues = append(labelNameValues, fingerprint)
s.labelNameToFingerprints[k] = labelNameValues
}
}
2013-03-07 01:16:39 +00:00
series.add(sample.Timestamp, sample.Value)
2013-02-08 17:03:26 +00:00
2013-05-16 14:02:07 +00:00
return nil
2013-02-08 17:03:26 +00:00
}
// Append raw sample, bypassing indexing. Only used to add data to views, which
// don't need to lookup by metric.
2013-05-17 10:58:15 +00:00
func (s *memorySeriesStorage) appendSampleWithoutIndexing(f *model.Fingerprint, timestamp time.Time, value model.SampleValue) {
series, ok := s.fingerprintToSeries[*f]
if !ok {
series = newStream(model.Metric{})
2013-05-17 10:58:15 +00:00
s.fingerprintToSeries[*f] = series
}
series.add(timestamp, value)
}
2013-05-16 14:02:07 +00:00
func (s *memorySeriesStorage) GetFingerprintsForLabelSet(l model.LabelSet) (fingerprints model.Fingerprints, err error) {
2013-02-08 17:03:26 +00:00
sets := []utility.Set{}
for k, v := range l {
values := s.labelPairToFingerprints[model.LabelPair{
Name: k,
Value: v,
}]
2013-02-08 17:03:26 +00:00
set := utility.Set{}
for _, fingerprint := range values {
2013-05-17 10:58:15 +00:00
set.Add(*fingerprint)
2013-02-08 17:03:26 +00:00
}
sets = append(sets, set)
}
setCount := len(sets)
if setCount == 0 {
return fingerprints, nil
2013-02-08 17:03:26 +00:00
}
base := sets[0]
for i := 1; i < setCount; i++ {
base = base.Intersection(sets[i])
}
for _, e := range base.Elements() {
fingerprint := e.(model.Fingerprint)
2013-05-17 10:58:15 +00:00
fingerprints = append(fingerprints, &fingerprint)
2013-02-08 17:03:26 +00:00
}
return fingerprints, nil
2013-02-08 17:03:26 +00:00
}
2013-05-20 18:31:58 +00:00
func (s *memorySeriesStorage) GetFingerprintsForLabelName(l model.LabelName) (fingerprints model.Fingerprints, _ error) {
2013-02-08 17:03:26 +00:00
values := s.labelNameToFingerprints[l]
fingerprints = append(fingerprints, values...)
return fingerprints, nil
2013-02-08 17:03:26 +00:00
}
2013-05-17 10:58:15 +00:00
func (s *memorySeriesStorage) GetMetricForFingerprint(f *model.Fingerprint) (model.Metric, error) {
series, ok := s.fingerprintToSeries[*f]
2013-02-08 17:03:26 +00:00
if !ok {
return nil, nil
2013-02-08 17:03:26 +00:00
}
metric := model.Metric{}
for label, value := range series.metric {
metric[label] = value
}
2013-02-08 17:03:26 +00:00
return metric, nil
2013-02-08 17:03:26 +00:00
}
2013-05-17 10:58:15 +00:00
func (s *memorySeriesStorage) GetValueAtTime(f *model.Fingerprint, t time.Time) (samples model.Values) {
series, ok := s.fingerprintToSeries[*f]
2013-02-08 17:03:26 +00:00
if !ok {
return samples
2013-02-08 17:03:26 +00:00
}
iterator := series.values.Seek(skipListTime(t))
if iterator == nil {
// If the iterator is nil, it means we seeked past the end of the series,
// so we seek to the last value instead. Due to the reverse ordering
// defined on skipListTime, this corresponds to the sample with the
// earliest timestamp.
iterator = series.values.SeekToLast()
if iterator == nil {
// The list is empty.
return samples
2013-02-08 17:03:26 +00:00
}
}
defer iterator.Close()
2013-02-08 17:03:26 +00:00
if iterator.Key() == nil || iterator.Value() == nil {
return samples
2013-02-08 17:03:26 +00:00
}
foundTime := time.Time(iterator.Key().(skipListTime))
samples = append(samples, model.SamplePair{
Timestamp: foundTime,
Value: iterator.Value().(value).get(),
})
if foundTime.Before(t) && iterator.Previous() {
samples = append(samples, model.SamplePair{
Timestamp: time.Time(iterator.Key().(skipListTime)),
Value: iterator.Value().(value).get(),
})
2013-02-08 17:03:26 +00:00
}
return samples
2013-02-08 17:03:26 +00:00
}
2013-05-17 10:58:15 +00:00
func (s *memorySeriesStorage) GetBoundaryValues(f *model.Fingerprint, i model.Interval) (model.Values, model.Values) {
return s.GetValueAtTime(f, i.OldestInclusive), s.GetValueAtTime(f, i.NewestInclusive)
2013-02-08 17:03:26 +00:00
}
2013-05-17 10:58:15 +00:00
func (s *memorySeriesStorage) GetRangeValues(f *model.Fingerprint, i model.Interval) (samples model.Values) {
series, ok := s.fingerprintToSeries[*f]
2013-02-08 17:03:26 +00:00
if !ok {
return samples
2013-02-08 17:03:26 +00:00
}
iterator := series.values.Seek(skipListTime(i.OldestInclusive))
2013-02-08 17:03:26 +00:00
if iterator == nil {
// If the iterator is nil, it means we seeked past the end of the series,
// so we seek to the last value instead. Due to the reverse ordering
// defined on skipListTime, this corresponds to the sample with the
// earliest timestamp.
iterator = series.values.SeekToLast()
if iterator == nil {
// The list is empty.
return samples
}
2013-02-08 17:03:26 +00:00
}
defer iterator.Close()
2013-02-08 17:03:26 +00:00
for {
timestamp := time.Time(iterator.Key().(skipListTime))
if timestamp.After(i.NewestInclusive) {
2013-02-08 17:03:26 +00:00
break
}
if !timestamp.Before(i.OldestInclusive) {
samples = append(samples, model.SamplePair{
2013-02-08 17:03:26 +00:00
Value: iterator.Value().(value).get(),
Timestamp: timestamp,
})
}
2013-02-08 17:03:26 +00:00
if !iterator.Previous() {
2013-02-08 17:03:26 +00:00
break
}
}
return samples
2013-02-08 17:03:26 +00:00
}
2013-05-20 18:31:58 +00:00
func (s *memorySeriesStorage) Close() {
s.fingerprintToSeries = map[model.Fingerprint]*stream{}
s.labelPairToFingerprints = map[model.LabelPair]model.Fingerprints{}
s.labelNameToFingerprints = map[model.LabelName]model.Fingerprints{}
2013-02-08 17:03:26 +00:00
}
2013-05-16 14:02:07 +00:00
func (s *memorySeriesStorage) GetAllValuesForLabel(labelName model.LabelName) (values model.LabelValues, err error) {
valueSet := map[model.LabelValue]bool{}
for _, series := range s.fingerprintToSeries {
if value, ok := series.metric[labelName]; ok {
2013-03-26 13:46:02 +00:00
if !valueSet[value] {
values = append(values, value)
valueSet[value] = true
}
}
}
return
2013-02-08 17:03:26 +00:00
}
2013-05-16 14:02:07 +00:00
func (s *memorySeriesStorage) ForEachSample(builder IteratorsForFingerprintBuilder) (err error) {
2013-02-08 17:03:26 +00:00
for _, stream := range s.fingerprintToSeries {
decoder, filter, operator := builder.ForStream(stream)
stream.forEach(decoder, filter, operator)
}
return
}
2013-05-20 18:31:58 +00:00
func NewMemorySeriesStorage() *memorySeriesStorage {
return &memorySeriesStorage{
fingerprintToSeries: make(map[model.Fingerprint]*stream),
labelPairToFingerprints: make(map[model.LabelPair]model.Fingerprints),
2013-02-08 17:03:26 +00:00
labelNameToFingerprints: make(map[model.LabelName]model.Fingerprints),
}
}