Merge pull request #5235 from WillardHu/mqtt-daemonset

Use the DaemonSet to manage the mqtt broker

5 files changed, 19 insertions, 7 deletions

diff --git a/keadm/cmd/keadm/app/cmd/common/content.go b/keadm/cmd/keadm/app/cmd/common/content.go
index 52a598872..0b661d339 100644
--- a/keadm/cmd/keadm/app/cmd/common/content.go
+++ b/keadm/cmd/keadm/app/cmd/common/content.go
@@ -43,7 +43,9 @@ WantedBy=multi-user.target
 func GenerateServiceFile(process string, execStartCmd string, withMqtt bool) error {
 	filename := fmt.Sprintf("%s.service", process)
 
-	content := fmt.Sprintf(serviceFileTemplate, process, execStartCmd, fmt.Sprintf("%s=%t", constants.DeployMqttContainerEnv, withMqtt))
+	content := fmt.Sprintf(serviceFileTemplate, process, execStartCmd,
+		// FIXME: cleanup Environment when the static pod mqtt broker no longer needs to be compatible
+		fmt.Sprintf("%s=%t", constants.DeployMqttContainerEnv, withMqtt))
 	serviceFilePath := fmt.Sprintf("/etc/systemd/system/%s", filename)
 	return os.WriteFile(serviceFilePath, []byte(content), os.ModePerm)
 }
diff --git a/keadm/cmd/keadm/app/cmd/common/types.go b/keadm/cmd/keadm/app/cmd/common/types.go
index 71c7efb74..36a16eb80 100644
--- a/keadm/cmd/keadm/app/cmd/common/types.go
+++ b/keadm/cmd/keadm/app/cmd/common/types.go
@@ -56,9 +56,11 @@ type JoinOptions struct {
 	CertPort              string
 	CGroupDriver          string
 	Labels                []string
-	WithMQTT              bool
-	ImageRepository       string
-	HubProtocol           string
+	// WithMQTT ...
+	// Deprecated: the mqtt broker is alreay managed by the DaemonSet in the cloud
+	WithMQTT        bool
+	ImageRepository string
+	HubProtocol     string
 }
 
 type CheckOptions struct {
diff --git a/keadm/cmd/keadm/app/cmd/config.go b/keadm/cmd/keadm/app/cmd/config.go
index f70c5cd32..b0eae8bc0 100644
--- a/keadm/cmd/keadm/app/cmd/config.go
+++ b/keadm/cmd/keadm/app/cmd/config.go
@@ -158,7 +158,7 @@ func GetKubeEdgeImages(cfg *Configuration) []string {
 		images = image.CloudSet(cfg.ImageRepository, cfg.KubeEdgeVersion).List()
 	case "edge":
 		images = image.EdgeSet(&cmdcommon.JoinOptions{
-			WithMQTT: true,
+			WithMQTT: false,
 			InitBaseOptions: cmdcommon.InitBaseOptions{
 				KubeEdgeVersion: cfg.KubeEdgeVersion,
 			},
@@ -168,7 +168,7 @@ func GetKubeEdgeImages(cfg *Configuration) []string {
 		// if not specified, will return all images used by both cloud part and edge part
 		cloudSet := image.CloudSet(cfg.ImageRepository, cfg.KubeEdgeVersion)
 		edgeSet := image.EdgeSet(&cmdcommon.JoinOptions{
-			WithMQTT: true,
+			WithMQTT: false,
 			InitBaseOptions: cmdcommon.InitBaseOptions{
 				KubeEdgeVersion: cfg.KubeEdgeVersion,
 			},
diff --git a/keadm/cmd/keadm/app/cmd/edge/join.go b/keadm/cmd/keadm/app/cmd/edge/join.go
index b0a880504..ca062daac 100644
--- a/keadm/cmd/keadm/app/cmd/edge/join.go
+++ b/keadm/cmd/keadm/app/cmd/edge/join.go
@@ -105,7 +105,7 @@ func NewEdgeJoin() *cobra.Command {
 
 func newOption() *common.JoinOptions {
 	joinOptions := &common.JoinOptions{}
-	joinOptions.WithMQTT = true
+	joinOptions.WithMQTT = false
 	joinOptions.CGroupDriver = v1alpha2.CGroupDriverCGroupFS
 	joinOptions.CertPath = common.DefaultCertPath
 	joinOptions.RemoteRuntimeEndpoint = constants.DefaultRemoteRuntimeEndpoint
diff --git a/keadm/cmd/keadm/app/cmd/edge/join_others.go b/keadm/cmd/keadm/app/cmd/edge/join_others.go
index d5561c71c..28dd90a0e 100644
--- a/keadm/cmd/keadm/app/cmd/edge/join_others.go
+++ b/keadm/cmd/keadm/app/cmd/edge/join_others.go
@@ -78,8 +78,13 @@ func AddJoinOtherFlags(cmd *cobra.Command, joinOptions *common.JoinOptions) {
 	cmd.Flags().StringSliceVarP(&joinOptions.Labels, common.Labels, "l", joinOptions.Labels,
 		`Use this key to set the customized labels for node, you can input customized labels like key1=value1,key2=value2`)
 
+	// FIXME: cleanup this code when the static pod mqtt broker no longer needs to be compatible
 	cmd.Flags().BoolVar(&joinOptions.WithMQTT, "with-mqtt", joinOptions.WithMQTT,
 		`Use this key to set whether to install and start MQTT Broker by default`)
+	if err := cmd.Flags().MarkDeprecated("with-mqtt",
+		"The mqtt broker is alreay managed by the DaemonSet in the cloud"); err != nil {
+		klog.Warning("falied to mark the flag with-mqtt to deprecated, err: %v", err)
+	}
 
 	cmd.Flags().StringVar(&joinOptions.ImageRepository, common.ImageRepository, joinOptions.ImageRepository,
 		`Use this key to decide which image repository to pull images from`,
@@ -237,10 +242,13 @@ func runEdgeCore(withMqtt bool) error {
 			common.EdgeCore, common.EdgeCore)
 	} else {
 		tip = fmt.Sprintf("KubeEdge edgecore is running, For logs visit: %s%s.log", util.KubeEdgeLogPath, util.KubeEdgeBinaryName)
+
+		// FIXME: cleanup this code when the static pod mqtt broker no longer needs to be compatible
 		err := os.Setenv(constants.DeployMqttContainerEnv, strconv.FormatBool(withMqtt))
 		if err != nil {
 			klog.Errorf("Set Environment %s failed, err: %v", constants.DeployMqttContainerEnv, err)
 		}
+
 		binExec = fmt.Sprintf(
 			"%s > %skubeedge/edge/%s.log 2>&1 &",
 			filepath.Join(util.KubeEdgeUsrBinPath, util.KubeEdgeBinaryName),