From 00c5690e47812f181cfd51b07bc6bcf29f594ccb Mon Sep 17 00:00:00 2001 From: NekoPunch Date: Mon, 3 Aug 2026 01:40:46 -0700 Subject: [PATCH] ateapi: classify atelet churn as Unavailable DialForWorker's transient failures (atelet pod being replaced, IP not yet assigned) surfaced as the interceptor's Internal default, telling clients not to retry an error that self-heals in seconds. Unavailable lets the router park and retry; its 503s now carry Retry-After: 1. Worker-pod-side lookups keep their current classification (#600), and a full workflow-step audit is #605's inverse direction. --- cmd/ateapi/internal/controlapi/dialer.go | 20 ++- cmd/ateapi/internal/controlapi/dialer_test.go | 129 ++++++++++++++++++ cmd/atenet/internal/router/extproc/errors.go | 35 +++-- .../internal/router/extproc/errors_test.go | 23 +++- 4 files changed, 190 insertions(+), 17 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/dialer.go b/cmd/ateapi/internal/controlapi/dialer.go index 43d6c4b865..7ea8559b0d 100644 --- a/cmd/ateapi/internal/controlapi/dialer.go +++ b/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. @@ -133,11 +145,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) @@ -149,7 +163,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)) diff --git a/cmd/ateapi/internal/controlapi/dialer_test.go b/cmd/ateapi/internal/controlapi/dialer_test.go index 73b9879f98..cfb3fa162f 100644 --- a/cmd/ateapi/internal/controlapi/dialer_test.go +++ b/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() +} diff --git a/cmd/atenet/internal/router/extproc/errors.go b/cmd/atenet/internal/router/extproc/errors.go index 0a6acb19aa..79edb21e91 100644 --- a/cmd/atenet/internal/router/extproc/errors.go +++ b/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}, }, }, } diff --git a/cmd/atenet/internal/router/extproc/errors_test.go b/cmd/atenet/internal/router/extproc/errors_test.go index ad8da7fe30..ade5080f21 100644 --- a/cmd/atenet/internal/router/extproc/errors_test.go +++ b/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) + } + } +}