add cluster ID to flow logs (#386)

api/v1beta1/flowcollector_types.go

@@ -448,6 +448,11 @@ type FlowCollectorFLP struct {
 	// `conversationTerminatingTimeout` is the time to wait from detected FIN flag to end a conversation. Only relevant for TCP flows.
 	ConversationTerminatingTimeout *metav1.Duration `json:"conversationTerminatingTimeout,omitempty"`
 
+	//+kubebuilder:default:=""
+	// +optional
+	// `clusterName` is the name of the cluster to appear in the flows data. This is useful in a multi-cluster context. When using OpenShift, leave empty to make it automatically determined.
+	ClusterName string `json:"clusterName,omitempty"`
+
 	// `debug` allows setting some aspects of the internal configuration of the flow processor.
 	// This section is aimed exclusively for debugging and fine-grained performance optimizations,
 	// such as GOGC and GOMAXPROCS env vars. Users setting its values do it at their own risk.

bundle/manifests/flows.netobserv.io_flowcollectors.yaml

@@ -3793,6 +3793,12 @@ spec:
                   and forwards them to the Loki persistence layer and/or any available
                   exporter.'
                 properties:
+                  clusterName:
+                    default: ""
+                    description: '`clusterName` is the name of the cluster to appear
+                      in the flows data. This is useful in a multi-cluster context.
+                      When using OpenShift, leave empty to make it automatically determined.'
+                    type: string
                   conversationEndTimeout:
                     default: 10s
                     description: '`conversationEndTimeout` is the time to wait after

bundle/manifests/netobserv-operator.clusterserviceversion.yaml

@@ -516,6 +516,14 @@ spec:
           - patch
           - update
           - watch
+        - apiGroups:
+          - config.openshift.io
+          resources:
+          - clusterversions
+          verbs:
+          - get
+          - list
+          - watch
         - apiGroups:
           - console.openshift.io
           resources:
@@ -614,24 +622,15 @@ spec:
           - rbac.authorization.k8s.io
           resources:
           - clusterrolebindings
-          - rolebindings
-          verbs:
-          - create
-          - delete
-          - get
-          - list
-          - update
-          - watch
-        - apiGroups:
-          - rbac.authorization.k8s.io
-          resources:
           - clusterroles
+          - rolebindings
           - roles
           verbs:
           - create
           - delete
           - get
           - list
+          - update
           - watch
         - apiGroups:
           - security.openshift.io

config/crd/bases/flows.netobserv.io_flowcollectors.yaml

@@ -3780,6 +3780,12 @@ spec:
                   and forwards them to the Loki persistence layer and/or any available
                   exporter.'
                 properties:
+                  clusterName:
+                    default: ""
+                    description: '`clusterName` is the name of the cluster to appear
+                      in the flows data. This is useful in a multi-cluster context.
+                      When using OpenShift, leave empty to make it automatically determined.'
+                    type: string
                   conversationEndTimeout:
                     default: 10s
                     description: '`conversationEndTimeout` is the time to wait after

config/rbac/role.yaml

@@ -57,6 +57,14 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - config.openshift.io
+  resources:
+  - clusterversions
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - console.openshift.io
   resources:
@@ -155,24 +163,15 @@ rules:
   - rbac.authorization.k8s.io
   resources:
   - clusterrolebindings
-  - rolebindings
-  verbs:
-  - create
-  - delete
-  - get
-  - list
-  - update
-  - watch
-- apiGroups:
-  - rbac.authorization.k8s.io
-  resources:
   - clusterroles
+  - rolebindings
   - roles
   verbs:
   - create
   - delete
   - get
   - list
+  - update
   - watch
 - apiGroups:
   - security.openshift.io

config/samples/flows_v1beta1_flowcollector.yaml

@@ -58,6 +58,8 @@ spec:
     conversationTerminatingTimeout: 5s
     conversationHeartbeatInterval: 30s
     conversationEndTimeout: 10s
+    # Append a unique cluster name to each record
+    # clusterName: <CLUSTER NAME>
   kafka:
     address: "kafka-cluster-kafka-bootstrap.netobserv"
     topic: network-flows

controllers/flowcollector_controller.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"net"
 
+	"github.com/netobserv/network-observability-operator/controllers/globals"
+	configv1 "github.com/openshift/api/config/v1"
 	osv1alpha1 "github.com/openshift/api/console/v1alpha1"
 	securityv1 "github.com/openshift/api/security/v1"
 	appsv1 "k8s.io/api/apps/v1"
@@ -64,8 +66,7 @@ func NewFlowCollectorReconciler(client client.Client, scheme *runtime.Scheme, co
 //+kubebuilder:rbac:groups=apps,resources=deployments;daemonsets,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=core,resources=namespaces;services;serviceaccounts;configmaps;secrets,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=core,resources=endpoints,verbs=get;list;watch
-//+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterroles;roles,verbs=get;create;delete;watch;list
-//+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterrolebindings;rolebindings,verbs=get;list;create;delete;update;watch
+//+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterrolebindings;clusterroles;rolebindings;roles,verbs=get;list;create;delete;update;watch
 //+kubebuilder:rbac:groups=console.openshift.io,resources=consoleplugins,verbs=get;create;delete;update;patch;list;watch
 //+kubebuilder:rbac:groups=operator.openshift.io,resources=consoles,verbs=get;update;list;update;watch
 //+kubebuilder:rbac:groups=flows.netobserv.io,resources=flowcollectors,verbs=get;list;watch;create;update;patch;delete
@@ -75,6 +76,7 @@ func NewFlowCollectorReconciler(client client.Client, scheme *runtime.Scheme, co
 //+kubebuilder:rbac:groups=security.openshift.io,resources=securitycontextconstraints,verbs=list;create;update;watch
 //+kubebuilder:rbac:groups=apiregistration.k8s.io,resources=apiservices,verbs=list;get;watch
 //+kubebuilder:rbac:groups=monitoring.coreos.com,resources=servicemonitors;prometheusrules,verbs=get;create;delete;update;patch;list;watch
+//+kubebuilder:rbac:groups=config.openshift.io,resources=clusterversions,verbs=get;list;watch
 //+kubebuilder:rbac:urls="/metrics",verbs=get
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
@@ -101,6 +103,18 @@ func (r *FlowCollectorReconciler) Reconcile(ctx context.Context, _ ctrl.Request)
 
 	var didChange, isInProgress bool
 	previousNamespace := desired.Status.Namespace
+
+	// obtain default cluster ID - api is specific to openshift
+	if r.permissions.Vendor(ctx) == discover.VendorOpenShift && globals.DefaultClusterID == "" {
+		cversion := &configv1.ClusterVersion{}
+		key := client.ObjectKey{Name: "version"}
+		if err := r.Client.Get(ctx, key, cversion); err != nil {
+			log.Error(err, "unable to obtain cluster ID")
+		} else {
+			globals.DefaultClusterID = cversion.Spec.ClusterID
+		}
+	}
+
 	reconcilersInfo := r.newCommonInfo(ctx, desired, ns, previousNamespace, func(b bool) { didChange = b }, func(b bool) { isInProgress = b })
 
 	err = r.reconcileOperator(ctx, &reconcilersInfo, desired)

controllers/flowcollector_controller_iso_test.go

@@ -53,6 +53,7 @@ func flowCollectorIsoSpecs() {
 				ConversationHeartbeatInterval:  &metav1.Duration{Duration: time.Second},
 				ConversationEndTimeout:         &metav1.Duration{Duration: time.Second},
 				ConversationTerminatingTimeout: &metav1.Duration{Duration: time.Second},
+				ClusterName:                    "testCluster",
 				Debug:                          flowslatest.DebugConfig{},
 				LogTypes:                       &outputRecordTypes,
 				Metrics: flowslatest.FLPMetrics{