Implement converting new API to FLP config

Provide some examples

api/v1alpha1/flowmetric_types.go

@@ -40,6 +40,7 @@ type MetricFilter struct {
 	Field string `json:"field"`
 
 	// Value to filter on
+	// +optional
 	Value string `json:"value"`
 
 	// Type of matching to apply
@@ -55,12 +56,14 @@ type MetricFilter struct {
 // To check the cardinality of all NetObserv metrics, run as `promql`: `count({__name__=~"netobserv.*"}) by (__name__)`.
 type FlowMetricSpec struct {
 	// Name of the metric in Prometheus. It will be automatically prefixed with "netobserv_".
+	// +required
 	MetricName string `json:"metricName"`
 
 	// Metric type: "Counter" or "Histogram".
 	// Use "Counter" for any value that increases over time and on which you can compute a rate, such as Bytes or Packets.
 	// Use "Histogram" for any value that must be sampled independently, such as latencies.
 	// +kubebuilder:validation:Enum:="Counter";"Histogram"
+	// +required
 	Type MetricType `json:"type"`
 
 	// `valueField` is the flow field that must be used as a value for this metric. This field must hold numeric values.
@@ -80,7 +83,7 @@ type FlowMetricSpec struct {
 	// It must be done carefully as it impacts the metric cardinality (cf https://rhobs-handbook.netlify.app/products/openshiftmonitoring/telemetry.md/#what-is-the-cardinality-of-a-metric).
 	// In general, avoid setting very high cardinality labels such as IP or MAC addresses.
 	// "SrcK8S_OwnerName" or "DstK8S_OwnerName" should be preferred over "SrcK8S_Name" or "DstK8S_Name" as much as possible.
-	// Refer to the documentation for the list of available fields: https://docs.openshift.com/container-platform/latest/networking/network_observability/json-flows-format-reference.html.
+	// Refer to the documentation for the list of available fields: https://docs.openshift.com/container-platform/latest/network_observability/json-flows-format-reference.html.
 	// +optional
 	Labels []string `json:"labels"`
 

bundle/manifests/netobserv-operator.clusterserviceversion.yaml

@@ -637,6 +637,14 @@ spec:
           - get
           - patch
           - update
+        - apiGroups:
+          - flows.netobserv.io
+          resources:
+          - flowmetrics
+          verbs:
+          - get
+          - list
+          - watch
         - apiGroups:
           - loki.grafana.com
           resourceNames:

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

@@ -75,7 +75,6 @@ spec:
                   required:
                   - field
                   - matchType
-                  - value
                   type: object
                 type: array
               labels:
@@ -87,7 +86,7 @@ spec:
                   In general, avoid setting very high cardinality labels such as IP
                   or MAC addresses. "SrcK8S_OwnerName" or "DstK8S_OwnerName" should
                   be preferred over "SrcK8S_Name" or "DstK8S_Name" as much as possible.
-                  Refer to the documentation for the list of available fields: https://docs.openshift.com/container-platform/latest/networking/network_observability/json-flows-format-reference.html.'
+                  Refer to the documentation for the list of available fields: https://docs.openshift.com/container-platform/latest/network_observability/json-flows-format-reference.html.'
                 items:
                   type: string
                 type: array

config/rbac/role.yaml

@@ -137,6 +137,14 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - flows.netobserv.io
+  resources:
+  - flowmetrics
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - loki.grafana.com
   resourceNames:

config/samples/flowmetrics/pods_accept_outside.yaml

@@ -0,0 +1,19 @@
+apiVersion: flows.netobserv.io/v1alpha1
+kind: FlowMetric
+metadata:
+  name: flowmetric-pods-accept-outside
+spec:
+  metricName: pods_accept_outside
+  type: Counter
+  valueField: Bytes
+  labels: [DstK8S_Name,DstK8S_Namespace,DstK8S_OwnerName,DstK8S_OwnerType]
+  filters:
+  - field: SrcK8S_OwnerType
+    matchType: Absence
+  - field: DstK8S_Type
+    value: Pod
+  - field: Duplicate
+    value: "false"
+  - field: FlowDirection
+    value: "0|2"
+    matchType: Regex

config/samples/flowmetrics/pods_openshift_ingress.yaml

@@ -0,0 +1,19 @@
+apiVersion: flows.netobserv.io/v1alpha1
+kind: FlowMetric
+metadata:
+  name: flowmetric-pods-openshift-ingress
+spec:
+  metricName: pods_openshift_ingress
+  type: Counter
+  valueField: Bytes
+  labels: [DstK8S_Name,DstK8S_Namespace,DstK8S_OwnerName,DstK8S_OwnerType]
+  filters:
+  - field: SrcK8S_Namespace
+    value: openshift-ingress
+  - field: DstK8S_Type
+    value: Pod
+  - field: Duplicate
+    value: "false"
+  - field: FlowDirection
+    value: "0|2"
+    matchType: Regex

config/samples/flowmetrics/pods_talking_outside.yaml

@@ -0,0 +1,19 @@
+apiVersion: flows.netobserv.io/v1alpha1
+kind: FlowMetric
+metadata:
+  name: flowmetric-pods-talking-outside
+spec:
+  metricName: pods_talking_outside
+  type: Counter
+  valueField: Bytes
+  labels: [SrcK8S_Name,SrcK8S_Namespace,SrcK8S_OwnerName,SrcK8S_OwnerType,DstAddr]
+  filters:
+  - field: DstK8S_OwnerType
+    matchType: Absence
+  - field: SrcK8S_Type
+    value: Pod
+  - field: Duplicate
+    value: "false"
+  - field: FlowDirection
+    value: "1|2"
+    matchType: Regex

config/samples/flows_v1alpha1_flowmetric.yaml

@@ -9,4 +9,17 @@ metadata:
     app.kubernetes.io/created-by: netobserv-operator
   name: flowmetric-sample
 spec:
-  # TODO(user): Add fields here
+  # Example, counting flows per port <= 9999
+  # More examples in config/samples/flowmetrics
+  metricName: service_ports_total
+  type: Counter
+  labels: [DstPort]
+  filters:
+  - field: DstPort
+    value: "^\\d\\d?\\d?\\d?$"
+    matchType: Regex
+  - field: Duplicate
+    value: "false"
+  - field: FlowDirection
+    value: "1|2"
+    matchType: Regex

controllers/flp/flp_common_objects.go

@@ -14,6 +14,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 
+	"github.com/netobserv/network-observability-operator/api/v1alpha1"
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/constants"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
@@ -49,20 +50,21 @@ var FlpConfSuffix = map[ConfKind]string{
 }
 
 type Builder struct {
-	info     *reconcilers.Instance
-	labels   map[string]string
-	selector map[string]string
-	desired  *flowslatest.FlowCollectorSpec
-	promTLS  *flowslatest.CertificateReference
-	confKind ConfKind
-	volumes  volumes.Builder
-	loki     *helper.LokiConfig
-	pipeline *PipelineBuilder
+	info        *reconcilers.Instance
+	labels      map[string]string
+	selector    map[string]string
+	desired     *flowslatest.FlowCollectorSpec
+	flowMetrics *v1alpha1.FlowMetricList
+	promTLS     *flowslatest.CertificateReference
+	confKind    ConfKind
+	volumes     volumes.Builder
+	loki        *helper.LokiConfig
+	pipeline    *PipelineBuilder
 }
 
 type builder = Builder
 
-func NewBuilder(info *reconcilers.Instance, desired *flowslatest.FlowCollectorSpec, ck ConfKind) (Builder, error) {
+func NewBuilder(info *reconcilers.Instance, desired *flowslatest.FlowCollectorSpec, flowMetrics *v1alpha1.FlowMetricList, ck ConfKind) (Builder, error) {
 	version := helper.ExtractVersion(info.Image)
 	name := name(ck)
 	var promTLS *flowslatest.CertificateReference
@@ -89,10 +91,11 @@ func NewBuilder(info *reconcilers.Instance, desired *flowslatest.FlowCollectorSp
 		selector: map[string]string{
 			"app": name,
 		},
-		desired:  desired,
-		confKind: ck,
-		promTLS:  promTLS,
-		loki:     info.Loki,
+		desired:     desired,
+		flowMetrics: flowMetrics,
+		confKind:    ck,
+		promTLS:     promTLS,
+		loki:        info.Loki,
 	}, nil
 }
 
@@ -141,7 +144,7 @@ func (b *builder) NewKafkaPipeline() PipelineBuilder {
 }
 
 func (b *builder) initPipeline(ingest config.PipelineBuilderStage) PipelineBuilder {
-	pipeline := newPipelineBuilder(b.desired, b.info.Loki, b.info.ClusterID, &b.volumes, &ingest)
+	pipeline := newPipelineBuilder(b.desired, b.flowMetrics, b.info.Loki, b.info.ClusterID, &b.volumes, &ingest)
 	b.pipeline = &pipeline
 	return pipeline
 }

controllers/flp/flp_controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/netobserv/network-observability-operator/api/v1alpha1"
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
 	"github.com/netobserv/network-observability-operator/pkg/helper"
@@ -47,6 +48,7 @@ func Start(ctx context.Context, mgr *manager.Manager) error {
 		Owns(&corev1.Namespace{}).
 		Owns(&corev1.Service{}).
 		Owns(&corev1.ServiceAccount{})
+		// TODO: create a namespaced watch for FlowMetrics
 
 	ctrl, err := builder.Build(&r)
 	if err != nil {
@@ -58,9 +60,9 @@ func Start(ctx context.Context, mgr *manager.Manager) error {
 }
 
 type subReconciler interface {
-	context(ctx context.Context) context.Context
-	cleanupNamespace(ctx context.Context)
-	reconcile(ctx context.Context, desired *flowslatest.FlowCollector) error
+	context(context.Context) context.Context
+	cleanupNamespace(context.Context)
+	reconcile(context.Context, *flowslatest.FlowCollector, *v1alpha1.FlowMetricList) error
 	getStatus() *status.Instance
 }
 
@@ -115,6 +117,12 @@ func (r *Reconciler) reconcile(ctx context.Context) error {
 		}
 	}
 
+	// List custom metrics
+	fm := v1alpha1.FlowMetricList{}
+	if err := r.Client.List(ctx, &fm, &client.ListOptions{Namespace: ns}); err != nil {
+		return r.status.Error("CantListFlowMetrics", err)
+	}
+
 	// Create sub-reconcilers
 	// TODO: refactor to move these subReconciler allocations in `Start`. It will involve some decoupling work, as currently
 	// `reconcilers.Common` is dependent on the FlowCollector object, which isn't known at start time.
@@ -139,7 +147,7 @@ func (r *Reconciler) reconcile(ctx context.Context) error {
 	}
 
 	for _, sr := range reconcilers {
-		if err := sr.reconcile(sr.context(ctx), fc); err != nil {
+		if err := sr.reconcile(sr.context(ctx), fc, &fm); err != nil {
 			return sr.getStatus().Error("FLPReconcileError", err)
 		}
 	}

controllers/flp/flp_ingest_objects.go

@@ -6,6 +6,7 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/netobserv/network-observability-operator/api/v1alpha1"
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
 	"github.com/netobserv/network-observability-operator/pkg/helper"
@@ -15,8 +16,8 @@ type ingestBuilder struct {
 	generic builder
 }
 
-func newIngestBuilder(info *reconcilers.Instance, desired *flowslatest.FlowCollectorSpec) (ingestBuilder, error) {
-	gen, err := NewBuilder(info, desired, ConfKafkaIngester)
+func newIngestBuilder(info *reconcilers.Instance, desired *flowslatest.FlowCollectorSpec, flowMetrics *v1alpha1.FlowMetricList) (ingestBuilder, error) {
+	gen, err := NewBuilder(info, desired, flowMetrics, ConfKafkaIngester)
 	return ingestBuilder{
 		generic: gen,
 	}, err

controllers/flp/flp_ingest_reconciler.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
+	"github.com/netobserv/network-observability-operator/api/v1alpha1"
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/constants"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
@@ -61,7 +62,7 @@ func (r *ingesterReconciler) getStatus() *status.Instance {
 	return &r.Status
 }
 
-func (r *ingesterReconciler) reconcile(ctx context.Context, desired *flowslatest.FlowCollector) error {
+func (r *ingesterReconciler) reconcile(ctx context.Context, desired *flowslatest.FlowCollector, flowMetrics *v1alpha1.FlowMetricList) error {
 	// Retrieve current owned objects
 	err := r.Managed.FetchAll(ctx)
 	if err != nil {
@@ -76,7 +77,7 @@ func (r *ingesterReconciler) reconcile(ctx context.Context, desired *flowslatest
 
 	r.Status.SetReady() // will be overidden if necessary, as error or pending
 
-	builder, err := newIngestBuilder(r.Instance, &desired.Spec)
+	builder, err := newIngestBuilder(r.Instance, &desired.Spec, flowMetrics)
 	if err != nil {
 		return err
 	}

controllers/flp/flp_monolith_objects.go

@@ -6,6 +6,7 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/netobserv/network-observability-operator/api/v1alpha1"
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
 	"github.com/netobserv/network-observability-operator/pkg/helper"
@@ -15,8 +16,8 @@ type monolithBuilder struct {
 	generic builder
 }
 
-func newMonolithBuilder(info *reconcilers.Instance, desired *flowslatest.FlowCollectorSpec) (monolithBuilder, error) {
-	gen, err := NewBuilder(info, desired, ConfMonolith)
+func newMonolithBuilder(info *reconcilers.Instance, desired *flowslatest.FlowCollectorSpec, flowMetrics *v1alpha1.FlowMetricList) (monolithBuilder, error) {
+	gen, err := NewBuilder(info, desired, flowMetrics, ConfMonolith)
 	return monolithBuilder{
 		generic: gen,
 	}, err

controllers/flp/flp_monolith_reconciler.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
+	"github.com/netobserv/network-observability-operator/api/v1alpha1"
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/constants"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
@@ -63,7 +64,7 @@ func (r *monolithReconciler) getStatus() *status.Instance {
 	return &r.Status
 }
 
-func (r *monolithReconciler) reconcile(ctx context.Context, desired *flowslatest.FlowCollector) error {
+func (r *monolithReconciler) reconcile(ctx context.Context, desired *flowslatest.FlowCollector, flowMetrics *v1alpha1.FlowMetricList) error {
 	// Retrieve current owned objects
 	err := r.Managed.FetchAll(ctx)
 	if err != nil {
@@ -78,7 +79,7 @@ func (r *monolithReconciler) reconcile(ctx context.Context, desired *flowslatest
 
 	r.Status.SetReady() // will be overidden if necessary, as error or pending
 
-	builder, err := newMonolithBuilder(r.Instance, &desired.Spec)
+	builder, err := newMonolithBuilder(r.Instance, &desired.Spec, flowMetrics)
 	if err != nil {
 		return err
 	}