Fix nonreliable method of validating shardawareness in e2e's

assets/config/config.yaml

@@ -16,4 +16,4 @@ operatorTests:
   scyllaDBVersions:
     updateFrom: "6.2.0-rc2"
     upgradeFrom: "6.1.2"
-  nodeSetupImage: "quay.io/scylladb/scylla-operator-images:node-setup-v0.0.2@sha256:210b1dd9bd60a5bf4056783f3132bdeef0cf9ab0a19eff0b620b2dfa5c4e5d61"
+  nodeSetupImage: "quay.io/scylladb/scylla-operator-images:node-setup-v0.0.3@sha256:c6b3de240cc5c884d5c617485bae35c51572cdfd39b6431d2e1f759c7d7feea1"

pkg/util/cql/frame.go

@@ -0,0 +1,75 @@
+// Copyright (c) 2024 ScyllaDB.
+
+package cql
+
+import (
+	"bytes"
+	"fmt"
+)
+
+const (
+	headerLen = 9
+
+	// OptionsFrame is a minimal OPTIONS CQL frame.
+	// Ref: https://github.com/apache/cassandra/blob/f278f6774fc76465c182041e081982105c3e7dbb/doc/native_protocol_v4.spec
+	OptionsFrame = `\x04\x00\x00\x00\x05\x00\x00\x00\x00`
+)
+
+type FrameParser struct {
+	buf *bytes.Buffer
+}
+
+func NewFrameParser(buf *bytes.Buffer) *FrameParser {
+	return &FrameParser{
+		buf: buf,
+	}
+}
+
+func (fp *FrameParser) SkipHeader() {
+	_ = fp.readBytes(headerLen)
+}
+
+func (fp *FrameParser) readByte() byte {
+	p, err := fp.buf.ReadByte()
+	if err != nil {
+		panic(fmt.Errorf("can't read byte from buffer: %w", err))
+	}
+	return p
+}
+
+func (fp *FrameParser) ReadShort() uint16 {
+	return uint16(fp.readByte())<<8 | uint16(fp.readByte())
+}
+
+func (fp *FrameParser) ReadStringMultiMap() map[string][]string {
+	n := fp.ReadShort()
+	m := make(map[string][]string, n)
+	for i := uint16(0); i < n; i++ {
+		k := fp.ReadString()
+		v := fp.ReadStringList()
+		m[k] = v
+	}
+	return m
+}
+
+func (fp *FrameParser) readBytes(n int) []byte {
+	p := make([]byte, 0, n)
+	for i := 0; i < n; i++ {
+		p = append(p, fp.readByte())
+	}
+
+	return p
+}
+
+func (fp *FrameParser) ReadString() string {
+	return string(fp.readBytes(int(fp.ReadShort())))
+}
+
+func (fp *FrameParser) ReadStringList() []string {
+	n := fp.ReadShort()
+	l := make([]string, 0, n)
+	for i := uint16(0); i < n; i++ {
+		l = append(l, fp.ReadString())
+	}
+	return l
+}

pkg/util/cql/frame_test.go

@@ -0,0 +1,184 @@
+// Copyright (c) 2024 ScyllaDB.
+
+package cql
+
+import (
+	"bytes"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/api/equality"
+)
+
+func TestFrameParserReadShort(t *testing.T) {
+	tt := []struct {
+		name           string
+		buffer         []byte
+		expectedShort  uint16
+		expectedBuffer []byte
+	}{
+		{
+			name: "consumes two bytes and returns short number",
+			buffer: []byte{
+				0x01, 0x00,
+				0x02, 0x00,
+				0x03, 0x00,
+				0x04, 0x00,
+			},
+			expectedShort: 256,
+			expectedBuffer: []byte{
+				0x02, 0x00,
+				0x03, 0x00,
+				0x04, 0x00,
+			},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := bytes.NewBuffer(tc.buffer)
+			fp := NewFrameParser(buf)
+			gotShort := fp.ReadShort()
+			if gotShort != tc.expectedShort {
+				t.Errorf("got %v short, expected %v", gotShort, tc.expectedShort)
+			}
+			if !equality.Semantic.DeepEqual(buf.Bytes(), tc.expectedBuffer) {
+				t.Errorf("got %v buffer, expected %v", buf, tc.expectedBuffer)
+			}
+		})
+	}
+}
+
+func TestFrameParserSkipHeader(t *testing.T) {
+	tt := []struct {
+		name           string
+		buffer         []byte
+		expectedBuffer []byte
+	}{
+		{
+			name:           "consumes first 9 bytes",
+			buffer:         []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x10},
+			expectedBuffer: []byte{0x10},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := bytes.NewBuffer(tc.buffer)
+			fp := NewFrameParser(buf)
+			fp.SkipHeader()
+			if !equality.Semantic.DeepEqual(buf.Bytes(), tc.expectedBuffer) {
+				t.Errorf("got %v buffer, expected %v", buf, tc.expectedBuffer)
+			}
+		})
+	}
+}
+
+func TestFrameParserReadString(t *testing.T) {
+	tt := []struct {
+		name           string
+		buffer         []byte
+		expectedString string
+		expectedBuffer []byte
+	}{
+		{
+			name: "consumes bytes from buffer by reading string length (uint16) and content",
+			buffer: []byte{
+				0x00, 0x05, // 5 - string length
+				0x68, 0x65, 0x6c, 0x6c, 0x6f, // hello
+			},
+			expectedString: "hello",
+			expectedBuffer: []byte{},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := bytes.NewBuffer(tc.buffer)
+			fp := NewFrameParser(buf)
+			gotString := fp.ReadString()
+			if gotString != tc.expectedString {
+				t.Errorf("got %v string, expected %v", gotString, tc.expectedString)
+			}
+			if !equality.Semantic.DeepEqual(buf.Bytes(), tc.expectedBuffer) {
+				t.Errorf("got %v buffer, expected %v", buf, tc.expectedBuffer)
+			}
+		})
+	}
+}
+
+func TestFrameParserReadStringList(t *testing.T) {
+	tt := []struct {
+		name            string
+		buffer          []byte
+		expectedStrings []string
+		expectedBuffer  []byte
+	}{
+		{
+			name: "consumes bytes from buffer by reading string list length and strings",
+			buffer: []byte{
+				0x00, 0x02, // 2 - slice length
+				0x00, 0x05, // 5 - string length
+				0x68, 0x65, 0x6c, 0x6c, 0x6f, // hello
+				0x00, 0x05, // 5 - string length
+				0x77, 0x6f, 0x72, 0x6c, 0x64, // world
+			},
+			expectedStrings: []string{"hello", "world"},
+			expectedBuffer:  []byte{},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := bytes.NewBuffer(tc.buffer)
+			fp := NewFrameParser(buf)
+			gotString := fp.ReadStringList()
+			if !equality.Semantic.DeepEqual(gotString, tc.expectedStrings) {
+				t.Errorf("got %v strings, expected %v", gotString, tc.expectedStrings)
+			}
+			if !equality.Semantic.DeepEqual(buf.Bytes(), tc.expectedBuffer) {
+				t.Errorf("got %v buffer, expected %v", buf, tc.expectedBuffer)
+			}
+		})
+	}
+}
+
+func TestFrameParserReadStringMultiMap(t *testing.T) {
+	tt := []struct {
+		name             string
+		buffer           []byte
+		expectedMultiMap map[string][]string
+		expectedBuffer   []byte
+	}{
+		{
+			name: "consumes bytes from buffer by reading string list length and strings",
+			buffer: []byte{
+				0x00, 0x01, // 1 - map elements
+				0x00, 0x03, // 3 - key length
+				0x66, 0x6f, 0x6f, // foo
+				0x00, 0x02, // 2 - slice length
+				0x00, 0x05, // 5 - string length
+				0x68, 0x65, 0x6c, 0x6c, 0x6f, // hello
+				0x00, 0x05, // 5 - string length
+				0x77, 0x6f, 0x72, 0x6c, 0x64, // world
+			},
+			expectedMultiMap: map[string][]string{
+				"foo": {"hello", "world"},
+			},
+			expectedBuffer: []byte{},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := bytes.NewBuffer(tc.buffer)
+			fp := NewFrameParser(buf)
+			gotMultiMap := fp.ReadStringMultiMap()
+			if !equality.Semantic.DeepEqual(gotMultiMap, tc.expectedMultiMap) {
+				t.Errorf("got %v multimap, expected %v", gotMultiMap, tc.expectedMultiMap)
+			}
+			if !equality.Semantic.DeepEqual(buf.Bytes(), tc.expectedBuffer) {
+				t.Errorf("got %v buffer, expected %v", buf, tc.expectedBuffer)
+			}
+		})
+	}
+}