mirror of
https://github.com/cirruslabs/orchard.git
synced 2026-10-17 20:25:50 +02:00
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
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user