app/lib/actions/remote_execution/run_hosts_job.rb
module Actions
module RemoteExecution
class RunHostsJob < Actions::ActionWithSubPlans
include Actions::RecurringAction
include Actions::ObservableAction
include Actions::RemoteExecution::EventHelpers
middleware.use Actions::Middleware::BindJobInvocation
middleware.use Actions::Middleware::RecurringLogic
middleware.use Actions::Middleware::WatchDelegatedProxySubTasks
execution_plan_hooks.use :notify_on_success, :on => :success
execution_plan_hooks.use :notify_on_failure, :on => :failure
execution_plan_hooks.use :emit_running_event, :on => :running
class CheckOnProxyActions; end
def queue
ForemanRemoteExecution::DYNFLOW_QUEUE
end
def delay(delay_options, job_invocation)
task.add_missing_task_groups(job_invocation.task_group)
job_invocation.targeting.resolve_hosts! if job_invocation.targeting.static? && !job_invocation.targeting.resolved?
input.update :job_invocation => job_invocation.to_action_input
super delay_options, job_invocation
end
def plan(job_invocation)
job_invocation.task_group.save! if job_invocation.task_group.try(:new_record?)
task.add_missing_task_groups(job_invocation.task_group) if job_invocation.task_group
features = job_invocation.pattern_templates.flat_map { |t| t.remote_execution_features.pluck(:label) }.uniq
action_subject(job_invocation, job_features: features)
job_invocation.targeting.resolve_hosts! if job_invocation.targeting.dynamic? || !job_invocation.targeting.resolved?
set_up_concurrency_control job_invocation
input.update(:job_category => job_invocation.job_category)
plan_self(:job_invocation_id => job_invocation.id)
provider = job_invocation.pattern_template_invocations.first&.template&.provider
input[:proxy_batch_size] ||= provider&.proxy_batch_size || Setting['foreman_tasks_proxy_batch_size']
trigger_action = plan_action(Actions::TriggerProxyBatch, batch_size: proxy_batch_size, total_count: hosts.count)
input[:trigger_run_step_id] = trigger_action.run_step_id
end
def create_sub_plans
proxy_selector = RemoteExecutionProxySelector.new
current_batch.map do |host|
# composer creates just "pattern" for template_invocations because target is evaluated
# during actual run (here) so we build template invocations from these patterns
template_invocation = job_invocation.pattern_template_invocation_for_host(host).deep_clone
trigger(RunHostJob, job_invocation, host, template_invocation, proxy_selector)
end
end
def spawn_plans
super
ensure
trigger_remote_batch
end
def trigger_remote_batch
remaining = output[:planned_count] - output[:remote_triggered_count]
return if remaining.zero?
batches_ready = remaining / proxy_batch_size
if concurrency_limit
count = remaining
else
return unless batches_ready > 0
count = proxy_batch_size * batches_ready
end
batches_ready = [1, batches_ready].max
plan_event(Actions::TriggerProxyBatch::TriggerNextBatch[batches_ready], nil, step_id: input[:trigger_run_step_id])
output[:remote_triggered_count] += count
end
def on_planning_finished
plan_event(Actions::TriggerProxyBatch::TriggerLastBatch, nil, step_id: input[:trigger_run_step_id])
super
end
def finalize
job_invocation.password = job_invocation.key_passphrase = job_invocation.effective_user_password = nil
job_invocation.save!
Rails.logger.debug "cleaning cache for keys that begin with 'job_invocation_#{job_invocation.id}'"
Rails.cache.delete_matched(cache_deletion_query(job_invocation.id))
end
def notify_on_success(_plan)
job_invocation.build_notification.deliver!
if [RexMailNotification::SUCCEEDED_JOBS, RexMailNotification::ALL_JOBS].include?(mail_notification_preference&.interval)
RexJobMailer.job_finished(job_invocation, subject: _("REX job has succeeded - %s") % job_invocation.to_s).deliver_now
end
end
def notify_on_failure(_plan)
job_invocation.build_notification.deliver!
if [RexMailNotification::FAILED_JOBS, RexMailNotification::ALL_JOBS].include?(mail_notification_preference&.interval)
RexJobMailer.job_finished(job_invocation, subject: _("REX job has failed - %s") % job_invocation.to_s).deliver_now
end
end
def job_invocation_id
input[:job_invocation_id] || input.fetch(:job_invocation, {})[:id]
end
def job_invocation
@job_invocation ||= JobInvocation.find(job_invocation_id)
end
def batch(from, size)
hosts.offset(from).limit(size)
end
def initiate
output[:host_count] = total_count
output[:remote_triggered_count] = 0
super
end
def total_count
# For compatibility with already existing tasks
return output[:total_count] || hosts.count unless output.has_key?(:host_count) || task.pending?
output[:host_count] || hosts.count
end
def hosts
job_invocation.targeting.hosts.order("#{TargetingHost.table_name}.id")
end
def set_up_concurrency_control(invocation)
limit_concurrency_level! invocation.concurrency_level unless invocation.concurrency_level.nil?
end
def rescue_strategy
::Dynflow::Action::Rescue::Skip
end
def run(event = nil)
if event == Dynflow::Action::Skip
plan_event(Dynflow::Action::Skip, nil, step_id: input[:trigger_run_step_id])
else
super
end
end
def humanized_input
input.fetch(:job_invocation, {}).fetch(:description, '')
end
def humanized_name
'%s:' % _(super)
end
def proxy_batch_size
input[:proxy_batch_size]
end
def self.event_states
[:success, :failure, :running]
end
def emit_running_event(plan)
emit_event(plan, :running)
end
private
def mail_notification_preference
UserMailNotification.where(mail_notification_id: RexMailNotification.first, user_id: User.current.id).first
end
def cache_deletion_query(job_invocation_id)
return "#{JobInvocation::CACHE_PREFIX}_#{job_invocation_id}*" if Rails.cache.kind_of? ActiveSupport::Cache::RedisCacheStore
/\A#{JobInvocation::CACHE_PREFIX}_#{job_invocation_id}/
end
extend ApipieDSL::Class
apipie :class, "An action representing execution of a job against a set of hosts" do
name 'Actions::RemoteExecution::RunHostsJob'
refs 'Actions::RemoteExecution::RunHostsJob'
sections only: %w[all webhooks]
property :task, object_of: 'Task', desc: 'Returns the task to which this action belongs'
property :job_invocation_id, Integer, desc: "Returns the id of the job invocation"
property :job_invocation, object_of: 'JobInvocation', desc: "Returns the job invocation"
end
class Jail < ::Actions::ObservableAction::Jail
allow :job_invocation_id, :job_invocation
end
end
end
end