Getting online domains using peers health check

packages/grid/backend/grid/core/node.py

@@ -105,4 +105,5 @@ def seaweedfs_config() -> SeaweedFSConfig:
  smtp_port=settings.SMTP_PORT,
  smtp_host=settings.SMTP_HOST,
  association_request_auto_approval=settings.ASSOCIATION_REQUEST_AUTO_APPROVAL,
+ background_tasks=True,
 )

packages/hagrid/.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.3.119
+current_version = 0.3.121
 tag = False
 tag_name = {new_version}
 commit = True

packages/hagrid/hagrid/manifest_template.yml

@@ -1,9 +1,9 @@
 manifestVersion: 0.1
-hagrid_version: 0.3.119
+hagrid_version: 0.3.121
 syft_version: 0.8.7-beta.7
 dockerTag: 0.8.7-beta.7
 baseUrl: https://raw.githubusercontent.com/OpenMined/PySyft/
-hash: 90713c314a1ac09cb604d0efa7d414e9811f2691
+hash: 4333433d5bec7bb9bcd52db59029d3bcb23c74c2
 target_dir: ~/.hagrid/PySyft/
 files:
  grid:

packages/hagrid/hagrid/orchestra.py

@@ -238,6 +238,7 @@ def deploy_to_python(
  create_producer: bool = False,
  queue_port: int | None = None,
  association_request_auto_approval: bool = False,
+ background_tasks: bool = False,
 ) -> NodeHandle | None:
  stage_protocol_changes = ImportFromSyft.import_stage_protocol_changes()
  NodeType = ImportFromSyft.import_node_type()
@@ -272,7 +273,7 @@ def deploy_to_python(
  "n_consumers": n_consumers,
  "create_producer": create_producer,
  "association_request_auto_approval": association_request_auto_approval,
- "background_tasks": True,
+ "background_tasks": background_tasks,
  }
 
  if port:
@@ -493,6 +494,7 @@ def launch(
  queue_port: int | None = None,
  in_memory_workers: bool = True,
  association_request_auto_approval: bool = False,
+ background_tasks: bool = False,
  ) -> NodeHandle | None:
  NodeType = ImportFromSyft.import_node_type()
  os.environ["DEV_MODE"] = str(dev_mode)
@@ -540,6 +542,7 @@ def launch(
  create_producer=create_producer,
  queue_port=queue_port,
  association_request_auto_approval=association_request_auto_approval,
+ background_tasks=background_tasks,
  )
 
  elif deployment_type_enum == DeploymentType.K8S:

packages/hagrid/hagrid/version.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # HAGrid Version
-__version__ = "0.3.119"
+__version__ = "0.3.121"
 
 if __name__ == "__main__":
  print(__version__)

packages/hagrid/setup.py

@@ -5,7 +5,7 @@
 from setuptools import find_packages
 from setuptools import setup
 
-__version__ = "0.3.119"
+__version__ = "0.3.121"
 
 DATA_FILES = {"img": ["hagrid/img/*.png"], "hagrid": ["*.yml"]}
 

packages/syft/src/syft/client/registry.py

@@ -13,7 +13,8 @@
 
 # relative
 from ..service.metadata.node_metadata import NodeMetadataJSON
-from ..service.network.network_service import NodePeer
+from ..service.network.node_peer import NodePeer
+from ..service.network.node_peer import NodePeerConnectionStatus
 from ..service.response import SyftException
 from ..types.grid_url import GridURL
 from ..util.constants import DEFAULT_TIMEOUT
@@ -80,10 +81,10 @@ def check_network(network: dict) -> dict[Any, Any] | None:
  except Exception:
  online = False
 
- # networks without frontend have a /ping route in 0.7.0
+ # networks without frontend
  if not online:
  try:
- ping_url = url + "ping"
+ ping_url = url + "api/v2/"
  res = requests.get(ping_url, timeout=DEFAULT_TIMEOUT) # nosec
  online = res.status_code == 200
  except Exception:
@@ -120,13 +121,40 @@ def _repr_html_(self) -> str:
  on = self.online_networks
  if len(on) == 0:
  return "(no gateways online - try syft.gateways.all_networks to see offline gateways)"
- return pd.DataFrame(on)._repr_html_() # type: ignore
+ df = pd.DataFrame(on)
+ total_df = pd.DataFrame(
+ [
+ [
+ f"{len(on)} / {len(self.all_networks)} (online networks / all networks)"
+ ]
+ + [""] * (len(df.columns) - 1)
+ ],
+ columns=df.columns,
+ index=["Total"],
+ )
+ df = pd.concat([df, total_df])
+ return df._repr_html_() # type: ignore
 
  def __repr__(self) -> str:
  on = self.online_networks
  if len(on) == 0:
  return "(no gateways online - try syft.gateways.all_networks to see offline gateways)"
- return pd.DataFrame(on).to_string()
+ df = pd.DataFrame(on)
+ total_df = pd.DataFrame(
+ [
+ [
+ f"{len(on)} / {len(self.all_networks)} (online networks / all networks)"
+ ]
+ + [""] * (len(df.columns) - 1)
+ ],
+ columns=df.columns,
+ index=["Total"],
+ )
+ df = pd.concat([df, total_df])
+ return df.to_string()
+
+ def __len__(self) -> int:
+ return len(self.all_networks)
 
  @staticmethod
  def create_client(network: dict[str, Any]) -> Client:
@@ -189,10 +217,10 @@ def check_network(network: dict) -> dict[Any, Any] | None:
  except Exception:
  online = False
 
- # networks without frontend have a /ping route in 0.7.0
+ # networks without frontend
  if not online:
  try:
- ping_url = url + "ping"
+ ping_url = url + "api/v2/"
  res = requests.get(ping_url, timeout=DEFAULT_TIMEOUT)
  online = res.status_code == 200
  except Exception:
@@ -228,32 +256,25 @@ def check_network(network: dict) -> dict[Any, Any] | None:
 
  @property
  def online_domains(self) -> list[tuple[NodePeer, NodeMetadataJSON | None]]:
- def check_domain(
- peer: NodePeer,
- ) -> tuple[NodePeer, NodeMetadataJSON | None] | None:
- try:
- guest_client = peer.guest_client
- metadata = guest_client.metadata
- return peer, metadata
- except Exception as e: # nosec
- print(f"Error in checking domain with exception {e}")
- return None
-
  networks = self.online_networks
 
- # We can use a with statement to ensure threads are cleaned up promptly
- with futures.ThreadPoolExecutor(max_workers=20) as executor:
- # map
- _all_online_domains = []
- for network in networks:
+ _all_online_domains = []
+ for network in networks:
+ try:
  network_client = NetworkRegistry.create_client(network)
- domains: list[NodePeer] = network_client.domains.retrieve_nodes()
- for domain in domains:
- self.all_domains[str(domain.id)] = domain
- _online_domains = list(
- executor.map(lambda domain: check_domain(domain), domains)
- )
- _all_online_domains += _online_domains
+ except Exception as e:
+ print(f"Error in creating network client with exception {e}")
+ continue
+
+ domains: list[NodePeer] = network_client.domains.retrieve_nodes()
+ for domain in domains:
+ self.all_domains[str(domain.id)] = domain
+
+ _all_online_domains += [
+ (domain, domain.guest_client.metadata)
+ for domain in domains
+ if domain.ping_status == NodePeerConnectionStatus.ACTIVE
+ ]
 
  return [domain for domain in _all_online_domains if domain is not None]
 
@@ -281,13 +302,33 @@ def _repr_html_(self) -> str:
  on: list[dict[str, Any]] = self.__make_dict__()
  if len(on) == 0:
  return "(no domains online - try syft.domains.all_domains to see offline domains)"
- return pd.DataFrame(on)._repr_html_() # type: ignore
+ df = pd.DataFrame(on)
+ total_df = pd.DataFrame(
+ [
+ [f"{len(on)} / {len(self.all_domains)} (online domains / all domains)"]
+ + [""] * (len(df.columns) - 1)
+ ],
+ columns=df.columns,
+ index=["Total"],
+ )
+ df = pd.concat([df, total_df])
+ return df._repr_html_() # type: ignore
 
  def __repr__(self) -> str:
  on: list[dict[str, Any]] = self.__make_dict__()
  if len(on) == 0:
  return "(no domains online - try syft.domains.all_domains to see offline domains)"
- return pd.DataFrame(on).to_string()
+ df = pd.DataFrame(on)
+ total_df = pd.DataFrame(
+ [
+ [f"{len(on)} / {len(self.all_domains)} (online domains / all domains)"]
+ + [""] * (len(df.columns) - 1)
+ ],
+ columns=df.columns,
+ index=["Total"],
+ )
+ df = pd.concat([df, total_df])
+ return df._repr_html_() # type: ignore
 
  def create_client(self, peer: NodePeer) -> Client:
  try:

packages/syft/src/syft/node/server.py

@@ -79,6 +79,7 @@ def run_uvicorn(
  create_producer: bool,
  association_request_auto_approval: bool,
  n_consumers: int,
+ background_tasks: bool,
 ) -> None:
  async def _run_uvicorn(
  name: str,
@@ -112,6 +113,7 @@ async def _run_uvicorn(
  create_producer=create_producer,
  n_consumers=n_consumers,
  association_request_auto_approval=association_request_auto_approval,
+ background_tasks=background_tasks,
  )
  else:
  worker = worker_class(
@@ -127,6 +129,7 @@ async def _run_uvicorn(
  create_producer=create_producer,
  n_consumers=n_consumers,
  association_request_auto_approval=association_request_auto_approval,
+ background_tasks=background_tasks,
  )
  router = make_routes(worker=worker)
  app = make_app(worker.name, router=router)
@@ -186,6 +189,7 @@ def serve_node(
  create_producer: bool = False,
  n_consumers: int = 0,
  association_request_auto_approval: bool = False,
+ background_tasks: bool = False,
 ) -> tuple[Callable, Callable]:
  server_process = multiprocessing.Process(
  target=run_uvicorn,
@@ -204,6 +208,7 @@ def serve_node(
  "create_producer": create_producer,
  "n_consumers": n_consumers,
  "association_request_auto_approval": association_request_auto_approval,
+ "background_tasks": background_tasks,
  },
  )
 

packages/syft/src/syft/service/network/node_peer.py

@@ -71,7 +71,7 @@ class NodePeer(SyftObject):
  "name",
  "node_type",
  "admin_email",
- "ping_status.value",
+ "ping_status",
  "ping_status_message",
  "pinged_timestamp",
  ]

packages/syft/src/syft/service/network/utils.py

@@ -19,7 +19,7 @@
 
 @serializable(without=["thread"])
 class PeerHealthCheckTask:
- repeat_time = 60 # in seconds
+ repeat_time = 10 # in seconds
 
  def __init__(self) -> None:
  self.thread: threading.Thread | None = None

packages/syft/src/syft/store/sqlite_document_store.py

@@ -351,9 +351,9 @@ def __iter__(self) -> Any:
  def __del__(self) -> None:
  try:
  self._close()
- except BaseException:
- print("Could not close connection")
- pass
+ except Exception as e:
+ print(f"Could not close connection. Error: {e}")
+ raise e
 
 
 @serializable()

scripts/hagrid_hash

@@ -1 +1 @@
-bcc1cc0354932a7d2cf4f35f781fde47
+56f89d45a711a6bf79a460fc8cd4ae20