fix tf apply errors

Author: agnesgaroux

catalogue_graph/terraform/iam_state_machines.tf

@@ -8,7 +8,10 @@ resource "aws_iam_role" "state_machine_execution_role" {
       {
         Effect = "Allow",
         Principal = {
-          Service = "states.amazonaws.com"
+          Service = [
+            "states.amazonaws.com",
+            "scheduler.amazonaws.com"
+          ]
         },
         Action = "sts:AssumeRole"
       }
@@ -31,8 +34,12 @@ resource "aws_iam_policy" "state_machine_policy" {
         Resource = [
           aws_sfn_state_machine.catalogue_graph_extractor.arn,
           aws_sfn_state_machine.catalogue_graph_extractors.arn,
+          aws_sfn_state_machine.catalogue_graph_extractors_monthly.arn,
+          aws_sfn_state_machine.catalogue_graph_extractors_daily.arn,
           aws_sfn_state_machine.catalogue_graph_bulk_loader.arn,
           aws_sfn_state_machine.catalogue_graph_bulk_loaders.arn,
+          aws_sfn_state_machine.catalogue_graph_bulk_loaders_monthly.arn,
+          aws_sfn_state_machine.catalogue_graph_bulk_loaders_daily.arn,
           aws_sfn_state_machine.catalogue_graph_ingestor.arn,
           aws_sfn_state_machine.concepts_pipeline_monthly.arn,
           aws_sfn_state_machine.concepts_pipeline_daily.arn

catalogue_graph/terraform/state_machine_bulk_loaders.tf

@@ -1,3 +1,32 @@
+resource "aws_sfn_state_machine" "catalogue_graph_bulk_loaders" {
+  name     = "catalogue-graph-bulk-loaders"
+  role_arn = aws_iam_role.state_machine_execution_role.arn
+
+  definition = jsonencode({
+    Comment = "Trigger the catalogue-graph-bulk-loader state machine in sequence for each combination of inputs."
+    StartAt = "Load ${var.state_machine_inputs[0].label}"
+    States = merge(tomap({
+      for index, task_input in var.state_machine_inputs :
+      "Load ${task_input.label}" => {
+        Type     = "Task"
+        Resource = "arn:aws:states:::states:startExecution.sync:2",
+        Parameters = {
+          StateMachineArn = aws_sfn_state_machine.catalogue_graph_bulk_loader.arn
+          Input = {
+            "transformer_type" = task_input.transformer_type,
+            "entity_type"      = task_input.entity_type
+          }
+        }
+        Next = index == length(var.state_machine_inputs) - 1 ? "Success" : "Load ${var.state_machine_inputs[index + 1].label}"
+      }
+      }), {
+      Success = {
+        Type = "Succeed"
+      }
+    })
+
+  })
+}
 resource "aws_sfn_state_machine" "catalogue_graph_bulk_loaders_monthly" {
   name     = "catalogue-graph-bulk-loaders_monthly"
   role_arn = aws_iam_role.state_machine_execution_role.arn

catalogue_graph/terraform/state_machine_concepts_pipeline.tf

@@ -4,7 +4,6 @@ resource "aws_sfn_state_machine" "concepts_pipeline_monthly" {
 
   definition = jsonencode({
     Comment = "Extract raw concepts from external sources, transform them into nodes and edges, and load them into the graph",
-    QueryLanguage = "JSONata"
     StartAt = "Extractors"
     States = {
       "Extractors" = {
@@ -36,7 +35,6 @@ resource "aws_sfn_state_machine" "concepts_pipeline_daily" {
 
   definition = jsonencode({
     Comment = "Extract concepts from catalogue works, load them into the graph, and ingests into ES index.",
-    QueryLanguage = "JSONata"
     StartAt = "Extractors"
     States = {
       "Extractors" = {
@@ -58,7 +56,7 @@ resource "aws_sfn_state_machine" "concepts_pipeline_daily" {
       "Concepts ingestor" = {
         Type     = "Task"
         Resource = "arn:aws:states:::states:startExecution.sync:2",
-        Arguments = {
+        Parameters = {
           StateMachineArn = aws_sfn_state_machine.catalogue_graph_ingestor.arn,
         }
         Next = "Success"

catalogue_graph/terraform/state_machine_extractors.tf

@@ -1,3 +1,34 @@
+resource "aws_sfn_state_machine" "catalogue_graph_extractors" {
+  name     = "catalogue-graph-extractors"
+  role_arn = aws_iam_role.state_machine_execution_role.arn
+
+  definition = jsonencode({
+    Comment = "Extract raw concepts from all sources, transform them into nodes and edges, and stream them into an S3 bucket."
+    StartAt = "Extract ${var.state_machine_inputs[0].label}"
+
+    States = merge(tomap({
+      for index, task_input in var.state_machine_inputs :
+      "Extract ${task_input.label}" => {
+        Type     = "Task"
+        Resource = "arn:aws:states:::states:startExecution.sync:2",
+        Parameters = {
+          StateMachineArn = aws_sfn_state_machine.catalogue_graph_extractor.arn
+          Input = {
+            "stream_destination" : "s3",
+            "transformer_type.$" : "$$.Execution.Input.transformer_type",
+            "entity_type.$" : "$$.Execution.Input.entity_type",
+            "sample_size.$" : "$$.Execution.Input.sample_size"
+          }
+        }
+        Next = index == length(var.state_machine_inputs) - 1 ? "Success" : "Extract ${var.state_machine_inputs[index + 1].label}"
+      }
+      }), {
+      Success = {
+        Type = "Succeed"
+      }
+    })
+  })
+}
 resource "aws_sfn_state_machine" "catalogue_graph_extractors_monthly" {
   name     = "catalogue-graph-extractors_monthly"
   role_arn = aws_iam_role.state_machine_execution_role.arn