[KYUUBI #6866] Add metrics for SSL keystore expiration time

### Why are the changes needed?

Add metrics for SSL keystore expiration, then we can add alert if the keystore will expire in 1 month.

### How was this patch tested?

Integration testing.
<img width="1721" alt="image" src="https://github.com/user-attachments/assets/f4ef6af6-923b-403c-a80d-06dbb80dbe1c" />

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #6866 from turboFei/keystore_expire.

Closes #6866

77c6db0 [Wang, Fei] Add metrics for SSL keystore expiration time #6866

Authored-by: Wang, Fei <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

Author: turboFei

kyuubi-common/src/main/scala/org/apache/kyuubi/service/TBinaryFrontendService.scala

@@ -53,6 +53,10 @@ abstract class TBinaryFrontendService(name: String)
   private var _actualPort: Int = _
   override protected lazy val actualPort: Int = _actualPort
 
+  protected var keyStorePath: Option[String] = None
+  protected var keyStorePassword: Option[String] = None
+  protected var keyStoreType: Option[String] = None
+
   // Removed OOM hook since Kyuubi #1800 to respect the hive server2 #2383
 
   override def initialize(conf: KyuubiConf): Unit = synchronized {
@@ -73,9 +77,9 @@ abstract class TBinaryFrontendService(name: String)
       val tServerSocket =
         // only enable ssl for server side
         if (isServer() && conf.get(FRONTEND_THRIFT_BINARY_SSL_ENABLED)) {
-          val keyStorePath = conf.get(FRONTEND_SSL_KEYSTORE_PATH)
-          val keyStorePassword = conf.get(FRONTEND_SSL_KEYSTORE_PASSWORD)
-          val keyStoreType = conf.get(FRONTEND_SSL_KEYSTORE_TYPE)
+          keyStorePath = conf.get(FRONTEND_SSL_KEYSTORE_PATH)
+          keyStorePassword = conf.get(FRONTEND_SSL_KEYSTORE_PASSWORD)
+          keyStoreType = conf.get(FRONTEND_SSL_KEYSTORE_TYPE)
           val keyStoreAlgorithm = conf.get(FRONTEND_SSL_KEYSTORE_ALGORITHM)
           val disallowedSslProtocols = conf.get(FRONTEND_THRIFT_BINARY_SSL_DISALLOWED_PROTOCOLS)
           val includeCipherSuites = conf.get(FRONTEND_THRIFT_BINARY_SSL_INCLUDE_CIPHER_SUITES)

kyuubi-metrics/src/main/scala/org/apache/kyuubi/metrics/MetricsConstants.scala

@@ -37,6 +37,8 @@ object MetricsConstants {
   final private val THRIFT_BINARY_CONN = KYUUBI + "thrift.binary.connection."
   final private val REST_CONN = KYUUBI + "rest.connection."
 
+  final val THRIFT_SSL_CERT_EXPIRATION = KYUUBI + "thrift.ssl.cert.expiration"
+
   final val CONN_OPEN: String = CONN + "opened"
   final val CONN_FAIL: String = CONN + "failed"
   final val CONN_TOTAL: String = CONN + "total"

kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTBinaryFrontendService.scala

@@ -35,6 +35,7 @@ import org.apache.kyuubi.session.KyuubiSessionImpl
 import org.apache.kyuubi.shaded.hive.service.rpc.thrift._
 import org.apache.kyuubi.shaded.thrift.protocol.TProtocol
 import org.apache.kyuubi.shaded.thrift.server.ServerContext
+import org.apache.kyuubi.util.SSLUtils
 
 final class KyuubiTBinaryFrontendService(
     override val serverable: Serverable)
@@ -122,4 +123,9 @@ final class KyuubiTBinaryFrontendService(
     resp.setStatus(notSupportTokenErrorStatus)
     resp
   }
+
+  override def start(): Unit = {
+    super.start()
+    SSLUtils.tracingThriftSSLCertExpiration(keyStorePath, keyStorePassword, keyStoreType)
+  }
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala

@@ -50,7 +50,7 @@ import org.apache.kyuubi.service.TFrontendService.{CURRENT_SERVER_CONTEXT, OK_ST
 import org.apache.kyuubi.session.KyuubiSessionImpl
 import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq, TOpenSessionResp}
 import org.apache.kyuubi.shaded.thrift.protocol.TBinaryProtocol
-import org.apache.kyuubi.util.NamedThreadFactory
+import org.apache.kyuubi.util.{NamedThreadFactory, SSLUtils}
 
 /**
  * Apache Thrift based hive service rpc
@@ -67,6 +67,10 @@ final class KyuubiTHttpFrontendService(
   override protected lazy val actualPort: Int = portNum
   override protected lazy val serverSocket: ServerSocket = null
 
+  private var keyStorePath: Option[String] = None
+  private var keyStorePassword: Option[String] = None
+  private var keyStoreType: Option[String] = None
+
   private var server: Option[Server] = None
   private val APPLICATION_THRIFT = "application/x-thrift"
 
@@ -122,15 +126,15 @@ final class KyuubiTHttpFrontendService(
       // Change connector if SSL is used
       val connector =
         if (useSsl) {
-          val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+          keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
 
           if (keyStorePath.isEmpty) {
             throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
               " Not configured for SSL connection, please set the key with: " +
               FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
           }
 
-          val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+          keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
           if (keyStorePassword.isEmpty) {
             throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
               " Not configured for SSL connection. please set the key with: " +
@@ -140,7 +144,7 @@ final class KyuubiTHttpFrontendService(
           val sslContextFactory = new SslContextFactory.Server
           val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST)
           val excludeCipherSuites = conf.get(FRONTEND_THRIFT_HTTP_SSL_EXCLUDE_CIPHER_SUITES)
-          val keyStoreType = conf.get(FRONTEND_SSL_KEYSTORE_TYPE)
+          keyStoreType = conf.get(FRONTEND_SSL_KEYSTORE_TYPE)
           val keyStoreAlgorithm = conf.get(FRONTEND_SSL_KEYSTORE_ALGORITHM)
           info("Thrift HTTP Server SSL: adding excluded protocols: " +
             String.join(",", excludedProtocols: _*))
@@ -359,4 +363,9 @@ final class KyuubiTHttpFrontendService(
 
     ret
   }
+
+  override def start(): Unit = {
+    super.start()
+    SSLUtils.tracingThriftSSLCertExpiration(keyStorePath, keyStorePassword, keyStoreType)
+  }
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/util/SSLUtils.scala

@@ -0,0 +1,70 @@
+/*
+ * 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.kyuubi.util
+import java.io.FileInputStream
+import java.security.KeyStore
+import java.security.cert.X509Certificate
+
+import scala.collection.JavaConverters._
+
+import org.apache.kyuubi.{Logging, Utils}
+import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem}
+
+object SSLUtils extends Logging {
+
+  /**
+   * Get the keystore certificate latest expiration time.
+   */
+  private def getKeyStoreExpirationTime(
+      keyStorePath: String,
+      keyStorePassword: String,
+      keyStoreType: Option[String]): Option[Long] = {
+    try {
+      val keyStore = KeyStore.getInstance(keyStoreType.getOrElse(KeyStore.getDefaultType))
+      keyStore.load(new FileInputStream(keyStorePath), keyStorePassword.toCharArray)
+      keyStore.aliases().asScala.toSeq.map { alias =>
+        keyStore.getCertificate(alias).asInstanceOf[X509Certificate].getNotAfter.getTime
+      }.sorted.headOption
+    } catch {
+      case e: Throwable =>
+        error("Error getting keystore expiration time.", e)
+        None
+    }
+  }
+
+  def tracingThriftSSLCertExpiration(
+      keyStorePath: Option[String],
+      keyStorePassword: Option[String],
+      keyStoreType: Option[String]): Unit = {
+    if (keyStorePath.isDefined && keyStorePassword.isDefined) {
+      SSLUtils.getKeyStoreExpirationTime(
+        keyStorePath.get,
+        keyStorePassword.get,
+        keyStoreType).foreach { expiration =>
+        info(s"Thrift SSL Serve KeyStore ${keyStorePath.get} will expire at:" +
+          s" ${Utils.getDateFromTimestamp(expiration)}")
+        MetricsSystem.tracing { ms =>
+          ms.registerGauge(
+            MetricsConstants.THRIFT_SSL_CERT_EXPIRATION,
+            expiration - System.currentTimeMillis(),
+            0L)
+        }
+      }
+    }
+  }
+}