http://sensuapp.org/
sensuは、2012-2013年ごろに登場した監視フレームワークで、nagiosの問題点を解決しうるモダンな設計を搭載して注目されている. 日本でもスタートアップでの採用事例が増えている点とrubyで実装されている点を考慮してsensuを導入してみることにした.

sensuの基本構成

sensuを導入するといろんなものがインストールされていろんなプロセスが立ち上がるが、この中で実際に監視業務をおこなっているのはsensu-client, sensu-serverの2つ. この両者がrabbitmqを介してどのようにメッセージのやりとりを行っているか理解することでsensuの概要が見えてくる.

SENSU SERVER: 監視サーバ
SENSU CLIENT: 監視対象サーバ

図１

rabbitmq概要

https://www.rabbitmq.com/
rabbitmqはAMQPを実装したミドルウェアのひとつでデファクトスタンダードになりつつあるらしい.
AMQP(Advanced Message Queuing Protocol)についての説明は以下が分かりやすい.
AMQPによるメッセージング

rabbitmqのTCP接続

図１のようにsensu-serverとsensu-clientが通信をおこなうためには、監視サーバのrabbitmq-serverの口が開いていればよい. TCP接続が実施されるのは新たなsensu-clientが起動されたときのみで、その後は常時ESTABLISHED状態になる. rabbitmq-serverプロセスが停止すると一旦接続が切れるが、復旧が確認されるとsensu-clientが自動で接続を再開する.
当然、sensu-serverは監視対象サーバの数だけ常時ESTABLISHEDなTCP接続を保持することになる.

##
## rabbitmq接続状態を確認
## 
netstat -an | grep ESTABLISHED | grep 5672
  # ※5672はrabbitmqのデフォルトlistenポート

rabbitmqのメッセージ送信と受信

rabbitmqではメッセージのやりとりがおこなわれるバッファをキューとよんでいる.
キューをsubscribeすることでメッセージを受信し、キューへpublishすることでメッセージを送信できる.
exchangeとbindという仕組みによってキューをまとめてひとつひとつのキュー名を知らなくても特定のクラスタへブロードキャストするようなことができる.

sensuではsensu-server, sensu-clientがそれぞれ以下のようにメッセージの送受信をおこなう.
(production, stagingというのはclient.jsonのsubscriptionおよびsensu checkのsubscribersで自由に定義可能)

sensu-server => sensu-clientの通信

監視用のコマンドなど(Sensu Checks)がsensu-server => sensu-clientへメッセージとして送られる.
sensu-serverは多数のsensu-clientを相手にするので、このときsensu-serverはメッセージを送る相手ひとりひとりを意識しない. 代わりにexchange名（Sensu Checks jsonで指定するsubscribers)を通じて特定のクラスタへメッセージを送る.

sensu-serverが送信しているメッセージを傍受するスクリプト
※exchange名(subscribers名)を知っていさえすれば、rabbitmq-serverにESTABLISHEDしているどのホストからでも内容をsubscribeできる.

client_listen.rb

# !/usr/bin/env ruby

require 'amqp'

queue_name    = 'test' # キュー名（sensu-client側は適当で問題ない）
subscription  = 'production' # 

EventMachine.run do
  connection = AMQP.connect({
	:host  => '<rabbitmq-serverのIPアドレス>', 
	:vhost => '/sensu', 
	:user  => 'sensu', 
	:pass  => '<sensuユーザパスワード>'
  })
  puts "Connecting to RabbitMQ. Running #{AMQP::VERSION} version of the gem..."

  ch  = AMQP::Channel.new(connection)
  
  ## 
  ## exchange宛に発信されたメッセージをlistenする
  ## 
  q   = ch.queue(queue_name, :auto_delete => true)
  q.bind(ch.method(:fanout).call(subscription))
  
  q.subscribe do |metadata, payload|
    puts "Received a message: #{payload}"
  end

  ## 以下はCtrl+C, killコマンドで安全に終了するためのもの
  Signal.trap(:INT) do
    connection.close { EM.stop }
  end
  Signal.trap(:TERM) do
    connection.close { EM.stop }
  end
end

■productionへのメッセージsubscribe結果

ruby client_listen.rb
# ----
# Connecting to RabbitMQ. Running 1.5.0 version of the gem...
# Received a message: {"name":"memory","issued":1430033390,"command":"/etc/sensu/plugins/check-mem.sh -w 128 -c 64"}
# Received a message: {"name":"cpu_metrics","issued":1430033399,"command":"/etc/sensu/plugins/cpu-metrics.rb"}
# Received a message: {"name":"memory","issued":1430033400,"command":"/etc/sensu/plugins/check-mem.sh -w 128 -c 64"}

sensu-client => sensu-serverの通信

sensu-client => sensu-serverへ送信されるメッセージは２種類ある.

keepalive
監視用コマンド(Sensu Checks)の実行結果

各sensu-clientが相手にするsensu-serverは1つと決まっているので、この方向はキュー名を指定した直接の通信になる.

sensu-clientが送信しているメッセージを傍受するスクリプト
※おもしろいのはキュー名を知っていさえすれば、rabbitmq-serverにESTABLISHEDしているどのホストからでも内容をsubscribeできる.

server_listen.rb

# !/usr/bin/env ruby

require 'amqp'

## 
## 確認したいメッセージによってresults/keepalivesを切り替えてください
## 

# sensu-clientのcheckコマンドの実行結果が返されるキュー
queue_name    = 'results'
# sensu-clientのkeepalive結果が返されるキュー
# queue_name    = 'keepalives'

EventMachine.run do
  connection = AMQP.connect({
	:host  => '<rabbitmq-serverのIPアドレス>',
	:vhost => '/sensu',
	:user  => 'sensu',
	:pass  => '<sensuユーザパスワード>'
  })
  puts "Connecting to RabbitMQ. Running #{AMQP::VERSION} version of the gem..."

  ch  = AMQP::Channel.new(connection)
  
  ## 
  ## queueを直接listenする
  ## 
  q   = ch.queue(queue_name, :auto_delete => true)
  q.subscribe do |metadata, payload|
    puts "Received a message: #{payload}"
  end

  ## 以下はCtrl+C, killコマンドで安全に終了するためのもの
  Signal.trap(:INT) do
    connection.close { EM.stop }
  end
  Signal.trap(:TERM) do
    connection.close { EM.stop }
  end
end

■keepalivesキューのsubscribe結果

ruby server_listen.rb
# ---
# Connecting to RabbitMQ. Running 1.5.0 version of the gem...
# Received a message: {"name":"sensu-client-1","address":"192.168.33.13","subscriptions":["production"],"version":"0.17.2","timestamp":1430033267}

■resultsキューのsubscribe結果

ruby server_listen.rb
# ---
# Connecting to RabbitMQ. Running 1.5.0 version of the gem...
# Received a message: {"client":"rvprov-agile","check":{"name":"disk","issued":1430033519,"command":"/etc/sensu/plugins/check-disk.rb","executed":1430033518,"duration":0.059,"output":"CheckDisk OK: All disk usage under 85% and inode usage under 85%\n","status":0}}
# Received a message: {"client":"rvprov-agile","check":{"thresholds":{"warning":120,"critical":180},"name":"keepalive","issued":1430033539,"executed":1430033539,"output":"Keepalive sent from client 11 seconds ago","status":0}}

rabbitmq通信確認スクリプトについて

rubyのrabbitmqクライアントとして、sensuではamqp gemを使っている.
CentOSの場合、sensuをインストールするとamqpは以下にインストールされた.
/opt/sensu/embedded/lib/ruby/gems/2.0.0/gems/amqp-1.5.0/
このパスを追加するか、別途gem install amqpでパスが通るところにamqpをインストールする必要がある.

References