ciomek/colony
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixes + docker

Browse Source
This commit is contained in:
ciomek committed 2026-09-19 20:57:32 +02:00
1 parent b2bd4d25f9
commit 06a1e360e1
22 files changed
+175 -296
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.dockerignore
+14
View File
@@ -0,0 +1,14 @@
.git
.gitea
.idea
.claude
.venv
__pycache__
*.pyc
docs
tests
logs
storage
state.json
state.json.corrupt
.tmp-*.json
.gitea/workflows/docker.yml
+25
View File
@@ -0,0 +1,25 @@
name: Build and Push Docker Image
on:
push:
branches:
- main
jobs:
build-and-push:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Log in to private registry
run: echo "${{ secrets.REGISTRY_PASSWORD }}" | docker login registry.ciomek.pl -u "${{ secrets.REGISTRY_USERNAME }}" --password-stdin
- name: Build Docker image
run: docker build -t registry.ciomek.pl/colony:latest .
- name: Push Docker image
run: docker push registry.ciomek.pl/colony:latest
.gitignore
+2
View File
@@ -2,6 +2,8 @@
logs logs
storage storage
state.json state.json
state.json.corrupt
.tmp-*.json
# Created by https://www.toptal.com/developers/gitignore/api/python # Created by https://www.toptal.com/developers/gitignore/api/python
# Edit at https://www.toptal.com/developers/gitignore?templates=python # Edit at https://www.toptal.com/developers/gitignore?templates=python
.idea/.gitignore
Generated
-5
View File
@@ -1,5 +0,0 @@
# Default ignored files
/shelf/
/workspace.xml
# Editor-based HTTP Client requests
/httpRequests/
.idea/colony.iml
Generated
-11
View File
@@ -1,11 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$" isTestSource="false" />
<excludeFolder url="file://$MODULE_DIR$/.venv" />
</content>
<orderEntry type="jdk" jdkName="Python 3.13 (colony)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>
.idea/inspectionProfiles/profiles_settings.xml
Generated
-6
View File
@@ -1,6 +0,0 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>
</component>
.idea/misc.xml
Generated
-7
View File
@@ -1,7 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="Python 3.13 (colony)" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.13 (colony)" project-jdk-type="Python SDK" />
</project>
.idea/modules.xml
Generated
-8
View File
@@ -1,8 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/colony.iml" filepath="$PROJECT_DIR$/.idea/colony.iml" />
</modules>
</component>
</project>
.idea/vcs.xml
Generated
-6
View File
@@ -1,6 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>
.idea/workspace.sync-conflict-20260111-130327-UFTLO5A.xml
Generated
-90
View File
@@ -1,90 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ChangeListManager">
<list default="true" id="a5f8e4e7-da52-40b9-ac09-6eedaaa69b69" name="Changes" comment="" />
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
<option name="LAST_RESOLUTION" value="IGNORE" />
</component>
<component name="FileTemplateManagerImpl">
<option name="RECENT_TEMPLATES">
<list>
<option value="Python Script" />
</list>
</option>
</component>
<component name="ProjectColorInfo">{
&quot;associatedIndex&quot;: 3
}</component>
<component name="ProjectId" id="37rDRQ5zG8SbPZaUPDbrYGJVv81" />
<component name="ProjectViewState">
<option name="hideEmptyMiddlePackages" value="true" />
<option name="showLibraryContents" value="true" />
</component>
<component name="PropertiesComponent"><![CDATA[{
"keyToString": {
"ModuleVcsDetector.initialDetectionPerformed": "true",
"Python.main.executor": "Debug",
"RunOnceActivity.ShowReadmeOnStart": "true",
"RunOnceActivity.TerminalTabsStorage.copyFrom.TerminalArrangementManager.252": "true",
"last_opened_file_path": "/home/ciomek/Repositories/colony"
}
}]]></component>
<component name="RecentsManager">
<key name="MoveFile.RECENT_KEYS">
<recent name="$PROJECT_DIR$/colony/core" />
<recent name="$PROJECT_DIR$/colony/core/storage" />
<recent name="$PROJECT_DIR$/storage" />
<recent name="$PROJECT_DIR$/src/domain" />
<recent name="$PROJECT_DIR$/src/core" />
</key>
</component>
<component name="RunManager">
<configuration name="main" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
<module name="colony" />
<option name="ENV_FILES" value="" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="RUN_TOOL" value="" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/main.py" />
<option name="PARAMETERS" value="--role Explorer --listen --discover" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<recent_temporary>
<list>
<item itemvalue="Python.main" />
</list>
</recent_temporary>
</component>
<component name="SharedIndexes">
<attachedChunks>
<set>
<option value="bundled-python-sdk-f2b7a9f6281b-6e1f45a539f7-com.jetbrains.pycharm.pro.sharedIndexes.bundled-PY-253.29346.142" />
</set>
</attachedChunks>
</component>
<component name="TaskManager">
<task active="true" id="Default" summary="Default task">
<changelist id="a5f8e4e7-da52-40b9-ac09-6eedaaa69b69" name="Changes" comment="" />
<created>1767652287935</created>
<option name="number" value="Default" />
<option name="presentableId" value="Default" />
<updated>1767652287935</updated>
</task>
<servers />
</component>
</project>
.idea/workspace.sync-conflict-20260111-130328-UFTLO5A.xml
Generated
-90
View File
@@ -1,90 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ChangeListManager">
<list default="true" id="a5f8e4e7-da52-40b9-ac09-6eedaaa69b69" name="Changes" comment="" />
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
<option name="LAST_RESOLUTION" value="IGNORE" />
</component>
<component name="FileTemplateManagerImpl">
<option name="RECENT_TEMPLATES">
<list>
<option value="Python Script" />
</list>
</option>
</component>
<component name="ProjectColorInfo">{
&quot;associatedIndex&quot;: 3
}</component>
<component name="ProjectId" id="37rDRQ5zG8SbPZaUPDbrYGJVv81" />
<component name="ProjectViewState">
<option name="hideEmptyMiddlePackages" value="true" />
<option name="showLibraryContents" value="true" />
</component>
<component name="PropertiesComponent"><![CDATA[{
"keyToString": {
"ModuleVcsDetector.initialDetectionPerformed": "true",
"Python.main.executor": "Debug",
"RunOnceActivity.ShowReadmeOnStart": "true",
"RunOnceActivity.TerminalTabsStorage.copyFrom.TerminalArrangementManager.252": "true",
"last_opened_file_path": "/home/ciomek/Repositories/colony"
}
}]]></component>
<component name="RecentsManager">
<key name="MoveFile.RECENT_KEYS">
<recent name="$PROJECT_DIR$/colony/core" />
<recent name="$PROJECT_DIR$/colony/core/storage" />
<recent name="$PROJECT_DIR$/storage" />
<recent name="$PROJECT_DIR$/src/domain" />
<recent name="$PROJECT_DIR$/src/core" />
</key>
</component>
<component name="RunManager">
<configuration name="main" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
<module name="colony" />
<option name="ENV_FILES" value="" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="RUN_TOOL" value="" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/main.py" />
<option name="PARAMETERS" value="--role Explorer --listen --discover" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<recent_temporary>
<list>
<item itemvalue="Python.main" />
</list>
</recent_temporary>
</component>
<component name="SharedIndexes">
<attachedChunks>
<set>
<option value="bundled-python-sdk-f2b7a9f6281b-6e1f45a539f7-com.jetbrains.pycharm.pro.sharedIndexes.bundled-PY-253.29346.142" />
</set>
</attachedChunks>
</component>
<component name="TaskManager">
<task active="true" id="Default" summary="Default task">
<changelist id="a5f8e4e7-da52-40b9-ac09-6eedaaa69b69" name="Changes" comment="" />
<created>1767652287935</created>
<option name="number" value="Default" />
<option name="presentableId" value="Default" />
<updated>1767652287935</updated>
</task>
<servers />
</component>
</project>
Dockerfile
+18
View File
@@ -0,0 +1,18 @@
FROM python:3.14-slim
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Runtime state lives in the working directory: state.json, storage/, logs/
VOLUME ["/app/storage", "/app/logs"]
EXPOSE 65334/tcp 28149/udp
CMD ["python", "main.py"]
colony/core/network/core.py
+3 -2
View File
@@ -65,10 +65,11 @@ class Network:
@staticmethod @staticmethod
def execute_to_unregistered_node(ip, port, command: Enum, *args, **kwargs): def execute_to_unregistered_node(ip, port, command: Enum, *args, timeout=None, **kwargs):
request = CommandRequest.form(command, *args, **kwargs) request = CommandRequest.form(command, *args, **kwargs)
return Whisperer(ip, port).send(request.collapse()) options = {} if timeout is None else {"timeout": timeout}
return Whisperer(ip, port).send(request.collapse(), **options)
def load_network_from_snapshot(self, file_path: str = "storage/colony_snapshot.json"): def load_network_from_snapshot(self, file_path: str = "storage/colony_snapshot.json"):
colony/core/network/membership.py
+24 -18
View File
@@ -1,5 +1,6 @@
import copy import copy
import ipaddress import ipaddress
import math
import threading import threading
import time import time
from dataclasses import dataclass, field, asdict from dataclasses import dataclass, field, asdict
@@ -19,7 +20,12 @@ def _is_int(value):
def _is_number(value): def _is_number(value):
return isinstance(value, (int, float)) and not isinstance(value, bool) if isinstance(value, bool) or not isinstance(value, (int, float)):
return False
try:
return math.isfinite(value)
except OverflowError:
return False
def _text(value, name): def _text(value, name):
@@ -108,16 +114,15 @@ class Membership:
admission=None, clock=time.monotonic, wall=time.time): admission=None, clock=time.monotonic, wall=time.time):
self.clock = clock self.clock = clock
self.wall = wall self.wall = wall
self.member_timeout = setting("membership.member_timeout") if member_timeout is None else member_timeout self.member_timeout = setting("membership.member_timeout", member_timeout)
self.forget_after = setting("membership.forget_after") if forget_after is None else forget_after self.forget_after = setting("membership.forget_after", forget_after)
self.max_members = setting("membership.max_members") if max_members is None else max_members self.max_members = setting("membership.max_members", max_members)
self._admission = resolve_admission(setting("membership.admission") if admission is None else admission) self._admission = resolve_admission(setting("membership.admission", admission))
self.denied = False self.denied = False
self._lock = threading.RLock() self._lock = threading.RLock()
self._self = MemberRecord(ip, port, hostname, incarnation=int(wall() * 1000), seq=1) self._self = MemberRecord(ip, port, hostname, incarnation=int(wall() * 1000), seq=1)
self._records = {} self._entries = {}
self._advanced = {}
self._declines = {} self._declines = {}
@property @property
@@ -172,35 +177,36 @@ class Membership:
now = self.clock() now = self.clock()
for record in records: for record in records:
if record.key == self.key: if record.key == self.key:
if record.version > self._self.version:
self._self.incarnation = record.incarnation
self._self.seq = record.seq + 1
continue continue
current = self._records.get(record.key) current = self._entries.get(record.key)
if current is not None and record.version <= current.version: if current is not None and record.version <= current[0].version:
continue continue
if current is None and len(self._records) >= self.max_members: if current is None and len(self._entries) >= self.max_members:
continue continue
self._records[record.key] = record self._entries[record.key] = (record, now)
self._advanced[record.key] = now
changed += 1 changed += 1
self._purge(now) self._purge(now)
return changed return changed
def _purge(self, now): def _purge(self, now):
for key in [key for key, advanced in self._advanced.items() if now - advanced > self.forget_after]: for key in [key for key, (_, advanced) in self._entries.items() if now - advanced > self.forget_after]:
del self._records[key] del self._entries[key]
del self._advanced[key]
def alive_records(self): def alive_records(self):
with self._lock: with self._lock:
now = self.clock() now = self.clock()
alive = [self._current_self()] alive = [self._current_self()]
alive += [record for key, record in self._records.items() alive += [record for record, advanced in self._entries.values()
if now - self._advanced[key] <= self.member_timeout] if now - advanced <= self.member_timeout]
return sorted(alive, key=lambda record: record.key) return sorted(alive, key=lambda record: record.key)
def known_addresses(self): def known_addresses(self):
with self._lock: with self._lock:
self._purge(self.clock()) self._purge(self.clock())
return sorted(self._records) return sorted(self._entries)
def export(self): def export(self):
return [record.to_dict() for record in self.alive_records()] return [record.to_dict() for record in self.alive_records()]
colony/core/roles/gossip/role.py
+31 -24
View File
@@ -27,6 +27,7 @@ class Gossip(Role):
BOOTSTRAP_REPROBE = 5 BOOTSTRAP_REPROBE = 5
DENIED_BACKOFF_CAP = 60 DENIED_BACKOFF_CAP = 60
SCORES_TTL = 30 SCORES_TTL = 30
MAX_SEEDS = 64
class COMMANDS(Enum): class COMMANDS(Enum):
HELLO = 'Gossip.Hello' HELLO = 'Gossip.Hello'
@@ -41,6 +42,7 @@ class Gossip(Role):
self.reprobe_interval = setting("membership.reprobe_interval") self.reprobe_interval = setting("membership.reprobe_interval")
self.fanout = setting("membership.fanout") self.fanout = setting("membership.fanout")
self.hello_timeout = setting("membership.hello_timeout")
self.max_members = setting("membership.max_members") self.max_members = setting("membership.max_members")
self.discover_port = setting("discovery.discover_port") self.discover_port = setting("discovery.discover_port")
self.probe_timeout = setting("discovery.probe_timeout") self.probe_timeout = setting("discovery.probe_timeout")
@@ -50,7 +52,6 @@ class Gossip(Role):
self.random = random.Random() self.random = random.Random()
self.next_reprobe = 0.0 self.next_reprobe = 0.0
self.denied_attempts = 0 self.denied_attempts = 0
self.outcome = None
self._scores = {} self._scores = {}
self._scores_at = None self._scores_at = None
self._persisted_peers = None self._persisted_peers = None
@@ -91,17 +92,13 @@ class Gossip(Role):
seeds.append((entry[0], entry[1])) seeds.append((entry[0], entry[1]))
return seeds return seeds
def probe_targets(self): def probe_targets(self, alive):
membership = self.membership remembered = (set(self.membership.known_addresses()) | set(self.seed_addresses())) - alive
alive = {record.key for record in membership.alive_records()}
remembered = (set(membership.known_addresses()) | set(self.seed_addresses())) - alive
targets = list(self.scan_targets()) targets = dict.fromkeys(self.scan_targets())
for ip, port in sorted(remembered): for ip, port in sorted(remembered):
target = self.discover_target(ip, port) targets[self.discover_target(ip, port)] = None
if target not in targets: return list(targets)
targets.append(target)
return targets
def _probe(self, target): def _probe(self, target):
host, discover_port = target host, discover_port = target
@@ -120,7 +117,7 @@ class Gossip(Role):
def find_peers(self): def find_peers(self):
membership = self.membership membership = self.membership
alive = {record.key for record in membership.alive_records()} alive = {record.key for record in membership.alive_records()}
found = self.scan(self.probe_targets()) found = self.scan(self.probe_targets(alive))
return [address for address in dict.fromkeys(found) if address not in alive] return [address for address in dict.fromkeys(found) if address not in alive]
def exchange(self, ip, port): def exchange(self, ip, port):
@@ -129,7 +126,9 @@ class Gossip(Role):
known = [record.to_dict() for record in membership.alive_records() if record.key != membership.key] known = [record.to_dict() for record in membership.alive_records() if record.key != membership.key]
try: try:
resp = self.context.execute_to_unregistered_node(ip, port, self.COMMANDS.HELLO, sender, known) resp = self.context.execute_to_unregistered_node(
ip, port, self.COMMANDS.HELLO, sender, known, timeout=self.hello_timeout
)
except Exception as e: except Exception as e:
self.logger.debug(f"Hello to {ip}:{port} failed: {e}") self.logger.debug(f"Hello to {ip}:{port} failed: {e}")
return FAILED return FAILED
@@ -153,6 +152,7 @@ class Gossip(Role):
now = self.membership.now() now = self.membership.now()
if self._scores_at is None or now - self._scores_at >= self.SCORES_TTL: if self._scores_at is None or now - self._scores_at >= self.SCORES_TTL:
self._scores = self.collect_scores() self._scores = self.collect_scores()
if self._scores:
self._scores_at = now self._scores_at = now
return self._scores return self._scores
@@ -160,9 +160,15 @@ class Gossip(Role):
return collect_scores() return collect_scores()
def refresh_self(self): def refresh_self(self):
self.membership.update_self(roles=list(self.context.self().roles), scores=self.scores()) self.membership.update_self(scores=self.scores())
def _after_probe(self, results, now): 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 membership = self.membership
has_peers = any(record.key != membership.key for record in membership.alive_records()) has_peers = any(record.key != membership.key for record in membership.alive_records())
@@ -179,28 +185,27 @@ class Gossip(Role):
outcome, delay = BOOTSTRAPPED, self.BOOTSTRAP_REPROBE outcome, delay = BOOTSTRAPPED, self.BOOTSTRAP_REPROBE
membership.denied = False membership.denied = False
self.next_reprobe = now + delay self.next_reprobe = membership.now() + delay
self.outcome = outcome
return outcome return outcome
def join_or_bootstrap(self): def join_or_bootstrap(self):
self.refresh_self() self.refresh_self()
results = [self.exchange(ip, port) for ip, port in self.find_peers()] results = self.exchange_all(self.find_peers())
outcome = self._after_probe(results, self.membership.now()) outcome = self._after_probe(results)
self.logger.info(f"Startup probe finished: {outcome}") self.logger.info(f"Startup probe finished: {outcome}")
return outcome return outcome
def reprobe(self): def reprobe(self):
results = [self.exchange(ip, port) for ip, port in self.find_peers()] results = self.exchange_all(self.find_peers())
self._after_probe(results, self.membership.now()) self._after_probe(results)
def tick(self): def tick(self):
membership = self.membership membership = self.membership
self.refresh_self() self.refresh_self()
peers = [record for record in membership.alive_records() if record.key != membership.key] peers = [record for record in membership.alive_records() if record.key != membership.key]
for record in self.random.sample(peers, min(self.fanout, len(peers))): chosen = self.random.sample(peers, min(self.fanout, len(peers)))
self.exchange(*record.key) self.exchange_all([record.key for record in chosen])
if membership.now() >= self.next_reprobe: if membership.now() >= self.next_reprobe:
self.reprobe() self.reprobe()
@@ -209,7 +214,9 @@ class Gossip(Role):
self.release_roles() self.release_roles()
def persist_peers(self): def persist_peers(self):
addresses = self.membership.known_addresses() 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: if addresses == self._persisted_peers:
return return
state = self.context.get_role_manager().state_manager state = self.context.get_role_manager().state_manager
@@ -234,4 +241,4 @@ class Gossip(Role):
drop = set(forbidden) | set(over_held) drop = set(forbidden) | set(over_held)
if drop: if drop:
self.logger.info(f"Releasing roles {sorted(drop)}") self.logger.info(f"Releasing roles {sorted(drop)}")
manager.set_roles([name for name in held if name not in drop]) manager.drop_roles(drop)
colony/core/roles/manager.py
+28 -12
View File
@@ -1,9 +1,9 @@
import threading import threading
from colony.core.context import RoleContext from colony.core.context import RoleContext
from colony.core.network.advice import assignable_roles
from colony.core.roles.policy import DefaultRolePolicy from colony.core.roles.policy import DefaultRolePolicy
from colony.core.state_manager import StateManager from colony.core.state_manager import StateManager
from colony.role_registry import RolePriority
from colony.core.logger.logging import get_logger from colony.core.logger.logging import get_logger
@@ -18,10 +18,8 @@ class RoleManager:
self._lock = threading.RLock() self._lock = threading.RLock()
def ensure_general_roles(self): def ensure_general_roles(self):
general = sorted( roles = self.role_registry.roles
name for name, stats in self.role_registry.roles.items() general = sorted(set(roles) - set(assignable_roles(roles)))
if stats.priority == RolePriority.GENERAL
)
current = self.state_manager.get(StateManager.Data.Roles) current = self.state_manager.get(StateManager.Data.Roles)
wanted = list(dict.fromkeys(general + current)) wanted = list(dict.fromkeys(general + current))
if wanted != current: if wanted != current:
@@ -42,11 +40,14 @@ class RoleManager:
self.logger.debug(f"Creating instance of role {role_name}. ") self.logger.debug(f"Creating instance of role {role_name}. ")
roles[role_name] = role_cls(context=context) roles[role_name] = role_cls(context=context)
self.network.self_node.roles = list(roles) self._publish(roles)
self.network.self_node.set_role_instances(roles)
self.network.membership.update_self(roles=list(roles))
self.logger.info(f"Roles initialized: {list(roles.keys())}. ") self.logger.info(f"Roles initialized: {list(roles.keys())}. ")
def _publish(self, roles):
self.network.self_node.set_role_instances(roles)
self.network.self_node.roles = list(roles)
self.network.membership.update_self(roles=list(roles))
def get_role_instances(self): def get_role_instances(self):
return self.network.self_node.get_role_instances() return self.network.self_node.get_role_instances()
@@ -92,9 +93,7 @@ class RoleManager:
} }
self.state_manager.set(StateManager.Data.Roles, list(final_roles.keys())) self.state_manager.set(StateManager.Data.Roles, list(final_roles.keys()))
self.network.self_node.set_role_instances(final_roles) self._publish(final_roles)
self.network.self_node.roles = list(final_roles.keys())
self.network.membership.update_self(roles=list(final_roles))
for role_name in roles_to_remove: for role_name in roles_to_remove:
self.logger.debug(f"Stopping removed role {role_name}") self.logger.debug(f"Stopping removed role {role_name}")
@@ -106,6 +105,17 @@ class RoleManager:
self.logger.info(f"Roles updated to: {list(final_roles.keys())}") self.logger.info(f"Roles updated to: {list(final_roles.keys())}")
def drop_roles(self, role_names):
with self._lock:
assignable = assignable_roles(self.role_registry.roles)
held = list(self.get_role_instances())
remaining = [name for name in held if name not in role_names or name not in assignable]
if remaining != held:
self.set_roles(remaining)
def _decline(self, role_name):
self.network.membership.decline(role_name, self.role_policy.decline_cooldown)
def request_role(self, role_name): def request_role(self, role_name):
membership = self.network.membership membership = self.network.membership
@@ -124,9 +134,15 @@ class RoleManager:
accepted, reason = self.role_policy.decide(role_name, held, scores, self.role_registry.roles) accepted, reason = self.role_policy.decide(role_name, held, scores, self.role_registry.roles)
if not accepted: if not accepted:
self.logger.info(f"Declining role {role_name}: {reason}") self.logger.info(f"Declining role {role_name}: {reason}")
membership.decline(role_name, self.role_policy.decline_cooldown) self._decline(role_name)
return {"accepted": False, "reason": reason} return {"accepted": False, "reason": reason}
try:
self.set_roles(held + [role_name]) self.set_roles(held + [role_name])
except Exception:
self.logger.exception(f"Could not take role {role_name}")
self._decline(role_name)
return {"accepted": False, "reason": "role could not be started"}
membership.clear_decline(role_name) membership.clear_decline(role_name)
return {"accepted": True, "reason": reason} return {"accepted": True, "reason": reason}
colony/core/roles/policy.py
+8 -6
View File
@@ -1,4 +1,4 @@
from colony.role_registry import RolePriority from colony.core.network.advice import assignable_roles
from colony.settings import setting from colony.settings import setting
@@ -14,11 +14,11 @@ class RolePolicy:
class DefaultRolePolicy(RolePolicy): class DefaultRolePolicy(RolePolicy):
def __init__(self, max_roles=None, deny=None, min_scores=None, decline_cooldown=None): def __init__(self, max_roles=None, deny=None, min_scores=None, decline_cooldown=None):
self.max_roles = setting("role_policy.max_roles") if max_roles is None else max_roles self.max_roles = setting("role_policy.max_roles", max_roles)
self.deny = set(setting("role_policy.deny") if deny is None else deny) self.deny = set(setting("role_policy.deny", deny))
raw_minimums = setting("role_policy.min_scores") if min_scores is None else min_scores raw_minimums = setting("role_policy.min_scores", min_scores)
self.min_scores = {role: dict(minimums) for role, minimums in dict(raw_minimums).items()} self.min_scores = {role: dict(minimums) for role, minimums in dict(raw_minimums).items()}
self.decline_cooldown = setting("role_policy.decline_cooldown") if decline_cooldown is None else decline_cooldown self.decline_cooldown = setting("role_policy.decline_cooldown", decline_cooldown)
def _violation(self, role_name, scores): def _violation(self, role_name, scores):
if role_name in self.deny: if role_name in self.deny:
@@ -34,11 +34,13 @@ class DefaultRolePolicy(RolePolicy):
return False, reason return False, reason
if self.max_roles is not None: if self.max_roles is not None:
assigned = [name for name in held if name in roles and roles[name].priority != RolePriority.GENERAL] assigned = [name for name in held if name in assignable_roles(roles)]
if len(assigned) >= self.max_roles: if len(assigned) >= self.max_roles:
return False, "node already holds the maximum number of roles" return False, "node already holds the maximum number of roles"
return True, "accepted" return True, "accepted"
def should_keep(self, role_name, scores): def should_keep(self, role_name, scores):
if not scores:
return role_name not in self.deny
return self._violation(role_name, scores) is None return self._violation(role_name, scores) is None
colony/core/roles/role.py
+5 -1
View File
@@ -19,6 +19,7 @@ class Role:
self._stop_event = threading.Event() self._stop_event = threading.Event()
self._thread = None self._thread = None
self._start_lock = threading.Lock()
if self.CONTINUOUS and self.LOOP_SECONDS_WAIT is None: if self.CONTINUOUS and self.LOOP_SECONDS_WAIT is None:
raise NotImplementedError("You made this role to run continuously but have not set seconds to wait. Please set constant variable LOOP_SECONDS_WAIT. ") raise NotImplementedError("You made this role to run continuously but have not set seconds to wait. Please set constant variable LOOP_SECONDS_WAIT. ")
@@ -31,9 +32,12 @@ class Role:
return self.__class__.__name__ return self.__class__.__name__
def start(self): def start(self):
if not self.CONTINUOUS or self._thread is not None: if not self.CONTINUOUS:
return return
with self._start_lock:
if self._thread is not None:
return
self._thread = threading.Thread(target=self._run_loop, daemon=True, name=str(self)) self._thread = threading.Thread(target=self._run_loop, daemon=True, name=str(self))
self._thread.start() self._thread.start()
colony/settings.py
+4 -1
View File
@@ -17,6 +17,7 @@ DEFAULTS = {
"membership.reprobe_interval": 15, "membership.reprobe_interval": 15,
"membership.fanout": 3, "membership.fanout": 3,
"membership.forget_after": 600, "membership.forget_after": 600,
"membership.hello_timeout": 3,
"membership.max_members": 1024, "membership.max_members": 1024,
"membership.admission": "open", "membership.admission": "open",
"foreman.interval": 5, "foreman.interval": 5,
@@ -30,5 +31,7 @@ DEFAULTS = {
} }
def setting(key): def setting(key, override=None):
if override is not None:
return override
return configuration.get(key, DEFAULTS[key]) return configuration.get(key, DEFAULTS[key])
config.json
+2 -1
View File
@@ -12,7 +12,7 @@
"keyfile": "certs/server.key" "keyfile": "certs/server.key"
}, },
"discovery": { "discovery": {
"range_start": "192.168.65.110", "range_start": "192.168.65.100",
"range_end": "192.168.65.120", "range_end": "192.168.65.120",
"discover_port": 28149, "discover_port": 28149,
"probe_timeout": 1.0, "probe_timeout": 1.0,
@@ -24,6 +24,7 @@
"reprobe_interval": 15, "reprobe_interval": 15,
"fanout": 3, "fanout": 3,
"forget_after": 600, "forget_after": 600,
"hello_timeout": 3,
"max_members": 1024, "max_members": 1024,
"admission": "open" "admission": "open"
}, },
docker-compose.yml
+11
View File
@@ -0,0 +1,11 @@
services:
colony:
image: registry.ciomek.pl/colony:latest
container_name: colony
restart: unless-stopped
network_mode: host
volumes:
- colony-storage:/app/storage
volumes:
colony-storage:
state.json
-8
View File
@@ -1,8 +0,0 @@
{
"roles": [
"Unit",
"Foreman",
"VaultWarden",
"Explorer"
]
}