diff --git a/cmd/ateapi/internal/controlapi/dialer.go b/cmd/ateapi/internal/controlapi/dialer.go index 43d6c4b86..7ea8559b0 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 73b9879f9..cfb3fa162 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 0a6acb19a..79edb21e9 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 ad8da7fe3..ade5080f2 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) + } + } +}