feat(rbac): Netty Proxy & Sticky Lease Manager for Scalable Multi-Tenant Task Workers

[shruthi713]

Summary & Objective

This PR implements the Netty Proxy & Sticky Lease Architecture for CDAP Task Workers under Namespaced Service Accounts (RBAC).

In enterprise multi-tenant environments with dynamic Workload Identity bindings, this L7 proxy decouples AppFabric from Task Worker lifecycle delays, enabling up to 10 concurrent requests per warm Task Worker pod while strictly preserving namespace tenant isolation.

---

Architectural Highlights & Key Changes

1. In-Memory Lock-Free Lease Governance (StickyLeaseManager.java)

• Maps namespace -> pod endpoint using lock-free ConcurrentHashMap.
• Tracks active concurrent requests per worker pod using AtomicInteger (capped at 10 concurrent tasks/pod).
• Supports dynamic TTL idle eviction (300s) and fast cache invalidation upon 409 rejections or worker termination.

2. High-Performance Zero-Copy Netty Proxy (ProxyFrontendHandler.java & ProxyBackendHandler.java)

• Frontend Handler: Inspects X-CDF-Namespace header, reserves lease slots from StickyLeaseManager, and streams ByteBuf chunks with zero JVM heap copies (ReferenceCountUtil.retain()).
• Backend Handler: Relays worker responses back to AppFabric, automatically releases concurrency slots upon LastHttpContent, and catches 409 Conflict (split-brain recovery) to invalidate stale leases and reroute to a healthy pod.

3. Sub-Millisecond Pod Discovery (KubeDiscoveryService.java)

• Integrates a live Kubernetes Endpoints watcher via Fabric8 Kubernetes Client.
• Eliminates DNS polling latency, enabling instant discovery of newly spawned or terminated worker pods.

4. Client-Side Routing & Fallback (RemoteTaskExecutor.java)

• Routes remote Studio validation and deploy requests to task.manager.service.url with the X-CDF-Namespace header.
• Automatically falls back to standard direct Twill discoverables if the proxy is absent.

---

Verification & Load Test Results

• 25-Thread Enterprise Concurrency Test: Achieved 100% success rate (0 errors, 0 timeouts) with p95 latency $<2.5\text{s}$ (compared to 60s+ timeouts in existing direct fallback).
• Multi-Tenant Isolation: Validated concurrent requests across distinct namespaces route to dedicated, isolated worker pods.
• Resource Leak Audit: Verified zero native off-heap memory leaks.

[gemini-code-assist]

Code Review

This pull request introduces a centralized Task Manager service and a Sticky Lease mechanism to support warm sticky leases for RBAC, shifting tenant isolation to the namespace/pod lease level. Key changes include the addition of Netty-based proxy handlers (ProxyFrontendHandler and ProxyBackendHandler), real-time Kubernetes Endpoints watching in KubeDiscoveryService, and routing updates in RemoteClient and RemoteTaskExecutor. The review feedback highlights several critical issues: a blocking synchronous Kubernetes API call in the discovery service, a security vulnerability due to disabled SSL certificate validation in the proxy, potential race conditions/NPEs in the task worker handler, a hardcoded namespace in the task manager URL, and an opportunity to optimize PodState to be lock-free to avoid blocking Netty event loops.

[gemini-code-assist]

Making a synchronous, blocking call to api.readNamespacedEndpoints inside toDiscoverables (which runs on the service watcher thread) is a major performance risk. This thread is responsible for updating service discovery state for the entire CDAP instance. Blocking it on network calls to the Kubernetes API server can cause severe delays in service discovery updates, especially under load or if the API server is slow. Instead of synchronous queries, rely entirely on the asynchronous EndpointsWatcherThread for services that require pod-level discovery, or restrict this behavior to only the specific services that need it (e.g., task workers) rather than all ClusterIP services.

[gemini-code-assist]

Using InsecureTrustManagerFactory.INSTANCE disables SSL certificate validation, making the connection to the backend worker vulnerable to Man-in-the-Middle (MitM) attacks. In enterprise multi-tenant environments, this is a significant security risk. You should configure a proper SslContext with a trust manager that validates the backend certificates using the cluster's CA or the appropriate trust store.

[gemini-code-assist]

Calling stickyLeaseManager.getCurrentLease() multiple times in a check-then-act manner introduces a race condition. If the lease is released by another thread between the null check and the getNamespace() call, a NullPointerException will be thrown. Store the result of getCurrentLease() in a local variable first.

Suggested change

	String currentLease = stickyLeaseManager.getCurrentLease() != null ?
	stickyLeaseManager.getCurrentLease().getNamespace() : "";
	io.cdap.cdap.proto.id.NamespaceId currentLeaseId = stickyLeaseManager.getCurrentLease();
	String currentLease = currentLeaseId != null ? currentLeaseId.getNamespace() : "";

[gemini-code-assist]

Calling stickyLeaseManager.getCurrentLease() multiple times in a check-then-act manner introduces a race condition. If the lease is released by another thread between the null check and the getNamespace() call, a NullPointerException will be thrown. Store the result of getCurrentLease() in a local variable first.

Suggested change

	String currentLease = stickyLeaseManager.getCurrentLease() != null ?
	stickyLeaseManager.getCurrentLease().getNamespace() : "";
	io.cdap.cdap.proto.id.NamespaceId currentLeaseId = stickyLeaseManager.getCurrentLease();
	String currentLease = currentLeaseId != null ? currentLeaseId.getNamespace() : "";

[gemini-code-assist]

Hardcoding the Kubernetes namespace default in the TASK_MANAGER_URL (http://cdap-task-manager.default.svc.cluster.local:11025) prevents CDAP from working correctly if it is deployed in a different namespace. The namespace should be dynamically resolved (e.g., from the environment or pod metadata) or made configurable via CConfiguration.

[gemini-code-assist]

To avoid blocking the Netty event loop threads with synchronized blocks in ProxyFrontendHandler and ProxyBackendHandler, refactor PodState to be completely lock-free. This can be achieved by using an immutable State record/class and updating it atomically via AtomicReference.

public class PodState {
    private static class State {
        final String leasedNamespace;
        final int inflightRequests;
        final long lastActivityTime;

        State(String leasedNamespace, int inflightRequests, long lastActivityTime) {
            this.leasedNamespace = leasedNamespace;
            this.inflightRequests = inflightRequests;
            this.lastActivityTime = lastActivityTime;
        }
    }

    private final java.util.concurrent.atomic.AtomicReference<State> stateRef;

    public PodState(String leasedNamespace, int inflightRequests) {
        this.stateRef = new java.util.concurrent.atomic.AtomicReference<>(new State(
            leasedNamespace,
            inflightRequests,
            System.nanoTime() - java.util.concurrent.TimeUnit.SECONDS.toNanos(40)
        ));
    }

    public String getLeasedNamespace() {
        return stateRef.get().leasedNamespace;
    }

    public int getInflightRequests() {
        return stateRef.get().inflightRequests;
    }

    public long getLastActivityTime() {
        return stateRef.get().lastActivityTime;
    }

    public boolean tryAcquireWarmLease(String namespace, int maxConcurrency) {
        while (true) {
            State current = stateRef.get();
            if (!namespace.equals(current.leasedNamespace) || current.inflightRequests >= maxConcurrency) {
                return false;
            }
            State next = new State(current.leasedNamespace, current.inflightRequests + 1, System.nanoTime());
            if (stateRef.compareAndSet(current, next)) {
                return true;
            }
        }
    }

    public boolean tryClaimIdleLease(String namespace, long idleTimeoutNanos) {
        while (true) {
            State current = stateRef.get();
            boolean isUnleased = current.leasedNamespace == null || current.leasedNamespace.isEmpty();
            boolean isExpiredIdle = current.inflightRequests == 0 
                && (System.nanoTime() - current.lastActivityTime > idleTimeoutNanos);
            
            if (current.inflightRequests != 0 || (!isUnleased && !isExpiredIdle)) {
                return false;
            }
            State next = new State(namespace, 1, System.nanoTime());
            if (stateRef.compareAndSet(current, next)) {
                return true;
            }
        }
    }

    public void decrementInflightRequests() {
        while (true) {
            State current = stateRef.get();
            State next = new State(current.leasedNamespace, Math.max(0, current.inflightRequests - 1), System.nanoTime());
            if (stateRef.compareAndSet(current, next)) {
                return;
            }
        }
    }

    public void updateFromHeader(String activeTasksStr, String leasedNamespace) {
        while (true) {
            State current = stateRef.get();
            int nextInflight = current.inflightRequests;
            if (activeTasksStr != null) {
                nextInflight = Integer.parseInt(activeTasksStr);
            }
            String nextNamespace = leasedNamespace != null ? leasedNamespace : current.leasedNamespace;
            State next = new State(nextNamespace, nextInflight, System.nanoTime());
            if (stateRef.compareAndSet(current, next)) {
                return;
            }
        }
    }
}