random demo intent installer

Change-Id: I1ac2f5a6b7efbc7c940c6d7c371af4f5befcd676

using a timer to pace ourselves

Change-Id: Ia58698f950bbbc958ad002ed56dfe54b90f317ab

all good blasts intents

Change-Id: Ia85df5ad211c01d22d4088403d789b3d6a2292f7

clean up

Change-Id: I1a6dde05f57d0e4866d3255fc28836dfa7e7c190

@@ -109,7 +109,9 @@
                             com.google.common.*,
                             org.onlab.packet.*,
                             org.onlab.rest.*,
-                            org.onlab.onos.*
+                            org.onlab.onos.*,
+                            org.onlab.util.*,
+                            org.jboss.netty.util.*
                         </Import-Package>
                         <Web-ContextPath>${web.context}</Web-ContextPath>
                     </instructions>

 package org.onlab.onos.demo;
 
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.Optional;
+
 /**
  * Simple demo api interface.
  */
@@ -11,7 +15,7 @@ public interface DemoAPI {
      * Installs intents based on the installation type.
      * @param type the installation type.
      */
-    void setup(InstallType type);
+    void setup(InstallType type, Optional<JsonNode> runParams);
 
     /**
      * Uninstalls all existing intents.

@@ -15,7 +15,11 @@
  */
 package org.onlab.onos.demo;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.base.Predicate;
+import com.google.common.collect.FluentIterable;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
@@ -23,9 +27,14 @@ 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.onos.cluster.ClusterService;
 import org.onlab.onos.core.ApplicationId;
 import org.onlab.onos.core.CoreService;
+import org.onlab.onos.mastership.MastershipService;
 import org.onlab.onos.net.Host;
+import org.onlab.onos.net.HostId;
+import org.onlab.onos.net.MastershipRole;
 import org.onlab.onos.net.flow.DefaultTrafficSelector;
 import org.onlab.onos.net.flow.DefaultTrafficTreatment;
 import org.onlab.onos.net.flow.TrafficSelector;
@@ -34,15 +43,27 @@ import org.onlab.onos.net.host.HostService;
 import org.onlab.onos.net.intent.Constraint;
 import org.onlab.onos.net.intent.HostToHostIntent;
 import org.onlab.onos.net.intent.Intent;
+import org.onlab.onos.net.intent.IntentBatchService;
+import org.onlab.onos.net.intent.IntentOperations;
 import org.onlab.onos.net.intent.IntentService;
 import org.slf4j.Logger;
 
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
 
 import static org.slf4j.LoggerFactory.getLogger;
 
@@ -64,17 +85,31 @@ public class DemoInstaller implements DemoAPI {
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected HostService hostService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected MastershipService mastershipService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected IntentBatchService intentBatchService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ClusterService clusterService;
+
     private ExecutorService worker;
 
+    private ExecutorService randomWorker;
+
     private ApplicationId appId;
 
     private final Set<Intent> existingIntents = new HashSet<>();
+    private RandomInstaller randomInstaller;
 
 
 
     @Activate
     public void activate() {
-        appId = coreService.registerApplication("org.onlab.onos.demo.installer");
+        String nodeId = clusterService.getLocalNode().ip().toString();
+        appId = coreService.registerApplication("org.onlab.onos.demo.installer."
+                                                        + nodeId);
         worker = Executors.newFixedThreadPool(1,
                                               new ThreadFactoryBuilder()
                                                       .setNameFormat("demo-app-worker")
@@ -84,19 +119,34 @@ public class DemoInstaller implements DemoAPI {
 
     @Deactivate
     public void deactivate() {
-        worker.shutdownNow();
+        shutdownAndAwaitTermination(worker);
+        if (!randomWorker.isShutdown()) {
+            shutdownAndAwaitTermination(randomWorker);
+        }
         log.info("Stopped");
     }
 
     @Override
-    public void setup(InstallType type) {
+    public void setup(InstallType type, Optional<JsonNode> runParams) {
         switch (type) {
             case MESH:
                 log.debug("Installing mesh intents");
                 worker.execute(new MeshInstaller());
                 break;
             case RANDOM:
-                throw new IllegalArgumentException("Not yet implemented.");
+                //check that we do not have a random installer running
+                if (randomWorker == null || randomWorker.isShutdown()) {
+                    randomWorker = Executors.newFixedThreadPool(1,
+                                                   new ThreadFactoryBuilder()
+                                                           .setNameFormat("random-worker")
+                                                           .build());
+                    log.debug("Installing random sequence of intents");
+                    randomInstaller = new RandomInstaller(runParams);
+                    randomWorker.execute(randomInstaller);
+                } else {
+                    log.warn("Random installer is already running");
+                }
+                break;
             default:
                 throw new IllegalArgumentException("What is it you want exactly?");
         }
@@ -108,6 +158,9 @@ public class DemoInstaller implements DemoAPI {
     }
 
 
+    /**
+     * Simply installs a mesh of intents from all the hosts existing in the network.
+     */
     private class MeshInstaller implements Runnable {
 
         @Override
@@ -129,15 +182,295 @@ public class DemoInstaller implements DemoAPI {
         }
     }
 
+    /**
+     * Randomly installs and withdraws intents.
+     */
+    private class RandomInstaller implements Runnable {
+
+        private final boolean isLocal;
+        private final Set<Host> hosts;
+
+        private final Random random = new Random(System.currentTimeMillis());
+
+        private Set<HostPair> uninstalledOrWithdrawn;
+        private Set<HostPair> installed;
+
+        private CountDownLatch latch;
+
+        //used to wait on a batch to be processed.
+        private static final int ITERATIONMAX = 50000000;
+
+
+        public RandomInstaller(Optional<JsonNode> runParams) {
+            /*
+                Check if we have params and honour them. Otherwise
+                    set defaults to processing only local stuff and
+                    all local hosts.
+             */
+            if (runParams.isPresent()) {
+                JsonNode node = runParams.get();
+                isLocal = node.get("local").asBoolean();
+                hosts = node.get("hosts") == null ? Sets.newHashSet(hostService.getHosts()) :
+                        constructHostIds(node.get("hosts").elements());
+            } else {
+                isLocal = true;
+                hosts = Sets.newHashSet(hostService.getHosts());
+            }
+
+            //construct list of intents.
+            installed = Sets.newHashSet();
+            if (isLocal) {
+                uninstalledOrWithdrawn = buildPairs(pruneHostsByMasterShip());
+            } else {
+                uninstalledOrWithdrawn = buildPairs(hosts);
+            }
+
+        }
+
+        private Set<Host> constructHostIds(Iterator<JsonNode> elements) {
+            Set<Host> hostIds = Sets.newHashSet();
+            JsonNode n;
+            while (elements.hasNext()) {
+                n = elements.next();
+                hostIds.add(hostService.getHost(HostId.hostId(n.textValue())));
+            }
+            return hostIds;
+        }
+
+        @Override
+        public void run() {
+            if (!randomWorker.isShutdown()) {
+                randomize();
+                latch = new CountDownLatch(1);
+                try {
+                    trackIntents();
+                } catch (InterruptedException e) {
+                    shutdown();
+                }
+            }
+
+        }
+
+
+        /**
+         *   Check whether the previously submitted batch is in progress
+         *   and if yes submit the next one. If things hang, wait for at
+         *   most 5 seconds and bail.
+         * @throws InterruptedException if the thread go interupted
+         */
+        private void trackIntents() throws InterruptedException {
+            int count = 0;
+            while (!latch.await(100, TimeUnit.NANOSECONDS)) {
+                if (intentBatchService.getPendingOperations().isEmpty()) {
+                    latch.countDown();
+                }
+                count++;
+                if (count > ITERATIONMAX) {
+                    log.warn("A batch is stuck processing. current : {}" +
+                                     ", pending : {}",
+                             intentBatchService.getCurrentOperations(),
+                             intentBatchService.getPendingOperations());
+                    shutdownAndAwaitTermination(randomWorker);
+                }
+            }
+            //if everyting is good proceed.
+            if (!randomWorker.isShutdown()) {
+                randomWorker.execute(this);
+            }
+
+        }
+
+        public void shutdown() {
+            log.warn("Shutting down random installer!");
+            cleanUp();
+        }
+
+
+        /**
+         *   Shuffle the uninstalled and installed list (separately) and select
+         *   a random number of them and install or uninstall them respectively.
+         */
+        private void randomize() {
+            List<HostPair> hostList = new LinkedList<>(uninstalledOrWithdrawn);
+            Collections.shuffle(hostList);
+            List<HostPair> toInstall = hostList.subList(0,
+                                                        random.nextInt(hostList.size() - 1));
+            List<HostPair> toRemove;
+            if (!installed.isEmpty()) {
+                hostList = new LinkedList<>(installed);
+                Collections.shuffle(hostList);
+                toRemove = hostList.subList(0,
+                                            random.nextInt(hostList.size() - 1));
+                uninstallIntents(toRemove);
+            }
+            installIntents(toInstall);
+
+        }
+
+        private void installIntents(List<HostPair> toInstall) {
+            IntentOperations.Builder builder = IntentOperations.builder();
+            for (HostPair pair : toInstall) {
+                installed.add(pair);
+                uninstalledOrWithdrawn.remove(pair);
+                builder.addSubmitOperation(pair.h2hIntent());
+            }
+            intentBatchService.addIntentOperations(builder.build());
+        }
 
+        private void uninstallIntents(Collection<HostPair> toRemove) {
+            IntentOperations.Builder builder = IntentOperations.builder();
+            for (HostPair pair : toRemove) {
+                installed.remove(pair);
+                uninstalledOrWithdrawn.add(pair);
+                builder.addWithdrawOperation(pair.h2hIntent().id());
+            }
+            intentBatchService.addIntentOperations(builder.build());
+        }
+
+        /**
+         *  Take everything and remove it all.
+         */
+        private void cleanUp() {
+            List<HostPair> allPairs = Lists.newArrayList(installed);
+            allPairs.addAll(uninstalledOrWithdrawn);
+            IntentOperations.Builder builder = IntentOperations.builder();
+            for (HostPair pair : allPairs) {
+                builder.addWithdrawOperation(pair.h2hIntent().id());
+            }
+            intentBatchService.addIntentOperations(builder.build());
+        }
+
+
+        private Set<HostPair> buildPairs(Set<Host> hosts) {
+            Set<HostPair> pairs = Sets.newHashSet();
+            Iterator<Host> it = Sets.newHashSet(hosts).iterator();
+            while (it.hasNext()) {
+                Host src = it.next();
+                it.remove();
+                for (Host dst : hosts) {
+                    pairs.add(new HostPair(src, dst));
+                }
+            }
+            return pairs;
+        }
+
+        private Set<Host> pruneHostsByMasterShip() {
+            return FluentIterable.from(hosts)
+                    .filter(hasLocalMaster())
+                    .toSet();
+
+        }
+
+        private Predicate<? super Host> hasLocalMaster() {
+            return new Predicate<Host>() {
+                @Override
+                public boolean apply(Host host) {
+                    return mastershipService.getLocalRole(
+                            host.location().deviceId()).equals(MastershipRole.MASTER);
+                }
+            };
+        }
+
+
+        /**
+         * Simple class representing a pair of hosts and precomputes the associated
+         * h2h intent.
+         */
+        private class HostPair {
+
+            private final Host src;
+            private final Host dst;
+
+            private final TrafficSelector selector = DefaultTrafficSelector.builder().build();
+            private final TrafficTreatment treatment = DefaultTrafficTreatment.builder().build();
+            private final List<Constraint> constraint = Lists.newArrayList();
+            private final HostToHostIntent intent;
+
+            public HostPair(Host src, Host dst) {
+                this.src = src;
+                this.dst = dst;
+                this.intent = new HostToHostIntent(appId, src.id(), dst.id(),
+                                                   selector, treatment, constraint);
+            }
+
+            public HostToHostIntent h2hIntent() {
+                return intent;
+            }
+
+            @Override
+            public boolean equals(Object o) {
+                if (this == o) {
+                    return true;
+                }
+                if (o == null || getClass() != o.getClass()) {
+                    return false;
+                }
+
+                HostPair hostPair = (HostPair) o;
+
+                return Objects.equals(src, hostPair.src) &&
+                        Objects.equals(dst, hostPair.dst);
+
+            }
+
+            @Override
+            public int hashCode() {
+                return Objects.hash(src, dst);
+            }
+
+
+        }
+
+    }
+
+    /**
+     * Remove anything that is running and clear it all out.
+     */
     private class UnInstaller implements Runnable {
         @Override
         public void run() {
+            if (!existingIntents.isEmpty()) {
+                clearExistingIntents();
+            }
+
+            if (randomWorker != null && !randomWorker.isShutdown()) {
+                shutdownAndAwaitTermination(randomWorker);
+                randomInstaller.shutdown();
+            }
+        }
+
+        private void clearExistingIntents() {
             for (Intent i : existingIntents) {
                 intentService.withdraw(i);
             }
+            existingIntents.clear();
         }
     }
+
+    /**
+     * Shutdown a pool cleanly if possible.
+     *
+     * @param pool an executorService
+     */
+    private void shutdownAndAwaitTermination(ExecutorService pool) {
+        pool.shutdown(); // Disable new tasks from being submitted
+        try {
+            // Wait a while for existing tasks to terminate
+            if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
+                pool.shutdownNow(); // Cancel currently executing tasks
+                // Wait a while for tasks to respond to being cancelled
+                if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
+                    log.error("Pool did not terminate");
+                }
+            }
+        } catch (Exception ie) {
+            // (Re-)Cancel if current thread also interrupted
+            pool.shutdownNow();
+            // Preserve interrupt status
+            Thread.currentThread().interrupt();
+        }
+    }
+
 }
 
 

@@ -13,6 +13,7 @@ import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Optional;
 
 /**
  * Rest API for demos.
@@ -33,10 +34,11 @@ public class DemoResource extends BaseResource {
                     .entity("Expected type field containing either mesh or random.").build();
         }
 
+
         DemoAPI.InstallType type = DemoAPI.InstallType.valueOf(
                 cfg.get("type").asText().toUpperCase());
         DemoAPI demo = get(DemoAPI.class);
-        demo.setup(type);
+        demo.setup(type, Optional.ofNullable(cfg.get("runParams")));
 
         return Response.ok(mapper.createObjectNode().toString()).build();
     }

@@ -231,6 +231,7 @@
     <feature name="onos-app-demo" version="1.0.0"
              description="ONOS demo applications">
         <feature>onos-api</feature>
+        <bundle>mvn:org.onlab.onos/onlab-misc/1.0.0-SNAPSHOT</bundle>
         <bundle>mvn:org.onlab.onos/onos-app-demo/1.0.0-SNAPSHOT</bundle>
     </feature>