feat: Add tenant_id to monitor metadata and propagate to AlertService SDK calls (opensearch-project#2131)

- Store tenant_id in monitor.metadata map during create flow in
  TransportIndexMonitorAction when multi-tenancy is enabled
- Set x-tenant-id header in MonitorJobPoller.populateThreadContext()
  from monitor metadata before executing monitors from SQS
- Add tenantId(currentTenantId()) to all AlertService SDK client calls
  (search, put, delete) for proper multi-tenant routing
- Add debug logging in AlertService for tenant context visibility
- Add unit tests for metadata key logic

Signed-off-by: Surya Sashank Nistala <snistala@amazon.com>

Author: eirsep

alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt

@@ -175,6 +175,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
         @JvmField val OPEN_SEARCH_DASHBOARDS_USER_AGENT = "OpenSearch-Dashboards"
         @JvmField val UI_METADATA_EXCLUDE = arrayOf("monitor.${Monitor.UI_METADATA_FIELD}")
         @JvmField val TENANT_ID_HEADER = "x-tenant-id"
+        @JvmField val TENANT_ID_METADATA_KEY = "tenant_id"
         @JvmField val MONITOR_BASE_URI = "/_plugins/_alerting/monitors"
         @JvmField val WORKFLOW_BASE_URI = "/_plugins/_alerting/workflows"
         @JvmField val REMOTE_BASE_URI = "/_plugins/_alerting/remote"

alerting/src/main/kotlin/org/opensearch/alerting/service/MonitorJobPoller.kt

@@ -12,6 +12,7 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import org.apache.logging.log4j.LogManager
+import org.opensearch.alerting.AlertingPlugin
 import org.opensearch.alerting.action.ExecuteMonitorAction
 import org.opensearch.alerting.action.ExecuteMonitorRequest
 import org.opensearch.alerting.action.ExecuteMonitorResponse
@@ -22,7 +23,6 @@ import org.opensearch.common.xcontent.LoggingDeprecationHandler
 import org.opensearch.common.xcontent.XContentType
 import org.opensearch.commons.alerting.model.Monitor
 import org.opensearch.commons.alerting.model.ScheduleJobPayload
-import org.opensearch.commons.alerting.model.Target
 import org.opensearch.commons.alerting.util.AlertingException
 import org.opensearch.commons.utils.scheduler.JobQueueAccountIdProvider
 import org.opensearch.core.xcontent.NamedXContentRegistry
@@ -145,7 +145,7 @@ class MonitorJobPoller(
     private suspend fun executeMonitor(monitor: Monitor, jobStartTime: Instant) {
         // populate thread context for downstream request interception the moment
         // Monitor config is in hand
-        populateThreadContext(monitor.target)
+        populateThreadContext(monitor)
 
         val request = ExecuteMonitorRequest(
             dryrun = false,
@@ -196,7 +196,8 @@ class MonitorJobPoller(
     // populates thread context with KVs that downstream interception will
     // need when intercepting search or PPL calls to external customer
     // data source
-    internal fun populateThreadContext(target: Target?) {
+    internal fun populateThreadContext(monitor: Monitor) {
+        val target = monitor.target
         if (target == null) {
             throw AlertingException.wrap(
                 IllegalStateException("Monitor received by Job Poller did not contain target")
@@ -230,6 +231,12 @@ class MonitorJobPoller(
 
         // populated upstream in AlertingPlugin.kt with REMOTE_METADATA_REGION.get(settings)
         threadContext.putHeader(REGION_HEADER, region)
+
+        // Set tenant_id header from monitor metadata for downstream SDK calls
+        val tenantId = monitor.metadata?.get(AlertingPlugin.TENANT_ID_METADATA_KEY)
+        if (!tenantId.isNullOrEmpty()) {
+            threadContext.putHeader(AlertingPlugin.TENANT_ID_HEADER, tenantId)
+        }
     }
 
     private fun mapTargetTypeToServiceName(targetType: String): String {

alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt

@@ -822,6 +822,12 @@ class TransportIndexMonitorAction @Inject constructor(
 
             log.info("Creating new monitor: ${request.monitor.name}, type: ${request.monitor.monitorType}")
 
+            if (!tenantId.isNullOrEmpty()) {
+                val updatedMetadata = (request.monitor.metadata.orEmpty()) +
+                    (AlertingPlugin.TENANT_ID_METADATA_KEY to tenantId)
+                request.monitor = request.monitor.copy(metadata = updatedMetadata)
+            }
+
             val monitorObj = ToXContentObject { builder, params ->
                 request.monitor.toXContentWithUser(builder, ToXContent.MapParams(mapOf("with_type" to "true")))
             }

alerting/src/test/kotlin/org/opensearch/alerting/service/MonitorJobPollerTests.kt

@@ -377,8 +377,9 @@ class MonitorJobPollerTests : OpenSearchTestCase() {
 
         val mockTargetType = mappingProvider().entries.first().key
         val target = Target(type = mockTargetType, endpoint = "https://test.aoss.amazonaws.com")
+        val monitor = org.opensearch.alerting.randomQueryLevelMonitor().copy(target = target)
 
-        poller.populateThreadContext(target)
+        poller.populateThreadContext(monitor)
 
         assertEquals("true", mockThreadContext.getHeader(MonitorJobPoller.IS_BACKGROUND_JOB_HEADER))
         assertEquals(mappingProvider()[mockTargetType], mockThreadContext.getHeader(MonitorJobPoller.SERVICE_NAME_HEADER))
@@ -398,9 +399,10 @@ class MonitorJobPollerTests : OpenSearchTestCase() {
         )
 
         val target = Target(type = "non_existent_type", endpoint = "https://test.aoss.amazonaws.com")
+        val monitor = org.opensearch.alerting.randomQueryLevelMonitor().copy(target = target)
 
         expectThrows(Exception::class.java) {
-            poller.populateThreadContext(target)
+            poller.populateThreadContext(monitor)
         }
 
         poller.close()

alerting/src/test/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorActionTests.kt

@@ -10,6 +10,7 @@ import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters
 import org.junit.Before
 import org.mockito.Mockito
 import org.opensearch.action.support.ActionFilters
+import org.opensearch.alerting.AlertingPlugin
 import org.opensearch.alerting.core.ScheduledJobIndices
 import org.opensearch.alerting.service.ExternalSchedulerService
 import org.opensearch.alerting.settings.AlertingSettings
@@ -190,4 +191,41 @@ class TransportIndexMonitorActionTests : OpenSearchTestCase() {
         assertEquals("us-west-2", info.region)
         assertEquals("monitor-m1", info.name)
     }
+
+    fun `test tenant_id added to monitor metadata when tenantId is present`() {
+        val tenantId = "test-tenant-123"
+        val existingMetadata: Map<String, String> = mapOf("existing_key" to "existing_value")
+        val updatedMetadata = existingMetadata +
+            (AlertingPlugin.TENANT_ID_METADATA_KEY to tenantId)
+        assertEquals(tenantId, updatedMetadata[AlertingPlugin.TENANT_ID_METADATA_KEY])
+        assertEquals("existing_value", updatedMetadata["existing_key"])
+    }
+
+    fun `test tenant_id not added to monitor metadata when tenantId is null`() {
+        val tenantId: String? = null
+        val existingMetadata: Map<String, String> = mapOf("existing_key" to "existing_value")
+        val updatedMetadata = if (!tenantId.isNullOrEmpty()) {
+            existingMetadata + (AlertingPlugin.TENANT_ID_METADATA_KEY to tenantId)
+        } else {
+            existingMetadata
+        }
+        assertNull(updatedMetadata[AlertingPlugin.TENANT_ID_METADATA_KEY])
+        assertEquals(1, updatedMetadata.size)
+    }
+
+    fun `test tenant_id not added to monitor metadata when tenantId is empty`() {
+        val tenantId = ""
+        val existingMetadata: Map<String, String> = mapOf("existing_key" to "existing_value")
+        val updatedMetadata = if (!tenantId.isNullOrEmpty()) {
+            existingMetadata + (AlertingPlugin.TENANT_ID_METADATA_KEY to tenantId)
+        } else {
+            existingMetadata
+        }
+        assertNull(updatedMetadata[AlertingPlugin.TENANT_ID_METADATA_KEY])
+        assertEquals(1, updatedMetadata.size)
+    }
+
+    fun `test tenant_id metadata key constant value`() {
+        assertEquals("tenant_id", AlertingPlugin.TENANT_ID_METADATA_KEY)
+    }
 }