Run Maelstrom in CI (oxia-db#336)

Author: merlimat

.github/workflows/pr_build_and_test.yaml

@@ -26,6 +26,9 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: true
 
+env:
+  MAELSTROM_VERSION: "0.2.3"
+
 jobs:
 
   build:
@@ -108,3 +111,45 @@ jobs:
         run: kubectl -n oxia logs -l app.kubernetes.io/component=coordinator
       - name: Print perf logs
         run: kubectl -n oxia logs perf
+
+  maelstrom:
+    name: Maelstrom
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: 1.19
+
+      - name: Set up JDK 17
+        uses: actions/setup-java@v3
+        with:
+          distribution: temurin
+          java-version: 17
+
+      - name: Install Gnuplot
+        run: |
+          sudo apt-get install -y gnuplot
+
+      - run: make
+      - run: make maelstrom
+
+      - name: Download Maelstrom
+        run: |
+          curl -O -L https://github.com/jepsen-io/maelstrom/releases/download/v${MAELSTROM_VERSION}/maelstrom.tar.bz2
+          tar xf maelstrom.tar.bz2
+
+      - name: Run Maelstrom test
+        run: |
+          cd maelstrom
+          ./maelstrom test -w lin-kv  --bin ../bin/oxia-maelstrom \
+                --time-limit 60  --concurrency 2n  --latency 10 --latency-dist uniform
+
+      - name: Upload test results
+        uses: actions/upload-artifact@v3
+        if: failure()
+        with:
+          name: maelstrom-result
+          path: maelstrom/store

maelstrom/main.go

@@ -17,6 +17,7 @@ package main
 import (
 	"bufio"
 	"fmt"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
@@ -64,6 +65,16 @@ var thisNode string
 var allNodes []string
 
 func handleInit(scanner *bufio.Scanner) {
+	for {
+		if err := receiveInit(scanner); err != nil {
+			continue
+		}
+
+		return
+	}
+}
+
+func receiveInit(scanner *bufio.Scanner) error {
 	if !scanner.Scan() {
 		log.Fatal().Msg("no init received")
 	}
@@ -72,7 +83,9 @@ func handleInit(scanner *bufio.Scanner) {
 	log.Info().RawJSON("line", []byte(line)).Msg("Got line")
 	reqType, req, _ := parseRequest(line)
 	if reqType != MsgTypeInit {
-		log.Fatal().Msg("unexpected request")
+		log.Error().Interface("req", req).
+			Msg("Unexpected request while waiting for init")
+		return errors.New("invalid message type")
 	}
 
 	init := req.(*Message[Init])
@@ -94,6 +107,8 @@ func handleInit(scanner *bufio.Scanner) {
 			InReplyTo: &init.Body.MsgId,
 		}},
 	})
+
+	return nil
 }
 
 func main() {