feat(e2e test): add e2e test to verify udp recovery

Makefile

@@ -323,6 +323,13 @@ test-upgrade: $(JUNITREPORT)
 	IMAGE_LOGGING_EVENTROUTER=$(IMAGE_LOGGING_EVENTROUTER) \
 	exit 0
 
+.PHONY: test-e2e-from-ci-bundle
+test-e2e-from-ci-bundle: $(JUNITREPORT)
+	RELATED_IMAGE_VECTOR=$(IMAGE_LOGGING_VECTOR) \
+	RELATED_IMAGE_LOG_FILE_METRIC_EXPORTER=$(IMAGE_LOGFILEMETRICEXPORTER) \
+	IMAGE_LOGGING_EVENTROUTER=$(IMAGE_LOGGING_EVENTROUTER) \
+	LOG_LEVEL=3 hack/test-e2e-from-ci-bundle.sh
+
 .PHONY: test-e2e
 test-e2e: $(JUNITREPORT)
 	RELATED_IMAGE_VECTOR=$(IMAGE_LOGGING_VECTOR) \

hack/manifests/clusterlogforwarder/base/clusterlogforwarder.yaml

@@ -2,7 +2,6 @@ apiVersion: "observability.openshift.io/v1"
 kind: ClusterLogForwarder
 metadata:
   name: collector
-  namespace: openshift-logging 
 spec:
   collector:
     resources:
@@ -14,6 +13,6 @@ spec:
   - name: infra
     inputRefs:
     - application
-    - infrastructure
+#    - infrastructure
     outputRefs:
       - replaceme

hack/manifests/clusterlogforwarder/overlays/syslog/udp/kustomization.yaml

@@ -0,0 +1,27 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+resources:
+- ../../../base
+
+patches:
+- patch: |-
+    - op: replace
+      path: /spec/pipelines/0/outputRefs/0
+      value: rsyslog
+    - op: add
+      path: /spec/outputs
+      value:
+      - name: rsyslog
+        type: syslog
+        syslog:
+          tuning:
+            deliveryMode: AtLeastOnce
+          url: 'udp://rsyslog-server-service.openshift-logging.svc.cluster.local:514'
+          rfc: RFC5424
+          facility: local0
+          enrichment: KubernetesMinimal
+          appName: '{.systemd.u.SYSLOG_IDENTIFIER||.log_type||"-"}'
+          procId: '{.systemd.t.PID||"-"}'
+          msgId: '{.systemd.u.MESSAGE_ID||"-"}'
+  target:
+    kind: ClusterLogForwarder

hack/manifests/log-generator/base/kustomization.yaml

@@ -0,0 +1,13 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+labels:
+  - pairs:
+      app.kubernetes.io/name: log-generator
+      app.kubernetes.io/component: log-generator
+      app.kubernetes.io/part-of: cluster-logging
+    includeSelectors: true 
+    includeTemplates: true 
+resources:
+- log_generator_deployment.yaml
+- servicemonitor.yaml
+- service.yaml

hack/manifests/log-generator/base/log_generator_deployment.yaml

@@ -0,0 +1,35 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: log-generator
+spec:
+  replicas: 1
+  template:
+    spec:
+      securityContext:
+        runAsNonRoot: true
+        seccompProfile:
+          type: RuntimeDefault
+      containers:
+      - name: log-generator
+        env:
+        - name: LOGS_PER_SECOND
+          value: "1"
+        - name: PAYLOAD_SIZE
+          value: "512"
+        image: quay.io/openshift-logging/cluster-logging-load-client:latest
+        imagePullPolicy: Always
+        args:
+        - --logs-per-second=$(LOGS_PER_SECOND)
+        - --synthetic-payload-size=$(PAYLOAD_SIZE)
+        - --log-type=synthetic
+        ports:
+        - containerPort: 8081
+          name: metrics
+          protocol: TCP
+        securityContext:
+          allowPrivilegeEscalation: false
+          readOnlyRootFilesystem: true
+          capabilities:
+            drop:
+            - ALL

hack/manifests/log-generator/base/service.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: log-generator
+spec:
+  ports:
+  - name: metrics
+    port: 8081
+    protocol: TCP
+    targetPort: metrics

hack/manifests/log-generator/base/servicemonitor.yaml

@@ -0,0 +1,16 @@
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: log-generator
+spec:
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: log-generator
+      app.kubernetes.io/component: log-generator
+      app.kubernetes.io/part-of: cluster-logging
+  endpoints:
+  - port: metrics
+  jobLabel: app.kubernetes.io/name
+  podTargetLabels:
+  - app.kubernetes.io/name
+  - app.kubernetes.io/component

hack/manifests/log-generator/overlays/by_payload_size/1024b_0001lps/kustomization.yaml

@@ -0,0 +1,16 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+resources:
+- ../../../base
+
+patches:
+- patch: |-
+    - op: replace
+      path: /spec/template/spec/containers/0/env
+      value:
+        - name: LOGS_PER_SECOND
+          value: "1"
+        - name: PAYLOAD_SIZE
+          value: "1024"
+  target:
+    kind: Deployment

hack/manifests/log-generator/overlays/logtype/application/kustomization.yaml

@@ -0,0 +1,15 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+resources:
+- ../../../base
+
+patches:
+- patch: |-
+    - op: replace
+      path: /spec/template/spec/containers/0/args
+      value:
+        - --logs-per-second=$(LOGS_PER_SECOND)
+        - --log-type=application
+        - --log-format=raw
+  target:
+    kind: Deployment

hack/test-e2e-from-ci-bundle.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+current_dir=$(dirname "${BASH_SOURCE[0]}" )
+GOFLAGS=-mod=mod NO_COLOR=1 go test -v -timeout=90m -ginkgo.v -ginkgo.trace \
+   -ginkgo.poll-progress-after=300s \
+   -ginkgo.poll-progress-interval=30s \
+   "${current_dir}/../test/e2e/..."

internal/factory/deployment.go

@@ -17,7 +17,7 @@ func NewDeployment(namespace, deploymentName, component, impl string, replicas i
 
 	dpl := runtime.NewDeployment(namespace, deploymentName, visitors...)
 	runtime.NewDeploymentBuilder(dpl).WithTemplateAnnotations(annotations).
-		WithTemplateLabels(dpl.Labels).
+		WithTemplateLabels(selectors).
 		WithSelector(selectors).
 		WithPodSpec(podSpec).
 		WithReplicas(utils.GetPtr(replicas))

test/e2e/logforwarding/syslog/recovery_test.go

@@ -0,0 +1,142 @@
+package syslog
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"github.com/openshift/cluster-logging-operator/internal/constants"
+	"github.com/openshift/cluster-logging-operator/internal/runtime"
+	obsruntime "github.com/openshift/cluster-logging-operator/internal/runtime/observability"
+	"github.com/openshift/cluster-logging-operator/internal/utils"
+	framework "github.com/openshift/cluster-logging-operator/test/framework/e2e"
+	"github.com/openshift/cluster-logging-operator/test/helpers/rand"
+	helpersyslog "github.com/openshift/cluster-logging-operator/test/helpers/syslog"
+	testruntime "github.com/openshift/cluster-logging-operator/test/runtime"
+	testruntimeobs "github.com/openshift/cluster-logging-operator/test/runtime/observability"
+	apps "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var _ = Describe("[ClusterLogForwarder] Syslog UDP connection recovery", func() {
+	var (
+		err              error
+		e2e              *framework.E2ETestFramework
+		forwarder        *obs.ClusterLogForwarder
+		forwarderName    = "my-forwarder"
+		deployNS         string
+		syslogDeployment *apps.Deployment
+		serviceAccount   *corev1.ServiceAccount
+	)
+
+	Describe("with vector collector over UDP", func() {
+		BeforeEach(func() {
+			e2e = framework.NewE2ETestFramework()
+			deployNS = e2e.Test.NS.Name
+
+			By("Create log generator first so it starts producing logs")
+			msg := rand.Word(1024)
+			logGenerator := testruntime.NewLogGeneratorDeployment(deployNS, "log-generator", int32(60), 1*time.Second, string(msg))
+			Expect(e2e.Test.Create(logGenerator)).To(Succeed(), "failed to create log generator")
+
+			if serviceAccount, err = e2e.BuildAuthorizationFor(deployNS, forwarderName).
+				AllowClusterRole(framework.ClusterRoleCollectApplicationLogs).
+				AllowClusterRole(framework.ClusterRoleCollectInfrastructureLogs).
+				AllowClusterRole(framework.ClusterRoleCollectAuditLogs).Create(); err != nil {
+				Fail(err.Error())
+			}
+
+			By("Deploy syslog UDP receiver first")
+			syslogDeployment, err = e2e.DeploySyslogReceiver(deployNS, corev1.ProtocolUDP, false, helpersyslog.RFC5424)
+			Expect(err).To(BeNil(), "should successfully deploy syslog UDP receiver")
+
+			By("Create forwarder with the syslog receiver running")
+			forwarder = testruntimeobs.NewClusterLogForwarderBuilder(obsruntime.NewClusterLogForwarder(deployNS, forwarderName, runtime.Initialize), func(clf *obs.ClusterLogForwarder) {
+				clf.Spec.Collector = &obs.CollectorSpec{
+					Resources: &corev1.ResourceRequirements{
+						Limits: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("1"),
+							corev1.ResourceMemory: resource.MustParse("2Gi"),
+						},
+						Requests: corev1.ResourceList{
+							corev1.ResourceCPU:    resource.MustParse("100m"),
+							corev1.ResourceMemory: resource.MustParse("64Mi"),
+						},
+					},
+				}
+				clf.Spec.ServiceAccount.Name = serviceAccount.Name
+			}).
+				FromInput(obs.InputTypeApplication).
+				ToSyslogOutput(obs.SyslogRFC5424, func(output *obs.OutputSpec) {
+					output.Syslog = &obs.Syslog{
+						URL: fmt.Sprintf("udp://%s.%s.svc:514", framework.SyslogReceiverName, deployNS),
+						RFC: obs.SyslogRFC5424,
+						Tuning: &obs.SyslogTuningSpec{
+							DeliveryMode: obs.DeliveryModeAtLeastOnce,
+						},
+						Facility:   "local0",
+						Enrichment: obs.EnrichmentTypeKubernetesMinimal,
+						AppName:    `{.systemd.u.SYSLOG_IDENTIFIER||.log_type||"-"}`,
+						ProcId:     `{.systemd.t.PID||"-"}`,
+						MsgId:      `{.systemd.u.MESSAGE_ID||"-"}`,
+					}
+				}).End()
+
+			if err := e2e.CreateObservabilityClusterLogForwarder(forwarder); err != nil {
+				Fail(fmt.Sprintf("Unable to create an instance of logforwarder: %v", err))
+			}
+
+			By("Waiting for the collector daemonset")
+			if err := e2e.WaitForDaemonSet(forwarder.Namespace, forwarder.Name); err != nil {
+				Fail(err.Error())
+			}
+		})
+
+		Context("should recover and send logs after syslog UDP receiver restarts", func() {
+			var podDeleteTest = func() {
+				By("Verify initial logs are flowing")
+				logStore := e2e.LogStores[syslogDeployment.GetName()]
+				Expect(logStore.HasApplicationLogs(time.Minute*2)).To(BeTrue(), "expected to collect application logs initially")
+
+				ctx := context.TODO()
+				labelSelector := fmt.Sprintf("%s=%s", constants.LabelK8sComponent, syslogDeployment.Name)
+
+				By("Delete syslog receiver pod repeatedly to trigger ICMP error caching on vector's UDP socket")
+				for i := 0; i < 10; i++ {
+					err = e2e.KubeClient.CoreV1().Pods(deployNS).DeleteCollection(ctx,
+						metav1.DeleteOptions{GracePeriodSeconds: utils.GetPtr[int64](0)},
+						metav1.ListOptions{LabelSelector: labelSelector},
+					)
+					Expect(err).To(BeNil(), "should be able to delete syslog receiver pods")
+					time.Sleep(5 * time.Second)
+				}
+				Expect(e2e.WaitForDeployment(deployNS, syslogDeployment.Name, time.Second, time.Second*30)).To(Succeed(), "replicas should become available ")
+
+				By("Verify that logs resume flowing after recovery")
+				Expect(logStore.HasApplicationLogs(1*time.Minute)).To(BeTrue(), "expected to collect application logs after receiver recovers")
+				//Fail("I should never get here")
+			}
+
+			It("when using a ClusterIP service", func() {
+				podDeleteTest()
+			})
+
+			It("when using a NodePort service", func() {
+				Expect(e2e.Test.Delete(runtime.NewService(deployNS, framework.SyslogReceiverName))).To(Succeed(), "should delete syslog receiver service")
+				By("Replacing the ClusterIP service with a NodePort service")
+				_, err = e2e.CreateSyslogService(syslogDeployment, corev1.ProtocolUDP, corev1.ServiceTypeNodePort)
+				Expect(err).To(BeNil(), "should successfully create NodePort service")
+				podDeleteTest()
+			})
+		})
+
+		AfterEach(func() {
+			e2e.Cleanup()
+		})
+	})
+})

test/e2e/logforwarding/syslog/syslog_suite_test.go

@@ -0,0 +1,13 @@
+package syslog
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestSyslog(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Syslog Log Forwarding E2E Suite")
+}

test/framework/e2e/framework.go

@@ -237,7 +237,7 @@ func (tc *E2ETestFramework) Create(obj crclient.Object) error {
 	tc.AddCleanup(func() error {
 		return tc.Test.Delete(obj)
 	})
-	clolog.Info("Creating object", "obj", string(body))
+	clolog.V(2).Info("Creating object", "obj", string(body))
 	return tc.Test.Recreate(obj)
 }
 
@@ -247,7 +247,7 @@ func (tc *E2ETestFramework) CreateObservabilityClusterLogForwarder(forwarder *ob
 
 func DoCleanup() bool {
 	doCleanup := strings.TrimSpace(os.Getenv("DO_CLEANUP"))
-	clolog.Info("Running Cleanup script ....", "DO_CLEANUP", doCleanup)
+	clolog.V(1).Info("Running Cleanup script ....", "DO_CLEANUP", doCleanup)
 	return doCleanup == "" || strings.ToLower(doCleanup) == "true"
 }
 
@@ -264,7 +264,7 @@ func (tc *E2ETestFramework) Cleanup() {
 	} else {
 		clolog.V(1).Info("Test passed. Skipping artifacts gathering")
 	}
-	clolog.Info("Running e2e cleanup functions, ", "number", len(tc.CleanupFns))
+	clolog.V(1).Info("Running e2e cleanup functions, ", "number", len(tc.CleanupFns))
 	for _, cleanup := range tc.CleanupFns {
 		clolog.V(5).Info("Running an e2e cleanup function")
 		if err := cleanup(); err != nil {
@@ -282,14 +282,14 @@ func RunCleanupScript() {
 			clolog.Info("No cleanup script provided")
 			return
 		}
-		clolog.Info("Script", "CLEANUP_CMD", value)
+		clolog.V(1).Info("Script", "CLEANUP_CMD", value)
 		args := strings.Split(value, " ")
 		// #nosec G204
 		cmd := exec.Command(args[0], args[1:]...)
 		cmd.Env = nil
 		result, err := cmd.CombinedOutput()
-		clolog.Info("RunCleanupScript output: ", "output", string(result))
-		clolog.Info("RunCleanupScript err: ", "error", err)
+		clolog.V(2).Info("RunCleanupScript output: ", "output", string(result))
+		clolog.V(2).Info("RunCleanupScript err: ", "error", err)
 	}
 }