Add ansible supervisory code

confluent_server/confluent/runansible.py

@@ -0,0 +1,94 @@
+#!/usr/bin/python
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2014 IBM Corporation
+# Copyright 2015-2018 Lenovo
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import confluent.sshutil as sshutil
+import eventlet.green.subprocess as subprocess
+import msgpack
+import os
+import sys
+
+running_status = {}
+
+def run_playbooks(playfiles, nodes):
+    sshutil.prep_ssh_key('/etc/confluent/ssh/automation')
+    targnodes = ','.join(nodes)
+    for playfilename in playfiles:
+        worker = subprocess.Popen(
+            [sys.executable, __file__, targnodes, playfilename])
+        for node in nodes:
+            running_status[node] = worker
+
+
+if __name__ == '__main__':
+    from ansible.inventory.manager import InventoryManager
+    from ansible.parsing.dataloader import DataLoader
+    from ansible.executor.task_queue_manager import TaskQueueManager
+    from ansible.vars.manager import VariableManager
+    from ansible.playbook.play import Play
+    from ansible import context
+    from ansible.module_utils.common.collections import ImmutableDict
+    from ansible.plugins.callback import CallbackBase
+    import yaml
+
+    class ResultsCollector(CallbackBase):
+        def v2_runner_on_unreachable(self, result):
+            print(dir(result))
+            print(repr(result._result))
+
+        def v2_runner_on_ok(self, result, *args, **kwargs):
+            print(repr(result))
+            print(repr(result.task_name))
+            print(repr(result.is_changed()))
+            print(repr(result.is_skipped()))
+            print(repr(result.is_failed()))
+            print(repr(result.is_unreachable()))
+            print(dir(result))
+            print(repr(result._result))
+            print(repr(args))
+            print(repr(kwargs))
+
+        def v2_runner_on_failed(self, result, *args, **kwargs):
+            print(repr(result))
+            print(repr(args))
+            print(repr(kwargs))
+
+    context.CLIARGS = ImmutableDict(
+        connection='smart', module_path=['/usr/share/ansible'], forks=10,
+        become=None, become_method=None, become_user=None, check=False,
+        diff=False, verbosity=0, remote_user='root')
+
+
+    invlist = sys.argv[1] + ','
+    loader = DataLoader()
+    invman = InventoryManager(loader=loader, sources=invlist)
+    varman = VariableManager(loader=loader, inventory=invman)
+
+    plays = yaml.safe_load(open(sys.argv[2]))
+    if isinstance(plays, dict):
+        plays = [plays]
+    taskman = TaskQueueManager(inventory=invman, loader=loader, passwords={},
+        variable_manager=varman, stdout_callback=ResultsCollector())
+    for currplay in plays:
+        currplay['hosts'] = sys.argv[1]
+        play = Play().load(currplay, loader=loader)
+        try:
+            taskman.run(play)
+        finally:
+            taskman.cleanup()
+            if loader:
+                loader.cleanup_all_tmp_files()

confluent_server/confluent/selfservice.py

@@ -1,3 +1,4 @@
+import confluent.runansible as runansible
 import confluent.config.configmanager as configmanager
 import confluent.collective.manager as collective
 import confluent.netutil as netutil
@@ -272,14 +273,16 @@ def handle_request(env, start_response):
         for filename in slist:
             if filename.endswith('.yaml') or filename.endswith('.yml'):
                 playlist.append(os.path.join(dirname, filename))
-        if not playlist:
+        if playlist:
+            runansible.run_playbooks(playlist, [nodename])
+            start_response('202 Queued', ())
+            yield ''
+        else:
             start_response('200 OK', ())
             yield ''
             return
-        
     elif env['PATH_INFO'].startswith('/self/remoteconfig/status'):
         scriptcat = env['PATH_INFO'].replace('/self/remoteconfig/', '')
-
     elif env['PATH_INFO'].startswith('/self/scriptlist/'):
         scriptcat = env['PATH_INFO'].replace('/self/scriptlist/', '')
         slist, _ = get_scriptlist(
@@ -309,7 +312,7 @@ def get_scriptlist(scriptcat, cfg, nodename, pathtemplate):
     if not profile:
         profile = deployinfo.get(
         'deployment.profile', {}).get('value', '')
-    slist = None
+    slist = []
     try:
         slist = os.listdir(pathtemplate.format(profile, scriptcat))
     except OSError: