uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-13 19:56:38 +05:00
Code Issues Packages Projects Releases Wiki Activity

Add set_versions flag to cgr-engine

Browse Source 
If set to true, it sets/overwrites all versions to the
current ones.
This commit is contained in:
ionutboangiu
2024-10-31 18:38:54 +02:00
committed by Dan Christian Bogos
parent a251547408
commit 792bbb3aff
47 changed files with 115 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
services/accounts_it_test.go
View File

@@ -50,7 +50,7 @@ func TestAccountSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
acctRPC := make(chan birpc.ClientConnector, 1)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
acctS := NewAccountService(cfg, db, css, filterSChan, nil, server, acctRPC, anz, srvDep)

2
services/accounts_test.go
View File

@@ -39,7 +39,7 @@ func TestAccountSCoverage(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
actRPC := make(chan birpc.ClientConnector, 1)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)

2
services/actions_it_test.go
View File

@@ -50,7 +50,7 @@ func TestActionSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
actRPC := make(chan birpc.ClientConnector, 1)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
actS := NewActionService(cfg, db, css, filterSChan, nil, server, actRPC, anz, srvDep)

2
services/actions_test.go
View File

@@ -38,7 +38,7 @@ func TestActionSCoverage(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
actRPC := make(chan birpc.ClientConnector, 1)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)

2
services/analyzers_it_test.go
View File

@@ -49,7 +49,7 @@ func TestAnalyzerSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anzRPC := make(chan birpc.ClientConnector, 1)
anz := NewAnalyzerService(cfg, server, filterSChan, anzRPC, srvDep)
engine.NewConnManager(cfg)

4
services/asteriskagent_it_test.go
View File

@@ -51,7 +51,7 @@ func TestAsteriskAgentReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)
@@ -109,7 +109,7 @@ func TestAsteriskAgentReload2(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/attributes_it_test.go
View File

@@ -50,7 +50,7 @@ func TestAttributeSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
attrRPC := make(chan birpc.ClientConnector, 1)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
attrS := NewAttributeService(cfg, db,

2
services/attributes_test.go
View File

@@ -38,7 +38,7 @@ func TestAttributeSCoverage(t *testing.T) {
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
attrRPC := make(chan birpc.ClientConnector, 1)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
attrS := NewAttributeService(cfg, db, chS, filterSChan, server, attrRPC, anz, &DispatcherService{srvsReload: map[string]chan struct{}{}}, srvDep)

6
services/cdrs_it_test.go
View File

@@ -53,9 +53,9 @@ func TestCdrsReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
cfg.StorDbCfg().Type = utils.Internal
stordb := NewStorDBService(cfg, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
cfg.StorDbCfg().Type = utils.MetaInternal
stordb := NewStorDBService(cfg, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chrS := NewChargerService(cfg, db, css, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
cdrsRPC := make(chan birpc.ClientConnector, 1)

6
services/cdrs_test.go
View File

@@ -37,11 +37,11 @@ func TestCdrsCoverage(t *testing.T) {
cfg.ChargerSCfg().Enabled = true
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
cdrsRPC := make(chan birpc.ClientConnector, 1)
cfg.StorDbCfg().Type = utils.Internal
stordb := NewStorDBService(cfg, srvDep)
cfg.StorDbCfg().Type = utils.MetaInternal
stordb := NewStorDBService(cfg, false, srvDep)
cdrS := NewCDRServer(cfg, db, stordb, filterSChan, server,
cdrsRPC, nil, anz, srvDep)
if cdrS.IsRunning() {

8
services/cgr-engine.go
View File

@@ -130,7 +130,7 @@ func (cgr *CGREngine) AddService(service servmanager.Service, connName, apiPrefi
cgr.cM.AddInternalConn(connName, apiPrefix, iConnCh)
}
func (cgr *CGREngine) InitServices(cpuPrfFl *os.File) {
func (cgr *CGREngine) InitServices(setVersions bool, cpuPrfFl *os.File) {
if len(cgr.cfg.HTTPCfg().RegistrarSURL) != 0 {
cgr.server.RegisterHTTPFunc(cgr.cfg.HTTPCfg().RegistrarSURL, registrarc.Registrar)
}
@@ -195,8 +195,8 @@ func (cgr *CGREngine) InitServices(cpuPrfFl *os.File) {
cgr.cM.AddInternalConn(utils.ConcatenatedKey(utils.MetaInternal, utils.MetaERs), utils.ErSv1, iERsCh)
cgr.gvS = NewGlobalVarS(cgr.cfg, cgr.srvDep)
cgr.dmS = NewDataDBService(cgr.cfg, cgr.cM, cgr.srvDep)
cgr.sdbS = NewStorDBService(cgr.cfg, cgr.srvDep)
cgr.dmS = NewDataDBService(cgr.cfg, cgr.cM, setVersions, cgr.srvDep)
cgr.sdbS = NewStorDBService(cgr.cfg, setVersions, cgr.srvDep)
cgr.anzS = NewAnalyzerService(cgr.cfg, cgr.server,
cgr.iFilterSCh, iAnalyzerSCh, cgr.srvDep) // init AnalyzerS
@@ -382,7 +382,7 @@ func (cgr *CGREngine) Init(ctx *context.Context, shtDw context.CancelFunc, flags
}
efs.SetFailedPostCacheTTL(cgr.cfg.EFsCfg().FailedPostsTTL) // init failedPosts to posts loggers/exporters in case of failing
utils.Logger.Info(fmt.Sprintf("<CoreS> starting version <%s><%s>", vers, runtime.Version()))
cgr.InitServices(cpuPrfF)
cgr.InitServices(*flags.SetVersions, cpuPrfF)
return nil
}

2
services/chargers_it_test.go
View File

@@ -54,7 +54,7 @@ func TestChargerSReload(t *testing.T) {
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
attrS := NewAttributeService(cfg, db, css, filterSChan, server, make(chan birpc.ClientConnector, 1), anz, &DispatcherService{srvsReload: make(map[string]chan struct{})}, srvDep)
chrS := NewChargerService(cfg, db, css, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)

2
services/chargers_test.go
View File

@@ -38,7 +38,7 @@ func TestChargerSCoverage(t *testing.T) {
filterSChan <- nil
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
server := cores.NewServer(nil)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
chrS1 := NewChargerService(cfg, db, chS,

2
services/cores_it_test.go
View File

@@ -44,7 +44,7 @@ func TestCoreSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
coreRPC := make(chan birpc.ClientConnector, 1)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
caps := engine.NewCaps(1, "test_caps")

37
services/datadb.go
View File

@@ -29,13 +29,14 @@ import (
)
// NewDataDBService returns the DataDB Service
func NewDataDBService(cfg *config.CGRConfig, connMgr *engine.ConnManager,
func NewDataDBService(cfg *config.CGRConfig, connMgr *engine.ConnManager, setVersions bool,
srvDep map[string]*sync.WaitGroup) *DataDBService {
return &DataDBService{
cfg: cfg,
dbchan: make(chan *engine.DataManager, 1),
connMgr: connMgr,
srvDep: srvDep,
cfg: cfg,
dbchan: make(chan *engine.DataManager, 1),
connMgr: connMgr,
setVersions: setVersions,
srvDep: srvDep,
}
}
@@ -46,12 +47,14 @@ type DataDBService struct {
oldDBCfg *config.DataDbCfg
connMgr *engine.ConnManager
dm *engine.DataManager
dbchan chan *engine.DataManager
dm *engine.DataManager
dbchan chan *engine.DataManager
setVersions bool
srvDep map[string]*sync.WaitGroup
}
// Start should handle the sercive start
// Start handles the service start.
func (db *DataDBService) Start(*context.Context, context.CancelFunc) (err error) {
if db.IsRunning() {
return utils.ErrServiceAlreadyRunning
@@ -59,8 +62,7 @@ func (db *DataDBService) Start(*context.Context, context.CancelFunc) (err error)
db.Lock()
defer db.Unlock()
db.oldDBCfg = db.cfg.DataDbCfg().Clone()
var d engine.DataDBDriver
d, err = engine.NewDataDBConn(db.cfg.DataDbCfg().Type,
dbConn, err := engine.NewDataDBConn(db.cfg.DataDbCfg().Type,
db.cfg.DataDbCfg().Host, db.cfg.DataDbCfg().Port,
db.cfg.DataDbCfg().Name, db.cfg.DataDbCfg().User,
db.cfg.DataDbCfg().Password, db.cfg.GeneralCfg().DBDataEncoding,
@@ -69,10 +71,17 @@ func (db *DataDBService) Start(*context.Context, context.CancelFunc) (err error)
utils.Logger.Crit(fmt.Sprintf("Could not configure dataDb: %s exiting!", err))
return
}
db.dm = engine.NewDataManager(d, db.cfg.CacheCfg(), db.connMgr)
if err = engine.CheckVersions(db.dm.DataDB()); err != nil {
return
db.dm = engine.NewDataManager(dbConn, db.cfg.CacheCfg(), db.connMgr)
if db.setVersions {
err = engine.OverwriteDBVersions(dbConn)
} else {
err = engine.CheckVersions(db.dm.DataDB())
}
if err != nil {
return err
}
db.dbchan <- db.dm
return
}
@@ -121,7 +130,7 @@ func (db *DataDBService) Shutdown() (_ error) {
func (db *DataDBService) IsRunning() bool {
db.RLock()
defer db.RUnlock()
return db != nil && db.dm != nil && db.dm.DataDB() != nil
return db.dm != nil && db.dm.DataDB() != nil
}
// ServiceName returns the service name

14
services/datadb_it_test.go
View File

@@ -52,7 +52,7 @@ func TestDataDBReload(t *testing.T) {
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
srvMngr.AddServices(NewAttributeService(cfg, db,
css, filterSChan, server, make(chan birpc.ClientConnector, 1), anz, &DispatcherService{srvsReload: make(map[string]chan struct{})}, srvDep),
@@ -183,7 +183,7 @@ func TestDataDBReloadBadType(t *testing.T) {
filterSChan <- nil
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
db.oldDBCfg = &config.DataDbCfg{
Type: utils.MetaMongo,
Host: "127.0.0.1",
@@ -236,7 +236,7 @@ func TestDataDBReloadErrorMarsheler(t *testing.T) {
filterSChan <- nil
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
if db.IsRunning() {
t.Errorf("Expected service to be down")
@@ -325,7 +325,7 @@ func TestDataDBStartVersion(t *testing.T) {
filterSChan <- nil
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
ctx, cancel := context.WithCancel(context.TODO())
err = db.Start(ctx, cancel)
if err == nil || err.Error() != "Migration needed: please backup cgr data and run : <cgr-migrator -exec=*attributes>" {
@@ -377,7 +377,7 @@ func TestDataDBReloadCastError(t *testing.T) {
filterSChan <- nil
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
db.oldDBCfg = &config.DataDbCfg{
Type: utils.MetaMongo,
Host: "127.0.0.1",
@@ -427,7 +427,7 @@ func TestDataDBStartAttributeSCfgErr(t *testing.T) {
cfg := config.NewDefaultCGRConfig()
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
cfg.DataDbCfg().Type = "badtype"
cfg.AttributeSCfg().Enabled = true
cfg.SessionSCfg().ListenBijson = ""
@@ -442,7 +442,7 @@ func TestDataDBReloadError(t *testing.T) {
cfg := config.NewDefaultCGRConfig()
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
cfg.GeneralCfg().DBDataEncoding = utils.JSON
db.oldDBCfg = &config.DataDbCfg{
Type: utils.MetaMongo,

2
services/datadb_test.go
View File

@@ -36,7 +36,7 @@ func TestDataDBCoverage(t *testing.T) {
filterSChan <- nil
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
cM := engine.NewConnManager(cfg)
db := NewDataDBService(cfg, cM, srvDep)
db := NewDataDBService(cfg, cM, false, srvDep)
if db.IsRunning() {
t.Errorf("Expected service to be down")
}

2
services/diameteragent_it_test.go
View File

@@ -45,7 +45,7 @@ func TestDiameterAgentReload1(t *testing.T) {
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/dispatchers_it_test.go
View File

@@ -55,7 +55,7 @@ func TestDispatcherSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
srv := NewDispatcherService(cfg, db, css, filterSChan, server,
make(chan birpc.ClientConnector, 1), nil, anz, srvDep)

2
services/dispatchers_test.go
View File

@@ -39,7 +39,7 @@ func TestDispatcherSCoverage(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
srv := NewDispatcherService(cfg, db, chS, filterSChan, server,

4
services/dnsagent_it_test.go
View File

@@ -65,7 +65,7 @@ func TestDNSAgentStartReloadShut(t *testing.T) {
shdWg := new(sync.WaitGroup)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
engine.NewConnManager(cfg)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
server := cores.NewServer(nil)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
@@ -112,7 +112,7 @@ func TestDNSAgentReloadFirst(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/ees_it_test.go
View File

@@ -55,7 +55,7 @@ func TestEventExporterSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
chS := engine.NewCacheS(cfg, nil, nil, nil)
close(chS.GetPrecacheChannel(utils.CacheAttributeProfiles))
close(chS.GetPrecacheChannel(utils.CacheAttributeFilterIndexes))

2
services/ers_it_test.go
View File

@@ -63,7 +63,7 @@ func TestEventReaderSReload(t *testing.T) {
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
intERsConn := make(chan birpc.ClientConnector, 1)
erS := NewEventReaderService(cfg, filterSChan, nil, server, intERsConn, anz, srvDep)

2
services/freeswitchagent_it_test.go
View File

@@ -53,7 +53,7 @@ func TestFreeSwitchAgentReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/httpagent_it_test.go
View File

@@ -51,7 +51,7 @@ func TestHTTPAgentReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/kamailioagent_it_test.go
View File

@@ -50,7 +50,7 @@ func TestKamailioAgentReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/libcgr-engine.go
View File

@@ -60,6 +60,7 @@ func NewCGREngineFlags() *CGREngineFlags {
LogLevel: fs.Int(utils.LogLevelCfg, -1, "Log level (0-emergency to 7-debug)"),
Preload: fs.String(utils.PreloadCgr, utils.EmptyString, "LoaderIDs used to load the data before the engine starts"),
CheckConfig: fs.Bool(utils.CheckCfgCgr, false, "Verify the config without starting the engine"),
SetVersions: fs.Bool(utils.SetVersionsCgr, false, "Overwrite database versions (equivalent to cgr-migrator -exec=*set_versions)"),
}
}
@@ -81,6 +82,7 @@ type CGREngineFlags struct {
LogLevel *int
Preload *string
CheckConfig *bool
SetVersions *bool
}
func cgrSingnalHandler(ctx *context.Context, shutdown context.CancelFunc,

7
services/libcgr-engine_test.go
View File

@@ -141,6 +141,13 @@ func TestCgrEngineFlags(t *testing.T) {
defaultVal: false,
want: true,
},
{
name: "setVersions",
flags: []string{"-set_versions"},
flagVar: ngFlags.SetVersions,
defaultVal: false,
want: true,
},
}
for _, tt := range tests {

6
services/loaders_it_test.go
View File

@@ -73,7 +73,7 @@ func TestLoaderSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
conMngr := engine.NewConnManager(cfg)
srv := NewLoaderService(cfg, db, filterSChan,
@@ -152,7 +152,7 @@ func TestLoaderSReload2(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
db.dbchan <- new(engine.DataManager)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
srv := NewLoaderService(cfg, db, filterSChan,
@@ -177,7 +177,7 @@ func TestLoaderSReload3(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
db.dbchan <- new(engine.DataManager)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
srv := NewLoaderService(cfg, db, filterSChan,

2
services/loaders_test.go
View File

@@ -37,7 +37,7 @@ func TestLoaderSCoverage(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
internalLoaderSChan := make(chan birpc.ClientConnector, 1)
cM := engine.NewConnManager(cfg)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)

4
services/radiusagent_it_test.go
View File

@@ -54,7 +54,7 @@ func TestRadiusAgentReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)
@@ -114,7 +114,7 @@ func TestRadiusAgentReload2(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/rates_it_test.go
View File

@@ -45,7 +45,7 @@ func TestRateSReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
chS := engine.NewCacheS(cfg, nil, nil, nil)
close(chS.GetPrecacheChannel(utils.CacheRateProfiles))
close(chS.GetPrecacheChannel(utils.CacheRateProfilesFilterIndexes))

2
services/rates_test.go
View File

@@ -39,7 +39,7 @@ func TestRateSCoverage(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
rS := NewRateService(cfg, chS, filterSChan, db, server, make(chan birpc.ClientConnector, 1), anz, srvDep)

2
services/registrarc_it_test.go
View File

@@ -52,7 +52,7 @@ func TestDispatcherHReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
connMngr := engine.NewConnManager(cfg)
srv := NewRegistrarCService(cfg, server, connMngr, anz, srvDep)

2
services/resources_it_test.go
View File

@@ -58,7 +58,7 @@ func TestResourceSReload(t *testing.T) {
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
tS := NewThresholdService(cfg, db, css, filterSChan, nil, server, make(chan birpc.ClientConnector, 1), anz, srvDep)
reS := NewResourceService(cfg, db, css, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
engine.NewConnManager(cfg)

2
services/resources_test.go
View File

@@ -39,7 +39,7 @@ func TestResourceSCoverage(t *testing.T) {
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
reS := NewResourceService(cfg, db, chS, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)

2
services/routes_it_test.go
View File

@@ -53,7 +53,7 @@ func TestRouteSReload(t *testing.T) {
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
routeS := NewRouteService(cfg, db, css, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
engine.NewConnManager(cfg)
srvMngr.AddServices(routeS,

2
services/routes_test.go
View File

@@ -38,7 +38,7 @@ func TestSupplierSCoverage(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
supS := NewRouteService(cfg, db, chS, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)

6
services/sessions_it_test.go
View File

@@ -95,7 +95,7 @@ func TestSessionSReload1(t *testing.T) {
conMng := engine.NewConnManager(cfg)
conMng.AddInternalConn(utils.ConcatenatedKey(utils.MetaInternal, utils.MetaChargers), utils.ChargerSv1, clientConect)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, conMng, srvDep)
db := NewDataDBService(cfg, conMng, false, srvDep)
db.dbchan = make(chan *engine.DataManager, 1)
db.dbchan <- nil
srv := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1), conMng, anz, srvDep)
@@ -156,7 +156,7 @@ func TestSessionSReload2(t *testing.T) {
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
srv := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
engine.NewConnManager(cfg)
@@ -202,7 +202,7 @@ func TestSessionSReload3(t *testing.T) {
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
srv := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)

2
services/sessions_test.go
View File

@@ -49,7 +49,7 @@ func TestSessionSCoverage(t *testing.T) {
filterSChan <- nil
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
srv := NewSessionService(cfg, db, nil, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
engine.NewConnManager(cfg)

2
services/sipagent_it_test.go
View File

@@ -48,7 +48,7 @@ func TestSIPAgentReload(t *testing.T) {
server := cores.NewServer(nil)
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
sS := NewSessionService(cfg, db, filterSChan, server, make(chan birpc.ClientConnector, 1),
nil, anz, srvDep)

2
services/stats_it_test.go
View File

@@ -58,7 +58,7 @@ func TestStatSReload(t *testing.T) {
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
tS := NewThresholdService(cfg, db, css, filterSChan, nil, server, make(chan birpc.ClientConnector, 1), anz, srvDep)
sS := NewStatService(cfg, db, css, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
engine.NewConnManager(cfg)

2
services/stats_test.go
View File

@@ -40,7 +40,7 @@ func TestStatSCoverage(t *testing.T) {
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
sS := NewStatService(cfg, db, chS, filterSChan, server, make(chan birpc.ClientConnector, 1), nil, anz, srvDep)
if sS.IsRunning() {

26
services/stordb.go
View File

@@ -29,11 +29,12 @@ import (
)
// NewStorDBService returns the StorDB Service
func NewStorDBService(cfg *config.CGRConfig,
func NewStorDBService(cfg *config.CGRConfig, setVersions bool,
srvDep map[string]*sync.WaitGroup) *StorDBService {
return &StorDBService{
cfg: cfg,
srvDep: srvDep,
cfg: cfg,
setVersions: setVersions,
srvDep: srvDep,
}
}
@@ -43,8 +44,9 @@ type StorDBService struct {
cfg *config.CGRConfig
oldDBCfg *config.StorDbCfg
db engine.StorDB
syncChans []chan engine.StorDB
db engine.StorDB
syncChans []chan engine.StorDB
setVersions bool
srvDep map[string]*sync.WaitGroup
}
@@ -57,7 +59,7 @@ func (db *StorDBService) Start(*context.Context, context.CancelFunc) (err error)
db.Lock()
defer db.Unlock()
db.oldDBCfg = db.cfg.StorDbCfg().Clone()
d, err := engine.NewStorDBConn(db.cfg.StorDbCfg().Type, db.cfg.StorDbCfg().Host,
dbConn, err := engine.NewStorDBConn(db.cfg.StorDbCfg().Type, db.cfg.StorDbCfg().Host,
db.cfg.StorDbCfg().Port, db.cfg.StorDbCfg().Name, db.cfg.StorDbCfg().User,
db.cfg.StorDbCfg().Password, db.cfg.GeneralCfg().DBDataEncoding,
db.cfg.StorDbCfg().StringIndexedFields, db.cfg.StorDbCfg().PrefixIndexedFields,
@@ -66,7 +68,17 @@ func (db *StorDBService) Start(*context.Context, context.CancelFunc) (err error)
utils.Logger.Crit(fmt.Sprintf("Could not configure storDB: %s exiting!", err))
return
}
db.db = d
db.db = dbConn
if db.setVersions {
err = engine.OverwriteDBVersions(dbConn)
} else {
err = engine.CheckVersions(db.db)
}
if err != nil {
return err
}
db.sync()
return
}

4
services/thresholds_it_test.go
View File

@@ -54,7 +54,7 @@ func TestThresholdSReload(t *testing.T) {
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
tS := NewThresholdService(cfg, db, css, filterSChan, nil, server, make(chan birpc.ClientConnector, 1), anz, srvDep)
engine.NewConnManager(cfg)
srvMngr.AddServices(tS,
@@ -123,7 +123,7 @@ func TestThresholdSReload2(t *testing.T) {
srvMngr := servmanager.NewServiceManager(shdWg, nil, cfg)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
tS := NewThresholdService(cfg, db, css, filterSChan, nil, server, make(chan birpc.ClientConnector, 1), anz, srvDep)
engine.NewConnManager(cfg)
srvMngr.AddServices(tS,

2
services/thresholds_test.go
View File

@@ -38,7 +38,7 @@ func TestThresholdSCoverage(t *testing.T) {
server := cores.NewServer(nil)
srvDep := map[string]*sync.WaitGroup{utils.DataDB: new(sync.WaitGroup)}
anz := NewAnalyzerService(cfg, server, filterSChan, make(chan birpc.ClientConnector, 1), srvDep)
db := NewDataDBService(cfg, nil, srvDep)
db := NewDataDBService(cfg, nil, false, srvDep)
chS := NewCacheService(cfg, db, nil, server, make(chan context.ClientConnector, 1), anz, nil, srvDep)
tS := NewThresholdService(cfg, db, chS, filterSChan, nil, server, make(chan birpc.ClientConnector, 1), anz, srvDep)
if tS.IsRunning() {

1
utils/consts.go
View File

@@ -2719,6 +2719,7 @@ const (
ScheduledShutdownCgr = "scheduled_shutdown"
SingleCpuCgr = "singlecpu"
PreloadCgr = "preload"
SetVersionsCgr = "set_versions"
MemProfFinalFile = "mem_final.prof"
CpuPathCgr = "cpu.prof"
//Cgr loader