DC

Source code: https://github.com/minipada/ros2_data_collection

Rolling

For detailed instructions, see the navigation sidebar, or browse doc/src/dc on GitHub. Security policy.

Introduction

The DC (Data Collection) project aims at integrating data collection pipelines into ROS 2. The goal is to integrate data collection pipelines with existing APIs to enable data analytics, rather than live monitoring, which already has excellent tools available. As companies increasingly turn to autonomous robots, the ability to understand and improve operations for any type of machine in any environment has become crucial. This involves mostly pick and drop and inspection operations. This framework aims at helping collecting, validating (through JSON schemas) and sending reliably the data to create such APIs and dashboards.

DC uses a modular approach, based on pluginlib and greatly inspired by Nav2 for its architecture. Pluginlib is used to configure which Measurements are collected. Data leaves the robot through the Bridge (dc_bridge), a thin ROS 2 node that renders and supervises an external Shipper, Vector: Vector is a fast, lightweight observability data pipeline, distributed as a single static binary, with native sinks for PostgreSQL, S3-compatible storage, and many more. DC gets its performance, reliability, and data integrity (backpressure handling and disk buffering) without embedding or forking it. Five Destination types are configured natively from ROS parameters; every other Vector sink is reachable by passing raw Shipper configuration through, with no DC code.

Why collect data from robots?

1. Performance Monitoring: Collecting data from a robot allows you to monitor its performance and identify areas for improvement. For example, you can use data to analyze the robot's motion and identify areas where it may be experiencing issues or inefficiencies.
2. Fault Diagnosis: Data collection can also be used to diagnose faults and troubleshoot issues with the robot. By collecting data on various aspects of the robot's behavior, you can identify patterns or anomalies that may indicate problems with the system.
3. Machine Learning: Data collected from robots can be used to train machine learning models, which can be used to improve the robot's performance and behavior. For example, you can use data collected from sensors to train models for object detection or path planning.
4. Research and Development: Data collection is important for research and development in robotics. By collecting data on the behavior of robots in different scenarios, researchers can gain insights into how robots can be designed and optimized for different applications.
5. Inventory Management: Data collection can be used to monitor inventory levels and track the movement of goods within a warehouse. This can help managers identify which products are in high demand and optimize the placement of products to improve order fulfillment times.
6. Resource Allocation: Data collection can also help managers allocate resources more efficiently. For example, by monitoring the movement of people and goods within a warehouse, managers can identify bottlenecks and areas of congestion and adjust staffing and equipment allocation to address these issues.
7. Process Improvement: Data collection can be used to monitor and analyze the performance of various processes within a warehouse. By identifying areas of inefficiency or errors, managers can develop strategies for improving these processes and increasing productivity.
8. Predictive Maintenance: Data collection can be used to monitor the performance of equipment and identify potential maintenance issues before they occur. This can help managers schedule maintenance more effectively and avoid costly downtime due to equipment failure.

Main features

• Open source: Currently all tools on the market are not open source. This project is in MPL-2.0 license, in summary you can use without asking permission and without paying
• Modular approach: based on pluginlib and greatly inspired by Nav2 for its architecture
• Reliable data collection: validate and send Records to create APIs and dashboards
• Flexible data collection: set polling interval for each Measurement or collect every Measurement with StringStamped messages
• Customizable validation: validate Records using existing or customized JSON schemas
• Easy to extend: add new Measurements by writing a plugin; add new Destinations with configuration alone
• Flexible data collection conditions: collect data based on conditions such as whether the robot is moving or if a field is equal to a value
• Condition-based data collection: collect data when a defined set of combination of all, any, or no condition are met
• Customizable record collection: configure the number of records to collect at the start and when a condition is activated.
• Data inspection: inspect data from camera input including barcode and QR codes
• Fast and efficient: high performance, using an external Shipper for delivery, and designed to minimize code duplication and reduce human errors
• Grouped Measurements: Records can be merged into Groups using the group node, based on the ApproximateTimeSynchronizer
• File uploads: Files — map_server maps, camera images, videos, anything a Measurement produces — are uploaded to object storage with verified, resumable transfers, and their metadata is recorded as a Record
• Easy to use: designed to be easy to learn and use
• No C++ 3rd party library required: all 3rd party libraries have a vendor package in the repository

And inherited from the Vector shipper:

• Backpressure handling
• Disk buffering, persisting Records across Destination outages and reboots

Here is an example of a pipeline, for an AGV doing pick-and-drop and inspection work:

Security

Found a vulnerability? Do not open a public issue — report it privately through GitHub private vulnerability reporting. The security policy covers the supported branches, the response targets, and what is in scope.

License

This program is under the terms of the Mozilla Public License Version 2.0.

About and Contact

For any inquiry, please contact David (d.bensoussan@proton.me). If your inquiry relates to bugs or open-source feature requests, consider posting a ticket on our GitHub project. If your inquiry relates to configuration support or private feature development, reach out and we will be able to support you in your projects.

Setup

DC 2.0 ships as published container images by default — podman compose or podman run, no ROS 2 toolchain to install locally. It is also an ordinary ROS 2 workspace if you'd rather build natively: rosdep install, colcon build, done — no forked shipper to compile, no Go toolchain, nothing needs root.

Containerized

The default, recommended way to run DC — nothing to build, nothing but Podman required.

Quick run

No build needed — the published :rolling images run the all-in-one shape (every ROS node, the Bridge and the Shipper in one dc-ros container) directly against a local Postgres, with RustFS available alongside it for a files/S3 Destination. Isolated network by default — dc-ros reaches the stores by container name, not the host's network — so the commands below run from deploy/robot/.

podman compose, all three containers, one command — the fastest path if you don't need to watch each container's own output separately:

podman compose -f compose.aio.yaml -f compose.isolated-network.yaml -f compose.local-destinations.yaml up

podman run, one step at a time — same three containers, split so each command can be run and checked before the next, in its own terminal:

1. Network and volumes:

   podman network create dc_robot_net
   podman volume create dc_robot_pgdata
   podman volume create dc_robot_rustfs_data
   podman volume create dc_robot_aio_buffer
   

2. Postgres, pinned by digest — bump deliberately, check https://hub.docker.com/_/postgres/tags?name=13 for a newer one:

   # PostgreSQL 13.23
   podman run --rm -it --network dc_robot_net --name dc_robot_postgres \
     -e POSTGRES_USER=dc -e POSTGRES_PASSWORD=password -e POSTGRES_DB=dc \
     -p 5432:5432 -v dc_robot_pgdata:/var/lib/postgresql/data \
     docker.io/library/postgres@sha256:4689940c683801b4ab839ab3b0a0a3555a5fe425371422310944e89eca7d8068
   

   Wait for database system is ready to accept connections before moving on.

3. RustFS, pinned by digest (matches compose.local-destinations.yaml — bump deliberately, check https://hub.docker.com/r/rustfs/rustfs/tags for the new one):

   # RustFS v1.0.0-beta.11
   podman run --rm -it --network dc_robot_net --name dc_robot_rustfs \
     -p 9000:9000 -v dc_robot_rustfs_data:/data \
     docker.io/rustfs/rustfs@sha256:84ce557a0245a06a9aae5516f55ee0f007fca78d41df356f419306fdc0cb168c
   

4. dc-ros — on the isolated network, 127.0.0.1 no longer reaches Postgres, so this needs a params override pointed at the hostname postgres instead of the image's own baked-in 127.0.0.1 default:

   podman run --rm -it --network dc_robot_net --name dc_robot_aio \
     -v dc_robot_aio_buffer:/root/.dc/buffer \
     -v "$(pwd)/params/aio_params_local.yaml:/opt/dc/dc_params.yaml:ro" \
     ghcr.io/minipada/ros2_data_collection/dc-ros:rolling \
     dc_params_file:=/opt/dc/dc_params.yaml
   

5. Check Records are landing, from a fourth terminal:

   psql -h 127.0.0.1 -U dc -d dc -c 'select * from dc order by date desc limit 5;'
   

   Password password.

See deploy/robot/README.md for the three-container split topology, the host-network variant, and trying it against a real store before fleet rollout.

Building the workspace image

The repository builds a full workspace image with Podman — the same one CI uses:

IMAGE_TAG=dc-workspace:local ./tools/e2e/scripts/build.sh

tools/e2e/scripts/test.sh runs colcon test against that image, and tools/e2e/scripts/run.sh drives the zero-loss end-to-end harness. See tools/e2e/README.md.

Deployment renderings and a local Kubernetes loop

deploy/robot/ describes the three-container robot tier (dc-ros, vector, dc-uploader — see Deployment modes) as Compose, Podman Quadlet and Kubernetes manifests, for whichever a site already runs. For iterating on the Kubernetes rendering itself, a loop of plain podman build, k3d and kubectl commands brings up a disposable local cluster in seconds — no wrapper script, no registry, just the commands themselves. See deploy/robot/README.md for the full command sequence and what each step is for.

Native

For developing DC itself, or wherever containers aren't an option.

Requirements

• ROS 2 Rolling (ros-rolling-ros-base or larger), on Ubuntu 26.04 or a Debian equivalent
• colcon, rosdep, git, vcstool (python3-vcstool), a C++17 compiler
• x86-64 or aarch64 — the architectures vector_vendor has a pinned Vector binary for

Build

1. Clone into a workspace:

   mkdir -p ~/ws/src && cd ~/ws/src
   git clone https://github.com/minipada/ros2_data_collection.git
   

2. Pull in vector_vendor, aws_sdk_vendor and nlohmann_json_schema_validator_vendor (each its own repo — see ADR-0002's amendment, ADR-0012 and ros2_data_collection.repos), register DC's local rosdep rules (header-only C++ libraries upstream rosdistro has no key for), then resolve dependencies:

   cd ~/ws
   vcs import src < src/ros2_data_collection/ros2_data_collection.repos
   echo "yaml file://$PWD/src/ros2_data_collection/rosdep/dc.yaml" \
     | sudo tee /etc/ros/rosdep/sources.list.d/10-dc.list
   rosdep update
   rosdep install --from-paths src --ignore-src -r -y
   

3. Build:

   source /opt/ros/rolling/setup.bash
   colcon build
   

That is the whole install. colcon build also runs vector_vendor, which fetches a pinned, checksummed Vector release tarball live — the external Shipper the Bridge supervises at runtime (ADR-0002) — and aws_sdk_vendor, which fetches and builds the AWS SDK for C++ (core + s3) dc_uploader uses (ADR-0014) live from github.com/aws/aws-sdk-cpp at a pinned tag; both steps need network access (ADR-0002, ADR-0012), and aws_sdk_vendor's takes several minutes the first time.

Python dependencies

Only some Measurement plugins (camera inspection, QR code detection) need Python packages beyond what ROS 2 installs. rosdep covers the ones with rosdistro keys; for the rest, uv installs pyproject.toml's pins into a project virtualenv:

uv sync --no-dev   # drop --no-dev to add the tooling and the demo dashboard's packages

Run

source install/setup.bash
ros2 launch dc_bringup dc_bringup.launch.py

The default parameters file (dc_bringup/params/dc_params.yaml) collects uptime and writes it to a local PostgreSQL Destination. To run your own:

ros2 launch dc_bringup dc_bringup.launch.py dc_params_file:=/path/to/my_params.yaml

See Configuration examples for configurations you can copy, and Destinations for the full Bridge configuration contract.

Useful launch arguments

What starts, in what order

dc_bringup.launch.py brings the pipeline up deterministically (ADR-0006): the Bridge and its Shipper first, then a readiness gate, and only then the collection nodes. If the Shipper never becomes ready, the launch shuts down loudly instead of collecting data nowhere. Data Pipeline describes this in full.

Advanced build options

colcon build --cmake-args -Dvector_path=/usr/bin/vector

Infrastructure

DC delivers to systems you run yourself. tools/infrastructure/docker/ has compose files that bring up PostgreSQL and RustFS (S3-compatible object storage) preconfigured for the demos; see Infrastructure setup.

Issues

If you run into problems building DC, search the issue tracker on GitHub and feel free to open a ticket.

Requirements

Requirements are tracked using Strictdoc.

The goal is to list tasks and do test based requirements.

Demos

We will go together through some demos to get started with DC. You shall find them in the dc_demos package

The demos are grouped in three tiers. Start at Beginner and work down: each demo assumes the concepts explained in the ones before it.

How the tiers work

A demo's tier is the heavier of the two things it asks of you: the infrastructure you have to run alongside DC, and the DC machinery you have to understand or write. Neither axis alone is enough — Custom plugin needs no infrastructure at all but has you writing a C++ Measurement plugin, and MCAP recording is a single ros2 launch away but is built on the ADR-0003 passthrough.

When adding a demo, find the heaviest thing it asks of the reader — a service to stand up, or code to write — and file it under the matching tier.

Beginner

Everything prints to your terminal: nothing to install, start or clean up afterwards, and nothing to go and look at in another tool.

Prerequisites: a built and sourced workspace, and nothing else — no containers, no databases. Roughly 5 minutes each, 15 for the Turtlebot3 one including simulator startup.

Uptime to stdout

This is the most minimal example to run DC, it collects the system uptime every 5 seconds and sends it to Stdout.

Copy the passthrough sink into place, then run it:

mkdir -p ~/.dc && cp "$(ros2 pkg prefix dc_demos)/share/dc_demos/config/uptime_stdout_sink.toml" ~/.dc/
ros2 launch dc_demos uptime_stdout.launch.py

At the end, the data is displayed. Every Destination — the passthrough console sink below included — goes through the external Vector Shipper (ADR-0002, Destinations), so this is Vector's own event object after ingesting the Record over the Fluent-forward protocol: source_type, tag, host and timestamp are Vector's, not the Bridge's:

[dc_bridge-2] {"custom_keys":["robot_name","time"],"date":1788476609.152106,"flattened":false,"host":"127.0.0.1","name":"uptime","nested":false,"robot_name":"C3PO","run_id":"169","source_type":"fluent","tag":"dc.measurement.uptime","time":1608993,"timestamp":"2026-09-03T23:03:29.152105868Z"}
[dc_bridge-2] {"custom_keys":["robot_name","time"],"date":1788476614.1494331,"flattened":false,"host":"127.0.0.1","name":"uptime","nested":false,"robot_name":"C3PO","run_id":"169","source_type":"fluent","tag":"dc.measurement.uptime","time":1608998,"timestamp":"2026-09-03T23:03:34.149433151Z"}
[dc_bridge-2] {"custom_keys":["robot_name","time"],"date":1788476619.149286,"flattened":false,"host":"127.0.0.1","name":"uptime","nested":false,"robot_name":"C3PO","run_id":"169","source_type":"fluent","tag":"dc.measurement.uptime","time":1609003,"timestamp":"2026-09-03T23:03:39.149286154Z"}

This launchfile is a wrapper of dc_bringup/launch/dc_bringup.launch.py which loads a custom yaml configuration

Configuration

Measurement

measurement_server:
  ros__parameters:
    measurement_plugins: ["uptime"]
    uptime:
      plugin: "dc_measurements/Uptime"
      topic_output: "/dc/measurement/uptime"
      polling_interval: 5000
      enable_validator: true
      debug: true
      init_collect: true
    custom_key_str_list: ["robot_name", "id"]
    custom_keys_str:
      robot_name:
        name: robot_name
        value: C3PO
      id:
        name: id
        value_from_file: /etc/machine-id
    run_id:
      enabled: true
      counter: true
      counter_path: "$HOME/run_id"
      uuid: false

measurement_plugins (Mandatory): List all the plugins to enable. This is a custom string that is equal to the measurement plugin dictionary present in the same level. If not listed, will not be loaded.

uptime.plugin (Mandatory): Name of the plugin, if you are not sure which plugin is available, use the CLI tool to list them

uptime.polling_interval (Optional): Interval to which data is collected in milliseconds

uptime.enable_validator (Optional): Will validate the data against a JSON schema. This file is located in the dc_measurements package. You can provide your own using the json_schema_path parameter, which we will explore later on

uptime.debug (Optional): More verbose output

uptime.init_collect (Optional): Collect when the node starts instead of waiting for the polling_interval time to pass

run_id.enabled (Optional): Identify which run the robot is. A new one is generated at every start of the node. Uses either a counter that increment at each restart of the node or UUID

run_id.counter (Optional): Enable counter for the run_id

run_id.counter_path (Optional): Path to store the last run. It is expanded with environment variables id

run_id.uuid (Optional): Generate a new run ID by using a random UUID

This will collect the uptime every 5 seconds (including when the node starts), will forward it to the console passthrough sink below.

Inject custom data for each record

Here, we want to append some content in every record: the robot name and its ID. While the robot name comes from a fixed variable in the parameter file, the id comes from the machine-id file.

custom_key_str_list (Optional): Look for those keys in this configuration to add them as keys and values in each record.

custom_keys_str.robot_name (Optional): This parameter is loaded since it is mentioned in custom_key_str_list

custom_keys_str.robot_name.name (Optional): Key in the dictionary to add

custom_keys_str.robot_name.value (Optional): Value associated to the key in the dictionary to add

custom_keys_str.id.name (Optional): Key in the dictionary to add

custom_keys_str.id.value_from_file (Optional): Value associated to the key in the dictionary to add taken from the content of a file

Find the complete measurements documentation here

Destination

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["records_log"]
    records_log:
      type: file
      receives: records
      inputs: ["/dc/measurement/uptime"]
      path: "/tmp/dc/uptime_stdout_records.ndjson"
      time_key: "date"
      time_format: "double"
    custom_config_files: ["$HOME/.dc/uptime_stdout_sink.toml"]
    vector_forward_host: "127.0.0.1"
    vector_forward_port: 24224

# ~/.dc/uptime_stdout_sink.toml
[sinks.debug_console]
type = "console"
inputs = ["dc.dc.measurement.uptime"]
target = "stdout"

[sinks.debug_console.encoding]
codec = "json"

Destinations

Let's analyze piece by piece. dc_bridge is the single C++ node that owns every Destination; each entry in the destinations list names a section, defined below it, that describes where data goes. We need the topic list on each Destination because the Bridge subscribes to those topics itself and forwards what it receives to an external Vector process over the shipper ingest protocol.

destinations (Mandatory): List all the Destinations to enable. Each name must have a matching section at the same level.

records_log.type (Mandatory): One of the blessed Destination types (postgres, s3, file, console, vector). file writes each Record as a JSON line to path, and is the cheapest anchor to give destinations when the actual output you want comes from a custom_config_files passthrough sink below — dc_bridge derives its ROS subscriptions and dc.<tag> routes from destinations alone, never from a passthrough snippet's inputs. (Printing straight to stdout used a blessed console Destination in earlier versions of this demo; per ADR-0003, console — along with postgres and s3 — has moved to the passthrough recipe below, since it's a pure Vector-sink wrapper with no DC-specific logic. See Destinations: Recipes.)

records_log.receives (Optional): records (default) or files.

records_log.inputs (Mandatory): Topics to which to listen to get the data.

records_log.path (Mandatory for file): Absolute path Vector writes each JSON line to (Vector, not the Bridge, expands this — no $HOME).

records_log.time_format (Optional): Format the data's timestamp will be printed as (epoch_nanos (default), iso8601 or double).

records_log.time_key (Optional): Dictionary key the timestamp is written under.

custom_config_files (Optional): Raw Vector config snippets, merged as-is alongside what dc_bridge itself renders. The uptime_stdout_sink.toml above defines a plain Vector console sink consuming the public dc.dc.measurement.uptime route that records_log's inputs created — this is what actually prints to stdout; records_log itself just writes the same Records to disk as the passthrough's required anchor. See Destinations: Passthrough.

dc_bridge itself needs no engine tuning of the kind the old embedded Fluent Bit shipper required (buffering, scheduler backoff, HTTP stats server, …) — Vector, the external shipper process it forwards to, owns its own on-disk buffering and is configured from the dc_bridge/destinations block above; see ADR-0002 for why that split exists.

Inject run id at each record

Finally, we set the run id. This is used later on when fetching data for a run. It can come from a counter which is incremented at each start of the node or from a random UUID generated. The counter mechanism writes and read on a file on the system (take care of not deleting it), you can set its path as a parameter.

Find the complete destinations documentation here

Console output

Now that the node started, let us see what's displayed in the console.

Measurement server and dc_bridge are started in the Lifecycle, you can read more about it here. Per ADR-0006, the lifecycle manager waits on a bridge_ready_gate before activating the collection nodes:

[INFO] [bridge_ready_gate-3]: process started with pid [31]
[dc_bridge-2] [INFO] [1788478636.751773322] [dc_bridge]: dc_bridge up: 1 subscribed topic(s), supervising /root/ws/install/vector_vendor/lib/vector_vendor/vector
[bridge_ready_gate-3] [INFO] [1788478637.212874385] [bridge_ready_gate]: Bridge is ready: vector is accepting connections
[INFO] [bridge_ready_gate-3]: process has finished cleanly [pid 31]
[INFO] [launch.user]: dc_bridge reports ready; activating collection nodes.

dc_bridge renders the destinations block above into a Vector config and launches (or reloads) the external Vector process pointed at it; bridge_ready_gate only lets the launch continue once Vector is actually accepting connections — see ADR-0002 for why Vector runs as its own process rather than embedded in the Bridge.

Finally, we see the data, now printed by the passthrough console sink rather than by the Bridge itself:

[dc_bridge-2] {"custom_keys":["robot_name","time"],"date":1788476609.152106,"flattened":false,"host":"127.0.0.1","name":"uptime","nested":false,"robot_name":"C3PO","run_id":"169","source_type":"fluent","tag":"dc.measurement.uptime","time":1608993,"timestamp":"2026-09-03T23:03:29.152105868Z"}
[dc_bridge-2] {"custom_keys":["robot_name","time"],"date":1788476614.1494331,"flattened":false,"host":"127.0.0.1","name":"uptime","nested":false,"robot_name":"C3PO","run_id":"169","source_type":"fluent","tag":"dc.measurement.uptime","time":1608998,"timestamp":"2026-09-03T23:03:34.149433151Z"}
[dc_bridge-2] {"custom_keys":["robot_name","time"],"date":1788476619.149286,"flattened":false,"host":"127.0.0.1","name":"uptime","nested":false,"robot_name":"C3PO","run_id":"169","source_type":"fluent","tag":"dc.measurement.uptime","time":1609003,"timestamp":"2026-09-03T23:03:39.149286154Z"}

So...what happened?

1. The measurement plugin starts publishing data to /dc/measurement/uptime, which contains the JSON and timestamp of the message
2. Run ID and robot_name is appended in the JSON
3. dc_bridge, which subscribes to this topic directly, receives the data and forwards it to Vector over the shipper ingest protocol
4. Vector's generated config applies a remap transform that writes the configured time_key in the requested time_format, and routes the Record onto its public dc.dc.measurement.uptime route
5. The passthrough console sink from uptime_stdout_sink.toml, consuming that route, prints the JSON to stdout — records_log's own file sink writes the same Record to disk in parallel

Group memory and uptime

This demo will introduce the group node. It subscribes to multiple nodes and group for each its data and republishes on a new topic.

Let's run it:

ros2 launch dc_demos group_memory_uptime_stdout.launch.py

[dc_bridge-3] {"date":1788476662.509069,"host":"127.0.0.1","memory":{"flattened":false,"name":"memory","nested":false,"run_id":"169","used":95.77983856201172},"name":"memory_uptime","source_type":"fluent","tag":"dc.group.memory_uptime","tags":[""],"timestamp":"2026-09-03T23:04:22.509068935Z","uptime":{"flattened":false,"name":"uptime","nested":false,"run_id":"169","time":1609047}}
[dc_bridge-3] {"date":1788476665.5074596,"host":"127.0.0.1","memory":{"flattened":false,"name":"memory","nested":false,"run_id":"169","used":95.87796020507812},"name":"memory_uptime","source_type":"fluent","tag":"dc.group.memory_uptime","tags":[""],"timestamp":"2026-09-03T23:04:25.507459531Z","uptime":{"flattened":false,"name":"uptime","nested":false,"run_id":"169","time":1609050}}
[dc_bridge-3] {"date":1788476668.5055394,"host":"127.0.0.1","memory":{"flattened":false,"name":"memory","nested":false,"run_id":"169","used":96.04102325439453},"name":"memory_uptime","source_type":"fluent","tag":"dc.group.memory_uptime","tags":[""],"timestamp":"2026-09-03T23:04:28.505539347Z","uptime":{"flattened":false,"name":"uptime","nested":false,"run_id":"169","time":1609053}}

This launchfile is a wrapper of dc_bringup/launch/dc_bringup.launch.py which loads a custom yaml configuration

Configuration

Measurement

We collect data from 2 plugins: memory and uptime. The first every second and the latter every 3.

measurement_server:
  ros__parameters:
    measurement_plugins: ["memory", "uptime"]
    memory:
      plugin: "dc_measurements/Memory"
      group_key: "memory"
      topic_output: "/dc/measurement/memory"
      polling_interval: 1000
    uptime:
      plugin: "dc_measurements/Uptime"
      group_key: "uptime"
      topic_output: "/dc/measurement/uptime"
      polling_interval: 3000

Data is now published on 2 ROS topics: /dc/measurement/uptime and /dc/measurement/memory.

The group_key mentioned will be used by the group node to assign a key in the new dictionary

Group

This create a memory_uptime group, subscribes to /dc/measurement/memory and /dc/measurement/uptime topics and republish the result on /dc/group/memory_uptime. The sync_delay allows to wait in a 5 seconds window timeframe the data from each topic before throwing away the data if one topic does not publish it

group_server:
  ros__parameters:
    groups: ["memory_uptime"]
    memory_uptime:
      inputs: ["/dc/measurement/memory", "/dc/measurement/uptime"]
      output: "/dc/group/memory_uptime"
      sync_delay: 5.0
      group_key: "memory_uptime"

include_group_name (Optional, default true): Includes the name of the group in the JSON as a top-level name field — visible in the console output below — which makes it easier later on to fetch the data from your API. Left at its default here.

You can also notice that the group also has a "group_key". It means a group can be part of another.

Destination

Here, we only subscribe to the /dc/group/memory_uptime topic

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["console"]
    console:
      type: console
      receives: records
      inputs: ["/dc/group/memory_uptime"]
      time_key: "date"
      time_format: "double"
    vector_forward_host: "127.0.0.1"
    vector_forward_port: 24224

Console output

In the terminal, you can see the result, published every 3 seconds (see the date field), which the is timeframe defined by sync_delay and the maximum polling_interval of the measurements.

Finally, note the new dictionary uses the key defined in the group_key measurement_server plugin configuration. They are transferred through the ROS message.

[dc_bridge-3] {"date":1788476662.509069,"host":"127.0.0.1","memory":{"flattened":false,"name":"memory","nested":false,"run_id":"169","used":95.77983856201172},"name":"memory_uptime","source_type":"fluent","tag":"dc.group.memory_uptime","tags":[""],"timestamp":"2026-09-03T23:04:22.509068935Z","uptime":{"flattened":false,"name":"uptime","nested":false,"run_id":"169","time":1609047}}
[dc_bridge-3] {"date":1788476665.5074596,"host":"127.0.0.1","memory":{"flattened":false,"name":"memory","nested":false,"run_id":"169","used":95.87796020507812},"name":"memory_uptime","source_type":"fluent","tag":"dc.group.memory_uptime","tags":[""],"timestamp":"2026-09-03T23:04:25.507459531Z","uptime":{"flattened":false,"name":"uptime","nested":false,"run_id":"169","time":1609050}}
[dc_bridge-3] {"date":1788476668.5055394,"host":"127.0.0.1","memory":{"flattened":false,"name":"memory","nested":false,"run_id":"169","used":96.04102325439453},"name":"memory_uptime","source_type":"fluent","tag":"dc.group.memory_uptime","tags":[""],"timestamp":"2026-09-03T23:04:28.505539347Z","uptime":{"flattened":false,"name":"uptime","nested":false,"run_id":"169","time":1609053}}

Turtlebot3

In this example, we add a robot and start collecting robot data to Stdout.

You will also need 2 terminal windows, to:

1. Run the Nav2 turtlebot3 launchfile: it starts localization, navigation and RViz
2. Run DC

Since RViz is pretty verbose, using 2 terminal windows will help reading the JSON printed on the terminal window.

Setup the environment

In each, terminal, source your environment and setup turtlebot configuration:

source /opt/ros/rolling/setup.bash
source install/setup.bash

Nothing else has to be exported. Nav2's own nav2_minimal_tb3_sim ships the world, the robot and its ros_gz_bridge config, and puts them on GZ_SIM_RESOURCE_PATH itself — the Gazebo Classic GAZEBO_MODEL_PATH and TURTLEBOT3_MODEL variables are gone along with Classic.

Start Navigation

Then, start the Turtlebot launchfile:

ros2 launch nav2_bringup tb3_simulation_launch.py headless:=False

RViz and Gazebo will start: you should now see the robot in Gazebo, and the map on RViz.

Set the robot position using the "2D Pose Estimate" button.

Start DC

Execute

ros2 launch dc_demos tb3_simulation_stdout.launch.py

At the end, the data is displayed. Every Destination — console included — goes through the external Vector Shipper (ADR-0002, Destinations), so each line below is Vector's own event object, one bare JSON object per line (not an array):

[dc_bridge-3] {"custom_keys":["robot_name","id"],"date":1788479117.6297202,"flattened":false,"height":384,"host":"127.0.0.1","id":"e110a88ba1c24602bd2c116daf5b8287","local_paths":{"pgm":"/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.pgm","png":"/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.png","yaml":"/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.yaml"},"name":"map","nested":false,"origin":{"x":-10,"y":-10},"resolution":0.05000000074505806,"robot_name":"C3PO","run_id":"170","source_type":"fluent","tag":"dc.measurement.map","timestamp":"2026-09-03T23:45:17.629720410Z","width":384}
[dc_bridge-3] {"cmd_vel":{"angular":{"x":0,"y":0,"z":0.0299867},"computed":0.42023804783821106,"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","linear":{"x":0.420238,"y":0,"z":0},"name":"cmd_vel","nested":false,"robot_name":"C3PO","run_id":"170"},"date":1788479127.1009243,"host":"127.0.0.1","name":"robot","position":{"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","name":"position","nested":false,"robot_name":"C3PO","run_id":"170","x":-1.0834591164924419,"y":0.5133007443427651,"yaw":-0.008814692701341621},"source_type":"fluent","speed":{"angular":{"x":0,"y":0,"z":0.027356281873875128},"computed":0.33021257209388344,"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","linear":{"x":0.33021257209388344,"y":0,"z":0},"name":"speed","nested":false,"robot_name":"C3PO","run_id":"170"},"tag":"dc.group.robot","tags":[""],"timestamp":"2026-09-03T23:45:27.100924197Z"}

Given the JSON is quite large, let's analyze 2 different records:

The first one being the data published on the robot group. Note it's deeply nested now — each of cmd_vel, position and speed carries its own copy of the custom keys (robot_name, id, run_id), since those are applied per-measurement before the group node merges them, not once at the top level:

{
  "cmd_vel": {
    "angular": {
      "x": 0,
      "y": 0,
      "z": 0.0299867
    },
    "computed": 0.42023804783821106,
    "custom_keys": [
      "robot_name",
      "id"
    ],
    "flattened": false,
    "id": "e110a88ba1c24602bd2c116daf5b8287",
    "linear": {
      "x": 0.420238,
      "y": 0,
      "z": 0
    },
    "name": "cmd_vel",
    "nested": false,
    "robot_name": "C3PO",
    "run_id": "170"
  },
  "date": 1788479127.1009243,
  "host": "127.0.0.1",
  "name": "robot",
  "position": {
    "custom_keys": [
      "robot_name",
      "id"
    ],
    "flattened": false,
    "id": "e110a88ba1c24602bd2c116daf5b8287",
    "name": "position",
    "nested": false,
    "robot_name": "C3PO",
    "run_id": "170",
    "x": -1.0834591164924419,
    "y": 0.5133007443427651,
    "yaw": -0.008814692701341621
  },
  "source_type": "fluent",
  "speed": {
    "angular": {
      "x": 0,
      "y": 0,
      "z": 0.027356281873875128
    },
    "computed": 0.33021257209388344,
    "custom_keys": [
      "robot_name",
      "id"
    ],
    "flattened": false,
    "id": "e110a88ba1c24602bd2c116daf5b8287",
    "linear": {
      "x": 0.33021257209388344,
      "y": 0,
      "z": 0
    },
    "name": "speed",
    "nested": false,
    "robot_name": "C3PO",
    "run_id": "170"
  },
  "tag": "dc.group.robot",
  "tags": [
    ""
  ],
  "timestamp": "2026-09-03T23:45:27.100924197Z"
}

This record contains the speed, cmd_vel and position from the group "robot".

{
  "custom_keys": [
    "robot_name",
    "id"
  ],
  "date": 1788479117.6297202,
  "flattened": false,
  "height": 384,
  "host": "127.0.0.1",
  "id": "e110a88ba1c24602bd2c116daf5b8287",
  "local_paths": {
    "pgm": "/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.pgm",
    "png": "/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.png",
    "yaml": "/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.yaml"
  },
  "name": "map",
  "nested": false,
  "origin": {
    "x": -10,
    "y": -10
  },
  "resolution": 0.05000000074505806,
  "robot_name": "C3PO",
  "run_id": "170",
  "source_type": "fluent",
  "tag": "dc.measurement.map",
  "timestamp": "2026-09-03T23:45:17.629720410Z",
  "width": 384
}

This record contains the map data from the measurement. There's no remote_paths key — this demo's dc_bridge block only declares the console Destination below, not a rustfs one, so nothing actually matches map.remote_keys: ["rustfs"] and the Bridge never populates it.

Configuration

Measurement

measurement_server:
  ros__parameters:
    custom_keys_str: ["robot_name"]
    robot_name: "C3PO"
    measurement_plugins: ["cmd_vel", "map", "position", "speed"]
    run_id:
      enabled: true
      counter: true
      counter_path: "$HOME/run_id"
      uuid: false
    save_local_base_path: "$HOME/dc_data/"
    all_base_path: "=robot_name/%Y/%m/%d/%H"
    cmd_vel:
      plugin: "dc_measurements/CmdVel"
      group_key: "cmd_vel"
      enable_validator: true
      topic_output: "/dc/measurement/cmd_vel"
    position:
      plugin: "dc_measurements/Position"
      group_key: "position"
      topic_output: "/dc/measurement/position"
      polling_interval: 1000
      enable_validator: true
      init_collect: true
      global_frame: "map"
      robot_base_frame: "base_link"
      transform_timeout: 0.1
    speed:
      plugin: "dc_measurements/Speed"
      group_key: "speed"
      odom_topic: "/odom"
      topic_output: "/dc/measurement/speed"
    map:
      plugin: "dc_measurements/Map"
      group_key: "map"
      polling_interval: 5000
      save_path: "map/%Y-%m-%dT%H:%M:%S"
      topic_output: "/dc/measurement/map"
      save_map_timeout: 4.0
      remote_prefixes: [""]
      remote_keys: ["rustfs"]

save_local_base_path (Optional): Used as a common base for all saved files from measurement plugins. all_base_path is concatenated to it afterwards for defining the path where files are saved.

all_base_path (Optional): Used as a common base for some measurements to save files. Is concatenated to save_local_base_path. Note the =robot_name, which is later replaced by C3PO (the variable defined in custom_keys_str)

map.remote_keys: creates a dictionary inside remote_paths which is named by the strings in this field — each name must match a receives: files Destination in the dc_bridge block below, so dc_uploader (a separate process from the Bridge, see ADR-0014) knows where to send the file.

Group

group_server:
  ros__parameters:
    groups: ["robot"]
    robot:
      inputs:
        [
          "/dc/measurement/cmd_vel",
          "/dc/measurement/position",
          "/dc/measurement/speed",
        ]
      output: "/dc/group/robot"
      sync_delay: 5.0
      group_key: "robot"
      include_group_name: true

Create a group with data from cmd_vel, position and speed. Even though it appears there is nothing new here, I shall like to precise something important. In the previous demo, we mentioned that if all messages are not received by the group, it will drop it. It matters in this case because cmd_vel is not published all the time in this example (not when it is not moving), this means the data will be collected only when the robot moves (when a controller sends a command).

If you wished to collect the position and the speed constantly, you could take cmd_vel out of this group and add it in the destination.

Destinations

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["console"]
    console:
      type: console
      receives: records
      inputs: ["/dc/group/robot", "/dc/measurement/map"]
      time_key: "date"
      time_format: "double"
    vector_forward_host: "127.0.0.1"
    vector_forward_port: 24224

measurement_server:
  ros__parameters:
    custom_key_str_list: ["robot_name", "id"]
    custom_keys_str:
      robot_name:
        name: robot_name
        value: "C3PO"
      # Requires systemd package
      id:
        name: id
        value_from_file: /etc/machine-id

Nothing new here, we simply edited the console Destination's inputs to ["/dc/group/robot", "/dc/measurement/map"] to get the data from the robot group and the map.

Console output

Now that the node started, let us see what's displayed in the console. Measurement server and dc_bridge are started in the Lifecycle, you can read more about it here.

"Base save path" and "All Base path" are also saved and expanded. Note "=robot_name" has been replaced by C3PO. measurement_server runs composed inside the same process as every other DC node here, so the prefix is the container's, not the node's own name:

[component_container_isolated-1] [INFO] [1788479023.671013064] [measurement_server]: Base save path expanded to /root/dc_data/
[component_container_isolated-1] [INFO] [1788479023.671095239] [measurement_server]: All Base path expanded to C3PO/%Y/%m/%d/%H

Once dc_bridge reports ready (per ADR-0006's bridge_ready_gate), the measurement plugins and the "robot" group start publishing, and we see the data on Vector's console sink:

[dc_bridge-3] {"custom_keys":["robot_name","id"],"date":1788479117.6297202,"flattened":false,"height":384,"host":"127.0.0.1","id":"e110a88ba1c24602bd2c116daf5b8287","local_paths":{"pgm":"/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.pgm","png":"/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.png","yaml":"/root/dc_data/C3PO/2026/09/03/23/map/2026-09-03T23:45:17.yaml"},"name":"map","nested":false,"origin":{"x":-10,"y":-10},"resolution":0.05000000074505806,"robot_name":"C3PO","run_id":"170","source_type":"fluent","tag":"dc.measurement.map","timestamp":"2026-09-03T23:45:17.629720410Z","width":384}
[dc_bridge-3] {"cmd_vel":{"angular":{"x":0,"y":0,"z":0.0131854},"computed":0.17549346387386322,"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","linear":{"x":0.175493,"y":0,"z":0},"name":"cmd_vel","nested":false,"robot_name":"C3PO","run_id":"170"},"date":1788479126.1020856,"host":"127.0.0.1","name":"robot","position":{"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","name":"position","nested":false,"robot_name":"C3PO","run_id":"170","x":-1.1893664880015578,"y":0.5148037648787771,"yaw":-0.02119602699838526},"source_type":"fluent","speed":{"angular":{"x":0,"y":0,"z":0},"computed":0,"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","linear":{"x":0,"y":0,"z":0},"name":"speed","nested":false,"robot_name":"C3PO","run_id":"170"},"tag":"dc.group.robot","tags":[""],"timestamp":"2026-09-03T23:45:26.102085677Z"}
[dc_bridge-3] {"cmd_vel":{"angular":{"x":0,"y":0,"z":0.0299867},"computed":0.42023804783821106,"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","linear":{"x":0.420238,"y":0,"z":0},"name":"cmd_vel","nested":false,"robot_name":"C3PO","run_id":"170"},"date":1788479127.1009243,"host":"127.0.0.1","name":"robot","position":{"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","name":"position","nested":false,"robot_name":"C3PO","run_id":"170","x":-1.0834591164924419,"y":0.5133007443427651,"yaw":-0.008814692701341621},"source_type":"fluent","speed":{"angular":{"x":0,"y":0,"z":0.027356281873875128},"computed":0.33021257209388344,"custom_keys":["robot_name","id"],"flattened":false,"id":"e110a88ba1c24602bd2c116daf5b8287","linear":{"x":0.33021257209388344,"y":0,"z":0},"name":"speed","nested":false,"robot_name":"C3PO","run_id":"170"},"tag":"dc.group.robot","tags":[""],"timestamp":"2026-09-03T23:45:27.100924197Z"}

So...what happened?

1. The Nav2 turtlebot3 simulation starts, a robot is able to localize and move (once you use the 2-D pose estimate on RViz)
2. The measurement plugins start publishing data to /dc/measurement/map, /dc/measurement/cmd_vel, /dc/measurement/position and /dc/measurement/speed, which contain the JSON and timestamp of the message
3. In parallel, each time the map plugin sends a ROS message, it also saves the files on the filesystem. Open a file browser to the path you set in the configuration to a path mentioned in the map JSON
4. The "robot" group node subscribes to /dc/measurement/cmd_vel, /dc/measurement/position and /dc/measurement/speed and publish on /dc/group/robot when it collects data from all 3 topics
5. Run ID and robot_name is appended in the JSON of each
6. dc_bridge, which subscribes to /dc/group/robot and /dc/measurement/map directly, receives the data and forwards it to the external Vector process over the shipper ingest protocol
7. Vector's generated config applies a remap transform that writes the configured time_key in the requested time_format
8. Vector's console sink, the only one matching the console Destination we configured, prints the JSON to stdout

Intermediate

The Records leave the terminal: they land in a database, a search index, a bucket or a file, and you go and read them back there. This is the tier where Destinations — blessed and passthrough — are introduced.

Prerequisites: the Beginner tier, plus the one service each demo sends to, started from tools/infrastructure/docker/ as described on its Infrastructure setup page. The two Turtlebot3 demos also need the AWS warehouse world on top of the simulator. Roughly 20 to 30 minutes each, plus a one-off container image pull the first time you bring a stack up.

MCAP recording (passthrough)

dc_bridge blesses exactly five Destination types — postgres, s3, file, console, vector (see Destinations) — and MCAP is not one of them; Vector, the Shipper, has no MCAP sink at all. This tutorial is the worked example for #210: the ADR-0003 passthrough plus a small standalone process, dc_mcap_writer (ADR-0009), that consumes the same public dc.<tag> routes a blessed Destination consumes and writes them as rotated .mcap files, ready to open with ros2 bag info or Foxglove.

Unlike the Elasticsearch/InfluxDB passthrough demos, there is no raw Vector TOML to hand-author and no second terminal to start a companion process in: dc_bringup.launch.py reads a dc_mcap_writer: block from the same params file every other node's parameters live in, generates the passthrough sink from it, and starts dc_mcap_writer itself automatically. From the params file it looks like configuring a Destination — one block, inputs, done — even though it structurally isn't one (see "Understanding the configuration" below for why).

Run it

Hardware-free: four system Measurements, one dc_mcap_writer: block in the params file (dc_demos/params/mcap_recording.yaml), one launch command:

colcon build
ros2 launch dc_demos mcap_recording.launch.py

dc_bringup.launch.py starts dc_mcap_writer before the rest of the stack, so its TCP listener is up before Vector's generated socket sink (re)connects to it — the very first Record lands instead of relying on Vector's own retry. The records_log file Destination writes every Record as it is shipped to /tmp/dc/mcap_recording_records.ndjson, so tail -f on that path doubles as a local view of what dc_mcap_writer is receiving (an earlier version of this demo used a blessed console Destination for the same job — see Destinations: Recipes for why that moved to a passthrough recipe); dc_mcap_writer's own log lines (on this terminal — it runs alongside dc_bridge, not in the background) show each file it opens and closes.

Verify the recording

Stop the stack (Ctrl-C) once a few Records have been collected — dc_mcap_writer finishes the file it has open on shutdown, so every .mcap under the configured output_dir (~/dc_mcap_out in the demo params) is independently valid, including every earlier rotation:

ros2 bag info ~/dc_mcap_out/records_<timestamp>_<pid>_0001.mcap

Files:             /root/dc_mcap_out/records_20260904T064025Z_40_0001.mcap
Bag size:          3.8 KiB
Storage id:        mcap
ROS Distro:        unknown
Duration:          9.998903549s
Start:             Sep  4 2026 06:40:26.274123868 (1788504026.274123868)
End:               Sep  4 2026 06:40:36.273027417 (1788504036.273027417)
Messages:          10
Topic information: Topic: dc.measurement.cpu | Type: dc/record | Count: 3 | Serialization Format: json
                    Topic: dc.measurement.memory | Type: dc/record | Count: 3 | Serialization Format: json
                    Topic: dc.measurement.os | Type: dc/record | Count: 1 | Serialization Format: json
                    Topic: dc.measurement.uptime | Type: dc/record | Count: 3 | Serialization Format: json
Service:           0
Service information:

Type: dc/record — not unknown — because dc_mcap_writer registers its JSON-schema Channels under that schema name (writer.py's register_schema(name="dc/record", ...)); ros2 bag info shows it verbatim rather than resolving a ROS message type, since there isn't one. Message counts and per-topic breakdown come from the file's Statistics/Channel records regardless. Also note the topic name is dc.measurement.cpu, not dc.dc.measurement.cpu — a single dc. prefix, matching the dc.<tag> routing contract in Destinations.

Opening the same file in Foxglove Studio (Open local file…) lists each dc.<tag> Channel and renders its JSON fields (cpu.average, memory.used, uptime.time, …) in the Raw Messages and Plot panels like any other topic.

Rotation

dc_mcap_writer rotates to a new .mcap file once the current one hits max_bytes (default 128 MiB) or has been open max_duration_secs (default 300s), whichever comes first — the same "whichever limit first" shape used for files.retention (ADR-0005), applied here to dc_mcap_writer's own output rather than dc_uploader's intent queue (ADR-0009 explains why this stays outside dc_bridge). Filenames are <prefix>_<UTC timestamp>_<pid>_<rotation index>.mcap; the PID and counter together guarantee a unique name even across a process restart landing in the same wall-clock second as the previous process's last rotation — without both, the new process could compute the identical name and its open(..., "wb") would silently truncate the file the previous process had already finished.

Understanding the configuration

dc_demos/params/mcap_recording.yaml's dc_mcap_writer: block is not nested inside dc_bridge's destinations list, and can't be made to look exactly like postgres/s3/file/console/vector there: destinations is parsed and validated by dc_bridge itself, in C++, and an unrecognized type is a hard startup error by design (see Destinations) — teaching it a sixth type would mean changing dc_bridge, which ADR-0009 explicitly decided against (Vector's own at-least-once/disk-buffered guarantees already cover what would have justified that). So it's a sibling top-level block instead, structurally shaped like a Destination (inputs, a handful of scalar settings) without literally being one.

dc_bringup.launch.py's build_bridge_and_mcap_actions() reads that block at launch time (not compiled in — editing the params file and relaunching is how you change it) and, when enabled: true:

1. Renders a Vector socket sink (mode = "tcp", encoding.codec = "json", framing.method = "newline_delimited") from inputs, converting each ROS topic to its public dc.<tag> route the same way the Bridge itself does, to ~/.dc/generated_mcap_sink.toml — regenerated every launch, not meant to be hand-edited — and merges that path into whatever dc_bridge.custom_config_files the params file already lists (a second parameters=[...] entry for a list-valued ROS parameter replaces the file's value rather than appending to it, so this merge has to happen in the launch file's own Python, not by relying on launch_ros).
2. Starts dc_mcap_writer as a plain ExecuteProcess — deliberately not ros2 run dc_mcap_writer dc_mcap_writer: ros2 run spawns its target as a child of its own process and does not forward signals to it, so launch's own respawn/shutdown handling would only ever reach the ros2 run wrapper, leaving the real dc_mcap_writer process running, orphaned, never getting the chance to finish its currently-open .mcap file. dc_mcap_writer has no rclpy/ROS-node dependency of its own, so nothing here actually needs ros2 run's node-launching machinery — a sourced workspace already puts it on PYTHONPATH, so python3 -m dc_mcap_writer.cli runs it directly.

Like every passthrough sink, the generated Vector sink gets Vector's default in-memory buffer, not the disk buffer blessed sinks get — see the "Passthrough: custom_config_files" section of Destinations.

dc_mcap_writer itself routes each incoming Record to a Channel by its tag field (falling back to name, then a fixed dc.unknown catch-all), and reads the Record's normalized timestamp from its own --time-key (default date, matching the Bridge's own time_key parameter — not currently exposed as its own dc_mcap_writer: field, set it via ros2 launch's underlying ExecuteProcess if you need a non-default value) to set the MCAP message's log_time/publish_time — it understands all three time_format values (epoch_nanos, iso8601, double), same as documented for Destinations.

Running dc_mcap_writer outside dc_bringup

The launch integration above covers the common case; python3 -m dc_mcap_writer.cli (see --help) is still the standalone entrypoint underneath it, useful for converting an existing NDJSON capture (--stdin) or running it against a hand-authored passthrough snippet the way the Elasticsearch tutorial shows for other sinks.

Elasticsearch (passthrough)

dc_bridge blesses exactly five Destination types — postgres, s3, file, console, vector (see Destinations) — and Elasticsearch is not one of them. This tutorial is the worked example for reaching everything else: the ADR-0003 passthrough, a raw Vector config snippet loaded through custom_config_files that consumes the same public dc.<tag> routes a blessed Destination consumes. Nothing in dc_bridge knows what Elasticsearch is, and no DC code was written to support it.

Elasticsearch is the example because it is the one most often asked for, but the shape generalises: swap the sink type and you have Kafka, Loki, ClickHouse, Datadog, or anything else in Vector's catalog. The InfluxDB demo is the same mechanism against a Turtlebot3 simulation; this page needs no simulator and no robot — four system Measurements, one compose file, one terminal.

Setup the infrastructure

Start Elasticsearch and Kibana by following the steps:

podman compose -f tools/infrastructure/docker/docker-compose.elasticsearch.yaml up -d

Confirm the cluster is up before launching DC — the sink retries a store that isn't there yet, but a green cluster makes the first Record land immediately:

curl -s http://localhost:9200/_cluster/health

One-time: install the passthrough sink config

Because there is no Elasticsearch Destination to configure through ROS parameters, the Vector sink itself ships as a plain file in this package, dc_demos/config/elasticsearch_sink.toml, installed to the package's share directory. Copy it into place once before the first launch:

mkdir -p ~/.dc
cp "$(ros2 pkg prefix dc_demos)/share/dc_demos/config/elasticsearch_sink.toml" ~/.dc/

The params file's custom_config_files points at that path. Editing the copy in ~/.dc/ is how you change the sink — it is read at Bridge startup, not compiled in.

Run it

colcon build
ros2 launch dc_demos elasticsearch.launch.py

The file Destination writes every Record as it is shipped to /tmp/dc/elasticsearch_records.ndjson, so tail -f on that path doubles as a local view of what Elasticsearch is receiving:

{"cpu":{"average":0,"processes":9,"sorted":[]},"custom_keys":["robot_name"],"date":1788504548.4015868,"flattened":false,"host":"127.0.0.1","name":"cpu","nested":true,"robot_name":"C3PO","run_id":"170","source_type":"fluent","tag":"dc.measurement.cpu","timestamp":"2026-09-04T06:49:08.401586806Z"}
{"custom_keys":["robot_name"],"date":1788504548.404685,"flattened":false,"host":"127.0.0.1","memory":{"used":96.89765167236328},"name":"memory","nested":true,"robot_name":"C3PO","run_id":"170","source_type":"fluent","tag":"dc.measurement.memory","timestamp":"2026-09-04T06:49:08.404685157Z"}
{"custom_keys":["robot_name"],"date":1788504548.4057963,"flattened":false,"host":"127.0.0.1","name":"uptime","nested":true,"robot_name":"C3PO","run_id":"170","source_type":"fluent","tag":"dc.measurement.uptime","timestamp":"2026-09-04T06:49:08.405796359Z","uptime":{"time":1636933}}

Visualize the data

With curl

The sink writes one index per UTC day, so dc-records-* is every day's data:

curl -s 'http://localhost:9200/dc-records-*/_count'

{"count":25,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0}}

Count the documents per Measurement — cpu, memory and uptime poll every 5 seconds, while os is configured to collect once (init_max_measurements: 1), which is exactly what shows up:

curl -s -H 'Content-Type: application/json' 'http://localhost:9200/dc-records-*/_search' \
  -d '{"size":0,"aggs":{"by_name":{"terms":{"field":"name.keyword"}}}}'

"buckets": [
  { "key": "cpu",    "doc_count": 8 },
  { "key": "memory", "doc_count": 8 },
  { "key": "uptime", "doc_count": 8 },
  { "key": "os",     "doc_count": 1 }
]

And one document in full:

curl -s -H 'Content-Type: application/json' 'http://localhost:9200/dc-records-*/_search' \
  -d '{"size":1,"query":{"term":{"name.keyword":"os"}}}'

{
  "@timestamp": "2026-08-07T15:47:09Z",
  "custom_keys": ["robot_name"],
  "date": 1786117629.0,
  "flattened": false,
  "host": "127.0.0.1",
  "name": "os",
  "nested": true,
  "os": {
    "cpus": 8,
    "kernel": "6.12.96+deb13-amd64",
    "memory": 23.23,
    "os": "Ubuntu 24.04.4 LTS"
  },
  "robot_name": "C3PO",
  "run_id": "1",
  "source_type": "fluent",
  "tag": "dc.measurement.os",
  "timestamp": "2026-08-07T15:47:09Z"
}

The Record's own fields (os.*, date, robot_name, run_id) are joined by four the Shipper adds: tag — the Tag the Record was routed under, useful as a filter — plus host, source_type and timestamp from Vector's ingest source. @timestamp is added by the snippet; see below.

With Kibana

Open http://localhost:5601, then Stack Management → Data Views → Create data view:

Discover then shows Records as they arrive, and cpu.average, memory.used and uptime.time are all numeric fields ready to chart in Lens — no mapping to declare, because Elasticsearch inferred it from the first document of each index.

Understanding the configuration

Measurements

Four system Measurements, none of which need hardware, a simulator, or a running robot:

1. CPU
2. Memory
3. OS
4. Uptime

measurement_server:
  ros__parameters:
    measurement_plugins: ["cpu", "memory", "os", "uptime"]
    cpu:
      plugin: "dc_measurements/Cpu"
      topic_output: "/dc/measurement/cpu"
      polling_interval: 5000
      include_measurement_name: true
      init_collect: true
      max_processes: 5
      cpu_min: 5.0
      nested: true
      flatten: false

The one destination-specific choice here is flatten: false. With nested: true the Measurement wraps its values under its own key; flatten then decides whether that structure is collapsed into keys like /cpu/average (what the InfluxDB and PostgreSQL demos use, because a line-protocol or column-shaped store wants flat fields) or left as real nested JSON. Elasticsearch stores JSON natively and maps a nested object to dotted field names on its own, so leaving it nested gives you cpu.average — a field Kibana can aggregate — instead of a literal field named /cpu/average, which every KQL query would have to escape.

Destination: what actually creates the routes

This is the part of the passthrough that surprises people:

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["records_log"]
    records_log:
      type: file
      receives: records
      inputs:
        [
          "/dc/measurement/cpu",
          "/dc/measurement/memory",
          "/dc/measurement/os",
          "/dc/measurement/uptime",
        ]
      path: "/tmp/dc/elasticsearch_records.ndjson"
      time_key: "date"
      time_format: "double"
    custom_config_files: ["$HOME/.dc/elasticsearch_sink.toml"]
    vector_forward_host: "127.0.0.1"
    vector_forward_port: 24224

destinations still names a blessed Destination, and it is not decoration. dc_bridge derives two things from destinations and nothing else:

• which ROS topics it subscribes to, and
• which dc.<tag> route branches exist in the Vector config it renders.

It never reads a passthrough snippet's inputs. So a topic that appears only in the snippet is a topic the Bridge never subscribes to and never routes — the snippet's inputs would resolve to routes that don't exist. Every topic you want in Elasticsearch must therefore appear in some blessed Destination's inputs too.

file is the anchor here — it needs no infrastructure and no credentials, and writes path to disk rather than your terminal. An earlier version of this demo used a blessed console Destination for the same job (console also gave you a local view of what was landing in Elasticsearch), but per ADR-0003 console — along with postgres and s3 — has since moved to the passthrough recipe itself, being a pure Vector-sink wrapper with no DC-specific logic; see Destinations: Recipes for that recipe if you still want a terminal echo alongside Elasticsearch.

The passthrough snippet

[transforms.elasticsearch_prepare]
type = "remap"
inputs = [
  "dc.dc.measurement.cpu",
  "dc.dc.measurement.memory",
  "dc.dc.measurement.os",
  "dc.dc.measurement.uptime",
]
source = '''
."@timestamp" = .timestamp

.doc_id = sha2(
  to_string!(.tag) + "|" + to_string!(.date) + "|" + to_string!(.run_id),
  variant: "SHA-256"
)
'''

[sinks.elasticsearch]
type = "elasticsearch"
inputs = ["elasticsearch_prepare"]
endpoints = ["http://127.0.0.1:9200"]
api_version = "v8"
mode = "bulk"
bulk.index = "dc-records-%Y.%m.%d"
bulk.action = "index"
id_key = "doc_id"

[sinks.elasticsearch.buffer]
type = "disk"
max_size = 268435488

Each inputs entry is one of the stable dc.<tag> routes. The Tag is the topic name with the leading / dropped and the remaining / turned into . (/dc/measurement/cpu → dc.measurement.cpu), and the route is dc.<tag> — so dc.dc.measurement.cpu. These names are public API; see Destinations.

Note that a snippet is not limited to sinks. elasticsearch_prepare is a transform, and it is merged into the topology like any other component — the only rule is that a snippet must not define a component id the generated config owns (dc_bridge_in, dc_bridge_normalize, dc, or a configured Destination's name). Consuming those ids is fine; redefining one is a loud Bridge startup error naming the file.

Four things in there are worth explaining, because three of them are what separate a snippet that works from a snippet that works and survives contact with a real deployment:

."@timestamp" = .timestamp — Elasticsearch's convention for a document's time field is @timestamp, and Kibana offers it as the default when you create a data view. Vector calls it timestamp. This copies it across. Copy, not move: bulk.index is a strftime template rendered from the event's own timestamp, and deleting that field makes every document fail to render an index name and get dropped.

bulk.index = "dc-records-%Y.%m.%d" — one index per UTC day, the usual shape for time-series data. dc-records-* is then a single Kibana data view, and expiring old data is deleting whole indices rather than running delete-by-query.

id_key + a deterministic doc_id — the Shipper is at-least-once (ADR-0002). After an outage, Records that were in flight when it started are re-sent on recovery, and with Elasticsearch's default auto-generated _id every re-send becomes a new document. Hashing (Tag, Record timestamp, run id) into a stable id makes the write an upsert instead, so a re-delivery overwrites its earlier copy. Vector moves the doc_id field into the document's _id rather than storing it in the body.

The disk buffer — dc_bridge gives every blessed sink a disk buffer, but a passthrough sink gets Vector's default, which is in-memory and 500 events deep. That is not a correctness problem (see below), but a shallow buffer starts applying backpressure within seconds of a store going away, which is what turns a short outage into a large burst of re-deliveries. max_size is in bytes and Vector rejects anything below ~256 MiB; the buffer lives under the Bridge's own shipper.data_dir.

What happens when Elasticsearch goes away

Worth doing once, because it is the question every passthrough raises — the Bridge's durability guarantees are documented for blessed Destinations, and it is reasonable to wonder whether a sink DC knows nothing about still gets them.

Leave DC running and stop the store for two minutes:

podman stop elasticsearch
# ...wait...
podman start elasticsearch

Vector logs the failure and retries with exponential backoff, so the terminal fills with:

WARN sink{component_id=elasticsearch component_type=elasticsearch}: vector::sinks::util::retries: Retrying after error. error=Failed to make HTTP(S) request: error trying to connect: tcp connect error: Connection refused (os error 111)
WARN sink{component_id=elasticsearch component_type=elasticsearch}: vector::sinks::util::service::health: Endpoint is unhealthy. endpoint=http://127.0.0.1:9200

uptime.time increases by exactly 5 every Record, which makes gaps and duplicates countable. Across a 2-minute outage, with the sink above, all 110 Records spanning the outage arrive: none missing, none duplicated.

Two things produced that result, and both are choices in the snippet rather than anything DC does for you:

• No loss comes from the Shipper's end-to-end acknowledgements. dc_bridge holds a Record until Vector confirms it, so a full buffer means the Bridge retries rather than drops — this part you get for free, disk buffer or not.
• No duplicates comes from the deterministic _id. Running the same outage with Vector's default in-memory buffer and Elasticsearch's auto-generated ids also loses nothing, but indexes each Record up to a dozen times: backpressure hits within seconds, the Bridge re-sends unacknowledged Records, and every re-send lands as its own document.

Adapting it

• Another store. Replace [sinks.elasticsearch] with any other type from Vector's sink catalog. The inputs, the blessed-Destination requirement, and the at-least-once caveat are the same; only the sink block changes.
• A real robot's data. Add the topics to both the blessed Destination's inputs and the snippet's. The Turtlebot3 AWS Warehouse params file is a good source of Measurement configuration to copy from.
• A secured cluster. Add an auth block to the sink — there is a commented example at the bottom of elasticsearch_sink.toml. Vector expands $VAR references in its own config at startup, so the password can come from the environment.

Cleanup

podman compose -f tools/infrastructure/docker/docker-compose.elasticsearch.yaml down -v

Turtlebot3

In this example, we add a robot and start collecting robot data to Stdout.

You will also need 3 terminal windows, to:

1. Run the Nav2 turtlebot3 launchfile: it starts localization, navigation and RViz
2. Run navigation inspection demo
3. Run DC

Using a different terminal window for DC helps reading its information.

Setup the environment

Python dependencies

For this tutorial, we will need to install all dependencies, the demo dashboard's included (uv owns them; see Setup):

uv sync

Setup Infrastructure

RustFS

RustFS will be used as storage for images and other files. To start it, follow the steps — a single RustFS container plus a one-shot container that bootstraps the dc-files bucket, using the default rustfsadmin/rustfsadmin credentials this demo's params file already assumes. No manual bucket or key setup is needed, unlike the old 4-node MinIO cluster + nginx console this replaces.

PostgreSQL

PostgreSQL will be used as database storage for our JSON. Later on, backend engineers can make requests on those JSON based on measurement requested and time range. To start it, follow the steps

The default yaml configuration file does not need change as it also uses default values.

Setup simulation environment

In the terminal 1, source your environment:

source /opt/ros/rolling/setup.bash
source install/setup.bash

Nothing else has to be exported. dc_simulation puts its own models and worlds on GZ_SIM_RESOURCE_PATH through its environment hook.

Terminal 1: Start Navigation

Then, in the same terminal (1), start the simulation and Nav2. DC comes up separately in terminal 2, so turn it off here:

ros2 launch dc_demos tb3_qrcodes.launch.py \
    headless:=False \
    use_dc:=False

RViz and gz-sim will start: now you see the robot in the warehouse, and the map on RViz. AMCL sets the initial pose itself, so there is no need to click "2D Pose Estimate".

Terminal 2: Start DC

Run colcon build to compile the workspace:

colcon build

Now, start the demo:

ros2 launch dc_demos tb3_simulation_pgsql_minio.launch.py use_sim_time:=True

The robot will start collecting data.

Terminal 3: Start autonomous navigation

Execute

ros2 run nav2_simple_commander demo_security

The robot will start moving and you will be able to see all visualizations activated in RViz:

Visualize the data

In the database

Navigate to localhost:8080

1. Select dc database
2. Select dc table
3. Click on Select data

You will see rows filling the database. You can click on one to see its content:

With Grafana

Four Grafana dashboards ship with the DC 2.0 infrastructure now — Home, Robot, KPI and Fast DDS statistics (the System/Environment/Infrastructure dashboards were dropped as out of scope for this minimal rework). Open http://localhost:3000 (admin/admin) and pick the Robot dashboard: its panels are backed by SQL queries against the dc/dc_files PostgreSQL tables — the Grafana datasource is PostgreSQL (uid dc_postgres) now, not InfluxDB. It shows, among other things, speed/command velocity over time and a table of uploaded inspection files with their RustFS + PostgreSQL upload status:

SELECT to_timestamp(updated_at) AS "time", group_name, robot_name, storage_type, remote_path, content_type, size, uploaded
FROM dc_files WHERE kind = 'file_status' ORDER BY updated_at DESC LIMIT 100

See tools/infrastructure/docker/config/grafana/dashboards/robot.json for every panel's exact query.

The KPI dashboard is the operations view of the same data: availability and uptime per robot over whatever time range the dashboard is set to, computed by the KPI views rather than by any panel query of its own. It populates from this demo's uptime Measurement with no extra configuration.

That's it! Now you can collect your data!

Understanding the configuration

Measurement server

Measurements

measurement_plugins sets which plugin to load. We collect

System measurements:

1. CPU
2. OS
3. Memory
4. Uptime

Robot measurements:

1. Camera images
2. Command velocities
3. Distance traveled
4. Driving type
5. Positions
6. Speed

Environment measurements:

1. Map

Infrastructure measurements:

1. RustFS health
2. PostgreSQL health

Each has their own configuration: polling interval, source topic, destination paths, topics used as input etc. Going through each of them would be too long here but you can check for each measurement its documentation and the general documentation of measurements

measurement_server:
  ros__parameters:
    ...
    camera:
      plugin: "dc_measurements/Camera"
      cam_topic: "/intel_realsense_r200_depth/image_raw"
      cam_name: "Intel Realsense"
      save_raw_img: true
      save_raw_path: "camera/raw/%Y-%m-%dT%H-%M-%S"
      remote_prefixes: [""]
      remote_keys: ["rustfs"]
      ...
    map:
      plugin: "dc_measurements/Map"
      save_path: "map/%Y-%m-%dT%H-%M-%S"
      topic_output: "/dc/measurement/map"
      remote_prefixes: [""]
      remote_keys: ["rustfs"]
      ...
    rustfs_health:
      plugin: "dc_measurements/TCPHealth"
      topic_output: "/dc/measurement/rustfs_health"
      polling_interval: 5000
      host: "127.0.0.1"
      port: 9000
      name: "RustFS"
      include_measurement_plugin: true
    pgsql_health:
      plugin: "dc_measurements/TCPHealth"
      topic_output: "/dc/measurement/pgsql_health"
      polling_interval: 5000
      host: "127.0.0.1"
      port: 5432
      name: "PostgreSQL"
      include_measurement_plugin: true

Conditions

We also initialize conditions:

1. min_distance_traveled
2. max_distance_traveled

They are used in the distance traveled measurement to only take values in a certain range.

Destination server

Here we enable the records_log and rustfs Destinations, plus a passthrough for PostgreSQL:

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["records_log", "rustfs"]
    records_log:
      type: file
      receives: records
      inputs: [
          # System
          "/dc/measurement/cpu",
          "/dc/measurement/memory",
          "/dc/measurement/os",
          "/dc/measurement/uptime",
          # Robot
          "/dc/measurement/camera",
          "/dc/measurement/cmd_vel",
          "/dc/measurement/distance_traveled",
          "/dc/measurement/driving_type",
          "/dc/measurement/position",
          "/dc/measurement/speed",
          # Environment
          "/dc/measurement/map",
          # Infrastructure
          "/dc/measurement/rustfs_health",
          "/dc/measurement/pgsql_health",
        ]
      path: "/tmp/dc/tb3_simulation_pgsql_minio_records.ndjson"
      time_key: "date"
    rustfs:
      type: s3
      receives: files
      inputs: ["/dc/measurement/map", "/dc/measurement/camera"]
      bucket: "dc-files"
      endpoint: "http://127.0.0.1:9000"
      region: "us-east-1"
      access_key_id: "rustfsadmin"
      secret_access_key: "rustfsadmin"
      force_path_style: true
    files:
      delete_when_sent: true
      metadata_destination: "records_log"
    custom_config_files: ["$HOME/.dc/tb3_simulation_pgsql_minio_sink.toml"]

# ~/.dc/tb3_simulation_pgsql_minio_sink.toml
[sinks.pgsql]
type = "postgres"
inputs = [
  "dc.dc.measurement.cpu", "dc.dc.measurement.memory", "dc.dc.measurement.os",
  "dc.dc.measurement.uptime", "dc.dc.measurement.camera", "dc.dc.measurement.cmd_vel",
  "dc.dc.measurement.distance_traveled", "dc.dc.measurement.driving_type",
  "dc.dc.measurement.position", "dc.dc.measurement.speed", "dc.dc.measurement.map",
  "dc.dc.measurement.rustfs_health", "dc.dc.measurement.pgsql_health",
]
endpoint = "postgres://dc:password@127.0.0.1:5432/dc"
table = "dc"

[sinks.pgsql.buffer]
type = "disk"
max_size = 268435488

[sinks.pgsql_files]
type = "postgres"
inputs = ["dc.dc.files"]
endpoint = "postgres://dc:password@127.0.0.1:5432/dc"
table = "dc_files"

[sinks.pgsql_files.buffer]
type = "disk"
max_size = 268435488

Copy the snippet into place before launching:

mkdir -p ~/.dc && cp "$(ros2 pkg prefix dc_demos)/share/dc_demos/config/tb3_simulation_pgsql_minio_sink.toml" ~/.dc/

PostgreSQL Destinations

PostgreSQL is reached through the ADR-0003 passthrough now, not a blessed postgres Destination — see Destinations: Recipes. The passthrough pgsql sink carries almost every measurement and the two infrastructure health checks as plain Records. Note that not all data needs to go to PostgreSQL — only topics whose dc.<tag> route exists (i.e. listed in records_log's inputs) reach it.

pgsql_files is a second, dedicated postgres sink in the same passthrough snippet for dc_uploader's status Records — it has no ROS topic inputs of its own; it consumes the dc.dc.files route instead, which records_log gains from being named as files.metadata_destination (that parameter must name a configured receives: records Destination, so it can't point at a passthrough-only sink id directly). See Destinations, ADR-0005 and ADR-0014 for the full split, and the QR codes demo for a worked example of the same pattern.

RustFS Destination

We list only map and camera in rustfs's inputs since those are the only measurements referencing Files. rustfs's type: s3 and receives: files mark it as owned by dc_uploader — a separate process from dc_bridge (ADR-0014) — rather than a Vector sink target: it uploads whatever File the measurement's remote_keys: ["rustfs"] pointed at it, verifies the object landed, and (with files.delete_when_sent: true) deletes the local copy only once that's confirmed. Unlike PostgreSQL, rustfs stays a blessed Destination: receives: files is served entirely by dc_uploader reading these same ROS params, never by a Vector sink, so there is no passthrough equivalent for it to migrate to.

Turtlebot3 AWS Warehouse InfluxDB

dc_bridge blesses exactly five Destination types — postgres, s3, file, console, vector (see Destinations) — and InfluxDB is not one of them. This demo is not a peer of the PostgreSQL/RustFS demos: it exists to show how to reach a destination dc_bridge doesn't bless directly, via the ADR-0003 passthrough escape hatch — a raw Vector sink config loaded through custom_config_files, consuming the same public dc.<tag> routes a blessed Destination would. Read Destinations's "Passthrough" section first if you haven't already; this page only covers what's specific to InfluxDB.

You will also need 3 terminal windows, to:

1. Run the Nav2 turtlebot3 launchfile: it starts localization, navigation and RViz
2. Run navigation inspection demo
3. Run DC

Using a different terminal window for DC helps reading its information.

Setup the environment

Python dependencies

For this tutorial, we will need to install all dependencies, the demo dashboard's included (uv owns them; see Setup):

uv sync

Setup the infrastructure

InfluxDB

InfluxDB will be used to store our data and timestamps. To start it, follow the steps (tools/infrastructure/docker/docker-compose.influxdb.yaml, unchanged by the DC 2.0 rework — this demo still needs a real InfluxDB instance to point the passthrough sink at).

One-time: install the passthrough sink config

dc_bridge has no InfluxDB Destination to configure through ROS parameters, so the Vector sink itself ships as a plain file in this package, dc_demos/config/tb3_simulation_influxdb_sink.toml, installed to the package's share directory. Copy it into place once before the first launch:

mkdir -p ~/.dc
cp "$(ros2 pkg prefix dc_demos)/share/dc_demos/config/tb3_simulation_influxdb_sink.toml" ~/.dc/

dc_params_file's custom_config_files (see below) points at this path.

Setup simulation environment

In the terminal 1, source your environment:

source /opt/ros/rolling/setup.bash
source install/setup.bash

Nothing else has to be exported. dc_simulation puts its own models and worlds on GZ_SIM_RESOURCE_PATH through its environment hook.

Terminal 1: Start Navigation

Then, in the same terminal (1), start the simulation and Nav2. DC comes up separately in terminal 2, so turn it off here:

ros2 launch dc_demos tb3_qrcodes.launch.py \
    headless:=False \
    use_dc:=False

RViz and gz-sim will start: now you see the robot in the warehouse, and the map on RViz. AMCL sets the initial pose itself, so there is no need to click "2D Pose Estimate".

Terminal 2: Start DC

Run colcon build to compile the workspace:

colcon build

Now, start the demo:

ros2 launch dc_demos tb3_simulation_influxdb.launch.py use_sim_time:=True

The robot will start collecting data.

Terminal 3: Start autonomous navigation

Execute

ros2 run nav2_simple_commander demo_security

The robot will start moving and you will be able to see all visualizations activated in RViz:

Visualize the data

Grafana's own datasource is PostgreSQL now (see PostgreSQL/RustFS demo), and every dashboard shipped with the DC 2.0 infrastructure — Home, Robot, KPI and Fast DDS statistics — is backed by it: there is no Grafana dashboard for this demo's data, since it never touches PostgreSQL. To look at what landed in InfluxDB, use InfluxDB's own tooling instead:

influx -database dc -execute "SELECT * FROM dc ORDER BY time DESC LIMIT 20"

or query its HTTP API directly:

curl -G 'http://127.0.0.1:8086/query' --data-urlencode "db=dc" --data-urlencode "q=SELECT * FROM dc ORDER BY time DESC LIMIT 20"

Understanding the configuration

Measurement server

Measurements

measurement_plugins sets which plugin to load. We collect

System measurements:

1. CPU
2. OS
3. Memory
4. Uptime

Robot measurements:

1. Camera images
2. Command velocities
3. Distance traveled
4. Positions
5. Speed

Environment measurements:

1. Map

Infrastructure measurements:

1. InfluxDB health, a TCPHealth check against InfluxDB's own port (8086) — unrelated to the passthrough mechanism, this is the same kind of infrastructure health-check measurement the PostgreSQL/RustFS demo uses for its own destinations.

None of this changed from the destination-agnostic measurement configuration used elsewhere in DC 2.0 — nested/flatten still shape the JSON for InfluxDB's line-protocol-oriented storage, and images (map, camera) are still stored as base64 strings since that's the only field type Grafana (or any consumer reading straight out of InfluxDB) can render from a database column. What changed is only how the Records reach InfluxDB in the first place — see the Destination section below.

measurement_server:
  ros__parameters:
    ...
    camera:
      plugin: "dc_measurements/Camera"
      topic_output: "/dc/measurement/camera"
      save_raw_base64: true
      nested: true
      flatten: true
      ...
    influxdb_health:
      plugin: "dc_measurements/TCPHealth"
      topic_output: "/dc/measurement/influxdb_health"
      polling_interval: 5000
      host: "127.0.0.1"
      port: 8086
      name: "InfluxDB"
      include_measurement_plugin: true
      nested: true
      flatten: true

An example camera Record, now without a tags field (that mechanism no longer exists — a Destination's inputs list decides routing instead). nested: true + flatten: true together produce /-prefixed dotted-path keys such as /camera/camera_name, which is also exactly what lands as the InfluxDB column name, since nothing downstream renames them:

{
  "/camera/base64/raw": "/9j/4AAQSkZJRgABAQAA...",
  "/camera/camera_name": "Intel Realsense",
  "custom_keys": ["robot_name", "id"],
  "date": 1677668926.700422,
  "flattened": true,
  "host": "127.0.0.1",
  "id": "be781e5ffb1e7ee4f817fe7b63e92c32",
  "name": "camera",
  "nested": true,
  "robot_name": "Turtlebot",
  "run_id": "218",
  "source_type": "fluent",
  "tag": "dc.measurement.camera",
  "timestamp": "2026-09-04T00:00:40.252Z"
}

Conditions

We also initialize conditions:

1. min_distance_traveled
2. max_distance_traveled

They are used in the distance traveled measurement to only take values in a certain range.

Destination: the passthrough

There is no InfluxDB Destination to bless, but destinations is not empty — it names a file Destination whose inputs are what put these topics on the dc.<tag> routes the snippet consumes:

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["records_log"]
    records_log:
      type: file
      receives: records
      inputs: ["/dc/measurement/cpu", "/dc/measurement/memory", ...]
      path: "/tmp/dc/tb3_simulation_influxdb_records.ndjson"
      time_key: "date"
      time_format: "double"
    custom_config_files: ["$HOME/.dc/tb3_simulation_influxdb_sink.toml"]
    vector_forward_host: "127.0.0.1"
    vector_forward_port: 24224

dc_bridge derives both its ROS subscriptions and its dc.<tag> route branches from destinations, and never reads a passthrough snippet's inputs — so a passthrough always accompanies at least one blessed Destination covering the same topics. file is the anchor in every passthrough demo in this repo now, including this one — blessed console moved to a passthrough recipe of its own (ADR-0003, Destinations: Recipes) — but it would have been the wrong anchor for this demo regardless, even back when it was still blessed: this demo collects base64 camera and map images, which make for unreadable terminal output. The Elasticsearch tutorial explains the file anchor mechanics in more detail.

custom_config_files lists raw Vector config snippets that are merged as-is alongside whatever dc_bridge itself renders (here, nothing) — see Destinations's passthrough section for the full contract (naming collisions, vector validate as a startup backstop, etc.). The snippet installed above:

# ~/.dc/tb3_simulation_influxdb_sink.toml
[sinks.influxdb]
type = "influxdb_logs"
inputs = [
  "dc.dc.measurement.cpu",
  "dc.dc.measurement.memory",
  "dc.dc.measurement.os",
  "dc.dc.measurement.uptime",
  "dc.dc.measurement.camera",
  "dc.dc.measurement.cmd_vel",
  "dc.dc.measurement.distance_traveled",
  "dc.dc.measurement.position",
  "dc.dc.measurement.speed",
  "dc.dc.measurement.map",
  "dc.dc.measurement.influxdb_health",
]
endpoint = "http://127.0.0.1:8086"
measurement = "dc"

[sinks.influxdb.influxdb1_settings]
database = "dc"
username = "dc"

Each inputs entry is one of the stable dc.<tag> routes dc_bridge exposes for every topic that appears in any Destination's inputs or, as here, any custom_config_files snippet's inputs — the Tag is the topic name with the leading / dropped and the rest of the /s turned into . (/dc/measurement/cpu → dc.measurement.cpu), and the route is dc.<tag> (so dc.dc.measurement.cpu). Vector's influxdb_logs sink type is what actually understands InfluxDB 1.x's write API; nothing in this snippet is DC-specific beyond the dc.<tag> inputs.

Advanced

Either you run the whole inspection pipeline at once — Measurements, Conditions, Groups, File uploads and the Grafana dashboards reading them back — or you leave YAML behind and write DC code yourself.

Prerequisites: the Intermediate tier, plus several services running together for the QR codes pipeline, and a C++ toolchain for the custom plugin. Roughly 45 minutes for the QR codes pipeline, 30 for the custom plugin including the rebuild.

Note that each demo assumes concepts explained in previous demos will be acknowledged.

QRCodes

In this example, we add a robot and start collecting robot data to PostgreSQL, and maps and scanned QR codes to RustFS as image files.

You will also need 3 terminal windows, to:

1. Run the simulation + Nav2 launchfile: it starts gz-sim, localization, navigation and RViz
2. Run DC
3. Drive the robot past the QR codes

Keeping them separate helps reading the JSON printed on the DC terminal, which RViz and gz-sim would otherwise drown out.

Setup RustFS and PostgreSQL

RustFS

RustFS will be used as storage for the map and camera image Files. To start it, follow the steps — a single RustFS container plus a one-shot container that bootstraps the dc-files bucket the Destinations below upload into, using the default rustfsadmin/rustfsadmin credentials this demo's params file already assumes.

PostgreSQL

PostgreSQL will be used as database storage for our JSON. Later on, backend engineers can make requests on those JSON based on measurement requested and time range. To start it, follow the steps

Setup the ROS environment

In each terminal, source your environment:

source /opt/ros/rolling/setup.bash
source install/setup.bash

Nothing else has to be exported. dc_simulation's environment hook puts its own models/ and worlds/ on GZ_SIM_RESOURCE_PATH, and warehouse.launch.py adds nav2_minimal_tb3_sim's models (where the TurtleBot3 meshes live now — the Gazebo Classic turtlebot3_gazebo package has no Jazzy release).

Start Navigation

dc_demos' own launch file brings up gz-sim with the QR-code warehouse world, spawns dc_simulation's TurtleBot3-Waffle with its ros_gz_bridge, and starts Nav2 (map_server + AMCL + planners) against dc_simulation/maps/qrcodes.yaml. It also starts DC, which this demo drives separately, so turn that off here:

ros2 launch dc_demos tb3_qrcodes.launch.py \
    headless:=False \
    use_dc:=False

gz-sim and RViz will start: you should see the robot in the warehouse, and the map in RViz. AMCL sets the demo's initial pose itself (set_initial_pose in qrcodes_nav.yaml), so there is no need to click "2D Pose Estimate" — wait until the laser scan lines up with the map before starting the run below.

Start DC

Execute

ros2 launch dc_demos tb3_qrcodes_minio_pgsql.launch.py use_sim_time:=True

With this, all data will be transmitted

Drive the robot past the QR codes

In a fourth terminal, run the waypoint follower. It sends the robot down each aisle, stopping in front of every QR-coded pallet so both cameras can read them, and exits once the pass is complete:

ros2 run dc_demos qrcodes_waypoint_follower

|lateral error| + standoff * tan(|yaw error|) + 0.181 <= standoff * tan(30°)

Understanding the configuration

For this demo, we will reconstruct the yaml configuration element by element, given how large it is. Go through the explanation to understand how it works.

Collect command velocity, position and speed to PostgreSQL as a group

Similarly to the previous tutorial:

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["records_log", "rustfs"]
    records_log:
      type: file
      receives: records
      inputs:
        [
          "/dc/measurement/map",
          "/dc/measurement/right_camera",
          "/dc/measurement/left_camera",
          "/dc/group/robot",
        ]
      path: "/tmp/dc/qrcodes_minio_pgsql_records.ndjson"
      time_key: "date"
    custom_config_files: ["$HOME/.dc/qrcodes_minio_pgsql_sink.toml"]

group_server:
  ros__parameters:
    groups: ["robot"]
    robot:
      inputs:
        [
          "/dc/measurement/cmd_vel",
          "/dc/measurement/position",
          "/dc/measurement/speed",
        ]
      output: "/dc/group/robot"
      sync_delay: 5.0
      group_key: "robot"

measurement_server:
  ros__parameters:
    custom_keys_str: ["robot_name"]
    robot_name: "C3PO"
    measurement_plugins: ["cmd_vel", "position", "speed"]
    custom_key_str_list: ["robot_name", "id"]
    custom_keys_str:
      robot_name:
        name: robot_name
        value: "C3PO"
      # Requires systemd package
      id:
        name: id
        value_from_file: /etc/machine-id
    run_id:
      enabled: true
      counter: true
      counter_path: "$HOME/run_id"
      uuid: false
    moving:
      plugin: "dc_conditions/Moving"
    cmd_vel:
      plugin: "dc_measurements/CmdVel"
      group_key: "cmd_vel"
      enable_validator: true
      topic_output: "/dc/measurement/cmd_vel"
      include_measurement_name: true
    position:
      plugin: "dc_measurements/Position"
      group_key: "position"
      topic_output: "/dc/measurement/position"
      enable_validator: true
      global_frame: "map"
      robot_base_frame: "base_link"
      transform_timeout: 0.1
      include_measurement_name: true
    speed:
      plugin: "dc_measurements/Speed"
      group_key: "speed"
      odom_topic: "/odom"
      topic_output: "/dc/measurement/speed"
      include_measurement_name: true

In the measurement server, we set 3 measurements: cmd_vel, position and speed being collected once per second, are validated with their respective JSON schemas and publish on their own topics.

Note the include_measurement_name which include measurement name in the JSON, which is used when grouping. The group collects the data from those 3 measurement and republishes it on the group topic /dc/group/robot.

dc_bridge owns every Destination this demo uses. records_log's inputs list already names every topic this demo produces (/dc/group/robot for this section, plus /dc/measurement/map, /dc/measurement/right_camera and /dc/measurement/left_camera for the sections below): unlike the retired per-measurement tags: [...] mechanism, a Destination's inputs is the single place that decides what reaches it, so we declare it once and simply grow the measurements that feed those topics as we go.

records_log is a file Destination, not postgres — PostgreSQL is reached through the ADR-0003 passthrough instead. The actual postgres sink lives in qrcodes_minio_pgsql_sink.toml, a raw Vector config snippet loaded via custom_config_files:

# ~/.dc/qrcodes_minio_pgsql_sink.toml
[sinks.pgsql]
type = "postgres"
inputs = [
  "dc.dc.measurement.map",
  "dc.dc.measurement.right_camera",
  "dc.dc.measurement.left_camera",
  "dc.dc.group.robot",
]
endpoint = "postgres://dc:password@127.0.0.1:5432/dc"
table = "dc"

[sinks.pgsql.buffer]
type = "disk"
max_size = 268435488

dc_bridge derives its ROS subscriptions and dc.<tag> routes from destinations alone, never from a passthrough snippet's inputs — that's why records_log still lists every topic even though the snippet above is what actually reaches PostgreSQL. Copy the snippet into place before launching:

mkdir -p ~/.dc && cp "$(ros2 pkg prefix dc_demos)/share/dc_demos/config/qrcodes_minio_pgsql_sink.toml" ~/.dc/

You can find more about the postgres sink recipe here

To take a look at records, go to Adminer. It is by default started at http://localhost:8080, it is a database GUI.:

You can then click on a record, to take a look, edit or delete it:

Send the map image and YAML from nav2_map_server to RustFS

First, we add the map measurement. Its remote_keys names the rustfs Destination, which is what actually uploads the pgm/yaml Files — the map measurement only records where they are, locally and (once uploaded) remotely:

measurement_server:
  ros__parameters:
  ...
  measurement_plugins: ["map"]
  map:
    plugin: "dc_measurements/Map"
    group_key: "map"
    polling_interval: 5000
    save_path: "map/%Y-%m-%dT%H:%M:%S"
    topic_output: "/dc/measurement/map"
    save_map_timeout: 4.0
    remote_prefixes: [""]
    remote_keys: ["rustfs"]
    enable_validator: true
    include_measurement_name: true
  ...

Then, the Destinations that make this work:

dc_bridge:
  ros__parameters:
    ...
    destinations: ["records_log", "rustfs"]
    rustfs:
      type: s3
      receives: files
      inputs:
        [
          "/dc/measurement/map",
          "/dc/measurement/right_camera",
          "/dc/measurement/left_camera",
        ]
      bucket: "dc-files"
      endpoint: "http://127.0.0.1:9000"
      region: "us-east-1"
      access_key_id: "rustfsadmin"
      secret_access_key: "rustfsadmin"
      force_path_style: true
    files:
      delete_when_sent: true
      metadata_destination: "records_log"

# ~/.dc/qrcodes_minio_pgsql_sink.toml (continued)
[sinks.pgsql_files]
type = "postgres"
inputs = ["dc.dc.files"]
endpoint = "postgres://dc:password@127.0.0.1:5432/dc"
table = "dc_files"

[sinks.pgsql_files.buffer]
type = "disk"
max_size = 268435488

This introduces the two-way PostgreSQL split this demo relies on:

• pgsql (the passthrough sink declared above) carries the map's own metadata Record (dimensions, resolution, local/remote paths) like any other measurement.
• pgsql_files is a second postgres sink in the same passthrough snippet, dedicated to the Uploader's own bookkeeping. It has no ROS topic inputs of its own — it consumes the dc.dc.files route instead, which records_log gains from being named as files.metadata_destination below. files.metadata_destination must name a configured receives: records Destination (dc_bridge rejects anything else at startup), which is why it names records_log rather than the passthrough sink directly — a passthrough-only sink id isn't eligible.
• rustfs is the s3 Destination that actually uploads the pgm/yaml bytes. receives: files marks it as owned by dc_uploader — a separate process from dc_bridge (ADR-0014) — rather than a Vector sink: dc_bridge subscribes to rustfs's inputs, and durably enqueues an intent for any Record with remote_paths entries whose key matches a Destination name — here rustfs, matching the map measurement's remote_keys above. dc_uploader reads that intent, uploads the referenced Files, verifies they landed, and emits a status Record under dc.files, routed to records_log (and, from there, consumed by the pgsql_files passthrough sink). Unlike pgsql/pgsql_files, rustfs stays blessed: receives: files is served entirely by dc_uploader reading these same ROS params, never by a Vector sink, so there is no passthrough equivalent for it to migrate to.

See Destinations for the full files:/Uploader contract, and ADR-0005 / ADR-0014 for why file uploads are a DC responsibility rather than a Vector sink, and why that logic now runs as its own process.

files.delete_when_sent: true means the local pgm/yaml are removed only once RustFS confirms the upload — never before.

You can check upload status and completion in Grafana's Robot dashboard, whose "Uploaded inspection files" and "Group completion status" panels query the dc_files table pgsql_files writes into:

SELECT to_timestamp(updated_at) AS "time", group_name, robot_name, storage_type, remote_path, content_type, size, uploaded
FROM dc_files WHERE kind = 'file_status' ORDER BY updated_at DESC LIMIT 100

Then, similarly, on Adminer, you can browse the dc table's map rows.

Send QR code images to RustFS

We want to collect pictures taken by the cameras

measurement_server:
  ros__parameters:
  ...
  measurement_plugins: ["cmd_vel", "position", "speed", "map", "right_camera", "left_camera"]
  condition_plugins: ["moving", "inspected_exists"]
  custom_key_str_list: ["robot_name", "id"]
  custom_keys_str:
    robot_name:
      name: robot_name
      value: "C3PO"
    # Requires systemd package
    id:
      name: id
      value_from_file: /etc/machine-id
  run_id:
    enabled: true
    counter: true
    counter_path: "$HOME/run_id"
    uuid: false
  moving:
    plugin: "dc_conditions/Moving"
  inspected_exists:
    plugin: "dc_conditions/Compare"
    key: "inspected"
    comparison: "exists"
  right_camera:
    plugin: "dc_measurements/Camera"
    group_key: "right_camera"
    if_none_conditions: ["moving"]
    if_all_conditions: ["inspected_exists"]
    topic_output: "/dc/measurement/right_camera"
    init_collect: false
    init_max_measurements: -1
    condition_max_measurements: 1
    node_name: "dc_measurement_camera"
    cam_topic: "/right_intel_realsense_r200_depth/image_raw"
    cam_name: right_camera
    enable_validator: true
    draw_det_barcodes: true
    save_raw_img: false
    save_rotated_img: false
    save_detections_img: true
    save_inspected_path: "right_camera/inspected/%Y-%m-%dT%H-%M-%S"
    rotation_angle: 0
    detection_modules: ["barcode"]
    remote_prefixes: [""]
    remote_keys: ["rustfs"]
    include_measurement_name: true
  left_camera:
    plugin: "dc_measurements/Camera"
    group_key: "left_camera"
    if_none_conditions: ["moving"]
    if_all_conditions: ["inspected_exists"]
    topic_output: "/dc/measurement/left_camera"
    init_collect: true
    init_max_measurements: -1
    condition_max_measurements: 1
    node_name: "dc_measurement_camera"
    cam_topic: "/left_intel_realsense_r200_depth/image_raw"
    cam_name: left_camera
    enable_validator: true
    draw_det_barcodes: true
    save_raw_img: false
    save_rotated_img: false
    save_detections_img: true
    save_inspected_path: "left_camera/inspected/%Y-%m-%dT%H-%M-%S"
    rotation_angle: 0
    detection_modules: ["barcode"]
    remote_prefixes: [""]
    remote_keys: ["rustfs"]
    include_measurement_name: true
  ...

Taking a look at the cameras, we can understand that:

1. Data is only collected when the robot is not moving: if_none_conditions: ["moving"]
2. Data is only collected when there is inspected data, so only when a QR code is detected: if_all_conditions: ["inspected_exists"]
3. Data is not collected constantly: init_max_measurements: -1
4. Only one record is collected when conditions are triggered: condition_max_measurements: 1
5. Only images with inspected data are collected:
   1. save_raw_img: false
   2. save_rotated_img: false
   3. save_detections_img: true
6. Barcodes are scanned in each image: detection_modules: ["barcode"]

include_measurement_name matters here too: the Uploader relies on it to know which top-level field of the Record holds the local_paths/remote_paths it should act on.

No new Destination block is needed for the cameras: records_log and rustfs already list /dc/measurement/right_camera and /dc/measurement/left_camera in their inputs (see the first section above), and the passthrough pgsql sink already consumes the matching dc.<tag> routes those inputs create — a Destination's inputs is a single, global list of topics rather than something declared per measurement, so adding a measurement that feeds an already-configured Destination requires no dc_bridge change at all.

Here, we collect images with the rustfs Destination, and their metadata (which record they belong to, remote path once uploaded, image dimensions where relevant) through the passthrough pgsql sink. pgsql_files, fed by the Uploader, tracks when each image is sent to RustFS and — with files.delete_when_sent: true — is deleted locally once confirmed. Note that a Record's remote_paths can name several Destinations at once; the Uploader sends to every one whose name appears there.

An example Record for right_camera, once the image is inspected:

{
  "camera_name": "right_camera",
  "date": 1677668926.700422,
  "flattened": false,
  "id": "be781e5ffb1e7ee4f817fe7b63e92c32",
  "nested": false,
  "robot_name": "C3PO",
  "run_id": "218",
  "local_img_paths": {
    "inspected": "/root/dc_data/C3PO/2023/03/01/17/right_camera/inspected/2023-03-01T17-12-57.jpg"
  },
  "remote_paths": {
    "rustfs": {
      "inspected": "C3PO/2023/03/01/17/right_camera/inspected/2023-03-01T17-12-57.jpg"
    }
  },
  "inspected": {
    "barcode": [
      {
        "data": [81, 82, 99, 111, 100, 101, 45, 49],
        "height": 40,
        "width": 40,
        "top": 120,
        "left": 200,
        "type": "QRCODE"
      }
    ]
  }
}

Fast DDS statistics to PostgreSQL/Grafana

This is the smallest hardware-free way to see Fast DDS's own Statistics Module land in DC: one Measurement (Fast DDS statistics), a passthrough postgres sink (ADR-0003), and a Grafana dashboard provisioned automatically — the same convention #304 established for the KPI dashboard.

Setup Infrastructure

PostgreSQL

Follow the steps to start it. The default yaml configuration file does not need change.

Grafana

Follow the steps to start it.

Run the demo

colcon build
mkdir -p ~/.dc && cp "$(ros2 pkg prefix dc_demos)/share/dc_demos/config/fastdds_stats_pgsql_grafana_sink.toml" ~/.dc/
export FASTDDS_STATISTICS="HISTORY_LATENCY_TOPIC;PUBLICATION_THROUGHPUT_TOPIC;SUBSCRIPTION_THROUGHPUT_TOPIC;RTPS_SENT_TOPIC;RTPS_LOST_TOPIC"
ros2 launch dc_demos fastdds_stats_pgsql_grafana.launch.py

A Record captured from a real run, echoed straight off /dc/measurement/fastdds_stats (no PostgreSQL needed to see this — it's what dc_bridge forwards on):

{
  "custom_keys": ["robot_name"],
  "datareader_count": 1,
  "datawriter_count": 11,
  "domain_id": 0,
  "event": "sample",
  "flattened": false,
  "hosts": ["d:14058711922191368192"],
  "name": "fastdds_stats",
  "nested": false,
  "participant_count": 3,
  "participants": [
    { "guid": "01.0f.4d.26.9c.1d.72.46.00.00.00.00|0.0.1.c1", "name": "/" },
    { "guid": "01.0f.4d.26.13.27.ae.cb.00.00.00.00|0.0.1.c1", "name": "/" },
    { "guid": "01.0f.4d.26.25.27.c4.fd.00.00.00.00|0.0.1.c1", "name": "/" }
  ],
  "process_names": ["7580", "10003", "10021"],
  "robot_name": "C3PO",
  "run_id": "172",
  "users": ["root"]
}

latency_ns_mean and the throughput/RTPS fields are absent in this particular sample: nothing exchanged data on a matched DataWriter/DataReader pair within that 5-second poll window. See the Measurement's own page for why absence is what "nothing to report" looks like here.

Visualize the data

Open http://localhost:3000 (admin/admin) and pick the ROS 2 Data Collection - Fast DDS statistics dashboard. Its panels are backed by SQL queries against the dc PostgreSQL table (datasource uid dc_postgres), the same one every other PostgreSQL demo writes to — filtered to WHERE name = 'fastdds_stats':

• Write-to-notification latency, mean across matched DataWriter/DataReader pairs
• Publication and subscription throughput, mean
• Discovered participant/DataWriter/DataReader counts over time
• RTPS packets sent and lost
• The most recent sample's physical-layer data — which hosts, users and processes are behind the discovered participants

See tools/infrastructure/docker/config/grafana/dashboards/fastdds_stats.json for every panel's exact query.

Understanding the configuration

measurement_server:
  ros__parameters:
    measurement_plugins: ["fastdds_stats"]
    fastdds_stats:
      plugin: "dc_measurements/FastddsStats"
      topic_output: "/dc/measurement/fastdds_stats"
      polling_interval: 5000
      domain_id: 0
      include_measurement_name: true
      init_collect: true

dc_bridge:
  ros__parameters:
    destinations: ["records_log"]
    records_log:
      type: file
      receives: records
      inputs: ["/dc/measurement/fastdds_stats"]
      path: "/tmp/dc/fastdds_stats_pgsql_grafana_records.ndjson"
      time_key: "date"
      time_format: "double"
    custom_config_files: ["$HOME/.dc/fastdds_stats_pgsql_grafana_sink.toml"]

# ~/.dc/fastdds_stats_pgsql_grafana_sink.toml
[sinks.pgsql]
type = "postgres"
inputs = ["dc.dc.measurement.fastdds_stats"]
endpoint = "postgres://dc:password@127.0.0.1:5432/dc"
table = "dc"

[sinks.pgsql.buffer]
type = "disk"
max_size = 268435488

postgres — along with s3 and console — moved from the blessed ROS-param form to this passthrough recipe (ADR-0003); records_log is the cheap file anchor the passthrough still needs, since dc_bridge derives its ROS subscriptions and dc.<tag> routes from destinations alone. See Destinations: Recipes for the full recipe. domain_id is the DDS domain to monitor — the same value ROS_DOMAIN_ID would use for every other node in the deployment. include_measurement_name: true writes "name": "fastdds_stats" onto every Record, which every panel's WHERE name = 'fastdds_stats' clause relies on to tell this Measurement's rows apart from any other demo sharing the same dc table.

Custom uptime to Stdout

In this demo, we will go through a new use case. You want to create your own measurement or use an existing measurement provided by dc_measurements but not exactly, you want to add a field and also modify the JSON schema. That is what we will do here: create a new plugin, inside dc_demos, another package, create a plugin and load it

We are going to take the uptime measurement, change it slightly and collect the data.

To test it, run:

ros2 launch dc_demos uptime_custom_stdout.launch.py

JSON schema

Located in dc_demos/plugins/measurements/json/uptime_custom.json:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Uptime Custom",
  "description": "Time the system has been up. Intentionally failing to demonstrate customization and callback",
  "properties": {
    "time": {
      "description": "Time the system has been up",
      "type": "integer",
      "maximum": 0
    }
  },
  "type": "object"
}

It is almost the same as the standard uptime, but for the sake of the example, we will set the maximum value to 0, which will certainly make the validation fail!

CPP code

First, we create a hpp file in dc_demos/include/dc_demos/plugins/measurements:

#ifndef DC_DEMOS__PLUGINS__MEASUREMENTS__UPTIME_CUSTOM_HPP_
#define DC_DEMOS__PLUGINS__MEASUREMENTS__UPTIME_CUSTOM_HPP_

#include <nlohmann/json-schema.hpp>
#include <nlohmann/json.hpp>

#include "dc_measurements/measurement.hpp"
#include "dc_measurements/plugins/measurements/uptime.hpp"

namespace dc_demos
{
using json = nlohmann::json;

class UptimeCustom : public dc_measurements::Uptime
{
protected:
  void onFailedValidation(json data_json) override;
};

}  // namespace dc_demos

#endif  // DC_DEMOS__PLUGINS__MEASUREMENTS__UPTIME_CUSTOM_HPP_

We create a new class UptimeCustom, which inherits from dc_measurements::Uptime.

The method onFailedValidation is not mandatory but it is here to show it is possible to trigger a custom function when the validation fails.

Then the cpp code, currently located in dc_demos/plugins/measurements/uptime_custom.cpp:

#include "dc_demos/plugins/measurements/uptime_custom.hpp"

namespace dc_demos
{

void UptimeCustom::onFailedValidation(json data_json)
{
  (void)data_json;
  RCLCPP_INFO(logger_, "Callback! Validation failed for uptime custom");
}

}  // namespace dc_demos

#include "pluginlib/class_list_macros.hpp"
PLUGINLIB_EXPORT_CLASS(dc_demos::UptimeCustom, dc_core::Measurement)

We include the uptime_custom file header. Then, define the onFailedValidation function (triggered when validation fails).

The schema is picked up without any code: a Measurement validates against plugins/measurements/json/<plugin type>.json from the package registering it, so dc_demos/UptimeCustom reads dc_demos/plugins/measurements/json/uptime_custom.json. Set json_schema_path to point elsewhere, or enable_validator: false to turn validation off.

Plugin file

Create the xml file, here will be measurement_plugin.xml for us, at the source of the package. It defines the plugins of the package.

<class_libraries>
    <library path="dc_uptime_custom_measurement">
        <class name="dc_demos/UptimeCustom" type="dc_demos::UptimeCustom" base_class_type="dc_core::Measurement">
            <description>
                dc_measurement_uptime_custom
            </description>
        </class>
    </library>
</class_libraries>

CMakeLists.txt

Now that you have all files set up, you can add the build process to the CMakeLists.txt:

# Measurement plugins
add_library(dc_uptime_custom_measurement SHARED
  plugins/measurements/uptime_custom.cpp
)
list(APPEND dc_measurement_plugin_libs dc_uptime_custom_measurement)

foreach(measurement_plugin ${dc_measurement_plugin_libs})
  ament_target_dependencies(${measurement_plugin} ${dependencies})
  target_link_libraries(
    ${measurement_plugin}
    nlohmann_json::nlohmann_json
    nlohmann_json_schema_validator
  )
  target_compile_definitions(${measurement_plugin} PRIVATE BT_PLUGIN_EXPORT)
endforeach()

pluginlib_export_plugin_description_file(dc_core measurement_plugin.xml)

install(FILES measurement_plugin.xml
  DESTINATION share/${PROJECT_NAME}
)

install(DIRECTORY plugins/measurements/
  DESTINATION share/${PROJECT_NAME}/plugins/measurements/
)

It creates the library, installs and exports it.

Console output

In the measurement server log, the plugin is detected properly

[component_container_isolated-1] [INFO] [1788506243.714587305] [measurement_server]: Creating measurement plugin uptime_custom: Type dc_demos/UptimeCustom, Group key: uptime, Polling interval: 5000, Debug: 0, Validator enabled: 1, Schema path: , Tags: [], Init collect: 1, Init Max measurement: 0, Include measurement name: 1, Include measurement plugin name: 0, Remote keys: , Remote prefixes: , Nest: 0, Flatten: 0, Include measurement plugin name: 0, Max measurement on condition: 0, If all condition: , If any condition: , If none condition: , Gate condition: , Buffer duration sec: 0, Post roll duration sec: 0, Cooldown sec: 0, Max flush rate hz: 0, Flush topic: /dc/flush
[component_container_isolated-1] [INFO] [1788506243.724443694] [measurement_server]: Done configuring uptime_custom
[component_container_isolated-1] [INFO] [1788506243.724997657] [measurement_server]: Looking for schema at /root/ws/install/dc_demos/share/dc_demos/plugins/measurements/json/uptime_custom.json
[component_container_isolated-1] [INFO] [1788506243.725045689] [measurement_server]: schema: {"$schema":"http://json-schema.org/draft-07/schema#","description":"Time the system has been up. Intentionally failing to demonstrate customization and callback","properties":{"time":{"description":"Time the system has been up","maximum":0,"type":"integer"}},"title":"Uptime Custom","type":"object"}

Then, it fails as expected:

[component_container_isolated-1] [ERROR] [1788506243.726002578] [measurement_server]: Validation failed: At /time of 1638628 - instance exceeds maximum of 0
[component_container_isolated-1] data={"time":1638628}
[component_container_isolated-1] [INFO] [1788506243.726050184] [measurement_server]: Callback! Validation failed for uptime custom
[component_container_isolated-1] [ERROR] [1788506248.725121430] [measurement_server]: Validation failed: At /time of 1638633 - instance exceeds maximum of 0
[component_container_isolated-1] data={"time":1638633}
[component_container_isolated-1] [INFO] [1788506248.725329122] [measurement_server]: Callback! Validation failed for uptime custom

Even though every Record fails validation, dc_bridge still ships the raw data to the console Destination — validation failure only triggers onFailedValidation, it doesn't drop the Record:

[dc_bridge-2] {"date":1788506243.7255764,"flattened":false,"host":"127.0.0.1","name":"uptime_custom","nested":false,"source_type":"fluent","tag":"dc.measurement.uptime_custom","time":1638628,"timestamp":"2026-09-04T07:17:23.725576334Z"}

Concepts

DC has a small, fixed vocabulary. Using it precisely makes the rest of the documentation — and the configuration files — unambiguous.

Glossary

Relationships between them:

• A Measurement emits Records, optionally gated by one or more Conditions
• A Group merges Records from several Measurements into one Record
• The Bridge forwards every Record to the Shipper
• The Shipper delivers Records to one or more Destinations
• A File is uploaded to an object-storage Destination; its metadata becomes a Record
• A Record carries a Tag; each Tag names a route Destinations subscribe to
• A Trigger fires a FlushEvent; every Measurement listening for it releases its buffered window as one Incident

ROS 2

ROS 2 is the core middleware used for DC. If you are unfamiliar with it, visit the ROS 2 documentation before continuing.

Records

A Record is a single data unit in JSON, published on a ROS topic as a dc_interfaces/msg/StringStamped message. For example, a Record from the Memory Measurement:

{
    "date": "2022-12-04T14:16:06.810999008",
    "flattened": false,
    "memory": {
        "used": 76.007431
    },
    "nested": true
}

flattened and nested are added to every Record unconditionally by the base Measurement class — they report whether nested/flatten were configured for this plugin, not something you opt out of showing.

The StringStamped message carries:

1. header: ROS timestamp as std_msgs/Header
2. data: the Record, as a JSON string
3. group_key: the key this Record is nested under when merged into a Group
4. incident_id: the Incident a released Record belongs to; empty outside an incident

The data string must be valid JSON:

"Robot_X:1.5, Robot_Y:1.8" # Not a valid Record
"{'x': 1.5, 'y': 1.8}"     # Valid Record

Timestamps

The timestamp is the time the Record was created. Measurements convert ROS time to a UTC timestamp. Each Destination normalizes it into one field before delivery, controlled by time_key (default date) and time_format (epoch_nanos, the default, for exact integer nanoseconds since the epoch; iso8601 for a string; double for fractional seconds, which rounds — see Destinations).

JSON validation

Each Record is validated against a JSON schema by default, following the JSON Schema Validation specification. Each Measurement ships its own schema, which can be overridden from a custom package or disabled per Measurement with enable_validator: false. See Data validation.

Bridge and Shipper

DC's data plane is an external Shipper process, Vector, fed by the Bridge (dc_bridge) over a local socket. The Bridge renders the Shipper's entire configuration from plain ROS parameters — see Destinations — spawns and supervises the Shipper process, and forwards every Record it receives on its configured input topics.

DC gets the Shipper's disk buffering, backpressure handling, retries, and native Destination support (PostgreSQL, S3-compatible storage, and many more) without embedding or forking it (ADR-0001, ADR-0002).

Tags

Every Record carries a Tag, derived mechanically from the topic it was published on: the leading / is dropped and the remaining / become ..

Tags are what the Shipper routes on, and the dc.<tag> route names are stable public API that passthrough Destinations consume. A Destination selects what it receives through its own inputs list of topics — Tags are derived from that, not configured directly.

Destinations

A Destination is where data ends up: a file is blessed (rendered from plain ROS parameters), and so is another Shipper, and (only for File uploads, not Records) S3-compatible storage. PostgreSQL, S3-compatible storage for Records, and the console are reached through a passthrough recipe instead of the ROS-param form (per ADR-0003); any other Vector sink is reachable through the passthrough too. See Destinations for the full contract.

dc_bridge:
  ros__parameters:
    destinations: ["records_log"]             # Destination names to enable
    records_log:                              # a name you choose
      type: file                              # a blessed type
      receives: records
      inputs: ["/dc/measurement/uptime"]      # the topics this Destination receives
      path: "/tmp/dc/records.ndjson"

measurement_server:
  ros__parameters:
    measurement_plugins: ["uptime"]
    uptime:
      plugin: "dc_measurements/Uptime"
      topic_output: "/dc/measurement/uptime"  # matched by the Destination's `inputs`

Conditions

A Condition enables or disables one or more Measurements. For example, collect camera images only when the robot is stopped. A Measurement can require that all, any, or none of a set of Conditions are active. See Conditions.

Triggers and incidents

A Condition can only give you data from the moment it turned true. When what matters is the run-up to an event — the seconds before an emergency brake — a Trigger is the mechanism: a Measurement holds its recent output in a rolling buffer instead of publishing it, and a Trigger firing on the false→true edge of a Condition composition broadcasts a FlushEvent that releases that buffer. Everything released by one firing, across every Measurement listening, shares one incident_id. See Triggers.

Lifecycle Nodes and Bond

(Source: Nav2 documentation)

Lifecycle (or Managed, more correctly) nodes are unique to ROS 2. More information can be found here. They are nodes that contain state machine transitions for bringup and teardown of ROS 2 servers. This helps in deterministic behavior of ROS systems in startup and shutdown. It also helps users structure their programs in reasonable ways for commercial uses and debugging.

When a node is started, it is in the unconfigured state, only processing the node's constructor which should not contain any ROS networking setup or parameter reading. By the launch system, or the supplied lifecycle manager, the nodes need to be transitioned to inactive by configuring. After, it is possible to activate the node by transitioning through the activating stage.

This state will allow the node to process information and be fully setup to run. The configuration stage, triggering the on_configure() method, will setup all parameters, ROS networking interfaces, and for safety systems, all dynamically allocated memory. The activation stage, triggering the on_activate() method, will active the ROS networking interfaces and set any states in the program to start processing information.

To shutdown, we transition into deactivating, cleaning up, shutting down and end in the finalized state. The networking interfaces are deactivated and stop processing, deallocate memory, exit cleanly, in those stages, respectively.

The lifecycle node framework is used extensively through out this project and all servers utilize it. It is best convention for all ROS systems to use lifecycle nodes if it is possible.

Within DC, we use a wrapper of LifecycleNodes, nav2_util LifecycleNode from Nav2. This wrapper wraps much of the complexities of LifecycleNodes for typical applications. It also includes a bond connection for the lifecycle manager to ensure that after a server transitions up, it also remains active. If a server crashes, it lets the lifecycle manager know and transition down the system to prevent a critical failure. See Eloquent to Foxy for details.

See Lifecycle Manager for the managed-node list, the bond heartbeat and state-transition diagrams, and the autostart flow that ties into the Bridge readiness gate above.

Buffering and data persistence

The Shipper manages its own disk buffer (shipper.data_dir on the Bridge's parameters) with a documented minimum size (shipper.buffer_max_bytes). This buffer persists across reboots: Records accepted by the Bridge but not yet delivered to a Destination survive an outage of that Destination, a robot reboot, or a Bridge restart, and are delivered — with end-to-end acknowledgements — once the Destination is reachable again.

Scheduling and retries

The Shipper retries delivery to a Destination on failure with its own backoff, independent of DC code; see Vector's documentation for details. The Bridge itself is supervised by DC (launch respawn) and, in turn, supervises its own Shipper child process — including a Linux parent-death signal so the Shipper can never outlive the Bridge across a crash or SIGKILL.

Data Pipeline

This page follows one piece of data from the sensor that produced it to the external system that stores it. The vocabulary — Measurement, Record, Group, File, Bridge, Shipper, Destination, Tag — is defined in Concepts and used consistently throughout.

C4 model

Three levels, each zooming further into the pipeline: DC as a single system among external actors and destinations (Context), the processes that make it up (Container), and the pieces inside the Bridge (Component). The flowchart in the next section stays as the at-a-glance narrative view of a single Record's journey; these diagrams complement it rather than replace it.

DC and the systems it exchanges data with (C1)

DC runs as one system on the robot. A robot operator configures it; analytics and dashboard consumers read from whatever Destinations it was configured to write to — PostgreSQL, S3-compatible object storage, and (via passthrough, ADR-0003) any other Shipper-supported sink.

Inside DC (C2)

Inside DC, measurement_server is the node lifecycle-managed by dc_lifecycle_manager (see Lifecycle Manager for the managed-node list and why it's just the one node today). group_server runs as a plain node alongside it, not under lifecycle management. The Bridge (dc_bridge) and its supervised Shipper child are deliberately outside that boundary too (ADR-0006) — the Bridge has no meaningful deactivated state, so its readiness comes from launch ordering (bridge_ready_gate) instead of a lifecycle transition. See Deterministic startup ordering for the sequence this diagram's bridge_ready_gate → dc_lifecycle_manager relationship summarizes.

dc_uploader is a fourth, independent process dc_bringup.launch.py starts alongside the rest of the pipeline when a receives: files Destination is configured (ADR-0014) — it uploads Files and reports their status directly to the Shipper over its own connection, so an Uploader crash or restart never touches Record collection.

Inside dc_bridge and dc_uploader (C3)

The pieces added across #244–#267, now invisible from the outside: BridgeNode wires a Forwarder (Records → Shipper), a Supervisor (owns the Vector child process), a Config renderer (ADR-0003's shipper/destinations params → Vector TOML, including passthrough snippet validation), Readiness (backs ~/ready), and — for receives: files Destinations (ADR-0005) — a durable on-disk IntentQueue it enqueues into and forgets.

That queue is where the Bridge's responsibility for a File ends. dc_uploader (ADR-0014) is a separate process — its own executable, no rclcpp/rclpy dependency — that rescans the same on-disk queue, uploads Files against an S3-compatible ObjectStore, and reports status Records over its own Forwarder/Shipper connection under the dc.files Tag, entirely independent of the Bridge's own Forwarder. Killing or restarting dc_uploader never touches Record collection, since there is no shared address space left for it to take down.

The path of a Record

flowchart LR
    subgraph ros["ROS 2 graph"]
        meas["Measurement plugins<br/>(measurement_server)"]
        cond["Conditions"]
        group["Group node<br/>(group_server)"]
    end
    subgraph bridge["Bridge (dc_bridge)"]
        fwd["Forwarder"]
        iq[("Intent queue<br/>(disk)")]
    end
    subgraph uploader["dc_uploader (own process)"]
        upl["Uploader"]
        ufwd["Forwarder<br/>(own connection)"]
    end
    subgraph shipper["Shipper (Vector)"]
        route["dc.&lt;tag&gt; routes"]
        buf[("Disk buffer")]
    end
    subgraph dest["Destinations"]
        pg["PostgreSQL"]
        s3["S3-compatible storage"]
        other["Any Vector sink<br/>(passthrough)"]
    end

    cond -- gate --> meas
    meas -- "Records (StringStamped)" --> fwd
    meas -- "Records" --> group
    group -- "merged Records" --> fwd
    meas -. "Files on disk" .-> fwd
    fwd -. "enqueues intent" .-> iq
    iq -. "rescans (poll)" .-> upl
    fwd -- "shipper ingest protocol" --> route
    route --> buf
    buf --> pg
    buf --> other
    upl -- "File bytes" --> s3
    upl -- "status Record" --> ufwd
    ufwd -- "dc.files, its own connection" --> route

1. A Measurement produces a Record. Each Measurement plugin samples its source on a timer (or on an input topic) and publishes one timestamped JSON document as a dc_interfaces/msg/StringStamped on its topic_output. Conditions can gate whether the Measurement collects at all.
2. Optionally, a Group merges Records. The Group node subscribes to several Measurement topics and publishes one merged Record on /dc/group/<name> once their timestamps line up (sync_delay).
3. The Bridge forwards every Record. dc_bridge subscribes to every topic listed in any Destination's inputs, derives that topic's Tag, and hands the Record to the Shipper over the local shipper ingest socket (default 127.0.0.1:24224), with receipt acknowledgement.
4. The Shipper routes, buffers and delivers. Vector normalizes the Record's timestamp field, exposes it on the public dc.<tag> route, writes it to a persistent disk buffer, and delivers it to each Destination wired to that route — retrying with its own backoff until it succeeds.
5. Files take a different path. A File (camera image, map, video) is never sent through the Shipper. dc_bridge parses the local_paths/remote_paths references embedded in the Record and durably enqueues an intent to a shared on-disk queue — then forgets it. dc_uploader, a separate process (ADR-0014), rescans that same queue, uploads the bytes to object storage, verifies them, and emits a status Record under the dc.files Tag over its own Shipper connection — which then travels the ordinary Record path. See File uploads.

Where each piece is configured

Routing is decided in exactly one place: a Destination's inputs list names the topics it receives. Nothing on the producing side selects a Destination.

Deterministic startup ordering

dc_bringup.launch.py brings the pipeline up in a fixed order (ADR-0006), so no Record can be emitted before the pipeline is able to accept it:

1. Bridge first. dc_bridge starts as a plain node (outside the lifecycle manager) and spawns the Vector Shipper as a supervised child process. dc_uploader starts alongside it at this same step, as its own process, unless the run_uploader launch argument is False (ADR-0014) — it isn't gated by readiness the way the collection nodes are, since it has nothing to wait for beyond the on-disk intent queue it rescans. The measurement server also starts here, but stays unconfigured and inactive — its publishers cannot emit anything yet.
2. Readiness gate. A bridge_ready_gate process blocks, polling the Bridge's ~/ready service (std_srvs/Trigger), which answers success=True only once the Shipper is accepting connections on its ingest socket. The gate's service, timeout_s (default 120 s), and poll_interval_s parameters are configurable from the params file under bridge_ready_gate:.
3. Activation. Only when the gate exits successfully does the launch start lifecycle_manager_dc, which configures and then activates the collection nodes. If the Bridge never becomes ready before the gate's deadline, the whole launch shuts down loudly instead of leaving a half-started pipeline running.

See Lifecycle Manager for the diagrammed version of this sequence, plus the state transitions and bond-heartbeat recovery behavior it drives once activated.

Durability and supervision

• Disk buffering. The Shipper owns a persistent disk buffer at shipper.data_dir. A Record the Bridge has handed over survives a Destination outage, a Bridge restart, and a robot reboot; delivery resumes — with end-to-end acknowledgements — once the Destination is reachable again.
• Backpressure. When a Destination is slow, the Shipper stops acknowledging, and the Bridge propagates that backwards rather than dropping data silently.
• Supervision. The launch file respawns dc_bridge unconditionally (independent of use_respawn), and the Bridge supervises its Shipper child — including a Linux parent-death signal, so the Shipper can never outlive the Bridge even across a SIGKILL or crash.
• The one lossy window. Records published while the Bridge is down are dropped: ROS topics are fire-and-forget and nothing buffers upstream of the Bridge. Delivery resumes as soon as the respawned Bridge is ready.
• Delivery semantics. At-least-once. After a crash or an induced outage, a boundary Record may be re-sent; deduplicate on read if that matters to you.
• File uploads survive an Uploader crash. An intent is only removed from the disk queue after a successful upload is acknowledged; killing dc_uploader mid-upload loses nothing; the next start replays the same intent from disk.

Data validation

Model

Each Measurement can validate the Records it emits against a model.

We use use JSON schema validator for JSON for Modern C++.

Schemas follow the JSON 7 model:

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Uptime",
    "description": "Time the system has been up",
    "properties": {
        "time": {
            "description": "Time the system has been up",
            "type": "integer",
            "minimum": 0
        }
    },
    "type": "object"
}

Validation is enabled by default (enable_validator: true); disable it per plugin with enable_validator: false when fields are filtered out or added dynamically (e.g. remote paths) in a way the shipped schema doesn't account for.

Failed validation callback

You might want to trigger some actions when a validation fails, e.g send the data to another database to later on debug it.

In this case, you will need to write your own plugin (inherit from an existing one or start from scratch) and define the onFailedValidation function in the class.

This case is covered by the custom plugin demo

Use a different Schema

For each plugin, a default path is provided but this can be changed by passing the json_schema_path parameter in the measurement plugin parameter to the absolute path of your schema.

Groups

Description

A Group merges the Records of several Measurements into one Record, based on time proximity. For example, grouping the cpu and position Measurements publishes a single merged Record on /dc/group/my_group:

{
    "cpu": ...,
    "position": ...,
}

A merged Record is an ordinary Record from there on: it carries the Tag derived from its output topic, and a Destination receives it by listing that topic in inputs.

Envelope fields do not get nested under a member's key: a member's tags is replaced by the Group's own, and the incident_id on its envelope — the id a Measurement stamps on a Record it released as part of an incident — is carried onto the merged Record's own envelope (first non-empty wins, so a partial Record built from a mix of released and live members still carries it). The Bridge lifts that envelope field into the payload's top level, where a postgres Destination has a column for it; buried under <group_key>.incident_id it would simply be dropped by that sink. A member payload that is not a JSON object — a bare scalar or an array — has no such fields to lift, and is merged as it is under its group_key.

Node parameters

Group parameters

Incomplete sets

A Group only publishes once every one of its inputs has produced a Record close enough in time. If one Measurement stops publishing — a camera unplugged, a plugin crashed, a gate_condition that never opens — the whole Group goes silent, and the Records of the inputs that are working never reach a Destination.

sync_timeout puts a deadline on that wait. It is measured from the arrival of the first Record of a set, and it is a different thing from sync_delay: sync_delay is the synchroniser's slop, the maximum spread between the timestamps of Records that may be merged together, whereas sync_timeout is real time spent waiting for a set to complete.

drop is the default, so a Group that sets nothing new behaves exactly as it did before. emit_partial needs a positive sync_timeout to have a deadline to fire on; configured without one, the Group logs an error and falls back to drop.

Warnings and throttling

Every timeout logs a warning naming the group, the deadline it missed, and the input topics that produced nothing — including under drop, where the warning is the only trace left since nothing is published:

[WARN] [group_server]: Group 'memory_cpu': no complete set after 10.0s, dropping the
incomplete set (no Record from: ['/dc/measurement/memory'])

A Measurement that dies stays dead, so the group keeps timing out and an unthrottled warning would repeat every sync_timeout for as long as the robot runs. sync_timeout_log_throttle caps this at one warning per group per window (60s by default); set it to 0.0 to log every timeout. The first timeout after a quiet window always logs immediately, and reports how many warnings the throttle swallowed since the last one:

[WARN] [group_server]: Group 'memory_cpu': no complete set after 10.0s, dropping the
incomplete set (no Record from: ['/dc/measurement/memory']) [+5 more in the last 60.0s]

The throttle is per group, so a permanently broken group does not silence the warnings of a healthy one that starts failing. Only the log line is rate-limited — under emit_partial every timeout still publishes its partial Record, throttled or not.

What a partial Record looks like

A partial Record has the same shape as a complete one, minus the inputs that did not report, plus two keys that mark it as partial:

{
    "cpu": { "used": 12.0 },
    "partial": true,
    "missing_inputs": ["/dc/measurement/memory"],
    "tags": [""],
    "name": "memory_cpu"
}

• A missing input's key is omitted, not present-and-null. The Group node knows a missing input's topic and nothing else: a Record's group_key and its fields both travel inside the Record itself, so there is no correctly shaped null placeholder the Group could invent for a Measurement that never published. Consumers with a flat column schema — the postgres sink maps a Record's top-level keys onto existing columns — see those columns stay NULL for that row, the same as any other column a Record does not populate.
• partial and missing_inputs appear only on partial Records. A complete Record is byte-for-byte what the Group published before this option existed, so nothing downstream has to change to keep consuming complete Records. Treat the absence of partial as false.
• missing_inputs lists the input topics that contributed no Record, in inputs order.
• tags, name and plugins behave as they do on a complete Record — plugins only lists the plugins that actually reported.
• incident_id behaves the same way: if any member that did arrive was released as part of an incident, the merged Record carries its incident_id, so a partial Record of an incident is still queryable as one.

Example

group_server:
  ros__parameters:
    groups: ["memory_cpu", "memory_uptime", "cameras", "map"]
    memory_cpu:
      inputs: ["/dc/measurement/memory", "/dc/measurement/cpu"]
      output: "/dc/group/memory_cpu"
      sync_delay: 5.0
      group_key: "memory_cpu"
    memory_uptime:
      inputs: ["/dc/measurement/memory", "/dc/measurement/uptime"]
      output: "/dc/group/memory_uptime"
      sync_delay: 5.0
      # Publish whatever arrived rather than nothing if one of the two is 10s late
      sync_timeout: 10.0
      on_sync_timeout: "emit_partial"
      group_key: "memory_uptime"
    cameras:
      inputs: ["/dc/measurement/camera"]
      output: "/dc/group/cameras"
      sync_delay: 5.0
      group_key: "cameras"
    map:
      inputs: ["/dc/measurement/map"]
      output: "/dc/group/map"
      sync_delay: 5.0
      group_key: "map"

Lifecycle Manager

dc_lifecycle_manager is DC's nav2-style manager for the nodes that have a meaningful deactivated state. It walks each managed node through the ROS 2 managed-node state machine (configure, activate, deactivate, cleanup, shutdown), watches a bond heartbeat on every node it has activated, and can bring the whole managed set up automatically at launch. See Concepts for the general nav2_util LifecycleNode background this page builds on.

What it manages

The set of managed nodes is the node_names parameter, walked in list order for bring-up (configure then activate, each transition applied to every node before the next one starts) and in reverse order for teardown. In every params file this repo ships — dc_bringup.launch.py's inline lifecycle_manager_params and both E2E harness params files — that list is:

lifecycle_manager_dc:
  ros__parameters:
    node_names: ["measurement_server"]
    transitions: [configure, activate]

measurement_server is the only node under lifecycle management today. group_server, when enabled, is launched as a plain Node alongside it — not added to node_names — so it is not lifecycle-managed. Nothing stops a future node_names entry from adding it if it ever needs a deactivated state; the manager is already list-driven for exactly that reason.

The boundary and the autostart flow

dc_bringup.launch.py brings the pipeline up in a fixed order (ADR-0006) so that no Record can be emitted before the Bridge can accept it. The Bridge, its Vector Shipper child, and the bridge_ready_gate process that polls the Bridge's ~/ready (std_srvs/Trigger) service all run outside lifecycle_manager_dc — only once the gate exits 0 does the lifecycle manager exist at all in the launch graph.

flowchart TB
    subgraph outside["Outside the lifecycle manager (ADR-0006)"]
        direction TB
        vector["Vector<br/>(Shipper child process)"]
        bridge["dc_bridge<br/>(plain node, launch respawn)"]
        gate["bridge_ready_gate<br/>(polls ~/ready)"]
        bridge -- "spawns and supervises" --> vector
        bridge -- "~/ready (std_srvs/Trigger)" --> gate
    end
    subgraph managed["Managed by lifecycle_manager_dc"]
        direction TB
        lm["lifecycle_manager_dc<br/>(autostart=true by default)"]
        meas["measurement_server<br/>(only entry in node_names)"]
        lm -- "configure, then activate" --> meas
        meas -- "bond heartbeat (10 Hz)" --> lm
    end

    gate -- "exit 0: Bridge ready" --> lm
    gate -. "exit != 0: never ready" .-> abort["Shutdown(reason=...)<br/>whole launch aborts"]

1. Bridge first. dc_bridge starts as a plain node and spawns Vector as a supervised child process.
2. Readiness gate. bridge_ready_gate blocks until ~/ready answers success=True (Vector accepting connections on its ingest socket), or its own timeout_s (default 120 s) expires.
3. Activation, only on success. dc_bringup.launch.py registers an OnProcessExit handler on the gate: exit 0 starts lifecycle_manager_dc; any other exit code shuts the whole launch down (Shutdown(reason=...)) instead of leaving collection nodes running against a Bridge that never came up.
4. Autostart. dc_bringup.launch.py's autostart launch argument defaults to True and is always passed through explicitly — overriding the code's own default of false for a manager launched some other way. With autostart=true, lifecycle_manager_dc calls its own startup() as soon as it has constructed lifecycle service clients for every managed node, with no external service call needed. autostart=false leaves the managed set in Unconfigured until something calls the manager's ~/manage_nodes service (nav2_msgs/srv/ManageLifecycleNodes, command STARTUP).

State transitions and bond heartbeats

Each managed node moves through the same five primary states nav2 uses. The manager's five service-level operations (startup, shutdown, reset, pause, resume) are each just a sequence of these per-node transitions applied across every managed node — pause/resume reuse the same deactivate/activate edges startup uses, not separate ones.

A bond (10 Hz heartbeat, bond_timeout — 10 s in dc_bringup.launch.py, 4 s default otherwise) is created for a node the moment it activates and torn down the moment it deactivates. The manager polls every bond every 200 ms; a missed heartbeat past bond_timeout is treated as that node having crashed.

stateDiagram-v2
    [*] --> Unconfigured
    Unconfigured --> Inactive: configure
    Inactive --> Active: activate / resume
    Active --> Inactive: deactivate / pause
    Inactive --> Unconfigured: cleanup
    Unconfigured --> Finalized: shutdown
    Finalized --> [*]

    Active --> Unconfigured: bond heartbeat lost -> reset(hard_reset=true)
    Unconfigured --> Active: node reachable again within\nbond_respawn_max_duration -> startup()

A lost heartbeat does not just deactivate the one node that crashed — checkBondConnections hard-resets every managed node (deactivate then cleanup, continuing past per-node failures since hard_reset=true) and clears all bonds, on the principle that a half-alive managed set is worse than a fully torn-down one. If attempt_respawn_reconnection (default true) is set, a 1 s-period timer then polls whether every managed node's lifecycle service is reachable again:

• Reachable within bond_respawn_max_duration (default 10 s): the manager calls startup() again — a full configure + activate pass — and resumes normal operation, bonds included.
• Still unreachable once bond_respawn_max_duration elapses: the manager gives up and leaves the managed set Unconfigured. Recovery from there needs an explicit STARTUP call to ~/manage_nodes (or a relaunch).

~/is_active (std_srvs/srv/Trigger) reports whether the managed set is currently Active, and a diagnostic_updater entry surfaces the same status on /diagnostics.

Measurements

Description

A Measurement is a source of sampled data that emits Records — timestamped JSON documents — on its own ROS topic. For example, a Record from the Memory Measurement:

{
    "flattened": false,
    "memory": {
        "used": 76.007431
    },
    "nested": true,
    "run_id": "358"
}

flattened/nested/run_id are added unconditionally by this node — see below. Fields like robot_name only appear once configured as a custom key (see custom_key_str_list above), and date/host/source_type/tag/ timestamp are added later, by Vector once the Bridge forwards the Record — see a captured example on any demo page for the full shape a Destination actually receives.

Node parameters

This node collects data and publishes it as Records. Each Measurement is a pluginlib plugin loaded into this node and publishes on its own topic_output; the Bridge (dc_bridge) subscribes to those topics and forwards the Records to the Destinations that list them in inputs (see Destinations). Conditions are pluginlib plugins loaded here too — optional predicates that gate whether a Measurement collects, e.g. only when the robot is not moving.

robot_name resolution

robot_name is a custom key like any other, but when it appears in custom_key_str_list its value resolves in a fixed order rather than always being a literal string, so a fleet does not need one hand-edited params file per robot:

1. custom_keys_str.robot_name.value — a literal string, unchanged from before.
2. custom_keys_str.robot_name.value_from_file — the contents of a file, e.g. one written by the provisioning process.
3. The machine's hostname — the default when neither of the above is set.

A value_from_file that names a file that cannot be read, or any source that resolves to an empty string, fails node configuration with a clear error rather than shipping Records with a missing or blank robot_name.

Custom keys on Files

The keys listed in custom_key_str_list label a Measurement's Files as well as its Records: dc_uploader (ADR-0014, a separate process from the Bridge) writes them into the file_status and group_complete Records it emits for that Measurement's Files, so both sides of a Destination carry the same labelling. A Record names its custom keys in a custom_keys field for that purpose.

Two limits are worth knowing. A custom key whose name is one the Uploader computes itself (group_name, local_path, remote_path, storage_type, uploaded, size, …) is not written — the Uploader's own value is kept and dc_uploader logs the collision. The keys the rows already carry, robot_name and id (as robot_id), are likewise not repeated, and are not reported: those values are in the row either way. And the column still has to exist in the Destination: the PostgreSQL sink maps JSON keys onto existing columns 1:1, so a new custom key needs an ALTER TABLE on dc_files the same way it needs one on dc_records.

Plugin parameters

Each measurement is collected through a node and has these configuration parameters:

Available plugins:

Battery

Description

Records the state of one battery pack from a sensor_msgs/BatteryState topic: charge percentage, voltage and current on the polling interval, plus a Record when a charging session starts and another when it ends. Charging is unavailable time, so the session boundaries are what turn battery data into a shift-utilisation number downstream.

The Measurement emits facts, not metrics: "charging started at 14:02:11 after a discharge of 62 %", never "battery availability is 87 %". Aggregation belongs in SQL views, where the time window is a query parameter.

Every Record carries an event field naming which of the three it is:

Sessions are delimited by the pack's power_supply_status, never by a percentage threshold, so a noisy percentage cannot open and close sessions repeatedly; a status of unknown carries no information and leaves an open session open. Completed cycles are accumulated from discharge depth rather than counted as full discharges — two half discharges are one cycle, not two — so a robot topped up at every dock still reports the wear it actually did. That accounting lives in dc_common::BatteryCycleAccumulator, which has no ROS dependency and is tested on its own; the plugin subscribes, delegates and serialises.

sensor_msgs/BatteryState leaves most fields optional and signals "unmeasured" with NaN. A field the hardware doesn't fill is left out of the Record rather than written as null, so a pack that reports only a voltage still produces a valid Record. Until the input topic publishes at all, no Record is emitted: a gap in the data simply means the battery hasn't reported yet.

Battery health is reported the way the hardware reports it — power_supply_health when the pack sends one, and health_percentage (capacity against design_capacity) only when it sends both.

A robot with two packs runs one Measurement per pack, each with its own topic and topic_output. Like every other Measurement, it can be gated by Conditions and merged into a Group.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Battery",
  "description": "Battery state of one pack: charge percentage, voltage and current on the polling interval, plus a Record at each charging session boundary",
  "properties": {
    "event": { "type": "string", "enum": ["sample", "charge_session_start", "charge_session_end"] },
    "power_supply_status": { "type": "string", "enum": ["unknown", "charging", "discharging", "not_charging", "full"] },
    "percentage": { "type": "number", "minimum": 0, "maximum": 100 },
    "voltage": { "type": "number" },
    "current": { "type": "number" },
    "charge": { "type": "number" },
    "capacity": { "type": "number" },
    "design_capacity": { "type": "number" },
    "health_percentage": { "type": "number", "minimum": 0 },
    "temperature": { "type": "number" },
    "present": { "type": "boolean" },
    "power_supply_health": { "type": "string" },
    "power_supply_technology": { "type": "string" },
    "location": { "type": "string" },
    "serial_number": { "type": "string" },
    "completed_cycles": { "type": "integer", "minimum": 0 },
    "session_id": { "type": "integer", "minimum": 1 },
    "duration_sec": { "type": "number", "minimum": 0 },
    "discharge_depth_percent": { "type": "number", "minimum": 0 },
    "start_percentage": { "type": "number", "minimum": 0, "maximum": 100 },
    "end_percentage": { "type": "number", "minimum": 0, "maximum": 100 },
    "charged_percent": { "type": "number" }
  },
  "required": ["event"],
  "type": "object"
}

The full file (plugins/measurements/json/battery.json) also requires power_supply_status on a sample; session_id and discharge_depth_percent on a session start; and session_id plus duration_sec on a session end — the fields the views depend on. Everything else is hardware dependent and therefore optional.

Configuration

...
battery:
  plugin: "dc_measurements/Battery"
  topic_output: "/dc/measurement/battery"
  polling_interval: 10000
  topic: "/battery_state"

Two packs:

...
battery_left:
  plugin: "dc_measurements/Battery"
  topic_output: "/dc/measurement/battery_left"
  topic: "/left/battery_state"
battery_right:
  plugin: "dc_measurements/Battery"
  topic_output: "/dc/measurement/battery_right"
  topic: "/right/battery_state"

Example output

One sample:

{
  "event": "sample",
  "percentage": 62.0,
  "voltage": 48.4,
  "current": -12.5,
  "power_supply_status": "discharging",
  "power_supply_health": "good",
  "capacity": 42.0,
  "design_capacity": 50.0,
  "health_percentage": 84.0,
  "present": true,
  "completed_cycles": 3,
  "serial_number": "PACK-A"
}

A charging session, start and end:

{
  "event": "charge_session_start",
  "session_id": 4,
  "percentage": 31.0,
  "discharge_depth_percent": 62.0
}

{
  "event": "charge_session_end",
  "session_id": 4,
  "duration_sec": 3600.0,
  "start_percentage": 31.0,
  "end_percentage": 97.0,
  "charged_percent": 66.0
}

Camera

Description

Save camera image files: raw, rotated and/or inspected. Images can be inspected using different detection modules (e.g. barcode/QR detection, in-process via ZXing-C++)

Parameters

Code pose estimation

With estimate_pose: true, every detected code carries a pose alongside its bounding box. The pose is solved from the code's four detected corners with cv::solvePnP (SOLVEPNP_IPPE_SQUARE, the solver for four coplanar corners of a square), which needs two things detection alone does not: the code's physical side length (code_size, in meters) and the camera intrinsics, read from camera_info_topic. It is off by default and costs nothing when off — no camera_info subscription is created and Records are unchanged.

Frame and convention

The pose is the code's pose, not the robot's — where the code is as seen from the robot. By default it is expressed in the camera optical frame (the frame_id of the camera_info message, following REP 103: X right, Y down, Z forward along the lens axis), so z is the depth of the code in front of the camera. The code's own frame is centered on the code and uses those same axes — X right, Y down, Z into its printed face — so a code seen square-on has the identity orientation, and roll/pitch/yaw read as how far off square-on it was. (This is the ArUco/OpenCV marker frame turned 180° about X: that convention puts Z out of the face towards the camera, which would make a square-on read a 180° roll.)

Set pose_frame to have the pose transformed into a robot frame (base_link, map, …) via TF before it is written. Every Record says which frame it is in: the emitted pose.frame_id is the frame actually used, so if the transform is unavailable within transform_timeout the pose is still reported — in the camera optical frame, with a warning logged, rather than dropped. pose.distance is the camera-to-code range in meters and is unaffected by pose_frame.

rotation_angle is handled: the corners are mapped back to raw-image coordinates before the solve, so the intrinsics still describe the image they were calibrated on.

camera:
  plugin: "dc_measurements/Camera"
  cam_topic: "/front_camera/image_raw"
  # camera_info_topic defaults to /front_camera/camera_info, next to cam_topic
  cam_name: my_camera_with_codes
  detection_modules: ["barcode"]
  estimate_pose: true
  code_size: 0.2       # meters, side length of the printed code
  pose_frame: "base_link"
  transform_timeout: 0.1

{
  "camera_name": "my_camera_with_codes",
  "inspected": {
    "barcode": [
      {
        "data": "0001", "type": "QRCode",
        "top": 210, "left": 295, "width": 84, "height": 84,
        "pose": {
          "frame_id": "base_link",
          "x": 1.482, "y": 0.037, "z": 0.611,
          "roll": 0.0, "pitch": 0.0, "yaw": 3.139,
          "distance": 1.483
        }
      }
    ]
  }
}

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Camera",
  "description": "Camera images with detected objects",
  "properties": {
    "camera_name": {
      "description": "Name of the camera",
      "type": "string"
    },
    "local_paths": {
      "description": "Paths of saved images",
      "type": "object",
      "items": {
        "$ref": "#/$defs/paths"
      }
    },
    "remote_paths": {
      "description": "Dictionary of paths where metadata and images will be remotely stored",
      "type": "object",
      "additionalProperties": {
        "type": "object",
        "items": {
          "$ref": "#/$defs/paths"
        }
      }
    },
    "inspected": {
      "description": "Inspected content of an image",
      "type": "object",
      "items": {
        "$ref": "#/$defs/inspected"
      }
    }
  },
  "$defs": {
    "paths": {
      "type": "object",
      "properties": {
        "raw": {
          "description": "Raw image",
          "type": "string"
        },
        "rotated": {
          "description": "Rotated image",
          "type": "string"
        },
        "inspected": {
          "description": "Inspected image",
          "type": "string"
        }
      }
    },
    "inspected": {
      "type": "object",
      "properties": {
        "barcode": {
          "description": "Barcode inspected data",
          "type": "array",
          "items": {
            "$ref": "#/$defs/barcode"
          }
        }
      }
    },
    "barcode": {
      "type": "object",
      "properties": {
        "data": {
          "description": "Barcode data",
          "type": "string"
        },
        "height": {
          "description": "Barcode height",
          "type": "integer"
        },
        "width": {
          "description": "Barcode width",
          "type": "integer"
        },
        "top": {
          "description": "Barcode top position",
          "type": "integer"
        },
        "left": {
          "description": "Barcode left position",
          "type": "integer"
        },
        "type": {
          "description": "Barcode type",
          "type": "string"
        },
        "pose": {
          "description": "Pose of the code, present only when estimate_pose is enabled",
          "$ref": "#/$defs/pose"
        }
      }
    },
    "pose": {
      "type": "object",
      "properties": {
        "frame_id": {
          "description": "Frame the pose is expressed in",
          "type": "string"
        },
        "x": {
          "description": "Code position along the frame's X axis, in meters",
          "type": "number"
        },
        "y": {
          "description": "Code position along the frame's Y axis, in meters",
          "type": "number"
        },
        "z": {
          "description": "Code position along the frame's Z axis, in meters",
          "type": "number"
        },
        "roll": {
          "description": "Code orientation about the frame's X axis, in radians",
          "type": "number"
        },
        "pitch": {
          "description": "Code orientation about the frame's Y axis, in radians",
          "type": "number"
        },
        "yaw": {
          "description": "Code orientation about the frame's Z axis, in radians",
          "type": "number"
        },
        "distance": {
          "description": "Straight-line distance from the camera to the code, in meters",
          "type": "number"
        }
      }
    }
  },
  "type": "object"
}

Configuration

The remote paths are also saved in the JSON under <measurement_name>._img_paths.(raw|rotated|inspected). If images want to be sent to a self-hosted S3-compatible store such as RustFS, add "rustfs" in remote_keys. This will add a remote path that can later be used in your API.

Note that this remote key is not included in the JSON schema, which only contains the local paths. If you want to enforce the schema with your custom remote key, you will need to write it and load it manually.

...
camera:
  plugin: "dc_measurements/Camera"
  group_key: "camera_with_codes"
  topic_output: "/dc/measurement/camera_with_codes"
  polling_interval: 10000
  init_collect: true
  node_name: "dc_measurement_camera"
  cam_topic: "/camera_with_codes"
  cam_name: my_camera_with_codes
  enable_validator: false
  draw_det_barcodes: true
  save_raw_img: true
  save_rotated_img: false
  save_detections_img: true
  save_raw_path: "camera_with_codes/raw/%Y-%m-%dT%H-%M-%S"
  save_rotated_path: "camera_with_codes/rotated/%Y-%m-%dT%H-%M-%S"
  save_inspected_path: "camera_with_codes/inspected/%Y-%m-%dT%H-%M-%S"
  rotation_angle: 0
  detection_modules: ["barcode"]
  remote_prefixes: [""]
  remote_keys: ["rustfs"] # Will create paths for RustFS, does not send the file

Destination (dc_bridge) configuration

Now that the path is set, it can be used to know where to send the image. The Destination name (rustfs) must match the remote_keys entry above — the Uploader matches a Record's remote_paths keys against receives: files Destination names (see Destinations):

dc_bridge:
  ros__parameters:
    destinations: ["rustfs", "pgsql"]
    rustfs:
      type: s3
      receives: files
      inputs: ["/dc/group/cameras"]
      endpoint: "http://127.0.0.1:9000"
      access_key_id: "XEYqG4ZcPY5jiq5i"
      secret_access_key: "ji011KCtI82ZeQS6UwsQAg8x9VR4lSaQ"
      force_path_style: true
      bucket: "mybucket"
    files:
      metadata_destination: "pgsql"  # a receives: records Destination for status rows

Example output

With estimate_pose: true (same sample as Code pose estimation above):

{
  "camera_name": "my_camera_with_codes",
  "inspected": {
    "barcode": [
      {
        "data": "0001", "type": "QRCode",
        "top": 210, "left": 295, "width": 84, "height": 84,
        "pose": {
          "frame_id": "base_link",
          "x": 1.482, "y": 0.037, "z": 0.611,
          "roll": 0.0, "pitch": 0.0, "yaw": 3.139,
          "distance": 1.483
        }
      }
    ]
  }
}

A plain capture with estimate_pose: false (no pose field, otherwise identical shape) has not been added yet.

Cmd_vel

Description

Collect command velocity sent to the robot by subscribing to cmd_vel topic.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Cmd_vel",
    "description": "Command velocity sent to the robot",
    "properties": {
        "computed": {
            "description": "Computed command velocity in meter/s",
            "type": "number"
        },
        "linear": {
            "description": "Linear velocity as a vector",
            "type": "object",
            "items": {
                "$ref": "#/$defs/vector3"
            }
        },
        "angular": {
            "description": "Angular velocity as a vector",
            "type": "object",
            "items": {
                "$ref": "#/$defs/vector3"
            }
        }
    },
    "$defs": {
        "vector3": {
            "type": "object",
            "properties": {
                "x": {
                    "description": "X speed",
                    "type": "number"
                },
                "y": {
                    "description": "Y speed",
                    "type": "number"
                },
                "z": {
                    "description": "Z speed",
                    "type": "number"
                }
            }
        }
    },
    "type": "object"
}

Configuration

...
cmd_vel:
  plugin: "dc_measurements/CmdVel"
  group_key: "cmd_vel"
  topic_output: "/dc/measurement/cmd_vel"

Example output

{
  "angular": {
    "x": 0.0,
    "y": 0.0,
    "z": 0.1
  },
  "computed": 0.3,
  "flattened": false,
  "linear": {
    "x": 0.3,
    "y": 0.0,
    "z": 0.0
  },
  "name": "cmd_vel",
  "nested": false,
  "run_id": "169"
}

CPU

Description

Collect cpu usage: average cpu, number of processes running and processes sorted by cpu usage.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Cpu",
    "description": "CPU statistics",
    "properties": {
        "average": {
            "description": "Average CPU",
            "type": "number",
            "minimum": 0
        },
        "processes": {
            "description": "Number of processes running",
            "type": "integer",
            "minimum": 0
        },
        "sorted": {
            "description": "Processes sorted by CPU usage",
            "type": "array",
            "items": {
                "$ref": "#/$defs/process"
            }
        }
    },
    "$defs": {
        "process": {
            "type": "object",
            "description": "Process information",
            "properties": {
                "pid": {
                    "description": "Process ID of the process",
                    "type": "integer"
                },
                "user": {
                    "description": "User who started the process",
                    "type": "string"
                },
                "cmd": {
                    "description": "Command that launched the process",
                    "type": "string"
                },
                "cpu": {
                    "description": "Process' current utilization as a percentage of total CPU time",
                    "type": "number"
                },
                "ram": {
                    "description": "Memory in use by this process in kb",
                    "type": "integer"
                },
                "uptime": {
                    "description": "Age of the process in seconds",
                    "type": "integer"
                }
            }
        }
    },
    "type": "object"
}

Configuration

...
cpu:
  plugin: "dc_measurements/Cpu"
  topic_output: "/dc/measurement/cpu"
  max_processes: 10
  cpu_min: 10.0

Example output

{
  "average": 20.487500071525574,
  "flattened": false,
  "name": "cpu",
  "nested": false,
  "processes": 8,
  "run_id": "169",
  "sorted": []
}

sorted is empty here because no process crossed cpu_min (10%) in this poll window.

Diagnostics

Description

Subscribes to /diagnostics (diagnostic_msgs/DiagnosticArray) and converts matching DiagnosticStatus entries into a Record, so hardware/driver health reaches Destinations and dashboards like any other Measurement. Each DiagnosticStatus.values key/value pair is preserved as-is in the Record rather than being flattened into the message string.

/diagnostics is typically high-volume and mostly unchanging, so level_threshold and names are provided to shrink what gets collected. Pair this Measurement with the same_as_previous condition (if_none_conditions) to also skip republishing when nothing has changed since the previous collection.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Diagnostics",
  "description": "Diagnostic statuses collected from /diagnostics",
  "properties": {
    "statuses": {
      "description": "Diagnostic statuses matching the configured level threshold and name allowlist",
      "type": "array",
      "items": {
        "title": "DiagnosticStatus",
        "description": "A single diagnostic_msgs/DiagnosticStatus entry",
        "properties": {
          "name": {
            "description": "Reporting component name",
            "type": "string"
          },
          "message": {
            "description": "Human-readable status summary",
            "type": "string"
          },
          "hardware_id": {
            "description": "Hardware identifier",
            "type": "string"
          },
          "level": {
            "description": "Status level: 0=OK, 1=WARN, 2=ERROR, 3=STALE",
            "type": "integer",
            "minimum": 0,
            "maximum": 3
          },
          "values": {
            "description": "Key/value pairs reported by the status, preserved as-is",
            "type": "object"
          }
        },
        "required": ["name", "message", "level", "values"],
        "type": "object"
      }
    }
  },
  "required": ["statuses"],
  "type": "object"
}

Configuration

...
diagnostics:
  plugin: "dc_measurements/Diagnostics"
  topic_output: "/dc/measurement/diagnostics"
  level_threshold: "WARN"
  names: ["motor_driver", "battery"]
  if_none_conditions: ["diagnostics_unchanged"]

diagnostics_unchanged:
  plugin: "dc_conditions/SameAsPrevious"
  keys: []
  exclude: []

Example output

{
  "flattened": false,
  "name": "diagnostics",
  "nested": false,
  "run_id": "169",
  "statuses": [
    {
      "hardware_id": "motor_driver_01",
      "level": 2,
      "message": "Overcurrent detected",
      "name": "motor_driver",
      "values": {
        "current_amps": "8.4"
      }
    }
  ]
}

Dummy

Description

The dummy measurement, generates dummy events. It is useful for testing, debugging, benchmarking and getting started with ROS 2 Data collection.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Dummy",
    "description": "Dummy JSON",
    "properties": {
        "message": {
            "description": "Dummy message",
            "type": "string"
        }
    },
    "type": "object"
}

Configuration

...
dummy:
  plugin: "dc_measurements/Dummy"
  topic_output: "/dc/measurement/dummy"

Example output

{
  "flattened": false,
  "message": "Hello from ROS 2 DC",
  "name": "dummy",
  "nested": false,
  "run_id": "169"
}

Distance traveled

Description

Reports the straight-line distance moved between the robot's current TF pose (robot_base_frame in global_frame) and its pose at the previous poll — a per-poll delta, despite the field's name. Sum it downstream (e.g. SUM(distance_traveled) in a SQL view) to get a cumulative total since power-on.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Distance traveled",
    "description": "Total distance traveled in meters by the robot",
    "properties": {
        "distance_traveled": {
            "description": "Total distance traveled in meters",
            "type": "number"
        }
    },
    "type": "object"
}

Configuration

...
distance_traveled:
  plugin: "dc_measurements/DistanceTraveled"
  topic_output: "/dc/measurement/distance_traveled"
  global_frame: "map"
  robot_base_frame: "base_link"
  transform_timeout: 0.1

Example output

The poll during which the robot's TF pose moved 3 m since the previous poll:

{
  "distance_traveled": 3.0,
  "flattened": false,
  "name": "distance_traveled",
  "nested": false,
  "run_id": "169"
}

The next poll, stationary:

{
  "distance_traveled": 0.0,
  "flattened": false,
  "name": "distance_traveled",
  "nested": false,
  "run_id": "169"
}

Driving type

Description

Reports the robot's current operating mode (autonomous, manual, teleop, or unknown) as a single Record, so every other Measurement can be segmented by driving mode downstream. There is no standard ROS message for "current driving mode", so this Measurement is entirely configuration driven and supports two common shapes, chosen by which parameters are set (configuring both is a configuration error):

• Dedicated mode topic: subscribes to mode_topic (std_msgs/String) and maps each raw value it carries to a mode through value_mapping_from/value_mapping_to. A raw value with no entry in the mapping is ignored (the previous mode is kept) rather than treated as unknown, since an unrecognized value is more likely an upstream hiccup than an actual mode change.
• Velocity source inference: subscribes to velocity_topics (geometry_msgs/Twist, e.g. one topic per command source such as a Nav2 output and a joystick teleop node) and reports the mode of whichever configured source last published, mapped through the parallel velocity_modes list. A source that hasn't published within velocity_timeout_s is no longer considered active.

The emitted mode is always one of the four values above -- a documented, closed set -- so downstream grouping/dashboards never see an unbounded string. Before any mode has been observed (no Measurement configured, a dedicated mode topic that hasn't published yet, or every velocity source past its timeout) the Measurement reports "unknown" rather than skipping the Record: a Record is always published on every poll, so a gap in driving_type data means the plugin itself stopped, not "mode currently unknown".

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "DrivingType",
  "description": "Current driving/operating mode of the robot, so every other metric can be segmented by mode",
  "properties": {
    "mode": {
      "description": "Driving mode: 'unknown' until a mode has been observed (dedicated mode topic) or a configured velocity source has published within 'velocity_timeout_s' (velocity-source inference)",
      "type": "string",
      "enum": ["autonomous", "manual", "teleop", "unknown"]
    }
  },
  "required": ["mode"],
  "type": "object"
}

Configuration

Dedicated mode topic:

...
driving_type:
  plugin: "dc_measurements/DrivingType"
  topic_output: "/dc/measurement/driving_type"
  mode_topic: "/driving_mode_raw"
  value_mapping_from: ["0", "1", "2"]
  value_mapping_to: ["manual", "autonomous", "teleop"]

Velocity source inference:

...
driving_type:
  plugin: "dc_measurements/DrivingType"
  topic_output: "/dc/measurement/driving_type"
  velocity_topics: ["/nav2/cmd_vel", "/teleop/cmd_vel"]
  velocity_modes: ["autonomous", "teleop"]
  velocity_timeout_s: 1.0

Example output

{
  "mode": "autonomous"
}

Fast DDS statistics

Description

Reads eProsima Fast DDS's own Statistics Module — latency, throughput and RTPS packet counts between the DomainParticipants, DataWriters and DataReaders it discovers on a DDS domain — through Fast-DDS-statistics-backend, and emits one sample Record per polling interval. It follows the same periodic-sample convention as Battery's sample event and Uptime: everything reported is scoped to the window since the previous poll, reset each time rather than accumulated since the Measurement started.

Unlike every other Measurement, fastdds_stats has no input topic: it starts a StatisticsBackend monitor on a DDS domain at onConfigure() and queries that domain's own statistics registry directly on each poll, rather than subscribing to anything. A robot's whole DDS graph (every node's participants, on whatever domain it runs) is visible to one Measurement instance, so normally one fastdds_stats Measurement per robot is enough.

Each sample reports:

• Discovered entity counts (participant_count, datawriter_count, datareader_count)
• Mean write-to-notification latency across every matched DataWriter/DataReader pair (latency_ns_mean, nanoseconds — Fast DDS's own unit for this statistic)
• Mean data rate sent and received (publication_throughput_bytes_per_sec_mean, subscription_throughput_bytes_per_sec_mean, bytes/second)
• RTPS packets sent and lost, summed across every participant (rtps_packets_sent, rtps_packets_lost)
• Physical-layer data: each discovered participant's name/guid, and the names of the hosts, OS users and processes running them (hosts, users, process_names — not processes, which the CPU Measurement's total process count already owns in the shared dc table) — useful for spotting which machine or process is actually behind a noisy participant on a multi-process robot

A field tied to a DataKind (latency, throughput, packet counts) is absent when nothing reported data in the window — a domain with one lonely participant and no matched DataWriter/DataReader pair yet still produces a valid Record, just without a latency_ns_mean.

Prerequisites

1. Fast DDS itself built with -DFASTDDS_STATISTICS=ON — the Statistics Module is compiled out by default, and most distro/apt Fast DDS builds do not enable it. ROS 2 Rolling pairs with Fast-DDS 3.x — rebuild the Fast-DDS set upstream's ros2.repos pins for this distro in a colcon workspace overlay with that cmake arg, then source the overlay before the rest of the workspace so it shadows the apt-installed Fast DDS.

2. fastdds_statistics_backend built and installed against that same Fast-DDS. Pin a v2.x tag — that line is what targets Fast-DDS 3.0.0 (find_package(fastdds 3.0.0 REQUIRED)), which is what Rolling ships; the older v1.0.0/v1.1.0 releases target Fast-DDS >=2.13.0 and won't build against it. No rosdep/apt key exists for any distro, so rosdep install never pulls it in and dc_measurements/package.xml deliberately does not list it as a <depend>.

3. The FASTDDS_STATISTICS environment variable, set on every process before it creates its first DomainParticipant — the library only emits the DataKinds named in it (semicolon- separated topic aliases), regardless of whether the plugin is built and running:

   export FASTDDS_STATISTICS="HISTORY_LATENCY_TOPIC;PUBLICATION_THROUGHPUT_TOPIC;SUBSCRIPTION_THROUGHPUT_TOPIC;RTPS_SENT_TOPIC;RTPS_LOST_TOPIC"
   

   Without it, participant_count/datawriter_count/datareader_count/participants/hosts/ users/process_names still populate (basic discovery data), but latency_ns_mean and every throughput/RTPS-packet field stay permanently absent — not intermittently, every single poll.

Without #1/#2, find_package(fastdds_statistics_backend) fails at CMake configure time and the plugin — and its test — are skipped from the build entirely.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "FastddsStats",
  "properties": {
    "event": { "type": "string", "enum": ["sample"] },
    "domain_id": { "type": "integer", "minimum": 0 },
    "participant_count": { "type": "integer", "minimum": 0 },
    "datawriter_count": { "type": "integer", "minimum": 0 },
    "datareader_count": { "type": "integer", "minimum": 0 },
    "latency_ns_mean": { "type": "number", "minimum": 0 },
    "publication_throughput_bytes_per_sec_mean": { "type": "number", "minimum": 0 },
    "subscription_throughput_bytes_per_sec_mean": { "type": "number", "minimum": 0 },
    "rtps_packets_sent": { "type": "integer", "minimum": 0 },
    "rtps_packets_lost": { "type": "integer", "minimum": 0 },
    "participants": { "type": "array" },
    "hosts": { "type": "array", "items": { "type": "string" } },
    "users": { "type": "array", "items": { "type": "string" } },
    "process_names": { "type": "array", "items": { "type": "string" } }
  },
  "required": ["event", "domain_id", "participant_count", "datawriter_count", "datareader_count"],
  "type": "object"
}

The full file (plugins/measurements/json/fastdds_stats.json) also spells out each participants[] entry's name/guid properties.

Configuration

...
fastdds_stats:
  plugin: "dc_measurements/FastddsStats"
  topic_output: "/dc/measurement/fastdds_stats"
  polling_interval: 5000
  domain_id: 0

Example output

Captured from a real run (all three prerequisites above met):

{
  "custom_keys": ["robot_name"],
  "datareader_count": 1,
  "datawriter_count": 11,
  "domain_id": 0,
  "event": "sample",
  "flattened": false,
  "hosts": ["d:14058711922191368192"],
  "name": "fastdds_stats",
  "nested": false,
  "participant_count": 3,
  "participants": [
    { "guid": "01.0f.4d.26.9c.1d.72.46.00.00.00.00|0.0.1.c1", "name": "/" },
    { "guid": "01.0f.4d.26.13.27.ae.cb.00.00.00.00|0.0.1.c1", "name": "/" },
    { "guid": "01.0f.4d.26.25.27.c4.fd.00.00.00.00|0.0.1.c1", "name": "/" }
  ],
  "process_names": ["7580", "10003", "10021"],
  "robot_name": "C3PO",
  "run_id": "172",
  "users": ["root"]
}

latency_ns_mean and every throughput/RTPS field are absent here: nothing exchanged data on a matched DataWriter/DataReader pair within this particular 5-second poll window, and per the Statistics Backend's own contract, absence is how "nothing to report" is signaled. A busier DDS graph (more topics, higher rate) makes them appear more often, though never guaranteed on every poll.

<footer id="open-on-gh">Found a bug? <a href="https://github.com/minipada/ros2_data_collection/edit/main/doc/src/dc/measurements/fastdds_stats.md">Edit this page on GitHub.</a></footer>

Fault

Description

Reports the transitions diagnostics's periodic snapshots leave a consumer to reconstruct: diagnostics writes the current DiagnosticStatus level on every poll, so recovering "when did this component break, and when was it fixed" means scanning for where a value changed in SQL -- exactly the failure mode transition Records exist to prevent. Fault subscribes to the same /diagnostics topic (diagnostic_msgs/DiagnosticArray) and emits one Record per component only when its level actually changes, feeding each watched component's stream of (level, timestamp) samples to its own dc_common::StateTransitionDetector, a header-only dc_common type generic over the state being tracked and owning no clock of its own.

Every Record carries the component, the level it left (from_level) and entered (to_level), how long the previous level had been held (previous_level_duration_s), and the status message as the reason. Levels are the DiagnosticStatus constants OK, WARN, ERROR and STALE -- reported by name, so a silent (STALE) component is never confused with a broken (ERROR) one.

Each transition is also classified as an event:

• raise -- the component left OK. A new fault opens ("state": "open"), and fault_started_at records when.
• change -- a transition between two non-OK levels (e.g. WARN to ERROR). The fault stays open; fault_started_at is unchanged from the raise that opened it.
• clear -- the component returned to OK. The fault closes ("state": "closed") and the Record carries duration_s, the length of the whole fault, alongside the same fault_started_at the raise carried -- so the pair can be joined downstream for MTTR. MTBF is the gap between successive raise events for a component.

A fault still open when collection stops keeps its last Record's "state": "open" and no duration_s: no closing Record is invented, so it cannot be read as a zero-length outage. Every Record carries a seq that increments by one across every watched component, so a dropped Record is a detectable gap rather than a silently shortened outage. A component observed already faulted on its very first sample has no fault_started_at on that Record -- the detector treats the first sample as a baseline rather than a transition, since it never saw the fault actually start.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Fault",
  "properties": {
    "seq": { "type": "integer", "minimum": 1 },
    "component": { "type": "string" },
    "event": { "type": "string", "enum": ["raise", "clear", "change"] },
    "from_level": { "type": "string", "enum": ["OK", "WARN", "ERROR", "STALE"] },
    "to_level": { "type": "string", "enum": ["OK", "WARN", "ERROR", "STALE"] },
    "previous_level_duration_s": { "type": "number", "minimum": 0 },
    "reason": { "type": "string" },
    "state": { "type": "string", "enum": ["open", "closed"] },
    "fault_started_at": { "type": "string" },
    "duration_s": { "type": "number", "minimum": 0 }
  },
  "required": ["seq", "component", "event", "from_level", "to_level", "previous_level_duration_s", "reason", "state"],
  "type": "object"
}

Configuration

...
fault:
  plugin: "dc_measurements/Fault"
  topic_output: "/dc/measurement/fault"
  group_key: "fault"
  topic: "/diagnostics"
  names: ["motor_driver", "battery"]

Example output

A raise:

{
  "seq": 1,
  "component": "motor_driver",
  "event": "raise",
  "from_level": "OK",
  "to_level": "ERROR",
  "previous_level_duration_s": 3612.4,
  "reason": "Motor fault",
  "state": "open",
  "fault_started_at": "2026-08-18T09:12:33.123456Z"
}

and its clear:

{
  "seq": 2,
  "component": "motor_driver",
  "event": "clear",
  "from_level": "ERROR",
  "to_level": "OK",
  "previous_level_duration_s": 214.9,
  "reason": "Motor nominal",
  "state": "closed",
  "fault_started_at": "2026-08-18T09:12:33.123456Z",
  "duration_s": 214.9
}

Intervention

Description

Reports human takeovers: how often somebody had to step in, how long the robot had been running itself beforehand, and how long the takeover lasted. It is a projection of driving-mode transitions -- it reads the same mode signal Driving type reads (the shared DrivingModeSource, so both Measurements are configured with the same parameters) and hands it to dc_common::StateTransitionDetector (#360), rather than running its own separate detection. The two therefore cannot disagree about when the robot was autonomous.

A takeover is a transition between autonomous and one of the two human-driven modes, manual or teleop -- this set is fixed and has no configuration knob, because dc_kpi_intervention_events (tools/infrastructure/sql/kpi_views.sql, #369) already matches these values literally; a custom mode name would silently never count downstream.

Two Records per takeover:

• a start Record, when the mode leaves autonomous for manual or teleop. open is true.
• an end Record, when the mode returns to autonomous. open is false.

Both carry from_mode, to_mode, and previous_duration: the dwell of from_mode, in seconds. On a start Record that is how long the robot had been autonomous before the takeover; on an end Record it is exactly how long the takeover itself lasted, because from_mode is the mode that just ended either way. sequence is the transition's own monotonically increasing number (from StateTransitionDetector), so a dropped Record shows up as a gap rather than silently corrupting a duration.

A takeover still open when collection stops simply never gets a matching end Record: nothing downstream can average an interval that was never closed as a zero, because there is no end Record to average. Nothing is published on a poll that saw no takeover boundary, so unlike driving_type this Measurement is silent most of the time. A transition that is not a boundary -- autonomy handing over to unknown because the mode signal went stale, or one human mode replacing another directly -- produces no Record.

Rates (interventions per autonomous hour, per kilometre travelled) are deliberately not computed here: they are views over these Records (dc_kpi_intervention_rate), so the denominator can change without touching a robot. Set group_key to merge an intervention with position in a Group and put the takeover on the site map.

Parameters

The mode-source parameters are the same ones Driving type takes, with the same meaning: mode_topic, value_mapping_from, value_mapping_to, velocity_topics, velocity_modes, velocity_timeout_s. Configuring both mode_topic and velocity_topics is a configuration error.

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Intervention",
  "description": "A human takeover, derived from driving-mode transitions: one Record when it starts, one when it ends",
  "properties": {
    "event": { "type": "string", "enum": ["start", "end"] },
    "from_mode": { "type": "string", "enum": ["autonomous", "manual", "teleop", "unknown"] },
    "to_mode": { "type": "string", "enum": ["autonomous", "manual", "teleop", "unknown"] },
    "previous_duration": { "type": "number", "minimum": 0 },
    "sequence": { "type": "integer", "minimum": 1 },
    "open": { "type": "boolean" }
  },
  "required": ["event", "from_mode", "to_mode", "previous_duration", "sequence", "open"],
  "type": "object"
}

Configuration

...
intervention:
  plugin: "dc_measurements/Intervention"
  topic_output: "/dc/measurement/intervention"
  group_key: "intervention"
  velocity_topics: ["/cmd_vel_smoothed", "/teleop/cmd_vel"]
  velocity_modes: ["autonomous", "teleop"]
  velocity_timeout_s: 1.0

Example output

Start:

{
  "event": "start",
  "from_mode": "autonomous",
  "to_mode": "teleop",
  "previous_duration": 412.5,
  "sequence": 7,
  "open": true
}

and end:

{
  "event": "end",
  "from_mode": "teleop",
  "to_mode": "autonomous",
  "previous_duration": 37.2,
  "sequence": 8,
  "open": false
}

Ip Camera

Description

Records video in small segments (in case of a cut) and store it locally. They are first stored in a temporary folder. Once the record (of e.g 10 seconds) is done, it is moved to another directory.

Compared to other plugins, the collect function only moves the files from the temporary location, it does not start the recording. It takes some time to establish connection, so we avoid doing this every time. Recording is started at initialization by an ffmpeg process and saved in HLS format.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Ip Camera",
    "description": "Local and remote path where the remote camera video is recorded",
    "properties": {
        "local_path": {
            "description": "Local video path",
            "type": "string"
        },
        "remote_path": {
            "description": "Remote video path",
            "type": "string"
        }
    },
    "type": "object"
}

Configuration

...
ip_camera:
  plugin: "dc_measurements/IpCamera"
  topic_output: "/dc/measurement/ip_camera"
  input: "rtsp://192.168.0.10:554/stream1"
  segment_time: 10
  save_path: "ip_camera/%Y-%m-%dT%H-%M-%S"

Example output

Captured from a real run, input pointed at a real-time MPEG-TS/TCP test stream (save_local_base_path overridden per the warning above):

{
  "data_src": "ip_camera",
  "flattened": false,
  "local_path": "/root/dc_capture_out/ip_camera/ffmpeg_2026-09-06T00:02:30.ts",
  "name": "ip_camera",
  "nested": false,
  "remote_path": "/ip_camera/ffmpeg_2026-09-06T00:02:30.ts",
  "run_id": "169",
  "timestamp": "ffmpeg_2026-09-06T00:02:30"
}

Manipulation

Description

Reports one MoveIt MoveGroup goal's lifecycle: a manipulation_start Record when the goal is accepted, a manipulation_end Record when it reaches a terminal state. This is a manipulation goal — it moves a robot arm for one planning group. MoveIt has its own outcome vocabulary here: one flat signed MoveItErrorCodes space, distinct from nav2's succeeded/failed/cancelled/aborted split used by the Mission Measurements.

The Measurement is a passive observer: it never sends the goal itself. It reads the action's status topic (<action_name>/_action/status, published for every goal the server knows about, regardless of which client sent it) to see a goal appear and reach a terminal state, then calls the action's get_result service -- a plain service any client may call given the goal's UUID, not only the one that sent it -- to read MoveGroup::Result's error_code and planning_time. group_name is this instance's own configuration rather than something read off the goal: MoveGroup's Goal (the only place a group name appears) is sent privately to the action server and never broadcast, so a passive observer structurally cannot see it. Run one Manipulation Measurement instance per planning group your robot moves.

outcome on the end Record is derived from error_code, not from the action's own terminal status: SUCCESS maps to succeeded, PREEMPTED maps to cancelled (the closest thing MoveIt has), and every other -- necessarily negative -- MoveItErrorCodes value maps to failed. sequence is a monotonically increasing counter across every Record this instance emits, so a dropped Record shows up as a gap rather than silently corrupting a downstream count.

A goal still executing when collection stops simply never gets a matching manipulation_end Record: nothing downstream can average an interval that was never closed as a zero, because there is no end Record to average.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Manipulation",
  "description": "One MoveIt MoveGroup goal's lifecycle: a Record when it is accepted, one when it reaches a terminal state",
  "properties": {
    "event": {
      "description": "Whether this Record is the start or the end of one manipulation goal",
      "type": "string",
      "enum": ["manipulation_start", "manipulation_end"]
    },
    "goal_id": {
      "description": "The MoveGroup goal's own UUID, as a canonical hex string, correlating a start Record with its end",
      "type": "string"
    },
    "group_name": {
      "description": "The planning group this Measurement instance watches, as configured -- not read off the goal, which is never broadcast",
      "type": "string"
    },
    "sequence": {
      "description": "Monotonically increasing across every Record from this Measurement instance, so a dropped Record shows up as a gap",
      "type": "integer",
      "minimum": 1
    },
    "outcome": {
      "description": "Derived from MoveIt's own error_code: SUCCESS succeeds, PREEMPTED is the closest thing MoveIt has to cancelled, everything else fails",
      "type": "string",
      "enum": ["succeeded", "cancelled", "failed"]
    },
    "error_code": {
      "description": "MoveIt's own numeric MoveItErrorCodes, carried through verbatim",
      "type": "integer"
    },
    "planning_time": {
      "description": "Seconds MoveGroup spent planning, from MoveGroup::Result",
      "type": "number",
      "minimum": 0
    },
    "duration_sec": {
      "description": "How long the goal took from accepted to terminal",
      "type": "number",
      "minimum": 0
    }
  },
  "required": ["event", "goal_id", "group_name", "sequence"],
  "type": "object"
}

The full file (plugins/measurements/json/manipulation.json) also requires outcome, error_code, planning_time and duration_sec on a manipulation_end Record.

Configuration

...
manipulation:
  plugin: "dc_measurements/Manipulation"
  topic_output: "/dc/measurement/manipulation"
  group_key: "manipulation"
  action_name: "move_action"
  group_name: "arm"

Example output

Start:

{
  "event": "manipulation_start",
  "goal_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "group_name": "arm",
  "sequence": 5
}

and end:

{
  "event": "manipulation_end",
  "goal_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "group_name": "arm",
  "sequence": 6,
  "outcome": "succeeded",
  "error_code": 1,
  "planning_time": 0.842,
  "duration_sec": 3.15
}

Map

Description

Save map using nav2_map_server and collect the map of the local map saved. The measurement also includes metadata: width, height and x and y origin.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Map",
    "description": "Map saved metadata and paths",
    "properties": {
        "resolution": {
            "description": "Resolution of the map, meters/pixel",
            "type": "number",
            "minimum": 0
        },
        "local_paths": {
            "description": "Paths where metadata and image are stored",
            "type": "object",
            "items": {
                "$ref": "#/$defs/paths"
            }
        },
        "remote_paths": {
            "description": "Dictionary of paths where metadata and image will be remotely stored",
            "type": "object",
            "additionalProperties": {
                "type": "object",
                "items": {
                    "$ref": "#/$defs/paths"
                }
            }
        },
        "origin": {
            "description": "Robot origin position in meters",
            "type": "object",
            "items": {
                "$ref": "#/$defs/origin"
            }
        },
        "width": {
            "description": "Width of the PGM",
            "type": "integer",
            "minimum": 0
        },
        "height": {
            "description": "Height of the PGM",
            "type": "integer",
            "minimum": 0
        }
    },
    "$defs": {
        "origin": {
            "type": "object",
            "description": "The 2-D pose of the lower-left pixel in the map, as (x, y, yaw), with yaw as counterclockwise rotation (yaw=0 means no rotation). Many parts of the system currently ignore yaw.",
            "properties": {
                "x": {
                    "description": "X origin of the robot",
                    "type": "number"
                },
                "y": {
                    "description": "Y origin of the robot",
                    "type": "number"
                }
            }
        },
        "paths": {
            "type": "object",
            "properties": {
                "png": {
                    "description": "Path to the map PNG file containing the image",
                    "type": "string"
                },
                "yaml": {
                    "description": "Path to the map YAML file containing map metadata",
                    "type": "string"
                },
                "pgm": {
                    "description": "Path to the map PGM file containing the gray-scale image",
                    "type": "string"
                }
            }
        }
    },
    "type": "object"
}

Configuration

...
map:
  plugin: "dc_measurements/Map"
  topic_output: "/dc/measurement/map"
  topic: "/map"
  save_path: "map/%Y-%m-%dT%H:%M:%S"
  save_map_timeout: 0.2
  quiet: true
  remote_keys: ["s3"]

Example output

A 4x4 test grid, captured without remote_keys configured (so no remote_paths):

{
  "flattened": false,
  "height": 4,
  "local_paths": {
    "pgm": "/tmp/dc_capture/map_out/2026-09-05T23:46:20.pgm",
    "png": "/tmp/dc_capture/map_out/2026-09-05T23:46:20.png",
    "yaml": "/tmp/dc_capture/map_out/2026-09-05T23:46:20.yaml"
  },
  "name": "map",
  "nested": false,
  "origin": {
    "x": 0.0,
    "y": 0.0
  },
  "resolution": 0.05000000074505806,
  "run_id": "169",
  "width": 4
}

Memory

Description

Collect memory used in percentage.

Parameters

This Measurement has no parameters beyond the common Plugin parameters.

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Memory",
    "description": "Memory used",
    "properties": {
        "used": {
            "description": "Memory used in percent",
            "type": "number",
            "minimum": 0
        }
    },
    "type": "object"
}

Configuration

...
memory:
  plugin: "dc_measurements/Memory"
  topic_output: "/dc/measurement/memory"

Example output

{
  "flattened": false,
  "name": "memory",
  "nested": false,
  "run_id": "169",
  "used": 91.44149017333984
}

Mission (nav2 FollowWaypoints)

Description

Reports the outcome of a nav2 FollowWaypoints mission -- a patrol or waypoint-following run, including nav2's own per-waypoint failure reporting -- as one mission_start Record when a goal is accepted and one mission_end Record when it reaches a terminal state. It is the FollowWaypoints sibling of the NavigateToPose Mission Measurement (#387): same Record schema, same mission_id/sequence conventions, its own mission_type.

This Measurement is a passive watcher: it never sends a FollowWaypoints goal itself. Whatever already dispatches waypoint-following missions on the robot -- nav2_simple_commander, a WMS integration, a teleop panel -- keeps doing exactly that; this Measurement only observes. That is a deliberate match to every other Measurement's read-only relationship to the systems it reports on, and it is also why it does not use rclcpp_action::Client's typed goal-tracking API: that API only reports on goals the client itself sent. Instead it subscribes directly to the action's _action/status topic (action_msgs/msg/GoalStatusArray, the same message type for every action) to see a goal get accepted or reach a terminal state, and calls the action's _action/get_result service directly once it does, to fetch the Result -- both standard, public parts of the ROS 2 action wire protocol that any client may use, sender or not.

Known limitation: number_of_loops

nav2's FollowWaypoints goal carries a number_of_loops field (how many times to repeat the route), but the goal itself is never re-published anywhere a third party can observe it -- only the sender and the action server ever see it, and nothing in the ROS 2 action protocol (nor rclcpp_action::Client's public API) lets a client that did not send a goal read it back. A passive watcher therefore cannot report number_of_loops, and this Measurement's Records do not carry it. Sending the goal itself instead of watching it would make this Measurement responsible for driving navigation, contradicting DC's role as a telemetry pipeline (see CONTEXT.md) and directly competing with whatever else already commands the same action -- e.g. dc_demos/dc_demos/qrcodes_waypoint_follower.py, which already sends FollowWaypoints goals of its own. If a deployment needs number_of_loops on this Record, it has to come from the mission commander via a documented escape hatch (#305's territory), not from this adapter.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "MissionNav2FollowWaypoints",
  "properties": {
    "event": { "type": "string", "enum": ["mission_start", "mission_end"] },
    "mission_id": { "type": "string" },
    "mission_type": { "type": "string", "const": "follow_waypoints" },
    "sequence": { "type": "integer", "minimum": 1 },
    "outcome": { "type": "string", "enum": ["succeeded", "failed", "cancelled", "aborted"] },
    "reason": { "type": "string" },
    "error_code": { "type": "integer", "minimum": 0 },
    "duration_sec": { "type": "number", "minimum": 0 },
    "missed_waypoints": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "index": { "type": "integer", "minimum": 0 },
          "error_code": { "type": "integer", "minimum": 0 }
        },
        "required": ["index", "error_code"]
      }
    }
  },
  "required": ["event", "mission_id", "mission_type", "sequence"],
  "type": "object"
}

outcome, duration_sec and missed_waypoints are required on mission_end only. reason and error_code are required on mission_end when outcome is failed or aborted: failed is a goal that reached nav2's GoalStatus.SUCCEEDED but whose FollowWaypoints::Result.error_code is still non-zero -- nav2's WaypointFollower can finish a route having missed a waypoint without the action itself aborting.

Configuration

...
mission:
  plugin: "dc_measurements/MissionNav2FollowWaypoints"
  topic_output: "/dc/measurement/mission"
  group_key: "mission"
  action_name: "follow_waypoints"

Example output

Start:

{
  "event": "mission_start",
  "mission_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "mission_type": "follow_waypoints",
  "sequence": 5
}

end, succeeded:

{
  "event": "mission_end",
  "mission_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "mission_type": "follow_waypoints",
  "sequence": 6,
  "outcome": "succeeded",
  "duration_sec": 184.2,
  "missed_waypoints": []
}

and end, failed on a missed waypoint:

{
  "event": "mission_end",
  "mission_id": "6c9f6a3e-2d1a-4e3a-9f7a-1b2c3d4e5f60",
  "mission_type": "follow_waypoints",
  "sequence": 8,
  "outcome": "failed",
  "reason": "task executor failed",
  "error_code": 601,
  "duration_sec": 92.6,
  "missed_waypoints": [
    { "index": 2, "error_code": 601 }
  ]
}

error_code is FollowWaypoints::Result.error_code verbatim -- in nav2_msgs, its only non-zero values are UNKNOWN (600) and TASK_EXECUTOR_FAILED (601); missed_waypoints[].error_code is nav2's per-waypoint error_code from the same table. nav2's WaypointStatus (new since Lyrical) also carries the waypoint's pose, a status enum and a free-text error_msg -- none part of this schema (the index key keeps the Jazzy-line Record shape downstream consumers already receive); reason/error_msg above are always the mission-level FollowWaypoints::Result.error_msg, not anything per-waypoint.

Mission Nav2 (NavigateToPose)

Description

The nav2 adapter of the Mission Measurement for nav2's NavigateToPose action -- a single-pose navigation goal, the base case the mission lifecycle contract (#305, recorded in ADR-0010) was agreed against. Emits a mission_start Record when a goal is accepted and a mission_end Record once it reaches a terminal state. NavigateThroughPoses and FollowWaypoints are its siblings (#388/#389): same Record schema, same mission_id/sequence conventions, each with its own mission_type.

This Measurement is a passive watcher: it never sends a NavigateToPose goal itself. Whatever already dispatches navigation missions on the robot -- nav2's bt_navigator, a fleet orchestrator, an operator command -- keeps doing exactly that; this Measurement only observes. That is a deliberate match to every other Measurement's read-only relationship to the systems it reports on, and it is also why it does not use rclcpp_action::Client's typed goal-tracking API: that API only reports on goals the client itself sent. Instead it subscribes directly to the action's _action/status (action_msgs/msg/GoalStatusArray, the same message type for every action) and _action/feedback topics to see a goal get accepted, pick up its recovery count, and reach a terminal state, and calls the action's _action/get_result service directly once it does, to fetch the Result -- all standard, public parts of the ROS 2 action wire protocol that any client may use, sender or not.

nav2's NavigateToPose action server processes one goal at a time, so this Measurement only ever tracks one mission at once. A second goal accepted before the first reaches a terminal state is logged and otherwise ignored -- it produces no Record of its own, and the mission already being tracked is unaffected.

outcome on the end Record is one of succeeded, failed, cancelled, aborted: CANCELED maps to cancelled, ABORTED to aborted (both carrying nav2's own error_msg/error_code as reason/error_code), and SUCCEEDED maps to succeeded unless NavigateToPose::Result.error_code is non-zero, in which case it is failed -- an application-level failure nav2 reported without aborting the goal status itself.

recoveries (from NavigateToPose::Feedback.number_of_recoveries, when feedback was seen before completion) is carried on the end Record when available.

A mission still running when collection stops simply never gets a matching mission_end Record: nothing downstream can average an interval that was never closed as a zero, because there is no mission_end Record to average. sequence is a monotonically increasing counter across every Record this Measurement instance emits (start and end alike, not per mission_id), so a dropped Record shows up as a gap rather than silently corrupting a duration or a mission-success-rate denominator.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "MissionNav2",
  "properties": {
    "event": { "type": "string", "enum": ["mission_start", "mission_end"] },
    "mission_id": { "type": "string" },
    "mission_type": { "type": "string", "const": "navigate_to_pose" },
    "sequence": { "type": "integer", "minimum": 1 },
    "outcome": { "type": "string", "enum": ["succeeded", "failed", "cancelled", "aborted"] },
    "reason": { "type": "string" },
    "error_code": { "type": "integer", "minimum": 0 },
    "duration_sec": { "type": "number", "minimum": 0 },
    "recoveries": { "type": "integer", "minimum": 0 }
  },
  "required": ["event", "mission_id", "mission_type", "sequence"],
  "type": "object"
}

outcome and duration_sec are required on mission_end only. reason and error_code are required on mission_end when outcome is failed or aborted.

Configuration

...
mission:
  plugin: "dc_measurements/MissionNav2"
  topic_output: "/dc/measurement/mission"
  group_key: "mission"
  action_name: "navigate_to_pose"

Example output

Start:

{
  "event": "mission_start",
  "mission_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "mission_type": "navigate_to_pose",
  "sequence": 7
}

and end (succeeded):

{
  "event": "mission_end",
  "mission_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "mission_type": "navigate_to_pose",
  "sequence": 8,
  "outcome": "succeeded",
  "duration_sec": 96.4,
  "recoveries": 1
}

and end (aborted):

{
  "event": "mission_end",
  "mission_id": "6c9f6a3e-2d1a-4e3a-9f7a-1b2c3d4e5f60",
  "mission_type": "navigate_to_pose",
  "sequence": 10,
  "outcome": "aborted",
  "reason": "tf timeout",
  "error_code": 9102,
  "duration_sec": 12.1
}

Mission Nav2 (NavigateThroughPoses)

Description

The nav2 adapter of the Mission Measurement for nav2's NavigateThroughPoses action -- the action a deployment issues when a mission is a single job through several hard-constraint poses in one call, rather than a chain of separate NavigateToPose goals (that sibling adapter is mission_nav2, #387). Emits a mission_start Record when a goal is first observed and a mission_end Record once it reaches a terminal state, in the same Record schema #387 defines, with mission_type: "navigate_through_poses".

This Measurement is a passive observer of the action server's single active goal. It never sends a goal itself. Instead it subscribes to the action's own _action/status and _action/feedback topics and calls its _action/get_result service directly for whichever goal_id just reached a terminal status -- the same standard per-action topics/services every rclcpp_action::Server (nav2's bt_navigator included) exposes. Whatever in the deployment actually dispatches missions (a BT navigator, a fleet orchestrator, an operator command) keeps doing so exactly as before; this Measurement only watches.

Two Records per mission:

• a mission_start Record, the first time a goal_id is observed on the action's status topic.
• a mission_end Record, once that goal_id reaches a terminal status (SUCCEEDED, CANCELED, or ABORTED) and its result has been fetched.

outcome on the end Record is one of succeeded, failed, cancelled, aborted: CANCELED maps to cancelled, ABORTED to aborted (both carrying nav2's own error_msg/error_code as reason/error_code), and SUCCEEDED maps to succeeded unless NavigateThroughPoses::Result.error_code is non-zero, in which case it is failed -- an application-level failure nav2 reported without aborting the goal status itself.

recoveries (from NavigateThroughPoses::Feedback.number_of_recoveries, when feedback was seen before completion) is carried on the end Record when available.

No per-waypoint outcome is reported. The acceptance criteria this Measurement was built against ask for the final waypoint_statuses (which poses in the job succeeded versus failed) to be represented on mission_end, following an upstream nav2 NavigateThroughPoses.action result field of that name. That field exists on the nav2 this repository actually builds against (nav2_msgs from navigation2's main branch, pinned in ros2_data_collection.repos -- its NavigateThroughPoses::Result carries WaypointStatus[] waypoint_statuses) but this Measurement does not surface it yet: an earlier implementation reported it, and was trimmed when the repository tracked a distro branch whose NavigateThroughPoses::Result had no such field (fatal error: nav2_msgs/msg/waypoint_status.hpp: No such file or directory). Surfacing it again against the current pin is a follow-up, not something this Measurement does today.

A mission still running when collection stops simply never gets a matching mission_end Record: nothing downstream can average an interval that was never closed as a zero, because there is no mission_end Record to average. sequence is a monotonically increasing counter across every Record this Measurement instance emits (start and end alike, not per mission_id), so a dropped Record shows up as a gap rather than silently corrupting a duration or a mission-success-rate denominator.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "MissionNav2ThroughPoses",
  "properties": {
    "event": { "type": "string", "enum": ["mission_start", "mission_end"] },
    "mission_id": { "type": "string" },
    "mission_type": { "type": "string" },
    "sequence": { "type": "integer", "minimum": 1 },
    "outcome": { "type": "string", "enum": ["succeeded", "failed", "cancelled", "aborted"] },
    "reason": { "type": "string" },
    "error_code": { "type": "integer", "minimum": 0 },
    "duration_sec": { "type": "number", "minimum": 0 },
    "recoveries": { "type": "integer", "minimum": 0 }
  },
  "required": ["event", "mission_id", "sequence"],
  "type": "object"
}

Configuration

...
mission_nav2_through_poses:
  plugin: "dc_measurements/MissionNav2ThroughPoses"
  topic_output: "/dc/measurement/mission_nav2_through_poses"
  group_key: "mission"
  action_name: "navigate_through_poses"

Example output

Start:

{
  "event": "mission_start",
  "mission_id": "3f2504e04f8911d39a0c0305e82c3301",
  "mission_type": "navigate_through_poses",
  "sequence": 7
}

and end (succeeded):

{
  "event": "mission_end",
  "mission_id": "3f2504e04f8911d39a0c0305e82c3301",
  "mission_type": "navigate_through_poses",
  "sequence": 8,
  "outcome": "succeeded",
  "duration_sec": 96.4,
  "recoveries": 1
}

and end (aborted):

{
  "event": "mission_end",
  "mission_id": "9f86d081884c7d659a2feaa0c55ad015",
  "mission_type": "navigate_through_poses",
  "sequence": 10,
  "outcome": "aborted",
  "reason": "tf timeout",
  "error_code": 9102,
  "duration_sec": 12.1
}

Mission (Open-RMF)

Description

The Open-RMF adapter of the Mission Measurement for Open-RMF's TaskState (the document rmf-web's API server maintains per task). Emits a mission_start Record when a task's status first leaves queued/standby into an active state, and a mission_end Record once it reaches a terminal status, in the same mission_start/mission_end Record schema #387 defines -- mission_id sourced from Open-RMF's own booking.id, mission_type from category (Open-RMF tasks have a real category/type, unlike nav2's bare goal).

Unlike the nav2 adapters (mission_nav2/mission_nav2_follow_waypoints/ mission_nav2_through_poses), Open-RMF exposes no ROS action or topic for task lifecycle. This Measurement instead consumes TaskState as a stream of JSON documents over a plain ws:// connection, using the reusable dc_common::WebSocketJsonClient. The websocket protocol rmf-web's API server actually speaks on its own endpoints (Socket.IO framing) is out of scope -- websocket_url must point at an endpoint that re-emits the task-state feed as one TaskState JSON object per plain text frame; bridging the real rmf-web wire protocol is a separate piece of infrastructure this Measurement assumes already exists, not something it does itself.

This Measurement is a passive observer. It never dispatches, cancels, or otherwise mutates any Open-RMF task -- it only watches the state stream.

Status mapping

Open-RMF's task model is richer than a flat start/end pair: TaskState.status is a continuously-updated 12-value enum (uninitialized, blocked, error, failed, queued, standby, underway, delayed, skipped, canceled, killed, completed). This Measurement still boils it down to two Records per mission:

• a mission_start Record, the first time a booking.id is observed leaving queued/standby into an active state -- underway, delayed, or (once already active) blocked/error.
• a mission_end Record, once that booking.id reaches a terminal status: completed, failed, canceled, killed, or skipped.

outcome on the end Record is one of succeeded, failed, cancelled, aborted: completed→succeeded, failed→failed, canceled→cancelled, killed→aborted.

Two cases the acceptance criteria asked to be resolved explicitly, not silently defaulted:

• skipped is treated as task-terminal, and maps to outcome cancelled. It is defined in the same status enum task_state.json uses for the task's own top-level status field, the same enum that also carries in-progress phases, so a task can legitimately end its life with status: "skipped". Open-RMF's skipped means the task's work was bypassed rather than performed, which is closer to DC's cancelled (the work simply didn't complete) than to succeeded (the work was done) or a failure outcome. DC's outcome contract (#305/#387) has exactly four values; this Measurement maps into those four rather than adding a fifth skipped outcome.
• blocked and error are treated as transient. Both describe a task Open-RMF is still actively trying to resolve or recover (a blocked path, a recoverable fault) -- they are absent from task_state.json's terminal set, so its task manager never settles on either as the end of the task's life. A task observed as blocked/error stays open; only a later terminal status closes it, and it may still resolve back to underway.

reason is populated on a best-effort basis, carried verbatim from whichever part of TaskState Open-RMF actually populated for that outcome -- dispatch.errors (or the top-level detail) for failed, cancellation.labels for cancelled, killed.labels for aborted. Unlike nav2's always-present error_msg, Open-RMF does not guarantee one of these for every outcome, so reason (and error_code, only ever populated for failed) may be absent even on a mission_end Record.

duration_sec prefers Open-RMF's own unix_millis_start_time/unix_millis_finish_time when the source provided both; it falls back to this Measurement's own locally observed start/end timestamps otherwise.

What this Measurement does not represent

• No phase-level detail. TaskState.phases/active/completed/pending (which step of a multi-phase task is running) is not carried into the Record. Only the task's own top-level status drives mission_start/mission_end.
• No interruptions. TaskState.interruptions (temporary holds placed on a task, distinct from cancellation) is not represented at all -- an interrupted-then-resumed task simply keeps running from this Measurement's point of view.
• No dispatch/assignment detail. Which fleet or robot a task was assigned to (TaskState.assigned_to/dispatch.assignment) is not carried into the Record.

A task still active when the websocket connection drops or collection stops simply never gets a matching mission_end Record: nothing downstream can average an interval that was never closed, because there is no mission_end Record to average -- matching #387's open-interval handling. Symmetrically, a task whose first-ever observed sample is already active (started before this Measurement connected) or already terminal (finished before this Measurement connected) is not tracked at all: no mission_start Record (the active-transition boundary was never observed) and consequently no mission_end Record either.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "MissionOpenRmf",
  "properties": {
    "event": { "type": "string", "enum": ["mission_start", "mission_end"] },
    "mission_id": { "type": "string" },
    "mission_type": { "type": "string" },
    "sequence": { "type": "integer", "minimum": 1 },
    "outcome": { "type": "string", "enum": ["succeeded", "failed", "cancelled", "aborted"] },
    "reason": { "type": "string" },
    "error_code": { "type": "integer", "minimum": 0 },
    "duration_sec": { "type": "number", "minimum": 0 }
  },
  "required": ["event", "mission_id", "sequence"],
  "type": "object"
}

Configuration

...
mission_open_rmf:
  plugin: "dc_measurements/MissionOpenRmf"
  topic_output: "/dc/measurement/mission_open_rmf"
  group_key: "mission"
  websocket_url: "ws://rmf-task-state-bridge:8080/task_states"

Example output

Start:

{
  "event": "mission_start",
  "mission_id": "delivery.dispenser_1.dispatch-14",
  "mission_type": "delivery",
  "sequence": 3
}

end (succeeded):

{
  "event": "mission_end",
  "mission_id": "delivery.dispenser_1.dispatch-14",
  "mission_type": "delivery",
  "sequence": 4,
  "outcome": "succeeded",
  "duration_sec": 214.7
}

end (cancelled, cancellation labels present):

{
  "event": "mission_end",
  "mission_id": "patrol.loop_a.dispatch-22",
  "mission_type": "patrol",
  "sequence": 9,
  "outcome": "cancelled",
  "reason": "operator; dashboard",
  "duration_sec": 42.1
}

end (failed, dispatch error present):

{
  "event": "mission_end",
  "mission_id": "delivery.dispenser_2.dispatch-31",
  "mission_type": "delivery",
  "sequence": 15,
  "outcome": "failed",
  "reason": "dispenser_unavailable: dispenser_1 did not respond",
  "error_code": 12,
  "duration_sec": 8.9
}

Network

Description

Collects ping value, whether or not the PC is online and interfaces available. ping() opens an unprivileged ICMP "ping" socket (socket(AF_INET, SOCK_DGRAM, IPPROTO_ICMP)) — no root or CAP_NET_RAW needed, only a permissive net.ipv4.ping_group_range (the Linux default is permissive for the root group, which is what containers and most robot processes run as). This works unmodified in a plain rootless Podman/Docker container, with no --cap-add or --privileged needed.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Network",
    "description": "Network accessibility and information",
    "properties": {
        "ping": {
            "description": "Time to ping the host in ms",
            "type": "integer",
            "minimum": -1
        },
        "online": {
            "description": "If the pc is online",
            "type": "boolean"
        },
        "interfaces": {
            "description": "List of network interfaces",
            "type": "array",
            "items": {
                "type": "string"
            }
        }
    },
    "type": "object"
}

Configuration

...
network:
  plugin: "dc_measurements/Network"
  topic_output: "/dc/measurement/network"
  ping_address: 192.168.0.1
  ping_timeout: 500

Example output

Captured from a plain rootless Podman container, no special flags, ping_address pointed at localhost:

{
  "flattened": false,
  "interfaces": ["lo", "tunl0", "enp0s31f6"],
  "name": "network",
  "nested": false,
  "online": true,
  "ping": 0,
  "run_id": "175"
}

The same container against a real external host (ping_address: 8.8.8.8):

{
  "flattened": false,
  "interfaces": ["lo", "tunl0", "enp0s31f6"],
  "name": "network",
  "nested": false,
  "online": true,
  "ping": 6,
  "run_id": "176"
}

And against an unreachable one (ping_address: 192.0.2.1, the timeout dropped to 300ms for a fast test):

{
  "flattened": false,
  "interfaces": ["lo", "tunl0", "enp0s31f6"],
  "name": "network",
  "nested": false,
  "online": false,
  "ping": -1,
  "run_id": "177"
}

OS

Description

Collects the Operating System information: cpus, operating system name and kernel information

Parameters

This Measurement has no parameters beyond the common Plugin parameters.

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "OS",
  "description": "OS, kernel and CPUs information",
  "properties": {
    "os": {
      "description": "Host distribution name",
      "type": "string"
    },
    "kernel": {
      "description": "Kernel version",
      "type": "string"
    },
    "cpu": {
      "description": "Number of CPUs",
      "type": "integer",
      "minimum": 0
    },
    "memory": {
      "description": "System memory",
      "type": "number",
      "minimum": 0
    }
  },
  "type": "object"
}

Configuration

...
os:
  plugin: "dc_measurements/OS"
  topic_output: "/dc/measurement/os"

Example output

{
  "cpus": 8,
  "flattened": false,
  "kernel": "6.12.101+deb13-amd64",
  "memory": 23.23,
  "name": "os",
  "nested": false,
  "os": "Ubuntu 24.04.4 LTS",
  "run_id": "169"
}

Permissions

Description

Collect UID, GID, if a file or directory exists and its permissions (in rwx or integer format).

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Permissions",
    "description": "Permissions of a file/directory",
    "properties": {
        "uid": {
            "description": "File/directory User IDentifier",
            "type": "integer"
        },
        "gid": {
            "description": "File/directory Group IDentifier",
            "type": "integer"
        },
        "exists": {
            "description": "File/directory exists",
            "type": "boolean"
        },
        "permissions": {
            "description": "Permissions as rwx or integer",
            "type": "string"
        }
    },
    "type": "object"
}

Configuration

...
permission_home_dc:
  plugin: "dc_measurements/Permissions"
  topic_output: "/dc/measurement/permissions_home_dc"
  path: "$HOME/dc"
  format: "rwx"

Example output

{
  "exists": true,
  "flattened": false,
  "gid": 0,
  "group": "root",
  "name": "permission_home_dc",
  "nested": false,
  "permissions": "rwx------",
  "run_id": "169",
  "uid": 0,
  "user": "root"
}

Position

Description

Collect x, y and yaw of the robot.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Position",
    "description": "Position and orientation of the robot",
    "properties": {
        "x": {
            "description": "X position of the robot",
            "type": "number"
        },
        "y": {
            "description": "Y position of the robot",
            "type": "number"
        },
        "yaw": {
            "description": "Yaw angle of the robot",
            "type": "number"
        }
    },
    "type": "object"
}

Configuration

...
position:
  plugin: "dc_measurements/Position"
  topic_output: "/dc/measurement/position"

Example output

{
  "flattened": false,
  "name": "position",
  "nested": false,
  "run_id": "169",
  "x": 0.0,
  "y": 0.0,
  "yaw": 0.0
}

Random

Description

Emits a randomly generated value on every polling interval. Useful for exercising the pipeline and Destinations without any robot infrastructure — the simplest possible Measurement plugin, and a deterministic, infrastructure-free Record source for load and backpressure testing.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Random",
    "description": "Randomly generated value",
    "properties": {
        "value": {
            "description": "Randomly generated value within the configured range",
            "type": "number"
        }
    },
    "type": "object"
}

Configuration

...
random:
  plugin: "dc_measurements/Random"
  topic_output: "/dc/measurement/random"
  type: "double"
  min: 0.0
  max: 1.0
  seed: 42

Example output

{
  "flattened": false,
  "name": "random",
  "nested": false,
  "run_id": "169",
  "value": 0.8265497558024156
}

ROS2 control status

Description

Reports when a ros2_control controller or hardware component crosses into or out of the active lifecycle state -- the state in which it is actually commanding or reading hardware. Rather than polling list_controllers/list_hardware_components, Ros2ControlStatus subscribes to the controller manager's own ~/activity topic (controller_manager_msgs/ControllerManagerActivity), republished with transient-local QoS on every controller/hardware-component lifecycle change, and feeds each component's stream of (state, timestamp) samples to its own dc_common::StateTransitionDetector -- one detector per component, mirroring Fault's per-component detector map, so components activate and deactivate independently.

A Record is produced only for the boundary Intervention already reports in the same shape: a transition into active starts one ("event": "start"), and a transition out of active ends one ("event": "end"). A transition between two non-active states (e.g. unconfigured to inactive during startup) crosses no boundary and produces nothing.

Every Record carries the component type (controller or hardware_component), its name, the state it left (from_state) and entered (to_state), and how long the previous state had been held (previous_state_duration_s) -- on a start Record, how long the component was out of active before this; on an end Record, how long it had just been active. open is true on a start Record and false on an end Record, so a component still active when collection stops simply never gets a matching end Record and cannot be averaged as a zero. Every Record carries a seq that increments by one across every controller and hardware component, so a dropped Record is a detectable gap.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Ros2ControlStatus",
  "properties": {
    "seq": { "type": "integer", "minimum": 1 },
    "component_type": { "type": "string", "enum": ["controller", "hardware_component"] },
    "component": { "type": "string" },
    "event": { "type": "string", "enum": ["start", "end"] },
    "from_state": { "type": "string", "enum": ["unconfigured", "inactive", "active", "finalized", "configuring", "cleaningup", "shuttingdown", "activating", "deactivating", "errorprocessing", "unknown"] },
    "to_state": { "type": "string", "enum": ["unconfigured", "inactive", "active", "finalized", "configuring", "cleaningup", "shuttingdown", "activating", "deactivating", "errorprocessing", "unknown"] },
    "previous_state_duration_s": { "type": "number", "minimum": 0 },
    "open": { "type": "boolean" }
  },
  "required": ["seq", "component_type", "component", "event", "from_state", "to_state", "previous_state_duration_s", "open"],
  "type": "object"
}

Configuration

...
ros2_control_status:
  plugin: "dc_measurements/Ros2ControlStatus"
  topic_output: "/dc/measurement/ros2_control_status"
  group_key: "ros2_control_status"
  topic: "/controller_manager/activity"

Example output

A start:

{
  "seq": 1,
  "component_type": "controller",
  "component": "diff_drive_controller",
  "event": "start",
  "from_state": "inactive",
  "to_state": "active",
  "previous_state_duration_s": 4.2,
  "open": true
}

and its end:

{
  "seq": 2,
  "component_type": "controller",
  "component": "diff_drive_controller",
  "event": "end",
  "from_state": "active",
  "to_state": "inactive",
  "previous_state_duration_s": 612.9,
  "open": false
}

Serial interface

Description

Reads line-delimited data off a configurable serial port (baud rate configurable) and parses each line into named fields, publishing them as a Record — for custom robot sensors/boards that talk over a UART/USB-serial link and never reach a ROS topic on their own.

The port is opened lazily, on the first poll after activation, and never on onConfigure() — an unplugged or not-yet-connected device does not fail activation. Disconnection is detected two ways: a hard read error (e.g. EIO on some USB-serial removal paths) closes the port immediately; a hangup with no error (the more common case on a raw-mode, non-blocking read, where read() returning 0 alone can't be told apart from "no data yet") is caught via poll()'s POLLHUP/POLLERR. Either way the Measurement logs a warning, closes the file descriptor, and keeps polling; the next poll after the device reappears reopens and resumes normally, with no operator action needed and no busy-looping in between (opening only happens once per polling_interval tick).

Only line-delimited framing (\n, with an optional trailing \r stripped) is implemented today; framing is still a configuration knob for future framing modes. Only the most recently completed line in a given poll is parsed — if several lines arrive within one polling_interval, earlier ones are dropped, the same lossy-between-polls behavior other subscription/poll-driven Measurements (e.g. cmd_vel, diagnostics) already have.

Two parsing modes are supported, both producing the same fields object shape:

• delimiter: splits the line on delimiter and assigns tokens to fields in order.
• regex: matches the line against regex and assigns capture groups to fields in order. std::regex/ECMAScript has no native named-capture-group syntax, so "named groups" here means pairing each positional capture group with a name from fields, in capture order.

If the token/capture count doesn't match the configured fields count, a warning is logged and whatever fields do line up are still published — a malformed line degrades rather than drops.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "SerialInterface",
  "description": "A line parsed off a configured serial port",
  "properties": {
    "raw": {
      "description": "The raw line read from the serial port, with any trailing CR/LF stripped",
      "type": "string"
    },
    "fields": {
      "description": "Named fields extracted from 'raw' per the configured parsing (delimiter split or regex capture groups)",
      "type": "object"
    }
  },
  "required": ["raw", "fields"],
  "type": "object"
}

Configuration

Delimiter split, e.g. a board emitting 23.5,60\n (temperature, humidity):

...
serial_sensor:
  plugin: "dc_measurements/SerialInterface"
  topic_output: "/dc/measurement/serial_sensor"
  port: "/dev/ttyUSB0"
  baud_rate: 9600
  parsing_type: "delimiter"
  delimiter: ","
  fields: ["temperature", "humidity"]

Regex capture groups, e.g. a board emitting T:23.5 H:60\n:

...
serial_sensor:
  plugin: "dc_measurements/SerialInterface"
  topic_output: "/dc/measurement/serial_sensor"
  port: "/dev/ttyUSB0"
  baud_rate: 9600
  parsing_type: "regex"
  regex: "^T:(\\d+\\.\\d+) H:(\\d+)$"
  fields: ["temperature", "humidity"]

Example output

Derived from the delimiter example above (23.5,60 split on , into fields: ["temperature", "humidity"]):

{
  "raw": "23.5,60",
  "fields": {
    "temperature": "23.5",
    "humidity": "60"
  }
}

Testing without hardware

The gtest suite (test/test_measurement_serial_interface.cpp) verifies this Measurement against a virtual serial pair created with socat, no hardware required:

socat -d -d pty,raw,echo=0,link=/tmp/dc_serial_dev pty,raw,echo=0,link=/tmp/dc_serial_peer

The Measurement is pointed at /tmp/dc_serial_dev; the test writes fixture lines to /tmp/dc_serial_peer. Killing and restarting socat against the same link= paths simulates an unplug/replug cycle and exercises the reconnect path.

slam_toolbox quality

Description

Localization quality from two of slam_toolbox's native ROS 2 topics: /pose (geometry_msgs/PoseWithCovarianceStamped), polled on the same interval as every other Measurement, for a sample Record carrying the pose and its covariance; and /slam_toolbox/loop_closure_event (slam_toolbox/LoopClosureEvent on slam_toolbox's development branch — see the note below on why this Measurement doesn't depend on that type directly) for a single-shot loop_closure Record per occurrence.

LoopClosureEvent carries nothing but its own timestamp — slam_toolbox doesn't say which nodes closed the loop or by how much, only that one happened. The Record reflects that: loop_closure has no fields beyond event. The KPI value is entirely in the timing of these Records against each other and against the samples: loop closures per hour is a rate over how often loop_closure Records land, and time since the last loop closure — how long the map has gone without a correction — is a localization-drift risk indicator on its own, the way a growing gap between battery samples reads as data loss rather than a battery at 0 %.

Every Record carries an event field naming which of the two it is:

The covariance terms are the diagonal of PoseWithCovariance's 6x6 matrix at the indices for x, y and yaw — the same terms AMCL/robot_localization dashboards already chart, and the ones that read as confidence on their own axis without needing the off-diagonal correlations.

Until /pose publishes at all, no sample Record is emitted: a gap means no localization data, not a robot at the origin. A loop closure is queued the moment it's reported and leaves on the next poll, one Record per poll, so it travels the same path as every other Record (Conditions, incident buffering, Group) — the same convention Battery's charging-session boundaries and Intervention's takeovers use.

Parameters

Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "SlamToolboxQuality",
  "description": "Localization quality from slam_toolbox: a pose sample with covariance on the polling interval, plus a Record on every loop closure",
  "properties": {
    "event": { "type": "string", "enum": ["sample", "loop_closure"] },
    "x": { "type": "number" },
    "y": { "type": "number" },
    "yaw": { "type": "number" },
    "covariance_x": { "type": "number", "minimum": 0 },
    "covariance_y": { "type": "number", "minimum": 0 },
    "covariance_yaw": { "type": "number", "minimum": 0 }
  },
  "required": ["event"],
  "type": "object"
}

The full file (plugins/measurements/json/slam_toolbox_quality.json) also requires x, y, yaw and all three covariance terms on a sample — loop_closure requires nothing beyond event.

Configuration

...
slam_quality:
  plugin: "dc_measurements/SlamToolboxQuality"
  topic_output: "/dc/measurement/slam_quality"
  polling_interval: 1000
  pose_topic: "/pose"
  loop_closure_topic: "/slam_toolbox/loop_closure_event"

Example output

One sample:

{
  "event": "sample",
  "x": 1.42,
  "y": -0.63,
  "yaw": 0.71,
  "covariance_x": 0.008,
  "covariance_y": 0.011,
  "covariance_yaw": 0.004
}

A loop closure:

{
  "event": "loop_closure"
}

KPIs

tools/infrastructure/sql/kpi_views.sql defines dc_kpi_loop_closure_rate() (loop closures per hour and seconds since the last one, over an arbitrary window) and dc_kpi_loop_closures_1h (the same count bucketed hourly for charting) over this Measurement's loop_closure Records, following the same convention as dc_kpi_intervention_rate(). The demo Grafana KPI dashboard charts loop closures per hour; the robot dashboard charts the covariance trend directly from sample Records.

Speed

Description

Collect robot speed using the Odom topic.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Speed",
    "description": "Computed, linear and angular speed of the robot",
    "properties": {
        "computed": {
            "description": "Computed speed in meter/s",
            "type": "number"
        },
        "linear": {
            "description": "Linear velocity as a vector",
            "type": "object",
            "items": {
                "$ref": "#/$defs/vector3"
            }
        },
        "angular": {
            "description": "Angular velocity as a vector",
            "type": "object",
            "items": {
                "$ref": "#/$defs/vector3"
            }
        }
    },
    "$defs": {
        "vector3": {
            "type": "object",
            "properties": {
                "x": {
                    "description": "X speed",
                    "type": "number"
                },
                "y": {
                    "description": "Y speed",
                    "type": "number"
                },
                "z": {
                    "description": "Z speed",
                    "type": "number"
                }
            }
        }
    },
    "type": "object"
}

Configuration

...
speed:
  plugin: "dc_measurements/Speed"
  topic_output: "/dc/measurement/speed"
  odom_topic: "/odom"

Example output

{
  "angular": {
    "x": 0.0,
    "y": 0.0,
    "z": 0.0
  },
  "computed": 0.5,
  "flattened": false,
  "linear": {
    "x": 0.5,
    "y": 0.0,
    "z": 0.0
  },
  "name": "speed",
  "nested": false,
  "run_id": "169"
}

Storage

Description

Collect storage information on a directory.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Storage",
    "description": "Storage information of a directory",
    "properties": {
        "free_percent": {
            "description": "Free space on the filesystem, in percent",
            "type": "number",
            "minimum": 0
        },
        "free": {
            "description": "Free space on the filesystem, in bytes",
            "type": "integer",
            "minimum": 0
        },
        "capacity": {
            "description": "Total size of the filesystem, in bytes",
            "type": "integer",
            "minimum": 0
        }
    },
    "type": "object"
}

Configuration

...
storage_home:
  plugin: "dc_measurements/Storage"
  topic_output: "/dc/measurement/storage_home"
  path: "$HOME"

Example output

{
  "capacity": 460041527296,
  "flattened": false,
  "free": 74388291584,
  "free_percent": 11.073844909667969,
  "name": "storage_home",
  "nested": false,
  "run_id": "169"
}

String stamped

Description

Collect generic data from a topic publishing a StringStamped message and republish it. It allows to fetch data from rclc, rclpy and your custom ROS 2 nodes that don't have use a plugin.

Parameters

Schema

Given that the data is customized here, there is no default schema.

Configuration

...
my_data:
  plugin: "dc_measurements/StringStamped"
  topic_output: "/dc/measurement/my_data"
  topic: "/hello_world"
  timer_based: true
  enable_validator: false # no schema exists for custom, per-topic data -- see Schema above

Example output

{
  "flattened": false,
  "name": "string_stamped",
  "nested": false,
  "run_id": "169",
  "temperature": 21.5
}

TCP Health

Description

Collects status of a TCP server.

Parameters

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "TCP Health",
    "description": "Status of a TCP server",
    "properties": {
        "host": {
            "description": "Server hostname",
            "type": "string"
        },
        "port": {
            "description": "Port number",
            "type": "integer",
            "minimum": 1,
            "maximum": 65536
        },
        "server_name": {
            "description": "Server alias, from the 'name' parameter",
            "type": "string"
        },
        "active": {
            "description": "Whether the TCP connection check succeeded",
            "type": "boolean"
        }
    },
    "type": "object"
}

Configuration

...
tcp_health:
  plugin: "dc_measurements/TCPHealth"
  topic_output: "/dc/measurement/rustfs_health"
  group_key: "rustfs_health"
  host: "127.0.0.1"
  port: 9000
  name: "rustfs_api"

Example output

{
  "active": true,
  "flattened": false,
  "host": "127.0.0.1",
  "name": "tcp_health",
  "nested": false,
  "port": 9000,
  "run_id": "169",
  "server_name": "test_service"
}

Thermal

Description

Reports temperatures (CPU, GPU, board, ...) from the kernel's thermal sysfs interface (/sys/class/thermal/thermal_zone*/), so overheating trends reach dashboards before they become failures. Zones are auto-discovered by default; each zone's type string (e.g. x86_pkg_temp, cpu-thermal) is used as the Record key, not its numeric index, since zone numbering is platform-specific and differs between x86_64 and ARM targets. A missing or unreadable /sys/class/thermal (e.g. a container without the host's thermal sysfs mounted) degrades gracefully: activation still succeeds, and if no zone can currently be read the Measurement publishes nothing that cycle (the same "silent, retry next poll" contract other hardware-optional Measurements like Serial interface use) instead of failing or publishing an empty Record.

Parameters

Schema

Zone type strings are the field names, so they can't be listed ahead of time — the schema constrains their shape instead: at least one entry (the Measurement publishes nothing rather than an empty Record), a non-empty type string as key, and a temperature in degrees Celsius above absolute zero as value.

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Thermal",
    "description": "One entry per thermal zone read this cycle, keyed by the zone's type string",
    "propertyNames": {
        "description": "The zone's type string, e.g. x86_pkg_temp or cpu-thermal",
        "minLength": 1
    },
    "additionalProperties": {
        "description": "Temperature of the zone in degrees Celsius",
        "type": "number",
        "minimum": -273.15
    },
    "minProperties": 1,
    "type": "object"
}

Configuration

...
thermal:
  plugin: "dc_measurements/Thermal"
  topic_output: "/dc/measurement/thermal"

Example output

On a host exposing a CPU package zone and a GPU zone:

{
  "x86_pkg_temp": 52.0,
  "gpu-thermal": 61.5
}

Uptime

Description

Time since when the robot PC has been on.

Parameters

This Measurement has no parameters beyond the common Plugin parameters.

Schema

{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "Uptime",
    "description": "Time the system has been up",
    "properties": {
        "time": {
            "description": "Time the system has been up",
            "type": "integer",
            "minimum": 0
        }
    },
    "type": "object"
}

Configuration

...
uptime:
  plugin: "dc_measurements/Uptime"
  topic_output: "/dc/measurement/uptime"

Example output

{
  "flattened": false,
  "name": "uptime",
  "nested": false,
  "run_id": "169",
  "time": 1783191
}

Overview

Description

A condition enables or disables one or multiple measurements to be published and thus collected. We could for example enable collecting camera images only when a robot is stopped.

Each condition is enabled or disabled through a pluginlib plugin. It has these configuration parameters.

Wiring a Condition to a Measurement

Configuring a Condition is a two-step wire-up, both under measurement_server:

1. Declare it in condition_plugins and give it its own named block, exactly like a Measurement — plugin names the pluginlib class, and any other keys are that Condition's own parameters.
2. Reference that name from a Measurement's if_all_conditions, if_any_conditions, if_none_conditions, or gate_condition (see Measurements). The Measurement is what names the Condition — nothing on the Condition side says which Measurement it gates, so one Condition can gate several Measurements at once.

The camera-only-when-stopped example above, taken from a real demo (dc_demos/params/qrcodes_stdout.yaml):

measurement_server:
  ros__parameters:
    condition_plugins: ["moving"]
    moving:
      plugin: "dc_conditions/Moving"
    right_camera:
      plugin: "dc_measurements/Camera"
      if_none_conditions: ["moving"] # collect only while NOT moving
      topic_output: "/dc/measurement/right_camera"
      cam_topic: "/right_intel_realsense_r200_depth/image_raw"

Available plugins:

Compare

Description

Compare the value of a JSON key against a configured operand and activate while the comparison holds. One plugin covers the whole family: comparison picks the operator and the type you give value picks the operand type (the two axes thirteen former per-operator plugins each hard-coded).

Parameters

The JSON comparison is type-strict: a Record field written as 5 is an integer and never matches a float operand (and 5.0 never matches an int one). gt/ge/lt/le need an int or float operand; ne inverts eq for every operand type.

Configuration

Only forward a TCP Health Record when the check actually failed — suppressing the steady stream of active: true polls and keeping only outage alerts:

measurement_server:
  ros__parameters:
    condition_plugins: ["endpoint_down"]
    endpoint_down:
      plugin: "dc_conditions/Compare"
      key: "active"
      value: false
      comparison: "eq"
    rustfs_health:
      plugin: "dc_measurements/TCPHealth"
      if_all_conditions: ["endpoint_down"]
      topic_output: "/dc/measurement/rustfs_health"
      host: "127.0.0.1"
      port: 9000
      name: "rustfs_api"

A dead-band on Distance traveled — filters out both "parked" noise and implausibly large jumps (dc_demos/params/tb3_simulation_pgsql_minio.yaml):

measurement_server:
  ros__parameters:
    condition_plugins: ["min_distance_traveled", "max_distance_traveled"]
    min_distance_traveled:
      plugin: "dc_conditions/Compare"
      key: "distance_traveled"
      value: 0.01
      comparison: "ge"
    max_distance_traveled:
      plugin: "dc_conditions/Compare"
      key: "distance_traveled"
      value: 2.0
      comparison: "le"
    distance_traveled:
      plugin: "dc_measurements/DistanceTraveled"
      if_all_conditions: ["min_distance_traveled", "max_distance_traveled"]
      topic_output: "/dc/measurement/distance_traveled"

Only publish Fast DDS statistics once its discovered hosts list no longer matches the expected single-host fingerprint — flags an unexpected extra participant joining the DDS graph:

measurement_server:
  ros__parameters:
    condition_plugins: ["known_hosts"]
    known_hosts:
      plugin: "dc_conditions/Compare"
      key: "hosts"
      value: ["d:14058711922191368192"]
      comparison: "eq"
      order_matters: false
    fastdds_stats:
      plugin: "dc_measurements/FastddsStats"
      if_none_conditions: ["known_hosts"]
      topic_output: "/dc/measurement/fastdds_stats"

Only collect a Battery Record while it is actively charging (match uses a full regex match, and exists tests only that key is present, taking no value):

measurement_server:
  ros__parameters:
    condition_plugins: ["actively_charging", "inspected_exists"]
    actively_charging:
      plugin: "dc_conditions/Compare"
      key: "power_supply_status"
      comparison: "match"
      regex: "charging|full"
    inspected_exists:
      plugin: "dc_conditions/Compare"
      key: "inspected"
      comparison: "exists"
    battery:
      plugin: "dc_measurements/Battery"
      if_all_conditions: ["actively_charging", "inspected_exists"]
      topic_output: "/dc/measurement/battery"

Migration from the former per-operator plugins

ne (value differs) and scalar str equality are new with Compare.

Moving

Description

Use a hysteresis on robot position to know whether the robot is moving.

Parameters

Configuration

Only record camera images while the robot is stationary (dc_demos/params/qrcodes_stdout.yaml):

measurement_server:
  ros__parameters:
    condition_plugins: ["moving"]
    moving:
      plugin: "dc_conditions/Moving"
    right_camera:
      plugin: "dc_measurements/Camera"
      if_none_conditions: ["moving"]
      topic_output: "/dc/measurement/right_camera"
      cam_topic: "/right_intel_realsense_r200_depth/image_raw"

Same as previous

Description

Compare JSON key value to the value passed in parameter and returns true if match.

Parameters

Configuration

Skip republishing Diagnostics when nothing has changed since the last collection — /diagnostics is typically high-volume and mostly unchanging:

measurement_server:
  ros__parameters:
    condition_plugins: ["diagnostics_unchanged"]
    diagnostics_unchanged:
      plugin: "dc_conditions/SameAsPrevious"
      keys: []
      exclude: []
    diagnostics:
      plugin: "dc_measurements/Diagnostics"
      if_none_conditions: ["diagnostics_unchanged"]
      topic_output: "/dc/measurement/diagnostics"
      level_threshold: "WARN"
      names: ["motor_driver", "battery"]

Triggers

Description

A Trigger fires a one-shot signal — a dc_interfaces/msg/FlushEvent — when a composition of Conditions goes from false to true. Measurements listening for that event release the window of recent data they have been holding back, so the seconds leading up to an event are collected too, alongside the seconds after it.

The motivating case is incident review: when an autonomous robot emergency-brakes, what matters is what happened before the brake. A Condition cannot express that — it gates collection going forward, and by the time it turns true the run-up is gone. A Trigger can, because the Measurement was already buffering.

Trigger vs Condition

Both are described with the same if_all/if_any/if_none Condition lists, and a Trigger is built out of Condition plugins — but they answer different questions.

They compose: a Trigger is made of Conditions, and a buffering Measurement can still be gated by its own Conditions. Nothing about a Trigger changes what a Condition means.

The flush broadcast

The trigger_broadcast_node loads the Condition plugins named in condition_plugins (the same Condition plugins Measurements use) plus one Trigger plugin, polls the Trigger on a timer, and publishes a FlushEvent on a configurable topic (/dc/flush by default) each time it fires:

                    ┌──────────────────────┐
   /odom, … ───────▶│  Condition plugins   │
                    └──────────┬───────────┘
                               │ if_all / if_any / if_none
                    ┌──────────▼───────────┐
                    │  Trigger (EdgeTrigger)│  false → true?
                    └──────────┬───────────┘
                               │ FlushEvent { incident_id }
                    ┌──────────▼───────────┐
                    │      /dc/flush       │
                    └───┬──────────────┬───┘
                        │              │
              ┌─────────▼────┐  ┌──────▼───────┐
              │ Measurement A│  │ Measurement B│   release their buffered windows,
              │  (buffering) │  │  (buffering) │   both tagged with the same incident_id
              └──────────────┘  └──────────────┘

The node mints a fresh incident_id (a UUID) for every firing; nothing else generates one, so every subscriber of one event shares one ID. Several Measurements can point at the same flush_topic, which is the whole reason the signal is broadcast on a topic rather than wired per Measurement: adding a sensor stream to an existing incident-capture setup needs no change to the Trigger.

ros2 run dc_triggers trigger_broadcast_node --ros-args --params-file <your_params.yaml>

FlushEvent

dc_interfaces/msg/FlushEvent is the entire contract between the Trigger side and the Measurement side:

A Measurement adopts the incident_id it receives; it never generates its own. An event that arrives while a Measurement is already flushing, in post-roll, or in cooldown is ignored.

incident_id

incident_id is a typed field of the dc_interfaces/msg/StringStamped Record envelope, beside group_key — not a key nested inside the measurement's own data. Every Record and File released by one flush cycle carries the same value, so "everything from this one event" is a single query rather than a timestamp range reconstructed by hand:

SELECT * FROM dc_records WHERE incident_id = '3f2b1c7e-…' ORDER BY date;

• A postgres Destination writes it to its own incident_id column — the column must exist in the table beforehand; see Destinations.
• A Record collected outside an incident carries an empty incident_id, leaving the column NULL.
• A Group carries a member's envelope incident_id onto its own output envelope, so grouping does not bury it.
• Every Destination receives it as a top-level key of the JSON the Bridge ships: the Bridge lifts the envelope field into the payload before handing a Record to its Destinations.

Available plugins

Node parameters

Plugin parameters

Every Trigger plugin loaded by this node is namespaced under trigger and shares these parameters:

Measurement parameters

The other half of the feature lives on each Measurement: what it buffers, and what it does once a FlushEvent releases it. All five are optional, and a Measurement that leaves buffer_duration_sec at 0 behaves exactly as it always has — this feature is entirely opt-in. They are documented in full, with the state machine and how Files are staged, in Measurements.

An armed Measurement moves through four states per incident — Buffering (accumulate, publish nothing) → Flushing (release the window, oldest first, rate-limited) → PostRoll (publish live, same incident_id) → Cooldown (ignore further events) → back to Buffering, with no manual re-arm.

Example

An emergency-brake capture: the robot buffers the last 10 seconds of uptime data, and the moment it starts moving the buffer is released and collection continues live for 5 more seconds. Swap the moving Condition for whatever expresses your event.

trigger_broadcast_node's parameters:

trigger_broadcast_node:
  ros__parameters:
    condition_plugins: ["moving"]

    moving:
      plugin: "dc_conditions/Moving"
      odom_topic: "/odom"

    trigger:
      plugin: "dc_triggers/EdgeTrigger"
      if_all_conditions: ["moving"]
      topic: "/dc/flush"
      polling_interval: 100

The Measurements listening for it:

measurement_server:
  ros__parameters:
    measurement_plugins: ["uptime", "memory"]

    # Armed: publishes nothing until a FlushEvent arrives on /dc/flush, then releases the
    # last 10 seconds of collection tagged with that event's incident_id.
    uptime:
      plugin: "dc_measurements/Uptime"
      polling_interval: 500
      group_key: "uptime"
      buffer_duration_sec: 10.0
      post_roll_duration_sec: 5.0
      cooldown_sec: 30.0
      max_flush_rate_hz: 20.0
      flush_topic: "/dc/flush"

    # Not armed: collects live as usual, and its Records leave incident_id NULL.
    memory:
      plugin: "dc_measurements/Memory"
      polling_interval: 1000
      group_key: "memory"

A runnable end-to-end version of this — two Measurements, one armed, both landing in the same Postgres table, with the incident_id column asserted by query — is tools/e2e/scripts/run_incident.sh.

Edge trigger

Description

Composes the Condition plugins named in if_all_conditions/if_any_conditions/ if_none_conditions — the exact same rules Measurements already use — and fires exactly once on the false→true rising edge of the composed result. A sustained true fires only once; the Trigger stays quiet until the composed result falls back to false and rises again. No rule-evaluation logic of its own: composition is entirely delegated to the same ConditionSet evaluator dc_measurements::Measurement uses for live gating.

Parameters

See Plugin parameters — EdgeTrigger has no parameters of its own beyond the shared trigger.* ones every Trigger plugin reads.

Trigger vs Condition

EdgeTrigger fires once per false→true transition; a Condition stays true and keeps gating for as long as its predicate holds. See Trigger vs Condition.

Destinations

A Destination is an external system that receives Records or Files. In the DC 2.0 architecture (ADR-0003), the pluginlib destination-plugin layer is retired. The Bridge (dc_bridge) renders the external Vector Shipper's configuration from plain ROS parameters for a blessed set of receives: records Destination types — file, vector — and every other destination in Vector's sink catalog, including postgres, s3 and console, is available through the passthrough: raw Vector config snippets listed in the custom_config_files parameter, merged into the same rendered config (see Passthrough below).

type: s3 also stays blessed, but only for receives: files Destinations — object storage there is served entirely by the separate Uploader process reading these same ROS params (ADR-0005), never by a Vector sink, so there is no passthrough equivalent for it (see File uploads below). postgres/s3 (for receives: records)/console were blessed types through DC 2.0's early releases; their dedicated ROS-param templating was removed in #471/#472 once passthrough recipes for all three shipped (#470) — see Recipes below for the equivalent snippets.

Verified sink versions

console and file write to local stdout/disk — no external version to track. Any other passthrough sink is whatever's current in Vector's own sink catalog for the pinned 0.57.0 release above.

Bridge node parameters

Deployment modes: shipper.managed

shipper.managed picks who owns the Shipper's process lifecycle. The Bridge renders the same config and connects over the shipper ingest protocol the same way in both modes — only whether it also locates, spawns and supervises the Vector binary changes.

Use shipper.managed: true (the default) for:

• Single-robot / simulation / demos — one process tree on one machine, apt install or a workspace build, nothing else to run or orchestrate. This is what every demo under Demos and the dc_simulation warehouse simulation use.
• Local development and testing — fewer moving parts: no container runtime, no volumes to wire up, ros2 launch dc_bringup dc_bringup.launch.py is the whole story.

Nothing about this mode changes with shipper.managed added — it is, and remains, the default, and an existing deployment that never sets the parameter is unaffected.

Use shipper.managed: false for:

• Multi-container / orchestrator-managed deployments — Vector runs as its own container (the upstream image, not the vendored binary) under Compose, Podman Quadlet or Kubernetes, alongside the ROS/Bridge container on the same host. Set shipper.config_path to a path on a volume both containers mount, so the Bridge writes the config where the Shipper container reads it; the atomic write (below) is what makes that handoff safe even while the Shipper is already watching the file.
• Per-component operations — separate logs (no ROS/Vector log interleaving), independent restarts, and orchestrator-native resource limits/credentials for the Shipper, without changing anything about how the Bridge renders or validates its config.

This parameter is the Bridge-side building block for the larger split-deployment and fleet topologies tracked in #440. dc_uploader as its own process (ADR-0014) and the blessed vector Destination type for robot→edge forwarding have both since shipped; running dc_uploader as its own container, with its own volumes and credentials, remains #447's work, not shipped by shipper.managed alone.

Atomic config write

The rendered config is always written atomically — a full write to <shipper.config_path>.tmp, then rename() over the real path — in both modes. A reader polling the path (an unmanaged Shipper container watching a shared volume; vector validate in managed mode) can therefore never observe a partial write, regardless of how large the config is or how the two containers' write/read timing lines up.

Configuration contract

Every Destination is declared in the destinations list of the dc_bridge node's parameters, with a parameter block named after it (see dc_bringup/params/dc_params.yaml for a complete commented example):

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"   # Vector's disk-buffer directory
      # buffer_max_bytes: 268435488  # optional; Vector's disk-buffer minimum
    destinations: ["records_log", "to_aggregator", "rustfs"]
    records_log:
      type: file
      receives: records
      inputs: ["/dc/measurement/uptime"]   # ROS topics feeding this Destination
      path: "/var/log/dc/records-%Y-%m-%d.log"   # Vector template syntax allowed
    to_aggregator:
      type: vector
      receives: records
      inputs: ["/dc/measurement/uptime"]
      host: "edge-aggregator.local"
      port: 6000
    rustfs:
      type: s3               # blessed only for `receives: files` — see below
      receives: files
      bucket: "dc-records"
      endpoint: "http://127.0.0.1:9000"    # omit for AWS S3
      region: "us-east-1"
      access_key_id: "rustfsadmin"         # omit both to use ambient AWS credentials
      secret_access_key: "$DC_S3_SECRET"
      force_path_style: true               # path-style addressing for self-hosted stores
      key_prefix: "robot1/"

The vector type forwards to another Shipper over Vector's own native inter-instance protocol (type = "vector" sink → type = "vector" source) — the standard way to chain a robot's local Shipper to an edge aggregator's Shipper in a fleet deployment. host and port name the downstream Shipper; there is no default for either, since there's no sensible address to assume for another Shipper.

The s3 type — used above with receives: files — works with any S3-compatible store. For self-hosting, RustFS (Apache 2.0, S3-compatible, a drop-in MinIO replacement) is the recommended choice — MinIO's community edition was archived upstream in 2026 and no longer receives maintenance — but existing MinIO or Ceph RGW deployments work identically: set endpoint, explicit credentials, and (typically) force_path_style: true. See File uploads below for the full receives: files contract; batch_timeout_secs above is a receives: records passthrough recipe field only (see Recipes), not something the Uploader's own S3 client uses.

Common parameters for every blessed type: type, receives (records | files, see the File uploads section below), inputs (ROS topic names — required for receives: records; optional for receives: files, see below), time_key (default date) and time_format (default epoch_nanos, receives: records only) controlling the normalized timestamp field written into each Record before routing.

time_format

The Bridge forwards each Record's timestamp at full nanosecond resolution, taken from the ROS message header. time_format decides how that is written into the Record:

double is lossy by construction, not by implementation: a float64 has ~15–16 significant digits and current epoch seconds already consume 10 of them. It remains available for consumers that want a fractional-seconds column, but it cannot represent the resolution the pipeline delivers.

incident_id

A Measurement configured for incident capture tags every Record it releases with the incident_id of the FlushEvent that released it. That field is typed on the StringStamped Record envelope (#506), and the Bridge lifts it into a top-level key of the payload JSON before handing a Record to its Destinations — no per-Destination configuration is needed to expose it. A postgres sink (via the passthrough recipe below) can therefore write it straight into its own incident_id column with no extra mapping — "everything from this one event" is a plain WHERE incident_id = '…' query — since Vector's postgres sink has no column options of its own: a top-level event key lands in the same-named column and everything else is dropped. The column has to exist in the table before the first incident, exactly like every other column (tools/e2e/sql/init.sql and tools/infrastructure/docker/config/postgresql/init.sql both carry it):

ALTER TABLE dc ADD COLUMN incident_id text;

Records collected outside an incident carry an empty incident_id, and an empty envelope field is never lifted, leaving the column NULL.

Type-specific parameters:

postgres, s3 (for receives: records) and console are configured via the passthrough instead — see Recipes below.

Invalid parameters (unknown type, missing required field, half a credential pair, an out-of-range port…) are rejected with a clear error at Bridge startup — before Vector is ever started.

$VAR environment references are expanded by the Bridge in shipper.data_dir, custom_config_files, and the secret_access_key credential — and nowhere else. In particular file's path is handed to Vector verbatim, and Vector does not expand environment variables there: path: "$HOME/records.ndjson" silently writes to a literal $HOME directory beside the Bridge's working directory. Use an absolute path (Vector creates missing parent directories).

The dc.<tag> routing contract (public API)

The Bridge exposes one Shipper route per Tag. The Tag for a topic is its name with the leading / dropped and the remaining / turned into . (/dc/measurement/uptime → dc.measurement.uptime), and its route is named dc.<tag> — the leading dc. names the Bridge's route transform, the rest is the Tag verbatim (so /dc/measurement/uptime's Records are consumable as dc.dc.measurement.uptime). These names are stable public API: blessed sinks are wired to them internally, and custom snippets consume them the same way. A route exists for every topic listed in any configured Destination's inputs.

One route is not per-Tag: dc.dc.raw carries the whole dc.raw. Tag namespace when raw topic collection is enabled. Raw mode discovers topics — and so mints Tags — while the Shipper is already running, which no fixed set of exact-Tag branches can cover, so that branch matches on the Tag prefix instead. It is consumable from a snippet exactly like the others; each event still carries its own tag field.

Passthrough: custom_config_files

Any Vector sink type ships Records with zero DC code by consuming dc.<tag> routes from a raw Vector config snippet (TOML):

# ~/.dc/custom_sinks.toml — an un-blessed sink type (http), pure Vector config
[sinks.my_http]
type = "http"
inputs = ["dc.dc.measurement.uptime"]   # the public dc.<tag> route
uri = "http://127.0.0.1:8080/ingest"
encoding.codec = "json"

dc_bridge:
  ros__parameters:
    custom_config_files: ["$HOME/.dc/custom_sinks.toml"]

Snippets must not re-define component ids owned by the generated config (the dc_bridge_in source, the dc_bridge_normalize and dc transforms, or any configured Destination's name) or by another snippet. An invalid or colliding snippet is a loud Bridge startup error naming the offending file. The Bridge then folds every validated snippet into the same file it writes to shipper.config_path — one self-contained file regardless of deployment mode, since in unmanaged mode a snippet's own filesystem path is never wired into the separately-run Shipper container — and, in managed mode, runs vector validate over that merged file as a backstop before starting Vector. A snippet may define transforms as well as sinks — only defining a reserved id is rejected, consuming one is not.

A snippet cannot route a topic on its own. The Bridge derives both its ROS subscriptions and its dc.<tag> route branches from destinations, and never reads a snippet's inputs — so every topic a snippet consumes must also appear in some blessed Destination's inputs, and destinations must name at least one (a file Destination is the cheapest way to satisfy that). Two consequences follow from the passthrough being outside the rendered config: the snippet's sink gets Vector's default in-memory buffer, not the disk buffer dc_bridge gives every blessed sink, and it is the snippet's job to make re-delivery idempotent if the store cares — the Shipper is at-least-once (ADR-0002) either way.

The Elasticsearch tutorial is the worked example for all of this, end to end; the InfluxDB demo is the same mechanism against a simulated robot. MCAP recording (ADR-0009, issue #210) is the same passthrough consumed by a standalone process instead of a Vector-native sink — the shape to follow for any store Vector has no sink for at all.

Recipes: postgres, s3, console via passthrough

Per ADR-0003, postgres, s3 (for receives: records) and console are pure Vector-sink wrappers with no DC-specific logic layered on top — Vector's own vector validate already gives clear, field-level errors for these three sinks, so passthrough loses nothing on the validation front for this subset. This is now the only way to configure them: their dedicated ROS-param templating was removed in #471/#472. Each recipe below reproduces exactly what dc_bridge used to render for the equivalent formerly-blessed configuration — confirmed by running Vector 0.57.0 against a live PostgreSQL and RustFS instance, not just written by inspection.

As with any passthrough, at least one blessed Destination is still needed to create the dc.<tag> route the snippet consumes — file is the cheapest (see above). If you're migrating the console Destination itself, keep a file Destination (or another cheap blessed type) as the route anchor instead.

postgres — the former blessed form's host/port/user/password/database collapse into a single connection-string endpoint; table is unchanged:

# ~/.dc/postgres_sink.toml — passthrough equivalent of the blessed `pgsql` Destination
[sinks.pgsql]
type = "postgres"
inputs = ["dc.dc.measurement.uptime"]   # the public dc.<tag> route
endpoint = "postgres://dc:${DC_PG_PASSWORD}@127.0.0.1:5432/dc"  # user:password@host:port/database
table = "dc"

[sinks.pgsql.buffer]
type = "disk"
max_size = 268435488   # Vector's disk-buffer minimum; a passthrough sink gets none by default

If user or password contain characters reserved in a URI (:, @, /, %), percent-encode them yourself — the former blessed form did this automatically, but a passthrough endpoint is handed to Vector verbatim.

s3 — Vector's own sink id is aws_s3, not s3; credentials move under [sinks.<name>.auth] and batch_timeout_secs becomes [sinks.<name>.batch] timeout_secs:

# ~/.dc/s3_sink.toml — passthrough equivalent of the blessed `rustfs` Destination
[sinks.rustfs]
type = "aws_s3"                          # Vector's sink id — not "s3"
inputs = ["dc.dc.measurement.uptime"]
bucket = "dc-records"
endpoint = "http://127.0.0.1:9000"       # omit for AWS S3
region = "us-east-1"
key_prefix = "robot1/"
force_path_style = true                  # path-style addressing for self-hosted stores

[sinks.rustfs.auth]
access_key_id = "rustfsadmin"
secret_access_key = "${DC_S3_SECRET}"

[sinks.rustfs.batch]
timeout_secs = 60                        # object write interval; Vector default 300

[sinks.rustfs.buffer]
type = "disk"
max_size = 268435488

[sinks.rustfs.encoding]
codec = "json"

console — had no required fields either way, so this recipe mostly matters for consistency with the other two:

# ~/.dc/console_sink.toml — passthrough equivalent of the blessed `console` Destination
[sinks.debug_console]
type = "console"
inputs = ["dc.dc.measurement.uptime"]
target = "stdout"                        # or "stderr"

[sinks.debug_console.encoding]
codec = "json"

File uploads: receives: files (the Uploader, ADR-0005, ADR-0014)

A Destination with receives: files (only type: s3 qualifies) is served by the Uploader — dc_uploader, a separate process with its own Shipper connection (ADR-0014) — not by a Vector sink: Records arriving on its inputs topics are durably enqueued by the Bridge, then scanned by dc_uploader for the local_paths/remote_paths File references Measurements embed (camera, map, …), and each referenced File is uploaded (multipart + resumable for large Files) to every receives: files Destination whose name appears as a key in the Record's remote_paths — so the Destination name in the params file must match the remote key the Measurement is configured with. The YAML config shape below is unchanged by the split: dc_bringup.launch.py reads these same files.*/uploader.* parameters from the dc_bridge params block and translates them into dc_uploader's own DC_UPLOADER_* environment variables when it starts that process.

{
  "name": "map",
  "local_paths":  { "yaml": "/tmp/map.yaml", "pgm": "/tmp/map.pgm" },
  "remote_paths": { "rustfs": { "yaml": "robot/map.yaml", "pgm": "robot/map.pgm" } }
}

Per File the Uploader: uploads (multipart and resumable for large Files — an interrupted transfer picks up from the last completed part after a reconnect or Bridge restart, using progress checkpointed under <shipper.data_dir>/uploader/), verifies the object landed (head + size comparison), extracts metadata (content type, size, and duration via ffprobe for video/* Files), and emits a status Record under the dc.files Tag, routed like any Record to the Destination named by files.metadata_destination. Status rows preserve the Humble files_metrics shape — group_name, robot_name/robot_id (when present in the Record), local_path, remote_path (s3://bucket/key), storage_type, uploaded, on_filesystem, deleted, content_type, size, duration, updated_at, plus thumbnail_path when thumbnails are enabled — but the log is append-only: deletion appends a deleted: true row instead of updating one, and PostgreSQL is only ever written through the Shipper's parameterized sink (no SQL strings anywhere in DC).

Two guarantees consumers can rely on:

• Group completion markers: when one Record references several Files (map = pgm+yaml, camera batches), a kind: group_complete Record listing every File is emitted only after all of them are verified on all their Destinations — a consumer querying mid-upload sees per-File rows but no marker, and never has to guess whether a group fully arrived.
• Deletion only after verified upload: with files.delete_when_sent: true, a local File is removed only once it is verified on every Destination that receives it. Retries are idempotent — already-verified objects are not re-uploaded and status rows are never duplicated.

Durability: the disk-backed intent queue

Neither a Bridge nor a dc_uploader restart ever forgets a pending upload — the intent queue is a shared on-disk directory, not in-process state either owns exclusively. Before a files-Destination's Record is ever handed to the Uploader, it is written to disk as one intent — a JSON file ({version: 1, tag, timestamp, payload}) at <shipper.data_dir>/queue/upload/<monotonic_ts>-<seq>.json, via tmp-write-then-rename (crash-atomic; no fsync, matching Humble Fluent Bit's own storage.sync normal). The intent leaves the queue only when its Record has been fully processed — every File verified everywhere, or reported missing — never on a timer, a cap, or a drop-oldest policy: an upload intent and its Files live and die together. On startup every intent left over from a previous run is replayed, oldest-first, alongside live traffic; because processing is idempotent (an already-verified object short-circuits, multipart resumes from its checkpointed sidecar), a replayed intent can never re-upload or duplicate a status row.

A permanently-failing intent cannot starve the rest of the backlog: each intent gets its own exponential backoff (5 s, doubling, capped at 2000 s — Humble Fluent Bit's own scheduler defaults) and the worker sweeps oldest-first, skipping whatever is still backing off rather than retrying the same head-of-line entry forever. The queue's depth is surfaced as a cheap observability hook in the ~/ready service's message text.

This is at-least-once, not exactly-once: a crash between a status Record being forwarded and its intent being acked can replay that status Record after restart (the Uploader's own dedup set is in-memory, so a fresh process may re-emit one it already sent). Consumers should key on the latest row per (local_path, storage_type), same as the Records path.

A Record referencing no Files is enqueued and acked in the same pass with no retries, so it never lingers in the queue's steady state. Note that base64-heavy Records (e.g. a map's save_base64: on) inline file content directly into the payload and so consume queue-directory bytes proportionally faster than a plain File reference.

Thumbnails: optional previews for image and video Files

A dashboard drawing a grid of inspection photos otherwise pulls every full-size camera JPEG — and every video file — just to render tiles. With files.thumbnails enabled, the Uploader derives a small JPEG preview from each image or video File and uploads it next to the original as <remote_path>.thumb.jpg, recording it in that File's status Record as thumbnail_path (same s3://bucket/key shape as remote_path; the field is simply absent when there is no preview).

Default is off, and enabling it cannot change what happens to your data:

dc_bridge:
  ros__parameters:
    files:
      thumbnails:
        enabled: true
        max_dimension: 320       # bound on the preview's longest side, in pixels
        ffmpeg_binary: "ffmpeg"  # override if it isn't on PATH

A preview is a derived artefact and is treated as strictly secondary to the File it comes from:

• It is generated only after the original is uploaded and verified on that Destination, never before or in parallel.
• Every failure — no ffmpeg on PATH, an undecodable File, a store that rejects the preview — is swallowed. It never fails, delays, retries, or blocks the primary upload; the File uploads exactly as it would with the feature off, and the only visible difference is a missing thumbnail_path plus a Bridge warning counting how many previews couldn't be generated.
• Local previews are scratch files under <shipper.data_dir>/uploader/thumbs/, removed before the Record finishes processing. So a preview never joins the un-uploaded pool retention measures and can never outlive its original: an intent retention sheds never uploaded, so no preview was ever derived from it.
• Remote previews are idempotent by key. A Bridge restart mid-upload replays the intent, finds the preview already on the store, and neither re-decodes nor re-uploads it.

max_dimension bounds the longest side; aspect ratio is preserved and a File already smaller than the bound is never upscaled. Images become a scaled still; videos become their first frame. Files that are neither (map YAML, logs) are skipped entirely rather than handed to a decoder.

The decode dependency: previews are generated by shelling out to the ffmpeg CLI, the same toolchain files.ffprobe_binary already uses for the duration column — one tool covering both stills and video first-frames, kept out of dc_uploader's link line, and crash-isolated in a child process. dc_uploader is built from the dc_bridge colcon package, and that package's package.xml declares ffmpeg as an exec_depend, so rosdep install provides it; if it is missing at runtime dc_uploader still uploads normally and simply produces no previews.

Retention: bounded local storage for un-uploaded Files

The intent queue above never abandons an intent by design — an intent and its Files live and die together, and deleting the only copy of data as a side effect of queue management is forbidden. That means a robot offline for weeks, or pointed at a store that's unreachable for weeks, accumulates un-uploaded Files on disk without bound, same as Humble did. A full disk takes down more than DC, so files.retention is the one, explicit, opt-in, audited mechanism allowed to abandon data under disk pressure — nothing else in the system deletes a File it hasn't verified was uploaded.

Default is off: with files.retention absent (or both limits at their defaults), behavior is unchanged — accumulation, matching Humble. Configuring it is a deliberate choice to shed data rather than run out of disk:

dc_bridge:
  ros__parameters:
    files:
      retention:
        max_bytes: 10737418240   # cap on the un-uploaded Files pool; 0/absent = unlimited (default)
        max_age_days: 30         # optional; whichever limit is hit first

Scope is exactly the pool of Files referenced by intents still pending in <shipper.data_dir>/queue/upload/ — the same disk-backed queue the durability section above describes. A File the Uploader has already verified everywhere but is still waiting to physically delete (files.delete_when_sent, e.g. after a filesystem error) is never retention's victim; that's delete_when_sent's own retry loop to resolve, not retention's.

When either limit is exceeded, dc_uploader sheds the oldest eligible intent — its File(s) and its queue entry, always together, never one without the other — and repeats until back under both limits. Each shed File gets its own audit row through the normal dc.files metadata path: deleted: true, uploaded: false is the queryable "shed without upload" signature, distinct from a normal delete_when_sent deletion (which is always uploaded: true, since that path only ever deletes a File already confirmed on every Destination). Emitting the audit row is best-effort — a Shipper outage doesn't block the shed itself — but every shed always logs a rate-limited warning, since data was just abandoned without ever leaving the robot.

A Destination named by metadata_destination may declare no inputs at all — being named there is itself what routes the dc.files Tag to it. That is the usual shape when the status log lives in its own table: one blessed file Destination (typically anchoring a passthrough postgres sink, Recipes above) with inputs for the Records, and a second one with no inputs for the File status log.

inputs is otherwise mandatory: a Destination that neither lists a topic nor receives dc.files would have nothing to deliver, and is rejected at Bridge startup.

Raw topic collection

Sometimes the point is not to measure three specific things well, but to capture everything on the ROS graph and put it somewhere — a bring-up session, a field incident, a robot whose interesting topics you don't know yet.

For that, dc_bridge has a generic-subscription mode. It subscribes to topics it was never compiled against, converts each message to JSON, and ships it to a Destination like any other Record. No Measurement plugin, no .msg header, no rebuild of DC when the robot's message packages change.

dc_bridge:
  ros__parameters:
    shipper:
      data_dir: "$HOME/.dc/buffer"
    destinations: ["raw_log"]
    raw_log:
      type: file              # blessed types: file | vector — or postgres/s3/console via
                               # its passthrough recipe (destinations.md#recipes-postgres-s3-console-via-passthrough)
      receives: records
      path: "/tmp/dc/raw_records.ndjson"
      time_key: "date"
    raw:
      enabled: true
      destination: "raw_log"

ros2 launch dc_bringup dc_raw.launch.py

dc_raw.launch.py starts the Bridge alone — no measurement_server, no group_server, no lifecycle manager, no readiness gate (there are no collection nodes to gate). The default params file is dc_bringup/params/dc_raw_params.yaml; point the launch file at your own with dc_params_file:=…. The same raw: block also works inside a normal dc_bringup.launch.py params file, next to your Measurements.

How it works

DC uses exactly what ROS 2 offers for this — the same mechanism ros2 bag record -a and PlotJuggler use:

1. get_topic_names_and_types() lists the graph. Topics are filtered by the include/exclude patterns below.
2. rclcpp::GenericSubscription subscribes to a topic given only its name and type string, receiving raw serialized bytes.
3. The type's two type support libraries are loaded at run time — rosidl_typesupport_cpp to deserialize, rosidl_typesupport_introspection_cpp to describe the fields — and the message is walked into JSON.
4. The Record is Tagged dc.raw.<topic> and routed to raw.destination.

The only requirement is that the message package is on the Bridge's AMENT_PREFIX_PATH — the same requirement ros2 topic echo has. A type that can't be resolved is logged once and skipped; the rest of the graph is still collected.

Tags and routing

A raw Record's Tag is the topic name with the leading / dropped and the rest turned into dots, under the dc.raw. namespace: