mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-21 19:56:33 +00:00
fix(eks): map transient JetStream errors in ListClusters to ServiceUnavailable (#302)
aws eks list-clusters returned InternalErrorException for a few seconds after a daemon restart, while describe-cluster (by name) succeeded. During the post-restart window the per-account KV stream's Raft group has no leader yet, so kv.Keys() (an ordered consumer over the warming stream) fails with a transient NATS error. ListClusters wrapped it as a fmt.Errorf string the daemon's ValidErrorCode cannot map, so it sanitized to ServerInternal — a misleading 500 that tells AWS clients the request is non-retryable. Classify transient NATS conditions (no-responders, timeout, no-stream-response, connection-closed) and return a retryable ServiceUnavailable (503) instead; genuine faults still pass through wrapped. describe-cluster survived only because kv.Get on the same bucket is cheaper than the consumer kv.Keys spins up.
This commit is contained in:
@@ -1,10 +1,14 @@
|
||||
package handlers_eks
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/eks"
|
||||
"github.com/mulgadc/spinifex/spinifex/awserrors"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
@@ -53,3 +57,26 @@ func TestEKSServiceImpl_ListClustersMaxResultsClamped(t *testing.T) {
|
||||
assert.Nil(t, out.NextToken, "MaxResults=%d single page must not advertise a NextToken", mr)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEKSReadUnavailableClassification pins the post-restart behaviour: a
|
||||
// transient JetStream error (no leader yet) surfaces as a retryable
|
||||
// ServiceUnavailable, while a genuine fault stays wrapped (the daemon maps it
|
||||
// to InternalError). Regression guard for list-clusters returning a misleading
|
||||
// 500 during the KV warmup window.
|
||||
func TestEKSReadUnavailableClassification(t *testing.T) {
|
||||
transient := []error{
|
||||
nats.ErrNoResponders,
|
||||
nats.ErrTimeout,
|
||||
nats.ErrNoStreamResponse,
|
||||
nats.ErrConnectionClosed,
|
||||
fmt.Errorf("kv keys: %w", nats.ErrNoResponders), // wrapped, as ListClusters wraps it
|
||||
}
|
||||
for _, e := range transient {
|
||||
assert.True(t, natsTransient(e), "%v must be classified transient", e)
|
||||
assert.EqualError(t, eksReadUnavailableOr(e, "op"), awserrors.ErrorServiceUnavailable)
|
||||
}
|
||||
|
||||
genuine := errors.New("corrupt cluster meta")
|
||||
assert.False(t, natsTransient(genuine))
|
||||
assert.ErrorContains(t, eksReadUnavailableOr(genuine, "list cluster names"), "list cluster names: corrupt cluster meta")
|
||||
}
|
||||
|
||||
@@ -533,15 +533,15 @@ func (s *EKSServiceImpl) DescribeCluster(input *eks.DescribeClusterInput, accoun
|
||||
func (s *EKSServiceImpl) ListClusters(input *eks.ListClustersInput, accountID string) (*eks.ListClustersOutput, error) {
|
||||
js, err := s.deps.NATSConn.JetStream()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("jetstream: %w", err)
|
||||
return nil, eksReadUnavailableOr(err, "jetstream")
|
||||
}
|
||||
acctKV, err := GetOrCreateAccountBucket(js, accountID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get account bucket: %w", err)
|
||||
return nil, eksReadUnavailableOr(err, "get account bucket")
|
||||
}
|
||||
names, err := listClusterNames(acctKV)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, eksReadUnavailableOr(err, "list cluster names")
|
||||
}
|
||||
sort.Strings(names)
|
||||
|
||||
@@ -1234,6 +1234,30 @@ func clusterMetaToAWS(meta *ClusterMeta) *eks.Cluster {
|
||||
return out
|
||||
}
|
||||
|
||||
// natsTransient reports whether err is a transient NATS/JetStream condition
|
||||
// rather than a genuine fault. These occur in the post-restart window before
|
||||
// the KV stream's Raft group has elected a leader: JetStream requests get no
|
||||
// responder, time out, or the stream is briefly unreachable.
|
||||
func natsTransient(err error) bool {
|
||||
return err != nil && (errors.Is(err, nats.ErrNoResponders) ||
|
||||
errors.Is(err, nats.ErrTimeout) ||
|
||||
errors.Is(err, nats.ErrNoStreamResponse) ||
|
||||
errors.Is(err, nats.ErrConnectionClosed))
|
||||
}
|
||||
|
||||
// eksReadUnavailableOr maps a transient NATS/JetStream failure to a retryable
|
||||
// ServiceUnavailable (503) so AWS clients back off and retry, instead of the
|
||||
// daemon sanitizing an unrecognised wrapped error to a misleading 500
|
||||
// InternalError. Non-transient errors pass through wrapped so real faults stay
|
||||
// visible.
|
||||
func eksReadUnavailableOr(err error, op string) error {
|
||||
if natsTransient(err) {
|
||||
slog.Warn("EKS read temporarily unavailable", "op", op, "err", err)
|
||||
return errors.New(awserrors.ErrorServiceUnavailable)
|
||||
}
|
||||
return fmt.Errorf("%s: %w", op, err)
|
||||
}
|
||||
|
||||
func listClusterNames(kv nats.KeyValue) ([]string, error) {
|
||||
keys, err := kv.Keys()
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user