ateapi: classify atelet churn as Unavailable

cmd/ateapi/internal/controlapi/dialer.go

@@ -31,7 +31,9 @@ import (
 	"github.com/spiffe/go-spiffe/v2/svid/x509svid"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/lru"
@@ -45,6 +47,16 @@ var ErrWorkerPodNotFound = errors.New("worker pod not found")
 // this one is retryable.
 var ErrNoAteletOnNode = errors.New("no atelet pod found on node")
 
+// unavailableError keeps a sentinel reachable via errors.Is while carrying
+// codes.Unavailable, so the interceptor doesn't default it to Internal.
+type unavailableError struct{ error }
+
+func (e *unavailableError) GRPCStatus() *status.Status {
+	return status.New(codes.Unavailable, e.Error())
+}
+
+func (e *unavailableError) Unwrap() error { return e.error }
+
 // The SPIFFE identity that atelet serving certs carry, as minted by the
 // podidentity signer (cmd/podcertcontroller/internal/podidentitysigner).
 // The namespace part is ateletNamespace, declared in informer.go.
@@ -119,11 +131,13 @@ func (d *AteletDialer) DialForAteletOnNode(nodeName string) (*grpc.ClientConn, e
 		return nil, fmt.Errorf("while finding atelet on node %q: %w", nodeName, err)
 	}
 
+	// Atelet churn self-heals in seconds; Unavailable lets the router park
+	// and retry instead of failing fast.
 	if len(matchingAtelets) == 0 {
-		return nil, fmt.Errorf("%w: %q", ErrNoAteletOnNode, nodeName)
+		return nil, &unavailableError{fmt.Errorf("%w: %q", ErrNoAteletOnNode, nodeName)}
 	}
 	if len(matchingAtelets) > 1 {
-		return nil, fmt.Errorf("found %d atelet pods on node %q, expected 1", len(matchingAtelets), nodeName)
+		return nil, status.Errorf(codes.Unavailable, "found %d atelet pods on node %q, expected 1", len(matchingAtelets), nodeName)
 	}
 
 	selectedAtelet := matchingAtelets[0].(*corev1.Pod)
@@ -135,7 +149,7 @@ func (d *AteletDialer) DialForAteletOnNode(nodeName string) (*grpc.ClientConn, e
 	}
 
 	if len(selectedAtelet.Status.PodIPs) == 0 {
-		return nil, fmt.Errorf("selected atelet %q has no assigned IPs", selectedAtelet.ObjectMeta.Namespace+"/"+selectedAtelet.ObjectMeta.Name)
+		return nil, status.Errorf(codes.Unavailable, "selected atelet %q has no assigned IPs", selectedAtelet.ObjectMeta.Namespace+"/"+selectedAtelet.ObjectMeta.Name)
 	}
 
 	creds, err := d.dialCredentials(string(selectedAtelet.ObjectMeta.UID))

cmd/ateapi/internal/controlapi/dialer_test.go

@@ -22,6 +22,7 @@ import (
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"errors"
+	"fmt"
 	"math/big"
 	"net/url"
 	"testing"
@@ -30,8 +31,10 @@ import (
 	"github.com/agent-substrate/substrate/internal/substratex509"
 	"github.com/spiffe/go-spiffe/v2/bundle/x509bundle"
 	"github.com/spiffe/go-spiffe/v2/spiffeid"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -406,3 +409,129 @@ func TestDialForAteletOnNode(t *testing.T) {
 		}
 	})
 }
+
+func newDialerTestIndexers(t *testing.T, workers, atelets []*corev1.Pod) (cache.Indexer, cache.Indexer) {
+	t.Helper()
+
+	workerIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{
+		byNamespaceAndName: func(obj any) ([]string, error) {
+			pod := obj.(*corev1.Pod)
+			return []string{pod.ObjectMeta.Namespace + "/" + pod.ObjectMeta.Name}, nil
+		},
+	})
+	for _, pod := range workers {
+		if err := workerIndexer.Add(pod); err != nil {
+			t.Fatalf("adding worker pod: %v", err)
+		}
+	}
+
+	ateletIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{
+		byNode: func(obj any) ([]string, error) {
+			pod := obj.(*corev1.Pod)
+			return []string{pod.Spec.NodeName}, nil
+		},
+	})
+	for _, pod := range atelets {
+		if err := ateletIndexer.Add(pod); err != nil {
+			t.Fatalf("adding atelet pod: %v", err)
+		}
+	}
+
+	return workerIndexer, ateletIndexer
+}
+
+func dialerTestWorkerPod() *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "worker-1", UID: "worker-uid"},
+		Spec:       corev1.PodSpec{NodeName: "node-1"},
+	}
+}
+
+func dialerTestAteletPod(name string, ips []corev1.PodIP) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: ateletNamespace, Name: name, UID: types.UID(name + "-uid")},
+		Spec:       corev1.PodSpec{NodeName: "node-1"},
+		Status:     corev1.PodStatus{PodIPs: ips},
+	}
+}
+
+// The router parks and retries only FailedPrecondition, Aborted, and
+// Unavailable (docs/request-parking.md). Transient atelet churn during node
+// upgrades must therefore surface as Unavailable, not the interceptor's
+// Internal default — even after the workflow engine wraps the error.
+func assertUnavailableThroughWrapping(t *testing.T, err error) {
+	t.Helper()
+
+	if err == nil {
+		t.Fatal("DialForWorker returned nil error")
+	}
+	wrapped := fmt.Errorf("workflow failed at step CallAteletSuspend: %w", err)
+	var statusErr interface{ GRPCStatus() *status.Status }
+	if !errors.As(wrapped, &statusErr) {
+		t.Fatalf("error %v carries no gRPC status; the server interceptor will default it to Internal", wrapped)
+	}
+	if code := statusErr.GRPCStatus().Code(); code != codes.Unavailable {
+		t.Fatalf("error code = %v, want %v", code, codes.Unavailable)
+	}
+}
+
+func TestDialForWorkerNoAteletOnNodeIsUnavailable(t *testing.T) {
+	workerIndexer, ateletIndexer := newDialerTestIndexers(t, []*corev1.Pod{dialerTestWorkerPod()}, nil)
+	dialer := NewAteletDialer(workerIndexer, ateletIndexer, "", "")
+
+	_, err := dialer.DialForWorker("default", "worker-1")
+	assertUnavailableThroughWrapping(t, err)
+}
+
+func TestDialForWorkerAteletWithoutIPsIsUnavailable(t *testing.T) {
+	workerIndexer, ateletIndexer := newDialerTestIndexers(t,
+		[]*corev1.Pod{dialerTestWorkerPod()},
+		[]*corev1.Pod{dialerTestAteletPod("atelet-1", nil)})
+	dialer := NewAteletDialer(workerIndexer, ateletIndexer, "", "")
+
+	_, err := dialer.DialForWorker("default", "worker-1")
+	assertUnavailableThroughWrapping(t, err)
+}
+
+func TestDialForWorkerMultipleAteletsOnNodeIsUnavailable(t *testing.T) {
+	workerIndexer, ateletIndexer := newDialerTestIndexers(t,
+		[]*corev1.Pod{dialerTestWorkerPod()},
+		[]*corev1.Pod{
+			dialerTestAteletPod("atelet-old", []corev1.PodIP{{IP: "10.0.0.1"}}),
+			dialerTestAteletPod("atelet-new", []corev1.PodIP{{IP: "10.0.0.2"}}),
+		})
+	dialer := NewAteletDialer(workerIndexer, ateletIndexer, "", "")
+
+	_, err := dialer.DialForWorker("default", "worker-1")
+	assertUnavailableThroughWrapping(t, err)
+}
+
+func TestDialForWorkerWorkerPodNotFoundKeepsSentinel(t *testing.T) {
+	workerIndexer, ateletIndexer := newDialerTestIndexers(t, nil, nil)
+	dialer := NewAteletDialer(workerIndexer, ateletIndexer, "", "")
+
+	_, err := dialer.DialForWorker("default", "worker-1")
+	if !errors.Is(err, ErrWorkerPodNotFound) {
+		t.Fatalf("error = %v, want ErrWorkerPodNotFound", err)
+	}
+	var statusErr interface{ GRPCStatus() *status.Status }
+	if errors.As(err, &statusErr) && statusErr.GRPCStatus().Code() == codes.Unavailable {
+		t.Fatal("ErrWorkerPodNotFound became Unavailable; call sites crash the actor on this sentinel and must keep fail-fast semantics")
+	}
+}
+
+func TestDialForWorkerSucceedsWithHealthyAtelet(t *testing.T) {
+	workerIndexer, ateletIndexer := newDialerTestIndexers(t,
+		[]*corev1.Pod{dialerTestWorkerPod()},
+		[]*corev1.Pod{dialerTestAteletPod("atelet-1", []corev1.PodIP{{IP: "10.0.0.1"}})})
+	dialer := NewAteletDialer(workerIndexer, ateletIndexer, "", "")
+	dialer.dialCredentials = func(string) (credentials.TransportCredentials, error) {
+		return insecure.NewCredentials(), nil
+	}
+
+	conn, err := dialer.DialForWorker("default", "worker-1")
+	if err != nil {
+		t.Fatalf("DialForWorker: %v", err)
+	}
+	defer conn.Close()
+}

cmd/atenet/internal/router/extproc/errors.go

@@ -57,25 +57,34 @@ func WrapReqError(code envoy_type.StatusCode, cause error, format string, args .
 // ImmediateResponse tells the dataplane to answer the request itself, without
 // going upstream.
 func ImmediateResponse(statusCode envoy_type.StatusCode, message string) *extprocv3.ProcessingResponse {
+	headers := []*corev3.HeaderValueOption{
+		{
+			// Using RawValues instead of Value: newer versions of Envoy
+			// drop Value and use RawValue
+			Header: &corev3.HeaderValue{
+				Key:      "content-type",
+				RawValue: []byte("text/plain"),
+			},
+		},
+	}
+	if statusCode == envoy_type.StatusCode_ServiceUnavailable {
+		// 503s here are transient (atelet churn, park budget spent); tell
+		// clients to retry shortly rather than back off.
+		headers = append(headers, &corev3.HeaderValueOption{
+			Header: &corev3.HeaderValue{
+				Key:      "retry-after",
+				RawValue: []byte("1"),
+			},
+		})
+	}
 	return &extprocv3.ProcessingResponse{
 		Response: &extprocv3.ProcessingResponse_ImmediateResponse{
 			ImmediateResponse: &extprocv3.ImmediateResponse{
 				Status: &envoy_type.HttpStatus{
 					Code: statusCode,
 				},
-				Body: []byte(message),
-				Headers: &extprocv3.HeaderMutation{
-					SetHeaders: []*corev3.HeaderValueOption{
-						{
-							// Using RawValues instead of Value: newer versions of Envoy
-							// drop Value and use RawValue
-							Header: &corev3.HeaderValue{
-								Key:      "content-type",
-								RawValue: []byte("text/plain"),
-							},
-						},
-					},
-				},
+				Body:    []byte(message),
+				Headers: &extprocv3.HeaderMutation{SetHeaders: headers},
 			},
 		},
 	}

cmd/atenet/internal/router/extproc/errors_test.go

@@ -47,7 +47,7 @@ func TestNewReqError(t *testing.T) {
 func TestImmediateResponseHeaderEncoding(t *testing.T) {
 	t.Parallel()
 
-	resp := ImmediateResponse(envoy_type.StatusCode_ServiceUnavailable, "body")
+	resp := ImmediateResponse(envoy_type.StatusCode_InternalServerError, "body")
 	set := resp.GetImmediateResponse().GetHeaders().GetSetHeaders()
 	if len(set) != 1 {
 		t.Fatalf("SetHeaders count = %d, want 1", len(set))
@@ -60,3 +60,24 @@ func TestImmediateResponseHeaderEncoding(t *testing.T) {
 		t.Errorf("header uses Value (%q); must use RawValue only", h.GetValue())
 	}
 }
+
+func TestImmediateResponseRetryAfter(t *testing.T) {
+	t.Parallel()
+
+	headers := func(code envoy_type.StatusCode) map[string]string {
+		got := map[string]string{}
+		for _, o := range ImmediateResponse(code, "body").GetImmediateResponse().GetHeaders().GetSetHeaders() {
+			got[o.GetHeader().GetKey()] = string(o.GetHeader().GetRawValue())
+		}
+		return got
+	}
+
+	if v, ok := headers(envoy_type.StatusCode_ServiceUnavailable)["retry-after"]; !ok || v != "1" {
+		t.Errorf("503 retry-after = %q, %v; want \"1\", true", v, ok)
+	}
+	for _, code := range []envoy_type.StatusCode{envoy_type.StatusCode_InternalServerError, envoy_type.StatusCode_GatewayTimeout, envoy_type.StatusCode_Forbidden} {
+		if v, ok := headers(code)["retry-after"]; ok {
+			t.Errorf("%d unexpectedly has retry-after %q", code, v)
+		}
+	}
+}