add integration test for pubsub. (#455)

* add integration test for pubsub.

Signed-off-by: Morven Cao <lcao@redhat.com>

* address comment.

Signed-off-by: Morven Cao <lcao@redhat.com>

* fix tekton integration test.

Signed-off-by: Morven Cao <lcao@redhat.com>

* fix typo.

Signed-off-by: Morven Cao <lcao@redhat.com>

---------

Signed-off-by: Morven Cao <lcao@redhat.com>

Author: morvencao

.gitignore

@@ -47,6 +47,7 @@
 secrets/db.password
 secrets/mqtt.password
 secrets/mqtt.config
+secrets/pubsub.config
 hack/mosquitto-passwd.txt
 
 # Ignore image environment file

.tekton/integration-test.yaml

@@ -62,7 +62,9 @@ spec:
             TARGET_COMPONENT_NAME="maestro"
             REPO_URL=$(echo $SNAPSHOT | jq -r '.components[] | select(.name == "maestro").source.git.url')
             REPO_COMMIT=$(echo $SNAPSHOT | jq -r '.components[] | select(.name == "maestro").source.git.revision')
-            test_output_file=${PWD}/test_output.json
+            mqtt_test_output_file=${PWD}/mqtt_test_output.json
+            pubsub_test_output_file=${PWD}/pubsub_test_output.json
+            grpc_test_output_file=${PWD}/grpc_test_output.json
 
             # Log Vars for Tracibility
             echo "REPO_URL: $REPO_URL"
@@ -84,16 +86,38 @@ spec:
             # Generate mqtt config file
             echo '{"brokerHost":"localhost:1883","username":"","password":"","topics":{"sourceEvents":"sources/maestro/consumers/+/sourceevents","agentEvents":"sources/maestro/consumers/+/agentevents"}}' > secrets/mqtt.config
 
+            # Generate pubsub config file
+            echo '{"projectID":"maestro-test","endpoint":"localhost:8085","insecure":true,"topics":{"sourceEvents":"projects/maestro-test/topics/sourceevents","sourceBroadcast":"projects/maestro-test/topics/sourcebroadcast"},"subscriptions":{"agentEvents":"projects/maestro-test/subscriptions/agentevents-maestro","agentBroadcast":"projects/maestro-test/subscriptions/agentbroadcast-maestro"}}' > secrets/pubsub.config
+
             # Generate db.password
             echo 'foobar-bizz-buzz' > secrets/db.password
 
+            # Wait for the pubsub emulator to be ready
+            echo "[INFO] Waiting for pubsub emulator to be ready for connection"
+            timeout 5m bash -c 'until curl -s http://localhost:8085 > /dev/null 2>&1; do sleep 2s; done'
+
+            # Initialize pubsub topics and subscriptions
+            export PUBSUB_EMULATOR_HOST=localhost:8085
+            export PUBSUB_PROJECT_ID=maestro-test
+            bash hack/init-pubsub-emulator.sh
+
             # Run integration tests with JSON output
-            make test-integration integration_test_json_output="$test_output_file"
+            make test-integration \
+              mqtt_integration_test_json_output="$mqtt_test_output_file" \
+              pubsub_integration_test_json_output="$pubsub_test_output_file" \
+              grpc_integration_test_json_output="$grpc_test_output_file"
 
             # Read, process, and write output in accepted format
             # Formatting docs: https://redhat-appstudio.github.io/book/ADR/0030-tekton-results-naming-convention.html#test_output-schema-validation
-            FAILURES=$(cat $test_output_file | jq 'select(.Action == "fail")' | jq -r --slurp 'length')
-            SUCCESSES=$(cat $test_output_file | jq 'select(.Action == "pass")' | jq -r --slurp 'length')
+            # Combine results from all three broker tests
+            MQTT_FAILURES=$(cat $mqtt_test_output_file | jq 'select(.Action == "fail")' | jq -r --slurp 'length')
+            MQTT_SUCCESSES=$(cat $mqtt_test_output_file | jq 'select(.Action == "pass")' | jq -r --slurp 'length')
+            PUBSUB_FAILURES=$(cat $pubsub_test_output_file | jq 'select(.Action == "fail")' | jq -r --slurp 'length')
+            PUBSUB_SUCCESSES=$(cat $pubsub_test_output_file | jq 'select(.Action == "pass")' | jq -r --slurp 'length')
+            GRPC_FAILURES=$(cat $grpc_test_output_file | jq 'select(.Action == "fail")' | jq -r --slurp 'length')
+            GRPC_SUCCESSES=$(cat $grpc_test_output_file | jq 'select(.Action == "pass")' | jq -r --slurp 'length')
+            FAILURES=$((MQTT_FAILURES + PUBSUB_FAILURES + GRPC_FAILURES))
+            SUCCESSES=$((MQTT_SUCCESSES + PUBSUB_SUCCESSES + GRPC_SUCCESSES))
             # Hard-code warnings
             WARNINGS=0
             RESULT="$(if [[ $FAILURES == 0 ]]; then echo "SUCCESS"; else echo "FAILURE"; fi)"
@@ -129,6 +153,16 @@ spec:
             volumeMounts:
               - mountPath: /mosquitto/data
                 name: mqttdata
+          - image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
+            name: pubsub-test
+            command:
+              - gcloud
+              - beta
+              - emulators
+              - pubsub
+              - start
+              - --host-port=0.0.0.0:8085
+              - --project=maestro-test
         volumes:
           - name: pgdata
             emptyDir: {}

Makefile

@@ -59,13 +59,25 @@ mqtt_port ?= 1883
 mqtt_user ?= maestro
 mqtt_password_file ?= ${PWD}/secrets/mqtt.password
 mqtt_config_file ?= ${PWD}/secrets/mqtt.config
+mqtt_image ?= quay.io/maestro/eclipse-mosquitto:2.0.18
+
+# Pub/Sub emulator configuration
+pubsub_host ?= maestro-pubsub.$(namespace)
+pubsub_port ?= 8085
+pubsub_project_id ?= maestro-test
+pubsub_config_file ?= ${PWD}/secrets/pubsub.config
+pubsub_image ?= gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
 
 # Log verbosity level
 klog_v:=2
 
+# message driver type, mqtt, grpc or pubsub, default is mqtt.
+MESSAGE_DRIVER_TYPE ?= mqtt
+
 # Test output files
 unit_test_json_output ?= ${PWD}/unit-test-results.json
 mqtt_integration_test_json_output ?= ${PWD}/mqtt-integration-test-results.json
+pubsub_integration_test_json_output ?= ${PWD}/pubsub-integration-test-results.json
 grpc_integration_test_json_output ?= ${PWD}/grpc-integration-test-results.json
 
 # Prints a list of useful targets.
@@ -77,6 +89,13 @@ help:
 	@echo "make lint                 run golangci-lint"
 	@echo "make binary               compile binaries"
 	@echo "make install              compile binaries and install in GOPATH bin"
+	@echo "make db/setup             setup local PostgreSQL database"
+	@echo "make db/teardown          teardown local PostgreSQL database"
+	@echo "make mqtt/setup           setup local MQTT broker"
+	@echo "make mqtt/teardown        teardown local MQTT broker"
+	@echo "make pubsub/setup         setup local Pub/Sub emulator"
+	@echo "make pubsub/init          initialize Pub/Sub topics and subscriptions"
+	@echo "make pubsub/teardown      teardown local Pub/Sub emulator"
 	@echo "make run                  run the application"
 	@echo "make run/docs             run swagger and host the api spec"
 	@echo "make test                 run unit tests"
@@ -206,16 +225,21 @@ test:
 #   make test-integration TESTFLAGS="-run TestAccounts"     acts as TestAccounts* and run TestAccountsGet, TestAccountsPost, etc.
 #   make test-integration TESTFLAGS="-run TestAccountsGet"  runs TestAccountsGet
 #   make test-integration TESTFLAGS="-short"                skips long-run tests
-test-integration: test-integration-mqtt test-integration-grpc
+test-integration: test-integration-mqtt test-integration-pubsub test-integration-grpc
 .PHONY: test-integration
 
 test-integration-mqtt:
-	BROKER=mqtt MAESTRO_ENV=testing gotestsum --jsonfile-timing-events=$(mqtt_integration_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -p 1 -ldflags -s -v -timeout 1h $(TESTFLAGS) \
+	MESSAGE_DRIVER_TYPE=mqtt MESSAGE_DRIVER_CONFIG=$(PWD)/secrets/mqtt.config MAESTRO_ENV=testing gotestsum --jsonfile-timing-events=$(mqtt_integration_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -count=1 -p 1 -ldflags -s -v -timeout 1h $(TESTFLAGS) \
 			./test/integration
 .PHONY: test-integration-mqtt
 
+test-integration-pubsub:
+	MESSAGE_DRIVER_TYPE=pubsub MESSAGE_DRIVER_CONFIG=$(PWD)/secrets/pubsub.config PUBSUB_EMULATOR_HOST=localhost:$(pubsub_port) MAESTRO_ENV=testing gotestsum --jsonfile-timing-events=$(pubsub_integration_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -count=1 -p 1 -ldflags -s -v -timeout 1h $(TESTFLAGS) \
+			./test/integration
+.PHONY: test-integration-pubsub
+
 test-integration-grpc:
-	BROKER=grpc MAESTRO_ENV=testing gotestsum --jsonfile-timing-events=$(grpc_integration_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -count=1 -p 1 -ldflags -s -v -timeout 1h $(TESTFLAGS) \
+	MESSAGE_DRIVER_TYPE=grpc MAESTRO_ENV=testing gotestsum --jsonfile-timing-events=$(grpc_integration_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -count=1 -p 1 -ldflags -s -v -timeout 1h $(TESTFLAGS) \
 			./test/integration
 .PHONY: test-integration-grpc
 
@@ -230,7 +254,11 @@ generate:
 
 run: install
 	maestro migration
-	maestro server
+	@if [ "$(MESSAGE_DRIVER_TYPE)" = "grpc" ]; then \
+		maestro server --message-broker-type=$(MESSAGE_DRIVER_TYPE); \
+	else \
+		maestro server --message-broker-type=$(MESSAGE_DRIVER_TYPE) --message-broker-config-file=./secrets/$(MESSAGE_DRIVER_TYPE).config; \
+	fi
 .PHONY: run
 
 # Run Swagger and host the api docs
@@ -325,14 +353,29 @@ mqtt/prepare:
 .PHONY: mqtt/setup
 mqtt/setup: mqtt/prepare
 	@echo '{"brokerHost":"localhost:1883","username":"$(mqtt_user)","password":"$(shell cat $(mqtt_password_file))","topics":{"sourceEvents":"sources/maestro/consumers/+/sourceevents","agentEvents":"sources/maestro/consumers/+/agentevents"}}' > $(mqtt_config_file)
-	$(container_tool) run --rm -v $(shell pwd)/hack:/mosquitto/data:z $(MQTT_IMAGE) mosquitto_passwd -c -b /mosquitto/data/mosquitto-passwd.txt $(mqtt_user) $(shell cat $(mqtt_password_file))
-	$(container_tool) run --name mqtt-maestro -p 1883:1883 -v $(shell pwd)/hack/mosquitto-passwd.txt:/mosquitto/config/password.txt -v $(shell pwd)/hack/mosquitto.conf:/mosquitto/config/mosquitto.conf -d $(MQTT_IMAGE)
+	$(container_tool) run --rm -v $(shell pwd)/hack:/mosquitto/data:z $(mqtt_image) mosquitto_passwd -c -b /mosquitto/data/mosquitto-passwd.txt $(mqtt_user) $(shell cat $(mqtt_password_file))
+	$(container_tool) run --name mqtt-maestro -p 1883:1883 -v $(shell pwd)/hack/mosquitto-passwd.txt:/mosquitto/config/password.txt -v $(shell pwd)/hack/mosquitto.conf:/mosquitto/config/mosquitto.conf -d $(mqtt_image)
 
 .PHONY: mqtt/teardown
 mqtt/teardown:
 	$(container_tool) stop mqtt-maestro
 	$(container_tool) rm mqtt-maestro
 
+.PHONY: pubsub/setup
+pubsub/setup:
+	@mkdir -p ${PWD}/secrets
+	@echo '{"projectID":"$(pubsub_project_id)","endpoint":"localhost:$(pubsub_port)","insecure":true,"topics":{"sourceEvents":"projects/$(pubsub_project_id)/topics/sourceevents","sourceBroadcast":"projects/$(pubsub_project_id)/topics/sourcebroadcast"},"subscriptions":{"agentEvents":"projects/$(pubsub_project_id)/subscriptions/agentevents-maestro","agentBroadcast":"projects/$(pubsub_project_id)/subscriptions/agentbroadcast-maestro"}}' > $(pubsub_config_file)
+	$(container_tool) run --name pubsub-maestro -p $(pubsub_port):8085 -e PUBSUB_PROJECT_ID=$(pubsub_project_id) -d $(pubsub_image) gcloud beta emulators pubsub start --host-port=0.0.0.0:8085 --project=$(pubsub_project_id)
+
+.PHONY: pubsub/teardown
+pubsub/teardown:
+	$(container_tool) stop pubsub-maestro
+	$(container_tool) rm pubsub-maestro
+
+.PHONY: pubsub/init
+pubsub/init:
+	@PUBSUB_EMULATOR_HOST=localhost:$(pubsub_port) PUBSUB_PROJECT_ID=$(pubsub_project_id) bash hack/init-pubsub-emulator.sh
+
 crc/login:
 	@echo "Logging into CRC"
 	@crc console --credentials -ojson | jq -r .clusterConfig.adminCredentials.password | oc login --username kubeadmin --insecure-skip-tls-verify=true https://api.crc.testing:6443

README.md

@@ -42,30 +42,40 @@ reducing the need for direct access to clusters.
 
 ## Run in Local Environment
 
-### Make a build, run postgres and mqtt broker
+### Make a build, run postgres and message broker (MQTT or Pub/Sub)
 
 ```shell
 
 # 1. build the project
 
-$ go install gotest.tools/gotestsum@latest  
+$ go install gotest.tools/gotestsum@latest
 $ make binary
 
-# 2. run a postgres database locally in docker 
+# 2. run a postgres database locally in docker
 
 $ make db/setup
 $ make db/login
-        
+
     root@f076ddf94520:/# psql -h localhost -U maestro maestro
     psql (14.4 (Debian 14.4-1.pgdg110+1))
     Type "help" for help.
-    
+
     maestro=# \dt
     Did not find any relations.
 
-# 3. run a mqtt broker locally in docker
+# 3a. run a MQTT broker locally in docker
 
 $ make mqtt/setup
+
+# OR
+
+# 3b. run a Pub/Sub emulator locally in docker
+
+$ make pubsub/setup
+
+# Initialize topics and subscriptions in the emulator
+# Note: Requires google-cloud-pubsub Python package (pip3 install google-cloud-pubsub)
+$ make pubsub/init
 ```
 
 ### Run database migrations
@@ -101,7 +111,12 @@ maestro=# \dt
 ### Running the Service
 
 ```shell
+# Run with MQTT broker (default)
 $ make run
+
+# OR run with Pub/Sub emulator
+# First, ensure the Pub/Sub emulator is running and configured
+$ MESSAGE_DRIVER_TYPE=pubsub make run
 ```
 
 #### List the consumers

go.mod

@@ -3,6 +3,7 @@ module github.com/openshift-online/maestro
 go 1.25.0
 
 require (
+	cloud.google.com/go/pubsub/v2 v2.3.0
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.12.0
 	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.6.0
 	github.com/Masterminds/squirrel v1.5.4
@@ -42,6 +43,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.39.0
 	go.opentelemetry.io/otel/trace v1.39.0
 	golang.org/x/oauth2 v0.32.0
+	google.golang.org/api v0.255.0
 	google.golang.org/grpc v1.78.0
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/resty.v1 v1.12.0
@@ -68,7 +70,6 @@ require (
 	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
 	cloud.google.com/go/compute/metadata v0.9.0 // indirect
 	cloud.google.com/go/iam v1.5.2 // indirect
-	cloud.google.com/go/pubsub/v2 v2.3.0 // indirect
 	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.9.0 // indirect
 	github.com/AzureAD/microsoft-authentication-library-for-go v1.2.2 // indirect
@@ -184,7 +185,6 @@ require (
 	golang.org/x/time v0.14.0 // indirect
 	golang.org/x/tools v0.39.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/api v0.255.0 // indirect
 	google.golang.org/genproto v0.0.0-20250603155806-513f23925822 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect