core: create Major Tom

Author: voorloopnul

core/services/install-services.sh

@@ -35,6 +35,7 @@ SERVICES=(
     ping
     versionchooser
     wifi
+    major_tom
 )
 
 # We need to install loguru and appdirs since they may be used inside setup.py

core/services/major_tom/major_tom.py

@@ -0,0 +1,96 @@
+#! /usr/bin/env python3
+import datetime
+import time
+import uuid
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, Optional
+from zoneinfo import ZoneInfo
+
+from src.core import DefaultPayload, TelemetryEngine, get_latency
+from src.metrics import Metrics
+from src.typedefs import ExtensionInfo, VersionInfo
+
+LOG_SESSION_UUID = str(uuid.uuid4())
+
+SERVICE_NAME = "major_tom"
+LOG_FOLDER_PATH = f"/var/logs/blueos/services/{SERVICE_NAME}/"
+
+TELEMETRY_ENDPOINT = "https://telemetry.blueos.cloud/api/v1/anonymous/"
+S3_TELEMETRY_ENDPOINT = "https://telemetry.blueos.cloud/api/v1/anonymous/s3/"
+
+
+def get_hardware_uuid() -> str:
+    try:
+        with open("/etc/blueos/hardware-uuid", "r", encoding="utf-8") as f:
+            return f.read().strip()
+    except Exception as error:
+        print(f"unable to read machine id: {error}")
+        return ""
+
+
+def get_blueos_uuid() -> str:
+    try:
+        with open("/etc/blueos/uuid", "r", encoding="utf-8") as f:
+            return f.read().strip()
+    except Exception as error:
+        print(f"unable to read blueos id: {error}")
+        return ""
+
+
+# pylint: disable=too-many-instance-attributes
+@dataclass
+class AnonymousTelemetryRecord:
+    uptime: float
+    latency: float
+    memory_size: int
+    memory_usage: int
+    disk_size: int
+    disk_usage: int
+    extensions: Optional[list[ExtensionInfo]]
+    blueos_version: Optional[VersionInfo]
+    probe_time: float
+
+    def json(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def compose_default_record(order: int) -> Dict[str, Any]:
+    date_time_utc = datetime.datetime.now(ZoneInfo("UTC")).isoformat()
+    payload = DefaultPayload(
+        log_session_uuid=LOG_SESSION_UUID,
+        order=order,
+        timestamp=date_time_utc,
+        hardware_id=get_hardware_uuid(),
+        blueos_id=get_blueos_uuid(),
+        data={},
+    )
+
+    start_probing = time.time()
+    metrics = Metrics()
+    record = AnonymousTelemetryRecord(
+        time.clock_gettime(time.CLOCK_BOOTTIME),
+        get_latency(),
+        metrics.memory.total,
+        metrics.memory.used,
+        metrics.disk.total,
+        metrics.disk.used,
+        metrics.installed_extensions,
+        metrics.installed_version,
+        0,
+    )
+    record.probe_time = time.time() - start_probing
+    payload.data = record.json()
+    return payload.json()
+
+
+if __name__ == "__main__":
+    TelemetryEngine(
+        label="anonymous",  # used to tag telemetry type. we may have non-anonymous telemetry in the future
+        endpoint=TELEMETRY_ENDPOINT,
+        s3_endpoint=S3_TELEMETRY_ENDPOINT,
+        create_record=compose_default_record,
+        interval=60 * 5,  # 5 minutes
+        max_file_size=1024 * 1024,  # 1Mb
+        max_file_retention=10,
+        buffer_folder=LOG_FOLDER_PATH,
+    )()

core/services/major_tom/requirements.txt

@@ -0,0 +1,6 @@
+psutil==5.9.5
+requests==2.31.0
+speedtest-cli==2.1.3
+Flask==2.3.2
+py-machineid==0.4.3
+loguru==0.7.0

core/services/major_tom/setup.py

@@ -0,0 +1,147 @@
+import copy
+import gzip
+import json
+import os
+import shutil
+import time
+from dataclasses import asdict, dataclass
+from typing import Any, Callable, Dict, List
+
+import loguru
+import requests
+import speedtest
+
+loguru.logger.remove()
+
+
+def formatter(record: "loguru.Record") -> str:
+    # Note this function returns the string to be formatted, not the actual message to be logged
+    record["extra"]["serialized"] = json.dumps(record["message"])
+    return "{extra[serialized]}\n"
+
+
+def is_online() -> bool:
+    return get_latency() > 0
+
+
+def get_latency() -> float:
+    try:
+        servers: List[str] = []
+        st = speedtest.Speedtest()
+        st.get_servers(servers)
+        best_server = st.get_best_server()
+        ping = best_server["latency"]
+        return float(ping)
+    except Exception:
+        return -1.0
+
+
+@dataclass
+class DefaultPayload:
+    log_session_uuid: str
+    order: int
+    timestamp: str
+    hardware_id: str
+    blueos_id: str
+    data: dict[str, Any]
+
+    def json(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+class TelemetryEngine:
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        label: str,
+        endpoint: str,
+        s3_endpoint: str,
+        create_record: Callable[[Any], Any],
+        interval: float,
+        max_file_size: int,
+        max_file_retention: int,
+        buffer_folder: str,
+    ):
+        self.buffer_file = f"{buffer_folder}/{label}_usage.log"
+        self.buffer_folder = buffer_folder
+
+        self.telemetry_endpoint = endpoint
+        self.telemetry_s3_endpoint = s3_endpoint
+        self.create_record = create_record
+        self.interval = interval
+
+        self.buffer = copy.deepcopy(loguru.logger)
+        self.buffer.add(
+            self.buffer_file,
+            rotation=max_file_size,
+            retention=max_file_retention,
+            format=formatter,
+            compression="gz",
+        )
+
+    def __call__(self) -> None:
+        order = 0
+        while True:
+            order += 1
+            record = self.create_record(order)
+            if self.save(record) == "online":
+                self.process_buffered_records()
+            time.sleep(self.interval)
+
+    def upload_file(self, file: str) -> bool:
+        """
+        This method requests to telemetry API a presigned url and upload the local archived files.
+        """
+        print(f"uploading file... {file}")
+        try:
+            response = requests.get(self.telemetry_s3_endpoint, timeout=5).json()
+            with open(file, "rb") as fh:
+                files = {"file": (file, fh)}
+                r = requests.post(response["url"], data=response["fields"], files=files, timeout=300)
+                if r.status_code == 204:
+                    print("[Success!]")
+                    return True
+        except Exception:
+            pass
+
+        return False
+
+    def process_buffered_records(self) -> None:
+        """
+        Check in the buffered folder if there are archived logs to upload. If the agent connects before an archive
+        is created it will also archive the current buffer file and upload it.
+        """
+        for file in os.listdir(self.buffer_folder):
+            file_path = os.path.join(self.buffer_folder, file)
+
+            # Upload regular archive
+            if file_path.endswith(".log.gz"):
+                if self.upload_file(file_path):
+                    os.remove(file_path)
+
+            # Archive current buffer and upload it
+            if file_path.endswith(".log") and os.path.getsize(file_path):
+                timestamp = int(time.time())
+                tmp_name = self.buffer_file.replace(".log", f".{timestamp}.log.gz")
+                with open(self.buffer_file, "rb") as f_in:
+                    with gzip.open(tmp_name, "wb") as f_out:
+                        shutil.copyfileobj(f_in, f_out)
+                if self.upload_file(tmp_name):
+                    os.remove(tmp_name)
+        with open(self.buffer_file, "w", encoding="utf-8"):
+            # create new empty file if not there
+            pass
+
+    def save(self, record: Dict[str, Any]) -> str:
+        """
+        Try to POST the telemetry payload, if it fails for any reason, we buffer it locally.
+        """
+        try:
+            r = requests.post(self.telemetry_endpoint, json=record, timeout=5)
+            if r.status_code == 201:
+                return "online"
+        except Exception as err:
+            print(err)
+
+        self.buffer.info(record)
+        return "offline"

core/services/major_tom/src/__init__.py

@@ -0,0 +1,45 @@
+from functools import cached_property
+from typing import List, Optional
+
+import psutil
+import requests
+
+from src.typedefs import ExtensionInfo, VersionInfo
+
+
+class Metrics:
+    @cached_property
+    def installed_extensions(self) -> Optional[List[ExtensionInfo]]:
+        try:
+            req = requests.get("http://localhost/kraken/v1.0/installed_extensions", timeout=3)
+            if req.status_code == 200:
+                return [ExtensionInfo(identifier=rec["identifier"], tag=rec["tag"]) for rec in req.json()]
+        except Exception:
+            return None
+        return []
+
+    @cached_property
+    def disk(self) -> psutil._common.sdiskusage:
+        return psutil.disk_usage("/")
+
+    @cached_property
+    def memory(self) -> psutil._pslinux.svmem:
+        return psutil.virtual_memory()
+
+    @cached_property
+    def installed_version(self) -> Optional[VersionInfo]:
+        try:
+            req = requests.get("http://localhost/version-chooser/v1.0/version/current", timeout=3)
+            if req.status_code == 200:
+                data = req.json()
+                return VersionInfo(
+                    repository=data["repository"],
+                    tag=data["tag"],
+                    last_modified=data["last_modified"],
+                    sha=data["sha"],
+                    architecture=data["architecture"],
+                )
+
+        except Exception:
+            return None
+        return None

core/services/major_tom/src/core.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class ExtensionInfo:
+    identifier: str
+    tag: str
+
+
+class VersionInfo:
+    repository: str
+    tag: str
+    last_modified: str
+    sha: str
+    architecture: str

core/services/major_tom/src/metrics.py

@@ -93,6 +93,7 @@ SERVICES=(
     'nginx',"nice -18 nginx -g \"daemon off;\" -c $TOOLS_PATH/nginx/nginx.conf"
     'log_zipper',"nice -20 $SERVICES_PATH/log_zipper/main.py '/shortcuts/system_logs/**/*.log' --max-age-minutes 60"
     'bag_of_holding',"$SERVICES_PATH/bag_of_holding/main.py"
+    'major_tom',"$SERVICES_PATH/major_tom/major_tom.py"
 )
 
 tmux -f /etc/tmux.conf start-server