robot-bringup

ロボットの起動スキル

このスキルを使用する場面

• systemd を使って、起動時にロボットの ROS2 スタック全体を自動的に開始するよう設定する
• ROS2 ワークスペースを正しく読み込み、DDS 環境を設定する systemd ユニットファイルを作成する
• ハードウェア、ドライバー、認識、アプリケーションの各レイヤーの launch ファイルを組み合わせて、単一の bringup を構成する
• 依存するノード間の競合状態を避けるため、ヘルスチェックを伴う順序付き起動を設定する
• カメラ、LiDAR、シリアルデバイスに対して、デバイス名を確定できる udev ルールを作成する
• ロボットと基地局間で複数マシンの ROS2 ディスカバリーを行うため、CycloneDDS または FastDDS を設定する
• 本番環境のロボットシステムにウォッチドッグとハートビート監視を実装する
• 長期間稼働するロボットのデプロイ向けに、ログローテーションと構造化ロギングを設定する
• 終了前にアクチュエーターを安全な状態にする、グレースフルシャットダウンハンドラーを作成する
• 起動時の障害、サービスの順序に関する問題、デバイス列挙の競合をデバッグする

ロボットの Bringup スタック

本番環境のロボットの bringup では、ハードウェアの初期化からアプリケーションレベルのノードまで、レイヤー化された起動シーケンスに従います。各レイヤーは、その下のレイヤーに依存します。

┌─────────────────────────────────────────────────────────────────────┐
│                        APPLICATION LAYER                            │
│  Navigation, manipulation, mission planning, HRI                    │
├─────────────────────────────────────────────────────────────────────┤
│                        PERCEPTION LAYER                             │
│  Object detection, SLAM, point cloud filtering, sensor fusion       │
├─────────────────────────────────────────────────────────────────────┤
│                         DRIVER LAYER                                │
│  Camera drivers, LiDAR drivers, motor controllers, IMU              │
├─────────────────────────────────────────────────────────────────────┤
│                        HARDWARE LAYER                               │
│  udev rules, device enumeration, USB reset, firmware check          │
├─────────────────────────────────────────────────────────────────────┤
│                      ROS2 ENVIRONMENT                               │
│  Source workspace, set RMW, ROS_DOMAIN_ID, DDS config               │
├─────────────────────────────────────────────────────────────────────┤
│                    SYSTEMD TARGETS & SERVICES                       │
│  network-online.target → robot-hw.target → robot-bringup.target     │
├─────────────────────────────────────────────────────────────────────┤
│                      LINUX BOOT (systemd)                           │
│  BIOS/UEFI → GRUB → kernel → systemd init                          │
├─────────────────────────────────────────────────────────────────────┤
│                         HARDWARE BOOT                               │
│  Power supply, onboard computer, peripherals                        │
└─────────────────────────────────────────────────────────────────────┘

ROS2 の systemd サービスユニット

基本的な ROS2 サービスユニット

サービスファイルを /etc/systemd/system/ に配置します。このテンプレートは、ウォッチドッグをサポートしながら、ROS2 の launch ファイルを長時間稼働するサービスとして起動します。

# /etc/systemd/system/robot-bringup.service
[Unit]
Description=Robot ROS2 Bringup Stack
Documentation=https://github.com/my-org/my-robot
After=network-online.target robot-hw.target
Wants=network-online.target
Requires=robot-hw.target

[Service]
Type=notify
User=robot
Group=robot
WorkingDirectory=/home/robot

# Load ROS2 environment variables from a dedicated env file
EnvironmentFile=/etc/robot/ros2.env

# Pre-start check: verify critical devices exist
ExecStartPre=/usr/local/bin/robot-device-check.sh

# Start the ROS2 launch file via bash so we can source the workspace
ExecStart=/bin/bash -c '\
  source /opt/ros/${ROS_DISTRO}/setup.bash && \
  source /home/robot/ros2_ws/install/setup.bash && \
  exec ros2 launch my_robot_bringup bringup.launch.py'

# Graceful shutdown: send SIGINT first (Ctrl+C equivalent for ROS2)
ExecStop=/bin/kill -INT $MAINPID
TimeoutStopSec=30

# Restart on failure, but not on clean exit
Restart=on-failure
RestartSec=5

# systemd watchdog: service must call sd_notify(WATCHDOG=1) within this interval
WatchdogSec=30

# Process management
KillMode=mixed
KillSignal=SIGINT
FinalKillSignal=SIGKILL
TimeoutStartSec=60

# Logging
StandardOutput=journal
StandardError=journal
SyslogIdentifier=robot-bringup

[Install]
WantedBy=multi-user.target

systemd での環境設定

systemd では .bashrc が読み込まれないため、環境変数は .bashrc を読み込むのではなく、専用ファイルに保存します。

# /etc/robot/ros2.env
# ROS2 distribution
ROS_DISTRO=humble

# DDS middleware selection
RMW_IMPLEMENTATION=rmw_cyclonedds_cpp

# Domain isolation: unique per robot to avoid cross-talk
ROS_DOMAIN_ID=42

# CycloneDDS configuration file path
CYCLONEDDS_URI=file:///etc/robot/cyclonedds.xml

# Disable localhost-only mode for multi-machine setups
ROS_LOCALHOST_ONLY=0

# Logging configuration
ROS_LOG_DIR=/var/log/ros2
RCUTILS_LOGGING_USE_STDOUT=0
RCUTILS_COLORIZED_OUTPUT=0

# Robot-specific configuration
ROBOT_NAME=my_robot_01
ROBOT_CONFIG_DIR=/etc/robot/config

サービス間の依存関係

明示的な順序設定を行い、ロボットスタックを複数の systemd サービスに分割します。これにより、各レイヤーを個別に再起動でき、障害の切り分けも容易になります。

# /etc/systemd/system/robot-drivers.service
[Unit]
Description=Robot Hardware Drivers (cameras, LiDAR, IMU, motors)
After=network-online.target robot-hw.target
Wants=network-online.target
Requires=robot-hw.target

[Service]
Type=notify
User=robot
EnvironmentFile=/etc/robot/ros2.env
ExecStart=/bin/bash -c '\
  source /opt/ros/${ROS_DISTRO}/setup.bash && \
  source /home/robot/ros2_ws/install/setup.bash && \
  exec ros2 launch my_robot_bringup drivers.launch.py'
Restart=on-failure
RestartSec=5
WatchdogSec=30
KillMode=mixed
KillSignal=SIGINT
TimeoutStopSec=20
StandardOutput=journal
SyslogIdentifier=robot-drivers

[Install]
WantedBy=robot-bringup.target

# /etc/systemd/system/robot-perception.service
[Unit]
Description=Robot Perception Stack (SLAM, detection, sensor fusion)
After=robot-drivers.service
Requires=robot-drivers.service
PartOf=robot-drivers.service

[Service]
Type=notify
User=robot
EnvironmentFile=/etc/robot/ros2.env
ExecStart=/bin/bash -c '\
  source /opt/ros/${ROS_DISTRO}/setup.bash && \
  source /home/robot/ros2_ws/install/setup.bash && \
  exec ros2 launch my_robot_bringup perception.launch.py'
Restart=on-failure
RestartSec=5
WatchdogSec=30
KillMode=mixed
KillSignal=SIGINT
TimeoutStopSec=20
StandardOutput=journal
SyslogIdentifier=robot-perception

[Install]
WantedBy=robot-bringup.target

# /etc/systemd/system/robot-application.service
[Unit]
Description=Robot Application Layer (navigation, planning, HRI)
After=robot-perception.service
Requires=robot-perception.service
PartOf=robot-perception.service

[Service]
Type=notify
User=robot
EnvironmentFile=/etc/robot/ros2.env
ExecStart=/bin/bash -c '\
  source /opt/ros/${ROS_DISTRO}/setup.bash && \
  source /home/robot/ros2_ws/install/setup.bash && \
  exec ros2 launch my_robot_bringup application.launch.py'
Restart=on-failure
RestartSec=10
WatchdogSec=30
KillMode=mixed
KillSignal=SIGINT
TimeoutStopSec=20
StandardOutput=journal
SyslogIdentifier=robot-application

[Install]
WantedBy=robot-bringup.target

再起動ポリシーと障害復旧

サービスに根本的な問題（デバイスの欠落や設定エラーなど）がある場合に再起動ループを防ぐため、レート制限を設定します。

# Add to the [Service] section of any robot service
Restart=on-failure
RestartSec=5

# Allow at most 5 restart attempts within 120 seconds
StartLimitIntervalSec=120
StartLimitBurst=5

# Ramp up restart delay to avoid thrashing
# RestartSec can also be set dynamically via drop-in overrides:
#   RestartSec=5   (first few retries, fast recovery)
#   After StartLimitBurst is hit, the unit enters failed state
#   Use systemctl reset-failed robot-drivers.service to retry

# On final failure, trigger an alert
OnFailure=robot-alert@%n.service

リソース制限と cgroups

暴走したノードがシステムの残りの部分を圧迫しないよう、リソース使用量を制限します。

# Add to the [Service] section
# Limit memory to 2 GB (hard kill at 2.5 GB)
MemoryMax=2G
MemoryHigh=1800M

# Limit CPU to 300% (3 cores on a multi-core system)
CPUQuota=300%

# Set real-time scheduling priority for time-critical drivers
# Requires the user to have rtprio permissions in /etc/security/limits.d/
Nice=-5
IOSchedulingClass=realtime
IOSchedulingPriority=0

# Restrict filesystem access
ProtectHome=read-only
ProtectSystem=strict
ReadWritePaths=/var/log/ros2 /tmp
PrivateTmp=true

Launch ファイルの構成とレイヤー化

Launch レイヤーのアーキテクチャ

launch ファイルを、systemd サービスのアーキテクチャに対応するレイヤーに整理します。各レイヤーは、単独でテストできる独立した launch ファイルです。

bringup.launch.py  (top-level: composes all layers)
├── hardware.launch.py     (udev checks, device readiness)
├── drivers.launch.py      (camera, LiDAR, IMU, motor drivers)
│   ├── camera.launch.py
│   ├── lidar.launch.py
│   └── motors.launch.py
├── perception.launch.py   (SLAM, detection, fusion)
│   ├── slam.launch.py
│   └── detection.launch.py
└── application.launch.py  (navigation, planning, HRI)
    ├── navigation.launch.py
    └── mission.launch.py

ハードウェアレイヤーの Launch

# my_robot_bringup/launch/hardware.launch.py
from launch import LaunchDescription
from launch.actions import LogInfo, ExecuteProcess, TimerAction
from launch.conditions import IfCondition
from launch.substitutions import LaunchConfiguration, EnvironmentVariable

def generate_launch_description():
    # Declare arguments for hardware configuration
    robot_name = LaunchConfiguration('robot_name',
        default=EnvironmentVariable('ROBOT_NAME', default_value='default_robot'))

    # Check that critical devices are present
    check_camera = ExecuteProcess(
        cmd=['test', '-e', '/dev/robot/camera_front'],
        name='check_camera_front',
        output='screen',
    )

    check_lidar = ExecuteProcess(
        cmd=['test', '-e', '/dev/robot/lidar'],
        name='check_lidar',
        output='screen',
    )

    check_imu = ExecuteProcess(
        cmd=['test', '-e', '/dev/robot/imu'],
        name='check_imu',
        output='screen',
    )

    log_ready = TimerAction(
        period=2.0,
        actions=[LogInfo(msg='Hardware checks passed, devices ready')],
    )

    return LaunchDescription([
        check_camera,
        check_lidar,
        check_imu,
        log_ready,
    ])

ドライバーレイヤーの Launch

# my_robot_bringup/launch/drivers.launch.py
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription, GroupAction
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node, SetRemap
from launch_ros.substitutions import FindPackageShare

def generate_launch_description():
    use_sim = LaunchConfiguration('use_sim', default='false')
    camera_config = LaunchConfiguration('camera_config', default='default')

    # Camera driver
    camera_node = Node(
        package='usb_cam',
        executable='usb_cam_node_exe',
        name='camera_front',
        parameters=[PathJoinSubstitution([
            FindPackageShare('my_robot_bringup'), 'config', 'camera_front.yaml'
        ])],
        remappings=[('/image_raw', '/camera/front/image_raw')],
    )

    # LiDAR driver
    lidar_node = Node(
        package='sllidar_ros2',
        executable='sllidar_node',
        name='lidar',
        parameters=[{
            'serial_port': '/dev/robot/lidar',
            'serial_baudrate': 460800,
            'frame_id': 'lidar_link',
            'angle_compensate': True,
        }],
    )

    # IMU driver
    imu_node = Node(
        package='imu_driver',
        executable='imu_node',
        name='imu',
        parameters=[{
            'port': '/dev/robot/imu',
            'frame_id': 'imu_link',
            'publish_rate': 100.0,
        }],
    )

    # Motor controller driver
    motor_node = Node(
        package='motor_driver',
        executable='motor_controller_node',
        name='motor_controller',
        parameters=[PathJoinSubstitution([
            FindPackageShare('my_robot_bringup'), 'config', 'motors.yaml'
        ])],
    )

    return LaunchDescription([
        DeclareLaunchArgument('use_sim', default_value='false'),
        DeclareLaunchArgument('camera_config', default_value='default'),
        camera_node,
        lidar_node,
        imu_node,
        motor_node,
    ])

認識レイヤーの Launch

# my_robot_bringup/launch/perception.launch.py
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, GroupAction
from launch.conditions import IfCondition
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node, ComposableNodeContainer, LoadComposableNode
from launch_ros.descriptions import ComposableNode
from launch_ros.substitutions import FindPackageShare

def generate_launch_description():
    enable_slam = LaunchConfiguration('enable_slam', default='true')
    enable_detection = LaunchConfiguration('enable_detection', default='true')

    # Use a composable node container for zero-copy perception pipeline
    perception_container = ComposableNodeContainer(
        name='perception_container',
        namespace='',
        package='rclcpp_components',
        executable='component_container_mt',
        composable_node_descriptions=[
            ComposableNode(
                package='image_proc',
                plugin='image_proc::RectifyNode',
                name='rectify',
                remappings=[('image', '/camera/front/image_raw')],
            ),
            ComposableNode(
                package='my_detection',
                plugin='my_detection::DetectorNode',
                name='detector',
                parameters=[PathJoinSubstitution([
                    FindPackageShare('my_robot_bringup'), 'config', 'detector.yaml'
                ])],
            ),
        ],
        condition=IfCondition(enable_detection),
    )

    # SLAM node
    slam_node = Node(
        package='slam_toolbox',
        executable='async_slam_toolbox_node',
        name='slam',
        parameters=[PathJoinSubstitution([
            FindPackageShare('my_robot_bringup'), 'config', 'slam.yaml'
        ])],
        condition=IfCondition(enable_slam),
    )

    return LaunchDescription([
        DeclareLaunchArgument('enable_slam', default_value='true'),
        DeclareLaunchArgument('enable_detection', default_value='true'),
        perception_container,
        slam_node,
    ])

アプリケーションレイヤーの Launch

# my_robot_bringup/launch/application.launch.py
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node
from launch_ros.substitutions import FindPackageShare

def generate_launch_description():
    nav_params = LaunchConfiguration('nav_params', default=PathJoinSubstitution([
        FindPackageShare('my_robot_bringup'), 'config', 'nav2_params.yaml'
    ]))

    # Include Nav2 bringup
    nav2_bringup = IncludeLaunchDescription(
        PythonLaunchDescriptionSource(PathJoinSubstitution([
            FindPackageShare('nav2_bringup'), 'launch', 'bringup_launch.py'
        ])),
        launch_arguments={
            'params_file': nav_params,
            'use_sim_time': LaunchConfiguration('use_sim', default='false'),
        }.items(),
    )

    # Mission planner
    mission_node = Node(
        package='my_mission',
        executable='mission_planner',
        name='mission_planner',
        parameters=[PathJoinSubstitution([
            FindPackageShare('my_robot_bringup'), 'config', 'mission.yaml'
        ])],
    )

    return LaunchDescription([
        DeclareLaunchArgument('nav_params', default_value=''),
        DeclareLaunchArgument('use_sim', default_value='false'),
        nav2_bringup,
        mission_node,
    ])

トップレベルの Bringup Launch

この launch ファイルは、シミュレーションと実機の切り替えや、ロボットのバリアント選択に使う条件付き引数を指定し、すべてのレイヤーを単一のエントリーポイントにまとめます。

# my_robot_bringup/launch/bringup.launch.py
from launch import LaunchDescription
from launch.actions import (
    DeclareLaunchArgument, IncludeLaunchDescription,
    GroupAction, LogInfo, TimerAction,
)
from launch.conditions import IfCondition, UnlessCondition
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import (
    LaunchConfiguration, PathJoinSubstitution, PythonExpression,
)
from launch_ros.actions import PushRosNamespace, SetParameter
from launch_ros.substitutions import FindPackageShare

def generate_launch_description():
    pkg_share = FindPackageShare('my_robot_bringup')
    use_sim = LaunchConfiguration('use_sim')
    robot_variant = LaunchConfiguration('robot_variant')
    enable_perception = LaunchConfiguration('enable_perception')
    enable_navigation = LaunchConfiguration('enable_navigation')

    # Hardware layer (skip in simulation)
    hardware_launch = GroupAction(
        actions=[
            LogInfo(msg='Starting hardware layer...'),
            IncludeLaunchDescription(
                PythonLaunchDescriptionSource(
                    PathJoinSubstitution([pkg_share, 'launch', 'hardware.launch.py'])
                ),
            ),
        ],
        condition=UnlessCondition(use_sim),
    )

    # Driver layer
    drivers_launch = GroupAction(
        actions=[
            LogInfo(msg='Starting driver layer...'),
            IncludeLaunchDescription(
                PythonLaunchDescriptionSource(
                    PathJoinSubstitution([pkg_share, 'launch', 'drivers.launch.py'])
                ),
                launch_arguments={'use_sim': use_sim}.items(),
            ),
        ],
    )

    # Perception layer (conditional)
    perception_launch = GroupAction(
        actions=[
            LogInfo(msg='Starting perception layer...'),
            IncludeLaunchDescription(
                PythonLaunchDescriptionSource(
                    PathJoinSubstitution([pkg_share, 'launch', 'perception.launch.py'])
                ),
            ),
        ],
        condition=IfCondition(enable_perception),
    )

    # Application layer (conditional)
    application_launch = GroupAction(
        actions=[
            LogInfo(msg='Starting application layer...'),
            IncludeLaunchDescription(
                PythonLaunchDescriptionSource(
                    PathJoinSubstitution([pkg_share, 'launch', 'application.launch.py'])
                ),
                launch_arguments={'use_sim': use_sim}.items(),
            ),
        ],
        condition=IfCondition(enable_navigation),
    )

    return LaunchDescription([
        DeclareLaunchArgument('use_sim', default_value='false',
            description='Use simulation instead of real hardware'),
        DeclareLaunchArgument('robot_variant', default_value='standard',
            description='Robot variant: standard, heavy_payload, outdoor'),
        DeclareLaunchArgument('enable_perception', default_value='true',
            description='Enable perception stack'),
        DeclareLaunchArgument('enable_navigation', default_value='true',
            description='Enable navigation and application stack'),

        # Set use_sim_time globally
        SetParameter(name='use_sim_time', value=use_sim),

        LogInfo(msg=['Bringing up robot variant: ', robot_variant]),

        hardware_launch,
        drivers_launch,
        perception_launch,
        application_launch,
    ])

条件付き読み込み（シミュレーションと実機、ロボットのバリアント）

IfCondition、UnlessCondition、PythonExpression を使って、実行時引数に基づいて設定を切り替えます。

from launch.conditions import IfCondition, UnlessCondition
from launch.substitutions import PythonExpression, LaunchConfiguration

robot_variant = LaunchConfiguration('robot_variant')

# Load a variant-specific config file
variant_config = PathJoinSubstitution([
    FindPackageShare('my_robot_bringup'), 'config', 'variants',
    PythonExpression(["'", robot_variant, "' + '.yaml'"]),
])

# Conditional node: only load the arm driver for heavy_payload variant
arm_driver = Node(
    package='arm_driver',
    executable='arm_controller_node',
    name='arm_controller',
    condition=IfCondition(
        PythonExpression(["'", robot_variant, "' == 'heavy_payload'"])
    ),
)

ヘルスチェックを伴う順序付き起動

起動依存関係グラフ

まだ存在しないトピックを購読したり、利用可能になる前にサービスを呼び出したりしないよう、ノードは特定の順序で起動する必要があります。

                    ┌──────────────┐
                    │  motors.srv  │
                    └──────┬───────┘
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ camera   │ │  lidar   │ │   imu    │
        └────┬─────┘ └────┬─────┘ └────┬─────┘
             │            │            │
             ▼            ▼            ▼
        ┌──────────────────────────────────┐
        │        perception / SLAM         │
        └──────────────┬───────────────────┘
                       │
                       ▼
        ┌──────────────────────────────────┐
        │      navigation / planning       │
        └──────────────────────────────────┘

ヘルスチェック用スクリプト

ExecStartPre でヘルスチェック用スクリプトを使い、依存先の準備が整うまでサービスの起動を保留します。

#!/bin/bash
# /usr/local/bin/robot-device-check.sh
# Verifies that all required hardware devices are present before starting drivers.
# Exit code 0 = all devices found, non-zero = missing device.

set -euo pipefail

REQUIRED_DEVICES=(
    "/dev/robot/camera_front"
    "/dev/robot/lidar"
    "/dev/robot/imu"
    "/dev/robot/motor_controller"
)

TIMEOUT=30
POLL_INTERVAL=1
elapsed=0

for device in "${REQUIRED_DEVICES[@]}"; do
    elapsed=0
    while [ ! -e "$device" ]; do
        if [ "$elapsed" -ge "$TIMEOUT" ]; then
            echo "ERROR: Device $device not found after ${TIMEOUT}s" >&2
            exit 1
        fi
        echo "Waiting for $device... (${elapsed}s/${TIMEOUT}s)"
        sleep "$POLL_INTERVAL"
        elapsed=$((elapsed + POLL_INTERVAL))
    done
    echo "Found device: $device"
done

echo "All required devices are present."
exit 0

#!/bin/bash
# /usr/local/bin/wait-for-ros2-nodes.sh
# Blocks until specified ROS2 nodes are active.
# Usage: wait-for-ros2-nodes.sh node_name1 node_name2 ...

set -euo pipefail

source /opt/ros/${ROS_DISTRO}/setup.bash
source /home/robot/ros2_ws/install/setup.bash

TIMEOUT=60
POLL_INTERVAL=2

for node_name in "$@"; do
    elapsed=0
    while ! ros2 node list 2>/dev/null | grep -q "$node_name"; do
        if [ "$elapsed" -ge "$TIMEOUT" ]; then
            echo "ERROR: Node $node_name not found after ${TIMEOUT}s" >&2
            exit 1
        fi
        echo "Waiting for node $node_name... (${elapsed}s/${TIMEOUT}s)"
        sleep "$POLL_INTERVAL"
        elapsed=$((elapsed + POLL_INTERVAL))
    done
    echo "Node active: $node_name"
done

echo "All required nodes are active."
exit 0

トピック待機パターン

トピックがパブリッシュされるまで待機する再利用可能な Python ユーティリティです。launch ファイルでの順序付き起動に役立ちます。

#!/usr/bin/env python3
# wait_for_topic.py
# Usage: python3 wait_for_topic.py /scan sensor_msgs/msg/LaserScan --timeout 30

import argparse
import sys
import time
import importlib

import rclpy
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data


class TopicWaiter(Node):
    def __init__(self, topic_name, msg_type_str, timeout):
        super().__init__('topic_waiter')
        self.received = False
        self.timeout = timeout
        self.start_time = time.time()

        # Dynamically import the message type
        module_name, class_name = msg_type_str.rsplit('/', 1)
        module_name = module_name.replace('/', '.')
        module = importlib.import_module(module_name)
        msg_type = getattr(module, class_name)

        self.sub = self.create_subscription(
            msg_type, topic_name, self._callback, qos_profile_sensor_data)
        self.timer = self.create_timer(1.0, self._check_timeout)
        self.get_logger().info(f'Waiting for topic {topic_name}...')

    def _callback(self, msg):
        self.get_logger().info('Topic is active, message received.')
        self.received = True

    def _check_timeout(self):
        if self.received:
            raise SystemExit(0)
        elapsed = time.time() - self.start_time
        if elapsed > self.timeout:
            self.get_logger().error(f'Timeout after {self.timeout}s')
            raise SystemExit(1)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('topic', help='Topic name to wait for')
    parser.add_argument('msg_type', help='Message type (e.g., sensor_msgs/msg/LaserScan)')
    parser.add_argument('--timeout', type=float, default=30.0)
    args = parser.parse_args()

    rclpy.init()
    node = TopicWaiter(args.topic, args.msg_type, args.timeout)
    rclpy.spin(node)


if __name__ == '__main__':
    main()

順序付き起動のためのライフサイクルノードのオーケストレーション

ライフサイクル（マネージド）ノードを使って、起動順序を保証します。ライフサイクルマネージャーがノードを順番に設定してアクティベートし、各ノードの設定が完了してから次のノードを起動します。

# lifecycle_manager.launch.py
from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description():
    # Lifecycle manager controls the startup/shutdown order
    lifecycle_manager = Node(
        package='nav2_lifecycle_manager',
        executable='lifecycle_manager',
        name='lifecycle_manager',
        output='screen',
        parameters=[{
            # Nodes are transitioned in order: configure, then activate
            'node_names': [
                'motor_controller',
                'camera_driver',
                'lidar_driver',
                'slam',
                'navigation',
            ],
            'autostart': True,
            # Timeout for each node transition
            'bond_timeout': 10.0,
            # Check period for node bonds
            'bond_respawn_max_duration': 2.0,
        }],
    )

    return LaunchDescription([
        lifecycle_manager,
    ])

デバイス名を確定するための udev ルール

カメラ用 udev ルールの作成

USB カメラは起動時に任意の順序で列挙されるため、/dev/video0 が予測できない場合があります。デバイス属性に基づいて安定したシンボリックリンクを割り当てるには、udev ルールを使用します。

# /etc/udev/rules.d/99-robot-cameras.rules
# Assign stable device names based on USB port path (physical location).
# Find attributes with: udevadm info --name=/dev/video0 --attribute-walk

# Front camera: USB hub port 1, interface 0 (video capture)
SUBSYSTEM=="video4linux", ATTRS{idVendor}=="1234", ATTRS{idProduct}=="5678", \
  KERNELS=="1-1.2:1.0", ATTR{index}=="0", \
  SYMLINK+="robot/camera_front", MODE="0666", GROUP="video"

# Rear camera: USB hub port 2, interface 0 (video capture)
SUBSYSTEM=="video4linux", ATTRS{idVendor}=="1234", ATTRS{idProduct}=="5678", \
  KERNELS=="1-1.3:1.0", ATTR{index}=="0", \
  SYMLINK+="robot/camera_rear", MODE="0666", GROUP="video"

# Depth camera (RealSense): by serial number
SUBSYSTEM=="video4linux", ATTRS{idVendor}=="8086", ATTRS{idProduct}=="0b3a", \
  ATTRS{serial}=="123456789", ATTR{index}=="0", \
  SYMLINK+="robot/camera_depth", MODE="0666", GROUP="video"

シリアルデバイス向けのudevルールの作成

シリアルデバイス（IMU、モーターコントローラー、GPS）にも、/dev/ttyUSB* の番号は一定ではないため、安定した名前が必要です。

# /etc/udev/rules.d/99-robot-serial.rules
# IMU on FTDI serial adapter (identified by serial number)
SUBSYSTEM=="tty", ATTRS{idVendor}=="0403", ATTRS{idProduct}=="6001", \
  ATTRS{serial}=="AB0CDEFG", \
  SYMLINK+="robot/imu", MODE="0666", GROUP="dialout"

# Motor controller on USB port path
SUBSYSTEM=="tty", ATTRS{idVendor}=="1a86", ATTRS{idProduct}=="7523", \
  KERNELS=="1-1.4:1.0", \
  SYMLINK+="robot/motor_controller", MODE="0666", GROUP="dialout"

# GPS receiver
SUBSYSTEM=="tty", ATTRS{idVendor}=="1546", ATTRS{idProduct}=="01a7", \
  SYMLINK+="robot/gps", MODE="0666", GROUP="dialout"

# LiDAR (CP2102 adapter)
SUBSYSTEM=="tty", ATTRS{idVendor}=="10c4", ATTRS{idProduct}=="ea60", \
  ATTRS{serial}=="0001", \
  SYMLINK+="robot/lidar", MODE="0666", GROUP="dialout"

再読み込みとテスト

# Reload udev rules without rebooting
sudo udevadm control --reload-rules
sudo udevadm trigger

# Test a rule against a specific device
sudo udevadm test $(udevadm info --query=path --name=/dev/ttyUSB0)

# View all attributes for a device (use to find idVendor, serial, etc.)
udevadm info --name=/dev/ttyUSB0 --attribute-walk

# Monitor udev events in real time (plug/unplug devices to see events)
udevadm monitor --subsystem-match=tty --property

複数マシンでのROS2向けネットワーク設定

静的IPアドレスの設定

netplanを使用して、ロボットの有線インターフェースに静的IPアドレスを割り当てます（Ubuntu 22.04以降）。

# /etc/netplan/01-robot-network.yaml
network:
  version: 2
  ethernets:
    eth0:
      addresses:
        - 10.0.0.10/24
      routes:
        - to: default
          via: 10.0.0.1
      nameservers:
        addresses:
          - 8.8.8.8
          - 8.8.4.4
  wifis:
    wlan0:
      dhcp4: true
      access-points:
        "RobotNetwork":
          password: "securepassword"

# Apply netplan configuration
sudo netplan apply

複数マシン間のDDSディスカバリー

マルチキャストはネットワークセグメントをまたいで機能しない場合があるため、CycloneDDSでは複数マシン構成に明示的なピア設定が必要です。

<CycloneDDS>
  <Domain>
    <General>
      <Interfaces>
        <NetworkInterface name="eth0" priority="default" multicast="false"/>
      </Interfaces>
      <AllowMulticast>false</AllowMulticast>
    </General>
    <Discovery>
      <ParticipantIndex>auto</ParticipantIndex>
      <Peers>
        
        <Peer address="10.0.0.10"/>
        
        <Peer address="10.0.0.20"/>
        
        <Peer address="10.0.0.11"/>
      </Peers>
      <MaxAutoParticipantIndex>30</MaxAutoParticipantIndex>
    </Discovery>
    <Internal>
      <SocketReceiveBufferSize min="10MB"/>
    </Internal>
  </Domain>
</CycloneDDS>

ファイアウォールルール

DDSはディスカバリーとデータ交換に複数のUDPポートを使用します。ロボットとベースステーションの両方で、これらのポートを開放してください。

#!/bin/bash
# /usr/local/bin/robot-firewall-setup.sh
# Open firewall ports for CycloneDDS discovery and data exchange.

# DDS discovery (SPDP) uses UDP port 7400 + (250 * domain_id) + participant_id
# For domain_id=42: base port = 7400 + 250*42 = 17900
DOMAIN_ID=42
BASE_PORT=$((7400 + 250 * DOMAIN_ID))

# Allow discovery (SPDP) multicast/unicast
sudo ufw allow proto udp from 10.0.0.0/24 to any port $BASE_PORT:$((BASE_PORT + 100))

# Allow data exchange (SEDP) user traffic ports
DATA_PORT=$((BASE_PORT + 1))
sudo ufw allow proto udp from 10.0.0.0/24 to any port $DATA_PORT:$((DATA_PORT + 200))

# Allow all traffic on the robot subnet (simpler alternative)
# sudo ufw allow from 10.0.0.0/24

sudo ufw reload
echo "Firewall configured for ROS2 DDS on domain $DOMAIN_ID"

ROS_DOMAIN_IDとROS_LOCALHOST_ONLY

# Isolate robots on the same network by domain ID (0-232)
export ROS_DOMAIN_ID=42

# Lock DDS traffic to localhost only (useful for single-machine development)
export ROS_LOCALHOST_ONLY=1

# For multi-machine setups, ensure ROS_LOCALHOST_ONLY is 0 on all machines
export ROS_LOCALHOST_ONLY=0

# Verify DDS discovery across machines
ros2 daemon stop && ros2 daemon start
ros2 topic list  # Should see topics from both machines

ウォッチドッグとハートビートの監視

systemdウォッチドッグとの統合

サービスユニットにWatchdogSecが設定されている場合、プロセスは定期的にsystemdへ稼働中であることを通知する必要があります。通知が届かないと、systemdがサービスを再起動します。

#!/usr/bin/env python3
# watchdog_node.py
# A ROS2 node that integrates with systemd watchdog via sd_notify.

import os
import socket
import time

import rclpy
from rclpy.node import Node
from diagnostic_msgs.msg import DiagnosticArray, DiagnosticStatus


class WatchdogNode(Node):
    """Notifies systemd that the ROS2 process is alive."""

    def __init__(self):
        super().__init__('watchdog_node')

        # Read the watchdog interval from systemd environment
        watchdog_usec = os.environ.get('WATCHDOG_USEC')
        if watchdog_usec:
            # Notify at half the watchdog interval for safety margin
            interval_sec = int(watchdog_usec) / 1_000_000 / 2.0
        else:
            interval_sec = 10.0
            self.get_logger().warn('WATCHDOG_USEC not set, using 10s interval')

        # Connect to systemd notification socket
        self.notify_socket = os.environ.get('NOTIFY_SOCKET')

        # Signal that startup is complete
        self._sd_notify('READY=1')
        self.get_logger().info(
            f'Watchdog node started, notify interval: {interval_sec:.1f}s')

        # Periodically send watchdog keepalive
        self.create_timer(interval_sec, self._watchdog_tick)

        # Subscribe to system diagnostics to detect failures
        self.diag_sub = self.create_subscription(
            DiagnosticArray, '/diagnostics', self._diag_callback, 10)
        self.system_healthy = True

    def _watchdog_tick(self):
        """Send watchdog keepalive to systemd if system is healthy."""
        if self.system_healthy:
            self._sd_notify('WATCHDOG=1')
        else:
            self.get_logger().error(
                'System unhealthy, withholding watchdog notification')

    def _diag_callback(self, msg):
        """Monitor diagnostics for critical errors."""
        for status in msg.status:
            if status.level == DiagnosticStatus.ERROR:
                self.get_logger().error(f'Critical error: {status.name}: {status.message}')
                self.system_healthy = False
                return
        self.system_healthy = True

    def _sd_notify(self, state):
        """Send notification to systemd."""
        if not self.notify_socket:
            return
        addr = self.notify_socket
        if addr[0] == '@':
            addr = '\0' + addr[1:]
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        try:
            sock.connect(addr)
            sock.sendall(state.encode())
        finally:
            sock.close()


def main():
    rclpy.init()
    node = WatchdogNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

ROS2レベルのハートビート監視ノード

重要なサブシステムからのハートビートトピックをサブスクライブし、システム全体の状態をパブリッシュする監視ノードです。ハートビートを受信できない場合、安全停止を実行します。

#!/usr/bin/env python3
# heartbeat_monitor.py
# Monitors heartbeats from critical nodes and triggers safe stop if any go silent.

import time

import rclpy
from rclpy.node import Node
from rclpy.qos import QoSProfile, ReliabilityPolicy, DurabilityPolicy
from std_msgs.msg import Bool, String
from geometry_msgs.msg import Twist
from diagnostic_msgs.msg import DiagnosticArray, DiagnosticStatus, KeyValue


class HeartbeatMonitor(Node):
    def __init__(self):
        super().__init__('heartbeat_monitor')

        # Declare parameters for monitored nodes and timeout
        self.declare_parameter('monitored_nodes', [
            'motor_controller', 'camera_driver', 'lidar_driver', 'slam'
        ])
        self.declare_parameter('heartbeat_timeout_sec', 5.0)
        self.declare_parameter('check_period_sec', 1.0)

        self.monitored_nodes = self.get_parameter('monitored_nodes').value
        self.timeout = self.get_parameter('heartbeat_timeout_sec').value
        check_period = self.get_parameter('check_period_sec').value

        # Track last heartbeat time for each monitored node
        self.last_heartbeat = {name: time.time() for name in self.monitored_nodes}

        # Subscribe to each node's heartbeat topic
        reliable_qos = QoSProfile(
            reliability=ReliabilityPolicy.RELIABLE,
            durability=DurabilityPolicy.VOLATILE,
            depth=1,
        )
        for node_name in self.monitored_nodes:
            self.create_subscription(
                Bool, f'/{node_name}/heartbeat',
                lambda msg, n=node_name: self._heartbeat_callback(n, msg),
                reliable_qos,
            )

        # Publishers
        self.health_pub = self.create_publisher(
            DiagnosticArray, '/system_health', 10)
        self.estop_pub = self.create_publisher(
            Bool, '/emergency_stop', reliable_qos)
        self.cmd_vel_pub = self.create_publisher(
            Twist, '/cmd_vel', 10)

        # Periodic health check
        self.create_timer(check_period, self._check_health)
        self.get_logger().info(
            f'Monitoring heartbeats for: {self.monitored_nodes}')

    def _heartbeat_callback(self, node_name, msg):
        """Record heartbeat reception time."""
        self.last_heartbeat[node_name] = time.time()

    def _check_health(self):
        """Check all heartbeats and publish diagnostics."""
        now = time.time()
        diag_array = DiagnosticArray()
        diag_array.header.stamp = self.get_clock().now().to_msg()
        all_healthy = True

        for node_name in self.monitored_nodes:
            elapsed = now - self.last_heartbeat[node_name]
            status = DiagnosticStatus()
            status.name = f'heartbeat/{node_name}'

            if elapsed < self.timeout:
                status.level = DiagnosticStatus.OK
                status.message = f'Alive ({elapsed:.1f}s ago)'
            else:
                status.level = DiagnosticStatus.ERROR
                status.message = f'TIMEOUT ({elapsed:.1f}s since last heartbeat)'
                all_healthy = False
                self.get_logger().error(
                    f'Heartbeat timeout for {node_name}: {elapsed:.1f}s')

            status.values = [
                KeyValue(key='elapsed_sec', value=f'{elapsed:.2f}'),
                KeyValue(key='timeout_sec', value=f'{self.timeout:.2f}'),
            ]
            diag_array.status.append(status)

        self.health_pub.publish(diag_array)

        if not all_healthy:
            self._trigger_safe_stop()

    def _trigger_safe_stop(self):
        """Send zero velocity and emergency stop signal."""
        self.get_logger().warn('Triggering safe stop due to heartbeat failure')
        # Publish zero velocity
        self.cmd_vel_pub.publish(Twist())
        # Publish emergency stop
        estop_msg = Bool()
        estop_msg.data = True
        self.estop_pub.publish(estop_msg)


def main():
    rclpy.init()
    node = HeartbeatMonitor()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

ハードウェアウォッチドッグとの統合

多くのロボット搭載コンピューターには、ハードウェアウォッチドッグタイマー（Intel TCO、iTCO_wdtなど）が搭載されています。ソフトウェアがウォッチドッグへのキックに失敗すると、ハードウェアが強制的に再起動します。

# Enable the hardware watchdog in systemd
# /etc/systemd/system.conf
# RuntimeWatchdogSec=30
# RebootWatchdogSec=10min
# ShutdownWatchdogSec=10min

# Or configure per-service in the unit file:
# WatchdogSec=30 triggers systemd to restart the service
# The hardware watchdog (configured via RuntimeWatchdogSec) reboots
# the entire machine if systemd itself becomes unresponsive.

# Verify hardware watchdog is active
sudo cat /sys/class/watchdog/watchdog0/state
# Should output: active

# Check watchdog timeout
sudo cat /sys/class/watchdog/watchdog0/timeout

ロギングとログローテーション

ROS2ログの設定

# Set log level via environment
export RCUTILS_LOGGING_USE_STDOUT=0          # Log to stderr (captured by journald)
export RCUTILS_COLORIZED_OUTPUT=0            # Disable color codes in log files
export RCUTILS_CONSOLE_OUTPUT_FORMAT="[{severity}] [{time}] [{name}]: {message}"

# Set log level at runtime
ros2 run my_pkg my_node --ros-args --log-level debug
ros2 run my_pkg my_node --ros-args --log-level my_node:=debug

# Set log level via parameter (Humble+)
ros2 param set /my_node use_sim_time false
ros2 service call /my_node/set_logger_level rcl_interfaces/srv/SetLoggerLevel \
  "{logger_name: 'my_node', level: 10}"

ROS2サービス向けのjournald設定

# /etc/systemd/journald.conf.d/robot.conf
[Journal]
# Persist logs across reboots
Storage=persistent

# Limit total journal size to 1 GB
SystemMaxUse=1G

# Limit per-file size to 100 MB
SystemMaxFileSize=100M

# Keep logs for 30 days
MaxRetentionSec=30day

# Rate limit: allow bursts during startup
RateLimitIntervalSec=10s
RateLimitBurst=10000

# Forward to syslog for remote logging
ForwardToSyslog=yes

# View logs for a specific robot service
journalctl -u robot-drivers.service -f

# View logs since last boot
journalctl -u robot-bringup.service -b

# View logs with priority filtering (error and above)
journalctl -u robot-bringup.service -p err

# Export logs for analysis
journalctl -u robot-bringup.service --since "2024-01-01" --output=json > logs.json

ROS2ログファイル向けのlogrotate

ROS2はデフォルトで~/.ros/log/にログファイルを書き込みます。または$ROS_LOG_DIRに書き込みます。ローテーションしないと、これらのファイルは無制限に増加します。

# /etc/logrotate.d/ros2
/var/log/ros2/*.log {
    daily
    rotate 7
    compress
    delaycompress
    missingok
    notifempty
    create 0644 robot robot
    maxsize 100M
    dateext
    dateformat -%Y%m%d
    postrotate
        # Notify ROS2 nodes to reopen log files (if using file logging)
        systemctl kill --signal=HUP robot-bringup.service 2>/dev/null || true
    endscript
}

/home/robot/.ros/log/**/*.log {
    daily
    rotate 3
    compress
    missingok
    notifempty
    maxsize 50M
}

本番環境向けの構造化ロギング

import json
import logging
from rclpy.node import Node


class StructuredLogger:
    """Wraps ROS2 logger with structured JSON output for production monitoring."""

    def __init__(self, node: Node):
        self.node = node
        self.logger = node.get_logger()

    def log_event(self, event_type: str, level: str = 'info', **kwargs):
        """Log a structured event with key-value metadata."""
        entry = {
            'event': event_type,
            'node': self.node.get_name(),
            'namespace': self.node.get_namespace(),
            'stamp': self.node.get_clock().now().nanoseconds,
            **kwargs,
        }
        message = json.dumps(entry)
        getattr(self.logger, level)(message)


# Usage in a node:
# self.slog = StructuredLogger(self)
# self.slog.log_event('detection', count=5, latency_ms=12.3)
# self.slog.log_event('motor_fault', level='error', motor_id=2, code=0x0A)

安全なシャットダウン手順

ROS2ノードでのシグナル処理

ROS2ノードは、終了前にアクチュエーターを安全な状態にするため、SIGINTとSIGTERMを処理する必要があります。

#!/usr/bin/env python3
# safe_shutdown_node.py
# Demonstrates graceful shutdown with safe state transitions.

import signal
import sys

import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from std_msgs.msg import Bool


class SafeShutdownNode(Node):
    def __init__(self):
        super().__init__('safe_shutdown_node')

        self.cmd_vel_pub = self.create_publisher(Twist, '/cmd_vel', 10)
        self.brake_pub = self.create_publisher(Bool, '/brakes/engage', 10)

        # Register signal handlers for graceful shutdown
        signal.signal(signal.SIGTERM, self._shutdown_handler)
        signal.signal(signal.SIGINT, self._shutdown_handler)

        self.get_logger().info('Node started with graceful shutdown handler')

    def _shutdown_handler(self, signum, frame):
        """Handle shutdown signals by commanding safe state."""
        sig_name = signal.Signals(signum).name
        self.get_logger().warn(f'Received {sig_name}, initiating safe shutdown...')

        # Step 1: Command zero velocity immediately
        zero_twist = Twist()  # All fields default to 0.0
        for _ in range(5):
            self.cmd_vel_pub.publish(zero_twist)

        # Step 2: Engage brakes
        brake_msg = Bool()
        brake_msg.data = True
        self.brake_pub.publish(brake_msg)

        # Step 3: Wait briefly for commands to be received
        self.get_logger().info('Safe state commanded, shutting down...')

        # Step 4: Clean exit
        self.destroy_node()
        rclpy.shutdown()
        sys.exit(0)


def main():
    rclpy.init()
    node = SafeShutdownNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

systemdの依存関係による順序付きシャットダウン

PartOfとBeforeディレクティブにより、ドライバーより先にアプリケーションレベルのサービスが停止します。これにより、モータードライバーの終了後にナビゲーションノードが速度指令を送信する状況を防げます。

# /etc/systemd/system/robot-application.service
[Unit]
# ...
PartOf=robot-perception.service
Before=robot-perception.service

# When robot-perception stops, robot-application is stopped FIRST
# (Before= reverses the stop order relative to start order)

[Service]
# Use SIGINT for ROS2's signal handler, then SIGTERM, then SIGKILL
KillSignal=SIGINT
TimeoutStopSec=15
FinalKillSignal=SIGTERM
SendSIGKILL=yes

シャットダウン時の安全状態

# Use rclpy context shutdown callback for cleanup
import rclpy
from rclpy.node import Node


class ActuatorNode(Node):
    def __init__(self):
        super().__init__('actuator_node')
        self.cmd_pub = self.create_publisher(Twist, '/cmd_vel', 10)
        # Register a callback that runs during rclpy.shutdown()
        context = self.context
        context.on_shutdown(self._on_shutdown)

    def _on_shutdown(self):
        """Called automatically during rclpy.shutdown()."""
        self.get_logger().info('Shutdown callback: commanding zero velocity')
        self.cmd_pub.publish(Twist())

リモートでの監視とデバッグ

ROS2トピック向けのSSHトンネリング

直接ネットワーク接続を利用できない場合（例：ロボットが携帯電話回線に接続されている場合）、SSH経由でDDSトラフィックを転送します。

#!/bin/bash
# ssh-ros2-tunnel.sh
# Creates an SSH tunnel for ROS2 DDS traffic between local machine and robot.
# Usage: ./ssh-ros2-tunnel.sh [email protected]

set -euo pipefail

ROBOT_HOST="${1:?Usage: $0 robot@host}"
DOMAIN_ID="${ROS_DOMAIN_ID:-0}"
BASE_PORT=$((7400 + 250 * DOMAIN_ID))

echo "Setting up SSH tunnel for ROS2 domain $DOMAIN_ID (ports $BASE_PORT-$((BASE_PORT + 200)))"

# Forward DDS discovery and data ports
ssh -N \
  -L ${BASE_PORT}:localhost:${BASE_PORT} \
  -L $((BASE_PORT + 1)):localhost:$((BASE_PORT + 1)) \
  -L $((BASE_PORT + 10)):localhost:$((BASE_PORT + 10)) \
  -L $((BASE_PORT + 11)):localhost:$((BASE_PORT + 11)) \
  "$ROBOT_HOST" &

SSH_PID=$!
echo "SSH tunnel PID: $SSH_PID"

# Set environment for local ROS2 to use localhost-only discovery
export ROS_LOCALHOST_ONLY=1
echo "Run: export ROS_LOCALHOST_ONLY=1"
echo "Then use ros2 topic list, ros2 topic echo, etc."
echo "Press Ctrl+C to close tunnel."

wait $SSH_PID

リモートからのjournalctlとサービス管理

# View live robot logs remotely
ssh [email protected] 'journalctl -u robot-bringup.service -f'

# Check service status
ssh [email protected] 'systemctl status robot-drivers.service robot-perception.service'

# Restart a single layer without rebooting
ssh [email protected] 'sudo systemctl restart robot-perception.service'

# View boot-time service ordering
ssh [email protected] 'systemd-analyze blame | head -20'

# Check for failed services
ssh [email protected] 'systemctl --failed'

# Stream structured logs as JSON
ssh [email protected] 'journalctl -u robot-bringup.service -o json --follow'

SSH経由での更新のデプロイ

#!/bin/bash
# deploy-to-robot.sh
# Build locally, copy to robot, and restart services.
# Usage: ./deploy-to-robot.sh [email protected]

set -euo pipefail

ROBOT_HOST="${1:?Usage: $0 robot@host}"
WORKSPACE="/home/robot/ros2_ws"

echo "=== Building workspace locally ==="
colcon build --cmake-args -DCMAKE_BUILD_TYPE=Release --packages-select my_robot_bringup

echo "=== Syncing to robot ==="
rsync -avz --delete \
  --exclude='build/' --exclude='log/' \
  src/ "${ROBOT_HOST}:${WORKSPACE}/src/"

echo "=== Building on robot ==="
ssh "$ROBOT_HOST" "cd ${WORKSPACE} && \
  source /opt/ros/\${ROS_DISTRO}/setup.bash && \
  colcon build --cmake-args -DCMAKE_BUILD_TYPE=Release"

echo "=== Restarting robot services ==="
ssh "$ROBOT_HOST" 'sudo systemctl restart robot-bringup.target'

echo "=== Checking service status ==="
ssh "$ROBOT_HOST" 'sleep 3 && systemctl status robot-bringup.target --no-pager'

echo "Deploy complete."

ロボットの起動におけるアンチパターン

1. systemd向けにsetup.bashを.bashrcで読み込む

問題： systemdサービスは~/.bashrcや~/.profileを読み込みません。そこで設定した環境変数はサービスから参照できないため、ROS2コマンドが「command not found」やパッケージが見つからないというエラーで失敗します。

# BAD: Relying on .bashrc for systemd services
# ~/.bashrc
source /opt/ros/humble/setup.bash  # systemd will never see this

# GOOD: Use EnvironmentFile in the service unit and source explicitly in ExecStart
# /etc/robot/ros2.env
ROS_DISTRO=humble
RMW_IMPLEMENTATION=rmw_cyclonedds_cpp

# In the service unit:
# EnvironmentFile=/etc/robot/ros2.env
# ExecStart=/bin/bash -c 'source /opt/ros/${ROS_DISTRO}/setup.bash && ...'

2. 起動順序を設定しない

問題： すべてのROS2ノードを同時に起動すると、競合状態が発生します。ナビゲーションノードが、ドライバーによるサービスの公開前にそのサービスを呼び出そうとする場合があり、起動が不安定になります。

解決策： systemdユニットでAfter=とRequires=を使用するか、ライフサイクルマネージャーを使用して順序付きの遷移を徹底します。

# BAD: All services start in parallel with no ordering
[Unit]
Description=Robot Navigation
# No After= or Requires= directives

# GOOD: Explicit dependency chain
[Unit]
Description=Robot Navigation
After=robot-perception.service
Requires=robot-perception.service

3. レート制限なしでRestart=alwaysを使用する

問題： 起動時にクラッシュするサービス（設定ファイルがない、ハードウェアが切断されているなど）は、短い間隔で再起動を繰り返し、CPUを消費してジャーナルを大量のログで埋め尽くす可能性があります。

解決策： StartLimitIntervalSecとStartLimitBurstを使用して、再起動の試行回数を制限します。

# BAD: Infinite restart loop
[Service]
Restart=always
RestartSec=1

# GOOD: Rate-limited restarts with failure notification
[Service]
Restart=on-failure
RestartSec=5
StartLimitIntervalSec=120
StartLimitBurst=5

4. network-online.target ではなく network.target に依存する

問題： network.target はネットワーク設定が開始するとすぐに到達しますが、実際に接続が確立した時点ではありません。ネットワークインターフェースにまだ IP アドレスが割り当てられていないため、DDS の検出に失敗します。

修正： network-online.target を使用し、systemd-networkd-wait-online.service または NetworkManager-wait-online.service が有効になっていることを確認します。

# BAD: network.target does not guarantee connectivity
[Unit]
After=network.target

# GOOD: Wait for actual network connectivity
[Unit]
After=network-online.target
Wants=network-online.target

5. ログローテーションがない

問題： ~/.ros/log/ 内の ROS2 ログファイルと journald のエントリーが無制限に増加し、最終的にストレージ容量の限られた組み込みシステムのディスクを使い切ってしまいます。

修正： ROS2 ログファイル用に logrotate を設定し、journald のサイズ制限を設定します。

# BAD: No log management
# Logs in ~/.ros/log/ grow forever, disk fills up after weeks of operation

# GOOD: logrotate config + journald limits
# /etc/logrotate.d/ros2 (see Logging section above)
# /etc/systemd/journald.conf: SystemMaxUse=1G

6. デバイスパスをハードコードする（/dev/ttyUSB0）

問題： /dev/ttyUSB0 は、列挙順によって任意の USB シリアルデバイスに割り当てられることがあります。再起動後に、IMU が /dev/ttyUSB1、モーターコントローラーが /dev/ttyUSB0 となり、割り当てが逆転する場合があります。

修正： udev ルールを使用して、安定したシンボリックリンクを作成します。

# BAD: Hardcoded device path in ROS2 params
serial_port: "/dev/ttyUSB0"  # Which device is this? It changes on reboot!

# GOOD: Stable symlink via udev rule
serial_port: "/dev/robot/imu"  # Always points to the correct device

7. スタック全体を root として実行する

問題： ROS2 を root として実行するとセキュリティリスクが生じ、rosbag2、ログファイル、パラメーターの永続化で権限の問題が発生することがあります。ROS2 ノードのバグによって、オペレーティングシステムが損傷する可能性もあります。

修正： 専用の robot ユーザーを作成し、udev の GROUP および MODE ルールを使って、必要なデバイス権限のみを付与します。

# BAD: Running as root
# ExecStart=/bin/bash -c 'source /opt/ros/humble/setup.bash && ros2 launch ...'
# (runs as root because no User= is specified)

# GOOD: Dedicated user with minimal privileges
# Create robot user
sudo useradd -r -m -s /bin/bash robot
sudo usermod -aG dialout,video,plugdev robot

# In the service unit:
# User=robot
# Group=robot

# udev rules grant device access to the robot user's groups:
# MODE="0666", GROUP="dialout"

8. グレースフルシャットダウンハンドラーがない

問題： systemd が ROS2 ノードの停止時に SIGTERM または SIGINT を送信すると、ノードはゼロ速度を指令したりブレーキを作動させたりせずに直ちに終了します。その結果、ロボットが惰性で進んだり、最後に指令された速度で動き続けたりする可能性があります。

修正： シグナルハンドラーを登録するか、rclpy のシャットダウンコールバックを使用して安全な状態を指令します。

# BAD: No shutdown handling, node just exits
def main():
    rclpy.init()
    node = MotorControlNode()
    rclpy.spin(node)
    # Robot is still moving with last commanded velocity!

# GOOD: Shutdown handler commands safe state
def main():
    rclpy.init()
    node = MotorControlNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.command_zero_velocity()
        node.engage_brakes()
        node.destroy_node()
        rclpy.shutdown()

ロボット起動チェックリスト

1. すべての USB デバイス（カメラ、LiDAR、シリアルアダプター）について、/dev/robot/ 配下に安定したシンボリックリンクを作成する udev ルールを記述してテスト済み
2. 各レイヤー（ドライバー、認識、アプリケーション）用の systemd サービスユニットを作成済み。After=／Requires= の順序も正しく設定
3. ROS_DISTRO、RMW_IMPLEMENTATION、ROS_DOMAIN_ID、CYCLONEDDS_URI を設定した ROS2 環境ファイル（/etc/robot/ros2.env）を構成済み
4. 複数マシンでの検出に備え、明示的なピアリストを設定した CycloneDDS または FastDDS の XML を構成済み
5. シミュレーション／実機やロボットのバリエーションに応じた条件付き引数を使い、起動ファイルを階層化して組み合わせ可能に設定済み
6. ドライバーの起動前にデバイスの存在を確認する ExecStartPre 用の ヘルスチェック スクリプトを作成済み
7. ウォッチドッグ連携を構成済み：サービスユニットの WatchdogSec と ROS2 プロセス内の sd_notify(WATCHDOG=1)
8. ノードの障害を検出して安全停止をトリガーする ハートビート監視ノードを配置済み
9. すべてのアクチュエーターノードに グレースフルシャットダウンハンドラーを登録済み（SIGINT／SIGTERM 時にゼロ速度を指令し、ブレーキを作動）
10. $ROS_LOG_DIR 用の logrotate と journald の SystemMaxUse 制限で ログローテーションを構成済み
11. 再起動ループを防ぐため、StartLimitIntervalSec と StartLimitBurst で 再起動ポリシーにレート制限を設定済み
12. 暴走したノードによるシステムリソースの占有を防ぐため、MemoryMax と CPUQuota で リソース制限を設定済み
13. 固定 IP、DDS ポートのルール、適切に設定した ROS_LOCALHOST_ONLY を含めて ネットワークとファイアウォールを構成済み
14. 電源オフの状態から自律動作に至るまでの 完全な起動テストを実施済み。サービスの起動順序と、障害をシミュレーションした場合の復旧を確認