RepoMirrors/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus synced 2024-12-27 09:02:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update example remote adapters for change in proto location.

Browse Source
This commit is contained in:
Tom Wilkie 2017-07-19 16:39:02 +01:00
parent ec999ff397
commit cf105f9d57
3 changed files with 38 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
documentation/examples/remote_storage/example_write_adapter/server.go
View File

@ -22,7 +22,7 @@ import (
"github.com/golang/snappy"
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/storage/remote"
"github.com/prometheus/prometheus/prompb"
)
func main() {
@ -39,7 +39,7 @@ func main() {
return
}
var req remote.WriteRequest
var req prompb.WriteRequest
if err := proto.Unmarshal(reqBuf, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
@ -53,7 +53,7 @@ func main() {
fmt.Println(m)
for _, s := range ts.Samples {
fmt.Printf(" %f %d\n", s.Value, s.TimestampMs)
fmt.Printf(" %f %d\n", s.Value, s.Timestamp)
}
}
})

54
documentation/examples/remote_storage/remote_storage_adapter/influxdb/client.go
View File

@ -22,7 +22,7 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/common/log"
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/storage/remote"
"github.com/prometheus/prometheus/prompb"
influx "github.com/influxdata/influxdb/client/v2"
)
@ -101,8 +101,8 @@ func (c *Client) Write(samples model.Samples) error {
return c.client.Write(bps)
}
func (c *Client) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) {
labelsToSeries := map[string]*remote.TimeSeries{}
func (c *Client) Read(req *prompb.ReadRequest) (*prompb.ReadResponse, error) {
labelsToSeries := map[string]*prompb.TimeSeries{}
for _, q := range req.Queries {
command, err := c.buildCommand(q)
if err != nil {
@ -123,9 +123,9 @@ func (c *Client) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) {
}
}
resp := remote.ReadResponse{
Results: []*remote.QueryResult{
{Timeseries: make([]*remote.TimeSeries, 0, len(labelsToSeries))},
resp := prompb.ReadResponse{
Results: []*prompb.QueryResult{
{Timeseries: make([]*prompb.TimeSeries, 0, len(labelsToSeries))},
},
}
for _, ts := range labelsToSeries {
@ -134,7 +134,7 @@ func (c *Client) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) {
return &resp, nil
}
func (c *Client) buildCommand(q *remote.Query) (string, error) {
func (c *Client) buildCommand(q *prompb.Query) (string, error) {
matchers := make([]string, 0, len(q.Matchers))
// If we don't find a metric name matcher, query all metrics
// (InfluxDB measurements) by default.
@ -142,9 +142,9 @@ func (c *Client) buildCommand(q *remote.Query) (string, error) {
for _, m := range q.Matchers {
if m.Name == model.MetricNameLabel {
switch m.Type {
case remote.MatchType_EQUAL:
case prompb.LabelMatcher_EQ:
from = fmt.Sprintf("FROM %q.%q", c.retentionPolicy, m.Value)
case remote.MatchType_REGEX_MATCH:
case prompb.LabelMatcher_RE:
from = fmt.Sprintf("FROM %q./^%s$/", c.retentionPolicy, escapeSlashes(m.Value))
default:
// TODO: Figure out how to support these efficiently.
@ -154,13 +154,13 @@ func (c *Client) buildCommand(q *remote.Query) (string, error) {
}
switch m.Type {
case remote.MatchType_EQUAL:
case prompb.LabelMatcher_EQ:
matchers = append(matchers, fmt.Sprintf("%q = '%s'", m.Name, escapeSingleQuotes(m.Value)))
case remote.MatchType_NOT_EQUAL:
case prompb.LabelMatcher_NEQ:
matchers = append(matchers, fmt.Sprintf("%q != '%s'", m.Name, escapeSingleQuotes(m.Value)))
case remote.MatchType_REGEX_MATCH:
case prompb.LabelMatcher_RE:
matchers = append(matchers, fmt.Sprintf("%q =~ /^%s$/", m.Name, escapeSlashes(m.Value)))
case remote.MatchType_REGEX_NO_MATCH:
case prompb.LabelMatcher_NRE:
matchers = append(matchers, fmt.Sprintf("%q !~ /^%s$/", m.Name, escapeSlashes(m.Value)))
default:
return "", fmt.Errorf("unknown match type %v", m.Type)
@ -180,13 +180,13 @@ func escapeSlashes(str string) string {
return strings.Replace(str, `/`, `\/`, -1)
}
func mergeResult(labelsToSeries map[string]*remote.TimeSeries, results []influx.Result) error {
func mergeResult(labelsToSeries map[string]*prompb.TimeSeries, results []influx.Result) error {
for _, r := range results {
for _, s := range r.Series {
k := concatLabels(s.Tags)
ts, ok := labelsToSeries[k]
if !ok {
ts = &remote.TimeSeries{
ts = &prompb.TimeSeries{
Labels: tagsToLabelPairs(s.Name, s.Tags),
}
labelsToSeries[k] = ts
@ -214,8 +214,8 @@ func concatLabels(labels map[string]string) string {
return strings.Join(pairs, separator)
}
func tagsToLabelPairs(name string, tags map[string]string) []*remote.LabelPair {
pairs := make([]*remote.LabelPair, 0, len(tags))
func tagsToLabelPairs(name string, tags map[string]string) []*prompb.Label {
pairs := make([]*prompb.Label, 0, len(tags))
for k, v := range tags {
if v == "" {
// If we select metrics with different sets of labels names,
@ -226,20 +226,20 @@ func tagsToLabelPairs(name string, tags map[string]string) []*remote.LabelPair {
// to make the result correct.
continue
}
pairs = append(pairs, &remote.LabelPair{
pairs = append(pairs, &prompb.Label{
Name: k,
Value: v,
})
}
pairs = append(pairs, &remote.LabelPair{
pairs = append(pairs, &prompb.Label{
Name: model.MetricNameLabel,
Value: name,
})
return pairs
}
func valuesToSamples(values [][]interface{}) ([]*remote.Sample, error) {
samples := make([]*remote.Sample, 0, len(values))
func valuesToSamples(values [][]interface{}) ([]*prompb.Sample, error) {
samples := make([]*prompb.Sample, 0, len(values))
for _, v := range values {
if len(v) != 2 {
return nil, fmt.Errorf("bad sample tuple length, expected [<timestamp>, <value>], got %v", v)
@ -265,8 +265,8 @@ func valuesToSamples(values [][]interface{}) ([]*remote.Sample, error) {
return nil, fmt.Errorf("unable to convert sample value to float64: %v", err)
}
samples = append(samples, &remote.Sample{
TimestampMs: timestamp,
samples = append(samples, &prompb.Sample{
Timestamp: timestamp,
Value: value,
})
}
@ -275,14 +275,14 @@ func valuesToSamples(values [][]interface{}) ([]*remote.Sample, error) {
// mergeSamples merges two lists of sample pairs and removes duplicate
// timestamps. It assumes that both lists are sorted by timestamp.
func mergeSamples(a, b []*remote.Sample) []*remote.Sample {
result := make([]*remote.Sample, 0, len(a)+len(b))
func mergeSamples(a, b []*prompb.Sample) []*prompb.Sample {
result := make([]*prompb.Sample, 0, len(a)+len(b))
i, j := 0, 0
for i < len(a) && j < len(b) {
if a[i].TimestampMs < b[j].TimestampMs {
if a[i].Timestamp < b[j].Timestamp {
result = append(result, a[i])
i++
} else if a[i].TimestampMs > b[j].TimestampMs {
} else if a[i].Timestamp > b[j].Timestamp {
result = append(result, b[j])
j++
} else {

14
documentation/examples/remote_storage/remote_storage_adapter/main.go
View File

@ -36,7 +36,7 @@ import (
"github.com/prometheus/prometheus/documentation/examples/remote_storage/remote_storage_adapter/graphite"
"github.com/prometheus/prometheus/documentation/examples/remote_storage/remote_storage_adapter/influxdb"
"github.com/prometheus/prometheus/documentation/examples/remote_storage/remote_storage_adapter/opentsdb"
"github.com/prometheus/prometheus/storage/remote"
"github.com/prometheus/prometheus/prompb"
)
type config struct {
@ -146,7 +146,7 @@ type writer interface {
}
type reader interface {
Read(req *remote.ReadRequest) (*remote.ReadResponse, error)
Read(req *prompb.ReadRequest) (*prompb.ReadResponse, error)
Name() string
}
@ -196,7 +196,7 @@ func serve(addr string, writers []writer, readers []reader) error {
return
}
var req remote.WriteRequest
var req prompb.WriteRequest
if err := proto.Unmarshal(reqBuf, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
@ -229,7 +229,7 @@ func serve(addr string, writers []writer, readers []reader) error {
return
}
var req remote.ReadRequest
var req prompb.ReadRequest
if err := proto.Unmarshal(reqBuf, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
@ -242,7 +242,7 @@ func serve(addr string, writers []writer, readers []reader) error {
}
reader := readers[0]
var resp *remote.ReadResponse
var resp *prompb.ReadResponse
resp, err = reader.Read(&req)
if err != nil {
log.With("query", req).With("storage", reader.Name()).With("err", err).Warnf("Error executing query")
@ -269,7 +269,7 @@ func serve(addr string, writers []writer, readers []reader) error {
return http.ListenAndServe(addr, nil)
}
func protoToSamples(req *remote.WriteRequest) model.Samples {
func protoToSamples(req *prompb.WriteRequest) model.Samples {
var samples model.Samples
for _, ts := range req.Timeseries {
metric := make(model.Metric, len(ts.Labels))
@ -281,7 +281,7 @@ func protoToSamples(req *remote.WriteRequest) model.Samples {
samples = append(samples, &model.Sample{
Metric: metric,
Value: model.SampleValue(s.Value),
Timestamp: model.Time(s.TimestampMs),
Timestamp: model.Time(s.Timestamp),
})
}
}