diff --git a/.github/workflows/pr_build_and_test.yaml b/.github/workflows/pr_build_and_test.yaml index 78da371a..38cff958 100644 --- a/.github/workflows/pr_build_and_test.yaml +++ b/.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 diff --git a/maelstrom/main.go b/maelstrom/main.go index 03bc8cf7..e3223253 100644 --- a/maelstrom/main.go +++ b/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() {