Module: Sfp::Agent

Defined in:
lib/sfpagent/agent.rb

Defined Under Namespace

Classes: Handler

Constant Summary collapse

NetHelper = 
Object.new.extend(Nuri::Net::Helper)
CachedDir = 
(Process.euid == 0 ? '/var/sfpagent' : File.expand_path('~/.sfpagent'))
DefaultPort = 
1314
PIDFile = 
"#{CachedDir}/sfpagent.pid"
LogFile = 
"#{CachedDir}/sfpagent.log"
ModelFile = 
"#{CachedDir}/sfpagent.model"
AgentsDataFile = 
"#{CachedDir}/sfpagent.agents"
BSigFile = 
"#{CachedDir}/bsig.model"
BSigPIDFile = 
"#{CachedDir}/bsig.pid"
@@logger = 
WEBrick::Log.new(LogFile, WEBrick::BasicLog::INFO ||
WEBrick::BasicLog::ERROR ||
WEBrick::BasicLog::FATAL ||
WEBrick::BasicLog::WARN)
@@current_model_hash = 
nil
@@bsig = 
nil
@@bsig_modified_time = 
nil
@@bsig_engine =

create BSig engine instance

Sfp::BSig.new
@@runtime_lock = 
Mutex.new
@@agents_data = 
nil
@@agents_data_modified_time = 
nil

Class Method Summary collapse

Class Method Details

.bsig_engine ⇒ Object 



273
274
275
 
# File 'lib/sfpagent/agent.rb', line 273

def self.bsig_engine
  @@bsig_engine
end

.build_model(p = {}) ⇒ Object

Reload the model from cached file.



214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
 
# File 'lib/sfpagent/agent.rb', line 214

def self.build_model(p={})
  if not File.exist?(ModelFile)
    Sfp::Agent.logger.info "There is no model in cache."
  else
    begin
      @@runtime_lock.synchronize {
        data = File.read(ModelFile)
        @@current_model_hash = Digest::MD5.hexdigest(data)
        if !defined?(@@runtime) or @@runtime.nil? or p[:complete]
          @@runtime = Sfp::Runtime.new(JSON[data])
        else
          @@runtime.set_model(JSON[data])
        end
      }
      Sfp::Agent.logger.info "Reloading the model in cache [OK]"
    rescue Exception => e
      Sfp::Agent.logger.error "Reloading the model in cache [Failed] #{e}\n#{e.backtrace.join("\n")}"
    end
  end
end

.execute_action(action) ⇒ Object

 Execute an action

Parameters:

  • action —

    contains the action’s schema.



335
336
337
338
339
340
341
342
343
344
345
346
 
# File 'lib/sfpagent/agent.rb', line 335

def self.execute_action(action)
  logger = (@@config[:daemon] ? Sfp::Agent.logger : Logger.new(STDOUT))
  action_string = "#{action['name']} #{JSON.generate(action['parameters'])}"
  begin
    result = @@runtime.execute_action(action)
    logger.info "Executing #{action_string} " + (result ? "[OK]" : "[Failed]")
    return result
  rescue Exception => e
    logger.error "Executing #{action_string} [Failed] #{e}\n#{e.backtrace.join("\n")}"
  end
  false
end

.get_agents ⇒ Object 



497
498
499
500
501
 
# File 'lib/sfpagent/agent.rb', line 497

def self.get_agents
  return {} if not File.exist?(AgentsDataFile)
  return @@agents_data if File.mtime(AgentsDataFile) == @@agents_data_modified_time
  @@agents_data = JSON[File.read(AgentsDataFile)]
end

.get_bsig ⇒ Object

Return a BSig model from cached file



258
259
260
261
262
263
264
265
266
267
268
269
270
271
 
# File 'lib/sfpagent/agent.rb', line 258

def self.get_bsig
  return nil if not File.exist?(BSigFile)
  return @@bsig if File.mtime(BSigFile) == @@bsig_modified_time

  begin
    data = File.read(BSigFile)
    @@bsig = (data.length > 0 ? JSON[data] : nil)
    @@bsig_modified_time = File.mtime(BSigFile)
    return @@bsig
  rescue Exception => e
    Sfp::Agent.logger.error "Get the BSig model [Failed] #{e}\n#{e.backtrace.join("\n")}"
  end
  false
end

.get_log(n = 0) ⇒ Object 



467
468
469
470
471
472
473
474
 
# File 'lib/sfpagent/agent.rb', line 467

def self.get_log(n=0)
  return '' if not File.exist?(LogFile)
  if n <= 0
    File.read(LogFile)
  else
    `tail -n #{n} #{LogFile}`
  end
end

.get_module_hash(name) ⇒ Object 



386
387
388
389
390
391
392
393
394
395
396
397
398
 
# File 'lib/sfpagent/agent.rb', line 386

def self.get_module_hash(name)
  return nil if @@config[:modules_dir].to_s == ''

  module_dir = "#{@@config[:modules_dir]}/#{name}"
  if File.directory? module_dir
    if `which md5sum`.strip.length > 0
      return `find #{module_dir} -type f -exec md5sum {} + | awk '{print $1}' | sort | md5sum | awk '{print $1}'`.strip
    elsif `which md5`.strip.length > 0
      return `find #{module_dir} -type f -exec md5 {} + | awk '{print $4}' | sort | md5`.strip
    end
  end
  nil
end

.get_modules ⇒ Object 



400
401
402
403
404
405
 
# File 'lib/sfpagent/agent.rb', line 400

def self.get_modules
  return [] if not (defined? @@modules and @@modules.is_a? Array)
  data = {}
  @@modules.each { |m| data[m] = get_module_hash(m) }
  data
end

.get_schemata(module_name) ⇒ Object 



377
378
379
380
381
382
383
384
 
# File 'lib/sfpagent/agent.rb', line 377

def self.get_schemata(module_name)
  dir = @@config[:modules_dir]

  filepath = "#{dir}/#{module_name}/#{module_name}.sfp"
  sfp = parse(filepath).root
  sfp.accept(Sfp::Visitor::ParentEliminator.new)
  JSON.generate(sfp)
end

.get_state(as_sfp = true) ⇒ Object

Return the current state of the model.



284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/sfpagent/agent.rb', line 284

def self.get_state(as_sfp=true)
  @@runtime_lock.synchronize {
    return nil if !defined?(@@runtime) or @@runtime.nil?
    begin
      return @@runtime.get_state(as_sfp)
    rescue Exception => e
      Sfp::Agent.logger.error "Get state [Failed] #{e}\n#{e.backtrace.join("\n")}"
    end
  }
  false
end

.install_module(name, data) ⇒ Object 



432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
 
# File 'lib/sfpagent/agent.rb', line 432

def self.install_module(name, data)
  return false if @@config[:modules_dir].to_s == ''

  if !File.directory? @@config[:modules_dir]
    File.delete @@config[:modules_dir] if File.exist? @@config[:modules_dir]
    Dir.mkdir(@@config[:modules_dir], 0700)
  end

  # delete old files
  module_dir = "#{@@config[:modules_dir]}/#{name}"
  system("rm -rf #{module_dir}") if File.exist? module_dir

  # save the archive
  Dir.mkdir("#{module_dir}", 0700)
  File.open("#{module_dir}/data.tgz", 'wb', 0600) { |f| f.syswrite data }

  # extract the archive and the files
  system("cd #{module_dir}; tar xvf data.tgz")
  Dir.entries(module_dir).each { |name|
    next if name == '.' or name == '..'
    if File.directory? "#{module_dir}/#{name}"
      system("cd #{module_dir}/#{name}; mv * ..; mv .* .. 2>/dev/null; cd ..; rm -rf #{name}")
    end
    system("cd #{module_dir}; rm data.tgz")
  }
  load_modules(@@config)
  
  # rebuild the model
  build_model({:complete => true})

  Sfp::Agent.logger.info "Installing module #{name} [OK]"

  true
end

.load_modules(p = {}) ⇒ Object

 Load all modules in given directory.

options: :dir => directory that holds all modules



353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
 
# File 'lib/sfpagent/agent.rb', line 353

def self.load_modules(p={})
  dir = p[:modules_dir]

  @@modules = []
  counter = 0
  if dir != '' and File.exist?(dir)
    Sfp::Agent.logger.info "Modules directory: #{dir}"
    Dir.entries(dir).each { |name|
      next if name == '.' or name == '..' or File.file?("#{dir}/#{name}")
      module_file = "#{dir}/#{name}/#{name}.rb"
      next if not File.exist?(module_file)
      begin
        load module_file #require module_file
        Sfp::Agent.logger.info "Loading module #{dir}/#{name} [OK]"
        counter += 1
        @@modules << name
      rescue Exception => e
        Sfp::Agent.logger.warn "Loading module #{dir}/#{name} [Failed]\n#{e}"
      end
    }
  end
  Sfp::Agent.logger.info "Successfully loading #{counter} modules."
end

.logger ⇒ Object 



42
43
44
 
# File 'lib/sfpagent/agent.rb', line 42

def self.logger
  @@logger
end

.resolve(path, as_sfp = true) ⇒ Object 



296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
 
# File 'lib/sfpagent/agent.rb', line 296

def self.resolve(path, as_sfp=true)
  return Sfp::Undefined.new if !defined?(@@runtime) or @@runtime.nil? or @@runtime.root.nil?
  begin
    path = path.simplify
    _, node, _ = path.split('.', 3)
    if @@runtime.root.has_key?(node)
      # local resolve
      parent, attribute = path.pop_ref
      mod = @@runtime.root.at?(parent)
      if mod.is_a?(Hash)
        mod[:_self].update_state
        state = mod[:_self].state
        return state[attribute] if state.has_key?(attribute)
      end
      return Sfp::Undefined.new
    end

    agents = get_agents
    if agents[node].is_a?(Hash)
      # remote resolve
      agent = agents[node]
      path = path[1, path.length-1].gsub /\./, '/'
      code, data = NetHelper.get_data(agent['sfpAddress'], agent['sfpPort'], "/state#{path}")
      if code.to_i == 200
        state = JSON[data]['state']
        return Sfp::Unknown.new if state == '<sfp::unknown>'
        return state if !state.is_a?(String) or state[0,15] != '<sfp::undefined'
      end
    end
  rescue Exception => e
    Sfp::Agent.logger.error "Resolve #{path} [Failed] #{e}\n#{e.backtrace.join("\n")}"
  end
  Sfp::Undefined.new
end

.set_agents(agents) ⇒ Object 



476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
 
# File 'lib/sfpagent/agent.rb', line 476

def self.set_agents(agents)
  File.open(AgentsDataFile, 'w', 0600) do |f|
    raise Exception, "Invalid agents list." if not agents.is_a?(Hash)
    buffer = {}
    agents.each { |name,data|
      raise Exception, "Invalid agents list." if not data.is_a?(Hash) or
        not data.has_key?('sfpAddress') or data['sfpAddress'].to_s.strip == '' or
        not data.has_key?('sfpPort')
      buffer[name] = {}
      buffer[name]['sfpAddress'] = data['sfpAddress'].to_s
      buffer[name]['sfpPort'] = data['sfpPort'].to_s.strip.to_i
      buffer[name]['sfpPort'] = DefaultPort if buffer[name]['sfpPort'] == 0
    }
    f.write(JSON.generate(buffer))
    f.flush
  end
  true
end

.set_bsig(bsig) ⇒ Object

Setting a new BSig model: set @@bsig variable, and save in cached file



237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/sfpagent/agent.rb', line 237

def self.set_bsig(bsig)
  begin
    File.open(BSigFile, File::RDWR|File::CREAT, 0600) { |f|
      f.flock(File::LOCK_EX)
      Sfp::Agent.logger.info "Setting the BSig model [Wait]"
      f.rewind
      data = (bsig.nil? ? '' : JSON.generate(bsig))
      f.write(data)
      f.flush
      f.truncate(f.pos)
    }
    Sfp::Agent.logger.info "Setting the BSig model [OK]"
    return true
  rescue Exception => e
    Sfp::Agent.logger.error "Setting the BSig model [Failed] #{e}\n#{e.backtrace.join("\n")}"
  end
  false
end

.set_model(model) ⇒ Object

Save given model to cached file, and then reload the model.



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/sfpagent/agent.rb', line 184

def self.set_model(model)
  begin
    # generate MD5 hash for the new model
    data = JSON.generate(model)
    new_model_hash = Digest::MD5.hexdigest(data)

    # save the new model if it's not same with the existing one
    if Digest::MD5.hexdigest(data) != @@current_model_hash
      Sfp::Agent.logger.info "Setting new model [Wait]"
      File.open(ModelFile, File::RDWR|File::CREAT, 0600) { |f|
        f.flock(File::LOCK_EX)
        f.rewind
        f.write(data)
        f.flush
        f.truncate(f.pos)
      }
      build_model
      Sfp::Agent.logger.info "Setting the model [OK]"
    else
      #Sfp::Agent.logger.info "The model is not changed."
    end
    return true
  rescue Exception => e
    Sfp::Agent.logger.error "Setting the model [Failed] #{e}\n#{e.backtrace.join("\n")}"
  end
  false
end

.start(p = {}) ⇒ Object

Start the agent.

options: :daemon => true if running as a daemon, false if as a normal application :port :ssl :certfile :keyfile



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/sfpagent/agent.rb', line 55

def self.start(p={})
  Process.daemon

  begin
    # check modules directory, and create it if it's not exist
    p[:modules_dir] = File.expand_path(p[:modules_dir].to_s.strip != '' ? p[:modules_dir].to_s : "#{CachedDir}/modules")
    Dir.mkdir(p[:modules_dir], 0700) if not File.exist?(p[:modules_dir])
    @@config = p
Sfp::Agent.logger.info "modules dir: " + p[:modules_dir]

    # load modules from cached directory
    load_modules(p)

    # reload model
    build_model({:complete => true})

    # create web server
    server_type = (p[:daemon] ? WEBrick::Daemon : WEBrick::SimpleServer)
    port = (p[:port] ? p[:port] : DefaultPort)
    config = { :Host => '0.0.0.0',
               :Port => port,
               :ServerType => server_type,
               :pid => '/tmp/webrick.pid',
               :Logger => Sfp::Agent.logger }
    if p[:ssl]
      config[:SSLEnable] = true
      config[:SSLVerifyClient] = OpenSSL::SSL::VERIFY_NONE
      config[:SSLCertificate] = OpenSSL::X509::Certificate.new(File.open(p[:certfile]).read)
      config[:SSLPrivateKey] = OpenSSL::PKey::RSA.new(File.open(p[:keyfile]).read)
      config[:SSLCertName] = [["CN", WEBrick::Utils::getservername]]
    end
    server = WEBrick::HTTPServer.new(config)
    server.mount("/", Sfp::Agent::Handler, Sfp::Agent.logger)

    # trap signal
    ['INT', 'KILL', 'HUP'].each { |signal|
      trap(signal) {
        Sfp::Agent.logger.info "Shutting down web server"
        server.shutdown
      }
    }

    # send request to local web server to save its PID
    fork {
      sleep 0.5
      1.upto(5) do |i|
        begin
          NetHelper.get_data('127.0.0.1', config[:Port], '/pid')
          break if File.exist?(PIDFile)
        rescue
          sleep (i*i)
        end
      end
      puts "SFP Agent is running with PID #{File.read(PIDFile)}" if File.exist?(PIDFile)
    }

    # start BSig's main thread in a separate process
    fork {
      bsig_engine.enable({:mode => :main})
    }

    # enable BSig's satisfier
    bsig_engine.enable({:mode => :satisfier})

    # start web server
    server.start

  rescue Exception => e
    Sfp::Agent.logger.error "Starting the agent [Failed] #{e}\n#{e.backtrace.join("\n")}"
    raise e
  end
end

.status ⇒ Object

Print the status of the agent.



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/sfpagent/agent.rb', line 156

def self.status
  if not File.exist?(PIDFile)
    puts "SFP Agent is not running."
  else
    pid = File.read(PIDFile).to_i
    if `ps hf #{pid}`.strip =~ /.*sfpagent.*/
      puts "SFP Agent is running with PID #{pid}"
    else
      File.delete(PIDFile)
      puts "SFP Agent is not running."
    end
  end

  if not File.exist?(BSigPIDFile)
    puts "BSig engine is not running."
  else
    pid = File.read(BSigPIDFile).to_i
    if `ps hf #{pid}`.strip =~ /.*sfpagent.*/
      puts "BSig engine is running with PID #{pid}"
    else
      File.delete(BSigPIDFile)
      puts "BSig engine is not running."
    end
  end
end

.stop ⇒ Object

Stop the agent’s daemon.



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/sfpagent/agent.rb', line 130

def self.stop
  # stopping web server (main thread)
  pid = (File.exist?(PIDFile) ? File.read(PIDFile).to_i : nil)
  if not pid.nil? and `ps h #{pid}`.strip =~ /.*sfpagent.*/
    Process.kill('HUP', pid)
    puts "Stopping SFP Agent with PID #{pid}"
    File.delete(PIDFile) if File.exist?(PIDFile)
  else
    puts "SFP Agent is not running."
  end

  # stopping BSig engine
  pid_bsig = (File.exist?(BSigPIDFile) ? File.read(BSigPIDFile).to_i : nil)
  if not pid_bsig.nil? and `ps h #{pid_bsig}`.strip =~ /.*sfpagent.*/
    Process.kill('HUP', pid_bsig)
    puts "Stopping BSig engine with PID #{pid_bsig}"
    File.delete(BSigPIDFile) if File.exist?(BSigPIDFile)
  else
    puts "BSig engine is not running."
  end

  Sfp::Agent.logger.info "SFP Agent daemon has been stopped."
end

.uninstall_all_modules(p = {}) ⇒ Object 



407
408
409
410
411
412
413
414
415
416
 
# File 'lib/sfpagent/agent.rb', line 407

def self.uninstall_all_modules(p={})
  return true if @@config[:modules_dir] == ''
  if system("rm -rf #{@@config[:modules_dir]}/*")
    load_modules(@@config)
    Sfp::Agent.logger.info "Deleting all modules [OK]"
    return true
  end
  Sfp::Agent.logger.info "Deleting all modules [Failed]"
  false
end

.uninstall_module(name) ⇒ Object 



418
419
420
421
422
423
424
425
426
427
428
429
430
 
# File 'lib/sfpagent/agent.rb', line 418

def self.uninstall_module(name)
  return false if @@config[:modules_dir] == ''
  
  module_dir = "#{@@config[:modules_dir]}/#{name}"
  if File.directory?(module_dir)
    result = !!system("rm -rf #{module_dir}")
  else
    result = true
  end
  load_modules(@@config)
  Sfp::Agent.logger.info "Deleting module #{name} " + (result ? "[OK]" : "[Failed]")
  result
end

.whoami? ⇒ Boolean

Returns:

  • (Boolean) 


277
278
279
280
 
# File 'lib/sfpagent/agent.rb', line 277

def self.whoami?
  return nil if !defined?(@@runtime) or @@runtime.nil?
  @@runtime.whoami?
end