Add monitoring and metrics

config.example.toml

+# Entries in comments are optional
+
 SECRET_KEY = "your-secret-key"
 UPLOAD_FOLDER = "uploads"
 MAX_CONTENT_LENGTH = 10 # in GB
@@ -6,6 +8,9 @@ ENABLED_LOCALISATIONS = [ "de", "en" ]
 DEFAULT_LANGUAGE = "de"
 MAIL_DOMAIN = "@example.com"
 
+MAILSERVICE_INTERVAL = 300 # in seconds
+# MONITORING_MAIL = "john.smith@example.com"
+
 [ CONTACT ]
 ORG = "Fun Inc."
 NAME = "John Smith"
@@ -15,4 +20,9 @@ MAIL = "john.smith@example.com"
 FROM = "funinc@example.com"
 SERVER = "smtp.example.com"
 PORT = 25
-# LOCAL_HOSTNAME: Set local hostname when talking to SMTP Server
+# LOCAL_HOSTNAME: Set local hostname when talking to SMTP Server
\ No newline at end of file
+
+#[ METRICS ]
+#URL = "http://localhost:8080/telegraf"
+#USER = "basic_auth_user"
+#PASS = "basic_auth_password"
\ No newline at end of file

mailservice.py

@@ -5,72 +5,138 @@ import tomllib
 from email.message import EmailMessage
 from os import scandir
 from pathlib import Path
-import signal
-import time
 
-run = True
-
-
-def handler(signum, frame):
-    global run
-    run = False
-
-
-signal.signal(signal.SIGINT, handler)
-signal.signal(signal.SIGTERM, handler)
-
-
-with open("config.toml", "rb") as f:
-    config = tomllib.load(f)
-
-with open("localisations.toml", "rb") as f:
-    localisations = tomllib.load(f)
-
-
-while run:
-    # gather jobs
-    finished_jobs = []
-    with scandir(config["UPLOAD_FOLDER"]) as uploads:
-        for entry in uploads:
-            if entry.is_dir():
-                with scandir(entry.path) as job:
-                    for file in job:
-                        if file.is_file() and file.name == "done":
-                            finished_jobs.append(entry.path)
-                            break
-
-    # send emails
-    sent = 0
-    with smtplib.SMTP(
-        host=config["MAIL"]["SERVER"],
-        port=config["MAIL"]["PORT"],
-        local_hostname=config["MAIL"]["LOCAL_HOSTNAME"] if config["MAIL"]["LOCAL_HOSTNAME"] else None,
-    ) as s:
-        for job in finished_jobs:
-            sent += 1
-            with open(Path(job).joinpath("metadata.json")) as f:
-                metadata = json.load(f)
-
-            language = metadata["language"]
+import requests
+from requests.auth import HTTPBasicAuth
+
+
+def main(end):
+    with open("config.toml", "rb") as f:
+        config = tomllib.load(f)
+
+    with open("localisations.toml", "rb") as f:
+        localisations = tomllib.load(f)
+
+    while not end.is_set():
+        metrics = {}
+
+        # gather jobs
+        completed_jobs = []
+        error_jobs = []
+        with scandir(config["UPLOAD_FOLDER"]) as uploads:
+            for entry in uploads:
+                if entry.is_dir():
+                    with scandir(entry.path) as job:
+                        for file in job:
+                            if file.is_file():
+                                if file.name == "done":
+                                    completed_jobs.append(entry.path)
+                                    break
+                                elif file.name == "error":
+                                    error_jobs.append(entry.path)
+                                    break
+
+        metrics["total_finished_jobs"] = len(completed_jobs)
+        metrics["current_job_errors"] = len(error_jobs)
+
+        try:
+            local_hostname = config["MAIL"]["LOCAL_HOSTNAME"]
+        except KeyError:
+            local_hostname = None
+
+        s = smtplib.SMTP(
+            host=config["MAIL"]["SERVER"], port=config["MAIL"]["PORT"], local_hostname=local_hostname
+        )
+
+        sent = 0
+        with s:
+            if len(completed_jobs) > 0:
+                metrics["completed_job_languages"] = {}
+                for job in completed_jobs:
+                    sent += 1
+                    with open(Path(job).joinpath("metadata.json")) as f:
+                        metadata = json.load(f)
+
+                    try:
+                        metrics["completed_job_languages"][metadata["video_language"]] += 1
+                    except KeyError:
+                        metrics["completed_job_languages"][metadata["video_language"]] = 1
+
+                    language = metadata["language"]
+
+                    msg = EmailMessage()
+                    msg["Subject"] = localisations["mail"]["subject"][language]
+                    msg["From"] = config["MAIL"]["FROM"]
+                    msg["To"] = metadata["email"]
+
+                    msg.set_content(
+                        localisations["mail"]["content"][language].format(metadata["filename"])
+                    )
+
+                    # filename.language.vtt
+                    filename = (
+                        Path(metadata["filename"])
+                        .with_suffix(".{}.vtt".format(metadata["video_language"]))
+                        .name
+                    )
+
+                    with open(Path(job).joinpath("subtitles.vtt")) as f:
+                        msg.add_attachment(f.read(), filename=filename)
+
+                    s.send_message(msg)
+
+                    shutil.rmtree(job)
+
+            if len(error_jobs) > 0:
+                try:
+                    msg = EmailMessage()
+                    msg["Subject"] = "Subtitle Service Error Report"
+                    msg["From"] = config["MAIL"]["FROM"]
+                    msg["To"] = config["MONITORING"]["MAIL"]
+
+                    job_uuids = []
+                    for job in error_jobs:
+                        job_uuids.append(Path(job).name)
+
+                    msg.set_content(
+                        "The following jobs currently have errors:\n{}".format("\n - ".join(job_uuids))
+                    )
+
+                except KeyError:
+                    pass
+
+        try:
+            try:
+                auth = HTTPBasicAuth(config["METRICS"]["USER"], config["MONITORING"]["PASS"])
+            except KeyError:
+                auth = None
+
+            requests.post(config["METRICS"]["URL"], json=metrics, auth=auth)
+
+        except KeyError:
+            pass
+
+        print(
+            "[MAILSERVICE] Sent {} mails. Sleeping for {} seconds.".format(
+                sent, config["MAILSERVICE_INTERVAL"]
+            )
+        )
 
-            msg = EmailMessage()
-            msg["Subject"] = localisations["mail"]["subject"][language]
-            msg["From"] = config["MAIL"]["FROM"]
-            msg["To"] = metadata["email"]
+        end.wait(config["MAILSERVICE_INTERVAL"])
 
-            msg.set_content(localisations["mail"]["content"][language].format(metadata["filename"]))
 
-            # filename.language.vtt
-            filename = (
-                Path(metadata["filename"]).with_suffix(".{}.vtt".format(metadata["video_language"])).name
-            )
+if __name__ == "__main__":
+    import signal
+    from threading import Event
 
-            with open(Path(job).joinpath("subtitles.vtt")) as f:
-                msg.add_attachment(f.read(), filename=filename)
+    end = Event()
 
-            s.send_message(msg)
+    def handler(signum, frame):
+        global end
+        print(signum)
+        end.set()
 
-            shutil.rmtree(job)
+    signal.signal(signal.SIGINT, handler)
+    signal.signal(signal.SIGTERM, handler)
 
-    print("[MAILSERVICE] Sent {} mails. Sleeping for 5 minutes.".format(sent))
-    time.sleep(300)
+    main(end)

monitoring.md

+# Monitoring
+
+## Job errors
+The mail service can send out emails for jobs with errors.
+Currently, it will send out a summary of all jobs with errors every time it runs.
+
+## Metrics
+If configured, the mailservice will export metrics to a webserver by POSTing a summary of the executed jobs and errors as json.
+
+```json
+{
+  "completed_job_languages": {
+    "de": 3,
+    "en": 1
+  },
+  "total_finished_jobs": 4,
+  "current_job_errors": 0
+}
+```
\ No newline at end of file

requirements.txt

@@ -2,4 +2,5 @@ av~=10.0.0
 Flask~=2.3.2
 Flask-WTF~=1.1.1
 wtforms[email]~=3.0.1
-whitenoise~=6.5.0
+whitenoise~=6.5.0
\ No newline at end of file
+requests
\ No newline at end of file