Merge remote-tracking branch 'upstream/main'

# Conflicts:
#	cmd/ateapi/internal/controlapi/workflow.go
#	cmd/atenet/internal/router/router.go
#	cmd/atenet/internal/router/xds.go
#	cmd/atenet/internal/router/xds_test.go
#	internal/e2e/suites/demo/demo_test.go
#	manifests/ate-install/ate-controller.yaml
#	manifests/ate-install/atenet-router.yaml

Author: EItanya

benchmarking/automation/manifests/runner-job.yaml.tmpl

@@ -71,6 +71,9 @@ spec:
         - name: servicedns-ca
           mountPath: /run/servicedns-ca
           readOnly: true
+        - name: podidentity
+          mountPath: /run/podidentity.podcert.ate.dev
+          readOnly: true
         resources:
           requests:
             cpu: "500m"
@@ -85,3 +88,10 @@ spec:
                 matchLabels:
                   podcert.ate.dev/canarying: live
               path: ca.crt
+      - name: podidentity
+        projected:
+          sources:
+          - podCertificate:
+              signerName: podidentity.podcert.ate.dev/identity
+              keyType: ECDSAP256
+              credentialBundlePath: credential-bundle.pem

benchmarking/locust/build_and_push.sh

@@ -31,10 +31,13 @@ fi
 
 IMAGE="us-docker.pkg.dev/${PROJECT_ID}/gcr.io/ate-images/locust-test:latest"
 
-echo "Building Docker image: $IMAGE"
+# Target platform must match the cluster's nodes, not the build host.
+PLATFORM="${LOCUST_IMAGE_PLATFORM:-linux/amd64}"
+
+echo "Building Docker image: $IMAGE (platform: $PLATFORM)"
 # Build context is the monorepo root because the Dockerfile compiles the
 # boomer-glutton Go binary alongside the Python install (see Dockerfile).
-docker build -t "$IMAGE" -f benchmarking/locust/Dockerfile .
+docker build --platform "$PLATFORM" -t "$IMAGE" -f benchmarking/locust/Dockerfile .
 
 echo "Pushing Docker image..."
 docker push "$IMAGE"

benchmarking/locust/manifests/locust.yaml

@@ -90,7 +90,12 @@ spec:
         - "-f"
         - "/app/tests"
         - "--worker"
-        - "--master-host=localhost"
+        # Literal IPv4, not "localhost". The master binds 0.0.0.0:5557, which
+        # is IPv4-only; locust sets ZMQ_IPV6 on the worker socket whenever the
+        # master host resolves to anything AF_INET6, and the pod's /etc/hosts
+        # maps localhost to ::1 as well as 127.0.0.1. The worker would then
+        # dial [::1]:5557 and retry forever against a silent master.
+        - "--master-host=127.0.0.1"
         env:
         - name: OTEL_EXPORTER_OTLP_ENDPOINT
           value: http://opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317

charts/substrate-crds/templates/ate.dev_workerpools.yaml

@@ -411,6 +411,17 @@ spec:
                     type: array
                     x-kubernetes-list-type: atomic
                 type: object
+              terminationGracePeriodSeconds:
+                default: 300
+                description: |-
+                  TerminationGracePeriodSeconds is the termination grace period applied to
+                  this pool's worker pods. On eviction, ateom traps SIGTERM and forwards it
+                  to the actor so it can save state and exit cleanly before the kubelet
+                  sends SIGKILL. Tune this to the maximum time your actors need to shut
+                  down gracefully. Defaults to 300 (5 minutes).
+                format: int32
+                minimum: 1
+                type: integer
             required:
             - ateomImage
             - replicas

charts/substrate/templates/ate-controller.yaml

@@ -85,6 +85,11 @@ spec:
         - "--ateapi-ca-file=/run/ateapi-ca/ca.crt"
         - "--ateapi-server-name={{ include "substrate.fullname" (list "api" .) }}.{{ .Release.Namespace }}.svc"
         - "--ateapi-token-file=/var/run/secrets/tokens/ateapi/token"
+{{- end }}
+{{- if .Values.otel.endpoint }}
+        env:
+        - name: OTEL_EXPORTER_OTLP_ENDPOINT
+          value: {{ .Values.otel.endpoint | quote }}
 {{- end }}
         ports:
         - name: metrics

charts/substrate/templates/atenet-router.yaml

@@ -294,3 +294,7 @@ spec:
     port: 443
     targetPort: 8443
     protocol: TCP
+  - name: status
+    port: 4040
+    targetPort: status
+    protocol: TCP

charts/substrate/templates/role.yaml

@@ -80,3 +80,15 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - networking.k8s.io
+  resources:
+  - networkpolicies
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch

cmd/ateapi/internal/controlapi/crash.go

@@ -22,33 +22,34 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/internal/ateerrors"
+	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
 // maybeCrashActor inspects err returned by an atelet RPC, it crashes
 // the actor if the err carries the actorCrashed=true metadata directive.
-func maybeCrashActor(ctx context.Context, st store.Interface, atespace, actorName string, err error, wrapMsg string) error {
+func maybeCrashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef, err error, wrapMsg string) error {
 	if err == nil {
 		return nil
 	}
 
 	if ateerrors.ActorCrashRequested(err) {
 		slog.ErrorContext(ctx, "Setting Actor to crashed due to error", slog.Any("error", err))
-		if cerr := crashActor(ctx, st, atespace, actorName); cerr != nil {
+		if cerr := crashActor(ctx, st, actorRef); cerr != nil {
 			slog.ErrorContext(ctx, "Failed to crash actor", slog.Any("cerr", cerr))
 			return cerr
 		}
-		return status.Errorf(codes.DataLoss, "actor %s crashed", actorName)
+		return status.Errorf(codes.DataLoss, "actor %s crashed", actorRef)
 	}
 	return fmt.Errorf("%s: %w", wrapMsg, err)
 }
 
 // crashActor moves the actor to CRASHED state and frees the worker it was
 // assigned to, if any, so the worker can host other actors.
-func crashActor(ctx context.Context, st store.Interface, atespace, actorName string) error {
-	actor, err := st.GetActor(ctx, atespace, actorName)
+func crashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef) error {
+	actor, err := st.GetActor(ctx, actorRef)
 	if err != nil {
 		return fmt.Errorf("while loading actor to crash: %w", err)
 	}
@@ -102,7 +103,7 @@ func releaseWorker(ctx context.Context, st store.Interface, actor *ateapipb.Acto
 		return nil
 	}
 	// Only free it if it still belongs to us
-	if wass.GetActor().GetAtespace() != actor.GetMetadata().GetAtespace() || wass.GetActor().GetName() != actor.GetMetadata().GetName() {
+	if resources.ActorRefFromObjectRef(wass.GetActor()) != resources.ActorRefFromActor(actor) {
 		slog.WarnContext(ctx, "Worker already assigned to another Actor", slog.String("worker", podUid))
 		return nil
 	}

cmd/ateapi/internal/controlapi/crash_test.go

@@ -23,17 +23,18 @@ import (
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
 	"github.com/agent-substrate/substrate/internal/ateerrors"
+	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
 // seedActor stores a running actor with all worker-binding fields populated, so
 // tests can assert they are cleared when the actor crashes.
-func seedActor(t *testing.T, ctx context.Context, st store.Interface, atespace, actorName string) {
+func seedActor(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) {
 	t.Helper()
 	if _, err := st.CreateActor(ctx, &ateapipb.Actor{
-		Metadata:           &ateapipb.ResourceMetadata{Name: actorName, Atespace: atespace},
+		Metadata:           &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace},
 		Status:             ateapipb.Actor_STATUS_RUNNING,
 		AteomPodNamespace:  "ns",
 		AteomPodName:       "pod",
@@ -47,17 +48,17 @@ func seedActor(t *testing.T, ctx context.Context, st store.Interface, atespace,
 }
 
 // seedWorker registers the worker referenced by seedActor's binding fields,
-// assigned to the given actor in atespace (unassigned if assignedActor is "").
-func seedWorker(t *testing.T, ctx context.Context, st store.Interface, atespace, assignedActor string) {
+// assigned to the given actor (unassigned if assigned is the zero ActorRef).
+func seedWorker(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) {
 	t.Helper()
 	worker := &ateapipb.Worker{
 		WorkerNamespace: "ns",
 		WorkerPool:      "pool",
 		WorkerPod:       "pod",
 	}
-	if assignedActor != "" {
+	if actorRef != (resources.ActorRef{}) {
 		worker.Assignment = &ateapipb.Assignment{
-			Actor: &ateapipb.ObjectRef{Atespace: atespace, Name: assignedActor},
+			Actor: actorRef.ToObjectRef(),
 		}
 	}
 	if err := st.CreateWorker(ctx, worker); err != nil {
@@ -67,10 +68,10 @@ func seedWorker(t *testing.T, ctx context.Context, st store.Interface, atespace,
 
 // seedUnboundActor stores a running actor whose worker-binding fields were
 // already cleared, e.g. by a prior release.
-func seedUnboundActor(t *testing.T, ctx context.Context, st store.Interface, atespace, actorName string) {
+func seedUnboundActor(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) {
 	t.Helper()
 	if _, err := st.CreateActor(ctx, &ateapipb.Actor{
-		Metadata:           &ateapipb.ResourceMetadata{Name: actorName, Atespace: atespace},
+		Metadata:           &ateapipb.ResourceMetadata{Name: actorRef.Name, Atespace: actorRef.Atespace},
 		Status:             ateapipb.Actor_STATUS_RUNNING,
 		InProgressSnapshot: "gs://snapshots/actor-1/reserved",
 	}); err != nil {
@@ -80,11 +81,11 @@ func seedUnboundActor(t *testing.T, ctx context.Context, st store.Interface, ate
 
 // assertCrashed reloads the actor and verifies it is CRASHED with its worker
 // binding cleared.
-func assertCrashed(t *testing.T, ctx context.Context, st store.Interface, atespace, actorName string) {
+func assertCrashed(t *testing.T, ctx context.Context, st store.Interface, actorRef resources.ActorRef) {
 	t.Helper()
-	got, err := st.GetActor(ctx, atespace, actorName)
+	got, err := st.GetActor(ctx, actorRef)
 	if err != nil {
-		t.Fatalf("GetActor(%q, %q) = %v, want nil", atespace, actorName, err)
+		t.Fatalf("GetActor(%v) = %v, want nil", actorRef, err)
 	}
 	if got.GetStatus() != ateapipb.Actor_STATUS_CRASHED {
 		t.Errorf("status = %v, want %v", got.GetStatus(), ateapipb.Actor_STATUS_CRASHED)
@@ -107,10 +108,7 @@ func assertCrashed(t *testing.T, ctx context.Context, st store.Interface, atespa
 }
 
 func TestCrashActor(t *testing.T) {
-	const (
-		atespace  = "team-a"
-		actorName = "actor-1"
-	)
+	actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"}
 
 	tests := []struct {
 		name string
@@ -127,20 +125,20 @@ func TestCrashActor(t *testing.T) {
 				if err != nil {
 					t.Fatalf("crashActor() = %v, want nil", err)
 				}
-				assertCrashed(t, ctx, st, atespace, actorName)
+				assertCrashed(t, ctx, st, actorRef)
 			},
 		},
 		{
 			name: "releases worker assigned to crashed actor",
 			seed: true,
 			setup: func(t *testing.T, ctx context.Context, st store.Interface) {
-				seedWorker(t, ctx, st, atespace, actorName)
+				seedWorker(t, ctx, st, actorRef)
 			},
 			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
 				if err != nil {
 					t.Fatalf("crashActor() = %v, want nil", err)
 				}
-				assertCrashed(t, ctx, st, atespace, actorName)
+				assertCrashed(t, ctx, st, actorRef)
 				worker, gerr := st.GetWorker(ctx, "ns", "pool", "pod")
 				if gerr != nil {
 					t.Fatalf("GetWorker() = %v, want nil", gerr)
@@ -154,13 +152,13 @@ func TestCrashActor(t *testing.T) {
 			name: "keeps worker assigned to another actor",
 			seed: true,
 			setup: func(t *testing.T, ctx context.Context, st store.Interface) {
-				seedWorker(t, ctx, st, atespace, "actor-2")
+				seedWorker(t, ctx, st, resources.ActorRef{Atespace: actorRef.Atespace, Name: "actor-2"})
 			},
 			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
 				if err != nil {
 					t.Fatalf("crashActor() = %v, want nil", err)
 				}
-				assertCrashed(t, ctx, st, atespace, actorName)
+				assertCrashed(t, ctx, st, actorRef)
 				worker, gerr := st.GetWorker(ctx, "ns", "pool", "pod")
 				if gerr != nil {
 					t.Fatalf("GetWorker() = %v, want nil", gerr)
@@ -174,14 +172,14 @@ func TestCrashActor(t *testing.T) {
 			name: "skips release for actor with no worker binding",
 			seed: false,
 			setup: func(t *testing.T, ctx context.Context, st store.Interface) {
-				seedUnboundActor(t, ctx, st, atespace, actorName)
-				seedWorker(t, ctx, st, atespace, actorName)
+				seedUnboundActor(t, ctx, st, actorRef)
+				seedWorker(t, ctx, st, actorRef)
 			},
 			check: func(t *testing.T, ctx context.Context, st store.Interface, err error) {
 				if err != nil {
 					t.Fatalf("crashActor() = %v, want nil", err)
 				}
-				assertCrashed(t, ctx, st, atespace, actorName)
+				assertCrashed(t, ctx, st, actorRef)
 				// Without a binding the worker cannot be looked up, so its
 				// assignment must be left untouched even though it names
 				// the crashed actor.
@@ -218,24 +216,21 @@ func TestCrashActor(t *testing.T) {
 			defer cleanup()
 
 			if tt.seed {
-				seedActor(t, ctx, st, atespace, actorName)
+				seedActor(t, ctx, st, actorRef)
 			}
 			if tt.setup != nil {
 				tt.setup(t, ctx, st)
 			}
 
-			err := crashActor(ctx, st, atespace, actorName)
+			err := crashActor(ctx, st, actorRef)
 			tt.check(t, ctx, st, err)
 		})
 	}
 }
 
 func TestMaybeCrashActor(t *testing.T) {
-	const (
-		atespace  = "team-a"
-		actorName = "actor-1"
-		wrapMsg   = "calling atelet"
-	)
+	const wrapMsg = "calling atelet"
+	actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-1"}
 
 	crashErr := ateerrors.NewGRPCError(context.Background(), codes.NotFound, ateerrors.ReasonTerminalFileSystemError, ateerrors.ActorCrashedMetadata(), errors.New("boom"))
 	// A structured error carrying a reason but no actorCrashed directive must be
@@ -271,7 +266,7 @@ func TestMaybeCrashActor(t *testing.T) {
 				if got := status.Code(err); got != codes.DataLoss {
 					t.Errorf("status code = %v, want %v", got, codes.DataLoss)
 				}
-				assertCrashed(t, ctx, st, atespace, actorName)
+				assertCrashed(t, ctx, st, actorRef)
 			},
 		},
 		{
@@ -305,7 +300,7 @@ func TestMaybeCrashActor(t *testing.T) {
 					t.Errorf("maybeCrashActor() error = %q, want prefix %q", err, wrapMsg)
 				}
 				// The actor must not have been crashed.
-				got, gerr := st.GetActor(ctx, atespace, actorName)
+				got, gerr := st.GetActor(ctx, actorRef)
 				if gerr != nil {
 					t.Fatalf("GetActor() = %v, want nil", gerr)
 				}
@@ -329,7 +324,7 @@ func TestMaybeCrashActor(t *testing.T) {
 					t.Errorf("maybeCrashActor() error = %q, want prefix %q", err, wrapMsg)
 				}
 				// The actor must not have been crashed.
-				got, gerr := st.GetActor(ctx, atespace, actorName)
+				got, gerr := st.GetActor(ctx, actorRef)
 				if gerr != nil {
 					t.Fatalf("GetActor() = %v, want nil", gerr)
 				}
@@ -347,10 +342,10 @@ func TestMaybeCrashActor(t *testing.T) {
 			defer cleanup()
 
 			if tt.seed {
-				seedActor(t, ctx, st, atespace, actorName)
+				seedActor(t, ctx, st, actorRef)
 			}
 
-			err := maybeCrashActor(ctx, st, atespace, actorName, tt.err, wrapMsg)
+			err := maybeCrashActor(ctx, st, actorRef, tt.err, wrapMsg)
 			tt.check(t, ctx, st, err)
 		})
 	}