Collect Logs and Diagnostic Files from Distributed Servers into a Central Location

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

import base64
import time


def encode_path(device_id, raw_path):
    normalized = str(raw_path or "").replace("\\", "/")
    token = base64.b64encode(normalized.encode("utf-8")).decode("ascii")
    return f"{device_id}_ino_{token}"


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())

def build_filter(exts=None, min_size=None, exclude=None):
    file_option = {}

    if exts:
        # extension whitelist, without the leading dot
        file_option["extension"] = {
            "extension": [e.lstrip(".").lower() for e in exts],
            "allow": True,
        }

    if min_size is not None:
        # over and equal both True means size or larger
        file_option["fileSize"] = {"size": min_size, "over": True, "equal": True}

    if exclude:
        # allow=False excludes files whose name contains this. Server matching is case sensitive.
        file_option["fileName"] = {"name": exclude, "allow": False}

    return {"send-fileoption": file_option} if file_option else {}

LOG_FILTER = build_filter(exts=["log", "gz"], exclude=".lck")
DUMP_FILTER = build_filter(exts=["core", "dmp", "hprof"])

page = api("POST", f"/api/devices/{device_id}/files/search",
           {"path": "/var/log/application", "pageSize": 500})

matched = [i for i in page["items"]
           if i["type"] == "file" and i["name"].endswith((".log", ".gz"))]

print(f"{len(matched)} matched")

def build_collection(name, source, source_path, target, target_path,
                     schedule, options):
    return {
        "name": name,
        "flowName": name,
        "transferType": "normal",
        "timezone": "Asia/Seoul",
        "step": 1,
        "isUpcoming": False,
        "details": [
            {
                "senderId": source,
                "receiverId": target,
                "sourceItem": [
                    {
                        "hash": encode_path(source, source_path),
                        "filePath": source_path,
                        "isDir": True,
                    }
                ],
                "targetPath": encode_path(target, target_path),
                "step": 1,
                "transferOptions": {
                    "noSchedule": False,
                    "target-action": "numbering",
                    "send-fileoption": {},
                    **options,
                },
            }
        ],
        "schedules": [schedule],
    }


DAILY_4AM = {
    "type": "day",
    "startDateType": "now",
    "hour": "04",
    "minute": "00",
    "ampm": "am",
    "startDate": now_iso(),
    "timezone": "Asia/Seoul",
}

api("POST", "/api/automations", build_collection(
    "daily log", "device-app-01", "/var/log/application",
    "device-central-01", "/collect/device-app-01",
    DAILY_4AM, LOG_FILTER))

body = build_collection(
    "core dump", "device-app-01", "/var/crash",
    "device-central-01", "/collect/device-app-01/dump",
    {"type": "none", "startDateType": "now",
     "startDate": now_iso(), "timezone": "Asia/Seoul"},
    {**DUMP_FILTER, "syncType": 1, "watchFolderType": 1})

body["transferType"] = "sync"
body["details"][0]["transferOptions"]["noSchedule"] = True

api("POST", "/api/automations", body)

SOURCES = [
    ("device-app-01", "/var/log/application"),
    ("device-app-02", "/var/log/application"),
    ("device-linux-01", "/var/log/system"),
    ("device-edge-01", "/data/diagnostics"),
]

for source, path in SOURCES:
    api("POST", "/api/automations", build_collection(
        f"collect {source}", source, path,
        "device-central-01", f"/collect/{source}",
        DAILY_4AM, LOG_FILTER))

body["processors"] = [{
    "category": "run",
    "type": "http",
    "config": {
        "url": "https://internal.example.com/analyze",
        "method": "POST",
    },
}]

def on_collect_hook(payload):
    monitor_id = payload.get("monitorId")

    # If you subscribed to the completed event only, this check can be skipped
    if monitor_id:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if detail["status"] != STATUS_COMPLETE:
            return skip_failed_collection(payload)

    start_analysis(payload)

def wait(monitor_id, timeout=3600, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

from collections import Counter
from datetime import datetime, timedelta, timezone


def history(device_id, days=1):
    end = datetime.now(timezone.utc)
    fmt = "%Y-%m-%dT%H:%M:%SZ"

    return list(paginate(f"/api/devices/{device_id}/transfer-history", params={
        "startDate": (end - timedelta(days=days)).strftime(fmt),
        "endDate": end.strftime(fmt),
    }))


for source, _ in SOURCES:
    rows = history(source)
    failed = [r for r in rows if r.get("status") in NOT_SUCCEEDED]
    mark = "" if not failed else "  <- needs attention"
    print(f"{source:20} collected {len(rows):>4} failed {len(failed):>3}{mark}")