Merge branch 'no-int-test/dac-2784-redshift-txma-new-model-step-funct…

…ion' into production-preview-release
govuk-one-login · Apr 29, 2024 · 5ad3903 · 5ad3903
2 parents 4648977 + d051a52
commit 5ad3903
Show file tree

Hide file tree

Showing 3 changed files with 127 additions and 0 deletions.
diff --git a/iac/main/resources/state-machine.yml b/iac/main/resources/state-machine.yml
@@ -707,6 +707,7 @@ StepFunctionRedshiftProcessRole:
             - Effect: Allow
               Resource:
                 - !Sub 'arn:aws:states:${AWS::Region}:${AWS::AccountId}:stateMachine:${Environment}-dap-redshift-processing'
+                - !Sub 'arn:aws:states:${AWS::Region}:${AWS::AccountId}:stateMachine:${Environment}-dap-consolidated-stage-layer-to-redshift-processing'
 
               Action:
                 - states:ListExecutions
@@ -744,6 +745,7 @@ TxmaRawLayerConsolidatedSchemaProcessingStateMachine:
       GlueCrawlerRawName: !Ref RawLayerSingleTableCrawler
       RawToStageProcessingGlueJobname: !Ref RawStageTransformProcessPythonGlueJob
       DataQualityGlueJobname: !Ref DataQualityStageLayerOptimisedPythonGlueJob
+      RedshiftProcessingStepFunctionArn: !Ref RedshiftProcessingStateMachine
     Events:
       DailySchedule:
         Type: Schedule # More info about Schedule Event Source: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-property-statemachine-schedule.html
@@ -885,3 +887,28 @@ DataQualityStageLayerOptimisedPythonGlueJob:
     SecurityConfiguration: !Ref GlueSecurityConfig
     Name: !Sub ${Environment}-dap-data-quality-new-stage-metrics-generation
     Role: !Ref GlueScriptsExecutionRole
+
+RedshiftConsolidatedModelProcessingLogGroup:
+  Type: AWS::Logs::LogGroup
+  Properties:
+    LogGroupName: /aws/stepfunction/dap-consolidated-stage-layer-to-redshift
+    KmsKeyId: !GetAtt KmsKey.Arn
+    RetentionInDays: 30
+
+RedshiftConsolidatedModelProcessingStateMachine:
+  Type: AWS::Serverless::StateMachine # More info about State Machine Resource: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-statemachine.html
+  Properties:
+    DefinitionUri: statemachine/txma_redshift_consolidated_schema_processing.asl.json
+    Name: !Sub ${Environment}-dap-consolidated-stage-layer-to-redshift-processing
+    Role: !GetAtt StepFunctionRedshiftProcessRole.Arn
+    Logging:
+      Destinations:
+        - CloudWatchLogsLogGroup:
+            LogGroupArn: !GetAtt RedshiftConsolidatedModelProcessingLogGroup.Arn
+      IncludeExecutionData: true
+      Level: ALL
+    DefinitionSubstitutions:
+      S3MetaDataBucketName: !Ref ELTMetadataBucket
+      RedshiftWorkgroup: !Ref RedshiftServerlessWorkgroup
+      RedshiftDatabaseName: dap_txma_reporting_db_refactored
+      StateMachineResultsBucket: !Ref StateMachineResultsBucket
diff --git a/statemachine/txma_raw_layer_consolidated_schema_processing.asl.json b/statemachine/txma_raw_layer_consolidated_schema_processing.asl.json
@@ -98,6 +98,18 @@
       "Parameters": {
         "JobName": "${DataQualityGlueJobname}"
       },
+      "Next": "RedshiftProcessingStepFunction"
+    },
+    "RedshiftProcessingStepFunction": {
+      "Type": "Task",
+      "Resource": "arn:aws:states:::states:startExecution.sync:2",
+      "Parameters": {
+        "StateMachineArn": "${RedshiftProcessingStepFunctionArn}",
+        "Input": {
+          "StatePayload": "Triggered from raw to stage consolidated step function",
+          "AWS_STEP_FUNCTIONS_STARTED_BY_EXECUTION_ID.$": "$$.Execution.Id"
+        }
+      },
       "Next": "StopProcessing"
     },
     "StopProcessing": {

diff --git a/statemachine/txma_redshift_consolidated_schema_processing.asl.json b/statemachine/txma_redshift_consolidated_schema_processing.asl.json
@@ -0,0 +1,88 @@
+{
+    "Comment": "Redshit ELT processing workflow for loading dimension and fact tables",
+    "StartAt": "ListExecutions",
+    "States": {
+      "ListExecutions": {
+        "Type": "Task",
+        "Next": "ValidateRunningInstances",
+        "Parameters": {
+          "StateMachineArn.$": "$$.StateMachine.Id",
+          "StatusFilter": "RUNNING"
+        },
+        "Resource": "arn:aws:states:::aws-sdk:sfn:listExecutions",
+        "ResultSelector": {
+          "runningExecutionsCount.$": "States.ArrayLength($.Executions)"
+        }
+      },
+      "ValidateRunningInstances": {
+        "Type": "Choice",
+        "Choices": [
+          {
+            "Variable": "$.runningExecutionsCount",
+            "NumericGreaterThan": 1,
+            "Next": "RunningInstanceDetected"
+          }
+        ],
+        "Default": "dap_datamart_update"
+      },
+      "RunningInstanceDetected": {
+        "Type": "Fail",
+        "Error": "RunningInstanceDetected"
+      },
+      "dap_datamart_update": {
+        "Comment": "Invoke Redshift Dap DataMart update script",
+        "Type": "Task",
+        "Resource": "arn:aws:states:::aws-sdk:redshiftdata:executeStatement",
+        "ResultPath": "$.sql_output",
+        "Parameters": {
+          "WorkgroupName": "dev-redshift-serverless-workgroup",
+          "Database": "dap_txma_reporting_db_refactored",
+          "Sql": "call conformed_refactored.update_dap_data_mart()"
+        },
+        "Next": "wait_on_dap_datamart_update"
+      },
+      "wait_on_dap_datamart_update": {
+        "Comment": "Wait before status check",
+        "Type": "Wait",
+        "Seconds": 5,
+        "Next": "dap_datamart_update_status_check"
+      },
+      "dap_datamart_update_status_check": {
+        "Comment": "Check Task Status",
+        "Type": "Task",
+        "Resource": "arn:aws:states:::aws-sdk:redshiftdata:describeStatement",
+        "ResultPath": "$.sql_output",
+        "Parameters": {
+          "Id.$": "$.sql_output.Id"
+        },
+        "Next": "is_dap_datamart_update_complete"
+      },
+      "is_dap_datamart_update_complete": {
+        "Comment": "check if DAP Datamart update step is complete",
+        "Type": "Choice",
+        "Choices": [
+          {
+            "Variable": "$.sql_output.Status",
+            "StringEquals": "FAILED",
+            "Next": "dap_datamart_update_failure"
+          },
+          {
+            "Variable": "$.sql_output.Status",
+            "StringEquals": "FINISHED",
+            "Next": "StopProcessing"
+          }
+        ],
+        "Default": "wait_on_dap_datamart_update"
+      },
+      "dap_datamart_update_failure": {
+        "Type": "Fail",
+        "Cause": "Failure within DAP Datamart update_step",
+        "Error": "Error"
+      },
+      "StopProcessing": {
+        "Type": "Pass",
+        "Result": "pass",
+        "End": true
+      }
+    }
+  }