fix: switch to using EMQX MQTT broker for e2e tests

Signed-off-by: Kate Goldenring <[email protected]>

Author: kate-goldenring

scripts/deploy-workloads.sh

@@ -9,16 +9,29 @@ if ! command -v kubectl &> /dev/null; then
     sudo install -o root -g root -m 0755 kubectl /usr/local/bin/kubectl;
 fi
 
+update_mqtt_workload_with_broker_cluster_ip() {
+    local dir=$1
+    echo "Waiting for emqx pod to be ready"
+    kubectl wait --for=condition=ready --timeout=20s pod/emqx
+    # The MQTT trigger cannot do DNS resolution, so we need to use the IP address of the MQTT broker
+    # Replace "EMQX_CLUSTER_IP" with the actual ClusterIP of the EMQX service
+    local cluster_ip=$(kubectl get svc emqx -o jsonpath='{.spec.clusterIP}')
+    sed -i "s/EMQX_CLUSTER_IP/$cluster_ip/g" $dir/workloads.yaml
+    echo "Updated workloads.yaml with ClusterIP: $cluster_ip"
+}
+
 # apply the workloads
 echo ">>> apply workloads"
 kubectl apply -f tests/workloads-common
 # wait for all the pods to be ready
 kubectl wait --for=condition=ready --timeout=120s pod --all
 
 if [ "$1" == "workloads-pushed-using-spin-registry-push" ]; then
+    update_mqtt_workload_with_broker_cluster_ip "tests/workloads-pushed-using-spin-registry-push"
     echo "deploying spin apps pushed to registry using 'spin registry push' command"
     kubectl apply -f tests/workloads-pushed-using-spin-registry-push
 else
+    update_mqtt_workload_with_broker_cluster_ip "tests/workloads-pushed-using-docker-build-push"
     echo "deploying spin apps pushed to registry using 'docker build && k3d image import' command"
     kubectl apply -f tests/workloads-pushed-using-docker-build-push
 fi

tests/src/integration_test.rs

@@ -174,8 +174,11 @@ mod test {
             anyhow::bail!("kubectl is not installed");
         }
 
-        // Publish a message to the MQTT broker
-        let mut mqttoptions = rumqttc::MqttOptions::new("123", "test.mosquitto.org", mqtt_port);
+        // Port forward the emqx mqtt broker
+        let forward_port = port_forward_emqx(mqtt_port).await?;
+
+        // Publish a message to the emqx broker
+        let mut mqttoptions = rumqttc::MqttOptions::new("123", "127.0.0.1", forward_port);
         mqttoptions.set_keep_alive(std::time::Duration::from_secs(1));
 
         let (client, mut eventloop) = rumqttc::AsyncClient::new(mqttoptions, 10);
@@ -216,6 +219,20 @@ mod test {
         Ok(())
     }
 
+    async fn port_forward_emqx(emqx_port: u16) -> Result<u16> {
+        let port = get_random_port()?;
+
+        println!(" >>> kubectl portforward emqx {}:{} ", port, emqx_port);
+
+        Command::new("kubectl")
+            .arg("port-forward")
+            .arg("emqx")
+            .arg(format!("{}:{}", port, emqx_port))
+            .spawn()?;
+        tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
+        Ok(port)
+    }
+
     #[tokio::test]
     async fn spin_static_assets_test() -> Result<()> {
         let host_port = 8082;

tests/workloads-common/mqtt-broker.yaml

@@ -0,0 +1,25 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: emqx
+  labels:
+    app: emqx
+spec:
+  containers:
+  - name: emqx
+    image: emqx/emqx
+    ports:
+    - containerPort: 1883
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: emqx
+spec:
+  selector:
+    app: emqx
+  ports:
+  - protocol: TCP
+    port: 1883
+    targetPort: 1883
+  type: ClusterIP

tests/workloads-pushed-using-docker-build-push/workloads.yaml

@@ -287,7 +287,7 @@ spec:
           value: containerd-shim-spin/mqtt-test-17h24d
         # The MQTT trigger cannot do DNS resolution, so we need to use the IP address of the MQTT broker
         - name: SPIN_VARIABLE_MQTT_BROKER_URI
-          value: "mqtt://test.mosquitto.org"
+          value: "mqtt://EMQX_CLUSTER_IP:1883"
 ---
 apiVersion: v1
 kind: Service

tests/workloads-pushed-using-spin-registry-push/workloads.yaml

@@ -287,7 +287,7 @@ spec:
             value: containerd-shim-spin/mqtt-test-17h24d
           # The MQTT trigger cannot do DNS resolution, so we need to use the IP address of the MQTT broker
           - name: SPIN_VARIABLE_MQTT_BROKER_URI
-            value: "mqtt://test.mosquitto.org"
+            value: "mqtt://EMQX_CLUSTER_IP:1883"
 ---
 apiVersion: v1
 kind: Service