add e2e tests: label based routing works

Makefile

@@ -186,3 +186,7 @@ run-delve: generate fmt vet manifests
 .PHONY: tidy
 tidy: ## Tidy Go modules
 	find . -iname "go.mod" -not -path "./.devcontainer/*" | xargs -L1 sh -c 'cd $$(dirname $$0); go mod tidy'
+
+.PHONY: e2e-test
+e2e-test: ## Run e2e tests, make sure subscription operator is running somewhere
+	cd e2e && timeout --foreground 15m ./e2e_test.sh || (echo "E2E test failed"; exit 1)

docs/examples/one_tenant_two_subscriptions.yaml → docs/examples/simple-demo/one_tenant_two_subscriptions.yaml

@@ -37,7 +37,7 @@ spec:
 apiVersion: telemetry.kube-logging.dev/v1alpha1
 kind: Subscription
 metadata:
-  name: subscription-sample
+  name: subscription-sample-1
   namespace: example-tenant-ns
 spec:
   ottl: 'route()'
@@ -48,7 +48,7 @@ spec:
 apiVersion: telemetry.kube-logging.dev/v1alpha1
 kind: Subscription
 metadata:
-  name: another-subscription-sample
+  name: subscription-sample-2
   namespace: example-tenant-ns
 spec:
   ottl: 'route()'

e2e/e2e_test.sh

@@ -0,0 +1,74 @@
+#!/bin/env bash
+
+set -eou pipefail
+
+create_if_does_not_exist() {
+  local resource_type=$1
+  local resource_name=$2
+  kubectl create  "${resource_type}" "${resource_name}" --dry-run=client -o yaml | kubectl apply -f-
+}
+
+KIND_CLUSTER_NAME=${KIND_CLUSTER_NAME_E2E:-so-e2e}
+# Backup current kubernetes context
+CURRENT_K8S_CTX=$(kubectl config view | grep "current" | cut -f 2 -d : | xargs)
+
+# Prepare env
+kind create cluster --name "${KIND_CLUSTER_NAME}" --wait 5m
+kubectl config set-context kind-"${KIND_CLUSTER_NAME}"
+
+# Install prerequisites
+
+helm upgrade \
+  --install \
+  --repo https://charts.jetstack.io \
+  cert-manager cert-manager \
+  --namespace cert-manager \
+  --create-namespace \
+  --version v1.13.3 \
+  --set installCRDs=true \
+  --wait
+
+kubectl apply -f https://github.com/open-telemetry/opentelemetry-operator/releases/latest/download/opentelemetry-operator.yaml --wait
+echo "Wait until otel operator pod is in ready state..."
+kubectl wait --namespace opentelemetry-operator-system --for=condition=available deployment/opentelemetry-operator-controller-manager --timeout=300s
+
+# Create subscription operator resources
+(cd .. && make manifests generate install)
+
+# Use example
+kubectl apply -f ../docs/examples/simple-demo
+
+
+(cd .. && timeout 5m make run &)
+
+# Create log-generator
+helm install --wait --create-namespace --namespace example-tenant-ns --generate-name oci://ghcr.io/kube-logging/helm-charts/log-generator
+
+
+# Check for received messages - subscription-sample
+while
+  echo "Checking for subscription-sample-1 in deployments/receiver-collector logs"
+  kubectl logs --namespace example-tenant-ns deployments/receiver-collector | grep -q "subscription-sample-1"
+
+  [[ $? -ne 0 ]]
+do true; done
+
+# Check for received messages - subscription-sample-2
+while
+  echo "Checking for subscription-sample-2 in deployments/receiver-collector logs"
+  kubectl logs --namespace example-tenant-ns deployments/receiver-collector | grep -q "subscription-sample-2"
+
+  [[ $? -ne 0 ]]
+do true; done
+
+echo "E2E test: PASSED"
+
+
+# Check if cluster should be removed, ctx restored
+if [[ -z "${NO_KIND_CLEANUP}" ]]; then
+  kind delete cluster --name "${KIND_CLUSTER_NAME}"
+fi
+
+if [[ "${CURRENT_K8S_CTX}" != "" ]]; then
+  kubectl config get-contexts -o name | grep -q "${CURRENT_K8S_CTX}" && kubectl config set-context "${CURRENT_K8S_CTX}"
+fi

internal/controller/telemetry/otel_conf_gen.go

@@ -41,7 +41,7 @@ type RoutingConnectorTableItem struct {
 
 type RoutingConnector struct {
 	Name             string                      `yaml:"-"`
-	DefaultPipelines []string                    `yaml:"default_pipelines,flow"`
+	DefaultPipelines []string                    `yaml:"default_pipelines,flow,omitempty"`
 	Table            []RoutingConnectorTableItem `yaml:"table"`
 }
 
@@ -84,6 +84,9 @@ type OtelColConfigIR struct {
 
 func (cfgInput *OtelColConfigInput) generateExporters() map[string]any {
 	exporters := cfgInput.generateOTLPExporters()
+	exporters["logging/debug"] = map[string]any{
+		"verbosity": "detailed",
+	}
 	return exporters
 }
 
@@ -353,6 +356,7 @@ func (cfgInput *OtelColConfigInput) generateDefaultKubernetesProcessor() map[str
 
 func (cfgInput *OtelColConfigInput) generateDefaultKubernetesReceiver() map[string]any {
 
+	// TODO: fix parser-crio
 	operators := []map[string]any{
 		{
 			"type": "router",
@@ -362,27 +366,12 @@ func (cfgInput *OtelColConfigInput) generateDefaultKubernetesReceiver() map[stri
 					"output": "parser-docker",
 					"expr":   `body matches "^\\{"`,
 				},
-				{
-					"output": "parser-crio",
-					"expr":   `body matches "^[^ Z]+ "`,
-				},
 				{
 					"output": "parser-containerd",
 					"expr":   `body matches "^[^ Z]+Z"`,
 				},
 			},
 		},
-		{
-			"type":   "regex_parser",
-			"id":     "parser-crio",
-			"regex":  `'^(?P<time>[^ Z]+) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$'`,
-			"output": "extract_metadata_from_filepath",
-			"timestamp": map[string]string{
-				"parse_from":  "attributes.time",
-				"layout_type": "gotime",
-				"layout":      "2006-01-02T15:04:05.999999999Z07:00",
-			},
-		},
 		{
 			"type":   "regex_parser",
 			"id":     "parser-containerd",
@@ -479,6 +468,8 @@ func (cfgInput *OtelColConfigInput) ToIntermediateRepresentation() *OtelColConfi
 
 	result.Services.Pipelines.NamedPipelines = cfgInput.generateNamedPipelines()
 
+	result.Services.Telemetry = make(map[string]any)
+
 	return &result
 }