Updated analyzers

analyzers/analyzers.go

@@ -20,17 +20,46 @@ package analyzers
 
 import (
 	"fmt"
+	"os"
+	"strconv"
+	"time"
 
+	"github.com/blevesearch/bleve"
+	// import the bleve packages in order to register the indextype and storagetype
+	"github.com/blevesearch/bleve/document"
+	_ "github.com/blevesearch/bleve/index/scorch"
+	_ "github.com/blevesearch/bleve/index/store/boltdb"
+	_ "github.com/blevesearch/bleve/index/store/goleveldb"
+	_ "github.com/blevesearch/bleve/index/store/moss"
+	_ "github.com/blevesearch/bleve/index/upsidedown"
+	"github.com/cgrates/cgrates/config"
 	"github.com/cgrates/cgrates/utils"
 )
 
 // NewAnalyzerService initializes a AnalyzerService
-func NewAnalyzerService() (*AnalyzerService, error) {
-	return &AnalyzerService{}, nil
+func NewAnalyzerService(cfg *config.CGRConfig) (aS *AnalyzerService, err error) {
+	aS = &AnalyzerService{cfg: cfg}
+	err = aS.initDB()
+	return
 }
 
 // AnalyzerService is the service handling analyzer
 type AnalyzerService struct {
+	db  bleve.Index
+	cfg *config.CGRConfig
+}
+
+func (aS *AnalyzerService) initDB() (err error) {
+	fmt.Println(aS.cfg.AnalyzerSCfg().DBPath)
+	if _, err = os.Stat(aS.cfg.AnalyzerSCfg().DBPath); err == nil {
+		fmt.Println("exista")
+		aS.db, err = bleve.Open(aS.cfg.AnalyzerSCfg().DBPath)
+	} else if os.IsNotExist(err) {
+		fmt.Println("nu exista")
+		aS.db, err = bleve.NewUsing(aS.cfg.AnalyzerSCfg().DBPath, bleve.NewIndexMapping(),
+			aS.cfg.AnalyzerSCfg().IndexType, aS.cfg.AnalyzerSCfg().StoreType, nil)
+	}
+	return
 }
 
 // ListenAndServe will initialize the service
@@ -44,6 +73,55 @@ func (aS *AnalyzerService) ListenAndServe(exitChan chan bool) error {
 // Shutdown is called to shutdown the service
 func (aS *AnalyzerService) Shutdown() error {
 	utils.Logger.Info(fmt.Sprintf("<%s> service shutdown initialized", utils.AnalyzerS))
+	aS.db.Close()
 	utils.Logger.Info(fmt.Sprintf("<%s> service shutdown complete", utils.AnalyzerS))
 	return nil
 }
+
+func (aS *AnalyzerService) logTrafic(id uint64, method string,
+	params, result, err interface{},
+	info *extraInfo, sTime, eTime time.Time) error {
+	var e interface{}
+	switch val := err.(type) {
+	default:
+	case nil:
+	case string:
+		e = val
+	case error:
+		e = val.Error()
+	}
+	return aS.db.Index(utils.ConcatenatedKey(method, strconv.FormatInt(sTime.Unix(), 10)),
+		InfoRPC{
+			Duration:  eTime.Sub(sTime),
+			StartTime: sTime,
+			EndTime:   eTime,
+
+			Encoding: info.enc,
+			From:     info.from,
+			To:       info.to,
+
+			ID:     id,
+			Method: method,
+			Params: params,
+			Result: result,
+			Error:  e,
+		})
+}
+
+func (aS *AnalyzerService) V1Search(searchstr string, reply *[]*document.Document) error {
+	s := bleve.NewSearchRequest(bleve.NewQueryStringQuery(searchstr))
+	searchResults, err := aS.db.Search(s)
+	if err != nil {
+		return err
+	}
+	rply := make([]*document.Document, searchResults.Hits.Len())
+	for i, obj := range searchResults.Hits {
+		fmt.Println(obj.ID)
+		fmt.Println(obj.Index)
+		d, _ := aS.db.Document(obj.ID)
+		fmt.Println(d.Fields[0].Name())
+		rply[i] = d
+	}
+	*reply = rply
+	return nil
+}

analyzers/codec.go

@@ -20,34 +20,60 @@ package analyzers
 
 import (
 	"net/rpc"
+	"sync"
+	"time"
 )
 
-func NewAnalyzeServerCodec(sc rpc.ServerCodec) rpc.ServerCodec {
-	return &AnalyzeServerCodec{sc: sc, req: new(RPCServerRequest)}
+func (aS *AnalyzerService) NewServerCodec(sc rpc.ServerCodec, enc, from, to string) rpc.ServerCodec {
+	return &AnalyzeServerCodec{
+		sc:   sc,
+		reqs: make(map[uint64]*rpcAPI),
+		aS:   aS,
+		extrainfo: &extraInfo{
+			enc:  enc,
+			from: from,
+			to:   to,
+		},
+	}
 }
 
 type AnalyzeServerCodec struct {
 	sc rpc.ServerCodec
-	// keep the information about the header so we handle this when the body is readed
-	// the ReadRequestHeader and ReadRequestBody are called in pairs
-	req *RPCServerRequest
+
+	// keep the API in memory because the write is async
+	reqs      map[uint64]*rpcAPI
+	reqIdx    uint64
+	reqsLk    sync.RWMutex
+	aS        *AnalyzerService
+	extrainfo *extraInfo
 }
 
 func (c *AnalyzeServerCodec) ReadRequestHeader(r *rpc.Request) (err error) {
-	c.req.reset()
 	err = c.sc.ReadRequestHeader(r)
-	c.req.Method = r.ServiceMethod
-	c.req.ID = r.Seq
+	c.reqsLk.Lock()
+	c.reqIdx = r.Seq
+	c.reqs[c.reqIdx] = &rpcAPI{
+		ID:        r.Seq,
+		Method:    r.ServiceMethod,
+		StartTime: time.Now(),
+	}
+	c.reqsLk.Unlock()
 	return
 }
 
 func (c *AnalyzeServerCodec) ReadRequestBody(x interface{}) (err error) {
 	err = c.sc.ReadRequestBody(x)
-	go h.handleRequest(c.req.ID, c.req.Method, x)
+	c.reqsLk.Lock()
+	c.reqs[c.reqIdx].Params = x
+	c.reqsLk.Unlock()
 	return
 }
 func (c *AnalyzeServerCodec) WriteResponse(r *rpc.Response, x interface{}) error {
-	go h.handleResponse(r.Seq, x, r.Error)
+	c.reqsLk.Lock()
+	api := c.reqs[c.reqIdx]
+	delete(c.reqs, c.reqIdx)
+	c.reqsLk.Unlock()
+	go c.aS.logTrafic(api.ID, api.Method, api.Params, x, r.Error, c.extrainfo, api.StartTime, time.Now())
 	return c.sc.WriteResponse(r, x)
 }
 func (c *AnalyzeServerCodec) Close() error { return c.sc.Close() }

analyzers/connector.go

@@ -19,28 +19,33 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package analyzers
 
 import (
-	"sync"
+	"time"
 
 	"github.com/cgrates/rpcclient"
 )
 
-func NewAnalyzeConnector(sc rpcclient.ClientConnector) rpcclient.ClientConnector {
-	return &AnalyzeConnector{conn: sc}
+func (aS *AnalyzerService) NewAnalyzeConnector(sc rpcclient.ClientConnector, enc, from, to string) rpcclient.ClientConnector {
+	return &AnalyzeConnector{
+		conn: sc,
+		aS:   aS,
+		extrainfo: &extraInfo{
+			enc:  enc,
+			from: from,
+			to:   to,
+		},
+	}
 }
 
 type AnalyzeConnector struct {
-	conn  rpcclient.ClientConnector
-	seq   uint64
-	seqLk sync.Mutex
+	conn rpcclient.ClientConnector
+
+	aS        *AnalyzerService
+	extrainfo *extraInfo
 }
 
 func (c *AnalyzeConnector) Call(serviceMethod string, args interface{}, reply interface{}) (err error) {
-	c.seqLk.Lock()
-	id := c.seq
-	c.seq++
-	c.seqLk.Unlock()
-	go h.handleRequest(id, serviceMethod, args)
+	sTime := time.Now()
 	err = c.conn.Call(serviceMethod, args, reply)
-	go h.handleResponse(id, reply, err)
+	go c.aS.logTrafic(0, serviceMethod, args, reply, err, c.extrainfo, sTime, time.Now())
 	return
 }

analyzers/utils.go

@@ -19,61 +19,34 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package analyzers
 
 import (
-	"log"
-
-	"github.com/cgrates/cgrates/utils"
+	"time"
 )
 
-var h = new(Handler)
-
-type Handler struct {
+type extraInfo struct {
 	enc  string
 	from string
 	to   string
 }
 
-func (h *Handler) handleTrafic(x interface{}) {
-	log.Println(utils.ToJSON(x))
-}
+type InfoRPC struct {
+	Duration  time.Duration
+	StartTime time.Time
+	EndTime   time.Time
 
-func (h *Handler) handleRequest(id uint64, method string, args interface{}) {
-	h.handleTrafic(&RPCServerRequest{
-		ID:     id,
-		Method: method,
-		Params: []interface{}{args},
-	})
-}
-func (h *Handler) handleResponse(id uint64, x, err interface{}) {
-	var e interface{}
-	switch val := x.(type) {
-	default:
-	case nil:
-	case string:
-		e = val
-	case error:
-		e = val.Error()
-	}
-	h.handleTrafic(&RPCServerResponse{
-		ID:     id,
-		Result: x,
-		Error:  e,
-	})
-}
+	Encoding string
+	From     string
+	To       string
 
-type RPCServerRequest struct {
-	Method string        `json:"method"`
-	Params []interface{} `json:"params"`
-	ID     uint64        `json:"id"`
+	ID     uint64
+	Method string
+	Params interface{}
+	Result interface{}
+	Error  interface{}
 }
-
-func (r *RPCServerRequest) reset() {
-	r.Method = ""
-	r.Params = nil
-	r.ID = 0
-}
-
-type RPCServerResponse struct {
+type rpcAPI struct {
 	ID     uint64      `json:"id"`
-	Result interface{} `json:"result"`
-	Error  interface{} `json:"error"`
+	Method string      `json:"method"`
+	Params interface{} `json:"params"`
+
+	StartTime time.Time
 }