update sdk-go (#376)

Signed-off-by: Wei Liu <liuweixa@redhat.com>

go.mod

@@ -36,7 +36,7 @@ require (
 	k8s.io/utils v0.0.0-20240310230437-4693a0247e57
 	open-cluster-management.io/addon-framework v0.9.1-0.20240311065811-974d3377ecbd
 	open-cluster-management.io/api v0.13.0
-	open-cluster-management.io/sdk-go v0.13.1-0.20240312062935-0163292c290b
+	open-cluster-management.io/sdk-go v0.13.1-0.20240313075541-00a94671ced1
 	sigs.k8s.io/controller-runtime v0.17.2
 	sigs.k8s.io/kube-storage-version-migrator v0.0.6-0.20230721195810-5c8923c5ff96
 )

go.sum

@@ -427,8 +427,8 @@ open-cluster-management.io/addon-framework v0.9.1-0.20240311065811-974d3377ecbd
 open-cluster-management.io/addon-framework v0.9.1-0.20240311065811-974d3377ecbd/go.mod h1:ZZvGRA/zmIjIww0reatmOxcXy0Eoghnat/Opszgo2NA=
 open-cluster-management.io/api v0.13.0 h1:dlcJEZlNlE0DmSDctK2s7iWKg9l+Tgb0V78Z040nMuk=
 open-cluster-management.io/api v0.13.0/go.mod h1:CuCPEzXDvOyxBB0H1d1eSeajbHqaeGEKq9c63vQc63w=
-open-cluster-management.io/sdk-go v0.13.1-0.20240312062935-0163292c290b h1:pqW+eqzQmnpgZdZid6EeH/+Mkz9ei5ko+ubb3guEWQE=
-open-cluster-management.io/sdk-go v0.13.1-0.20240312062935-0163292c290b/go.mod h1:sq+amR9Ls9JzMP5dypvlCx4jIGfDg45gicS67Z/MnlI=
+open-cluster-management.io/sdk-go v0.13.1-0.20240313075541-00a94671ced1 h1:s3dJdi1eol+/8ek6JQuaEuoGPkK/wRyM9zowqzKHPDY=
+open-cluster-management.io/sdk-go v0.13.1-0.20240313075541-00a94671ced1/go.mod h1:sq+amR9Ls9JzMP5dypvlCx4jIGfDg45gicS67Z/MnlI=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.28.0 h1:TgtAeesdhpm2SGwkQasmbeqDo8th5wOBA5h/AjTKA4I=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.28.0/go.mod h1:VHVDI/KrK4fjnV61bE2g3sA7tiETLn8sooImelsCx3Y=
 sigs.k8s.io/controller-runtime v0.17.2 h1:FwHwD1CTUemg0pW2otk7/U5/i5m2ymzvOXdbeGOUvw0=

vendor/modules.txt

@@ -1577,7 +1577,7 @@ open-cluster-management.io/api/utils/work/v1/workapplier
 open-cluster-management.io/api/utils/work/v1/workvalidator
 open-cluster-management.io/api/work/v1
 open-cluster-management.io/api/work/v1alpha1
-# open-cluster-management.io/sdk-go v0.13.1-0.20240312062935-0163292c290b
+# open-cluster-management.io/sdk-go v0.13.1-0.20240313075541-00a94671ced1
 ## explicit; go 1.21
 open-cluster-management.io/sdk-go/pkg/apis/cluster/v1alpha1
 open-cluster-management.io/sdk-go/pkg/apis/cluster/v1beta1

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt/agentoptions.go

@@ -38,6 +38,15 @@ func NewAgentOptions(mqttOptions *MQTTOptions, clusterName, agentID string) *opt
 }
 
 func (o *mqttAgentOptions) WithContext(ctx context.Context, evtCtx cloudevents.EventContext) (context.Context, error) {
+	topic, err := getAgentPubTopic(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	if topic != nil {
+		return cloudeventscontext.WithTopic(ctx, string(*topic)), nil
+	}
+
 	eventType, err := types.ParseCloudEventsType(evtCtx.GetType())
 	if err != nil {
 		return nil, fmt.Errorf("unsupported event type %s, %v", eventType, err)

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt/options.go

@@ -20,6 +20,14 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
+type TopicKey string
+type PubTopic string
+
+const (
+	MQTT_SOURCE_PUB_TOPIC_KEY TopicKey = "mqtt_source_pub_topic"
+	MQTT_AGENT_PUB_TOPIC_KEY  TopicKey = "mqtt_agent_pub_topic"
+)
+
 // MQTTOptions holds the options that are used to build MQTT client.
 type MQTTOptions struct {
 	Topics         types.Topics
@@ -272,3 +280,47 @@ func replaceLast(str, old, new string) string {
 	}
 	return str[:last] + new + str[last+len(old):]
 }
+
+func getSourcePubTopic(ctx context.Context) (*PubTopic, error) {
+	ctxTopic := ctx.Value(MQTT_SOURCE_PUB_TOPIC_KEY)
+	if ctxTopic == nil {
+		return nil, nil
+	}
+
+	topic, ok := ctxTopic.(PubTopic)
+	if !ok {
+		return nil, fmt.Errorf("source pub topic should be a string")
+	}
+
+	if regexp.MustCompile(types.SourceEventsTopicPattern).MatchString(string(topic)) {
+		return &topic, nil
+	}
+
+	if regexp.MustCompile(types.SourceBroadcastTopicPattern).MatchString(string(topic)) {
+		return &topic, nil
+	}
+
+	return nil, fmt.Errorf("invalid source pub topic")
+}
+
+func getAgentPubTopic(ctx context.Context) (*PubTopic, error) {
+	ctxTopic := ctx.Value(MQTT_AGENT_PUB_TOPIC_KEY)
+	if ctxTopic == nil {
+		return nil, nil
+	}
+
+	topic, ok := ctxTopic.(PubTopic)
+	if !ok {
+		return nil, fmt.Errorf("agent pub topic should be a string")
+	}
+
+	if regexp.MustCompile(types.AgentEventsTopicPattern).MatchString(string(topic)) {
+		return &topic, nil
+	}
+
+	if regexp.MustCompile(types.AgentBroadcastTopicPattern).MatchString(string(topic)) {
+		return &topic, nil
+	}
+
+	return nil, fmt.Errorf("invalid agent pub topic")
+}

vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt/sourceoptions.go

@@ -36,6 +36,15 @@ func NewSourceOptions(mqttOptions *MQTTOptions, clientID, sourceID string) *opti
 }
 
 func (o *mqttSourceOptions) WithContext(ctx context.Context, evtCtx cloudevents.EventContext) (context.Context, error) {
+	topic, err := getSourcePubTopic(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	if topic != nil {
+		return cloudeventscontext.WithTopic(ctx, string(*topic)), nil
+	}
+
 	eventType, err := types.ParseCloudEventsType(evtCtx.GetType())
 	if err != nil {
 		return nil, fmt.Errorf("unsupported event type %s, %v", eventType, err)