diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt
index 93c17ab14..87d62e395 100644
--- a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt
@@ -15,11 +15,15 @@ import org.opensearch.alerting.action.GetRemoteIndexesAction
 import org.opensearch.alerting.action.SearchEmailAccountAction
 import org.opensearch.alerting.action.SearchEmailGroupAction
 import org.opensearch.alerting.actionv2.DeleteMonitorV2Action
+import org.opensearch.alerting.actionv2.GetAlertsV2Action
 import org.opensearch.alerting.actionv2.GetMonitorV2Action
 import org.opensearch.alerting.actionv2.IndexMonitorV2Action
 import org.opensearch.alerting.actionv2.SearchMonitorV2Action
 import org.opensearch.alerting.alerts.AlertIndices
 import org.opensearch.alerting.alerts.AlertIndices.Companion.ALL_ALERT_INDEX_PATTERN
+import org.opensearch.alerting.alertsv2.AlertV2Indices
+import org.opensearch.alerting.alertsv2.AlertV2Indices.Companion.ALL_ALERT_V2_INDEX_PATTERN
+import org.opensearch.alerting.alertsv2.AlertV2Mover
 import org.opensearch.alerting.comments.CommentsIndices
 import org.opensearch.alerting.comments.CommentsIndices.Companion.ALL_COMMENTS_INDEX_PATTERN
 import org.opensearch.alerting.core.JobSweeper
@@ -29,6 +33,7 @@ import org.opensearch.alerting.core.action.node.ScheduledJobsStatsTransportActio
 import org.opensearch.alerting.core.lock.LockService
 import org.opensearch.alerting.core.resthandler.RestScheduledJobStatsHandler
 import org.opensearch.alerting.core.schedule.JobScheduler
+import org.opensearch.alerting.core.settings.AlertingV2Settings
 import org.opensearch.alerting.core.settings.LegacyOpenDistroScheduledJobSettings
 import org.opensearch.alerting.core.settings.ScheduledJobSettings
 import org.opensearch.alerting.modelv2.MonitorV2
@@ -57,6 +62,7 @@ import org.opensearch.alerting.resthandler.RestSearchEmailAccountAction
 import org.opensearch.alerting.resthandler.RestSearchEmailGroupAction
 import org.opensearch.alerting.resthandler.RestSearchMonitorAction
 import org.opensearch.alerting.resthandlerv2.RestDeleteMonitorV2Action
+import org.opensearch.alerting.resthandlerv2.RestGetAlertsV2Action
 import org.opensearch.alerting.resthandlerv2.RestGetMonitorV2Action
 import org.opensearch.alerting.resthandlerv2.RestIndexMonitorV2Action
 import org.opensearch.alerting.resthandlerv2.RestSearchMonitorV2Action
@@ -93,6 +99,7 @@ import org.opensearch.alerting.transport.TransportSearchEmailAccountAction
 import org.opensearch.alerting.transport.TransportSearchEmailGroupAction
 import org.opensearch.alerting.transport.TransportSearchMonitorAction
 import org.opensearch.alerting.transportv2.TransportDeleteMonitorV2Action
+import org.opensearch.alerting.transportv2.TransportGetAlertsV2Action
 import org.opensearch.alerting.transportv2.TransportGetMonitorV2Action
 import org.opensearch.alerting.transportv2.TransportIndexMonitorV2Action
 import org.opensearch.alerting.transportv2.TransportSearchMonitorV2Action
@@ -194,8 +201,10 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
     lateinit var docLevelMonitorQueries: DocLevelMonitorQueries
     lateinit var threadPool: ThreadPool
     lateinit var alertIndices: AlertIndices
+    lateinit var alertV2Indices: AlertV2Indices
     lateinit var clusterService: ClusterService
     lateinit var destinationMigrationCoordinator: DestinationMigrationCoordinator
+    lateinit var alertV2Mover: AlertV2Mover
     var monitorTypeToMonitorRunners: MutableMap<String, RemoteMonitorRegistry> = mutableMapOf()
 
     override fun getRestHandlers(
@@ -239,6 +248,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             RestDeleteMonitorV2Action(),
             RestGetMonitorV2Action(),
             RestSearchMonitorV2Action(settings, clusterService),
+            RestGetAlertsV2Action(),
         )
     }
 
@@ -278,6 +288,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             ActionPlugin.ActionHandler(GetMonitorV2Action.INSTANCE, TransportGetMonitorV2Action::class.java),
             ActionPlugin.ActionHandler(SearchMonitorV2Action.INSTANCE, TransportSearchMonitorV2Action::class.java),
             ActionPlugin.ActionHandler(DeleteMonitorV2Action.INSTANCE, TransportDeleteMonitorV2Action::class.java),
+            ActionPlugin.ActionHandler(GetAlertsV2Action.INSTANCE, TransportGetAlertsV2Action::class.java)
         )
     }
 
@@ -314,6 +325,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         val settings = environment.settings()
         val lockService = LockService(client, clusterService)
         alertIndices = AlertIndices(settings, client, threadPool, clusterService)
+        alertV2Indices = AlertV2Indices(settings, client, threadPool, clusterService)
         val alertService = AlertService(client, xContentRegistry, alertIndices)
         val triggerService = TriggerService(scriptService)
         runner = MonitorRunnerService
@@ -325,6 +337,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             .registerSettings(settings)
             .registerThreadPool(threadPool)
             .registerAlertIndices(alertIndices)
+            .registerAlertV2Indices(alertV2Indices)
             .registerInputService(
                 InputService(
                     client,
@@ -351,6 +364,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         scheduler = JobScheduler(threadPool, runner)
         sweeper = JobSweeper(environment.settings(), client, clusterService, threadPool, xContentRegistry, scheduler, ALERTING_JOB_TYPES)
         destinationMigrationCoordinator = DestinationMigrationCoordinator(client, clusterService, threadPool, scheduledJobIndices)
+        alertV2Mover = AlertV2Mover(environment.settings(), client, threadPool, clusterService, xContentRegistry)
         this.threadPool = threadPool
         this.clusterService = clusterService
 
@@ -378,6 +392,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             commentsIndices,
             docLevelMonitorQueries,
             destinationMigrationCoordinator,
+            alertV2Mover,
             lockService,
             alertService,
             triggerService
@@ -475,7 +490,8 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             AlertingSettings.ALERT_V2_QUERY_RESULTS_MAX_SIZE,
             AlertingSettings.ALERT_V2_PER_RESULT_TRIGGER_MAX_ALERTS,
             AlertingSettings.NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH,
-            AlertingSettings.NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH
+            AlertingSettings.NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH,
+            AlertingV2Settings.ALERTING_V2_ENABLED
         )
     }
 
@@ -494,6 +510,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
             SystemIndexDescriptor(ALL_ALERT_INDEX_PATTERN, "Alerting Plugin system index pattern"),
             SystemIndexDescriptor(SCHEDULED_JOBS_INDEX, "Alerting Plugin Configuration index"),
             SystemIndexDescriptor(ALL_COMMENTS_INDEX_PATTERN, "Alerting Comments system index pattern"),
+            SystemIndexDescriptor(ALL_ALERT_V2_INDEX_PATTERN, "Alerting V2 Alerts index pattern")
         )
     }
 
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerExecutionContext.kt b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerExecutionContext.kt
index a890ec1a6..5c5e24070 100644
--- a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerExecutionContext.kt
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerExecutionContext.kt
@@ -7,6 +7,7 @@ package org.opensearch.alerting
 
 import org.opensearch.action.bulk.BackoffPolicy
 import org.opensearch.alerting.alerts.AlertIndices
+import org.opensearch.alerting.alertsv2.AlertV2Indices
 import org.opensearch.alerting.core.lock.LockService
 import org.opensearch.alerting.model.destination.DestinationContextFactory
 import org.opensearch.alerting.remote.monitors.RemoteMonitorRegistry
@@ -35,6 +36,7 @@ data class MonitorRunnerExecutionContext(
     var settings: Settings? = null,
     var threadPool: ThreadPool? = null,
     var alertIndices: AlertIndices? = null,
+    var alertV2Indices: AlertV2Indices? = null,
     var inputService: InputService? = null,
     var triggerService: TriggerService? = null,
     var alertService: AlertService? = null,
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerService.kt b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerService.kt
index f8703aec2..6d8687623 100644
--- a/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerService.kt
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/MonitorRunnerService.kt
@@ -23,11 +23,14 @@ import org.opensearch.alerting.action.ExecuteWorkflowRequest
 import org.opensearch.alerting.action.ExecuteWorkflowResponse
 import org.opensearch.alerting.alerts.AlertIndices
 import org.opensearch.alerting.alerts.AlertMover.Companion.moveAlerts
+import org.opensearch.alerting.alertsv2.AlertV2Indices
+import org.opensearch.alerting.alertsv2.AlertV2Mover.Companion.moveAlertV2s
 import org.opensearch.alerting.core.JobRunner
 import org.opensearch.alerting.core.ScheduledJobIndices
 import org.opensearch.alerting.core.lock.LockModel
 import org.opensearch.alerting.core.lock.LockService
 import org.opensearch.alerting.model.destination.DestinationContextFactory
+import org.opensearch.alerting.modelv2.MonitorV2
 import org.opensearch.alerting.opensearchapi.retry
 import org.opensearch.alerting.opensearchapi.suspendUntil
 import org.opensearch.alerting.remote.monitors.RemoteDocumentLevelMonitorRunner
@@ -137,6 +140,11 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
         return this
     }
 
+    fun registerAlertV2Indices(alertV2Indices: AlertV2Indices): MonitorRunnerService {
+        this.monitorCtx.alertV2Indices = alertV2Indices
+        return this
+    }
+
     fun registerInputService(inputService: InputService): MonitorRunnerService {
         this.monitorCtx.inputService = inputService
         return this
@@ -316,6 +324,18 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
                     logger.error("Failed to move active alerts for monitor [${job.id}].", e)
                 }
             }
+        } else if (job is MonitorV2) {
+            launch {
+                try {
+                    monitorCtx.moveAlertsRetryPolicy!!.retry(logger) {
+                        if (monitorCtx.alertV2Indices!!.isAlertV2Initialized()) {
+                            moveAlertV2s(job.id, job, monitorCtx)
+                        }
+                    }
+                } catch (e: Exception) {
+                    logger.error("Failed to move active alertV2s for monitorV2 [${job.id}].", e)
+                }
+            }
         } else {
             throw IllegalArgumentException("Invalid job type")
         }
@@ -339,6 +359,15 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
             } catch (e: Exception) {
                 logger.error("Failed to move active alerts for monitor [$jobId].", e)
             }
+            try {
+                monitorCtx.moveAlertsRetryPolicy!!.retry(logger) {
+                    if (monitorCtx.alertV2Indices!!.isAlertV2Initialized()) {
+                        moveAlertV2s(jobId, null, monitorCtx)
+                    }
+                }
+            } catch (e: Exception) {
+                logger.error("Failed to move active alertV2s for monitorV2 [$jobId].", e)
+            }
         }
     }
 
@@ -433,20 +462,7 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
     ): MonitorRunResult<*> {
         // Updating the scheduled job index at the start of monitor execution runs for when there is an upgrade the the schema mapping
         // has not been updated.
-        if (!IndexUtils.scheduledJobIndexUpdated && monitorCtx.clusterService != null && monitorCtx.client != null) {
-            IndexUtils.updateIndexMapping(
-                ScheduledJob.SCHEDULED_JOBS_INDEX,
-                ScheduledJobIndices.scheduledJobMappings(), monitorCtx.clusterService!!.state(), monitorCtx.client!!.admin().indices(),
-                object : ActionListener<AcknowledgedResponse> {
-                    override fun onResponse(response: AcknowledgedResponse) {
-                    }
-
-                    override fun onFailure(t: Exception) {
-                        logger.error("Failed to update config index schema", t)
-                    }
-                }
-            )
-        }
+        updateAlertingConfigIndexSchema()
 
         if (job is Workflow) {
             logger.info("Executing scheduled workflow - id: ${job.id}, periodStart: $periodStart, periodEnd: $periodEnd, dryrun: $dryrun")
@@ -582,4 +598,21 @@ object MonitorRunnerService : JobRunner, CoroutineScope, AbstractLifecycleCompon
             .newInstance(template.params + mapOf("ctx" to ctx.asTemplateArg()))
             .execute()
     }
+
+    private fun updateAlertingConfigIndexSchema() {
+        if (!IndexUtils.scheduledJobIndexUpdated && monitorCtx.clusterService != null && monitorCtx.client != null) {
+            IndexUtils.updateIndexMapping(
+                ScheduledJob.SCHEDULED_JOBS_INDEX,
+                ScheduledJobIndices.scheduledJobMappings(), monitorCtx.clusterService!!.state(), monitorCtx.client!!.admin().indices(),
+                object : ActionListener<AcknowledgedResponse> {
+                    override fun onResponse(response: AcknowledgedResponse) {
+                    }
+
+                    override fun onFailure(t: Exception) {
+                        logger.error("Failed to update config index schema", t)
+                    }
+                }
+            )
+        }
+    }
 }
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Action.kt
new file mode 100644
index 000000000..e656d6a71
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Action.kt
@@ -0,0 +1,15 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.actionv2
+
+import org.opensearch.action.ActionType
+
+class GetAlertsV2Action private constructor() : ActionType<GetAlertsV2Response>(NAME, ::GetAlertsV2Response) {
+    companion object {
+        val INSTANCE = GetAlertsV2Action()
+        const val NAME = "cluster:admin/opensearch/alerting/v2/alerts/get"
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Request.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Request.kt
new file mode 100644
index 000000000..008057aa4
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Request.kt
@@ -0,0 +1,47 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.actionv2
+
+import org.opensearch.action.ActionRequest
+import org.opensearch.action.ActionRequestValidationException
+import org.opensearch.commons.alerting.model.Table
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import java.io.IOException
+
+class GetAlertsV2Request : ActionRequest {
+    val table: Table
+    val severityLevel: String
+    val monitorV2Ids: List<String>?
+
+    constructor(
+        table: Table,
+        severityLevel: String,
+        monitorV2Ids: List<String>? = null,
+    ) : super() {
+        this.table = table
+        this.severityLevel = severityLevel
+        this.monitorV2Ids = monitorV2Ids
+    }
+
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        table = Table.readFrom(sin),
+        severityLevel = sin.readString(),
+        monitorV2Ids = sin.readOptionalStringList(),
+    )
+
+    override fun validate(): ActionRequestValidationException? {
+        return null
+    }
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        table.writeTo(out)
+        out.writeString(severityLevel)
+        out.writeOptionalStringCollection(monitorV2Ids)
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Response.kt b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Response.kt
new file mode 100644
index 000000000..0de492496
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2Response.kt
@@ -0,0 +1,52 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.actionv2
+
+import org.opensearch.alerting.modelv2.AlertV2
+import org.opensearch.commons.notifications.action.BaseResponse
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.XContentBuilder
+import java.io.IOException
+import java.util.Collections
+
+class GetAlertsV2Response : BaseResponse {
+    val alertV2s: List<AlertV2>
+
+    // totalAlertV2s is not the same as the size of alertV2s because there can be 30 alerts from the request, but
+    // the request only asked for 5 alerts, so totalAlertV2s will be 30, but alertV2s will only contain 5 alerts
+    val totalAlertV2s: Int?
+
+    constructor(
+        alertV2s: List<AlertV2>,
+        totalAlertV2s: Int?
+    ) : super() {
+        this.alertV2s = alertV2s
+        this.totalAlertV2s = totalAlertV2s
+    }
+
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        alertV2s = Collections.unmodifiableList(sin.readList(::AlertV2)),
+        totalAlertV2s = sin.readOptionalInt()
+    )
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeCollection(alertV2s)
+        out.writeOptionalInt(totalAlertV2s)
+    }
+
+    @Throws(IOException::class)
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        builder.startObject()
+            .field("alerts_v2", alertV2s)
+            .field("total_alerts_v2", totalAlertV2s)
+
+        return builder.endObject()
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/alertsv2/AlertV2Indices.kt b/alerting/src/main/kotlin/org/opensearch/alerting/alertsv2/AlertV2Indices.kt
new file mode 100644
index 000000000..1c50cd94f
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/alertsv2/AlertV2Indices.kt
@@ -0,0 +1,413 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.alertsv2
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import org.apache.logging.log4j.LogManager
+import org.opensearch.ExceptionsHelper
+import org.opensearch.ResourceAlreadyExistsException
+import org.opensearch.action.admin.cluster.state.ClusterStateRequest
+import org.opensearch.action.admin.cluster.state.ClusterStateResponse
+import org.opensearch.action.admin.indices.alias.Alias
+import org.opensearch.action.admin.indices.create.CreateIndexRequest
+import org.opensearch.action.admin.indices.create.CreateIndexResponse
+import org.opensearch.action.admin.indices.delete.DeleteIndexRequest
+import org.opensearch.action.admin.indices.exists.indices.IndicesExistsRequest
+import org.opensearch.action.admin.indices.exists.indices.IndicesExistsResponse
+import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest
+import org.opensearch.action.admin.indices.rollover.RolloverRequest
+import org.opensearch.action.admin.indices.rollover.RolloverResponse
+import org.opensearch.action.support.IndicesOptions
+import org.opensearch.action.support.clustermanager.AcknowledgedResponse
+import org.opensearch.alerting.opensearchapi.suspendUntil
+import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_V2_HISTORY_ENABLED
+import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_V2_HISTORY_INDEX_MAX_AGE
+import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_V2_HISTORY_MAX_DOCS
+import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_V2_HISTORY_RETENTION_PERIOD
+import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_V2_HISTORY_ROLLOVER_PERIOD
+import org.opensearch.alerting.settings.AlertingSettings.Companion.REQUEST_TIMEOUT
+import org.opensearch.alerting.util.IndexUtils
+import org.opensearch.cluster.ClusterChangedEvent
+import org.opensearch.cluster.ClusterStateListener
+import org.opensearch.cluster.metadata.IndexMetadata
+import org.opensearch.cluster.service.ClusterService
+import org.opensearch.common.settings.Settings
+import org.opensearch.common.unit.TimeValue
+import org.opensearch.common.xcontent.XContentType
+import org.opensearch.commons.alerting.util.AlertingException
+import org.opensearch.core.action.ActionListener
+import org.opensearch.threadpool.Scheduler.Cancellable
+import org.opensearch.threadpool.ThreadPool
+import org.opensearch.transport.client.Client
+import java.time.Instant
+
+private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO)
+private val logger = LogManager.getLogger(AlertV2Indices::class.java)
+
+class AlertV2Indices(
+    settings: Settings,
+    private val client: Client,
+    private val threadPool: ThreadPool,
+    private val clusterService: ClusterService
+) : ClusterStateListener {
+
+    init {
+        clusterService.addListener(this)
+        clusterService.clusterSettings.addSettingsUpdateConsumer(ALERT_V2_HISTORY_ENABLED) { alertV2HistoryEnabled = it }
+        clusterService.clusterSettings.addSettingsUpdateConsumer(ALERT_V2_HISTORY_MAX_DOCS) { alertV2HistoryMaxDocs = it }
+        clusterService.clusterSettings.addSettingsUpdateConsumer(ALERT_V2_HISTORY_INDEX_MAX_AGE) { alertV2HistoryMaxAge = it }
+        clusterService.clusterSettings.addSettingsUpdateConsumer(ALERT_V2_HISTORY_ROLLOVER_PERIOD) {
+            alertV2HistoryRolloverPeriod = it
+            rescheduleAlertRollover()
+        }
+        clusterService.clusterSettings.addSettingsUpdateConsumer(ALERT_V2_HISTORY_RETENTION_PERIOD) {
+            alertV2HistoryRetentionPeriod = it
+        }
+        clusterService.clusterSettings.addSettingsUpdateConsumer(REQUEST_TIMEOUT) { requestTimeout = it }
+    }
+
+    companion object {
+
+        /** The in progress alert history index. */
+        const val ALERT_V2_INDEX = ".opensearch-alerting-v2-alerts"
+
+        /** The alias of the index in which to write alert history */
+        const val ALERT_V2_HISTORY_WRITE_INDEX = ".opensearch-alerting-v2-alert-history-write"
+
+        /** The index name pattern referring to all alert history indices */
+        const val ALERT_V2_HISTORY_ALL = ".opensearch-alerting-v2-alert-history*"
+
+        /** The index name pattern to create alert history indices */
+        const val ALERT_V2_HISTORY_INDEX_PATTERN = "<.opensearch-alerting-v2-alert-history-{now/d}-1>"
+
+        /** The index name pattern to query all alerts, history and current alerts. */
+        const val ALL_ALERT_V2_INDEX_PATTERN = ".opensearch-alerting-v2-alert*"
+
+        @JvmStatic
+        fun alertV2Mapping() =
+            AlertV2Indices::class.java.getResource("alert_v2_mapping.json").readText()
+    }
+
+    @Volatile private var alertV2HistoryEnabled = ALERT_V2_HISTORY_ENABLED.get(settings)
+
+    @Volatile private var alertV2HistoryMaxDocs = ALERT_V2_HISTORY_MAX_DOCS.get(settings)
+
+    @Volatile private var alertV2HistoryMaxAge = ALERT_V2_HISTORY_INDEX_MAX_AGE.get(settings)
+
+    @Volatile private var alertV2HistoryRolloverPeriod = ALERT_V2_HISTORY_ROLLOVER_PERIOD.get(settings)
+
+    @Volatile private var alertV2HistoryRetentionPeriod = ALERT_V2_HISTORY_RETENTION_PERIOD.get(settings)
+
+    @Volatile private var requestTimeout = REQUEST_TIMEOUT.get(settings)
+
+    @Volatile private var isClusterManager = false
+
+    // for JobsMonitor to report
+    var lastRolloverTime: TimeValue? = null
+
+    private var alertV2HistoryIndexInitialized: Boolean = false
+
+    private var alertV2IndexInitialized: Boolean = false
+
+    private var scheduledAlertV2Rollover: Cancellable? = null
+
+    fun onClusterManager() {
+        try {
+            // try to rollover immediately as we might be restarting the cluster
+            rolloverAlertV2HistoryIndex()
+
+            // schedule the next rollover for approx MAX_AGE later
+            scheduledAlertV2Rollover = threadPool
+                .scheduleWithFixedDelay({ rolloverAndDeleteAlertV2HistoryIndices() }, alertV2HistoryRolloverPeriod, executorName())
+        } catch (e: Exception) {
+            logger.error("Error rolling over alerts v2 history index.", e)
+        }
+    }
+
+    fun offClusterManager() {
+        scheduledAlertV2Rollover?.cancel()
+    }
+
+    private fun executorName(): String {
+        return ThreadPool.Names.MANAGEMENT
+    }
+
+    override fun clusterChanged(event: ClusterChangedEvent) {
+        // Instead of using a LocalNodeClusterManagerListener to track clustermanager changes, this service will
+        // track them here to avoid conditions where clustermanager listener events run after other
+        // listeners that depend on what happened in the clustermanager listener
+        if (this.isClusterManager != event.localNodeClusterManager()) {
+            this.isClusterManager = event.localNodeClusterManager()
+            if (this.isClusterManager) {
+                onClusterManager()
+            } else {
+                offClusterManager()
+            }
+        }
+
+        // if the indexes have been deleted they need to be reinitialized
+        alertV2IndexInitialized = event.state().routingTable().hasIndex(ALERT_V2_INDEX)
+        alertV2HistoryIndexInitialized = event.state().metadata().hasAlias(ALERT_V2_HISTORY_WRITE_INDEX)
+    }
+
+    private fun rescheduleAlertRollover() {
+        if (clusterService.state().nodes.isLocalNodeElectedClusterManager) {
+            scheduledAlertV2Rollover?.cancel()
+            scheduledAlertV2Rollover = threadPool
+                .scheduleWithFixedDelay({ rolloverAndDeleteAlertV2HistoryIndices() }, alertV2HistoryRolloverPeriod, executorName())
+        }
+    }
+
+    suspend fun createOrUpdateAlertV2Index() {
+        if (!alertV2IndexInitialized) {
+            alertV2IndexInitialized = createIndex(ALERT_V2_INDEX, alertV2Mapping())
+            if (alertV2IndexInitialized) IndexUtils.alertIndexUpdated()
+        } else {
+            if (!IndexUtils.alertIndexUpdated) updateIndexMapping(ALERT_V2_INDEX, alertV2Mapping())
+        }
+        alertV2IndexInitialized
+    }
+
+    suspend fun createOrUpdateInitialAlertV2HistoryIndex() {
+        if (!alertV2HistoryIndexInitialized) {
+            alertV2HistoryIndexInitialized = createIndex(ALERT_V2_HISTORY_INDEX_PATTERN, alertV2Mapping(), ALERT_V2_HISTORY_WRITE_INDEX)
+            if (alertV2HistoryIndexInitialized)
+                IndexUtils.lastUpdatedAlertV2HistoryIndex = IndexUtils.getIndexNameWithAlias(
+                    clusterService.state(),
+                    ALERT_V2_HISTORY_WRITE_INDEX
+                )
+        } else {
+            updateIndexMapping(ALERT_V2_HISTORY_WRITE_INDEX, alertV2Mapping(), true)
+        }
+        alertV2HistoryIndexInitialized
+    }
+
+    fun isAlertV2Initialized(): Boolean {
+        return alertV2IndexInitialized && alertV2HistoryIndexInitialized
+    }
+
+    private fun rolloverAndDeleteAlertV2HistoryIndices() {
+        if (alertV2HistoryEnabled) rolloverAlertV2HistoryIndex()
+        deleteOldIndices("History", ALERT_V2_HISTORY_ALL)
+    }
+
+    private suspend fun createIndex(index: String, schemaMapping: String, alias: String? = null): Boolean {
+        // This should be a fast check of local cluster state. Should be exceedingly rare that the local cluster
+        // state does not contain the index and multiple nodes concurrently try to create the index.
+        // If it does happen that error is handled we catch the ResourceAlreadyExistsException
+        val existsResponse: IndicesExistsResponse = client.admin().indices().suspendUntil {
+            exists(IndicesExistsRequest(index).local(true), it)
+        }
+        if (existsResponse.isExists) return true
+
+        logger.debug("index: [$index] schema mappings: [$schemaMapping]")
+        val request = CreateIndexRequest(index)
+            .mapping(schemaMapping)
+            .settings(Settings.builder().put("index.hidden", true).build())
+
+        if (alias != null) request.alias(Alias(alias))
+        return try {
+            val createIndexResponse: CreateIndexResponse = client.admin().indices().suspendUntil { create(request, it) }
+            createIndexResponse.isAcknowledged
+        } catch (t: Exception) {
+            if (ExceptionsHelper.unwrapCause(t) is ResourceAlreadyExistsException) {
+                true
+            } else {
+                throw AlertingException.wrap(t)
+            }
+        }
+    }
+
+    private suspend fun updateIndexMapping(index: String, mapping: String, alias: Boolean = false) {
+        val clusterState = clusterService.state()
+        var targetIndex = index
+        if (alias) {
+            targetIndex = IndexUtils.getIndexNameWithAlias(clusterState, index)
+        }
+
+        if (targetIndex == IndexUtils.lastUpdatedAlertV2HistoryIndex) {
+            return
+        }
+
+        val putMappingRequest: PutMappingRequest = PutMappingRequest(targetIndex)
+            .source(mapping, XContentType.JSON)
+        val updateResponse: AcknowledgedResponse = client.admin().indices().suspendUntil { putMapping(putMappingRequest, it) }
+        if (updateResponse.isAcknowledged) {
+            logger.info("Index mapping of $targetIndex is updated")
+            setIndexUpdateFlag(index, targetIndex)
+        } else {
+            logger.info("Failed to update index mapping of $targetIndex")
+        }
+    }
+
+    private fun setIndexUpdateFlag(index: String, targetIndex: String) {
+        when (index) {
+            ALERT_V2_INDEX -> IndexUtils.alertV2IndexUpdated()
+            ALERT_V2_HISTORY_WRITE_INDEX -> IndexUtils.lastUpdatedAlertV2HistoryIndex = targetIndex
+        }
+    }
+
+    private fun rolloverIndex(
+        initialized: Boolean,
+        index: String,
+        pattern: String,
+        map: String,
+        docsCondition: Long,
+        ageCondition: TimeValue,
+        writeIndex: String
+    ) {
+        if (!initialized) {
+            return
+        }
+
+        // We have to pass null for newIndexName in order to get Elastic to increment the index count.
+        val request = RolloverRequest(index, null)
+        request.createIndexRequest.index(pattern)
+            .mapping(map)
+            .settings(Settings.builder().put("index.hidden", true).build())
+        request.addMaxIndexDocsCondition(docsCondition)
+        request.addMaxIndexAgeCondition(ageCondition)
+        client.admin().indices().rolloverIndex(
+            request,
+            object : ActionListener<RolloverResponse> {
+                override fun onResponse(response: RolloverResponse) {
+                    if (!response.isRolledOver) {
+                        logger.info("$writeIndex not rolled over. Conditions were: ${response.conditionStatus}")
+                    } else {
+                        lastRolloverTime = TimeValue.timeValueMillis(threadPool.absoluteTimeInMillis())
+                    }
+                }
+                override fun onFailure(e: Exception) {
+                    logger.error("$writeIndex not roll over failed.")
+                }
+            }
+        )
+    }
+
+    private fun rolloverAlertV2HistoryIndex() {
+        rolloverIndex(
+            alertV2HistoryIndexInitialized,
+            ALERT_V2_HISTORY_WRITE_INDEX,
+            ALERT_V2_HISTORY_INDEX_PATTERN,
+            alertV2Mapping(),
+            alertV2HistoryMaxDocs,
+            alertV2HistoryMaxAge,
+            ALERT_V2_HISTORY_WRITE_INDEX
+        )
+    }
+
+    private fun deleteOldIndices(tag: String, indices: String) {
+        val clusterStateRequest = ClusterStateRequest()
+            .clear()
+            .indices(indices)
+            .metadata(true)
+            .local(true)
+            .indicesOptions(IndicesOptions.strictExpand())
+        client.admin().cluster().state(
+            clusterStateRequest,
+            object : ActionListener<ClusterStateResponse> {
+                override fun onResponse(clusterStateResponse: ClusterStateResponse) {
+                    if (clusterStateResponse.state.metadata.indices.isNotEmpty()) {
+                        scope.launch {
+                            val indicesToDelete = getIndicesToDelete(clusterStateResponse)
+                            logger.info("Deleting old $tag indices viz $indicesToDelete")
+                            deleteAllOldHistoryIndices(indicesToDelete)
+                        }
+                    } else {
+                        logger.info("No Old $tag Indices to delete")
+                    }
+                }
+                override fun onFailure(e: Exception) {
+                    logger.error("Error fetching cluster state")
+                }
+            }
+        )
+    }
+
+    private fun getIndicesToDelete(clusterStateResponse: ClusterStateResponse): List<String> {
+        val indicesToDelete = mutableListOf<String>()
+        for (entry in clusterStateResponse.state.metadata.indices) {
+            val indexMetaData = entry.value
+            getHistoryIndexToDelete(
+                indexMetaData,
+                alertV2HistoryRetentionPeriod.millis,
+                ALERT_V2_HISTORY_WRITE_INDEX,
+                alertV2HistoryEnabled
+            )?.let { indicesToDelete.add(it) }
+        }
+        return indicesToDelete
+    }
+
+    private fun getHistoryIndexToDelete(
+        indexMetadata: IndexMetadata,
+        retentionPeriodMillis: Long,
+        writeIndex: String,
+        historyEnabled: Boolean
+    ): String? {
+        val creationTime = indexMetadata.creationDate
+        if ((Instant.now().toEpochMilli() - creationTime) > retentionPeriodMillis) {
+            val alias = indexMetadata.aliases.entries.firstOrNull { writeIndex == it.value.alias }
+            if (alias != null) {
+                if (historyEnabled) {
+                    // If the index has the write alias and history is enabled, don't delete the index
+                    return null
+                } else if (writeIndex == ALERT_V2_HISTORY_WRITE_INDEX) {
+                    // Otherwise reset alertHistoryIndexInitialized since index will be deleted
+                    alertV2HistoryIndexInitialized = false
+                }
+            }
+
+            return indexMetadata.index.name
+        }
+        return null
+    }
+
+    private fun deleteAllOldHistoryIndices(indicesToDelete: List<String>) {
+        if (indicesToDelete.isNotEmpty()) {
+            val deleteIndexRequest = DeleteIndexRequest(*indicesToDelete.toTypedArray())
+            client.admin().indices().delete(
+                deleteIndexRequest,
+                object : ActionListener<AcknowledgedResponse> {
+                    override fun onResponse(deleteIndicesResponse: AcknowledgedResponse) {
+                        if (!deleteIndicesResponse.isAcknowledged) {
+                            logger.error(
+                                "Could not delete one or more Alerting V2 history indices: $indicesToDelete. Retrying one by one."
+                            )
+                            deleteOldHistoryIndex(indicesToDelete)
+                        }
+                    }
+                    override fun onFailure(e: Exception) {
+                        logger.error("Delete for Alerting V2 History Indices $indicesToDelete Failed. Retrying one by one.")
+                        deleteOldHistoryIndex(indicesToDelete)
+                    }
+                }
+            )
+        }
+    }
+
+    private fun deleteOldHistoryIndex(indicesToDelete: List<String>) {
+        for (index in indicesToDelete) {
+            val singleDeleteRequest = DeleteIndexRequest(*indicesToDelete.toTypedArray())
+            client.admin().indices().delete(
+                singleDeleteRequest,
+                object : ActionListener<AcknowledgedResponse> {
+                    override fun onResponse(acknowledgedResponse: AcknowledgedResponse?) {
+                        if (acknowledgedResponse != null) {
+                            if (!acknowledgedResponse.isAcknowledged) {
+                                logger.error("Could not delete one or more Alerting V2 history indices: $index")
+                            }
+                        }
+                    }
+                    override fun onFailure(e: Exception) {
+                        logger.error("Exception ${e.message} while deleting the index $index")
+                    }
+                }
+            )
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/alertsv2/AlertV2Mover.kt b/alerting/src/main/kotlin/org/opensearch/alerting/alertsv2/AlertV2Mover.kt
new file mode 100644
index 000000000..df297a9b6
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/alertsv2/AlertV2Mover.kt
@@ -0,0 +1,458 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.alertsv2
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import org.apache.logging.log4j.LogManager
+import org.opensearch.action.bulk.BulkRequest
+import org.opensearch.action.bulk.BulkResponse
+import org.opensearch.action.delete.DeleteRequest
+import org.opensearch.action.index.IndexRequest
+import org.opensearch.action.search.SearchRequest
+import org.opensearch.action.search.SearchResponse
+import org.opensearch.alerting.MonitorRunnerExecutionContext
+import org.opensearch.alerting.alertsv2.AlertV2Indices.Companion.ALERT_V2_HISTORY_WRITE_INDEX
+import org.opensearch.alerting.alertsv2.AlertV2Indices.Companion.ALERT_V2_INDEX
+import org.opensearch.alerting.modelv2.AlertV2
+import org.opensearch.alerting.modelv2.AlertV2.Companion.TRIGGERED_TIME_FIELD
+import org.opensearch.alerting.modelv2.AlertV2.Companion.TRIGGER_V2_ID_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.MONITOR_V2_TYPE
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.TRIGGERS_FIELD
+import org.opensearch.alerting.modelv2.PPLSQLMonitor.Companion.PPL_SQL_MONITOR_TYPE
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.EXPIRE_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.ID_FIELD
+import org.opensearch.alerting.opensearchapi.suspendUntil
+import org.opensearch.alerting.settings.AlertingSettings.Companion.ALERT_V2_HISTORY_ENABLED
+import org.opensearch.alerting.util.MAX_SEARCH_SIZE
+import org.opensearch.cluster.ClusterChangedEvent
+import org.opensearch.cluster.ClusterStateListener
+import org.opensearch.cluster.service.ClusterService
+import org.opensearch.common.settings.Settings
+import org.opensearch.common.unit.TimeValue
+import org.opensearch.common.xcontent.LoggingDeprecationHandler
+import org.opensearch.common.xcontent.XContentFactory
+import org.opensearch.common.xcontent.XContentHelper
+import org.opensearch.common.xcontent.XContentType
+import org.opensearch.commons.alerting.model.ScheduledJob.Companion.SCHEDULED_JOBS_INDEX
+import org.opensearch.core.common.bytes.BytesReference
+import org.opensearch.core.rest.RestStatus
+import org.opensearch.core.xcontent.NamedXContentRegistry
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.XContentParser
+import org.opensearch.index.VersionType
+import org.opensearch.index.query.QueryBuilders
+import org.opensearch.search.builder.SearchSourceBuilder
+import org.opensearch.threadpool.Scheduler
+import org.opensearch.threadpool.ThreadPool
+import org.opensearch.transport.client.Client
+import java.time.Instant
+import java.util.concurrent.TimeUnit
+
+private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO)
+private val logger = LogManager.getLogger(AlertV2Mover::class.java)
+
+class AlertV2Mover(
+    settings: Settings,
+    private val client: Client,
+    private val threadPool: ThreadPool,
+    private val clusterService: ClusterService,
+    private val xContentRegistry: NamedXContentRegistry,
+) : ClusterStateListener {
+    init {
+        clusterService.addListener(this)
+        clusterService.clusterSettings.addSettingsUpdateConsumer(ALERT_V2_HISTORY_ENABLED) { alertV2HistoryEnabled = it }
+    }
+
+    @Volatile private var isClusterManager = false
+
+    private var alertV2IndexInitialized = false
+
+    private var alertV2HistoryIndexInitialized = false
+
+    private var alertV2HistoryEnabled = ALERT_V2_HISTORY_ENABLED.get(settings)
+
+    private var scheduledAlertsV2CheckAndExpire: Scheduler.Cancellable? = null
+
+    private val executorName = ThreadPool.Names.MANAGEMENT
+
+    private val checkForExpirationInterval = TimeValue(1L, TimeUnit.MINUTES)
+
+    override fun clusterChanged(event: ClusterChangedEvent) {
+        if (this.isClusterManager != event.localNodeClusterManager()) {
+            this.isClusterManager = event.localNodeClusterManager()
+            if (this.isClusterManager) {
+                onManager()
+            } else {
+                offManager()
+            }
+        }
+
+        alertV2IndexInitialized = event.state().routingTable().hasIndex(ALERT_V2_INDEX)
+        alertV2HistoryIndexInitialized = event.state().metadata().hasAlias(ALERT_V2_HISTORY_WRITE_INDEX)
+    }
+
+    fun onManager() {
+        try {
+            // try to sweep current AlertV2s for expiration immediately as we might be restarting the cluster
+            moveOrDeleteAlertV2s()
+            // schedule expiration checks and expirations to happen repeatedly at some interval
+            scheduledAlertsV2CheckAndExpire = threadPool
+                .scheduleWithFixedDelay({ moveOrDeleteAlertV2s() }, checkForExpirationInterval, executorName)
+        } catch (e: Exception) {
+            // This should be run on cluster startup
+            logger.error(
+                "Error sweeping AlertV2s for expiration. This cannot be done until clustermanager node is restarted.",
+                e
+            )
+        }
+    }
+
+    fun offManager() {
+        scheduledAlertsV2CheckAndExpire?.cancel()
+    }
+
+    // if alertV2 history is enabled, move expired alerts to alertV2 history indices
+    // if alertV2 history is disabled, permanently delete expired alerts
+    private fun moveOrDeleteAlertV2s() {
+        if (!areAlertV2IndicesPresent()) {
+            return
+        }
+
+        scope.launch {
+            val expiredAlerts = searchForExpiredAlerts()
+
+            var copyResponse: BulkResponse? = null
+            val deleteResponse: BulkResponse?
+            if (!alertV2HistoryEnabled) {
+                deleteResponse = deleteExpiredAlerts(expiredAlerts)
+            } else {
+                copyResponse = copyExpiredAlerts(expiredAlerts)
+                deleteResponse = deleteExpiredAlertsThatWereCopied(copyResponse, expiredAlerts)
+            }
+            checkForFailures(copyResponse)
+            checkForFailures(deleteResponse)
+        }
+    }
+
+    private suspend fun searchForExpiredAlerts(): List<AlertV2> {
+        /* first collect all triggers and their expire durations */
+        // when searching the alerting-config index, only trigger IDs and their expire durations are needed
+        val monitorV2sSearchQuery = SearchSourceBuilder.searchSource()
+            .query(QueryBuilders.existsQuery(MONITOR_V2_TYPE))
+            .fetchSource(
+                arrayOf(
+                    "$MONITOR_V2_TYPE.$PPL_SQL_MONITOR_TYPE.$TRIGGERS_FIELD.$ID_FIELD",
+                    "$MONITOR_V2_TYPE.$PPL_SQL_MONITOR_TYPE.$TRIGGERS_FIELD.$EXPIRE_FIELD"
+                ),
+                null
+            )
+            .size(MAX_SEARCH_SIZE)
+            .version(true)
+        val monitorV2sRequest = SearchRequest(SCHEDULED_JOBS_INDEX)
+            .source(monitorV2sSearchQuery)
+        val searchMonitorV2sResponse: SearchResponse = client.suspendUntil { search(monitorV2sRequest, it) }
+
+        // construct a map that stores each trigger's expiration time
+        // TODO: create XContent parser specifically for responses to the above search to avoid casting
+        val triggerToExpireDuration = mutableMapOf<String, Long>()
+        searchMonitorV2sResponse.hits.forEach { hit ->
+            val monitorV2Obj = hit.sourceAsMap[MONITOR_V2_TYPE] as Map<String, Any>
+            val pplMonitorObj = monitorV2Obj[PPL_SQL_MONITOR_TYPE] as Map<String, Any>
+            val triggers = pplMonitorObj[TRIGGERS_FIELD] as List<Map<String, Any>>
+            triggers.forEach { trigger ->
+                val triggerId = trigger[ID_FIELD] as String
+                val expireDuration = (trigger[EXPIRE_FIELD] as Int).toLong()
+                triggerToExpireDuration[triggerId] = expireDuration
+            }
+        }
+
+        /* now collect all expired alerts */
+        val now = Instant.now().toEpochMilli()
+
+        val expiredAlertsBoolQuery = QueryBuilders.boolQuery()
+
+        // collect, in an overarching should clause, each trigger and its expiration time.
+        // any alert that matches both the trigger ID and the expiration time check should
+        // be returned by the search query
+        triggerToExpireDuration.forEach { (triggerId, expireDuration) ->
+            val expireDurationMillis = expireDuration * 60 * 1000
+            val maxValidTime = now - expireDurationMillis
+
+            expiredAlertsBoolQuery.should(
+                QueryBuilders.boolQuery()
+                    .must(QueryBuilders.termQuery(TRIGGER_V2_ID_FIELD, triggerId))
+                    .must(QueryBuilders.rangeQuery(TRIGGERED_TIME_FIELD).lte(maxValidTime))
+            )
+        }
+
+        // add orphaned alerts to should clause as well (i.e. alerts whose trigger IDs cannot
+        // be found in the list of currently existent triggers), since orphaned alerts should be expired.
+        // note: this is a redundancy with MonitorRunnerService's
+        // postIndex and postDelete, which handles moving alerts in response
+        // to a monitor update or delete event. this cleanly handles the case
+        // that even with those measures in place, an alert that came from a
+        // now nonexistent trigger was somehow found
+        expiredAlertsBoolQuery.should(
+            QueryBuilders.boolQuery()
+                .mustNot(QueryBuilders.termsQuery(TRIGGER_V2_ID_FIELD, triggerToExpireDuration.keys.toList()))
+        )
+
+        // Explicitly specify that at least one should clause must match
+        expiredAlertsBoolQuery.minimumShouldMatch(1)
+
+        // only alerts' monitor IDs should be fetched, the ID of the alert
+        // itself will be the document ID, which is part of the doc's metadata,
+        // not the doc's source, so it doesn't need to be fetched in the query
+        val expiredAlertsSearchQuery = SearchSourceBuilder.searchSource()
+            .query(expiredAlertsBoolQuery)
+            .size(MAX_SEARCH_SIZE)
+            .version(true)
+        val expiredAlertsRequest = SearchRequest(ALERT_V2_INDEX)
+            .source(expiredAlertsSearchQuery)
+        val expiredAlertsResponse: SearchResponse = client.suspendUntil { search(expiredAlertsRequest, it) }
+
+        val expiredAlertV2s = mutableListOf<AlertV2>()
+        expiredAlertsResponse.hits.forEach { hit ->
+            expiredAlertV2s.add(
+                AlertV2.parse(alertV2ContentParser(hit.sourceRef), hit.id, hit.version)
+            )
+        }
+
+        return expiredAlertV2s
+    }
+
+    private suspend fun deleteExpiredAlerts(expiredAlerts: List<AlertV2>): BulkResponse? {
+        // If no expired alerts are found, simply return
+        if (expiredAlerts.isEmpty()) {
+            return null
+        }
+
+        val deleteRequests = expiredAlerts.map {
+            DeleteRequest(ALERT_V2_INDEX, it.id)
+                .routing(it.monitorId)
+                .version(it.version)
+                .versionType(VersionType.EXTERNAL_GTE)
+        }
+
+        val deleteRequest = BulkRequest().add(deleteRequests)
+        val deleteResponse: BulkResponse = client.suspendUntil { bulk(deleteRequest, it) }
+
+        return deleteResponse
+    }
+
+    private suspend fun copyExpiredAlerts(expiredAlerts: List<AlertV2>): BulkResponse? {
+        // If no expired alerts are found, simply return
+        if (expiredAlerts.isEmpty()) {
+            return null
+        }
+
+        val indexRequests = expiredAlerts.map {
+            IndexRequest(ALERT_V2_HISTORY_WRITE_INDEX)
+                .routing(it.monitorId)
+                .source(it.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS))
+                .version(it.version)
+                .versionType(VersionType.EXTERNAL_GTE)
+                .id(it.id)
+        }
+
+        val copyRequest = BulkRequest().add(indexRequests)
+        val copyResponse: BulkResponse = client.suspendUntil { bulk(copyRequest, it) }
+
+        return copyResponse
+    }
+
+    private suspend fun deleteExpiredAlertsThatWereCopied(copyResponse: BulkResponse?, expiredAlerts: List<AlertV2>): BulkResponse? {
+        // if there were no expired alerts to copy, skip deleting anything
+        if (copyResponse == null) {
+            return null
+        }
+
+        // pre-index the alerts so retrieving their
+        // monitor IDs for routing is easier
+        val alertsById: Map<String, AlertV2> = expiredAlerts.associateBy { it.id }
+
+        val deleteRequests = copyResponse.items.filterNot { it.isFailed }.map {
+            DeleteRequest(ALERT_V2_INDEX, it.id)
+                .routing(alertsById[it.id]!!.monitorId)
+                .version(it.version)
+                .versionType(VersionType.EXTERNAL_GTE)
+        }
+        val deleteRequest = BulkRequest().add(deleteRequests)
+        val deleteResponse: BulkResponse = client.suspendUntil { bulk(deleteRequest, it) }
+
+        return deleteResponse
+    }
+
+    private fun checkForFailures(bulkResponse: BulkResponse?) {
+        bulkResponse?.let {
+            if (bulkResponse.hasFailures()) {
+                val retryCause = bulkResponse.items.filter { it.isFailed }
+                    .firstOrNull { it.status() == RestStatus.TOO_MANY_REQUESTS }
+                    ?.failure?.cause
+                logger.error(
+                    "Failed to move or delete alert v2s: ${bulkResponse.buildFailureMessage()}",
+                    retryCause
+                )
+            }
+        }
+    }
+
+    private fun alertV2ContentParser(bytesReference: BytesReference): XContentParser {
+        return XContentHelper.createParser(
+            NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE,
+            bytesReference, XContentType.JSON
+        )
+    }
+    private fun scheduledJobContentParser(bytesReference: BytesReference): XContentParser {
+        return XContentHelper.createParser(
+            xContentRegistry, LoggingDeprecationHandler.INSTANCE,
+            bytesReference, XContentType.JSON
+        )
+    }
+
+    private fun areAlertV2IndicesPresent(): Boolean {
+        return alertV2IndexInitialized && alertV2HistoryIndexInitialized
+    }
+
+    companion object {
+        // this method is used by MonitorRunnerService's postIndex and postDelete
+        // functions to move (in the case of alert v2 history enabled) or delete
+        // (in the case of alert v2 history disabled) the alerts generated by
+        // a monitor in response to the event that the monitor gets updated
+        // or deleted
+        suspend fun moveAlertV2s(monitorV2Id: String, monitorV2: MonitorV2?, monitorCtx: MonitorRunnerExecutionContext) {
+            val client = monitorCtx.client!!
+
+            // first collect all alerts that came from this updated or deleted monitor
+            val boolQuery = QueryBuilders.boolQuery()
+                .filter(QueryBuilders.termQuery(AlertV2.MONITOR_V2_ID_FIELD, monitorV2Id))
+
+            /*
+             this monitorV2 != null case happens when this function is called by postIndex. if the monitor is updated,
+             we don't want to expire alerts that were generated by triggers that still exist
+             in the updated monitor, so filter those out. only expire alerts from triggers in
+             this monitor that may no longer exist in the updated version of the monitor.
+             edge case: user can edit the trigger itself while explicitly keeping the ID the same,
+             which means alerts generated by that trigger will (incorrectly) not be filtered out by this logic
+             even though it was edited. to mitigate this, recall that callers of the update monitor API
+             must supply the full MonitorV2 object of the updated monitor config. this is important
+             because it means they don't have to reference the triggers by ID when updating the triggers,
+             they simply declare a whole new monitor with whatever new triggers they want it to have, and when doing this,
+             likely won't explicitly pass in trigger IDs for their updated triggers that exactly match
+             the IDs of the old triggers. this means Alerting will generate a new ID for the updated triggers by default,
+             meaning this logic will pick up those updated triggers and correctly move/delete the alerts
+            */
+            if (monitorV2 != null) {
+                boolQuery.mustNot(QueryBuilders.termsQuery(TRIGGER_V2_ID_FIELD, monitorV2.triggers.map { it.id }))
+            }
+
+            val alertsSearchQuery = SearchSourceBuilder.searchSource()
+                .query(boolQuery)
+                .size(MAX_SEARCH_SIZE)
+                .version(true)
+            val activeAlertsRequest = SearchRequest(ALERT_V2_INDEX)
+                .source(alertsSearchQuery)
+            val searchAlertsResponse: SearchResponse = client.suspendUntil { search(activeAlertsRequest, it) }
+
+            // If no alerts are found, simply return
+            if (searchAlertsResponse.hits.totalHits?.value == 0L) return
+
+            val activeAlerts = mutableListOf<AlertV2>()
+            searchAlertsResponse.hits.forEach { hit ->
+                activeAlerts.add(
+                    AlertV2.parse(
+                        XContentHelper.createParser(
+                            NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE,
+                            hit.sourceRef, XContentType.JSON
+                        ),
+                        hit.id,
+                        hit.version
+                    )
+                )
+            }
+
+            // pre-index the alerts so retrieving their
+            // monitor IDs for routing is easier
+            val alertsById: Map<String, AlertV2> = activeAlerts.associateBy { it.id }
+
+            val alertV2HistoryEnabled = monitorCtx.clusterService!!.clusterSettings.get(ALERT_V2_HISTORY_ENABLED)
+
+            // if alert v2 history is enabled, migrate the relevant alerts
+            // to the alert v2 history index pattern instead of hard deleting them
+            var copyResponse: BulkResponse? = null
+            if (alertV2HistoryEnabled) {
+                val indexRequests = searchAlertsResponse.hits.map { hit ->
+                    val xcp = XContentHelper.createParser(
+                        NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE,
+                        hit.sourceRef, XContentType.JSON
+                    )
+
+                    IndexRequest(ALERT_V2_HISTORY_WRITE_INDEX)
+                        .routing(monitorV2Id)
+                        .source(
+                            AlertV2.parse(xcp, hit.id, hit.version)
+                                .toXContentWithUser(XContentFactory.jsonBuilder())
+                        )
+                        .version(hit.version)
+                        .versionType(VersionType.EXTERNAL_GTE)
+                        .id(hit.id)
+                }
+                val copyRequest = BulkRequest().add(indexRequests)
+                copyResponse = client.suspendUntil { bulk(copyRequest, it) }
+
+                if (copyResponse!!.hasFailures()) {
+                    val retryCause = copyResponse.items.filter { it.isFailed }
+                        .firstOrNull { it.status() == RestStatus.TOO_MANY_REQUESTS }
+                        ?.failure?.cause
+                    throw RuntimeException(
+                        "Failed to copy alertV2s for [$monitorV2Id, ${monitorV2?.triggers?.map { it.id }}]: " +
+                            copyResponse.buildFailureMessage(),
+                        retryCause
+                    )
+                }
+            }
+
+            // prepare deletion request
+            val deleteRequests = if (alertV2HistoryEnabled) {
+                // if alerts were to be migrated, delete only the ones
+                // that were successfully copied over
+                copyResponse!!.items.filterNot { it.isFailed }.map {
+                    DeleteRequest(ALERT_V2_INDEX, it.id)
+                        .routing(alertsById[it.id]!!.monitorId)
+                        .version(it.version)
+                        .versionType(VersionType.EXTERNAL_GTE)
+                }
+            } else {
+                // otherwise just directly get the original
+                // set of alerts
+                searchAlertsResponse.hits.map { hit ->
+                    DeleteRequest(ALERT_V2_INDEX, hit.id)
+                        .routing(alertsById[hit.id]!!.monitorId)
+                        .version(hit.version)
+                        .versionType(VersionType.EXTERNAL_GTE)
+                }
+            }
+
+            // execute delete request
+            val deleteRequest = BulkRequest().add(deleteRequests)
+            val deleteResponse: BulkResponse = client.suspendUntil { bulk(deleteRequest, it) }
+
+            if (deleteResponse.hasFailures()) {
+                val retryCause = deleteResponse.items.filter { it.isFailed }
+                    .firstOrNull { it.status() == RestStatus.TOO_MANY_REQUESTS }
+                    ?.failure?.cause
+                throw RuntimeException(
+                    "Failed to delete alertV2s for [$monitorV2Id, ${monitorV2?.triggers?.map { it.id }}]: " +
+                        deleteResponse.buildFailureMessage(),
+                    retryCause
+                )
+            }
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestGetAlertsV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestGetAlertsV2Action.kt
new file mode 100644
index 000000000..912b8e93f
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/resthandlerv2/RestGetAlertsV2Action.kt
@@ -0,0 +1,70 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.resthandlerv2
+
+import org.apache.logging.log4j.LogManager
+import org.opensearch.alerting.AlertingPlugin
+import org.opensearch.alerting.actionv2.GetAlertsV2Action
+import org.opensearch.alerting.actionv2.GetAlertsV2Request
+import org.opensearch.commons.alerting.model.Table
+import org.opensearch.rest.BaseRestHandler
+import org.opensearch.rest.RestHandler.Route
+import org.opensearch.rest.RestRequest
+import org.opensearch.rest.RestRequest.Method.GET
+import org.opensearch.rest.action.RestToXContentListener
+import org.opensearch.transport.client.node.NodeClient
+
+/**
+ * This class consists of the REST handler to retrieve alerts .
+ */
+class RestGetAlertsV2Action : BaseRestHandler() {
+
+    private val log = LogManager.getLogger(RestGetAlertsV2Action::class.java)
+
+    override fun getName(): String {
+        return "get_alerts_v2_action"
+    }
+
+    override fun routes(): List<Route> {
+        return listOf(
+            Route(
+                GET,
+                "${AlertingPlugin.MONITOR_V2_BASE_URI}/alerts"
+            )
+        )
+    }
+
+    override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer {
+        log.debug("${request.method()} ${AlertingPlugin.MONITOR_V2_BASE_URI}/alerts")
+
+        val sortString = request.param("sortString", "monitor_v2_name.keyword")
+        val sortOrder = request.param("sortOrder", "asc")
+        val missing: String? = request.param("missing")
+        val size = request.paramAsInt("size", 20)
+        val startIndex = request.paramAsInt("startIndex", 0)
+        val searchString = request.param("searchString", "")
+        val severityLevel = request.param("severityLevel", "ALL")
+        val monitorId: String? = request.param("monitorId")
+        val table = Table(
+            sortOrder,
+            sortString,
+            missing,
+            size,
+            startIndex,
+            searchString
+        )
+
+        val getAlertsV2Request = GetAlertsV2Request(
+            table,
+            severityLevel,
+            monitorId?.let { listOf(monitorId) }
+        )
+        return RestChannelConsumer {
+                channel ->
+            client.execute(GetAlertsV2Action.INSTANCE, getAlertsV2Request, RestToXContentListener(channel))
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt b/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt
index d48552646..404db9f67 100644
--- a/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt
@@ -295,65 +295,65 @@ class AlertingSettings {
         )
 
         val ALERT_V2_HISTORY_ENABLED = Setting.boolSetting(
-            "plugins.alerting_v2.alert_history_enabled",
+            "plugins.alerting.v2.alert_history_enabled",
             true,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERT_V2_HISTORY_ROLLOVER_PERIOD = Setting.positiveTimeSetting(
-            "plugins.alerting_v2.alert_history_rollover_period",
+            "plugins.alerting.v2.alert_history_rollover_period",
             TimeValue(12, TimeUnit.HOURS),
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERT_V2_HISTORY_INDEX_MAX_AGE = Setting.positiveTimeSetting(
-            "plugins.alerting_v2.alert_history_max_age",
+            "plugins.alerting.v2.alert_history_max_age",
             TimeValue(30, TimeUnit.DAYS),
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERT_V2_HISTORY_MAX_DOCS = Setting.longSetting(
-            "plugins.alerting_v2.alert_history_max_docs",
+            "plugins.alerting.v2.alert_history_max_docs",
             1000L, 0L,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERT_V2_HISTORY_RETENTION_PERIOD = Setting.positiveTimeSetting(
-            "plugins.alerting_v2.alert_history_retention_period",
+            "plugins.alerting.v2.alert_history_retention_period",
             TimeValue(60, TimeUnit.DAYS),
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERTING_V2_MAX_MONITORS = Setting.intSetting(
-            "plugins.alerting_v2.monitor.max_monitors",
+            "plugins.alerting.v2.monitor.max_monitors",
             1000,
             1,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERTING_V2_MAX_THROTTLE_DURATION = Setting.longSetting(
-            "plugins.alerting_v2.monitor.max_throttle_duration",
+            "plugins.alerting.v2.monitor.max_throttle_duration",
             7200L, // 5 days, 7200 minutes
             2L,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERTING_V2_MAX_EXPIRE_DURATION = Setting.longSetting(
-            "plugins.alerting_v2.monitor.max_expire_duration",
+            "plugins.alerting.v2.monitor.max_expire_duration",
             43200L, // 30 days, 43200 minutes
             2L,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERTING_V2_MAX_LOOK_BACK_WINDOW = Setting.longSetting(
-            "plugins.alerting_v2.monitor.max_look_back_window",
+            "plugins.alerting.v2.monitor.max_look_back_window",
             10080L, // 7 days, 10080 minutes
             2L,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERTING_V2_MAX_QUERY_LENGTH = Setting.longSetting(
-            "plugins.alerting_v2.monitor.max_query_length",
+            "plugins.alerting.v2.monitor.max_query_length",
             2000L,
             0L,
             Setting.Property.NodeScope, Setting.Property.Dynamic
@@ -362,7 +362,7 @@ class AlertingSettings {
         // max data rows to retrieve when executing PPL query against
         // SQL/PPL plugin during monitor execution
         val ALERTING_V2_QUERY_RESULTS_MAX_DATAROWS = Setting.longSetting(
-            "plugins.alerting_v2.query_results_max_datarows",
+            "plugins.alerting.v2.query_results_max_datarows",
             1000L,
             1L,
             Setting.Property.NodeScope, Setting.Property.Dynamic
@@ -370,28 +370,28 @@ class AlertingSettings {
 
         // max size of query results to store in alerts and notifications
         val ALERT_V2_QUERY_RESULTS_MAX_SIZE = Setting.longSetting(
-            "plugins.alerting_v2.query_results_max_size",
+            "plugins.alerting.v2.query_results_max_size",
             3000L,
             0L,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val ALERT_V2_PER_RESULT_TRIGGER_MAX_ALERTS = Setting.intSetting(
-            "plugins.alerting_v2.per_result_trigger_max_alerts",
+            "plugins.alerting.v2.per_result_trigger_max_alerts",
             10,
             1,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val NOTIFICATION_SUBJECT_SOURCE_MAX_LENGTH = Setting.intSetting(
-            "plugins.alerting_v2.notification_subject_source_max_length",
+            "plugins.alerting.v2.notification_subject_source_max_length",
             1000,
             100,
             Setting.Property.NodeScope, Setting.Property.Dynamic
         )
 
         val NOTIFICATION_MESSAGE_SOURCE_MAX_LENGTH = Setting.intSetting(
-            "plugins.alerting_v2.notification_message_source_max_length",
+            "plugins.alerting.v2.notification_message_source_max_length",
             3000,
             1000,
             Setting.Property.NodeScope, Setting.Property.Dynamic
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetAlertsAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetAlertsAction.kt
index 1fc3ef83a..00e47a000 100644
--- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetAlertsAction.kt
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportGetAlertsAction.kt
@@ -16,6 +16,7 @@ import org.opensearch.action.search.SearchRequest
 import org.opensearch.action.search.SearchResponse
 import org.opensearch.action.support.ActionFilters
 import org.opensearch.action.support.HandledTransportAction
+import org.opensearch.alerting.AlertingV2Utils.validateMonitorV1
 import org.opensearch.alerting.alerts.AlertIndices
 import org.opensearch.alerting.opensearchapi.addFilter
 import org.opensearch.alerting.opensearchapi.suspendUntil
@@ -211,7 +212,11 @@ class TransportGetAlertsAction @Inject constructor(
                 xContentRegistry, LoggingDeprecationHandler.INSTANCE,
                 getResponse.sourceAsBytesRef, XContentType.JSON
             )
-            return ScheduledJob.parse(xcp, getResponse.id, getResponse.version) as Monitor
+            val scheduledJob = ScheduledJob.parse(xcp, getResponse.id, getResponse.version)
+            validateMonitorV1(scheduledJob)?.let {
+                throw it
+            }
+            return scheduledJob as Monitor
         } catch (t: Exception) {
             log.error("Failure in fetching monitor ${getAlertsRequest.monitorId} to resolve alert index in get alerts action", t)
             return null
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportGetAlertsV2Action.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportGetAlertsV2Action.kt
new file mode 100644
index 000000000..86ec85b8f
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/transportv2/TransportGetAlertsV2Action.kt
@@ -0,0 +1,204 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.transportv2
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import org.apache.logging.log4j.LogManager
+import org.opensearch.OpenSearchStatusException
+import org.opensearch.action.search.SearchRequest
+import org.opensearch.action.search.SearchResponse
+import org.opensearch.action.support.ActionFilters
+import org.opensearch.action.support.HandledTransportAction
+import org.opensearch.alerting.actionv2.GetAlertsV2Action
+import org.opensearch.alerting.actionv2.GetAlertsV2Request
+import org.opensearch.alerting.actionv2.GetAlertsV2Response
+import org.opensearch.alerting.alertsv2.AlertV2Indices
+import org.opensearch.alerting.core.settings.AlertingV2Settings.Companion.ALERTING_V2_ENABLED
+import org.opensearch.alerting.modelv2.AlertV2
+import org.opensearch.alerting.modelv2.AlertV2.Companion.MONITOR_V2_NAME_FIELD
+import org.opensearch.alerting.modelv2.AlertV2.Companion.MONITOR_V2_USER_FIELD
+import org.opensearch.alerting.modelv2.AlertV2.Companion.TRIGGER_V2_NAME_FIELD
+import org.opensearch.alerting.opensearchapi.addFilter
+import org.opensearch.alerting.settings.AlertingSettings
+import org.opensearch.alerting.transport.SecureTransportAction
+import org.opensearch.alerting.util.use
+import org.opensearch.cluster.service.ClusterService
+import org.opensearch.common.inject.Inject
+import org.opensearch.common.settings.Settings
+import org.opensearch.common.xcontent.LoggingDeprecationHandler
+import org.opensearch.common.xcontent.XContentHelper
+import org.opensearch.common.xcontent.XContentType
+import org.opensearch.commons.alerting.util.AlertingException
+import org.opensearch.commons.authuser.User
+import org.opensearch.commons.authuser.User.BACKEND_ROLES_FIELD
+import org.opensearch.core.action.ActionListener
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry
+import org.opensearch.core.rest.RestStatus
+import org.opensearch.core.xcontent.NamedXContentRegistry
+import org.opensearch.index.query.Operator
+import org.opensearch.index.query.QueryBuilders
+import org.opensearch.search.builder.SearchSourceBuilder
+import org.opensearch.search.sort.SortBuilders
+import org.opensearch.search.sort.SortOrder
+import org.opensearch.tasks.Task
+import org.opensearch.transport.TransportService
+import org.opensearch.transport.client.Client
+import java.io.IOException
+
+private val log = LogManager.getLogger(TransportGetAlertsV2Action::class.java)
+private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO)
+
+class TransportGetAlertsV2Action @Inject constructor(
+    transportService: TransportService,
+    val client: Client,
+    clusterService: ClusterService,
+    actionFilters: ActionFilters,
+    val settings: Settings,
+    val xContentRegistry: NamedXContentRegistry,
+    val namedWriteableRegistry: NamedWriteableRegistry
+) : HandledTransportAction<GetAlertsV2Request, GetAlertsV2Response>(
+    GetAlertsV2Action.NAME,
+    transportService,
+    actionFilters,
+    ::GetAlertsV2Request
+),
+    SecureTransportAction {
+
+    @Volatile private var alertingV2Enabled = ALERTING_V2_ENABLED.get(settings)
+
+    @Volatile
+    override var filterByEnabled = AlertingSettings.FILTER_BY_BACKEND_ROLES.get(settings)
+
+    init {
+        clusterService.clusterSettings.addSettingsUpdateConsumer(ALERTING_V2_ENABLED) { alertingV2Enabled = it }
+        listenFilterBySettingChange(clusterService)
+    }
+
+    override fun doExecute(
+        task: Task,
+        getAlertsV2Request: GetAlertsV2Request,
+        actionListener: ActionListener<GetAlertsV2Response>,
+    ) {
+        if (!alertingV2Enabled) {
+            actionListener.onFailure(
+                AlertingException.wrap(
+                    OpenSearchStatusException(
+                        "Alerting V2 is currently disabled, please enable it with the " +
+                            "cluster setting: ${ALERTING_V2_ENABLED.key}",
+                        RestStatus.FORBIDDEN
+                    ),
+                )
+            )
+            return
+        }
+
+        val user = readUserFromThreadContext(client)
+
+        val tableProp = getAlertsV2Request.table
+        val sortBuilder = SortBuilders
+            .fieldSort(tableProp.sortString)
+            .order(SortOrder.fromString(tableProp.sortOrder))
+        if (!tableProp.missing.isNullOrBlank()) {
+            sortBuilder.missing(tableProp.missing)
+        }
+
+        val queryBuilder = QueryBuilders.boolQuery()
+
+        if (getAlertsV2Request.severityLevel != "ALL") {
+            queryBuilder.filter(QueryBuilders.termQuery("severity", getAlertsV2Request.severityLevel))
+        }
+
+        if (!getAlertsV2Request.monitorV2Ids.isNullOrEmpty()) {
+            queryBuilder.filter(QueryBuilders.termsQuery("monitor_id", getAlertsV2Request.monitorV2Ids))
+        }
+
+        if (!tableProp.searchString.isNullOrBlank()) {
+            queryBuilder
+                .must(
+                    QueryBuilders
+                        .queryStringQuery(tableProp.searchString)
+                        .defaultOperator(Operator.AND)
+                        .field(MONITOR_V2_NAME_FIELD)
+                        .field(TRIGGER_V2_NAME_FIELD)
+                )
+        }
+        val searchSourceBuilder = SearchSourceBuilder()
+            .version(true)
+            .seqNoAndPrimaryTerm(true)
+            .query(queryBuilder)
+            .sort(sortBuilder)
+            .size(tableProp.size)
+            .from(tableProp.startIndex)
+
+        client.threadPool().threadContext.stashContext().use {
+            scope.launch {
+                try {
+                    getAlerts(AlertV2Indices.ALERT_V2_INDEX, searchSourceBuilder, actionListener, user)
+                } catch (t: Exception) {
+                    log.error("Failed to get alerts", t)
+                    if (t is AlertingException) {
+                        actionListener.onFailure(t)
+                    } else {
+                        actionListener.onFailure(AlertingException.wrap(t))
+                    }
+                }
+            }
+        }
+    }
+
+    fun getAlerts(
+        alertIndex: String,
+        searchSourceBuilder: SearchSourceBuilder,
+        actionListener: ActionListener<GetAlertsV2Response>,
+        user: User?
+    ) {
+        try {
+            // if user is null, security plugin is disabled or user is super-admin
+            // if doFilterForUser() is false, security is enabled but filterby is disabled
+            if (user != null && doFilterForUser(user)) {
+                // if security is enabled and filterby is enabled, add search filter
+                log.info("Filtering result by: ${user.backendRoles}")
+                addFilter(user, searchSourceBuilder, "$MONITOR_V2_USER_FIELD.$BACKEND_ROLES_FIELD.keyword")
+            }
+
+            search(alertIndex, searchSourceBuilder, actionListener)
+        } catch (ex: IOException) {
+            actionListener.onFailure(AlertingException.wrap(ex))
+        }
+    }
+
+    fun search(alertIndex: String, searchSourceBuilder: SearchSourceBuilder, actionListener: ActionListener<GetAlertsV2Response>) {
+        val searchRequest = SearchRequest()
+            .indices(alertIndex)
+            .source(searchSourceBuilder)
+
+        client.search(
+            searchRequest,
+            object : ActionListener<SearchResponse> {
+                override fun onResponse(response: SearchResponse) {
+                    val totalAlertCount = response.hits.totalHits?.value?.toInt()
+                    val alerts = response.hits.map { hit ->
+                        val xcp = XContentHelper.createParser(
+                            xContentRegistry,
+                            LoggingDeprecationHandler.INSTANCE,
+                            hit.sourceRef,
+                            XContentType.JSON
+                        )
+                        val alertV2 = AlertV2.parse(xcp, hit.id, hit.version)
+                        alertV2
+                    }
+                    actionListener.onResponse(GetAlertsV2Response(alerts, totalAlertCount))
+                }
+
+                override fun onFailure(t: Exception) {
+                    actionListener.onFailure(t)
+                }
+            }
+        )
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt b/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt
index 994293f1d..b388ae757 100644
--- a/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/util/IndexUtils.kt
@@ -10,6 +10,7 @@ import org.opensearch.action.index.IndexResponse
 import org.opensearch.action.support.IndicesOptions
 import org.opensearch.action.support.clustermanager.AcknowledgedResponse
 import org.opensearch.alerting.alerts.AlertIndices
+import org.opensearch.alerting.alertsv2.AlertV2Indices
 import org.opensearch.alerting.comments.CommentsIndices
 import org.opensearch.alerting.core.ScheduledJobIndices
 import org.opensearch.cluster.ClusterState
@@ -39,6 +40,8 @@ class IndexUtils {
             private set
         var alertingCommentIndexSchemaVersion: Int
             private set
+        var alertV2IndexSchemaVersion: Int
+            private set
 
         var scheduledJobIndexUpdated: Boolean = false
             private set
@@ -48,16 +51,20 @@ class IndexUtils {
             private set
         var commentsIndexUpdated: Boolean = false
             private set
+        var alertV2IndexUpdated: Boolean = false
+            private set
 
         var lastUpdatedAlertHistoryIndex: String? = null
         var lastUpdatedFindingHistoryIndex: String? = null
         var lastUpdatedCommentsHistoryIndex: String? = null
+        var lastUpdatedAlertV2HistoryIndex: String? = null
 
         init {
             scheduledJobIndexSchemaVersion = getSchemaVersion(ScheduledJobIndices.scheduledJobMappings())
             alertIndexSchemaVersion = getSchemaVersion(AlertIndices.alertMapping())
             findingIndexSchemaVersion = getSchemaVersion(AlertIndices.findingMapping())
             alertingCommentIndexSchemaVersion = getSchemaVersion(CommentsIndices.commentsMapping())
+            alertV2IndexSchemaVersion = getSchemaVersion(AlertV2Indices.alertV2Mapping())
         }
 
         @JvmStatic
@@ -80,6 +87,11 @@ class IndexUtils {
             commentsIndexUpdated = true
         }
 
+        @JvmStatic
+        fun alertV2IndexUpdated() {
+            commentsIndexUpdated = true
+        }
+
         @JvmStatic
         fun getSchemaVersion(mapping: String): Int {
             val xcp = XContentType.JSON.xContent().createParser(
diff --git a/alerting/src/main/resources/org/opensearch/alerting/alertsv2/alert_v2_mapping.json b/alerting/src/main/resources/org/opensearch/alerting/alertsv2/alert_v2_mapping.json
new file mode 100644
index 000000000..5543a289c
--- /dev/null
+++ b/alerting/src/main/resources/org/opensearch/alerting/alertsv2/alert_v2_mapping.json
@@ -0,0 +1,118 @@
+{
+  "dynamic": "strict",
+  "_routing": {
+    "required": true
+  },
+  "_meta" : {
+    "schema_version": 1
+  },
+  "properties": {
+    "schema_version": {
+      "type": "integer"
+    },
+    "monitor_v2_id": {
+      "type": "keyword"
+    },
+    "monitor_v2_version": {
+      "type": "long"
+    },
+    "id": {
+      "type": "keyword"
+    },
+    "version": {
+      "type": "long"
+    },
+    "severity": {
+      "type": "keyword"
+    },
+    "monitor_v2_name": {
+      "type": "text",
+      "fields": {
+        "keyword": {
+          "type": "keyword",
+          "ignore_above": 256
+        }
+      }
+    },
+    "monitor_v2_user": {
+      "properties": {
+        "name": {
+          "type": "text",
+          "fields": {
+            "keyword": {
+              "type": "keyword",
+              "ignore_above": 256
+            }
+          }
+        },
+        "backend_roles": {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "type" : "keyword"
+            }
+          }
+        },
+        "roles": {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "type" : "keyword"
+            }
+          }
+        },
+        "custom_attribute_names": {
+          "type" : "text",
+          "fields" : {
+            "keyword" : {
+              "type" : "keyword"
+            }
+          }
+        }
+      }
+    },
+    "execution_id": {
+      "type": "keyword"
+    },
+    "trigger_v2_id": {
+      "type": "keyword"
+    },
+    "trigger_v2_name": {
+      "type": "text",
+      "fields": {
+        "keyword": {
+          "type": "keyword",
+          "ignore_above": 256
+        }
+      }
+    },
+    "triggered_time": {
+      "type": "date"
+    },
+    "error_message": {
+      "type": "text"
+    },
+    "query": {
+      "type": "text"
+    },
+    "query_results":  {
+      "type": "nested",
+      "properties": {
+        "schema": {
+          "type": "nested",
+          "dynamic": true
+        },
+        "datarows": {
+          "type": "object",
+          "enabled": false
+        },
+        "total": {
+          "type": "integer"
+        },
+        "size": {
+          "type": "integer"
+        }
+      }
+    }
+  }
+}
\ No newline at end of file
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt b/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt
index 84433c779..434f01666 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/AlertingRestTestCase.kt
@@ -20,6 +20,7 @@ import org.opensearch.alerting.AlertingPlugin.Companion.EMAIL_GROUP_BASE_URI
 import org.opensearch.alerting.AlertingPlugin.Companion.MONITOR_V2_BASE_URI
 import org.opensearch.alerting.alerts.AlertIndices
 import org.opensearch.alerting.alerts.AlertIndices.Companion.FINDING_HISTORY_WRITE_INDEX
+import org.opensearch.alerting.alertsv2.AlertV2Indices
 import org.opensearch.alerting.core.settings.ScheduledJobSettings
 import org.opensearch.alerting.model.destination.Chime
 import org.opensearch.alerting.model.destination.CustomWebhook
@@ -27,6 +28,7 @@ import org.opensearch.alerting.model.destination.Destination
 import org.opensearch.alerting.model.destination.Slack
 import org.opensearch.alerting.model.destination.email.EmailAccount
 import org.opensearch.alerting.model.destination.email.EmailGroup
+import org.opensearch.alerting.modelv2.AlertV2
 import org.opensearch.alerting.modelv2.MonitorV2
 import org.opensearch.alerting.modelv2.PPLSQLMonitor
 import org.opensearch.alerting.settings.AlertingSettings
@@ -843,6 +845,35 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
         }
     }
 
+    protected fun searchAlertV2s(
+        monitorV2Id: String,
+        indices: String = AlertV2Indices.ALERT_V2_INDEX,
+        refresh: Boolean = true
+    ): List<AlertV2> {
+        try {
+            if (refresh) refreshIndex(indices)
+        } catch (e: Exception) {
+            logger.warn("Could not refresh index $indices because: ${e.message}")
+            return emptyList()
+        }
+
+        // If this is a test monitor (it doesn't have an ID) and no alerts will be saved for it.
+        val searchParams = if (monitorV2Id != MonitorV2.NO_ID) mapOf("routing" to monitorV2Id) else mapOf()
+        val request = """
+                { "version" : true,
+                  "query" : { "term" : { "${AlertV2.MONITOR_V2_ID_FIELD}" : "$monitorV2Id" } }
+                }
+        """.trimIndent()
+        val httpResponse = adminClient().makeRequest("GET", "/$indices/_search", searchParams, StringEntity(request, APPLICATION_JSON))
+        assertEquals("Search failed", RestStatus.OK, httpResponse.restStatus())
+
+        val searchResponse = SearchResponse.fromXContent(createParser(jsonXContent, httpResponse.entity.content))
+        return searchResponse.hits.hits.map {
+            val xcp = createParser(jsonXContent, it.sourceRef)
+            AlertV2.parse(xcp, it.id, it.version)
+        }
+    }
+
     protected fun acknowledgeAlerts(monitor: Monitor, vararg alerts: Alert): Response {
         val request = XContentFactory.jsonBuilder().startObject()
             .array("alerts", *alerts.map { it.id }.toTypedArray())
@@ -898,7 +929,7 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
     protected fun getAlertV2s(): Response {
         val response = client().makeRequest(
             "GET",
-            "$MONITOR_V2_BASE_URI/alerts?",
+            "$MONITOR_V2_BASE_URI/alerts",
             null,
             BasicHeader(HttpHeaders.CONTENT_TYPE, "application/json")
         )
@@ -1409,6 +1440,14 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
         createIndex(encodedHistoryIndex, settings, mappingHack, "\"${AlertIndices.FINDING_HISTORY_WRITE_INDEX}\" : {}")
     }
 
+    fun putAlertV2Mappings(mapping: String? = null) {
+        val mappingHack = if (mapping != null) mapping else AlertV2Indices.alertV2Mapping().trimStart('{').trimEnd('}')
+        val encodedHistoryIndex = URLEncoder.encode(AlertV2Indices.ALERT_V2_HISTORY_INDEX_PATTERN, Charsets.UTF_8.toString())
+        val settings = Settings.builder().put("index.hidden", true).build()
+        createIndex(AlertV2Indices.ALERT_V2_INDEX, settings, mappingHack)
+        createIndex(encodedHistoryIndex, settings, mappingHack, "\"${AlertV2Indices.ALERT_V2_HISTORY_WRITE_INDEX}\" : {}")
+    }
+
     fun scheduledJobMappings(): String {
         return javaClass.classLoader.getResource("mappings/scheduled-jobs.json").readText()
     }
@@ -2197,7 +2236,7 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
         val search = SearchSourceBuilder().query(QueryBuilders.matchAllQuery()).toString()
         val searchResponse = client().makeRequest(
             "POST", "$MONITOR_V2_BASE_URI/_search",
-            StringEntity(search, ContentType.APPLICATION_JSON)
+            StringEntity(search, APPLICATION_JSON)
         )
 
         assertEquals("Search monitor failed", RestStatus.OK, searchResponse.restStatus())
@@ -2218,6 +2257,22 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
     // takes in a get alerts API response and returns the current number of active alerts
     protected fun numAlerts(getAlertsResponse: Response): Int {
         logger.info("get alerts response: ${entityAsMap(getAlertsResponse)}")
-        return entityAsMap(getAlertsResponse)["totalAlertV2s"] as Int
+        return entityAsMap(getAlertsResponse)["total_alerts_v2"] as Int
+    }
+
+    protected fun getAlertV2HistoryDocCount(): Long {
+        val request = """
+            {
+                "query": {
+                    "match_all": {}
+                }
+            }
+        """.trimIndent()
+        val response = adminClient().makeRequest(
+            "POST", "${AlertV2Indices.ALERT_V2_HISTORY_ALL}/_search", emptyMap(),
+            StringEntity(request, APPLICATION_JSON)
+        )
+        assertEquals("Request to get alert v2 history failed", RestStatus.OK, response.restStatus())
+        return SearchResponse.fromXContent(createParser(jsonXContent, response.entity.content)).hits.totalHits!!.value
     }
 }
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt b/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt
index 6f8c05d09..75819d6aa 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt
@@ -507,7 +507,7 @@ fun randomAction(
     name: String = OpenSearchRestTestCase.randomUnicodeOfLength(10),
     template: Script = randomTemplateScript("Hello World"),
     subjectTemplate: Script = template,
-    destinationId: String = "",
+    destinationId: String = "abc",
     throttleEnabled: Boolean = false,
     throttle: Throttle = randomThrottle()
 ) = Action(name, destinationId, subjectTemplate, template, throttleEnabled, throttle, actionExecutionPolicy = null)
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2RequestTests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2RequestTests.kt
new file mode 100644
index 000000000..84a6c8fb8
--- /dev/null
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2RequestTests.kt
@@ -0,0 +1,49 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.actionv2
+
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.commons.alerting.model.Table
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.test.OpenSearchTestCase
+
+class GetAlertsV2RequestTests : OpenSearchTestCase() {
+    fun `test get alerts request`() {
+        val table = Table("asc", "sortString", null, 1, 0, "")
+
+        val req = GetAlertsV2Request(
+            table = table,
+            severityLevel = "1",
+            monitorV2Ids = listOf("1", "2"),
+        )
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetAlertsV2Request(sin)
+
+        assertEquals("1", newReq.severityLevel)
+        assertEquals(table, newReq.table)
+        assertTrue(newReq.monitorV2Ids!!.contains("1"))
+        assertTrue(newReq.monitorV2Ids!!.contains("2"))
+    }
+
+    fun `test get alerts request with filter`() {
+        val table = Table("asc", "sortString", null, 1, 0, "")
+        val req = GetAlertsV2Request(table, "1", null)
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetAlertsV2Request(sin)
+
+        assertEquals("1", newReq.severityLevel)
+        assertNull(newReq.monitorV2Ids)
+        assertEquals(table, newReq.table)
+    }
+}
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2ResponseTests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2ResponseTests.kt
new file mode 100644
index 000000000..8f803bc00
--- /dev/null
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/actionv2/GetAlertsV2ResponseTests.kt
@@ -0,0 +1,93 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.actionv2
+
+import org.opensearch.alerting.modelv2.AlertV2
+import org.opensearch.alerting.modelv2.TriggerV2
+import org.opensearch.alerting.randomUser
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.common.xcontent.XContentType
+import org.opensearch.commons.alerting.util.string
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.XContentBuilder
+import org.opensearch.test.OpenSearchTestCase
+import java.time.Instant
+import java.util.Collections
+
+class GetAlertsV2ResponseTests : OpenSearchTestCase() {
+    fun `test get alerts response with no alerts`() {
+        val req = GetAlertsV2Response(Collections.emptyList(), 0)
+        assertNotNull(req)
+
+        val out = BytesStreamOutput()
+        req.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newReq = GetAlertsV2Response(sin)
+        assertTrue(newReq.alertV2s.isEmpty())
+        assertEquals(0, newReq.totalAlertV2s)
+    }
+
+    fun `test get alerts response with alerts`() {
+        val alert = AlertV2(
+            monitorId = "id",
+            monitorName = "name",
+            monitorVersion = AlertV2.NO_VERSION,
+            monitorUser = randomUser(),
+            triggerId = "triggerId",
+            triggerName = "triggerNamer",
+            query = "source = some_index",
+            queryResults = mapOf(),
+            triggeredTime = Instant.now(),
+            errorMessage = null,
+            severity = TriggerV2.Severity.LOW,
+            executionId = "executionId"
+        )
+        val res = GetAlertsV2Response(listOf(alert), 1)
+        assertNotNull(res)
+
+        val out = BytesStreamOutput()
+        res.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newRes = GetAlertsV2Response(sin)
+        assertEquals(1, newRes.alertV2s.size)
+        assertEquals(alert, newRes.alertV2s[0])
+        assertEquals(1, newRes.totalAlertV2s)
+    }
+
+    fun `test toXContent for get alerts response`() {
+        val now = Instant.now()
+        val alert = AlertV2(
+            monitorId = "id",
+            monitorName = "name",
+            monitorVersion = AlertV2.NO_VERSION,
+            monitorUser = randomUser(),
+            triggerId = "triggerId",
+            triggerName = "triggerName",
+            query = "source = some_index",
+            queryResults = mapOf(),
+            triggeredTime = now,
+            errorMessage = null,
+            severity = TriggerV2.Severity.LOW,
+            executionId = "executionId"
+        )
+
+        val req = GetAlertsV2Response(listOf(alert), 1)
+        var actualXContentString = req.toXContent(
+            XContentBuilder.builder(XContentType.JSON.xContent()),
+            ToXContent.EMPTY_PARAMS
+        ).string()
+        val expectedXContentString = "{\"alerts_v2\":[{\"id\":\"\",\"version\":-1,\"monitor_v2_id\":\"id\",\"schema_version\":0," +
+            "\"monitor_v2_version\":-1,\"monitor_v2_name\":\"name\",\"execution_id\":\"executionId\",\"trigger_v2_id\":\"triggerId\"," +
+            "\"trigger_v2_name\":\"triggerName\",\"query\":\"source = some_index\",\"query_results\":{},\"error_message\":null," +
+            "\"severity\":\"low\",\"triggered_time\":${now.toEpochMilli()}}],\"total_alerts_v2\":1}"
+
+        logger.info("expected: $expectedXContentString")
+        logger.info("actual: $actualXContentString")
+
+        assertEquals(expectedXContentString, actualXContentString)
+    }
+}
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/alerts/AlertIndicesIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/alerts/AlertIndicesIT.kt
index 69a7e0363..9e1e2437f 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/alerts/AlertIndicesIT.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/alerts/AlertIndicesIT.kt
@@ -64,7 +64,7 @@ class AlertIndicesIT : AlertingRestTestCase() {
         executeMonitor(createRandomMonitor())
         assertIndexExists(AlertIndices.ALERT_INDEX)
         assertIndexExists(AlertIndices.ALERT_HISTORY_WRITE_INDEX)
-        verifyIndexSchemaVersion(ScheduledJob.SCHEDULED_JOBS_INDEX, 8)
+        verifyIndexSchemaVersion(ScheduledJob.SCHEDULED_JOBS_INDEX, 9)
         verifyIndexSchemaVersion(AlertIndices.ALERT_INDEX, 5)
         verifyIndexSchemaVersion(AlertIndices.ALERT_HISTORY_WRITE_INDEX, 5)
     }
@@ -88,7 +88,7 @@ class AlertIndicesIT : AlertingRestTestCase() {
         val trueMonitor = createMonitor(randomDocumentLevelMonitor(inputs = listOf(docLevelInput), triggers = listOf(trigger)))
         executeMonitor(trueMonitor.id)
         assertIndexExists(AlertIndices.FINDING_HISTORY_WRITE_INDEX)
-        verifyIndexSchemaVersion(ScheduledJob.SCHEDULED_JOBS_INDEX, 8)
+        verifyIndexSchemaVersion(ScheduledJob.SCHEDULED_JOBS_INDEX, 9)
         verifyIndexSchemaVersion(AlertIndices.FINDING_HISTORY_WRITE_INDEX, 4)
     }
 
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt
index 737dbed6e..4bd478782 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorV2RestApiIT.kt
@@ -7,11 +7,13 @@ package org.opensearch.alerting.resthandler
 
 import org.apache.hc.core5.http.ContentType
 import org.apache.hc.core5.http.io.entity.StringEntity
+import org.junit.Before
 import org.opensearch.alerting.AlertingPlugin.Companion.MONITOR_V2_BASE_URI
 import org.opensearch.alerting.AlertingRestTestCase
 import org.opensearch.alerting.TEST_INDEX_MAPPINGS
 import org.opensearch.alerting.TEST_INDEX_NAME
 import org.opensearch.alerting.assertPplMonitorsEqual
+import org.opensearch.alerting.core.settings.AlertingV2Settings
 import org.opensearch.alerting.makeRequest
 import org.opensearch.alerting.modelv2.MonitorV2
 import org.opensearch.alerting.modelv2.PPLSQLMonitor
@@ -53,6 +55,10 @@ import java.time.temporal.ChronoUnit.MINUTES
 @TestLogging("level:DEBUG", reason = "Debug for tests.")
 @Suppress("UNCHECKED_CAST")
 class MonitorV2RestApiIT : AlertingRestTestCase() {
+    @Before
+    fun enableAlertingV2() {
+        client().updateSettings(AlertingV2Settings.ALERTING_V2_ENABLED.key, "true")
+    }
 
     /* Simple Case Tests */
     fun `test create ppl monitor`() {
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt
index bc4178784..2fc0420fa 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureMonitorV2RestApiIT.kt
@@ -19,6 +19,7 @@ import org.opensearch.alerting.AlertingRestTestCase
 import org.opensearch.alerting.PPL_FULL_ACCESS_ROLE
 import org.opensearch.alerting.ROLE_TO_PERMISSION_MAPPING
 import org.opensearch.alerting.TEST_INDEX_NAME
+import org.opensearch.alerting.core.settings.AlertingV2Settings
 import org.opensearch.alerting.makeRequest
 import org.opensearch.alerting.randomPPLMonitor
 import org.opensearch.client.ResponseException
@@ -52,6 +53,7 @@ class SecureMonitorV2RestApiIT : AlertingRestTestCase() {
 
     @Before
     fun create() {
+        client().updateSettings(AlertingV2Settings.ALERTING_V2_ENABLED.key, "true")
         if (userClient == null) {
             createUser(user, arrayOf())
             userClient = SecureRestClientBuilder(clusterHosts.toTypedArray(), isHttps(), user, password)
diff --git a/core/src/main/kotlin/org/opensearch/alerting/core/settings/AlertingV2Settings.kt b/core/src/main/kotlin/org/opensearch/alerting/core/settings/AlertingV2Settings.kt
new file mode 100644
index 000000000..cbb9fa9df
--- /dev/null
+++ b/core/src/main/kotlin/org/opensearch/alerting/core/settings/AlertingV2Settings.kt
@@ -0,0 +1,17 @@
+package org.opensearch.alerting.core.settings
+
+import org.opensearch.common.settings.Setting
+
+/**
+ *  This class exclusively houses the Alerting V2 enabled setting, so that both Monitor V2 Stats
+ *  and the rest of the CRUD APIs can read it
+ */
+class AlertingV2Settings {
+    companion object {
+        val ALERTING_V2_ENABLED = Setting.boolSetting(
+            "plugins.alerting.v2.enabled",
+            true,
+            Setting.Property.NodeScope, Setting.Property.Dynamic
+        )
+    }
+}
diff --git a/core/src/main/resources/mappings/scheduled-jobs.json b/core/src/main/resources/mappings/scheduled-jobs.json
index 6e3d31c51..ba5c0010d 100644
--- a/core/src/main/resources/mappings/scheduled-jobs.json
+++ b/core/src/main/resources/mappings/scheduled-jobs.json
@@ -1,6 +1,6 @@
 {
   "_meta" : {
-    "schema_version": 8
+    "schema_version": 9
   },
   "properties": {
     "monitor": {
@@ -450,6 +450,201 @@
         }
       }
     },
+    "monitor_v2": {
+      "dynamic": "false",
+      "properties": {
+        "ppl_monitor": {
+          "dynamic": "false",
+          "properties": {
+            "schema_version": {
+              "type": "integer"
+            },
+            "name": {
+              "type": "text",
+              "fields": {
+                "keyword": {
+                  "type": "keyword",
+                  "ignore_above": 256
+                }
+              }
+            },
+            "enabled": {
+              "type": "boolean"
+            },
+            "schedule": {
+              "properties": {
+                "period": {
+                  "properties": {
+                    "interval": {
+                      "type": "integer"
+                    },
+                    "unit": {
+                      "type": "keyword"
+                    }
+                  }
+                },
+                "cron": {
+                  "properties": {
+                    "expression": {
+                      "type": "text"
+                    },
+                    "timezone": {
+                      "type": "keyword"
+                    }
+                  }
+                }
+              }
+            },
+            "look_back_window_minutes": {
+              "type": "long"
+            },
+            "timestamp_field": {
+              "type": "text",
+              "fields": {
+                "keyword": {
+                  "type": "keyword",
+                  "ignore_above": 256
+                }
+              }
+            },
+            "last_update_time": {
+              "type": "date",
+              "format": "strict_date_time||epoch_millis"
+            },
+            "enabled_time": {
+              "type": "date",
+              "format": "strict_date_time||epoch_millis"
+            },
+            "description": {
+              "type": "text"
+            },
+            "user": {
+              "properties": {
+                "name": {
+                  "type": "text",
+                  "fields": {
+                    "keyword": {
+                      "type": "keyword",
+                      "ignore_above": 256
+                    }
+                  }
+                },
+                "backend_roles": {
+                  "type" : "text",
+                  "fields" : {
+                    "keyword" : {
+                      "type" : "keyword"
+                    }
+                  }
+                },
+                "roles": {
+                  "type" : "text",
+                  "fields" : {
+                    "keyword" : {
+                      "type" : "keyword"
+                    }
+                  }
+                },
+                "custom_attribute_names": {
+                  "type" : "text",
+                  "fields" : {
+                    "keyword" : {
+                      "type" : "keyword"
+                    }
+                  }
+                }
+              }
+            },
+            "query_language": {
+              "type": "keyword"
+            },
+            "query": {
+              "type": "text"
+            },
+            "triggers": {
+              "type": "nested",
+              "properties": {
+                "name": {
+                  "type": "text",
+                  "fields": {
+                    "keyword": {
+                      "type": "keyword",
+                      "ignore_above": 256
+                    }
+                  }
+                },
+                "severity": {
+                  "type": "keyword"
+                },
+                "throttle_minutes": {
+                  "type": "long"
+                },
+                "expires_minutes": {
+                  "type": "long"
+                },
+                "last_triggered_time": {
+                  "type": "date",
+                  "format": "strict_date_time||epoch_millis"
+                },
+                "mode": {
+                  "type": "keyword"
+                },
+                "type": {
+                  "type": "keyword"
+                },
+                "num_results_condition": {
+                  "type": "keyword"
+                },
+                "num_results_value": {
+                  "type": "long"
+                },
+                "custom_condition": {
+                  "type": "text"
+                },
+                "actions": {
+                  "type": "nested",
+                  "properties": {
+                    "name": {
+                      "type": "text",
+                      "fields": {
+                        "keyword": {
+                          "type": "keyword",
+                          "ignore_above": 256
+                        }
+                      }
+                    },
+                    "destination_id": {
+                      "type": "keyword"
+                    },
+                    "subject_template": {
+                      "type": "object",
+                      "enabled": false
+                    },
+                    "message_template": {
+                      "type": "object",
+                      "enabled": false
+                    },
+                    "throttle_enabled": {
+                      "type": "boolean"
+                    },
+                    "throttle": {
+                      "properties": {
+                        "value": {
+                          "type": "integer"
+                        },
+                        "unit": {
+                          "type": "keyword"
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "destination": {
       "dynamic": "false",
       "properties": {