Bug/256 fix flaky system tests

src/main/resources/application.yml

@@ -22,6 +22,15 @@ management:
       exposure:
         # This property enables the openapi and swagger-ui endpoints to be exposed beneath the actuator base path.
         include: "health,openapi,swagger-ui"
+  endpoint:
+    health:
+      probes:
+        enabled: true
+  health:
+    livenessstate:
+      enabled: true
+    readinessstate:
+      enabled: true
 logging:
   level:
     org.springframework.web: ${LOG_LEVEL:info}

system-tests/README.md

@@ -5,6 +5,7 @@ environment. This includes the communication with the Hub but _excludes_ analysi
 simulated.
 
 ## Requirements
+
 The following dependencies need to be satisfied in order to run the tests:
 
 - Golang in version ~1.23.3
@@ -101,12 +102,16 @@ In order to run the tests you have to:
 
 1. build the message broker (if not already built)
     ```shell
-    mvn mvn -B -DskipTests package 
+    mvn -B -DskipTests package 
     ```
-2. run the tests
+2. build the CLI for running system tests
+   ```shell
+   go build -C ./system-tests/tests/
+   ```
+3. run the tests
     ```shell
    ./system-tests/run.sh
     ```
 
-The second step will spin up the whole test environment and subsequently run the tests against it. The script also takes
+The third step will spin up the whole test environment and subsequently run the tests against it. The script also takes
 care of cleaning up any resources in the end.

system-tests/environment/node/node-docker-compose.tpl.yml

@@ -8,6 +8,7 @@ services:
       keycloak:
         condition: service_healthy
     ports:
+      - "19088:8090" # management server
       - "18088:18088"
     restart: always
     secrets:
@@ -25,6 +26,7 @@ services:
       PERSISTENCE_HOSTNAME: "node-a-db"
       PERSISTENCE_PORT: 27017
       PERSISTENCE_DATABASE_NAME: "node-message-broker"
+      LOG_LEVEL: "debug"
     networks:
       node:
       hub-node-intercomm:
@@ -42,6 +44,7 @@ services:
       keycloak:
         condition: service_healthy
     ports:
+      - "19089:8090" # management server
       - "18089:18089"
     restart: always
     secrets:
@@ -59,6 +62,7 @@ services:
       PERSISTENCE_HOSTNAME: "node-b-db"
       PERSISTENCE_PORT: 27017
       PERSISTENCE_DATABASE_NAME: "node-message-broker"
+      LOG_LEVEL: "debug"
     networks:
       node:
       hub-node-intercomm:
@@ -76,6 +80,7 @@ services:
       keycloak:
         condition: service_healthy
     ports:
+      - "19090:8090" # management server
       - "18090:18090"
     restart: always
     secrets:
@@ -93,6 +98,7 @@ services:
       PERSISTENCE_HOSTNAME: "node-c-db"
       PERSISTENCE_PORT: 27017
       PERSISTENCE_DATABASE_NAME: "node-message-broker"
+      LOG_LEVEL: "debug"
     networks:
       node:
       hub-node-intercomm:

system-tests/run.sh

@@ -17,6 +17,7 @@ echo "Running system test for sending messages to dedicated recipients..."
 "$BASE_DIR"/tests/mb-test-cli send-dedicated-messages \
   --analysis-id="$ANALYSIS_ID" \
   --bootstrap-nodes="http://localhost:18088,http://localhost:18089" \
+  --bootstrap-management-nodes="http://localhost:19088,http://localhost:19089" \
   --node-auth-base-url="http://localhost:18080" \
   --node-auth-client-id="message-broker" \
   --node-auth-client-secret="thtiFoImj6rvrfTvKkiOlSigRcYLbQwf"
@@ -33,6 +34,7 @@ echo "Running system test for sending broadcast messages to various recipients..
 "$BASE_DIR"/tests/mb-test-cli send-broadcast-messages \
   --analysis-id="$ANALYSIS_ID" \
   --bootstrap-nodes="http://localhost:18088,http://localhost:18089,http://localhost:18090" \
+  --bootstrap-management-nodes="http://localhost:19088,http://localhost:19089,http://localhost:19090" \
   --node-auth-base-url="http://localhost:18080" \
   --node-auth-client-id="message-broker" \
   --node-auth-client-secret="thtiFoImj6rvrfTvKkiOlSigRcYLbQwf"

system-tests/tests/cmd/global.go

@@ -3,28 +3,31 @@ package cmd
 import "github.com/spf13/cobra"
 
 type GlobalFlags struct {
-	AnalysisId             string
-	BootstrapNodes         []string
-	NodeAuthBaseUrl        string
-	NodeAuthClientId       string
-	NodeAuthClientSecret   string
-	NumberOfMessagesToSend uint16
+	AnalysisId               string
+	BootstrapNodes           []string
+	BootstrapManagementNodes []string
+	NodeAuthBaseUrl          string
+	NodeAuthClientId         string
+	NodeAuthClientSecret     string
+	NumberOfMessagesToSend   uint16
 }
 
 func getGlobalFlags(cmd *cobra.Command) *GlobalFlags {
 	analysisId, _ := cmd.Flags().GetString("analysis-id")
-	BootstrapNodes, _ := cmd.Flags().GetStringSlice("bootstrap-nodes")
+	bootstrapNodes, _ := cmd.Flags().GetStringSlice("bootstrap-nodes")
+	bootstrapManagementNodes, _ := cmd.Flags().GetStringSlice("bootstrap-management-nodes")
 	nodeAuthBaseUrl, _ := cmd.Flags().GetString("node-auth-base-url")
 	nodeAuthClientId, _ := cmd.Flags().GetString("node-auth-client-id")
 	nodeAuthClientSecret, _ := cmd.Flags().GetString("node-auth-client-secret")
 	numberOfMessagesToSend, _ := cmd.Flags().GetUint16("number-of-messages-to-send")
 
 	return &GlobalFlags{
-		AnalysisId:             analysisId,
-		BootstrapNodes:         BootstrapNodes,
-		NodeAuthBaseUrl:        nodeAuthBaseUrl,
-		NodeAuthClientId:       nodeAuthClientId,
-		NodeAuthClientSecret:   nodeAuthClientSecret,
-		NumberOfMessagesToSend: numberOfMessagesToSend,
+		AnalysisId:               analysisId,
+		BootstrapNodes:           bootstrapNodes,
+		BootstrapManagementNodes: bootstrapManagementNodes,
+		NodeAuthBaseUrl:          nodeAuthBaseUrl,
+		NodeAuthClientId:         nodeAuthClientId,
+		NodeAuthClientSecret:     nodeAuthClientSecret,
+		NumberOfMessagesToSend:   numberOfMessagesToSend,
 	}
 }

system-tests/tests/cmd/root.go

@@ -5,12 +5,13 @@ import (
 )
 
 var (
-	AnalysisId             string
-	BootstrapNodes         []string
-	NodeAuthBaseUrl        string
-	NodeAuthClientId       string
-	NodeAuthClientSecret   string
-	NumberOfMessagesToSend uint16
+	AnalysisId               string
+	BootstrapNodes           []string
+	BootstrapManagementNodes []string
+	NodeAuthBaseUrl          string
+	NodeAuthClientId         string
+	NodeAuthClientSecret     string
+	NumberOfMessagesToSend   uint16
 
 	globalFlags = GlobalFlags{}
 	RootCmd     = &cobra.Command{
@@ -26,13 +27,15 @@ func Execute() error {
 func init() {
 	RootCmd.PersistentFlags().StringVar(&globalFlags.AnalysisId, "analysis-id", "", "unique identifier of the analysis to be used")
 	RootCmd.PersistentFlags().StringSliceVar(&globalFlags.BootstrapNodes, "bootstrap-nodes", nil, "list of bootstrap nodes")
+	RootCmd.PersistentFlags().StringSliceVar(&globalFlags.BootstrapManagementNodes, "bootstrap-management-nodes", nil, "list of management nodes for the specified bootstrap nodes (equal to the management server of a node)")
 	RootCmd.PersistentFlags().StringVar(&globalFlags.NodeAuthBaseUrl, "node-auth-base-url", "", "base url for the auth service used for authentication")
 	RootCmd.PersistentFlags().StringVar(&globalFlags.NodeAuthClientId, "node-auth-client-id", "", "client id used for authentication")
 	RootCmd.PersistentFlags().StringVar(&globalFlags.NodeAuthClientSecret, "node-auth-client-secret", "", "client secret used for authentication")
 	RootCmd.PersistentFlags().Uint16Var(&globalFlags.NumberOfMessagesToSend, "number-of-messages-to-send", 10, "number of messages to be send during a test case")
 
 	_ = RootCmd.MarkPersistentFlagRequired("analysis-id")
 	_ = RootCmd.MarkPersistentFlagRequired("bootstrap-nodes")
+	_ = RootCmd.MarkPersistentFlagRequired("bootstrap-management-nodes")
 	_ = RootCmd.MarkPersistentFlagRequired("node-auth-base-url")
 	_ = RootCmd.MarkPersistentFlagRequired("node-auth-client-id")
 	_ = RootCmd.MarkPersistentFlagRequired("node-auth-client-secret")

system-tests/tests/cmd/send-broadcast-messages.go

@@ -34,30 +34,39 @@ func sendBroadcastMessagesFunc(cmd *cobra.Command, _ []string) error {
 	log.Info("Running system test for sending messages to various recipients using broadcast")
 	log.Infof("Running test case with analysis ID: `%s`", globalFlags.AnalysisId)
 	log.Infof("Running test case with bootstrap nodes: `%s`", globalFlags.BootstrapNodes)
+	log.Infof("Running test case with bootstrap management nodes: `%s`", globalFlags.BootstrapManagementNodes)
 
 	if len(globalFlags.BootstrapNodes) < 3 {
 		return fmt.Errorf("at least 3 bootstrap nodes have to be specified")
 	}
+	if len(globalFlags.BootstrapNodes) != len(globalFlags.BootstrapManagementNodes) {
+		return fmt.Errorf("number of bootstrap nodes has to be equal to the number of bootstrap management nodes")
+	}
 
 	authClient := messagebroker.NewMessageBrokerAuthClient(globalFlags.NodeAuthBaseUrl, globalFlags.NodeAuthClientId, globalFlags.NodeAuthClientSecret)
-	testNodeClients := getBroadcastMessagesTestNodeClients(globalFlags.BootstrapNodes, authClient)
+	testNodeClients := getBroadcastMessagesTestNodeClients(globalFlags.BootstrapNodes, globalFlags.BootstrapManagementNodes, authClient)
+
+	log.Info("waiting for all test nodes to get ready")
+	if err := waitForTestNodesToBeReady([]messagebroker.MessageBrokerClient{testNodeClients.ReceiverNodeAClient, testNodeClients.ReceiverNodeBClient, testNodeClients.SenderNodeClient}, 30*time.Second); err != nil {
+		log.Fatalf("failed to wait for test nodes to get ready: %v", err)
+	}
 
-	log.Infof("setting up subscription for result server at reciever node with base URL: `%s`", testNodeClients.ReceiverNodeAClient.GetBaseUrl())
+	log.Infof("setting up subscription for result server at receiver node with base URL: `%s`", testNodeClients.ReceiverNodeAClient.GetBaseUrl())
 	// TODO: remove hard coded value
 	subscriptionIdA, err := configureReceiverNodeSubscription(testNodeClients.ReceiverNodeAClient, globalFlags.AnalysisId, "http://host.docker.internal:8080/results")
 	if err != nil {
-		return fmt.Errorf("could not configure receiver node")
+		return fmt.Errorf("could not configure receiver node: %v", err)
 	}
 	defer func() {
 		log.Info("cleaning up receiver node subscription")
 		cleanUpReceiverNodeSubscription(testNodeClients.ReceiverNodeAClient, globalFlags.AnalysisId, subscriptionIdA)
 	}()
 
-	log.Infof("setting up subscription for result server at reciever node with base URL: `%s`", testNodeClients.ReceiverNodeBClient.GetBaseUrl())
+	log.Infof("setting up subscription for result server at receiver node with base URL: `%s`", testNodeClients.ReceiverNodeBClient.GetBaseUrl())
 	// TODO: remove hard coded value
 	subscriptionIdB, err := configureReceiverNodeSubscription(testNodeClients.ReceiverNodeBClient, globalFlags.AnalysisId, "http://host.docker.internal:8080/results")
 	if err != nil {
-		return fmt.Errorf("could not configure receiver node")
+		return fmt.Errorf("could not configure receiver node: %v", err)
 	}
 	defer func() {
 		log.Info("cleaning up receiver node subscription")
@@ -132,11 +141,11 @@ func sendBroadcastMessagesFunc(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
-func getBroadcastMessagesTestNodeClients(bootstrapNodes []string, authClient messagebroker.MessageBrokerAuthClient) TestNodeClients {
+func getBroadcastMessagesTestNodeClients(bootstrapNodes []string, bootstrapManagementNodes []string, authClient messagebroker.MessageBrokerAuthClient) TestNodeClients {
 	return TestNodeClients{
-		SenderNodeClient:    messagebroker.NewMessageBrokerClient(bootstrapNodes[0], authClient.AcquireAccessToken),
-		ReceiverNodeAClient: messagebroker.NewMessageBrokerClient(bootstrapNodes[1], authClient.AcquireAccessToken),
-		ReceiverNodeBClient: messagebroker.NewMessageBrokerClient(bootstrapNodes[2], authClient.AcquireAccessToken),
+		SenderNodeClient:    messagebroker.NewMessageBrokerClient(bootstrapNodes[0], bootstrapManagementNodes[0], authClient.AcquireAccessToken),
+		ReceiverNodeAClient: messagebroker.NewMessageBrokerClient(bootstrapNodes[1], bootstrapManagementNodes[1], authClient.AcquireAccessToken),
+		ReceiverNodeBClient: messagebroker.NewMessageBrokerClient(bootstrapNodes[2], bootstrapManagementNodes[2], authClient.AcquireAccessToken),
 	}
 }
 

system-tests/tests/cmd/send-dedicated-messages.go

@@ -33,19 +33,28 @@ func sendDedicatedMessagesFunc(cmd *cobra.Command, _ []string) error {
 	log.Info("Running system test for sending messages to dedicated recipients")
 	log.Infof("Running test case with analysis ID: `%s`", globalFlags.AnalysisId)
 	log.Infof("Running test case with bootstrap nodes: `%s`", globalFlags.BootstrapNodes)
+	log.Infof("Running test case with bootstrap management nodes: `%s`", globalFlags.BootstrapManagementNodes)
 
 	if len(globalFlags.BootstrapNodes) < 2 {
 		return fmt.Errorf("at least 2 bootstrap nodes have to be specified")
 	}
+	if len(globalFlags.BootstrapNodes) != len(globalFlags.BootstrapManagementNodes) {
+		return fmt.Errorf("number of bootstrap nodes has to be equal to the number of bootstrap management nodes")
+	}
 
 	authClient := messagebroker.NewMessageBrokerAuthClient(globalFlags.NodeAuthBaseUrl, globalFlags.NodeAuthClientId, globalFlags.NodeAuthClientSecret)
-	testNodeClients := getDedicatedMessagesTestNodeClients(globalFlags.BootstrapNodes, authClient)
+	testNodeClients := getDedicatedMessagesTestNodeClients(globalFlags.BootstrapNodes, globalFlags.BootstrapManagementNodes, authClient)
+
+	log.Info("waiting for all test nodes to get ready")
+	if err := waitForTestNodesToBeReady([]messagebroker.MessageBrokerClient{testNodeClients.ReceiverNodeClient, testNodeClients.SenderNodeClient}, 30*time.Second); err != nil {
+		log.Fatalf("failed to wait for test nodes to get ready: %v", err)
+	}
 
-	log.Infof("setting up subscription for result server at reciever node with base URL: `%s`", testNodeClients.ReceiverNodeClient.GetBaseUrl())
+	log.Infof("setting up subscription for result server at receiver node with base URL: `%s`", testNodeClients.ReceiverNodeClient.GetBaseUrl())
 	// TODO: remove hard coded value
 	subscriptionId, err := configureReceiverNodeSubscription(testNodeClients.ReceiverNodeClient, globalFlags.AnalysisId, "http://host.docker.internal:8080/results")
 	if err != nil {
-		return fmt.Errorf("could not configure receiver node")
+		return fmt.Errorf("could not configure receiver node: %v", err)
 	}
 	defer func() {
 		log.Info("cleaning up receiver node subscription")
@@ -123,10 +132,10 @@ func sendDedicatedMessagesFunc(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
-func getDedicatedMessagesTestNodeClients(bootstrapNodes []string, authClient messagebroker.MessageBrokerAuthClient) dedicatedMessagesTestNodeClients {
+func getDedicatedMessagesTestNodeClients(bootstrapNodes []string, bootstrapManagementNodes []string, authClient messagebroker.MessageBrokerAuthClient) dedicatedMessagesTestNodeClients {
 	return dedicatedMessagesTestNodeClients{
-		SenderNodeClient:   messagebroker.NewMessageBrokerClient(bootstrapNodes[0], authClient.AcquireAccessToken),
-		ReceiverNodeClient: messagebroker.NewMessageBrokerClient(bootstrapNodes[1], authClient.AcquireAccessToken),
+		SenderNodeClient:   messagebroker.NewMessageBrokerClient(bootstrapNodes[0], bootstrapManagementNodes[0], authClient.AcquireAccessToken),
+		ReceiverNodeClient: messagebroker.NewMessageBrokerClient(bootstrapNodes[1], bootstrapManagementNodes[1], authClient.AcquireAccessToken),
 	}
 }
 

system-tests/tests/cmd/util.go

@@ -4,10 +4,48 @@ import (
 	"fmt"
 	messagebroker "mb-test-cli/message-broker"
 	"sort"
+	"sync"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 )
 
+func waitForTestNodesToBeReady(clients []messagebroker.MessageBrokerClient, timeout time.Duration) error {
+	var readiness sync.WaitGroup
+	readiness.Add(len(clients))
+
+	for _, client := range clients {
+		go func() {
+			defer readiness.Done()
+
+			for {
+				ready, err := client.IsReady()
+				if err != nil {
+					log.Warn("could not check readiness for broker at `%s`", client.GetBaseUrl())
+					continue
+				}
+				if !ready {
+					continue
+				}
+				break
+			}
+		}()
+	}
+
+	completionCh := make(chan struct{})
+	go func() {
+		defer close(completionCh)
+		readiness.Wait()
+	}()
+
+	select {
+	case <-completionCh:
+		return nil
+	case <-time.After(timeout):
+		return fmt.Errorf("timed out while waiting for test nodes to get ready")
+	}
+}
+
 func configureReceiverNodeSubscription(receiverNodeClient messagebroker.MessageBrokerClient, analysisId string, webhookUrl string) (messagebroker.SubscriptionId, error) {
 	subscription, err := receiverNodeClient.CreateSubscription(analysisId, webhookUrl)
 	if err != nil {

system-tests/tests/message-broker/client.go

@@ -19,19 +19,25 @@ type messageAPI interface {
 	SendBroadcastMessage(analysisId string, message TestMessage) error
 }
 
-type intance interface {
+type healthAPI interface {
+	IsReady() (bool, error)
+}
+
+type instance interface {
 	GetBaseUrl() string
 }
 
 type MessageBrokerClient interface {
 	subscriptionAPI
 	discoveryAPI
 	messageAPI
-	intance
+	healthAPI
+	instance
 }
 
 type messageBrokerClient struct {
 	baseUrl             string
+	managementBaseUrl   string // base url to reach the broker's management server
 	accessTokenAcquirer func() (string, error)
 	httpClient          http.Client
 }
@@ -40,9 +46,10 @@ func (mbc *messageBrokerClient) GetBaseUrl() string {
 	return mbc.baseUrl
 }
 
-func NewMessageBrokerClient(baseUrl string, accessTokenAcquirer func() (string, error)) MessageBrokerClient {
+func NewMessageBrokerClient(baseUrl string, managementBaseUrl string, accessTokenAcquirer func() (string, error)) MessageBrokerClient {
 	return &messageBrokerClient{
 		baseUrl:             baseUrl,
+		managementBaseUrl:   managementBaseUrl,
 		accessTokenAcquirer: accessTokenAcquirer,
 		httpClient:          http.Client{Timeout: time.Duration(1) * time.Second},
 	}