-
Notifications
You must be signed in to change notification settings - Fork 487
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Convert to using prometheus appendable (#2431)
* initial commit of switching over to using appendable * comments * simplify fanout * Switch to using a common fanout appendable. * Merge changes * Fix lint * Add check * remove panics * PR feedback * fix error * separate fanout and intercept patterns (#2447) * separate fanout and intercept patterns * testing flow * Add middleware style next to interceptor. * Simplify the remote_write and add multierror.
- Loading branch information
1 parent
37d333a
commit 0513789
Showing
14 changed files
with
385 additions
and
430 deletions.
There are no files selected for viewing
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| package prometheus | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sync" | ||
|
|
||
| "github.com/hashicorp/go-multierror" | ||
|
|
||
| "github.com/prometheus/prometheus/model/exemplar" | ||
| "github.com/prometheus/prometheus/model/labels" | ||
| "github.com/prometheus/prometheus/model/metadata" | ||
|
|
||
| "github.com/prometheus/prometheus/storage" | ||
| ) | ||
|
|
||
| var _ storage.Appendable = (*Fanout)(nil) | ||
|
|
||
| // Fanout supports the default Flow style of appendables since it can go to multiple outputs. It also allows the intercepting of appends. | ||
| type Fanout struct { | ||
| mut sync.RWMutex | ||
| // children is where to fan out. | ||
| children []storage.Appendable | ||
| // ComponentID is what component this belongs to. | ||
| componentID string | ||
| } | ||
|
|
||
| // NewFanout creates a fanout appendable. | ||
| func NewFanout(children []storage.Appendable, componentID string) *Fanout { | ||
| return &Fanout{ | ||
| children: children, | ||
| componentID: componentID, | ||
| } | ||
| } | ||
|
|
||
| // UpdateChildren allows changing of the children of the fanout. | ||
| func (f *Fanout) UpdateChildren(children []storage.Appendable) { | ||
| f.mut.Lock() | ||
| defer f.mut.Unlock() | ||
| f.children = children | ||
| } | ||
|
|
||
| // Appender satisfies the Appendable interface. | ||
| func (f *Fanout) Appender(ctx context.Context) storage.Appender { | ||
| f.mut.RLock() | ||
| defer f.mut.RUnlock() | ||
|
|
||
| app := &appender{ | ||
| children: make([]storage.Appender, 0), | ||
| componentID: f.componentID, | ||
| } | ||
| for _, x := range f.children { | ||
| if x == nil { | ||
| continue | ||
| } | ||
| app.children = append(app.children, x.Appender(ctx)) | ||
| } | ||
| return app | ||
| } | ||
|
|
||
| var _ storage.Appender = (*appender)(nil) | ||
|
|
||
| type appender struct { | ||
| children []storage.Appender | ||
| componentID string | ||
| } | ||
|
|
||
| // Append satisfies the Appender interface. | ||
| func (a *appender) Append(ref storage.SeriesRef, l labels.Labels, t int64, v float64) (storage.SeriesRef, error) { | ||
| if ref == 0 { | ||
| ref = storage.SeriesRef(GlobalRefMapping.GetOrAddGlobalRefID(l)) | ||
| } | ||
| var multiErr error | ||
| for _, x := range a.children { | ||
| _, err := x.Append(ref, l, t, v) | ||
| if err != nil { | ||
| multiErr = multierror.Append(multiErr, err) | ||
| } | ||
| } | ||
| return ref, multiErr | ||
| } | ||
|
|
||
| // Commit satisfies the Appender interface. | ||
| func (a *appender) Commit() error { | ||
| for _, x := range a.children { | ||
| _ = x.Commit() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Rollback satisifies the Appender interface. | ||
| func (a *appender) Rollback() error { | ||
| for _, x := range a.children { | ||
| _, _ = x, a.Rollback() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // AppendExemplar satisfies the Appender interface. | ||
| func (a *appender) AppendExemplar(ref storage.SeriesRef, l labels.Labels, e exemplar.Exemplar) (storage.SeriesRef, error) { | ||
| return 0, fmt.Errorf("appendExemplar not supported yet") | ||
| } | ||
|
|
||
| // UpdateMetadata satisifies the Appender interface. | ||
| func (a *appender) UpdateMetadata(ref storage.SeriesRef, l labels.Labels, m metadata.Metadata) (storage.SeriesRef, error) { | ||
| return 0, fmt.Errorf("updateMetadata not supported yet") | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.