azure: add routing system test and fake EventHub server - #20581
Conversation
efd6
added
enhancement
Add a script-based system test that verifies each Azure log category routes to the correct data stream. The test starts a fake EventHub server that speaks AMQP 1.0 on port 5672 and serves Azure Blob Storage REST on port 10000, delivers one event per Azure log category, then checks that each event lands in the expected index. The fake server (_dev/scripts/azure-eh/main.go) handles the full AMQP 1.0 handshake including CBS token authentication, link flow control, and message transfer. Messages include the annotations x-opt-enqueued-time, x-opt-sequence-number, and x-opt-offset that the azeventhubs receiver expects. The blob storage implementation generates correct XML for list-blobs responses, including Last-Modified and Etag in <Properties> and per-key child elements in <Metadata>, which the checkpoint store relies on to track partition ownership. Add a storage_account_connection_string package variable so the test can point the checkpoint store at the fake server's blob endpoint. When set, it is used as-is; the connection string built from storage_account and storage_account_key is used otherwise.
✅ Elastic Docs Style Checker (Vale)No issues found on modified lines! The Vale linter checks documentation changes against the Elastic Docs style guide. To use Vale locally or report issues, refer to Elastic style guide for Vale. |
|
✅ All changelog entries have the correct PR link. |
🚀 Benchmarks reportTo see the full report comment with |
💚 Build Succeeded
cc @efd6 |
|
Pinging @elastic/security-service-integrations (Team:Security-Service Integrations) |
![vera-review-bot[bot]](https://avatars.githubusercontent.com/in/3297022?s=60&v=4){kind=small}
| required: false | ||
| show_user: false | ||
| description: >- | ||
| (Optional) A full connection string for the Storage Account, including the endpoint. When set, this overrides the connection string built from **Storage Account** and **Storage Account Key**. |
There was a problem hiding this comment.
Severity: 🟡 Medium confidence: high path: packages/azure/manifest.yml:89
This package-level variable is only read by the events data stream, so the description promising it overrides the built connection string is wrong for the other 13 azure-eventhub streams. Scope the description to the v2 events stream (or add the same override to the other templates).
Details
Vars declared at the package root apply to every policy template in the package, so this option is settable on adlogs, activitylogs, platformlogs, firewall_logs, eventhub, etc. But only data_stream/events/agent/stream/stream.yml.hbs consults it. Every other stream template (data_stream/*/agent/stream/azure-eventhub.yml.hbs and data_stream/eventhub/agent/stream/stream.yml.hbs, all at line 42) still unconditionally emits storage_account_connection_string: DefaultEndpointsProtocol=https;AccountName={{storage_account}};AccountKey={{storage_account_key}};EndpointSuffix={{endpoint_suffix}}. Setting the new var on any of those is therefore a silent no-op, while the description states it "overrides the connection string built from Storage Account and Storage Account Key". Separately, the closing sentence on the next line calls the target "a custom S3-compatible store" — the input speaks the Azure Blob Storage REST API, not S3, so that example is misleading.
Recommendation:
Either scope the wording to the data stream that honours it, or extend the other azure-eventhub.yml.hbs templates with the same {{#if}}/{{else}} block. The cheap fix:
- name: storage_account_connection_string
type: password
secret: true
title: Storage Account Connection String
multi: false
required: false
show_user: false
description: >-
(Optional, Azure Logs (v2 preview) only) A full connection string for the Storage Account, including the endpoint.
When set, it replaces the connection string built from **Storage Account** and **Storage Account Key** for the `events` data stream;
the v1 data streams ignore it. Use this to point at a non-standard blob endpoint, for example a local Azurite emulator.🤖 AI-Generated Review | Vera Review Bot | 📚 Knowledge base: integration-skills
⚠️ Automated review — verify suggestions before applying.
| // since copyOwnershipPropsFromBlob dereferences both without nil checks. | ||
| var buf strings.Builder | ||
| buf.WriteString(`<?xml version="1.0" encoding="utf-8"?><EnumerationResults><Blobs>`) | ||
| for name, item := range blobs { |
There was a problem hiding this comment.
Severity: 🟡 Medium confidence: high path: packages/azure/_dev/scripts/azure-eh/main.go:157
listBlobs ranges over the container map after dropping the mutex while putBlob inserts into that same map under it, which Go aborts with an uncatchable fatal error. Hold the lock across the whole iteration.
Details
Lines 147-149 take s.mu only long enough to fetch the inner map, then this loop iterates it with the lock released; putBlob (line 205) writes s.containers[container][blob] = ... while holding the lock. The Azure blob checkpoint store used by the v2 processor lists ownership blobs from its load-balancer goroutine while partition clients create the ownership/checkpoint blobs from another goroutine, so the two overlap during startup. A concurrent map read and map write is a Go runtime fatal error that recover cannot catch: the fake server dies and the routing assertions fail — intermittently, which is the worst failure mode for CI. getBlob (lines 251-262) has the same shape, reading item.metadata and item.data after unlocking while setMetadata (lines 232-237) reassigns those fields under the lock; that one is a plain data race rather than a fatal abort, but it can also serve a torn view of a checkpoint blob.
Recommendation:
Keep the mutex for the whole read in both methods — the fake server does no slow work while holding it:
func (s *blobStore) listBlobs(w http.ResponseWriter, r *http.Request, container string) {
prefix := r.URL.Query().Get("prefix")
s.mu.Lock()
defer s.mu.Unlock()
blobs := s.containers[container]
// ... unchanged ...
}
func (s *blobStore) getBlob(w http.ResponseWriter, container, blob string) {
s.mu.Lock()
defer s.mu.Unlock()
blobs := s.containers[container]
var item *blobItem
if blobs != nil {
item = blobs[blob]
}
// ... unchanged ...
}🤖 AI-Generated Review | Vera Review Bot | 📚 Knowledge base: integration-skills
⚠️ Automated review — verify suggestions before applying.
| {{/if}} | ||
| {{#if storage_account_connection_string}} | ||
| storage_account_connection_string: {{storage_account_connection_string}} | ||
| {{else}} |
There was a problem hiding this comment.
Severity: 🔵 Low confidence: medium path: packages/azure/data_stream/events/agent/stream/stream.yml.hbs:42
Putting storage_account_key inside the {{else}} branch means setting the new override also drops the key, which processor v1 requires — emit the key independently of the connection-string override.
Details
In filebeat's azure-eventhub input, validateStorageAccountConfig returns no storage account key configured (config: storage_account_key) when processor_version: v1 and storage_account_key is empty; only v2 accepts a bare storage_account_connection_string (and warns that the key is unused). processor_version is still a selectable var on this data stream (default v2, with a v1 (legacy) option), so a policy that sets storage_account_connection_string while pinned to v1 now renders a config the input rejects at startup, where before the change the key was always emitted alongside the built connection string. Both vars are show_user: false, so the blast radius is small — hence low severity — but the regression is in the rendered template as written.
Recommendation:
Emit the key on its own and let the override apply only to the connection string:
If you would rather not emit the key for v2 policies (filebeat logs a deprecation warning when it sees it), gate the first block on {{#contains "v1" processor_version}} instead.
🤖 AI-Generated Review | Vera Review Bot | 📚 Knowledge base: integration-skills
⚠️ Automated review — verify suggestions before applying.
Review summaryIssues found across the latest commits 247d00a — 2 medium, 1 low
🤖 AI-Generated Review | Vera Review Bot | 📚 Knowledge base: integration-skills
|
1 participant
Proposed commit message
Checklist
changelog.ymlfile.Author's Checklist
How to test this PR locally
Related issues
Screenshots