feat: cache InfluxDB Prometheus remote reads

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.

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

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=

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

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=

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)
+	}
+}

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():

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
 }

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{