fix(ess_billing): Improve reliability and prevent API errors

[andrewkroh]

NOTE: A commit-by-commit review is recommended because of the celfmt changes.

Proposed commit message

fix(ess_billing): Improve reliability and prevent API errors

This commit introduces several improvements to the ess_billing integration to
enhance its reliability and prevent common API errors.

- To prevent repeated failing requests on non-200 HTTP status codes, the CEL
  program now sets want_more to false. This allows the input to retry at the
  next periodic interval instead of exhausting the execution budget.
- A validation rule has been added to enforce a minimum 'from' date of
  2021-01-01. This clamps the calculated timestamp to the API's minimum allowed
  value, preventing 'Bad Request' errors caused by lookbehind configurations
  creating excessively early dates. This uses the max() function which requires
  Elastic Agent 8.18 or greater so the Kibana constraint was raised as a proxy.
- All instances of the now() function in the CEL program have been replaced
  with the now variable.  This ensures a stable time reference throughout a
  single execution, leading to more consistent and predictable time-based
  calculations.

  A system test has been added to cover these scenarios.

  Fixes #14743
  Fixes #14755

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

How to test this PR locally

elastic-package -C packages/ess_billing test system -v

Related issues

• Fixes [ESS Billing]: CEL Configuration Issue #14743
• Fixes [ESS Billing] Invalid 'from' query param #14755

[efd6]

This all seems sane, though it looks like the system test needs attention since it's failing. I would be nice to understand where the invalid timestamp in the bugged case came from, but this will fix it.

[andrewkroh]

Looks like the CI problem is my usage of max with 8.15.0.

fleet-server-1 | {"log.level":"info","@timestamp":"2025-07-31T03:37:26.000Z","message":"applying new components data","component":{"binary":"fleet-server","dataset":"elastic_agent.fleet_server","id":"fleet-server-default","type":"fleet-server"},"log":{"source":"fleet-server-default"},"service.name":"fleet-server","req.Components":[{"id":"cel-default","message":"Healthy: communicating with pid '316'","status":"HEALTHY","type":"cel","units":[{"id":"cel-default-cel-ESS Billing-4f94630c-28ef-4096-a8ba-2306872c53e2","message":"failed to check program: failed compilation: ERROR: <input>:11:8: found no matching overload for 'max' applied to '(timestamp, timestamp)'\n | max(from, timestamp(\"2021-01-01T00:00:00Z\"))\n | .......^ accessing config","payload":{"streams":{"cel-ess_billing.billing-4f94630c-28ef-4096-a8ba-2306872c53e2":{"error":"","status":"STARTING"}}},"status":"FAILED","type":"input"},{"id":"cel-default","message":"Healthy","status":"HEALTHY","type":"output"}]}],"service.type":"fleet-server","http.request.id":"01K1F8WQZJQW54PFJT18VCHVNX","server.address":"","fleet.agent.id":"64fa78d6-ba93-421e-b6bf-f44e31225444","fleet.access.apikey.id":"wx2JXpgBcJYJrjvG18FE","@timestamp":"2025-07-31T03:37:26Z","ecs.version":"1.6.0","ecs.version":"1.6.0"}

[elastic-sonarqube]

Quality Gate passed

Issues
0 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
100.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube

[elasticmachine]

💚 Build Succeeded

• Buildkite Build
• Commit: 583c2ac

History

• 💔 Build #29347 failed 7a3e86d
• 💔 Build #29343 failed 0a7a2bc
• 💚 Build #29322 succeeded 9e53bb8
• 💚 Build #29307 succeeded cc1cdce

[andrewkroh]

@3kt would you be able to review my changes (since you have contributed and are a CODEOWNER)? Thanks!

[3kt]

Out of curiosity, what's the difference between now() and now (apart from the obvious that one is a function and the other a variable)?

[andrewkroh]

https://pkg.go.dev/github.com/elastic/mito/lib#Time

now is global var set at execution time. now() is function that returns the current time. By invoking now() you could have a different view of the current time in all of the places that it is used.

[3kt]

Why would we stop there?
Let's say we have a 429 or 503 - shouldn't we retry if we're not caught up?

[andrewkroh]

This is the crux of the problem. For instance, by returning want_more: true on HTTP 500, it will trigger 999 more CEL program executions in just a few seconds (assuming the server keeps responding with non-200 status codes). There is already a default built-in retry request mechanism that attempts the request up to 5 times.

My goal is to stop the program and have the next iteration retry the same request. However, it has been some time since I made this change, so I need to revisit it to ensure the cursor isn't advanced. I might need to remove the cursor from this failure case to prevent advancing it. I'll test this and provide an update.

[3kt]

The other issue is that we'll wait 24 hours before retrying I guess? 24h being the execution schedule.

[3kt]

The drop processor then needs to be removed from the ingest pipeline as well:

  - drop:
      if: "ctx.ess?.billing == null && ctx.error?.message == null"

Its only purpose was to drop this fake event, but I got heavy-handed and stated "anything that doesn't have an error or the billing namespace should be dropped".

[andrewkroh]

It can be removed, but to be safe it should probably be removed at a later time to give Agents a chance to upgrade and get the local beat drop_event processor. The package upgrade and the agent policy upgrade are not an atomic operation. If the package upgrades, but the related Agent policies are not immediately upgraded then this could offer a window where fake: true events sneak through.

[3kt]

This is a very good point actually, I forgot the local and remote parts aren't necessarily atomically tied.
Maybe worth commenting in the pipeline that this should be removed at a later date then?