remote: take code from master

This commit is contained in:
Fabian Reinartz 2017-03-07 11:43:32 +01:00
parent d9fb57cde4
commit 9eb1d6c927

View file

@ -1,8 +1,4 @@
<<<<<<< HEAD
// Copyright 2016 The Prometheus Authors
=======
// Copyright 2017 The Prometheus Authors // Copyright 2017 The Prometheus Authors
>>>>>>> master
// Licensed under the Apache License, Version 2.0 (the "License"); // Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License. // you may not use this file except in compliance with the License.
// You may obtain a copy of the License at // You may obtain a copy of the License at
@ -23,29 +19,12 @@ import (
"github.com/prometheus/common/model" "github.com/prometheus/common/model"
"github.com/prometheus/prometheus/config" "github.com/prometheus/prometheus/config"
<<<<<<< HEAD
"github.com/prometheus/prometheus/relabel"
=======
>>>>>>> master
) )
// Storage allows queueing samples for remote writes. // Storage allows queueing samples for remote writes.
type Storage struct { type Storage struct {
<<<<<<< HEAD
mtx sync.RWMutex
externalLabels model.LabelSet
conf config.RemoteWriteConfig
queue *StorageQueueManager
}
// New returns a new remote Storage.
func New() *Storage {
return &Storage{}
=======
mtx sync.RWMutex mtx sync.RWMutex
queues []*QueueManager queues []*QueueManager
>>>>>>> master
} }
// ApplyConfig updates the state as the new config requires. // ApplyConfig updates the state as the new config requires.
@ -53,28 +32,6 @@ func (s *Storage) ApplyConfig(conf *config.Config) error {
s.mtx.Lock() s.mtx.Lock()
defer s.mtx.Unlock() defer s.mtx.Unlock()
<<<<<<< HEAD
// TODO: we should only stop & recreate queues which have changes,
// as this can be quite disruptive.
var newQueue *StorageQueueManager
if conf.RemoteWriteConfig.URL != nil {
c, err := NewClient(conf.RemoteWriteConfig)
if err != nil {
return err
}
newQueue = NewStorageQueueManager(c, nil)
}
if s.queue != nil {
s.queue.Stop()
}
s.queue = newQueue
s.conf = conf.RemoteWriteConfig
s.externalLabels = conf.GlobalConfig.ExternalLabels
if s.queue != nil {
s.queue.Start()
=======
newQueues := []*QueueManager{} newQueues := []*QueueManager{}
// TODO: we should only stop & recreate queues which have changes, // TODO: we should only stop & recreate queues which have changes,
// as this can be quite disruptive. // as this can be quite disruptive.
@ -97,15 +54,14 @@ func (s *Storage) ApplyConfig(conf *config.Config) error {
s.queues = newQueues s.queues = newQueues
for _, q := range s.queues { for _, q := range s.queues {
q.Start() q.Start()
>>>>>>> master
} }
return nil return nil
} }
// Stop the background processing of the storage queues. // Stop the background processing of the storage queues.
func (s *Storage) Stop() { func (s *Storage) Stop() {
if s.queue != nil { for _, q := range s.queues {
s.queue.Stop() q.Stop()
} }
} }
@ -114,32 +70,9 @@ func (s *Storage) Append(smpl *model.Sample) error {
s.mtx.RLock() s.mtx.RLock()
defer s.mtx.RUnlock() defer s.mtx.RUnlock()
<<<<<<< HEAD
if s.queue == nil {
return nil
}
var snew model.Sample
snew = *smpl
snew.Metric = smpl.Metric.Clone()
for ln, lv := range s.externalLabels {
if _, ok := smpl.Metric[ln]; !ok {
snew.Metric[ln] = lv
}
}
snew.Metric = model.Metric(
relabel.Process(model.LabelSet(snew.Metric), s.conf.WriteRelabelConfigs...))
if snew.Metric == nil {
return nil
}
s.queue.Append(&snew)
=======
for _, q := range s.queues { for _, q := range s.queues {
q.Append(smpl) q.Append(smpl)
} }
>>>>>>> master
return nil return nil
} }