Skip to content

Commit

Permalink
Implement reconnect logic for AMQP EventReader
Browse files Browse the repository at this point in the history
Implemented functionality that handles reconnecting to the amqp server
and reinitializing the amqp channel in case of errors and timeouts. This
is handled by a goroutine created in the client constructor (it also
handles the initial connect/init).

Reconnects and reinits will use a fibonacci backoff strategy, and the
attempt amount and max waiting interval can be adjusted by the
'reconnects' and 'max_reconnect_interval' config options.

Messages that fail processing are now dropped instead of being requeued,
preventing infinite processing loops. However, this means that the
messages are lost. Handling failed messages will need to be addressed
separately.

'concurrent_requests' will now set the prefetch count. Setting the
prefetch count using the Qos function was able to replace our old
approach that was using channels. Default value is 1024 which,
according to the rabbitmq docs, 'runs into the law of diminishing
returns'. The recommended value is between 100-300. Source:
https://www.rabbitmq.com/confirms.html#channel-qos-prefetch-throughput

Fix test compilation errors and failing tests caused by these changes.

References #4160
  • Loading branch information
ionutboangiu authored and danbogos committed Nov 8, 2023
1 parent f696164 commit c282d40
Show file tree
Hide file tree
Showing 9 changed files with 803 additions and 594 deletions.
100 changes: 54 additions & 46 deletions config/config_it_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -596,17 +596,19 @@ func testCGRConfigReloadERs(t *testing.T) {
SessionSConns: []string{utils.MetaLocalHost},
Readers: []*EventReaderCfg{
{
ID: utils.MetaDefault,
Type: utils.MetaNone,
RunDelay: 0,
ConcurrentReqs: 1024,
SourcePath: "/var/spool/cgrates/ers/in",
ProcessedPath: "/var/spool/cgrates/ers/out",
Filters: []string{},
Flags: flagsDefault,
Fields: content,
CacheDumpFields: []*FCTemplate{},
PartialCommitFields: []*FCTemplate{},
ID: utils.MetaDefault,
Type: utils.MetaNone,
RunDelay: 0,
ConcurrentReqs: 1024,
SourcePath: "/var/spool/cgrates/ers/in",
ProcessedPath: "/var/spool/cgrates/ers/out",
Filters: []string{},
Flags: flagsDefault,
Fields: content,
CacheDumpFields: []*FCTemplate{},
PartialCommitFields: []*FCTemplate{},
Reconnects: -1,
MaxReconnectInterval: 5 * time.Minute,
Opts: &EventReaderOpts{
CSV: &CSVROpts{
FieldSeparator: utils.StringPointer(utils.FieldsSep),
Expand All @@ -625,17 +627,19 @@ func testCGRConfigReloadERs(t *testing.T) {
},
},
{
ID: "file_reader1",
Type: utils.MetaFileCSV,
RunDelay: -1,
ConcurrentReqs: 1024,
SourcePath: "/tmp/ers/in",
ProcessedPath: "/tmp/ers/out",
Filters: []string{},
Flags: flags,
Fields: content,
CacheDumpFields: []*FCTemplate{},
PartialCommitFields: []*FCTemplate{},
ID: "file_reader1",
Type: utils.MetaFileCSV,
RunDelay: -1,
ConcurrentReqs: 1024,
SourcePath: "/tmp/ers/in",
ProcessedPath: "/tmp/ers/out",
Filters: []string{},
Flags: flags,
Fields: content,
CacheDumpFields: []*FCTemplate{},
PartialCommitFields: []*FCTemplate{},
Reconnects: -1,
MaxReconnectInterval: 5 * time.Minute,
Opts: &EventReaderOpts{
CSV: &CSVROpts{
FieldSeparator: utils.StringPointer(utils.FieldsSep),
Expand Down Expand Up @@ -827,18 +831,20 @@ func testCgrCfgV1ReloadConfigSection(t *testing.T) {
"partial_cache_ttl": "1s",
"readers": []any{
map[string]any{
"id": utils.MetaDefault,
"cache_dump_fields": []any{},
"concurrent_requests": 1024,
"fields": content,
"filters": []string{},
"flags": []string{},
"run_delay": "0",
"source_path": "/var/spool/cgrates/ers/in",
"processed_path": "/var/spool/cgrates/ers/out",
"tenant": "",
"timezone": "",
"type": utils.MetaNone,
"id": utils.MetaDefault,
"cache_dump_fields": []any{},
"concurrent_requests": 1024,
"fields": content,
"filters": []string{},
"flags": []string{},
"run_delay": "0",
"source_path": "/var/spool/cgrates/ers/in",
"processed_path": "/var/spool/cgrates/ers/out",
"tenant": "",
"timezone": "",
"type": utils.MetaNone,
utils.ReconnectsCfg: -1,
utils.MaxReconnectIntervalCfg: "5m0s",
"opts": map[string]any{
"csvFieldSeparator": ",",
"csvHeaderDefineChar": ":",
Expand All @@ -850,18 +856,20 @@ func testCgrCfgV1ReloadConfigSection(t *testing.T) {
"partial_commit_fields": []any{},
},
map[string]any{
"cache_dump_fields": []any{},
"concurrent_requests": 1024,
"filters": []string{},
"flags": []string{"*dryrun"},
"id": "file_reader1",
"processed_path": "/tmp/ers/out",
"run_delay": "-1",
"source_path": "/tmp/ers/in",
"tenant": "",
"timezone": "",
"type": "*file_csv",
"fields": content,
"cache_dump_fields": []any{},
"concurrent_requests": 1024,
"filters": []string{},
"flags": []string{"*dryrun"},
"id": "file_reader1",
"processed_path": "/tmp/ers/out",
"run_delay": "-1",
"source_path": "/tmp/ers/in",
"tenant": "",
"timezone": "",
"type": "*file_csv",
"fields": content,
utils.ReconnectsCfg: -1,
utils.MaxReconnectIntervalCfg: "5m0s",
"opts": map[string]any{
"csvFieldSeparator": ",",
"csvHeaderDefineChar": ":",
Expand Down
28 changes: 15 additions & 13 deletions config/config_json_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2003,19 +2003,21 @@ func TestDfEventReaderCfg(t *testing.T) {
Sessions_conns: &[]string{utils.MetaInternal},
Readers: &[]*EventReaderJsonCfg{
{
Id: utils.StringPointer(utils.MetaDefault),
Type: utils.StringPointer(utils.MetaNone),
Run_delay: utils.StringPointer("0"),
Concurrent_requests: utils.IntPointer(1024),
Source_path: utils.StringPointer("/var/spool/cgrates/ers/in"),
Processed_path: utils.StringPointer("/var/spool/cgrates/ers/out"),
Tenant: utils.StringPointer(utils.EmptyString),
Timezone: utils.StringPointer(utils.EmptyString),
Filters: &[]string{},
Flags: &[]string{},
Fields: &cdrFields,
Cache_dump_fields: &[]*FcTemplateJsonCfg{},
Partial_commit_fields: &[]*FcTemplateJsonCfg{},
Id: utils.StringPointer(utils.MetaDefault),
Type: utils.StringPointer(utils.MetaNone),
Run_delay: utils.StringPointer("0"),
Concurrent_requests: utils.IntPointer(1024),
Source_path: utils.StringPointer("/var/spool/cgrates/ers/in"),
Processed_path: utils.StringPointer("/var/spool/cgrates/ers/out"),
Tenant: utils.StringPointer(utils.EmptyString),
Timezone: utils.StringPointer(utils.EmptyString),
Filters: &[]string{},
Flags: &[]string{},
Fields: &cdrFields,
Cache_dump_fields: &[]*FcTemplateJsonCfg{},
Partial_commit_fields: &[]*FcTemplateJsonCfg{},
Reconnects: utils.IntPointer(-1),
Max_reconnect_interval: utils.StringPointer("5m"),
Opts: &EventReaderOptsJson{
CSVFieldSeparator: utils.StringPointer(utils.FieldsSep),
CSVHeaderDefineChar: utils.StringPointer(utils.InInFieldSep),
Expand Down
104 changes: 56 additions & 48 deletions config/config_test.go

Large diffs are not rendered by default.

0 comments on commit c282d40

Please sign in to comment.