Merge pull request #56335 from skrthomas/OSDOCS-5288

OSDOCS-5288:Exporting flow data to Kafka

Author: skrthomas

_topic_maps/_topic_map.yml

@@ -1444,6 +1444,8 @@ Topics:
     File: observing-network-traffic
   - Name: API reference
     File: flowcollector-api
+  - Name: JSON flows format reference
+    File: json-flows-format-reference
   - Name: Troubleshooting Network Observability
     File: troubleshooting-network-observability
 ---

modules/network-observability-enriched-flows-kafka.adoc

@@ -0,0 +1,39 @@
+// Module included in the following assemblies:
+//
+// network_observability/configuring-operator.adoc
+
+:_content-type: PROCEDURE
+[id="network-observability-enriched-flows-kafka_{context}"]
+= Export enriched network flow data
+
+You can send network flows to Kafka, so that they can be consumed by any processor or storage that supports Kafka input, such as Splunk, Elasticsearch, or Fluentd.
+
+.Prerequisites
+* Installed Kafka
+
+.Procedure
+
+. In the web console, navigate to *Operators* -> *Installed Operators*.
+. Under the *Provided APIs* heading for the *NetObserv Operator*, select *Flow Collector*. 
+. Select *cluster* and then select the *YAML* tab.
+. Edit the `FlowCollector` to configure `spec.exporters` as follows:
++
+[source,yaml]
+----
+apiVersion: flows.netobserv.io/v1alpha1
+kind: FlowCollector
+metadata:
+  name: cluster
+spec:
+  exporters:
+  - type: KAFKA
+      kafka:
+        address: "kafka-cluster-kafka-bootstrap.netobserv"
+        topic: netobserv-flows-export   <1>
+        tls: 
+          enable: false                 <2>
+  
+----
+<1> The Network Observability Operator exports all flows to the configured Kafka topic.
+<2> You can encrypt all communications to and from Kafka with SSL/TLS or mTLS. When enabled, the Kafka CA certificate must be available as a ConfigMap or a Secret, both in the namespace where the `flowlogs-pipeline` processor component is deployed (default: netobserv). It must be referenced with `spec.exporters.tls.caCert`. When using mTLS, client secrets must be available in these namespaces as well (they can be generated for instance using the AMQ Streams User Operator) and referenced with `spec.exporters.tls.userCert`.
+. After configuration, network flows data can be sent to an available output in a JSON format. For more information, see _Network flows format reference_

modules/network-observability-flowcollector-kafka-config.adoc

@@ -23,4 +23,4 @@ The following example shows how to modify the `FlowCollector` resource for {prod
 <1> Set `spec.deploymentModel` to `KAFKA` instead of `DIRECT` to enable the Kafka deployment model.
 <2> `spec.kafka.address` refers to the Kafka bootstrap server address. You can specify a port if needed, for instance `kafka-cluster-kafka-bootstrap.netobserv:9093` for using TLS on port 9093.
 <3> `spec.kafka.topic` should match the name of a topic created in Kafka.
-<4> `spec.kafka.tls` can be used to encrypt all communications to and from Kafka with TLS or mTLS. When enabled, the Kafka CA certificate must be available as a ConfigMap or a Secret, both in the namespace where the flowlogs-pipeline processor component is deployed (default: `netobserv`) and where the eBPF agents are deployed (default: `netobserv-privileged`). It must be referenced with `spec.kafka.tls.caCert`. When using mTLS, client secrets must be available in these namespaces as well (they can be generated for instance using the AMQ Streams User Operator) and referenced with `spec.kafka.tls.userCert`.
+<4> `spec.kafka.tls` can be used to encrypt all communications to and from Kafka with TLS or mTLS. When enabled, the Kafka CA certificate must be available as a ConfigMap or a Secret, both in the namespace where the `flowlogs-pipeline` processor component is deployed (default: `netobserv`) and where the eBPF agents are deployed (default: `netobserv-privileged`). It must be referenced with `spec.kafka.tls.caCert`. When using mTLS, client secrets must be available in these namespaces as well (they can be generated for instance using the AMQ Streams User Operator) and referenced with `spec.kafka.tls.userCert`.

modules/network-observability-flows-format.adoc

@@ -0,0 +1,322 @@
+// Automatically generated by 'hack/asciidoc-flows-gen.sh'. Do not edit.
+// Module included in the following assemblies:
+// json-flows-format-reference.adoc
+
+:_content-type: REFERENCE
+[id="network-observability-flows-format_{context}"]
+= Network Flows format reference
+The document is organized in two main categories: _Labels_ and regular _Fields_. This distinction only matters when querying Loki. This is because _Labels_, unlike _Fields_, must be used in link:https://grafana.com/docs/loki/latest/logql/log_queries/#log-stream-selector[stream selectors].
+
+If you are reading this specification as a reference for the Kafka export feature, you must treat all _Labels_ and _Fields_ as regualr fields and ignore any distinctions between them that are specific to Loki.
+
+
+== Labels
+
+'''
+
+SrcK8S_Namespace::
+
+• `Optional` *SrcK8S_Namespace*: `string`
+
+Source namespace
+
+'''
+
+DstK8S_Namespace::
+
+• `Optional` *DstK8S_Namespace*: `string`
+
+Destination namespace
+
+'''
+
+SrcK8S_OwnerName::
+
+• `Optional` *SrcK8S_OwnerName*: `string`
+
+Source owner, such as Deployment, StatefulSet, etc.
+
+'''
+
+DstK8S_OwnerName::
+
+• `Optional` *DstK8S_OwnerName*: `string`
+
+Destination owner, such as Deployment, StatefulSet, etc.
+
+'''
+
+FlowDirection::
+
+• *FlowDirection*: see the following section, _Enumeration: FlowDirection_ for more details.
+
+Flow direction from the node observation point
+
+'''
+
+_RecordType::
+
+• `Optional` *_RecordType*: `RecordType`
+
+Type of record: 'flowLog' for regular flow logs, or 'allConnections',
+'newConnection', 'heartbeat', 'endConnection' for conversation tracking
+
+== Fields
+
+'''
+
+SrcAddr::
+
+• *SrcAddr*: `string`
+
+Source IP address (ipv4 or ipv6)
+
+'''
+
+DstAddr::
+
+• *DstAddr*: `string`
+
+Destination IP address (ipv4 or ipv6)
+
+'''
+
+SrcMac::
+
+• *SrcMac*: `string`
+
+Source MAC address
+
+'''
+
+DstMac::
+
+• *DstMac*: `string`
+
+Destination MAC address
+
+'''
+
+SrcK8S_Name::
+
+• `Optional` *SrcK8S_Name*: `string`
+
+Name of the source matched Kubernetes object, such as Pod name, Service name, etc.
+
+'''
+
+DstK8S_Name::
+
+• `Optional` *DstK8S_Name*: `string`
+
+Name of the destination matched Kubernetes object, such as Pod name, Service name, etc.
+
+'''
+
+SrcK8S_Type::
+
+• `Optional` *SrcK8S_Type*: `string`
+
+Kind of the source matched Kubernetes object, such as Pod, Service, etc.
+
+'''
+
+DstK8S_Type::
+
+• `Optional` *DstK8S_Type*: `string`
+
+Kind of the destination matched Kubernetes object, such as Pod name, Service name, etc.
+
+'''
+
+SrcPort::
+
+• *SrcPort*: `number`
+
+Source port
+
+'''
+
+DstPort::
+
+• *DstPort*: `number`
+
+Destination port
+
+'''
+
+SrcK8S_OwnerType::
+
+• `Optional` *SrcK8S_OwnerType*: `string`
+
+Kind of the source Kubernetes owner, such as Deployment, StatefulSet, etc.
+
+'''
+
+DstK8S_OwnerType::
+
+• `Optional` *DstK8S_OwnerType*: `string`
+
+Kind of the destination Kubernetes owner, such as Deployment, StatefulSet, etc.
+
+'''
+
+SrcK8S_HostIP::
+
+• `Optional` *SrcK8S_HostIP*: `string`
+
+Source node IP
+
+'''
+
+DstK8S_HostIP::
+
+• `Optional` *DstK8S_HostIP*: `string`
+
+Destination node IP
+
+'''
+
+SrcK8S_HostName::
+
+• `Optional` *SrcK8S_HostName*: `string`
+
+Source node name
+
+'''
+
+DstK8S_HostName::
+
+• `Optional` *DstK8S_HostName*: `string`
+
+Destination node name
+
+'''
+
+Proto::
+
+• *Proto*: `number`
+
+L4 protocol
+
+'''
+
+Interface::
+
+• `Optional` *Interface*: `string`
+
+Network interface
+
+'''
+Packets::
+
+• *Packets*: `number`
+
+Number of packets in this flow
+
+
+'''
+
+Packets_AB::
+
+• `Optional` *Packets_AB*: `number`
+
+In conversation tracking, A to B packets counter per conversation
+
+'''
+
+Packets_BA::
+
+• `Optional` *Packets_BA*: `number`
+
+In conversation tracking, B to A packets counter per conversation
+
+'''
+
+Bytes::
+
+• *Bytes*: `number`
+
+Number of bytes in this flow
+
+'''
+
+Bytes_AB::
+
+• `Optional` *Bytes_AB*: `number`
+
+In conversation tracking, A to B bytes counter per conversation
+
+'''
+
+Bytes_BA::
+
+• `Optional` *Bytes_BA*: `number`
+
+In conversation tracking, B to A bytes counter per conversation
+
+'''
+
+TimeFlowStartMs::
+
+• *TimeFlowStartMs*: `number`
+
+Start timestamp of this flow, in milliseconds
+
+'''
+
+TimeFlowEndMs::
+
+• *TimeFlowEndMs*: `number`
+
+End timestamp of this flow, in milliseconds
+
+'''
+
+TimeReceived::
+
+• *TimeReceived*: `number`
+
+Timestamp when this flow was received and processed by the flow collector, in seconds
+
+'''
+
+_HashId::
+
+• `Optional` *_HashId*: `string`
+
+In conversation tracking, the conversation identifier
+
+'''
+
+_IsFirst::
+
+• `Optional` *_IsFirst*: `string`
+
+In conversation tracking, a flag identifying the first flow
+
+'''
+
+numFlowLogs::
+
+• `Optional` *numFlowLogs*: `number`
+
+In conversation tracking, a counter of flow logs per conversation
+
+== Enumeration: FlowDirection
+
+'''
+
+Ingress::
+
+• *Ingress* = `"0"`
+
+Incoming traffic, from node observation point
+
+'''
+
+Egress::
+
+• *Egress* = `"1"`
+
+Outgoing traffic, from node observation point

modules/network-observability-operator-install.adoc

@@ -27,6 +27,14 @@ This documentation assumes that your `LokiStack` instance name is `loki`. Using
 +
 * *spec.agent.ebpf.Sampling* : Specify a sampling size for flows. Lower sampling sizes will have higher impact on resource utilization. For more information, see the `FlowCollector` API reference, under spec.agent.ebpf. 
 * *spec.deploymentModel*: If you are using Kafka, verify Kafka is selected.
+* *spec.exporters*: If you are using Kafka, you can optionally send network flows to Kafka, so that they can be consumed by any processor or storage that supports Kafka input, such as Splunk, Elasticsearch, or Fluentd. To do this, set the following specifications:
+** Set the *type* to `KAFKA`. 
+** Set the *address* as `kafka-cluster-kafka-bootstrap.netobserv`. 
+** Set the *topic* as `netobserv-flows-export`. The Operator exports all flows to the configured Kafka topic. 
+** Set the following *tls* specifications: 
+*** *certFile*: `service-ca.crt`, *name*: `kafka-gateway-ca-bundle`, and *type*: `configmap`.
++
+You can also configure this option at a later time by directly editing the YAML. For more information, see _Export enriched network flow data_.
 * *loki.url*: Since authentication is specified separately, this URL needs to be updated to `https://loki-gateway-http.netobserv.svc:8080/api/logs/v1/network`. The first part of the URL, "loki", should match the name of your LokiStack.
 * *loki.statusUrl*: Set this to `https://loki-query-frontend-http.netobserv.svc:3100/`. The first part of the URL, "loki", should match the name of your LokiStack.
 * *loki.authToken*: Select the `FORWARD` value.