Add metrics for batch pending max elapse time

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

@@ -2023,6 +2023,16 @@ object KyuubiConf {
       .intConf
       .createWithDefault(65536)
 
+  val METADATA_SEARCH_WINDOW: OptionalConfigEntry[Long] =
+    buildConf("kyuubi.metadata.search.window")
+      .doc("The time window to restrict user queries to metadata within a specific period. " +
+        "For example, if the window is set to P7D, only metadata from the past 7 days can be " +
+        "queried. If not set, it allows searching all metadata information in the metadata store.")
+      .version("1.10.1")
+      .timeConf
+      .checkValue(_ > 0, "must be positive number")
+      .createOptional
+
   val ENGINE_EXEC_WAIT_QUEUE_SIZE: ConfigEntry[Int] =
     buildConf("kyuubi.backend.engine.exec.pool.wait.queue.size")
       .doc("Size of the wait queue for the operation execution thread pool in SQL engine" +

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

@@ -64,6 +64,7 @@ object MetricsConstants {
   final val OPERATION_TOTAL: String = OPERATION + "total"
   final val OPERATION_STATE: String = OPERATION + "state"
   final val OPERATION_EXEC_TIME: String = OPERATION + "exec_time"
+  final val OPERATION_BATCH_PENDING_MAX_ELAPSE: String = OPERATION + "batch_pending_max_elapse"
 
   final private val BACKEND_SERVICE = KYUUBI + "backend_service."
   final val BS_FETCH_LOG_ROWS_RATE = BACKEND_SERVICE + "fetch_log_rows_rate"

kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala

@@ -424,6 +424,14 @@ class BatchJobSubmission(
       Utils.deleteDirectoryRecursively(session.resourceUploadFolderPath.toFile)
     }
   }
+
+  def getPendingElapsedTime: Long = {
+    if (state == OperationState.PENDING) {
+      System.currentTimeMillis() - createTime
+    } else {
+      0L
+    }
+  }
 }
 
 object BatchJobSubmission {

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

@@ -31,12 +31,14 @@ import org.eclipse.jetty.servlet.{ErrorPageErrorHandler, FilterHolder}
 import org.apache.kyuubi.{KyuubiException, Utils}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.OPERATION_BATCH_PENDING_MAX_ELAPSE
+import org.apache.kyuubi.metrics.MetricsSystem
 import org.apache.kyuubi.server.api.v1.ApiRootResource
 import org.apache.kyuubi.server.http.authentication.{AuthenticationFilter, KyuubiHttpAuthenticationFactory}
 import org.apache.kyuubi.server.ui.{JettyServer, JettyUtils}
 import org.apache.kyuubi.service.{AbstractFrontendService, Serverable, Service, ServiceUtils}
 import org.apache.kyuubi.service.authentication.{AuthTypes, AuthUtils}
-import org.apache.kyuubi.session.{KyuubiSessionManager, SessionHandle}
+import org.apache.kyuubi.session.{KyuubiBatchSession, KyuubiSessionManager, SessionHandle}
 import org.apache.kyuubi.util.{JavaUtils, ThreadUtils}
 import org.apache.kyuubi.util.ThreadUtils.scheduleTolerableRunnableWithFixedDelay
 
@@ -202,6 +204,14 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
     }
   }
 
+  private def getBatchPendingMaxElapse(): Long = {
+    val batchPendingElapseTimes = sessionManager.allSessions().map {
+      case session: KyuubiBatchSession => session.batchJobSubmissionOp.getPendingElapsedTime
+      case _ => 0L
+    }
+    if (batchPendingElapseTimes.isEmpty) 0L else batchPendingElapseTimes.max
+  }
+
   def waitForServerStarted(): Unit = {
     // block until the HTTP server is started, otherwise, we may get
     // the wrong HTTP server port -1
@@ -220,6 +230,9 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
         isStarted.set(true)
         startBatchChecker()
         recoverBatchSessions()
+        MetricsSystem.tracing { ms =>
+          ms.registerGauge(OPERATION_BATCH_PENDING_MAX_ELAPSE, getBatchPendingMaxElapse, 0)
+        }
       } catch {
         case e: Exception => throw new KyuubiException(s"Cannot start $getName", e)
       }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -67,6 +67,7 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
     fe.getConf.get(ENGINE_SECURITY_ENABLED)
   private lazy val resourceFileMaxSize = fe.getConf.get(BATCH_RESOURCE_FILE_MAX_SIZE)
   private lazy val extraResourceFileMaxSize = fe.getConf.get(BATCH_EXTRA_RESOURCE_FILE_MAX_SIZE)
+  private lazy val metadataSearchWindow = fe.getConf.get(METADATA_SEARCH_WINDOW)
 
   private def batchV2Enabled(reqConf: Map[String, String]): Boolean = {
     fe.getConf.get(BATCH_SUBMITTER_ENABLED) &&
@@ -420,13 +421,15 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
         s"The valid batch state can be one of the following: ${VALID_BATCH_STATES.mkString(",")}")
     }
 
+    val createTimeFilter =
+      math.max(createTime, metadataSearchWindow.map(System.currentTimeMillis() - _).getOrElse(0L))
     val filter = MetadataFilter(
       sessionType = SessionType.BATCH,
       engineType = batchType,
       username = batchUser,
       state = batchState,
       requestName = batchName,
-      createTime = createTime,
+      createTime = createTimeFilter,
       endTime = endTime)
     val batches = sessionManager.getBatchesFromMetadataStore(filter, from, size, desc)
     new GetBatchesResponse(from, batches.size, batches.asJava)

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala

@@ -139,8 +139,13 @@ class MetadataManager extends AbstractService("MetadataManager") {
       from: Int,
       size: Int,
       desc: Boolean = false): Seq[Batch] = {
-    withMetadataRequestMetrics(_metadataStore.getMetadataList(filter, from, size, desc)).map(
-      buildBatch)
+    withMetadataRequestMetrics(_metadataStore.getMetadataList(
+      filter,
+      from,
+      size,
+      // if create_file field is set, order by create_time, which is faster, otherwise by key_id
+      orderBy = if (filter.createTime > 0) Some("create_time") else Some("key_id"),
+      direction = if (desc) "DESC" else "ASC")).map(buildBatch)
   }
 
   def countBatch(

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataStore.scala

@@ -58,13 +58,16 @@ trait MetadataStore extends Closeable {
    * @param from the metadata offset.
    * @param size the size to get.
    * @param desc the order of metadata list.
+   * @param orderBy the order by column, default is the auto increment primary key, `key_id`.
+   * @param direction the order direction, default is `ASC`.
    * @return selected metadata list.
    */
   def getMetadataList(
       filter: MetadataFilter,
       from: Int,
       size: Int,
-      desc: Boolean = false): Seq[Metadata]
+      orderBy: Option[String] = Some("key_id"),
+      direction: String = "ASC"): Seq[Metadata]
 
   /**
    * Count the metadata list with filter conditions.

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStore.scala

@@ -257,15 +257,15 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
       filter: MetadataFilter,
       from: Int,
       size: Int,
-      desc: Boolean = false): Seq[Metadata] = {
+      orderBy: Option[String] = Some("key_id"),
+      direction: String = "ASC"): Seq[Metadata] = {
     val queryBuilder = new StringBuilder
     val params = ListBuffer[Any]()
     queryBuilder.append("SELECT ")
     queryBuilder.append(METADATA_COLUMNS)
     queryBuilder.append(s" FROM $METADATA_TABLE")
     queryBuilder.append(s" ${assembleWhereClause(filter, params)}")
-    queryBuilder.append(" ORDER BY key_id ")
-    queryBuilder.append(if (desc) "DESC " else "ASC ")
+    orderBy.foreach(o => queryBuilder.append(s" ORDER BY $o $direction "))
     queryBuilder.append(dialect.limitClause(size, from))
     val query = queryBuilder.toString
     JdbcUtils.withConnection { connection =>

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala

@@ -486,6 +486,7 @@ abstract class BatchesResourceSuiteBase extends KyuubiFunSuite
       .queryParam("from", "0")
       .queryParam("size", "1")
       .queryParam("desc", "true")
+      .queryParam("createTime", "1")
       .request(MediaType.APPLICATION_JSON_TYPE)
       .header(AUTHORIZATION_HEADER, basicAuthorizationHeader("anonymous"))
       .get()