Merge pull request #5235 from WillardHu/mqtt-daemonset

Use the DaemonSet to manage the mqtt broker

3 files changed, 12 insertions, 2 deletions

diff --git a/edge/pkg/edged/edged.go b/edge/pkg/edged/edged.go
index 6b2ceee08..11521841d 100644
--- a/edge/pkg/edged/edged.go
+++ b/edge/pkg/edged/edged.go
@@ -121,13 +121,18 @@ func (e *edged) Enable() bool {
 func (e *edged) Start() {
 	klog.Info("Starting edged...")
 
+	// FIXME: cleanup this code when the static pod mqtt broker no longer needs to be compatible
 	// edged saves the data of mqtt container in sqlite3 and starts it. This is a temporary workaround and will be modified in v1.15.
 	withMqtt, err := strconv.ParseBool(os.Getenv(constants.DeployMqttContainerEnv))
 	if err == nil && withMqtt {
-		err := dao.SaveMQTTMeta(e.nodeName)
-		if err != nil {
+		if err := dao.SaveMQTTMeta(e.nodeName); err != nil {
 			klog.ErrorS(err, "Start mqtt container failed")
 		}
+	} else {
+		// Delete a not exists key does not return an error
+		if err := dao.DeleteMetaByKey(fmt.Sprintf("default/pod/%s", constants.DefaultMosquittoContainerName)); err != nil {
+			klog.ErrorS(err, "delete mqtt container failed")
+		}
 	}
 
 	go func() {
diff --git a/edge/pkg/metamanager/client/pod.go b/edge/pkg/metamanager/client/pod.go
index e21296097..7ebc38fad 100644
--- a/edge/pkg/metamanager/client/pod.go
+++ b/edge/pkg/metamanager/client/pod.go
@@ -110,9 +110,12 @@ func (c *pods) Get(name string) (*corev1.Pod, error) {
 
 func (c *pods) Patch(name string, patchBytes []byte) (*corev1.Pod, error) {
 	resource := fmt.Sprintf("%s/%s/%s", c.namespace, model.ResourceTypePodPatch, name)
+
+	// FIXME: cleanup this code when the static pod mqtt broker no longer needs to be compatible
 	if name == constants.DefaultMosquittoContainerName {
 		return handleMqttMeta()
 	}
+
 	podMsg := message.BuildMsg(modules.MetaGroup, "", modules.EdgedModuleName, resource, model.PatchOperation, string(patchBytes))
 	resp, err := c.send.SendSync(podMsg)
 	if err != nil {
@@ -189,6 +192,7 @@ func updatePodDB(resource string, pod *corev1.Pod) error {
 	return dao.InsertOrUpdate(meta)
 }
 
+// FIXME: cleanup this code when the static pod mqtt broker no longer needs to be compatible
 func handleMqttMeta() (*corev1.Pod, error) {
 	var pod corev1.Pod
 	metas, err := dao.QueryMeta("key", fmt.Sprintf("default/pod/%s", constants.DefaultMosquittoContainerName))
diff --git a/edge/pkg/metamanager/dao/meta.go b/edge/pkg/metamanager/dao/meta.go
index 42d0c98c3..112d9d1e7 100644
--- a/edge/pkg/metamanager/dao/meta.go
+++ b/edge/pkg/metamanager/dao/meta.go
@@ -120,6 +120,7 @@ func QueryAllMeta(key string, condition string) (*[]Meta, error) {
 
 // SaveMQTTMeta saves mqtt container data in sqlites
 // When egdecore starts, edged will start mqtt container
+// FIXME: cleanup this code when the static pod mqtt broker no longer needs to be compatible
 func SaveMQTTMeta(nodeName string) error {
 	flag := true
 	mqttData := coreV1.Pod{