diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java index accc6c4b4dff..96f1db1db2b6 100644 --- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java +++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java @@ -1980,6 +1980,17 @@ public enum ConfVars { "hive.metastore.iceberg.catalog.metrics.reporters", "org.apache.iceberg.rest.metrics.LoggingMetricsReporter", "A comma separated list of custom Iceberg Metrics Reporting plugins." ), + CATALOG_SERVLET_UGI_CACHE_SIZE("metastore.catalog.servlet.ugi.cache.size", + "hive.metastore.catalog.servlet.ugi.cache.size", 1000L, + "Maximum number of proxy UserGroupInformation instances to keep in the catalog servlet UGI cache. " + + "Entries displaced by this limit trigger FileSystem resource cleanup for the evicted UGI." + ), + CATALOG_SERVLET_UGI_CACHE_EXPIRY("metastore.catalog.servlet.ugi.cache.expiry", + "hive.metastore.catalog.servlet.ugi.cache.expiry", 3600, TimeUnit.SECONDS, + "Idle-expiry time for cached proxy UserGroupInformation instances in the catalog servlet. " + + "After this period of inactivity, the entry is evicted and FileSystem.closeAllForUGI is called " + + "to release associated IPC and RPC resources. Set to 0 to disable expiry-based eviction." + ), HTTPSERVER_THREADPOOL_MIN("hive.metastore.httpserver.threadpool.min", "hive.metastore.httpserver.threadpool.min", 8, "HMS embedded HTTP server minimum number of threads." diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java index 944c490e9787..f75dcdfade58 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java @@ -21,12 +21,22 @@ import static javax.ws.rs.core.HttpHeaders.WWW_AUTHENTICATE; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.RemovalListener; +import com.github.benmanes.caffeine.cache.Ticker; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import java.security.KeyStore; +import java.time.Duration; import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.hive.metastore.auth.HttpAuthenticationException; import org.apache.hadoop.hive.metastore.auth.jwt.SimpleJWTAuthenticator; import org.apache.hadoop.hive.metastore.auth.oauth2.OAuth2Authenticator; @@ -106,6 +116,13 @@ public static AuthType fromString(String type) { private final Function> scopeProvider; private SimpleJWTAuthenticator jwtAuthenticator = null; private OAuth2Authenticator oAuth2Authenticator = null; + private final Cache proxyUserCache; + + /** + * Cache key for a proxy {@link UserGroupInformation}. A proxy UGI is bound to both the effective user it + * impersonates and the server login user acting as its real user, so both participate in identity. + */ + record UgiKey(String effectiveUser, String loginUser) {} public ServletSecurity(AuthType authType, Configuration conf) { this(authType, conf, null); @@ -113,10 +130,99 @@ public ServletSecurity(AuthType authType, Configuration conf) { public ServletSecurity(AuthType authType, Configuration conf, Function> scopeProvider) { + this(authType, conf, scopeProvider, ForkJoinPool.commonPool()); + } + + @VisibleForTesting + ServletSecurity(AuthType authType, Configuration conf, + Function> scopeProvider, Executor cacheCleanupExecutor) { this.conf = conf; this.isSecurityEnabled = UserGroupInformation.isSecurityEnabled(); this.authType = authType; this.scopeProvider = scopeProvider; + this.proxyUserCache = createCacheWithConfig( + MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.CATALOG_SERVLET_UGI_CACHE_EXPIRY, TimeUnit.MILLISECONDS), + MetastoreConf.getLongVar(conf, MetastoreConf.ConfVars.CATALOG_SERVLET_UGI_CACHE_SIZE), + cacheCleanupExecutor); + } + + /** + * Creates a UGI cache with the specified expiration time and maximum size. + * + * @param expirationMs Time in milliseconds after which entries expire due to inactivity + * @param maxSize Maximum number of entries the cache can hold + * @param cacheCleanupExecutor executor on which removal-listener cleanup ({@link FileSystem#closeAllForUGI}) + * runs; production uses {@link ForkJoinPool#commonPool()} so cleanup stays off the + * request thread + * @return A configured Caffeine cache for UGI objects + */ + private Cache createCacheWithConfig(long expirationMs, long maxSize, + Executor cacheCleanupExecutor) { + // Note: eviction closes the UGI's FileSystems. If an entry is evicted while a request is still inside doAs, + // that in-flight operation could see a "FileSystem closed" error. We don't reference-count to prevent this; + // instead we rely on generous margins: expiry is idle-based (expireAfterAccess), and both the expiry window + // and maximumSize should be kept well above the longest operation / peak concurrent distinct users. + RemovalListener cleanupListener = + (key, ugi, cause) -> { + if (ugi != null) { + try { + FileSystem.closeAllForUGI(ugi); + if (LOG.isDebugEnabled()) { + LOG.debug("Cleaned up FileSystem handles for evicted UGI: {} (cause: {})", + ugi.getUserName(), cause); + } + } catch (IOException cleanupException) { + LOG.error("Failed to clean up FileSystem handles for evicted UGI: {} (cause: {})", + ugi, cause, cleanupException); + } + } + }; + + Caffeine builder = Caffeine.newBuilder() + .maximumSize(maxSize) + .executor(cacheCleanupExecutor) + .removalListener(cleanupListener); + + if (expirationMs > 0) { + builder.expireAfterAccess(Duration.ofMillis(expirationMs)) + .ticker(Ticker.systemTicker()); + } + + return builder.build(); + } + + /** + * Returns the (cached) proxy {@link UserGroupInformation} for the given user, creating it on a cache miss. + *

Caching the proxy UGI prevents Hadoop{@literal '}s {@code FileSystem.CACHE} from accumulating a distinct + * entry (and its RPC/IPC resources) for every request; evicted entries are cleaned up through + * {@link FileSystem#closeAllForUGI(UserGroupInformation)}.

+ * @param userName the effective user name extracted from the request + * @param loginUser the server login user that acts as the real user of the proxy + * @return the cached or freshly created proxy user + */ + @VisibleForTesting + UserGroupInformation getProxyUser(String userName, UserGroupInformation loginUser) { + return proxyUserCache.get(new UgiKey(userName, loginUser.getUserName()), key -> { + LOG.debug("Creating proxy user for: {}", key.effectiveUser()); + return UserGroupInformation.createProxyUser(key.effectiveUser(), loginUser); + }); + } + + /** + * Forces the proxy user cache to run any pending maintenance (eviction and cleanup). + */ + @VisibleForTesting + void cleanUpProxyUserCache() { + proxyUserCache.cleanUp(); + } + + /** + * @return the approximate number of entries currently held in the proxy user cache + */ + @VisibleForTesting + long proxyUserCacheSize() { + proxyUserCache.cleanUp(); + return proxyUserCache.estimatedSize(); } /** @@ -237,8 +343,7 @@ public void execute(HttpServletRequest request, HttpServletResponse response, Me // Temporary, and useless for now. Here only to allow this to work on an otherwise kerberized // server. if (isSecurityEnabled || authType == AuthType.JWT || authType == AuthType.OAUTH2) { - LOG.info("Creating proxy user for: {}", userFromHeader); - clientUgi = UserGroupInformation.createProxyUser(userFromHeader, UserGroupInformation.getLoginUser()); + clientUgi = getProxyUser(userFromHeader, UserGroupInformation.getLoginUser()); } else { // Unreachable in the case of NONE Preconditions.checkState(authType == AuthType.SIMPLE); diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java new file mode 100644 index 000000000000..e0146f8b6cf1 --- /dev/null +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hadoop.hive.metastore; + +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars; +import org.apache.hadoop.security.UserGroupInformation; +import org.junit.Assert; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +/** + * Unit tests for the proxy {@link UserGroupInformation} cache used by {@link ServletSecurity} to bound the number of + * proxy UGIs (and their associated FileSystem resources) that would otherwise leak through Hadoop's FileSystem cache. + */ +@Category(MetastoreUnitTest.class) +public class TestServletSecurity { + + private static Configuration confWithCache(long maxSize, long expirySeconds) { + Configuration conf = MetastoreConf.newMetastoreConf(); + MetastoreConf.setLongVar(conf, ConfVars.CATALOG_SERVLET_UGI_CACHE_SIZE, maxSize); + MetastoreConf.setTimeVar(conf, ConfVars.CATALOG_SERVLET_UGI_CACHE_EXPIRY, expirySeconds, TimeUnit.SECONDS); + return conf; + } + + @Test + public void testProxyUserIsCachedPerUser() throws Exception { + ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(100, 3600)); + UserGroupInformation loginUser = UserGroupInformation.getCurrentUser(); + + UserGroupInformation first = security.getProxyUser("alice", loginUser); + UserGroupInformation second = security.getProxyUser("alice", loginUser); + + Assert.assertSame("Repeated requests for the same user must reuse the cached proxy UGI", first, second); + Assert.assertEquals("alice", first.getShortUserName()); + } + + @Test + public void testDistinctUsersGetDistinctProxies() throws Exception { + ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(100, 3600)); + UserGroupInformation loginUser = UserGroupInformation.getCurrentUser(); + + UserGroupInformation alice = security.getProxyUser("alice", loginUser); + UserGroupInformation bob = security.getProxyUser("bob", loginUser); + + Assert.assertNotSame(alice, bob); + Assert.assertEquals(2, security.proxyUserCacheSize()); + } + + @Test + public void testEvictionClosesFileSystemForUgi() throws Exception { + // A cache bounded to a single entry: inserting a second user evicts the first. + // Runnable::run makes removal-listener cleanup run synchronously on this thread, where the static mock is active. + ServletSecurity security = + new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(1, 3600), null, Runnable::run); + UserGroupInformation loginUser = UserGroupInformation.getCurrentUser(); + + try (MockedStatic fsMock = Mockito.mockStatic(FileSystem.class)) { + UserGroupInformation alice = security.getProxyUser("alice", loginUser); + security.getProxyUser("bob", loginUser); + // Force any pending size-based eviction (and its synchronous cleanup) to run. + security.cleanUpProxyUserCache(); + + fsMock.verify(() -> FileSystem.closeAllForUGI(alice)); + } + } + + @Test + public void testExpiryDisabledWhenNonPositive() throws Exception { + // expiry == 0 disables time-based eviction; the size bound still applies and entries remain until displaced. + ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(100, 0)); + UserGroupInformation loginUser = UserGroupInformation.getCurrentUser(); + + UserGroupInformation first = security.getProxyUser("carol", loginUser); + UserGroupInformation second = security.getProxyUser("carol", loginUser); + + Assert.assertSame(first, second); + } +}