Skip to content

🐛 Remove cluster label #142

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

Merged
openshift-merge-bot merged 3 commits into from
Sep 15, 2025
Merged

🐛 Remove cluster label #142

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
02daddd
Remove cluster label
clyang82 Sep 12, 2025
525b638
rename cluster to consumer
clyang82 Sep 12, 2025
6700532
cannot share labels by agent and manager
clyang82 Sep 12, 2025
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
6 changes: 3 additions & 3 deletions pkg/cloudevents/generic/agentclient.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ func (c *CloudEventAgentClient[T]) Resync(ctx context.Context, source string) er
return err
}

increaseCloudEventsSentCounter(evt.Source(), source, c.clusterName, c.codec.EventDataType().String(), string(eventType.SubResource), string(eventType.Action))
increaseCloudEventsSentFromAgentCounter(evt.Source(), source, c.codec.EventDataType().String(), string(eventType.SubResource), string(eventType.Action))

return nil
}
Expand All @@ -132,7 +132,7 @@ func (c *CloudEventAgentClient[T]) Publish(ctx context.Context, eventType types.
}

originalSource, _ := cloudeventstypes.ToString(evt.Context.GetExtensions()[types.ExtensionOriginalSource])
increaseCloudEventsSentCounter(evt.Source(), originalSource, c.clusterName, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))
increaseCloudEventsSentFromAgentCounter(evt.Source(), originalSource, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))

return nil
}
Expand All @@ -153,7 +153,7 @@ func (c *CloudEventAgentClient[T]) receive(ctx context.Context, evt cloudevents.
return
}

increaseCloudEventsReceivedCounter(evt.Source(), c.clusterName, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))
increaseCloudEventsReceivedByAgentCounter(evt.Source(), eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))

if eventType.Action == types.ResyncRequestAction {
if eventType.SubResource != types.SubResourceStatus {
Expand Down
8 changes: 4 additions & 4 deletions pkg/cloudevents/generic/agentclient_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ func TestAgentResync(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())

lister := newMockResourceLister(c.resources...)
agent, err := NewCloudEventAgentClient[*mockResource](ctx, fake.NewAgentOptions(gochan.New(), nil, c.clusterName, testAgentName), lister, statusHash, newMockResourceCodec())
agent, err := NewCloudEventAgentClient(ctx, fake.NewAgentOptions(gochan.New(), nil, c.clusterName, testAgentName), lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

// start a cloudevents receiver client go to receive the event
Expand Down Expand Up @@ -129,7 +129,7 @@ func TestAgentPublish(t *testing.T) {

agentOptions := fake.NewAgentOptions(gochan.New(), nil, c.clusterName, testAgentName)
lister := newMockResourceLister()
agent, err := NewCloudEventAgentClient[*mockResource](context.TODO(), agentOptions, lister, statusHash, newMockResourceCodec())
agent, err := NewCloudEventAgentClient(context.TODO(), agentOptions, lister, statusHash, newMockResourceCodec())
require.Nil(t, err)

// start a cloudevents receiver client go to receive the event
Expand Down Expand Up @@ -283,7 +283,7 @@ func TestStatusResyncResponse(t *testing.T) {

agentOptions := fake.NewAgentOptions(gochan.New(), nil, c.clusterName, testAgentName)
lister := newMockResourceLister(c.resources...)
agent, err := NewCloudEventAgentClient[*mockResource](ctx, agentOptions, lister, statusHash, newMockResourceCodec())
agent, err := NewCloudEventAgentClient(ctx, agentOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

// start receiver
Expand Down Expand Up @@ -465,7 +465,7 @@ func TestReceiveResourceSpec(t *testing.T) {
t.Run(c.name, func(t *testing.T) {
agentOptions := fake.NewAgentOptions(gochan.New(), nil, c.clusterName, testAgentName)
lister := newMockResourceLister(c.resources...)
agent, err := NewCloudEventAgentClient[*mockResource](context.TODO(), agentOptions, lister, statusHash, newMockResourceCodec())
agent, err := NewCloudEventAgentClient(context.TODO(), agentOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

var actualEvent types.ResourceAction
Expand Down
225 changes: 148 additions & 77 deletions pkg/cloudevents/generic/metrics_collector.go
View file
Open in desktop

Large diffs are not rendered by default.

74 changes: 49 additions & 25 deletions pkg/cloudevents/generic/metrics_collector_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,8 @@ func TestCloudEventsMetrics(t *testing.T) {
}
for _, c := range cases {
// reset metrics
ResetCloudEventsMetrics()
ResetSourceCloudEventsMetrics()
ResetClientCloudEventsMetrics()
// run test
t.Run(c.name, func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
Expand All @@ -69,13 +70,13 @@ func TestCloudEventsMetrics(t *testing.T) {
// initialize source client
sourceOptions := fake.NewSourceOptions(sendReceiver, c.sourceID)
lister := newMockResourceLister([]*mockResource{}...)
source, err := NewCloudEventSourceClient[*mockResource](ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
source, err := NewCloudEventSourceClient(ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

// initialize agent client
agentOptions := fake.NewAgentOptions(sendReceiver, nil, c.clusterName, testAgentName)
agentLister := newMockResourceLister([]*mockResource{}...)
agent, err := NewCloudEventAgentClient[*mockResource](ctx, agentOptions, agentLister, statusHash, newMockResourceCodec())
agent, err := NewCloudEventAgentClient(ctx, agentOptions, agentLister, statusHash, newMockResourceCodec())
require.NoError(t, err)

// start agent subscription
Expand All @@ -99,9 +100,9 @@ func TestCloudEventsMetrics(t *testing.T) {
time.Sleep(time.Second)

// ensure metrics are updated
sentTotal := cloudeventsSentCounterMetric.WithLabelValues(c.sourceID, noneOriginalSource, c.clusterName, c.dataType.String(), string(c.subresource), string(c.action))
sentTotal := cloudeventsSentFromSourceCounterMetric.WithLabelValues(c.sourceID, noneOriginalSource, c.clusterName, c.dataType.String(), string(c.subresource), string(c.action))
require.Equal(t, len(c.resources), int(toFloat64Counter(sentTotal)))
receivedTotal := cloudeventsReceivedCounterMetric.WithLabelValues(c.sourceID, c.clusterName, c.dataType.String(), string(c.subresource), string(c.action))
receivedTotal := cloudeventsReceivedByClientCounterMetric.WithLabelValues(c.sourceID, c.dataType.String(), string(c.subresource), string(c.action))
require.Equal(t, len(c.resources), int(toFloat64Counter(receivedTotal)))

cancel()
Expand All @@ -111,7 +112,7 @@ func TestCloudEventsMetrics(t *testing.T) {

func TestReconnectMetrics(t *testing.T) {
// reset metrics
ResetCloudEventsMetrics()
ResetSourceCloudEventsMetrics()
ctx, cancel := context.WithCancel(context.Background())
Comment on lines 113 to 116

@coderabbitai coderabbitai Bot Sep 12, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

🛠️ Refactor suggestion

Reset the right registry set for reconnect test

Reconnect metric is client-side. Use client reset to avoid stale samples.

- ResetSourceCloudEventsMetrics()
+ ResetClientCloudEventsMetrics()
📝 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
func TestReconnectMetrics(t *testing.T) {
// reset metrics
ResetCloudEventsMetrics()
ResetSourceCloudEventsMetrics()
ctx, cancel := context.WithCancel(context.Background())
func TestReconnectMetrics(t *testing.T) {
// reset metrics
ResetClientCloudEventsMetrics()
ctx, cancel := context.WithCancel(context.Background())
🤖 Prompt for AI Agents 
In pkg/cloudevents/generic/metrics_collector_test.go around lines 112 to 115,
the test resets the source metrics but the reconnect metric is client-side;
replace the call to ResetSourceCloudEventsMetrics() with
ResetClientCloudEventsMetrics() (or the correct client metrics reset function)
so the test clears the client registry and avoids stale samples before creating
the client context; keep the rest of the setup intact.


originalDelayFn := DelayFn
Expand All @@ -125,7 +126,7 @@ func TestReconnectMetrics(t *testing.T) {
errChan := make(chan error)
agentOptions := fake.NewAgentOptions(gochan.New(), errChan, "cluster1", testAgentName)
agentLister := newMockResourceLister([]*mockResource{}...)
_, err := NewCloudEventAgentClient[*mockResource](ctx, agentOptions, agentLister, statusHash, newMockResourceCodec())
_, err := NewCloudEventAgentClient(ctx, agentOptions, agentLister, statusHash, newMockResourceCodec())
require.NoError(t, err)

// mimic agent disconnection by sending an error
Expand Down Expand Up @@ -174,7 +175,7 @@ func toFloat64Counter(c prometheus.Counter) float64 {
panic(fmt.Errorf("collected a non-counter metric: %s", pb))
}

func TestResyncMetrics(t *testing.T) {
func TestResyncSpecMetrics(t *testing.T) {
cases := []struct {
name string
resyncType testResyncType
Expand All @@ -194,29 +195,19 @@ func TestResyncMetrics(t *testing.T) {
},
dataType: mockEventDataType,
},
{
name: "resync status",
resyncType: testStatusResync,
clusterName: "cluster1",
sourceID: "source1",
resources: []*mockResource{
{Namespace: "cluster1", UID: kubetypes.UID("test1"), ResourceVersion: "2", Status: "test1"},
},
dataType: mockEventDataType,
},
}

for _, c := range cases {
// reset metrics
ResetCloudEventsMetrics()
ResetSourceCloudEventsMetrics()
// run test
t.Run(c.name, func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())

if c.resyncType == testSpecResync {
sourceOptions := fake.NewSourceOptions(gochan.New(), c.sourceID)
lister := newMockResourceLister(c.resources...)
source, err := NewCloudEventSourceClient[*mockResource](ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
source, err := NewCloudEventSourceClient(ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

eventType := types.CloudEventsType{
Expand All @@ -235,7 +226,7 @@ func TestResyncMetrics(t *testing.T) {
// receive resync request and publish associated resources
source.receive(ctx, evt)

receivedTotal := cloudeventsReceivedCounterMetric.WithLabelValues(c.clusterName, c.clusterName, c.dataType.String(), string(types.SubResourceSpec), string(types.ResyncRequestAction))
receivedTotal := cloudeventsReceivedBySourceCounterMetric.WithLabelValues(c.clusterName, c.clusterName, c.dataType.String(), string(types.SubResourceSpec), string(types.ResyncRequestAction))
require.Equal(t, 1, int(toFloat64Counter(receivedTotal)))

// wait 1 seconds to respond to the spec resync request
Expand All @@ -248,14 +239,47 @@ func TestResyncMetrics(t *testing.T) {
require.Greater(t, sum, 0.0)
require.Less(t, sum, 1.0)

sentTotal := cloudeventsSentCounterMetric.WithLabelValues(c.sourceID, noneOriginalSource, c.clusterName, c.dataType.String(), string(types.SubResourceSpec), string(types.ResyncResponseAction))
sentTotal := cloudeventsSentFromSourceCounterMetric.WithLabelValues(c.sourceID, noneOriginalSource, c.clusterName, c.dataType.String(), string(types.SubResourceSpec), string(types.ResyncResponseAction))
require.Equal(t, len(c.resources), int(toFloat64Counter(sentTotal)))
}

cancel()
})
}
}

func TestResyncStatusMetrics(t *testing.T) {
cases := []struct {
name string
resyncType testResyncType
clusterName string
sourceID string
resources []*mockResource
dataType types.CloudEventsDataType
}{
{
name: "resync status",
resyncType: testStatusResync,
clusterName: "cluster1",
sourceID: "source1",
resources: []*mockResource{
{Namespace: "cluster1", UID: kubetypes.UID("test1"), ResourceVersion: "2", Status: "test1"},
},
dataType: mockEventDataType,
},
}

for _, c := range cases {
// reset metrics
ResetClientCloudEventsMetrics()
// run test
t.Run(c.name, func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())

if c.resyncType == testStatusResync {
agentOptions := fake.NewAgentOptions(gochan.New(), nil, c.clusterName, testAgentName)
lister := newMockResourceLister(c.resources...)
agent, err := NewCloudEventAgentClient[*mockResource](ctx, agentOptions, lister, statusHash, newMockResourceCodec())
agent, err := NewCloudEventAgentClient(ctx, agentOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

eventType := types.CloudEventsType{
Expand All @@ -274,7 +298,7 @@ func TestResyncMetrics(t *testing.T) {
// receive resync request and publish associated resources
agent.receive(ctx, evt)

receivedTotal := cloudeventsReceivedCounterMetric.WithLabelValues(c.sourceID, c.clusterName, c.dataType.String(), string(types.SubResourceStatus), string(types.ResyncRequestAction))
receivedTotal := cloudeventsReceivedByClientCounterMetric.WithLabelValues(c.sourceID, c.dataType.String(), string(types.SubResourceStatus), string(types.ResyncRequestAction))
require.Equal(t, 1, int(toFloat64Counter(receivedTotal)))

// wait 1 seconds to respond to the resync request
Expand All @@ -287,7 +311,7 @@ func TestResyncMetrics(t *testing.T) {
require.Greater(t, sum, 0.0)
require.Less(t, sum, 1.0)

sentTotal := cloudeventsSentCounterMetric.WithLabelValues(testAgentName, noneOriginalSource, c.clusterName, c.dataType.String(), string(types.SubResourceStatus), string(types.ResyncResponseAction))
sentTotal := cloudeventsSentFromClientCounterMetric.WithLabelValues(testAgentName, noneOriginalSource, c.dataType.String(), string(types.SubResourceStatus), string(types.ResyncResponseAction))
require.Equal(t, len(c.resources), int(toFloat64Counter(sentTotal)))
}

Expand Down
8 changes: 4 additions & 4 deletions pkg/cloudevents/generic/sourceclient.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ func (c *CloudEventSourceClient[T]) Resync(ctx context.Context, clusterName stri
return err
}

increaseCloudEventsSentCounter(evt.Source(), "", clusterName, c.codec.EventDataType().String(), string(eventType.SubResource), string(eventType.Action))
increaseCloudEventsSentFromSourceCounter(evt.Source(), "", clusterName, c.codec.EventDataType().String(), string(eventType.SubResource), string(eventType.Action))

return nil
}
Expand All @@ -130,7 +130,7 @@ func (c *CloudEventSourceClient[T]) Publish(ctx context.Context, eventType types
}

clusterName := evt.Context.GetExtensions()[types.ExtensionClusterName].(string)
increaseCloudEventsSentCounter(evt.Source(), "", clusterName, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))
increaseCloudEventsSentFromSourceCounter(evt.Source(), "", clusterName, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))

return nil
}
Expand Down Expand Up @@ -158,7 +158,7 @@ func (c *CloudEventSourceClient[T]) receive(ctx context.Context, evt cloudevents
cn = ""
}

increaseCloudEventsReceivedCounter(evt.Source(), cn, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))
increaseCloudEventsReceivedBySourceCounter(evt.Source(), cn, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))

if eventType.Action == types.ResyncRequestAction {
if eventType.SubResource != types.SubResourceSpec {
Expand Down Expand Up @@ -290,7 +290,7 @@ func (c *CloudEventSourceClient[T]) respondResyncSpecRequest(
if err := c.publish(ctx, evt); err != nil {
return err
}
increaseCloudEventsSentCounter(evt.Source(), "", fmt.Sprintf("%s", clusterName), evtDataType.String(), string(eventType.SubResource), string(eventType.Action))
increaseCloudEventsSentFromSourceCounter(evt.Source(), "", fmt.Sprintf("%s", clusterName), evtDataType.String(), string(eventType.SubResource), string(eventType.Action))
}

return nil
Expand Down
8 changes: 4 additions & 4 deletions pkg/cloudevents/generic/sourceclient_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ func TestSourceResync(t *testing.T) {

sourceOptions := fake.NewSourceOptions(gochan.New(), testSourceName)
lister := newMockResourceLister(c.resources...)
source, err := NewCloudEventSourceClient[*mockResource](ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
source, err := NewCloudEventSourceClient(ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

eventChan := make(chan receiveEvent)
Expand Down Expand Up @@ -108,7 +108,7 @@ func TestSourcePublish(t *testing.T) {

sourceOptions := fake.NewSourceOptions(gochan.New(), testSourceName)
lister := newMockResourceLister()
source, err := NewCloudEventSourceClient[*mockResource](ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
source, err := NewCloudEventSourceClient(ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

eventChan := make(chan receiveEvent)
Expand Down Expand Up @@ -291,7 +291,7 @@ func TestSpecResyncResponse(t *testing.T) {

sourceOptions := fake.NewSourceOptions(gochan.New(), testSourceName)
lister := newMockResourceLister(c.resources...)
source, err := NewCloudEventSourceClient[*mockResource](ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
source, err := NewCloudEventSourceClient(ctx, sourceOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

// start receiver
Expand Down Expand Up @@ -398,7 +398,7 @@ func TestReceiveResourceStatus(t *testing.T) {
t.Run(c.name, func(t *testing.T) {
sourceOptions := fake.NewSourceOptions(gochan.New(), testSourceName)
lister := newMockResourceLister(c.resources...)
source, err := NewCloudEventSourceClient[*mockResource](context.TODO(), sourceOptions, lister, statusHash, newMockResourceCodec())
source, err := NewCloudEventSourceClient(context.TODO(), sourceOptions, lister, statusHash, newMockResourceCodec())
require.NoError(t, err)

var actualEvent types.ResourceAction
Expand Down
Loading