[KYUUBI #3104] Support SSL for Etcd

### _Why are the changes needed?_

Support SSL for Etcd

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [X] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #3105 from hddong/support-etcd-ssl.

Closes #3104

49aadb9 [hongdongdong] change enable to enabled
87fa626 [hongdongdong] [KYUUBI #3104] Support SSL for Etcd

Authored-by: hongdongdong <hongdongdong@cmss.chinamobile.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: hddong

docs/deployment/settings.md

@@ -312,6 +312,10 @@ Key | Default | Meaning | Type | Since
 kyuubi.ha.addresses||The connection string for the discovery ensemble|string|1.6.0
 kyuubi.ha.client.class|org.apache.kyuubi.ha.client.zookeeper.ZookeeperDiscoveryClient|Class name for service discovery client.<ul> <li>Zookeeper: org.apache.kyuubi.ha.client.zookeeper.ZookeeperDiscoveryClient</li> <li>Etcd: org.apache.kyuubi.ha.client.etcd.EtcdDiscoveryClient</li></ul>|string|1.6.0
 kyuubi.ha.etcd.lease.timeout|PT10S|Timeout for etcd keep alive lease. The kyuubi server will known unexpected loss of engine after up to this seconds.|duration|1.6.0
+kyuubi.ha.etcd.ssl.ca.path|&lt;undefined&gt;|Where the etcd CA certificate file is stored.|string|1.6.0
+kyuubi.ha.etcd.ssl.client.certificate.path|&lt;undefined&gt;|Where the etcd SSL certificate file is stored.|string|1.6.0
+kyuubi.ha.etcd.ssl.client.key.path|&lt;undefined&gt;|Where the etcd SSL key file is stored.|string|1.6.0
+kyuubi.ha.etcd.ssl.enabled|false|When set to true, will build a ssl secured etcd client.|boolean|1.6.0
 kyuubi.ha.namespace|kyuubi|The root directory for the service to deploy its instance uri|string|1.6.0
 kyuubi.ha.zookeeper.acl.enabled|false|Set to true if the zookeeper ensemble is kerberized|boolean|1.0.0
 kyuubi.ha.zookeeper.auth.digest|&lt;undefined&gt;|The digest auth string is used for zookeeper authentication, like: username:password.|string|1.3.2

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/HighAvailabilityConf.scala

@@ -191,4 +191,32 @@ object HighAvailabilityConf {
       .timeConf
       .checkValue(_ > 0, "Must be positive")
       .createWithDefault(Duration.ofSeconds(10).toMillis)
+
+  val HA_ETCD_SSL_ENABLED: ConfigEntry[Boolean] =
+    buildConf("kyuubi.ha.etcd.ssl.enabled")
+      .doc("When set to true, will build a ssl secured etcd client.")
+      .version("1.6.0")
+      .booleanConf
+      .createWithDefault(false)
+
+  val HA_ETCD_SSL_CA_PATH: OptionalConfigEntry[String] =
+    buildConf("kyuubi.ha.etcd.ssl.ca.path")
+      .doc("Where the etcd CA certificate file is stored.")
+      .version("1.6.0")
+      .stringConf
+      .createOptional
+
+  val HA_ETCD_SSL_CLINET_CRT_PATH: OptionalConfigEntry[String] =
+    buildConf("kyuubi.ha.etcd.ssl.client.certificate.path")
+      .doc("Where the etcd SSL certificate file is stored.")
+      .version("1.6.0")
+      .stringConf
+      .createOptional
+
+  val HA_ETCD_SSL_CLINET_KEY_PATH: OptionalConfigEntry[String] =
+    buildConf("kyuubi.ha.etcd.ssl.client.key.path")
+      .doc("Where the etcd SSL key file is stored.")
+      .version("1.6.0")
+      .stringConf
+      .createOptional
 }

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/client/etcd/EtcdDiscoveryClient.scala

@@ -17,6 +17,7 @@
 
 package org.apache.kyuubi.ha.client.etcd
 
+import java.io.File
 import java.nio.charset.StandardCharsets.UTF_8
 import java.util.concurrent.TimeUnit
 
@@ -36,6 +37,7 @@ import io.etcd.jetcd.options.GetOption
 import io.etcd.jetcd.options.PutOption
 import io.etcd.jetcd.watch.WatchEvent
 import io.etcd.jetcd.watch.WatchResponse
+import io.grpc.netty.GrpcSslContexts
 import io.grpc.stub.StreamObserver
 
 import org.apache.kyuubi.KYUUBI_VERSION
@@ -44,7 +46,7 @@ import org.apache.kyuubi.KyuubiSQLException
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.ENGINE_INIT_TIMEOUT
 import org.apache.kyuubi.ha.HighAvailabilityConf
-import org.apache.kyuubi.ha.HighAvailabilityConf.HA_ENGINE_REF_ID
+import org.apache.kyuubi.ha.HighAvailabilityConf._
 import org.apache.kyuubi.ha.client.DiscoveryClient
 import org.apache.kyuubi.ha.client.DiscoveryPaths
 import org.apache.kyuubi.ha.client.ServiceDiscovery
@@ -63,9 +65,32 @@ class EtcdDiscoveryClient(conf: KyuubiConf) extends DiscoveryClient {
 
   var leaseTTL: Long = _
 
+  private def buildClient(): Client = {
+    val endpoints = conf.get(HA_ADDRESSES).split(",")
+    val sslEnabled = conf.get(HA_ETCD_SSL_ENABLED)
+    if (!sslEnabled) {
+      Client.builder.endpoints(endpoints: _*).build
+    } else {
+      val caPath = conf.getOption(HA_ETCD_SSL_CA_PATH.key).getOrElse(
+        throw new IllegalArgumentException(s"${HA_ETCD_SSL_CA_PATH.key} is not defined"))
+      val crtPath = conf.getOption(HA_ETCD_SSL_CLINET_CRT_PATH.key).getOrElse(
+        throw new IllegalArgumentException(s"${HA_ETCD_SSL_CLINET_CRT_PATH.key} is not defined"))
+      val keyPath = conf.getOption(HA_ETCD_SSL_CLINET_KEY_PATH.key).getOrElse(
+        throw new IllegalArgumentException(s"${HA_ETCD_SSL_CLINET_KEY_PATH.key} is not defined"))
+
+      val context = GrpcSslContexts.forClient()
+        .trustManager(new File(caPath))
+        .keyManager(new File(crtPath), new File(keyPath))
+        .build()
+      Client.builder()
+        .endpoints(endpoints: _*)
+        .sslContext(context)
+        .build()
+    }
+  }
+
   def createClient(): Unit = {
-    val endpoints = conf.get(HighAvailabilityConf.HA_ADDRESSES).split(",")
-    client = Client.builder.endpoints(endpoints: _*).build
+    client = buildClient()
     kvClient = client.getKVClient()
     lockClient = client.getLockClient()
     leaseClient = client.getLeaseClient()