diff --git a/README.md b/README.md
index 3a8fddb01..8a672164f 100644
--- a/README.md
+++ b/README.md
@@ -49,6 +49,8 @@ Trickster works with virtually any Dashboard application that makes queries to a
InfluxDB
+Elasticsearch
+
See the [Supported TSDB Providers](./docs/supported-backend-providers.md) document for full details
### How Trickster Accelerates Time Series
diff --git a/deploy/kube/configmap.yaml b/deploy/kube/configmap.yaml
index 3c8792542..868901a1e 100644
--- a/deploy/kube/configmap.yaml
+++ b/deploy/kube/configmap.yaml
@@ -229,7 +229,7 @@ data:
default:
# provider identifies the backend provider.
- # Valid options are: prometheus, influxdb, clickhouse, reverseproxycache (or just rpc)
+ # Valid options are: prometheus, influxdb, clickhouse, elasticsearch, reverseproxycache (or just rpc)
# provider is a required configuration value
provider: prometheus
diff --git a/docs/developer/environment/README.md b/docs/developer/environment/README.md
index 7bbf7bb6a..e47a509c6 100644
--- a/docs/developer/environment/README.md
+++ b/docs/developer/environment/README.md
@@ -21,6 +21,9 @@ You can combine these make actions `make developer-start serve-dev` if you want.
Once you have the Docker Compose running, and Trickster running locally, visit
the Grafana Dashboard at .
+The Kibana frontend is available at and is configured
+to use Trickster's `es1` Elasticsearch backend at
+`http://host.docker.internal:8480/es1`.
The data in this dashboard is polled by Prometheus from your local Trickster
dev instance. So the longer you keep this dashboard up and refreshing, the more
@@ -28,6 +31,15 @@ you can test out Trickster acceleration features. You can change the Data Source
selector to go between various Trickster configs, or bypass Trickster altogether
for verification purposes.
+Elasticsearch is seeded on startup with the `trickster-dev-logs` index. The seed
+data includes recent and older `@timestamp` values so developers can verify
+Elasticsearch date histogram caching through Trickster.
+After Kibana connects through Trickster, the Kibana seed container creates the
+`Trickster Dev Logs` data view, saved search, and dashboard for that index. The
+dashboard includes a minute-aligned log-volume histogram that exercises Delta
+Proxy Cache requests, followed by the generated log documents. It is available at
+.
+
You can stop the developer environment by running `make developer-stop`. To
delete the developer environment, run `make developer-delete` which will destroy
all data.
diff --git a/docs/developer/environment/docker-compose-data/elasticsearch-config/seeding/seed.sh b/docs/developer/environment/docker-compose-data/elasticsearch-config/seeding/seed.sh
new file mode 100644
index 000000000..6a47171a3
--- /dev/null
+++ b/docs/developer/environment/docker-compose-data/elasticsearch-config/seeding/seed.sh
@@ -0,0 +1,120 @@
+#!/bin/sh
+
+#
+# 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.
+#
+
+set -eu
+
+ES_URL="${ES_URL:-http://elasticsearch:9200}"
+ES_INDEX="${ES_INDEX:-trickster-dev-logs}"
+
+wait_for_elasticsearch() {
+ echo "waiting for Elasticsearch at ${ES_URL}"
+ until curl -fsS "${ES_URL}/_cluster/health?wait_for_status=yellow&timeout=1s" >/dev/null 2>&1; do
+ sleep 2
+ done
+}
+
+iso_utc() {
+ date -u -d "@$1" '+%Y-%m-%dT%H:%M:%SZ'
+}
+
+create_index() {
+ echo "creating ${ES_INDEX}"
+ curl -fsS -X DELETE "${ES_URL}/${ES_INDEX}" >/dev/null 2>&1 || true
+ curl -fsS -X PUT "${ES_URL}/${ES_INDEX}" \
+ -H 'Content-Type: application/json' \
+ -d '{
+ "mappings": {
+ "properties": {
+ "@timestamp": { "type": "date" },
+ "event_id": { "type": "keyword" },
+ "service": { "type": "keyword" },
+ "level": { "type": "keyword" },
+ "status_code": { "type": "integer" },
+ "duration_ms": { "type": "integer" },
+ "cacheable": { "type": "boolean" },
+ "message": { "type": "text" }
+ }
+ }
+}' >/dev/null
+}
+
+append_log() {
+ event_id="$1"
+ timestamp="$2"
+ service="$3"
+ level="$4"
+ status_code="$5"
+ duration_ms="$6"
+ cacheable="$7"
+ message="$8"
+
+ printf '{"index":{"_index":"%s","_id":"%s"}}\n' "${ES_INDEX}" "${event_id}" >> "${BULK_FILE}"
+ printf '{"@timestamp":"%s","event_id":"%s","service":"%s","level":"%s","status_code":%s,"duration_ms":%s,"cacheable":%s,"message":"%s"}\n' \
+ "${timestamp}" "${event_id}" "${service}" "${level}" "${status_code}" "${duration_ms}" "${cacheable}" "${message}" >> "${BULK_FILE}"
+}
+
+seed_recent_logs() {
+ now="$(date -u +%s)"
+ i=0
+ while [ "${i}" -lt 180 ]; do
+ ts="$(iso_utc "$((now - (i * 60)))")"
+ case $((i % 4)) in
+ 0) service="api"; level="info"; status=200; cacheable=true ;;
+ 1) service="worker"; level="info"; status=202; cacheable=true ;;
+ 2) service="edge"; level="warn"; status=429; cacheable=false ;;
+ *) service="checkout"; level="error"; status=500; cacheable=false ;;
+ esac
+ duration="$((25 + (i % 45)))"
+ append_log "recent-${i}" "${ts}" "${service}" "${level}" "${status}" "${duration}" "${cacheable}" "recent trickster developer log ${i}"
+ i=$((i + 1))
+ done
+}
+
+seed_older_logs() {
+ now="$(date -u +%s)"
+ i=0
+ while [ "${i}" -lt 72 ]; do
+ ts="$(iso_utc "$((now - (3 * 86400) - (i * 3600)))")"
+ case $((i % 3)) in
+ 0) service="api"; level="info"; status=200; cacheable=true ;;
+ 1) service="edge"; level="warn"; status=404; cacheable=false ;;
+ *) service="worker"; level="error"; status=503; cacheable=false ;;
+ esac
+ duration="$((50 + (i % 75)))"
+ append_log "older-${i}" "${ts}" "${service}" "${level}" "${status}" "${duration}" "${cacheable}" "older trickster developer log ${i}"
+ i=$((i + 1))
+ done
+}
+
+load_bulk_data() {
+ echo "loading generated log data into ${ES_INDEX}"
+ curl -fsS -X POST "${ES_URL}/_bulk?refresh=true" \
+ -H 'Content-Type: application/x-ndjson' \
+ --data-binary "@${BULK_FILE}" >/dev/null
+}
+
+BULK_FILE="$(mktemp)"
+trap 'rm -f "${BULK_FILE}"' EXIT
+
+wait_for_elasticsearch
+create_index
+seed_recent_logs
+seed_older_logs
+load_bulk_data
+
+echo "seeded ${ES_INDEX}"
diff --git a/docs/developer/environment/docker-compose-data/kibana-config/seeding/seed.sh b/docs/developer/environment/docker-compose-data/kibana-config/seeding/seed.sh
new file mode 100644
index 000000000..7dcdaf00e
--- /dev/null
+++ b/docs/developer/environment/docker-compose-data/kibana-config/seeding/seed.sh
@@ -0,0 +1,173 @@
+#!/bin/sh
+
+#
+# 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.
+#
+
+set -eu
+
+KIBANA_URL="${KIBANA_URL:-http://kibana:5601}"
+KIBANA_VERSION="${KIBANA_VERSION:-8.17.3}"
+
+wait_for_kibana() {
+ echo "waiting for Kibana at ${KIBANA_URL}"
+ i=0
+ until curl -fsS "${KIBANA_URL}/api/status" 2>/dev/null | grep -q '"level":"available"'; do
+ i=$((i + 1))
+ if [ "${i}" -gt 450 ]; then
+ echo "timed out waiting for Kibana" >&2
+ return 1
+ fi
+ sleep 2
+ done
+}
+
+post_json() {
+ path="$1"
+ file="$2"
+ curl -fsS -X POST "${KIBANA_URL}${path}" \
+ -H 'kbn-xsrf: true' \
+ -H 'Content-Type: application/json' \
+ --data-binary "@${file}" >/dev/null
+}
+
+create_kibana_config() {
+ cat > "${CONFIG_FILE}" < "${DATA_VIEW_FILE}" < "${SEARCH_FILE}" < "${VISUALIZATION_FILE}" < "${DASHBOARD_FILE}" </dev/null || exit 1"]
+ interval: 5s
+ timeout: 3s
+ retries: 20
+ restart: always
+
+ # elasticsearch seed data
+ elasticsearch_seed:
+ links:
+ - elasticsearch
+ image: curlimages/curl:8.11.1
+ environment:
+ - ES_URL=http://elasticsearch:9200
+ volumes:
+ - ./docker-compose-data/elasticsearch-config/seeding:/seeding
+ entrypoint: sh /seeding/seed.sh
+ restart: on-failure:10
+ depends_on:
+ - elasticsearch
+
+ # elasticsearch frontend
+ kibana:
+ image: docker.elastic.co/kibana/kibana:8.17.3
+ extra_hosts:
+ - host.docker.internal:host-gateway
+ environment:
+ - ELASTICSEARCH_HOSTS=["http://host.docker.internal:8480/es1"]
+ - XPACK_SECURITY_ENABLED=false
+ - TELEMETRY_ENABLED=false
+ - SERVER_PUBLICBASEURL=http://127.0.0.1:5601
+ ports:
+ - 5601:5601
+ restart: always
+ depends_on:
+ - elasticsearch
+
+ # kibana saved objects for elasticsearch demo data
+ kibana_seed:
+ links:
+ - kibana
+ image: curlimages/curl:8.11.1
+ environment:
+ - KIBANA_URL=http://kibana:5601
+ volumes:
+ - ./docker-compose-data/kibana-config/seeding:/seeding
+ entrypoint: sh /seeding/seed.sh
+ restart: on-failure:10
+ depends_on:
+ - kibana
+
# Mocks
mockster:
@@ -157,5 +221,6 @@ services:
- influxdb2
volumes:
+ elasticsearch-data:
influxdbv2:
prometheus-data:
diff --git a/docs/developer/environment/trickster-config/trickster.yaml b/docs/developer/environment/trickster-config/trickster.yaml
index b7e6f3354..3623aa83a 100644
--- a/docs/developer/environment/trickster-config/trickster.yaml
+++ b/docs/developer/environment/trickster-config/trickster.yaml
@@ -96,6 +96,14 @@ backends:
cache_name: mem1
backfill_tolerance: 60s
timeseries_retention_factor: 2048
+ es1:
+ provider: elasticsearch
+ origin_url: 'http://127.0.0.1:9200'
+ cache_name: mem1
+ backfill_tolerance: 60s
+ timeseries_retention_factor: 2048
+ elasticsearch:
+ timestamp_field: '@timestamp'
rp1:
provider: proxy
origin_url: 'http://127.0.0.1:9090'
diff --git a/docs/elasticsearch.md b/docs/elasticsearch.md
new file mode 100644
index 000000000..8e3200cf7
--- /dev/null
+++ b/docs/elasticsearch.md
@@ -0,0 +1,55 @@
+# Elasticsearch Support
+
+Trickster can accelerate Elasticsearch search requests that return date histogram time series, such as Grafana Elasticsearch panels. Acceleration uses the Time Series Delta Proxy Cache to avoid re-querying histogram buckets that are already cached.
+
+## Configuration
+
+Use `provider: elasticsearch` and point `origin_url` at the Elasticsearch HTTP endpoint:
+
+```yaml
+backends:
+ elasticsearch:
+ provider: elasticsearch
+ origin_url: http://elasticsearch:9200
+```
+
+Elasticsearch deployments commonly use `@timestamp` as the event time field. Trickster uses that field by default. If your index uses a different date field, set `elasticsearch.timestamp_field`:
+
+```yaml
+backends:
+ elasticsearch:
+ provider: elasticsearch
+ origin_url: http://elasticsearch:9200
+ elasticsearch:
+ timestamp_field: event_time
+```
+
+## Cacheable Search Requests
+
+Trickster applies Delta Proxy Cache acceleration to `GET` or `POST` requests whose path is `/_search`, `//_search`, `/_msearch`, or `//_msearch` when each search body contains:
+
+- `size: 0`, so the response is aggregation-only
+- exactly one mandatory `range` over the configured timestamp field, using `gte` with either `lt` or `lte`
+- exactly one top-level `date_histogram` aggregation over that same timestamp field
+- a positive `fixed_interval` that evenly divides a UTC day, or a UTC calendar interval of one minute, hour, or day
+- complete, interval-aligned buckets; an inclusive `lte` must identify the final millisecond of the last bucket, while an exclusive `lt` identifies the next bucket boundary
+- array-form buckets (`keyed` is absent or false), ascending key order, no bucket `offset`, and either no `time_zone` or a UTC time zone
+- no pipeline aggregations, because their values depend on buckets outside an independently fetched cache extent
+
+Elasticsearch returns empty buckets between the first and last matching document by default. Trickster supports `min_doc_count` values of zero or one. To preserve empty buckets when Trickster fetches missing extents independently, a histogram with `min_doc_count: 0` (or no `min_doc_count`) must include both `extended_bounds.min` and `extended_bounds.max` covering the requested bucket range. Any `extended_bounds` or `hard_bounds` values must resolve to the same first and last buckets as the timestamp range.
+
+The range can use RFC3339 timestamps or epoch milliseconds. Numeric dates use Elasticsearch's default epoch-millisecond interpretation unless the range `format` selects `epoch_second`; epoch-second ranges must use boundaries that can represent every bucket exactly. Trickster normalizes the range values in the cache key and rewrites only the range, `extended_bounds`, and `hard_bounds` values when fetching missing cache extents. Each returned bucket is retained as Elasticsearch JSON, including ordinary nested bucket and metric aggregations.
+
+Variable calendar intervals, legacy `interval`, partial edge buckets, shifted or non-UTC bucket boundaries, multiple timestamp ranges, multiple top-level aggregations, pipeline aggregations, descending bucket order, response-shaping search options, and searches that request document hits fall back to Object Proxy Cache. This avoids changing query semantics or caching a partial bucket as a complete extent.
+
+## Multi Search
+
+For `_msearch`, Trickster supports the standard newline-delimited header/body pair format. Every search body in the request must match the cacheable shape above and describe the same time range and histogram interval so a single cache extent can be applied safely.
+
+## Fallback Behavior
+
+Requests that do not match the supported time-series shape are not forced through the time-series modeler. Elasticsearch `GET` requests use Object Proxy Cache when possible, unsupported search requests fall back to Object Proxy Cache using the exact request body in the cache key, and write-oriented or unsafe methods are proxied without caching.
+
+Delta-cached responses preserve the histogram buckets but reconstruct query-level metadata. `took` is reported as zero and `hits.total` is calculated from bucket `doc_count` values. For an exact total, the configured timestamp field must be a single-valued millisecond-resolution `date` field and documents must use normal document counts.
+
+For narrowly scoped reverse-proxy caching of custom Elasticsearch endpoints, you can still use `provider: reverseproxycache` and configure path-level cache key behavior explicitly.
diff --git a/docs/supported-backend-providers.md b/docs/supported-backend-providers.md
index 57c5db37a..06fe96b33 100644
--- a/docs/supported-backend-providers.md
+++ b/docs/supported-backend-providers.md
@@ -25,3 +25,9 @@ See the [InfluxDB Support Document](./influxdb.md) for more information.
Trickster supports accelerating ClickHouse time series. Specify `'clickhouse'` as the Provider when configuring Trickster.
See the [ClickHouse Support Document](./clickhouse.md) for more information.
+
+### Elasticsearch
+
+Trickster supports accelerating Elasticsearch date histogram time series. Specify `'elasticsearch'` as the Provider when configuring Trickster.
+
+See the [Elasticsearch Support Document](./elasticsearch.md) for more information.
diff --git a/examples/conf/example.full.yaml b/examples/conf/example.full.yaml
index c7cc51e6b..87f8485e1 100644
--- a/examples/conf/example.full.yaml
+++ b/examples/conf/example.full.yaml
@@ -240,7 +240,7 @@ backends:
default:
# provider identifies the backend provider.
- # Valid options are: prometheus, influxdb, clickhouse, reverseproxycache (or just rpc)
+ # Valid options are: prometheus, influxdb, clickhouse, elasticsearch, reverseproxycache (or just rpc)
# provider is a required configuration value
provider: prometheus
@@ -249,6 +249,11 @@ backends:
# labels:
# labelname: value
+ # for elasticsearch backends, you can configure the timestamp field used
+ # for date_histogram acceleration. default is @timestamp:
+ # elasticsearch:
+ # timestamp_field: '@timestamp'
+
# origin_url provides the base upstream URL for all proxied requests to this origin.
# it can be as simple as http://example.com or as complex as https://example.com:8443/path/prefix
# origin_url is a required configuration value
diff --git a/pkg/backends/elasticsearch/elasticsearch.go b/pkg/backends/elasticsearch/elasticsearch.go
new file mode 100644
index 000000000..013a536f0
--- /dev/null
+++ b/pkg/backends/elasticsearch/elasticsearch.go
@@ -0,0 +1,64 @@
+/*
+ * 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 elasticsearch provides the Elasticsearch backend provider.
+package elasticsearch
+
+import (
+ "net/http"
+
+ "github.com/trickstercache/trickster/v2/pkg/backends"
+ eso "github.com/trickstercache/trickster/v2/pkg/backends/elasticsearch/options"
+ bo "github.com/trickstercache/trickster/v2/pkg/backends/options"
+ "github.com/trickstercache/trickster/v2/pkg/backends/providers/registry/types"
+ "github.com/trickstercache/trickster/v2/pkg/cache"
+)
+
+var _ backends.TimeseriesBackend = (*Client)(nil)
+
+// Client implements the backend client interface for Elasticsearch.
+type Client struct {
+ backends.TimeseriesBackend
+}
+
+var _ types.NewBackendClientFunc = NewClient
+
+// NewClient returns a new Client instance.
+func NewClient(name string, o *bo.Options, router http.Handler,
+ cache cache.Cache, _ backends.Backends, _ types.Lookup,
+) (backends.Backend, error) {
+ if o != nil {
+ o.FastForwardDisable = true
+ if o.Elasticsearch == nil {
+ o.Elasticsearch = eso.New()
+ } else {
+ _ = o.Elasticsearch.Initialize("")
+ }
+ }
+ c := &Client{}
+ b, err := backends.NewTimeseriesBackend(name, o, c.RegisterHandlers,
+ router, cache, NewModeler())
+ c.TimeseriesBackend = b
+ return c, err
+}
+
+func (c *Client) elasticsearchOptions() *eso.Options {
+ if c == nil || c.TimeseriesBackend == nil ||
+ c.Configuration() == nil || c.Configuration().Elasticsearch == nil {
+ return eso.New()
+ }
+ return c.Configuration().Elasticsearch
+}
diff --git a/pkg/backends/elasticsearch/handler_query.go b/pkg/backends/elasticsearch/handler_query.go
new file mode 100644
index 000000000..4e4a7d81c
--- /dev/null
+++ b/pkg/backends/elasticsearch/handler_query.go
@@ -0,0 +1,52 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "net/http"
+
+ "github.com/trickstercache/trickster/v2/pkg/proxy/engines"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/urls"
+)
+
+// QueryHandler handles Elasticsearch search requests. Time-series
+// date_histogram searches use the delta proxy cache; other GET/POST requests
+// are safely handled by object cache or proxy fallback.
+func (c *Client) QueryHandler(w http.ResponseWriter, r *http.Request) {
+ if !isSearchPath(r.URL.Path) && !isMSearchPath(r.URL.Path) {
+ if r.Method == http.MethodGet || r.Method == http.MethodHead {
+ c.ObjectProxyCacheHandler(w, r)
+ return
+ }
+ c.ProxyHandler(w, r)
+ return
+ }
+ r.URL = urls.BuildUpstreamURL(r, c.BaseUpstreamURL())
+ engines.DeltaProxyCacheRequest(w, r, c.Modeler())
+}
+
+// ObjectProxyCacheHandler routes an Elasticsearch request through object cache.
+func (c *Client) ObjectProxyCacheHandler(w http.ResponseWriter, r *http.Request) {
+ r.URL = urls.BuildUpstreamURL(r, c.BaseUpstreamURL())
+ engines.ObjectProxyCacheRequest(w, r)
+}
+
+// ProxyHandler sends a request through the basic reverse proxy.
+func (c *Client) ProxyHandler(w http.ResponseWriter, r *http.Request) {
+ r.URL = urls.BuildUpstreamURL(r, c.BaseUpstreamURL())
+ engines.DoProxy(w, r, true)
+}
diff --git a/pkg/backends/elasticsearch/handler_query_test.go b/pkg/backends/elasticsearch/handler_query_test.go
new file mode 100644
index 000000000..3df6f0fbf
--- /dev/null
+++ b/pkg/backends/elasticsearch/handler_query_test.go
@@ -0,0 +1,265 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/trickstercache/trickster/v2/pkg/backends/providers"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/request"
+ tu "github.com/trickstercache/trickster/v2/pkg/testutil"
+)
+
+func TestQueryHandlerUnsupportedSearchUsesObjectProxyCache(t *testing.T) {
+ const originBody = `{"hits":{"hits":[]}}`
+ backendClient, err := NewClient("test", nil, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ts, w, r, _, err := tu.NewTestInstance("", backendClient.DefaultPathConfigs,
+ http.StatusOK, originBody, nil, providers.Elasticsearch, "/_search", "debug")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer ts.Close()
+
+ r.Method = http.MethodPost
+ request.SetBody(r, []byte(`{"query":{"match_all":{}}}`))
+ rsc := request.GetResources(r)
+ backendClient, err = NewClient("test", rsc.BackendOptions, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ client := backendClient.(*Client)
+ rsc.BackendClient = client
+ rsc.BackendOptions.HTTPClient = backendClient.HTTPClient()
+
+ client.QueryHandler(w, r)
+ resp := w.Result()
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("status = %d, want 200", resp.StatusCode)
+ }
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := string(body); got != originBody {
+ t.Fatalf("body = %s, want %s", got, originBody)
+ }
+}
+
+func TestQueryHandlerFallbackCacheSeparatesTimeRanges(t *testing.T) {
+ var originHits atomic.Int64
+ origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ originHits.Add(1)
+ body, err := io.ReadAll(r.Body)
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ w.Header().Set("Cache-Control", "public, max-age=60")
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(map[string]string{"request": string(body)})
+ }))
+ defer origin.Close()
+
+ prototype, err := NewClient("test", nil, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ testOrigin, _, baseRequest, _, err := tu.NewTestInstance("", prototype.DefaultPathConfigs,
+ http.StatusOK, `{}`, nil, providers.Elasticsearch, "/_search", "debug")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer testOrigin.Close()
+
+ base := request.GetResources(baseRequest)
+ base.BackendOptions.OriginURL = origin.URL
+ if err := base.BackendOptions.Initialize("test"); err != nil {
+ t.Fatal(err)
+ }
+ backendClient, err := NewClient("test", base.BackendOptions, nil, base.CacheClient, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ client := backendClient.(*Client)
+ base.BackendOptions.HTTPClient = backendClient.HTTPClient()
+
+ serve := func(body string) string {
+ t.Helper()
+ r := httptest.NewRequest(http.MethodPost, "/_search", strings.NewReader(body))
+ rsc := request.NewResources(base.BackendOptions, base.PathConfig, base.CacheConfig,
+ base.CacheClient, client, base.Tracer)
+ r = request.SetResources(r, rsc)
+ w := httptest.NewRecorder()
+ client.QueryHandler(w, r)
+ if w.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body=%s", w.Code, w.Body.String())
+ }
+ return w.Body.String()
+ }
+
+ firstRequest := `{"size":0,"query":{"range":{"@timestamp":` +
+ `{"gte":1704067200000,"lte":1704067800000,"format":"epoch_millis"}}}}`
+ secondRequest := `{"size":0,"query":{"range":{"@timestamp":` +
+ `{"gte":1704067800000,"lte":1704068400000,"format":"epoch_millis"}}}}`
+ first := serve(firstRequest)
+ second := serve(secondRequest)
+ firstCached := serve(firstRequest)
+
+ if first == second {
+ t.Fatalf("different time ranges returned the same cached body: %s", first)
+ }
+ if firstCached != first {
+ t.Fatalf("repeated first range did not use its cached response: first=%s cached=%s", first, firstCached)
+ }
+ if got := originHits.Load(); got != 2 {
+ t.Fatalf("origin requests = %d, want 2", got)
+ }
+}
+
+func TestQueryHandlerCachesOnlyCompleteHistogramBuckets(t *testing.T) {
+ type observedRange struct {
+ start int64
+ end int64
+ }
+ var mu sync.Mutex
+ var observed []observedRange
+ origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ var body map[string]any
+ if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+ t.Error(err)
+ return
+ }
+ rangeBody := firstTimestampRange(body)
+ start := int64(rangeBody["gte"].(float64))
+ end := int64(rangeBody["lte"].(float64))
+ mu.Lock()
+ observed = append(observed, observedRange{start: start, end: end})
+ mu.Unlock()
+
+ buckets := make([]map[string]any, 0)
+ for key := start; key <= end; key += time.Minute.Milliseconds() {
+ buckets = append(buckets, map[string]any{"key": key, "doc_count": 1})
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(map[string]any{
+ "took": 1,
+ "timed_out": false,
+ "_shards": map[string]any{"total": 1, "successful": 1, "failed": 0},
+ "hits": map[string]any{
+ "total": map[string]any{"value": len(buckets), "relation": "eq"},
+ "hits": []any{},
+ },
+ "aggregations": map[string]any{"2": map[string]any{"buckets": buckets}},
+ })
+ }))
+ defer origin.Close()
+
+ prototype, err := NewClient("test", nil, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ testOrigin, _, baseRequest, _, err := tu.NewTestInstance("", prototype.DefaultPathConfigs,
+ http.StatusOK, `{}`, nil, providers.Elasticsearch, "/_search", "debug")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer testOrigin.Close()
+
+ base := request.GetResources(baseRequest)
+ base.BackendOptions.OriginURL = origin.URL
+ if err := base.BackendOptions.Initialize("test"); err != nil {
+ t.Fatal(err)
+ }
+ backendClient, err := NewClient("test", base.BackendOptions, nil, base.CacheClient, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ client := backendClient.(*Client)
+ base.BackendOptions.HTTPClient = backendClient.HTTPClient()
+
+ start := time.Now().UTC().Add(-2 * time.Hour).Truncate(time.Minute).UnixMilli()
+ search := func(bucketCount int) string {
+ lastBucket := start + int64(bucketCount-1)*time.Minute.Milliseconds()
+ return fmt.Sprintf(`{"size":0,"query":{"bool":{"filter":[{"range":{"@timestamp":`+
+ `{"gte":%d,"lte":%d,"format":"epoch_millis"}}}]}},"aggs":{"2":{"date_histogram":`+
+ `{"field":"@timestamp","fixed_interval":"1m","min_doc_count":0,`+
+ `"extended_bounds":{"min":%d,"max":%d}}}}}`,
+ start, lastBucket+time.Minute.Milliseconds()-1, start, lastBucket)
+ }
+ serve := func(bucketCount int) map[string]any {
+ t.Helper()
+ r := httptest.NewRequest(http.MethodPost, "/_search", strings.NewReader(search(bucketCount)))
+ r.Header.Set("Content-Type", "application/json")
+ rsc := request.NewResources(base.BackendOptions, base.PathConfig, base.CacheConfig,
+ base.CacheClient, client, base.Tracer)
+ r = request.SetResources(r, rsc)
+ w := httptest.NewRecorder()
+ client.QueryHandler(w, r)
+ if w.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body=%s", w.Code, w.Body.String())
+ }
+ var response map[string]any
+ if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
+ t.Fatal(err)
+ }
+ return response
+ }
+
+ first := serve(3)
+ second := serve(4)
+ again := serve(3)
+ for _, tt := range []struct {
+ response map[string]any
+ want int
+ }{{first, 3}, {second, 4}, {again, 3}} {
+ buckets := tt.response["aggregations"].(map[string]any)["2"].(map[string]any)["buckets"].([]any)
+ if got := len(buckets); got != tt.want {
+ t.Fatalf("buckets = %d, want %d", got, tt.want)
+ }
+ total := tt.response["hits"].(map[string]any)["total"].(map[string]any)["value"].(float64)
+ if got := int(total); got != tt.want {
+ t.Fatalf("total hits = %d, want %d", got, tt.want)
+ }
+ }
+
+ mu.Lock()
+ defer mu.Unlock()
+ if got := len(observed); got != 2 {
+ t.Fatalf("origin requests = %d, want 2", got)
+ }
+ wantRanges := []observedRange{
+ {start: start, end: start + 3*time.Minute.Milliseconds() - 1},
+ {start: start + 3*time.Minute.Milliseconds(), end: start + 4*time.Minute.Milliseconds() - 1},
+ }
+ for i := range wantRanges {
+ if observed[i] != wantRanges[i] {
+ t.Fatalf("origin range %d = %+v, want %+v", i, observed[i], wantRanges[i])
+ }
+ }
+}
diff --git a/pkg/backends/elasticsearch/health.go b/pkg/backends/elasticsearch/health.go
new file mode 100644
index 000000000..53f9ad20d
--- /dev/null
+++ b/pkg/backends/elasticsearch/health.go
@@ -0,0 +1,29 @@
+/*
+ * 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 elasticsearch
+
+import ho "github.com/trickstercache/trickster/v2/pkg/backends/healthcheck/options"
+
+// DefaultHealthCheckConfig returns the default health check config for Elasticsearch.
+func (c *Client) DefaultHealthCheckConfig() *ho.Options {
+ o := ho.New()
+ u := c.BaseUpstreamURL()
+ o.Scheme = u.Scheme
+ o.Host = u.Host
+ o.Path = u.Path + "/_cluster/health"
+ return o
+}
diff --git a/pkg/backends/elasticsearch/model.go b/pkg/backends/elasticsearch/model.go
new file mode 100644
index 000000000..b6671a260
--- /dev/null
+++ b/pkg/backends/elasticsearch/model.go
@@ -0,0 +1,494 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "bytes"
+ "cmp"
+ "encoding/json"
+ "io"
+ "math"
+ "net/http"
+ "slices"
+ "strconv"
+ "time"
+
+ "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"
+)
+
+const (
+ esStatusSuccess = "success"
+ fieldBucketJSON = "bucket"
+)
+
+// NewModeler returns Elasticsearch modeling functions.
+func NewModeler() *timeseries.Modeler {
+ return ×eries.Modeler{
+ WireUnmarshalerReader: UnmarshalTimeseriesReader,
+ WireMarshaler: MarshalTimeseries,
+ WireMarshalWriter: MarshalTimeseriesWriter,
+ WireUnmarshaler: UnmarshalTimeseries,
+ CacheMarshaler: dataset.MarshalDataSet,
+ CacheUnmarshaler: dataset.UnmarshalDataSet,
+ }
+}
+
+// UnmarshalTimeseries converts an Elasticsearch response into a DataSet.
+func UnmarshalTimeseries(data []byte, trq *timeseries.TimeRangeQuery) (timeseries.Timeseries, error) {
+ return UnmarshalTimeseriesReader(bytes.NewReader(data), trq)
+}
+
+// UnmarshalTimeseriesReader converts an Elasticsearch response into a DataSet.
+func UnmarshalTimeseriesReader(reader io.Reader, trq *timeseries.TimeRangeQuery) (timeseries.Timeseries, error) {
+ if trq == nil {
+ return nil, timeseries.ErrNoTimerangeQuery
+ }
+ if reader == nil {
+ return nil, io.ErrUnexpectedEOF
+ }
+ plan, _ := trq.ParsedQuery.(*RequestPlan)
+ if plan == nil {
+ return nil, timeseries.ErrNoTimerangeQuery
+ }
+ switch plan.Kind {
+ case requestKindMSearch:
+ return unmarshalMSearchResponse(reader, trq, plan)
+ default:
+ return unmarshalSearchResponse(reader, trq, plan)
+ }
+}
+
+func unmarshalSearchResponse(reader io.Reader, trq *timeseries.TimeRangeQuery,
+ plan *RequestPlan,
+) (timeseries.Timeseries, error) {
+ if len(plan.Searches) != 1 || plan.Searches[0] == nil {
+ return nil, timeseries.ErrInvalidBody
+ }
+ var body map[string]json.RawMessage
+ if err := decodeSingleJSON(reader, &body); err != nil {
+ return nil, err
+ }
+ ds := newDataSet(trq)
+ result, err := resultFromSearchBody(body, trq, plan.Searches[0], 0)
+ if err != nil {
+ return nil, err
+ }
+ ds.Results = []*dataset.Result{result}
+ return ds, nil
+}
+
+func unmarshalMSearchResponse(reader io.Reader, trq *timeseries.TimeRangeQuery,
+ plan *RequestPlan,
+) (timeseries.Timeseries, error) {
+ var body struct {
+ Responses []json.RawMessage `json:"responses"`
+ }
+ if err := decodeSingleJSON(reader, &body); err != nil {
+ return nil, err
+ }
+ if len(body.Responses) != len(plan.Searches) {
+ return nil, timeseries.ErrInvalidBody
+ }
+ ds := newDataSet(trq)
+ ds.Results = make([]*dataset.Result, 0, len(body.Responses))
+ for i, raw := range body.Responses {
+ if i >= len(plan.Searches) {
+ break
+ }
+ var item map[string]json.RawMessage
+ if err := json.Unmarshal(raw, &item); err != nil {
+ return nil, err
+ }
+ result, err := resultFromSearchBody(item, trq, plan.Searches[i], i)
+ if err != nil {
+ return nil, err
+ }
+ ds.Results = append(ds.Results, result)
+ }
+ return ds, nil
+}
+
+func decodeSingleJSON(reader io.Reader, value any) error {
+ dec := json.NewDecoder(reader)
+ if err := dec.Decode(value); err != nil {
+ return err
+ }
+ var trailing any
+ if err := dec.Decode(&trailing); err != io.EOF {
+ return timeseries.ErrInvalidBody
+ }
+ return nil
+}
+
+func newDataSet(trq *timeseries.TimeRangeQuery) *dataset.DataSet {
+ return &dataset.DataSet{
+ Status: esStatusSuccess,
+ TimeRangeQuery: trq,
+ ExtentList: timeseries.ExtentList{trq.Extent},
+ }
+}
+
+func resultFromSearchBody(body map[string]json.RawMessage, trq *timeseries.TimeRangeQuery,
+ sp *SearchPlan, idx int,
+) (*dataset.Result, error) {
+ if sp == nil {
+ return nil, timeseries.ErrInvalidBody
+ }
+ if !isCompleteSearchResponse(body) {
+ return nil, timeseries.ErrInvalidBody
+ }
+ aggBody, err := aggregationResponse(body, sp.DateHistogramName)
+ if err != nil {
+ return nil, err
+ }
+ var agg struct {
+ Buckets []json.RawMessage `json:"buckets"`
+ }
+ if err := json.Unmarshal(aggBody, &agg); err != nil {
+ return nil, err
+ }
+ points := make(dataset.Points, 0, len(agg.Buckets))
+ seen := make(map[int64]struct{}, len(agg.Buckets))
+ for _, bucket := range agg.Buckets {
+ pt, timestamp, err := pointFromBucket(bucket)
+ if err != nil {
+ return nil, err
+ }
+ keyNs := timestamp.UnixNano()
+ if timestamp.Before(sp.Extent.Start) || timestamp.After(sp.Extent.End) ||
+ !timestamp.Equal(timestamp.Truncate(sp.Step)) {
+ return nil, timeseries.ErrInvalidBody
+ }
+ if _, exists := seen[keyNs]; exists {
+ return nil, timeseries.ErrInvalidBody
+ }
+ seen[keyNs] = struct{}{}
+ points = append(points, pt)
+ }
+ slices.SortFunc(points, func(a, b dataset.Point) int {
+ return cmp.Compare(a.Epoch, b.Epoch)
+ })
+ sh := dataset.SeriesHeader{
+ Name: sp.DateHistogramName,
+ Tags: dataset.Tags{
+ "aggregation": sp.DateHistogramName,
+ },
+ TimestampField: timeseries.FieldDefinition{
+ Name: sp.TimestampField,
+ DataType: timeseries.DateTimeUnixMilli,
+ Role: timeseries.RoleTimestamp,
+ },
+ ValueFieldsList: []timeseries.FieldDefinition{
+ {Name: fieldBucketJSON, DataType: timeseries.String, Role: timeseries.RoleValue},
+ },
+ QueryStatement: trq.Statement,
+ }
+ sh.CalculateSize()
+ return &dataset.Result{
+ StatementID: idx,
+ Name: sp.DateHistogramName,
+ SeriesList: []*dataset.Series{
+ {
+ Header: sh,
+ Points: points,
+ PointSize: points.Size(),
+ },
+ },
+ }, nil
+}
+
+func isCompleteSearchResponse(body map[string]json.RawMessage) bool {
+ raw, ok := body["timed_out"]
+ if !ok {
+ return false
+ }
+ var timedOut bool
+ if json.Unmarshal(raw, &timedOut) != nil || timedOut {
+ return false
+ }
+ if raw, ok := body["terminated_early"]; ok {
+ var terminatedEarly bool
+ if json.Unmarshal(raw, &terminatedEarly) != nil || terminatedEarly {
+ return false
+ }
+ }
+ raw, ok = body["_shards"]
+ if !ok || !hasCompleteShards(raw) {
+ return false
+ }
+ if raw, ok := body["_clusters"]; ok && !hasCompleteClusters(raw) {
+ return false
+ }
+ if raw, ok := body["hits"]; !ok || !hasSearchHits(raw) {
+ return false
+ }
+ if raw, ok := body["error"]; ok && len(bytes.TrimSpace(raw)) > 0 &&
+ !bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
+ return false
+ }
+ if raw, ok := body["status"]; ok {
+ var status int
+ if json.Unmarshal(raw, &status) != nil || status < http.StatusOK || status >= http.StatusMultipleChoices {
+ return false
+ }
+ }
+ return true
+}
+
+func hasCompleteShards(raw json.RawMessage) bool {
+ var shards struct {
+ Total *int `json:"total"`
+ Successful *int `json:"successful"`
+ Failed *int `json:"failed"`
+ }
+ return json.Unmarshal(raw, &shards) == nil && shards.Total != nil &&
+ shards.Successful != nil && shards.Failed != nil && *shards.Failed == 0 &&
+ *shards.Successful == *shards.Total
+}
+
+func hasCompleteClusters(raw json.RawMessage) bool {
+ var clusters struct {
+ Total *int `json:"total"`
+ Successful *int `json:"successful"`
+ Skipped int `json:"skipped"`
+ Running int `json:"running"`
+ Partial int `json:"partial"`
+ Failed int `json:"failed"`
+ }
+ return json.Unmarshal(raw, &clusters) == nil && clusters.Total != nil &&
+ clusters.Successful != nil && clusters.Skipped == 0 && clusters.Running == 0 &&
+ clusters.Partial == 0 && clusters.Failed == 0 &&
+ *clusters.Successful == *clusters.Total
+}
+
+func hasSearchHits(raw json.RawMessage) bool {
+ var hits map[string]json.RawMessage
+ if json.Unmarshal(raw, &hits) != nil || hits == nil {
+ return false
+ }
+ var documents []json.RawMessage
+ return json.Unmarshal(hits["hits"], &documents) == nil
+}
+
+func aggregationResponse(body map[string]json.RawMessage, name string) (json.RawMessage, error) {
+ var aggs map[string]json.RawMessage
+ if raw, ok := body[aggKeyAggregations]; ok {
+ if err := json.Unmarshal(raw, &aggs); err != nil {
+ return nil, err
+ }
+ } else if raw, ok := body[aggKeyAggs]; ok {
+ if err := json.Unmarshal(raw, &aggs); err != nil {
+ return nil, err
+ }
+ }
+ if aggs == nil {
+ return nil, timeseries.ErrInvalidBody
+ }
+ raw, ok := aggs[name]
+ if !ok {
+ return nil, timeseries.ErrInvalidBody
+ }
+ return raw, nil
+}
+
+func pointFromBucket(raw json.RawMessage) (dataset.Point, time.Time, error) {
+ var bucket map[string]any
+ dec := json.NewDecoder(bytes.NewReader(raw))
+ dec.UseNumber()
+ if err := dec.Decode(&bucket); err != nil {
+ return dataset.Point{}, time.Time{}, err
+ }
+ t, ok := bucketTime(bucket)
+ if !ok {
+ return dataset.Point{}, time.Time{}, timeseries.ErrInvalidBody
+ }
+ s := string(raw)
+ return dataset.Point{
+ Epoch: epoch.Epoch(t.UnixNano()),
+ Size: len(s) + 32,
+ Values: []any{s},
+ }, t, nil
+}
+
+func bucketTime(bucket map[string]any) (time.Time, bool) {
+ if v, ok := bucket["key"]; ok {
+ switch x := v.(type) {
+ case json.Number:
+ i, err := strconv.ParseInt(x.String(), 10, 64)
+ return time.UnixMilli(i), err == nil
+ case float64:
+ return time.UnixMilli(int64(x)), true
+ }
+ }
+ if v, ok := bucket["key_as_string"].(string); ok {
+ t, err := time.Parse(time.RFC3339Nano, v)
+ return t, err == nil
+ }
+ return time.Time{}, false
+}
+
+// MarshalTimeseries marshals a DataSet to Elasticsearch response JSON.
+func MarshalTimeseries(ts timeseries.Timeseries, rlo *timeseries.RequestOptions, status int) ([]byte, error) {
+ buf := bytes.NewBuffer(nil)
+ err := MarshalTimeseriesWriter(ts, rlo, status, buf)
+ return buf.Bytes(), err
+}
+
+// MarshalTimeseriesWriter marshals a DataSet to Elasticsearch response JSON.
+func MarshalTimeseriesWriter(ts timeseries.Timeseries, rlo *timeseries.RequestOptions,
+ status int, w io.Writer,
+) error {
+ if w == nil {
+ return io.ErrClosedPipe
+ }
+ ds, ok := ts.(*dataset.DataSet)
+ if !ok || ds == nil || rlo == nil {
+ return timeseries.ErrUnknownFormat
+ }
+ plan, _ := rlo.ProviderRequest.(*RequestPlan)
+ if plan == nil {
+ return timeseries.ErrUnknownFormat
+ }
+ if hw, ok := w.(http.ResponseWriter); ok && hw != nil {
+ hw.Header().Set(headers.NameContentType, headers.ValueApplicationJSON)
+ }
+ switch plan.Kind {
+ case requestKindMSearch:
+ return marshalMSearchResponse(ds, plan, status, w)
+ default:
+ if len(plan.Searches) != 1 || plan.Searches[0] == nil {
+ return timeseries.ErrUnknownFormat
+ }
+ return marshalSearchResponse(ds, plan.Searches[0], 0, status, w)
+ }
+}
+
+func marshalMSearchResponse(ds *dataset.DataSet, plan *RequestPlan, status int, w io.Writer) error {
+ if len(plan.Searches) == 0 {
+ return timeseries.ErrUnknownFormat
+ }
+ responses := make([]json.RawMessage, 0, len(plan.Searches))
+ for i, sp := range plan.Searches {
+ if sp == nil {
+ return timeseries.ErrUnknownFormat
+ }
+ b, err := marshalSearchResponseBytes(ds, sp, i, status, true)
+ if err != nil {
+ return err
+ }
+ responses = append(responses, b)
+ }
+ return json.NewEncoder(w).Encode(map[string]any{"responses": responses})
+}
+
+func marshalSearchResponse(ds *dataset.DataSet, sp *SearchPlan, statementID, status int, w io.Writer) error {
+ b, err := marshalSearchResponseBytes(ds, sp, statementID, status, false)
+ if err != nil {
+ return err
+ }
+ _, err = w.Write(b)
+ return err
+}
+
+func marshalSearchResponseBytes(ds *dataset.DataSet, sp *SearchPlan, statementID,
+ status int, includeStatus bool,
+) ([]byte, error) {
+ buckets, err := bucketsForStatement(ds, statementID)
+ if err != nil {
+ return nil, err
+ }
+ total, err := totalDocCount(buckets)
+ if err != nil {
+ return nil, err
+ }
+ if status == 0 {
+ status = http.StatusOK
+ }
+ aggregation := map[string]any{"buckets": buckets}
+ if sp.AggregationMeta != nil {
+ aggregation["meta"] = sp.AggregationMeta
+ }
+ out := map[string]any{
+ "took": 0,
+ "timed_out": false,
+ "hits": map[string]any{
+ "total": map[string]any{"value": total, "relation": "eq"},
+ "max_score": nil,
+ "hits": []any{},
+ },
+ "aggregations": map[string]any{
+ sp.DateHistogramName: aggregation,
+ },
+ }
+ if includeStatus || status != http.StatusOK {
+ out["status"] = status
+ }
+ return json.Marshal(out)
+}
+
+func totalDocCount(buckets []json.RawMessage) (int64, error) {
+ var total int64
+ for _, raw := range buckets {
+ var bucket struct {
+ DocCount json.Number `json:"doc_count"`
+ }
+ dec := json.NewDecoder(bytes.NewReader(raw))
+ dec.UseNumber()
+ if err := dec.Decode(&bucket); err != nil {
+ return 0, err
+ }
+ count, err := strconv.ParseInt(bucket.DocCount.String(), 10, 64)
+ if err != nil || count < 0 || total > math.MaxInt64-count {
+ return 0, timeseries.ErrInvalidBody
+ }
+ total += count
+ }
+ return total, nil
+}
+
+func bucketsForStatement(ds *dataset.DataSet, statementID int) ([]json.RawMessage, error) {
+ for _, r := range ds.Results {
+ if r == nil || r.StatementID != statementID {
+ continue
+ }
+ for _, s := range r.SeriesList {
+ if s == nil {
+ continue
+ }
+ pts := s.Points.Clone()
+ slices.SortFunc(pts, func(a, b dataset.Point) int {
+ return cmp.Compare(a.Epoch, b.Epoch)
+ })
+ out := make([]json.RawMessage, 0, len(pts))
+ for _, p := range pts {
+ if len(p.Values) == 0 {
+ continue
+ }
+ raw, ok := p.Values[0].(string)
+ if !ok {
+ continue
+ }
+ out = append(out, json.RawMessage(raw))
+ }
+ return out, nil
+ }
+ }
+ return []json.RawMessage{}, nil
+}
diff --git a/pkg/backends/elasticsearch/model_test.go b/pkg/backends/elasticsearch/model_test.go
new file mode 100644
index 000000000..93809b1b4
--- /dev/null
+++ b/pkg/backends/elasticsearch/model_test.go
@@ -0,0 +1,204 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "bytes"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+
+ "github.com/trickstercache/trickster/v2/pkg/timeseries"
+ "github.com/trickstercache/trickster/v2/pkg/timeseries/dataset"
+)
+
+const searchResponse = `{
+ "took": 1,
+ "timed_out": false,
+ "_shards": {
+ "total": 2,
+ "successful": 2,
+ "skipped": 0,
+ "failed": 0
+ },
+ "hits": {
+ "total": {
+ "value": 0,
+ "relation": "eq"
+ },
+ "hits": []
+ },
+ "aggregations": {
+ "2": {
+ "meta": {
+ "source": "query"
+ },
+ "buckets": [
+ {
+ "key": 1704067200000,
+ "doc_count": 1,
+ "1": {
+ "value": 10
+ }
+ },
+ {
+ "key": 1704067260000,
+ "doc_count": 2,
+ "1": {
+ "value": 20
+ }
+ }
+ ]
+ }
+ }
+}`
+
+func TestUnmarshalMarshalSearchResponse(t *testing.T) {
+ trq, ro := parseSearchForModelTest(t)
+ ts, err := UnmarshalTimeseries([]byte(searchResponse), trq)
+ if err != nil {
+ t.Fatalf("UnmarshalTimeseries returned error: %v", err)
+ }
+ ds := ts.(*dataset.DataSet)
+ if got := ds.ValueCount(); got != 2 {
+ t.Fatalf("value count = %d, want 2", got)
+ }
+ var buf bytes.Buffer
+ if err := MarshalTimeseriesWriter(ts, ro, http.StatusOK, &buf); err != nil {
+ t.Fatalf("MarshalTimeseriesWriter returned error: %v", err)
+ }
+ var out map[string]any
+ if err := json.Unmarshal(buf.Bytes(), &out); err != nil {
+ t.Fatal(err)
+ }
+ buckets := out["aggregations"].(map[string]any)["2"].(map[string]any)["buckets"].([]any)
+ if len(buckets) != 2 {
+ t.Fatalf("buckets = %d, want 2", len(buckets))
+ }
+ if got := int64(buckets[0].(map[string]any)["key"].(float64)); got != 1704067200000 {
+ t.Fatalf("first key = %d, want 1704067200000", got)
+ }
+ meta := out["aggregations"].(map[string]any)["2"].(map[string]any)["meta"].(map[string]any)
+ if got := meta["source"]; got != "query" {
+ t.Fatalf("aggregation meta source = %v, want query", got)
+ }
+ total := out["hits"].(map[string]any)["total"].(map[string]any)["value"].(float64)
+ if got, want := int64(total), int64(3); got != want {
+ t.Fatalf("total hits = %d, want %d", got, want)
+ }
+}
+
+func TestUnmarshalMarshalMSearchResponse(t *testing.T) {
+ trq, ro := parseMSearchForModelTest(t)
+ resp := `{"responses":[` + searchResponse + `,` + searchResponse + `]}`
+ ts, err := UnmarshalTimeseries([]byte(resp), trq)
+ if err != nil {
+ t.Fatalf("UnmarshalTimeseries returned error: %v", err)
+ }
+ var buf bytes.Buffer
+ if err := MarshalTimeseriesWriter(ts, ro, http.StatusOK, &buf); err != nil {
+ t.Fatalf("MarshalTimeseriesWriter returned error: %v", err)
+ }
+ var out map[string][]map[string]any
+ if err := json.Unmarshal(buf.Bytes(), &out); err != nil {
+ t.Fatal(err)
+ }
+ if got := len(out["responses"]); got != 2 {
+ t.Fatalf("responses = %d, want 2", got)
+ }
+ if got := int(out["responses"][0]["status"].(float64)); got != http.StatusOK {
+ t.Fatalf("response status = %d, want 200", got)
+ }
+}
+
+func TestUnmarshalMSearchResponseRequiresOneResponsePerSearch(t *testing.T) {
+ trq, _ := parseMSearchForModelTest(t)
+ resp := `{"responses":[` + searchResponse + `]}`
+ if _, err := UnmarshalTimeseries([]byte(resp), trq); err == nil {
+ t.Fatal("expected response-count mismatch to fail instead of fabricating an empty search result")
+ }
+}
+
+func TestUnmarshalTimeseriesRejectsPartialSearchResponses(t *testing.T) {
+ trq, _ := parseSearchForModelTest(t)
+ for _, body := range []string{
+ strings.Replace(searchResponse, `"timed_out": false`, `"timed_out": true`, 1),
+ strings.Replace(searchResponse, `"successful": 2`, `"successful": 1`, 1),
+ strings.Replace(searchResponse, `"failed": 0`, `"failed": 1`, 1),
+ strings.Replace(searchResponse, `"timed_out": false`,
+ `"timed_out": false, "terminated_early": true`, 1),
+ strings.Replace(searchResponse, `"timed_out": false`,
+ `"timed_out": false, "_clusters": {"total": 2, "successful": 1, "skipped": 1}`, 1),
+ strings.Replace(searchResponse, ` "_shards": {`+
+ "\n"+` "total": 2,`+"\n"+` "successful": 2,`+"\n"+
+ ` "skipped": 0,`+"\n"+` "failed": 0`+"\n"+` },`+"\n", "", 1),
+ } {
+ if _, err := UnmarshalTimeseries([]byte(body), trq); err == nil {
+ t.Fatal("expected partial Elasticsearch response to be rejected")
+ }
+ }
+}
+
+func TestUnmarshalTimeseriesRejectsTrailingResponseJSON(t *testing.T) {
+ trq, _ := parseSearchForModelTest(t)
+ if _, err := UnmarshalTimeseries([]byte(searchResponse+` {"extra":true}`), trq); err == nil {
+ t.Fatal("expected trailing response JSON to be rejected")
+ }
+}
+
+func TestUnmarshalTimeseriesRejectsInvalidBucketKeys(t *testing.T) {
+ trq, _ := parseSearchForModelTest(t)
+ for _, body := range []string{
+ strings.Replace(searchResponse, `"key": 1704067200000`, `"key": 1704067200001`, 1),
+ strings.Replace(searchResponse, `"key": 1704067200000`, `"key": 1704067140000`, 1),
+ strings.Replace(searchResponse, `"key": 1704067260000`, `"key": 1704067200000`, 1),
+ } {
+ if _, err := UnmarshalTimeseries([]byte(body), trq); err == nil {
+ t.Fatal("expected invalid or duplicate bucket key to be rejected")
+ }
+ }
+}
+
+func parseSearchForModelTest(t *testing.T) (*timeseries.TimeRangeQuery, *timeseries.RequestOptions) {
+ t.Helper()
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", strings.NewReader(searchBody))
+ trq, ro, _, err := c.ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return trq, ro
+}
+
+func parseMSearchForModelTest(t *testing.T) (*timeseries.TimeRangeQuery, *timeseries.RequestOptions) {
+ t.Helper()
+ searchLine := compactJSON(t, searchBody)
+ var body bytes.Buffer
+ body.WriteString(`{"index":"metrics"}` + "\n")
+ body.WriteString(searchLine + "\n")
+ body.WriteString(`{"index":"metrics"}` + "\n")
+ body.WriteString(searchLine + "\n")
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodPost, "/_msearch", &body)
+ trq, ro, _, err := c.ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return trq, ro
+}
diff --git a/pkg/backends/elasticsearch/options/options.go b/pkg/backends/elasticsearch/options/options.go
new file mode 100644
index 000000000..aadf87636
--- /dev/null
+++ b/pkg/backends/elasticsearch/options/options.go
@@ -0,0 +1,60 @@
+/*
+ * 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 options
+
+import "github.com/trickstercache/trickster/v2/pkg/config/types"
+
+const (
+ // DefaultTimestampField is the Elasticsearch timestamp field used when a
+ // backend does not provide an explicit timestamp_field.
+ DefaultTimestampField = "@timestamp"
+)
+
+// Options holds Elasticsearch-specific backend options.
+type Options struct {
+ // TimestampField is the Elasticsearch date field Trickster uses to detect,
+ // normalize, and rewrite time range filters for date_histogram requests.
+ TimestampField string `yaml:"timestamp_field,omitempty"`
+}
+
+var _ types.ConfigOptions[Options] = &Options{}
+
+// New returns a default Elasticsearch options object.
+func New() *Options {
+ return &Options{TimestampField: DefaultTimestampField}
+}
+
+// Clone returns an exact copy of the subject Options.
+func (o *Options) Clone() *Options {
+ if o == nil {
+ return nil
+ }
+ return &Options{TimestampField: o.TimestampField}
+}
+
+// Initialize sets option defaults.
+func (o *Options) Initialize(_ string) error {
+ if o.TimestampField == "" {
+ o.TimestampField = DefaultTimestampField
+ }
+ return nil
+}
+
+// Validate validates the Elasticsearch options.
+func (o *Options) Validate() (bool, error) {
+ return true, nil
+}
diff --git a/pkg/backends/elasticsearch/path.go b/pkg/backends/elasticsearch/path.go
new file mode 100644
index 000000000..da779963b
--- /dev/null
+++ b/pkg/backends/elasticsearch/path.go
@@ -0,0 +1,32 @@
+/*
+ * 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 elasticsearch
+
+import "strings"
+
+const (
+ pathSearch = "/_search"
+ pathMSearch = "/_msearch"
+)
+
+func isSearchPath(path string) bool {
+ return path == pathSearch || strings.HasSuffix(path, pathSearch)
+}
+
+func isMSearchPath(path string) bool {
+ return path == pathMSearch || strings.HasSuffix(path, pathMSearch)
+}
diff --git a/pkg/backends/elasticsearch/query.go b/pkg/backends/elasticsearch/query.go
new file mode 100644
index 000000000..2b60ba697
--- /dev/null
+++ b/pkg/backends/elasticsearch/query.go
@@ -0,0 +1,1017 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "bytes"
+ "encoding/json"
+ stderrors "errors"
+ "fmt"
+ "io"
+ "maps"
+ "net/http"
+ "slices"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/trickstercache/trickster/v2/pkg/proxy/errors"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/methods"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/request"
+ "github.com/trickstercache/trickster/v2/pkg/timeseries"
+)
+
+const (
+ queryParamSource = "source"
+
+ aggKeyAggs = "aggs"
+ aggKeyAggregations = "aggregations"
+
+ rangeStartToken = "<$START$>"
+ rangeEndToken = "<$END$>"
+)
+
+type requestKind byte
+
+const (
+ requestKindSearch requestKind = iota
+ requestKindMSearch
+)
+
+// RequestPlan is the parsed provider request used by SetExtent and response marshaling.
+type RequestPlan struct {
+ Kind requestKind
+ Searches []*SearchPlan
+ SourceBody bool
+}
+
+// SearchPlan is one Elasticsearch search body inside a request.
+type SearchPlan struct {
+ Header map[string]any
+ Body map[string]any
+ AggregationMeta map[string]any
+ DateHistogramName string
+ TimestampField string
+ TimestampValueKind timestampValueKind
+ RangeEndExclusive bool
+ Statement string
+ Extent timeseries.Extent
+ Step time.Duration
+}
+
+type timestampValueKind byte
+
+const (
+ timestampValueRFC3339 timestampValueKind = iota
+ timestampValueEpochSeconds
+ timestampValueEpochMillis
+)
+
+func (c *Client) ParseTimeRangeQuery(r *http.Request) (*timeseries.TimeRangeQuery,
+ *timeseries.RequestOptions, bool, error,
+) {
+ if r == nil || r.URL == nil {
+ return nil, nil, false, errors.ErrNotTimeRangeQuery
+ }
+ isSearch := isSearchPath(r.URL.Path)
+ isMSearch := isMSearchPath(r.URL.Path)
+ if !isSearch && !isMSearch {
+ return nil, nil, false, errors.ErrNotTimeRangeQuery
+ }
+
+ body, sourceBody, err := readSearchBody(r)
+ if err != nil {
+ return nil, nil, false, err
+ }
+ if len(bytes.TrimSpace(body)) == 0 {
+ return nil, nil, false, errors.ErrNotTimeRangeQuery
+ }
+
+ opts := c.elasticsearchOptions()
+ plan, trq, ro, err := parseRequestPlan(body, sourceBody, isMSearch, opts.TimestampField)
+ if err == nil && hasUnsafeSearchQueryParams(r, sourceBody) {
+ setFallbackCacheKey(trq, body, sourceBody)
+ err = errors.ErrNotTimeRangeQuery
+ }
+ if trq != nil {
+ trq.OriginalBody = body
+ trq.ParsedQuery = plan
+ }
+ if ro != nil {
+ ro.ProviderRequest = plan
+ }
+ return trq, ro, true, err
+}
+
+func readSearchBody(r *http.Request) ([]byte, bool, error) {
+ if methods.HasBody(r.Method) {
+ body, err := request.GetBody(r)
+ if err != nil && !stderrors.Is(err, io.EOF) {
+ return nil, false, err
+ }
+ if len(bytes.TrimSpace(body)) > 0 {
+ return body, false, nil
+ }
+ }
+ if r.Method == http.MethodGet && r.Body != nil {
+ body, err := io.ReadAll(r.Body)
+ if err != nil {
+ return nil, false, err
+ }
+ request.SetBody(r, body)
+ if len(bytes.TrimSpace(body)) > 0 {
+ return body, false, nil
+ }
+ }
+ if source := r.URL.Query().Get(queryParamSource); source != "" {
+ return []byte(source), true, nil
+ }
+ return nil, false, nil
+}
+
+func parseRequestPlan(body []byte, sourceBody, isMSearch bool, timestampField string) (*RequestPlan,
+ *timeseries.TimeRangeQuery, *timeseries.RequestOptions, error,
+) {
+ if isMSearch {
+ plan, trq, ro, err := parseMSearchPlan(body, timestampField)
+ if err != nil {
+ if plan == nil {
+ plan = &RequestPlan{Kind: requestKindMSearch}
+ }
+ trq = timeRangeQueryFromSearches(plan.Searches, exactRequestStatement(body))
+ ro = ×eries.RequestOptions{ProviderRequest: plan}
+ }
+ return plan, trq, ro, err
+ }
+ sp, normalized, err := parseSearchPlan(nil, body, timestampField)
+ plan := &RequestPlan{Kind: requestKindSearch, Searches: []*SearchPlan{sp}, SourceBody: sourceBody}
+ statement := normalized
+ if err != nil {
+ statement = exactRequestStatement(body)
+ }
+ trq := timeRangeQueryFromSearches(plan.Searches, statement)
+ setStatementCacheKey(trq, sourceBody)
+ ro := ×eries.RequestOptions{ProviderRequest: plan}
+ if err != nil {
+ return plan, trq, ro, err
+ }
+ return plan, trq, ro, nil
+}
+
+func parseMSearchPlan(body []byte, timestampField string) (*RequestPlan,
+ *timeseries.TimeRangeQuery, *timeseries.RequestOptions, error,
+) {
+ lines := splitNDJSON(body)
+ if len(lines) == 0 || len(lines)%2 != 0 {
+ return nil, nil, nil, timeseries.ErrInvalidBody
+ }
+ plan := &RequestPlan{Kind: requestKindMSearch, Searches: make([]*SearchPlan, 0, len(lines)/2)}
+ normalizedParts := make([]string, 0, len(lines))
+ for i := 0; i < len(lines); i += 2 {
+ header, err := decodeObject(lines[i])
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ sp, normalized, err := parseSearchPlan(header, lines[i+1], timestampField)
+ if sp != nil {
+ plan.Searches = append(plan.Searches, sp)
+ }
+ hb, _ := canonicalJSON(header)
+ normalizedParts = append(normalizedParts, string(hb), normalized)
+ if err != nil {
+ trq := timeRangeQueryFromSearches(plan.Searches, strings.Join(normalizedParts, "\n"))
+ return plan, trq, ×eries.RequestOptions{ProviderRequest: plan}, err
+ }
+ }
+ normalized := strings.Join(normalizedParts, "\n")
+ trq := timeRangeQueryFromSearches(plan.Searches, normalized)
+ ro := ×eries.RequestOptions{ProviderRequest: plan}
+ if err := validateCommonRange(plan.Searches); err != nil {
+ return plan, trq, ro, err
+ }
+ return plan, trq, ro, nil
+}
+
+func splitNDJSON(body []byte) [][]byte {
+ raw := bytes.Split(body, []byte{'\n'})
+ out := make([][]byte, 0, len(raw))
+ for i, line := range raw {
+ line = bytes.TrimSpace(line)
+ if len(line) == 0 {
+ if i == len(raw)-1 {
+ continue
+ }
+ return nil
+ }
+ out = append(out, line)
+ }
+ return out
+}
+
+func parseSearchPlan(header map[string]any, body []byte, timestampField string) (*SearchPlan, string, error) {
+ m, err := decodeObject(body)
+ if err != nil {
+ return nil, "", err
+ }
+ sp := &SearchPlan{Header: header, Body: m, TimestampField: timestampField}
+ if !isAggregationOnlySearch(m) {
+ return sp, normalizedBodyForCache(m, timestampField), errors.ErrNotTimeRangeQuery
+ }
+ documentExtent, kind, endExclusive, ok := extractTimeRange(m, timestampField)
+ if !ok {
+ normalized := normalizedBodyForCache(m, timestampField)
+ return sp, normalized, errors.ErrNotTimeRangeQuery
+ }
+ aggName, step, histogram, ok := extractDateHistogram(m, timestampField)
+ if !ok {
+ normalized := normalizedBodyForCache(m, timestampField)
+ return sp, normalized, errors.ErrNotTimeRangeQuery
+ }
+ extent, ok := completeBucketExtent(documentExtent, step, kind, endExclusive)
+ if !ok || !histogramSupportsIndependentExtents(histogram, extent, step) {
+ normalized := normalizedBodyForCache(m, timestampField)
+ return sp, normalized, errors.ErrNotTimeRangeQuery
+ }
+ sp.Extent = extent
+ sp.Step = step
+ sp.TimestampValueKind = kind
+ sp.RangeEndExclusive = endExclusive
+ sp.DateHistogramName = aggName
+ sp.AggregationMeta = aggregationMeta(m, aggName)
+ normalized := normalizedBodyForCache(m, timestampField)
+ sp.Statement = normalized
+ return sp, normalized, nil
+}
+
+func decodeObject(data []byte) (map[string]any, error) {
+ var m map[string]any
+ dec := json.NewDecoder(bytes.NewReader(data))
+ dec.UseNumber()
+ if err := dec.Decode(&m); err != nil {
+ return nil, err
+ }
+ if m == nil {
+ return nil, timeseries.ErrInvalidBody
+ }
+ var trailing any
+ if err := dec.Decode(&trailing); err != io.EOF {
+ return nil, timeseries.ErrInvalidBody
+ }
+ return m, nil
+}
+
+func timeRangeQueryFromSearches(searches []*SearchPlan, statement string) *timeseries.TimeRangeQuery {
+ trq := ×eries.TimeRangeQuery{
+ Statement: statement,
+ CacheKeyElements: map[string]string{
+ "query": statement,
+ },
+ }
+ if len(searches) == 0 || searches[0] == nil {
+ return trq
+ }
+ trq.Extent = searches[0].Extent
+ trq.Step = searches[0].Step
+ trq.TimestampDefinition = timeseries.FieldDefinition{
+ Name: searches[0].TimestampField,
+ DataType: timeseries.DateTimeUnixMilli,
+ Role: timeseries.RoleTimestamp,
+ }
+ return trq
+}
+
+func exactRequestStatement(body []byte) string {
+ return string(bytes.TrimSpace(body))
+}
+
+func setFallbackCacheKey(trq *timeseries.TimeRangeQuery, body []byte, sourceBody bool) {
+ if trq == nil {
+ return
+ }
+ trq.Statement = exactRequestStatement(body)
+ trq.CacheKeyElements = map[string]string{"query": trq.Statement}
+ setStatementCacheKey(trq, sourceBody)
+}
+
+func setStatementCacheKey(trq *timeseries.TimeRangeQuery, sourceBody bool) {
+ if trq == nil || !sourceBody {
+ return
+ }
+ delete(trq.CacheKeyElements, "query")
+ trq.CacheKeyElements[queryParamSource] = trq.Statement
+}
+
+func isAggregationOnlySearch(root map[string]any) bool {
+ size, ok := root["size"].(json.Number)
+ if !ok {
+ return false
+ }
+ n, err := strconv.ParseInt(size.String(), 10, 64)
+ if err != nil || n != 0 {
+ return false
+ }
+ for key := range root {
+ switch key {
+ case "size", "query", aggKeyAggs, aggKeyAggregations, "runtime_mappings":
+ default:
+ return false
+ }
+ }
+ return true
+}
+
+func hasUnsafeSearchQueryParams(r *http.Request, sourceBody bool) bool {
+ if r == nil || r.URL == nil {
+ return true
+ }
+ for key := range r.URL.Query() {
+ switch key {
+ case "allow_no_indices", "allow_partial_search_results", "batched_reduce_size",
+ "ccs_minimize_roundtrips", "error_trace", "expand_wildcards", "human",
+ "ignore_throttled", "ignore_unavailable", "max_concurrent_shard_requests",
+ "pre_filter_shard_size", "preference", "pretty", "request_cache", "routing",
+ "source_content_type":
+ case queryParamSource:
+ if !sourceBody {
+ return true
+ }
+ default:
+ return true
+ }
+ }
+ return false
+}
+
+func validateCommonRange(searches []*SearchPlan) error {
+ if len(searches) == 0 || searches[0] == nil {
+ return errors.ErrNotTimeRangeQuery
+ }
+ first := searches[0]
+ for _, sp := range searches[1:] {
+ if sp == nil || sp.Step != first.Step ||
+ !sp.Extent.Start.Equal(first.Extent.Start) ||
+ !sp.Extent.End.Equal(first.Extent.End) {
+ return errors.ErrNotTimeRangeQuery
+ }
+ }
+ return nil
+}
+
+func extractTimeRange(root map[string]any, timestampField string) (timeseries.Extent,
+ timestampValueKind, bool, bool,
+) {
+ query, ok := root["query"]
+ if !ok {
+ return timeseries.Extent{}, timestampValueRFC3339, false, false
+ }
+ ranges := requiredTimestampRanges(query, timestampField)
+ if len(ranges) != 1 || countTimestampRanges(root, timestampField) != 1 {
+ return timeseries.Extent{}, timestampValueRFC3339, false, false
+ }
+ fieldNode := ranges[0]
+ if hasAnyKey(fieldNode, "gt", "from", "to", "time_zone", "relation") {
+ return timeseries.Extent{}, timestampValueRFC3339, false, false
+ }
+ _, hasLTE := fieldNode["lte"]
+ _, hasLT := fieldNode["lt"]
+ if hasLTE == hasLT {
+ return timeseries.Extent{}, timestampValueRFC3339, false, false
+ }
+ start, kind, ok := parseRangeTime(fieldNode["gte"], fieldNode)
+ if !ok {
+ return timeseries.Extent{}, timestampValueRFC3339, false, false
+ }
+ endValue := fieldNode["lte"]
+ if hasLT {
+ endValue = fieldNode["lt"]
+ }
+ end, endKind, ok := parseRangeTime(endValue, fieldNode)
+ if !ok || endKind != kind || !start.Before(end) {
+ return timeseries.Extent{}, timestampValueRFC3339, false, false
+ }
+ return timeseries.Extent{Start: start, End: end}, kind, hasLT, true
+}
+
+func requiredTimestampRanges(v any, timestampField string) []map[string]any {
+ m, ok := v.(map[string]any)
+ if !ok {
+ return nil
+ }
+ if rangeNode, ok := m["range"].(map[string]any); ok {
+ if fieldNode, ok := rangeNode[timestampField].(map[string]any); ok {
+ return []map[string]any{fieldNode}
+ }
+ return nil
+ }
+ var out []map[string]any
+ if boolNode, ok := m["bool"].(map[string]any); ok {
+ for _, key := range []string{"filter", "must"} {
+ out = append(out, requiredTimestampRangesFromClauses(boolNode[key], timestampField)...)
+ }
+ }
+ if constantScore, ok := m["constant_score"].(map[string]any); ok {
+ out = append(out, requiredTimestampRanges(constantScore["filter"], timestampField)...)
+ }
+ if functionScore, ok := m["function_score"].(map[string]any); ok {
+ out = append(out, requiredTimestampRanges(functionScore["query"], timestampField)...)
+ }
+ return out
+}
+
+func requiredTimestampRangesFromClauses(v any, timestampField string) []map[string]any {
+ switch clauses := v.(type) {
+ case []any:
+ var out []map[string]any
+ for _, clause := range clauses {
+ out = append(out, requiredTimestampRanges(clause, timestampField)...)
+ }
+ return out
+ default:
+ return requiredTimestampRanges(v, timestampField)
+ }
+}
+
+func countTimestampRanges(root map[string]any, timestampField string) int {
+ var count int
+ walkMaps(root, func(m map[string]any) bool {
+ rangeNode, ok := m["range"].(map[string]any)
+ if !ok {
+ return true
+ }
+ if _, ok := rangeNode[timestampField].(map[string]any); ok {
+ count++
+ }
+ return true
+ })
+ return count
+}
+
+func completeBucketExtent(documentExtent timeseries.Extent, step time.Duration,
+ kind timestampValueKind, endExclusive bool,
+) (timeseries.Extent, bool) {
+ if step < time.Millisecond || (24*time.Hour)%step != 0 ||
+ !documentExtent.Start.Equal(documentExtent.Start.Truncate(step)) {
+ return timeseries.Extent{}, false
+ }
+ if kind == timestampValueEpochSeconds && step%time.Second != 0 {
+ return timeseries.Extent{}, false
+ }
+ endBoundary := documentExtent.End
+ if !endExclusive {
+ endBoundary = endBoundary.Add(time.Millisecond)
+ }
+ if !endBoundary.Equal(endBoundary.Truncate(step)) || !endBoundary.After(documentExtent.Start) {
+ return timeseries.Extent{}, false
+ }
+ return timeseries.Extent{
+ Start: documentExtent.Start,
+ End: endBoundary.Add(-step),
+ }, true
+}
+
+func histogramSupportsIndependentExtents(histogram map[string]any, extent timeseries.Extent,
+ step time.Duration,
+) bool {
+ minDocCount := int64(0)
+ if raw, exists := histogram["min_doc_count"]; exists {
+ n, ok := raw.(json.Number)
+ if !ok {
+ return false
+ }
+ var err error
+ minDocCount, err = strconv.ParseInt(n.String(), 10, 64)
+ if err != nil || minDocCount < 0 || minDocCount > 1 {
+ return false
+ }
+ }
+
+ for _, key := range []string{"extended_bounds", "hard_bounds"} {
+ raw, exists := histogram[key]
+ if !exists {
+ if key == "extended_bounds" && minDocCount == 0 {
+ return false
+ }
+ continue
+ }
+ bounds, ok := raw.(map[string]any)
+ if !ok {
+ return false
+ }
+ if key == "extended_bounds" && minDocCount == 0 &&
+ (!hasAnyKey(bounds, "min") || !hasAnyKey(bounds, "max")) {
+ return false
+ }
+ for bound, expected := range map[string]time.Time{"min": extent.Start, "max": extent.End} {
+ value, exists := bounds[bound]
+ if !exists {
+ continue
+ }
+ parsed, _, ok := parseRangeTime(value, histogram)
+ if !ok || !parsed.Truncate(step).Equal(expected) {
+ return false
+ }
+ }
+ }
+ return true
+}
+
+func hasAnyKey(m map[string]any, keys ...string) bool {
+ for _, key := range keys {
+ if _, ok := m[key]; ok {
+ return true
+ }
+ }
+ return false
+}
+
+func parseRangeTime(v any, fieldNode map[string]any) (time.Time, timestampValueKind, bool) {
+ if v == nil {
+ return time.Time{}, timestampValueRFC3339, false
+ }
+ format, _ := fieldNode["format"].(string)
+ switch x := v.(type) {
+ case json.Number:
+ return parseNumericTime(x.String(), format)
+ case float64:
+ return parseNumericTime(strconv.FormatFloat(x, 'f', -1, 64), format)
+ case int64:
+ return parseNumericTime(strconv.FormatInt(x, 10), format)
+ case string:
+ if t, err := time.Parse(time.RFC3339Nano, x); err == nil {
+ return t, timestampValueRFC3339, true
+ }
+ return parseNumericTime(x, format)
+ }
+ return time.Time{}, timestampValueRFC3339, false
+}
+
+func parseNumericTime(input, format string) (time.Time, timestampValueKind, bool) {
+ i, err := strconv.ParseInt(input, 10, 64)
+ if err != nil {
+ return time.Time{}, timestampValueRFC3339, false
+ }
+ secondPos := strings.Index(format, "epoch_second")
+ millisPos := strings.Index(format, "epoch_millis")
+ if secondPos >= 0 && (millisPos < 0 || secondPos < millisPos) {
+ return time.Unix(i, 0), timestampValueEpochSeconds, true
+ }
+ return time.UnixMilli(i), timestampValueEpochMillis, true
+}
+
+func extractDateHistogram(root map[string]any, timestampField string) (string, time.Duration,
+ map[string]any, bool,
+) {
+ _, hasAggs := root[aggKeyAggs]
+ _, hasAggregations := root[aggKeyAggregations]
+ if hasAggs && hasAggregations {
+ return "", 0, nil, false
+ }
+ aggs, ok := aggregationMap(root)
+ if !ok || len(aggs) != 1 {
+ return "", 0, nil, false
+ }
+ keys := make([]string, 0, len(aggs))
+ for key := range aggs {
+ keys = append(keys, key)
+ }
+ slices.Sort(keys)
+ for _, key := range keys {
+ agg, ok := aggs[key].(map[string]any)
+ if !ok {
+ continue
+ }
+ if meta, exists := agg["meta"]; exists {
+ if _, ok := meta.(map[string]any); !ok {
+ continue
+ }
+ }
+ dh, ok := agg["date_histogram"].(map[string]any)
+ if !ok {
+ continue
+ }
+ if field, _ := dh["field"].(string); field != timestampField {
+ continue
+ }
+ if keyed, _ := dh["keyed"].(bool); keyed {
+ continue
+ }
+ if _, hasOffset := dh["offset"]; hasOffset {
+ continue
+ }
+ if _, hasScript := dh["script"]; hasScript {
+ continue
+ }
+ if !hasAscendingKeyOrder(dh) || containsPipelineAggregation(agg) {
+ continue
+ }
+ if tzValue, hasTimeZone := dh["time_zone"]; hasTimeZone {
+ tz, ok := tzValue.(string)
+ if !ok || !isUTCTimeZone(tz) {
+ continue
+ }
+ }
+ step, ok := parseHistogramInterval(dh)
+ if ok {
+ return key, step, dh, true
+ }
+ }
+ return "", 0, nil, false
+}
+
+func aggregationMeta(root map[string]any, name string) map[string]any {
+ aggs, ok := aggregationMap(root)
+ if !ok {
+ return nil
+ }
+ agg, ok := aggs[name].(map[string]any)
+ if !ok {
+ return nil
+ }
+ meta, _ := agg["meta"].(map[string]any)
+ if meta == nil {
+ return nil
+ }
+ return cloneMap(meta)
+}
+
+func hasAscendingKeyOrder(histogram map[string]any) bool {
+ value, exists := histogram["order"]
+ if !exists {
+ return true
+ }
+ order, ok := value.(map[string]any)
+ if !ok || len(order) != 1 {
+ return false
+ }
+ direction, ok := order["_key"].(string)
+ return ok && strings.EqualFold(direction, "asc")
+}
+
+func containsPipelineAggregation(aggregation map[string]any) bool {
+ nested, ok := aggregationMap(aggregation)
+ if !ok {
+ return false
+ }
+ for _, raw := range nested {
+ definition, ok := raw.(map[string]any)
+ if !ok {
+ return true
+ }
+ if isPipelineAggregation(definition) || containsPipelineAggregation(definition) {
+ return true
+ }
+ }
+ return false
+}
+
+func isPipelineAggregation(definition map[string]any) bool {
+ for key := range definition {
+ switch key {
+ case "avg_bucket", "bucket_script", "bucket_selector", "bucket_sort",
+ "cumulative_cardinality", "cumulative_sum", "derivative",
+ "extended_stats_bucket", "inference", "max_bucket", "min_bucket",
+ "moving_avg", "moving_fn", "moving_percentiles", "normalize",
+ "percentiles_bucket", "serial_diff", "stats_bucket", "sum_bucket":
+ return true
+ }
+ }
+ return false
+}
+
+func aggregationMap(root map[string]any) (map[string]any, bool) {
+ if aggs, ok := root[aggKeyAggs].(map[string]any); ok {
+ return aggs, true
+ }
+ if aggs, ok := root[aggKeyAggregations].(map[string]any); ok {
+ return aggs, true
+ }
+ return nil, false
+}
+
+func parseHistogramInterval(dh map[string]any) (time.Duration, bool) {
+ value, hasFixed := dh["fixed_interval"]
+ calendarValue, hasCalendar := dh["calendar_interval"]
+ if hasFixed == hasCalendar || hasAnyKey(dh, "interval") {
+ return 0, false
+ }
+ if hasCalendar {
+ return parseFixedCalendarInterval(fmt.Sprint(calendarValue))
+ }
+ if d, ok := parseESDuration(fmt.Sprint(value)); ok {
+ return d, true
+ }
+ return 0, false
+}
+
+func parseFixedCalendarInterval(input string) (time.Duration, bool) {
+ input = strings.TrimSpace(input)
+ switch input {
+ case "1m":
+ return time.Minute, true
+ case "1h":
+ return time.Hour, true
+ case "1d":
+ return 24 * time.Hour, true
+ }
+ switch strings.ToLower(input) {
+ case "minute":
+ return time.Minute, true
+ case "hour":
+ return time.Hour, true
+ case "day":
+ return 24 * time.Hour, true
+ default:
+ return 0, false
+ }
+}
+
+func parseESDuration(input string) (time.Duration, bool) {
+ input = strings.TrimSpace(input)
+ if input == "" {
+ return 0, false
+ }
+ if d, err := time.ParseDuration(input); err == nil {
+ return d, d > 0
+ }
+ unit := input[len(input)-1]
+ n, err := strconv.Atoi(input[:len(input)-1])
+ if err != nil {
+ return 0, false
+ }
+ switch unit {
+ case 'd':
+ return checkedDuration(n, 24*time.Hour)
+ case 'w':
+ return checkedDuration(n, 7*24*time.Hour)
+ }
+ return 0, false
+}
+
+func checkedDuration(n int, unit time.Duration) (time.Duration, bool) {
+ if n <= 0 {
+ return 0, false
+ }
+ d := time.Duration(n) * unit
+ return d, d > 0 && d/time.Duration(n) == unit
+}
+
+func isUTCTimeZone(value string) bool {
+ switch strings.ToUpper(strings.TrimSpace(value)) {
+ case "UTC", "Z", "+00:00", "-00:00":
+ return true
+ default:
+ return false
+ }
+}
+
+func walkMaps(v any, fn func(map[string]any) bool) bool {
+ switch x := v.(type) {
+ case map[string]any:
+ if !fn(x) {
+ return false
+ }
+ for _, child := range x {
+ if !walkMaps(child, fn) {
+ return false
+ }
+ }
+ case []any:
+ for _, child := range x {
+ if !walkMaps(child, fn) {
+ return false
+ }
+ }
+ }
+ return true
+}
+
+func normalizedBodyForCache(body map[string]any, timestampField string) string {
+ clone := cloneMap(body)
+ replaceRangeValues(clone, timestampField, rangeStartToken, rangeEndToken)
+ replaceExtendedBounds(clone, timestampField, rangeStartToken, rangeEndToken)
+ b, _ := canonicalJSON(clone)
+ return string(b)
+}
+
+func cloneMap(in map[string]any) map[string]any {
+ out := maps.Clone(in)
+ for k, v := range out {
+ out[k] = cloneValue(v)
+ }
+ return out
+}
+
+func cloneValue(v any) any {
+ switch x := v.(type) {
+ case map[string]any:
+ return cloneMap(x)
+ case []any:
+ out := make([]any, len(x))
+ for i := range x {
+ out[i] = cloneValue(x[i])
+ }
+ return out
+ default:
+ return x
+ }
+}
+
+func canonicalJSON(v any) ([]byte, error) {
+ var buf bytes.Buffer
+ enc := json.NewEncoder(&buf)
+ enc.SetEscapeHTML(false)
+ if err := enc.Encode(v); err != nil {
+ return nil, err
+ }
+ return bytes.TrimSpace(buf.Bytes()), nil
+}
+
+func replaceRangeValues(root map[string]any, timestampField string, start, end any) bool {
+ var replaced bool
+ walkMaps(root, func(m map[string]any) bool {
+ rangeNode, ok := m["range"].(map[string]any)
+ if !ok {
+ return true
+ }
+ fieldNode, ok := rangeNode[timestampField].(map[string]any)
+ if !ok {
+ return true
+ }
+ setRangeBound(fieldNode, start, "gte", "gt", "from")
+ setRangeBound(fieldNode, end, "lte", "lt", "to")
+ replaced = true
+ return true
+ })
+ return replaced
+}
+
+func setRangeBound(m map[string]any, value any, keys ...string) {
+ for _, key := range keys {
+ if _, ok := m[key]; ok {
+ m[key] = value
+ return
+ }
+ }
+}
+
+func replaceExtendedBounds(root map[string]any, timestampField string, start, end any) {
+ aggs, ok := aggregationMap(root)
+ if !ok {
+ return
+ }
+ for _, v := range aggs {
+ agg, ok := v.(map[string]any)
+ if !ok {
+ continue
+ }
+ dh, ok := agg["date_histogram"].(map[string]any)
+ if !ok {
+ continue
+ }
+ if field, _ := dh["field"].(string); field != "" && field != timestampField {
+ continue
+ }
+ for _, key := range []string{"extended_bounds", "hard_bounds"} {
+ if bounds, ok := dh[key].(map[string]any); ok {
+ if _, ok := bounds["min"]; ok {
+ bounds["min"] = start
+ }
+ if _, ok := bounds["max"]; ok {
+ bounds["max"] = end
+ }
+ }
+ }
+ }
+}
+
+// SetExtent rewrites the Elasticsearch search body to request the provided extent.
+func (c *Client) SetExtent(r *http.Request, trq *timeseries.TimeRangeQuery,
+ extent *timeseries.Extent,
+) {
+ if r == nil || trq == nil || extent == nil {
+ return
+ }
+ plan, ok := trq.ParsedQuery.(*RequestPlan)
+ if !ok || plan == nil {
+ return
+ }
+ out := plan.bodyForExtent(extent)
+ if plan.SourceBody {
+ q := r.URL.Query()
+ q.Set(queryParamSource, string(out))
+ r.URL.RawQuery = q.Encode()
+ return
+ }
+ request.SetBody(r, out)
+}
+
+func (p *RequestPlan) bodyForExtent(extent *timeseries.Extent) []byte {
+ if p == nil {
+ return nil
+ }
+ switch p.Kind {
+ case requestKindMSearch:
+ var buf bytes.Buffer
+ for _, sp := range p.Searches {
+ hb, _ := json.Marshal(sp.Header)
+ bb, _ := json.Marshal(sp.bodyForExtent(extent))
+ buf.Write(hb)
+ _ = buf.WriteByte('\n')
+ buf.Write(bb)
+ _ = buf.WriteByte('\n')
+ }
+ return buf.Bytes()
+ default:
+ if len(p.Searches) == 0 {
+ return nil
+ }
+ b, _ := json.Marshal(p.Searches[0].bodyForExtent(extent))
+ return b
+ }
+}
+
+func (sp *SearchPlan) bodyForExtent(extent *timeseries.Extent) map[string]any {
+ body := cloneMap(sp.Body)
+ rangeStart, rangeEnd := sp.formatRangeExtent(extent)
+ replaceRangeValues(body, sp.TimestampField, rangeStart, rangeEnd)
+ rewriteHistogramBounds(body, sp.TimestampField, extent.Start, extent.End)
+ return body
+}
+
+func (sp *SearchPlan) formatRangeExtent(extent *timeseries.Extent) (any, any) {
+ end := extent.End.Add(sp.Step)
+ if !sp.RangeEndExclusive {
+ end = end.Add(-time.Millisecond)
+ }
+ return sp.formatTimestamp(extent.Start), sp.formatTimestamp(end)
+}
+
+func (sp *SearchPlan) formatTimestamp(t time.Time) any {
+ return formatTimestamp(t, sp.TimestampValueKind)
+}
+
+func formatTimestamp(t time.Time, kind timestampValueKind) any {
+ switch kind {
+ case timestampValueEpochSeconds:
+ return t.Unix()
+ case timestampValueEpochMillis:
+ return t.UnixMilli()
+ default:
+ return t.UTC().Format(time.RFC3339Nano)
+ }
+}
+
+func rewriteHistogramBounds(root map[string]any, timestampField string, start, end time.Time) {
+ aggs, ok := aggregationMap(root)
+ if !ok {
+ return
+ }
+ for _, value := range aggs {
+ agg, ok := value.(map[string]any)
+ if !ok {
+ continue
+ }
+ histogram, ok := agg["date_histogram"].(map[string]any)
+ if !ok {
+ continue
+ }
+ if field, _ := histogram["field"].(string); field != timestampField {
+ continue
+ }
+ for _, key := range []string{"extended_bounds", "hard_bounds"} {
+ bounds, ok := histogram[key].(map[string]any)
+ if !ok {
+ continue
+ }
+ for name, timestamp := range map[string]time.Time{"min": start, "max": end} {
+ original, exists := bounds[name]
+ if !exists {
+ continue
+ }
+ _, kind, ok := parseRangeTime(original, histogram)
+ if ok {
+ bounds[name] = formatTimestamp(timestamp, kind)
+ }
+ }
+ }
+ }
+}
+
+// FastForwardRequest is not supported for Elasticsearch.
+func (c *Client) FastForwardRequest(_ *http.Request) (*http.Request, error) {
+ return nil, nil
+}
diff --git a/pkg/backends/elasticsearch/query_test.go b/pkg/backends/elasticsearch/query_test.go
new file mode 100644
index 000000000..513f216da
--- /dev/null
+++ b/pkg/backends/elasticsearch/query_test.go
@@ -0,0 +1,686 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/trickstercache/trickster/v2/pkg/timeseries"
+)
+
+const searchBody = `{
+ "size": 0,
+ "query": {
+ "bool": {
+ "filter": [
+ {
+ "range": {
+ "@timestamp": {
+ "gte": 1704067200000,
+ "lte": 1704067859999,
+ "format": "epoch_millis"
+ }
+ }
+ }
+ ]
+ }
+ },
+ "aggs": {
+ "2": {
+ "meta": {
+ "source": "query"
+ },
+ "date_histogram": {
+ "field": "@timestamp",
+ "fixed_interval": "1m",
+ "min_doc_count": 0,
+ "extended_bounds": {
+ "min": 1704067200000,
+ "max": 1704067800000
+ }
+ },
+ "aggs": {
+ "1": {
+ "avg": {
+ "field": "value"
+ }
+ }
+ }
+ }
+ }
+}`
+
+func TestParseTimeRangeQuerySearch(t *testing.T) {
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", strings.NewReader(searchBody))
+ trq, ro, canOPC, err := c.ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatalf("ParseTimeRangeQuery returned error: %v", err)
+ }
+ if !canOPC {
+ t.Fatal("expected canOPC fallback to be true")
+ }
+ if ro == nil || ro.ProviderRequest == nil {
+ t.Fatal("expected provider request in request options")
+ }
+ if trq.Step != time.Minute {
+ t.Fatalf("step = %s, want 1m", trq.Step)
+ }
+ if got, want := trq.Extent.Start.UnixMilli(), int64(1704067200000); got != want {
+ t.Fatalf("start = %d, want %d", got, want)
+ }
+ if got, want := trq.Extent.End.UnixMilli(), int64(1704067800000); got != want {
+ t.Fatalf("end = %d, want %d", got, want)
+ }
+ if trq.TimestampDefinition.Name != "@timestamp" {
+ t.Fatalf("timestamp field = %q, want @timestamp", trq.TimestampDefinition.Name)
+ }
+ if !strings.Contains(trq.Statement, rangeStartToken) || !strings.Contains(trq.Statement, rangeEndToken) {
+ t.Fatalf("normalized statement does not contain range tokens: %s", trq.Statement)
+ }
+}
+
+func TestParseTimeRangeQueryGetBody(t *testing.T) {
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodGet, "/metrics/_search", strings.NewReader(searchBody))
+ trq, _, _, err := c.ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatalf("ParseTimeRangeQuery returned error: %v", err)
+ }
+ if got, want := trq.Step, time.Minute; got != want {
+ t.Fatalf("step = %s, want %s", got, want)
+ }
+}
+
+func TestParseTimeRangeQuerySourceParam(t *testing.T) {
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodGet,
+ "/metrics/_search?source="+url.QueryEscape(searchBody), nil)
+ trq, ro, _, err := c.ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatalf("ParseTimeRangeQuery returned error: %v", err)
+ }
+ plan := ro.ProviderRequest.(*RequestPlan)
+ if !plan.SourceBody {
+ t.Fatal("expected source query parameter to be marked as source body")
+ }
+ if _, ok := trq.CacheKeyElements[queryParamSource]; !ok {
+ t.Fatalf("expected normalized %q cache key element: %v", queryParamSource, trq.CacheKeyElements)
+ }
+ if _, ok := trq.CacheKeyElements["query"]; ok {
+ t.Fatalf("source request must not retain a separate query cache key: %v", trq.CacheKeyElements)
+ }
+ next := ×eries.Extent{
+ Start: trq.Extent.Start.Add(time.Minute),
+ End: trq.Extent.Start.Add(2 * time.Minute),
+ }
+ c.SetExtent(r, trq, next)
+ if got := r.URL.Query().Get(queryParamSource); !strings.Contains(got, "1704067260000") {
+ t.Fatalf("rewritten source query does not include new start: %s", got)
+ }
+}
+
+func TestSetExtentSearchBody(t *testing.T) {
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", strings.NewReader(searchBody))
+ trq, _, _, err := c.ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ next := ×eries.Extent{
+ Start: time.UnixMilli(1704067320000),
+ End: time.UnixMilli(1704067440000),
+ }
+ c.SetExtent(r, trq, next)
+ var out map[string]any
+ if err := json.NewDecoder(r.Body).Decode(&out); err != nil {
+ t.Fatal(err)
+ }
+ rangeNode := out["query"].(map[string]any)["bool"].(map[string]any)["filter"].([]any)[0].(map[string]any)["range"].(map[string]any)["@timestamp"].(map[string]any)
+ if got, want := int64(rangeNode["gte"].(float64)), int64(1704067320000); got != want {
+ t.Fatalf("gte = %d, want %d", got, want)
+ }
+ if got, want := int64(rangeNode["lte"].(float64)), int64(1704067499999); got != want {
+ t.Fatalf("lte = %d, want %d", got, want)
+ }
+ bounds := out["aggs"].(map[string]any)["2"].(map[string]any)["date_histogram"].(map[string]any)["extended_bounds"].(map[string]any)
+ if got, want := int64(bounds["min"].(float64)), int64(1704067320000); got != want {
+ t.Fatalf("extended min = %d, want %d", got, want)
+ }
+ if got, want := int64(bounds["max"].(float64)), int64(1704067440000); got != want {
+ t.Fatalf("extended max = %d, want %d", got, want)
+ }
+}
+
+func TestParseUnsupportedSearchFallsBackToObjectCache(t *testing.T) {
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search",
+ strings.NewReader(`{"query":{"match_all":{}}}`))
+ trq, _, canOPC, err := c.ParseTimeRangeQuery(r)
+ if err == nil {
+ t.Fatal("expected unsupported query error")
+ }
+ if !canOPC {
+ t.Fatal("expected object cache fallback")
+ }
+ if trq == nil || trq.CacheKeyElements["query"] == "" {
+ t.Fatal("expected normalized cache key elements for object cache fallback")
+ }
+}
+
+func TestUnsupportedSearchFallbackKeepsExactTimeRangeInCacheKey(t *testing.T) {
+ const bodyTemplate = `{"size":0,"query":{"range":{"@timestamp":` +
+ `{"gte":%d,"lte":%d,"format":"epoch_millis"}}}}`
+ c := &Client{}
+ parse := func(start, end int64) *timeseries.TimeRangeQuery {
+ t.Helper()
+ body := fmt.Sprintf(bodyTemplate, start, end)
+ r := httptest.NewRequest(http.MethodPost, "/logs/_search", strings.NewReader(body))
+ trq, _, canOPC, err := c.ParseTimeRangeQuery(r)
+ if err == nil || !canOPC {
+ t.Fatalf("expected object-cache fallback, canOPC=%v err=%v", canOPC, err)
+ }
+ if strings.Contains(trq.Statement, rangeStartToken) || strings.Contains(trq.Statement, rangeEndToken) {
+ t.Fatalf("fallback statement must retain exact range values: %s", trq.Statement)
+ }
+ return trq
+ }
+
+ first := parse(1704067200000, 1704067800000)
+ second := parse(1704067800000, 1704068400000)
+ if first.CacheKeyElements["query"] == second.CacheKeyElements["query"] {
+ t.Fatalf("different fallback time ranges share a cache key: %q", first.CacheKeyElements["query"])
+ }
+}
+
+func TestParseTimeRangeQueryRejectsLossyShapes(t *testing.T) {
+ tests := []struct {
+ name string
+ mutate func(map[string]any)
+ }{
+ {
+ name: "search hits requested",
+ mutate: func(body map[string]any) {
+ body["size"] = json.Number("10")
+ },
+ },
+ {
+ name: "multiple timestamp ranges",
+ mutate: func(body map[string]any) {
+ filters := body["query"].(map[string]any)["bool"].(map[string]any)["filter"].([]any)
+ filters = append(filters, cloneValue(filters[0]))
+ body["query"].(map[string]any)["bool"].(map[string]any)["filter"] = filters
+ },
+ },
+ {
+ name: "exclusive range",
+ mutate: func(body map[string]any) {
+ rangeBody := firstTimestampRange(body)
+ rangeBody["gt"] = rangeBody["gte"]
+ delete(rangeBody, "gte")
+ },
+ },
+ {
+ name: "multiple top-level aggregations",
+ mutate: func(body map[string]any) {
+ aggs := body[aggKeyAggs].(map[string]any)
+ aggs["3"] = cloneValue(aggs["2"])
+ },
+ },
+ {
+ name: "variable calendar interval",
+ mutate: func(body map[string]any) {
+ dh := firstDateHistogram(body)
+ dh["calendar_interval"] = "1M"
+ delete(dh, "fixed_interval")
+ },
+ },
+ {
+ name: "zero interval",
+ mutate: func(body map[string]any) {
+ firstDateHistogram(body)["fixed_interval"] = "0m"
+ },
+ },
+ {
+ name: "keyed buckets",
+ mutate: func(body map[string]any) {
+ firstDateHistogram(body)["keyed"] = true
+ },
+ },
+ {
+ name: "non-UTC bucket alignment",
+ mutate: func(body map[string]any) {
+ firstDateHistogram(body)["time_zone"] = "America/New_York"
+ },
+ },
+ {
+ name: "partial final bucket",
+ mutate: func(body map[string]any) {
+ firstTimestampRange(body)["lte"] = json.Number("1704067800000")
+ },
+ },
+ {
+ name: "optional timestamp range",
+ mutate: func(body map[string]any) {
+ rangeClause := body["query"].(map[string]any)["bool"].(map[string]any)["filter"].([]any)[0]
+ body["query"] = map[string]any{"bool": map[string]any{
+ "should": []any{rangeClause, map[string]any{"match_all": map[string]any{}}},
+ }}
+ },
+ },
+ {
+ name: "descending buckets",
+ mutate: func(body map[string]any) {
+ firstDateHistogram(body)["order"] = map[string]any{"_key": "desc"}
+ },
+ },
+ {
+ name: "pipeline aggregation",
+ mutate: func(body map[string]any) {
+ agg := body[aggKeyAggs].(map[string]any)["2"].(map[string]any)
+ agg[aggKeyAggs].(map[string]any)["rate"] = map[string]any{
+ "derivative": map[string]any{"buckets_path": "1"},
+ }
+ },
+ },
+ {
+ name: "mismatched extended bounds",
+ mutate: func(body map[string]any) {
+ bounds := firstDateHistogram(body)["extended_bounds"].(map[string]any)
+ bounds["max"] = json.Number("1704067860000")
+ },
+ },
+ {
+ name: "empty buckets without extended bounds",
+ mutate: func(body map[string]any) {
+ delete(firstDateHistogram(body), "extended_bounds")
+ },
+ },
+ {
+ name: "empty buckets with partial extended bounds",
+ mutate: func(body map[string]any) {
+ bounds := firstDateHistogram(body)["extended_bounds"].(map[string]any)
+ delete(bounds, "max")
+ },
+ },
+ {
+ name: "minimum document count above one",
+ mutate: func(body map[string]any) {
+ firstDateHistogram(body)["min_doc_count"] = json.Number("2")
+ },
+ },
+ {
+ name: "range time zone",
+ mutate: func(body map[string]any) {
+ firstTimestampRange(body)["time_zone"] = "+01:00"
+ },
+ },
+ {
+ name: "date range relation",
+ mutate: func(body map[string]any) {
+ firstTimestampRange(body)["relation"] = "intersects"
+ },
+ },
+ {
+ name: "scripted histogram",
+ mutate: func(body map[string]any) {
+ firstDateHistogram(body)["script"] = "emit(doc['@timestamp'].value.toInstant().toEpochMilli())"
+ },
+ },
+ {
+ name: "legacy interval",
+ mutate: func(body map[string]any) {
+ histogram := firstDateHistogram(body)
+ histogram["interval"] = histogram["fixed_interval"]
+ delete(histogram, "fixed_interval")
+ },
+ },
+ {
+ name: "response profile requested",
+ mutate: func(body map[string]any) {
+ body["profile"] = true
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ body := decodedSearchBody(t)
+ tt.mutate(body)
+ encoded, err := json.Marshal(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", bytes.NewReader(encoded))
+ trq, _, canOPC, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err == nil || !canOPC {
+ t.Fatalf("expected safe object-cache fallback, canOPC=%v err=%v", canOPC, err)
+ }
+ if trq == nil || trq.Statement == "" {
+ t.Fatal("fallback must retain an exact request statement for the object-cache key")
+ }
+ })
+ }
+}
+
+func TestParseTimeRangeQueryAcceptsExclusiveCompleteBuckets(t *testing.T) {
+ body := decodedSearchBody(t)
+ rangeBody := firstTimestampRange(body)
+ delete(rangeBody, "lte")
+ rangeBody["lt"] = json.Number("1704067860000")
+ encoded, err := json.Marshal(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", bytes.NewReader(encoded))
+ trq, _, _, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got, want := trq.Extent.End.UnixMilli(), int64(1704067800000); got != want {
+ t.Fatalf("bucket extent end = %d, want %d", got, want)
+ }
+ (&Client{}).SetExtent(r, trq, &trq.Extent)
+ var rewritten map[string]any
+ if err := json.NewDecoder(r.Body).Decode(&rewritten); err != nil {
+ t.Fatal(err)
+ }
+ if got, want := firstTimestampRange(rewritten)["lt"].(float64), float64(1704067860000); got != want {
+ t.Fatalf("lt = %.0f, want %.0f", got, want)
+ }
+}
+
+func TestParseTimeRangeQueryAcceptsSafeHistogramVariants(t *testing.T) {
+ tests := []struct {
+ name string
+ mutate func(map[string]any)
+ }{
+ {
+ name: "UTC calendar minute",
+ mutate: func(body map[string]any) {
+ histogram := firstDateHistogram(body)
+ histogram["calendar_interval"] = "1m"
+ delete(histogram, "fixed_interval")
+ },
+ },
+ {
+ name: "positive min doc count without bounds",
+ mutate: func(body map[string]any) {
+ histogram := firstDateHistogram(body)
+ histogram["min_doc_count"] = json.Number("1")
+ delete(histogram, "extended_bounds")
+ },
+ },
+ {
+ name: "bounds inside edge buckets",
+ mutate: func(body map[string]any) {
+ bounds := firstDateHistogram(body)["extended_bounds"].(map[string]any)
+ bounds["max"] = json.Number("1704067859999")
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ body := decodedSearchBody(t)
+ tt.mutate(body)
+ encoded, err := json.Marshal(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", bytes.NewReader(encoded))
+ if _, _, _, err := (&Client{}).ParseTimeRangeQuery(r); err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+func TestParseTimeRangeQueryNumericDateFormats(t *testing.T) {
+ t.Run("numeric dates default to epoch millis", func(t *testing.T) {
+ body := decodedSearchBody(t)
+ rangeBody := firstTimestampRange(body)
+ rangeBody["gte"] = json.Number("946684800000")
+ rangeBody["lte"] = json.Number("946685459999")
+ delete(rangeBody, "format")
+ bounds := firstDateHistogram(body)["extended_bounds"].(map[string]any)
+ bounds["min"] = json.Number("946684800000")
+ bounds["max"] = json.Number("946685400000")
+ encoded, err := json.Marshal(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", bytes.NewReader(encoded))
+ trq, _, _, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got, want := trq.Extent.Start.UnixMilli(), int64(946684800000); got != want {
+ t.Fatalf("start = %d, want %d", got, want)
+ }
+ })
+
+ t.Run("epoch seconds use exclusive bucket boundary", func(t *testing.T) {
+ body := decodedSearchBody(t)
+ rangeBody := firstTimestampRange(body)
+ rangeBody["gte"] = json.Number("1704067200")
+ delete(rangeBody, "lte")
+ rangeBody["lt"] = json.Number("1704067860")
+ rangeBody["format"] = "epoch_second"
+ histogram := firstDateHistogram(body)
+ histogram["format"] = "epoch_second"
+ bounds := histogram["extended_bounds"].(map[string]any)
+ bounds["min"] = json.Number("1704067200")
+ bounds["max"] = json.Number("1704067800")
+ encoded, err := json.Marshal(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", bytes.NewReader(encoded))
+ trq, _, _, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ (&Client{}).SetExtent(r, trq, &trq.Extent)
+ var rewritten map[string]any
+ if err := json.NewDecoder(r.Body).Decode(&rewritten); err != nil {
+ t.Fatal(err)
+ }
+ if got, want := firstTimestampRange(rewritten)["lt"].(float64), float64(1704067860); got != want {
+ t.Fatalf("lt = %.0f, want %.0f", got, want)
+ }
+ })
+}
+
+func TestSetExtentPreservesHistogramBoundRepresentation(t *testing.T) {
+ body := decodedSearchBody(t)
+ rangeBody := firstTimestampRange(body)
+ rangeBody["gte"] = "2024-01-01T00:00:00Z"
+ rangeBody["lte"] = "2024-01-01T00:10:59.999Z"
+ delete(rangeBody, "format")
+ encoded, err := json.Marshal(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search", bytes.NewReader(encoded))
+ trq, _, _, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ next := ×eries.Extent{
+ Start: time.UnixMilli(1704067320000),
+ End: time.UnixMilli(1704067440000),
+ }
+ (&Client{}).SetExtent(r, trq, next)
+ var rewritten map[string]any
+ if err := json.NewDecoder(r.Body).Decode(&rewritten); err != nil {
+ t.Fatal(err)
+ }
+ if _, ok := firstTimestampRange(rewritten)["gte"].(string); !ok {
+ t.Fatal("range boundary did not retain its RFC3339 representation")
+ }
+ bounds := firstDateHistogram(rewritten)["extended_bounds"].(map[string]any)
+ if _, ok := bounds["min"].(float64); !ok {
+ t.Fatal("histogram boundary did not retain its numeric representation")
+ }
+}
+
+func TestParseTimeRangeQueryRejectsUnsafeURLParameters(t *testing.T) {
+ r := httptest.NewRequest(http.MethodPost, "/metrics/_search?q=service:api", strings.NewReader(searchBody))
+ trq, _, canOPC, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err == nil || !canOPC {
+ t.Fatalf("expected object-cache fallback, canOPC=%v err=%v", canOPC, err)
+ }
+ if trq == nil || strings.Contains(trq.Statement, rangeStartToken) {
+ t.Fatal("fallback must keep the exact request body in its cache key")
+ }
+}
+
+func TestUnsupportedMSearchFallbackUsesCompleteBody(t *testing.T) {
+ unsupported := `{"size":0,"query":{"match_all":{}}}`
+ build := func(index string) string {
+ return `{"index":"first"}` + "\n" + unsupported + "\n" +
+ fmt.Sprintf(`{"index":%q}`, index) + "\n" + unsupported + "\n"
+ }
+ parse := func(body string) *timeseries.TimeRangeQuery {
+ t.Helper()
+ r := httptest.NewRequest(http.MethodPost, "/_msearch", strings.NewReader(body))
+ trq, _, canOPC, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err == nil || !canOPC {
+ t.Fatalf("expected object-cache fallback, canOPC=%v err=%v", canOPC, err)
+ }
+ return trq
+ }
+ first := parse(build("logs-a"))
+ second := parse(build("logs-b"))
+ if first.CacheKeyElements["query"] == second.CacheKeyElements["query"] {
+ t.Fatal("msearch fallback key omitted request pairs after the first unsupported search")
+ }
+}
+
+func TestMalformedSearchBodiesFallBackWithoutRewriting(t *testing.T) {
+ tests := []struct {
+ name string
+ path string
+ body string
+ }{
+ {
+ name: "trailing search document",
+ path: "/_search",
+ body: compactJSON(t, searchBody) + ` {"extra":true}`,
+ },
+ {
+ name: "blank line inside msearch",
+ path: "/_msearch",
+ body: `{"index":"metrics"}` + "\n\n" + compactJSON(t, searchBody) + "\n",
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ r := httptest.NewRequest(http.MethodPost, tt.path, strings.NewReader(tt.body))
+ trq, _, canOPC, err := (&Client{}).ParseTimeRangeQuery(r)
+ if err == nil || !canOPC {
+ t.Fatalf("expected object-cache fallback, canOPC=%v err=%v", canOPC, err)
+ }
+ if trq == nil || trq.Statement != strings.TrimSpace(tt.body) {
+ t.Fatal("fallback cache key did not retain the malformed body exactly")
+ }
+ })
+ }
+}
+
+func TestParseMSearchPlan(t *testing.T) {
+ searchLine := compactJSON(t, searchBody)
+ body := bytes.NewBuffer(nil)
+ body.WriteString(`{"index":"metrics"}` + "\n")
+ body.WriteString(searchLine + "\n")
+ body.WriteString(`{"index":"metrics"}` + "\n")
+ body.WriteString(searchLine + "\n")
+
+ c := &Client{}
+ r := httptest.NewRequest(http.MethodPost, "/_msearch", body)
+ trq, ro, _, err := c.ParseTimeRangeQuery(r)
+ if err != nil {
+ t.Fatalf("ParseTimeRangeQuery returned error: %v", err)
+ }
+ plan := ro.ProviderRequest.(*RequestPlan)
+ if plan.Kind != requestKindMSearch {
+ t.Fatalf("plan kind = %d, want msearch", plan.Kind)
+ }
+ if len(plan.Searches) != 2 {
+ t.Fatalf("searches = %d, want 2", len(plan.Searches))
+ }
+ next := ×eries.Extent{
+ Start: trq.Extent.Start.Add(2 * time.Minute),
+ End: trq.Extent.Start.Add(3 * time.Minute),
+ }
+ c.SetExtent(r, trq, next)
+ lines := splitNDJSONFromString(t, r.Body)
+ if len(lines) != 4 {
+ t.Fatalf("rewritten msearch lines = %d, want 4", len(lines))
+ }
+}
+
+func splitNDJSONFromString(t *testing.T, body any) [][]byte {
+ t.Helper()
+ var buf bytes.Buffer
+ switch x := body.(type) {
+ case interface{ Read([]byte) (int, error) }:
+ _, _ = buf.ReadFrom(x)
+ default:
+ t.Fatalf("unsupported body type %T", body)
+ }
+ return splitNDJSON(buf.Bytes())
+}
+
+func compactJSON(t *testing.T, input string) string {
+ t.Helper()
+ var buf bytes.Buffer
+ if err := json.Compact(&buf, []byte(input)); err != nil {
+ t.Fatal(err)
+ }
+ return buf.String()
+}
+
+func decodedSearchBody(t *testing.T) map[string]any {
+ t.Helper()
+ var body map[string]any
+ dec := json.NewDecoder(strings.NewReader(searchBody))
+ dec.UseNumber()
+ if err := dec.Decode(&body); err != nil {
+ t.Fatal(err)
+ }
+ return body
+}
+
+func firstTimestampRange(body map[string]any) map[string]any {
+ query := body["query"].(map[string]any)
+ boolQuery := query["bool"].(map[string]any)
+ filter := boolQuery["filter"].([]any)[0].(map[string]any)
+ return filter["range"].(map[string]any)["@timestamp"].(map[string]any)
+}
+
+func firstDateHistogram(body map[string]any) map[string]any {
+ return body[aggKeyAggs].(map[string]any)["2"].(map[string]any)["date_histogram"].(map[string]any)
+}
diff --git a/pkg/backends/elasticsearch/routes.go b/pkg/backends/elasticsearch/routes.go
new file mode 100644
index 000000000..696bf9582
--- /dev/null
+++ b/pkg/backends/elasticsearch/routes.go
@@ -0,0 +1,78 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "net/http"
+
+ 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"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/methods"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/paths/matching"
+ po "github.com/trickstercache/trickster/v2/pkg/proxy/paths/options"
+)
+
+const (
+ handlerHealth = "health"
+ handlerQuery = "query"
+ handlerProxyCache = "proxycache"
+)
+
+func (c *Client) RegisterHandlers(handlers.Lookup) {
+ c.TimeseriesBackend.RegisterHandlers(
+ handlers.Lookup{
+ handlerHealth: http.HandlerFunc(c.HealthHandler),
+ handlerQuery: http.HandlerFunc(c.QueryHandler),
+ handlerProxyCache: http.HandlerFunc(c.ObjectProxyCacheHandler),
+ providers.Proxy: http.HandlerFunc(c.ProxyHandler),
+ },
+ )
+}
+
+// DefaultPathConfigs returns the default path configs for Elasticsearch.
+func (c *Client) DefaultPathConfigs(_ *bo.Options) po.List {
+ return po.List{
+ {
+ Path: "/",
+ HandlerName: handlerQuery,
+ Methods: methods.GetAndPost(),
+ CacheKeyParams: []string{"*"},
+ MatchType: matching.PathMatchTypePrefix,
+ MatchTypeName: matching.PathMatchNamePrefix,
+ },
+ {
+ Path: "/",
+ HandlerName: handlerProxyCache,
+ Methods: []string{http.MethodHead},
+ CacheKeyParams: []string{"*"},
+ MatchType: matching.PathMatchTypePrefix,
+ MatchTypeName: matching.PathMatchNamePrefix,
+ },
+ {
+ Path: "/",
+ HandlerName: providers.Proxy,
+ Methods: []string{
+ http.MethodPut, http.MethodDelete, http.MethodConnect,
+ http.MethodOptions, http.MethodTrace, http.MethodPatch,
+ methods.MethodPurge,
+ },
+ MatchType: matching.PathMatchTypePrefix,
+ MatchTypeName: matching.PathMatchNamePrefix,
+ },
+ }
+}
diff --git a/pkg/backends/elasticsearch/routes_test.go b/pkg/backends/elasticsearch/routes_test.go
new file mode 100644
index 000000000..5b2b23dfa
--- /dev/null
+++ b/pkg/backends/elasticsearch/routes_test.go
@@ -0,0 +1,97 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "net/http"
+ "slices"
+ "testing"
+
+ "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"
+ tu "github.com/trickstercache/trickster/v2/pkg/testutil"
+)
+
+func TestRegisterHandlers(t *testing.T) {
+ c, err := NewClient("test", nil, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c.RegisterHandlers(nil)
+ for _, name := range []string{handlerHealth, handlerQuery, handlerProxyCache, providers.Proxy} {
+ if _, ok := c.Handlers()[name]; !ok {
+ t.Fatalf("expected to find handler named %q", name)
+ }
+ }
+}
+
+func TestDefaultPathConfigs(t *testing.T) {
+ backendClient, err := NewClient("test", nil, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ts, _, r, _, err := tu.NewTestInstance("", backendClient.DefaultPathConfigs, http.StatusOK, "{}",
+ nil, providers.Elasticsearch, "/_search", "debug")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer ts.Close()
+
+ rsc := request.GetResources(r)
+ backendClient, err = NewClient("test", rsc.BackendOptions, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rsc.BackendClient = backendClient.(*Client)
+ rsc.BackendOptions.HTTPClient = backendClient.HTTPClient()
+
+ if !slices.ContainsFunc([]*po.Options(backendClient.Configuration().Paths),
+ func(pathConfig *po.Options) bool {
+ return pathConfig.Path == "/" && pathConfig.HandlerName == handlerQuery
+ }) {
+ t.Fatal("expected to find query path config")
+ }
+
+ const expectedLen = 3
+ if len(backendClient.Configuration().Paths) != expectedLen {
+ t.Fatalf("paths = %d, want %d", len(backendClient.Configuration().Paths), expectedLen)
+ }
+}
+
+func TestDefaultHealthCheckConfig(t *testing.T) {
+ backendClient, err := NewClient("test", nil, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ts, _, r, _, err := tu.NewTestInstance("", backendClient.DefaultPathConfigs, http.StatusOK, "{}",
+ nil, providers.Elasticsearch, "/_search", "debug")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer ts.Close()
+
+ rsc := request.GetResources(r)
+ backendClient, err = NewClient("test", rsc.BackendOptions, nil, nil, nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hc := backendClient.(*Client).DefaultHealthCheckConfig()
+ if got := hc.Path; got != "/_cluster/health" {
+ t.Fatalf("health path = %q, want /_cluster/health", got)
+ }
+}
diff --git a/pkg/backends/options/options.go b/pkg/backends/options/options.go
index 29741d546..72e974920 100644
--- a/pkg/backends/options/options.go
+++ b/pkg/backends/options/options.go
@@ -27,6 +27,7 @@ import (
"github.com/prometheus/common/sigv4"
ao "github.com/trickstercache/trickster/v2/pkg/backends/alb/options"
+ eso "github.com/trickstercache/trickster/v2/pkg/backends/elasticsearch/options"
ho "github.com/trickstercache/trickster/v2/pkg/backends/healthcheck/options"
prop "github.com/trickstercache/trickster/v2/pkg/backends/prometheus/options"
"github.com/trickstercache/trickster/v2/pkg/backends/providers"
@@ -160,6 +161,8 @@ type Options struct {
ALBOptions *ao.Options `yaml:"alb,omitempty"`
// Prometheus holds options specific to prometheus backends
Prometheus *prop.Options `yaml:"prometheus,omitempty"`
+ // Elasticsearch holds options specific to elasticsearch backends
+ Elasticsearch *eso.Options `yaml:"elasticsearch,omitempty"`
// TLS is the TLS Configuration for the Frontend and Backend
TLS *to.Options `yaml:"tls,omitempty"`
@@ -316,6 +319,9 @@ func (o *Options) Clone() *Options {
if o.Prometheus != nil {
out.Prometheus = o.Prometheus.Clone()
}
+ if o.Elasticsearch != nil {
+ out.Elasticsearch = o.Elasticsearch.Clone()
+ }
if o.AuthOptions != nil {
out.AuthOptions = o.AuthOptions.Clone()
@@ -614,6 +620,11 @@ func (o *Options) Initialize(name string) error {
}
}
}
+ if o.Elasticsearch != nil {
+ if err := o.Elasticsearch.Initialize(""); err != nil {
+ return err
+ }
+ }
if o.HealthCheck != nil {
if err := o.HealthCheck.Initialize(""); err != nil {
diff --git a/pkg/backends/options/options_test.go b/pkg/backends/options/options_test.go
index 04dd85597..ac7dc34e8 100644
--- a/pkg/backends/options/options_test.go
+++ b/pkg/backends/options/options_test.go
@@ -24,6 +24,7 @@ import (
"time"
"github.com/stretchr/testify/require"
+ eso "github.com/trickstercache/trickster/v2/pkg/backends/elasticsearch/options"
ho "github.com/trickstercache/trickster/v2/pkg/backends/healthcheck/options"
"github.com/trickstercache/trickster/v2/pkg/backends/providers"
ro "github.com/trickstercache/trickster/v2/pkg/backends/rule/options"
@@ -382,6 +383,38 @@ func TestInitialize(t *testing.T) {
}
}
+func TestInitializeElasticsearchOptions(t *testing.T) {
+ o := New()
+ o.Provider = providers.Elasticsearch
+ o.OriginURL = "http://elasticsearch:9200"
+ o.Elasticsearch = &eso.Options{}
+
+ if err := o.Initialize("test"); err != nil {
+ t.Fatal(err)
+ }
+ if got := o.Elasticsearch.TimestampField; got != eso.DefaultTimestampField {
+ t.Fatalf("timestamp_field = %q, want %q", got, eso.DefaultTimestampField)
+ }
+
+ o2, err := fromYAML(`
+backends:
+ test:
+ provider: elasticsearch
+ origin_url: http://elasticsearch:9200
+ elasticsearch:
+ timestamp_field: event_time
+`, "test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := o2.Initialize("test"); err != nil {
+ t.Fatal(err)
+ }
+ if got := o2.Elasticsearch.TimestampField; got != "event_time" {
+ t.Fatalf("timestamp_field = %q, want event_time", got)
+ }
+}
+
func TestValidateTLSConfigs(t *testing.T) {
o, err := fromTestYAML()
if err != nil {
diff --git a/pkg/backends/providers/providers.go b/pkg/backends/providers/providers.go
index f474a52c1..2c7a4fef1 100644
--- a/pkg/backends/providers/providers.go
+++ b/pkg/backends/providers/providers.go
@@ -40,6 +40,8 @@ const (
InfluxDBID
// ClickHouse represents the ClickHouse backend provider
ClickHouseID
+ // Elasticsearch represents the Elasticsearch backend provider
+ ElasticsearchID
Backends = "backends"
@@ -55,6 +57,8 @@ const (
Prometheus = "prometheus"
ClickHouse = "clickhouse"
InfluxDB = "influxdb"
+
+ Elasticsearch = "elasticsearch"
)
// Names is a map of Providers keyed by string name
@@ -66,6 +70,7 @@ var Names = map[string]Provider{
Prometheus: PrometheusID,
InfluxDB: InfluxDBID,
ClickHouse: ClickHouseID,
+ Elasticsearch: ElasticsearchID,
Proxy: RPID,
ReverseProxy: RPID,
ReverseProxyShort: RPID,
@@ -85,9 +90,10 @@ func init() {
}
var supportedTimeSeries = map[string]Provider{
- Prometheus: PrometheusID,
- InfluxDB: InfluxDBID,
- ClickHouse: ClickHouseID,
+ Prometheus: PrometheusID,
+ InfluxDB: InfluxDBID,
+ ClickHouse: ClickHouseID,
+ Elasticsearch: ElasticsearchID,
}
// IsSupportedTimeSeriesProvider returns true if the provided time series is supported by Trickster
diff --git a/pkg/backends/providers/providers_test.go b/pkg/backends/providers/providers_test.go
index 718bfb488..c19a2925e 100644
--- a/pkg/backends/providers/providers_test.go
+++ b/pkg/backends/providers/providers_test.go
@@ -49,6 +49,7 @@ func TestIsValidProvider(t *testing.T) {
{"", false},
{"invalid", false},
{InfluxDB, true},
+ {Elasticsearch, true},
}
for i, test := range tests {
@@ -73,4 +74,10 @@ func TestIsSupportedTimeSeriesProvider(t *testing.T) {
if !ok {
t.Error("expected true")
}
+
+ name = Elasticsearch
+ ok = IsSupportedTimeSeriesProvider(name)
+ if !ok {
+ t.Error("expected true")
+ }
}
diff --git a/pkg/backends/providers/registry/registry.go b/pkg/backends/providers/registry/registry.go
index 5cfb32e75..12a61065d 100644
--- a/pkg/backends/providers/registry/registry.go
+++ b/pkg/backends/providers/registry/registry.go
@@ -19,6 +19,7 @@ package registry
import (
"github.com/trickstercache/trickster/v2/pkg/backends/alb"
"github.com/trickstercache/trickster/v2/pkg/backends/clickhouse"
+ "github.com/trickstercache/trickster/v2/pkg/backends/elasticsearch"
"github.com/trickstercache/trickster/v2/pkg/backends/influxdb"
"github.com/trickstercache/trickster/v2/pkg/backends/prometheus"
"github.com/trickstercache/trickster/v2/pkg/backends/providers"
@@ -32,6 +33,7 @@ func SupportedProviders() types.Lookup {
return types.Lookup{
providers.ALB: alb.NewClient,
providers.ClickHouse: clickhouse.NewClient,
+ providers.Elasticsearch: elasticsearch.NewClient,
providers.InfluxDB: influxdb.NewClient,
providers.Prometheus: prometheus.NewClient,
providers.Rule: rule.NewClient,
diff --git a/pkg/proxy/authenticator/registry/registry.go b/pkg/proxy/authenticator/registry/registry.go
index 9e35e910b..3c5798070 100644
--- a/pkg/proxy/authenticator/registry/registry.go
+++ b/pkg/proxy/authenticator/registry/registry.go
@@ -59,7 +59,7 @@ func NewObserverFromProviderName(backendProvider string, data map[string]any) (t
var a types.Authenticator
var err error
switch backendProvider {
- case providers.Prometheus, providers.ReverseProxy, providers.Proxy,
+ case providers.Prometheus, providers.Elasticsearch, providers.ReverseProxy, providers.Proxy,
providers.ReverseProxyCache, providers.ReverseProxyCacheShort,
providers.ReverseProxyShort:
a, err = basic.New(data)
diff --git a/pkg/proxy/authenticator/registry/registry_test.go b/pkg/proxy/authenticator/registry/registry_test.go
new file mode 100644
index 000000000..a4d0e8b34
--- /dev/null
+++ b/pkg/proxy/authenticator/registry/registry_test.go
@@ -0,0 +1,45 @@
+/*
+ * 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 registry
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/trickstercache/trickster/v2/pkg/backends/providers"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/authenticator/options"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/authenticator/types"
+)
+
+func TestNewObserverFromElasticsearchProvider(t *testing.T) {
+ a, err := NewObserverFromProviderName(providers.Elasticsearch, map[string]any{
+ "options": &options.Options{ObserveOnly: true},
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ req := httptest.NewRequest(http.MethodGet, "http://example.com/", nil)
+ req.SetBasicAuth("alice", "secret")
+ result, err := a.Authenticate(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if result.Username != "alice" || result.Status != types.AuthObserved {
+ t.Fatalf("Authenticate() = %+v, want observed user alice", result)
+ }
+}
diff --git a/pkg/proxy/handlers/elasticsearch/elasticsearch.go b/pkg/proxy/handlers/elasticsearch/elasticsearch.go
new file mode 100644
index 000000000..1a4ad437b
--- /dev/null
+++ b/pkg/proxy/handlers/elasticsearch/elasticsearch.go
@@ -0,0 +1,73 @@
+/*
+ * 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 elasticsearch
+
+import (
+ "net/http"
+ "net/url"
+
+ "github.com/trickstercache/trickster/v2/pkg/backends/elasticsearch"
+ bo "github.com/trickstercache/trickster/v2/pkg/backends/options"
+ "github.com/trickstercache/trickster/v2/pkg/backends/providers"
+ co "github.com/trickstercache/trickster/v2/pkg/cache/options"
+ "github.com/trickstercache/trickster/v2/pkg/cache/registry"
+ "github.com/trickstercache/trickster/v2/pkg/config"
+ fopt "github.com/trickstercache/trickster/v2/pkg/frontend/options"
+ "github.com/trickstercache/trickster/v2/pkg/proxy/router/lm"
+ "github.com/trickstercache/trickster/v2/pkg/routing"
+)
+
+// NewAccelerator returns a new Elasticsearch Accelerator. only baseURL is required.
+func NewAccelerator(baseURL string) (http.Handler, error) {
+ return NewAcceleratorWithOptions(baseURL, nil, nil)
+}
+
+// NewAcceleratorWithOptions returns a new Elasticsearch Accelerator.
+func NewAcceleratorWithOptions(baseURL string, o *bo.Options, c *co.Options) (http.Handler, error) {
+ u, err := url.Parse(baseURL)
+ if err != nil {
+ return nil, err
+ }
+ if c == nil {
+ c = co.New()
+ c.Name = "default"
+ }
+ cache := registry.NewCache(c.Name, c)
+ err = cache.Connect()
+ if err != nil {
+ return nil, err
+ }
+ if o == nil {
+ o = bo.New()
+ o.Name = "default"
+ }
+ o.Provider = providers.Elasticsearch
+ o.CacheName = c.Name
+ o.Scheme = u.Scheme
+ o.Host = u.Host
+ o.PathPrefix = u.Path
+ r := lm.NewRouter()
+ cl, err := elasticsearch.NewClient("default", o, lm.NewRouter(), cache, nil, nil)
+ if err != nil {
+ return nil, err
+ }
+ o.HTTPClient = cl.HTTPClient()
+ o.Paths = cl.DefaultPathConfigs(o)
+ barecfg := &config.Config{Frontend: fopt.New()}
+ routing.RegisterPathRoutes(r, barecfg, cl.Handlers(), cl, o, cache, nil)
+ return r, nil
+}