New config option - attributes.process_runs

cmd/cgr-engine/cgr-engine.go

@@ -582,7 +582,9 @@ func startAttributeService(internalAttributeSChan chan rpcclient.RpcClientConnec
 	<-cacheS.GetPrecacheChannel(utils.CacheAttributeProfiles)
 
 	aS, err := engine.NewAttributeService(dm, filterS,
-		cfg.AttributeSCfg().StringIndexedFields, cfg.AttributeSCfg().PrefixIndexedFields)
+		cfg.AttributeSCfg().StringIndexedFields,
+		cfg.AttributeSCfg().PrefixIndexedFields,
+		cfg.AttributeSCfg().ProcessRuns)
 	if err != nil {
 		utils.Logger.Crit(
 			fmt.Sprintf("<%s> Could not init, error: %s",

config/attributescfg.go

@@ -23,6 +23,7 @@ type AttributeSCfg struct {
 	Enabled             bool
 	StringIndexedFields *[]string
 	PrefixIndexedFields *[]string
+	ProcessRuns         int
 }
 
 func (alS *AttributeSCfg) loadFromJsonCfg(jsnCfg *AttributeSJsonCfg) (err error) {
@@ -46,5 +47,8 @@ func (alS *AttributeSCfg) loadFromJsonCfg(jsnCfg *AttributeSJsonCfg) (err error)
 		}
 		alS.PrefixIndexedFields = &pif
 	}
+	if jsnCfg.Process_runs != nil {
+		alS.ProcessRuns = *jsnCfg.Process_runs
+	}
 	return
 }

config/config.go

@@ -658,6 +658,11 @@ func (self *CGRConfig) checkConfigSanity() error {
 			}
 		}
 	}
+	if self.attributeSCfg != nil && self.attributeSCfg.Enabled {
+		if self.attributeSCfg.ProcessRuns < 1 {
+			return fmt.Errorf("<%s> process_runs needs to be bigger than 0", utils.AttributeS)
+		}
+	}
 	if self.chargerSCfg != nil && self.chargerSCfg.Enabled {
 		for _, connCfg := range self.chargerSCfg.AttributeSConns {
 			if connCfg.Address == utils.MetaInternal &&

config/config_defaults.go

@@ -430,6 +430,7 @@ const CGRATES_CFG_JSON = `
 	"enabled": false,						// starts attribute service: <true|false>.
 	//"string_indexed_fields": [],			// query indexes based on these fields for faster processing
 	"prefix_indexed_fields": [],			// query indexes based on these fields for faster processing
+	"process_runs": 1,						// number of run loops when processing event
 },
 
 

config/config_json_test.go

@@ -705,6 +705,7 @@ func TestDfAttributeServJsonCfg(t *testing.T) {
 		Enabled:               utils.BoolPointer(false),
 		String_indexed_fields: nil,
 		Prefix_indexed_fields: &[]string{},
+		Process_runs:          utils.IntPointer(1),
 	}
 	if cfg, err := dfCgrJsonCfg.AttributeServJsonCfg(); err != nil {
 		t.Error(err)

config/config_test.go

@@ -805,6 +805,7 @@ func TestCgrCfgJSONDefaultSAttributeSCfg(t *testing.T) {
 		Enabled:             false,
 		StringIndexedFields: nil,
 		PrefixIndexedFields: &[]string{},
+		ProcessRuns:         1,
 	}
 	if !reflect.DeepEqual(eAliasSCfg, cgrCfg.attributeSCfg) {
 		t.Errorf("received: %+v, expecting: %+v", eAliasSCfg, cgrCfg.attributeSCfg)

config/libconfig_json.go

@@ -427,6 +427,7 @@ type AttributeSJsonCfg struct {
 	Enabled               *bool
 	String_indexed_fields *[]string
 	Prefix_indexed_fields *[]string
+	Process_runs          *int
 }
 
 // ChargerSJsonCfg service config section

engine/attributes.go

@@ -25,10 +25,12 @@ import (
 )
 
 func NewAttributeService(dm *DataManager, filterS *FilterS,
-	stringIndexedFields, prefixIndexedFields *[]string) (*AttributeService, error) {
+	stringIndexedFields, prefixIndexedFields *[]string,
+	processRuns int) (*AttributeService, error) {
 	return &AttributeService{dm: dm, filterS: filterS,
 		stringIndexedFields: stringIndexedFields,
-		prefixIndexedFields: prefixIndexedFields}, nil
+		prefixIndexedFields: prefixIndexedFields,
+		processRuns:         processRuns}, nil
 }
 
 type AttributeService struct {
@@ -36,6 +38,7 @@ type AttributeService struct {
 	filterS             *FilterS
 	stringIndexedFields *[]string
 	prefixIndexedFields *[]string
+	processRuns         int
 }
 
 // ListenAndServe will initialize the service

engine/attributes_test.go

@@ -182,7 +182,7 @@ func TestAttributePopulateAttrService(t *testing.T) {
 	if err != nil {
 		t.Errorf("Error: %+v", err)
 	}
-	attrService, err = NewAttributeService(dmAtr, &FilterS{dm: dmAtr, cfg: defaultCfg}, nil, nil)
+	attrService, err = NewAttributeService(dmAtr, &FilterS{dm: dmAtr, cfg: defaultCfg}, nil, nil, 1)
 	if err != nil {
 		t.Errorf("Error: %+v", err)
 	}