LOG-7572: NetworkPolicy for ClusterLogForwarder

Clee2691 · Clee2691 · commit 01efb83d6bbc · 2025-09-04T00:32:14.000-04:00
diff --git a/api/observability/v1/output_types.go b/api/observability/v1/output_types.go
@@ -744,7 +744,7 @@ type Kafka struct {
 	Brokers []BrokerURL `json:"brokers,omitempty"`
 }
 
-// +kubebuilder:validation:XValidation:rule="self == '' || (isURL(self) && (self.startsWith('tcp://') || self.startsWith('tls://')))",message="each broker must be a valid URL with a tcp or tls scheme"
+// +kubebuilder:validation:XValidation:rule="isURL(self) && (self.startsWith('tcp://') || self.startsWith('tls://'))",message="each broker must be a valid URL with a tcp or tls scheme"
 type BrokerURL string
 
 type LokiTuningSpec struct {
diff --git a/bundle/manifests/cluster-logging.clusterserviceversion.yaml b/bundle/manifests/cluster-logging.clusterserviceversion.yaml
@@ -82,7 +82,7 @@ metadata:
     categories: OpenShift Optional, Logging & Tracing, Observability
     certified: "false"
     containerImage: quay.io/openshift-logging/cluster-logging-operator:latest
-    createdAt: "2025-07-29T17:44:27Z"
+    createdAt: "2025-09-03T01:08:59Z"
     description: The Red Hat OpenShift Logging Operator for OCP provides a means for
       configuring and managing log collection and forwarding.
     features.operators.openshift.io/cnf: "false"
@@ -215,15 +215,15 @@ spec:
 
           Examples:
 
-           - `.kubernetes.namespace_name`
+           - `.kubernetes.namespace_id`
 
-           - `.log_type`
+           - `.hostname`
 
            - '.kubernetes.labels.foobar'
 
            - `.kubernetes.labels."foo-bar/baz"`
 
-          NOTE1: `In` CANNOT contain `.log_type` or `.message` as those fields are required and cannot be pruned.
+          NOTE1: `In` CANNOT contain `.log_type`, `.log_source` or `.message` as those fields are required and cannot be pruned.
 
           NOTE2: If this filter is used in a pipeline with GoogleCloudLogging, `.hostname` CANNOT be added to this list as it is a required field.
         displayName: Fields to be dropped
@@ -243,11 +243,13 @@ spec:
 
            - `.log_type`
 
-           - '.kubernetes.labels.foobar'
+           - '.log_source'
+
+           - '.message'
 
            - `.kubernetes.labels."foo-bar/baz"`
 
-          NOTE1: `NotIn` MUST contain `.log_type` and `.message` as those fields are required and cannot be pruned.
+          NOTE1: `NotIn` MUST contain `.log_type`, `.log_source` and `.message` as those fields are required and cannot be pruned.
 
           NOTE2: If this filter is used in a pipeline with GoogleCloudLogging, `.hostname` MUST be added to this list as it is a required field.
         displayName: Fields to be kept
@@ -1991,6 +1993,9 @@ spec:
           verbs:
           - create
           - delete
+          - list
+          - get
+          - watch
         - apiGroups:
           - observability.openshift.io
           resources:
diff --git a/bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml b/bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml
@@ -2535,8 +2535,7 @@ spec:
                             x-kubernetes-validations:
                             - message: each broker must be a valid URL with a tcp
                                 or tls scheme
-                              rule: self == '' || (isURL(self) && (self.startsWith('tcp://')
-                                || self.startsWith('tls://')))
+                              rule: isURL(self) && (self.startsWith('tcp://') || self.startsWith('tls://'))
                           type: array
                         topic:
                           description: |-
diff --git a/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml b/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml
@@ -2535,8 +2535,7 @@ spec:
                             x-kubernetes-validations:
                             - message: each broker must be a valid URL with a tcp
                                 or tls scheme
-                              rule: self == '' || (isURL(self) && (self.startsWith('tcp://')
-                                || self.startsWith('tls://')))
+                              rule: isURL(self) && (self.startsWith('tcp://') || self.startsWith('tls://'))
                           type: array
                         topic:
                           description: |-
diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml
@@ -78,6 +78,9 @@ rules:
   verbs:
   - create
   - delete
+  - list
+  - get
+  - watch
 - apiGroups:
   - observability.openshift.io
   resources:
diff --git a/docs/features/collection.adoc b/docs/features/collection.adoc
@@ -95,6 +95,7 @@ a|
 
 |https://issues.redhat.com/browse/LOG-3270[TLS Security Profile Compliance]
 |Comply with OCP cluster-wide cryptographic profiles for internal communication and allow configuration of outbound connection profiles. See link:./tls_security_profile.adoc[details]
+|https://issues.redhat.com/browse/LOG-7571[Network Policy]| Network policy in place for the collectors that allows all egress and ingress.
 |======
 
 === Tuning
diff --git a/docs/features/network_policy.adoc b/docs/features/network_policy.adoc
@@ -0,0 +1,104 @@
+= NetworkPolicy for Collector Pods
+
+The Cluster Logging Operator automatically creates and manages a `NetworkPolicy` for its collector pods to ensure they function in restrictive network environments, even if a cluster-wide `AdminNetworkPolicy` would otherwise block their traffic.
+
+== Overview
+
+When a `ClusterLogForwarder` is deployed, the operator creates a permissive `NetworkPolicy` that allows all ingress and egress traffic for the collector pods. This ensures that log collection can function properly even when:
+
+* Restrictive default `NetworkPolicies` are in place
+* `AdminNetworkPolicy` configurations limit pod communications
+* Namespace-level network restrictions are applied
+
+The `NetworkPolicy` is automatically created and removed along with the collector deployment lifecycle. 
+
+The `NetworkPolicy` can directly be edited by the cluster administrator and won't be reconciled by the operator upon updates.
+
+== NetworkPolicy Configuration
+
+The operator creates a `NetworkPolicy` for the collector with the following characteristics:
+
+```yaml
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: <COLLECTOR-INSTANCE-NAME>
+  namespace: <COLLECTOR-NAMESPACE>
+  labels:
+    app.kubernetes.io/name: vector
+    app.kubernetes.io/instance: <COLLECTOR-INSTANCE-NAME>
+    app.kubernetes.io/component: collector
+    app.kubernetes.io/part-of: cluster-logging
+    app.kubernetes.io/managed-by: cluster-logging-operator
+    app.kubernetes.io/version: <CLO-VERSION>
+spec:
+  podSelector:
+    matchLabels:
+      app.kubernetes.io/name: vector
+      app.kubernetes.io/instance: <COLLECTOR-INSTANCE-NAME>
+      app.kubernetes.io/component: collector
+      app.kubernetes.io/part-of: cluster-logging
+      app.kubernetes.io/managed-by: cluster-logging-operator
+  policyTypes:
+    - Ingress
+    - Egress
+  ingress:
+    - {}  # Allow all ingress traffic
+  egress:
+    - {}  # Allow all egress traffic
+```
+
+== AdminNetworkPolicy Delegation
+
+When an `AdminNetworkPolicy` (ANP) is used in your cluster to enforce network restrictions, you may need to configure delegation to allow the collector's `NetworkPolicy` to take precedence for log collection traffic.
+
+=== Understanding the Hierarchy
+
+OpenShift network policy precedence (highest to lowest priority):
+
+1. **AdminNetworkPolicy** - Cluster-admin controlled, highest priority
+2. **BaselineAdminNetworkPolicy** - Default fallback rules  
+3. **NetworkPolicy** - Namespace-level policies (where collector policies reside)
+
+=== Delegation Configuration
+
+To ensure collector pods can communicate properly when an `AdminNetworkPolicy` is blocking traffic, create an `AdminNetworkPolicy` rule that delegates to `NetworkPolicy` for collector traffic:
+
+==== Example: Delegating Collector Traffic
+
+```yaml
+apiVersion: policy.networking.k8s.io/v1alpha1
+kind: AdminNetworkPolicy
+metadata:
+  name: allow-logging-collector-delegation
+spec:
+  priority: 50  # Adjust based on your cluster's ANP priority scheme. Lower number means higher priority
+  subject:
+    pods: # Target the collector pods
+      namespaceSelector:
+        matchLabels:
+          kubernetes.io/metadata.name: openshift-logging # or collector namespace
+      podSelector:
+        matchLabels:
+          app.kubernetes.io/name: vector
+          app.kubernetes.io/instance: my-clf # or collector instance name
+          app.kubernetes.io/managed-by: cluster-logging-operator
+          app.kubernetes.io/part-of: cluster-logging
+          app.kubernetes.io/component: collector
+  ingress:
+    - name: "delegate-to-collector-ingress"
+      action: "Pass"  # Pass to collector NetworkPolicy
+      from:
+        - {}  # Delegate decisions for traffic coming from any source
+  egress:
+    - name: "delegate-to-collector-egress"
+      action: "Pass"  # Pass to collector NetworkPolicy
+      to:
+        - {}  # Delegate decisions for traffic going to any destination
+```
+
+== References
+
+- https://docs.redhat.com/en/documentation/openshift_container_platform/4.19/html/network_security/admin-network-policy#adminnetworkpolicy_ovn-k-anp[Openshift AdminNetworkPolicy]
+- https://docs.openshift.com/container-platform/latest/networking/network_policy/about-network-policy.html[OpenShift Network Policy]
+- https://kubernetes.io/docs/concepts/services-networking/network-policies/[Kubernetes NetworkPolicy Documentation]
diff --git a/docs/reference/operator/api_observability_v1.adoc b/docs/reference/operator/api_observability_v1.adoc
@@ -1238,15 +1238,15 @@ If segments contain characters outside of this range, the segment must be quoted
 
 Examples:
 
-- `.kubernetes.namespace_name`
+- `.kubernetes.namespace_id`
 
-- `.log_type`
+- `.hostname`
 
 - &#39;.kubernetes.labels.foobar&#39;
 
 - `.kubernetes.labels.&#34;foo-bar/baz&#34;`
 
-NOTE1: `In` CANNOT contain `.log_type` or `.message` as those fields are required and cannot be pruned.
+NOTE1: `In` CANNOT contain `.log_type`, `.log_source` or `.message` as those fields are required and cannot be pruned.
 
 NOTE2: If this filter is used in a pipeline with GoogleCloudLogging, `.hostname` CANNOT be added to this list as it is a required field.
 
@@ -1264,11 +1264,13 @@ Examples:
 
 - `.log_type`
 
-- &#39;.kubernetes.labels.foobar&#39;
+- &#39;.log_source&#39;
+
+- &#39;.message&#39;
 
 - `.kubernetes.labels.&#34;foo-bar/baz&#34;`
 
-NOTE1: `NotIn` MUST contain `.log_type` and `.message` as those fields are required and cannot be pruned.
+NOTE1: `NotIn` MUST contain `.log_type`, `.log_source` and `.message` as those fields are required and cannot be pruned.
 
 NOTE2: If this filter is used in a pipeline with GoogleCloudLogging, `.hostname` MUST be added to this list as it is a required field.
 
diff --git a/internal/controller/observability/clusterlogforwarder_controller.go b/internal/controller/observability/clusterlogforwarder_controller.go
@@ -7,6 +7,7 @@ import (
 
 	v1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	appsv1 "k8s.io/api/apps/v1"
+	networkingv1 "k8s.io/api/networking/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -206,6 +207,7 @@ func (r *ClusterLogForwarderReconciler) SetupWithManager(mgr ctrl.Manager) error
 		Owns(&corev1.ConfigMap{}).
 		Owns(&appsv1.DaemonSet{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
 		Owns(&appsv1.Deployment{}).
+		Owns(&networkingv1.NetworkPolicy{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
 		Owns(&rbacv1.Role{}).
 		Owns(&rbacv1.RoleBinding{}).
 		Owns(&corev1.Secret{}).
diff --git a/internal/controller/observability/collector.go b/internal/controller/observability/collector.go
@@ -1,6 +1,7 @@
 package observability
 
 import (
+	"fmt"
 	"strings"
 	"time"
 
@@ -139,6 +140,12 @@ func ReconcileCollector(context internalcontext.ForwarderContext, pollInterval,
 		return err
 	}
 
+	// Reconcile NetworkPolicy for the collector daemonset
+	if err := network.ReconcileNetworkPolicy(context.Client, context.Forwarder.Namespace, fmt.Sprintf("%s-%s", constants.CollectorName, resourceNames.CommonName), context.Forwarder.Name, ownerRef, collectorFactory.CommonLabelInitializer); err != nil {
+		log.Error(err, "collector.ReconcileNetworkPolicy")
+		return err
+	}
+
 	// Reconcile resources to support metrics gathering
 	if err := network.ReconcileService(context.Client, context.Forwarder.Namespace, resourceNames.CommonName, context.Forwarder.Name, constants.CollectorName, collector.MetricsPortName, resourceNames.SecretMetrics, collector.MetricsPort, ownerRef, collectorFactory.CommonLabelInitializer); err != nil {
 		log.Error(err, "collector.ReconcileService")
diff --git a/internal/factory/network_policy.go b/internal/factory/network_policy.go
@@ -0,0 +1,39 @@
+package factory
+
+import (
+	"github.com/openshift/cluster-logging-operator/internal/constants"
+	"github.com/openshift/cluster-logging-operator/internal/runtime"
+	networkingv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// NewNetworkPolicy creates a NetworkPolicy for clfs.
+// It configures the policy to allow all ingress and egress traffic for the collector pods.
+func NewNetworkPolicy(namespace, policyName, instanceName string, visitors ...func(o runtime.Object)) *networkingv1.NetworkPolicy {
+	// Create the base NetworkPolicy
+	np := runtime.NewNetworkPolicy(namespace, policyName, visitors...)
+
+	// Set up pod selector to match collector pods for this instance
+	podSelector := runtime.Selectors(instanceName, constants.CollectorName, constants.VectorName)
+
+	// Configure the NetworkPolicy spec
+	np.Spec = networkingv1.NetworkPolicySpec{
+		PodSelector: metav1.LabelSelector{
+			MatchLabels: podSelector,
+		},
+		PolicyTypes: []networkingv1.PolicyType{
+			networkingv1.PolicyTypeIngress,
+			networkingv1.PolicyTypeEgress,
+		},
+		// Allow all ingress traffic
+		Ingress: []networkingv1.NetworkPolicyIngressRule{
+			{}, // Empty rule allows all ingress
+		},
+		// Allow all egress traffic
+		Egress: []networkingv1.NetworkPolicyEgressRule{
+			{}, // Empty rule allows all egress
+		},
+	}
+
+	return np
+}
diff --git a/internal/factory/network_policy_test.go b/internal/factory/network_policy_test.go
@@ -0,0 +1,77 @@
+package factory
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/openshift/cluster-logging-operator/internal/constants"
+	"github.com/openshift/cluster-logging-operator/internal/runtime"
+	"github.com/openshift/cluster-logging-operator/version"
+	networkingv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var _ = Describe("#NewNetworkPolicy", func() {
+
+	var (
+		np           *networkingv1.NetworkPolicy
+		namespace    = "openshift-logging"
+		policyName   = "policy-name"
+		instanceName = "instance-name"
+		commonLabels = func(o runtime.Object) {
+			runtime.SetCommonLabels(o, constants.VectorName, instanceName, constants.CollectorName)
+		}
+	)
+
+	BeforeEach(func() {
+		np = NewNetworkPolicy(namespace, policyName, instanceName, commonLabels)
+	})
+
+	It("should set name and namespace correctly", func() {
+		Expect(np.Name).To(Equal(policyName))
+		Expect(np.Namespace).To(Equal(namespace))
+	})
+
+	It("should set labels correctly", func() {
+		expectedLabels := map[string]string{
+			constants.LabelK8sName:      constants.VectorName,
+			constants.LabelK8sInstance:  instanceName,
+			constants.LabelK8sComponent: constants.CollectorName,
+			constants.LabelK8sPartOf:    constants.ClusterLogging,
+			constants.LabelK8sManagedBy: constants.ClusterLoggingOperator,
+			constants.LabelK8sVersion:   version.Version,
+		}
+
+		Expect(np.Labels).To(Equal(expectedLabels))
+	})
+
+	It("should set pod selector to match collector pods", func() {
+		expectedPodSelector := metav1.LabelSelector{
+			MatchLabels: map[string]string{
+				constants.LabelK8sName:      constants.VectorName,
+				constants.LabelK8sInstance:  instanceName,
+				constants.LabelK8sComponent: constants.CollectorName,
+				constants.LabelK8sPartOf:    constants.ClusterLogging,
+				constants.LabelK8sManagedBy: constants.ClusterLoggingOperator,
+			},
+		}
+		Expect(np.Spec.PodSelector).To(Equal(expectedPodSelector))
+	})
+
+	It("should include both Ingress and Egress policy types", func() {
+		expectedPolicyTypes := []networkingv1.PolicyType{
+			networkingv1.PolicyTypeIngress,
+			networkingv1.PolicyTypeEgress,
+		}
+		Expect(np.Spec.PolicyTypes).To(Equal(expectedPolicyTypes))
+	})
+
+	It("should have ingress rules that allow all traffic", func() {
+		Expect(np.Spec.Ingress).To(HaveLen(1))
+		Expect(np.Spec.Ingress[0]).To(Equal(networkingv1.NetworkPolicyIngressRule{}))
+	})
+
+	It("should have egress rules that allow all traffic", func() {
+		Expect(np.Spec.Egress).To(HaveLen(1))
+		Expect(np.Spec.Egress[0]).To(Equal(networkingv1.NetworkPolicyEgressRule{}))
+	})
+})
diff --git a/internal/network/network_policy.go b/internal/network/network_policy.go
diff --git a/internal/network/network_policy_test.go b/internal/network/network_policy_test.go
diff --git a/internal/reconcile/networkpolicy.go b/internal/reconcile/networkpolicy.go
diff --git a/internal/runtime/core.go b/internal/runtime/core.go

Original file line number	Diff line number	Diff line change
`@@ -744,7 +744,7 @@ type Kafka struct {`
`744`	`744`	Brokers []BrokerURL `json:"brokers,omitempty"`
`745`	`745`	`}`
`746`	`746`
`747`		`-// +kubebuilder:validation:XValidation:rule="self == '' \|\| (isURL(self) && (self.startsWith('tcp://') \|\| self.startsWith('tls://')))",message="each broker must be a valid URL with a tcp or tls scheme"`
	`747`	`+// +kubebuilder:validation:XValidation:rule="isURL(self) && (self.startsWith('tcp://') \|\| self.startsWith('tls://'))",message="each broker must be a valid URL with a tcp or tls scheme"`
`748`	`748`	`type BrokerURL string`
`749`	`749`
`750`	`750`	`type LokiTuningSpec struct {`