From 5ec7c6ec1f5cd766b60a030d16c47a372560dd92 Mon Sep 17 00:00:00 2001 From: Trial97 Date: Mon, 16 Dec 2019 10:37:47 +0200 Subject: [PATCH 1/3] Updated gob integration tests --- data/conf/samples/actions_gob/cgradmin.json | 2 +- .../dispatchers/dispatchers_gob/cgrates.json | 73 ++++++++++++++++ .../dispatchers_mongo_gob/cgrates.json | 84 +++++++++++++++++++ data/conf/samples/smg_gob/cgrates.json | 7 ++ .../dispatchers_gob/Attributes.csv | 2 +- dispatchers/sessions_it_test.go | 4 +- engine/connmanager.go | 2 +- general_tests/broadcast_client_it_test.go | 2 +- general_tests/cdrs_processevent_it_test.go | 1 - 9 files changed, 170 insertions(+), 7 deletions(-) create mode 100755 data/conf/samples/dispatchers/dispatchers_gob/cgrates.json create mode 100644 data/conf/samples/dispatchers/dispatchers_mongo_gob/cgrates.json diff --git a/data/conf/samples/actions_gob/cgradmin.json b/data/conf/samples/actions_gob/cgradmin.json index 9226ad587..0a33ef910 100644 --- a/data/conf/samples/actions_gob/cgradmin.json +++ b/data/conf/samples/actions_gob/cgradmin.json @@ -38,7 +38,7 @@ "cdrs": { "enabled": true, // start the CDR Server service: - "chargers_conns":["*conn1"], + "chargers_conns":["conn1"], }, "chargers": { diff --git a/data/conf/samples/dispatchers/dispatchers_gob/cgrates.json b/data/conf/samples/dispatchers/dispatchers_gob/cgrates.json new file mode 100755 index 000000000..b8e1c7de1 --- /dev/null +++ b/data/conf/samples/dispatchers/dispatchers_gob/cgrates.json @@ -0,0 +1,73 @@ +{ + +// Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments +// Copyright (C) ITsysCOM GmbH +// +// This file contains the default configuration hardcoded into CGRateS. +// This is what you get when you load CGRateS with an empty configuration file. + +"general": { + "node_id": "DispatcherS1", + "log_level": 7, + "reconnects": 1, +}, + + +"listen": { + "rpc_json": ":2012", + "rpc_gob": ":2013", + "http": ":2080", +}, + +"stor_db": { + "db_type":"*internal", +}, + +"cache":{ + "*dispatcher_routes": {"limit": -1, "ttl": "2s"} +}, + + +"attributes": { + "enabled": true +}, + +"scheduler": { + "enabled": true, +}, + +"rals": { + "enabled": true, +}, + +"chargers": { + "enabled": true, +}, + +"rpc_conns": { + "conn1": { + "strategy": "*first", + "conns": [{"address": "127.0.0.1:2013", "transport":"*gob"}], + }, +}, + +"sessions": { + "enabled": true, + "attributes_conns": ["conn1"], + "rals_conns": ["conn1"], + "resources_conns": ["conn1"], + "chargers_conns": ["conn1"], + "listen_bijson": ":3014", +}, + +"dispatchers":{ + "enabled": true, + "attributes_conns": [ + {"address": "*internal"}, + ], +}, + +"apier": { + "scheduler_conns": ["*internal"], +}, +} \ No newline at end of file diff --git a/data/conf/samples/dispatchers/dispatchers_mongo_gob/cgrates.json b/data/conf/samples/dispatchers/dispatchers_mongo_gob/cgrates.json new file mode 100644 index 000000000..2a28cb711 --- /dev/null +++ b/data/conf/samples/dispatchers/dispatchers_mongo_gob/cgrates.json @@ -0,0 +1,84 @@ +{ + +// Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments +// Copyright (C) ITsysCOM GmbH +// +// This file contains the default configuration hardcoded into CGRateS. +// This is what you get when you load CGRateS with an empty configuration file. + +"general": { + "node_id": "DispatcherS1", + "log_level": 7, + "reconnects": 1, +}, + + +"listen": { + "rpc_json": ":2012", + "rpc_gob": ":2013", + "http": ":2080", +}, + +"data_db": { + "db_type": "mongo", + "db_name": "10", + "db_port": 27017, +}, + + +"stor_db": { + "db_type": "mongo", + "db_name": "cgrates", + "db_port": 27017, +}, + +"cache":{ + "*dispatcher_routes": {"limit": -1, "ttl": "2s"} +}, + +"scheduler": { + "enabled": true, +}, + +"attributes": { + "enabled": true +}, + +"rals": { + "enabled": true, +}, + + +"chargers": { + "enabled": true, +}, + +"rpc_conns": { + "conn1": { + "strategy": "*first", + "conns": [{"address": "127.0.0.1:2013", "transport":"*gob"}], + }, +}, + +"sessions": { + "enabled": true, + "attributes_conns": ["conn1"], + "rals_conns": ["conn1"], + "resources_conns": ["conn1"], + "chargers_conns": ["conn1"], + "listen_bijson": ":3014", +}, + + +"dispatchers":{ + "enabled": true, + "attributes_conns": [ + {"address": "*internal"}, + ], +}, + + +"apier": { + "scheduler_conns": ["*internal"], +}, +} \ No newline at end of file diff --git a/data/conf/samples/smg_gob/cgrates.json b/data/conf/samples/smg_gob/cgrates.json index 0162b9956..c87afe23e 100644 --- a/data/conf/samples/smg_gob/cgrates.json +++ b/data/conf/samples/smg_gob/cgrates.json @@ -14,6 +14,13 @@ "http": ":2080", // HTTP listening address }, +"rpc_conns": { + "conn1": { + "strategy": "*first", + "conns": [{"address": "127.0.0.1:2013", "transport":"*gob"}], + }, +}, + "stor_db": { // database used to store offline tariff plans and CDRs "db_password": "CGRateS.org", // password to use when connecting to stordb }, diff --git a/data/tariffplans/dispatchers_gob/Attributes.csv b/data/tariffplans/dispatchers_gob/Attributes.csv index e73eeaa13..3f4e362d4 100644 --- a/data/tariffplans/dispatchers_gob/Attributes.csv +++ b/data/tariffplans/dispatchers_gob/Attributes.csv @@ -14,7 +14,7 @@ cgrates.org,ATTR_API_RSP_AUTH,*auth,*string:~*req.APIKey:rsp12345,,,APIMethods,* cgrates.org,ATTR_API_CHC_AUTH,*auth,*string:~*req.APIKey:chc12345,,,APIMethods,*constant,CacheSv1.Ping&CacheSv1.GetCacheStats&CacheSv1.LoadCache&CacheSv1.PrecacheStatus&CacheSv1.GetItemIDs&CacheSv1.HasItem&CacheSv1.GetItemExpiryTime&CacheSv1.ReloadCache&CacheSv1.RemoveItem&CacheSv1.FlushCache&CacheSv1.Clear,false,20 cgrates.org,ATTR_API_GRD_AUTH,*auth,*string:~*req.APIKey:grd12345,,,APIMethods,*constant,GuardianSv1.Ping&GuardianSv1.RemoteLock&GuardianSv1.RemoteUnlock,false,20 cgrates.org,ATTR_API_SCHD_AUTH,*auth,*string:~*req.APIKey:sched12345,,,APIMethods,*constant,SchedulerSv1.Ping,false,20 -cgrates.org,ATTR_API_CDRS_AUTH,*auth,*string:~*req.APIKey:cdrs12345,,,APIMethods,*constant,CDRsV1.Ping&CDRsV1.ProcessEvent&CDRsV1.GetCDRs&CDRsV1.CountCDRs&CDRsV1.ProcessCDR&CDRsV1.ProcessExternalCDR,false,20 +cgrates.org,ATTR_API_CDRS_AUTH,*auth,*string:~*req.APIKey:cdrs12345,,,APIMethods,*constant,CDRsV1.Ping&CDRsV1.ProcessEvent&CDRsV1.GetCDRs&CDRsV1.GetCDRsCount&CDRsV1.ProcessCDR&CDRsV1.ProcessExternalCDR,false,20 cgrates.org,ATTR_API_DSP_AUTH,*auth,*string:~*req.APIKey:dsp12345,,,APIMethods,*constant,DispatcherSv1.Ping&DispatcherSv1.GetProfileForEvent,false,20 cgrates.org,ATTR_API_PSE_AUTH,*auth,*string:~*req.APIKey:pse12345,,,APIMethods,*constant,SessionSv1.Ping&SessionSv1.AuthorizeEvent&SessionSv1.AuthorizeEventWithDigest&SessionSv1.InitiateSession&SessionSv1.InitiateSessionWithDigest&SessionSv1.UpdateSession&SessionSv1.SyncSessions&SessionSv1.TerminateSession&SessionSv1.ProcessCDR&SessionSv1.ProcessMessage&SessionSv1.GetActiveSessions&SessionSv1.GetActiveSessionsCount&SessionSv1.ForceDisconnect&SessionSv1.GetPassiveSessions&SessionSv1.GetPassiveSessionsCount&SessionSv1.ReplicateSessions&SessionSv1.SetPassiveSession&AttributeSv1.ProcessEvent&Responder.Debit&ResourceSv1.AllocateResources&ChargerSv1.ProcessEvent&Responder.MaxDebit,false,20 cgrates.org,ATTR_API_CFG_AUTH,*auth,*string:~*req.APIKey:cfg12345,,,APIMethods,*constant,ConfigSv1.GetJSONSection&ConfigSv1.ReloadConfig,false,20 diff --git a/dispatchers/sessions_it_test.go b/dispatchers/sessions_it_test.go index 63353c017..2cf8f7912 100755 --- a/dispatchers/sessions_it_test.go +++ b/dispatchers/sessions_it_test.go @@ -60,7 +60,7 @@ var sTestsDspSession = []func(t *testing.T){ //Test start here func TestDspSessionSTMySQL(t *testing.T) { if *encoding == utils.MetaGOB { - testDsp(t, sTestsDspSession, "TestDspSessionS", "all", "all2", "dispatchers", "testit", "tutorial", "dispatchers_gob") + testDsp(t, sTestsDspSession, "TestDspSessionS", "all", "all2", "dispatchers_gob", "testit", "tutorial", "dispatchers_gob") } else { testDsp(t, sTestsDspSession, "TestDspSessionS", "all", "all2", "dispatchers", "testit", "tutorial", "dispatchers") } @@ -68,7 +68,7 @@ func TestDspSessionSTMySQL(t *testing.T) { func TestDspSessionSMongo(t *testing.T) { if *encoding == utils.MetaGOB { - testDsp(t, sTestsDspSession, "TestDspSessionS", "all", "all2", "dispatchers_mongo", "testit", "tutorial", "dispatchers_gob") + testDsp(t, sTestsDspSession, "TestDspSessionS", "all", "all2", "dispatchers_mongo_gob", "testit", "tutorial", "dispatchers_gob") } else { testDsp(t, sTestsDspSession, "TestDspSessionS", "all", "all2", "dispatchers_mongo", "testit", "tutorial", "dispatchers") } diff --git a/engine/connmanager.go b/engine/connmanager.go index c64cef8b8..bd04eb5f8 100644 --- a/engine/connmanager.go +++ b/engine/connmanager.go @@ -92,7 +92,7 @@ func (cM *ConnManager) getConn(connID string, biRPCClient rpcclient.ClientConnec func (cM *ConnManager) Call(connIDs []string, biRPCClient rpcclient.ClientConnector, method string, arg, reply interface{}) (err error) { if len(connIDs) == 0 { - return utils.ErrMandatoryIeMissing + return utils.NewErrMandatoryIeMissing("connIDs") } var conn rpcclient.ClientConnector for _, connID := range connIDs { diff --git a/general_tests/broadcast_client_it_test.go b/general_tests/broadcast_client_it_test.go index 004bd0a23..a3c2c2113 100644 --- a/general_tests/broadcast_client_it_test.go +++ b/general_tests/broadcast_client_it_test.go @@ -68,7 +68,7 @@ func testbrodcastItLoadConfig(t *testing.T) { t.Error(err) } brodcastInternalCfgPath = path.Join(*dataDir, "conf", "samples", "tutinternal") - if brodcastInternalCfg, err = config.NewCGRConfigFromPath(brodcastCfgPath); err != nil { + if brodcastInternalCfg, err = config.NewCGRConfigFromPath(brodcastInternalCfgPath); err != nil { t.Error(err) } } diff --git a/general_tests/cdrs_processevent_it_test.go b/general_tests/cdrs_processevent_it_test.go index 4c6177589..7143cb2f5 100644 --- a/general_tests/cdrs_processevent_it_test.go +++ b/general_tests/cdrs_processevent_it_test.go @@ -223,7 +223,6 @@ func testV1CDRsProcessEventAttrS(t *testing.T) { } func testV1CDRsProcessEventChrgS(t *testing.T) { - argsEv := &engine.ArgV1ProcessEvent{ Flags: []string{utils.MetaChargers, "*attributes:false"}, CGREvent: utils.CGREvent{ From 7c8e3810b2dacf4d1cc8d8d76e6a5f9c79ae4d28 Mon Sep 17 00:00:00 2001 From: Trial97 Date: Mon, 16 Dec 2019 18:08:53 +0200 Subject: [PATCH 2/3] Added *sql Event Reader --- ers/kafka_it_test.go | 3 +- ers/reader.go | 2 + ers/sql.go | 309 +++++++++++++++++++++++++++++++++++++++++++ ers/sql_it_test.go | 196 +++++++++++++++++++++++++++ utils/consts.go | 3 + 5 files changed, 512 insertions(+), 1 deletion(-) create mode 100644 ers/sql.go create mode 100644 ers/sql_it_test.go diff --git a/ers/kafka_it_test.go b/ers/kafka_it_test.go index 565f4e51c..341a01cf9 100644 --- a/ers/kafka_it_test.go +++ b/ers/kafka_it_test.go @@ -76,7 +76,6 @@ func TestKafkaER(t *testing.T) { rdrErr, new(engine.FilterS), rdrExit); err != nil { t.Fatal(err) } - kfk.Serve() w := kafka.NewWriter(kafka.WriterConfig{ Brokers: []string{"localhost:9092"}, Topic: defaultTopic, @@ -90,6 +89,8 @@ func TestKafkaER(t *testing.T) { ) w.Close() + kfk.Serve() + select { case err = <-rdrErr: t.Error(err) diff --git a/ers/reader.go b/ers/reader.go index bc520e2d3..97800312d 100644 --- a/ers/reader.go +++ b/ers/reader.go @@ -42,6 +42,8 @@ func NewEventReader(cfg *config.CGRConfig, cfgIdx int, return NewCSVFileER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit) case utils.MetaKafkajsonMap: return NewKafkaER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit) + case utils.MetaSQL: + return NewSQLEventReader(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit) } return } diff --git a/ers/sql.go b/ers/sql.go new file mode 100644 index 000000000..5e479dd6e --- /dev/null +++ b/ers/sql.go @@ -0,0 +1,309 @@ +/* +Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments +Copyright (C) ITsysCOM GmbH + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see +*/ + +package ers + +import ( + "fmt" + "net/url" + "strings" + "time" + + "github.com/cgrates/cgrates/agents" + "github.com/cgrates/cgrates/config" + "github.com/cgrates/cgrates/engine" + "github.com/cgrates/cgrates/utils" + + // libs for sql DBs + _ "github.com/go-sql-driver/mysql" + "github.com/jinzhu/gorm" + _ "github.com/lib/pq" +) + +const ( + dbName = "db_name" + tableName = "db_name" + sslMode = "sslmode" + defaultSSLMode = "disable" + defaultDBName = "cgrates" +) + +// NewSQLEventReader return a new kafka event reader +func NewSQLEventReader(cfg *config.CGRConfig, cfgIdx int, + rdrEvents chan *erEvent, rdrErr chan error, + fltrS *engine.FilterS, rdrExit chan struct{}) (er EventReader, err error) { + + rdr := &SQLEventReader{ + cgrCfg: cfg, + cfgIdx: cfgIdx, + fltrS: fltrS, + rdrEvents: rdrEvents, + rdrExit: rdrExit, + rdrErr: rdrErr, + } + if concReq := rdr.Config().ConcurrentReqs; concReq != -1 { + rdr.cap = make(chan struct{}, concReq) + for i := 0; i < concReq; i++ { + rdr.cap <- struct{}{} + } + } + if err = rdr.setURL(rdr.Config().SourcePath, rdr.Config().ProcessedPath); err != nil { + return + } + er = rdr + + return +} + +// SQLEventReader implements EventReader interface for sql +type SQLEventReader struct { + // sync.RWMutex + cgrCfg *config.CGRConfig + cfgIdx int // index of config instance within ERsCfg.Readers + fltrS *engine.FilterS + + connString string + connType string + tableName string + + expConnString string + expConnType string + expTableName string + + rdrEvents chan *erEvent // channel to dispatch the events created to + rdrExit chan struct{} + rdrErr chan error + cap chan struct{} +} + +// Config returns the curent configuration +func (rdr *SQLEventReader) Config() *config.EventReaderCfg { + return rdr.cgrCfg.ERsCfg().Readers[rdr.cfgIdx] +} + +// Serve will start the gorutines needed to watch the kafka topic +func (rdr *SQLEventReader) Serve() (err error) { + // connectString := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'", user, password, host, port, name) + var db *gorm.DB + if db, err = gorm.Open(strings.TrimPrefix(rdr.connType, utils.Meta), rdr.connString); err != nil { + return + } + if err = db.DB().Ping(); err != nil { + return + } + if rdr.Config().RunDelay == time.Duration(0) { // 0 disables the automatic read, maybe done per API + return + } + go rdr.readLoop(db) // read until the connection is closed + return +} + +func (rdr *SQLEventReader) readLoop(db *gorm.DB) { + for { + if db = db.Table(rdr.tableName).Select("*"); db.Error != nil { + rdr.rdrErr <- db.Error + return + } + rows, err := db.Rows() + if err != nil { + rdr.rdrErr <- err + return + } + colNames, err := rows.Columns() + if err != nil { + rdr.rdrErr <- err + return + } + for rows.Next() { + select { + case <-rdr.rdrExit: + utils.Logger.Info( + fmt.Sprintf("<%s> stop monitoring sql DB <%s>", + utils.ERs, rdr.Config().SourcePath)) + db.Close() + return + default: + } + if err := rows.Err(); err != nil { + rdr.rdrErr <- err + return + } + if rdr.Config().ConcurrentReqs != -1 { + <-rdr.cap // do not try to read if the limit is reached + } + columns := make([]interface{}, len(colNames)) + columnPointers := make([]interface{}, len(colNames)) + for i := range columns { + columnPointers[i] = &columns[i] + } + if err = rows.Scan(columnPointers...); err != nil { + rdr.rdrErr <- err + return + } + go func(columns []interface{}, colNames []string) { + msg := make(map[string]interface{}) + for i, colName := range colNames { + msg[colName] = columns[i] + } + if err := rdr.processMessage(msg); err != nil { + utils.Logger.Warning( + fmt.Sprintf("<%s> processing message %s error: %s", + utils.ERs, utils.ToJSON(msg), err.Error())) + } + db.Delete(msg) // to ensure we don't read it again + if rdr.Config().ProcessedPath != utils.EmptyString { + // post it + } + if rdr.Config().ConcurrentReqs != -1 { + rdr.cap <- struct{}{} + } + }(columns, colNames) + } + if rdr.Config().RunDelay < 0 { + return + } + time.Sleep(rdr.Config().RunDelay) + } +} + +func (rdr *SQLEventReader) processMessage(msg map[string]interface{}) (err error) { + reqVars := make(map[string]interface{}) + agReq := agents.NewAgentRequest( + config.NewNavigableMap(msg), reqVars, + nil, nil, rdr.Config().Tenant, + rdr.cgrCfg.GeneralCfg().DefaultTenant, + utils.FirstNonEmpty(rdr.Config().Timezone, + rdr.cgrCfg.GeneralCfg().DefaultTimezone), + rdr.fltrS) // create an AgentRequest + var pass bool + if pass, err = rdr.fltrS.Pass(agReq.Tenant, rdr.Config().Filters, + agReq); err != nil || !pass { + return + } + var navMp *config.NavigableMap + if navMp, err = agReq.AsNavigableMap(rdr.Config().ContentFields); err != nil { + return + } + rdr.rdrEvents <- &erEvent{ + cgrEvent: navMp.AsCGREvent(agReq.Tenant, utils.NestingSep), + rdrCfg: rdr.Config(), + } + return +} + +func (rdr *SQLEventReader) setURL(inURL, outURL string) (err error) { + // *dbtype:user:password@host:port?options + split := strings.SplitN(inURL, utils.InInFieldSep, 2) + if len(split) != 2 { + return utils.NewErrMandatoryIeMissing("db_type") + } + rdr.connType = split[0] + inURL = split[1] + + //outhpath if no meta is op[tions only + var u *url.URL + if u, err = url.Parse(inURL); err != nil { + return + } + password, _ := u.User.Password() + qry := u.Query() + + dbname := defaultDBName + if vals, has := qry[dbName]; has && len(vals) != 0 { + dbname = vals[0] + } + ssl := defaultSSLMode + if vals, has := qry[sslMode]; has && len(vals) != 0 { + ssl = vals[0] + } + + rdr.tableName = utils.CDRsTBL + if vals, has := qry[tableName]; has && len(vals) != 0 { + rdr.tableName = vals[0] + } + + switch rdr.connType { + case utils.MetaMySQL: + rdr.connString = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'", + u.User.Username(), password, u.Hostname(), u.Port(), dbname) + case utils.MetaPostgres: + rdr.connString = fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s sslmode=%s", u.Hostname(), u.Port(), dbname, u.User.Username(), password, ssl) + default: + return fmt.Errorf("unknown db_type %s", rdr.connType) + } + + // outURL + if len(outURL) == 0 { + return + } + var outUser, outPassword, outDBname, outSSL, outHost, outPort string + var oqry url.Values + if !strings.HasPrefix(outURL, utils.Meta) { + rdr.expConnType = rdr.connType + outUser = u.User.Username() + outPassword = password + outHost = u.Hostname() + outPort = u.Port() + if oqry, err = url.ParseQuery(outURL); err != nil { + return + } + } else { + split := strings.SplitN(inURL, utils.ConcatenatedKey(), 2) + if len(split) != 2 { + return utils.NewErrMandatoryIeMissing("db_type") + } + rdr.expConnType = split[0] + inURL = split[1] + + //outhpath if no meta is op[tions only + var outURL *url.URL + if outURL, err = url.Parse(inURL); err != nil { + return + } + outPassword, _ = outURL.User.Password() + outUser = outURL.User.Username() + outHost = outURL.Hostname() + outPort = outURL.Port() + oqry = outURL.Query() + } + + outDBname = defaultDBName + if vals, has := oqry[dbName]; has && len(vals) != 0 { + outDBname = vals[0] + } + outSSL = defaultSSLMode + if vals, has := oqry[sslMode]; has && len(vals) != 0 { + outSSL = vals[0] + } + rdr.expTableName = utils.CDRsTBL + if vals, has := qry[tableName]; has && len(vals) != 0 { + rdr.expTableName = vals[0] + } + + switch rdr.connType { + case utils.MetaMySQL: + rdr.expConnString = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'", + outUser, outPassword, outHost, outPort, outDBname) + case utils.MetaPostgres: + rdr.expConnString = fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s sslmode=%s", + outHost, outPort, outDBname, outUser, outPassword, outSSL) + default: + return fmt.Errorf("unknown db type") + } + return +} diff --git a/ers/sql_it_test.go b/ers/sql_it_test.go new file mode 100644 index 000000000..a15694c9b --- /dev/null +++ b/ers/sql_it_test.go @@ -0,0 +1,196 @@ +// +build integration + +/* +Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments +Copyright (C) ITsysCOM GmbH + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see +*/ + +package ers + +import ( + "reflect" + "testing" + "time" + + "github.com/cgrates/cgrates/config" + "github.com/cgrates/cgrates/engine" + "github.com/cgrates/cgrates/utils" + _ "github.com/go-sql-driver/mysql" + "github.com/jinzhu/gorm" +) + +var ( + sqlCfgPath string + sqlCfg *config.CGRConfig + sqlTests = []func(t *testing.T){ + testSQLInitConfig, + testSQLInitCdrDb, + testSQLInitDB, + testSQLReader, + testSQLEmptyTable, + testSQLInitDB, + testSQLReader2, + testSQLStop, + } + cdr = &engine.CDR{ + CGRID: "CGRID", + RunID: "RunID", + } + db *gorm.DB + dbConnString = "cgrates:CGRateS.org@tcp(127.0.0.1:3306)/cgrates?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'" +) + +func TestSQL(t *testing.T) { + // sqlCfgPath = path.Join(*dataDir, "conf", "samples", "ers_reload", "disabled") + for _, test := range sqlTests { + t.Run("TestSQL", test) + } +} + +func testSQLInitConfig(t *testing.T) { + var err error + if sqlCfg, err = config.NewCGRConfigFromJsonStringWithDefaults(`{ + "stor_db": { + "db_password": "CGRateS.org", + }, + "ers": { // EventReaderService + "enabled": true, // starts the EventReader service: + "readers": [ + { + "id": "mysql", // identifier of the EventReader profile + "type": "*sql", // reader type <*file_csv> + "run_delay": 1, // sleep interval in seconds between consecutive runs, -1 to use automation via inotify or 0 to disable running all together + "concurrent_requests": 1024, // maximum simultaneous requests/files to process, 0 for unlimited + "source_path": "*mysql:cgrates:CGRateS.org@127.0.0.1:3306", // read data from this path + // "processed_path": "/var/spool/cgrates/cdrc/out", // move processed data here + "tenant": "cgrates.org", // tenant used by import + "filters": [], // limit parsing based on the filters + "flags": [], // flags to influence the event processing + // "header_fields": [], // template of the import header fields + "content_fields":[ // import content_fields template, tag will match internally CDR field, in case of .csv value will be represented by index of the field value + {"tag": "CGRID", "type": "*composed", "value": "~*req.cgrid", "field_id": "CGRID"}, + ], + // "trailer_fields": [], // template of the import trailer fields + }, + ], + }, + }`); err != nil { + t.Fatal(err) + } + utils.Newlogger(utils.MetaSysLog, sqlCfg.GeneralCfg().NodeID) + utils.Logger.SetLogLevel(7) +} + +func testSQLInitCdrDb(t *testing.T) { + if err := engine.InitStorDb(sqlCfg); err != nil { + t.Fatal(err) + } +} + +func testSQLInitDB(t *testing.T) { + cdr.CGRID = utils.UUIDSha1Prefix() + var err error + db, err = gorm.Open("mysql", dbConnString) + if err != nil { + t.Fatal(err) + } + db = db.Table(utils.CDRsTBL) + + tx := db.Begin() + cdrSql := cdr.AsCDRsql() + cdrSql.CreatedAt = time.Now() + saved := tx.Save(cdrSql) + if saved.Error != nil { + tx.Rollback() + t.Fatal(err) + } + tx.Commit() +} + +func testSQLReader(t *testing.T) { + rdrEvents = make(chan *erEvent, 1) + rdrErr = make(chan error, 1) + rdrExit = make(chan struct{}, 1) + sqlER, err := NewEventReader(sqlCfg, 1, rdrEvents, rdrErr, new(engine.FilterS), rdrExit) + if err != nil { + t.Fatal(err) + } + sqlER.Serve() + + select { + case err = <-rdrErr: + t.Error(err) + case ev := <-rdrEvents: + if ev.rdrCfg.ID != "mysql" { + t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID) + } + expected := &utils.CGREvent{ + Tenant: "cgrates.org", + ID: ev.cgrEvent.ID, + Time: ev.cgrEvent.Time, + Event: map[string]interface{}{ + "CGRID": cdr.CGRID, + }, + } + if !reflect.DeepEqual(ev.cgrEvent, expected) { + t.Errorf("Expected %s ,received %s", utils.ToJSON(expected), utils.ToJSON(ev.cgrEvent)) + } + case <-time.After(time.Second): + t.Fatal("Timeout") + } +} + +func testSQLEmptyTable(t *testing.T) { + rows, err := db.Table(utils.CDRsTBL).Select("*").Rows() + if err != nil { + t.Error(err) + } + for rows.Next() { + t.Fatal("Expected empty table") + } +} + +func testSQLReader2(t *testing.T) { + + select { + case err := <-rdrErr: + t.Error(err) + case ev := <-rdrEvents: + if ev.rdrCfg.ID != "mysql" { + t.Errorf("Expected 'mysql' received `%s`", ev.rdrCfg.ID) + } + expected := &utils.CGREvent{ + Tenant: "cgrates.org", + ID: ev.cgrEvent.ID, + Time: ev.cgrEvent.Time, + Event: map[string]interface{}{ + "CGRID": cdr.CGRID, + }, + } + if !reflect.DeepEqual(ev.cgrEvent, expected) { + t.Errorf("Expected %s ,received %s", utils.ToJSON(expected), utils.ToJSON(ev.cgrEvent)) + } + case <-time.After(time.Second): + t.Fatal("Timeout") + } +} + +func testSQLStop(t *testing.T) { + rdrExit <- struct{}{} + if err := db.Close(); err != nil { + t.Error(err) + } +} diff --git a/utils/consts.go b/utils/consts.go index 56d90f47b..e979c9916 100755 --- a/utils/consts.go +++ b/utils/consts.go @@ -301,6 +301,9 @@ const ( MetaAMQPV1jsonMap = "*amqpv1_json_map" MetaSQSjsonMap = "*sqs_json_map" MetaKafkajsonMap = "*kafka_json_map" + MetaSQL = "*sql" + MetaMySQL = "*mysql" + MetaPostgres = "*postgres" MetaS3jsonMap = "*s3_json_map" CONFIG_PATH = "/etc/cgrates/" DISCONNECT_CAUSE = "DisconnectCause" From 656fb8be00ffd1f4c7434e0c4780e6d86a3bb9bc Mon Sep 17 00:00:00 2001 From: Trial97 Date: Tue, 17 Dec 2019 11:43:12 +0200 Subject: [PATCH 3/3] Updated *sql Event Reader --- ers/sql.go | 89 +++++++++++++++++++++++---------------- ers/sql_it_test.go | 103 ++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 148 insertions(+), 44 deletions(-) diff --git a/ers/sql.go b/ers/sql.go index 5e479dd6e..f6b4a88ce 100644 --- a/ers/sql.go +++ b/ers/sql.go @@ -37,7 +37,7 @@ import ( const ( dbName = "db_name" - tableName = "db_name" + tableName = "table_name" sslMode = "sslmode" defaultSSLMode = "disable" defaultDBName = "cgrates" @@ -98,9 +98,8 @@ func (rdr *SQLEventReader) Config() *config.EventReaderCfg { // Serve will start the gorutines needed to watch the kafka topic func (rdr *SQLEventReader) Serve() (err error) { - // connectString := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'", user, password, host, port, name) var db *gorm.DB - if db, err = gorm.Open(strings.TrimPrefix(rdr.connType, utils.Meta), rdr.connString); err != nil { + if db, err = gorm.Open(rdr.connType, rdr.connString); err != nil { return } if err = db.DB().Ping(); err != nil { @@ -165,9 +164,13 @@ func (rdr *SQLEventReader) readLoop(db *gorm.DB) { fmt.Sprintf("<%s> processing message %s error: %s", utils.ERs, utils.ToJSON(msg), err.Error())) } - db.Delete(msg) // to ensure we don't read it again + db = db.Delete(msg) // to ensure we don't read it again if rdr.Config().ProcessedPath != utils.EmptyString { - // post it + if err = rdr.postCDR(columns); err != nil { + utils.Logger.Warning( + fmt.Sprintf("<%s> posting message %s error: %s", + utils.ERs, utils.ToJSON(msg), err.Error())) + } } if rdr.Config().ConcurrentReqs != -1 { rdr.cap <- struct{}{} @@ -207,21 +210,14 @@ func (rdr *SQLEventReader) processMessage(msg map[string]interface{}) (err error } func (rdr *SQLEventReader) setURL(inURL, outURL string) (err error) { - // *dbtype:user:password@host:port?options - split := strings.SplitN(inURL, utils.InInFieldSep, 2) - if len(split) != 2 { - return utils.NewErrMandatoryIeMissing("db_type") - } - rdr.connType = split[0] - inURL = split[1] - - //outhpath if no meta is op[tions only + inURL = strings.TrimPrefix(inURL, utils.Meta) var u *url.URL if u, err = url.Parse(inURL); err != nil { return } password, _ := u.User.Password() qry := u.Query() + rdr.connType = u.Scheme dbname := defaultDBName if vals, has := qry[dbName]; has && len(vals) != 0 { @@ -236,12 +232,11 @@ func (rdr *SQLEventReader) setURL(inURL, outURL string) (err error) { if vals, has := qry[tableName]; has && len(vals) != 0 { rdr.tableName = vals[0] } - switch rdr.connType { - case utils.MetaMySQL: + case utils.MYSQL: rdr.connString = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'", u.User.Username(), password, u.Hostname(), u.Port(), dbname) - case utils.MetaPostgres: + case utils.POSTGRES: rdr.connString = fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s sslmode=%s", u.Hostname(), u.Port(), dbname, u.User.Username(), password, ssl) default: return fmt.Errorf("unknown db_type %s", rdr.connType) @@ -263,23 +258,17 @@ func (rdr *SQLEventReader) setURL(inURL, outURL string) (err error) { return } } else { - split := strings.SplitN(inURL, utils.ConcatenatedKey(), 2) - if len(split) != 2 { - return utils.NewErrMandatoryIeMissing("db_type") - } - rdr.expConnType = split[0] - inURL = split[1] - - //outhpath if no meta is op[tions only - var outURL *url.URL - if outURL, err = url.Parse(inURL); err != nil { + outURL = strings.TrimPrefix(outURL, utils.Meta) + var oURL *url.URL + if oURL, err = url.Parse(inURL); err != nil { return } - outPassword, _ = outURL.User.Password() - outUser = outURL.User.Username() - outHost = outURL.Hostname() - outPort = outURL.Port() - oqry = outURL.Query() + rdr.expConnType = oURL.Scheme + outPassword, _ = oURL.User.Password() + outUser = oURL.User.Username() + outHost = oURL.Hostname() + outPort = oURL.Port() + oqry = oURL.Query() } outDBname = defaultDBName @@ -291,19 +280,45 @@ func (rdr *SQLEventReader) setURL(inURL, outURL string) (err error) { outSSL = vals[0] } rdr.expTableName = utils.CDRsTBL - if vals, has := qry[tableName]; has && len(vals) != 0 { + if vals, has := oqry[tableName]; has && len(vals) != 0 { rdr.expTableName = vals[0] } - switch rdr.connType { - case utils.MetaMySQL: + switch rdr.expConnType { + case utils.MYSQL: rdr.expConnString = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'", outUser, outPassword, outHost, outPort, outDBname) - case utils.MetaPostgres: + case utils.POSTGRES: rdr.expConnString = fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s sslmode=%s", outHost, outPort, outDBname, outUser, outPassword, outSSL) default: - return fmt.Errorf("unknown db type") + return fmt.Errorf("unknown db_type") } return } + +func (rdr *SQLEventReader) postCDR(in []interface{}) (err error) { + sqlValues := make([]string, len(in)) + for i := range in { + sqlValues[i] = "?" + } + sqlStatement := fmt.Sprintf("INSERT INTO %s VALUES (%s); ", rdr.expTableName, strings.Join(sqlValues, ",")) + var db *gorm.DB + if db, err = gorm.Open(rdr.expConnType, rdr.expConnString); err != nil { + return + } + if err = db.DB().Ping(); err != nil { + return + } + tx := db.Begin() + _, err = db.DB().Exec(sqlStatement, in...) + if err != nil { + tx.Rollback() + if strings.Contains(err.Error(), "1062") || strings.Contains(err.Error(), "duplicate key") { // returns 1062/pq when key is duplicated + return utils.ErrExists + } + return + } + tx.Commit() + return +} diff --git a/ers/sql_it_test.go b/ers/sql_it_test.go index a15694c9b..b673da954 100644 --- a/ers/sql_it_test.go +++ b/ers/sql_it_test.go @@ -41,8 +41,11 @@ var ( testSQLInitDB, testSQLReader, testSQLEmptyTable, + testSQLPoster, + testSQLInitDB, testSQLReader2, + testSQLStop, } cdr = &engine.CDR{ @@ -50,7 +53,7 @@ var ( RunID: "RunID", } db *gorm.DB - dbConnString = "cgrates:CGRateS.org@tcp(127.0.0.1:3306)/cgrates?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'" + dbConnString = "cgrates:CGRateS.org@tcp(127.0.0.1:3306)/cgrates2?charset=utf8&loc=Local&parseTime=true&sql_mode='ALLOW_INVALID_DATES'" ) func TestSQL(t *testing.T) { @@ -74,8 +77,8 @@ func testSQLInitConfig(t *testing.T) { "type": "*sql", // reader type <*file_csv> "run_delay": 1, // sleep interval in seconds between consecutive runs, -1 to use automation via inotify or 0 to disable running all together "concurrent_requests": 1024, // maximum simultaneous requests/files to process, 0 for unlimited - "source_path": "*mysql:cgrates:CGRateS.org@127.0.0.1:3306", // read data from this path - // "processed_path": "/var/spool/cgrates/cdrc/out", // move processed data here + "source_path": "*mysql://cgrates:CGRateS.org@127.0.0.1:3306?db_name=cgrates2", // read data from this path + "processed_path": "db_name=cgrates2&table_name=cdrs2", // move processed data here "tenant": "cgrates.org", // tenant used by import "filters": [], // limit parsing based on the filters "flags": [], // flags to influence the event processing @@ -95,9 +98,47 @@ func testSQLInitConfig(t *testing.T) { } func testSQLInitCdrDb(t *testing.T) { + rdrsql := sqlCfg.StorDbCfg().Clone() if err := engine.InitStorDb(sqlCfg); err != nil { t.Fatal(err) } + sqlCfg.StorDbCfg().Name = "cgrates2" + if err := engine.InitStorDb(sqlCfg); err != nil { + t.Fatal(err) + } + *sqlCfg.StorDbCfg() = *rdrsql + +} + +type testModelSql struct { + ID int64 + Cgrid string + RunID string + OriginHost string + Source string + OriginID string + TOR string + RequestType string + Tenant string + Category string + Account string + Subject string + Destination string + SetupTime time.Time + AnswerTime time.Time + Usage int64 + ExtraFields string + CostSource string + Cost float64 + CostDetails string + ExtraInfo string + CreatedAt time.Time + UpdatedAt time.Time + DeletedAt *time.Time +} + +func (_ *testModelSql) TableName() string { + return "cdrs2" } func testSQLInitDB(t *testing.T) { @@ -107,8 +148,10 @@ func testSQLInitDB(t *testing.T) { if err != nil { t.Fatal(err) } + if !db.HasTable("cdrs2") { + db = db.CreateTable(&testModelSql{}) + } db = db.Table(utils.CDRsTBL) - tx := db.Begin() cdrSql := cdr.AsCDRsql() cdrSql.CreatedAt = time.Now() @@ -154,17 +197,33 @@ func testSQLReader(t *testing.T) { } func testSQLEmptyTable(t *testing.T) { + time.Sleep(10 * time.Millisecond) rows, err := db.Table(utils.CDRsTBL).Select("*").Rows() if err != nil { - t.Error(err) + t.Fatal(err) + } + colNames, err := rows.Columns() + if err != nil { + t.Fatal(err) } for rows.Next() { - t.Fatal("Expected empty table") + columns := make([]interface{}, len(colNames)) + columnPointers := make([]interface{}, len(colNames)) + for i := range columns { + columnPointers[i] = &columns[i] + } + if err = rows.Scan(columnPointers...); err != nil { + t.Fatal(err) + } + msg := make(map[string]interface{}) + for i, colName := range colNames { + msg[colName] = columns[i] + } + t.Fatal("Expected empty table ", utils.ToJSON(msg)) } } func testSQLReader2(t *testing.T) { - select { case err := <-rdrErr: t.Error(err) @@ -188,8 +247,38 @@ func testSQLReader2(t *testing.T) { } } +func testSQLPoster(t *testing.T) { + rows, err := db.Table("cdrs2").Select("*").Rows() + if err != nil { + t.Fatal(err) + } + colNames, err := rows.Columns() + if err != nil { + t.Fatal(err) + } + for rows.Next() { + columns := make([]interface{}, len(colNames)) + columnPointers := make([]interface{}, len(colNames)) + for i := range columns { + columnPointers[i] = &columns[i] + } + if err = rows.Scan(columnPointers...); err != nil { + t.Fatal(err) + } + msg := make(map[string]interface{}) + for i, colName := range colNames { + 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) + } + } +} + func testSQLStop(t *testing.T) { rdrExit <- struct{}{} + db = db.DropTable("cdrs2") if err := db.Close(); err != nil { t.Error(err) }