mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2024-11-09 23:24:05 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Beginnings of a tiered index implementation.

Browse source 
This reintroduces a LevelDB-based metrics index.

Change-Id: I4111540301c52255a07b2f570761707a32f72c05
This commit is contained in:
Julius Volz 2014-08-21 22:06:11 +02:00 committed by Bjoern Rabenstein
parent 8dfaa5ecd2
commit 7e85711df0
13 changed files with 1389 additions and 562 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.build/Makefile
View file

@ -45,7 +45,7 @@ cc-implementation-Linux-stamp:
[ -x "$$(which cc)" ] || $(APT_GET_INSTALL) build-essential
touch $@
dependencies-stamp: cache-stamp cc-stamp leveldb-stamp snappy-stamp godns-stamp
dependencies-stamp: cache-stamp cc-stamp leveldb-stamp snappy-stamp godns-stamp goleveldb-stamp
touch $@
goprotobuf-protoc-gen-go-stamp: protoc-stamp goprotobuf-stamp
@ -60,6 +60,10 @@ godns-stamp:
$(GO_GET) github.com/miekg/dns $(THIRD_PARTY_BUILD_OUTPUT)
touch $@
goleveldb-stamp:
$(GO_GET) github.com/syndtr/goleveldb/leveldb $(THIRD_PARTY_BUILD_OUTPUT)
touch $@
leveldb-stamp: cache-stamp cache/leveldb-$(LEVELDB_VERSION).tar.gz cc-stamp rsync-stamp snappy-stamp
tar xzvf cache/leveldb-$(LEVELDB_VERSION).tar.gz -C dirty $(THIRD_PARTY_BUILD_OUTPUT)
cd dirty/leveldb-$(LEVELDB_VERSION) && CFLAGS="$(CFLAGS) -lsnappy" CXXFLAGS="$(CXXFLAGS) -lsnappy $(LDFLAGS)" LDFLAGS="-lsnappy $(LDFLAGS)" bash -x ./build_detect_platform build_config.mk ./

21
storage/local/index/batch.go Normal file
View file

@ -0,0 +1,21 @@
package index
import (
"github.com/syndtr/goleveldb/leveldb"
)
type batch struct {
batch *leveldb.Batch
}
func (b *batch) Put(key, value encodable) {
b.batch.Put(key.encode(), value.encode())
}
func (b *batch) Delete(k encodable) {
b.batch.Delete(k.encode())
}
func (b *batch) Reset() {
b.batch.Reset()
}

200
storage/local/index/codec.go Normal file
View file

@ -0,0 +1,200 @@
package index
import (
"bytes"
"encoding/binary"
"io"
"sync"
clientmodel "github.com/prometheus/client_golang/model"
"github.com/prometheus/prometheus/storage/metric"
)
type codable interface {
encodable
decodable
}
type encodable interface {
encode() []byte
}
type decodable interface {
decode([]byte)
}
// TODO: yeah, this ain't ideal. A lot of locking and possibly even contention.
var tmpBufMtx sync.Mutex
var tmpBuf = make([]byte, binary.MaxVarintLen64)
func setTmpBufLen(l int) {
if cap(tmpBuf) >= l {
tmpBuf = tmpBuf[:l]
} else {
tmpBuf = make([]byte, l)
}
}
func encodeVarint(b *bytes.Buffer, i int) {
tmpBufMtx.Lock()
defer tmpBufMtx.Unlock()
bytesWritten := binary.PutVarint(tmpBuf, int64(i))
if _, err := b.Write(tmpBuf[:bytesWritten]); err != nil {
panic(err)
}
}
func encodeString(b *bytes.Buffer, s string) {
encodeVarint(b, len(s))
if _, err := b.WriteString(s); err != nil {
panic(err)
}
}
func decodeString(b *bytes.Reader) string {
length, err := binary.ReadVarint(b)
if err != nil {
panic(err)
}
tmpBufMtx.Lock()
defer tmpBufMtx.Unlock()
setTmpBufLen(int(length))
if _, err := io.ReadFull(b, tmpBuf); err != nil {
panic(err)
}
return string(tmpBuf)
}
type codableMetric clientmodel.Metric
func (m codableMetric) encode() []byte {
buf := &bytes.Buffer{}
encodeVarint(buf, len(m))
for l, v := range m {
encodeString(buf, string(l))
encodeString(buf, string(v))
}
return buf.Bytes()
}
func (m codableMetric) decode(buf []byte) {
r := bytes.NewReader(buf)
numLabelPairs, err := binary.ReadVarint(r)
if err != nil {
panic(err)
}
for ; numLabelPairs > 0; numLabelPairs-- {
ln := decodeString(r)
lv := decodeString(r)
m[clientmodel.LabelName(ln)] = clientmodel.LabelValue(lv)
}
}
type codableFingerprint clientmodel.Fingerprint
func (fp codableFingerprint) encode() []byte {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fp))
return b
}
func (fp *codableFingerprint) decode(buf []byte) {
*fp = codableFingerprint(binary.BigEndian.Uint64(buf))
}
type codableFingerprints clientmodel.Fingerprints
func (fps codableFingerprints) encode() []byte {
buf := bytes.NewBuffer(make([]byte, 0, binary.MaxVarintLen64+len(fps)*8))
encodeVarint(buf, len(fps))
tmpBufMtx.Lock()
defer tmpBufMtx.Unlock()
setTmpBufLen(8)
for _, fp := range fps {
binary.BigEndian.PutUint64(tmpBuf, uint64(fp))
if _, err := buf.Write(tmpBuf[:8]); err != nil {
panic(err)
}
}
return buf.Bytes()
}
func (fps *codableFingerprints) decode(buf []byte) {
r := bytes.NewReader(buf)
numFPs, err := binary.ReadVarint(r)
if err != nil {
panic(err)
}
*fps = make(codableFingerprints, numFPs)
offset := len(buf) - r.Len()
for i, _ := range *fps {
(*fps)[i] = clientmodel.Fingerprint(binary.BigEndian.Uint64(buf[offset+i*8:]))
}
}
type codableLabelPair metric.LabelPair
func (lp codableLabelPair) encode() []byte {
buf := &bytes.Buffer{}
encodeString(buf, string(lp.Name))
encodeString(buf, string(lp.Value))
return buf.Bytes()
}
func (lp *codableLabelPair) decode(buf []byte) {
r := bytes.NewReader(buf)
lp.Name = clientmodel.LabelName(decodeString(r))
lp.Value = clientmodel.LabelValue(decodeString(r))
}
type codableLabelName clientmodel.LabelName
func (l codableLabelName) encode() []byte {
buf := &bytes.Buffer{}
encodeString(buf, string(l))
return buf.Bytes()
}
func (l *codableLabelName) decode(buf []byte) {
r := bytes.NewReader(buf)
*l = codableLabelName(decodeString(r))
}
type codableLabelValues clientmodel.LabelValues
func (vs codableLabelValues) encode() []byte {
buf := &bytes.Buffer{}
encodeVarint(buf, len(vs))
for _, v := range vs {
encodeString(buf, string(v))
}
return buf.Bytes()
}
func (vs *codableLabelValues) decode(buf []byte) {
r := bytes.NewReader(buf)
numValues, err := binary.ReadVarint(r)
if err != nil {
panic(err)
}
*vs = make(codableLabelValues, numValues)
for i, _ := range *vs {
(*vs)[i] = clientmodel.LabelValue(decodeString(r))
}
}
type codableMembership struct{}
func (m codableMembership) encode() []byte {
return []byte{}
}
func (m codableMembership) decode(buf []byte) {}

112
storage/local/index/codec_test.go Normal file
View file

@ -0,0 +1,112 @@
package index
import (
"testing"
clientmodel "github.com/prometheus/client_golang/model"
)
func newCodableFingerprint(fp int64) *codableFingerprint {
cfp := codableFingerprint(fp)
return &cfp
}
func newCodableLabelName(ln string) *codableLabelName {
cln := codableLabelName(ln)
return &cln
}
func TestCodec(t *testing.T) {
scenarios := []struct {
in codable
out codable
equal func(in, out codable) bool
}{
{
in: codableMetric{
"label_1": "value_2",
"label_2": "value_2",
"label_3": "value_3",
},
out: codableMetric{},
equal: func(in, out codable) bool {
m1 := clientmodel.Metric(in.(codableMetric))
m2 := clientmodel.Metric(out.(codableMetric))
return m1.Equal(m2)
},
}, {
in: newCodableFingerprint(12345),
out: newCodableFingerprint(0),
equal: func(in, out codable) bool {
return *in.(*codableFingerprint) == *out.(*codableFingerprint)
},
}, {
in: &codableFingerprints{1, 2, 56, 1234},
out: &codableFingerprints{},
equal: func(in, out codable) bool {
fps1 := *in.(*codableFingerprints)
fps2 := *out.(*codableFingerprints)
if len(fps1) != len(fps2) {
return false
}
for i, _ := range fps1 {
if fps1[i] != fps2[i] {
return false
}
}
return true
},
}, {
in: &codableLabelPair{
Name: "label_name",
Value: "label_value",
},
out: &codableLabelPair{},
equal: func(in, out codable) bool {
lp1 := *in.(*codableLabelPair)
lp2 := *out.(*codableLabelPair)
return lp1 == lp2
},
}, {
in: newCodableLabelName("label_name"),
out: newCodableLabelName(""),
equal: func(in, out codable) bool {
ln1 := *in.(*codableLabelName)
ln2 := *out.(*codableLabelName)
return ln1 == ln2
},
}, {
in: &codableLabelValues{"value_1", "value_2", "value_3"},
out: &codableLabelValues{},
equal: func(in, out codable) bool {
lvs1 := *in.(*codableLabelValues)
lvs2 := *out.(*codableLabelValues)
if len(lvs1) != len(lvs2) {
return false
}
for i, _ := range lvs1 {
if lvs1[i] != lvs2[i] {
return false
}
}
return true
},
}, {
in: &codableMembership{},
out: &codableMembership{},
equal: func(in, out codable) bool {
// We don't care about the membership value. Just test if the
// encoding/decoding works at all.
return true
},
},
}
for i, s := range scenarios {
encoded := s.in.encode()
s.out.decode(encoded)
if !s.equal(s.in, s.out) {
t.Fatalf("%d. Got: %v; want %v; encoded bytes are: %v", i, s.out, s.in, encoded)
}
}
}

561
storage/local/index/index.go Normal file
View file

@ -0,0 +1,561 @@
package index
import (
"io"
"sync"
clientmodel "github.com/prometheus/client_golang/model"
"github.com/prometheus/prometheus/storage/metric"
"github.com/prometheus/prometheus/utility"
)
// FingerprintMetricMapping is an in-memory map of fingerprints to metrics.
type FingerprintMetricMapping map[clientmodel.Fingerprint]clientmodel.Metric
// FingerprintMetricIndex models a database mapping fingerprints to metrics.
type FingerprintMetricIndex struct {
KeyValueStore
}
// IndexBatch indexes a batch of mappings from fingerprints to metrics.
func (i *FingerprintMetricIndex) IndexBatch(mapping FingerprintMetricMapping) error {
b := i.NewBatch()
for fp, m := range mapping {
b.Put(codableFingerprint(fp), codableMetric(m))
}
return i.Commit(b)
}
// UnindexBatch unindexes a batch of mappings from fingerprints to metrics.
func (i *FingerprintMetricIndex) UnindexBatch(mapping FingerprintMetricMapping) error {
b := i.NewBatch()
for fp, _ := range mapping {
b.Delete(codableFingerprint(fp))
}
return i.Commit(b)
}
// Lookup looks up a metric by fingerprint.
func (i *FingerprintMetricIndex) Lookup(fp clientmodel.Fingerprint) (m clientmodel.Metric, ok bool, err error) {
m = clientmodel.Metric{}
if ok, err := i.Get(codableFingerprint(fp), codableMetric(m)); !ok {
return nil, false, nil
} else if err != nil {
return nil, false, err
}
return m, true, nil
}
// NewFingerprintMetricIndex returns a FingerprintMetricIndex
// object ready to use.
func NewFingerprintMetricIndex(db KeyValueStore) *FingerprintMetricIndex {
return &FingerprintMetricIndex{
KeyValueStore: db,
}
}
// LabelNameLabelValuesMapping is an in-memory map of label names to
// label values.
type LabelNameLabelValuesMapping map[clientmodel.LabelName]clientmodel.LabelValues
// LabelNameLabelValuesIndex models a database mapping label names to
// label values.
type LabelNameLabelValuesIndex struct {
KeyValueStore
}
// IndexBatch implements LabelNameLabelValuesIndex.
func (i *LabelNameLabelValuesIndex) IndexBatch(b LabelNameLabelValuesMapping) error {
batch := i.NewBatch()
for name, values := range b {
if len(values) == 0 {
batch.Delete(codableLabelName(name))
} else {
batch.Put(codableLabelName(name), codableLabelValues(values))
}
}
return i.Commit(batch)
}
// Lookup looks up all label values for a given label name.
func (i *LabelNameLabelValuesIndex) Lookup(l clientmodel.LabelName) (values clientmodel.LabelValues, ok bool, err error) {
ok, err = i.Get(codableLabelName(l), (*codableLabelValues)(&values))
if err != nil {
return nil, false, err
}
if !ok {
return nil, false, nil
}
return values, true, nil
}
// NewLabelNameLabelValuesIndex returns a LabelNameLabelValuesIndex
// ready to use.
func NewLabelNameLabelValuesIndex(db KeyValueStore) *LabelNameLabelValuesIndex {
return &LabelNameLabelValuesIndex{
KeyValueStore: db,
}
}
// LabelPairFingerprintsMapping is an in-memory map of label pairs to
// fingerprints.
type LabelPairFingerprintsMapping map[metric.LabelPair]clientmodel.Fingerprints
// LabelPairFingerprintIndex models a database mapping label pairs to
// fingerprints.
type LabelPairFingerprintIndex struct {
KeyValueStore
}
// IndexBatch indexes a batch of mappings from label pairs to fingerprints.
func (i *LabelPairFingerprintIndex) IndexBatch(m LabelPairFingerprintsMapping) error {
batch := i.NewBatch()
for pair, fps := range m {
if len(fps) == 0 {
batch.Delete(codableLabelPair(pair))
} else {
batch.Put(codableLabelPair(pair), codableFingerprints(fps))
}
}
return i.Commit(batch)
}
// Lookup looks up all fingerprints for a given label pair.
func (i *LabelPairFingerprintIndex) Lookup(p *metric.LabelPair) (fps clientmodel.Fingerprints, ok bool, err error) {
ok, err = i.Get((*codableLabelPair)(p), (*codableFingerprints)(&fps))
if !ok {
return nil, false, nil
}
if err != nil {
return nil, false, err
}
return fps, true, nil
}
// NewLabelPairFingerprintIndex returns a LabelPairFingerprintIndex
// object ready to use.
func NewLabelPairFingerprintIndex(db KeyValueStore) *LabelPairFingerprintIndex {
return &LabelPairFingerprintIndex{
KeyValueStore: db,
}
}
// FingerprintMembershipIndex models a database tracking the existence
// of metrics by their fingerprints.
type FingerprintMembershipIndex struct {
KeyValueStore
}
// IndexBatch indexes a batch of fingerprints.
func (i *FingerprintMembershipIndex) IndexBatch(b FingerprintMetricMapping) error {
batch := i.NewBatch()
for fp, _ := range b {
batch.Put(codableFingerprint(fp), codableMembership{})
}
return i.Commit(batch)
}
// UnindexBatch unindexes a batch of fingerprints.
func (i *FingerprintMembershipIndex) UnindexBatch(b FingerprintMetricMapping) error {
batch := i.NewBatch()
for fp, _ := range b {
batch.Delete(codableFingerprint(fp))
}
return i.Commit(batch)
}
// Has returns true if the given fingerprint is present.
func (i *FingerprintMembershipIndex) Has(fp clientmodel.Fingerprint) (ok bool, err error) {
return i.KeyValueStore.Has(codableFingerprint(fp))
}
// NewFingerprintMembershipIndex returns a FingerprintMembershipIndex object
// ready to use.
func NewFingerprintMembershipIndex(db KeyValueStore) *FingerprintMembershipIndex {
return &FingerprintMembershipIndex{
KeyValueStore: db,
}
}
// TODO(julius): Currently unused, is it needed?
// SynchronizedIndexer provides naive locking for any MetricIndexer.
type SynchronizedIndexer struct {
mu sync.Mutex
i MetricIndexer
}
// IndexMetrics calls IndexMetrics of the wrapped MetricIndexer after acquiring
// a lock.
func (i *SynchronizedIndexer) IndexMetrics(b FingerprintMetricMapping) error {
i.mu.Lock()
defer i.mu.Unlock()
return i.i.IndexMetrics(b)
}
type flusher interface {
Flush() error
}
// Flush calls Flush of the wrapped MetricIndexer after acquiring a lock. If the
// wrapped MetricIndexer has no Flush method, this is a no-op.
func (i *SynchronizedIndexer) Flush() error {
if flusher, ok := i.i.(flusher); ok {
i.mu.Lock()
defer i.mu.Unlock()
return flusher.Flush()
}
return nil
}
// Close calls Close of the wrapped MetricIndexer after acquiring a lock. If the
// wrapped MetricIndexer has no Close method, this is a no-op.
func (i *SynchronizedIndexer) Close() error {
if closer, ok := i.i.(io.Closer); ok {
i.mu.Lock()
defer i.mu.Unlock()
return closer.Close()
}
return nil
}
// NewSynchronizedIndexer returns a SynchronizedIndexer wrapping the given
// MetricIndexer.
func NewSynchronizedIndexer(i MetricIndexer) *SynchronizedIndexer {
return &SynchronizedIndexer{
i: i,
}
}
// BufferedIndexer provides unsynchronized index buffering.
type BufferedIndexer struct {
i MetricIndexer
limit int
buf []FingerprintMetricMapping
}
// IndexMetrics writes the entries in the given FingerprintMetricMapping to the
// index.
func (i *BufferedIndexer) IndexMetrics(b FingerprintMetricMapping) error {
if len(i.buf) < i.limit {
i.buf = append(i.buf, b)
return nil
}
return i.Flush()
}
// Flush writes all pending entries to the index.
func (i *BufferedIndexer) Flush() error {
if len(i.buf) == 0 {
return nil
}
union := FingerprintMetricMapping{}
for _, b := range i.buf {
for fp, m := range b {
union[fp] = m
}
}
i.buf = make([]FingerprintMetricMapping, 0, i.limit)
return i.i.IndexMetrics(union)
}
// Close flushes and closes the underlying buffer.
func (i *BufferedIndexer) Close() error {
if err := i.Flush(); err != nil {
return err
}
if closer, ok := i.i.(io.Closer); ok {
return closer.Close()
}
return nil
}
// NewBufferedIndexer returns a BufferedIndexer ready to use.
func NewBufferedIndexer(i MetricIndexer, limit int) *BufferedIndexer {
return &BufferedIndexer{
i: i,
limit: limit,
buf: make([]FingerprintMetricMapping, 0, limit),
}
}
// TotalIndexer is a MetricIndexer that indexes all standard facets of a metric
// that a user or the Prometheus subsystem would want to query against:
//
// <Fingerprint> -> <existence marker>
// <Label Name> -> {<Label Value>, ...}
// <Label Name> <Label Value> -> {<Fingerprint>, ...}
// <Fingerprint> -> <Metric>
//
// This type supports concurrent queries, but only single writes, and it has no
// locking semantics to enforce this.
type TotalIndexer struct {
FingerprintToMetric *FingerprintMetricIndex
LabelNameToLabelValues *LabelNameLabelValuesIndex
LabelPairToFingerprints *LabelPairFingerprintIndex
FingerprintMembership *FingerprintMembershipIndex
}
func findUnindexed(i *FingerprintMembershipIndex, b FingerprintMetricMapping) (FingerprintMetricMapping, error) {
out := FingerprintMetricMapping{}
for fp, m := range b {
has, err := i.Has(fp)
if err != nil {
return nil, err
}
if !has {
out[fp] = m
}
}
return out, nil
}
func findIndexed(i *FingerprintMembershipIndex, b FingerprintMetricMapping) (FingerprintMetricMapping, error) {
out := FingerprintMetricMapping{}
for fp, m := range b {
has, err := i.Has(fp)
if err != nil {
return nil, err
}
if has {
out[fp] = m
}
}
return out, nil
}
func extendLabelNameToLabelValuesIndex(i *LabelNameLabelValuesIndex, b FingerprintMetricMapping) (LabelNameLabelValuesMapping, error) {
collection := map[clientmodel.LabelName]utility.Set{}
for _, m := range b {
for l, v := range m {
set, ok := collection[l]
if !ok {
baseValues, _, err := i.Lookup(l)
if err != nil {
return nil, err
}
set = utility.Set{}
for _, baseValue := range baseValues {
set.Add(baseValue)
}
collection[l] = set
}
set.Add(v)
}
}
batch := LabelNameLabelValuesMapping{}
for l, set := range collection {
values := make(clientmodel.LabelValues, 0, len(set))
for e := range set {
val := e.(clientmodel.LabelValue)
values = append(values, val)
}
batch[l] = values
}
return batch, nil
}
func reduceLabelNameToLabelValuesIndex(i *LabelNameLabelValuesIndex, m LabelPairFingerprintsMapping) (LabelNameLabelValuesMapping, error) {
collection := map[clientmodel.LabelName]utility.Set{}
for lp, fps := range m {
if len(fps) != 0 {
continue
}
set, ok := collection[lp.Name]
if !ok {
baseValues, _, err := i.Lookup(lp.Name)
if err != nil {
return nil, err
}
set = utility.Set{}
for _, baseValue := range baseValues {
set.Add(baseValue)
}
collection[lp.Name] = set
}
set.Remove(lp.Value)
}
batch := LabelNameLabelValuesMapping{}
for l, set := range collection {
values := make(clientmodel.LabelValues, 0, len(set))
for e := range set {
val := e.(clientmodel.LabelValue)
values = append(values, val)
}
batch[l] = values
}
return batch, nil
}
func extendLabelPairIndex(i *LabelPairFingerprintIndex, b FingerprintMetricMapping, remove bool) (LabelPairFingerprintsMapping, error) {
collection := map[metric.LabelPair]utility.Set{}
for fp, m := range b {
for n, v := range m {
pair := metric.LabelPair{
Name: n,
Value: v,
}
set, ok := collection[pair]
if !ok {
baseFps, _, err := i.Lookup(&pair)
if err != nil {
return nil, err
}
set = utility.Set{}
for _, baseFp := range baseFps {
set.Add(baseFp)
}
collection[pair] = set
}
if remove {
set.Remove(fp)
} else {
set.Add(fp)
}
}
}
batch := LabelPairFingerprintsMapping{}
for pair, set := range collection {
fps := batch[pair]
for element := range set {
fp := element.(clientmodel.Fingerprint)
fps = append(fps, fp)
}
batch[pair] = fps
}
return batch, nil
}
// IndexMetrics adds the facets of all unindexed metrics found in the given
// FingerprintMetricMapping to the corresponding indices.
func (i *TotalIndexer) IndexMetrics(b FingerprintMetricMapping) error {
unindexed, err := findUnindexed(i.FingerprintMembership, b)
if err != nil {
return err
}
labelNames, err := extendLabelNameToLabelValuesIndex(i.LabelNameToLabelValues, unindexed)
if err != nil {
return err
}
if err := i.LabelNameToLabelValues.IndexBatch(labelNames); err != nil {
return err
}
labelPairs, err := extendLabelPairIndex(i.LabelPairToFingerprints, unindexed, false)
if err != nil {
return err
}
if err := i.LabelPairToFingerprints.IndexBatch(labelPairs); err != nil {
return err
}
if err := i.FingerprintToMetric.IndexBatch(unindexed); err != nil {
return err
}
return i.FingerprintMembership.IndexBatch(unindexed)
}
// UnindexMetrics removes the facets of all indexed metrics found in the given
// FingerprintMetricMapping to the corresponding indices.
func (i *TotalIndexer) UnindexMetrics(b FingerprintMetricMapping) error {
indexed, err := findIndexed(i.FingerprintMembership, b)
if err != nil {
return err
}
labelPairs, err := extendLabelPairIndex(i.LabelPairToFingerprints, indexed, true)
if err != nil {
return err
}
if err := i.LabelPairToFingerprints.IndexBatch(labelPairs); err != nil {
return err
}
labelNames, err := reduceLabelNameToLabelValuesIndex(i.LabelNameToLabelValues, labelPairs)
if err := i.LabelNameToLabelValues.IndexBatch(labelNames); err != nil {
return err
}
if err := i.FingerprintToMetric.UnindexBatch(indexed); err != nil {
return err
}
return i.FingerprintMembership.UnindexBatch(indexed)
}
// GetMetricForFingerprint returns the metric associated with the provided fingerprint.
func (i *TotalIndexer) GetMetricForFingerprint(clientmodel.Fingerprint) (clientmodel.Metric, error) {
// TODO: implement.
return nil, nil
}
// GetFingerprintsForLabelPair returns all fingerprints for the provided label pair.
func (i *TotalIndexer) GetFingerprintsForLabelPair(l clientmodel.LabelName, v clientmodel.LabelValue) (clientmodel.Fingerprints, error) {
// TODO: implement.
return nil, nil
}
// GetLabelValuesForLabelName returns all label values associated with a given label name.
func (i *TotalIndexer) GetLabelValuesForLabelName(clientmodel.LabelName) (clientmodel.LabelValues, error) {
// TODO: implement.
return nil, nil
}
// HasFingerprint returns true if a metric with the given fingerprint has been indexed.
func (i *TotalIndexer) HasFingerprint(clientmodel.Fingerprint) (bool, error) {
// TODO: implement.
return false, nil
}

252
storage/local/index/index_test.go Normal file
View file

@ -0,0 +1,252 @@
package index
import (
"sort"
"testing"
clientmodel "github.com/prometheus/client_golang/model"
"github.com/prometheus/prometheus/storage/metric"
"github.com/prometheus/prometheus/utility/test"
)
type incrementalBatch struct {
fpToMetric FingerprintMetricMapping
expectedLnToLvs LabelNameLabelValuesMapping
expectedLpToFps LabelPairFingerprintsMapping
}
func newTestDB(t *testing.T) (KeyValueStore, test.Closer) {
dir := test.NewTemporaryDirectory("test_db", t)
db, err := NewLevelDB(LevelDBOptions{
Path: dir.Path(),
})
if err != nil {
dir.Close()
t.Fatal("failed to create test DB: ", err)
}
return db, test.NewCallbackCloser(func() {
db.Close()
dir.Close()
})
}
func verifyIndexedState(i int, t *testing.T, b incrementalBatch, indexedFpsToMetrics FingerprintMetricMapping, indexer *TotalIndexer) {
for fp, m := range indexedFpsToMetrics {
// Compare indexed metrics with input metrics.
mOut, ok, err := indexer.FingerprintToMetric.Lookup(fp)
if err != nil {
t.Fatal(err)
}
if !ok {
t.Fatalf("%d. fingerprint %v not found", i, fp)
}
if !mOut.Equal(m) {
t.Fatalf("%i. %v: Got: %s; want %s", i, fp, mOut, m)
}
// Check that indexed metrics are in membership index.
ok, err = indexer.FingerprintMembership.Has(fp)
if err != nil {
t.Fatal(err)
}
if !ok {
t.Fatalf("%d. fingerprint %v not found", i, fp)
}
}
// Compare label name -> label values mappings.
for ln, lvs := range b.expectedLnToLvs {
outLvs, ok, err := indexer.LabelNameToLabelValues.Lookup(ln)
if err != nil {
t.Fatal(err)
}
if !ok {
t.Fatalf("%d. label name %s not found", i, ln)
}
sort.Sort(lvs)
sort.Sort(outLvs)
if len(lvs) != len(outLvs) {
t.Fatalf("%d. different number of label values. Got: %d; want %d", i, len(outLvs), len(lvs))
}
for j, _ := range lvs {
if lvs[j] != outLvs[j] {
t.Fatalf("%d.%d. label values don't match. Got: %s; want %s", i, j, outLvs[j], lvs[j])
}
}
}
// Compare label pair -> fingerprints mappings.
for lp, fps := range b.expectedLpToFps {
outFps, ok, err := indexer.LabelPairToFingerprints.Lookup(&lp)
if err != nil {
t.Fatal(err)
}
if !ok {
t.Fatalf("%d. label pair %v not found", i, lp)
}
sort.Sort(fps)
sort.Sort(outFps)
if len(fps) != len(outFps) {
t.Fatalf("%d. %v: different number of fingerprints. Got: %d; want %d", i, lp, len(outFps), len(fps))
}
for j, _ := range fps {
if fps[j] != outFps[j] {
t.Fatalf("%d.%d. %v: fingerprints don't match. Got: %d; want %d", i, j, lp, outFps[j], fps[j])
}
}
}
}
func TestIndexing(t *testing.T) {
batches := []incrementalBatch{
{
fpToMetric: FingerprintMetricMapping{
0: {
clientmodel.MetricNameLabel: "metric_0",
"label_1": "value_1",
},
1: {
clientmodel.MetricNameLabel: "metric_0",
"label_2": "value_2",
"label_3": "value_3",
},
2: {
clientmodel.MetricNameLabel: "metric_1",
"label_1": "value_2",
},
},
expectedLnToLvs: LabelNameLabelValuesMapping{
clientmodel.MetricNameLabel: clientmodel.LabelValues{"metric_0", "metric_1"},
"label_1": clientmodel.LabelValues{"value_1", "value_2"},
"label_2": clientmodel.LabelValues{"value_2"},
"label_3": clientmodel.LabelValues{"value_3"},
},
expectedLpToFps: LabelPairFingerprintsMapping{
metric.LabelPair{
Name: clientmodel.MetricNameLabel,
Value: "metric_0",
}: {0, 1},
metric.LabelPair{
Name: clientmodel.MetricNameLabel,
Value: "metric_1",
}: {2},
metric.LabelPair{
Name: "label_1",
Value: "value_1",
}: {0},
metric.LabelPair{
Name: "label_1",
Value: "value_2",
}: {2},
metric.LabelPair{
Name: "label_2",
Value: "value_2",
}: {1},
metric.LabelPair{
Name: "label_3",
Value: "value_3",
}: {1},
},
}, {
fpToMetric: FingerprintMetricMapping{
3: {
clientmodel.MetricNameLabel: "metric_0",
"label_1": "value_3",
},
4: {
clientmodel.MetricNameLabel: "metric_2",
"label_2": "value_2",
"label_3": "value_1",
},
5: {
clientmodel.MetricNameLabel: "metric_1",
"label_1": "value_3",
},
},
expectedLnToLvs: LabelNameLabelValuesMapping{
clientmodel.MetricNameLabel: clientmodel.LabelValues{"metric_0", "metric_1", "metric_2"},
"label_1": clientmodel.LabelValues{"value_1", "value_2", "value_3"},
"label_2": clientmodel.LabelValues{"value_2"},
"label_3": clientmodel.LabelValues{"value_1", "value_3"},
},
expectedLpToFps: LabelPairFingerprintsMapping{
metric.LabelPair{
Name: clientmodel.MetricNameLabel,
Value: "metric_0",
}: {0, 1, 3},
metric.LabelPair{
Name: clientmodel.MetricNameLabel,
Value: "metric_1",
}: {2, 5},
metric.LabelPair{
Name: clientmodel.MetricNameLabel,
Value: "metric_2",
}: {4},
metric.LabelPair{
Name: "label_1",
Value: "value_1",
}: {0},
metric.LabelPair{
Name: "label_1",
Value: "value_2",
}: {2},
metric.LabelPair{
Name: "label_1",
Value: "value_3",
}: {3, 5},
metric.LabelPair{
Name: "label_2",
Value: "value_2",
}: {1, 4},
metric.LabelPair{
Name: "label_3",
Value: "value_1",
}: {4},
metric.LabelPair{
Name: "label_3",
Value: "value_3",
}: {1},
},
},
}
fpToMetricDB, fpToMetricCloser := newTestDB(t)
defer fpToMetricCloser.Close()
lnToLvsDB, lnToLvsCloser := newTestDB(t)
defer lnToLvsCloser.Close()
lpToFpDB, lpToFpCloser := newTestDB(t)
defer lpToFpCloser.Close()
fpMsDB, fpMsCloser := newTestDB(t)
defer fpMsCloser.Close()
indexer := TotalIndexer{
FingerprintToMetric: NewFingerprintMetricIndex(fpToMetricDB),
LabelNameToLabelValues: NewLabelNameLabelValuesIndex(lnToLvsDB),
LabelPairToFingerprints: NewLabelPairFingerprintIndex(lpToFpDB),
FingerprintMembership: NewFingerprintMembershipIndex(fpMsDB),
}
indexedFpsToMetrics := FingerprintMetricMapping{}
for i, b := range batches {
indexer.IndexMetrics(b.fpToMetric)
for fp, m := range b.fpToMetric {
indexedFpsToMetrics[fp] = m
}
verifyIndexedState(i, t, b, indexedFpsToMetrics, &indexer)
}
for i := len(batches) - 1; i >= 0; i-- {
b := batches[i]
verifyIndexedState(i, t, batches[i], indexedFpsToMetrics, &indexer)
indexer.UnindexMetrics(b.fpToMetric)
for fp, _ := range b.fpToMetric {
delete(indexedFpsToMetrics, fp)
}
}
}

43
storage/local/index/interface.go Normal file
View file

@ -0,0 +1,43 @@
package index
import (
clientmodel "github.com/prometheus/client_golang/model"
)
// MetricIndexer indexes facets of a clientmodel.Metric. The interface makes no
// assumptions about the concurrency safety of the underlying implementer.
type MetricIndexer interface {
// IndexMetrics adds metrics to the index.
IndexMetrics(FingerprintMetricMapping) error
// UnindexMetrics removes metrics from the index.
UnindexMetrics(FingerprintMetricMapping) error
// GetMetricForFingerprint returns the metric associated with the provided fingerprint.
GetMetricForFingerprint(clientmodel.Fingerprint) (clientmodel.Metric, error)
// GetFingerprintsForLabelPair returns all fingerprints for the provided label pair.
GetFingerprintsForLabelPair(l clientmodel.LabelName, v clientmodel.LabelValue) (clientmodel.Fingerprints, error)
// GetLabelValuesForLabelName returns all label values associated with a given label name.
GetLabelValuesForLabelName(clientmodel.LabelName) (clientmodel.LabelValues, error)
// HasFingerprint returns true if a metric with the given fingerprint has been indexed.
HasFingerprint(clientmodel.Fingerprint) (bool, error)
}
// KeyValueStore persists key/value pairs.
type KeyValueStore interface {
Put(key, value encodable) error
Get(k encodable, v decodable) (bool, error)
Has(k encodable) (has bool, err error)
Delete(k encodable) error
NewBatch() Batch
Commit(b Batch) error
Close() error
}
// Batch allows KeyValueStore mutations to be pooled and committed together.
type Batch interface {
Put(key, value encodable)
Delete(key encodable)
Reset()
}

80
storage/local/index/leveldb.go Normal file
View file

@ -0,0 +1,80 @@
package index
import (
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/cache"
"github.com/syndtr/goleveldb/leveldb/filter"
"github.com/syndtr/goleveldb/leveldb/opt"
)
// LevelDB is a LevelDB-backed sorted key-value store.
type LevelDB struct {
storage *leveldb.DB
readOpts *opt.ReadOptions
writeOpts *opt.WriteOptions
}
type LevelDBOptions struct {
Path string
CacheSizeBytes int
}
func NewLevelDB(o LevelDBOptions) (*LevelDB, error) {
options := &opt.Options{
Compression: opt.SnappyCompression,
BlockCache: cache.NewLRUCache(o.CacheSizeBytes),
Filter: filter.NewBloomFilter(10),
}
storage, err := leveldb.OpenFile(o.Path, options)
if err != nil {
return nil, err
}
return &LevelDB{
storage: storage,
readOpts: &opt.ReadOptions{},
writeOpts: &opt.WriteOptions{},
}, nil
}
func (l *LevelDB) NewBatch() Batch {
return &batch{
batch: &leveldb.Batch{},
}
}
func (l *LevelDB) Close() error {
return l.storage.Close()
}
func (l *LevelDB) Get(k encodable, v decodable) (bool, error) {
raw, err := l.storage.Get(k.encode(), l.readOpts)
if err == leveldb.ErrNotFound {
return false, nil
}
if err != nil {
return false, err
}
if v == nil {
return true, nil
}
v.decode(raw)
return true, err
}
func (l *LevelDB) Has(k encodable) (has bool, err error) {
return l.Get(k, nil)
}
func (l *LevelDB) Delete(k encodable) error {
return l.storage.Delete(k.encode(), l.writeOpts)
}
func (l *LevelDB) Put(key, value encodable) error {
return l.storage.Put(key.encode(), value.encode(), l.writeOpts)
}
func (l *LevelDB) Commit(b Batch) error {
return l.storage.Write(b.(*batch).batch, l.writeOpts)
}

21
storage/local/interface.go
View file

@ -3,8 +3,8 @@ package storage_ng
import (
clientmodel "github.com/prometheus/client_golang/model"
"github.com/prometheus/prometheus/storage/local/index"
"github.com/prometheus/prometheus/storage/metric"
"github.com/prometheus/prometheus/utility"
)
type Storage interface {
@ -47,10 +47,6 @@ type SeriesIterator interface {
type Persistence interface {
// PersistChunk persists a single chunk of a series.
PersistChunk(clientmodel.Fingerprint, chunk) error
// PersistIndexes persists a Prometheus server's timeseries indexes. It
// is the caller's responsibility to not modify indexes while persisting
// is underway, and to not call this method multiple times concurrently.
PersistIndexes(i *Indexes) error
// PersistHeads persists all open (non-full) head chunks.
PersistHeads(map[clientmodel.Fingerprint]*memorySeries) error
@ -66,10 +62,11 @@ type Persistence interface {
LoadChunkDescs(fp clientmodel.Fingerprint, beforeTime clientmodel.Timestamp) (chunkDescs, error)
// LoadHeads loads all open (non-full) head chunks.
LoadHeads(map[clientmodel.Fingerprint]*memorySeries) error
// LoadIndexes loads and returns all timeseries indexes. It is the
// caller's responsibility to not modify indexes while loading is
// underway, and to not call this method multiple times concurrently.
LoadIndexes() (*Indexes, error)
// Close releases any held resources.
Close()
index.MetricIndexer
}
// A Preloader preloads series data necessary for a query into memory and pins
@ -94,9 +91,3 @@ type Closer interface {
// Close cleans up any used resources.
Close()
}
type Indexes struct {
FingerprintToSeries map[clientmodel.Fingerprint]*memorySeries
LabelPairToFingerprints map[metric.LabelPair]utility.Set
LabelNameToLabelValues map[clientmodel.LabelName]utility.Set
}

428
storage/local/persistence.go
View file

@ -3,7 +3,7 @@ package storage_ng
import (
"bufio"
"encoding/binary"
"encoding/gob"
"flag"
"fmt"
"io"
"os"
@ -11,25 +11,23 @@ import (
"github.com/golang/glog"
//"github.com/prometheus/prometheus/storage/metric"
//"github.com/prometheus/client_golang/prometheus"
clientmodel "github.com/prometheus/client_golang/model"
"github.com/prometheus/prometheus/storage/metric"
"github.com/prometheus/prometheus/utility"
"github.com/prometheus/prometheus/storage/local/index"
)
const (
seriesFileName = "series.db"
seriesTempFileName = "series.db.tmp"
headsFileName = "heads.db"
indexFileName = "index.db"
indexDirName = "index"
indexFormatVersion = 1
indexMagicString = "PrometheusIndexes"
indexBufSize = 1 << 15 // 32kiB. TODO: Tweak.
fingerprintToMetricDir = "fingerprint_to_metric"
labelNameToLabelValuesDir = "labelname_to_labelvalues"
labelPairToFingerprintsDir = "labelpair_to_fingerprints"
fingerprintMembershipDir = "fingerprint_membership"
chunkHeaderLen = 17
chunkHeaderTypeOffset = 0
@ -41,24 +39,73 @@ const (
headsHeaderTypeOffset = 8
)
var (
fingerprintToMetricCacheSize = flag.Int("storage.fingerprintToMetricCacheSizeBytes", 25*1024*1024, "The size in bytes for the fingerprint to metric index cache.")
labelNameToLabelValuesCacheSize = flag.Int("storage.labelNameToLabelValuesCacheSizeBytes", 25*1024*1024, "The size in bytes for the label name to label values index.")
labelPairToFingerprintsCacheSize = flag.Int("storage.labelPairToFingerprintsCacheSizeBytes", 25*1024*1024, "The size in bytes for the label pair to fingerprints index.")
fingerprintMembershipCacheSize = flag.Int("storage.fingerprintMembershipCacheSizeBytes", 5*1024*1024, "The size in bytes for the metric membership index.")
)
type diskPersistence struct {
index.MetricIndexer
indexDBs []index.KeyValueStore
basePath string
chunkLen int
buf []byte // Staging space for persisting indexes.
}
func NewDiskPersistence(basePath string, chunkLen int) (Persistence, error) {
gob.Register(clientmodel.Fingerprint(0))
gob.Register(clientmodel.LabelValue(""))
err := os.MkdirAll(basePath, 0700)
if err != nil {
return nil, err
}
fingerprintToMetricDB, err := index.NewLevelDB(index.LevelDBOptions{
Path: path.Join(basePath, fingerprintToMetricDir),
CacheSizeBytes: *fingerprintToMetricCacheSize,
})
if err != nil {
return nil, err
}
labelNameToLabelValuesDB, err := index.NewLevelDB(index.LevelDBOptions{
Path: path.Join(basePath, labelNameToLabelValuesDir),
CacheSizeBytes: *labelNameToLabelValuesCacheSize,
})
if err != nil {
return nil, err
}
labelPairToFingerprintsDB, err := index.NewLevelDB(index.LevelDBOptions{
Path: path.Join(basePath, labelPairToFingerprintsDir),
CacheSizeBytes: *labelPairToFingerprintsCacheSize,
})
if err != nil {
return nil, err
}
fingerprintMembershipDB, err := index.NewLevelDB(index.LevelDBOptions{
Path: path.Join(basePath, fingerprintMembershipDir),
CacheSizeBytes: *fingerprintMembershipCacheSize,
})
if err != nil {
return nil, err
}
return &diskPersistence{
basePath: basePath,
chunkLen: chunkLen,
buf: make([]byte, binary.MaxVarintLen64), // Also sufficient for uint64.
MetricIndexer: &index.TotalIndexer{
FingerprintToMetric: index.NewFingerprintMetricIndex(fingerprintToMetricDB),
LabelNameToLabelValues: index.NewLabelNameLabelValuesIndex(labelNameToLabelValuesDB),
LabelPairToFingerprints: index.NewLabelPairFingerprintIndex(labelPairToFingerprintsDB),
FingerprintMembership: index.NewFingerprintMembershipIndex(fingerprintMembershipDB),
},
indexDBs: []index.KeyValueStore{
fingerprintToMetricDB,
labelNameToLabelValuesDB,
labelPairToFingerprintsDB,
fingerprintMembershipDB,
},
}, nil
}
@ -231,340 +278,6 @@ func (p *diskPersistence) LoadChunkDescs(fp clientmodel.Fingerprint, beforeTime
return cds, nil
}
func (p *diskPersistence) indexPath() string {
return path.Join(p.basePath, indexFileName)
}
// PersistIndexes persists the indexes to disk. Do not call it concurrently with
// LoadIndexes as they share a buffer for staging. This method depends on the
// following type conversions being possible:
// clientmodel.LabelName -> string
// clientmodel.LabelValue -> string
// clientmodel.Fingerprint -> uint64
//
// Description of the on-disk format:
//
// Label names and label values are encoded as their varint-encoded length
// followed by their byte sequence.
//
// Fingerprints are encoded as big-endian uint64.
//
// The file starts with the 'magic' byte sequence "PrometheusIndexes", followed
// by a varint-encoded version number (currently 1).
//
// The indexes follow one after another in the order FingerprintToSeries,
// LabelPairToFingerprints, LabelNameToLabelValues. Each index starts with the
// varint-encoded number of entries in that index, followed by the corresponding
// number of entries.
//
// An entry in FingerprintToSeries consists of a fingerprint, followed by the
// number of label pairs, followed by those label pairs, each in order label
// name and then label value.
//
// An entry in LabelPairToFingerprints consists of a label name, then a label
// value, then a varint-encoded number of fingerprints, followed by those
// fingerprints.
//
// An entry in LabelNameToLabelValues consists of a label name, followed by the
// varint-encoded number of label values, followed by those label values.
func (p *diskPersistence) PersistIndexes(i *Indexes) error {
f, err := os.OpenFile(p.indexPath(), os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0640)
if err != nil {
return err
}
defer f.Close()
p.setBufLen(binary.MaxVarintLen64)
w := bufio.NewWriterSize(f, indexBufSize)
if _, err := w.WriteString(indexMagicString); err != nil {
return err
}
if err := p.persistVarint(w, indexFormatVersion); err != nil {
return err
}
if err := p.persistFingerprintToSeries(w, i.FingerprintToSeries); err != nil {
return err
}
if err := p.persistLabelPairToFingerprints(w, i.LabelPairToFingerprints); err != nil {
return err
}
if err := p.persistLabelNameToLabelValues(w, i.LabelNameToLabelValues); err != nil {
return err
}
return w.Flush()
}
func (p *diskPersistence) persistVarint(w io.Writer, i int) error {
bytesWritten := binary.PutVarint(p.buf, int64(i))
_, err := w.Write(p.buf[:bytesWritten])
return err
}
// persistFingerprint depends on clientmodel.Fingerprint to be convertible to
// uint64.
func (p *diskPersistence) persistFingerprint(w io.Writer, fp clientmodel.Fingerprint) error {
binary.BigEndian.PutUint64(p.buf, uint64(fp))
_, err := w.Write(p.buf[:8])
return err
}
func (p *diskPersistence) persistString(w *bufio.Writer, s string) error {
if err := p.persistVarint(w, len(s)); err != nil {
return err
}
_, err := w.WriteString(s)
return err
}
// persistFingerprintToSeries depends on clientmodel.LabelName and
// clientmodel.LabelValue to be convertible to string.
func (p *diskPersistence) persistFingerprintToSeries(w *bufio.Writer, index map[clientmodel.Fingerprint]*memorySeries) error {
if err := p.persistVarint(w, len(index)); err != nil {
return err
}
for fp, ms := range index {
if err := p.persistFingerprint(w, fp); err != nil {
return err
}
if err := p.persistVarint(w, len(ms.metric)); err != nil {
return err
}
for n, v := range ms.metric {
if err := p.persistString(w, string(n)); err != nil {
return err
}
if err := p.persistString(w, string(v)); err != nil {
return err
}
}
}
return nil
}
// persistLabelPairToFingerprints depends on clientmodel.LabelName and
// clientmodel.LabelValue to be convertible to string.
func (p *diskPersistence) persistLabelPairToFingerprints(w *bufio.Writer, index map[metric.LabelPair]utility.Set) error {
if err := p.persistVarint(w, len(index)); err != nil {
return err
}
for lp, fps := range index {
if err := p.persistString(w, string(lp.Name)); err != nil {
return err
}
if err := p.persistString(w, string(lp.Value)); err != nil {
return err
}
if err := p.persistVarint(w, len(fps)); err != nil {
return err
}
for fp := range fps {
if err := p.persistFingerprint(w, fp.(clientmodel.Fingerprint)); err != nil {
return err
}
}
}
return nil
}
// persistLabelNameToLabelValues depends on clientmodel.LabelValue to be convertible to string.
func (p *diskPersistence) persistLabelNameToLabelValues(w *bufio.Writer, index map[clientmodel.LabelName]utility.Set) error {
if err := p.persistVarint(w, len(index)); err != nil {
return err
}
for ln, lvs := range index {
if err := p.persistString(w, string(ln)); err != nil {
return err
}
if err := p.persistVarint(w, len(lvs)); err != nil {
return err
}
for lv := range lvs {
if err := p.persistString(w, string(lv.(clientmodel.LabelValue))); err != nil {
return err
}
}
}
return nil
}
// LoadIndexes loads the indexes from disk. See PersistIndexes for details about
// the disk format. Do not call LoadIndexes and PersistIndexes concurrently as
// they share a buffer for staging.
func (p *diskPersistence) LoadIndexes() (*Indexes, error) {
f, err := os.Open(p.indexPath())
if os.IsNotExist(err) {
return &Indexes{
FingerprintToSeries: map[clientmodel.Fingerprint]*memorySeries{},
LabelPairToFingerprints: map[metric.LabelPair]utility.Set{},
LabelNameToLabelValues: map[clientmodel.LabelName]utility.Set{},
}, nil
}
if err != nil {
return nil, err
}
defer f.Close()
r := bufio.NewReaderSize(f, indexBufSize)
p.setBufLen(len(indexMagicString))
if _, err := io.ReadFull(r, p.buf); err != nil {
return nil, err
}
magic := string(p.buf)
if magic != indexMagicString {
return nil, fmt.Errorf(
"unexpected magic string, want %q, got %q",
indexMagicString, magic,
)
}
if version, err := binary.ReadVarint(r); version != indexFormatVersion || err != nil {
return nil, fmt.Errorf("unknown index format version, want %d", indexFormatVersion)
}
i := &Indexes{}
if err := p.loadFingerprintToSeries(r, i); err != nil {
return nil, err
}
if err := p.loadLabelPairToFingerprints(r, i); err != nil {
return nil, err
}
if err := p.loadLabelNameToLabelValues(r, i); err != nil {
return nil, err
}
return i, nil
}
func (p *diskPersistence) loadFingerprintToSeries(r *bufio.Reader, i *Indexes) error {
length, err := binary.ReadVarint(r)
if err != nil {
return err
}
i.FingerprintToSeries = make(map[clientmodel.Fingerprint]*memorySeries, length)
for ; length > 0; length-- {
fp, err := p.loadFingerprint(r)
if err != nil {
return err
}
numLabelPairs, err := binary.ReadVarint(r)
if err != nil {
return err
}
m := make(clientmodel.Metric, numLabelPairs)
i.FingerprintToSeries[fp] = &memorySeries{metric: m}
for ; numLabelPairs > 0; numLabelPairs-- {
ln, err := p.loadString(r)
if err != nil {
return err
}
lv, err := p.loadString(r)
if err != nil {
return err
}
m[clientmodel.LabelName(ln)] = clientmodel.LabelValue(lv)
}
}
return nil
}
func (p *diskPersistence) loadLabelPairToFingerprints(r *bufio.Reader, i *Indexes) error {
length, err := binary.ReadVarint(r)
if err != nil {
return err
}
i.LabelPairToFingerprints = make(map[metric.LabelPair]utility.Set, length)
for ; length > 0; length-- {
ln, err := p.loadString(r)
if err != nil {
return err
}
lv, err := p.loadString(r)
if err != nil {
return err
}
numFPs, err := binary.ReadVarint(r)
if err != nil {
return err
}
s := make(utility.Set, numFPs)
i.LabelPairToFingerprints[metric.LabelPair{
Name: clientmodel.LabelName(ln),
Value: clientmodel.LabelValue(lv),
}] = s
for ; numFPs > 0; numFPs-- {
fp, err := p.loadFingerprint(r)
if err != nil {
return err
}
s.Add(fp)
}
}
return nil
}
func (p *diskPersistence) loadLabelNameToLabelValues(r *bufio.Reader, i *Indexes) error {
length, err := binary.ReadVarint(r)
if err != nil {
return err
}
i.LabelNameToLabelValues = make(map[clientmodel.LabelName]utility.Set, length)
for ; length > 0; length-- {
ln, err := p.loadString(r)
if err != nil {
return err
}
numLVs, err := binary.ReadVarint(r)
if err != nil {
return err
}
s := make(utility.Set, numLVs)
i.LabelNameToLabelValues[clientmodel.LabelName(ln)] = s
for ; numLVs > 0; numLVs-- {
lv, err := p.loadString(r)
if err != nil {
return err
}
s.Add(clientmodel.LabelValue(lv))
}
}
return nil
}
func (p *diskPersistence) loadFingerprint(r io.Reader) (clientmodel.Fingerprint, error) {
p.setBufLen(8)
if _, err := io.ReadFull(r, p.buf); err != nil {
return 0, err
}
return clientmodel.Fingerprint(binary.BigEndian.Uint64(p.buf)), nil
}
func (p *diskPersistence) loadString(r *bufio.Reader) (string, error) {
length, err := binary.ReadVarint(r)
if err != nil {
return "", err
}
p.setBufLen(int(length))
if _, err := io.ReadFull(r, p.buf); err != nil {
return "", err
}
return string(p.buf), nil
}
func (p *diskPersistence) setBufLen(l int) {
if cap(p.buf) >= l {
p.buf = p.buf[:l]
} else {
p.buf = make([]byte, l)
}
}
func (p *diskPersistence) headsPath() string {
return path.Join(p.basePath, headsFileName)
}
@ -681,3 +394,26 @@ func (p *diskPersistence) LoadHeads(fpToSeries map[clientmodel.Fingerprint]*memo
}
return nil
}
func (d *diskPersistence) Close() {
for _, db := range d.indexDBs {
if err := db.Close(); err != nil {
glog.Error("Error closing index DB: ", err)
}
}
}
// Get all of the label values that are associated with a given label name.
func (d *diskPersistence) GetFingerprintsForLabelPair(l clientmodel.LabelName, v clientmodel.LabelValue) (clientmodel.Fingerprints, error) {
return nil, nil
}
// Get all label values that are associated with a given label name.
func (d *diskPersistence) GetLabelValuesForLabelName(clientmodel.LabelName) (clientmodel.LabelValues, error) {
return nil, nil
}
// Get the metric associated with the provided fingerprint.
func (d *diskPersistence) GetMetricForFingerprint(clientmodel.Fingerprint) (clientmodel.Metric, error) {
return nil, nil
}

177
storage/local/persistence_test.go
View file

@ -1,98 +1,14 @@
package storage_ng
import (
"io/ioutil"
"os"
"testing"
clientmodel "github.com/prometheus/client_golang/model"
"github.com/prometheus/prometheus/storage/metric"
"github.com/prometheus/prometheus/utility"
"github.com/prometheus/prometheus/utility/test"
)
var (
indexes = Indexes{
FingerprintToSeries: map[clientmodel.Fingerprint]*memorySeries{
0: {
metric: clientmodel.Metric{
clientmodel.MetricNameLabel: "metric_0",
"label_1": "value_1",
},
},
1: {
metric: clientmodel.Metric{
clientmodel.MetricNameLabel: "metric_0",
"label_2": "value_2",
"label_3": "value_3",
},
},
2: {
metric: clientmodel.Metric{
clientmodel.MetricNameLabel: "metric_1",
"label_1": "value_2",
},
},
},
LabelPairToFingerprints: map[metric.LabelPair]utility.Set{
metric.LabelPair{
Name: clientmodel.MetricNameLabel,
Value: "metric_0",
}: {
clientmodel.Fingerprint(0): struct{}{},
clientmodel.Fingerprint(1): struct{}{},
},
metric.LabelPair{
Name: clientmodel.MetricNameLabel,
Value: "metric_1",
}: {
clientmodel.Fingerprint(2): struct{}{},
},
metric.LabelPair{
Name: "label_1",
Value: "value_1",
}: {
clientmodel.Fingerprint(0): struct{}{},
},
metric.LabelPair{
Name: "label_1",
Value: "value_2",
}: {
clientmodel.Fingerprint(2): struct{}{},
},
metric.LabelPair{
Name: "label_2",
Value: "value_2",
}: {
clientmodel.Fingerprint(1): struct{}{},
},
metric.LabelPair{
Name: "label_3",
Value: "value_2",
}: {
clientmodel.Fingerprint(1): struct{}{},
},
},
LabelNameToLabelValues: map[clientmodel.LabelName]utility.Set{
clientmodel.MetricNameLabel: {
clientmodel.LabelValue("metric_0"): struct{}{},
clientmodel.LabelValue("metric_1"): struct{}{},
},
"label_1": {
clientmodel.LabelValue("value_1"): struct{}{},
clientmodel.LabelValue("value_2"): struct{}{},
},
"label_2": {
clientmodel.LabelValue("value_2"): struct{}{},
},
"label_3": {
clientmodel.LabelValue("value_3"): struct{}{},
},
},
}
)
func newTestPersistence(t *testing.T) (Persistence, test.Closer) {
dir := test.NewTemporaryDirectory("test_persistence", t)
p, err := NewDiskPersistence(dir.Path(), 1024)
@ -100,95 +16,10 @@ func newTestPersistence(t *testing.T) (Persistence, test.Closer) {
dir.Close()
t.Fatal(err)
}
return p, dir
}
func TestIndexPersistence(t *testing.T) {
p, closer := newTestPersistence(t)
defer closer.Close()
if err := p.PersistIndexes(&indexes); err != nil {
t.Fatal(err)
}
actual, err := p.LoadIndexes()
if err != nil {
t.Fatal(err)
}
if len(actual.FingerprintToSeries) != len(indexes.FingerprintToSeries) {
t.Fatalf("Count mismatch: Got %d; want %d", len(actual.FingerprintToSeries), len(indexes.FingerprintToSeries))
}
for fp, actualSeries := range actual.FingerprintToSeries {
expectedSeries := indexes.FingerprintToSeries[fp]
if !expectedSeries.metric.Equal(actualSeries.metric) {
t.Fatalf("%v: Got %s; want %s", fp, actualSeries.metric, expectedSeries.metric)
}
}
if len(actual.LabelPairToFingerprints) != len(indexes.LabelPairToFingerprints) {
t.Fatalf("Count mismatch: Got %d; want %d", len(actual.LabelPairToFingerprints), len(indexes.LabelPairToFingerprints))
}
for lp, actualFps := range actual.LabelPairToFingerprints {
expectedFps := indexes.LabelPairToFingerprints[lp]
if len(actualFps) != len(actualFps.Intersection(expectedFps)) {
t.Fatalf("%s: Got %s; want %s", lp, actualFps, expectedFps)
}
}
if len(actual.LabelNameToLabelValues) != len(indexes.LabelNameToLabelValues) {
t.Fatalf("Count mismatch: Got %d; want %d", len(actual.LabelNameToLabelValues), len(indexes.LabelNameToLabelValues))
}
for name, actualVals := range actual.LabelNameToLabelValues {
expectedVals := indexes.LabelNameToLabelValues[name]
if len(actualVals) != len(actualVals.Intersection(expectedVals)) {
t.Fatalf("%s: Got %s; want %s", name, actualVals, expectedVals)
}
}
}
func BenchmarkPersistIndexes(b *testing.B) {
basePath, err := ioutil.TempDir("", "test_index_persistence")
if err != nil {
b.Fatal(err)
}
defer os.RemoveAll(basePath)
p, err := NewDiskPersistence(basePath, 1024)
if err != nil {
b.Fatal(err)
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
if err := p.PersistIndexes(&indexes); err != nil {
b.Fatal(err)
}
}
b.StopTimer()
}
func BenchmarkLoadIndexes(b *testing.B) {
basePath, err := ioutil.TempDir("", "test_index_persistence")
if err != nil {
b.Fatal(err)
}
defer os.RemoveAll(basePath)
p, err := NewDiskPersistence(basePath, 1024)
if err != nil {
b.Fatal(err)
}
if err := p.PersistIndexes(&indexes); err != nil {
b.Fatal(err)
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
_, err := p.LoadIndexes()
if err != nil {
b.Fatal(err)
}
}
b.StopTimer()
return p, test.NewCallbackCloser(func() {
p.Close()
dir.Close()
})
}
func buildTestChunks() map[clientmodel.Fingerprint]chunks {

28
storage/local/storage.go
View file

@ -52,20 +52,17 @@ type MemorySeriesStorageOptions struct {
}
func NewMemorySeriesStorage(o *MemorySeriesStorageOptions) (*memorySeriesStorage, error) { // TODO: change to return Storage?
glog.Info("Loading indexes...")
i, err := o.Persistence.LoadIndexes()
if err != nil {
return nil, err
}
glog.Info("Loading series head chunks...")
/*
if err := o.Persistence.LoadHeads(i.FingerprintToSeries); err != nil {
return nil, err
}
numSeries.Set(float64(len(i.FingerprintToSeries)))
*/
return &memorySeriesStorage{
fingerprintToSeries: i.FingerprintToSeries,
labelPairToFingerprints: i.LabelPairToFingerprints,
labelNameToLabelValues: i.LabelNameToLabelValues,
fingerprintToSeries: map[clientmodel.Fingerprint]*memorySeries{},
labelPairToFingerprints: map[metric.LabelPair]utility.Set{},
labelNameToLabelValues: map[clientmodel.LabelName]utility.Set{},
persistDone: make(chan bool),
stopServing: make(chan chan<- bool),
@ -241,12 +238,6 @@ func (s *memorySeriesStorage) Close() error {
}
glog.Info("Done persisting head chunks.")
glog.Info("Persisting indexes...")
if err := s.persistIndexes(); err != nil {
return err
}
glog.Info("Done persisting indexes.")
for _, series := range s.fingerprintToSeries {
series.close()
}
@ -261,15 +252,6 @@ func (s *memorySeriesStorage) persistHeads() error {
return s.persistence.PersistHeads(s.fingerprintToSeries)
}
func (s *memorySeriesStorage) persistIndexes() error {
err := s.persistence.PersistIndexes(&Indexes{
FingerprintToSeries: s.fingerprintToSeries,
LabelPairToFingerprints: s.labelPairToFingerprints,
LabelNameToLabelValues: s.labelNameToLabelValues,
})
return err
}
func (s *memorySeriesStorage) purgePeriodically(stop <-chan bool) {
purgeTicker := time.NewTicker(s.persistencePurgeInterval)
defer purgeTicker.Stop()

14
utility/test/directory.go
View file

@ -53,11 +53,25 @@ type (
path string
tester testing.TB
}
callbackCloser struct {
fn func()
}
)
func (c nilCloser) Close() {
}
func (c callbackCloser) Close() {
c.fn()
}
func NewCallbackCloser(fn func()) *callbackCloser {
return &callbackCloser{
fn: fn,
}
}
func (t temporaryDirectory) Close() {
err := os.RemoveAll(t.path)
if err != nil {