diff --git a/.ko.yaml b/.ko.yaml index 4afb2e7e1..fd8e85968 100644 --- a/.ko.yaml +++ b/.ko.yaml @@ -20,7 +20,6 @@ defaultPlatforms: baseImageOverrides: github.com/agent-substrate/substrate/demos/sandbox: alpine - github.com/agent-substrate/substrate/demos/agent-secret: alpine # ateom-microvm needs glibc (for the fetched cloud-hypervisor binary) and mount/umount # (to bind the image into the virtiofsd shared dir) — both in debian:stable-slim but # not in the distroless static default. diff --git a/README.md b/README.md index 231a4be09..1cedd0697 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,7 @@ This demo highlights the core developer experience and "Agentic Infrastructure" 2. **State Persistence:** Persistent working memory (volatile RAM) and filesystem state preserved perfectly across hibernation cycles via full-state snapshots. 3. **Agent Swarm Multiplexing:** Demonstrates 30x+ oversubscription by "juggling" a large registry of stateful actors onto a small pool of shared physical pods. -To reproduce this demo in your own cluster, please refer to the detailed walkthroughs in the **[Counter Demo](demos/counter/README.md)** and **[Secret Agent Demo](demos/agent-secret/README.md)**. +To reproduce this demo in your own cluster, please refer to the detailed walkthrough in the **[Counter Demo](demos/counter/README.md)**. For more videos and walkthroughs, visit our YouTube channel: **[agent-substrate](https://www.youtube.com/channel/UCN9PPqlTtVxlcpbQ-NWpfZQ)**. @@ -199,7 +199,6 @@ We provide several sample applications demonstrating Agent Substrate's capabilit 1. **[Counter Demo](demos/counter/README.md)**: A stateful Go HTTP server demonstrating state preservation across suspends/resumes, and dynamic CRD routing. 2. **[Sandbox Demo (Antigravity)](demos/sandbox/README.md)**: A secure, sandboxed execution environment (running Alpine Linux) that allows arbitrary shell execution while preserving filesystem state across sessions. 3. **[Claude Code Multiplex](demos/claude-code-multiplex/README.md)**: Demonstrates oversubscribing physical hardware by multiplexing multiple Claude Code agents onto a limited pool of workers. -4. **[Secret Agent](demos/agent-secret/README.md)**: Highlights Substrate's "Zero-Idle" self-suspension and re-animation of volatile process memory. ### Documentation & Guides * [API Configuration Guide](docs/api-guide.md): Detailed reference for configuring WorkerPools, ActorTemplates, Secrets, and Volumes. diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go index 7ddc25cad..fbc562636 100644 --- a/cmd/ateapi/main.go +++ b/cmd/ateapi/main.go @@ -51,6 +51,7 @@ import ( "golang.org/x/oauth2/google" "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" "google.golang.org/grpc/reflection" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" @@ -195,6 +196,14 @@ func main() { mux := grpc.NewServer( grpc.Creds(serverCreds), grpc.StatsHandler(otelgrpc.NewServerHandler()), + // Bounds every connection's lifetime so round_robin clients + // periodically re-resolve DNS and pick up replicas added since they + // last connected - without this, an existing connection never + // notices new replicas on its own (see https://github.com/grpc/grpc/issues/12295). + grpc.KeepaliveParams(keepalive.ServerParameters{ + MaxConnectionAge: 1 * time.Minute, + MaxConnectionAgeGrace: maxRPCDeadline + time.Minute, + }), grpc.ChainUnaryInterceptor( ateapiauth.UnaryServerInterceptor(authCfg), ateinterceptors.MaxDeadlineUnaryInterceptor(maxRPCDeadline), diff --git a/cmd/atenet/internal/router/cmd.go b/cmd/atenet/internal/router/cmd.go index 3cddd048d..3e7a6957b 100644 --- a/cmd/atenet/internal/router/cmd.go +++ b/cmd/atenet/internal/router/cmd.go @@ -43,7 +43,7 @@ func NewRouterCmd() *cobra.Command { cmd.Flags().BoolVar(&cfg.Standalone, "standalone", false, "Run in standalone mode, bypassing creation of managed deployment and services in Kubernetes cluster") cmd.Flags().StringVar(&cfg.Namespace, "namespace", "default", "Target operations namespace") cmd.Flags().StringVar(&cfg.Kubeconfig, "kubeconfig", "", "Absolute path to the kubeconfig configuration file") - cmd.Flags().StringVar(&cfg.AteapiAddr, "ateapi-address", "api.ate-system.svc:443", "gRPC host address of the cluster ateapi Control instance") + cmd.Flags().StringVar(&cfg.AteapiAddr, "ateapi-address", "dns:///api.ate-system.svc:443", "gRPC dial target for the cluster ateapi Control instance.") cmd.Flags().IntVar(&cfg.HttpPort, "port-http", 8080, "TCP port for workload traffic entering through the Envoy Router") cmd.Flags().IntVar(&cfg.XdsPort, "port-xds", 18000, "TCP port listening for the xDS dynamic Envoy connections") cmd.Flags().IntVar(&cfg.ExtprocPort, "port-extproc", 50051, "Listen port for the Envoy dynamic External Processing (ext_proc) server") diff --git a/cmd/benchmarking/boomer-glutton/main.go b/cmd/benchmarking/boomer-glutton/main.go index 9a5817bfd..efa921ef8 100644 --- a/cmd/benchmarking/boomer-glutton/main.go +++ b/cmd/benchmarking/boomer-glutton/main.go @@ -36,7 +36,7 @@ import ( func main() { var ( - apiEndpoint = flag.String("api-endpoint", "api.ate-system.svc.cluster.local:443", "ateapi gRPC endpoint host:port.") + apiEndpoint = flag.String("api-endpoint", "dns:///api.ate-system.svc.cluster.local:443", "ateapi gRPC dial target.") routerURL = flag.String("router-url", "http://atenet-router.ate-system.svc.cluster.local", "atenet HTTP router base URL (no trailing slash).") atespace = flag.String("atespace", "benchmark", "Atespace every actor this worker creates lives in. Ensured (CreateAtespace, AlreadyExists is ok) at startup.") promAddr = flag.String("prometheus-addr", ":8001", "Address for the Prometheus /metrics endpoint.") diff --git a/demos/agent-secret/README.md b/demos/agent-secret/README.md deleted file mode 100644 index e5da14966..000000000 --- a/demos/agent-secret/README.md +++ /dev/null @@ -1,92 +0,0 @@ -# Secret Agent Demo: Self-Suspension & Persistent Identity - -This demo showcases a "Zero-Idle" agentic lifecycle using a specialized Go server. It demonstrates how Agent Substrate treats process state as a portable asset, enabling both **Self-Suspension** and **Persistent Working Memory** (via a RAM-based secret). - -## Key Concepts Demonstrated -* **Self-Suspension:** The agent process calls the Substrate Control Plane API (`SuspendActor`) automatically after responding to a request, yielding compute resources as soon as it becomes idle. -* **Persistent Working Memory:** The agent generates a random secret in its volatile RAM on process start. Substrate preserves this exact memory state across physical resumptions, proving it doesn't just "restart" containers but actually rehydrates living processes. -* **High-Density Multiplexing:** Demonstrates how dozens of isolated, stateful sessions can coexist on a small pool of physical hardware. - ---- - -## 🛠️ Step-by-Step Setup - -### 1. Prerequisites -* A GKE cluster with **Agent Substrate** installed. -* `ko` installed (for building and deploying Go images). -* `kubectl-ate` CLI installed (can be installed via `go install ./cmd/kubectl-ate`). - -### 2. Deploy the Infrastructure -Use the core installation script to build the image and apply the manifests: -```bash -./hack/install-ate.sh --deploy-demo-agent-secret -``` - -### 3. Scale the Worker Pool -To demonstrate oversubscription and density, we recommend scaling the physical pool to 8 workers: -```bash -kubectl patch workerpool agent-secret -n ate-demo-secret-agent-v2 \ - --type='merge' -p '{"spec":{"replicas":8}}' -``` - ---- - -## 📽️ Interaction Guide - -### 1. Basic Interaction -Actors live in an **atespace**, which must exist first. Create one, then create a single actor and watch it automatically yield compute after use: -```bash -# Create the atespace (required before creating actors). -kubectl ate create atespace demo - -# Create the actor -kubectl ate create actor my-agent -a demo --template ate-demo-secret-agent-v2/agent-secret - -# Send a request via the Substrate Router (Note the official DNS suffix) -curl -H "Host: my-agent.demo.actors.resources.substrate.ate.dev" http://localhost:8000 -``` - -**What to observe:** -* In a separate terminal, run `watch kubectl ate get actors`. -* Notice that the actor status flips to `STATUS_RUNNING` instantly upon the request, and then **automatically** flips back to `STATUS_SUSPENDED` after the 7-second "visibility linger" period. - -### 2. Verify Identity Persistence -Send another request to the same actor: -```bash -curl -H "Host: my-agent.demo.actors.resources.substrate.ate.dev" http://localhost:8000 -``` -The "Identity" secret returned will be identical to the first response, even if the actor was resumed on a different physical pod. This proves the volatile RAM survived the hibernation cycle. - -### 3. Demonstrating Massive Density (The "Wave" Swarm) -To show the scale at which Substrate can manage sessions, run this loop to populate the registry with 23 additional stateful sessions: - -```bash -for i in {001..023}; do - kubectl ate create actor session-$i -a demo --template ate-demo-secret-agent-v2/agent-secret -done -``` - -Now, trigger a "Wave Pulse" to show multiplexing in action. This command sends requests in groups of 8, allowing you to see the physical grid fill and clear in a rhythmic "conveyor belt" fashion: - -```bash -# Pulse in 3 waves of 8 -for wave in 0 1 2; do - echo "Triggering Wave $((wave + 1))..." - for i in {1..8}; do - num=$(printf "%03d" $((wave * 8 + i))) - curl -s -H "Host: session-$num.demo.actors.resources.substrate.ate.dev" http://localhost:8000 & - done - sleep 8 # 7s linger + 1s buffer -done -``` - -### 4. Clean Up -Delete the actors, then the now-empty atespace: -```bash -kubectl ate delete actor my-agent -a demo -for i in {001..023}; do kubectl ate delete actor session-$i -a demo; done -kubectl ate delete atespace demo - -# Remove the demo infrastructure entirely: -./hack/install-ate.sh --delete-demo-agent-secret -``` diff --git a/demos/agent-secret/agent-secret.yaml.tmpl b/demos/agent-secret/agent-secret.yaml.tmpl deleted file mode 100644 index 442f2e2b6..000000000 --- a/demos/agent-secret/agent-secret.yaml.tmpl +++ /dev/null @@ -1,76 +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. - -apiVersion: v1 -kind: Namespace -metadata: - name: ate-demo-secret-agent-v2 ---- -# Grant ate-api-server read access to only the Secrets that this demo's -# ActorTemplate references for env resolution. Update resourceNames as new -# references are added. -apiVersion: rbac.authorization.k8s.io/v1 -kind: Role -metadata: - name: ate-api-server-env-sources - namespace: ate-demo-secret-agent-v2 -rules: -- apiGroups: [""] - resources: ["secrets"] - resourceNames: [] - verbs: ["get"] ---- -apiVersion: rbac.authorization.k8s.io/v1 -kind: RoleBinding -metadata: - name: ate-api-server-env-sources - namespace: ate-demo-secret-agent-v2 -subjects: -- kind: ServiceAccount - name: ate-api-server - namespace: ate-system -roleRef: - kind: Role - name: ate-api-server-env-sources - apiGroup: rbac.authorization.k8s.io ---- -apiVersion: ate.dev/v1alpha1 -kind: WorkerPool -metadata: - name: agent-secret - namespace: ate-demo-secret-agent-v2 - labels: - workload: agent-secret -spec: - replicas: 8 - ateomImage: ko://github.com/agent-substrate/substrate/cmd/ateom-gvisor ---- -apiVersion: ate.dev/v1alpha1 -kind: ActorTemplate -metadata: - name: agent-secret - namespace: ate-demo-secret-agent-v2 -spec: - pauseImage: "registry.k8s.io/pause:3.10.2@sha256:f548e0e8e3dc1896ca956272154dde3314e8cc4fde0a57577ee9fa1c63f5baf4" - containers: - - name: agent-secret - image: ko://github.com/agent-substrate/substrate/demos/agent-secret - env: - - name: PORT - value: "80" - workerSelector: - matchLabels: - workload: agent-secret - snapshotsConfig: - location: gs://${BUCKET_NAME}/ate-demo-secret-agent-v2/ diff --git a/demos/agent-secret/main.go b/demos/agent-secret/main.go deleted file mode 100644 index 9758d0170..000000000 --- a/demos/agent-secret/main.go +++ /dev/null @@ -1,125 +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. - -package main - -import ( - "context" - "crypto/tls" - "fmt" - "io" - "log" - "net/http" - "os" - "strings" - "time" - - "github.com/agent-substrate/substrate/internal/resources" - "github.com/agent-substrate/substrate/pkg/proto/ateapipb" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials" -) - -var residentSecret string - -func init() { - // Unique ID generated in volatile RAM on process start - residentSecret = fmt.Sprintf("SECRET-%d", time.Now().UnixNano()%10000) -} - -func main() { - mux := http.NewServeMux() - mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - w.Write([]byte("OK")) - }) - mux.HandleFunc("/", handleRequest) - - port := os.Getenv("PORT") - if port == "" { - port = "80" - } - - log.Printf("Self-Suspending Agent listening on :%s with Identity: %s", port, residentSecret) - if err := http.ListenAndServe(":"+port, mux); err != nil { - log.Fatalf("failed to serve: %v", err) - } -} - -func handleRequest(w http.ResponseWriter, r *http.Request) { - // 1. Identify Actor - actorName := r.Header.Get("X-AgentSet-Session") - if actorName == "" { - actorName = r.Header.Get("x-agentset-session") - } - var atespace string - if actorName == "" { - host := r.Host - if host == "" { - host = r.Header.Get("Host") - } - atespace, actorName, _ = resources.ParseActorDNSName(host) - } - - if actorName == "" { - actorName = "unknown" - } - - body, _ := io.ReadAll(r.Body) - message := string(body) - if message == "" { - message = "Status Check" - } - - // 2. Respond - var sb strings.Builder - sb.WriteString(fmt.Sprintf("Agent Response: [%s] | Identity: %s | Session: %s\n", message, residentSecret, actorName)) - response := sb.String() - w.Header().Set("Content-Type", "text/plain") - w.WriteHeader(http.StatusOK) - w.Write([]byte(response)) - - // 3. Self-Suspend (Zero-Idle) - if actorName != "" && actorName != "localhost" && !strings.Contains(actorName, ":") { - // Use a goroutine to avoid blocking the HTTP response - go func() { - // We linger for 7 seconds in this demo to make the multiplexing visible in the CLI. - time.Sleep(7 * time.Second) - suspendSelf(atespace, actorName) - }() - } -} - -func suspendSelf(atespace, actorName string) { - apiAddr := os.Getenv("ATE_API_ADDR") - if apiAddr == "" { - apiAddr = "api.ate-system.svc.cluster.local:443" - } - - creds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: true}) - conn, err := grpc.Dial(apiAddr, grpc.WithTransportCredentials(creds)) //nolint:staticcheck // SA1019: TODO migrate to grpc.NewClient. - if err != nil { - log.Printf("Failed to connect to ATE API: %v", err) - return - } - defer conn.Close() - - client := ateapipb.NewControlClient(conn) - - log.Printf("Yielding compute. Requesting self-suspension for actor %s...", actorName) - _, err = client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{Actor: &ateapipb.ObjectRef{Atespace: atespace, Name: actorName}}) - if err != nil { - log.Printf("Failed to self-suspend: %v", err) - } -} diff --git a/hack/install-ate.sh b/hack/install-ate.sh index dbc2f0ad0..c891db35a 100755 --- a/hack/install-ate.sh +++ b/hack/install-ate.sh @@ -42,7 +42,6 @@ ATE_DEMOS=() source "${ROOT}"/hack/install-demo-counter.sh source "${ROOT}"/hack/install-demo-sandbox.sh source "${ROOT}"/hack/install-demo-claude-code-multiplex.sh -source "${ROOT}"/hack/install-demo-agent-secret.sh source "${ROOT}"/hack/install-demo-multi-template.sh # ANSI color codes for prettier output diff --git a/hack/install-demo-agent-secret.sh b/hack/install-demo-agent-secret.sh deleted file mode 100644 index cfaca18d7..000000000 --- a/hack/install-demo-agent-secret.sh +++ /dev/null @@ -1,44 +0,0 @@ -#!/usr/bin/env bash - -# 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. -# -# This is sourced as part of install-ate.sh. Do not run directly. - -ATE_DEMOS+=(demo-agent-secret) # register demo-agent-secret - -demo-agent-secret_cmdline() { - case "${1}" in - --deploy-demo-agent-secret) demo-agent-secret_deploy ;; - --delete-demo-agent-secret) demo-agent-secret_delete ;; - *) - return 1 - ;; - esac - return 0 -} - -demo-agent-secret_deploy() { - log_step "demo-agent-secret_deploy" - ensure_crds - sed "s|\${BUCKET_NAME}|${BUCKET_NAME}|g" demos/agent-secret/agent-secret.yaml.tmpl \ - | run_ko apply -f - -} - -demo-agent-secret_delete() { - log_step "demo-agent-secret_delete" - delete_demo_actors ate-demo-secret-agent-v2 agent-secret - sed "s|\${BUCKET_NAME}|${BUCKET_NAME}|g" demos/agent-secret/agent-secret.yaml.tmpl \ - | run_kubectl delete --ignore-not-found -f - -} diff --git a/internal/ateapiauth/client.go b/internal/ateapiauth/client.go index aed3b5d71..70acb511b 100644 --- a/internal/ateapiauth/client.go +++ b/internal/ateapiauth/client.go @@ -32,6 +32,10 @@ const ( DefaultServiceAccountTokenFile = "/var/run/secrets/kubernetes.io/serviceaccount/token" ) +// roundRobinServiceConfig spreads RPCs across every address the resolver +// returns. +const roundRobinServiceConfig = `{"loadBalancingConfig": [{"round_robin":{}}]}` + // ClientConfig configures how to dial the ateapi gRPC server. The server // cert is always validated against CAFile. UseTokenAuth selects the client // credential: a client certificate from ClientCredBundle (mutual TLS, @@ -87,11 +91,13 @@ func DialOptions(cfg ClientConfig) ([]grpc.DialOption, error) { return []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)), grpc.WithPerRPCCredentials(&fileTokenCreds{path: cfg.TokenFile}), + grpc.WithDefaultServiceConfig(roundRobinServiceConfig), }, nil } tlsCfg.GetClientCertificate = credbundle.ClientLoader(cfg.ClientCredBundle) return []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)), + grpc.WithDefaultServiceConfig(roundRobinServiceConfig), }, nil } diff --git a/internal/ateclient/builder.go b/internal/ateclient/builder.go index ed72a7599..0b31df10a 100644 --- a/internal/ateclient/builder.go +++ b/internal/ateclient/builder.go @@ -19,11 +19,10 @@ import ( "crypto/tls" "crypto/x509" "fmt" - "io" - "net/http" "os" "sync" + "github.com/agent-substrate/substrate/internal/portforward" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "go.opentelemetry.io/otel" @@ -36,12 +35,9 @@ import ( authv1 "k8s.io/api/authentication/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/clientcmd" - "k8s.io/client-go/tools/portforward" - "k8s.io/client-go/transport/spdy" metricsv1beta1 "k8s.io/metrics/pkg/client/clientset/versioned" ) @@ -159,81 +155,15 @@ func dialPortForward(ctx context.Context, kubeconfigPath, k8sContext string, tra return nil, fmt.Errorf("failed to create k8s client: %w", err) } - // Look up the 'api' Service to dynamically get its pod selector - svc, err := clientset.CoreV1().Services("ate-system").Get(ctx, "api", metav1.GetOptions{}) + localPort, stopForward, err := portforward.ServicePortForward(ctx, config, clientset, "ate-system", "api", 443) if err != nil { - return nil, fmt.Errorf("failed to get api service: %w", err) - } - selector := labels.SelectorFromSet(svc.Spec.Selector).String() - - // Find the pods backing the service - pods, err := clientset.CoreV1().Pods("ate-system").List(ctx, metav1.ListOptions{ - LabelSelector: selector, - }) - if err != nil { - return nil, fmt.Errorf("failed to list ateapi pods: %w", err) - } - if len(pods.Items) == 0 { - return nil, fmt.Errorf("no ate-api-server pods found in ate-system namespace") - } - targetPod := pods.Items[0] - - // Setup port-forwarding - req := clientset.CoreV1().RESTClient().Post(). - Resource("pods"). - Namespace(targetPod.Namespace). - Name(targetPod.Name). - SubResource("portforward") - - transport, upgrader, err := spdy.RoundTripperFor(config) - if err != nil { - return nil, fmt.Errorf("failed to create SPDY transport: %w", err) - } - - dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, http.MethodPost, req.URL()) - - stopCh := make(chan struct{}) - readyCh := make(chan struct{}) - - ports := []string{"0:443"} // Port 0 asks OS for a random available local port - - fw, err := portforward.New(dialer, ports, stopCh, readyCh, io.Discard, io.Discard) - if err != nil { - return nil, fmt.Errorf("failed to create port forwarder: %w", err) - } - - errCh := make(chan error, 1) - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - if err := fw.ForwardPorts(); err != nil { - errCh <- fmt.Errorf("port forwarding failed: %w", err) - } - }() - - // Wait for the tunnel to be ready, an error, or context cancellation - select { - case <-readyCh: - // Tunnel is ready! - case err := <-errCh: return nil, err - case <-ctx.Done(): - return nil, ctx.Err() - } - - forwardedPorts, err := fw.GetPorts() - if err != nil || len(forwardedPorts) == 0 { - close(stopCh) - return nil, fmt.Errorf("failed to get forwarded ports: %w", err) } - - localPort := forwardedPorts[0].Local localEndpoint := fmt.Sprintf("127.0.0.1:%d", localPort) tlsCfg, err := serverTLSConfig(ctx, clientset) if err != nil { - close(stopCh) + stopForward() return nil, err } @@ -242,7 +172,7 @@ func dialPortForward(ctx context.Context, kubeconfigPath, k8sContext string, tra opts = append(opts, grpc.WithStatsHandler(otelgrpc.NewClientHandler())) tokenOpt, err := bearerTokenDialOption(ctx, clientset) if err != nil { - close(stopCh) + stopForward() return nil, err } opts = append(opts, tokenOpt) @@ -253,7 +183,7 @@ func dialPortForward(ctx context.Context, kubeconfigPath, k8sContext string, tra conn, err := grpc.NewClient(localEndpoint, opts...) if err != nil { - close(stopCh) + stopForward() return nil, fmt.Errorf("failed to dial gRPC over tunnel: %w", err) } @@ -261,10 +191,7 @@ func dialPortForward(ctx context.Context, kubeconfigPath, k8sContext string, tra ControlClient: ateapipb.NewControlClient(conn), DebugClient: ateapipb.NewDebugClient(conn), conn: conn, - cancel: func() { - close(stopCh) - wg.Wait() - }, + cancel: stopForward, }, nil } diff --git a/internal/benchmarking/boomer/glutton/grpcclient.go b/internal/benchmarking/boomer/glutton/grpcclient.go index b1a139cef..a99923ed8 100644 --- a/internal/benchmarking/boomer/glutton/grpcclient.go +++ b/internal/benchmarking/boomer/glutton/grpcclient.go @@ -33,6 +33,7 @@ func DialControl(endpoint string) (*grpc.ClientConn, ateapipb.ControlClient, err conn, err := grpc.NewClient(endpoint, grpc.WithTransportCredentials(creds), grpc.WithStatsHandler(otelgrpc.NewClientHandler()), + grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`), ) if err != nil { return nil, nil, fmt.Errorf("dial %s: %w", endpoint, err) diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go index 6249631a5..14d0cfe5b 100644 --- a/internal/e2e/collector_metrics.go +++ b/internal/e2e/collector_metrics.go @@ -23,6 +23,7 @@ import ( "time" "github.com/agent-substrate/substrate/internal/ateclient" + "github.com/agent-substrate/substrate/internal/portforward" "k8s.io/client-go/kubernetes" ) @@ -54,11 +55,7 @@ func ScrapeCollectorMetrics(ctx context.Context) (string, error) { return "", fmt.Errorf("creating k8s client: %w", err) } - pod, _, err := firstReadyPodForService(ctx, clientset, collectorNamespace, collectorService) - if err != nil { - return "", err - } - localPort, stop, err := podPortForward(ctx, config, clientset, collectorNamespace, pod.Name, collectorPromPort) + localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, collectorNamespace, collectorService, collectorPromPort) if err != nil { return "", err } diff --git a/internal/e2e/router_client.go b/internal/e2e/router_client.go index 0f16457d3..3c15405fc 100644 --- a/internal/e2e/router_client.go +++ b/internal/e2e/router_client.go @@ -21,9 +21,8 @@ import ( "time" "github.com/agent-substrate/substrate/internal/ateclient" + "github.com/agent-substrate/substrate/internal/portforward" "github.com/agent-substrate/substrate/internal/resources" - corev1 "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/kubernetes" ) @@ -54,20 +53,7 @@ func NewRouterClient(ctx context.Context) (*RouterClient, error) { return nil, fmt.Errorf("creating k8s client: %w", err) } - targetPod, svc, err := firstReadyPodForService(ctx, clientset, routerNamespace, routerService) - if err != nil { - return nil, err - } - - // Port-forward targets a pod's container port, so resolve the Service's - // HTTP port (80) to its backing targetPort (kubectl does this for us when - // forwarding a Service, but we forward the pod directly). - targetPort, err := resolveHTTPTargetPort(svc, targetPod) - if err != nil { - return nil, err - } - - localPort, stop, err := podPortForward(ctx, config, clientset, routerNamespace, targetPod.Name, targetPort) + localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, routerNamespace, routerService, 80) if err != nil { return nil, err } @@ -79,53 +65,6 @@ func NewRouterClient(ctx context.Context) (*RouterClient, error) { }, nil } -// isPodReady reports whether the pod is Running, not terminating, and has -// passed its readiness probe — i.e. actually serving, the same bar the Service -// uses to select endpoints. -func isPodReady(pod *corev1.Pod) bool { - if pod.Status.Phase != corev1.PodRunning || pod.DeletionTimestamp != nil { - return false - } - for _, c := range pod.Status.Conditions { - if c.Type == corev1.PodReady { - return c.Status == corev1.ConditionTrue - } - } - return false -} - -// resolveHTTPTargetPort maps the router Service's HTTP port (80) to the -// container port it targets on the given pod, resolving named targetPorts. -func resolveHTTPTargetPort(svc *corev1.Service, pod *corev1.Pod) (int32, error) { - for _, sp := range svc.Spec.Ports { - if sp.Port != 80 { - continue - } - var port int32 - switch sp.TargetPort.Type { - case intstr.Int: - port = sp.TargetPort.IntVal - case intstr.String: - for _, c := range pod.Spec.Containers { - for _, cp := range c.Ports { - if cp.Name == sp.TargetPort.StrVal { - port = cp.ContainerPort - } - } - } - if port == 0 { - return 0, fmt.Errorf("named targetPort %q not found on pod %s", sp.TargetPort.StrVal, pod.Name) - } - } - // Guard against an unset/zero targetPort, which would forward to nothing. - if port <= 0 { - return 0, fmt.Errorf("service %s port 80 has no usable targetPort", svc.Name) - } - return port, nil - } - return 0, fmt.Errorf("service %s has no port 80", svc.Name) -} - // Close stops the port-forward tunnel. func (c *RouterClient) Close() { c.stop() diff --git a/internal/e2e/portforward.go b/internal/portforward/portforward.go similarity index 55% rename from internal/e2e/portforward.go rename to internal/portforward/portforward.go index 50aa01073..8f7316293 100644 --- a/internal/e2e/portforward.go +++ b/internal/portforward/portforward.go @@ -12,7 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -package e2e +// Package portforward tunnels to the pods behind a Service. The +// Kubernetes port-forward API is pod-scoped, so "forwarding to a +// Service" means resolving its selector to one ready pod and its port +// mapping to a container port, then forwarding to that pod. +// This is the same thing kubectl does for `port-forward svc/`. +package portforward import ( "context" @@ -23,15 +28,31 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/portforward" "k8s.io/client-go/transport/spdy" ) -// firstReadyPodForService returns a ready pod backing service, plus the Service -// itself (callers need its port mapping). It refuses a selectorless Service -// rather than forward to an arbitrary pod in the namespace. +// ServicePortForward forwards a random local port to servicePort of the +// named Service. It picks a ready backing pod, resolves servicePort to +// the pod's container port (including named targetPorts), and tunnels +// to that pod. It returns the chosen local port and a stop func the +// caller must invoke to tear the tunnel down. The tunnel is pinned to +// the picked pod; it does not fail over if that pod later dies. +func ServicePortForward(ctx context.Context, config *rest.Config, clientset kubernetes.Interface, namespace, service string, servicePort int32) (int, func(), error) { + pod, svc, err := firstReadyPodForService(ctx, clientset, namespace, service) + if err != nil { + return 0, nil, err + } + targetPort, err := resolveTargetPort(svc, pod, servicePort) + if err != nil { + return 0, nil, err + } + return podPortForward(ctx, config, clientset, namespace, pod.Name, targetPort) +} + func firstReadyPodForService(ctx context.Context, clientset kubernetes.Interface, namespace, service string) (*corev1.Pod, *corev1.Service, error) { svc, err := clientset.CoreV1().Services(namespace).Get(ctx, service, metav1.GetOptions{}) if err != nil { @@ -46,16 +67,43 @@ func firstReadyPodForService(ctx context.Context, clientset kubernetes.Interface return nil, nil, fmt.Errorf("listing %s pods: %w", service, err) } for i := range pods.Items { - if isPodReady(&pods.Items[i]) { + if IsPodReady(&pods.Items[i]) { return &pods.Items[i], svc, nil } } return nil, nil, fmt.Errorf("no ready %s pods in %s", service, namespace) } -// podPortForward forwards a random local port to targetPort on the pod (local -// port 0 asks the OS for a free one), returning the chosen local port and a stop -// func the caller must invoke to tear the tunnel down. +func resolveTargetPort(svc *corev1.Service, pod *corev1.Pod, servicePort int32) (int32, error) { + for _, sp := range svc.Spec.Ports { + if sp.Port != servicePort { + continue + } + var port int32 + switch sp.TargetPort.Type { + case intstr.Int: + port = sp.TargetPort.IntVal + if port == 0 { + // targetPort defaults to port when omitted. + port = sp.Port + } + case intstr.String: + for _, c := range pod.Spec.Containers { + for _, cp := range c.Ports { + if cp.Name == sp.TargetPort.StrVal { + port = cp.ContainerPort + } + } + } + if port == 0 { + return 0, fmt.Errorf("named targetPort %q not found on pod %s", sp.TargetPort.StrVal, pod.Name) + } + } + return port, nil + } + return 0, fmt.Errorf("service %s has no port %d", svc.Name, servicePort) +} + func podPortForward(ctx context.Context, config *rest.Config, clientset kubernetes.Interface, namespace, podName string, targetPort int32) (int, func(), error) { req := clientset.CoreV1().RESTClient().Post(). Resource("pods"). @@ -98,3 +146,15 @@ func podPortForward(ctx context.Context, config *rest.Config, clientset kubernet } return int(ports[0].Local), func() { close(stopCh) }, nil } + +func IsPodReady(pod *corev1.Pod) bool { + if pod.Status.Phase != corev1.PodRunning || pod.DeletionTimestamp != nil { + return false + } + for _, c := range pod.Status.Conditions { + if c.Type == corev1.PodReady { + return c.Status == corev1.ConditionTrue + } + } + return false +} diff --git a/internal/e2e/router_client_test.go b/internal/portforward/portforward_test.go similarity index 88% rename from internal/e2e/router_client_test.go rename to internal/portforward/portforward_test.go index 632534515..a4b20e58c 100644 --- a/internal/e2e/router_client_test.go +++ b/internal/portforward/portforward_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package e2e +package portforward import ( "testing" @@ -22,7 +22,7 @@ import ( "k8s.io/apimachinery/pkg/util/intstr" ) -func TestResolveHTTPTargetPort(t *testing.T) { +func TestResolveTargetPort(t *testing.T) { pod := &corev1.Pod{ Spec: corev1.PodSpec{ Containers: []corev1.Container{{ @@ -59,12 +59,12 @@ func TestResolveHTTPTargetPort(t *testing.T) { {"named target port", svc(corev1.ServicePort{Port: 80, TargetPort: intstr.FromString("http")}), pod, 8080, false}, {"named target not found", svc(corev1.ServicePort{Port: 80, TargetPort: intstr.FromString("nope")}), pod, 0, true}, {"named target resolves to zero", svc(corev1.ServicePort{Port: 80, TargetPort: intstr.FromString("http")}), zeroPortPod, 0, true}, - {"unset target port", svc(corev1.ServicePort{Port: 80}), pod, 0, true}, - {"no port 80", svc(corev1.ServicePort{Port: 443, TargetPort: intstr.FromInt(8443)}), pod, 0, true}, + {"unset target port defaults to port", svc(corev1.ServicePort{Port: 80}), pod, 80, false}, + {"no matching service port", svc(corev1.ServicePort{Port: 443, TargetPort: intstr.FromInt(8443)}), pod, 0, true}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - got, err := resolveHTTPTargetPort(tc.svc, tc.pod) + got, err := resolveTargetPort(tc.svc, tc.pod, 80) if (err != nil) != tc.wantErr { t.Fatalf("err = %v, wantErr %v", err, tc.wantErr) } @@ -101,8 +101,8 @@ func TestIsPodReady(t *testing.T) { } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - if got := isPodReady(tc.pod); got != tc.want { - t.Errorf("isPodReady = %v, want %v", got, tc.want) + if got := IsPodReady(tc.pod); got != tc.want { + t.Errorf("IsPodReady = %v, want %v", got, tc.want) } }) } diff --git a/internal/volume/mock.go b/internal/volume/mock.go index 7dfb2b3f5..4846599e0 100644 --- a/internal/volume/mock.go +++ b/internal/volume/mock.go @@ -20,7 +20,6 @@ import ( "log/slog" "os" "path/filepath" - "sync" "github.com/agent-substrate/substrate/internal/ateompath" ) @@ -39,88 +38,48 @@ var ( // is scheduled to the same host. // // This plugin also does not cleanup the subdirectories, so that has to be done by the test infrastructure. -type MockVolumePlugin struct { - mu sync.Mutex - volumes map[string]*MockVolumeState - counter int -} - -// MockVolumeState tracks the state of a mock volume. -type MockVolumeState struct { - ID string - Name string - Capacity string - StorageClass string - Node string - Mounts map[string]bool // targetPath -> mounted -} +// +// The control-plane methods (Create/Delete/Attach/DetachVolume) are +// intentionally stateless no-ops: they log and succeed, with no bookkeeping +// of which volumes exist or which node they're attached to. Nothing reads +// that state - the e2e test that exercises this plugin verifies the volume +// actually worked by checking the real file MountVolume writes on the node, +// not by querying the plugin. A stateful mock used to track this in an +// in-process map, which broke once ate-api-server ran multiple replicas +// (whichever replica handled a given RPC had no idea what a different +// replica's map contained); going stateless removes the bug class instead +// of syncing the state, since nothing actually needs it. +type MockVolumePlugin struct{} // NewMockVolumePlugin creates a new MockVolumePlugin. func NewMockVolumePlugin() *MockVolumePlugin { - return &MockVolumePlugin{ - volumes: make(map[string]*MockVolumeState), - } + return &MockVolumePlugin{} } -// CreateVolume simulates volume provisioning. +// CreateVolume simulates volume provisioning. The returned volumeID is +// derived deterministically from name so repeated calls (retries) are +// idempotent by construction, with no state needed to enforce that. func (p *MockVolumePlugin) CreateVolume(ctx context.Context, name string, capacity string, storageClass string) (string, error) { - p.mu.Lock() - defer p.mu.Unlock() - p.counter++ - volumeID := fmt.Sprintf("mock-vol-%d", p.counter) + volumeID := "mock-vol-" + name slog.InfoContext(ctx, "MockVolumePlugin.CreateVolume", slog.String("name", name), slog.String("capacity", capacity), slog.String("storageClass", storageClass), slog.String("volumeID", volumeID)) - p.volumes[volumeID] = &MockVolumeState{ - ID: volumeID, - Name: name, - Capacity: capacity, - StorageClass: storageClass, - Mounts: make(map[string]bool), - } return volumeID, nil } // DeleteVolume simulates volume deletion. func (p *MockVolumePlugin) DeleteVolume(ctx context.Context, volumeID string) error { - p.mu.Lock() - defer p.mu.Unlock() slog.InfoContext(ctx, "MockVolumePlugin.DeleteVolume", slog.String("volumeID", volumeID)) - if _, ok := p.volumes[volumeID]; !ok { - slog.ErrorContext(ctx, "MockVolumePlugin.DeleteVolume failed: volume not found", slog.String("volumeID", volumeID)) - return fmt.Errorf("volume %s not found", volumeID) - } - delete(p.volumes, volumeID) return nil } // AttachVolume simulates volume attachment to a node. func (p *MockVolumePlugin) AttachVolume(ctx context.Context, volumeID string, node string) error { - p.mu.Lock() - defer p.mu.Unlock() slog.InfoContext(ctx, "MockVolumePlugin.AttachVolume", slog.String("volumeID", volumeID), slog.String("node", node)) - vol, ok := p.volumes[volumeID] - if !ok { - slog.ErrorContext(ctx, "MockVolumePlugin.AttachVolume failed: volume not found", slog.String("volumeID", volumeID)) - return fmt.Errorf("volume %s not found", volumeID) - } - vol.Node = node return nil } // DetachVolume simulates volume detachment from a node. func (p *MockVolumePlugin) DetachVolume(ctx context.Context, volumeID string, node string) error { - p.mu.Lock() - defer p.mu.Unlock() slog.InfoContext(ctx, "MockVolumePlugin.DetachVolume", slog.String("volumeID", volumeID), slog.String("node", node)) - vol, ok := p.volumes[volumeID] - if !ok { - slog.ErrorContext(ctx, "MockVolumePlugin.DetachVolume failed: volume not found", slog.String("volumeID", volumeID)) - return fmt.Errorf("volume %s not found", volumeID) - } - if vol.Node != node { - slog.ErrorContext(ctx, "MockVolumePlugin.DetachVolume failed: volume not attached to node", slog.String("volumeID", volumeID), slog.String("node", node), slog.String("attachedNode", vol.Node)) - return fmt.Errorf("volume %s not attached to node %s", volumeID, node) - } - vol.Node = "" return nil } @@ -158,29 +117,3 @@ func (p *MockVolumePlugin) UnmountVolume(ctx context.Context, volumeID string, t } return nil } - -// GetVolumeState returns the state of a mock volume for verification in tests. -// This only works for controller methods. -func (p *MockVolumePlugin) GetVolumeState(volumeID string) (*MockVolumeState, error) { - p.mu.Lock() - defer p.mu.Unlock() - slog.Info("MockVolumePlugin.GetVolumeState", slog.String("volumeID", volumeID)) - vol, ok := p.volumes[volumeID] - if !ok { - slog.Error("MockVolumePlugin.GetVolumeState failed: volume not found", slog.String("volumeID", volumeID)) - return nil, fmt.Errorf("volume %s not found", volumeID) - } - // Return a copy to avoid concurrent access issues in tests - mountsCopy := make(map[string]bool) - for k, v := range vol.Mounts { - mountsCopy[k] = v - } - return &MockVolumeState{ - ID: vol.ID, - Name: vol.Name, - Capacity: vol.Capacity, - StorageClass: vol.StorageClass, - Node: vol.Node, - Mounts: mountsCopy, - }, nil -} diff --git a/manifests/ate-install/ate-api-server.yaml b/manifests/ate-install/ate-api-server.yaml index ef2d863af..9df053eff 100644 --- a/manifests/ate-install/ate-api-server.yaml +++ b/manifests/ate-install/ate-api-server.yaml @@ -62,7 +62,12 @@ metadata: name: ate-api-server namespace: ate-system spec: - replicas: 1 + replicas: 2 + strategy: + # Update replicas one at a time, create a new one first, then delete the old one. + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 selector: matchLabels: app: ate-api-server @@ -208,6 +213,19 @@ spec: - key: "pool" path: "pool.json" --- +# Allow voluntary disruptions (node drains, cluster upgrades) to take +# at most one API server pod at a time, at any replica count. +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: ate-api-server + namespace: ate-system +spec: + maxUnavailable: 1 + selector: + matchLabels: + app: ate-api-server +--- # 6. Expose the Session Assigner apiVersion: v1 kind: Service @@ -215,7 +233,9 @@ metadata: name: api namespace: ate-system spec: - type: ClusterIP + # Headless (i.e. `clusterIP: None`) services make DNS return one A record per + # ready pod so gRPC clients can load balance client-side. + clusterIP: None selector: app: ate-api-server ports: diff --git a/manifests/ate-install/atenet-router.yaml b/manifests/ate-install/atenet-router.yaml index ad6f9dfd6..8fe146561 100644 --- a/manifests/ate-install/atenet-router.yaml +++ b/manifests/ate-install/atenet-router.yaml @@ -137,7 +137,7 @@ spec: - "--otlp-collector-address=opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317" # Client auth to ateapi (mtls): verify the serving cert against the # servicedns trust bundle and present the podidentity client cert. - - "--ateapi-address=api.ate-system.svc:443" + - "--ateapi-address=dns:///api.ate-system.svc:443" - "--ateapi-ca-file=/run/servicedns-ca/trust-bundle.pem" - "--ateapi-client-cert=/run/podidentity.podcert.ate.dev/credential-bundle.pem" env: