//go:build performance /* 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 Affero 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 Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see */ package agents import ( "bytes" "flag" "fmt" "path/filepath" "sync/atomic" "testing" "time" "github.com/cgrates/birpc/context" "github.com/cgrates/cgrates/engine" "github.com/cgrates/cgrates/utils" ) var ( capsLimit = flag.Int("caps_limit", 0, "caps limit") capsStrategy = flag.String("caps_strategy", "*busy", "caps strategy") parallelism = flag.Int("parallelism", 0, "parallelism") ) func BenchmarkDiameterCaps(b *testing.B) { // CoreS config is dynamic for this benchmark. jsonCfg := fmt.Sprintf(`{ "cores": { "caps": %d, "caps_strategy": "%s", // use shutdown_timeout option to set the diameter reply timeout. "shutdown_timeout": "100ms" } }`, *capsLimit, *capsStrategy) ng := engine.TestEngine{ ConfigJSON: jsonCfg, ConfigPath: filepath.Join(*utils.DataDir, "conf", "samples", "diambench"), TpFiles: map[string]string{ utils.AccountActionsCsv: `#Tenant,Account,ActionPlanId,ActionTriggersId,AllowNegative,Disabled cgrates.org,1001,PACKAGE_1001,,,`, utils.ActionPlansCsv: `#Id,ActionsId,TimingId,Weight PACKAGE_1001,ACT_TOPUP,*asap,10`, utils.ActionsCsv: `#ActionsId[0],Action[1],ExtraParameters[2],Filter[3],BalanceId[4],BalanceType[5],Categories[6],DestinationIds[7],RatingSubject[8],SharedGroup[9],ExpiryTime[10],TimingIds[11],Units[12],BalanceWeight[13],BalanceBlocker[14],BalanceDisabled[15],Weight[16] ACT_TOPUP,*topup_reset,,,balance_sms,*sms,,,,,*unlimited,,1000000,,,,`, utils.ChargersCsv: `#Tenant,ID,FilterIDs,ActivationInterval,RunID,AttributeIDs,Weight cgrates.org,DEFAULT,,,*default,*none,0`, }, DBCfg: engine.InternalDBCfg, } client, cfg := ng.Run(b) time.Sleep(10 * time.Millisecond) // wait for DiameterAgent service to start diamClient, err := NewDiameterClient(cfg.DiameterAgentCfg().Listeners[0].Address, "localhost", cfg.DiameterAgentCfg().OriginRealm, cfg.DiameterAgentCfg().VendorID, cfg.DiameterAgentCfg().ProductName, utils.DiameterFirmwareRevision, cfg.DiameterAgentCfg().DictionariesPath, cfg.DiameterAgentCfg().Listeners[0].Network) if err != nil { b.Fatal(err) } // actual benchmark var sent, completed atomic.Int64 b.SetParallelism(*parallelism) b.ResetTimer() b.RunParallel(func(pb *testing.PB) { for pb.Next() { sent.Add(1) if sendDiamCCR(b, diamClient, cfg.CoreSCfg().ShutdownTimeout, "2001") { completed.Add(1) } } }) // check results b.Logf("sent %d, completed %d", sent.Load(), completed.Load()) var acnt *engine.Account attrsAcnt := &utils.AttrGetAccount{ Tenant: "cgrates.org", Account: "1001", } expBalance := float64(1000000 - completed.Load()) if err = client.Call(context.Background(), utils.APIerSv2GetAccount, attrsAcnt, &acnt); err != nil { b.Errorf("APIerSv1.GetAccount unexpected err: %v", err) } else if rply := acnt.BalanceMap[utils.MetaSMS].GetTotalValue(); rply != expBalance { b.Errorf("APIerSv1.GetAccount: sms_balance: %f, want: %f", rply, expBalance) } } func BenchmarkDiameterReplication(b *testing.B) { // Setup for storage engine. rplNgJSONCfg := `{ "listen": { "rpc_json": "127.0.0.1:22012", "rpc_gob": "127.0.0.1:22013", "http": "127.0.0.1:22080", }, "data_db": { "db_host": "192.168.122.164" }, "apiers": { "enabled": true, "scheduler_conns": ["*internal"] }, "schedulers": { "enabled": true } }` rplNg := engine.TestEngine{ TpFiles: map[string]string{ utils.AccountActionsCsv: `#Tenant,Account,ActionPlanId,ActionTriggersId,AllowNegative,Disabled cgrates.org,1001,PACKAGE_1001,,,`, utils.ActionPlansCsv: `#Id,ActionsId,TimingId,Weight PACKAGE_1001,ACT_TOPUP,*asap,10`, utils.ActionsCsv: `#ActionsId[0],Action[1],ExtraParameters[2],Filter[3],BalanceId[4],BalanceType[5],Categories[6],DestinationIds[7],RatingSubject[8],SharedGroup[9],ExpiryTime[10],TimingIds[11],Units[12],BalanceWeight[13],BalanceBlocker[14],BalanceDisabled[15],Weight[16] ACT_TOPUP,*topup_reset,,,balance_sms,*sms,,,,,*unlimited,,1000000,,,,`, utils.ChargersCsv: `#Tenant,ID,FilterIDs,ActivationInterval,RunID,AttributeIDs,Weight cgrates.org,DEFAULT,*string:~*req.Account:1001,,*default,*none,10 cgrates.org,Raw,,,*raw,*constant:*req.RequestType:*none,0`, }, ConfigJSON: rplNgJSONCfg, DBCfg: engine.DBCfg{ DataDB: &engine.DBParams{ Host: utils.StringPointer("192.168.122.164"), }, StorDB: engine.InternalDBCfg.StorDB, }, LogBuffer: new(bytes.Buffer), } // defer fmt.Println(rplNg.LogBuffer) rplClient, _ := rplNg.Run(b) // Setup for rater. rplInterval := 5 * time.Millisecond failedDir := b.TempDir() jsonCfg := fmt.Sprintf(`{ "data_db": { "db_type": "*internal", "remote_conns": ["db_conn"], "replication_conns": ["db_conn"], "replication_filtered": false, "replication_cache": "", "replication_failed_dir": "%s", "replication_interval": "%s", "items":{ "*accounts": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": true}, "*reverse_destinations": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*destinations": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*rating_plans": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*rating_profiles": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*actions": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*action_plans": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*account_action_plans": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*action_triggers": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*shared_groups": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*timings": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*filters": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*attribute_profiles": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*charger_profiles": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*load_ids": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*versions": {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*attribute_filter_indexes" : {"limit": -1, "ttl": "", "static_ttl": false, "remote": false, "replicate": false}, "*charger_filter_indexes" : {"limit": -1, "ttl": "", "static_ttl": false, "remote": true, "replicate": false}, "*reverse_filter_indexes" : {"limit": -1, "ttl": "", "static_ttl": false, "remote": false, "replicate": false}, } }, "rpc_conns": { "db_conn": { "conns": [ { "address": "127.0.0.1:22013", "transport":"*gob", "connect_attempts": 5, "reconnects": 5, "max_reconnect_interval": "", "connect_timeout":"1s", "reply_timeout":"2s" } ] } } }`, failedDir, rplInterval) ng := engine.TestEngine{ ConfigJSON: jsonCfg, ConfigPath: filepath.Join(*utils.DataDir, "conf", "samples", "diambench"), LogBuffer: new(bytes.Buffer), } // defer fmt.Println(ng.LogBuffer) client, cfg := ng.Run(b) time.Sleep(50 * time.Millisecond) // wait for DiameterAgent service to start diamClient, err := NewDiameterClient(cfg.DiameterAgentCfg().Listeners[0].Address, "localhost", cfg.DiameterAgentCfg().OriginRealm, cfg.DiameterAgentCfg().VendorID, cfg.DiameterAgentCfg().ProductName, utils.DiameterFirmwareRevision, cfg.DiameterAgentCfg().DictionariesPath, cfg.DiameterAgentCfg().Listeners[0].Network) if err != nil { b.Fatal(err) } // actual benchmark var sent, completed atomic.Int64 b.SetParallelism(*parallelism) b.ResetTimer() b.RunParallel(func(pb *testing.PB) { for pb.Next() { sent.Add(1) if sendDiamCCR(b, diamClient, 5*time.Second, "2001") { completed.Add(1) } } }) // Wait for last batch of replications to execute. time.Sleep(rplInterval) // check results b.Logf("sent %d, completed %d", sent.Load(), completed.Load()) var acnt *engine.Account attrsAcnt := &utils.AttrGetAccount{ Tenant: "cgrates.org", Account: "1001", } expBalance := float64(1000000 - completed.Load()) if err = client.Call(context.Background(), utils.APIerSv2GetAccount, attrsAcnt, &acnt); err != nil { b.Errorf("APIerSv1.GetAccount unexpected err: %v", err) } else if rply := acnt.BalanceMap[utils.MetaSMS].GetTotalValue(); rply != expBalance { b.Errorf("APIerSv1.GetAccount: sms_balance: %f, want: %f", rply, expBalance) } if err = rplClient.Call(context.Background(), utils.APIerSv2GetAccount, attrsAcnt, &acnt); err != nil { b.Errorf("APIerSv1.GetAccount unexpected err: %v", err) } else if rply := acnt.BalanceMap[utils.MetaSMS].GetTotalValue(); rply != expBalance { b.Errorf("APIerSv1.GetAccount: sms_balance: %f, want: %f", rply, expBalance) } }