diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/AlertV2.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/AlertV2.kt
new file mode 100644
index 000000000..7f60201c6
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/AlertV2.kt
@@ -0,0 +1,255 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.core.util.nonOptionalTimeField
+import org.opensearch.alerting.modelv2.TriggerV2.Severity
+import org.opensearch.common.lucene.uid.Versions
+import org.opensearch.commons.alerting.util.instant
+import org.opensearch.commons.alerting.util.optionalUserField
+import org.opensearch.commons.authuser.User
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.common.io.stream.Writeable
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.XContentBuilder
+import org.opensearch.core.xcontent.XContentParser
+import org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken
+import java.io.IOException
+import java.time.Instant
+
+/**
+ * Alert generated by Alerting V2
+ * An alert is created when a Trigger's trigger conditions are met.
+ *
+ * @property id Alert ID. Defaults to [NO_ID].
+ * @property version Version number of the Alert. Defaults to [NO_VERSION].
+ * @property schemaVersion Version of the alerting-alerts index schema when this Alert was indexed. Defaults to [NO_SCHEMA_VERSION].
+ * @property monitorId ID of the Monitor that generated this Alert.
+ * @property monitorName Name of the Monitor that generated this Alert.
+ * @property monitorVersion Version of the Monitor at the time it generated this Alert.
+ * @property triggerId ID of the Trigger in the Monitor that generated this alert.
+ * @property triggerName Name of the trigger in the Monitor that generated this alert.
+ * @property queryResults Results from the Monitor's query that caused the Trigger to fire.
+ * @property triggeredTime Timestamp for when the Alert was generated.
+ * @property errorMessage Optional error message if there were issues during Trigger execution.
+ *                       Null indicates no errors occurred.
+ * @property severity Severity level of the alert (e.g., "HIGH", "MEDIUM", "LOW").
+ * @property executionId Optional ID for the Monitor execution that generated this Alert.
+ *
+ * @see MonitorV2 For the monitor that generates alerts
+ * @see TriggerV2 For the trigger conditions that create alerts
+ *
+ * Lifecycle:
+ * 1. AlertV2 is generated when a TriggerV2's condition is met. The TriggerV2 fires and forgets the AlertV2.
+ * 2. AlertV2 is stored in the alerts index. AlertV2s are stateless. (e.g. they are never ACTIVE or COMPLETED)
+ * 3. AlertV2 is soft deleted after its expire duration (determined by its trigger), and archived in an alert history index
+ * 4. Based on the alert v2 history retention period, the AlertV2 is permanently deleted
+ */
+data class AlertV2(
+    val id: String = NO_ID,
+    val version: Long = NO_VERSION,
+    val schemaVersion: Int = NO_SCHEMA_VERSION,
+    val monitorId: String,
+    val monitorName: String,
+    val monitorVersion: Long,
+    val monitorUser: User?,
+    val triggerId: String,
+    val triggerName: String,
+    val query: String,
+    val queryResults: Map<String, Any>,
+    val triggeredTime: Instant,
+    val errorMessage: String? = null,
+    val severity: Severity,
+    val executionId: String? = null
+) : Writeable, ToXContent {
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        id = sin.readString(),
+        version = sin.readLong(),
+        schemaVersion = sin.readInt(),
+        monitorId = sin.readString(),
+        monitorName = sin.readString(),
+        monitorVersion = sin.readLong(),
+        monitorUser = if (sin.readBoolean()) {
+            User(sin)
+        } else {
+            null
+        },
+        triggerId = sin.readString(),
+        triggerName = sin.readString(),
+        query = sin.readString(),
+        queryResults = sin.readMap(),
+        triggeredTime = sin.readInstant(),
+        errorMessage = sin.readOptionalString(),
+        severity = sin.readEnum(Severity::class.java),
+        executionId = sin.readOptionalString()
+    )
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(id)
+        out.writeLong(version)
+        out.writeInt(schemaVersion)
+        out.writeString(monitorId)
+        out.writeString(monitorName)
+        out.writeLong(monitorVersion)
+        out.writeBoolean(monitorUser != null)
+        monitorUser?.writeTo(out)
+        out.writeString(triggerId)
+        out.writeString(triggerName)
+        out.writeString(query)
+        out.writeMap(queryResults)
+        out.writeInstant(triggeredTime)
+        out.writeOptionalString(errorMessage)
+        out.writeEnum(severity)
+        out.writeOptionalString(executionId)
+    }
+
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        return createXContentBuilder(builder, false)
+    }
+
+    fun toXContentWithUser(builder: XContentBuilder): XContentBuilder {
+        return createXContentBuilder(builder, true)
+    }
+
+    private fun createXContentBuilder(builder: XContentBuilder, withUser: Boolean): XContentBuilder {
+        builder.startObject()
+            .field(ALERT_V2_ID_FIELD, id)
+            .field(ALERT_V2_VERSION_FIELD, version)
+            .field(MONITOR_V2_ID_FIELD, monitorId)
+            .field(SCHEMA_VERSION_FIELD, schemaVersion)
+            .field(MONITOR_V2_VERSION_FIELD, monitorVersion)
+            .field(MONITOR_V2_NAME_FIELD, monitorName)
+            .field(EXECUTION_ID_FIELD, executionId)
+            .field(TRIGGER_V2_ID_FIELD, triggerId)
+            .field(TRIGGER_V2_NAME_FIELD, triggerName)
+            .field(QUERY_FIELD, query)
+            .field(QUERY_RESULTS_FIELD, queryResults)
+            .field(ERROR_MESSAGE_FIELD, errorMessage)
+            .field(SEVERITY_FIELD, severity.value)
+            .nonOptionalTimeField(TRIGGERED_TIME_FIELD, triggeredTime)
+
+        if (withUser) {
+            builder.optionalUserField(MONITOR_V2_USER_FIELD, monitorUser)
+        }
+
+        builder.endObject()
+
+        return builder
+    }
+
+    fun asTemplateArg(): Map<String, Any?> {
+        return mapOf(
+            ALERT_V2_ID_FIELD to id,
+            ALERT_V2_VERSION_FIELD to version,
+            ERROR_MESSAGE_FIELD to errorMessage,
+            EXECUTION_ID_FIELD to executionId,
+            SEVERITY_FIELD to severity.value
+        )
+    }
+
+    companion object {
+        const val ALERT_V2_ID_FIELD = "id"
+        const val ALERT_V2_VERSION_FIELD = "version"
+        const val MONITOR_V2_ID_FIELD = "monitor_v2_id"
+        const val MONITOR_V2_VERSION_FIELD = "monitor_v2_version"
+        const val MONITOR_V2_NAME_FIELD = "monitor_v2_name"
+        const val MONITOR_V2_USER_FIELD = "monitor_v2_user"
+        const val TRIGGER_V2_ID_FIELD = "trigger_v2_id"
+        const val TRIGGER_V2_NAME_FIELD = "trigger_v2_name"
+        const val TRIGGERED_TIME_FIELD = "triggered_time"
+        const val QUERY_FIELD = "query"
+        const val QUERY_RESULTS_FIELD = "query_results"
+        const val ERROR_MESSAGE_FIELD = "error_message"
+        const val EXECUTION_ID_FIELD = "execution_id"
+        const val SEVERITY_FIELD = "severity"
+        const val SCHEMA_VERSION_FIELD = "schema_version"
+
+        const val NO_ID = ""
+        const val NO_VERSION = Versions.NOT_FOUND
+        const val NO_SCHEMA_VERSION = 0
+
+        @JvmStatic
+        @JvmOverloads
+        @Throws(IOException::class)
+        fun parse(xcp: XContentParser, id: String = NO_ID, version: Long = NO_VERSION): AlertV2 {
+            var schemaVersion = NO_SCHEMA_VERSION
+            lateinit var monitorId: String
+            lateinit var monitorName: String
+            var monitorVersion: Long = Versions.NOT_FOUND
+            var monitorUser: User? = null
+            lateinit var triggerId: String
+            lateinit var triggerName: String
+            lateinit var query: String
+            var queryResults: Map<String, Any> = mapOf()
+            lateinit var severity: Severity
+            var triggeredTime: Instant? = null
+            var errorMessage: String? = null
+            var executionId: String? = null
+
+            ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.nextToken(), xcp)
+            while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
+                val fieldName = xcp.currentName()
+                xcp.nextToken()
+
+                when (fieldName) {
+                    MONITOR_V2_ID_FIELD -> monitorId = xcp.text()
+                    SCHEMA_VERSION_FIELD -> schemaVersion = xcp.intValue()
+                    MONITOR_V2_NAME_FIELD -> monitorName = xcp.text()
+                    MONITOR_V2_VERSION_FIELD -> monitorVersion = xcp.longValue()
+                    MONITOR_V2_USER_FIELD ->
+                        monitorUser = if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                            null
+                        } else {
+                            User.parse(xcp)
+                        }
+                    TRIGGER_V2_ID_FIELD -> triggerId = xcp.text()
+                    TRIGGER_V2_NAME_FIELD -> triggerName = xcp.text()
+                    QUERY_FIELD -> query = xcp.text()
+                    QUERY_RESULTS_FIELD -> queryResults = xcp.map()
+                    TRIGGERED_TIME_FIELD -> triggeredTime = xcp.instant()
+                    ERROR_MESSAGE_FIELD -> errorMessage = xcp.textOrNull()
+                    EXECUTION_ID_FIELD -> executionId = xcp.textOrNull()
+                    TriggerV2.SEVERITY_FIELD -> {
+                        val input = xcp.text()
+                        val enumMatchResult = Severity.enumFromString(input)
+                            ?: throw IllegalArgumentException(
+                                "Invalid value for ${TriggerV2.SEVERITY_FIELD}: $input. " +
+                                    "Supported values are ${Severity.entries.map { it.value }}"
+                            )
+                        severity = enumMatchResult
+                    }
+                }
+            }
+
+            return AlertV2(
+                id = id,
+                version = version,
+                schemaVersion = schemaVersion,
+                monitorId = requireNotNull(monitorId),
+                monitorName = requireNotNull(monitorName),
+                monitorVersion = monitorVersion,
+                monitorUser = monitorUser,
+                triggerId = requireNotNull(triggerId),
+                triggerName = requireNotNull(triggerName),
+                query = requireNotNull(query),
+                queryResults = requireNotNull(queryResults),
+                triggeredTime = requireNotNull(triggeredTime),
+                errorMessage = errorMessage,
+                severity = severity,
+                executionId = executionId
+            )
+        }
+
+        @JvmStatic
+        @Throws(IOException::class)
+        fun readFrom(sin: StreamInput): AlertV2 {
+            return AlertV2(sin)
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/MonitorV2.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/MonitorV2.kt
new file mode 100644
index 000000000..69485b4c9
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/MonitorV2.kt
@@ -0,0 +1,152 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.modelv2.PPLSQLMonitor.Companion.PPL_SQL_MONITOR_TYPE
+import org.opensearch.common.CheckedFunction
+import org.opensearch.commons.alerting.model.Schedule
+import org.opensearch.commons.alerting.model.ScheduledJob
+import org.opensearch.commons.authuser.User
+import org.opensearch.core.ParseField
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.xcontent.NamedXContentRegistry
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.XContentBuilder
+import org.opensearch.core.xcontent.XContentParser
+import org.opensearch.core.xcontent.XContentParserUtils
+import java.io.IOException
+import java.time.Instant
+
+interface MonitorV2 : ScheduledJob {
+    override val id: String
+    override val version: Long
+    override val name: String
+    override val enabled: Boolean
+    override val schedule: Schedule
+    override val lastUpdateTime: Instant // required for scheduled job maintenance
+    override val enabledTime: Instant? // required for scheduled job maintenance
+    val description: String?
+    val user: User?
+    val triggers: List<TriggerV2>
+    val schemaVersion: Int // for updating monitors
+    val lookBackWindow: Long? // how far back to look when querying data during monitor execution
+    val timestampField: String? // field that will be used to inject lookback window time filter
+
+    fun asTemplateArg(): Map<String, Any?>
+
+    fun toXContentWithUser(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder
+
+    fun makeCopy(
+        id: String = this.id,
+        version: Long = this.version,
+        name: String = this.name,
+        enabled: Boolean = this.enabled,
+        schedule: Schedule = this.schedule,
+        lastUpdateTime: Instant = this.lastUpdateTime,
+        enabledTime: Instant? = this.enabledTime,
+        description: String? = this.description,
+        user: User? = this.user,
+        // no support for overriding triggers in interface-level makeCopy(),
+        // triggers can be copied at instance-level data class copy()
+        schemaVersion: Int = this.schemaVersion,
+        lookBackWindow: Long? = this.lookBackWindow,
+        timestampField: String? = this.timestampField
+    ): MonitorV2
+
+    enum class MonitorV2Type(val value: String) {
+        PPL_SQL_MONITOR(PPL_SQL_MONITOR_TYPE);
+
+        override fun toString(): String {
+            return value
+        }
+
+        companion object {
+            fun enumFromString(value: String): MonitorV2Type? {
+                return MonitorV2Type.entries.find { it.value == value }
+            }
+        }
+    }
+
+    companion object {
+        // scheduled job field names
+        const val MONITOR_V2_TYPE = "monitor_v2" // scheduled job type is MonitorV2
+
+        // field names
+        const val NAME_FIELD = "name"
+        const val ENABLED_FIELD = "enabled"
+        const val SCHEDULE_FIELD = "schedule"
+        const val LAST_UPDATE_TIME_FIELD = "last_update_time"
+        const val ENABLED_TIME_FIELD = "enabled_time"
+        const val DESCRIPTION_FIELD = "description"
+        const val USER_FIELD = "user"
+        const val TRIGGERS_FIELD = "triggers"
+        const val SCHEMA_VERSION_FIELD = "schema_version"
+        const val LOOK_BACK_WINDOW_FIELD = "look_back_window_minutes"
+        const val TIMESTAMP_FIELD = "timestamp_field"
+
+        // default values
+        const val NO_ID = ""
+        const val NO_VERSION = 1L
+
+        // hard, nonadjustable limits
+        const val MONITOR_V2_MAX_TRIGGERS = 10
+        const val MONITOR_V2_MIN_LOOK_BACK_WINDOW = 1L // 1 minute
+        const val ALERTING_V2_MAX_NAME_LENGTH = 30 // max length of any name for monitors, triggers, notif actions, etc
+        const val UUID_LENGTH = 20 // the length of a UUID generated by UUIDs.base64UUID()
+        const val DESCRIPTION_MAX_LENGTH = 2000
+
+        val XCONTENT_REGISTRY = NamedXContentRegistry.Entry(
+            ScheduledJob::class.java,
+            ParseField(MONITOR_V2_TYPE),
+            CheckedFunction { parse(it) }
+        )
+
+        @JvmStatic
+        @Throws(IOException::class)
+        fun parse(xcp: XContentParser, id: String = NO_ID, version: Long = NO_VERSION): MonitorV2 {
+            /* parse outer object for monitorV2 type, then delegate to correct monitorV2 parser */
+
+            XContentParserUtils.ensureExpectedToken( // outer monitor object start
+                XContentParser.Token.START_OBJECT,
+                xcp.currentToken(),
+                xcp
+            )
+
+            // monitor type field name
+            XContentParserUtils.ensureExpectedToken(XContentParser.Token.FIELD_NAME, xcp.nextToken(), xcp)
+            val monitorTypeText = xcp.currentName()
+            val monitorType = MonitorV2Type.enumFromString(monitorTypeText)
+                ?: throw IllegalStateException(
+                    "when parsing MonitorV2, received invalid monitor type: $monitorTypeText. " +
+                        "Please ensure monitor object is wrapped in an outer $PPL_SQL_MONITOR_TYPE object"
+                )
+
+            // inner monitor object start
+            XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.nextToken(), xcp)
+
+            return when (monitorType) {
+                MonitorV2Type.PPL_SQL_MONITOR -> PPLSQLMonitor.parse(xcp, id, version)
+            }
+        }
+
+        fun readFrom(sin: StreamInput): MonitorV2 {
+            return when (val monitorType = sin.readEnum(MonitorV2Type::class.java)) {
+                MonitorV2Type.PPL_SQL_MONITOR -> PPLSQLMonitor(sin)
+                else -> throw IllegalStateException("Unexpected input \"$monitorType\" when reading MonitorV2")
+            }
+        }
+
+        fun writeTo(out: StreamOutput, monitorV2: MonitorV2) {
+            when (monitorV2) {
+                is PPLSQLMonitor -> {
+                    out.writeEnum(MonitorV2Type.PPL_SQL_MONITOR)
+                    monitorV2.writeTo(out)
+                }
+            }
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/MonitorV2RunResult.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/MonitorV2RunResult.kt
new file mode 100644
index 000000000..f6707eb78
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/MonitorV2RunResult.kt
@@ -0,0 +1,43 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.common.io.stream.Writeable
+import org.opensearch.core.xcontent.ToXContent
+
+interface MonitorV2RunResult<out TriggerV2Result : TriggerV2RunResult> : Writeable, ToXContent {
+    val monitorName: String
+    val error: Exception?
+    val triggerResults: Map<String, TriggerV2Result>
+
+    enum class MonitorV2RunResultType() {
+        PPL_SQL_MONITOR_RUN_RESULT;
+    }
+
+    companion object {
+        const val ERROR_FIELD = "error"
+        const val TRIGGER_RESULTS_FIELD = "trigger_results"
+
+        fun readFrom(sin: StreamInput): MonitorV2RunResult<TriggerV2RunResult> {
+            val monitorRunResultType = sin.readEnum(MonitorV2RunResultType::class.java)
+            return when (monitorRunResultType) {
+                MonitorV2RunResultType.PPL_SQL_MONITOR_RUN_RESULT -> PPLSQLMonitorRunResult(sin)
+                else -> throw IllegalStateException("Unexpected input [$monitorRunResultType] when reading MonitorV2RunResult")
+            }
+        }
+
+        fun writeTo(out: StreamOutput, monitorV2RunResult: MonitorV2RunResult<TriggerV2RunResult>) {
+            when (monitorV2RunResult) {
+                is PPLSQLMonitorRunResult -> {
+                    out.writeEnum(MonitorV2RunResultType.PPL_SQL_MONITOR_RUN_RESULT)
+                    monitorV2RunResult.writeTo(out)
+                }
+            }
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLMonitor.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLMonitor.kt
new file mode 100644
index 000000000..edf8e7c4d
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLMonitor.kt
@@ -0,0 +1,408 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.core.util.nonOptionalTimeField
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ALERTING_V2_MAX_NAME_LENGTH
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.DESCRIPTION_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.DESCRIPTION_MAX_LENGTH
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ENABLED_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ENABLED_TIME_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.LAST_UPDATE_TIME_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.LOOK_BACK_WINDOW_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.MONITOR_V2_MAX_TRIGGERS
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.MONITOR_V2_MIN_LOOK_BACK_WINDOW
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.NAME_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.NO_ID
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.NO_VERSION
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.SCHEDULE_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.SCHEMA_VERSION_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.TIMESTAMP_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.TRIGGERS_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.USER_FIELD
+import org.opensearch.commons.alerting.model.CronSchedule
+import org.opensearch.commons.alerting.model.Schedule
+import org.opensearch.commons.alerting.util.AlertingException
+import org.opensearch.commons.alerting.util.IndexUtils
+import org.opensearch.commons.alerting.util.instant
+import org.opensearch.commons.alerting.util.optionalTimeField
+import org.opensearch.commons.alerting.util.optionalUserField
+import org.opensearch.commons.authuser.User
+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 org.opensearch.core.xcontent.XContentParser
+import org.opensearch.core.xcontent.XContentParserUtils
+import java.io.IOException
+import java.time.Instant
+
+/**
+ * PPL/SQL Monitor for OpenSearch Alerting V2
+ *
+ * @property id Monitor ID. Defaults to [NO_ID].
+ * @property version Version number of the monitor. Defaults to [NO_VERSION].
+ * @property name Display name of the monitor.
+ * @property enabled Boolean flag indicating whether the monitor is currently on or off.
+ * @property schedule Defines when and how often the monitor should run. Can be a CRON or interval schedule.
+ * @property lookBackWindow How far back each Monitor execution's query should look back when searching data.
+ * @property lastUpdateTime Timestamp of the last update to this monitor.
+ * @property enabledTime Timestamp when the monitor was last enabled. Null if never enabled.
+ * @property description Optional Monitor description.
+ * @property triggers List of [PPLTrigger]s associated with this monitor.
+ * @property schemaVersion Version of the alerting-config index schema used when this Monitor was indexed. Defaults to [NO_SCHEMA_VERSION].
+ * @property queryLanguage The query language used. Defaults to [QueryLanguage.PPL].
+ * @property query The query string to be executed by this monitor.
+ */
+data class PPLSQLMonitor(
+    override val id: String = NO_ID,
+    override val version: Long = NO_VERSION,
+    override val name: String,
+    override val enabled: Boolean,
+    override val schedule: Schedule,
+    override val lookBackWindow: Long?,
+    override val timestampField: String?,
+    override val lastUpdateTime: Instant,
+    override val enabledTime: Instant?,
+    override val description: String?,
+    override val user: User?,
+    override val triggers: List<PPLSQLTrigger>,
+    override val schemaVersion: Int = IndexUtils.NO_SCHEMA_VERSION,
+    val queryLanguage: QueryLanguage = QueryLanguage.PPL, // default to PPL, SQL not currently supported
+    val query: String
+) : MonitorV2 {
+
+    // specify scheduled job type
+    override val type = MonitorV2.MONITOR_V2_TYPE
+
+    override fun fromDocument(id: String, version: Long): PPLSQLMonitor = copy(id = id, version = version)
+
+    init {
+        // SQL monitors are not yet supported
+        if (this.queryLanguage == QueryLanguage.SQL) {
+            throw IllegalArgumentException("SQL queries are not supported. Please use a PPL query.")
+        }
+
+        require(this.name.length <= ALERTING_V2_MAX_NAME_LENGTH) {
+            "Monitor name too long, length must be less than $ALERTING_V2_MAX_NAME_LENGTH."
+        }
+
+        if (this.lookBackWindow != null) {
+            requireNotNull(this.timestampField) { "If look back window is specified, timestamp field must not be null." }
+        } else {
+            require(this.timestampField == null) { "If look back window is not specified, timestamp field must not be specified." }
+        }
+
+        require(this.triggers.isNotEmpty()) { "Monitor must include at least 1 trigger." }
+        require(this.triggers.size <= MONITOR_V2_MAX_TRIGGERS) { "Monitors can only have $MONITOR_V2_MAX_TRIGGERS triggers." }
+
+        lookBackWindow?.let {
+            require(this.lookBackWindow >= MONITOR_V2_MIN_LOOK_BACK_WINDOW) {
+                "Monitors look back windows must be at least $MONITOR_V2_MIN_LOOK_BACK_WINDOW minute."
+            }
+        }
+
+        this.description?.let {
+            require(this.description.length <= DESCRIPTION_MAX_LENGTH) { "Description must be under $DESCRIPTION_MAX_LENGTH characters." }
+        }
+
+        // for checking trigger ID uniqueness
+        val triggerIds = mutableSetOf<String>()
+        this.triggers.forEach { trigger ->
+            require(triggerIds.add(trigger.id)) { "Duplicate trigger id: ${trigger.id}. Trigger ids must be unique." }
+        }
+
+        if (this.enabled) {
+            requireNotNull(this.enabledTime)
+        } else {
+            require(this.enabledTime == null)
+        }
+    }
+
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        id = sin.readString(),
+        version = sin.readLong(),
+        name = sin.readString(),
+        enabled = sin.readBoolean(),
+        schedule = Schedule.readFrom(sin),
+        lookBackWindow = sin.readOptionalLong(),
+        timestampField = sin.readOptionalString(),
+        lastUpdateTime = sin.readInstant(),
+        enabledTime = sin.readOptionalInstant(),
+        description = sin.readOptionalString(),
+        user = if (sin.readBoolean()) {
+            User(sin)
+        } else {
+            null
+        },
+        triggers = sin.readList(PPLSQLTrigger.Companion::readFrom),
+        schemaVersion = sin.readInt(),
+        queryLanguage = sin.readEnum(QueryLanguage::class.java),
+        query = sin.readString()
+    )
+
+    override fun toXContentWithUser(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        return createXContentBuilder(builder, params, true)
+    }
+
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        return createXContentBuilder(builder, params, false)
+    }
+
+    private fun createXContentBuilder(builder: XContentBuilder, params: ToXContent.Params, withUser: Boolean): XContentBuilder {
+        builder.startObject() // overall start object
+
+        // if this is being written as ScheduledJob, add extra object layer and add ScheduledJob
+        // related metadata, default to false
+        if (params.paramAsBoolean("with_type", false)) {
+            builder.startObject(MonitorV2.MONITOR_V2_TYPE)
+        }
+
+        // wrap PPLSQLMonitor in outer object named after its monitor type
+        // required for MonitorV2 XContentParser to first encounter this,
+        // read in monitor type, then delegate to correct parse() function
+        builder.startObject(PPL_SQL_MONITOR_TYPE) // monitor type start object
+
+        builder.field(NAME_FIELD, name)
+        builder.field(SCHEDULE_FIELD, schedule)
+        builder.field(LOOK_BACK_WINDOW_FIELD, lookBackWindow)
+        builder.field(TIMESTAMP_FIELD, timestampField)
+        builder.field(ENABLED_FIELD, enabled)
+        builder.nonOptionalTimeField(LAST_UPDATE_TIME_FIELD, lastUpdateTime)
+        builder.optionalTimeField(ENABLED_TIME_FIELD, enabledTime)
+        builder.field(DESCRIPTION_FIELD, description)
+        builder.field(TRIGGERS_FIELD, triggers.toTypedArray())
+        builder.field(SCHEMA_VERSION_FIELD, schemaVersion)
+        builder.field(QUERY_LANGUAGE_FIELD, queryLanguage.value)
+        builder.field(QUERY_FIELD, query)
+
+        if (withUser) {
+            builder.optionalUserField(USER_FIELD, user)
+        }
+
+        builder.endObject() // monitor type end object
+
+        // if ScheduledJob metadata was added, end the extra object layer that was created
+        if (params.paramAsBoolean("with_type", false)) {
+            builder.endObject()
+        }
+
+        builder.endObject() // overall end object
+
+        return builder
+    }
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(id)
+        out.writeLong(version)
+        out.writeString(name)
+        out.writeBoolean(enabled)
+
+        if (schedule is CronSchedule) {
+            out.writeEnum(Schedule.TYPE.CRON)
+        } else {
+            out.writeEnum(Schedule.TYPE.INTERVAL)
+        }
+        schedule.writeTo(out)
+
+        out.writeOptionalLong(lookBackWindow)
+        out.writeOptionalString(timestampField)
+        out.writeInstant(lastUpdateTime)
+        out.writeOptionalInstant(enabledTime)
+        out.writeOptionalString(description)
+
+        out.writeBoolean(user != null)
+        user?.writeTo(out)
+
+        out.writeVInt(triggers.size)
+        triggers.forEach { it.writeTo(out) }
+        out.writeInt(schemaVersion)
+        out.writeEnum(queryLanguage)
+        out.writeString(query)
+    }
+
+    override fun asTemplateArg(): Map<String, Any?> {
+        return mapOf(
+            IndexUtils._ID to id,
+            IndexUtils._VERSION to version,
+            NAME_FIELD to name,
+            ENABLED_FIELD to enabled,
+            SCHEDULE_FIELD to schedule,
+            LOOK_BACK_WINDOW_FIELD to lookBackWindow,
+            LAST_UPDATE_TIME_FIELD to lastUpdateTime.toEpochMilli(),
+            ENABLED_TIME_FIELD to enabledTime?.toEpochMilli(),
+            QUERY_FIELD to query
+        )
+    }
+
+    override fun makeCopy(
+        id: String,
+        version: Long,
+        name: String,
+        enabled: Boolean,
+        schedule: Schedule,
+        lastUpdateTime: Instant,
+        enabledTime: Instant?,
+        description: String?,
+        user: User?,
+        schemaVersion: Int,
+        lookBackWindow: Long?,
+        timestampField: String?
+    ): PPLSQLMonitor {
+        return copy(
+            id = id,
+            version = version,
+            name = name,
+            enabled = enabled,
+            schedule = schedule,
+            lastUpdateTime = lastUpdateTime,
+            enabledTime = enabledTime,
+            description = description,
+            user = user,
+            schemaVersion = schemaVersion,
+            lookBackWindow = lookBackWindow,
+            timestampField = timestampField
+        )
+    }
+
+    enum class QueryLanguage(val value: String) {
+        PPL(PPL_QUERY_LANGUAGE),
+        SQL(SQL_QUERY_LANGUAGE);
+
+        companion object {
+            fun enumFromString(value: String): QueryLanguage? = QueryLanguage.entries.firstOrNull { it.value == value }
+        }
+    }
+
+    companion object {
+        // monitor type name
+        const val PPL_SQL_MONITOR_TYPE = "ppl_monitor"
+
+        // query languages
+        const val PPL_QUERY_LANGUAGE = "ppl"
+        const val SQL_QUERY_LANGUAGE = "sql"
+
+        // field names
+        const val QUERY_LANGUAGE_FIELD = "query_language"
+        const val QUERY_FIELD = "query"
+
+        @JvmStatic
+        @JvmOverloads
+        @Throws(IOException::class)
+        fun parse(xcp: XContentParser, id: String = NO_ID, version: Long = NO_VERSION): PPLSQLMonitor {
+            var name: String? = null
+            var enabled = true
+            var schedule: Schedule? = null
+            var lookBackWindow: Long? = null
+            var timestampField: String? = null
+            var lastUpdateTime: Instant? = null
+            var enabledTime: Instant? = null
+            var description: String? = null
+            var user: User? = null
+            val triggers: MutableList<PPLSQLTrigger> = mutableListOf()
+            var schemaVersion = IndexUtils.NO_SCHEMA_VERSION
+            var queryLanguage: QueryLanguage = QueryLanguage.PPL // default to PPL
+            var query: String? = null
+
+            /* parse */
+            XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
+            while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
+                val fieldName = xcp.currentName()
+                xcp.nextToken()
+
+                when (fieldName) {
+                    NAME_FIELD -> name = xcp.text()
+                    ENABLED_FIELD -> enabled = xcp.booleanValue()
+                    SCHEDULE_FIELD -> schedule = Schedule.parse(xcp)
+                    LOOK_BACK_WINDOW_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            lookBackWindow = xcp.longValue()
+                        }
+                    }
+                    TIMESTAMP_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            timestampField = xcp.text()
+                        }
+                    }
+                    LAST_UPDATE_TIME_FIELD -> lastUpdateTime = xcp.instant()
+                    ENABLED_TIME_FIELD -> enabledTime = xcp.instant()
+                    DESCRIPTION_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            description = xcp.text()
+                        }
+                    }
+                    USER_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            user = User.parse(xcp)
+                        }
+                    }
+                    TRIGGERS_FIELD -> {
+                        XContentParserUtils.ensureExpectedToken(
+                            XContentParser.Token.START_ARRAY,
+                            xcp.currentToken(),
+                            xcp
+                        )
+                        while (xcp.nextToken() != XContentParser.Token.END_ARRAY) {
+                            triggers.add(PPLSQLTrigger.parseInner(xcp))
+                        }
+                    }
+                    SCHEMA_VERSION_FIELD -> schemaVersion = xcp.intValue()
+                    QUERY_LANGUAGE_FIELD -> {
+                        val input = xcp.text()
+                        val enumMatchResult = QueryLanguage.enumFromString(input)
+                            ?: throw AlertingException.wrap(
+                                IllegalArgumentException(
+                                    "Invalid value for $QUERY_LANGUAGE_FIELD: $input. " +
+                                        "Supported values are ${QueryLanguage.entries.map { it.value }}"
+                                )
+                            )
+                        queryLanguage = enumMatchResult
+                    }
+                    QUERY_FIELD -> query = xcp.text()
+                    else -> throw IllegalArgumentException("Unexpected field when parsing PPL/SQL Monitor: $fieldName")
+                }
+            }
+
+            /* validations */
+
+            // if enabled, set time of MonitorV2 creation/update is set as enable time
+            if (enabled && enabledTime == null) {
+                enabledTime = Instant.now()
+            } else if (!enabled) {
+                enabledTime = null
+            }
+
+            lastUpdateTime = lastUpdateTime ?: Instant.now()
+
+            // check for required fields
+            requireNotNull(name) { "Monitor name is null" }
+            requireNotNull(schedule) { "Schedule is null" }
+            requireNotNull(query) { "Query is null" }
+            requireNotNull(lastUpdateTime) { "Last update time is null" }
+
+            /* return PPLSQLMonitor */
+            return PPLSQLMonitor(
+                id,
+                version,
+                name,
+                enabled,
+                schedule,
+                lookBackWindow,
+                timestampField,
+                lastUpdateTime,
+                enabledTime,
+                description,
+                user,
+                triggers,
+                schemaVersion,
+                queryLanguage,
+                query
+            )
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLMonitorRunResult.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLMonitorRunResult.kt
new file mode 100644
index 000000000..853ec58b8
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLMonitorRunResult.kt
@@ -0,0 +1,54 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.modelv2.AlertV2.Companion.MONITOR_V2_NAME_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2RunResult.Companion.ERROR_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2RunResult.Companion.TRIGGER_RESULTS_FIELD
+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
+
+data class PPLSQLMonitorRunResult(
+    override val monitorName: String,
+    override val error: Exception?,
+    override val triggerResults: Map<String, PPLSQLTriggerRunResult>,
+    val pplQueryResults: Map<String, Map<String, Any?>> // key: trigger id, value: query results
+) : MonitorV2RunResult<PPLSQLTriggerRunResult> {
+
+    @Throws(IOException::class)
+    @Suppress("UNCHECKED_CAST")
+    constructor(sin: StreamInput) : this(
+        sin.readString(), // monitorName
+        sin.readException(), // error
+        sin.readMap() as Map<String, PPLSQLTriggerRunResult>, // triggerResults
+        sin.readMap() as Map<String, Map<String, Any?>> // pplQueryResults
+    )
+
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        builder.startObject()
+        builder.field(MONITOR_V2_NAME_FIELD, monitorName)
+        builder.field(ERROR_FIELD, error?.message)
+        builder.field(TRIGGER_RESULTS_FIELD, triggerResults)
+        builder.field(PPL_QUERY_RESULTS_FIELD, pplQueryResults)
+        builder.endObject()
+        return builder
+    }
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(monitorName)
+        out.writeException(error)
+        out.writeMap(triggerResults)
+        out.writeMap(pplQueryResults)
+    }
+
+    companion object {
+        const val PPL_QUERY_RESULTS_FIELD = "ppl_query_results"
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLTrigger.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLTrigger.kt
new file mode 100644
index 000000000..c5c058310
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLTrigger.kt
@@ -0,0 +1,401 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ALERTING_V2_MAX_NAME_LENGTH
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.UUID_LENGTH
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.ACTIONS_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.DEFAULT_EXPIRE_DURATION
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.EXPIRE_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.ID_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.LAST_TRIGGERED_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.MONITOR_V2_MIN_EXPIRE_DURATION_MINUTES
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.MONITOR_V2_MIN_THROTTLE_DURATION_MINUTES
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.NAME_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.NOTIFICATIONS_ID_MAX_LENGTH
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.SEVERITY_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.THROTTLE_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Severity
+import org.opensearch.common.CheckedFunction
+import org.opensearch.common.UUIDs
+import org.opensearch.commons.alerting.model.action.Action
+import org.opensearch.commons.alerting.util.instant
+import org.opensearch.commons.alerting.util.optionalTimeField
+import org.opensearch.core.ParseField
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.core.common.io.stream.StreamOutput
+import org.opensearch.core.xcontent.NamedXContentRegistry
+import org.opensearch.core.xcontent.ToXContent
+import org.opensearch.core.xcontent.XContentBuilder
+import org.opensearch.core.xcontent.XContentParser
+import org.opensearch.core.xcontent.XContentParserUtils
+import java.io.IOException
+import java.time.Instant
+
+/**
+ * The PPL/SQL Trigger for PPL/SQL Monitors
+ *
+ * There are two types of PPLTrigger conditions: NUMBER_OF_RESULTS and CUSTOM
+ * NUMBER_OF_RESULTS: triggers based on whether the number of query results returned by the PPLSQLMonitor
+ *                    query meets some threshold
+ * CUSTOM: triggers based on a custom condition that user specifies (a single ppl eval statement)
+ *
+ * PPLTriggers can run on two modes: RESULT_SET and PER_RESULT
+ * RESULT_SET: exactly one Alert is generated when the Trigger condition is met
+ * PER_RESULT: one Alert is generated per trigger condition-meeting query result row
+ *
+ * @property id Trigger ID, defaults to a base64 UUID.
+ * @property name Display name of the Trigger.
+ * @property severity The severity level of the Trigger.
+ * @property throttleDuration Optional duration (in minutes) for which alerts from this Trigger should be throttled/suppressed.
+ *                            Null indicates no throttling.
+ * @property expireDuration Duration (in minutes) after which alerts from this Trigger should be deleted permanently.
+ * @property lastTriggeredTime The last time this Trigger generated an Alert. Null if Trigger hasn't generated an Alert yet.
+ * @property actions List of notification-sending actions to run when the Trigger condition is met.
+ * @property mode Specifies whether the trigger evaluates the entire result set or each result individually.
+ *               Can be either [TriggerMode.RESULT_SET] or [TriggerMode.PER_RESULT].
+ * @property conditionType The type of condition to evaluate.
+ *               Can be either [ConditionType.NUMBER_OF_RESULTS] or [ConditionType.CUSTOM].
+ * @property numResultsCondition The comparison operator for NUMBER_OF_RESULTS conditions. Required if using NUMBER_OF_RESULTS conditions,
+ *                               required to be null otherwise.
+ * @property numResultsValue The threshold value for NUMBER_OF_RESULTS conditions. Required if using NUMBER_OF_RESULTS conditions,
+ *                           required to be null otherwise.
+ * @property customCondition A custom condition expression. Required if using CUSTOM conditions,
+ *                           required to be null otherwise.
+ */
+data class PPLSQLTrigger(
+    override val id: String = UUIDs.base64UUID(),
+    override val name: String,
+    override val severity: Severity,
+    override val throttleDuration: Long?,
+    override val expireDuration: Long = DEFAULT_EXPIRE_DURATION,
+    override var lastTriggeredTime: Instant?,
+    override val actions: List<Action>,
+    val mode: TriggerMode, // RESULT_SET or PER_RESULT
+    val conditionType: ConditionType, // NUMBER_OF_RESULTS or CUSTOM
+    val numResultsCondition: NumResultsCondition?,
+    val numResultsValue: Long?,
+    val customCondition: String?
+) : TriggerV2 {
+
+    init {
+        requireNotNull(this.name) { "Trigger name must be included." }
+        requireNotNull(this.severity) { "Trigger severity must be included." }
+        requireNotNull(this.mode) { "Trigger mode must be included." }
+        requireNotNull(this.conditionType) { "Trigger condition type must be included." }
+
+        require(this.id.length <= UUID_LENGTH) {
+            "Trigger ID too long, length must be less than $UUID_LENGTH."
+        }
+
+        require(this.name.length <= ALERTING_V2_MAX_NAME_LENGTH) {
+            "Trigger name too long, length must be less than $ALERTING_V2_MAX_NAME_LENGTH."
+        }
+
+        require(this.expireDuration >= MONITOR_V2_MIN_EXPIRE_DURATION_MINUTES) {
+            "expire duration cannot be less than $MONITOR_V2_MIN_EXPIRE_DURATION_MINUTES, was $expireDuration."
+        }
+
+        this.throttleDuration?.let {
+            require(it >= MONITOR_V2_MIN_THROTTLE_DURATION_MINUTES) {
+                "Throttle duration cannot be less than $MONITOR_V2_MIN_THROTTLE_DURATION_MINUTES, was $throttleDuration."
+            }
+        }
+
+        this.actions.forEach {
+            require(it.name.length <= ALERTING_V2_MAX_NAME_LENGTH) {
+                "Name of action with ID ${it.id} too long, length must be less than $ALERTING_V2_MAX_NAME_LENGTH."
+            }
+            require(it.destinationId.length <= NOTIFICATIONS_ID_MAX_LENGTH) {
+                "Channel ID of action with ID ${it.id} too long, length must be less than $NOTIFICATIONS_ID_MAX_LENGTH."
+            }
+        }
+
+        when (this.conditionType) {
+            ConditionType.NUMBER_OF_RESULTS -> {
+                requireNotNull(this.numResultsCondition) {
+                    "if trigger condition is of type ${ConditionType.NUMBER_OF_RESULTS.value}, " +
+                        "$NUM_RESULTS_CONDITION_FIELD must be included."
+                }
+                requireNotNull(this.numResultsValue) {
+                    "if trigger condition is of type ${ConditionType.NUMBER_OF_RESULTS.value}, " +
+                        "$NUM_RESULTS_VALUE_FIELD must be included."
+                }
+                require(this.customCondition == null) {
+                    "if trigger condition is of type ${ConditionType.NUMBER_OF_RESULTS.value}, " +
+                        "$CUSTOM_CONDITION_FIELD must not be included."
+                }
+            }
+            ConditionType.CUSTOM -> {
+                requireNotNull(this.customCondition) {
+                    "if trigger condition is of type ${ConditionType.CUSTOM.value}, " +
+                        "$CUSTOM_CONDITION_FIELD must be included."
+                }
+                require(this.numResultsCondition == null) {
+                    "if trigger condition is of type ${ConditionType.CUSTOM.value}, " +
+                        "$NUM_RESULTS_CONDITION_FIELD must not be included."
+                }
+                require(this.numResultsValue == null) {
+                    "if trigger condition is of type ${ConditionType.CUSTOM.value}, " +
+                        "$NUM_RESULTS_VALUE_FIELD must not be included."
+                }
+            }
+        }
+
+        if (conditionType == ConditionType.NUMBER_OF_RESULTS) {
+            require(this.numResultsValue!! >= 0L) { "Number of results to check for cannot be negative." }
+        }
+    }
+
+    @Throws(IOException::class)
+    constructor(sin: StreamInput) : this(
+        sin.readString(), // id
+        sin.readString(), // name
+        sin.readEnum(Severity::class.java), // severity
+        sin.readOptionalLong(), // throttleDuration
+        sin.readLong(), // expireDuration
+        sin.readOptionalInstant(), // lastTriggeredTime
+        sin.readList(::Action), // actions
+        sin.readEnum(TriggerMode::class.java), // trigger mode
+        sin.readEnum(ConditionType::class.java), // condition type
+        if (sin.readBoolean()) sin.readEnum(NumResultsCondition::class.java) else null, // num results condition
+        sin.readOptionalLong(), // num results value
+        sin.readOptionalString() // custom condition
+    )
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(id)
+        out.writeString(name)
+        out.writeEnum(severity)
+        out.writeOptionalLong(throttleDuration)
+        out.writeLong(expireDuration)
+        out.writeOptionalInstant(lastTriggeredTime)
+        out.writeCollection(actions)
+        out.writeEnum(mode)
+        out.writeEnum(conditionType)
+
+        out.writeBoolean(numResultsCondition != null)
+        numResultsCondition?.let { out.writeEnum(numResultsCondition) }
+
+        out.writeOptionalLong(numResultsValue)
+        out.writeOptionalString(customCondition)
+    }
+
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params?): XContentBuilder {
+        builder.startObject()
+        builder.field(ID_FIELD, id)
+        builder.field(NAME_FIELD, name)
+        builder.field(SEVERITY_FIELD, severity.value)
+        throttleDuration?.let { builder.field(THROTTLE_FIELD, throttleDuration) }
+        builder.field(EXPIRE_FIELD, expireDuration)
+        builder.optionalTimeField(LAST_TRIGGERED_FIELD, lastTriggeredTime)
+        builder.field(ACTIONS_FIELD, actions.toTypedArray())
+        builder.field(MODE_FIELD, mode.value)
+        builder.field(CONDITION_TYPE_FIELD, conditionType.value)
+        numResultsCondition?.let { builder.field(NUM_RESULTS_CONDITION_FIELD, numResultsCondition.value) }
+        numResultsValue?.let { builder.field(NUM_RESULTS_VALUE_FIELD, numResultsValue) }
+        customCondition?.let { builder.field(CUSTOM_CONDITION_FIELD, customCondition) }
+        builder.endObject()
+        return builder
+    }
+
+    fun asTemplateArg(): Map<String, Any?> {
+        return mapOf(
+            ID_FIELD to id,
+            NAME_FIELD to name,
+            SEVERITY_FIELD to severity.value,
+            THROTTLE_FIELD to throttleDuration,
+            EXPIRE_FIELD to expireDuration,
+            ACTIONS_FIELD to actions.map { it.asTemplateArg() },
+            MODE_FIELD to mode.value,
+            CONDITION_TYPE_FIELD to conditionType.value,
+            NUM_RESULTS_CONDITION_FIELD to numResultsCondition?.value,
+            NUM_RESULTS_VALUE_FIELD to numResultsValue,
+            CUSTOM_CONDITION_FIELD to customCondition
+        )
+    }
+
+    enum class TriggerMode(val value: String) {
+        RESULT_SET("result_set"),
+        PER_RESULT("per_result");
+
+        companion object {
+            fun enumFromString(value: String): TriggerMode? = entries.firstOrNull { it.value == value }
+        }
+    }
+
+    enum class ConditionType(val value: String) {
+        NUMBER_OF_RESULTS("number_of_results"),
+        CUSTOM("custom");
+
+        companion object {
+            fun enumFromString(value: String): ConditionType? = entries.firstOrNull { it.value == value }
+        }
+    }
+
+    enum class NumResultsCondition(val value: String) {
+        GREATER_THAN(">"),
+        GREATER_THAN_EQUAL(">="),
+        LESS_THAN("<"),
+        LESS_THAN_EQUAL("<="),
+        EQUAL("=="),
+        NOT_EQUAL("!=");
+
+        companion object {
+            fun enumFromString(value: String): NumResultsCondition? = entries.firstOrNull { it.value == value }
+        }
+    }
+
+    companion object {
+        // trigger wrapper object field name
+        const val PPL_SQL_TRIGGER_FIELD = "ppl_sql_trigger"
+
+        // field names
+        const val MODE_FIELD = "mode"
+        const val CONDITION_TYPE_FIELD = "type"
+        const val NUM_RESULTS_CONDITION_FIELD = "num_results_condition"
+        const val NUM_RESULTS_VALUE_FIELD = "num_results_value"
+        const val CUSTOM_CONDITION_FIELD = "custom_condition"
+
+        val XCONTENT_REGISTRY = NamedXContentRegistry.Entry(
+            TriggerV2::class.java,
+            ParseField(PPL_SQL_TRIGGER_FIELD),
+            CheckedFunction { parseInner(it) }
+        )
+
+        @JvmStatic
+        @Throws(IOException::class)
+        fun parseInner(xcp: XContentParser): PPLSQLTrigger {
+            var id = UUIDs.base64UUID() // assign a default triggerId if one is not specified
+            var name: String? = null
+            var severity: Severity? = null
+            var throttleDuration: Long? = null
+            var expireDuration: Long = DEFAULT_EXPIRE_DURATION
+            var lastTriggeredTime: Instant? = null
+            val actions: MutableList<Action> = mutableListOf()
+            var mode: TriggerMode? = null
+            var conditionType: ConditionType? = null
+            var numResultsCondition: NumResultsCondition? = null
+            var numResultsValue: Long? = null
+            var customCondition: String? = null
+
+            /* parse */
+            XContentParserUtils.ensureExpectedToken( // outer trigger object start
+                XContentParser.Token.START_OBJECT,
+                xcp.currentToken(), xcp
+            )
+
+            while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
+                val fieldName = xcp.currentName()
+                xcp.nextToken()
+
+                when (fieldName) {
+                    ID_FIELD -> id = xcp.text()
+                    NAME_FIELD -> name = xcp.text()
+                    SEVERITY_FIELD -> {
+                        val input = xcp.text()
+                        val enumMatchResult = Severity.enumFromString(input)
+                            ?: throw IllegalArgumentException(
+                                "Invalid value for $SEVERITY_FIELD: $input. " +
+                                    "Supported values are ${Severity.entries.map { it.value }}"
+                            )
+                        severity = enumMatchResult
+                    }
+                    MODE_FIELD -> {
+                        val input = xcp.text()
+                        val enumMatchResult = TriggerMode.enumFromString(input)
+                            ?: throw IllegalArgumentException(
+                                "Invalid value for $MODE_FIELD: $input. " +
+                                    "Supported values are ${TriggerMode.entries.map { it.value }}"
+                            )
+                        mode = enumMatchResult
+                    }
+                    CONDITION_TYPE_FIELD -> {
+                        val input = xcp.text()
+                        val enumMatchResult = ConditionType.enumFromString(input)
+                            ?: throw IllegalArgumentException(
+                                "Invalid value for $CONDITION_TYPE_FIELD: $input. " +
+                                    "Supported values are ${ConditionType.entries.map { it.value }}"
+                            )
+                        conditionType = enumMatchResult
+                    }
+                    NUM_RESULTS_CONDITION_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            val input = xcp.text()
+                            val enumMatchResult = NumResultsCondition.enumFromString(input)
+                                ?: throw IllegalArgumentException(
+                                    "Invalid value for $NUM_RESULTS_CONDITION_FIELD: $input. " +
+                                        "Supported values are ${NumResultsCondition.entries.map { it.value }}"
+                                )
+                            numResultsCondition = enumMatchResult
+                        }
+                    }
+                    NUM_RESULTS_VALUE_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            numResultsValue = xcp.longValue()
+                        }
+                    }
+                    CUSTOM_CONDITION_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            customCondition = xcp.text()
+                        }
+                    }
+                    THROTTLE_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            throttleDuration = xcp.longValue()
+                        }
+                    }
+                    EXPIRE_FIELD -> {
+                        if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                            expireDuration = xcp.longValue()
+                        }
+                    }
+                    LAST_TRIGGERED_FIELD -> lastTriggeredTime = xcp.instant()
+                    ACTIONS_FIELD -> {
+                        XContentParserUtils.ensureExpectedToken(
+                            XContentParser.Token.START_ARRAY,
+                            xcp.currentToken(),
+                            xcp
+                        )
+                        while (xcp.nextToken() != XContentParser.Token.END_ARRAY) {
+                            actions.add(Action.parse(xcp))
+                        }
+                    }
+                    else -> throw IllegalArgumentException("Unexpected field when parsing PPL Trigger: $fieldName")
+                }
+            }
+
+            /* validations */
+            requireNotNull(name) { "Trigger name must be included" }
+            requireNotNull(severity) { "Trigger severity must be included" }
+            requireNotNull(mode) { "Trigger mode must be included" }
+            requireNotNull(conditionType) { "Trigger condition type must be included" }
+
+            // 3. prepare and return PPLTrigger object
+            return PPLSQLTrigger(
+                id,
+                name,
+                severity,
+                throttleDuration,
+                expireDuration,
+                lastTriggeredTime,
+                actions,
+                mode,
+                conditionType,
+                numResultsCondition,
+                numResultsValue,
+                customCondition
+            )
+        }
+
+        @JvmStatic
+        @Throws(IOException::class)
+        fun readFrom(sin: StreamInput): PPLSQLTrigger {
+            return PPLSQLTrigger(sin)
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLTriggerRunResult.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLTriggerRunResult.kt
new file mode 100644
index 000000000..70ea28a35
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/PPLSQLTriggerRunResult.kt
@@ -0,0 +1,54 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.modelv2.TriggerV2RunResult.Companion.ERROR_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2RunResult.Companion.NAME_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2RunResult.Companion.TRIGGERED_FIELD
+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
+
+data class PPLSQLTriggerRunResult(
+    override var triggerName: String,
+    override var triggered: Boolean,
+    override var error: Exception?,
+) : TriggerV2RunResult {
+
+    @Throws(IOException::class)
+    @Suppress("UNCHECKED_CAST")
+    constructor(sin: StreamInput) : this(
+        triggerName = sin.readString(),
+        triggered = sin.readBoolean(),
+        error = sin.readException()
+    )
+
+    override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
+        builder.startObject()
+        builder.field(NAME_FIELD, triggerName)
+        builder.field(TRIGGERED_FIELD, triggered)
+        builder.field(ERROR_FIELD, error?.message)
+        builder.endObject()
+        return builder
+    }
+
+    @Throws(IOException::class)
+    override fun writeTo(out: StreamOutput) {
+        out.writeString(triggerName)
+        out.writeBoolean(triggered)
+        out.writeException(error)
+    }
+
+    companion object {
+        @JvmStatic
+        @Throws(IOException::class)
+        fun readFrom(sin: StreamInput): TriggerV2RunResult {
+            return PPLSQLTriggerRunResult(sin)
+        }
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/TriggerV2.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/TriggerV2.kt
new file mode 100644
index 000000000..14e726800
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/TriggerV2.kt
@@ -0,0 +1,64 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.Companion.PPL_SQL_TRIGGER_FIELD
+import org.opensearch.commons.alerting.model.action.Action
+import org.opensearch.commons.notifications.model.BaseModel
+import java.time.Instant
+
+interface TriggerV2 : BaseModel {
+
+    val id: String
+    val name: String
+    val severity: Severity
+    val throttleDuration: Long?
+    val expireDuration: Long
+    var lastTriggeredTime: Instant?
+    val actions: List<Action>
+
+    enum class TriggerV2Type(val value: String) {
+        PPL_TRIGGER(PPL_SQL_TRIGGER_FIELD);
+
+        override fun toString(): String {
+            return value
+        }
+    }
+
+    enum class Severity(val value: String) {
+        INFO("info"),
+        ERROR("error"),
+        LOW("low"),
+        MEDIUM("medium"),
+        HIGH("high"),
+        CRITICAL("critical");
+
+        companion object {
+            fun enumFromString(value: String): Severity? {
+                return entries.find { it.value == value }
+            }
+        }
+    }
+
+    companion object {
+        // field names
+        const val ID_FIELD = "id"
+        const val NAME_FIELD = "name"
+        const val SEVERITY_FIELD = "severity"
+        const val THROTTLE_FIELD = "throttle_minutes"
+        const val LAST_TRIGGERED_FIELD = "last_triggered_time"
+        const val EXPIRE_FIELD = "expires_minutes"
+        const val ACTIONS_FIELD = "actions"
+
+        // hard, nonadjustable limits
+        const val MONITOR_V2_MIN_THROTTLE_DURATION_MINUTES = 1L // one minute min duration to match scheduled job interval granularity
+        const val MONITOR_V2_MIN_EXPIRE_DURATION_MINUTES = 1L // one minute min duration to match scheduled job interval granularity
+        const val NOTIFICATIONS_ID_MAX_LENGTH = 512 // length limit for notifications channel custom ID at channel creation time
+
+        // default fallback values of fields if none are passed in
+        const val DEFAULT_EXPIRE_DURATION = (7 * 24 * 60).toLong() // 7 days in minutes
+    }
+}
diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/TriggerV2RunResult.kt b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/TriggerV2RunResult.kt
new file mode 100644
index 000000000..5a09e4b7c
--- /dev/null
+++ b/alerting/src/main/kotlin/org/opensearch/alerting/modelv2/TriggerV2RunResult.kt
@@ -0,0 +1,22 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.core.common.io.stream.Writeable
+import org.opensearch.core.xcontent.ToXContent
+
+interface TriggerV2RunResult : Writeable, ToXContent {
+
+    val triggerName: String
+    val triggered: Boolean
+    val error: Exception?
+
+    companion object {
+        const val NAME_FIELD = "name"
+        const val TRIGGERED_FIELD = "triggered"
+        const val ERROR_FIELD = "error"
+    }
+}
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt b/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt
index 61276abde..6f8c05d09 100644
--- a/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/TestHelpers.kt
@@ -12,6 +12,14 @@ import org.opensearch.alerting.model.AlertContext
 import org.opensearch.alerting.model.destination.email.EmailAccount
 import org.opensearch.alerting.model.destination.email.EmailEntry
 import org.opensearch.alerting.model.destination.email.EmailGroup
+import org.opensearch.alerting.modelv2.AlertV2
+import org.opensearch.alerting.modelv2.PPLSQLMonitor
+import org.opensearch.alerting.modelv2.PPLSQLMonitor.QueryLanguage
+import org.opensearch.alerting.modelv2.PPLSQLTrigger
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.ConditionType
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.NumResultsCondition
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.TriggerMode
+import org.opensearch.alerting.modelv2.TriggerV2.Severity
 import org.opensearch.alerting.util.getBucketKeysHash
 import org.opensearch.client.Request
 import org.opensearch.client.RequestOptions
@@ -79,10 +87,18 @@ import org.opensearch.search.builder.SearchSourceBuilder
 import org.opensearch.test.OpenSearchTestCase.randomBoolean
 import org.opensearch.test.OpenSearchTestCase.randomInt
 import org.opensearch.test.OpenSearchTestCase.randomIntBetween
+import org.opensearch.test.OpenSearchTestCase.randomLongBetween
 import org.opensearch.test.rest.OpenSearchRestTestCase
+import org.opensearch.test.rest.OpenSearchRestTestCase.assertEquals
 import java.time.Instant
 import java.time.temporal.ChronoUnit
 
+// constants for PPL Alerting tests
+const val TIMESTAMP_FIELD = "timestamp"
+const val TEST_INDEX_NAME = "index"
+const val TEST_INDEX_MAPPINGS =
+    """"properties":{"timestamp":{"type":"date"},"abc":{"type":"keyword"},"number":{"type":"integer"}}"""
+
 fun randomQueryLevelMonitor(
     name: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
     user: User = randomUser(),
@@ -292,6 +308,36 @@ fun randomWorkflowWithDelegates(
     )
 }
 
+fun randomPPLMonitor(
+    name: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
+    enabled: Boolean = randomBoolean(),
+    schedule: Schedule = IntervalSchedule(interval = 5, unit = ChronoUnit.MINUTES),
+    lookBackWindow: Long? = randomLongBetween(10, 100),
+    timestampField: String? = lookBackWindow?.let { TIMESTAMP_FIELD },
+    lastUpdateTime: Instant = Instant.now().truncatedTo(ChronoUnit.MILLIS),
+    enabledTime: Instant? = if (enabled) Instant.now().truncatedTo(ChronoUnit.MILLIS) else null,
+    description: String? = if (randomBoolean()) "some description" else null,
+    triggers: List<PPLSQLTrigger> = List(randomIntBetween(1, 5)) { randomPPLTrigger() },
+    user: User? = randomUser(),
+    queryLanguage: QueryLanguage = QueryLanguage.PPL,
+    query: String = "source = $TEST_INDEX_NAME | head 10"
+): PPLSQLMonitor {
+    return PPLSQLMonitor(
+        name = name,
+        enabled = enabled,
+        schedule = schedule,
+        lookBackWindow = lookBackWindow,
+        timestampField = timestampField,
+        lastUpdateTime = lastUpdateTime,
+        enabledTime = enabledTime,
+        description = description,
+        triggers = triggers,
+        user = user,
+        queryLanguage = queryLanguage,
+        query = query
+    )
+}
+
 fun randomQueryLevelTrigger(
     id: String = UUIDs.base64UUID(),
     name: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
@@ -348,6 +394,42 @@ fun randomDocumentLevelTrigger(
     )
 }
 
+// random PPLTrigger defaults to a number_of_results trigger, because a custom condition
+// would require knowledge of the PPL Monitor's query
+// it is on the caller to be explicit and pass in valid arguments that would create either
+// a valid PPL Monitor or one that intentionally throws an error.
+// e.g. to create a valid PPL Monitor, if conditionType is CUSTOM,
+// numResultsCondition and numResultsValue must be null, while
+// customCondition must not be null.
+fun randomPPLTrigger(
+    id: String = UUIDs.base64UUID(),
+    name: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
+    severity: Severity = Severity.entries.random(),
+    throttleDuration: Long? = randomLongBetween(1, 100),
+    expireDuration: Long = randomLongBetween(1, 100),
+    actions: List<Action> = mutableListOf(),
+    mode: TriggerMode = TriggerMode.entries.random(),
+    conditionType: ConditionType = ConditionType.NUMBER_OF_RESULTS,
+    numResultsCondition: NumResultsCondition? = NumResultsCondition.entries.random(),
+    numResultsValue: Long? = randomLongBetween(1L, 50L),
+    customCondition: String? = null
+): PPLSQLTrigger {
+    return PPLSQLTrigger(
+        id = id,
+        name = name,
+        severity = severity,
+        throttleDuration = throttleDuration,
+        expireDuration = expireDuration,
+        lastTriggeredTime = null,
+        actions = actions,
+        mode = mode,
+        conditionType = conditionType,
+        numResultsCondition = numResultsCondition,
+        numResultsValue = numResultsValue,
+        customCondition = customCondition
+    )
+}
+
 fun randomBucketSelectorExtAggregationBuilder(
     name: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
     bucketsPathsMap: MutableMap<String, String> = mutableMapOf("avg" to "10"),
@@ -424,10 +506,11 @@ fun randomTemplateScript(
 fun randomAction(
     name: String = OpenSearchRestTestCase.randomUnicodeOfLength(10),
     template: Script = randomTemplateScript("Hello World"),
+    subjectTemplate: Script = template,
     destinationId: String = "",
     throttleEnabled: Boolean = false,
     throttle: Throttle = randomThrottle()
-) = Action(name, destinationId, template, template, throttleEnabled, throttle, actionExecutionPolicy = null)
+) = Action(name, destinationId, subjectTemplate, template, throttleEnabled, throttle, actionExecutionPolicy = null)
 
 fun randomActionWithPolicy(
     name: String = OpenSearchRestTestCase.randomUnicodeOfLength(10),
@@ -472,6 +555,42 @@ fun randomAlert(monitor: Monitor = randomQueryLevelMonitor()): Alert {
     )
 }
 
+fun randomAlertV2(
+    id: String = UUIDs.base64UUID(),
+    version: Long = randomLongBetween(1, 10),
+    schemaVersion: Int = randomIntBetween(1, 10),
+    monitorId: String = UUIDs.base64UUID(),
+    monitorName: String = UUIDs.base64UUID(),
+    monitorVersion: Long = randomLongBetween(1, 10),
+    monitorUser: User? = randomUser(),
+    triggerId: String = UUIDs.base64UUID(),
+    triggerName: String = UUIDs.base64UUID(),
+    query: String = "source = $TEST_INDEX_NAME | head 10",
+    queryResults: Map<String, Any> = mapOf(),
+    triggeredTime: Instant = Instant.now().truncatedTo(ChronoUnit.MILLIS),
+    errorMessage: String? = "sample error message",
+    severity: Severity = Severity.entries.random(),
+    executionId: String? = UUIDs.base64UUID()
+): AlertV2 {
+    return AlertV2(
+        id = id,
+        version = version,
+        schemaVersion = schemaVersion,
+        monitorId = monitorId,
+        monitorName = monitorName,
+        monitorVersion = monitorVersion,
+        monitorUser = monitorUser,
+        triggerId = triggerId,
+        triggerName = triggerName,
+        query = query,
+        queryResults = queryResults,
+        triggeredTime = triggeredTime,
+        errorMessage = errorMessage,
+        severity = severity,
+        executionId = executionId,
+    )
+}
+
 fun randomDocLevelQuery(
     id: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
     query: String = OpenSearchRestTestCase.randomAlphaOfLength(10),
@@ -810,3 +929,161 @@ fun randomAlertContext(
 fun Map<String, Any>.objectMap(key: String): Map<String, Map<String, Any>> {
     return this[key] as Map<String, Map<String, Any>>
 }
+
+fun assertPplMonitorsEqual(pplMonitor1: PPLSQLMonitor, pplMonitor2: PPLSQLMonitor) {
+    // note: Get and Search Monitor responses do not include User information by
+    // design, so that check is skipped
+
+    // note: Update Monitor API intentionally overrides the enabledTime of the new given monitor
+    // with the enabledTime of the existing monitor being updated to ensure execution correctness,
+    // so that check is skipped
+
+    assertEquals("Monitor enabled fields not equal", pplMonitor1.enabled, pplMonitor2.enabled)
+    assertEquals("Monitor schedules not equal", pplMonitor1.schedule, pplMonitor2.schedule)
+    assertEquals("Monitor lookback windows not equal", pplMonitor1.lookBackWindow, pplMonitor2.lookBackWindow)
+    assertEquals("Monitor timestamp fields not equal", pplMonitor1.timestampField, pplMonitor2.timestampField)
+    assertEquals("Monitor last updated times are not equal", pplMonitor1.lastUpdateTime, pplMonitor2.lastUpdateTime)
+    assertEquals("Monitor query languages not equal", pplMonitor1.queryLanguage, pplMonitor2.queryLanguage)
+    assertEquals("Monitor queries not equal", pplMonitor1.query, pplMonitor2.query)
+    assertEquals("Number of triggers in monitor not equal", pplMonitor1.triggers.size, pplMonitor2.triggers.size)
+
+    val sortedTriggers1 = pplMonitor1.triggers.sortedBy { it.id }
+    val sortedTriggers2 = pplMonitor2.triggers.sortedBy { it.id }
+    for (i in sortedTriggers1.indices) {
+        assertPplTriggersEqual(sortedTriggers1[i], sortedTriggers2[i])
+    }
+}
+
+fun assertPplTriggersEqual(pplTrigger1: PPLSQLTrigger, pplTrigger2: PPLSQLTrigger) {
+    assertEquals(
+        "Monitor trigger IDs not equal",
+        pplTrigger1.id,
+        pplTrigger2.id
+    )
+
+    val id = pplTrigger1.id
+
+    assertEquals(
+        "Monitor trigger $id names not equal",
+        pplTrigger1.name,
+        pplTrigger2.name
+    )
+    assertEquals(
+        "Monitor trigger $id severities not equal",
+        pplTrigger1.severity,
+        pplTrigger2.severity
+    )
+    assertEquals(
+        "Monitor trigger $id throttle durations not equal",
+        pplTrigger1.throttleDuration,
+        pplTrigger2.throttleDuration
+    )
+    assertEquals(
+        "Monitor trigger $id expire durations not equal",
+        pplTrigger1.expireDuration,
+        pplTrigger2.expireDuration
+    )
+    assertEquals(
+        "Monitor trigger $id modes not equal",
+        pplTrigger1.mode,
+        pplTrigger2.mode
+    )
+    assertEquals(
+        "Monitor trigger $id condition types not equal",
+        pplTrigger1.conditionType,
+        pplTrigger2.conditionType
+    )
+    assertEquals(
+        "Monitor trigger $id number_of_results conditions not equal",
+        pplTrigger1.numResultsCondition,
+        pplTrigger2.numResultsCondition
+    )
+    assertEquals(
+        "Monitor trigger $id number_of_results values not equal",
+        pplTrigger1.numResultsValue,
+        pplTrigger2.numResultsValue
+    )
+    assertEquals(
+        "Monitor trigger $id custom conditions not equal",
+        pplTrigger1.customCondition,
+        pplTrigger2.customCondition
+    )
+}
+
+fun assertAlertV2sEqual(alert1: AlertV2, alert2: AlertV2) {
+    assertEquals(
+        "AlertV2 IDs are not equal",
+        alert1.id,
+        alert2.id
+    )
+    assertEquals(
+        "AlertV2 versions are not equal",
+        alert1.version,
+        alert2.version
+    )
+    assertEquals(
+        "AlertV2 schema versions are not equal",
+        alert1.schemaVersion,
+        alert2.schemaVersion
+    )
+    assertEquals(
+        "AlertV2 monitor IDs are not equal",
+        alert1.monitorId,
+        alert2.monitorId
+    )
+    assertEquals(
+        "AlertV2 monitor names are not equal",
+        alert1.monitorName,
+        alert2.monitorName
+    )
+    assertEquals(
+        "AlertV2 monitor versions are not equal",
+        alert1.monitorVersion,
+        alert2.monitorVersion
+    )
+    assertEquals(
+        "AlertV2 monitor users are not equal",
+        alert1.monitorUser.toString(),
+        alert2.monitorUser.toString()
+    )
+    assertEquals(
+        "AlertV2 trigger IDs are not equal",
+        alert1.triggerId,
+        alert2.triggerId
+    )
+    assertEquals(
+        "AlertV2 trigger names are not equal",
+        alert1.triggerName,
+        alert2.triggerName
+    )
+    assertEquals(
+        "AlertV2 queries are not equal",
+        alert1.query,
+        alert2.query
+    )
+    assertEquals(
+        "AlertV2 query results are not equal",
+        alert1.queryResults,
+        alert2.queryResults
+    )
+    assertEquals(
+        "AlertV2 triggered times are not equal",
+        alert1.triggeredTime,
+        alert2.triggeredTime
+    )
+    assertEquals(
+        "AlertV2 error messages are not equal",
+        alert1.errorMessage,
+        alert2.errorMessage
+    )
+    assertEquals(
+        "AlertV2 severities are not equal",
+        alert1.severity,
+        alert2.severity
+    )
+    assertEquals(
+        "AlertV2 execution IDs are not equal",
+        alert1.executionId,
+        alert2.executionId
+    )
+}
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/AlertV2Tests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/AlertV2Tests.kt
new file mode 100644
index 000000000..fade152bc
--- /dev/null
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/AlertV2Tests.kt
@@ -0,0 +1,59 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.assertAlertV2sEqual
+import org.opensearch.alerting.modelv2.AlertV2.Companion.ALERT_V2_ID_FIELD
+import org.opensearch.alerting.modelv2.AlertV2.Companion.ALERT_V2_VERSION_FIELD
+import org.opensearch.alerting.modelv2.AlertV2.Companion.ERROR_MESSAGE_FIELD
+import org.opensearch.alerting.modelv2.AlertV2.Companion.EXECUTION_ID_FIELD
+import org.opensearch.alerting.modelv2.AlertV2.Companion.SEVERITY_FIELD
+import org.opensearch.alerting.randomAlertV2
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.test.OpenSearchTestCase
+
+class AlertV2Tests : OpenSearchTestCase() {
+    fun `test alertv2 as stream`() {
+        val alertV2 = randomAlertV2()
+        val out = BytesStreamOutput()
+        alertV2.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newAlertV2 = AlertV2(sin)
+        assertAlertV2sEqual(alertV2, newAlertV2)
+    }
+
+    fun `test alertv2 asTemplateArgs`() {
+        val alertV2 = randomAlertV2()
+        val templateArgs = alertV2.asTemplateArg()
+
+        assertEquals(
+            "Template args field $ALERT_V2_ID_FIELD doesn't match",
+            alertV2.id,
+            templateArgs[ALERT_V2_ID_FIELD]
+        )
+        assertEquals(
+            "Template args field $ALERT_V2_VERSION_FIELD doesn't match",
+            alertV2.version,
+            templateArgs[ALERT_V2_VERSION_FIELD]
+        )
+        assertEquals(
+            "Template args field $ERROR_MESSAGE_FIELD doesn't match",
+            alertV2.errorMessage,
+            templateArgs[ERROR_MESSAGE_FIELD]
+        )
+        assertEquals(
+            "Template args field $EXECUTION_ID_FIELD doesn't match",
+            alertV2.executionId,
+            templateArgs[EXECUTION_ID_FIELD]
+        )
+        assertEquals(
+            "Template args field $SEVERITY_FIELD doesn't match",
+            alertV2.severity.value,
+            templateArgs[SEVERITY_FIELD]
+        )
+    }
+}
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/MonitorV2Tests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/MonitorV2Tests.kt
new file mode 100644
index 000000000..a4c15a37a
--- /dev/null
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/MonitorV2Tests.kt
@@ -0,0 +1,161 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.assertPplMonitorsEqual
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ALERTING_V2_MAX_NAME_LENGTH
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ENABLED_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ENABLED_TIME_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.LAST_UPDATE_TIME_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.LOOK_BACK_WINDOW_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.MONITOR_V2_MIN_LOOK_BACK_WINDOW
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.NAME_FIELD
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.SCHEDULE_FIELD
+import org.opensearch.alerting.modelv2.PPLSQLMonitor.Companion.QUERY_FIELD
+import org.opensearch.alerting.randomPPLMonitor
+import org.opensearch.alerting.randomPPLTrigger
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.commons.alerting.util.IndexUtils.Companion._ID
+import org.opensearch.commons.alerting.util.IndexUtils.Companion._VERSION
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.test.OpenSearchTestCase
+import java.lang.IllegalArgumentException
+import java.time.Instant
+
+class MonitorV2Tests : OpenSearchTestCase() {
+    fun `test enabled time`() {
+        val pplMonitor = randomPPLMonitor(enabled = true, enabledTime = Instant.now())
+        try {
+            pplMonitor.makeCopy(enabled = false)
+            fail("Disabling monitor with enabled time set should fail.")
+        } catch (_: IllegalArgumentException) {}
+
+        val disabledMonitor = pplMonitor.copy(enabled = false, enabledTime = null)
+
+        try {
+            disabledMonitor.makeCopy(enabled = true)
+            fail("Enabling monitor without enabled time should fail")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test max triggers`() {
+        val tooManyTriggers = mutableListOf<PPLSQLTrigger>()
+        for (i in 0..10) { // 11 times
+            tooManyTriggers.add(randomPPLTrigger())
+        }
+
+        try {
+            randomPPLMonitor(triggers = tooManyTriggers)
+            fail("Monitor with too many triggers should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test monitor name too long`() {
+        var monitorName = ""
+        for (i in 0 until ALERTING_V2_MAX_NAME_LENGTH + 1) {
+            monitorName += "a"
+        }
+
+        try {
+            randomPPLMonitor(name = monitorName)
+            fail("Monitor with too long a name should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test monitor min look back window`() {
+        try {
+            randomPPLMonitor(
+                lookBackWindow = MONITOR_V2_MIN_LOOK_BACK_WINDOW - 1
+            )
+            fail("Monitor with too long a name should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test monitor no triggers`() {
+        try {
+            randomPPLMonitor(
+                triggers = listOf()
+            )
+            fail("Monitor without triggers be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test monitor with look back window without timestamp field`() {
+        try {
+            randomPPLMonitor(
+                lookBackWindow = randomLongBetween(1, 10),
+                timestampField = null
+            )
+            fail("Monitor with look back window but without timestamp field be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test monitor without look back window with timestamp field`() {
+        try {
+            randomPPLMonitor(
+                lookBackWindow = null,
+                timestampField = "some_timestamp_field"
+            )
+            fail("Monitor without look back window but with timestamp field be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test ppl monitor as stream`() {
+        val pplMonitor = randomPPLMonitor()
+        val out = BytesStreamOutput()
+        pplMonitor.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newPplMonitor = PPLSQLMonitor(sin)
+        assertPplMonitorsEqual(pplMonitor, newPplMonitor)
+    }
+
+    fun `test ppl monitor asTemplateArgs`() {
+        val pplMonitor = randomPPLMonitor()
+        val templateArgs = pplMonitor.asTemplateArg()
+
+        assertEquals(
+            "Template args field $_ID doesn't match",
+            pplMonitor.id,
+            templateArgs[_ID]
+        )
+        assertEquals(
+            "Template args field $_VERSION doesn't match",
+            pplMonitor.version,
+            templateArgs[_VERSION]
+        )
+        assertEquals(
+            "Template args field $NAME_FIELD doesn't match",
+            pplMonitor.name,
+            templateArgs[NAME_FIELD]
+        )
+        assertEquals(
+            "Template args field $ENABLED_FIELD doesn't match",
+            pplMonitor.enabled,
+            templateArgs[ENABLED_FIELD]
+        )
+        assertNotNull(templateArgs[SCHEDULE_FIELD])
+        assertEquals(
+            "Template args field $LOOK_BACK_WINDOW_FIELD doesn't match",
+            pplMonitor.lookBackWindow,
+            templateArgs[LOOK_BACK_WINDOW_FIELD]
+        )
+        assertEquals(
+            "Template args field $LAST_UPDATE_TIME_FIELD doesn't match",
+            pplMonitor.lastUpdateTime.toEpochMilli(),
+            templateArgs[LAST_UPDATE_TIME_FIELD]
+        )
+        assertEquals(
+            "Template args field $ENABLED_TIME_FIELD doesn't match",
+            pplMonitor.enabledTime?.toEpochMilli(),
+            templateArgs[ENABLED_TIME_FIELD]
+        )
+        assertEquals(
+            "Template args field $QUERY_FIELD doesn't match",
+            pplMonitor.query,
+            templateArgs[QUERY_FIELD]
+        )
+    }
+}
diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/TriggerV2Tests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/TriggerV2Tests.kt
new file mode 100644
index 000000000..430572254
--- /dev/null
+++ b/alerting/src/test/kotlin/org/opensearch/alerting/modelv2/TriggerV2Tests.kt
@@ -0,0 +1,254 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.modelv2
+
+import org.opensearch.alerting.assertPplTriggersEqual
+import org.opensearch.alerting.modelv2.MonitorV2.Companion.ALERTING_V2_MAX_NAME_LENGTH
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.Companion.CONDITION_TYPE_FIELD
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.Companion.CUSTOM_CONDITION_FIELD
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.Companion.MODE_FIELD
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.Companion.NUM_RESULTS_CONDITION_FIELD
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.Companion.NUM_RESULTS_VALUE_FIELD
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.ConditionType
+import org.opensearch.alerting.modelv2.PPLSQLTrigger.NumResultsCondition
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.ACTIONS_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.EXPIRE_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.ID_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.MONITOR_V2_MIN_EXPIRE_DURATION_MINUTES
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.MONITOR_V2_MIN_THROTTLE_DURATION_MINUTES
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.NAME_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.NOTIFICATIONS_ID_MAX_LENGTH
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.SEVERITY_FIELD
+import org.opensearch.alerting.modelv2.TriggerV2.Companion.THROTTLE_FIELD
+import org.opensearch.alerting.randomAction
+import org.opensearch.alerting.randomPPLTrigger
+import org.opensearch.common.io.stream.BytesStreamOutput
+import org.opensearch.core.common.io.stream.StreamInput
+import org.opensearch.test.OpenSearchTestCase
+import java.lang.IllegalArgumentException
+
+class TriggerV2Tests : OpenSearchTestCase() {
+    fun `test min throttle duration`() {
+        try {
+            randomPPLTrigger(
+                throttleDuration = MONITOR_V2_MIN_THROTTLE_DURATION_MINUTES - 1
+            )
+            fail("Trigger with throttle duration less than 1 should be rejected")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test min expire duration`() {
+        try {
+            randomPPLTrigger(
+                expireDuration = MONITOR_V2_MIN_EXPIRE_DURATION_MINUTES - 1
+            )
+            fail("Trigger with expire duration less than 1 should be rejected")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test trigger name too long`() {
+        var triggerName = ""
+        for (i in 0 until ALERTING_V2_MAX_NAME_LENGTH + 1) {
+            triggerName += "a"
+        }
+
+        try {
+            randomPPLTrigger(name = triggerName)
+            fail("Trigger with too long a name should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test number of results trigger with negative number of results value`() {
+        try {
+            randomPPLTrigger(
+                conditionType = ConditionType.NUMBER_OF_RESULTS,
+                numResultsValue = -1L,
+                numResultsCondition = NumResultsCondition.GREATER_THAN
+            )
+            fail("Number of results trigger with negative number of results value should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test trigger action name too long`() {
+        var actionName = ""
+        for (i in 0 until ALERTING_V2_MAX_NAME_LENGTH + 1) {
+            actionName += "a"
+        }
+
+        try {
+            randomPPLTrigger(
+                actions = listOf(
+                    randomAction(
+                        name = actionName
+                    )
+                )
+            )
+            fail("Trigger action with too long a name should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test trigger action channel ID too long`() {
+        var channelId = ""
+        for (i in 0 until NOTIFICATIONS_ID_MAX_LENGTH + 1) {
+            channelId += "a"
+        }
+
+        try {
+            randomPPLTrigger(
+                actions = listOf(
+                    randomAction(
+                        destinationId = channelId
+                    )
+                )
+            )
+            fail("Trigger action with too long a channel ID should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test number_of_results trigger has no number_of_results value field`() {
+        try {
+            randomPPLTrigger(
+                conditionType = ConditionType.NUMBER_OF_RESULTS,
+                numResultsCondition = NumResultsCondition.entries.random(),
+                numResultsValue = null,
+                customCondition = null
+            )
+            fail("Number of results trigger that has no number of results value should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test number_of_results trigger has no number_of_results condition field`() {
+        try {
+            randomPPLTrigger(
+                conditionType = ConditionType.NUMBER_OF_RESULTS,
+                numResultsCondition = null,
+                numResultsValue = randomLongBetween(1, 10),
+                customCondition = null
+            )
+            fail("Number of results trigger that has no number of results condition should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test number_of_results trigger has custom_condition value field`() {
+        try {
+            randomPPLTrigger(
+                conditionType = ConditionType.NUMBER_OF_RESULTS,
+                numResultsCondition = null,
+                numResultsValue = null,
+                customCondition = "eval result = something > 5"
+            )
+            fail("Number of results trigger that has custom condition should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test custom trigger has number_of_results value field`() {
+        try {
+            randomPPLTrigger(
+                conditionType = ConditionType.CUSTOM,
+                numResultsCondition = NumResultsCondition.entries.random(),
+                numResultsValue = null,
+                customCondition = null
+            )
+            fail("Number of results trigger that has no number of results value should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test custom trigger has number_of_results condition field`() {
+        try {
+            randomPPLTrigger(
+                conditionType = ConditionType.CUSTOM,
+                numResultsCondition = null,
+                numResultsValue = randomLongBetween(1, 10),
+                customCondition = null
+            )
+            fail("Number of results trigger that has no number of results condition should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test custom trigger has no custom_condition value field`() {
+        try {
+            randomPPLTrigger(
+                conditionType = ConditionType.CUSTOM,
+                numResultsCondition = null,
+                numResultsValue = null,
+                customCondition = null
+            )
+            fail("Number of results trigger that has custom condition should be rejected.")
+        } catch (_: IllegalArgumentException) {}
+    }
+
+    fun `test ppl trigger as stream`() {
+        val pplTrigger = randomPPLTrigger()
+        val out = BytesStreamOutput()
+        pplTrigger.writeTo(out)
+        val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
+        val newPplTrigger = PPLSQLTrigger(sin)
+        assertPplTriggersEqual(pplTrigger, newPplTrigger)
+    }
+
+    fun `test ppl trigger asTemplateArgs`() {
+        val pplTrigger = randomPPLTrigger()
+        val templateArgs = pplTrigger.asTemplateArg()
+
+        assertEquals(
+            "Template args field $ID_FIELD doesn't match",
+            pplTrigger.id,
+            templateArgs[ID_FIELD]
+        )
+        assertEquals(
+            "Template args field $NAME_FIELD doesn't match",
+            pplTrigger.name,
+            templateArgs[NAME_FIELD]
+        )
+        assertEquals(
+            "Template args field $SEVERITY_FIELD doesn't match",
+            pplTrigger.severity.value,
+            templateArgs[SEVERITY_FIELD]
+        )
+        assertEquals(
+            "Template args field $THROTTLE_FIELD doesn't match",
+            pplTrigger.throttleDuration,
+            templateArgs[THROTTLE_FIELD]
+        )
+        assertEquals(
+            "Template args field $EXPIRE_FIELD doesn't match",
+            pplTrigger.expireDuration,
+            templateArgs[EXPIRE_FIELD]
+        )
+        assertEquals(
+            "Template args field $EXPIRE_FIELD doesn't match",
+            pplTrigger.expireDuration,
+            templateArgs[EXPIRE_FIELD]
+        )
+        val actions = templateArgs[ACTIONS_FIELD] as List<*>
+        assertEquals("number of trigger actions doesn't match", pplTrigger.actions.size, actions.size)
+        assertEquals(
+            "Template args field $MODE_FIELD doesn't match",
+            pplTrigger.mode.value,
+            templateArgs[MODE_FIELD]
+        )
+        assertEquals(
+            "Template args field $CONDITION_TYPE_FIELD doesn't match",
+            pplTrigger.conditionType.value,
+            templateArgs[CONDITION_TYPE_FIELD]
+        )
+        assertEquals(
+            "Template args field $NUM_RESULTS_CONDITION_FIELD doesn't match",
+            pplTrigger.numResultsCondition?.value,
+            templateArgs[NUM_RESULTS_CONDITION_FIELD]
+        )
+        assertEquals(
+            "Template args field $NUM_RESULTS_VALUE_FIELD doesn't match",
+            pplTrigger.numResultsValue,
+            templateArgs[NUM_RESULTS_VALUE_FIELD]
+        )
+        assertEquals(
+            "Template args field $CUSTOM_CONDITION_FIELD doesn't match",
+            pplTrigger.customCondition,
+            templateArgs[CUSTOM_CONDITION_FIELD]
+        )
+    }
+}
diff --git a/core/src/main/kotlin/org/opensearch/alerting/core/util/XContentExtensions.kt b/core/src/main/kotlin/org/opensearch/alerting/core/util/XContentExtensions.kt
new file mode 100644
index 000000000..9ca03ed6b
--- /dev/null
+++ b/core/src/main/kotlin/org/opensearch/alerting/core/util/XContentExtensions.kt
@@ -0,0 +1,13 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.alerting.core.util
+
+import org.opensearch.core.xcontent.XContentBuilder
+import java.time.Instant
+
+fun XContentBuilder.nonOptionalTimeField(name: String, instant: Instant): XContentBuilder {
+    return this.timeField(name, "${name}_in_millis", instant.toEpochMilli())
+}