Repository: netobserv/flowlogs-pipeline
Pull request #609 (2024-02-26 13:04:56)

Bump google.golang.org/grpc from 1.61.0 to 1.62.0
Bumps [google.golang.org/grpc](https://github.com/grpc/grpc-go) from 1.61.0 to 1.62.0.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/grpc/grpc-go/releases">google.golang.org/grpc's releases</a>.</em></p>
<blockquote>
<h2>Release 1.62.0</h2>
<h1>New Features</h1>
<ul>
<li>grpc: Add StaticMethod CallOption as a signal to stats handler that a method is safe to use as an instrument key (<a href="https://redirect.github.com/grpc/grpc-go/issues/6986">#6986</a>)</li>
</ul>
<h1>Behavior Changes</h1>
<ul>
<li>grpc: Return canonical target string from ClientConn.Target() and resolver.Address.String() (<a href="https://redirect.github.com/grpc/grpc-go/issues/6923">#6923</a>)</li>
</ul>
<h1>Bug Fixes</h1>
<ul>
<li>server: wait to close connection until incoming socket is drained (with timeout) to prevent data loss on client-side (<a href="https://redirect.github.com/grpc/grpc-go/issues/6977">#6977</a>)
<ul>
<li>Special Thanks: <a href="https://github.com/s-matyukevich"><code>@​s-matyukevich</code></a> for discovering the root cause</li>
</ul>
</li>
</ul>
<h1>Performance Improvements</h1>
<ul>
<li>*: Allow building without <code>x/net/trace</code> by using <code>grpcnotrace</code> to enable dead code elimination (<a href="https://redirect.github.com/grpc/grpc-go/issues/6954">#6954</a>)
<ul>
<li>Special Thanks: <a href="https://github.com/hugelgupf"><code>@​hugelgupf</code></a></li>
</ul>
</li>
<li>rand: improve performance and simplify implementation of <code>grpcrand</code> by adopting <code>math/rand</code>'s top-level functions for go version 1.21.0 and newer. (<a href="https://redirect.github.com/grpc/grpc-go/issues/6925">#6925</a>)
<ul>
<li>Special Thanks: <a href="https://github.com/kmirzavaziri"><code>@​kmirzavaziri</code></a></li>
</ul>
</li>
</ul>
<h1>Dependencies</h1>
<ul>
<li>*: Use google.golang.org/protobuf/proto instead of github.com/golang/protobuf. (<a href="https://redirect.github.com/grpc/grpc-go/issues/6919">#6919</a>)
<ul>
<li>Special Thanks: <a href="https://github.com/Clement-Jean"><code>@​Clement-Jean</code></a></li>
</ul>
</li>
</ul>
<blockquote>
<p>[!NOTE]
The above change in proto library usage introduces a minor behavior change within those libraries.  The old <code>github.com/golang/protobuf</code> library would error if given a <code>nil</code> message to <code>Marshal</code>, while the new <code>google.golang.org/protobuf</code> library will successfully output zero bytes in this case.  This means server method handlers that did <code>return nil, nil</code> will now return an empty message and no error, while it used to return an error.  This also affects the client side, where clients sending <code>nil</code> messages used to fail without sending the RPC, and now they will send an empty message.</p>
</blockquote>
<h2>Release 1.61.1</h2>
<h1>Bug Fixes</h1>
<ul>
<li>server: wait to close connection until incoming socket is drained (with timeout) to prevent data loss on client-side (<a href="https://redirect.github.com/grpc/grpc-go/issues/6977">#6977</a>)
<ul>
<li>Special Thanks: <a href="https://github.com/s-matyukevich"><code>@​s-matyukevich</code></a> for discovering the root cause</li>
</ul>
</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/grpc/grpc-go/commit/d715b2ef0602fe2133221519cba0642ac282cc3f"><code>d715b2e</code></a> Change version to 1.62.0 (<a href="https://redirect.github.com/grpc/grpc-go/issues/6994">#6994</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/d706a42d29ab785cc1ad86a09fe828ef0a196f26"><code>d706a42</code></a> cherry-pick <a href="https://redirect.github.com/grpc/grpc-go/issues/6926">#6926</a> to 1.62 release branch (<a href="https://redirect.github.com/grpc/grpc-go/issues/6986">#6986</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/8a4ca0cc41ceb5bd0763c865cc4094650df9f95f"><code>8a4ca0c</code></a> cherry-pick <a href="https://redirect.github.com/grpc/grpc-go/issues/6977">#6977</a> to 1.62.x release branch (<a href="https://redirect.github.com/grpc/grpc-go/issues/6979">#6979</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/d41b01db97ca2e3627b2c9949fffe8f152a4255d"><code>d41b01d</code></a> encoding: fix typo (<a href="https://redirect.github.com/grpc/grpc-go/issues/6966">#6966</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/c2b50ee081682eca4b995d2fb79e642019f78aea"><code>c2b50ee</code></a> deps: fix backwards compatibility with encoding (<a href="https://redirect.github.com/grpc/grpc-go/issues/6965">#6965</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/05b4a8b8f779518bb0119042c8c5806fa33d2184"><code>05b4a8b</code></a> Revert &quot;xds/googlec2p: use xdstp names for LDS (<a href="https://redirect.github.com/grpc/grpc-go/issues/6949">#6949</a>)&quot; (<a href="https://redirect.github.com/grpc/grpc-go/issues/6964">#6964</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/03e76b3d2a467a6deebabe2b323780a1890e36d4"><code>03e76b3</code></a> grpc: add ability to compile with or without tracing (<a href="https://redirect.github.com/grpc/grpc-go/issues/6954">#6954</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/84b85babc00b4b8460e53b6ee110bfb49e9311cf"><code>84b85ba</code></a> xds/googledirectpath: Check if ipv6 address is non empty (<a href="https://redirect.github.com/grpc/grpc-go/issues/6959">#6959</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/6f63f05a5b98c484020c143d69450e8679603953"><code>6f63f05</code></a> internal/grpcrand: use Go top-level random functions for go1.21+ (<a href="https://redirect.github.com/grpc/grpc-go/issues/6925">#6925</a>)</li>
<li><a href="https://github.com/grpc/grpc-go/commit/cd69b5d0af6bd46ca17b94acaa09256a863ac277"><code>cd69b5d</code></a> .*: fix minor linter issues (<a href="https://redirect.github.com/grpc/grpc-go/issues/6958">#6958</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/grpc/grpc-go/compare/v1.61.0...v1.62.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=google.golang.org/grpc&package-manager=go_modules&previous-version=1.61.0&new-version=1.62.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</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 [ronensc](https://github.com/ronensc) 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":["ronensc"]} -->
openshift-ci[bot]: Hi @dependabot[bot]. 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>

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/609?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 67.03%. Comparing base [(`ba7ad1f`)](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/commit/ba7ad1ff7aa32f2d8026327d018a6d55797d3ce2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) to head [(`b1066e4`)](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/609?src=pr&el=desc&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     #609   +/-   ##
=======================================
  Coverage   67.03%   67.03%           
=======================================
  Files         103      103           
  Lines        7401     7401           
=======================================
  Hits         4961     4961           
  Misses       2147     2147           
  Partials      293      293           
```

| [Flag](https://app.codecov.io/gh/netobserv/flowlogs-pipeline/pull/609/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/609/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) | `67.03% <ø> (ø)` | |

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/609?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).


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/go.mod b/go.mod
index 95add854..f56360c5 100644
--- a/go.mod
+++ b/go.mod
@@ -39,7 +39,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.23.1
 	go.opentelemetry.io/otel/sdk/metric v1.23.1
 	golang.org/x/net v0.21.0
-	google.golang.org/grpc v1.61.0
+	google.golang.org/grpc v1.62.0
 	google.golang.org/protobuf v1.32.0
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.29.2
@@ -68,7 +68,7 @@ require (
 	github.com/google/gnostic-models v0.6.8 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
-	github.com/google/uuid v1.5.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.0 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
@@ -123,8 +123,8 @@ require (
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240102182953-50ed04b92917 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240102182953-50ed04b92917 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
diff --git a/go.sum b/go.sum
index 0e1fd0ed..d3002baf 100644
--- a/go.sum
+++ b/go.sum
@@ -401,8 +401,8 @@ github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJY
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
-github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gnostic v0.4.1/go.mod h1:LRhVm6pbyptWbWbuZ38d1eyptfvIytN3ir6b65WBswg=
@@ -1046,7 +1046,7 @@ golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
+golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -1295,11 +1295,11 @@ google.golang.org/genproto v0.0.0-20200729003335-053ba62fc06f/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20200804131852-c06518451d9c/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20200825200019-8632dd797987/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20200904004341-0bd0a958aa1d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
-google.golang.org/genproto v0.0.0-20231212172506-995d672761c0 h1:YJ5pD9rF8o9Qtta0Cmy9rdBwkSjrTCT6XTiUQVOtIos=
-google.golang.org/genproto/googleapis/api v0.0.0-20240102182953-50ed04b92917 h1:rcS6EyEaoCO52hQDupoSfrxI3R6C2Tq741is7X8OvnM=
-google.golang.org/genproto/googleapis/api v0.0.0-20240102182953-50ed04b92917/go.mod h1:CmlNWB9lSezaYELKS5Ym1r44VrrbPUa7JTvw+6MbpJ0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240102182953-50ed04b92917 h1:6G8oQ016D88m1xAKljMlBOOGWDZkes4kMhgGFlf8WcQ=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240102182953-50ed04b92917/go.mod h1:xtjpI3tXFPP051KaWnhvxkiubL/6dJ18vLVf7q2pTOU=
+google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 h1:KAeGQVN3M9nD0/bQXnr/ClcEMJ968gUXJQ9pwfSynuQ=
+google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 h1:Lj5rbfG876hIAYFjqiJnPHfhXbv+nzTWfm04Fg/XSVU=
+google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80/go.mod h1:4jWUdICTdgc3Ibxmr8nAJiiLHwQBY0UI0XZcEMaFKaA=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 h1:AjyfHzEPEFp/NpvfN5g+KDla3EMojjhRVZc1i7cj+oM=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80/go.mod h1:PAREbraiVEVGVdTZsVWjSbbTtSyGbAgIIvni8a8CD5s=
 google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.0/go.mod h1:chYK+tFQF0nDUGJgXMSgLCQk3phJEuONr2DCgLDdAQM=
@@ -1319,8 +1319,8 @@ google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM
 google.golang.org/grpc v1.31.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.31.1/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.32.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
-google.golang.org/grpc v1.61.0 h1:TOvOcuXn30kRao+gfcvsebNEa5iZIiLkisYEkf7R7o0=
-google.golang.org/grpc v1.61.0/go.mod h1:VUbo7IFqmF1QtCAstipjG0GIoq49KvMe9+h1jFLBNJs=
+google.golang.org/grpc v1.62.0 h1:HQKZ/fa1bXkX1oFOvSjmZEUL8wLSaZTjCcLAlmZRtdk=
+google.golang.org/grpc v1.62.0/go.mod h1:IWTG0VlJLCh1SkC58F7np9ka9mx/WNkjl4PGJaiq+QE=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
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
 }
diff --git a/vendor/github.com/golang/protobuf/jsonpb/decode.go b/vendor/github.com/golang/protobuf/jsonpb/decode.go
deleted file mode 100644
index 6c16c255..00000000
--- a/vendor/github.com/golang/protobuf/jsonpb/decode.go
+++ /dev/null
@@ -1,530 +0,0 @@
-// Copyright 2015 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-package jsonpb
-
-import (
-	"encoding/json"
-	"errors"
-	"fmt"
-	"io"
-	"math"
-	"reflect"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/golang/protobuf/proto"
-	"google.golang.org/protobuf/encoding/protojson"
-	protoV2 "google.golang.org/protobuf/proto"
-	"google.golang.org/protobuf/reflect/protoreflect"
-	"google.golang.org/protobuf/reflect/protoregistry"
-)
-
-const wrapJSONUnmarshalV2 = false
-
-// UnmarshalNext unmarshals the next JSON object from d into m.
-func UnmarshalNext(d *json.Decoder, m proto.Message) error {
-	return new(Unmarshaler).UnmarshalNext(d, m)
-}
-
-// Unmarshal unmarshals a JSON object from r into m.
-func Unmarshal(r io.Reader, m proto.Message) error {
-	return new(Unmarshaler).Unmarshal(r, m)
-}
-
-// UnmarshalString unmarshals a JSON object from s into m.
-func UnmarshalString(s string, m proto.Message) error {
-	return new(Unmarshaler).Unmarshal(strings.NewReader(s), m)
-}
-
-// Unmarshaler is a configurable object for converting from a JSON
-// representation to a protocol buffer object.
-type Unmarshaler struct {
-	// AllowUnknownFields specifies whether to allow messages to contain
-	// unknown JSON fields, as opposed to failing to unmarshal.
-	AllowUnknownFields bool
-
-	// AnyResolver is used to resolve the google.protobuf.Any well-known type.
-	// If unset, the global registry is used by default.
-	AnyResolver AnyResolver
-}
-
-// JSONPBUnmarshaler is implemented by protobuf messages that customize the way
-// they are unmarshaled from JSON. Messages that implement this should also
-// implement JSONPBMarshaler so that the custom format can be produced.
-//
-// The JSON unmarshaling must follow the JSON to proto specification:
-//	https://developers.google.com/protocol-buffers/docs/proto3#json
-//
-// Deprecated: Custom types should implement protobuf reflection instead.
-type JSONPBUnmarshaler interface {
-	UnmarshalJSONPB(*Unmarshaler, []byte) error
-}
-
-// Unmarshal unmarshals a JSON object from r into m.
-func (u *Unmarshaler) Unmarshal(r io.Reader, m proto.Message) error {
-	return u.UnmarshalNext(json.NewDecoder(r), m)
-}
-
-// UnmarshalNext unmarshals the next JSON object from d into m.
-func (u *Unmarshaler) UnmarshalNext(d *json.Decoder, m proto.Message) error {
-	if m == nil {
-		return errors.New("invalid nil message")
-	}
-
-	// Parse the next JSON object from the stream.
-	raw := json.RawMessage{}
-	if err := d.Decode(&raw); err != nil {
-		return err
-	}
-
-	// Check for custom unmarshalers first since they may not properly
-	// implement protobuf reflection that the logic below relies on.
-	if jsu, ok := m.(JSONPBUnmarshaler); ok {
-		return jsu.UnmarshalJSONPB(u, raw)
-	}
-
-	mr := proto.MessageReflect(m)
-
-	// NOTE: For historical reasons, a top-level null is treated as a noop.
-	// This is incorrect, but kept for compatibility.
-	if string(raw) == "null" && mr.Descriptor().FullName() != "google.protobuf.Value" {
-		return nil
-	}
-
-	if wrapJSONUnmarshalV2 {
-		// NOTE: If input message is non-empty, we need to preserve merge semantics
-		// of the old jsonpb implementation. These semantics are not supported by
-		// the protobuf JSON specification.
-		isEmpty := true
-		mr.Range(func(protoreflect.FieldDescriptor, protoreflect.Value) bool {
-			isEmpty = false // at least one iteration implies non-empty
-			return false
-		})
-		if !isEmpty {
-			// Perform unmarshaling into a newly allocated, empty message.
-			mr = mr.New()
-
-			// Use a defer to copy all unmarshaled fields into the original message.
-			dst := proto.MessageReflect(m)
-			defer mr.Range(func(fd protoreflect.FieldDescriptor, v protoreflect.Value) bool {
-				dst.Set(fd, v)
-				return true
-			})
-		}
-
-		// Unmarshal using the v2 JSON unmarshaler.
-		opts := protojson.UnmarshalOptions{
-			DiscardUnknown: u.AllowUnknownFields,
-		}
-		if u.AnyResolver != nil {
-			opts.Resolver = anyResolver{u.AnyResolver}
-		}
-		return opts.Unmarshal(raw, mr.Interface())
-	} else {
-		if err := u.unmarshalMessage(mr, raw); err != nil {
-			return err
-		}
-		return protoV2.CheckInitialized(mr.Interface())
-	}
-}
-
-func (u *Unmarshaler) unmarshalMessage(m protoreflect.Message, in []byte) error {
-	md := m.Descriptor()
-	fds := md.Fields()
-
-	if jsu, ok := proto.MessageV1(m.Interface()).(JSONPBUnmarshaler); ok {
-		return jsu.UnmarshalJSONPB(u, in)
-	}
-
-	if string(in) == "null" && md.FullName() != "google.protobuf.Value" {
-		return nil
-	}
-
-	switch wellKnownType(md.FullName()) {
-	case "Any":
-		var jsonObject map[string]json.RawMessage
-		if err := json.Unmarshal(in, &jsonObject); err != nil {
-			return err
-		}
-
-		rawTypeURL, ok := jsonObject["@type"]
-		if !ok {
-			return errors.New("Any JSON doesn't have '@type'")
-		}
-		typeURL, err := unquoteString(string(rawTypeURL))
-		if err != nil {
-			return fmt.Errorf("can't unmarshal Any's '@type': %q", rawTypeURL)
-		}
-		m.Set(fds.ByNumber(1), protoreflect.ValueOfString(typeURL))
-
-		var m2 protoreflect.Message
-		if u.AnyResolver != nil {
-			mi, err := u.AnyResolver.Resolve(typeURL)
-			if err != nil {
-				return err
-			}
-			m2 = proto.MessageReflect(mi)
-		} else {
-			mt, err := protoregistry.GlobalTypes.FindMessageByURL(typeURL)
-			if err != nil {
-				if err == protoregistry.NotFound {
-					return fmt.Errorf("could not resolve Any message type: %v", typeURL)
-				}
-				return err
-			}
-			m2 = mt.New()
-		}
-
-		if wellKnownType(m2.Descriptor().FullName()) != "" {
-			rawValue, ok := jsonObject["value"]
-			if !ok {
-				return errors.New("Any JSON doesn't have 'value'")
-			}
-			if err := u.unmarshalMessage(m2, rawValue); err != nil {
-				return fmt.Errorf("can't unmarshal Any nested proto %v: %v", typeURL, err)
-			}
-		} else {
-			delete(jsonObject, "@type")
-			rawJSON, err := json.Marshal(jsonObject)
-			if err != nil {
-				return fmt.Errorf("can't generate JSON for Any's nested proto to be unmarshaled: %v", err)
-			}
-			if err = u.unmarshalMessage(m2, rawJSON); err != nil {
-				return fmt.Errorf("can't unmarshal Any nested proto %v: %v", typeURL, err)
-			}
-		}
-
-		rawWire, err := protoV2.Marshal(m2.Interface())
-		if err != nil {
-			return fmt.Errorf("can't marshal proto %v into Any.Value: %v", typeURL, err)
-		}
-		m.Set(fds.ByNumber(2), protoreflect.ValueOfBytes(rawWire))
-		return nil
-	case "BoolValue", "BytesValue", "StringValue",
-		"Int32Value", "UInt32Value", "FloatValue",
-		"Int64Value", "UInt64Value", "DoubleValue":
-		fd := fds.ByNumber(1)
-		v, err := u.unmarshalValue(m.NewField(fd), in, fd)
-		if err != nil {
-			return err
-		}
-		m.Set(fd, v)
-		return nil
-	case "Duration":
-		v, err := unquoteString(string(in))
-		if err != nil {
-			return err
-		}
-		d, err := time.ParseDuration(v)
-		if err != nil {
-			return fmt.Errorf("bad Duration: %v", err)
-		}
-
-		sec := d.Nanoseconds() / 1e9
-		nsec := d.Nanoseconds() % 1e9
-		m.Set(fds.ByNumber(1), protoreflect.ValueOfInt64(int64(sec)))
-		m.Set(fds.ByNumber(2), protoreflect.ValueOfInt32(int32(nsec)))
-		return nil
-	case "Timestamp":
-		v, err := unquoteString(string(in))
-		if err != nil {
-			return err
-		}
-		t, err := time.Parse(time.RFC3339Nano, v)
-		if err != nil {
-			return fmt.Errorf("bad Timestamp: %v", err)
-		}
-
-		sec := t.Unix()
-		nsec := t.Nanosecond()
-		m.Set(fds.ByNumber(1), protoreflect.ValueOfInt64(int64(sec)))
-		m.Set(fds.ByNumber(2), protoreflect.ValueOfInt32(int32(nsec)))
-		return nil
-	case "Value":
-		switch {
-		case string(in) == "null":
-			m.Set(fds.ByNumber(1), protoreflect.ValueOfEnum(0))
-		case string(in) == "true":
-			m.Set(fds.ByNumber(4), protoreflect.ValueOfBool(true))
-		case string(in) == "false":
-			m.Set(fds.ByNumber(4), protoreflect.ValueOfBool(false))
-		case hasPrefixAndSuffix('"', in, '"'):
-			s, err := unquoteString(string(in))
-			if err != nil {
-				return fmt.Errorf("unrecognized type for Value %q", in)
-			}
-			m.Set(fds.ByNumber(3), protoreflect.ValueOfString(s))
-		case hasPrefixAndSuffix('[', in, ']'):
-			v := m.Mutable(fds.ByNumber(6))
-			return u.unmarshalMessage(v.Message(), in)
-		case hasPrefixAndSuffix('{', in, '}'):
-			v := m.Mutable(fds.ByNumber(5))
-			return u.unmarshalMessage(v.Message(), in)
-		default:
-			f, err := strconv.ParseFloat(string(in), 0)
-			if err != nil {
-				return fmt.Errorf("unrecognized type for Value %q", in)
-			}
-			m.Set(fds.ByNumber(2), protoreflect.ValueOfFloat64(f))
-		}
-		return nil
-	case "ListValue":
-		var jsonArray []json.RawMessage
-		if err := json.Unmarshal(in, &jsonArray); err != nil {
-			return fmt.Errorf("bad ListValue: %v", err)
-		}
-
-		lv := m.Mutable(fds.ByNumber(1)).List()
-		for _, raw := range jsonArray {
-			ve := lv.NewElement()
-			if err := u.unmarshalMessage(ve.Message(), raw); err != nil {
-				return err
-			}
-			lv.Append(ve)
-		}
-		return nil
-	case "Struct":
-		var jsonObject map[string]json.RawMessage
-		if err := json.Unmarshal(in, &jsonObject); err != nil {
-			return fmt.Errorf("bad StructValue: %v", err)
-		}
-
-		mv := m.Mutable(fds.ByNumber(1)).Map()
-		for key, raw := range jsonObject {
-			kv := protoreflect.ValueOf(key).MapKey()
-			vv := mv.NewValue()
-			if err := u.unmarshalMessage(vv.Message(), raw); err != nil {
-				return fmt.Errorf("bad value in StructValue for key %q: %v", key, err)
-			}
-			mv.Set(kv, vv)
-		}
-		return nil
-	}
-
-	var jsonObject map[string]json.RawMessage
-	if err := json.Unmarshal(in, &jsonObject); err != nil {
-		return err
-	}
-
-	// Handle known fields.
-	for i := 0; i < fds.Len(); i++ {
-		fd := fds.Get(i)
-		if fd.IsWeak() && fd.Message().IsPlaceholder() {
-			continue //  weak reference is not linked in
-		}
-
-		// Search for any raw JSON value associated with this field.
-		var raw json.RawMessage
-		name := string(fd.Name())
-		if fd.Kind() == protoreflect.GroupKind {
-			name = string(fd.Message().Name())
-		}
-		if v, ok := jsonObject[name]; ok {
-			delete(jsonObject, name)
-			raw = v
-		}
-		name = string(fd.JSONName())
-		if v, ok := jsonObject[name]; ok {
-			delete(jsonObject, name)
-			raw = v
-		}
-
-		field := m.NewField(fd)
-		// Unmarshal the field value.
-		if raw == nil || (string(raw) == "null" && !isSingularWellKnownValue(fd) && !isSingularJSONPBUnmarshaler(field, fd)) {
-			continue
-		}
-		v, err := u.unmarshalValue(field, raw, fd)
-		if err != nil {
-			return err
-		}
-		m.Set(fd, v)
-	}
-
-	// Handle extension fields.
-	for name, raw := range jsonObject {
-		if !strings.HasPrefix(name, "[") || !strings.HasSuffix(name, "]") {
-			continue
-		}
-
-		// Resolve the extension field by name.
-		xname := protoreflect.FullName(name[len("[") : len(name)-len("]")])
-		xt, _ := protoregistry.GlobalTypes.FindExtensionByName(xname)
-		if xt == nil && isMessageSet(md) {
-			xt, _ = protoregistry.GlobalTypes.FindExtensionByName(xname.Append("message_set_extension"))
-		}
-		if xt == nil {
-			continue
-		}
-		delete(jsonObject, name)
-		fd := xt.TypeDescriptor()
-		if fd.ContainingMessage().FullName() != m.Descriptor().FullName() {
-			return fmt.Errorf("extension field %q does not extend message %q", xname, m.Descriptor().FullName())
-		}
-
-		field := m.NewField(fd)
-		// Unmarshal the field value.
-		if raw == nil || (string(raw) == "null" && !isSingularWellKnownValue(fd) && !isSingularJSONPBUnmarshaler(field, fd)) {
-			continue
-		}
-		v, err := u.unmarshalValue(field, raw, fd)
-		if err != nil {
-			return err
-		}
-		m.Set(fd, v)
-	}
-
-	if !u.AllowUnknownFields && len(jsonObject) > 0 {
-		for name := range jsonObject {
-			return fmt.Errorf("unknown field %q in %v", name, md.FullName())
-		}
-	}
-	return nil
-}
-
-func isSingularWellKnownValue(fd protoreflect.FieldDescriptor) bool {
-	if fd.Cardinality() == protoreflect.Repeated {
-		return false
-	}
-	if md := fd.Message(); md != nil {
-		return md.FullName() == "google.protobuf.Value"
-	}
-	if ed := fd.Enum(); ed != nil {
-		return ed.FullName() == "google.protobuf.NullValue"
-	}
-	return false
-}
-
-func isSingularJSONPBUnmarshaler(v protoreflect.Value, fd protoreflect.FieldDescriptor) bool {
-	if fd.Message() != nil && fd.Cardinality() != protoreflect.Repeated {
-		_, ok := proto.MessageV1(v.Interface()).(JSONPBUnmarshaler)
-		return ok
-	}
-	return false
-}
-
-func (u *Unmarshaler) unmarshalValue(v protoreflect.Value, in []byte, fd protoreflect.FieldDescriptor) (protoreflect.Value, error) {
-	switch {
-	case fd.IsList():
-		var jsonArray []json.RawMessage
-		if err := json.Unmarshal(in, &jsonArray); err != nil {
-			return v, err
-		}
-		lv := v.List()
-		for _, raw := range jsonArray {
-			ve, err := u.unmarshalSingularValue(lv.NewElement(), raw, fd)
-			if err != nil {
-				return v, err
-			}
-			lv.Append(ve)
-		}
-		return v, nil
-	case fd.IsMap():
-		var jsonObject map[string]json.RawMessage
-		if err := json.Unmarshal(in, &jsonObject); err != nil {
-			return v, err
-		}
-		kfd := fd.MapKey()
-		vfd := fd.MapValue()
-		mv := v.Map()
-		for key, raw := range jsonObject {
-			var kv protoreflect.MapKey
-			if kfd.Kind() == protoreflect.StringKind {
-				kv = protoreflect.ValueOf(key).MapKey()
-			} else {
-				v, err := u.unmarshalSingularValue(kfd.Default(), []byte(key), kfd)
-				if err != nil {
-					return v, err
-				}
-				kv = v.MapKey()
-			}
-
-			vv, err := u.unmarshalSingularValue(mv.NewValue(), raw, vfd)
-			if err != nil {
-				return v, err
-			}
-			mv.Set(kv, vv)
-		}
-		return v, nil
-	default:
-		return u.unmarshalSingularValue(v, in, fd)
-	}
-}
-
-var nonFinite = map[string]float64{
-	`"NaN"`:       math.NaN(),
-	`"Infinity"`:  math.Inf(+1),
-	`"-Infinity"`: math.Inf(-1),
-}
-
-func (u *Unmarshaler) unmarshalSingularValue(v protoreflect.Value, in []byte, fd protoreflect.FieldDescriptor) (protoreflect.Value, error) {
-	switch fd.Kind() {
-	case protoreflect.BoolKind:
-		return unmarshalValue(in, new(bool))
-	case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind:
-		return unmarshalValue(trimQuote(in), new(int32))
-	case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
-		return unmarshalValue(trimQuote(in), new(int64))
-	case protoreflect.Uint32Kind, protoreflect.Fixed32Kind:
-		return unmarshalValue(trimQuote(in), new(uint32))
-	case protoreflect.Uint64Kind, protoreflect.Fixed64Kind:
-		return unmarshalValue(trimQuote(in), new(uint64))
-	case protoreflect.FloatKind:
-		if f, ok := nonFinite[string(in)]; ok {
-			return protoreflect.ValueOfFloat32(float32(f)), nil
-		}
-		return unmarshalValue(trimQuote(in), new(float32))
-	case protoreflect.DoubleKind:
-		if f, ok := nonFinite[string(in)]; ok {
-			return protoreflect.ValueOfFloat64(float64(f)), nil
-		}
-		return unmarshalValue(trimQuote(in), new(float64))
-	case protoreflect.StringKind:
-		return unmarshalValue(in, new(string))
-	case protoreflect.BytesKind:
-		return unmarshalValue(in, new([]byte))
-	case protoreflect.EnumKind:
-		if hasPrefixAndSuffix('"', in, '"') {
-			vd := fd.Enum().Values().ByName(protoreflect.Name(trimQuote(in)))
-			if vd == nil {
-				return v, fmt.Errorf("unknown value %q for enum %s", in, fd.Enum().FullName())
-			}
-			return protoreflect.ValueOfEnum(vd.Number()), nil
-		}
-		return unmarshalValue(in, new(protoreflect.EnumNumber))
-	case protoreflect.MessageKind, protoreflect.GroupKind:
-		err := u.unmarshalMessage(v.Message(), in)
-		return v, err
-	default:
-		panic(fmt.Sprintf("invalid kind %v", fd.Kind()))
-	}
-}
-
-func unmarshalValue(in []byte, v interface{}) (protoreflect.Value, error) {
-	err := json.Unmarshal(in, v)
-	return protoreflect.ValueOf(reflect.ValueOf(v).Elem().Interface()), err
-}
-
-func unquoteString(in string) (out string, err error) {
-	err = json.Unmarshal([]byte(in), &out)
-	return out, err
-}
-
-func hasPrefixAndSuffix(prefix byte, in []byte, suffix byte) bool {
-	if len(in) >= 2 && in[0] == prefix && in[len(in)-1] == suffix {
-		return true
-	}
-	return false
-}
-
-// trimQuote is like unquoteString but simply strips surrounding quotes.
-// This is incorrect, but is behavior done by the legacy implementation.
-func trimQuote(in []byte) []byte {
-	if len(in) >= 2 && in[0] == '"' && in[len(in)-1] == '"' {
-		in = in[1 : len(in)-1]
-	}
-	return in
-}
diff --git a/vendor/github.com/golang/protobuf/jsonpb/encode.go b/vendor/github.com/golang/protobuf/jsonpb/encode.go
deleted file mode 100644
index 685c80a6..00000000
--- a/vendor/github.com/golang/protobuf/jsonpb/encode.go
+++ /dev/null
@@ -1,559 +0,0 @@
-// Copyright 2015 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-package jsonpb
-
-import (
-	"encoding/json"
-	"errors"
-	"fmt"
-	"io"
-	"math"
-	"reflect"
-	"sort"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/golang/protobuf/proto"
-	"google.golang.org/protobuf/encoding/protojson"
-	protoV2 "google.golang.org/protobuf/proto"
-	"google.golang.org/protobuf/reflect/protoreflect"
-	"google.golang.org/protobuf/reflect/protoregistry"
-)
-
-const wrapJSONMarshalV2 = false
-
-// Marshaler is a configurable object for marshaling protocol buffer messages
-// to the specified JSON representation.
-type Marshaler struct {
-	// OrigName specifies whether to use the original protobuf name for fields.
-	OrigName bool
-
-	// EnumsAsInts specifies whether to render enum values as integers,
-	// as opposed to string values.
-	EnumsAsInts bool
-
-	// EmitDefaults specifies whether to render fields with zero values.
-	EmitDefaults bool
-
-	// Indent controls whether the output is compact or not.
-	// If empty, the output is compact JSON. Otherwise, every JSON object
-	// entry and JSON array value will be on its own line.
-	// Each line will be preceded by repeated copies of Indent, where the
-	// number of copies is the current indentation depth.
-	Indent string
-
-	// AnyResolver is used to resolve the google.protobuf.Any well-known type.
-	// If unset, the global registry is used by default.
-	AnyResolver AnyResolver
-}
-
-// JSONPBMarshaler is implemented by protobuf messages that customize the
-// way they are marshaled to JSON. Messages that implement this should also
-// implement JSONPBUnmarshaler so that the custom format can be parsed.
-//
-// The JSON marshaling must follow the proto to JSON specification:
-//	https://developers.google.com/protocol-buffers/docs/proto3#json
-//
-// Deprecated: Custom types should implement protobuf reflection instead.
-type JSONPBMarshaler interface {
-	MarshalJSONPB(*Marshaler) ([]byte, error)
-}
-
-// Marshal serializes a protobuf message as JSON into w.
-func (jm *Marshaler) Marshal(w io.Writer, m proto.Message) error {
-	b, err := jm.marshal(m)
-	if len(b) > 0 {
-		if _, err := w.Write(b); err != nil {
-			return err
-		}
-	}
-	return err
-}
-
-// MarshalToString serializes a protobuf message as JSON in string form.
-func (jm *Marshaler) MarshalToString(m proto.Message) (string, error) {
-	b, err := jm.marshal(m)
-	if err != nil {
-		return "", err
-	}
-	return string(b), nil
-}
-
-func (jm *Marshaler) marshal(m proto.Message) ([]byte, error) {
-	v := reflect.ValueOf(m)
-	if m == nil || (v.Kind() == reflect.Ptr && v.IsNil()) {
-		return nil, errors.New("Marshal called with nil")
-	}
-
-	// Check for custom marshalers first since they may not properly
-	// implement protobuf reflection that the logic below relies on.
-	if jsm, ok := m.(JSONPBMarshaler); ok {
-		return jsm.MarshalJSONPB(jm)
-	}
-
-	if wrapJSONMarshalV2 {
-		opts := protojson.MarshalOptions{
-			UseProtoNames:   jm.OrigName,
-			UseEnumNumbers:  jm.EnumsAsInts,
-			EmitUnpopulated: jm.EmitDefaults,
-			Indent:          jm.Indent,
-		}
-		if jm.AnyResolver != nil {
-			opts.Resolver = anyResolver{jm.AnyResolver}
-		}
-		return opts.Marshal(proto.MessageReflect(m).Interface())
-	} else {
-		// Check for unpopulated required fields first.
-		m2 := proto.MessageReflect(m)
-		if err := protoV2.CheckInitialized(m2.Interface()); err != nil {
-			return nil, err
-		}
-
-		w := jsonWriter{Marshaler: jm}
-		err := w.marshalMessage(m2, "", "")
-		return w.buf, err
-	}
-}
-
-type jsonWriter struct {
-	*Marshaler
-	buf []byte
-}
-
-func (w *jsonWriter) write(s string) {
-	w.buf = append(w.buf, s...)
-}
-
-func (w *jsonWriter) marshalMessage(m protoreflect.Message, indent, typeURL string) error {
-	if jsm, ok := proto.MessageV1(m.Interface()).(JSONPBMarshaler); ok {
-		b, err := jsm.MarshalJSONPB(w.Marshaler)
-		if err != nil {
-			return err
-		}
-		if typeURL != "" {
-			// we are marshaling this object to an Any type
-			var js map[string]*json.RawMessage
-			if err = json.Unmarshal(b, &js); err != nil {
-				return fmt.Errorf("type %T produced invalid JSON: %v", m.Interface(), err)
-			}
-			turl, err := json.Marshal(typeURL)
-			if err != nil {
-				return fmt.Errorf("failed to marshal type URL %q to JSON: %v", typeURL, err)
-			}
-			js["@type"] = (*json.RawMessage)(&turl)
-			if b, err = json.Marshal(js); err != nil {
-				return err
-			}
-		}
-		w.write(string(b))
-		return nil
-	}
-
-	md := m.Descriptor()
-	fds := md.Fields()
-
-	// Handle well-known types.
-	const secondInNanos = int64(time.Second / time.Nanosecond)
-	switch wellKnownType(md.FullName()) {
-	case "Any":
-		return w.marshalAny(m, indent)
-	case "BoolValue", "BytesValue", "StringValue",
-		"Int32Value", "UInt32Value", "FloatValue",
-		"Int64Value", "UInt64Value", "DoubleValue":
-		fd := fds.ByNumber(1)
-		return w.marshalValue(fd, m.Get(fd), indent)
-	case "Duration":
-		const maxSecondsInDuration = 315576000000
-		// "Generated output always contains 0, 3, 6, or 9 fractional digits,
-		//  depending on required precision."
-		s := m.Get(fds.ByNumber(1)).Int()
-		ns := m.Get(fds.ByNumber(2)).Int()
-		if s < -maxSecondsInDuration || s > maxSecondsInDuration {
-			return fmt.Errorf("seconds out of range %v", s)
-		}
-		if ns <= -secondInNanos || ns >= secondInNanos {
-			return fmt.Errorf("ns out of range (%v, %v)", -secondInNanos, secondInNanos)
-		}
-		if (s > 0 && ns < 0) || (s < 0 && ns > 0) {
-			return errors.New("signs of seconds and nanos do not match")
-		}
-		var sign string
-		if s < 0 || ns < 0 {
-			sign, s, ns = "-", -1*s, -1*ns
-		}
-		x := fmt.Sprintf("%s%d.%09d", sign, s, ns)
-		x = strings.TrimSuffix(x, "000")
-		x = strings.TrimSuffix(x, "000")
-		x = strings.TrimSuffix(x, ".000")
-		w.write(fmt.Sprintf(`"%vs"`, x))
-		return nil
-	case "Timestamp":
-		// "RFC 3339, where generated output will always be Z-normalized
-		//  and uses 0, 3, 6 or 9 fractional digits."
-		s := m.Get(fds.ByNumber(1)).Int()
-		ns := m.Get(fds.ByNumber(2)).Int()
-		if ns < 0 || ns >= secondInNanos {
-			return fmt.Errorf("ns out of range [0, %v)", secondInNanos)
-		}
-		t := time.Unix(s, ns).UTC()
-		// time.RFC3339Nano isn't exactly right (we need to get 3/6/9 fractional digits).
-		x := t.Format("2006-01-02T15:04:05.000000000")
-		x = strings.TrimSuffix(x, "000")
-		x = strings.TrimSuffix(x, "000")
-		x = strings.TrimSuffix(x, ".000")
-		w.write(fmt.Sprintf(`"%vZ"`, x))
-		return nil
-	case "Value":
-		// JSON value; which is a null, number, string, bool, object, or array.
-		od := md.Oneofs().Get(0)
-		fd := m.WhichOneof(od)
-		if fd == nil {
-			return errors.New("nil Value")
-		}
-		return w.marshalValue(fd, m.Get(fd), indent)
-	case "Struct", "ListValue":
-		// JSON object or array.
-		fd := fds.ByNumber(1)
-		return w.marshalValue(fd, m.Get(fd), indent)
-	}
-
-	w.write("{")
-	if w.Indent != "" {
-		w.write("\n")
-	}
-
-	firstField := true
-	if typeURL != "" {
-		if err := w.marshalTypeURL(indent, typeURL); err != nil {
-			return err
-		}
-		firstField = false
-	}
-
-	for i := 0; i < fds.Len(); {
-		fd := fds.Get(i)
-		if od := fd.ContainingOneof(); od != nil {
-			fd = m.WhichOneof(od)
-			i += od.Fields().Len()
-			if fd == nil {
-				continue
-			}
-		} else {
-			i++
-		}
-
-		v := m.Get(fd)
-
-		if !m.Has(fd) {
-			if !w.EmitDefaults || fd.ContainingOneof() != nil {
-				continue
-			}
-			if fd.Cardinality() != protoreflect.Repeated && (fd.Message() != nil || fd.Syntax() == protoreflect.Proto2) {
-				v = protoreflect.Value{} // use "null" for singular messages or proto2 scalars
-			}
-		}
-
-		if !firstField {
-			w.writeComma()
-		}
-		if err := w.marshalField(fd, v, indent); err != nil {
-			return err
-		}
-		firstField = false
-	}
-
-	// Handle proto2 extensions.
-	if md.ExtensionRanges().Len() > 0 {
-		// Collect a sorted list of all extension descriptor and values.
-		type ext struct {
-			desc protoreflect.FieldDescriptor
-			val  protoreflect.Value
-		}
-		var exts []ext
-		m.Range(func(fd protoreflect.FieldDescriptor, v protoreflect.Value) bool {
-			if fd.IsExtension() {
-				exts = append(exts, ext{fd, v})
-			}
-			return true
-		})
-		sort.Slice(exts, func(i, j int) bool {
-			return exts[i].desc.Number() < exts[j].desc.Number()
-		})
-
-		for _, ext := range exts {
-			if !firstField {
-				w.writeComma()
-			}
-			if err := w.marshalField(ext.desc, ext.val, indent); err != nil {
-				return err
-			}
-			firstField = false
-		}
-	}
-
-	if w.Indent != "" {
-		w.write("\n")
-		w.write(indent)
-	}
-	w.write("}")
-	return nil
-}
-
-func (w *jsonWriter) writeComma() {
-	if w.Indent != "" {
-		w.write(",\n")
-	} else {
-		w.write(",")
-	}
-}
-
-func (w *jsonWriter) marshalAny(m protoreflect.Message, indent string) error {
-	// "If the Any contains a value that has a special JSON mapping,
-	//  it will be converted as follows: {"@type": xxx, "value": yyy}.
-	//  Otherwise, the value will be converted into a JSON object,
-	//  and the "@type" field will be inserted to indicate the actual data type."
-	md := m.Descriptor()
-	typeURL := m.Get(md.Fields().ByNumber(1)).String()
-	rawVal := m.Get(md.Fields().ByNumber(2)).Bytes()
-
-	var m2 protoreflect.Message
-	if w.AnyResolver != nil {
-		mi, err := w.AnyResolver.Resolve(typeURL)
-		if err != nil {
-			return err
-		}
-		m2 = proto.MessageReflect(mi)
-	} else {
-		mt, err := protoregistry.GlobalTypes.FindMessageByURL(typeURL)
-		if err != nil {
-			return err
-		}
-		m2 = mt.New()
-	}
-
-	if err := protoV2.Unmarshal(rawVal, m2.Interface()); err != nil {
-		return err
-	}
-
-	if wellKnownType(m2.Descriptor().FullName()) == "" {
-		return w.marshalMessage(m2, indent, typeURL)
-	}
-
-	w.write("{")
-	if w.Indent != "" {
-		w.write("\n")
-	}
-	if err := w.marshalTypeURL(indent, typeURL); err != nil {
-		return err
-	}
-	w.writeComma()
-	if w.Indent != "" {
-		w.write(indent)
-		w.write(w.Indent)
-		w.write(`"value": `)
-	} else {
-		w.write(`"value":`)
-	}
-	if err := w.marshalMessage(m2, indent+w.Indent, ""); err != nil {
-		return err
-	}
-	if w.Indent != "" {
-		w.write("\n")
-		w.write(indent)
-	}
-	w.write("}")
-	return nil
-}
-
-func (w *jsonWriter) marshalTypeURL(indent, typeURL string) error {
-	if w.Indent != "" {
-		w.write(indent)
-		w.write(w.Indent)
-	}
-	w.write(`"@type":`)
-	if w.Indent != "" {
-		w.write(" ")
-	}
-	b, err := json.Marshal(typeURL)
-	if err != nil {
-		return err
-	}
-	w.write(string(b))
-	return nil
-}
-
-// marshalField writes field description and value to the Writer.
-func (w *jsonWriter) marshalField(fd protoreflect.FieldDescriptor, v protoreflect.Value, indent string) error {
-	if w.Indent != "" {
-		w.write(indent)
-		w.write(w.Indent)
-	}
-	w.write(`"`)
-	switch {
-	case fd.IsExtension():
-		// For message set, use the fname of the message as the extension name.
-		name := string(fd.FullName())
-		if isMessageSet(fd.ContainingMessage()) {
-			name = strings.TrimSuffix(name, ".message_set_extension")
-		}
-
-		w.write("[" + name + "]")
-	case w.OrigName:
-		name := string(fd.Name())
-		if fd.Kind() == protoreflect.GroupKind {
-			name = string(fd.Message().Name())
-		}
-		w.write(name)
-	default:
-		w.write(string(fd.JSONName()))
-	}
-	w.write(`":`)
-	if w.Indent != "" {
-		w.write(" ")
-	}
-	return w.marshalValue(fd, v, indent)
-}
-
-func (w *jsonWriter) marshalValue(fd protoreflect.FieldDescriptor, v protoreflect.Value, indent string) error {
-	switch {
-	case fd.IsList():
-		w.write("[")
-		comma := ""
-		lv := v.List()
-		for i := 0; i < lv.Len(); i++ {
-			w.write(comma)
-			if w.Indent != "" {
-				w.write("\n")
-				w.write(indent)
-				w.write(w.Indent)
-				w.write(w.Indent)
-			}
-			if err := w.marshalSingularValue(fd, lv.Get(i), indent+w.Indent); err != nil {
-				return err
-			}
-			comma = ","
-		}
-		if w.Indent != "" {
-			w.write("\n")
-			w.write(indent)
-			w.write(w.Indent)
-		}
-		w.write("]")
-		return nil
-	case fd.IsMap():
-		kfd := fd.MapKey()
-		vfd := fd.MapValue()
-		mv := v.Map()
-
-		// Collect a sorted list of all map keys and values.
-		type entry struct{ key, val protoreflect.Value }
-		var entries []entry
-		mv.Range(func(k protoreflect.MapKey, v protoreflect.Value) bool {
-			entries = append(entries, entry{k.Value(), v})
-			return true
-		})
-		sort.Slice(entries, func(i, j int) bool {
-			switch kfd.Kind() {
-			case protoreflect.BoolKind:
-				return !entries[i].key.Bool() && entries[j].key.Bool()
-			case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind, protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
-				return entries[i].key.Int() < entries[j].key.Int()
-			case protoreflect.Uint32Kind, protoreflect.Fixed32Kind, protoreflect.Uint64Kind, protoreflect.Fixed64Kind:
-				return entries[i].key.Uint() < entries[j].key.Uint()
-			case protoreflect.StringKind:
-				return entries[i].key.String() < entries[j].key.String()
-			default:
-				panic("invalid kind")
-			}
-		})
-
-		w.write(`{`)
-		comma := ""
-		for _, entry := range entries {
-			w.write(comma)
-			if w.Indent != "" {
-				w.write("\n")
-				w.write(indent)
-				w.write(w.Indent)
-				w.write(w.Indent)
-			}
-
-			s := fmt.Sprint(entry.key.Interface())
-			b, err := json.Marshal(s)
-			if err != nil {
-				return err
-			}
-			w.write(string(b))
-
-			w.write(`:`)
-			if w.Indent != "" {
-				w.write(` `)
-			}
-
-			if err := w.marshalSingularValue(vfd, entry.val, indent+w.Indent); err != nil {
-				return err
-			}
-			comma = ","
-		}
-		if w.Indent != "" {
-			w.write("\n")
-			w.write(indent)
-			w.write(w.Indent)
-		}
-		w.write(`}`)
-		return nil
-	default:
-		return w.marshalSingularValue(fd, v, indent)
-	}
-}
-
-func (w *jsonWriter) marshalSingularValue(fd protoreflect.FieldDescriptor, v protoreflect.Value, indent string) error {
-	switch {
-	case !v.IsValid():
-		w.write("null")
-		return nil
-	case fd.Message() != nil:
-		return w.marshalMessage(v.Message(), indent+w.Indent, "")
-	case fd.Enum() != nil:
-		if fd.Enum().FullName() == "google.protobuf.NullValue" {
-			w.write("null")
-			return nil
-		}
-
-		vd := fd.Enum().Values().ByNumber(v.Enum())
-		if vd == nil || w.EnumsAsInts {
-			w.write(strconv.Itoa(int(v.Enum())))
-		} else {
-			w.write(`"` + string(vd.Name()) + `"`)
-		}
-		return nil
-	default:
-		switch v.Interface().(type) {
-		case float32, float64:
-			switch {
-			case math.IsInf(v.Float(), +1):
-				w.write(`"Infinity"`)
-				return nil
-			case math.IsInf(v.Float(), -1):
-				w.write(`"-Infinity"`)
-				return nil
-			case math.IsNaN(v.Float()):
-				w.write(`"NaN"`)
-				return nil
-			}
-		case int64, uint64:
-			w.write(fmt.Sprintf(`"%d"`, v.Interface()))
-			return nil
-		}
-
-		b, err := json.Marshal(v.Interface())
-		if err != nil {
-			return err
-		}
-		w.write(string(b))
-		return nil
-	}
-}
diff --git a/vendor/github.com/golang/protobuf/jsonpb/json.go b/vendor/github.com/golang/protobuf/jsonpb/json.go
deleted file mode 100644
index 480e2448..00000000
--- a/vendor/github.com/golang/protobuf/jsonpb/json.go
+++ /dev/null
@@ -1,69 +0,0 @@
-// Copyright 2015 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-// Package jsonpb provides functionality to marshal and unmarshal between a
-// protocol buffer message and JSON. It follows the specification at
-// https://developers.google.com/protocol-buffers/docs/proto3#json.
-//
-// Do not rely on the default behavior of the standard encoding/json package
-// when called on generated message types as it does not operate correctly.
-//
-// Deprecated: Use the "google.golang.org/protobuf/encoding/protojson"
-// package instead.
-package jsonpb
-
-import (
-	"github.com/golang/protobuf/proto"
-	"google.golang.org/protobuf/reflect/protoreflect"
-	"google.golang.org/protobuf/reflect/protoregistry"
-	"google.golang.org/protobuf/runtime/protoimpl"
-)
-
-// AnyResolver takes a type URL, present in an Any message,
-// and resolves it into an instance of the associated message.
-type AnyResolver interface {
-	Resolve(typeURL string) (proto.Message, error)
-}
-
-type anyResolver struct{ AnyResolver }
-
-func (r anyResolver) FindMessageByName(message protoreflect.FullName) (protoreflect.MessageType, error) {
-	return r.FindMessageByURL(string(message))
-}
-
-func (r anyResolver) FindMessageByURL(url string) (protoreflect.MessageType, error) {
-	m, err := r.Resolve(url)
-	if err != nil {
-		return nil, err
-	}
-	return protoimpl.X.MessageTypeOf(m), nil
-}
-
-func (r anyResolver) FindExtensionByName(field protoreflect.FullName) (protoreflect.ExtensionType, error) {
-	return protoregistry.GlobalTypes.FindExtensionByName(field)
-}
-
-func (r anyResolver) FindExtensionByNumber(message protoreflect.FullName, field protoreflect.FieldNumber) (protoreflect.ExtensionType, error) {
-	return protoregistry.GlobalTypes.FindExtensionByNumber(message, field)
-}
-
-func wellKnownType(s protoreflect.FullName) string {
-	if s.Parent() == "google.protobuf" {
-		switch s.Name() {
-		case "Empty", "Any",
-			"BoolValue", "BytesValue", "StringValue",
-			"Int32Value", "UInt32Value", "FloatValue",
-			"Int64Value", "UInt64Value", "DoubleValue",
-			"Duration", "Timestamp",
-			"NullValue", "Struct", "Value", "ListValue":
-			return string(s.Name())
-		}
-	}
-	return ""
-}
-
-func isMessageSet(md protoreflect.MessageDescriptor) bool {
-	ms, ok := md.(interface{ IsMessageSet() bool })
-	return ok && ms.IsMessageSet()
-}
diff --git a/vendor/github.com/google/uuid/CHANGELOG.md b/vendor/github.com/google/uuid/CHANGELOG.md
index c9fb829d..7ec5ac7e 100644
--- a/vendor/github.com/google/uuid/CHANGELOG.md
+++ b/vendor/github.com/google/uuid/CHANGELOG.md
@@ -1,5 +1,18 @@
 # Changelog
 
+## [1.6.0](https://github.com/google/uuid/compare/v1.5.0...v1.6.0) (2024-01-16)
+
+
+### Features
+
+* add Max UUID constant ([#149](https://github.com/google/uuid/issues/149)) ([c58770e](https://github.com/google/uuid/commit/c58770eb495f55fe2ced6284f93c5158a62e53e3))
+
+
+### Bug Fixes
+
+* fix typo in version 7 uuid documentation ([#153](https://github.com/google/uuid/issues/153)) ([016b199](https://github.com/google/uuid/commit/016b199544692f745ffc8867b914129ecb47ef06))
+* Monotonicity in UUIDv7 ([#150](https://github.com/google/uuid/issues/150)) ([a2b2b32](https://github.com/google/uuid/commit/a2b2b32373ff0b1a312b7fdf6d38a977099698a6))
+
 ## [1.5.0](https://github.com/google/uuid/compare/v1.4.0...v1.5.0) (2023-12-12)
 
 
diff --git a/vendor/github.com/google/uuid/hash.go b/vendor/github.com/google/uuid/hash.go
index b404f4be..dc60082d 100644
--- a/vendor/github.com/google/uuid/hash.go
+++ b/vendor/github.com/google/uuid/hash.go
@@ -17,6 +17,12 @@ var (
 	NameSpaceOID  = Must(Parse("6ba7b812-9dad-11d1-80b4-00c04fd430c8"))
 	NameSpaceX500 = Must(Parse("6ba7b814-9dad-11d1-80b4-00c04fd430c8"))
 	Nil           UUID // empty UUID, all zeros
+
+	// The Max UUID is special form of UUID that is specified to have all 128 bits set to 1.
+	Max = UUID{
+		0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
+		0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
+	}
 )
 
 // NewHash returns a new UUID derived from the hash of space concatenated with
diff --git a/vendor/github.com/google/uuid/version7.go b/vendor/github.com/google/uuid/version7.go
index ba9dd5eb..3167b643 100644
--- a/vendor/github.com/google/uuid/version7.go
+++ b/vendor/github.com/google/uuid/version7.go
@@ -44,7 +44,7 @@ func NewV7FromReader(r io.Reader) (UUID, error) {
 
 // makeV7 fill 48 bits time (uuid[0] - uuid[5]), set version b0111 (uuid[6])
 // uuid[8] already has the right version number (Variant is 10)
-// see function  NewV7 and NewV7FromReader
+// see function NewV7 and NewV7FromReader
 func makeV7(uuid []byte) {
 	/*
 		 0                   1                   2                   3
@@ -52,7 +52,7 @@ func makeV7(uuid []byte) {
 		+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 		|                           unix_ts_ms                          |
 		+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-		|          unix_ts_ms           |  ver  |       rand_a          |
+		|          unix_ts_ms           |  ver  |  rand_a (12 bit seq)  |
 		+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 		|var|                        rand_b                             |
 		+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
@@ -61,7 +61,7 @@ func makeV7(uuid []byte) {
 	*/
 	_ = uuid[15] // bounds check
 
-	t := timeNow().UnixMilli()
+	t, s := getV7Time()
 
 	uuid[0] = byte(t >> 40)
 	uuid[1] = byte(t >> 32)
@@ -70,6 +70,35 @@ func makeV7(uuid []byte) {
 	uuid[4] = byte(t >> 8)
 	uuid[5] = byte(t)
 
-	uuid[6] = 0x70 | (uuid[6] & 0x0F)
-	// uuid[8] has already has right version
+	uuid[6] = 0x70 | (0x0F & byte(s>>8))
+	uuid[7] = byte(s)
+}
+
+// lastV7time is the last time we returned stored as:
+//
+//	52 bits of time in milliseconds since epoch
+//	12 bits of (fractional nanoseconds) >> 8
+var lastV7time int64
+
+const nanoPerMilli = 1000000
+
+// getV7Time returns the time in milliseconds and nanoseconds / 256.
+// The returned (milli << 12 + seq) is guarenteed to be greater than
+// (milli << 12 + seq) returned by any previous call to getV7Time.
+func getV7Time() (milli, seq int64) {
+	timeMu.Lock()
+	defer timeMu.Unlock()
+
+	nano := timeNow().UnixNano()
+	milli = nano / nanoPerMilli
+	// Sequence number is between 0 and 3906 (nanoPerMilli>>8)
+	seq = (nano - milli*nanoPerMilli) >> 8
+	now := milli<<12 + seq
+	if now <= lastV7time {
+		now = lastV7time + 1
+		milli = now >> 12
+		seq = now & 0xfff
+	}
+	lastV7time = now
+	return milli, seq
 }
diff --git a/vendor/google.golang.org/grpc/binarylog/grpc_binarylog_v1/binarylog.pb.go b/vendor/google.golang.org/grpc/binarylog/grpc_binarylog_v1/binarylog.pb.go
index e9e97d45..856c75dd 100644
--- a/vendor/google.golang.org/grpc/binarylog/grpc_binarylog_v1/binarylog.pb.go
+++ b/vendor/google.golang.org/grpc/binarylog/grpc_binarylog_v1/binarylog.pb.go
@@ -18,8 +18,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.31.0
-// 	protoc        v4.22.0
+// 	protoc-gen-go v1.32.0
+// 	protoc        v4.25.2
 // source: grpc/binlog/v1/binarylog.proto
 
 package grpc_binarylog_v1
diff --git a/vendor/google.golang.org/grpc/encoding/proto/proto.go b/vendor/google.golang.org/grpc/encoding/proto/proto.go
index 0ee3d3ba..66d5cdf0 100644
--- a/vendor/google.golang.org/grpc/encoding/proto/proto.go
+++ b/vendor/google.golang.org/grpc/encoding/proto/proto.go
@@ -23,8 +23,9 @@ package proto
 import (
 	"fmt"
 
-	"github.com/golang/protobuf/proto"
 	"google.golang.org/grpc/encoding"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/protoadapt"
 )
 
 // Name is the name registered for the proto compressor.
@@ -38,21 +39,34 @@ func init() {
 type codec struct{}
 
 func (codec) Marshal(v any) ([]byte, error) {
-	vv, ok := v.(proto.Message)
-	if !ok {
+	vv := messageV2Of(v)
+	if vv == nil {
 		return nil, fmt.Errorf("failed to marshal, message is %T, want proto.Message", v)
 	}
+
 	return proto.Marshal(vv)
 }
 
 func (codec) Unmarshal(data []byte, v any) error {
-	vv, ok := v.(proto.Message)
-	if !ok {
+	vv := messageV2Of(v)
+	if vv == nil {
 		return fmt.Errorf("failed to unmarshal, message is %T, want proto.Message", v)
 	}
+
 	return proto.Unmarshal(data, vv)
 }
 
+func messageV2Of(v any) proto.Message {
+	switch v := v.(type) {
+	case protoadapt.MessageV1:
+		return protoadapt.MessageV2Of(v)
+	case protoadapt.MessageV2:
+		return v
+	}
+
+	return nil
+}
+
 func (codec) Name() string {
 	return Name
 }
diff --git a/vendor/google.golang.org/grpc/health/grpc_health_v1/health.pb.go b/vendor/google.golang.org/grpc/health/grpc_health_v1/health.pb.go
index 24299efd..5bf880d4 100644
--- a/vendor/google.golang.org/grpc/health/grpc_health_v1/health.pb.go
+++ b/vendor/google.golang.org/grpc/health/grpc_health_v1/health.pb.go
@@ -17,8 +17,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.31.0
-// 	protoc        v4.22.0
+// 	protoc-gen-go v1.32.0
+// 	protoc        v4.25.2
 // source: grpc/health/v1/health.proto
 
 package grpc_health_v1
diff --git a/vendor/google.golang.org/grpc/health/grpc_health_v1/health_grpc.pb.go b/vendor/google.golang.org/grpc/health/grpc_health_v1/health_grpc.pb.go
index 4439cda0..4c46c098 100644
--- a/vendor/google.golang.org/grpc/health/grpc_health_v1/health_grpc.pb.go
+++ b/vendor/google.golang.org/grpc/health/grpc_health_v1/health_grpc.pb.go
@@ -18,7 +18,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.3.0
-// - protoc             v4.22.0
+// - protoc             v4.25.2
 // source: grpc/health/v1/health.proto
 
 package grpc_health_v1
diff --git a/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go b/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go
index 0f31274a..e8456a77 100644
--- a/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go
+++ b/vendor/google.golang.org/grpc/internal/binarylog/method_logger.go
@@ -25,11 +25,12 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/golang/protobuf/proto"
-	"github.com/golang/protobuf/ptypes"
 	binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type callIDGenerator struct {
@@ -88,7 +89,7 @@ func NewTruncatingMethodLogger(h, m uint64) *TruncatingMethodLogger {
 // in TruncatingMethodLogger as possible.
 func (ml *TruncatingMethodLogger) Build(c LogEntryConfig) *binlogpb.GrpcLogEntry {
 	m := c.toProto()
-	timestamp, _ := ptypes.TimestampProto(time.Now())
+	timestamp := timestamppb.Now()
 	m.Timestamp = timestamp
 	m.CallId = ml.callID
 	m.SequenceIdWithinCall = ml.idWithinCallGen.next()
@@ -178,7 +179,7 @@ func (c *ClientHeader) toProto() *binlogpb.GrpcLogEntry {
 		Authority:  c.Authority,
 	}
 	if c.Timeout > 0 {
-		clientHeader.Timeout = ptypes.DurationProto(c.Timeout)
+		clientHeader.Timeout = durationpb.New(c.Timeout)
 	}
 	ret := &binlogpb.GrpcLogEntry{
 		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_HEADER,
diff --git a/vendor/google.golang.org/grpc/internal/binarylog/sink.go b/vendor/google.golang.org/grpc/internal/binarylog/sink.go
index 264de387..9ea598b1 100644
--- a/vendor/google.golang.org/grpc/internal/binarylog/sink.go
+++ b/vendor/google.golang.org/grpc/internal/binarylog/sink.go
@@ -25,8 +25,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/golang/protobuf/proto"
 	binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
+	"google.golang.org/protobuf/proto"
 )
 
 var (
diff --git a/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
index aa97273e..0126d6b5 100644
--- a/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
+++ b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand.go
@@ -1,3 +1,8 @@
+//go:build !go1.21
+
+// TODO: when this file is deleted (after Go 1.20 support is dropped), delete
+// all of grpcrand and call the rand package directly.
+
 /*
  *
  * Copyright 2018 gRPC authors.
diff --git a/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand_go1.21.go b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand_go1.21.go
new file mode 100644
index 00000000..c37299af
--- /dev/null
+++ b/vendor/google.golang.org/grpc/internal/grpcrand/grpcrand_go1.21.go
@@ -0,0 +1,73 @@
+//go:build go1.21
+
+/*
+ *
+ * Copyright 2024 gRPC authors.
+ *
+ * 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 grpcrand implements math/rand functions in a concurrent-safe way
+// with a global random source, independent of math/rand's global source.
+package grpcrand
+
+import "math/rand"
+
+// This implementation will be used for Go version 1.21 or newer.
+// For older versions, the original implementation with mutex will be used.
+
+// Int implements rand.Int on the grpcrand global source.
+func Int() int {
+	return rand.Int()
+}
+
+// Int63n implements rand.Int63n on the grpcrand global source.
+func Int63n(n int64) int64 {
+	return rand.Int63n(n)
+}
+
+// Intn implements rand.Intn on the grpcrand global source.
+func Intn(n int) int {
+	return rand.Intn(n)
+}
+
+// Int31n implements rand.Int31n on the grpcrand global source.
+func Int31n(n int32) int32 {
+	return rand.Int31n(n)
+}
+
+// Float64 implements rand.Float64 on the grpcrand global source.
+func Float64() float64 {
+	return rand.Float64()
+}
+
+// Uint64 implements rand.Uint64 on the grpcrand global source.
+func Uint64() uint64 {
+	return rand.Uint64()
+}
+
+// Uint32 implements rand.Uint32 on the grpcrand global source.
+func Uint32() uint32 {
+	return rand.Uint32()
+}
+
+// ExpFloat64 implements rand.ExpFloat64 on the grpcrand global source.
+func ExpFloat64() float64 {
+	return rand.ExpFloat64()
+}
+
+// Shuffle implements rand.Shuffle on the grpcrand global source.
+var Shuffle = func(n int, f func(int, int)) {
+	rand.Shuffle(n, f)
+}
diff --git a/vendor/google.golang.org/grpc/internal/pretty/pretty.go b/vendor/google.golang.org/grpc/internal/pretty/pretty.go
index 70331913..52cfab1b 100644
--- a/vendor/google.golang.org/grpc/internal/pretty/pretty.go
+++ b/vendor/google.golang.org/grpc/internal/pretty/pretty.go
@@ -24,7 +24,6 @@ import (
 	"encoding/json"
 	"fmt"
 
-	"github.com/golang/protobuf/jsonpb"
 	protov1 "github.com/golang/protobuf/proto"
 	"google.golang.org/protobuf/encoding/protojson"
 	protov2 "google.golang.org/protobuf/proto"
@@ -38,15 +37,15 @@ const jsonIndent = "  "
 func ToJSON(e any) string {
 	switch ee := e.(type) {
 	case protov1.Message:
-		mm := jsonpb.Marshaler{Indent: jsonIndent}
-		ret, err := mm.MarshalToString(ee)
+		mm := protojson.MarshalOptions{Indent: jsonIndent}
+		ret, err := mm.Marshal(protov1.MessageV2(ee))
 		if err != nil {
 			// This may fail for proto.Anys, e.g. for xDS v2, LDS, the v2
 			// messages are not imported, and this will fail because the message
 			// is not found.
 			return fmt.Sprintf("%+v", ee)
 		}
-		return ret
+		return string(ret)
 	case protov2.Message:
 		mm := protojson.MarshalOptions{
 			Multiline: true,
diff --git a/vendor/google.golang.org/grpc/internal/status/status.go b/vendor/google.golang.org/grpc/internal/status/status.go
index 03ef2fed..c7dbc820 100644
--- a/vendor/google.golang.org/grpc/internal/status/status.go
+++ b/vendor/google.golang.org/grpc/internal/status/status.go
@@ -31,10 +31,11 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/golang/protobuf/proto"
-	"github.com/golang/protobuf/ptypes"
 	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/protoadapt"
+	"google.golang.org/protobuf/types/known/anypb"
 )
 
 // Status represents an RPC status code, message, and details.  It is immutable
@@ -130,14 +131,14 @@ func (s *Status) Err() error {
 
 // WithDetails returns a new status with the provided details messages appended to the status.
 // If any errors are encountered, it returns nil and the first error encountered.
-func (s *Status) WithDetails(details ...proto.Message) (*Status, error) {
+func (s *Status) WithDetails(details ...protoadapt.MessageV1) (*Status, error) {
 	if s.Code() == codes.OK {
 		return nil, errors.New("no error details for status with code OK")
 	}
 	// s.Code() != OK implies that s.Proto() != nil.
 	p := s.Proto()
 	for _, detail := range details {
-		any, err := ptypes.MarshalAny(detail)
+		any, err := anypb.New(protoadapt.MessageV2Of(detail))
 		if err != nil {
 			return nil, err
 		}
@@ -154,12 +155,12 @@ func (s *Status) Details() []any {
 	}
 	details := make([]any, 0, len(s.s.Details))
 	for _, any := range s.s.Details {
-		detail := &ptypes.DynamicAny{}
-		if err := ptypes.UnmarshalAny(any, detail); err != nil {
+		detail, err := any.UnmarshalNew()
+		if err != nil {
 			details = append(details, err)
 			continue
 		}
-		details = append(details, detail.Message)
+		details = append(details, detail)
 	}
 	return details
 }
diff --git a/vendor/google.golang.org/grpc/internal/transport/controlbuf.go b/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
index b330cced..83c38298 100644
--- a/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
+++ b/vendor/google.golang.org/grpc/internal/transport/controlbuf.go
@@ -535,8 +535,8 @@ const minBatchSize = 1000
 // size is too low to give stream goroutines a chance to fill it up.
 //
 // Upon exiting, if the error causing the exit is not an I/O error, run()
-// flushes and closes the underlying connection.  Otherwise, the connection is
-// left open to allow the I/O error to be encountered by the reader instead.
+// flushes the underlying connection.  The connection is always left open to
+// allow different closing behavior on the client and server.
 func (l *loopyWriter) run() (err error) {
 	defer func() {
 		if l.logger.V(logLevel) {
@@ -544,7 +544,6 @@ func (l *loopyWriter) run() (err error) {
 		}
 		if !isIOError(err) {
 			l.framer.writer.Flush()
-			l.conn.Close()
 		}
 		l.cbuf.finish()
 	}()
diff --git a/vendor/google.golang.org/grpc/internal/transport/handler_server.go b/vendor/google.golang.org/grpc/internal/transport/handler_server.go
index a9d70e2a..bd39ff9a 100644
--- a/vendor/google.golang.org/grpc/internal/transport/handler_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/handler_server.go
@@ -35,7 +35,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/golang/protobuf/proto"
 	"golang.org/x/net/http2"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
@@ -45,6 +44,7 @@ import (
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 // NewServerHandlerTransport returns a ServerTransport handling gRPC from
diff --git a/vendor/google.golang.org/grpc/internal/transport/http2_client.go b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
index c33ac596..eff87996 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_client.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_client.go
@@ -451,7 +451,13 @@ func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	}
 	go func() {
 		t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst, t.conn, t.logger)
-		t.loopy.run()
+		if err := t.loopy.run(); !isIOError(err) {
+			// Immediately close the connection, as the loopy writer returns
+			// when there are no more active streams and we were draining (the
+			// server sent a GOAWAY).  For I/O errors, the reader will hit it
+			// after draining any remaining incoming data.
+			t.conn.Close()
+		}
 		close(t.writerDone)
 	}()
 	return t, nil
diff --git a/vendor/google.golang.org/grpc/internal/transport/http2_server.go b/vendor/google.golang.org/grpc/internal/transport/http2_server.go
index f6bac0e8..3839c1ad 100644
--- a/vendor/google.golang.org/grpc/internal/transport/http2_server.go
+++ b/vendor/google.golang.org/grpc/internal/transport/http2_server.go
@@ -32,13 +32,13 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/golang/protobuf/proto"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcutil"
 	"google.golang.org/grpc/internal/pretty"
 	"google.golang.org/grpc/internal/syscall"
+	"google.golang.org/protobuf/proto"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
@@ -322,8 +322,24 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport,
 	go func() {
 		t.loopy = newLoopyWriter(serverSide, t.framer, t.controlBuf, t.bdpEst, t.conn, t.logger)
 		t.loopy.ssGoAwayHandler = t.outgoingGoAwayHandler
-		t.loopy.run()
+		err := t.loopy.run()
 		close(t.loopyWriterDone)
+		if !isIOError(err) {
+			// Close the connection if a non-I/O error occurs (for I/O errors
+			// the reader will also encounter the error and close).  Wait 1
+			// second before closing the connection, or when the reader is done
+			// (i.e. the client already closed the connection or a connection
+			// error occurred).  This avoids the potential problem where there
+			// is unread data on the receive side of the connection, which, if
+			// closed, would lead to a TCP RST instead of FIN, and the client
+			// encountering errors.  For more info:
+			// https://github.com/grpc/grpc-go/issues/5358
+			select {
+			case <-t.readerDone:
+			case <-time.After(time.Second):
+			}
+			t.conn.Close()
+		}
 	}()
 	go t.keepalive()
 	return t, nil
@@ -609,8 +625,8 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 // traceCtx attaches trace to ctx and returns the new context.
 func (t *http2Server) HandleStreams(ctx context.Context, handle func(*Stream)) {
 	defer func() {
-		<-t.loopyWriterDone
 		close(t.readerDone)
+		<-t.loopyWriterDone
 	}()
 	for {
 		t.controlBuf.throttle()
@@ -636,10 +652,6 @@ func (t *http2Server) HandleStreams(ctx context.Context, handle func(*Stream)) {
 				}
 				continue
 			}
-			if err == io.EOF || err == io.ErrUnexpectedEOF {
-				t.Close(err)
-				return
-			}
 			t.Close(err)
 			return
 		}
@@ -1329,6 +1341,7 @@ func (t *http2Server) outgoingGoAwayHandler(g *goAway) (bool, error) {
 		if err := t.framer.fr.WriteGoAway(sid, g.code, g.debugData); err != nil {
 			return false, err
 		}
+		t.framer.writer.Flush()
 		if retErr != nil {
 			return false, retErr
 		}
@@ -1349,7 +1362,7 @@ func (t *http2Server) outgoingGoAwayHandler(g *goAway) (bool, error) {
 		return false, err
 	}
 	go func() {
-		timer := time.NewTimer(time.Minute)
+		timer := time.NewTimer(5 * time.Second)
 		defer timer.Stop()
 		select {
 		case <-t.drainEvent.Done():
diff --git a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection.pb.go b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection.pb.go
index 6f5c786b..8953c9d8 100644
--- a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection.pb.go
+++ b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection.pb.go
@@ -21,8 +21,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.31.0
-// 	protoc        v4.22.0
+// 	protoc-gen-go v1.32.0
+// 	protoc        v4.25.2
 // source: grpc/reflection/v1/reflection.proto
 
 package grpc_reflection_v1
diff --git a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection_grpc.pb.go b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection_grpc.pb.go
index 62b56a8b..d6cdd5b5 100644
--- a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection_grpc.pb.go
+++ b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1/reflection_grpc.pb.go
@@ -22,7 +22,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.3.0
-// - protoc             v4.22.0
+// - protoc             v4.25.2
 // source: grpc/reflection/v1/reflection.proto
 
 package grpc_reflection_v1
diff --git a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection.pb.go b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection.pb.go
index 69fbfb62..929733e7 100644
--- a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection.pb.go
+++ b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection.pb.go
@@ -18,8 +18,8 @@
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.31.0
-// 	protoc        v4.22.0
+// 	protoc-gen-go v1.32.0
+// 	protoc        v4.25.2
 // grpc/reflection/v1alpha/reflection.proto is a deprecated file.
 
 package grpc_reflection_v1alpha
diff --git a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection_grpc.pb.go b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection_grpc.pb.go
index 367a029b..ef691406 100644
--- a/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection_grpc.pb.go
+++ b/vendor/google.golang.org/grpc/reflection/grpc_reflection_v1alpha/reflection_grpc.pb.go
@@ -19,7 +19,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.3.0
-// - protoc             v4.22.0
+// - protoc             v4.25.2
 // grpc/reflection/v1alpha/reflection.proto is a deprecated file.
 
 package grpc_reflection_v1alpha
diff --git a/vendor/google.golang.org/grpc/rpc_util.go b/vendor/google.golang.org/grpc/rpc_util.go
index a4b6bc68..d17ede0f 100644
--- a/vendor/google.golang.org/grpc/rpc_util.go
+++ b/vendor/google.golang.org/grpc/rpc_util.go
@@ -189,6 +189,20 @@ type EmptyCallOption struct{}
 func (EmptyCallOption) before(*callInfo) error      { return nil }
 func (EmptyCallOption) after(*callInfo, *csAttempt) {}
 
+// StaticMethod returns a CallOption which specifies that a call is being made
+// to a method that is static, which means the method is known at compile time
+// and doesn't change at runtime. This can be used as a signal to stats plugins
+// that this method is safe to include as a key to a measurement.
+func StaticMethod() CallOption {
+	return StaticMethodCallOption{}
+}
+
+// StaticMethodCallOption is a CallOption that specifies that a call comes
+// from a static method.
+type StaticMethodCallOption struct {
+	EmptyCallOption
+}
+
 // Header returns a CallOptions that retrieves the header metadata
 // for a unary RPC.
 func Header(md *metadata.MD) CallOption {
@@ -958,6 +972,7 @@ const (
 	SupportPackageIsVersion5 = true
 	SupportPackageIsVersion6 = true
 	SupportPackageIsVersion7 = true
+	SupportPackageIsVersion8 = true
 )
 
 const grpcUA = "grpc-go/" + Version
diff --git a/vendor/google.golang.org/grpc/server.go b/vendor/google.golang.org/grpc/server.go
index e89c5ac6..0bf5c78b 100644
--- a/vendor/google.golang.org/grpc/server.go
+++ b/vendor/google.golang.org/grpc/server.go
@@ -33,8 +33,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"golang.org/x/net/trace"
-
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/encoding"
@@ -131,7 +129,7 @@ type Server struct {
 	drain    bool
 	cv       *sync.Cond              // signaled when connections close for GracefulStop
 	services map[string]*serviceInfo // service name -> service info
-	events   trace.EventLog
+	events   traceEventLog
 
 	quit               *grpcsync.Event
 	done               *grpcsync.Event
@@ -670,7 +668,7 @@ func NewServer(opt ...ServerOption) *Server {
 	s.cv = sync.NewCond(&s.mu)
 	if EnableTracing {
 		_, file, line, _ := runtime.Caller(1)
-		s.events = trace.NewEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
+		s.events = newTraceEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
 	}
 
 	if s.opts.numServerWorkers > 0 {
@@ -1734,8 +1732,8 @@ func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Str
 	ctx = contextWithServer(ctx, s)
 	var ti *traceInfo
 	if EnableTracing {
-		tr := trace.New("grpc.Recv."+methodFamily(stream.Method()), stream.Method())
-		ctx = trace.NewContext(ctx, tr)
+		tr := newTrace("grpc.Recv."+methodFamily(stream.Method()), stream.Method())
+		ctx = newTraceContext(ctx, tr)
 		ti = &traceInfo{
 			tr: tr,
 			firstLine: firstLine{
diff --git a/vendor/google.golang.org/grpc/stream.go b/vendor/google.golang.org/grpc/stream.go
index d621f52b..814e9983 100644
--- a/vendor/google.golang.org/grpc/stream.go
+++ b/vendor/google.golang.org/grpc/stream.go
@@ -27,7 +27,6 @@ import (
 	"sync"
 	"time"
 
-	"golang.org/x/net/trace"
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/encoding"
@@ -431,7 +430,7 @@ func (cs *clientStream) newAttemptLocked(isTransparent bool) (*csAttempt, error)
 	var trInfo *traceInfo
 	if EnableTracing {
 		trInfo = &traceInfo{
-			tr: trace.New("grpc.Sent."+methodFamily(method), method),
+			tr: newTrace("grpc.Sent."+methodFamily(method), method),
 			firstLine: firstLine{
 				client: true,
 			},
@@ -440,7 +439,7 @@ func (cs *clientStream) newAttemptLocked(isTransparent bool) (*csAttempt, error)
 			trInfo.firstLine.deadline = time.Until(deadline)
 		}
 		trInfo.tr.LazyLog(&trInfo.firstLine, false)
-		ctx = trace.NewContext(ctx, trInfo.tr)
+		ctx = newTraceContext(ctx, trInfo.tr)
 	}
 
 	if cs.cc.parsedTarget.URL.Scheme == internal.GRPCResolverSchemeExtraMetadata {
diff --git a/vendor/google.golang.org/grpc/trace.go b/vendor/google.golang.org/grpc/trace.go
index 9ded7932..10f4f798 100644
--- a/vendor/google.golang.org/grpc/trace.go
+++ b/vendor/google.golang.org/grpc/trace.go
@@ -26,8 +26,6 @@ import (
 	"strings"
 	"sync"
 	"time"
-
-	"golang.org/x/net/trace"
 )
 
 // EnableTracing controls whether to trace RPCs using the golang.org/x/net/trace package.
@@ -44,9 +42,31 @@ func methodFamily(m string) string {
 	return m
 }
 
+// traceEventLog mirrors golang.org/x/net/trace.EventLog.
+//
+// It exists in order to avoid importing x/net/trace on grpcnotrace builds.
+type traceEventLog interface {
+	Printf(format string, a ...any)
+	Errorf(format string, a ...any)
+	Finish()
+}
+
+// traceLog mirrors golang.org/x/net/trace.Trace.
+//
+// It exists in order to avoid importing x/net/trace on grpcnotrace builds.
+type traceLog interface {
+	LazyLog(x fmt.Stringer, sensitive bool)
+	LazyPrintf(format string, a ...any)
+	SetError()
+	SetRecycler(f func(any))
+	SetTraceInfo(traceID, spanID uint64)
+	SetMaxEvents(m int)
+	Finish()
+}
+
 // traceInfo contains tracing information for an RPC.
 type traceInfo struct {
-	tr        trace.Trace
+	tr        traceLog
 	firstLine firstLine
 }
 
diff --git a/vendor/google.golang.org/grpc/trace_notrace.go b/vendor/google.golang.org/grpc/trace_notrace.go
new file mode 100644
index 00000000..1da3a230
--- /dev/null
+++ b/vendor/google.golang.org/grpc/trace_notrace.go
@@ -0,0 +1,52 @@
+//go:build grpcnotrace
+
+/*
+ *
+ * Copyright 2024 gRPC authors.
+ *
+ * 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 grpc
+
+// grpcnotrace can be used to avoid importing golang.org/x/net/trace, which in
+// turn enables binaries using gRPC-Go for dead code elimination, which can
+// yield 10-15% improvements in binary size when tracing is not needed.
+
+import (
+	"context"
+	"fmt"
+)
+
+type notrace struct{}
+
+func (notrace) LazyLog(x fmt.Stringer, sensitive bool) {}
+func (notrace) LazyPrintf(format string, a ...any)     {}
+func (notrace) SetError()                              {}
+func (notrace) SetRecycler(f func(any))                {}
+func (notrace) SetTraceInfo(traceID, spanID uint64)    {}
+func (notrace) SetMaxEvents(m int)                     {}
+func (notrace) Finish()                                {}
+
+func newTrace(family, title string) traceLog {
+	return notrace{}
+}
+
+func newTraceContext(ctx context.Context, tr traceLog) context.Context {
+	return ctx
+}
+
+func newTraceEventLog(family, title string) traceEventLog {
+	return nil
+}
diff --git a/vendor/google.golang.org/grpc/trace_withtrace.go b/vendor/google.golang.org/grpc/trace_withtrace.go
new file mode 100644
index 00000000..88d6e857
--- /dev/null
+++ b/vendor/google.golang.org/grpc/trace_withtrace.go
@@ -0,0 +1,39 @@
+//go:build !grpcnotrace
+
+/*
+ *
+ * Copyright 2024 gRPC authors.
+ *
+ * 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 grpc
+
+import (
+	"context"
+
+	t "golang.org/x/net/trace"
+)
+
+func newTrace(family, title string) traceLog {
+	return t.New(family, title)
+}
+
+func newTraceContext(ctx context.Context, tr traceLog) context.Context {
+	return t.NewContext(ctx, tr)
+}
+
+func newTraceEventLog(family, title string) traceEventLog {
+	return t.NewEventLog(family, title)
+}
diff --git a/vendor/google.golang.org/grpc/version.go b/vendor/google.golang.org/grpc/version.go
index 1ad1ba2a..df85a021 100644
--- a/vendor/google.golang.org/grpc/version.go
+++ b/vendor/google.golang.org/grpc/version.go
@@ -19,4 +19,4 @@
 package grpc
 
 // Version is the current grpc version.
-const Version = "1.61.0"
+const Version = "1.62.0"
diff --git a/vendor/google.golang.org/grpc/vet.sh b/vendor/google.golang.org/grpc/vet.sh
index 5da38a40..7a33c215 100644
--- a/vendor/google.golang.org/grpc/vet.sh
+++ b/vendor/google.golang.org/grpc/vet.sh
@@ -41,7 +41,7 @@ if [[ "$1" = "-install" ]]; then
   popd
   if [[ -z "${VET_SKIP_PROTO}" ]]; then
     if [[ "${GITHUB_ACTIONS}" = "true" ]]; then
-      PROTOBUF_VERSION=22.0 # a.k.a v4.22.0 in pb.go files.
+      PROTOBUF_VERSION=25.2 # a.k.a. v4.22.0 in pb.go files.
       PROTOC_FILENAME=protoc-${PROTOBUF_VERSION}-linux-x86_64.zip
       pushd /home/runner/go
       wget https://github.com/google/protobuf/releases/download/v${PROTOBUF_VERSION}/${PROTOC_FILENAME}
diff --git a/vendor/google.golang.org/protobuf/protoadapt/convert.go b/vendor/google.golang.org/protobuf/protoadapt/convert.go
new file mode 100644
index 00000000..ea276d15
--- /dev/null
+++ b/vendor/google.golang.org/protobuf/protoadapt/convert.go
@@ -0,0 +1,31 @@
+// Copyright 2023 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package protoadapt bridges the original and new proto APIs.
+package protoadapt
+
+import (
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/runtime/protoiface"
+	"google.golang.org/protobuf/runtime/protoimpl"
+)
+
+// MessageV1 is the original [github.com/golang/protobuf/proto.Message] type.
+type MessageV1 = protoiface.MessageV1
+
+// MessageV2 is the [google.golang.org/protobuf/proto.Message] type used by the
+// current [google.golang.org/protobuf] module, adding support for reflection.
+type MessageV2 = proto.Message
+
+// MessageV1Of converts a v2 message to a v1 message.
+// It returns nil if m is nil.
+func MessageV1Of(m MessageV2) MessageV1 {
+	return protoimpl.X.ProtoMessageV1Of(m)
+}
+
+// MessageV2Of converts a v1 message to a v2 message.
+// It returns nil if m is nil.
+func MessageV2Of(m MessageV1) MessageV2 {
+	return protoimpl.X.ProtoMessageV2Of(m)
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 72950e72..b5a26c5a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -84,7 +84,6 @@ github.com/gogo/protobuf/sortkeys
 github.com/gogo/protobuf/types
 # github.com/golang/protobuf v1.5.3
 ## explicit; go 1.9
-github.com/golang/protobuf/jsonpb
 github.com/golang/protobuf/proto
 github.com/golang/protobuf/ptypes
 github.com/golang/protobuf/ptypes/any
@@ -119,7 +118,7 @@ github.com/google/go-jsonnet/internal/program
 ## explicit; go 1.12
 github.com/google/gofuzz
 github.com/google/gofuzz/bytesource
-# github.com/google/uuid v1.5.0
+# github.com/google/uuid v1.6.0
 ## explicit
 github.com/google/uuid
 # github.com/gorilla/websocket v1.5.0
@@ -647,14 +646,14 @@ google.golang.org/appengine/internal/log
 google.golang.org/appengine/internal/remote_api
 google.golang.org/appengine/internal/urlfetch
 google.golang.org/appengine/urlfetch
-# google.golang.org/genproto/googleapis/api v0.0.0-20240102182953-50ed04b92917
+# google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80
 ## explicit; go 1.19
 google.golang.org/genproto/googleapis/api/httpbody
-# google.golang.org/genproto/googleapis/rpc v0.0.0-20240102182953-50ed04b92917
+# google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80
 ## explicit; go 1.19
 google.golang.org/genproto/googleapis/rpc/errdetails
 google.golang.org/genproto/googleapis/rpc/status
-# google.golang.org/grpc v1.61.0
+# google.golang.org/grpc v1.62.0
 ## explicit; go 1.19
 google.golang.org/grpc
 google.golang.org/grpc/attributes
@@ -738,6 +737,7 @@ google.golang.org/protobuf/internal/set
 google.golang.org/protobuf/internal/strs
 google.golang.org/protobuf/internal/version
 google.golang.org/protobuf/proto
+google.golang.org/protobuf/protoadapt
 google.golang.org/protobuf/reflect/protodesc
 google.golang.org/protobuf/reflect/protoreflect
 google.golang.org/protobuf/reflect/protoregistry


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