feat(reliability) Add flag to disable telegram notifications

pyth_observer/__init__.py

@@ -55,9 +55,10 @@ def __init__(
         config: Dict[str, Any],
         publishers: Dict[str, Publisher],
         coingecko_mapping: Dict[str, Symbol],
+        disable_telegram: bool = False,
     ):
         self.config = config
-        self.dispatch = Dispatch(config, publishers)
+        self.dispatch = Dispatch(config, publishers, disable_telegram)
         self.publishers = publishers
         self.pyth_client = PythClient(
             solana_endpoint=config["network"]["http_endpoint"],
@@ -75,6 +76,7 @@ def __init__(
         metrics.set_observer_info(
             network=config["network"]["name"],
             config=config,
+            telegram_enabled=not disable_telegram,
         )
 
     async def run(self):

pyth_observer/cli.py

@@ -37,7 +37,14 @@
     envvar="PROMETHEUS_PORT",
     default="9001",
 )
-def run(config, publishers, coingecko_mapping, prometheus_port):
+@click.option(
+    "--disable-telegram",
+    help="Disable sending Telegram notifications",
+    envvar="DISABLE_TELEGRAM",
+    is_flag=True,
+    default=False,
+)
+def run(config, publishers, coingecko_mapping, prometheus_port, disable_telegram):
     config_ = yaml.safe_load(open(config, "r"))
     # Load publishers YAML file and convert to dictionary of Publisher instances
     publishers_raw = yaml.safe_load(open(publishers, "r"))
@@ -54,11 +61,7 @@ def run(config, publishers, coingecko_mapping, prometheus_port):
         for publisher in publishers_raw
     }
     coingecko_mapping_ = yaml.safe_load(open(coingecko_mapping, "r"))
-    observer = Observer(
-        config_,
-        publishers_,
-        coingecko_mapping_,
-    )
+    observer = Observer(config_, publishers_, coingecko_mapping_, disable_telegram)
 
     start_http_server(int(prometheus_port))
 

pyth_observer/dispatch.py

@@ -29,9 +29,10 @@ class Dispatch:
     notifiers for the checks that failed.
     """
 
-    def __init__(self, config, publishers):
+    def __init__(self, config, publishers, disable_telegram=False):
         self.config = config
         self.publishers = publishers
+        self.disable_telegram = disable_telegram
         if "ZendutyEvent" in self.config["events"]:
             self.open_alerts_file = os.environ["OPEN_ALERTS_FILE"]
             self.open_alerts = self.load_alerts()
@@ -67,6 +68,8 @@ async def run(self, states: List[State]):
         current_time = datetime.now()
         for check in failed_checks:
             for event_type in self.config["events"]:
+                if event_type == "TelegramEvent" and self.disable_telegram:
+                    continue
                 event: Event = globals()[event_type](check, context)
 
                 if event_type in ["ZendutyEvent", "TelegramEvent"]:

pyth_observer/metrics.py

@@ -148,7 +148,9 @@ def __init__(self, registry: CollectorRegistry = REGISTRY):
             registry=registry,
         )
 
-    def set_observer_info(self, network: str, config: Dict[str, Any]):
+    def set_observer_info(
+        self, network: str, config: Dict[str, Any], telegram_enabled: bool = False
+    ):
         """Set static information about the observer instance."""
         self.observer_info.info(
             {
@@ -163,6 +165,7 @@ def set_observer_info(self, network: str, config: Dict[str, Any]):
                     )
                 ),
                 "event_handlers": ",".join(config.get("events", [])),
+                "telegram_enabled": str(int(telegram_enabled)),
             }
         )