core/store/dist/src/main/java/org/onosproject/store/link/impl/ECLinkStore.java

/*
 * Copyright 2015 Open Networking Laboratory
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.onosproject.store.link.impl;

import static com.google.common.base.Preconditions.checkNotNull;
import static org.onosproject.net.DefaultAnnotations.merge;
import static org.onosproject.net.DefaultAnnotations.union;
import static org.onosproject.net.Link.State.ACTIVE;
import static org.onosproject.net.Link.State.INACTIVE;
import static org.onosproject.net.Link.Type.DIRECT;
import static org.onosproject.net.Link.Type.INDIRECT;
import static org.onosproject.net.LinkKey.linkKey;
import static org.onosproject.net.link.LinkEvent.Type.LINK_ADDED;
import static org.onosproject.net.link.LinkEvent.Type.LINK_REMOVED;
import static org.onosproject.net.link.LinkEvent.Type.LINK_UPDATED;
import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.PUT;
import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.REMOVE;
import static org.slf4j.LoggerFactory.getLogger;

import java.util.Collection;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.util.KryoNamespace;
import org.onlab.util.SharedExecutors;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.NodeId;
import org.onosproject.mastership.MastershipService;
import org.onosproject.net.AnnotationKeys;
import org.onosproject.net.AnnotationsUtil;
import org.onosproject.net.ConnectPoint;
import org.onosproject.net.DefaultAnnotations;
import org.onosproject.net.DefaultLink;
import org.onosproject.net.DeviceId;
import org.onosproject.net.Link;
import org.onosproject.net.LinkKey;
import org.onosproject.net.Link.Type;
import org.onosproject.net.device.DeviceClockService;
import org.onosproject.net.link.DefaultLinkDescription;
import org.onosproject.net.link.LinkDescription;
import org.onosproject.net.link.LinkEvent;
import org.onosproject.net.link.LinkStore;
import org.onosproject.net.link.LinkStoreDelegate;
import org.onosproject.net.provider.ProviderId;
import org.onosproject.store.AbstractStore;
import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
import org.onosproject.store.cluster.messaging.MessageSubject;
import org.onosproject.store.impl.MastershipBasedTimestamp;
import org.onosproject.store.serializers.KryoNamespaces;
import org.onosproject.store.serializers.KryoSerializer;
import org.onosproject.store.serializers.custom.DistributedStoreSerializers;
import org.onosproject.store.service.EventuallyConsistentMap;
import org.onosproject.store.service.EventuallyConsistentMapEvent;
import org.onosproject.store.service.EventuallyConsistentMapListener;
import org.onosproject.store.service.StorageService;
import org.slf4j.Logger;

import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Futures;

/**
 * Manages the inventory of links using a {@code EventuallyConsistentMap}.
 */
@Component(immediate = true, enabled = true)
@Service
public class ECLinkStore
    extends AbstractStore<LinkEvent, LinkStoreDelegate>
    implements LinkStore {

    private final Logger log = getLogger(getClass());

    private final Map<LinkKey, Link> links = Maps.newConcurrentMap();
    private EventuallyConsistentMap<Provided<LinkKey>, LinkDescription> linkDescriptions;

    private static final MessageSubject LINK_INJECT_MESSAGE = new MessageSubject("inject-link-request");

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected StorageService storageService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected MastershipService mastershipService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected DeviceClockService deviceClockService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected ClusterCommunicationService clusterCommunicator;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected ClusterService clusterService;

    private EventuallyConsistentMapListener<Provided<LinkKey>, LinkDescription> linkTracker =
            new InternalLinkTracker();

    protected static final KryoSerializer SERIALIZER = new KryoSerializer() {
        @Override
        protected void setupKryoPool() {
            serializerPool = KryoNamespace.newBuilder()
                    .register(DistributedStoreSerializers.STORE_COMMON)
                    .nextId(DistributedStoreSerializers.STORE_CUSTOM_BEGIN)
                    .register(Provided.class)
                    .build();
        }
    };

    @Activate
    public void activate() {
        KryoNamespace.Builder serializer = KryoNamespace.newBuilder()
                .register(KryoNamespaces.API)
                .register(MastershipBasedTimestamp.class)
                .register(Provided.class);

        linkDescriptions = storageService.<Provided<LinkKey>, LinkDescription>eventuallyConsistentMapBuilder()
                .withName("onos-link-descriptions")
                .withSerializer(serializer)
                .withTimestampProvider((k, v) -> {
                    try {
                        return v == null ? null : deviceClockService.getTimestamp(v.dst().deviceId());
                    } catch (IllegalStateException e) {
                        return null;
                    }
                }).build();

        clusterCommunicator.addSubscriber(LINK_INJECT_MESSAGE,
                SERIALIZER::decode,
                this::injectLink,
                SERIALIZER::encode,
                SharedExecutors.getPoolThreadExecutor());

        linkDescriptions.addListener(linkTracker);

        log.info("Started");
    }

    @Deactivate
    public void deactivate() {
        linkDescriptions.removeListener(linkTracker);
        linkDescriptions.destroy();
        links.clear();
        clusterCommunicator.removeSubscriber(LINK_INJECT_MESSAGE);

        log.info("Stopped");
    }

    @Override
    public int getLinkCount() {
        return links.size();
    }

    @Override
    public Iterable<Link> getLinks() {
        return links.values();
    }

    @Override
    public Set<Link> getDeviceEgressLinks(DeviceId deviceId) {
        return filter(links.values(), link -> deviceId.equals(link.src().deviceId()));
    }

    @Override
    public Set<Link> getDeviceIngressLinks(DeviceId deviceId) {
        return filter(links.values(), link -> deviceId.equals(link.dst().deviceId()));
    }

    @Override
    public Link getLink(ConnectPoint src, ConnectPoint dst) {
        return links.get(linkKey(src, dst));
    }

    @Override
    public Set<Link> getEgressLinks(ConnectPoint src) {
        return filter(links.values(), link -> src.equals(link.src()));
    }

    @Override
    public Set<Link> getIngressLinks(ConnectPoint dst) {
        return filter(links.values(), link -> dst.equals(link.dst()));
    }

    @Override
    public LinkEvent createOrUpdateLink(ProviderId providerId,
                                        LinkDescription linkDescription) {
        final DeviceId dstDeviceId = linkDescription.dst().deviceId();
        final NodeId dstNodeId = mastershipService.getMasterFor(dstDeviceId);

        // Process link update only if we're the master of the destination node,
        // otherwise signal the actual master.
        if (clusterService.getLocalNode().id().equals(dstNodeId)) {
            LinkKey linkKey = linkKey(linkDescription.src(), linkDescription.dst());
            Provided<LinkKey> internalLinkKey = new Provided<>(linkKey, providerId);
            linkDescriptions.compute(internalLinkKey, (k, v) -> createOrUpdateLinkInternal(v  , linkDescription));
            return refreshLinkCache(linkKey);
        } else {
            if (dstNodeId == null) {
                return null;
            }
            return Futures.getUnchecked(clusterCommunicator.sendAndReceive(new Provided<>(linkDescription, providerId),
                    LINK_INJECT_MESSAGE,
                    SERIALIZER::encode,
                    SERIALIZER::decode,
                    dstNodeId));
        }
    }

    private LinkDescription createOrUpdateLinkInternal(LinkDescription current, LinkDescription updated) {
        if (current != null) {
            // we only allow transition from INDIRECT -> DIRECT
            return  new DefaultLinkDescription(
                        current.src(),
                        current.dst(),
                        current.type() == DIRECT ? DIRECT : updated.type(),
                        union(current.annotations(), updated.annotations()));
        }
        return updated;
    }

    private LinkEvent refreshLinkCache(LinkKey linkKey) {
        AtomicReference<LinkEvent.Type> eventType = new AtomicReference<>();
        Link link = links.compute(linkKey, (key, existingLink) -> {
            Link newLink = composeLink(linkKey);
            if (existingLink == null) {
                eventType.set(LINK_ADDED);
                return newLink;
            } else if (existingLink.state() != newLink.state() ||
                        (existingLink.type() == INDIRECT && newLink.type() == DIRECT) ||
                        !AnnotationsUtil.isEqual(existingLink.annotations(), newLink.annotations())) {
                    eventType.set(LINK_UPDATED);
                    return newLink;
            } else {
                return existingLink;
            }
        });
        return eventType.get() != null ? new LinkEvent(eventType.get(), link) : null;
    }

    private Set<ProviderId> getAllProviders(LinkKey linkKey) {
        return linkDescriptions.keySet()
                               .stream()
                               .filter(key -> key.key().equals(linkKey))
                               .map(key -> key.providerId())
                               .collect(Collectors.toSet());
    }

    private ProviderId getBaseProviderId(LinkKey linkKey) {
        Set<ProviderId> allProviders = getAllProviders(linkKey);
        if (allProviders.size() > 0) {
            return allProviders.stream()
                               .filter(p -> !p.isAncillary())
                               .findFirst()
                               .orElse(Iterables.getFirst(allProviders, null));
        }
        return null;
    }

    private Link composeLink(LinkKey linkKey) {

        ProviderId baseProviderId = checkNotNull(getBaseProviderId(linkKey));
        LinkDescription base = linkDescriptions.get(new Provided<>(linkKey, baseProviderId));

        ConnectPoint src = base.src();
        ConnectPoint dst = base.dst();
        Type type = base.type();
        AtomicReference<DefaultAnnotations> annotations = new AtomicReference<>(DefaultAnnotations.builder().build());
        annotations.set(merge(annotations.get(), base.annotations()));

        getAllProviders(linkKey).stream()
                                .map(p -> new Provided<>(linkKey, p))
                                .forEach(key -> {
                                    annotations.set(merge(annotations.get(),
                                                          linkDescriptions.get(key).annotations()));
        });

        boolean isDurable = Objects.equals(annotations.get().value(AnnotationKeys.DURABLE), "true");
        return new DefaultLink(baseProviderId, src, dst, type, ACTIVE, isDurable, annotations.get());
    }

    // Updates, if necessary the specified link and returns the appropriate event.
    // Guarded by linkDescs value (=locking each Link)
    private LinkEvent updateLink(LinkKey key, Link oldLink, Link newLink) {
        // Note: INDIRECT -> DIRECT transition only
        // so that BDDP discovered Link will not overwrite LDDP Link
        if (oldLink.state() != newLink.state() ||
            (oldLink.type() == INDIRECT && newLink.type() == DIRECT) ||
            !AnnotationsUtil.isEqual(oldLink.annotations(), newLink.annotations())) {

            links.put(key, newLink);
            return new LinkEvent(LINK_UPDATED, newLink);
        }
        return null;
    }

    @Override
    public LinkEvent removeOrDownLink(ConnectPoint src, ConnectPoint dst) {
        Link link = getLink(src, dst);
        if (link == null) {
            return null;
        }

        if (link.isDurable()) {
            // FIXME: this will not sync link state!!!
            return link.state() == INACTIVE ? null :
                    updateLink(linkKey(link.src(), link.dst()), link,
                               new DefaultLink(link.providerId(),
                                               link.src(), link.dst(),
                                               link.type(), INACTIVE,
                                               link.isDurable(),
                                               link.annotations()));
        }
        return removeLink(src, dst);
    }

    @Override
    public LinkEvent removeLink(ConnectPoint src, ConnectPoint dst) {
        final LinkKey linkKey = LinkKey.linkKey(src, dst);
        LinkDescription removedLinkDescription =
                linkDescriptions.remove(new Provided<>(linkKey, checkNotNull(getBaseProviderId(linkKey))));
        if (removedLinkDescription != null) {
            return purgeLinkCache(linkKey);
        }
        return null;
    }

    private LinkEvent purgeLinkCache(LinkKey linkKey) {
        Link removedLink = links.remove(linkKey);
        if (removedLink != null) {
            getAllProviders(linkKey).forEach(p -> linkDescriptions.remove(new Provided<>(linkKey, p)));
            return new LinkEvent(LINK_REMOVED, removedLink);
        }
        return null;
    }

    private Set<Link> filter(Collection<Link> links, Predicate<Link> predicate) {
        return links.stream().filter(predicate).collect(Collectors.toSet());
    }

    private LinkEvent injectLink(Provided<LinkDescription> linkInjectRequest) {
        log.trace("Received request to inject link {}", linkInjectRequest);

        ProviderId providerId = linkInjectRequest.providerId();
        LinkDescription linkDescription = linkInjectRequest.key();

        final DeviceId deviceId = linkDescription.dst().deviceId();
        if (!deviceClockService.isTimestampAvailable(deviceId)) {
            // workaround for ONOS-1208
            log.warn("Not ready to accept update. Dropping {}", linkInjectRequest);
            return null;
        }
        return createOrUpdateLink(providerId, linkDescription);
    }

    private class InternalLinkTracker implements EventuallyConsistentMapListener<Provided<LinkKey>, LinkDescription> {
        @Override
        public void event(EventuallyConsistentMapEvent<Provided<LinkKey>, LinkDescription> event) {
            if (event.type() == PUT) {
                notifyDelegate(refreshLinkCache(event.key().key()));
            } else if (event.type() == REMOVE) {
                notifyDelegate(purgeLinkCache(event.key().key()));
            }
        }
    }
}