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
- #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 |
# 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 @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 if @settings.set?('config.field_reference.parser') # TODO: i18n logger.warn("deprecated setting `config.field_reference.parser` set; field reference parsing is strict by default") end # 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
135 136 137 |
# File 'lib/logstash/agent.rb', line 135 def auto_reload? @auto_reload end |
#converge_state_and_update ⇒ Object
147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 |
# File 'lib/logstash/agent.rb', line 147 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 |
#execute ⇒ Object
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 127 128 129 130 131 132 133 |
# File 'lib/logstash/agent.rb', line 90 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
237 238 239 |
# File 'lib/logstash/agent.rb', line 237 def get_pipeline(pipeline_id) @pipelines_registry.get_pipeline(pipeline_id) end |
#id ⇒ Object
195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 |
# File 'lib/logstash/agent.rb', line 195 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
229 230 231 |
# File 'lib/logstash/agent.rb', line 229 def id_path @id_path ||= ::File.join(settings.get("path.data"), "uuid") end |
#no_pipeline? ⇒ Boolean
269 270 271 |
# File 'lib/logstash/agent.rb', line 269 def no_pipeline? @pipelines_registry.running_pipelines.empty? end |
#non_running_pipelines ⇒ Object
249 250 251 |
# File 'lib/logstash/agent.rb', line 249 def non_running_pipelines @pipelines_registry.non_running_pipelines end |
#pipelines_count ⇒ Object
241 242 243 |
# File 'lib/logstash/agent.rb', line 241 def pipelines_count @pipelines_registry.size end |
#running? ⇒ Boolean
139 140 141 |
# File 'lib/logstash/agent.rb', line 139 def running? @running.true? end |
#running_pipelines ⇒ Object
245 246 247 |
# File 'lib/logstash/agent.rb', line 245 def running_pipelines @pipelines_registry.running_pipelines end |
#running_pipelines? ⇒ Boolean
253 254 255 |
# File 'lib/logstash/agent.rb', line 253 def running_pipelines? @pipelines_registry.running_pipelines.any? end |
#running_pipelines_count ⇒ Object
257 258 259 |
# File 'lib/logstash/agent.rb', line 257 def running_pipelines_count @pipelines_registry.running_pipelines.size end |
#running_user_defined_pipelines ⇒ Object
265 266 267 |
# File 'lib/logstash/agent.rb', line 265 def running_user_defined_pipelines @pipelines_registry.running_user_defined_pipelines end |
#running_user_defined_pipelines? ⇒ Boolean
261 262 263 |
# File 'lib/logstash/agent.rb', line 261 def running_user_defined_pipelines? @pipelines_registry.running_user_defined_pipelines.any? end |
#shutdown ⇒ Object
183 184 185 186 187 188 189 190 191 192 193 |
# File 'lib/logstash/agent.rb', line 183 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
143 144 145 |
# File 'lib/logstash/agent.rb', line 143 def stopped? @running.false? end |
#uptime ⇒ Integer
Calculate the Logstash uptime in milliseconds
179 180 181 |
# File 'lib/logstash/agent.rb', line 179 def uptime ((Time.now.to_f - STARTED_AT.to_f) * 1000.0).to_i end |