POC: status updates

.dockerignore

@@ -9,3 +9,6 @@ terraform/.terraform/*
 .vscode/settings.json
 terraform/*/.terraform/*
 application-deployments/*/*/providers/*
+application-deployments/*
+terraform/application-wrapper/applications/echo-app/providers/registry.terraform.io/hashicorp/aws/5.36.0/linux_arm64/terraform-provider-aws_v5.36.0_x5
+terraform/application-wrapper/applications/*/.terraform/providers/*

.gitignore

@@ -41,3 +41,14 @@ terraform/remote-state-application/terraform.tfstate.backup
 terraform/remote-state-application/tfplan
 terraform/remote-state-application/.terraform/providers/registry.terraform.io/hashicorp/archive/2.4.2/linux_arm64/terraform-provider-archive_v2.4.2_x5
 terraform/remote-state-application/.terraform/providers/registry.terraform.io/hashicorp/random/3.6.0/linux_arm64/terraform-provider-random_v3.6.0_x5
+terraform/application-state/.terraform.lock.hcl
+terraform/application-state/terraform.tfstate.backup
+terraform/application-state/.terraform/providers/registry.terraform.io/hashicorp/archive/2.4.2/linux_arm64/terraform-provider-archive_v2.4.2_x5
+terraform/application-state/.terraform/providers/registry.terraform.io/hashicorp/random/3.6.0/linux_arm64/terraform-provider-random_v3.6.0_x5
+terraform/application-state/tfplan
+terraform/status-service/.terraform.lock.hcl
+terraform/status-service/tfplan
+terraform/status-service/.terraform/providers/registry.terraform.io/hashicorp/archive/2.4.2/linux_arm64/terraform-provider-archive_v2.4.2_x5
+terraform/status-service/.terraform/providers/registry.terraform.io/hashicorp/aws/5.36.0/linux_arm64/terraform-provider-aws_v5.36.0_x5
+terraform/status-service/.terraform/providers/registry.terraform.io/hashicorp/random/3.6.0/linux_arm64/terraform-provider-random_v3.6.0_x5
+terraform/status-service/terraform.tfstate.backup

Makefile

@@ -41,8 +41,14 @@ create-application:
 destroy-application:
 	docker-compose run app-deploy -cmd destroy-application
 
-create-remote-state-app:
-	docker-compose run app-deploy -cmd create-remote-state-app	
+create-application-state-app:
+	docker-compose run app-deploy -cmd create-application-state-app
+
+create-status-service:
+	docker-compose run app-deploy -cmd create-status-service
+
+destroy-status-service:
+	docker-compose run app-deploy -cmd destroy-status-service	
 
 deploy:
 	aws ecr get-login-password --profile ${AWS_PROFILE} --region ${AWS_DEFAULT_REGION} | docker login --username AWS --password-stdin ${ACCOUNT}.dkr.ecr.${AWS_DEFAULT_REGION}.amazonaws.com

main.go

@@ -8,9 +8,10 @@ import (
 )
 
 var TerraformStateDirectory = "/service/terraform/remote-state"
-var TerraformAppStateDirectory = "/service/terraform/remote-state-application"
+var TerraformAppStateDirectory = "/service/terraform/application-state"
 var TerraformGatewayDirectory = "/service/terraform/internet-gateway"
 var TerraformApplicationDirectory = "/service/terraform/application-wrapper"
+var TerraformStatusServiceDirectory = "/service/terraform/status-service"
 
 func main() {
 	cmdPtr := flag.String("cmd", "plan", "command to execute")
@@ -28,8 +29,8 @@ func main() {
 	}
 
 	// Remote State Application Management
-	if *cmdPtr == "create-remote-state-app" || *cmdPtr == "remote-state-app" {
-		cmd := exec.Command("/bin/sh", "./scripts/remote-state-application.sh", TerraformAppStateDirectory, *cmdPtr)
+	if *cmdPtr == "create-application-state-app" || *cmdPtr == "delete-application-state-app" {
+		cmd := exec.Command("/bin/sh", "./scripts/application-state-app.sh", TerraformAppStateDirectory, *cmdPtr)
 		out, err := cmd.Output()
 		if err != nil {
 			log.Fatalf("error %s", err)
@@ -71,5 +72,16 @@ func main() {
 		}
 	}
 
+	// status service
+	if *cmdPtr == "create-status-service" || *cmdPtr == "destroy-status-service" {
+		cmd := exec.Command("/bin/sh", "./scripts/status-service.sh", TerraformStatusServiceDirectory, *cmdPtr)
+		out, err := cmd.Output()
+		output := string(out)
+		fmt.Println(output)
+		if err != nil {
+			log.Fatalf("error %s", err.Error())
+		}
+	}
+
 	log.Println("done")
 }

scripts/remote-state-application.sh → scripts/application-state-app.sh

@@ -2,7 +2,7 @@
 
 cd $1
 
-if [ $2 = "create-remote-state-app" ]; then
+if [ $2 = "create-application-state-app" ]; then
   echo "creating ..."
   terraform init
   terraform plan -out=tfplan > plan.log

scripts/infrastructure.sh

@@ -44,6 +44,7 @@ post_processor_cpu = "${POST_PROCESSOR_CPU:-2048}"
 post_processor_memory = "${POST_PROCESSOR_MEMORY:-4096}"
 wm_cpu = "${WM_CPU:-2048}"
 wm_memory = "${WM_MEMORY:-4096}"
+status_api_host = "${STATUS_API_HOST}"
 EOL
 
 if [ $1 = "destroy" ]; then

scripts/status-service.sh

@@ -0,0 +1,16 @@
+#!/bin/sh
+
+cd $1
+
+if [ $2 = "create-status-service" ]; then
+  echo "creating ..."
+  # terraform validate
+  terraform init > init.log
+  export TF_LOG_PATH="error.log"
+  export TF_LOG=TRACE
+  terraform plan -out=tfplan > plan.log
+  terraform apply tfplan
+else
+  echo "deleting ..."
+  terraform apply -destroy -auto-approve
+fi

terraform/application-state/README.md

@@ -0,0 +1 @@
+Application that keeps tracko of applications state, listens for S3 events and inserts applicat details into a Dynamo DB table.

terraform/remote-state-application/lambda.tf → terraform/application-state/lambda.tf

@@ -52,7 +52,6 @@ resource "aws_iam_role" "iam_for_lambda" {
   assume_role_policy = data.aws_iam_policy_document.assume_role.json
 }
 
-// attach policy to allow state lambda to start an ECS task and to write to Cloudwatch
 resource "aws_iam_role_policy_attachment" "lambda_policy_ecs" {
   role       = aws_iam_role.iam_for_lambda.name
   policy_arn = aws_iam_policy.lambda_iam_policy.arn

terraform/ecs.tf

@@ -153,7 +153,8 @@ resource "aws_ecs_task_definition" "workflow-manager" {
       { name: "PENNSIEVE_API_HOST", value: var.api_host},
       { name: "PENNSIEVE_API_HOST2", value: var.api_host2},
       { name: "BASE_DIR", value: "/mnt/efs"},
-      { name: "REGION", value: var.region}
+      { name: "REGION", value: var.region},
+      { name: "PENNSIEVE_STATUS_HOST", value: var.status_api_host},
       ],
       essential = true
       portMappings = [

terraform/status-service/README.md

@@ -0,0 +1 @@
+Status service

terraform/status-service/dynamodb.tf

@@ -0,0 +1,15 @@
+resource "aws_dynamodb_table" "statuses_table" {
+  name           = "statuses-table"
+  billing_mode   = "PAY_PER_REQUEST"
+  hash_key       = "task_id"
+
+  attribute {
+    name = "task_id"
+    type = "S"
+  }
+
+  ttl {
+    attribute_name = "TimeToExist"
+    enabled        = true
+  }
+}

terraform/status-service/lambda.tf

@@ -0,0 +1,137 @@
+// Application Gateway Lambda
+resource "aws_lambda_function" "status_service" {
+  function_name = "status-service-${random_uuid.val.id}"
+  role          = aws_iam_role.iam_for_lambda.arn
+  handler       = "status-service.lambda_handler" # module is name of python file: application
+  description   = "Status Service"
+
+  s3_bucket = aws_s3_bucket.lambda_bucket.id
+  s3_key    = aws_s3_object.status_service_lambda.key
+
+  source_code_hash = data.archive_file.status_service_lambda.output_base64sha256
+
+  runtime = "python3.12"
+  timeout = 60
+
+  environment {
+    variables = {
+      STATUSES_TABLE = aws_dynamodb_table.statuses_table.name,
+    }
+  }
+}
+
+resource "aws_cloudwatch_log_group" "status_service-lambda" {
+  name = "/aws/lambda/${aws_lambda_function.status_service.function_name}"
+
+  retention_in_days = 30
+}
+
+resource "aws_lambda_function_url" "status_service" {
+  function_name      = aws_lambda_function.status_service.function_name
+  authorization_type = "NONE"
+}
+
+# IAM
+// Lambda gateway function
+// allow lambda to access resources in your AWS account
+data "aws_iam_policy_document" "assume_role" {
+  statement {
+    effect = "Allow"
+
+    principals {
+      type        = "Service"
+      identifiers = ["lambda.amazonaws.com"]
+    }
+
+    actions = ["sts:AssumeRole"]
+  }
+}
+
+resource "aws_iam_role" "iam_for_lambda" {
+  name               = "iam_for_lambda-${random_uuid.val.id}"
+  assume_role_policy = data.aws_iam_policy_document.assume_role.json
+}
+
+resource "aws_iam_role_policy_attachment" "status_lambda_policy" {
+  role       = aws_iam_role.iam_for_lambda.name
+  policy_arn = aws_iam_policy.lambda_iam_policy.arn
+}
+
+resource "aws_iam_policy" "lambda_iam_policy" {
+  name   = "lambda-iam-policy-${random_uuid.val.id}"
+  path   = "/"
+  policy = data.aws_iam_policy_document.iam_policy_document_status.json
+}
+
+# data
+// creates an archive and uploads to s3 bucket
+data "archive_file" "status_service_lambda" {
+  type = "zip"
+
+  source_dir  = "${path.module}/status-service-lambda"
+  output_path = "${path.module}/status-service-lambda.zip"
+}
+
+// provides an s3 object resource
+resource "aws_s3_object" "status_service_lambda" {
+  bucket = aws_s3_bucket.lambda_bucket.id
+
+  key    = "status-service-lambda.zip"
+  source = data.archive_file.status_service_lambda.output_path
+
+  etag = filemd5(data.archive_file.status_service_lambda.output_path)
+}
+
+// policy document - status service lambda
+data "aws_iam_policy_document" "iam_policy_document_status" {
+  statement {
+    sid    = "CloudwatchPermissions"
+    effect = "Allow"
+    actions = [
+      "logs:CreateLogGroup",
+      "logs:CreateLogStream",
+      "logs:PutLogEvents"
+    ]
+    resources = ["*"]
+  }
+
+   statement {
+    sid = "LambdaAccessToDynamoDB"
+    effect = "Allow"
+
+    actions = [
+      "dynamodb:BatchGetItem",
+      "dynamodb:GetItem",
+      "dynamodb:Query",
+      "dynamodb:Scan",
+      "dynamodb:BatchWriteItem",
+      "dynamodb:PutItem",
+      "dynamodb:UpdateItem"
+    ]
+
+    resources = [
+      aws_dynamodb_table.statuses_table.arn,
+      "${aws_dynamodb_table.statuses_table.arn}/*"
+    ]
+
+  }
+}
+
+// S3 bucket
+resource "aws_s3_bucket" "lambda_bucket" {
+  bucket = random_uuid.val.id
+}
+
+resource "aws_s3_bucket_ownership_controls" "lambda_bucket" {
+  bucket = aws_s3_bucket.lambda_bucket.id
+  rule {
+    object_ownership = "BucketOwnerPreferred"
+  }
+}
+
+resource "aws_s3_bucket_acl" "bucket_acl" {
+  depends_on = [aws_s3_bucket_ownership_controls.lambda_bucket]
+
+  bucket = aws_s3_bucket.lambda_bucket.id
+  acl    = "private"
+}

terraform/status-service/main.tf

@@ -0,0 +1,4 @@
+provider "aws" {}
+
+resource "random_uuid" "val" {
+}

terraform/status-service/outputs.tf

@@ -0,0 +1,5 @@
+output "lambda_function_url" {
+  description = "Lambda URL"
+
+  value = aws_lambda_function_url.status_service.function_url
+}

terraform/status-service/status-service-lambda/status-service.py

@@ -0,0 +1,35 @@
+from boto3 import client as boto3_client
+import json
+import base64
+import os
+
+dynamodb = boto3_client('dynamodb')
+
+def lambda_handler(event, context):
+
+    print(event)
+
+    if event['isBase64Encoded'] == True:
+        body = base64.b64decode(event['body']).decode('utf-8')
+        event['body'] = body
+        event['isBase64Encoded'] = False
+    json_body = json.loads(event['body'])
+    print(json_body)
+
+    table_name = os.environ['STATUSES_TABLE']
+
+    response = dynamodb.put_item(
+            TableName=table_name,
+            Item={
+                "task_id": {'S':json_body['task_id']} ,
+                "integration_id": {'S':json_body['integration_id']},
+                "description": {'S':json_body['description']},
+            }
+        )
+
+    print(response)
+
+    return {
+        'statusCode': 200,
+        'body': json.dumps(str('Status updated'))
+    }

terraform/variables.tf

@@ -61,4 +61,7 @@ variable "wm_cpu" {
 }
 variable "wm_memory" {
     type = number
-}
+}
+variable "status_api_host" {
+    type = string
+}

terraform/workflow-manager/main.go

@@ -123,7 +123,7 @@ func processSQS(ctx context.Context, sqsSvc *sqs.Client, queueUrl string, logger
 				logger.Error(err.Error(),
 					slog.String("error", stderr.String()))
 			}
-			log.Println(stdout.String())
+			fmt.Println(stdout.String())
 
 			_, err = sqsSvc.DeleteMessage(ctx, &sqs.DeleteMessageInput{
 				QueueUrl:      &queueUrl,

terraform/workflow-manager/taskRunner/main.py

@@ -14,8 +14,8 @@ def main():
     subnet_ids = os.environ['SUBNET_IDS']
     cluster_name = os.environ['CLUSTER_NAME']
     security_group = os.environ['SECURITY_GROUP_ID']
-    pennsieve_host = os.environ['PENNSIEVE_API_HOST']
     pennsieve_host2 = os.environ['PENNSIEVE_API_HOST2']
+    pennsieve_status_host = os.environ['PENNSIEVE_STATUS_HOST']
 
     inputDir = sys.argv[1]
     outputDir = sys.argv[2]
@@ -65,6 +65,12 @@ def main():
         })
         task_arn = response['tasks'][0]['taskArn']
 
+        # POST at start of task - check (for success) if task_arn is present
+        data = { 'task_id': task_arn, 'integration_id': integration_id, 'description': 'main'}
+        r = requests.post(pennsieve_status_host, json=data)
+        r.raise_for_status()
+        print(r.json())
+
         waiter = ecs_client.get_waiter('tasks_stopped')
         waiter.wait(
             cluster=cluster_name,