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

checkpoint.

Browse source
This commit is contained in:
Matt T. Proud 2013-03-06 17:16:39 -08:00
parent d5380897c3
commit 8cc5cdde0b
11 changed files with 250 additions and 144 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Makefile.TRAVIS
View file

@ -41,7 +41,7 @@ preparation-stamp: build-dependencies
build-dependencies: build-dependencies-stamp
build-dependencies-stamp: bison cc mercurial protoc goprotobuf gorest go instrumentation leveldb levigo skiplist vim-common
build-dependencies-stamp: bison cc mercurial protoc goprotobuf gorest goskiplist go instrumentation leveldb levigo
touch $@
overlay: overlay-stamp

58
main.go
View file

@ -18,7 +18,7 @@ import (
"fmt"
"github.com/prometheus/prometheus/appstate"
"github.com/prometheus/prometheus/config"
"github.com/prometheus/prometheus/model"
// "github.com/prometheus/prometheus/model"
"github.com/prometheus/prometheus/retrieval"
"github.com/prometheus/prometheus/retrieval/format"
"github.com/prometheus/prometheus/rules"
@ -50,25 +50,32 @@ func main() {
log.Fatalf("Error loading configuration from %s: %v", *configFile, err)
}
var persistence metric.MetricPersistence
var (
persistence metric.MetricPersistence
ts metric.Storage
)
if *memoryArena {
persistence = metric.NewMemorySeriesStorage()
} else {
persistence, err = metric.NewLevelDBMetricPersistence(*metricsStoragePath)
if err != nil {
log.Fatalf("Error opening storage: %v", err)
}
ts = metric.NewTieredStorage(5000, 5000, 100, time.Second*30, time.Second*1, time.Second*20, *metricsStoragePath)
go ts.Serve()
// persistence, err = metric.NewLevelDBMetricPersistence(*metricsStoragePath)
// if err != nil {
// log.Fatalf("Error opening storage: %v", err)
// }
}
go func() {
notifier := make(chan os.Signal)
signal.Notify(notifier, os.Interrupt)
<-notifier
persistence.Close()
// persistence.Close()
os.Exit(0)
}()
defer persistence.Close()
// defer persistence.Close()
// Queue depth will need to be exposed
scrapeResults := make(chan format.Result, *scrapeResultsQueueCapacity)
@ -94,26 +101,23 @@ func main() {
web.StartServing(appState)
ts := metric.NewTieredStorage(5000, 5000, 100, time.Second*30, time.Second*1, time.Second*20)
go ts.Serve()
// go func() {
// ticker := time.Tick(time.Second)
// for i := 0; i < 120; i++ {
// <-ticker
// if i%10 == 0 {
// fmt.Printf(".")
// }
// }
// fmt.Println()
// //f := model.NewFingerprintFromRowKey("9776005627788788740-g-131-0")
// f := model.NewFingerprintFromRowKey("09923616460706181007-g-131-0")
// v := metric.NewViewRequestBuilder()
// v.GetMetricAtTime(f, time.Now().Add(-120*time.Second))
go func() {
ticker := time.Tick(time.Second)
for i := 0; i < 120; i++ {
<-ticker
if i%10 == 0 {
fmt.Printf(".")
}
}
fmt.Println()
//f := model.NewFingerprintFromRowKey("9776005627788788740-g-131-0")
f := model.NewFingerprintFromRowKey("09923616460706181007-g-131-0")
v := metric.NewViewRequestBuilder()
v.GetMetricAtTime(f, time.Now().Add(-120*time.Second))
view, err := ts.MakeView(v, time.Minute)
fmt.Println(view, err)
}()
// view, err := ts.MakeView(v, time.Minute)
// fmt.Println(view, err)
// }()
for {
select {

36
model/fingerprinting.go
View file

@ -90,10 +90,10 @@ func NewFingerprintFromMetric(metric Metric) (f Fingerprint) {
labelValueLength := len(labelValue)
labelMatterLength += labelNameLength + labelValueLength
switch i {
case 0:
if i == 0 {
firstCharacterOfFirstLabelName = labelName[0:1]
case labelLength - 1:
}
if i == labelLength-1 {
lastCharacterOfLastLabelValue = string(labelValue[labelValueLength-2 : labelValueLength-1])
}
@ -146,25 +146,21 @@ func (f fingerprint) LastCharacterOfLastLabelValue() string {
return f.lastCharacterOfLastLabelValue
}
func (f fingerprint) Less(o Fingerprint) (before bool) {
before = f.Hash() <= o.Hash()
if !before {
return
func (f fingerprint) Less(o Fingerprint) bool {
if f.Hash() < o.Hash() {
return true
}
if f.FirstCharacterOfFirstLabelName() < o.FirstCharacterOfFirstLabelName() {
return true
}
if f.LabelMatterLength() < o.LabelMatterLength() {
return true
}
if f.LastCharacterOfLastLabelValue() < o.LastCharacterOfLastLabelValue() {
return true
}
before = sort.StringsAreSorted([]string{f.FirstCharacterOfFirstLabelName(), o.FirstCharacterOfFirstLabelName()})
if !before {
return
}
before = f.LabelMatterLength() <= o.LabelMatterLength()
if !before {
return
}
before = sort.StringsAreSorted([]string{f.LastCharacterOfLastLabelValue(), o.LastCharacterOfLastLabelValue()})
return
return false
}
func (f fingerprint) Equal(o Fingerprint) (equal bool) {

9
model/labelname.go
View file

@ -13,10 +13,6 @@
package model
import (
"sort"
)
// A LabelName is a key for a LabelSet or Metric. It has a value associated
// therewith.
type LabelName string
@ -28,10 +24,7 @@ func (l LabelNames) Len() int {
}
func (l LabelNames) Less(i, j int) bool {
return sort.StringsAreSorted([]string{
string(l[i]),
string(l[j]),
})
return l[i] < l[j]
}
func (l LabelNames) Swap(i, j int) {

21
model/labelpair.go
View file

@ -13,10 +13,6 @@
package model
import (
"sort"
)
type LabelPair struct {
Name LabelName
Value LabelValue
@ -29,20 +25,15 @@ func (l LabelPairs) Len() int {
}
func (l LabelPairs) Less(i, j int) (less bool) {
less = sort.StringsAreSorted([]string{
string(l[i].Name),
string(l[j].Name),
})
if !less {
return
if l[i].Name < l[j].Name {
return true
}
less = sort.StringsAreSorted([]string{
string(l[i].Value),
string(l[j].Value),
})
if l[i].Value < l[j].Value {
return true
}
return
return false
}
func (l LabelPairs) Swap(i, j int) {

15
model/sample.go
View file

@ -14,7 +14,6 @@
package model
import (
"sort"
"time"
)
@ -31,19 +30,15 @@ func (s Samples) Len() int {
}
func (s Samples) Less(i, j int) (less bool) {
fingerprints := Fingerprints{
NewFingerprintFromMetric(s[i].Metric),
NewFingerprintFromMetric(s[j].Metric),
if NewFingerprintFromMetric(s[i].Metric).Less(NewFingerprintFromMetric(s[j].Metric)) {
return true
}
less = sort.IsSorted(fingerprints)
if !less {
return
if s[i].Timestamp.Before(s[j].Timestamp) {
return true
}
less = s[i].Timestamp.Before(s[j].Timestamp)
return
return false
}
func (s Samples) Swap(i, j int) {

2
storage/metric/instrumentation.go
View file

@ -60,6 +60,7 @@ var (
storageOperations = metrics.NewCounter()
storageOperationDurations = metrics.NewCounter()
storageLatency = metrics.NewHistogram(diskLatencyHistogram)
queueSizes = metrics.NewGauge()
)
func recordOutcome(duration time.Duration, err error, success, failure map[string]string) {
@ -78,4 +79,5 @@ func init() {
registry.Register("prometheus_metric_disk_operations_total", "Total number of metric-related disk operations.", registry.NilLabels, storageOperations)
registry.Register("prometheus_metric_disk_latency_microseconds", "Latency for metric disk operations in microseconds.", registry.NilLabels, storageLatency)
registry.Register("prometheus_storage_operation_time_total_microseconds", "The total time spent performing a given storage operation.", registry.NilLabels, storageOperationDurations)
registry.Register("prometheus_storage_queue_sizes_total", "The various sizes and capacities of the storage queues.", registry.NilLabels, queueSizes)
}

6
storage/metric/interface.go
View file

@ -69,9 +69,9 @@ type StalenessPolicy struct {
// View provides view of the values in the datastore subject to the request of a
// preloading operation.
type View interface {
GetValueAtTime(model.Metric, time.Time, StalenessPolicy) (*model.Sample, error)
GetBoundaryValues(model.Metric, model.Interval, StalenessPolicy) (*model.Sample, *model.Sample, error)
GetRangeValues(model.Metric, model.Interval) (*model.SampleSet, error)
GetValueAtTime(model.Fingerprint, time.Time) []model.SamplePair
GetBoundaryValues(model.Fingerprint, model.Interval) []model.SamplePair
GetRangeValues(model.Fingerprint, model.Interval) []model.SamplePair
// Destroy this view.
Close()

27
storage/metric/memory.go
View file

@ -54,19 +54,20 @@ type stream struct {
values *skiplist.SkipList
}
func (s stream) add(sample model.Sample) {
s.values.Set(skipListTime(sample.Timestamp), singletonValue(sample.Value))
func (s stream) add(timestamp time.Time, value model.SampleValue) {
s.values.Set(skipListTime(timestamp), singletonValue(value))
}
func (s stream) forEach(decoder storage.RecordDecoder, filter storage.RecordFilter, operator storage.RecordOperator) (scannedEntireCorpus bool, err error) {
iterator := s.values.SeekToLast()
if iterator == nil {
panic("nil iterator")
if s.values.Len() == 0 {
return
}
iterator := s.values.SeekToLast()
defer iterator.Close()
for iterator.Previous() {
for !(iterator.Key() == nil || iterator.Value() == nil) {
decodedKey, decodeErr := decoder.DecodeKey(iterator.Key())
if decodeErr != nil {
continue
@ -90,6 +91,10 @@ func (s stream) forEach(decoder storage.RecordDecoder, filter storage.RecordFilt
break
}
}
if !iterator.Previous() {
break
}
}
scannedEntireCorpus = true
return
@ -117,9 +122,11 @@ func (s memorySeriesStorage) AppendSamples(samples model.Samples) (err error) {
}
func (s memorySeriesStorage) AppendSample(sample model.Sample) (err error) {
metric := sample.Metric
fingerprint := model.NewFingerprintFromMetric(metric)
series, ok := s.fingerprintToSeries[fingerprint]
var (
metric = sample.Metric
fingerprint = model.NewFingerprintFromMetric(metric)
series, ok = s.fingerprintToSeries[fingerprint]
)
if !ok {
series = newStream(metric)
@ -138,7 +145,7 @@ func (s memorySeriesStorage) AppendSample(sample model.Sample) (err error) {
}
}
series.add(sample)
series.add(sample.Timestamp, sample.Value)
return
}

129
storage/metric/tiered.go
View file

@ -20,6 +20,7 @@ import (
"github.com/prometheus/prometheus/model"
dto "github.com/prometheus/prometheus/model/generated"
"github.com/prometheus/prometheus/storage"
"sort"
"sync"
"time"
)
@ -58,10 +59,11 @@ type Storage interface {
Serve()
// Stops the storage subsystem, flushing all pending operations.
Drain()
Flush()
}
func NewTieredStorage(appendToMemoryQueueDepth, appendToDiskQueueDepth, viewQueueDepth uint, flushMemoryInterval, writeMemoryInterval, memoryTTL time.Duration) Storage {
diskStorage, err := NewLevelDBMetricPersistence("/tmp/metrics-foof")
func NewTieredStorage(appendToMemoryQueueDepth, appendToDiskQueueDepth, viewQueueDepth uint, flushMemoryInterval, writeMemoryInterval, memoryTTL time.Duration, root string) Storage {
diskStorage, err := NewLevelDBMetricPersistence(root)
if err != nil {
panic(err)
}
@ -90,7 +92,9 @@ func (t *tieredStorage) AppendSample(s model.Sample) (err error) {
}
func (t *tieredStorage) Drain() {
t.draining <- true
if len(t.draining) == 0 {
t.draining <- true
}
}
func (t *tieredStorage) MakeView(builder ViewRequestBuilder, deadline time.Duration) (view View, err error) {
@ -120,13 +124,13 @@ func (t *tieredStorage) MakeView(builder ViewRequestBuilder, deadline time.Durat
}
func (t *tieredStorage) rebuildDiskFrontier() (err error) {
fmt.Println("a1")
begin := time.Now()
defer func() {
duration := time.Now().Sub(begin)
recordOutcome(duration, err, map[string]string{operation: appendSample, result: success}, map[string]string{operation: rebuildDiskFrontier, result: failure})
}()
i, closer, err := t.diskStorage.metricSamples.GetIterator()
if closer != nil {
defer closer.Close()
@ -134,12 +138,10 @@ func (t *tieredStorage) rebuildDiskFrontier() (err error) {
if err != nil {
panic(err)
}
t.diskFrontier, err = newDiskFrontier(i)
if err != nil {
panic(err)
}
return
}
@ -150,6 +152,8 @@ func (t *tieredStorage) Serve() {
)
for {
t.reportQueues()
select {
case <-writeMemoryTicker:
t.writeMemory()
@ -159,11 +163,22 @@ func (t *tieredStorage) Serve() {
t.renderView(viewRequest)
case <-t.draining:
t.flush()
return
break
}
}
}
func (t *tieredStorage) reportQueues() {
queueSizes.Set(map[string]string{"queue": "append_to_disk", "facet": "occupancy"}, float64(len(t.appendToDiskQueue)))
queueSizes.Set(map[string]string{"queue": "append_to_disk", "facet": "capacity"}, float64(cap(t.appendToDiskQueue)))
queueSizes.Set(map[string]string{"queue": "append_to_memory", "facet": "occupancy"}, float64(len(t.appendToMemoryQueue)))
queueSizes.Set(map[string]string{"queue": "append_to_memory", "facet": "capacity"}, float64(cap(t.appendToMemoryQueue)))
queueSizes.Set(map[string]string{"queue": "view_generation", "facet": "occupancy"}, float64(len(t.viewQueue)))
queueSizes.Set(map[string]string{"queue": "view_generation", "facet": "capacity"}, float64(cap(t.viewQueue)))
}
func (t *tieredStorage) writeMemory() {
begin := time.Now()
defer func() {
@ -182,6 +197,10 @@ func (t *tieredStorage) writeMemory() {
}
}
func (t *tieredStorage) Flush() {
t.flush()
}
// Write all pending appends.
func (t *tieredStorage) flush() (err error) {
t.writeMemory()
@ -223,7 +242,6 @@ func (f memoryToDiskFlusherVisitor) Filter(key, value interface{}) (filterResult
return storage.ACCEPT
}
f.flusher.valuesRejected++
return storage.STOP
}
@ -312,6 +330,7 @@ func (t *tieredStorage) renderView(viewJob viewJob) (err error) {
scans = viewJob.builder.ScanJobs()
// standingOperations = ops{}
// lastTime = time.Time{}
view = newView()
)
// Rebuilding of the frontier should happen on a conditional basis if a
@ -332,12 +351,10 @@ func (t *tieredStorage) renderView(viewJob viewJob) (err error) {
for _, scanJob := range scans {
// XXX: Memoize the last retrieval for forward scans.
var (
standingOperations ops
// standingOperations ops
)
fmt.Printf("Starting scan of %s...\n", scanJob)
// If the fingerprint is outside of the known frontier for the disk, the
// disk won't be queried at this time.
if !(t.diskFrontier == nil || scanJob.fingerprint.Less(t.diskFrontier.firstFingerprint) || t.diskFrontier.lastFingerprint.Less(scanJob.fingerprint)) {
fmt.Printf("Using diskFrontier %s\n", t.diskFrontier)
seriesFrontier, err := newSeriesFrontier(scanJob.fingerprint, *t.diskFrontier, iterator)
@ -347,28 +364,28 @@ func (t *tieredStorage) renderView(viewJob viewJob) (err error) {
}
if seriesFrontier != nil {
var (
targetKey = &dto.SampleKey{}
foundKey = &dto.SampleKey{}
foundValue *dto.SampleValueSeries
)
for _, operation := range scanJob.operations {
if seriesFrontier.lastTime.Before(operation.StartsAt()) {
fmt.Printf("operation %s occurs after %s; aborting...\n", operation, seriesFrontier.lastTime)
break
}
scanJob.operations = scanJob.operations[1:len(scanJob.operations)]
// if operation.StartsAt().Before(seriesFrontier.firstSupertime) {
// fmt.Printf("operation %s occurs before %s; discarding...\n", operation, seriesFrontier.firstSupertime)
// continue
// }
// if seriesFrontier.lastTime.Before(operation.StartsAt()) {
// fmt.Printf("operation %s occurs after %s; discarding...\n", operation, seriesFrontier.lastTime)
// continue
// }
var (
targetKey = &dto.SampleKey{}
foundKey = &dto.SampleKey{}
)
if operation.StartsAt().Before(seriesFrontier.firstSupertime) {
fmt.Printf("operation %s occurs before %s; discarding...\n", operation, seriesFrontier.firstSupertime)
continue
}
targetKey.Fingerprint = scanJob.fingerprint.ToDTO()
targetKey.Timestamp = indexable.EncodeTime(operation.StartsAt())
fmt.Println("target (unencoded) ->", targetKey)
rawKey, _ := coding.NewProtocolBufferEncoder(targetKey).Encode()
iterator.Seek(rawKey)
@ -378,35 +395,45 @@ func (t *tieredStorage) renderView(viewJob viewJob) (err error) {
panic(err)
}
fmt.Printf("startAt -> %s\n", operation.StartsAt())
fmt.Println("target ->", rawKey)
fmt.Println("found ->", iterator.Key())
fst := indexable.DecodeTime(foundKey.Timestamp)
lst := time.Unix(*foundKey.LastTimestamp, 0)
fmt.Printf("(%s, %s)\n", fst, lst)
fmt.Println(rawKey)
fmt.Println(foundKey)
var (
fst = indexable.DecodeTime(foundKey.Timestamp)
lst = time.Unix(*foundKey.LastTimestamp, 0)
)
if !((operation.StartsAt().Before(fst)) || lst.Before(operation.StartsAt())) {
fmt.Printf("operation %s occurs inside of %s...\n", operation, foundKey)
} else {
for i := 0; i < 3; i++ {
iterator.Next()
fmt.Println(i)
foundKey, err = extractSampleKey(iterator)
if err != nil {
panic(err)
}
fst = indexable.DecodeTime(foundKey.Timestamp)
lst = time.Unix(*foundKey.LastTimestamp, 0)
fmt.Println("found ->", iterator.Key())
fmt.Printf("(%s, %s)\n", fst, lst)
fmt.Println(foundKey)
foundValue, err = extractSampleValue(iterator)
if err != nil {
panic(err)
}
standingOperations = append(standingOperations, operation)
fmt.Printf("f -> %s\n", foundValue)
} else if operation.StartsAt().Before(fst) {
fmt.Printf("operation %s may occur in next entity; fast forwarding...\n", operation)
panic("oops")
} else {
panic("illegal state")
}
var (
elementCount = len(foundValue.Value)
searcher = func(i int) bool {
return time.Unix(*foundValue.Value[i].Timestamp, 0).After(operation.StartsAt())
}
index = sort.Search(elementCount, searcher)
)
foundValue.Value = foundValue.Value[index:elementCount]
switch operation.(type) {
case getValuesAtTimeOp:
if len(foundValue.Value) > 0 {
view.appendSample(scanJob.fingerprint, time.Unix(*foundValue.Value[0].Timestamp, 0), model.SampleValue(*foundValue.Value[0].Value))
}
if len(foundValue.Value) > 1 {
}
default:
panic("unhandled")
}
}
}
@ -445,6 +472,8 @@ func (t *tieredStorage) renderView(viewJob viewJob) (err error) {
// s.operations = s.operations[1:len(s.operations)]
// }
viewJob.output <- view
return
}

89
storage/metric/view.go
View file

@ -14,12 +14,16 @@
package metric
import (
"fmt"
"github.com/prometheus/prometheus/model"
"github.com/ryszard/goskiplist/skiplist"
"sort"
"time"
)
var (
_ = fmt.Sprintf("")
// firstSupertime is the smallest valid supertime that may be seeked to.
firstSupertime = []byte{0, 0, 0, 0, 0, 0, 0, 0}
// lastSupertime is the largest valid supertime that may be seeked to.
@ -99,3 +103,88 @@ func (v viewRequestBuilder) ScanJobs() (j scanJobs) {
return
}
type view struct {
fingerprintToSeries map[model.Fingerprint]viewStream
}
func (v view) appendSample(fingerprint model.Fingerprint, timestamp time.Time, value model.SampleValue) {
var (
series, ok = v.fingerprintToSeries[fingerprint]
)
if !ok {
series = newViewStream()
v.fingerprintToSeries[fingerprint] = series
}
series.add(timestamp, value)
}
func (v view) Close() {
v.fingerprintToSeries = make(map[model.Fingerprint]viewStream)
}
func (v view) GetValueAtTime(f model.Fingerprint, t time.Time) (s []model.SamplePair) {
var (
series, ok = v.fingerprintToSeries[f]
)
if !ok {
return
}
var (
iterator = series.values.Seek(skipListTime(t))
)
if iterator == nil {
return
}
defer iterator.Close()
if iterator.Key() == nil || iterator.Value() == nil {
return
}
s = append(s, model.SamplePair{
Timestamp: time.Time(iterator.Key().(skipListTime)),
Value: iterator.Value().(model.SampleValue),
})
if iterator.Next() {
s = append(s, model.SamplePair{
Timestamp: time.Time(iterator.Key().(skipListTime)),
Value: iterator.Value().(model.SampleValue),
})
}
return
}
func (v view) GetBoundaryValues(f model.Fingerprint, i model.Interval) (s []model.SamplePair) {
return
}
func (v view) GetRangeValues(f model.Fingerprint, i model.Interval) (s []model.SamplePair) {
return
}
func newView() view {
return view{
fingerprintToSeries: make(map[model.Fingerprint]viewStream),
}
}
type viewStream struct {
values *skiplist.SkipList
}
func (s viewStream) add(timestamp time.Time, value model.SampleValue) {
s.values.Set(skipListTime(timestamp), singletonValue(value))
}
func newViewStream() viewStream {
return viewStream{
values: skiplist.New(),
}
}