HDFS-17974. Add pluggable DataNode affinity for tenant isolation in block placement - #8715
Open
rdhabalia wants to merge 1 commit into
Open
HDFS-17974. Add pluggable DataNode affinity for tenant isolation in block placement#8715rdhabalia wants to merge 1 commit into
rdhabalia wants to merge 1 commit into
Conversation
|
💔 -1 overall
This message was automatically generated. |
rdhabalia
force-pushed
the
affinity-tenant-isolation
branch
2 times, most recently
from
September 5, 2026 05:03
a42c953 to
f2c52f6
Compare
…lacement Motivation: Multi-tenant HDFS clusters often need to pin a subset of data (identified by its HDFS path) to a dedicated pool of DataNodes so that a noisy or isolation-sensitive tenant does not share write/read capacity with the rest of the cluster. The default BlockPlacementPolicy has no notion of such path-to-DataNode affinity, so operators resort to separate clusters or brittle rack hacks. Approach: Introduce a pluggable DatanodeAffinityManager abstraction resolved by reflection from `dfs.datanode.affinity.manager.classname`. An affinity group maps a source-path regex to a DataNode-hostname regex. On refresh the manager resolves each DataNode-hostname regex against the live cluster and builds, per group, a restricted NetworkTopology containing only that group's eligible DataNodes. - DatanodeManager instantiates the configured manager, removes affinity DataNodes from the default NetworkTopology as they register (so the default policy can never select them), prunes them from the affinity structures when they are decommissioned/removed, and re-triggers a refresh on `dfsadmin -refreshNodes`. - BlockManager keeps one BlockPlacementPolicies per affinity group, each backed by the group's restricted topology. Both initial block placement (chooseTarget4NewBlock) and mid-write pipeline recovery (chooseTarget4AdditionalDatanode) route through the matching group's policy so a replacement node for a failed pipeline member stays inside the isolated pool. When no group matches, the default policy is used. - A built-in FileDatanodeAffinityManager loads affinity groups from a JSON file (`dfs.datanode.affinity.file.path`), reloaded on `dfsadmin -refreshNodes`. - `hdfs fsck <path> -favored-nodes` prints the affinity-resolved DataNodes for a path as a dry-run, without requiring the path to exist. Isolation vs. availability: When an affinity group cannot place all requested replicas (under-provisioned group, all group DataNodes down, or a datanodesRegex that matched no nodes), the default behaviour falls back to the shared pool so the write still succeeds, logging the spillover at WARN. Setting `dfs.namenode.affinity.strict.isolation.enabled=true` makes initial placement fail closed unless ALL requested replicas fit inside the group -- placing only minReplication in-group would let the redundancy monitor repair the remainder on shared-pool nodes and leak data out of the group. Scope / limitations: Affinity governs initial block placement and in-pipeline recovery. Background replication / EC reconstruction still uses the default placement policy: making it affinity-aware requires a reverse block->path lookup inside a lock-sensitive hot loop and is intentionally left as follow-up work (documented in BlockManager). DataNode-regex patterns are matched (via find()) against each node's "hostname:port" address, so a fully anchored pattern must allow the trailing ":port" (e.g. `^dn-tenant-a[0-9]+\.example\.com(:\d+)?$`). Configuration: - dfs.datanode.affinity.manager.classname (empty/disabled by default) - dfs.datanode.affinity.file.path (JSON file for FileDatanodeAffinityManager) - dfs.namenode.affinity.strict.isolation.enabled (default false) Concurrency and correctness hardening: - Both the strict and non-strict modes now require the affinity group to place ALL requested replicas before returning in-group targets. Accepting a partial placement (only >= minReplication) would create the block under-replicated and let the redundancy monitor "repair" the remainder onto shared-pool nodes -- leaking data out of the isolated group (strict) or leaving a silent under-replication window (non-strict). On a shortfall strict fails closed and non-strict spills the WHOLE block to the shared pool via the default policy, so it is fully replicated at write time. - Symmetric registration/removal races during a lock-free config refresh are reconciled: internalRefresh() takes a live snapshot before publishing, then (a) postAffinityRefresh() re-evaluates every live DataNode against the current patterns so a node that registered mid-refresh stays isolated instead of leaking back into the default topology, and (b) a post-publish purge re-runs onDatanodeRemoved() for any snapshot node no longer live so a node removed mid-refresh cannot linger as a dead entry in the freshly published structures. The isolated-address set is a stable final set mutated in place (never swapped) so concurrent registration adds are not lost. - DataNode replacement (same host:port re-registering with a new descriptor / storage UUID) is reconciled in the per-group topology: because that topology stores descriptor OBJECTS but membership is address-keyed, a replacement could otherwise leave a STALE, dead descriptor selectable for placement. onDatanodeRegistered() now swaps the stale descriptor for the live one (object-identity guarded), removing every stale descriptor for the same transfer endpoint regardless of the topology path it sits under (so a replacement that also changes rack/IP cannot leave a dead leaf behind), and postAffinityRefresh() routes every live node through it (no address short-circuit) so the reconciliation always runs. - The per-group restricted topology is never a storage-type-aware DFSNetworkTopology. A DFSNetworkTopology captures each node's storage-type counts at add() time and updates them only through the descriptor's parent back-pointer; but an affinity node's shared descriptor has its parent nulled when it is removed from the default topology, so those counts could never update from later storage reports. Because a DataNode registers (and is added to its group) with an EMPTY storage map -- storage reports arrive afterwards -- a DFSNetworkTopology group would freeze the count at ZERO and, via storage-type-gated selection, return no target: group placement would silently break after every NameNode restart (all affinity nodes re-register fresh) until a manual `dfsadmin -refreshNodes`. So when the cluster uses DFSNetworkTopology (the default) the group topology is built as a plain NetworkTopology, which tracks no storage counts; the placement policy instead selects a random eligible leaf and validates the storage type against the LIVE descriptor, which is always current. When dfs.use.dfs.network.topology=false the configured net.topology.impl is honored instead (e.g. NetworkTopologyWithNodeGroup for a node-group replicator, whose placement policy's initialize() rejects any other clusterMap type) -- those impls track no storage counts, so they are immune to the staleness problem while keeping the isolation feature from silently failing open under a node-group topology. - Pipeline recovery (chooseTarget4AdditionalDatanode) now routes through the affinity group's policy ONLY when the surviving pipeline replicas actually live inside that group's restricted topology; otherwise it uses the default policy. In non-strict availability mode an under-provisioned group spills the WHOLE block to the shared pool, so the survivors are shared-pool nodes absent from the tiny group topology. Forcing the group policy would then count those out-of-topology survivors against the group in getMaxNodesPerRack (numChosen + numAdditional exceeds the group's leaf count), drive the additional count to 0, return no replacement, and -- with the client default best-effort=false -- fail the write. Recovering a spilled block through the default policy keeps the write alive and never leaks isolation (the block already lives in the shared pool, and affinity nodes are absent from the default topology). Testing: TestDatanodeAffinityBlockPlacement (14 tests) covers: all replicas of an affinity path landing only on the group's DataNodes; non-affinity paths using the full cluster; isolation surviving DataNode re-registration; stale-node pruning on removal; strict isolation failing both an unsatisfiable and a partially-provisioned write; non-strict fallback succeeding via the shared pool for both an empty group and a partially-provisioned group (whole block spilled, no replica left on the under-provisioned group); pipeline recovery of a spilled (shared-pool) block succeeding via the default policy while in-group recovery stays in-group; malformed (null-regex) records being skipped without aborting refresh; a DataNode that registers during a config refresh being reconciled into its group instead of leaking back into the default topology; a replaced DataNode (same address, new UUID) having its group topology reconciled to the live descriptor, both in place and when the replacement moves to a different rack/path; and the per-group topology being a plain NetworkTopology (not a storage-type-aware DFSNetworkTopology) so its node storage accounting cannot go stale for affinity nodes. The removal-race purge reuses onDatanodeRemoved(), whose pruning of the published structures is covered by the removal test; the exact mid-refresh interleaving is not unit-tested because it cannot be made deterministic without intrusive production test hooks. Results: Enables strict tenant isolation of write and read traffic onto a dedicated DataNode pool while leaving the rest of the cluster on the default placement policy, refreshable at runtime without a NameNode restart.
rdhabalia
force-pushed
the
affinity-tenant-isolation
branch
from
September 5, 2026 06:06
f2c52f6 to
2ae7b9d
Compare
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
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
2 participants
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.
Description of PR
Multi-tenant HDFS clusters often need to pin a subset of data, identified by its HDFS path, to a dedicated pool of DataNodes. This prevents noisy or isolation-sensitive tenants from sharing read/write capacity with the rest of the cluster.
The default
BlockPlacementPolicyhas no notion of path-to-DataNode affinity, so operators typically resort to separate clusters or brittle rack-based workarounds.Approach
Introduce a pluggable
DatanodeAffinityManagerabstraction, resolved via reflection from:An affinity group maps:
On refresh, the manager resolves each DataNode hostname regex against the live cluster and builds a restricted
NetworkTopologyfor each group containing only that group's eligible DataNodes.Components
DatanodeManagerNetworkTopologyas they register, ensuring the default placement policy cannot select them.dfsadmin -refreshNodes.BlockManagerBlockPlacementPolicyper affinity group.chooseTarget4NewBlock) through the matching group's policy.chooseTarget4AdditionalDatanode) through the matching group's policy.FileDatanodeAffinityManagerBuilt-in implementation that loads affinity groups from a JSON file.
Configuration:
Reloads the configuration on
dfsadmin -refreshNodes.hdfs fsckSupports:
Prints the affinity-resolved DataNodes for a path as a dry run.
Does not require the path to exist.
Isolation vs. Availability
If an affinity group cannot place all requested replicas because of:
datanodesRegexmatching no nodes,the default behavior falls back to the shared pool so the write can still succeed. Spillover is logged at
WARN.Strict isolation can be enabled with:
dfs.namenode.affinity.strict.isolation.enabled=trueIn strict mode, initial placement fails closed unless all requested replicas can be placed inside the affinity group.
This is important because placing only
minReplicationreplicas in-group would allow the redundancy monitor to repair the remaining replicas onto shared-pool DataNodes, leaking data outside the isolated group.Scope and Limitations
Affinity currently governs:
Background replication and EC reconstruction still use the default placement policy.
Making these paths affinity-aware would require a reverse block-to-path lookup inside a lock-sensitive hot loop and is intentionally left as follow-up work. This is documented in
BlockManager.DataNode Regex Matching
DataNode regex patterns are matched using
find()against each node's:Therefore, a fully anchored pattern must account for the trailing port.
For example:
Configuration
dfs.datanode.affinity.manager.classnamedfs.datanode.affinity.file.pathFileDatanodeAffinityManagerdfs.namenode.affinity.strict.isolation.enabledfalseConcurrency and Correctness Hardening
1. Full-Replica Placement
Both strict and non-strict modes require the affinity group to place all requested replicas before returning in-group targets.
Accepting a partial placement, even when it satisfies
minReplication, creates an under-replicated block. The redundancy monitor could then repair the remaining replicas onto shared-pool DataNodes.Behavior on shortfall:
This guarantees that the block is fully replicated at write time and avoids both isolation leaks and silent under-replication windows.
2. Registration/Removal Races During Refresh
Registration and removal races during lock-free configuration refresh are reconciled.
internalRefresh()first takes a live snapshot before publishing the refreshed state. After publishing:postAffinityRefresh()re-evaluates every live DataNode against the current affinity patterns.A post-publish purge re-runs
onDatanodeRemoved()for any snapshot node that is no longer live.The isolated-address set is a stable set that is mutated in place rather than replaced. This ensures concurrent registration updates are not lost.
3. DataNode Replacement
A DataNode can re-register with the same
host:portbut a new descriptor or storage UUID.Affinity group topologies store descriptor objects while membership is address-keyed. Without reconciliation, a replacement could leave a stale descriptor selectable for placement.
onDatanodeRegistered()now:postAffinityRefresh()routes every live DataNode through this reconciliation logic rather than using an address-based short circuit.4. Per-Group Topology and Storage Types
The per-group restricted topology is intentionally not a storage-type-aware
DFSNetworkTopology.DFSNetworkTopologycaptures each node's storage-type counts when the node is added and updates them through the descriptor's parent back-pointer.Affinity DataNodes are removed from the default topology, which nulls their parent back-pointer. Consequently, storage-type counts in a
DFSNetworkTopologyaffinity topology could become stale.This is particularly problematic because:
DFSNetworkTopologygroup would retain a storage count of zero.After a NameNode restart, this could cause affinity placement to silently fail until a manual:
To avoid this:
DFSNetworkTopology, each affinity group uses a plainNetworkTopology.This ensures storage information remains current.
When:
dfs.use.dfs.network.topology=falsethe configured
net.topology.implis honored instead.For example:
can be used for a node-group-based placement policy. Such implementations do not maintain storage counts and therefore do not have the same stale-storage-accounting issue.
The placement policy's
initialize()still validates that the supplied cluster map is compatible with the configured topology implementation.5. Pipeline Recovery
chooseTarget4AdditionalDatanode()routes through the affinity group's policy only when the surviving pipeline replicas actually belong to that group's restricted topology.Otherwise, it uses the default placement policy.
This distinction is required for non-strict availability mode:
getMaxNodesPerRack()calculation.numChosen + numAdditionalcould exceed the group's leaf count.best-effort=false, the write could fail.Using the default policy for a spilled block avoids this failure.
It also does not introduce an isolation leak because:
Testing
TestDatanodeAffinityBlockPlacementcontains 14 tests covering:All replicas for an affinity path landing only on the group's DataNodes.
Non-affinity paths using the full cluster.
Isolation surviving DataNode re-registration.
Stale-node pruning on DataNode removal.
Strict isolation failing for:
Non-strict fallback succeeding through the shared pool for:
Whole-block spillover with no replica remaining on the under-provisioned affinity group.
Pipeline recovery of a spilled shared-pool block succeeding through the default policy.
In-group pipeline recovery remaining within the affinity group.
Malformed records with null regex values being skipped without aborting refresh.
DataNodes registering during configuration refresh being reconciled into their affinity group instead of leaking into the default topology.
Replaced DataNodes with the same address but a new UUID being reconciled to the live descriptor.
DataNode replacement working both:
Per-group topology being a plain
NetworkTopologyrather than a storage-type-awareDFSNetworkTopology.Storage accounting remaining current for affinity DataNodes.
The removal-race purge reuses
onDatanodeRemoved(), whose pruning behavior is covered by the removal test.The exact mid-refresh interleaving is not unit-tested because reproducing it deterministically would require intrusive production test hooks.
Results
The implementation enables strict tenant isolation of write and read traffic onto a dedicated DataNode pool while keeping the rest of the cluster on the default placement policy.
Affinity configuration can be refreshed at runtime without requiring a NameNode restart.
How was this patch tested?
Tested by newly added unit test
For code changes:
declared according to the connector-specific documentation? Note: Automated CI
testing doesn't cover all cases so manual testing with cloud storage is still
required.
LICENSE,LICENSE-binary,NOTICE-binaryfiles?AI Tooling
If an AI tool was used:
where is the name of the AI tool used.
https://www.apache.org/legal/generative-tooling.html