replicator

.dockerignore

@@ -9,6 +9,8 @@ docs
 tests
 logs
 storage
+storages
+data
 state.json
 state.json.corrupt
 .tmp-*.json

.gitignore

@@ -1,6 +1,8 @@
 .stfolder
 logs
 storage
+/storages
+/data
 state.json
 state.json.corrupt
 .tmp-*.json

Dockerfile

@@ -12,7 +12,7 @@ COPY . .
 
 RUN chmod +x colonyctl.py && ln -s /app/colonyctl.py /usr/local/bin/colonyctl
 
-VOLUME ["/app/storage", "/app/logs"]
+VOLUME ["/app/data", "/app/storages", "/app/logs"]
 
 EXPOSE 65334/tcp 28149/udp 8080/tcp
 

colony/cli.py

@@ -111,13 +111,14 @@ def role_states(payload):
     for role in payload["roles"]:
         holders = sorted(address(member) for member in alive if role["name"] in member["roles"])
         required = role["required"]
+        limit = role.get("max", required)
         if role.get("on_demand"):
             state = "held" if holders else "off"
         elif required is None:
             state = "general"
         elif len(holders) < required:
             state = "unmet"
-        elif len(holders) > required:
+        elif limit is not None and len(holders) > limit:
             state = "duplicate"
         else:
             state = "held"

colony/core/context.py

@@ -10,7 +10,16 @@ class RoleContext:
     def execute(self, command, *args, **kwargs):
         return self._network.execute(command, *args, **kwargs)
 
+    def execute_on(self, target, command, *args, **kwargs):
+        """Run a command on one particular node, going through a gateway if it cannot be dialled.
+
+        `target` is a Node or a MemberRecord, anything with `ip` and `port`, for example an item of
+        `nodes_with_role(...)`. `execute` picks the node for you; this one does not.
+        """
+        return self._network.send_command(target.ip, target.port, command, *args, **kwargs)
+
     def execute_to_unregistered_node(self, ip, port, command, *args, **kwargs):
+        """The same as `execute_on` for a caller that only has an address."""
         return self._network.send_command(ip, port, command, *args, **kwargs)
 
     def send_direct(self, ip, port, payload, timeout=None, connect_timeout=None):
@@ -28,6 +37,9 @@ class RoleContext:
     def peers(self):
         return self._network.get_all_nodes()
 
+    def nodes_with_role(self, role):
+        return self._network.get_nodes_of_role(role if isinstance(role, str) else role.__name__)
+
     def membership(self):
         return self._network.membership
 

colony/core/network/advice.py

@@ -1,3 +1,5 @@
+import math
+
 from colony.role_registry import HardwareType, RolePriority
 
 HARDWARE_SCORE_KEY = {
@@ -9,12 +11,16 @@ HARDWARE_SCORE_KEY = {
 }
 
 
-def required_holders(role_stat):
-    return 0 if role_stat.on_demand else 1
+def required_holders(role_stat, cluster_size=1):
+    if role_stat.on_demand:
+        return 0
+    return max(1, math.ceil(cluster_size * role_stat.minimal_percentage / 100))
 
 
 def max_holders(role_stat):
-    return None if role_stat.on_demand else required_holders(role_stat)
+    if role_stat.on_demand or role_stat.minimal_percentage:
+        return None
+    return required_holders(role_stat)
 
 
 def assignable_roles(roles):
@@ -29,7 +35,7 @@ def claimants(records, role_name):
 def unmet_roles(records, roles):
     return [
         name for name in assignable_roles(roles)
-        if len(claimants(records, name)) < required_holders(roles[name])
+        if len(claimants(records, name)) < required_holders(roles[name], len(records))
     ]
 
 

colony/core/network/core.py

@@ -1,6 +1,6 @@
 import json
 from enum import Enum
-from typing import Optional
+from typing import List, Optional
 
 from colony.core.network.membership import Membership
 from colony.core.network.nodes import LocalNode, Node
@@ -56,11 +56,13 @@ class Network:
     def get_all_nodes(self):
         return self.nodes
 
+    def get_nodes_of_role(self, role_name: str) -> List[Node]:
+        return [node for node in self.nodes if role_name in node.roles]
+
     def get_node_of_role(self, role_name: str) -> Optional[Node]:
-        for node in self.nodes:
-            for role in node.roles:
-                if role == role_name:
-                    return node
+        holders = self.get_nodes_of_role(role_name)
+        if holders:
+            return holders[0]
 
         self.logger.warning(f"No node found with role: {role_name}")
         return None
@@ -190,7 +192,7 @@ class Network:
         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 = "storages/default/colony_snapshot.json"):
         self.logger.info(f"Loading network snapshot from {file_path}")
         nodes = []
         try:

colony/core/network/membership.py

@@ -8,7 +8,8 @@ from dataclasses import dataclass, field, asdict
 from colony.settings import setting
 
 ADMISSION_POLICIES = {
-    "open": lambda record: True,
+    "open": lambda membership, record: True,
+    "closed": lambda membership, record: membership.knows(record.key),
 }
 
 MAX_TEXT = 255
@@ -107,7 +108,7 @@ class MemberRecord:
 
 def resolve_admission(admission):
     if callable(admission):
-        return admission
+        return lambda membership, record: admission(record)
     try:
         return ADMISSION_POLICIES[admission]
     except KeyError:
@@ -138,7 +139,12 @@ class Membership:
         return self.clock()
 
     def admit(self, record):
-        return bool(self._admission(record))
+        return bool(self._admission(self, record))
+
+    def knows(self, key):
+        with self._lock:
+            self._purge(self.clock())
+            return key in self._entries
 
     def _current_self(self):
         now = self.clock()

colony/core/roles/replicator/role.py

@@ -0,0 +1,66 @@
+import base64
+import random
+from collections import Counter
+from enum import Enum
+
+from colony.core.roles.role import Role, Command
+from colony.core.storages.replica import ReplicaStorage
+
+
+class Replicator(Role):
+    class COMMANDS(Enum):
+        # Write to replicator
+        REPLICATE = 'Replicator.Replicate'
+        # Read from replicator
+        RETRIEVE = 'Replicator.Retrieve'
+        HASH_SUM = 'Replicator.HashSum'
+
+    def __init__(self, context, storage=None):
+        commands = {
+            self.COMMANDS.REPLICATE.value: Command(self.replicate),
+            self.COMMANDS.RETRIEVE.value: Command(self.retrieve),
+            self.COMMANDS.HASH_SUM.value: Command(self.hash_sum),
+        }
+
+        self.storage = storage or ReplicaStorage()
+
+        super().__init__(commands, context)
+
+    def start(self):
+        # catch up
+        replicator_checksums = dict()
+
+        for node in self.context.nodes_with_role(Replicator):
+            if (node.ip, node.port) == self.context.membership().key:
+                continue
+
+            response = self.context.execute_on(node, Replicator.COMMANDS.HASH_SUM)
+            if response['status'] != 'ok':
+                continue
+
+            checksum = response['result']
+
+            replicator_checksums[node] = checksum
+
+        if replicator_checksums:
+            checksum = Counter(replicator_checksums.values()).most_common(1)[0][0]
+            matching_nodes = [
+                node
+                for node, node_checksum in replicator_checksums.items()
+                if node_checksum == checksum
+            ]
+
+            if checksum != self.hash_sum():
+                node = random.choice(matching_nodes)
+                response = self.context.execute_on(node, Replicator.COMMANDS.RETRIEVE)
+                self.storage.restore(base64.b64decode(response['result']))
+
+    def replicate(self, filename, content):
+        self.storage.save(filename, content)
+        return True
+
+    def retrieve(self):
+        return base64.b64encode(self.storage.archive()).decode()
+
+    def hash_sum(self):
+        return self.storage.dir_checksum()

colony/core/roles/unit/role.py

@@ -35,13 +35,16 @@ class Unit(Role):
             via = self.context.route_of(key)
             member["via"] = list(via) if via else None
             member["direct"] = self.context.is_direct(key)
+        alive = sum(1 for member in members if member["alive"])
         return {
             "members": members,
             "roles": [
                 {
                     "name": name,
                     "priority": roles[name].priority.name.lower(),
-                    "required": None if roles[name].priority == RolePriority.GENERAL else advice.required_holders(roles[name]),
+                    "required": None if roles[name].priority == RolePriority.GENERAL
+                    else advice.required_holders(roles[name], alive),
+                    "max": advice.max_holders(roles[name]),
                     "on_demand": roles[name].on_demand,
                 }
                 for name in sorted(roles, key=lambda name: (roles[name].priority.value, name))

colony/core/roles/vaultwarden/role.py

@@ -1,10 +1,13 @@
 from enum import Enum
 
-from colony.core.storage import DefaultStorage
+from colony.core.storages.default import DefaultStorage
+from colony.core.roles.replicator.role import Replicator
 from colony.core.roles.role import Role, Command
 
 
 class VaultWarden(Role):
+    REPLICATE_TIMEOUT = 3
+
     class COMMANDS(Enum):
         LS = 'VaultWarden.ls'
         OPEN = 'VaultWarden.open'
@@ -28,4 +31,24 @@ class VaultWarden(Role):
         return self.storage.open(file)
 
     def overwrite(self, filename, content):
-        return self.storage.save(filename, content)
+        path = self.storage.save(filename, content)
+        if self.storage.SHARED:
+            self._push_to_replicators(filename, content)
+        return path
+
+    def _push_to_replicators(self, filename, content):
+        for record in self.context.membership().alive_records():
+            if Replicator.__name__ not in record.roles:
+                continue
+            target = f"{record.ip}:{record.port}"
+            try:
+                resp = self.context.execute_to_unregistered_node(
+                    record.ip, record.port, Replicator.COMMANDS.REPLICATE, filename, content,
+                    timeout=self.REPLICATE_TIMEOUT,
+                )
+            except Exception as e:
+                self.logger.warning(f"Could not replicate {filename} to {target}: {e}")
+                continue
+            if not isinstance(resp, dict) or resp.get("status") != "ok":
+                error = resp.get("error") if isinstance(resp, dict) else "bad reply"
+                self.logger.warning(f"{target} did not accept {filename}: {error}")

colony/core/roles/webview/static/model.js

@@ -51,11 +51,12 @@
   function roleStates(members, roles) {
     return roles.map((role) => {
       const holders = members.filter((node) => node.alive && node.roles.includes(role.name)).length;
+      const limit = role.max === undefined ? role.required : role.max;
       let state = "held";
       if (role.on_demand) state = holders ? "held" : "off";
       else if (role.required === null) state = "general";
       else if (holders < role.required) state = "unmet";
-      else if (holders > role.required) state = "duplicate";
+      else if (limit !== null && holders > limit) state = "duplicate";
       return { name: role.name, state: state };
     });
   }

colony/core/state_manager.py

@@ -20,10 +20,11 @@ class StateManager:
         Data.Peers: [],
     }
 
-    STATE_FILE = Path("state.json")
+    STATE_FILE = Path("data") / "state.json"
 
     def __init__(self, path=None):
         self._path = Path(path) if path is not None else self.STATE_FILE
+        self._path.parent.mkdir(parents=True, exist_ok=True)
         self.logger = get_logger(self.__class__.__name__)
         self._lock = threading.RLock()
         self._data = {}

colony/core/storages/core.py

@@ -1,16 +1,21 @@
+import hashlib
+import io
 import os
 import json
+import tarfile
 import threading
-from enum import Enum
+from pathlib import Path
 
 from colony.utils.files import atomic_write_json
 
-STORAGE_DIR = "storage"
+STORAGES_DIR = "storages"
 
 _write_lock = threading.Lock()
 
 
 class Storage:
+    SHARED = False
+
     def __init__(self, storage_dir):
         self.storage_dir = storage_dir
 
@@ -70,11 +75,27 @@ class Storage:
 
         return False
 
+    def dir_checksum(self):
+        h = hashlib.sha256()
+        path = Path(self.storage_dir)
 
-class DefaultStorage(Storage):
-    def __init__(self):
-        super().__init__(STORAGE_DIR)
+        for file in sorted(p for p in path.rglob("*") if p.is_file() and not p.name.startswith(".tmp-")):
+            h.update(str(file.relative_to(path)).encode())
+            with file.open("rb") as f:
+                for chunk in iter(lambda: f.read(1024 * 1024), b""):
+                    h.update(chunk)
 
+        return h.hexdigest()
 
-class DEFAULT_STORAGE_FILES(Enum):
-    COLONY_SNAPSHOT = 'colony_snapshot.json'
+    def archive(self):
+        buffer = io.BytesIO()
+
+        with tarfile.open(fileobj=buffer, mode="w:gz") as tar:
+            for file in sorted(p for p in Path(self.storage_dir).rglob("*") if p.is_file() and not p.name.startswith(".tmp-")):
+                tar.add(file, arcname=file.relative_to(self.storage_dir).as_posix())
+
+        return buffer.getvalue()
+
+    def restore(self, data):
+        with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
+            tar.extractall(self.storage_dir, filter="data")

colony/core/storages/default.py

@@ -0,0 +1,15 @@
+import os
+from enum import Enum
+
+from colony.core.storages.core import STORAGES_DIR, Storage
+
+STORAGE_DIR = os.path.join(STORAGES_DIR, "default")
+
+
+class DefaultStorage(Storage):
+    SHARED = True
+
+    COLONY_SNAPSHOT = 'colony_snapshot.json'
+
+    def __init__(self):
+        super().__init__(STORAGE_DIR)

colony/core/storages/replica.py

@@ -0,0 +1,10 @@
+import os
+
+from colony.core.storages.core import STORAGES_DIR, Storage
+
+REPLICA_DIR = os.path.join(STORAGES_DIR, "replica")
+
+
+class ReplicaStorage(Storage):
+    def __init__(self):
+        super().__init__(REPLICA_DIR)

colony/role_registry.py

@@ -31,12 +31,13 @@ class HardwarePreferences:
 
 
 class RoleStats:
-    def __init__(self, cls, priority: RolePriority, hardware_preferences: HardwarePreferences = None, on_demand: bool = False):
+    def __init__(self, cls, priority: RolePriority, hardware_preferences: HardwarePreferences = None, on_demand: bool = False, minimal_percentage: int = 0):
         self.name = cls.__name__
         self.cls = cls
         self.priority = priority
         self.hardware_preferences = hardware_preferences
         self.on_demand = on_demand
+        self.minimal_percentage = max(0, min(minimal_percentage, 100))
 
 
 class RoleRegistry:

colony/settings.py

@@ -38,6 +38,9 @@ DEFAULTS = {
         "Unit.ShowHardware",
         "Unit.ShowCluster",
         "Unit.RequestRole",
+        "Replicator.HashSum",
+        "Replicator.Retrieve",
+        "Replicator.Replicate",
     ],
     "web.host": "0.0.0.0",
     "web.port": 8080,

config.json

@@ -13,7 +13,7 @@
   },
   "discovery": {
     "range_start": "192.168.65.100",
-    "range_end": "192.168.65.120",
+    "range_end": "192.168.65.100",
     "discover_port": 28149,
     "probe_timeout": 1.0,
     "probe_workers": 64
@@ -25,8 +25,8 @@
     "fanout": 3,
     "forget_after": 600,
     "hello_timeout": 3,
-    "max_members": 1024,
-    "admission": "open"
+    "max_members": 10,
+    "admission": "closed"
   },
   "foreman": {
     "interval": 5,
@@ -38,8 +38,28 @@
     "host": "0.0.0.0",
     "port": 8080
   },
+  "explorer": {
+    "hosts": []
+  },
   "gateway": {
-    "max_attempts": 4
+    "client_fallback": true,
+    "connect_timeout": 2,
+    "direct_retry": 30,
+    "max_attempts": 4,
+    "forward_timeout": 8,
+    "max_concurrent": 16,
+    "max_payload_bytes": 4194304,
+    "allow_commands": [
+      "Gossip.Hello",
+      "Unit.ShowStats",
+      "Unit.ShowRoles",
+      "Unit.ShowHardware",
+      "Unit.ShowCluster",
+      "Unit.RequestRole",
+      "Replicator.HashSum",
+      "Replicator.Retrieve",
+      "Replicator.Replicate"
+    ]
   },
   "role_policy": {
     "max_roles": null,

docker-compose.yml

@@ -5,7 +5,9 @@ services:
     restart: unless-stopped
     network_mode: host
     volumes:
-      - colony-storage:/app/storage
+      - colony-data:/app/data
+      - colony-storages:/app/storages
 
 volumes:
-  colony-storage:
+  colony-data:
+  colony-storages:

role_settings.py

@@ -1,4 +1,5 @@
 from colony.core.roles.explorer.role import Explorer
+from colony.core.roles.replicator.role import Replicator
 from colony.core.roles.unit.role import Unit
 from colony.role_registry import RoleStats, HardwarePreferences, RolePriority, HardwareType
 from colony.core.roles.foreman.role import Foreman
@@ -48,4 +49,15 @@ available_roles = [
         cls=Gateway,
         priority=RolePriority.GENERAL,
     ),
+    RoleStats(
+        cls=Replicator,
+        priority=RolePriority.CRITICAL,
+        hardware_preferences=HardwarePreferences(
+            {
+                HardwareType.DISK_SPACE: 1,
+                HardwareType.LONGEVITY: 3,
+            }
+        ),
+        minimal_percentage=51
+    )
 ]