Support Kafka (#80)

* Support Kafka Signed-off-by: clyang82 <[email protected]> * Address review comments Signed-off-by: clyang82 <[email protected]> * update ocm dependency Signed-off-by: clyang82 <[email protected]> * Upgrade sdk-go to adopt new fix Signed-off-by: clyang82 <[email protected]> * update sdk-go Signed-off-by: clyang82 <[email protected]> --------- Signed-off-by: clyang82 <[email protected]>
openshift-online · May 17, 2024 · 8ad292b · 8ad292b
1 parent 38a76b9
commit 8ad292b
Show file tree

Hide file tree

Showing 11 changed files with 171 additions and 116 deletions.
diff --git a/cmd/maestro/agent/cmd.go b/cmd/maestro/agent/cmd.go
@@ -67,9 +67,6 @@ func runAgent(cmd *cobra.Command, args []string) {
 
 	// use mqtt as the default driver
 	agentOption.MaxJSONRawLength = maxJSONRawLength
-	agentOption.WorkloadSourceDriver = "mqtt"
-	agentOption.CloudEventsClientCodecs = []string{"manifest"}
-
 	cfg := spoke.NewWorkAgentConfig(commonOptions, agentOption)
 	cmdConfig := commonOptions.CommoOpts.
 		NewControllerCommandConfig("maestro-agent", version.Get(), cfg.RunWorkloadAgent)
@@ -90,9 +87,13 @@ func addFlags(fs *pflag.FlagSet) {
 		agentOption.AppliedManifestWorkEvictionGracePeriod, "Grace period for resource eviction")
 	fs.StringVar(&commonOptions.SpokeClusterName, "consumer-name",
 		commonOptions.SpokeClusterName, "Name of the consumer")
-	// mqtt config file
-	fs.StringVar(&agentOption.WorkloadSourceConfig, "mqtt-config-file",
-		agentOption.WorkloadSourceConfig, "The config file path of mqtt broker")
-	fs.StringVar(&agentOption.CloudEventsClientID, "mqtt-client-id",
-		agentOption.CloudEventsClientID, "The ID of the mqtt client, by default it is <consumer-id>-work-agent")
+	// message broker config file
+	fs.StringVar(&agentOption.WorkloadSourceConfig, "message-broker-config-file",
+		agentOption.WorkloadSourceConfig, "The config file path of the message broker, it can be mqtt broker or kafka broker")
+	fs.StringVar(&agentOption.WorkloadSourceDriver, "message-broker-type", "mqtt", "Message broker type (default: mqtt)")
+	fs.StringVar(&agentOption.CloudEventsClientID, "agent-client-id",
+		agentOption.CloudEventsClientID, "The ID of the agent client, by default it is <consumer-id>-work-agent")
+	fs.StringSliceVar(&agentOption.CloudEventsClientCodecs, "agent-client-codecs",
+		[]string{"manifest"}, "The codecs of the agent client. The valid codecs are manifest and manifestbundle")
+
 }
diff --git a/cmd/maestro/environments/framework.go b/cmd/maestro/environments/framework.go
@@ -7,14 +7,13 @@ import (
 
 	"github.com/getsentry/sentry-go"
 	"github.com/golang/glog"
-	"github.com/spf13/pflag"
-
 	"github.com/openshift-online/maestro/pkg/client/cloudevents"
 	"github.com/openshift-online/maestro/pkg/client/ocm"
 	"github.com/openshift-online/maestro/pkg/config"
 	"github.com/openshift-online/maestro/pkg/errors"
+	"github.com/spf13/pflag"
 
-	mqttoptions "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
 )
 
 func init() {
@@ -173,12 +172,27 @@ func (e *Env) LoadClients() error {
 		glog.Infof("Using Mock CloudEvents Source Client")
 		e.Clients.CloudEventsSource = cloudevents.NewSourceClientMock(e.Services.Resources())
 	} else {
-		cloudEventsSourceOptions := mqttoptions.NewSourceOptions(e.Config.MessageBroker.MQTTOptions, e.Config.MessageBroker.ClientID, e.Config.MessageBroker.SourceID)
-		e.Clients.CloudEventsSource, err = cloudevents.NewSourceClient(cloudEventsSourceOptions, e.Services.Resources())
+
+		_, config, err := generic.NewConfigLoader(e.Config.MessageBroker.MessageBrokerType, e.Config.MessageBroker.MessageBrokerConfig).
+			LoadConfig()
 		if err != nil {
-			glog.Errorf("Unable to create CloudEvents Source client: %s", err.Error())
+			glog.Errorf("Unable to load configuration: %s", err.Error())
 			return err
 		}
+
+		cloudEventsSourceOptions, err := generic.BuildCloudEventsSourceOptions(config,
+			e.Config.MessageBroker.ClientID, e.Config.MessageBroker.SourceID)
+		if err != nil {
+			glog.Errorf("Unable to build cloudevent source options: %s", err.Error())
+			return err
+		}
+		if cloudEventsSourceOptions != nil {
+			e.Clients.CloudEventsSource, err = cloudevents.NewSourceClient(cloudEventsSourceOptions, e.Services.Resources())
+			if err != nil {
+				glog.Errorf("Unable to create CloudEvents Source client: %s", err.Error())
+				return err
+			}
+		}
 	}
 
 	return nil

diff --git a/go.mod b/go.mod
@@ -9,7 +9,7 @@ require (
 	github.com/bwmarrin/snowflake v0.3.0
 	github.com/bxcodec/faker/v3 v3.2.0
 	github.com/cespare/xxhash v1.1.0
-	github.com/cloudevents/sdk-go/v2 v2.14.0
+	github.com/cloudevents/sdk-go/v2 v2.15.3-0.20240329120647-e6a74efbacbf
 	github.com/deckarep/golang-set/v2 v2.6.0
 	github.com/docker/go-healthcheck v0.1.0
 	github.com/getsentry/sentry-go v0.20.0
@@ -23,8 +23,8 @@ require (
 	github.com/jinzhu/inflection v1.0.0
 	github.com/lib/pq v1.10.7
 	github.com/mendsley/gojwk v0.0.0-20141217222730-4d5ec6e58103
-	github.com/onsi/ginkgo/v2 v2.15.0
-	github.com/onsi/gomega v1.31.1
+	github.com/onsi/ginkgo/v2 v2.17.1
+	github.com/onsi/gomega v1.32.0
 	github.com/openshift-online/ocm-sdk-go v0.1.334
 	github.com/prometheus/client_golang v1.18.0
 	github.com/segmentio/ksuid v1.0.2
@@ -37,13 +37,13 @@ require (
 	gorm.io/datatypes v1.2.0
 	gorm.io/driver/postgres v1.5.0
 	gorm.io/gorm v1.24.7-0.20230306060331-85eaf9eeda11
-	k8s.io/apimachinery v0.29.2
-	k8s.io/client-go v0.29.2
-	k8s.io/component-base v0.29.1
+	k8s.io/apimachinery v0.29.4
+	k8s.io/client-go v0.29.4
+	k8s.io/component-base v0.29.3
 	k8s.io/klog/v2 v2.120.1
-	open-cluster-management.io/api v0.13.0
-	open-cluster-management.io/ocm v0.13.1-0.20240313094829-1c0c0156e780
-	open-cluster-management.io/sdk-go v0.13.1-0.20240321032811-7dbdd1b5c63d
+	open-cluster-management.io/api v0.13.1-0.20240506072237-800b00d9f0db
+	open-cluster-management.io/ocm v0.13.1-0.20240514020334-4117a4b3027f
+	open-cluster-management.io/sdk-go v0.13.1-0.20240516092635-a00a7ab51fd2
 )
 
 require (
@@ -56,15 +56,17 @@ require (
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cloudevents/sdk-go/protocol/kafka_confluent/v2 v2.0.0-20240413090539-7fef29478991 // indirect
 	github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-20231030012137-0836a524e995 // indirect
+	github.com/confluentinc/confluent-kafka-go/v2 v2.3.0 // indirect
 	github.com/coreos/go-semver v0.3.1 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible // indirect
 	github.com/docker/distribution v2.8.1+incompatible // indirect
 	github.com/eclipse/paho.golang v0.11.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
-	github.com/evanphx/json-patch v5.7.0+incompatible // indirect
+	github.com/evanphx/json-patch v5.9.0+incompatible // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
@@ -127,17 +129,17 @@ require (
 	go.opentelemetry.io/otel/trace v1.19.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.26.0 // indirect
-	golang.org/x/crypto v0.19.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1 // indirect
-	golang.org/x/net v0.21.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
 	golang.org/x/oauth2 v0.16.0 // indirect
 	golang.org/x/sync v0.6.0 // indirect
-	golang.org/x/sys v0.17.0 // indirect
-	golang.org/x/term v0.17.0 // indirect
+	golang.org/x/sys v0.18.0 // indirect
+	golang.org/x/term v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
-	golang.org/x/tools v0.16.1 // indirect
+	golang.org/x/tools v0.17.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 // indirect
@@ -147,15 +149,15 @@ require (
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	gorm.io/driver/mysql v1.4.7 // indirect
-	k8s.io/api v0.29.2 // indirect
-	k8s.io/apiextensions-apiserver v0.29.0 // indirect
-	k8s.io/apiserver v0.29.0 // indirect
-	k8s.io/kms v0.29.0 // indirect
-	k8s.io/kube-aggregator v0.29.0 // indirect
+	k8s.io/api v0.29.4 // indirect
+	k8s.io/apiextensions-apiserver v0.29.3 // indirect
+	k8s.io/apiserver v0.29.3 // indirect
+	k8s.io/kms v0.29.3 // indirect
+	k8s.io/kube-aggregator v0.29.3 // indirect
 	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
 	k8s.io/utils v0.0.0-20240310230437-4693a0247e57 // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.28.0 // indirect
-	sigs.k8s.io/controller-runtime v0.17.2 // indirect
+	sigs.k8s.io/controller-runtime v0.17.3 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/kube-storage-version-migrator v0.0.6-0.20230721195810-5c8923c5ff96 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect