From 08f17fc155f44f9d5eef564b231127c30af6ffee Mon Sep 17 00:00:00 2001 From: Kevin Tavog <165935671+kevin-paladin@users.noreply.github.com> Date: Fri, 22 Nov 2024 13:28:14 -0800 Subject: [PATCH 1/2] Direct aws, azure & gcp to the asset state service --- .../pacman/commons/aws/sqs/SQSManager.java | 5 ++++ .../commons/dto/AssetStateStartEvent.java | 20 +++++++++++++ .../tmobile/cso/pacman/datashipper/Main.java | 29 +++++++++++++++---- 3 files changed, 49 insertions(+), 5 deletions(-) create mode 100644 commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java diff --git a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java index 93d6ed51c0..f87b54f2a0 100644 --- a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java +++ b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java @@ -23,6 +23,7 @@ import com.amazonaws.services.sqs.model.SendMessageResult; import com.fasterxml.jackson.databind.ObjectMapper; import com.tmobile.pacman.commons.aws.CredentialProvider; +import com.tmobile.pacman.commons.dto.AssetStateStartEvent; import com.tmobile.pacman.commons.dto.JobDoneMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,6 +42,10 @@ public static SQSManager getInstance() { return InstanceHolder.instance; } + public String sendMessage(AssetStateStartEvent message, String url) { + return sendMessage(message.toCommandLine(), url); + } + public String sendSQSMessage(JobDoneMessage jobDoneMessage, String url) { try { String sqsMessage = objectMapper.writeValueAsString(jobDoneMessage); diff --git a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java new file mode 100644 index 0000000000..2c3bac10a8 --- /dev/null +++ b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java @@ -0,0 +1,20 @@ +package com.tmobile.pacman.commons.dto; + +public class AssetStateStartEvent { + private final String tenantId; + private final String dataSource; + private final String[] assetTypes; + private final boolean isFromPolicyEngine; + + public AssetStateStartEvent(String tenantId, String dataSource, String[] assetTypes, boolean isFromPolicyEngine) { + this.tenantId = tenantId; + this.dataSource = dataSource; + this.assetTypes = assetTypes; + this.isFromPolicyEngine = isFromPolicyEngine; + } + + public String toCommandLine() { + return String.format("--tenant_id=%s --data_source=%s --asset_types=%s --is_from_policy_engine=%s", + tenantId, dataSource, String.join(",", assetTypes), isFromPolicyEngine); + } +} diff --git a/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java b/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java index 541175f5fa..c83ddf7c92 100644 --- a/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java +++ b/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java @@ -15,19 +15,20 @@ ******************************************************************************/ package com.tmobile.cso.pacman.datashipper; +import com.tmobile.cso.pacman.datashipper.config.ConfigManager; import com.tmobile.cso.pacman.datashipper.dto.DatasourceData; import com.tmobile.cso.pacman.datashipper.entity.*; import com.tmobile.cso.pacman.datashipper.es.ESManager; import com.tmobile.cso.pacman.datashipper.util.Constants; import com.tmobile.cso.pacman.datashipper.util.ErrorManageUtil; import com.tmobile.pacman.commons.aws.sqs.SQSManager; +import com.tmobile.pacman.commons.dto.AssetStateStartEvent; import com.tmobile.pacman.commons.dto.JobDoneMessage; import com.tmobile.pacman.commons.jobs.PacmanJob; +import java.util.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.*; - @PacmanJob(methodToexecute = "shipData", jobName = "data-shipper", desc = "Job to load data from s3 to OP", priority = 5) public class Main implements Constants { private static final Logger LOGGER = LoggerFactory.getLogger(Main.class); @@ -107,10 +108,28 @@ public static Map shipData(Map params) { errorList.add(errorMap); LOGGER.error("Error while updating stats", e); } + + // Route all assets through the asset state service, which will route to the policy engine + // once it's done processing. SQSManager sqsManager = SQSManager.getInstance(); - JobDoneMessage jobDoneMessage = new JobDoneMessage(ds + "-Shipper-Job", tenantId, ds, null); - String sqsMessageID = sqsManager.sendSQSMessage(jobDoneMessage, System.getenv("SHIPPER_SQS_QUEUE_URL")); - LOGGER.debug("Shipper done SQS message ID: {}", sqsMessageID); + boolean isAsset = ds.equalsIgnoreCase("azure") + || ds.equalsIgnoreCase("aws") + || ds.equalsIgnoreCase("gcp"); + if (isAsset) { + List assetTypes = new ArrayList<>(ConfigManager.getTypesWithDisplayName(ds).keySet()); + Collections.sort(assetTypes); + AssetStateStartEvent assetStateStartEvent = + new AssetStateStartEvent(tenantId, ds, assetTypes.toArray(new String[0]), false); + String sqsMessageID = + sqsManager.sendMessage(assetStateStartEvent, System.getenv("ASSET_STATE_QUEUE_URL")); + LOGGER.debug("AssetState Start SQS message ID: {}", sqsMessageID); + } else { + JobDoneMessage jobDoneMessage = new JobDoneMessage(ds + "-Shipper-Job", tenantId, ds, null); + String sqsMessageID = + sqsManager.sendSQSMessage(jobDoneMessage, System.getenv("SHIPPER_SQS_QUEUE_URL")); + LOGGER.debug("Shipper done SQS message ID: {}", sqsMessageID); + } + Map status = ErrorManageUtil.formErrorCode(jobName, errorList); LOGGER.info("Job Return Status {} ", status); return status; From 14e14b90f9cebbf384639c0ff6848dd6052348c8 Mon Sep 17 00:00:00 2001 From: Kevin Tavog <165935671+kevin-paladin@users.noreply.github.com> Date: Mon, 2 Dec 2024 14:59:05 -0800 Subject: [PATCH 2/2] Use the same event & format for the policy engine & asset state service --- .../pacman/commons/aws/sqs/SQSManager.java | 5 ----- .../commons/dto/AssetStateStartEvent.java | 20 ------------------ .../pacman/commons/dto/JobDoneMessage.java | 21 +++++++++++++++++++ .../tmobile/cso/pacman/datashipper/Main.java | 9 ++++---- 4 files changed, 25 insertions(+), 30 deletions(-) delete mode 100644 commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java diff --git a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java index f87b54f2a0..93d6ed51c0 100644 --- a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java +++ b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/aws/sqs/SQSManager.java @@ -23,7 +23,6 @@ import com.amazonaws.services.sqs.model.SendMessageResult; import com.fasterxml.jackson.databind.ObjectMapper; import com.tmobile.pacman.commons.aws.CredentialProvider; -import com.tmobile.pacman.commons.dto.AssetStateStartEvent; import com.tmobile.pacman.commons.dto.JobDoneMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,10 +41,6 @@ public static SQSManager getInstance() { return InstanceHolder.instance; } - public String sendMessage(AssetStateStartEvent message, String url) { - return sendMessage(message.toCommandLine(), url); - } - public String sendSQSMessage(JobDoneMessage jobDoneMessage, String url) { try { String sqsMessage = objectMapper.writeValueAsString(jobDoneMessage); diff --git a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java deleted file mode 100644 index 2c3bac10a8..0000000000 --- a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/AssetStateStartEvent.java +++ /dev/null @@ -1,20 +0,0 @@ -package com.tmobile.pacman.commons.dto; - -public class AssetStateStartEvent { - private final String tenantId; - private final String dataSource; - private final String[] assetTypes; - private final boolean isFromPolicyEngine; - - public AssetStateStartEvent(String tenantId, String dataSource, String[] assetTypes, boolean isFromPolicyEngine) { - this.tenantId = tenantId; - this.dataSource = dataSource; - this.assetTypes = assetTypes; - this.isFromPolicyEngine = isFromPolicyEngine; - } - - public String toCommandLine() { - return String.format("--tenant_id=%s --data_source=%s --asset_types=%s --is_from_policy_engine=%s", - tenantId, dataSource, String.join(",", assetTypes), isFromPolicyEngine); - } -} diff --git a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/JobDoneMessage.java b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/JobDoneMessage.java index 9af8fad8ba..102077edfe 100644 --- a/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/JobDoneMessage.java +++ b/commons/pac-batch-commons/src/main/java/com/tmobile/pacman/commons/dto/JobDoneMessage.java @@ -15,11 +15,22 @@ ******************************************************************************/ package com.tmobile.pacman.commons.dto; +import com.fasterxml.jackson.annotation.JsonProperty; + public class JobDoneMessage { private String jobName; + // NOTE: 'paladinCloudTenantId' is currently sent by the legacy shipper while 'tenantId' is sent + // by the delta engine; it's not clear if different recipients will use what's available so + // both are sent here. private String paladinCloudTenantId; + @JsonProperty("tenant_id") + private String tenantId; private String source; private String enricherSource; + @JsonProperty("tenant_name") + private String tenantName; + private String[] assetTypes; + public String getJobName() { return jobName; @@ -53,10 +64,20 @@ public void setEnricherSource(String enricherSource) { this.enricherSource = enricherSource; } + public String getTenantName() { return tenantName; } + public String[] getAssetTypes() { return assetTypes; } + public JobDoneMessage (String jobName, String paladinCloudTenantId, String source, String enricherSource) { + this(jobName, paladinCloudTenantId, source, enricherSource, null, null); + } + + public JobDoneMessage (String jobName, String paladinCloudTenantId, String source, String enricherSource, String tenantName, String[] assetTypes) { this.jobName = jobName; this.paladinCloudTenantId = paladinCloudTenantId; + this.tenantId = paladinCloudTenantId; + this.tenantName = tenantName; this.source = source; this.enricherSource = enricherSource; + this.assetTypes = assetTypes; } } diff --git a/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java b/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java index c83ddf7c92..09c372d5c9 100644 --- a/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java +++ b/jobs/pacman-data-shipper/src/main/java/com/tmobile/cso/pacman/datashipper/Main.java @@ -22,7 +22,6 @@ import com.tmobile.cso.pacman.datashipper.util.Constants; import com.tmobile.cso.pacman.datashipper.util.ErrorManageUtil; import com.tmobile.pacman.commons.aws.sqs.SQSManager; -import com.tmobile.pacman.commons.dto.AssetStateStartEvent; import com.tmobile.pacman.commons.dto.JobDoneMessage; import com.tmobile.pacman.commons.jobs.PacmanJob; import java.util.*; @@ -118,11 +117,11 @@ public static Map shipData(Map params) { if (isAsset) { List assetTypes = new ArrayList<>(ConfigManager.getTypesWithDisplayName(ds).keySet()); Collections.sort(assetTypes); - AssetStateStartEvent assetStateStartEvent = - new AssetStateStartEvent(tenantId, ds, assetTypes.toArray(new String[0]), false); + JobDoneMessage jobDoneMessage = new JobDoneMessage(ds + "-Shipper-Job", tenantId, + ds, null, null, assetTypes.toArray(new String[0])); String sqsMessageID = - sqsManager.sendMessage(assetStateStartEvent, System.getenv("ASSET_STATE_QUEUE_URL")); - LOGGER.debug("AssetState Start SQS message ID: {}", sqsMessageID); + sqsManager.sendSQSMessage(jobDoneMessage, System.getenv("ASSET_STATE_QUEUE_URL")); + LOGGER.debug("Done SQS message ID: {}", sqsMessageID); } else { JobDoneMessage jobDoneMessage = new JobDoneMessage(ds + "-Shipper-Job", tenantId, ds, null); String sqsMessageID =