Added plugins management to host handler

cloudcontrol/node/host/__init__.py

@@ -17,13 +17,14 @@ from subprocess import Popen, PIPE, STDOUT
 from sjrpc.utils import pass_connection, threadless
 from sjrpc.core.protocols import TunnelProtocol
 from sjrpc.core.exceptions import RpcError
-from cloudcontrol.common.client.tags import Tag, tag_inspector
+from cloudcontrol.common.client.tags import Tag, tag_inspector, TagDB
 from cloudcontrol.common.client.plugins import Base as BasePlugin
 from cloudcontrol.common.jobs import JobsManager, JobsStore
 from cloudcontrol.common.helpers.logger import patch_logging; patch_logging()
 
 from cloudcontrol.node.host import tags
 from cloudcontrol.node.host.jobs import NodeJobsManagerInterface, ScriptJob
+from cloudcontrol.node.host.plugins import PluginMethodJob, Plugin
 
 
 logger = logging.getLogger(__name__)
@@ -184,6 +185,9 @@ class Handler(BasePlugin):
             job_purge=self.job_purge,
             job_attachment=self.job_attachment,
             script_run=self.script_run,
+            plugin_install=self.plugin_install,
+            plugin_uninstall=self.plugin_uninstall,
+            plugin_run=self.plugin_run
         ))
         # running shells
         self.shells = dict()
@@ -192,6 +196,9 @@ class Handler(BasePlugin):
         self.jobs_manager = JobsManager(logger, NodeJobsManagerInterface(self),
                                        JobsStore(self.main.config.jobs_store_path))
 
+        #: loaded plugins
+        self.plugins = {}  # plugin name -> plugin object
+
     def stop(self):
         # kill all currently running shells
         for shell in self.shells.values():
@@ -322,3 +329,44 @@ class Handler(BasePlugin):
         :param name: attachement name
         """
         return self.jobs_manager.get(job_id).read_attachment(name)
+
+    @pass_connection
+    def plugin_install(self, conn, sha1, name):
+        # check if plugin is not already loaded and upgrade it if the sha1 hash
+        # has changed:
+        if name in self.plugins:
+            if self.plugins[name].sha1 != sha1:
+                self.plugins[name].uninstall()
+                del self.plugins[name]
+            else:
+                return
+
+        # get plugin from server:
+        sha1_get, content = conn.call('plugin_get', name)
+        if sha1 != sha1_get:
+            raise RuntimeError('Requested sha1 is not available on the server')
+
+        # load the plugin:
+        plugin_logger = logger.getChild('plugin.%s' % name)
+        self.plugins[name] = Plugin(plugin_logger, TagDB(self.tag_db),
+                                    name, sha1, content)
+
+    def plugin_uninstall(self, name):
+        plugin = self.plugins.pop(name, None)
+        if plugin is None:
+            raise KeyError('Plugin %r is not running' % name)
+        plugin.uninstall()
+
+    def plugin_run(self, name, method, owner, **kwargs):
+        if name not in self.plugins:
+            raise KeyError('Plugin %r is not running' % name)
+        try:
+            func = self.plugins[name].methods[method]
+        except KeyError:
+            raise KeyError('Unknown method %r' % method)
+        return self.jobs_manager.spawn(PluginMethodJob, owner, settings=dict(
+            plugin_name=name,
+            method_name=method,
+            method=func,
+            method_kwargs=kwargs,
+        )).id