🐛 should send the header immediately - #194
Conversation
WalkthroughPre-generates and emits a subscription-id header before registering subscribers in the gRPC broker; adds registerSubscriber and unregister methods; refactors Subscribe to register a handler that forwards CloudEvents via an internal channel; adds tests for header timing/reconnect/concurrency; updates client header handling and gRPC auth error types. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
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: 1
🤖 Fix all issues with AI agents
In `@pkg/cloudevents/server/grpc/broker_test.go`:
- Around line 302-307: The error reporting loop uses the loop index i which is
not the actual client ID because errors arrive out-of-order; change errCh from
chan error to a channel that carries both clientID and error (e.g., a struct
{clientID int; err error} or a tuple), update the client goroutine(s) that send
to errCh to send their clientID along with the error, then in the receiver loop
read the result (e.g., res := <-errCh) and use res.clientID when logging
failures instead of i; keep numClients and the loop bounds the same.
🧹 Nitpick comments (2)
pkg/cloudevents/server/grpc/broker.go (1)
127-149: TheregisterSubscribermethod always returnsnil- consider simplifying.The method signature returns
error, but the implementation always returnsnilat line 148. This makes the error check at lines 224-226 dead code currently. Either:
- Simplify to return nothing if no error conditions are expected, or
- Keep as-is for future-proofing if error conditions might be added later
Given this is a new internal helper, the current defensive pattern is acceptable for extensibility.
pkg/cloudevents/server/grpc/broker_test.go (1)
139-166: Consider extracting common test setup to a helper function.The server/broker/listener setup code (approximately lines 140-166) is duplicated across all four new tests and the existing
TestServer. Extracting this to a helper would reduce duplication and make tests easier to maintain.Example helper extraction
func setupTestBroker(t *testing.T, ctx context.Context) (*GRPCBroker, *testService, net.Listener, func()) { t.Helper() grpcServer := grpc.NewServer() broker := NewGRPCBroker(NewBrokerOptions()) pbv1.RegisterCloudEventServiceServer(grpcServer, broker) svc := &testService{evts: make(map[string]*cloudevents.Event)} broker.RegisterService(ctx, dataType, svc) lis, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("failed to listen: %v", err) } go func() { if err := grpcServer.Serve(lis); err != nil { t.Errorf("failed to serve: %v", err) } }() cleanup := func() { grpcServer.GracefulStop() _ = lis.Close() } return broker, svc, lis, cleanup }
| // Wait for all clients to complete | ||
| for i := 0; i < numClients; i++ { | ||
| if err := <-errCh; err != nil { | ||
| t.Errorf("client %d failed: %v", i, err) | ||
| } | ||
| } |
There was a problem hiding this comment.
Misleading client ID in error message.
The variable i in the error message is the loop counter for receiving from errCh, not the actual clientID that failed. Since errors arrive in non-deterministic order, this could report the wrong client number.
Proposed fix to track actual client ID
+type clientResult struct {
+ clientID int
+ err error
+}
+
// Create and subscribe multiple clients concurrently
numClients := 10
-errCh := make(chan error, numClients)
+resultCh := make(chan clientResult, numClients)
for i := 0; i < numClients; i++ {
go func(clientID int) {
grpcClientOptions := grpccli.NewGRPCOptions()
grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()}
agentOption := grpcv2.NewAgentOptions(grpcClientOptions, "cluster1", "agent1", dataType)
if err := agentOption.CloudEventsTransport.Connect(ctx); err != nil {
- errCh <- err
+ resultCh <- clientResult{clientID, err}
return
}
if err := agentOption.CloudEventsTransport.Subscribe(ctx); err != nil {
- errCh <- err
+ resultCh <- clientResult{clientID, err}
return
}
- errCh <- nil
+ resultCh <- clientResult{clientID, nil}
}(i)
}
// Wait for all clients to complete
for i := 0; i < numClients; i++ {
- if err := <-errCh; err != nil {
- t.Errorf("client %d failed: %v", i, err)
+ result := <-resultCh
+ if result.err != nil {
+ t.Errorf("client %d failed: %v", result.clientID, result.err)
}
}🤖 Prompt for AI Agents
In `@pkg/cloudevents/server/grpc/broker_test.go` around lines 302 - 307, The error
reporting loop uses the loop index i which is not the actual client ID because
errors arrive out-of-order; change errCh from chan error to a channel that
carries both clientID and error (e.g., a struct {clientID int; err error} or a
tuple), update the client goroutine(s) that send to errCh to send their clientID
along with the error, then in the receiver loop read the result (e.g., res :=
<-errCh) and use res.clientID when logging failures instead of i; keep
numClients and the loop bounds the same.
|
/hold |
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/server/grpc/server.go (1)
226-240: Avoid logging full request payloads on authz denial.Logging
reqcan leak sensitive data and inflate logs. Prefer method + request type instead.🔧 Suggested change
- klog.FromContext(ctx).Error(err, "access denied", "req", req) + klog.FromContext(ctx).Error(err, "access denied", "method", info.FullMethod, "reqType", fmt.Sprintf("%T", req))
♻️ Duplicate comments (1)
pkg/cloudevents/server/grpc/broker_test.go (1)
278-307: Use actual client IDs when reporting concurrent failures.Errors can arrive out of order, so the loop index doesn’t reflect the client that failed.
🔧 Suggested change
- // Create and subscribe multiple clients concurrently + // Create and subscribe multiple clients concurrently numClients := 10 - errCh := make(chan error, numClients) + type clientResult struct { + clientID int + err error + } + resultCh := make(chan clientResult, numClients) for i := 0; i < numClients; i++ { go func(clientID int) { grpcClientOptions := grpccli.NewGRPCOptions() grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()} agentOption := grpcv2.NewAgentOptions(grpcClientOptions, "cluster1", "agent1", dataType) if err := agentOption.CloudEventsTransport.Connect(ctx); err != nil { - errCh <- err + resultCh <- clientResult{clientID: clientID, err: err} return } if err := agentOption.CloudEventsTransport.Subscribe(ctx); err != nil { - errCh <- err + resultCh <- clientResult{clientID: clientID, err: err} return } - errCh <- nil + resultCh <- clientResult{clientID: clientID, err: nil} }(i) } // Wait for all clients to complete for i := 0; i < numClients; i++ { - if err := <-errCh; err != nil { - t.Errorf("client %d failed: %v", i, err) - } + result := <-resultCh + if result.err != nil { + t.Errorf("client %d failed: %v", result.clientID, result.err) + } }
🧹 Nitpick comments (1)
pkg/server/grpc/server.go (1)
307-322: Use the stream context for denial logs.
context.TODO()drops request-scoped values; preferss.Context()for traceability.🔧 Suggested change
- klog.FromContext(context.TODO()).Error(err, "stream access denied") + klog.FromContext(ss.Context()).Error(err, "stream access denied", "method", info.FullMethod)
| if err := grpcprotocol.WritePBMessage(handlerCtx, binding.ToMessage(evt), pbEvt); err != nil { | ||
| return fmt.Errorf("failed to convert cloudevent to protobuf for resource(%s): %v", evt.ID(), err) | ||
| } | ||
|
|
There was a problem hiding this comment.
should new a handlerLogger from handleCtx
| return handler(ctx, req) | ||
| case authz.DecisionDeny: | ||
| return nil, fmt.Errorf("access denied: %v", err) | ||
| klog.FromContext(ctx).Error(err, "access denied", "req", req) |
There was a problem hiding this comment.
maybe be V(4)? this can generate a lot of logs.
| return handler(srv, authorizedStream) | ||
| case authz.DecisionDeny: | ||
| return fmt.Errorf("access denied: %v", err) | ||
| klog.FromContext(context.TODO()).Error(err, "stream access denied") |
There was a problem hiding this comment.
same, and do we need log here?
| // Generate subscription ID and send header IMMEDIATELY, before any other operations | ||
| // This ensures the client receives the header as soon as possible after the stream is established | ||
| subID := uuid.NewString() | ||
| if err := subServer.SendHeader(metadata.Pairs(constants.GRPCSubscriptionIDKey, subID)); err != nil { |
There was a problem hiding this comment.
if this is an issue in grpc, let's create an issue and put link here.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@pkg/cloudevents/generic/options/v2/grpc/transport.go`:
- Around line 121-132: The Subscribe code currently calls subClient.Recv() while
holding t.mu which can block indefinitely if the server never sends an event;
change this to a bounded wait: spawn a goroutine to call subClient.Recv() (or
use Recv with a context/timeout) and select between the result and a timeout so
Subscribe does not hold t.mu waiting forever; on timeout close the stream (call
Close() / Cancel on the underlying client) and return a timeout error. Ensure
you update the logic around header.Get(constants.GRPCSubscriptionIDKey) so you
release t.mu before waiting or use a non-blocking pattern that closes the stream
on timeout to avoid stalling Close().
♻️ Duplicate comments (1)
pkg/cloudevents/server/grpc/broker_test.go (1)
278-307: Misleading client ID in concurrent error reporting.The loop index
iis not necessarily the client that failed because results arrive out of order. Track the actual client ID alongside the error to avoid misleading failures.🧪 Proposed fix
+type clientResult struct { + clientID int + err error +} + // Create and subscribe multiple clients concurrently numClients := 10 -errCh := make(chan error, numClients) +resultCh := make(chan clientResult, numClients) for i := 0; i < numClients; i++ { go func(clientID int) { grpcClientOptions := grpccli.NewGRPCOptions() grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: lis.Addr().String()} agentOption := grpcv2.NewAgentOptions(grpcClientOptions, "cluster1", "agent1", dataType) if err := agentOption.CloudEventsTransport.Connect(ctx); err != nil { - errCh <- err + resultCh <- clientResult{clientID, err} return } if err := agentOption.CloudEventsTransport.Subscribe(ctx); err != nil { - errCh <- err + resultCh <- clientResult{clientID, err} return } - errCh <- nil + resultCh <- clientResult{clientID, nil} }(i) } // Wait for all clients to complete for i := 0; i < numClients; i++ { - if err := <-errCh; err != nil { - t.Errorf("client %d failed: %v", i, err) + result := <-resultCh + if result.err != nil { + t.Errorf("client %d failed: %v", result.clientID, result.err) } }
87b0ebd to
a1760bc
Compare
Signed-off-by: Wei Liu <liuweixa@redhat.com>
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/server/grpc/server.go (1)
226-239: Fix gRPC status code semantics and unify error handling in authorization interceptors.Line 239 and 319 incorrectly return
codes.Unauthenticatedfor "no authorizer found"—this should becodes.PermissionDenied. Per gRPC spec:UNAUTHENTICATEDindicates the caller cannot be identified (missing/invalid credentials), whilePERMISSION_DENIEDindicates an authenticated caller lacks authorization. A caller with valid credentials but no matching authorizer policy is authorized (denied), not unauthenticated.Additionally, line 236 returns a bare aggregated error without a status code wrapper, while line 316 wraps it as
codes.Internal. Both should wrap authorization errors consistently—consider usingcodes.PermissionDeniedfor failed authorization checks rather thancodes.Internal(which implies an unexpected system failure).
🧹 Nitpick comments (1)
pkg/cloudevents/generic/clients/baseclient.go (1)
213-230: Consider moving backoff state to per-client instance for independent retry timing.DelayFn is a global shared backoff function used by multiple clients; while thread-safe by design, it causes all client retries to advance the same backoff progression. Moving the backoff into each
baseClientinstance provides independent backoff state per client, avoiding timing coupling between separate client reconnect and subscribe operations. This is particularly useful in multi-client scenarios where you want each client's retry behavior to be isolated.♻️ Suggested refactor (per-client backoff)
-// the reconnect backoff will stop at [5s, 1min) interval. If we don't backoff for 10min, we reset the backoff. -var DelayFn = wait.Backoff{ - Duration: 5 * time.Second, - Cap: 1 * time.Minute, - Steps: 12, // now a required argument - Factor: 5.0, - Jitter: 1.0, -}.DelayWithReset(&clock.RealClock{}, 10*time.Minute) +// backoff will stop at [5s, 1min) interval. If we don't backoff for 10min, we reset the backoff. + type baseClient struct { clientID string transport options.CloudEventTransport cloudEventsRateLimiter flowcontrol.RateLimiter receiverChan chan int subscribedChan chan struct{} subscribeChan chan struct{} connected atomic.Bool subscribed atomic.Bool + delayFn func() time.Duration } func newBaseClient(clientID string, transport options.CloudEventTransport, limit utils.EventRateLimit) *baseClient { return &baseClient{ clientID: clientID, transport: transport, cloudEventsRateLimiter: utils.NewRateLimiter(limit), subscribedChan: make(chan struct{}, 1), subscribeChan: make(chan struct{}, 1), receiverChan: make(chan int, 2), // Allow both stop and start signals to be buffered + delayFn: wait.Backoff{ + Duration: 5 * time.Second, + Cap: 1 * time.Minute, + Steps: 12, + Factor: 5.0, + Jitter: 1.0, + }.DelayWithReset(&clock.RealClock{}, 10*time.Minute), } }- <-wait.RealTimer(DelayFn()).C() + <-wait.RealTimer(c.delayFn()).C()- case <-wait.RealTimer(DelayFn()).C(): + case <-wait.RealTimer(c.delayFn()).C():
|
/approve |
|
[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 |
|
/unhold |
9e9f97a
into
open-cluster-management-io:main
2 participants
Summary
we meet the error
failed to subscribe after connection" err="expected exactly one subscription-id header, got 0"sometimes,Related issue(s)
Fixes #
Summary by CodeRabbit
Bug Fixes
Tests
✏️ Tip: You can customize this high-level summary in your review settings.