Prometheus integration (#198)

* Initial prometheus integration

* datasource_head_updated

* Callback time

* more duration metrics

* sync_duration (why?)

* Fix immediate stopping after oneshot execution

* metrics.index_realtime_queue_size

* total, synched, realtime

* Some renamings

* docs

* Cleanup

* Cleanup

* More metrics

* Fixes

* Fix index counter

* min/max/avg for durations

* averaged durations everywhere

* Refactoring

* Count rollbacks

* http_errors

* index_hit_ratio, refactoring

* Total matches

* Useful options

* Opt-in prometheus (dirty)

* Same for bigmap indexes

* Require host field

* typo

* Build images, cooler pyproject injectr

* Changelog

* Cleanup

* indexes_total

* Review, bump

* Do not MA metrics

Author: droserasprout

.github/workflows/build.yml

@@ -46,4 +46,4 @@ jobs:
           cache-from: type=gha
           cache-to: type=gha,mode=max
           tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
+          labels: ${{ steps.meta.outputs.labels }}

CHANGELOG.md

@@ -7,6 +7,7 @@ Please use [this](https://docs.gitlab.com/ee/development/changelog.html) documen
 ### Added
 
 * metadata: Added `metadata_interface` feature flag to expose metadata in TzKT format.
+* prometheus: Added ability to expose Prometheus metrics.
 * tzkt: Added missing fields to the `HeadBlockData` model.
 * tzkt: Added `iter_...` methods to iterate over item batches.
 

Dockerfile

@@ -19,9 +19,11 @@ WORKDIR /home/dipdup/source
 RUN poetry config virtualenvs.create false
 RUN make install DEV=0 PLUGINS="${PLUGINS}"
 
-COPY --chown=dipdup inject_pyproject.sh /home/dipdup/inject_pyproject
-RUN chmod +x /home/dipdup/inject_pyproject
-RUN echo 'dipdup ALL = NOPASSWD: /home/dipdup/inject_pyproject' >> /etc/sudoers
+COPY --chown=dipdup inject_pyproject.sh /usr/bin/inject_pyproject.sh
+RUN echo 'sudo /usr/bin/inject_pyproject.sh' >> /usr/bin/inject_pyproject
+RUN echo 'dipdup ALL = NOPASSWD: /usr/bin/inject_pyproject.sh' >> /etc/sudoers
+RUN chmod +x /usr/bin/inject_pyproject.sh
+RUN chmod +x /usr/bin/inject_pyproject
 
 COPY --chown=dipdup src /home/dipdup/source/src
 

poetry.lock

@@ -42,6 +42,7 @@ sqlparse = "^0.4.1"
 tabulate = "^0.8.9"
 tortoise-orm = "0.17.8"
 typing-inspect = "^0.6.0"
+prometheus-client = "^0.12.0"
 
 [tool.poetry.dev-dependencies]
 black = "^22.1.0"

pyproject.toml

@@ -906,6 +906,13 @@ class SentryConfig:
     debug: bool = False
 
 
+@dataclass
+class PrometheusConfig:
+    host: str
+    port: int = 8000
+    update_interval: float = 1.0
+
+
 @dataclass
 class HookConfig(CallbackMixin, kind='hook'):
     args: Dict[str, str] = field(default_factory=dict)
@@ -989,6 +996,7 @@ class DipDupConfig:
     hooks: Dict[str, HookConfig] = field(default_factory=dict)
     hasura: Optional[HasuraConfig] = None
     sentry: Optional[SentryConfig] = None
+    prometheus: Optional[PrometheusConfig] = None
     advanced: AdvancedConfig = AdvancedConfig()
 
     def __post_init_post_parse__(self):

src/dipdup/config.py

@@ -1,9 +1,9 @@
 import logging
 import os
 import sys
-import time
 from collections import deque
 from contextlib import AsyncExitStack
+from contextlib import ExitStack
 from contextlib import contextmanager
 from contextlib import suppress
 from os.path import exists
@@ -59,6 +59,7 @@
 from dipdup.models import ReindexingReason
 from dipdup.models import Schema
 from dipdup.models import TokenMetadata
+from dipdup.prometheus import Metrics
 from dipdup.utils import FormattedLogger
 from dipdup.utils import slowdown
 from dipdup.utils.database import execute_sql_scripts
@@ -455,11 +456,10 @@ async def execute_sql(self, ctx: 'DipDupContext', name: str) -> None:
     @contextmanager
     def _callback_wrapper(self, kind: str, name: str) -> Iterator[None]:
         try:
-            start = time.perf_counter()
-            yield
-            diff = time.perf_counter() - start
-            level = self._logger.warning if diff > 1 else self._logger.debug
-            level('`%s` %s callback executed in %s seconds', name, kind, diff)
+            with ExitStack() as stack:
+                if Metrics.enabled:
+                    stack.enter_context(Metrics.measure_callback_duration(name))
+                yield
         except Exception as e:
             if isinstance(e, ReindexingRequiredError):
                 raise

src/dipdup/context.py

@@ -19,6 +19,7 @@
 from typing import cast
 
 from apscheduler.events import EVENT_JOB_ERROR  # type: ignore
+from prometheus_client import start_http_server  # type: ignore
 from tortoise.exceptions import OperationalError
 from tortoise.transactions import get_connection
 
@@ -46,9 +47,7 @@
 from dipdup.index import HeadIndex
 from dipdup.index import Index
 from dipdup.index import OperationIndex
-from dipdup.index import block_cache
 from dipdup.index import extract_operation_subgroups
-from dipdup.index import head_cache
 from dipdup.models import BigMapData
 from dipdup.models import Contract
 from dipdup.models import Head
@@ -57,6 +56,7 @@
 from dipdup.models import IndexStatus
 from dipdup.models import OperationData
 from dipdup.models import Schema
+from dipdup.prometheus import Metrics
 from dipdup.scheduler import add_job
 from dipdup.scheduler import create_scheduler
 from dipdup.utils import slowdown
@@ -74,18 +74,18 @@ def __init__(self, ctx: DipDupContext) -> None:
         self._logger = logging.getLogger('dipdup')
         self._indexes: Dict[str, Index] = {}
         self._contracts: Set[ContractConfig] = set()
-        self._stopped: bool = False
         self._tasks: Deque[asyncio.Task] = deque()
 
         self._entrypoint_filter: Set[Optional[str]] = set()
         self._address_filter: Set[str] = set()
 
-    async def run(
-        self,
-        spawn_datasources_event: Event,
-        start_scheduler_event: Event,
-        early_realtime: bool = False,
-    ) -> None:
+    async def run(self, spawn_datasources_event: Event, start_scheduler_event: Event, early_realtime: bool = False) -> None:
+        tasks = [self._run(spawn_datasources_event, start_scheduler_event, early_realtime)]
+        if self._ctx.config.prometheus:
+            tasks.append(self._update_metrics(self._ctx.config.prometheus.update_interval))
+        await gather(*tasks)
+
+    async def _run(self, spawn_datasources_event: Event, start_scheduler_event: Event, early_realtime: bool = False) -> None:
         self._logger.info('Starting index dispatcher')
         await self._subscribe_to_datasource_events()
         await self._load_index_states()
@@ -96,7 +96,7 @@ async def run(
             if isinstance(index, OperationIndex):
                 self._apply_filters(index._config)
 
-        while not self._stopped:
+        while True:
             if not spawn_datasources_event.is_set():
                 if self._every_index_is(IndexStatus.REALTIME) or early_realtime:
                     spawn_datasources_event.set()
@@ -123,7 +123,7 @@ async def run(
                     self._apply_filters(index._config)
 
             if not indexes_spawned and self._every_index_is(IndexStatus.ONESHOT):
-                self.stop()
+                break
 
             if self._every_index_is(IndexStatus.REALTIME) and not indexes_spawned:
                 if not on_synchronized_fired:
@@ -136,8 +136,19 @@ async def run(
             else:
                 on_synchronized_fired = False
 
-    def stop(self) -> None:
-        self._stopped = True
+    async def _update_metrics(self, update_interval: float) -> None:
+        while True:
+            await asyncio.sleep(update_interval)
+
+            active, synced, realtime = 0, 0, 0
+            for index in tuple(self._indexes.values()) + tuple(pending_indexes):
+                active += 1
+                if index.synchronized:
+                    synced += 1
+                if index.realtime:
+                    realtime += 1
+
+            Metrics.set_indexes_count(active, synced, realtime)
 
     def _apply_filters(self, index_config: OperationIndexConfig) -> None:
         self._address_filter.update(index_config.address_filter)
@@ -215,10 +226,6 @@ async def _process(index_state: IndexState) -> None:
         tasks = (create_task(_process(index_state)) for index_state in await IndexState.all())
         await gather(*tasks)
 
-        # NOTE: Cached blocks used only on index state init
-        block_cache.clear()
-        head_cache.clear()
-
     async def _on_head(self, datasource: TzktDatasource, head: HeadBlockData) -> None:
         # NOTE: Do not await query results - blocked database connection may cause Websocket timeout.
         self._tasks.append(
@@ -233,6 +240,8 @@ async def _on_head(self, datasource: TzktDatasource, head: HeadBlockData) -> Non
                 ),
             )
         )
+        if Metrics.enabled:
+            Metrics.set_datasource_head_updated(datasource.name)
         for index in self._indexes.values():
             if isinstance(index, HeadIndex) and index.datasource == datasource:
                 index.push_head(head)
@@ -261,6 +270,8 @@ async def _on_big_maps(self, datasource: TzktDatasource, big_maps: Tuple[BigMapD
     async def _on_rollback(self, datasource: TzktDatasource, from_level: int, to_level: int) -> None:
         """Perform a single level rollback when possible, otherwise call `on_rollback` hook"""
         self._logger.warning('Datasource `%s` rolled back: %s -> %s', datasource.name, from_level, to_level)
+        if Metrics.enabled:
+            Metrics.set_datasource_rollback(datasource.name)
 
         # NOTE: Zero difference between levels means we received no operations/big_maps on this level and thus channel level hasn't changed
         zero_level_rollback = from_level - to_level == 0
@@ -341,6 +352,7 @@ async def run(self) -> None:
             await self._set_up_database(stack)
             await self._set_up_datasources(stack)
             await self._set_up_hooks(tasks)
+            await self._set_up_prometheus()
 
             await self._initialize_schema()
             await self._initialize_datasources()
@@ -446,6 +458,11 @@ async def _set_up_hooks(self, tasks: Optional[Set[Task]] = None) -> None:
         if tasks:
             tasks.add(create_task(self._ctx.callbacks.run()))
 
+    async def _set_up_prometheus(self) -> None:
+        if self._config.prometheus:
+            Metrics.enabled = True
+            start_http_server(self._config.prometheus.port, self._config.prometheus.host)
+
     async def _set_up_hasura(self, stack: AsyncExitStack) -> None:
         if not self._config.hasura:
             return

src/dipdup/dipdup.py

@@ -19,7 +19,8 @@
 from fcache.cache import FileCache  # type: ignore
 
 from dipdup import __version__
-from dipdup.config import HTTPConfig
+from dipdup.config import HTTPConfig  # type: ignore
+from dipdup.prometheus import Metrics
 
 safe_exceptions = (
     aiohttp.ClientConnectionError,
@@ -113,6 +114,7 @@ def _session(self) -> aiohttp.ClientSession:
             raise RuntimeError('aiohttp session is closed')
         return self.__session
 
+    # TODO: Move to separate method to cover SignalR negotiations too
     async def _retry_request(self, method: str, url: str, weight: int = 1, **kwargs):
         """Retry a request in case of failure sleeping according to config"""
         attempt = 1
@@ -134,13 +136,20 @@ async def _retry_request(self, method: str, url: str, weight: int = 1, **kwargs)
                     raise e
 
                 ratelimit_sleep: Optional[float] = None
-                if isinstance(e, aiohttp.ClientResponseError) and e.status == HTTPStatus.TOO_MANY_REQUESTS:
-                    # NOTE: Sleep at least 5 seconds on ratelimit
-                    ratelimit_sleep = 5
-                    # TODO: Parse Retry-After in UTC date format
-                    with suppress(KeyError, ValueError):
-                        e.headers = cast(Mapping, e.headers)
-                        ratelimit_sleep = int(e.headers['Retry-After'])
+                if isinstance(e, aiohttp.ClientResponseError):
+                    if Metrics.enabled:
+                        Metrics.set_http_error(self._url, e.status)
+
+                    if e.status == HTTPStatus.TOO_MANY_REQUESTS:
+                        # NOTE: Sleep at least 5 seconds on ratelimit
+                        ratelimit_sleep = 5
+                        # TODO: Parse Retry-After in UTC date format
+                        with suppress(KeyError, ValueError):
+                            e.headers = cast(Mapping, e.headers)
+                            ratelimit_sleep = int(e.headers['Retry-After'])
+                else:
+                    if Metrics.enabled:
+                        Metrics.set_http_error(self._url, 0)
 
                 self._logger.warning('HTTP request attempt %s/%s failed: %s', attempt, retry_count_str, e)
                 self._logger.info('Waiting %s seconds before retry', ratelimit_sleep or retry_sleep)