#55: Create an endpoint for grafana alerts

.github/workflows/master.yml

@@ -58,10 +58,11 @@ jobs:
         env:
           WS_URL_ALL_USERS: ${{ secrets.WS_URL_ALL_USERS }}
           WS_URL_POST_COMMENT: ${{ secrets.WS_URL_POST_COMMENT }}
-          WS_ADMIN_EMAIL: ${{ secrets.WS_EMAIL }}
+          WS_URL_POST_TASK: ${{ secrets.WS_URL_POST_TASK }}
+          WS_ADMIN_EMAIL: ${{ secrets.WS_ADMIN_EMAIL }}
           WS_ADMIN_USER_ID: ${{ secrets.WS_ADMIN_USER_ID }}
-          WS_PRJ_223728_HASH: ${{ secrets.WS_PRJ_223728_HASH }}
-          WS_PRJ_1010_HASH: ${{ secrets.WS_PRJ_1010_HASH }}
+          WS_PRJ_223728_POST_TASK_HASH: ${{ secrets.WS_PRJ_223728_POST_TASK_HASH }}
+          WS_PRJ_223728_POST_COMMENT_HASH: ${{ secrets.WS_PRJ_223728_POST_COMMENT_HASH }}
       - name: "Upload coverage to Codecov"
         uses: codecov/[email protected]
         # with:
@@ -85,10 +86,6 @@ jobs:
       - name: Run tests
         run: make test
         env:
-          WS_URL_ALL_USERS: ${{ secrets.WS_URL_ALL_USERS }}
-          WS_URL_POST_COMMENT: ${{ secrets.WS_URL_POST_COMMENT }}
-          WS_ADMIN_EMAIL: ${{ secrets.WS_EMAIL }}
-          WS_ADMIN_USER_ID: ${{ secrets.WS_ADMIN_USER_ID }}
           WS_INT_TESTS_DISABLED: true
 
   tests_win:
@@ -111,10 +108,6 @@ jobs:
       - name: run tests
         run: pytest -s -vvvv -l --tb=long tests
         env:
-          WS_URL_ALL_USERS: ${{ secrets.WS_URL_ALL_USERS }}
-          WS_URL_POST_COMMENT: ${{ secrets.WS_URL_POST_COMMENT }}
-          WS_ADMIN_EMAIL: ${{ secrets.WS_EMAIL }}
-          WS_ADMIN_USER_ID: ${{ secrets.WS_ADMIN_USER_ID }}
           WS_INT_TESTS_DISABLED: true
 
   docker:

.gitignore

@@ -73,6 +73,7 @@ docs/_build/
 
 # PyBuilder
 target/
+dist/
 
 # Jupyter Notebook
 .ipynb_checkpoints
@@ -130,3 +131,5 @@ dmypy.json
 
 # templates
 .github/templates/*
+
+.local

docker-compose.yml

@@ -6,13 +6,15 @@ services:
     environment:
       WS_URL_ALL_USERS: "https://xxx.worksection.com/xxxx"
       WS_URL_POST_COMMENT: "https://xxx.worksection.com/xxxx"
+      WS_URL_POST_TASK: "https://xxx.worksection.com/xxxx"
       WS_ADMIN_EMAIL: "[email protected]"
       WS_ADMIN_USER_ID: "370080"
-      WS_PRJ_223728_HASH: "xxx"
+      WS_PRJ_223728_POST_TASK_HASH: "23e1sdfj2323"
+      WS_PRJ_223728_POST_COMMENT_HASH: "2312jsafajsdf"
     build:
       dockerfile: Containerfile
       context: .
-    entrypoint: g2w --log=DEBUG
+    entrypoint: g2w
     ports:
       - "8080:8080"
     restart: always

g2w/__init__.py

@@ -1,5 +1,6 @@
-from .push import Push
+from .gitlab import Push
 from .ws import Ws
 from .api import LoggableRoute
+from .grafana import Alert
 
-__all__ = ["Push", "Ws", "LoggableRoute"]
+__all__ = ["Push", "Ws", "LoggableRoute", "Alert"]

g2w/__main__.py

@@ -5,30 +5,11 @@
 import argparse  # pragma: no cover
 
 import uvicorn  # pragma: no cover
-from fastapi import FastAPI
+from fastapi import Request, FastAPI
 from fastapi.routing import APIRouter
-
-from g2w import Push, Ws, LoggableRoute
-
+from g2w import Push, Ws, LoggableRoute, Alert
 
 # @todo #/DEV Add support of command line parser for program arguments
-
-
-def main() -> None:  # pragma: no cover
-    cmd = argparse.ArgumentParser()
-    cmd.add_argument(
-        "-p",
-        "--port",
-        type=int,
-        help="The port to listen REST API endpoints",
-        default=8080,
-        required=False,
-    )
-    uvicorn.run(app, host="0.0.0.0", port=cmd.parse_args().port)
-    # @todo #/DEV Add prometheus client library for app monitoring
-    #  https://github.com/prometheus/client_python
-
-
 ws = Ws()
 app = FastAPI()
 router = APIRouter(route_class=LoggableRoute)
@@ -39,7 +20,6 @@ def main() -> None:  # pragma: no cover
 
 @router.post("/gitlab/push/{project_id}")
 def push(event: Push, project_id: int) -> dict:
-
     author = ws.find_user(event.user_email)
     msg = event.comment(author)
     comments = []
@@ -49,7 +29,35 @@ def push(event: Push, project_id: int) -> dict:
     return {"comments": comments}
 
 
+@router.post("/grafana/alert/{project_id}")
+async def alert(event: Request, project_id: int) -> dict:
+    # @todo #/DEV Replace plain json in ticket summary by more sophisticated
+    #  object with proper formatting
+    alert = Alert()
+    return {
+        "created": ws.add_task(
+            project_id, alert.subject(), alert.desc(await event.json())
+        )
+    }
+
+
 app.include_router(router)
 
+
+def main() -> None:  # pragma: no cover
+    cmd = argparse.ArgumentParser()
+    cmd.add_argument(
+        "-p",
+        "--port",
+        type=int,
+        help="The port to listen REST API endpoints",
+        default=8080,
+        required=False,
+    )
+    uvicorn.run(app, host="0.0.0.0", port=cmd.parse_args().port)
+    # @todo #/DEV Add prometheus client library for app monitoring
+    #  https://github.com/prometheus/client_python
+
+
 if __name__ == "__main__":  # pragma: no cover
     main()

g2w/grafana.py

@@ -0,0 +1,31 @@
+import urllib.parse
+
+import airspeed
+from pydantic import BaseModel
+
+
+class Alert(BaseModel):
+    """
+    Alert event details from Grafana.
+    """
+
+    def desc(self, json) -> str:
+        """
+        Allows to transform Gitlab push event about multiple commits into HTML
+        comment for worksection.
+        """
+        return self.encode(
+            airspeed.Template(
+                """<pre><code class="code_init hljs json">$json
+                                 </code></pre>"""
+            ).merge(locals())
+        )
+
+    def subject(self):
+        return self.encode("Monitoring alert")
+
+    # @todo #/DEV Move encode function to generic place as it will be used for
+    #  all future Worksection requests:
+    #  https://stackoverflow.com/a/30045261/6916890
+    def encode(self, text: str) -> str:
+        return urllib.parse.quote_plus(text)

g2w/ws.py

@@ -5,89 +5,94 @@
 import requests  # pragma: no cover
 
 
-def get_hash(prj) -> str:
-    """
-    Fetch project hash from environment variable.
-    The naming format is 'WS_PRJ_0000_HASH', where '0000' is project id:
-      WS_PRJ_0001_HASH: 12fjasdfsdfhk34hsdf
-      WS_PRJ_0002_HASH: asf324i324jdfi23hfd
-      ...
-    """
-    val = os.getenv(f"WS_PRJ_{prj}_HASH")
+def env(key) -> str:
+    val = os.getenv(key)
     if val is None:
-        logging.error("g2w-003: No hash found for project id '%'", prj)
-        raise ValueError("g2w-003: No hash found for project id '%'", prj)
+        raise ValueError(f"g2w-003: Environment variable '{key}' not found")
     return val
 
 
+def ws_admin_email() -> str:
+    return env("WS_ADMIN_EMAIL")
+
+
+def ws_admin_userid() -> str:
+    return env("WS_ADMIN_USER_ID")
+
+
 def post(req) -> dict:
     """
     Send POST request to Worksection API.
     """
     resp = requests.post(req).json()
     # @todo #58/DEV Ensure that logging is enabled for this method as well.
     logging.debug("WS req: '%s', resp: '%s'", req, resp)
-    return resp
+    if resp["status"] == "ok":
+        return resp["data"]
+    else:
+        return resp
 
 
 class Ws:
     """
     Worksection client that allows manipulation with
     """
-    def __init__(
-        self,
-        email=os.getenv("WS_ADMIN_EMAIL"),
-        system_user_id=os.getenv("WS_ADMIN_USER_ID"),
-        all_users=os.getenv("WS_URL_ALL_USERS"),
-        post_comment=os.getenv("WS_URL_POST_COMMENT"),
-    ):
-        self.system_email = email
-        self.system_user_id = system_user_id
-        self.url_all_users = all_users
-        self.url_post_comment = post_comment
 
     users: List[dict] = []
 
+    def find_user(self, email: str) -> dict:
+        """
+        Find user details in Worksection by email.
+        Return user or system account (if not found).
+        """
+        user = next((u for u in self.all_users() if u["email"] == email), None)
+        if user is not None:
+            return user
+        else:
+            return next(
+                (u for u in self.all_users() if u["id"] == ws_admin_userid())
+            )
+
     def all_users(self) -> List[dict]:
         """
         Fetch all users from worksection space.
         """
         if not self.users:
             # @todo #/DEV use memorize feature/approach instead of own caching.
-            self.users.extend(requests.get(self.url_all_users).json()["data"])
+            self.users.extend(
+                requests.get(env("WS_URL_ALL_USERS")).json()["data"]
+            )
         return self.users
 
     def add_comment(self, prj: int, task: int, body: str) -> dict:
         """
         Add a comment to a particular worksection task id.
         """
-        resp = post(self.post_comment_url(prj, task, body))
-        if resp["status"] == "ok":
-            return resp["data"]
-        else:
-            return resp
+        return post(self.post_comment_url(prj, task, body))
 
     def post_comment_url(self, prj, task, body) -> str:
         """
         Construct URL for posting comments.
         """
-        return self.url_post_comment.format(
-            prj, task, self.system_email, body, get_hash(prj)
+        return env("WS_URL_POST_COMMENT").format(
+            prj,
+            task,
+            ws_admin_email(),
+            body,
+            env(f"WS_PRJ_{prj}_POST_COMMENT_HASH"),
         )
 
-    def find_user(self, email: str) -> dict:
+    def add_task(self, prj, subj, body) -> dict:
         """
-        Find user details in Worksection by email.
-        Return user or system account (if not found).
+        Add a ticket to a particular worksection project.
         """
-        user = next((u for u in self.all_users() if u["email"] == email), None)
-        if user is not None:
-            return user
-        if self.system_user_id is None:
-            raise ValueError(
-                "g2w-002: No user found with email {0}".format(email)
-            )
-        else:
-            return next(
-                (u for u in self.all_users() if u["id"] == self.system_user_id)
-            )
+        return post(self.post_task_url(prj, subj, body))
+
+    def post_task_url(self, prj, subj, body) -> str:
+        return env("WS_URL_POST_TASK").format(
+            prj,
+            subj,
+            ws_admin_email(),
+            body,
+            env(f"WS_PRJ_{prj}_POST_TASK_HASH"),
+        )

readme.md

@@ -56,9 +56,11 @@ Please note, that `GitLab` commit messages must have the following format: `#WS-
           # Mandatory environment variables (docker, podman, etc.)
           WS_URL_ALL_USERS: "https://xxx.worksection.com/xxxx"    # https://worksection.com/faq/api-user.html#q1572
           WS_URL_POST_COMMENT: "https://xxx.worksection.com/xxxx" # https://worksection.com/faq/api-comments.html#q1575
+          WS_URL_POST_TASK: "https://xxx.worksection.com/xxxx"    # https://worksection.com/faq/api-task.html#q1577
           WS_ADMIN_EMAIL: "[email protected]"         # plain worksection user email
           WS_ADMIN_USER_ID: "370080"                              # plain worksection user id
-          WS_PRJ_223728_HASH: "xxx"                               # HASH generated for a particular Worksection project
+          WS_PRJ_223728_POST_TASK_HASH: "23e1sdfj2323"            # HASH generated for new task action for a particular project
+          WS_PRJ_223728_POST_COMMENT_HASH: "2312jsafajsdf"        # HASH generated for new comment action for a particular project
         build:
           dockerfile: Containerfile
           context: .

tests/test_alert.py

@@ -0,0 +1,9 @@
+from g2w import Alert
+
+
+# @todo #/DEV Create fake grafana alert event object based on
+#  https://grafana.com/docs/grafana/latest/alerting/unified-alerting/contact-points/#webhook   # noqa: E501
+
+
+def test_comment():
+    assert Alert().desc('{"user":"Tom"}').find("user%22%3A%22Tom") > 0