summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--common/constants/default.go10
-rw-r--r--edge/pkg/edged/edged.go9
-rw-r--r--edge/pkg/metamanager/client/pod.go4
-rw-r--r--edge/pkg/metamanager/dao/meta.go1
-rw-r--r--keadm/cmd/keadm/app/cmd/common/content.go4
-rw-r--r--keadm/cmd/keadm/app/cmd/common/types.go8
-rw-r--r--keadm/cmd/keadm/app/cmd/config.go4
-rw-r--r--keadm/cmd/keadm/app/cmd/edge/join.go2
-rw-r--r--keadm/cmd/keadm/app/cmd/edge/join_others.go8
-rw-r--r--manifests/charts/cloudcore/templates/daemonset_mosquitto.yaml46
-rw-r--r--manifests/charts/cloudcore/values.yaml27
-rw-r--r--manifests/profiles/version.yaml27
-rwxr-xr-xtests/scripts/keadm_e2e.sh5
13 files changed, 143 insertions, 12 deletions
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
}
generated by cgit v1.2.3 (git 2.52.0) at 2026-08-20 01:18:04 +0000