updated tests and docs

CHANGELOG

@@ -1,6 +1,8 @@
 v1.4
 ----
 
+* added replication
+
 * removed action deployment tool, this now handled with Foxx and its manager or
   by kaerus node utility
 

UnitTests/Makefile.unittests

@@ -562,7 +562,7 @@ unittests-import:
 
 .PHONY: unittests-replication unittests-replication-common unittests-replication-http unittests-replication-data unittests-replication-logger
 	
-unittests-replication: unittests-replication-common unittests-replication-data unittests-replication-http unittests-replication-logger
+unittests-replication: unittests-replication-common unittests-replication-http unittests-replication-data unittests-replication-logger
 
 unittests-replication-common:
 	@echo
@@ -630,8 +630,8 @@ unittests-replication-logger:
 
 	@rm -rf "$(VOCDIR)"
 	@mkdir "$(VOCDIR)"
-
+	echo "{\"autoStart\" : true }" > "$(VOCDIR)/REPLICATION-LOGGER-CONFIG"
-	$(VALGRIND) @builddir@/bin/arangod "$(VOCDIR)" $(SERVER_OPT) --replication.enable-logger true $(UNITTESTS_SERVER) || test "x$(FORCE)" == "x1"
+	$(VALGRIND) @builddir@/bin/arangod "$(VOCDIR)" $(SERVER_OPT) $(UNITTESTS_SERVER) || test "x$(FORCE)" == "x1"
 
 	@rm -rf "$(VOCDIR)"
 	@echo

arangod/RestHandler/RestReplicationHandler.cpp

@@ -47,6 +47,7 @@ using namespace triagens::arango;
 
   
 const uint64_t RestReplicationHandler::minChunkSize = 512 * 1024;
+const uint64_t RestReplicationHandler::maxChunkSize = 128 * 1024 * 1024;
 
 // -----------------------------------------------------------------------------
 // --SECTION--                                      constructors and destructors
@@ -279,7 +280,7 @@ bool RestReplicationHandler::filterCollection (TRI_vocbase_col_t* collection,
 /// @brief insert the applier action into an action list
 ////////////////////////////////////////////////////////////////////////////////
 
-void RestReplicationHandler::insertClient () {
+void RestReplicationHandler::insertClient (TRI_voc_tick_t lastFoundTick) {
   bool found;
   char const* value;
 
@@ -289,7 +290,7 @@ void RestReplicationHandler::insertClient () {
     TRI_server_id_t serverId = (TRI_server_id_t) StringUtils::uint64(value);
 
     if (serverId > 0) {
-      TRI_UpdateClientReplicationLogger(_vocbase->_replicationLogger, serverId, _request->fullUrl().c_str());
+      TRI_UpdateClientReplicationLogger(_vocbase->_replicationLogger, serverId, lastFoundTick);
     }
   }
 }
@@ -306,12 +307,13 @@ uint64_t RestReplicationHandler::determineChunkSize () const {
   const char* value = _request->value("chunkSize", found);
 
   if (found) {
-    // url parameter "chunkSize" specified
+    // url parameter "chunkSize" was specified
-    chunkSize = (uint64_t) StringUtils::uint64(value);
+    chunkSize = StringUtils::uint64(value);
+
+    // don't allow overly big allocations
+    if (chunkSize > maxChunkSize) {
+      chunkSize = maxChunkSize;
     }
-  else {
-    // not specified, use default
-    chunkSize = minChunkSize;
   }
 
   return chunkSize;
@@ -472,7 +474,11 @@ void RestReplicationHandler::handleCommandLoggerStop () {
 /// - `clients`: a list of all replication clients that ever connected to
 ///   the logger since it was started. This list can be used to determine 
 ///   approximately how much data the individual clients have already fetched 
-///   from the logger server.
+///   from the logger server. Each entry in the list contains a `time` value
+///   indicating the server time the client last fetched data from the
+///   replication logger. The `lastFoundTick` value of each client indicates
+///   the latest tick value sent to the client upon a client request to the
+///   replication logger.
 ///
 /// @RESTRETURNCODES
 ///
@@ -1021,7 +1027,7 @@ void RestReplicationHandler::handleCommandLoggerFollow () {
       TRI_StealStringBuffer(dump._buffer);
     }
     
-    insertClient();
+    insertClient(dump._lastFoundTick);
   }
   else {
     generateError(HttpResponse::SERVER_ERROR, res);
@@ -1177,8 +1183,6 @@ void RestReplicationHandler::handleCommandInventory () {
   
   generateResult(&json);
   TRI_DestroyJson(TRI_CORE_MEM_ZONE, &json);
-    
-  insertClient();
 }
     
 ////////////////////////////////////////////////////////////////////////////////
@@ -1403,8 +1407,6 @@ void RestReplicationHandler::handleCommandDump () {
     _response->body().appendText(TRI_BeginStringBuffer(dump._buffer), length);
     // avoid double freeing
     TRI_StealStringBuffer(dump._buffer);
-    
-    insertClient();
   }
   else {
     generateError(HttpResponse::SERVER_ERROR, res);

arangod/RestHandler/RestReplicationHandler.h

@@ -162,7 +162,7 @@ namespace triagens {
 /// @brief insert the applier action into an action list
 ////////////////////////////////////////////////////////////////////////////////
 
-        void insertClient ();
+        void insertClient (TRI_voc_tick_t);
 
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief determine chunk size from request
@@ -281,6 +281,12 @@ namespace triagens {
   
         static const uint64_t minChunkSize;
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief maximum chunk size
+////////////////////////////////////////////////////////////////////////////////
+
+        static const uint64_t maxChunkSize;
+
      };
   }
 }

arangod/VocBase/replication-logger.c

@@ -64,7 +64,7 @@
 
 typedef struct logger_client_s {
   TRI_server_id_t _serverId;
-  char*           _url;
+  TRI_voc_tick_t  _lastFoundTick;
   char            _stamp[24];
 }
 logger_client_t;
@@ -122,10 +122,6 @@ static bool IsEqualKeyClient (TRI_associative_pointer_t* array,
 ////////////////////////////////////////////////////////////////////////////////
 
 static void FreeClient (logger_client_t* client) {
-  if (client->_url != NULL) {
-    TRI_Free(TRI_UNKNOWN_MEM_ZONE, client->_url);
-  }
-
   TRI_Free(TRI_UNKNOWN_MEM_ZONE, client);
 }
 
@@ -1605,14 +1601,18 @@ TRI_json_t* TRI_JsonClientsReplicationLogger (TRI_replication_logger_t* logger)
       TRI_json_t* element = TRI_CreateArrayJson(TRI_CORE_MEM_ZONE);
 
       if (element != NULL) {
-        char* serverId = TRI_StringUInt64(client->_serverId);
+        char* value;
         
-        if (serverId != NULL) {
+        value = TRI_StringUInt64(client->_serverId);
-          TRI_Insert3ArrayJson(TRI_CORE_MEM_ZONE, element, "serverId", TRI_CreateStringJson(TRI_CORE_MEM_ZONE, serverId));
+
+        if (value != NULL) {
+          TRI_Insert3ArrayJson(TRI_CORE_MEM_ZONE, element, "serverId", TRI_CreateStringJson(TRI_CORE_MEM_ZONE, value));
         }
         
-        if (client->_url != NULL) {
+        value = TRI_StringUInt64(client->_lastFoundTick);
-          TRI_Insert3ArrayJson(TRI_CORE_MEM_ZONE, element, "url", TRI_CreateStringCopyJson(TRI_CORE_MEM_ZONE, client->_url));
+
+        if (value != NULL) {
+          TRI_Insert3ArrayJson(TRI_CORE_MEM_ZONE, element, "lastFoundTick", TRI_CreateStringJson(TRI_CORE_MEM_ZONE, value));
         }
 
         TRI_Insert3ArrayJson(TRI_CORE_MEM_ZONE, element, "time", TRI_CreateStringCopyJson(TRI_CORE_MEM_ZONE, client->_stamp));
@@ -1633,7 +1633,7 @@ TRI_json_t* TRI_JsonClientsReplicationLogger (TRI_replication_logger_t* logger)
 
 void TRI_UpdateClientReplicationLogger (TRI_replication_logger_t* logger,
                                         TRI_server_id_t serverId,
-                                        char const* url) {
+                                        TRI_voc_tick_t lastFoundTick) {
 
   logger_client_t* client;
   void* found;
@@ -1647,13 +1647,7 @@ void TRI_UpdateClientReplicationLogger (TRI_replication_logger_t* logger,
   TRI_GetTimeStampReplication(client->_stamp, sizeof(client->_stamp) - 1);
 
   client->_serverId      = serverId;
-  client->_url = TRI_DuplicateStringZ(TRI_UNKNOWN_MEM_ZONE, url);
+  client->_lastFoundTick = lastFoundTick;
-
-  if (client->_url == NULL) {
-    // OOM
-    TRI_Free(TRI_UNKNOWN_MEM_ZONE, client);
-    return;
-  }
 
   TRI_WriteLockReadWriteLock(&logger->_clientsLock);
 

arangod/VocBase/replication-logger.h

@@ -193,7 +193,7 @@ struct TRI_json_s* TRI_JsonClientsReplicationLogger (TRI_replication_logger_t*);
 
 void TRI_UpdateClientReplicationLogger (TRI_replication_logger_t*,
                                         TRI_server_id_t,
-                                        char const*);
+                                        TRI_voc_tick_t);
 
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief start the replication logger

js/common/tests/shell-replication.js

@@ -81,6 +81,7 @@ function ReplicationLoggerSuite () {
 ////////////////////////////////////////////////////////////////////////////////
 
     setUp : function () {
+      replication.logger.properties({ maxEvents: 1048576 });
     },
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -89,6 +90,7 @@ function ReplicationLoggerSuite () {
 
     tearDown : function () {
       replication.logger.stop();
+      replication.logger.properties({ maxEvents: 1048576 });
       db._drop(cn);
       db._drop(cn2);
     },
@@ -423,6 +425,60 @@ function ReplicationLoggerSuite () {
       assertEqual(0, properties.maxEventsSize);
     },
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test capped logging
+////////////////////////////////////////////////////////////////////////////////
+
+    testCappedLogger : function () {
+      var state, tick;
+      
+      state = replication.logger.state().state;
+      assertFalse(state.running);
+      tick = state.lastLogTick;
+      assertTrue(typeof tick === 'string');
+
+      replication.logger.properties({ maxEvents: 5000 });
+      replication.logger.start();
+
+      // do something that will cause logging
+      var c = db._create(cn);
+
+      for (var i = 0; i < 50000; ++i) {
+        c.save({ value: i });
+      }
+
+      assertEqual(5000, db._collection('_replication').count());
+    },
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test uncapped logging
+////////////////////////////////////////////////////////////////////////////////
+
+    testUncappedLogger : function () {
+      var state, tick;
+      
+      state = replication.logger.state().state;
+      assertFalse(state.running);
+      tick = state.lastLogTick;
+      assertTrue(typeof tick === 'string');
+
+      // manually flush all entries in the collection
+      db._collection('_replication').truncate();
+
+      replication.logger.properties({ maxEvents: 0, maxEventsSize: 0 });
+      replication.logger.start();
+
+      // do something that will cause logging
+      var c = db._create(cn);
+
+      for (var i = 0; i < 50000; ++i) {
+        c.save({ value: i });
+      }
+
+      // 50000 + transaction start + transaction commit
+      assertEqual(50000 + 2, db._collection('_replication').count());
+    },
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief test actions
 ////////////////////////////////////////////////////////////////////////////////

js/server/tests/replication.js

@@ -136,6 +136,7 @@ function ReplicationSuite () {
       db._drop(cn);
       db._drop(cn2);
     
+      replication.logger.properties({ maxEvents: 1048576 });
       replication.logger.start();
     },
 
@@ -146,6 +147,7 @@ function ReplicationSuite () {
     tearDown : function () {
       connectToMaster();
       replication.logger.stop();
+      replication.logger.properties({ maxEvents: 1048576 });
       
       db._drop(cn);
       db._drop(cn2);
@@ -156,6 +158,62 @@ function ReplicationSuite () {
       db._drop(cn2);
     },
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test exceeding cap
+////////////////////////////////////////////////////////////////////////////////
+
+    testCapped : function () {
+      connectToMaster();
+      // set a low cap which we'll exceed easily
+      replication.logger.properties({ maxEvents: 4096 });
+
+      compare(
+        function (state) {
+          var c = db._create(cn), i;
+ 
+          for (i = 0; i < 50000; ++i) {
+            c.save({ "value" : i });
+          }
+      
+          state.checksum = collectionChecksum(cn);
+          state.count = collectionCount(cn);
+          assertEqual(50000, state.count);
+        },
+        function (state) {
+          assertEqual(state.checksum, collectionChecksum(cn));
+          assertEqual(state.count, collectionCount(cn));
+        }
+      );
+    },
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test not exceeding cap
+////////////////////////////////////////////////////////////////////////////////
+
+    testUncapped : function () {
+      connectToMaster();
+      // set a low cap which we'll exceed easily
+      replication.logger.properties({ maxEvents: 0, maxEventsSize: 0 });
+
+      compare(
+        function (state) {
+          var c = db._create(cn), i;
+ 
+          for (i = 0; i < 50000; ++i) {
+            c.save({ "value" : i });
+          }
+      
+          state.checksum = collectionChecksum(cn);
+          state.count = collectionCount(cn);
+          assertEqual(50000, state.count);
+        },
+        function (state) {
+          assertEqual(state.checksum, collectionChecksum(cn));
+          assertEqual(state.count, collectionCount(cn));
+        }
+      );
+    },
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief test documents
 ////////////////////////////////////////////////////////////////////////////////