fix(runner/gcp): correct gke_logs filter and reshape gke_traces BigQuery output

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.

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 {

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

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
 			}