Central Collection of Logs and Core Dumps

Getting Started

Core Concepts

Collect logs and diagnostic files from multiple systems into a single analysis environment

Servers and applications generate logs, core dumps, error reports, and various diagnostic files during operation.

Central log and core dump collection gathers files generated on each system according to configured criteria and automatically transfers them to a designated central analysis environment.

Connecting multiple servers and applications into a single collection flow lets you review system-specific diagnostic data centrally and use it for analysis.

Server A ──┐
           │
Server B ──┼──→ Central Collection ──→ Analysis System
           │
App Server ─┤
           │
Edge Device ─┘

Collection Flow

Automatically connect everything from file creation to the central analysis environment

When logs or diagnostic files are created on a collection target system, start the collection task based on file type, path, and configured execution conditions.

Transfer collected files to central storage or an analysis environment, then connect collection completion to the required analysis and monitoring tasks.

① Create Logs · Diagnostic Files

↓

② Identify Collection Targets

↓

③ Apply Collection Criteria

↓

④ Transfer to Central Analysis Environment

↓

⑤ Connect Analysis · Monitoring Tasks

↓

⑥ Check Execution Results

This flow lets you configure the collection and subsequent use of diagnostic files generated across multiple systems as a single operational workflow.

Operational Benefits

Review distributed diagnostic data centrally and use it in analysis workflows

Connecting diagnostic files from multiple systems to a central collection environment lets you manage file locations, collection status, and analysis targets in a single flow.

This creates a diagnostic-file workflow that connects file creation → central collection → analysis integration → result confirmation.

IT Engineers

Collection Environment

Connect systems where logs and diagnostic files are generated

First, connect the servers and application environments where logs, core dumps, and diagnostic files are generated to the collection flow.

Specify the locations where files are generated on each system to configure the source paths used by the central collection task.

Devices
   │
   ├── Application Server
   │       └── /var/log/application
   │
   ├── Linux Server
   │       └── /var/log/system
   │
   └── Edge Device
           └── /data/diagnostics

Collection Policy

Configure collection criteria based on file type and priority

After connecting the collection environment, define which files to collect and according to which criteria.

Specify collection targets based on file extension, path, creation, or modification conditions, and configure processing order and execution criteria according to file type and priority.

File Event
    │
    ▼
Collection Policy
    │
    ├── Log File ──────→ Standard Collection
    │
    ├── Error Report ──→ Analysis Flow
    │
    └── Core Dump ─────→ Priority Collection

Central Collection

Transfer diagnostic files from multiple systems to a central analysis environment

Transfer files from each system to the central storage location according to the configured collection criteria.

Bring files from multiple systems into a single central environment and connect subsequent tasks based on file type or analysis purpose.

Application ───┐
               │
Database ──────┼──→ Central Storage
               │           │
Server ────────┤           ├──→ Analysis
               │           │
Edge ──────────┘           └──→ Monitoring

Analysis Integration

Connect collected files to the next analysis and monitoring tasks

Once files are collected in the central environment, completed files can be used immediately by analysis systems or monitoring environments.

By using collection completion as the execution condition for the next task, you can configure the entire process after file transfer through analysis as a single workflow.

Collection Completed
        │
        ▼
   File Available
        │
   ┌────┴─────┐
   ▼          ▼
Analysis   Monitoring
   │          │
   └────┬─────┘
        ▼
   Result Tracking

Operational Review

Review collection status and execution results, then rerun required tasks

When a collection task runs, use Runs and detailed execution records to review file-processing status and results by system.

Review the collection environment, file paths, connection status, and processing results together. For tasks requiring additional review, take the necessary action based on the details and rerun the task.

Collection Run
      │
      ▼
Status Check
      │
      ├── Completed ─────→ Result Check
      │
      └── Review Required
               │
               ▼
          View Details
               │
               ▼
   Source / Path / Connection Check
               │
               ▼
          Run Again
               │
               ▼
          Result Check

Developers

Collect logs and core dumps by type and filter them before sending them to a central host

Integration Preparation

Prepare shared request code and path conventions

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())

Determine the transfer status using the values below. There are five terminal states, and the successful state is Complete (2).

Select Collection Targets

Select only the files to collect by extension and size

Log folders often contain files that do not need to be collected. Specify conditions to send only the required files.

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 {}

Use send-fileoption.extension for extension filters. The send-filetype-cus regular expression matches only filenames with the extension removed, so it does not work as an extension condition.

When multiple filters are provided, they are combined with AND. Only files that pass every filter are transferred.

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

Logs are often compressed to .gz during rotation, so specify both the original and compressed versions. Filter lock files using a name condition.

Verify in advance through a search that the conditions are being applied as intended.

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")

Register Recurring Collection

Collect general logs at a defined time

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))

There are four requirements that must always be followed when creating an automation request.

Registration succeeds even if all four items are omitted, but the behavior changes at execution time. If a recurring schedule was registered but runs only once, check isUpcoming first.

Diagnostic files must retain each run, so use numbering. With overwrite, rotating logs with the same name overwrite one another.

Register Immediate Collection

Collect core dumps immediately when they are created

Core dumps are needed immediately after an incident, so collect them as soon as they are created. Register a real-time watch automation by setting transferType to sync and placing syncType and watchFolderType in transferOptions.

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)

Put syncType and watchFolderType inside transferOptions. The watch path is read from sourceItem[0].filePath, so the plain-text path added by build_collection becomes the watch target.

The agent considers writing complete when file size stops changing and then sends the event. Large files take longer from detection to the event, which prevents incomplete files from being transferred.

Register Multiple Systems

Keep collection target servers in a list and register them in bulk

When there are dozens of servers, creating them one by one in the interface is difficult. Keep a list and iterate through it.

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))

Include the device identifier in the destination path. Since each server can have a file with the same name, such as application.log, collecting everything into one path makes the source indistinguishable.

/collect/
    device-app-01/
        application.log
    device-app-02/
        application.log

Analysis Integration

Call the analysis task after collection completes

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

Specify category and type, and put url, method, and body inside config.

The call occurs after transfer completion, and the receiving endpoint processes the request as follows.

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)

Review Collection Status

Review collection counts and failures by system

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}")

Diagnostic file collection is most needed during incidents, but collection itself can also fail at that moment. Monitor collection failures separately.