Merge pull request #122 from zeebe-io/zell-go-chaos-worker

New Zbchaos cli

.github/workflows/go-ci.yml

@@ -12,14 +12,16 @@ jobs:
   go-ci:
     name: Run Go CI
     runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: go-chaos
     steps:
     - uses: actions/checkout@v2
 
     - name: Set up Go
       uses: actions/setup-go@v2
       with:
         go-version: "${{ env.GO_VERSION }}"
-
     - name: Go Format check
       run: |
         make fmt

go-chaos/.gitignore

@@ -0,0 +1 @@
+zbchaos

Makefile → go-chaos/Makefile

@@ -1,29 +1,34 @@
 # Makefile for go-chaos
+directory=.
 
 # test: runs the tests without updating the golden files (runs checks against golden files)
 .PHONY: test
-test:	deps
-	go test ./...
+test:	
+	go test $(directory)/...
+# go build
+.PHONY: build
+build:
+	go build -o zbchaos main.go
 
 # golden: runs the tests with updating the golden files
 .PHONY: golden
-golden:	deps
-	go test ./... -args -update-golden 
+golden:	
+	go test $(directory)/... -args -update-golden 
 
 # fmt: runs the gofmt in order to format all go files
 .PHONY: fmt
 fmt:
-	go fmt ./... 
+	go fmt $(directory)/... 
 
 # addlicense: add license headers to go files
 .PHONY: addlicense
 addlicense:
-	addlicense -c 'Camunda Services GmbH' -l apache go-chaos/**/*.go
+	addlicense -c 'Camunda Services GmbH' -l apache ./**/*.go
 
 # checkLicense: checks that the go files contain license header
 .PHONY: checkLicense
 checkLicense:
-	addlicense -check -l apache go-chaos/**/*.go
+	addlicense -check -l apache ./**/*.go
 
 # installLicense: installs the addlicense tool
 .PHONY: installLicense

go-chaos/cmd/root.go

@@ -0,0 +1,40 @@
+// Copyright 2022 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 (
+	"fmt"
+	"os"
+
+	"github.com/spf13/cobra"
+)
+
+var rootCmd = &cobra.Command{
+	Use:   "zbchaos",
+	Short: "Zeebe chaos is a chaos experiment tool for Zeebe",
+	Long: `A chaos experimenting toolkit for Zeebe.
+    Perfect to inject some chaos into your brokers and gateways.`,
+}
+
+func NewCmd() *cobra.Command {
+	return rootCmd
+}
+
+func Execute() {
+	if err := rootCmd.Execute(); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}

go-chaos/cmd/terminate.go

@@ -0,0 +1,135 @@
+// Copyright 2022 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 (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/camunda-cloud/zeebe/clients/go/pkg/pb"
+	"github.com/camunda-cloud/zeebe/clients/go/pkg/zbc"
+	"github.com/spf13/cobra"
+	"github.com/zeebe-io/zeebe-chaos/go-chaos/internal"
+)
+
+var (
+	partitionId int
+	role        string
+)
+
+func init() {
+	rootCmd.AddCommand(terminateCmd)
+
+	terminateCmd.Flags().StringVar(&role, "role", "LEADER", "Specify the partition role [LEADER, FOLLOWER]")
+	terminateCmd.Flags().IntVar(&partitionId, "partitionId", 1, "Specify the id of the partition")
+
+	if err := terminateCmd.MarkFlagRequired("role"); err != nil {
+		panic(err)
+	}
+
+	if err := terminateCmd.MarkFlagRequired("partitionId"); err != nil {
+		panic(err)
+	}
+}
+
+var terminateCmd = &cobra.Command{
+	Use:   "terminate",
+	Short: "Terminates a Zeebe broker",
+	Long:  `Terminates a Zeebe broker with a certain role and given partition.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		port := 26500
+		k8Client := internal.CreateK8Client()
+		closeFn, err := k8Client.GatewayPortForward(port)
+		if err != nil {
+			panic(err.Error())
+		}
+		defer closeFn()
+
+		zbClient, err := internal.CreateZeebeClient(port)
+		if err != nil {
+			panic(err.Error())
+		}
+		defer zbClient.Close()
+		broker, err := getBrokerToTerminate(k8Client, zbClient)
+		if err != nil {
+			panic(err.Error())
+		}
+
+		err = k8Client.TerminatePod(broker)
+		if err != nil {
+			panic(err.Error())
+		}
+
+		fmt.Printf("\nDeleted %s", broker)
+		fmt.Println()
+	},
+}
+
+func getBrokerToTerminate(k8Client internal.K8Client, zbClient zbc.Client) (string, error) {
+	topologyResponse, err := zbClient.NewTopologyCommand().Send(context.TODO())
+	if err != nil {
+		return "", err
+	}
+
+	assertPartitionsCount(topologyResponse)
+
+	roleValue, exist := pb.Partition_PartitionBrokerRole_value[role]
+	assertRoleExists(exist)
+
+	nodeId := extractNodeId(topologyResponse, roleValue)
+
+	brokerPodNames, err := k8Client.GetBrokerPodNames()
+	if err != nil {
+		return "", err
+	}
+
+	broker := brokerPodNames[nodeId]
+	return broker, nil
+}
+
+func assertRoleExists(exist bool) {
+	if !exist {
+		errorMsg := fmt.Sprintf("Expected a partition role, which is part of [LEADER, FOLLOWER], but got %s.", role)
+		panic(errors.New(errorMsg))
+	}
+}
+
+func assertPartitionsCount(topologyResponse *pb.TopologyResponse) {
+	partitionsCount := topologyResponse.PartitionsCount
+	if partitionsCount < int32(partitionId) {
+		errorMsg := fmt.Sprintf("Expected that given partition id (%d) is smaller then the partitions count %d, but was greater.", partitionId, partitionsCount)
+		panic(errors.New(errorMsg))
+	}
+}
+
+func extractNodeId(topologyResponse *pb.TopologyResponse, roleValue int32) int32 {
+	nodeId := int32(-1)
+	for _, broker := range topologyResponse.Brokers {
+		for _, partition := range broker.Partitions {
+			if partition.PartitionId == int32(partitionId) &&
+				partition.Role == pb.Partition_PartitionBrokerRole(roleValue) {
+				nodeId = broker.NodeId
+				break
+			}
+		}
+	}
+
+	if nodeId == int32(-1) {
+		errorMsg := fmt.Sprintf("Expected to find broker with given partition id (%d) and role %s, but found nothing.", partitionId, role)
+		panic(errors.New(errorMsg))
+	}
+	return nodeId
+}

go-chaos/cmd/topology.go

@@ -0,0 +1,62 @@
+// Copyright 2022 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 (
+	"context"
+	"fmt"
+
+	"github.com/spf13/cobra"
+	"github.com/zeebe-io/zeebe-chaos/go-chaos/internal"
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+func init() {
+	rootCmd.AddCommand(topologyCmd)
+}
+
+var topologyCmd = &cobra.Command{
+	Use:   "topology",
+	Short: "Print the Zeebe topology deployed in the current namespace",
+	Long:  `Shows the current Zeebe topology, in the current kubernetes namespace.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		port := 26500
+		k8Client := internal.CreateK8Client()
+		closeFn, err := k8Client.GatewayPortForward(port)
+		if err != nil {
+			panic(err.Error())
+		}
+		defer closeFn()
+
+		client, err := internal.CreateZeebeClient(port)
+		if err != nil {
+			panic(err.Error())
+		}
+
+		response, err := client.NewTopologyCommand().Send(context.TODO())
+		if err != nil {
+			panic(err)
+		}
+
+		m := protojson.MarshalOptions{EmitUnpopulated: true, Indent: "  "}
+		valueJSON, err := m.Marshal(response)
+		if err != nil {
+			panic(err.Error())
+		}
+
+		fmt.Printf("\nResponse topology, %s", string(valueJSON))
+		fmt.Println()
+	},
+}

go-chaos/cmd/zeebePods.go

@@ -0,0 +1,43 @@
+// Copyright 2022 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 (
+	"fmt"
+
+	"github.com/spf13/cobra"
+	"github.com/zeebe-io/zeebe-chaos/go-chaos/internal"
+)
+
+func init() {
+	rootCmd.AddCommand(getZeebeBrokersCmd)
+}
+
+var getZeebeBrokersCmd = &cobra.Command{
+	Use:   "brokers",
+	Short: "Print the name of the Zeebe broker pods",
+	Long:  `Show all names of deployed Zeebe brokers, in the current kubernetes namespace.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		client := internal.CreateK8Client()
+		pods, err := client.GetBrokerPodNames()
+		if err != nil {
+			panic(err.Error())
+		}
+
+		for _, item := range pods {
+			fmt.Printf("%s\n", item)
+		}
+	},
+}