fix(runner/elasticsearch): restore legacy endpoints, PPL scoping, and header auth #525
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,9 +2,9 @@ | |
| // Same shape as pkg/observability/prometheus. | ||
| // | ||
| // Action surface: | ||
| // - query_es : POST {index}/_search with DSL, or /_plugins/_ppl with PPL | ||
| // - query_es_indices : GET _alias (object keyed by index name) | ||
| // - query_es_index_field : GET {index}/_field_caps?fields=* | ||
| // - query_es_field_index_values : terms aggregation on {field} | ||
| package elasticsearch | ||
|
|
||
|
|
@@ -56,24 +56,43 @@ func (c *Client) Search(ctx context.Context, index, queryType string, query any) | |
| return c.do(ctx, http.MethodPost, "/"+index+"/_search", body, "application/json") | ||
| case "ppl": | ||
| s, _ := query.(string) | ||
| // PPL statements are `source=<index> | <commands>`. The legacy client | ||
| // scoped every PPL query to the requested index; callers pass just the | ||
| // command tail (or a full statement). Prepend the source clause when | ||
| // the query doesn't already carry one. | ||
| s = strings.TrimSpace(s) | ||
| if !strings.Contains(s, "source=") { | ||
| if s == "" { | ||
| s = "source=" + index | ||
| } else { | ||
| s = "source=" + index + " | " + s | ||
| } | ||
| } | ||
|
Comment on lines
+64
to
+70
There was a problem hiding this comment. The current check We can perform a more robust, case-insensitive check by verifying if the query starts with lower := strings.ToLower(s)
hasSource := false
if strings.HasPrefix(lower, "source") {
tail := strings.TrimSpace(strings.TrimPrefix(lower, "source"))
if strings.HasPrefix(tail, "=") {
hasSource = true
}
}
if !hasSource {
if s == "" {
s = "source=" + index
} else {
s = "source=" + index + " | " + s
}
} |
||
| body, _ := json.Marshal(map[string]any{"query": s}) | ||
| return c.do(ctx, http.MethodPost, "/_plugins/_ppl", body, "application/json") | ||
| default: | ||
| return nil, fmt.Errorf("elasticsearch: unsupported query_type %q", queryType) | ||
| } | ||
| } | ||
|
|
||
| // Indices returns the cluster's index→alias map via GET /_alias. The | ||
| // response is an object keyed by index name (matching the legacy get_alias), | ||
| // which the backend composer iterates as map keys. (The old /_cat/indices | ||
| // endpoint returned an array, which the composer could not index by key.) | ||
| func (c *Client) Indices(ctx context.Context) (json.RawMessage, error) { | ||
| return c.do(ctx, http.MethodGet, "/_alias", nil, "") | ||
| } | ||
|
|
||
| // IndexFields returns the field capabilities for an index via | ||
| // GET /{index}/_field_caps?fields=*. The response carries a `fields` map | ||
| // keyed by field name (matching the legacy field_caps), which is the shape | ||
| // the backend composer expects. (The old /_mapping shape nested fields under | ||
| // mappings.properties and did not match.) | ||
| func (c *Client) IndexFields(ctx context.Context, index string) (json.RawMessage, error) { | ||
| if index == "" { | ||
| return nil, errors.New("elasticsearch: index is required") | ||
| } | ||
| return c.do(ctx, http.MethodGet, "/"+index+"/_field_caps?fields=*", nil, "") | ||
| } | ||
|
|
||
| // FieldValues returns distinct values for one field via a terms aggregation. | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -58,6 +58,59 @@ func TestSearch_PPL_PostsToPlugins(t *testing.T) { | |
| } | ||
| } | ||
|
|
||
| // TestSearch_PPL_PrependsSource: a PPL query without a source= clause is | ||
| // scoped to the requested index, matching the legacy client. | ||
| func TestSearch_PPL_PrependsSource(t *testing.T) { | ||
| var gotBody string | ||
| c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { | ||
| body, _ := io.ReadAll(r.Body) | ||
| gotBody = string(body) | ||
| _, _ = w.Write([]byte(`{}`)) | ||
| }) | ||
| defer srv.Close() | ||
| if _, err := c.Search(context.Background(), "logs-1", "ppl", "head 5"); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if !strings.Contains(gotBody, `"query":"source=logs-1 | head 5"`) { | ||
| t.Errorf("source clause not prepended: %s", gotBody) | ||
| } | ||
| } | ||
|
|
||
| // TestSearch_PPL_KeepsExistingSource: a query that already has a source= | ||
| // clause is left untouched (no double source=). | ||
| func TestSearch_PPL_KeepsExistingSource(t *testing.T) { | ||
| var gotBody string | ||
| c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { | ||
| body, _ := io.ReadAll(r.Body) | ||
| gotBody = string(body) | ||
| _, _ = w.Write([]byte(`{}`)) | ||
| }) | ||
| defer srv.Close() | ||
| if _, err := c.Search(context.Background(), "logs-1", "ppl", "source=other | head 5"); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if strings.Count(gotBody, "source=") != 1 { | ||
| t.Errorf("expected exactly one source= clause: %s", gotBody) | ||
| } | ||
| } | ||
|
Comment on lines
+79
to
+95
There was a problem hiding this comment. Let's expand the test suite to cover case-insensitivity and whitespace variations in the // TestSearch_PPL_KeepsExistingSource: a query that already has a source=
// clause is left untouched (no double source=).
func TestSearch_PPL_KeepsExistingSource(t *testing.T) {
tests := []struct {
name string
query string
}{
{"exact", "source=other | head 5"},
{"with spaces", "source = other | head 5"},
{"uppercase", "SOURCE=other | head 5"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var gotBody string
c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
gotBody = string(body)
_, _ = w.Write([]byte(`{}`))
})
defer srv.Close()
if _, err := c.Search(context.Background(), "logs-1", "ppl", tt.query); err != nil {
t.Fatal(err)
}
if strings.Contains(gotBody, "source=logs-1") {
t.Errorf("expected logs-1 not to be prepended: %s", gotBody)
}
})
}
} |
||
|
|
||
| // TestExtraHeaders_Sent verifies ELASTICSEARCH_HEADER values reach upstream. | ||
| func TestExtraHeaders_Sent(t *testing.T) { | ||
| var got string | ||
| c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { | ||
| got = r.Header.Get("X-Tenant") | ||
| _, _ = w.Write([]byte(`{}`)) | ||
| }) | ||
| defer srv.Close() | ||
| c.ExtraHeaders = http.Header{"X-Tenant": []string{"team-a"}} | ||
| if _, err := c.Indices(context.Background()); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if got != "team-a" { | ||
| t.Errorf("X-Tenant = %q; want team-a", got) | ||
| } | ||
| } | ||
|
|
||
| func TestSearch_RequiresIndex(t *testing.T) { | ||
| c := New("http://x", nil) | ||
| if _, err := c.Search(context.Background(), "", "dsl", map[string]any{}); err == nil { | ||
|
|
@@ -72,32 +125,34 @@ func TestSearch_RejectsUnknownQueryType(t *testing.T) { | |
| } | ||
| } | ||
|
|
||
| func TestIndices_GetsAliasEndpoint(t *testing.T) { | ||
| var path string | ||
| c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { | ||
| path = r.URL.RequestURI() | ||
| _, _ = w.Write([]byte(`{}`)) | ||
| }) | ||
| defer srv.Close() | ||
| if _, err := c.Indices(context.Background()); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if path != "/_alias" { | ||
| t.Errorf("path = %q; want /_alias", path) | ||
| } | ||
| } | ||
|
|
||
| func TestIndexFields_GetsFieldCaps(t *testing.T) { | ||
| var path string | ||
| c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { | ||
| path = r.URL.RequestURI() | ||
| _, _ = w.Write([]byte(`{"fields":{}}`)) | ||
| }) | ||
| defer srv.Close() | ||
| if _, err := c.IndexFields(context.Background(), "logs-1"); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| if path != "/logs-1/_field_caps?fields=*" { | ||
| t.Errorf("path = %q; want /logs-1/_field_caps?fields=*", path) | ||
| } | ||
| } | ||
|
|
||
| func TestIndexFields_RequiresIndex(t *testing.T) { | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Update the documentation to specify that
ELASTICSEARCH_HEADERis comma-separated, matching the actual parsing implementation ofconfig.ParseHeadersand maintaining consistency withPROMETHEUS_HEADERS.