initial commit

.env.template

@@ -0,0 +1,13 @@
+# slack settings
+SLACK_BOT_TOKEN=
+SLACK_CHANNEL=
+
+# postgres to monitor
+DB_HOST=
+DB_PORT=
+DB_USER=
+DB_PASSWORD=
+DB_NAME=
+
+# interval seconds
+INTERVAL_SECONDS=

README.md

@@ -1 +1,4 @@
-# pgslot-notify-bot
+# pgslot-notify-bot
+
+1. Copy `.env.template` to `.env` and fill in the values.
+2.

main.py

@@ -0,0 +1,56 @@
+import click
+import psycopg2
+import os
+import time
+from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError
+
+# PostgreSQL query to get replication slot size
+REPLICATION_SLOT_QUERY = "SELECT slot_name, pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS replication_lag_bytes FROM pg_replication_slots;"
+
+# Slack client initialization
+slack_token = os.environ["SLACK_BOT_TOKEN"]
+client = WebClient(token=slack_token)
+
+def query_replication_slot_size(conn):
+    with conn.cursor() as cur:
+        cur.execute(REPLICATION_SLOT_QUERY)
+        slots = cur.fetchall()
+        return slots
+
+def post_message_to_slack(channel, message):
+    try:
+        response = client.chat_postMessage(channel=channel, text=message)
+    except SlackApiError as e:
+        print(f"Error posting to Slack: {e.response['error']}")
+
+@click.command()
+@click.option('--db-host', default='localhost', help='Database host.')
+@click.option('--db-port', default=5432, help='Database port.')
+@click.option('--db-user', default='user', help='Database user.')
+@click.option('--db-password', default='password', help='Database password.')
+@click.option('--db-name', default='dbname', help='Database name.')
+@click.option('--slack-channel', default='#general', help='Slack channel to post messages to.')
+@click.option('--interval-seconds', default=60, help='Interval in seconds between each check.')
+def main(db_host, db_port, db_user, db_password, db_name, slack_channel, interval_seconds):
+    conn = psycopg2.connect(
+        host=db_host,
+        port=db_port,
+        user=db_user,
+        password=db_password,
+        dbname=db_name
+    )
+
+    while True:
+        slots = query_replication_slot_size(conn)
+        for slot in slots:
+            slot_name, size = slot
+            size_mb = size / 1024 / 1024
+            if size_mb > 100:
+                post_message_to_slack(slack_channel, f"Replication slot '{slot_name}' size is over 100MB: {size_mb} MB")
+
+        time.sleep(interval_seconds)
+        conn.close()
+
+if __name__ == '__main__':
+    main()

requirements.txt

@@ -0,0 +1,3 @@
+click==8.1.7
+psycopg2-binary==2.9.9
+slack-sdk==3.26.1

run.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+set -euo pipefail
+
+set -a
+source .env
+set +a
+
+python ./main.py \
+    --db-host="${DB_HOST}" \
+    --db-port="${DB_PORT}" \
+    --db-user="${DB_USER}" \
+    --db-password="${DB_PASSWORD}" \
+    --db-name="${DB_NAME}" \
+    --slack-channel="${SLACK_CHANNEL}" \
+    --interval-seconds=${INTERVAL_SECONDS}