fix race condition

iterating over keys & then non-atomically fetching value ...

inventory_provider/tasks/worker.py

@@ -238,38 +238,40 @@ def clear_cached_classifier_responses(hostname):
         r.delete(k)
 
 
-def refresh_ix_public_peers(hostname, netconf):
+def _refresh_peers(hostname, key_base, peers):
     task_logger = logging.getLogger(constants.TASK_LOGGER_NAME)
     task_logger.debug(
-        'removing cached ix public peers for %r' % hostname)
+        'removing cached %s for %r' % (key_base, hostname))
     r = get_redis(InventoryTask.config)
-    for k in r.keys('ix_public_peer:*'):
-        value = json.loads(r.get(k.decode('utf-8')).decode('utf-8'))
-        if value['router'] == hostname:
-            r.delete(k)
-
-    for peer in juniper.ix_public_peers(netconf):
+    for k in r.keys(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
+        value = r.get(k.decode('utf-8'))
+        if value:
+            value = json.loads(value.decode('utf-8'))
+            if value['router'] == hostname:
+                r.delete(k)
+
+    for peer in peers:
         peer['router'] = hostname
         r.set(
-            'ix_public_peer:' + peer['name'],
+            '%s:%s' % (key_base, peer['name']),
             json.dumps(peer))
 
 
-def refresh_vpn_rr_peers(hostname, netconf):
-    task_logger = logging.getLogger(constants.TASK_LOGGER_NAME)
-    task_logger.debug(
-        'removing cached vpn rr for %r' % hostname)
-    r = get_redis(InventoryTask.config)
-    for k in r.keys('vpn_rr_peer:*'):
-        value = json.loads(r.get(k.decode('utf-8')).decode('utf-8'))
-        if value['router'] == hostname:
-            r.delete(k)
+def refresh_ix_public_peers(hostname, netconf):
+    _refresh_peers(
+        hostname,
+        'ix_public_peer',
+        juniper.ix_public_peers(netconf))
 
-    for peer in juniper.vpn_rr_peers(netconf):
-        peer['router'] = hostname
-        r.set(
-            'vpn_rr_peer:' + peer['name'],
-            json.dumps(peer))
+
+def refresh_vpn_rr_peers(hostname, netconf):
+    _refresh_peers(
+        hostname,
+        'vpn_rr_peer',
+        juniper.vpn_rr_peers(netconf))
 
 
 @app.task