Fix empty log lines parsing in parse_cassandra_log, add automated tes…

.github/workflows/test_and_build_image.yaml

@@ -48,6 +48,11 @@ jobs:
             git status
             exit 1
           fi
+      - name: Run Vector tests
+        run: |
+          sudo apt-get install --yes curl bc
+          curl --proto '=https' --tlsv1.2 -sSfL https://sh.vector.dev | bash -s -- -y
+          make $VECTOR=~/.vector/bin/vector vector-test
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v3
   build_image:

CHANGELOG/CHANGELOG-1.21.md

@@ -15,5 +15,6 @@ When cutting a new release, update the `unreleased` heading to the tag being gen
 
 ## unreleased
 
-* [DOCS] [#1469](https://github.com/riptano/mission-control/issues/1469) Add docs for Reaper's Control Plane deployment mode
 * [CHANGE] Bump default Medusa version to 0.22.3
+* [BUGFIX] [#1409](https://github.com/k8ssandra/k8ssandra-operator/issues/1409) Vector would crash in the Cassandra log parsing if empty lines were present. Add automated tests for Vector parsing rules.
+* [DOCS] [#1469](https://github.com/riptano/mission-control/issues/1469) Add docs for Reaper's Control Plane deployment mode

Makefile

@@ -159,6 +159,15 @@ else
 	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) --bin-dir $(LOCALBIN) -p path)" go test $(GO_FLAGS) ./apis/... ./pkg/... ./test/yq/... ./controllers/... -covermode=atomic -coverprofile coverage.out
 endif
 
+.PHONY: vector-test
+vector-test: ## Run vector tests
+	@echo Generating test files for Vector tests
+	$(eval TMP := $(shell mktemp -d))
+	VECTOR_TEST_FILES=true OUTPUT_PATH=$(TMP) go test -v ./pkg/telemetry -run=TestGenerateTomlTestFiles
+	@echo Running vector test files
+	OUTPUT_PATH=$(TMP) VECTOR=$(VECTOR) scripts/run-vector-tests.sh
+	rm -rf $(TMP)
+
 E2E_TEST_TIMEOUT ?= 3600s
 
 PHONY: e2e-test
@@ -326,6 +335,7 @@ KUSTOMIZE ?= $(LOCALBIN)/kustomize
 CONTROLLER_GEN ?= $(LOCALBIN)/controller-gen
 ENVTEST ?= $(LOCALBIN)/setup-envtest
 GOLANGCI_LINT ?= $(LOCALBIN)/golangci-lint # TODO Add linting to the GHA also
+VECTOR ?= $(LOCALBIN)/vector
 
 ## Tool Versions
 CERT_MANAGER_VERSION ?= v1.12.2
@@ -346,6 +356,10 @@ cert-manager-multi: ## Install cert-manager to the clusters
 		make cert-manager;  \
 	done
 
+.PHONY: vector-install
+vector-install:
+	curl --proto '=https' --tlsv1.2 -sSfL https://sh.vector.dev | bash -s -- --prefix $(LOCALBIN) -y
+
 # Install NGINX in the current Kind cluster using Helm and a values file that is suitable for
 # running e2e tests locally with a cluster created with setup-kind-multicluster.sh. Helm must be
 # pre-installed on the system.

pkg/telemetry/vector.go

@@ -140,10 +140,14 @@ timeout_ms = 10000
 		Inputs: []string{"systemlog"},
 		Config: `source = '''
 del(.source_type)
+.message = string!(.message)
+.message = strip_whitespace(.message)
 . |= parse_groks!(.message, patterns: [
-  "%{LOGLEVEL:loglevel}\\s+\\[(?<thread>((.+)))\\]\\s+%{TIMESTAMP_ISO8601:timestamp}\\s+%{JAVACLASS:class}:%{NUMBER:line}\\s+-\\s+(?<message>(.+\\n?)+)",
+  "%{LOGLEVEL:loglevel}\\s+\\[(?<thread>((.+)))\\]\\s+%{TIMESTAMP_ISO8601:timestamp_raw}\\s+%{JAVACLASS:class}:%{NUMBER:line}\\s+-\\s+(?<message>(.+\\n?)+)",
   ]
 )
+.timestamp = parse_timestamp!(.timestamp_raw, format: "%Y-%m-%d %T,%3f")
+del(.timestamp_raw)
 pod_name, err = get_env_var("POD_NAME")
 if err == null {
   .pod_name = pod_name

pkg/telemetry/vector_test.go

@@ -1,6 +1,9 @@
 package telemetry
 
 import (
+	"fmt"
+	"os"
+	"strings"
 	"testing"
 
 	"github.com/go-logr/logr/testr"
@@ -366,3 +369,95 @@ func TestOverrideSourcePossible(t *testing.T) {
 
 	assert.Equal("stdin", sources[0].Type)
 }
+
+func TestGenerateTomlTestFiles(t *testing.T) {
+	if os.Getenv("VECTOR_TEST_FILES") == "" {
+		t.Skip("Set VECTOR_TEST_FILES to generate vector test files")
+	}
+	outputDir := os.Getenv("OUTPUT_PATH")
+	if outputDir == "" {
+		fmt.Printf("No OUTPUT_PATH env variable set")
+		t.FailNow()
+	}
+	assert := assert.New(t)
+	sources, transformers, sinks := BuildDefaultVectorComponents(vector.VectorConfig{})
+	assert.Equal(2, len(sources))
+	assert.Equal(2, len(transformers))
+	assert.Equal(1, len(sinks))
+
+	telemetrySpec := &telemetry.TelemetrySpec{
+		Vector: &telemetry.VectorSpec{
+			Enabled: ptr.To[bool](true),
+			Components: &telemetry.VectorComponentsSpec{
+				Sources:    sources,
+				Sinks:      sinks,
+				Transforms: transformers,
+			},
+		},
+	}
+
+	// Vector components are provided in the Telemetry spec, build the Vector sink config from them
+	vectorConfigToml := BuildCustomVectorToml(telemetrySpec)
+
+	b := strings.Builder{}
+	fmt.Fprint(&b, vectorConfigToml)
+
+	// Append tests
+
+	fmt.Fprint(&b, `
+[[tests]]
+name = "Test parsing normal Cassandra logs"
+
+[[tests.inputs]]
+insert_at = "parse_cassandra_log"
+value = "WARN  [ScheduledTasks:1] 2024-10-01 12:31:17,694  LeaksDetectorImpl.java:306 - LEAK: RandomAccessReader/RandomAccessReader was not released before it was garbage-collected. This resource is a debugging aid, no negative consequences follow from this leak. However, please report this nonetheless even if auto-cleaning succeeded. Auto cleaning result: not attempted, no cleaner."
+
+[[tests.outputs]]
+extract_from = "parse_cassandra_log"
+
+[[tests.outputs.conditions]]
+type = "vrl"
+source = '''
+assert_eq!(.loglevel, "WARN")
+assert_eq!(.thread, "ScheduledTasks:1")
+assert!(is_timestamp(.timestamp))
+assert!(is_string(.message))
+assert!(is_string(.line))
+assert!(exists(.class))
+'''
+`)
+
+	assert.NoError(os.WriteFile(fmt.Sprintf("%s/vector-simple.toml", outputDir), []byte(b.String()), 0644))
+
+	b.Reset()
+	fmt.Fprint(&b, vectorConfigToml)
+
+	fmt.Fprint(&b, `
+[[tests]]
+name = "Test parsing normal Cassandra logs"
+
+[[tests.inputs]]
+insert_at = "parse_cassandra_log"
+value = '''
+WARN  [ScheduledTasks:1] 2024-10-01 12:31:17,694  LeaksDetectorImpl.java:306 - LEAK: RandomAccessReader/RandomAccessReader was not released before it was garbage-collected. This resource is a debugging aid, no negative consequences follow from this leak. However, please report this nonetheless even if auto-cleaning succeeded. Auto cleaning result: not attempted, no cleaner."
+
+
+'''
+
+[[tests.outputs]]
+extract_from = "parse_cassandra_log"
+
+[[tests.outputs.conditions]]
+type = "vrl"
+source = '''
+assert_eq!(.loglevel, "WARN")
+assert_eq!(.thread, "ScheduledTasks:1")
+assert!(is_timestamp(.timestamp))
+assert!(is_string(.message))
+assert!(is_string(.line))
+assert!(exists(.class))
+'''
+`)
+
+	assert.NoError(os.WriteFile(fmt.Sprintf("%s/vector-emptyline.toml", outputDir), []byte(b.String()), 0644))
+}

scripts/run-vector-tests.sh

@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+for i in $OUTPUT_PATH/*; do
+   $VECTOR test $i
+   if [ "$?" -ne 0 ]; then
+      exit 1
+   fi
+done