From 806df9a77e946c97d673d7457b78b42931da74d6 Mon Sep 17 00:00:00 2001 From: Yufan Su Date: Sun, 23 Aug 2026 23:45:31 -0700 Subject: [PATCH 1/2] e2e: cover egress on grpc requests and bidi streaming --- demos/egress/README.md | 36 ++- demos/egress/main.go | 211 +++++++++++++- demos/egress/main_test.go | 211 ++++++++++++++ .../egressprobe/egressprobe.yaml.tmpl | 113 -------- internal/e2e/fixtures/grpcecho/main.go | 115 ++++++++ internal/e2e/fixtures/grpcecho/main_test.go | 194 +++++++++++++ internal/e2e/fixtures/serverpod.yaml.tmpl | 86 ++++++ internal/e2e/manifest.go | 117 ++++++++ internal/e2e/pod.go | 74 +++++ internal/e2e/probe.go | 33 +-- internal/e2e/sandbox.go | 41 +-- internal/e2e/serverpod.go | 153 ++++++++++ internal/e2e/serverpod_test.go | 222 +++++++++++++++ .../e2e/suites/networking/grpcegress_test.go | 157 +++++++++++ .../e2e/suites/networking/networking_test.go | 21 +- .../e2e/suites/sdsmint/actoridentity_test.go | 6 +- internal/e2e/suites/sdsmint/sdsmint_test.go | 125 +++++---- internal/proto/grpcechopb/gen.go | 17 ++ internal/proto/grpcechopb/grpcecho.pb.go | 265 ++++++++++++++++++ internal/proto/grpcechopb/grpcecho.proto | 72 +++++ internal/proto/grpcechopb/grpcecho_grpc.pb.go | 257 +++++++++++++++++ 21 files changed, 2276 insertions(+), 250 deletions(-) delete mode 100644 internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl create mode 100644 internal/e2e/fixtures/grpcecho/main.go create mode 100644 internal/e2e/fixtures/grpcecho/main_test.go create mode 100644 internal/e2e/fixtures/serverpod.yaml.tmpl create mode 100644 internal/e2e/manifest.go create mode 100644 internal/e2e/pod.go create mode 100644 internal/e2e/serverpod.go create mode 100644 internal/e2e/serverpod_test.go create mode 100644 internal/e2e/suites/networking/grpcegress_test.go create mode 100644 internal/proto/grpcechopb/gen.go create mode 100644 internal/proto/grpcechopb/grpcecho.pb.go create mode 100644 internal/proto/grpcechopb/grpcecho.proto create mode 100644 internal/proto/grpcechopb/grpcecho_grpc.pb.go diff --git a/demos/egress/README.md b/demos/egress/README.md index 8a1b6f7784..bbe4ccff0f 100644 --- a/demos/egress/README.md +++ b/demos/egress/README.md @@ -53,7 +53,7 @@ intercepted and carried over mTLS to a gateway that verifies who is making the r ## Components - **Egress app (`main.go`)** — the Actor: `POST /` with `{"url":"..."}` → fetches it → returns - status + body. + status + body. It also serves `POST /grpc`, described below. - **Egress gateway** — `manifests/ate-install/atenet-egress.yaml`. One pod, two containers: an Envoy (`envoy`) and the atenet router ext_proc (`ext-proc`, `--mode=egress`), called over localhost. In egress mode the router serves the egress ext_proc handler only — no xDS server, @@ -132,6 +132,40 @@ kubectl -n ate-system logs deploy/atenet-egress -c ext-proc | grep -i 'egress id The `whoami` body shows `RemoteAddr: ` — proof the request egressed *through* the gateway rather than directly. +## gRPC over the same tunnel + +`POST /grpc` with `{"target":":","message":"hello","streamCount":2,"bidiCount":2}` makes +the Actor dial that address as a cleartext-HTTP/2 gRPC server and return what came back: + +```json +{ + "message": "hello", + "stream": [{"message":"hello","index":0},{"message":"hello","index":1}], + "bidi": [{"message":"hello-0","index":0},{"message":"hello-1","index":1}], + "code": "OK" +} +``` + +One RPC per streaming shape, because each fails differently over a network path: + +- **unary `Echo`** — always run. Its status arrives in trailers, *after* the response body, which + is what `code` reports and what a path that dropped trailers or downgraded to HTTP/1.1 could not + produce. +- **server-streaming `EchoStream`** — when `streamCount` is positive. Many frames over one + held-open connection. +- **bidirectional `EchoBidi`** — when `bidiCount` is positive. The Actor sends each message only + after reading the response to the previous one, then half-closes the request direction while the + response direction is still open. A path that serialized the two directions hangs here rather + than answering short. + +The gateway terminates the `CONNECT` and relays opaque TCP, so all of this crosses it untouched. A +failed RPC answers `502` and still carries its gRPC code in the same field. + +`internal/e2e/suites/networking` drives this endpoint against the `grpcecho` fixture +(`internal/e2e/fixtures/grpcecho`, deployed by `e2e.DeployOriginPod` from the shared origin +manifest) in `TestActorEgressGRPC`; the same fixture, or any h2c gRPC server reachable from the +cluster, works for a manual run. + ## Notes / limitations - This milestone **authenticates** identity (is this a real, running actor?). **Authorizing** diff --git a/demos/egress/main.go b/demos/egress/main.go index 4c0288e12e..09c7b26772 100644 --- a/demos/egress/main.go +++ b/demos/egress/main.go @@ -13,18 +13,30 @@ // limitations under the License. // Command egress is a small HTTP service for demonstrating per-Actor egress -// policy. It accepts a URL, fetches it, and returns the upstream response. +// policy. It accepts a URL, fetches it, and returns the upstream response, and +// on a second endpoint it makes gRPC calls and returns what came back. package main import ( + "context" "encoding/json" + "errors" "fmt" "io" "log/slog" + "net" "net/http" "net/url" "os" + "strconv" "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + grpcstatus "google.golang.org/grpc/status" + + "github.com/agent-substrate/substrate/internal/proto/grpcechopb" ) const ( @@ -44,6 +56,42 @@ type fetchResponse struct { Error string `json:"error,omitempty"` } +// grpcRequest asks for one unary Echo against target, and additionally for a +// server-stream or a bidirectional stream when the matching count is positive. +type grpcRequest struct { + // Target is the gRPC server to dial, as host:port. Cleartext HTTP/2: the + // point of the demo is that the actor speaks plainly and the egress path + // carries it, so there is nothing here to configure TLS with. + Target string `json:"target"` + Message string `json:"message"` + StreamCount int32 `json:"streamCount,omitempty"` + // BidiCount is how many messages to send over a bidirectional stream, + // one at a time, each awaiting its response before the next goes out. + BidiCount int32 `json:"bidiCount,omitempty"` +} + +type grpcResponse struct { + // Message is what the unary Echo returned. + Message string `json:"message"` + // Stream is what EchoStream returned, in the order it arrived. Absent when + // the request did not ask for a stream. + Stream []streamedMessage `json:"stream,omitempty"` + // Bidi is what EchoBidi returned, in the order it arrived. Absent when the + // request did not ask for one. + Bidi []streamedMessage `json:"bidi,omitempty"` + // Code is the gRPC status of the last RPC attempted, as a string. It is the + // one field an HTTP status cannot stand in for: a gRPC status travels in + // trailers, after the response body, so a path that drops trailers or + // downgrades the connection to HTTP/1.1 cannot produce this at all. + Code string `json:"code,omitempty"` + Error string `json:"error,omitempty"` +} + +type streamedMessage struct { + Message string `json:"message"` + Index int32 `json:"index"` +} + func main() { slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil))) @@ -101,9 +149,147 @@ func newHandler(client *http.Client) http.Handler { } writeJSON(w, response.StatusCode, fetchResponse{StatusCode: response.StatusCode, Body: string(body)}) }) + mux.HandleFunc("/grpc", handleGRPC) return mux } +// handleGRPC dials the requested target and echoes back what the RPCs returned. +// It exists so an e2e can assert that gRPC survives the egress path: HTTP/2 +// framing end to end, and a status delivered in trailers. +func handleGRPC(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + writeJSON(w, http.StatusMethodNotAllowed, grpcResponse{Error: "method must be POST"}) + return + } + + var input grpcRequest + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxRequestBody)) + if err := decoder.Decode(&input); err != nil { + writeJSON(w, http.StatusBadRequest, grpcResponse{Error: fmt.Sprintf("invalid JSON payload: %v", err)}) + return + } + if err := validateTarget(input.Target); err != nil { + writeJSON(w, http.StatusBadRequest, grpcResponse{Error: err.Error()}) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), requestTimeout) + defer cancel() + + // Dialed per request and closed with it. The actor this runs inside is + // checkpointed and restored, and an HTTP/2 connection opened before a + // snapshot does not survive one: the peer is long gone by the time the + // actor resumes, and every RPC on it would fail in a way that looks like a + // broken gateway. + conn, err := grpc.NewClient(input.Target, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + writeJSON(w, http.StatusBadRequest, grpcResponse{Error: fmt.Sprintf("dialing %s: %v", input.Target, err)}) + return + } + defer conn.Close() + + client := grpcechopb.NewEchoClient(conn) + echoed, err := client.Echo(ctx, &grpcechopb.EchoRequest{Message: input.Message}) + if err != nil { + writeGRPCFailure(w, "Echo", err) + return + } + + response := grpcResponse{Message: echoed.GetMessage(), Code: codes.OK.String()} + if input.StreamCount > 0 { + response.Stream, err = echoStream(ctx, client, input) + if err != nil { + writeGRPCFailure(w, "EchoStream", err) + return + } + } + if input.BidiCount > 0 { + response.Bidi, err = echoBidi(ctx, client, input) + if err != nil { + writeGRPCFailure(w, "EchoBidi", err) + return + } + } + writeJSON(w, http.StatusOK, response) +} + +// echoStream drains a server-stream into the response's Stream field. +func echoStream(ctx context.Context, client grpcechopb.EchoClient, input grpcRequest) ([]streamedMessage, error) { + stream, err := client.EchoStream(ctx, &grpcechopb.EchoStreamRequest{Message: input.Message, Count: input.StreamCount}) + if err != nil { + return nil, err + } + var out []streamedMessage + for { + received, err := stream.Recv() + if errors.Is(err, io.EOF) { + return out, nil + } + if err != nil { + return nil, err + } + out = append(out, streamedMessage{Message: received.GetMessage(), Index: received.GetIndex()}) + } +} + +// echoBidi runs a bidirectional stream, sending BidiCount messages one at a +// time and waiting for each response before sending the next. That ordering is +// the whole point: sending everything and then reading it back would succeed +// over a path that carries one direction at a time, which is precisely the +// failure this endpoint exists to catch. +func echoBidi(ctx context.Context, client grpcechopb.EchoClient, input grpcRequest) ([]streamedMessage, error) { + stream, err := client.EchoBidi(ctx) + if err != nil { + return nil, err + } + + out := make([]streamedMessage, 0, input.BidiCount) + for i := range input.BidiCount { + // A distinct message per iteration, so a path that replays or holds on + // to a buffered frame shows up as a mismatch rather than as a pass. + message := fmt.Sprintf("%s-%d", input.Message, i) + if err := stream.Send(&grpcechopb.EchoRequest{Message: message}); err != nil { + // grpc-go reports a broken stream from Send as io.EOF and puts the + // real status on Recv, which is the code worth reporting. + if _, recvErr := stream.Recv(); recvErr != nil { + return nil, recvErr + } + return nil, err + } + received, err := stream.Recv() + if err != nil { + return nil, err + } + out = append(out, streamedMessage{Message: received.GetMessage(), Index: received.GetIndex()}) + } + + // Half-close the request direction while the response direction is still + // open, then drain it. A path that reads a one-directional END_STREAM as a + // teardown fails here and nowhere else. + if err := stream.CloseSend(); err != nil { + return nil, err + } + if _, err := stream.Recv(); !errors.Is(err, io.EOF) { + if err == nil { + return nil, fmt.Errorf("server sent more responses than the %d messages requested", input.BidiCount) + } + return nil, err + } + return out, nil +} + +// writeGRPCFailure reports a failed RPC. The HTTP status is 502 so that a +// caller polling through the ingress router keeps retrying -- the origin may +// simply not be up yet -- while the gRPC code travels in the body, where an +// HTTP status cannot flatten it into a generic "bad gateway". +func writeGRPCFailure(w http.ResponseWriter, rpc string, err error) { + writeJSON(w, http.StatusBadGateway, grpcResponse{ + Code: grpcstatus.Code(err).String(), + Error: fmt.Sprintf("%s failed: %v", rpc, err), + }) +} + func validateURL(raw string) error { parsed, err := url.Parse(raw) if err != nil { @@ -118,8 +304,27 @@ func validateURL(raw string) error { return nil } -func writeJSON(w http.ResponseWriter, status int, response fetchResponse) { +// validateTarget checks that raw is the host:port a gRPC dial needs. Unlike a +// URL there is no scheme to reject, so a caller that passes one -- or passes a +// bare hostname and lets the dial default the port -- finds out here rather +// than in a connection error from somewhere along the egress path. +func validateTarget(raw string) error { + host, port, err := net.SplitHostPort(raw) + if err != nil { + return fmt.Errorf("target must be host:port: %w", err) + } + if host == "" { + return fmt.Errorf("target must include a host") + } + number, err := strconv.Atoi(port) + if err != nil || number < 1 || number > 65535 { + return fmt.Errorf("target port must be a number between 1 and 65535, got %q", port) + } + return nil +} + +func writeJSON(w http.ResponseWriter, statusCode int, response any) { w.Header().Set("Content-Type", "application/json") - w.WriteHeader(status) + w.WriteHeader(statusCode) _ = json.NewEncoder(w).Encode(response) } diff --git a/demos/egress/main_test.go b/demos/egress/main_test.go index 6cd203732f..c5736f1f77 100644 --- a/demos/egress/main_test.go +++ b/demos/egress/main_test.go @@ -15,13 +15,21 @@ package main import ( + "context" "encoding/json" "errors" + "fmt" "io" + "net" "net/http" "net/http/httptest" "strings" "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + + "github.com/agent-substrate/substrate/internal/proto/grpcechopb" ) func TestFetch(t *testing.T) { @@ -107,3 +115,206 @@ type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { return f(request) } + +// The gRPC endpoint below is what an e2e reads as evidence that gRPC crossed +// the egress tunnel, so these tests pin its answers against a loopback server, +// where there is no tunnel to blame. + +// echoServer is the in-process stand-in for internal/e2e/fixtures/grpcecho. +type echoServer struct { + grpcechopb.UnimplementedEchoServer +} + +func (echoServer) Echo(_ context.Context, req *grpcechopb.EchoRequest) (*grpcechopb.EchoResponse, error) { + return &grpcechopb.EchoResponse{Message: req.GetMessage()}, nil +} + +func (echoServer) EchoStream(req *grpcechopb.EchoStreamRequest, stream grpc.ServerStreamingServer[grpcechopb.EchoResponse]) error { + for i := range req.GetCount() { + if err := stream.Send(&grpcechopb.EchoResponse{Message: req.GetMessage(), Index: i}); err != nil { + return err + } + } + return nil +} + +// EchoBidi answers each request as it arrives, like the fixture: the handler +// under test sends one message at a time and blocks on its response, so a +// stand-in that drained the request direction first would deadlock. +func (echoServer) EchoBidi(stream grpc.BidiStreamingServer[grpcechopb.EchoRequest, grpcechopb.EchoResponse]) error { + for index := int32(0); ; index++ { + req, err := stream.Recv() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + if err := stream.Send(&grpcechopb.EchoResponse{Message: req.GetMessage(), Index: index}); err != nil { + return err + } + } +} + +// startEchoServer serves Echo on loopback and returns its host:port. +func startEchoServer(t *testing.T) string { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening on loopback: %v", err) + } + server := grpc.NewServer() + grpcechopb.RegisterEchoServer(server, echoServer{}) + go func() { + if err := server.Serve(listener); err != nil { + t.Logf("serving: %v", err) + } + }() + t.Cleanup(server.Stop) + + return listener.Addr().String() +} + +// postGRPC drives the /grpc endpoint and returns the recorder and the decoded +// body, which is what the e2e asserts on. +func postGRPC(t *testing.T, body string) (*httptest.ResponseRecorder, grpcResponse) { + t.Helper() + + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/grpc", strings.NewReader(body)) + newHandler(http.DefaultClient).ServeHTTP(recorder, request) + + var decoded grpcResponse + if err := json.NewDecoder(recorder.Body).Decode(&decoded); err != nil { + t.Fatalf("decoding response (HTTP %d): %v", recorder.Code, err) + } + return recorder, decoded +} + +func TestGRPCUnary(t *testing.T) { + target := startEchoServer(t) + + recorder, got := postGRPC(t, fmt.Sprintf(`{"target":%q,"message":"hello"}`, target)) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body = %+v", recorder.Code, http.StatusOK, got) + } + if got.Message != "hello" { + t.Errorf("message = %q, want %q", got.Message, "hello") + } + if got.Code != codes.OK.String() { + t.Errorf("code = %q, want %q", got.Code, codes.OK.String()) + } + // A request that did not ask for a stream must not report one, so the e2e + // cannot pass its streaming assertion against leftover unary state. + if got.Stream != nil { + t.Errorf("stream = %+v, want none for a request with no streamCount", got.Stream) + } +} + +func TestGRPCStream(t *testing.T) { + target := startEchoServer(t) + const count = 3 + + recorder, got := postGRPC(t, fmt.Sprintf(`{"target":%q,"message":"streamed","streamCount":%d}`, target, count)) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body = %+v", recorder.Code, http.StatusOK, got) + } + if len(got.Stream) != count { + t.Fatalf("stream has %d messages, want %d: %+v", len(got.Stream), count, got.Stream) + } + for i, message := range got.Stream { + if message.Message != "streamed" { + t.Errorf("stream[%d].message = %q, want %q", i, message.Message, "streamed") + } + if int(message.Index) != i { + t.Errorf("stream[%d].index = %d, want %d", i, message.Index, i) + } + } +} + +func TestGRPCBidi(t *testing.T) { + target := startEchoServer(t) + const count = 3 + + recorder, got := postGRPC(t, fmt.Sprintf(`{"target":%q,"message":"duplex","bidiCount":%d}`, target, count)) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body = %+v", recorder.Code, http.StatusOK, got) + } + if len(got.Bidi) != count { + t.Fatalf("bidi has %d messages, want %d: %+v", len(got.Bidi), count, got.Bidi) + } + // The handler numbers its own messages, so the echoed text pins each + // response to the request it answered rather than to any response. + for i, message := range got.Bidi { + want := fmt.Sprintf("duplex-%d", i) + if message.Message != want { + t.Errorf("bidi[%d].message = %q, want %q", i, message.Message, want) + } + if int(message.Index) != i { + t.Errorf("bidi[%d].index = %d, want %d", i, message.Index, i) + } + } + // Asking only for a bidi stream must not report a server-stream, so the + // e2e's two assertions cannot pass on each other's data. + if got.Stream != nil { + t.Errorf("stream = %+v, want none for a request with no streamCount", got.Stream) + } +} + +// A failed RPC must still report its gRPC code. That code is the only part of +// the answer that proves trailers arrived, so collapsing it into the HTTP +// status would erase the thing the e2e is looking for. +func TestGRPCFailureReportsStatusCode(t *testing.T) { + // A port with nothing behind it: the dial fails, and grpc-go reports that + // as Unavailable. + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening on loopback: %v", err) + } + target := listener.Addr().String() + if err := listener.Close(); err != nil { + t.Fatalf("closing loopback listener: %v", err) + } + + recorder, got := postGRPC(t, fmt.Sprintf(`{"target":%q,"message":"hello"}`, target)) + + if recorder.Code != http.StatusBadGateway { + t.Errorf("status = %d, want %d", recorder.Code, http.StatusBadGateway) + } + if got.Code != codes.Unavailable.String() { + t.Errorf("code = %q, want %q; error = %s", got.Code, codes.Unavailable.String(), got.Error) + } +} + +func TestGRPCInvalidRequests(t *testing.T) { + tests := []struct { + name string + method string + body string + status int + }{ + {name: "method", method: http.MethodGet, body: `{}`, status: http.StatusMethodNotAllowed}, + {name: "malformed JSON", method: http.MethodPost, body: `{`, status: http.StatusBadRequest}, + {name: "missing target", method: http.MethodPost, body: `{"message":"hi"}`, status: http.StatusBadRequest}, + {name: "no port", method: http.MethodPost, body: `{"target":"grpcecho"}`, status: http.StatusBadRequest}, + {name: "no host", method: http.MethodPost, body: `{"target":":50051"}`, status: http.StatusBadRequest}, + {name: "non-numeric port", method: http.MethodPost, body: `{"target":"grpcecho:grpc"}`, status: http.StatusBadRequest}, + {name: "URL not host:port", method: http.MethodPost, body: `{"target":"http://grpcecho:50051"}`, status: http.StatusBadRequest}, + } + + handler := newHandler(http.DefaultClient) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(test.method, "/grpc", strings.NewReader(test.body)) + handler.ServeHTTP(recorder, request) + if recorder.Code != test.status { + t.Errorf("status = %d, want %d; body = %s", recorder.Code, test.status, recorder.Body.String()) + } + }) + } +} diff --git a/internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl b/internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl deleted file mode 100644 index a07f096294..0000000000 --- a/internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl +++ /dev/null @@ -1,113 +0,0 @@ -# Copyright 2026 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# The egress probe used by internal/e2e/suites/sdsmint. ${NAMESPACE} is -# substituted by the suite with the randomized namespace it created, so the -# probe is torn down with that namespace and leaves nothing behind. -# -# A plain Pod, not an actor: the suite is testing the gateway's MITM leg, and -# running the probe itself inside an actor would make a snapshot or restore -# failure look like an sdsmint failure. The suite does create one actor, but -# only for its identity -- the certificate mounted below is minted for it, and -# the actor's own workload is never contacted. - -apiVersion: v1 -kind: Pod -metadata: - name: egressprobe - namespace: ${NAMESPACE} - labels: - app: egressprobe -spec: - restartPolicy: Never - containers: - - name: egressprobe - image: ko://github.com/agent-substrate/substrate/internal/e2e/fixtures/egressprobe - args: - - "--listen=:8080" - ports: - - name: http - containerPort: 8080 - # The suite port-forwards to this pod, and port-forward targets the first - # READY pod behind the Service. Without a readiness gate the forward can - # attach before the listener exists and the first handshake fails as a - # connection refused that looks like a gateway problem. - readinessProbe: - httpGet: - path: /healthz - port: 8080 - periodSeconds: 2 - resources: - requests: - cpu: 10m - memory: 32Mi - # runAsUser must be spelled out: ko's distroless static base declares no - # USER, so runAsNonRoot on its own makes kubelet refuse to start the - # container ("image will run as root") rather than pick a uid. 65532 is - # distroless's nonroot uid, and the same one the egress gateway's pod uses. - securityContext: - allowPrivilegeEscalation: false - readOnlyRootFilesystem: true - runAsNonRoot: true - runAsUser: 65532 - runAsGroup: 65532 - capabilities: - drop: ["ALL"] - volumeMounts: - - name: "actor-identity" - mountPath: "/run/actor-identity" - - name: "actor-identity-unknown" - mountPath: "/run/actor-identity-unknown" - - name: "podidentity" - mountPath: "/run/podidentity.podcert.ate.dev" - - name: "servicedns-ca" - mountPath: "/run/servicedns.podcert.ate.dev" - volumes: - - name: "actor-identity" - secret: - secretName: egressprobe-actor-identity - - name: "actor-identity-unknown" - secret: - secretName: egressprobe-unknown-actor - - name: "podidentity" - projected: - sources: - - podCertificate: - signerName: podidentity.podcert.ate.dev/identity - keyType: ECDSAP256 - credentialBundlePath: credential-bundle.pem - - name: "servicedns-ca" - projected: - sources: - - clusterTrustBundle: - signerName: servicedns.podcert.ate.dev/identity - labelSelector: - matchLabels: - podcert.ate.dev/canarying: live - path: trust-bundle.pem - ---- - -apiVersion: v1 -kind: Service -metadata: - name: egressprobe - namespace: ${NAMESPACE} -spec: - selector: - app: egressprobe - ports: - - name: http - port: 8080 - targetPort: 8080 diff --git a/internal/e2e/fixtures/grpcecho/main.go b/internal/e2e/fixtures/grpcecho/main.go new file mode 100644 index 0000000000..5cab985773 --- /dev/null +++ b/internal/e2e/fixtures/grpcecho/main.go @@ -0,0 +1,115 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Command grpcecho is the gRPC origin the egress e2e suites dial through the +// egress gateway. It serves cleartext HTTP/2 -- no TLS anywhere -- because the +// leg under test is the tunnel, not the origin's identity, and because the +// gateway relays a terminated CONNECT as opaque TCP: whatever the actor speaks +// is what arrives here. +// +// It answers the grpc health service as well as Echo, which is what the pod's +// readinessProbe checks. A separate HTTP port for readiness would need an h2c +// handler multiplexed onto this listener, and the whole point of this fixture +// is that nothing between the actor and here parses HTTP. +package main + +import ( + "context" + "errors" + "flag" + "io" + "log" + "net" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/health" + healthpb "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/status" + + "github.com/agent-substrate/substrate/internal/proto/grpcechopb" +) + +var listenAddress = flag.String("listen", ":50051", "Address the gRPC server listens on, cleartext HTTP/2.") + +// maxStreamCount bounds EchoStream. A test asks for a handful of messages; a +// request for millions is a bug in the caller, and answering it would look like +// a hung tunnel rather than the mistake it is. +const maxStreamCount = 1000 + +type echoServer struct { + grpcechopb.UnimplementedEchoServer +} + +func (echoServer) Echo(_ context.Context, req *grpcechopb.EchoRequest) (*grpcechopb.EchoResponse, error) { + return &grpcechopb.EchoResponse{Message: req.GetMessage()}, nil +} + +func (echoServer) EchoStream(req *grpcechopb.EchoStreamRequest, stream grpc.ServerStreamingServer[grpcechopb.EchoResponse]) error { + count := req.GetCount() + if count <= 0 { + return status.Errorf(codes.InvalidArgument, "count must be positive, got %d", count) + } + if count > maxStreamCount { + return status.Errorf(codes.InvalidArgument, "count must be at most %d, got %d", maxStreamCount, count) + } + for i := range count { + if err := stream.Send(&grpcechopb.EchoResponse{Message: req.GetMessage(), Index: i}); err != nil { + return err + } + } + return nil +} + +// EchoBidi answers each request as it arrives rather than draining the request +// direction first. Batching the responses would deadlock a caller that waits +// for each one before sending the next, and would also stop this from testing +// anything a server-stream does not already cover: the point is frames moving +// both ways at once. +func (echoServer) EchoBidi(stream grpc.BidiStreamingServer[grpcechopb.EchoRequest, grpcechopb.EchoResponse]) error { + for index := int32(0); ; index++ { + req, err := stream.Recv() + // io.EOF is the client's half-close: the request direction is done, + // this direction still has to end cleanly with an OK status. + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + if err := stream.Send(&grpcechopb.EchoResponse{Message: req.GetMessage(), Index: index}); err != nil { + return err + } + } +} + +// newServer returns a server with everything registered on it, so the unit +// tests exercise the same registrations the pod serves. +func newServer() *grpc.Server { + server := grpc.NewServer() + grpcechopb.RegisterEchoServer(server, echoServer{}) + healthpb.RegisterHealthServer(server, health.NewServer()) + return server +} + +func main() { + flag.Parse() + + listener, err := net.Listen("tcp", *listenAddress) + if err != nil { + log.Fatalf("grpcecho: listening on %s: %v", *listenAddress, err) + } + log.Printf("grpcecho: serving on %s", listener.Addr()) + log.Fatal(newServer().Serve(listener)) +} diff --git a/internal/e2e/fixtures/grpcecho/main_test.go b/internal/e2e/fixtures/grpcecho/main_test.go new file mode 100644 index 0000000000..22f2c1e009 --- /dev/null +++ b/internal/e2e/fixtures/grpcecho/main_test.go @@ -0,0 +1,194 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + healthpb "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/status" + + "github.com/agent-substrate/substrate/internal/proto/grpcechopb" +) + +// The e2e suite reads this fixture's answers as evidence about the egress +// tunnel, so a wrong answer here would be read as a broken gateway. These tests +// pin the answers against a loopback listener, where no gateway is involved. + +// dialLocal starts the fixture's server on loopback and returns a connection to +// it. A real listener rather than an in-memory pipe, so the registrations and +// the HTTP/2 framing are the ones the pod serves. +func dialLocal(t *testing.T) *grpc.ClientConn { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening on loopback: %v", err) + } + server := newServer() + go func() { + if err := server.Serve(listener); err != nil { + t.Logf("serving: %v", err) + } + }() + t.Cleanup(server.Stop) + + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("dialing %s: %v", listener.Addr(), err) + } + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} + +func testContext(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + return ctx +} + +func TestEcho(t *testing.T) { + client := grpcechopb.NewEchoClient(dialLocal(t)) + + for _, message := range []string{"hello", "", "unicode: é世界", "with spaces and\nnewline"} { + response, err := client.Echo(testContext(t), &grpcechopb.EchoRequest{Message: message}) + if err != nil { + t.Fatalf("Echo(%q): %v", message, err) + } + if response.GetMessage() != message { + t.Errorf("Echo(%q) = %q, want the message back unchanged", message, response.GetMessage()) + } + } +} + +func TestEchoStream(t *testing.T) { + client := grpcechopb.NewEchoClient(dialLocal(t)) + + const ( + message = "streamed" + count = 3 + ) + stream, err := client.EchoStream(testContext(t), &grpcechopb.EchoStreamRequest{Message: message, Count: count}) + if err != nil { + t.Fatalf("EchoStream: %v", err) + } + + var got []*grpcechopb.EchoResponse + for { + response, err := stream.Recv() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatalf("EchoStream Recv after %d responses: %v", len(got), err) + } + got = append(got, response) + } + + if len(got) != count { + t.Fatalf("EchoStream sent %d responses, want %d", len(got), count) + } + // Indexes are what tell a reordered or deduplicated stream from an intact + // one, which is the only thing the e2e assertion can check about ordering. + for i, response := range got { + if response.GetMessage() != message { + t.Errorf("response %d message = %q, want %q", i, response.GetMessage(), message) + } + if int(response.GetIndex()) != i { + t.Errorf("response %d index = %d, want %d", i, response.GetIndex(), i) + } + } +} + +// The bidi handler must answer each request as it arrives. Sending everything +// and only then reading would pass against a server that drains first, which is +// exactly the implementation this test exists to rule out -- so this sends one +// message at a time and blocks on its response before sending the next. +func TestEchoBidi(t *testing.T) { + client := grpcechopb.NewEchoClient(dialLocal(t)) + + stream, err := client.EchoBidi(testContext(t)) + if err != nil { + t.Fatalf("EchoBidi: %v", err) + } + + const count = 3 + for i := range count { + message := fmt.Sprintf("message-%d", i) + if err := stream.Send(&grpcechopb.EchoRequest{Message: message}); err != nil { + t.Fatalf("EchoBidi Send %d: %v", i, err) + } + response, err := stream.Recv() + if err != nil { + t.Fatalf("EchoBidi Recv %d: %v", i, err) + } + if response.GetMessage() != message { + t.Errorf("response %d message = %q, want %q", i, response.GetMessage(), message) + } + if int(response.GetIndex()) != i { + t.Errorf("response %d index = %d, want %d", i, response.GetIndex(), i) + } + } + + // Half-close the request direction and drain the response direction. The + // server must end with OK here: a handler that treated the half-close as an + // error would still have echoed everything above. + if err := stream.CloseSend(); err != nil { + t.Fatalf("EchoBidi CloseSend: %v", err) + } + if _, err := stream.Recv(); !errors.Is(err, io.EOF) { + t.Errorf("EchoBidi Recv after CloseSend = %v, want io.EOF", err) + } +} + +// A non-positive count must fail loudly. Returning an empty stream instead +// would make a caller that forgot to set count look like a working one. +func TestEchoStreamRejectsBadCount(t *testing.T) { + client := grpcechopb.NewEchoClient(dialLocal(t)) + + for _, count := range []int32{0, -1, maxStreamCount + 1} { + stream, err := client.EchoStream(testContext(t), &grpcechopb.EchoStreamRequest{Message: "x", Count: count}) + if err == nil { + _, err = stream.Recv() + } + if got := status.Code(err); got != codes.InvalidArgument { + t.Errorf("EchoStream(count=%d) status = %s (%v), want InvalidArgument", count, got, err) + } + } +} + +// The pod's readinessProbe is a grpc probe, which checks the health service's +// empty service name. Nothing else in the fixture would notice if that +// registration disappeared, and the pod would simply never become ready. +func TestHealthServiceIsServing(t *testing.T) { + response, err := healthpb.NewHealthClient(dialLocal(t)).Check(testContext(t), &healthpb.HealthCheckRequest{}) + if err != nil { + t.Fatalf("health Check: %v", err) + } + if response.GetStatus() != healthpb.HealthCheckResponse_SERVING { + t.Errorf("health Check status = %s, want SERVING", response.GetStatus()) + } +} diff --git a/internal/e2e/fixtures/serverpod.yaml.tmpl b/internal/e2e/fixtures/serverpod.yaml.tmpl new file mode 100644 index 0000000000..733a8976fa --- /dev/null +++ b/internal/e2e/fixtures/serverpod.yaml.tmpl @@ -0,0 +1,86 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# One manifest for every plain server a suite stands up beside the code under +# test -- the thing an Actor's egress lands on, or the thing that dials the +# gateway itself. e2e.DeployServerPod renders this; see internal/e2e/serverpod.go +# for what each placeholder means. +# +# The differences between those servers are the placeholders below: a name, an +# image, a port, a readiness probe, and whatever credentials the server has to +# mount. Everything else was identical in every copy of this manifest that +# existed before it was shared, so a per-server file only offered somewhere for +# the copies to drift apart. +# +# A plain Pod, not an actor. These servers are scaffolding around the code under +# test, and running one inside an actor would make its snapshot or restore +# failures look like failures of the thing being tested. + +apiVersion: v1 +kind: Pod +metadata: + name: ${NAME} + namespace: ${NAMESPACE} + labels: + app: ${NAME} +spec: + restartPolicy: Never + containers: + - name: ${NAME} + image: ${IMAGE} + args: + - "--listen=:${PORT}" + ports: + - name: serve + containerPort: ${PORT} + # The suite starts dialing as soon as it has an address, so without a + # readiness gate the first connection can land before the listener exists + # and come back refused -- which reads as a failure of whatever sits between + # the caller and here, rather than as a pod that had not started yet. + readinessProbe: +${READINESS_PROBE} + periodSeconds: 2 + resources: + requests: + cpu: 10m + memory: 32Mi + # runAsUser must be spelled out: ko's distroless static base declares no + # USER, so runAsNonRoot on its own makes kubelet refuse to start the + # container ("image will run as root") rather than pick a uid. 65532 is + # distroless's nonroot uid, and the same one the egress gateway's pod uses. + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 65532 + runAsGroup: 65532 + capabilities: + drop: ["ALL"] +${VOLUME_MOUNTS} +${VOLUMES} + +--- + +apiVersion: v1 +kind: Service +metadata: + name: ${NAME} + namespace: ${NAMESPACE} +spec: + selector: + app: ${NAME} + ports: + - name: serve + port: ${PORT} + targetPort: ${PORT} diff --git a/internal/e2e/manifest.go b/internal/e2e/manifest.go new file mode 100644 index 0000000000..85d306eb26 --- /dev/null +++ b/internal/e2e/manifest.go @@ -0,0 +1,117 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "sigs.k8s.io/yaml" +) + +// renderManifest substitutes placeholders into the manifest template at relPath +// (repo-relative), writes the result into the test's temp dir and returns that +// path. Both an apply and a later delete can then consume the same file, with +// no shell involved. +// +// Templates carry two kinds of ${...} placeholder: +// +// - inline, substituted wherever they appear (an empty value just disappears); +// - block, which must be the entire content of their line. They expand to a +// YAML fragment that brings its own indentation, and an empty value takes +// the whole line with it — the same trick hack/install-demo-counter.sh +// plays with `sed /.../d`. Requiring the placeholder to be the whole line +// is what lets a comment mention one without being deleted. +func renderManifest(t *testing.T, relPath string, inline, blocks map[string]string) string { + t.Helper() + root, err := FindRepoRoot() + if err != nil { + t.Fatalf("FindRepoRoot: %v", err) + } + raw, err := os.ReadFile(filepath.Join(root, relPath)) + if err != nil { + t.Fatalf("reading manifest template %s: %v", relPath, err) + } + + var out []string + for line := range strings.SplitSeq(string(raw), "\n") { + if value, isBlock := blocks[strings.TrimSpace(line)]; isBlock { + if value != "" { + out = append(out, value) + } + continue + } + for placeholder, value := range inline { + line = strings.ReplaceAll(line, placeholder, value) + } + out = append(out, line) + } + + rendered := strings.TrimSuffix(filepath.Join(t.TempDir(), filepath.Base(relPath)), ".tmpl") + if err := os.WriteFile(rendered, []byte(strings.Join(out, "\n")), 0o644); err != nil { + t.Fatalf("writing rendered manifest %s: %v", rendered, err) + } + return rendered +} + +// yamlListBlock renders items as `key:` followed by their YAML, every line +// indented by indent spaces, for substitution into a block placeholder. An +// empty list renders to "", which takes the placeholder's whole line — key +// included, since a bare `volumes:` with nothing under it is not what the +// caller meant. +// +// Marshaling the real API types rather than asking callers for YAML text is +// what keeps a fragment honest: a misspelled field is a compile error here, +// where in a template it would apply cleanly and do nothing. +func yamlListBlock[T any](t *testing.T, key string, items []T, indent int) string { + t.Helper() + if len(items) == 0 { + return "" + } + raw, err := yaml.Marshal(items) + if err != nil { + t.Fatalf("marshaling %s for the manifest: %v", key, err) + } + pad := strings.Repeat(" ", indent) + out := []string{pad + key + ":"} + for line := range strings.SplitSeq(strings.TrimRight(string(raw), "\n"), "\n") { + out = append(out, pad+line) + } + return strings.Join(out, "\n") +} + +// koApply builds and pushes the ko:// images named in manifest and applies it. +// +// Through the repo's pinned ko (hack/run-tool.sh), because CI does not install +// ko on PATH and every other deploy in this repo goes through that wrapper. The +// trailing `-- --context=...` mirrors run_ko in hack/install-ate.sh: ko's apply +// subcommand forwards args after `--` to kubectl. KO_CONFIG_PATH is required +// because ko resolves .ko.yaml from its working directory, which is the test's +// package dir rather than the repo root; without it the build silently loses +// defaultPlatforms and produces images that cannot run on the cluster's nodes. +func koApply(t *testing.T, manifest string) { + t.Helper() + root, err := FindRepoRoot() + if err != nil { + t.Fatalf("FindRepoRoot: %v", err) + } + applyArgs := []string{"ko", "apply", "-f", manifest} + if KubeContext != "" { + applyArgs = append(applyArgs, "--", "--context="+KubeContext) + } + RunCmdWithEnv(t, []string{"KO_CONFIG_PATH=" + root}, filepath.Join(root, "hack/run-tool.sh"), applyArgs...) +} diff --git a/internal/e2e/pod.go b/internal/e2e/pod.go new file mode 100644 index 0000000000..baf9bff29f --- /dev/null +++ b/internal/e2e/pod.go @@ -0,0 +1,74 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/agent-substrate/substrate/internal/portforward" +) + +// WaitForPodReady blocks until the pod passes its readiness probe, and fails the +// test with the pod's last observed state if it does not within timeout. +// +// A suite that skipped this and dialed straight away would race the readiness +// probe, and a fixture that is still pulling its image or crash-looping would +// then be reported as whatever the code under test does with a refused +// connection. Reporting the container's own waiting/terminated reason instead is +// the whole point of the poll: it is the difference between "ImagePullBackOff" +// and an unexplained timeout. +func WaitForPodReady(t *testing.T, ctx context.Context, namespace, name string, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + var lastState string + for time.Now().Before(deadline) { + pod, err := GetClients().K8s.CoreV1().Pods(namespace).Get(ctx, name, metav1.GetOptions{}) + switch { + case err != nil: + lastState = err.Error() + case portforward.IsPodReady(pod): + t.Logf("pod %s/%s is ready", namespace, name) + return + default: + lastState = DescribePodState(pod) + } + time.Sleep(2 * time.Second) + } + t.Fatalf("timed out after %v waiting for pod %s/%s to become ready: %s", timeout, namespace, name, lastState) +} + +// DescribePodState summarizes why a pod is not ready yet, one clause per +// container, for a timeout message. +func DescribePodState(pod *corev1.Pod) string { + parts := []string{"phase=" + string(pod.Status.Phase)} + for _, cs := range pod.Status.ContainerStatuses { + switch { + case cs.State.Waiting != nil: + parts = append(parts, fmt.Sprintf("%s waiting: %s: %s", cs.Name, cs.State.Waiting.Reason, cs.State.Waiting.Message)) + case cs.State.Terminated != nil: + parts = append(parts, fmt.Sprintf("%s terminated: %s: %s", cs.Name, cs.State.Terminated.Reason, cs.State.Terminated.Message)) + default: + parts = append(parts, fmt.Sprintf("%s running, ready=%t", cs.Name, cs.Ready)) + } + } + return strings.Join(parts, "; ") +} diff --git a/internal/e2e/probe.go b/internal/e2e/probe.go index 3ee570f16c..e5cafd9acb 100644 --- a/internal/e2e/probe.go +++ b/internal/e2e/probe.go @@ -14,11 +14,7 @@ package e2e -import ( - "context" - "path/filepath" - "testing" -) +import "testing" // ProbeName is the name of the probe fixture's WorkerPool and ActorTemplate, // inside the namespace DeployProbe returns. @@ -32,33 +28,14 @@ const ProbeName = "probe" func DeployProbe(t *testing.T, bucket, name string) string { t.Helper() - root, err := FindRepoRoot() - if err != nil { - t.Fatalf("FindRepoRoot: %v", err) - } - - // The probe template projects the egress trust bundle, and every actor — - // including the fixture's golden boot — fails closed while the bundle is - // missing, so make sure it exists whatever suite is deploying. - EnsureEgressTrustBundle(t, context.Background(), GetClients()) - // One manifest, rendered for the sandbox class under test, so both apply // and delete consume the same file without any shell involved. manifest := RenderFixtureManifest(t, "internal/e2e/fixtures/probe/probe.yaml.tmpl", bucket, name) + koApply(t, manifest) - // Build/push the probe image and apply through the repo's pinned ko; CI - // does not install ko on PATH. The trailing `-- --context=...` mirrors - // run_ko in hack/install-ate.sh: ko's apply subcommand forwards args after - // `--` to kubectl. KO_CONFIG_PATH is required because ko resolves .ko.yaml - // from its working directory, which is the test's package dir rather than - // the repo root; without it the build silently loses defaultPlatforms and - // produces images that cannot run on the cluster's nodes. - applyArgs := []string{"ko", "apply", "-f", manifest} - if KubeContext != "" { - applyArgs = append(applyArgs, "--", "--context="+KubeContext) - } - RunCmdWithEnv(t, []string{"KO_CONFIG_PATH=" + root}, filepath.Join(root, "hack/run-tool.sh"), applyArgs...) - + // Unlike the fixtures that live in a namespace CreateNamespace tears down, + // this one installs into a fixed namespace it shares with nothing, so it has + // to clean up after itself. t.Cleanup(func() { // Deletion needs no image build, so go straight to kubectl. `ko delete` // rejects this arg shape ("you may not specify resource arguments as diff --git a/internal/e2e/sandbox.go b/internal/e2e/sandbox.go index 29778f10de..bc9dc24cfa 100644 --- a/internal/e2e/sandbox.go +++ b/internal/e2e/sandbox.go @@ -16,8 +16,6 @@ package e2e import ( "os" - "path/filepath" - "strings" "testing" "time" ) @@ -136,45 +134,12 @@ func TemplateReadyTimeout(t *testing.T) time.Duration { // out from under another. // // One template serves both sandbox classes so the two variants of a fixture -// cannot drift apart. Templates carry two kinds of ${...} placeholder: -// -// - inline, substituted wherever they appear (an empty value just disappears); -// - block, which must be the entire content of their line. They expand to a -// YAML fragment that brings its own indentation, and an empty value takes -// the whole line with it — the same trick hack/install-demo-counter.sh -// plays with `sed /.../d`. Requiring the placeholder to be the whole line -// is what lets a comment mention one without being deleted. +// cannot drift apart. See renderManifest for the placeholder kinds a template +// can carry. func RenderFixtureManifest(t *testing.T, relPath, bucket, name string) string { t.Helper() - root, err := FindRepoRoot() - if err != nil { - t.Fatalf("FindRepoRoot: %v", err) - } - raw, err := os.ReadFile(filepath.Join(root, relPath)) - if err != nil { - t.Fatalf("reading fixture manifest %s: %v", relPath, err) - } - inline, blocks := fixtureSubstitutions(bucket, name) - var out []string - for line := range strings.SplitSeq(string(raw), "\n") { - if value, isBlock := blocks[strings.TrimSpace(line)]; isBlock { - if value != "" { - out = append(out, value) - } - continue - } - for placeholder, value := range inline { - line = strings.ReplaceAll(line, placeholder, value) - } - out = append(out, line) - } - - rendered := strings.TrimSuffix(filepath.Join(t.TempDir(), filepath.Base(relPath)), ".tmpl") - if err := os.WriteFile(rendered, []byte(strings.Join(out, "\n")), 0o644); err != nil { - t.Fatalf("writing rendered fixture manifest %s: %v", rendered, err) - } - return rendered + return renderManifest(t, relPath, inline, blocks) } // fixtureSubstitutions is the placeholder set the internal/e2e/fixtures diff --git a/internal/e2e/serverpod.go b/internal/e2e/serverpod.go new file mode 100644 index 0000000000..e16b2c7df5 --- /dev/null +++ b/internal/e2e/serverpod.go @@ -0,0 +1,153 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e + +import ( + "context" + "fmt" + "net" + "strconv" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// serverPodTemplate is the one manifest every ServerPod is rendered from. +const serverPodTemplate = "internal/e2e/fixtures/serverpod.yaml.tmpl" + +// serverPodReadyTimeout covers a cold image pull on a kind node, which is what +// dominates: the servers themselves listen immediately. +const serverPodReadyTimeout = 3 * time.Minute + +// ServerPod describes a plain server to stand up beside the code under test: +// the origin an Actor's egress lands on, or the probe that dials a gateway. +// Everything those have in common — the pod shape, the Service, the security +// context — lives in the shared template, so a suite only names what differs. +type ServerPod struct { + // Name names the Pod, its container and the Service alike, and is what + // appears in kubectl output when a test fails. + Name string + // ImportPath is the server binary's package, as a ko:// reference. The + // template's contract is that the binary takes --listen=:, which is + // how one manifest serves fixtures that share nothing else. + ImportPath string + // Port is what the binary listens on and what the Service publishes, + // unchanged, so an address a suite grafts into an assertion — a CONNECT + // authority in a gateway's access log, say — is this number. + Port int + // Namespace deploys into an existing namespace instead of a fresh one, for + // a suite that has to populate that namespace first: credentials the pod + // mounts have to exist before it is scheduled, and DeployServerPod cannot + // hand back a namespace it has not created yet. + Namespace string + // GRPCProbe asks kubelet to probe with the gRPC health protocol instead of + // an HTTP GET. A gRPC server answers an HTTP request with a protocol error, + // so a server speaking grpc must set this and register the health service. + GRPCProbe bool + // HealthPath is the HTTP readiness path, defaulting to /healthz. Ignored + // when GRPCProbe is set. + HealthPath string + // Volumes and VolumeMounts carry whatever credentials the server needs. + // Typed, rather than more YAML in the template, so the Secret names here + // sit beside the code that creates them instead of drifting from it. + Volumes []corev1.Volume + VolumeMounts []corev1.VolumeMount +} + +// Server is a deployed ServerPod, as the address a caller dials it at. +type Server struct { + // Namespace is the namespace the server was deployed into, for a suite that + // wants to port-forward to it or read its logs on failure. + Namespace string + // ClusterIP is the Service's address. Deliberately not its DNS name: an IP + // keeps a caller inside a sandbox off that sandbox's resolver, and makes + // the authority in a gateway's access log exactly what the test deployed. + ClusterIP string + Port int +} + +// Address is the host:port to dial the server at. +func (s Server) Address() string { + return net.JoinHostPort(s.ClusterIP, strconv.Itoa(s.Port)) +} + +// DeployServerPod builds spec's image, applies the shared server manifest, waits +// for readiness and returns the address to dial. +// +// It registers no cleanup: everything the manifest creates is namespaced, so it +// goes with the namespace CreateNamespace made — and, on failure, is retained +// with it for `kubectl logs`. +func DeployServerPod(t *testing.T, ctx context.Context, spec ServerPod) Server { + t.Helper() + if _, err := CheckEnv("KO_DOCKER_REPO"); err != nil { + t.Fatalf("CheckEnv failed: %v", err) + } + namespace := spec.Namespace + if namespace == "" { + namespace = CreateNamespace(t).Name + } + + koApply(t, renderServerPod(t, spec, namespace)) + WaitForPodReady(t, ctx, namespace, spec.Name, serverPodReadyTimeout) + + service, err := GetClients().K8s.CoreV1().Services(namespace).Get(ctx, spec.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("getting service %s/%s: %v", namespace, spec.Name, err) + } + if service.Spec.ClusterIP == "" || service.Spec.ClusterIP == corev1.ClusterIPNone { + t.Fatalf("service %s/%s has no ClusterIP to dial: %q", namespace, spec.Name, service.Spec.ClusterIP) + } + + server := Server{Namespace: namespace, ClusterIP: service.Spec.ClusterIP, Port: spec.Port} + t.Logf("server %s is serving at %s (namespace %s)", spec.Name, server.Address(), namespace) + return server +} + +// renderServerPod writes spec's manifest into the test's temp dir and returns +// the path. Split out of DeployServerPod so the rendering has a unit test that +// does not need a cluster. +func renderServerPod(t *testing.T, spec ServerPod, namespace string) string { + t.Helper() + port := strconv.Itoa(spec.Port) + inline := map[string]string{ + "${NAME}": spec.Name, + "${NAMESPACE}": namespace, + "${IMAGE}": "ko://" + spec.ImportPath, + "${PORT}": port, + } + blocks := map[string]string{ + "${READINESS_PROBE}": serverReadinessProbe(spec, port), + // Indented to their parents: volumeMounts is a container field, volumes + // a pod one. An empty list takes its whole line, key included. + "${VOLUME_MOUNTS}": yamlListBlock(t, "volumeMounts", spec.VolumeMounts, 4), + "${VOLUMES}": yamlListBlock(t, "volumes", spec.Volumes, 2), + } + return renderManifest(t, serverPodTemplate, inline, blocks) +} + +// serverReadinessProbe renders the probe fragment for spec, indented to sit +// under the template's `readinessProbe:` key. +func serverReadinessProbe(spec ServerPod, port string) string { + if spec.GRPCProbe { + return " grpc:\n port: " + port + } + path := spec.HealthPath + if path == "" { + path = "/healthz" + } + return fmt.Sprintf(" httpGet:\n path: %s\n port: %s", path, port) +} diff --git a/internal/e2e/serverpod_test.go b/internal/e2e/serverpod_test.go new file mode 100644 index 0000000000..1b63a87b07 --- /dev/null +++ b/internal/e2e/serverpod_test.go @@ -0,0 +1,222 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e + +import ( + "os" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + "sigs.k8s.io/yaml" +) + +// renderServerPodDocs renders spec and decodes the Pod and Service out of it. +// +// Strict decoding against the real API types is the point: the probe and the +// volumes are injected as pre-indented text, so a fragment at the wrong depth +// yields YAML that still parses but hangs its fields off the wrong parent — +// which strict mode reports as an unknown field, instead of applying a pod that +// silently has no readiness gate or no credentials. +func renderServerPodDocs(t *testing.T, spec ServerPod) (*corev1.Pod, *corev1.Service) { + t.Helper() + raw, err := os.ReadFile(renderServerPod(t, spec, "test-namespace")) + if err != nil { + t.Fatalf("reading the rendered server manifest: %v", err) + } + if strings.Contains(string(raw), "${") { + t.Errorf("rendered server manifest still carries a placeholder:\n%s", raw) + } + + pod, service := &corev1.Pod{}, &corev1.Service{} + for doc := range strings.SplitSeq(string(raw), "\n---\n") { + if strings.TrimSpace(doc) == "" { + continue + } + var meta struct { + Kind string `json:"kind"` + } + if err := yaml.Unmarshal([]byte(doc), &meta); err != nil { + t.Fatalf("rendered server manifest is not valid YAML: %v\n%s", err, doc) + } + var into any + switch meta.Kind { + case "Pod": + into = pod + case "Service": + into = service + default: + continue + } + if err := yaml.UnmarshalStrict([]byte(doc), into); err != nil { + t.Fatalf("rendered server %s does not match the API type: %v\n%s", meta.Kind, err, doc) + } + } + if pod.Name == "" || service.Name == "" { + t.Fatalf("rendered server manifest is missing a Pod or a Service:\n%s", raw) + } + return pod, service +} + +// TestRenderServerPod_GRPCProbe covers the shape the networking suite deploys: +// a bare origin, no credentials. +func TestRenderServerPod_GRPCProbe(t *testing.T) { + pod, service := renderServerPodDocs(t, ServerPod{ + Name: "grpcecho", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/grpcecho", + Port: 50051, + GRPCProbe: true, + }) + + container := pod.Spec.Containers[0] + if got, want := container.Image, "ko://github.com/agent-substrate/substrate/internal/e2e/fixtures/grpcecho"; got != want { + t.Errorf("container image = %q, want %q", got, want) + } + // The port has to reach the binary, the container port and the Service + // alike: the gateway's access log records whatever the caller dialed, and + // the networking suite greps for exactly this number. + if got, want := container.Args, []string{"--listen=:50051"}; len(got) != 1 || got[0] != want[0] { + t.Errorf("container args = %v, want %v", got, want) + } + if got := container.Ports[0].ContainerPort; got != 50051 { + t.Errorf("containerPort = %d, want 50051", got) + } + if got := service.Spec.Ports[0].Port; got != 50051 { + t.Errorf("service port = %d, want 50051", got) + } + if got := service.Spec.Selector["app"]; got != pod.Labels["app"] { + t.Errorf("service selects app=%q but the pod is labelled app=%q", got, pod.Labels["app"]) + } + + probe := container.ReadinessProbe + if probe == nil || probe.GRPC == nil { + t.Fatalf("readinessProbe = %+v, want a gRPC probe", probe) + } + if probe.GRPC.Port != 50051 { + t.Errorf("gRPC probe port = %d, want 50051", probe.GRPC.Port) + } + if probe.HTTPGet != nil { + t.Errorf("readinessProbe also carries an httpGet: %+v", probe.HTTPGet) + } + // A distroless base declares no USER, so runAsNonRoot alone makes kubelet + // refuse to start the container rather than pick a uid. + if sc := container.SecurityContext; sc == nil || sc.RunAsUser == nil || *sc.RunAsUser != 65532 { + t.Errorf("container securityContext = %+v, want an explicit runAsUser 65532", sc) + } + + // An empty list must take its whole line, `volumes:` key included: a key + // with nothing under it decodes as null, which is not what a caller that + // asked for no volumes meant. + if len(pod.Spec.Volumes) != 0 || len(container.VolumeMounts) != 0 { + t.Errorf("a server that asked for no credentials got volumes %+v / mounts %+v", + pod.Spec.Volumes, container.VolumeMounts) + } +} + +// TestRenderServerPod_HTTPProbe covers the other probe kind, and the default +// health path a server gets when it does not name one. +func TestRenderServerPod_HTTPProbe(t *testing.T) { + pod, _ := renderServerPodDocs(t, ServerPod{ + Name: "httporigin", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/egressprobe", + Port: 8080, + }) + + probe := pod.Spec.Containers[0].ReadinessProbe + if probe == nil || probe.HTTPGet == nil { + t.Fatalf("readinessProbe = %+v, want an httpGet probe", probe) + } + if got, want := probe.HTTPGet.Path, "/healthz"; got != want { + t.Errorf("probe path = %q, want the default %q", got, want) + } + if got := probe.HTTPGet.Port.IntValue(); got != 8080 { + t.Errorf("probe port = %d, want 8080", got) + } + if probe.GRPC != nil { + t.Errorf("readinessProbe also carries a gRPC probe: %+v", probe.GRPC) + } +} + +// TestRenderServerPod_Volumes covers the credential-carrying shape the sdsmint +// suite deploys, with both volume kinds it needs: a plain Secret and a +// projection. A projection is the interesting one — it nests three levels, so +// it is what an off-by-two in the block indentation shows up in. +func TestRenderServerPod_Volumes(t *testing.T) { + pod, _ := renderServerPodDocs(t, ServerPod{ + Name: "egressprobe", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/egressprobe", + Port: 8080, + VolumeMounts: []corev1.VolumeMount{ + {Name: "actor-identity", MountPath: "/run/actor-identity"}, + {Name: "podidentity", MountPath: "/run/podidentity.podcert.ate.dev"}, + }, + Volumes: []corev1.Volume{{ + Name: "actor-identity", + VolumeSource: corev1.VolumeSource{Secret: &corev1.SecretVolumeSource{SecretName: "probe-actor-identity"}}, + }, { + Name: "podidentity", + VolumeSource: corev1.VolumeSource{Projected: &corev1.ProjectedVolumeSource{ + Sources: []corev1.VolumeProjection{{ + ClusterTrustBundle: &corev1.ClusterTrustBundleProjection{ + SignerName: ptr.To("servicedns.podcert.ate.dev/identity"), + LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"podcert.ate.dev/canarying": "live"}}, + Path: "trust-bundle.pem", + }, + }}, + }}, + }}, + }) + + mounts := pod.Spec.Containers[0].VolumeMounts + if len(mounts) != 2 { + t.Fatalf("rendered %d volumeMounts, want 2: %+v", len(mounts), mounts) + } + if got, want := mounts[0].MountPath, "/run/actor-identity"; got != want { + t.Errorf("first mountPath = %q, want %q", got, want) + } + + if len(pod.Spec.Volumes) != 2 { + t.Fatalf("rendered %d volumes, want 2: %+v", len(pod.Spec.Volumes), pod.Spec.Volumes) + } + secret := pod.Spec.Volumes[0].Secret + if secret == nil || secret.SecretName != "probe-actor-identity" { + t.Errorf("first volume = %+v, want the actor-identity Secret", pod.Spec.Volumes[0]) + } + projected := pod.Spec.Volumes[1].Projected + if projected == nil || len(projected.Sources) != 1 || projected.Sources[0].ClusterTrustBundle == nil { + t.Fatalf("second volume = %+v, want a clusterTrustBundle projection", pod.Spec.Volumes[1]) + } + bundle := projected.Sources[0].ClusterTrustBundle + if got, want := ptr.Deref(bundle.SignerName, ""), "servicedns.podcert.ate.dev/identity"; got != want { + t.Errorf("projected signerName = %q, want %q", got, want) + } + if got := bundle.LabelSelector.MatchLabels["podcert.ate.dev/canarying"]; got != "live" { + t.Errorf("projected label selector = %+v, want the live canary label", bundle.LabelSelector) + } + + // Every mount has to name a volume that exists: a typo here yields a pod + // kubelet refuses to start, long after the manifest applied cleanly. + volumes := map[string]bool{} + for _, v := range pod.Spec.Volumes { + volumes[v.Name] = true + } + for _, m := range mounts { + if !volumes[m.Name] { + t.Errorf("volumeMount %q names no volume; the pod has %v", m.Name, volumes) + } + } +} diff --git a/internal/e2e/suites/networking/grpcegress_test.go b/internal/e2e/suites/networking/grpcegress_test.go new file mode 100644 index 0000000000..cd81193661 --- /dev/null +++ b/internal/e2e/suites/networking/grpcegress_test.go @@ -0,0 +1,157 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package networking + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/agent-substrate/substrate/internal/e2e" + "github.com/agent-substrate/substrate/internal/resources" +) + +// grpcEcho is the origin this test deploys: a cleartext-HTTP/2 gRPC server, in +// its own namespace, so nothing here depends on the internet. +var grpcEcho = e2e.ServerPod{ + Name: "grpcecho", + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/grpcecho", + Port: 50051, + // A gRPC server answers an HTTP GET with a protocol error, so readiness has + // to go through the health service the fixture registers. + GRPCProbe: true, +} + +// grpcEchoResponse mirrors the egress demo Actor's /grpc answer. Kept as a +// local copy rather than imported: the demo is a separate module-internal +// command, and what this suite is really pinning is the wire shape between the +// two, which a shared struct would hide. +type grpcEchoResponse struct { + Message string `json:"message"` + Stream []grpcEchoStreamedMsg `json:"stream"` + Bidi []grpcEchoStreamedMsg `json:"bidi"` + Code string `json:"code"` + Error string `json:"error"` +} + +type grpcEchoStreamedMsg struct { + Message string `json:"message"` + Index int32 `json:"index"` +} + +// TestActorEgressGRPC covers the egress path with gRPC, which fails in ways the +// HTTP tests cannot see. atenet-egress terminates the Actor's CONNECT and +// relays opaque TCP, so HTTP/2 framing has to survive end to end and the gRPC +// status has to arrive in trailers, after the response body. An egress path +// that parsed the traffic as HTTP/1.1, or dropped trailers, would still pass +// TestActorEgress and fail here. +// +// All three streaming shapes in one request, because each one fails +// differently: unary is a status in trailers, a server-stream is many frames +// over a held-open connection, and a bidirectional stream has both directions +// carrying frames at once and then half-closes one of them. +// +// The origin is an in-cluster cleartext-HTTP/2 server, deployed per test into +// its own namespace, so nothing in this test depends on the internet. +func TestActorEgressGRPC(t *testing.T) { + ctx := context.Background() + target := e2e.DeployServerPod(t, ctx, grpcEcho).Address() + + actorName, _ := createAndResumeActor(t, ctx, "egress-grpc", e2e.EgressFixture()) + router := mustRouterClient(t, ctx) + defer router.Close() + + // Bound the access-log scan below to lines this test could have produced. + // The slack absorbs clock skew between here and the gateway's node. + since := metav1.NewTime(time.Now().Add(-1 * time.Minute)) + + const ( + message = "hello over grpc" + streamCount = 3 + bidiCount = 3 + ) + payload, err := json.Marshal(map[string]any{ + "target": target, + "message": message, + "streamCount": streamCount, + "bidiCount": bidiCount, + }) + if err != nil { + t.Fatalf("marshaling the gRPC request for %s: %v", target, err) + } + + actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName} + status, body := postThroughEgressActor(t, ctx, router, actorRef, "/grpc", payload) + if status != http.StatusOK { + t.Fatalf("Actor gRPC egress to %s returned HTTP %d, want 200; body: %s", target, status, body) + } + + var got grpcEchoResponse + if err := json.Unmarshal(body, &got); err != nil { + t.Fatalf("decoding the Actor's gRPC response: %v; body: %s", err, body) + } + if got.Message != message { + t.Errorf("unary Echo returned %q, want %q", got.Message, message) + } + // The status is the trailer assertion: a gRPC status is sent after the + // response body, so a path that dropped trailers could deliver the message + // above and still not produce this. + if got.Code != "OK" { + t.Errorf("gRPC status = %q, want OK; error: %s", got.Code, got.Error) + } + if len(got.Stream) != streamCount { + t.Fatalf("EchoStream returned %d responses, want %d: %+v", len(got.Stream), streamCount, got.Stream) + } + for i, response := range got.Stream { + if response.Message != message { + t.Errorf("stream response %d message = %q, want %q", i, response.Message, message) + } + if int(response.Index) != i { + t.Errorf("stream response %d index = %d, want %d", i, response.Index, i) + } + } + + // The bidi leg is the one that needs frames moving in both directions at + // once: the Actor sends each message only after reading the response to the + // previous one, so a path that carried one direction at a time would not + // return a short answer here, it would hang until the Actor's own 15s + // timeout and come back as a 502. + if len(got.Bidi) != bidiCount { + t.Fatalf("EchoBidi returned %d responses, want %d: %+v", len(got.Bidi), bidiCount, got.Bidi) + } + for i, response := range got.Bidi { + // The Actor numbers each message it sends, so this pins every response + // to the request it answered. + want := fmt.Sprintf("%s-%d", message, i) + if response.Message != want { + t.Errorf("bidi response %d message = %q, want %q", i, response.Message, want) + } + if int(response.Index) != i { + t.Errorf("bidi response %d index = %d, want %d", i, response.Index, i) + } + } + t.Logf("Actor gRPC egress to %s succeeded; body: %s", target, body) + + // Everything above would also pass if the Actor's traffic had been + // masqueraded straight out instead of tunneled. This is what says it went + // through the gateway, on this Actor's own certificate. + assertEgressGatewayConnect(t, ctx, since, actorName, strconv.Itoa(grpcEcho.Port)) +} diff --git a/internal/e2e/suites/networking/networking_test.go b/internal/e2e/suites/networking/networking_test.go index 8984fda3b8..13240b0a49 100644 --- a/internal/e2e/suites/networking/networking_test.go +++ b/internal/e2e/suites/networking/networking_test.go @@ -97,22 +97,31 @@ func TestActorEgressHTTPS(t *testing.T) { } // fetchThroughEgressActor asks the egress demo Actor to fetch url and returns -// the status and body it echoes back. Retries a non-200 response for up to -// 30s: ResumeActor can return before its route reaches atenet-router's xDS -// snapshot, and a request sent in that window sees a transient 503. +// the status and body it echoes back. func fetchThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, url string) (int, []byte) { t.Helper() payload, err := json.Marshal(map[string]string{"url": url}) if err != nil { t.Fatalf("marshaling the fetch request for %s: %v", url, err) } + return postThroughEgressActor(t, ctx, router, actorRef, "/", payload) +} + +// postThroughEgressActor POSTs payload to path on the egress demo Actor and +// returns the status and body it answered with. Retries a non-200 response for +// up to 30s: ResumeActor can return before its route reaches atenet-router's +// xDS snapshot, and a request sent in that window sees a transient 503. The +// retry also rides out an origin that is reachable but not yet answering, which +// the Actor reports as a 502. +func postThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, path string, payload []byte) (int, []byte) { + t.Helper() const timeout = 30 * time.Second deadline := time.Now().Add(timeout) for { - response, err := router.PostJSON(ctx, actorRef, "/", payload) + response, err := router.PostJSON(ctx, actorRef, path, payload) if err != nil { - t.Fatalf("POST %s to egress Actor through ingress: %v", url, err) + t.Fatalf("POST %s to egress Actor through ingress: %v", path, err) } body, err := io.ReadAll(response.Body) response.Body.Close() @@ -122,7 +131,7 @@ func fetchThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.Rout if response.StatusCode == http.StatusOK || time.Now().After(deadline) { return response.StatusCode, body } - t.Logf("fetch through egress Actor returned HTTP %d; retrying...", response.StatusCode) + t.Logf("POST %s through egress Actor returned HTTP %d; retrying... body: %s", path, response.StatusCode, body) time.Sleep(1 * time.Second) } } diff --git a/internal/e2e/suites/sdsmint/actoridentity_test.go b/internal/e2e/suites/sdsmint/actoridentity_test.go index 8f2ee175eb..09158ae52f 100644 --- a/internal/e2e/suites/sdsmint/actoridentity_test.go +++ b/internal/e2e/suites/sdsmint/actoridentity_test.go @@ -56,9 +56,9 @@ const ( // gateway rather than reproduce it. actorCertificateLifetime = time.Hour - // Where the probe pod finds the credentials the suite mints for it. Kept in - // step with egressprobe.yaml.tmpl and the --credential-bundle default in - // the probe. + // Where the probe pod finds the credentials the suite mints for it: + // probeServerPod mounts this Secret, and the probe's --credential-bundle + // default reads out of that mount. actorCredentialSecret = "egressprobe-actor-identity" unknownActorCredentialSecret = "egressprobe-unknown-actor" diff --git a/internal/e2e/suites/sdsmint/sdsmint_test.go b/internal/e2e/suites/sdsmint/sdsmint_test.go index 92adc69a69..821604f8a8 100644 --- a/internal/e2e/suites/sdsmint/sdsmint_test.go +++ b/internal/e2e/suites/sdsmint/sdsmint_test.go @@ -58,8 +58,6 @@ import ( "io" "net/http" "net/url" - "os" - "path/filepath" "strings" "sync" "testing" @@ -68,6 +66,7 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" + "k8s.io/utils/ptr" "github.com/agent-substrate/substrate/internal/ateclient" "github.com/agent-substrate/substrate/internal/e2e" @@ -91,6 +90,9 @@ const ( leafSkew = 5 * time.Minute probeName = "egressprobe" + // probePort is the probe's --listen default, which the suite port-forwards + // to and the shared server manifest publishes. + probePort = 8080 ) // skipUntilPresubmit disables this suite. Every test here needs the sdsmint @@ -351,34 +353,18 @@ func sharedProbe(t *testing.T, ctx context.Context) *probeClient { // and deploys the probe, waits for it to be ready, and returns a client for it. func startProbe(t *testing.T, ctx context.Context) *probeClient { t.Helper() + // DeployServerPod checks this too, but only after provisionProbeCredentials + // has created an actor and minted certificates against it. Fail on the + // missing variable before doing that work. if _, err := e2e.CheckEnv("KO_DOCKER_REPO"); err != nil { t.Fatalf("CheckEnv failed: %v", err) } ns := e2e.CreateNamespace(t).Name + // Before the pod: the Secrets below are mounted, so a pod scheduled ahead of + // them sits in ContainerCreating until they appear. provisionProbeCredentials(t, ctx, ns) - root, err := e2e.FindRepoRoot() - if err != nil { - t.Fatalf("FindRepoRoot: %v", err) - } - - tmpl, err := os.ReadFile(filepath.Join(root, "internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl")) - if err != nil { - t.Fatalf("reading egressprobe manifest template: %v", err) - } - manifest := filepath.Join(t.TempDir(), "egressprobe.yaml") - rendered := strings.ReplaceAll(string(tmpl), "${NAMESPACE}", ns) - if err := os.WriteFile(manifest, []byte(rendered), 0o644); err != nil { - t.Fatalf("writing rendered egressprobe manifest: %v", err) - } - - applyArgs := []string{"ko", "apply", "-f", manifest} - if e2e.KubeContext != "" { - applyArgs = append(applyArgs, "--", "--context="+e2e.KubeContext) - } - e2e.RunCmdWithEnv(t, []string{"KO_CONFIG_PATH=" + root}, filepath.Join(root, "hack/run-tool.sh"), applyArgs...) - - waitForProbeReady(t, ctx, ns) + e2e.DeployServerPod(t, ctx, probeServerPod(ns)) config, err := ateclient.LoadConfig(e2e.KubeConfig, e2e.KubeContext) if err != nil { @@ -388,7 +374,7 @@ func startProbe(t *testing.T, ctx context.Context) *probeClient { if err != nil { t.Fatalf("creating k8s client for port-forward: %v", err) } - localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, ns, probeName, 8080) + localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, ns, probeName, probePort) if err != nil { t.Fatalf("port-forwarding %s/%s: %v", ns, probeName, err) } @@ -401,40 +387,63 @@ func startProbe(t *testing.T, ctx context.Context) *probeClient { } } -func waitForProbeReady(t *testing.T, ctx context.Context, ns string) { - t.Helper() - const timeout = 3 * time.Minute - deadline := time.Now().Add(timeout) - var lastState string - for time.Now().Before(deadline) { - pod, err := e2e.GetClients().K8s.CoreV1().Pods(ns).Get(ctx, probeName, metav1.GetOptions{}) - switch { - case err != nil: - lastState = err.Error() - case portforward.IsPodReady(pod): - t.Logf("probe pod %s/%s is ready", ns, probeName) - return - default: - lastState = describeProbeState(pod) - } - time.Sleep(2 * time.Second) - } - t.Fatalf("timed out after %v waiting for probe pod %s/%s to become ready: %s", timeout, ns, probeName, lastState) -} - -func describeProbeState(pod *corev1.Pod) string { - parts := []string{"phase=" + string(pod.Status.Phase)} - for _, cs := range pod.Status.ContainerStatuses { - switch { - case cs.State.Waiting != nil: - parts = append(parts, fmt.Sprintf("%s waiting: %s: %s", cs.Name, cs.State.Waiting.Reason, cs.State.Waiting.Message)) - case cs.State.Terminated != nil: - parts = append(parts, fmt.Sprintf("%s terminated: %s: %s", cs.Name, cs.State.Terminated.Reason, cs.State.Terminated.Message)) - default: - parts = append(parts, fmt.Sprintf("%s running, ready=%t", cs.Name, cs.Ready)) - } +// probeServerPod describes the probe: a plain pod holding every credential the +// suite wants to watch the gateway's front door judge. +// +// The mount paths are the probe's own flag defaults and the paths the tests +// name in handshakeAs, and the Secret names are the ones +// provisionProbeCredentials writes. Both used to live in a manifest of the +// probe's own, one file away from the constants they had to agree with. +func probeServerPod(namespace string) e2e.ServerPod { + return e2e.ServerPod{ + Name: probeName, + Namespace: namespace, + ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/egressprobe", + Port: probePort, + VolumeMounts: []corev1.VolumeMount{ + {Name: "actor-identity", MountPath: "/run/actor-identity"}, + {Name: "actor-identity-unknown", MountPath: "/run/actor-identity-unknown"}, + {Name: "podidentity", MountPath: "/run/podidentity.podcert.ate.dev"}, + {Name: "servicedns-ca", MountPath: "/run/servicedns.podcert.ate.dev"}, + }, + Volumes: []corev1.Volume{{ + // The credential that gets through: an actor-identity leaf the + // suite mints off the CA the gateway trusts. + Name: "actor-identity", + VolumeSource: corev1.VolumeSource{Secret: &corev1.SecretVolumeSource{SecretName: actorCredentialSecret}}, + }, { + // The same shape, for an actor the control plane has never heard + // of: it clears the handshake and must be refused by ext_proc. + Name: "actor-identity-unknown", + VolumeSource: corev1.VolumeSource{Secret: &corev1.SecretVolumeSource{SecretName: unknownActorCredentialSecret}}, + }, { + // The probe's own workload identity -- valid, and not an actor, so + // the gateway must refuse it at the handshake. + Name: "podidentity", + VolumeSource: corev1.VolumeSource{Projected: &corev1.ProjectedVolumeSource{ + Sources: []corev1.VolumeProjection{{ + PodCertificate: &corev1.PodCertificateProjection{ + SignerName: "podidentity.podcert.ate.dev/identity", + KeyType: "ECDSAP256", + CredentialBundlePath: "credential-bundle.pem", + }, + }}, + }}, + }, { + // Verifies the gateway's serving certificate, so a refusal is the + // gateway's decision rather than the probe declining to trust it. + Name: "servicedns-ca", + VolumeSource: corev1.VolumeSource{Projected: &corev1.ProjectedVolumeSource{ + Sources: []corev1.VolumeProjection{{ + ClusterTrustBundle: &corev1.ClusterTrustBundleProjection{ + SignerName: ptr.To("servicedns.podcert.ate.dev/identity"), + LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"podcert.ate.dev/canarying": "live"}}, + Path: "trust-bundle.pem", + }, + }}, + }}, + }}, } - return strings.Join(parts, "; ") } // handshake asks the probe to complete one inner TLS handshake for sni, diff --git a/internal/proto/grpcechopb/gen.go b/internal/proto/grpcechopb/gen.go new file mode 100644 index 0000000000..93a0ceb859 --- /dev/null +++ b/internal/proto/grpcechopb/gen.go @@ -0,0 +1,17 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package grpcechopb + +//go:generate bash -c "../../../hack/protoc.sh --plugin=protoc-gen-go=$(bash ../../../hack/run-tool.sh --print-bin-path protoc-gen-go) --plugin=protoc-gen-go-grpc=$(bash ../../../hack/run-tool.sh --print-bin-path protoc-gen-go-grpc) --go_out=paths=source_relative:. --go-grpc_out=paths=source_relative:. grpcecho.proto" diff --git a/internal/proto/grpcechopb/grpcecho.pb.go b/internal/proto/grpcechopb/grpcecho.pb.go new file mode 100644 index 0000000000..d0daaf535f --- /dev/null +++ b/internal/proto/grpcechopb/grpcecho.pb.go @@ -0,0 +1,265 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11-devel +// protoc v4.25.3 +// source: grpcecho.proto + +package grpcechopb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type EchoRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // The message to be echoed back. + Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EchoRequest) Reset() { + *x = EchoRequest{} + mi := &file_grpcecho_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EchoRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EchoRequest) ProtoMessage() {} + +func (x *EchoRequest) ProtoReflect() protoreflect.Message { + mi := &file_grpcecho_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use EchoRequest.ProtoReflect.Descriptor instead. +func (*EchoRequest) Descriptor() ([]byte, []int) { + return file_grpcecho_proto_rawDescGZIP(), []int{0} +} + +func (x *EchoRequest) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + +type EchoStreamRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // The message to be echoed back in every response. + Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"` + // How many responses to send. Must be positive. + Count int32 `protobuf:"varint,2,opt,name=count,proto3" json:"count,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EchoStreamRequest) Reset() { + *x = EchoStreamRequest{} + mi := &file_grpcecho_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EchoStreamRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EchoStreamRequest) ProtoMessage() {} + +func (x *EchoStreamRequest) ProtoReflect() protoreflect.Message { + mi := &file_grpcecho_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use EchoStreamRequest.ProtoReflect.Descriptor instead. +func (*EchoStreamRequest) Descriptor() ([]byte, []int) { + return file_grpcecho_proto_rawDescGZIP(), []int{1} +} + +func (x *EchoStreamRequest) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + +func (x *EchoStreamRequest) GetCount() int32 { + if x != nil { + return x.Count + } + return 0 +} + +type EchoResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // The message echoed back. + Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"` + // The response's zero-based position in the stream, so a caller can tell a + // reordered or deduplicated stream from an intact one. Always 0 for Echo. + Index int32 `protobuf:"varint,2,opt,name=index,proto3" json:"index,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EchoResponse) Reset() { + *x = EchoResponse{} + mi := &file_grpcecho_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EchoResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EchoResponse) ProtoMessage() {} + +func (x *EchoResponse) ProtoReflect() protoreflect.Message { + mi := &file_grpcecho_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use EchoResponse.ProtoReflect.Descriptor instead. +func (*EchoResponse) Descriptor() ([]byte, []int) { + return file_grpcecho_proto_rawDescGZIP(), []int{2} +} + +func (x *EchoResponse) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + +func (x *EchoResponse) GetIndex() int32 { + if x != nil { + return x.Index + } + return 0 +} + +var File_grpcecho_proto protoreflect.FileDescriptor + +const file_grpcecho_proto_rawDesc = "" + + "\n" + + "\x0egrpcecho.proto\x12\bgrpcecho\"'\n" + + "\vEchoRequest\x12\x18\n" + + "\amessage\x18\x01 \x01(\tR\amessage\"C\n" + + "\x11EchoStreamRequest\x12\x18\n" + + "\amessage\x18\x01 \x01(\tR\amessage\x12\x14\n" + + "\x05count\x18\x02 \x01(\x05R\x05count\">\n" + + "\fEchoResponse\x12\x18\n" + + "\amessage\x18\x01 \x01(\tR\amessage\x12\x14\n" + + "\x05index\x18\x02 \x01(\x05R\x05index2\xc7\x01\n" + + "\x04Echo\x127\n" + + "\x04Echo\x12\x15.grpcecho.EchoRequest\x1a\x16.grpcecho.EchoResponse\"\x00\x12E\n" + + "\n" + + "EchoStream\x12\x1b.grpcecho.EchoStreamRequest\x1a\x16.grpcecho.EchoResponse\"\x000\x01\x12?\n" + + "\bEchoBidi\x12\x15.grpcecho.EchoRequest\x1a\x16.grpcecho.EchoResponse\"\x00(\x010\x01B@Z>github.com/agent-substrate/substrate/internal/proto/grpcechopbb\x06proto3" + +var ( + file_grpcecho_proto_rawDescOnce sync.Once + file_grpcecho_proto_rawDescData []byte +) + +func file_grpcecho_proto_rawDescGZIP() []byte { + file_grpcecho_proto_rawDescOnce.Do(func() { + file_grpcecho_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_grpcecho_proto_rawDesc), len(file_grpcecho_proto_rawDesc))) + }) + return file_grpcecho_proto_rawDescData +} + +var file_grpcecho_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_grpcecho_proto_goTypes = []any{ + (*EchoRequest)(nil), // 0: grpcecho.EchoRequest + (*EchoStreamRequest)(nil), // 1: grpcecho.EchoStreamRequest + (*EchoResponse)(nil), // 2: grpcecho.EchoResponse +} +var file_grpcecho_proto_depIdxs = []int32{ + 0, // 0: grpcecho.Echo.Echo:input_type -> grpcecho.EchoRequest + 1, // 1: grpcecho.Echo.EchoStream:input_type -> grpcecho.EchoStreamRequest + 0, // 2: grpcecho.Echo.EchoBidi:input_type -> grpcecho.EchoRequest + 2, // 3: grpcecho.Echo.Echo:output_type -> grpcecho.EchoResponse + 2, // 4: grpcecho.Echo.EchoStream:output_type -> grpcecho.EchoResponse + 2, // 5: grpcecho.Echo.EchoBidi:output_type -> grpcecho.EchoResponse + 3, // [3:6] is the sub-list for method output_type + 0, // [0:3] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_grpcecho_proto_init() } +func file_grpcecho_proto_init() { + if File_grpcecho_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_grpcecho_proto_rawDesc), len(file_grpcecho_proto_rawDesc)), + NumEnums: 0, + NumMessages: 3, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_grpcecho_proto_goTypes, + DependencyIndexes: file_grpcecho_proto_depIdxs, + MessageInfos: file_grpcecho_proto_msgTypes, + }.Build() + File_grpcecho_proto = out.File + file_grpcecho_proto_goTypes = nil + file_grpcecho_proto_depIdxs = nil +} diff --git a/internal/proto/grpcechopb/grpcecho.proto b/internal/proto/grpcechopb/grpcecho.proto new file mode 100644 index 0000000000..099d186324 --- /dev/null +++ b/internal/proto/grpcechopb/grpcecho.proto @@ -0,0 +1,72 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package grpcecho; + +option go_package = "github.com/agent-substrate/substrate/internal/proto/grpcechopb"; + +// Echo is the smallest service that can prove a gRPC request and its response +// crossed a network path intact. It exists for the egress e2e coverage: the +// origin is internal/e2e/fixtures/grpcecho and the client is the egress demo +// actor, which reaches it through the egress gateway's CONNECT tunnel. +// +// One method per streaming shape, because they fail differently. A unary +// response carries its status in trailers sent immediately after the message; a +// server-stream holds the connection open across many frames; a bidirectional +// stream has both directions carrying frames at once and then half-closes one +// of them. A path that buffers, downgrades to HTTP/1.1, serializes the two +// directions, or reaps an idle tunnel breaks one of these without touching the +// others. +service Echo { + // Echo returns the message it was sent. + rpc Echo(EchoRequest) returns (EchoResponse) {} + + // EchoStream returns the message it was sent count times, one response per + // message, each numbered with its position in the stream. + rpc EchoStream(EchoStreamRequest) returns (stream EchoResponse) {} + + // EchoBidi answers every request with one response, as each request arrives, + // numbered with its position in the stream. The caller decides how many to + // send and half-closes when it is done. + // + // Unlike EchoStream, the server must not wait for the client to finish: a + // caller that sends its next request only after reading the previous response + // deadlocks against an implementation -- or a relay -- that reads the request + // direction to exhaustion before writing anything back. + rpc EchoBidi(stream EchoRequest) returns (stream EchoResponse) {} +} + +message EchoRequest { + // The message to be echoed back. + string message = 1; +} + +message EchoStreamRequest { + // The message to be echoed back in every response. + string message = 1; + + // How many responses to send. Must be positive. + int32 count = 2; +} + +message EchoResponse { + // The message echoed back. + string message = 1; + + // The response's zero-based position in the stream, so a caller can tell a + // reordered or deduplicated stream from an intact one. Always 0 for Echo. + int32 index = 2; +} diff --git a/internal/proto/grpcechopb/grpcecho_grpc.pb.go b/internal/proto/grpcechopb/grpcecho_grpc.pb.go new file mode 100644 index 0000000000..bdec199976 --- /dev/null +++ b/internal/proto/grpcechopb/grpcecho_grpc.pb.go @@ -0,0 +1,257 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.1 +// - protoc v4.25.3 +// source: grpcecho.proto + +package grpcechopb + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + Echo_Echo_FullMethodName = "/grpcecho.Echo/Echo" + Echo_EchoStream_FullMethodName = "/grpcecho.Echo/EchoStream" + Echo_EchoBidi_FullMethodName = "/grpcecho.Echo/EchoBidi" +) + +// EchoClient is the client API for Echo service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +// +// Echo is the smallest service that can prove a gRPC request and its response +// crossed a network path intact. It exists for the egress e2e coverage: the +// origin is internal/e2e/fixtures/grpcecho and the client is the egress demo +// actor, which reaches it through the egress gateway's CONNECT tunnel. +// +// One method per streaming shape, because they fail differently. A unary +// response carries its status in trailers sent immediately after the message; a +// server-stream holds the connection open across many frames; a bidirectional +// stream has both directions carrying frames at once and then half-closes one +// of them. A path that buffers, downgrades to HTTP/1.1, serializes the two +// directions, or reaps an idle tunnel breaks one of these without touching the +// others. +type EchoClient interface { + // Echo returns the message it was sent. + Echo(ctx context.Context, in *EchoRequest, opts ...grpc.CallOption) (*EchoResponse, error) + // EchoStream returns the message it was sent count times, one response per + // message, each numbered with its position in the stream. + EchoStream(ctx context.Context, in *EchoStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[EchoResponse], error) + // EchoBidi answers every request with one response, as each request arrives, + // numbered with its position in the stream. The caller decides how many to + // send and half-closes when it is done. + // + // Unlike EchoStream, the server must not wait for the client to finish: a + // caller that sends its next request only after reading the previous response + // deadlocks against an implementation -- or a relay -- that reads the request + // direction to exhaustion before writing anything back. + EchoBidi(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[EchoRequest, EchoResponse], error) +} + +type echoClient struct { + cc grpc.ClientConnInterface +} + +func NewEchoClient(cc grpc.ClientConnInterface) EchoClient { + return &echoClient{cc} +} + +func (c *echoClient) Echo(ctx context.Context, in *EchoRequest, opts ...grpc.CallOption) (*EchoResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(EchoResponse) + err := c.cc.Invoke(ctx, Echo_Echo_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *echoClient) EchoStream(ctx context.Context, in *EchoStreamRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[EchoResponse], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &Echo_ServiceDesc.Streams[0], Echo_EchoStream_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[EchoStreamRequest, EchoResponse]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Echo_EchoStreamClient = grpc.ServerStreamingClient[EchoResponse] + +func (c *echoClient) EchoBidi(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[EchoRequest, EchoResponse], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &Echo_ServiceDesc.Streams[1], Echo_EchoBidi_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[EchoRequest, EchoResponse]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Echo_EchoBidiClient = grpc.BidiStreamingClient[EchoRequest, EchoResponse] + +// EchoServer is the server API for Echo service. +// All implementations must embed UnimplementedEchoServer +// for forward compatibility. +// +// Echo is the smallest service that can prove a gRPC request and its response +// crossed a network path intact. It exists for the egress e2e coverage: the +// origin is internal/e2e/fixtures/grpcecho and the client is the egress demo +// actor, which reaches it through the egress gateway's CONNECT tunnel. +// +// One method per streaming shape, because they fail differently. A unary +// response carries its status in trailers sent immediately after the message; a +// server-stream holds the connection open across many frames; a bidirectional +// stream has both directions carrying frames at once and then half-closes one +// of them. A path that buffers, downgrades to HTTP/1.1, serializes the two +// directions, or reaps an idle tunnel breaks one of these without touching the +// others. +type EchoServer interface { + // Echo returns the message it was sent. + Echo(context.Context, *EchoRequest) (*EchoResponse, error) + // EchoStream returns the message it was sent count times, one response per + // message, each numbered with its position in the stream. + EchoStream(*EchoStreamRequest, grpc.ServerStreamingServer[EchoResponse]) error + // EchoBidi answers every request with one response, as each request arrives, + // numbered with its position in the stream. The caller decides how many to + // send and half-closes when it is done. + // + // Unlike EchoStream, the server must not wait for the client to finish: a + // caller that sends its next request only after reading the previous response + // deadlocks against an implementation -- or a relay -- that reads the request + // direction to exhaustion before writing anything back. + EchoBidi(grpc.BidiStreamingServer[EchoRequest, EchoResponse]) error + mustEmbedUnimplementedEchoServer() +} + +// UnimplementedEchoServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedEchoServer struct{} + +func (UnimplementedEchoServer) Echo(context.Context, *EchoRequest) (*EchoResponse, error) { + return nil, status.Error(codes.Unimplemented, "method Echo not implemented") +} +func (UnimplementedEchoServer) EchoStream(*EchoStreamRequest, grpc.ServerStreamingServer[EchoResponse]) error { + return status.Error(codes.Unimplemented, "method EchoStream not implemented") +} +func (UnimplementedEchoServer) EchoBidi(grpc.BidiStreamingServer[EchoRequest, EchoResponse]) error { + return status.Error(codes.Unimplemented, "method EchoBidi not implemented") +} +func (UnimplementedEchoServer) mustEmbedUnimplementedEchoServer() {} +func (UnimplementedEchoServer) testEmbeddedByValue() {} + +// UnsafeEchoServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to EchoServer will +// result in compilation errors. +type UnsafeEchoServer interface { + mustEmbedUnimplementedEchoServer() +} + +func RegisterEchoServer(s grpc.ServiceRegistrar, srv EchoServer) { + // If the following call panics, it indicates UnimplementedEchoServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&Echo_ServiceDesc, srv) +} + +func _Echo_Echo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(EchoRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(EchoServer).Echo(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Echo_Echo_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(EchoServer).Echo(ctx, req.(*EchoRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Echo_EchoStream_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(EchoStreamRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(EchoServer).EchoStream(m, &grpc.GenericServerStream[EchoStreamRequest, EchoResponse]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Echo_EchoStreamServer = grpc.ServerStreamingServer[EchoResponse] + +func _Echo_EchoBidi_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(EchoServer).EchoBidi(&grpc.GenericServerStream[EchoRequest, EchoResponse]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Echo_EchoBidiServer = grpc.BidiStreamingServer[EchoRequest, EchoResponse] + +// Echo_ServiceDesc is the grpc.ServiceDesc for Echo service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var Echo_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "grpcecho.Echo", + HandlerType: (*EchoServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "Echo", + Handler: _Echo_Echo_Handler, + }, + }, + Streams: []grpc.StreamDesc{ + { + StreamName: "EchoStream", + Handler: _Echo_EchoStream_Handler, + ServerStreams: true, + }, + { + StreamName: "EchoBidi", + Handler: _Echo_EchoBidi_Handler, + ServerStreams: true, + ClientStreams: true, + }, + }, + Metadata: "grpcecho.proto", +} From cd52520b8ccc8c32f301618876c50032fa97cacc Mon Sep 17 00:00:00 2001 From: Yufan Su Date: Mon, 24 Aug 2026 11:04:13 -0700 Subject: [PATCH 2/2] revert the change in sdsmint test suite and egressprobe yaml --- demos/egress/README.md | 2 +- .../egressprobe/egressprobe.yaml.tmpl | 113 ++++++++++++++++ .../e2e/suites/sdsmint/actoridentity_test.go | 6 +- internal/e2e/suites/sdsmint/sdsmint_test.go | 125 ++++++++---------- 4 files changed, 175 insertions(+), 71 deletions(-) create mode 100644 internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl diff --git a/demos/egress/README.md b/demos/egress/README.md index bbe4ccff0f..58e07acd3a 100644 --- a/demos/egress/README.md +++ b/demos/egress/README.md @@ -162,7 +162,7 @@ The gateway terminates the `CONNECT` and relays opaque TCP, so all of this cross failed RPC answers `502` and still carries its gRPC code in the same field. `internal/e2e/suites/networking` drives this endpoint against the `grpcecho` fixture -(`internal/e2e/fixtures/grpcecho`, deployed by `e2e.DeployOriginPod` from the shared origin +(`internal/e2e/fixtures/grpcecho`, deployed by `e2e.DeployServerPod` from the shared server-pod manifest) in `TestActorEgressGRPC`; the same fixture, or any h2c gRPC server reachable from the cluster, works for a manual run. diff --git a/internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl b/internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl new file mode 100644 index 0000000000..a07f096294 --- /dev/null +++ b/internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl @@ -0,0 +1,113 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# The egress probe used by internal/e2e/suites/sdsmint. ${NAMESPACE} is +# substituted by the suite with the randomized namespace it created, so the +# probe is torn down with that namespace and leaves nothing behind. +# +# A plain Pod, not an actor: the suite is testing the gateway's MITM leg, and +# running the probe itself inside an actor would make a snapshot or restore +# failure look like an sdsmint failure. The suite does create one actor, but +# only for its identity -- the certificate mounted below is minted for it, and +# the actor's own workload is never contacted. + +apiVersion: v1 +kind: Pod +metadata: + name: egressprobe + namespace: ${NAMESPACE} + labels: + app: egressprobe +spec: + restartPolicy: Never + containers: + - name: egressprobe + image: ko://github.com/agent-substrate/substrate/internal/e2e/fixtures/egressprobe + args: + - "--listen=:8080" + ports: + - name: http + containerPort: 8080 + # The suite port-forwards to this pod, and port-forward targets the first + # READY pod behind the Service. Without a readiness gate the forward can + # attach before the listener exists and the first handshake fails as a + # connection refused that looks like a gateway problem. + readinessProbe: + httpGet: + path: /healthz + port: 8080 + periodSeconds: 2 + resources: + requests: + cpu: 10m + memory: 32Mi + # runAsUser must be spelled out: ko's distroless static base declares no + # USER, so runAsNonRoot on its own makes kubelet refuse to start the + # container ("image will run as root") rather than pick a uid. 65532 is + # distroless's nonroot uid, and the same one the egress gateway's pod uses. + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 65532 + runAsGroup: 65532 + capabilities: + drop: ["ALL"] + volumeMounts: + - name: "actor-identity" + mountPath: "/run/actor-identity" + - name: "actor-identity-unknown" + mountPath: "/run/actor-identity-unknown" + - name: "podidentity" + mountPath: "/run/podidentity.podcert.ate.dev" + - name: "servicedns-ca" + mountPath: "/run/servicedns.podcert.ate.dev" + volumes: + - name: "actor-identity" + secret: + secretName: egressprobe-actor-identity + - name: "actor-identity-unknown" + secret: + secretName: egressprobe-unknown-actor + - name: "podidentity" + projected: + sources: + - podCertificate: + signerName: podidentity.podcert.ate.dev/identity + keyType: ECDSAP256 + credentialBundlePath: credential-bundle.pem + - name: "servicedns-ca" + projected: + sources: + - clusterTrustBundle: + signerName: servicedns.podcert.ate.dev/identity + labelSelector: + matchLabels: + podcert.ate.dev/canarying: live + path: trust-bundle.pem + +--- + +apiVersion: v1 +kind: Service +metadata: + name: egressprobe + namespace: ${NAMESPACE} +spec: + selector: + app: egressprobe + ports: + - name: http + port: 8080 + targetPort: 8080 diff --git a/internal/e2e/suites/sdsmint/actoridentity_test.go b/internal/e2e/suites/sdsmint/actoridentity_test.go index 09158ae52f..8f2ee175eb 100644 --- a/internal/e2e/suites/sdsmint/actoridentity_test.go +++ b/internal/e2e/suites/sdsmint/actoridentity_test.go @@ -56,9 +56,9 @@ const ( // gateway rather than reproduce it. actorCertificateLifetime = time.Hour - // Where the probe pod finds the credentials the suite mints for it: - // probeServerPod mounts this Secret, and the probe's --credential-bundle - // default reads out of that mount. + // Where the probe pod finds the credentials the suite mints for it. Kept in + // step with egressprobe.yaml.tmpl and the --credential-bundle default in + // the probe. actorCredentialSecret = "egressprobe-actor-identity" unknownActorCredentialSecret = "egressprobe-unknown-actor" diff --git a/internal/e2e/suites/sdsmint/sdsmint_test.go b/internal/e2e/suites/sdsmint/sdsmint_test.go index 821604f8a8..92adc69a69 100644 --- a/internal/e2e/suites/sdsmint/sdsmint_test.go +++ b/internal/e2e/suites/sdsmint/sdsmint_test.go @@ -58,6 +58,8 @@ import ( "io" "net/http" "net/url" + "os" + "path/filepath" "strings" "sync" "testing" @@ -66,7 +68,6 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" - "k8s.io/utils/ptr" "github.com/agent-substrate/substrate/internal/ateclient" "github.com/agent-substrate/substrate/internal/e2e" @@ -90,9 +91,6 @@ const ( leafSkew = 5 * time.Minute probeName = "egressprobe" - // probePort is the probe's --listen default, which the suite port-forwards - // to and the shared server manifest publishes. - probePort = 8080 ) // skipUntilPresubmit disables this suite. Every test here needs the sdsmint @@ -353,18 +351,34 @@ func sharedProbe(t *testing.T, ctx context.Context) *probeClient { // and deploys the probe, waits for it to be ready, and returns a client for it. func startProbe(t *testing.T, ctx context.Context) *probeClient { t.Helper() - // DeployServerPod checks this too, but only after provisionProbeCredentials - // has created an actor and minted certificates against it. Fail on the - // missing variable before doing that work. if _, err := e2e.CheckEnv("KO_DOCKER_REPO"); err != nil { t.Fatalf("CheckEnv failed: %v", err) } ns := e2e.CreateNamespace(t).Name - // Before the pod: the Secrets below are mounted, so a pod scheduled ahead of - // them sits in ContainerCreating until they appear. provisionProbeCredentials(t, ctx, ns) - e2e.DeployServerPod(t, ctx, probeServerPod(ns)) + root, err := e2e.FindRepoRoot() + if err != nil { + t.Fatalf("FindRepoRoot: %v", err) + } + + tmpl, err := os.ReadFile(filepath.Join(root, "internal/e2e/fixtures/egressprobe/egressprobe.yaml.tmpl")) + if err != nil { + t.Fatalf("reading egressprobe manifest template: %v", err) + } + manifest := filepath.Join(t.TempDir(), "egressprobe.yaml") + rendered := strings.ReplaceAll(string(tmpl), "${NAMESPACE}", ns) + if err := os.WriteFile(manifest, []byte(rendered), 0o644); err != nil { + t.Fatalf("writing rendered egressprobe manifest: %v", err) + } + + applyArgs := []string{"ko", "apply", "-f", manifest} + if e2e.KubeContext != "" { + applyArgs = append(applyArgs, "--", "--context="+e2e.KubeContext) + } + e2e.RunCmdWithEnv(t, []string{"KO_CONFIG_PATH=" + root}, filepath.Join(root, "hack/run-tool.sh"), applyArgs...) + + waitForProbeReady(t, ctx, ns) config, err := ateclient.LoadConfig(e2e.KubeConfig, e2e.KubeContext) if err != nil { @@ -374,7 +388,7 @@ func startProbe(t *testing.T, ctx context.Context) *probeClient { if err != nil { t.Fatalf("creating k8s client for port-forward: %v", err) } - localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, ns, probeName, probePort) + localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, ns, probeName, 8080) if err != nil { t.Fatalf("port-forwarding %s/%s: %v", ns, probeName, err) } @@ -387,63 +401,40 @@ func startProbe(t *testing.T, ctx context.Context) *probeClient { } } -// probeServerPod describes the probe: a plain pod holding every credential the -// suite wants to watch the gateway's front door judge. -// -// The mount paths are the probe's own flag defaults and the paths the tests -// name in handshakeAs, and the Secret names are the ones -// provisionProbeCredentials writes. Both used to live in a manifest of the -// probe's own, one file away from the constants they had to agree with. -func probeServerPod(namespace string) e2e.ServerPod { - return e2e.ServerPod{ - Name: probeName, - Namespace: namespace, - ImportPath: "github.com/agent-substrate/substrate/internal/e2e/fixtures/egressprobe", - Port: probePort, - VolumeMounts: []corev1.VolumeMount{ - {Name: "actor-identity", MountPath: "/run/actor-identity"}, - {Name: "actor-identity-unknown", MountPath: "/run/actor-identity-unknown"}, - {Name: "podidentity", MountPath: "/run/podidentity.podcert.ate.dev"}, - {Name: "servicedns-ca", MountPath: "/run/servicedns.podcert.ate.dev"}, - }, - Volumes: []corev1.Volume{{ - // The credential that gets through: an actor-identity leaf the - // suite mints off the CA the gateway trusts. - Name: "actor-identity", - VolumeSource: corev1.VolumeSource{Secret: &corev1.SecretVolumeSource{SecretName: actorCredentialSecret}}, - }, { - // The same shape, for an actor the control plane has never heard - // of: it clears the handshake and must be refused by ext_proc. - Name: "actor-identity-unknown", - VolumeSource: corev1.VolumeSource{Secret: &corev1.SecretVolumeSource{SecretName: unknownActorCredentialSecret}}, - }, { - // The probe's own workload identity -- valid, and not an actor, so - // the gateway must refuse it at the handshake. - Name: "podidentity", - VolumeSource: corev1.VolumeSource{Projected: &corev1.ProjectedVolumeSource{ - Sources: []corev1.VolumeProjection{{ - PodCertificate: &corev1.PodCertificateProjection{ - SignerName: "podidentity.podcert.ate.dev/identity", - KeyType: "ECDSAP256", - CredentialBundlePath: "credential-bundle.pem", - }, - }}, - }}, - }, { - // Verifies the gateway's serving certificate, so a refusal is the - // gateway's decision rather than the probe declining to trust it. - Name: "servicedns-ca", - VolumeSource: corev1.VolumeSource{Projected: &corev1.ProjectedVolumeSource{ - Sources: []corev1.VolumeProjection{{ - ClusterTrustBundle: &corev1.ClusterTrustBundleProjection{ - SignerName: ptr.To("servicedns.podcert.ate.dev/identity"), - LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"podcert.ate.dev/canarying": "live"}}, - Path: "trust-bundle.pem", - }, - }}, - }}, - }}, +func waitForProbeReady(t *testing.T, ctx context.Context, ns string) { + t.Helper() + const timeout = 3 * time.Minute + deadline := time.Now().Add(timeout) + var lastState string + for time.Now().Before(deadline) { + pod, err := e2e.GetClients().K8s.CoreV1().Pods(ns).Get(ctx, probeName, metav1.GetOptions{}) + switch { + case err != nil: + lastState = err.Error() + case portforward.IsPodReady(pod): + t.Logf("probe pod %s/%s is ready", ns, probeName) + return + default: + lastState = describeProbeState(pod) + } + time.Sleep(2 * time.Second) + } + t.Fatalf("timed out after %v waiting for probe pod %s/%s to become ready: %s", timeout, ns, probeName, lastState) +} + +func describeProbeState(pod *corev1.Pod) string { + parts := []string{"phase=" + string(pod.Status.Phase)} + for _, cs := range pod.Status.ContainerStatuses { + switch { + case cs.State.Waiting != nil: + parts = append(parts, fmt.Sprintf("%s waiting: %s: %s", cs.Name, cs.State.Waiting.Reason, cs.State.Waiting.Message)) + case cs.State.Terminated != nil: + parts = append(parts, fmt.Sprintf("%s terminated: %s: %s", cs.Name, cs.State.Terminated.Reason, cs.State.Terminated.Message)) + default: + parts = append(parts, fmt.Sprintf("%s running, ready=%t", cs.Name, cs.Ready)) + } } + return strings.Join(parts, "; ") } // handshake asks the probe to complete one inner TLS handshake for sni,