Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fixing snapshot bug #1257

Merged
merged 3 commits into from
Sep 17, 2024
Merged

Fixing snapshot bug #1257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f1d43be
Fixing snapshot bug
tandonks Sep 16, 2024
f7a9164
Merge branch 'main' into main
tandonks Sep 16, 2024
96cd9c4
Merge branch 'main' into main
bowenlan-amzn Sep 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 22 additions & 25 deletions .../org/opensearch/indexmanagement/indexstatemanagement/step/snapshot/WaitForSnapshotStep.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,16 @@ package org.opensearch.indexmanagement.indexstatemanagement.step.snapshot

import org.apache.logging.log4j.LogManager
import org.opensearch.ExceptionsHelper
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotStatus
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotsStatusRequest
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotsStatusResponse
import org.opensearch.cluster.SnapshotsInProgress.State
import org.opensearch.action.admin.cluster.snapshots.get.GetSnapshotsRequest
import org.opensearch.action.admin.cluster.snapshots.get.GetSnapshotsResponse
import org.opensearch.indexmanagement.indexstatemanagement.action.SnapshotAction
import org.opensearch.indexmanagement.opensearchapi.suspendUntil
import org.opensearch.indexmanagement.spi.indexstatemanagement.Step
import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ActionProperties
import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ManagedIndexMetaData
import org.opensearch.indexmanagement.spi.indexstatemanagement.model.StepMetaData
import org.opensearch.snapshots.SnapshotInfo
import org.opensearch.snapshots.SnapshotState
import org.opensearch.transport.RemoteTransportException

class WaitForSnapshotStep(private val action: SnapshotAction) : Step(name) {
Expand All @@ -33,32 +33,31 @@ class WaitForSnapshotStep(private val action: SnapshotAction) : Step(name) {

try {
val snapshotName = getSnapshotName(managedIndexMetadata, indexName) ?: return this
val request =
SnapshotsStatusRequest()
.snapshots(arrayOf(snapshotName))
.repository(repository)
val response: SnapshotsStatusResponse = context.client.admin().cluster().suspendUntil { snapshotsStatus(request, it) }
val status: SnapshotStatus? =
val newRequest = GetSnapshotsRequest()
.snapshots(arrayOf(snapshotName))
.repository(repository)
val response: GetSnapshotsResponse = context.client.admin().cluster().suspendUntil { getSnapshots(newRequest, it) }
val status: SnapshotInfo? =
response
.snapshots
.find { snapshotStatus ->
snapshotStatus.snapshot.snapshotId.name == snapshotName && snapshotStatus.snapshot.repository == repository
.find { snapshotInfo ->
snapshotInfo.snapshotId().name == snapshotName
}
if (status != null) {
when (status.state) {
State.INIT, State.STARTED -> {
when (status.state()) {
SnapshotState.IN_PROGRESS -> {
stepStatus = StepStatus.CONDITION_NOT_MET
vikasvb90 marked this conversation as resolved.
Show resolved Hide resolved
info = mapOf("message" to getSnapshotInProgressMessage(indexName), "state" to status.state.name)
info = mapOf("message" to getSnapshotInProgressMessage(indexName), "state" to status.state().toString())
}
State.SUCCESS -> {
SnapshotState.SUCCESS -> {
stepStatus = StepStatus.COMPLETED
info = mapOf("message" to getSuccessMessage(indexName), "state" to status.state.name)
info = mapOf("message" to getSuccessMessage(indexName), "state" to status.state().toString())
}
else -> { // State.FAILED, State.ABORTED
val message = getFailedExistsMessage(indexName)
logger.warn(message)
stepStatus = StepStatus.FAILED
info = mapOf("message" to message, "state" to status.state.name)
info = mapOf("message" to message, "state" to status.state().toString())
}
}
} else {
Expand Down Expand Up @@ -98,13 +97,11 @@ class WaitForSnapshotStep(private val action: SnapshotAction) : Step(name) {
return actionProperties.snapshotName
}

override fun getUpdatedManagedIndexMetadata(currentMetadata: ManagedIndexMetaData): ManagedIndexMetaData {
return currentMetadata.copy(
stepMetaData = StepMetaData(name, getStepStartTime(currentMetadata).toEpochMilli(), stepStatus),
transitionTo = null,
info = info,
)
}
override fun getUpdatedManagedIndexMetadata(currentMetadata: ManagedIndexMetaData): ManagedIndexMetaData = currentMetadata.copy(
stepMetaData = StepMetaData(name, getStepStartTime(currentMetadata).toEpochMilli(), stepStatus),
transitionTo = null,
info = info,
)

override fun isIdempotent(): Boolean = true

Expand Down
58 changes: 29 additions & 29 deletions ...tlin/org/opensearch/indexmanagement/indexstatemanagement/step/WaitForSnapshotStepTests.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,26 +11,26 @@ import com.nhaarman.mockitokotlin2.doReturn
import com.nhaarman.mockitokotlin2.mock
import com.nhaarman.mockitokotlin2.whenever
import kotlinx.coroutines.runBlocking
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotStatus
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotsStatusResponse
import org.opensearch.action.admin.cluster.snapshots.get.GetSnapshotsResponse
import org.opensearch.client.AdminClient
import org.opensearch.client.Client
import org.opensearch.client.ClusterAdminClient
import org.opensearch.cluster.SnapshotsInProgress
import org.opensearch.cluster.service.ClusterService
import org.opensearch.common.settings.Settings
import org.opensearch.core.action.ActionListener
import org.opensearch.indexmanagement.indexstatemanagement.action.SnapshotAction
import org.opensearch.indexmanagement.indexstatemanagement.step.snapshot.WaitForSnapshotStep
import org.opensearch.indexmanagement.snapshotmanagement.mockInProgressSnapshotInfo
import org.opensearch.indexmanagement.snapshotmanagement.mockSnapshotInfo
import org.opensearch.indexmanagement.spi.indexstatemanagement.Step
import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ActionMetaData
import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ActionProperties
import org.opensearch.indexmanagement.spi.indexstatemanagement.model.ManagedIndexMetaData
import org.opensearch.indexmanagement.spi.indexstatemanagement.model.StepContext
import org.opensearch.jobscheduler.spi.utils.LockService
import org.opensearch.script.ScriptService
import org.opensearch.snapshots.Snapshot
import org.opensearch.snapshots.SnapshotId
import org.opensearch.snapshots.SnapshotInfo
import org.opensearch.snapshots.SnapshotState
import org.opensearch.test.OpenSearchTestCase
import org.opensearch.transport.RemoteTransportException

Expand Down Expand Up @@ -70,13 +70,10 @@ class WaitForSnapshotStepTests : OpenSearchTestCase() {
}

fun `test snapshot status states`() {
val snapshotStatus: SnapshotStatus = mock()
val response: SnapshotsStatusResponse = mock()
whenever(response.snapshots).doReturn(listOf(snapshotStatus))
whenever(snapshotStatus.snapshot).doReturn(Snapshot("repo", SnapshotId("snapshot-name", "some_uuid")))
val snapshotInfo: SnapshotInfo = mockInProgressSnapshotInfo(snapshot)
val response: GetSnapshotsResponse = mock()
whenever(response.snapshots).doReturn(listOf(snapshotInfo))
val client = getClient(getAdminClient(getClusterAdminClient(response, null)))

whenever(snapshotStatus.state).doReturn(SnapshotsInProgress.State.INIT)
runBlocking {
val snapshotAction = SnapshotAction("repo", snapshot, 0)
val metadata = ManagedIndexMetaData("test", "indexUuid", "policy_id", null, null, null, null, null, null, null, ActionMetaData(WaitForSnapshotStep.name, 1, 0, false, 0, null, ActionProperties(snapshotName = "snapshot-name")), null, null, null)
Expand All @@ -88,31 +85,34 @@ class WaitForSnapshotStepTests : OpenSearchTestCase() {
assertEquals("Did not get snapshot in progress message", WaitForSnapshotStep.getSnapshotInProgressMessage("test"), updatedManagedIndexMetaData.info!!["message"])
}

whenever(snapshotStatus.state).doReturn(SnapshotsInProgress.State.STARTED)
val snapshotInfo2: SnapshotInfo = mockSnapshotInfo(snapshot, SnapshotState.SUCCESS)
whenever(response.snapshots).doReturn(listOf(snapshotInfo2))
runBlocking {
val snapshotAction = SnapshotAction("repo", snapshot, 0)
val metadata = ManagedIndexMetaData("test", "indexUuid", "policy_id", null, null, null, null, null, null, null, ActionMetaData(WaitForSnapshotStep.name, 1, 0, false, 0, null, ActionProperties(snapshotName = "snapshot-name")), null, null, null)
val step = WaitForSnapshotStep(snapshotAction)
val context = StepContext(metadata, clusterService, client, null, null, scriptService, settings, lockService)
step.preExecute(logger, context).execute()
val updatedManagedIndexMetaData = step.getUpdatedManagedIndexMetadata(metadata)
assertEquals("Step status is not CONDITION_NOT_MET", Step.StepStatus.CONDITION_NOT_MET, updatedManagedIndexMetaData.stepMetaData?.stepStatus)
assertEquals("Did not get snapshot in progress message", WaitForSnapshotStep.getSnapshotInProgressMessage("test"), updatedManagedIndexMetaData.info!!["message"])
assertEquals("Step status is not COMPLETED", Step.StepStatus.COMPLETED, updatedManagedIndexMetaData.stepMetaData?.stepStatus)
assertEquals("Did not get snapshot completed message", WaitForSnapshotStep.getSuccessMessage("test"), updatedManagedIndexMetaData.info!!["message"])
}

whenever(snapshotStatus.state).doReturn(SnapshotsInProgress.State.SUCCESS)
val snapshotInfo3: SnapshotInfo = mockSnapshotInfo(snapshot, SnapshotState.FAILED)
whenever(response.snapshots).doReturn(listOf(snapshotInfo3))
runBlocking {
val snapshotAction = SnapshotAction("repo", snapshot, 0)
val metadata = ManagedIndexMetaData("test", "indexUuid", "policy_id", null, null, null, null, null, null, null, ActionMetaData(WaitForSnapshotStep.name, 1, 0, false, 0, null, ActionProperties(snapshotName = "snapshot-name")), null, null, null)
val step = WaitForSnapshotStep(snapshotAction)
val context = StepContext(metadata, clusterService, client, null, null, scriptService, settings, lockService)
step.preExecute(logger, context).execute()
val updatedManagedIndexMetaData = step.getUpdatedManagedIndexMetadata(metadata)
assertEquals("Step status is not COMPLETED", Step.StepStatus.COMPLETED, updatedManagedIndexMetaData.stepMetaData?.stepStatus)
assertEquals("Did not get snapshot completed message", WaitForSnapshotStep.getSuccessMessage("test"), updatedManagedIndexMetaData.info!!["message"])
assertEquals("Step status is not FAILED", Step.StepStatus.FAILED, updatedManagedIndexMetaData.stepMetaData?.stepStatus)
assertEquals("Did not get snapshot failed message", WaitForSnapshotStep.getFailedExistsMessage("test"), updatedManagedIndexMetaData.info!!["message"])
}

whenever(snapshotStatus.state).doReturn(SnapshotsInProgress.State.ABORTED)
val snapshotInfo4: SnapshotInfo = mockSnapshotInfo(snapshot, SnapshotState.PARTIAL)
whenever(response.snapshots).doReturn(listOf(snapshotInfo4))
runBlocking {
val snapshotAction = SnapshotAction("repo", snapshot, 0)
val metadata = ManagedIndexMetaData("test", "indexUuid", "policy_id", null, null, null, null, null, null, null, ActionMetaData(WaitForSnapshotStep.name, 1, 0, false, 0, null, ActionProperties(snapshotName = "snapshot-name")), null, null, null)
Expand All @@ -124,7 +124,8 @@ class WaitForSnapshotStepTests : OpenSearchTestCase() {
assertEquals("Did not get snapshot failed message", WaitForSnapshotStep.getFailedExistsMessage("test"), updatedManagedIndexMetaData.info!!["message"])
}

whenever(snapshotStatus.state).doReturn(SnapshotsInProgress.State.FAILED)
val snapshotInfo5: SnapshotInfo = mockSnapshotInfo(snapshot, SnapshotState.INCOMPATIBLE)
whenever(response.snapshots).doReturn(listOf(snapshotInfo5))
runBlocking {
val snapshotAction = SnapshotAction("repo", snapshot, 0)
val metadata = ManagedIndexMetaData("test", "indexUuid", "policy_id", null, null, null, null, null, null, null, ActionMetaData(WaitForSnapshotStep.name, 1, 0, false, 0, null, ActionProperties(snapshotName = "snapshot-name")), null, null, null)
Expand All @@ -138,10 +139,9 @@ class WaitForSnapshotStepTests : OpenSearchTestCase() {
}

fun `test snapshot not in response list`() {
val snapshotStatus: SnapshotStatus = mock()
val response: SnapshotsStatusResponse = mock()
whenever(response.snapshots).doReturn(listOf(snapshotStatus))
whenever(snapshotStatus.snapshot).doReturn(Snapshot("repo", SnapshotId("snapshot-different-name", "some_uuid")))
val snapshotInfo: SnapshotInfo = mockSnapshotInfo("snapshot-different-name")
val response: GetSnapshotsResponse = mock()
whenever(response.snapshots).doReturn(listOf(snapshotInfo))
val client = getClient(getAdminClient(getClusterAdminClient(response, null)))

runBlocking {
Expand Down Expand Up @@ -190,17 +190,17 @@ class WaitForSnapshotStepTests : OpenSearchTestCase() {

private fun getAdminClient(clusterAdminClient: ClusterAdminClient): AdminClient = mock { on { cluster() } doReturn clusterAdminClient }

private fun getClusterAdminClient(snapshotsStatusResponse: SnapshotsStatusResponse?, exception: Exception?): ClusterAdminClient {
assertTrue("Must provide one and only one response or exception", (snapshotsStatusResponse != null).xor(exception != null))
private fun getClusterAdminClient(getSnapshotsResponse: GetSnapshotsResponse?, exception: Exception?): ClusterAdminClient {
assertTrue("Must provide one and only one response or exception", (getSnapshotsResponse != null).xor(exception != null))
return mock {
doAnswer { invocationOnMock ->
val listener = invocationOnMock.getArgument<ActionListener<SnapshotsStatusResponse>>(1)
if (snapshotsStatusResponse != null) {
listener.onResponse(snapshotsStatusResponse)
val listener = invocationOnMock.getArgument<ActionListener<GetSnapshotsResponse>>(1)
if (getSnapshotsResponse != null) {
listener.onResponse(getSnapshotsResponse)
} else {
listener.onFailure(exception)
}
}.whenever(this.mock).snapshotsStatus(any(), any())
}.whenever(this.mock).getSnapshots(any(), any())
}
}
}
25 changes: 21 additions & 4 deletions src/test/kotlin/org/opensearch/indexmanagement/snapshotmanagement/TestUtils.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule
import org.opensearch.snapshots.Snapshot
import org.opensearch.snapshots.SnapshotId
import org.opensearch.snapshots.SnapshotInfo
import org.opensearch.snapshots.SnapshotState
import org.opensearch.test.OpenSearchTestCase.randomAlphaOfLength
import org.opensearch.test.OpenSearchTestCase.randomBoolean
import org.opensearch.test.OpenSearchTestCase.randomIntBetween
Expand Down Expand Up @@ -177,11 +178,14 @@ fun randomSMState(): SMState = SMState.values()[randomIntBetween(0, SMState.valu

fun randomNotificationConfig(): NotificationConfig = NotificationConfig(randomChannel(), randomConditions())

fun randomConditions(): NotificationConfig.Conditions = NotificationConfig.Conditions(randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean())
fun randomConditions(): NotificationConfig.Conditions =
NotificationConfig.Conditions(randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean())

fun ToXContent.toJsonString(params: ToXContent.Params = ToXContent.EMPTY_PARAMS): String = this.toXContent(XContentFactory.jsonBuilder(), params).string()
fun ToXContent.toJsonString(params: ToXContent.Params = ToXContent.EMPTY_PARAMS): String =
this.toXContent(XContentFactory.jsonBuilder(), params).string()

fun ToXContent.toMap(params: ToXContent.Params = ToXContent.EMPTY_PARAMS): Map<String, Any> = this.toXContent(XContentFactory.jsonBuilder(), params).toMap()
fun ToXContent.toMap(params: ToXContent.Params = ToXContent.EMPTY_PARAMS): Map<String, Any> =
this.toXContent(XContentFactory.jsonBuilder(), params).toMap()

fun mockIndexResponse(status: RestStatus = RestStatus.OK): IndexResponse {
val indexResponse: IndexResponse = mock()
Expand Down Expand Up @@ -263,6 +267,18 @@ fun mockInProgressSnapshotInfo(
return SnapshotInfo(entry)
}

fun mockSnapshotInfo(
name: String = randomAlphaOfLength(10),
snapshotState: SnapshotState,
): SnapshotInfo {
return SnapshotInfo(
SnapshotId(name, UUIDs.randomBase64UUID()),
emptyList(),
emptyList(),
snapshotState,
)
}

fun mockGetSnapshotResponse(num: Int): GetSnapshotsResponse {
val getSnapshotsRes: GetSnapshotsResponse = mock()
whenever(getSnapshotsRes.snapshots).doReturn(mockSnapshotInfoList(num))
Expand All @@ -281,4 +297,5 @@ fun mockSnapshotInfoList(num: Int, namePrefix: String = randomAlphaOfLength(10))
return result.toList()
}

fun String.parser(): XContentParser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, this)
fun String.parser(): XContentParser =
XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, this)
Loading