Groundwork for pull provider

Author: adamruzicka

lib/smart_proxy_remote_execution_ssh.rb

@@ -51,6 +51,10 @@ def validate_ssh_log_level!
 
         Plugin.settings.ssh_log_level = Plugin.settings.ssh_log_level.to_sym
       end
+
+      def job_storage
+        @job_storage ||= Proxy::MemoryStore.new
+      end
     end
   end
 end

lib/smart_proxy_remote_execution_ssh/actions/pull_script.rb

@@ -0,0 +1,84 @@
+require 'mqtt'
+require 'json'
+
+module Proxy::RemoteExecution::Ssh
+  class PullScript < Proxy::Dynflow::Action::Runner
+    JobDelivered = Class.new
+
+    execution_plan_hooks.use :cleanup, :on => :stopped
+
+    def plan(action_input, mqtt: false)
+      super(action_input)
+      input[:with_mqtt] = mqtt
+    end
+
+    def run(event = nil)
+      if event == JobDelivered
+        output[:state] = :delivered
+        suspend
+      else
+        super
+      end
+    end
+
+    def init_run
+      Proxy::RemoteExecution::Ssh::Plugin.job_storage["#{input[:hostname]}-#{execution_plan_id}", run_step_id, 'script.sh'] = input[:script]
+      output[:state] = :ready_for_pickup
+      mqtt_start if input[:with_mqtt]
+      suspend
+    end
+
+    def cleanup(_plan = nil)
+      Proxy::RemoteExecution::Ssh::Plugin.job_storage.delete("#{input[:hostname]}-#{execution_plan_id}")
+    end
+
+    def process_external_event(event)
+      output[:state] = :running
+      data = event.data
+      continuous_output = Proxy::Dynflow::ContinuousOutput.new
+      continuous_output.add_output(lines, 'stdout') if data.key?('output')
+      exit_code = data['exit_code'].to_i if data['exit_code']
+      process_update(Proxy::Dynflow::Runner::Update.new(continuous_output, exit_code))
+    end
+
+    def kill_run
+      case output[:state]
+      when :ready_for_pickup
+        # If the job is not running yet on the client, wipe it from storage
+        cleanup
+        # TODO: Stop the action
+      when :notified, :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]
+          cleanup
+          payload = {} # TODO
+          mqtt_notify payload
+        end
+      end
+      suspend
+    end
+
+    def mqtt_start
+      payload = {
+        type: 'data',
+        message_id: SecureRandom.uuid,
+        version: 1,
+        sent: DateTime.now.iso8601,
+        directive: 'foreman',
+        content: "#{input[:proxy_url]}/job/store/#{execution_plan_id}/#{run_step_id}/script.sh",
+      }
+      mqtt_notify payload
+      output[:state] = :notified
+    end
+
+    def mqtt_notify(payload)
+      broker = 'localhost' # TODO
+      broker_port = 1883 # TODO
+      MQTT::Client.connect(broker, broker_port) do |c|
+        c.publish("yggdrasil/#{input[:hostname]}/data/in", JSON.dump(payload), false, 1)
+      end
+    end
+  end
+end

lib/smart_proxy_remote_execution_ssh/api.rb

@@ -1,5 +1,6 @@
 require 'net/ssh'
 require 'base64'
+require 'smart_proxy_dynflow/runner'
 
 module Proxy::RemoteExecution
   module Ssh
@@ -37,6 +38,43 @@ class Api < ::Sinatra::Base
           end
         204
       end
+
+      # Payload is a hash where
+      # exit_code: Integer | NilClass
+      # output: String
+      post '/job/:task_id/:step_id/update' do |task_id, step_id|
+        do_authorize_with_ssl_client
+
+        path = job_path(https_cert_cn, task_id, nil, nil).first
+        if Proxy::RemoteExecution::Ssh.job_storage[path].nil?
+          status 404
+          return ''
+        end
+
+        data = MultiJson.load(request.body.read)
+        world.event(task_id, step_id, ::Proxy::Dynflow::Runner::ExternalEvent.new(data))
+      end
+
+      get "/job/store/:task_id/:step_id/:file" do |task_id, step_id, file|
+        do_authorize_with_ssl_client
+
+        path = job_path(https_cert_cn, task_id, step_id.to_i, file)
+        content = Proxy::RemoteExecution::Ssh.job_storage[*path]
+        if content
+          world.event(task_id, step_id.to_i, Proxy::RemoteExecution::Ssh::PullScript::JobDelivered)
+          return content
+        end
+
+        status 404
+        ''
+      end
+
+      def job_path(hostname, task_id, step_id, file)
+        ["#{hostname}-#{task_id}",
+         step_id,
+         file,
+        ]
+      end
     end
   end
 end

lib/smart_proxy_remote_execution_ssh/plugin.rb

@@ -24,6 +24,7 @@ class Plugin < Proxy::Plugin
       require 'smart_proxy_remote_execution_ssh/cockpit'
       require 'smart_proxy_remote_execution_ssh/api'
       require 'smart_proxy_remote_execution_ssh/actions/run_script'
+      require 'smart_proxy_remote_execution_ssh/actions/pull_script'
       require 'smart_proxy_remote_execution_ssh/dispatcher'
       require 'smart_proxy_remote_execution_ssh/log_filter'
       require 'smart_proxy_remote_execution_ssh/runners'

smart_proxy_remote_execution_ssh.gemspec

@@ -31,4 +31,5 @@ Gem::Specification.new do |gem|
 
   gem.add_runtime_dependency('smart_proxy_dynflow', '~> 0.5')
   gem.add_runtime_dependency('net-ssh', '>= 4.2.0')
+  gem.add_runtime_dependency('mqtt')
 end

test/api_test.rb

@@ -1,5 +1,6 @@
 require 'test_helper'
 require 'tempfile'
+require 'smart_proxy_remote_execution_ssh/actions/pull_script'
 
 KNOWN_HOSTS = <<EOF
 c7s62.lxc,192.168.122.200 ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBFAMEcFeBHeY8AD7xw2weF6vIE0BZXBk0oOm7sM+iJ4ld7BvQDf0mF6EeyyjzDmMUTyR2q9q0OdYiTbyEiKHQF4=
@@ -78,5 +79,71 @@ def with_known_hosts
         end
       end
     end
+
+    describe '/job/update' do
+      let(:hostname) { 'myhost.example.com' }
+
+      before { Proxy::RemoteExecution::Ssh.job_storage["#{hostname}-12345", 1, 'message'] = 'hello' }
+      after  { Proxy::RemoteExecution::Ssh.job_storage.delete("#{hostname}-12345") }
+
+      it 'returns 403 if HTTPS is used and no cert is provided' do
+        post '/job/12345/1/update', {}, 'HTTPS' => 1
+        _(last_response.status).must_equal 403
+      end
+
+      it 'returns 404 if job does not exist' do
+        Proxy::RemoteExecution::Ssh::Api.any_instance.expects(:https_cert_cn).returns(hostname)
+        post '/job/12346/1/update'
+        _(last_response.status).must_equal 404
+      end
+
+      it 'dispatches an event' do
+        Proxy::RemoteExecution::Ssh::Api.any_instance.expects(:https_cert_cn).returns(hostname)
+        fake_world = mock
+        fake_world.expects(:event) do |task_id, step_id, _payload|
+          task_id == '12345' && step_id == 1
+        end
+        Proxy::RemoteExecution::Ssh::Api.any_instance.expects(:world).returns(fake_world)
+
+        post '/job/12345/1/update', '{}'
+        _(last_response.status).must_equal 200
+      end
+    end
+
+    describe '/job/store' do
+      let(:hostname) { 'myhost.example.com' }
+
+      before { Proxy::RemoteExecution::Ssh.job_storage["#{hostname}-12345", 1, 'message'] = 'hello' }
+      after  { Proxy::RemoteExecution::Ssh.job_storage.delete("#{hostname}-12345") }
+
+      it 'returns 403 if HTTPS is used and no cert is provided' do
+        get '/job/store/12345/1/message', {}, 'HTTPS' => 1
+        _(last_response.status).must_equal 403
+      end
+
+      it 'returns content if there is some and notifies the action' do
+        Proxy::RemoteExecution::Ssh::Api.any_instance.expects(:https_cert_cn).returns(hostname)
+        fake_world = mock
+        fake_world.expects(:event).with('12345', 1, Proxy::RemoteExecution::Ssh::PullScript::JobDelivered)
+        Proxy::RemoteExecution::Ssh::Api.any_instance.expects(:world).returns(fake_world)
+
+        get '/job/store/12345/1/message'
+        _(last_response.status).must_equal 200
+        _(last_response.body).must_equal 'hello'
+      end
+
+      it 'returns 404 if there is no content' do
+        Proxy::RemoteExecution::Ssh::Api.any_instance.expects(:https_cert_cn).times(3).returns(hostname)
+
+        get '/job/store/12345/1/something.tar.gz'
+        _(last_response.status).must_equal 404
+
+        get '/job/store/12345/2/something.tar.gz'
+        _(last_response.status).must_equal 404
+
+        get '/job/store/12346/2/something.tar.gz'
+        _(last_response.status).must_equal 404
+      end
+    end
   end
 end