# Main Flow step function *Please refer the [Step Function Diagram](./step_function_diagramm.md)* ## Step function input To execute the step function one has to pass the proper input data. Here is the example input data structure: ```javascript { "settings" : { "isReprocessing" : "False", "PartnerKey" : "Spotify", "UnitOfWorkID" : "spotify-20190203-theorchard-streams-v2", "config_bucket" : "slz-test", "config_file" : "dummy.conf" } } ``` **isProcessing** ("True" or "False") - specifies if this is a reprocessing execution. In case of reprocessing more parameters should be additionally provided to specify the time frame etc. **PartnerKey** - unique identifier of data provider **UnitOfWorkID** - job identifier describing the exact job for the step function execution **config_bucket** and **config_file** parameters specify location of the main configuration file containing additional parameters ofr the step funaction execution ## Step 1: "Workflow setup" This is the main flow entry step where we can setup main workflow parameters (currently just the execution time) All parameters defined in thos step added to the output as a separate ``info`` section with the help of the following directive: ``"ResultPath": "$.info"`` Example step output: ```javascript { "name": "SetupWorkflow", "output": { "settings": { "isReprocessing": "False", "PartnerKey": "Spotify", "UnitOfWorkID": "spotify-20190203-theorchard-streams-v2", "config_bucket": "slz-test", "config_file": "dummy.conf" }, "info": { "note": "State machine workflow setup", "start_time": "2019-07-10T12:53:17.126Z" } } } ``` ## Step2: "Check Fargate Limits" This step executes Lambda function to check if we are running good inside the limit of started Fargate tasks. We expect that output of Lambda funaction contains top level element ``validate`` with ``status`` and ``message`` attributes: ```javascript "validate" : { "status" : "success", "message" : "Fargate limits OK" } ``` ``status`` must be ``success`` in positive case and anything else otherwise ``message`` contains notification string to be logged and sent to the support Slack channel The ``validate`` element will be added in the step output. As you will see below, it's a common approach in the Main Flow step function to gather result of a Lambda function and take a decision of further routing. Example step output: ```javascript { "name": "CheckFargateLimits", "output": { "settings": { "isReprocessing": "False", "PartnerKey": "Spotify", "UnitOfWorkID": "spotify-20190203-theorchard-streams-v2", "config_bucket": "slz-test", "config_file": "dummy.conf" }, "info": { "note": "State machine workflow setup", "start_time": "2019-07-10T12:53:17.126Z" }, "result": { "validate": { "status": "success", "message": "Fargate limits OK" } } } } ``` ## Step 3: Validate Fargate limits Conditional routing step. If check Fargate limits result status is “success” - we proceed to the next step (Read Configuration). Otherwise step function routes to the "Failure chain" - several steps responsible to update task status in DynamoDB, send Slack notification and stop the step function with the "Failed" status. ## Step 4: Read configuration This step executes Lambda function which locates additional configuration file in S3 using the **config_bucket** and **config_file** parameters and adds it's content to the step output. Lambda function must return the configuration details added in the top level ``config`` and task result - in the top level ``validate`` sections. These sections added in the ``result`` section of the step output with the help of the ``"ResultPath": "$.result"`` directive. ``validate`` section must contain usual attributes: ``status`` and ``message`` describing the Lambda execution results. Example step output: ```javascript { "name": "ReadConfiguration", "output": { "settings": { "isReprocessing": "False", "PartnerKey": "Spotify", "UnitOfWorkID": "spotify-20190203-theorchard-streams-v2", "config_bucket": "slz-test", "config_file": "dummy.conf" }, "info": { "note": "State machine workflow setup", "start_time": "2019-07-10T12:53:17.126Z" }, "result": { "config": { "Command": [ "--url=http://sdm.lbl.gov/fastbit/data/star2000.csv.gz", "--s3-bucket=stepf-poc" ], "LaunchType": "FARGATE", "Subnets": [ "subnet-01646dcbdf59567c6" ], "Cluster": "arn:aws:ecs:us-east-1:475275892927:cluster/slz", "TaskDefinition": "arn:aws:ecs:us-east-1:475275892927:task-definition/sfpoc-downloader:2", "ReprocessingStartMessage": { "message": "start reprocessing" }, "ReprocessingEndMessage": { "message": "end reprocessing" } }, "validate": { "status": "success", "message": "configuration parameters successfully downloaded" } } } } ``` ## Step 5: Validate Configuration Conditional routing step. If read configuration result status is “success” - we proceed to the next step (Check Reprocessing). Otherwise step function routes to the "Failure chain" - several steps responsible to update task status in DynamoDB, send Slack notification and stop the step function with the "Failed" status. ## Step 6: Check Reprocessing Conditional routing step checking the ``settings/isProcessing`` variable (see **Step Function input** chapter). If this variable exists and has a value of ``"True"`` then execution switches to the *Reprocessing start* route. Otherwise execution continues to the **Start Download** step ## Steps 7-8: Reprocessing start chain If our step function has been invoked with the ``"isReprocessing" : "True"`` we have to perform two additional steps before continue: **Reprocessing Start Notification** - here we send message to the special SNS topic for all possible subscribers who might be interested in the reProcessing event. > Note: notification message JSON object need to be specified in settings or configuration parameters before. **CleanUp** - here we invoke the Lambda function which stops all running step function executions and deletes all teh file related to the particular job reprocessing. Lambda function should inform us on its results in any form. > Final output structure of the CleanUp Lambda is not yet defined Example CleanUp Lambda output: ```javascript "cleanup" : { "jobs_deleted" : 2, "files_deleted" : 5 }, "validate" : { "status" : "success", "message" : "jobs cleaned OK" } ``` ## Step 9: Start Download ## This step starts content download in docker container as a Fargate task. All the parameters of Fargate task need to be specified in configuration file. For example: ```javascript "Command" : [ "--url=http://sdm.lbl.gov/fastbit/data/star2000.csv.gz", "--s3-bucket=stepf-poc" ], "LaunchType" : "FARGATE", "Subnets" : ["subnet-01646dcbdf59567c6"], "Cluster" : "arn:aws:ecs:us-east-1:475275892927:cluster/slz", "TaskDefinition" : "arn:aws:ecs:us-east-1:475275892927:task-definition/sfpoc-downloader:2" ``` > Parameters in this example define PoC simple download task. They will be different for the real task imlementation. Inside the step function this task started with the ``waitForTaskToken`` option. So the step function execution will be paused before the download task resume it by remote call with the Task Token. The Task Token delivered to the download task via environment variable ``SFN_TASK_TOKEN`` Step function expects that Fargate task returns result as JSON with the top-level ``result`` element containg ``status`` attribute. ``"status" : "success"`` means file was downloaded OK. Additional attributes might be returned inside the ``download`` top-level element. > Currently the PoC Fargate task returns ``status`` as a separate element without ``result`` section. So step function taking care of this with the help of the ``"ResultPath": "$.result.validate"`` directive instead of usual ``"ResultPath": "$.result"`` This should be changed later. ## Step 10: Validate Download ## Conditional routing step. If download result status is “success” - we proceed to the next step (Check Job Done). Otherwise step function routes to the "Failure chain" - several steps responsible to update task status in DynamoDB, send Slack notification and stop the step function with the "Failed" status. ## Step 11: Check Job Done ## This step executes Lambda function to check completness criteria of the current job - depending on the list of downloaded file and/or current time job completeness status could be defined as "Not done", "Min done" or "Done". This step expects the ``settings.UnitOfWorkID`` parameter specified before the step function execution. Lambda function must return JSON output with two top-level elements" ``job`` containg job details and ``validate`` containing execution status result. Example lambda output: ```javascript "job" : { "job_status" : "In progress", "job_complete" : "Done", "PartnerKey" : "Spotify", "UnitOfWorkID" : "spotify-20191115-theorchard-streams-v2" }, "validate" : { "status" : "success", "message" : "job status checked OK" } ``` Example step output: ```javascript { "name": "CheckJobDone", "output": { "settings": { "isReprocessing": "False", "PartnerKey": "Spotify", "UnitOfWorkID": "spotify-20190203-theorchard-streams-v2", "config_bucket": "slz-test", "config_file": "dummy.conf" }, "info": { "note": "State machine workflow setup", "start_time": "2019-07-10T12:53:17.126Z" }, "result": { "job": { "job_status": "In progress", "job_complete": "Done", "PartnerKey": "Spotify", "UnitOfWorkID": "spotify-20191115-theorchard-streams-v2" }, "validate": { "status": "success", "message": "job status checked OK" } } } } ``` ## Step 12: Validate UoW Done ## Conditional routing step. If CheckJobDone lambda returns "Done" as the completeness status for the current job - we need to update it in the DynamoDB. So in this case step function routes to the "UpdateJobDoneStatus" step. Otherwise this step could be skiped and execution continues with the next step - "Reprocessing End chain" ## Step 13: Update UoW Done Status ## This step directly updates the *UoWCompletenessStatus* attribute in the DynamoDB table *slz_uow_status* using the primary index (*PartnerKey, UnitOfWorkID*) Value of the *UoWCompletenessStatus* is taken from the previos step output(``result/job/job_complete``) ## Steps 14-15: Reprocessing end chain If our step function has been invoked with the ``"isReprocessing" : "True"`` we have to notify all possible subscribers that reprocessing is now finished: **Reprocessing End Notification** - here we send message to the special SNS topic for all possible subscribers who might be interested in the reProcessing event. > Note: notification message JSON object need to be specified in settings or configuration parameters before. ## Step 16: Success Stop This is the final step successfully closing the main flow ## "Failure Chain" In case of any external failure when executed lambda function returns result status different from ``"success`` we are routing step function to these chain containing the following steps: **UpdateUoWStatusIdle** - here we update the current job item in the DynamoDB *slz_uow_status* table setting job_status to ``IDLE`` **FailureSlackNotify** - executes special Lambda function publishing the failure message in Slack channel > Note: this lambda function may require additional configuration parameters not specified in current step function **Fail Stop** - the final step closing the main flow with *Failed* status # Migration from DynamoDB to Postgres Migration will be done in a multiple stages. Stage should be assigned to the environment variable and used everywhere across the application. 1. Read and write from DynamoDB (STORAGE_MIGRATION_STAGE=0 - default) 2. Read and write from DynamoDB, write to Postgres (STORAGE_MIGRATION_STAGE=1) 3. Read and write from Postgres, write to DynamoDB (STORAGE_MIGRATION_STAGE=2) 4. Read and write from Postgres (STORAGE_MIGRATION_STAGE=3)