Repository: netobserv/flowlogs-pipeline
Pull request #630 (2024-03-13 13:26:00)

NETOBSERV-1566: ipfix: make RTT optional
## Description

- refactor IPFIX fields mapping / definition
- allow optional fields
- add interfaces and directions (plural) fields to IPFIX template
- add tests for partial records & non-enriched records

<!-- Fill-in description here -->

## Dependencies

<!-- List here any related PRs with links, that need to be pulled also for testing -->
n/a

## Checklist

If you are not familiar with our processes or don't know what to answer in the list below, let us know in a comment: the maintainers will take care of that.

* [x] Will this change affect NetObserv / Network Observability operator? If not, you can ignore the rest of this checklist.
* [x] Is this PR backed with a JIRA ticket? If so, make sure it is written as a title prefix _(in general, PRs affecting the NetObserv/Network Observability product should be backed with a JIRA ticket - especially if they bring user facing changes)._
* [ ] Does this PR require product documentation?
  * [ ] If so, make sure the JIRA epic is labelled with "documentation" and provides a description relevant for doc writers, such as use cases or scenarios. Any required step to activate or configure the feature should be documented there, such as new CRD knobs.
* [ ] Does this PR require a product release notes entry?
  * [ ] If so, fill in "Release Note Text" in the JIRA.
* [ ] Is there anything else the QE team should know before testing? E.g: configuration changes, environment setup, etc.
  * [ ] If so, make sure it is described in the JIRA ticket.
* QE requirements (check 1 from the list):
  * [x] Standard QE validation, with pre-merge tests unless stated otherwise.
  * [ ] Regression tests only (e.g. refactoring with no user-facing change).
  * [ ] No QE (e.g. trivial change with high reviewer's confidence, or per agreement with the QE team).


openshift-ci-robot: @jotak: This pull request references NETOBSERV-1566 which is a valid jira issue.

Warning: The referenced jira issue has an invalid target version for the target branch this PR targets: expected the bug to target the "4.16.0" version, but no target version was set.

<details>

In response to [this](https://github.com/netobserv/flowlogs-pipeline/pull/630):

>## Description
>
>- refactor IPFIX fields mapping / definition
>- allow optional fields
>- add tests for partial records & non-enriched records
>
><!-- Fill-in description here -->
>
>## Dependencies
>
><!-- List here any related PRs with links, that need to be pulled also for testing -->
>n/a
>
>## Checklist
>
>If you are not familiar with our processes or don't know what to answer in the list below, let us know in a comment: the maintainers will take care of that.
>
>* [x] Will this change affect NetObserv / Network Observability operator? If not, you can ignore the rest of this checklist.
>* [x] Is this PR backed with a JIRA ticket? If so, make sure it is written as a title prefix _(in general, PRs affecting the NetObserv/Network Observability product should be backed with a JIRA ticket - especially if they bring user facing changes)._
>* [ ] Does this PR require product documentation?
>  * [ ] If so, make sure the JIRA epic is labelled with "documentation" and provides a description relevant for doc writers, such as use cases or scenarios. Any required step to activate or configure the feature should be documented there, such as new CRD knobs.
>* [ ] Does this PR require a product release notes entry?
>  * [ ] If so, fill in "Release Note Text" in the JIRA.
>* [ ] Is there anything else the QE team should know before testing? E.g: configuration changes, environment setup, etc.
>  * [ ] If so, make sure it is described in the JIRA ticket.
>* QE requirements (check 1 from the list):
>  * [x] Standard QE validation, with pre-merge tests unless stated otherwise.
>  * [ ] Regression tests only (e.g. refactoring with no user-facing change).
>  * [ ] No QE (e.g. trivial change with high reviewer's confidence, or per agreement with the QE team).
>


Instructions for interacting with me using PR comments are available [here](https://prow.ci.openshift.org/command-help?repo=netobserv%2Fflowlogs-pipeline).  If you have questions or suggestions related to my behavior, please file an issue against the [openshift-eng/jira-lifecycle-plugin](https://github.com/openshift-eng/jira-lifecycle-plugin/issues/new) repository.
</details>
openshift-ci[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [jotak](https://github.com/jotak). For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=netobserv%2Fflowlogs-pipeline).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/netobserv/flowlogs-pipeline/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["jotak"]} -->
openshift-ci-robot: @jotak: This pull request references NETOBSERV-1566 which is a valid jira issue.

Warning: The referenced jira issue has an invalid target version for the target branch this PR targets: expected the bug to target the "4.16.0" version, but no target version was set.

<details>

In response to [this](https://github.com/netobserv/flowlogs-pipeline/pull/630):

>## Description
>
>- refactor IPFIX fields mapping / definition
>- allow optional fields
>- add interfaces and directions (plural) fields to IPFIX template
>- add tests for partial records & non-enriched records
>
><!-- Fill-in description here -->
>
>## Dependencies
>
><!-- List here any related PRs with links, that need to be pulled also for testing -->
>n/a
>
>## Checklist
>
>If you are not familiar with our processes or don't know what to answer in the list below, let us know in a comment: the maintainers will take care of that.
>
>* [x] Will this change affect NetObserv / Network Observability operator? If not, you can ignore the rest of this checklist.
>* [x] Is this PR backed with a JIRA ticket? If so, make sure it is written as a title prefix _(in general, PRs affecting the NetObserv/Network Observability product should be backed with a JIRA ticket - especially if they bring user facing changes)._
>* [ ] Does this PR require product documentation?
>  * [ ] If so, make sure the JIRA epic is labelled with "documentation" and provides a description relevant for doc writers, such as use cases or scenarios. Any required step to activate or configure the feature should be documented there, such as new CRD knobs.
>* [ ] Does this PR require a product release notes entry?
>  * [ ] If so, fill in "Release Note Text" in the JIRA.
>* [ ] Is there anything else the QE team should know before testing? E.g: configuration changes, environment setup, etc.
>  * [ ] If so, make sure it is described in the JIRA ticket.
>* QE requirements (check 1 from the list):
>  * [x] Standard QE validation, with pre-merge tests unless stated otherwise.
>  * [ ] Regression tests only (e.g. refactoring with no user-facing change).
>  * [ ] No QE (e.g. trivial change with high reviewer's confidence, or per agreement with the QE team).
>


Instructions for interacting with me using PR comments are available [here](https://prow.ci.openshift.org/command-help?repo=netobserv%2Fflowlogs-pipeline).  If you have questions or suggestions related to my behavior, please file an issue against the [openshift-eng/jira-lifecycle-plugin](https://github.com/openshift-eng/jira-lifecycle-plugin/issues/new) repository.
</details>
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) Report
Attention: Patch coverage is `0%` with `113 lines` in your changes are missing coverage. Please review.
> Project coverage is 67.41%. Comparing base [(`ddc1f67`)](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/commit/ddc1f67022a5890098c263f4537de0f352709c11?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) to head [(`1a3119a`)](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv).

| [Files](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) | Patch % | Lines |
|---|---|---|
| [pkg/pipeline/write/write\_ipfix.go](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv#diff-cGtnL3BpcGVsaW5lL3dyaXRlL3dyaXRlX2lwZml4Lmdv) | 0.00% | [113 Missing :warning: ](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #630      +/-   ##
==========================================
+ Coverage   67.02%   67.41%   +0.38%     
==========================================
  Files         110      110              
  Lines        7676     7632      -44     
==========================================
  Hits         5145     5145              
+ Misses       2217     2173      -44     
  Partials      314      314              
```

| [Flag](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) | Coverage Δ | |
|---|---|---|
| [unittests](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) | `67.41% <0.00%> (+0.38%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv#carryforward-flags-in-the-pull-request-comment) to find out more.


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/630?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv).

github-actions[bot]: New image:
quay.io/netobserv/flowlogs-pipeline:a2b8863

It will expire after two weeks.

To deploy this build, run from the operator repo, assuming the operator is running:
```bash
USER=netobserv VERSION=a2b8863 make set-flp-image
```

Amoghrd: @jotak The export is still failing with ipfix collextor pod showing logs
`E0313 17:46:49.328657 1 tcp.go:87] error in decoding message: template 256 with obsDomainID 1 does not exist
E0313 17:46:49.328693 1 tcp.go:87] error in decoding message: template 256 with obsDomainID 1 does not exist
E0313 17:46:49.328717 1 tcp.go:87] error in decoding message: template 256 with obsDomainID 1 does not exist
E0313 17:46:49.328860 1 tcp.go:87] error in decoding message: template 256 with obsDomainID 1 does not exist`
jotak: @Amoghrd I tried using another FLP instance deployed as a collector and I don't see any error. I'm wondering if this could be due to your collector.
Do you know what exactly is this, that you're using: https://github.com/Amoghrd/go-ipfix/blob/custom-image/build/yamls/ipfix-collector.yaml#L50-L54 ? Is it pre-setting some IPFIX template? That could be the cause of the issue, whener new fields are added to our template, if you force a pre-registered template. cc @jpinsonneau  do you know? (since it seems you created this custom image)

FWIW we can also use a custom FLP deployment to ingest IPFIX (with a workflow like:
`agent -> usual netobserv' FLP + IPFIX exporter -> Custom FLP`)
See my PR https://github.com/netobserv/flowlogs-pipeline/pull/633 for such as deployment
Amoghrd: Yeah this template was built using @jpinsonneau help. He had mentioned that a custom image was built to cater to the needs of the IBM folks as far as I remember. But yeah for all IPFIX export testing we have been using this YAML. @jpinsonneau Could you confirm if we should move away from this YAML and use @jotak suggestion of a custom FLP deployment?
Amoghrd: Tried the FLP collector and it worked fine.
/label qe-approved
openshift-ci[bot]: New changes are detected. LGTM label has been removed.
jotak: (rebased)

Patch:
diff --git a/contrib/kubernetes/ipfix-collector-stdout.yaml b/contrib/kubernetes/ipfix-collector-stdout.yaml
new file mode 100644
index 00000000..d7211b00
--- /dev/null
+++ b/contrib/kubernetes/ipfix-collector-stdout.yaml
@@ -0,0 +1,68 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: flowlogs-pipeline
+  labels:
+    app: flowlogs-pipeline
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: flowlogs-pipeline
+  template:
+    metadata:
+      labels:
+        app: flowlogs-pipeline
+    spec:
+      containers:
+        - name: flowlogs-pipeline
+          image: quay.io/netobserv/flowlogs-pipeline:main
+          args:
+          - "--config=/etc/flowlogs-pipeline/config.yaml"
+          ports:
+            - containerPort: 2055
+          imagePullPolicy: IfNotPresent
+          volumeMounts:
+            - name: configuration
+              mountPath: "/etc/flowlogs-pipeline/"
+      volumes:
+        - name: configuration
+          configMap:
+            name: flp-config
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: flowlogs-pipeline
+  labels:
+    app: flowlogs-pipeline
+spec:
+  ports:
+    - port: 2055
+      targetPort: 2055
+      protocol: UDP
+      name: ipfix
+  selector:
+    app: flowlogs-pipeline
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: flp-config
+data:
+  config.yaml: |
+    log-level: info
+    pipeline:
+      - name: ingest
+      - name: write
+        follows: ingest
+    parameters:
+      - name: ingest
+        ingest:
+          type: collector
+          collector:
+            hostName: 0.0.0.0
+            port: 2055
+      - name: write
+        write:
+          type: stdout
diff --git a/pkg/pipeline/write/write_ipfix.go b/pkg/pipeline/write/write_ipfix.go
index cf73b83d..5ecb7f6a 100644
--- a/pkg/pipeline/write/write_ipfix.go
+++ b/pkg/pipeline/write/write_ipfix.go
@@ -20,6 +20,8 @@ package write
 import (
 	"fmt"
 	"net"
+	"strconv"
+	"strings"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
@@ -40,6 +42,14 @@ type writeIpfix struct {
 	entitiesV6         []entities.InfoElementWithValue
 }
 
+type FieldMap struct {
+	Key      string
+	Getter   func(entities.InfoElementWithValue) any
+	Setter   func(entities.InfoElementWithValue, any)
+	Matcher  func(entities.InfoElementWithValue, any) bool
+	Optional bool
+}
+
 // IPv6Type value as defined in IEEE 802: https://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml
 const IPv6Type = 0x86DD
 
@@ -78,6 +88,191 @@ var (
 	}
 	CustomNetworkFields = []string{
 		"timeFlowRttNs",
+		"interfaces",
+		"directions",
+	}
+
+	MapIPFIXKeys = map[string]FieldMap{
+		"sourceIPv4Address": {
+			Key:    "SrcAddr",
+			Getter: func(elt entities.InfoElementWithValue) any { return elt.GetIPAddressValue().String() },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetIPAddressValue(net.ParseIP(rec.(string))) },
+		},
+		"destinationIPv4Address": {
+			Key:    "DstAddr",
+			Getter: func(elt entities.InfoElementWithValue) any { return elt.GetIPAddressValue().String() },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetIPAddressValue(net.ParseIP(rec.(string))) },
+		},
+		"sourceIPv6Address": {
+			Key:    "SrcAddr",
+			Getter: func(elt entities.InfoElementWithValue) any { return elt.GetIPAddressValue().String() },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetIPAddressValue(net.ParseIP(rec.(string))) },
+		},
+		"destinationIPv6Address": {
+			Key:    "DstAddr",
+			Getter: func(elt entities.InfoElementWithValue) any { return elt.GetIPAddressValue().String() },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetIPAddressValue(net.ParseIP(rec.(string))) },
+		},
+		"nextHeaderIPv6": {
+			Key:    "Proto",
+			Getter: func(elt entities.InfoElementWithValue) any { return uint32(elt.GetUnsigned8Value()) },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned8Value(uint8(rec.(uint32))) },
+		},
+		"sourceMacAddress": {
+			Key: "SrcMac",
+			Setter: func(elt entities.InfoElementWithValue, rec any) {
+				elt.SetMacAddressValue(net.HardwareAddr(rec.(string)))
+			},
+			Matcher: func(_ entities.InfoElementWithValue, _ any) bool {
+				// Getting some discrepancies here, need to figure out why
+				return true
+			},
+		},
+		"destinationMacAddress": {
+			Key: "DstMac",
+			Setter: func(elt entities.InfoElementWithValue, rec any) {
+				elt.SetMacAddressValue(net.HardwareAddr(rec.(string)))
+			},
+			Matcher: func(_ entities.InfoElementWithValue, _ any) bool {
+				// Getting some discrepancies here, need to figure out why
+				return true
+			},
+		},
+		"ethernetType": {
+			Key:    "Etype",
+			Getter: func(elt entities.InfoElementWithValue) any { return uint32(elt.GetUnsigned16Value()) },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned16Value(uint16(rec.(uint32))) },
+		},
+		"flowDirection": {
+			Key: "IfDirections",
+			Setter: func(elt entities.InfoElementWithValue, rec any) {
+				if dirs, ok := rec.([]int); ok && len(dirs) > 0 {
+					elt.SetUnsigned8Value(uint8(dirs[0]))
+				}
+			},
+			Matcher: func(elt entities.InfoElementWithValue, expected any) bool {
+				ifdirs := expected.([]int)
+				return int(elt.GetUnsigned8Value()) == ifdirs[0]
+			},
+		},
+		"directions": {
+			Key: "IfDirections",
+			Getter: func(elt entities.InfoElementWithValue) any {
+				var dirs []int
+				for _, dir := range strings.Split(elt.GetStringValue(), ",") {
+					d, _ := strconv.Atoi(dir)
+					dirs = append(dirs, d)
+				}
+				return dirs
+			},
+			Setter: func(elt entities.InfoElementWithValue, rec any) {
+				if dirs, ok := rec.([]int); ok && len(dirs) > 0 {
+					var asStr []string
+					for _, dir := range dirs {
+						asStr = append(asStr, strconv.Itoa(dir))
+					}
+					elt.SetStringValue(strings.Join(asStr, ","))
+				}
+			},
+		},
+		"protocolIdentifier": {
+			Key:    "Proto",
+			Getter: func(elt entities.InfoElementWithValue) any { return uint32(elt.GetUnsigned8Value()) },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned8Value(uint8(rec.(uint32))) },
+		},
+		"sourceTransportPort": {
+			Key:    "SrcPort",
+			Getter: func(elt entities.InfoElementWithValue) any { return uint32(elt.GetUnsigned16Value()) },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned16Value(uint16(rec.(uint32))) },
+		},
+		"destinationTransportPort": {
+			Key:    "DstPort",
+			Getter: func(elt entities.InfoElementWithValue) any { return uint32(elt.GetUnsigned16Value()) },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned16Value(uint16(rec.(uint32))) },
+		},
+		"octetDeltaCount": {
+			Key:    "Bytes",
+			Getter: func(elt entities.InfoElementWithValue) any { return elt.GetUnsigned64Value() },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned64Value(rec.(uint64)) },
+		},
+		"flowStartMilliseconds": {
+			Key:    "TimeFlowStartMs",
+			Getter: func(elt entities.InfoElementWithValue) any { return int64(elt.GetUnsigned64Value()) },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned64Value(uint64(rec.(int64))) },
+		},
+		"flowEndMilliseconds": {
+			Key:    "TimeFlowEndMs",
+			Getter: func(elt entities.InfoElementWithValue) any { return int64(elt.GetUnsigned64Value()) },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned64Value(uint64(rec.(int64))) },
+		},
+		"packetDeltaCount": {
+			Key:    "Packets",
+			Getter: func(elt entities.InfoElementWithValue) any { return elt.GetUnsigned64Value() },
+			Setter: func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned64Value(rec.(uint64)) },
+		},
+		"interfaceName": {
+			Key: "Interfaces",
+			Setter: func(elt entities.InfoElementWithValue, rec any) {
+				if ifs, ok := rec.([]string); ok && len(ifs) > 0 {
+					elt.SetStringValue(ifs[0])
+				}
+			},
+			Matcher: func(elt entities.InfoElementWithValue, expected any) bool {
+				ifs := expected.([]string)
+				return elt.GetStringValue() == ifs[0]
+			},
+		},
+		"interfaces": {
+			Key:    "Interfaces",
+			Getter: func(elt entities.InfoElementWithValue) any { return strings.Split(elt.GetStringValue(), ",") },
+			Setter: func(elt entities.InfoElementWithValue, rec any) {
+				if ifs, ok := rec.([]string); ok {
+					elt.SetStringValue(strings.Join(ifs, ","))
+				}
+			},
+		},
+		"sourcePodNamespace": {
+			Key:      "SrcK8S_Namespace",
+			Getter:   func(elt entities.InfoElementWithValue) any { return elt.GetStringValue() },
+			Setter:   func(elt entities.InfoElementWithValue, rec any) { elt.SetStringValue(rec.(string)) },
+			Optional: true,
+		},
+		"sourcePodName": {
+			Key:      "SrcK8S_Name",
+			Getter:   func(elt entities.InfoElementWithValue) any { return elt.GetStringValue() },
+			Setter:   func(elt entities.InfoElementWithValue, rec any) { elt.SetStringValue(rec.(string)) },
+			Optional: true,
+		},
+		"destinationPodNamespace": {
+			Key:      "DstK8S_Namespace",
+			Getter:   func(elt entities.InfoElementWithValue) any { return elt.GetStringValue() },
+			Setter:   func(elt entities.InfoElementWithValue, rec any) { elt.SetStringValue(rec.(string)) },
+			Optional: true,
+		},
+		"destinationPodName": {
+			Key:      "DstK8S_Name",
+			Getter:   func(elt entities.InfoElementWithValue) any { return elt.GetStringValue() },
+			Setter:   func(elt entities.InfoElementWithValue, rec any) { elt.SetStringValue(rec.(string)) },
+			Optional: true,
+		},
+		"sourceNodeName": {
+			Key:      "SrcK8S_HostName",
+			Getter:   func(elt entities.InfoElementWithValue) any { return elt.GetStringValue() },
+			Setter:   func(elt entities.InfoElementWithValue, rec any) { elt.SetStringValue(rec.(string)) },
+			Optional: true,
+		},
+		"destinationNodeName": {
+			Key:      "DstK8S_HostName",
+			Getter:   func(elt entities.InfoElementWithValue) any { return elt.GetStringValue() },
+			Setter:   func(elt entities.InfoElementWithValue, rec any) { elt.SetStringValue(rec.(string)) },
+			Optional: true,
+		},
+		"timeFlowRttNs": {
+			Key:      "TimeFlowRttNs",
+			Getter:   func(elt entities.InfoElementWithValue) any { return int64(elt.GetUnsigned64Value()) },
+			Setter:   func(elt entities.InfoElementWithValue, rec any) { elt.SetUnsigned64Value(uint64(rec.(int64))) },
+			Optional: true,
+		},
 	}
 )
 
@@ -155,6 +350,16 @@ func loadCustomRegistry(EnterpriseID uint32) error {
 		ilog.WithError(err).Errorf("Failed to register element")
 		return err
 	}
+	err = registry.PutInfoElement((*entities.NewInfoElement("interfaces", 7741, entities.String, EnterpriseID, 65535)), EnterpriseID)
+	if err != nil {
+		ilog.WithError(err).Errorf("Failed to register element")
+		return err
+	}
+	err = registry.PutInfoElement((*entities.NewInfoElement("directions", 7742, entities.String, EnterpriseID, 65535)), EnterpriseID)
+	if err != nil {
+		ilog.WithError(err).Errorf("Failed to register element")
+		return err
+	}
 	return nil
 }
 
@@ -237,173 +442,27 @@ func SendTemplateRecordv6(exporter *ipfixExporter.ExportingProcess, enrichEnterp
 }
 
 //nolint:cyclop
-func setStandardIEValue(record config.GenericMap, ieValPtr *entities.InfoElementWithValue) error {
+func setElementValue(record config.GenericMap, ieValPtr *entities.InfoElementWithValue) error {
 	ieVal := *ieValPtr
-	switch ieVal.GetName() {
-	case "ethernetType":
-		if record["Etype"] != nil {
-			ieVal.SetUnsigned16Value(uint16(record["Etype"].(uint32)))
-		} else {
-			return fmt.Errorf("unable to find ethernet type (Etype) in record")
-		}
-	case "flowDirection":
-		dirs := record["IfDirections"].([]int)
-		if len(dirs) > 0 {
-			ieVal.SetUnsigned8Value(uint8(dirs[0]))
-		} else {
-			return fmt.Errorf("unable to find flow direction (flowDirection) in record")
-		}
-	case "sourceMacAddress":
-		if record["SrcMac"] != nil {
-			ieVal.SetMacAddressValue(net.HardwareAddr(record["SrcMac"].(string)))
-		} else {
-			return fmt.Errorf("unable to find source mac address (SrcMac) in record")
-		}
-	case "destinationMacAddress":
-		if record["DstMac"] != nil {
-			ieVal.SetMacAddressValue(net.HardwareAddr(record["DstMac"].(string)))
-		} else {
-			return fmt.Errorf("unable to find dest mac address (DstMac) in record")
-		}
-	case "sourceIPv4Address":
-		if record["SrcAddr"] != nil {
-			ieVal.SetIPAddressValue(net.ParseIP(record["SrcAddr"].(string)))
-		} else {
-			return fmt.Errorf("unable to find source IPv4 address (SrcAddr) in record")
-		}
-	case "destinationIPv4Address":
-		if record["DstAddr"] != nil {
-			ieVal.SetIPAddressValue(net.ParseIP(record["DstAddr"].(string)))
-		} else {
-			return fmt.Errorf("unable to find dest IPv4 address (DstAddr) in record")
-		}
-	case "sourceIPv6Address":
-		if record["SrcAddr"] != nil {
-			ieVal.SetIPAddressValue(net.ParseIP(record["SrcAddr"].(string)))
-		} else {
-			return fmt.Errorf("unable to find source IPv6 address (SrcAddr) in record")
-		}
-	case "destinationIPv6Address":
-		if record["DstAddr"] != nil {
-			ieVal.SetIPAddressValue(net.ParseIP(record["DstAddr"].(string)))
-		} else {
-			return fmt.Errorf("unable to find dest IPv6 address (DstAddr) in record")
-		}
-	case "protocolIdentifier":
-		if record["Proto"] != nil {
-			ieVal.SetUnsigned8Value(uint8(record["Proto"].(uint32)))
-		} else {
-			return fmt.Errorf("unable to find protocol identifier (Proto) in record")
-		}
-	case "nextHeaderIPv6":
-		if record["Proto"] != nil {
-			ieVal.SetUnsigned8Value(uint8(record["Proto"].(uint32)))
-		} else {
-			return fmt.Errorf("unable to find next header (Proto) in record")
-		}
-	case "sourceTransportPort":
-		if record["SrcPort"] != nil {
-			ieVal.SetUnsigned16Value(uint16(record["SrcPort"].(uint32)))
-		} else {
-			return fmt.Errorf("unable to find source port (SrcPort) in record")
-		}
-	case "destinationTransportPort":
-		if record["DstPort"] != nil {
-			ieVal.SetUnsigned16Value(uint16(record["DstPort"].(uint32)))
-		} else {
-			return fmt.Errorf("unable to find dest port (DstPort) in record")
-		}
-	case "octetDeltaCount":
-		if record["Bytes"] != nil {
-			ieVal.SetUnsigned64Value(record["Bytes"].(uint64))
-		} else {
-			return fmt.Errorf("unable to find bytes in record")
-		}
-	case "flowStartMilliseconds":
-		if record["TimeFlowStartMs"] != nil {
-			ieVal.SetUnsigned64Value(uint64(record["TimeFlowStartMs"].(int64)))
-		} else {
-			return fmt.Errorf("unable to find flow start time (TimeFlowStartMs) in record")
-		}
-	case "flowEndMilliseconds":
-		if record["TimeFlowEndMs"] != nil {
-			ieVal.SetUnsigned64Value(uint64(record["TimeFlowEndMs"].(int64)))
-		} else {
-			return fmt.Errorf("unable to find flow end time (TimeFlowEndMs) in record")
-		}
-	case "packetDeltaCount":
-		if record["Packets"] != nil {
-			ieVal.SetUnsigned64Value(record["Packets"].(uint64))
-		} else {
-			return fmt.Errorf("unable to find packets in record")
-		}
-	case "interfaceName":
-		interfaces := record["Interfaces"].([]string)
-		if len(interfaces) > 0 {
-			ieVal.SetStringValue(interfaces[0])
-		} else {
-			return fmt.Errorf("unable to find interface in record")
-		}
-	case "timeFlowRttNs":
-		if record["TimeFlowRttNs"] != nil {
-			ieVal.SetUnsigned64Value(uint64(record["TimeFlowRttNs"].(int64)))
-		} else {
-			return fmt.Errorf("unable to find timeflowrtt in record")
-		}
+	name := ieVal.GetName()
+	mapping, ok := MapIPFIXKeys[name]
+	if !ok {
+		return nil
+	}
+	if value := record[mapping.Key]; value != nil {
+		mapping.Setter(ieVal, value)
+	} else if !mapping.Optional {
+		return fmt.Errorf("unable to find %s (%s) in record", name, mapping.Key)
 	}
 	return nil
 }
 
-func setKubeIEValue(record config.GenericMap, ieValPtr *entities.InfoElementWithValue) {
-	ieVal := *ieValPtr
-	switch ieVal.GetName() {
-	case "sourcePodNamespace":
-		if record["SrcK8S_Namespace"] != nil {
-			ieVal.SetStringValue(record["SrcK8S_Namespace"].(string))
-		} else {
-			ieVal.SetStringValue("none")
-		}
-	case "sourcePodName":
-		if record["SrcK8S_Name"] != nil {
-			ieVal.SetStringValue(record["SrcK8S_Name"].(string))
-		} else {
-			ieVal.SetStringValue("none")
-		}
-	case "destinationPodNamespace":
-		if record["DstK8S_Namespace"] != nil {
-			ieVal.SetStringValue(record["DstK8S_Namespace"].(string))
-		} else {
-			ieVal.SetStringValue("none")
-		}
-	case "destinationPodName":
-		if record["DstK8S_Name"] != nil {
-			ieVal.SetStringValue(record["DstK8S_Name"].(string))
-		} else {
-			ieVal.SetStringValue("none")
-		}
-	case "sourceNodeName":
-		if record["SrcK8S_HostName"] != nil {
-			ieVal.SetStringValue(record["SrcK8S_HostName"].(string))
-		} else {
-			ieVal.SetStringValue("none")
-		}
-	case "destinationNodeName":
-		if record["DstK8S_HostName"] != nil {
-			ieVal.SetStringValue(record["DstK8S_HostName"].(string))
-		} else {
-			ieVal.SetStringValue("none")
-		}
-	}
-}
-func setEntities(record config.GenericMap, enrichEnterpriseID uint32, elements *[]entities.InfoElementWithValue) error {
+func setEntities(record config.GenericMap, elements *[]entities.InfoElementWithValue) error {
 	for _, ieVal := range *elements {
-		err := setStandardIEValue(record, &ieVal)
+		err := setElementValue(record, &ieVal)
 		if err != nil {
 			return err
 		}
-		if enrichEnterpriseID != 0 {
-			setKubeIEValue(record, &ieVal)
-		}
 	}
 	return nil
 }
@@ -412,13 +471,13 @@ func (t *writeIpfix) sendDataRecord(record config.GenericMap, v6 bool) error {
 	var templateID uint16
 	if v6 {
 		templateID = t.templateIDv6
-		err := setEntities(record, t.enrichEnterpriseID, &t.entitiesV6)
+		err := setEntities(record, &t.entitiesV6)
 		if err != nil {
 			return err
 		}
 	} else {
 		templateID = t.templateIDv4
-		err := setEntities(record, t.enrichEnterpriseID, &t.entitiesV4)
+		err := setEntities(record, &t.entitiesV4)
 		if err != nil {
 			return err
 		}


Test patch:
diff --git a/pkg/pipeline/write/testnorace/write_ipfix_test.go b/pkg/pipeline/write/testnorace/write_ipfix_test.go
index 33a5d4a9..f589f149 100644
--- a/pkg/pipeline/write/testnorace/write_ipfix_test.go
+++ b/pkg/pipeline/write/testnorace/write_ipfix_test.go
@@ -51,7 +51,7 @@ var (
 		},
 		TimeFlowStart: timestamppb.New(startTime),
 		TimeFlowEnd:   timestamppb.New(endTime),
-		Interface:     "eth0",
+
 		AgentIp: &pbflow.IP{
 			IpFamily: &pbflow.IP_Ipv4{Ipv4: 0x0a090807},
 		},
@@ -73,6 +73,10 @@ var (
 				Interface: "eth0",
 				Direction: pbflow.Direction_EGRESS,
 			},
+			{
+				Interface: "a1234567",
+				Direction: pbflow.Direction_INGRESS,
+			},
 		},
 	}
 )
@@ -115,11 +119,14 @@ func TestEnrichedIPFIXFlow(t *testing.T) {
 	cp.CloseMsgChan()
 	cp.Stop()
 
+	expectedFields := append(write.IPv4IANAFields, write.KubeFields...)
+	expectedFields = append(expectedFields, write.CustomNetworkFields...)
+
 	// Check template
 	assert.Equal(t, uint16(10), tplv4Msg.GetVersion())
 	templateSet := tplv4Msg.GetSet()
 	templateElements := templateSet.GetRecords()[0].GetOrderedElementList()
-	assert.Len(t, templateElements, 21)
+	assert.Len(t, templateElements, len(expectedFields))
 	assert.Equal(t, uint32(0), templateElements[0].GetInfoElement().EnterpriseId)
 
 	// Check data
@@ -127,9 +134,67 @@ func TestEnrichedIPFIXFlow(t *testing.T) {
 	dataSet := dataMsg.GetSet()
 	record := dataSet.GetRecords()[0]
 
+	for _, name := range expectedFields {
+		element, _, exist := record.GetInfoElementWithValue(name)
+		assert.Truef(t, exist, "element with name %s should exist in the record", name)
+		assert.NotNil(t, element)
+		matchElement(t, element, flow)
+	}
+}
+
+func TestEnrichedIPFIXPartialFlow(t *testing.T) {
+	cp := startCollector(t)
+	addr := cp.GetAddress().(*net.UDPAddr)
+
+	flow := decode.PBFlowToMap(&FullPBFlow)
+
+	// Add partial enrichment
+	flow["SrcK8S_Name"] = "pod A"
+	flow["SrcK8S_Namespace"] = "ns1"
+	flow["SrcK8S_HostName"] = "node1"
+
+	// Remove a field
+	delete(flow, "TimeFlowRttNs")
+
+	writer, err := write.NewWriteIpfix(config.StageParam{
+		Write: &config.Write{
+			Ipfix: &api.WriteIpfix{
+				TargetHost:   addr.IP.String(),
+				TargetPort:   addr.Port,
+				Transport:    addr.Network(),
+				EnterpriseID: 9999,
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	writer.Write(flow)
+
+	// Read collector
+	// 1st = IPv4 template
+	tplv4Msg := <-cp.GetMsgChan()
+	// 2nd = IPv6 template (ignore)
+	<-cp.GetMsgChan()
+	// 3rd = data record
+	dataMsg := <-cp.GetMsgChan()
+	cp.CloseMsgChan()
+	cp.Stop()
+
 	expectedFields := append(write.IPv4IANAFields, write.KubeFields...)
 	expectedFields = append(expectedFields, write.CustomNetworkFields...)
 
+	// Check template
+	assert.Equal(t, uint16(10), tplv4Msg.GetVersion())
+	templateSet := tplv4Msg.GetSet()
+	templateElements := templateSet.GetRecords()[0].GetOrderedElementList()
+	assert.Len(t, templateElements, len(expectedFields))
+	assert.Equal(t, uint32(0), templateElements[0].GetInfoElement().EnterpriseId)
+
+	// Check data
+	assert.Equal(t, uint16(10), dataMsg.GetVersion())
+	dataSet := dataMsg.GetSet()
+	record := dataSet.GetRecords()[0]
+
 	for _, name := range expectedFields {
 		element, _, exist := record.GetInfoElementWithValue(name)
 		assert.Truef(t, exist, "element with name %s should exist in the record", name)
@@ -138,52 +203,86 @@ func TestEnrichedIPFIXFlow(t *testing.T) {
 	}
 }
 
+func TestBasicIPFIXFlow(t *testing.T) {
+	cp := startCollector(t)
+	addr := cp.GetAddress().(*net.UDPAddr)
+
+	flow := decode.PBFlowToMap(&FullPBFlow)
+
+	// Add partial enrichment (must be ignored)
+	flow["SrcK8S_Name"] = "pod A"
+	flow["SrcK8S_Namespace"] = "ns1"
+	flow["SrcK8S_HostName"] = "node1"
+
+	writer, err := write.NewWriteIpfix(config.StageParam{
+		Write: &config.Write{
+			Ipfix: &api.WriteIpfix{
+				TargetHost: addr.IP.String(),
+				TargetPort: addr.Port,
+				Transport:  addr.Network(),
+				// No enterprise ID here
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	writer.Write(flow)
+
+	// Read collector
+	// 1st = IPv4 template
+	tplv4Msg := <-cp.GetMsgChan()
+	// 2nd = IPv6 template (ignore)
+	<-cp.GetMsgChan()
+	// 3rd = data record
+	dataMsg := <-cp.GetMsgChan()
+	cp.CloseMsgChan()
+	cp.Stop()
+
+	// Check template
+	assert.Equal(t, uint16(10), tplv4Msg.GetVersion())
+	templateSet := tplv4Msg.GetSet()
+	templateElements := templateSet.GetRecords()[0].GetOrderedElementList()
+	assert.Len(t, templateElements, len(write.IPv4IANAFields))
+	assert.Equal(t, uint32(0), templateElements[0].GetInfoElement().EnterpriseId)
+
+	// Check data
+	assert.Equal(t, uint16(10), dataMsg.GetVersion())
+	dataSet := dataMsg.GetSet()
+	record := dataSet.GetRecords()[0]
+
+	for _, name := range write.IPv4IANAFields {
+		element, _, exist := record.GetInfoElementWithValue(name)
+		assert.Truef(t, exist, "element with name %s should exist in the record", name)
+		assert.NotNil(t, element)
+		matchElement(t, element, flow)
+	}
+
+	// Make sure enriched fields are absent
+	for _, name := range write.KubeFields {
+		element, _, exist := record.GetInfoElementWithValue(name)
+		assert.Falsef(t, exist, "element with name %s should NOT exist in the record", name)
+		assert.Nil(t, element)
+	}
+}
+
 //nolint:cyclop
 func matchElement(t *testing.T, element entities.InfoElementWithValue, flow config.GenericMap) {
-	switch element.GetName() {
-	case "sourceIPv4Address":
-		assert.Equal(t, flow["SrcAddr"], element.GetIPAddressValue().String())
-	case "destinationIPv4Address":
-		assert.Equal(t, flow["DstAddr"], element.GetIPAddressValue().String())
-	case "ethernetType":
-		assert.Equal(t, flow["Etype"], uint32(element.GetUnsigned16Value()))
-	case "flowDirection":
-		assert.Equal(t, flow["IfDirections"], []int{int(element.GetUnsigned8Value())})
-	case "protocolIdentifier":
-		assert.Equal(t, flow["Proto"], uint32(element.GetUnsigned8Value()))
-	case "sourceTransportPort":
-		assert.Equal(t, flow["SrcPort"], uint32(element.GetUnsigned16Value()))
-	case "destinationTransportPort":
-		assert.Equal(t, flow["DstPort"], uint32(element.GetUnsigned16Value()))
-	case "octetDeltaCount":
-		assert.Equal(t, flow["Bytes"], element.GetUnsigned64Value())
-	case "flowStartMilliseconds":
-		assert.Equal(t, flow["TimeFlowStartMs"], int64(element.GetUnsigned64Value()))
-	case "flowEndMilliseconds":
-		assert.Equal(t, flow["TimeFlowEndMs"], int64(element.GetUnsigned64Value()))
-	case "packetDeltaCount":
-		assert.Equal(t, flow["Packets"], element.GetUnsigned64Value())
-	case "interfaceName":
-		assert.Equal(t, flow["Interfaces"], []string{element.GetStringValue()})
-	case "sourcePodNamespace":
-		assert.Equal(t, flow["SrcK8S_Namespace"], element.GetStringValue())
-	case "sourcePodName":
-		assert.Equal(t, flow["SrcK8S_Name"], element.GetStringValue())
-	case "destinationPodNamespace":
-		assert.Equal(t, flow["DstK8S_Namespace"], element.GetStringValue())
-	case "destinationPodName":
-		assert.Equal(t, flow["DstK8S_Name"], element.GetStringValue())
-	case "sourceNodeName":
-		assert.Equal(t, flow["SrcK8S_HostName"], element.GetStringValue())
-	case "destinationNodeName":
-		assert.Equal(t, flow["DstK8S_HostName"], element.GetStringValue())
-	case "timeFlowRttNs":
-		assert.Equal(t, uint64(flow["TimeFlowRttNs"].(int64)), element.GetUnsigned64Value())
-	case "sourceMacAddress":
-	case "destinationMacAddress":
-		// Getting some discrepancies here, need to figure out why
-	default:
-		assert.Fail(t, "missing check on element", element.GetName())
+	name := element.GetName()
+	mapping, ok := write.MapIPFIXKeys[name]
+	if !ok {
+		assert.Fail(t, "missing check on element", name)
+		return
+	}
+	expected := flow[mapping.Key]
+	if mapping.Matcher != nil {
+		assert.True(t, mapping.Matcher(element, expected), "unexpected "+name)
+	} else {
+		value := mapping.Getter(element)
+		if expected == nil {
+			assert.Empty(t, value, "unexpected "+name)
+		} else {
+			assert.Equal(t, expected, value, "unexpected "+name)
+		}
 	}
 }