Merge pull request #85 from matburt/fact_scans_and_caching

Fact scans and caching

awx/main/management/commands/run_callback_receiver.py

@@ -87,7 +87,7 @@ class CallbackReceiver(object):
                     queue_worker[2] = w
             if workers_changed:
                 signal.signal(signal.SIGINT, shutdown_handler([p[2] for p in worker_queues] + [main_process]))
-                signal.signal(signal.SIGTERM, shutdown_handler([p[2] for p in worker_queues] + [main_process]))        
+                signal.signal(signal.SIGTERM, shutdown_handler([p[2] for p in worker_queues] + [main_process]))
             if not main_process.is_alive():
                 sys.exit(1)
             time.sleep(0.1)
@@ -239,7 +239,7 @@ class Command(NoArgsCommand):
     Save Job Callback receiver (see awx.plugins.callbacks.job_event_callback)
     Runs as a management command and receives job save events.  It then hands
     them off to worker processors (see Worker) which writes them to the database
-    '''    
+    '''
     help = 'Launch the job callback receiver'
 
     def handle_noargs(self, **options):

awx/main/management/commands/run_fact_cache_receiver.py

@@ -0,0 +1,55 @@
+# Copyright (c) 2015 Ansible, Inc.
+# All Rights Reserved
+
+import logging
+
+from django.core.management.base import NoArgsCommand
+
+from awx.main.models import * # noqa
+from awx.main.socket import Socket
+
+from pymongo import MongoClient
+
+logger = logging.getLogger('awx.main.commands.run_fact_cache_receiver')
+
+class FactCacheReceiver(object):
+
+    def __init__(self):
+        self.client = MongoClient('localhost', 27017)
+
+    def process_fact_message(self, message):
+        host = message['host']
+        facts = message['facts']
+        date_key = message['date_key']
+        host_db = self.client.host_facts
+        host_collection = host_db[host]
+        facts.update(dict(tower_host=host, datetime=date_key))
+        rec = host_collection.find({"datetime": date_key})
+        if rec.count():
+            this_fact = rec.next()
+            this_fact.update(facts)
+            host_collection.save(this_fact)
+        else:
+            host_collection.insert(facts)
+
+    def run_receiver(self):
+        with Socket('fact_cache', 'r') as facts:
+            for message in facts.listen():
+                print("Message received: " + str(message))
+                if 'host' not in message or 'facts' not in message or 'date_key' not in message:
+                    continue
+                self.process_fact_message(message)
+
+class Command(NoArgsCommand):
+    '''
+    blah blah
+    '''
+    help = 'Launch the Fact Cache Receiver'
+
+    def handle_noargs(self, **options):
+        fcr = FactCacheReceiver()
+        try:
+            fcr.run_receiver()
+        except KeyboardInterrupt:
+            pass
+

awx/main/socket.py

@@ -63,6 +63,7 @@ class Socket(object):
                                         settings.CALLBACK_CONSUMER_PORT),
             'task_commands': settings.TASK_COMMAND_PORT,
             'websocket': settings.SOCKETIO_NOTIFICATION_PORT,
+            'fact_cache': settings.FACT_CACHE_PORT,
         }[self._bucket]
 
     def connect(self):

awx/main/tasks.py

@@ -550,6 +550,10 @@ class RunJob(BaseTask):
         env['JOB_ID'] = str(job.pk)
         env['INVENTORY_ID'] = str(job.inventory.pk)
         env['ANSIBLE_CALLBACK_PLUGINS'] = plugin_dir
+        # TODO: env['ANSIBLE_LIBRARY'] # plugins/library
+        # TODO: env['ANSIBLE_CACHE_PLUGINS'] # plugins/fact_caching
+        # TODD: env['ANSIBLE_CACHE_PLUGIN'] # tower
+        # TODO: env['ANSIBLE_CACHE_PLUGIN_CONNECTION'] # connection to tower service
         env['REST_API_URL'] = settings.INTERNAL_API_URL
         env['REST_API_TOKEN'] = job.task_auth_token or ''
         env['CALLBACK_CONSUMER_PORT'] = str(settings.CALLBACK_CONSUMER_PORT)

awx/playbooks/scan_facts.yml

@@ -0,0 +1,4 @@
+- hosts: all
+  tasks:
+    - scan_packages:
+

awx/plugins/fact_caching/tower.py

@@ -0,0 +1,80 @@
+# Copyright (c) 2015 Ansible, Inc.
+# This file is a utility Ansible plugin that is not part of the AWX or Ansible
+# packages.  It does not import any code from either package, nor does its
+# license apply to Ansible or AWX.
+# 
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+# 
+# Redistributions of source code must retain the above copyright notice, this
+# list of conditions and the following disclaimer.
+#
+#    Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+#
+#    Neither the name of the <ORGANIZATION> nor the names of its contributors
+#    may be used to endorse or promote products derived from this software
+#    without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+import sys
+import time
+import datetime
+from ansible import constants as C
+from ansible.cache.base import BaseCacheModule
+
+try:
+    import zmq
+except ImportError:
+    print("pyzmq is required")
+    sys.exit(1)
+
+class CacheModule(BaseCacheModule):
+
+    def __init__(self, *args, **kwargs):
+
+        # This is the local tower zmq connection
+        self._tower_connection = C.CACHE_PLUGIN_CONNECTION
+        self.date_key = time.mktime(datetime.datetime.utcnow().timetuple())
+        try:
+            self.context = zmq.Context()
+            self.socket = self.context.socket(zmq.REQ)
+            self.socket.connect(self._tower_connection)
+        except Exception, e:
+            print("Connection to zeromq failed at %s with error: %s" % (str(self._tower_connection),
+                                                                        str(e)))
+            sys.exit(1)
+
+    def get(self, key):
+        return {} # Temporary until we have some tower retrieval endpoints
+
+    def set(self, key, value):
+        self.socket.send_json(dict(host=key, facts=value, date_key=self.date_key))
+        self.socket.recv()
+
+    def keys(self):
+        return []
+
+    def contains(self, key):
+        return False
+
+    def delete(self, key):
+        pass
+
+    def flush(self):
+        pass
+
+    def copy(self):
+        return dict()

awx/plugins/library/scan_packages.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+
+import os
+from ansible.module_utils.basic import *
+
+def rpm_package_list():
+    import rpm
+    trans_set = rpm.TransactionSet()
+    installed_packages = {}
+    for package in trans_set.dbMatch():
+        package_details = dict(name=package[rpm.RPMTAG_NAME],
+                               version=package[rpm.RPMTAG_VERSION],
+                               release=package[rpm.RPMTAG_RELEASE],
+                               epoch=package[rpm.RPMTAG_EPOCH],
+                               arch=package[rpm.RPMTAG_ARCH],
+                               source='rpm')
+        if package['name'] not in installed_packages:
+            installed_packages[package['name']] = [package_details]
+        else:
+            installed_packages[package['name']].append(package_details)
+    return installed_packages
+
+def deb_package_list():
+    import apt
+    apt_cache = apt.Cache()
+    installed_packages = {}
+    apt_installed_packages = [pk for pk in apt_cache.keys() if apt_cache[pk].is_installed]
+    for package in apt_installed_packages:
+        ac_pkg = apt_cache[package].installed
+        package_details = dict(name=package,
+                               version=ac_pkg.version,
+                               architecture=ac_pkg.architecture,
+                               source='apt')
+        installed_packages[package] = [package_details]
+    return installed_packages
+
+def main():
+    module = AnsibleModule(
+        argument_spec = dict())
+
+    packages = []
+    if os.path.exists("/etc/redhat-release"):
+        packages = rpm_package_list()
+    elif os.path.exists("/etc/os-release"):
+        packages = deb_package_list()
+    results = dict(ansible_facts=dict(packages=packages))
+    module.exit_json(**results)
+
+main()

awx/settings/defaults.py

@@ -510,6 +510,8 @@ TASK_COMMAND_PORT = 6559
 SOCKETIO_NOTIFICATION_PORT = 6557
 SOCKETIO_LISTEN_PORT = 8080
 
+FACT_CACHE_PORT = 6564
+
 ORG_ADMINS_CAN_SEE_ALL_USERS = True
 
 # Logging configuration.