use get_next/current_redis

inventory_provider/routes/classifier.py

@@ -50,7 +50,7 @@ def base_interface_name(interface):
 
 
 def related_interfaces(hostname, interface):
-    r = common.get_redis()
+    r = common.get_current_redis()
     prefix = 'netconf-interfaces:%s:' % hostname
     for k in r.keys(prefix + base_interface_name(interface) + '.*'):
         k = k.decode('utf-8')
@@ -63,7 +63,7 @@ def related_interfaces(hostname, interface):
               methods=['GET', 'POST'])
 @common.require_accepts_json
 def get_juniper_link_info(source_equipment, interface):
-    r = common.get_redis()
+    r = common.get_current_redis()
 
     cache_key = 'classifier-cache:juniper:%s:%s' % (
         source_equipment, interface)
@@ -137,7 +137,7 @@ def ix_peering_info(peer_info):
     protocol = type(address).__name__
     keyword = description.split(' ')[0]  # regex needed??? (e.g. tabs???)
 
-    r = common.get_redis()
+    r = common.get_current_redis()
 
     for k in r.keys('ix_public_peer:*'):
         other = r.get(k.decode('utf-8')).decode('utf-8')
@@ -165,7 +165,7 @@ def find_interfaces(address):
     :param address: an ipaddress object
     :return:
     """
-    r = common.get_redis()
+    r = common.get_current_redis()
     for k in r.keys('reverse_interface_addresses:*'):
         info = r.get(k.decode('utf-8')).decode('utf-8')
         info = json.loads(info)
@@ -187,7 +187,7 @@ def find_interfaces_and_services(address_str):
         raise ClassifierProcessingError(
             'unable to parse %r as an ip address' % address_str)
 
-    r = common.get_redis()
+    r = common.get_current_redis()
     for interface in find_interfaces(address):
 
         services = r.get(
@@ -209,7 +209,7 @@ def find_interfaces_and_services(address_str):
 @common.require_accepts_json
 def peer_info(address):
 
-    r = common.get_redis()
+    r = common.get_current_redis()
 
     cache_key = 'classifier-cache:peer:%s' % address
 
@@ -257,7 +257,7 @@ def get_trap_metadata(source_equipment, interface, circuit_id):
     cache_key = 'classifier-cache:infinera:%s:%s' % (
         source_equipment, interface)
 
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = r.get(cache_key)
 
     if result:
@@ -294,7 +294,7 @@ def get_trap_metadata(source_equipment, interface, circuit_id):
               methods=['GET', 'POST'])
 @common.require_accepts_json
 def get_coriant_info(equipment_name, entity_string):
-    r = common.get_redis()
+    r = common.get_current_redis()
 
     cache_key = 'classifier-cache:coriant:%s:%s' % (
         equipment_name, entity_string)

inventory_provider/routes/data.py

@@ -18,7 +18,7 @@ def after_request(resp):
 @routes.route("/routers", methods=['GET', 'POST'])
 @common.require_accepts_json
 def routers():
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = []
     for k in r.keys('netconf:*'):
         m = re.match('^netconf:(.+)$', k.decode('utf-8'))
@@ -30,7 +30,7 @@ def routers():
 @routes.route("/interfaces/<hostname>", methods=['GET', 'POST'])
 @common.require_accepts_json
 def router_interfaces(hostname):
-    r = common.get_redis()
+    r = common.get_current_redis()
     interfaces = []
     for k in r.keys('netconf-interfaces:%s:*' % hostname):
         ifc = r.get(k.decode('utf-8'))

inventory_provider/routes/poller.py

@@ -16,7 +16,7 @@ def after_request(resp):
 @routes.route('/interfaces/<hostname>', methods=['GET', 'POST'])
 @common.require_accepts_json
 def poller_interface_oids(hostname):
-    r = common.get_redis()
+    r = common.get_current_redis()
 
     netconf_string = r.get('netconf:' + hostname)
     if not netconf_string:

inventory_provider/routes/testing.py

@@ -14,7 +14,7 @@ routes = Blueprint("inventory-data-testing-support-routes", __name__)
 
 @routes.route("flushdb", methods=['GET', 'POST'])
 def flushdb():
-    common.get_redis().flushdb()
+    common.get_current_redis().flushdb()
     return Response('OK')
 
 
@@ -46,7 +46,7 @@ def update_interface_statuses():
 @common.require_accepts_json
 def juniper_addresses():
     # TODO: this route (and corant, infinera routes) can be removed
-    r = common.get_redis()
+    r = common.get_current_redis()
     routers = []
     for k in r.keys('junosspace:*'):
         info = r.get(k.decode('utf-8'))
@@ -58,7 +58,7 @@ def juniper_addresses():
 
 @routes.route("opsdb/interfaces")
 def get_all_interface_details():
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = collections.defaultdict(list)
     for k in r.keys('opsdb:interface_services:*'):
         m = re.match(
@@ -71,7 +71,7 @@ def get_all_interface_details():
 
 @routes.route("opsdb/interfaces/<equipment_name>")
 def get_interface_details_for_equipment(equipment_name):
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = []
     for k in r.keys('opsdb:interface_services:%s:*' % equipment_name):
         m = re.match(
@@ -84,7 +84,7 @@ def get_interface_details_for_equipment(equipment_name):
 
 @routes.route("opsdb/interfaces/<equipment_name>/<path:interface>")
 def get_interface_details(equipment_name, interface):
-    r = common.get_redis()
+    r = common.get_current_redis()
     key = 'opsdb:interface_services:%s:%s' % (equipment_name, interface)
     # TODO: handle None (return 404)
     return jsonify(json.loads(r.get(key).decode('utf-8')))
@@ -92,7 +92,7 @@ def get_interface_details(equipment_name, interface):
 
 @routes.route("opsdb/equipment-location")
 def get_all_equipment_locations():
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = {}
     for k in r.keys('opsdb:location:*'):
         k = k.decode('utf-8')
@@ -104,7 +104,7 @@ def get_all_equipment_locations():
 
 @routes.route("opsdb/equipment-location/<path:equipment_name>")
 def get_equipment_location(equipment_name):
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = r.get('opsdb:location:' + equipment_name)
     # TODO: handle None (return 404)
     return jsonify(json.loads(result.decode('utf-8')))
@@ -112,7 +112,7 @@ def get_equipment_location(equipment_name):
 
 @routes.route("opsdb/circuit-hierarchy/children/<int:parent_id>")
 def get_children(parent_id):
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = r.get('opsdb:services:children:%d' % parent_id)
     # TODO: handle None (return 404)
     return jsonify(json.loads(result.decode('utf-8')))
@@ -120,7 +120,7 @@ def get_children(parent_id):
 
 @routes.route("opsdb/circuit-hierarchy/parents/<int:child_id>")
 def get_parents(child_id):
-    r = common.get_redis()
+    r = common.get_current_redis()
     result = r.get('opsdb:services:parents:%d' % child_id)
     # TODO: handle None (return 404)
     return jsonify(json.loads(result.decode('utf-8')))
@@ -129,7 +129,7 @@ def get_parents(child_id):
 @routes.route("bgp/<hostname>", methods=['GET', 'POST'])
 @common.require_accepts_json
 def bgp_configs(hostname):
-    r = common.get_redis()
+    r = common.get_current_redis()
     netconf_string = r.get('netconf:' + hostname)
     if not netconf_string:
         return Response(
@@ -151,7 +151,7 @@ def bgp_configs(hostname):
 @routes.route("snmp/<hostname>", methods=['GET', 'POST'])
 @common.require_accepts_json
 def snmp_ids(hostname):
-    r = common.get_redis()
+    r = common.get_next_redis()
     ifc_data_string = r.get('snmp-interfaces:' + hostname)
     ifc_data = json.loads(ifc_data_string.decode('utf-8'))
     return jsonify(ifc_data)

inventory_provider/tasks/worker.py

@@ -10,7 +10,7 @@ from collections import defaultdict
 from lxml import etree
 
 from inventory_provider.tasks.app import app
-from inventory_provider.tasks.common import get_redis
+from inventory_provider.tasks.common import get_next_redis
 from inventory_provider import config
 from inventory_provider import environment
 from inventory_provider.db import db, opsdb
@@ -55,38 +55,16 @@ class InventoryTask(Task):
         super().update_state(**kwargs)
 
 
-def _save_value(key, value):
-    assert isinstance(value, str), \
-        "sanity failure: expected string data as value"
-    r = get_redis(InventoryTask.config)
-    r.set(name=key, value=value)
-    # InventoryTask.logger.debug("saved %s" % key)
-    return "OK"
-
-
-def _save_value_json(key, data_obj):
-    _save_value(
-        key,
-        json.dumps(data_obj))
-
-
-def _save_value_etree(key, xml_doc):
-    _save_value(
-        key,
-        etree.tostring(xml_doc, encoding='unicode'))
-
-
 @app.task
 def snmp_refresh_interfaces(hostname, community):
     logger = logging.getLogger(__name__)
     logger.debug(
         '>>> snmp_refresh_interfaces(%r, %r)' % (hostname, community))
 
-    _save_value_json(
-        'snmp-interfaces:' + hostname,
-        list(snmp.get_router_snmp_indexes(
-            hostname,
-            community)))
+    value = list(snmp.get_router_snmp_indexes(hostname, community))
+
+    r = get_next_redis(InventoryTask.config)
+    r.set('snmp-interfaces:' + hostname, json.dumps(value))
 
     logger.debug(
         '<<< snmp_refresh_interfaces(%r, %r)' % (hostname, community))
@@ -97,9 +75,11 @@ def netconf_refresh_config(hostname):
     logger = logging.getLogger(__name__)
     logger.debug('>>> netconf_refresh_config(%r)' % hostname)
 
-    _save_value_etree(
-        'netconf:' + hostname,
-        juniper.load_config(hostname, InventoryTask.config["ssh"]))
+    netconf_doc = juniper.load_config(hostname, InventoryTask.config["ssh"])
+    netconf_str = etree.tostring(netconf_doc, encoding='unicode')
+
+    r = get_next_redis(InventoryTask.config)
+    r.set('netconf:' + hostname, netconf_str)
 
     logger.debug('<<< netconf_refresh_config(%r)' % hostname)
 
@@ -116,7 +96,7 @@ def update_interfaces_to_services():
                 service['equipment'], service['interface_name'])
             interface_services[equipment_interface].append(service)
 
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
     for key in r.scan_iter('opsdb:interface_services:*'):
         r.delete(key)
     for equipment_interface, services in interface_services.items():
@@ -132,7 +112,7 @@ def update_equipment_locations():
     logger = logging.getLogger(__name__)
     logger.debug('>>> update_equipment_locations')
 
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
     for k in r.scan_iter('opsdb:location:*'):
         r.delete(k)
 
@@ -161,7 +141,7 @@ def update_circuit_hierarchy():
             parent_to_children[parent_id].append(relation)
             child_to_parents[child_id].append(relation)
 
-        r = get_redis(InventoryTask.config)
+        r = get_next_redis(InventoryTask.config)
         for key in r.scan_iter('opsdb:services:parents:*'):
             r.delete(key)
         for cid, parents in child_to_parents.items():
@@ -180,7 +160,7 @@ def update_geant_lambdas():
     logger = logging.getLogger(__name__)
     logger.debug('>>> update_geant_lambdas')
 
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
     for key in r.scan_iter('opsdb:geant_lambdas:*'):
         r.delete(key)
     with db.connection(InventoryTask.config["ops-db"]) as cx:
@@ -203,7 +183,7 @@ def update_junosspace_device_list(self):
             'message': 'querying junosspace for managed routers'
         })
 
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
 
     routers = {}
     for d in juniper.load_routers_from_junosspace(
@@ -237,7 +217,7 @@ def load_netconf_data(hostname):
     :param hostname:
     :return:
     """
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
     netconf = r.get('netconf:' + hostname)
     if not netconf:
         raise InventoryTaskError('no netconf data found for %r' % hostname)
@@ -252,7 +232,7 @@ def clear_cached_classifier_responses(hostname=None):
     else:
         logger.debug('removing all cached classifier responses')
 
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
 
     def _hostname_keys():
         for k in r.keys('classifier-cache:juniper:%s:*' % hostname):
@@ -281,7 +261,7 @@ def _refresh_peers(hostname, key_base, peers):
     logger = logging.getLogger(__name__)
     logger.debug(
         'removing cached %s for %r' % (key_base, hostname))
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
     for k in r.keys(key_base + ':*'):
         # potential race condition: another proc could have
         # delete this element between the time we read the
@@ -325,7 +305,7 @@ def refresh_juniper_interface_list(hostname, netconf):
     logger.debug(
         'removing cached netconf-interfaces for %r' % hostname)
 
-    r = get_redis(InventoryTask.config)
+    r = get_next_redis(InventoryTask.config)
     for k in r.keys('netconf-interfaces:%s:*' % hostname):
         r.delete(k)
 
@@ -433,7 +413,7 @@ def reload_router_config(self, hostname):
 
 def _derive_router_hostnames(config):
     logger = logging.getLogger(__name__)
-    r = get_redis(config)
+    r = get_next_redis(config)
     junosspace_equipment = set()
     for k in r.keys('junosspace:*'):
         m = re.match('^junosspace:(.*)$', k.decode('utf-8'))