fix(runner/elasticsearch): restore legacy endpoints, PPL scoping, and header auth

[mayankpande88]

Three parity fixes so the ES primitives match what the backend composer
consumes (the composer is unchanged):

• query_es_indices: GET /_alias (object keyed by index) instead of
  /_cat/indices?format=json (an array). The composer iterates the result
  as index-keyed map entries, so the array shape yielded nothing.
• query_es_index_field: GET /{index}/_field_caps?fields=* instead of
  /{index}/_mapping, so the response carries the fields map the
  composer expects (legacy field_caps).
• query_es PPL: prepend source=<index> | when the query lacks a
  source= clause, scoping the query to the requested index as the legacy
  client did.

Also wire ELASTICSEARCH_HEADER (";"-separated) into the client's
ExtraHeaders, which were already applied per-request but never populated.

Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
Claude-Session: https://claude.ai/code/session_01L2HAXQH1gEtX7h4sUJsi9j

[gemini-code-assist]

Code Review

This pull request introduces support for custom Elasticsearch headers via the ELASTICSEARCH_HEADER environment variable and updates the Elasticsearch client to use the _alias and _field_caps endpoints for retrieving indices and fields. Additionally, it adds logic to automatically prepend a source clause to PPL queries when missing. Feedback highlights a discrepancy in the documentation and comments regarding header separation (which should be comma-separated rather than semicolon-separated) and suggests making the PPL source= check case-insensitive and whitespace-tolerant to prevent syntax errors, along with expanding the test suite to cover these cases.

[gemini-code-assist]

The current check !strings.Contains(s, "source=") is case-sensitive and does not account for optional whitespace around the = operator (e.g., source = index or SOURCE=index), which are both valid in OpenSearch PPL. If a query contains spaces or uppercase characters, this check will incorrectly evaluate to true and prepend another source= clause, leading to a syntax error.

We can perform a more robust, case-insensitive check by verifying if the query starts with source followed by optional spaces and =.

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

[gemini-code-assist]

There is a discrepancy between the comment/documentation and the actual implementation. The comment states that ELASTICSEARCH_HEADER is semicolon-separated, but the agent parses it using config.ParseHeaders (in runner/cmd/agent/main.go), which splits headers by commas (,).

To maintain consistency with PROMETHEUS_HEADERS and LOKI_EXTRA_HEADER (which are both comma-separated), we should update the comment and documentation to specify comma-separation.

Suggested change

	ElasticsearchHeaders string // ELASTICSEARCH_HEADER; ";"-separated "Header: value" pairs
	ElasticsearchHeaders string // ELASTICSEARCH_HEADER; comma-separated "Header: value" pairs

[gemini-code-assist]

Update the documentation to specify that ELASTICSEARCH_HEADER is comma-separated, matching the actual parsing implementation of config.ParseHeaders and maintaining consistency with PROMETHEUS_HEADERS.

Suggested change

	| `ELASTICSEARCH_HEADER` | optional | Semicolon-separated `Header: value` pairs sent on every ES request (same format as PROMETHEUS_HEADERS) |
	| `ELASTICSEARCH_HEADER` | optional | Comma-separated `Header: value` pairs sent on every ES request (same format as PROMETHEUS_HEADERS) |

[gemini-code-assist]

Let's expand the test suite to cover case-insensitivity and whitespace variations in the source= clause of PPL queries to ensure our robust parser behaves correctly.

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