CDS: Add stress test RPC to the cars model

For stress testing the CDS, I've been using an RPC that continuously
creates cars at a specified per second rate. I thought it might be
useful to submit it.

Change-Id: I33b9c2e304884b9541774a12ee248082de60f72e
Signed-off-by: Tom Pantelis <tpanteli@brocade.com>
(cherry picked from commit b6a63eea6eb164a0c925716ae398bf64b06958c3)

diff --git a/opendaylight/md-sal/samples/clustering-test-app/model/src/main/yang/car.yang b/opendaylight/md-sal/samples/clustering-test-app/model/src/main/yang/car.yang
index d9cfb6b1d5872f79a695410cf52d100fd6094e70..f421133c620c20bbeebe1358ddeacd3ba955c819 100644 (file)
--- a/opendaylight/md-sal/samples/clustering-test-app/model/src/main/yang/car.yang
+++ b/opendaylight/md-sal/samples/clustering-test-app/model/src/main/yang/car.yang
@@ -59,6 +59,15 @@ module car {
        uses car-entry;
       }
     }
-
     
+    rpc stress-test {
+        input {
+            leaf rate {
+              type uint16;
+            }
+        }
+    }
+    
+    rpc stop-stress-test {
+    }
 }
\ No newline at end of file

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/CarProvider.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/CarProvider.java
new file mode 100644 (file)
index 0000000..2a068af
--- /dev/null
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/CarProvider.java
@@ -0,0 +1,116 @@
+/*
+ * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  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,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.clustering.it.provider;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.Futures;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.opendaylight.controller.md.sal.binding.api.DataBroker;
+import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.CarId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.CarService;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.Cars;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.CarsBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.StressTestInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.cars.CarEntry;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.cars.CarEntryBuilder;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author Thomas Pantelis
+ */
+public class CarProvider implements CarService {
+    private static final Logger log = LoggerFactory.getLogger(PurchaseCarProvider.class);
+
+    private final DataBroker dataProvider;
+
+    private volatile Thread testThread;
+    private volatile boolean stopThread;
+
+    public CarProvider(DataBroker dataProvider) {
+        this.dataProvider = dataProvider;
+    }
+
+    private void stopThread() {
+        if(testThread != null) {
+            stopThread = true;
+            testThread.interrupt();
+            try {
+                testThread.join();
+            } catch (InterruptedException e) {}
+            testThread = null;
+        }
+    }
+
+    @Override
+    public Future<RpcResult<Void>> stressTest(StressTestInput input) {
+        log.info("stressTest starting : rate: {}", input.getRate());
+
+        stopThread();
+
+        WriteTransaction tx = dataProvider.newWriteOnlyTransaction();
+        InstanceIdentifier<Cars> carsId = InstanceIdentifier.<Cars>builder(Cars.class).build();
+        tx.merge(LogicalDatastoreType.CONFIGURATION, carsId, new CarsBuilder().build());
+        try {
+            tx.submit().checkedGet(5, TimeUnit.SECONDS);
+        } catch (TransactionCommitFailedException | TimeoutException e) {
+            log.error("Put Cars failed",e);
+            return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        }
+
+        stopThread = false;
+        final long sleep = TimeUnit.NANOSECONDS.convert(1000,TimeUnit.MILLISECONDS) / input.getRate();
+        final Stopwatch sw = Stopwatch.createUnstarted();
+        testThread = new Thread() {
+            @Override
+            public void run() {
+                sw.start();
+                AtomicLong count = new AtomicLong();
+                while(!stopThread) {
+                    long id = count.incrementAndGet();
+                    WriteTransaction tx = dataProvider.newWriteOnlyTransaction();
+                    CarEntry car = new CarEntryBuilder().setId(new CarId("car"+id)).build();
+                    tx.put(LogicalDatastoreType.CONFIGURATION,
+                            InstanceIdentifier.<Cars>builder(Cars.class).child(CarEntry.class, car.getKey()).build(),
+                            car);
+                    tx.submit();
+                    try {
+                        TimeUnit.NANOSECONDS.sleep(sleep);
+                    } catch (InterruptedException e) {
+                        break;
+                    }
+
+                    if((count.get() % 1000) == 0) {
+                        log.info("Cars created {}, time: {}",count.get(),sw.elapsed(TimeUnit.SECONDS));
+                    }
+                }
+
+                log.info("Stress test thread stopping");
+            }
+        };
+        testThread.start();
+
+        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+    }
+
+    @Override
+    public Future<RpcResult<Void>> stopStressTest() {
+        stopThread();
+        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+    }
+
+}

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/config/yang/config/clustering_it_provider/ClusteringItProviderModule.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/config/yang/config/clustering_it_provider/ClusteringItProviderModule.java
index d91d40a34dfb9efaeb0e8c6bf2817362fed4458c..aa24d91063f332bd53ffe8ca11f03cd8d96219af 100644 (file)
--- a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/config/yang/config/clustering_it_provider/ClusteringItProviderModule.java
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/config/yang/config/clustering_it_provider/ClusteringItProviderModule.java
@@ -10,12 +10,14 @@ package org.opendaylight.controller.config.yang.config.clustering_it_provider;
 
 
 import org.opendaylight.controller.clustering.it.listener.PeopleCarListener;
+import org.opendaylight.controller.clustering.it.provider.CarProvider;
 import org.opendaylight.controller.clustering.it.provider.PeopleProvider;
 import org.opendaylight.controller.clustering.it.provider.PurchaseCarProvider;
 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.purchase.rev140818.CarPurchaseService;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.CarService;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.people.rev140818.PeopleService;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.yang.binding.NotificationListener;
@@ -52,6 +54,9 @@ public class ClusteringItProviderModule extends org.opendaylight.controller.conf
 
       people.setRpcRegistration(purchaseCarRpc);
 
+      CarProvider carProvider = new CarProvider(dataBrokerService);
+      getRpcRegistryDependency().addRpcImplementation(CarService.class, carProvider);
+
       final BindingAwareBroker.RpcRegistration<PeopleService> peopleRpcReg = getRpcRegistryDependency()
           .addRpcImplementation(PeopleService.class, people);