Class: SoarAuditingProvider::AuditingWorker

Inherits:
SoarThreadWorker::ThreadWorker
  • Object
show all
Defined in:
lib/soar_auditing_provider/auditing_worker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ AuditingWorker

Returns a new instance of AuditingWorker.



11
12
13
14
15
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 11

def initialize
  @queue = Queue.new
  @start_mutex = Mutex.new
  initialize_metrics
end

Instance Attribute Details

#failed_audit_attempts ⇒ Object (readonly)

Returns the value of attribute failed_audit_attempts.



5
6
7
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 5

def failed_audit_attempts
  @failed_audit_attempts
end

#latest_audit_timespan ⇒ Object (readonly)

Returns the value of attribute latest_audit_timespan.



8
9
10
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 8

def latest_audit_timespan
  @latest_audit_timespan
end

#latest_audit_timestamp ⇒ Object (readonly)

Returns the value of attribute latest_audit_timestamp.



9
10
11
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 9

def latest_audit_timestamp
  @latest_audit_timestamp
end

#successful_audits ⇒ Object (readonly)

Returns the value of attribute successful_audits.



6
7
8
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 6

def successful_audits
  @successful_audits
end

#total_audits ⇒ Object (readonly)

Returns the value of attribute total_audits.



7
8
9
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 7

def total_audits
  @total_audits
end

Instance Method Details

#configure(queue_worker_configuration:, auditor_audit_method:) ⇒ Object 



17
18
19
20
21
22
23
24
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 17

def configure(queue_worker_configuration: ,auditor_audit_method: )
  validate_configuration(queue_worker_configuration)
  @maximum_queue_size = queue_worker_configuration['queue_size'].to_i
  @initial_back_off_in_seconds = queue_worker_configuration['initial_back_off_in_seconds'].to_i
  @back_off_multiplier = queue_worker_configuration['back_off_multiplier'].to_i
  @maximum_back_off_attempts = queue_worker_configuration['back_off_attempts'].to_i
  @auditor_audit_method = auditor_audit_method
end

#enqueue(level, data) ⇒ Object 



26
27
28
29
30
31
32
33
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 26

def enqueue(level, data)
  if @queue.size < @maximum_queue_size then
    @queue.push({:level => level, :data => data})
  else
    raise AuditingOverflowError
  end
  ensure_worker_is_running
end

#execute ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 44

def execute
  audit_event = @queue.pop
  @total_audits += 1
  failed_before = false
  begin
    if @stopping
      @queue.push(audit_event) if audit_event #push the event back into the queue so that fallback flush mechanism can deal with this audit event
      return true #indicates to thread worder that we are done executing since we are in the process of stopping
    end
    exponential_back_off(start_at_last_attempt: failed_before) {
      time_before_audit = Time.now
      @auditor_audit_method.call(audit_event[:level],audit_event[:data])
      @latest_audit_timespan = Time.now - time_before_audit
      @successful_audits += 1
    }
  rescue Exception => e
    print_exception_with_message_to_stderr(nil,e)
    failed_before = true
    retry
  end
  @latest_audit_timestamp = Time.now.utc
  return false #indicates to thread worder that we are not done executing
end

#flush(timeout = 1) ⇒ Object 



68
69
70
71
72
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 68

def flush(timeout = 1)
  ensure_worker_is_running
  wait_for_worker_to_clear_queue(timeout)
  fallback_flush_to_stderr if @queue.size > 0
end

#start(verbose: false) ⇒ Object 



35
36
37
38
39
40
41
42
 
# File 'lib/soar_auditing_provider/auditing_worker.rb', line 35

def start(verbose: false)
  @start_mutex.synchronize {
    if not running? then
      super()
      $stderr.puts("Auditing worker was not running and respawned") if verbose
    end
  }
end