pem: direct-query gRPC endpoint — stub + TDD contract

.github/workflows/vizier_release.yaml

@@ -15,7 +15,7 @@ jobs:
       image-base-name: "dev_image_with_extras"
   build-release:
     name: Build Release
-    runs-on: oracle-16cpu-64gb-x86-64
+    runs-on: oracle-vm-16cpu-64gb-x86-64
     needs: get-dev-image
     permissions:
       contents: read
@@ -140,7 +140,7 @@ jobs:
         git commit -s -m "Release Helm chart Vizier ${VERSION}"
         git push origin "gh-pages"
   update-gh-artifacts-manifest:
-    runs-on: oracle-8cpu-32gb-x86-64
+    runs-on: oracle-vm-16cpu-64gb-x86-64
     needs: [get-dev-image, create-github-release]
     container:
       image: ${{ needs.get-dev-image.outputs.image-with-tag }}

src/api/go/pxapi/opts.go

@@ -82,3 +82,17 @@ func WithDirectCredsInsecure() ClientOption {
 		c.insecureDirect = true
 	}
 }
+
+// WithDirectTLSSkipVerify is the secure-by-default option for direct (standalone /
+// node-local PEM) connections: the transport IS TLS-encrypted, but the server cert
+// is not chain/hostname-verified. Use this instead of WithDirectCredsInsecure when
+// the direct endpoint serves TLS with a self-signed / service cert whose SAN does
+// not match the node IP (e.g. vizier-pem's direct-query port served with
+// service-tls-certs, dialed at HOST_IP). Unlike WithDisableTLSVerification it does
+// NOT require a "cluster.local" address, so it works for the node-IP direct dial.
+// Bearer creds (the minted JWT) therefore ride an encrypted channel, never plaintext.
+func WithDirectTLSSkipVerify() ClientOption {
+	return func(c *Client) {
+		c.disableTLSVerification = true
+	}
+}

src/carnot/BUILD.bazel

@@ -44,6 +44,14 @@ pl_cc_library(
             "carnot_executable.cc",
         ],
     ),
+    # entlein/dx#29 — PEM's direct_query_server.cc needs concrete Carnot +
+    # EngineState defs to compile and execute PxL on the live engine. Same
+    # access pattern //src/experimental/standalone_pem already uses.
+    visibility = [
+        "//src/carnot:__subpackages__",
+        "//src/experimental:__subpackages__",
+        "//src/vizier/services/agent/pem:__pkg__",
+    ],
     deps = [
         "//src/carnot/exec:cc_library",
         "//src/carnot/exec/ml:cc_library",

src/carnot/exec/BUILD.bazel

@@ -36,6 +36,16 @@ pl_cc_library(
         "clickhouse_source_node.h",
         "exec_node.h",
         "exec_state.h",
+        # entlein/dx#29 — direct_query_server_test in
+        # //src/vizier/services/agent/pem needs LocalGRPCResultSinkServer
+        # for its Carnot fixture (same pattern as //src/carnot:carnot_test).
+        # Per-target visibility extension below keeps the rest of cc_library
+        # package-internal.
+        "local_grpc_result_server.h",
+    ],
+    visibility = [
+        "//src/carnot:__subpackages__",
+        "//src/vizier/services/agent/pem:__pkg__",
     ],
     deps = [
         "//src/carnot/carnotpb:carnot_pl_cc_proto",
@@ -61,6 +71,10 @@ pl_cc_library(
 pl_cc_test_library(
     name = "test_utils",
     hdrs = glob(["*test_utils.h"]),
+    visibility = [
+        "//src/carnot:__subpackages__",
+        "//src/vizier/services/agent/pem:__pkg__",  # entlein/dx#29 fixture
+    ],
     deps = [
         ":exec_node_test_helpers",
         "@com_github_apache_arrow//:arrow",

src/carnot/udf/BUILD.bazel

@@ -19,6 +19,9 @@ load("//bazel:pl_build_system.bzl", "pl_cc_binary", "pl_cc_library", "pl_cc_test
 package(default_visibility = [
     "//src/carnot:__subpackages__",
     "//src/vizier/funcs:__subpackages__",
+    # entlein/dx#29 — PEM's direct_query_server_test builds a CarnotTest-style
+    # fixture with a udf::Registry, same pattern as //src/carnot:carnot_test.
+    "//src/vizier/services/agent/pem:__pkg__",
 ])
 
 pl_cc_library(

src/stirling/source_connectors/socket_tracer/testing/container_images/BUILD.bazel

@@ -24,29 +24,29 @@ package(default_visibility = [
 
 # Generate all Go container library permutations for supported Go versions.
 go_container_libraries(
-    container_type = "grpc_server",
     bazel_sdk_versions = pl_all_supported_go_sdk_versions,
+    container_type = "grpc_server",
     prebuilt_container_versions = pl_go_test_versions,
 )
 
 # Stirling test cases usually test server side tracing. Therefore
 # we only need to provide the bazel SDK versions for the client containers.
 go_container_libraries(
-    container_type = "grpc_client",
     bazel_sdk_versions = pl_all_supported_go_sdk_versions,
+    container_type = "grpc_client",
 )
 
 go_container_libraries(
-    container_type = "tls_server",
     bazel_sdk_versions = pl_all_supported_go_sdk_versions,
+    container_type = "tls_server",
     prebuilt_container_versions = pl_go_test_versions,
 )
 
 # Stirling test cases usually test server side tracing. Therefore
 # we only need to provide the bazel SDK versions for the client containers.
 go_container_libraries(
-    container_type = "tls_client",
     bazel_sdk_versions = pl_all_supported_go_sdk_versions,
+    container_type = "tls_client",
 )
 
 pl_cc_test_library(

src/utils/shared/k8s/apply.go

@@ -30,8 +30,8 @@ import (
 	"strings"
 
 	log "github.com/sirupsen/logrus"
-	"k8s.io/apimachinery/pkg/api/meta"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"

src/utils/shared/k8s/delete.go

@@ -123,8 +123,8 @@ func (o *ObjectDeleter) DeleteNamespace() error {
 // groups are skipped during cluster-wide deletion sweeps because aggregated
 // servers frequently advertise the delete verb on read-only virtual resources
 // and fail the call with "operation not supported".
-func (o *ObjectDeleter) getAggregatedGroupVersions() (sets.String, error) {
-	out := sets.NewString()
+func (o *ObjectDeleter) getAggregatedGroupVersions() (sets.Set[string], error) {
+	out := sets.New[string]()
 	list, err := o.dynamicClient.Resource(apiServiceGVR).List(context.TODO(), metav1.ListOptions{})
 	if err != nil {
 		if errors.IsNotFound(err) || meta.IsNoMatchError(err) {
@@ -173,7 +173,7 @@ func (o *ObjectDeleter) getDeletableResourceTypes() ([]string, error) {
 			if len(resource.Verbs) == 0 {
 				continue
 			}
-			if !sets.NewString(resource.Verbs...).HasAll("delete") {
+			if !sets.New[string](resource.Verbs...).HasAll("delete") {
 				continue
 			}
 			resources = append(resources, resource.Name)

src/vizier/services/agent/pem/BUILD.bazel

@@ -20,6 +20,23 @@ load("//bazel:pl_build_system.bzl", "pl_cc_binary", "pl_cc_library", "pl_cc_test
 
 package(default_visibility = ["//src/vizier:__subpackages__"])
 
+# entlein/dx#29 — compile-time kill switch for the direct-query endpoint.
+# Operators who do not want the feature available IN THE BINARY at all build
+# with `bazel build … --//src/vizier/services/agent/pem:direct_query=disabled`
+# (or equivalently `--define=PX_PEM_DIRECT_QUERY=disabled` if invoking via
+# bazel's legacy --define flag). When disabled, this propagates
+# `-DPX_PEM_DIRECT_QUERY_DISABLED` into every translation unit in cc_library;
+# direct_query_server.cc's body, pem_manager.cc's flag DEFINEs, and
+# MaybeStartDirectQueryServer's implementation are all `#ifndef`'d out, and
+# the stub class methods return UNIMPLEMENTED for any caller that still
+# routes through DirectQueryServer. The runtime flag --direct_query_enabled
+# is the soft, per-deploy toggle; this is the hard, per-build toggle. See
+# DIRECT_QUERY_SECURITY.md "Disabling the feature".
+config_setting(
+    name = "direct_query_disabled",
+    define_values = {"PX_PEM_DIRECT_QUERY": "disabled"},
+)
+
 pl_cc_library(
     name = "cc_library",
     srcs = glob(
@@ -30,7 +47,34 @@ pl_cc_library(
         ],
     ),
     hdrs = glob(["*.h"]),
+    defines = select({
+        ":direct_query_disabled": ["PX_PEM_DIRECT_QUERY_DISABLED"],
+        "//conditions:default": [],
+    }),
     deps = [
+        # entlein/dx#29 direct-query server deps (direct_query_server.{h,cc}):
+        "//src/api/proto/vizierpb:vizier_pl_cc_proto",
+        # Use //src/carnot:carnot (the public header target) for forward decls
+        # in direct_query_server.h. Step 2 (#29) needs concrete types — pull
+        # additional carnot sub-targets exposed via the fork's PEM visibility:
+        "//src/carnot",
+        "//src/carnot:cc_library",  # Carnot, EngineState concrete defs
+        "//src/carnot/carnotpb:carnot_pl_cc_proto",  # TransferResultChunkRequest
+        "//src/carnot/exec:cc_library",  # LocalGRPCResultSinkServer
+        "//src/carnot/funcs:cc_library",  # Step 9: funcs::RegisterFuncsOrDie
+        "//src/carnot/planner/compiler:cc_library",  # Compiler().Compile(...)
+        "//src/carnot/planpb:plan_pl_cc_proto",  # planpb::Plan, OperatorType
+        "//src/carnot/udf:cc_library",  # Step 9: udf::Registry
+        # Step 1 (#29): HS256 verify uses BoringSSL HMAC directly and parses
+        # claims via rapidjson. cpp_jwt's own HMAC verify path calls
+        # BIO_f_base64 which is in boringssl/decrepit/ — not exposed as a
+        # bazel target on this fork. Native HMAC is ~50 LoC vs. a boringssl
+        # patch; the mint side (test only, via cpp_jwt) still uses the lib.
+        "@boringssl//:crypto",
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_github_rlyeh_sole//:sole",  # sole::uuid4 for query_id
+        "@com_github_tencent_rapidjson//:rapidjson",
+        # existing PEM deps:
         "//src/carnot/planner/dynamic_tracing/ir/logicalpb:logical_pl_cc_proto",
         "//src/integrations/grpc_clocksync:cc_library",
         "//src/shared/tracepoint_translation:cc_library",
@@ -51,6 +95,32 @@ pl_cc_test(
     ],
 )
 
+# entlein/dx#29 — TDD contract for the PEM direct-query endpoint. dx-agent authored
+# the spec; pem-agent makes it pass (port the standalone_pem execution path + JWT
+# verify + Carnot fixture). See DIRECT_QUERY_CONTRACT.md.
+#
+# Note: pl_cc_test auto-injects //src/common/testing:cc_library for gtest/gmock,
+# so we must NOT list it explicitly here — same shape as tracepoint_manager_test
+# above (the dx-agent flagged this as a likely nit in the kickoff comment).
+pl_cc_test(
+    name = "direct_query_server_test",
+    srcs = ["direct_query_server_test.cc"],
+    deps = [
+        ":cc_library",
+        "//src/api/proto/vizierpb:vizier_pl_cc_proto",
+        # Step 2b (#29) — Carnot exec fixture for ValidToken_TrivialQuery_StreamsRows
+        "//src/carnot",
+        "//src/carnot:cc_library",
+        "//src/carnot/exec:cc_library",  # LocalGRPCResultSinkServer
+        "//src/carnot/funcs:cc_library",  # RegisterFuncsOrDie
+        "//src/carnot/udf:cc_library",  # udf::Registry
+        "//src/table_store:cc_library",
+        "@com_github_arun11299_cpp_jwt//:cpp_jwt",  # Step 1: MakeBearerToken mints HS256
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_github_rlyeh_sole//:sole",
+    ],
+)
+
 pl_cc_binary(
     name = "pem",
     srcs = ["pem_main.cc"],

src/vizier/services/agent/pem/DIRECT_QUERY_CONTRACT.md

@@ -0,0 +1,88 @@
+# PEM direct-query gRPC endpoint — contract (entlein/dx#29)
+
+**Status:** stub PR. dx-agent owns this contract + the dx-side integration; the
+**pem-agent** (on a bazel-capable build VM) implements the C++ to make the TDD
+targets in `direct_query_server_test.cc` pass.
+
+## Why
+
+Today dx reads evidence by querying the in-cluster **vizier-query-broker** directly
+(entlein/dx#28, live). That works and is the MVP path. #29 adds the *durable
+per-node* path: make the **normal `vizier-pem`** itself serve `ExecuteScript`
+directly over gRPC, so dx on each node can query its node-local PEM with no broker
+hop and no cloud dependency.
+
+This is the capability the **experimental `standalone_pem`** already proved
+(`src/experimental/standalone_pem/vizier_server.h` — `px::vizier::agent::VizierServer`
+implementing `api::vizierpb::VizierService::ExecuteScript` against a local Carnot).
+The two differences for the real PEM:
+
+1. **Metadata-connected.** The normal PEM has the metadata service, so per-pod PxL
+   filters (`df[df.ctx['pod'] == ...]`) resolve — the gap that made standalone_pem
+   return empty per-pod results (old #15). Reuse the PEM's existing Carnot +
+   table_store + metadata state; do **not** stand up a second Carnot.
+2. **Authenticated.** standalone_pem was insecure (`WithDirectCredsInsecure`). The
+   real PEM is in `pl` and must require a **valid cluster service JWT** (the same
+   `jwt-signing-key` dx already mints with — see entlein/dx `cmd/dx-daemon/pxbroker.go`).
+
+## The endpoint
+
+- Service: `px.api.vizierpb.VizierService` (the generated gRPC service).
+- Method implemented: **`ExecuteScript`** (server-streaming). Mutations/tracepoints
+  are **out of scope** for #29 — return `UNIMPLEMENTED` for `req.mutation()==true`.
+  (standalone_pem handles mutations; the dx read path never mutates.)
+- Transport: gRPC over TLS. TLS may use the in-cluster self-signed CA (dx sets
+  `PX_DISABLE_TLS=1` to skip-verify, matching the broker path).
+
+## Config (flags / env) — gated OFF by default
+
+| flag / env                          | default | meaning                                            |
+|-------------------------------------|---------|----------------------------------------------------|
+| `--direct_query_enabled` / `PL_PEM_DIRECT_QUERY_ENABLED` | `false` | master switch; when false the port is never opened |
+| `--direct_query_port` / `PL_PEM_DIRECT_QUERY_PORT`       | `50305` | gRPC listen port for the direct-query service      |
+| `--direct_query_jwt_signing_key` / `PL_JWT_SIGNING_KEY`  | `""`    | HMAC key the bearer JWT must verify against         |
+
+Default-off so existing PEM deployments are byte-for-byte unchanged until opted in.
+
+## Auth contract
+
+Every `ExecuteScript` call MUST present `authorization: Bearer <jwt>` metadata.
+The JWT is verified with `PL_JWT_SIGNING_KEY` and must:
+- have a valid signature (HS256) against the signing key,
+- be unexpired (`exp` in the future),
+- carry a service/audience claim acceptable to vizier (dx mints
+  `GenerateJWTForService("dx", "vizier")` — see `src/shared/services/utils`).
+
+Missing/invalid/expired token → `grpc::StatusCode::UNAUTHENTICATED`. No token must
+ever fall through to query execution.
+
+## Behavioral contract (the executable spec → `direct_query_server_test.cc`)
+
+1. **flag-off → no listener.** With `direct_query_enabled=false`, nothing listens on
+   the port; the PEM starts exactly as today.
+2. **flag-on → serves ExecuteScript.** With it enabled + a signing key set, a gRPC
+   client with a valid bearer JWT gets a streamed response (status OK) for a trivial
+   PxL (e.g. `import px; px.display(px.DataFrame('http_events'))`).
+3. **auth required.** Same call with (a) no token, (b) a token signed by the wrong
+   key, (c) an expired token → each `UNAUTHENTICATED`, no rows.
+4. **metadata-connected filter.** A PxL with a per-pod filter returns only that pod's
+   rows (proves the metadata gap #15 is closed on the real PEM). May be an
+   integration test tagged `requires_metadata` if a unit Carnot fixture can't supply
+   pod context.
+5. **mutation rejected.** `req.mutation()==true` → `UNIMPLEMENTED` (scope guard).
+6. **no regression.** The existing PEM agent registration / Carnot / Stirling path is
+   unchanged when the flag is off (assert via the existing PEM smoke/unit tests).
+
+## dx-side integration (owned by dx-agent — informational)
+
+dx already has the client: `cmd/dx-daemon/pxbroker.go`. Pointing it at a per-PEM
+addr is a one-line switch — add `DX_BENCH=pemdirect` selecting
+`DX_PEM_DIRECT_ADDR=<HOST_IP>:50305` (HOST_IP via downward API), reusing the exact
+JWT mint + `WithBearerAuth` + `WithDisableTLSVerification` path proven against the
+broker. dx-agent adds this once #29's endpoint is live; no PEM-side work needed for it.
+
+## Done =
+
+`direct_query_server_test.cc` green under `bazel test`, PEM image builds via the
+vizier-release workflow, and a live PG shows dx (`DX_BENCH=pemdirect`) ruling in the
+log4shell e2e off the node-local PEM with the same verdict it gets via the broker.