Finished release 0.23.

changelog

@@ -37,3 +37,4 @@
 0.20: included both v4 & v6 addresses in peering info
 0.21: added parsing of 'logical-systems' (DBOARD3-150)
 0.22: return a skeleton response for unknown interfaces (DBOARD3-169)
+0.23: use redis pipelines where possible

inventory_provider/tasks/worker.py

@@ -105,10 +105,12 @@ def update_interfaces_to_services():
     r = get_next_redis(InventoryTask.config)
     for key in r.scan_iter('opsdb:interface_services:*'):
         r.delete(key)
+    rp = r.pipeline()
     for equipment_interface, services in interface_services.items():
-        r.set(
+        rp.set(
             'opsdb:interface_services:' + equipment_interface,
             json.dumps(services))
+    rp.execute()
 
     logger.debug('<<< update_interfaces_to_services')
 
@@ -150,13 +152,15 @@ def update_circuit_hierarchy():
         r = get_next_redis(InventoryTask.config)
         for key in r.scan_iter('opsdb:services:parents:*'):
             r.delete(key)
-        for cid, parents in parent_to_children.items():
-            r.set('opsdb:services:parents:%d' % cid, json.dumps(parents))
-
         for key in r.scan_iter('opsdb:services:children:*'):
             r.delete(key)
+
+        rp = r.pipeline()
+        for cid, parents in parent_to_children.items():
+            rp.set('opsdb:services:parents:%d' % cid, json.dumps(parents))
         for cid, children in child_to_parents.items():
-            r.set('opsdb:services:children:%d' % cid, json.dumps(children))
+            rp.set('opsdb:services:children:%d' % cid, json.dumps(children))
+        rp.execute()
 
     logger.debug('<<< update_circuit_hierarchy')
 
@@ -170,9 +174,12 @@ def update_geant_lambdas():
     for key in r.scan_iter('opsdb:geant_lambdas:*'):
         r.delete(key)
     with db.connection(InventoryTask.config["ops-db"]) as cx:
+        rp = r.pipeline()
         for ld in opsdb.get_geant_lambdas(cx):
-            r.set('opsdb:geant_lambdas:%s' % ld['name'].lower(),
-                  json.dumps(ld))
+            rp.set(
+                'opsdb:geant_lambdas:%s' % ld['name'].lower(),
+                json.dumps(ld))
+        rp.execute()
 
     logger.debug('<<< geant_lambdas')
 
@@ -203,10 +210,12 @@ def update_junosspace_device_list(self):
             'message': 'found %d routers, saving details' % len(routers)
         })
 
-    for k in r.keys('junosspace:*'):
+    for k in r.scan_iter('junosspace:*'):
         r.delete(k)
+    rp = r.pipeline()
     for k, v in routers.items():
-        r.set(k, v)
+        rp.set(k, v)
+    rp.execute()
 
     logger.debug('<<< update_junosspace_device_list')
 
@@ -268,7 +277,7 @@ def _refresh_peers(hostname, key_base, peers):
     logger.debug(
         'removing cached %s for %r' % (key_base, hostname))
     r = get_next_redis(InventoryTask.config)
-    for k in r.keys(key_base + ':*'):
+    for k in r.scan_iter(key_base + ':*'):
         # potential race condition: another proc could have
         # delete this element between the time we read the
         # keys and the next statement ... check for None below
@@ -278,11 +287,13 @@ def _refresh_peers(hostname, key_base, peers):
             if value['router'] == hostname:
                 r.delete(k)
 
+    rp = r.pipeline()
     for peer in peers:
         peer['router'] = hostname
-        r.set(
+        rp.set(
             '%s:%s' % (key_base, peer['name']),
             json.dumps(peer))
+    rp.execute()
 
 
 def refresh_ix_public_peers(hostname, netconf):
@@ -312,26 +323,27 @@ def refresh_juniper_interface_list(hostname, netconf):
         'removing cached netconf-interfaces for %r' % hostname)
 
     r = get_next_redis(InventoryTask.config)
-    for k in r.keys('netconf-interfaces:%s:*' % hostname):
+    for k in r.scan_iter('netconf-interfaces:%s:*' % hostname):
         r.delete(k)
-
     for k in r.keys('netconf-interface-bundles:%s:*' % hostname):
         r.delete(k)
 
     all_bundles = defaultdict(list)
+
+    rp = r.pipeline()
     for ifc in juniper.list_interfaces(netconf):
         bundles = ifc.get('bundle', None)
         for bundle in bundles:
             if bundle:
                 all_bundles[bundle].append(ifc['name'])
-
-        r.set(
+        rp.set(
             'netconf-interfaces:%s:%s' % (hostname, ifc['name']),
             json.dumps(ifc))
     for k, v in all_bundles.items():
-        r.set(
+        rp.set(
             'netconf-interface-bundles:%s:%s' % (hostname, k),
             json.dumps(v))
+    rp.execute()
 
 
 @app.task(base=InventoryTask, bind=True)
@@ -427,7 +439,7 @@ def _derive_router_hostnames(config):
         junosspace_equipment.add(m.group(1))
 
     opsdb_equipment = set()
-    for k in r.keys('opsdb:interface_services:*'):
+    for k in r.scan_iter('opsdb:interface_services:*'):
         m = re.match(
             'opsdb:interface_services:([^:]+):.*$',
             k.decode('utf-8'))
@@ -558,8 +570,10 @@ def _build_subnet_db(update_callback=lambda s: None):
 
     update_callback('saving {} subnets'.format(len(subnets)))
 
+    rp = r.pipeline()
     for k, v in subnets.items():
-        r.set('subnets:' + k, json.dumps(v))
+        rp.set('subnets:' + k, json.dumps(v))
+    rp.execute()
 
 
 def check_task_status(task_id):

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name='inventory-provider',
-    version="0.22",
+    version="0.23",
     author='GEANT',
     author_email='swd@geant.org',
     description='Dashboard inventory provider',

test/conftest.py

@@ -98,12 +98,12 @@ class MockedRedis(object):
 
     def scan_iter(self, glob=None):
         if not glob:
-            for k in MockedRedis.db.keys():
+            for k in list(MockedRedis.db.keys()):
                 yield k.encode('utf-8')
 
         m = re.match(r'^([^*]+)\*$', glob)
         assert m  # all expected globs are like this
-        for k in MockedRedis.db.keys():
+        for k in list(MockedRedis.db.keys()):
             if k.startswith(m.group(1)):
                 yield k.encode('utf-8')
 
@@ -114,6 +114,12 @@ class MockedRedis(object):
         # only called from testing routes (hopefully)
         pass
 
+    def execute(self):
+        pass
+
+    def pipeline(self, *args, **kwargs):
+        return self
+
 
 @pytest.fixture
 def cached_test_data():

test/test_classifier_routes.py

@@ -140,6 +140,26 @@ def test_juniper_link_info(client):
     jsonschema.validate(response_data, JUNIPER_LINK_METADATA)
 
 
+def test_juniper_link_info_not_found(client):
+    rv = client.get(
+        '/classifier/juniper-link-info/'
+        'mx1.ams.nl.geant.net/unknown-interface-name',
+        headers=DEFAULT_REQUEST_HEADERS)
+    assert rv.status_code == 200
+    assert rv.is_json
+    response_data = json.loads(rv.data.decode('utf-8'))
+    jsonschema.validate(response_data, JUNIPER_LINK_METADATA)
+    assert response_data == {
+        'interface': {
+            'name': 'unknown-interface-name',
+            'description': '',
+            'ipv4': [],
+            'ipv6': [],
+            'bundle': []
+        }
+    }
+
+
 VPN_RR_PEER_INFO_KEYS = {'vpn-rr-peer-info'}
 IX_PUBLIC_PEER_INFO_KEYS = {'ix-public-peer-info', 'interfaces'}