Certains le savent peut-être, je travaille depuis quelques mois sur mon projet personnel Nodecast. Pour résumer, ce projet a l’ambition de proposer un outil de monitoring simple à mettre en œuvre mais aussi un outil de recensement façon Linux counter. Il n’a cependant pas pour objectif de concurrencer un logiciel de type Nagios. Outre le challenge du développement de la partie web, il y a également celui du client desktop en Qt, mais qui fera peut-être l’objet d’un futur billet.
Lors d’une précédente expérience professionnelle en 2007 (AF83) j’avais mis en oeuvre des techniques de répartition de charge via des traitements asynchrones pour le développement d’un microblog/chat web. Je cherchais depuis à réutiliser ces technologies, ce qui m’a poussé au développement de Nodecast.
Or si à l’époque le domaine des serveurs de message queues était encore balbutiant, il a nettement évolué depuis. En effet à l’époque après en avoir testé quelques uns, j’avais fini par me résoudre à utiliser le protocole XMPP via un serveur Jabber. Il faut bien avouer que la mise en œuvre a été plutôt ardu. XMPP est un bon protocole mais il est au final peu adapté pour un simple système de file d’attente, trop verbeux et la librairie Ruby XMPP4R faiblarde, à l’époque en tout cas.
Depuis les serveurs de file d’attente ont poussé comme des champignons, et certains sont même dérivés de serveur type clé/valeur. J’ai pour ma part choisi Gearman qui me parait une bonne technologie depuis sa réécriture en C, et c’est un bon compromis fonctionnalités / simplicité / performances. Avant d’aller plus loin je préfère présenter le schéma de l’architecture du site, ce qui rendra plus aisé les explications.
Architecture
La partie droite en orange représente le site web. Nginx transmet les requêtes HTTP vers un pool de services Thin qui est un serveur web applicatif, il est en charge d’exécuter l’application en Ruby on Rails. Pour des raisons de performances et efficacité, j’ai choisi d’utiliser une base de données NoSQL, mongoDB.
La partie intéressante est celle à gauche représentée par le nuage gris. En effet il représente les services en charge de l’API.
API asynchrone
L’API est ici une API REST. Le client en Qt effectue donc simplement des requêtes HTTP afin de communiquer avec le service web. Il fait des POST pour l’ajout de données, des PUT pour la modification et des GET pour la consultation.
Plus le client envoi des données sur un court délai, plus les statistiques seront détaillées. Le problème est que cela génère une multitude de connexions et donc de traitement. Et plus il y en a, plus la réponse au client sera longue, ce qui va dégrader la qualité de service. Bien entendu la solution est de rendre asynchrone les traitements afin que la réponse au client soit la plus rapide possible.
Cela est représenté par l’échange 1 et 2 entre le serveur Nginx et le serveur thin.
Client <-> serveur
Lors d’un POST, le client va demander la création d’une donnée au service. Ici il s’agit de la création d’une nouvelle machine (host) à monitorer. Le workflow est le suivant :
- client Qt POST les datas
- serveur sinatra authentifie le client
- serveur sinatra génère un identifiant unique (UUID)
- serveur sinatra sérialize les datas, et transmet la charge à Gearman dans la file d’attente du dispatcher
- serveur sinatra renvoi au client un XML contenant l’identifiant unique
- le client stocke cet identifiant et l’utilisera pour tous les prochains échanges
Le principe ici est que le serveur sinatra ne fasse que le strict minimum, afin de répondre le plus rapidement et d’être disponible pour une prochaine requête. Ainsi même sous une charge importante, les requêtes transmises par les clients Qt, seront empilé et mise en attente au chaud dans le serveur de file Gearman.
Pour une requête PUT le workflow est plus simple puisque l’étape 3 est supprimée. Le serveur sinatra renvoi dans tous les cas un XML contenant le status “proceed” afin de signaler au client que sa requête a été prise en compte.
L’intérêt de Sinatra est qu’il est très simple à mettre en oeuvre ce qui en fait à mon avis un candidat idéal pour servir une API. Le code du serveur tiens d’ailleurs dans un seul fichier :
#!/usr/bin/env ruby
require 'rubygems'
require 'gearman'
require "bundler"
Bundler.setup
Bundler.require(:default)
RAILS_ENV="production"
#Gearman::Util.debug = true
SERVERS = ['localhost:4730']
@@logger = Logger.new('log/server.log', 'daily')
@@logger.debug("Created logger")
File.open(File.join('../config/database.mongo.yml'), 'r') do |f|
@settings = YAML.load(f)[RAILS_ENV]
end
Mongoid.configure do |config|
name = @settings["database"]
host = @settings["host"]
config.use_object_ids = @settings["use_object_ids"]
@@logger.info "database : #{name}"
@@logger.info "host : #{host}"
config.master = Mongo::Connection.new.db(name)
# config.slaves = [
# Mongo::Connection.new(host, @settings["slave_one"]["port"], :slave_ok => true).db(name)
# ]
end
require 'models_mongoid/user.rb'
require 'models_mongoid/profil.rb'
require 'models_mongoid/host.rb'
set :logging, true
helpers do
def protected!
unless authorized?
response['WWW-Authenticate'] = %(Basic realm="Nodecast HTTP Auth")
throw(:halt, [401, "Not authorized\n"])
end
end
def authorized?
@auth ||= Rack::Auth::Basic::Request.new(request.env)
@current_user = User.where(:email => @auth.credentials.first, :authentication_token => @auth.credentials.last).first
@auth.provided? && @auth.basic? && @current_user
end
end
post '/hosts.xml' do
protected!
xml = Crack::XML.parse(request.body.read)
@@logger.info("#{Time.now} : RECEIVE CREATE")
uuid = UUIDTools::UUID.timestamp_create.to_s
host = {
:user => @current_user.email,
:uuid => uuid,
:timestamp => Time.now.utc,
:datas => xml
}
payload('dispatcher_add', host)
builder do |xml|
xml.instruct!
xml.host do
xml.uuid host[:uuid]
end
end
end
put '/host/update/:id' do
protected!
xml = Crack::XML.parse(request.body.read)
@@logger.info("#{Time.now} : RECEIVE UPDATE")
host = {
:user => @current_user.email,
:uuid => params[:id],
:timestamp => Time.now.utc,
:datas => xml
}
payload('dispatcher_update', host)
builder do |xml|
xml.instruct!
xml.host do
xml.status "proceed"
end
end
end
private
def payload(worker, data)
client = Gearman::Client.new(SERVERS)
taskset = Gearman::TaskSet.new(client)
#task = Gearman::Task.new('update', Marshal.dump(host), :background => true, :poll_status_interval => 1)
task = Gearman::Task.new(worker, Marshal.dump(data))
task.on_complete {|d| @@logger.info "complete : #{d}" }
task.on_warning {|w| @@logger.info "[client] warn: #{w}" }
task.on_fail {|f| @@logger.info "[client] calculation failed : #{f}" }
taskset.add_task(task)
end
On voit bien ici que “post ‘/hosts.xml’ do” et ‘put ‘/host/update/:id’ do” permettent de répondre très simplement aux requêtes POST et PUT des clients. Pour le reste on voit bien que le serveur ne fait que préparer un Hash avec l’XML transmis par le client, un timestamp, l’uuid du host et l’id de l’utilisateur. Ce Hast est ensuite sérialisé puis la charge est envoyée dans la file d’attente du dispatcher.
Dispatcher <-> workers
Comme on le voit dans le code les traitements sont envoyés dans la file d’attente sur 2 canaux : “dispatcher_update” et “dispatcher_add”. Il y a donc un processus Ruby qui attend des traitements sur ces canaux afin de les préparer puis les transmettre à chaque worker. J’avais tout d’abord un seul worker, mais il est beaucoup plus intéressant de le découper en plusieurs workers spécifiques à une tâche. En effet des utilisateurs peuvent décider de ne pas envoyer les informations relatives au CPU ou bien au réseau. De fait les worker auront la charge qui correspond au contenu des requêtes transmises.
Le dispatcher ne dépend pas des traitements à effectuer, son rôle est de découper le traitement en de multiples sous-traitement qu’il transmet dans le canal de chaque worker. Si les traitements sont lourds il peut malgré tout continuer à les répartir dans chacun des canaux, quelque soit la charge en cours.
Les traitements se font réellement dans chaque worker. Ils désérialisent les données reçues de leur file d’attente, puis les stocke dans mongodb. Voici un des workers dédié au traitement des load :
#!/usr/bin/env ruby
require 'rubygems'
gem 'mongoid', '1.9.0'
require 'mongoid'
gem 'gearman-ruby', '3.0.1'
require 'gearman'
require 'uuidtools'
require 'logger'
require 'yaml'
require 'optparse'
require "pp"
options = {}
optparse = OptionParser.new do |opts|
opts.on('-w', '--work WORK', 'path to the work directory') do |work|
options[:work] = work
end
opts.on('-m', '--mongo MONGO', 'path to the mongo file') do |mongo|
options[:mongo] = mongo
end
opts.on('-e', '--env ENV', 'rails environment') do |env|
options[:env] = env
end
end
begin
optparse.parse!
mandatory = [:work, :mongo, :env]
missing = mandatory.select{ |param| options[param].nil? }
if not missing.empty?
puts "Missing options: #{missing.join(', ')}"
puts optparse
exit
end
rescue OptionParser::InvalidOption, OptionParser::MissingArgument
puts $!.to_s
puts optparse
exit
end
puts "Performing task with options: #{options.inspect}"
#Gearman::Util.debug = true if options[:env] == "development"
servers = ['localhost:4730']
@@worker = Gearman::Worker.new(servers)
logger = Logger.new("#{options[:work]}/log/worker_stats_load.log", 'daily')
logger.debug("Created logger")
File.open(File.join("#{options[:mongo]}/database.mongo.yml"), 'r') do |f|
@settings = YAML.load(f)[options[:env]]
end
Mongoid.configure do |config|
name = @settings["database"]
host = @settings["host"]
config.use_object_ids = @settings["use_object_ids"]
logger.info "database : #{name}"
logger.info "host : #{host}"
config.master = Mongo::Connection.new.db(name)
# config.slaves = [
# Mongo::Connection.new(host, @settings["slave_one"]["port"], :slave_ok => true).db(name)
# ]
end
require "#{options[:work]}/models_mongoid/user.rb"
require "#{options[:work]}/models_mongoid/profil.rb"
require "#{options[:work]}/models_mongoid/host.rb"
require "#{options[:work]}/models_mongoid/osystem.rb"
require "#{options[:work]}/models_mongoid/host_ram.rb"
require "#{options[:work]}/models_mongoid/host_cpu.rb"
require "#{options[:work]}/models_mongoid/host_network.rb"
require "#{options[:work]}/models_mongoid/host_last_comment.rb"
require "#{options[:work]}/models_mongoid/host_stats_uptime.rb"
require "#{options[:work]}/models_mongoid/host_stats_load.rb"
require "#{options[:work]}/models_mongoid/host_stats_network.rb"
require "#{options[:work]}/models_mongoid/host_stats_cpu.rb"
require "#{options[:work]}/models_mongoid/host_stats_memory.rb"
require "#{options[:work]}/models_mongoid/load_statistic.rb"
########## JOB UPDATE STAT ############
@@worker.add_ability('update_load') do |data,job|
dump = Marshal.load(data)
xml = dump[:push]
host = Host.where(:uuid => dump[:uuid]).first
host_load = {
:created_at => dump[:timestamp],
:updated_at => dump[:timestamp],
:loadavg0 => xml[:loadavg0],
:loadavg1 => xml[:loadavg1],
:loadavg2 => xml[:loadavg2]
}
begin
ls = host.load_statistics.create(host_load)
logger.info "LOAD stats created"
loadavg0 = 0.0
loadavg1 = 0.0
loadavg2 = 0.0
host.load_statistics.each do |stat|
loadavg0 += stat.loadavg0
loadavg1 += stat.loadavg1
loadavg2 += stat.loadavg2
end
if !host.stats_load
host.create_stats_load(
:created_at => dump[:timestamp],
:updated_at => dump[:timestamp],
:number => 1,
:loadavg0 => ls.loadavg0,
:loadavg1 => ls.loadavg1,
:loadavg2 => ls.loadavg2,
:max0 => ls.loadavg0,
:max1 => ls.loadavg1,
:max2 => ls.loadavg2
)
else
host.stats_load.updated_at = dump[:timestamp]
host.stats_load.number += 1
host.stats_load.max0 = ls.loadavg0 if ls.loadavg0 > host.stats_load.max0
host.stats_load.max1 = ls.loadavg1 if ls.loadavg1 > host.stats_load.max1
host.stats_load.max2 = ls.loadavg2 if ls.loadavg2 > host.stats_load.max2
host.stats_load.average0 = loadavg0 / host.stats_load.number
host.stats_load.average1 = loadavg1 / host.stats_load.number
host.stats_load.average2 = loadavg2 / host.stats_load.number
host.stats_load.loadavg0 = ls.loadavg0
host.stats_load.loadavg1 = ls.loadavg1
host.stats_load.loadavg2 = ls.loadavg2
host.save
end
logger.info "Embedded host load stats updated"
rescue => e
logger.info "failed on update : #{e}"
raise Exception.new("failed on update : #{e}")
end
end
loop do
@@worker.work
end
Avantages
En cas de soucis sur les workers, ceux-ci pourront être stoppé sans problème même en production. Grâce au découplage mis en place, les traitements sont tous simplement en attente dans leur file d’attente. Il y a intérêt à avoir un serveur d’API très basique (ici le code dans Sinatra) afin d’éviter au maximum les éventuels plantages ou corruption des données, et de plus cela le rend, comme je l’ai déjà dis, plus rapide.
Evolution
Cette architecture est encore basique car ne tourne que sur une seule machine. Cependant les technologies employées permettront de la répartir sur plusieurs machines très simplement. Tout d’abord au niveau de mongoDB via des slaves. Puis en démarrant plusieurs démons Gearman. Les workers pourront ainsi dépiler leurs jobs sur l’un des serveurs Gearman du cluster.
Ensuite il pourra être intéressant que chaque worker log dans mongoDB ses tâches et ses temps de traitement afin de détecter les éventuels bottleneck. Gearman possède également une multitude d’options comme pouvoir rendre prioritaire un job.
Enfin il est bien entendu approprié que le front web puisse utiliser l’API afin d’effectuer des traitements lourds demandés par l’utilisateur.
