Cloudflare Workers - サーバレス環境で Rust を動かす件

Cloudflare Workersとは？

簡単に言えば、Cloudflare社が提供するサーバレス環境です。誤解を恐れずに言えば、AWS Lambdaみたいなものです。こんな感じのJavaScriptのHTTPリクエスト・ハンドラーをサーバを構築することなく実行できます。

Cloudflare Workers HTTPハンドラー

export default {
	async fetch(request, env, ctx) {
		return new Response('Hello World!');
	},
};

そして、JavaScriptだけでなく、なんとRustも動きます!

Rustハンドラー

use worker::*;

#[event(fetch)]
async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
    let  a = req.headers().get("x-forwarded-for").unwrap();

    Response::ok("Hello, World!")
}

厳密にはGoogleが開発したV8 Isolatesをランタイムに使用しています。fsなど一部のAPIは実装していませんが、基本的にNode.js互換です。V8 Isolatesの売りはコールド・スタートがゼロであることです。
Vercel Next.jsではCloudflare WorkersとAWS Lambdaの２つのランタイムを使用しています。

WebAssembly

がっかりされるかもしれませんが、ご察しの通り、V8はJavaScriptのランタイムでありますから、Rustのネイティブコードを動かすには、WebAssembly(Wasm)を使用する必要があります。RustからほぼネイティブのWASMコードを生成して動かします。

とはいっても、Wasmの知識はいっさい不要です。workers-rsを使用すると、そのあたりのwasm関連のコードの生成を自動的にやってくれるので、開発者はrustのコードだけに注力できます。

実際、私はWebAssemblyを知らずこの記事を書いています。もちろん、Rustの知識は必要です。

async/awaitのシンタックスで非同期で記述できます。tokioのライブラリはmacroとio-utilのみ使用できます。スレッドの生成はできません。

サポートしているcrateはこちら。

また、RustからTCPソケット接続することもできます。WebSocketではなく、TCP直のソケットです。HTTPハンドシェイクが不要になるので接続が速くなります。PostgresとTCPソケットレベルで接続するサンプルなどあります。

https://blog.cloudflare.com/workers-tcp-socket-api-connect-databases/

Hello, World!

ではさっそく、プロジェクトを作成しましょう。pnpmを使用した方法です。Cloudflareのアカウントの作成は済ませておいてください。無料プランでクレカの登録は不要です。
ソースコードをこちらに置きます。

プロジェクトの作成

mkdir my-rust-worker-prj
cd my-rust-worker-prj
pnpm init
pnpm install wrangler --save-dev
pnpm wrangler generate my-project https://github.com/cloudflare/workers-sdk/templates/experimental/worker-rust
cd my-project

生成されるmy-project/package.jsonが古いので、以下のように修正します。

package.json

  "devDependencies": {
    "@cloudflare/workers-types": "^4.20230419.0",
    "typescript": "^5.0.4",
    "wrangler": "^3.0.0"
  }

インストール

pnpm install (my-projectフォルダ配下)

my-rust-worker-prj/my-projectにRustのプロジェクトが作成されます。以下、このmy-projectをカレントディレクトリとして作業します。VSCodeやRustRoverでmy-projectを起点にして開きコーディングしていきます。

wranglerとは、GCP gutilやVercel CLIのような、Cloudflareクラウド環境とやりとりするためのCLIツールです。ローカルでサーバを起動したり、Cloudにデプロイするのに使用します。

サンプルのHello, World!プログラムが自動生成されてますので、そのままローカルでビルド&起動します。8787ポートでサービスが起動します。

ローカルでサーバ起動

pnpm run dev (my-projectの直下)

cURLで叩いてみます。あっさりと動いてしまいます!

ページを開く

➜  my-project git:(main) ✗ curl http://localhost:8787

Hello, World!

Wrangler V3でローカル開発環境(Miniflare V3)でWorkerを動かせるようになりました。Wasmも動くようです。Miniflareとはminikube(ローカルKubernetes)のような、ローカルでWorkersランタイム環境を再現するものです。Wranglerに包括されています。KV, R2, D1, Cache and Queuesのシミュレータがローカルで動きます。wrangler dev --remoteとすれば、クラウド環境のマネージド(KV等)をポイントすることもできます。

workers-rsの公式ドキュメントはWrangler V2をベースに記述されていますが、V3でも動きました。私の環境はMacbook Air M2です。
ブラウザからアクセスしても良いですが、/favicon.icoなど余計なトラフィックが発生したりするので、APIのテストはcURLを使用するのが良いでしょう。

コーディング

src/lib.rsが本体のコードです。少し書き換えてみましょう。Hello, World333!とか、適当に書き換えてください。

src/lib.rs

use worker::*;

#[event(fetch)]
async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
    Response::ok("Hello, World333!")
}

Wranglerはファイルの更新を検出し、再ビルド、再起動してくれます。(たまにおかしくなります。)

ページを開く

curl http://localhost:8787/

Hello, World333!

プログラムが正常に動くことを確認したら、クラウド環境にデプロイしましょう。

デプロイ

npx wrangler deploy (pnpm run deploy)

⚡ Done in 2ms
Total Upload: 361.72 KiB / gzip: 154.46 KiB
Uploaded worker-rust (3.86 sec)
Published worker-rust (1.53 sec)
  https://worker-rust.xxx.workers.dev
Current Deployment ID: b7b07901-a1fb-4054-b200-efa92b8636b6

リモートのWorkerの実行ログをローカルに出力します。

ログ

npx wrangler tail (pnpm run tail)

実行します。

ページを開く

curl https://worker-rust.xxx.workers.dev

こんな感じで簡単に開発を進めることができます。

ビルド自体はローカルで行われます。ファイルのサイズが大きくなるとコードのパースに時間がかかるため、コールド・スタート時間の問題が顕著になります。

KVとD1

CloudflareではAmazon RDSのようなDB系のマネージド・サービスがあり、Rustからもアクセスできます。ここでは無料版でも使用できるKVとD1をご紹介します。

KV

KVとはRedisの廉価版のようなキー・バリューストアです。eventually-consistentで書き込みが反映されるのに最大60秒かかるようです。strong-consistentが求められる場合はDurable Objectsという別のサービスを使用します。

開発用のKVをクラウド上に作成しましょう。--previewオプションを指定すると開発用のKVになります。まずは名前空間(namespace)を作成します。お好みの名前をつけてください。ここではmyrustkvとしました。

KVの名前空間の作成

npx wrangler kv:namespace create myrustkv --preview

Add the following to your configuration file in your kv_namespaces array:
{ binding = "myrustkv", preview_id = "b06bc387f865433291255ea31c2c51c6" }

プロジェクト名-名前空間_previewでKVの名前空間が作成されます。ID(preview_id)が発行されますのでメモしてください。

Cloudflareのダッシュボードから、正常に作成されたか確認しましょう。

一つハマりポイントがあります。開発用の他に本番用も作成する必要があります。--preview falseと指定すると本番用の名前空間が作成されます。

本番用も作る必要あり

npx wrangler kv:namespace create myrustkv --preview false

次に、wrangler.tomlファイルに[[kv_namespaces]]を定義します。idとpreview_idに先ほどのコマンドの出力結果のIDの値を指定します。

wrangler.toml

name = "worker-rust"
main = "build/worker/shim.mjs"
compatibility_date = "2023-03-22"

[build]
command = "cargo install -q worker-build && worker-build --release"

[[kv_namespaces]]
binding = "myrustkv" <- Rustから参照するキー (なんでも良い)
id="e7b97c05be5440f2b0e4efff640c14c5" <- 本番用 preview falseの結果
preview_id = "b06bc387f865433291255ea31c2c51c6" <- 開発用 preview trueの結果

binding名はRustのコードから参照する名前になります。したがって、名前空間に合わせる必要はありませんが、ややこしいのでmyrustkvと設定しました。

確認のため、コマンドラインからキー・バリュー(mykey1:myval1)をインサートしてみましょう。

npx wrangler kv:key put --binding=myrustkv "mykey1" "myval1" --preview

ダッシュボードから正常に登録できたか確認してください。

次はRustからインサートしてみましょう。

KV

use worker::*;
use reqwest;

#[event(fetch, respond_with_errors)]
async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
    let kv = env.kv("myrustkv")?; // wrangler.tomlのbindingを参照

    kv.put("key3", "value3")?.execute().await?; // セット
    let val = kv.get("key3").text().await?; // 取得
    console_log!("key3: {:?}", val);

    Response::empty()
}

Workerサービスを起動します。

RUST_LOG=info npx wrangler dev --remote

⚡ Done in 13ms
Your worker has access to the following bindings:
- KV Namespaces:
  - myrustkv: b06bc387f865433291255ea31c2c51c6 <- バインドされたか確認！
⬣ Listening at http://0.0.0.0:8787
- http://127.0.0.1:8787
- http://192.168.13.2:8787

正常に起動すると、このようにコンソールにKVにアクセス可能になったとのメッセージが表示されます。cURLで叩いて確認してください。

--remoteオプションをつけると、worker(rust)はローカル環境で動かしつつ、KVはクラウド上のサービスへ接続することができます。

OSシングルスレッドで実行しますから、awaitでスレッドを譲る必要があります。私のこちらの記事も参考にしてください。

https://zenn.dev/tfutada/articles/5e87d6e7131e8e

D1

D1とはCloudflareのマネージドなSQLiteベースのRDBです(まだベータです)。サーバレス環境から他社のデータベース・サービスを利用するのは、物理的なロケーション(ネットワーク遅延)や、コネクションプーリングの問題などから現実的ではありません。Vercel PostgresやSupabase Supavisorもそうですが、自社提供で最適化したデータベース・サービスやSDKの利用が望ましいです。

mizchiさんもヤバいといっていたのですごいんだと思います。

d1データベースを作成します。

npx wrangler d1 create d1-example

次にテーブルの作成とデータの仕込み作業をします。schema.sqlファイルを用意します。

schema.sql

DROP TABLE IF EXISTS Customers;
CREATE TABLE IF NOT EXISTS Customers (customer_id INTEGER PRIMARY KEY, company_name TEXT, contact_name TEXT);
INSERT INTO Customers (CustomerID, company_name, contact_name) VALUES (1, 'Alfreds Futterkiste', 'Maria Anders'), (4, 'Around the Horn', 'Thomas Hardy'), (11, 'Bs Beverages', 'Victoria Ashworth'), (13, 'Bs Beverages', 'Random Name');

--localオプションを指定してローカル開発環境のDBにテーブルを作成します。

npx wrangler d1 execute d1-example --local --file=./schema.sql

正しくデータがインサートされたかSELECT文で確認します。

npx wrangler d1 execute d1-example --local --command='SELECT * FROM Customers'

Customersテーブルに以下の４つのレコードがインサートされていればOKです。

では、コーディングをしていきましょう。

wrangler.tomlにd1のバインディングを定義します。databse_idは先ほどのSELECTの結果に表示されています。パッと見、昔風のSQLをベタに書いていくインターフェースです。

wrangler.toml

[[d1_databases]]
binding = "customer-db" # i.e. available in your Worker on env.DB
database_name = "d1-example"
database_id = "9aa38d15-8716-43f9-8cb1-31f633c7638b"

d1のライブラリを追加します。

Cargo.toml

worker = { version = "0.0.15", features = ["d1"] }

Rust Workerのコードを書きます。お約束でserdeでテーブルとマッピングします。

use serde::{Deserialize, Serialize};
use worker::*;

#[derive(Deserialize, Serialize, Debug)]
struct Customer {
    #[serde(rename = "CustomerId")]
    customer_id: u32,
    #[serde(rename = "CompanyName")]
    company_name: String,
    #[serde(rename = "ContactName")]
    contact_name: String,
}

#[event(fetch, respond_with_errors)]
pub async fn main(request: Request, env: Env, ctx: Context) -> Result<Response> {
    let company_name = "Bs Beverages".to_string();
    let d1 = env.d1("customer-db")?;
    let statement = d1.prepare("SELECT * FROM Customers WHERE CompanyName = ?1");
    let query = statement.bind(&[company_name.into()])?;
    let result = query.first::<Customer>(None).await?;
    console_log!("result: {:?}", result);

    Response::empty()
}

ローカルで起動します。

RUST_LOG=info npx wrangler dev

Your worker has access to the following bindings:
- KV Namespaces:
  - myrustkv: b06bc387f865433291255ea31c2c51c6
- D1 Databases:
  - things-db: d1-example (9aa38d15-8716-43f9-8cb1-31f633c7638b)

D1データベースがバインドされたことを確認してください。

curlで呼び出します。以下の出力があればOKです。

実行結果

result: Some(Customer { customer_id: 11, company_name: "Bs Beverages", contact_name: "Victoria Ashworth" })

外部APIのfetch

reqwestを使用して、Rust Workerから外部のAPIを呼ぶことができます。JavaScriptのfetchを経由して外部のAPIを呼び出します。ここでは、pokemon API(https://pokeapi.co/api/v2/pokemon)を利用したサンプルを紹介します。

use worker::*;
use reqwest;

// proxy to upstream with reqwest
#[event(fetch)]
async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
    // Make a GET request to Pokemon API
    let response = reqwest::get("https://pokeapi.co/api/v2/pokemon/ditto").await;

    match response {
        Ok(res) => {
            let body = res.text().await.unwrap_or_else(|_| "Failed to parse Pokemon response".to_string());
            Response::ok(body)
        }
        Err(_) => Response::ok("Failed to fetch"),
    }
}

このようにプロキシー(ゲートウェイ)的なサービスが簡単に作成できます。

WasmですのでJavaScriptとのやりとりのオーバーヘッドがあります。

試しに、呼ばれる側のAPIでスリープを60秒入れてみましたが、Workerはタイムアウトすることなく正常に終了しました。I/O待ちの時間はカウントされないためです。CPU Time 10ms

tokio::join!でpromise allライクにコンカレントにリクエストを投げることもできます。一つのOSスレッドで処理されます。

tokio：：join!

#[event(fetch)]
async fn main(req: Request, env: Env, ctx: Context) -> Result<Response> {
    // Make a GET request to Yahoo and Google
    let req1 = reqwest::get("https://www.yahoo.co.jp");
    let req2 = reqwest::get("https://www.google.com");

    // Send both requests concurrently
    let (yahoo_response, google_response) = tokio::join!(req1, req2);

TCPソケット

最後にTCPソケットを使用したサンプルを紹介します。Cloudflare WorkersからTCPソケットで直接TCPベースの外部サービスと通信することができます。PostgresなどHTTP上ではなく、TCP直で通信が可能です。HTTPプロトコルのオーバーヘッドが無くなるので速くなります。

Today, we are excited to announce a new API in Cloudflare Workers for creating outbound TCP sockets, making it possible to connect directly to any TCP-based service from Workers.
公式ブログ

ここでは簡単なエコー・サービスを作ってみたいと思います。

まず、サーバ側を作成します。言語はなんでも良いですが、私はtokioで実装しました。よくあるエコー・サービスです。

TCPサーバー

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

mod constants; // pub const SERVER_ADDR: &str = "127.0.0.1:5000";

// TCP server using Tokio, echo server
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind(constants::SERVER_ADDR).await?;
    println!("Server is listening on {}", constants::SERVER_ADDR);

    while let Ok((mut socket, _)) = listener.accept().await {
        tokio::spawn(async move {
            let (mut rd, mut wr) = socket.split();
            tokio::io::copy(&mut rd, &mut wr).await.expect("TODO: panic message");
        });
    }
    Ok(())
}

さて本題のWorkerから上記のサーバにTCP接続してみます。worker::Socketを使用して先ほどのTCPサーバーに接続します。

worker、TCPクライアント

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use worker::*;


#[event(fetch)]
async fn main(_req: Request, _env: Env, _ctx: Context) -> worker::Result<Response> {
    let mut socket = worker::Socket::builder()
        // ngrokでlocalhostにプロキシーしてます。
        .connect("xxx.ngrok-free.app", 443)?;

    // 送信
    socket.write_all(b"Hello, server!").await?;

    // 受信
    let mut buffer = Vec::new();
    let bytes_read = socket.read_to_end(&mut buffer).await?;

    // ログ出力
    let received_data = String::from_utf8_lossy(&buffer[..bytes_read]);
    console_log!("Received from server: {}", received_data);

    Response::ok("done!")
}

worker::SocketはAsyncReadトレイトを実装していますので、blanket implementationsのAsyncReadExtが有効になります。

上記のサーバ、クライアントをローカルで起動しcURLで叩いてみます。どうでしょう？うまくいきましたか？このように、TCPでメッセージの送受信が簡単でできます。

wrangler環境のせいかlocalhostだとうまく繋がりませんでした。回避策として、ngrok経由で繋ぐようにしたらうまくいきました。Docker環境みたく、ホスト側にはlocalhostで繋がらないのかもしれません。ngrokは固定のホスト名が無料で貰えて良いですね。

もちろん、サーバレス環境のためソケットの接続と開放を関数内で完結させる必要があります。サーバレス環境はインスタンスが短命であり、メモリ・リソースが限られているためコネクション・プーリングなどは向いていません。

HyperでこのTCPソケットを使用することもできるようです。他にPostgresも可能のようです。

hyper

// workersのTCPソケットを利用
let socket = worker::Socket::builder()
        .secure_transport(SecureTransport::On)
        .connect("example.com", 443)?;

    let (sender, connection) = hyper::client::conn::handshake(socket)
        .await
        .map_err(map_hyper_error)?;

    let request = hyper::Request::builder()
        .header("Host", "example.com")
        .method("GET")
        .body(hyper::Body::from(""))
        .map_err(map_hyper_http_error)?;

Rustで記述する意味はあるのか？

RustはLinuxカーネルや、Googleの組み込みOSのKataOSなどの実装に利用されているように、リソースが限られた環境にもってこいの言語です。私もActixWebやQdrantなどのRust実装のアプリケーションをKubernetesで動かしていますが、リソースを食わないのでクラウドの料金が安く済み助かります。
また、サーバレスではデバッグが難しいですが、その点でも型安全でGCのないRustは手離れが良く本番環境に強いです。先日も某システムがメモリ不足で障害を発生させましたが、明日は我が身と感じます。

無料プランの場合、WorkersのCPU時間は10ms、メモリ使用量は128MBまでの制約があります。
CPU timeとはI/O待ちを含まない、CPUサイクルを消費した時間になります。ちなみにI/Oを含めた全体の処理時間のことをwall-clock timeと言います。wall-clock timeに制限はありません。詳細はWorkersのリミットをご覧ください。

他の制約として、外サービスの呼び出しは同時に６つまでになります。R2やKVの呼び出しもこれにカウントされます。
また、コールド・スタート時にworkerのコードを400ms以内でパーシングする必要があります。workerのファイルサイズは1MB(圧縮後)になります。workerは同時に100個まで実行できます。
console_logログ出力は128kbまでになります。

Wasmのオーバーヘッドはあります。JavaScriptからの呼び出し、パーシングなど。

最後に

正直、無料でここまで遊べるのかと感動しました。本当に良い時代になりましたね。今の学生さんが羨ましいです。

次回は、Web3連携を試してみたいと思います。実はCloudflareではWeb3のゲートウェイ・サービスもベータで開始しています。SolanaのWasmなど動かしてみたいです。