Repository: banzaicloud/logging-operator
Pull request #357 (2020-02-12 15:22:30)

Label router with Match field for Exclude and Select
| Q               | A
| --------------- | ---
| Bug fix?        | no
| New feature?    | yes
| API breaks?     | no
| Deprecations?   | yes
| License         | Apache 2.0


### What's in this PR?
This PR contains major changes in the CRD spec!

- Deprecation of `selector` in `Flow` and `ClusterFlow`
   The attribute will still be available however using the `match` field is preferable.
- New field: `match`
   Differently from `selectors`, match is a *list* of `select` and `esclude` statement. The expressions
evaluated in order and the first positive or negative match will break the evaluation.

Detailed documentation will be available in the docs.

### Additional context
With this change we can handle several so far unavailable scenarios of log routing.

### Checklist
- [x] Manual testing upgrade and actual workflows
- [x] Error handling code meets the [guideline](https://github.com/banzaicloud/pipeline/blob/master/docs/error-handling-guide.md)
- [x] Logging code meets the guideline (TODO)
- [x] User guide and development docs updated (if needed)
- [x] Related Helm chart(s) updated (if needed)



Patch:
diff --git a/charts/logging-demo/Chart.yaml b/charts/logging-demo/Chart.yaml
index 015c8dfd..632d8f17 100644
--- a/charts/logging-demo/Chart.yaml
+++ b/charts/logging-demo/Chart.yaml
@@ -1,6 +1,6 @@
 apiVersion: v1
-appVersion: "2.7.0"
+appVersion: "3.0.0"
 description: A Helm chart for Kubernetes
 name: logging-demo
-version: 2.7.2
+version: 3.0.0
 icon: https://raw.githubusercontent.com/banzaicloud/logging-operator/master/docs/img/icon.png
diff --git a/charts/logging-demo/templates/flow.yaml b/charts/logging-demo/templates/flow.yaml
index 2a5a40ca..72266b42 100644
--- a/charts/logging-demo/templates/flow.yaml
+++ b/charts/logging-demo/templates/flow.yaml
@@ -13,9 +13,11 @@ spec:
         reserve_data: true
         parse:
           type: nginx
-  selectors:
-    app.kubernetes.io/name: log-generator
-    app.kubernetes.io/instance: {{ .Release.Name }}
+  match:
+    - select:
+        labels:
+          app.kubernetes.io/name: log-generator
+          app.kubernetes.io/instance: {{ .Release.Name }}
   outputRefs:
 
 {{- if or .Values.file.enabled (and (not .Values.cloudwatch.enabled) (not .Values.elasticsearch.enabled) (not .Values.kafka.enabled) (not .Values.loki.enabled) (not .Values.minio.enabled)) }}
diff --git a/charts/logging-operator/crds/logging.banzaicloud.io_clusterflows.yaml b/charts/logging-operator/crds/logging.banzaicloud.io_clusterflows.yaml
index 502c1150..cba81780 100644
--- a/charts/logging-operator/crds/logging.banzaicloud.io_clusterflows.yaml
+++ b/charts/logging-operator/crds/logging.banzaicloud.io_clusterflows.yaml
@@ -522,6 +522,25 @@ spec:
               type: array
             loggingRef:
               type: string
+            match:
+              items:
+                properties:
+                  exclude:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                  select:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                type: object
+              type: array
             outputRefs:
               items:
                 type: string
@@ -529,10 +548,10 @@ spec:
             selectors:
               additionalProperties:
                 type: string
+              description: Deprecated
               type: object
           required:
           - outputRefs
-          - selectors
           type: object
         status:
           description: FlowStatus defines the observed state of Flow
diff --git a/charts/logging-operator/crds/logging.banzaicloud.io_flows.yaml b/charts/logging-operator/crds/logging.banzaicloud.io_flows.yaml
index 19f1ebd3..96d90e9b 100644
--- a/charts/logging-operator/crds/logging.banzaicloud.io_flows.yaml
+++ b/charts/logging-operator/crds/logging.banzaicloud.io_flows.yaml
@@ -527,6 +527,25 @@ spec:
               type: array
             loggingRef:
               type: string
+            match:
+              items:
+                properties:
+                  exclude:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                  select:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                type: object
+              type: array
             outputRefs:
               items:
                 type: string
@@ -534,10 +553,10 @@ spec:
             selectors:
               additionalProperties:
                 type: string
+              description: Deprecated
               type: object
           required:
           - outputRefs
-          - selectors
           type: object
         status:
           description: FlowStatus defines the observed state of Flow
diff --git a/config/crd/bases/logging.banzaicloud.io_clusterflows.yaml b/config/crd/bases/logging.banzaicloud.io_clusterflows.yaml
index 502c1150..cba81780 100644
--- a/config/crd/bases/logging.banzaicloud.io_clusterflows.yaml
+++ b/config/crd/bases/logging.banzaicloud.io_clusterflows.yaml
@@ -522,6 +522,25 @@ spec:
               type: array
             loggingRef:
               type: string
+            match:
+              items:
+                properties:
+                  exclude:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                  select:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                type: object
+              type: array
             outputRefs:
               items:
                 type: string
@@ -529,10 +548,10 @@ spec:
             selectors:
               additionalProperties:
                 type: string
+              description: Deprecated
               type: object
           required:
           - outputRefs
-          - selectors
           type: object
         status:
           description: FlowStatus defines the observed state of Flow
diff --git a/config/crd/bases/logging.banzaicloud.io_flows.yaml b/config/crd/bases/logging.banzaicloud.io_flows.yaml
index 19f1ebd3..96d90e9b 100644
--- a/config/crd/bases/logging.banzaicloud.io_flows.yaml
+++ b/config/crd/bases/logging.banzaicloud.io_flows.yaml
@@ -527,6 +527,25 @@ spec:
               type: array
             loggingRef:
               type: string
+            match:
+              items:
+                properties:
+                  exclude:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                  select:
+                    properties:
+                      labels:
+                        additionalProperties:
+                          type: string
+                        type: object
+                    type: object
+                type: object
+              type: array
             outputRefs:
               items:
                 type: string
@@ -534,10 +553,10 @@ spec:
             selectors:
               additionalProperties:
                 type: string
+              description: Deprecated
               type: object
           required:
           - outputRefs
-          - selectors
           type: object
         status:
           description: FlowStatus defines the observed state of Flow
diff --git a/controllers/logging_controller.go b/controllers/logging_controller.go
index c61cbf53..98726dad 100644
--- a/controllers/logging_controller.go
+++ b/controllers/logging_controller.go
@@ -40,7 +40,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
-	loggingv1alpha2 "github.com/banzaicloud/logging-operator/pkg/sdk/api/v1beta1"
+	loggingv1beta1 "github.com/banzaicloud/logging-operator/pkg/sdk/api/v1beta1"
 )
 
 // LoggingReconciler reconciles a Logging object
@@ -57,7 +57,7 @@ func (r *LoggingReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 	_ = context.Background()
 	log := r.Log.WithValues("logging", req.NamespacedName)
 
-	logging := &loggingv1alpha2.Logging{}
+	logging := &loggingv1beta1.Logging{}
 	err := r.Client.Get(context.TODO(), req.NamespacedName, logging)
 	if err != nil {
 		// Object not found, return.  Created objects are automatically garbage collected.
@@ -113,7 +113,7 @@ func (r *LoggingReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 	return ctrl.Result{}, nil
 }
 
-func (r *LoggingReconciler) clusterConfiguration(logging *loggingv1alpha2.Logging) (string, *secret.MountSecrets, error) {
+func (r *LoggingReconciler) clusterConfiguration(logging *loggingv1beta1.Logging) (string, *secret.MountSecrets, error) {
 	if logging.Spec.FlowConfigOverride != "" {
 		return logging.Spec.FlowConfigOverride, nil, nil
 	}
@@ -143,10 +143,10 @@ func (r *LoggingReconciler) clusterConfiguration(logging *loggingv1alpha2.Loggin
 
 // SetupLoggingWithManager setup logging manager
 func SetupLoggingWithManager(mgr ctrl.Manager, logger logr.Logger) *ctrl.Builder {
-	clusterOutputSource := &source.Kind{Type: &loggingv1alpha2.ClusterOutput{}}
-	clusterFlowSource := &source.Kind{Type: &loggingv1alpha2.ClusterFlow{}}
-	outputSource := &source.Kind{Type: &loggingv1alpha2.Output{}}
-	flowSource := &source.Kind{Type: &loggingv1alpha2.Flow{}}
+	clusterOutputSource := &source.Kind{Type: &loggingv1beta1.ClusterOutput{}}
+	clusterFlowSource := &source.Kind{Type: &loggingv1beta1.ClusterFlow{}}
+	outputSource := &source.Kind{Type: &loggingv1beta1.Output{}}
+	flowSource := &source.Kind{Type: &loggingv1beta1.Flow{}}
 	secretSource := &source.Kind{Type: &corev1.Secret{}}
 
 	requestMapper := &handler.EnqueueRequestsFromMapFunc{
@@ -156,7 +156,7 @@ func SetupLoggingWithManager(mgr ctrl.Manager, logger logr.Logger) *ctrl.Builder
 				return nil
 			}
 			// get all the logging resources from the cache
-			loggingList := &loggingv1alpha2.LoggingList{}
+			loggingList := &loggingv1beta1.LoggingList{}
 			err = mgr.GetCache().List(context.TODO(), loggingList)
 			if err != nil {
 				logger.Error(err, "failed to list logging resources")
@@ -179,16 +179,16 @@ func SetupLoggingWithManager(mgr ctrl.Manager, logger logr.Logger) *ctrl.Builder
 				}
 				return requestList
 			}
-			if o, ok := object.(*loggingv1alpha2.ClusterOutput); ok {
+			if o, ok := object.(*loggingv1beta1.ClusterOutput); ok {
 				return reconcileRequestsForLoggingRef(loggingList, o.Spec.LoggingRef)
 			}
-			if o, ok := object.(*loggingv1alpha2.Output); ok {
+			if o, ok := object.(*loggingv1beta1.Output); ok {
 				return reconcileRequestsForLoggingRef(loggingList, o.Spec.LoggingRef)
 			}
-			if o, ok := object.(*loggingv1alpha2.Flow); ok {
+			if o, ok := object.(*loggingv1beta1.Flow); ok {
 				return reconcileRequestsForLoggingRef(loggingList, o.Spec.LoggingRef)
 			}
-			if o, ok := object.(*loggingv1alpha2.ClusterFlow); ok {
+			if o, ok := object.(*loggingv1beta1.ClusterFlow); ok {
 				return reconcileRequestsForLoggingRef(loggingList, o.Spec.LoggingRef)
 			}
 			return nil
@@ -196,7 +196,7 @@ func SetupLoggingWithManager(mgr ctrl.Manager, logger logr.Logger) *ctrl.Builder
 	}
 
 	builder := ctrl.NewControllerManagedBy(mgr).
-		For(&loggingv1alpha2.Logging{}).
+		For(&loggingv1beta1.Logging{}).
 		Owns(&corev1.Pod{}).
 		Watches(clusterOutputSource, requestMapper).
 		Watches(clusterFlowSource, requestMapper).
@@ -210,7 +210,7 @@ func SetupLoggingWithManager(mgr ctrl.Manager, logger logr.Logger) *ctrl.Builder
 	return builder
 }
 
-func reconcileRequestsForLoggingRef(loggingList *loggingv1alpha2.LoggingList, loggingRef string) []reconcile.Request {
+func reconcileRequestsForLoggingRef(loggingList *loggingv1beta1.LoggingList, loggingRef string) []reconcile.Request {
 	filtered := make([]reconcile.Request, 0)
 	for _, l := range loggingList.Items {
 		if l.Spec.LoggingRef == loggingRef {
@@ -248,11 +248,11 @@ func FluentbitWatches(builder *ctrl.Builder) *ctrl.Builder {
 }
 
 // GetResources collect all resources referenced by logging resource
-func (r *LoggingReconciler) GetResources(logging *loggingv1alpha2.Logging) (*model.LoggingResources, error) {
+func (r *LoggingReconciler) GetResources(logging *loggingv1beta1.Logging) (*model.LoggingResources, error) {
 	loggingResources := model.NewLoggingResources(logging, r.Client, r.Log)
 	var err error
 
-	clusterFlows := &loggingv1alpha2.ClusterFlowList{}
+	clusterFlows := &loggingv1beta1.ClusterFlowList{}
 	err = r.List(context.TODO(), clusterFlows, client.InNamespace(logging.Spec.ControlNamespace))
 	if err != nil {
 		return nil, err
@@ -265,7 +265,7 @@ func (r *LoggingReconciler) GetResources(logging *loggingv1alpha2.Logging) (*mod
 		}
 	}
 
-	clusterOutputs := &loggingv1alpha2.ClusterOutputList{}
+	clusterOutputs := &loggingv1beta1.ClusterOutputList{}
 	err = r.List(context.TODO(), clusterOutputs, client.InNamespace(logging.Spec.ControlNamespace))
 	if err != nil {
 		return nil, err
@@ -292,7 +292,7 @@ func (r *LoggingReconciler) GetResources(logging *loggingv1alpha2.Logging) (*mod
 	}
 
 	for _, ns := range watchNamespaces {
-		flows := &loggingv1alpha2.FlowList{}
+		flows := &loggingv1beta1.FlowList{}
 		err = r.List(context.TODO(), flows, client.InNamespace(ns))
 		if err != nil {
 			return nil, err
@@ -304,7 +304,7 @@ func (r *LoggingReconciler) GetResources(logging *loggingv1alpha2.Logging) (*mod
 				}
 			}
 		}
-		outputs := &loggingv1alpha2.OutputList{}
+		outputs := &loggingv1beta1.OutputList{}
 		err = r.List(context.TODO(), outputs, client.InNamespace(ns))
 		if err != nil {
 			return nil, err
diff --git a/docs/crds.md b/docs/crds.md
index 6380a4dc..3fd4dab1 100644
--- a/docs/crds.md
+++ b/docs/crds.md
@@ -413,7 +413,7 @@ Flows define a `logging flow` that defines the `filters` and `outputs`.
 ### Parameters
 | Name                    | Type              | Default | Description |
 |-------------------------|-------------------|---------|-------------|
-| selectors               | map[string]string | {}      | Kubernetes label selectors for the log. |
+| selectors (DEPRECATED)  | map[string]string | {}      | DEPRECATED inf favor of match. Kubernetes label selectors for the log. |
 | filters                 | [][Filter](./plugins/filters)          | []      | List of applied [filter](./plugins/filters).  |
 | loggingRef              | string | "" | Specified `logging` resource reference to connect `FLow` and `ClusterFlow` to |
 | outputRefs              | []string | [] | List of [Outputs](#Defining-outputs) or [ClusterOutputs](#Defining-outputs) names |
@@ -435,8 +435,10 @@ spec:
         format: ${namespace_name}.${pod_name}.${container_name}
   outputRefs:
     - s3-output
-  selectors:
-    app: nginx
+  match:
+    - select:
+        labels:
+          app: nginx
 ```
 
 
diff --git a/docs/crds/v1beta1/clusterflow_types.md b/docs/crds/v1beta1/clusterflow_types.md
index 694db6f2..529683e3 100644
--- a/docs/crds/v1beta1/clusterflow_types.md
+++ b/docs/crds/v1beta1/clusterflow_types.md
@@ -7,6 +7,30 @@
 | metadata | metav1.ObjectMeta | No | - |  |
 | spec | FlowSpec | No | - | Name of the logging cluster to be attached<br> |
 | status | FlowStatus | No | - |  |
+### ClusterMatch
+| Variable Name | Type | Required | Default | Description |
+|---|---|---|---|---|
+| select | *ClusterSelect | No | - |  |
+| exclude | *ClusterExclude | No | - |  |
+### ClusterSelect
+| Variable Name | Type | Required | Default | Description |
+|---|---|---|---|---|
+| namespaces | []string | No | - |  |
+| labels | map[string]string | No | - |  |
+### ClusterExclude
+| Variable Name | Type | Required | Default | Description |
+|---|---|---|---|---|
+| namespaces | []string | No | - |  |
+| labels | map[string]string | No | - |  |
+### ClusterFlowSpec
+#### FlowSpec is the Kubernetes spec for Flows
+
+| Variable Name | Type | Required | Default | Description |
+|---|---|---|---|---|
+| match | []ClusterMatch | No | - |  |
+| filters | []Filter | No | - |  |
+| loggingRef | string | No | - |  |
+| outputRefs | []string | Yes | - |  |
 ### ClusterFlowList
 #### ClusterFlowList contains a list of ClusterFlow
 
diff --git a/docs/crds/v1beta1/flow_types.md b/docs/crds/v1beta1/flow_types.md
index 3c94b344..3adfb7f5 100644
--- a/docs/crds/v1beta1/flow_types.md
+++ b/docs/crds/v1beta1/flow_types.md
@@ -3,10 +3,24 @@
 
 | Variable Name | Type | Required | Default | Description |
 |---|---|---|---|---|
-| selectors | map[string]string | Yes | - |  |
+| selectors | map[string]string | No | - | Deprecated<br> |
+| match | []Match | No | - |  |
 | filters | []Filter | No | - |  |
 | loggingRef | string | No | - |  |
 | outputRefs | []string | Yes | - |  |
+### Match
+| Variable Name | Type | Required | Default | Description |
+|---|---|---|---|---|
+| select | *Select | No | - |  |
+| exclude | *Exclude | No | - |  |
+### Select
+| Variable Name | Type | Required | Default | Description |
+|---|---|---|---|---|
+| labels | map[string]string | No | - |  |
+### Exclude
+| Variable Name | Type | Required | Default | Description |
+|---|---|---|---|---|
+| labels | map[string]string | No | - |  |
 ### Filter
 #### Filter definition for FlowSpec
 
diff --git a/docs/examples/logging_flow_geoip.yaml b/docs/examples/logging_flow_geoip.yaml
index e98b801c..33f1bce9 100644
--- a/docs/examples/logging_flow_geoip.yaml
+++ b/docs/examples/logging_flow_geoip.yaml
@@ -11,14 +11,17 @@ spec:
         parse:
           type: nginx
     - geoip:
+        geoip_lookup_keys: remote
+        backend_library: geoip2_c
         records:
           - city: ${city.names.en["remote"]}
-            latitude: ${location.latitude["remote"]}
-            longitude: ${location.longitude["remote"]}
+            location_array: '''[${location.longitude["remote"]},${location.latitude["remote"]}]'''
             country: ${country.iso_code["remote"]}
             country_name: ${country.names.en["remote"]}
             postal_code:  ${postal.code["remote"]}
   outputRefs:
-    - forward-output-sample
-  selectors:
-    app: nginx
+    - null-output-sample
+  match:
+    - select:
+        labels:
+          app: nginx
\ No newline at end of file
diff --git a/docs/examples/logging_flow_multiple_output.yaml b/docs/examples/logging_flow_multiple_output.yaml
index 07e35b48..ff66ed64 100644
--- a/docs/examples/logging_flow_multiple_output.yaml
+++ b/docs/examples/logging_flow_multiple_output.yaml
@@ -7,5 +7,7 @@ spec:
   outputRefs:
     - gcs-output-sample
     - s3-output-example
-  selectors:
-    app: nginx
+  match:
+    - select:
+        labels:
+          app: nginx
diff --git a/docs/examples/logging_flow_single_output.yaml b/docs/examples/logging_flow_single_output.yaml
index c4b18bc5..8a21d76a 100644
--- a/docs/examples/logging_flow_single_output.yaml
+++ b/docs/examples/logging_flow_single_output.yaml
@@ -6,5 +6,7 @@ metadata:
 spec:
   outputRefs:
     - forward-output-sample
-  selectors:
-    app: nginx
+  match:
+    - select:
+        labels:
+          app: nginx
diff --git a/docs/examples/logging_flow_with_dedot.yaml b/docs/examples/logging_flow_with_dedot.yaml
index 3ca0d151..82afde18 100644
--- a/docs/examples/logging_flow_with_dedot.yaml
+++ b/docs/examples/logging_flow_with_dedot.yaml
@@ -14,5 +14,7 @@ spec:
     - dedot: {}
   outputRefs:
     - s3-output
-  selectors:
-    app: nginx
+  match:
+    - select:
+        labels:
+          app: nginx
\ No newline at end of file
diff --git a/docs/examples/logging_flow_with_filters.yaml b/docs/examples/logging_flow_with_filters.yaml
index 597c2cba..bcfff586 100644
--- a/docs/examples/logging_flow_with_filters.yaml
+++ b/docs/examples/logging_flow_with_filters.yaml
@@ -13,5 +13,7 @@ spec:
         format: ${namespace_name}.${pod_name}.${container_name}
   outputRefs:
     - s3-output
-  selectors:
-    app: nginx
+  match:
+    - select:
+        labels:
+          app: nginx
diff --git a/docs/examples/logging_flow_with_multi_format.yaml b/docs/examples/logging_flow_with_multi_format.yaml
index 51b021df..c96e6002 100644
--- a/docs/examples/logging_flow_with_multi_format.yaml
+++ b/docs/examples/logging_flow_with_multi_format.yaml
@@ -16,6 +16,8 @@ spec:
       reserve_data: true
   outputRefs:
   - s3-output
-  selectors:
-    app.kubernetes.io/instance: nginx-demo
-    app.kubernetes.io/name: nginx-logging-demo
+  match:
+    - select:
+        labels:
+          app.kubernetes.io/instance: nginx-demo
+          app.kubernetes.io/name: nginx-logging-demo
diff --git a/docs/examples/logging_output_file.yaml b/docs/examples/logging_output_file.yaml
index 8ebce0f3..cdc6482e 100644
--- a/docs/examples/logging_output_file.yaml
+++ b/docs/examples/logging_output_file.yaml
@@ -2,11 +2,10 @@ apiVersion: logging.banzaicloud.io/v1beta1
 kind: Output
 metadata:
   name: file-output-sample
-spec:
 spec:
   file:
-    path: /tmp/logs/
+    path: /tmp/logs/${tag}/%Y/%m/%d/%H.%M
     buffer:
       timekey: 1m
       timekey_wait: 10s
-      timekey_use_utc: true
+      timekey_use_utc: true
\ No newline at end of file
diff --git a/docs/log-routing.md b/docs/log-routing.md
new file mode 100644
index 00000000..656e9912
--- /dev/null
+++ b/docs/log-routing.md
@@ -0,0 +1,142 @@
+<p align="center"><img src="./img/lo.svg" width="260"></p>
+
+# Routing your logs with match directive
+
+---
+
+The first step to process your logs is to select what logs goes to where.
+The logging operator uses Kubernetes labels and namespaces to separate
+different log flows.
+
+#### Match statement
+
+To select or exclude logs you can use the `match` statement. Match is a collection
+of `select` and `exclude` expressions. In both expression you can use the `labels`
+attribute to filter for pod's labels. Moreover in Cluster flow you can use `namespaces`
+as a selecting or excluding criteria.
+
+The list of `select` and `exclude` statements are evaluated **in order**!
+
+Flow
+```
+  kind: Flow
+  metadata:
+    name: flow-sample
+  spec:
+    match:
+      - exclude:
+          labels:
+            exclude-this: label
+      - select:
+          labels:
+            app: nginx
+            label/xxx: example
+```
+
+ClusterFlow
+```
+  kind: ClusterFlow
+  metadata:
+    name: flow-sample
+  spec:
+    match:
+      - exclude:
+          labels:
+            exclude-this: label
+          namespaces: developer 
+      - select:
+          labels:
+            app: nginx
+            label/xxx: example
+          namespaces: production,beta
+```
+
+## Examples
+
+#### 1. Select logs with `app: nginx` labels from the namespace
+
+  ```
+  apiVersion: logging.banzaicloud.io/v1beta1
+  kind: Flow
+  metadata:
+    name: flow-sample
+    namespace: default
+  spec:
+    outputRefs:
+      - forward-output-sample
+    match:
+      - select:
+          labels:
+            app: nginx
+  ```
+
+#### 2. Exclude logs with `app: nginx` labels from the namespace
+  ```
+  apiVersion: logging.banzaicloud.io/v1beta1
+  kind: Flow
+  metadata:
+    name: flow-sample
+    namespace: default
+  spec:
+    outputRefs:
+      - forward-output-sample
+    match:
+      - exclude:
+          labels:
+            app: nginx
+  ```
+
+#### 3. Exclude logs with `env: dev` labels but select `app: nginx` labels from the namespace
+  ```
+  apiVersion: logging.banzaicloud.io/v1beta1
+  kind: Flow
+  metadata:
+    name: flow-sample
+    namespace: default
+  spec:
+    outputRefs:
+      - forward-output-sample
+    match:
+      - exclude:
+          labels:
+            env: dev
+      - select:
+          labels:
+            app: nginx
+  ```
+
+#### 4. Exclude cluster logs from  `dev`, `sandbox` namespaces and select `app: nginx` from all namespaces
+  ```
+  apiVersion: logging.banzaicloud.io/v1beta1
+  kind: ClusterFlow
+  metadata:
+    name: clusterflow-sample
+  spec:
+    outputRefs:
+      - forward-output-sample
+    match:
+      - exclude:
+          namespaces: dev,sandbox
+      - select:
+          labels:
+            app: nginx
+  ```
+
+
+#### 5. Exclude cluster logs from  `dev`, `sandbox` namespaces and select `app: nginx` from all `prod` and `infra` namespaces
+  ```
+  apiVersion: logging.banzaicloud.io/v1beta1
+  kind: ClusterFlow
+  metadata:
+    name: clusterflow-sample
+  spec:
+    outputRefs:
+      - forward-output-sample
+    match:
+      - exclude:
+          namespaces: dev,sandbox
+      - select:
+          labels:
+            app: nginx
+          namespaces: prod,infra
+  ```
\ No newline at end of file
diff --git a/docs/quickstarts/cloudwatch-nginx.md b/docs/quickstarts/cloudwatch-nginx.md
index 09bfe19d..ecfe4d4d 100644
--- a/docs/quickstarts/cloudwatch-nginx.md
+++ b/docs/quickstarts/cloudwatch-nginx.md
@@ -139,8 +139,10 @@ Or set up the secret manually.
              reserve_data: true
              parse:
                type: nginx
-       selectors:
-         app.kubernetes.io/name: log-generator
+        match:
+          - select:
+              labels:
+                app.kubernetes.io/name: log-generator         
        outputRefs:
          - cloudwatch-output
      EOF
diff --git a/docs/quickstarts/es-nginx.md b/docs/quickstarts/es-nginx.md
index 6c53a9b6..1fd52bd6 100644
--- a/docs/quickstarts/es-nginx.md
+++ b/docs/quickstarts/es-nginx.md
@@ -150,8 +150,10 @@ To deploy the Logging operator using Kubernetes manifests, complete these steps.
              reserve_data: true
              parse:
                type: nginx
-       selectors:
-         app.kubernetes.io/name: log-generator
+        match:
+          - select:
+              labels:
+                app.kubernetes.io/name: log-generator
        outputRefs:
          - es-output
      EOF
@@ -164,9 +166,10 @@ To deploy the Logging operator using Kubernetes manifests, complete these steps.
     metadata:
       name: log-generator
     spec:
-      selector:
-        matchLabels:
-          app.kubernetes.io/name: log-generator
+      match:
+        - select:
+            labels:
+              app.kubernetes.io/name: log-generator
       replicas: 1
       template:
         metadata:
diff --git a/docs/quickstarts/example-s3.md b/docs/quickstarts/example-s3.md
index 9c61dd99..7f334526 100644
--- a/docs/quickstarts/example-s3.md
+++ b/docs/quickstarts/example-s3.md
@@ -122,8 +122,10 @@ Or set up the secret manually.
      spec:
        filters:
          - tag_normaliser: {}
-       selectors: {}
-         app.kubernetes.io/name: log-generator
+       match:
+         - select:
+             labels:
+               app.kubernetes.io/name: log-generator
        outputRefs:
          - s3-output
      EOF
diff --git a/docs/quickstarts/kafka-nginx.md b/docs/quickstarts/kafka-nginx.md
index 3f0a932a..750cb88e 100644
--- a/docs/quickstarts/kafka-nginx.md
+++ b/docs/quickstarts/kafka-nginx.md
@@ -102,8 +102,10 @@ To deploy the Logging operator using Kubernetes manifests, complete these steps.
              reserve_data: true
              parse:
                type: nginx
-       selectors:
-         app.kubernetes.io/name: log-generator
+       match:
+         - select:
+             labels:
+               app.kubernetes.io/name: log-generator
        outputRefs:
          - kafka-output
      EOF
diff --git a/docs/quickstarts/loki-nginx.md b/docs/quickstarts/loki-nginx.md
index f483870c..fdd82246 100644
--- a/docs/quickstarts/loki-nginx.md
+++ b/docs/quickstarts/loki-nginx.md
@@ -121,8 +121,10 @@ To deploy the Logging operator using Kubernetes manifests, complete these steps.
              reserve_data: true
              parse:
                type: nginx
-       selectors:
-         app.kubernetes.io/name: log-generator
+       match:
+         - select:
+           labels:
+             app.kubernetes.io/name: log-generator
        outputRefs:
          - loki-output
      EOF
diff --git a/main.go b/main.go
index 43ac3f80..bfbd45b2 100644
--- a/main.go
+++ b/main.go
@@ -22,7 +22,7 @@ import (
 	"emperror.dev/errors"
 	"github.com/banzaicloud/logging-operator/controllers"
 	"github.com/banzaicloud/logging-operator/pkg/k8sutil"
-	loggingv1alpha2 "github.com/banzaicloud/logging-operator/pkg/sdk/api/v1beta1"
+	loggingv1beta1 "github.com/banzaicloud/logging-operator/pkg/sdk/api/v1beta1"
 	"github.com/banzaicloud/logging-operator/pkg/sdk/model/types"
 	prometheusOperator "github.com/coreos/prometheus-operator/pkg/apis/monitoring/v1"
 	apiextensions "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
@@ -43,11 +43,11 @@ var (
 )
 
 func init() {
-	clientgoscheme.AddToScheme(scheme)
-	_ = loggingv1alpha2.AddToScheme(scheme)
+	_ = clientgoscheme.AddToScheme(scheme)
+	_ = loggingv1beta1.AddToScheme(scheme)
 	// +kubebuilder:scaffold:scheme
-	prometheusOperator.AddToScheme(scheme)
-	apiextensions.AddToScheme(scheme)
+	_ = prometheusOperator.AddToScheme(scheme)
+	_ = apiextensions.AddToScheme(scheme)
 }
 
 func main() {
diff --git a/pkg/resources/model/system.go b/pkg/resources/model/system.go
index 35273f67..98d94f17 100644
--- a/pkg/resources/model/system.go
+++ b/pkg/resources/model/system.go
@@ -15,6 +15,8 @@
 package model
 
 import (
+	"fmt"
+
 	"emperror.dev/errors"
 	"github.com/banzaicloud/logging-operator/pkg/resources/fluentd"
 	"github.com/banzaicloud/logging-operator/pkg/sdk/api/v1beta1"
@@ -71,8 +73,8 @@ func (l *LoggingResources) CreateModel() (*types.Builder, error) {
 		return nil, errors.WrapIf(err, "failed to create root input")
 	}
 	system := types.NewSystem(rootInput, types.NewRouter("main"))
-	for _, flow := range l.Flows {
-		flow, err := l.CreateFlowFromCustomResource(flow, flow.Namespace)
+	for _, flowCr := range l.Flows {
+		flow, err := l.CreateFlowFromCustomResource(flowCr)
 		if err != nil {
 			// TODO set flow status to error?
 			return nil, err
@@ -83,12 +85,7 @@ func (l *LoggingResources) CreateModel() (*types.Builder, error) {
 		}
 	}
 	for _, flowCr := range l.ClusterFlows {
-		flow, err := l.CreateFlowFromCustomResource(v1beta1.Flow{
-			TypeMeta:   flowCr.TypeMeta,
-			ObjectMeta: flowCr.ObjectMeta,
-			Spec:       flowCr.Spec,
-			Status:     flowCr.Status,
-		}, "")
+		flow, err := l.CreateFlowFromCustomResource(flowCr)
 		if err != nil {
 			// TODO set flow status to error?
 			return nil, err
@@ -104,21 +101,132 @@ func (l *LoggingResources) CreateModel() (*types.Builder, error) {
 	return system, nil
 }
 
-func (l *LoggingResources) CreateFlowFromCustomResource(flowCr v1beta1.Flow, namespace string) (*types.Flow, error) {
-	flow, err := types.NewFlow(namespace, flowCr.Spec.Selectors)
+type CommonFlow struct {
+	Name       string
+	Namespace  string
+	Scope      string
+	OutputRefs []string
+	Filters    []v1beta1.Filter
+	Flow       *types.Flow
+}
+
+// Create []FlowMatch from v1beta1.Match and v1beta1.ClusterMatch
+func GetFlowMatchFromSpec(namespace string, matches interface{}) []types.FlowMatch {
+	var flowMatches []types.FlowMatch
+	switch matchList := matches.(type) {
+	case []v1beta1.Match:
+		for _, match := range matchList {
+			if match.Select != nil {
+				flowMatches = append(flowMatches, types.FlowMatch{
+					Labels:     match.Select.Labels,
+					Namespaces: []string{namespace},
+					Negate:     false,
+				})
+			}
+			if match.Exclude != nil {
+				flowMatches = append(flowMatches, types.FlowMatch{
+					Labels:     match.Exclude.Labels,
+					Namespaces: []string{namespace},
+					Negate:     true,
+				})
+			}
+		}
+	case []v1beta1.ClusterMatch:
+		for _, match := range matchList {
+			if match.ClusterSelect != nil {
+				flowMatches = append(flowMatches, types.FlowMatch{
+					Labels:     match.ClusterSelect.Labels,
+					Namespaces: match.ClusterSelect.Namespaces,
+					Negate:     false,
+				})
+			}
+			if match.ClusterExclude != nil {
+				flowMatches = append(flowMatches, types.FlowMatch{
+					Labels:     match.ClusterExclude.Labels,
+					Namespaces: match.ClusterExclude.Namespaces,
+					Negate:     true,
+				})
+			}
+		}
+	}
+	return flowMatches
+}
+
+func FlowDispatcher(flowCr interface{}) (*CommonFlow, error) {
+	var commonFlow *CommonFlow
+	switch f := flowCr.(type) {
+	case v1beta1.ClusterFlow:
+		var matches []types.FlowMatch
+		commonFlow = &CommonFlow{
+			Name:       f.Name,
+			Namespace:  f.Namespace,
+			Scope:      "",
+			OutputRefs: f.Spec.OutputRefs,
+			Filters:    f.Spec.Filters,
+		}
+		if f.Spec.Match != nil {
+			matches = GetFlowMatchFromSpec(f.Namespace, f.Spec.Match)
+		} else {
+			matches = []types.FlowMatch{
+				{
+					Labels:     f.Spec.Selectors,
+					Namespaces: []string{""},
+					Negate:     false,
+				},
+			}
+		}
+		flow, err := types.NewFlow(matches)
+		if err != nil {
+			return nil, err
+		}
+		commonFlow.Flow = flow
+		return commonFlow, nil
+	case v1beta1.Flow:
+		var matches []types.FlowMatch
+		commonFlow = &CommonFlow{
+			Name:       f.Name,
+			Namespace:  f.Namespace,
+			Scope:      f.Namespace,
+			OutputRefs: f.Spec.OutputRefs,
+			Filters:    f.Spec.Filters,
+		}
+		if f.Spec.Match != nil {
+			matches = GetFlowMatchFromSpec(f.Namespace, f.Spec.Match)
+		} else {
+			matches = []types.FlowMatch{
+				{
+					Labels:     f.Spec.Selectors,
+					Namespaces: []string{f.Namespace},
+					Negate:     false,
+				},
+			}
+		}
+		flow, err := types.NewFlow(matches)
+		commonFlow.Flow = flow
+		if err != nil {
+			return nil, err
+		}
+		return commonFlow, nil
+	}
+	return nil, fmt.Errorf("unsupported type: %t", flowCr)
+}
+
+func (l *LoggingResources) CreateFlowFromCustomResource(flowCr interface{}) (*types.Flow, error) {
+	commonFlow, err := FlowDispatcher(flowCr)
 	if err != nil {
 		return nil, err
 	}
+	flow := commonFlow.Flow
 	outputs := []types.Output{}
 	var multierr error
 FindOutputForAllRefs:
-	for _, outputRef := range flowCr.Spec.OutputRefs {
+	for _, outputRef := range commonFlow.OutputRefs {
 		// only namespaced flows should use namespaced outputs
-		if namespace != "" {
+		if commonFlow.Scope != "" {
 			for _, output := range l.Outputs {
 				// only an output from the same namespace can be used with a matching name
-				if output.Namespace == namespace && outputRef == output.Name {
-					outputId := namespace + "_" + flowCr.Name + "_" + output.Name
+				if output.Namespace == commonFlow.Scope && outputRef == output.Name {
+					outputId := commonFlow.Scope + "_" + commonFlow.Name + "_" + output.Name
 					plugin, err := plugins.CreateOutput(output.Spec, outputId, secret.NewSecretLoader(l.client, output.Namespace, fluentd.OutputSecretPath, l.Secrets))
 					if err != nil {
 						multierr = errors.Combine(multierr, errors.WrapIff(err, "failed to create configured output %s", outputRef))
@@ -131,7 +239,7 @@ FindOutputForAllRefs:
 		}
 		for _, clusterOutput := range l.ClusterOutputs {
 			if outputRef == clusterOutput.Name {
-				outputId := namespace + "_" + flowCr.Name + "_" + clusterOutput.Name
+				outputId := commonFlow.Namespace + "_" + commonFlow.Name + "_" + clusterOutput.Name
 				plugin, err := plugins.CreateOutput(clusterOutput.Spec.OutputSpec, outputId, secret.NewSecretLoader(l.client, clusterOutput.Namespace, fluentd.OutputSecretPath, l.Secrets))
 				if err != nil {
 					multierr = errors.Combine(multierr, errors.WrapIff(err, "failed to create configured output %s", outputRef))
@@ -147,10 +255,10 @@ FindOutputForAllRefs:
 
 	// Filter
 	var filters []types.Filter
-	for i, f := range flowCr.Spec.Filters {
-		filter, err := plugins.CreateFilter(f, flowCr.Name, i, secret.NewSecretLoader(l.client, flowCr.Namespace, fluentd.OutputSecretPath, l.Secrets))
+	for i, f := range commonFlow.Filters {
+		filter, err := plugins.CreateFilter(f, commonFlow.Name, i, secret.NewSecretLoader(l.client, commonFlow.Namespace, fluentd.OutputSecretPath, l.Secrets))
 		if err != nil {
-			multierr = errors.Combine(multierr, errors.WrapIff(err, "failed to create filter with index %d for flow %s", i, flowCr.Name))
+			multierr = errors.Combine(multierr, errors.WrapIff(err, "failed to create filter with index %d for flow %s", i, commonFlow.Name))
 			continue
 		}
 		filters = append(filters, filter)
diff --git a/pkg/sdk/api/v1beta1/clusterflow_types.go b/pkg/sdk/api/v1beta1/clusterflow_types.go
index c5089dda..de647db0 100644
--- a/pkg/sdk/api/v1beta1/clusterflow_types.go
+++ b/pkg/sdk/api/v1beta1/clusterflow_types.go
@@ -30,6 +30,29 @@ type ClusterFlow struct {
 	Status FlowStatus `json:"status,omitempty"`
 }
 
+type ClusterMatch struct {
+	*ClusterSelect  `json:"select,omitempty"`
+	*ClusterExclude `json:"exclude,omitempty"`
+}
+
+type ClusterSelect struct {
+	Namespaces []string          `json:"namespaces,omitempty"`
+	Labels     map[string]string `json:"labels,omitempty"`
+}
+
+type ClusterExclude struct {
+	Namespaces []string          `json:"namespaces,omitempty"`
+	Labels     map[string]string `json:"labels,omitempty"`
+}
+
+// FlowSpec is the Kubernetes spec for Flows
+type ClusterFlowSpec struct {
+	Match      []ClusterMatch `json:"match,omitempty"`
+	Filters    []Filter       `json:"filters,omitempty"`
+	LoggingRef string         `json:"loggingRef,omitempty"`
+	OutputRefs []string       `json:"outputRefs"`
+}
+
 // +kubebuilder:object:root=true
 
 // ClusterFlowList contains a list of ClusterFlow
diff --git a/pkg/sdk/api/v1beta1/flow_types.go b/pkg/sdk/api/v1beta1/flow_types.go
index 5433eb95..1adfae4f 100644
--- a/pkg/sdk/api/v1beta1/flow_types.go
+++ b/pkg/sdk/api/v1beta1/flow_types.go
@@ -21,12 +21,27 @@ import (
 
 // FlowSpec is the Kubernetes spec for Flows
 type FlowSpec struct {
-	Selectors  map[string]string `json:"selectors"`
+	// Deprecated
+	Selectors  map[string]string `json:"selectors,omitempty"`
+	Match      []Match           `json:"match,omitempty"`
 	Filters    []Filter          `json:"filters,omitempty"`
 	LoggingRef string            `json:"loggingRef,omitempty"`
 	OutputRefs []string          `json:"outputRefs"`
 }
 
+type Match struct {
+	*Select  `json:"select,omitempty"`
+	*Exclude `json:"exclude,omitempty"`
+}
+
+type Select struct {
+	Labels map[string]string `json:"labels,omitempty"`
+}
+
+type Exclude struct {
+	Labels map[string]string `json:"labels,omitempty"`
+}
+
 // Filter definition for FlowSpec
 type Filter struct {
 	StdOut            *filter.StdOutFilterConfig `json:"stdout,omitempty"`
diff --git a/pkg/sdk/api/v1beta1/zz_generated.deepcopy.go b/pkg/sdk/api/v1beta1/zz_generated.deepcopy.go
index c08ecdc8..1e5a3674 100644
--- a/pkg/sdk/api/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/sdk/api/v1beta1/zz_generated.deepcopy.go
@@ -40,6 +40,33 @@ func (in *BufferStorage) DeepCopy() *BufferStorage {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterExclude) DeepCopyInto(out *ClusterExclude) {
+	*out = *in
+	if in.Namespaces != nil {
+		in, out := &in.Namespaces, &out.Namespaces
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Labels != nil {
+		in, out := &in.Labels, &out.Labels
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterExclude.
+func (in *ClusterExclude) DeepCopy() *ClusterExclude {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterExclude)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ClusterFlow) DeepCopyInto(out *ClusterFlow) {
 	*out = *in
@@ -99,6 +126,65 @@ func (in *ClusterFlowList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterFlowSpec) DeepCopyInto(out *ClusterFlowSpec) {
+	*out = *in
+	if in.Match != nil {
+		in, out := &in.Match, &out.Match
+		*out = make([]ClusterMatch, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	if in.Filters != nil {
+		in, out := &in.Filters, &out.Filters
+		*out = make([]Filter, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	if in.OutputRefs != nil {
+		in, out := &in.OutputRefs, &out.OutputRefs
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterFlowSpec.
+func (in *ClusterFlowSpec) DeepCopy() *ClusterFlowSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterFlowSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterMatch) DeepCopyInto(out *ClusterMatch) {
+	*out = *in
+	if in.ClusterSelect != nil {
+		in, out := &in.ClusterSelect, &out.ClusterSelect
+		*out = new(ClusterSelect)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.ClusterExclude != nil {
+		in, out := &in.ClusterExclude, &out.ClusterExclude
+		*out = new(ClusterExclude)
+		(*in).DeepCopyInto(*out)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterMatch.
+func (in *ClusterMatch) DeepCopy() *ClusterMatch {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterMatch)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ClusterOutput) DeepCopyInto(out *ClusterOutput) {
 	*out = *in
@@ -179,6 +265,55 @@ func (in *ClusterOutputSpec) DeepCopy() *ClusterOutputSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ClusterSelect) DeepCopyInto(out *ClusterSelect) {
+	*out = *in
+	if in.Namespaces != nil {
+		in, out := &in.Namespaces, &out.Namespaces
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Labels != nil {
+		in, out := &in.Labels, &out.Labels
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterSelect.
+func (in *ClusterSelect) DeepCopy() *ClusterSelect {
+	if in == nil {
+		return nil
+	}
+	out := new(ClusterSelect)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Exclude) DeepCopyInto(out *Exclude) {
+	*out = *in
+	if in.Labels != nil {
+		in, out := &in.Labels, &out.Labels
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Exclude.
+func (in *Exclude) DeepCopy() *Exclude {
+	if in == nil {
+		return nil
+	}
+	out := new(Exclude)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Filter) DeepCopyInto(out *Filter) {
 	*out = *in
@@ -333,6 +468,13 @@ func (in *FlowSpec) DeepCopyInto(out *FlowSpec) {
 			(*out)[key] = val
 		}
 	}
+	if in.Match != nil {
+		in, out := &in.Match, &out.Match
+		*out = make([]Match, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	if in.Filters != nil {
 		in, out := &in.Filters, &out.Filters
 		*out = make([]Filter, len(*in))
@@ -726,6 +868,31 @@ func (in *LoggingStatus) DeepCopy() *LoggingStatus {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Match) DeepCopyInto(out *Match) {
+	*out = *in
+	if in.Select != nil {
+		in, out := &in.Select, &out.Select
+		*out = new(Select)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.Exclude != nil {
+		in, out := &in.Exclude, &out.Exclude
+		*out = new(Exclude)
+		(*in).DeepCopyInto(*out)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Match.
+func (in *Match) DeepCopy() *Match {
+	if in == nil {
+		return nil
+	}
+	out := new(Match)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Metrics) DeepCopyInto(out *Metrics) {
 	*out = *in
@@ -935,6 +1102,28 @@ func (in *Security) DeepCopy() *Security {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Select) DeepCopyInto(out *Select) {
+	*out = *in
+	if in.Labels != nil {
+		in, out := &in.Labels, &out.Labels
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Select.
+func (in *Select) DeepCopy() *Select {
+	if in == nil {
+		return nil
+	}
+	out := new(Select)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *VolumeMount) DeepCopyInto(out *VolumeMount) {
 	*out = *in
diff --git a/pkg/sdk/go.sum b/pkg/sdk/go.sum
index 25fe4df7..d80f5891 100644
--- a/pkg/sdk/go.sum
+++ b/pkg/sdk/go.sum
@@ -33,6 +33,7 @@ github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5
 github.com/asaskevich/govalidator v0.0.0-20180720115003-f9ffefc3facf/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
 github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
 github.com/banzaicloud/k8s-objectmatcher v1.1.0/go.mod h1:gGaElvgkqa0Lk1khRr+jel/nsCLfzhLnD3CEWozpk9k=
+github.com/banzaicloud/logging-operator v0.0.0-20200212081810-c4fd64a8ce42 h1:sl4WPmvEYDoZameWb8c9PexCw2JGn0QC5y4ha2u74CE=
 github.com/banzaicloud/operator-tools v0.4.1 h1:+pAQ67l7eGTNR/D2FMKyLEgNv9l6U6H21qFNrp1VhnQ=
 github.com/banzaicloud/operator-tools v0.4.1/go.mod h1:12rKcmPZkK8CaESI6YI3UGFBBBQvYZH9i8Gn5cfCi+U=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
diff --git a/pkg/sdk/model/render/json.go b/pkg/sdk/model/render/json.go
deleted file mode 100644
index 86d1b105..00000000
--- a/pkg/sdk/model/render/json.go
+++ /dev/null
@@ -1,47 +0,0 @@
-// Copyright © 2019 Banzai Cloud
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//    http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package render
-
-import (
-	"encoding/json"
-	"fmt"
-	"io"
-	"strings"
-
-	"emperror.dev/errors"
-
-	"github.com/banzaicloud/logging-operator/pkg/sdk/model/types"
-)
-
-type JsonRender struct {
-	out    io.Writer
-	indent int
-}
-
-func (t *JsonRender) Render(config types.FluentConfig) error {
-	var out []byte
-	var err error
-	if t.indent > 0 {
-		out, err = json.MarshalIndent(config, "", strings.Repeat(" ", t.indent))
-	} else {
-		out, err = json.Marshal(config)
-	}
-
-	if err != nil {
-		return errors.WrapIf(err, "Failed to marshal model into yaml")
-	}
-	fmt.Fprintf(t.out, "%s", out)
-	return nil
-}
diff --git a/pkg/sdk/model/types/builder.go b/pkg/sdk/model/types/builder.go
index 20fcb413..08002fa9 100644
--- a/pkg/sdk/model/types/builder.go
+++ b/pkg/sdk/model/types/builder.go
@@ -16,7 +16,6 @@ package types
 
 import (
 	"errors"
-	"reflect"
 )
 
 type Builder struct {
@@ -34,7 +33,7 @@ func NewSystem(input Input, router *Router) *Builder {
 
 func (s *Builder) RegisterFlow(f *Flow) error {
 	for _, e := range s.flows {
-		if e.Namespace == f.Namespace && reflect.DeepEqual(e.Labels, f.Labels) {
+		if e.FlowLabel == f.FlowLabel {
 			return errors.New("Flow already exists")
 		}
 	}
diff --git a/pkg/sdk/model/types/flow.go b/pkg/sdk/model/types/flow.go
index 6fddca6c..722911d8 100644
--- a/pkg/sdk/model/types/flow.go
+++ b/pkg/sdk/model/types/flow.go
@@ -50,10 +50,8 @@ type Flow struct {
 	// List of Outputs that will emit the event, at least one output is required.
 	Outputs []Output `json:"outputs"`
 
-	// Optional set of kubernetes labels
-	Labels map[string]string `json:"-"`
-	// Optional namespace
-	Namespace string `json:"-"`
+	// Matches for select or exclude
+	Matches []FlowMatch `json:"matches,omitempty"`
 
 	// Fluentd label
 	FlowLabel string `json:"-"`
@@ -94,8 +92,8 @@ func (f *Flow) WithOutputs(output ...Output) *Flow {
 	return f
 }
 
-func NewFlow(namespace string, labels map[string]string) (*Flow, error) {
-	flowLabel, err := calculateFlowLabel(namespace, labels)
+func NewFlow(matches []FlowMatch) (*Flow, error) {
+	flowLabel, err := calculateFlowLabel(matches)
 	if err != nil {
 		return nil, err
 	}
@@ -105,31 +103,35 @@ func NewFlow(namespace string, labels map[string]string) (*Flow, error) {
 			Tag:       flowLabel,
 		},
 		FlowLabel: flowLabel,
-		Labels:    labels,
-		Namespace: namespace,
+		Matches:   matches,
 	}, nil
 }
 
-func calculateFlowLabel(namespace string, labels map[string]string) (string, error) {
+func calculateFlowLabel(matches []FlowMatch) (string, error) {
 	b := md5.New()
-	if _, err := io.WriteString(b, namespace); err != nil {
-		return "", err
-	}
-
-	// Make sure the generated label is consistent
-	keys := []string{}
-	for k := range labels {
-		keys = append(keys, k)
-	}
-	sort.Strings(keys)
-
-	for _, k := range keys {
-		if _, err := io.WriteString(b, k); err != nil {
-			return "", err
+	for _, match := range matches {
+		sort.Strings(match.Namespaces)
+		for _, n := range match.Namespaces {
+			if _, err := io.WriteString(b, n); err != nil {
+				return "", err
+			}
+		}
+		// Make sure the generated label is consistent
+		keys := []string{}
+		for k := range match.Labels {
+			keys = append(keys, k)
 		}
-		if _, err := io.WriteString(b, labels[k]); err != nil {
-			return "", err
+		sort.Strings(keys)
+
+		for _, k := range keys {
+			if _, err := io.WriteString(b, k); err != nil {
+				return "", err
+			}
+			if _, err := io.WriteString(b, match.Labels[k]); err != nil {
+				return "", err
+			}
 		}
 	}
+
 	return fmt.Sprintf("@%x", b.Sum(nil)), nil
 }
diff --git a/pkg/sdk/model/types/router.go b/pkg/sdk/model/types/router.go
index 0145cda1..8626cf9d 100644
--- a/pkg/sdk/model/types/router.go
+++ b/pkg/sdk/model/types/router.go
@@ -15,6 +15,7 @@
 package types
 
 import (
+	"strconv"
 	"strings"
 
 	util "github.com/banzaicloud/operator-tools/pkg/utils"
@@ -38,22 +39,24 @@ func (r *Router) GetSections() []Directive {
 	return r.Routes
 }
 
-type FlowRoute struct {
-	PluginMeta
+type FlowMatch struct {
 	// Optional set of kubernetes labels
 	Labels map[string]string `json:"labels,omitempty"`
 	// Optional namespace
-	Namespace string `json:"namespace,omitempty"`
+	Namespaces []string `json:"namespaces,omitempty"`
+	// Negate
+	Negate bool `json:"negate,omitempty"`
 }
 
-func (f *FlowRoute) GetPluginMeta() *PluginMeta {
-	return &f.PluginMeta
+func (f FlowMatch) GetPluginMeta() *PluginMeta {
+	return &PluginMeta{
+		Directive: "match",
+	}
 }
-
-func (f *FlowRoute) GetParams() map[string]string {
-	params := map[string]string{}
-	if f.Namespace != "" {
-		params["namespace"] = f.Namespace
+func (f FlowMatch) GetParams() map[string]string {
+	params := map[string]string{
+		"namespaces": strings.Join(f.Namespaces, ","),
+		"negate":     strconv.FormatBool(f.Negate),
 	}
 	if len(f.Labels) > 0 {
 		var sb []string
@@ -65,19 +68,38 @@ func (f *FlowRoute) GetParams() map[string]string {
 	return params
 }
 
-func (f *FlowRoute) GetSections() []Directive {
+func (f FlowMatch) GetSections() []Directive {
 	return nil
 }
 
+type FlowRoute struct {
+	PluginMeta
+	Matches []Directive
+}
+
+func (f *FlowRoute) GetPluginMeta() *PluginMeta {
+	return &f.PluginMeta
+}
+
+func (f *FlowRoute) GetParams() map[string]string {
+	return nil
+}
+
+func (f *FlowRoute) GetSections() []Directive {
+	return f.Matches
+}
+
 func (r *Router) AddRoute(flow *Flow) *Router {
-	r.Routes = append(r.Routes, &FlowRoute{
+	route := &FlowRoute{
 		PluginMeta: PluginMeta{
 			Directive: "route",
 			Label:     flow.FlowLabel,
 		},
-		Labels:    flow.Labels,
-		Namespace: flow.Namespace,
-	})
+	}
+	for _, f := range flow.Matches {
+		route.Matches = append(route.Matches, f)
+	}
+	r.Routes = append(r.Routes, route)
 	return r
 }
 


Test patch:
diff --git a/pkg/sdk/api/v1beta1/clusterflow_types_test.go b/pkg/sdk/api/v1beta1/clusterflow_types_test.go
index 630cbb71..189f5e34 100644
--- a/pkg/sdk/api/v1beta1/clusterflow_types_test.go
+++ b/pkg/sdk/api/v1beta1/clusterflow_types_test.go
@@ -58,7 +58,7 @@ var _ = Describe("ClusterFlow", func() {
 					Namespace: "foo",
 				},
 				Spec: FlowSpec{
-					Selectors:  map[string]string{},
+					Selectors:  nil,
 					OutputRefs: []string{},
 				},
 				Status: FlowStatus{},
diff --git a/pkg/sdk/api/v1beta1/flow_types_test.go b/pkg/sdk/api/v1beta1/flow_types_test.go
index 817ca767..8b6c514d 100644
--- a/pkg/sdk/api/v1beta1/flow_types_test.go
+++ b/pkg/sdk/api/v1beta1/flow_types_test.go
@@ -58,7 +58,7 @@ var _ = Describe("Flow", func() {
 					Namespace: "default",
 				},
 				Spec: FlowSpec{
-					Selectors:  map[string]string{},
+					Selectors:  nil,
 					OutputRefs: []string{},
 				},
 				Status: FlowStatus{},
diff --git a/pkg/sdk/model/render/fluent_test.go b/pkg/sdk/model/render/fluent_test.go
index ce3fff8b..f626ed13 100644
--- a/pkg/sdk/model/render/fluent_test.go
+++ b/pkg/sdk/model/render/fluent_test.go
@@ -209,6 +209,10 @@ func TestRenderDirective(t *testing.T) {
               @id test_label_router
               <route>
                 @label @d41d8cd98f00b204e9800998ecf8427e
+				<match>
+				  namespaces
+				  negate false
+				</match>
               </route>
             </match>`,
 			),
@@ -225,7 +229,10 @@ func TestRenderDirective(t *testing.T) {
               @id test_label_router
               <route>
                 @label @098f6bcd4621d373cade4e832627b4f6
-                namespace test
+				  <match>
+				    namespaces test
+				    negate false
+				  </match>
               </route>
             </match>`,
 			),
@@ -242,8 +249,11 @@ func TestRenderDirective(t *testing.T) {
               @id test_label_router
               <route>
                 @label @092f5fa58e4f619d739f5b65f2ed38bc
-                labels a:b,c:d
-                namespace test
+        		  <match>
+        		    labels a:b,c:d
+        		    namespaces test
+        		    negate false
+        		  </match>
               </route>
             </match>`,
 			),
@@ -272,10 +282,11 @@ func TestMultipleOutput(t *testing.T) {
 	system := types.NewSystem(toDirective(t, input.NewTailInputConfig("input.log")), types.NewRouter("test"))
 
 	flowObj, err := types.NewFlow(
-		"ns-test",
-		map[string]string{
-			"key1": "val1",
-			"key2": "val2",
+		[]types.FlowMatch{
+			{Labels: map[string]string{
+				"key1": "val1",
+				"key2": "val2"},
+				Namespaces: []string{"ns-test"}},
 		})
 	if err != nil {
 		t.Fatal(err)
@@ -316,8 +327,11 @@ func TestMultipleOutput(t *testing.T) {
           @id test_label_router
           <route>
             @label @901f778f9602a78e8fd702c1973d8d8d
-            labels key1:val1,key2:val2
-            namespace ns-test
+        	<match>
+        	  labels key1:val1,key2:val2
+        	  namespaces ns-test
+        	  negate false
+        	</match>
           </route>
         </match>
         <label @901f778f9602a78e8fd702c1973d8d8d>
@@ -347,10 +361,11 @@ func TestRenderFullFluentConfig(t *testing.T) {
 	system := types.NewSystem(toDirective(t, input.NewTailInputConfig("input.log")), types.NewRouter("test"))
 
 	flowObj, err := types.NewFlow(
-		"ns-test",
-		map[string]string{
-			"key1": "val1",
-			"key2": "val2",
+		[]types.FlowMatch{
+			{Labels: map[string]string{
+				"key1": "val1",
+				"key2": "val2"},
+				Namespaces: []string{"ns-test"}},
 		})
 	if err != nil {
 		t.Fatal(err)
@@ -390,8 +405,11 @@ func TestRenderFullFluentConfig(t *testing.T) {
           @id test_label_router
           <route>
             @label @901f778f9602a78e8fd702c1973d8d8d
-            labels key1:val1,key2:val2
-            namespace ns-test
+			  <match>
+			    labels key1:val1,key2:val2
+			    namespaces ns-test
+			    negate false
+			  </match>
           </route>
         </match>
         <label @901f778f9602a78e8fd702c1973d8d8d>
@@ -517,10 +535,11 @@ func ValidateRenderS3(t *testing.T, s3Config plugins.DirectiveConverter, expecte
 		return err
 	}
 	flowObj, err := types.NewFlow(
-		"ns-test",
-		map[string]string{
-			"key1": "val1",
-			"key2": "val2",
+		[]types.FlowMatch{
+			{Labels: map[string]string{
+				"key1": "val1",
+				"key2": "val2"},
+				Namespaces: []string{"ns-test"}},
 		})
 	if err != nil {
 		return err
@@ -558,8 +577,11 @@ func ValidateRenderS3(t *testing.T, s3Config plugins.DirectiveConverter, expecte
           @id test_label_router
           <route>
             @label @901f778f9602a78e8fd702c1973d8d8d
-            labels key1:val1,key2:val2
-            namespace ns-test
+			<match>
+			  labels key1:val1,key2:val2
+			  namespaces ns-test
+			  negate false
+			</match>
           </route>
         </match>
         <label @901f778f9602a78e8fd702c1973d8d8d>
@@ -574,7 +596,7 @@ func ValidateRenderS3(t *testing.T, s3Config plugins.DirectiveConverter, expecte
 }
 
 func newFlowOrPanic(namespace string, labels map[string]string) *types.Flow {
-	flowObj, err := types.NewFlow(namespace, labels)
+	flowObj, err := types.NewFlow([]types.FlowMatch{{Labels: labels, Namespaces: []string{namespace}}})
 	if err != nil {
 		panic(err)
 	}
diff --git a/pkg/sdk/model/render/json_test.go b/pkg/sdk/model/render/json_test.go
deleted file mode 100644
index 7e3f0a05..00000000
--- a/pkg/sdk/model/render/json_test.go
+++ /dev/null
@@ -1,129 +0,0 @@
-// Copyright © 2019 Banzai Cloud
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//    http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package render
-
-import (
-	"bytes"
-	"testing"
-
-	"github.com/andreyvit/diff"
-	"github.com/banzaicloud/logging-operator/pkg/sdk/model/filter"
-	"github.com/banzaicloud/logging-operator/pkg/sdk/model/input"
-	"github.com/banzaicloud/logging-operator/pkg/sdk/model/output"
-	"github.com/banzaicloud/logging-operator/pkg/sdk/model/types"
-	"github.com/banzaicloud/operator-tools/pkg/secret"
-)
-
-func TestJsonRender(t *testing.T) {
-	commonId := "test"
-	input, err := input.NewTailInputConfig("input.log").ToDirective(secret.NewSecretLoader(nil, "", "", nil), commonId)
-	if err != nil {
-		t.Fatalf("%+v", err)
-	}
-	system := types.NewSystem(input, types.NewRouter("test"))
-
-	flow, err := types.NewFlow(
-		"ns-test",
-		map[string]string{
-			"key1": "val1",
-			"key2": "val2",
-		})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	filter, err := filter.NewStdOutFilterConfig().ToDirective(secret.NewSecretLoader(nil, "", "", nil), commonId)
-	if err != nil {
-		t.Fatalf("%+v", err)
-	}
-
-	nullOut, err := output.NewNullOutputConfig().ToDirective(secret.NewSecretLoader(nil, "", "", nil), commonId)
-	if err != nil {
-		t.Fatalf("%+v", err)
-	}
-
-	flow.WithFilters(filter).
-		WithOutputs(nullOut)
-
-	err = system.RegisterFlow(flow)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	configuredSystem, err := system.Build()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	b := &bytes.Buffer{}
-	jsonRender := JsonRender{out: b, indent: 2}
-	err = jsonRender.Render(configuredSystem)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	expected := `{
-          "input": {
-            "type": "tail",
-            "id": "test_tail",
-            "directive": "source",
-			"params": {
-            	"path": "input.log"
-			}
-          },
-          "router": {
-			"type": "label_router",
-            "id": "test_label_router",
-            "directive": "match",
-			"tag": "**",
-            "routes": [
-              {
-                "directive": "route",
-                "label": "@901f778f9602a78e8fd702c1973d8d8d",
-                "labels": {
-                  "key1": "val1",
-                  "key2": "val2"
-                },
-                "namespace": "ns-test"
-              }
-            ]
-          },
-          "flows": [
-            {
-              "directive": "label",
-              "tag": "@901f778f9602a78e8fd702c1973d8d8d",
-              "filters": [
-                {
-                  "type": "stdout",
-                  "id": "test_stdout",
-                  "directive": "filter",
-                  "tag": "**"
-                }
-              ],
-              "outputs": [
-                {
-                  "type": "null",
-                  "id": "test_null",
-                  "directive": "match",
-                  "tag": "**"
-                }
-              ]
-            }
-          ]
-        }`
-	if a, e := diff.TrimLinesInString(b.String()), diff.TrimLinesInString(expected); a != e {
-		t.Errorf("Result not as expected:\n%v \nActual: %s", diff.LineDiff(a, e), b.String())
-	}
-}
diff --git a/pkg/sdk/model/render/test_helper.go b/pkg/sdk/model/render/test_helper.go
index 5ce2814d..4ff432a0 100644
--- a/pkg/sdk/model/render/test_helper.go
+++ b/pkg/sdk/model/render/test_helper.go
@@ -74,7 +74,10 @@ func NewOutputPluginTest(t *testing.T, plugin plugins.DirectiveConverter) *Plugi
   @id test_label_router
   <route>
     @label @a42fd8d29c181fcf9887280c4a51bd1e
-    namespace ns-test
+    <match>
+      namespaces ns-test
+      negate false
+    </match>
   </route>
 </match>
 <label @a42fd8d29c181fcf9887280c4a51bd1e>`,
@@ -82,8 +85,9 @@ func NewOutputPluginTest(t *testing.T, plugin plugins.DirectiveConverter) *Plugi
 	suite.System = types.NewSystem(toDirective(t, input.NewTailInputConfig("input.log")), types.NewRouter("test"))
 
 	flowObj, err := types.NewFlow(
-		"ns-test",
-		map[string]string{})
+		[]types.FlowMatch{
+			{Namespaces: []string{"ns-test"}},
+		})
 	if err != nil {
 		t.Fatal(err)
 	}