DiscoveryManager.java

Go to the documentation of this file.

/* Copyright 2018 Telstra Open Source
 *
 *   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.openkilda.wfm.isl;

import org.openkilda.messaging.model.DiscoveryLink;
import org.openkilda.messaging.model.NetworkEndpoint;
import org.openkilda.messaging.model.SwitchId;

import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.map.PassiveExpiringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

public class DiscoveryManager {
    private final Logger logger = LoggerFactory.getLogger(DiscoveryManager.class);

    private final int islHealthCheckInterval;
    private final int islConsecutiveFailureLimit;
    private final int maxAttempts;

    private final Map<SwitchId, Set<DiscoveryLink>> linksBySwitch;

    private final Map<NetworkEndpoint, DiscoveryLink> removedFromDiscovery;

    public DiscoveryManager(Map<SwitchId, Set<DiscoveryLink>> linksBySwitch,
                            int islHealthCheckInterval, int islConsecutiveFailureLimit,
                            int maxAttempts, Integer minutesKeepRemovedIsl) {
        this.islHealthCheckInterval = islHealthCheckInterval;
        this.islConsecutiveFailureLimit = islConsecutiveFailureLimit;
        this.maxAttempts = maxAttempts;
        this.linksBySwitch = linksBySwitch;
        this.removedFromDiscovery = new PassiveExpiringMap<>(minutesKeepRemovedIsl, TimeUnit.MINUTES, new HashMap<>());
    }

    public Plan makeDiscoveryPlan() {
        Plan result = new Plan();
        int unsentDiscoPackets = 0;

        List<DiscoveryLink> links = linksBySwitch.values()
                .stream()
                .flatMap(Set::stream)
                .collect(Collectors.toList());

        for (DiscoveryLink link : links) {
            if (!link.isNewAttemptAllowed()) {
                logger.trace("Disco packet from {} is not sent due to exceeded limit of consecutive failures: {}",
                        link.getSource(), link.getConsecutiveFailure());
                continue;
            }

            /*
             * If we get a response from FL, we clear the attempts. Otherwise, no response, and
             * number of attempts grows.
             *
             * Further, consecutivefailures = attempts - failure limit (we wait until attempt limit before increasing)
             */
            NetworkEndpoint node = link.getSource();
            if (link.isAckAttemptsLimitExceeded(islConsecutiveFailureLimit)) {
                // We've attempted to get the health multiple times, with no response.
                // Time to mark it as a failure and send a failure notice ** if ** it was an ISL.
                if (link.isActive() && link.getConsecutiveFailure() == 0) {
                    // It is a discovery failure if it was previously a success.
                    result.discoveryFailure.add(node);
                    logger.info("ISL IS DOWN (NO RESPONSE): {}", link);
                }
                // Increment Failure = 1 after isAttemptsLimitExceeded failure, then increases every attempt.
                logger.trace("No response to the disco packet from {}", link.getSource());
                link.fail();
                // NB: this node can be in both discoveryFailure and needDiscovery
            }

            if (link.isAttemptsLimitExceeded(islConsecutiveFailureLimit) && link.isActive()) {
                logger.info("Speaker doesn't send disco packet for {}", link);
                unsentDiscoPackets++;
            }

            /*
             * If you get here, the following are true:
             *  - it isn't in some filter
             *  - it hasn't reached failure limit (forlorn)
             *  - it is either time to send discovery or not
             *  - NB: we'll keep trying to send discovery, even if we don't get a response.
             */
            if (link.timeToCheck()) {
                link.incAttempts();
                link.resetTickCounter();
                result.needDiscovery.add(node);

                logger.trace("Added to discovery plan: {}", link);
            } else {
                link.tick();
            }

        }

        if (unsentDiscoPackets > 0) {
            logger.warn("Speaker does not send discovery packets. Affected links amount: {}", unsentDiscoPackets);
        }

        return result;
    }

    public boolean handleDiscovered(SwitchId srcSwitch, int srcPort, SwitchId dstSwitch, int dstPort) {
        boolean stateChanged = false;
        NetworkEndpoint node = new NetworkEndpoint(srcSwitch, srcPort);
        Optional<DiscoveryLink> matchedLink = findBySourceEndpoint(node);

        if (!matchedLink.isPresent()) {
            logger.warn("Ignore \"AVAIL\" request for {}: node not found", node);
        } else {
            DiscoveryLink link = matchedLink.get();
            if (!link.isActive() || link.isDestinationChanged(dstSwitch, dstPort)) {
                // we've found newly discovered or moved/replugged isl
                link.activate(new NetworkEndpoint(dstSwitch, dstPort));

                stateChanged = true;
                logger.info("FOUND ISL: {}", link);
            } else if (link.getConsecutiveFailure() > 0) {
                // We've found failures, but now we've had success, so that is a state change.
                // To repeat, current model for state change is just 1 failure. If we change this
                // policy, then change the test above.
                stateChanged = true;
                logger.info("ISL IS UP: {}", link);
            }
            link.renew();
            link.success();
            link.clearConsecutiveFailure();
            // If one of the logs above wasn't reachd, don't log anything .. ISL was up and is still up
        }

        if (stateChanged) {
            // Add logic to ensure we send a discovery packet for the opposite direction.
            // TODO: in order to do this here, we need more information (ie the other end of the ISL)
            //      Since that isn't passed in and isn't available in our state, have to rely on the
            //      calling function.

        }
        return stateChanged;
    }

    public boolean handleFailed(SwitchId switchId, int portId) {
        boolean stateChanged = false;
        NetworkEndpoint endpoint = new NetworkEndpoint(switchId, portId);
        Optional<DiscoveryLink> matchedLink = findBySourceEndpoint(endpoint);

        if (!matchedLink.isPresent()) {
            logger.warn("Ignoring \"FAILED\" request. There is no link found from {}", endpoint);
        } else {
            DiscoveryLink link = matchedLink.get();
            if (link.isActive() && link.getConsecutiveFailure() == 0) {
                // This is the first failure for an ISL. That is a state change.
                // IF this isn't an ISL and we receive a failure, that isn't a state change.
                stateChanged = true;
                logger.info("ISL IS DOWN (GOT RESPONSE): {}", link);
            }
            link.renew();
            link.fail();
            link.deactivate();
        }
        return stateChanged;
    }

    public void handleSentDiscoPacket(NetworkEndpoint endpoint) {
        findBySourceEndpoint(endpoint)
                .ifPresent(DiscoveryLink::incAcknowledgedAttempts);

        logger.debug("Received acknowledge of sending disco from {}", endpoint);
    }

    public void handleSwitchUp(SwitchId switchId) {
        logger.info("Register switch {} into ISL discovery manager", switchId);
        // TODO: this method *use to not* do anything .. but it should register the switch.
        //          At least, it seems like it should do something to register a switch, even
        //          though this can be lazily done when the first port event arrives.

        /*
         * If a switch comes up, clear any "isFoundIsl" flags, in case something has changed,
         * and/or if the TE has cleared it's state .. this will pass along the ISL.
         */
        Set<DiscoveryLink> subjectList = findAllBySwitch(switchId);

        if (!subjectList.isEmpty()) {
            logger.info("Received SWITCH UP (id:{}) with EXISTING NODES.  Clearing isFoundISL flags", switchId);
            for (DiscoveryLink subject : subjectList) {
                subject.deactivate();
            }
        }
    }

    public void handleSwitchDown(SwitchId switchId) {
        logger.info("Deregister switch {} from ISL discovery manager", switchId);
        removeFromDiscovery(new NetworkEndpoint(switchId, 0));
    }

    public void handlePortUp(SwitchId switchId, int portId) {
        DiscoveryLink link = registerPort(switchId, portId);
        if (link.isActive() || !link.isNewAttemptAllowed()) {
            // Similar to SwitchUp, if we have a PortUp on an existing port, either we are receiving
            // a duplicate, or we missed the port down, or a new discovery has occurred.
            // NB: this should cause an ISL discovery packet to be sent.
            // TODO: we should probably separate "port up" from "do discovery". ATM, one would call
            //          this function just to get the "do discovery" functionality.
            logger.info("Port UP on existing node {};  clear failures and ISLFound", link);
            link.deactivate();
        } else {
            logger.info("Port UP on new node: {}", link);
        }
    }

    public DiscoveryLink registerPort(SwitchId switchId, int portId) {
        NetworkEndpoint node = new NetworkEndpoint(switchId, portId);
        return findBySourceEndpoint(node)
                .orElseGet(() -> registerDiscoveryLink(node));
    }

    private DiscoveryLink registerDiscoveryLink(NetworkEndpoint node) {
        DiscoveryLink link = new DiscoveryLink(node.getDatapath(), node.getPortNumber(),
                this.islHealthCheckInterval, this.maxAttempts);
        linksBySwitch.computeIfAbsent(node.getDatapath(), key -> new HashSet<>())
                .add(link);

        logger.info("The link has been registered for discovery: {}", link);

        return link;
    }

    public void handlePortDown(SwitchId switchId, int portId) {
        NetworkEndpoint node = new NetworkEndpoint(switchId, portId);
        Optional<DiscoveryLink> discoveryLink = findBySourceEndpoint(node);

        if (!discoveryLink.isPresent()) {
            logger.warn("Can't update discovery {} -> node not found", node);
            return;
        }

        removeFromDiscovery(node);
    }

    @VisibleForTesting
    Set<DiscoveryLink> findAllBySwitch(SwitchId dpid) {
        return linksBySwitch.getOrDefault(dpid, Collections.emptySet());
    }

    @VisibleForTesting
    Optional<DiscoveryLink> findBySourceEndpoint(NetworkEndpoint endpoint) {
        Set<DiscoveryLink> links = linksBySwitch.getOrDefault(endpoint.getDatapath(), Collections.emptySet());

        return links.stream()
                .filter(link -> endpoint.equals(link.getSource()))
                .findFirst();
    }

    private void removeFromDiscovery(NetworkEndpoint endpoint) {
        if (endpoint.getPortNumber() == 0) {
            Set<DiscoveryLink> links = linksBySwitch.remove(endpoint.getDatapath());
            if (!CollectionUtils.isEmpty(links)) {
                Map<NetworkEndpoint, DiscoveryLink> removedLinks = links.stream()
                        .collect(Collectors.toMap(DiscoveryLink::getSource, Function.identity()));
                removedFromDiscovery.putAll(removedLinks);

                logger.info("Removed switch {} from discovery", endpoint.getDatapath());
            }
        } else {
            Set<DiscoveryLink> links = linksBySwitch.get(endpoint.getDatapath());

            Optional<DiscoveryLink> matchedLink = links.stream()
                    .filter(discoveryLink -> endpoint.equals(discoveryLink.getSource()))
                    .findFirst();

            matchedLink.ifPresent(link -> {
                links.remove(link);
                removedFromDiscovery.put(endpoint, link);

                logger.info("The link has been removed from discovery: {}", link);
            });
        }
    }

    public boolean isIslMoved(SwitchId srcSwitch, int srcPort, SwitchId dstSwitch, int dstPort) {
        NetworkEndpoint endpoint = new NetworkEndpoint(srcSwitch, srcPort);
        DiscoveryLink link = findBySourceEndpoint(endpoint)
                .orElseGet(() -> removedFromDiscovery.get(endpoint));

        if (link != null && link.isDestinationChanged(dstSwitch, dstPort)) {
            logger.info("ISL Event: the link has been moved: {} to {}_{}", link, dstSwitch, dstPort);
            return true;
        }

        return false;
    }

    public NetworkEndpoint getLinkDestination(SwitchId srcSwitch, int srcPort) {
        NetworkEndpoint srcEndpoint = new NetworkEndpoint(srcSwitch, srcPort);
        DiscoveryLink link = findBySourceEndpoint(srcEndpoint)
                .orElseGet(() -> removedFromDiscovery.get(srcEndpoint));

        if (link != null && link.getDestination() != null) {
            return link.getDestination();
        }

        throw new IllegalStateException(String.format("Not found link from %s_%s", srcSwitch, srcPort));
    }

    public void deactivateLinkFromEndpoint(NetworkEndpoint endpoint) {
        DiscoveryLink link = findBySourceEndpoint(endpoint)
                .orElseGet(() -> removedFromDiscovery.remove(endpoint));
        if (link != null) {
            link.deactivate();

            logger.info("The link has been deactivated: {}", link);
        }
    }

    public boolean isInDiscoveryPlan(SwitchId switchId, int portId) {
        Optional<DiscoveryLink> link = findBySourceEndpoint(new NetworkEndpoint(switchId, portId));

        return link.isPresent() && link.get().isNewAttemptAllowed();
    }

    public final class Plan {
        public final List<NetworkEndpoint> needDiscovery;
        public final List<NetworkEndpoint> discoveryFailure;

        private Plan() {
            this.needDiscovery = new LinkedList<>();
            this.discoveryFailure = new LinkedList<>();
        }
    }
}