snyk: add rate limit handling to audit_logs and issues data streams by efd6 · Pull Request #16184 · elastic/integrations

efd6 · 2025-12-02T02:26:24Z

Proposed commit message

snyk: add rate limit handling to audit_logs and issues data streams

This enhancement implements comprehensive rate limiting for Snyk API calls
to prevent quota exhaustion and improve collection reliability.

Checklist

I have reviewed tips for building integrations and this pull request is aligned with them.
I have verified that all data streams collect metrics or logs.
I have added an entry to my package's changelog.yml file.
I have verified that Kibana version constraints are current according to guidelines.
I have verified that any added dashboard complies with Kibana's Dashboard good practices

Author's Checklist

[ ]

How to test this PR locally

The system tests exercise the rate limit handling path and you will see this in the pattern of event publications in the issues data stream when tests are run with -v.

To see the rate limit values and the extracted headers, make the following changes to the CEL programs

Details

diff --git a/packages/snyk/data_stream/audit_logs/agent/stream/cel.yml.hbs b/packages/snyk/data_stream/audit_logs/agent/stream/cel.yml.hbs
index 2a3e949a55..c3270bfd5b 100644
--- a/packages/snyk/data_stream/audit_logs/agent/stream/cel.yml.hbs
+++ b/packages/snyk/data_stream/audit_logs/agent/stream/cel.yml.hbs
@@ -124,14 +124,14 @@ program: |-
                 // Threading rate-limit results obtained here through to the final
                 // result is not tenable, so we do not do any work to allow them
                 // to be interpreted by the input and just drop them.
-                rate_limit(
+                debug("RATE_LIMIT_1", rate_limit(
                   headers,
                   "X-Ratelimit",
                   false,
                   true,
                   duration(string(headers[?"X-Ratelimit-Reset"][0].orValue("1")) + "s"),
                   0
-                )
+                ))
               ).drop(["rate", "next"])
             )
           ).as(resp, (resp.StatusCode == 200) ?
@@ -198,14 +198,14 @@ program: |-
           resp.with(
             resp.Header.as(headers,
               // Calculate and apply rate limits.
-              rate_limit(
+              debug("RATE_LIMIT_2", rate_limit(
                 headers,
                 "X-Ratelimit",
                 false,
                 true,
                 duration(string(headers[?"X-Ratelimit-Reset"][0].orValue("1")) + "s"),
                 0
-              )
+              ))
             ).as(rate_headers,
               {
                // Rate limit side-effects have already been applied to the
diff --git a/packages/snyk/data_stream/issues/agent/stream/cel.yml.hbs b/packages/snyk/data_stream/issues/agent/stream/cel.yml.hbs
index dfeccaa246..3f1d0ff8c3 100644
--- a/packages/snyk/data_stream/issues/agent/stream/cel.yml.hbs
+++ b/packages/snyk/data_stream/issues/agent/stream/cel.yml.hbs
@@ -134,14 +134,14 @@ program: |-
         resp.with(
           resp.Header.as(headers,
             // Calculate and apply rate limits.
-            rate_limit(
+            debug("RATE_LIMIT_1", rate_limit(
               headers,
               "X-Ratelimit",
               false,
               true,
               duration(string(headers[?"X-Ratelimit-Reset"][0].orValue("1")) + "s"),
               0
-            )
+            ))
           ).as(rate_headers,
               {
                // Rate limit side-effects have already been applied to the
@@ -195,14 +195,14 @@ program: |-
                                     // Threading rate-limit results obtained here through to the final
                                     // result is not tenable, so we do not do any work to allow them
                                     // to be interpreted by the input and just drop them.
-                                    rate_limit(
+                                    debug("RATE_LIMIT_2", rate_limit(
                                       headers,
                                       "X-Ratelimit",
                                       false,
                                       true,
                                       duration(string(headers[?"X-Ratelimit-Reset"][0].orValue("1")) + "s"),
                                       0
-                                    )
+                                    ))
                                   ).drop(["rate", "next"])
                                 )
                               ).as(resp, (resp.StatusCode == 200) ?

And examine the agent logs (either via docker logs or by getting a diagnostic).

The RATE_LIMIT_1 and RATE_LIMIT_2 tags will appear in the agent debug logs with the computed rate, burst, next, and reset fields from the rate limiter. Check that the values of these fields are sane relative to the HTTP headers which will also be present in the debug output in the message.

Related issues

Closes snyk: enhance snyk integration to properly handle rate limits #15608

Screenshots

elastic-vault-github-plugin-prod · 2025-12-02T02:49:39Z

🚀 Benchmarks report

To see the full report comment with /test benchmark fullreport

botelastic · 2026-01-01T03:48:43Z

Hi! We just realized that we haven't looked into this PR in a while. We're sorry! We're labeling this issue as Stale to make it hit our filters and make sure we get back to it as soon as possible. In the meantime, it'd be extremely helpful if you could take a look at it as well and confirm its relevance. A simple comment with a nice emoji will be enough :+1. Thank you for your contribution!

elasticmachine · 2026-01-18T22:18:40Z

Pinging @elastic/security-service-integrations (Team:Security-Service Integrations)

chrisberkhout

What do you think about testing this? I think the system test could be modified to exercise the new logic, but that's probably not worthwhile.

I think it would be nice if the commit message or at least the PR's "How to test this PR locally" section had some advice about how to manually exercise it.

I'm thinking I'd run the system test's stream config outside of a container, with extra response headers, then modify the CEL program to run in miko/mito and add some debug() calls to inspect the debugging value.

packages/snyk/data_stream/audit_logs/agent/stream/cel.yml.hbs

chrisberkhout · 2026-02-16T10:53:33Z

packages/snyk/data_stream/audit_logs/agent/stream/cel.yml.hbs

+                rate_headers.with(
+                  {
+                    // Work around inf detection in input.
+                    // If the headers are missing or rate_limit failed, rate and
+                    // next may be missing. So use optional types.
+                    ?"rate": (rate_headers.?rate == optional.of(double("Infinity"))) ? optional.of("inf") : optional.none(),
+                    ?"next": (rate_headers.?next == optional.of(double("Infinity"))) ? optional.of("inf") : optional.none(),
+                  }
+                )


This part will also only affect the debugging value, right?

I think it'd be better to combine it the the following by adding the {"rate_limit": ...} wrapper, so we have one block for side effects and one block for a debugging value.

Not quite, but you have identified a nest of bugs that are addressed in 94b456a.

efd6 · 2026-02-17T00:14:02Z

@chrisberkhout there is significant work to do to get this to merge due to conflicts. I'll ping you when it's ready.

* dynamically applied rate limits * canonical_mime_header_key overload

* fix propagation of rate_limit header where possible

* fix propagation of rate_limit header where possible * explain what is happening better * in-line rate_limit construction or drop unused values

* add rate-limit headers

This should not have been converted to a timestamp.

…omment removal

efd6 · 2026-02-17T05:45:29Z

/test

efd6 · 2026-02-17T06:03:28Z

/test

chrisberkhout

Looks good.

I put a few comments that don't need any response.

One thing left is that the "We are doing this work for the side-effects of the rate_limit call" comments are a bit misleading in the cases where the values are added to the response and returned in the final result.

Maybe some could be changed to:

We are doing this work only for the side-effects of the rate_limit call.

and the others could be removed or changed to something like:

The rate_limit call has side effects as well as generating values for the final result.

chrisberkhout · 2026-02-17T09:08:22Z

packages/snyk/data_stream/issues/agent/stream/cel.yml.hbs

@@ -144,26 +143,29 @@ program: |-
              0
            )
          ).as(rate_headers,


Could use resp.Header directly rather than as headers.

Much clearer. Done

chrisberkhout · 2026-02-17T09:31:08Z

packages/snyk/data_stream/issues/agent/stream/cel.yml.hbs

+                    ?"rate": (rate_headers.?rate == optional.of(double("Infinity"))) ? optional.of("inf") : optional.none(),
+                    ?"next": (rate_headers.?next == optional.of(double("Infinity"))) ? optional.of("inf") : optional.none(),


Just thinking...

Looks like this checks for an Infinity valued double and switches it to the string "inf", which in the input will be switched back to rate.Inf, which is the finite math.MaxFloat64 rather than infinite.

Seems like it could have been made simpler, especially in x/time/rate.

But given that's the way it is, maybe at some point rate_limit should handle avoiding infinite rates?

Yeah, this is unfortunate history. It's a conflict between between encoding/json, mito/lib, x/time/rate and cel-go. The origin of the "inf" is from JSON serialisation of math.Inf(1) that results from divisions. We could conceivably condition the return values from the rate limit extensions so that if either of these end up being infinite, it gets replaced with rate.Inf. I think this would be backwards compatible; the relevant text in the documentation is 'The map returned by the policy functions should have "rate" and "next" fields with type rate.Limit or string with the value "inf", a "burst" field with type int and a "reset" field with type time.Time in the UTC location. The semantics of "rate" and "burst" are described in the documentation for the golang.org/x/time/rate package.' This would remain true, but the second option would never happen.

Do you want to file an issue in mito?

chrisberkhout · 2026-02-17T09:45:48Z

packages/snyk/data_stream/issues/agent/stream/cel.yml.hbs

+                                    // Threading rate-limit results obtained here through to the final
+                                    // result is not tenable, so we do not do any work to allow them
+                                    // to be interpreted by the input and just drop them.


Just thinking about the existing structure of the CEL program...

Doing multiple requests per eval means you have to reduce the results with complicated logic and accept compromises regarding the feedback that can go back to the input. This is a case of that.

I think the work list approach like we have now in o365 is easier to reason about and can provide better feedback and error handling. The downside is the overhead of extra evals, but I think it's usually worth it. Or maybe I'm missing something?

I think this would be worthwhile. I also think it's work for another PR.

…ue calls

elasticmachine · 2026-02-17T21:54:35Z

💚 Build Succeeded

Buildkite Build
Commit: 8509377

History

💚 Build #38323 succeeded c08cf7b
💔 Build #38321 failed c08cf7b
💔 Build #38319 failed c08cf7b
💚 Build #38314 succeeded 94b456a
💔 Build #38310 failed 14e09a1
💔 Build #38309 failed 486b884

cc @efd6

efd6 self-assigned this Dec 2, 2025

efd6 added enhancement New feature or request Integration:snyk Snyk Team:Security-Service Integrations Security Service Integrations team [elastic/security-service-integrations] labels Dec 2, 2025

botelastic bot added the Stalled label Jan 1, 2026

efd6 removed the Stalled label Jan 18, 2026

efd6 marked this pull request as ready for review January 18, 2026 22:18

efd6 requested a review from a team as a code owner January 18, 2026 22:18

chrisberkhout reviewed Feb 16, 2026

View reviewed changes

efd6 force-pushed the e24453-snyk-again branch from 486b884 to 14e09a1 Compare February 16, 2026 23:44

efd6 added 11 commits February 17, 2026 12:00

snyk: run celfmt

4e2fa9f

snyk: invert order of ternaries

89b164f

snyk: remove unnecessary bytes conversions

3bd311b

snyk: remove unnecessary arrayification in drop calls

b5b1f89

snyk: add 429 failout

ce903a9

snyk: add rate limit handling to issues data stream

7739f8f

snyk: add rate limit handling to audit_logs data stream

311ee04

snyk: bump kibana.version to support required features

1c195e0

* dynamically applied rate limits * canonical_mime_header_key overload

snyk: add changelog entry

b2c3842

address pr comment

2c62956

* fix propagation of rate_limit header where possible

address pr comment

94b456a

* fix propagation of rate_limit header where possible * explain what is happening better * in-line rate_limit construction or drop unused values

efd6 force-pushed the e24453-snyk-again branch from 14e09a1 to 94b456a Compare February 17, 2026 02:43

efd6 added 3 commits February 17, 2026 13:50

address pr comment

bd12270

* add rate-limit headers

fix rate-limit reset handling

fe0c8ee

This should not have been converted to a timestamp.

apply rate-limit changes in audit_logs to issues and fix up lagging c…

c08cf7b

…omment removal

efd6 requested a review from chrisberkhout February 17, 2026 06:37

chrisberkhout requested changes Feb 17, 2026

View reviewed changes

efd6 added 2 commits February 18, 2026 07:41

do not unnecessarily complicate the code

d3d501d

differentiate pure side-effect calls from side-effect plus return val…

8509377

…ue calls

efd6 requested a review from chrisberkhout February 17, 2026 21:26

		?"rate": (rate_headers.?rate == optional.of(double("Infinity"))) ? optional.of("inf") : optional.none(),
		?"next": (rate_headers.?next == optional.of(double("Infinity"))) ? optional.of("inf") : optional.none(),

Conversation

efd6 commented Dec 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Proposed commit message

Checklist

Author's Checklist

How to test this PR locally

Related issues

Screenshots

Uh oh!

elastic-vault-github-plugin-prod bot commented Dec 2, 2025

🚀 Benchmarks report

Uh oh!

botelastic bot commented Jan 1, 2026

Uh oh!

elasticmachine commented Jan 18, 2026

Uh oh!

chrisberkhout left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

efd6 commented Feb 17, 2026

Uh oh!

efd6 commented Feb 17, 2026

Uh oh!

efd6 commented Feb 17, 2026

Uh oh!

chrisberkhout left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

elasticmachine commented Feb 17, 2026

💚 Build Succeeded

History

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

efd6 commented Dec 2, 2025 •

edited

Loading

chrisberkhout left a comment •

edited

Loading