ResourceS automatic storage mechanism with tests

apier/v1/resourcesv1.go

@@ -62,7 +62,7 @@ func (rsv1 *ResourceSV1) Call(serviceMethod string, args interface{}, reply inte
 }
 
 // GetResourcesForEvent returns Resources matching a specific event
-func (rsv1 *ResourceSV1) GetResourcesForEvent(args utils.ArgRSv1ResourceUsage, reply *[]*engine.ResourceProfile) error {
+func (rsv1 *ResourceSV1) GetResourcesForEvent(args utils.ArgRSv1ResourceUsage, reply *engine.Resources) error {
 	return rsv1.rls.V1ResourcesForEvent(args, reply)
 }
 

apier/v1/resourcesv1_it_test.go

@@ -49,10 +49,11 @@ var sTestsRLSV1 = []func(t *testing.T){
 	testV1RsRpcConn,
 	testV1RsFromFolder,
 	testV1RsGetResourcesForEvent,
-	/*testV1RsAllocateResource,
+	testV1RsAllocateResource,
 	testV1RsAllowUsage,
 	testV1RsReleaseResource,
-	testV1RsGetResourceConfigBeforeSet,
+	testV1RsDBStore,
+	/*testV1RsGetResourceConfigBeforeSet,
 	testV1RsSetResourceConfig,
 	testV1RsGetResourceConfigAfterSet,
 	testV1RsUpdateResourceConfig,
@@ -60,7 +61,7 @@ var sTestsRLSV1 = []func(t *testing.T){
 	testV1RsRemResourceCOnfig,
 	testV1RsGetResourceConfigAfterDelete,
 	*/
-	testV1RsStopEngine,
+	//testV1RsStopEngine,
 }
 
 //Test start here
@@ -174,96 +175,248 @@ func testV1RsGetResourcesForEvent(t *testing.T) {
 }
 
 func testV1RsAllocateResource(t *testing.T) {
+	// first event matching Resource1
 	var reply string
-
-	attrRU := utils.ArgRSv1ResourceUsage{
+	argsRU := utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
 		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e51",
-		Event:   map[string]interface{}{"Account": "1002", "Subject": "1001", "Destination": "1002"},
-		Units:   3,
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
+		Units: 3,
 	}
-	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", attrRU, &reply); err != nil {
+	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", argsRU, &reply); err != nil {
 		t.Error(err)
 	}
-	if reply != "ResGroup1" {
-		t.Errorf("Expecting: %+v, received: %+v", "ResGroup1", reply)
+	eAllocationMsg := "ResGroup1"
+	if reply != eAllocationMsg {
+		t.Errorf("Expecting: %+v, received: %+v", eAllocationMsg, reply)
 	}
-
-	time.Sleep(time.Duration(1000) * time.Millisecond)
-
-	attrRU = utils.ArgRSv1ResourceUsage{
+	// Second event to test matching of exact limit of first resource
+	argsRU = utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
 		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e52",
-		Event:   map[string]interface{}{"Destination": "100"},
-		Units:   5,
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
+		Units: 4,
 	}
-	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", attrRU, &reply); err != nil {
+	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", argsRU, &reply); err != nil {
 		t.Error(err)
 	}
-	if reply != "ResGroup2" {
-		t.Errorf("Expecting: %+v, received: %+v", "ResGroup2", reply)
+	eAllocationMsg = "ResGroup1"
+	if reply != eAllocationMsg {
+		t.Errorf("Expecting: %+v, received: %+v", eAllocationMsg, reply)
 	}
-
-	time.Sleep(time.Duration(1000) * time.Millisecond)
-
-	attrRU = utils.ArgRSv1ResourceUsage{
+	// Third event testing overflow to second resource which still has one resource available
+	argsRU = utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
 		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e53",
-		Event:   map[string]interface{}{"Account": "1002", "Subject": "1001", "Destination": "1002"},
-		Units:   3,
+		Event: map[string]interface{}{
+			"Account":     "dan",
+			"Subject":     "dan",
+			"Destination": "1002"},
+		Units: 1,
 	}
-	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", attrRU, &reply); err != nil {
+	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", argsRU, &reply); err != nil {
 		t.Error(err)
 	}
-	if reply != "ResGroup1" {
-		t.Errorf("Expecting: %+v, received: %+v", "ResGroup1", reply)
+	eAllocationMsg = "ResGroup2"
+	if reply != eAllocationMsg {
+		t.Errorf("Expecting: %+v, received: %+v", eAllocationMsg, reply)
 	}
+	// Test resource unavailable
+	argsRU = utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
+		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e54", // same ID should be accepted by first group since the previous resource should be expired
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
+		Units: 1,
+	}
+	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", argsRU, &reply); err == nil || err.Error() != utils.ErrResourceUnavailable.Error() {
+		t.Error(err)
+	}
+	eAllocationMsg = "ResGroup1"
+	time.Sleep(time.Duration(1000) * time.Millisecond) // Give time for allocations on first resource to expire
 
+	argsRU = utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
+		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e55", // same ID should be accepted by first group since the previous resource should be expired
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
+		Units: 1,
+	}
+	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", argsRU, &reply); err != nil {
+		t.Error(err)
+	}
+	eAllocationMsg = "ResGroup1"
+	if reply != eAllocationMsg {
+		t.Errorf("Expecting: %+v, received: %+v", eAllocationMsg, reply)
+	}
 }
 
 func testV1RsAllowUsage(t *testing.T) {
 	var allowed bool
-	attrRU := utils.ArgRSv1ResourceUsage{
-		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e51",
-		Event:   map[string]interface{}{"Account": "1002", "Subject": "1001", "Destination": "1002"},
-		Units:   1,
+	argsRU := utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
+		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e61",
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
+		Units: 6,
 	}
-	if err := rlsV1Rpc.Call("ResourceSV1.AllowUsage", attrRU, &allowed); err != nil {
+	if err := rlsV1Rpc.Call("ResourceSV1.AllowUsage", argsRU, &allowed); err != nil {
 		t.Error(err)
-	} else if !allowed {
-		t.Errorf("Expecting: %+v, received: %+v", true, allowed)
+	} else if !allowed { // already 3 usages active before allow call, we should have now more than allowed
+		t.Error("resource is not allowed")
 	}
-
-	attrRU = utils.ArgRSv1ResourceUsage{
-		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e51",
-		Event:   map[string]interface{}{"Account": "1002", "Subject": "1001", "Destination": "1002"},
-		Units:   2,
+	argsRU = utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
+		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e61",
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
+		Units: 7,
 	}
-	if err := rlsV1Rpc.Call("ResourceSV1.AllowUsage", attrRU, &allowed); err != nil { // already
+	if err := rlsV1Rpc.Call("ResourceSV1.AllowUsage", argsRU, &allowed); err != nil {
 		t.Error(err)
 	} else if allowed { // already 3 usages active before allow call, we should have now more than allowed
-		t.Error("Resource allowed")
+		t.Error("resource should not be allowed")
 	}
 }
 
 func testV1RsReleaseResource(t *testing.T) {
+	// relase the only resource active for Resource1
 	var reply string
-	attrRU := utils.ArgRSv1ResourceUsage{
-		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e52",
-		Event:   map[string]interface{}{"Destination": "100"},
-		Units:   2,
+	argsRU := utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
+		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e55", // same ID should be accepted by first group since the previous resource should be expired
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
 	}
-	if err := rlsV1Rpc.Call("ResourceSV1.ReleaseResource", attrRU, &reply); err != nil {
+	if err := rlsV1Rpc.Call("ResourceSV1.ReleaseResource", argsRU, &reply); err != nil {
 		t.Error(err)
 	}
+	// try reserving with full units for Resource1, case which did not work in previous test
+	// only match Resource1 since we don't want for storing of the resource2 bellow
+	argsRU = utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
+		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e61",
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "2002"},
+		Units: 7,
+	}
 	var allowed bool
-	if err := rlsV1Rpc.Call("ResourceSV1.AllowUsage", attrRU, &allowed); err != nil {
+	if err := rlsV1Rpc.Call("ResourceSV1.AllowUsage", argsRU, &allowed); err != nil {
 		t.Error(err)
 	} else if !allowed {
-		t.Error("not allowed")
+		t.Error("resource should be allowed")
 	}
-	attrRU.Units += 7
-	if err := rlsV1Rpc.Call("ResourceSV1.AllowUsage", attrRU, &allowed); err != nil {
+	var rs *engine.Resources
+	args := &utils.ArgRSv1ResourceUsage{
+		Tenant: "cgrates.org",
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"}}
+	if err := rlsV1Rpc.Call("ResourceSV1.GetResourcesForEvent", args, &rs); err != nil {
 		t.Error(err)
-	} else if allowed {
-		t.Error("Resource should not be allowed")
+	} else if len(*rs) != 2 {
+		t.Errorf("Resources: %+v", rs)
+	}
+	// make sure Resource1 have no more active resources
+	for _, r := range *rs {
+		if r.ID == "ResGroup1" &&
+			(len(r.Usages) != 0 || len(r.TTLIdx) != 0) {
+			t.Errorf("Unexpected resource: %+v", r)
+		}
+	}
+}
+
+func testV1RsDBStore(t *testing.T) {
+	// Save one event in Resource1
+	argsRU := utils.ArgRSv1ResourceUsage{
+		Tenant:  "cgrates.org",
+		UsageID: "651a8db2-4f67-4cf8-b622-169e8a482e71",
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"},
+		Units: 1,
+	}
+	var reply string
+	if err := rlsV1Rpc.Call("ResourceSV1.AllocateResource", argsRU, &reply); err != nil {
+		t.Error(err)
+	}
+	var rs *engine.Resources
+	args := &utils.ArgRSv1ResourceUsage{
+		Tenant: "cgrates.org",
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"}}
+	if err := rlsV1Rpc.Call("ResourceSV1.GetResourcesForEvent", args, &rs); err != nil {
+		t.Error(err)
+	} else if len(*rs) != 2 {
+		t.Errorf("Resources: %+v", rs)
+	}
+	// count resources before restart
+	for _, r := range *rs {
+		switch r.ID {
+		case "ResGroup1":
+			if len(r.Usages) != 1 || len(r.TTLIdx) != 1 {
+				t.Errorf("Unexpected resource: %+v", r)
+			}
+		case "ResGroup2":
+			if len(r.Usages) != 4 || len(r.TTLIdx) != 4 {
+				t.Errorf("Unexpected resource: %+v", r)
+			}
+		}
+	}
+	if _, err := engine.StopStartEngine(rlsV1CfgPath, resDelay); err != nil {
+		t.Fatal(err)
+	}
+	var err error
+	rlsV1Rpc, err = jsonrpc.Dial("tcp", rlsV1Cfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
+	if err != nil {
+		t.Fatal("Could not connect to rater: ", err.Error())
+	}
+	rs = new(engine.Resources)
+	args = &utils.ArgRSv1ResourceUsage{
+		Tenant: "cgrates.org",
+		Event: map[string]interface{}{
+			"Account":     "1002",
+			"Subject":     "1001",
+			"Destination": "1002"}}
+	if err := rlsV1Rpc.Call("ResourceSV1.GetResourcesForEvent", args, &rs); err != nil {
+		t.Error(err)
+	} else if len(*rs) != 2 {
+		t.Errorf("Resources: %+v", rs)
+	}
+	// count resources after restart
+	for _, r := range *rs {
+		switch r.ID {
+		case "ResGroup1":
+			if len(r.Usages) != 0 || len(r.TTLIdx) != 0 {
+				t.Errorf("Unexpected resource: %+v", r)
+			}
+		case "ResGroup2":
+			if len(r.Usages) != 3 || len(r.TTLIdx) != 3 {
+				t.Errorf("Unexpected resource: %+v", r)
+			}
+		}
 	}
 }
 

data/conf/samples/tutmongo/cgrates.json

@@ -115,6 +115,7 @@
 
 "resources": {
 	"enabled": true,
+	"store_interval": "1s",
 },
 
 

data/conf/samples/tutmysql/cgrates.json

@@ -108,6 +108,7 @@
 
 "resources": {
 	"enabled": true,
+	"store_interval": "1s",
 },
 
 

data/conf/samples/tutpostgres/cgrates.json

@@ -72,6 +72,7 @@
 
 "resources": {
 	"enabled": true,
+	"store_interval": "1s",
 },
 
 

data/tariffplans/tutorial/Resources.csv

@@ -1,8 +1,8 @@
 #Tenant[0],Id[1],FilterType[2],FilterFieldName[3],FilterFieldValues[4],ActivationInterval[5],TTL[6],Limit[7],AllocationMessage[8],Blocker[9],Stored[10],Weight[11],Thresholds[12]
-cgrates.org,ResGroup1,*string,Account,1001;1002,2014-07-29T15:00:00Z,1s,7,,true,true,20,
+cgrates.org,ResGroup1,*string,Account,1001;1002,2014-07-29T15:00:00Z,1s,7,,false,false,20,
 cgrates.org,ResGroup1,*string_prefix,Destination,10;20,,,,,,,,
 cgrates.org,ResGroup1,*rsr_fields,,Subject(~^1.*1$);Destination(1002),,,,,,,,
-cgrates.org,ResGroup2,*destinations,Destination,DST_FS,2014-07-29T15:00:00Z,3600s,8,SPECIAL_1002,true,true,10,
-cgrates.org,ResGroup3,*string,Account,3001,2014-07-29T15:00:00Z,1s,3,,true,true,20,
+cgrates.org,ResGroup2,*destinations,Destination,DST_FS,2014-07-29T15:00:00Z,3600s,8,SPECIAL_1002,false,true,10,
+cgrates.org,ResGroup3,*string,Account,3001,2014-07-29T15:00:00Z,0s,1,,true,false,20,
 #cgrates.org,ResGroup3,*timings,SetupTime,PEAK,,,,,,,,
 #cgrates.org,ResGroup3,*stats,,CDRST1:*min_ASR:34;CDRST_1001:*min_ASR:20,,,,,,,,

engine/resources.go

@@ -125,6 +125,7 @@ func (r *Resource) removeExpiredUnits() {
 		}
 	}
 	r.TTLIdx = r.TTLIdx[firstActive:]
+	r.tUsage = nil
 }
 
 // totalUsage returns the sum of all usage units
@@ -236,10 +237,10 @@ func (rs Resources) tenatIDsStr() []string {
 	return ids
 }
 
-// AllocateResource attempts allocating resources for a *ResourceUsage
+// allocateResource attempts allocating resources for a *ResourceUsage
 // simulates on dryRun
 // returns utils.ErrResourceUnavailable if allocation is not possible
-func (rs Resources) AllocateResource(ru *ResourceUsage, dryRun bool) (alcMessage string, err error) {
+func (rs Resources) allocateResource(ru *ResourceUsage, dryRun bool) (alcMessage string, err error) {
 	if len(rs) == 0 {
 		return "", utils.ErrResourceUnavailable
 	}
@@ -248,6 +249,7 @@ func (rs Resources) AllocateResource(ru *ResourceUsage, dryRun bool) (alcMessage
 	defer guardian.Guardian.UnguardIDs(lockIDs...)
 	// Simulate resource usage
 	for _, r := range rs {
+		r.removeExpiredUnits()
 		if r.rPrf.Limit >= r.totalUsage()+ru.Units {
 			if alcMessage == "" {
 				if r.rPrf.AllocationMessage != "" {
@@ -363,10 +365,12 @@ func (rS *ResourceService) runBackup() {
 		select {
 		case <-rS.stopBackup:
 			return
+		default:
 		}
 		rS.storeResources()
+		time.Sleep(rS.storeInterval)
 	}
-	time.Sleep(rS.storeInterval)
+
 }
 
 // cachedResourcesForEvent attempts to retrieve cached resources for an event
@@ -454,7 +458,7 @@ func (rS *ResourceService) matchingResourcesForEvent(tenant string, ev map[strin
 		if err != nil {
 			return nil, err
 		}
-		if rPrf.Stored {
+		if rPrf.Stored && r.dirty == nil {
 			r.dirty = utils.BoolPointer(false)
 		}
 		if rPrf.UsageTTL > 0 {
@@ -481,7 +485,7 @@ func (rS *ResourceService) matchingResourcesForEvent(tenant string, ev map[strin
 }
 
 // V1ResourcesForEvent returns active resource configs matching the event
-func (rS *ResourceService) V1ResourcesForEvent(args utils.ArgRSv1ResourceUsage, reply *[]*ResourceProfile) (err error) {
+func (rS *ResourceService) V1ResourcesForEvent(args utils.ArgRSv1ResourceUsage, reply *Resources) (err error) {
 	if args.Tenant == "" {
 		return utils.NewErrMandatoryIeMissing("Tenant")
 	}
@@ -498,10 +502,8 @@ func (rS *ResourceService) V1ResourcesForEvent(args utils.ArgRSv1ResourceUsage,
 	if len(mtcRLs) == 0 {
 		return utils.ErrNotFound
 	}
-	for _, r := range mtcRLs {
-		*reply = append(*reply, r.rPrf)
-	}
-	return nil
+	*reply = mtcRLs
+	return
 }
 
 // V1AllowUsage queries service to find if an Usage is allowed
@@ -516,7 +518,7 @@ func (rS *ResourceService) V1AllowUsage(args utils.ArgRSv1ResourceUsage, allow *
 		}
 		cache.Set(utils.EventResourcesPrefix+args.TenantID(), mtcRLs.tenantIDs(), true, "")
 	}
-	if _, err = mtcRLs.AllocateResource(
+	if _, err = mtcRLs.allocateResource(
 		&ResourceUsage{
 			Tenant: args.Tenant,
 			ID:     args.UsageID,
@@ -546,7 +548,8 @@ func (rS *ResourceService) V1AllocateResource(args utils.ArgRSv1ResourceUsage, r
 	} else {
 		wasCached = true
 	}
-	alcMsg, err := mtcRLs.AllocateResource(&ResourceUsage{ID: args.UsageID, Units: args.Units}, false)
+	alcMsg, err := mtcRLs.allocateResource(
+		&ResourceUsage{Tenant: args.Tenant, ID: args.UsageID, Units: args.Units}, false)
 	if err != nil {
 		return err
 	}
@@ -572,8 +575,8 @@ func (rS *ResourceService) V1AllocateResource(args utils.ArgRSv1ResourceUsage, r
 			rS.StoreResource(r)
 		} else if r.dirty != nil {
 			*r.dirty = true // mark it to be saved
+			rS.storedResources[r.TenantID()] = true
 		}
-		rS.storedResources[r.ID] = true
 	}
 	rS.srMux.Unlock()
 	*reply = alcMsg
@@ -604,7 +607,7 @@ func (rS *ResourceService) V1ReleaseResource(args utils.ArgRSv1ResourceUsage, re
 				rS.StoreResource(r)
 			} else {
 				*r.dirty = true // mark it to be saved
-				rS.storedResources[r.ID] = true
+				rS.storedResources[r.TenantID()] = true
 			}
 		}
 	}