Collecting ROS data with ReductBridge

This guide explains how the bridge.toml configuration from the ROS applications guide turns ROS 2 topics into time-indexed ReductStore records. The complete configuration is available as the bridge.toml example.

ReductBridge connects three layers: a ROS 2 input subscribes to topics, a pipeline enriches and routes records, and a ReductStore remote writes them to a bucket.

# ROS 2 input named "robot".
[inputs.ros2.robot]
# Use the same domain as the publishers.
domain_id = 0
node_name = "reduct_bridge_robot"
queue_size = 128
# Resolve ROS message schemas from this installation.
schema_paths = ["/opt/ros/jazzy"]

# Subscribe to camera images.
[[inputs.ros2.robot.topics]]
name = "/camera/image_raw"
entry_name = "/camera/image_raw"
# Use the message acquisition time.
# timestamp-start
timestamp = { field = "header.stamp", format = "ros_stamp" }
# timestamp-end
# Add labels that are constant for this topic.
labels = [
  # camera-static-start
  { static = { source = "ros2", topic = "/camera/image_raw", sensor = "camera" } }
  # camera-static-end
]

# Subscribe to GPS fixes.
[[inputs.ros2.robot.topics]]
name = "/gps/fix"
entry_name = "/gps/fix"
timestamp = { field = "header.stamp", format = "ros_stamp" }
# Extract coordinates and add constant labels.
labels = [
  # gps-dynamic-start
  { field = "latitude", label = "x" },
  { field = "longitude", label = "y" },
  { field = "altitude", label = "z" },
  # gps-dynamic-end
  { static = { source = "ros2", topic = "/gps/fix", sensor = "gps" } }
]

# Subscribe to IMU measurements.
[[inputs.ros2.robot.topics]]
name = "/imu/data"
entry_name = "/imu/data"
timestamp = { field = "header.stamp", format = "ros_stamp" }
labels = [
  { static = { source = "ros2", topic = "/imu/data", sensor = "imu" } }
]

# Route the ROS input to the local remote.
# pipeline-route-start
[pipelines.ros_to_reductstore]
remote = "local"
inputs = ["robot"]
# pipeline-route-end
# Copy the latest GPS coordinates and identify the robot.
labels = [
  # gps-propagation-start
  { from = "/gps/fix", labels = ["x", "y", "z"], to = "/camera/image_raw" },
  { from = "/gps/fix", labels = ["x", "y", "z"], to = "/imu/data" },
  # gps-propagation-end
  # robot-label-start
  { static = { robot = "robot-1" }, to = "*" }
  # robot-label-end
]

# Write records to the local ReductStore instance.
[remotes.reduct.local]
url = "http://127.0.0.1:8383"
token_api = "my-token"
bucket = "robot-data"
prefix = ""

# Create a rolling bucket when it is missing.
[remotes.reduct.local.create_bucket]
quota_type = "FIFO"
quota_size = "20GB"

Data flow

The configuration moves each ROS message through the input, pipeline, and remote stages before writing it to the local ReductStore bucket.

Connecting the ROS 2 input

The [inputs.ros2.robot] table creates a ROS 2 input named robot. domain_id = 0 must match the domain used by the publishers, while node_name is the unique ROS node name used by ReductBridge. queue_size = 128 configures the keep-last depth of each subscription.

ReductBridge needs the ROS message definitions to decode fields and attach schemas to stored entries. schema_paths = ["/opt/ros/jazzy"] tells it to look under /opt/ros/jazzy/share/<package>/msg/, which is the ROS installation in the Jazzy Docker image. Add a workspace install prefix to this list when topics use custom message types. See the ROS 2 input reference for schema lookup and runtime details.

Subscribing to topics

Every [[inputs.ros2.robot.topics]] block creates a subscription. This example subscribes to /camera/image_raw, /gps/fix, and /imu/data. name is the ROS topic, while entry_name is the ReductStore entry that receives its records. They are deliberately identical here, producing one entry per topic in the robot-data bucket.

Topic names can also contain *, for example /camera/*, when the same rules should apply to several discovered topics. If entry_name is omitted, ReductBridge uses the resolved topic name. Each payload is kept in its original serialized CDR form with content type application/cdr.

Handling ROS schemas

When ReductBridge subscribes to a topic, it discovers the ROS message type and resolves its .msg definition from schema_paths or the ROS environment. It also resolves dependent message definitions, so nested types can be decoded. The original message remains a binary CDR record; schema handling does not convert or replace the stored payload.

ReductBridge automatically stores the resolved metadata in the $schema attachment of the destination entry. This is an entry-level attachment shared by all records in that entry, so the schema is stored once alongside the data rather than repeated in every record. It contains:

• encoding: cdr for ROS 2 records.
• topic: the original ROS topic name.
• schema_name: the ROS message type, such as sensor_msgs/msg/Image.
• schema: the full message definition needed to decode the payload.

The same resolved schema is used during ingestion when a timestamp or dynamic label refers to a message field. ReductBridge decodes the CDR payload in memory, reads paths such as header.stamp or latitude, and stores the selected values as the record timestamp or labels. Only the configured values are extracted; the record content written to ReductStore remains the original CDR bytes.

The $schema attachment also keeps the raw records self-describing for later queries. The ReductROS Raw Messages extension uses it to select the correct decoder, return individual ROS messages as JSON, or export records from one or more entries into MCAP episodes. This allows the same stored data to support filtering and inspection as JSON as well as export to standard robotics tooling without converting it during ingestion.

Assigning record timestamps

Each topic uses the same timestamp mapping:

timestamp = { field = "header.stamp", format = "ros_stamp" }

The field path selects the standard ROS header timestamp, and ros_stamp converts its sec and nanosec fields into the timestamp indexed by ReductStore. This records when the sensor message was produced rather than when ReductBridge received it. In the example publisher, all three messages in a cycle share one header timestamp, which makes time-range queries across the entries consistent.

note

Timestamp extraction requires ReductBridge to decode the CDR payload using the resolved schema. If the field cannot be decoded, ReductBridge logs a warning and falls back to the timestamp supplied by the ROS middleware, or the ingest time when no middleware timestamp is available.

Adding static and dynamic labels

Static labels assign known values to every record from a topic. For example, camera records receive source=ros2, topic=/camera/image_raw, and sensor=camera:

{ static = { source = "ros2", topic = "/camera/image_raw", sensor = "camera" } }

Dynamic labels read values from the decoded message. The GPS subscription maps the latitude, longitude, and altitude fields to labels named x, y, and z:

{ field = "latitude", label = "x" },
{ field = "longitude", label = "y" },
{ field = "altitude", label = "z" },

Field paths use dot notation for nested values and numeric segments for array indexes. ReductBridge applies static labels first and dynamic field labels afterward, so a field label overrides a static label with the same name. If a field is absent or cannot be decoded, that dynamic label is omitted while the static labels are still stored.

note

Configuring dynamic field labels enables CDR payload parsing for that topic. ReductBridge needs a valid ROS message schema to resolve and read the configured field paths, just as it does for timestamp assignment.

Propagating labels between topics

The pipeline connects the robot input to the ReductStore remote named local:

[pipelines.ros_to_reductstore]
remote = "local"
inputs = ["robot"]

Pipeline label rules can enrich one topic with context extracted from another. These two rules remember the latest x, y, and z labels seen on the GPS entry and add them to subsequent camera and IMU records:

{ from = "/gps/fix", labels = ["x", "y", "z"], to = "/camera/image_raw" },
{ from = "/gps/fix", labels = ["x", "y", "z"], to = "/imu/data" },

The from and to patterns match entry_name, not the original topic name. The rule is stateful: when a /gps/fix record arrives, the pipeline caches its selected labels; when a matching target record arrives later, the pipeline copies the cached values onto it. This is a last-seen-value operation, not a timestamp join. Target records received before the first GPS record have no copied coordinates, and a target record may receive coordinates from the previous publishing cycle when it arrives before the current GPS update.

The final pipeline rule adds a robot identifier to every entry because * matches all entry names:

{ static = { robot = "robot-1" }, to = "*" }

This combination makes camera and IMU records directly filterable by robot and the latest known position without changing their original payloads.

Writing to ReductStore

The [remotes.reduct.local] table defines the local destination referenced by the pipeline. It connects through the host port exposed by Docker Compose, authenticates with my-token, and writes every topic to the robot-data bucket. An empty prefix preserves the configured entry names; set it to a value such as robot-1/ to place all entries under that path.

ReductBridge batches records before sending them to ReductStore over HTTP. Combining multiple records into fewer requests reduces HTTP overhead and makes communication more efficient, especially for high-rate ROS topics. A batch is flushed when its configured record count, payload size, or time interval is reached.

The [remotes.reduct.local.create_bucket] table creates the bucket if it does not exist. quota_type = "FIFO" with quota_size = "20GB" makes it a rolling 20 GB store: after the quota is reached, ReductStore removes the oldest records to make room for new data. Existing bucket settings are not changed. See the ReductStore remote reference for batching and attachment options.