NETOBSERV-627 merge Agent and FLP

netobserv · Nov 27, 2023 · 31c6c21 · 31c6c21
1 parent 9ae09a5
commit 31c6c21
Show file tree

Hide file tree

Showing 11 changed files with 103 additions and 27 deletions.
diff --git a/controllers/ebpf/agent_controller.go b/controllers/ebpf/agent_controller.go
@@ -9,6 +9,7 @@ import (
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/constants"
 	"github.com/netobserv/network-observability-operator/controllers/ebpf/internal/permissions"
+	"github.com/netobserv/network-observability-operator/controllers/flowlogspipeline"
 	"github.com/netobserv/network-observability-operator/controllers/operator"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
 	"github.com/netobserv/network-observability-operator/pkg/helper"
@@ -50,6 +51,7 @@ const (
 	envKafkaSASLSecretPath        = "KAFKA_SASL_CLIENT_SECRET_PATH"
 	envLogLevel                   = "LOG_LEVEL"
 	envDedupe                     = "DEDUPER"
+	envFLPConfig                  = "FLP_CONFIG"
 	dedupeDefault                 = "firstCome"
 	envDedupeJustMark             = "DEDUPER_JUST_MARK"
 	dedupeJustMarkDefault         = "true"
@@ -62,7 +64,7 @@ const (
 
 const (
 	exportKafka        = "kafka"
-	exportGRPC         = "grpc"
+	exportFLP          = "direct-flp"
 	kafkaCerts         = "kafka-certs"
 	averageMessageSize = 100
 	bpfTraceMountName  = "bpf-kernel-debug"
@@ -324,25 +326,32 @@ func (c *AgentController) envConfig(ctx context.Context, coll *flowslatest.FlowC
 			)
 		}
 	} else {
-		config = append(config, corev1.EnvVar{Name: envExport, Value: exportGRPC})
-		// When flowlogs-pipeline is deployed as a daemonset, each agent must send
-		// data to the pod that is deployed in the same host
+		flpConfig, err := c.buildFLPConfig(&coll.Spec)
+		if err != nil {
+			return nil, err
+		}
+		config = append(config, corev1.EnvVar{Name: envExport, Value: exportFLP})
 		config = append(config, corev1.EnvVar{
-			Name: envFlowsTargetHost,
-			ValueFrom: &corev1.EnvVarSource{
-				FieldRef: &corev1.ObjectFieldSelector{
-					APIVersion: "v1",
-					FieldPath:  "status.hostIP",
-				},
-			},
-		}, corev1.EnvVar{
-			Name:  envFlowsTargetPort,
-			Value: strconv.Itoa(int(coll.Spec.Processor.Port)),
+			Name:  envFLPConfig,
+			Value: flpConfig,
 		})
 	}
 	return config, nil
 }
 
+func (c *AgentController) buildFLPConfig(desired *flowslatest.FlowCollectorSpec) (string, error) {
+	flpBuilder, err := flowlogspipeline.NewBuilder(c.NewInstance(c.config.EBPFAgentImage), desired, flowlogspipeline.ConfMonolith)
+	if err != nil {
+		return "", err
+	}
+	pipeline := flpBuilder.NewInProcessPipeline()
+	err = pipeline.AddProcessorStages()
+	if err != nil {
+		return "", err
+	}
+	return flpBuilder.GetJSONConfig()
+}
+
 func requiredAction(current, desired *v1.DaemonSet) reconcileAction {
 	if desired == nil {
 		return actionNone

diff --git a/controllers/ebpf/internal/permissions/permissions.go b/controllers/ebpf/internal/permissions/permissions.go
@@ -6,11 +6,13 @@ import (
 
 	flowslatest "github.com/netobserv/network-observability-operator/api/v1beta2"
 	"github.com/netobserv/network-observability-operator/controllers/constants"
+	"github.com/netobserv/network-observability-operator/controllers/flowlogspipeline"
 	"github.com/netobserv/network-observability-operator/controllers/reconcilers"
 	"github.com/netobserv/network-observability-operator/pkg/helper"
 
 	osv1 "github.com/openshift/api/security/v1"
 	v1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -45,6 +47,9 @@ func (c *Reconciler) Reconcile(ctx context.Context, desired *flowslatest.FlowCol
 	if err := c.reconcileVendorPermissions(ctx, desired); err != nil {
 		return fmt.Errorf("reconciling vendor permissions: %w", err)
 	}
+	if err := c.reconcileRoles(ctx); err != nil {
+		return fmt.Errorf("reconciling roles: %w", err)
+	}
 	return nil
 }
 
@@ -222,3 +227,26 @@ func (c *Reconciler) cleanupPreviousNamespace(ctx context.Context) error {
 	}
 	return nil
 }
+
+func (c *Reconciler) reconcileRoles(ctx context.Context) error {
+	cr := flowlogspipeline.BuildClusterRoleTransformer()
+	if err := c.ReconcileClusterRole(ctx, cr); err != nil {
+		return err
+	}
+	crb := &rbacv1.ClusterRoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: cr.Name + "-agent",
+		},
+		RoleRef: rbacv1.RoleRef{
+			APIGroup: "rbac.authorization.k8s.io",
+			Kind:     "ClusterRole",
+			Name:     cr.Name,
+		},
+		Subjects: []rbacv1.Subject{{
+			Kind:      "ServiceAccount",
+			Name:      constants.EBPFServiceAccount,
+			Namespace: c.PrivilegedNamespace(),
+		}},
+	}
+	return c.ReconcileClusterRoleBinding(ctx, crb)
+}
diff --git a/controllers/flowlogspipeline/flp_common_objects.go b/controllers/flowlogspipeline/flp_common_objects.go
@@ -140,6 +140,10 @@ func (b *builder) NewKafkaPipeline() PipelineBuilder {
 	}))
 }
 
+func (b *builder) NewInProcessPipeline() PipelineBuilder {
+	return b.initPipeline(config.NewPresetIngesterPipeline())
+}
+
 func (b *builder) initPipeline(ingest config.PipelineBuilderStage) PipelineBuilder {
 	pipeline := newPipelineBuilder(b.desired, b.info.Loki, &b.volumes, &ingest)
 	b.pipeline = &pipeline
@@ -291,7 +295,9 @@ func (b *builder) ConfigMap() (*corev1.ConfigMap, string, error) {
 
 func (b *builder) GetJSONConfig() (string, error) {
 	metricsSettings := config.MetricsSettings{
-		Port:    int(b.desired.Processor.Metrics.Server.Port),
+		PromConnectionInfo: api.PromConnectionInfo{
+			Port: int(b.desired.Processor.Metrics.Server.Port),
+		},
 		Prefix:  "netobserv_",
 		NoPanic: true,
 	}

diff --git a/controllers/flowlogspipeline/flp_monolith_reconciler.go b/controllers/flowlogspipeline/flp_monolith_reconciler.go
@@ -82,7 +82,7 @@ func (r *flpMonolithReconciler) reconcile(ctx context.Context, desired *flowslat
 	}
 
 	// Monolith only used without Kafka
-	if helper.UseKafka(&desired.Spec) {
+	if helper.UseKafka(&desired.Spec) || helper.UseMergedAgentFLP(&desired.Spec) {
 		r.Managed.TryDeleteAll(ctx)
 		return nil
 	}

diff --git a/go.mod b/go.mod
@@ -5,7 +5,7 @@ go 1.20
 require (
 	github.com/go-logr/logr v1.3.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231108130854-fac8a619b238
+	github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231123152750-f3b03fa192aa
 	github.com/onsi/ginkgo/v2 v2.13.1
 	github.com/onsi/gomega v1.30.0
 	github.com/openshift/api v0.0.0-20220112145620-704957ce4980

diff --git a/go.sum b/go.sum
@@ -155,8 +155,8 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8m
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f h1:KUppIJq7/+SVif2QVs3tOP0zanoHgBEVAwHxUSIzRqU=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw=
-github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231108130854-fac8a619b238 h1:kBz9gPV19sBfgHx9SXEbDZsRHurRKUrBtmlGOAS9YDE=
-github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231108130854-fac8a619b238/go.mod h1:gl/lTacCn1Hb8i+ULrTbieXbxXVimRYHJBQrbGz9r0I=
+github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231123152750-f3b03fa192aa h1:g09mCEph6ujskD/1EOTeJt49zqLq4zieZjpaXsGzG6s=
+github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231123152750-f3b03fa192aa/go.mod h1:OZPLp6ypAR9NBsAGveBdGHcVQbuZKR2PY/JuXfng2yM=
 github.com/netobserv/prometheus-common v0.44.0-netobserv h1:1DEcYfG8UQcDRjHax+MBJGpwbnsQNB+fiiMh54mW4yU=
 github.com/netobserv/prometheus-common v0.44.0-netobserv/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=

diff --git a/pkg/helper/flowcollector.go b/pkg/helper/flowcollector.go
@@ -28,6 +28,10 @@ func UseKafka(spec *flowslatest.FlowCollectorSpec) bool {
 	return spec.DeploymentModel == flowslatest.DeploymentModelKafka
 }
 
+func UseMergedAgentFLP(spec *flowslatest.FlowCollectorSpec) bool {
+	return spec.DeploymentModel == flowslatest.DeploymentModelDirect && spec.Agent.Type == flowslatest.AgentEBPF
+}
+
 func HasKafkaExporter(spec *flowslatest.FlowCollectorSpec) bool {
 	for _, ex := range spec.Exporters {
 		if ex.Type == flowslatest.KafkaExporter {

diff --git a/vendor/github.com/netobserv/flowlogs-pipeline/pkg/api/encode_prom.go b/vendor/github.com/netobserv/flowlogs-pipeline/pkg/api/encode_prom.go
diff --git a/vendor/github.com/netobserv/flowlogs-pipeline/pkg/config/config.go b/vendor/github.com/netobserv/flowlogs-pipeline/pkg/config/config.go
diff --git a/vendor/github.com/netobserv/flowlogs-pipeline/pkg/config/pipeline_builder.go b/vendor/github.com/netobserv/flowlogs-pipeline/pkg/config/pipeline_builder.go
diff --git a/vendor/modules.txt b/vendor/modules.txt
@@ -113,7 +113,7 @@ github.com/munnerz/goautoneg
 # github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f
 ## explicit
 github.com/mwitkow/go-conntrack
-# github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231108130854-fac8a619b238
+# github.com/netobserv/flowlogs-pipeline v0.1.11-0.20231123152750-f3b03fa192aa
 ## explicit; go 1.19
 github.com/netobserv/flowlogs-pipeline/pkg/api
 github.com/netobserv/flowlogs-pipeline/pkg/config