class Proxy::RemoteExecution::Ssh::Actions::PullScript
Constants
- DELIVERED
The host has picked up the job
- JobDelivered
- NOTIFIED
The host was notified over
MQTT
at least once- PickupTimeout
- READY_FOR_PICKUP
The proxy has the job stored in its job storage
- RUNNING
We received at least one output from the host
Public Instance Methods
cleanup(_plan = nil)
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 67 def cleanup(_plan = nil) job_storage.drop_job(execution_plan_id, run_step_id) Proxy::Dynflow::OtpManager.passwords.delete(execution_plan_id) Proxy::RemoteExecution::Ssh::MQTT::Dispatcher.instance.done(input[:job_uuid]) end
host_name()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 179 def host_name alternative_names = input.fetch(:alternative_names, {}) alternative_names[:consumer_uuid] || alternative_names[:fqdn] || input[:hostname] end
init_run()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 50 def init_run otp_password = if input[:with_mqtt] ::Proxy::Dynflow::OtpManager.generate_otp(execution_plan_id) end plan_event(PickupTimeout, input[:time_to_pickup], optional: true) if input[:time_to_pickup] input[:job_uuid] = job_storage.store_job(host_name, execution_plan_id, run_step_id, input[:script].tr("\r", ''), effective_user: input[:effective_user]) output[:state] = READY_FOR_PICKUP output[:result] = [] mqtt_start(otp_password) if input[:with_mqtt] suspend end
job_storage()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 191 def job_storage Proxy::RemoteExecution::Ssh.job_storage end
kill_run(fail_msg = 'The job was cancelled by the user')
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 121 def kill_run(fail_msg = 'The job was cancelled by the user') continuous_output = Proxy::Dynflow::ContinuousOutput.new exit_code = nil case output[:state] when READY_FOR_PICKUP, NOTIFIED # If the job is not running yet on the client, wipe it from storage cleanup exit_code = 'EXCEPTION' when DELIVERED, RUNNING # Client was notified or is already running, dealing with this situation # is only supported if mqtt is available # Otherwise we have to wait it out if input[:with_mqtt] mqtt_cancel fail_msg += ', notifying the host over MQTT' else fail_msg += ', however the job was triggered without MQTT and cannot be stopped' end end continuous_output.add_output(fail_msg + "\n") process_update(Proxy::Dynflow::Runner::Update.new(continuous_output, exit_code)) suspend unless exit_code end
mqtt_cancel()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 164 def mqtt_cancel payload = mqtt_payload_base.merge( content: "#{input[:proxy_url]}/ssh/jobs/#{input[:job_uuid]}/cancel", metadata: { 'event': 'cancel', 'job_uuid': input[:job_uuid] } ) mqtt_notify payload end
mqtt_notify(payload)
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 175 def mqtt_notify(payload) Proxy::RemoteExecution::Ssh::MQTT.publish(mqtt_topic, JSON.dump(payload)) end
mqtt_payload_base()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 195 def mqtt_payload_base { type: 'data', message_id: SecureRandom.uuid, version: 1, sent: DateTime.now.iso8601, directive: 'foreman' } end
mqtt_start(otp_password)
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 147 def mqtt_start(otp_password) payload = mqtt_payload_base.merge( content: "#{input[:proxy_url]}/ssh/jobs/#{input[:job_uuid]}", metadata: { 'event': 'start', 'job_uuid': input[:job_uuid], 'username': execution_plan_id, 'password': otp_password, 'return_url': "#{input[:proxy_url]}/ssh/jobs/#{input[:job_uuid]}/update", 'version': 'v1', 'effective_user': input[:effective_user] }, ) Proxy::RemoteExecution::Ssh::MQTT::Dispatcher.instance.new(input[:job_uuid], mqtt_topic, payload) output[:state] = NOTIFIED end
mqtt_topic()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 187 def mqtt_topic "yggdrasil/#{host_name}/data/in" end
plan(action_input, mqtt: false)
click to toggle source
Calls superclass method
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 23 def plan(action_input, mqtt: false) super(action_input) input[:with_mqtt] = mqtt end
process_external_event(event)
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 73 def process_external_event(event) output[:state] = RUNNING data = event.data case data['version'] when nil process_external_unversioned(data) when '1' process_external_v1(data) else raise "Unexpected update message version '#{data['version']}'" end end
process_external_unversioned(payload)
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 86 def process_external_unversioned(payload) continuous_output = Proxy::Dynflow::ContinuousOutput.new if payload.key?('output') Array(payload['output']).each do |line| continuous_output.add_output(line, payload['type']) end end exit_code = payload['exit_code'].to_i if payload['exit_code'] process_update(Proxy::Dynflow::Runner::Update.new(continuous_output, exit_code)) end
process_external_v1(payload)
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 97 def process_external_v1(payload) continuous_output = Proxy::Dynflow::ContinuousOutput.new exit_code = nil payload['updates'].each do |update| time = Time.parse update['timestamp'] type = update['type'] case type when 'output' continuous_output.add_output(update['content'], update['stream'], time) when 'exit' exit_code = update['exit_code'].to_i else raise "Unexpected update type '#{update['type']}'" end end process_update(Proxy::Dynflow::Runner::Update.new(continuous_output, exit_code)) end
process_pickup_timeout()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 205 def process_pickup_timeout suspend unless [READY_FOR_PICKUP, NOTIFIED].include? output[:state] kill_run 'The job was not picked up in time' end
process_timeout()
click to toggle source
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 117 def process_timeout kill_run "Execution timeout exceeded" end
run(event = nil)
click to toggle source
Calls superclass method
# File lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb, line 28 def run(event = nil) if event == JobDelivered output[:state] = DELIVERED suspend elsif event == PickupTimeout process_pickup_timeout elsif event == ::Dynflow::Action::Timeouts::Timeout process_timeout elsif event.nil? if (timeout = input['execution_timeout_interval']) schedule_timeout(timeout, optional: true) end super else super end rescue => e cleanup action_logger.error(e) process_update(Proxy::Dynflow::Runner::Update.encode_exception('Proxy error', e)) end