uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-11 18:16:24 +05:00
Code Issues Packages Projects Releases Wiki Activity

Commented CGRID where it is not needed

Browse Source
This commit is contained in:
andronache98
2022-01-25 14:27:57 +02:00
committed by Dan Christian Bogos
parent ebfd371bb7
commit 2e4b77e4df
17 changed files with 124 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
agents/fsevent.go
View File

@@ -193,7 +193,7 @@ func (fsev FSEvent) MissingParameter(timezone string) string {
func (fsev FSEvent) GetSetupTime(fieldName, timezone string) (t time.Time, err error) {
fsSTimeStr, hasKey := fsev[SETUP_TIME]
if hasKey && fsSTimeStr != "0" {
// Discard the nanoseconds information since MySQL cannot store them in early versions and csv uses default seconds so CGRID will not corelate
// Discard the nanoseconds information since MySQL cannot store them in early versions and csv uses default seconds so originID will not corelate
fsSTimeStr = fsSTimeStr[:len(fsSTimeStr)-6]
}
sTimeStr := utils.FirstNonEmpty(fsev[fieldName], fsSTimeStr)
@@ -206,7 +206,7 @@ func (fsev FSEvent) GetSetupTime(fieldName, timezone string) (t time.Time, err e
func (fsev FSEvent) GetAnswerTime(fieldName, timezone string) (t time.Time, err error) {
fsATimeStr, hasKey := fsev[ANSWER_TIME]
if hasKey && fsATimeStr != "0" {
// Discard the nanoseconds information since MySQL cannot store them in early versions and csv uses default seconds so CGRID will not corelate
// Discard the nanoseconds information since MySQL cannot store them in early versions and csv uses default seconds so originID will not corelate
fsATimeStr = fsATimeStr[:len(fsATimeStr)-6]
}
aTimeStr := utils.FirstNonEmpty(fsev[fieldName], fsATimeStr)

2
agents/kamagent.go
View File

@@ -391,7 +391,7 @@ func (ka *KamailioAgent) V1DisconnectSession(ctx *context.Context, args utils.At
return
}
// V1GetActiveSessionIDs returns a list of CGRIDs based on active sessions from agent
// V1GetActiveSessionIDs returns a list of originIDs based on active sessions from agent
func (ka *KamailioAgent) V1GetActiveSessionIDs(ctx *context.Context, _ string, sessionIDs *[]*sessions.SessionID) (err error) {
kamEv := utils.ToJSON(map[string]string{utils.Event: CGR_DLG_LIST})
var sentDLG int

32
dispatchers/cdrs_it_test.go
View File

@@ -188,8 +188,8 @@ func testDspCDRsGetCDR(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
// t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
}
}
@@ -209,8 +209,8 @@ func testDspCDRsGetCDRWithoutTenant(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
// t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
}
}
@@ -258,8 +258,8 @@ func testDspCDRsGetCDR2(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "f08dfd32930b6bea326bb8ec4e38ab03d781c0bf" {
t.Errorf("Expected: f08dfd32930b6bea326bb8ec4e38ab03d781c0bf , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "f08dfd32930b6bea326bb8ec4e38ab03d781c0bf" {
// t.Errorf("Expected: f08dfd32930b6bea326bb8ec4e38ab03d781c0bf , received:%v", reply[0].CGRID)
}
}
@@ -311,8 +311,8 @@ func testDspCDRsGetCDR3(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "8ae63781b39f3265d014d2ba6a70437172fba46d" {
t.Errorf("Expected: 8ae63781b39f3265d014d2ba6a70437172fba46d , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "8ae63781b39f3265d014d2ba6a70437172fba46d" {
// t.Errorf("Expected: 8ae63781b39f3265d014d2ba6a70437172fba46d , received:%v", reply[0].CGRID)
}
}
@@ -428,8 +428,8 @@ func testDspCDRsGetCDRNoAuth(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
// t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
}
}
@@ -446,8 +446,8 @@ func testDspCDRsGetCDRNoAuthWithoutTenant(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "9ee4c71fcd67eef5fb25a4bb3f190487de3073f5" {
// t.Errorf("Expected: 9ee4c71fcd67eef5fb25a4bb3f190487de3073f5 , received:%v", reply[0].CGRID)
}
}
@@ -489,8 +489,8 @@ func testDspCDRsGetCDR2NoAuth(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "f08dfd32930b6bea326bb8ec4e38ab03d781c0bf" {
t.Errorf("Expected: f08dfd32930b6bea326bb8ec4e38ab03d781c0bf , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "f08dfd32930b6bea326bb8ec4e38ab03d781c0bf" {
// t.Errorf("Expected: f08dfd32930b6bea326bb8ec4e38ab03d781c0bf , received:%v", reply[0].CGRID)
}
}
@@ -536,7 +536,7 @@ func testDspCDRsGetCDR3NoAuth(t *testing.T) {
t.Error(err)
} else if len(reply) != 1 {
t.Errorf("Received: %+v", reply)
} else if reply[0].CGRID != "8ae63781b39f3265d014d2ba6a70437172fba46d" {
t.Errorf("Expected: 8ae63781b39f3265d014d2ba6a70437172fba46d , received:%v", reply[0].CGRID)
// } else if reply[0].CGRID != "8ae63781b39f3265d014d2ba6a70437172fba46d" {
// t.Errorf("Expected: 8ae63781b39f3265d014d2ba6a70437172fba46d , received:%v", reply[0].CGRID)
}
}

6
ees/ee.go
View File

@@ -139,13 +139,13 @@ func newEEMetrics(location string) (*utils.SafeMapStorage, error) {
}}, nil
}
func updateEEMetrics(dc *utils.SafeMapStorage, cgrID string, ev engine.MapEvent, hasError bool, timezone string) {
func updateEEMetrics(dc *utils.SafeMapStorage, originID string, ev engine.MapEvent, hasError bool, timezone string) {
dc.Lock()
defer dc.Unlock()
if hasError {
dc.MapStorage[utils.NegativeExports].(utils.StringSet).Add(cgrID)
dc.MapStorage[utils.NegativeExports].(utils.StringSet).Add(originID)
} else {
dc.MapStorage[utils.PositiveExports].(utils.StringSet).Add(cgrID)
dc.MapStorage[utils.PositiveExports].(utils.StringSet).Add(originID)
}
if aTime, err := ev.GetTime(utils.AnswerTime, timezone); err == nil {
if _, has := dc.MapStorage[utils.FirstEventATime]; !has {

1
engine/cdr.go
View File

@@ -312,7 +312,6 @@ func NewCDRFromSQL(cdrSQL *CDRsql) (cdr *CDR, err error) {
}
type ExternalCDR struct {
CGRID string
RunID string
OrderID int64
OriginHost string

6
engine/cdr_test.go
View File

@@ -29,7 +29,7 @@ import (
func TestNewCDRFromExternalCDR(t *testing.T) {
extCdr := &ExternalCDR{
CGRID: utils.Sha1("dsafdsaf", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
// CGRID: utils.Sha1("dsafdsaf", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
OrderID: 123, ToR: utils.MetaVoice, OriginID: "dsafdsaf", OriginHost: "192.168.1.1",
Source: utils.UnitTest, RequestType: utils.MetaRated,
Tenant: "cgrates.org", Category: "call", Account: "1001", Subject: "1001", Destination: "1002",
@@ -161,7 +161,7 @@ func TestCDRNewCDRFromSQL(t *testing.T) {
extraFields := map[string]string{"field_extr1": "val_extr1", "fieldextr2": "valextr2"}
cdrSQL := &CDRsql{
ID: 123,
Cgrid: "abecd993d06672714c4218a6dcf8278e0589a171",
// Cgrid: "abecd993d06672714c4218a6dcf8278e0589a171",
RunID: utils.MetaDefault,
OriginID: "dsafdsaf",
TOR: utils.MetaVoice,
@@ -297,7 +297,7 @@ func TestCDRAddDefaults(t *testing.T) {
func TestNewCDRFromExternalCDRSetupTimeError(t *testing.T) {
extCdr := &ExternalCDR{
CGRID: utils.Sha1("dsafdsaf", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
// CGRID: utils.Sha1("dsafdsaf", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
OrderID: 123, ToR: utils.MetaVoice, OriginID: "dsafdsaf", OriginHost: "192.168.1.1",
Source: utils.UnitTest, RequestType: utils.MetaRated,
Tenant: "cgrates.org", Category: "call", Account: "1001", Subject: "1001", Destination: "1002",

2
engine/fscdr.go
View File

@@ -69,7 +69,7 @@ type FSCdr struct {
body map[string]interface{} // keeps the loaded body for extra field search
}
func (fsCdr FSCdr) getCGRID() string {
func (fsCdr FSCdr) getOriginID() string {
return utils.Sha1(fsCdr.vars[fsUUID],
utils.FirstNonEmpty(fsCdr.vars[utils.CGROriginHost], fsCdr.vars[fsIPv4]))
}

4
engine/models.go
View File

@@ -106,7 +106,6 @@ func (FilterMdl) TableName() string {
type CDRsql struct {
ID int64
Cgrid string
RunID string
OriginHost string
Source string
@@ -138,7 +137,7 @@ func (t CDRsql) TableName() string {
func (t CDRsql) AsMapStringInterface() (out map[string]interface{}) {
out = make(map[string]interface{})
// out["id"] = t.ID // ignore ID
out["cgrid"] = t.Cgrid
out["run_id"] = t.RunID
out["origin_host"] = t.OriginHost
out["source"] = t.Source
@@ -167,7 +166,6 @@ func (t CDRsql) AsMapStringInterface() (out map[string]interface{}) {
type SessionCostsSQL struct {
ID int64
Cgrid string
RunID string
OriginHost string
OriginID string

6
engine/models_test.go
View File

@@ -28,8 +28,8 @@ import (
func TestModelsAsMapStringInterface(t *testing.T) {
testCdrSql := CDRsql{
ID: 1,
Cgrid: "testCgrID1",
ID: 1,
// Cgrid: "testCgrID1",
RunID: "testRunID",
OriginHost: "testOriginHost",
Source: "testSource",
@@ -54,7 +54,7 @@ func TestModelsAsMapStringInterface(t *testing.T) {
DeletedAt: utils.TimePointer(time.Date(2021, 3, 3, 3, 3, 3, 3, time.UTC)),
}
expected := map[string]interface{}{
"cgrid": testCdrSql.Cgrid,
// "cgrid": testCdrSql.Cgrid,
"run_id": testCdrSql.RunID,
"origin_host": testCdrSql.OriginHost,
"source": testCdrSql.Source,

43
ers/sql_it_test.go
View File

@@ -67,7 +67,6 @@ var (
testSQLStop2,
}
cdr = &engine.CDR{
CGRID: "CGRID",
RunID: "RunID",
}
db *gorm.DB
@@ -173,7 +172,7 @@ func testSQLInitDBs(t *testing.T) {
}
}
func testSQLInitDB(t *testing.T) {
cdr.CGRID = utils.UUIDSha1Prefix()
var err error
if db, err = gorm.Open(mysql.Open(fmt.Sprintf(dbConnString, "cgrates2")),
&gorm.Config{
@@ -244,11 +243,9 @@ func testSQLReader(t *testing.T) {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{
"CGRID": cdr.CGRID,
},
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{},
APIOpts: map[string]interface{}{},
}
if !reflect.DeepEqual(ev.cgrEvent, expected) {
@@ -277,11 +274,9 @@ func testSQLReader2(t *testing.T) {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{
"CGRID": cdr.CGRID,
},
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{},
APIOpts: map[string]interface{}{},
}
if !reflect.DeepEqual(ev.cgrEvent, expected) {
@@ -315,9 +310,6 @@ func testSQLPoster(t *testing.T) {
msg[colName] = columns[i]
}
db.Table("cdrs2").Delete(msg)
if cgrid := utils.IfaceAsString(msg["cgrid"]); cgrid != cdr.CGRID {
t.Errorf("Expected: %s ,receieved: %s", cgrid, cdr.CGRID)
}
}
}
@@ -415,7 +407,6 @@ func testSQLInitDBs2(t *testing.T) {
}
}
func testSQLInitDB2(t *testing.T) {
cdr.CGRID = utils.UUIDSha1Prefix()
var err error
if db, err = gorm.Open(mysql.Open(fmt.Sprintf(dbConnString, "cgrates2")),
&gorm.Config{
@@ -487,11 +478,9 @@ func testSQLReader3(t *testing.T) {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{
"CGRID": cdr.CGRID,
},
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{},
APIOpts: map[string]interface{}{},
}
if !reflect.DeepEqual(ev.cgrEvent, expected) {
@@ -520,11 +509,9 @@ func testSQLReader4(t *testing.T) {
t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID)
}
expected := &utils.CGREvent{
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{
"CGRID": cdr.CGRID,
},
Tenant: "cgrates.org",
ID: ev.cgrEvent.ID,
Event: map[string]interface{}{},
APIOpts: map[string]interface{}{},
}
if !reflect.DeepEqual(ev.cgrEvent, expected) {
@@ -558,9 +545,7 @@ func testSQLPoster2(t *testing.T) {
msg[colName] = columns[i]
}
db.Table("cdrs2").Delete(msg)
if cgrid := utils.IfaceAsString(msg["cgrid"]); cgrid != cdr.CGRID {
t.Errorf("Expected: %s ,receieved: %s", cgrid, cdr.CGRID)
}
}
}

6
general_tests/session_graceful_shutdown_it_test.go
View File

@@ -202,7 +202,7 @@ func testSessionSRplcApierGetInitateSessions(t *testing.T) {
func testSessionSRplcApierGetActiveSessions(t *testing.T) {
expected := []*sessions.ExternalSession{
{
CGRID: "testSessionRplCGRID",
//CGRID: "testSessionRplCGRID",
RunID: "*default",
ToR: "",
OriginID: "testSessionRplORIGINID",
@@ -241,7 +241,7 @@ func testSessionSRplcApierGetActiveSessions(t *testing.T) {
func testSessionSRplcApierGetPassiveSessions(t *testing.T) {
expected := []*sessions.ExternalSession{
{
CGRID: "testSessionRplCGRID",
//CGRID: "testSessionRplCGRID",
RunID: "",
ToR: "",
OriginID: "testSessionRplORIGINID",
@@ -291,7 +291,7 @@ func testSessionSRplcApierStopSession2(t *testing.T) {
func testSessionSRplcApierGetPassiveSessionsAfterStop(t *testing.T) {
expected := []*sessions.ExternalSession{
{
CGRID: "testSessionRplCGRID",
//CGRID: "testSessionRplCGRID",
RunID: "",
ToR: "",
OriginID: "testSessionRplORIGINID",

5
migrator/storage_sql.go
View File

@@ -60,12 +60,11 @@ func (mgSQL *migratorSQL) renameV1SMCosts() (err error) {
}
func (mgSQL *migratorSQL) createV1SMCosts() (err error) {
qry := fmt.Sprint("CREATE TABLE sm_costs ( id int(11) NOT NULL AUTO_INCREMENT, cgrid varchar(40) NOT NULL, run_id varchar(64) NOT NULL, origin_host varchar(64) NOT NULL, origin_id varchar(128) NOT NULL, cost_source varchar(64) NOT NULL, `usage` BIGINT NOT NULL, cost_details MEDIUMTEXT, created_at TIMESTAMP NULL,deleted_at TIMESTAMP NULL, PRIMARY KEY (`id`),UNIQUE KEY costid (cgrid, run_id),KEY origin_idx (origin_host, origin_id),KEY run_origin_idx (run_id, origin_id),KEY deleted_at_idx (deleted_at));")
qry := fmt.Sprint("CREATE TABLE sm_costs ( id int(11) NOT NULL AUTO_INCREMENT, run_id varchar(64) NOT NULL, origin_host varchar(64) NOT NULL, origin_id varchar(128) NOT NULL, cost_source varchar(64) NOT NULL, `usage` BIGINT NOT NULL, cost_details MEDIUMTEXT, created_at TIMESTAMP NULL,deleted_at TIMESTAMP NULL, PRIMARY KEY (`id`),UNIQUE KEY costid ( run_id),KEY origin_idx (origin_host, origin_id),KEY run_origin_idx (run_id, origin_id),KEY deleted_at_idx (deleted_at));")
if mgSQL.StorDB().GetStorageType() == utils.Postgres {
qry = `
CREATE TABLE sm_costs (
id SERIAL PRIMARY KEY,
cgrid VARCHAR(40) NOT NULL,
run_id VARCHAR(64) NOT NULL,
origin_host VARCHAR(64) NOT NULL,
origin_id VARCHAR(128) NOT NULL,
@@ -74,7 +73,7 @@ func (mgSQL *migratorSQL) createV1SMCosts() (err error) {
cost_details jsonb,
created_at TIMESTAMP WITH TIME ZONE,
deleted_at TIMESTAMP WITH TIME ZONE NULL,
UNIQUE (cgrid, run_id)
UNIQUE ( run_id)
);
`
}

11
sessions/libsessions.go
View File

@@ -51,15 +51,15 @@ type BiRPClient interface {
V1WarnDisconnect(ctx *context.Context, args map[string]interface{}, reply *string) (err error)
}
// GetSetOptsOriginID will populate the CGRID key if not present and return it
func GetSetOptsOriginID(ev engine.MapEvent, opt map[string]interface{}) (cgrID string) {
// GetSetOptsOriginID will populate the originID key if not present and return it
func GetSetOptsOriginID(ev engine.MapEvent, opt map[string]interface{}) (originID string) {
fldIface, has := opt[utils.MetaOriginID]
if has {
cgrID = utils.IfaceAsString(fldIface)
originID = utils.IfaceAsString(fldIface)
} else {
cgrID = utils.Sha1(ev.GetStringIgnoreErrors(utils.OriginID),
originID = utils.Sha1(ev.GetStringIgnoreErrors(utils.OriginID),
ev.GetStringIgnoreErrors(utils.OriginHost))
opt[utils.MetaOriginID] = cgrID
opt[utils.MetaOriginID] = originID
}
return
}
@@ -608,7 +608,6 @@ func (v1Rply *V1UpdateSessionReply) AsNavigableMap() map[string]*utils.DataNode
// ArgsReplicateSessions used to specify wich Session to replicate over the given connections
type ArgsReplicateSessions struct {
CGRID string
Passive bool
ConnIDs []string
APIOpts map[string]interface{}

6
sessions/session.go
View File

@@ -40,7 +40,7 @@ func (s *SessionID) OptsOriginID() string {
// ExternalSession is used when displaying active sessions via RPC
type ExternalSession struct {
CGRID string
//CGRID string
RunID string
ToR string // type of record, meta-field, should map to one of the TORs hardcoded inside the server <*voice|*data|*sms|*generic>
OriginID string // represents the unique accounting id given by the telecom switch generating the CDR
@@ -134,7 +134,7 @@ func (s *Session) AsExternalSessions(tmz, nodeID string) (aSs []*ExternalSession
aSs = make([]*ExternalSession, len(s.SRuns))
for i, sr := range s.SRuns {
aSs[i] = &ExternalSession{
CGRID: utils.IfaceAsString(s.OptsStart[utils.MetaOriginID]),
//CGRID: utils.IfaceAsString(s.OptsStart[utils.MetaOriginID]),
RunID: sr.RunID,
ToR: sr.Event.GetStringIgnoreErrors(utils.ToR),
OriginID: s.EventStart.GetStringIgnoreErrors(utils.OriginID),
@@ -164,7 +164,7 @@ func (s *Session) AsExternalSessions(tmz, nodeID string) (aSs []*ExternalSession
// AsExternalSession returns the session as an ExternalSession using the SRuns given
func (s *Session) AsExternalSession(sr *SRun, tmz, nodeID string) (aS *ExternalSession) {
aS = &ExternalSession{
CGRID: utils.IfaceAsString(s.OptsStart[utils.MetaOriginID]),
//CGRID: utils.IfaceAsString(s.OptsStart[utils.MetaOriginID]),
RunID: sr.RunID,
ToR: sr.Event.GetStringIgnoreErrors(utils.ToR),
OriginID: s.EventStart.GetStringIgnoreErrors(utils.OriginID),

2
sessions/session_test.go
View File

@@ -282,7 +282,7 @@ func TestSessionAsExternalSessions(t *testing.T) {
}},
}
exp := []*ExternalSession{{
CGRID: "RandomCGRID",
//CGRID: "RandomCGRID",
ToR: utils.MetaVoice,
OriginID: "123451",
// OriginHost: s.EventStart.GetStringIgnoreErrors(utils.OriginHost),

126
sessions/sessions.go
View File

@@ -83,14 +83,14 @@ type SessionS struct {
aSessions map[string]*Session // group sessions per sessionId
aSIMux sync.RWMutex // protects aSessionsIdx
aSessionsIdx map[string]map[string]map[string]utils.StringSet // map[fieldName]map[fieldValue][cgrID]utils.StringSet[runID]sID
aSessionsIdx map[string]map[string]map[string]utils.StringSet // map[fieldName]map[fieldValue][originID]utils.StringSet[runID]sID
aSessionsRIdx map[string][]*riFieldNameVal // reverse indexes for active sessions, used on remove
pSsMux sync.RWMutex // protects pSessions
pSessions map[string]*Session // group passive sessions based on cgrID
pSessions map[string]*Session // group passive sessions based on originID
pSIMux sync.RWMutex // protects pSessionsIdx
pSessionsIdx map[string]map[string]map[string]utils.StringSet // map[fieldName]map[fieldValue][cgrID]utils.StringSet[runID]sID
pSessionsIdx map[string]map[string]map[string]utils.StringSet // map[fieldName]map[fieldValue][originID]utils.StringSet[runID]sID
pSessionsRIdx map[string][]*riFieldNameVal // reverse indexes for passive sessions, used on remove
}
@@ -324,7 +324,7 @@ func (sS *SessionS) forceSTerminate(ctx *context.Context, s *Session, extraUsage
if _, err = sS.debitSession(ctx, s, i, extraUsage, lastUsed); err != nil {
utils.Logger.Warning(
fmt.Sprintf(
"<%s> failed debitting cgrID %s, sRunIdx: %d, err: %s",
"<%s> failed debitting originID %s, sRunIdx: %d, err: %s",
utils.SessionS, s.originID(), i, err.Error()))
}
}
@@ -810,7 +810,7 @@ func (sS *SessionS) filterSessions(ctx *context.Context, sf *utils.SessionFilter
if len(indx) != 0 && len(originIDs) == 0 { // no sessions matched the indexed filters
return
}
ss := sS.getSessionsFromCGRIDs(psv, originIDs...)
ss := sS.getSessionsFromOriginIDs(psv, originIDs...)
pass := func(filterRules []*engine.FilterRule,
me engine.MapEvent) (pass bool) {
pass = true
@@ -862,11 +862,11 @@ func (sS *SessionS) filterSessionsCount(ctx *context.Context, sf *utils.SessionF
}
tenant := utils.FirstNonEmpty(sf.Tenant, sS.cfg.GeneralCfg().DefaultTenant)
indx, unindx := sS.getIndexedFilters(ctx, tenant, sf.Filters)
cgrIDs, _ /* matchingSRuns*/ := sS.getSessionIDsMatchingIndexes(indx, psv)
if len(indx) != 0 && len(cgrIDs) == 0 { // no sessions matched the indexed filters
originIDs, _ /* matchingSRuns*/ := sS.getSessionIDsMatchingIndexes(indx, psv)
if len(indx) != 0 && len(originIDs) == 0 { // no sessions matched the indexed filters
return
}
ss := sS.getSessionsFromCGRIDs(psv, cgrIDs...)
ss := sS.getSessionsFromOriginIDs(psv, originIDs...)
pass := func(filterRules []*engine.FilterRule,
me engine.MapEvent) (pass bool) {
pass = true
@@ -885,9 +885,9 @@ func (sS *SessionS) filterSessionsCount(ctx *context.Context, sf *utils.SessionF
}
for _, s := range ss {
s.RLock()
// runIDs := matchingSRuns[s.CGRID]
// runIDs := matchingSRuns[s.OptsStart[utils.MetaOriginID]]
for _, sr := range s.SRuns {
// if len(cgrIDs) != 0 && !runIDs.Has(sr.CD.RunID) {
// if len(originIDs) != 0 && !runIDs.Has(sr.CD.RunID) {
// continue
// }
if pass(unindx, sr.Event) {
@@ -988,7 +988,7 @@ func (sS *SessionS) getSessions(originID string, pSessions bool) (ss []*Session)
}
// getSessions is used to return in a thread-safe manner active or passive sessions
func (sS *SessionS) getSessionsFromCGRIDs(pSessions bool, originIDs ...string) (ss []*Session) {
func (sS *SessionS) getSessionsFromOriginIDs(pSessions bool, originIDs ...string) (ss []*Session) {
ssMux := &sS.aSsMux // get the pointer so we don't copy, otherwise locks will not work
ssMp := sS.aSessions // reference it so we don't overwrite the new map without protection
if pSessions {
@@ -1106,7 +1106,7 @@ func (sS *SessionS) syncSessions(ctx *context.Context) {
replys <- &reply
}(clnt)
}
queriedCGRIDs := utils.StringSet{}
queriedOriginIDs := utils.StringSet{}
for range biClnts {
reply := <-replys
if reply.err != nil {
@@ -1117,14 +1117,14 @@ func (sS *SessionS) syncSessions(ctx *context.Context) {
continue
}
for _, sessionID := range reply.reply {
queriedCGRIDs.Add(sessionID.OptsOriginID())
queriedOriginIDs.Add(sessionID.OptsOriginID())
}
}
var toBeRemoved []string
sS.aSsMux.RLock()
for cgrid := range sS.aSessions {
if !queriedCGRIDs.Has(cgrid) {
toBeRemoved = append(toBeRemoved, cgrid)
for optOriginID := range sS.aSessions {
if !queriedOriginIDs.Has(optOriginID) {
toBeRemoved = append(toBeRemoved, optOriginID)
}
}
sS.aSsMux.RUnlock()
@@ -1133,8 +1133,8 @@ func (sS *SessionS) syncSessions(ctx *context.Context) {
// Extracted from syncSessions in order to test all cases
func (sS *SessionS) terminateSyncSessions(ctx *context.Context, toBeRemoved []string) {
for _, cgrID := range toBeRemoved {
ss := sS.getSessions(cgrID, false)
for _, optOriginID := range toBeRemoved {
ss := sS.getSessions(optOriginID, false)
if len(ss) == 0 {
continue
}
@@ -1142,7 +1142,7 @@ func (sS *SessionS) terminateSyncSessions(ctx *context.Context, toBeRemoved []st
if err := sS.forceSTerminate(ctx, ss[0], 0, nil, nil); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> failed force-terminating session: <%s>, err: <%s>",
utils.SessionS, cgrID, err.Error()))
utils.SessionS, optOriginID, err.Error()))
}
ss[0].Unlock()
}
@@ -1330,7 +1330,7 @@ func (sS *SessionS) endSession(ctx *context.Context, s *Session, tUsage, lastUsa
// APIOpts: s.OptsStart,
// }, cc); err == nil {
// sr.EventCost.Merge(
// engine.NewEventCostFromCallCost(cc, s.CGRID,
// engine.NewEventCostFromCallCost(cc, s.OptsStart[utils.MetaOriginID],
// sr.Event.GetStringIgnoreErrors(utils.RunID)))
// }
// }
@@ -1339,13 +1339,13 @@ func (sS *SessionS) endSession(ctx *context.Context, s *Session, tUsage, lastUsa
// utils.Logger.Warning(
// fmt.Sprintf(
// "<%s> failed refunding session: <%s>, srIdx: <%d>, error: <%s>",
// utils.SessionS, s.CGRID, sRunIdx, err.Error()))
// utils.SessionS, s.OptsStart[utils.MetaOriginID], sRunIdx, err.Error()))
// }
// }
// if err := sS.roundCost(s, sRunIdx); err != nil { // will round the cost and refund the extra increment
// utils.Logger.Warning(
// fmt.Sprintf("<%s> failed rounding session cost for <%s>, srIdx: <%d>, error: <%s>",
// utils.SessionS, s.CGRID, sRunIdx, err.Error()))
// utils.SessionS, s.OptsStart[utils.MetaOriginID], sRunIdx, err.Error()))
// }
// }
// // compute the event cost before saving the SessionCost
@@ -1355,7 +1355,7 @@ func (sS *SessionS) endSession(ctx *context.Context, s *Session, tUsage, lastUsa
// if err := sS.storeSCost(s, sRunIdx); err != nil {
// utils.Logger.Warning(
// fmt.Sprintf("<%s> failed storing session cost for <%s>, srIdx: <%d>, error: <%s>",
// utils.SessionS, s.CGRID, sRunIdx, err.Error()))
// utils.SessionS, s.OptsStart[utils.MetaOriginID], sRunIdx, err.Error()))
// }
// }
@@ -1513,10 +1513,10 @@ func (sS *SessionS) BiRPCv1SetPassiveSession(ctx *context.Context,
}
// BiRPCv1ReplicateSessions will replicate active sessions to either args.Connections or the internal configured ones
// args.Filter is used to filter the sessions which are replicated, CGRID is the only one possible for now
// args.Filter is used to filter the sessions which are replicated, originID is the only one possible for now
func (sS *SessionS) BiRPCv1ReplicateSessions(ctx *context.Context,
args ArgsReplicateSessions, reply *string) (err error) {
sS.replicateSessions(ctx, args.CGRID, args.Passive, args.ConnIDs)
sS.replicateSessions(ctx, utils.IfaceAsString(args.APIOpts[utils.MetaOriginID]), args.Passive, args.ConnIDs)
*reply = utils.OK
return
}
@@ -2002,8 +2002,8 @@ func (sS *SessionS) BiRPCv1UpdateSession(ctx *context.Context,
config.SessionsDebitIntervalDftOpt, utils.OptsSesDebitInterval); err != nil {
return err
}
cgrID := GetSetOptsOriginID(ev, args.APIOpts)
s := sS.getRelocateSession(ctx, cgrID,
optOriginID := GetSetOptsOriginID(ev, args.APIOpts)
s := sS.getRelocateSession(ctx, optOriginID,
ev.GetStringIgnoreErrors(utils.InitialOriginID),
ev.GetStringIgnoreErrors(utils.OriginID),
ev.GetStringIgnoreErrors(utils.OriginHost))
@@ -2086,7 +2086,7 @@ func (sS *SessionS) BiRPCv1TerminateSession(ctx *context.Context,
ev := engine.MapEvent(args.Event)
opts := engine.MapEvent(args.APIOpts)
cgrID := GetSetOptsOriginID(ev, opts)
optOriginID := GetSetOptsOriginID(ev, opts)
originID := ev.GetStringIgnoreErrors(utils.OriginID)
if termS {
if originID == "" {
@@ -2106,7 +2106,7 @@ func (sS *SessionS) BiRPCv1TerminateSession(ctx *context.Context,
return
}
for i := 0; i < sS.cfg.SessionSCfg().TerminateAttempts; i++ {
if s = sS.getRelocateSession(ctx, cgrID,
if s = sS.getRelocateSession(ctx, optOriginID,
ev.GetStringIgnoreErrors(utils.InitialOriginID),
ev.GetStringIgnoreErrors(utils.OriginID),
ev.GetStringIgnoreErrors(utils.OriginHost)); s != nil {
@@ -2677,7 +2677,7 @@ func (sS *SessionS) BiRPCv1ProcessEvent(ctx *context.Context,
// }
// cd := &engine.CallDescriptor{
// CgrID: cgrEv.ID,
// //CgrID: cgrEv.ID,
// RunID: ev.GetStringIgnoreErrors(utils.RunID),
// ToR: ev.GetStringIgnoreErrors(utils.ToR),
// Tenant: cgrEv.Tenant,
@@ -2844,21 +2844,21 @@ func (sS *SessionS) BiRPCv1ForceDisconnect(ctx *context.Context,
if len(aSs) == 0 {
return utils.ErrNotFound
}
for _, as := range aSs {
ss := sS.getSessions(as.CGRID, false)
if len(ss) == 0 {
continue
}
ss[0].Lock()
if errTerm := sS.forceSTerminate(ctx, ss[0], 0, nil, nil); errTerm != nil {
utils.Logger.Warning(
fmt.Sprintf(
"<%s> failed force-terminating session with id: <%s>, err: <%s>",
utils.SessionS, ss[0].originID(), errTerm.Error()))
err = utils.ErrPartiallyExecuted
}
ss[0].Unlock()
}
// for _, as := range aSs {
// ss := sS.getSessions(as.CGRID, false)
// if len(ss) == 0 {
// continue
// }
// ss[0].Lock()
// if errTerm := sS.forceSTerminate(ctx, ss[0], 0, nil, nil); errTerm != nil {
// utils.Logger.Warning(
// fmt.Sprintf(
// "<%s> failed force-terminating session with id: <%s>, err: <%s>",
// utils.SessionS, ss[0].originID(), errTerm.Error()))
// err = utils.ErrPartiallyExecuted
// }
// ss[0].Unlock()
// }
if err == nil {
*reply = utils.OK
} else {
@@ -3058,24 +3058,24 @@ func (sS *SessionS) BiRPCv1ReAuthorize(ctx *context.Context,
if len(aSs) == 0 {
return utils.ErrNotFound
}
cache := utils.NewStringSet(nil)
for _, as := range aSs {
if cache.Has(as.CGRID) {
continue
}
cache.Add(as.CGRID)
ss := sS.getSessions(as.CGRID, false)
if len(ss) == 0 {
continue
}
if errTerm := sS.sendRar(ctx, ss[0]); errTerm != nil {
utils.Logger.Warning(
fmt.Sprintf(
"<%s> failed sending RAR for session with id: <%s>, err: <%s>",
utils.SessionS, ss[0].originID(), errTerm.Error()))
err = utils.ErrPartiallyExecuted
}
}
// cache := utils.NewStringSet(nil)
// for _, as := range aSs {
// if cache.Has(as.CGRID) {
// continue
// }
// cache.Add(as.CGRID)
// ss := sS.getSessions(as.CGRID, false)
// if len(ss) == 0 {
// continue
// }
// if errTerm := sS.sendRar(ctx, ss[0]); errTerm != nil {
// utils.Logger.Warning(
// fmt.Sprintf(
// "<%s> failed sending RAR for session with id: <%s>, err: <%s>",
// utils.SessionS, ss[0].originID(), errTerm.Error()))
// err = utils.ErrPartiallyExecuted
// }
// }
if err != nil {
return
}

6
utils/apitpdata.go
View File

@@ -32,7 +32,7 @@ type ArgsItemIDs struct {
}
type AttrGetCdrs struct {
CgrIds []string // If provided, it will filter based on the cgrids present in list
//CgrIds []string // If provided, it will filter based on the cgrids present in list
MediationRunIds []string // If provided, it will filter on mediation runid
TORs []string // If provided, filter on TypeOfRecord
@@ -88,8 +88,8 @@ type ExportedTPStats struct {
// RPCCDRsFilter is a filter used in Rpc calls
// RPCCDRsFilter is slightly different than CDRsFilter by using string instead of Time filters
type RPCCDRsFilter struct {
CGRIDs []string // If provided, it will filter based on the cgrids present in list
NotCGRIDs []string // Filter specific CgrIds out
//CGRIDs []string // If provided, it will filter based on the cgrids present in list
//NotCGRIDs []string // Filter specific CgrIds out
RunIDs []string // If provided, it will filter on mediation runid
NotRunIDs []string // Filter specific runIds out
OriginIDs []string // If provided, it will filter on OriginIDs