diff --git a/linkis-commons/linkis-hadoop-common/src/main/java/org/apache/linkis/hadoop/common/utils/KerberosTgtUtils.java b/linkis-commons/linkis-hadoop-common/src/main/java/org/apache/linkis/hadoop/common/utils/KerberosTgtUtils.java new file mode 100644 index 00000000000..5c42c13bab0 --- /dev/null +++ b/linkis-commons/linkis-hadoop-common/src/main/java/org/apache/linkis/hadoop/common/utils/KerberosTgtUtils.java @@ -0,0 +1,99 @@ +/* + * 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.linkis.hadoop.common.utils; + +import org.apache.hadoop.security.UserGroupInformation; + +import javax.security.auth.Subject; +import javax.security.auth.kerberos.KerberosTicket; + +import java.lang.reflect.Method; +import java.util.Set; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class KerberosTgtUtils { + + private static final Logger logger = LoggerFactory.getLogger(KerberosTgtUtils.class); + + private static volatile Method getSubjectMethod = null; + private static volatile boolean getSubjectMethodResolved = false; + + /** + * Get the Subject from UserGroupInformation via reflection, since getSubject() is protected in + * Hadoop 2.7.x. + */ + private static Subject getSubjectFromUgi(UserGroupInformation ugi) { + if (!getSubjectMethodResolved) { + synchronized (KerberosTgtUtils.class) { + if (!getSubjectMethodResolved) { + try { + Method method = UserGroupInformation.class.getDeclaredMethod("getSubject"); + method.setAccessible(true); + getSubjectMethod = method; + } catch (NoSuchMethodException e) { + logger.warn("Failed to find getSubject method on UserGroupInformation", e); + getSubjectMethod = null; + } + getSubjectMethodResolved = true; + } + } + } + if (getSubjectMethod == null) { + return null; + } + try { + return (Subject) getSubjectMethod.invoke(ugi); + } catch (Exception e) { + logger.warn("Failed to invoke getSubject on UserGroupInformation", e); + return null; + } + } + + /** + * Check if the Kerberos TGT in the given UGI is still valid (not expired). For proxy UGI, the + * real user's TGT is checked. + * + * @param ugi the UserGroupInformation to check + * @return true if UGI is null, security not enabled, or TGT is still valid; false if TGT has + * expired + */ + public static boolean isTgtValid(UserGroupInformation ugi) { + if (ugi == null || !UserGroupInformation.isSecurityEnabled()) { + return true; + } + // For proxy UGI, check the real user's tickets + UserGroupInformation checkUgi = ugi.getRealUser() != null ? ugi.getRealUser() : ugi; + Subject subject = getSubjectFromUgi(checkUgi); + if (subject == null) { + return true; + } + Set tickets = subject.getPrivateCredentials(KerberosTicket.class); + if (tickets == null || tickets.isEmpty()) { + return true; + } + long now = System.currentTimeMillis(); + for (KerberosTicket ticket : tickets) { + if (ticket.getEndTime().getTime() <= now) { + return false; + } + } + return true; + } +} diff --git a/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/conf/HadoopConf.scala b/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/conf/HadoopConf.scala index 864b8f65057..56ca8a89ff0 100644 --- a/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/conf/HadoopConf.scala +++ b/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/conf/HadoopConf.scala @@ -81,4 +81,13 @@ object HadoopConf { */ val KEYTAB_TEMP_DIR = CommonVars("linkis.keytab.temp.dir", "/tmp/keytab") + /** + * Enable proactive Kerberos TGT refresh when retrieving cached HDFS FileSystem. When enabled, + * before returning a cached FileSystem, Hadoop's `reloginFromKeytab()` is called on the UGI to + * refresh the TGT if near expiry. If refresh fails (TGT expired and keytab unavailable), the + * cached entry is removed and a new FileSystem with fresh TGT is created. + */ + val HDFS_TGT_PROACTIVE_CHECK_ENABLE = + CommonVars("linkis.hadoop.hdfs.tgt.proactive.check.enable", false) + } diff --git a/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/entity/HDFSFileSystemContainer.scala b/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/entity/HDFSFileSystemContainer.scala index f87f89393e9..10bca1c4cd2 100644 --- a/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/entity/HDFSFileSystemContainer.scala +++ b/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/entity/HDFSFileSystemContainer.scala @@ -18,10 +18,17 @@ package org.apache.linkis.hadoop.common.entity import org.apache.linkis.hadoop.common.conf.HadoopConf +import org.apache.linkis.hadoop.common.utils.KerberosTgtUtils import org.apache.hadoop.fs.FileSystem +import org.apache.hadoop.security.UserGroupInformation -class HDFSFileSystemContainer(fs: FileSystem, user: String, label: String) { +class HDFSFileSystemContainer( + fs: FileSystem, + user: String, + label: String, + ugi: UserGroupInformation = null +) { private var lastAccessTime: Long = System.currentTimeMillis() @@ -33,6 +40,8 @@ class HDFSFileSystemContainer(fs: FileSystem, user: String, label: String) { def getLabel: String = this.label + def getUgi: UserGroupInformation = this.ugi + def getLastAccessTime: Long = this.lastAccessTime def updateLastAccessTime: Unit = { @@ -51,4 +60,14 @@ class HDFSFileSystemContainer(fs: FileSystem, user: String, label: String) { idleTime > HadoopConf.HDFS_ENABLE_CACHE_MAX_TIME || ((idleTime > HadoopConf.HDFS_ENABLE_CACHE_IDLE_TIME) && count <= 0) } + /** + * Check if the Kerberos TGT in this container's UGI is still valid. Delegates to + * KerberosTgtUtils.isTgtValid() which checks the KerberosTicket end time against current time. + * + * @return + * true if TGT is valid (or UGI is null / non-Kerberos); false if TGT has expired → caller + * should remove from cache and recreate FileSystem + */ + def isTgtValid(): Boolean = KerberosTgtUtils.isTgtValid(ugi) + } diff --git a/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/utils/HDFSUtils.scala b/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/utils/HDFSUtils.scala index 45a03f0dc93..0ae9cf7bfce 100644 --- a/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/utils/HDFSUtils.scala +++ b/linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/utils/HDFSUtils.scala @@ -233,9 +233,23 @@ object HDFSUtils extends Logging { locker.intern().synchronized { if (fileSystemCache.containsKey(cacheKey)) { val hdfsFileSystemContainer = fileSystemCache.get(cacheKey) - hdfsFileSystemContainer.addAccessCount() - hdfsFileSystemContainer.updateLastAccessTime - hdfsFileSystemContainer.getFileSystem + // Proactive TGT expiry check: if expired, remove from cache and recreate + if ( + HadoopConf.HDFS_TGT_PROACTIVE_CHECK_ENABLE.getValue && !hdfsFileSystemContainer + .isTgtValid() + ) { + logger.info( + s"Cached HDFS FileSystem TGT expired - user: $userName, label: $cacheLabel, removing from cache and recreating" + ) + fileSystemCache.remove(cacheKey) + IOUtils.closeQuietly(hdfsFileSystemContainer.getFileSystem) + // Fall through to create new FileSystem with fresh TGT + getHDFSUserFileSystem(userName, label, getConfigurationByLabel(userName, label)) + } else { + hdfsFileSystemContainer.addAccessCount() + hdfsFileSystemContainer.updateLastAccessTime + hdfsFileSystemContainer.getFileSystem + } } else { getHDFSUserFileSystem(userName, label, getConfigurationByLabel(userName, label)) } @@ -259,25 +273,33 @@ object HDFSUtils extends Logging { val cacheLabel = if (label == null) DEFAULT_CACHE_LABEL else label val cacheKey = userName + JOINT + cacheLabel locker.intern().synchronized { - val hdfsFileSystemContainer = if (fileSystemCache.containsKey(cacheKey)) { - fileSystemCache.get(cacheKey) - } else { - // we use cacheLabel to create HDFSFileSystemContainer, and in the rest part of HDFSUtils, we consistently - // use the same cacheLabel to operate HDFSFileSystemContainer, like close or remove. - // At the same time, we don't want to change the behavior of createFileSystem which is out of HDFSUtils, - // so we continue to use the original label to createFileSystem. - val newHDFSFileSystemContainer = - new HDFSFileSystemContainer( - createFileSystem(userName, label, conf), - userName, - cacheLabel + if (fileSystemCache.containsKey(cacheKey)) { + val hdfsFileSystemContainer = fileSystemCache.get(cacheKey) + // Proactive TGT expiry check: if expired, remove from cache and recreate + if ( + HadoopConf.HDFS_TGT_PROACTIVE_CHECK_ENABLE.getValue && !hdfsFileSystemContainer + .isTgtValid() + ) { + logger.info( + s"Cached HDFS FileSystem TGT expired - user: $userName, label: $cacheLabel, removing from cache and recreating" ) - fileSystemCache.put(cacheKey, newHDFSFileSystemContainer) - newHDFSFileSystemContainer + fileSystemCache.remove(cacheKey) + IOUtils.closeQuietly(hdfsFileSystemContainer.getFileSystem) + // Fall through to create new FileSystem with fresh TGT + } else { + hdfsFileSystemContainer.addAccessCount() + hdfsFileSystemContainer.updateLastAccessTime + return hdfsFileSystemContainer.getFileSystem + } } - hdfsFileSystemContainer.addAccessCount() - hdfsFileSystemContainer.updateLastAccessTime - hdfsFileSystemContainer.getFileSystem + // Cache miss or TGT expired → create new FileSystem with UGI + val (newFs, newUgi) = createFileSystemWithUgi(userName, label, conf) + val newHDFSFileSystemContainer = + new HDFSFileSystemContainer(newFs, userName, cacheLabel, newUgi) + fileSystemCache.put(cacheKey, newHDFSFileSystemContainer) + newHDFSFileSystemContainer.addAccessCount() + newHDFSFileSystemContainer.updateLastAccessTime + newHDFSFileSystemContainer.getFileSystem } } else { createFileSystem(userName, label, conf) @@ -287,6 +309,45 @@ object HDFSUtils extends Logging { def createFileSystem(userName: String, conf: org.apache.hadoop.conf.Configuration): FileSystem = createFileSystem(userName, null, conf) + /** + * Create a FileSystem and also return the UserGroupInformation used to create it. The UGI is + * needed for proactive TGT validity checking in the cache layer. + */ + private def createFileSystemWithUgi( + userName: String, + label: String, + conf: org.apache.hadoop.conf.Configuration + ): (FileSystem, UserGroupInformation) = { + val createCount = count.getAndIncrement() + val startTime = System.currentTimeMillis() + val labelInfo = if (label == null) "default" else label + logger.info( + s"Creating Hadoop FileSystem - user: $userName, label: $labelInfo, createCount: $createCount" + ) + try { + val ugi = getUserGroupInformation(userName, label) + val fs = ugi + .doAs(new PrivilegedExceptionAction[FileSystem] { + def run: FileSystem = FileSystem.newInstance(conf) + }) + val duration = System.currentTimeMillis() - startTime + logger.info( + s"Hadoop FileSystem created successfully - user: $userName, label: $labelInfo, duration: ${ByteTimeUtils + .msDurationToString(duration)}, createCount: $createCount" + ) + (fs, ugi) + } catch { + case e: Exception => + val duration = System.currentTimeMillis() - startTime + logger.error( + s"Failed to create Hadoop FileSystem - user: $userName, label: $labelInfo, duration: ${ByteTimeUtils + .msDurationToString(duration)}, createCount: $createCount", + e + ) + throw e + } + } + def createFileSystem( userName: String, label: String, diff --git a/linkis-commons/linkis-hadoop-common/src/test/scala/org/apache/linkis/hadoop/common/conf/HadoopConfTest.scala b/linkis-commons/linkis-hadoop-common/src/test/scala/org/apache/linkis/hadoop/common/conf/HadoopConfTest.scala index 7c2c7b38355..9c3c008adbc 100644 --- a/linkis-commons/linkis-hadoop-common/src/test/scala/org/apache/linkis/hadoop/common/conf/HadoopConfTest.scala +++ b/linkis-commons/linkis-hadoop-common/src/test/scala/org/apache/linkis/hadoop/common/conf/HadoopConfTest.scala @@ -38,6 +38,7 @@ class HadoopConfTest { ) Assertions.assertFalse(HadoopConf.HDFS_ENABLE_CACHE) Assertions.assertTrue(180000 == HadoopConf.HDFS_ENABLE_CACHE_IDLE_TIME) + Assertions.assertFalse(HadoopConf.HDFS_TGT_PROACTIVE_CHECK_ENABLE.getValue) }