Added test

config/proto/config.proto

@@ -482,6 +482,8 @@ message FrontendResourceControl {
     // client info from cache too frequently can result in a lot more
     // IO. Default size of this cache is the expected_clients above.
     uint64 client_info_lru_ttl = 27;
+
+    // How often to sync client info records (ms)
     uint64 client_info_sync_time = 29;
 
     // The journal files are used to queue messages between event

services/client_info/client_info.go

@@ -38,7 +38,6 @@ package client_info
 import (
 	"context"
 	"errors"
-	"fmt"
 	"sync"
 	"time"
 
@@ -201,7 +200,6 @@ func (self *CachedInfo) Flush() error {
 		return err
 	}
 
-	fmt.Printf("Writing client info for %v\n", client_id)
 	// A blind write will eventually hit the disk.
 	client_path_manager := paths.NewClientPathManager(client_id)
 	db.SetSubjectWithCompletion(
@@ -307,7 +305,7 @@ func (self *ClientInfoManager) MutationSync(
 	sync_time := time.Duration(10) * time.Second
 	if config_obj.Frontend != nil && config_obj.Frontend.Resources != nil &&
 		config_obj.Frontend.Resources.ClientInfoSyncTime > 0 {
-		sync_time = time.Duration(config_obj.Frontend.Resources.ClientInfoSyncTime) * time.Second
+		sync_time = time.Duration(config_obj.Frontend.Resources.ClientInfoSyncTime) * time.Millisecond
 	}
 
 	journal, err := services.GetJournal()

services/client_info/client_info_test.go

@@ -27,6 +27,9 @@ type ClientInfoTestSuite struct {
 }
 
 func (self *ClientInfoTestSuite) SetupTest() {
+	self.ConfigObj = self.TestSuite.LoadConfig()
+	self.ConfigObj.Frontend.Resources.ClientInfoSyncTime = 1
+
 	self.TestSuite.SetupTest()
 
 	// Create a client in the datastore
@@ -125,6 +128,21 @@ func (self *ClientInfoTestSuite) TestMasterMinion() {
 		assert.NoError(self.T(), err)
 		return client_info.IpAddress == "127.0.0.1"
 	})
+
+	// Make sure we actually write the result in the datastore.
+	client_path_manager := paths.NewClientPathManager(self.client_id)
+
+	// Read the main client record
+	db, err := datastore.GetDB(self.ConfigObj)
+	assert.NoError(self.T(), err)
+
+	vtesting.WaitUntil(2*time.Second, self.T(), func() bool {
+		ping_info := &services.ClientInfo{}
+		db.GetSubject(self.ConfigObj, client_path_manager.Ping(),
+			ping_info)
+		utils.Debug(ping_info)
+		return ping_info.IpAddress == "127.0.0.1"
+	})
 }
 
 func TestClientInfoService(t *testing.T) {

services/journal/replication.go

@@ -501,8 +501,9 @@ func (self *ReplicationService) watchOnce(
 	subctx, cancel := context.WithCancel(ctx)
 
 	stream, err := self.api_client.WatchEvent(subctx, &api_proto.EventRequest{
-		Queue:       queue,
-		WatcherName: watcher_name,
+		Queue: queue,
+		WatcherName: watcher_name + "_" +
+			services.GetNodeName(self.config_obj.Frontend),
 	})
 	if err != nil {
 		close(output_chan)