Event Tracking: For statement (#767)

Generate KStatement info. This object includes the following elements:
1. statement
2. statementId
3. appId
4. sessionId
5. executionId
6. physicalPlan
7. stateTime: contains each state and the time occurrence

Those data was packaged in KStatement and we store those object in mem.
You can get some summary data from this object.

externals/kyuubi-spark-monitor/pom.xml

@@ -31,6 +31,26 @@
     <packaging>jar</packaging>
     <name>Kyuubi Project Spark Monitor</name>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.kyuubi</groupId>
+            <artifactId>kyuubi-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_${scala.binary.version}</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_${scala.binary.version}</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
     <build>
         <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
         <testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory>

externals/kyuubi-spark-monitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/KyuubiStatementMonitor.scala

@@ -0,0 +1,66 @@
+/*
+ * 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.engine.spark.monitor
+
+import java.util.concurrent.ArrayBlockingQueue
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.monitor.entity.KyuubiStatementInfo
+
+// TODO: Thread Safe need to consider
+object KyuubiStatementMonitor extends Logging{
+
+  /**
+   * This blockingQueue store kyuubiStatementInfo.
+   *
+   * Notice:
+   *    1. When we remove items from this queue, we should ensure those statements have finished.
+   *       If not, we should put them into this queue again.
+   *    2. There have two kinds of threshold to trigger when to remove items from this queue:
+   *      a. time
+   *      b. this queue's current size
+   */
+  // TODO: Capacity should make configurable
+  private val kyuubiStatementQueue = new ArrayBlockingQueue[KyuubiStatementInfo](10)
+
+  /**
+   * This function is used for putting kyuubiStatementInfo into blockingQueue(statementQueue).
+   * Every time we put an item into this queue, we should judge this queue's current size at first.
+   * If the size is less than threshold, we need to remove items from this queue.
+   * @param kyuubiStatementInfo
+   */
+  // TODO: Lack size type threshold and time type threshold
+  def putStatementInfoIntoQueue(kyuubiStatementInfo: KyuubiStatementInfo): Unit = {
+    if (kyuubiStatementQueue.size() >= 7) {
+      removeAndDumpStatementInfoFromQueue()
+    }
+    val isSuccess = kyuubiStatementQueue.add(kyuubiStatementInfo)
+    info(s"Add kyuubiStatementInfo into queue is [$isSuccess], " +
+      s"statementId is [${kyuubiStatementInfo.statementId}]")
+  }
+
+  /**
+   * This function is used for removing kyuubiStatementInfo from blockingQueue(statementQueue)
+   * and dumpping them to a file by threshold.
+   */
+  // TODO: Need ensure those items have finished. If not, we should put them into this queue again.
+  private def removeAndDumpStatementInfoFromQueue(): Unit = {
+    // TODO: Just for test
+    kyuubiStatementQueue.clear()
+  }
+}

externals/kyuubi-spark-monitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/entity/KyuubiStatementInfo.scala

@@ -0,0 +1,48 @@
+/*
+ * 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.engine.spark.monitor.entity
+
+import scala.collection.mutable.Map
+
+import org.apache.spark.sql.execution.QueryExecution
+
+import org.apache.kyuubi.KyuubiSQLException
+import org.apache.kyuubi.cli.HandleIdentifier
+import org.apache.kyuubi.operation.OperationState.OperationState
+
+/**
+ * This object store the summary infomation about statement.
+ * You can use statementId to get all jobs' or stages' metric that this statement has.
+ * @param statementId
+ * @param statement
+ * @param appId
+ * @param sessionId
+ * @param queryExecution: contains physicalPlan, logicPlan and so on
+ * @param exception
+ * @param stateToTime: store this statement's every state and the time of occurrence
+ */
+case class KyuubiStatementInfo(
+    statementId: String,
+    statement: String,
+    appId: String,
+    sessionId: HandleIdentifier,
+    stateToTime: Map[OperationState, Long]) {
+
+  var queryExecution: QueryExecution = null
+  var exception: KyuubiSQLException = null
+}

externals/kyuubi-spark-monitor/src/test/resources/log4j.properties

@@ -0,0 +1,40 @@
+#
+# 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.
+#
+
+# Set everything to be logged to the file target/unit-tests.log
+log4j.rootLogger=DEBUG, CA, FA
+
+#Console Appender
+log4j.appender.CA=org.apache.log4j.ConsoleAppender
+log4j.appender.CA.layout=org.apache.log4j.PatternLayout
+log4j.appender.CA.layout.ConversionPattern=%d{HH:mm:ss.SSS} %p %c: %m%n
+log4j.appender.CA.Threshold = FATAL
+
+#File Appender
+log4j.appender.FA=org.apache.log4j.FileAppender
+log4j.appender.FA.append=false
+log4j.appender.FA.file=target/unit-tests.log
+log4j.appender.FA.layout=org.apache.log4j.PatternLayout
+log4j.appender.FA.layout.ConversionPattern=%d{HH:mm:ss.SSS} %t %p %c{2}: %m%n
+
+# Set the logger level of File Appender to WARN
+log4j.appender.FA.Threshold = DEBUG
+
+# SPARK-34128：Suppress undesirable TTransportException warnings involved in THRIFT-4805
+log4j.appender.console.filter.1=org.apache.log4j.varia.StringMatchFilter
+log4j.appender.console.filter.1.StringToMatch=Thrift error occurred during processing of message
+log4j.appender.console.filter.1.AcceptOnMatch=false

externals/kyuubi-spark-sql-engine/pom.xml

@@ -49,7 +49,7 @@
             <artifactId>kyuubi-spark-monitor</artifactId>
             <version>${project.version}</version>
         </dependency>
-      
+
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-core_${scala.binary.version}</artifactId>

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

@@ -19,6 +19,8 @@ package org.apache.kyuubi.engine.spark.operation
 
 import java.util.concurrent.{RejectedExecutionException, ScheduledExecutorService, TimeUnit}
 
+import scala.collection.mutable.Map
+
 import org.apache.spark.kyuubi.SQLOperationListener
 import org.apache.spark.sql.{DataFrame, SparkSession}
 import org.apache.spark.sql.types._
@@ -26,6 +28,8 @@ import org.apache.spark.sql.types._
 import org.apache.kyuubi.{KyuubiSQLException, Logging}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.engine.spark.{ArrayFetchIterator, KyuubiSparkUtil}
+import org.apache.kyuubi.engine.spark.monitor.KyuubiStatementMonitor
+import org.apache.kyuubi.engine.spark.monitor.entity.KyuubiStatementInfo
 import org.apache.kyuubi.operation.{OperationState, OperationType}
 import org.apache.kyuubi.operation.OperationState.OperationState
 import org.apache.kyuubi.operation.log.OperationLog
@@ -58,6 +62,11 @@ class ExecuteStatement(
 
   private val operationListener: SQLOperationListener = new SQLOperationListener(this, spark)
 
+  private val kyuubiStatementInfo = KyuubiStatementInfo(
+    statementId, statement, spark.sparkContext.applicationId,
+    session.getTypeInfo.identifier, Map(state -> lastAccessTime))
+  KyuubiStatementMonitor.putStatementInfoIntoQueue(kyuubiStatementInfo)
+
   override protected def resultSchema: StructType = {
     if (result == null || result.schema.isEmpty) {
       new StructType().add("Result", "string")
@@ -84,6 +93,7 @@ class ExecuteStatement(
       // TODO: Make it configurable
       spark.sparkContext.addSparkListener(operationListener)
       result = spark.sql(statement)
+      kyuubiStatementInfo.queryExecution = result.queryExecution
       debug(result.queryExecution)
       iter = new ArrayFetchIterator(result.collect())
       setState(OperationState.FINISHED)
@@ -156,6 +166,16 @@ class ExecuteStatement(
     spark.sparkContext.removeSparkListener(operationListener)
     super.cleanup(targetState)
   }
+
+  override def setState(newState: OperationState): Unit = {
+    super.setState(newState)
+    kyuubiStatementInfo.stateToTime.put(newState, lastAccessTime)
+  }
+
+  override def setOperationException(opEx: KyuubiSQLException): Unit = {
+    super.setOperationException(opEx)
+    kyuubiStatementInfo.exception = opEx
+  }
 }
 
 object ExecuteStatement {

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/KyuubiStatementMonitorSuite.scala

@@ -0,0 +1,86 @@
+/*
+ * 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.engine.spark
+
+import java.util.concurrent.ArrayBlockingQueue
+
+import org.apache.hive.service.rpc.thrift.{TExecuteStatementReq, TGetOperationStatusReq, TOperationHandle}
+import org.apache.hive.service.rpc.thrift.TCLIService.Iface
+import org.apache.hive.service.rpc.thrift.TOperationState._
+import org.scalatest.PrivateMethodTester
+
+import org.apache.kyuubi.engine.spark.monitor.KyuubiStatementMonitor
+import org.apache.kyuubi.engine.spark.monitor.entity.KyuubiStatementInfo
+import org.apache.kyuubi.operation.HiveJDBCTests
+
+class KyuubiStatementMonitorSuite extends WithSparkSQLEngine with HiveJDBCTests
+    with PrivateMethodTester {
+
+  override protected def jdbcUrl: String = getJdbcUrl
+  override def withKyuubiConf: Map[String, String] = Map.empty
+
+  test("add kyuubiStatementInfo into queue and remove them by size type threshold") {
+    val sql = "select timestamp'2021-06-01'"
+    val total: Int = 7
+    // Clear kyuubiStatementQueue first
+    val getQueue = PrivateMethod[
+      ArrayBlockingQueue[KyuubiStatementInfo]](Symbol("kyuubiStatementQueue"))()
+    val kyuubiStatementQueue = KyuubiStatementMonitor.invokePrivate(getQueue)
+    kyuubiStatementQueue.clear()
+    withSessionHandle { (client, handle) =>
+      for ( a <- 1 to total ) {
+        val req = new TExecuteStatementReq()
+        req.setSessionHandle(handle)
+        req.setStatement(sql)
+        val tExecuteStatementResp = client.ExecuteStatement(req)
+        val operationHandle = tExecuteStatementResp.getOperationHandle
+        waitForOperationToComplete(client, operationHandle)
+      }
+
+      var iterator = kyuubiStatementQueue.iterator()
+      while (iterator.hasNext) {
+        val kyuubiStatementInfo = iterator.next()
+        assert(kyuubiStatementInfo.statement !== null)
+        assert(kyuubiStatementInfo.statementId !== null)
+        assert(kyuubiStatementInfo.sessionId !== null)
+        assert(kyuubiStatementInfo.queryExecution !== null)
+        assert(kyuubiStatementInfo.stateToTime.size === 4)
+      }
+      iterator = null
+
+      // Test for clear kyuubiStatementQueue
+      // This function is used for avoiding mem leak
+      val req = new TExecuteStatementReq()
+      req.setSessionHandle(handle)
+      req.setStatement(sql)
+      val tExecuteStatementResp = client.ExecuteStatement(req)
+      val operationHandle = tExecuteStatementResp.getOperationHandle
+      waitForOperationToComplete(client, operationHandle)
+
+      assert(kyuubiStatementQueue.size() === 1)
+    }
+  }
+
+  private def waitForOperationToComplete(client: Iface, op: TOperationHandle): Unit = {
+    val req = new TGetOperationStatusReq(op)
+    var state = client.GetOperationStatus(req).getOperationState
+    while (state == INITIALIZED_STATE || state == PENDING_STATE || state == RUNNING_STATE) {
+      state = client.GetOperationStatus(req).getOperationState
+    }
+  }
+}