This repository has been archived by the owner on Apr 2, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 170
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement adapters for queryable and appendable.
Signed-off-by: Harkishen-Singh <harkishensingh@hotmail.com> This commits adds adapters to convert our db-ingestor and samples querier into the appendable and queryable that is expected by the rules manager module.
- Loading branch information
1 parent
bf1a706
commit 816c9f9
Showing
13 changed files
with
314 additions
and
70 deletions.
There are no files selected for viewing
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
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
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
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,74 @@ | ||
| package adapters | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/pkg/errors" | ||
|
|
||
| "github.com/prometheus/prometheus/model/exemplar" | ||
| "github.com/prometheus/prometheus/model/labels" | ||
| "github.com/prometheus/prometheus/storage" | ||
|
|
||
| "github.com/timescale/promscale/pkg/pgmodel/ingestor" | ||
| "github.com/timescale/promscale/pkg/pgmodel/metrics" | ||
| "github.com/timescale/promscale/pkg/pgmodel/model" | ||
| "github.com/timescale/promscale/pkg/prompb" | ||
| "github.com/timescale/promscale/pkg/util" | ||
| ) | ||
|
|
||
| var samplesIngested = metrics.IngestorItems.With(map[string]string{"type": "metric", "kind": "sample", "subsystem": "rules"}) | ||
|
|
||
| type ingestAdapter struct { | ||
| ingestor *ingestor.DBIngestor | ||
| } | ||
|
|
||
| func NewIngestAdapter(ingestor *ingestor.DBIngestor) *ingestAdapter { | ||
| return &ingestAdapter{ingestor} | ||
| } | ||
|
|
||
| type appenderAdapter struct { | ||
| data map[string][]model.Insertable | ||
| ingestor *ingestor.DBIngestor | ||
| } | ||
|
|
||
| func (a ingestAdapter) Appender(_ context.Context) storage.Appender { | ||
| return &appenderAdapter{ | ||
| data: make(map[string][]model.Insertable), | ||
| ingestor: a.ingestor, | ||
| } | ||
| } | ||
|
|
||
| func (app *appenderAdapter) Append(_ storage.SeriesRef, l labels.Labels, t int64, v float64) (storage.SeriesRef, error) { | ||
| series, metricName, err := app.ingestor.SCache.GetSeriesFromProtos(util.LabelToPrompbLabels(l)) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("error creating series: %w", err) | ||
| } | ||
|
|
||
| samples := model.NewPromSamples(series, []prompb.Sample{{Timestamp: t, Value: v}}) | ||
| if _, found := app.data[metricName]; !found { | ||
| app.data[metricName] = make([]model.Insertable, 0) | ||
| } | ||
| app.data[metricName] = append(app.data[metricName], samples) | ||
| return 0, nil | ||
| } | ||
|
|
||
| func (app *appenderAdapter) AppendExemplar(_ storage.SeriesRef, l labels.Labels, e exemplar.Exemplar) (storage.SeriesRef, error) { | ||
| // We do not support appending exemplars in recording rules since this is not yet implemented upstream. | ||
| // Once upstream implements this feature, we can modify this function. | ||
| return 0, nil | ||
| } | ||
|
|
||
| func (app *appenderAdapter) Commit() error { | ||
| numInsertablesIngested, err := app.ingestor.Dispatcher().InsertTs(context.Background(), model.Data{Rows: app.data, ReceivedTime: time.Now()}) | ||
| if err == nil { | ||
| samplesIngested.Add(float64(numInsertablesIngested)) | ||
| } | ||
| return errors.WithMessage(err, "rules: error ingesting data into db-ingestor") | ||
| } | ||
|
|
||
| func (app *appenderAdapter) Rollback() error { | ||
| app.data = nil | ||
| return nil | ||
| } |
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,51 @@ | ||
| package adapters | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/prometheus/prometheus/model/labels" | ||
| "github.com/prometheus/prometheus/storage" | ||
| "github.com/timescale/promscale/pkg/promql" | ||
| ) | ||
|
|
||
| type queryAdapter struct { | ||
| queryable promql.Queryable | ||
| } | ||
|
|
||
| func NewQueryAdapter(q promql.Queryable) *queryAdapter { | ||
| return &queryAdapter{q} | ||
| } | ||
|
|
||
| func (q *queryAdapter) Querier(ctx context.Context, mint int64, maxt int64) (storage.Querier, error) { | ||
| qr, err := q.queryable.SamplesQuerier(ctx, mint, maxt) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("samples-querier: %w", err) | ||
| } | ||
| return querierAdapter{qr}, nil | ||
| } | ||
|
|
||
| type querierAdapter struct { | ||
| qr promql.SamplesQuerier | ||
| } | ||
|
|
||
| func (q querierAdapter) Select(sortSeries bool, hints *storage.SelectHints, matchers ...*labels.Matcher) storage.SeriesSet { | ||
| // Pushdowns are not supported here. This is fine as Prometheus rule-manager only uses queryable to know | ||
| // the previous state of the alert. This function is not used in recording/alerting rules evaluation. | ||
| seriesSet, _ := q.qr.Select(sortSeries, hints, nil, nil, matchers...) | ||
| return seriesSet | ||
| } | ||
|
|
||
| func (q querierAdapter) LabelValues(name string, _ ...*labels.Matcher) ([]string, storage.Warnings, error) { | ||
| // Weak TODO: We need to implement the matchers. | ||
| return q.qr.LabelValues(name) | ||
| } | ||
|
|
||
| func (q querierAdapter) LabelNames(matchers ...*labels.Matcher) ([]string, storage.Warnings, error) { | ||
| return q.qr.LabelNames(matchers...) | ||
| } | ||
|
|
||
| func (q querierAdapter) Close() error { | ||
| q.qr.Close() | ||
| return nil | ||
| } |
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
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.