264 lines
9.6 KiB
Python
264 lines
9.6 KiB
Python
import random
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from enum import Enum
|
|
|
|
from colony.core.hardware_scores import collect_hardware
|
|
from colony.core.network import advice
|
|
from colony.core.network.membership import MemberRecord
|
|
from colony.core.roles.role import Role, Command
|
|
from colony.core.state_manager import StateManager
|
|
from colony.settings import setting
|
|
from colony.transport.discover import Discover
|
|
from colony.utils.network import hosts_in_range
|
|
|
|
ADMITTED = "admitted"
|
|
REFUSED = "refused"
|
|
FAILED = "failed"
|
|
|
|
JOINED = "joined"
|
|
BOOTSTRAPPED = "bootstrapped"
|
|
DENIED = "denied"
|
|
|
|
|
|
class Gossip(Role):
|
|
CONTINUOUS = True
|
|
LOOP_SECONDS_WAIT = 5
|
|
|
|
BOOTSTRAP_REPROBE = 5
|
|
DENIED_BACKOFF_CAP = 60
|
|
HARDWARE_TTL = 5
|
|
MAX_SEEDS = 64
|
|
|
|
class COMMANDS(Enum):
|
|
HELLO = 'Gossip.Hello'
|
|
|
|
def __init__(self, context):
|
|
commands = {
|
|
self.COMMANDS.HELLO.value: Command(self.hello),
|
|
}
|
|
|
|
self.LOOP_SECONDS_WAIT = setting("membership.gossip_interval")
|
|
super().__init__(commands, context)
|
|
|
|
self.reprobe_interval = setting("membership.reprobe_interval")
|
|
self.fanout = setting("membership.fanout")
|
|
self.hello_timeout = setting("membership.hello_timeout")
|
|
self.max_members = setting("membership.max_members")
|
|
self.discover_port = setting("discovery.discover_port")
|
|
self.probe_timeout = setting("discovery.probe_timeout")
|
|
self.probe_workers = setting("discovery.probe_workers")
|
|
self.range = (setting("discovery.range_start"), setting("discovery.range_end"))
|
|
|
|
self.random = random.Random()
|
|
self.next_reprobe = 0.0
|
|
self.denied_attempts = 0
|
|
self._scores = {}
|
|
self._stats = {}
|
|
self._hardware_at = None
|
|
self._persisted_peers = None
|
|
|
|
@property
|
|
def membership(self):
|
|
return self.context.membership()
|
|
|
|
def loop(self):
|
|
self.tick()
|
|
|
|
def hello(self, sender, known):
|
|
record = MemberRecord.from_dict(sender)
|
|
if record.key == self.membership.key:
|
|
return {"admitted": False, "members": [], "self": True}
|
|
if not self.membership.admit(record):
|
|
return {"admitted": False, "members": []}
|
|
|
|
records = self._parse_records(known)
|
|
self.membership.merge([record] + records)
|
|
return {"admitted": True, "members": self.membership.export(), "responder": self.membership.self_record().to_dict()}
|
|
|
|
def _parse_records(self, raw):
|
|
if not isinstance(raw, list) or len(raw) > self.max_members:
|
|
raise ValueError("Member list must be a list of bounded size")
|
|
return [MemberRecord.from_dict(item) for item in raw]
|
|
|
|
def scan_targets(self):
|
|
own_ip = self.context.self().ip
|
|
return [(host, self.discover_port) for host in hosts_in_range(*self.range, exclude=[own_ip])]
|
|
|
|
def discover_target(self, ip, port):
|
|
return (ip, self.discover_port)
|
|
|
|
def seed_addresses(self):
|
|
state = self.context.get_role_manager().state_manager
|
|
seeds = []
|
|
for entry in state.get(StateManager.Data.Peers):
|
|
if isinstance(entry, list) and len(entry) == 2 and isinstance(entry[0], str) and isinstance(entry[1], int):
|
|
seeds.append((entry[0], entry[1]))
|
|
return seeds
|
|
|
|
def probe_targets(self, alive):
|
|
remembered = (set(self.membership.known_addresses()) | set(self.seed_addresses())) - alive
|
|
|
|
targets = dict.fromkeys(self.scan_targets())
|
|
for ip, port in sorted(remembered):
|
|
targets[self.discover_target(ip, port)] = None
|
|
return list(targets)
|
|
|
|
def _probe(self, target):
|
|
host, discover_port = target
|
|
try:
|
|
return host, Discover.request(host, timeout=self.probe_timeout, discover_port=discover_port)
|
|
except Exception:
|
|
return host, None
|
|
|
|
def scan(self, targets):
|
|
if not targets:
|
|
return []
|
|
with ThreadPoolExecutor(max_workers=min(self.probe_workers, len(targets))) as pool:
|
|
results = list(pool.map(self._probe, targets))
|
|
return [(host, port) for host, port in results if port is not None]
|
|
|
|
def find_peers(self):
|
|
membership = self.membership
|
|
alive = {record.key for record in membership.alive_records()}
|
|
found = self.scan(self.probe_targets(alive))
|
|
return [address for address in dict.fromkeys(found) if address not in alive]
|
|
|
|
def exchange(self, ip, port):
|
|
membership = self.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, self.COMMANDS.HELLO, sender, known, timeout=self.hello_timeout
|
|
)
|
|
except Exception as e:
|
|
self.logger.debug(f"Hello to {ip}:{port} failed: {e}")
|
|
return FAILED
|
|
|
|
result = resp.get("result") if isinstance(resp, dict) and resp.get("status") == "ok" else None
|
|
if not isinstance(result, dict):
|
|
self.logger.debug(f"Hello to {ip}:{port} was answered with an error")
|
|
return FAILED
|
|
if result.get("self"):
|
|
self.logger.debug(f"{ip}:{port} is this node")
|
|
return FAILED
|
|
if not result.get("admitted"):
|
|
self.logger.warning(f"{ip}:{port} refused to admit this node")
|
|
return REFUSED
|
|
|
|
try:
|
|
membership.merge(self._parse_records(result.get("members", [])))
|
|
except ValueError as e:
|
|
self.logger.warning(f"Ignoring bad member list from {ip}:{port}: {e}")
|
|
return FAILED
|
|
|
|
self.remember_dial_address(result.get("responder"), (ip, port))
|
|
return ADMITTED
|
|
|
|
def remember_dial_address(self, responder, address):
|
|
if not isinstance(responder, dict):
|
|
return
|
|
try:
|
|
key = MemberRecord.from_dict(responder).key
|
|
except ValueError:
|
|
return
|
|
if key != address:
|
|
self.context.set_dial_alias(key, address)
|
|
|
|
def hardware(self):
|
|
now = self.membership.now()
|
|
if self._hardware_at is None or now - self._hardware_at >= self.HARDWARE_TTL:
|
|
self._scores, self._stats = self.collect_hardware()
|
|
if self._scores:
|
|
self._hardware_at = now
|
|
return self._scores, self._stats
|
|
|
|
def collect_hardware(self):
|
|
return collect_hardware()
|
|
|
|
def refresh_self(self):
|
|
scores, stats = self.hardware()
|
|
self.membership.update_self(scores=scores, stats=stats)
|
|
|
|
def exchange_all(self, addresses):
|
|
if not addresses:
|
|
return []
|
|
with ThreadPoolExecutor(max_workers=min(self.probe_workers, len(addresses))) as pool:
|
|
return list(pool.map(lambda address: self.exchange(*address), addresses))
|
|
|
|
def _after_probe(self, results):
|
|
membership = self.membership
|
|
has_peers = any(record.key != membership.key for record in membership.alive_records())
|
|
|
|
if ADMITTED in results or has_peers:
|
|
outcome, delay = JOINED, self.reprobe_interval
|
|
membership.denied = False
|
|
self.denied_attempts = 0
|
|
elif REFUSED in results:
|
|
self.denied_attempts += 1
|
|
outcome = DENIED
|
|
delay = min(self.BOOTSTRAP_REPROBE * 2 ** self.denied_attempts, self.DENIED_BACKOFF_CAP)
|
|
membership.denied = True
|
|
else:
|
|
outcome, delay = BOOTSTRAPPED, self.BOOTSTRAP_REPROBE
|
|
membership.denied = False
|
|
|
|
self.next_reprobe = membership.now() + delay
|
|
return outcome
|
|
|
|
def join_or_bootstrap(self):
|
|
self.refresh_self()
|
|
results = self.exchange_all(self.find_peers())
|
|
outcome = self._after_probe(results)
|
|
self.logger.info(f"Startup probe finished: {outcome}")
|
|
return outcome
|
|
|
|
def reprobe(self):
|
|
results = self.exchange_all(self.find_peers())
|
|
self._after_probe(results)
|
|
|
|
def tick(self):
|
|
membership = self.membership
|
|
self.refresh_self()
|
|
|
|
peers = [record for record in membership.alive_records() if record.key != membership.key]
|
|
chosen = self.random.sample(peers, min(self.fanout, len(peers)))
|
|
self.exchange_all([record.key for record in chosen])
|
|
|
|
if membership.now() >= self.next_reprobe:
|
|
self.reprobe()
|
|
|
|
self.persist_peers()
|
|
self.release_roles()
|
|
|
|
def persist_peers(self):
|
|
known = self.membership.known_addresses()
|
|
seeds = [address for address in self.seed_addresses() if address not in known]
|
|
addresses = (known + seeds)[:self.MAX_SEEDS]
|
|
if addresses == self._persisted_peers:
|
|
return
|
|
state = self.context.get_role_manager().state_manager
|
|
state.set(StateManager.Data.Peers, [list(address) for address in addresses])
|
|
self._persisted_peers = addresses
|
|
|
|
def release_roles(self):
|
|
manager = self.context.get_role_manager()
|
|
membership = self.membership
|
|
roles = self.context.get_role_registry().roles
|
|
policy = manager.role_policy
|
|
|
|
held = list(manager.get_role_instances())
|
|
scores = membership.self_record().scores
|
|
assignable = advice.assignable_roles(roles)
|
|
forbidden = [name for name in held if name in assignable and not policy.should_keep(name, scores)]
|
|
over_held = advice.roles_to_release(membership.key, membership.alive_records(), roles)
|
|
|
|
for name in forbidden:
|
|
membership.decline(name, policy.decline_cooldown)
|
|
|
|
drop = set(forbidden) | set(over_held)
|
|
if drop:
|
|
self.logger.info(f"Releasing roles {sorted(drop)}")
|
|
manager.drop_roles(drop)
|