Bug 7947: Move subscribers to a separate cache
[lispflowmapping.git] / mappingservice / implementation / src / main / java / org / opendaylight / lispflowmapping / implementation / MappingSystem.java
index 3446930da7b84be1c409bde73c528247f39e6bb4..7de94bb2c9231ffbaa9454a72807461b52f86be1 100644 (file)
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015 Cisco Systems, Inc.  All rights reserved.
+ * Copyright (c) 2015, 2017 Cisco Systems, Inc.  All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
@@ -8,47 +8,66 @@
 
 package org.opendaylight.lispflowmapping.implementation;
 
+import com.google.common.collect.Sets;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.EnumMap;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
-
+import java.util.concurrent.ConcurrentHashMap;
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
 import org.opendaylight.lispflowmapping.config.ConfigIni;
 import org.opendaylight.lispflowmapping.dsbackend.DataStoreBackEnd;
 import org.opendaylight.lispflowmapping.implementation.timebucket.implementation.TimeBucketMappingTimeoutService;
 import org.opendaylight.lispflowmapping.implementation.timebucket.interfaces.ISouthBoundMappingTimeoutService;
 import org.opendaylight.lispflowmapping.implementation.util.DSBEInputUtil;
+import org.opendaylight.lispflowmapping.implementation.util.LoggingUtil;
+import org.opendaylight.lispflowmapping.implementation.util.MSNotificationInputUtil;
 import org.opendaylight.lispflowmapping.implementation.util.MappingMergeUtil;
 import org.opendaylight.lispflowmapping.interfaces.dao.ILispDAO;
 import org.opendaylight.lispflowmapping.interfaces.dao.SubKeys;
+import org.opendaylight.lispflowmapping.interfaces.dao.Subscriber;
 import org.opendaylight.lispflowmapping.interfaces.mapcache.IAuthKeyDb;
 import org.opendaylight.lispflowmapping.interfaces.mapcache.ILispMapCache;
 import org.opendaylight.lispflowmapping.interfaces.mapcache.IMapCache;
 import org.opendaylight.lispflowmapping.interfaces.mapcache.IMappingSystem;
 import org.opendaylight.lispflowmapping.interfaces.mappingservice.IMappingService;
+import org.opendaylight.lispflowmapping.lisp.type.LispMessage;
 import org.opendaylight.lispflowmapping.lisp.type.MappingData;
 import org.opendaylight.lispflowmapping.lisp.util.LispAddressStringifier;
 import org.opendaylight.lispflowmapping.lisp.util.LispAddressUtil;
+import org.opendaylight.lispflowmapping.lisp.util.MaskUtil;
+import org.opendaylight.lispflowmapping.lisp.util.SourceDestKeyHelper;
 import org.opendaylight.lispflowmapping.mapcache.AuthKeyDb;
 import org.opendaylight.lispflowmapping.mapcache.MultiTableMapCache;
 import org.opendaylight.lispflowmapping.mapcache.SimpleMapCache;
+import org.opendaylight.lispflowmapping.mapcache.lisp.LispMapCacheStringifier;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.lisp.address.types.rev151105.SimpleAddress;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.lisp.address.types.rev151105.lisp.address.address.ExplicitLocatorPath;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.lisp.address.types.rev151105.lisp.address.address.Ipv4;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.lisp.address.types.rev151105.lisp.address.address.Ipv6;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.lisp.address.types.rev151105.lisp.address.address.ServicePath;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.lisp.address.types.rev151105.lisp.address.address.SourceDestKey;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.lisp.address.types.rev151105.lisp.address.address.explicit.locator.path.explicit.locator.path.Hop;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.inet.binary.types.rev160303.IpAddressBinary;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.binary.address.types.rev160504.Ipv4PrefixBinaryAfi;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.binary.address.types.rev160504.Ipv6PrefixBinaryAfi;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.binary.address.types.rev160504.augmented.lisp.address.address.Ipv4PrefixBinary;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.binary.address.types.rev160504.augmented.lisp.address.address.Ipv6PrefixBinary;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.XtrId;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.eid.container.Eid;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.locatorrecords.LocatorRecord;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.locatorrecords.LocatorRecordBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.mapping.authkey.container.MappingAuthkey;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.mapping.record.container.MappingRecord;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.mapping.record.container.MappingRecordBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.lisp.proto.rev151105.rloc.container.Rloc;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.mappingservice.rev150906.MappingChange;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.mappingservice.rev150906.MappingChanged;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.mappingservice.rev150906.MappingOrigin;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.mappingservice.rev150906.db.instance.AuthenticationKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.mappingservice.rev150906.db.instance.Mapping;
@@ -60,17 +79,21 @@ import org.slf4j.LoggerFactory;
  * mapping lookups.
  *
  * @author Florin Coras
+ * @author Lorand Jakab
  *
  */
 public class MappingSystem implements IMappingSystem {
     private static final Logger LOG = LoggerFactory.getLogger(MappingSystem.class);
     private static final String AUTH_KEY_TABLE = "authentication";
-    private boolean notificationService;
+    //private static final int TTL_RLOC_TIMED_OUT = 1;
+    private static final int TTL_NO_RLOC_KNOWN = ConfigIni.getInstance().getNegativeMappingTTL();
+    private NotificationPublishService notificationPublishService;
     private boolean mappingMerge;
     private ILispDAO dao;
     private ILispDAO sdao;
     private ILispMapCache smc;
     private IMapCache pmc;
+    private ConcurrentHashMap<Eid, Set<Subscriber>> subscriberdb = new ConcurrentHashMap<>();
     private IAuthKeyDb akdb;
     private final EnumMap<MappingOrigin, IMapCache> tableMap = new EnumMap<>(MappingOrigin.class);
     private DataStoreBackEnd dsbe;
@@ -78,9 +101,9 @@ public class MappingSystem implements IMappingSystem {
 
     private ISouthBoundMappingTimeoutService sbMappingTimeoutService;
 
-    public MappingSystem(ILispDAO dao, boolean iterateMask, boolean notifications, boolean mappingMerge) {
+    public MappingSystem(ILispDAO dao, boolean iterateMask, NotificationPublishService nps, boolean mappingMerge) {
         this.dao = dao;
-        this.notificationService = notifications;
+        this.notificationPublishService = nps;
         this.mappingMerge = mappingMerge;
         buildMapCaches();
 
@@ -155,6 +178,7 @@ public class MappingSystem implements IMappingSystem {
         tableMap.get(origin).addMapping(key, mappingData);
     }
 
+    @SuppressWarnings("unchecked")
     private void clearPresentXtrIdMappings(Eid key) {
         List<MappingData> allXtrMappingList = (List<MappingData>) (List<?>) smc.getAllXtrIdMappings(key);
 
@@ -181,6 +205,38 @@ public class MappingSystem implements IMappingSystem {
         smc.addData(key, SubKeys.TIME_BUCKET_ID, updatedBucketId);
     }
 
+    @Override
+    public MappingData addNegativeMapping(Eid key) {
+        MappingRecord mapping = buildNegativeMapping(key);
+        MappingData mappingData = new MappingData(mapping);
+        LOG.debug("Adding negative mapping for EID {}", LispAddressStringifier.getString(key));
+        LOG.trace(mappingData.getString());
+        smc.addMapping(mapping.getEid(), mappingData);
+        dsbe.addMapping(DSBEInputUtil.toMapping(MappingOrigin.Southbound, mapping.getEid(), null, mappingData));
+        return mappingData;
+    }
+
+    private MappingRecord buildNegativeMapping(Eid eid) {
+        MappingRecordBuilder recordBuilder = new MappingRecordBuilder();
+        recordBuilder.setAuthoritative(false);
+        recordBuilder.setMapVersion((short) 0);
+        recordBuilder.setEid(eid);
+        if (eid.getAddressType().equals(Ipv4PrefixBinaryAfi.class)
+                || eid.getAddressType().equals(Ipv6PrefixBinaryAfi.class)) {
+            Eid widestNegativePrefix = getWidestNegativePrefix(eid);
+            if (widestNegativePrefix != null) {
+                recordBuilder.setEid(widestNegativePrefix);
+            }
+        }
+        recordBuilder.setAction(LispMessage.NEGATIVE_MAPPING_ACTION);
+        //if (getAuthenticationKey(eid) != null) {
+        //    recordBuilder.setRecordTtl(TTL_RLOC_TIMED_OUT);
+        //} else {
+        recordBuilder.setRecordTtl(TTL_NO_RLOC_KNOWN);
+        //}
+        return recordBuilder.build();
+    }
+
     /*
      * Since this method is only called when there is a hit in the southbound Map-Register cache, and that cache is
      * not used when merge is on, it's OK to ignore the effects of timestamp changes on merging for now.
@@ -373,9 +429,13 @@ public class MappingSystem implements IMappingSystem {
         if (mappingData != null && mappingData.getXtrId() != null) {
             removeSbXtrIdSpecificMapping(key, mappingData.getXtrId(), mappingData);
         }
+
         removeFromSbTimeoutService(key);
+        Set<Subscriber> subscribers = getSubscribers(key);
         smc.removeMapping(key);
         dsbe.removeMapping(DSBEInputUtil.toMapping(MappingOrigin.Southbound, key, mappingData));
+        notifyChange(mappingData, subscribers, null, MappingChange.Removed);
+        removeSubscribers(key);
     }
 
     private void removeFromSbTimeoutService(Eid key) {
@@ -387,15 +447,30 @@ public class MappingSystem implements IMappingSystem {
 
     @Override
     public Eid getWidestNegativePrefix(Eid key) {
+        if (!MaskUtil.isMaskable(key.getAddress())) {
+            LOG.warn("Widest negative prefix only makes sense for maskable addresses!");
+            return null;
+        }
+
+        // We assume that ILispMapCache#getWidestNegativeMapping() returns null for positive mappings, and 0/0
+        // for empty cache.
         Eid nbPrefix = pmc.getWidestNegativeMapping(key);
         if (nbPrefix == null) {
+            LOG.trace("getWidestNegativePrefix NB: positive mapping, returning null");
             return null;
         }
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("getWidestNegativePrefix NB: {}", LispAddressStringifier.getString(nbPrefix));
+        }
 
         Eid sbPrefix = smc.getWidestNegativeMapping(key);
         if (sbPrefix == null) {
+            LOG.trace("getWidestNegativePrefix SB: positive mapping, returning null");
             return null;
         }
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("getWidestNegativePrefix SB: {}", LispAddressStringifier.getString(sbPrefix));
+        }
 
         // since prefixes overlap, just return the more specific (larger mask)
         if (LispAddressUtil.getIpPrefixMask(nbPrefix) < LispAddressUtil.getIpPrefixMask(sbPrefix)) {
@@ -407,13 +482,156 @@ public class MappingSystem implements IMappingSystem {
 
     @Override
     public void removeMapping(MappingOrigin origin, Eid key) {
+        Set<Subscriber> subscribers = null;
+        Set<Subscriber> dstSubscribers = null;
+        MappingData mapping = (MappingData) tableMap.get(origin).getMapping(null, key);
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Removing mapping for EID {} from {}",
+                    LispAddressStringifier.getString(key), origin);
+        }
+        if (LOG.isTraceEnabled()) {
+            LOG.trace(mapping.getString());
+        }
+
+        MappingData notificationMapping = mapping;
+
+        if (mapping != null) {
+            subscribers = getSubscribers(key);
+            // For SrcDst LCAF also send SMRs to Dst prefix
+            if (key.getAddress() instanceof SourceDestKey) {
+                Eid dstAddr = SourceDestKeyHelper.getDstBinary(key);
+                dstSubscribers = getSubscribers(dstAddr);
+                if (!(mapping.getRecord().getEid().getAddress() instanceof SourceDestKey)) {
+                    notificationMapping = new MappingData(new MappingRecordBuilder().setEid(key).build());
+                }
+            }
+        }
+
+        removeSubscribers(key);
+
         if (origin == MappingOrigin.Southbound) {
             removeFromSbTimeoutService(key);
         }
-        tableMap.get(origin).removeMapping(key);
-        if (notificationService) {
-            // TODO
+
+        if (origin == MappingOrigin.Southbound && mapping != null && mapping.isPositive().or(false)) {
+            mergeNegativePrefixes(key);
+        } else {
+            // mergeNegativePrefixes() above removes the mapping, so addNegativeMapping() will work correctly
+            tableMap.get(origin).removeMapping(key);
+        }
+
+        if (notificationMapping != null) {
+            notifyChange(notificationMapping, subscribers, dstSubscribers, MappingChange.Removed);
+        }
+    }
+
+    private void notifyChange(MappingData mapping, Set<Subscriber> subscribers, Set<Subscriber> dstSubscribers,
+            MappingChange mappingChange) {
+        MappingChanged notification = MSNotificationInputUtil.toMappingChanged(mapping, subscribers, dstSubscribers,
+                mappingChange);
+        try {
+            notificationPublishService.putNotification(notification);
+        } catch (InterruptedException e) {
+            LOG.warn("Notification publication interrupted!");
+        }
+    }
+
+
+    /*
+     * Merges adjacent negative prefixes and notifies their subscribers.
+     */
+    private void mergeNegativePrefixes(Eid eid) {
+        LOG.debug("Merging negative prefixes starting from EID {}", LispAddressStringifier.getString(eid));
+
+        // If we delete nodes while we walk up the radix trie the algorithm will give incorrect results, because
+        // removals rearrange relationships in the trie. So we save prefixes to be removed into a HashMap.
+        Map<Eid, MappingData> mergedMappings = new HashMap<>();
+
+        Eid currentNode = smc.getSiblingPrefix(eid);
+        MappingData mapping = (MappingData) smc.getMapping(null, currentNode);
+        if (mapping != null && mapping.isNegative().or(false)) {
+            mergedMappings.put(currentNode, mapping);
+        } else {
+            return;
+        }
+
+        Eid previousNode = currentNode;
+        currentNode = smc.getVirtualParentSiblingPrefix(currentNode);
+        while (currentNode != null) {
+            mapping = (MappingData) smc.getMapping(null, currentNode);
+            if (mapping != null && mapping.isNegative().or(false)) {
+                mergedMappings.put(currentNode, mapping);
+            } else {
+                break;
+            }
+            previousNode = currentNode;
+            currentNode = smc.getVirtualParentSiblingPrefix(previousNode);
+        }
+
+        for (Eid key : mergedMappings.keySet()) {
+            removeSbMapping(key, mergedMappings.get(key));
+        }
+        smc.removeMapping(eid);
+
+        addNegativeMapping(getVirtualParent(previousNode));
+    }
+
+    private static Eid getVirtualParent(Eid eid) {
+        if (eid.getAddress() instanceof Ipv4PrefixBinary) {
+            Ipv4PrefixBinary prefix = (Ipv4PrefixBinary) eid.getAddress();
+            short parentPrefixLength = (short) (prefix.getIpv4MaskLength() - 1);
+            byte[] parentPrefix = MaskUtil.normalizeByteArray(prefix.getIpv4AddressBinary().getValue(),
+                    parentPrefixLength);
+            return LispAddressUtil.asIpv4PrefixBinaryEid(eid, parentPrefix, parentPrefixLength);
+        } else if (eid.getAddress() instanceof Ipv6PrefixBinary) {
+            Ipv6PrefixBinary prefix = (Ipv6PrefixBinary) eid.getAddress();
+            short parentPrefixLength = (short) (prefix.getIpv6MaskLength() - 1);
+            byte[] parentPrefix = MaskUtil.normalizeByteArray(prefix.getIpv6AddressBinary().getValue(),
+                    parentPrefixLength);
+            return LispAddressUtil.asIpv6PrefixBinaryEid(eid, parentPrefix, parentPrefixLength);
+        }
+        return null;
+    }
+
+    @Override
+    public synchronized void subscribe(Subscriber subscriber, Eid subscribedEid) {
+        Set<Subscriber> subscribers = getSubscribers(subscribedEid);
+        if (subscribers == null) {
+            subscribers = Sets.newConcurrentHashSet();
+        } else if (subscribers.contains(subscriber)) {
+            // If there is an entry already for this subscriber, remove it, so that it gets the new timestamp
+            subscribers.remove(subscriber);
+        }
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Adding new subscriber {} for EID {}", subscriber.getString(),
+                    LispAddressStringifier.getString(subscribedEid));
+        }
+        subscribers.add(subscriber);
+        addSubscribers(subscribedEid, subscribers);
+    }
+
+    private void addSubscribers(Eid address, Set<Subscriber> subscribers) {
+        LoggingUtil.logSubscribers(LOG, address, subscribers);
+        subscriberdb.put(address, subscribers);
+    }
+
+    @Override
+    public Set<Subscriber> getSubscribers(Eid address) {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Retrieving subscribers for EID {}", LispAddressStringifier.getString(address));
+        }
+
+        Set<Subscriber> subscribers = subscriberdb.get(address);
+        LoggingUtil.logSubscribers(LOG, address, subscribers);
+        return subscribers;
+    }
+
+    private void removeSubscribers(Eid address) {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Removing subscribers for EID {}", LispAddressStringifier.getString(address));
         }
+        subscriberdb.remove(address);
     }
 
     @Override
@@ -442,7 +660,7 @@ public class MappingSystem implements IMappingSystem {
     @Override
     public void addData(MappingOrigin origin, Eid key, String subKey, Object data) {
         if (LOG.isDebugEnabled()) {
-            LOG.debug("Add data of class {} for key {} and subkey {}", data.getClass(),
+            LOG.debug("Add data of {} for key {} and subkey {}", data.getClass(),
                     LispAddressStringifier.getString(key), subKey);
         }
         tableMap.get(origin).addData(key, subKey, data);
@@ -514,27 +732,53 @@ public class MappingSystem implements IMappingSystem {
     @Override
     public String printMappings() {
         final StringBuffer sb = new StringBuffer();
-        sb.append("PolicyMapCache\n--------------\n");
+        sb.append("Policy map-cache\n----------------\n");
         sb.append(pmc.printMappings());
-        sb.append("SbMapCache\n----------\n");
+        sb.append("\nSouthbound map-cache\n--------------------\n");
         sb.append(smc.printMappings());
         return sb.toString();
     }
 
+    @Override
+    public String prettyPrintMappings() {
+        final StringBuffer sb = new StringBuffer();
+        sb.append("Policy map-cache\n----------------\n");
+        sb.append(pmc.prettyPrintMappings());
+        sb.append("\nSouthbound map-cache\n--------------------\n");
+        sb.append(smc.prettyPrintMappings());
+        sb.append("\nSubscribers\n-----------\n");
+        sb.append(prettyPrintSubscribers(subscriberdb));
+        return sb.toString();
+    }
+
+    private static String prettyPrintSubscribers(Map<Eid, Set<Subscriber>> subscribers) {
+        final StringBuffer sb = new StringBuffer();
+        for (Eid eid: subscribers.keySet()) {
+            sb.append("\n  ");
+            sb.append(LispAddressStringifier.getString(eid));
+            sb.append("\n");
+            sb.append(LispMapCacheStringifier.prettyPrintSubscriberSet(subscribers.get(eid), 4));
+            sb.append("\n");
+        }
+        return sb.toString();
+    }
+
     @Override
     public String printKeys() {
         return akdb.printKeys();
     }
 
+    @Override
+    public String prettyPrintKeys() {
+        return akdb.prettyPrintKeys();
+    }
+
     public void cleanCaches() {
         dao.removeAll();
+        subscriberdb.clear();
         buildMapCaches();
     }
 
-    /*
-     * XXX  Mappings and keys should be separated for this to work properly, as is it will remove northbound originated
-     * authentication keys too, since they are currently stored in smc.
-     */
     public void cleanSBMappings() {
         smc = new SimpleMapCache(sdao);
     }