Add debug counters

logback.xml

@@ -11,4 +11,5 @@
   <logger name="LogService" level="WARN"/> <!-- Restlet access logging -->
   <logger name="net.floodlightcontroller" level="INFO"/>
   <logger name="net.floodlightcontroller.logging" level="WARN"/>
+  <logger name="org.sdnplatform" level="INFO"/>
 </configuration>

src/main/java/org/sdnplatform/sync/internal/StoreRegistry.java

@@ -111,6 +111,7 @@ public class StoreRegistry {
             dstore = new InMemoryStorageEngine<ByteArray, byte[]>(storeName);
         }
         store = new SynchronizingStorageEngine(dstore, syncManager,
+                                               syncManager.debugCounter,
                                                scope);
         localStores.put(storeName, store);
         return store;

src/main/java/org/sdnplatform/sync/internal/SyncManager.java

@@ -57,6 +57,7 @@ import net.floodlightcontroller.core.module.FloodlightModuleContext;
 import net.floodlightcontroller.core.module.FloodlightModuleException;
 import net.floodlightcontroller.core.module.IFloodlightService;
 import net.floodlightcontroller.core.util.SingletonTask;
+import net.floodlightcontroller.debugcounter.IDebugCounterService;
 import net.floodlightcontroller.storage.IStorageSourceService;
 import net.floodlightcontroller.threadpool.IThreadPoolService;
 
@@ -72,7 +73,8 @@ public class SyncManager extends AbstractSyncManager {
     protected static final Logger logger =
             LoggerFactory.getLogger(SyncManager.class.getName());
 
-    private IThreadPoolService threadPool;
+    protected IThreadPoolService threadPool;
+    protected IDebugCounterService debugCounter;
     
     /**
      * The store registry holds the storage engines that provide 
@@ -137,6 +139,15 @@ public class SyncManager extends AbstractSyncManager {
     private Map<Integer, Cursor> cursorMap = 
             new ConcurrentHashMap<Integer, Cursor>(); 
     
+    private static final String PACKAGE = 
+            ISyncService.class.getPackage().getName();
+    public static final String COUNTER_HINTS = PACKAGE + "-hints";
+    public static final String COUNTER_SENT_VALUES = PACKAGE + "-sent-values";
+    public static final String COUNTER_RECEIVED_VALUES = PACKAGE + "-received-values";
+    public static final String COUNTER_PUTS = PACKAGE + "-puts";
+    public static final String COUNTER_GETS = PACKAGE + "-gets";
+    public static final String COUNTER_ITERATORS = PACKAGE + "-iterators";
+
     // ************
     // ISyncService
     // ************
@@ -478,6 +489,7 @@ public class SyncManager extends AbstractSyncManager {
     public void init(FloodlightModuleContext context)
             throws FloodlightModuleException {
         threadPool = context.getServiceImpl(IThreadPoolService.class);
+        debugCounter = context.getServiceImpl(IDebugCounterService.class);
         Map<String, String> config = context.getConfigParams(this);
 
         String[] configProviders =
@@ -524,7 +536,7 @@ public class SyncManager extends AbstractSyncManager {
     public void startUp(FloodlightModuleContext context) 
             throws FloodlightModuleException {
         updateConfiguration();
-        rpcService = new RPCService(this);
+        rpcService = new RPCService(this, debugCounter);
         rpcService.run();
 
         cleanupTask = new SingletonTask(threadPool.getScheduledExecutor(), 
@@ -566,6 +578,7 @@ public class SyncManager extends AbstractSyncManager {
                 new ArrayList<Class<? extends IFloodlightService>>();
         l.add(IThreadPoolService.class);
         l.add(IStorageSourceService.class);
+        l.add(IDebugCounterService.class);
         return l;
     }
 
@@ -694,6 +707,7 @@ public class SyncManager extends AbstractSyncManager {
                     // XXX - todo - handle hints targeted to specific nodes
                     storeRegistry.takeHints(tasks, 50);
                     for (Hint task : tasks) {
+                        debugCounter.updateCounter(COUNTER_HINTS);
                         SynchronizingStorageEngine store = 
                                 storeRegistry.get(task.getHintKey().
                                                   getStoreName());
@@ -722,10 +736,11 @@ public class SyncManager extends AbstractSyncManager {
                             svm.getHeader().
                             setTransactionId(rpcService.
                                              getTransactionId());
+                            debugCounter.updateCounter(COUNTER_SENT_VALUES);
                             rpcService.writeToNode(n.getNodeId(), bsm);
                         }
                     }
-
+                    debugCounter.flushCounters();
                     tasks.clear(); 
                     clearMessages();
 

src/main/java/org/sdnplatform/sync/internal/rpc/RPCChannelHandler.java

@@ -11,6 +11,7 @@ import net.floodlightcontroller.core.annotations.LogMessageDoc;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelHandlerContext;
 import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.MessageEvent;
 import org.sdnplatform.sync.IVersion;
 import org.sdnplatform.sync.Versioned;
 import org.sdnplatform.sync.ISyncService.Scope;
@@ -70,6 +71,13 @@ public class RPCChannelHandler extends AbstractRPCChannelHandler {
     // AbstractRPCChannelHandler message handlers
     // ******************************************
 
+    @Override
+    public void messageReceived(ChannelHandlerContext ctx,
+                                MessageEvent e) throws Exception {
+        super.messageReceived(ctx, e);
+        rpcService.debugCounter.flushCounters();
+    }
+        
     @Override
     @LogMessageDoc(level="ERROR",
               message="[{id}->{id}] Attempted connection from unrecognized " +
@@ -259,6 +267,8 @@ public class RPCChannelHandler extends AbstractRPCChannelHandler {
             SyncMessage bsm = 
                     new SyncMessage(MessageType.SYNC_VALUE_RESPONSE);
             bsm.setSyncValueResponse(m);
+            
+            updateCounter(SyncManager.COUNTER_RECEIVED_VALUES);
             channel.write(bsm);
         } catch (Exception e) {
 
@@ -339,9 +349,11 @@ public class RPCChannelHandler extends AbstractRPCChannelHandler {
                 svm.addToValues(kv);
             }
             
-            if (svm.isSetValues())
+            if (svm.isSetValues()) {
+                updateCounter(SyncManager.COUNTER_SENT_VALUES);
                 rpcService.syncQueue.add(new NodeMessage(getRemoteNodeId(),
                                                          bsm));
+            }
         } catch (Exception e) {
             channel.write(getError(request.getHeader().getTransactionId(), e, 
                                    MessageType.SYNC_REQUEST));
@@ -461,6 +473,10 @@ public class RPCChannelHandler extends AbstractRPCChannelHandler {
     // Utility functions
     // *****************
 
+    protected void updateCounter(String counter) {
+        rpcService.debugCounter.updateCounter(counter);
+    }
+    
     protected void startAntientropy() {
         // Run antientropy in a background task so we don't use up an I/O
         // thread.  Note that this task will result in lots of traffic

src/main/java/org/sdnplatform/sync/internal/rpc/RPCService.java

@@ -20,6 +20,8 @@ import net.floodlightcontroller.core.annotations.LogMessageCategory;
 import net.floodlightcontroller.core.annotations.LogMessageDoc;
 import net.floodlightcontroller.core.annotations.LogMessageDocs;
 import net.floodlightcontroller.core.util.SingletonTask;
+import net.floodlightcontroller.debugcounter.IDebugCounterService;
+
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.bootstrap.ServerBootstrap;
 import org.jboss.netty.channel.Channel;
@@ -55,6 +57,11 @@ public class RPCService {
      */
     protected SyncManager syncManager;
 
+    /**
+     * Debug counter service
+     */
+    protected IDebugCounterService debugCounter;
+
     /**
      * Channel group that will hold all our channels
      */
@@ -156,9 +163,11 @@ public class RPCService {
      */
     protected static final int MAX_PENDING_MESSAGES = 500;
 
-    public RPCService(SyncManager syncManager) {
+    public RPCService(SyncManager syncManager, 
+                      IDebugCounterService debugCounter) {
         super();
         this.syncManager = syncManager;
+        this.debugCounter = debugCounter;
 
         messageWindows = new ConcurrentHashMap<Short, MessageWindow>();
     }

src/main/java/org/sdnplatform/sync/internal/store/ListenerStorageEngine.java

@@ -6,10 +6,13 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map.Entry;
 
+import net.floodlightcontroller.debugcounter.IDebugCounterService;
+
 import org.sdnplatform.sync.IClosableIterator;
 import org.sdnplatform.sync.IVersion;
 import org.sdnplatform.sync.Versioned;
 import org.sdnplatform.sync.error.SyncException;
+import org.sdnplatform.sync.internal.SyncManager;
 import org.sdnplatform.sync.internal.util.ByteArray;
 
 
@@ -32,10 +35,21 @@ public class ListenerStorageEngine
      */
     protected IStorageEngine<ByteArray, byte[]> localStorage;
 
+    /**
+     * Debug counter service
+     */
+    protected IDebugCounterService debugCounter;
     
+    /**
+     * Allocate new {@link ListenerStorageEngine}
+     * @param localStorage the delegate store
+     * @param debugCounter debug counter service
+     */
     public ListenerStorageEngine(IStorageEngine<ByteArray,
-                                                byte[]> localStorage) {
+                                                byte[]> localStorage,
+                                                IDebugCounterService debugCounter) {
         this.localStorage = localStorage;
+        this.debugCounter = debugCounter;
     }
 
     // *************************
@@ -44,17 +58,20 @@ public class ListenerStorageEngine
 
     @Override
     public List<Versioned<byte[]>> get(ByteArray key) throws SyncException {
+        updateCounter(SyncManager.COUNTER_GETS);
         return localStorage.get(key);
     }
 
     @Override
     public IClosableIterator<Entry<ByteArray,List<Versioned<byte[]>>>> entries() {
+        updateCounter(SyncManager.COUNTER_ITERATORS);
         return localStorage.entries();
     }
 
     @Override
     public void put(ByteArray key, Versioned<byte[]> value)
             throws SyncException {
+        updateCounter(SyncManager.COUNTER_PUTS);
         localStorage.put(key, value);
         notifyListeners(key);
     }
@@ -124,4 +141,10 @@ public class ListenerStorageEngine
             msl.notify(keys);
         }
     }
+
+    protected void updateCounter(String counterName) {
+        if (debugCounter != null) {
+            debugCounter.updateCounter(counterName);
+        }
+    }    
 }

src/main/java/org/sdnplatform/sync/internal/store/SynchronizingStorageEngine.java

 package org.sdnplatform.sync.internal.store;
 
+import net.floodlightcontroller.debugcounter.IDebugCounterService;
+
 import org.sdnplatform.sync.Versioned;
 import org.sdnplatform.sync.ISyncService.Scope;
 import org.sdnplatform.sync.error.SyncException;
@@ -32,6 +34,7 @@ public class SynchronizingStorageEngine extends ListenerStorageEngine {
      * Allocate a synchronizing storage engine
      * @param localStorage the local storage
      * @param syncManager the sync manager
+     * @param debugCounter the debug counter service
      * @param scope the scope for this store
      * @param rpcService the RPC service
      * @param storeName the name of the store
@@ -39,8 +42,9 @@ public class SynchronizingStorageEngine extends ListenerStorageEngine {
     public SynchronizingStorageEngine(IStorageEngine<ByteArray,
                                                     byte[]> localStorage,
                                       SyncManager syncManager,
+                                      IDebugCounterService debugCounter, 
                                       Scope scope) {
-        super(localStorage);
+        super(localStorage, debugCounter);
         this.localStorage = localStorage;
         this.syncManager = syncManager;
         this.scope = scope;

src/test/java/org/sdnplatform/sync/client/ClientTest.java

@@ -7,6 +7,8 @@ import java.io.PrintStream;
 import java.util.ArrayList;
 
 import net.floodlightcontroller.core.module.FloodlightModuleContext;
+import net.floodlightcontroller.debugcounter.IDebugCounterService;
+import net.floodlightcontroller.debugcounter.NullDebugCounter;
 import net.floodlightcontroller.threadpool.IThreadPoolService;
 import net.floodlightcontroller.threadpool.ThreadPool;
 
@@ -39,6 +41,7 @@ public class ClientTest {
         
         FloodlightModuleContext fmc = new FloodlightModuleContext();
         fmc.addService(IThreadPoolService.class, tp);
+        fmc.addService(IDebugCounterService.class, new NullDebugCounter());
         
         fmc.addConfigParam(syncManager, "nodes", nodeString);
         fmc.addConfigParam(syncManager, "thisNode", ""+1);

src/test/java/org/sdnplatform/sync/internal/SyncManagerTest.java

@@ -13,6 +13,8 @@ import java.util.Map.Entry;
 
 import net.floodlightcontroller.core.module.FloodlightModuleContext;
 import net.floodlightcontroller.core.module.FloodlightModuleException;
+import net.floodlightcontroller.debugcounter.IDebugCounterService;
+import net.floodlightcontroller.debugcounter.NullDebugCounter;
 import net.floodlightcontroller.threadpool.IThreadPoolService;
 import net.floodlightcontroller.threadpool.ThreadPool;
 
@@ -58,6 +60,7 @@ public class SyncManagerTest {
                                     SyncManager syncManager, Node thisNode)
             throws FloodlightModuleException {
         fmc.addService(IThreadPoolService.class, tp);
+        fmc.addService(IDebugCounterService.class, new NullDebugCounter());
         fmc.addConfigParam(syncManager, "configProviders", 
                            PropertyCCProvider.class.getName());
         fmc.addConfigParam(syncManager, "nodes", nodeString);

src/test/java/org/sdnplatform/sync/test/MockSyncService.java

@@ -42,7 +42,7 @@ public class MockSyncService extends AbstractSyncManager {
         if (store != null) return;
         IStorageEngine<ByteArray, byte[]> memstore =
                 new InMemoryStorageEngine<ByteArray, byte[]>(storeName);
-        store = new ListenerStorageEngine(memstore);
+        store = new ListenerStorageEngine(memstore, null);
         localStores.put(storeName, store);
     }