はじめに

Cloud Run Job はリスニングポートを持たず、外部からのインバウンド接続を受け付けることができません。 この制約はバッチ処理や CI ジョブといったワンショットなタスクでは問題になりませんが、Job 内でブラウザのリモート表示や映像ストリーミングのような双方向通信が必要な場面では課題が生じます。

直感的な解決策として Job の中でも HTTP サーバを立ち上げるという発想が浮かぶかもしれませんが、Cloud Run Job のコンテナには外部からアクセスできるエンドポイント（URL）が割り当てられないため、ポートをバインドしてもブラウザからそのポートに到達できません。

そこで Job 側から専用の中継サーバへアウトバウンド接続を確立させ、中継サーバがブラウザとの通信を橋渡しするアーキテクチャ（以後 Relay パターンと呼ぶ）を取ります。（参考）

今回のブログでは、Job 内で動作するブラウザの画面をリモートストリーミングするユースケース（一時的に仮想ブラウザを起動して完了後は破棄したい）を例に、Relay パターンで Cloud Run Job の WebSocket 双方向通信を実現する方法について紹介したいと思います。

WebSocket とは

まず、WebSocket とは何なのかという前提を説明しておきます。 WebSocket は RFC 6455 で定義されたプロトコルで、HTTP のハンドシェイクを出発点としてコネクションを確立した後、全二重（Full-Duplex）の TCP 通信に切り替えます。 一度接続が確立されると、クライアントとサーバのどちら側からでも任意のタイミングでメッセージを送り合える状態になります。

WebSocket の接続確立は HTTP/1.1 の アップグレード機構 を利用します。

サーバが 101 Switching Protocols を返すと、TCP コネクションは HTTP から WebSocket プロトコルに切り替わり、以後は HTTP のオーバーヘッドなしにフレーム単位でメッセージを送受信できます。

なぜ WebSocket が必要か

HTTP のリクエスト / レスポンス型（クライアントサーバモデル）の通信では、クライアントからリクエストがなければサーバからデータを送ることはできません。 故にリモートデスクトップの画面更新や映像ストリーミングのように、サーバ側で発生したイベントをリアルタイムにクライアントへ届ける 場合、WebSocket は自然な選択になるでしょう。

フレームタイプ

WebSocket は HTTP とは異なり、フレーム単位 でデータを送受信します。 フレームにはいくつかのタイプがあり、用途によって使い分けます。

Relay パターンでは、テキストフレームで制御メッセージ（{"type":"stop"} 等）をやり取りし、バイナリフレームで映像・入力データを中継するという使い分けができます。 フレームタイプを中継サーバ（以下 Hub と呼ぶ）が変換することなく透過的にリレーできるため、Job 側とブラウザ側の実装がフレームタイプの詳細を意識せず通信できます。

ステートフル接続

HTTP と WebSocket の大きな違いは、接続がステートフルかどうかです。 HTTP はそれ自体で完結するため、サーバはリクエスト間の状態を持たずに済みます。

一方、WebSocket は TCP コネクションが確立されている間、その接続状態をプロセスのメモリ上に保持し続ける 必要があります。 これは中継サーバ（Hub）を設計する上で重要な制約になります。

例えば、以下の例ではブラウザと Job のそれぞれが同じ Hub インスタンスに接続していなければペアリングは成立しません。

このため Hub は以下のいずれかの構成を取る必要があります。

小規模であれば単一インスタンス固定が最もシンプルですが、セッション数が多くなる場合は、Sticky session（Session affinity）やセッション情報自体を外部で管理するような構成を検討します。

Cloud Run Job の制約

Cloud Run Job と Service の違い

Cloud Run には「Service」と「Job」の 2 種類の実行モデルがあります。

インバウンド接続を受け付けられない

Cloud Run Job のコンテナは、アウトバウンドのネットワーク接続（例：外部 API の呼び出し、Cloud Storage へのアップロード等）は行えますが、外部からの接続を待ち受けるリスニングポートを持たないためインバウンド接続は受け付けられません。

Unlike a Cloud Run service, which listens for and serves requests, a Cloud Run job only runs its tasks and exits when finished. A job does not listen for or serve requests.

Because jobs shouldn't serve requests, the container shouldn't listen on a port or start a web server.

Cloud Run Service であれば受信リクエストを処理できますが、Job はその設計になっていないわけです。 なので本来であれば、WebSocket のような双方向通信が必要になる場合は前者（Cloud Run Service）を利用するのが一般的 です。

どのような場面で問題になるか

この制約が顕在化するのは、Job がリアルタイムに画面やデータをブラウザへ中継する必要がある場面です。 代表的なユースケースとして以下が挙げられます。

• VNC / リモートデスクトップ：Job 内で動作するブラウザ（headful Chrome 等）の画面を noVNC 等でブラウザに表示したい
• ターミナルストリーミング：Job 内のシェル出力をブラウザのターミナルエミュレータにリアルタイムで流したい
• 進捗ストリーミング：長時間バッチの進捗をブラウザへリアルタイムに通知したい

いずれも「Job → ブラウザ方向」だけでなく、「ブラウザ → Job 方向（キーボード・マウス入力等）」も必要なため、WebSocket の全二重通信が求められます。

HTTP ポーリングでは解決できない理由

これだけ聞くと、WebSocket ではなく HTTP ポーリングで代替できないかという疑問が出てきそうですが、VNC やターミナルのような用途では以下の理由で現実的ではありません。

通信のオーバーヘッド

映像フレームのデータ量は 1 フレームあたり数十〜数百 KB になります。 30fps でポーリングしようとすると 1 秒間に 30 回のリクエストを送り続けることになり、HTTP のオーバーヘッド（ヘッダのやり取り等）が積み重なって帯域を圧迫します。 WebSocket であれば接続確立後のオーバーヘッドはフレームヘッダの数バイトに留まります。

双方向通信の困難性

キーボード・マウス入力のような「ブラウザ → Job」方向の送信はポーリングでは実現できません。 Server-Sent Events（SSE）で映像を流しながら、操作入力は別途 HTTP POST で送るという分離も考えられますが、2 種類のコネクションを扱う複雑さが生じます。 WebSocket は単一の接続で双方向を実現できるため、実装もシンプルになります。

Relay パターン

Cloud Run Job のインバウンド制約と WebSocket のステートフル性という 2 つの制約を踏まえ、ここからは Relay パターンを使った具体的な解決策を見ていきます。

以下は、Job 側から Hub へアウトバウンド接続を確立させ、Hub がブラウザとの通信を橋渡しするアーキテクチャの一例です。

API サーバはセッション ID を生成して Cloud Run Job を起動し、セッション ID をブラウザへ返します。 ブラウザと Job はそれぞれ Hub に WebSocket で接続して Hub が 2 本の接続を中継することで全二重通信が成立します。 Job はアウトバウンド接続しか行わないため、インバウンドを受け付けられない Cloud Run Job の制約を回避できます。

インバウンド接続を受け付けられないなら、Job 側からアウトバウンドで Hub に接続させればよいというのが Relay パターンの発想です。

同様の発想は Microsoft Azure の Azure Relay というサービスでも採用されています。 オンプレミスのサービスがファイアウォール越しにアウトバウンドでリレーサーバへ接続し、外部クライアントはそのリレー経由で通信する構成で、いわゆる NAPT 越えの問題 を解決します。

NAPT 越え問題については こちらのブログ で紹介しています。

Hub はリクエストを常時受け付ける常駐サーバとして起動しておく（例：Cloud Run Service）ため、ブラウザからのインバウンド接続も、Job からのアウトバウンド接続も受け付けられます。 Job は起動後に自分から Hub へ接続し、Hub はブラウザとの接続が揃った時点で双方のメッセージを中継し始めます。

セッション単位で 2 本の接続を管理する

Hub には複数のセッションが同時に存在するため、どの Job とどのブラウザが対応するかを識別する仕組みが必要です。 セッション ID をエンドポイントのパスに含め、Hub がセッション ID をキーとして 2 本の接続をペアリングします。

このペアリング方式は、前述した Azure Relay では Hybrid Connection と呼ばれており、この他 ngrok や frp 等でも取り入れている逆方向トンネル確立時の慣用パターンです。

セッション ID を Job へ渡す手段はいくつか考えられますが、例えば、Cloud Run Job であれば起動時の環境変数として渡す方法が簡便ですが、他にも起動引数・設定ファイル・起動前に書き込んだストレージ経由等、Job の実行環境に応じた方法を選択できます。 いずれの方法でも、Job は起動直後にセッション ID を読み取り、対応する Hub のエンドポイントへアウトバウンド接続するという流れは同じです。

Job の起動フロー

ユーザがセッションを開始する際のフロー全体は以下のようになります。

Job の起動処理（Cloud Run Job の submit / execute API 呼び出し）は非同期で完了するため、ブラウザと Job の接続は不定の順序で Hub に届きます。 Hub がどちらの順番でも正しくペアリングできるよう、先着でセッションを作成、後着が join する という設計が肝になります。

Job とブラウザのどちらが先に Hub へ到達するかは不定ですが、一般に Cloud Run Job の起動には数秒〜十数秒かかるため、必然的にブラウザが先に Hub に接続して待機するケースが多くなります。

Job が先に接続した場合

ブラウザが先に接続した場合

Hub の実装ではセッション作成と接続の登録をアトミックに行い（例：mutex ロック機構）、どちらのケースでも同じセッションオブジェクトにアクセスできるようにします。 ブラウザが接続したとき Job 接続が既に登録済みであれば即座に browser_connected を送信し、未登録であれば何もしません。（Job が後から接続したときに Hub 側で送る）

メッセージ制御プロトコル

WebSocket はフレームの送受信 channel を提供しますが、「ブラウザが接続した」「処理を終了してよい」といったアプリケーションレベルのライフサイクルイベントは定義していません。 Hub・Job・ブラウザ間でこれらの状態を伝えるためには、アプリケーション側で制御メッセージを設計する必要があります。

制御メッセージは JSON で表現し、type フィールドで種別を区別します。

例えば、Socket.IO や WAMP といった WebSocket アプリケーションプロトコルでは同様の設計が採用されています。 Socket.IO はパケット先頭の整数コードで種別を識別し（例：2["event","data"] → EVENT=2）、WAMP は JSON 配列の第 1 要素に整数コードを置く形式で（例：[48, ...] → CALL）メッセージ種別を表現します。 いずれも、種別を表すフィールドを先頭に置きます。

メッセージの具体的な種別や意味はアプリケーションの要件に応じて自由に定義できます。 以下は Relay パターンの最小構成の一例です。

Hub はバイナリフレームを内容に関わらずそのまま転送します。 テキストフレームは type フィールドを確認し、制御メッセージと一致すれば Hub 自身が処理し、それ以外は同様に転送します。 制御メッセージをデータと同じ channel で送ることで、別途 WebSocket 接続を設ける必要がなく、実装をシンプルに保てます。

セッションのクリーンアップ

セッションは Job・ブラウザの両方が切断した時点で削除する必要があります。 どちらか一方だけが切断した状態では、もう一方の接続がまだ生きている可能性があるため、セッションオブジェクト自体は保持し続けます。

ブラウザが切断した際は、Job に対して {"type":"stop"} を送ることで、Job が処理を完了して終了するトリガーにします。 Job 側はこのメッセージを受け取ると処理を完了し、{"type":"stopped"} を返して切断します。 その後の Job 切断で「job=nil, browser=nil」になった時点でセッションを削除します。

実装で意識すること

並行書き込みの排他制御

Go で WebSocket を扱う際のデファクトスタンダードとなっている gorilla/websocket のコネクションは、同時に書き込める Goroutine は 1 つまで という制約があります。

Connections support one concurrent reader and one concurrent writer.

gorilla/websocket では upgrader.Upgrade() を呼ぶことで HTTP コネクションを WebSocket にアップグレードし、*websocket.Conn を取得します。

Hub の実装では、受信したフレームを転送する Goroutine と制御メッセージを送信する Goroutine の 2 つが同一の conn へ書き込む可能性があります。 そのため WriteMessage の呼び出しを sync.Mutex で保護するラッパを定義します。

conn.WriteMessage を直接呼ぶ代わりに c.write() 経由にすることで、どの Goroutine から呼んでも自動的に排他制御が行われます。

セッションマップへの並行アクセスも同様に sync.Mutex で保護する必要があります。 ただし、ロックを保持したまま I/O に入ると別セッションの操作が全てブロックされるため、ロックはセッションの取得・更新に限定し、ポインタをローカル変数にコピーしてからロックを解放してから処理を行うようにします。

Goroutine のライフサイクル

Hub の各エンドポイントの Handler は WebSocket ハンドシェイク後もその Goroutine を使い、フレームの受信ループを実行します。 Handler が return すると Goroutine が終了し、接続もクローズされます。

受信ループ内ではフレームタイプで処理を振り分けます。 テキストフレームは制御メッセージとして解析し、バイナリフレームはそのままブラウザへ転送します。

Job 側とブラウザ側それぞれに 1 つずつ Goroutine が割り当てられ、それぞれが独立して動きます。 一方が切断したら他方も切断するという動作は、Goroutine 終了時の defer 内でもう一方へ制御メッセージを送ることで実現します。

例えば、ブラウザが切断した場合、ブラウザ側 Goroutine の defer で Job へ {"type":"stop"} を送信します。

停止シグナルのバッファ

外部から Job に停止を命令する場合、Hub を経由して {"type":"stop"} を送り、Job からの {"type":"stopped"} を受け取るまで待ち受けます。

ここで、API リクエストを処理する Goroutine（SendStop）と、Job トンネルの受信ループを実行する Goroutine（{"type":"stopped"} を受け取る側）は別々に動作します。 前者には stoppedCh: make(chan int, 1) のようにバッファチャネルを準備しておくことで、SendStop がタイムアウトで抜けた後に Job が応答してきても、Job トンネル側の送信をブロックしないようにします。

また、Hub プロセス自体が SIGTERM を受けてシャットダウンする場合は、管理下の全 WebSocket 接続に対して Close Frame を送信する実装が必要になります。 WebSocket は net/http の Graceful Shutdown の管理外（Hijacked Connection）となるため、アプリケーション側で明示的にクローズ処理を実装しなければなりません。 Hub プロセスの Graceful Shutdown については こちらのブログ で紹介しています。

まとめ

今回のブログでは、Cloud Run Job のインバウンド接続制約を Relay パターンで回避し、ブラウザと Job 間の WebSocket 双方向通信を実現する方法について紹介しました。

インバウンドを受け付けられないなら Job 側からアウトバウンドで Hub に接続させればよいというのが Relay パターンの発想で、Hub が常駐サーバとしてブラウザと Job の 2 経路の接続を中継することで WebSocket による双方向通信が成立します。

前提として、WebSocket のような双方向通信が必要な場合は、本来は Cloud Run Service のような常駐型サービスを利用するのが一般的です。 Relay パターンはあくまで Job を使わなければならない制約がある場合の回避策なので、特別そういった制約がなければ常駐型の方が実装はシンプルです。

参考・引用