Explorer + cli

colony/core/roles/explorer/role.py

@@ -2,7 +2,9 @@ import datetime
 from concurrent.futures import ThreadPoolExecutor
 from enum import Enum
 
+from colony.core.network.membership import MemberRecord
 from colony.core.storage import DEFAULT_STORAGE_FILES
+from colony.core.roles.gossip.role import Gossip
 from colony.core.roles.role import Role
 from colony.core.roles.unit.role import Unit
 from colony.core.roles.vaultwarden.role import VaultWarden
@@ -66,6 +68,7 @@ class Explorer(Role):
             return
 
         self.save_snapshot(snapshot)
+        self.introduce_unknown_peers(discovered_peers)
 
     def load_snapshot(self):
         try:
@@ -137,3 +140,81 @@ class Explorer(Role):
             self.logger.warning(f"Peer {ip}:{port} answered SHOW_STATS with an error")
             return None
         return resp['result']
+
+    def fetch_members_from_peer(self, ip, port):
+        try:
+            resp = self.context.execute_to_unregistered_node(ip, port, Unit.COMMANDS.SHOW_CLUSTER)
+        except Exception as e:
+            self.logger.warning(f"Could not fetch cluster members from {ip}:{port}: {e}")
+            return []
+
+        result = resp.get('result') if isinstance(resp, dict) and resp.get('status') == 'ok' else None
+        rows = result.get('members') if isinstance(result, dict) else None
+        if not isinstance(rows, list):
+            self.logger.warning(f"Peer {ip}:{port} answered SHOW_CLUSTER with an error")
+            return []
+
+        members = []
+        for row in rows[:setting("membership.max_members")]:
+            try:
+                record = MemberRecord.from_dict(row)
+            except ValueError as e:
+                self.logger.debug(f"Ignoring bad member from {ip}:{port}: {e}")
+                continue
+            if row.get('alive') is True:
+                members.append(record.key)
+        return members
+
+    def _ping(self, address):
+        ip, _ = address
+        try:
+            port = Discover.request(ip, timeout=self.SCAN_TIMEOUT, discover_port=setting("discovery.discover_port"))
+        except Exception:
+            return None
+        return (ip, port) if port is not None else None
+
+    def introduce_unknown_peers(self, discovered_peers):
+        membership = self.context.membership()
+        known = {record.key for record in membership.alive_records()} | set(discovered_peers.items())
+
+        candidates = {}
+        for ip, port in discovered_peers.items():
+            for address in self.fetch_members_from_peer(ip, port):
+                if address not in known:
+                    candidates[address] = None
+        if not candidates or self._stop_event.is_set():
+            return
+
+        addresses = list(candidates)[:setting("membership.max_members")]
+        with ThreadPoolExecutor(max_workers=min(self.SCAN_WORKERS, len(addresses))) as pool:
+            pinged = [address for address in pool.map(self._ping, addresses) if address is not None]
+
+        for ip, port in pinged:
+            self.logger.info(f"Peers know {ip}:{port}, asking it to become our peer")
+            self.send_hello(ip, port)
+
+    def send_hello(self, ip, port):
+        membership = self.context.membership()
+        sender = membership.self_record().to_dict()
+        known = [record.to_dict() for record in membership.alive_records() if record.key != membership.key]
+
+        try:
+            resp = self.context.execute_to_unregistered_node(
+                ip, port, Gossip.COMMANDS.HELLO, sender, known, timeout=setting("membership.hello_timeout")
+            )
+        except Exception as e:
+            self.logger.warning(f"Hello to {ip}:{port} failed: {e}")
+            return False
+
+        result = resp.get('result') if isinstance(resp, dict) and resp.get('status') == 'ok' else None
+        if not isinstance(result, dict) or not result.get('admitted'):
+            self.logger.warning(f"{ip}:{port} did not admit this node")
+            return False
+
+        try:
+            members = [MemberRecord.from_dict(item) for item in result.get('members', [])]
+        except (ValueError, TypeError) as e:
+            self.logger.warning(f"Ignoring bad member list from {ip}:{port}: {e}")
+            return False
+        membership.merge(members[:setting("membership.max_members")])
+        return True