From 8a310d2ee2a77a9c800bbb30ad586e6be0228461 Mon Sep 17 00:00:00 2001 From: Craig Perkins Date: Tue, 26 Nov 2024 12:59:18 -0500 Subject: [PATCH] Update snapshot management requests Signed-off-by: Craig Perkins --- .../api/transport/start/StartSMRequest.kt | 20 ++++++++++++------- .../transport/start/TransportStartSMAction.kt | 12 ++++++----- .../api/transport/stop/StopSMRequest.kt | 20 ++++++++++++------- .../transport/stop/TransportStopSMAction.kt | 12 ++++++----- .../snapshotmanagement/action/RequestTests.kt | 9 +++++---- 5 files changed, 45 insertions(+), 28 deletions(-) diff --git a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/StartSMRequest.kt b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/StartSMRequest.kt index 23e00e16b..b64e876fa 100644 --- a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/StartSMRequest.kt +++ b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/StartSMRequest.kt @@ -5,25 +5,31 @@ package org.opensearch.indexmanagement.snapshotmanagement.api.transport.start +import org.opensearch.action.ActionRequest import org.opensearch.action.ActionRequestValidationException -import org.opensearch.action.ValidateActions -import org.opensearch.action.update.UpdateRequest +import org.opensearch.action.ValidateActions.addValidationError import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import java.io.IOException -class StartSMRequest : UpdateRequest { +class StartSMRequest : ActionRequest { + + val id: String + get() = field + @Throws(IOException::class) - constructor(sin: StreamInput) : super(sin) + constructor(sin: StreamInput) : super(sin) { + this.id = sin.readString() + } constructor(id: String) { - super.id(id) + this.id = id } override fun validate(): ActionRequestValidationException? { var validationException: ActionRequestValidationException? = null - if (super.id().isEmpty()) { - validationException = ValidateActions.addValidationError("id is missing", validationException) + if (this.id.isEmpty()) { + validationException = addValidationError("id is missing", validationException) } return validationException } diff --git a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/TransportStartSMAction.kt b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/TransportStartSMAction.kt index ac5e58ab5..6e81bb6b7 100644 --- a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/TransportStartSMAction.kt +++ b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/start/TransportStartSMAction.kt @@ -11,6 +11,7 @@ import org.opensearch.OpenSearchStatusException import org.opensearch.action.DocWriteResponse import org.opensearch.action.support.ActionFilters import org.opensearch.action.support.master.AcknowledgedResponse +import org.opensearch.action.update.UpdateRequest import org.opensearch.action.update.UpdateResponse import org.opensearch.client.Client import org.opensearch.cluster.service.ClusterService @@ -57,7 +58,7 @@ constructor( user: User?, threadContext: ThreadContext.StoredContext, ): AcknowledgedResponse { - val smPolicy = client.getSMPolicy(request.id()) + val smPolicy = client.getSMPolicy(request.id) // Check if the requested user has permission on the resource, throwing an exception if the user does not verifyUserHasPermissionForResource(user, smPolicy.user, filterByEnabled, "snapshot management policy", smPolicy.policyName) @@ -71,7 +72,8 @@ constructor( private suspend fun enableSMPolicy(updateRequest: StartSMRequest): Boolean { val now = Instant.now().toEpochMilli() - updateRequest.index(INDEX_MANAGEMENT_INDEX).doc( + val updateReq = UpdateRequest(INDEX_MANAGEMENT_INDEX, updateRequest.id) + updateReq.doc( mapOf( SMPolicy.SM_TYPE to mapOf( @@ -83,12 +85,12 @@ constructor( ) val updateResponse: UpdateResponse = try { - client.suspendUntil { update(updateRequest, it) } + client.suspendUntil { update(updateReq, it) } } catch (e: VersionConflictEngineException) { - log.error("VersionConflictEngineException while trying to enable snapshot management policy id [${updateRequest.id()}]: $e") + log.error("VersionConflictEngineException while trying to enable snapshot management policy id [${updateRequest.id}]: $e") throw OpenSearchStatusException(conflictExceptionMessage, RestStatus.INTERNAL_SERVER_ERROR) } catch (e: Exception) { - log.error("Failed trying to enable snapshot management policy id [${updateRequest.id()}]: $e") + log.error("Failed trying to enable snapshot management policy id [${updateRequest.id}]: $e") throw OpenSearchStatusException("Failed while trying to enable SM Policy", RestStatus.INTERNAL_SERVER_ERROR) } return updateResponse.result == DocWriteResponse.Result.UPDATED diff --git a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/StopSMRequest.kt b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/StopSMRequest.kt index 4d70096b6..2d537b251 100644 --- a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/StopSMRequest.kt +++ b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/StopSMRequest.kt @@ -5,25 +5,31 @@ package org.opensearch.indexmanagement.snapshotmanagement.api.transport.stop +import org.opensearch.action.ActionRequest import org.opensearch.action.ActionRequestValidationException -import org.opensearch.action.ValidateActions -import org.opensearch.action.update.UpdateRequest +import org.opensearch.action.ValidateActions.addValidationError import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.core.common.io.stream.StreamOutput import java.io.IOException -class StopSMRequest : UpdateRequest { +class StopSMRequest : ActionRequest { + + val id: String + get() = field + @Throws(IOException::class) - constructor(sin: StreamInput) : super(sin) + constructor(sin: StreamInput) : super(sin) { + this.id = sin.readString() + } constructor(id: String) { - super.id(id) + this.id = id } override fun validate(): ActionRequestValidationException? { var validationException: ActionRequestValidationException? = null - if (super.id().isEmpty()) { - validationException = ValidateActions.addValidationError("id is missing", validationException) + if (this.id.isEmpty()) { + validationException = addValidationError("id is missing", validationException) } return validationException } diff --git a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/TransportStopSMAction.kt b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/TransportStopSMAction.kt index 4e4b0b7e6..6d45735d6 100644 --- a/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/TransportStopSMAction.kt +++ b/src/main/kotlin/org/opensearch/indexmanagement/snapshotmanagement/api/transport/stop/TransportStopSMAction.kt @@ -11,6 +11,7 @@ import org.opensearch.OpenSearchStatusException import org.opensearch.action.DocWriteResponse import org.opensearch.action.support.ActionFilters import org.opensearch.action.support.master.AcknowledgedResponse +import org.opensearch.action.update.UpdateRequest import org.opensearch.action.update.UpdateResponse import org.opensearch.client.Client import org.opensearch.cluster.service.ClusterService @@ -57,7 +58,7 @@ constructor( user: User?, threadContext: ThreadContext.StoredContext, ): AcknowledgedResponse { - val smPolicy = client.getSMPolicy(request.id()) + val smPolicy = client.getSMPolicy(request.id) // Check if the requested user has permission on the resource, throwing an exception if the user does not verifyUserHasPermissionForResource(user, smPolicy.user, filterByEnabled, "snapshot management policy", smPolicy.policyName) @@ -71,7 +72,8 @@ constructor( private suspend fun disableSMPolicy(updateRequest: StopSMRequest): Boolean { val now = Instant.now().toEpochMilli() - updateRequest.index(INDEX_MANAGEMENT_INDEX).doc( + val updateReq = UpdateRequest(INDEX_MANAGEMENT_INDEX, updateRequest.id) + updateReq.doc( mapOf( SMPolicy.SM_TYPE to mapOf( @@ -83,12 +85,12 @@ constructor( ) val updateResponse: UpdateResponse = try { - client.suspendUntil { update(updateRequest, it) } + client.suspendUntil { update(updateReq, it) } } catch (e: VersionConflictEngineException) { - log.error("VersionConflictEngineException while trying to disable snapshot management policy id [${updateRequest.id()}]: $e") + log.error("VersionConflictEngineException while trying to disable snapshot management policy id [${updateRequest.id}]: $e") throw OpenSearchStatusException(conflictExceptionMessage, RestStatus.INTERNAL_SERVER_ERROR) } catch (e: Exception) { - log.error("Failed trying to disable snapshot management policy id [${updateRequest.id()}]: $e") + log.error("Failed trying to disable snapshot management policy id [${updateRequest.id}]: $e") throw OpenSearchStatusException("Failed while trying to disable SM Policy", RestStatus.INTERNAL_SERVER_ERROR) } // TODO update metadata diff --git a/src/test/kotlin/org/opensearch/indexmanagement/snapshotmanagement/action/RequestTests.kt b/src/test/kotlin/org/opensearch/indexmanagement/snapshotmanagement/action/RequestTests.kt index 440aca087..9117ac0d5 100644 --- a/src/test/kotlin/org/opensearch/indexmanagement/snapshotmanagement/action/RequestTests.kt +++ b/src/test/kotlin/org/opensearch/indexmanagement/snapshotmanagement/action/RequestTests.kt @@ -7,6 +7,7 @@ package org.opensearch.indexmanagement.snapshotmanagement.action import org.opensearch.action.DocWriteRequest import org.opensearch.action.support.WriteRequest +import org.opensearch.action.update.UpdateRequest import org.opensearch.common.io.stream.BytesStreamOutput import org.opensearch.core.common.io.stream.StreamInput import org.opensearch.index.seqno.SequenceNumbers @@ -82,22 +83,22 @@ class RequestTests : OpenSearchTestCase() { fun `test start sm policy request`() { val id = "some_id" - val req = StartSMRequest(id).index(INDEX_MANAGEMENT_INDEX) + val req = UpdateRequest(INDEX_MANAGEMENT_INDEX, id) val out = BytesStreamOutput().apply { req.writeTo(this) } val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) val streamedReq = StartSMRequest(sin) - assertEquals(id, streamedReq.id()) + assertEquals(id, streamedReq.id) } fun `test stop sm policy request`() { val id = "some_id" - val req = StopSMRequest(id).index(INDEX_MANAGEMENT_INDEX) + val req = UpdateRequest(INDEX_MANAGEMENT_INDEX, id) val out = BytesStreamOutput().apply { req.writeTo(this) } val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes) val streamedReq = StopSMRequest(sin) - assertEquals(id, streamedReq.id()) + assertEquals(id, streamedReq.id) } fun `test explain sm policy request`() {