Added Prometheus metrics support with opt-in CLI flag.

crabstero/__main__.py

@@ -100,9 +100,29 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
             " never respond."
         ),
     )
+    parser.add_argument(
+        "--listen-metrics",
+        default=os.environ.get("LISTEN_METRICS"),
+        help=(
+            "Enable Prometheus metrics endpoint on HOST:PORT"
+            " (e.g. 127.0.0.1:9090). Disabled by default."
+            " (default: LISTEN_METRICS environment variable)."
+        ),
+    )
 
     args = parser.parse_args(argv)
 
+    if args.listen_metrics is not None:
+        host, sep, port_str = args.listen_metrics.rpartition(":")
+        if not sep or not host:
+            parser.error(
+                "--listen-metrics must be in HOST:PORT format (e.g. 127.0.0.1:9090)"
+            )
+        try:
+            args.listen_metrics = (host, int(port_str))
+        except ValueError:
+            parser.error(f"--listen-metrics port must be an integer, got '{port_str}'")
+
     if args.token is None:
         parser.error(
             "a Discord bot token is required via --token,"
@@ -129,6 +149,7 @@ def main() -> None:
         database_path=args.database_path,
         ingestion_workers=args.ingestion_workers,
         ingest_only=args.ingest_only,
+        metrics_address=args.listen_metrics,
     )
 
     async def _run() -> None:

crabstero/bot.py

@@ -28,6 +28,13 @@ from discord.ext import commands
 from crabstero import __version__ as crabstero_version
 from crabstero.database import Database
 from crabstero.listeners import interaction, message, server_events
+from crabstero.metrics import (
+    DISCORD_EVENTS,
+    DISCORD_LATENCY,
+    GUILD_COUNT,
+    INGESTION_BACKLOG,
+    MetricsServer,
+)
 from crabstero.tasks.ingestion import ingest_channel
 
 logger = logging.getLogger(__name__)
@@ -46,6 +53,7 @@ class Crabstero(commands.Bot):
         database_path: str,
         ingestion_workers: int = 4,
         ingest_only: bool = False,
+        metrics_address: tuple[str, int] | None = None,
     ) -> None:
         """Configure intents, store configuration, and prepare ingestion queue state.
 
@@ -53,6 +61,7 @@ class Crabstero(commands.Bot):
         :param database_path: The file path to the SQLite database.
         :param ingestion_workers: The number of concurrent ingestion workers.
         :param ingest_only: When True, the bot only ingests data and never responds.
+        :param metrics_address: Optional (host, port) for the Prometheus metrics server.
         """
         intents = discord.Intents.default()
         intents.guilds = True
@@ -75,6 +84,12 @@ class Crabstero(commands.Bot):
         self._ingestion_workers: list[asyncio.Task[None]] = []
         self.db: Database
 
+        self._metrics_address = metrics_address
+
+        DISCORD_LATENCY.set_function(lambda: self.latency)
+        GUILD_COUNT.set_function(lambda: len(self.guilds))
+        INGESTION_BACKLOG.set_function(lambda: self._ingestion_queue.qsize())
+
         repo_url = "https://git.logal.dev/LogalDeveloper/Crabstero"
         self.http.user_agent = f"DiscordBot ({repo_url}, {crabstero_version})"
 
@@ -83,6 +98,10 @@ class Crabstero(commands.Bot):
         self.db = await Database.connect(self._database_path)
         self._start_ingestion_workers()
 
+        if self._metrics_address is not None:
+            self._metrics_server = MetricsServer(*self._metrics_address)
+            await self._metrics_server.start()
+
         if not self._ingest_only:
             await interaction.setup(self)
 
@@ -109,6 +128,14 @@ class Crabstero(commands.Bot):
                 logger.info("Slash command tree has changed, syncing with Discord.")
                 await self.tree.sync()
 
+    def dispatch(self, event: str, /, *args: object, **kwargs: object) -> None:
+        """Dispatch an event, incrementing the events counter.
+
+        :param event: The event name.
+        """
+        DISCORD_EVENTS.labels(event=event).inc()
+        super().dispatch(event, *args, **kwargs)
+
     async def on_ready(self) -> None:
         """Log that the bot has started successfully."""
         logger.info("Crabstero started!")
@@ -130,6 +157,8 @@ class Crabstero(commands.Bot):
             worker.cancel()
         await asyncio.gather(*self._ingestion_workers, return_exceptions=True)
         self._ingestion_workers.clear()
+        if hasattr(self, "_metrics_server"):
+            await self._metrics_server.stop()
         if hasattr(self, "db"):
             await self.db.close()
         await super().close()

crabstero/listeners/interaction.py

@@ -24,7 +24,7 @@ import discord
 from discord import app_commands
 from discord.ext import commands
 
-from crabstero import flags
+from crabstero import flags, metrics
 from crabstero.flags import EntityType, Flag
 
 if TYPE_CHECKING:
@@ -63,6 +63,7 @@ class InteractionCog(commands.Cog):
                 await flags.clear_flag(
                     self.bot.db, interaction.user, EntityType.USER, Flag.ALLOW_PINGS
                 )
+                metrics.PINGME_OPTED_OUT.inc()
                 await interaction.response.send_message(
                     "I will no longer ping you for messages which"
                     " mention you. If you decide to opt back in,"
@@ -73,6 +74,7 @@ class InteractionCog(commands.Cog):
                 await flags.set_flag(
                     self.bot.db, interaction.user, EntityType.USER, Flag.ALLOW_PINGS
                 )
+                metrics.PINGME_OPTED_IN.inc()
                 await interaction.response.send_message(
                     "I will now ping you for messages which mention"
                     " you. If you change your mind, run"
@@ -81,6 +83,7 @@ class InteractionCog(commands.Cog):
                 )
 
         except Exception:
+            metrics.PINGME_ERRORS.inc()
             logger.exception(
                 "An exception occurred while attempting to execute"
                 " slash command responder for command 'pingme'."

crabstero/listeners/message.py

@@ -23,6 +23,7 @@ from typing import TYPE_CHECKING
 import discord
 from discord.ext import commands
 
+from crabstero import metrics
 from crabstero.messages import ingest_message, reply_to_message
 
 if TYPE_CHECKING:
@@ -55,6 +56,8 @@ class MessageCog(commands.Cog):
         if message.author.bot or message.author == self.bot.user:
             return
 
+        metrics.MESSAGES_PROCESSED.inc()
+
         if self.bot._ingest_only:
             # In ingest-only mode, never reply — only ingest eligible messages.
             if message.type == discord.MessageType.default and not isinstance(

crabstero/markov.py

@@ -22,6 +22,8 @@ LIMIT 1 naturally preserves this weighting.
 import re
 from typing import TYPE_CHECKING
 
+from crabstero import metrics
+
 if TYPE_CHECKING:
     from crabstero.database import Database
 
@@ -106,46 +108,49 @@ async def generate(
     :param hard_limit: Character count to hard-cut the sentence at.
     :return: A new generated sentence.
     """
-    word = await db.get_random_start_word(channel_id)
+    with metrics.GENERATION_DURATION.time():
+        word = await db.get_random_start_word(channel_id)
 
-    # Seed the chain with a fallback sentence if the channel has no data yet.
-    if word is None:
-        await _ingest_sentence(db, channel_id, 0, "Hello world!")
-        word = "Hello"  # Known start word from the fallback sentence above.
-
-    parts: list[str] = []
-    parts.append(word)
-    current_length = len(word)
-
-    # The loop is skipped if the starting word already ends a sentence (e.g. "Yes.").
-    while not is_complete_sentence(word):
-        # Past the soft limit, prefer a sentence-ending word to wrap up.
-        if current_length >= soft_limit:
-            next_word = await db.get_random_completing_next_word(channel_id, word)
-            if next_word is None:
-                next_word = await db.get_random_next_word(channel_id, word)
-        else:
-            next_word = await db.get_random_next_word(channel_id, word)
-
-        if next_word is None:
-            break
-
-        word = next_word
+        # Seed the chain with a fallback sentence if the channel has no data yet.
+        if word is None:
+            await _ingest_sentence(db, channel_id, 0, "Hello world!")
+            word = "Hello"  # Known start word from the fallback sentence above.
 
+        parts: list[str] = []
         parts.append(word)
-        current_length += 1 + len(word)  # +1 for the joining space.
+        current_length = len(word)
 
-        if current_length >= hard_limit:
-            result = " ".join(parts)[:hard_limit]
-            # Strip the internal sentence-end marker if it ended up at the boundary.
-            if result and result[-1] == DEFAULT_SENTENCE_END:
-                return result[:-1]
-            return result
+        # The loop is skipped if the start word already ends a sentence (e.g. "Yes.").
+        while not is_complete_sentence(word):
+            # Past the soft limit, prefer a sentence-ending word to wrap up.
+            if current_length >= soft_limit:
+                next_word = await db.get_random_completing_next_word(channel_id, word)
+                if next_word is None:
+                    next_word = await db.get_random_next_word(channel_id, word)
+            else:
+                next_word = await db.get_random_next_word(channel_id, word)
 
-    result = " ".join(parts)
+            if next_word is None:
+                break
 
-    # Strip the internal sentence-end marker so it never appears in output.
-    if result and result[-1] == DEFAULT_SENTENCE_END:
-        return result[:-1]
+            word = next_word
 
-    return result
+            parts.append(word)
+            current_length += 1 + len(word)  # +1 for the joining space.
+
+            if current_length >= hard_limit:
+                result = " ".join(parts)[:hard_limit]
+                # Strip the internal sentence-end marker if it ended up at the boundary.
+                if result and result[-1] == DEFAULT_SENTENCE_END:
+                    result = result[:-1]
+                metrics.GENERATED_MESSAGE_LENGTH.observe(len(result))
+                return result
+
+        result = " ".join(parts)
+
+        # Strip the internal sentence-end marker so it never appears in output.
+        if result and result[-1] == DEFAULT_SENTENCE_END:
+            result = result[:-1]
+
+        metrics.GENERATED_MESSAGE_LENGTH.observe(len(result))
+        return result

crabstero/messages.py

@@ -25,7 +25,7 @@ from typing import TYPE_CHECKING
 
 import discord
 
-from crabstero import flags, markov
+from crabstero import flags, markov, metrics
 from crabstero.flags import EntityType, Flag
 
 if TYPE_CHECKING:
@@ -105,12 +105,14 @@ async def reply_to_message(db: Database, message: discord.Message) -> None:
             allowed_mentions=allowed_mentions,
             mention_author=False,
         )
+        metrics.EMBEDS_GENERATED.inc()
     else:
         await message.reply(
             content=body,
             allowed_mentions=allowed_mentions,
             mention_author=False,
         )
+    metrics.REPLIES_SENT.inc()
 
 
 async def ingest_message(db: Database, message: discord.Message) -> None:
@@ -136,11 +138,17 @@ async def ingest_message(db: Database, message: discord.Message) -> None:
     channel_id = message.channel.id
     user_id = message.author.id
 
-    if message.content:
-        await markov.ingest(db, channel_id, user_id, message.content)
+    if not message.content and not message.embeds:
+        return
 
-    for embed in message.embeds:
-        await _ingest_embed(db, channel_id, user_id, embed)
+    with metrics.MESSAGE_INGESTION_DURATION.time():
+        if message.content:
+            await markov.ingest(db, channel_id, user_id, message.content)
+
+        for embed in message.embeds:
+            await _ingest_embed(db, channel_id, user_id, embed)
+
+    metrics.MESSAGES_INGESTED.inc()
 
 
 async def _ingest_embed(

crabstero/metrics.py

@@ -0,0 +1,171 @@
+# Copyright 2026 Logan Fick
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Prometheus metrics definitions and HTTP server for Crabstero.
+
+All metric objects are defined at module level using the default global
+registry. The MetricsServer class wraps an aiohttp application that serves
+the ``/metrics`` scrape endpoint.
+"""
+
+import logging
+
+from aiohttp import web
+from prometheus_client import Counter, Gauge, Histogram, Info
+from prometheus_client.aiohttp import make_aiohttp_handler
+
+from crabstero import __version__
+
+logger = logging.getLogger(__name__)
+
+_FAST_BUCKETS = (0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
+_SLOW_BUCKETS = (0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0)
+_LENGTH_BUCKETS = (10, 50, 100, 200, 500, 750, 1000, 1500, 2000)
+_MESSAGES_BUCKETS = (100, 500, 1000, 5000, 10000, 25000, 50000)
+
+BUILD_INFO = Info("crabstero_build", "Build information")
+BUILD_INFO.info({"version": __version__})
+
+MESSAGES_INGESTED = Counter(
+    "crabstero_messages_ingested_total",
+    "Messages ingested into a Markov chain",
+)
+REPLIES_SENT = Counter(
+    "crabstero_replies_sent_total",
+    "Markov replies sent",
+)
+EMBEDS_GENERATED = Counter(
+    "crabstero_embeds_generated_total",
+    "Embeds included in replies",
+)
+MESSAGES_PROCESSED = Counter(
+    "crabstero_messages_processed_total",
+    "Messages seen by the listener",
+)
+CHANNELS_INGESTED = Counter(
+    "crabstero_channels_ingested_total",
+    "Channels fully ingested successfully",
+)
+CHANNEL_INGESTION_ERRORS = Counter(
+    "crabstero_channel_ingestion_errors_total",
+    "Channel ingestions that failed",
+)
+PINGME_OPTED_IN = Counter(
+    "crabstero_pingme_opted_in_total",
+    "/pingme opt-ins",
+)
+PINGME_OPTED_OUT = Counter(
+    "crabstero_pingme_opted_out_total",
+    "/pingme opt-outs",
+)
+PINGME_ERRORS = Counter(
+    "crabstero_pingme_errors_total",
+    "/pingme exceptions",
+)
+DISCORD_EVENTS = Counter(
+    "crabstero_discord_events_total",
+    "Discord gateway events received",
+    ["event"],
+)
+
+DISCORD_LATENCY = Gauge(
+    "crabstero_discord_latency_seconds",
+    "Discord WebSocket heartbeat latency",
+)
+GUILD_COUNT = Gauge(
+    "crabstero_guild_count",
+    "Guilds the bot is currently in",
+)
+INGESTION_BACKLOG = Gauge(
+    "crabstero_ingestion_backlog_channels",
+    "Channels currently queued for ingestion",
+)
+
+MESSAGE_INGESTION_DURATION = Histogram(
+    "crabstero_message_ingestion_duration_seconds",
+    "Time to ingest one message",
+    buckets=_FAST_BUCKETS,
+)
+CHANNEL_INGESTION_DURATION = Histogram(
+    "crabstero_channel_ingestion_duration_seconds",
+    "Time to fully ingest a channel",
+    buckets=_SLOW_BUCKETS,
+)
+GENERATION_DURATION = Histogram(
+    "crabstero_generation_duration_seconds",
+    "Time to generate a Markov response",
+    buckets=_FAST_BUCKETS,
+)
+GENERATED_MESSAGE_LENGTH = Histogram(
+    "crabstero_generated_message_length_characters",
+    "Length of generated reply bodies in characters",
+    buckets=_LENGTH_BUCKETS,
+)
+CHANNEL_INGESTION_MESSAGES = Histogram(
+    "crabstero_channel_ingestion_messages",
+    "Messages ingested per channel run",
+    buckets=_MESSAGES_BUCKETS,
+)
+
+
+class MetricsServer:
+    """HTTP server that exposes a Prometheus ``/metrics`` scrape endpoint.
+
+    Uses ``aiohttp.web.AppRunner`` and ``TCPSite`` for async-native serving.
+    The handler is provided by ``prometheus_client.aiohttp.make_aiohttp_handler``,
+    which handles compression and content negotiation automatically.
+    """
+
+    def __init__(self, host: str, port: int) -> None:
+        """Store the listen address.
+
+        :param host: The hostname or IP to bind to.
+        :param port: The TCP port to bind to. Use 0 for OS-assigned.
+        """
+        self._host = host
+        self._port = port
+        self._runner: web.AppRunner | None = None
+
+    @property
+    def port(self) -> int:
+        """The TCP port the server is bound to.
+
+        After ``start()``, this reflects the actual port (useful when
+        the constructor received port 0 for OS assignment).
+        """
+        return self._port
+
+    async def start(self) -> None:
+        """Create the aiohttp application and start listening."""
+        if self._runner is not None:
+            raise RuntimeError("Metrics server is already running")
+        app = web.Application()
+        app.router.add_get("/metrics", make_aiohttp_handler())
+        self._runner = web.AppRunner(app)
+        await self._runner.setup()
+        site = web.TCPSite(self._runner, self._host, self._port)
+        await site.start()
+
+        # Resolve the actual bound port when the OS assigned one.
+        if self._port == 0:
+            self._port = self._runner.addresses[0][1]
+
+        logger.info("Metrics server listening on %s:%d.", self._host, self._port)
+
+    async def stop(self) -> None:
+        """Shut down the HTTP server and release resources."""
+        if self._runner is not None:
+            await self._runner.cleanup()
+            self._runner = None
+            logger.info("Metrics server stopped.")

crabstero/tasks/ingestion.py

@@ -22,6 +22,7 @@ from typing import TYPE_CHECKING
 
 import discord
 
+from crabstero import metrics
 from crabstero.messages import ingest_message
 
 if TYPE_CHECKING:
@@ -74,18 +75,23 @@ async def ingest_channel(
 
         logger.info("[%s] Starting ingestion of textable channel history.", channel.id)
 
-        count = 0
-        async for message in channel.history(limit=MAXIMUM_MESSAGES_PER_CHANNEL):
-            count += 1
-            await ingest_message(db, message)
+        with metrics.CHANNEL_INGESTION_DURATION.time():
+            count = 0
+            async for message in channel.history(limit=MAXIMUM_MESSAGES_PER_CHANNEL):
+                count += 1
+                await ingest_message(db, message)
+
+        metrics.CHANNEL_INGESTION_MESSAGES.observe(count)
 
         logger.info(
             "[%s] Ingestion of channel history complete. %d messages ingested.",
             channel.id,
             count,
         )
+        metrics.CHANNELS_INGESTED.inc()
 
     except Exception:
+        metrics.CHANNEL_INGESTION_ERRORS.inc()
         logger.exception(
             "[%s] An error occurred while ingesting textable channel history!",
             channel.id,