test(buffer): add functional test for disk buffer corruption recovery (LOG-9386)

Verify that Vector does not CrashLoop when the disk buffer contains
corrupted protobuf records. The test corrupts buffer record payloads
while the collector is running, then kills it so the restart reads the
corrupted buffer. Asserts that Vector logs a warning about corrupted
records during seek, starts successfully, and continues delivering
logs after recovery.

Signed-off-by: Vitalii Parfonov <vparfono@redhat.com>

Author: vparfonov

test/framework/functional/framework.go

@@ -192,6 +192,22 @@ func (f *CollectorFunctionalFramework) AddOutputContainersVisitors() []runtime.P
 	return visitors
 }
 
+const ToolsContainerName = "tools"
+
+func (f *CollectorFunctionalFramework) AddToolsContainerVisitor(volumeName, mountPath string) runtime.PodBuilderVisitor {
+	return func(b *runtime.PodBuilder) error {
+		b.AddEmptyDirVolume(volumeName)
+		b.GetContainer(constants.CollectorName).
+			AddVolumeMount(volumeName, mountPath, "", false).
+			Update()
+		b.AddContainer(ToolsContainerName, "registry.access.redhat.com/ubi9/ubi-minimal:latest").
+			AddVolumeMount(volumeName, mountPath, "", false).
+			WithCmd([]string{"sleep", "infinity"}).
+			End()
+		return nil
+	}
+}
+
 // Deploy the objects needed to functional Test
 func (f *CollectorFunctionalFramework) Deploy() (err error) {
 	return f.DeployWithVisitors(f.AddOutputContainersVisitors())

test/functional/misc/corrupt_buffer.go

@@ -0,0 +1,93 @@
+package misc
+
+import (
+	"encoding/binary"
+	"fmt"
+	"hash/crc32"
+	"os"
+)
+
+// corruptDiskBufferPayload corrupts protobuf payloads in a Vector disk buffer
+// data file while keeping CRC32 checksums valid, causing ReaderError::Decode
+// on startup seek. The last record is left intact (writer validates it).
+// Returns the number of corrupted records.
+func corruptDiskBufferPayload(datFilePath string) (int, error) {
+	data, err := os.ReadFile(datFilePath)
+	if err != nil {
+		return 0, fmt.Errorf("read file: %w", err)
+	}
+
+	type record struct{ start, length int }
+	var records []record
+	offset := 0
+	for offset+8 < len(data) {
+		recLen := int(binary.BigEndian.Uint64(data[offset : offset+8]))
+		recStart := offset + 8
+		if recLen < 32 || recStart+recLen > len(data) {
+			break
+		}
+		records = append(records, record{recStart, recLen})
+		offset = recStart + recLen
+	}
+
+	if len(records) < 2 {
+		return 0, fmt.Errorf("need at least 2 records to corrupt (found %d); last record must stay intact for writer validation", len(records))
+	}
+
+	corruptUpTo := len(records) - 1
+	corrupted := 0
+	for i := 0; i < corruptUpTo; i++ {
+		r := records[i]
+		if err := corruptRecordPayload(data[r.start : r.start+r.length]); err != nil {
+			return 0, fmt.Errorf("corrupt record %d: %w", i, err)
+		}
+		corrupted++
+	}
+
+	if err := os.WriteFile(datFilePath, data, 0o640); err != nil {
+		return 0, err
+	}
+	return corrupted, nil
+}
+
+// corruptRecordPayload XOR-flips the payload bytes of a single rkyv-serialized
+// record and recalculates the CRC32 checksum so the record passes checksum
+// validation but fails protobuf decode.
+func corruptRecordPayload(recordData []byte) error {
+	if len(recordData) < 32 {
+		return fmt.Errorf("record too small: %d bytes", len(recordData))
+	}
+
+	// ArchivedRecord layout (repr(C), little-endian, 32 bytes at end of record):
+	//   +0: checksum u32, +4: padding, +8: id u64,
+	//   +16: metadata u32, +20: rel_ptr i32, +24: payload_len u32, +28: padding
+	structOff := len(recordData) - 32
+
+	id := binary.LittleEndian.Uint64(recordData[structOff+8 : structOff+16])
+	metadata := binary.LittleEndian.Uint32(recordData[structOff+16 : structOff+20])
+	relPtr := int32(binary.LittleEndian.Uint32(recordData[structOff+20 : structOff+24]))
+	payloadLen := binary.LittleEndian.Uint32(recordData[structOff+24 : structOff+28])
+
+	payloadStart := int(structOff+20) + int(relPtr)
+	if payloadStart < 0 || payloadStart+int(payloadLen) > len(recordData) {
+		return fmt.Errorf("payload out of bounds: start=%d len=%d record=%d", payloadStart, payloadLen, len(recordData))
+	}
+
+	payload := recordData[payloadStart : payloadStart+int(payloadLen)]
+	for i := range payload {
+		payload[i] ^= 0xFF
+	}
+
+	// Recalculate CRC32-IEEE: BE(id) + BE(metadata) + payload
+	h := crc32.NewIEEE()
+	idBE := make([]byte, 8)
+	binary.BigEndian.PutUint64(idBE, id)
+	_, _ = h.Write(idBE)
+	metaBE := make([]byte, 4)
+	binary.BigEndian.PutUint32(metaBE, metadata)
+	_, _ = h.Write(metaBE)
+	_, _ = h.Write(payload)
+	binary.LittleEndian.PutUint32(recordData[structOff:structOff+4], h.Sum32())
+
+	return nil
+}