Class: LogStash::Agent
- Inherits:
-
Object
- Object
- LogStash::Agent
- Includes:
- Util::Loggable
- Defined in:
- lib/logstash/agent.rb
Constant Summary collapse
- STARTED_AT =
Time.now.freeze
Instance Attribute Summary collapse
-
#dispatcher ⇒ Object
readonly
Returns the value of attribute dispatcher.
-
#ephemeral_id ⇒ Object
readonly
Returns the value of attribute ephemeral_id.
-
#logger ⇒ Object
Returns the value of attribute logger.
-
#metric ⇒ Object
readonly
Returns the value of attribute metric.
-
#name ⇒ Object
readonly
Returns the value of attribute name.
-
#pipeline_bus ⇒ Object
readonly
Returns the value of attribute pipeline_bus.
-
#settings ⇒ Object
readonly
Returns the value of attribute settings.
Instance Method Summary collapse
- #auto_reload? ⇒ Boolean
- #converge_state_and_update ⇒ Object
- #exclusive(&block) ⇒ Object
- #execute ⇒ Object
-
#get_pipeline(pipeline_id) ⇒ Object
Backward compatibility proxies to the PipelineRegistry.
- #id ⇒ Object
- #id_path ⇒ Object
-
#initialize(settings = LogStash::SETTINGS, source_loader = nil) ⇒ Agent
constructor
initialize method for LogStash::Agent.
- #no_pipeline? ⇒ Boolean
- #non_running_pipelines ⇒ Object
- #pipelines_count ⇒ Object
- #running? ⇒ Boolean
- #running_pipelines ⇒ Object
- #running_pipelines? ⇒ Boolean
- #running_pipelines_count ⇒ Object
- #running_user_defined_pipelines ⇒ Object
- #running_user_defined_pipelines? ⇒ Boolean
- #shutdown ⇒ Object
- #stopped? ⇒ Boolean
-
#uptime ⇒ Integer
Calculate the Logstash uptime in milliseconds.
Constructor Details
#initialize(settings = LogStash::SETTINGS, source_loader = nil) ⇒ Agent
initialize method for LogStash::Agent
31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 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 |
# File 'lib/logstash/agent.rb', line 31 def initialize(settings = LogStash::SETTINGS, source_loader = nil) @logger = self.class.logger @settings = settings @auto_reload = setting("config.reload.automatic") @ephemeral_id = SecureRandom.uuid # Mutex to synchonize in the exclusive method # Initial usage for the Ruby pipeline initialization which is not thread safe @exclusive_lock = Mutex.new @webserver_control_lock = Mutex.new @convergence_lock = Mutex.new # Special bus object for inter-pipelines communications. Used by the `pipeline` input/output @pipeline_bus = org.logstash.plugins.pipeline.PipelineBus.new @pipelines_registry = LogStash::PipelinesRegistry.new @name = setting("node.name") @http_host = setting("http.host") @http_port = setting("http.port") @http_environment = setting("http.environment") # Generate / load the persistent uuid id # Set the global FieldReference parsing mode parsing_mode = case setting('config.field_reference.parser') when 'COMPAT' then org.logstash.FieldReference::ParsingMode::COMPAT; when 'LEGACY' then org.logstash.FieldReference::ParsingMode::LEGACY; when 'STRICT' then org.logstash.FieldReference::ParsingMode::STRICT; else fail('Unsupported FieldReference parsing mode') end logger.debug("Setting global FieldReference parsing mode: #{parsing_mode}") org.logstash.FieldReference::set_parsing_mode(parsing_mode) # This is for backward compatibility in the tests if source_loader.nil? @source_loader = LogStash::Config::SourceLoader.new @source_loader.add_source(LogStash::Config::Source::Local.new(@settings)) else @source_loader = source_loader end # Normalize time interval to seconds @reload_interval = setting("config.reload.interval") / 1_000_000_000.0 @collect_metric = setting("metric.collect") # Create the collectors and configured it with the library configure_metrics_collectors @state_resolver = LogStash::StateResolver.new(metric) @pipeline_reload_metric = metric.namespace([:stats, :pipelines]) @instance_reload_metric = metric.namespace([:stats, :reloads]) initialize_agent_metrics @dispatcher = LogStash::EventDispatcher.new(self) LogStash::PLUGIN_REGISTRY.hooks.register_emitter(self.class, dispatcher) dispatcher.fire(:after_initialize) @running = Concurrent::AtomicBoolean.new(false) end |
Instance Attribute Details
#dispatcher ⇒ Object (readonly)
Returns the value of attribute dispatcher.
23 24 25 |
# File 'lib/logstash/agent.rb', line 23 def dispatcher @dispatcher end |
#ephemeral_id ⇒ Object (readonly)
Returns the value of attribute ephemeral_id.
23 24 25 |
# File 'lib/logstash/agent.rb', line 23 def ephemeral_id @ephemeral_id end |
#logger ⇒ Object
Returns the value of attribute logger.
24 25 26 |
# File 'lib/logstash/agent.rb', line 24 def logger @logger end |
#metric ⇒ Object (readonly)
Returns the value of attribute metric.
23 24 25 |
# File 'lib/logstash/agent.rb', line 23 def metric @metric end |
#name ⇒ Object (readonly)
Returns the value of attribute name.
23 24 25 |
# File 'lib/logstash/agent.rb', line 23 def name @name end |
#pipeline_bus ⇒ Object (readonly)
Returns the value of attribute pipeline_bus.
23 24 25 |
# File 'lib/logstash/agent.rb', line 23 def pipeline_bus @pipeline_bus end |
#settings ⇒ Object (readonly)
Returns the value of attribute settings.
23 24 25 |
# File 'lib/logstash/agent.rb', line 23 def settings @settings end |
Instance Method Details
#auto_reload? ⇒ Boolean
144 145 146 |
# File 'lib/logstash/agent.rb', line 144 def auto_reload? @auto_reload end |
#converge_state_and_update ⇒ Object
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 181 182 183 |
# File 'lib/logstash/agent.rb', line 156 def converge_state_and_update results = @source_loader.fetch unless results.success? if auto_reload? logger.debug("Could not fetch the configuration to converge, will retry", :message => results.error, :retrying_in => @reload_interval) return else raise "Could not fetch the configuration, message: #{results.error}" end end converge_result = resolve_actions_and_converge_state(results.response) update_metrics(converge_result) logger.info( "Pipelines running", :count => running_pipelines.size, :running_pipelines => running_pipelines.keys, :non_running_pipelines => non_running_pipelines.keys ) if converge_result.success? && converge_result.total > 0 dispatch_events(converge_result) converge_result rescue => e logger.error("An exception happened when converging configuration", :exception => e.class, :message => e., :backtrace => e.backtrace) end |
#exclusive(&block) ⇒ Object
95 96 97 |
# File 'lib/logstash/agent.rb', line 95 def exclusive(&block) @exclusive_lock.synchronize { block.call } end |
#execute ⇒ Object
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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
# File 'lib/logstash/agent.rb', line 99 def execute @thread = Thread.current # this var is implicitly used by Stud.stop? logger.debug("Starting agent") transition_to_running converge_state_and_update start_webserver if auto_reload? # `sleep_then_run` instead of firing the interval right away Stud.interval(@reload_interval, :sleep_then_run => true) do # TODO(ph) OK, in reality, we should get out of the loop, but I am # worried about the implication of that change so instead when we are stopped # we don't converge. # # Logstash currently expect to be block here, the signal will force a kill on the agent making # the agent thread unblock # # Actually what we really need is one more state: # # init => running => stopping => stopped converge_state_and_update unless stopped? end else # exit with error status if the initial converge_state_and_update did not create any pipeline return 1 if @pipelines_registry.empty? while !Stud.stop? # exit if all pipelines are terminated and none are reloading break if no_pipeline? # exit if there are no user defined pipelines (not system pipeline) and none are reloading break if !running_user_defined_pipelines? sleep(0.5) end end return 0 ensure transition_to_stopped end |
#get_pipeline(pipeline_id) ⇒ Object
Backward compatibility proxies to the PipelineRegistry
246 247 248 |
# File 'lib/logstash/agent.rb', line 246 def get_pipeline(pipeline_id) @pipelines_registry.get_pipeline(pipeline_id) end |
#id ⇒ Object
204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 |
# File 'lib/logstash/agent.rb', line 204 def id return @id if @id uuid = nil if ::File.exists?(id_path) begin uuid = ::File.open(id_path) {|f| f.each_line.first.chomp } rescue => e logger.warn("Could not open persistent UUID file!", :path => id_path, :error => e., :class => e.class.name) end end if !uuid uuid = SecureRandom.uuid logger.info("No persistent UUID file found. Generating new UUID", :uuid => uuid, :path => id_path) begin ::File.open(id_path, 'w') {|f| f.write(uuid) } rescue => e logger.warn("Could not write persistent UUID file! Will use ephemeral UUID", :uuid => uuid, :path => id_path, :error => e., :class => e.class.name) end end @id = uuid end |
#id_path ⇒ Object
238 239 240 |
# File 'lib/logstash/agent.rb', line 238 def id_path @id_path ||= ::File.join(settings.get("path.data"), "uuid") end |
#no_pipeline? ⇒ Boolean
278 279 280 |
# File 'lib/logstash/agent.rb', line 278 def no_pipeline? @pipelines_registry.running_pipelines.empty? end |
#non_running_pipelines ⇒ Object
258 259 260 |
# File 'lib/logstash/agent.rb', line 258 def non_running_pipelines @pipelines_registry.non_running_pipelines end |
#pipelines_count ⇒ Object
250 251 252 |
# File 'lib/logstash/agent.rb', line 250 def pipelines_count @pipelines_registry.size end |
#running? ⇒ Boolean
148 149 150 |
# File 'lib/logstash/agent.rb', line 148 def running? @running.true? end |
#running_pipelines ⇒ Object
254 255 256 |
# File 'lib/logstash/agent.rb', line 254 def running_pipelines @pipelines_registry.running_pipelines end |
#running_pipelines? ⇒ Boolean
262 263 264 |
# File 'lib/logstash/agent.rb', line 262 def running_pipelines? @pipelines_registry.running_pipelines.any? end |
#running_pipelines_count ⇒ Object
266 267 268 |
# File 'lib/logstash/agent.rb', line 266 def running_pipelines_count @pipelines_registry.running_pipelines.size end |
#running_user_defined_pipelines ⇒ Object
274 275 276 |
# File 'lib/logstash/agent.rb', line 274 def running_user_defined_pipelines @pipelines_registry.running_user_defined_pipelines end |
#running_user_defined_pipelines? ⇒ Boolean
270 271 272 |
# File 'lib/logstash/agent.rb', line 270 def running_user_defined_pipelines? @pipelines_registry.running_user_defined_pipelines.any? end |
#shutdown ⇒ Object
192 193 194 195 196 197 198 199 200 201 202 |
# File 'lib/logstash/agent.rb', line 192 def shutdown # Since we're shutting down we need to shutdown the DAG of pipelines that are talking to each other # in order of dependency. pipeline_bus.setBlockOnUnlisten(true) stop_collecting_metrics stop_webserver transition_to_stopped converge_result = shutdown_pipelines converge_result end |
#stopped? ⇒ Boolean
152 153 154 |
# File 'lib/logstash/agent.rb', line 152 def stopped? @running.false? end |
#uptime ⇒ Integer
Calculate the Logstash uptime in milliseconds
188 189 190 |
# File 'lib/logstash/agent.rb', line 188 def uptime ((Time.now.to_f - STARTED_AT.to_f) * 1000.0).to_i end |