85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
from enum import Enum
|
|
|
|
from colony.core.network import advice
|
|
from colony.core.roles.role import Role
|
|
from colony.core.roles.unit.role import Unit
|
|
from colony.settings import setting
|
|
|
|
|
|
class Foreman(Role):
|
|
CONTINUOUS = True
|
|
LOOP_SECONDS_WAIT = 5
|
|
|
|
class COMMANDS(Enum):
|
|
pass
|
|
|
|
def __init__(self, context):
|
|
self.LOOP_SECONDS_WAIT = setting("foreman.interval")
|
|
super().__init__({}, context)
|
|
|
|
self.acquire_delay = setting("foreman.acquire_delay")
|
|
self.proposal_stagger = setting("foreman.proposal_stagger")
|
|
self.proposal_cooldown = setting("foreman.proposal_cooldown")
|
|
|
|
self._needed_since = {}
|
|
self._proposals = {}
|
|
|
|
def loop(self):
|
|
self.tick()
|
|
|
|
def tick(self):
|
|
membership = self.context.membership()
|
|
if membership.denied:
|
|
return None
|
|
|
|
now = membership.now()
|
|
records = membership.alive_records()
|
|
roles = self.context.get_role_registry().roles
|
|
|
|
unmet = advice.unmet_roles(records, roles)
|
|
self._needed_since = {name: self._needed_since.get(name, now) for name in unmet}
|
|
|
|
self._proposals = {
|
|
key: (at, accepted) for key, (at, accepted) in self._proposals.items()
|
|
if now - at < self.proposal_cooldown
|
|
}
|
|
|
|
alone = len(records) == 1
|
|
rank = advice.rank_of(membership.key, records)
|
|
wait = 0 if alone else self.acquire_delay + min(rank, 3) * self.proposal_stagger
|
|
|
|
for role_name in unmet:
|
|
if now - self._needed_since[role_name] < wait:
|
|
continue
|
|
|
|
recent = {key: accepted for (name, key), (_, accepted) in self._proposals.items() if name == role_name}
|
|
if any(recent.values()):
|
|
continue
|
|
|
|
target = advice.pick_candidate(role_name, roles[role_name], records, exclude=set(recent))
|
|
if target is None:
|
|
if not recent:
|
|
self.logger.warning(f"No node is willing to take role {role_name}")
|
|
continue
|
|
|
|
result = self.propose(role_name, target)
|
|
self._proposals[(role_name, target.key)] = (now, result[2] is True)
|
|
return result
|
|
|
|
return None
|
|
|
|
def propose(self, role_name, target):
|
|
self.logger.info(f"Asking {target.ip}:{target.port} to take role {role_name}")
|
|
try:
|
|
resp = self.context.execute_to_unregistered_node(
|
|
target.ip, target.port, Unit.COMMANDS.REQUEST_ROLE, role_name
|
|
)
|
|
except Exception as e:
|
|
self.logger.warning(f"Could not ask {target.ip}:{target.port} for role {role_name}: {e}")
|
|
return (role_name, target.key, None)
|
|
|
|
result = resp.get("result") if isinstance(resp, dict) and resp.get("status") == "ok" else None
|
|
accepted = result.get("accepted") if isinstance(result, dict) else None
|
|
self.logger.info(f"{target.ip}:{target.port} answered {role_name}: {result}")
|
|
return (role_name, target.key, accepted)
|