Automatically place servers using learned resource capacity

static/agent/ec-agent.py

@@ -9,6 +9,8 @@ import argparse
 import base64
 from concurrent.futures import ThreadPoolExecutor, as_completed
 import contextlib
+from datetime import datetime, timezone
+import math
 import hashlib
 import fcntl
 import json
@@ -30,7 +32,7 @@ import urllib.parse
 import urllib.request
 import uuid
 
-VERSION = "1.4.1"
+VERSION = "1.4.3"
 LABEL = "io.electric-cherry"
 MAX_FILE = 32 * 1024 * 1024
 LIBRARY_DOWNLOAD_WORKERS = 4
@@ -710,6 +712,8 @@ class Agent:
         self.docker = docker or Docker()
         self.stopping = threading.Event()
         self.current_operation = None
+        self.cpu_sample_lock = threading.Lock()
+        self.cpu_sample = None
         self.instance_locks = {}
         self.read_gate = ReadWriteGate()
         self.read_claim_lock = threading.Lock()
@@ -899,7 +903,88 @@ class Agent:
         except (OSError, ValueError):
             raise AgentError("Controller request failed") from None
 
+    def cpu_usage_percent(self):
+        """Host-wide busy time since the previous inventory sample, without waiting."""
+        with self.cpu_sample_lock:
+            previous = self.cpu_sample
+            self.cpu_sample = None
+            try:
+                with Path("/proc/stat").open() as source:
+                    fields = source.readline().split()
+                if not fields or fields[0] != "cpu" or len(fields) < 9:
+                    return None
+                # Guest time is already included in user/nice. Idle and I/O
+                # wait are not busy; steal counts as unavailable CPU capacity.
+                counters = tuple(int(value) for value in fields[1:9])
+                if any(value < 0 for value in counters):
+                    return None
+            except (OSError, ValueError):
+                return None
+            self.cpu_sample = counters
+            if previous is None:
+                return None
+            deltas = [current - old for current, old in zip(counters, previous)]
+            total = sum(deltas)
+            # Counters can reset and Linux can revise I/O wait downwards.
+            # Rebase rather than reporting a misleading utilization value.
+            if total <= 0 or any(delta < 0 for delta in deltas):
+                return None
+            return round(100 * (total - deltas[3] - deltas[4]) / total, 1)
+
+    def instance_resources(self):
+        """One bounded Docker stats batch; unavailable samples are never zero load."""
+        ids = self.docker.call("container", "ls", "--filter", LABEL + ".managed=true",
+                               "--no-trunc", "--format", "{{.ID}}", timeout=10).splitlines()[:1000]
+        if not ids:
+            return []
+        containers = json.loads(self.docker.call("container", "inspect", *ids, timeout=10))
+        owned = {}
+        for info in containers:
+            try:
+                labels = info.get("Config", {}).get("Labels", {}) or {}
+                if labels.get(LABEL + ".managed") != "true" or not info.get("State", {}).get("Running"):
+                    continue
+                instance_id, generation = instance_identity(labels.get(LABEL + ".instance"), int(labels.get(LABEL + ".generation", "")))
+                name = self.container_name(instance_id, generation)
+                if info.get("Name", "").lstrip("/") != name:
+                    continue
+                expected = str(self.instance_dir(instance_id, generation) / "app")
+                if not any(mount.get("Type") == "bind" and mount.get("Source") == expected and mount.get("Destination") == "/home/plutainer/app" for mount in info.get("Mounts", [])):
+                    continue
+                started = datetime.fromisoformat(info["State"]["StartedAt"].replace("Z", "+00:00"))
+                uptime = max(0, int((datetime.now(timezone.utc) - started).total_seconds()))
+                owned[name] = {"instanceId": instance_id, "generation": generation, "uptimeSeconds": uptime}
+            except (AgentError, ValueError, KeyError, TypeError):
+                continue
+        if not owned:
+            return []
+        rows = self.docker.call("stats", "--no-stream", "--no-trunc", "--format", "{{json .}}", *owned, timeout=15)
+        samples = []
+        units = {"B": 1, "KiB": 1024, "MiB": 1024 ** 2, "GiB": 1024 ** 3,
+                 "TiB": 1024 ** 4, "kB": 1000, "MB": 1000 ** 2, "GB": 1000 ** 3}
+        for line in rows.splitlines():
+            try:
+                row = json.loads(line)
+                identity = owned.get(row.get("Name"))
+                if not identity or int(row["PIDs"]) <= 0:
+                    continue
+                # Docker Linux CPU percent is core-based: 150% means 1.5 cores.
+                percent = row["CPUPerc"]
+                if not percent.endswith("%"):
+                    continue
+                cores = float(percent[:-1]) / 100
+                used = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)\s*([A-Za-z]+)", row["MemUsage"].split("/", 1)[0].strip())
+                if not used or used[2] not in units or not math.isfinite(cores) or cores < 0:
+                    continue
+                memory = math.ceil(float(used[1]) * units[used[2]])
+                if memory > 0:
+                    samples.append({**identity, "cpuCores": cores, "memoryBytes": memory})
+            except (ValueError, KeyError, TypeError, AttributeError, OverflowError):
+                continue
+        return samples
+
     def inventory(self):
+        cpu_usage = self.cpu_usage_percent()
         usage = shutil.disk_usage(self.root)
         memory = {}
         try:
@@ -997,6 +1082,10 @@ class Agent:
                 if aliases.exists():
                     with contextlib.suppress(OSError, ValueError):
                         images.extend(alias for alias, image_id in json.loads(aliases.read_text()).items() if image_id in image_ids)
+        resources = {}
+        if docker_running:
+            with contextlib.suppress(AgentError, ValueError, TypeError, KeyError):
+                resources["instanceResources"] = self.instance_resources()
         os_name = platform.system()
         with contextlib.suppress(OSError):
             for line in Path("/etc/os-release").read_text().splitlines():
@@ -1004,7 +1093,7 @@ class Agent:
                     os_name = line.split("=", 1)[1].strip('"')
         update = self.update_inventory()
         capabilities = ["instance.cancel-create", "operations.read-lane"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
-        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
+        return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
 
     def safe_to_disrupt(self, instance_id, generation):
         health = self.inspect(instance_id, generation)

static/agent/manifest.json

@@ -1,8 +1,8 @@
 {
-  "version": "1.4.1",
+  "version": "1.4.3",
   "protocolVersion": 1,
   "platform": "debian-amd64",
-  "sha256": "0691d8457f52ef8b53691b627733352b2f187529a4ecc465cdaaa6e27ddaaaa7",
+  "sha256": "9ca9bff1a42ab838b175329c6a8458e18053429efe2f050f477a9ad9c598410f",
   "updaterProtocolVersion": 1,
   "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8"
 }