apps/vpls/src/main/java/org/onosproject/vpls/VplsOperationManager.java

/*
 * Copyright 2017-present Open Networking Foundation
 *
 * 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.vpls;

import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.google.common.collect.Sets;
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.Tools;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.LeadershipEvent;
import org.onosproject.cluster.LeadershipEventListener;
import org.onosproject.cluster.LeadershipService;
import org.onosproject.cluster.NodeId;
import org.onosproject.core.ApplicationId;
import org.onosproject.core.CoreService;
import org.onosproject.incubator.net.intf.Interface;
import org.onosproject.net.Host;
import org.onosproject.net.host.HostService;
import org.onosproject.net.intent.Intent;
import org.onosproject.net.intent.IntentEvent;
import org.onosproject.net.intent.IntentException;
import org.onosproject.net.intent.IntentListener;
import org.onosproject.net.intent.IntentService;
import org.onosproject.net.intent.IntentUtils;
import org.onosproject.net.intent.Key;
import org.onosproject.net.intent.MultiPointToSinglePointIntent;
import org.onosproject.net.intent.SinglePointToMultiPointIntent;
import org.onosproject.vpls.api.VplsData;
import org.onosproject.vpls.api.VplsOperationException;
import org.onosproject.vpls.api.VplsOperationService;
import org.onosproject.vpls.api.VplsOperation;
import org.onosproject.vpls.api.VplsStore;
import org.onosproject.vpls.intent.VplsIntentUtility;
import org.slf4j.Logger;

import java.util.Collection;
import java.util.Deque;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.onlab.util.BoundedThreadPool.newFixedThreadPool;
import static org.onlab.util.Tools.groupedThreads;
import static org.slf4j.LoggerFactory.getLogger;

/**
 * An implementation of VplsOperationService.
 * Handles the execution order of the VPLS operations generated by the
 * application.
 */
@Component(immediate = true)
@Service
public class VplsOperationManager implements VplsOperationService {
    private static final int NUM_THREADS = 4;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected CoreService coreService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected IntentService intentService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected LeadershipService leadershipService;

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

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected HostService hostService;

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected VplsStore vplsStore;

    private final Logger log = getLogger(getClass());
    protected Map<String, Deque<VplsOperation>> pendingVplsOperations;
    protected final Map<String, VplsOperation> runningOperations = Maps.newHashMap();
    protected ScheduledExecutorService schedulerExecutor;
    protected ExecutorService workerExecutor;
    protected ApplicationId appId;
    protected boolean isLeader;
    protected NodeId localNodeId;
    protected LeadershipEventListener leadershipEventListener;

    @Activate
    public void activate() {
        appId = coreService.registerApplication(VplsManager.VPLS_APP);
        localNodeId = clusterService.getLocalNode().id();

        leadershipEventListener = new InternalLeadershipListener();
        leadershipService.addListener(leadershipEventListener);
        leadershipService.runForLeadership(appId.name());
        pendingVplsOperations = Maps.newConcurrentMap();

        // Thread pool for VplsOperationExecutor
        workerExecutor = newFixedThreadPool(NUM_THREADS,
                                            groupedThreads("onos/apps/vpls",
                                                           "worker-%d",
                                                           log));
        // A single thread pool for VplsOperationScheduler
        schedulerExecutor = Executors.newScheduledThreadPool(1,
                                                             groupedThreads("onos/apps/vpls",
                                                                            "scheduler-%d",
                                                                            log));
        // Start the scheduler
        schedulerExecutor.scheduleAtFixedRate(new VplsOperationScheduler(),
                                              0,
                                              500,
                                              TimeUnit.MILLISECONDS);

    }

    @Deactivate
    public void deactivate() {
        pendingVplsOperations.clear();
        runningOperations.clear();
        leadershipService.removeListener(leadershipEventListener);
        schedulerExecutor.shutdown();
        workerExecutor.shutdown();

        // remove all intents from VPLS application when deactivated
        Tools.stream(intentService.getIntents())
                .filter(intent -> intent.appId().equals(appId))
                .forEach(intentService::withdraw);
    }

    @Override
    public void submit(VplsOperation vplsOperation) {
        if (isLeader) {
            // Only leader can execute operation
            addVplsOperation(vplsOperation);
        }
    }

    /**
     * Adds a VPLS operation to the queue of pending operations.
     *
     * @param vplsOperation the VPLS operation to add
     */
    private void addVplsOperation(VplsOperation vplsOperation) {
        VplsData vplsData = vplsOperation.vpls();
        pendingVplsOperations.compute(vplsData.name(), (name, opQueue) -> {
            opQueue = opQueue == null ? Queues.newArrayDeque() : opQueue;

            // If the operation already exist in queue, ignore it.
            if (opQueue.contains(vplsOperation)) {
                return opQueue;
            }
            opQueue.add(vplsOperation);
            return opQueue;
        });
    }

    /**
     * Optimizes the VPLS operation queue and return a single VPLS operation to
     * execute.
     *
     * @param operations the queue to be optimized
     * @return optimized VPLS operation from the queue
     */
    protected static VplsOperation getOptimizedVplsOperation(Deque<VplsOperation> operations) {
        if (operations.isEmpty()) {
            return null;
        }
        // no need to optimize if the queue contains only one operation
        if (operations.size() == 1) {
            return operations.getFirst();
        }

        final VplsOperation firstOperation = operations.peekFirst();
        final VplsOperation lastOperation = operations.peekLast();
        final VplsOperation.Operation firstOp = firstOperation.op();
        final VplsOperation.Operation lastOp = lastOperation.op();

        if (firstOp.equals(VplsOperation.Operation.REMOVE)) {
            if (lastOp.equals(VplsOperation.Operation.REMOVE)) {
                // case 1: both first and last operation are REMOVE; do remove
                return firstOperation;
            } else if (lastOp.equals(VplsOperation.Operation.ADD)) {
                // case 2: if first is REMOVE, and last is ADD; do update
                return VplsOperation.of(lastOperation.vpls(),
                                                 VplsOperation.Operation.UPDATE);
            } else {
                // case 3: first is REMOVE, last is UPDATE; do update
                return lastOperation;
            }
        } else if (firstOp.equals(VplsOperation.Operation.ADD)) {
            if (lastOp.equals(VplsOperation.Operation.REMOVE)) {
                // case 4: first is ADD, last is REMOVE; nothing to do
                return null;
            } else if (lastOp.equals(VplsOperation.Operation.ADD)) {
                // case 5: both first and last are ADD, do add
                return VplsOperation.of(lastOperation.vpls(),
                                                 VplsOperation.Operation.ADD);
            } else {
                // case 6: first is ADD and last is update, do add
                return VplsOperation.of(lastOperation.vpls(),
                                                 VplsOperation.Operation.ADD);
            }
        } else {
            if (lastOp.equals(VplsOperation.Operation.REMOVE)) {
                // case 7: last is remove, do remove
                return lastOperation;
            } else if (lastOp.equals(VplsOperation.Operation.ADD)) {
                // case 8: do update only
                return VplsOperation.of(lastOperation.vpls(),
                                                 VplsOperation.Operation.UPDATE);
            } else {
                // case 9: from UPDATE to UPDATE
                // only need last UPDATE operation
                return VplsOperation.of(lastOperation.vpls(),
                                                 VplsOperation.Operation.UPDATE);
            }
        }
    }

    /**
     * Scheduler for VPLS operation.
     * Processes a batch of VPLS operations in a period.
     */
    class VplsOperationScheduler implements Runnable {
        private static final String UNKNOWN_STATE =
                "Unknown state {} for success consumer";
        private static final String OP_EXEC_ERR =
                "Error when executing VPLS operation {}, error: {}";

        /**
         * Process a batch of VPLS operations.
         */
        @Override
        public void run() {
            Set<String> vplsNames = pendingVplsOperations.keySet();
            vplsNames.forEach(vplsName -> {
                VplsOperation operation;
                synchronized (runningOperations) {
                    // Only one operation for a VPLS at the same time
                    if (runningOperations.containsKey(vplsName)) {
                        return;
                    }
                    Deque<VplsOperation> operations = pendingVplsOperations.remove(vplsName);
                    operation = getOptimizedVplsOperation(operations);
                    if (operation == null) {
                        // Nothing to do, this only happened when we add a VPLS
                        // and remove it before batch operations been processed.
                        return;
                    }
                    runningOperations.put(vplsName, operation);
                }

                VplsOperationExecutor operationExecutor =
                        new VplsOperationExecutor(operation);
                operationExecutor.setConsumers(
                        (vplsOperation) -> {
                            // Success consumer
                            VplsData vplsData = vplsOperation.vpls();
                            log.debug("VPLS operation success: {}", vplsOperation);
                            switch (vplsData.state()) {
                                case ADDING:
                                case UPDATING:
                                    vplsData.state(VplsData.VplsState.ADDED);
                                    vplsStore.updateVpls(vplsData);
                                    break;
                                case REMOVING:
                                    // The VPLS information does not exists in
                                    // store. No need to update the store.
                                    break;
                                default:
                                    log.warn(UNKNOWN_STATE, vplsData.state());
                                    vplsData.state(VplsData.VplsState.FAILED);
                                    vplsStore.updateVpls(vplsData);
                                    break;
                            }
                            runningOperations.remove(vplsName);
                        },
                        (vplsOperationException) -> {
                            // Error consumer
                            VplsOperation vplsOperation =
                                    vplsOperationException.vplsOperation();
                            log.warn(OP_EXEC_ERR,
                                     vplsOperation.toString(),
                                     vplsOperationException.getMessage());
                            VplsData vplsData = vplsOperation.vpls();
                            vplsData.state(VplsData.VplsState.FAILED);
                            vplsStore.updateVpls(vplsData);
                            runningOperations.remove(vplsName);
                        });
                log.debug("Applying operation: {}", operation);
                workerExecutor.execute(operationExecutor);
            });
        }
    }

    /**
     * Direction for Intent installation.
     */
    private enum Direction {
        ADD,
        REMOVE
    }

    /**
     * VPLS operation executor.
     * Installs, updates or removes Intents according to the given VPLS operation.
     */
    class VplsOperationExecutor implements Runnable {
        private static final String UNKNOWN_OP = "Unknown operation.";
        private static final String UNKNOWN_INTENT_DIR = "Unknown Intent install direction.";
        private static final int OPERATION_TIMEOUT = 10;
        private VplsOperation vplsOperation;
        private Consumer<VplsOperation> successConsumer;
        private Consumer<VplsOperationException> errorConsumer;
        private VplsOperationException error;

        public VplsOperationExecutor(VplsOperation vplsOperation) {
            this.vplsOperation = vplsOperation;
            this.error = null;
        }

        /**
         * Sets success consumer and error consumer for this executor.
         *
         * @param successConsumer the success consumer
         * @param errorConsumer the error consumer
         */
        public void setConsumers(Consumer<VplsOperation> successConsumer,
                                 Consumer<VplsOperationException> errorConsumer) {
            this.successConsumer = successConsumer;
            this.errorConsumer = errorConsumer;

        }

        @Override
        public void run() {
            switch (vplsOperation.op()) {
                case ADD:
                    installVplsIntents();
                    break;
                case REMOVE:
                    removeVplsIntents();
                    break;
                case UPDATE:
                    updateVplsIntents();
                    break;
                default:
                    this.error = new VplsOperationException(vplsOperation,
                                                            UNKNOWN_OP);
                    break;
            }

            if (this.error != null) {
                errorConsumer.accept(this.error);
            } else {
                successConsumer.accept(vplsOperation);
            }
        }

        /**
         * Updates Intents of the VPLS.
         */
        private void updateVplsIntents() {
            // check which part we need to update
            // if we update host only, we don't need to reinstall
            // every Intents
            Set<Intent> intentsToInstall = Sets.newHashSet();
            Set<Intent> intentsToUninstall = Sets.newHashSet();
            VplsData vplsData = vplsOperation.vpls();
            Set<Intent> currentIntents = getCurrentIntents();

            // Compares broadcast Intents
            Set<Intent> currentBrcIntents = currentIntents.stream()
                    .filter(intent -> intent instanceof SinglePointToMultiPointIntent)
                    .collect(Collectors.toSet());
            Set<Intent> targetBrcIntents = VplsIntentUtility.buildBrcIntents(vplsData, appId);
            if (!intentSetEquals(currentBrcIntents, targetBrcIntents)) {
                // If broadcast Intents changes, it means some network
                // interfaces or encapsulation constraint changed; Need to
                // reinstall all intents
                removeVplsIntents();
                installVplsIntents();
                return;
            }

            // Compares unicast Intents
            Set<Intent> currentUniIntents = currentIntents.stream()
                    .filter(intent -> intent instanceof MultiPointToSinglePointIntent)
                    .collect(Collectors.toSet());
            Set<Intent> targetUniIntents = VplsIntentUtility.buildUniIntents(vplsData,
                                                                             hostsFromVpls(),
                                                                             appId);

            // New unicast Intents to install
            targetUniIntents.forEach(intent -> {
                if (!currentUniIntents.contains(intent)) {
                    intentsToInstall.add(intent);
                }
            });

            // Old unicast Intents to remove
            currentUniIntents.forEach(intent -> {
                if (!targetUniIntents.contains(intent)) {
                    intentsToUninstall.add(intent);
                }
            });
            applyIntentsSync(intentsToUninstall, Direction.REMOVE);
            applyIntentsSync(intentsToInstall, Direction.ADD);
        }

        private Set<Host> hostsFromVpls() {
            VplsData vplsData = vplsOperation.vpls();
            Set<Interface> interfaces = vplsData.interfaces();
            return interfaces.stream()
                    .map(this::hostsFromInterface)
                    .flatMap(Collection::stream)
                    .collect(Collectors.toSet());
        }

        private Set<Host> hostsFromInterface(Interface iface) {
            return hostService.getConnectedHosts(iface.connectPoint())
                    .stream()
                    .filter(host -> host.vlan().equals(iface.vlan()))
                    .collect(Collectors.toSet());
        }

        /**
         * Applies Intents synchronously with a specific direction.
         *
         * @param intents the Intents
         * @param direction the direction
         */
        private void applyIntentsSync(Set<Intent> intents, Direction direction) {
            Set<Key> pendingIntentKeys = intents.stream()
                    .map(Intent::key).collect(Collectors.toSet());
            IntentCompleter completer;

            switch (direction) {
                case ADD:
                    completer = new IntentCompleter(pendingIntentKeys,
                                                    IntentEvent.Type.INSTALLED);
                    intentService.addListener(completer);
                    intents.forEach(intentService::submit);
                    break;
                case REMOVE:
                    completer = new IntentCompleter(pendingIntentKeys,
                                                    IntentEvent.Type.WITHDRAWN);
                    intentService.addListener(completer);
                    intents.forEach(intentService::withdraw);
                    break;
                default:
                    this.error = new VplsOperationException(this.vplsOperation,
                                                            UNKNOWN_INTENT_DIR);
                    return;
            }

            try {
                // Wait until Intent operation completed
                completer.complete();
            } catch (VplsOperationException e) {
                this.error = e;
            } finally {
                intentService.removeListener(completer);
            }
        }

        /**
         * Checks if two sets of Intents are equal.
         *
         * @param intentSet1 the first set of Intents
         * @param intentSet2 the second set of Intents
         * @return true if both set of Intents are equal; otherwise false
         */
        private boolean intentSetEquals(Set<Intent> intentSet1, Set<Intent> intentSet2) {
            if (intentSet1.size() != intentSet2.size()) {
                return false;
            }
            for (Intent intent1 : intentSet1) {
                if (intentSet2.stream()
                        .noneMatch(intent2 -> IntentUtils.intentsAreEqual(intent1, intent2))) {
                    return false;
                }
            }
            return true;
        }

        /**
         * Retrieves installed Intents from IntentService which related to
         * specific VPLS.
         *
         * @return the Intents which related to the VPLS
         */
        private Set<Intent> getCurrentIntents() {
            VplsData vplsData = vplsOperation.vpls();
            String vplsName = vplsData.name();
            return Tools.stream(intentService.getIntents())
                    .filter(intent -> intent.key().toString().startsWith(vplsName))
                    .collect(Collectors.toSet());
        }

        /**
         * Generates unicast Intents and broadcast Intents for the VPLS.
         *
         * @return Intents for the VPLS
         */
        private Set<Intent> generateVplsIntents() {
            VplsData vplsData = vplsOperation.vpls();
            Set<Intent> brcIntents = VplsIntentUtility.buildBrcIntents(vplsData, appId);
            Set<Intent> uniIntent = VplsIntentUtility.buildUniIntents(vplsData, hostsFromVpls(), appId);

            return Stream.concat(brcIntents.stream(), uniIntent.stream())
                    .collect(Collectors.toSet());
        }

        /**
         * Removes all Intents from the VPLS.
         */
        private void removeVplsIntents() {
            Set<Intent> intentsToWithdraw = getCurrentIntents();
            applyIntentsSync(intentsToWithdraw, Direction.REMOVE);
            intentsToWithdraw.forEach(intentService::purge);
        }

        /**
         * Installs Intents of the VPLS.
         */
        private void installVplsIntents() {
            Set<Intent> intentsToInstall = generateVplsIntents();
            applyIntentsSync(intentsToInstall, Direction.ADD);
        }

        /**
         * Helper class which monitors if all Intent operations are completed.
         */
        class IntentCompleter implements IntentListener {
            private static final String INTENT_COMPILE_ERR = "Got {} from intent completer";
            private CompletableFuture<Void> completableFuture;
            private Set<Key> pendingIntentKeys;
            private IntentEvent.Type expectedEventType;

            /**
             * Initialize completer with given Intent keys and expect Intent
             * event type.
             *
             * @param pendingIntentKeys the Intent keys to wait
             * @param expectedEventType expect Intent event type
             */
            public IntentCompleter(Set<Key> pendingIntentKeys,
                                   IntentEvent.Type expectedEventType) {
                this.completableFuture = new CompletableFuture<>();
                this.pendingIntentKeys = Sets.newConcurrentHashSet(pendingIntentKeys);
                this.expectedEventType = expectedEventType;
            }

            @Override
            public void event(IntentEvent event) {
                Intent intent = event.subject();
                Key key = intent.key();
                if (!pendingIntentKeys.contains(key)) {
                    // ignore Intent events from other VPLS
                    return;
                }
                // Intent failed, throw an exception to completable future
                if (event.type() == IntentEvent.Type.CORRUPT ||
                        event.type() == IntentEvent.Type.FAILED) {
                    completableFuture.completeExceptionally(new IntentException(intent.toString()));
                    return;
                }
                // If event type matched to expected type, remove from pending
                if (event.type() == expectedEventType) {
                    pendingIntentKeys.remove(key);
                }
                if (pendingIntentKeys.isEmpty()) {
                    completableFuture.complete(null);
                }
            }

            /**
             * Waits until all pending Intents completed ot timeout.
             */
            public void complete() {
                // If no pending Intent keys, complete directly
                if (pendingIntentKeys.isEmpty()) {
                    return;
                }
                try {
                    completableFuture.get(OPERATION_TIMEOUT, TimeUnit.SECONDS);
                } catch (TimeoutException | InterruptedException |
                         ExecutionException | IntentException e) {
                    // TODO: handle errors more carefully
                    log.warn(INTENT_COMPILE_ERR, e.toString());
                    throw new VplsOperationException(vplsOperation, e.toString());
                }
            }
        }
    }

    /**
     * A listener for leadership events.
     * Only the leader can process VPLS operation in the ONOS cluster.
     */
    private class InternalLeadershipListener implements LeadershipEventListener {
        private static final String LEADER_CHANGE = "Change leader to {}";

        @Override
        public void event(LeadershipEvent event) {
            switch (event.type()) {
                case LEADER_CHANGED:
                case LEADER_AND_CANDIDATES_CHANGED:
                    isLeader = localNodeId.equals(event.subject().leaderNodeId());
                    if (isLeader) {
                        log.debug(LEADER_CHANGE, localNodeId);
                    }
                    break;
                default:
                    break;
            }
        }

        @Override
        public boolean isRelevant(LeadershipEvent event) {
            return event.subject().topic().equals(appId.name());
        }
    }
}