feat(obs): pluggable observability-sink framework + Aliyun SLS exporter - #528
Merged
Conversation
Introduce the capability-typed `ObservabilitySink` adapter contract in aisix-obs (AISIX-Cloud#692, phase F1) that the shared delivery pipeline and concrete sinks (Aliyun SLS first) build on: - `ObservabilitySink` trait (#[async_trait], object-safe `Arc<dyn>`): encode-and-deliver one batch, plus committed-marker resume (default `None` for at-least-once sinks) and a health probe. - `SinkCapabilities` + `IdempotencyScheme`/`IdempotencyMarker` + `ChannelKey` + `OrderingScope`/`BatchUnit` — the capability matrix that makes HTTP / object-store / warehouse differences data, not a fork in the pipeline. - `SinkRecord` wraps the existing `UsageEvent` metadata (single-sourced) plus opt-in `SinkContent`; metadata-only is the default so the path cannot carry a prompt. `EventBatch` Arc-shares records for fan-out. - `SinkError` (Transient/Permanent), `SinkAck`, `SinkHealth`. No wiring into the request hot path yet (phase F2). Unit tests cover the types and dyn-compatibility; existing otlp_http_sink tests unaffected.
Generalise the aisix-server telemetry worker (bounded mpsc -> batch -> flush) into a reusable `SinkPipeline` that every sink runs behind (AISIX-Cloud#692, phase F2), and add what telemetry deliberately skipped: - retry with exponential backoff for `SinkError::Transient` batches, capped at `max_backoff`; - drop-with-metric backpressure — queue-full, retry-exhausted, and permanent failures are all counted (`SinkStats`) and logged, not silently lost like the per-event otlp fan-out; - one pipeline per sink (independent queue + stats) so a slow/down sink can't stall another or the request hot path; - graceful shutdown drain. `SinkHandle::try_enqueue` is non-blocking and drops on a full bounded queue. Flow control lives in the pipeline; wire encoding and per-request byte chunking stay in the sink. Not wired into the request path yet (phase F3). 8 unit tests cover batching, retry, drop accounting, age-flush and backoff; workspace compile and existing tests unaffected.
Make OTLP/HTTP-JSON traces a first-class `ObservabilitySink` so it can run behind the shared `SinkPipeline` (AISIX-Cloud#692, phase F3a): - split the span encoder into `build_otlp_span` (one span) + `otlp_export_request` (wrap N spans into one ExportTraceServiceRequest); `build_otlp_traces_payload` stays a thin wrapper with byte-identical output, so the existing fan-out and its tests are untouched. - add `OtlpSink`: batches every record's span into a single export request and one POST (vs the fan-out's N per-event spawns), and classifies failures — 5xx/408/429 and network errors are Transient (retryable), other 4xx are Permanent. No hot-path rewiring yet; the existing `OtlpHttpFanOut` is unchanged and still in use (the per-exporter pipeline manager + call-site migration is phase F3b). 3 new tests (batched single request, 5xx->transient, 4xx->permanent); the 8 existing otlp tests pass as the regression gate.
Add `ExporterPipelines` (AISIX-Cloud#692, phase F3b) — the shared, sink-agnostic manager that turns the snapshot's exporter set into running `SinkPipeline`s. Reused by every pipeline-backed sink family (otlp, http_batch, ...): the caller supplies a `build` closure mapping a desired exporter spec to an `ObservabilitySink`. - `reconcile(desired, key, fingerprint, build)` starts a pipeline per new exporter, stops pipelines whose key disappeared (cancel -> drain -> exit, not aborted), and rebuilds those whose fingerprint changed. `build` runs only for new/changed exporters, so unchanged pipelines keep running. - `enqueue_to_all(&record)` fans an Arc-shared record into every pipeline, non-blocking on the request hot path (a full queue drops + counts there). - `stats()` exposes per-exporter delivery counters; `shutdown()` drains all. Self-contained and unit-tested (start/stop/rebuild/idempotent reconcile + fan-out delivery); not wired into the request path yet (phase F3b-2).
Wire the otlp_http exporter path onto the framework (AISIX-Cloud#692, phase F3b): `OtlpHttpFanOut::fan_out` now resolves a per-exporter `SinkPipeline` via the manager and enqueues one span record into it, instead of spawning a fire-and-forget POST per (event, exporter). The `fan_out` signature is unchanged, so all 8 proxy hot-path call sites stay untouched. - Reshape `ExporterPipelines` from a reconcile-from-loop model to lazy `get_or_create` + `retain`. Lazy-on-first-sighting (like the old permit map) is immediately consistent with the snapshot — a just-added exporter receives the very next request — which a reconcile poll would race (and flake the otlp e2e). - Delivery now batches (1s flush), retries transient failures with backoff, and drops-with-metric under backpressure (the pipeline's bounded queue), replacing the per-exporter Semaphore(64) cap. Removed the obsolete MAX_INFLIGHT / permits_for / in_flight_for / post_one and the two #113 semaphore tests; added one that asserts the new path delivers a span to a real receiver. Validated: clippy -D warnings clean, 64 unit tests, and the usage-client-source real-chain e2e (register exporter -> real chat -> assert span at the receiver) passes against the built DP + a real etcd. Follow-up (not blocking): wire the server GC loop (gc/retain for removed exporters) + shutdown drain; otlp is best-effort and the exporter set is bounded, matching the prior never-pruned behavior.
Add `aliyun_sls` as the second observability-exporter kind, delivering
request events to Aliyun SLS via a signed PutLogs call.
- AliyunSlsSink (http_batch family): one LogGroup per batch, lz4 block
compression, SLS signature v1 (HMAC-SHA1), POST to
/logstores/<ls>/shards/lb. Uses the official aliyun-log-sdk-protobuf /
aliyun-log-sdk-sign sub-crates over the existing rustls reqwest, so no
OpenSSL / native-tls is pulled in.
- Field mapping serializes UsageEvent (single-sources the schema and its
skip_serializing_if rules); opt-in prompt/response land as flat fields,
absent on the default metadata-only path so prompts cannot leak.
- Error classification mirrors OtlpSink (5xx/408/429 -> transient) plus
SLS errorCode-aware promotion, so a 4xx quota/busy/skew is retried
rather than dropped.
- ExporterKind::AliyunSls carries a credential_ref, never the AccessKey:
the plaintext key never rides the kine path. The DP resolves the ref
from its local env (SLS_CRED_<REF>_AK_{ID,SECRET}; deliberately not the
AISIX_ prefix, which the config loader claims via deny_unknown_fields).
- fan_out generalized to dispatch per kind over one shared
ExporterPipelines + client (the #692 design); call sites unchanged.
The OtlpHttpFanOut -> ExporterFanOut rename is deferred as a mechanical
follow-up.
- Adds an env-gated, #[ignore] real-SLS smoke test that validates the
signature / lz4 / protobuf against the live endpoint.
Refs: api7/AISIX-Cloud#687, api7/AISIX-Cloud#692
- L2 e2e: a dashboard-configured aliyun_sls exporter makes the real DP post a signed lz4-protobuf PutLogs to a mock SLS receiver on each chat. Asserts the logstore path, SLS headers, lz4 framing, and that the Authorization carries the env-resolved AccessKey id (proving the credential_ref -> env path), with the secret never on the wire. Body field-mapping is covered by the Rust round-trip unit test; real-endpoint acceptance by the smoke. - harness: AppOverrides.extraEnv injects non-config secrets (the SLS AccessKey) past the AISIX_* strip. - ci: a sls-smoke job runs the #[ignore] real-SLS smoke with the SLS secrets injected as env; fork PRs (no secrets) skip cleanly via the test's env-gate. Refs: api7/AISIX-Cloud#687
ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Free Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds a pluggable observability-sink framework, refactors OTLP delivery into pipeline-backed batching/retry, and introduces an Aliyun SLS sink with credential-ref resolution, signing, lz4+protobuf encoding, plus schema, tests, and CI/e2e integrations. ChangesPluggable Observability Sink Platform with Aliyun SLS Backend
🎯 4 (Complex) | ⏱️ ~75 minutes Note 🎁 Summarized by CodeRabbit FreeYour organization has reached its limit of developer seats under the Pro Plan. For new users, CodeRabbit will generate a high-level summary and a walkthrough for each pull request. For a comprehensive line-by-line review, please add seats to your subscription by visiting https://app.coderabbit.ai/login.If you believe this is a mistake and have available seats, please assign one to the pull request author through the subscription management page using the link above. Comment |
Independent audit of #528 (finding M1): SinkCapabilities.max_batch_bytes is advertised but enforced nowhere — SinkPipeline batches by record count, and AliyunSlsSink encodes the whole batch into one PutLogs. Harmless on today's metadata-only path, but a silent Permanent-drop once content capture makes a batch exceed SLS's PutLogs body limit. Downgrade the capability to its honest current state (batch_unit: Records, max_batch_bytes: None — parity with OtlpSink) so it no longer promises an unenforced ceiling. Byte-aware chunking (the real fix, in append_batch per the pipeline's sink-self-chunks design) is filed as \#529 and blocks the content-capture milestone.
Collaborator
Author
Independent audit (CLAUDE.md §7) — results + dispositionsA fresh general-purpose agent reviewed this PR cold (no shared context), verifying the wire/secret/regression claims against the actual Aliyun SLS SDK source, not just the description. No HIGH findings. Wire-signing order, lz4-block choice, MEDIUM
LOW (non-blocking)
Bottom line from the audit: safe to merge after M1, which is now addressed. Remaining items are LOW and justified above. |
This was referenced Jun 6, 2026
moonming
added a commit
that referenced
this pull request
Jun 8, 2026
The real-credential Aliyun SLS validation moved to the control plane as a full-chain e2e (cp-api config → live DP → real logstore readback, api7/AISIX-Cloud#709). Removes the now-dead mod smoke (sls_smoke_putlogs_and_readback + helpers) and its CI job; the mock-SLS L2 e2e remains the DP-side sink coverage. Refs #528, #532, api7/AISIX-Cloud#709.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Introduces a pluggable observability-sink framework and lands its first real vendor — Aliyun SLS — so the DP can ship per-request telemetry (and, later, full prompt/response) to customer log/analytics backends without a bespoke connector per vendor.
Driven by the pre-sales requirement on api7/AISIX-Cloud#687 (customer wants AI Gateway logs in Aliyun SLS) and the generic-integration design in api7/AISIX-Cloud#692.
Framework (one implementation per family, not per vendor)
ObservabilitySinktrait +SinkCapabilitiesmatrix (idempotency / ordering / batch-unit / partial-batch / streaming) so the shared pipeline adapts to a sink's wire contract as data, not a fork.SinkPipeline— bounded mpsc → batch → retry/backoff → drop-with-metric, one per exporter.ExporterPipelinesmanager — lazyget_or_create(immediate consistency) +retainGC.OtlpSink), behaviour-preserving (theusage-client-sourcee2e is the regression gate).Aliyun SLS exporter
AliyunSlsSink(http_batch family): one protobufLogGroupper batch → lz4 block compression → SLS signature v1 (HMAC-SHA1) →POST /logstores/<ls>/shards/lb. Built on the officialaliyun-log-sdk-protobuf/aliyun-log-sdk-signsub-crates over the existing rustls reqwest — no OpenSSL / native-tls is pulled in.ExporterKind::AliyunSlscarries acredential_ref, never the AccessKey — the plaintext key never rides the kine path. The DP resolves the ref from its local env (SLS_CRED_<REF>_AK_{ID,SECRET}, deliberately not theAISIX_prefix the config loader claims). BYOK variants slot into the same resolution point later.fan_outgeneralized to dispatch per kind over one sharedExporterPipelines+ client; all 7 proxy call sites unchanged.Key design decisions
credential_ref+ local env resolution;rejects_plaintext_credentials_in_configpins it at the serde + schema layers.UsageEvent(single-sources the schema + itsskip_serializing_ifrules) rather than hand-listing 35 fields; opt-in prompt/response land as flat fields, absent on the default metadata-only path so prompts cannot leak.OtlpSink(5xx/408/429 → transient) plus SLSerrorCode-aware promotion, so a 4xx quota/busy/clock-skew is retried, not dropped (a log sink must not drop under throttling).https://<project>.<host>; a scheme-qualified loopback (the e2e mock) is used verbatim.Test plan
aisix-core204 passed,aisix-obs75 passed (sink encode/sign/lz4 round-trip decode, error classification incl. quota→transient, credential resolver, schema validators). clippy-D warningsclean;schemas/regenerated (schema-drift gate).aliyun-sls-exporter-e2e): a dashboard-configuredaliyun_slsexporter makes the realaisixbinary post a signed lz4-protobuf PutLogs to a mock SLS receiver on a real chat — asserts logstore path, SLS headers, lz4 framing, and thatAuthorizationcarries the env-resolved AccessKey id (secret never on the wire). ✅ passing.usage-client-source-e2e): ✅ 2/2 — confirms the per-kind dispatch generalization preserved the OTLP path.sls_smoke,#[ignore]+ env-gated): validates the v1 signature / lz4 / protobuf against the live SLS endpoint (the thing a mock cannot check). Runs in the newsls-smokeCI job with the SLS secrets injected; fork PRs skip cleanly. PutLogs→2xx is the hard assertion (Permanent error fails fast, Transient retries); GetLogs readback is best-effort with diagnostics pending its first real run.Deliberately out of scope (follow-ups)
aliyun_slssupport (AISIX-Cloud / Go) — the DP accepts the kind via admin API + kine now; the dashboard/CP flow is separate per feat(routing): sticky weighted routing (A/B / canary) #687 ("DP first, then CP").OtlpHttpFanOut → ExporterFanOutand wiring the fan-out GC/shutdown loop — mechanical follow-ups kept out to avoid churn.tsc --noEmiterrors in unrelated e2e files tracked in e2e: pre-existingtsc --noEmiterrors in cache-scenarios / request-id-uuid tests #527 (not touched here; my additions typecheck clean).Refs: api7/AISIX-Cloud#687, api7/AISIX-Cloud#692
Summary by CodeRabbit
New Features
Tests
Chores