Data collectors

A collector turns what the equipment says (registers, files, HTTP responses) into records (structured values) or blobs (files) and puts them into the local queue. Sending them to Central is the job of the queue and the uploader, so collectors keep running the same way while the link is down.

You declare them under collectors: in edge.yaml. No code needed.

Common fields#

collectors:
  - type: http          # type (required)
    name: gateway-1     # name. The collector list on screen and the record's collector value
    enabled: true       # false skips it
    priority: 50        # 0-100. Default 50
    options: { ... }    # type-specific options

priority is read by both retention and upload. When a limit is exceeded, the lowest are dropped first; when sending, the highest go first. At sync.urgent_priority (default 90) and above, the upload window restriction is ignored too. Unless something is especially important, leave it at 50.

A single misconfigured collector is logged and skipped. The other collectors and communication with Central keep running. Collector state (running, disconnected, stopped) and the last error appear in the Collectors (수집기) card on the device detail page and in geo-mlops-edge status.

You can write options under options: or directly on the same level (both are accepted).

http — poll another server#

- type: http
  name: gateway-1
  options:
    url: http://10.0.0.7/api/current
    interval_ms: 1000
    timeout_s: 10
    verify_tls: true
    headers: { Authorization: "Bearer <your-secret>" }   # optional
    auth: { username: edge, password: <your-secret> }    # optional, HTTP Basic
    ts_field: measured_at      # the field holding the observation time, if the response has one
    max_body_bytes: 1MiB       # response size limit. 0 disables the check

The record content is {"collector": "gateway-1", "body": <response JSON>}.

The same value still makes one record every time. "It was still 21.5 at 12:00:01" is data too. So volume is predictable as interval × response size — reading 2 KiB every second is about 177 MB a day, and with the default retention limits (50 GiB / 30 days) the age limit is hit first.

Endpoints that return a list of past entries#

An endpoint that returns a list of past entries, such as "the last N alarms", sends rows you already have on every poll. Tell it the array location and the key, and each item is queued only once.

- type: http
  name: alarms
  priority: 70
  options:
    url: http://10.0.0.7/api/alarms
    interval_ms: 5000
    items_path: data.items
    id_field: id

There is one test: is this response the current state, or a list of past entries? For the current state, one record per poll is right. For a list of past entries, name the key with items_path and id_field.

The other server being down, slow or returning odd values is treated as normal. The collector logs the error, marks itself disconnected, and recovers on its own, stretching the interval from 1 second to 30 seconds.

push — an entry point for pushed records#

Use it when a robot controller, a vision PC or a service written in another language pushes records to the agent over the LAN.

- type: push
  name: robot-1
  priority: 60
  options:
    kind: robot

The sender PUTs with an id it chooses.

curl -X PUT http://edge-pc:8600/api/v1/collectors/robot-1/records/evt-1 \
     -H 'content-type: application/json' \
     -d '{"payload": {"step": 3}, "ts": "2026-09-16T01:02:03Z"}'

• 201 = stored, 200 + "duplicate": true = id already received. If the response is lost and you send again, only one record remains.
• ts is optional. Without it, the receive time is used.
• The YAML decides kind and priority. A sender cannot put data in under a kind nobody looks at.
• A declared entry point shows in the collector list with its last receive time. If the sender dies, not even an error arrives, so this time stopping is the only signal.

Pushing without a declaration#

For processes not in the config, POST /api/v1/records and POST /api/v1/blobs are also open. But the agent creates the id, so a retry makes two records, and they do not appear in the collector list.

modbus — read PLC registers#

Needs pip install 'geo-mlops-sdk[edge,modbus]'.

- type: modbus
  name: line-1
  priority: 50
  options:
    host: 10.0.0.5
    port: 502
    unit_id: 1
    interval_ms: 1000
    schema: /etc/geo-mlops/plc.yaml   # register map file (can also be written inline in the YAML)

The register map is plain YAML.

# /etc/geo-mlops/plc.yaml
version: "1"
unit_id: 1

fields:
  # type: holding | input | coil | discrete
  # dtype: bool | uint16 | int16 | uint32 | int32 | float32
  - { name: temperature.zone1, type: holding, address: 100, dtype: uint16, scale: 0.1 }
  - { name: temperature.zone2, type: holding, address: 101, dtype: uint16, scale: 0.1 }

  # 32-bit values take two registers. A wrong word_order gives plausible garbage instead of an error
  - { name: flow_rate, type: holding, address: 110, dtype: float32, word_order: big }
  - { name: cycle_count, type: holding, address: 112, dtype: uint32 }

  - { name: running, type: coil, address: 5 }
  - { name: fault, type: discrete, address: 12 }

Contiguous addresses are read in one go. The record content is {"collector": "line-1", "version": "1", "fields": {"temperature.zone1": 21.3, ...}}.

If the PLC is off, the collector is marked disconnected and reconnects at intervals from 1 second up to 30 seconds.

watchdir — files dropped in a folder#

Picks up files that cameras, lidars or legacy tools write into a folder. The files go to the Files (파일) tab.

- type: watchdir
  name: cam-0
  priority: 20
  options:
    path: /data/incoming
    pattern: "*.jpg"
    interval_s: 2
    delete_after: true

So as not to pick up a file that is still being written, it takes only files whose size stayed the same for one interval. The agent's account also needs write permission on that folder. If it cannot delete the original, it does not pick up the same file repeatedly, but the collector shows an error.

Custom collectors#

For equipment the four built-in types cannot handle (serial ports, vendor SDKs and so on), write your own collector and register it. register_collector must be called in the same process before the agent starts, so use a small launcher script instead of geo-mlops-edge run.

# my_edge.py
import asyncio
import sys
from datetime import datetime, timezone
from pathlib import Path

from geo_mlops_sdk.edge.collectors import CollectorBase, register_collector
from geo_mlops_sdk.edge.daemon import run
from geo_mlops_sdk.edge.settings import EdgeSettings


class CounterCollector(CollectorBase):
    type_name = "counter"          # type shown on screen

    def __init__(self, name, *, priority=50, interval_s=5.0):
        super().__init__(name, priority=priority)
        self.interval_s = interval_s
        self._task = None

    @classmethod
    def from_options(cls, *, name, priority, options):
        return cls(name, priority=priority,
                   interval_s=float(options.get("interval_s", 5)))

    async def start(self, sink):
        self.state = "running"
        self._task = asyncio.create_task(self._loop(sink))

    async def stop(self):
        if self._task:
            self._task.cancel()
        self.state = "stopped"

    async def _loop(self, sink):
        value = 0
        while True:
            value += 1
            now = datetime.now(timezone.utc)
            await sink.record("counter", {"value": value},
                              priority=self.priority, ts=now)
            self.note_emit(now)        # updates 'last received' on the collector card
            await asyncio.sleep(self.interval_s)


register_collector("counter", CounterCollector.from_options)

if __name__ == "__main__":
    config = Path(sys.argv[1]) if len(sys.argv) > 1 else None
    sys.exit(run(EdgeSettings.load(config)))

collectors:
  - type: counter
    name: counter-1
    options:
      interval_s: 15

python my_edge.py /etc/geo-mlops/edge.yaml

• The factory is called as factory(name=..., priority=..., options=...).
• sink.record(kind, payload, priority=, ts=, meta=, record_id=) queues a record, and sink.blob(kind, path|bytes, filename=, priority=, move=) queues a file. If you pass record_id, the same id goes in only once.
• You do not have to subclass CollectorBase; name, start(sink), stop() and status() are enough. If you do subclass it, note_emit() and note_error() fill in the collector state on screen correctly.
• Pass the exit code returned by run() (3 for a restart request) straight to sys.exit so systemd restarts it correctly.

The example above was run and verified. It shows up as counter-1 (counter) running in the Collectors (수집기) card on the device detail page, and as records like {"value": 8} in the Collected data (수집 데이터) tab.