Move EBSCO Adapter to ECS (#2675)

* initial shift to ecs for ebsco adapter

* add task definition

* remove lambda traces, and add alerting

* remove deployment hacks

* Apply auto-formatting rules

---------

Co-authored-by: Buildkite on behalf of Wellcome Collection <wellcomedigitalplatform@wellcome.ac.uk>

.buildkite/pipeline.deploy-adapters.yml

@@ -1,9 +1,8 @@
 steps:
   - label: deploy EBSCO adapter
     command: |
-      ./builds/deploy_lambda_zip.sh \
-        ebsco_adapter/ebsco_adapter \
-        ebsco-adapter-ftp
+      ENV_TAG=env.prod ./builds/update_ecr_image_tag.sh \
+        uk.ac.wellcome/ebsco_adapter 
 
     plugins:
       - wellcomecollection/aws-assume-role#v0.2.2:

.buildkite/pipeline.yml

@@ -151,11 +151,26 @@ steps:
     matrix:
       - "calm_adapter/calm_window_generator"
       - "calm_adapter/calm_deletion_check_initiator"
-      - "ebsco_adapter/ebsco_adapter"
       - "ebsco_adapter/ebsco_indexer"
       - "common/window_generator"
       - "tei_adapter/tei_updater"
 
+  - label: "{{ matrix }} (Publish/Image)"
+    branches: "main"
+    plugins:
+      - wellcomecollection/aws-assume-role#v0.2.2:
+          role: "arn:aws:iam::760097843905:role/platform-ci"
+      - ecr#v2.5.0:
+          login: true
+      - docker-compose#v4.16.0:
+          config: "{{ matrix }}/docker-compose.yml"
+          cli-version: 2
+          push:
+            - publish:760097843905.dkr.ecr.eu-west-1.amazonaws.com/uk.ac.wellcome/ebsco_adapter:ref.${BUILDKITE_COMMIT}
+            - publish:760097843905.dkr.ecr.eu-west-1.amazonaws.com/uk.ac.wellcome/ebsco_adapter:latest
+    matrix:
+      - "ebsco_adapter/ebsco_adapter"
+
   - wait
 
   - label: trigger adapter deployments

ebsco_adapter/ebsco_adapter/Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.10
+
+WORKDIR /app
+
+ADD src /app
+RUN pip install -r requirements.txt
+
+ENTRYPOINT ["python", "main.py"]

ebsco_adapter/ebsco_adapter/docker-compose.yml

@@ -4,3 +4,24 @@ services:
       context: .
       dockerfile: ../../builds/test.python.Dockerfile
     command: ["py.test"]
+  dev:
+    build:
+      context: .
+      dockerfile: ./Dockerfile
+    volumes:
+      - $HOME/.aws:/root/.aws:ro
+    environment:
+      - AWS_PROFILE=platform-developer
+      - OUTPUT_TOPIC_ARN=${OUTPUT_TOPIC_ARN}
+      - CUSTOMER_ID=${CUSTOMER_ID}
+      - S3_BUCKET=${S3_BUCKET}
+      - S3_PREFIX=${S3_PREFIX}
+      - FTP_SERVER=${FTP_SERVER}
+      - FTP_USERNAME=${FTP_USERNAME}
+      - FTP_PASSWORD=${FTP_PASSWORD}
+      - FTP_REMOTE_DIR=${FTP_REMOTE_DIR}
+    command: ["python", "main.py"]
+  publish:
+    build:
+      context: .
+      dockerfile: ./Dockerfile

ebsco_adapter/ebsco_adapter/run_local.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+
+# Usage: ./run_local.sh <args>
+
+AWS_PROFILE=platform-developer
+
+export AWS_PROFILE
+
+FTP_PASSWORD=$(aws ssm get-parameter --name /catalogue_pipeline/ebsco_adapter/ftp_password --with-decryption --query "Parameter.Value" --output text)
+FTP_SERVER=$(aws ssm get-parameter --name /catalogue_pipeline/ebsco_adapter/ftp_server --query "Parameter.Value" --output text)
+FTP_USERNAME=$(aws ssm get-parameter --name /catalogue_pipeline/ebsco_adapter/ftp_username --query "Parameter.Value" --output text)
+CUSTOMER_ID=$(aws ssm get-parameter --name /catalogue_pipeline/ebsco_adapter/customer_id --query "Parameter.Value" --output text)
+FTP_REMOTE_DIR=$(aws ssm get-parameter --name /catalogue_pipeline/ebsco_adapter/ftp_remote_dir --query "Parameter.Value" --output text)
+S3_BUCKET=$(aws ssm get-parameter --name /catalogue_pipeline/ebsco_adapter/bucket_name --query "Parameter.Value" --output text)
+OUTPUT_TOPIC_ARN=$(aws ssm get-parameter --name /catalogue_pipeline/ebsco_adapter/output_topic_arn --query "Parameter.Value" --output text)
+
+# Update the S3_PREFIX to be the environment (use dev for local testing)
+S3_PREFIX=prod
+
+export FTP_PASSWORD FTP_SERVER FTP_USERNAME CUSTOMER_ID FTP_REMOTE_DIR S3_BUCKET S3_PREFIX OUTPUT_TOPIC_ARN
+
+# Ensure the docker image is up to date
+docker-compose --log-level ERROR build dev
+
+docker-compose run dev "$@"

ebsco_adapter/ebsco_adapter/src/ebsco_ftp.py

@@ -8,11 +8,13 @@ def __init__(self, ftp_server, ftp_username, ftp_password, ftp_remote_dir):
         self.ftp_username = ftp_username
         self.ftp_password = ftp_password
         self.ftp_remote_dir = ftp_remote_dir
+        self.ftp_connection_open = False
 
     def __enter__(self):
         self.ftp = FTP(self.ftp_server)
         self.ftp.login(self.ftp_username, self.ftp_password)
         self.ftp.cwd(self.ftp_remote_dir)
+        self.ftp_connection_open = True
         return self
 
     def list_files(self, valid_suffixes):
@@ -32,5 +34,10 @@ def download_file(self, file, temp_dir):
 
         return os.path.join(temp_dir, file)
 
-    def __exit__(self, exc_type, exc_val, exc_tb):
+    def quit(self):
         self.ftp.quit()
+        self.ftp_connection_open = False
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.ftp_connection_open:
+            self.ftp.quit()

ebsco_adapter/ebsco_adapter/src/main.py

@@ -13,6 +13,7 @@
 from extract_marc import extract_marc_records
 from compare_uploads import compare_uploads, find_notified_and_completed_flag
 from update_notifier import update_notifier
+from metrics import ProcessMetrics
 
 ftp_server = os.environ.get("FTP_SERVER")
 ftp_username = os.environ.get("FTP_USERNAME")
@@ -30,6 +31,12 @@
 def run_process(temp_dir, ebsco_ftp, s3_store, sns_publisher, invoked_at):
     print("Running regular process ...")
     available_files = sync_and_list_files(temp_dir, ftp_s3_prefix, ebsco_ftp, s3_store)
+
+    # Holding the connection open for the next step
+    # is unnecessary, if we close here we avoid any
+    # potential timeout issues with the connection.
+    ebsco_ftp.quit()
+
     updates = compare_uploads(
         available_files, extract_marc_records, xml_s3_prefix, temp_dir, s3_store
     )
@@ -109,34 +116,6 @@ def _get_iso8601_invoked_at():
     return invoked_at
 
 
-def lambda_handler(event, context):
-    invoked_at = _get_iso8601_invoked_at()
-    if "invoked_at" in event:
-        invoked_at = event["invoked_at"]
-
-    print(f"Starting lambda_handler @ {invoked_at}, got event: {event}")
-
-    with tempfile.TemporaryDirectory() as temp_dir:
-        with EbscoFtp(
-            ftp_server, ftp_username, ftp_password, ftp_remote_dir
-        ) as ebsco_ftp:
-            s3_store = S3Store(s3_bucket)
-            sns_publisher = SnsPublisher(sns_topic_arn)
-
-            if event is not None and "reindex_type" in event:
-                return run_reindex(
-                    s3_store,
-                    sns_publisher,
-                    invoked_at,
-                    event["reindex_type"],
-                    event.get("reindex_ids"),
-                )
-            else:
-                return run_process(
-                    temp_dir, ebsco_ftp, s3_store, sns_publisher, invoked_at
-                )
-
-
 if __name__ == "__main__":
     event = None
     context = SimpleNamespace(invoked_function_arn=None)
@@ -154,16 +133,40 @@ def lambda_handler(event, context):
         type=str,
         help="Comma-separated list of IDs to reindex (for partial)",
     )
+    parser.add_argument(
+        "--scheduled-invoke",
+        action="store_true",
+        help="To run a regular process invocation, without reindexing.",
+    )
 
+    invoked_at = _get_iso8601_invoked_at()
     args = parser.parse_args()
+
+    process_type = None
+    reindex_ids = None
     if args.reindex_type:
-        reindex_ids = None
+        process_type = f"reindex-{args.reindex_type}"
         if args.reindex_ids:
             reindex_ids = args.reindex_ids.split(",")
             reindex_ids = [rid.strip() for rid in reindex_ids]
-
-    # This is the event that will be passed to the lambda handler.
-    # When invoking the function, use this structure to trigger reindexing.
-    event = {"reindex_type": args.reindex_type, "reindex_ids": reindex_ids}
-
-    lambda_handler(event, None)
+    elif args.scheduled_invoke:
+        process_type = "scheduled"
+
+    with ProcessMetrics(
+        process_type
+    ) as metrics, tempfile.TemporaryDirectory() as temp_dir, EbscoFtp(
+        ftp_server, ftp_username, ftp_password, ftp_remote_dir
+    ) as ebsco_ftp:
+        s3_store = S3Store(s3_bucket)
+        sns_publisher = SnsPublisher(sns_topic_arn)
+
+        if args.reindex_type:
+            run_reindex(
+                s3_store,
+                sns_publisher,
+                invoked_at,
+                args.reindex_type,
+                reindex_ids,
+            )
+        elif args.scheduled_invoke:
+            run_process(temp_dir, ebsco_ftp, s3_store, sns_publisher, invoked_at)

ebsco_adapter/ebsco_adapter/src/metrics.py

@@ -0,0 +1,36 @@
+import boto3
+import time
+
+
+class ProcessMetrics:
+    def __init__(self, process_name: str):
+        self.process_name = process_name
+
+    def __enter__(self):
+        self.start_time = time.time()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        end_time = time.time()
+        duration = end_time - self.start_time
+
+        if exc_type is not None:
+            self.put_metric("ProcessDurationFailure", duration)
+        else:
+            self.put_metric("ProcessDurationSuccess", duration)
+
+    def put_metric(self, metric_name: str, value: float):
+        print(f"Putting metric {metric_name} ({self.process_name}) with value {value}s")
+        boto3.client("cloudwatch").put_metric_data(
+            Namespace="ebsco_adapter",
+            MetricData=[
+                {
+                    "MetricName": metric_name,
+                    "Dimensions": [
+                        {"Name": "process_name", "Value": self.process_name}
+                    ],
+                    "Value": value,
+                    "Unit": "Seconds",
+                }
+            ],
+        )

ebsco_adapter/ebsco_adapter/src/test_fixtures.py

@@ -46,6 +46,9 @@ def download_file(self, file, temp_dir):
             f.write(self.files[file])
         return os.path.join(temp_dir, file)
 
+    def quit(self):
+        pass
+
     def __exit__(self, exc_type, exc_val, exc_tb):
         pass
 

ebsco_adapter/terraform/alarm_lambda.tf

@@ -0,0 +1,62 @@
+module "cloudwatch_alarm_to_slack_lambda" {
+  source = "git@github.com:wellcomecollection/terraform-aws-lambda?ref=v1.2.0"
+
+  name        = "cloudwatch-alarm-to-slack"
+  description = "Sends CloudWatch alarms to a Slack channel."
+  runtime     = "python3.10"
+
+  filename    = data.archive_file.cloudwatch_alarm_to_slack.output_path
+  handler     = "cloudwatch_alarm_to_slack.lambda_handler"
+  memory_size = 512
+  timeout     = 60 // 1 minute
+
+  source_code_hash = data.archive_file.cloudwatch_alarm_to_slack.output_base64sha256
+
+  error_alarm_topic_arn = data.terraform_remote_state.monitoring.outputs["platform_lambda_error_alerts_topic_arn"]
+
+  environment = {
+    variables = {
+      HOOK_URL      = aws_ssm_parameter.cloudwatch_alarm_to_slack_hook_url.value
+      SLACK_CHANNEL = aws_ssm_parameter.cloudwatch_alarm_to_slack_channel.value
+    }
+  }
+}
+
+resource "aws_ssm_parameter" "cloudwatch_alarm_to_slack_hook_url" {
+  name        = "/catalogue_pipeline/ebsco_adapter/cloudwatch_alarm_to_slack_hook_url"
+  description = "The URL of the Slack webhook to send messages to"
+  type        = "String"
+  value       = "placeholder"
+
+  lifecycle {
+    ignore_changes = [
+      value
+    ]
+  }
+}
+
+resource "aws_ssm_parameter" "cloudwatch_alarm_to_slack_channel" {
+  name        = "/catalogue_pipeline/ebsco_adapter/cloudwatch_alarm_to_slack_channel"
+  description = "The Slack channel to send messages to"
+  type        = "String"
+  value       = "wc-platform-alerts"
+}
+
+module "cloudwatch_alarm_to_slack_topic" {
+  source = "github.com/wellcomecollection/terraform-aws-sns-topic.git?ref=v1.0.0"
+  name   = "cloudwatch_alarm_to_slack"
+}
+
+resource "aws_sns_topic_subscription" "cloudwatch_alarm_to_slack_subscription" {
+  topic_arn = module.cloudwatch_alarm_to_slack_topic.arn
+  protocol  = "lambda"
+  endpoint  = module.cloudwatch_alarm_to_slack_lambda.lambda.arn
+}
+
+resource "aws_lambda_permission" "allow_execution_from_sns" {
+  statement_id  = "AllowExecutionFromSNS"
+  action        = "lambda:InvokeFunction"
+  function_name = module.cloudwatch_alarm_to_slack_lambda.lambda.function_name
+  principal     = "sns.amazonaws.com"
+  source_arn    = module.cloudwatch_alarm_to_slack_topic.arn
+}

ebsco_adapter/terraform/alarms.tf

@@ -0,0 +1,15 @@
+resource "aws_cloudwatch_metric_alarm" "ebsco_adapter_no_success_metric" {
+  alarm_name          = "ebsco-adapter-no-success-metric"
+  comparison_operator = "LessThanThreshold"
+  evaluation_periods  = 1
+  metric_name         = "ProcessDurationSuccess"
+  namespace           = "ebsco_adapter"
+  period              = 86400
+  statistic           = "Sum"
+  threshold           = 1
+  alarm_description   = "No success metrics have been sent in the last 24 hours"
+  alarm_actions       = [module.cloudwatch_alarm_to_slack_topic.arn]
+  dimensions = {
+    process_name = "scheduled"
+  }
+}

ebsco_adapter/terraform/cluster.tf

@@ -0,0 +1,3 @@
+resource "aws_ecs_cluster" "cluster" {
+  name = local.namespace
+}

ebsco_adapter/terraform/data.tf

@@ -6,3 +6,35 @@ data "archive_file" "empty_zip" {
     filename = "lambda.py"
   }
 }
+
+data "archive_file" "cloudwatch_alarm_to_slack" {
+  output_path = "data/cloudwatch_alarm_to_slack.zip"
+  type        = "zip"
+  source {
+    content  = file("${path.module}/data/cloudwatch_alarm_to_slack.py")
+    filename = "cloudwatch_alarm_to_slack.py"
+  }
+}
+
+data "terraform_remote_state" "accounts_catalogue" {
+  backend = "s3"
+
+  config = {
+    role_arn = "arn:aws:iam::760097843905:role/platform-read_only"
+
+    bucket = "wellcomecollection-platform-infra"
+    key    = "terraform/aws-account-infrastructure/catalogue.tfstate"
+    region = "eu-west-1"
+  }
+}
+
+data "terraform_remote_state" "shared_infra" {
+  backend = "s3"
+
+  config = {
+    role_arn = "arn:aws:iam::760097843905:role/platform-read_only"
+    bucket   = "wellcomecollection-platform-infra"
+    key      = "terraform/platform-infrastructure/shared.tfstate"
+    region   = "eu-west-1"
+  }
+}

ebsco_adapter/terraform/data/.gitignore

@@ -0,0 +1,2 @@
+# This directory contains generated files, ignore the zip files
+*.zip