From c63cd0b5bdb0b6c0aa159dde07c0dfc3b7b7b6db Mon Sep 17 00:00:00 2001 From: WillardHu Date: Wed, 6 Dec 2023 18:21:49 +0800 Subject: Use the DaemonSet to manage the mqtt broker Signed-off-by: WillardHu --- common/constants/default.go | 10 ++++- edge/pkg/edged/edged.go | 9 ++++- edge/pkg/metamanager/client/pod.go | 4 ++ edge/pkg/metamanager/dao/meta.go | 1 + keadm/cmd/keadm/app/cmd/common/content.go | 4 +- keadm/cmd/keadm/app/cmd/common/types.go | 8 ++-- keadm/cmd/keadm/app/cmd/config.go | 4 +- keadm/cmd/keadm/app/cmd/edge/join.go | 2 +- keadm/cmd/keadm/app/cmd/edge/join_others.go | 8 ++++ .../cloudcore/templates/daemonset_mosquitto.yaml | 46 ++++++++++++++++++++++ manifests/charts/cloudcore/values.yaml | 27 +++++++++++++ manifests/profiles/version.yaml | 27 +++++++++++++ tests/scripts/keadm_e2e.sh | 5 ++- 13 files changed, 143 insertions(+), 12 deletions(-) create mode 100644 manifests/charts/cloudcore/templates/daemonset_mosquitto.yaml diff --git a/common/constants/default.go b/common/constants/default.go index 07b642816..305a38071 100644 --- a/common/constants/default.go +++ b/common/constants/default.go @@ -38,7 +38,9 @@ const ( DefaultDockershimRootDir = "/var/lib/dockershim" DefaultRuntimeType = "remote" DefaultEdgedMemoryCapacity = 7852396000 - DefaultMosquittoImage = "eclipse-mosquitto:1.6.15" + // DefaultMosquittoImage ... + // Deprecated: the mqtt broker is alreay managed by the DaemonSet in the cloud + DefaultMosquittoImage = "eclipse-mosquitto:1.6.15" // update PodSandboxImage version when bumping k8s vendor version, consistent with vendor/k8s.io/kubernetes/cmd/kubelet/app/options/container_runtime.go defaultPodSandboxImageVersion // When this value are updated, also update comments in pkg/apis/componentconfig/edgecore/v1alpha1/types.go DefaultPodSandboxImage = "kubeedge/pause:3.6" @@ -155,8 +157,12 @@ const ( EdgeNodeRoleKey = "node-role.kubernetes.io/edge" EdgeNodeRoleValue = "" + // DefaultMosquittoContainerName ... + // Deprecated: the mqtt broker is alreay managed by the DaemonSet in the cloud DefaultMosquittoContainerName = "mqtt-kubeedge" - DeployMqttContainerEnv = "DEPLOY_MQTT_CONTAINER" + // DeployMqttContainerEnv ... + // Deprecated: the mqtt broker is alreay managed by the DaemonSet in the cloud + DeployMqttContainerEnv = "DEPLOY_MQTT_CONTAINER" // EdgeHub DefaultWebSocketPort = 10000 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{ 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 6fc674690..f6eee0581 100644 --- a/keadm/cmd/keadm/app/cmd/edge/join_others.go +++ b/keadm/cmd/keadm/app/cmd/edge/join_others.go @@ -77,8 +77,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`, @@ -230,10 +235,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), diff --git a/manifests/charts/cloudcore/templates/daemonset_mosquitto.yaml b/manifests/charts/cloudcore/templates/daemonset_mosquitto.yaml new file mode 100644 index 000000000..85561309b --- /dev/null +++ b/manifests/charts/cloudcore/templates/daemonset_mosquitto.yaml @@ -0,0 +1,46 @@ +{{- if .Values.mosquitto.enable }} +kind: DaemonSet +apiVersion: apps/v1 +metadata: + name: edge-eclipse-mosquitto + {{- with .Values.mosquitto.labels }} + labels: {{- toYaml . | nindent 4 }} + {{- end }} + {{- with .Values.mosquitto.annotations }} + annotations: {{- toYaml . | nindent 4 }} + {{- end }} +spec: + selector: + matchLabels: + {{- with .Values.mosquitto.labels }} + {{- toYaml . | nindent 6 }} + {{- end }} + template: + metadata: + {{- with .Values.mosquitto.labels }} + labels: {{- toYaml . | nindent 8 }} + {{- end }} + spec: + hostNetwork: true + restartPolicy: Always + volumes: + - name: mqtt-data-path + hostPath: + path: /var/lib/kubeedge/mqtt/data + containers: + - name: edge-eclipse-mosquitto + image: {{ .Values.mosquitto.image.repository }}:{{ .Values.mosquitto.image.tag }} + imagePullPolicy: {{ .Values.mosquitto.image.pullPolicy }} + {{- with .Values.mosquitto.resources }} + resources: {{ toYaml . | nindent 10 }} + {{- end }} + volumeMounts: + - name: mqtt-data-path + mountPath: /mosquitto/data + {{- with .Values.mosquitto.affinity }} + affinity: {{ toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.mosquitto.tolerations }} + tolerations: {{ toYaml . | nindent 8 }} + {{- end }} +{{- end }} diff --git a/manifests/charts/cloudcore/values.yaml b/manifests/charts/cloudcore/values.yaml index d33dac1b6..a09482c03 100644 --- a/manifests/charts/cloudcore/values.yaml +++ b/manifests/charts/cloudcore/values.yaml @@ -139,3 +139,30 @@ controllerManager: requests: cpu: 100m memory: 25Mi + +mosquitto: + enable: true + image: + repository: eclipse-mosquitto + tag: "1.6.15" + pullPolicy: "IfNotPresent" + pullSecrets: [] + labels: + k8s-app: eclipse-mosquitto + kubeedge: eclipse-mosquitto + annotations: {} + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: node-role.kubernetes.io/edge + operator: Exists + tolerations: [] + resources: + limits: + cpu: 200m + memory: 100Mi + requests: + cpu: 100m + memory: 50Mi diff --git a/manifests/profiles/version.yaml b/manifests/profiles/version.yaml index a959dfb55..5b3deb7e1 100644 --- a/manifests/profiles/version.yaml +++ b/manifests/profiles/version.yaml @@ -132,3 +132,30 @@ controllerManager: requests: cpu: 100m memory: 25Mi + +mosquitto: + enable: true + image: + repository: eclipse-mosquitto + tag: "1.6.15" + pullPolicy: "IfNotPresent" + pullSecrets: [] + labels: + k8s-app: eclipse-mosquitto + kubeedge: eclipse-mosquitto + annotations: {} + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: node-role.kubernetes.io/edge + operator: Exists + tolerations: [] + resources: + limits: + cpu: 200m + memory: 100Mi + requests: + cpu: 100m + memory: 50Mi \ No newline at end of file diff --git a/tests/scripts/keadm_e2e.sh b/tests/scripts/keadm_e2e.sh index 6329ca7ed..cc0e188ab 100755 --- a/tests/scripts/keadm_e2e.sh +++ b/tests/scripts/keadm_e2e.sh @@ -24,7 +24,10 @@ source "${KUBEEDGE_ROOT}/hack/lib/install.sh" function cleanup() { sudo pkill edgecore || true - helm uninstall cloudcore -n kubeedge && kubectl delete ns kubeedge || true + helm uninstall cloudcore -n kubeedge || true + # The namespace cleanup timeout may occur if pods is not clearned + kubectl get pods -n kubeedge | awk '{print $1}' | grep -v NAME | xargs kubectl delete pods -n kubeedge --force --grace-period=0 || true + kubectl delete ns kubeedge --force --grace-period=0 || true kind delete cluster --name test sudo rm -rf /var/log/kubeedge /etc/kubeedge /etc/systemd/system/edgecore.service $E2E_DIR/e2e_keadm/e2e_keadm.test $E2E_DIR/config.json } -- cgit v1.2.3