NETOBSERV-101 R&D: Kube enricher write path for downstream operator

api/v1alpha1/flowcollector_types.go

@@ -308,6 +308,11 @@ type FlowCollectorLoki struct {
 	// it will be ignored if instanceSpec is specified
 	TenantID string `json:"tenantID,omitempty"`
 
+	//+kubebuilder:default:=false
+	// SendAuthToken is a flag to enable or disable Authorization header from service account secret
+	// It allows authentication to loki operator gateway
+	SendAuthToken bool `json:"sendAuthToken,omitempty"`
+
 	//+kubebuilder:default:="1s"
 	// BatchWait is max time to wait before sending a batch
 	BatchWait metav1.Duration `json:"batchWait,omitempty"`

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

@@ -1609,6 +1609,12 @@ spec:
                       If empty, the URL value will be used (assuming that the Loki
                       ingester and querier are int he same host).
                     type: string
+                  sendAuthToken:
+                    default: false
+                    description: SendAuthToken is a flag to enable or disable Authorization
+                      header from service account secret It allows authentication
+                      to loki operator gateway
+                    type: boolean
                   staticLabels:
                     additionalProperties:
                       type: string

controllers/consoleplugin/consoleplugin_objects.go

@@ -27,6 +27,7 @@ const configFile = "config.yaml"
 const configVolume = "config-volume"
 const configPath = "/opt/app-root/"
 const lokiCerts = "loki-certs"
+const tokensPath = "/var/run/secrets/tokens/"
 
 // PodConfigurationDigest is an annotation name to facilitate pod restart after
 // any external configuration change
@@ -99,6 +100,13 @@ func (b *builder) deployment(cmDigest string) *appsv1.Deployment {
 	}
 }
 
+func tokenPath(desiredLoki *flowsv1alpha1.FlowCollectorLoki) string {
+	if desiredLoki.SendAuthToken {
+		return tokensPath + constants.PluginName
+	}
+	return ""
+}
+
 func buildArgs(desired *flowsv1alpha1.FlowCollectorConsolePlugin, desiredLoki *flowsv1alpha1.FlowCollectorLoki) []string {
 	args := []string{
 		"-cert", "/var/serving-cert/tls.crt",
@@ -116,6 +124,9 @@ func buildArgs(desired *flowsv1alpha1.FlowCollectorConsolePlugin, desiredLoki *f
 			args = append(args, "--loki-ca-path", helper.GetCACertPath(&desiredLoki.TLS, lokiCerts))
 		}
 	}
+	if desiredLoki.SendAuthToken {
+		args = append(args, "-loki-token-path", tokenPath(desiredLoki))
+	}
 	return args
 }
 
@@ -143,16 +154,20 @@ func (b *builder) podTemplate(cmDigest string) *corev1.PodTemplateSpec {
 		Name:      secretName,
 		MountPath: "/var/serving-cert",
 		ReadOnly:  true,
+	}, {
+		Name:      configVolume,
+		MountPath: configPath,
+		ReadOnly:  true,
 	},
-		{
-			Name:      configVolume,
-			MountPath: configPath,
-			ReadOnly:  true,
-		}}
+	}
 
 	args := buildArgs(b.desired, b.desiredLoki)
 	if b.desiredLoki != nil && b.desiredLoki.TLS.Enable && !b.desiredLoki.TLS.InsecureSkipVerify {
-		helper.AppendCertVolumes(volumes, volumeMounts, &b.desiredLoki.TLS, lokiCerts)
+		volumes, volumeMounts = helper.AppendCertVolumes(volumes, volumeMounts, &b.desiredLoki.TLS, lokiCerts)
+	}
+
+	if b.desiredLoki.SendAuthToken {
+		volumes, volumeMounts = helper.AppendTokenVolume(volumes, volumeMounts, constants.PluginName, constants.PluginName)
 	}
 
 	return &corev1.PodTemplateSpec{

controllers/flowlogspipeline/flp_objects.go

@@ -181,6 +181,10 @@ func (b *builder) podTemplate(hostNetwork bool, configDigest string) corev1.PodT
 		volumes, volumeMounts = helper.AppendCertVolumes(volumes, volumeMounts, &b.desiredLoki.TLS, lokiCerts)
 	}
 
+	if b.desiredLoki.SendAuthToken {
+		volumes, volumeMounts = helper.AppendTokenVolume(volumes, volumeMounts, constants.FLPName+b.confKindSuffix, constants.FLPName)
+	}
+
 	container := corev1.Container{
 		Name:            constants.FLPName + b.confKindSuffix,
 		Image:           b.desired.Image,
@@ -333,20 +337,35 @@ func (b *builder) addTransformStages(stage *config.PipelineBuilderStage) error {
 		lokiWrite.TimestampLabel = "TimeFlowEndMs"
 		lokiWrite.TimestampScale = "1ms"
 		lokiWrite.TenantID = b.desiredLoki.TenantID
+
+		var authorization *promConfig.Authorization
+		if b.desiredLoki.SendAuthToken {
+			authorization = &promConfig.Authorization{
+				Type:            "Bearer",
+				CredentialsFile: helper.TokensPath + constants.FLPName,
+			}
+		}
+
 		if b.desiredLoki != nil && b.desiredLoki.TLS.Enable {
 			if b.desiredLoki.TLS.InsecureSkipVerify {
 				lokiWrite.ClientConfig = &promConfig.HTTPClientConfig{
+					Authorization: authorization,
 					TLSConfig: promConfig.TLSConfig{
 						InsecureSkipVerify: true,
 					},
 				}
 			} else {
 				lokiWrite.ClientConfig = &promConfig.HTTPClientConfig{
+					Authorization: authorization,
 					TLSConfig: promConfig.TLSConfig{
 						CAFile: helper.GetCACertPath(&b.desiredLoki.TLS, lokiCerts),
 					},
 				}
 			}
+		} else {
+			lokiWrite.ClientConfig = &promConfig.HTTPClientConfig{
+				Authorization: authorization,
+			}
 		}
 	}
 	enrichedStage.WriteLoki("loki", lokiWrite)

docs/FlowCollector.md

@@ -2857,6 +2857,15 @@ Settings related to the Loki client, used as a flow store.
           QuerierURL specifies the address of the Loki querier service, in case it is different from the Loki ingester URL. If empty, the URL value will be used (assuming that the Loki ingester and querier are int he same host).<br/>
         </td>
         <td>false</td>
+      </tr><tr>
+        <td><b>sendAuthToken</b></td>
+        <td>boolean</td>
+        <td>
+          SendAuthToken is a flag to enable or disable Authorization header from service account secret It allows authentication to loki operator gateway<br/>
+          <br/>
+            <i>Default</i>: false<br/>
+        </td>
+        <td>false</td>
       </tr><tr>
         <td><b>staticLabels</b></td>
         <td>map[string]string</td>

pkg/helper/tokens.go

@@ -0,0 +1,33 @@
+package helper
+
+import (
+	corev1 "k8s.io/api/core/v1"
+)
+
+const TokensPath = "/var/run/secrets/tokens/"
+
+// AppendTokenVolume will add a volume + volume mount for a service account token if defined
+func AppendTokenVolume(volumes []corev1.Volume, volumeMounts []corev1.VolumeMount, name string, fileName string) ([]corev1.Volume, []corev1.VolumeMount) {
+	volOut := append(volumes,
+		corev1.Volume{
+			Name: name,
+			VolumeSource: corev1.VolumeSource{
+				Projected: &corev1.ProjectedVolumeSource{
+					Sources: []corev1.VolumeProjection{
+						{
+							ServiceAccountToken: &corev1.ServiceAccountTokenProjection{
+								Path: fileName,
+							},
+						},
+					},
+				},
+			},
+		})
+	vmOut := append(volumeMounts,
+		corev1.VolumeMount{
+			MountPath: TokensPath,
+			Name:      name,
+		},
+	)
+	return volOut, vmOut
+}