diff --git a/cmd/adapter/main.go b/cmd/adapter/main.go index 9d5a478b..4e51f21f 100644 --- a/cmd/adapter/main.go +++ b/cmd/adapter/main.go @@ -557,7 +557,7 @@ func runServe(flags *pflag.FlagSet) error { } // Create the event handler and subscribe to broker - handler := executor.AlwaysAck(executor.WithMetrics(exec.CreateHandler(), metricsRecorder, log)) + handler := executor.AlwaysAck(executor.WithMetrics(exec.CreateHandler(), metricsRecorder, log), log) // Handle signals for graceful shutdown sigCh := make(chan os.Signal, 1) diff --git a/go.mod b/go.mod index 811fcacf..7394a1ca 100644 --- a/go.mod +++ b/go.mod @@ -27,11 +27,12 @@ require ( go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/trace v1.44.0 golang.org/x/text v0.39.0 + google.golang.org/grpc v1.82.0 gopkg.in/yaml.v3 v3.0.1 k8s.io/apimachinery v0.36.2 k8s.io/client-go v0.36.2 open-cluster-management.io/api v1.3.0 - open-cluster-management.io/sdk-go v1.3.0 + open-cluster-management.io/sdk-go v1.3.1-0.20260630085947-ac9666c85f0a sigs.k8s.io/controller-runtime v0.24.1 sigs.k8s.io/yaml v1.6.0 ) @@ -172,7 +173,6 @@ require ( google.golang.org/genproto v0.0.0-20260630182238-925bb5da69e7 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7 // indirect - google.golang.org/grpc v1.82.0 // indirect google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect diff --git a/go.sum b/go.sum index d2957332..c8d36fe1 100644 --- a/go.sum +++ b/go.sum @@ -519,8 +519,8 @@ k8s.io/utils v0.0.0-20260626114624-be93311217bd h1:Ea7fgQ5we8Y9T0OX5o0dAHzQOBRI0 k8s.io/utils v0.0.0-20260626114624-be93311217bd/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk= open-cluster-management.io/api v1.3.0 h1:Q3miH38BE3N5+PesHQ0kcFi5nhX5350m7OJWapZcVqY= open-cluster-management.io/api v1.3.0/go.mod h1:t0DsBv4gjIo9ojd7GYfA2tcEMpNf0h5Ix68pFDXwNSk= -open-cluster-management.io/sdk-go v1.3.0 h1:03VnnN3c10FwWnsSl9+apevKadLUw1hrKIu91JHL364= -open-cluster-management.io/sdk-go v1.3.0/go.mod h1:+zL1hpfT73F4kxq3k293H6qsgm8k4YiC1giSwn36fAo= +open-cluster-management.io/sdk-go v1.3.1-0.20260630085947-ac9666c85f0a h1:wdFYMV/zpSlapCadoyALzAS+mLslKJtdhe8M0i3n1kQ= +open-cluster-management.io/sdk-go v1.3.1-0.20260630085947-ac9666c85f0a/go.mod h1:iPACd3YK3e4UMCoNzMDSdYGwKgLEPjM0VYl2nlIpE58= pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= sigs.k8s.io/controller-runtime v0.24.1 h1:miPEwrmirImAvgME1L9qebGHrOnGJoVmVdtOU9fRfo4= diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index 958a0a27..380c3d90 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -925,7 +925,7 @@ func TestCreateHandler_MetricsRecording(t *testing.T) { Build() require.NoError(t, err) - handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger())) + handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger()), logger.NewTestLogger()) evt := event.New() evt.SetID("test-event-1") @@ -974,7 +974,7 @@ func TestCreateHandler_MetricsRecording_Failed(t *testing.T) { Build() require.NoError(t, err) - handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger())) + handler := AlwaysAck(WithMetrics(exec.CreateHandler(), recorder, logger.NewTestLogger()), logger.NewTestLogger()) evt := event.New() evt.SetID("test-event-fail") @@ -1013,7 +1013,7 @@ func TestCreateHandler_NilMetricsRecorder(t *testing.T) { Build() require.NoError(t, err) - handler := AlwaysAck(WithMetrics(exec.CreateHandler(), nil, logger.NewTestLogger())) + handler := AlwaysAck(WithMetrics(exec.CreateHandler(), nil, logger.NewTestLogger()), logger.NewTestLogger()) evt := event.New() evt.SetID("test-event-nil") @@ -1172,7 +1172,7 @@ func TestAlwaysAck_AlwaysReturnsNil(t *testing.T) { return tt.result, tt.err }) - handler := AlwaysAck(inner) + handler := AlwaysAck(inner, logger.NewTestLogger()) evt := event.New() evt.SetID("test-ack") diff --git a/internal/executor/handler.go b/internal/executor/handler.go index 602e58ff..a915248b 100644 --- a/internal/executor/handler.go +++ b/internal/executor/handler.go @@ -11,7 +11,7 @@ import ( // HandlerFunc is a composable event handler. Build a broker-compatible handler with: // -// handler := AlwaysAck(WithMetrics(exec.CreateHandler(), metricsRecorder, log)) +// handler := AlwaysAck(WithMetrics(exec.CreateHandler(), metricsRecorder, log), log) type HandlerFunc func(ctx context.Context, evt *event.Event) (*ExecutionResult, error) // WithMetrics wraps a HandlerFunc to record Prometheus metrics after execution. @@ -46,10 +46,25 @@ func WithMetrics(h HandlerFunc, recorder *metrics.Recorder, log logger.Logger) H // AlwaysAck wraps a HandlerFunc into a broker compatible handler that always returns nil, // preventing infinite retry loops for non-recoverable errors. -// Any error returned by HandlerFunc is discarded, callers must log or record errors before this layer. -func AlwaysAck(h HandlerFunc) func(ctx context.Context, evt *event.Event) error { +// Errors are logged at warn level before being discarded. +func AlwaysAck(h HandlerFunc, log logger.Logger) func(ctx context.Context, evt *event.Event) error { return func(ctx context.Context, evt *event.Event) error { - _, _ = h(ctx, evt) //nolint:errcheck // errors are handled by inner wrappers before this layer + result, err := h(ctx, evt) + errCtx := logger.WithLogFields(ctx, logger.LogFields{ + "event_id": evt.ID(), + "event_type": evt.Type(), + }) + if err != nil { + errCtx = logger.WithErrorField(errCtx, err) + log.Warn(errCtx, "event handler error (acked)") + } else if result != nil && result.Status == StatusFailed { + phases := make([]string, 0, len(result.Errors)) + for phase := range result.Errors { + phases = append(phases, string(phase)) + } + errCtx = logger.WithLogField(errCtx, "failed_phases", phases) + log.Warn(errCtx, "event handler failed (acked)") + } return nil } } diff --git a/internal/maestroclient/client.go b/internal/maestroclient/client.go index acb45753..ab6d95fb 100644 --- a/internal/maestroclient/client.go +++ b/internal/maestroclient/client.go @@ -21,6 +21,8 @@ import ( "github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/version" "github.com/openshift-online/maestro/pkg/api/openapi" "github.com/openshift-online/maestro/pkg/client/cloudevents/grpcsource" + "google.golang.org/grpc" + "google.golang.org/grpc/backoff" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" @@ -35,6 +37,7 @@ import ( const ( DefaultHTTPTimeout = 10 * time.Second DefaultServerHealthinessTimeout = 20 * time.Second + DefaultGRPCBackoffMaxDelay = 5 * time.Second ) // Client is the Maestro client for managing ManifestWorks via CloudEvents gRPC @@ -187,7 +190,19 @@ func NewMaestroClient(ctx context.Context, config *Config, log logger.Logger) (* // Create gRPC options grpcOptions := &grpcopts.GRPCOptions{ - Dialer: &grpcopts.GRPCDialer{}, + Dialer: &grpcopts.GRPCDialer{ + ExtraDialOpts: []grpc.DialOption{ + grpc.WithConnectParams(grpc.ConnectParams{ + Backoff: backoff.Config{ + BaseDelay: 1 * time.Second, + Multiplier: 1.6, + Jitter: 0.2, + MaxDelay: DefaultGRPCBackoffMaxDelay, + }, + MinConnectTimeout: 3 * time.Second, + }), + }, + }, ServerHealthinessTimeout: &serverHealthinessTimeout, } grpcOptions.Dialer.URL = config.GRPCServerAddr diff --git a/internal/maestroclient/operations.go b/internal/maestroclient/operations.go index f96d01ca..0aa24cb4 100644 --- a/internal/maestroclient/operations.go +++ b/internal/maestroclient/operations.go @@ -4,20 +4,62 @@ import ( "context" "encoding/json" "strings" + "time" "github.com/openshift-hyperfleet/hyperfleet-adapter/internal/manifest" "github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/constants" apperrors "github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/errors" "github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/logger" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" kubetypes "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" workv1 "open-cluster-management.io/api/work/v1" ) +const ( + grpcRetryMaxAttempts = 3 + grpcRetryBaseDelay = 1 * time.Second + grpcRetryMultiplier = 2 +) + +func isTransientGRPCError(err error) bool { + if err == nil { + return false + } + return status.Code(err) == codes.Unavailable +} + +func (c *Client) retryOnTransientGRPC(ctx context.Context, fn func() error) error { + var lastErr error + backoff := wait.Backoff{ + Duration: grpcRetryBaseDelay, + Factor: float64(grpcRetryMultiplier), + Steps: grpcRetryMaxAttempts, + } + waitErr := wait.ExponentialBackoffWithContext(ctx, backoff, func(ctx context.Context) (bool, error) { + lastErr = fn() + if lastErr == nil { + return true, nil + } + if !isTransientGRPCError(lastErr) { + return false, lastErr + } + retryCtx := logger.WithErrorField(ctx, lastErr) + c.log.Warn(retryCtx, "transient gRPC error, retrying") + return false, nil + }) + if wait.Interrupted(waitErr) && lastErr != nil { + return lastErr + } + return waitErr +} + // CreateManifestWork creates a new ManifestWork for a target cluster (consumer) // // The ManifestWork object should be pre-constructed from a template with: @@ -54,8 +96,13 @@ func (c *Client) CreateManifestWork( // Set namespace to consumer name (required by Maestro) work.Namespace = consumerName - // Create via the work client - created, err := c.workClient.ManifestWorks(consumerName).Create(ctx, work, metav1.CreateOptions{}) + // Create via the work client with retry on transient gRPC errors + var created *workv1.ManifestWork + err := c.retryOnTransientGRPC(ctx, func() error { + var createErr error + created, createErr = c.workClient.ManifestWorks(consumerName).Create(ctx, work, metav1.CreateOptions{}) + return createErr + }) if err != nil { if isConsumerNotFoundError(err) { return nil, apperrors.NotFound("consumer %q is not registered in Maestro", consumerName) @@ -76,9 +123,13 @@ func (c *Client) GetManifestWork( ctx = logger.WithMaestroConsumer(ctx, consumerName) ctx = logger.WithLogField(ctx, "manifestwork", workName) - work, err := c.workClient.ManifestWorks(consumerName).Get(ctx, workName, metav1.GetOptions{}) + var work *workv1.ManifestWork + err := c.retryOnTransientGRPC(ctx, func() error { + var getErr error + work, getErr = c.workClient.ManifestWorks(consumerName).Get(ctx, workName, metav1.GetOptions{}) + return getErr + }) if err != nil { - // Return not found error without wrapping for callers to check if apierrors.IsNotFound(err) { return nil, err } @@ -99,13 +150,18 @@ func (c *Client) PatchManifestWork( ctx = logger.WithMaestroConsumer(ctx, consumerName) ctx = logger.WithLogField(ctx, "manifestwork", workName) - patched, err := c.workClient.ManifestWorks(consumerName).Patch( - ctx, - workName, - kubetypes.MergePatchType, - patchData, - metav1.PatchOptions{}, - ) + var patched *workv1.ManifestWork + err := c.retryOnTransientGRPC(ctx, func() error { + var patchErr error + patched, patchErr = c.workClient.ManifestWorks(consumerName).Patch( + ctx, + workName, + kubetypes.MergePatchType, + patchData, + metav1.PatchOptions{}, + ) + return patchErr + }) if err != nil { return nil, apperrors.MaestroError("failed to patch ManifestWork %s/%s: %v", consumerName, workName, err) diff --git a/internal/maestroclient/operations_test.go b/internal/maestroclient/operations_test.go index a5031ab3..bb441c7e 100644 --- a/internal/maestroclient/operations_test.go +++ b/internal/maestroclient/operations_test.go @@ -6,11 +6,15 @@ package maestroclient import ( + "context" "fmt" "testing" "github.com/openshift-hyperfleet/hyperfleet-adapter/internal/manifest" "github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/constants" + "github.com/openshift-hyperfleet/hyperfleet-adapter/pkg/logger" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" workv1 "open-cluster-management.io/api/work/v1" ) @@ -207,3 +211,126 @@ func TestGenerationComparison(t *testing.T) { }) } } + +func TestIsTransientGRPCError(t *testing.T) { + tests := []struct { + err error + name string + expected bool + }{ + { + name: "nil returns false", + err: nil, + expected: false, + }, + { + name: "Unavailable returns true", + err: status.Error(codes.Unavailable, "connection refused"), + expected: true, + }, + { + name: "InvalidArgument returns false", + err: status.Error(codes.InvalidArgument, "bad request"), + expected: false, + }, + { + name: "NotFound returns false", + err: status.Error(codes.NotFound, "not found"), + expected: false, + }, + { + name: "PermissionDenied returns false", + err: status.Error(codes.PermissionDenied, "denied"), + expected: false, + }, + { + name: "non-gRPC error returns false", + err: fmt.Errorf("plain error"), + expected: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isTransientGRPCError(tt.err); got != tt.expected { + t.Errorf("isTransientGRPCError() = %v, want %v", got, tt.expected) + } + }) + } +} + +func TestRetryOnTransientGRPC(t *testing.T) { + client := &Client{log: logger.NewTestLogger()} + ctx := context.Background() + + t.Run("succeeds first try", func(t *testing.T) { + calls := 0 + err := client.retryOnTransientGRPC(ctx, func() error { + calls++ + return nil + }) + if err != nil { + t.Errorf("expected nil, got %v", err) + } + if calls != 1 { + t.Errorf("expected 1 call, got %d", calls) + } + }) + + t.Run("succeeds on second try", func(t *testing.T) { + calls := 0 + err := client.retryOnTransientGRPC(ctx, func() error { + calls++ + if calls == 1 { + return status.Error(codes.Unavailable, "transient") + } + return nil + }) + if err != nil { + t.Errorf("expected nil, got %v", err) + } + if calls != 2 { + t.Errorf("expected 2 calls, got %d", calls) + } + }) + + t.Run("gives up after max attempts", func(t *testing.T) { + calls := 0 + err := client.retryOnTransientGRPC(ctx, func() error { + calls++ + return status.Error(codes.Unavailable, "always failing") + }) + if err == nil { + t.Error("expected error, got nil") + } + if calls != grpcRetryMaxAttempts { + t.Errorf("expected %d calls, got %d", grpcRetryMaxAttempts, calls) + } + }) + + t.Run("no retry on non-transient error", func(t *testing.T) { + calls := 0 + err := client.retryOnTransientGRPC(ctx, func() error { + calls++ + return status.Error(codes.InvalidArgument, "bad request") + }) + if err == nil { + t.Error("expected error, got nil") + } + if calls != 1 { + t.Errorf("expected 1 call, got %d", calls) + } + }) + + t.Run("respects context cancellation", func(t *testing.T) { + cancelCtx, cancel := context.WithCancel(ctx) + cancel() + calls := 0 + err := client.retryOnTransientGRPC(cancelCtx, func() error { + calls++ + return status.Error(codes.Unavailable, "transient") + }) + if err != context.Canceled { + t.Errorf("expected context.Canceled, got %v", err) + } + }) +} diff --git a/test/Dockerfile.integration b/test/Dockerfile.integration index 41fdb6c3..b8e83137 100644 --- a/test/Dockerfile.integration +++ b/test/Dockerfile.integration @@ -2,7 +2,7 @@ # Pre-built integration test image with envtest binaries # This eliminates the need for dynamic installation during test runtime -FROM registry.access.redhat.com/ubi9/go-toolset:1.25 +FROM registry.access.redhat.com/ubi9/go-toolset:1.26 # Install runtime dependencies as root, then switch back to non-root. USER root diff --git a/test/integration/executor/executor_integration_test.go b/test/integration/executor/executor_integration_test.go index ee683232..0b68ac3c 100644 --- a/test/integration/executor/executor_integration_test.go +++ b/test/integration/executor/executor_integration_test.go @@ -708,7 +708,7 @@ func TestExecutor_Handler_Integration(t *testing.T) { } // Get the handler function - handler := executor.AlwaysAck(exec.CreateHandler()) + handler := executor.AlwaysAck(exec.CreateHandler(), logger.NewTestLogger()) // Simulate broker calling the handler evt := createTestEvent("cluster-handler-test") @@ -759,7 +759,7 @@ func TestExecutor_Handler_PreconditionNotMet_ReturnsNil(t *testing.T) { t.Fatalf("Failed to create executor: %v", err) } - handler := executor.AlwaysAck(exec.CreateHandler()) + handler := executor.AlwaysAck(exec.CreateHandler(), logger.NewTestLogger()) evt := createTestEvent("cluster-skip") // Handler should return nil even when precondition not met @@ -862,7 +862,7 @@ func TestExecutor_MissingRequiredParam(t *testing.T) { } // Test handler behavior: should ACK (not NACK) invalid events - handler := executor.AlwaysAck(exec.CreateHandler()) + handler := executor.AlwaysAck(exec.CreateHandler(), logger.NewTestLogger()) err = handler(context.Background(), evt) if err != nil { t.Errorf("Handler should ACK (return nil) for param extraction failures, got error: %v", err) @@ -918,7 +918,7 @@ func TestExecutor_InvalidEventJSON(t *testing.T) { assert.Empty(t, result.ResourceResults, "Resources should not execute for invalid event") // Test handler behavior: should ACK (not NACK) invalid events - handler := executor.AlwaysAck(exec.CreateHandler()) + handler := executor.AlwaysAck(exec.CreateHandler(), logger.NewTestLogger()) err = handler(context.Background(), &evt) assert.Nil(t, err, "Handler should ACK (return nil) for invalid events, not NACK") @@ -976,7 +976,7 @@ func TestExecutor_MissingEventFields(t *testing.T) { assert.Empty(t, result.ResourceResults, "Resources should not execute for missing required field") // Test handler behavior: should ACK (not NACK) events with missing required fields - handler := executor.AlwaysAck(exec.CreateHandler()) + handler := executor.AlwaysAck(exec.CreateHandler(), logger.NewTestLogger()) errPhase = handler(context.Background(), &evt) assert.Nil(t, errPhase, "Handler should ACK (return nil) for missing required fields, not NACK")