Merge branch 'opensearch-project:main' into zips-to-maven

.github/ISSUE_TEMPLATE/documentation.md

@@ -0,0 +1,11 @@
+**Is your feature request related to a problem?**
+A new feature has been added.
+
+**What solution would you like?**
+Document the usage of the new feature.
+
+**What alternatives have you considered?**
+N/A
+
+**Do you have any additional context?**
+_Add any other context or screenshots about the feature request here._

.github/workflows/create-documentation-issue.yml

@@ -0,0 +1,41 @@
+name: Create Documentation Issue
+on:
+  pull_request:
+    types:
+      - labeled
+env:
+  PR_NUMBER: ${{ github.event.number }}
+
+jobs:
+  create-issue:
+    if: ${{ github.event.label.name == 'needs-documentation' }}
+    runs-on: ubuntu-latest
+    name: Create Documentation Issue
+    steps:
+      - name: GitHub App token
+        id: github_app_token
+        uses: tibdex/[email protected]
+        with:
+          app_id: ${{ secrets.APP_ID }}
+          private_key: ${{ secrets.APP_PRIVATE_KEY }}
+          installation_id: 22958780
+
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Edit the issue template
+        run: |
+          echo "https://github.com/opensearch-project/index-management/pull/${{ env.PR_NUMBER }}." >> ./.github/ISSUE_TEMPLATE/documentation.md
+
+      - name: Create Issue From File
+        id: create-issue
+        uses: peter-evans/create-issue-from-file@v4
+        with:
+          title: Add documentation related to new feature
+          content-filepath: ./.github/ISSUE_TEMPLATE/documentation.md
+          labels: documentation
+          repository: opensearch-project/documentation-website
+          token: ${{ steps.github_app_token.outputs.token }}
+
+      - name: Print Issue
+        run: echo Created related documentation issue ${{ steps.create-issue.outputs.issue-number }}

.gitignore

@@ -11,4 +11,6 @@ http
 .project
 .settings
 src/test/resources/job-scheduler/
-src/test/resources/bwc/
+src/test/resources/bwc/
+src/test/resources/notifications-core/
+src/test/resources/notifications/

build.gradle

@@ -14,9 +14,9 @@ import java.util.function.Predicate
 buildscript {
     ext {
         isSnapshot = "true" == System.getProperty("build.snapshot", "true")
-        opensearch_version = System.getProperty("opensearch.version", "2.0.0-rc1-SNAPSHOT")
-        buildVersionQualifier = System.getProperty("build.version_qualifier", "rc1")
-        // 2.0.0-rc1-SNAPSHOT -> 2.0.0.0-rc1-SNAPSHOT
+        opensearch_version = System.getProperty("opensearch.version", "2.0.0-SNAPSHOT")
+        buildVersionQualifier = System.getProperty("build.version_qualifier", "")
+        // 2.0.0-SNAPSHOT -> 2.0.0.0-SNAPSHOT
         version_tokens = opensearch_version.tokenize('-')
         opensearch_build = version_tokens[0] + '.0'
         job_scheduler_no_snapshot = opensearch_build

src/main/kotlin/org/opensearch/indexmanagement/indexstatemanagement/ManagedIndexCoordinator.kt

@@ -633,28 +633,49 @@ class ManagedIndexCoordinator(
     suspend fun sweepManagedIndexJobs(client: Client): List<String> {
         val managedIndexUuids = mutableListOf<String>()
 
-        val managedIndexSearchRequest = getSweptManagedIndexSearchRequest()
-        var response: SearchResponse = client.suspendUntil { search(managedIndexSearchRequest, it) }
-        var uuids = response.hits.map { it.id }
-        val scrollIDsToClear = mutableSetOf<String>()
-
-        while (uuids.isNotEmpty()) {
-            managedIndexUuids.addAll(uuids)
-            val scrollID = response.scrollId
-            scrollIDsToClear.add(scrollID)
-            val scrollRequest = SearchScrollRequest().scrollId(scrollID).scroll(TimeValue.timeValueMinutes(1))
-            response = client.suspendUntil { searchScroll(scrollRequest, it) }
-            uuids = response.hits.map { it.id }
-        }
+        // if # of documents below 10k, don't use scroll search
+        val countReq = getSweptManagedIndexSearchRequest(size = 0)
+        val countRes: SearchResponse = client.suspendUntil { search(countReq, it) }
+        val totalHits = countRes.hits.totalHits ?: return managedIndexUuids
 
-        if (scrollIDsToClear.isNotEmpty()) {
-            val clearScrollRequest = ClearScrollRequest()
-            clearScrollRequest.scrollIds(scrollIDsToClear.toList())
-            val clearScrollResponse: ClearScrollResponse =
-                client.suspendUntil { execute(ClearScrollAction.INSTANCE, clearScrollRequest, it) }
+        if (totalHits.value >= MAX_HITS) {
+            val scrollIDsToClear = mutableSetOf<String>()
+            try {
+                val managedIndexSearchRequest = getSweptManagedIndexSearchRequest(scroll = true)
+                var response: SearchResponse = client.suspendUntil { search(managedIndexSearchRequest, it) }
+                var uuids = transformManagedIndexSearchRes(response)
+
+                while (uuids.isNotEmpty()) {
+                    managedIndexUuids.addAll(uuids)
+                    val scrollID = response.scrollId
+                    scrollIDsToClear.add(scrollID)
+                    val scrollRequest = SearchScrollRequest().scrollId(scrollID).scroll(TimeValue.timeValueMinutes(1))
+                    response = client.suspendUntil { searchScroll(scrollRequest, it) }
+                    uuids = transformManagedIndexSearchRes(response)
+                }
+            } finally {
+                if (scrollIDsToClear.isNotEmpty()) {
+                    val clearScrollRequest = ClearScrollRequest()
+                    clearScrollRequest.scrollIds(scrollIDsToClear.toList())
+                    val clearScrollResponse: ClearScrollResponse =
+                        client.suspendUntil { execute(ClearScrollAction.INSTANCE, clearScrollRequest, it) }
+                }
+            }
+            return managedIndexUuids
         }
 
-        return managedIndexUuids
+        val response: SearchResponse = client.suspendUntil { search(getSweptManagedIndexSearchRequest(), it) }
+        return transformManagedIndexSearchRes(response)
+    }
+
+    fun transformManagedIndexSearchRes(response: SearchResponse): List<String> {
+        if (response.isTimedOut || response.failedShards > 0 || response.skippedShards > 0) {
+            val errorMsg = "Sweep managed indices failed. Timed out: ${response.isTimedOut} | " +
+                "Failed shards: ${response.failedShards} | Skipped shards: ${response.skippedShards}."
+            logger.error(errorMsg)
+            throw ISMCoordinatorSearchException(message = errorMsg)
+        }
+        return response.hits.map { it.id }
     }
 
     /**
@@ -736,3 +757,5 @@ class ManagedIndexCoordinator(
         const val BUFFER = 20L
     }
 }
+
+class ISMCoordinatorSearchException(message: String, cause: Throwable? = null) : Exception(message, cause)