Golangで非同期処理のGraceful Shutdownを独自実装する

現在、キュー処理のコードを書いていて、これは後からでも使えるなと思ったので記事作成。

これはまあ言ってしまえばどんな言語に限らずなのですが・・・

通常、処理が走っている状態のままプロセスを再起動や停止を掛けてしまうと、処理途中でプロセスが無くなってしまうため、その後の処理もそのまま消えてしまいます。

処理の中でファイルをAPIでアップロードした後にデータベースにファイルメタデータを書き込む、みたいな処理にて、APIアップロードの後にデータベース書き込みが終わっていない場合に、シャットダウンするとデータベースの書き込みが無くなってしまいます。当然です。

この問題に対処するため、シャットダウンを行おうとしても既存の処理を行われるまで待って、再起動を掛ける__graceful shutdown__というものが存在します。

Graceful Shutdownはhttpサーバに関してはGolangで標準実装されています。 httpサーバの場合はリクエスト処理を全て処理したのち、リクエストが0になったということを確認して停止することが出来ます。

// httpサーバ定義
srv := &http.Server{Addr: ":8080"}

// goroutine で実行
go func() {
    if err := srv.ListenAndServe(); err != nil {
        log.Print(err)
    }
}()

// httpサーバのGraceful Shutdown
if err := srv.Shutdown(ctx); err != nil {
    log.Print(err)
}

このServer.Shutdownという関数はGo1.8にて追加され、より安全に停止することが可能になりました。

しかし、今回は非同期キュー処理のコードだったので、httpサーバではないため使えません。 なので、今回は独自にGraceful Shutdownのような機構を独自実装します。

この非同期キューはPub-Sub型のキューです。実装しているときはstomp(ActiveMQ)でしたが、ここでは語りません。かなり簡易な実装と置き換えています。

仕組みはこんな感じです。

• キュー処理の購読を開始する。メッセージが飛んで来たら処理を行う。実行している処理はsync.WaitGroupで管理する
• signal.Notifyを使って停止シグナル(SIGTERMなど)を待つ
• Ctrl+Cを押すことでSIGTERMを受信する
• キュー処理の購読を削除する
• sync.WaitGroupを使用して残処理を管理し、全ての処理が終わるまで待つ
• 0になれば残処理が無くなったと判断し、プロセスを終了する。

package main

import (
  "mq"
    "fmt"
    "os/signal"
    "sync"
    "syscall"
)

func main() {
    fmt.Println("Start!")
    var wg sync.WaitGroup
    queue := mq.New()
    // キュー処理の購読を開始する
    queue.Subscibe(func(){
      // 非同期
      // wgに1つ追加
      wg.Add(1)
    　fmt.Println("Queue Start!")
      // 非同期処理(長い処理)
      time.sleep(20 * time.Second)
      // 処理終了
      fmt.Println("Queue End!")
      wg.Done()
    })
    
    // ここでシグナル待ち状態になる
    waitSignal()
    
    // キュー処理の購読を解除する
    queue.Unsubscibe()
    
    // シグナル受信後、全ての処理が終わるまで待つ
    fmt.Println("Graceful ShutDown...")
    wg.Wait()
    // 終了
    fmt.Println("Exit")
}

func waitSignal() {
    var endWaiter sync.WaitGroup
    endWaiter.Add(1)
    var signalChannel chan os.Signal
    signalChannel = make(chan os.Signal, 1)
    // signal.Notifyを使ってシグナルを待つ
    signal.Notify(signalChannel, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
    go func() {
        <-signalChannel
        endWaiter.Done()
    }()
    // シグナルが来ればここのwaitが解除される
    endWaiter.Wait()
}

"mq"というパッケージはないのでこのコードはそのまま動きませんが、実行するとこんな感じになるはず。

Start!
Queue Start! //ここでSIGINT送信
Graceful ShutDown...
Queue End!
Exit

ちゃんと処理が停止してから終了しています。

キャッチ可能なシグナル

ctrl+cを押したときはSIGINT、killコマンドなど一般的な終了はSIGTERM、ハングアップがSIGHUP、強制終了がSIGKILL。

SIGKILLに関しては強制終了なだけあって、キャッチできずに強制的に消されるようです。

https://golang.org/pkg/os/signal/

The signals SIGKILL and SIGSTOP may not be caught by a program, and therefore cannot be affected by this package.

他にもいろいろありますが、一般的なのを抑えておけば大丈夫かな？

Dockerコンテナでの再起動の場合

ちなみにDockerコンテナでもちゃんと動作します。Dockerの再起動や停止のコマンドはこのようになっているようで・・・

http://docs.docker.jp/compose/faq.html

Compose の停止（stop）とはコンテナに SIGTERM を送信して停止することです。 デフォルトのタイムアウトは 10 秒間です 。タイムアウトしたら、コンテナを強制停止するために SIGKILL を送信します。タイムアウトで待っているとは、つまり、コンテナが SIGTERM シグナルを受信しても停止しないのを意味します。

つまり、タイムアウトしなければSIGTERMで処理してくれるとか。デフォルトのタイムアウトも超えてしまうことも考えられる場合は、-tオプションで伸ばすこともできますね。

実際にしっかり問題なく動作しました。

こういうのはセンシティブな処理では、途中で消えてエラーログすら出ずに見落としてしまう危険性を考えると結構重要なので、ぜひ覚えておきたいです。