logstash-core/lib/logstash/agent.rb
# We depends on a series of task derived from the internal state and what
# need to be run, theses actions are applied to the current pipelines to converge to
# the desired state.
#
# The current actions are simple and favor composition, allowing us to experiment with different
# way to making them and also test them in isolation with the current running agent.
#
# Currently only action related to pipeline exist, but nothing prevent us to use the same logic
# for other tasks.
#
We depends on a series of task derived from the internal state and what
need to be run, theses actions are applied to the current pipelines to converge to
the desired state.

The current actions are simple and favor composition, allowing us to experiment with different
way to making them and also test them in isolation with the current running agent.

Currently only action related to pipeline exist, but nothing prevent us to use the same logic
for other tasks.
We depends on a series of task derived from the internal state and what need to be run theses actions are applied to the current pipelines to converge to the desired state . The current actions are simple and favor composition allowing us to experiment with different way to making them and also test them in isolation with the current running agent . Currently only action related to pipeline exist but nothing prevent us to use the same logic for other tasks .
def converge_state(pipeline_actions)
    logger.debug("Converging pipelines state", :actions_count => pipeline_actions.size)
    converge_result = LogStash::ConvergeResult.new(pipeline_actions.size)
    pipeline_actions.each do |action|
       The ConvergeResultadd, will accept the following values
        - boolean
        - FailedAction
        - SuccessfulAction
        - Exception
       This give us a bit more extensibility with the current startup/validation model
       that we currently have.
      with_pipelines do |pipelines|
        begin
          logger.debug("Executing action", :action => action)
            action_result = action.execute(self, pipelines)
          converge_result.add(action, action_result)
          unless action_result.successful?
            logger.error("Failed to execute action", :id => action.pipeline_id,
                        :action_type => action_result.class, :message => action_result.message,
                        :backtrace => action_result.backtrace)
          end
        rescue SystemExit => e
          converge_result.add(action, e)
        rescue Exception => e
          logger.error("Failed to execute action", :action => action, :exception => e.class.name, :message => e.message, :backtrace => e.backtrace)
          converge_result.add(action, e)
        end
      end
    end
    if logger.trace?
      logger.trace("Converge results", :success => converge_result.success?,
                   :failed_actions => converge_result.failed_actions.collect { |a, r| "id: {a.pipeline_id}, action_type: {a.class}, message: {r.message}" },
                   :successful_actions => converge_result.successful_actions.collect { |a, r| "id: {a.pipeline_id}, action_type: {a.class}" })
    end
    converge_result
  end
def converge_state ( pipeline_actions ) logger . debug ( "Converging pipelines state" , :actions_count => pipeline_actions . size ) converge_result = LogStash :: ConvergeResult . new ( pipeline_actions . size ) pipeline_actions . each do | action | with_pipelines do | pipelines | begin logger . debug ( "Executing action" , :action => action ) action_result = action . execute ( self , pipelines ) converge_result . add ( action , action_result ) unless action_result . successful? logger . error ( "Failed to execute action" , :id => action . pipeline_id , :action_type => action_result . class , :message => action_result . message , :backtrace => action_result . backtrace ) end rescue SystemExit => e converge_result . add ( action , e ) rescue Exception => e logger . error ( "Failed to execute action" , :action => action , :exception => e . class . name , :message => e . message , :backtrace => e . backtrace ) converge_result . add ( action , e ) end end end if logger . trace? logger . trace ( "Converge results" , :success => converge_result . success? , :failed_actions => converge_result . failed_actions . collect { | a , r | "id: #{a.pipeline_id}, action_type: #{a.class}, message: #{r.message}" } , :successful_actions => converge_result . successful_actions . collect { | a , r | "id: #{a.pipeline_id}, action_type: #{a.class}" } ) end converge_result end
We depends on a series of task derived from the internal state and what
need to be run, theses actions are applied to the current pipelines to converge to
the desired state.
We depends on a series of task derived from the internal state and what need to be run theses actions are applied to the current pipelines to converge to the desired state .
# We execute every task we need to converge the current state of pipelines # for every task we will record the action result, that will help us # the results of all the task will determine if the converge was successful or not # # The ConvergeResult#add, will accept the following values #  - boolean #  - FailedAction #  - SuccessfulAction #  - Exception # # This give us a bit more extensibility with the current startup/validation model # that we currently have.
pipeline_actions
<not_specific>