feat(hip-3-pusher): Basic metrics, other fixes

Author: merolish

apps/hip-3-pusher/config/config.toml

@@ -1,4 +1,5 @@
 stale_price_threshold_seconds = 5
+prometheus_port = 9090
 
 [hyperliquid]
 market_name = ""
@@ -15,7 +16,7 @@ aws_region_name = "ap-northeast-1"
 
 [lazer]
 lazer_urls = ["wss://pyth-lazer-0.dourolabs.app/v1/stream", "wss://pyth-lazer-1.dourolabs.app/v1/stream"]
-api_key = "lazer_api_key"
+lazer_api_key = "lazer_api_key"
 base_feed_id = 1   # BTC
 base_feed_exponent = -8
 quote_feed_id = 8  # USDT

apps/hip-3-pusher/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "hip-3-pusher"
-version = "0.1.1"
+version = "0.1.2"
 description = "Hyperliquid HIP-3 market oracle pusher"
 readme = "README.md"
 requires-python = ">=3.13"
@@ -10,6 +10,9 @@ dependencies = [
     "cryptography>=45.0.7",
     "hyperliquid-python-sdk>=0.19.0",
     "loguru>=0.7.3",
+    "opentelemetry-exporter-prometheus>=0.58b0",
+    "opentelemetry-sdk>=1.37.0",
+    "prometheus-client>=0.23.1",
     "toml>=0.10.2",
     "websockets>=15.0.1",
 ]

apps/hip-3-pusher/src/main.py

@@ -10,6 +10,7 @@
 from hermes_listener import HermesListener
 from price_state import PriceState
 from publisher import Publisher
+from metrics import Metrics
 
 
 def load_config():
@@ -38,7 +39,9 @@ async def main():
     config = load_config()
 
     price_state = PriceState(config)
-    publisher = Publisher(config, price_state)
+    metrics = Metrics(config)
+
+    publisher = Publisher(config, price_state, metrics)
     hyperliquid_listener = HyperliquidListener(config, price_state)
     lazer_listener = LazerListener(config, price_state)
     hermes_listener = HermesListener(config, price_state)

apps/hip-3-pusher/src/metrics.py

@@ -0,0 +1,37 @@
+from prometheus_client import start_http_server
+from opentelemetry.exporter.prometheus import PrometheusMetricReader
+from opentelemetry.metrics import get_meter_provider, set_meter_provider
+from opentelemetry.sdk.metrics import MeterProvider
+
+METER_NAME = "hip3pusher"
+
+
+class Metrics:
+    def __init__(self, config):
+        # Adapted from opentelemetry-exporter-prometheus example code.
+        # Start Prometheus client
+        start_http_server(port=config["prometheus_port"], addr="localhost")
+        # Exporter to export metrics to Prometheus
+        reader = PrometheusMetricReader()
+        # Meter is responsible for creating and recording metrics
+        set_meter_provider(MeterProvider(metric_readers=[reader]))
+        # TODO: sync version number and add?
+        self.meter = get_meter_provider().get_meter(METER_NAME)
+
+        self._init_metrics()
+
+    def _init_metrics(self):
+        self.no_oracle_price_counter = self.meter.create_counter(
+            name="hip_3_pusher_no_oracle_price_count",
+            description="Number of failed push attempts with no valid oracle price",
+        )
+        self.successful_push_counter = self.meter.create_counter(
+            name="hip_3_pusher_successful_push_count",
+            description="Number of successful push attempts",
+        )
+        self.failed_push_counter = self.meter.create_counter(
+            name="hip_3_pusher_failed_push_count",
+            description="Number of failed push attempts",
+        )
+
+        # TODO: labels/attributes

apps/hip-3-pusher/src/publisher.py

@@ -8,6 +8,7 @@
 from hyperliquid.utils.constants import TESTNET_API_URL, MAINNET_API_URL
 
 from kms_signer import KMSSigner
+from metrics import Metrics
 from price_state import PriceState
 
 
@@ -17,7 +18,7 @@ class Publisher:
 
     See https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/hip-3-deployer-actions
     """
-    def __init__(self, config: dict, price_state: PriceState):
+    def __init__(self, config: dict, price_state: PriceState, metrics: Metrics):
         self.publish_interval = float(config["hyperliquid"]["publish_interval"])
         self.kms_signer = None
         self.enable_kms = False
@@ -43,40 +44,54 @@ def __init__(self, config: dict, price_state: PriceState):
         self.enable_publish = config["hyperliquid"].get("enable_publish", False)
 
         self.price_state = price_state
+        self.metrics = metrics
 
     async def run(self):
         while True:
             await asyncio.sleep(self.publish_interval)
+            try:
+                self.publish()
+            except Exception as e:
+                logger.exception("Publisher.publish() exception: {}", e)
 
-            oracle_pxs = {}
-            oracle_px = self.price_state.get_current_oracle_price()
-            if not oracle_px:
-                logger.error("No valid oracle price available!")
-                return
-            else:
-                logger.debug("Current oracle price: {}", oracle_px)
-                oracle_pxs[self.market_symbol] = oracle_px
+    def publish(self):
+        oracle_pxs = {}
+        oracle_px = self.price_state.get_current_oracle_price()
+        if not oracle_px:
+            logger.error("No valid oracle price available")
+            self.metrics.no_oracle_price_counter.add(1)
+            return
+        else:
+            logger.debug("Current oracle price: {}", oracle_px)
+            oracle_pxs[self.market_symbol] = oracle_px
 
-            mark_pxs = []
-            #if self.price_state.hl_mark_price:
-            #    mark_pxs.append({self.market_symbol: self.price_state.hl_mark_price})
+        mark_pxs = []
+        #if self.price_state.hl_mark_price:
+        #    mark_pxs.append({self.market_symbol: self.price_state.hl_mark_price})
 
-            external_perp_pxs = {}
+        external_perp_pxs = {}
+
+        if self.enable_publish:
+            if self.enable_kms:
+                push_response = self.kms_signer.set_oracle(
+                    dex=self.market_name,
+                    oracle_pxs=oracle_pxs,
+                    all_mark_pxs=mark_pxs,
+                    external_perp_pxs=external_perp_pxs,
+                )
+            else:
+                push_response = self.oracle_publisher_exchange.perp_deploy_set_oracle(
+                    dex=self.market_name,
+                    oracle_pxs=oracle_pxs,
+                    all_mark_pxs=mark_pxs,
+                    external_perp_pxs=external_perp_pxs,
+                )
 
-            if self.enable_publish:
-                if self.enable_kms:
-                    push_response = self.kms_signer.set_oracle(
-                        dex=self.market_name,
-                        oracle_pxs=oracle_pxs,
-                        all_mark_pxs=mark_pxs,
-                        external_perp_pxs=external_perp_pxs,
-                    )
-                else:
-                    push_response = self.oracle_publisher_exchange.perp_deploy_set_oracle(
-                        dex=self.market_name,
-                        oracle_pxs=oracle_pxs,
-                        all_mark_pxs=mark_pxs,
-                        external_perp_pxs=external_perp_pxs,
-                    )
-                # TODO: Look at specific error responses and log/alert accordingly
-                logger.info("publish: push response: {}", push_response)
+            # TODO: Look at specific error responses and log/alert accordingly
+            logger.debug("publish: push response: {} {}", push_response, type(push_response))
+            status = push_response.get("status", "")
+            if status == "ok":
+                self.metrics.successful_push_counter.add(1)
+            elif status == "err":
+                self.metrics.failed_push_counter.add(1)
+                logger.error("publish: publish error: {}", push_response)