mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-01-12 22:37:27 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Storage interface to TieredStorage.

Browse source 
This commit drops the Storage interface and just replaces it with a
publicized TieredStorage type.  Storage had been anticipated to be
used as a wrapper for testability but just was not used due to
practicality.  Merely overengineered.  My bad.  Anyway, we will
eventually instantiate the TieredStorage dependencies in main.go and
pass them in for more intelligent lifecycle management.

These changes will pave the way for managing the curators without
Law of Demeter violations.
This commit is contained in:
Matt T. Proud 2013-05-02 18:27:12 +02:00
parent 59e5ddeed3
commit ce45787dbf
8 changed files with 55 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
appstate/appstate.go
View file

@ -26,7 +26,7 @@ import (
type ApplicationState struct { type ApplicationState struct {
Config config.Config Config config.Config
RuleManager rules.RuleManager RuleManager rules.RuleManager
Storage metric.Storage Storage metric.TieredStorage
TargetManager retrieval.TargetManager TargetManager retrieval.TargetManager
BuildInfo map[string]string BuildInfo map[string]string
CurationState chan metric.CurationState CurationState chan metric.CurationState

11
main.go
View file

@ -42,7 +42,7 @@ var (
) )
type prometheus struct { type prometheus struct {
storage metric.Storage storage metric.TieredStorage
// TODO: Refactor channels to work with arrays of results for better chunking. // TODO: Refactor channels to work with arrays of results for better chunking.
scrapeResults chan format.Result scrapeResults chan format.Result
ruleResults chan *rules.Result ruleResults chan *rules.Result
@ -86,12 +86,15 @@ func main() {
if err != nil { if err != nil {
log.Fatalf("Error opening storage: %s", err) log.Fatalf("Error opening storage: %s", err)
} }
if ts == nil {
log.Fatalln("Nil tiered storage.")
}
scrapeResults := make(chan format.Result, *scrapeResultsQueueCapacity) scrapeResults := make(chan format.Result, *scrapeResultsQueueCapacity)
ruleResults := make(chan *rules.Result, *ruleResultsQueueCapacity) ruleResults := make(chan *rules.Result, *ruleResultsQueueCapacity)
prometheus := prometheus{ prometheus := prometheus{
storage: ts, storage: *ts,
scrapeResults: scrapeResults, scrapeResults: scrapeResults,
ruleResults: ruleResults, ruleResults: ruleResults,
} }
@ -105,7 +108,7 @@ func main() {
targetManager := retrieval.NewTargetManager(scrapeResults, *concurrentRetrievalAllowance) targetManager := retrieval.NewTargetManager(scrapeResults, *concurrentRetrievalAllowance)
targetManager.AddTargetsFromConfig(conf) targetManager.AddTargetsFromConfig(conf)
ast.SetStorage(ts) ast.SetStorage(*ts)
ruleManager := rules.NewRuleManager(ruleResults, conf.EvaluationInterval()) ruleManager := rules.NewRuleManager(ruleResults, conf.EvaluationInterval())
err = ruleManager.AddRulesFromConfig(conf) err = ruleManager.AddRulesFromConfig(conf)
@ -118,7 +121,7 @@ func main() {
Config: conf, Config: conf,
CurationState: make(chan metric.CurationState), CurationState: make(chan metric.CurationState),
RuleManager: ruleManager, RuleManager: ruleManager,
Storage: ts, Storage: *ts,
TargetManager: targetManager, TargetManager: targetManager,
} }

8
rules/ast/persistence_adapter.go
View file

@ -24,7 +24,9 @@ var defaultStalenessDelta = flag.Int("defaultStalenessDelta", 300, "Default stal
// AST-global storage to use for operations that are not supported by views // AST-global storage to use for operations that are not supported by views
// (i.e. metric->fingerprint lookups). // (i.e. metric->fingerprint lookups).
var queryStorage metric.Storage = nil //
// BUG(julius): Wrap this into non-global state.
var queryStorage *metric.TieredStorage
// Describes the lenience limits to apply to values from the materialized view. // Describes the lenience limits to apply to values from the materialized view.
type StalenessPolicy struct { type StalenessPolicy struct {
@ -167,8 +169,8 @@ func (v *viewAdapter) GetRangeValues(fingerprints model.Fingerprints, interval *
return sampleSets, nil return sampleSets, nil
} }
func SetStorage(storage metric.Storage) { func SetStorage(storage metric.TieredStorage) {
queryStorage = storage queryStorage = &storage
} }
func NewViewAdapter(view metric.View) *viewAdapter { func NewViewAdapter(view metric.View) *viewAdapter {

9
rules/rules_test.go
View file

@ -49,10 +49,13 @@ func vectorComparisonString(expected []string, actual []string) string {
separator) separator)
} }
func newTestStorage(t test.Tester) (storage metric.Storage, closer test.Closer) { func newTestStorage(t test.Tester) (storage *metric.TieredStorage, closer test.Closer) {
storage, closer = metric.NewTestTieredStorage(t) storage, closer = metric.NewTestTieredStorage(t)
ast.SetStorage(storage) if storage == nil {
storeMatrix(storage, testMatrix) t.Fatal("storage == nil")
}
ast.SetStorage(*storage)
storeMatrix(*storage, testMatrix)
return return
} }

2
rules/testdata.go
View file

@ -51,7 +51,7 @@ func getTestVectorFromTestMatrix(matrix ast.Matrix) ast.Vector {
return vector return vector
} }
func storeMatrix(storage metric.Storage, matrix ast.Matrix) (err error) { func storeMatrix(storage metric.TieredStorage, matrix ast.Matrix) (err error) {
pendingSamples := model.Samples{} pendingSamples := model.Samples{}
for _, sampleSet := range matrix { for _, sampleSet := range matrix {
for _, sample := range sampleSet.Values { for _, sample := range sampleSet.Values {

4
storage/metric/test_helper.go
View file

@ -78,7 +78,7 @@ func buildMemoryTestPersistence(f func(p MetricPersistence, t test.Tester)) func
} }
type testTieredStorageCloser struct { type testTieredStorageCloser struct {
storage Storage storage *TieredStorage
directory test.Closer directory test.Closer
} }
@ -87,7 +87,7 @@ func (t testTieredStorageCloser) Close() {
t.directory.Close() t.directory.Close()
} }
func NewTestTieredStorage(t test.Tester) (storage Storage, closer test.Closer) { func NewTestTieredStorage(t test.Tester) (storage *TieredStorage, closer test.Closer) {
var directory test.TemporaryDirectory var directory test.TemporaryDirectory
directory = test.NewTemporaryDirectory("test_tiered_storage", t) directory = test.NewTemporaryDirectory("test_tiered_storage", t)
storage, err := NewTieredStorage(appendQueueSize, 2500, 1000, 5*time.Second, 15*time.Second, 0*time.Second, directory.Path()) storage, err := NewTieredStorage(appendQueueSize, 2500, 1000, 5*time.Second, 15*time.Second, 0*time.Second, directory.Path())

71
storage/metric/tiered.go
View file

@ -27,9 +27,9 @@ import (
"time" "time"
) )
// tieredStorage both persists samples and generates materialized views for // TieredStorage both persists samples and generates materialized views for
// queries. // queries.
type tieredStorage struct { type TieredStorage struct {
appendToDiskQueue chan model.Samples appendToDiskQueue chan model.Samples
appendToMemoryQueue chan model.Samples appendToMemoryQueue chan model.Samples
diskFrontier *diskFrontier diskFrontier *diskFrontier
@ -51,36 +51,13 @@ type viewJob struct {
err chan error err chan error
} }
// Provides a unified means for batch appending values into the datastore along func NewTieredStorage(appendToMemoryQueueDepth, appendToDiskQueueDepth, viewQueueDepth uint, flushMemoryInterval, writeMemoryInterval, memoryTTL time.Duration, root string) (storage *TieredStorage, err error) {
// with querying for values in an efficient way.
type Storage interface {
// Enqueues Samples for storage.
AppendSamples(model.Samples) error
// Enqueus a ViewRequestBuilder for materialization, subject to a timeout.
MakeView(request ViewRequestBuilder, timeout time.Duration) (View, error)
// Starts serving requests.
Serve()
// Stops the storage subsystem, flushing all pending operations.
Drain()
Flush()
Close()
// Get all label values that are associated with the provided label name.
GetAllValuesForLabel(model.LabelName) (model.LabelValues, error)
// Get all of the metric fingerprints that are associated with the provided
// label set.
GetFingerprintsForLabelSet(model.LabelSet) (model.Fingerprints, error)
// Get the metric associated with the provided fingerprint.
GetMetricForFingerprint(model.Fingerprint) (m *model.Metric, err error)
}
func NewTieredStorage(appendToMemoryQueueDepth, appendToDiskQueueDepth, viewQueueDepth uint, flushMemoryInterval, writeMemoryInterval, memoryTTL time.Duration, root string) (storage Storage, err error) {
diskStorage, err := NewLevelDBMetricPersistence(root) diskStorage, err := NewLevelDBMetricPersistence(root)
if err != nil { if err != nil {
return return
} }
storage = &tieredStorage{ storage = &TieredStorage{
appendToDiskQueue: make(chan model.Samples, appendToDiskQueueDepth), appendToDiskQueue: make(chan model.Samples, appendToDiskQueueDepth),
appendToMemoryQueue: make(chan model.Samples, appendToMemoryQueueDepth), appendToMemoryQueue: make(chan model.Samples, appendToMemoryQueueDepth),
diskStorage: diskStorage, diskStorage: diskStorage,
@ -94,7 +71,8 @@ func NewTieredStorage(appendToMemoryQueueDepth, appendToDiskQueueDepth, viewQueu
return return
} }
func (t tieredStorage) AppendSamples(s model.Samples) (err error) { // Enqueues Samples for storage.
func (t TieredStorage) AppendSamples(s model.Samples) (err error) {
if len(t.draining) > 0 { if len(t.draining) > 0 {
return fmt.Errorf("Storage is in the process of draining.") return fmt.Errorf("Storage is in the process of draining.")
} }
@ -104,7 +82,8 @@ func (t tieredStorage) AppendSamples(s model.Samples) (err error) {
return return
} }
func (t tieredStorage) Drain() { // Stops the storage subsystem, flushing all pending operations.
func (t TieredStorage) Drain() {
log.Println("Starting drain...") log.Println("Starting drain...")
drainingDone := make(chan bool) drainingDone := make(chan bool)
if len(t.draining) == 0 { if len(t.draining) == 0 {
@ -114,7 +93,8 @@ func (t tieredStorage) Drain() {
log.Println("Done.") log.Println("Done.")
} }
func (t *tieredStorage) MakeView(builder ViewRequestBuilder, deadline time.Duration) (view View, err error) { // Enqueus a ViewRequestBuilder for materialization, subject to a timeout.
func (t TieredStorage) MakeView(builder ViewRequestBuilder, deadline time.Duration) (view View, err error) {
if len(t.draining) > 0 { if len(t.draining) > 0 {
err = fmt.Errorf("Storage is in the process of draining.") err = fmt.Errorf("Storage is in the process of draining.")
return return
@ -148,7 +128,7 @@ func (t *tieredStorage) MakeView(builder ViewRequestBuilder, deadline time.Durat
return return
} }
func (t *tieredStorage) rebuildDiskFrontier(i leveldb.Iterator) (err error) { func (t *TieredStorage) rebuildDiskFrontier(i leveldb.Iterator) (err error) {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Since(begin) duration := time.Since(begin)
@ -163,7 +143,8 @@ func (t *tieredStorage) rebuildDiskFrontier(i leveldb.Iterator) (err error) {
return return
} }
func (t *tieredStorage) Serve() { // Starts serving requests.
func (t TieredStorage) Serve() {
flushMemoryTicker := time.NewTicker(t.flushMemoryInterval) flushMemoryTicker := time.NewTicker(t.flushMemoryInterval)
defer flushMemoryTicker.Stop() defer flushMemoryTicker.Stop()
writeMemoryTicker := time.NewTicker(t.writeMemoryInterval) writeMemoryTicker := time.NewTicker(t.writeMemoryInterval)
@ -193,7 +174,7 @@ func (t *tieredStorage) Serve() {
} }
} }
func (t *tieredStorage) reportQueues() { 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": "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_disk", "facet": "capacity"}, float64(cap(t.appendToDiskQueue)))
@ -204,7 +185,7 @@ func (t *tieredStorage) reportQueues() {
queueSizes.Set(map[string]string{"queue": "view_generation", "facet": "capacity"}, float64(cap(t.viewQueue))) queueSizes.Set(map[string]string{"queue": "view_generation", "facet": "capacity"}, float64(cap(t.viewQueue)))
} }
func (t *tieredStorage) writeMemory() { func (t *TieredStorage) writeMemory() {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Since(begin) duration := time.Since(begin)
@ -222,11 +203,11 @@ func (t *tieredStorage) writeMemory() {
} }
} }
func (t tieredStorage) Flush() { func (t TieredStorage) Flush() {
t.flush() t.flush()
} }
func (t tieredStorage) Close() { func (t TieredStorage) Close() {
log.Println("Closing tiered storage...") log.Println("Closing tiered storage...")
t.Drain() t.Drain()
t.diskStorage.Close() t.diskStorage.Close()
@ -239,7 +220,7 @@ func (t tieredStorage) Close() {
} }
// Write all pending appends. // Write all pending appends.
func (t tieredStorage) flush() (err error) { func (t TieredStorage) flush() (err error) {
// Trim any old values to reduce iterative write costs. // Trim any old values to reduce iterative write costs.
t.flushMemory() t.flushMemory()
t.writeMemory() t.writeMemory()
@ -330,7 +311,7 @@ func (f memoryToDiskFlusher) Close() {
} }
// Persist a whole bunch of samples to the datastore. // Persist a whole bunch of samples to the datastore.
func (t *tieredStorage) flushMemory() { func (t *TieredStorage) flushMemory() {
begin := time.Now() begin := time.Now()
defer func() { defer func() {
duration := time.Since(begin) duration := time.Since(begin)
@ -353,7 +334,7 @@ func (t *tieredStorage) flushMemory() {
return return
} }
func (t *tieredStorage) renderView(viewJob viewJob) { func (t TieredStorage) renderView(viewJob viewJob) {
// Telemetry. // Telemetry.
var err error var err error
begin := time.Now() begin := time.Now()
@ -482,7 +463,7 @@ func (t *tieredStorage) renderView(viewJob viewJob) {
return return
} }
func (t *tieredStorage) loadChunkAroundTime(iterator leveldb.Iterator, frontier *seriesFrontier, fingerprint model.Fingerprint, ts time.Time) (chunk model.Values) { func (t TieredStorage) loadChunkAroundTime(iterator leveldb.Iterator, frontier *seriesFrontier, fingerprint model.Fingerprint, ts time.Time) (chunk model.Values) {
var ( var (
targetKey = &dto.SampleKey{ targetKey = &dto.SampleKey{
Fingerprint: fingerprint.ToDTO(), Fingerprint: fingerprint.ToDTO(),
@ -558,7 +539,8 @@ func (t *tieredStorage) loadChunkAroundTime(iterator leveldb.Iterator, frontier
return return
} }
func (t *tieredStorage) GetAllValuesForLabel(labelName model.LabelName) (values model.LabelValues, err error) { // Get all label values that are associated with the provided label name.
func (t TieredStorage) GetAllValuesForLabel(labelName model.LabelName) (values model.LabelValues, err error) {
diskValues, err := t.diskStorage.GetAllValuesForLabel(labelName) diskValues, err := t.diskStorage.GetAllValuesForLabel(labelName)
if err != nil { if err != nil {
return return
@ -579,7 +561,9 @@ func (t *tieredStorage) GetAllValuesForLabel(labelName model.LabelName) (values
return return
} }
func (t *tieredStorage) GetFingerprintsForLabelSet(labelSet model.LabelSet) (fingerprints model.Fingerprints, err error) { // Get all of the metric fingerprints that are associated with the provided
// label set.
func (t TieredStorage) GetFingerprintsForLabelSet(labelSet model.LabelSet) (fingerprints model.Fingerprints, err error) {
memFingerprints, err := t.memoryArena.GetFingerprintsForLabelSet(labelSet) memFingerprints, err := t.memoryArena.GetFingerprintsForLabelSet(labelSet)
if err != nil { if err != nil {
return return
@ -599,7 +583,8 @@ func (t *tieredStorage) GetFingerprintsForLabelSet(labelSet model.LabelSet) (fin
return return
} }
func (t *tieredStorage) GetMetricForFingerprint(f model.Fingerprint) (m *model.Metric, err error) { // Get the metric associated with the provided fingerprint.
func (t TieredStorage) GetMetricForFingerprint(f model.Fingerprint) (m *model.Metric, err error) {
m, err = t.memoryArena.GetMetricForFingerprint(f) m, err = t.memoryArena.GetMetricForFingerprint(f)
if err != nil { if err != nil {
return return

10
storage/metric/tiered_test.go
View file

@ -348,7 +348,7 @@ func testMakeView(t test.Tester, flushToDisk bool) {
if flushToDisk { if flushToDisk {
tiered.Flush() tiered.Flush()
} else { } else {
tiered.(*tieredStorage).writeMemory() tiered.writeMemory()
} }
requestBuilder := NewViewRequestBuilder() requestBuilder := NewViewRequestBuilder()
@ -480,12 +480,12 @@ func TestGetAllValuesForLabel(t *testing.T) {
Metric: model.Metric{model.MetricNameLabel: model.LabelValue(metric.metricName)}, Metric: model.Metric{model.MetricNameLabel: model.LabelValue(metric.metricName)},
} }
if metric.appendToMemory { if metric.appendToMemory {
if err := tiered.(*tieredStorage).memoryArena.AppendSample(sample); err != nil { if err := tiered.memoryArena.AppendSample(sample); err != nil {
t.Fatalf("%d.%d. failed to add fixture data: %s", i, j, err) t.Fatalf("%d.%d. failed to add fixture data: %s", i, j, err)
} }
} }
if metric.appendToDisk { if metric.appendToDisk {
if err := tiered.(*tieredStorage).diskStorage.AppendSample(sample); err != nil { if err := tiered.diskStorage.AppendSample(sample); err != nil {
t.Fatalf("%d.%d. failed to add fixture data: %s", i, j, err) t.Fatalf("%d.%d. failed to add fixture data: %s", i, j, err)
} }
} }
@ -517,10 +517,10 @@ func TestGetFingerprintsForLabelSet(t *testing.T) {
diskSample := model.Sample{ diskSample := model.Sample{
Metric: model.Metric{model.MetricNameLabel: "http_requests", "method": "/bar"}, Metric: model.Metric{model.MetricNameLabel: "http_requests", "method": "/bar"},
} }
if err := tiered.(*tieredStorage).memoryArena.AppendSample(memorySample); err != nil { if err := tiered.memoryArena.AppendSample(memorySample); err != nil {
t.Fatalf("Failed to add fixture data: %s", err) t.Fatalf("Failed to add fixture data: %s", err)
} }
if err := tiered.(*tieredStorage).diskStorage.AppendSample(diskSample); err != nil { if err := tiered.diskStorage.AppendSample(diskSample); err != nil {
t.Fatalf("Failed to add fixture data: %s", err) t.Fatalf("Failed to add fixture data: %s", err)
} }
tiered.Flush() tiered.Flush()