Repository: netobserv/flowlogs-pipeline
Pull request #603 (2024-02-20 18:58:58)

Add new field to ipfix exporter for flowRtt
## Description

I am working on implementing a fix for 
https://github.com/netobserv/flowlogs-pipeline/issues/544

There are many new fields added to the pipeline but not yet written via IPFIX. I attempted to start with a single field and get feedback before I added several fields using a bad approach. 

## Dependencies

None

## Checklist

** I am not familiar with the processes **

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.

* [ ] Will this change affect NetObserv / Network Observability operator? If not, you can ignore the rest of this checklist.
* [ ] 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):
  * [ ] 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[bot]: Hi @bhale. Thanks for your PR.

I'm waiting for a [netobserv](https://github.com/orgs/netobserv/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/netobserv/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=netobserv%2Fflowlogs-pipeline).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) 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 assign [kalmanmeth](https://github.com/kalmanmeth) for approval. 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":["kalmanmeth"]} -->
bhale: I think this pattern is going badly, partly because there isn't a logical place to put the new fields... the previous Kubernetes enrichment section added strings that don't change or contain values, like source namespace. With the new data being produced upstream in the ebpf agent, there are new fields we want to export that are not static data / enrichment, or data that already exists in an IANA data model  - such as timeFlowRttNs. Should we be looking to create a third set of fields, or am I missing a better solution? 
jotak: Hi @bhale , thanks for the contrib!

I am not super familiar with the ipfix code, perhaps if @praveingk is around he can better answer that me.
I'm not sure to understand what you mean when saying "the previous Kubernetes enrichment section added strings that don't change or contain values" ... kube enrichment should set values and vary according to the src/dst IPs found in the flows, this is not supposed to be static ... or I'm misunderstanding something.

That said, it seems indeed wrong to tie agent-specific features such as RTT to the K8s model, as this isn't related to k8s. I'm not sure how to do that (as said I'm not very familiar with this code) but I can take a look and try
praveingk: @bhale Thanks for addressing this. I had previously added _enrichEnterpriseID_ to support fields that are not supported by IANA registry and a custom registry. In this case, RTT isn't supported by IANA, and it makes sense to add a separate function  and not adding as part of K8s enrichment. 
jotak: @bhale my PR https://github.com/netobserv/flowlogs-pipeline/pull/606 adds a test with ipfix that should be helpful here
There will be a small conflict with your PR, as you'll then need to add any additional field managed in a global list here: https://github.com/netobserv/flowlogs-pipeline/pull/606/files#diff-9f1ae1c930f06df67c0493ad41f8774ae4fa31832f12f7a8d4f38f0ffa534b74R79-R81
bhale: > @bhale my PR #606 adds a test with ipfix that should be helpful here There will be a small conflict with your PR, as you'll then need to add any additional field managed in a global list here: https://github.com/netobserv/flowlogs-pipeline/pull/606/files#diff-9f1ae1c930f06df67c0493ad41f8774ae4fa31832f12f7a8d4f38f0ffa534b74R79-R81

This is a huge help thank you! I will rebase on these changes tomorrow, it looks like it will simplify things. 
bhale: Thanks for all your help and advice, @praveingk and @jotak! I incorporated the refactoring and tests from jotak, integrated my changes to export timeFlowRttNs, and added a test for it. The test passes, and I can see valid data for RTT in IBM SevOne for the new RTT field. 

Regarding the comment on using field id 7739 - my colleague has access to a database at IBM that tracks the fields for our enterpriseId. Someone had already claimed 7739, so I skipped it. 
jotak: > Thanks for all your help and advice, @praveingk and @jotak! I incorporated the refactoring and tests from jotak, integrated my changes to export timeFlowRttNs, and added a test for it. The test passes, and I can see valid data for RTT in IBM SevOne for the new RTT field.

Nice 👍

> 
> Regarding the comment on using field id 7739 - my colleague has access to a database at IBM that tracks the fields for our enterpriseId. Someone had already claimed 7739, so I skipped it.

ok .. I don't know why exactly it starts at `7733`, it would probably have been better if we had a range that wouldn't overlap with other's using the same enterprise id
openshift-merge-robot: PR needs rebase.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
jotak: btw I've merged my PR, this one should be rebased
bhale: > btw I've merged my PR, this one should be rebased

Rebase good to go!
rupertgregoryibm: @jotak FYI, yes, between myself & Dale Bowie, we track the field ID allocations. The 7xxx starting number was partly looking for a range that doesn't overlap with other IBM products. We should probably switch it from private to public git at some point.

Patch:
diff --git a/docs/api.md b/docs/api.md
index d45a2f5e..26ae6e93 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -17,22 +17,16 @@ Following is the supported API format for prometheus encode:
                      counter: monotonically increasing counter whose value can only increase
                      histogram: counts samples in configurable buckets
                      agg_histogram: counts samples in configurable buckets, pre-aggregated via an Aggregate stage
-                 filter: an optional criterion to filter entries by. Deprecated: use filters instead.
-                     key: the key to match and filter by
-                     value: the value to match and filter by
-                     type: (enum) the type of filter match: exact (default), presence, absence or regex
-                         exact: match exactly the provided fitler value
-                         presence: filter key must be present (filter value is ignored)
-                         absence: filter key must be absent (filter value is ignored)
-                         regex: match filter value as a regular expression
                  filters: a list of criteria to filter entries by
                          key: the key to match and filter by
                          value: the value to match and filter by
-                         type: (enum) the type of filter match: exact (default), presence, absence or regex
-                             exact: match exactly the provided fitler value
+                         type: (enum) the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex
+                             equal: match exactly the provided filter value
+                             not_equal: the value must be different from the provided filter
                              presence: filter key must be present (filter value is ignored)
                              absence: filter key must be absent (filter value is ignored)
-                             regex: match filter value as a regular expression
+                             match_regex: match filter value as a regular expression
+                             not_match_regex: the filter value must not match the provided regular expression
                  valueKey: entry key from which to resolve metric value
                  labels: labels to be associated with the metric
                  buckets: histogram buckets
@@ -350,22 +344,16 @@ Following is the supported API format for writing metrics to an OpenTelemetry co
                      counter: monotonically increasing counter whose value can only increase
                      histogram: counts samples in configurable buckets
                      agg_histogram: counts samples in configurable buckets, pre-aggregated via an Aggregate stage
-                 filter: an optional criterion to filter entries by. Deprecated: use filters instead.
-                     key: the key to match and filter by
-                     value: the value to match and filter by
-                     type: (enum) the type of filter match: exact (default), presence, absence or regex
-                         exact: match exactly the provided fitler value
-                         presence: filter key must be present (filter value is ignored)
-                         absence: filter key must be absent (filter value is ignored)
-                         regex: match filter value as a regular expression
                  filters: a list of criteria to filter entries by
                          key: the key to match and filter by
                          value: the value to match and filter by
-                         type: (enum) the type of filter match: exact (default), presence, absence or regex
-                             exact: match exactly the provided fitler value
+                         type: (enum) the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex
+                             equal: match exactly the provided filter value
+                             not_equal: the value must be different from the provided filter
                              presence: filter key must be present (filter value is ignored)
                              absence: filter key must be absent (filter value is ignored)
-                             regex: match filter value as a regular expression
+                             match_regex: match filter value as a regular expression
+                             not_match_regex: the filter value must not match the provided regular expression
                  valueKey: entry key from which to resolve metric value
                  labels: labels to be associated with the metric
                  buckets: histogram buckets
diff --git a/pkg/api/api.go b/pkg/api/api.go
index 0bbc0d72..83f921e8 100644
--- a/pkg/api/api.go
+++ b/pkg/api/api.go
@@ -50,10 +50,12 @@ const (
 	AddKubernetesRuleType        = "add_kubernetes"
 	AddKubernetesInfraRuleType   = "add_kubernetes_infra"
 	ReinterpretDirectionRuleType = "reinterpret_direction"
-	PromFilterExact              = "exact"
+	PromFilterEqual              = "equal"
+	PromFilterNotEqual           = "not_equal"
 	PromFilterPresence           = "presence"
 	PromFilterAbsence            = "absence"
-	PromFilterRegex              = "regex"
+	PromFilterRegex              = "match_regex"
+	PromFilterNotRegex           = "not_match_regex"
 
 	TagYaml = "yaml"
 	TagDoc  = "doc"
diff --git a/pkg/api/encode_prom.go b/pkg/api/encode_prom.go
index 07213d61..001dab3f 100644
--- a/pkg/api/encode_prom.go
+++ b/pkg/api/encode_prom.go
@@ -50,7 +50,6 @@ type PromConnectionInfo struct {
 type MetricsItem struct {
 	Name       string          `yaml:"name" json:"name" doc:"the metric name"`
 	Type       string          `yaml:"type" json:"type" enum:"MetricEncodeOperationEnum" doc:"one of the following:"`
-	Filter     MetricsFilter   `yaml:"filter,omitempty" json:"filter,omitempty" doc:"an optional criterion to filter entries by. Deprecated: use filters instead."`
 	Filters    []MetricsFilter `yaml:"filters" json:"filters" doc:"a list of criteria to filter entries by"`
 	ValueKey   string          `yaml:"valueKey" json:"valueKey" doc:"entry key from which to resolve metric value"`
 	Labels     []string        `yaml:"labels" json:"labels" doc:"labels to be associated with the metric"`
@@ -58,26 +57,21 @@ type MetricsItem struct {
 	ValueScale float64         `yaml:"valueScale" json:"valueScale" doc:"scale factor of the value (MetricVal := FlowVal / Scale)"`
 }
 
-func (i *MetricsItem) GetFilters() []MetricsFilter {
-	if len(i.Filters) == 0 && i.Filter.Key != "" {
-		return []MetricsFilter{i.Filter}
-	}
-	return i.Filters
-}
-
 type MetricsItems []MetricsItem
 
 type MetricsFilter struct {
 	Key   string `yaml:"key" json:"key" doc:"the key to match and filter by"`
 	Value string `yaml:"value" json:"value" doc:"the value to match and filter by"`
-	Type  string `yaml:"type" json:"type" enum:"MetricEncodeFilterTypeEnum" doc:"the type of filter match: exact (default), presence, absence or regex"`
+	Type  string `yaml:"type" json:"type" enum:"MetricEncodeFilterTypeEnum" doc:"the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex"`
 }
 
 type MetricEncodeFilterTypeEnum struct {
-	Exact    string `yaml:"exact" json:"exact" doc:"match exactly the provided fitler value"`
-	Presence string `yaml:"presence" json:"presence" doc:"filter key must be present (filter value is ignored)"`
-	Absence  string `yaml:"absence" json:"absence" doc:"filter key must be absent (filter value is ignored)"`
-	Regex    string `yaml:"regex" json:"regex" doc:"match filter value as a regular expression"`
+	Equal         string `yaml:"equal" json:"equal" doc:"match exactly the provided filter value"`
+	NotEqual      string `yaml:"not_equal" json:"not_equal" doc:"the value must be different from the provided filter"`
+	Presence      string `yaml:"presence" json:"presence" doc:"filter key must be present (filter value is ignored)"`
+	Absence       string `yaml:"absence" json:"absence" doc:"filter key must be absent (filter value is ignored)"`
+	MatchRegex    string `yaml:"match_regex" json:"match_regex" doc:"match filter value as a regular expression"`
+	NotMatchRegex string `yaml:"not_match_regex" json:"not_match_regex" doc:"the filter value must not match the provided regular expression"`
 }
 
 func MetricEncodeFilterTypeName(t string) string {
diff --git a/pkg/pipeline/encode/encode_prom.go b/pkg/pipeline/encode/encode_prom.go
index 9780e1be..9cdef8e1 100644
--- a/pkg/pipeline/encode/encode_prom.go
+++ b/pkg/pipeline/encode/encode_prom.go
@@ -18,229 +18,82 @@
 package encode
 
 import (
-	"fmt"
-	"strings"
 	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/operational"
-	putils "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
 	promserver "github.com/netobserv/flowlogs-pipeline/pkg/prometheus"
-	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
-
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 )
 
 const defaultExpiryTime = time.Duration(2 * time.Minute)
 
-type gaugeInfo struct {
-	gauge *prometheus.GaugeVec
-	info  *MetricInfo
-}
-
-type counterInfo struct {
-	counter *prometheus.CounterVec
-	info    *MetricInfo
-}
-
-type histoInfo struct {
-	histo *prometheus.HistogramVec
-	info  *MetricInfo
-}
-
 type EncodeProm struct {
-	cfg              *api.PromEncode
-	registerer       prometheus.Registerer
-	gauges           []gaugeInfo
-	counters         []counterInfo
-	histos           []histoInfo
-	aggHistos        []histoInfo
-	expiryTime       time.Duration
-	mCache           *putils.TimedCache
-	mChacheLenMetric prometheus.Gauge
-	exitChan         <-chan struct{}
-	metricsProcessed prometheus.Counter
-	metricsDropped   prometheus.Counter
-	errorsCounter    *prometheus.CounterVec
+	cfg          *api.PromEncode
+	registerer   prometheus.Registerer
+	metricCommon *MetricsCommonStruct
 }
 
-var (
-	MetricsProcessed = operational.DefineMetric(
-		"metrics_processed",
-		"Number of metrics processed",
-		operational.TypeCounter,
-		"stage",
-	)
-	MetricsDropped = operational.DefineMetric(
-		"metrics_dropped",
-		"Number of metrics dropped",
-		operational.TypeCounter,
-		"stage",
-	)
-	EncodePromErrors = operational.DefineMetric(
-		"encode_prom_errors",
-		"Total errors during metrics generation",
-		operational.TypeCounter,
-		"error", "metric", "key",
-	)
-	mChacheLen = operational.DefineMetric(
-		"encode_prom_metrics_reported",
-		"Total number of prometheus metrics reported by this stage",
-		operational.TypeGauge,
-		"stage",
-	)
-)
-
-// Encode encodes a metric before being stored
+// Encode encodes a metric before being stored; the heavy work is done by the MetricCommonEncode
 func (e *EncodeProm) Encode(metricRecord config.GenericMap) {
 	log.Tracef("entering EncodeMetric. metricRecord = %v", metricRecord)
-
-	// Process counters
-	for _, mInfo := range e.counters {
-		labels, value := e.prepareMetric(metricRecord, mInfo.info, mInfo.counter.MetricVec)
-		if labels == nil {
-			continue
-		}
-		m, err := mInfo.counter.GetMetricWith(labels)
-		if err != nil {
-			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
-			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
-			continue
-		}
-		m.Add(value)
-		e.metricsProcessed.Inc()
-	}
-
-	// Process gauges
-	for _, mInfo := range e.gauges {
-		labels, value := e.prepareMetric(metricRecord, mInfo.info, mInfo.gauge.MetricVec)
-		if labels == nil {
-			continue
-		}
-		m, err := mInfo.gauge.GetMetricWith(labels)
-		if err != nil {
-			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
-			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
-			continue
-		}
-		m.Set(value)
-		e.metricsProcessed.Inc()
-	}
-
-	// Process histograms
-	for _, mInfo := range e.histos {
-		labels, value := e.prepareMetric(metricRecord, mInfo.info, mInfo.histo.MetricVec)
-		if labels == nil {
-			continue
-		}
-		m, err := mInfo.histo.GetMetricWith(labels)
-		if err != nil {
-			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
-			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
-			continue
-		}
-		m.Observe(value)
-		e.metricsProcessed.Inc()
-	}
-
-	// Process pre-aggregated histograms
-	for _, mInfo := range e.aggHistos {
-		labels, values := e.prepareAggHisto(metricRecord, mInfo.info, mInfo.histo.MetricVec)
-		if labels == nil {
-			continue
-		}
-		m, err := mInfo.histo.GetMetricWith(labels)
-		if err != nil {
-			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
-			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
-			continue
-		}
-		for _, v := range values {
-			m.Observe(v)
-		}
-		e.metricsProcessed.Inc()
-	}
+	e.metricCommon.MetricCommonEncode(e, metricRecord)
 }
 
-func (e *EncodeProm) prepareMetric(flow config.GenericMap, info *MetricInfo, m *prometheus.MetricVec) (map[string]string, float64) {
-	val := e.extractGenericValue(flow, info)
-	if val == nil {
-		return nil, 0
-	}
-	floatVal, err := utils.ConvertToFloat64(val)
+func (e *EncodeProm) ProcessCounter(m interface{}, labels map[string]string, value float64) error {
+	counter := m.(*prometheus.CounterVec)
+	mm, err := counter.GetMetricWith(labels)
 	if err != nil {
-		e.errorsCounter.WithLabelValues("ValueConversionError", info.Name, info.ValueKey).Inc()
-		return nil, 0
-	}
-	if info.ValueScale != 0 {
-		floatVal = floatVal / info.ValueScale
+		return err
 	}
-
-	entryLabels, key := ExtractLabelsAndKey(flow, &info.MetricsItem)
-	// Update entry for expiry mechanism (the entry itself is its own cleanup function)
-	_, ok := e.mCache.UpdateCacheEntry(key, func() { m.Delete(entryLabels) })
-	if !ok {
-		e.metricsDropped.Inc()
-		return nil, 0
-	}
-	return entryLabels, floatVal
+	mm.Add(value)
+	return nil
 }
 
-func (e *EncodeProm) prepareAggHisto(flow config.GenericMap, info *MetricInfo, m *prometheus.MetricVec) (map[string]string, []float64) {
-	val := e.extractGenericValue(flow, info)
-	if val == nil {
-		return nil, nil
-	}
-	values, ok := val.([]float64)
-	if !ok {
-		e.errorsCounter.WithLabelValues("HistoValueConversionError", info.Name, info.ValueKey).Inc()
-		return nil, nil
+func (e *EncodeProm) ProcessGauge(m interface{}, labels map[string]string, value float64, key string) error {
+	gauge := m.(*prometheus.GaugeVec)
+	mm, err := gauge.GetMetricWith(labels)
+	if err != nil {
+		return err
 	}
+	mm.Set(value)
+	return nil
+}
 
-	entryLabels, key := ExtractLabelsAndKey(flow, &info.MetricsItem)
-	// Update entry for expiry mechanism (the entry itself is its own cleanup function)
-	_, ok = e.mCache.UpdateCacheEntry(key, func() { m.Delete(entryLabels) })
-	if !ok {
-		e.metricsDropped.Inc()
-		return nil, nil
+func (e *EncodeProm) ProcessHist(m interface{}, labels map[string]string, value float64) error {
+	hist := m.(*prometheus.HistogramVec)
+	mm, err := hist.GetMetricWith(labels)
+	if err != nil {
+		return err
 	}
-	return entryLabels, values
+	mm.Observe(value)
+	return nil
 }
 
-func (e *EncodeProm) extractGenericValue(flow config.GenericMap, info *MetricInfo) interface{} {
-	for _, pred := range info.FilterPredicates {
-		if !pred(flow) {
-			return nil
-		}
-	}
-	if info.ValueKey == "" {
-		// No value key means it's a records / flows counter (1 flow = 1 increment), so just return 1
-		return 1
+func (e *EncodeProm) ProcessAggHist(m interface{}, labels map[string]string, values []float64) error {
+	hist := m.(*prometheus.HistogramVec)
+	mm, err := hist.GetMetricWith(labels)
+	if err != nil {
+		return err
 	}
-	val, found := flow[info.ValueKey]
-	if !found {
-		e.errorsCounter.WithLabelValues("RecordKeyMissing", info.Name, info.ValueKey).Inc()
-		return nil
+	for _, v := range values {
+		mm.Observe(v)
 	}
-	return val
+	return nil
 }
 
-func ExtractLabelsAndKey(flow config.GenericMap, info *api.MetricsItem) (map[string]string, string) {
-	entryLabels := make(map[string]string, len(info.Labels))
-	key := strings.Builder{}
-	key.WriteString(info.Name)
-	key.WriteRune('|')
-	for _, t := range info.Labels {
-		entryLabels[t] = ""
-		if v, ok := flow[t]; ok {
-			entryLabels[t] = fmt.Sprintf("%v", v)
-		}
-		key.WriteString(entryLabels[t])
-		key.WriteRune('|')
+func (e *EncodeProm) GetChacheEntry(entryLabels map[string]string, m interface{}) interface{} {
+	switch mv := m.(type) {
+	case *prometheus.CounterVec:
+		return func() { mv.Delete(entryLabels) }
+	case *prometheus.GaugeVec:
+		return func() { mv.Delete(entryLabels) }
+	case *prometheus.HistogramVec:
+		return func() { mv.Delete(entryLabels) }
 	}
-	return entryLabels, key.String()
+	return nil
 }
 
 // callback function from lru cleanup
@@ -248,19 +101,6 @@ func (e *EncodeProm) Cleanup(cleanupFunc interface{}) {
 	cleanupFunc.(func())()
 }
 
-func (e *EncodeProm) cleanupExpiredEntriesLoop() {
-	ticker := time.NewTicker(e.expiryTime)
-	for {
-		select {
-		case <-e.exitChan:
-			log.Debugf("exiting cleanupExpiredEntriesLoop because of signal")
-			return
-		case <-ticker.C:
-			e.mCache.CleanupExpiredEntries(e.expiryTime, e.Cleanup)
-		}
-	}
-}
-
 func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (Encoder, error) {
 	cfg := api.PromEncode{}
 	if params.Encode != nil && params.Encode.Prom != nil {
@@ -282,11 +122,13 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 	} else {
 		registerer = prometheus.DefaultRegisterer
 	}
+	w := &EncodeProm{
+		cfg:        params.Encode.Prom,
+		registerer: registerer,
+	}
 
-	counters := []counterInfo{}
-	gauges := []gaugeInfo{}
-	histos := []histoInfo{}
-	aggHistos := []histoInfo{}
+	metricCommon := NewMetricsCommonStruct(opMetrics, cfg.MaxMetrics, params.Name, expiryTime, w.Cleanup)
+	w.metricCommon = metricCommon
 
 	for _, mCfg := range cfg.Metrics {
 		fullMetricName := cfg.Prefix + mCfg.Name
@@ -302,10 +144,7 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 				log.Errorf("error during prometheus.Register: %v", err)
 				return nil, err
 			}
-			counters = append(counters, counterInfo{
-				counter: counter,
-				info:    mInfo,
-			})
+			metricCommon.AddCounter(counter, mInfo)
 		case api.MetricEncodeOperationName("Gauge"):
 			gauge := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: fullMetricName, Help: ""}, labels)
 			err := registerer.Register(gauge)
@@ -313,10 +152,7 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 				log.Errorf("error during prometheus.Register: %v", err)
 				return nil, err
 			}
-			gauges = append(gauges, gaugeInfo{
-				gauge: gauge,
-				info:  mInfo,
-			})
+			metricCommon.AddGauge(gauge, mInfo)
 		case api.MetricEncodeOperationName("Histogram"):
 			log.Debugf("buckets = %v", mCfg.Buckets)
 			hist := prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: fullMetricName, Help: "", Buckets: mCfg.Buckets}, labels)
@@ -325,10 +161,7 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 				log.Errorf("error during prometheus.Register: %v", err)
 				return nil, err
 			}
-			histos = append(histos, histoInfo{
-				histo: hist,
-				info:  mInfo,
-			})
+			metricCommon.AddHist(hist, mInfo)
 		case api.MetricEncodeOperationName("AggHistogram"):
 			log.Debugf("buckets = %v", mCfg.Buckets)
 			hist := prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: fullMetricName, Help: "", Buckets: mCfg.Buckets}, labels)
@@ -337,38 +170,11 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 				log.Errorf("error during prometheus.Register: %v", err)
 				return nil, err
 			}
-			aggHistos = append(aggHistos, histoInfo{
-				histo: hist,
-				info:  mInfo,
-			})
+			metricCommon.AddAggHist(hist, mInfo)
 		case "default":
 			log.Errorf("invalid metric type = %v, skipping", mCfg.Type)
 			continue
 		}
 	}
-
-	log.Debugf("counters = %v", counters)
-	log.Debugf("gauges = %v", gauges)
-	log.Debugf("histos = %v", histos)
-	log.Debugf("aggHistos = %v", aggHistos)
-
-	mChacheLenMetric := opMetrics.NewGauge(&mChacheLen, params.Name)
-
-	w := &EncodeProm{
-		cfg:              params.Encode.Prom,
-		registerer:       registerer,
-		counters:         counters,
-		gauges:           gauges,
-		histos:           histos,
-		aggHistos:        aggHistos,
-		expiryTime:       expiryTime.Duration,
-		mCache:           putils.NewTimedCache(cfg.MaxMetrics, mChacheLenMetric),
-		mChacheLenMetric: mChacheLenMetric,
-		exitChan:         putils.ExitChannel(),
-		metricsProcessed: opMetrics.NewCounter(&MetricsProcessed, params.Name),
-		metricsDropped:   opMetrics.NewCounter(&MetricsDropped, params.Name),
-		errorsCounter:    opMetrics.NewCounterVec(&EncodePromErrors),
-	}
-	go w.cleanupExpiredEntriesLoop()
 	return w, nil
 }
diff --git a/pkg/pipeline/encode/encode_prom_metric.go b/pkg/pipeline/encode/encode_prom_metric.go
index 15fe7854..618ccd2a 100644
--- a/pkg/pipeline/encode/encode_prom_metric.go
+++ b/pkg/pipeline/encode/encode_prom_metric.go
@@ -3,6 +3,7 @@ package encode
 import (
 	"fmt"
 	"regexp"
+	"strings"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
@@ -10,6 +11,8 @@ import (
 
 type Predicate func(flow config.GenericMap) bool
 
+var variableExtractor, _ = regexp.Compile(`\$\(([^\)]+)\)`)
+
 type MetricInfo struct {
 	api.MetricsItem
 	FilterPredicates []Predicate
@@ -29,20 +32,30 @@ func Absence(filter api.MetricsFilter) Predicate {
 	}
 }
 
-func Exact(filter api.MetricsFilter) Predicate {
+func Equal(filter api.MetricsFilter) Predicate {
+	varLookups := extractVarLookups(filter.Value)
 	return func(flow config.GenericMap) bool {
 		if val, found := flow[filter.Key]; found {
 			sVal, ok := val.(string)
 			if !ok {
 				sVal = fmt.Sprint(val)
 			}
-			return sVal == filter.Value
+			value := filter.Value
+			if len(varLookups) > 0 {
+				value = injectVars(flow, value, varLookups)
+			}
+			return sVal == value
 		}
 		return false
 	}
 }
 
-func regex(filter api.MetricsFilter) Predicate {
+func NotEqual(filter api.MetricsFilter) Predicate {
+	pred := Equal(filter)
+	return func(flow config.GenericMap) bool { return !pred(flow) }
+}
+
+func Regex(filter api.MetricsFilter) Predicate {
 	r, _ := regexp.Compile(filter.Value)
 	return func(flow config.GenericMap) bool {
 		if val, found := flow[filter.Key]; found {
@@ -56,26 +69,60 @@ func regex(filter api.MetricsFilter) Predicate {
 	}
 }
 
+func NotRegex(filter api.MetricsFilter) Predicate {
+	pred := Regex(filter)
+	return func(flow config.GenericMap) bool { return !pred(flow) }
+}
+
 func filterToPredicate(filter api.MetricsFilter) Predicate {
 	switch filter.Type {
-	case api.PromFilterExact:
-		return Exact(filter)
+	case api.PromFilterEqual:
+		return Equal(filter)
+	case api.PromFilterNotEqual:
+		return NotEqual(filter)
 	case api.PromFilterPresence:
 		return Presence(filter)
 	case api.PromFilterAbsence:
 		return Absence(filter)
 	case api.PromFilterRegex:
-		return regex(filter)
+		return Regex(filter)
+	case api.PromFilterNotRegex:
+		return NotRegex(filter)
 	}
 	// Default = Exact
-	return Exact(filter)
+	return Equal(filter)
+}
+
+func extractVarLookups(value string) [][]string {
+	// Extract list of variables to lookup
+	// E.g: filter "$(SrcAddr):$(SrcPort)" would return [SrcAddr,SrcPort]
+	if len(value) > 0 {
+		return variableExtractor.FindAllStringSubmatch(value, -1)
+	}
+	return nil
+}
+
+func injectVars(flow config.GenericMap, filterValue string, varLookups [][]string) string {
+	injected := filterValue
+	for _, matchGroup := range varLookups {
+		var value string
+		if rawVal, found := flow[matchGroup[1]]; found {
+			if sVal, ok := rawVal.(string); ok {
+				value = sVal
+			} else {
+				value = fmt.Sprint(rawVal)
+			}
+		}
+		injected = strings.ReplaceAll(injected, matchGroup[0], value)
+	}
+	return injected
 }
 
 func CreateMetricInfo(def api.MetricsItem) *MetricInfo {
 	mi := MetricInfo{
 		MetricsItem: def,
 	}
-	for _, f := range def.GetFilters() {
+	for _, f := range def.Filters {
 		mi.FilterPredicates = append(mi.FilterPredicates, filterToPredicate(f))
 	}
 	return &mi
diff --git a/pkg/pipeline/encode/metrics_common.go b/pkg/pipeline/encode/metrics_common.go
new file mode 100644
index 00000000..c90a5a2e
--- /dev/null
+++ b/pkg/pipeline/encode/metrics_common.go
@@ -0,0 +1,279 @@
+/*
+ * Copyright (C) 2024 IBM, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package encode
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/netobserv/flowlogs-pipeline/pkg/api"
+	"github.com/netobserv/flowlogs-pipeline/pkg/config"
+	"github.com/netobserv/flowlogs-pipeline/pkg/operational"
+	putils "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
+	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
+	"github.com/prometheus/client_golang/prometheus"
+	log "github.com/sirupsen/logrus"
+)
+
+type mInfoStruct struct {
+	genericMetric interface{} // can be a counter, gauge, or histogram pointer
+	info          *MetricInfo
+}
+
+type MetricsCommonStruct struct {
+	gauges           []mInfoStruct
+	counters         []mInfoStruct
+	histos           []mInfoStruct
+	aggHistos        []mInfoStruct
+	mCache           *putils.TimedCache
+	mChacheLenMetric prometheus.Gauge
+	metricsProcessed prometheus.Counter
+	metricsDropped   prometheus.Counter
+	errorsCounter    *prometheus.CounterVec
+	expiryTime       time.Duration
+	exitChan         <-chan struct{}
+}
+
+type MetricsCommonInterface interface {
+	GetChacheEntry(entryLabels map[string]string, m interface{}) interface{}
+	ProcessCounter(m interface{}, labels map[string]string, value float64) error
+	ProcessGauge(m interface{}, labels map[string]string, value float64, key string) error
+	ProcessHist(m interface{}, labels map[string]string, value float64) error
+	ProcessAggHist(m interface{}, labels map[string]string, value []float64) error
+}
+
+var (
+	metricsProcessed = operational.DefineMetric(
+		"metrics_processed",
+		"Number of metrics processed",
+		operational.TypeCounter,
+		"stage",
+	)
+	metricsDropped = operational.DefineMetric(
+		"metrics_dropped",
+		"Number of metrics dropped",
+		operational.TypeCounter,
+		"stage",
+	)
+	encodePromErrors = operational.DefineMetric(
+		"encode_prom_errors",
+		"Total errors during metrics generation",
+		operational.TypeCounter,
+		"error", "metric", "key",
+	)
+	mChacheLen = operational.DefineMetric(
+		"encode_prom_metrics_reported",
+		"Total number of prometheus metrics reported by this stage",
+		operational.TypeGauge,
+		"stage",
+	)
+)
+
+func (m *MetricsCommonStruct) AddCounter(g interface{}, info *MetricInfo) {
+	mStruct := mInfoStruct{genericMetric: g, info: info}
+	m.counters = append(m.counters, mStruct)
+}
+
+func (m *MetricsCommonStruct) AddGauge(g interface{}, info *MetricInfo) {
+	mStruct := mInfoStruct{genericMetric: g, info: info}
+	m.gauges = append(m.gauges, mStruct)
+}
+
+func (m *MetricsCommonStruct) AddHist(g interface{}, info *MetricInfo) {
+	mStruct := mInfoStruct{genericMetric: g, info: info}
+	m.histos = append(m.histos, mStruct)
+}
+
+func (m *MetricsCommonStruct) AddAggHist(g interface{}, info *MetricInfo) {
+	mStruct := mInfoStruct{genericMetric: g, info: info}
+	m.aggHistos = append(m.aggHistos, mStruct)
+}
+
+func (e *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, metricRecord config.GenericMap) {
+	log.Tracef("entering MetricCommonEncode. metricRecord = %v", metricRecord)
+
+	// Process counters
+	for _, mInfo := range e.counters {
+		labels, value, _ := e.prepareMetric(mci, metricRecord, mInfo.info, mInfo.genericMetric)
+		if labels == nil {
+			continue
+		}
+		err := mci.ProcessCounter(mInfo.genericMetric, labels, value)
+		if err != nil {
+			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
+			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
+			continue
+		}
+		e.metricsProcessed.Inc()
+	}
+
+	// Process gauges
+	for _, mInfo := range e.gauges {
+		labels, value, key := e.prepareMetric(mci, metricRecord, mInfo.info, mInfo.genericMetric)
+		if labels == nil {
+			continue
+		}
+		err := mci.ProcessGauge(mInfo.genericMetric, labels, value, key)
+		if err != nil {
+			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
+			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
+			continue
+		}
+		e.metricsProcessed.Inc()
+	}
+
+	// Process histograms
+	for _, mInfo := range e.histos {
+		labels, value, _ := e.prepareMetric(mci, metricRecord, mInfo.info, mInfo.genericMetric)
+		if labels == nil {
+			continue
+		}
+		err := mci.ProcessHist(mInfo.genericMetric, labels, value)
+		if err != nil {
+			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
+			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
+			continue
+		}
+		e.metricsProcessed.Inc()
+	}
+
+	// Process pre-aggregated histograms
+	for _, mInfo := range e.aggHistos {
+		labels, values := e.prepareAggHisto(mci, metricRecord, mInfo.info, mInfo.genericMetric)
+		if labels == nil {
+			continue
+		}
+		err := mci.ProcessAggHist(mInfo.genericMetric, labels, values)
+		if err != nil {
+			log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
+			e.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
+			continue
+		}
+		e.metricsProcessed.Inc()
+	}
+}
+
+func (m *MetricsCommonStruct) prepareMetric(mci MetricsCommonInterface, flow config.GenericMap, info *MetricInfo, mv interface{}) (map[string]string, float64, string) {
+	val := m.extractGenericValue(flow, info)
+	if val == nil {
+		return nil, 0, ""
+	}
+	floatVal, err := utils.ConvertToFloat64(val)
+	if err != nil {
+		m.errorsCounter.WithLabelValues("ValueConversionError", info.Name, info.ValueKey).Inc()
+		return nil, 0, ""
+	}
+	if info.ValueScale != 0 {
+		floatVal = floatVal / info.ValueScale
+	}
+
+	entryLabels, key := extractLabelsAndKey(flow, &info.MetricsItem)
+	// Update entry for expiry mechanism (the entry itself is its own cleanup function)
+	cacheEntry := mci.GetChacheEntry(entryLabels, mv)
+	_, ok := m.mCache.UpdateCacheEntry(key, cacheEntry)
+	if !ok {
+		m.metricsDropped.Inc()
+		return nil, 0, ""
+	}
+	return entryLabels, floatVal, key
+}
+
+func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow config.GenericMap, info *MetricInfo, mc interface{}) (map[string]string, []float64) {
+	val := m.extractGenericValue(flow, info)
+	if val == nil {
+		return nil, nil
+	}
+	values, ok := val.([]float64)
+	if !ok {
+		m.errorsCounter.WithLabelValues("HistoValueConversionError", info.Name, info.ValueKey).Inc()
+		return nil, nil
+	}
+
+	entryLabels, key := extractLabelsAndKey(flow, &info.MetricsItem)
+	// Update entry for expiry mechanism (the entry itself is its own cleanup function)
+	cacheEntry := mci.GetChacheEntry(entryLabels, mc)
+	_, ok = m.mCache.UpdateCacheEntry(key, cacheEntry)
+	if !ok {
+		m.metricsDropped.Inc()
+		return nil, nil
+	}
+	return entryLabels, values
+}
+
+func (m *MetricsCommonStruct) extractGenericValue(flow config.GenericMap, info *MetricInfo) interface{} {
+	for _, pred := range info.FilterPredicates {
+		if !pred(flow) {
+			return nil
+		}
+	}
+	if info.ValueKey == "" {
+		// No value key means it's a records / flows counter (1 flow = 1 increment), so just return 1
+		return 1
+	}
+	val, found := flow[info.ValueKey]
+	if !found {
+		m.errorsCounter.WithLabelValues("RecordKeyMissing", info.Name, info.ValueKey).Inc()
+		return nil
+	}
+	return val
+}
+
+func extractLabelsAndKey(flow config.GenericMap, info *api.MetricsItem) (map[string]string, string) {
+	entryLabels := make(map[string]string, len(info.Labels))
+	key := strings.Builder{}
+	key.WriteString(info.Name)
+	key.WriteRune('|')
+	for _, t := range info.Labels {
+		entryLabels[t] = ""
+		if v, ok := flow[t]; ok {
+			entryLabels[t] = fmt.Sprintf("%v", v)
+		}
+		key.WriteString(entryLabels[t])
+		key.WriteRune('|')
+	}
+	return entryLabels, key.String()
+}
+
+func (m *MetricsCommonStruct) cleanupExpiredEntriesLoop(callback putils.CacheCallback) {
+	ticker := time.NewTicker(m.expiryTime)
+	for {
+		select {
+		case <-m.exitChan:
+			log.Debugf("exiting cleanupExpiredEntriesLoop because of signal")
+			return
+		case <-ticker.C:
+			m.mCache.CleanupExpiredEntries(m.expiryTime, callback)
+		}
+	}
+}
+
+func NewMetricsCommonStruct(opMetrics *operational.Metrics, maxCacheEntries int, name string, expiryTime api.Duration, callback putils.CacheCallback) *MetricsCommonStruct {
+	mChacheLenMetric := opMetrics.NewGauge(&mChacheLen, name)
+	m := &MetricsCommonStruct{
+		mCache:           putils.NewTimedCache(maxCacheEntries, mChacheLenMetric),
+		mChacheLenMetric: mChacheLenMetric,
+		metricsProcessed: opMetrics.NewCounter(&metricsProcessed, name),
+		metricsDropped:   opMetrics.NewCounter(&metricsDropped, name),
+		errorsCounter:    opMetrics.NewCounterVec(&encodePromErrors),
+		expiryTime:       expiryTime.Duration,
+		exitChan:         putils.ExitChannel(),
+	}
+	go m.cleanupExpiredEntriesLoop(callback)
+	return m
+}
diff --git a/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go b/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go
index 739a7d4e..c645bd53 100644
--- a/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go
+++ b/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go
@@ -25,9 +25,6 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/operational"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/encode"
-	putils "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
-	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
-	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -36,116 +33,60 @@ import (
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
-// TODO: Refactor the code that is common with encode_prom
-const (
-	defaultExpiryTime = time.Duration(2 * time.Minute)
-	flpMeterName      = "flp_meter"
-)
-
-type counterInfo struct {
-	counter *metric.Float64Counter
-	info    *encode.MetricInfo
-}
-
-type gaugeInfo struct {
-	gauge *metric.Float64ObservableGauge
-	info  *encode.MetricInfo
-	obs   Float64Gauge
-}
-
-// TBD: Handle histograms
-/*
-type histoInfo struct {
-	histo *metric.Float64Histogram
-	info  *encode.MetricInfo
-}
-*/
+const defaultExpiryTime = time.Duration(2 * time.Minute)
+const flpMeterName = "flp_meter"
 
 type EncodeOtlpMetrics struct {
-	cfg      api.EncodeOtlpMetrics
-	ctx      context.Context
-	res      *resource.Resource
-	mp       *sdkmetric.MeterProvider
-	counters []counterInfo
-	gauges   []gaugeInfo
-	//histos           []histoInfo
-	//aggHistos        []histoInfo
-	expiryTime       time.Duration
-	mCache           *putils.TimedCache
-	exitChan         <-chan struct{}
-	meter            metric.Meter
-	metricsProcessed prometheus.Counter
-	metricsDropped   prometheus.Counter
-	errorsCounter    *prometheus.CounterVec
+	cfg          api.EncodeOtlpMetrics
+	ctx          context.Context
+	res          *resource.Resource
+	mp           *sdkmetric.MeterProvider
+	meter        metric.Meter
+	metricCommon *encode.MetricsCommonStruct
 }
 
 // Encode encodes a metric to be exported
 func (e *EncodeOtlpMetrics) Encode(metricRecord config.GenericMap) {
 	log.Tracef("entering EncodeOtlpMetrics. entry = %v", metricRecord)
+	e.metricCommon.MetricCommonEncode(e, metricRecord)
+}
 
-	// Process counters
-	for _, mInfo := range e.counters {
-		labels, value, _ := e.prepareMetric(metricRecord, mInfo.info)
-		if labels == nil {
-			continue
-		}
-		// set attributes using the labels
-		attributes := obtainAttributesFromLabels(labels)
-		(*mInfo.counter).Add(e.ctx, value, metric.WithAttributes(attributes...))
-		e.metricsProcessed.Inc()
-	}
+func (e *EncodeOtlpMetrics) ProcessCounter(m interface{}, labels map[string]string, value float64) error {
+	counter := m.(metric.Float64Counter)
+	// set attributes using the labels
+	attributes := obtainAttributesFromLabels(labels)
+	counter.Add(e.ctx, value, metric.WithAttributes(attributes...))
+	return nil
+}
 
-	// Process gauges
-	for _, mInfo := range e.gauges {
-		labels, value, key := e.prepareMetric(metricRecord, mInfo.info)
-		if labels == nil {
-			continue
-		}
-		// set attributes using the labels
-		attributes := obtainAttributesFromLabels(labels)
-		mInfo.obs.Set(key, value, attributes)
-		e.metricsProcessed.Inc()
-	}
-	// TBD: Process histograms
+func (e *EncodeOtlpMetrics) ProcessGauge(m interface{}, labels map[string]string, value float64, key string) error {
+	obs := m.(Float64Gauge)
+	// set attributes using the labels
+	attributes := obtainAttributesFromLabels(labels)
+	obs.Set(key, value, attributes)
+	return nil
 }
 
-func (e *EncodeOtlpMetrics) prepareMetric(flow config.GenericMap, info *encode.MetricInfo) (map[string]string, float64, string) {
-	val := e.extractGenericValue(flow, info)
-	if val == nil {
-		return nil, 0, ""
-	}
-	floatVal, err := utils.ConvertToFloat64(val)
-	if err != nil {
-		e.errorsCounter.WithLabelValues("ValueConversionError", info.Name, info.ValueKey).Inc()
-		return nil, 0, ""
-	}
+func (e *EncodeOtlpMetrics) ProcessHist(m interface{}, labels map[string]string, value float64) error {
+	histo := m.(metric.Float64Histogram)
+	// set attributes using the labels
+	attributes := obtainAttributesFromLabels(labels)
+	histo.Record(e.ctx, value, metric.WithAttributes(attributes...))
+	return nil
+}
 
-	entryLabels, key := encode.ExtractLabelsAndKey(flow, &info.MetricsItem)
-	// Update entry for expiry mechanism (the entry itself is its own cleanup function)
-	_, ok := e.mCache.UpdateCacheEntry(key, entryLabels)
-	if !ok {
-		e.metricsDropped.Inc()
-		return nil, 0, ""
+func (e *EncodeOtlpMetrics) ProcessAggHist(m interface{}, labels map[string]string, values []float64) error {
+	histo := m.(metric.Float64Histogram)
+	// set attributes using the labels
+	attributes := obtainAttributesFromLabels(labels)
+	for _, v := range values {
+		histo.Record(e.ctx, v, metric.WithAttributes(attributes...))
 	}
-	return entryLabels, floatVal, key
+	return nil
 }
 
-func (e *EncodeOtlpMetrics) extractGenericValue(flow config.GenericMap, info *encode.MetricInfo) interface{} {
-	for _, pred := range info.FilterPredicates {
-		if !pred(flow) {
-			return nil
-		}
-	}
-	if info.ValueKey == "" {
-		// No value key means it's a records / flows counter (1 flow = 1 increment), so just return 1
-		return 1
-	}
-	val, found := flow[info.ValueKey]
-	if !found {
-		e.errorsCounter.WithLabelValues("RecordKeyMissing", info.Name, info.ValueKey).Inc()
-		return nil
-	}
-	return val
+func (e *EncodeOtlpMetrics) GetChacheEntry(entryLabels map[string]string, m interface{}) interface{} {
+	return entryLabels
 }
 
 func NewEncodeOtlpMetrics(opMetrics *operational.Metrics, params config.StageParam) (encode.Encoder, error) {
@@ -173,8 +114,17 @@ func NewEncodeOtlpMetrics(opMetrics *operational.Metrics, params config.StagePar
 	}
 
 	meterFactory := otel.Meter(flpMeterName)
-	counters := []counterInfo{}
-	gauges := []gaugeInfo{}
+
+	w := &EncodeOtlpMetrics{
+		cfg:   cfg,
+		ctx:   ctx,
+		res:   res,
+		mp:    mp,
+		meter: meterFactory,
+	}
+
+	metricCommon := encode.NewMetricsCommonStruct(opMetrics, 0, params.Name, expiryTime, nil)
+	w.metricCommon = metricCommon
 
 	for _, mCfg := range cfg.Metrics {
 		fullMetricName := cfg.Prefix + mCfg.Name
@@ -189,10 +139,7 @@ func NewEncodeOtlpMetrics(opMetrics *operational.Metrics, params config.StagePar
 				log.Errorf("error during counter creation: %v", err)
 				return nil, err
 			}
-			counters = append(counters, counterInfo{
-				counter: &counter,
-				info:    mInfo,
-			})
+			metricCommon.AddCounter(counter, mInfo)
 		case api.MetricEncodeOperationName("Gauge"):
 			// at implementation time, only asynchronous gauges are supported by otel in golang
 			obs := Float64Gauge{observations: make(map[string]Float64GaugeEntry)}
@@ -204,56 +151,31 @@ func NewEncodeOtlpMetrics(opMetrics *operational.Metrics, params config.StagePar
 				log.Errorf("error during gauge creation: %v", err)
 				return nil, err
 			}
-			gInfo := gaugeInfo{
-				info:  mInfo,
-				obs:   obs,
-				gauge: &gauge,
+			metricCommon.AddGauge(gauge, mInfo)
+		case api.MetricEncodeOperationName("Histogram"):
+			var histo metric.Float64Histogram
+			if len(mCfg.Buckets) == 0 {
+				histo, err = meter.Float64Histogram(fullMetricName)
+			} else {
+				histo, err = meter.Float64Histogram(fullMetricName,
+					metric.WithExplicitBucketBoundaries(mCfg.Buckets...),
+				)
+
+			}
+			if err != nil {
+				log.Errorf("error during histogram creation: %v", err)
+				return nil, err
 			}
-			gauges = append(gauges, gInfo)
-		// TBD: handle histograms
+			metricCommon.AddHist(histo, mInfo)
 		case "default":
 			log.Errorf("invalid metric type = %v, skipping", mCfg.Type)
 			continue
 		}
 	}
 
-	w := &EncodeOtlpMetrics{
-		cfg:              cfg,
-		ctx:              ctx,
-		res:              res,
-		mp:               mp,
-		meter:            meterFactory,
-		counters:         counters,
-		gauges:           gauges,
-		expiryTime:       expiryTime.Duration,
-		mCache:           putils.NewTimedCache(0, nil),
-		exitChan:         putils.ExitChannel(),
-		metricsProcessed: opMetrics.NewCounter(&encode.MetricsProcessed, params.Name),
-		metricsDropped:   opMetrics.NewCounter(&encode.MetricsDropped, params.Name),
-		errorsCounter:    opMetrics.NewCounterVec(&encode.EncodePromErrors),
-	}
-	go w.cleanupExpiredEntriesLoop()
 	return w, nil
 }
 
-// Cleanup - callback function from lru cleanup
-func (e *EncodeOtlpMetrics) Cleanup(cleanupFunc interface{}) {
-	// nothing more to do
-}
-
-func (e *EncodeOtlpMetrics) cleanupExpiredEntriesLoop() {
-	ticker := time.NewTicker(e.expiryTime)
-	for {
-		select {
-		case <-e.exitChan:
-			log.Debugf("exiting cleanupExpiredEntriesLoop because of signal")
-			return
-		case <-ticker.C:
-			e.mCache.CleanupExpiredEntries(e.expiryTime, e.Cleanup)
-		}
-	}
-}
-
 // At present, golang only supports asynchronous gauge, so we have some function here to support this
 
 type Float64GaugeEntry struct {
diff --git a/pkg/pipeline/write/write_ipfix.go b/pkg/pipeline/write/write_ipfix.go
index 5cf78f8b..e76bb9e1 100644
--- a/pkg/pipeline/write/write_ipfix.go
+++ b/pkg/pipeline/write/write_ipfix.go
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2022 IBM, Inc.
+ * Copyright (C) 2024 IBM, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -77,7 +77,7 @@ var (
 		"destinationNodeName",
 	}
 	CustomNetworkFields = []string{
-		// TODO
+		"timeFlowRttNs",
 	}
 )
 
@@ -150,6 +150,11 @@ func loadCustomRegistry(EnterpriseID uint32) error {
 		ilog.WithError(err).Errorf("Failed to register element")
 		return err
 	}
+	err = registry.PutInfoElement((*entities.NewInfoElement("timeFlowRttNs", 7740, entities.Unsigned64, EnterpriseID, 8)), EnterpriseID)
+	if err != nil {
+		ilog.WithError(err).Errorf("Failed to register element")
+		return err
+	}
 	return nil
 }
 
@@ -336,6 +341,12 @@ func setStandardIEValue(record config.GenericMap, ieValPtr *entities.InfoElement
 		} 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")
+		}
 	}
 	return nil
 }


Test patch:
diff --git a/cmd/flowlogs-pipeline/main_test.go b/cmd/flowlogs-pipeline/main_test.go
index fca40f8e..b9f45082 100644
--- a/cmd/flowlogs-pipeline/main_test.go
+++ b/cmd/flowlogs-pipeline/main_test.go
@@ -50,7 +50,7 @@ func TestPipelineConfigSetup(t *testing.T) {
 
 	js := `{
     "PipeLine": "[{\"name\":\"grpc\"},{\"follows\":\"grpc\",\"name\":\"enrich\"},{\"follows\":\"enrich\",\"name\":\"loki\"},{\"follows\":\"enrich\",\"name\":\"prometheus\"}]",
-    "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"input\":\"SrcAddr\",\"output\":\"SrcK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstAddr\",\"output\":\"DstK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstPort\",\"output\":\"Service\",\"parameters\":\"Proto\",\"type\":\"add_service\"},{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"parameters\":\"/16\",\"type\":\"add_subnet\"}]},\"type\":\"network\"}},{\"name\":\"loki\",\"write\":{\"loki\":{\"batchSize\":102400,\"batchWait\":\"1s\",\"clientConfig\":{\"follow_redirects\":false,\"proxy_url\":null,\"tls_config\":{\"insecure_skip_verify\":false}},\"labels\":[\"SrcK8S_Namespace\",\"SrcK8S_OwnerName\",\"DstK8S_Namespace\",\"DstK8S_OwnerName\",\"FlowDirection\"],\"maxBackoff\":\"5m0s\",\"maxRetries\":10,\"minBackoff\":\"1s\",\"staticLabels\":{\"app\":\"netobserv-flowcollector\"},\"tenantID\":\"netobserv\",\"timeout\":\"10s\",\"timestampLabel\":\"TimeFlowEndMs\",\"timestampScale\":\"1ms\",\"url\":\"http://loki.netobserv.svc:3100/\"},\"type\":\"loki\"}},{\"encode\":{\"prom\":{\"metrics\":[{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
+    "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"input\":\"SrcAddr\",\"output\":\"SrcK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstAddr\",\"output\":\"DstK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstPort\",\"output\":\"Service\",\"parameters\":\"Proto\",\"type\":\"add_service\"},{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"parameters\":\"/16\",\"type\":\"add_subnet\"}]},\"type\":\"network\"}},{\"name\":\"loki\",\"write\":{\"loki\":{\"batchSize\":102400,\"batchWait\":\"1s\",\"clientConfig\":{\"follow_redirects\":false,\"proxy_url\":null,\"tls_config\":{\"insecure_skip_verify\":false}},\"labels\":[\"SrcK8S_Namespace\",\"SrcK8S_OwnerName\",\"DstK8S_Namespace\",\"DstK8S_OwnerName\",\"FlowDirection\"],\"maxBackoff\":\"5m0s\",\"maxRetries\":10,\"minBackoff\":\"1s\",\"staticLabels\":{\"app\":\"netobserv-flowcollector\"},\"tenantID\":\"netobserv\",\"timeout\":\"10s\",\"timestampLabel\":\"TimeFlowEndMs\",\"timestampScale\":\"1ms\",\"url\":\"http://loki.netobserv.svc:3100/\"},\"type\":\"loki\"}},{\"encode\":{\"prom\":{\"metrics\":[{\"buckets\":null,\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
     "Health": {
         "Port": "8080"
     },
diff --git a/pkg/config/pipeline_builder_test.go b/pkg/config/pipeline_builder_test.go
index d2ef9ebe..d5670de4 100644
--- a/pkg/config/pipeline_builder_test.go
+++ b/pkg/config/pipeline_builder_test.go
@@ -166,7 +166,7 @@ func TestKafkaPromPipeline(t *testing.T) {
 
 	b, err = json.Marshal(params[4])
 	require.NoError(t, err)
-	require.JSONEq(t, `{"name":"prom","encode":{"type":"prom","prom":{"expiryTime":"50s", "metrics":[{"name":"connections_per_source_as","type":"counter","filter":{"key":"","type":"","value":""},"filters":[{"key":"name","type":"","value":"src_as_connection_count"}],"valueKey":"recent_count","valueScale":0,"labels":["by","aggregate"],"buckets":[]}],"prefix":"flp_"}}}`, string(b))
+	require.JSONEq(t, `{"name":"prom","encode":{"type":"prom","prom":{"expiryTime":"50s", "metrics":[{"name":"connections_per_source_as","type":"counter","filters":[{"key":"name","type":"","value":"src_as_connection_count"}],"valueKey":"recent_count","valueScale":0,"labels":["by","aggregate"],"buckets":[]}],"prefix":"flp_"}}}`, string(b))
 }
 
 func TestForkPipeline(t *testing.T) {
diff --git a/pkg/pipeline/aggregate_prom_test.go b/pkg/pipeline/aggregate_prom_test.go
index 6c820682..776fc3c8 100644
--- a/pkg/pipeline/aggregate_prom_test.go
+++ b/pkg/pipeline/aggregate_prom_test.go
@@ -68,19 +68,19 @@ parameters:
        metrics:
          - name: flow_count
            type: counter
-           filter: {key: name, value: bandwidth_count}
+           filters: [{key: name, value: bandwidth_count}]
            valueKey: recent_count
            labels:
              - service
          - name: bytes_sum
            type: counter
-           filter: {key: name, value: bandwidth_sum}
+           filters: [{key: name, value: bandwidth_sum}]
            valueKey: recent_op_value
            labels:
              - service
          - name: bytes_histogram
            type: agg_histogram
-           filter: {key: name, value: bandwidth_raw_values}
+           filters: [{key: name, value: bandwidth_raw_values}]
            valueKey: recent_raw_values
            labels:
              - service
diff --git a/pkg/pipeline/encode/encode_prom_test.go b/pkg/pipeline/encode/encode_prom_test.go
index 6a2f81a8..0820a81b 100644
--- a/pkg/pipeline/encode/encode_prom_test.go
+++ b/pkg/pipeline/encode/encode_prom_test.go
@@ -46,7 +46,7 @@ parameters:
         metrics:
           - name: Bytes
             type: gauge
-            filter: {key: dstAddr, value: 10.1.2.4}
+            filters: [{key: dstAddr, value: 10.1.2.4}]
             valueKey: bytes
             labels:
               - srcAddr
@@ -54,7 +54,7 @@ parameters:
               - srcPort
           - name: Packets
             type: counter
-            filter: {key: dstAddr, value: 10.1.2.4}
+            filters: [{key: dstAddr, value: 10.1.2.4}]
             valueKey: packets
             labels:
               - srcAddr
@@ -93,30 +93,30 @@ func Test_NewEncodeProm(t *testing.T) {
 	encodeProm, err := initProm(cfg.Parameters[0].Encode.Prom)
 	require.NoError(t, err)
 
-	require.Equal(t, 1, len(encodeProm.counters))
-	require.Equal(t, 1, len(encodeProm.gauges))
-	require.Equal(t, 1, len(encodeProm.histos))
-	require.Equal(t, 1, len(encodeProm.aggHistos))
-	require.Equal(t, time.Second, encodeProm.expiryTime)
+	require.Equal(t, 1, len(encodeProm.metricCommon.counters))
+	require.Equal(t, 1, len(encodeProm.metricCommon.gauges))
+	require.Equal(t, 1, len(encodeProm.metricCommon.histos))
+	require.Equal(t, 1, len(encodeProm.metricCommon.aggHistos))
+	require.Equal(t, time.Second, encodeProm.metricCommon.expiryTime)
 
-	require.Equal(t, encodeProm.gauges[0].info.Name, "Bytes")
+	require.Equal(t, encodeProm.metricCommon.gauges[0].info.Name, "Bytes")
 	expectedList := []string{"srcAddr", "dstAddr", "srcPort"}
-	require.Equal(t, encodeProm.gauges[0].info.Labels, expectedList)
+	require.Equal(t, encodeProm.metricCommon.gauges[0].info.Labels, expectedList)
 
-	require.Equal(t, encodeProm.counters[0].info.Name, "Packets")
+	require.Equal(t, encodeProm.metricCommon.counters[0].info.Name, "Packets")
 	expectedList = []string{"srcAddr", "dstAddr", "dstPort"}
-	require.Equal(t, encodeProm.counters[0].info.Labels, expectedList)
+	require.Equal(t, encodeProm.metricCommon.counters[0].info.Labels, expectedList)
 	entry := test.GetExtractMockEntry()
 	encodeProm.Encode(entry)
 
 	// verify entries are in cache; one for the gauge and one for the counter
-	entriesMapLen := encodeProm.mCache.GetCacheLen()
+	entriesMapLen := encodeProm.metricCommon.mCache.GetCacheLen()
 	require.Equal(t, 2, entriesMapLen)
 
 	// wait a couple seconds so that the entry will expire
 	time.Sleep(2 * time.Second)
-	encodeProm.mCache.CleanupExpiredEntries(encodeProm.expiryTime, encodeProm.Cleanup)
-	entriesMapLen = encodeProm.mCache.GetCacheLen()
+	encodeProm.metricCommon.mCache.CleanupExpiredEntries(encodeProm.metricCommon.expiryTime, encodeProm.Cleanup)
+	entriesMapLen = encodeProm.metricCommon.mCache.GetCacheLen()
 	require.Equal(t, 0, entriesMapLen)
 }
 
@@ -361,7 +361,7 @@ func Test_FilterDirection(t *testing.T) {
 				Name:     "ingress_or_inner_packets_total",
 				Type:     "counter",
 				ValueKey: "packets",
-				Filters:  []api.MetricsFilter{{Key: "dir", Value: "0|2", Type: "regex"}},
+				Filters:  []api.MetricsFilter{{Key: "dir", Value: "0|2", Type: "match_regex"}},
 			},
 		},
 	}
@@ -380,6 +380,63 @@ func Test_FilterDirection(t *testing.T) {
 	require.Contains(t, exposed, `test_ingress_or_inner_packets_total 1010`)
 }
 
+func Test_FilterSameOrDifferentNamespace(t *testing.T) {
+	metrics := []config.GenericMap{
+		{
+			"src-ns":  "a",
+			"dst-ns":  "b",
+			"packets": 10,
+		},
+		{
+			"src-ns":  "b",
+			"dst-ns":  "a",
+			"packets": 200,
+		},
+		{
+			"src-ns":  "a",
+			"dst-ns":  "a",
+			"packets": 3000,
+		},
+		{
+			"src-ns":  "b",
+			"dst-ns":  "b",
+			"packets": 40000,
+		},
+	}
+	params := api.PromEncode{
+		Prefix: "test_",
+		ExpiryTime: api.Duration{
+			Duration: time.Duration(60 * time.Second),
+		},
+		Metrics: []api.MetricsItem{
+			{
+				Name:     "packets_same_namespace_total",
+				Type:     "counter",
+				ValueKey: "packets",
+				Filters:  []api.MetricsFilter{{Key: "src-ns", Value: "$(dst-ns)"}},
+			},
+			{
+				Name:     "packets_different_namespace_total",
+				Type:     "counter",
+				ValueKey: "packets",
+				Filters:  []api.MetricsFilter{{Key: "src-ns", Value: "$(dst-ns)", Type: "not_equal"}},
+			},
+		},
+	}
+
+	encodeProm, err := initProm(&params)
+	require.NoError(t, err)
+	for _, metric := range metrics {
+		encodeProm.Encode(metric)
+	}
+	time.Sleep(100 * time.Millisecond)
+
+	exposed := test.ReadExposedMetrics(t)
+
+	require.Contains(t, exposed, `test_packets_same_namespace_total 43000`)
+	require.Contains(t, exposed, `test_packets_different_namespace_total 210`)
+}
+
 func Test_ValueScale(t *testing.T) {
 	metrics := []config.GenericMap{{"rtt": 15_000_000} /*15ms*/, {"rtt": 110_000_000} /*110ms*/}
 	params := api.PromEncode{
@@ -631,18 +688,18 @@ func Test_MultipleProm(t *testing.T) {
 	encodeProm3, err := initProm(cfg.Parameters[2].Encode.Prom)
 	require.NoError(t, err)
 
-	require.Equal(t, 0, len(encodeProm1.counters))
-	require.Equal(t, 1, len(encodeProm1.gauges))
-	require.Equal(t, 0, len(encodeProm1.histos))
-	require.Equal(t, 0, len(encodeProm1.aggHistos))
-	require.Equal(t, 1, len(encodeProm2.counters))
-	require.Equal(t, 0, len(encodeProm2.gauges))
-	require.Equal(t, 0, len(encodeProm2.histos))
-	require.Equal(t, 0, len(encodeProm2.aggHistos))
-	require.Equal(t, 0, len(encodeProm3.counters))
-	require.Equal(t, 0, len(encodeProm3.gauges))
-	require.Equal(t, 1, len(encodeProm3.histos))
-	require.Equal(t, 0, len(encodeProm3.aggHistos))
+	require.Equal(t, 0, len(encodeProm1.metricCommon.counters))
+	require.Equal(t, 1, len(encodeProm1.metricCommon.gauges))
+	require.Equal(t, 0, len(encodeProm1.metricCommon.histos))
+	require.Equal(t, 0, len(encodeProm1.metricCommon.aggHistos))
+	require.Equal(t, 1, len(encodeProm2.metricCommon.counters))
+	require.Equal(t, 0, len(encodeProm2.metricCommon.gauges))
+	require.Equal(t, 0, len(encodeProm2.metricCommon.histos))
+	require.Equal(t, 0, len(encodeProm2.metricCommon.aggHistos))
+	require.Equal(t, 0, len(encodeProm3.metricCommon.counters))
+	require.Equal(t, 0, len(encodeProm3.metricCommon.gauges))
+	require.Equal(t, 1, len(encodeProm3.metricCommon.histos))
+	require.Equal(t, 0, len(encodeProm3.metricCommon.aggHistos))
 
 	require.Equal(t, "test1_", encodeProm1.cfg.Prefix)
 	require.Equal(t, "test2_", encodeProm2.cfg.Prefix)
@@ -650,3 +707,12 @@ func Test_MultipleProm(t *testing.T) {
 
 	// TODO: Add test for different addresses, but need to deal with StartPromServer (ListenAndServe)
 }
+
+func Test_Filters_extractVarLookups(t *testing.T) {
+	variables := extractVarLookups("$(abc)--$(def)")
+
+	require.Equal(t, [][]string{{"$(abc)", "abc"}, {"$(def)", "def"}}, variables)
+
+	variables = extractVarLookups("")
+	require.Empty(t, variables)
+}
diff --git a/pkg/pipeline/encode/opentelemetry/encode_otlp_test.go b/pkg/pipeline/encode/opentelemetry/encode_otlp_test.go
index 5d882ebe..1864f328 100644
--- a/pkg/pipeline/encode/opentelemetry/encode_otlp_test.go
+++ b/pkg/pipeline/encode/opentelemetry/encode_otlp_test.go
@@ -50,10 +50,10 @@ type fakeOltpLoggerProvider struct {
 type fakeOltpLogger struct {
 }
 
-var receivedData []logs.LogRecord
+var otlpReceivedData []logs.LogRecord
 
 func (f *fakeOltpLogger) Emit(msg logs.LogRecord) {
-	receivedData = append(receivedData, msg)
+	otlpReceivedData = append(otlpReceivedData, msg)
 }
 
 func (f *fakeOltpLoggerProvider) Logger(name string, options ...logs.LoggerOption) logs.Logger {
@@ -61,7 +61,7 @@ func (f *fakeOltpLoggerProvider) Logger(name string, options ...logs.LoggerOptio
 }
 
 func initNewEncodeOtlpLogs(t *testing.T) encode.Encoder {
-	receivedData = []logs.LogRecord{}
+	otlpReceivedData = []logs.LogRecord{}
 	v, cfg := test.InitConfig(t, testOtlpConfig)
 	require.NotNil(t, v)
 
@@ -88,14 +88,14 @@ func Test_EncodeOtlpLogs(t *testing.T) {
 	newEncode.Encode(entry1)
 	newEncode.Encode(entry2)
 	// verify contents of the output
-	require.Len(t, receivedData, 2)
+	require.Len(t, otlpReceivedData, 2)
 	expected1, _ := json.Marshal(entry1)
 	expected2, _ := json.Marshal(entry2)
-	require.Equal(t, string(expected1), *receivedData[0].Body())
-	require.Equal(t, string(expected2), *receivedData[1].Body())
+	require.Equal(t, string(expected1), *otlpReceivedData[0].Body())
+	require.Equal(t, string(expected2), *otlpReceivedData[1].Body())
 }
 
-func Test_TLSConfigEmpty(t *testing.T) {
+func Test_EncodeOtlpTLSConfigEmpty(t *testing.T) {
 	cfg := config.StageParam{
 		Encode: &config.Encode{
 			OtlpLogs: &api.EncodeOtlpLogs{OtlpConnectionInfo: &api.OtlpConnectionInfo{
@@ -111,7 +111,7 @@ func Test_TLSConfigEmpty(t *testing.T) {
 	require.NotNil(t, newEncode)
 }
 
-func Test_TLSConfigNotEmpty(t *testing.T) {
+func Test_EncodeOtlpTLSConfigNotEmpty(t *testing.T) {
 	cfg := config.StageParam{
 		Encode: &config.Encode{
 			OtlpLogs: &api.EncodeOtlpLogs{OtlpConnectionInfo: &api.OtlpConnectionInfo{
@@ -193,9 +193,5 @@ func Test_EncodeOtlpMetrics(t *testing.T) {
 	newEncode, err := NewEncodeOtlpMetrics(operational.NewMetrics(&config.MetricsSettings{}), cfg)
 	require.NoError(t, err)
 	require.NotNil(t, newEncode)
-	em := newEncode.(*EncodeOtlpMetrics)
-	require.Equal(t, 2, len(em.counters))
-	require.Equal(t, 1, len(em.gauges))
-	require.Equal(t, "metric3", em.counters[1].info.Name)
-	require.Equal(t, []string{"label21", "label22"}, em.gauges[0].info.Labels)
+	// TODO: add more tests
 }
diff --git a/pkg/pipeline/encode/prom_cache_test.go b/pkg/pipeline/encode/prom_cache_test.go
index bab6d71e..649b310e 100644
--- a/pkg/pipeline/encode/prom_cache_test.go
+++ b/pkg/pipeline/encode/prom_cache_test.go
@@ -113,12 +113,12 @@ func Test_Prom_Cache1(t *testing.T) {
 	entries = utils.GenerateConnectionFlowEntries(10)
 	require.Equal(t, 10, len(entries))
 	encodeEntries(promEncode, entries)
-	require.Equal(t, 10, promEncode.mCache.GetCacheLen())
+	require.Equal(t, 10, promEncode.metricCommon.mCache.GetCacheLen())
 
 	entries = utils.GenerateConnectionFlowEntries(40)
 	require.Equal(t, 40, len(entries))
 	encodeEntries(promEncode, entries)
-	require.Equal(t, 30, promEncode.mCache.GetCacheLen())
+	require.Equal(t, 30, promEncode.metricCommon.mCache.GetCacheLen())
 }
 
 func Test_Prom_Cache2(t *testing.T) {
@@ -133,12 +133,12 @@ func Test_Prom_Cache2(t *testing.T) {
 	entries = utils.GenerateConnectionFlowEntries(10)
 	require.Equal(t, 10, len(entries))
 	encodeEntries(promEncode, entries)
-	require.Equal(t, 20, promEncode.mCache.GetCacheLen())
+	require.Equal(t, 20, promEncode.metricCommon.mCache.GetCacheLen())
 
 	entries = utils.GenerateConnectionFlowEntries(40)
 	require.Equal(t, 40, len(entries))
 	encodeEntries(promEncode, entries)
-	require.Equal(t, 30, promEncode.mCache.GetCacheLen())
+	require.Equal(t, 30, promEncode.metricCommon.mCache.GetCacheLen())
 }
 
 func Test_Prom_Cache3(t *testing.T) {
@@ -153,10 +153,10 @@ func Test_Prom_Cache3(t *testing.T) {
 	entries = utils.GenerateConnectionFlowEntries(10)
 	require.Equal(t, 10, len(entries))
 	encodeEntries(promEncode, entries)
-	require.Equal(t, 20, promEncode.mCache.GetCacheLen())
+	require.Equal(t, 20, promEncode.metricCommon.mCache.GetCacheLen())
 
 	entries = utils.GenerateConnectionFlowEntries(40)
 	require.Equal(t, 40, len(entries))
 	encodeEntries(promEncode, entries)
-	require.Equal(t, 80, promEncode.mCache.GetCacheLen())
+	require.Equal(t, 80, promEncode.metricCommon.mCache.GetCacheLen())
 }
diff --git a/pkg/pipeline/write/testnorace/write_ipfix_test.go b/pkg/pipeline/write/testnorace/write_ipfix_test.go
index f67ca993..075764d4 100644
--- a/pkg/pipeline/write/testnorace/write_ipfix_test.go
+++ b/pkg/pipeline/write/testnorace/write_ipfix_test.go
@@ -119,7 +119,7 @@ func TestEnrichedIPFIXFlow(t *testing.T) {
 	assert.Equal(t, uint16(10), tplv4Msg.GetVersion())
 	templateSet := tplv4Msg.GetSet()
 	templateElements := templateSet.GetRecords()[0].GetOrderedElementList()
-	assert.Len(t, templateElements, 20)
+	assert.Len(t, templateElements, 21)
 	assert.Equal(t, uint32(0), templateElements[0].GetInfoElement().EnterpriseId)
 
 	// Check data
@@ -176,6 +176,8 @@ func matchElement(t *testing.T, element entities.InfoElementWithValue, flow conf
 		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