ROSAENG-62444: feat: replace DynamoDB Streams informer with EventBridge Pipes for specs and status paths - #291
Conversation
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: rrp-bot The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
openshift-ci
Bot
added
the
needs-ok-to-test
|
Hi @rrp-bot. Thanks for your PR. I'm waiting for a openshift-online member to verify that this patch is reasonable to test. If it is, they should reply with Tip We noticed you've done this a few times! Consider joining the org to skip this step and gain Once the patch is verified, the new status will be reflected by the I understand the commands that are listed here. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
WalkthroughThe operator replaces DynamoDB Streams status watching with per-replica SQS consumption. Helm and manager configuration resolve queue URLs. Integration simulators dispatch events after successful status writes. ChangesSQS status migration
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant StatefulSet
participant Manager
participant SQS
participant StatusSQSConsumer
participant EventRouter
StatefulSet->>Manager: Provide SQS queue URL prefix
Manager->>Manager: Resolve queue URL from pod ordinal
Manager->>SQS: Create SQS client
Manager->>StatusSQSConsumer: Start consumer
SQS->>StatusSQSConsumer: Return status notification
StatusSQSConsumer->>EventRouter: Dispatch document ID
StatusSQSConsumer->>SQS: Delete processed message
Important Pre-merge checks failedPlease resolve all errors before merging. Addressing warnings is optional. ❌ Failed checks (1 error, 2 warnings)
✅ Passed checks (8 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@hyperfleet-operator/cmd/manager/main.go`:
- Around line 100-105: Update the pod ordinal handling around podOrdinal() so
Atoi parse failures are propagated instead of treating the returned zero as
valid. Ensure the manager exits or returns the error before constructing
sqsStatusQueueURL, while preserving normal prefix URL construction for valid
ordinals.
- Around line 219-226: Delay starting statusConsumer.Run until
mgr.Start(signalCtx) has completed initial route registration, or otherwise
buffer messages until EventRouter can accept them; update the startup flow
around EventRouter.Dispatch and statusConsumer.Run while preserving cancellation
via watchCtx and watchCancel.
- Around line 224-226: Update the watch context initialization near
statusConsumer.Run to derive watchCtx from signalCtx instead of
context.Background(), while retaining its cancellation cleanup. This ensures the
consumer stops when manager shutdown begins through signalCtx.
In `@hyperfleet-operator/internal/dynamo/statussqsconsumer/consumer.go`:
- Around line 117-120: Update the JSON unmarshal error handling in the SQS
consumer to stop logging the complete message body. Keep logging the decode
error and include only safe metadata, such as the message identifier or receipt
context, while preserving deletion through deleteMessage.
In `@hyperfleet-operator/test/helpers_test.go`:
- Around line 39-62: Update the cleanup retry callback in Eventually to return
the first error from any c.List, SetFinalizers-related c.Update, or c.Delete
operation instead of discarding errors. Propagate each failure immediately so
retries report the failed Kubernetes operation, and return the success value
only after all ClusterList, NodePoolList, and ManifestList cleanup completes.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository: openshift-online/coderabbit/.coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 2469c074-dbd7-45b4-8a7b-1c49f11e1bf0
⛔ Files ignored due to path filters (1)
hyperfleet-operator/go.sumis excluded by!**/*.sum
📒 Files selected for processing (12)
hyperfleet-operator/charts/templates/statefulset.yamlhyperfleet-operator/charts/values.yamlhyperfleet-operator/cmd/manager/main.gohyperfleet-operator/go.modhyperfleet-operator/internal/dynamo/client.gohyperfleet-operator/internal/dynamo/statussqsconsumer/consumer.gohyperfleet-operator/internal/dynamo/statussqsconsumer/consumer_test.gohyperfleet-operator/internal/dynamo/statusstream/manager.gohyperfleet-operator/internal/dynamo/statusstream/watcher.gohyperfleet-operator/internal/dynamo/statusstream/watcher_test.gohyperfleet-operator/test/helpers_test.gohyperfleet-operator/test/suite_test.go
💤 Files with no reviewable changes (3)
- hyperfleet-operator/internal/dynamo/statusstream/watcher.go
- hyperfleet-operator/internal/dynamo/statusstream/manager.go
- hyperfleet-operator/internal/dynamo/statusstream/watcher_test.go
| // If a prefix was given, construct the full queue URL by appending the | ||
| // pod ordinal — matching the queue naming convention: | ||
| // <prefix><ordinal> e.g. https://sqs…/regional-hyperfleet-operator-2 | ||
| if sqsStatusQueueURLPrefix != "" { | ||
| sqsStatusQueueURL = fmt.Sprintf("%s%d", sqsStatusQueueURLPrefix, ordinal) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fail when the pod ordinal is invalid.
podOrdinal() returns (0, nil) when strconv.Atoi fails. This prefix path then consumes replica 0's queue. Return the parse error so an invalid hostname cannot create competing consumers for the same queue.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@hyperfleet-operator/cmd/manager/main.go` around lines 100 - 105, Update the
pod ordinal handling around podOrdinal() so Atoi parse failures are propagated
instead of treating the returned zero as valid. Ensure the manager exits or
returns the error before constructing sqsStatusQueueURL, while preserving normal
prefix URL construction for valid ordinals.
8172b92 to
6848565
Compare
psav
changed the title
openshift-ci-robot
added
the
jira/valid-reference
|
@rrp-bot: This pull request references ROSAENG-62444 which is a valid jira issue. Warning: The referenced jira issue has an invalid target version for the target branch this PR targets: expected the story to target the "5.0.0" version, but no target version was set. DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the openshift-eng/jira-lifecycle-plugin repository. |
…ecs and status paths Remove the dynamo stream-based polling from the hyperfleet-operator entirely. The operator previously relied on DynamoDB Streams informers for desire-write notifications on the specs path, and kube-applier published status updates back. Both paths are now driven by EventBridge Pipes reading directly from DynamoDB Streams and delivering to SQS queues. Changes: - Delete hyperfleet-operator/internal/dynamo/snspublisher/ package entirely - Remove SNSPublisher interface and NewClientWithSNS constructor from dynamo client - Remove --sns-status-topic-arn flag from operator options - Add statussqsconsumer package: polls per-replica SQS queue for status documentID notifications delivered by EventBridge Pipes - Wire statussqsconsumer into manager startup; consumer index derived from pod name suffix for deterministic queue assignment - go.mod/go.sum: add aws-sdk-go-v2/service/sqs dependency Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…nsumer Add a guard in handleMessage that detects when documentID is a literal <$.path> placeholder — caused by jsonencode() in the pipe's input_template. Previously these messages would pass the empty-string check and be dispatched to EventRouter with a nonsense key, silently dropped with no error logged. Also update StatusNotification comment to reflect the SQS delivery path and import strings. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- consumer.go: remove SQS message body from error log to avoid leaking customer data in log output - main.go: derive consumer context from signalCtx so the consumer shuts down cleanly on manager termination; delay consumer start until mgr.GetCache().WaitForCacheSync() so all controllers have registered their EventRouter routes before messages are dispatched (messages dispatched before route registration are silently dropped and deleted, losing the notification) - test/helpers_test.go: return errors from c.List, c.Update, c.Delete in the Eventually cleanup callback instead of discarding them, so test retries report the failing Kubernetes operation Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
3dba255 to
dbac28c
Compare
|
/ok-to-test |
openshift-ci
Bot
added
ok-to-test
|
@rrp-bot: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
4 participants
Summary
Remove the DynamoDB Streams-based status polling from the hyperfleet-operator and replace with EventBridge Pipes → SQS delivery. Introduces
statussqsconsumerfor per-replica SQS queue polling.Why: The previous approach required the operator to poll DynamoDB Streams directly for status change events, which is complex to manage across replicas and requires stream-level IAM access. EventBridge Pipes now reads INSERT/MODIFY events from the RC status DynamoDB Streams and delivers documentIDs to per-replica SQS queues. Each operator replica polls its own queue (index derived from pod name suffix).
Commits
feat: replace DynamoDB Streams informer with EventBridge Pipes for specs and status paths— deletesdynamo/snspublisherpackage, deletesdynamo/statusstreampackage, introducesdynamo/statussqsconsumer, wires into manager startupfix: detect unresolved EventBridge JSONPath placeholders in status consumer— guards against literal<$.path>placeholder documentIDs, updatesStatusNotificationcommentRelated PRs
Testing
Unit tests cover the consumer happy path and placeholder guard. Integration tests exercise the full operator stack.
Co-Authored-By: Claude Sonnet 4.6 noreply@anthropic.com
Summary by CodeRabbit
New Features
Bug Fixes