emahiro/b.log

日々の勉強の記録とか育児の記録とか。

local で Cloud Pub/Sub の動作をエミュレートする

Go GCP PubSub

Overview

ema-hiro.hatenablog.com

上記で書いた gRPC サーバーを local で立ち上げる方法を下に、local で起動したその gRPC サーバーを利用して pubsub の動作(今回は subscriber の動作）をエミュレートするテストを書いてみます。

local でテスト用の pubsub クライアントを実装する

// grpc サーバーを起動する
listener, err := net.Listen("tcp", "localhost:0")
if err != nil {
    t.Fatal(err)
}

ts := grpc.NewServer()
go func() {
    if err := ts.Serve(listener); err != nil {
        panic(err)
    }
}()


// 略

// テスト用の pubsub の Fake サービスを用意する。
type testSubscribeServer struct {
    pubsubpb.UnimplementedSubscriberServer
}

// subscriber の fake サーバーを起動した gPRC のサーバーに登録する
tss := &testSubscribeServer{}
pubsubpb.RegisterSubscriberServer(ts, tss)

// local の gRPC に向けた grpc のコネクションを作成し、そのコネクションを使って pubsub クライアントを生成する。
conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithInsecure())
if err != nil {
    t.Fatal(err)
}
client, _ = pubsub.NewClient(ctx, "test-project", option.WithGRPCConn(conn), option.WithoutAuthentication())

subscriber のサーバーを用意する場合以下のようにエミュレートしたい RPC のプロセスを mock で実装しておく必要があります。

type testSubscribeServer struct {
    pubsubpb.UnimplementedSubscriberServer
}

func (s testSubscribeServer) StreamingPull(r pubsubpb.Subscriber_StreamingPullServer) error {
    r.Send(&pubsubpb.StreamingPullResponse{
        ReceivedMessages: []*pubsubpb.ReceivedMessage{
            {
                AckId: "ack-id",
                Message: &pubsubpb.PubsubMessage{
                    Data: []byte("test-message"),
                },
            },
        },
    })
    return nil
}

今回は streaming-pull の API をエミュレートするために StreamingPull の interface を Fake サービスに実装しました。 pubsub 含め GCP の各種サービスでエミュレートしたいプロセスがあるときは Fake サービスに interface を実装すればよいです。
なお、StreamingPull のプロセスをエミュレートできるとこの Receive 処理の検証が local で可能です。

pubsub の場合は https://github.com/googleapis/google-cloud-go/blob/main/pubsub/apiv1/pubsubpb/pubsub.pb.go にある Unimplemented{$Hoge}Server に生えている interface を見ればエミュレートしたいプロセスをテストサーバーに生やせばよいです。
GCP のサービス全般、gRPC が必要なことが多いのでこの google-cloud-go の中にあるサンプルを探してみるのは良いかもしれません。

まとめ

pubsub に限らず gRPC サーバーが必要になるととたんにテストが多少難解になります。
pubsub のようにそもそもリモートでもテストがしづらいサービスなどを利用してるときは簡易的な動作検証は local で出来ると開発効率は上がることもあるので覚えてて損はない手法かなと思いました。

単体テストのために local で gRPC サーバーを起動する

Go gRPC

Overview

タイトルのとおりです。
このエントリは local で簡単に gRPC サーバーを起動して GCP のライブラリの簡易的な動作検証環境を整えたときの備忘録になります。
モチベーションとしては、単体テストと書いてますが、主に GCP のライブラリの動作を local で簡単にエミュレートしたいとき、GCP のライブラリには gRPC サーバーがセットで必要なことが多く、良く HTTP サーバーのエミュレートで利用される httptest package がそのまま利用できません。
リモート環境にデプロイすることなく local で簡易的な動作チェックが出来ると開発も捗るのでその手順を記載します。

local での起動手順

listener, err := net.Listen("tcp", "localhost:0")
if err != nil {
    t.Fatal(err)
}
ts := grpc.NewServer()
go func() {
    if err := ts.Serve(listener); err != nil {
        panic(err)
    }
}()

これだけです。
※ port を 0 にするとその時空いている適当な port を利用してくれます。

実はこれと同じように gPRC のサーバーを起動するサンプルコードは GCP のライブラリの internal の中に testutil としてコードは公開されていました（これをあとから知りました...)

github.com

local で gRPC サーバーを起動して GCP のライブラリの挙動をテストするときはこのサーバーの実装をまるっと fork してくるほうが早いかもしれません。

また grcp-ecosystem というリポジトリがあって、ここで簡易的な local での gRPC サーバーの立ち上げ方も書かれているのでこういうとこを見て実装方法を調べるのは良いですね。 -> https://github.com/grpc-ecosystem/grpc-cloud-run-example/blob/master/golang/server/main.go

ISUCON14 に参加し、惨敗した

ISUCON 振り返り

Overview

  • isucon 14 に参加した。
  • 今年の目標は去年の点数を超えることと最終30位に入ることだった。
  • 結果: 未達（惨敗）

やったこと

事前準備

  • SRE のメンバーにインフラ関係のワンパン設定 playbook 及び監視環境、デプロイスクリプト等々を準備してもらう。
  • 「イスワン(1時間等制限の中で初期準備を高速化するisucon)」を実施して環境設定周りの手順の統一化。
    • pprotein 等メトリクス系のライブラリ導入練習もここで済ませる。
  • LLM にコードや改善点を吐き出させるためのプロンプト用意。

speakerdeck.com

当日

  • アプリケーションコードの改善
  • DB 分割 (web server 2台構成はうまくいかず撤退)
  • インフラパラメータチューニング
  • アプリケーション側のパラメータチューニング一部
  • 一部データのキャッシュ化
    • ベンチマーカーの整合性チェックが合わず減点対象に。
    • この結果環境が壊れてしまい直せず。

結果

  • 最終300点台。初期スコアより落ちる。
  • 一時は上記改善をやって 14000点台くらいまで伸びて結構いい感じだったが、キャッシュ周りの実装をしたときに本番環境が壊れてしまいベンチマーカーが走ってもスコアが上がりきらなかった。

振り返り

良かった(と思ってる)こと

  • 過去問解く時間を捻出はできなかったが、1時間でやることをチームで共通認識持ってすり合わせる 「イスワン」 はとても良い訓練だった。
    • 業務時間中でも1時間程度であれば取りやすく思いの他効果があった（と思う）
  • 最初から DB を分割しなかったこと。ある程度1台でチューニングしきれるところまでやりきってお昼すぎくらいにDB分割したらスコアが伸びた。もともと MySQL のプロセスが支配的だったのはわかっていたけどその状態である程度チューニングして分割することでちゃんと因果を意識した改善ができた点は良かった。（反省点でも一部触れる）
  • LLM を実戦投入できたこと。
    • アプリケーションコード(と SQL ファイル)をまるっと食わせて Index 提案して、というあたりは精度高くそのまま出力された create index を適用してもスコアには反映された。
    • N+1改善も Copilot と話しながらコード生成してもらったらそこそこいいコードが吐き出す事ができた。このへんは普段の業務でも Copilot に頼ってるいい面が出た。クエリ改善 -> コード改善系も難しいコードがなかったのも今年は良かったと思うけど Copilot は本当に偉大。

反省点

  • アプリケーションコードであまり大きなボトルネックじゃないところを潰すのに時間を割いてしまっていた。特に全体のワークロード（アプリの仕様とも言えるけど）を理解せずにとりあえず非効率なクエリ直したり N+1 を改善しても数百点~1000点位の積み上げにしかならなかった = ボトルネックじゃないところを直しても意味はなかった。
  • MySQL が支配的というのは最初にベンチ流したタイミングでわかっていたので今回に限っては DB を先に分割しても良かったかもなと思った（アプローチ自体が間違っていたとは思わないのでケースバイケースだけど）
  • 「台数不足」のアプローチとして先に owner 側に手を付けず、ユーザーの体験を改善させる（レビューが悪いのはベンチマーカーのログからもわかっていた）方を先に対応するべきだった。
    • notification の改善が終わったタイミングで台数が足りないことがわかっていたので owner 登録改善でいけるのでは？と思ったけど、仕様には売上 UP -> 台数追加と書いてあるので売上 UP のための配車・乗車の改善にアプローチするべきだった。
      • ここはしたにはしたけど効果的な改善方法が見つからなかったというのはある。仕様の理解不足。
  • マッチングの改善以外にもレギュレーションに記載されている内容の精査や理解の時間を取るべきだった。
  • ベンチは常に回せるようにする。
    • 積極的に revert する。
    • ベンチ壊れた状態を長時間保持しない（改善を merge してもベンチ落ちるので意味がない。手持ち無沙汰の時間を減らす）
    • 可能であれば複数台構成にするまでは1台を dev 環境として使う等のルールを決めておく（デプロイスクリプトはそうしてもらってたけどちゃんとは使えなかった）
  • パラメータチューニングもう少し頑張るべきだった。
    • 終盤、ちょっと行き詰まって次の一手が浮かばない時間があってもったいなかったので、実装側のパラメータチューニングにはもう少し意識を向けるべきだった。
      • retry interval を調整したことでスコアが伸びた箇所があったがそこしかチューニングしてなかった。
      • MySQL にしろ nginx にしろスコアに目を向けるのであればパラメータチューニングは王道なので、チューニングの余地があるところはもう少し手を出しても良かった。実際話を聞く限り 5000 点程度の上積みにはなったっぽい。

その他 (感想戦)

当日終了後社内の isucon 参加メンバーで打ち上げをした際にも話したし、Discord の会話も眺めてましたけど DB 分割と web サーバー複数台構成をもう少し早いタイミングでやっても良かったなということと、仕様（アプリケーションのレギュレーション）を理解する時間をちゃんと取るべきだったなと感じた。
仕様の理解については毎年課題を感じているし、今年も取っていなかった訳では無いが、8時間という短い時間の中でルールから加点ポイントを導き出すスキルはまだまだ足りてないと感じる。
ルール理解のプロセスも実際やっていたから一部のパラメータをチューンングしてみようという発想に至ったのは良いことでもあるけどそれ以外にも目を向けるべきだった。
この辺りは ISUCON14 が終わったあとメンバーとも話したけど、ベンチマーカーのログが大きなヒントになっているので、ベンチマーカーのログとレギュレーションを突き合わせて、スコアに効きそうなポイントを洗い出す時間は環境構築してる間にやったりとか、8時間全体のマネジメントには改善の余地があるなと思った。

txtar コマンドで Go のファイルをテキストファイルにアーカイブする

Go

pkg.go.dev

Go のファイルを text ファイルにアーカイブするコマンドが純正で用意されているということを教えてもらって初めて知りました。

以下のように使うと指定したディレクトリ配下の Go のファイルをすべて任意のファイル形式でにアーカイブ出力してくれます。（なおその逆も可能です。）

例えばテキスト形式でアーカイブしたいときは以下のようにします。

# 出力をスキップする
$ txtar ./*.go > ./sample.txt < /dev/null

# echo で空文字を stdin にパイプで渡す
$ echo "" | txtar ./*.go > ./sample.txt

回答するときは --extract(-x) オプションを利用します。

$ txtar -x <./sample.txt

Go のファイルに限らずあるディレクトリすべてのファイルをアーカイブ出力するときは

$ echo "" | txtar ./*.*

というようにすれば良いです。出力結果をパイプで繋いで pbcopy 等に渡せば LLM へのコピペもはかどりますね。

400記事継続の振り返りとこれから

振り返り

この記事がちょうど400記事目です。

300記事のときの振り返りはこちら。

ema-hiro.hatenablog.com

300記事から 400 記事まではだいたい 2年半くらい時間が空いていました。200 記事から 300 記事までは2年ちょっとくらいだったので、エントリを各頻度というものは多少落ちているようです。

300 -> 400 の間は自分の人生のターニングポイントが多く（結婚したり、家族が増えたり、家買ったり）、かなり激動の期間でした。

ブログの内容も技術的な備忘録や読書録だけでなく、子育て周りのコンテンツが増えたというのはこの期間の変化でした。

ブログを書けるようなネタに触れる時間もブログを書く時間も減っていたりするので次の 100 記事はもっと時間がかかるかもしれませんが、このブログのモットーである「無理をしない」「"続ける"ことを続ける」ということを目標にやっていこうと思います。

Go の range over func は nil で panic する

Go

Go 1.23 から導入された range-over func を使うときの注意点として、range にわたす iterator 型のメソッドは nil を返してはいけない、というものがありました。
具体的には以下のコードで range-over func にアサインしてる NilFn は返り値が nil になるときに nilpo で panic します。

package main

import (
    "fmt"
    "iter"
)

func main() {
    for i := range Fn([]int64{1}) {
        fmt.Println(i)
    }
    for i := range Fn([]int64{}) {
        fmt.Println(i)
    }
    for i := range NilFn([]int64{}) {
        fmt.Println(i)
    }
}

func Fn(userIDs []int64) iter.Seq2[int64, error] {
    if len(userIDs) == 0 {
        return func(func(int64, error) bool) {}
    }
    return func(func(int64, error) bool) {}
}

func NilFn(userIDs []int64) iter.Seq2[int64, error] {
    if len(userIDs) == 0 {
        return nil
    }
    return func(func(int64, error) bool) {}
}

https://go.dev/play/p/VU4kff9_evd

range-over func に割り当てる iterator 型の関数を実装するときに、「何も処理をしない」ということをする場合、からの iter 型を返す必要がありました。

Go の slices.Contains はカスタム struct の slice も比較できる

Go

ということを知りませんでした。

// You can edit this code!
// Click here and start typing.
package main

import (
    "fmt"
    "slices"
)

type X struct {
    ID   int
    Name string
}

type xarr []X

func main() {
    x := X{
        ID:   1,
        Name: "taro",
    }

    xx := X{
        ID:   2,
        Name: "jiro",
    }

    xarr1 := []X{x}

    fmt.Println(slices.Contains(xarr1, x))

    xarr2 := []X{xx}

    fmt.Println(slices.Contains(xarr2, x))
}

https://go.dev/play/p/usPRZnNm8rK

また、ある特定のフィールドを持つ struct があるかどうかは slices.ContainsFunc が使えました。

// You can edit this code!
// Click here and start typing.
package main

import (
    "fmt"
    "slices"
)

type X struct {
    ID   int
    Name string
}

type xarr []X

func main() {
    x := X{
        ID:   1,
        Name: "taro",
    }

    xarr := []X{x}

    fmt.Println(slices.ContainsFunc(xarr, func(e X) bool {
        return e.Name == "taro"
    }))
}

https://go.dev/play/p/RgBNAISaNDB

slices package での比較処理は線形探索になるので、あまりに大きな slice だとパフォーマンスに影響が出そうですが、結構柔軟に使うことができて便利です。

『アーキテクトの教科書』を読んでソフトスキルの重要性について考えた

読書Note 備忘録 マネジメント キャリア ひとりごと

Overview

『アーキテクトの教科書』を読んで、アーキテクトに紐づくソフトスキルの重要性について考えてみた。

アーキテクトの教科書 価値を生むソフトウェアのアーキテクチャ構築

アーキテクトの教科書 価値を生むソフトウェアのアーキテクチャ構築

Amazon

本書は全体を通してソフトウェア開発においける「アーキテクト」とどういうもので、どういう役割、業務をこなしていくのかということがわかりやすくまとまっており、ソフトウェアエンジニアのキャリアの先に「アーキテクト」というものを目指してる人は一読する価値がある書籍だと思う。

実際断片的に知っている知識（一部は深堀りもしているかもしれない）が、アーキテクトという業務において点と点が線になるような書き方がされており、ソフトウェアエンジニアとしての業務経験が次のキャリアの礎になっていることを実感できると思う。

私自身は「アーキテクト」になりたいかと言われるとよくわからないが、それに近い振る舞いを業務ですることもあり、自分の業務内容の振り返りと、業務全体をメタ的に認知して自分の日々考えていることを抽象化してみようというモチベーションで読んでみた。
その結果、ソフトウェア開発におけるアーキテクトにはソフトスキルが否応にも求められるのではないか？ということを考えたので、その考えを備忘録として記載している。

ソフトウェア開発におけるアーキテクト

本書の最後の章、6章には以下のように記載されている。

アーキテクトはアーキテクティングを専門領域とするスペシャリストであると同時に、ソフトウェアエンジニアリング全般の知識や経験を有するジェネラリスト

そしてジェネラリストであるがゆえに「ソフトスキル」を一定有することが求められ、その習得についても記載されています。

アーキテクトとソフトスキル

本書を読むとアーキテクトの役割とソフトウェアプロダクトの開発にあたって最初から最後までアーキテクトが密に関わっていくことがわかりますが、その中の一つに「開発プロセスの平準化」の仕事があり、これがちょうど最近自分も読んでいた『Software Design 』のドキュメント回の内容とも重なっていたので、いくつか自分の頭にあったことを言語化して、その中で 「プロセスをチームに落とし込むにはソフトスキルが必要である」 という仮説に行き着きました。

Software Design (ソフトウェアデザイン) 2024年10月号 [雑誌]

Software Design (ソフトウェアデザイン) 2024年10月号 [雑誌]

Amazon

Software Design のこの回でテーマに上がっていたのは、ドキュメントの課題を各社どう解決しているのか？ということだったのですが、ざっと目を通してみて、方向性は様々あれど、各社とも概ね各々ドキュメントを始めとした「開発プロセスの平準化」には取り組んでいて、それがワークフローとして組織の中で回っているのだろうな、ということはわかりました。

そのうえで、このプロセスをどうやって組織にインストールしたのか？ということが疑問だったのですが、結局のところこれは言い出しっぺ（多くはアーキテクトやテックリードといった上位職）がちゃんとワークフローとして組織に取り入れるまでコミットした以外にはなく、このプロセスの組織へのインストールというのはソフトウェアエンジニアリングのスキルではなく、その役割の人が持っていたソフトスキルに依存していたんじゃないか？と考えます。

なぜソフトスキルがあってこそそれができたと考えるのか？というと、 プロセスの導入は骨の折れる仕事で、言っただけでは徹底されず、日々の業務のワークフローとして取り入れ鉄の掟として守り続けるためのコミットメントが必須 だからです。

あくまで僕の一人のソフトウェアエンジニアとしての経験の上でのはなしですが、プロダクト開発（これはPdM側であれTech側であれ）において、何かしらのプロセスを導入しようとすると十中八九ハレーションが発生します。全員が賛成するなんてことはないです。なぜなら、プロセス導入し、 その事によって発生するオーバーヘッドを許容できない立場の人間が組織には一定数存在する からです。

「自由」を志向し開発プロセスと言った画一的な物事の進め方に対してネガティブな反応を示すたちがの人がいることを理解し、個人最適が全体最適にはならないことを徹底して説得、時には上位職（CTOや VPoE といった組織の意思決定権を持つ立場の人) の力を借りながら粘り強くプロセスを入れた際のベネフィットを伝えるコミュニケーションし、時には警察業みたいな嫌われ役を買って出ても組織にプロセスをインストールするにはソフトスキルがより大事になってきます。
ただやろうと言っただけでは人は動きません。鉄の掟を作るというのはそう簡単なことではないからこそ、開発全体に関わり、ソフトウェア開発のプロセスを俯瞰できるアーキテクトにこそできる役割なんだと考えます。そしてその役割をまっとうするにはソフトウェアエンジニアリングのスキルだけでなく、ソフトスキルも求められます。

まとめ

  • アーキテクトはジェネラリストの側面がある。
  • ジェネラリストである以上ソフトスキルからは逃れられない。
  • 組織に鉄の掟を作るときにはコミットメントが必要。
  • そのコミットメントはアーキテクトが発揮するべきである。

考えていたことはこんなことです。たまたま連続して目を通していた本の内容に良いつながりを見つけられたので文章に起こしてみました。

Cloud Pub/Sub における Dead Letter Topic を用いたメッセージのエラー処理

GCP PubSub

Overview

Cloud Pub/Sub でトピックを立てて、メッセージの送受信を行うケースにおいて、メッセージの処理プロセスの中でエラーが発生した場合、そのメッセージのハンドリングが必要になります。
Cloud Pub/Sub にはこのエラー処理の仕組みとして Dead Letter Topic というメッセージのエラー処理の機構が実装されているので、この Dead Leatter Topic を用いて、メッセージの処理プロセスでエラーが発生した場合の復旧方法について検討します。

公式のドキュメントとしては以下にかかれている内容になります。

cloud.google.com

Dead Letter Topic とは？

Cloud Pub/Sub が実装している Dead Letter Queue の仕組みの名称です。

en.wikipedia.org

AWS SQS にも Dead Letter Queue という仕組みが用意されており、その Pub/Sub 版と考えるのがわかりやすいかもしれません。

この仕組みがないとどうなるか？というとキューイングされたあるメッセージにはそのキューに残り続けていられる TTL が存在します。
そのため、メッセージを取り出すことなく TTL を過ぎるとそのメッセージは消失して二度と処理に回すことはできません。

通常はこの TTL 以内にメッセージを取り出し、そのメッセージの情報を使って特定の処理を進めます。(ex SQS で UserID を送信して、AWS Lambda を hook し Lambda 上の処理経由でメールを送信する等）
ただし、この処理がいつも正常に完了するとは限りません。

プロセスが動いているランタイム上で異常により処理を進めなくなるかもしれません。このときメッセージングサービス上では何もしないとエラーが発生したメッセージはキューの中に戻ります。明示的に処理を正常系で完了させたり等、メッセージを削除するマーカーを付けなければメッセージは上記の TTL まで生き残っています。ただ、再度処理をしなければ TTL を過ぎたら消えてしまいます。また同じメッセージをもう一度取り出しても同様のエラーが発生してキューに戻る、という操作を繰り返す可能性もあります。

もしこのメッセージの情報が欠損してはいけない情報（決済の情報であったり CRM の配信情報であったり等）の場合、上記のような杜撰なメッセージ管理をするわけには行きません。必ず、あるメッセージが処理を正常に完了できたのか、できなかったのか？といったことを監視し、状態を把握しておく必要があります。

この状態把握の一つの方法として、規定回数以上の試行に失敗したら「失敗をためておく専用の箱」に移動しておき、この箱の中身を監視しておくことで、メッセージの処理の失敗に素早く気付けるようになります。この箱とのことを DLQ (Pub/Sub では DLT) と呼称しています。

Pub/Sub における DLT の設定方法

Terraform では以下の Example のような設定を追記します。

https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/pubsub_subscription#example-usage---pubsub-subscription-dead-letter

このとき必要になる GCP のリソースが以下です。

  • 通常のTopic (この Topic が DLT の対象になる。)
  • 通常の Topic に紐づく Subscription
  • Dead Letter Topic
  • Dead Letter Topic に紐づく Subscription

※ Dead Letter Topic の識別子は Subscription に紐づける。こうすることで Subscription 経由でメッセージを受信し、そのメッセージに紐づく処理を進めていて異常が発生しても、Retry 上限以上に失敗したら DLT に入ります。

DLT に入ったメッセージを最初に回す方法

これが少しややこしいです。
Cloud Pub/Sub には SQS のような DLQ 再送ボタンと呼ばれるものがありません。AWS SQS の場合、例え DLQ に落ちたとしても AWS のコンソールから再処理のワークフローにメッセージを遷移させることは可能ですが、Pub/Sub にはないのでこのメッセージの再処理フローを時前で用意しないといけません。

大きく再処理フローを実装するパターンとしてあるのは２パターンがあると思います。

  • DLT に紐づく Subscription を用意し、DLT にメッセージが入ったらその Subscription 経由で GCP 上のランタイムを hook して起動させる。（ex. CloudRun)
  • DLT に紐づく Subscription を用意し、定期的（毎分ごとなど）に Message を Pull して処理を進めていく。

1つ目のパターンの場合

シーケンスはざっくり以下のようになるかなと思います。

sequenceDiagram
    participant p1 as publisher
    participant p2 as pub/sub(DLT)
    participant p3 as subscription
    participant p4 as cloudRun(その他のruntime)

p1->>p2: send dead letter message
p2->>p3: send message
p3->>p4: execute runtime (message)
p4->>p4: handling dead letter message

何かしらのメッセージの処理が失敗し DLT にメッセージが入った場合にその最初利用の Runtime (ここでは CloudRun など）内でメッセージの再処理のハンドリングをしたり、このランタイムをプロキシとして他のサービス内でメッセージを処理させます。
GCP 内で完結させる場合、Pub/Sub と CloudRun の連携が簡単にできるので、DLQ に入ったメッセージをハンドリングするランタイムには CloudRun 等 GCP のリソースを選ぶことが多いと思います。（Cloud Function 他）

2つ目のパターンの場合

シーケンスは以下のようになるかなと思います。

sequenceDiagram
    participant p1 as publisher
    participant p2 as pub/sub(DLT)
    participant p3 as subscription
    participant p4 as runtime

p1->>p2: send dead letter message
p2->>p3: send message
p4-)p4: execute runtime
p4->>p3: pull request
p3-->>p4: message
p4->>p4: handle dead letter message

この場合はパターン１とは異なり、Dead Letter Queue に入ったメッセージの再処理をハンドリングするランタイム側で Pub/Sub の Subscription に pull request を送信し、DLQ に入ったメッセージを取得して再処理をするかどうかのハンドリングをします。
このケースでは、pull request を送信する側は GCP の認証を通してリクエストを送信することができればいいので、GCP のリソースに限らずともランタイムの選択肢が何でも良いのはメリットです。

まとめ

Pub/Sub における Dead Letter Queue の処理方法について調べたのでまとめてみました。AWS SQS と違って再処理ボタンがないことで個人的にはめんどくさい実装を挟まないとメッセージのエラー処理フローを構築できない、というのは若干デメリットに感じました。

AWS SDK for Go V2 で Endpoint Resolver の実装方法が変わった話

AWS Go

Overview

タイトルのとおりです。
AWS SDK for Go V2 を使ってるケースで従来の Endpoint Resolver の実装方法が非推奨になり、新しい実装方法が公開されていました。

aws.github.io

Endpoint Resolver とは？

文字通り AWS SDK 経由で AWS のサービスにアクセスするときの Endpoint の向き先を解決するための設定（config) です。

例えば Local 環境で AWS に依存したサービスを立ち上げたり、テストを回したりするときに、Local で起動してる AWS のエミュレーターや Fake サービスのコンテナにアクセス先を切り替えるときなどに利用します。

実装方法

ドキュメントでは S3 の実装方法について記載してますが以下では SQS を使ったときの Endpoint Resolver の実装方法を記載します。

type resolverV2 struct{}

func (*resolverV2) ResolveEndpoint(ctx context.Context, params sqs.EndpointParameters) (smithyendpoints.Endpoint, error) {    
    if IsLocal() {
        // local 環境では local で立ち上がっている sqs のコンテナ、ないしエミュレーターへルーティングする
        u, err := url.Parse("$localSqsEndpoint")
        if err != nil {
            return smithyendpoints.Endpoint{}, err
        }
        return smithyendpoints.Endpoint{URI: *u}, nil
    }
    return sqs.NewDefaultEndpointResolverV2().ResolveEndpoint(ctx, params)
}

func main(){
    opt := []func(*sqs.Options){
        func(o *sqs.Options) {
            o.EndpointResolverV2 = &resolverV2{}
        },
    }
    client := sqs.NewFromConfig(cfg, append(opt, optFns...)...)
    // use sqs client
}

上記のような実装になります。

  1. resolver の型を定義し、サービスごとに ResolveEndpoint interface を実装する。
  2. AWS サービスのクライアントをインスタンス化するときに、interface を実装した resolver を option で指定する。このとき EndpointResolverV2 に、実装した resolver を渡す。

Go で複数の文字を一括で別の文字に置換する

Go

Overview

Go で文字列の中で特定の文字を別の文字に置換したいケース、例えば改行コードやタブが混じっている文字列の中で特殊文字だけをエスケープしたいケースなどで strings.Replace を多用するのではなく、一括で置換する実装について記載します。

strings.NewReplacer を使う

https://pkg.go.dev/strings#example-NewReplacer を使います。

使い方

以下のような実装でタブと改行コードを一括でエスケープ可能です。

strings.NewReplacer("\t", "\\t", "\n", "\\n", "\r", "\\r").Replace("hogehoge...")

上記は公式のドキュメントに記載されてるとおりですが、今回は簡単なユースケースとして http のリクエストとレスポンスをダンプしたときの出力を考えます。

http のリクエストとレスポンスを dump したとき、その中には改行コードが複数入っています。これらをそのまま "" で出力してしまうと、特殊文字が評価されて、dump 結果の改行コード移行が別の標準出力になってしまい、正しくログに出力されない（トレースした結果に出力されない）といったことが発生します。

そのため、出力する結果は適切にエスケープされていてほしいです。このとき何度も置換の実装を書くのではなく、 Replacer を使うと1行で置換の処理を複数の特殊文字に割り当てることができます（便利）

httptest で server を立てるときに Port を固定する

Go

Overview

endpoint のテストをするときに httpserver でサーバーを立ち上げて request と response の検証をする、ということをするときに、httptest で立ち上げるサーバーに割り当てられる port はランダムに決まります。
ただ、テストによっては port を固定したい時もあるので、固定する方法を実装してみました。

port を固定する方法

func TestServer(t *testing.T) {
    listener, err := net.Listen("tcp", "127.0.0.1:9999")
    if err != nil {
        t.Fatal(err)
    }

    mux := http.NewServeMux()
    ts := httptest.NewUnstartedServer(mux)
    ts.Listener = listener
    ts.Start()
    t.Cleanup(func() {
        ts.Close()
    })
}

上記のような方法で固定できます。
固定できるというか、指定した Listener を httptest server で立ち上げた server で指定することでその port で TCP のやり取りをできるようになる、という感じです。

やってみると簡単でした。

GCP で Container Registory から Artifact Registory に移行する

GCP Docker

Overview

GCP の Container Registory が廃止されるので Artifact Registory に移行しました。

cloud.google.com

移行手順

以下のドキュメントに沿って進めました。

cloud.google.com

なお、移行するときに Container Registory -> Artifact Registory へコピーするための権限をリソースに渡す必要があります。

  gcloud projects add-iam-policy-binding $projectID --member=user:$username --role='roles/storage.admin'
  gcloud projects add-iam-policy-binding $projectID --member='$serviceAccountName' --role='roles/storage.objectViewer'

CloudRun をデプロイするときに terminated: Application failed to start: failed to load /docker-entrypoint.sh: exec format error エラーが発生して起動しない

Docker Cloud Run

Mac で作ったイメージを使って CloudRun をデプロイするときに terminated: Application failed to start: failed to load /docker-entrypoint.sh: exec format error というエラーが発生してコンテナが起動しない、ということがあったのですが、これはビルドするときに platform を指定しない状態でビルドしていたためでした。

docker build --platform linux/amd64 -t $tagName:latest としないと Apple Silicon でビルドしたイメージを使ってコンテナは動きません。

zenn.dev

buf CLI を v1.32 以上に上げて設定ファイルを v2 にマイグレーションする

Go buf proto

概要

タイトルのとおりです。
Buf CLI を最新(v1.36) にしたところ、v1.32 以上で大きな変更があった模様で、それに対応しました。

buf.build

zenn.dev

対応

github.com

ハマったところ

新しい Mac で作業していたところ、Connect (依存含め）を動かすためのライブラリを go get 経由で入れて損ねてました。