Support device modules in apply statements

bolt.gemspec

@@ -40,6 +40,7 @@ Gem::Specification.new do |spec|
   spec.add_dependency "net-ssh", ">= 4.0"
   spec.add_dependency "orchestrator_client", "~> 0.3.1"
   spec.add_dependency "puppet", [">= 6.0.1", "< 7"]
+  spec.add_dependency "puppet-resource_api"
   spec.add_dependency "r10k", "~> 2.6"
   spec.add_dependency "terminal-table", "~> 1.8"
   spec.add_dependency "winrm", "~> 2.0"

lib/bolt/config.rb

@@ -10,14 +10,16 @@
 require 'bolt/transport/orch'
 require 'bolt/transport/local'
 require 'bolt/transport/docker'
+require 'bolt/transport/remote'
 
 module Bolt
   TRANSPORTS = {
     ssh: Bolt::Transport::SSH,
     winrm: Bolt::Transport::WinRM,
     pcp: Bolt::Transport::Orch,
     local: Bolt::Transport::Local,
-    docker: Bolt::Transport::Docker
+    docker: Bolt::Transport::Docker,
+    remote: Bolt::Transport::Remote,
   }.freeze
 
   class UnknownTransportError < Bolt::Error
@@ -41,6 +43,7 @@ class Config
       'tty' => false
     }.freeze
 
+    # TODO: move these to the transport themselves
     TRANSPORT_SPECIFIC_DEFAULTS = {
       ssh: {
         'host-key-check' => true
@@ -53,7 +56,10 @@ class Config
         'task-environment' => 'production'
       },
       local: {},
-      docker: {}
+      docker: {},
+      remote: {
+        'run-on' => 'localhost'
+      }
     }.freeze
 
     def self.default

lib/bolt/executor.rb

@@ -32,8 +32,14 @@ def initialize(concurrency = 1,
       @load_config = load_config
 
       @transports = Bolt::TRANSPORTS.each_with_object({}) do |(key, val), coll|
-        coll[key.to_s] = Concurrent::Delay.new do
-          val.new
+        if key == :remote
+          coll[key.to_s] = Concurrent::Delay.new do
+            val.new(self)
+          end
+        else
+          coll[key.to_s] = Concurrent::Delay.new do
+            val.new
+          end
         end
       end
       @reported_transports = Set.new

lib/bolt/inventory.rb

@@ -136,6 +136,17 @@ def data_hash
       }
     end
 
+    # TODO: This does two things because the applicator bypasses run_task
+    # It would probably be cleaner to give the exectutor access to inventory
+    # and handle this there.
+    def run_on_target(target, params)
+      if target.remote?
+        [get_target(target.run_on || 'localhost'), params.merge('_target' => target.hash)]
+      else
+        [target, params]
+      end
+    end
+
     #### PRIVATE ####
     #
     # For debugging only now
@@ -207,6 +218,7 @@ def resolve_name(target)
     private :resolve_name
 
     def expand_targets(targets)
+      # TODO: inventory may not be set in this case?
       if targets.is_a? Bolt::Target
         targets
       elsif targets.is_a? Array

lib/bolt/node/errors.rb

@@ -7,7 +7,8 @@ class Node
     class BaseError < Bolt::Error
       attr_reader :issue_code
 
-      def initialize(message, issue_code)
+      # TODO can we just drop issue code here?
+      def initialize(message, issue_code = nil)
         super(message, kind, nil, issue_code)
       end
 
@@ -34,6 +35,12 @@ def kind
       end
     end
 
+    class RemoteError < BaseError
+      def kind
+        'puppetlabs.tasks/remote-task-error'
+      end
+    end
+
     class EnvironmentVarError < BaseError
       def initialize(var, val)
         message = "Could not set environment variable '#{var}' to '#{val}'"

lib/bolt/target.rb

@@ -85,6 +85,18 @@ def to_s
       "Target('#{@uri}', #{opts})"
     end
 
+    def to_h
+      options.merge({
+        'name' => name,
+        'uri' => uri,
+        'protocol' => protocol,
+        'user' => user,
+        'password' => password,
+        'host' => host,
+        'port' => port,
+      })
+    end
+
     def host
       @uri_obj.hostname
     end

lib/bolt/task.rb

@@ -36,6 +36,10 @@ def module_name
       name.split('::').first
     end
 
+    def remote?
+      !!metadata['remote']
+    end
+
     def tasks_dir
       File.join(module_name, 'tasks')
     end

lib/bolt/transport/base.rb

@@ -39,6 +39,7 @@ module Transport
     class Base
       STDIN_METHODS       = %w[both stdin].freeze
       ENVIRONMENT_METHODS = %w[both environment].freeze
+      DEFAULT_INPUT_METHOD = 'both'
 
       attr_reader :logger
 
@@ -68,6 +69,33 @@ def with_events(target, callback)
         result
       end
 
+      def provided_features
+        []
+      end
+
+      def default_input_method
+        'both'
+      end
+
+      def select_implementation(target, task)
+        impl =task.select_implementation(target, provided_features)
+        impl['input_method'] ||= default_input_method
+        impl
+      end
+
+      def add_target_param(target, task, params)
+        # TODO: How should metadata impact this?
+        #if target.remote? && !task.remote?
+        #  raise Bolt::Node::RemoteError.new('Remote targets can only run Remotable tasks')
+        #end
+        #if !target.remote? && task.remote?
+        #  raise Bolt::Node::RemoteError.new('Remote tasks can only be run on remote targets')
+        #end
+
+        # TODO: wrap sensitive values
+        target.remote? ? params.merge('_target' => target.to_h) : params
+      end
+
       def filter_options(target, options)
         if target.options['run-as']
           options.reject { |k, _v| k == '_run_as' }

lib/bolt/transport/local.rb

@@ -13,7 +13,9 @@ def self.options
         %w[tmpdir]
       end
 
-      PROVIDED_FEATURES = ['shell'].freeze
+      def provided_features
+        ['shell']
+      end
 
       def self.validate(_options); end
 
@@ -82,9 +84,9 @@ def run_script(target, script, arguments, _options = {})
       end
 
       def run_task(target, task, arguments, _options = {})
-        implementation = task.select_implementation(target, PROVIDED_FEATURES)
+        implementation = select_implementation(target, task)
         executable = implementation['path']
-        input_method = implementation['input_method'] || 'both'
+        input_method = implementation['input_method']
         extra_files = implementation['files']
 
         in_tmpdir(target.options['tmpdir']) do |dir|

lib/bolt/transport/orch.rb

@@ -29,7 +29,9 @@ def self.options
         %w[service-url cacert token-file task-environment]
       end
 
-      PROVIDED_FEATURES = ['puppet-agent'].freeze
+      def provided_features
+        ['puppet-agent']
+      end
 
       def self.validate(options); end
 

lib/bolt/transport/remote.rb

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+require 'bolt/transport/base'
+
+
+module Bolt
+  module Transport
+    class Remote < Base
+      def self.options
+        # TODO: We should accept arbitrary options here
+        %w[host port user password connnect-timeout device-type run-on]
+      end
+
+      def self.validate(options)
+        # This will fail when validating global config
+        #unless options['device-type']
+        #  raise Bolt::ValidationError, 'Must specify device-type for devices'
+        #end
+      end
+
+      # TODO: this should have access to inventory so target doesn't have to
+      def initialize(executor)
+        super()
+
+        @executor = executor
+      end
+
+      def get_proxy(target)
+        # TODO: This needs to have access to the inventory
+        inventory = target.instance_variable_get(:@inventory)
+        raise "Target was creates without inventory? Not get_targets?" unless inventory
+        inventory.get_targets(target.options['run-on'] || 'localhost').first
+      end
+
+      # Cannot batch because arugments differ
+      def run_task(target, task, arguments, options = {})
+        proxy_target = get_proxy(target)
+        transport = @executor.transport(proxy_target.protocol)
+        arguments = arguments.merge('_target' => target.to_h.reject {|_, v| v.nil?})
+
+        # TODO: add support for device-type and feature checking here.
+        # * tasks/task_implementations must have the same device-type as the target
+        # * Does one implementation need to support multiple device types?
+        # * Do we need multiple implementations for the same device based on proxy features?
+        # TODO doesn't support transports with only batch exec(orchestrator). update orch
+        result = transport.run_task(proxy_target, task, arguments, options)
+
+        Bolt::Result.new(target, value: result.value)
+      end
+    end
+  end
+end

lib/bolt/transport/ssh.rb

@@ -12,7 +12,9 @@ def self.options
         %w[port user password sudo-password private-key host-key-check connect-timeout tmpdir run-as tty run-as-command]
       end
 
-      PROVIDED_FEATURES = ['shell'].freeze
+      def provided_features
+        ['shell']
+      end
 
       def self.validate(options)
         logger = Logging.logger[self]
@@ -122,11 +124,10 @@ def run_script(target, script, arguments, options = {})
       end
 
       def run_task(target, task, arguments, options = {})
-        implementation = task.select_implementation(target, PROVIDED_FEATURES)
+        implementation = select_implementation(target, task)
         executable = implementation['path']
         input_method = implementation['input_method']
         extra_files = implementation['files']
-        input_method ||= 'both'
 
         # unpack any Sensitive data
         arguments = unwrap_sensitive_args(arguments)

lib/bolt/transport/winrm.rb

@@ -14,7 +14,13 @@ def self.options
         %w[port user password connect-timeout ssl ssl-verify tmpdir cacert extensions]
       end
 
-      PROVIDED_FEATURES = ['powershell'].freeze
+      def provided_features
+        ['powershell']
+      end
+
+      def default_input_method
+        'powershell'
+      end
 
       def self.validate(options)
         ssl_flag = options['ssl']
@@ -108,7 +114,7 @@ def run_script(target, script, arguments, _options = {})
       end
 
       def run_task(target, task, arguments, _options = {})
-        implementation = task.select_implementation(target, PROVIDED_FEATURES)
+        implementation = select_implementation(target, task)
         executable = implementation['path']
         input_method = implementation['input_method']
         extra_files = implementation['files']

libexec/apply_catalog.rb

@@ -31,6 +31,23 @@
 
 Puppet[:default_file_terminus] = :file_server
 
+if args['_target']
+  # TODO: move some of this logic to puppet
+  special_keys = ['type', 'debug']
+  connection  = conn_info.reject { |k, _| special_keys.include?(k) }
+  device = OpenStruct.new(connection)
+  device.provider = conn_info['type']
+  device.options[:debug] = true if conn_info['debug']
+  Puppet[:certname] = device.name
+  Puppet::Util::NetworkDevice.init(device)
+
+  Puppet[:facts_terminus] = :network_device
+  Puppet[:node_terminus] = :plain
+  Puppet[:catalog_terminus] = :compiler
+  # TODO this shouldn't be device speciifc
+  Puppet[:catalog_cache_terminus] = nil
+end
+
 exit_code = 0
 begin
   # This happens implicitly when running the Configurer, but we make it explicit here. It creates the
@@ -49,6 +66,7 @@
   end
 
   # Ensure custom facts are available for provider suitability tests
+  # TODO: skip this for devices?
   facts = Puppet::Node::Facts.indirection.find(SecureRandom.uuid, environment: env)
 
   report = if Puppet::Util::Package.versioncmp(Puppet.version, '5.0.0') > 0
@@ -57,8 +75,11 @@
              Puppet::Transaction::Report.new('apply')
            end
 
-  Puppet.override(current_environment: env,
-                  loaders: Puppet::Pops::Loaders.new(env)) do
+  overrides = { current_environment: env,
+                loaders: Puppet::Pops::Loaders.new(env) }
+  ovierrides[:network_device] = true if args['_target']
+
+  Puppet.override(overrides) do
     catalog = Puppet::Resource::Catalog.from_data_hash(args['catalog'])
     catalog.environment = env.name.to_s
     catalog.environment_instance = env