|
| 1 | +// Package queue provides helpers for working with client-go's `workqueues` and control flow handlers. |
| 2 | +// |
| 3 | +// This file provides handlers that integrate with the state package to provide clean |
| 4 | +// control flow patterns for controllers. Instead of manually calling queue operations |
| 5 | +// and returning, controllers can now use: |
| 6 | +// |
| 7 | +// return queue.Done() |
| 8 | +// return queue.Requeue() |
| 9 | +// return queue.RequeueAfter(5 * time.Second) |
| 10 | +// return queue.RequeueErr(err) |
| 11 | +// |
| 12 | +// These handlers automatically call the appropriate queue operations and terminate |
| 13 | +// the handler pipeline by returning nil. |
| 14 | +// |
| 15 | +// # Error Propagation Pattern |
| 16 | +// |
| 17 | +// Errors in this system are communicated via: |
| 18 | +// |
| 19 | +// 1. **Context Cancellation**: ctx.Err() is checked by Continue() and parallel execution. |
| 20 | +// When context is cancelled, the pipeline stops (unless WithErrorHandler is set). |
| 21 | +// |
| 22 | +// 2. **Explicit Error Handlers**: queue.RequeueErr(err) and queue.RequeueAPIErr(err) |
| 23 | +// take an explicit error and pass it to the queue for retry logic. |
| 24 | +// |
| 25 | +// 3. **Context Values**: Steps can store errors in context using typedctx.Box or |
| 26 | +// context.WithValue, then check them in subsequent steps or OnError handlers. |
| 27 | +// |
| 28 | +// 4. **Decision-based Error Handling**: Use Decision() to branch based on whether |
| 29 | +// an error occurred, or handle errors inline: |
| 30 | +// |
| 31 | +// riskyOperation := state.NewStepFunc(func(ctx context.Context, next state.Step) state.Step { |
| 32 | +// result, err := doSomethingRisky() |
| 33 | +// if err != nil { |
| 34 | +// return queue.RequeueErr(err).Step().Run(ctx) // Handle error inline |
| 35 | +// } |
| 36 | +// // Store result and continue |
| 37 | +// ctx = context.WithValue(ctx, "result", result) |
| 38 | +// return state.Continue(ctx, next) |
| 39 | +// }) |
| 40 | +// |
| 41 | +// Or with Decision for conditional logic: |
| 42 | +// |
| 43 | +// Sequence( |
| 44 | +// validateInput, |
| 45 | +// Decision( |
| 46 | +// func(ctx context.Context) bool { return isValid(ctx) }, |
| 47 | +// continueProcessing, |
| 48 | +// queue.RequeueErr(fmt.Errorf("validation failed")), |
| 49 | +// ), |
| 50 | +// ) |
| 51 | +package queue |
| 52 | + |
| 53 | +import ( |
| 54 | + "context" |
| 55 | + "time" |
| 56 | + |
| 57 | + "github.com/authzed/controller-idioms/state" |
| 58 | +) |
| 59 | + |
| 60 | +// Done creates a handler that marks the current queue key as finished and terminates the pipeline. |
| 61 | +// This is equivalent to calling queue.NewQueueOperationsCtx().Done(ctx) and returning. |
| 62 | +// |
| 63 | +// Usage: |
| 64 | +// |
| 65 | +// pipeline := state.Sequence( |
| 66 | +// validateInput, |
| 67 | +// processResource, |
| 68 | +// queue.Done(), // Stop here - processing complete |
| 69 | +// ) |
| 70 | +func Done() state.NewStep { |
| 71 | + return state.NewTerminalStepFunc(func(ctx context.Context) { |
| 72 | + NewQueueOperationsCtx().Done(ctx) |
| 73 | + }) |
| 74 | +} |
| 75 | + |
| 76 | +// Requeue creates a handler that requeues the current key immediately and terminates the pipeline. |
| 77 | +// This is equivalent to calling queue.NewQueueOperationsCtx().Requeue(ctx) and returning. |
| 78 | +// |
| 79 | +// Usage: |
| 80 | +// |
| 81 | +// pipeline := state.Decision( |
| 82 | +// resourceReady, |
| 83 | +// continueProcessing, |
| 84 | +// queue.Requeue(), // Not ready - try again immediately |
| 85 | +// ) |
| 86 | +func Requeue() state.NewStep { |
| 87 | + return state.NewTerminalStepFunc(func(ctx context.Context) { |
| 88 | + NewQueueOperationsCtx().Requeue(ctx) |
| 89 | + }) |
| 90 | +} |
| 91 | + |
| 92 | +// RequeueAfter creates a handler that requeues the current key after the specified duration |
| 93 | +// and terminates the pipeline. |
| 94 | +// This is equivalent to calling queue.NewQueueOperationsCtx().RequeueAfter(ctx, duration) and returning. |
| 95 | +// |
| 96 | +// Usage: |
| 97 | +// |
| 98 | +// pipeline := state.Decision( |
| 99 | +// resourceReady, |
| 100 | +// continueProcessing, |
| 101 | +// queue.RequeueAfter(30 * time.Second), // Not ready - try again in 30s |
| 102 | +// ) |
| 103 | +func RequeueAfter(duration time.Duration) state.NewStep { |
| 104 | + return state.NewTerminalStepFunc(func(ctx context.Context) { |
| 105 | + NewQueueOperationsCtx().RequeueAfter(ctx, duration) |
| 106 | + }) |
| 107 | +} |
| 108 | + |
| 109 | +// RequeueErr creates a handler that records an error and requeues the current key immediately, |
| 110 | +// then terminates the pipeline. |
| 111 | +// This is equivalent to calling queue.NewQueueOperationsCtx().RequeueErr(ctx, err) and returning. |
| 112 | +// |
| 113 | +// Usage - return directly from within a step when error occurs: |
| 114 | +// |
| 115 | +// validateInput := state.NewStepFunc(func(ctx context.Context, next state.Step) state.Step { |
| 116 | +// obj := getObjectFromContext(ctx) |
| 117 | +// if err := validate(obj); err != nil { |
| 118 | +// return queue.RequeueErr(fmt.Errorf("validation failed: %w", err)).Step().Run(ctx) |
| 119 | +// } |
| 120 | +// return state.Continue(ctx, next) |
| 121 | +// }) |
| 122 | +// |
| 123 | +// Or use in Decision branches when the error is known statically: |
| 124 | +// |
| 125 | +// state.Decision( |
| 126 | +// inputValid, |
| 127 | +// continueProcessing, |
| 128 | +// queue.RequeueErr(fmt.Errorf("validation failed")), // Error known at composition time |
| 129 | +// ) |
| 130 | +func RequeueErr(err error) state.NewStep { |
| 131 | + return state.NewTerminalStepFunc(func(ctx context.Context) { |
| 132 | + NewQueueOperationsCtx().RequeueErr(ctx, err) |
| 133 | + }) |
| 134 | +} |
| 135 | + |
| 136 | +// RequeueAPIErr creates a handler that handles API errors with appropriate retry logic |
| 137 | +// and terminates the pipeline. |
| 138 | +// This checks if the error contains retry information from the API server and requeues |
| 139 | +// accordingly, equivalent to calling queue.NewQueueOperationsCtx().RequeueAPIErr(ctx, err). |
| 140 | +// |
| 141 | +// Usage - return directly from within a step when error occurs: |
| 142 | +// |
| 143 | +// callKubernetesAPI := state.NewStepFunc(func(ctx context.Context, next state.Step) state.Step { |
| 144 | +// result, err := clientset.AppsV1().Deployments(ns).Get(ctx, name, metav1.GetOptions{}) |
| 145 | +// if err != nil { |
| 146 | +// return queue.RequeueAPIErr(err).Step().Run(ctx) // execute inline, terminating the pipeline |
| 147 | +// } |
| 148 | +// // Store result in context and continue |
| 149 | +// ctx = context.WithValue(ctx, "deployment", result) |
| 150 | +// return state.Continue(ctx, next) |
| 151 | +// }) |
| 152 | +// |
| 153 | +// Note: .Step() is equivalent to calling the NewStep with nil: queue.RequeueAPIErr(err)(nil) |
| 154 | +// but is more readable and explicit about the conversion. |
| 155 | +// |
| 156 | +// This pattern keeps errors local to where they occur, avoiding the need to store |
| 157 | +// errors in context or check them in Decision branches. |
| 158 | +func RequeueAPIErr(err error) state.NewStep { |
| 159 | + return state.NewTerminalStepFunc(func(ctx context.Context) { |
| 160 | + NewQueueOperationsCtx().RequeueAPIErr(ctx, err) |
| 161 | + }) |
| 162 | +} |
| 163 | + |
| 164 | +// OnError creates a handler that executes different queue operations based on whether |
| 165 | +// an error occurred in the context. |
| 166 | +// |
| 167 | +// Usage: |
| 168 | +// |
| 169 | +// pipeline := state.Sequence( |
| 170 | +// riskyOperation, |
| 171 | +// queue.OnError( |
| 172 | +// queue.RequeueErr(fmt.Errorf("operation failed")), // If error |
| 173 | +// queue.Done(), // If success |
| 174 | +// ), |
| 175 | +// ) |
| 176 | +func OnError(errorHandler, successHandler state.NewStep) state.NewStep { |
| 177 | + return func(next state.Step) state.Step { |
| 178 | + errStep := errorHandler(next) |
| 179 | + okStep := successHandler(next) |
| 180 | + return state.StepFunc(func(ctx context.Context) state.Step { |
| 181 | + if ctx.Err() != nil { |
| 182 | + if errStep != nil { |
| 183 | + return errStep.Run(ctx) |
| 184 | + } |
| 185 | + return nil |
| 186 | + } |
| 187 | + if okStep != nil { |
| 188 | + return okStep.Run(ctx) |
| 189 | + } |
| 190 | + return nil |
| 191 | + }) |
| 192 | + } |
| 193 | +} |
0 commit comments