diff --git a/general_tests/cdr_compressed_it_test.go b/general_tests/cdr_compressed_it_test.go
index d7ba72274..e18825bc9 100644
--- a/general_tests/cdr_compressed_it_test.go
+++ b/general_tests/cdr_compressed_it_test.go
@@ -21,87 +21,19 @@ along with this program. If not, see
package general_tests
import (
+ "fmt"
"path"
"testing"
"time"
+ "github.com/cgrates/birpc"
"github.com/cgrates/birpc/context"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
func TestCDRCompressed(t *testing.T) {
- var dbCfg engine.DBCfg
- switch *utils.DBType {
- case utils.MetaInternal:
- dbCfg = engine.DBCfg{
- StorDB: &engine.DBParams{
- Type: utils.StringPointer("*internal"),
- },
- }
- case utils.MetaMySQL:
- dbCfg = engine.DBCfg{
- StorDB: &engine.DBParams{
- Type: utils.StringPointer("*mysql"),
- Password: utils.StringPointer("CGRateS.org"),
- },
- }
- case utils.MetaMongo:
- dbCfg = engine.DBCfg{
- StorDB: &engine.DBParams{
- Type: utils.StringPointer("*mongo"),
- Name: utils.StringPointer("cgrates"),
- Port: utils.IntPointer(27017),
- },
- }
- case utils.MetaPostgres:
- t.SkipNow()
- default:
- t.Fatal("unsupported dbtype value")
- }
-
- content := `{
-
- "data_db": {
- "db_type": "*internal"
- },
-
- "stor_db": {
- "db_type": "*internal"
- },
-
- "attributes":{
- "enabled": true,
- "indexed_selects": false,
- },
-
- "rals": {
- "enabled": true,
- },
-
- "cdrs": {
- "enabled": true,
- "rals_conns": ["*internal"],
- "compress_stored_cost":true,
- },
-
- "schedulers": {
- "enabled": true
- },
-
- "apiers": {
- "enabled": true,
- "scheduler_conns": ["*internal"]
- }
-
- }`
-
- ng := engine.TestEngine{
- ConfigJSON: content,
- DBCfg: dbCfg,
- TpPath: path.Join(*utils.DataDir, "tariffplans", "reratecdrs"),
- }
- client, _ := ng.Run(t)
+ client := newTestClient(t, true)
t.Run("ProcessAndSetCDR", func(t *testing.T) {
var reply string
err := client.Call(context.Background(), utils.CDRsV1ProcessEvent,
@@ -142,3 +74,119 @@ func TestCDRCompressed(t *testing.T) {
}
})
}
+
+func newTestClient(t testing.TB, compress bool) *birpc.Client {
+ // set up DBCfg exactly as in your TestCDRCompressed
+ var dbCfg engine.DBCfg
+ switch *utils.DBType {
+ case utils.MetaInternal:
+ dbCfg = engine.DBCfg{StorDB: &engine.DBParams{Type: utils.StringPointer("*internal")}}
+ case utils.MetaMySQL:
+ dbCfg = engine.DBCfg{StorDB: &engine.DBParams{
+ Type: utils.StringPointer("*mysql"),
+ Password: utils.StringPointer("CGRateS.org"),
+ }}
+ case utils.MetaMongo:
+ dbCfg = engine.DBCfg{StorDB: &engine.DBParams{
+ Type: utils.StringPointer("*mongo"),
+ Name: utils.StringPointer("cgrates"),
+ Port: utils.IntPointer(27017),
+ }}
+ case utils.MetaPostgres:
+ t.SkipNow()
+ default:
+ t.Fatalf("unsupported dbtype %v", *utils.DBType)
+ }
+
+ content := fmt.Sprintf(`{
+
+ "data_db": {
+ "db_type": "*internal"
+ },
+
+ "stor_db": {
+ "db_type": "*internal"
+ },
+
+ "attributes":{
+ "enabled": true,
+ "indexed_selects": false,
+ },
+
+ "rals": {
+ "enabled": true,
+ },
+
+ "cdrs": {
+ "enabled": true,
+ "rals_conns": ["*internal"],
+ "compress_stored_cost": %t,
+ },
+
+ "schedulers": {
+ "enabled": true
+ },
+
+ "apiers": {
+ "enabled": true,
+ "scheduler_conns": ["*internal"]
+ }
+
+ }`, compress)
+ cfgJSON := fmt.Sprintf(content, compress)
+ ng := engine.TestEngine{
+ ConfigJSON: cfgJSON,
+ DBCfg: dbCfg,
+ TpPath: path.Join(*utils.DataDir, "tariffplans", "reratecdrs"),
+ }
+ client, _ := ng.Run(t)
+ return client
+}
+
+func benchmarkProcessCDR(b *testing.B, compress bool) {
+ b.ReportAllocs()
+ client := newTestClient(b, compress)
+
+ for b.Loop() {
+ // pre-build the RPC arg
+ arg := &engine.ArgV1ProcessEvent{
+ Flags: []string{utils.MetaRALs},
+ CGREvent: utils.CGREvent{
+ Tenant: "cgrates.org",
+ ID: "eventX",
+ Event: map[string]any{
+ utils.RunID: "run",
+ utils.CGRID: utils.GenUUID(),
+ utils.Tenant: "cgrates.org",
+ utils.Category: "call",
+ utils.ToR: utils.MetaVoice,
+ utils.OriginID: "bench",
+ utils.OriginHost: "bench-host",
+ utils.RequestType: utils.MetaRated,
+ utils.AccountField: "1001",
+ utils.Destination: "1002",
+ utils.SetupTime: time.Date(2021, time.February, 2, 16, 14, 50, 0, time.UTC),
+ utils.AnswerTime: time.Date(2021, time.February, 2, 16, 15, 0, 0, time.UTC),
+ utils.Usage: 2 * time.Minute,
+ },
+ },
+ }
+ var reply string
+ if err := client.Call(context.Background(), utils.CDRsV1ProcessEvent, arg, &reply); err != nil {
+ b.Fatalf("ProcessEvent failed: %v", err)
+ }
+ var cdrs []*engine.CDR
+ if err := client.Call(context.Background(), utils.CDRsV1GetCDRs, &utils.RPCCDRsFilterWithAPIOpts{
+ RPCCDRsFilter: &utils.RPCCDRsFilter{}}, &cdrs); err != nil {
+ b.Fatalf("GetCDRs failed: %v", err)
+ }
+ }
+}
+
+func BenchmarkCDRCompressed(b *testing.B) {
+ benchmarkProcessCDR(b, true)
+}
+
+func BenchmarkCDRUncompressed(b *testing.B) {
+ benchmarkProcessCDR(b, false)
+}
diff --git a/general_tests/resources_cps_it_test.go b/general_tests/resources_cps_it_test.go
new file mode 100644
index 000000000..ebbb68620
--- /dev/null
+++ b/general_tests/resources_cps_it_test.go
@@ -0,0 +1,163 @@
+//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 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
+*/
+
+package general_tests
+
+import (
+ "bytes"
+ "fmt"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/cgrates/birpc/context"
+ "github.com/cgrates/cgrates/engine"
+ "github.com/cgrates/cgrates/utils"
+)
+
+func TestStressResourceProcessEvent(t *testing.T) {
+ var dbConfig engine.DBCfg
+ switch *utils.DBType {
+ case utils.MetaInternal:
+ dbConfig = engine.DBCfg{
+ DataDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ StorDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ }
+ case utils.MetaMySQL:
+ case utils.MetaMongo, utils.MetaPostgres:
+ t.SkipNow()
+ default:
+ t.Fatal("unsupported dbtype value")
+ }
+ content := `{
+
+ "general": {
+ "log_level": 7,
+ },
+ "data_db": {
+ "db_type": "redis",
+ "db_port": 6379,
+ "db_name": "10",
+ },
+ "stor_db": {
+ "db_password": "CGRateS.org",
+ },
+ "apiers": {
+ "enabled": true
+ },
+ "stats": {
+ "enabled": true,
+ "store_interval": "-1",
+ },
+ "resources": {
+ "enabled": true,
+ "store_interval": "-1",
+ "thresholds_conns": ["*internal"]
+ },
+ "thresholds": {
+ "enabled": true,
+ "store_interval": "-1",
+ },
+ }`
+
+ ng := engine.TestEngine{
+ ConfigJSON: content,
+ DBCfg: dbConfig,
+ LogBuffer: bytes.NewBuffer(nil),
+ }
+ client, _ := ng.Run(t)
+ t.Run("SetResourceProfile", func(t *testing.T) {
+ var result string
+ for i := 1; i <= 10; i++ {
+ rls := &engine.ResourceProfileWithAPIOpts{
+ ResourceProfile: &engine.ResourceProfile{
+ Tenant: "cgrates.org",
+ ID: fmt.Sprintf("RES_%d", i),
+ FilterIDs: []string{fmt.Sprintf("*string:~*req.Account:100%d", i)},
+ UsageTTL: -1,
+ Limit: -1,
+ AllocationMessage: "Account1Channels",
+ Weight: 20,
+ ThresholdIDs: []string{utils.MetaNone},
+ },
+ }
+ if err := client.Call(context.Background(), utils.APIerSv1SetResourceProfile, rls, &result); err != nil {
+ t.Error(err)
+ } else if result != utils.OK {
+ t.Error("Unexpected reply returned", result)
+ }
+ }
+ })
+
+ t.Run("StatExportEvent", func(t *testing.T) {
+ var wg sync.WaitGroup
+ start := time.Now()
+ var sucessEvent atomic.Int32
+ errCH := make(chan error, *count)
+ for i := 1; i <= *count; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ args := &utils.CGREvent{
+ Tenant: "cgrates.org",
+ ID: utils.UUIDSha1Prefix(),
+ Event: map[string]any{
+ utils.AccountField: fmt.Sprintf("100%d", ((i-1)%10)+1),
+ utils.Destination: "3420340",
+ },
+ APIOpts: map[string]any{
+ utils.OptsResourcesUsageID: "651a8db2-4f67-4cf8-b622-169e8a482e45",
+ utils.OptsResourcesUnits: 6,
+ },
+ }
+ var reply string
+ if err := client.Call(context.Background(), utils.ResourceSv1AllocateResources,
+ args, &reply); err != nil {
+ errCH <- fmt.Errorf("no loop %d event id %s failed with err: %v", i, args.Event[utils.AccountField], err)
+ return
+ } else if reply != "Account1Channels" {
+ t.Error("Unexpected reply returned", reply)
+ }
+ sucessEvent.Add(1)
+ }(i)
+ }
+ doneCH := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCH)
+ }()
+ select {
+ case <-doneCH:
+ case <-time.After(10 * time.Minute):
+ t.Error("timeout")
+ }
+ close(errCH)
+ for err := range errCH {
+ t.Error(err)
+ }
+ t.Logf("Processed %v events in %v (%.2f events/sec)", sucessEvent.Load(), time.Since(start), float64(*count)/time.Since(start).Seconds())
+ })
+
+}
diff --git a/general_tests/stat_cps_it_test.go b/general_tests/stat_cps_it_test.go
new file mode 100644
index 000000000..9d4304658
--- /dev/null
+++ b/general_tests/stat_cps_it_test.go
@@ -0,0 +1,239 @@
+//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 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
+*/
+
+package general_tests
+
+import (
+ "bytes"
+ "flag"
+ "fmt"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/cgrates/birpc/context"
+ "github.com/cgrates/cgrates/engine"
+ "github.com/cgrates/cgrates/utils"
+)
+
+var (
+ count = flag.Int("no_events", 1000, "no_events")
+)
+
+func TestStressStatsProcessEvent(t *testing.T) {
+ var dbConfig engine.DBCfg
+ switch *utils.DBType {
+ case utils.MetaInternal:
+ dbConfig = engine.DBCfg{
+ DataDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ StorDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ }
+ case utils.MetaMySQL:
+ case utils.MetaMongo, utils.MetaPostgres:
+ t.SkipNow()
+ default:
+ t.Fatal("unsupported dbtype value")
+ }
+ content := `{
+
+ "general": {
+ "log_level": 7,
+ },
+ "data_db": {
+ "db_type": "redis",
+ "db_port": 6379,
+ "db_name": "10",
+ },
+ "stor_db": {
+ "db_password": "CGRateS.org",
+ },
+ "apiers": {
+ "enabled": true
+ },
+
+ "stats": {
+ "enabled": true,
+ "store_interval": "-1",
+ },
+
+ }`
+
+ ng := engine.TestEngine{
+ ConfigJSON: content,
+ DBCfg: dbConfig,
+ LogBuffer: bytes.NewBuffer(nil),
+ }
+ client, _ := ng.Run(t)
+ t.Run("SetStatProfile", func(t *testing.T) {
+ var reply string
+ for i := 1; i <= 10; i++ {
+ statConfig := &engine.StatQueueProfileWithAPIOpts{
+ StatQueueProfile: &engine.StatQueueProfile{
+ Tenant: "cgrates.org",
+ ID: fmt.Sprintf("STAT_%v", i),
+ FilterIDs: []string{fmt.Sprintf("*string:~*req.Account:100%d", i)},
+ QueueLength: -1,
+ TTL: 1 * time.Hour,
+ Metrics: []*engine.MetricWithFilters{
+ {
+ MetricID: utils.MetaTCD,
+ },
+ {
+ MetricID: utils.MetaASR,
+ },
+ {
+ MetricID: utils.MetaACC,
+ },
+ {
+ MetricID: "*sum#~*req.Usage",
+ },
+ },
+ Blocker: true,
+ Stored: true,
+ Weight: 20,
+ },
+ }
+ if err := client.Call(context.Background(), utils.APIerSv1SetStatQueueProfile, statConfig, &reply); err != nil {
+ t.Error(err)
+ }
+ }
+ })
+
+ t.Run("StatExportEvent", func(t *testing.T) {
+ var wg sync.WaitGroup
+ start := time.Now()
+ var sucessEvent atomic.Int32
+ errCH := make(chan error, *count)
+ for i := 1; i <= *count; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ args := &engine.CGREventWithEeIDs{
+ CGREvent: &utils.CGREvent{
+ Tenant: "cgrates.org",
+ ID: "voiceEvent",
+ Time: utils.TimePointer(time.Now()),
+ Event: map[string]any{
+ utils.AccountField: fmt.Sprintf("100%d", ((i-1)%10)+1),
+ utils.AnswerTime: utils.TimePointer(time.Now()),
+ utils.Usage: 45,
+ utils.Cost: 12.1,
+ },
+ },
+ }
+ var reply []string
+ if err := client.Call(context.Background(), utils.StatSv1ProcessEvent, args, &reply); err != nil {
+ errCH <- fmt.Errorf("no loop %d event id %s failed with err: %v", i, args.Event[utils.AccountField], err)
+ return
+ }
+ sucessEvent.Add(1)
+ }(i)
+ }
+ doneCH := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCH)
+ }()
+ select {
+ case <-doneCH:
+ case <-time.After(10 * time.Minute):
+ t.Error("timeout")
+ }
+ close(errCH)
+ for err := range errCH {
+ t.Error(err)
+ }
+ t.Logf("Processed %v events in %v (%.2f events/sec)", sucessEvent.Load(), time.Since(start), float64(*count)/time.Since(start).Seconds())
+ })
+}
+
+func BenchmarkStatQueueCompress(b *testing.B) {
+ sizes := []int{1000, 10000, 50000}
+ origQueues := make(map[int][]engine.SQItem, len(sizes))
+ now := time.Now()
+ for _, size := range sizes {
+ items := make([]engine.SQItem, size)
+ for i := range size {
+ t := now.Add(time.Duration(i) * time.Second)
+ items[i] = engine.SQItem{EventID: fmt.Sprintf("e%d", i), ExpiryTime: &t}
+ }
+ origQueues[size] = items
+ }
+
+ for _, size := range sizes {
+ size := size
+ orig := origQueues[size]
+ b.Run(fmt.Sprintf("Compress_N=%d", size), func(b *testing.B) {
+ threshold := int64(size / 2)
+ round := 2
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ items := make([]engine.SQItem, len(orig))
+ copy(items, orig)
+ sq := &engine.StatQueue{
+ Tenant: "cgrates.org",
+ ID: fmt.Sprintf("Q-%d", size),
+ SQItems: items,
+ SQMetrics: make(map[string]engine.StatMetric),
+ }
+ sq.Compress(threshold, round)
+ }
+ })
+ }
+}
+
+func BenchmarkStatQueueExpand(b *testing.B) {
+ sizes := []int{1000, 10000, 50000}
+ origQueues := make(map[int][]engine.SQItem, len(sizes))
+ now := time.Now()
+ for _, size := range sizes {
+ items := make([]engine.SQItem, size/10)
+ for i := range items {
+ t := now
+ items[i] = engine.SQItem{EventID: fmt.Sprintf("e%d", i), ExpiryTime: &t}
+ }
+ origQueues[size] = items
+ }
+ for _, size := range sizes {
+ size := size
+ orig := origQueues[size]
+ b.Run(fmt.Sprintf("Expand_N=%d", size), func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ items := make([]engine.SQItem, len(orig))
+ copy(items, orig)
+ sq := &engine.StatQueue{
+ Tenant: "cgrates.org",
+ ID: fmt.Sprintf("Q-%d", size),
+ SQItems: items,
+ SQMetrics: make(map[string]engine.StatMetric),
+ }
+ sq.Expand()
+ }
+ })
+ }
+}
diff --git a/general_tests/thresholds_cps_it_test.go b/general_tests/thresholds_cps_it_test.go
new file mode 100644
index 000000000..cbef2a2c9
--- /dev/null
+++ b/general_tests/thresholds_cps_it_test.go
@@ -0,0 +1,154 @@
+//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 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
+*/
+
+package general_tests
+
+import (
+ "bytes"
+ "fmt"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/cgrates/birpc/context"
+ "github.com/cgrates/cgrates/engine"
+ "github.com/cgrates/cgrates/utils"
+)
+
+func TestStressThresholdsProcessEvent(t *testing.T) {
+ var dbConfig engine.DBCfg
+ switch *utils.DBType {
+ case utils.MetaInternal:
+ dbConfig = engine.DBCfg{
+ DataDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ StorDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ }
+ case utils.MetaMySQL:
+ case utils.MetaMongo, utils.MetaPostgres:
+ t.SkipNow()
+ default:
+ t.Fatal("unsupported dbtype value")
+ }
+ content := `{
+
+ "general": {
+ "log_level": 7,
+ },
+ "data_db": {
+ "db_type": "redis",
+ "db_port": 6379,
+ "db_name": "10",
+ },
+ "stor_db": {
+ "db_password": "CGRateS.org",
+ },
+ "apiers": {
+ "enabled": true
+ },
+ "thresholds": {
+ "enabled": true,
+ "store_interval": "-1",
+ },
+ }`
+
+ ng := engine.TestEngine{
+ ConfigJSON: content,
+ DBCfg: dbConfig,
+ LogBuffer: bytes.NewBuffer(nil),
+ }
+ defer fmt.Println(ng.LogBuffer)
+ client, _ := ng.Run(t)
+ t.Run("SetThresholdProfile", func(t *testing.T) {
+ var reply string
+ for i := 1; i <= 10; i++ {
+ thresholdPrf := &engine.ThresholdProfileWithAPIOpts{
+ ThresholdProfile: &engine.ThresholdProfile{
+ Tenant: "cgrates.org",
+ ID: fmt.Sprintf("THD_%d", i),
+ FilterIDs: []string{fmt.Sprintf("*string:~*req.Account:100%d", i)},
+ ActivationInterval: &utils.ActivationInterval{
+ ActivationTime: time.Date(2024, 7, 14, 14, 35, 0, 0, time.UTC),
+ },
+ MaxHits: -1,
+ Blocker: false,
+ Weight: 20.0,
+ Async: true,
+ },
+ }
+ if err := client.Call(context.Background(), utils.APIerSv1SetThresholdProfile, thresholdPrf, &reply); err != nil {
+ t.Error(err)
+ }
+ }
+ })
+
+ t.Run("ThresholdExportEvent", func(t *testing.T) {
+ var wg sync.WaitGroup
+ start := time.Now()
+ var sucessEvent atomic.Int32
+ errCH := make(chan error, *count)
+ for i := 1; i <= *count; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ args := &utils.CGREvent{
+ Tenant: "cgrates.org",
+ ID: utils.GenUUID(),
+ Time: utils.TimePointer(time.Now()),
+ Event: map[string]any{
+ utils.AccountField: fmt.Sprintf("100%d", ((i-1)%10)+1),
+ utils.AnswerTime: utils.TimePointer(time.Now()),
+ utils.Usage: 45,
+ utils.Cost: 12.1,
+ utils.Tenant: "cgrates.org",
+ utils.Category: "call",
+ },
+ }
+
+ var reply []string
+ if err := client.Call(context.Background(), utils.ThresholdSv1ProcessEvent, args, &reply); err != nil {
+ errCH <- fmt.Errorf("no loop %d event id %s failed with err: %v", i, args.Event[utils.AccountField], err)
+ return
+ }
+ sucessEvent.Add(1)
+ }(i)
+ }
+ doneCH := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCH)
+ }()
+ select {
+ case <-doneCH:
+ case <-time.After(10 * time.Minute):
+ t.Error("timeout")
+ }
+ close(errCH)
+ for err := range errCH {
+ t.Error(err)
+ }
+ t.Logf("Processed %v events in %v (%.2f events/sec)", sucessEvent.Load(), time.Since(start), float64(*count)/time.Since(start).Seconds())
+ })
+
+}
diff --git a/general_tests/trends_cps_it_test.go b/general_tests/trends_cps_it_test.go
new file mode 100644
index 000000000..4557e75d7
--- /dev/null
+++ b/general_tests/trends_cps_it_test.go
@@ -0,0 +1,226 @@
+//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 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
+*/
+
+package general_tests
+
+import (
+ "bytes"
+ "fmt"
+ "math/rand"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/cgrates/birpc/context"
+ "github.com/cgrates/cgrates/engine"
+ "github.com/cgrates/cgrates/utils"
+)
+
+func TestStressTrendsProcessEvent(t *testing.T) {
+ var dbConfig engine.DBCfg
+ switch *utils.DBType {
+ case utils.MetaInternal:
+ dbConfig = engine.DBCfg{
+ DataDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ StorDB: &engine.DBParams{
+ Type: utils.StringPointer(utils.MetaInternal),
+ },
+ }
+ case utils.MetaMySQL:
+ case utils.MetaMongo, utils.MetaPostgres:
+ t.SkipNow()
+ default:
+ t.Fatal("unsupported dbtype value")
+ }
+ content := `{
+
+ "general": {
+ "log_level": 7,
+ },
+ "data_db": {
+ "db_type": "redis",
+ "db_port": 6379,
+ "db_name": "10",
+ },
+ "stor_db": {
+ "db_password": "CGRateS.org",
+ },
+ "apiers": {
+ "enabled": true
+ },
+ "stats": {
+ "enabled": true,
+ "store_interval": "-1",
+ },
+ "trends": {
+ "enabled": true,
+ "store_interval": "-1",
+ "stats_conns":["*localhost"],
+ "thresholds_conns": [],
+ "ees_conns": [],
+ },
+ }`
+
+ ng := engine.TestEngine{
+ ConfigJSON: content,
+ DBCfg: dbConfig,
+ LogBuffer: bytes.NewBuffer(nil),
+ }
+ //defer fmt.Println(ng.LogBuffer)
+ client, _ := ng.Run(t)
+
+ numProfiles := 1000
+ t.Run("SetStatAndTrendProfiles", func(t *testing.T) {
+ var reply string
+ for i := 1; i <= numProfiles; i++ {
+ statConfig := &engine.StatQueueProfileWithAPIOpts{
+ StatQueueProfile: &engine.StatQueueProfile{
+ Tenant: "cgrates.org",
+ ID: fmt.Sprintf("STAT_%v", i),
+ FilterIDs: []string{fmt.Sprintf("*string:~*req.Account:100%d", i)},
+ QueueLength: -1,
+ TTL: 1 * time.Hour,
+ Metrics: []*engine.MetricWithFilters{
+ {
+ MetricID: utils.MetaTCD,
+ },
+ {
+ MetricID: utils.MetaASR,
+ },
+ {
+ MetricID: utils.MetaACC,
+ },
+ {
+ MetricID: "*sum#~*req.Usage",
+ },
+ },
+ Blocker: true,
+ Stored: true,
+ Weight: 20,
+ },
+ }
+ if err := client.Call(context.Background(), utils.APIerSv1SetStatQueueProfile, statConfig, &reply); err != nil {
+ t.Error(err)
+ }
+ trendProfile := &engine.TrendProfileWithAPIOpts{
+ TrendProfile: &engine.TrendProfile{
+ Tenant: "cgrates.org",
+ ID: fmt.Sprintf("Trend%d", i),
+ ThresholdIDs: []string{""},
+ CorrelationType: utils.MetaAverage,
+ TTL: -1,
+ QueueLength: -1,
+ StatID: fmt.Sprintf("STAT_%d", i),
+ Schedule: "@every 1s",
+ },
+ }
+ if err := client.Call(context.Background(), utils.APIerSv1SetTrendProfile, trendProfile, &reply); err != nil {
+ t.Error(err)
+ } else if reply != utils.OK {
+ t.Errorf("Expected: %v,Received: %v", utils.OK, reply)
+ }
+
+ var scheduled int
+ if err := client.Call(context.Background(), utils.TrendSv1ScheduleQueries,
+ &utils.ArgScheduleTrendQueries{TrendIDs: []string{trendProfile.ID}, TenantIDWithAPIOpts: utils.TenantIDWithAPIOpts{TenantID: &utils.TenantID{Tenant: "cgrates.org"}}}, &scheduled); err != nil {
+ t.Fatal(err)
+ } else if scheduled != 1 {
+ t.Errorf("expected 1, got %d", scheduled)
+ }
+
+ }
+ })
+ t.Run("ContinuousEventLoad", func(t *testing.T) {
+ var wg sync.WaitGroup
+ errCh := make(chan error, numProfiles*10)
+ stopCh := make(chan struct{}) // To stop event generation after duration
+
+ // Run event generation for 30 seconds
+ go func() {
+ time.Sleep(30 * time.Second)
+ close(stopCh)
+ }()
+
+ start := time.Now()
+ var evcount atomic.Int64
+ for i := 0; i < 50; i++ { // 50 concurrent workers
+ wg.Add(1)
+ go func(workerID int) {
+ defer wg.Done()
+ for {
+ select {
+ case <-stopCh:
+ return
+ default:
+ // Generate events for all accounts cyclically
+ accID := rand.Intn(numProfiles) + 1 // Random account matching a profile
+ args := &engine.CGREventWithEeIDs{
+ CGREvent: &utils.CGREvent{
+ Tenant: "cgrates.org",
+ ID: utils.GenUUID(),
+ Event: map[string]any{
+ utils.AccountField: fmt.Sprintf("100%d", accID),
+ utils.Usage: rand.Intn(60) + 1,
+ utils.Cost: rand.Float64() * 10,
+ },
+ },
+ }
+ var reply []string
+ if err := client.Call(context.Background(), utils.StatSv1ProcessEvent, args, &reply); err != nil {
+ errCh <- fmt.Errorf("worker %d: %v", workerID, err)
+ }
+ evcount.Add(1)
+ }
+ }
+ }(i)
+ }
+
+ // Wait for all workers to finish
+ go func() {
+ wg.Wait()
+ close(errCh)
+ }()
+
+ // Collect errors
+ for err := range errCh {
+ t.Error(err)
+ }
+ t.Logf("Generated %v events for %v", evcount.Load(), time.Since(start))
+ })
+
+ // t.Run("VerifyTrendExecution", func(t *testing.T) {
+ // time.Sleep(5 * time.Second) // Allow final trend executions
+
+ // // Verify each trend profile executed ~300 times (30s test * 10 executions/s)
+ // // This requires trend executions to be logged in a identifiable way
+ // // Example: Count occurrences of "processed trend Trend123"
+ // expectedExecutions := 30 * 10 // 30 seconds * 10 executions/second
+ // for i := 1; i <= numProfiles; i++ {
+ // trendID := fmt.Sprintf("Trend%d", i)
+ // if count < expectedExecutions*0.8 { // Allow 20% scheduling variance
+ // t.Errorf("Trend %s executed %d times, expected ~%d", trendID, count, expectedExecutions)
+ // }
+ // }
+ // })
+
+}