diff --git a/charts/nudgebee-agent/values.yaml b/charts/nudgebee-agent/values.yaml index bbf14b10..e3beaa5d 100644 --- a/charts/nudgebee-agent/values.yaml +++ b/charts/nudgebee-agent/values.yaml @@ -65,7 +65,7 @@ runnerServiceAccount: runner: image: repository: ghcr.io/nudgebee/nudgebee-agent - tag: 2026-08-12T06-01-54_7ec58f7701909a3ce172ad2a9235f8b15255e363 + tag: 2026-08-12T08-10-04_e13ce92bc8c238e883d050b1276c6128b2c8fb44 # Image template the pod_profiler action launches debugger pods from. # The agent substitutes `{}` for the variant (bpf, jvm, python, perf, ruby). # Surfaces as PROFILER_IMAGE; leave empty to fall back to the binary default. diff --git a/runner/cmd/agent/main.go b/runner/cmd/agent/main.go index bafb2ec8..8e8d59c8 100644 --- a/runner/cmd/agent/main.go +++ b/runner/cmd/agent/main.go @@ -375,6 +375,7 @@ func run(ctx context.Context, logger *slog.Logger, cfg *config.Config) error { ec.Username = cfg.ElasticsearchUser ec.Password = cfg.ElasticsearchPassword ec.APIKey = cfg.ElasticsearchAPIKey + ec.ExtraHeaders = config.ParseHeaders(cfg.ElasticsearchHeaders) registerProxy("elasticsearch", esEnabled, elasticsearch.Handlers(ec)) if esEnabled { logger.Info("elasticsearch enabled", "url", cfg.ElasticsearchURL) diff --git a/runner/docs/configuration.md b/runner/docs/configuration.md index 0ebf338b..99e05e7e 100644 --- a/runner/docs/configuration.md +++ b/runner/docs/configuration.md @@ -55,6 +55,7 @@ If a K8s subsystem is enabled but the agent fails to build a K8s client (no kube | `ELASTICSEARCH_URL` | optional | Enables `query_es*` actions | | `ELASTICSEARCH_USERNAME` / `_PASSWORD` | optional | Basic auth (one of these or `_APIKEY` is needed if ES requires auth) | | `ELASTICSEARCH_APIKEY` | optional | Alternative to user/pass | +| `ELASTICSEARCH_HEADER` | optional | Semicolon-separated `Header: value` pairs sent on every ES request (same format as PROMETHEUS_HEADERS) | | `ELASTICSEARCH_SSL_VERIFY` | optional | Verify the ES server TLS cert (https only). Default `false` (skip verify), matching the legacy client | | `SIGNOZ_URL` / `SIGNOZ_API_KEY` | optional | `signoz_*` actions; API key sent as `SIGNOZ-API-KEY` header | | `SIGNOZ_USER` / `SIGNOZ_PASSWORD` | optional | Alternative auth: JWT minted via `/api/v1/login` (used when `SIGNOZ_API_KEY` is unset) | diff --git a/runner/pkg/config/config.go b/runner/pkg/config/config.go index 366eb4f5..3366f60e 100644 --- a/runner/pkg/config/config.go +++ b/runner/pkg/config/config.go @@ -54,8 +54,9 @@ type Config struct { ElasticsearchUser string ElasticsearchPassword string ElasticsearchAPIKey string - ElasticsearchEnabled bool // ELASTICSEARCH_ENABLED; default true when URL is set - ElasticsearchSSLVerify bool // ELASTICSEARCH_SSL_VERIFY; default false (skip cert verify) to match legacy + ElasticsearchHeaders string // ELASTICSEARCH_HEADER; ";"-separated "Header: value" pairs + ElasticsearchEnabled bool // ELASTICSEARCH_ENABLED; default true when URL is set + ElasticsearchSSLVerify bool // ELASTICSEARCH_SSL_VERIFY; default false (skip cert verify) to match legacy // Signoz SignozURL string @@ -231,6 +232,7 @@ func FromEnv() (*Config, error) { ElasticsearchUser: os.Getenv("ELASTICSEARCH_USERNAME"), ElasticsearchPassword: os.Getenv("ELASTICSEARCH_PASSWORD"), ElasticsearchAPIKey: os.Getenv("ELASTICSEARCH_APIKEY"), + ElasticsearchHeaders: os.Getenv("ELASTICSEARCH_HEADER"), // ELASTICSEARCH_ENABLED is the explicit opt-in for ES as the logs // provider, defaulting false to match the legacy agent // (env_vars.ELASTICSEARCH_ENABLED = load_bool(..., False)) and the prod diff --git a/runner/pkg/observability/elasticsearch/client.go b/runner/pkg/observability/elasticsearch/client.go index e292e88e..0b9ac4d1 100644 --- a/runner/pkg/observability/elasticsearch/client.go +++ b/runner/pkg/observability/elasticsearch/client.go @@ -2,9 +2,9 @@ // Same shape as pkg/observability/prometheus. // // Action surface: -// - query_es : POST {index}/_search with DSL, or {index}/_plugins/_ppl with PPL -// - query_es_indices : GET _cat/indices?format=json -// - query_es_index_field : GET {index}/_mapping +// - 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,6 +56,18 @@ 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= | `. 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 + } + } body, _ := json.Marshal(map[string]any{"query": s}) return c.do(ctx, http.MethodPost, "/_plugins/_ppl", body, "application/json") default: @@ -63,17 +75,24 @@ func (c *Client) Search(ctx context.Context, index, queryType string, query any) } } -// Indices returns _cat/indices in JSON form. +// 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, "/_cat/indices?format=json", nil, "") + return c.do(ctx, http.MethodGet, "/_alias", nil, "") } -// IndexFields returns mapping for an index. +// 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+"/_mapping", nil, "") + return c.do(ctx, http.MethodGet, "/"+index+"/_field_caps?fields=*", nil, "") } // FieldValues returns distinct values for one field via a terms aggregation. diff --git a/runner/pkg/observability/elasticsearch/elasticsearch_test.go b/runner/pkg/observability/elasticsearch/elasticsearch_test.go index d0235272..b62d3feb 100644 --- a/runner/pkg/observability/elasticsearch/elasticsearch_test.go +++ b/runner/pkg/observability/elasticsearch/elasticsearch_test.go @@ -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) + } +} + +// 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_GetsCatEndpoint(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(`[]`)) + _, _ = w.Write([]byte(`{}`)) }) defer srv.Close() if _, err := c.Indices(context.Background()); err != nil { t.Fatal(err) } - if path != "/_cat/indices?format=json" { - t.Errorf("path = %q", path) + if path != "/_alias" { + t.Errorf("path = %q; want /_alias", path) } } -func TestIndexFields_GetsMapping(t *testing.T) { +func TestIndexFields_GetsFieldCaps(t *testing.T) { + var path string c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path != "/logs-1/_mapping" { - t.Errorf("path = %q", r.URL.Path) - } - _, _ = w.Write([]byte(`{}`)) + 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) {