k6でSocket.IOを扱う - enechain Tech Blog

ogp

この記事はenechain Advent Calendar 2024の2日目の記事です。

はじめに

こんにちは！enechainでソフトウェアエンジニアをしている@taniyarnです。現在は『eSquare Live』のバックエンドを主に担当しています。

『eSquare Live』は、電力卸取引のオンラインマーケットプレイスであり、10か月の立ち上げ期間で開発した新規プロダクトです。バックエンドはGoで構築していますが、リアルタイムに取引情報を表示するため、ストリーミングサーバーにはNestJSを用い、Socket.IOを使ってフロントエンドとリアルタイム通信をしています。

以前の記事で紹介した通り、eSquare Liveでは負荷試験ツールにk6を採用しています。本記事では、Socket.IOサーバーに対する負荷試験をする際にk6を使う方法について解説したいと思います。

はじめに
測定したい項目
k6とは
なぜk6を選定したのか
k6はSocket.IOを標準でサポートしていない
websocket: close 1005 (no status)の回避方法
Socket.IO + gRPC-Webを使ったサンプルスクリプト
さいごに

測定したい項目

負荷試験ツールの選定の前に、今回の負荷試験で何を測定したいのかを詳しく説明したいと思います。今回の負荷試験で注目したいポイントは以下のとおりです。

ストリーミングサーバーのリソース使用状況の測定
- 複数のクライアントを同時に立ち上げて、ストリーミングサーバーに負荷をかける
- サーバーのCPU使用率やメモリ使用率をDatadogでモニタリングする
- 高負荷時のリソース消費と同時接続数を評価する
イベント処理時のAPIレスポンス性能の評価
- イベント受信をトリガーに単一または複数のgRPC-WebのAPIを呼び出すシナリオをテストする
- レイテンシーやエラーレートを測定し、システムの応答性能を評価する
- リアルタイム性が求められる取引情報の更新のパフォーマンスを検証する

これらの測定項目を通じて、システム全体のパフォーマンスやボトルネックを明確にし、必要な最適化ポイントを特定しようと考えました。

k6とは

k6は、HTTP/HTTPSなどの主要なプロトコルを標準でサポートしている負荷テストツールです。さらに、Goを使ったエクステンションxk6で機能を拡張できるため、デフォルトでは扱えないプロトコルやカスタムのテストシナリオにも対応可能です。この柔軟性により、さまざまなシステムに対して効率的かつ効果的なパフォーマンステストを行うことができます。

なぜk6を選定したのか

Socket.IOサーバーに対する負荷試験を行う際の主な選択肢として、以下の2つが考えられます。

ツールを使う
- Artilleryを使用する
  - 公式ドキュメントでも推奨されている
- その他の負荷テストツールを使用する
  - k6、Gatling、Locustなどが挙げられる
自前でスクリプトを書く
- 公式ドキュメントで推奨されている方法の1つ
- 実装コストは多少かかるが、取得するメトリクスを自由にカスタマイズできる

当初、Artilleryを検討しましたが、gRPC-Webのサポートがなかったため見送りました。

一方、k6は標準ではgRPC-Webをサポートしていないものの、xk6を使用することで対応させることが可能です。また、APIの単体テストで既にk6を使用することが決定していたため、一貫性を持たせるためにもk6を試してみることにしました。

もしk6で実現できないことがあれば、自前でスクリプトを書くことも検討しますが、まずはk6での実装に挑戦してみることにしました。

k6はSocket.IOを標準でサポートしていない

ここで1つ問題がありました。それは、k6はWebSocketをサポートしているものの、標準でSocket.IOをサポートしていないということです。

この問題に対するアプローチとして、以下の2つが考えられます。

xk6を使ってSocket.IOに対応させる
WebSocketを使ってSocket.IOに接続する

xk6で拡張するには、GoでのSocket.IOクライアント実装が必要になります。ところが、GoのSocket.IOライブラリであるgoogollee/go-Socket.IOは、クライアント対応が2024年12月現在はプレリリースの段階で正式にはサポートされていません。そのため、適切なライブラリが見つからず、WebSocketを使ってSocket.IOに接続するアプローチを取ることにしました。

幸いなことに、k6のSocket.IO対応に関するIssueがあり、そこに掲載されているスクリプトを利用することで、接続できました。

websocket: close 1005 (no status)の回避方法

接続は成功したものの、約30秒後に websocket: close 1005 (no status)というエラーが発生し、接続が切断されてしまいました。この現象はIssueでも報告されています。

このエラーの原因は、Socket.IOのPing/Pongメカニズムにクライアント側が対応していないためです。 Socket.IOでは、サーバーが定期的にクライアントにPingパケットを送信し、クライアントがPongパケットで応答することにより接続を維持します。クライアントがPongを返さない場合、サーバーは接続を切断します。

解決策として、クライアント側でPingを受信した際にPongを返す処理を追加する必要があります。これにより、websocket: close 1005 (no status) のエラーを回避し、接続を安定させることができます。

Socket.IO + gRPC-Webを使ったサンプルスクリプト

上記の回避方法を対応したサンプルスクリプトがこちらです。 gRPC-Webの拡張は一緒に負荷試験を実施した@shota3506がシュッと作ってくれてxk6-grpc-webとして公開しています。

import grpcweb from "k6/x/grpc-web";
import ws from "k6/ws";
import { check } from "k6";

const GRPC_ADDR = "xxx";
const STREAMING_ADDR = "xxx";
const TOKEN = "xxx";

let grpcWebClient = new grpcweb.Client();

grpcWebClient.load(["proto"], "xxx.proto");

const TEST_DURATION_SECOND = 30;
export const options = {
  vus: 10,
  iterations: 10,
};

export default function () {
  grpcWebClient.connect(GRPC_ADDR);

  var response = ws.connect(STREAMING_ADDR, {}, function (socket) {
    socket.on("open", function open() {
      console.log("connected");
    });

    socket.send(`40{"token":"${TOKEN}"}`);

    socket.on("message", function incoming(msg) {
      if (!msg) {
        return;
      }
      if (msg.startsWith("2")) {
        // 2 is ping
        socket.send("3"); // 3 is pong
      }
      if (msg.startsWith(`42`)) {
        // parse message

        const response = grpcWebClient.invoke(
          "/xxx.xxx/xxx",
          {
            requestParam: xxx,
          },
          {
            metadata: {
              Authorization: `Bearer ${TOKEN}`,
            },
          }
        );
        check(response, {
          "status is OK": (r) => {
            const isOk = r && Number(r.status) === 0;
            if (!isOk) {
              console.log(r);
            }
            return isOk;
          },
        });
      }
    });

    socket.on("close", function close() {
      console.log("disconnected");
    });

    socket.on("error", function (e) {
      if (e.error() != "websocket: close sent") {
        console.log("An unexpected error occured: ", e.error());
      }
    });

    socket.setTimeout(function () {
      console.log(`${TEST_DURATION_SECOND} seconds passed, closing the socket`);
      socket.close();
    }, 1000 * TEST_DURATION_SECOND);
  });

  check(response, {
    "status is 101": (r) => {
      return r && r.status === 101;
    },
  });
}