Merge branch 'feature/convert-data-extraction-to-tasks' into develop

inventory_provider/opsdb.py

@@ -18,7 +18,14 @@ equipment_location_query = """SELECT
                           ON g.absid = p.PTR_geocoding
                         WHERE
                           e.status != 'terminated'
-                          AND e.status != 'disposed'"""
+                          AND e.status != 'disposed'
+                        ORDER BY
+                         FIELD(e.status,
+                          'spare',
+                          'planned',
+                          'ordered',
+                          'installed',
+                          'operational')"""
 
 
 circuit_hierarchy_query = """SELECT

inventory_provider/routes/jobs.py

 import logging
+
 from flask import Blueprint, Response, current_app
 
-import inventory_provider.storage.external_inventory as external_inventory
-from inventory_provider import db, opsdb
 from inventory_provider.tasks.app import app
 from inventory_provider.constants import TASK_LOGGER_NAME
 
@@ -49,56 +48,45 @@ def startup_update():
     return Response("OK")
 
 
-@routes.route("update-services", methods=['GET'])
-def update_service():
-    config = current_app.config['INVENTORY_PROVIDER_CONFIG']
-
-    with db.connection(config['ops-db']) as connection:
-        result = opsdb.get_circuits(connection)
-    external_inventory.update_services_to_monitor(result)
+@routes.route("update-services-to-monitor", methods=['GET'])
+def update_services_to_monitor():
+    app.send_task(
+        'inventory_provider.tasks.worker.update_service_to_monitor')
     return Response("OK")
 
 
-@routes.route("update-interfaces", methods=['GET'])
-def update_interfaces():
-    config = current_app.config['INVENTORY_PROVIDER_CONFIG']
-
-    with db.connection(config['ops-db']) as connection:
-        result = opsdb.get_circuits(connection)
-    external_inventory.update_interfaces_to_services(result)
+@routes.route("update-interfaces-to-services", methods=['GET'])
+def update_interfaces_to_services():
+    app.send_task(
+        'inventory_provider.tasks.worker.update_interfaces_to_services')
     return Response("OK")
 
 
 @routes.route("update-service-hierarchy", methods=['GET'])
 def update_service_hierarchy():
-    config = current_app.config['INVENTORY_PROVIDER_CONFIG']
-
-    with db.connection(config['ops-db']) as connection:
-        result = opsdb.get_circuit_hierarchy(connection)
-    external_inventory.update_service_hierarchy(result)
+    app.send_task('inventory_provider.tasks.worker.update_circuit_hierarchy')
     return Response("OK")
 
 
 @routes.route("update-equipment-locations", methods=['GET'])
 def update_equipment_locations():
-    config = current_app.config['INVENTORY_PROVIDER_CONFIG']
-
-    with db.connection(config['ops-db']) as connection:
-        result = opsdb.get_equipment_location_data(connection)
-    external_inventory.update_equipment_locations(result)
+    app.send_task('inventory_provider.tasks.worker.update_equipment_locations')
     return Response("OK")
 
 
 @routes.route("update-from-inventory-system", methods=['GET'])
 def update_from_inventory_system():
-    config = current_app.config['INVENTORY_PROVIDER_CONFIG']
-
-    with db.connection(config['ops-db']) as connection:
-        circuits = opsdb.get_circuits(connection)
-        hierarchy = opsdb.get_circuit_hierarchy(connection)
-        equipment_locations = opsdb.get_equipment_location_data(connection)
-    external_inventory.update_services_to_monitor(circuits)
-    external_inventory.update_interfaces_to_services(circuits)
-    external_inventory.update_service_hierarchy(hierarchy)
-    external_inventory.update_equipment_locations(equipment_locations)
+    app.send_task(
+        'inventory_provider.tasks.worker.update_service_to_monitor')
+    app.send_task(
+        'inventory_provider.tasks.worker.update_interfaces_to_services')
+    app.send_task('inventory_provider.tasks.worker.update_circuit_hierarchy')
+    app.send_task('inventory_provider.tasks.worker.update_equipment_locations')
+    return Response("OK")
+
+
+@routes.route("update-interface-statuses")
+def update_interface_statuses():
+    app.send_task(
+        'inventory_provider.tasks.worker.update_interface_statuses')
     return Response("OK")

inventory_provider/routes/opsdb.py

@@ -3,13 +3,15 @@ import json
 
 from flask import Blueprint, request, Response
 from inventory_provider import db
-from inventory_provider.storage import external_inventory
 
 routes = Blueprint("inventory-opsdb-query-routes", __name__)
 
-services_key = "inv_services"
-interfaces_key = "inv_interfaces"
-equipment_locations_key = "inv_eq_locations"
+monitored_services_key = "monitored_services"
+interfaces_key = "interface_services"
+equipment_locations_key = "equipment_locations"
+service_child_to_parents_key = "child_to_parent_circuit_relations"
+service_parent_to_children_key = "parent_to_children_circuit_relations"
+interface_status_key = "interface_statuses"
 
 
 def require_accepts_json(f):
@@ -93,7 +95,7 @@ def get_children(parent_id):
     r = db.get_redis()
     return Response(
         r.hget(
-            external_inventory.service_parent_to_children_key,
+            service_parent_to_children_key,
             parent_id),
         mimetype="application/json")
 
@@ -103,6 +105,6 @@ def get_parents(child_id):
     r = db.get_redis()
     return Response(
         r.hget(
-            external_inventory.service_child_to_parents_key,
+            service_child_to_parents_key,
             child_id),
         mimetype="application/json")

inventory_provider/tasks/worker.py

@@ -2,6 +2,7 @@ import json
 import logging
 
 from celery import bootsteps, Task
+from collections import defaultdict
 import redis
 from lxml import etree
 
@@ -11,6 +12,7 @@ from inventory_provider import config
 from inventory_provider import constants
 from inventory_provider import db
 from inventory_provider import environment
+from inventory_provider import opsdb
 from inventory_provider import snmp
 from inventory_provider import juniper
 
@@ -81,6 +83,14 @@ class WorkerArgs(bootsteps.Step):
         InventoryTask.logger = logging.getLogger(constants.TASK_LOGGER_NAME)
 
 
+monitored_services_key = "monitored_services"
+interfaces_key = "interface_services"
+equipment_locations_key = "equipment_locations"
+service_child_to_parents_key = "child_to_parent_circuit_relations"
+service_parent_to_children_key = "parent_to_children_circuit_relations"
+interface_status_key = "interface_statuses"
+
+
 def worker_args(parser):
     parser.add_argument(
         "--config_filename",
@@ -135,3 +145,106 @@ def update_alarmsdb_cache(self):
             InventoryTask.save_value_json('alarmsdb:%s' % table_name, data)
 
     logger.debug('FINISHED: update_alarmsdb_cache')
+
+
+@app.task()
+def update_service_to_monitor():
+    # todo - factor this connection stuff out
+    r = redis.StrictRedis(
+        host=InventoryTask.config["redis"]["hostname"],
+        port=InventoryTask.config["redis"]["port"])
+    r.delete(monitored_services_key)
+
+    relevant_types = ("path", "service", "l2circuit")
+    with db.connection(InventoryTask.config["ops-db"]) as cx:
+        for circuit in opsdb.get_circuits(cx):
+            if circuit["circuit_type"].lower() in relevant_types:
+                r.hset(
+                    monitored_services_key, circuit["id"], json.dumps(circuit))
+
+
+@app.task()
+def update_interfaces_to_services():
+    # todo - factor this connection stuff out
+    r = redis.StrictRedis(
+        host=InventoryTask.config["redis"]["hostname"],
+        port=InventoryTask.config["redis"]["port"])
+    with db.connection(InventoryTask.config["ops-db"]) as cx:
+        services = opsdb.get_circuits(cx)
+
+    mapped_interfaces = defaultdict(list)
+    for service in services:
+        key = "{}::{}".format(
+            service["equipment"],
+            service["interface_name"]
+        )
+        mapped_interfaces[key].append(service)
+
+    r.delete(interfaces_key)
+    for key, value in mapped_interfaces.items():
+        r.hset(interfaces_key, key, json.dumps(value))
+
+
+@app.task()
+def update_equipment_locations():
+    # todo - factor this connection stuff out
+    r = redis.StrictRedis(
+        host=InventoryTask.config["redis"]["hostname"],
+        port=InventoryTask.config["redis"]["port"])
+    r.delete(equipment_locations_key)
+
+    with db.connection(InventoryTask.config["ops-db"]) as cx:
+        for ld in opsdb.get_equipment_location_data(cx):
+            r.hset(
+                equipment_locations_key, ld["equipment_name"], json.dumps(ld))
+
+
+@app.task()
+def update_circuit_hierarchy():
+    # todo - factor this connection stuff out
+    r = redis.StrictRedis(
+        host=InventoryTask.config["redis"]["hostname"],
+        port=InventoryTask.config["redis"]["port"])
+    children_to_parents = defaultdict(list)
+    parents_to_children = defaultdict(list)
+    with db.connection(InventoryTask.config["ops-db"]) as cx:
+        records = opsdb.get_circuit_hierarchy(cx)
+        for relation in records:
+            parent_id = relation["parent_circuit_id"]
+            child_id = relation["child_circuit_id"]
+            parents_to_children[parent_id].append(relation)
+            children_to_parents[child_id].append(relation)
+
+        r.delete(service_child_to_parents_key)
+        for child, parents in children_to_parents.items():
+            r.hset(service_child_to_parents_key, child, json.dumps(parents))
+
+        r.delete(service_parent_to_children_key)
+        for parent, children in parents_to_children.items():
+            r.hset(
+                service_parent_to_children_key, parent, json.dumps(children))
+
+
+@app.task()
+def update_interface_statuses():
+    # todo - factor this connection stuff out
+    r = redis.StrictRedis(
+        host=InventoryTask.config["redis"]["hostname"],
+        port=InventoryTask.config["redis"]["port"])
+    with db.connection(InventoryTask.config["ops-db"]) as cx:
+            services = opsdb.get_circuits(cx)
+    with db.connection(InventoryTask.config["alarms-db"]) as cx:
+        with db.cursor(cx) as csr:
+            for service in services:
+                key = "{}::{}".format(
+                    service["equipment"],
+                    service["interface_name"]
+                )
+                if not r.hexists(interface_status_key, key):
+                    r.hset(interface_status_key,
+                           key,
+                           alarmsdb.get_last_known_interface_status(
+                               csr,
+                               service["equipment"],
+                               service["interface_name"]
+                           ))

test/test_external_inventory_routes.py

 import json
-from inventory_provider.storage import external_inventory
+from inventory_provider.tasks import worker
 
 DEFAULT_REQUEST_HEADERS = {
     "Content-type": "application/json",
@@ -31,7 +31,7 @@ def test_get_one_equipment_location(mocker, client):
     assert dummy_data == json.loads(rv.data.decode("utf-8"))
 
     mocked_hget.assert_called_with(
-        external_inventory.equipment_locations_key,
+        worker.equipment_locations_key,
         "dummy-equipment"
     )
 
@@ -48,7 +48,7 @@ def test_get_equipment_location(mocker, client):
     assert rv.is_json
 
     mocked_hgetall.assert_called_with(
-        external_inventory.equipment_locations_key
+        worker.equipment_locations_key
     )
 
 
@@ -64,7 +64,7 @@ def test_get_interface_info(mocker, client):
     assert rv.is_json
 
     mocked_hgetall.assert_called_with(
-        external_inventory.interfaces_key
+        worker.interfaces_key
     )
 
 
@@ -80,7 +80,7 @@ def test_get_interface_info_for_equipment(mocker, client):
     assert rv.is_json
 
     mocked_hscan_iter.assert_called_with(
-        external_inventory.interfaces_key, "dummy-equipment::*"
+        worker.interfaces_key, "dummy-equipment::*"
     )
 
 
@@ -96,7 +96,7 @@ def test_get_interface_info_for_equipment_and_interface(mocker, client):
     assert rv.is_json
 
     mocked_hget.assert_called_with(
-        external_inventory.interfaces_key, "dummy-equipment::xe-2/3/1"
+        worker.interfaces_key, "dummy-equipment::xe-2/3/1"
     )
 
 
@@ -112,7 +112,7 @@ def test_get_children(mocker, client):
     assert rv.is_json
 
     mocked_hget.assert_called_with(
-        external_inventory.service_parent_to_children_key,
+        worker.service_parent_to_children_key,
         "22987"
     )
 
@@ -129,6 +129,6 @@ def test_get_parents(mocker, client):
     assert rv.is_json
 
     mocked_hget.assert_called_with(
-        external_inventory.service_parent_to_children_key,
+        worker.service_parent_to_children_key,
         "22987"
     )