Class: Rookout::ComWs::Output
- Inherits:
-
Object
- Object
- Rookout::ComWs::Output
- Defined in:
- lib/rookout/com_ws/output.rb
Instance Attribute Summary collapse
-
#agent_com ⇒ Object
Returns the value of attribute agent_com.
-
#agent_id ⇒ Object
Returns the value of attribute agent_id.
Instance Method Summary collapse
- #close ⇒ Object
- #flush_messages ⇒ Object
-
#initialize ⇒ Output
constructor
A new instance of Output.
- #send_log_message(level, time, filename, lineno, text, formatted_message, arguments) ⇒ Object
- #send_output_queue_full_warning(aug_id) ⇒ Object
- #send_rule_status(rule_id, active, error) ⇒ Object
- #send_user_message(aug_id, report_id, arguments) ⇒ Object
- #send_warning(rule_id, error) ⇒ Object
- #user_messages_queue_full? ⇒ Boolean
Constructor Details
#initialize ⇒ Output
Returns a new instance of Output.
17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 |
# File 'lib/rookout/com_ws/output.rb', line 17 def initialize @agent_id = nil @agent_com = nil @skipped_aug_ids = Concurrent::Set.new @rule_status_update_bucket = TokenBucket.new Config.output_max_status_updates, Config.output_bucket_refresh_rate do Logger.instance.error "Limit reached, dropping status updates" end @user_message_bucket = TokenBucket.new Config., Config.output_bucket_refresh_rate do Logger.instance.error "Limit reached, dropping aug report messages" end @log_message_bucket = TokenBucket.new Config.output_max_log_items, Config.output_bucket_refresh_rate do 3, Time.new, __FILE__, 0, text, "Limit reached, dropping log messages", "Limit reached, dropping log messages" end Logger.instance.register_output self @closing = false end |
Instance Attribute Details
#agent_com ⇒ Object
Returns the value of attribute agent_com.
50 51 52 |
# File 'lib/rookout/com_ws/output.rb', line 50 def agent_com @agent_com end |
#agent_id ⇒ Object
Returns the value of attribute agent_id.
49 50 51 |
# File 'lib/rookout/com_ws/output.rb', line 49 def agent_id @agent_id end |
Instance Method Details
#close ⇒ Object
52 53 54 55 56 |
# File 'lib/rookout/com_ws/output.rb', line 52 def close @closing = true Logger.instance.remove_output self end |
#flush_messages ⇒ Object
142 143 144 145 |
# File 'lib/rookout/com_ws/output.rb', line 142 def return unless @agent_com @agent_com.flush end |
#send_log_message(level, time, filename, lineno, text, formatted_message, arguments) ⇒ Object
135 136 137 138 139 140 |
# File 'lib/rookout/com_ws/output.rb', line 135 def level, time, filename, lineno, text, , arguments return if @closing || !@agent_com @log_message_bucket.if_available do level, time, filename, lineno, text, , arguments end end |
#send_output_queue_full_warning(aug_id) ⇒ Object
62 63 64 65 66 67 68 69 70 71 |
# File 'lib/rookout/com_ws/output.rb', line 62 def send_output_queue_full_warning aug_id if @skipped_aug_ids.include? aug_id return end @skipped_aug_ids.add aug_id error = Processor::RookError.new Exceptions::RookOutputQueueFull.new send_rule_status aug_id, :Warning, error Logger.instance.warning "Skipping aug-\t#{aug_id} execution because the queue is full" end |
#send_rule_status(rule_id, active, error) ⇒ Object
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 |
# File 'lib/rookout/com_ws/output.rb', line 77 def send_rule_status rule_id, active, error return if @closing || !@agent_com @rule_status_update_bucket.if_available do if active == "Deleted" @skipped_aug_ids.delete? rule_id end status = Com::Rookout::RuleStatusMessage.new agent_id: @agent_id, rule_id: rule_id, active: active if error status.error = error.dumps end envelope_wrapper = EnvelopeWrapper.new status begin @agent_com.add envelope_wrapper rescue Exceptions::RookOutputQueueFull # Ignored end end end |
#send_user_message(aug_id, report_id, arguments) ⇒ Object
103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
# File 'lib/rookout/com_ws/output.rb', line 103 def aug_id, report_id, arguments return if @closing || !@agent_com @user_message_bucket.if_available do if Config.protobuf_version2 envelope_wrapper = Variant2EnvelopeWrapper.new @agent_id, aug_id, report_id, arguments else if arguments.nil? || arguments.call_method("size", "") == 0 protobuf_arguments = nil else protobuf_arguments = Processor::NamespaceSerializer.dump arguments, true end envelope_wrapper = EnvelopeWrapper.new( Com::Rookout::AugReportMessage.new( agent_id: @agent_id, aug_id: aug_id, report_id: report_id, arguments: protobuf_arguments ) ) end begin @agent_com.add envelope_wrapper @skipped_aug_ids.delete? aug_id rescue Exceptions::RookOutputQueueFull send_output_queue_full_warning aug_id end end end |
#send_warning(rule_id, error) ⇒ Object
73 74 75 |
# File 'lib/rookout/com_ws/output.rb', line 73 def send_warning rule_id, error send_rule_status rule_id, :Warning, error end |
#user_messages_queue_full? ⇒ Boolean
58 59 60 |
# File 'lib/rookout/com_ws/output.rb', line 58 def @user_message_bucket.exhausted? || (!@agent_com.nil? && @agent_com.queue_full?) end |