diff --git a/controllers/logging_controller.go b/controllers/logging_controller.go
index ea120fc31..26c6148bc 100644
--- a/controllers/logging_controller.go
+++ b/controllers/logging_controller.go
@@ -263,8 +263,14 @@ func (r *LoggingReconciler) GetResources(logging *loggingv1beta1.Logging) (*mode
 	loggingResources := model.NewLoggingResources(logging, r.Client, r.Log)
 	var err error
 
+	listOptsForClusterResources := []client.ListOption{}
+
+	if !logging.Spec.AllowClusterResourcesFromAllNamespaces {
+		listOptsForClusterResources = append(listOptsForClusterResources, client.InNamespace(logging.Spec.ControlNamespace))
+	}
+
 	clusterFlows := &loggingv1beta1.ClusterFlowList{}
-	err = r.List(context.TODO(), clusterFlows, client.InNamespace(logging.Spec.ControlNamespace))
+	err = r.List(context.TODO(), clusterFlows, listOptsForClusterResources...)
 	if err != nil {
 		return nil, err
 	}
@@ -288,7 +294,7 @@ func (r *LoggingReconciler) GetResources(logging *loggingv1beta1.Logging) (*mode
 	}
 
 	clusterOutputs := &loggingv1beta1.ClusterOutputList{}
-	err = r.List(context.TODO(), clusterOutputs, client.InNamespace(logging.Spec.ControlNamespace))
+	err = r.List(context.TODO(), clusterOutputs, listOptsForClusterResources...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/controllers/logging_controller_test.go b/controllers/logging_controller_test.go
index dd15306b5..381a2e333 100644
--- a/controllers/logging_controller_test.go
+++ b/controllers/logging_controller_test.go
@@ -264,6 +264,58 @@ func TestSingleFlowWithClusterOutput(t *testing.T) {
 	g.Expect(string(secret.Data[fluentd.AppConfigKey])).Should(gomega.ContainSubstring("a:b"))
 }
 
+func TestSingleClusterFlowWithClusterOutputFromExternalNamespace(t *testing.T) {
+	g := gomega.NewGomegaWithT(t)
+	defer beforeEach(t)()
+
+	logging := &v1beta1.Logging{
+		ObjectMeta: v1.ObjectMeta{
+			Name: "test-" + uuid.New()[:8],
+		},
+		Spec: v1beta1.LoggingSpec{
+			WatchNamespaces:                        []string{testNamespace},
+			FluentdSpec:                            &v1beta1.FluentdSpec{},
+			FlowConfigCheckDisabled:                true,
+			ControlNamespace:                       controlNamespace,
+			AllowClusterResourcesFromAllNamespaces: true,
+		},
+	}
+
+	output := &v1beta1.ClusterOutput{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "test-cluster-output",
+			Namespace: testNamespace,
+		},
+		Spec: v1beta1.ClusterOutputSpec{
+			OutputSpec: v1beta1.OutputSpec{
+				NullOutputConfig: output.NewNullOutputConfig(),
+			},
+		},
+	}
+
+	flow := &v1beta1.ClusterFlow{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "test-flow",
+			Namespace: testNamespace,
+		},
+		Spec: v1beta1.ClusterFlowSpec{
+			Selectors: map[string]string{
+				"a": "b",
+			},
+			OutputRefs: []string{"test-cluster-output"},
+		},
+	}
+
+	defer ensureCreated(t, logging)()
+	defer ensureCreated(t, output)()
+	defer ensureCreated(t, flow)()
+
+	secret := &corev1.Secret{}
+	defer ensureCreatedEventually(t, controlNamespace, logging.QualifiedName(fluentd.AppSecretConfigName), secret)()
+
+	g.Expect(string(secret.Data[fluentd.AppConfigKey])).Should(gomega.ContainSubstring("a:b"))
+}
+
 func TestClusterFlowWithNamespacedOutput(t *testing.T) {
 	defer beforeEach(t)()
 
diff --git a/pkg/sdk/api/v1beta1/clusterflow_types.go b/pkg/sdk/api/v1beta1/clusterflow_types.go
index 40314606b..9cc1660ce 100644
--- a/pkg/sdk/api/v1beta1/clusterflow_types.go
+++ b/pkg/sdk/api/v1beta1/clusterflow_types.go
@@ -61,6 +61,7 @@ type ClusterExclude struct {
 
 // FlowSpec is the Kubernetes spec for Flows
 type ClusterFlowSpec struct {
+	// Deprecated
 	Selectors  map[string]string `json:"selectors,omitempty"`
 	Match      []ClusterMatch    `json:"match,omitempty"`
 	Filters    []Filter          `json:"filters,omitempty"`
diff --git a/pkg/sdk/api/v1beta1/logging_types.go b/pkg/sdk/api/v1beta1/logging_types.go
index 3d93d51c7..31d9c4d9d 100644
--- a/pkg/sdk/api/v1beta1/logging_types.go
+++ b/pkg/sdk/api/v1beta1/logging_types.go
@@ -41,14 +41,15 @@ type _metaLoggingSpec interface{}
 
 // LoggingSpec defines the desired state of Logging
 type LoggingSpec struct {
-	LoggingRef              string           `json:"loggingRef,omitempty"`
-	FlowConfigCheckDisabled bool             `json:"flowConfigCheckDisabled,omitempty"`
-	FlowConfigOverride      string           `json:"flowConfigOverride,omitempty"`
-	FluentbitSpec           *FluentbitSpec   `json:"fluentbit,omitempty"`
-	FluentdSpec             *FluentdSpec     `json:"fluentd,omitempty"`
-	DefaultFlowSpec         *DefaultFlowSpec `json:"defaultFlow,omitempty"`
-	WatchNamespaces         []string         `json:"watchNamespaces,omitempty"`
-	ControlNamespace        string           `json:"controlNamespace"`
+	LoggingRef                             string           `json:"loggingRef,omitempty"`
+	FlowConfigCheckDisabled                bool             `json:"flowConfigCheckDisabled,omitempty"`
+	FlowConfigOverride                     string           `json:"flowConfigOverride,omitempty"`
+	FluentbitSpec                          *FluentbitSpec   `json:"fluentbit,omitempty"`
+	FluentdSpec                            *FluentdSpec     `json:"fluentd,omitempty"`
+	DefaultFlowSpec                        *DefaultFlowSpec `json:"defaultFlow,omitempty"`
+	WatchNamespaces                        []string         `json:"watchNamespaces,omitempty"`
+	ControlNamespace                       string           `json:"controlNamespace"`
+	AllowClusterResourcesFromAllNamespaces bool             `json:"allowClusterResourcesFromAllNamespaces"`
 
 	// EnableRecreateWorkloadOnImmutableFieldChange enables the operator to recreate the
 	// fluentbit daemonset and the fluentd statefulset (and possibly other resource in the future)