Skip to content

ROSAENG-62444: feat: replace DynamoDB Streams informer with EventBridge Pipes for specs and status paths #291

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rrp-bot wants to merge 4 commits into
base: main
Choose a base branch
from
Open

ROSAENG-62444: feat: replace DynamoDB Streams informer with EventBridge Pipes for specs and status paths #291

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ff759f9
feat: replace DynamoDB Streams informer with EventBridge Pipes for sp…
rrp-bot Jul 24, 2026
f301c32
fix: detect unresolved EventBridge JSONPath placeholders in status co…
rrp-bot Aug 7, 2026
77ece00
Up go
psav Aug 7, 2026
dbac28c
fix: address CodeRabbit review comments on statussqsconsumer
rrp-bot Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions hyperfleet-operator/charts/templates/statefulset.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ spec:
- --aws-region={{ required "awsRegion is required" .Values.awsRegion }}
- --base-domain={{ required "baseDomain is required" .Values.baseDomain }}
- --health-probe-bind-address=:8081
- --sqs-status-queue-url-prefix={{ required "sqsQueueUrlPrefix is required" .Values.sqsQueueUrlPrefix }}
env:
- name: POSTGRES_DSN
valueFrom:
Expand Down
7 changes: 7 additions & 0 deletions hyperfleet-operator/charts/values.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,13 @@ image:
awsRegion: ""
baseDomain: ""

# sqsQueueUrlPrefix is the SQS queue URL prefix for the per-replica status
# queues. The pod ordinal (derived from the StatefulSet hostname) is appended
# to form the full queue URL, e.g.:
# https://sqs.us-east-1.amazonaws.com/123456789012/regional-hyperfleet-operator-2
# Must end with a hyphen before the ordinal digit.
sqsQueueUrlPrefix: ""

postgres:
secretName: postgres-dsn
secretKey: dsn
Expand Down
66 changes: 47 additions & 19 deletions hyperfleet-operator/cmd/manager/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,17 +20,14 @@ import (
"context"
"flag"
"fmt"
"log/slog"
"os"
"strconv"
"strings"
"time"

awsconfig "github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
"github.com/aws/aws-sdk-go-v2/service/dynamodbstreams"
"github.com/aws/aws-sdk-go-v2/service/sqs"
hyperfleetdb "github.com/openshift-online/rosa-hyperfleet-api/hyperfleet-db"

"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
ctrl "sigs.k8s.io/controller-runtime"
Expand All @@ -41,7 +38,7 @@ import (
v1alpha1 "github.com/openshift-online/rosa-hyperfleet-api/api/v1alpha1"
"github.com/openshift-online/rosa-hyperfleet-api/hyperfleet-operator/internal/controller"
"github.com/openshift-online/rosa-hyperfleet-api/hyperfleet-operator/internal/dynamo"
"github.com/openshift-online/rosa-hyperfleet-api/hyperfleet-operator/internal/dynamo/statusstream"
"github.com/openshift-online/rosa-hyperfleet-api/hyperfleet-operator/internal/dynamo/statussqsconsumer"
"github.com/openshift-online/rosa-hyperfleet-api/hyperfleet-operator/internal/render"
)

Expand All @@ -53,13 +50,16 @@ func main() {
var awsRegion string
var baseDomain string
var maxConcurrentReconciles int
var sqsStatusQueueURL string
var sqsStatusQueueURLPrefix string

flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metrics endpoint binds to.")
flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
flag.StringVar(&awsRegion, "aws-region", "", "AWS region for DynamoDB and EKS (required).")
flag.StringVar(&awsRegion, "aws-region", "", "AWS region for DynamoDB, SQS, and EKS (required).")
flag.StringVar(&baseDomain, "base-domain", "", "DNS base domain for hosted clusters (required).")
flag.IntVar(&maxConcurrentReconciles, "max-concurrent-reconciles", 10,
"Maximum number of concurrent reconciles per controller.")
flag.StringVar(&sqsStatusQueueURL, "sqs-status-queue-url", "", "Full SQS queue URL for receiving status change notifications from kube-applier. Mutually exclusive with --sqs-status-queue-url-prefix.")
flag.StringVar(&sqsStatusQueueURLPrefix, "sqs-status-queue-url-prefix", "", "SQS queue URL prefix; the pod ordinal (from hostname) is appended to form the full queue URL. Mutually exclusive with --sqs-status-queue-url.")
flag.IntVar(&maxConcurrentReconciles, "max-concurrent-reconciles", 10, "Maximum number of concurrent reconciles per controller.")

opts := zap.Options{Development: true}
opts.BindFlags(flag.CommandLine)
Expand All @@ -75,6 +75,14 @@ func main() {
setupLog.Error(nil, "--base-domain is required")
os.Exit(1)
}
if sqsStatusQueueURL != "" && sqsStatusQueueURLPrefix != "" {
setupLog.Error(nil, "--sqs-status-queue-url and --sqs-status-queue-url-prefix are mutually exclusive")
os.Exit(1)
}
if sqsStatusQueueURL == "" && sqsStatusQueueURLPrefix == "" {
setupLog.Error(nil, "one of --sqs-status-queue-url or --sqs-status-queue-url-prefix is required")
os.Exit(1)
}

dsn := os.Getenv("POSTGRES_DSN")
if dsn == "" {
Expand All @@ -89,6 +97,13 @@ func main() {
os.Exit(1)
}

// 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)
}
Comment on lines +100 to +105

@coderabbitai coderabbitai Bot Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ 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.


setupLog.Info("shard config",
"replicaCount", replicaCount,
"ordinal", ordinal,
Expand Down Expand Up @@ -131,8 +146,9 @@ func main() {
}

dynamoDBClient := dynamodb.NewFromConfig(awsCfg)

sqsClient := sqs.NewFromConfig(awsCfg)
dynamoClient := dynamo.NewClient(dynamoDBClient)
streamsClient := dynamodbstreams.NewFromConfig(awsCfg)

rcfg := render.RegionalConfig{
BaseDomain: baseDomain,
Expand Down Expand Up @@ -196,19 +212,31 @@ func main() {
os.Exit(1)
}

streamMgr := statusstream.NewManager(
dynamoDBClient,
streamsClient,
mgr.GetClient(),
[]string{dynamo.TableSuffixStatusApplyDesires, dynamo.TableSuffixStatusReadDesires},
// Each operator replica drains its own pre-provisioned SQS queue.
// EventBridge Pipes deliver status change notifications to the queue after
// each DynamoDB status write. EventRouter.Dispatch silently drops document
// IDs it does not own, so no per-MC filtering is required here.
statusConsumer := statussqsconsumer.New(
sqsClient,
sqsStatusQueueURL,
func(documentID string) { eventRouter.Dispatch(documentID) },
slog.Default().With("component", "statusstream"),
)
watchCtx, watchCancel := context.WithCancel(context.Background())
defer watchCancel()
go streamMgr.Run(watchCtx, 5*time.Second)
// Derive the consumer context from signalCtx so the consumer shuts down
// cleanly when the manager receives a termination signal.
// Start the consumer only after the cache has synced so that all
// controllers have registered their EventRouter routes. Messages received
// before routes are registered would be dispatched to nobody and then
// deleted, losing the notification.
go func() {
if !mgr.GetCache().WaitForCacheSync(signalCtx) {
return
}
statusConsumer.Run(signalCtx)
}()

setupLog.Info("Starting pgruntime manager")
setupLog.Info("Starting pgruntime manager",
"sqsStatusQueueURL", sqsStatusQueueURL,
)
if err := mgr.Start(signalCtx); err != nil {
setupLog.Error(err, "Failed to run manager")
os.Exit(1)
Expand Down
5 changes: 3 additions & 2 deletions hyperfleet-operator/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ require (
github.com/aws/aws-sdk-go-v2/credentials v1.19.30
github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.54
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.62.0
github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.36.0
github.com/aws/aws-sdk-go-v2/service/sqs v1.37.8
github.com/google/uuid v1.6.0
github.com/jackc/pgx/v5 v5.10.0
github.com/onsi/ginkgo/v2 v2.27.4
Expand All @@ -23,7 +23,7 @@ require (
github.com/openshift/api v0.0.0-20260416105050-3c6b218b8a80
github.com/openshift/hypershift/api v0.0.0-20260625052409-9acec4759a16
github.com/prometheus/client_golang v1.23.2
github.com/rrp-bot/rosa-hyperfleet-kube-applier v0.0.0-20260716171749-c71c7549f8db
github.com/rrp-bot/rosa-hyperfleet-kube-applier v0.0.0-20260730140449-9106cf5c01ed
k8s.io/api v0.36.1
k8s.io/apimachinery v0.36.1
k8s.io/client-go v0.36.1
Expand All @@ -37,6 +37,7 @@ require (
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.31 // indirect
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.31 // indirect
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.32 // indirect
github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.36.0 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.13 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.12.8 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.31 // indirect
Expand Down
6 changes: 4 additions & 2 deletions hyperfleet-operator/go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.31 h1:w2SIhW92
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.31/go.mod h1:wAhpCQbkov+IcvjozJbd2xRCoZybUEHNkcFunssNACg=
github.com/aws/aws-sdk-go-v2/service/signin v1.5.0 h1:OHH5iTQvVGmfHjX/5Q+vFuA/Rf2x6/95aJ/75QCQSm4=
github.com/aws/aws-sdk-go-v2/service/signin v1.5.0/go.mod h1:mCF3AK9PpL49oOrhniUXWAfhVBVQ/XbytoE5eccZUIs=
github.com/aws/aws-sdk-go-v2/service/sqs v1.37.8 h1:70G7GI+dwy3tydU6ig6jyMOhtigYk80OafPDfWyqmlU=
github.com/aws/aws-sdk-go-v2/service/sqs v1.37.8/go.mod h1:VS6v7DyZL6dnc6Lz850vFzW+Nhzpcgj+P1ftJEBngyE=
github.com/aws/aws-sdk-go-v2/service/sso v1.33.0 h1:CaJyYhxBE0M/HJX/YvSaSmQlsI91VHB0lKU8LtLxL3A=
github.com/aws/aws-sdk-go-v2/service/sso v1.33.0/go.mod h1:+e6BMRMPjBQoCw/WovYR9GLy2IU0z4Q77smOB1DraSg=
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.38.0 h1:tC323YV77QdafeBr6LUhLDTsboyuyHLNRwAyCP44kGU=
Expand Down Expand Up @@ -170,8 +172,8 @@ github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
github.com/rrp-bot/rosa-hyperfleet-kube-applier v0.0.0-20260716171749-c71c7549f8db h1:B2TcZMSBzIq1Jf3R3qJHG6UytQxsk31bdLyAVFg5Oyw=
github.com/rrp-bot/rosa-hyperfleet-kube-applier v0.0.0-20260716171749-c71c7549f8db/go.mod h1:fQG90W6c4XJ2SrpWo5gy64AeHHP2W8XI5/VQ5z+MGsY=
github.com/rrp-bot/rosa-hyperfleet-kube-applier v0.0.0-20260730140449-9106cf5c01ed h1:4lBinC7nznkY2LfrEVwHU7F4PDALv28nOOU1Eg5yvOM=
github.com/rrp-bot/rosa-hyperfleet-kube-applier v0.0.0-20260730140449-9106cf5c01ed/go.mod h1:akfFjlRC94AbZUaa6haV0glQ618TYPCPHKSef85ITRQ=
github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
Expand Down
17 changes: 13 additions & 4 deletions hyperfleet-operator/internal/dynamo/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@ import (
var ErrNotFound = errors.New("desire not found")

const (
TableSuffixApplyDesires = "-applydesires"
TableSuffixReadDesires = "-readdesires"
TableSuffixApplyDesires = "-applydesires"
TableSuffixReadDesires = "-readdesires"
TableSuffixStatusApplyDesires = "-status-applydesires"
TableSuffixStatusReadDesires = "-status-readdesires"
attributeDocumentID = "documentID"
Expand Down Expand Up @@ -66,18 +66,27 @@ type Client struct {

var _ DesireClient = (*Client)(nil)

// NewClient returns a Client backed by the given DynamoDB API.
func NewClient(db dynamoAPI) *Client {
return &Client{db: db}
}

// UpsertApplyDesire writes an ApplyDesire spec only when content has changed.
func (c *Client) UpsertApplyDesire(ctx context.Context, specsPrefix string, desire *ApplyDesire) (UpsertResult, error) {
return c.upsertDesire(ctx, specsPrefix+TableSuffixApplyDesires, desire.DocumentID, desire.Spec)
result, err := c.upsertDesire(ctx, specsPrefix+TableSuffixApplyDesires, desire.DocumentID, desire.Spec)
if err != nil {
return result, err
}
return result, nil
}

// UpsertReadDesire writes a ReadDesire spec only when content has changed.
func (c *Client) UpsertReadDesire(ctx context.Context, specsPrefix string, desire *ReadDesire) (UpsertResult, error) {
return c.upsertDesire(ctx, specsPrefix+TableSuffixReadDesires, desire.DocumentID, desire.Spec)
result, err := c.upsertDesire(ctx, specsPrefix+TableSuffixReadDesires, desire.DocumentID, desire.Spec)
if err != nil {
return result, err
}
return result, nil
}

// GetApplyDesireStatus reads an ApplyDesire from the status table.
Expand Down
Loading