Decomplect the slack logging by putting it directly into the ingestio…

…n job. #1564

.github/workflows/ingestion-functions-deploy.yml

@@ -41,6 +41,7 @@ jobs:
         ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
         ECR_REPOSITORY: gdh-ingestor
         IMAGE_TAG: ${{ github.sha }}
+        SLACK_LOGS_WEBHOOK: ${{ secrets.SLACK_WEBHOOK_INGESTION_LOGS }}
       run: |
         docker build --build-arg NOTIFY_WEBHOOK_URL -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG -t $ECR_REGISTRY/$ECR_REPOSITORY .
         docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
@@ -54,6 +55,7 @@ jobs:
         ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
         ECR_REPOSITORY: gdh-ingestor
         IMAGE_TAG: ${{ github.sha }}
+        SLACK_LOGS_WEBHOOK: ${{ secrets.SLACK_WEBHOOK_INGESTION_LOGS_TEST }}
       run: |
         docker build --build-arg NOTIFY_WEBHOOK_URL -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG -t $ECR_REGISTRY/$ECR_REPOSITORY:stable .
         docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG

ingestion/functions/common/ingestion_logging.py

@@ -1,8 +1,26 @@
 import logging
+import os
+import requests
 import sys
 
 handlers = set()
 
+class SlackHandler(logging.Handler):
+    def __init__(self, webhook_url, level=logging.NOTSET):
+        super().__init__(level)
+        self.slack_url = webhook_url
+
+    def emit(self, record):
+        message_header = {'Content-Type': 'application/json'}
+        message = {'text': f"[{record.levelname}] {record.message}"}
+        response = requests.post(url=self.slack_url, data=json.dumps(message), headers=message_header)
+        if response.status_code == 429 and response['error'] == 'rate_limited':
+            sleep(response['retry_after'])
+        elif response.status_code != 200:
+            raise ValueError(
+                f"Request to slack returned an error {response.status_code}, the response is:\n{response.text}"
+            )
+
 def getLogger(module_name):
     logger = logging.getLogger(module_name)
     logger.setLevel(logging.DEBUG)
@@ -11,6 +29,10 @@ def getLogger(module_name):
         handler.setLevel(logging.DEBUG)
         logger.addHandler(handler)
         handlers.add(handler)
+        if slack_webhook := os.getenv('SLACK_LOGS_WEBHOOK'):
+            slackHandler = SlackHandler(slack_webhook, logging.WARNING)
+            logger.addHandler(slackHandler)
+            handlers.add(slackHandler)
     return logger
 
 def flushAll():

ingestion/monitoring/README.md

@@ -1,20 +1,3 @@
-# Error monitoring
-
-The `errorLogsToSlack.py` script reads log messages from a given Cloudwatch stream
-and posts any errors to Slack. It has three inputs, all passed via the environment:
-
- - `SLACK_WEBHOOK` is the webhook URL to post messages to Slack.
- - `INGESTION_LOG_GROUP` is the Cloudwatch log group name.
- - `INGESTION_LOG_STREAM` is the Cloudwatch log stream name.
-
-Typically, all would be set up EventBridge in AWS when it's run in Batch.
-
-## To set up for a new instance
-
-1. see https://api.slack.com/messaging/webhooks for details on creating a Slack app and enabling web hooks.
-2. change the Slack user IDs in the script to ones that represent users in your workspace (who should get notified on ingestion errors).
-3. deploy to Batch
-
 # Data monitoring
 
 Data monitoring scripts, currently there's a script to alert daily about