diff --git a/docs/influxdb.md b/docs/influxdb.md index ee5ab9783..5b53bda8e 100644 --- a/docs/influxdb.md +++ b/docs/influxdb.md @@ -10,6 +10,32 @@ Trickster uses InfluxDB-provided packages to parse and normalize queries for cac Trickster supports integrations with InfluxDB 1.x and 2.0. +### Prometheus Remote Read + +For InfluxDB 1.x, Trickster accelerates Prometheus remote-read requests sent to +`POST /api/v1/prom/read`. Point Prometheus at the same endpoint on Trickster and +retain the InfluxDB query parameters, for example: + +```yaml +remote_read: + - url: http://trickster:8480/api/v1/prom/read?db=metrics&rp=autogen +``` + +The InfluxDB endpoint supports one query per request and the Prometheus +`SAMPLES` response type. Trickster delta-caches requests with that shape. Other +request shapes continue through the normal proxy path so that InfluxDB remains +responsible for its response and error behavior. + +Raw remote-read samples do not advertise a guaranteed interval, so point-count +sharding cannot split their extents without risking gaps. Requests use the +normal proxy path when `shard_max_size_points` is enabled; time-based sharding +remains supported. Point-based backfill tolerance uses a positive +`hints.step_ms`; a request without that hint also uses the normal proxy path. + +The `db`, `rp`, `u`, and `p` query parameters and the `Authorization` header are +part of the cache identity. They are forwarded unchanged unless a path-level +request rewrite or header configuration overrides them. + ### A note on Flux Language Support: Trickster supports the Flux Query Language for general/basic usage. Trickster does not support advanced union-style queries (e.g., with multiple `from` clauses). In this rare use case, these responses will currently provide invalid data, however, a subsequent beta will proxy unsupported requests. diff --git a/go.mod b/go.mod index dd7563a50..5d6e5e2bb 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 github.com/dgraph-io/badger/v4 v4.9.5 github.com/dgraph-io/ristretto/v2 v2.4.2 + github.com/golang/snappy v0.0.4 github.com/influxdata/influxdb v1.12.4 github.com/influxdata/influxql v1.4.1 github.com/klauspost/compress v1.19.1 diff --git a/go.sum b/go.sum index 1a0ded693..5c79c76b0 100644 --- a/go.sum +++ b/go.sum @@ -289,6 +289,8 @@ github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= +github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/golangci/asciicheck v0.5.0 h1:jczN/BorERZwK8oiFBOGvlGPknhvq0bjnysTj4nUfo0= github.com/golangci/asciicheck v0.5.0/go.mod h1:5RMNAInbNFw2krqN6ibBxN/zfRFa9S6tA1nPdM0l8qQ= github.com/golangci/dupl v0.0.0-20250308024227-f665c8d69b32 h1:WUvBfQL6EW/40l6OmeSBYQJNSif4O11+bmWEz+C7FYw= diff --git a/integration/go.mod b/integration/go.mod index 88a78f270..758f77bd5 100644 --- a/integration/go.mod +++ b/integration/go.mod @@ -34,6 +34,7 @@ require ( github.com/go-logr/logr v1.4.4 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/golang-jwt/jwt/v5 v5.3.1 // indirect + github.com/golang/snappy v0.0.4 // indirect github.com/google/flatbuffers v25.12.19+incompatible // indirect github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect diff --git a/integration/go.sum b/integration/go.sum index fea6f51e4..5118d454e 100644 --- a/integration/go.sum +++ b/integration/go.sum @@ -143,6 +143,8 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= +github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/flatbuffers v25.12.19+incompatible h1:haMV2JRRJCe1998HeW/p0X9UaMTK6SDo0ffLn2+DbLs= diff --git a/pkg/backends/influxdb/handler_promremote_test.go b/pkg/backends/influxdb/handler_promremote_test.go new file mode 100644 index 000000000..241f131c7 --- /dev/null +++ b/pkg/backends/influxdb/handler_promremote_test.go @@ -0,0 +1,267 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package influxdb + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "strconv" + "sync" + "testing" + "time" + + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote/prompb" + "github.com/trickstercache/trickster/v2/pkg/backends/providers" + "github.com/trickstercache/trickster/v2/pkg/proxy/headers" + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + tu "github.com/trickstercache/trickster/v2/pkg/testutil" + + "github.com/golang/snappy" + "google.golang.org/protobuf/proto" +) + +type promReadTransport struct { + mu sync.Mutex + extents [][2]int64 +} + +func (p *promReadTransport) RoundTrip(r *http.Request) (*http.Response, error) { + body, err := io.ReadAll(r.Body) + if err != nil { + return nil, err + } + decoded, err := snappy.Decode(nil, body) + if err != nil { + return nil, err + } + readRequest := &prompb.ReadRequest{} + if err := proto.Unmarshal(decoded, readRequest); err != nil { + return nil, err + } + if len(readRequest.Queries) != 1 { + return &http.Response{ + Status: "400 Bad Request", + StatusCode: http.StatusBadRequest, + Header: make(http.Header), + Body: io.NopCloser(bytes.NewBufferString("one query required")), + Request: r, + }, nil + } + query := readRequest.Queries[0] + p.mu.Lock() + p.extents = append(p.extents, [2]int64{query.StartTimestampMs, query.EndTimestampMs}) + p.mu.Unlock() + + // InfluxDB's storage ReadFilter treats the end timestamp as exclusive. + samples := make([]*prompb.Sample, 0, query.EndTimestampMs-query.StartTimestampMs) + for timestamp := query.StartTimestampMs; timestamp < query.EndTimestampMs; timestamp++ { + samples = append(samples, &prompb.Sample{ + Timestamp: timestamp, + Value: float64(timestamp), + }) + } + readResponse := &prompb.ReadResponse{Results: []*prompb.QueryResult{{ + Timeseries: []*prompb.TimeSeries{{ + Labels: []*prompb.Label{ + {Name: "__name__", Value: "requests_total"}, + {Name: "job", Value: "api"}, + }, + Samples: samples, + }}, + }}} + encoded, err := proto.Marshal(readResponse) + if err != nil { + return nil, err + } + compressed := snappy.Encode(nil, encoded) + header := make(http.Header) + header.Set(headers.NameContentType, promremote.ContentType) + header.Set(headers.NameContentEncoding, promremote.ContentEncoding) + header.Set(headers.NameContentLength, strconv.Itoa(len(compressed))) + return &http.Response{ + Status: "200 OK", + StatusCode: http.StatusOK, + Header: header, + Body: io.NopCloser(bytes.NewReader(compressed)), + ContentLength: int64(len(compressed)), + Request: r, + }, nil +} + +func (p *promReadTransport) requestedExtents() [][2]int64 { + p.mu.Lock() + defer p.mu.Unlock() + out := make([][2]int64, len(p.extents)) + copy(out, p.extents) + return out +} + +func encodePromReadRequest(t *testing.T, start, end int64, queries int) []byte { + t.Helper() + readRequest := &prompb.ReadRequest{AcceptedResponseTypes: []prompb.ReadRequest_ResponseType{ + prompb.ReadRequest_SAMPLES, + }} + for range queries { + readRequest.Queries = append(readRequest.Queries, &prompb.Query{ + StartTimestampMs: start, + EndTimestampMs: end, + Matchers: []*prompb.LabelMatcher{{ + Type: prompb.LabelMatcher_EQ, Name: "__name__", Value: "requests_total", + }}, + Hints: &prompb.ReadHints{StepMs: 15_000}, + }) + } + encoded, err := proto.Marshal(readRequest) + if err != nil { + t.Fatal(err) + } + return snappy.Encode(nil, encoded) +} + +func decodePromReadResponse(t *testing.T, body []byte) *prompb.ReadResponse { + t.Helper() + decoded, err := snappy.Decode(nil, body) + if err != nil { + t.Fatal(err) + } + response := &prompb.ReadResponse{} + if err := proto.Unmarshal(decoded, response); err != nil { + t.Fatal(err) + } + return response +} + +func newPromReadHandlerTest(t *testing.T) (*Client, *request.Resources, *promReadTransport) { + t.Helper() + prototype, err := NewClient("test", nil, nil, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + ts, _, seed, _, err := tu.NewTestInstance("", prototype.DefaultPathConfigs, + http.StatusOK, "", nil, providers.InfluxDB, promremote.Path, "error") + if err != nil { + t.Fatal(err) + } + rsc := request.GetResources(seed) + clientBackend, err := NewClient("test", rsc.BackendOptions, nil, rsc.CacheClient, nil, nil) + if err != nil { + t.Fatal(err) + } + client := clientBackend.(*Client) + transport := &promReadTransport{} + client.HTTPClient().Transport = transport + rsc.BackendClient = client + rsc.BackendOptions.HTTPClient = client.HTTPClient() + + t.Cleanup(func() { + _ = rsc.CacheClient.Close() + ts.CloseClientConnections() + ts.Close() + }) + return client, rsc, transport +} + +func runPromReadHandler(t *testing.T, client *Client, rsc *request.Resources, + body []byte, +) *httptest.ResponseRecorder { + t.Helper() + r := httptest.NewRequest(http.MethodPost, + "http://trickster.local"+promremote.Path+"?db=metrics&rp=autogen", bytes.NewReader(body)) + r.Header.Set(headers.NameContentType, promremote.ContentType) + r.Header.Set(headers.NameContentEncoding, promremote.ContentEncoding) + r = request.SetResources(r, rsc.Clone()) + w := httptest.NewRecorder() + client.QueryHandler(w, r) + return w +} + +func TestPromRemoteReadDeltaCache(t *testing.T) { + client, rsc, transport := newPromReadHandlerTest(t) + start := time.Now().Add(-time.Minute).Truncate(time.Millisecond).UnixMilli() + + first := runPromReadHandler(t, client, rsc, encodePromReadRequest(t, start, start+3, 1)) + if first.Code != http.StatusOK { + t.Fatalf("first status = %d, body = %s", first.Code, first.Body.String()) + } + if got := decodePromReadResponse(t, first.Body.Bytes()).Results[0].Timeseries[0].Samples; len(got) != 3 { + t.Fatalf("first sample count = %d", len(got)) + } + + second := runPromReadHandler(t, client, rsc, encodePromReadRequest(t, start, start+4, 1)) + if second.Code != http.StatusOK { + t.Fatalf("second status = %d, body = %s", second.Code, second.Body.String()) + } + if second.Header().Get(headers.NameContentType) != promremote.ContentType || + second.Header().Get(headers.NameContentEncoding) != promremote.ContentEncoding { + t.Fatalf("second headers = %#v", second.Header()) + } + if got := decodePromReadResponse(t, second.Body.Bytes()).Results[0].Timeseries[0].Samples; len(got) != 4 { + t.Fatalf("second sample count = %d", len(got)) + } + if got := second.Header().Get(headers.NameContentLength); got != "" { + t.Fatalf("second content length = %s", got) + } + + third := runPromReadHandler(t, client, rsc, encodePromReadRequest(t, start, start+4, 1)) + if third.Code != http.StatusOK { + t.Fatalf("third status = %d", third.Code) + } + if got := decodePromReadResponse(t, third.Body.Bytes()).Results[0].Timeseries[0].Samples; len(got) != 4 { + t.Fatalf("third sample count = %d", len(got)) + } + + extents := transport.requestedExtents() + want := [][2]int64{{start, start + 3}, {start + 3, start + 4}} + if len(extents) != len(want) || extents[0] != want[0] || extents[1] != want[1] { + t.Fatalf("origin extents = %#v, want %#v", extents, want) + } +} + +func TestPromRemoteReadUnsupportedShapePassesThrough(t *testing.T) { + client, rsc, transport := newPromReadHandlerTest(t) + start := time.Now().Add(-time.Minute).Truncate(time.Millisecond).UnixMilli() + w := runPromReadHandler(t, client, rsc, encodePromReadRequest(t, start, start+1, 2)) + if w.Code != http.StatusBadRequest || w.Body.String() != "one query required" { + t.Fatalf("status = %d, body = %q", w.Code, w.Body.String()) + } + if len(transport.requestedExtents()) != 0 { + t.Fatal("unsupported request was parsed as a cacheable query") + } +} + +func TestPromRemoteReadPointShardingPassesThrough(t *testing.T) { + client, rsc, transport := newPromReadHandlerTest(t) + rsc.BackendOptions.MaxShardSizePoints = 1 + rsc.BackendOptions.DoesShard = true + start := time.Now().Add(-time.Minute).Truncate(time.Millisecond).UnixMilli() + body := encodePromReadRequest(t, start, start+2, 1) + + for range 2 { + w := runPromReadHandler(t, client, rsc, body) + if w.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", w.Code, w.Body.String()) + } + } + extents := transport.requestedExtents() + want := [2]int64{start, start + 2} + if len(extents) != 2 || extents[0] != want || extents[1] != want { + t.Fatalf("origin extents = %#v, want two passthrough requests for %#v", extents, want) + } +} diff --git a/pkg/backends/influxdb/handler_query.go b/pkg/backends/influxdb/handler_query.go index ef6132bd8..7dd764f6d 100644 --- a/pkg/backends/influxdb/handler_query.go +++ b/pkg/backends/influxdb/handler_query.go @@ -23,6 +23,7 @@ import ( "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/flux" "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/influxql" "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/iofmt" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote" "github.com/trickstercache/trickster/v2/pkg/errors" "github.com/trickstercache/trickster/v2/pkg/proxy/engines" "github.com/trickstercache/trickster/v2/pkg/proxy/params" @@ -58,6 +59,9 @@ func (c *Client) QueryHandler(w http.ResponseWriter, r *http.Request) { func (c *Client) ParseTimeRangeQuery(r *http.Request) (*timeseries.TimeRangeQuery, *timeseries.RequestOptions, bool, error, ) { + if promremote.IsRequest(r) { + return promremote.ParseTimeRangeQuery(r) + } f := iofmt.Detect(r) switch { case f.IsInfluxQL(): diff --git a/pkg/backends/influxdb/iofmt/iofmt.go b/pkg/backends/influxdb/iofmt/iofmt.go index 9123e82e8..6070e5dfd 100644 --- a/pkg/backends/influxdb/iofmt/iofmt.go +++ b/pkg/backends/influxdb/iofmt/iofmt.go @@ -35,6 +35,7 @@ const ( isFlux // 4 isFluxInputJSON // 8 isFluxOutputJSON // 16 + isPromRemoteRead // 32 InfluxqlGet = isInfluxql InfluxqlPost = isInfluxql + isInfluxqlPost @@ -44,6 +45,8 @@ const ( FluxRawJSON = isFlux + isFluxOutputJSON FluxRawCsv = isFlux + + PromRemoteRead = isPromRemoteRead ) var ErrSupportedQueryLanguage = errors.New("unsupported query language") @@ -64,6 +67,10 @@ func (f Format) IsFluxOutputJSON() bool { return f&isFluxOutputJSON == isFluxOutputJSON } +func (f Format) IsPromRemoteRead() bool { + return f&isPromRemoteRead == isPromRemoteRead +} + func (f Format) IsPost() bool { return f&isFlux == isFlux || f&isInfluxqlPost == isInfluxqlPost } diff --git a/pkg/backends/influxdb/iofmt/iofmt_test.go b/pkg/backends/influxdb/iofmt/iofmt_test.go index 24dfb9a45..82caa91a7 100644 --- a/pkg/backends/influxdb/iofmt/iofmt_test.go +++ b/pkg/backends/influxdb/iofmt/iofmt_test.go @@ -65,6 +65,15 @@ func TestIsFluxOutputJSON(t *testing.T) { } } +func TestIsPromRemoteRead(t *testing.T) { + if !PromRemoteRead.IsPromRemoteRead() { + t.Fatal("expected remote-read format") + } + if FluxRawCsv.IsPromRemoteRead() { + t.Fatal("flux must not be detected as remote read") + } +} + func TestDetect(t *testing.T) { emptyHeader := make(http.Header) fluxHeader := http.Header{ diff --git a/pkg/backends/influxdb/promremote/prompb/remote_read.pb.go b/pkg/backends/influxdb/promremote/prompb/remote_read.pb.go new file mode 100644 index 000000000..2b3033ffc --- /dev/null +++ b/pkg/backends/influxdb/promremote/prompb/remote_read.pb.go @@ -0,0 +1,769 @@ +// Copyright 2016 Prometheus Team +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v7.34.1 +// source: pkg/backends/influxdb/promremote/prompb/remote_read.proto + +package prompb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type ReadRequest_ResponseType int32 + +const ( + ReadRequest_SAMPLES ReadRequest_ResponseType = 0 + ReadRequest_STREAMED_XOR_CHUNKS ReadRequest_ResponseType = 1 +) + +// Enum value maps for ReadRequest_ResponseType. +var ( + ReadRequest_ResponseType_name = map[int32]string{ + 0: "SAMPLES", + 1: "STREAMED_XOR_CHUNKS", + } + ReadRequest_ResponseType_value = map[string]int32{ + "SAMPLES": 0, + "STREAMED_XOR_CHUNKS": 1, + } +) + +func (x ReadRequest_ResponseType) Enum() *ReadRequest_ResponseType { + p := new(ReadRequest_ResponseType) + *p = x + return p +} + +func (x ReadRequest_ResponseType) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (ReadRequest_ResponseType) Descriptor() protoreflect.EnumDescriptor { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_enumTypes[0].Descriptor() +} + +func (ReadRequest_ResponseType) Type() protoreflect.EnumType { + return &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_enumTypes[0] +} + +func (x ReadRequest_ResponseType) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use ReadRequest_ResponseType.Descriptor instead. +func (ReadRequest_ResponseType) EnumDescriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{0, 0} +} + +type LabelMatcher_Type int32 + +const ( + LabelMatcher_EQ LabelMatcher_Type = 0 + LabelMatcher_NEQ LabelMatcher_Type = 1 + LabelMatcher_RE LabelMatcher_Type = 2 + LabelMatcher_NRE LabelMatcher_Type = 3 +) + +// Enum value maps for LabelMatcher_Type. +var ( + LabelMatcher_Type_name = map[int32]string{ + 0: "EQ", + 1: "NEQ", + 2: "RE", + 3: "NRE", + } + LabelMatcher_Type_value = map[string]int32{ + "EQ": 0, + "NEQ": 1, + "RE": 2, + "NRE": 3, + } +) + +func (x LabelMatcher_Type) Enum() *LabelMatcher_Type { + p := new(LabelMatcher_Type) + *p = x + return p +} + +func (x LabelMatcher_Type) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (LabelMatcher_Type) Descriptor() protoreflect.EnumDescriptor { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_enumTypes[1].Descriptor() +} + +func (LabelMatcher_Type) Type() protoreflect.EnumType { + return &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_enumTypes[1] +} + +func (x LabelMatcher_Type) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use LabelMatcher_Type.Descriptor instead. +func (LabelMatcher_Type) EnumDescriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{7, 0} +} + +type ReadRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Queries []*Query `protobuf:"bytes,1,rep,name=queries,proto3" json:"queries,omitempty"` + AcceptedResponseTypes []ReadRequest_ResponseType `protobuf:"varint,2,rep,packed,name=accepted_response_types,json=acceptedResponseTypes,proto3,enum=prometheus.ReadRequest_ResponseType" json:"accepted_response_types,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ReadRequest) Reset() { + *x = ReadRequest{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ReadRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ReadRequest) ProtoMessage() {} + +func (x *ReadRequest) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ReadRequest.ProtoReflect.Descriptor instead. +func (*ReadRequest) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{0} +} + +func (x *ReadRequest) GetQueries() []*Query { + if x != nil { + return x.Queries + } + return nil +} + +func (x *ReadRequest) GetAcceptedResponseTypes() []ReadRequest_ResponseType { + if x != nil { + return x.AcceptedResponseTypes + } + return nil +} + +type ReadResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Results []*QueryResult `protobuf:"bytes,1,rep,name=results,proto3" json:"results,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ReadResponse) Reset() { + *x = ReadResponse{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ReadResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ReadResponse) ProtoMessage() {} + +func (x *ReadResponse) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ReadResponse.ProtoReflect.Descriptor instead. +func (*ReadResponse) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{1} +} + +func (x *ReadResponse) GetResults() []*QueryResult { + if x != nil { + return x.Results + } + return nil +} + +type Query struct { + state protoimpl.MessageState `protogen:"open.v1"` + StartTimestampMs int64 `protobuf:"varint,1,opt,name=start_timestamp_ms,json=startTimestampMs,proto3" json:"start_timestamp_ms,omitempty"` + EndTimestampMs int64 `protobuf:"varint,2,opt,name=end_timestamp_ms,json=endTimestampMs,proto3" json:"end_timestamp_ms,omitempty"` + Matchers []*LabelMatcher `protobuf:"bytes,3,rep,name=matchers,proto3" json:"matchers,omitempty"` + Hints *ReadHints `protobuf:"bytes,4,opt,name=hints,proto3" json:"hints,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Query) Reset() { + *x = Query{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Query) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Query) ProtoMessage() {} + +func (x *Query) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Query.ProtoReflect.Descriptor instead. +func (*Query) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{2} +} + +func (x *Query) GetStartTimestampMs() int64 { + if x != nil { + return x.StartTimestampMs + } + return 0 +} + +func (x *Query) GetEndTimestampMs() int64 { + if x != nil { + return x.EndTimestampMs + } + return 0 +} + +func (x *Query) GetMatchers() []*LabelMatcher { + if x != nil { + return x.Matchers + } + return nil +} + +func (x *Query) GetHints() *ReadHints { + if x != nil { + return x.Hints + } + return nil +} + +type QueryResult struct { + state protoimpl.MessageState `protogen:"open.v1"` + Timeseries []*TimeSeries `protobuf:"bytes,1,rep,name=timeseries,proto3" json:"timeseries,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *QueryResult) Reset() { + *x = QueryResult{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *QueryResult) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*QueryResult) ProtoMessage() {} + +func (x *QueryResult) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use QueryResult.ProtoReflect.Descriptor instead. +func (*QueryResult) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{3} +} + +func (x *QueryResult) GetTimeseries() []*TimeSeries { + if x != nil { + return x.Timeseries + } + return nil +} + +type Sample struct { + state protoimpl.MessageState `protogen:"open.v1"` + Value float64 `protobuf:"fixed64,1,opt,name=value,proto3" json:"value,omitempty"` + Timestamp int64 `protobuf:"varint,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Sample) Reset() { + *x = Sample{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Sample) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Sample) ProtoMessage() {} + +func (x *Sample) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Sample.ProtoReflect.Descriptor instead. +func (*Sample) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{4} +} + +func (x *Sample) GetValue() float64 { + if x != nil { + return x.Value + } + return 0 +} + +func (x *Sample) GetTimestamp() int64 { + if x != nil { + return x.Timestamp + } + return 0 +} + +type TimeSeries struct { + state protoimpl.MessageState `protogen:"open.v1"` + Labels []*Label `protobuf:"bytes,1,rep,name=labels,proto3" json:"labels,omitempty"` + Samples []*Sample `protobuf:"bytes,2,rep,name=samples,proto3" json:"samples,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TimeSeries) Reset() { + *x = TimeSeries{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TimeSeries) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TimeSeries) ProtoMessage() {} + +func (x *TimeSeries) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TimeSeries.ProtoReflect.Descriptor instead. +func (*TimeSeries) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{5} +} + +func (x *TimeSeries) GetLabels() []*Label { + if x != nil { + return x.Labels + } + return nil +} + +func (x *TimeSeries) GetSamples() []*Sample { + if x != nil { + return x.Samples + } + return nil +} + +type Label struct { + state protoimpl.MessageState `protogen:"open.v1"` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Label) Reset() { + *x = Label{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Label) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Label) ProtoMessage() {} + +func (x *Label) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Label.ProtoReflect.Descriptor instead. +func (*Label) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{6} +} + +func (x *Label) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *Label) GetValue() string { + if x != nil { + return x.Value + } + return "" +} + +type LabelMatcher struct { + state protoimpl.MessageState `protogen:"open.v1"` + Type LabelMatcher_Type `protobuf:"varint,1,opt,name=type,proto3,enum=prometheus.LabelMatcher_Type" json:"type,omitempty"` + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + Value string `protobuf:"bytes,3,opt,name=value,proto3" json:"value,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LabelMatcher) Reset() { + *x = LabelMatcher{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LabelMatcher) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LabelMatcher) ProtoMessage() {} + +func (x *LabelMatcher) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LabelMatcher.ProtoReflect.Descriptor instead. +func (*LabelMatcher) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{7} +} + +func (x *LabelMatcher) GetType() LabelMatcher_Type { + if x != nil { + return x.Type + } + return LabelMatcher_EQ +} + +func (x *LabelMatcher) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *LabelMatcher) GetValue() string { + if x != nil { + return x.Value + } + return "" +} + +type ReadHints struct { + state protoimpl.MessageState `protogen:"open.v1"` + StepMs int64 `protobuf:"varint,1,opt,name=step_ms,json=stepMs,proto3" json:"step_ms,omitempty"` + Func string `protobuf:"bytes,2,opt,name=func,proto3" json:"func,omitempty"` + StartMs int64 `protobuf:"varint,3,opt,name=start_ms,json=startMs,proto3" json:"start_ms,omitempty"` + EndMs int64 `protobuf:"varint,4,opt,name=end_ms,json=endMs,proto3" json:"end_ms,omitempty"` + Grouping []string `protobuf:"bytes,5,rep,name=grouping,proto3" json:"grouping,omitempty"` + By bool `protobuf:"varint,6,opt,name=by,proto3" json:"by,omitempty"` + RangeMs int64 `protobuf:"varint,7,opt,name=range_ms,json=rangeMs,proto3" json:"range_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ReadHints) Reset() { + *x = ReadHints{} + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ReadHints) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ReadHints) ProtoMessage() {} + +func (x *ReadHints) ProtoReflect() protoreflect.Message { + mi := &file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ReadHints.ProtoReflect.Descriptor instead. +func (*ReadHints) Descriptor() ([]byte, []int) { + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP(), []int{8} +} + +func (x *ReadHints) GetStepMs() int64 { + if x != nil { + return x.StepMs + } + return 0 +} + +func (x *ReadHints) GetFunc() string { + if x != nil { + return x.Func + } + return "" +} + +func (x *ReadHints) GetStartMs() int64 { + if x != nil { + return x.StartMs + } + return 0 +} + +func (x *ReadHints) GetEndMs() int64 { + if x != nil { + return x.EndMs + } + return 0 +} + +func (x *ReadHints) GetGrouping() []string { + if x != nil { + return x.Grouping + } + return nil +} + +func (x *ReadHints) GetBy() bool { + if x != nil { + return x.By + } + return false +} + +func (x *ReadHints) GetRangeMs() int64 { + if x != nil { + return x.RangeMs + } + return 0 +} + +var File_pkg_backends_influxdb_promremote_prompb_remote_read_proto protoreflect.FileDescriptor + +const file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDesc = "" + + "\n" + + "9pkg/backends/influxdb/promremote/prompb/remote_read.proto\x12\n" + + "prometheus\"\xce\x01\n" + + "\vReadRequest\x12+\n" + + "\aqueries\x18\x01 \x03(\v2\x11.prometheus.QueryR\aqueries\x12\\\n" + + "\x17accepted_response_types\x18\x02 \x03(\x0e2$.prometheus.ReadRequest.ResponseTypeR\x15acceptedResponseTypes\"4\n" + + "\fResponseType\x12\v\n" + + "\aSAMPLES\x10\x00\x12\x17\n" + + "\x13STREAMED_XOR_CHUNKS\x10\x01\"A\n" + + "\fReadResponse\x121\n" + + "\aresults\x18\x01 \x03(\v2\x17.prometheus.QueryResultR\aresults\"\xc2\x01\n" + + "\x05Query\x12,\n" + + "\x12start_timestamp_ms\x18\x01 \x01(\x03R\x10startTimestampMs\x12(\n" + + "\x10end_timestamp_ms\x18\x02 \x01(\x03R\x0eendTimestampMs\x124\n" + + "\bmatchers\x18\x03 \x03(\v2\x18.prometheus.LabelMatcherR\bmatchers\x12+\n" + + "\x05hints\x18\x04 \x01(\v2\x15.prometheus.ReadHintsR\x05hints\"E\n" + + "\vQueryResult\x126\n" + + "\n" + + "timeseries\x18\x01 \x03(\v2\x16.prometheus.TimeSeriesR\n" + + "timeseries\"<\n" + + "\x06Sample\x12\x14\n" + + "\x05value\x18\x01 \x01(\x01R\x05value\x12\x1c\n" + + "\ttimestamp\x18\x02 \x01(\x03R\ttimestamp\"e\n" + + "\n" + + "TimeSeries\x12)\n" + + "\x06labels\x18\x01 \x03(\v2\x11.prometheus.LabelR\x06labels\x12,\n" + + "\asamples\x18\x02 \x03(\v2\x12.prometheus.SampleR\asamples\"1\n" + + "\x05Label\x12\x12\n" + + "\x04name\x18\x01 \x01(\tR\x04name\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value\"\x95\x01\n" + + "\fLabelMatcher\x121\n" + + "\x04type\x18\x01 \x01(\x0e2\x1d.prometheus.LabelMatcher.TypeR\x04type\x12\x12\n" + + "\x04name\x18\x02 \x01(\tR\x04name\x12\x14\n" + + "\x05value\x18\x03 \x01(\tR\x05value\"(\n" + + "\x04Type\x12\x06\n" + + "\x02EQ\x10\x00\x12\a\n" + + "\x03NEQ\x10\x01\x12\x06\n" + + "\x02RE\x10\x02\x12\a\n" + + "\x03NRE\x10\x03\"\xb1\x01\n" + + "\tReadHints\x12\x17\n" + + "\astep_ms\x18\x01 \x01(\x03R\x06stepMs\x12\x12\n" + + "\x04func\x18\x02 \x01(\tR\x04func\x12\x19\n" + + "\bstart_ms\x18\x03 \x01(\x03R\astartMs\x12\x15\n" + + "\x06end_ms\x18\x04 \x01(\x03R\x05endMs\x12\x1a\n" + + "\bgrouping\x18\x05 \x03(\tR\bgrouping\x12\x0e\n" + + "\x02by\x18\x06 \x01(\bR\x02by\x12\x19\n" + + "\brange_ms\x18\a \x01(\x03R\arangeMsBPZNgithub.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote/prompbb\x06proto3" + +var ( + file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescOnce sync.Once + file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescData []byte +) + +func file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescGZIP() []byte { + file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescOnce.Do(func() { + file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDesc), len(file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDesc))) + }) + return file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDescData +} + +var file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_enumTypes = make([]protoimpl.EnumInfo, 2) +var file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes = make([]protoimpl.MessageInfo, 9) +var file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_goTypes = []any{ + (ReadRequest_ResponseType)(0), // 0: prometheus.ReadRequest.ResponseType + (LabelMatcher_Type)(0), // 1: prometheus.LabelMatcher.Type + (*ReadRequest)(nil), // 2: prometheus.ReadRequest + (*ReadResponse)(nil), // 3: prometheus.ReadResponse + (*Query)(nil), // 4: prometheus.Query + (*QueryResult)(nil), // 5: prometheus.QueryResult + (*Sample)(nil), // 6: prometheus.Sample + (*TimeSeries)(nil), // 7: prometheus.TimeSeries + (*Label)(nil), // 8: prometheus.Label + (*LabelMatcher)(nil), // 9: prometheus.LabelMatcher + (*ReadHints)(nil), // 10: prometheus.ReadHints +} +var file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_depIdxs = []int32{ + 4, // 0: prometheus.ReadRequest.queries:type_name -> prometheus.Query + 0, // 1: prometheus.ReadRequest.accepted_response_types:type_name -> prometheus.ReadRequest.ResponseType + 5, // 2: prometheus.ReadResponse.results:type_name -> prometheus.QueryResult + 9, // 3: prometheus.Query.matchers:type_name -> prometheus.LabelMatcher + 10, // 4: prometheus.Query.hints:type_name -> prometheus.ReadHints + 7, // 5: prometheus.QueryResult.timeseries:type_name -> prometheus.TimeSeries + 8, // 6: prometheus.TimeSeries.labels:type_name -> prometheus.Label + 6, // 7: prometheus.TimeSeries.samples:type_name -> prometheus.Sample + 1, // 8: prometheus.LabelMatcher.type:type_name -> prometheus.LabelMatcher.Type + 9, // [9:9] is the sub-list for method output_type + 9, // [9:9] is the sub-list for method input_type + 9, // [9:9] is the sub-list for extension type_name + 9, // [9:9] is the sub-list for extension extendee + 0, // [0:9] is the sub-list for field type_name +} + +func init() { file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_init() } +func file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_init() { + if File_pkg_backends_influxdb_promremote_prompb_remote_read_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDesc), len(file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_rawDesc)), + NumEnums: 2, + NumMessages: 9, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_goTypes, + DependencyIndexes: file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_depIdxs, + EnumInfos: file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_enumTypes, + MessageInfos: file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_msgTypes, + }.Build() + File_pkg_backends_influxdb_promremote_prompb_remote_read_proto = out.File + file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_goTypes = nil + file_pkg_backends_influxdb_promremote_prompb_remote_read_proto_depIdxs = nil +} diff --git a/pkg/backends/influxdb/promremote/prompb/remote_read.proto b/pkg/backends/influxdb/promremote/prompb/remote_read.proto new file mode 100644 index 000000000..0c59614b4 --- /dev/null +++ b/pkg/backends/influxdb/promremote/prompb/remote_read.proto @@ -0,0 +1,86 @@ +// Copyright 2016 Prometheus Team +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package prometheus; + +option go_package = "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote/prompb"; + +// This is the SAMPLES subset of Prometheus' remote-read protocol used by +// InfluxDB 1.x. Field numbers and enum values are wire-compatible with the +// upstream Prometheus schema. + +message ReadRequest { + repeated Query queries = 1; + + enum ResponseType { + SAMPLES = 0; + STREAMED_XOR_CHUNKS = 1; + } + + repeated ResponseType accepted_response_types = 2; +} + +message ReadResponse { + repeated QueryResult results = 1; +} + +message Query { + int64 start_timestamp_ms = 1; + int64 end_timestamp_ms = 2; + repeated LabelMatcher matchers = 3; + ReadHints hints = 4; +} + +message QueryResult { + repeated TimeSeries timeseries = 1; +} + +message Sample { + double value = 1; + int64 timestamp = 2; +} + +message TimeSeries { + repeated Label labels = 1; + repeated Sample samples = 2; +} + +message Label { + string name = 1; + string value = 2; +} + +message LabelMatcher { + enum Type { + EQ = 0; + NEQ = 1; + RE = 2; + NRE = 3; + } + + Type type = 1; + string name = 2; + string value = 3; +} + +message ReadHints { + int64 step_ms = 1; + string func = 2; + int64 start_ms = 3; + int64 end_ms = 4; + repeated string grouping = 5; + bool by = 6; + int64 range_ms = 7; +} diff --git a/pkg/backends/influxdb/promremote/promremote.go b/pkg/backends/influxdb/promremote/promremote.go new file mode 100644 index 000000000..ce1496232 --- /dev/null +++ b/pkg/backends/influxdb/promremote/promremote.go @@ -0,0 +1,220 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// Package promremote models InfluxDB's Prometheus remote-read endpoint. +package promremote + +import ( + "cmp" + "encoding/json" + "errors" + "fmt" + "math" + "net/http" + "slices" + "strings" + "time" + + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/iofmt" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote/prompb" + "github.com/trickstercache/trickster/v2/pkg/proxy/headers" + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + "github.com/trickstercache/trickster/v2/pkg/timeseries" + + "github.com/golang/snappy" + "google.golang.org/protobuf/proto" +) + +const ( + // Path is InfluxDB's Prometheus remote-read endpoint. + Path = "/api/v1/prom/read" + + // ContentType is the Prometheus remote-read protobuf media type. + ContentType = "application/x-protobuf" + // ContentEncoding is the Prometheus remote-read compression scheme. + ContentEncoding = "snappy" + + cacheKeyQuery = "prom_remote_read_query" +) + +var ( + errInvalidExtent = errors.New("invalid Prometheus remote-read extent") + errMissingMatcher = errors.New("prometheus remote-read query requires at least one matcher") + errPointSharding = errors.New("prometheus remote-read cannot safely use point-count sharding") + errPointPolicyStep = errors.New("prometheus remote-read point policies require a positive query step hint") + errSingleQuery = errors.New("the InfluxDB Prometheus remote-read endpoint supports exactly one query") + errUnsupportedResult = errors.New("prometheus remote-read request does not accept sample responses") +) + +type canonicalMatcher struct { + Type prompb.LabelMatcher_Type `json:"type"` + Name string `json:"name"` + Value string `json:"value"` +} + +type parsedRequest struct { + readRequest *prompb.ReadRequest + decodeLimit int +} + +// IsRequest reports whether r targets InfluxDB's Prometheus remote-read protocol. +func IsRequest(r *http.Request) bool { + if r == nil || r.Method != http.MethodPost { + return false + } + if r.URL != nil && strings.HasSuffix(strings.TrimSuffix(r.URL.Path, "/"), Path) { + return true + } + contentType := strings.TrimSpace(strings.Split(r.Header.Get(headers.NameContentType), ";")[0]) + return strings.EqualFold(contentType, ContentType) && + strings.EqualFold(strings.TrimSpace(r.Header.Get(headers.NameContentEncoding)), ContentEncoding) +} + +// IsParsedQuery reports whether q is a parsed Prometheus remote-read request. +func IsParsedQuery(q any) bool { + parsed, ok := q.(*parsedRequest) + return ok && parsed != nil && parsed.readRequest != nil +} + +// ParseTimeRangeQuery converts a remote-read body into Trickster's time-range model. +func ParseTimeRangeQuery(r *http.Request) (*timeseries.TimeRangeQuery, + *timeseries.RequestOptions, bool, error, +) { + body, err := request.GetBody(r) + if err != nil { + return nil, nil, false, fmt.Errorf("read Prometheus remote-read request: %w", err) + } + trq := ×eries.TimeRangeQuery{OriginalBody: body, Step: time.Millisecond} + + decodeLimit := requestDecodeLimit(r) + decoded, err := decodeSnappyBlock(body, decodeLimit) + if err != nil { + return trq, nil, false, fmt.Errorf("decode Prometheus remote-read request: %w", err) + } + readRequest := &prompb.ReadRequest{} + if err := proto.Unmarshal(decoded, readRequest); err != nil { + return trq, nil, false, fmt.Errorf("unmarshal Prometheus remote-read request: %w", err) + } + if len(readRequest.Queries) != 1 || readRequest.Queries[0] == nil { + return trq, nil, false, errSingleQuery + } + if !acceptsSamples(readRequest.AcceptedResponseTypes) { + return trq, nil, false, errUnsupportedResult + } + + query := readRequest.Queries[0] + // InfluxDB's storage read range is [start, end), while Trickster models + // cached extents as inclusive. Empty ranges cannot be represented safely. + if query.EndTimestampMs <= query.StartTimestampMs { + return trq, nil, false, errInvalidExtent + } + statement, err := canonicalQuery(query) + if err != nil { + return trq, nil, false, err + } + + trq.Statement = statement + trq.Extent = timeseries.Extent{ + Start: time.UnixMilli(query.StartTimestampMs), + End: time.UnixMilli(query.EndTimestampMs - 1), + } + trq.ParsedQuery = &parsedRequest{readRequest: readRequest, decodeLimit: decodeLimit} + trq.CacheKeyElements = map[string]string{cacheKeyQuery: statement} + if query.Hints != nil && query.Hints.StepMs > 0 && + query.Hints.StepMs <= int64((1<<63-1)/time.Millisecond) { + trq.PolicyStep = time.Duration(query.Hints.StepMs) * time.Millisecond + } + if rsc := request.GetResources(r); rsc != nil && rsc.BackendOptions != nil { + if rsc.BackendOptions.MaxShardSizePoints > 0 { + return trq, nil, false, errPointSharding + } + if rsc.BackendOptions.BackfillTolerancePoints > 0 && trq.PolicyStep <= 0 { + return trq, nil, false, errPointPolicyStep + } + } + + rlo := ×eries.RequestOptions{ + OutputFormat: byte(iofmt.PromRemoteRead), + FastForwardDisable: true, + ProviderRequest: readRequest, + } + return trq, rlo, false, nil +} + +func acceptsSamples(types []prompb.ReadRequest_ResponseType) bool { + return len(types) == 0 || slices.Contains(types, prompb.ReadRequest_SAMPLES) +} + +func canonicalQuery(query *prompb.Query) (string, error) { + if len(query.Matchers) == 0 { + return "", errMissingMatcher + } + matchers := make([]canonicalMatcher, len(query.Matchers)) + for i, matcher := range query.Matchers { + if matcher == nil { + return "", errMissingMatcher + } + matchers[i] = canonicalMatcher{ + Type: matcher.Type, + Name: matcher.Name, + Value: matcher.Value, + } + } + slices.SortFunc(matchers, func(a, b canonicalMatcher) int { + if c := cmp.Compare(a.Type, b.Type); c != 0 { + return c + } + if c := cmp.Compare(a.Name, b.Name); c != 0 { + return c + } + return cmp.Compare(a.Value, b.Value) + }) + b, err := json.Marshal(matchers) + if err != nil { + return "", fmt.Errorf("marshal canonical Prometheus remote-read query: %w", err) + } + return string(b), nil +} + +// SetExtent rewrites a remote-read request body to the supplied inclusive extent. +func SetExtent(r *http.Request, trq *timeseries.TimeRangeQuery, + extent *timeseries.Extent, +) error { + if r == nil || trq == nil || extent == nil { + return errInvalidExtent + } + parsed, ok := trq.ParsedQuery.(*parsedRequest) + if !ok || parsed == nil || parsed.readRequest == nil { + return errSingleQuery + } + clone := proto.Clone(parsed.readRequest).(*prompb.ReadRequest) + if len(clone.Queries) != 1 || clone.Queries[0] == nil { + return errSingleQuery + } + end := extent.End.UnixMilli() + if end == math.MaxInt64 { + return errInvalidExtent + } + clone.Queries[0].StartTimestampMs = extent.Start.UnixMilli() + clone.Queries[0].EndTimestampMs = end + 1 + + encoded, err := proto.Marshal(clone) + if err != nil { + return fmt.Errorf("marshal Prometheus remote-read request: %w", err) + } + request.SetBody(r, snappy.Encode(nil, encoded)) + return nil +} diff --git a/pkg/backends/influxdb/promremote/promremote_test.go b/pkg/backends/influxdb/promremote/promremote_test.go new file mode 100644 index 000000000..3b110c025 --- /dev/null +++ b/pkg/backends/influxdb/promremote/promremote_test.go @@ -0,0 +1,288 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package promremote + +import ( + "bytes" + "encoding/binary" + "errors" + "io" + "net/http" + "testing" + "time" + + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/iofmt" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote/prompb" + bo "github.com/trickstercache/trickster/v2/pkg/backends/options" + "github.com/trickstercache/trickster/v2/pkg/proxy/headers" + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + "github.com/trickstercache/trickster/v2/pkg/timeseries" + + "github.com/golang/snappy" + "google.golang.org/protobuf/proto" +) + +func encodeReadRequest(t *testing.T, readRequest *prompb.ReadRequest) []byte { + t.Helper() + b, err := proto.Marshal(readRequest) + if err != nil { + t.Fatal(err) + } + return snappy.Encode(nil, b) +} + +func decodeReadRequest(t *testing.T, body []byte) *prompb.ReadRequest { + t.Helper() + b, err := snappy.Decode(nil, body) + if err != nil { + t.Fatal(err) + } + readRequest := &prompb.ReadRequest{} + if err := proto.Unmarshal(b, readRequest); err != nil { + t.Fatal(err) + } + return readRequest +} + +func newReadRequest(t *testing.T, body []byte) *http.Request { + t.Helper() + r, err := http.NewRequest(http.MethodPost, + "http://example.com/api/v1/prom/read?db=metrics&rp=autogen", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + r.Header.Set(headers.NameContentType, ContentType) + r.Header.Set(headers.NameContentEncoding, ContentEncoding) + return r +} + +func sampleReadRequest(start, end, step int64) *prompb.ReadRequest { + return &prompb.ReadRequest{ + Queries: []*prompb.Query{{ + StartTimestampMs: start, + EndTimestampMs: end, + Matchers: []*prompb.LabelMatcher{ + {Type: prompb.LabelMatcher_EQ, Name: "job", Value: "api"}, + {Type: prompb.LabelMatcher_RE, Name: "__name__", Value: "http_.+"}, + }, + Hints: &prompb.ReadHints{StepMs: step, Func: "rate"}, + }}, + AcceptedResponseTypes: []prompb.ReadRequest_ResponseType{prompb.ReadRequest_SAMPLES}, + } +} + +func TestParseTimeRangeQuery(t *testing.T) { + const ( + start = int64(1_700_000_000_001) + end = int64(1_700_000_060_001) + step = int64(15_000) + ) + body := encodeReadRequest(t, sampleReadRequest(start, end, step)) + r := newReadRequest(t, body) + + trq, rlo, canObjectCache, err := ParseTimeRangeQuery(r) + if err != nil { + t.Fatal(err) + } + if canObjectCache { + t.Fatal("remote read must not fall back to object caching") + } + if trq.Extent.Start.UnixMilli() != start || trq.Extent.End.UnixMilli() != end-1 { + t.Fatalf("extent = %d..%d", trq.Extent.Start.UnixMilli(), trq.Extent.End.UnixMilli()) + } + if trq.Step != time.Millisecond { + t.Fatalf("step = %s", trq.Step) + } + if trq.PolicyStep != 15*time.Second { + t.Fatalf("policy step = %s", trq.PolicyStep) + } + if !bytes.Equal(trq.OriginalBody, body) { + t.Fatal("original body was not preserved") + } + if !IsParsedQuery(trq.ParsedQuery) { + t.Fatalf("parsed query type = %T", trq.ParsedQuery) + } + if rlo == nil || iofmt.Format(rlo.OutputFormat) != iofmt.PromRemoteRead || + !rlo.FastForwardDisable { + t.Fatalf("request options = %#v", rlo) + } + if trq.CacheKeyElements[cacheKeyQuery] != trq.Statement || trq.Statement == "" { + t.Fatalf("canonical query = %q", trq.Statement) + } + + gotBody, err := io.ReadAll(r.Body) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(gotBody, body) { + t.Fatal("parsing did not reset the request body") + } +} + +func TestParseTimeRangeQueryCanonicalizesMatchers(t *testing.T) { + first := sampleReadRequest(100, 200, 15) + second := sampleReadRequest(300, 400, 30) + second.Queries[0].Matchers[0], second.Queries[0].Matchers[1] = + second.Queries[0].Matchers[1], second.Queries[0].Matchers[0] + + trq1, _, _, err := ParseTimeRangeQuery(newReadRequest(t, encodeReadRequest(t, first))) + if err != nil { + t.Fatal(err) + } + trq2, _, _, err := ParseTimeRangeQuery(newReadRequest(t, encodeReadRequest(t, second))) + if err != nil { + t.Fatal(err) + } + if trq1.Statement != trq2.Statement { + t.Fatalf("equivalent matchers produced different keys:\n%s\n%s", trq1.Statement, trq2.Statement) + } +} + +func TestParseTimeRangeQueryUnsupportedBodies(t *testing.T) { + tests := map[string][]byte{ + "not snappy": []byte("not-snappy"), + "no query": encodeReadRequest(t, &prompb.ReadRequest{}), + "two queries": encodeReadRequest(t, &prompb.ReadRequest{Queries: []*prompb.Query{ + {Matchers: []*prompb.LabelMatcher{{Name: "a"}}}, + {Matchers: []*prompb.LabelMatcher{{Name: "b"}}}, + }}), + "no matcher": encodeReadRequest(t, &prompb.ReadRequest{Queries: []*prompb.Query{{}}}), + "invalid extent": encodeReadRequest(t, &prompb.ReadRequest{Queries: []*prompb.Query{{ + StartTimestampMs: 2, + EndTimestampMs: 1, + Matchers: []*prompb.LabelMatcher{{Name: "a"}}, + }}}), + "empty extent": encodeReadRequest(t, &prompb.ReadRequest{Queries: []*prompb.Query{{ + StartTimestampMs: 1, + EndTimestampMs: 1, + Matchers: []*prompb.LabelMatcher{{Name: "a"}}, + }}}), + "chunks only": encodeReadRequest(t, &prompb.ReadRequest{ + Queries: []*prompb.Query{{Matchers: []*prompb.LabelMatcher{{Name: "a"}}}}, + AcceptedResponseTypes: []prompb.ReadRequest_ResponseType{ + prompb.ReadRequest_STREAMED_XOR_CHUNKS, + }, + }), + } + for name, body := range tests { + t.Run(name, func(t *testing.T) { + trq, _, _, err := ParseTimeRangeQuery(newReadRequest(t, body)) + if err == nil { + t.Fatal("expected an error") + } + if trq != nil && !bytes.Equal(trq.OriginalBody, body) { + t.Fatal("error path did not preserve the original body") + } + }) + } +} + +func TestParseTimeRangeQueryDecodeLimit(t *testing.T) { + body := binary.AppendUvarint(nil, 17) + r := newReadRequest(t, body) + r = request.SetResources(r, request.NewResources( + &bo.Options{MaxCaptureBytes: 16}, nil, nil, nil, nil, nil, + )) + + trq, _, _, err := ParseTimeRangeQuery(r) + if !errors.Is(err, errBodyTooLarge) { + t.Fatalf("error = %v", err) + } + if trq == nil || !bytes.Equal(trq.OriginalBody, body) { + t.Fatal("oversized body was not preserved for proxy fallback") + } +} + +func TestParseTimeRangeQueryPointPolicyFallbacks(t *testing.T) { + tests := map[string]struct { + options *bo.Options + step int64 + want error + }{ + "point sharding": { + options: &bo.Options{MaxShardSizePoints: 100}, + step: 15_000, + want: errPointSharding, + }, + "backfill points without hints": { + options: &bo.Options{BackfillTolerancePoints: 2}, + want: errPointPolicyStep, + }, + } + for name, test := range tests { + t.Run(name, func(t *testing.T) { + r := newReadRequest(t, encodeReadRequest(t, sampleReadRequest(100, 200, test.step))) + r = request.SetResources(r, request.NewResources( + test.options, nil, nil, nil, nil, nil, + )) + trq, _, _, err := ParseTimeRangeQuery(r) + if !errors.Is(err, test.want) { + t.Fatalf("error = %v", err) + } + if trq == nil || len(trq.OriginalBody) == 0 { + t.Fatal("fallback did not preserve the original body") + } + }) + } +} + +func TestSetExtent(t *testing.T) { + original := sampleReadRequest(100, 200, 15) + body := encodeReadRequest(t, original) + r := newReadRequest(t, body) + trq, _, _, err := ParseTimeRangeQuery(r) + if err != nil { + t.Fatal(err) + } + extent := ×eries.Extent{Start: time.UnixMilli(125), End: time.UnixMilli(175)} + if err := SetExtent(r, trq, extent); err != nil { + t.Fatal(err) + } + rewrittenBody, err := io.ReadAll(r.Body) + if err != nil { + t.Fatal(err) + } + rewritten := decodeReadRequest(t, rewrittenBody) + if rewritten.Queries[0].StartTimestampMs != 125 || rewritten.Queries[0].EndTimestampMs != 176 { + t.Fatalf("rewritten extent = %d..%d", rewritten.Queries[0].StartTimestampMs, + rewritten.Queries[0].EndTimestampMs) + } + parsed := trq.ParsedQuery.(*parsedRequest).readRequest + if parsed.Queries[0].StartTimestampMs != 100 || parsed.Queries[0].EndTimestampMs != 200 { + t.Fatal("extent rewrite mutated the parsed request") + } + if r.ContentLength != int64(len(rewrittenBody)) { + t.Fatalf("content length = %d, body = %d", r.ContentLength, len(rewrittenBody)) + } +} + +func TestIsRequest(t *testing.T) { + standard, _ := http.NewRequest(http.MethodPost, "http://example.com/base"+Path, nil) + if !IsRequest(standard) { + t.Fatal("standard path was not detected") + } + custom, _ := http.NewRequest(http.MethodPost, "http://example.com/custom", nil) + custom.Header.Set(headers.NameContentType, ContentType+"; proto=prometheus.ReadRequest") + custom.Header.Set(headers.NameContentEncoding, ContentEncoding) + if !IsRequest(custom) { + t.Fatal("protocol headers were not detected") + } + get, _ := http.NewRequest(http.MethodGet, "http://example.com"+Path, nil) + if IsRequest(get) { + t.Fatal("GET must not be detected as remote read") + } +} diff --git a/pkg/backends/influxdb/promremote/serialization.go b/pkg/backends/influxdb/promremote/serialization.go new file mode 100644 index 000000000..2557fd1e3 --- /dev/null +++ b/pkg/backends/influxdb/promremote/serialization.go @@ -0,0 +1,244 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package promremote + +import ( + "bytes" + "errors" + "fmt" + "io" + "math" + "net/http" + "slices" + "time" + + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote/prompb" + "github.com/trickstercache/trickster/v2/pkg/proxy/headers" + "github.com/trickstercache/trickster/v2/pkg/timeseries" + "github.com/trickstercache/trickster/v2/pkg/timeseries/dataset" + "github.com/trickstercache/trickster/v2/pkg/timeseries/epoch" + + "github.com/golang/snappy" + "google.golang.org/protobuf/proto" +) + +var ( + errDuplicateLabel = errors.New("prometheus remote-read series contains duplicate labels") + errInvalidDataSet = errors.New("invalid Prometheus remote-read dataset") + errInvalidSample = errors.New("invalid Prometheus remote-read sample") + errInvalidTimestamp = errors.New("prometheus remote-read timestamp exceeds nanosecond range") + errUnexpectedResults = errors.New("prometheus remote-read response must contain exactly one result") + errUnexpectedTimeStep = errors.New("prometheus remote-read sample timestamp is not millisecond-aligned") +) + +const nanosPerMillisecond = int64(time.Millisecond) + +// UnmarshalTimeseries converts a snappy-compressed remote-read response into a DataSet. +func UnmarshalTimeseries(data []byte, + trq *timeseries.TimeRangeQuery, +) (timeseries.Timeseries, error) { + return UnmarshalTimeseriesReader(bytes.NewReader(data), trq) +} + +// UnmarshalTimeseriesReader converts a snappy-compressed remote-read response into a DataSet. +func UnmarshalTimeseriesReader(reader io.Reader, + trq *timeseries.TimeRangeQuery, +) (timeseries.Timeseries, error) { + if reader == nil || trq == nil { + return nil, timeseries.ErrNoTimerangeQuery + } + decoded, err := readSnappyBlock(reader, responseDecodeLimit(trq.ParsedQuery)) + if err != nil { + return nil, fmt.Errorf("decode Prometheus remote-read response: %w", err) + } + response := &prompb.ReadResponse{} + if err := proto.Unmarshal(decoded, response); err != nil { + return nil, fmt.Errorf("unmarshal Prometheus remote-read response: %w", err) + } + if len(response.Results) != 1 || response.Results[0] == nil { + return nil, errUnexpectedResults + } + + result := &dataset.Result{StatementID: 0} + if response.Results[0].Timeseries != nil { + result.SeriesList = make(dataset.SeriesList, len(response.Results[0].Timeseries)) + } + for i, wireSeries := range response.Results[0].Timeseries { + series, err := fromWireSeries(wireSeries, trq.Statement) + if err != nil { + return nil, err + } + result.SeriesList[i] = series + } + + return &dataset.DataSet{ + Results: dataset.Results{result}, + TimeRangeQuery: trq, + ExtentList: timeseries.ExtentList{trq.Extent}, + }, nil +} + +func fromWireSeries(wireSeries *prompb.TimeSeries, + statement string, +) (*dataset.Series, error) { + if wireSeries == nil { + return nil, errInvalidDataSet + } + tags := make(dataset.Tags, len(wireSeries.Labels)) + for _, label := range wireSeries.Labels { + if label == nil { + return nil, errInvalidDataSet + } + if _, ok := tags[label.Name]; ok { + return nil, errDuplicateLabel + } + tags[label.Name] = label.Value + } + + tagNames := tags.Keys() + tagFields := make(timeseries.FieldDefinitions, len(tagNames)) + for i, name := range tagNames { + tagFields[i] = timeseries.FieldDefinition{ + Name: name, + DataType: timeseries.String, + Role: timeseries.RoleTag, + } + } + header := dataset.SeriesHeader{ + Name: tags["__name__"], + Tags: tags, + TagFieldsList: tagFields, + QueryStatement: statement, + TimestampField: timeseries.FieldDefinition{Name: "timestamp", DataType: timeseries.DateTimeUnixMilli, Role: timeseries.RoleTimestamp}, + ValueFieldsList: timeseries.FieldDefinitions{{Name: "value", DataType: timeseries.Float64, Role: timeseries.RoleValue}}, + } + header.CalculateSize() + + points := make(dataset.Points, len(wireSeries.Samples)) + var pointSize int64 + for i, sample := range wireSeries.Samples { + if sample == nil { + return nil, errInvalidSample + } + if sample.Timestamp > math.MaxInt64/nanosPerMillisecond || + sample.Timestamp < math.MinInt64/nanosPerMillisecond { + return nil, errInvalidTimestamp + } + points[i] = dataset.Point{ + Epoch: epoch.Epoch(sample.Timestamp * nanosPerMillisecond), + Size: 20, + Values: []any{sample.Value}, + } + pointSize += int64(points[i].Size) + } + slices.SortFunc(points, func(a, b dataset.Point) int { + return cmpEpoch(a.Epoch, b.Epoch) + }) + + return &dataset.Series{Header: header, Points: points, PointSize: pointSize}, nil +} + +func cmpEpoch(a, b epoch.Epoch) int { + if a < b { + return -1 + } + if a > b { + return 1 + } + return 0 +} + +// MarshalTimeseries converts a DataSet into a snappy-compressed remote-read response. +func MarshalTimeseries(ts timeseries.Timeseries, + _ *timeseries.RequestOptions, _ int, +) ([]byte, error) { + response, err := toWireResponse(ts) + if err != nil { + return nil, err + } + encoded, err := proto.Marshal(response) + if err != nil { + return nil, fmt.Errorf("marshal Prometheus remote-read response: %w", err) + } + return snappy.Encode(nil, encoded), nil +} + +// MarshalTimeseriesWriter writes a snappy-compressed remote-read response. +func MarshalTimeseriesWriter(ts timeseries.Timeseries, + rlo *timeseries.RequestOptions, status int, writer io.Writer, +) error { + if writer == nil { + return errInvalidDataSet + } + body, err := MarshalTimeseries(ts, rlo, status) + if err != nil { + return err + } + if responseWriter, ok := writer.(http.ResponseWriter); ok { + responseWriter.Header().Set(headers.NameContentType, ContentType) + responseWriter.Header().Set(headers.NameContentEncoding, ContentEncoding) + } + _, err = writer.Write(body) + return err +} + +func toWireResponse(ts timeseries.Timeseries) (*prompb.ReadResponse, error) { + ds, ok := ts.(*dataset.DataSet) + if !ok || ds == nil || len(ds.Results) != 1 || ds.Results[0] == nil { + return nil, errInvalidDataSet + } + queryResult := &prompb.QueryResult{} + if ds.Results[0].SeriesList != nil { + queryResult.Timeseries = make([]*prompb.TimeSeries, 0, len(ds.Results[0].SeriesList)) + } + for _, series := range ds.Results[0].SeriesList { + if series == nil { + continue + } + wireSeries, err := toWireSeries(series) + if err != nil { + return nil, err + } + queryResult.Timeseries = append(queryResult.Timeseries, wireSeries) + } + return &prompb.ReadResponse{Results: []*prompb.QueryResult{queryResult}}, nil +} + +func toWireSeries(series *dataset.Series) (*prompb.TimeSeries, error) { + labels := make([]*prompb.Label, 0, len(series.Header.Tags)) + for _, name := range series.Header.Tags.Keys() { + labels = append(labels, &prompb.Label{Name: name, Value: series.Header.Tags[name]}) + } + samples := make([]*prompb.Sample, len(series.Points)) + for i, point := range series.Points { + if len(point.Values) != 1 { + return nil, errInvalidSample + } + value, ok := point.Values[0].(float64) + if !ok { + return nil, errInvalidSample + } + if int64(point.Epoch)%nanosPerMillisecond != 0 { + return nil, errUnexpectedTimeStep + } + samples[i] = &prompb.Sample{ + Timestamp: int64(point.Epoch) / nanosPerMillisecond, + Value: value, + } + } + return &prompb.TimeSeries{Labels: labels, Samples: samples}, nil +} diff --git a/pkg/backends/influxdb/promremote/serialization_test.go b/pkg/backends/influxdb/promremote/serialization_test.go new file mode 100644 index 000000000..eb67c64fe --- /dev/null +++ b/pkg/backends/influxdb/promremote/serialization_test.go @@ -0,0 +1,226 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package promremote + +import ( + "bytes" + "encoding/binary" + "errors" + "math" + "net/http/httptest" + "testing" + "time" + + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote/prompb" + "github.com/trickstercache/trickster/v2/pkg/proxy/headers" + "github.com/trickstercache/trickster/v2/pkg/timeseries" + "github.com/trickstercache/trickster/v2/pkg/timeseries/dataset" + "github.com/trickstercache/trickster/v2/pkg/timeseries/epoch" + + "github.com/golang/snappy" + "google.golang.org/protobuf/proto" +) + +func encodeReadResponse(t *testing.T, response *prompb.ReadResponse) []byte { + t.Helper() + b, err := proto.Marshal(response) + if err != nil { + t.Fatal(err) + } + return snappy.Encode(nil, b) +} + +func decodeReadResponse(t *testing.T, body []byte) *prompb.ReadResponse { + t.Helper() + b, err := snappy.Decode(nil, body) + if err != nil { + t.Fatal(err) + } + response := &prompb.ReadResponse{} + if err := proto.Unmarshal(b, response); err != nil { + t.Fatal(err) + } + return response +} + +func TestTimeseriesRoundTrip(t *testing.T) { + trq := ×eries.TimeRangeQuery{ + Statement: "query", + Step: time.Millisecond, + Extent: timeseries.Extent{ + Start: time.UnixMilli(100), + End: time.UnixMilli(200), + }, + } + wire := &prompb.ReadResponse{Results: []*prompb.QueryResult{{ + Timeseries: []*prompb.TimeSeries{{ + Labels: []*prompb.Label{ + {Name: "job", Value: "api"}, + {Name: "__name__", Value: "requests_total"}, + }, + Samples: []*prompb.Sample{ + {Timestamp: 200, Value: 2}, + {Timestamp: 100, Value: 1}, + }, + }}, + }}} + + ts, err := UnmarshalTimeseries(encodeReadResponse(t, wire), trq) + if err != nil { + t.Fatal(err) + } + ds := ts.(*dataset.DataSet) + if len(ds.Results) != 1 || len(ds.Results[0].SeriesList) != 1 { + t.Fatalf("dataset shape = %#v", ds.Results) + } + series := ds.Results[0].SeriesList[0] + if series.Header.Name != "requests_total" || series.Header.Tags["job"] != "api" { + t.Fatalf("series header = %#v", series.Header) + } + if len(series.Header.TagFieldsList) != 2 || + series.Header.TagFieldsList[0].Name != "__name__" || + series.Header.TagFieldsList[1].Name != "job" { + t.Fatalf("tag fields = %#v", series.Header.TagFieldsList) + } + if series.Points[0].Epoch != epoch.Epoch(100*int64(time.Millisecond)) || + series.Points[1].Epoch != epoch.Epoch(200*int64(time.Millisecond)) { + t.Fatalf("points were not sorted: %#v", series.Points) + } + + body, err := MarshalTimeseries(ds, nil, 200) + if err != nil { + t.Fatal(err) + } + roundTrip := decodeReadResponse(t, body) + got := roundTrip.Results[0].Timeseries[0] + if got.Labels[0].Name != "__name__" || got.Labels[1].Name != "job" { + t.Fatalf("labels were not sorted: %#v", got.Labels) + } + if got.Samples[0].Timestamp != 100 || got.Samples[1].Timestamp != 200 { + t.Fatalf("samples = %#v", got.Samples) + } +} + +func TestMarshalTimeseriesWriterHeaders(t *testing.T) { + ds := &dataset.DataSet{Results: dataset.Results{{}}} + w := httptest.NewRecorder() + if err := MarshalTimeseriesWriter(ds, nil, 200, w); err != nil { + t.Fatal(err) + } + if w.Header().Get(headers.NameContentType) != ContentType || + w.Header().Get(headers.NameContentEncoding) != ContentEncoding { + t.Fatalf("headers = %#v", w.Header()) + } + response := decodeReadResponse(t, w.Body.Bytes()) + if len(response.Results) != 1 || response.Results[0] == nil { + t.Fatalf("response = %#v", response) + } +} + +func TestUnmarshalTimeseriesErrors(t *testing.T) { + trq := ×eries.TimeRangeQuery{} + tests := map[string][]byte{ + "bad compression": []byte("not-snappy"), + "no results": encodeReadResponse(t, &prompb.ReadResponse{}), + "two results": encodeReadResponse(t, &prompb.ReadResponse{Results: []*prompb.QueryResult{ + {}, {}, + }}), + "duplicate label": encodeReadResponse(t, &prompb.ReadResponse{Results: []*prompb.QueryResult{{ + Timeseries: []*prompb.TimeSeries{{Labels: []*prompb.Label{ + {Name: "job", Value: "a"}, {Name: "job", Value: "b"}, + }}}, + }}}), + "timestamp overflow": encodeReadResponse(t, &prompb.ReadResponse{Results: []*prompb.QueryResult{{ + Timeseries: []*prompb.TimeSeries{{Samples: []*prompb.Sample{{ + Timestamp: math.MaxInt64, + }}}}, + }}}), + } + for name, body := range tests { + t.Run(name, func(t *testing.T) { + if _, err := UnmarshalTimeseries(body, trq); err == nil { + t.Fatal("expected an error") + } + }) + } + if _, err := UnmarshalTimeseriesReader(nil, trq); err == nil { + t.Fatal("nil reader must fail") + } +} + +func TestUnmarshalTimeseriesDecodeLimit(t *testing.T) { + trq := ×eries.TimeRangeQuery{ParsedQuery: &parsedRequest{decodeLimit: 16}} + + t.Run("decoded size", func(t *testing.T) { + body := binary.AppendUvarint(nil, 17) + if _, err := UnmarshalTimeseries(body, trq); !errors.Is(err, errBodyTooLarge) { + t.Fatalf("error = %v", err) + } + }) + + t.Run("compressed size", func(t *testing.T) { + body := bytes.Repeat([]byte{0}, snappy.MaxEncodedLen(16)+1) + if _, err := UnmarshalTimeseries(body, trq); !errors.Is(err, errBodyTooLarge) { + t.Fatalf("error = %v", err) + } + }) +} + +func TestMarshalTimeseriesErrors(t *testing.T) { + if _, err := MarshalTimeseries(nil, nil, 200); err == nil { + t.Fatal("nil dataset must fail") + } + ds := &dataset.DataSet{Results: dataset.Results{{SeriesList: dataset.SeriesList{{ + Header: dataset.SeriesHeader{Tags: dataset.Tags{"job": "api"}}, + Points: dataset.Points{{Epoch: 1, Values: []any{float64(1)}}}, + }}}}} + if _, err := MarshalTimeseries(ds, nil, 200); err == nil { + t.Fatal("sub-millisecond timestamp must fail") + } + ds.Results[0].SeriesList[0].Points[0] = dataset.Point{ + Epoch: epoch.Epoch(time.Millisecond), + Values: []any{"1"}, + } + if _, err := MarshalTimeseries(ds, nil, 200); err == nil { + t.Fatal("non-float sample must fail") + } +} + +func TestEmptyResultRoundTrip(t *testing.T) { + wire := &prompb.ReadResponse{Results: []*prompb.QueryResult{{}}} + ts, err := UnmarshalTimeseries(encodeReadResponse(t, wire), ×eries.TimeRangeQuery{}) + if err != nil { + t.Fatal(err) + } + body, err := MarshalTimeseries(ts, nil, 200) + if err != nil { + t.Fatal(err) + } + got := decodeReadResponse(t, body) + if len(got.Results) != 1 || len(got.Results[0].Timeseries) != 0 { + t.Fatalf("response = %#v", got) + } +} + +func TestMarshalTimeseriesToBuffer(t *testing.T) { + ds := &dataset.DataSet{Results: dataset.Results{{}}} + var buf bytes.Buffer + if err := MarshalTimeseriesWriter(ds, nil, 200, &buf); err != nil { + t.Fatal(err) + } + decodeReadResponse(t, buf.Bytes()) +} diff --git a/pkg/backends/influxdb/promremote/snappy.go b/pkg/backends/influxdb/promremote/snappy.go new file mode 100644 index 000000000..92812bf61 --- /dev/null +++ b/pkg/backends/influxdb/promremote/snappy.go @@ -0,0 +1,83 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package promremote + +import ( + "errors" + "fmt" + "io" + "math" + "net/http" + + bo "github.com/trickstercache/trickster/v2/pkg/backends/options" + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + + "github.com/golang/snappy" +) + +var errBodyTooLarge = errors.New("prometheus remote-read body exceeds decode limit") + +func requestDecodeLimit(r *http.Request) int { + if rsc := request.GetResources(r); rsc != nil && rsc.BackendOptions != nil && + rsc.BackendOptions.MaxCaptureBytes > 0 { + return rsc.BackendOptions.MaxCaptureBytes + } + return bo.DefaultMaxCaptureBytes +} + +func responseDecodeLimit(trq any) int { + if parsed, ok := trq.(*parsedRequest); ok && parsed != nil && parsed.decodeLimit > 0 { + return parsed.decodeLimit + } + return bo.DefaultMaxCaptureBytes +} + +func decodeSnappyBlock(compressed []byte, maxDecoded int) ([]byte, error) { + if maxDecoded <= 0 { + maxDecoded = bo.DefaultMaxCaptureBytes + } + decodedLen, err := snappy.DecodedLen(compressed) + if err != nil { + return nil, err + } + if decodedLen > maxDecoded { + return nil, fmt.Errorf("%w: %d > %d", errBodyTooLarge, decodedLen, maxDecoded) + } + return snappy.Decode(nil, compressed) +} + +func readSnappyBlock(reader io.Reader, maxDecoded int) ([]byte, error) { + if maxDecoded <= 0 { + maxDecoded = bo.DefaultMaxCaptureBytes + } + maxEncoded := snappy.MaxEncodedLen(maxDecoded) + if maxEncoded < 0 { + maxEncoded = maxDecoded + } + readLimit := int64(maxEncoded) + if readLimit < math.MaxInt64 { + readLimit++ + } + compressed, err := io.ReadAll(io.LimitReader(reader, readLimit)) + if err != nil { + return nil, err + } + if len(compressed) > maxEncoded { + return nil, fmt.Errorf("%w: compressed size exceeds %d", errBodyTooLarge, maxEncoded) + } + return decodeSnappyBlock(compressed, maxDecoded) +} diff --git a/pkg/backends/influxdb/routes.go b/pkg/backends/influxdb/routes.go index 65e81d2fe..9236ca865 100644 --- a/pkg/backends/influxdb/routes.go +++ b/pkg/backends/influxdb/routes.go @@ -20,6 +20,7 @@ import ( "net/http" "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/influxql" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote" bo "github.com/trickstercache/trickster/v2/pkg/backends/options" "github.com/trickstercache/trickster/v2/pkg/backends/providers" "github.com/trickstercache/trickster/v2/pkg/proxy/handlers" @@ -43,6 +44,15 @@ func (c *Client) RegisterHandlers(handlers.Lookup) { // DefaultPathConfigs returns the default PathConfigs for the given Provider func (c *Client) DefaultPathConfigs(_ *bo.Options) po.List { return po.List{ + { + Path: promremote.Path, + HandlerName: mnQuery, + Methods: []string{http.MethodPost}, + CacheKeyParams: []string{"db", "rp", "u", "p"}, + CacheKeyHeaders: []string{}, + MatchTypeName: matching.PathMatchNameExact, + MatchType: matching.PathMatchTypeExact, + }, { Path: "/" + mnQuery, HandlerName: mnQuery, diff --git a/pkg/backends/influxdb/routes_test.go b/pkg/backends/influxdb/routes_test.go index 1683a114f..b86d14c70 100644 --- a/pkg/backends/influxdb/routes_test.go +++ b/pkg/backends/influxdb/routes_test.go @@ -17,9 +17,11 @@ package influxdb import ( + "net/http" "slices" "testing" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote" "github.com/trickstercache/trickster/v2/pkg/backends/providers" po "github.com/trickstercache/trickster/v2/pkg/proxy/paths/options" "github.com/trickstercache/trickster/v2/pkg/proxy/request" @@ -65,8 +67,15 @@ func TestDefaultPathConfigs(t *testing.T) { t.Errorf("expected to find path named: %s", "/") } - const expectedLen = 3 + const expectedLen = 4 if len(rsc.BackendOptions.Paths) != expectedLen { t.Errorf("expected ordered length to be: %d, got: %d", expectedLen, len(rsc.BackendOptions.Paths)) } + if !slices.ContainsFunc([]*po.Options(rsc.BackendOptions.Paths), + func(pathConfig *po.Options) bool { + return pathConfig.Path == promremote.Path && pathConfig.HandlerName == mnQuery && + len(pathConfig.Methods) == 1 && pathConfig.Methods[0] == http.MethodPost + }) { + t.Errorf("expected remote-read path named: %s", promremote.Path) + } } diff --git a/pkg/backends/influxdb/serialization.go b/pkg/backends/influxdb/serialization.go index f7735ad7c..b9d6751d2 100644 --- a/pkg/backends/influxdb/serialization.go +++ b/pkg/backends/influxdb/serialization.go @@ -23,6 +23,7 @@ import ( "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/flux" "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/influxql" "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/iofmt" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote" "github.com/trickstercache/trickster/v2/pkg/errors" "github.com/trickstercache/trickster/v2/pkg/timeseries" "github.com/trickstercache/trickster/v2/pkg/timeseries/dataset" @@ -46,6 +47,9 @@ func UnmarshalTimeseries(data []byte, if len(data) == 0 || trq == nil { return nil, errors.ErrBadRequest } + if promremote.IsParsedQuery(trq.ParsedQuery) { + return promremote.UnmarshalTimeseries(data, trq) + } if strings.Contains(strings.ToLower(trq.Statement), flux.FuncRange) { return flux.UnmarshalTimeseries(data, trq) } @@ -58,6 +62,9 @@ func UnmarshalTimeseriesReader(reader io.Reader, if reader == nil || trq == nil { return nil, errors.ErrBadRequest } + if promremote.IsParsedQuery(trq.ParsedQuery) { + return promremote.UnmarshalTimeseriesReader(reader, trq) + } if strings.Contains(strings.ToLower(trq.Statement), flux.FuncRange) { return flux.UnmarshalTimeseriesReader(reader, trq) } @@ -70,6 +77,9 @@ func MarshalTimeseries(ts timeseries.Timeseries, if ts == nil || rlo == nil { return nil, errors.ErrBadRequest } + if iofmt.Format(rlo.OutputFormat).IsPromRemoteRead() { + return promremote.MarshalTimeseries(ts, rlo, status) + } if iofmt.Format(rlo.OutputFormat).IsInfluxQL() { return influxql.MarshalTimeseries(ts, rlo, status) } @@ -82,6 +92,9 @@ func MarshalTimeseriesWriter(ts timeseries.Timeseries, if ts == nil || rlo == nil || w == nil { return errors.ErrBadRequest } + if iofmt.Format(rlo.OutputFormat).IsPromRemoteRead() { + return promremote.MarshalTimeseriesWriter(ts, rlo, status, w) + } if rlo.OutputFormat < 4 { return influxql.MarshalTimeseriesWriter(ts, rlo, status, w) } diff --git a/pkg/backends/influxdb/url.go b/pkg/backends/influxdb/url.go index 3f57a0777..94be4d153 100644 --- a/pkg/backends/influxdb/url.go +++ b/pkg/backends/influxdb/url.go @@ -23,6 +23,7 @@ import ( "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/flux" ti "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/influxql" + "github.com/trickstercache/trickster/v2/pkg/backends/influxdb/promremote" "github.com/trickstercache/trickster/v2/pkg/timeseries" "github.com/influxdata/influxql" @@ -51,6 +52,9 @@ func (c *Client) SetExtent(r *http.Request, trq *timeseries.TimeRangeQuery, } trq.ParsedQuery = t2.ParsedQuery } + if promremote.IsParsedQuery(trq.ParsedQuery) { + return promremote.SetExtent(r, trq, extent) + } switch q := trq.ParsedQuery.(type) { case *influxql.Query: ti.SetExtent(r, trq, extent, q) diff --git a/pkg/proxy/engines/deltaproxycache.go b/pkg/proxy/engines/deltaproxycache.go index ac4409c4a..ed9a4d601 100644 --- a/pkg/proxy/engines/deltaproxycache.go +++ b/pkg/proxy/engines/deltaproxycache.go @@ -260,12 +260,13 @@ func DeltaProxyCacheRequest(w http.ResponseWriter, r *http.Request, modeler *tim OldestRetainedTimestamp := time.Time{} if o.TimeseriesEvictionMethod == evictionmethods.EvictionMethodOldest { - OldestRetainedTimestamp = now.Truncate(trq.Step).Add(-(trq.Step * time.Duration(o.TimeseriesRetention))) + retentionStep := trq.CachePolicyStep() + OldestRetainedTimestamp = now.Truncate(retentionStep).Add(-(retentionStep * time.Duration(o.TimeseriesRetention))) if trq.Extent.End.Before(OldestRetainedTimestamp) { logger.Debug("timerange end is too old to consider caching", logging.Pairs{ "oldestRetainedTimestamp": OldestRetainedTimestamp, - "step": trq.Step, "retention": o.TimeseriesRetention, + "step": retentionStep, "retention": o.TimeseriesRetention, }) if trq.OriginalBody != nil { request.SetBody(r, trq.OriginalBody) diff --git a/pkg/timeseries/timerangequery.go b/pkg/timeseries/timerangequery.go index 1922e787a..e65713af0 100644 --- a/pkg/timeseries/timerangequery.go +++ b/pkg/timeseries/timerangequery.go @@ -37,6 +37,9 @@ type TimeRangeQuery struct { Extent Extent `msg:"ex"` // Step indicates the amount of time in seconds between each datapoint in a TimeRangeQuery's resulting timeseries Step time.Duration `msg:"-"` + // PolicyStep optionally overrides Step for cache policies expressed in logical query points. + // It does not affect timestamp-grid operations such as extent normalization and gap detection. + PolicyStep time.Duration `msg:"-"` // Phase is the bucket offset from the Unix epoch Phase time.Duration `msg:"-"` // TemplateURL is used by some Backend providers for templatization of url parameters containing timestamps @@ -66,6 +69,7 @@ func (trq *TimeRangeQuery) Clone() *TimeRangeQuery { t := &TimeRangeQuery{ Statement: trq.Statement, Step: trq.Step, + PolicyStep: trq.PolicyStep, Phase: trq.Phase, StepNS: trq.StepNS, Extent: Extent{Start: trq.Extent.Start, End: trq.Extent.End}, @@ -95,6 +99,14 @@ func (trq *TimeRangeQuery) Clone() *TimeRangeQuery { return t } +// CachePolicyStep returns the duration used by point-based cache policies. +func (trq *TimeRangeQuery) CachePolicyStep() time.Duration { + if trq.PolicyStep > 0 { + return trq.PolicyStep + } + return trq.Step +} + // NormalizeExtent adjusts the Start and End of a TimeRangeQuery's Extent to align against normalized boundaries. func (trq *TimeRangeQuery) NormalizeExtent() { if trq.Step > 0 { @@ -152,7 +164,7 @@ func (trq *TimeRangeQuery) GetBackfillTolerance(def time.Duration, points int) t } if points > 0 { - sd := time.Duration(points) * trq.Step + sd := time.Duration(points) * trq.CachePolicyStep() if sd > def { return sd } @@ -163,7 +175,7 @@ func (trq *TimeRangeQuery) GetBackfillTolerance(def time.Duration, points int) t // Size returns the memory usage in bytes of the TimeRangeQuery func (trq *TimeRangeQuery) Size() int { - return len(trq.Statement) + 24 + 16 + trq.TimestampDefinition.Size() + // Extent=24 + Step=8 + Phase=8 + return len(trq.Statement) + 24 + 24 + trq.TimestampDefinition.Size() + // Extent=24 + Step=8 + PolicyStep=8 + Phase=8 urls.Size(trq.TemplateURL) + 11 // FFwDisable=1 IsOffset=1 StepNS=8 CustomData=1 } diff --git a/pkg/timeseries/timerangequery_test.go b/pkg/timeseries/timerangequery_test.go index 2c6fbc701..47a4ac74c 100644 --- a/pkg/timeseries/timerangequery_test.go +++ b/pkg/timeseries/timerangequery_test.go @@ -165,6 +165,23 @@ func TestClone(t *testing.T) { }) } +func TestCachePolicyStep(t *testing.T) { + trq := &TimeRangeQuery{Step: time.Millisecond} + if got := trq.CachePolicyStep(); got != time.Millisecond { + t.Fatalf("policy step = %s", got) + } + trq.PolicyStep = 15 * time.Second + if got := trq.CachePolicyStep(); got != 15*time.Second { + t.Fatalf("policy step = %s", got) + } + if got := trq.Clone().PolicyStep; got != trq.PolicyStep { + t.Fatalf("cloned policy step = %s", got) + } + if got := trq.GetBackfillTolerance(0, 2); got != 30*time.Second { + t.Fatalf("backfill tolerance = %s", got) + } +} + func TestSizeTRQ(t *testing.T) { u, _ := url.Parse("http://127.0.0.1/") trq := &TimeRangeQuery{Statement: "1234", Extent: Extent{ @@ -172,8 +189,8 @@ func TestSizeTRQ(t *testing.T) { End: time.Unix(10, 0), }, Step: time.Duration(5) * time.Second, TemplateURL: u} size := trq.Size() - if size != 127 { - t.Errorf("expected %d got %d", 127, size) + if size != 135 { + t.Errorf("expected %d got %d", 135, size) } }