Updated config reload

services/apierv1.go

@@ -64,6 +64,9 @@ type ApierV1Service struct {
 
 	api      *v1.ApierV1
 	connChan chan rpcclient.ClientConnector
+
+	syncStop   chan struct{}
+	storDBChan chan engine.StorDB
 }
 
 // Start should handle the sercive start
@@ -79,10 +82,15 @@ func (api *ApierV1Service) Start() (err error) {
 	api.Lock()
 	defer api.Unlock()
 
+	api.storDBChan = make(chan engine.StorDB, 1)
+	api.syncStop = make(chan struct{})
+	api.storDB.RegisterSyncChan(api.storDBChan)
+	stordb := <-api.storDBChan
+
 	api.api = &v1.ApierV1{
 		DataManager:      api.dm.GetDM(),
-		CdrDb:            api.storDB.GetDM(),
-		StorDb:           api.storDB.GetDM(),
+		CdrDb:            stordb,
+		StorDb:           stordb,
 		Config:           api.cfg,
 		Responder:        api.responderService.GetResponder(),
 		SchedulerService: api.schedService,
@@ -102,6 +110,7 @@ func (api *ApierV1Service) Start() (err error) {
 	utils.RegisterRpcParams("", api.api)
 
 	api.connChan <- api.api
+	go api.sync()
 
 	return
 }
@@ -113,20 +122,16 @@ func (api *ApierV1Service) GetIntenternalChan() (conn chan rpcclient.ClientConne
 
 // Reload handles the change of config
 func (api *ApierV1Service) Reload() (err error) {
-	api.Lock()
-	if api.storDB.WasReconnected() { // rewrite the connection if was changed
-		api.api.SetStorDB(api.storDB.GetDM())
-	}
-	api.Unlock()
 	return
 }
 
 // Shutdown stops the service
 func (api *ApierV1Service) Shutdown() (err error) {
 	api.Lock()
-	defer api.Unlock()
+	close(api.syncStop)
 	api.api = nil
 	<-api.connChan
+	api.Unlock()
 	return
 }
 
@@ -153,3 +158,22 @@ func (api *ApierV1Service) GetApierV1() *v1.ApierV1 {
 func (api *ApierV1Service) ShouldRun() bool {
 	return api.cfg.RalsCfg().Enabled
 }
+
+// sync handles stordb sync
+func (api *ApierV1Service) sync() {
+	for {
+		select {
+		case <-api.syncStop:
+			return
+		case stordb, ok := <-api.storDBChan:
+			if !ok { // the chanel was closed by the shutdown of stordbService
+				return
+			}
+			api.Lock()
+			if api.api != nil {
+				api.api.SetStorDB(stordb)
+			}
+			api.Unlock()
+		}
+	}
+}

services/cdrs.go

@@ -61,6 +61,9 @@ type CDRServer struct {
 	rpcv2    *v2.CDRsV2
 	connChan chan rpcclient.ClientConnector
 	connMgr  *engine.ConnManager
+
+	syncStop   chan struct{}
+	storDBChan chan engine.StorDB
 }
 
 // Start should handle the sercive start
@@ -76,7 +79,13 @@ func (cdrS *CDRServer) Start() (err error) {
 
 	cdrS.Lock()
 	defer cdrS.Unlock()
-	cdrS.cdrS = engine.NewCDRServer(cdrS.cfg, cdrS.storDB.GetDM(), cdrS.dm.GetDM(),
+
+	cdrS.storDBChan = make(chan engine.StorDB, 1)
+	cdrS.syncStop = make(chan struct{})
+	cdrS.storDB.RegisterSyncChan(cdrS.storDBChan)
+	stordb := <-cdrS.storDBChan
+
+	cdrS.cdrS = engine.NewCDRServer(cdrS.cfg, stordb, cdrS.dm.GetDM(),
 		filterS, cdrS.connMgr)
 	utils.Logger.Info("Registering CDRS HTTP Handlers.")
 	cdrS.cdrS.RegisterHandlersToServer(cdrS.server)
@@ -88,6 +97,7 @@ func (cdrS *CDRServer) Start() (err error) {
 	// Make the cdr server available for internal communication
 	cdrS.server.RpcRegister(cdrS.cdrS) // register CdrServer for internal usage (TODO: refactor this)
 	cdrS.connChan <- cdrS.cdrS         // Signal that cdrS is operational
+	go cdrS.sync()
 	return
 }
 
@@ -98,18 +108,13 @@ func (cdrS *CDRServer) GetIntenternalChan() (conn chan rpcclient.ClientConnector
 
 // Reload handles the change of config
 func (cdrS *CDRServer) Reload() (err error) {
-
-	cdrS.Lock()
-	if cdrS.storDB.WasReconnected() { // rewrite the connection if was changed
-		cdrS.cdrS.SetStorDB(cdrS.storDB.GetDM())
-	}
-	cdrS.Unlock()
 	return
 }
 
 // Shutdown stops the service
 func (cdrS *CDRServer) Shutdown() (err error) {
 	cdrS.Lock()
+	close(cdrS.syncStop)
 	cdrS.cdrS = nil
 	cdrS.rpcv1 = nil
 	cdrS.rpcv2 = nil
@@ -134,3 +139,22 @@ func (cdrS *CDRServer) ServiceName() string {
 func (cdrS *CDRServer) ShouldRun() bool {
 	return cdrS.cfg.CdrsCfg().Enabled
 }
+
+// sync handles stordb sync
+func (cdrS *CDRServer) sync() {
+	for {
+		select {
+		case <-cdrS.syncStop:
+			return
+		case stordb, ok := <-cdrS.storDBChan:
+			if !ok { // the chanel was closed by the shutdown of stordbService
+				return
+			}
+			cdrS.Lock()
+			if cdrS.cdrS != nil {
+				cdrS.cdrS.SetStorDB(stordb)
+			}
+			cdrS.Unlock()
+		}
+	}
+}

services/cdrs_it_test.go

@@ -58,7 +58,6 @@ func TestCdrsReload(t *testing.T) {
 	close(chS.GetPrecacheChannel(utils.CacheTimings))
 
 	cfg.ChargerSCfg().Enabled = true
-	cfg.RalsCfg().Enabled = true
 	internalChan := make(chan rpcclient.ClientConnector, 1)
 	internalChan <- nil
 	cacheSChan := make(chan rpcclient.ClientConnector, 1)
@@ -95,6 +94,7 @@ func TestCdrsReload(t *testing.T) {
 	if stordb.IsRunning() {
 		t.Errorf("Expected service to be down")
 	}
+	cfg.RalsCfg().Enabled = true
 	var reply string
 	if err := cfg.V1ReloadConfigFromPath(&config.ConfigReloadWithArgDispatcher{
 		Path:    path.Join("/usr", "share", "cgrates", "conf", "samples", "tutmongo"),

services/stordb.go

@@ -32,8 +32,7 @@ import (
 // NewStorDBService returns the StorDB Service
 func NewStorDBService(cfg *config.CGRConfig) *StorDBService {
 	return &StorDBService{
-		cfg:    cfg,
-		dbchan: make(chan engine.StorDB, 1),
+		cfg: cfg,
 		// db:     engine.NewInternalDB([]string{}, []string{}), // to be removed
 	}
 }
@@ -44,8 +43,8 @@ type StorDBService struct {
 	cfg      *config.CGRConfig
 	oldDBCfg *config.StorDbCfg
 
-	db     engine.StorDB
-	dbchan chan engine.StorDB
+	db        engine.StorDB
+	syncChans []chan engine.StorDB
 
 	reconnected bool
 }
@@ -73,7 +72,7 @@ func (db *StorDBService) Start() (err error) {
 		fmt.Println(err)
 		return
 	}
-	db.dbchan <- db.db
+	db.sync()
 	return
 }
 
@@ -98,6 +97,7 @@ func (db *StorDBService) Reload() (err error) {
 		db.db.Close()
 		db.db = d
 		db.oldDBCfg = db.cfg.StorDbCfg().Clone()
+		db.sync() // sync only if needed
 		return
 	}
 	if db.cfg.StorDbCfg().Type == utils.MONGO {
@@ -135,6 +135,10 @@ func (db *StorDBService) Shutdown() (err error) {
 	db.db.Close()
 	db.db = nil
 	db.reconnected = false
+	for _, c := range db.syncChans {
+		close(c)
+	}
+	db.syncChans = nil
 	db.Unlock()
 	return
 }
@@ -143,6 +147,11 @@ func (db *StorDBService) Shutdown() (err error) {
 func (db *StorDBService) IsRunning() bool {
 	db.RLock()
 	defer db.RUnlock()
+	return db.isRunning()
+}
+
+// isRunning returns if the service is running (not thread safe)
+func (db *StorDBService) isRunning() bool {
 	return db != nil && db.db != nil
 }
 
@@ -156,11 +165,25 @@ func (db *StorDBService) ShouldRun() bool {
 	return db.cfg.RalsCfg().Enabled || db.cfg.CdrsCfg().Enabled
 }
 
-// GetDM returns the StorDB
-func (db *StorDBService) GetDM() engine.StorDB {
-	db.RLock()
-	defer db.RUnlock()
-	return db.db
+// RegisterSyncChan used by dependent subsystems to register a chanel to reload only the storDB(thread safe)
+func (db *StorDBService) RegisterSyncChan(c chan engine.StorDB) {
+	db.Lock()
+	db.syncChans = append(db.syncChans, c)
+	if db.isRunning() {
+		for _, c := range db.syncChans {
+			c <- db.db
+		}
+	}
+	db.Unlock()
+}
+
+// sync sends the storDB over syncChansv (not thrad safe)
+func (db *StorDBService) sync() {
+	if db.isRunning() {
+		for _, c := range db.syncChans {
+			c <- db.db
+		}
+	}
 }
 
 // needsConnectionReload returns if the DB connection needs to reloaded
@@ -179,17 +202,3 @@ func (db *StorDBService) needsConnectionReload() bool {
 	}
 	return false
 }
-
-// GetStorDBchan returns the StorDB chanel
-func (db *StorDBService) GetStorDBchan() chan engine.StorDB {
-	db.RLock()
-	defer db.RUnlock()
-	return db.dbchan
-}
-
-// WasReconnected returns if after reload the DB was recreated
-func (db *StorDBService) WasReconnected() bool {
-	db.RLock()
-	defer db.RUnlock()
-	return db.reconnected
-}