🐛 fix grpc broker resync bugs - #198
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughSwaps codec resource identifiers to UIDs, centralizes resourceVersion handling, adds deletion-aware encode/decode utilities, updates store/informer flows to UID-based deletion handling, refactors gRPC broker resync to use event-derived resource versions, and adds extensive unit and integration tests across affected packages. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Suggested labels
Suggested reviewers
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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: 8
🤖 Fix all issues with AI agents
In `@pkg/cloudevents/clients/addon/v1alpha1/codec.go`:
- Around line 40-50: The event codec currently sets the resource ID using
event.Name; change it to use the UID to match other codecs: when building the
CloudEvent in pkg/cloudevents/clients/event/codec.go (the code path that calls
types.NewEventBuilder(...).WithResourceID(...)), replace event.Name with
string(event.UID) so the ExtensionResourceID extension is UID-based and
consistent with addon/cluster/work/csr/lease/serviceaccount codecs.
In `@pkg/cloudevents/clients/store/informer.go`:
- Around line 70-87: The metadata mutations
(SetDeletionTimestamp/SetFinalizers/SetGeneration) are being applied to
cachedMetaObj (the original cached object) while cachedRuntimeObj (used in
s.Update) was created earlier via utils.ToRuntimeObject, so the deep copy passed
to s.Update doesn't include the changes; fix by applying the mutations to a deep
copy or by creating cachedRuntimeObj after performing the
SetDeletionTimestamp/SetFinalizers/SetGeneration changes on the meta accessor,
e.g. obtain meta.Accessor(cachedResource), perform the three Set* calls, then
call utils.ToRuntimeObject to produce cachedRuntimeObj and pass that into
s.Update so the updated metadata is persisted and emitted.
In `@pkg/cloudevents/clients/store/simplestore.go`:
- Around line 62-85: The branch that handles deletion-with-finalizers returns
early when newMetaObj has finalizers, which prevents the cached object's
DeletionTimestamp/Finalizers from being updated; instead, after finding
cachedResource via s.findObjByUID and obtaining cachedMetaObj (meta.Accessor)
and cachedRuntimeObj (utils.ToRuntimeObject), copy the deletion timestamp and
finalizers from newMetaObj into cachedMetaObj and persist the change with
s.Update(cachedRuntimeObj) (falling back to returning nil if cached resource
doesn't exist), mirroring the behavior used by the informer update logic rather
than silently discarding the update.
In `@pkg/cloudevents/clients/utils/decode.go`:
- Around line 47-50: The error return in Decode (when calling evt.DataAs(obj))
has a typo ("even" → "event") and creates a new shadowed zero value `var zero
T`; fix by replacing the error string with "failed to unmarshal event: %v" (or
similar) and remove the local `var zero T` so the function returns the
already-declared zero value `zero` from the top of the function; update the
fmt.Errorf call in the block containing `evt.DataAs(obj)` to use the corrected
message and the existing `zero` identifier.
In `@pkg/cloudevents/generic/utils/resourceversion.go`:
- Line 9: The import alias in resourceversion.go is misspelled as workpaylaod;
change the import alias to workpayload and update all usages in this file that
reference workpaylaod. Locate the import line that declares workpaylaod
"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload" and
rename it to workpayload, then replace every occurrence of workpaylaod.<symbol>
with workpayload.<symbol> (e.g., any functions, types, or constants referenced
via the alias) to keep identifiers consistent.
- Around line 48-52: GetResourceVersionFromEvent currently uses
cloudeventstypes.ToInteger (int32) to read the extension
types.ExtensionResourceVersion which causes truncation for generations >
MaxInt32; update GetResourceVersionFromEvent to read the extension value as a
full int64 (or parse it as int64) instead of using ToInteger — e.g., use a
ToInt64/ToInteger64 helper if available, or fetch
evt.Extensions()[types.ExtensionResourceVersion] as an interface and
convert/parse it to int64 safely, then return that int64 and any parsing error;
ensure the change aligns with how SetResourceVersion wrote obj.GetGeneration()
(int64).
In `@pkg/cloudevents/server/grpc/broker.go`:
- Around line 376-381: Replace the hardcoded "source" literal passed to
types.NewEventBuilder with the actual source from the incoming event by calling
evt.Source(); specifically update the call to types.NewEventBuilder(...) in the
broker.go block that builds the deletion event (the NewEventBuilder call that
currently uses "source" and respEventType) to use evt.Source() so the
constructed event (evt :=
types.NewEventBuilder(...).WithResourceID(...).WithClusterName(...).WithDeletionTimestamp(...).NewEvent())
carries the correct source metadata.
- Around line 331-333: The loop mutates events in-place via
evt.SetType(respEventType.String()), which alters the original pointers returned
by Service.List() and can persist into service caches; fix by cloning each event
before any mutation (e.g., create respEvt := evt.Clone() or equivalent) and then
call respEvt.SetType(...) and use respEvt (or &respEvt as needed) and evtLogger
derived from the cloned instance throughout the loop body so the original evt
objects remain unchanged.
🧹 Nitpick comments (8)
pkg/cloudevents/clients/store/simplestore.go (1)
75-83: Roundabout conversion:ToRuntimeObject(cachedMetaObj)on ametav1.Object.
cachedResourceis already of typeT(which satisfiesruntime.Object). You could callutils.ToRuntimeObject(cachedResource)directly instead of going throughmeta.Accessorand then back toToRuntimeObject(cachedMetaObj). Themeta.Accessorcall is unnecessary on this code path since neitherDeletionTimestamp,Finalizers, norGenerationare being set on the cached object here (unlikeinformer.go).♻️ Proposed simplification
- cachedMetaObj, err := meta.Accessor(cachedResource) - if err != nil { - return err - } - - cachedRuntimeObj, err := utils.ToRuntimeObject(cachedMetaObj) - if err != nil { - return err - } - return s.Delete(cachedRuntimeObj) + cachedRuntimeObj, err := utils.ToRuntimeObject(cachedResource) + if err != nil { + return err + } + return s.Delete(cachedRuntimeObj)pkg/cloudevents/clients/utils/decode.go (1)
39-44: Deletion-with-data path does not explicitly setDeletionTimestampfrom the extension.When a deletion event carries data (line 40-44), the code relies entirely on
DataAsto populate theDeletionTimestamp. If the serialized payload omits it (e.g., the source only signals deletion via the CloudEvent extension), the returned object won't reflect the deletion state. Consider explicitly setting the deletion timestamp from the extension after unmarshalling, as done in the no-data branch (lines 30-36), for defensive consistency.🛡️ Proposed defensive fix
if err := evt.DataAs(obj); err != nil { return zero, fmt.Errorf("failed to unmarshal event, %v", err) } + if obj.GetDeletionTimestamp() == nil { + deletionTimestamp, err := cloudeventstypes.ToTime(evt.Extensions()[types.ExtensionDeletionTimestamp]) + if err != nil { + return zero, err + } + obj.SetDeletionTimestamp(&metav1.Time{Time: deletionTimestamp}) + } + return obj, nilpkg/cloudevents/clients/utils/decode_test.go (2)
30-36:expectedErrfield is declared but unused in the test runner.The
expectedErrfield is defined in the test struct but the test loop (lines 230-236) never checks it — all error assertions are handled inside thevalidatecallback. Consider removing the field to avoid confusion.
76-93: "Deletion event with data" test does not assertDeletionTimestamp.This test validates
Name,Namespace,UID, andData, but doesn't check thatDeletionTimestampis set on the decoded object. Given that the extension signals deletion, verifying this field would strengthen coverage and would also surface the gap noted indecode.gowhere the deletion-with-data path doesn't explicitly setDeletionTimestampfrom the extension.pkg/cloudevents/generic/utils/resourceversion.go (1)
24-30: Redundant call toobj.GetResourceVersion()on line 29.
rvalready holds the value from line 24.♻️ Proposed fix
rv := obj.GetResourceVersion() if rv == "" { return } - evt.SetExtension(types.ExtensionResourceVersion, obj.GetResourceVersion()) + evt.SetExtension(types.ExtensionResourceVersion, rv)pkg/cloudevents/clients/lease/codec_test.go (1)
197-199: Nil-pointer dereference risk onHolderIdentity.If
Decodeever fails to restore theHolderIdentitypointer, dereferencingdecoded.Spec.HolderIdentity(line 197) and*decoded.Spec.HolderIdentity(line 274) will panic instead of producing a clear test failure. A nil guard would make failures easier to diagnose.Proposed guard
+ if decoded.Spec.HolderIdentity == nil { + t.Fatal("expected non-nil HolderIdentity") + } if *decoded.Spec.HolderIdentity != "holder-1" { t.Errorf("expected HolderIdentity holder-1, got %s", *decoded.Spec.HolderIdentity) }pkg/cloudevents/server/grpc/broker_resync_test.go (2)
76-81: Unchecked type assertion onExtensionResourceIDwill produce a confusing panic if the extension is missing.Line 79 does
evt.Extensions()[cetypes.ExtensionResourceID].(string)without a comma-ok check. The same pattern appears in the receive handler at lines 283 and 464. If an event lacks this extension (e.g., due to an encoding bug), the test will panic with an unhelpful stack trace rather than a clear failure message.Proposed guard for setResource
func (s *testResyncService) setResource(evt *cloudevents.Event) { s.mu.Lock() defer s.mu.Unlock() - resourceID := evt.Extensions()[cetypes.ExtensionResourceID].(string) + resourceID, ok := evt.Extensions()[cetypes.ExtensionResourceID].(string) + if !ok { + panic(fmt.Sprintf("event missing resourceid extension: %v", evt.Extensions())) + } s.resources[resourceID] = evt }
297-302: Silently discarding errors from event creation functions.Throughout the test (e.g.,
work1, _ := createWorkEvent(...)at line 297,res1, _ := createEventFn(...)at line 485), errors from event creation are ignored. If encoding fails,nilevents will be passed tosvc.setResource, causing a panic at the type assertion. Check the error to get clear test failures.Example fix
- res1, _ := createEventFn(clusterName, "res-1", 100, nil) + res1, err := createEventFn(clusterName, "res-1", 100, nil) + if err != nil { + t.Fatalf("failed to create event: %v", err) + }Also applies to: 485-490
| if !newMetaObj.GetDeletionTimestamp().IsZero() { | ||
| cachedResource, exists, err := s.findObjByUID(ctx, newMetaObj.GetUID()) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if !exists { | ||
| return nil | ||
| } | ||
|
|
||
| if len(newMetaObj.GetFinalizers()) != 0 { | ||
| return nil | ||
| } | ||
|
|
||
| cachedMetaObj, err := meta.Accessor(cachedResource) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| cachedRuntimeObj, err := utils.ToRuntimeObject(cachedMetaObj) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return s.Delete(cachedRuntimeObj) | ||
| } |
There was a problem hiding this comment.
Deletion-with-finalizers silently discards state updates, unlike informer.go.
When newMetaObj has finalizers (line 71-73), this store returns nil without updating the cached object's DeletionTimestamp/Finalizers. In contrast, informer.go (lines 82-87) updates the cached object with these fields and persists via s.Update(cachedRuntimeObj).
This means a subsequent Get from this store won't reflect that the resource is being deleted (has a DeletionTimestamp set), which could lead to stale reads. If SimpleStore consumers never inspect deletion metadata, this is fine — but the inconsistency is worth confirming as intentional.
🤖 Prompt for AI Agents
In `@pkg/cloudevents/clients/store/simplestore.go` around lines 62 - 85, The
branch that handles deletion-with-finalizers returns early when newMetaObj has
finalizers, which prevents the cached object's DeletionTimestamp/Finalizers from
being updated; instead, after finding cachedResource via s.findObjByUID and
obtaining cachedMetaObj (meta.Accessor) and cachedRuntimeObj
(utils.ToRuntimeObject), copy the deletion timestamp and finalizers from
newMetaObj into cachedMetaObj and persist the change with
s.Update(cachedRuntimeObj) (falling back to returning nil if cached resource
doesn't exist), mirroring the behavior used by the informer update logic rather
than silently discarding the update.
| func GetResourceVersionFromEvent(eventType types.CloudEventsType, evt cloudevents.Event) (int64, error) { | ||
| if eventType.CloudEventsDataType == workpaylaod.ManifestBundleEventDataType { | ||
| gen, err := cloudeventstypes.ToInteger(evt.Extensions()[types.ExtensionResourceVersion]) | ||
| return int64(gen), err | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
cloudevents sdk-go ToInteger return type
💡 Result:
In cloudevents/sdk-go (v2), types.ToInteger returns:
func ToInteger(v interface{}) (int32, error)So the value type is int32 (plus an error). [1]
Sources: [1]
🏁 Script executed:
cd /
find . -path ./tmp -prune -o -name "resourceversion.go" -type f -print 2>/dev/null | head -20Repository: open-cluster-management-io/sdk-go
Length of output: 145
🏁 Script executed:
git ls-files | grep -E "resourceversion\.go|types\.go" | head -20Repository: open-cluster-management-io/sdk-go
Length of output: 1078
🏁 Script executed:
cat -n pkg/cloudevents/generic/utils/resourceversion.goRepository: open-cluster-management-io/sdk-go
Length of output: 3115
🏁 Script executed:
wc -l pkg/cloudevents/generic/utils/resourceversion.goRepository: open-cluster-management-io/sdk-go
Length of output: 128
🏁 Script executed:
grep -r "GetGeneration" pkg/cloudevents/generic/types.goRepository: open-cluster-management-io/sdk-go
Length of output: 142
🏁 Script executed:
grep -r "type ResourceObject" pkg/cloudevents/generic/ --include="*.go"Repository: open-cluster-management-io/sdk-go
Length of output: 145
🏁 Script executed:
cat -n pkg/cloudevents/generic/interface.goRepository: open-cluster-management-io/sdk-go
Length of output: 4185
Address int32 truncation for generation values in ManifestBundleEvent.
SetResourceVersion stores obj.GetGeneration() (int64) as the extension value, but GetResourceVersionFromEvent reads it back with cloudeventstypes.ToInteger() which returns int32. If generation exceeds math.MaxInt32, the value will be silently truncated when read, despite the subsequent cast to int64 at line 51.
🤖 Prompt for AI Agents
In `@pkg/cloudevents/generic/utils/resourceversion.go` around lines 48 - 52,
GetResourceVersionFromEvent currently uses cloudeventstypes.ToInteger (int32) to
read the extension types.ExtensionResourceVersion which causes truncation for
generations > MaxInt32; update GetResourceVersionFromEvent to read the extension
value as a full int64 (or parse it as int64) instead of using ToInteger — e.g.,
use a ToInt64/ToInteger64 helper if available, or fetch
evt.Extensions()[types.ExtensionResourceVersion] as an interface and
convert/parse it to int64 safely, then return that int64 and any parsing error;
ensure the change aligns with how SetResourceVersion wrote obj.GetGeneration()
(int64).
There was a problem hiding this comment.
I think this convert is fine to us for now, I added a todo for this, we need to use the string as the resourcevesion extension in cloudevent to avoid type conversion
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check if a TODO was added to the resourceversion.go file
rg -n "TODO|FIXME" pkg/cloudevents/generic/utils/resourceversion.go
Length of output: 61
@skeeey, sounds good! I can see the TODO comment you added at line 50. Using a string for the resource version extension will indeed avoid the type conversion concerns altogether. Thanks for documenting this for future work! 👍
| // for deletion, we don't care about the resourceVersion. | ||
| evt := types.NewEventBuilder("source", respEventType). | ||
| WithResourceID(rv.ResourceID). | ||
| WithResourceVersion(rv.ResourceVersion). | ||
| WithClusterName(clusterName). | ||
| WithDeletionTimestamp(time.Now()). | ||
| NewEvent() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check how NewEventBuilder is called elsewhere to see what source values are typically used
rg -n 'NewEventBuilder\(' --type go -C 2Repository: open-cluster-management-io/sdk-go
Length of output: 12836
🏁 Script executed:
cat -n pkg/cloudevents/server/grpc/broker.go | sed -n '360,390p'Repository: open-cluster-management-io/sdk-go
Length of output: 1321
🏁 Script executed:
cat -n pkg/cloudevents/server/grpc/broker.go | sed -n '1,100p' | head -80Repository: open-cluster-management-io/sdk-go
Length of output: 3448
🏁 Script executed:
cat -n pkg/cloudevents/server/grpc/broker.go | sed -n '300,377p'Repository: open-cluster-management-io/sdk-go
Length of output: 3339
🏁 Script executed:
cat -n pkg/cloudevents/server/grpc/broker.go | sed -n '300,330p'Repository: open-cluster-management-io/sdk-go
Length of output: 1296
🏁 Script executed:
cat -n pkg/cloudevents/server/grpc/broker.go | grep -A 5 -B 5 "evt.Source\|evt.Context\|Source()"Repository: open-cluster-management-io/sdk-go
Length of output: 741
🏁 Script executed:
rg -n "evt\.Source\(\)" --type go pkg/cloudevents/server/grpc/Repository: open-cluster-management-io/sdk-go
Length of output: 59
🏁 Script executed:
# Check if there's a source extraction pattern elsewhere
rg -n "\.Source\(\)" --type go -C 2 | head -40Repository: open-cluster-management-io/sdk-go
Length of output: 2973
🏁 Script executed:
cat -n pkg/cloudevents/generic/clients/sourceclient.go | sed -n '280,300p'Repository: open-cluster-management-io/sdk-go
Length of output: 969
🏁 Script executed:
cat -n pkg/cloudevents/server/grpc/broker.go | sed -n '300,315p'Repository: open-cluster-management-io/sdk-go
Length of output: 771
Replace hardcoded "source" literal with evt.Source().
types.NewEventBuilder("source", respEventType) at line 377 uses the string literal "source" as a placeholder. The incoming event parameter has the actual source available via evt.Source(), which should be used instead to ensure proper event metadata for downstream consumers.
🤖 Prompt for AI Agents
In `@pkg/cloudevents/server/grpc/broker.go` around lines 376 - 381, Replace the
hardcoded "source" literal passed to types.NewEventBuilder with the actual
source from the incoming event by calling evt.Source(); specifically update the
call to types.NewEventBuilder(...) in the broker.go block that builds the
deletion event (the NewEventBuilder call that currently uses "source" and
respEventType) to use evt.Source() so the constructed event (evt :=
types.NewEventBuilder(...).WithResourceID(...).WithClusterName(...).WithDeletionTimestamp(...).NewEvent())
carries the correct source metadata.
6abd1b9 to
04df29b
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@pkg/cloudevents/server/grpc/broker_resync_test.go`:
- Around line 280-293: The goroutine that calls
agentOptions.CloudEventsTransport.Receive currently calls t.Errorf directly
which can race if the test ends; modify the goroutine to capture any error and
send it on a buffered error channel (e.g., errCh := make(chan error, 1)) instead
of calling t.Errorf inside the goroutine, then ensure the main test waits for
the goroutine to finish (e.g., via canceling the context with
t.Cleanup(ctxCancel) or a sync.WaitGroup) and after the goroutine returns read
from errCh and call t.Errorf/fatal in the main test goroutine; apply the same
pattern to the other occurrence in testBrokerResync so all Receive errors are
reported from the main test goroutine.
🧹 Nitpick comments (5)
pkg/cloudevents/clients/store/simplestore.go (1)
75-83: Eliminate unnecessary round-trip throughmeta.AccessorbeforeToRuntimeObject.
cachedResourcefromfindObjByUIDis already of typeT(ageneric.ResourceObject), which can be passed directly toutils.ToRuntimeObject. The detour throughmeta.Accessor(returningmetav1.Object) is redundant sinceToRuntimeObjectis defined to acceptT generic.ResourceObjectdirectly.♻️ Suggested simplification
- cachedMetaObj, err := meta.Accessor(cachedResource) - if err != nil { - return err - } - - cachedRuntimeObj, err := utils.ToRuntimeObject(cachedMetaObj) + cachedRuntimeObj, err := utils.ToRuntimeObject(cachedResource) if err != nil { return err } return s.Delete(cachedRuntimeObj)pkg/cloudevents/generic/utils/resourceversion.go (1)
24-29: Reuse the already-fetchedrvvariable instead of callingGetResourceVersion()twice.Line 29 calls
obj.GetResourceVersion()again when the value is already stored inrvfrom line 24.♻️ Proposed fix
- evt.SetExtension(types.ExtensionResourceVersion, obj.GetResourceVersion()) + evt.SetExtension(types.ExtensionResourceVersion, rv)pkg/cloudevents/server/grpc/broker.go (1)
309-313: Consider usingcloudeventstypes.ToStringfor cluster name extraction consistency.Line 313 uses
fmt.Sprintf("%s", clusterNameValue)to convert the extension to a string, while line 345 usescloudeventstypes.ToString(...)with proper error handling forresourceID. UsingToStringhere too would provide consistent error handling if the extension is unexpectedly missing or of the wrong type.Proposed fix
- clusterNameValue, err := evt.Context.GetExtension(types.ExtensionClusterName) - if err != nil { + clusterNameValue, err := evt.Context.GetExtension(types.ExtensionClusterName) + if err != nil { return err } - clusterName := fmt.Sprintf("%s", clusterNameValue) + clusterName, err := cloudeventstypes.ToString(clusterNameValue) + if err != nil { + return fmt.Errorf("failed to get clustername extension: %v", err) + }pkg/cloudevents/server/grpc/broker_resync_test.go (2)
295-336: TestBrokerResync_Work subtests clean up at the end instead of the beginning.Lines 331–335 clean up
receivedEventsandsvcat the end of each subtest. If a subtest fails viat.Fatalf(e.g., line 320), cleanup is skipped and the next subtest inherits stale state. The generictestBrokerResyncfunction correctly cleans up at the start of each subtest (e.g., lines 477–482). Align this test with the same pattern.Proposed fix (example for the first subtest)
t.Run("work: empty resync gets all resources", func(t *testing.T) { + // Clean state first + svc.clear() + drainChannel(receivedCh) + mu.Lock() + receivedEvents = make(map[string]*cloudevents.Event) + mu.Unlock() + // Setup: Create 3 resources on source work1, _ := createWorkEvent(clusterName, "work-1", 100, nil) ... if count != 3 { t.Errorf("expected 3 resources, got %d", count) } - - // Cleanup - mu.Lock() - receivedEvents = make(map[string]*cloudevents.Event) - mu.Unlock() - svc.clear() })Apply the same pattern to the other Work subtests.
297-302: Errors fromcreateWorkEventare silently discarded.If
createWorkEvent(or anycreateEventFn) returns an error, the event will benil, andsvc.setResource(work1)will panic on the unguarded type assertion at line 79 (evt.Extensions()[cetypes.ExtensionResourceID].(string)). Uset.Fatalon error for clearer test failures. The same pattern applies to the generictestBrokerResync(e.g., lines 485–490).Proposed fix
- work1, _ := createWorkEvent(clusterName, "work-1", 100, nil) - work2, _ := createWorkEvent(clusterName, "work-2", 200, nil) - work3, _ := createWorkEvent(clusterName, "work-3", 300, nil) + work1, err := createWorkEvent(clusterName, "work-1", 100, nil) + if err != nil { + t.Fatalf("failed to create work event: %v", err) + } + work2, err := createWorkEvent(clusterName, "work-2", 200, nil) + if err != nil { + t.Fatalf("failed to create work event: %v", err) + } + work3, err := createWorkEvent(clusterName, "work-3", 300, nil) + if err != nil { + t.Fatalf("failed to create work event: %v", err) + }
| go func() { | ||
| if err := agentOptions.CloudEventsTransport.Receive(ctx, func(ctx context.Context, event cloudevents.Event) { | ||
| mu.Lock() | ||
| resourceID := event.Extensions()[cetypes.ExtensionResourceID].(string) | ||
| receivedEvents[resourceID] = &event | ||
| mu.Unlock() | ||
| select { | ||
| case receivedCh <- struct{}{}: | ||
| default: | ||
| } | ||
| }); err != nil && err != context.Canceled { | ||
| t.Errorf("receive error: %v", err) | ||
| } | ||
| }() |
There was a problem hiding this comment.
t.Errorf called from a goroutine may race with test completion.
If this goroutine outlives the test (e.g., the Receive callback fires after the test returns), calling t.Errorf on a finished test panics. The same issue exists in the generic testBrokerResync at line 472. Consider capturing errors into a channel and asserting after the goroutine is joined, or using t.Cleanup to ensure the context is cancelled before the test returns.
🤖 Prompt for AI Agents
In `@pkg/cloudevents/server/grpc/broker_resync_test.go` around lines 280 - 293,
The goroutine that calls agentOptions.CloudEventsTransport.Receive currently
calls t.Errorf directly which can race if the test ends; modify the goroutine to
capture any error and send it on a buffered error channel (e.g., errCh :=
make(chan error, 1)) instead of calling t.Errorf inside the goroutine, then
ensure the main test waits for the goroutine to finish (e.g., via canceling the
context with t.Cleanup(ctxCancel) or a sync.WaitGroup) and after the goroutine
returns read from errCh and call t.Errorf/fatal in the main test goroutine;
apply the same pattern to the other occurrence in testBrokerResync so all
Receive errors are reported from the main test goroutine.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@pkg/cloudevents/clients/utils/decode_test.go`:
- Around line 219-235: The test "regular event without data" calls
DecodeWithDeletionHandling which currently may invoke evt.DataAs(obj) on an
event with empty payload; change the test to avoid undefined DataAs behavior by
either (A) setting explicit non-empty valid data on the event before calling
DecodeWithDeletionHandling, or (B) update the code path under test
(DecodeWithDeletionHandling) to check if len(evt.Data()) == 0 before calling
evt.DataAs and treat empty-data events as empty-success (mirroring deletion
handling). Refer to DecodeWithDeletionHandling, evt.Data(), and evt.DataAs when
applying the fix so the test does not rely on decoder-specific behavior.
🧹 Nitpick comments (8)
pkg/cloudevents/clients/addon/v1alpha1/codec_test.go (2)
204-249: "addon with labels and annotations" test case never asserts labels or annotations survive the round-trip.The test sets
LabelsandAnnotationson the input but only checksUID,Name, andNamespaceafter decoding. Either add assertions for the metadata maps or rename the case to avoid implying coverage that isn't there.Proposed additional assertions
if decoded.Namespace != tt.addon.Namespace { t.Errorf("Namespace mismatch: expected %s, got %s", tt.addon.Namespace, decoded.Namespace) } + if tt.addon.Labels != nil { + for k, v := range tt.addon.Labels { + if decoded.Labels[k] != v { + t.Errorf("Label %s mismatch: expected %s, got %s", k, v, decoded.Labels[k]) + } + } + } + if tt.addon.Annotations != nil { + for k, v := range tt.addon.Annotations { + if decoded.Annotations[k] != v { + t.Errorf("Annotation %s mismatch: expected %s, got %s", k, v, decoded.Annotations[k]) + } + } + }
143-183: Consider adding a decode test for a deletion event.The encode path for deletion is tested (
TestManagedClusterAddOnCodec_Encode_WithDeletionTimestamp), but there is no corresponding decode test that verifies whatDecodereturns for an event carrying adeletiontimestampextension and empty payload. Given the PR introducesDecodeWithDeletionHandling, a round-trip through a deletion event would strengthen confidence in that path.pkg/cloudevents/clients/utils/decode_test.go (2)
30-36: UnusedexpectedErrfield in test table.The
expectedErrfield is defined in the test case struct but never read in the test runner loop (lines 238–244). Thevalidatecallback handles all assertions, makingexpectedErrdead code. Either remove the field or use it in the runner to guard thevalidatecall.♻️ Suggested fix: remove unused field
cases := []struct { name string event func() *cloudevents.Event factory func() *corev1.ConfigMap validate func(t *testing.T, obj *corev1.ConfigMap, err error) - expectedErr bool }{
64-96: Consider asserting DeletionTimestamp behavior for deletion-with-data case.When the deletion extension is present but data is also present, the implementation skips setting
DeletionTimestampfrom the extension and relies entirely onDataAs. SincetestConfigMaphas noDeletionTimestamp, the decoded object won't have one either. It may be worth adding an assertion thatobj.GetDeletionTimestamp() == nilto make this behavior explicit, or testing with atestConfigMapthat does have aDeletionTimestampto verify it round-trips through the data payload.pkg/cloudevents/server/grpc/broker_resync_test.go (4)
76-81: Unchecked type assertion insetResourcewill panic on missing extension.Line 79 performs
evt.Extensions()[cetypes.ExtensionResourceID].(string)without the comma-ok form. If a test ever passes an event withoutExtensionResourceID(or with a non-string value), this panics with an unhelpful stack trace. Since this is test-only code the risk is contained, but a guard would improve debuggability.♻️ Suggested fix
func (s *testResyncService) setResource(evt *cloudevents.Event) { s.mu.Lock() defer s.mu.Unlock() - resourceID := evt.Extensions()[cetypes.ExtensionResourceID].(string) + resourceID, ok := evt.Extensions()[cetypes.ExtensionResourceID].(string) + if !ok { + panic(fmt.Sprintf("test bug: event missing or non-string %s extension", cetypes.ExtensionResourceID)) + } s.resources[resourceID] = evt }
280-293: Receive goroutine also performs an unchecked type assertion.Lines 283 and 470 do
event.Extensions()[cetypes.ExtensionResourceID].(string)inside the Receive callback. If the broker ever sends an event without this extension (e.g., an error/control event), this panics inside the goroutine, which will crash the test process rather than producing a clear failure message.♻️ Suggested fix (apply to both occurrences)
if err := agentOptions.CloudEventsTransport.Receive(ctx, func(ctx context.Context, event cloudevents.Event) { mu.Lock() - resourceID := event.Extensions()[cetypes.ExtensionResourceID].(string) + resourceID, ok := event.Extensions()[cetypes.ExtensionResourceID].(string) + if !ok { + mu.Unlock() + return + } receivedEvents[resourceID] = &event mu.Unlock()Also applies to: 467-480
491-496: Errors fromcreateEventFn/createWorkEventare silently discarded.Throughout the test, event creation errors are ignored with
_, _ :=pattern (e.g., lines 297–299, 491–493, 535, 574, 669, 722–724). If a codec change causes these to fail, the test would pass anilevent tosvc.setResource, which would then panic at the type assertion — masking the real issue.Use
t.Helper()+t.Fatalor a local helper that fails immediately on creation errors.♻️ Example helper
mustCreateEvent := func(t *testing.T, fn func(string, string, int64, *metav1.Time) (*cloudevents.Event, error), cluster, id string, ver int64, ts *metav1.Time) *cloudevents.Event { t.Helper() evt, err := fn(cluster, id, ver, ts) if err != nil { t.Fatalf("failed to create event %s: %v", id, err) } return evt }Also applies to: 297-302
608-658: Deletion detection test relies ontime.Sleepfor synchronization.Line 616 uses
time.Sleep(100 * time.Millisecond)as a workaround for potential event delivery from prior subtests. This is fragile under CI load. Since each subtest uses unique resource IDs (res-deletedis not used elsewhere), the sleep may be unnecessary. If it is needed, it indicates the drain/clear pattern isn't sufficient, which could affect other subtests too.
| { | ||
| name: "regular event without data", | ||
| event: func() *cloudevents.Event { | ||
| evt := cloudevents.NewEvent() | ||
| return &evt | ||
| }, | ||
| factory: func() *corev1.ConfigMap { | ||
| return &corev1.ConfigMap{} | ||
| }, | ||
| validate: func(t *testing.T, obj *corev1.ConfigMap, err error) { | ||
| // Empty data should be handled gracefully | ||
| if err != nil { | ||
| t.Errorf("unexpected error: %v", err) | ||
| } | ||
| }, | ||
| expectedErr: false, | ||
| }, |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
cloudevents sdk-go v2 DataAs nil empty data behavior documentation
💡 Result:
In cloudevents/sdk-go/v2, Event.DataAs(&out) is only documented as “attempts to populate the provided data object with the event payload” (the argument must be a pointer). It does not document any special-case behavior for “no data” / nil / empty payloads. [1]
What is documented is the mechanism:
Event.DataAs()decodes the event’sdatausing a registered datacodec Decoder selected from the event’sdatacontenttype. [2]datacodec.Decode(ctx, contentType, in, out)returns an error if no decoder is registered for that content type. [3]
So the “nil vs empty” outcome is effectively decoder-defined:
- For
application/json, the JSON decoder’s contract is simply that it “takesinas []byte” and decodes it intoout. Ifinis empty/nil, the underlying JSON unmarshal will fail (commonly with an “unexpected end of JSON input”-style error). [4]
Practical guidance (since this isn’t explicitly specified in docs):
- If “missing data” should be treated as “no-op”, check
len(event.Data()) == 0before callingDataAs. (Event.Data()returns the raw data bytes.) [1][2]
Sources:
[1] CloudEvents SDK-Go v2 event.Event docs (Data(), DataAs()) (pkg.go.dev)
[2] “Event data structure” (explains DataAs() uses datacodec decoders) (cloudevents.github.io)
[3] event/datacodec docs (Decode behavior when no decoder registered) (pkg.go.dev)
[4] event/datacodec/json docs (JSON decoder takes in []byte) (pkg.go.dev)
Citations:
- 1: https://pkg.go.dev/github.com/cloudevents/sdk-go/v2/event?utm_source=openai
- 2: https://cloudevents.github.io/sdk-go/event_data_structure.html?utm_source=openai
- 3: https://pkg.go.dev/github.com/cloudevents/sdk-go/v2/event/datacodec?utm_source=openai
- 4: https://pkg.go.dev/github.com/cloudevents/sdk-go/v2/event/datacodec/json?utm_source=openai
🏁 Script executed:
# Find the CloudEvents SDK version and check NewEvent() defaults
cd /tmp && git clone --depth 1 https://github.com/cloudevents/sdk-go 2>/dev/null || true
cd sdk-go && git log --oneline -1 2>/dev/null || echo "Clone check"
# Check what NewEvent() returns
rg "func NewEvent" -A 10 --type goRepository: open-cluster-management-io/sdk-go
Length of output: 2213
🏁 Script executed:
# Now check the actual test file and decode implementation
rg "DecodeWithDeletionHandling" -B 5 -A 15 --type go | head -100Repository: open-cluster-management-io/sdk-go
Length of output: 7130
🏁 Script executed:
# Check what happens with empty data and no content type
rg "DataAs" pkg/cloudevents/clients/utils/ -B 3 -A 3 --type goRepository: open-cluster-management-io/sdk-go
Length of output: 914
Ensure empty events are handled safely without relying on undefined DataAs behavior.
The test expects DecodeWithDeletionHandling to succeed on a non-deletion event with no data. However, calling evt.DataAs(obj) with empty data is not guaranteed to succeed—the behavior is decoder-dependent and not specified by the CloudEvents SDK. For example, the JSON decoder fails on empty payloads.
The implementation already uses the safe pattern if len(evt.Data()) == 0 when handling deletion events. Apply the same check to non-deletion events before calling DataAs, or explicitly set valid test data to avoid undefined behavior.
🤖 Prompt for AI Agents
In `@pkg/cloudevents/clients/utils/decode_test.go` around lines 219 - 235, The
test "regular event without data" calls DecodeWithDeletionHandling which
currently may invoke evt.DataAs(obj) on an event with empty payload; change the
test to avoid undefined DataAs behavior by either (A) setting explicit non-empty
valid data on the event before calling DecodeWithDeletionHandling, or (B) update
the code path under test (DecodeWithDeletionHandling) to check if
len(evt.Data()) == 0 before calling evt.DataAs and treat empty-data events as
empty-success (mirroring deletion handling). Refer to
DecodeWithDeletionHandling, evt.Data(), and evt.DataAs when applying the fix so
the test does not rely on decoder-specific behavior.
| func GetResourceVersionFromEvent(eventType types.CloudEventsType, evt cloudevents.Event) (int64, error) { | ||
| if eventType.CloudEventsDataType == workpayload.ManifestBundleEventDataType { | ||
| gen, err := cloudeventstypes.ToInteger(evt.Extensions()[types.ExtensionResourceVersion]) | ||
| return int64(gen), err |
There was a problem hiding this comment.
why we need to treat this differently? It seems it differs only for type conversion?
There was a problem hiding this comment.
We need this for now because the cloudevents sdk SetExtension accepts any type for extension values, but when retrieving extensions, the sdk internally converts them to a specific type—even though the return type is any
I think we need to use the string as the resourcevesion extension in cloudevent to avoid type conversion, consider maestro already use int32 as this extension, I added todo for this, we may need a new pr to handle this
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@pkg/cloudevents/generic/utils/resourceversion.go`:
- Around line 24-29: The code redundantly calls obj.GetResourceVersion() again
when you've already stored it in rv; update the call to
evt.SetExtension(types.ExtensionResourceVersion, obj.GetResourceVersion()) to
use the cached rv variable instead (i.e., pass rv), ensuring you still return
early when rv == "" as currently implemented.
Signed-off-by: Wei Liu <liuweixa@redhat.com>
84f48f5 to
8309c40
Compare
|
/assign @qiujian16 |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: qiujian16, skeeey The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
797891e
into
open-cluster-management-io:main
2 participants
Summary
Related issue(s)
Fixes #
Summary by CodeRabbit
New Features
Bug Fixes / Behavior
Tests