自動切り替えの仕組み:障害検知と選択操作を分ける

ノードの自動切り替えは、プロキシグループの種類とコアが提供する外部コントローラーAPIを使って実装できます。基本の流れは、現在選択中のノードを取得し、決めた接続先への応答を確認し、失敗が一定回数続いた場合に別のノードをテストして、正常な候補をグループに設定する、というものです。画面上の選択を監視するだけではなく、実際にプロキシ経由の疎通を確認することが重要です。

APIは、Clash系コアの外部コントローラーが有効になっている場合に利用できます。Clash Verge、Clash Verge Revなどのクライアントは、コアの起動や設定を管理する画面であり、APIの機能やURLの細部は実際に動作しているコアに依存します。オリジナル版Clash、Clash Premium、mihomoでは利用できる項目や対応状況に差があるため、まずコア名とバージョンを確認してください。

切り替え先として想定するのは、select型のプロキシグループです。url-testやfallbackなどの自動選択グループは、コア自身が遅延測定や優先順位に基づいて選択するため、外部スクリプトから手動選択する対象とは性質が異なります。設定にあるグループの型と、APIが返すグループ情報を確認してから実装してください。

external-controller: 127.0.0.1:9090
secret: "ここに十分長いランダムな文字列"

上記は設定の一部です。すでに同じ項目がある場合、重複したキーを追加せず、現在の設定を確認して変更してください。127.0.0.1はコントローラーを同じ端末からだけ利用する指定です。ポート番号の9090は例であり、実際の値は使用中のクライアントと設定に合わせます。ローカル以外から管理する必要がある場合も、ネットワークへ無条件に公開せず、接続元の制限やファイアウォールを検討してください。

APIでグループ、現在のノード、疎通状態を確認する

Clash互換のREST APIでは、/proxiesからプロキシやグループの一覧を取得できます。グループ名を指定した/proxies/{名前}では、グループの型、候補一覧、現在の選択先などを確認できます。URLのパスにグループ名やノード名を入れるときは、空白や記号を含む名前をURLエンコードしてください。日本語名を使う場合も、文字列をそのまま連結せず、必ずURLエンコードする実装にします。

疎通確認には、ノードの遅延テストAPIを利用できます。たとえば/proxies/{ノード名}/delayに、テストURLとタイムアウトをクエリとして渡すと、指定したノードを経由した遅延測定を要求できます。応答に遅延値が返る場合はテスト成功の目安になりますが、失敗時にはHTTPエラーになることもあります。HTTPステータスだけでなく、応答時間やエラー内容も記録してください。

GET http://127.0.0.1:9090/proxies
GET http://127.0.0.1:9090/proxies/自動切替グループ
GET http://127.0.0.1:9090/proxies/ノード名/delay?url=https%3A%2F%2Fwww.gstatic.com%2Fgenerate_204&timeout=5000
PUT http://127.0.0.1:9090/proxies/自動切替グループ
Content-Type: application/json

{"name":"切り替え先のノード名"}

最後のPUTは、選択型グループの現在のノードを変更する操作の例です。コアやAPIバージョンによって、返されるフィールドや細かな挙動が異なる場合があります。実運用前に、まずGETで対象グループが存在することと型がSelectorであることを確認し、手動のPUTで選択が変わるかテストしてください。APIの成功応答だけで終わらせず、もう一度グループ情報を取得し、nowが意図したノード名になったことも確認します。

監視先は運用目的に合わせて選ぶ

遅延テストのURLは、利用する地域やネットワークで安定して到達できるものを選びます。特定のWebサイトだけをテスト先にすると、そのサイト側の一時障害や地域制限をノード障害と誤認することがあります。逆に、テスト先が応答しやすくても、実際に必要なサービスへ接続できるとは限りません。重要な運用では、用途に合った複数の確認方法を組み合わせ、特定のサービスに依存しすぎないようにします。

  • テストURL:HTTPSで安定して応答するURLを指定します。URLはコアから各候補ノード経由で到達可能である必要があります。
  • タイムアウト:例では5000ミリ秒を使います。短すぎる値は一時的な混雑で誤判定し、長すぎる値は障害時の切り替えを遅らせます。
  • 許容遅延:応答があっても、運用上使えないほど遅ければ候補から外す、といった基準を定めます。
  • 観測回数:単発の失敗で切り替えず、複数回連続した失敗を条件にします。候補ノードについても同じ基準で判定します。

監視スクリプトを作成して段階的に切り替える

以下はPython標準ライブラリだけを使う最小構成の例です。現在のグループ名と候補ノード名を、実際の設定にある名前へ置き換えてください。候補リストは、グループAPIから取得した全ノードを無条件に使うのではなく、接続先として認めたノードだけを登録する方法が安全です。サブスクリプションの更新後に名前が変わることがあるため、候補一覧とAPIの応答を定期的に照合してください。

import json
import os
import time
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen

API = os.environ.get("CLASH_API", "http://127.0.0.1:9090")
SECRET = os.environ["CLASH_SECRET"]
GROUP = "自動切替グループ"
NODES = ["ノードA", "ノードB", "ノードC"]

TEST_URL = "https://www.gstatic.com/generate_204"
TIMEOUT_MS = 5000
MAX_DELAY_MS = 2500
FAIL_LIMIT = 3
POLL_SECONDS = 15
SWITCH_COOLDOWN = 60


def api(method, path, payload=None):
    body = None if payload is None else json.dumps(payload).encode("utf-8")
    request = Request(
        API.rstrip("/") + path,
        data=body,
        method=method,
        headers={
            "Authorization": "Bearer " + SECRET,
            "Content-Type": "application/json",
        },
    )
    with urlopen(request, timeout=8) as response:
        data = response.read()
        return json.loads(data) if data else {}


def encoded_name(name):
    return quote(name, safe="")


def get_current_node():
    group = api("GET", "/proxies/" + encoded_name(GROUP))
    if group.get("type", "").lower() not in ("selector", "select"):
        raise RuntimeError("対象グループが選択型ではありません")
    return group.get("now")


def probe(node):
    query = urlencode({"url": TEST_URL, "timeout": str(TIMEOUT_MS)})
    result = api(
        "GET",
        "/proxies/" + encoded_name(node) + "/delay?" + query,
    )
    delay = result.get("delay")
    return isinstance(delay, (int, float)) and 0 < delay <= MAX_DELAY_MS


def switch_to(node):
    api(
        "PUT",
        "/proxies/" + encoded_name(GROUP),
        {"name": node},
    )
    current = get_current_node()
    if current != node:
        raise RuntimeError("切り替え後の選択先を確認できません")
    print(time.strftime("%Y-%m-%d %H:%M:%S"), "switched to", node)


failures = 0
next_switch_time = 0

while True:
    try:
        current = get_current_node()
        if current and probe(current):
            failures = 0
            print(time.strftime("%Y-%m-%d %H:%M:%S"), "healthy", current)
        else:
            failures += 1
            print(time.strftime("%Y-%m-%d %H:%M:%S"),
                  "unhealthy", current, "count", failures)

        if failures >= FAIL_LIMIT and time.time() >= next_switch_time:
            selected = None
            for node in NODES:
                if node == current:
                    continue
                try:
                    if probe(node):
                        selected = node
                        break
                except (HTTPError, URLError, TimeoutError, ValueError) as error:
                    print("candidate check failed:", node, type(error).__name__)

            if selected:
                switch_to(selected)
                failures = 0
                next_switch_time = time.time() + SWITCH_COOLDOWN
            else:
                print("no healthy candidate; keeping current selection")
                next_switch_time = time.time() + SWITCH_COOLDOWN

    except (HTTPError, URLError, TimeoutError, ValueError, RuntimeError) as error:
        failures += 1
        print(time.strftime("%Y-%m-%d %H:%M:%S"),
              "monitor error:", type(error).__name__)

    time.sleep(POLL_SECONDS)

このコードは仕組みを説明するための例です。Pythonでは比較演算子の<と>を、実ファイルでは通常の<と>として記述します。HTML表示からコードをコピーする場合、画面上で文字列がHTMLエスケープされて見えることがあります。実行前にPythonファイルとして保存し、構文チェックを行ってください。たとえば次のコマンドで構文を確認できます。

python -m py_compile monitor.py

CLASH_SECRET環境変数を設定してから実行します。設定されていない場合、例では起動時にエラーとなり、認証なしでAPIへアクセスすることはありません。スクリプトをサービスとして常駐させる場合は、OSのサービス管理機能で起動・再起動の方針を設定し、出力先のログを管理します。監視プロセス自体が停止している状態を検知する方法も別途用意してください。

  • 失敗の連続回数:FAIL_LIMIT = 3なら、15秒間隔で連続失敗した場合、概ね3回分の確認後に候補テストを始めます。実際の判定時間には各API要求のタイムアウトも加わります。
  • 切り替え後の待機:SWITCH_COOLDOWNは頻繁な再試行を抑えます。ノードを交互に選び続ける状態を防ぐため、切り替え先の再評価と元のノードへの復帰条件も運用方針として決めてください。
  • 候補の順序:リストの先頭から検査し、最初に基準を満たしたノードへ切り替えます。優先順位を変えたい場合はリストの順序を変更します。
  • 一時的なAPI障害:コア再起動やコントローラーの応答遅延でも監視処理は失敗します。APIに接続できないことと、現在のノードが故障したことをログ上で区別します。

API保護とログ調査:誤切り替えを減らす

APIを使うときは、到達性と認証を別々に確認します。まずコアが起動しているか、設定したポートで待ち受けているかを調べます。次に、ループバックアドレスへ認証付きのGETを送り、グループ一覧が返るか確認します。認証エラーならシークレットの読み込みとヘッダー形式を、接続拒否ならコントローラーの待ち受けアドレス、ポート、クライアントが使用する実際の設定を調べてください。

シークレットが設定にあるのにAPI要求が認証エラーになる場合は、違うコアや違うポートに接続していないか確認します。複数のClash系クライアントをインストールしている環境では、想定外のコアが同じポートを使っていることがあります。設定ファイルを編集しただけで実行中のコアに反映されたと思い込まず、クライアントの適用済み設定と起動ログを確認してください。

ノードの遅延テストに失敗しても、直ちにノード障害とは限りません。テストURLへの経路、DNSの状態、コアが使うタイムアウト、ネットワーク側の一時的な制限を確認します。接続ログにdial tcpがある場合は、表示された宛先がノードサーバーかテスト先かを見分けてください。複数の候補が同時に失敗するなら、ローカルネットワーク、コア、テスト先の問題も考えられます。

ログには時刻、対象グループ、ノード名、遅延値、失敗回数、HTTPステータス、実行した切り替え結果を残します。一方で、APIシークレット、サブスクリプションURLのトークン、ノードの認証情報は記録しません。デバッグログを共有する場合も、機密情報を削除し、時刻とエラーの種類、対象ポートなど原因調査に必要な情報は残してください。

本番運用前のチェック項目

  1. 対象グループが選択型で、候補ノードがAPIのグループ情報に存在することを確認します。
  2. スクリプトを動かす端末から、認証付きGET、遅延テスト、手動のノード選択を個別に試します。
  3. 失敗回数、最大遅延、監視間隔、タイムアウト、切り替え後の待機時間を、ネットワークの実測結果に合わせて調整します。
  4. 意図的に誤ったテスト先を使うなど、運用に影響しない範囲で失敗時のログと切り替え動作を確認します。
  5. コアやクライアントを再起動した後、監視プロセスの再接続、グループ選択の維持、APIキーの保護を確認します。

自動切り替えは、通信を確実に復旧させる仕組みではなく、あらかじめ決めた条件に基づいてノード選択を補助するものです。利用できる候補がない場合、APIが停止している場合、すべての候補が同じ上流障害の影響を受けている場合には切り替えられません。手動で戻す手順と監視を停止する方法も用意し、異常な選択を検知したときに人が介入できる状態を保ってください。

よくある質問

どのプロキシグループでもAPIから切り替えられますか?

いいえ。外部スクリプトから選択先を指定する用途には、一般に選択型グループを使います。url-testやfallbackはコアが自動選択するグループです。対象の型は設定とAPI応答の両方で確認し、自動選択型の状態を無理に手動管理しないでください。

失敗が何回続いたら切り替えるべきですか?

一律の正解はありません。短い監視間隔で1回の失敗だけを条件にすると、一時的な混雑で切り替わる可能性があります。まずは15秒程度の監視間隔と複数回の連続失敗を試し、実測した遅延と切り替え許容時間に合わせて調整します。

遅延テストが成功すれば、すべての通信が正常ですか?

いいえ。成功は指定したURLに対するテストが通ったことを示します。ほかのサイト、UDP通信、DNS、アプリ固有の認証やプロトコルまで正常とは限りません。必要な用途に合った確認先を選び、必要なら利用者側の実通信も別に監視してください。

外部コントローラーにシークレットは必須ですか?

APIを運用する場合は設定し、APIへのアクセスを認証で保護してください。待ち受けアドレスも必要以上に広げず、可能ならループバックに限定します。シークレットをスクリプトへ直書きしたり、ログや共有用の診断情報に残したりしないことが重要です。