Merge pull request #2612 from royalhuang/issue_2596_release-1.2

perf: 日志模块采用Rest Client方式访问ES #2596

src/backend/ci/build.gradle

@@ -39,7 +39,7 @@ buildscript {
         swaggerVersion = "1.5.17"
         jaxrsVersion = "2.0"
         jacksonVersion = "2.9.10"
-        esVersion = "5.6.14"
+        esVersion = "7.0.0"
         searchGuardSSLVersion = "5.6.14-23"
         hashidsVersion = "1.0.3"
         feignVersion = "9.5.1"
@@ -236,8 +236,8 @@ subprojects {
             dependency "com.google.guava:guava:$guavaVersion"
             dependency "com.floragunn:search-guard-ssl:$searchGuardSSLVersion"
             dependency "org.elasticsearch:elasticsearch:$esVersion"
-            dependency "org.elasticsearch.client:transport:$esVersion"
-            dependency "org.elasticsearch.plugin:transport-netty4-client:$esVersion"
+            dependency "org.elasticsearch.client:elasticsearch-rest-client:$esVersion"
+            dependency "org.elasticsearch.client:elasticsearch-rest-high-level-client:$esVersion"
             dependency "org.hashids:hashids:$hashidsVersion"
             dependency "javax.ws.rs:javax.ws.rs-api:$jaxrsVersion"
             dependency "com.squareup.okhttp3:okhttp:$okHttpVersion"

src/backend/ci/core/common/common-es/build.gradle

@@ -27,7 +27,8 @@
 dependencies {
     compile project(":core:common:common-web")
     compile "org.elasticsearch:elasticsearch"
-    compile "org.elasticsearch.client:transport"
+    compile "org.elasticsearch.client:elasticsearch-rest-client"
+    compile "org.elasticsearch.client:elasticsearch-rest-high-level-client"
     compile "org.apache.logging.log4j:log4j-core"
     compile "org.apache.logging.log4j:log4j-api"
     compile "com.floragunn:search-guard-ssl"

src/backend/ci/core/common/common-es/src/main/kotlin/com/tencent/devops/common/es/ESAutoConfiguration.kt

@@ -26,12 +26,15 @@
 
 package com.tencent.devops.common.es
 
-import com.floragunn.searchguard.ssl.SearchGuardSSLPlugin
-import com.floragunn.searchguard.ssl.util.SSLConfigConstants
 import com.tencent.devops.common.web.WebAutoConfiguration
-import org.elasticsearch.common.settings.Settings
-import org.elasticsearch.common.transport.InetSocketTransportAddress
-import org.elasticsearch.transport.client.PreBuiltTransportClient
+import org.apache.http.HttpHost
+import org.apache.http.auth.AuthScope
+import org.apache.http.auth.UsernamePasswordCredentials
+import org.apache.http.impl.client.BasicCredentialsProvider
+import org.apache.http.ssl.SSLContexts
+import org.elasticsearch.client.RestClient
+import org.elasticsearch.client.RestHighLevelClient
+import org.springframework.beans.factory.DisposableBean
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.autoconfigure.AutoConfigureBefore
 import org.springframework.boot.autoconfigure.AutoConfigureOrder
@@ -40,21 +43,30 @@ import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.context.annotation.Primary
 import org.springframework.core.Ordered
-import java.net.InetAddress
+import java.io.File
+import java.io.FileInputStream
+import java.security.KeyStore
+import javax.net.ssl.SSLContext
 
 @Configuration
 @AutoConfigureOrder(Ordered.HIGHEST_PRECEDENCE)
 @AutoConfigureBefore(WebAutoConfiguration::class)
 @EnableConfigurationProperties(ESProperties::class)
-class ESAutoConfiguration {
+class ESAutoConfiguration : DisposableBean {
     @Value("\${elasticsearch.ip}")
     private val ip: String? = null
     @Value("\${elasticsearch.port}")
-    private val port: Int? = 0
+    private val port: Int? = 9200
     @Value("\${elasticsearch.cluster}")
     private val cluster: String? = null
     @Value("\${elasticsearch.name}")
     private val name: String? = null
+    @Value("\${elasticsearch.username:#{null}}")
+    private val username: String? = null
+    @Value("\${elasticsearch.password:#{null}}")
+    private val password: String? = null
+    @Value("\${elasticsearch.https}")
+    private val https: String? = null
     @Value("\${elasticsearch.keystore.filePath:#{null}}")
     private val keystoreFilePath: String? = null
     @Value("\${elasticsearch.keystore.password:#{null}}")
@@ -64,6 +76,8 @@ class ESAutoConfiguration {
     @Value("\${elasticsearch.truststore.password:#{null}}")
     private val truststorePassword: String? = null
 
+    private var client: RestHighLevelClient? = null
+
     @Bean
     @Primary
     fun transportClient(): ESClient {
@@ -80,14 +94,24 @@ class ESAutoConfiguration {
             throw IllegalArgumentException("ES唯一名称尚未配置: elasticsearch.name")
         }
 
-        val builder = Settings.builder()
-            .put("cluster.name", cluster)
-            .put("client.transport.sniff", true)
-        val searchGuard =
-            !keystoreFilePath.isNullOrBlank() || !truststoreFilePath.isNullOrBlank() ||
-                !keystorePassword.isNullOrBlank() || !truststorePassword.isNullOrBlank()
+        var httpHost = HttpHost(ip, port ?: 9200, "http")
+        var sslContext: SSLContext? = null
+
+        // 基础鉴权 - 账号密码
+        val credentialsProvider = if (!username.isNullOrBlank() || !password.isNullOrBlank()) {
+            if (username.isNullOrBlank()) {
+                throw IllegalArgumentException("缺少配置: elasticsearch.username")
+            }
+            if (password.isNullOrBlank()) {
+                throw IllegalArgumentException("缺少配置: elasticsearch.password")
+            }
+            val provider = BasicCredentialsProvider()
+            provider.setCredentials(AuthScope.ANY, UsernamePasswordCredentials(username, password))
+            provider
+        } else null
 
-        val client = if (searchGuard) {
+        // HTTPS鉴权 - SSL证书
+        if (enableSSL(https)) {
             if (keystoreFilePath.isNullOrBlank()) {
                 throw IllegalArgumentException("SearchGuard认证缺少配置: elasticsearch.keystore.filePath")
             }
@@ -101,23 +125,50 @@ class ESAutoConfiguration {
                 throw IllegalArgumentException("SearchGuard认证缺少配置: elasticsearch.truststore.password")
             }
 
-            builder.put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_ENFORCE_HOSTNAME_VERIFICATION, false)
-                .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_ENABLED, true)
-                .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_ENFORCE_HOSTNAME_VERIFICATION_RESOLVE_HOST_NAME, true)
-                .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_KEYSTORE_FILEPATH, keystoreFilePath)
-                .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_TRUSTSTORE_FILEPATH, truststoreFilePath)
-                .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_KEYSTORE_PASSWORD, keystorePassword)
-                .put(SSLConfigConstants.SEARCHGUARD_SSL_TRANSPORT_TRUSTSTORE_PASSWORD, truststorePassword)
+            val keystoreFile = File(keystoreFilePath!!)
+            if (!keystoreFile.exists()) {
+                throw IllegalArgumentException("未找到 keystore 文件，请检查路径是否正确: $keystoreFilePath")
+            }
+            val truststoreFile = File(truststoreFilePath!!)
+            if (!truststoreFile.exists()) {
+                throw IllegalArgumentException("未找到 truststore 文件，请检查路径是否正确: $truststoreFilePath")
+            }
 
-            PreBuiltTransportClient(builder.build(), SearchGuardSSLPlugin::class.java)
-        } else {
-            PreBuiltTransportClient(builder.build())
+            val keyStore = KeyStore.getInstance(KeyStore.getDefaultType())
+            val keystorePasswordCharArray = keystorePassword!!.toCharArray()
+            keyStore.load(FileInputStream(keystoreFile), keystorePasswordCharArray)
+            val truststore = KeyStore.getInstance(KeyStore.getDefaultType())
+            val truststorePasswordCharArray = truststorePassword!!.toCharArray()
+            truststore.load(FileInputStream(truststoreFile), truststorePasswordCharArray)
+
+            httpHost = HttpHost(ip, port ?: 9200, "https")
+            sslContext = SSLContexts.custom()
+                .loadTrustMaterial(truststore, null)
+                .loadKeyMaterial(keyStore, keystorePasswordCharArray)
+                .build()
         }
 
-        val ips = ip!!.split(",".toRegex()).dropLastWhile { it.isEmpty() }.toTypedArray()
-        for (ipAddress in ips) {
-            client.addTransportAddress(InetSocketTransportAddress(InetAddress.getByName(ipAddress), port!!))
+        // 初始化 RestClient 配置
+        val builder = RestClient.builder(httpHost)
+        builder.setHttpClientConfigCallback { httpClientBuilder ->
+            if (sslContext != null) httpClientBuilder.setSSLContext(sslContext)
+            if (credentialsProvider != null) httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
+            httpClientBuilder
+        }
+
+        client = RestHighLevelClient(builder)
+        return ESClient(name!!, client!!)
+    }
+
+    override fun destroy() {
+        client?.close()
+    }
+
+    private fun enableSSL(https: String?): Boolean {
+        return if (!https.isNullOrBlank()) {
+            https!!.toBoolean()
+        } else {
+            false
         }
-        return ESClient(name!!, client)
     }
 }

src/backend/ci/core/common/common-es/src/main/kotlin/com/tencent/devops/common/es/ESClient.kt

@@ -26,10 +26,10 @@
 
 package com.tencent.devops.common.es
 
-import org.elasticsearch.client.Client
+import org.elasticsearch.client.RestHighLevelClient
 
 data class ESClient(
     val name: String,
-    val client: Client,
+    val client: RestHighLevelClient,
     val mainCluster: Boolean? = false
 )

src/backend/ci/core/log/biz-log/src/main/kotlin/com/tencent/devops/log/client/LogClient.kt

@@ -27,23 +27,13 @@
 package com.tencent.devops.log.client
 
 import com.tencent.devops.common.es.ESClient
-import org.elasticsearch.client.Client
+import org.elasticsearch.client.RestHighLevelClient
 
 interface LogClient {
 
-    fun admin(buildId: String) = getClient(buildId).admin()
+    fun restClient(buildId: String) = getClient(buildId)
 
-    fun prepareBulk(buildId: String) = getClient(buildId).prepareBulk()
-
-    fun prepareSearch(buildId: String, index: String) = getClient(buildId).prepareSearch(index)
-
-    fun prepareMultiSearch(buildId: String) = getClient(buildId).prepareMultiSearch()
-
-    fun prepareSearchScroll(buildId: String, scrollId: String) = getClient(buildId).prepareSearchScroll(scrollId)
-
-    fun prepareIndex(buildId: String, index: String, type: String) = getClient(buildId).prepareIndex(index, type)
-
-    private fun getClient(buildId: String): Client {
+    private fun getClient(buildId: String): RestHighLevelClient {
         return hashClient(buildId).client
     }
 

src/backend/ci/core/log/biz-log/src/main/kotlin/com/tencent/devops/log/cron/ESIndexCloseJob.kt

@@ -31,7 +31,11 @@ import com.tencent.devops.common.redis.RedisLock
 import com.tencent.devops.common.redis.RedisOperation
 import com.tencent.devops.log.client.LogClient
 import com.tencent.devops.log.util.IndexNameUtils.LOG_PREFIX
-import org.elasticsearch.client.Client
+import org.elasticsearch.action.admin.indices.close.CloseIndexRequest
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest
+import org.elasticsearch.client.indices.GetIndexRequest
+import org.elasticsearch.client.RequestOptions
+import org.elasticsearch.client.RestHighLevelClient
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.scheduling.annotation.Scheduled
@@ -83,10 +87,9 @@ class ESIndexCloseJob @Autowired constructor(
 
     private fun closeESIndexes() {
         client.getActiveClients().forEach { c ->
-            val indexes = c.client.admin()
+            val indexes = c.client
                 .indices()
-                .prepareGetIndex()
-                .get()
+                .get(GetIndexRequest(), RequestOptions.DEFAULT)
 
             if (indexes.indices.isEmpty()) {
                 return
@@ -103,21 +106,18 @@ class ESIndexCloseJob @Autowired constructor(
         }
     }
 
-    private fun closeESIndex(c: Client, index: String) {
+    private fun closeESIndex(c: RestHighLevelClient, index: String) {
         logger.info("[$index] Start to close ES index")
-        val resp = c.admin()
-            .indices()
-            .prepareClose(index)
-            .get()
+        val resp = c.indices()
+            .close(CloseIndexRequest(index), RequestOptions.DEFAULT)
         logger.info("Get the close es response - ${resp.isAcknowledged}")
     }
 
     private fun deleteESIndexes() {
         client.getActiveClients().forEach { c ->
-            val indexes = c.client.admin()
+            val indexes = c.client
                 .indices()
-                .prepareGetIndex()
-                .get()
+                .get(GetIndexRequest(), RequestOptions.DEFAULT)
 
             if (indexes.indices.isEmpty()) {
                 return
@@ -134,12 +134,10 @@ class ESIndexCloseJob @Autowired constructor(
         }
     }
 
-    private fun deleteESIndex(c: Client, index: String) {
+    private fun deleteESIndex(c: RestHighLevelClient, index: String) {
         logger.info("[$index] Start to delete ES index")
-        val resp = c.admin()
-            .indices()
-            .prepareDelete(index)
-            .get()
+        val resp = c.indices()
+            .delete(DeleteIndexRequest(index), RequestOptions.DEFAULT)
         logger.info("Get the delete es response - ${resp.isAcknowledged}")
     }