[RORDEV-1927][RORDEV-1929][RORDEV-1931] ES 9.3.0, 9.2.5, 8.19.11 support (#1209)

Author: coutoPL

ci/supported-es-versions/es8x.txt

@@ -1,3 +1,4 @@
+8.19.11
 8.19.10
 8.19.9
 8.19.8

ci/supported-es-versions/es9x.txt

@@ -1,3 +1,5 @@
+9.3.0
+9.2.5
 9.2.4
 9.2.3
 9.2.2

es818x/gradle.properties

@@ -1 +1 @@
-latestSupportedEsVersion=8.19.10
+latestSupportedEsVersion=8.19.11

es92x/build.gradle

@@ -38,7 +38,9 @@ dependencies {
     implementation  project(path: ':core')
     implementation  project(path: ':ror-tools', configuration: 'shadow')
     implementation  project(path: ':ror-tools-core')
-    implementation  group: 'org.elasticsearch',                 name: 'elasticsearch' ,                         version: moduleEsVersion
+    implementation (group: 'org.elasticsearch',                 name: 'elasticsearch' ,                         version: moduleEsVersion) {
+        exclude group: 'org.elasticsearch', module: 'exponential-histogram'
+    }
     implementation  group: 'org.elasticsearch.client',          name: 'elasticsearch-rest-client',              version: moduleEsVersion
     implementation  group: 'org.apache.httpcomponents',         name: 'httpclient',                             version: '4.5.14'
     implementation  group: 'org.apache.logging.log4j',          name: 'log4j-core',                             version: '2.11.1'

es92x/gradle.properties

@@ -1 +1 @@
-latestSupportedEsVersion=9.2.4
+latestSupportedEsVersion=9.3.0

es92x/src/main/scala/tech/beshu/ror/es/IndexLevelActionFilter.scala

@@ -167,7 +167,8 @@ class IndexLevelActionFilter(clusterService: ClusterService,
               request,
               rorActionListener,
               chain,
-              JavaConverters.flattenPair(threadPool.getThreadContext.getResponseHeaders).toCovariantSet
+              JavaConverters.flattenPair(threadPool.getThreadContext.getResponseHeaders).toCovariantSet,
+              esEnv.esVersion
             )
           )
         } recover {

es92x/src/main/scala/tech/beshu/ror/es/handler/AclAwareRequestFilter.scala

@@ -75,7 +75,7 @@ import tech.beshu.ror.es.handler.request.context.types.repositories.*
 import tech.beshu.ror.es.handler.request.context.types.ror.*
 import tech.beshu.ror.es.handler.request.context.types.snapshots.*
 import tech.beshu.ror.es.handler.request.context.types.templates.*
-import tech.beshu.ror.es.{HidingInternalErrorDetailsRorActionListener, RorActionListener, RorClusterService, RorRestChannel, AtEsLevelUpdateActionResponseListener}
+import tech.beshu.ror.es.{EsVersion, HidingInternalErrorDetailsRorActionListener, RorActionListener, RorClusterService, RorRestChannel, AtEsLevelUpdateActionResponseListener}
 import tech.beshu.ror.implicits.*
 import tech.beshu.ror.syntax.*
 
@@ -278,7 +278,8 @@ object AclAwareRequestFilter {
                         val actionRequest: ActionRequest,
                         val listener: RorActionListener[ActionResponse],
                         val chain: EsChain,
-                        val threadContextResponseHeaders: Set[(String, String)]) {
+                        val threadContextResponseHeaders: Set[(String, String)],
+                        val esVersion: EsVersion) {
 
     val timestamp: Instant = Instant.now()
 

es92x/src/main/scala/tech/beshu/ror/es/handler/request/context/types/EsqlIndicesEsRequestContext.scala

@@ -40,12 +40,13 @@ class EsqlIndicesEsRequestContext private(actionRequest: ActionRequest with Comp
                                           esContext: EsContext,
                                           aclContext: AccessControlStaticContext,
                                           clusterService: RorClusterService,
-                                          override val threadPool: ThreadPool)
+                                          override val threadPool: ThreadPool,
+                                          esqlRequestHelper: EsqlRequestHelper)
   extends BaseFilterableEsRequestContext[ActionRequest with CompositeIndicesRequest](actionRequest, esContext, aclContext, clusterService, threadPool) {
 
   override protected def requestFieldsUsage: RequestFieldsUsage = RequestFieldsUsage.NotUsingFields
 
-  private lazy val requestClassification = EsqlRequestHelper.classifyEsqlRequest(actionRequest)
+  private lazy val requestClassification = esqlRequestHelper.classifyEsqlRequest(actionRequest)
 
   override protected def requestedIndicesFrom(request: ActionRequest with CompositeIndicesRequest): Set[RequestedIndex[ClusterIndexName]] = {
     requestClassification match {
@@ -74,7 +75,7 @@ class EsqlIndicesEsRequestContext private(actionRequest: ActionRequest with Comp
       case Right(r@EsqlRequestClassification.IndicesRelated(tables)) =>
         val filteredIndicesStrings = filteredIndices.stringify.toCovariantSet
         if (filteredIndicesStrings != r.indices) {
-          EsqlRequestHelper.modifyIndicesOf(request, tables, filteredIndicesStrings)
+          esqlRequestHelper.modifyIndicesOf(request, tables, filteredIndicesStrings)
         } else {
           request
         }
@@ -103,7 +104,7 @@ class EsqlIndicesEsRequestContext private(actionRequest: ActionRequest with Comp
   private def applyFieldLevelSecurityTo(response: ActionResponse,
                                         fieldLevelSecurity: Option[FieldLevelSecurity]) = {
     fieldLevelSecurity match {
-      case Some(fls) => EsqlRequestHelper.modifyResponseAccordingToFieldLevelSecurity(response, fls)
+      case Some(fls) => esqlRequestHelper.modifyResponseAccordingToFieldLevelSecurity(response, fls)
       case None => response
     }
   }
@@ -126,7 +127,8 @@ object EsqlIndicesEsRequestContext {
         arg.esContext,
         arg.aclContext,
         arg.clusterService,
-        arg.threadPool
+        arg.threadPool,
+        new EsqlRequestHelper(arg.esContext.esVersion)
       ))
     } else {
       None

es92x/src/main/scala/tech/beshu/ror/es/utils/EsqlRequestHelper.scala

@@ -23,8 +23,10 @@ import org.joor.Reflect.*
 import org.joor.ReflectException
 import tech.beshu.ror.accesscontrol.domain.FieldLevelSecurity
 import tech.beshu.ror.accesscontrol.domain.FieldLevelSecurity.FieldsRestrictions
+import tech.beshu.ror.es.EsVersion
 import tech.beshu.ror.es.handler.response.FieldsFiltering
 import tech.beshu.ror.es.handler.response.FieldsFiltering.NonMetadataDocumentFields
+import tech.beshu.ror.es.utils.EsqlRequestHelper.{ClassificationError, EsqlRequestClassification, IndexTable}
 import tech.beshu.ror.syntax.*
 import tech.beshu.ror.utils.ScalaOps.*
 
@@ -33,7 +35,7 @@ import java.util.List as JList
 import scala.jdk.CollectionConverters.*
 import scala.util.{Failure, Success, Try}
 
-object EsqlRequestHelper {
+class EsqlRequestHelper(esVersion: EsVersion) {
 
   def modifyIndicesOf(request: CompositeIndicesRequest,
                       requestTables: NonEmptyList[IndexTable],
@@ -117,15 +119,36 @@ object EsqlRequestHelper {
     }
 
     private def indicesFrom(statement: Any) = {
-      val preAnalyze =  doPreAnalyze(newPreAnalyzer, statement)
-      val indexPattern = indexPatternFrom(preAnalyze)
+      val plan = esVersion match {
+        case v if v >= EsVersion(9, 3, 0) => on(statement).call("plan").get[Any]()
+        case _ => statement
+      }
+      val preAnalysis = doPreAnalyze(newPreAnalyzer, plan)
+      esVersion match {
+        case v if v >= EsVersion(9, 3, 0) => indicesFromPreAnalysisForEsEqualOrAbove930(preAnalysis)
+        case _ => indicesFromPreAnalysisForEsBelow930(preAnalysis)
+      }
+    }
+
+    private def indicesFromPreAnalysisForEsBelow930(preAnalysis: Any) = {
+      val indexPattern = indexPatternFrom(preAnalysis)
       val indexPatternString = indexPatternStringFrom(indexPattern)
       NonEmptyList
         .fromList(splitIntoIndices(indexPatternString))
         .map(IndexTable(indexPatternString, _))
         .toList
     }
 
+    private def indicesFromPreAnalysisForEsEqualOrAbove930(preAnalysis: Any) = {
+      val indexesMap = on(preAnalysis).get[java.util.Map[Any, Any]]("indexes")
+      indexesMap.keySet().asScala.toList.flatMap { indexPattern =>
+        val indexPatternString = on(indexPattern).call("indexPattern").get[String]()
+        NonEmptyList
+          .fromList(splitIntoIndices(indexPatternString))
+          .map(IndexTable(indexPatternString, _))
+      }
+    }
+
     private def splitIntoIndices(tableString: String) = {
       tableString.split(',').asSafeList.filter(_.nonEmpty)
     }
@@ -243,6 +266,9 @@ object EsqlRequestHelper {
       }
     }
   }
+}
+
+object EsqlRequestHelper {
 
   final case class IndexTable(tableStringInQuery: String, indices: NonEmptyList[String])
 
@@ -258,4 +284,4 @@ object EsqlRequestHelper {
   object ClassificationError {
     final case class ParsingException(cause: Throwable) extends ClassificationError
   }
-}
+}

es92x/src/main/scala/tech/beshu/ror/es/utils/XPackSecurityAuthenticationHeader.scala

@@ -17,7 +17,7 @@
 package tech.beshu.ror.es.utils
 
 import eu.timepit.refined.types.string.NonEmptyString
-import org.elasticsearch.{TransportVersion, TransportVersions}
+import org.elasticsearch.TransportVersion
 import org.elasticsearch.common.bytes.BytesReference
 import org.elasticsearch.common.io.stream.BytesStreamOutput
 import tech.beshu.ror.accesscontrol.domain.Header
@@ -58,18 +58,10 @@ object XPackSecurityAuthenticationHeader {
     output.writeString(nodeName)
     output.writeString("__attach")
     output.writeString("__attach")
-    if(output.getTransportVersion.onOrAfter(TransportVersions.V_8_2_0)) {
-      output.writeBoolean(false)
-    }
     output.writeBoolean(false)
-    if (output.getTransportVersion.onOrAfter(TransportVersions.V_7_0_0)) {
-      output.writeVInt(4) // Internal
-      if(output.getTransportVersion.onOrAfter(TransportVersions.V_8_8_0)) {
-        output.writeVInt(0)
-      } else {
-        output.writeGenericMap(Map.empty[String, Object].asJava)
-      }
-    }
+    output.writeBoolean(false)
+    output.writeVInt(4) // Internal
+    output.writeVInt(0)
     NonEmptyString.unsafeFrom {
       Base64.getEncoder.encodeToString(BytesReference.toBytes(output.bytes()))
     }