[security][Audit] Add structured audit logging for RPC operations

[litiliu]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Fluss already has pluggable authentication and authorization support, but it does not have a first-class, structured audit logging facility for security-sensitive RPC operations.

Production users need to answer questions such as:

• Who accessed or modified a Fluss resource?
• Which operation was attempted, from which client address, and when?
• Was authentication successful?
• Was an authorization decision allowed or denied?
• Did an administrative operation eventually succeed or fail?

This is important for security monitoring, incident investigation, access reviews, and regulatory compliance.

This proposal is different from the $changelog and $binlog virtual tables in FIP-20. Those tables expose row-level data changes, while security audit logs should capture the principal, request context, resource, operation, authorization decision, and operation outcome.

Solution

Introduce a structured and extensible audit logging framework around the Fluss RPC authentication, authorization, and request-processing paths.

An audit event could include:

• Event ID, schema version, and timestamp
• Event category, such as authentication, authorization, management, or data access
• FlussPrincipal, authentication mechanism, and authentication result when available
• Client address, listener name, and whether the request is internal
• RPC/API name, request ID, correlation ID, and client ID when available
• Resource type and resource name
• Operation type
• Authorization result (ALLOWED or DENIED) and the matched rule when appropriate
• Operation outcome and a sanitized error code

The audit framework should:

1. Emit machine-readable, versioned events, for example structured JSON.
2. Provide a pluggable sink API so deployments can route events to a dedicated log, Kafka/Fluss system table, or an external security platform.
3. Support filtering by event category, resource, principal, internal/external traffic, and allowed/denied result.
4. Avoid blocking latency-sensitive RPC request threads. A bounded asynchronous delivery path and explicit overload behavior should be considered.
5. Never include record payloads, credentials, tokens, or sensitive configuration values.
6. Expose metrics and a fallback strategy when audit event delivery fails.

Possible initial event categories:

• Authentication: successful and failed authentication attempts
• Authorization: allowed and denied resource/operation checks
• Management: create/drop/alter database, table, partition, ACL, and cluster configuration operations
• Data access: produce, fetch, put, lookup, and scan operations, disabled or sampled by default because of their volume
• Internal operations: optionally record or exclude server-to-server RPCs

Open design questions include:

• Should the default implementation write structured logs, a persistent system table, or both?
• Should authorization decisions and final operation outcomes be separate events?
• Which event categories should be enabled by default?
• How should ordering, retention, backpressure, and delivery failures be handled?
• Which fields can be added to the existing Session and RPC context without exposing sensitive data?

Related work:

• FIP-2: Fluss Authentication and Authorization
• #399: Introduce Fluss Authentication and authorization
• #2007: Add missing authentication for RPC calls
• FIP-12: Global Server Dynamic Config, which mentions a lightweight configuration change audit log

Anything else?

The initial scope should focus on auditable metadata and security decisions rather than message contents or complete data lineage.

If the community agrees on the direction, a follow-up FIP can define the event schema, extension API, default sink, performance guarantees, and compatibility policy in detail.

Willingness to contribute

• I'm willing to submit a PR!