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

Added support for in memory db for analyzers

Browse Source
This commit is contained in:
Trial97
2021-09-14 16:26:16 +03:00
committed by Dan Christian Bogos
parent 968ba9c694
commit 7fa6ea769c
7 changed files with 54 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
analyzers/analyzers.go
View File

@@ -61,6 +61,10 @@ func (aS *AnalyzerService) SetFilterS(fS *engine.FilterS) {
}
func (aS *AnalyzerService) initDB() (err error) {
if aS.cfg.AnalyzerSCfg().DBPath == utils.EmptyString {
aS.db, err = bleve.NewMemOnly(bleve.NewIndexMapping())
return
}
dbPath := path.Join(aS.cfg.AnalyzerSCfg().DBPath, utils.AnzDBDir)
if _, err = os.Stat(dbPath); err == nil {
aS.db, err = bleve.Open(dbPath)

45
analyzers/analyzers_test.go
View File

@@ -440,3 +440,48 @@ func TestAnalyzersV1Search(t *testing.T) {
t.Fatal(err)
}
}
func TestAnalyzerSLogTrafficInternalDB(t *testing.T) {
cfg := config.NewDefaultCGRConfig()
cfg.AnalyzerSCfg().DBPath = utils.EmptyString
cfg.AnalyzerSCfg().TTL = 30 * time.Minute
anz, err := NewAnalyzerService(cfg)
if err != nil {
t.Fatal(err)
}
t1 := time.Now().Add(-time.Hour)
if err = anz.logTrafic(0, utils.AnalyzerSv1Ping, "status", "result", "error",
utils.MetaJSON, "127.0.0.1:5565", "127.0.0.1:2012", t1, t1.Add(time.Second)); err != nil {
t.Fatal(err)
}
if err = anz.logTrafic(0, utils.CoreSv1Status, "status", "result", "error",
utils.MetaJSON, "127.0.0.1:5565", "127.0.0.1:2012", t1, t1.Add(time.Second)); err != nil {
t.Fatal(err)
}
t1 = time.Now().Add(-10 * time.Minute)
if err = anz.logTrafic(0, utils.CoreSv1Status, "status", "result", "error",
utils.MetaJSON, "127.0.0.1:5565", "127.0.0.1:2012", t1, t1.Add(time.Second)); err != nil {
t.Fatal(err)
}
if cnt, err := anz.db.DocCount(); err != nil {
t.Fatal(err)
} else if cnt != 2 {
t.Errorf("Expected only 2 documents received:%v", cnt)
}
if err = anz.clenaUp(); err != nil {
t.Fatal(err)
}
if cnt, err := anz.db.DocCount(); err != nil {
t.Fatal(err)
} else if cnt != 1 {
t.Errorf("Expected only one document received:%v", cnt)
}
if err = anz.db.Close(); err != nil {
t.Fatal(err)
}
if err = anz.clenaUp(); err != bleve.ErrorIndexClosed {
t.Errorf("Expected error: %v,received: %+v", bleve.ErrorIndexClosed, err)
}
}

3
config/configsanity.go
View File

@@ -1004,9 +1004,6 @@ func (cfg *CGRConfig) checkConfigSanity() error {
}
if cfg.analyzerSCfg.Enabled {
if _, err := os.Stat(cfg.analyzerSCfg.DBPath); err != nil && os.IsNotExist(err) {
return fmt.Errorf("<%s> nonexistent DB folder: %q", utils.AnalyzerS, cfg.analyzerSCfg.DBPath)
}
if !utils.AnzIndexType.Has(cfg.analyzerSCfg.IndexType) {
return fmt.Errorf("<%s> unsuported index type: %q", utils.AnalyzerS, cfg.analyzerSCfg.IndexType)
}

5
config/configsanity_test.go
View File

@@ -1567,11 +1567,6 @@ func TestConfigSanityAnalyzer(t *testing.T) {
t.Errorf("Expecting: %+q received: %+q", expected, err)
}
cfg.analyzerSCfg.DBPath = "/inexistent/Path"
expected = "<AnalyzerS> nonexistent DB folder: \"/inexistent/Path\""
if err := cfg.checkConfigSanity(); err == nil || err.Error() != expected {
t.Errorf("Expecting: %+q received: %+q", expected, err)
}
cfg.analyzerSCfg.DBPath = "/"
cfg.analyzerSCfg.IndexType = utils.MetaScorch

3
services/analyzers.go
View File

@@ -76,7 +76,6 @@ func (anz *AnalyzerService) Start(_ *context.Context, shtDwn context.CancelFunc)
utils.Logger.Crit(fmt.Sprintf("<%s> Error: %s listening for packets", utils.AnalyzerS, err.Error()))
shtDwn()
}
return
}(anz.anz)
anz.server.SetAnalyzer(anz.anz)
go anz.start()
@@ -115,7 +114,7 @@ func (anz *AnalyzerService) Shutdown() (err error) {
anz.server.SetAnalyzer(nil)
anz.anz.Shutdown()
anz.anz = nil
<-anz.connChan
// <-anz.connChan
anz.Unlock()
return
}

4
services/datadb.go
View File

@@ -141,8 +141,8 @@ func (db *DataDBService) ServiceName() string {
}
// ShouldRun returns if the service should be running
func (db *DataDBService) ShouldRun() bool {
return db.mandatoryDB() || db.cfg.SessionSCfg().Enabled
func (db *DataDBService) ShouldRun() bool { // db should allways run
return true // ||db.mandatoryDB() || db.cfg.SessionSCfg().Enabled
}
// mandatoryDB returns if the current configuration needs the DB

4
services/datadb_test.go
View File

@@ -131,7 +131,7 @@ func TestDataDBCoverage(t *testing.T) {
t.Errorf("\nExpecting <%+v>,\n Received <%+v>", utils.DataDB, serviceName)
}
shouldRun := db.ShouldRun()
if !reflect.DeepEqual(shouldRun, false) {
t.Errorf("\nExpecting <false>,\n Received <%+v>", shouldRun)
if !reflect.DeepEqual(shouldRun, true) {
t.Errorf("\nExpecting <true>,\n Received <%+v>", shouldRun)
}
}