feat(hip-3-pusher): improved metrics (#3243)

Author: merolish

apps/hip-3-pusher/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "hip-3-pusher"
-version = "0.2.1"
+version = "0.2.3"
 description = "Hyperliquid HIP-3 market oracle pusher"
 readme = "README.md"
 requires-python = "==3.13.*"

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

@@ -27,6 +27,7 @@ def load_config():
         config_toml = tomllib.load(config_file)
         config = Config(**config_toml)
         logger.debug("Config loaded: {}", config)
+        logger.info("Price config: {}", config.price)
     return config
 
 
@@ -44,6 +45,7 @@ async def main():
 
     price_state = PriceState(config)
     metrics = Metrics(config)
+    metrics.set_price_configs(config.hyperliquid.market_name, config.price)
 
     publisher = Publisher(config, price_state, metrics)
     hyperliquid_listener = HyperliquidListener(config, price_state.hl_oracle_state, price_state.hl_mark_state, price_state.hl_mid_state)

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

@@ -3,7 +3,8 @@
 from opentelemetry.metrics import get_meter_provider, set_meter_provider
 from opentelemetry.sdk.metrics import MeterProvider
 
-from pusher.config import Config
+from pusher.config import Config, PriceConfig, PriceSourceConfig, ConstantSourceConfig, SingleSourceConfig, \
+    PairSourceConfig, OracleMidAverageConfig, PriceSource
 
 METER_NAME = "hip3pusher"
 
@@ -21,20 +22,62 @@ def __init__(self, config: Config):
         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",
+        # labels: dex, symbol
+        self.last_pushed_time = self.meter.create_gauge(
+            name="hip_3_relayer_last_published_time",
+            description="Time of last successful oracle update",
         )
-        self.successful_push_counter = self.meter.create_counter(
-            name="hip_3_pusher_successful_push_count",
-            description="Number of successful push attempts",
+        # labels: dex, symbol, status, error_reason
+        self.update_attempts_total = self.meter.create_counter(
+            name="hip_3_relayer_update_attempts_total",
+            description="Number of update attempts",
         )
-        self.failed_push_counter = self.meter.create_counter(
-            name="hip_3_pusher_failed_push_count",
-            description="Number of failed push attempts",
+        # labels: dex
+        self.no_oracle_price_counter = self.meter.create_counter(
+            name="hip_3_relayer_no_oracle_price_count",
+            description="Number of failed push attempts with no valid oracle price",
         )
+        # labels: dex
         self.push_interval_histogram = self.meter.create_histogram(
-            name="hip_3_pusher_push_interval",
+            name="hip_3_relayer_push_interval",
             description="Interval between push requests (seconds)",
             unit="s",
         )
+        # labels: dex, price_type, symbol
+        self.price_config_counter = self.meter.create_counter(
+            name="hip_3_relayer_price_config",
+            description="Price source config",
+        )
+
+    def set_price_configs(self, dex: str, price_config: PriceConfig):
+        self._set_price_config_type(dex, price_config.oracle, "oracle")
+        self._set_price_config_type(dex, price_config.mark, "mark")
+        self._set_price_config_type(dex, price_config.external, "external")
+
+    def _set_price_config_type(self, dex: str, price_source_config: dict[str, list[PriceSourceConfig]], price_type: str):
+        for symbol in price_source_config:
+            source_config_str = ' | '.join(self._get_source_config_str(source_config) for source_config in price_source_config[symbol])
+            labels = {
+                "dex": dex,
+                "symbol": symbol,
+                "price_type": price_type,
+                "config": source_config_str,
+            }
+            self.price_config_counter.add(1, labels)
+
+    def _get_source_config_str(self, source_config: PriceSourceConfig):
+        if isinstance(source_config, ConstantSourceConfig):
+            return f"constant({source_config.value})"
+        elif isinstance(source_config, SingleSourceConfig):
+            return self._get_price_source_str(source_config.source)
+        elif isinstance(source_config, PairSourceConfig):
+            base_str = self._get_price_source_str(source_config.base_source)
+            quote_str = self._get_price_source_str(source_config.quote_source)
+            return f"pair({base_str},{quote_str})"
+        elif isinstance(source_config, OracleMidAverageConfig):
+            return f"oracle_mid_average({source_config.symbol})"
+        else:
+            return "unknown"
+
+    def _get_price_source_str(self, price_source: PriceSource):
+        return f"{price_source.source_name}({str(price_source.source_id)[:8]})"

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

@@ -1,4 +1,5 @@
 import asyncio
+from enum import StrEnum
 import time
 
 from loguru import logger
@@ -16,6 +17,20 @@
 from pusher.price_state import PriceState
 
 
+class PushErrorReason(StrEnum):
+    """ setOracle push failure modes """
+    # 2.5s rate limit reject, expected with redundant relayers
+    RATE_LIMIT = "rate_limit"
+    # Per-account limit, need to purchase more transactions with reserveRequestWeight
+    USER_LIMIT = "user_limit"
+    # Some exception thrown internally
+    INTERNAL_ERROR = "internal_error"
+    # Invalid nonce, if the pusher account pushes multiple transactions with the same ms timestamp
+    INVALID_NONCE = "invalid_nonce"
+    # Some error string we haven't categorized yet
+    UNKNOWN = "unknown"
+
+
 class Publisher:
     """
     HIP-3 oracle publisher handler
@@ -51,6 +66,8 @@ def __init__(self, config: Config, price_state: PriceState, metrics: Metrics):
             if not config.multisig.multisig_address:
                 raise Exception("Multisig enabled but missing multisig address")
             self.multisig_address = config.multisig.multisig_address
+        else:
+            self.multisig_address = None
 
         self.market_name = config.hyperliquid.market_name
         self.enable_publish = config.hyperliquid.enable_publish
@@ -79,9 +96,6 @@ def publish(self):
         # markPxs is a list of dicts of length 0-2, and so can be empty
         mark_pxs = [mark_pxs] if mark_pxs else []
 
-        # TODO: "Each update can change oraclePx and markPx by at most 1%."
-        # TODO: "The markPx cannot be updated such that open interest would be 10x the open interest cap."
-
         if self.enable_publish:
             try:
                 if self.enable_kms:
@@ -103,12 +117,13 @@ def publish(self):
                         all_mark_pxs=mark_pxs,
                         external_perp_pxs=external_perp_pxs,
                     )
-                self._handle_response(push_response)
+                self._handle_response(push_response, list(oracle_pxs.keys()))
             except PushError:
-                logger.error("All push attempts failed")
-                self.metrics.failed_push_counter.add(1, self.metrics_labels)
+                # since rate limiting is expected, don't necessarily log
+                pass
             except Exception as e:
                 logger.exception("Unexpected exception in push request: {}", repr(e))
+                self._update_attempts_total("error", PushErrorReason.INTERNAL_ERROR, list(oracle_pxs.keys()))
         else:
             logger.debug("push disabled")
 
@@ -128,14 +143,27 @@ def _send_update(self, oracle_pxs, all_mark_pxs, external_perp_pxs):
 
         raise PushError("all push endpoints failed")
 
-    def _handle_response(self, response):
+    def _handle_response(self, response, symbols: list[str]):
         logger.debug("oracle update response: {}", response)
         status = response.get("status")
         if status == "ok":
-            self.metrics.successful_push_counter.add(1, self.metrics_labels)
+            self._update_attempts_total("success", None, symbols)
+            time_secs = int(time.time())
+
+            # update last publish time for each symbol in dex
+            for symbol in symbols:
+                labels = {**self.metrics_labels, "symbol": symbol}
+                self.metrics.last_pushed_time.set(time_secs, labels)
+
+            # log any data in the ok response (likely price clamping issues)
+            ok_data = response.get("response", {}).get("data")
+            if ok_data:
+                logger.info("ok response data: {}", ok_data)
         elif status == "err":
-            self.metrics.failed_push_counter.add(1, self.metrics_labels)
-            logger.error("oracle update error response: {}", response)
+            error_reason = self._get_error_reason(response)
+            self._update_attempts_total("error", error_reason, symbols)
+            if error_reason != "rate_limit":
+                logger.error("Error response: {}", response)
 
     def _record_push_interval_metric(self):
         now = time.time()
@@ -183,3 +211,29 @@ def _send_single_multisig_update(self, exchange, oracle_pxs, all_mark_pxs, exter
             outer_signer=self.oracle_account.address,
         )]
         return exchange.multi_sig(self.multisig_address, action, signatures, timestamp)
+
+    def _update_attempts_total(self, status: str, error_reason: str | None, symbols: list[str]):
+        labels = {**self.metrics_labels, "status": status}
+        if error_reason:
+            # don't flag rate limiting as this is expected with redundancy
+            if error_reason == "rate_limit":
+                return
+            labels["error_reason"] = error_reason
+
+        for symbol in symbols:
+            labels["symbol"] = symbol
+            self.metrics.update_attempts_total.add(1, labels)
+
+    def _get_error_reason(self, response):
+        response = response.get("response")
+        if not response:
+            return None
+        elif "Oracle price update too often" in response:
+            return PushErrorReason.RATE_LIMIT
+        elif "Too many cumulative requests" in response:
+            return PushErrorReason.USER_LIMIT
+        elif "Invalid nonce" in response:
+            return PushErrorReason.INVALID_NONCE
+        else:
+            logger.warning("Unrecognized error response: {}", response)
+            return PushErrorReason.UNKNOWN

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

@@ -17,7 +17,7 @@ class SedaListener:
     """
     def __init__(self, config: Config, seda_state: PriceSourceState):
         self.url = config.seda.url
-        self.api_key = Path(config.seda.api_key_path).read_text().strip()
+        self.api_key = Path(config.seda.api_key_path).read_text().strip() if config.seda.api_key_path else None
         self.feeds = config.seda.feeds
         self.poll_interval = config.seda.poll_interval
         self.poll_failure_interval = config.seda.poll_failure_interval