Added *sql Event Reader

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)

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
 }

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 <http://www.gnu.org/licenses/>
+*/
+
+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
+}

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 <http://www.gnu.org/licenses/>
+*/
+
+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: <true|false>
+			"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)
+	}
+}

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"