feat(zbchaos): query current cluster topology

go-chaos/cmd/cluster.go

@@ -99,29 +99,55 @@ func waitForChange(flags *Flags) error {
 		if err != nil {
 			return err
 		}
-		if topology.LastChange != nil && topology.LastChange.Id == int64(flags.changeId) {
-			if topology.LastChange.Status == "COMPLETED" {
-				internal.LogInfo("Change %d completed successfully", flags.changeId)
-				return nil
-			} else {
-				return fmt.Errorf("change %d failed with status %s", flags.changeId, topology.LastChange.Status)
-			}
-		} else if topology.LastChange != nil && topology.LastChange.Id > int64(flags.changeId) {
+		changeStatus := describeChangeStatus(topology, int64(flags.changeId))
+		switch changeStatus {
+		case ChangeStatusCompleted:
+			internal.LogInfo("Change %d completed successfully", flags.changeId)
+			return nil
+		case ChangeStatusFailed:
+			internal.LogInfo("Change %d failed with status %s", flags.changeId, topology.LastChange.Status)
+			return fmt.Errorf("change %d failed with status %s", flags.changeId, topology.LastChange.Status)
+		case ChangeStatusOutdated:
 			internal.LogInfo("Change %d is outdated but was most likely completed successfully, latest change is %d", flags.changeId, topology.LastChange.Id)
 			return nil
-		} else if topology.PendingChange != nil && topology.PendingChange.Id == int64(flags.changeId) {
+		case ChangeStatusPending:
 			competed := len(topology.PendingChange.Completed)
 			pending := len(topology.PendingChange.Pending)
 			total := competed + pending
 			internal.LogInfo("Change %d is %s with %d/%d operations complete", flags.changeId, topology.PendingChange.Status, competed, total)
-		} else {
+		case ChangeStatusUnknown:
 			internal.LogInfo("Change %d not yet started", flags.changeId)
 		}
-
 		time.Sleep(5 * time.Second)
 	}
 }
 
+type ChangeStatus string
+
+const (
+	ChangeStatusOutdated  ChangeStatus = "OUTDATED"
+	ChangeStatusCompleted ChangeStatus = "COMPLETED"
+	ChangeStatusFailed    ChangeStatus = "FAILED"
+	ChangeStatusPending   ChangeStatus = "PENDING"
+	ChangeStatusUnknown   ChangeStatus = "UNKNOWN"
+)
+
+func describeChangeStatus(topology *CurrentTopology, changeId int64) ChangeStatus {
+	if topology.LastChange != nil && topology.LastChange.Id == changeId {
+		if topology.LastChange.Status == "COMPLETED" {
+			return ChangeStatusCompleted
+		} else {
+			return ChangeStatusFailed
+		}
+	} else if topology.LastChange != nil && topology.LastChange.Id > changeId {
+		return ChangeStatusOutdated
+	} else if topology.PendingChange != nil && topology.PendingChange.Id == changeId {
+		return ChangeStatusPending
+	} else {
+		return ChangeStatusUnknown
+	}
+}
+
 func queryTopology(port int) (*CurrentTopology, error) {
 	url := fmt.Sprintf("http://localhost:%d/actuator/cluster", port)
 	resp, err := http.Get(url)

go-chaos/cmd/cluster_test.go

@@ -0,0 +1,107 @@
+// Copyright 2023 Camunda Services GmbH
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_DescribeChangeStatusWithPending(t *testing.T) {
+	// given
+	topology := CurrentTopology{
+		Version: 1,
+		Brokers: []BrokerState{},
+		LastChange: &LastChange{
+			Id:          2,
+			Status:      "COMPLETED",
+			StartedAt:   "2021-09-01T12:00:00.000Z",
+			CompletedAt: "2021-09-01T12:00:00.000Z",
+		},
+		PendingChange: &TopologyChange{
+			Id:              3,
+			Status:          "IN_PROGRESS",
+			StartedAt:       "2021-09-01T12:00:00.000Z",
+			CompletedAt:     "2021-09-01T12:00:00.000Z",
+			InternalVersion: 1,
+			Completed: []Operation{
+				{
+					Operation: "ADD",
+					BrokerId:  1,
+				},
+			},
+			Pending: []Operation{
+				{
+					Operation: "ADD_BROKER",
+					BrokerId:  2,
+				},
+			},
+		},
+	}
+
+	// then
+	assert.Equal(t, ChangeStatusPending, describeChangeStatus(&topology, 3))
+	assert.Equal(t, ChangeStatusCompleted, describeChangeStatus(&topology, 2))
+	assert.Equal(t, ChangeStatusOutdated, describeChangeStatus(&topology, 1))
+	assert.Equal(t, ChangeStatusUnknown, describeChangeStatus(&topology, 4))
+}
+
+func Test_DescribeChangeStatusWithoutChanges(t *testing.T) {
+	// given
+	topology := CurrentTopology{
+		Version:       1,
+		Brokers:       []BrokerState{},
+		LastChange:    nil,
+		PendingChange: nil,
+	}
+
+	// then
+	assert.Equal(t, ChangeStatusUnknown, describeChangeStatus(&topology, 1))
+	assert.Equal(t, ChangeStatusUnknown, describeChangeStatus(&topology, 2))
+}
+
+func Test_DescribeChangeStatusWithoutCompleted(t *testing.T) {
+	// given
+	topology := CurrentTopology{
+		Version:    1,
+		Brokers:    []BrokerState{},
+		LastChange: nil,
+		PendingChange: &TopologyChange{
+			Id:              3,
+			Status:          "IN_PROGRESS",
+			StartedAt:       "2021-09-01T12:00:00.000Z",
+			CompletedAt:     "2021-09-01T12:00:00.000Z",
+			InternalVersion: 1,
+			Completed: []Operation{
+				{
+					Operation: "ADD",
+					BrokerId:  1,
+				},
+			},
+			Pending: []Operation{
+				{
+					Operation: "ADD_BROKER",
+					BrokerId:  2,
+				},
+			},
+		},
+	}
+
+	// then
+	assert.Equal(t, ChangeStatusUnknown, describeChangeStatus(&topology, 1))
+	assert.Equal(t, ChangeStatusPending, describeChangeStatus(&topology, 3))
+	assert.Equal(t, ChangeStatusUnknown, describeChangeStatus(&topology, 4))
+}