Skip to content

ROSAENG-62444: feat: replace DynamoDB Streams with SNS/SQS for status notifications #184

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

Closed
rrp-bot wants to merge 11 commits into from
Closed

ROSAENG-62444: feat: replace DynamoDB Streams with SNS/SQS for status notifications #184

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
5da29bc
feat: replace DynamoDB Streams with SNS/SQS for status notifications
rrp-bot Jul 24, 2026
bed56f8
fix: port integration test fixes correctly onto monorepo versions
rrp-bot Jul 24, 2026
a6be8b9
fix: restore hyperfleetdb in cmd/manager/main.go, remove jmelis/pgrun…
rrp-bot Jul 24, 2026
2b03d7d
fix: drop dynamodbstreams dep, bump kube-applier to feature/sns-sqs, …
rrp-bot Jul 24, 2026
cdf2aa6
Update go mod
psav Jul 27, 2026
52c449c
feat: add --sqs-status-queue-url-prefix flag for StatefulSet pod self…
rrp-bot Jul 28, 2026
ccbc4c9
fix: correct compactor binary path to /app/compactor
rrp-bot Jul 28, 2026
7abe27d
chore: bump kube-applier to 9106cf5 (Up go.mod)
rrp-bot Jul 30, 2026
42ea423
fix: correct pseudo-version timestamp for kube-applier 9106cf5
rrp-bot Jul 30, 2026
bfe629c
Updated
psav Jul 30, 2026
c756faf
fix: correct SNS topic ARN format in snspublisher to match Terraform …
rrp-bot Jul 30, 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
2 changes: 1 addition & 1 deletion hyperfleet-operator/charts/templates/compactor-deployment.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ spec:
- name: compactor
image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
command: ["/compactor"]
command: ["/app/compactor"]
args:
- --interval={{ .Values.compactor.interval }}
- --retention={{ .Values.compactor.retention }}
Expand Down
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
73 changes: 55 additions & 18 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,16 @@ 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/sns"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sts"
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 +40,8 @@ import (
v1alpha1 "github.com/openshift-online/rosa-hyperfleet-api/hyperfleet-operator/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/snspublisher"
"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 +53,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, SNS, 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 +78,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 +100,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 +103 to +109

@coderabbitai coderabbitai Bot Jul 28, 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

Queue-URL construction silently misroutes if podOrdinal() can't parse the hostname suffix.

This new logic makes ordinal load-bearing for correct per-replica SQS routing: sqsStatusQueueURL = prefix + ordinal. podOrdinal() (below, unchanged) swallows strconv.Atoi failures and returns (0, nil) instead of an error:

ordinal, err := strconv.Atoi(last)
if err != nil {
    return 0, nil
}

The existing if err != nil { os.Exit(1) } check at the call site never fires in that case, so a malformed/non-StatefulSet hostname silently resolves to ordinal 0 instead of failing fast. Since two replicas defaulting to ordinal 0 would poll the same SQS queue while another replica's queue goes undrained, this can silently cause missed status notifications for the misrouted replica. Recommend propagating the parse error instead of masking it.

🐛 Suggested fix in podOrdinal (outside this diff range) 
 	ordinal, err := strconv.Atoi(last)
 	if err != nil {
-		return 0, nil
+		return 0, fmt.Errorf("parse ordinal from hostname %q: %w", hostname, err)
 	}
 	return ordinal, nil
🤖 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 103 - 109, Update
podOrdinal to propagate the strconv.Atoi failure instead of returning ordinal 0
with a nil error. Preserve the existing successful parsing behavior, so the
caller’s existing err check exits on malformed or non-StatefulSet hostnames
before constructing sqsStatusQueueURL.

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

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

// Discover the AWS account ID to construct SNS topic ARNs without an
// explicit CLI flag.
stsClient := sts.NewFromConfig(awsCfg)
identity, err := stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
if err != nil {
setupLog.Error(err, "Failed to get AWS caller identity for SNS ARN construction")
os.Exit(1)
}
awsAccountID := *identity.Account
setupLog.Info("Resolved AWS account ID for SNS topic ARNs", "accountID", awsAccountID)

snsClient := sns.NewFromConfig(awsCfg)
sqsClient := sqs.NewFromConfig(awsCfg)
publisher := snspublisher.New(snsClient, awsRegion, awsAccountID)
dynamoClient := dynamo.NewClientWithSNS(dynamoDBClient, publisher)
Comment on lines +152 to +167

@coderabbitai coderabbitai Bot Jul 28, 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.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== locate main.go =="
fd -a 'main.go$' . | sed 's#^\./##'

echo "== git diff stat =="
git diff --stat || true

echo "== inspect relevant main.go sections =="
if [ -f hyperfleet-operator/cmd/manager/main.go ]; then
  wc -l hyperfleet-operator/cmd/manager/main.go
  cat -n hyperfleet-operator/cmd/manager/main.go | sed -n '1,240p'
fi

echo "== search context usage in manager main =="
rg -n "context|Background|Deadline|WithTimeout|GetCallerIdentity|stsClient|snsClient|sqsClient|dynamodDBClient|NewClientWithSNS" hyperfleet-operator/cmd/manager/main.go || true

Repository: openshift-online/rosa-hyperfleet-api

Length of output: 11601

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== inspect relevant main.go sections with line numbers =="
if [ -f hyperfleet-operator/cmd/manager/main.go ]; then
  wc -l hyperfleet-operator/cmd/manager/main.go
  sed -n '1,240p' hyperfleet-operator/cmd/manager/main.go | nl -ba -v1
fi

Repository: openshift-online/rosa-hyperfleet-api

Length of output: 316

Add a timeout to the startup STS caller-identity call.

At hyperfleet-operator/cmd/manager/main.go:116, ctx is plain context.Background(), so the GetCalleridentity request used to construct SNS ARNs can block startup indefinitely if STS is slow or unreachable, before controllers or health endpoints come up.

Suggested bounded startup call 
-	identity, err := stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
+	stsCtx, stsCancel := context.WithTimeout(ctx, 10*time.Second)
+	defer stsCancel()
+	identity, err := stsClient.GetCallerIdentity(stsCtx, &sts.GetCallerIdentityInput{})
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// Discover the AWS account ID to construct SNS topic ARNs without an
// explicit CLI flag.
stsClient := sts.NewFromConfig(awsCfg)
identity, err := stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
if err != nil {
setupLog.Error(err, "Failed to get AWS caller identity for SNS ARN construction")
os.Exit(1)
}
awsAccountID := *identity.Account
setupLog.Info("Resolved AWS account ID for SNS topic ARNs", "accountID", awsAccountID)
snsClient := sns.NewFromConfig(awsCfg)
sqsClient := sqs.NewFromConfig(awsCfg)
publisher := snspublisher.New(snsClient, awsRegion, awsAccountID)
dynamoClient := dynamo.NewClientWithSNS(dynamoDBClient, publisher)
// Discover the AWS account ID to construct SNS topic ARNs without an
// explicit CLI flag.
stsClient := sts.NewFromConfig(awsCfg)
stsCtx, stsCancel := context.WithTimeout(ctx, 10*time.Second)
defer stsCancel()
identity, err := stsClient.GetCallerIdentity(stsCtx, &sts.GetCallerIdentityInput{})
if err != nil {
setupLog.Error(err, "Failed to get AWS caller identity for SNS ARN construction")
os.Exit(1)
}
awsAccountID := *identity.Account
setupLog.Info("Resolved AWS account ID for SNS topic ARNs", "accountID", awsAccountID)
snsClient := sns.NewFromConfig(awsCfg)
sqsClient := sqs.NewFromConfig(awsCfg)
publisher := snspublisher.New(snsClient, awsRegion, awsAccountID)
dynamoClient := dynamo.NewClientWithSNS(dynamoDBClient, publisher)
🤖 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 152 - 167, Bound the
startup STS request in the caller-identity initialization flow by creating a
timeout context before stsClient.GetCallerIdentity, use that context for the
call, and ensure the timeout cancellation is released after the request.
Preserve the existing error handling and AWS account ID resolution behavior.


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

streamMgr := statusstream.NewManager(
dynamoDBClient,
streamsClient,
mgr.GetClient(),
[]string{dynamo.TableSuffixStatusApplyDesires, dynamo.TableSuffixStatusReadDesires},
// Replace DynamoDB Streams-based statusstream.Manager with a single
// pre-provisioned SQS queue per operator replica. kube-applier publishes
// a status notification to SNS after each status write; SNS delivers to
// all per-replica queues; the operator drains its own queue only.
// 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)
go statusConsumer.Run(watchCtx)

setupLog.Info("Starting pgruntime manager")
setupLog.Info("Starting pgruntime manager",
"sqsStatusQueueURL", sqsStatusQueueURL,
)
Comment on lines +231 to +248

@coderabbitai coderabbitai Bot Jul 28, 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.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Locate and inspect the relevant main.go and SQS consumer implementation.
printf '--- main.go excerpt ---\n'
sed -n '200,270p' hyperfleet-operator/cmd/manager/main.go 2>/dev/null || true

printf '\n--- statussqsconsumer definitions/usages ---\n'
fd -a '.*sqs.*|.*status.*' hyperfleet-operator | sed 's#^\./##' | head -100
rg -n "type .*Consumer|func .*Run\\(|func .*handleMessage|Recover|runtime/debug|Add\\(|mgr\\.Add|Run\\(watchCtx\\)" hyperfleet-operator -S

printf '\n--- go.mod controller-runtime version ---\n'
rg -n "controller-runtime|sigs.k8s.io/controller-runtime" -S go.mod go.sum hyperfleet-operator/go.mod 2>/dev/null || true

Repository: openshift-online/rosa-hyperfleet-api

Length of output: 265

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Locate and inspect the relevant main.go and SQS consumer implementation.
printf '%s\n' '--- main.go excerpt ---'
sed -n '200,270p' hyperfleet-operator/cmd/manager/main.go 2>/dev/null || true

printf '%s\n' ''
printf '%s\n' '--- statussqsconsumer definitions/usages ---'
fd -a '.*sqs.*|.*status.*' hyperfleet-operator | sed 's#^\./##' | head -100
rg -n "type .*Consumer|func .*Run\\(|func .*handleMessage|Recover|runtime/debug|Add\\(|mgr\\.Add|Run\\(watchCtx\\)" hyperfleet-operator -S || true

printf '%s\n' ''
printf '%s\n' '--- go.mod controller-runtime version ---'
rg -n "controller-runtime|sigs.k8s.io/controller-runtime" -S go.mod go.sum hyperfleet-operator/go.mod 2>/dev/null || true

Repository: openshift-online/rosa-hyperfleet-api

Length of output: 4334

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '%s\n' '--- SQS consumer implementation ---'
sed -n '1,180p' hyperfleet-operator/internal/dynamo/statussqsconsumer/consumer.go

printf '%s\n' ''
printf '%s\n' '--- main.go imports / manager lifecycle section ---'
sed -n '1,120p' hyperfleet-operator/cmd/manager/main.go
sed -n '240,260p' hyperfleet-operator/cmd/manager/main.go

Repository: openshift-online/rosa-hyperfleet-api

Length of output: 10413

Attach the status SQS consumer to the manager lifecycle and isolate it with recovery.

statusConsumer.Run(watchCtx) runs outside mgr.Start(signalCtx), while watchCtx is driven by a manual context.Background()-derived cancellation. A panic in Run/handleMessage/Dispatch can take down the whole operator process. Implement a bare controller-runtime Runnable for the consumer, add it with mgr.Add(...), or at least add a recover() in the goroutine so a consumer-side bug cannot crash unrelated reconcilers.

🤖 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 231 - 248, Attach the
status SQS consumer to the controller-runtime manager lifecycle instead of
running it from a manually canceled background context. Update the setup around
statusConsumer and use a bare Runnable with Run(ctx), or add it via
mgr.Add(...), ensuring panics from Run, handleMessage, or Dispatch are recovered
so they cannot terminate the operator process.

if err := mgr.Start(signalCtx); err != nil {
setupLog.Error(err, "Failed to run manager")
os.Exit(1)
Expand Down
8 changes: 5 additions & 3 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,9 @@ require (
github.com/aws/aws-sdk-go-v2/credentials v1.19.24
github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.48
github.com/aws/aws-sdk-go-v2/service/dynamodb v1.59.0
github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.34.0
github.com/aws/aws-sdk-go-v2/service/sns v1.34.8
github.com/aws/aws-sdk-go-v2/service/sqs v1.37.8
github.com/aws/aws-sdk-go-v2/service/sts v1.43.3
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 +25,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,13 +39,13 @@ require (
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.29 // indirect
github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.29 // indirect
github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.30 // indirect
github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.34.0 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.12.6 // indirect
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.29 // indirect
github.com/aws/aws-sdk-go-v2/service/signin v1.2.0 // indirect
github.com/aws/aws-sdk-go-v2/service/sso v1.31.3 // indirect
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.6 // indirect
github.com/aws/aws-sdk-go-v2/service/sts v1.43.3 // indirect
github.com/aws/smithy-go v1.27.4 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
Expand Down
8 changes: 6 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,10 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.29 h1:DRebniUG
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.29/go.mod h1:LfRkPCD8YHDM2E5eTkos2UpwYeZnBcVarTa8L59bJHA=
github.com/aws/aws-sdk-go-v2/service/signin v1.2.0 h1:3nXpRcFwRCW8n7HgO2QGy0Dc20eQNfBuUemGQhpF8m8=
github.com/aws/aws-sdk-go-v2/service/signin v1.2.0/go.mod h1:LxYujSTLPRlp2vTtcUO/+1ilrew8ytt6SvQyOgejzFQ=
github.com/aws/aws-sdk-go-v2/service/sns v1.34.8 h1:8o7NvBkjmMaX1Cv4vztOx83aFDV6uiU8VM9pTVochng=
github.com/aws/aws-sdk-go-v2/service/sns v1.34.8/go.mod h1:FjsDzsEw55AFHFERIaeE82KqpwA2GUYhtA7yvcVCHnM=
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.31.3 h1:ey1XLTYXb9PcLt4535632o5kCGXNXEhNb620Dqwuylo=
github.com/aws/aws-sdk-go-v2/service/sso v1.31.3/go.mod h1:Lk7PlmoTYryQmyBG0EXqj5BcUbj3whXdU2s3yGI3EAc=
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.6 h1:yLr03zQE/5Eu5l3QU0Si+xMbLMbSDF2YXsigqXngs6g=
Expand Down Expand Up @@ -153,8 +157,8 @@ github.com/prometheus/procfs v0.21.0 h1:Qh/e6TlBjZf+XLLqNCqFGmCU6Kj/2Bu7kj3oAc0U
github.com/prometheus/procfs v0.21.0/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
71 changes: 67 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 @@ -7,6 +7,7 @@ import (
"errors"
"fmt"
"log/slog"
"strings"
"sync"
"time"

Expand All @@ -21,8 +22,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 All @@ -34,6 +35,16 @@ type dynamoAPI interface {
DeleteItem(ctx context.Context, params *dynamodb.DeleteItemInput, optFns ...func(*dynamodb.Options)) (*dynamodb.DeleteItemOutput, error)
}

// SNSPublisher publishes a spec change notification after a desire is written.
// Implementations must be safe for concurrent use. A nil SNSPublisher disables
// notifications (useful for tests and local development).
type SNSPublisher interface {
// Publish sends a notification for the given document to the SNS topic for
// mcName. tableSuffix identifies the table type (e.g. "-applydesires").
// Errors are best-effort: callers should log them but need not propagate.
Publish(ctx context.Context, mcName, documentID, tableSuffix string) error
}

// UpsertResult reports whether an upsert changed the item and the updateTime
// that should be used for staleness tracking. When Changed is false, UpdateTime
// reflects the existing item's time so callers never need to fabricate one.
Expand Down Expand Up @@ -62,22 +73,74 @@ type cacheEntry struct {
type Client struct {
db dynamoAPI
cache sync.Map // table/documentID → cacheEntry
sns SNSPublisher
}

var _ DesireClient = (*Client)(nil)

// NewClient returns a Client with no SNS publisher. Desire writes succeed but
// no SNS notifications are sent. Use NewClientWithSNS for production.
func NewClient(db dynamoAPI) *Client {
return &Client{db: db}
}

// NewClientWithSNS returns a Client that publishes a spec change notification
// to SNS after every desire write where the spec actually changed.
func NewClientWithSNS(db dynamoAPI, publisher SNSPublisher) *Client {
return &Client{db: db, sns: publisher}
}

// UpsertApplyDesire writes an ApplyDesire spec only when content has changed.
// If the spec changed and an SNSPublisher is configured, it publishes a
// notification so kube-applier learns about the change without polling Streams.
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
}
if result.Changed {
c.publishNotification(ctx, specsPrefix, desire.DocumentID, TableSuffixApplyDesires)
}
return result, nil
}

// UpsertReadDesire writes a ReadDesire spec only when content has changed.
// If the spec changed and an SNSPublisher is configured, it publishes a
// notification so kube-applier learns about the change without polling Streams.
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
}
if result.Changed {
c.publishNotification(ctx, specsPrefix, desire.DocumentID, TableSuffixReadDesires)
}
return result, nil
}

// publishNotification sends an SNS notification for a changed desire write.
// It is a no-op when no SNSPublisher is configured. Errors are logged but not
// propagated — kube-applier's 5-minute safety-net poll covers missed events.
func (c *Client) publishNotification(ctx context.Context, specsPrefix, documentID, tableSuffix string) {
if c.sns == nil {
return
}
mcName := mcNameFromPrefix(specsPrefix)
if err := c.sns.Publish(ctx, mcName, documentID, tableSuffix); err != nil {
slog.Error("Failed to publish desire change notification to SNS",
"mcName", mcName,
"documentID", documentID,
"tableSuffix", tableSuffix,
"error", err,
)
}
}
Comment on lines +121 to +137

@coderabbitai coderabbitai Bot Jul 28, 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.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf 'Files matching client.go:\n'
fd -a 'client\.go$' . || true

printf '\nLocate target file and inspect:\n'
if [ -f "hyperfleet-operator/internal/dynamo/client.go" ]; then
  wc -l hyperfleet-operator/internal/dynamo/client.go
  ast-grep outline hyperfleet-operator/internal/dynamo/client.go || true
  printf '\nRelevant lines 1-220:\n'
  sed -n '1,220p' hyperfleet-operator/internal/dynamo/client.go | nl -ba
else
  printf 'target file not found\n'
fi

printf '\nSearch SNSPublisher/Publish usages:\n'
rg -n "type SNSPublisher|Publish\\(|publishNotification|UpsertApplyDesire|UpsertReadDesire|WithTimeout|Context\\.Func" hyperfleet-operator/internal/dynamo -S || true

Repository: openshift-online/rosa-hyperfleet-api

Length of output: 3055

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf 'Relevant client.go lines:\n'
sed -n '1,180p' hyperfleet-operator/internal/dynamo/client.go

printf '\nSearch SNSPublisher/Publish/upsert calls in repository:\n'
rg -n "type SNSPublisher|interface\\s*\\{[^}]*Publish|Publish\\(|publishNotification\\(|UpsertApplyDesire\\(|UpsertReadDesire\\(|NewClientWithSNS|NewClient\\(" hyperfleet-operator -S || true

python3 - <<'PY'
from pathlib import Path
src = Path("hyperfleet-operator/internal/dynamo/client.go").read_text()
for name in ["func (c *Client) publishNotification", "func (c *Client) UpsertApplyDesire", "func (c *Client) UpsertReadDesire", "func (c *Client) upsertDesire"]:
    idx = src.find(name)
    print(f"\n--- {name} offset {idx} ---")
    if idx >= 0:
        line = src[:idx].count("\n") + 1
        print("start line:", line)
PY

Repository: openshift-online/rosa-hyperfleet-api

Length of output: 13677

Bound SNS publish latency before returning from a desire upsert.

UpsertApplyDesire and UpsertReadDesire call publishNotification with the caller’s ctx, and the SNS publisher forwards that context directly. A degraded SNS endpoint can therefore add latency/backpressure to reconciliation waits on the same contexts; publish the notification with its own short timeout/deadline so this best-effort path cannot stall the desire write.

🕐 Suggested bound on publish latency 
 func (c *Client) publishNotification(ctx context.Context, specsPrefix, documentID, tableSuffix string) {
 	if c.sns == nil {
 		return
 	}
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
 	mcName := mcNameFromPrefix(specsPrefix)
 	if err := c.sns.Publish(ctx, mcName, documentID, tableSuffix); err != nil {
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// publishNotification sends an SNS notification for a changed desire write.
// It is a no-op when no SNSPublisher is configured. Errors are logged but not
// propagated — kube-applier's 5-minute safety-net poll covers missed events.
func (c *Client) publishNotification(ctx context.Context, specsPrefix, documentID, tableSuffix string) {
if c.sns == nil {
return
}
mcName := mcNameFromPrefix(specsPrefix)
if err := c.sns.Publish(ctx, mcName, documentID, tableSuffix); err != nil {
slog.Error("Failed to publish desire change notification to SNS",
"mcName", mcName,
"documentID", documentID,
"tableSuffix", tableSuffix,
"error", err,
)
}
}
// publishNotification sends an SNS notification for a changed desire write.
// It is a no-op when no SNSPublisher is configured. Errors are logged but not
// propagated — kube-applier's 5-minute safety-net poll covers missed events.
func (c *Client) publishNotification(ctx context.Context, specsPrefix, documentID, tableSuffix string) {
if c.sns == nil {
return
}
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
mcName := mcNameFromPrefix(specsPrefix)
if err := c.sns.Publish(ctx, mcName, documentID, tableSuffix); err != nil {
slog.Error("Failed to publish desire change notification to SNS",
"mcName", mcName,
"documentID", documentID,
"tableSuffix", tableSuffix,
"error", err,
)
}
}
🤖 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/internal/dynamo/client.go` around lines 121 - 137, Update
publishNotification to derive a short, independent timeout context for
c.sns.Publish instead of forwarding the caller’s ctx, ensuring SNS publishing
cannot stall UpsertApplyDesire or UpsertReadDesire. Use the timeout context only
for the best-effort publish, preserve the existing no-op and error logging
behavior, and ensure its cancellation is released.


// mcNameFromPrefix strips the "-specs" suffix from a specsPrefix to recover the
// management cluster name. E.g. "eph-45df5708-mc01-specs" → "eph-45df5708-mc01",
// "mc01-specs" → "mc01". SpecsPrefix is the inverse: SpecsPrefix(mc) = mc+"-specs".
func mcNameFromPrefix(specsPrefix string) string {
return strings.TrimSuffix(specsPrefix, "-specs")
}

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