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"