[sdlf-stage-glue] remove pDataset parameter

it was wrongly used in place of pBucketPrefix

and unnecessarily used in resource names

Author: cnfait

sdlf-stage-glue/src/glue.yaml

@@ -32,6 +32,10 @@ Parameters:
     Description: KMS key set as infra key for the dataset
     Type: String
     Default: "" # if not provided, pDatasetDeploymentInstance must be specified
+  pBucketPrefix:
+    Description: Bucket prefix for the dataset
+    Type: String
+    Default: "" # if not provided, pDatasetDeploymentInstance must be specified
   pEventBus:
     Description: EventBridge bus for the dataset
     Type: String
@@ -56,10 +60,6 @@ Parameters:
   #   Description: Analytics bucket
   #   Type: String
   #   Default: "" # if not provided, pStorageDeploymentInstance must be specified
-  pDataset:
-    Description: The name of the dataset (all lowercase, no symbols or spaces)
-    Type: String
-    AllowedPattern: "[a-z0-9]{2,14}"
   pStageEnabled:
     Description: Whether the stage is enabled or not
     Type: String
@@ -158,7 +158,7 @@ Globals:
       - "{{resolve:ssm:/SDLF/Lambda/LatestDatalakeLibraryLayer}}"
     Environment:
       Variables:
-        DATASET: !Ref pDataset
+        S3_PREFIX: !If [FetchFromDatasetSsm, !Sub "{{resolve:ssm:/sdlf/dataset/rS3Prefix/${pDatasetDeploymentInstance}}}", !Ref pBucketPrefix]
         DEPLOYMENT_INSTANCE: !Ref pDeploymentInstance
         STORAGE_DEPLOYMENT_INSTANCE: !Ref pStorageDeploymentInstance
         DATASET_DEPLOYMENT_INSTANCE: !Ref pDatasetDeploymentInstance
@@ -182,7 +182,6 @@ Resources:
       pInfraKmsKey: !Ref pInfraKmsKey
       pEventBus: !Ref pEventBus
       pScheduleGroup: !Ref pScheduleGroup
-      pDataset: !Ref pDataset
       pStageEnabled: !Ref pStageEnabled
       pTriggerType: !Ref pTriggerType
       pSchedule: !Ref pSchedule
@@ -193,7 +192,7 @@ Resources:
   rLambdaCommonPolicy:
     Type: AWS::IAM::ManagedPolicy
     Properties:
-      Path: !Sub /sdlf-${pDataset}/
+      Path: !Sub /sdlf-${pDeploymentInstance}/
       PolicyDocument:
         Version: "2012-10-17"
         Statement:
@@ -203,7 +202,7 @@ Resources:
               - logs:CreateLogStream
               - logs:PutLogEvents
             Resource:
-              - !Sub arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/sdlf-${pDataset}-${pDeploymentInstance}-*
+              - !Sub arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/sdlf-${pDeploymentInstance}-*
           - Effect: Allow
             Action:
               - ssm:GetParameter
@@ -238,7 +237,7 @@ Resources:
   rRoleLambdaExecutionRoutingStep:
     Type: AWS::IAM::Role
     Properties:
-      Path: !Sub /sdlf-${pDataset}/
+      Path: !Sub /sdlf-${pDeploymentInstance}/
       # PermissionsBoundary: !Sub "{{resolve:ssm:/SDLF/IAM/${pDataset}/TeamPermissionsBoundary}}"
       ManagedPolicyArns:
         - !Ref rLambdaCommonPolicy
@@ -254,7 +253,7 @@ Resources:
               Service: lambda.amazonaws.com
             Action: sts:AssumeRole
       Policies:
-        - PolicyName: !Sub sdlf-${pDataset}-${pDeploymentInstance}-routing
+        - PolicyName: !Sub sdlf-${pDeploymentInstance}-routing
           PolicyDocument:
             Version: "2012-10-17"
             Statement:
@@ -274,14 +273,14 @@ Resources:
                   - sqs:ReceiveMessage
                   - sqs:SendMessage
                 Resource:
-                  - !Sub arn:${AWS::Partition}:sqs:${AWS::Region}:${AWS::AccountId}:sdlf-${pDataset}-${pDeploymentInstance}-queue-*
-                  - !Sub arn:${AWS::Partition}:sqs:${AWS::Region}:${AWS::AccountId}:sdlf-${pDataset}-${pDeploymentInstance}-dlq-*
+                  - !Sub arn:${AWS::Partition}:sqs:${AWS::Region}:${AWS::AccountId}:sdlf-${pDeploymentInstance}-queue.fifo
+                  - !Sub arn:${AWS::Partition}:sqs:${AWS::Region}:${AWS::AccountId}:sdlf-${pDeploymentInstance}-dlq.fifo
 
   # Metadata Step Role (fetch metadata, update pipeline execution history...)
   rRoleLambdaExecutionMetadataStep:
     Type: AWS::IAM::Role
     Properties:
-      Path: !Sub /sdlf-${pDataset}/
+      Path: !Sub /sdlf-${pDeploymentInstance}/
       # PermissionsBoundary: !Sub "{{resolve:ssm:/SDLF/IAM/${pDataset}/TeamPermissionsBoundary}}"
       ManagedPolicyArns:
         - !Ref rLambdaCommonPolicy
@@ -301,7 +300,7 @@ Resources:
   rRoleLambdaExecutionErrorStep:
     Type: AWS::IAM::Role
     Properties:
-      Path: !Sub /sdlf-${pDataset}/
+      Path: !Sub /sdlf-${pDeploymentInstance}/
       # PermissionsBoundary: !Sub "{{resolve:ssm:/SDLF/IAM/${pDataset}/TeamPermissionsBoundary}}"
       ManagedPolicyArns:
         - !Ref rLambdaCommonPolicy
@@ -317,7 +316,7 @@ Resources:
               Service: lambda.amazonaws.com
             Action: sts:AssumeRole
       Policies:
-        - PolicyName: !Sub sdlf-${pDataset}-${pDeploymentInstance}-error
+        - PolicyName: !Sub sdlf-${pDeploymentInstance}-error
           PolicyDocument:
             Version: "2012-10-17"
             Statement:
@@ -332,7 +331,7 @@ Resources:
                   - sqs:ReceiveMessage
                   - sqs:SendMessage
                 Resource:
-                  - !Sub arn:${AWS::Partition}:sqs:${AWS::Region}:${AWS::AccountId}:sdlf-${pDataset}-${pDeploymentInstance}-dlq-*
+                  - !Sub arn:${AWS::Partition}:sqs:${AWS::Region}:${AWS::AccountId}:sdlf-${pDeploymentInstance}-dlq.fifo
 
   ######## LAMBDA FUNCTIONS #########
   rLambdaRoutingStep:
@@ -344,7 +343,7 @@ Resources:
             reason: Permissions to write CloudWatch Logs are granted by rLambdaCommonPolicy
     Properties:
       CodeUri: ./lambda/routing/src
-      FunctionName: !Sub sdlf-${pDataset}-${pDeploymentInstance}-routing
+      FunctionName: !Sub sdlf-${pDeploymentInstance}-routing
       Description: Checks if items are to be processed and route them to state machine
       Environment:
         Variables:
@@ -362,7 +361,7 @@ Resources:
             reason: Permissions to write CloudWatch Logs are granted by rLambdaCommonPolicy
     Properties:
       CodeUri: ./lambda/redrive/src
-      FunctionName: !Sub sdlf-${pDataset}-${pDeploymentInstance}-redrive
+      FunctionName: !Sub sdlf-${pDeploymentInstance}-redrive
       Description: Redrives Failed messages to the routing queue
       MemorySize: 192
       Timeout: 300
@@ -377,7 +376,7 @@ Resources:
             reason: Permissions to write CloudWatch Logs are granted by rLambdaCommonPolicy
     Properties:
       CodeUri: ./lambda/postupdate-metadata/src
-      FunctionName: !Sub sdlf-${pDataset}-${pDeploymentInstance}-postupdate
+      FunctionName: !Sub sdlf-${pDeploymentInstance}-postupdate
       Description: Post-Update the metadata in the DynamoDB Catalog table
       MemorySize: 192
       Timeout: 300
@@ -392,7 +391,7 @@ Resources:
             reason: Permissions to write CloudWatch Logs are granted by rLambdaCommonPolicy
     Properties:
       CodeUri: ./lambda/error/src
-      FunctionName: !Sub sdlf-${pDataset}-${pDeploymentInstance}-error
+      FunctionName: !Sub sdlf-${pDeploymentInstance}-error
       Description: Fallback lambda to handle messages which failed processing
       MemorySize: 192
       Timeout: 300
@@ -405,7 +404,7 @@ Resources:
       Name: !Sub /sdlf/pipeline/rLambdaRoutingStep/${pDeploymentInstance}
       Type: String
       Value: !GetAtt rLambdaRoutingStep.Arn
-      Description: !Sub "ARN of the ${pDataset} ${pDeploymentInstance} Routing Lambda" # TODO
+      Description: !Sub "ARN of the ${pDeploymentInstance} Routing Lambda" # TODO
 
   rLambdaRoutingStepLogGroup:
     Type: AWS::Logs::LogGroup
@@ -452,7 +451,7 @@ Resources:
           - id: W11
             reason: The actions with "*" are all ones that do not have resource limitations associated with them
     Properties:
-      Path: !Sub /sdlf-${pDataset}/
+      Path: !Sub /sdlf-${pDeploymentInstance}/
       # PermissionsBoundary: !Sub "{{resolve:ssm:/SDLF/IAM/${pDataset}/TeamPermissionsBoundary}}"
       AssumeRolePolicyDocument:
         Version: "2012-10-17"
@@ -466,26 +465,26 @@ Resources:
               StringEquals:
                 "aws:SourceAccount": !Sub ${AWS::AccountId}
       Policies:
-        - PolicyName: !Sub sdlf-${pDataset}-${pDeploymentInstance}-sm
+        - PolicyName: !Sub sdlf-${pDeploymentInstance}-sm
           PolicyDocument:
             Version: "2012-10-17"
             Statement:
               - Effect: Allow
                 Action:
                   - lambda:InvokeFunction
-                Resource: !Sub arn:${AWS::Partition}:lambda:${AWS::Region}:${AWS::AccountId}:function:sdlf-${pDataset}-${pDeploymentInstance}-*
+                Resource: !Sub arn:${AWS::Partition}:lambda:${AWS::Region}:${AWS::AccountId}:function:sdlf-${pDeploymentInstance}-*
               - Effect: Allow
                 Action:
                   - glue:StartJobRun
                   - glue:GetJobRun
                   - glue:GetJobRuns
                   - glue:BatchStopJobRun
-                Resource: !Sub arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:job/sdlf-${pDataset}-*
+                Resource: !Sub arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:job/sdlf-${pDeploymentInstance}-*
               - Effect: Allow
                 Action:
                   - glue:StartCrawler
                   - glue:GetCrawler
-                Resource: !Sub arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:crawler/sdlf-${pDataset}-*
+                Resource: !Sub arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:crawler/sdlf-${pDatasetDeploymentInstance}-*
               - Effect: Allow
                 Action:
                   - xray:PutTraceSegments # W11 exception
@@ -514,7 +513,7 @@ Resources:
   rStateMachine:
     Type: AWS::Serverless::StateMachine
     Properties:
-      Name: !Sub sdlf-${pDataset}-${pDeploymentInstance}-sm
+      Name: !Sub sdlf-${pDeploymentInstance}-sm
       DefinitionUri: ./state-machine/stage-glue.asl.json
       DefinitionSubstitutions:
         lPostMetadata: !GetAtt rLambdaPostMetadataStep.Arn
@@ -537,7 +536,7 @@ Resources:
       Name: !Sub /sdlf/pipeline/rStateMachine/${pDeploymentInstance}
       Type: String
       Value: !Ref rStateMachine
-      Description: !Sub "ARN of the ${pDataset} ${pDeploymentInstance} State Machine" # TODO
+      Description: !Sub "ARN of the ${pDeploymentInstance} State Machine" # TODO
 
 Outputs:
   oPipelineReference:

sdlf-stage-glue/src/lambda/error/src/lambda_function.py

@@ -6,9 +6,6 @@
 from datalake_library.sdlf import SQSConfiguration
 
 logger = init_logger(__name__)
-dataset = os.environ["DATASET"]
-pipeline = os.environ["PIPELINE"]
-pipeline_stage = os.environ["PIPELINE_STAGE"]
 deployment_instance = os.environ["DEPLOYMENT_INSTANCE"]
 
 

sdlf-stage-glue/src/lambda/postupdate-metadata/src/lambda_function.py

@@ -4,9 +4,6 @@
 from datalake_library.sdlf import PipelineExecutionHistoryAPI
 
 logger = init_logger(__name__)
-dataset = os.environ["DATASET"]
-pipeline = os.environ["PIPELINE"]
-pipeline_stage = os.environ["PIPELINE_STAGE"]
 deployment_instance = os.environ["DEPLOYMENT_INSTANCE"]
 peh_table_instance = os.environ["DATASET_DEPLOYMENT_INSTANCE"]
 manifests_table_instance = os.environ["DATASET_DEPLOYMENT_INSTANCE"]
@@ -42,7 +39,7 @@ def lambda_handler(event, context):
 
         if not partial_failure:
             pipeline_execution.update_pipeline_execution(
-                status=f"{pipeline}-{pipeline_stage} {component} Processing", component=component
+                status=f"{deployment_instance} {component} Processing", component=component
             )
             pipeline_execution.end_pipeline_execution_success()
         else:
@@ -51,6 +48,6 @@ def lambda_handler(event, context):
     except Exception as e:
         logger.error("Fatal error", exc_info=True)
         pipeline_execution.end_pipeline_execution_failed(
-            component=component, issue_comment=f"{pipeline}-{pipeline_stage} {component} Error: {repr(e)}"
+            component=component, issue_comment=f"{deployment_instance} {component} Error: {repr(e)}"
         )
         raise e

sdlf-stage-glue/src/lambda/redrive/src/lambda_function.py

@@ -12,12 +12,8 @@
 )
 
 logger = init_logger(__name__)
-dataset = os.environ["DATASET"]
-pipeline = os.environ["PIPELINE"]
-pipeline_stage = os.environ["PIPELINE_STAGE"]
 deployment_instance = os.environ["DEPLOYMENT_INSTANCE"]
-peh_table_instance = os.environ["DATASET_DEPLOYMENT_INSTANCE"]
-manifests_table_instance = os.environ["DATASET_DEPLOYMENT_INSTANCE"]
+dataset_deployment_instance = peh_table_instance = manifests_table_instance = os.environ["DATASET_DEPLOYMENT_INSTANCE"]
 
 
 def serializer(obj):
@@ -31,8 +27,8 @@ def serializer(obj):
 
 def pipeline_start(pipeline_execution, event):
     peh_id = pipeline_execution.start_pipeline_execution(
-        pipeline_name=f"{pipeline}-{pipeline_stage}",
-        dataset_name=f"{dataset}",
+        pipeline_name=deployment_instance,
+        dataset_name=dataset_deployment_instance,
         comment=event,  # TODO test maximum size
     )
     logger.info(f"peh_id: {peh_id}")
@@ -53,14 +49,14 @@ def get_source_records(event):
         logger.info("Stage trigger: event-schedule")
         min_items_to_process = 1
         max_items_to_process = 100
-        logger.info(f"Pipeline is {pipeline}, stage is {pipeline_stage}")
+        logger.info(f"Pipeline is {deployment_instance}")
         logger.info(
             f"Pipeline stage configuration: min_items_to_process {min_items_to_process}, max_items_to_process {max_items_to_process}"
         )
 
         sqs_config = SQSConfiguration(instance=deployment_instance)
         queue_interface = SQSInterface(sqs_config.stage_queue)
-        logger.info(f"Querying {dataset} {pipeline}-{pipeline_stage} objects waiting for processing")
+        logger.info(f"Querying {deployment_instance} objects waiting for processing")
         messages = queue_interface.receive_min_max_messages(min_items_to_process, max_items_to_process)
         logger.info(f"{len(messages)} Objects ready for processing")
 
@@ -115,7 +111,7 @@ def lambda_handler(event, context):
                 state_config.stage_state_machine, json.dumps(records, default=serializer)
             )
             pipeline_execution.update_pipeline_execution(
-                status=f"{pipeline}-{pipeline_stage} Transform Processing", component="Transform"
+                status=f"{deployment_instance} Transform Processing", component="Transform"
             )
         else:
             logger.info("Nothing to process, exiting pipeline")
@@ -126,6 +122,6 @@ def lambda_handler(event, context):
         component = context.function_name.split("-")[-2].title()
         pipeline_execution.end_pipeline_execution_failed(
             component=component,
-            issue_comment=f"{pipeline}-{pipeline_stage} {component} Error: {repr(e)}",
+            issue_comment=f"{deployment_instance} {component} Error: {repr(e)}",
         )
         raise e