Add slack webhook integration

librarian_server/logger.py

@@ -5,6 +5,7 @@
 import inspect
 import logging as log
 
+import requests
 from sqlalchemy.orm import Session
 
 from hera_librarian.errors import ErrorCategory, ErrorSeverity
@@ -29,6 +30,34 @@
 log.debug("Logging set up.")
 
 
+def post_error_to_slack(error: "Error") -> None:
+    if not server_settings.slack_webhook_enable:
+        return
+
+    if error.severity not in server_settings.slack_webhook_post_error_severity:
+        return
+
+    if error.category not in server_settings.slack_webhook_post_error_category:
+        return
+
+    requests.post(
+        server_settings.slack_webhook_url,
+        json={
+            "username": server_settings.displayed_site_name,
+            "icon_emoji": ":ledger:",
+            "text": (
+                f"*New Librarian Error at {server_settings.name}*\n"
+                f"> _Error Severity_: {error.severity.name}\n"
+                f"> _Error Category_: {error.category.name}\n"
+                f"> _Error Message_: {error.message}\n"
+                f"> _Error ID_: {error.id}\n"
+                f"> _Error Raised Time_: {error.raised_time}\n"
+                f"`{error.caller}`"
+            ),
+        },
+    )
+
+
 def log_to_database(
     severity: ErrorSeverity, category: ErrorCategory, message: str, session: Session
 ) -> None:
@@ -72,3 +101,5 @@ def log_to_database(
 
     session.add(error)
     session.commit()
+
+    post_error_to_slack(error)

librarian_server/settings.py

@@ -11,6 +11,8 @@
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from sqlalchemy import URL
 
+from hera_librarian.errors import ErrorCategory, ErrorSeverity
+
 from .stores import StoreNames
 
 if TYPE_CHECKING:
@@ -80,6 +82,13 @@ class ServerSettings(BaseSettings):
 
     max_search_results: int = 64
 
+    # Slack integration; by default disable this. You will need a slack
+    # webhook url, and by default we raise all log_to_database alerts to slack too.
+    slack_webhook_enable: bool = False
+    slack_webhook_url: Optional[str] = None
+    slack_webhook_post_error_severity: list[ErrorSeverity] = list(ErrorSeverity)
+    slack_webhook_post_error_category: list[ErrorCategory] = list(ErrorCategory)
+
     model_config = SettingsConfigDict(env_prefix="librarian_server_")
 
     @property

tests/server_unit_test/test_post_to_slack.py

@@ -0,0 +1,50 @@
+"""
+Tests the 'post to slack' functionality.
+"""
+
+import datetime
+import inspect
+
+
+def test_post_to_slack():
+    # This is a manual-only test. You will need to have a valid install.
+    # If you want to run this test, you will need to set the following
+    # env vars: LIBRARIAN_SERVER_SLACK_WEBHOOK_ENABLE,
+    # LIBRARIAN_SERVER_SLACK_WEBHOOK_URL
+    try:
+        from librarian_server.settings import server_settings
+    except:
+        return
+
+    if not server_settings.slack_webhook_enable:
+        return
+
+    from librarian_server.logger import (
+        ErrorCategory,
+        ErrorSeverity,
+        post_error_to_slack,
+    )
+
+    class MockError:
+        def __init__(self, severity, category, message, id):
+            self.message = message
+            self.category = category
+            self.severity = severity
+            self.id = id
+            self.raised_time = datetime.datetime.utcnow()
+            self.caller = (
+                inspect.stack()[1].filename
+                + ":"
+                + inspect.stack()[1].function
+                + ":"
+                + str(inspect.stack()[1].lineno)
+            )
+
+    post_error_to_slack(
+        MockError(
+            ErrorSeverity.CRITICAL,
+            ErrorCategory.DATA_AVAILABILITY,
+            "This is a test message, please ignore.",
+            12345,
+        )
+    )