From f6b48b7c42904286b4597b3759c7799f26835ad1 Mon Sep 17 00:00:00 2001
From: Fedor Korotkov <fedor.korotkov@gmail.com>
Date: Mon, 5 Jun 2023 21:01:12 +0400
Subject: [PATCH] Change event prefix to preserve order under load (#89)

* Change event prefix to preserve order under load

When there are a lot of events streamed from a worker, it's possible to have two batches coming for the same timestamp (which is a timestamp of the event on the worker). This way the existing logic would mess up the order because `index` and the random number doesn't guarantee the order.

To fix this I've changed the format of the prefix for the event to include tro things:

1. Timestamp in nanoseconds of the injection time on the controller so two sequential batches will have guaranteed order unless they are processed within a nanosecond.
2. Made the `index` being fixed length with trailing zeros, so they are properly lexicographically sorted (`000001`, `000002`, ...).

* No need to disable linting
---
 internal/controller/store/badger/badger_events.go | 9 +++++----
 internal/tests/integration_test.go                | 9 +++++++--
 2 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/internal/controller/store/badger/badger_events.go b/internal/controller/store/badger/badger_events.go
index 7237dea..df48c84 100644
--- a/internal/controller/store/badger/badger_events.go
+++ b/internal/controller/store/badger/badger_events.go
@@ -5,8 +5,8 @@ import (
 	"fmt"
 	"github.com/cirruslabs/orchard/pkg/resource/v1"
 	"github.com/dgraph-io/badger/v3"
-	"math/rand"
 	"path"
+	"time"
 )
 
 const SpaceEvents = "/events"
@@ -21,16 +21,17 @@ func (txn *Transaction) AppendEvents(events []v1.Event, scope ...string) (err er
 		err = mapErr(err)
 	}()
 
+	injectionTime := time.Now().UnixNano()
+
 	for index, event := range events {
 		valueBytes, err := json.Marshal(event)
 		if err != nil {
 			return err
 		}
-		//nolint:gosec
-		eventUID := fmt.Sprintf("/%d-%d-%d",
+		eventUID := fmt.Sprintf("/%d-%d-%06d",
 			event.Timestamp,
+			injectionTime, // to preserve order in case two sequential batches have events with the same timestamp
 			index,         // to preserve order in case a batch of events has some events with the same timestamp
-			rand.Uint32(), // extra cautions to avoid collisions in case several batches of events are appended at the same time
 		)
 
 		eventKey := scopePrefix(scope)
diff --git a/internal/tests/integration_test.go b/internal/tests/integration_test.go
index a9edef7..6d8b955 100644
--- a/internal/tests/integration_test.go
+++ b/internal/tests/integration_test.go
@@ -18,6 +18,7 @@ import (
 	"golang.org/x/exp/slices"
 	"net"
 	"net/http"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -41,7 +42,7 @@ func TestSingleVM(t *testing.T) {
 		Headless: true,
 		Status:   v1.VMStatusPending,
 		StartupScript: &v1.VMScript{
-			ScriptContent: "echo \"Hello, $FOO!\"",
+			ScriptContent: "echo \"Hello, $FOO!\"\nfor i in $(seq 1 1000); do echo \"$i\"; done",
 			Env:           map[string]string{"FOO": "Bar"},
 		},
 	})
@@ -73,7 +74,11 @@ func TestSingleVM(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	assert.Equal(t, []string{"Hello, Bar!"}, logLines)
+	expectedLogs := []string{"Hello, Bar!"}
+	for i := 1; i <= 1000; i++ {
+		expectedLogs = append(expectedLogs, strconv.Itoa(i))
+	}
+	assert.Equal(t, expectedLogs, logLines)
 
 	// Ensure that the VM exists on disk before deleting it
 	require.True(t, hasVMByPredicate(t, func(info tart.VMInfo) bool {