molpako

もるぱこのブログです

Ceph-CSI の Changed Block Tracking を試してみた

KEP-3314 にて CSI に Changed Block Tracking という機能が提案されました。この機能を使えば効率的なバックアップが可能なため、今回はこれを試します。

想定読者: K8s に Rook/Ceph をデプロイできる

CBT (Changed Block Tracking) 概要

k8s blog に紹介記事がありますので、まずはこちらを確認することをおすすめします。

Announcing Changed Block Tracking API support (alpha) | Kubernetes

CBT は特定ブロックのメタデータ（オフセットと長さ）の取得のみを行います。 取得したメタデータが示したブロックの取得は範囲外なので、バックアップやリストアを行うアプリケーションはユーザーが実装する必要があります。

CSI Driver が CBT に対応するためには、以下２つのgRPCメソッドが定義された SnapshotMetata Service を実装する必要があります。

  • GetMetadataAllocated
  • GetMetadataDelta

GetMetadataAllocated gRPC はあるスナップショットの割り当て済みのブロックのメタデータを取得できます。 つまり書き込まれたデータのみを扱えることに使用できるので、シンプロビジョニングされたブロックデバイスのフルバックアップに最適です。

GetMetadataDelta gRPC は異なる２つのスナップショット間で変更されたブロックのメタデータを教えてくれます。 つまり変更されたデータのみを扱えることに使用できるので、差分バックアップに最適です。

Ceph-CSI v3.7.0 から CBT に対応済み

この記事の執筆時点では、SnapshotMetadata Service に対応している CSI ドライバーは Ceph-CSI のみです。 Ceph-CSI が内部で呼び出す librbd API rbd_diff_iterate3 は比較的最近追加されたもので、この機能を利用するには Ceph v18.2.5 または v19.2.3 以上が必要です。*1

kubernetes-csi/external-snapshot-metadata

GitHub - kubernetes-csi/external-snapshot-metadata: This repo contains sidecar controller for the snapshot metadata service.

kubernetes-csi org には CSI Driver のためのサイドカーがいくつかあります。今回 CBT のために external-snapshot-metadata リポジトリも追加されました。このリポジトリには CRD、サイドカーのソースコード、サンプルプログラムが含まれており、今回の実験でも使用します。

バックアップ検証

ここから実際の検証を行います。以下の環境を前提として手順を進めます。合わせて StorageClass や VolumeSnapshotClass も作成済みとします。

  • K8s v1.33
  • Rook v1.18.4
    • Ceph: v19.2.3
    • ceph-csi-operator v0.4.1

以下の手順で検証を進めます。

  1. K8s クラスタ準備
  2. SnapshotMetadataService インストール
  3. Rook で Ceph クラスタをデプロイ

SnapshotMetadataService デプロイ

まずは SnapshotMetadataService CRD のインストールを行います。手順は kubernetes-csi/external-snapshot-metadata にあるインストール手順 にならいます。

$ git clone git@github.com:kubernetes-csi/external-snapshot-metadata.git ~/external-snapshot-metadata/
$ ~/external-snapshot-metadata/
$ kubectl apply -f deploy/snapshot-metadata-cluster-role.yaml
$ kubectl apply -f deploy/snapshot-metadata-client-cluster-role.yaml
$ kubectl apply -f deploy/cbt.storage.k8s.io_snapshotmetadataservices.yaml

次に SnapshotMetadataService CR を作成します。手順は https://github.com/kubernetes-csi/external-snapshot-metadata/tree/main/deploy/example/csi-driver#installation にならいます。

サービスを作成します。

$ kubectl get service -nrook-ceph rbd-csi-ceph-com-metadata -o wide 
NAME                        TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)    AGE   SELECTOR
rbd-csi-ceph-com-metadata   ClusterIP   10.102.141.49   <none>        6443/TCP   12m   app=rook-ceph.rbd.csi.ceph.com-ctrlplugin

Secret リソースを作成して、 spec.caCert に CA の値を設定してください

$ cat <<EOF| kubectl apply -f -
apiVersion: cbt.storage.k8s.io/v1alpha1
kind: SnapshotMetadataService
metadata:
  name: rook-ceph.rbd.csi.ceph.com
spec:
  address: rbd-csi-ceph-com-metadata.rook-ceph:6443
  audience: rbd-csi-ceph-com-metadata
  caCert: $(base64 -i tls.crt -w 0)
EOF
snapshotmetadataservice.cbt.storage.k8s.io/rook-ceph.rbd.csi.ceph.com created

RBD 用の Driver リソースを変更して先程作った Secret をマウントします。これにより RBD用の Controller Plugin に external-snapshot-metadata sidecar が生えます。

$ kubectl edit driver rook-ceph.rbd.csi.ceph.com -nrook-ceph
+    volumes:
+    - mount:
+        mountPath: /tmp/certificates
+        name: tls-key
+      volume:
+        name: tls-key
+        secret:
+          secretName: rbd-csi-ceph-com-metadata

この SnapshotMetadataService CR を使って CSI Driver にリクエストを送ります。そのために ServiceAccount と ClusterRoleBinding を作ります。 （ClusterRole はすでに作成しています）

$ kubectl get clusterrolebindings backup-app-cluster-role-binding -o wide
NAME                              ROLE                                                   AGE     USERS   GROUPS   SERVICEACCOUNTS
backup-app-cluster-role-binding   ClusterRole/external-snapshot-metadata-client-runner   3m49s                    default/backup-app-service-account

これによって SMS へのリクエストが送れるようになりました。

RBD の VolumeSnapshot を取得

SnapshotMetadataService の準備ができたので、フルバックアップ・増分バックアップのために2つのスナップショットを作成していきます。

PVC のブロックデバイスをアタッチした Pod から 100MBオフセットから 6MB分のブロックを \xFF 埋めします。

$ kubectl exec -it pod-with-raw-block-volume -- bash
root@pod-with-raw-block-volume:/# tr '\000' '\377' < /dev/zero | dd of=/dev/xvda bs=1M count=6 seek=100 conv=notrunc  
...

この状態で VolumeSnapshot CR を作成し、RBD スナップショットを取得します。

$ kubectl get volumesnapshot
NAME                 READYTOUSE   SOURCEPVC       SOURCESNAPSHOTCONTENT   RESTORESIZE   SNAPSHOTCLASS             SNAPSHOTCONTENT                                    CREATIONTIME   AGE
rbd-pvc-snapshot-1   true         raw-block-pvc                           1Gi           csi-rbdplugin-snapclass   snapcontent-61fa8a7f-e7be-4f66-ab15-b91dbc829708   19h            19h

次は300MB オフセットから 6MB 分のブロックを \xAA で埋めて、スナップショットを作成しました。

$ kubectl exec -it pod-with-raw-block-volume -- bash
root@pod-with-raw-block-volume:/# tr '\000' '\252' < /dev/zero | dd of=/dev/xvda bs=1M count=6 seek=300 conv=notrunc  
...

$ kubectl get volumesnapshot -o wide
NAME                 READYTOUSE   SOURCEPVC       SOURCESNAPSHOTCONTENT   RESTORESIZE   SNAPSHOTCLASS             SNAPSHOTCONTENT                                    CREATIONTIME   AGE
rbd-pvc-snapshot-1   true         raw-block-pvc                           1Gi           csi-rbdplugin-snapclass   snapcontent-d067b36a-5ff2-4647-967d-ef7ea8f40209   2m21s          2m21s
rbd-pvc-snapshot-2   true         raw-block-pvc                           1Gi           csi-rbdplugin-snapclass   snapcontent-193d5866-c499-4652-ba24-d2ec7ff0e7b5   31s            31s

実験

external-snapshot-metadata にはサイドカーにリクエストするサンプルプログラムが2つ用意されています。今回はこれらを使います。 - snapshot-metadata-lister: 割り当て済みブロックまたは差分ブロックを表示する - snapshot-metadata-verifier: 割り当て済みブロックまたは差分ブロックをブロックデバイスにコピーする

snapshot-metadata-lister

snapshot-metadata-lister は、引数に与えられた VolumeSnapshot の割り当て済みブロックや２つの VolumeSnapshot の差分ブロックのメタデータを表示します。

external-snapshot-metadata にサンプル Pod のマニフェストがあるのでそれを使います。

$ cd ~/external-snapshot-metadata
$ vim examples/snapshot-metadata-lister/deploy/snapshot-metadata-lister-pod.yaml
-  serviceAccountName: csi-client-sa
+  serviceAccountName: backup-app-service-account

$ kubectl apply -f examples/snapshot-metadata-lister/deploy/snapshot-metadata-lister-pod.yaml
pod/csi-client created

では、実際に実行していきます。まずは rbd-pvc-snapshot-1 の割り当て済みブロックの表示です。ちなみに、4MiB ずつ表示されているのは、RBDイメージのオブジェクトサイズが 4MiB だからです（デフォルト値）

$ kubectl exec csi-client -c run-client -- \
  /tools/snapshot-metadata-lister \
  -namespace default -snapshot rbd-pvc-snapshot-1
Record#   VolCapBytes  BlockMetadataType   ByteOffset     SizeBytes
------- -------------- ----------------- -------------- --------------
      1     1073741824   VARIABLE_LENGTH      104857600        4194304
      1     1073741824   VARIABLE_LENGTH      109051904        2097152

300MiBオフセットから6MiB の変更も確認できました。

$ kubectl exec csi-client -c run-client -- \
  /tools/snapshot-metadata-lister \
  -namespace default -p rbd-pvc-snapshot-1 -s rbd-pvc-snapshot-2
Record#   VolCapBytes  BlockMetadataType   ByteOffset     SizeBytes
------- -------------- ----------------- -------------- --------------
      1     1073741824   VARIABLE_LENGTH      314572800        4194304
      1     1073741824   VARIABLE_LENGTH      318767104        2097152

snapshot-metadata-verifier

まずはバックアップを行うためのブロックデバイスを用意します。変更分のデータのみ書き込まれたかがわかりやすいシンプロビジョニングされたブロックデバイスを用意するために qcow2 イメージファイルを作ります。

docker@minikube:~$ sudo qemu-img create -f qcow2 /mnt/disks/backup.img 1G
docker@minikube:~$ sudo qemu-nbd --format qcow2 -c /dev/nbd15  /mnt/disks/backup.img

docker@minikube:~$ sudo qemu-img info -U /mnt/disks/backup.img  | grep " size:"
virtual size: 1 GiB (1073741824 bytes)
disk size: 196 KiB
cluster_size: 65536

上記のブロックデバイスにアクセスできるように local plugin を使って PV,PVC も作成しました。後述のサンプルマニフェストのために、バックアップ先 PVC の名前を target-device にします。

$ kubectl get pv backup-pv -o wide 
NAME        CAPACITY   ACCESS MODES   RECLAIM POLICY   STATUS   CLAIM                   STORAGECLASS   VOLUMEATTRIBUTESCLASS   REASON   AGE   VOLUMEMODE
backup-pv   1Gi        RWO            Retain           Bound    default/target-device                  <unset>                          70s   Block

まずは rbd-pvc-snapshot-1 のデータをバックアップするので rbd-pvc-snapshot-1 から PVC を作成します。こちらも同様マニフェストのために source-device にします。

cat <<EOF | kubectl create -f -
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: source-device
spec:
  accessModes:
  - ReadWriteOnce
  resources:
    requests:
      storage: 1Gi
  dataSource:
    name: rbd-pvc-snapshot-1
    kind: VolumeSnapshot
    apiGroup: snapshot.storage.k8s.io
  storageClassName: rook-ceph-block
  volumeMode: Block
EOF
persistentvolumeclaim/source-device created

ここまでやったら、リポジトリにあるマニフェストを使ってサンプルポッドを立ち上げます。この Pod は snapshot-metadata-verifier 用に source-device, target-devic pvc をアタッチしています。

$ kubectl apply -f deploy/example/backup-app/testdata/backup-app-pod.yaml
pod/backup-app-client created

$ kubectl get pod backup-app-client -o jsonpath='{.spec.volumes}' | yq -P | head -6
- name: source-device
  persistentVolumeClaim:
    claimName: source-device
- name: target-device
  persistentVolumeClaim:
    claimName: target-devic

$ kubectl get pod backup-app-client -o jsonpath='{.spec.containers[0].volumeDevices}' | yq -P | head -6
- devicePath: /dev/source
  name: source-device
- devicePath: /dev/target
  name: target-device

snapshot-metadata-verifier コマンドをコンテナへコピーして、実行します。 snapshot-metadata-verifier は GetMetadataAllocated を呼び出し、割り当て済みブロックのメタデータを取得します。そのメタデータから src ブロックデバイスからデータを読み出し tgt ブロックデバイスにコピーします。

$ go build -o snapshot-metadata-verifier ./examples/snapshot-metadata-verifier/main.go

$ kubectl cp ./snapshot-metadata-verifier default/backup-app-client:/snapshot-metadata-verify -c run-client

$ kubectl exec backup-app-client -- \
  /snapshot-metadata-verify \
  -namespace default -snapshot rbd-pvc-snapshot-1 \
  -src /dev/source -tgt /dev/target

バックアップ先を確認します。まず、イメージファイルを見ると 6 MB が割り当てされたことがわかります。つまり書き込みされたデータだけがバックアップされました。

docker@minikube:~$ sudo qemu-img info -U /mnt/disks/backup.img  | grep " size:"
virtual size: 1 GiB (1073741824 bytes)
disk size: 6.25 MiB

実際にブロックデバイスを見てみると 100 MiB から 106 MiB まで 0xFF で埋められていて、期待通りです。

docker@minikube:~$ sudo dd if=/dev/nbd15 bs=1M count=1000 2>/dev/null | od -Ax -tx1     
000000 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00
*
6400000 ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff
*
6a00000 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00
*
3e800000

では次のスナップショットの差分だけをバックアップします。先程と同じように２つ目の VolumeSnapshot から PVC を source-device という名前で作成、サンプルポッドを作り直します。 今度は --previous-snapshot を指定して snapshot-metadata-verify を実行します。snapshot-metadata-verify はこのオプションが指定されると GetMetadataDelta を呼び出し、2つのスナップショットの差分メタデータを取得します。そのあとは同様に src から tgt へデータコピーを行います。

$ kubectl exec backup-app-client -- \
  /snapshot-metadata-verify -namespace default \
  --previous-snapshot rbd-pvc-snapshot-1 -snapshot rbd-pvc-snapshot-2 \
  -src /dev/source -tgt /dev/target

イメージファイルを確認すると、差分データである 6MiB が追加で書き込まれたことがわかります。

docker@minikube:~$ sudo qemu-img info -U /mnt/disks/backup.img  | grep " size:"
virtual size: 1 GiB (1073741824 bytes)
disk size: 12.3 MiB

ブロックデバイスを見てみると 300 MiB から 306 MiB まで 0xAA で埋められていることがわかります。

docker@minikube:~$ sudo dd if=/dev/nbd15 bs=1M count=1000 2>/dev/null | od -Ax -tx1     
000000 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00
*
6400000 ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff
*
6a00000 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00
*
12c00000 aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa
*
13200000 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00
*
3e800000

まとめ

  • Ceph-CSI を通して CSI Changed Block Tracking を試しました
  • 割り当て済みブロックや差分ブロックのメタデータを利用して、効率的なバックアップができました

今回はサンプルプログラムがあったので、一行もコードを書きませんでした。SnapshotMetadataService gRPC には、途中で失敗してしまった場合にその途中から再開させるための starting_offset 引数があります。これを試すにはコードを書く必要があるため、次回以降でチャレンジしようと思います。

*1:この API は、スナップショット ID を指定することで、各スナップショット間の差分に対して操作を行えます。なぜこの API が必要かというと、Ceph-CSI の VolumeSnapshot で取得される RBD スナップショットが 元のイメージとは別イメージとして扱われるためです。具体的には、ターゲットの RBD イメージに一時スナップショットを作成し、そのスナップショットを基に イメージクローンで生成された新しい RBD イメージ が提供されます。そのため、異なるイメージ間に存在するスナップショット同士を比較が可能である rbd_diff_iterate3 が必要になるのです

イレイジャーコーディングの雰囲気

もるぱこです。最近は分散ストレージを勉強しています。データの冗長性を高める手法としてRAIDやレプリケーションがよく知られていますが、イレイジャーコーディングという手法もあるそうです。今回はそのイレイジャーコーディングを試してみました。

イレイジャーコーディング（Erasure Coding）とは

イレイジャーコーディングでは、データを k 個のデータシャードと復元用の m 個のパリティシャードに分割・符号化し、分散保存します。 シャードを最大 m 個失っても、残りのシャードから計算でデータを復元することができ、これによって冗長性を高めます。

イレイジャーコーディングが優れている点は、ストレージコストです。例えば、耐障害数を2とするストレージシステムを考えてみます。レプリケーションとイレイジャーコーディングのオーバーヘッドはそれぞれ以下になります。

  • レプリケーション: r - 1 (3レプリカ → +200%)
  • イレイジャーコーディング: m / k (4+2 → 2/4 = +50%)

つまり、イレイジャーコーディングの場合、3レプリカと比べて 1/2 のストレージコストで同じ耐障害性を満たせます。

デメリットとしては、write/read 時に符号化・復元の計算コストは発生することと、小さいファイルではメタデータやパディングのオーバーヘッドが相対的に大きいことが考えられます。

また、符号化には一般的にリードソロモン符号が使われます。これは「データを k 個のシャードに分割し、m 個のパリティを生成して、最大 m 個のシャード欠損から計算で復元できる」代表的な方式です。リードソロモン符号は、CD や QR コードのエラー訂正にも使われており、多少の傷や汚れでも読み取れるのはこの仕組みのおかげらしいです。すごい。

リードソロモン符号のアルゴリズムを理解するのは難しかったので、この記事では解説しません（できません）が、概要を掴むには以下のブログがわかりやすかったです。

試してみる

イレイジャーコーディングを使う分散ストレージ MinIO を参考にして、検証用ツールを作りました。これを用いて、本当にディスク障害が起きてもデータを読めるのか試してみます。

このツールの動作ざっくりこういう感じ。

  • save コマンド: 入力ファイルを読み込んで 4 個のデータシャード + 2 個のパリティシャード に分割し、複数ディスクに書き出す
  • load コマンド: 各ディスク上のシャードを読み、足りない部分を Reed-Solomon の再構築で補完し、標準出力へ出力する

1. 準備

今回 k=4 (データシャード数), m=2 (パリティ数) という構成である 4+2 のイレイジャーコーディングを行います。 まずは、テスト用に k+m=6個の loopback filesystem を作り、これを分散ディスクとして扱います

$ ./scripts/disk.sh prepare .workdir 6
...
$ ls -1d .workdir/mnt/disk*
.workdir/mnt/disk1 # xfs
.workdir/mnt/disk2
.workdir/mnt/disk3
.workdir/mnt/disk4
.workdir/mnt/disk5
.workdir/mnt/disk6

データを4つに分割するため、わかりやすさのために、1行100Bのテキストファイルを ChatGPT に作ってもらいました。 （スペースパディングしています）

$ cat ./examples/sample.txt
The quick brown fox jumps over the lazy dog while counting stars in the endless midnight sky.       
Programming in Go allows developers to build efficient systems with concurrency and simplicity.     
Distributed storage systems like Ceph ensure reliability and scalability across many clusters.      
Unit tests provide a safety net that helps developers refactor confidently without breaking code.

2. シャード生成

save サブコマンドを実行して、上記のファイルを分割 & 保存 (k=4, parity=2) します。

$ go run main.go save ./examples/sample.txt

disk1..4 までは分割されたデータシャードが保存されて、disk5..6 にはパリティシャードが保存されました。

$ head .workdir/mnt/disk{1,2,3,4}/examples/sample.txt/shard
==> .workdir/mnt/disk1/examples/sample.txt/shard <==
The quick brown fox jumps over the lazy dog while counting stars in the endless midnight sky.       

==> .workdir/mnt/disk2/examples/sample.txt/shard <==
Programming in Go allows developers to build efficient systems with concurrency and simplicity.     

==> .workdir/mnt/disk3/examples/sample.txt/shard <==
Distributed storage systems like Ceph ensure reliability and scalability across many clusters.      

==> .workdir/mnt/disk4/examples/sample.txt/shard <==
Unit tests provide a safety net that helps developers refactor confidently without breaking code.   

$ file .workdir/mnt/disk{5,6}/examples/sample.txt/shard
.workdir/mnt/disk5/examples/sample.txt/shard: data # パリティシャードはバイナリデータ
.workdir/mnt/disk6/examples/sample.txt/shard: data

3. シャード欠損（１つ目）

データシャードが保存されている disk1 が故障したことをシミュレートするために、アンマウントします。 これにより disk1 にあるファイルを読めなくなりました。

$ umount .workdir/mnt/disk1
$ find .workdir/mnt/disk* -type f 
.workdir/mnt/disk2/examples/sample.txt/shard
.workdir/mnt/disk3/examples/sample.txt/shard
.workdir/mnt/disk4/examples/sample.txt/shard
.workdir/mnt/disk5/examples/sample.txt/shard
.workdir/mnt/disk6/examples/sample.txt/shard

disk1 が故障した状態で load サブコマンドでファイルを読み込んでみると、無事に元データと同じデータが出力されました。 これによりデータシャードが1つ消失しても、パリティシャードを使って元データを復元できることがわかりました。

$ go run main.go load ./examples/sample.txt
2025/09/23 19:34:50 WARN failed to open shard err="open .workdir/mnt/disk1/examples/sample.txt/shard: no such file or directory"
The quick brown fox jumps over the lazy dog while counting stars in the endless midnight sky.       
Programming in Go allows developers to build efficient systems with concurrency and simplicity.     
Distributed storage systems like Ceph ensure reliability and scalability across many clusters.      
Unit tests provide a safety net that helps developers refactor confidently without breaking code.

4. シャード欠損（２つ目）

4+2 のイレイジャーコーディングは耐障害数（失ってよいシャード数）が2です。 さらにパリティシャードを保管している disk5 が故障したとしても、元データを復元することができます。

$ umount .workdir/mnt/disk5
$ go run main.go load ./examples/sample.txt
2025/09/23 19:34:51 WARN failed to open shard err="open .workdir/mnt/disk1/examples/sample.txt/shard: no such file or directory"
2025/09/23 19:34:51 WARN failed to open shard err="open .workdir/mnt/disk5/examples/sample.txt/shard: no such file or directory"
The quick brown fox jumps over the lazy dog while counting stars in the endless midnight sky.       
Programming in Go allows developers to build efficient systems with concurrency and simplicity.     
Distributed storage systems like Ceph ensure reliability and scalability across many clusters.      
Unit tests provide a safety net that helps developers refactor confidently without breaking code.

5. シャード欠損（３つ目）

3 シャード欠損になると、耐障害数である m=2 を超過し、復元に必要な情報が足りない状態になるので、復元に失敗します。 （パリティ数を増やせば耐障害数は増えますが、ストレージコストと符号化/再構築時の計算コストが増えるトレードオフになります）

$ go run main.go load ./examples/sample.txt
2025/09/23 19:34:53 WARN failed to open shard err="open .workdir/mnt/disk1/examples/sample.txt/shard: no such file or directory"
2025/09/23 19:34:53 WARN failed to open shard err="open .workdir/mnt/disk3/examples/sample.txt/shard: no such file or directory"
2025/09/23 19:34:53 WARN failed to open shard err="open .workdir/mnt/disk5/examples/sample.txt/shard: no such file or directory"
2025/09/23 19:34:53 ERROR failed to load file error="failed to reconstruct data from disks(3/6): too few shards given"

まとめ

  • イレイジャーコーディングはデータを k 個のデータシャードと m 個のパリティシャードに分割・符号化し、最大 m シャード欠損まで復元できる
  • イレイジャーコードのストレージオーバーヘッドはレプリケーションより優れている
    • 代わりに、計算コストが発生する
  • リードソロモン符号はその実装で広く使われるアルゴリズム

次はリードソロモン符号を深堀りしたブログを書きたくて、符号理論やエラー訂正を勉強しているんですが本当に難しい！本当に難しいので気晴らしに「高可用性なTODOアプリを作る」のような無駄なことをするかもしれません。

slog.Error は第2引数で error を受け取らない

試験的なリポジトリにある golang.org/x/exp/slog の Error 関数は、2023年3月ごろの変更で第2引数として error を受け取らなくなりました。これにより golang.org/x/exp/slog を紹介している記事が古くなっている可能性があるので、参考にする場合は注意してください。

変更前

変更前では slog.Error 関数は以下のように第2引数に error オブジェクトを直接渡すことができました。 便利ですね。

slog.Error("an error occurred", err)

内部では slog.Any("err", err) のような処理が実装されていて、ログに err キーとして err の内容が出力されます（ キーは slog.ErrorKey で定義されており exported なので変更できます）

変更後

変更後では slog.Error は第2引数に error を受け取らなくなり Info, Warn 関数と同じインタフェースに統一されました。 キーとバリューを指定するか slog.Attr 構造体を渡してあげる必要があります。

slog.Error("an error occurred", "err", err)
slog.Error("an error occurred", slog.Any("err", err))

この変更の後に golang.org/x/exp/slog は 標準ライブラリ log/slog に導入されました。

注意点

これから log/slogを使う場合、紹介記事などを参考にするかもしれません。参考にした記事が log/slog ではなく golang.org/x/exp/slog を紹介している場合 slog.Error のインタフェースが標準パッケージのものと異なる可能性がありますので、なるべく公式ドキュメントもあわせて確認してください。

まとめ

  • 使用しているライブラリのドキュメントは確認しよう

補足

この変更が入った経緯を把握するために commit https://go.dev/cl/475295 を確認しましたが、理由が書かれていませんでした。なので https://go.dev/cl/475295 で入った差分の紹介だけします。

変更以前では、Errorレベルのために、全てのレベルで呼び出される共通の log メソッドが以下のように error を受け取っていました。そのため Error 以外の全ての関数の内部で log メソッドに nil を渡していました。

// WarnCtx logs at LevelWarn with the given context.
func (l *Logger) WarnCtx(ctx context.Context, msg string, args ...any) {
    l.log(ctx, nil, LevelWarn, msg, args...)
}

func (l *Logger) log(ctx context.Context, err error, level Level, msg string, args ...any) {

逆に Error 関数を利用する場合では、第2引数に error があるせいで、error がない場合に nil を渡す必要があったのです。

slog.Error("Error level, but no errors", nil)

そして、共通の log メソッドから error 引数が削除する変更が入りました。これにより必要のない nil を渡す必要がなくなりました。Error関数で error を渡したい場合には キーとバリューを交互に指定するか slog.Attr を渡す必要が生まれたというわけです。

[Go] TCP接続でハーフクローズを行う

Go

概要

GoでTCPプロキシーを書いていたら、接続先に書き込み完了を伝えるために、ハーフクローズが必要だった。メモ書きレベルだが、まとめておく。

この記事でしないこと

  • ACKとFINなどのパケットについての説明
  • ネットワークコネクション構造体のCloseメソッドについての説明
  • 検証作業

TCPのハーフクローズ

  • TCPは全二重通信である
  • 送受信の双方向からの切断が必要
  • 片側の切断をハーフクローズという

Go net.TCPConn

Go言語でネットワークプログラミングを行う場合、net.Conn インターフェースが一般的に使用される。しかし net.Conn は標準的なネットワークコネクションであるため、全二重通信に必要なハーフクローズに関するメソッドは含まれていない。

ハーフクローズが必要な場合は、net.TCPConnやnet.UnixConn構造体をそのまま扱う必要がある。

読み込み側のクローズ

読み込みの完了後 CloseRead を実行する。接続先から読み込みができなくなる。

func (c *TCPConn) CloseRead() error

書き込み側のクローズ

書き込みの完了後 CloseWrite を実行する。接続先には書き込みができなくなる。

func (c *TCPConn) CloseWrite() error

少しソースを追っかけてみると、shutdown(socket, SHUT_WR) を呼び出されていて、それによって、socketに対する書き込み側通信が閉じられる。という感じだった。（読み取り側も同様）

まとめ

ドキュメントにもある通り基本的にはCloseで済むパターンが多いと思う。しかし、何かしらの理由で片側通信だけをクローズしたいというときはCloseReadやCloseWriteメソッドを実行しハーフクローズすることが必要。

systemd の socket activation を試す

systemd では .socket ファイルを作成することによって socket activation という機能を使えることができます。socket activation はサービスの代わりに socket がリッスンし、リクエストをサービスに渡します。socketがリッスンしているため、サービスのリスタートなどでサービスが落ちている時間もリクエストを受け、サービスが上がったら接続を渡すことができます。

試したコードは molpako/go-server-systemd に置いていて、docker-compose も用意しているのですぐ試せる状態になっていると思います。

サーバーを作る

golang で簡単なサーバーを作ります。なんとなく時刻を返すサーバーにします。

func HelloServer(w http.ResponseWriter, req *http.Request) {
    io.WriteString(w, time.Now().Format(time.Stamp))
    io.WriteString(w, "\n")
}

coreos/go-systemd の activation.Listeners() は systemd から渡されたファイルディスクリプタを元に net.Listener を作り []net.Listener を返します。

func main() {
    listeners, err := activation.Listeners()
    if err != nil {
        log.Fatal(err)
    }

    if len(listeners) == 0 {
        log.Fatal("Unexpected number of socket activation fds")
    }
    mux := http.NewServeMux()
    mux.HandleFunc("/", HelloServer)
    srv := &http.Server{
        Handler: mux,
    }
    if err := srv.Serve(listeners[0]); err != http.ErrServerClosed {
        log.Fatalf("HTTP server ListenAndServe: %v", err)
    }
}

systemd

.socket ファイルにはリッスンするアドレスを記載します

# /etc/systemd/system/hello.socket

[Socket]
ListenStream=0.0.0.0:8076

[Install]
WantedBy=sockets.target

.serviceファイルには先ほどgoで作ったサーバーのバイナリを指定します

# /etc/systemd/system/hello.service

[Unit]
Description=Hello World HTTP
Requires=network.target
After=multi-user.target

[Service]
Type=simple
ExecStart=/app/go-server-systemd

[Install]
WantedBy=multi-user.target

それぞれのファイルを配置したあと socket を起動して、リクエストを飛ばしてみます。

root@5f8340a6141a:/app# systemctl start hello.socket
root@5f8340a6141a:/app# curl 127.0.0.1:8076
Dec 10 15:56:55

返ってきた^^

socket activation を試してみる

ここから本題です。実際にサービス側を落としてリクエストがどうなるか見てみます。

root@5f8340a6141a:/app# systemctl stop hello.service
Warning: Stopping hello.service, but it can still be activated by:
  hello.socket

socket はまだ生きているよというメッセージが出ました。

socketにリクエストを飛ばすと、レスポンスが返ってきました。socket がリクエストを受けたときに service を立ち上げてくれたようです。

root@5f8340a6141a:/app# curl 127.0.0.1:8076
Dec 10 16:15:49
root@5f8340a6141a:/app# systemctl is-active hello.service
active

go の graceful shutdown を組み合わせてみる。

上記だけでは、接続が残っている時に service を restart した場合に、その接続が切れてしまいます。無理矢理ですが接続を残すためにsleep処理を入れて試してみます。

func HelloServer(w http.ResponseWriter, req *http.Request) {
    time.Sleep(5 * time.Second)
    io.WriteString(w, time.Now().Format(time.Stamp))
    io.WriteString(w, "\n")
}

root@5f8340a6141a:/app# curl 127.0.0.1:8076
curl: (52) Empty reply from server   # systemctl restart hello.service

なので本題とはそれますが、ついでに go で作ったhttpサーバーに graceful shutdown を 導入してみます。

main.goでは HUP を受け取ると server.Shutdown() を実行してプログラムを終了するようにします。

// main.go
...
    idleConnsClosed := make(chan struct{})
    go func() {
        sigint := make(chan os.Signal, 1)
        signal.Notify(sigint, syscall.SIGHUP)
        s := <-sigint
        log.Printf("HTTP server Reload: %v", s)

        if err := srv.Shutdown(context.Background()); err != nil {
            log.Printf("HTTP server Shutdown: %v", err)
        }
        close(idleConnsClosed)
    }()

    if err := srv.Serve(listeners[0]); err != http.ErrServerClosed {
        log.Fatalf("HTTP server ListenAndServe: %v", err)
    }

    <-idleConnsClosed
}

.service ファイルに reload の処理を追加します

# /etc/systemd/system/hello.service

[Service]
...
ExecReload=kill -HUP $MAINPID

go の再ビルドと service を再起動して、準備完了です。

graceful shutdown 確認

リクエストを飛ばし続けるスクリプトを実行しておきます。

root@5f8340a6141a:/app# cat test.sh
#!/bin/bash

for i in {1..100}
do
    curl -s http://127.0.0.1:8076
done
root@5f8340a6141a:/app# nohup bash test.sh &
[1] 902
root@5f8340a6141a:/app# tail -f  nohup.out
Dec 10 16:44:23
Dec 10 16:44:28
Dec 10 16:44:33

service を reload することによって graceful shutdown され、接続が切れることは無くなりました。

root@5f8340a6141a:/app# tail -f  nohup.out
Dec 10 16:51:41
Dec 10 16:51:46
^C
root@5f8340a6141a:/app# systemctl reload hello.service
root@5f8340a6141a:/app# systemctl reload hello.service
root@5f8340a6141a:/app# systemctl reload hello.service
root@5f8340a6141a:/app# tail -f  nohup.out
Dec 10 16:51:41
Dec 10 16:51:46
Dec 10 16:51:51
Dec 10 16:51:56
Dec 10 16:52:01
Dec 10 16:52:06

# restart では graceful shutdown にならないので接続が切れる
root@5f8340a6141a:/app# systemctl restart hello.service
root@5f8340a6141a:/app# tail -f  nohup.out
Dec 10 16:54:07
Dec 10 16:54:12
curl: (52) Empty reply from server

以上です。

Pythonの並行処理 concurrentモジュール

python 並行処理 concurrent

@molpako です！

前回 では、multiprocessingモジュールを勉強しました。 今回は、concurrentパッケージを勉強していきます！

concurrentパッケージには、一つだけモジュールがあります。それが、並列実行のための concurrent.futures です。

concurrent.futures

concurrent.futuresは、主に二つのクラスを提供しています。

  • ThreadPoolExecutor
  • ProcessPoolExecutor

この二つのクラスは前回紹介したthreadingとmultiprocessingを呼び出していて、スレッドやプロセスのプールを使用して非同期に実行します。また、両方ともExecutorのサブクラスで同じインターフェースを実装しているので、同じメソッドを提供しています。

では、 前回と前々回で書いた処理をconcurrent.futuresを使用し実装していきます！

ThreadPoolExecutor

まずは並列に実行する関数の作成。引数によって待つ時間を変えれるようにしています。

import select
import socket

def slow_syscall(timeout=1):
    """遅いシステムコールを実行する関数"""
    select.select([socket.socket()], [], [], timeout)

ThreadPoolExecutorを使用して、slow_syscall()を並列に実行します。並列処理の終了を同期するために、withステートメントを使用しています。これは、内部で shutdown(wait=True) を呼び出していて、実行されたプール内のスレッドが全て終わるまで待機させています。

from time import time
from concurrent.futures import ThreadPoolExecutor

start = time()
# with を使うことで、pool内の実行がすべて終わるまで待つ
with ThreadPoolExecutor(max_workers=10) as pool:
    for _ in range(10):
        pool.submit(slow_syscall)

print('Took %.3f seconds' % (time() - start))

>>>
Took 1.006 seconds

ProcessPoolExecutor

関数の作成。

def factorize(number):
    """素因数分解する関数"""
    for i in range(1, number + 1):
        if number % i == 0:
            yield i

def call_factorize(number):
    """イテレーターをリストに変換する"""
    return list(factorize(number))

ThreadPoolExecutorと同じインタフェースを実装しているため、同じように扱うことができます。ついでに計算した結果も出力しておきます。

from concurrent.futures import ProcessPoolExecutor

numbers = [53541233, 21235343, 11421443, 5423123]
start = time()

with ProcessPoolExecutor(max_workers=2) as pool:
    # mapは呼び出す関数をイテラブルな要素それぞれに対して実行する。
    results = pool.map(call_factorize, numbers)

    for result in results:
        print(result)

print('Took %.3f seconds' % (time() - start))

>>>
[1, 5501, 9733, 53541233]
[1, 21235343]
[1, 11, 383, 2711, 4213, 29821, 1038313, 11421443]
[1, 5423123]
Took 4.070 seconds

まとめ

  • threadingやmultiprocessingを直接扱わずとも、concurrent.futuresを使用して並列処理ができる。
  • 両方とも簡単なインターフェースを実装していてとても扱いやすい。

次回は、データのやり取りを安全に行うための queueモジュールについて勉強しますー！

参考文献

Pythonの並行処理 multiprocessingモジュール

python multiprocessing 並行処理

@molpako です！

Pythonを勉強していて並行処理あたりが難しいと感じたので、Golangと比較しながらまとめていきます。

前回 では、threadingモジュールを勉強しました。 今回は、multiprocessingを勉強していきます！

multiprocessing

multiprocessingのサンプルコードをみると start() や join() というメソッドがあるしthreadingと同じじゃん！マルチスレッドとマルチプロセスはどっちを使えばいいんだ！と感じましたが、ドキュメントを見ると答えが書いてありました。multiprocessingモジュールの目的は 並列処理 ということです。

threadingの所で説明しましたが、PythonはGILという仕組みがあって、それがスレッドを同時に一つのスレッドしか動かさないようにしています。multiprocessingはその問題を解決するモジュールらしく、名の通り複数のプロセスを使いマルチコアの恩恵を受け、並列処理ができるみたいです。早速CPUバウンドな処理を並列にして高速化してみましょう。まずは並列ではなく、順番に実行します。

def factorize(number):
    """素因数分解する関数"""
    for i in range(1, number + 1):
        if number % i == 0:
            yield i

numbers = [53541233, 21235343, 11421443, 5423123]

from time import time
start = time()
for number in numbers:
    list(factorize(number))

print('Took %.3f seconds' % (time() - start))

>>>
Took 7.344 seconds

処理時間は約7秒でした。次に、プロセスクラスを作成し並列に実行していきます。複数プロセスの終了を待機するには、Threadクラスと同じように join() を使います。

import multiprocessing

class FactorizeProcess(multiprocessing.Process):
    """計算するプロセスの各処理を表すクラス"""
    def __init__(self, number):
        super().__init__()
        self.number = number
    
    def run(self):
        self.factors = list(factorize(self.number))

# プロセスの開始
start = time()
procs = []
for number in numbers:
    proc = FactorizeProcess(number)
    proc.start()
    procs.append(proc)

for proc in procs:
    proc.join()

print('Took %.3f seconds' % (time() - start))

>>>
Took 4.885 seconds

並列実行の場合の処理時間は約4.9秒！正直パフォーマンス的にもっと早くなるものかと思っていましたが、CPUバウンドな処理でも並列実行され時間が短縮できたのが確認できました。これは、多分、おそらく、予想ですが、プロセスはスレッドより重くオーバーヘッドがありメモリ使用量も多いから？と思います。

ちなみに、multiprocessingのPoolクラスを使用すると上記よりも少ないコード量でワーカープロセスのプールを制御し複数のプロセスを並列に動かすことができます。作成した素因数分解する関数 factorize() を使用して試してみましょう。

def call_factorize(number):
    """イテレーターをリストに変換する"""
    return list(factorize(number))


start = time()

# 計算する要素分プロセスを立ち上げる
with multiprocessing.Pool(len(numbers)) as pool:
    results = pool.map(call_factorize, numbers)

    for result in results:
        print(result)

print('Took %.3f seconds' % (time() - start))

>>>
[1, 5501, 9733, 53541233]
[1, 21235343]
[1, 11, 383, 2711, 4213, 29821, 1038313, 11421443]
[1, 5423123]
Took 5.252 seconds

ちなみにgolangでは、前回と同じようにgoroutine 1 を使えば計算処理もはやくなります。

メモリの共有

2つのプロセス間でデータのやり取りをするためには、Pipeクラスを使用します。（Queueクラスもありますが、別の記事で紹介します！）2

Pipe()が返すコネクションオブジェクトは send() , recv() などのメソッドがあり、socketオブジェクトに似ていますね。早速Pipeクラスを使用して2つのプロセス噛んでデータをやり取りしてみましょう。

FactorizeProcess を少し変更して、コネクションオブジェクトを扱えるようにします。プロセスが開始されると、計算をし、結果をパイプ先のプロセスへと送信します。

class PipeFactorizeProcess(multiprocessing.Process):
    """計算するプロセスの各処理を表すクラス
    結果をパイプ先のプロセスに送信する"""

    def __init__(self, numbers, conn):
        super().__init__()
        self.numbers = numbers
        self.conn = conn

    def run(self):
        for number in self.numbers:
            self.conn.send(list(factorize(number)))
        self.conn.close()

受信用のプロセスは、5秒間データが受信できるか確認します。確認ができたら受信をして、データの受信がなければコネクションを閉じます。

# Pipe()は、パイプの両端を表すConnectionオブジェクトのペアを返す！
parent_conn, child_conn = multiprocessing.Pipe()
p = PipeFactorizeProcess(numbers, child_conn)
p.start()
while True:
    if parent_conn.poll(5):
        print('receive: {}'.format(parent_conn.recv()))
    else:
        parent_conn.close()
        break
p.join()

>>>
receive: [1, 5501, 9733, 53541233]
receive: [1, 21235343]
receive: [1, 11, 383, 2711, 4213, 29821, 1038313, 11421443]
receive: [1, 5423123]

ちなみにgolangでは、プロセスやスレッドを扱わず、goroutineを扱います。goroutine間でのデータのやり取りはチャネルの通信によってデータのやり取りを行います。

// 素因数分解する関数
func factorize(numbers []int, c chan<- []int) {
    for _, number := range numbers {
        var a []int
        for i := 1; i < number+1; i++ {
            if number%i == 0 {
                a = append(a, i)
            }
        }
        c <- a
    }
    // 送信側がチャネルをクローズする
    close(c)
}


func main() {
    numbers := []int{53541233, 21235343, 11421443, 5423123}
    c := make(chan []int)
    go factorize(numbers, c)

    for i := range c {
        fmt.Printf("receive: %d\n", i)
    }
}

>>>
receive: [1 5501 9733 53541233]
receive: [1 21235343]
receive: [1 11 383 2711 4213 29821 1038313 11421443]
receive: [1 5423123]

まとめ

  • start()やjoin()などthreadingとAPIが似ている。（ので、移行がしやすい）
  • threadingと違い、マルチコア実行ができる。
  • Poolクラスにより複数プロセスの管理が簡単になる。
  • Pipeクラスにより二つのプロセスでデータをやり取りできる。

次回は 並列性 のための councurrent モジュールについて勉強しますー！

参考文献

  1. スレッドより小さい軽量スレッド。Golangで並行・並列処理する場合には、goroutineを扱う。↩

  2. データをやり取りするためにメモリを共有する方法があるが、デフォルトではメモリを共有しない。プロセス同士でメモリを共有したい場合は、ValueクラスやArrayクラスもしくは multiprocessing.sharedctypes を使用する。↩