Support sending email message via Notifications passthrough API

notifications/core/src/main/kotlin/org/opensearch/notifications/core/setting/PluginSettings.kt

@@ -32,6 +32,12 @@ internal object PluginSettings {
      */
     private const val EMAIL_KEY_PREFIX = "$KEY_PREFIX.email"
 
+    /**
+     * Legacy Email Destination setting prefix used by Alerting.
+     * Defining this here to be used as a fallback for the Notification plugin setting to account for migrated Email Destinations.
+     */
+    private const val LEGACY_EMAIL_DESTINATION_SETTING_PREFIX = "plugins.alerting.destination.email."
+
     /**
      * Settings Key prefix for Email. Note: Should contain . at the end for secure settings
      */
@@ -310,16 +316,38 @@ internal object PluginSettings {
         NodeScope, Dynamic
     )
 
+    private val LEGACY_EMAIL_USERNAME: Setting.AffixSetting<SecureString> = Setting.affixKeySetting(
+        LEGACY_EMAIL_DESTINATION_SETTING_PREFIX,
+        "username",
+        { key: String -> SecureSetting.secureString(key, null) }
+    )
+
+    private val LEGACY_EMAIL_PASSWORD: Setting.AffixSetting<SecureString> = Setting.affixKeySetting(
+        LEGACY_EMAIL_DESTINATION_SETTING_PREFIX,
+        "password",
+        { key: String -> SecureSetting.secureString(key, null) }
+    )
+
     private val EMAIL_USERNAME: Setting.AffixSetting<SecureString> = Setting.affixKeySetting(
         EMAIL_DESTINATION_SETTING_PREFIX,
         "username",
-        { key: String -> SecureSetting.secureString(key, null) }
+        { key: String ->
+            SecureSetting.secureString(
+                key,
+                fallback(key, LEGACY_EMAIL_USERNAME, "opensearch\\.notifications\\.core", "plugins.alerting.destination")
+            )
+        }
     )
 
     private val EMAIL_PASSWORD: Setting.AffixSetting<SecureString> = Setting.affixKeySetting(
         EMAIL_DESTINATION_SETTING_PREFIX,
         "password",
-        { key: String -> SecureSetting.secureString(key, null) }
+        { key: String ->
+            SecureSetting.secureString(
+                key,
+                fallback(key, LEGACY_EMAIL_PASSWORD, "opensearch\\.notifications\\.core", "plugins.alerting.destination")
+            )
+        }
     )
 
     /**
@@ -465,7 +493,10 @@ internal object PluginSettings {
         // If this logic needs to be expanded to support other Destinations, different groups can be retrieved similar
         // to emailAccountNames based on the setting namespace and SecureDestinationSettings should be expanded to support
         // these new settings.
-        val emailAccountNames: Set<String> = settings.getGroups(EMAIL_DESTINATION_SETTING_PREFIX, true).keys
+        var emailAccountNames: Set<String> = settings.getGroups(EMAIL_DESTINATION_SETTING_PREFIX, true).keys
+        // Retrieve the email account names defined under the legacy Alerting prefix as well, otherwise the fallback setting won't be checked for them
+        val legacyEmailAccountNames: Set<String> = settings.getGroups(LEGACY_EMAIL_DESTINATION_SETTING_PREFIX, true).keys
+        emailAccountNames = emailAccountNames.union(legacyEmailAccountNames)
         val emailAccounts: MutableMap<String, SecureDestinationSettings> = mutableMapOf()
         for (emailAccountName in emailAccountNames) {
             // Only adding the settings if they exist
@@ -493,6 +524,14 @@ internal object PluginSettings {
         return concreteSetting.get(settings)
     }
 
+    private fun <T> fallback(key: String, affixSetting: Setting.AffixSetting<T>, regex: String, replacement: String): Setting<T>? {
+        return if ("_na_" == key) {
+            affixSetting.getConcreteSettingForNamespace(key)
+        } else {
+            affixSetting.getConcreteSetting(key.replace(regex.toRegex(), replacement))
+        }
+    }
+
     // reset the settings values to default values for testing purpose
     @OpenForTesting
     fun reset() {

notifications/notifications/src/main/kotlin/org/opensearch/notifications/send/SendMessageActionHelper.kt

@@ -15,6 +15,7 @@ import org.opensearch.commons.authuser.User
 import org.opensearch.commons.destination.message.LegacyBaseMessage
 import org.opensearch.commons.destination.message.LegacyCustomWebhookMessage
 import org.opensearch.commons.destination.message.LegacyDestinationType
+import org.opensearch.commons.destination.message.LegacyEmailMessage
 import org.opensearch.commons.destination.response.LegacyDestinationResponse
 import org.opensearch.commons.notifications.action.LegacyPublishNotificationRequest
 import org.opensearch.commons.notifications.action.LegacyPublishNotificationResponse
@@ -265,7 +266,7 @@ object SendMessageActionHelper {
      * @return notification delivery status for the legacy destination
      */
     private fun sendMessageToLegacyDestination(baseMessage: LegacyBaseMessage): LegacyDestinationResponse {
-        val message =
+        var message =
             MessageContent(title = "Legacy Notification", textDescription = baseMessage.messageContent)
         // These legacy destination calls do not have reference Ids, just passing 'legacy' constant
         return when (baseMessage.channelType) {
@@ -291,6 +292,45 @@ object SendMessageActionHelper {
                 LegacyDestinationResponse.Builder().withStatusCode(status.statusCode)
                     .withResponseContent(status.statusText).build()
             }
+            LegacyDestinationType.LEGACY_EMAIL -> {
+                val recipients = (baseMessage as LegacyEmailMessage).recipients
+                // The naming is a little confusing but need to use the subject from LegacyEmailMessage as the title,
+                // so it appears as the subject of the email
+                message = MessageContent(title = baseMessage.subject, textDescription = baseMessage.messageContent)
+                runBlocking {
+                    val emailRecipientStatus: List<EmailRecipientStatus> = recipients.map {
+                        async(Dispatchers.IO) {
+                            val destination = SmtpDestination(
+                                baseMessage.accountName,
+                                baseMessage.host,
+                                baseMessage.port,
+                                baseMessage.method,
+                                baseMessage.from,
+                                it
+                            )
+                            val status = sendMessageThroughSpi(destination, message, "legacy")
+                            EmailRecipientStatus(
+                                it,
+                                DeliveryStatus(status.statusCode.toString(), status.statusText)
+                            )
+                        }
+                    }.awaitAll()
+
+                    val firstStatus = emailRecipientStatus.first()
+                    var overallStatus = firstStatus.deliveryStatus.statusCode
+                    var overallStatusText = firstStatus.deliveryStatus.statusText
+                    emailRecipientStatus.forEach {
+                        val status = it.deliveryStatus
+                        log.info("$LOG_PREFIX:Legacy email:statusCode=${status.statusCode}, statusText=${status.statusText}")
+                        if (overallStatus != status.statusCode) {
+                            overallStatus = RestStatus.MULTI_STATUS.status.toString()
+                            overallStatusText = "Errors"
+                        }
+                    }
+                    LegacyDestinationResponse.Builder().withStatusCode(overallStatus.toInt())
+                        .withResponseContent(overallStatusText).build()
+                }
+            }
             null -> {
                 log.warn("No channel type given (null) for publishing to legacy destination")
                 LegacyDestinationResponse.Builder().withStatusCode(400)