[AUTHENTICATION] Add UnboundID-backed LDAP connection pool with failover and cache

LICENSE-binary

@@ -213,6 +213,7 @@ org.apache.commons:commons-lang3
 com.google.errorprone:error_prone_annotations
 dev.failsafe:failsafe
 com.jakewharton.fliptables:fliptables
+com.unboundid:unboundid-ldapsdk
 io.grpc:grpc-api
 io.grpc:grpc-context
 io.grpc:grpc-core

dev/dependencyList

@@ -184,6 +184,7 @@ swagger-integration/2.2.1//swagger-integration-2.2.1.jar
 swagger-jaxrs2/2.2.1//swagger-jaxrs2-2.2.1.jar
 swagger-models/2.2.1//swagger-models-2.2.1.jar
 trino-client/411//trino-client-411.jar
+unboundid-ldapsdk/6.0.5//unboundid-ldapsdk-6.0.5.jar
 units/1.7//units-1.7.jar
 vertx-core/4.5.3//vertx-core-4.5.3.jar
 vertx-grpc/4.5.3//vertx-grpc-4.5.3.jar

kyuubi-common/pom.xml

@@ -150,7 +150,6 @@
         <dependency>
             <groupId>com.unboundid</groupId>
             <artifactId>unboundid-ldapsdk</artifactId>
-            <scope>test</scope>
         </dependency>
 
         <dependency>

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -1089,6 +1089,136 @@ object KyuubiConf {
       .stringConf
       .createOptional
 
+  val AUTHENTICATION_LDAP_POOL_ENABLED: ConfigEntry[Boolean] =
+    buildConf("kyuubi.authentication.ldap.pool.enabled")
+      .doc("Whether to use a persistent UnboundID LDAP connection pool for LDAP authentication. " +
+        "When enabled, the server maintains warm pre-authenticated connections to LDAP, " +
+        "eliminating per-request TCP/TLS/BIND overhead. Requires bind user credentials to be " +
+        "configured. If the bind user is not set the pool is silently skipped and ephemeral " +
+        "connections are used. " +
+        "Setting this to false falls back to the legacy JNDI implementation; that path is " +
+        "scheduled for removal in a future release. Both paths derive SSL purely from the URL " +
+        "scheme (ldap:// vs ldaps://) -- StartTLS and LDAP referrals are not supported on " +
+        "either path, matching the behaviour of the legacy JNDI implementation. " +
+        "All kyuubi.authentication.ldap.* settings are read once on the first authentication " +
+        "request after server start and cached for the JVM lifetime; changes require a Kyuubi " +
+        "restart to take effect.")
+      .version("1.12.0")
+      .serverOnly
+      .booleanConf
+      .createWithDefault(false)
+
+  val AUTHENTICATION_LDAP_POOL_MIN_CONNECTIONS: ConfigEntry[Int] =
+    buildConf("kyuubi.authentication.ldap.pool.minConnections")
+      .doc("Minimum number of connections to keep open in the LDAP connection pool.")
+      .version("1.12.0")
+      .serverOnly
+      .intConf
+      .createWithDefault(3)
+
+  val AUTHENTICATION_LDAP_POOL_MAX_CONNECTIONS: ConfigEntry[Int] =
+    buildConf("kyuubi.authentication.ldap.pool.maxConnections")
+      .doc("Maximum number of connections in the LDAP connection pool.")
+      .version("1.12.0")
+      .serverOnly
+      .intConf
+      .createWithDefault(10)
+
+  val AUTHENTICATION_LDAP_POOL_CONNECT_TIMEOUT: ConfigEntry[Long] =
+    buildConf("kyuubi.authentication.ldap.pool.connectTimeout")
+      .doc("Timeout for establishing a TCP connection to an LDAP server.")
+      .version("1.12.0")
+      .serverOnly
+      .timeConf
+      .createWithDefaultString("PT5S")
+
+  val AUTHENTICATION_LDAP_POOL_RESPONSE_TIMEOUT: ConfigEntry[Long] =
+    buildConf("kyuubi.authentication.ldap.pool.responseTimeout")
+      .doc("Timeout for an LDAP operation response after the connection is established.")
+      .version("1.12.0")
+      .serverOnly
+      .timeConf
+      .createWithDefaultString("PT10S")
+
+  val AUTHENTICATION_LDAP_POOL_HEALTH_CHECK_INTERVAL: ConfigEntry[Long] =
+    buildConf("kyuubi.authentication.ldap.pool.healthCheck.interval")
+      .doc("Interval between background health checks on idle LDAP pool connections. " +
+        "Each check fetches the root DSE; connections that fail are discarded and replaced.")
+      .version("1.12.0")
+      .serverOnly
+      .timeConf
+      .createWithDefaultString("PT30S")
+
+  val AUTHENTICATION_LDAP_POOL_CHECKOUT_TIMEOUT: ConfigEntry[Long] =
+    buildConf("kyuubi.authentication.ldap.pool.checkoutTimeout")
+      .doc("Maximum time a request thread will wait for a connection to become available " +
+        "when the LDAP pool is fully utilised. The UnboundID SDK default is to skip waiting " +
+        "and immediately try to grow the pool: if a new connection can be created the " +
+        "request succeeds, otherwise it fails fast (typically with CONNECT_ERROR, or with " +
+        "the result code from the underlying connection attempt if creation was tried). " +
+        "A positive value here adds a grace window so brief contention spikes do not " +
+        "surface as authentication failures. Set to 0 to revert to the SDK default no-wait " +
+        "behaviour. The default is intentionally short -- long waits pile up front-end " +
+        "Thrift/HTTP handler threads under a degraded LDAP. Operators who raise this should " +
+        s"also raise ${"kyuubi.authentication.ldap.pool.maxConnections"}.")
+      .version("1.12.0")
+      .serverOnly
+      .timeConf
+      .createWithDefaultString("PT5S")
+
+  val AUTHENTICATION_LDAP_POOL_SEARCH_MAX_RETRIES: ConfigEntry[Int] =
+    buildConf("kyuubi.authentication.ldap.pool.search.maxRetries")
+      .doc("Maximum number of times a pooled LDAP search is retried against a freshly " +
+        "checked-out connection when the original connection fails mid-request with a " +
+        "connection-level error (server down, dropped socket, idle connection reset). " +
+        "Bounded retry covers the common case where one LDAP server in the failover list " +
+        "rotates or briefly disconnects, without amplifying load during a real outage. " +
+        "Set to 0 to disable retry and propagate the first failure. Only applies to the " +
+        "pooled (bind-user) path; the ephemeral end-user bind path does not retry because " +
+        "its connection is tied to specific credentials that cannot be re-established " +
+        "transparently.")
+      .version("1.12.0")
+      .serverOnly
+      .intConf
+      .checkValue(_ >= 0, "must be non-negative")
+      .createWithDefault(1)
+
+  val AUTHENTICATION_LDAP_CACHE_ENABLED: ConfigEntry[Boolean] =
+    buildConf("kyuubi.authentication.ldap.cache.enabled")
+      .doc("Whether to cache successful LDAP authentications within a short TTL to reduce " +
+        "LDAP round-trips for repeated requests from the same user. " +
+        "Only successful authentications are cached; failed attempts always reach LDAP.")
+      .version("1.12.0")
+      .serverOnly
+      .booleanConf
+      .createWithDefault(true)
+
+  val AUTHENTICATION_LDAP_CACHE_TTL: ConfigEntry[Long] =
+    buildConf("kyuubi.authentication.ldap.cache.ttl")
+      .doc("Time-to-live for a cached successful LDAP authentication result. " +
+        "After this period the entry is evicted and the next request re-authenticates " +
+        "against LDAP. Time-based eviction applies in addition to the size cap configured " +
+        "by kyuubi.authentication.ldap.cache.maxSize. " +
+        "Operational note: after an LDAP-side password rotation, account lockout, or user " +
+        "disable, the cached (user, password) pair continues to be accepted by Kyuubi until " +
+        "this TTL expires. Lower this value in environments with strict revocation SLAs; " +
+        "raise it to reduce LDAP load.")
+      .version("1.12.0")
+      .serverOnly
+      .timeConf
+      .createWithDefaultString("PT30S")
+
+  val AUTHENTICATION_LDAP_CACHE_MAX_SIZE: ConfigEntry[Int] =
+    buildConf("kyuubi.authentication.ldap.cache.maxSize")
+      .doc("Maximum number of entries in the LDAP authentication cache. " +
+        "When the limit is reached, least-recently-used entries are evicted. " +
+        "Size-based eviction applies in addition to the TTL configured by " +
+        "kyuubi.authentication.ldap.cache.ttl.")
+      .version("1.12.0")
+      .serverOnly
+      .intConf
+      .createWithDefault(1000)
+
   val AUTHENTICATION_JDBC_DRIVER: OptionalConfigEntry[String] =
     buildConf("kyuubi.authentication.jdbc.driver.class")
       .doc("Driver class name for JDBC Authentication Provider.")

kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/AuthenticationProviderFactory.scala

@@ -17,18 +17,28 @@
 
 package org.apache.kyuubi.service.authentication
 
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicLong
 import javax.security.sasl.AuthenticationException
 
 import org.apache.commons.lang3.StringUtils
 
+import org.apache.kyuubi.Logging
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.service.authentication.AuthMethods.AuthMethod
+import org.apache.kyuubi.service.authentication.ldap.{CachingLdapAuthenticationProvider, LdapAuthCache, LdapAuthFailureClassifier, LdapSearchFactory, UnboundIdConnectionPool, UnboundIdDirSearchFactory}
 import org.apache.kyuubi.util.ClassUtils
 
 /**
  * This class helps select a [[PasswdAuthenticationProvider]] for a given [[AuthMethods]]
  */
-object AuthenticationProviderFactory {
+object AuthenticationProviderFactory extends Logging {
+
+  // Singletons keyed by auth scope ("server" or "http"). The LDAP stack must be a singleton
+  // so the connection pool and auth cache are shared across all authentication requests -- not
+  // recreated per request. JDBC providers are stateless, so per-call creation is fine there.
+  private val ldapProviders = new ConcurrentHashMap[String, PasswdAuthenticationProvider]()
+
   @throws[AuthenticationException]
   def getAuthenticationProvider(
       method: AuthMethod,
@@ -45,7 +55,7 @@ object AuthenticationProviderFactory {
       method: AuthMethod,
       conf: KyuubiConf): PasswdAuthenticationProvider = method match {
     case AuthMethods.NONE => new AnonymousAuthenticationProviderImpl
-    case AuthMethods.LDAP => new LdapAuthenticationProviderImpl(conf)
+    case AuthMethods.LDAP => ldapProviders.computeIfAbsent("server", _ => buildLdapProvider(conf))
     case AuthMethods.JDBC => new JdbcAuthenticationProviderImpl(conf)
     case AuthMethods.CUSTOM =>
       val className = conf.get(KyuubiConf.AUTHENTICATION_CUSTOM_CLASS)
@@ -68,7 +78,7 @@ object AuthenticationProviderFactory {
       method: AuthMethod,
       conf: KyuubiConf): PasswdAuthenticationProvider = method match {
     case AuthMethods.NONE => new AnonymousAuthenticationProviderImpl
-    case AuthMethods.LDAP => new LdapAuthenticationProviderImpl(conf)
+    case AuthMethods.LDAP => ldapProviders.computeIfAbsent("http", _ => buildLdapProvider(conf))
     case AuthMethods.JDBC => new JdbcAuthenticationProviderImpl(conf)
     case AuthMethods.CUSTOM =>
       val className = conf.get(KyuubiConf.AUTHENTICATION_CUSTOM_BASIC_CLASS)
@@ -87,4 +97,164 @@ object AuthenticationProviderFactory {
       "kyuubi.authentication.custom.bearer.class must be set for http bearer authentication.")
     ClassUtils.createInstance(providerClass, classOf[TokenAuthenticationProvider], conf)
   }
+
+  /**
+   * Builds a [[PasswdAuthenticationProvider]] for LDAP.
+   *
+   * When pool is enabled, the stack is:
+   *   CachingLdapAuthenticationProvider
+   *     -> LdapAuthenticationProviderImpl(UnboundIdDirSearchFactory + connection pool)
+   *
+   * When pool is disabled (the default), uses the legacy JNDI path without caching. Enabling
+   * the pool opts into warm, health-checked, failover-aware connections; the legacy path is
+   * retained for environments where per-request connections are required (e.g. firewall
+   * rules that disallow long-lived connections).
+   */
+  private def buildLdapProvider(conf: KyuubiConf): PasswdAuthenticationProvider = {
+    if (conf.get(KyuubiConf.AUTHENTICATION_LDAP_POOL_ENABLED)) {
+      val poolOpt = UnboundIdConnectionPool.getOrCreate(conf)
+      val factory = new UnboundIdDirSearchFactory(conf, poolOpt)
+      val core = new LdapAuthenticationProviderImpl(conf, factory)
+      if (conf.get(KyuubiConf.AUTHENTICATION_LDAP_CACHE_ENABLED)) {
+        new CachingLdapAuthenticationProvider(core, conf)
+      } else {
+        core
+      }
+    } else {
+      // The auth cache is only wired into the UnboundID/pool path; warn loudly when an
+      // operator turns the pool off but leaves caching enabled, so they're not silently
+      // running without the cache they thought they had configured.
+      if (conf.get(KyuubiConf.AUTHENTICATION_LDAP_CACHE_ENABLED)) {
+        warn(
+          s"${KyuubiConf.AUTHENTICATION_LDAP_CACHE_ENABLED.key} has no effect when " +
+            s"${KyuubiConf.AUTHENTICATION_LDAP_POOL_ENABLED.key}=false: " +
+            "the legacy JNDI auth path does not use the LDAP auth cache.")
+      }
+      new LdapAuthenticationProviderImpl(conf, new LdapSearchFactory)
+    }
+  }
+
+  /** Called on server shutdown to release the LDAP connection pool. */
+  def close(): Unit = {
+    UnboundIdConnectionPool.close()
+    LdapAuthCache.resetCache()
+    ldapProviders.clear()
+    // Reset auth counters so a re-init after close() (e.g. between test cases) starts
+    // from a clean slate. In production close() runs once at JVM shutdown, so the reset
+    // is a no-op there.
+    ldapAuthSuccessCounter.set(0L)
+    ldapAuthFailureInvalidCredentialsCounter.set(0L)
+    ldapAuthFailureInvalidInputCounter.set(0L)
+    ldapAuthFailureInfrastructureCounter.set(0L)
+    ldapAuthCacheHitCounter.set(0L)
+    ldapAuthCacheMissCounter.set(0L)
+    // Drop any installed metrics recorder so subsequent server starts in the same JVM
+    // (notably tests) do not see events emitted to a stale destination.
+    ldapAuthMetricsRecorder = LdapAuthMetricsRecorder.NoOp
+  }
+
+  /**
+   * Live gauge accessors for the LDAP connection pool. Each method returns 0 when the pool
+   * has not yet been created (e.g. before the first authentication request, or when LDAP
+   * pooling is disabled). Designed to be wired into [[org.apache.kyuubi.metrics.MetricsSystem]]
+   * gauges from `kyuubi-server`, which depends on `kyuubi-metrics`; `kyuubi-common` does not.
+   */
+  def ldapPoolNumAvailableConnections: Long = ldapPoolStat(_.numAvailableConnections)
+  def ldapPoolNumFailedCheckouts: Long = ldapPoolStat(_.numFailedCheckouts)
+  def ldapPoolNumSuccessfulCheckouts: Long = ldapPoolStat(_.numSuccessfulCheckouts)
+  def ldapPoolNumConnectionsClosedDefunct: Long = ldapPoolStat(_.numConnectionsClosedDefunct)
+  def ldapPoolNumConnectionsClosedExpired: Long = ldapPoolStat(_.numConnectionsClosedExpired)
+
+  private def ldapPoolStat(read: UnboundIdConnectionPool => Long): Long =
+    UnboundIdConnectionPool.currentInstance.map(read).getOrElse(0L)
+
+  // ---------------------------------------------------------------------------------------
+  // LDAP authentication request counters.
+  //
+  // Held as AtomicLong here so they can be incremented from kyuubi-common code paths
+  // (CachingLdapAuthenticationProvider, LdapAuthenticationProviderImpl) without taking a
+  // dependency on kyuubi-metrics. KyuubiServer reads them via the Long getters below and
+  // registers them as MetricsSystem gauges, which DropwizardExports surfaces 1:1 in
+  // Prometheus. Counters are reset on close() so test isolation is automatic.
+  //
+  // Semantics:
+  //   ldapAuthSuccess              = ldapAuthCacheHit + (cache-miss attempts that succeeded)
+  //   sum(ldapAuthFailure_*)       = cache-miss attempts that failed
+  //   ldapAuthCacheMiss            = (success - cache-hit) + sum(failure_*)   [caching ON]
+  //   ldapAuthCacheHit / cache-miss = both 0 when caching is disabled
+  // ---------------------------------------------------------------------------------------
+
+  private val ldapAuthSuccessCounter = new AtomicLong(0L)
+  private val ldapAuthFailureInvalidCredentialsCounter = new AtomicLong(0L)
+  private val ldapAuthFailureInvalidInputCounter = new AtomicLong(0L)
+  private val ldapAuthFailureInfrastructureCounter = new AtomicLong(0L)
+  private val ldapAuthCacheHitCounter = new AtomicLong(0L)
+  private val ldapAuthCacheMissCounter = new AtomicLong(0L)
+
+  // Installable callback for surfacing auth events to a metrics system. See
+  // [[LdapAuthMetricsRecorder]] for why this is a hook rather than a direct call -- short
+  // version: cumulative counters must surface in Prometheus as Counter (not Gauge) so
+  // rate() / increase() handle process restarts correctly, and kyuubi-common does not
+  // depend on kyuubi-metrics. KyuubiServer installs a real implementation at startup.
+  @volatile private var ldapAuthMetricsRecorder: LdapAuthMetricsRecorder =
+    LdapAuthMetricsRecorder.NoOp
+
+  /**
+   * Installs the recorder that surfaces LDAP auth events to the metrics subsystem.
+   * Passing `null` resets to the no-op default. Idempotent across calls.
+   */
+  def setLdapAuthMetricsRecorder(recorder: LdapAuthMetricsRecorder): Unit = {
+    ldapAuthMetricsRecorder = if (recorder == null) LdapAuthMetricsRecorder.NoOp else recorder
+  }
+
+  /** Records a successful LDAP authentication (cache hit OR cache-miss success). */
+  private[authentication] def recordLdapAuthSuccess(): Unit = {
+    ldapAuthSuccessCounter.incrementAndGet()
+    ldapAuthMetricsRecorder.recordSuccess()
+  }
+
+  /**
+   * Records a failed LDAP authentication, classified by `reason`. Unknown reasons fall
+   * into the `infrastructure` bucket so a typo at a call site does not silently drop the
+   * event.
+   */
+  private[authentication] def recordLdapAuthFailure(reason: String): Unit = {
+    val normalised = reason match {
+      case LdapAuthFailureClassifier.INVALID_CREDENTIALS =>
+        ldapAuthFailureInvalidCredentialsCounter.incrementAndGet()
+        LdapAuthFailureClassifier.INVALID_CREDENTIALS
+      case LdapAuthFailureClassifier.INVALID_INPUT =>
+        ldapAuthFailureInvalidInputCounter.incrementAndGet()
+        LdapAuthFailureClassifier.INVALID_INPUT
+      case _ =>
+        ldapAuthFailureInfrastructureCounter.incrementAndGet()
+        LdapAuthFailureClassifier.INFRASTRUCTURE
+    }
+    ldapAuthMetricsRecorder.recordFailure(normalised)
+  }
+
+  /** Records a cache hit at the LDAP authentication caching layer. */
+  private[authentication] def recordLdapAuthCacheHit(): Unit = {
+    ldapAuthCacheHitCounter.incrementAndGet()
+    ldapAuthMetricsRecorder.recordCacheHit()
+  }
+
+  /** Records a cache miss at the LDAP authentication caching layer. */
+  private[authentication] def recordLdapAuthCacheMiss(): Unit = {
+    ldapAuthCacheMissCounter.incrementAndGet()
+    ldapAuthMetricsRecorder.recordCacheMiss()
+  }
+
+  // In-process counter accessors retained for test inspection. Tightened to
+  // `private[authentication]` because production metric exposure now flows through the
+  // recorder above -- no caller outside this package needs these.
+  private[authentication] def ldapAuthSuccess: Long = ldapAuthSuccessCounter.get()
+  private[authentication] def ldapAuthFailureInvalidCredentials: Long =
+    ldapAuthFailureInvalidCredentialsCounter.get()
+  private[authentication] def ldapAuthFailureInvalidInput: Long =
+    ldapAuthFailureInvalidInputCounter.get()
+  private[authentication] def ldapAuthFailureInfrastructure: Long =
+    ldapAuthFailureInfrastructureCounter.get()
+  private[authentication] def ldapAuthCacheHit: Long = ldapAuthCacheHitCounter.get()
+  private[authentication] def ldapAuthCacheMiss: Long = ldapAuthCacheMissCounter.get()
 }