[
https://issues.apache.org/jira/browse/RANGER-5655?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ramachandran Krishnan updated RANGER-5655:
------------------------------------------
Description:
Implement a dynamic unified ingestor registry for Ranger audit-ingestor so
operators can change Kafka partition routing and per-repo service allowlists at
runtime — without restarting ingestor pods.
The registry is stored in a Kafka compacted topic
({{ranger_audit_partition_plan}}) and managed via REST. All ingestor replicas
converge on the same versioned plan through {{PartitionPlanWatcher}};
{{AuditPartitioner}} routes audit records on the hot path from in-memory state
only.
*Feature flag (default off):*
{{ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled=false}}
*Pull request:* [GitHub PR #1032|https://github.com/apache/ranger/pull/1032]
----
h2. Problem
Today (static mode), audit-ingestor loads two kinds of configuration from XML
at startup only:
|| Job || Question || Static behavior ||
| Service allowlist | May this Kerberos principal POST audits for repo R? |
{{ranger.audit.ingestor.service.<repo>.allowed.users}} in site XML |
| Partition routing | After accept, which {{ranger_audits}} partition? |
{{kafka.configured.plugins}} + per-plugin overrides in site XML |
Changing either requires editing XML and restarting every ingestor replica.
Contiguous-range static allocation can also reshuffle later plugins when an
early plugin's partition count changes.
*Goals:*
* Onboard new plugins/repos and scale hot plugins without ingestor restart
* Append-only partition growth (no reshuffle of existing plugin assignments)
* One shared source of truth across all ingestor pods
* No new infra (no Postgres / ZooKeeper for the registry)
----
h2. Solution
Introduce a unified partition plan document (versioned JSON) in Kafka topic
{{ranger_audit_partition_plan}} (1 partition, compacted). One document holds:
* {{plugins}} — dedicated partition IDs per plugin id (Kafka record key / agent
id)
* {{buffer}} — partition pool for not-yet-promoted plugins (sticky hash)
* {{services}} — per-repo {{allowedUsers}} for {{POST /api/audit/access}}
(optional {{pluginId}} for ownership)
* {{topicPartitionCount}} — must match live {{ranger_audits}} partition count
* {{version}} — optimistic locking for REST mutations
{code}
Control plane: GET/PATCH /api/audit/partition-plan → Kafka registry topic →
PartitionPlanWatcher
Data plane: POST /api/audit/access (Kerberos) → allowlist check
→ optional auto-onboard
→ AuditPartitioner → ranger_audits
{code}
Solr/HDFS dispatchers are unchanged; they consume all partitions of
{{ranger_audits}}.
h3. REST API (v2 — two public endpoints)
Plan admin endpoints are *unauthenticated* (same as health/status):
{{permitAll()}} in Spring Security and bypass in
{{AuditDelegationTokenFilter}}. Audit ingestion remains Kerberos-protected on
{{POST /api/audit/access}}.
All mutations use optimistic locking via {{expectedVersion}} (from GET). Stale
writes return *409* with the current plan body.
|| Method || Endpoint || Purpose ||
| GET | {{/api/audit/partition-plan}} | Read current plan (503 when dynamic
mode off) |
| PATCH | {{/api/audit/partition-plan}} | Partial update via
{{PartitionPlanReplacement}} JSON body |
*PATCH body semantics* ({{PartitionPlanReplacement}}):
* {{expectedVersion}} — required; stale → 409
* {{plugins}} / {{services}} — merge *new keys only* (promote/onboard entries
not already in plan)
* {{pluginScales}} — map of {{pluginId → additionalPartitions}}; append-only
tail growth
* {{buffer}} / {{topicPartitionCount}} — optional; omitted fields inherited
from current plan
*Removed (superseded by PATCH + auto-onboard):*
* {{POST /api/audit/partition-plan/plugins}}
* {{PATCH /api/audit/partition-plan/plugins/\{pluginId\}}}
h3. Auto-onboard on audit access
When dynamic mode is enabled and a Kerberos-authenticated {{POST
/api/audit/access}} includes {{appId}} for a plugin not yet in the plan:
# Allowlist check passes for {{serviceName}} + authenticated principal
# Ingestor promotes the plugin from buffer (default partition count from
config) and upserts {{services[serviceName]}} in one plan version bump
# Audit batch is then produced using the updated in-memory plan
No separate onboard REST call is required for first-time plugin discovery.
Operators can still pre-promote plugins and allowlists via PATCH.
h3. Allowlist updates
{{services}} entries in the registry drive {{POST /access}} authorization. In
dynamic mode, {{AuthToLocalRuleComposer}} recomposes {{auth_to_local}} from the
union of all {{services[*].allowedUsers}} whenever the plan changes. Removing
or tightening a repo's allowlist causes *403* on subsequent access attempts
without pod restart.
----
h2. Key deliverables (PR #1032)
* Kafka registry — {{KafkaPartitionPlanRegistry}}, bootstrap from XML on
greenfield, brownfield pre-seed support
* REST API — {{GET}} / {{PATCH}} {{/api/audit/partition-plan}} with
{{PartitionPlanReplacement}} and {{expectedVersion}} (409 on stale)
* Auto-onboard — {{PartitionPlanService.ensurePluginOnboarded()}} on {{POST
/api/audit/access}} when {{appId}} is unknown
* Dynamic partitioner — {{AuditPartitioner}} reads {{PartitionPlanHolder}};
round-robin for promoted plugins; buffer sticky hash for unknown plugins;
post-scale routing uses {{max(cluster, plan)}} when metadata lags
* Unified allowlist — {{services}} map in same registry; per-repo POST checks
use each repo's own allowlist
* Topic grow — grow {{ranger_audits}} before registry write when
promoting/scaling
* Unit tests — partition-plan, allowlist, partitioner, and bootstrap coverage
in {{audit-common}} + {{audit-ingestor}}
*Out of scope for PR #1032 (follow-up / lab validation):* Docker E2E harness,
ops runbooks, brownfield migration guide (tracked separately in dev-support /
audit-server docs).
----
h2. Configuration
{code:xml}
<property>
<name>ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled</name>
<value>true</value>
</property>
<property>
<name>ranger.audit.ingestor.kafka.partition.plan.topic</name>
<value>ranger_audit_partition_plan</value>
</property>
{code}
When dynamic mode is on and the registry topic is empty, the first ingestor pod
bootstraps an initial plan from existing XML properties
({{kafka.configured.plugins}}, buffer/per-plugin counts, service allowlists).
h3. Example plan JSON
{code:json}
{
"topic": "ranger_audits",
"version": 12,
"topicPartitionCount": 48,
"plugins": {
"hdfs": { "partitions": [0, 1, 2, 3, 4, 5] },
"hiveServer2": { "partitions": [6, 7, 8, 9, 10, 11] }
},
"buffer": { "partitions": [12, 13, "..."] },
"services": {
"dev_hive": { "allowedUsers": ["hive"], "pluginId": "hiveServer2" },
"dev_ozone": { "allowedUsers": ["om", "ozone"], "pluginId": "ozone" },
"dev_hdfs": { "allowedUsers": ["hdfs", "nn"], "pluginId": "hdfs" }
}
}
{code}
h3. Example PATCH (promote + scale)
{code:json}
{
"expectedVersion": 12,
"plugins": {
"storm": { "partitions": [12, 13, 14] }
},
"services": {
"dev_storm": { "allowedUsers": ["storm"], "pluginId": "storm" }
},
"pluginScales": {
"hdfs": 2
}
}
{code}
----
h2. Testing
h3. Unit tests and quality gates
{code:bash}
mvn verify -pl audit-server/audit-common,audit-server/audit-ingestor
-Drat.skip=true
{code}
|| Gate || Result ||
| Unit tests | 104 passed (audit-ingestor module) |
| Checkstyle | Pass |
| PMD / SpotBugs | Pass (existing audit-ingestor findings unchanged) |
h3. Validated scenarios (manual + lab)
|| # || Scenario || Pass criteria ||
| 1 | Static mode (default off) | {{GET /partition-plan}} → 503; health 200;
plugin audits unchanged |
| 2 | Dynamic bootstrap | Plan topic created; watcher active; {{GET
/partition-plan}} → 200 with v≥1 |
| 3 | Public plan REST | {{GET}} / {{PATCH}} without Kerberos succeed when
dynamic mode on |
| 4 | PATCH promote/onboard | New {{plugins}} + {{services}} keys merged in one
version bump |
| 5 | PATCH {{pluginScales}} | Tail partitions appended; {{ranger_audits}}
grown; no reshuffle |
| 6 | Optimistic locking | Stale {{expectedVersion}} → 409 with current plan
body |
| 7 | Auto-onboard on {{/access}} | Unknown {{appId}} + allowlisted principal →
plan updated, audit accepted |
| 8 | Multi-pod convergence | All replicas report same version after PATCH on
one pod |
| 9 | Routing | Kafka record partition ∈ plugin assignment from plan |
| 10 | Allowlist enforcement | Allow/deny/cross-repo → 200/403; recomposition
on plan change |
----
h2. Acceptance criteria
# With {{dynamic.enabled=false}}, behavior matches existing static XML
partitioning; {{GET /partition-plan}} returns 503
# With {{dynamic.enabled=true}}, registry topic created (1 partition,
compacted); bootstrap plan published on greenfield
# {{PATCH /api/audit/partition-plan}} promotes/scales/mutates allowlists; stale
{{expectedVersion}} returns 409
# Unknown {{appId}} on authenticated {{POST /access}} auto-onboards plugin when
allowlist permits
# All ingestor replicas converge to same plan within watcher refresh interval
# Audits for promoted plugin land only on assigned {{ranger_audits}} partitions
# Service allowlists in registry drive {{/access}} authorization; unauthorized
principal returns 403
# Post-scale routing works when Kafka metadata lags plan ({{AuditPartitioner}}
bound logic)
was:
Implement a dynamic unified ingestor registry for Ranger audit-ingestor so
operators can change Kafka partition routing and per-repo service allowlists at
runtime — without restarting ingestor pods.
The registry is stored in a Kafka compacted topic
({{{}ranger_audit_partition_plan{}}}) and managed via REST
({{{}/api/audit/partition-plan/*{}}}). All ingestor replicas converge on the
same versioned plan through a background watcher; {{AuditPartitioner}} routes
audit records on the hot path from in-memory state only.
Feature flag (default off):
{{ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled=false}}
----
h2. Problem
Today (static mode), audit-ingestor loads two kinds of configuration from XML
at startup only:
||Job||Question||Static behavior||
|Service allowlist|May this Kerberos principal POST audits for repo
R?|{{ranger.audit.ingestor.service.<repo>.allowed.users}} in site XML|
|Partition routing|After accept, which {{ranger_audits}}
partition?|{{kafka.configured.plugins}} + per-plugin overrides in site XML|
Changing either requires editing XML and restarting every ingestor replica.
Contiguous-range static allocation can also reshuffle later plugins when an
early plugin's partition count changes.
*Goals:*
* Onboard new plugins/repos and scale hot plugins without ingestor restart
* Append-only partition growth (no reshuffle of existing plugin assignments)
* One shared source of truth across all ingestor pods
* No new infra (no Postgres / ZooKeeper for the registry)
----
h2. Solution
Introduce a unified partition plan document (versioned JSON) in Kafka topic
{{ranger_audit_partition_plan}} (1 partition, compacted). One document holds:
* {{plugins}} — dedicated partition IDs per plugin id (Kafka record key /
agent id)
* {{buffer}} — partition pool for not-yet-promoted plugins (sticky hash)
* {{services}} — per-repo {{allowedUsers}} for POST {{/api/audit/access}}
(optional {{pluginId}} for ownership)
* {{topicPartitionCount}} — must match live {{ranger_audits}} partition count
* {{version}} — optimistic locking for REST mutations
*Control plane:* REST +registry topic+ {{PartitionPlanWatcher}}
*Data plane:* plugins POST {{/api/audit/access}} → allowlist check →
{{AuditPartitioner}} → {{ranger_audits}}
Solr/HDFS dispatchers are unchanged; they consume all partitions of
{{{}ranger_audits{}}}.
h2. REST API (3 endpoints)
All mutations use optimistic locking via {{expectedVersion}} (from
{{{}GET{}}}). Stale writes return *409* with the current plan body.
||Method|| ||Endpoint||Purpose||
|{{GET}}| |{{/api/audit/partition-plan}}|Read current plan (503 when dynamic
mode off)|
|{{POST}}| |{{/api/audit/partition-plan/plugins}}|*Onboard* plugin: promote
from buffer + register service allowlists in one version bump|
|{{PATCH}}| |{{{}/api/audit/partition-plan/plugins/{pluginId{}}}}|*Update*
onboarded plugin: scale partitions and/or mutate allowlists
|
| | | | |
*{{removeServices}} semantics:* removes the whole {{services[repo]}} entry
(including {{allowedUsers}} and {{{}pluginId{}}}). Does *not* change
{{{}plugins.{*}.partitions{*}{}}}{*}. On each ingestor pod,
{{AuthToLocalRuleComposer}} recomposes {{auth_to_local}} from the *union of
remaining{*} {{{}services[*].allowedUsers{}}}; short names only on removed
repos no longer map → *403* on {{{}/access{}}}.
*Future work (not in this change):* bootstrap *v0* split — seeding {{services}}
from XML separately from plugin partition assignments. Current bootstrap still
publishes a single v1 plan from XML.
----
h2. Key deliverables
* Kafka registry — {{{}KafkaPartitionPlanRegistry{}}}, bootstrap from XML on
greenfield, brownfield pre-seed support
* REST API — *3 endpoints* above with {{expectedVersion}} (409 on stale)
* Dynamic partitioner — {{AuditPartitioner}} reads
{{{}PartitionPlanHolder{}}}; round-robin for promoted plugins; buffer sticky
hash for unknown plugins; post-scale routing uses max(cluster, plan) when
metadata lags
* Unified allowlist — {{services}} map in same registry; {{auth_to_local}}
rules recomposed from allowlist union in dynamic mode; per-repo POST checks use
each repo's own allowlist
* Topic grow — grow {{ranger_audits}} before registry write when
onboarding/scaling
* E2E harness — Docker Tier 3 scripts for partition-plan REST, plugin
onboard/routing, {{{}auth_to_local{}}}, full plugin→Solr pipelines (HDFS,
Ozone, Hive, etc.)
* Documentation — implementation guide, ops runbook, brownfield migration, E2E
test plan
----
h2. Configuration
{code:java}
<property>
<name>ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled</name>
<value>true</value>
</property>
<property>
<name>ranger.audit.ingestor.kafka.partition.plan.topic</name>
<value>ranger_audit_partition_plan</value>
</property> {code}
When dynamic mode is on and registry is empty, the first ingestor pod
bootstraps an initial plan from existing XML properties
({{{}kafka.configured.plugins{}}}, buffer/per-plugin counts, service
allowlists).
h2. Example plan JSON
{code:java}
{
"topic": "ranger_audits",
"version": 12,
"topicPartitionCount": 48,
"plugins": {
"hdfs": { "partitions": [0, 1, 2, 3, 4, 5] },
"hiveServer2": { "partitions": [6, 7, 8, 9, 10, 11] }
},
"buffer": { "partitions": [12, 13, "..."] },
"services": {
"dev_hive": { "allowedUsers": ["hive"], "pluginId": "hiveServer2" },
"dev_ozone": { "allowedUsers": ["om", "ozone"], "pluginId": "ozone" },
"dev_hdfs": { "allowedUsers": ["hdfs", "nn"], "pluginId": "hdfs" }
}
}{code}
h2. Testing
h3. Unit tests and quality gates
export MAVEN_OPTS="-Xmx4g"
mvn verify -pl audit-server/audit-common,audit-server/audit-ingestor
-Drat.skip=true}}
||Gate||Result||
|Unit tests|*94 passed* — {{audit-common}} + {{audit-ingestor}}|
|Checkstyle|*0 violations*|
|PMD|*0 rule violations*|
h3. Validated scenarios
h4. 1. Static vs dynamic mode
What it proves: The feature flag works and existing deployments are not broken.
||Mode||Config||Expected behavior||
|Static (default)|{{kafka.partition.plan.dynamic.enabled=false}}|Routing and
allowlists from XML at startup only. {{GET /api/audit/partition-plan}} returns
{*}503{*}. Ingestor health {*}200{*}. Plugin audits still flow to Solr.|
|Dynamic|{{dynamic.enabled=true}}|{{PartitionPlanWatcher}} active; routing from
in-memory plan. {{GET /partition-plan}} returns JSON.|
h4. 2. Greenfield bootstrap
What it proves: On a new cluster with an empty plan topic, the first ingestor
pod creates version 1 of the partition plan automatically — no manual REST call.
Flow:
# Enable dynamic mode; restart ingestor.
# Plan topic {{ranger_audit_partition_plan}} is created (1 partition,
compacted).
# Ingestor publishes v1 from existing XML ({{{}kafka.configured.plugins{}}},
buffer size, per-plugin overrides, service allowlists).
# Plan's {{topicPartitionCount}} matches live {{ranger_audits}} partition
count (e.g. 48 = 13 plugins × 3 + 9 buffer on full lab layout).
Success: {{GET /partition-plan}} shows v1, correct plugin list, partition count
aligned with Kafka. Audits still work after enable.
h4. 3. REST onboard / update (200 / 409 / 400)
What it proves: Operators can change routing and allowlists at runtime via REST
without restarting ingestor. The API rejects bad requests.
||Action||REST||Success||Failure cases||
|Onboard plugin from buffer|{{POST /api/audit/partition-plan/plugins}} (with
non-empty {{{}services{}}})|200, version increments, plugin in {{plugins}} map,
services merged|400 if plugin already promoted; 400 if {{services}}
missing/empty|
|Scale hot plugin (+N partitions)|{{{}PATCH .../plugins/{pluginId{}}}} with
{{additionalPartitions}}|200, new partition IDs appended at tail;
{{ranger_audits}} grown first if needed|400 if scaling buffer-only / unknown
plugin|
|Mutate allowlists|{{{}PATCH .../plugins/{pluginId{}}}} with {{addServices}} /
{{updateServices}} / {{removeServices}}|200, {{services}} updated;
{{auth_to_local}} recomposed on plan install|400 if repo owned by another
plugin|
Key property: Changes are append-only — existing plugin partition lists are
never reshuffled.
h4. 4. Multi-pod convergence
What it proves: In a real cluster with multiple ingestor replicas behind a load
balancer, all pods see the same plan after an admin change.
Flow:
# Primary ingestor on :7081, second replica on :7082 (same Kafka plan topic,
different Kerberos identity).
# Both pods report the same version at startup.
# Admin promotes or updates a plugin on primary only.
# Within one watcher cycle (~30s), replica on :7082 shows the same new version
without any REST call on the replica.
Success: No drift between pods; routing is consistent cluster-wide.
h4. 5. Brownfield pre-seed
What it proves: Existing production clusters can cut over to dynamic mode
safely by writing the plan into Kafka before enabling the feature — ingestor
must not overwrite it with a fresh XML bootstrap.
Flow (Path A migration):
# Capture current plan JSON while dynamic is briefly on.
# Turn dynamic off; delete plan topic (simulate “still on static”).
# Operator pre-seeds plan to Kafka with {{version=1}} and marker
{{{}updatedBy=brownfield-e2e-seed{}}}.
# Enable dynamic; restart ingestor.
# Ingestor adopts pre-seeded plan — not auto-bootstrap from XML.
# Rollback: turn dynamic off → partition-plan API {*}503{*}, health {*}200{*},
audits still OK in static mode.
h4. 6. Plugin onboard + Kafka partition routing
What it proves: A real plugin can be onboarded via REST and its audit events
land on the correct Kafka partitions defined in the plan.
Flow:
# Enable dynamic mode (greenfield buffer-only layout).
# For each running plugin container (HDFS, Ozone, Hive, etc.): {{POST
/api/audit/partition-plan/plugins}} with {{{}pluginId{}}},
{{{}partitionCount{}}}, {{{}services{}}}.
# Plugin authenticates with Kerberos and POST {{{}/api/audit/access{}}}.
# Read Kafka record for that event → partition number must be in the plugin's
assigned list in the plan (not buffer, after onboard).
h4. 7. auth_to_local recomposition
What it proves: The unified registry {{services}} map controls who may POST
audits, and Kerberos principals are mapped to short usernames correctly.
Flow (per plugin repo):
# Plugin calls POST {{/access}} with Kerberos principal (e.g.
{{{}hdfs/[email protected]{}}}) → ingestor maps via
{{auth_to_local}} → short name {{hdfs}} → *200* if in
{{{}services[dev_hdfs].allowedUsers{}}}.
# Remove user from allowlist via {{{}PATCH .../plugins/{pluginId{}}}}
({{{}updateServices{}}} or {{{}removeServices{}}}) → same principal gets
{*}403{*}; {{removeServices}} drops the whole repo entry and recomposes
{{auth_to_local}} from remaining allowlists.
# Re-add allowlist → *200* again.
# Cross-repo denial: HDFS principal posting to {{dev_kms}} repo → {*}403{*}.
Why it matters: In dynamic mode, allowlists live in the registry (not XML
restart). {{auth_to_local}} rules must stay in sync with
{{{}services[].allowedUsers{}}}.
h4. 8. HDFS / Ozone / Hive full audit pipelines
What it proves: Dynamic partition plan changes do not break the end-to-end
audit path: plugin → ingestor → Kafka → dispatcher → Solr (and Admin Audit UI
when {{{}audit_store=solr{}}}).
||Plugin||What is exercised||
|HDFS|Real NameNode operation → plugin audit → ingestor → correct partition →
Solr doc|
|Ozone|OM principal ({{{}om{}}}, {{{}ozone{}}}) → {{dev_ozone}} repo → full
pipeline|
|Hive|HS2 principal ({{{}hive{}}}) → {{dev_hive}} repo → full pipeline|
Validated after onboard/scale (partition plan changed live), not only at
bootstrap.
Verify Solr:
{{curl -s
'[http://localhost:8983/solr/ranger_audits/select?q=repo:dev_hdfs&rows=3&wt=json'|http://localhost:8983/solr/ranger_audits/select?q=repo:dev_hdfs&rows=3&wt=json%27]}}
h3. Summary table
||Scenario||Plain question||How verified||Pass criteria||
|Static vs dynamic|Does default-off behavior still work?|Feature flag + REST
503/200|Static unchanged; dynamic enables plan API|
|Greenfield bootstrap|Who creates v1 on empty topic?|First ingestor start|Plan
v1 in Kafka; counts match|
|REST onboard/update|Can ops change routing + allowlists live?|REST
mutations|200 on valid; 409 stale; 400 illegal|
|Multi-pod convergence|Do all replicas agree?|2 ingestors, update on one|Same
version ≤35s on both|
|Brownfield pre-seed|Safe production cutover?|Pre-write plan, then
enable|Pre-seed preserved; rollback OK|
|Plugin onboard + routing|Do audits hit right partitions?|POST access + Kafka
inspect|Partition ∈ plan list|
|auth_to_local|Does allowlist enforcement work?|Allow/deny via services
map|200/403 per principal+repo; recomposition on remove|
|Full pipelines|Is audit delivery intact?|HDFS/Ozone/Hive → Solr|Docs in
{{ranger_audits}} Solr collection
|
----
h2. Acceptance criteria
# With {{{}dynamic.enabled=false{}}}, behavior matches existing static XML
partitioning; {{GET /partition-plan}} returns *503*
# With {{{}dynamic.enabled=true{}}}, registry topic created (1 partition,
compacted); bootstrap plan published on greenfield
# {{POST .../plugins}} onboards plugin + services; {{{}PATCH
.../plugins/{pluginId{}}}} scales and/or mutates allowlists; stale
{{expectedVersion}} returns *409*
# All ingestor replicas converge to same plan within watcher refresh interval
# Audits for promoted plugin land only on assigned {{ranger_audits}} partitions
# Service allowlists updated via {{{}PATCH .../plugins/{pluginId{}}}}
({{{}addServices{}}} / {{updateServices}} / {{{}removeServices{}}});
unauthorized principal returns {*}403{*}; {{removeServices}} recomposes
{{auth_to_local}} from remaining allowlists
# Post-scale routing works when Kafka metadata lags plan
({{{}AuditPartitioner{}}} bound logic)
> Implement dynamic unified ingestor registry for audit-ingestor: runtime Kafka
> partition routing and per-repo service allowlists via compacted topic + REST,
> without ingestor restarts. Feature flag default off.
> ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: RANGER-5655
> URL: https://issues.apache.org/jira/browse/RANGER-5655
> Project: Ranger
> Issue Type: Improvement
> Components: Ranger
> Reporter: Ramachandran Krishnan
> Assignee: Ramachandran Krishnan
> Priority: Major
> Fix For: 3.0.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Implement a dynamic unified ingestor registry for Ranger audit-ingestor so
> operators can change Kafka partition routing and per-repo service allowlists
> at runtime — without restarting ingestor pods.
> The registry is stored in a Kafka compacted topic
> ({{ranger_audit_partition_plan}}) and managed via REST. All ingestor replicas
> converge on the same versioned plan through {{PartitionPlanWatcher}};
> {{AuditPartitioner}} routes audit records on the hot path from in-memory
> state only.
> *Feature flag (default off):*
> {{ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled=false}}
> *Pull request:* [GitHub PR #1032|https://github.com/apache/ranger/pull/1032]
> ----
> h2. Problem
> Today (static mode), audit-ingestor loads two kinds of configuration from XML
> at startup only:
> || Job || Question || Static behavior ||
> | Service allowlist | May this Kerberos principal POST audits for repo R? |
> {{ranger.audit.ingestor.service.<repo>.allowed.users}} in site XML |
> | Partition routing | After accept, which {{ranger_audits}} partition? |
> {{kafka.configured.plugins}} + per-plugin overrides in site XML |
> Changing either requires editing XML and restarting every ingestor replica.
> Contiguous-range static allocation can also reshuffle later plugins when an
> early plugin's partition count changes.
> *Goals:*
> * Onboard new plugins/repos and scale hot plugins without ingestor restart
> * Append-only partition growth (no reshuffle of existing plugin assignments)
> * One shared source of truth across all ingestor pods
> * No new infra (no Postgres / ZooKeeper for the registry)
> ----
> h2. Solution
> Introduce a unified partition plan document (versioned JSON) in Kafka topic
> {{ranger_audit_partition_plan}} (1 partition, compacted). One document holds:
> * {{plugins}} — dedicated partition IDs per plugin id (Kafka record key /
> agent id)
> * {{buffer}} — partition pool for not-yet-promoted plugins (sticky hash)
> * {{services}} — per-repo {{allowedUsers}} for {{POST /api/audit/access}}
> (optional {{pluginId}} for ownership)
> * {{topicPartitionCount}} — must match live {{ranger_audits}} partition count
> * {{version}} — optimistic locking for REST mutations
> {code}
> Control plane: GET/PATCH /api/audit/partition-plan → Kafka registry topic
> → PartitionPlanWatcher
> Data plane: POST /api/audit/access (Kerberos) → allowlist check
> → optional auto-onboard
>
> → AuditPartitioner → ranger_audits
> {code}
> Solr/HDFS dispatchers are unchanged; they consume all partitions of
> {{ranger_audits}}.
> h3. REST API (v2 — two public endpoints)
> Plan admin endpoints are *unauthenticated* (same as health/status):
> {{permitAll()}} in Spring Security and bypass in
> {{AuditDelegationTokenFilter}}. Audit ingestion remains Kerberos-protected on
> {{POST /api/audit/access}}.
> All mutations use optimistic locking via {{expectedVersion}} (from GET).
> Stale writes return *409* with the current plan body.
> || Method || Endpoint || Purpose ||
> | GET | {{/api/audit/partition-plan}} | Read current plan (503 when dynamic
> mode off) |
> | PATCH | {{/api/audit/partition-plan}} | Partial update via
> {{PartitionPlanReplacement}} JSON body |
> *PATCH body semantics* ({{PartitionPlanReplacement}}):
> * {{expectedVersion}} — required; stale → 409
> * {{plugins}} / {{services}} — merge *new keys only* (promote/onboard entries
> not already in plan)
> * {{pluginScales}} — map of {{pluginId → additionalPartitions}}; append-only
> tail growth
> * {{buffer}} / {{topicPartitionCount}} — optional; omitted fields inherited
> from current plan
> *Removed (superseded by PATCH + auto-onboard):*
> * {{POST /api/audit/partition-plan/plugins}}
> * {{PATCH /api/audit/partition-plan/plugins/\{pluginId\}}}
> h3. Auto-onboard on audit access
> When dynamic mode is enabled and a Kerberos-authenticated {{POST
> /api/audit/access}} includes {{appId}} for a plugin not yet in the plan:
> # Allowlist check passes for {{serviceName}} + authenticated principal
> # Ingestor promotes the plugin from buffer (default partition count from
> config) and upserts {{services[serviceName]}} in one plan version bump
> # Audit batch is then produced using the updated in-memory plan
> No separate onboard REST call is required for first-time plugin discovery.
> Operators can still pre-promote plugins and allowlists via PATCH.
> h3. Allowlist updates
> {{services}} entries in the registry drive {{POST /access}} authorization. In
> dynamic mode, {{AuthToLocalRuleComposer}} recomposes {{auth_to_local}} from
> the union of all {{services[*].allowedUsers}} whenever the plan changes.
> Removing or tightening a repo's allowlist causes *403* on subsequent access
> attempts without pod restart.
> ----
> h2. Key deliverables (PR #1032)
> * Kafka registry — {{KafkaPartitionPlanRegistry}}, bootstrap from XML on
> greenfield, brownfield pre-seed support
> * REST API — {{GET}} / {{PATCH}} {{/api/audit/partition-plan}} with
> {{PartitionPlanReplacement}} and {{expectedVersion}} (409 on stale)
> * Auto-onboard — {{PartitionPlanService.ensurePluginOnboarded()}} on {{POST
> /api/audit/access}} when {{appId}} is unknown
> * Dynamic partitioner — {{AuditPartitioner}} reads {{PartitionPlanHolder}};
> round-robin for promoted plugins; buffer sticky hash for unknown plugins;
> post-scale routing uses {{max(cluster, plan)}} when metadata lags
> * Unified allowlist — {{services}} map in same registry; per-repo POST checks
> use each repo's own allowlist
> * Topic grow — grow {{ranger_audits}} before registry write when
> promoting/scaling
> * Unit tests — partition-plan, allowlist, partitioner, and bootstrap coverage
> in {{audit-common}} + {{audit-ingestor}}
> *Out of scope for PR #1032 (follow-up / lab validation):* Docker E2E harness,
> ops runbooks, brownfield migration guide (tracked separately in dev-support /
> audit-server docs).
> ----
> h2. Configuration
> {code:xml}
> <property>
> <name>ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled</name>
> <value>true</value>
> </property>
> <property>
> <name>ranger.audit.ingestor.kafka.partition.plan.topic</name>
> <value>ranger_audit_partition_plan</value>
> </property>
> {code}
> When dynamic mode is on and the registry topic is empty, the first ingestor
> pod bootstraps an initial plan from existing XML properties
> ({{kafka.configured.plugins}}, buffer/per-plugin counts, service allowlists).
> h3. Example plan JSON
> {code:json}
> {
> "topic": "ranger_audits",
> "version": 12,
> "topicPartitionCount": 48,
> "plugins": {
> "hdfs": { "partitions": [0, 1, 2, 3, 4, 5] },
> "hiveServer2": { "partitions": [6, 7, 8, 9, 10, 11] }
> },
> "buffer": { "partitions": [12, 13, "..."] },
> "services": {
> "dev_hive": { "allowedUsers": ["hive"], "pluginId": "hiveServer2" },
> "dev_ozone": { "allowedUsers": ["om", "ozone"], "pluginId": "ozone" },
> "dev_hdfs": { "allowedUsers": ["hdfs", "nn"], "pluginId": "hdfs" }
> }
> }
> {code}
> h3. Example PATCH (promote + scale)
> {code:json}
> {
> "expectedVersion": 12,
> "plugins": {
> "storm": { "partitions": [12, 13, 14] }
> },
> "services": {
> "dev_storm": { "allowedUsers": ["storm"], "pluginId": "storm" }
> },
> "pluginScales": {
> "hdfs": 2
> }
> }
> {code}
> ----
> h2. Testing
> h3. Unit tests and quality gates
> {code:bash}
> mvn verify -pl audit-server/audit-common,audit-server/audit-ingestor
> -Drat.skip=true
> {code}
> || Gate || Result ||
> | Unit tests | 104 passed (audit-ingestor module) |
> | Checkstyle | Pass |
> | PMD / SpotBugs | Pass (existing audit-ingestor findings unchanged) |
> h3. Validated scenarios (manual + lab)
> || # || Scenario || Pass criteria ||
> | 1 | Static mode (default off) | {{GET /partition-plan}} → 503; health 200;
> plugin audits unchanged |
> | 2 | Dynamic bootstrap | Plan topic created; watcher active; {{GET
> /partition-plan}} → 200 with v≥1 |
> | 3 | Public plan REST | {{GET}} / {{PATCH}} without Kerberos succeed when
> dynamic mode on |
> | 4 | PATCH promote/onboard | New {{plugins}} + {{services}} keys merged in
> one version bump |
> | 5 | PATCH {{pluginScales}} | Tail partitions appended; {{ranger_audits}}
> grown; no reshuffle |
> | 6 | Optimistic locking | Stale {{expectedVersion}} → 409 with current plan
> body |
> | 7 | Auto-onboard on {{/access}} | Unknown {{appId}} + allowlisted principal
> → plan updated, audit accepted |
> | 8 | Multi-pod convergence | All replicas report same version after PATCH on
> one pod |
> | 9 | Routing | Kafka record partition ∈ plugin assignment from plan |
> | 10 | Allowlist enforcement | Allow/deny/cross-repo → 200/403; recomposition
> on plan change |
> ----
> h2. Acceptance criteria
> # With {{dynamic.enabled=false}}, behavior matches existing static XML
> partitioning; {{GET /partition-plan}} returns 503
> # With {{dynamic.enabled=true}}, registry topic created (1 partition,
> compacted); bootstrap plan published on greenfield
> # {{PATCH /api/audit/partition-plan}} promotes/scales/mutates allowlists;
> stale {{expectedVersion}} returns 409
> # Unknown {{appId}} on authenticated {{POST /access}} auto-onboards plugin
> when allowlist permits
> # All ingestor replicas converge to same plan within watcher refresh interval
> # Audits for promoted plugin land only on assigned {{ranger_audits}}
> partitions
> # Service allowlists in registry drive {{/access}} authorization;
> unauthorized principal returns 403
> # Post-scale routing works when Kafka metadata lags plan
> ({{AuditPartitioner}} bound logic)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
