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/pkg/observability/gcp/client.go b/runner/pkg/observability/gcp/client.go index f560ef21..3f1a79fd 100644 --- a/runner/pkg/observability/gcp/client.go +++ b/runner/pkg/observability/gcp/client.go @@ -9,8 +9,8 @@ // gcloud user creds. // // Action surface: -// - gke_logs : Cloud Logging entries for a GKE node pool in a zone -// - gke_traces : arbitrary BigQuery SQL (used for traces stored in BQ) +// - gke_logs : GKE cluster-autoscaler visibility logs for a node pool +// - gke_traces : arbitrary BigQuery SQL (reshaped to {data,columns,column_types}) package gcp import ( @@ -63,8 +63,12 @@ func NewWithHTTP(c *http.Client) *Client { return &Client{HTTP: c} } -// FetchNodePoolLogs queries Cloud Logging for GCE-instance entries scoped -// to a zone. +// FetchNodePoolLogs queries Cloud Logging for the GKE cluster-autoscaler +// visibility logs (node-pool scale-up/scale-down decisions), matching the +// legacy gcloud_client. These logs live under the k8s_cluster resource and +// are keyed by the cluster's location — zonal clusters use the zone, regional +// clusters use the region — so we try the zone first and fall back to the +// derived region. // // projectID : GCP project (required) // zone : compute zone, e.g. "us-central1-a" (required) @@ -80,22 +84,72 @@ func (c *Client) FetchNodePoolLogs(ctx context.Context, projectID, zone string, limit = 100 } - body := map[string]any{ + locations := []string{zone} + if region := zoneToRegion(zone); region != "" && region != zone { + locations = append(locations, region) + } + + var last json.RawMessage + for _, loc := range locations { + raw, err := c.postJSON(ctx, c.loggingURL()+"/v2/entries:list", nodePoolLogsBody(projectID, loc, limit)) + if err != nil { + return nil, err + } + last = raw + if hasLogEntries(raw) { + return raw, nil + } + } + return last, nil +} + +// nodePoolLogsBody builds the entries:list request scoped to one location. +func nodePoolLogsBody(projectID, location string, limit int) map[string]any { + return map[string]any{ "resourceNames": []string{"projects/" + projectID}, "filter": fmt.Sprintf( - `resource.type="gce_instance" AND resource.labels.zone="%s"`, - zone, + `resource.type="k8s_cluster" AND resource.labels.project_id="%s" `+ + `AND resource.labels.location="%s" `+ + `AND logName="projects/%s/logs/container.googleapis.com%%2Fcluster-autoscaler-visibility" `+ + `AND severity>=DEFAULT`, + projectID, location, projectID, ), "pageSize": limit, "orderBy": "timestamp desc", } - return c.postJSON(ctx, c.loggingURL()+"/v2/entries:list", body) } -// QueryBigQuery runs an arbitrary SQL query as a synchronous query job. -// Does not perform server-side pagination (callers needing pagination iterate -// with maxResults + pageToken via a follow-up action). -func (c *Client) QueryBigQuery(ctx context.Context, projectID, query string) (json.RawMessage, error) { +// zoneToRegion strips the trailing zone suffix, e.g. "us-central1-a" -> +// "us-central1". Returns the input unchanged when it has no suffix. +func zoneToRegion(zone string) string { + if len(zone) > 2 && zone[len(zone)-2] == '-' { + lastChar := zone[len(zone)-1] + if lastChar >= 'a' && lastChar <= 'z' { + return zone[:len(zone)-2] + } + } + return zone +} + +// hasLogEntries reports whether a Cloud Logging entries:list response carries +// at least one entry. +func hasLogEntries(raw json.RawMessage) bool { + var resp struct { + Entries []json.RawMessage `json:"entries"` + } + if err := json.Unmarshal(raw, &resp); err != nil { + return false + } + return len(resp.Entries) > 0 +} + +// QueryBigQuery runs an arbitrary SQL query as a synchronous query job and +// reshapes the BigQuery REST response into the {data, columns, column_types} +// envelope the backend warehouse consumer expects (mirroring the legacy +// run_bigquery). location, when non-empty, scopes the job to the dataset's +// region (required for non-US/EU datasets). Does not perform server-side +// pagination. +func (c *Client) QueryBigQuery(ctx context.Context, projectID, query, location string) (json.RawMessage, error) { if projectID == "" { return nil, errors.New("gcp: project_id required") } @@ -108,8 +162,63 @@ func (c *Client) QueryBigQuery(ctx context.Context, projectID, query string) (js "timeoutMs": 30000, // No maxResults — let BigQuery default; backend can cap on its end. } + if location != "" { + body["location"] = location + } url := c.bigQueryURL() + "/bigquery/v2/projects/" + projectID + "/queries" - return c.postJSON(ctx, url, body) + raw, err := c.postJSON(ctx, url, body) + if err != nil { + return nil, err + } + return reshapeBigQuery(raw) +} + +// reshapeBigQuery converts the BigQuery jobs.query REST response +// ({schema:{fields:[{name,type}]}, rows:[{f:[{v}]}]}) into the +// {data, columns, column_types} shape the backend warehouse consumer reads. +func reshapeBigQuery(raw json.RawMessage) (json.RawMessage, error) { + var resp struct { + Schema struct { + Fields []struct { + Name string `json:"name"` + Type string `json:"type"` + } `json:"fields"` + } `json:"schema"` + Rows []struct { + F []struct { + V json.RawMessage `json:"v"` + } `json:"f"` + } `json:"rows"` + } + if err := json.Unmarshal(raw, &resp); err != nil { + return nil, fmt.Errorf("gcp: parse bigquery response: %w", err) + } + + columns := make([]string, len(resp.Schema.Fields)) + columnTypes := make([]string, len(resp.Schema.Fields)) + for i, f := range resp.Schema.Fields { + columns[i] = f.Name + columnTypes[i] = f.Type + } + + data := make([][]any, 0, len(resp.Rows)) + for _, row := range resp.Rows { + vals := make([]any, len(row.F)) + for i, cell := range row.F { + var v any + // BigQuery cell values are JSON scalars (usually strings); keep the + // decoded value, or nil on an unexpected shape. + _ = json.Unmarshal(cell.V, &v) + vals[i] = v + } + data = append(data, vals) + } + + return json.Marshal(map[string]any{ + "data": data, + "columns": columns, + "column_types": columnTypes, + }) } func (c *Client) loggingURL() string { diff --git a/runner/pkg/observability/gcp/gcp_test.go b/runner/pkg/observability/gcp/gcp_test.go index dcc70061..b9ee4e98 100644 --- a/runner/pkg/observability/gcp/gcp_test.go +++ b/runner/pkg/observability/gcp/gcp_test.go @@ -24,7 +24,8 @@ func TestFetchNodePoolLogs_PostsExpectedFilter(t *testing.T) { path = r.URL.Path b, _ := io.ReadAll(r.Body) body = string(b) - _, _ = w.Write([]byte(`{"entries":[]}`)) + // Return an entry so the zone→region fallback does not fire. + _, _ = w.Write([]byte(`{"entries":[{"textPayload":"x"}]}`)) })) defer srv.Close() @@ -39,14 +40,54 @@ func TestFetchNodePoolLogs_PostsExpectedFilter(t *testing.T) { if !strings.Contains(body, `"projects/my-proj"`) { t.Errorf("missing resourceNames: %s", body) } - if !strings.Contains(body, `resource.labels.zone=\"us-central1-a\"`) { - t.Errorf("missing zone filter: %s", body) + if !strings.Contains(body, `resource.type=\"k8s_cluster\"`) { + t.Errorf("missing k8s_cluster resource filter: %s", body) + } + if !strings.Contains(body, `resource.labels.location=\"us-central1-a\"`) { + t.Errorf("missing location filter: %s", body) + } + if !strings.Contains(body, `cluster-autoscaler-visibility`) { + t.Errorf("missing autoscaler-visibility logName: %s", body) + } + // The severity clause uses `>=` which JSON-escapes the `>`; match the + // stable parts to avoid depending on the escaping. + if !strings.Contains(body, `severity`) || !strings.Contains(body, `=DEFAULT`) { + t.Errorf("missing severity>=DEFAULT filter: %s", body) } if !strings.Contains(body, `"pageSize":50`) { t.Errorf("missing pageSize: %s", body) } } +// TestFetchNodePoolLogs_ZoneRegionFallback: when the zone-scoped query +// returns no entries, the client retries with the derived region. +func TestFetchNodePoolLogs_ZoneRegionFallback(t *testing.T) { + var locations []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + s := string(b) + if strings.Contains(s, `location=\"us-central1-a\"`) { + locations = append(locations, "zone") + _, _ = w.Write([]byte(`{"entries":[]}`)) // empty → triggers fallback + return + } + if strings.Contains(s, `location=\"us-central1\"`) { + locations = append(locations, "region") + } + _, _ = w.Write([]byte(`{"entries":[{"textPayload":"x"}]}`)) + })) + defer srv.Close() + + c := newTestClient() + c.LoggingBaseURL = srv.URL + if _, err := c.FetchNodePoolLogs(context.Background(), "p", "us-central1-a", 10); err != nil { + t.Fatal(err) + } + if len(locations) != 2 || locations[0] != "zone" || locations[1] != "region" { + t.Errorf("fallback order = %v; want [zone region]", locations) + } +} + func TestFetchNodePoolLogs_DefaultLimit(t *testing.T) { var body string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -86,7 +127,7 @@ func TestQueryBigQuery_PostsToProjectQueriesEndpoint(t *testing.T) { defer srv.Close() c := newTestClient() c.BigQueryBaseURL = srv.URL - if _, err := c.QueryBigQuery(context.Background(), "my-proj", "SELECT 1"); err != nil { + if _, err := c.QueryBigQuery(context.Background(), "my-proj", "SELECT 1", "US"); err != nil { t.Fatal(err) } if path != "/bigquery/v2/projects/my-proj/queries" { @@ -95,14 +136,52 @@ func TestQueryBigQuery_PostsToProjectQueriesEndpoint(t *testing.T) { if !strings.Contains(body, `"query":"SELECT 1"`) || !strings.Contains(body, `"useLegacySql":false`) { t.Errorf("body = %s", body) } + if !strings.Contains(body, `"location":"US"`) { + t.Errorf("location not passed in job body: %s", body) + } +} + +// TestQueryBigQuery_ReshapesResponse verifies the BQ REST response is +// transformed into the {data, columns, column_types} envelope. +func TestQueryBigQuery_ReshapesResponse(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{ + "schema":{"fields":[{"name":"svc","type":"STRING"},{"name":"cnt","type":"INTEGER"}]}, + "rows":[{"f":[{"v":"frontend"},{"v":"42"}]},{"f":[{"v":"cart"},{"v":"7"}]}] + }`)) + })) + defer srv.Close() + c := newTestClient() + c.BigQueryBaseURL = srv.URL + raw, err := c.QueryBigQuery(context.Background(), "p", "SELECT svc, cnt FROM t", "") + if err != nil { + t.Fatal(err) + } + var out struct { + Data [][]any `json:"data"` + Columns []string `json:"columns"` + ColumnTypes []string `json:"column_types"` + } + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatalf("reshaped result not the expected envelope: %v", err) + } + if len(out.Columns) != 2 || out.Columns[0] != "svc" || out.Columns[1] != "cnt" { + t.Errorf("columns = %v", out.Columns) + } + if len(out.ColumnTypes) != 2 || out.ColumnTypes[0] != "STRING" || out.ColumnTypes[1] != "INTEGER" { + t.Errorf("column_types = %v", out.ColumnTypes) + } + if len(out.Data) != 2 || out.Data[0][0] != "frontend" || out.Data[1][1] != "7" { + t.Errorf("data = %v", out.Data) + } } func TestQueryBigQuery_RequiresProjectAndQuery(t *testing.T) { c := newTestClient() - if _, err := c.QueryBigQuery(context.Background(), "", "SELECT 1"); err == nil { + if _, err := c.QueryBigQuery(context.Background(), "", "SELECT 1", ""); err == nil { t.Error("missing project should error") } - if _, err := c.QueryBigQuery(context.Background(), "p", ""); err == nil { + if _, err := c.QueryBigQuery(context.Background(), "p", "", ""); err == nil { t.Error("missing query should error") } } diff --git a/runner/pkg/observability/gcp/handlers.go b/runner/pkg/observability/gcp/handlers.go index 5ab80b9b..9771d390 100644 --- a/runner/pkg/observability/gcp/handlers.go +++ b/runner/pkg/observability/gcp/handlers.go @@ -25,7 +25,14 @@ func Handlers(c *Client, defaultProjectID string) map[string]dispatch.Handler { }, "gke_traces": func(ctx context.Context, p map[string]any) (any, error) { project := strOrDefault(p, "project_id", defaultProjectID) - raw, err := c.QueryBigQuery(ctx, project, str(p, "query")) + // Dataset location: explicit `location`, else derive from `zone`. + location := str(p, "location") + if location == "" { + if z := str(p, "zone"); z != "" { + location = zoneToRegion(z) + } + } + raw, err := c.QueryBigQuery(ctx, project, str(p, "query"), location) if err != nil { return nil, err }