From 37787469c1d1465188501bede9e00d53375d95e1 Mon Sep 17 00:00:00 2001 From: Rene Krenn Date: Sun, 2 Apr 2017 15:28:59 +0200 Subject: [PATCH] TT#13821 process acc trash records fix tool to update acc_trash and re-enqueue records accumulated during 3.x->4.x updates due to introduction of cdr lnp prefix fields. Change-Id: Iea6140edccc1aeb456f6584b7d8335e18ba1b69d --- .../BulkProcessor/Dao/Trunk/accounting/cdr.pm | 203 ++++++++++ .../BulkProcessor/Dao/Trunk/kamailio/acc.pm | 152 ++++++++ .../Dao/Trunk/kamailio/acc_trash.pm | 228 ++++++++++++ .../Projects/Disaster/Acc/AccTrash.pm | 352 ++++++++++++++++++ .../Projects/Disaster/Acc/Check.pm | 118 ++++++ .../Projects/Disaster/Acc/Settings.pm | 134 +++++++ .../Projects/Disaster/Acc/config.cfg | 62 +++ .../Projects/Disaster/Acc/process.pl | 250 +++++++++++++ .../Projects/Disaster/Acc/settings.cfg | 12 + lib/NGCP/BulkProcessor/SqlProcessor.pm | 16 +- 10 files changed, 1520 insertions(+), 7 deletions(-) create mode 100644 lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc.pm create mode 100644 lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc_trash.pm create mode 100644 lib/NGCP/BulkProcessor/Projects/Disaster/Acc/AccTrash.pm create mode 100644 lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Check.pm create mode 100644 lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Settings.pm create mode 100644 lib/NGCP/BulkProcessor/Projects/Disaster/Acc/config.cfg create mode 100644 lib/NGCP/BulkProcessor/Projects/Disaster/Acc/process.pl create mode 100644 lib/NGCP/BulkProcessor/Projects/Disaster/Acc/settings.cfg diff --git a/lib/NGCP/BulkProcessor/Dao/Trunk/accounting/cdr.pm b/lib/NGCP/BulkProcessor/Dao/Trunk/accounting/cdr.pm index e69de29b..1655bb92 100644 --- a/lib/NGCP/BulkProcessor/Dao/Trunk/accounting/cdr.pm +++ b/lib/NGCP/BulkProcessor/Dao/Trunk/accounting/cdr.pm @@ -0,0 +1,203 @@ +package NGCP::BulkProcessor::Dao::Trunk::accounting::cdr; +use strict; + +## no critic + +use NGCP::BulkProcessor::Logging qw( + getlogger + rowsdeleted +); + +use NGCP::BulkProcessor::ConnectorPool qw( + get_accounting_db +); + +use NGCP::BulkProcessor::SqlProcessor qw( + checktableinfo + + copy_row +); +use NGCP::BulkProcessor::SqlRecord qw(); + +require Exporter; +our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord); +our @EXPORT_OK = qw( + gettablename + check_table + + delete_callids + +); +#process_records +#delete_ids + +my $tablename = 'cdr'; +my $get_db = \&get_accounting_db; + +my $expected_fieldnames = [ +"id", +"update_time", +"source_user_id", +"source_provider_id", +"source_external_subscriber_id", +"source_external_contract_id", +"source_account_id", +"source_user", +"source_domain", +"source_cli", +"source_clir", +"source_ip", +"source_gpp0", +"source_gpp1", +"source_gpp2", +"source_gpp3", +"source_gpp4", +"source_gpp5", +"source_gpp6", +"source_gpp7", +"source_gpp8", +"source_gpp9", +"source_lnp_prefix", +"destination_user_id", +"destination_provider_id", +"destination_external_subscriber_id", +"destination_external_contract_id", +"destination_account_id", +"destination_user", +"destination_domain", +"destination_user_dialed", +"destination_user_in", +"destination_domain_in", +"destination_gpp0", +"destination_gpp1", +"destination_gpp2", +"destination_gpp3", +"destination_gpp4", +"destination_gpp5", +"destination_gpp6", +"destination_gpp7", +"destination_gpp8", +"destination_gpp9", +"destination_lnp_prefix", +"peer_auth_user", +"peer_auth_realm", +"call_type", +"call_status", +"call_code", +"init_time", +"start_time", +"duration", +"call_id", +"source_carrier_cost", +"source_reseller_cost", +"source_customer_cost", +"source_carrier_free_time", +"source_reseller_free_time", +"source_customer_free_time", +"source_carrier_billing_fee_id", +"source_reseller_billing_fee_id", +"source_customer_billing_fee_id", +"source_carrier_billing_zone_id", +"source_reseller_billing_zone_id", +"source_customer_billing_zone_id", +"destination_carrier_cost", +"destination_reseller_cost", +"destination_customer_cost", +"destination_carrier_free_time", +"destination_reseller_free_time", +"destination_customer_free_time", +"destination_carrier_billing_fee_id", +"destination_reseller_billing_fee_id", +"destination_customer_billing_fee_id", +"destination_carrier_billing_zone_id", +"destination_reseller_billing_zone_id", +"destination_customer_billing_zone_id", +"frag_carrier_onpeak", +"frag_reseller_onpeak", +"frag_customer_onpeak", +"is_fragmented", +"split", +"rated_at", +"rating_status", +"exported_at", +"export_status", +]; + +my $indexes = {}; + +my $insert_unique_fields = []; + +sub new { + + my $class = shift; + my $self = NGCP::BulkProcessor::SqlRecord->new($class,$get_db, + $tablename,$expected_fieldnames,$indexes); + + copy_row($self,shift,$expected_fieldnames); + + return $self; + +} + + +sub delete_callids { + + my ($xa_db,$callids) = @_; + + check_table(); + my $db = &$get_db(); + $xa_db //= $db; + my $table = $db->tableidentifier($tablename); + + my $stmt = 'DELETE FROM ' . $table . ' WHERE ' . + $db->columnidentifier('call_id') . ' IN (' . substr(',?' x scalar @$callids,1) . ')'; + my @params = @$callids; + + my $count; + if ($count = $xa_db->db_do($stmt,@params)) { + rowsdeleted($db,$tablename,$count,$count,getlogger(__PACKAGE__)); + return 1; + } else { + rowsdeleted($db,$tablename,0,0,getlogger(__PACKAGE__)); + return 0; + } + +} + +sub buildrecords_fromrows { + + my ($rows,$load_recursive) = @_; + + my @records = (); + my $record; + + if (defined $rows and ref $rows eq 'ARRAY') { + foreach my $row (@$rows) { + $record = __PACKAGE__->new($row); + + # transformations go here ... + + push @records,$record; + } + } + + return \@records; + +} + +sub gettablename { + + return $tablename; + +} + +sub check_table { + + return checktableinfo($get_db, + __PACKAGE__,$tablename, + $expected_fieldnames, + $indexes); + +} + +1; diff --git a/lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc.pm b/lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc.pm new file mode 100644 index 00000000..1856616f --- /dev/null +++ b/lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc.pm @@ -0,0 +1,152 @@ +package NGCP::BulkProcessor::Dao::Trunk::kamailio::acc; +use strict; + +## no critic + +use NGCP::BulkProcessor::Logging qw( + getlogger + rowinserted +); + +use NGCP::BulkProcessor::ConnectorPool qw( + get_kamailio_db +); + +use NGCP::BulkProcessor::SqlProcessor qw( + checktableinfo + insert_record + copy_row +); +use NGCP::BulkProcessor::SqlRecord qw(); + +require Exporter; +our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord); +our @EXPORT_OK = qw( + gettablename + check_table + + insert_row + count + count_callids +); + +my $tablename = 'acc'; +my $get_db = \&get_kamailio_db; + +my $expected_fieldnames = [ +"id", +"method", +"from_tag", +"to_tag", +"callid", +"sip_code", +"sip_reason", +"time", +"time_hires", +"src_leg", +"dst_leg", +"dst_user", +"dst_ouser", +"dst_domain", +"src_user", +"src_domain", +]; + +my $indexes = {}; + +my $insert_unique_fields = []; + +sub new { + + my $class = shift; + my $self = NGCP::BulkProcessor::SqlRecord->new($class,$get_db, + $tablename,$expected_fieldnames,$indexes); + + copy_row($self,shift,$expected_fieldnames); + + return $self; + +} + +sub insert_row { + + my $db = &$get_db(); + my $xa_db = shift // $db; + #if ('HASH' eq ref $_[0]) { + my ($data,$insert_ignore) = @_; + check_table(); + if (insert_record($db,$xa_db,__PACKAGE__,$data,$insert_ignore,$insert_unique_fields)) { + return $xa_db->db_last_insert_id(); + } + #} + return undef; + +} + +sub count { + + my ($xa_db) = @_; + + check_table(); + my $db = &$get_db(); + $xa_db //= $db; + my $table = $db->tableidentifier($tablename); + + my $stmt = 'SELECT COUNT(*) FROM ' . $table; + + return $db->db_get_value($stmt); + +} + +sub count_callids { + + my ($xa_db) = @_; + + check_table(); + my $db = &$get_db(); + $xa_db //= $db; + my $table = $db->tableidentifier($tablename); + + my $stmt = 'SELECT COUNT(DISTINCT ' . $db->columnidentifier('callid') . ') FROM ' . $table; + + return $db->db_get_value($stmt); + +} + +sub buildrecords_fromrows { + + my ($rows,$load_recursive) = @_; + + my @records = (); + my $record; + + if (defined $rows and ref $rows eq 'ARRAY') { + foreach my $row (@$rows) { + $record = __PACKAGE__->new($row); + + # transformations go here ... + + push @records,$record; + } + } + + return \@records; + +} + +sub gettablename { + + return $tablename; + +} + +sub check_table { + + return checktableinfo($get_db, + __PACKAGE__,$tablename, + $expected_fieldnames, + $indexes); + +} + +1; diff --git a/lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc_trash.pm b/lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc_trash.pm new file mode 100644 index 00000000..84e5dc81 --- /dev/null +++ b/lib/NGCP/BulkProcessor/Dao/Trunk/kamailio/acc_trash.pm @@ -0,0 +1,228 @@ +package NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash; +use strict; + +## no critic + +use NGCP::BulkProcessor::Logging qw( + getlogger + rowsdeleted +); + +use NGCP::BulkProcessor::ConnectorPool qw( + get_kamailio_db + destroy_dbs +); + +use NGCP::BulkProcessor::SqlProcessor qw( + checktableinfo + process_table + copy_row +); +use NGCP::BulkProcessor::SqlRecord qw(); + +require Exporter; +our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord); +our @EXPORT_OK = qw( + gettablename + check_table + + process_records + delete_callids + findby_callids + count + count_callids +); + +my $tablename = 'acc_trash'; +my $get_db = \&get_kamailio_db; + +my $expected_fieldnames = [ +"id", +"method", +"from_tag", +"to_tag", +"callid", +"sip_code", +"sip_reason", +"time", +"time_hires", +"src_leg", +"dst_leg", +"dst_user", +"dst_ouser", +"dst_domain", +"src_user", +"src_domain", +]; + +my $indexes = {}; + +my $insert_unique_fields = []; + +sub new { + + my $class = shift; + my $self = NGCP::BulkProcessor::SqlRecord->new($class,$get_db, + $tablename,$expected_fieldnames,$indexes); + + copy_row($self,shift,$expected_fieldnames); + + return $self; + +} + + +sub delete_callids { + + my ($xa_db,$callids) = @_; + + check_table(); + my $db = &$get_db(); + $xa_db //= $db; + my $table = $db->tableidentifier($tablename); + + my $stmt = 'DELETE FROM ' . $table . ' WHERE ' . + $db->columnidentifier('callid') . ' IN (' . substr(',?' x scalar @$callids,1) . ')'; + my @params = @$callids; + + my $count; + if ($count = $xa_db->db_do($stmt,@params)) { + rowsdeleted($db,$tablename,$count,$count,getlogger(__PACKAGE__)); + return 1; + } else { + rowsdeleted($db,$tablename,0,0,getlogger(__PACKAGE__)); + return 0; + } + +} + +sub count { + + my ($xa_db) = @_; + + check_table(); + my $db = &$get_db(); + $xa_db //= $db; + my $table = $db->tableidentifier($tablename); + + my $stmt = 'SELECT COUNT(*) FROM ' . $table; + + return $db->db_get_value($stmt); + +} + +sub count_callids { + + my ($xa_db) = @_; + + check_table(); + my $db = &$get_db(); + $xa_db //= $db; + my $table = $db->tableidentifier($tablename); + + my $stmt = 'SELECT COUNT(DISTINCT ' . $db->columnidentifier('callid') . ') FROM ' . $table; + + return $db->db_get_value($stmt); + +} + +sub findby_callids { + + my ($xa_db,$callids,$load_recursive) = @_; + + check_table(); + my $db = &$get_db(); + $xa_db //= $db; + my $table = $db->tableidentifier($tablename); + + my $stmt = 'SELECT * FROM ' . $table . ' WHERE ' . + $db->columnidentifier('callid') . ' IN (' . substr(',?' x scalar @$callids,1) . ')'; + my @params = @$callids; + my $rows = $xa_db->db_get_all_arrayref($stmt,@params); + + return buildrecords_fromrows($rows,$load_recursive); + +} + + +sub process_records { + + my %params = @_; + my ($process_code, + $static_context, + $init_process_context_code, + $uninit_process_context_code, + $multithreading, + $blocksize, + $numofthreads, + $load_recursive) = @params{qw/ + process_code + static_context + init_process_context_code + uninit_process_context_code + multithreading + blocksize + numofthreads + load_recursive + /}; + + check_table(); + my $db = &$get_db(); + my $table = $db->tableidentifier($tablename); + + return process_table( + get_db => $get_db, + class => __PACKAGE__, + process_code => sub { + my ($context,$rowblock,$row_offset) = @_; + return &$process_code($context,buildrecords_fromrows($rowblock,$load_recursive),$row_offset); + }, + static_context => $static_context, + init_process_context_code => $init_process_context_code, + uninit_process_context_code => $uninit_process_context_code, + destroy_reader_dbs_code => \&destroy_dbs, + multithreading => $multithreading, + blocksize => $blocksize, + tableprocessing_threads => $numofthreads, + 'select' => 'SELECT * FROM ' . $table . ' GROUP BY ' . $db->columnidentifier('callid'), + 'selectcount' => 'SELECT COUNT(DISTINCT ' . $db->columnidentifier('callid') . ') FROM ' . $table, + ); +} + +sub buildrecords_fromrows { + + my ($rows,$load_recursive) = @_; + + my @records = (); + my $record; + + if (defined $rows and ref $rows eq 'ARRAY') { + foreach my $row (@$rows) { + $record = __PACKAGE__->new($row); + + # transformations go here ... + + push @records,$record; + } + } + + return \@records; + +} + +sub gettablename { + + return $tablename; + +} + +sub check_table { + + return checktableinfo($get_db, + __PACKAGE__,$tablename, + $expected_fieldnames, + $indexes); + +} + +1; diff --git a/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/AccTrash.pm b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/AccTrash.pm new file mode 100644 index 00000000..81577bf2 --- /dev/null +++ b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/AccTrash.pm @@ -0,0 +1,352 @@ +package NGCP::BulkProcessor::Projects::Disaster::Acc::AccTrash; +use strict; + +## no critic + +use threads::shared qw(); +#use List::Util qw(); +use DateTime qw(); +use Time::HiRes qw(sleep); + +use NGCP::BulkProcessor::Projects::Disaster::Acc::Settings qw( + $dry + $skip_errors + + $process_acc_trash_multithreading + $process_acc_trash_numofthreads + $process_acc_trash_blocksize + + $delete_cdr + + $sleep_secs + $acc_record_limit +); +#$set_preference_bulk_numofthreads + +use NGCP::BulkProcessor::Logging qw ( + getlogger + processing_info + processing_debug +); +use NGCP::BulkProcessor::LogError qw( + rowprocessingerror + rowprocessingwarn +); + +use NGCP::BulkProcessor::Dao::Trunk::accounting::cdr qw(); +use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc qw(); +use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash qw(); + + + +use NGCP::BulkProcessor::ConnectorPool qw( + + get_kamailio_db + get_accounting_db +); +#get_xa_db + +use NGCP::BulkProcessor::ConnectorPool qw( + destroy_dbs +); + +use NGCP::BulkProcessor::Utils qw(threadid); + +require Exporter; +our @ISA = qw(Exporter); +our @EXPORT_OK = qw( + process_acc_trash +); + +sub process_acc_trash { + + my ($process_acc_trash,$fix_lnp_prefix_tokens) = @_; + my $static_context = { + delete_cdr => $delete_cdr, + sleep_secs => $sleep_secs, + acc_record_limit => $acc_record_limit, + process_acc_trash => $process_acc_trash, + fix_lnp_prefix_tokens => $fix_lnp_prefix_tokens, + }; + my $result = _process_acc_trash_checks($static_context); + + destroy_dbs(); + my $warning_count :shared = 0; + return ($result && NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::process_records( + static_context => $static_context, + process_code => sub { + my ($context,$records,$row_offset) = @_; + my $rownum = $row_offset; + if ($context->{sleep_secs} > 0.0 and $context->{acc_record_limit} > 0) { + my $sleep = $context->{sleep_secs}; + while (((my $acc_count = NGCP::BulkProcessor::Dao::Trunk::kamailio::acc::count($context->{acc_db})) + scalar @$records) > $context->{acc_record_limit}) { + _info($context,"$acc_count acc records (limit $context->{acc_record_limit}), sleep for $sleep secs ..."); + sleep($sleep); + $sleep *= 2.0 if $sleep < 30.0; #manchester + } + } + foreach my $acc_trash (@$records) { + $rownum++; + next unless _reset_process_acc_trash_context($context,$acc_trash,$rownum); + #if (_delete_cdr($context)) { + # _process_acc_trash($context); + #} + _delete_cdr($context) if $context->{delete_cdr}; + _process_acc_trash($context) if $context->{process_acc_trash}; + } + + #return 0; + return 1; + }, + init_process_context_code => sub { + my ($context)= @_; + $context->{acc_db} = &get_kamailio_db(); + $context->{cdr_db} = &get_accounting_db(); + $context->{error_count} = 0; + $context->{warning_count} = 0; + # below is not mandatory.. + _check_insert_tables(); + }, + uninit_process_context_code => sub { + my ($context)= @_; + undef $context->{acc_db}; + undef $context->{cdr_db}; + destroy_dbs(); + { + lock $warning_count; + $warning_count += $context->{warning_count}; + } + }, + load_recursive => 0, + multithreading => $process_acc_trash_multithreading, + numofthreads => $process_acc_trash_numofthreads, + blocksize => $process_acc_trash_blocksize, + ),$warning_count); +} + + +sub _check_insert_tables { + + #NGCP::BulkProcessor::Dao::mr38::provisioning::voip_usr_preferences::check_table(); + +} + +sub _process_acc_trash { + my ($context) = @_; + + eval { + $context->{acc_db}->db_begin(); + + my $incomplete = 0; + foreach my $acc_trash (@{$context->{acc_trash}}) { + my $src_leg = $acc_trash->{src_leg}; + #my $move = 1; + if ($context->{fix_lnp_prefix_tokens}) { + if (defined $src_leg and length($src_leg) > 0) { + my @tokens = split(/\|/,$src_leg,-1); + my $tokencount = scalar @tokens; + if ($tokencount == 24) { + $acc_trash->{src_leg} .= '|'; + } elsif ($tokencount == 25) { + _info($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " seems to be correct (lnp_prefix), src_leg token count = $tokencount"); + } elsif ($tokencount > 0) { + if ($skip_errors) { + _warn($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), src_leg token count = $tokencount"); + } else { + _error($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), src_leg token count = $tokencount"); + } + $incomplete = 1; + #$move = 0; + #last; + } + } + my $dst_leg = $acc_trash->{dst_leg}; + if (defined $dst_leg and length($dst_leg) > 0) { + my @tokens = split(/\|/,$dst_leg,-1); + my $tokencount = scalar @tokens; + if ($tokencount == 22) { + $acc_trash->{dst_leg} .= '|'; + } elsif ($tokencount == 23) { + _info($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " seems to be correct (lnp_prefix), dst_leg token count = $tokencount"); + } elsif ($tokencount > 0) { + if ($skip_errors) { + _warn($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), dst_leg token count = $tokencount"); + } else { + _error($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " cannot be fixed (lnp_prefix), dst_leg token count = $tokencount"); + } + $incomplete = 1; + #$move = 0; + #last; + } + } + } + unless ($incomplete) { + NGCP::BulkProcessor::Dao::Trunk::kamailio::acc::insert_row($context->{acc_db},$acc_trash); + _info($context,"($context->{rownum}) " . 'acc trash record id ' . $acc_trash->{id} . " copied",1); + } + } + unless ($incomplete) { + NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::delete_callids($context->{acc_db},[ $context->{call_id} ]); + _info($context,"($context->{rownum}) " . 'acc trash records for callid ' . $context->{call_id} . " deleted",1); + } + + if ($dry or $incomplete) { + $context->{acc_db}->db_rollback(0); + } else { + $context->{acc_db}->db_commit(); + } + + }; + my $err = $@; + if ($err) { + eval { + $context->{acc_db}->db_rollback(1); + }; + if ($skip_errors) { + _warn($context,"($context->{rownum}) " . 'kamailio database error with acc trash callid ' . $context->{call_id} . ': ' . $err); + } else { + _error($context,"($context->{rownum}) " . 'kamailio database error with acc trash callid ' . $context->{call_id} . ': ' . $err); + } + } +} + +sub _delete_cdr { + my ($context) = @_; + + eval { + $context->{cdr_db}->db_begin(); + + if (NGCP::BulkProcessor::Dao::Trunk::accounting::cdr::delete_callids($context->{cdr_db},[ $context->{call_id} ])) { + _info($context,'cdr id ' . $context->{call_id} . " deleted"); + } else { + _info($context,'no cdr id ' . $context->{call_id} . " to delete",1); + } + + if ($dry) { + $context->{cdr_db}->db_rollback(0); + } else { + $context->{cdr_db}->db_commit(); + } + + }; + my $err = $@; + if ($err) { + eval { + $context->{cdr_db}->db_rollback(1); + }; + if ($skip_errors) { + _warn($context,"($context->{rownum}) " . 'accounting database error with acc trash callid ' . $context->{call_id} . ': ' . $err); + } else { + _error($context,"($context->{rownum}) " . 'accounting database error with acc trash callid ' . $context->{call_id} . ': ' . $err); + } + #return 0; + } + #return 1; + +} + +sub _process_acc_trash_checks { + my ($context) = @_; + + my $result = _checks($context); + + return $result; +} + +sub _reset_process_acc_trash_context { + + my ($context,$acc_trash,$rownum) = @_; + + my $result = _reset_context($context,$acc_trash,$rownum); + + $context->{call_id} = $acc_trash->{callid}; + $context->{acc_trash} = NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::findby_callids($context->{acc_db},[ $context->{call_id} ]); + + ##$context->{barring_profile} = $imported_subscriber->{barring_profile}; + ##$context->{ncos_level} = $context->{ncos_level_map}->{$context->{barring_profile}}; + + ##delete $context->{adm_ncos_id_preference_id}; + + return $result; + +} + + +sub _checks { + + my ($context) = @_; + + my $result = 1; + my $acctrashcount = 0; + eval { + $acctrashcount = NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::count(); + }; + if ($@ or $acctrashcount == 0) { + rowprocessingerror(threadid(),'no acc trash records',getlogger(__PACKAGE__)); + $result = 0; #even in skip-error mode.. + } + #my $userpasswordcount = 0; + #eval { + # $userpasswordcount = NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::UsernamePassword::countby_fqdn(); + #}; + #if ($@ or $userpasswordcount == 0) { + # rowprocessingerror(threadid(),'please import user passwords first',getlogger(__PACKAGE__)); + # $result = 0; #even in skip-error mode.. + #} + #my $subscribercount = 0; + #my $subscriber_barring_profiles = []; + #eval { + # $subscribercount = NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::Subscriber::countby_subscribernumber(); + # $subscriber_barring_profiles = NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::import::Subscriber::list_barringprofiles(); + #}; + #if ($@ or $subscribercount == 0) { + # rowprocessingerror(threadid(),'please import subscribers first',getlogger(__PACKAGE__)); + # $result = 0; #even in skip-error mode.. + #} + + return $result; + +} + +sub _reset_context { + + my ($context,$acc_trash,$rownum) = @_; + + my $result = 1; + + $context->{rownum} = $rownum; + + undef $context->{call_id}; + undef $context->{acc_trash}; # = $acc_trash; + + return $result; + +} + +sub _error { + + my ($context,$message) = @_; + $context->{error_count} = $context->{error_count} + 1; + rowprocessingerror($context->{tid},$message,getlogger(__PACKAGE__)); + +} + +sub _warn { + + my ($context,$message) = @_; + $context->{warning_count} = $context->{warning_count} + 1; + rowprocessingwarn($context->{tid},$message,getlogger(__PACKAGE__)); + +} + +sub _info { + + my ($context,$message,$debug) = @_; + if ($debug) { + processing_debug($context->{tid},$message,getlogger(__PACKAGE__)); + } else { + processing_info($context->{tid},$message,getlogger(__PACKAGE__)); + } +} + +1; diff --git a/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Check.pm b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Check.pm new file mode 100644 index 00000000..10eeefea --- /dev/null +++ b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Check.pm @@ -0,0 +1,118 @@ +package NGCP::BulkProcessor::Projects::Disaster::Acc::Check; +use strict; + +## no critic + +no strict 'refs'; + +use NGCP::BulkProcessor::Dao::Trunk::accounting::cdr qw(); +use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc qw(); +use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash qw(); + +require Exporter; +our @ISA = qw(Exporter); +our @EXPORT_OK = qw( + check_accounting_db_tables + check_kamailio_db_tables +); +#check_rest_get_items + +my $NOK = 'NOK'; +my $OK = 'ok'; + +sub check_kamailio_db_tables { + + my ($messages) = @_; + + my $result = 1; + my $check_result; + my $message; + + my $message_prefix = 'NGCP kamailio db tables - '; + + ($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Dao::Trunk::kamailio::acc'); + $result &= $check_result; push(@$messages,$message); + + ($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash'); + $result &= $check_result; push(@$messages,$message); + + return $result; + +} + +sub check_accounting_db_tables { + + my ($messages) = @_; + + my $result = 1; + my $check_result; + my $message; + + my $message_prefix = 'NGCP accounting db tables - '; + + ($check_result,$message) = _check_table($message_prefix,'NGCP::BulkProcessor::Dao::Trunk::accounting::cdr'); + $result &= $check_result; push(@$messages,$message); + + return $result; + +} + +sub _check_table { + + my ($message_prefix,$module) = @_; + my $result = 0; + my $message = ($message_prefix // '') . &{$module . '::gettablename'}() . ': '; + eval { + $result = &{$module . '::check_table'}(); + }; + if (@$ or not $result) { + return (0,$message . $NOK); + } else { + return (1,$message . $OK); + } + +} + +#sub check_rest_get_items { +# +# my ($messages) = @_; +# +# my $result = 1; +# my $check_result; +# my $message; +# +# my $message_prefix = 'NGCP id\'s/constants - '; +# +# ($check_result,$message, my $reseller) = _check_rest_get_item($message_prefix, +# 'NGCP::BulkProcessor::RestRequests::mr38::Resellers', +# $reseller_id, +# 'name'); +# $result &= $check_result; push(@$messages,$message); +# +# +# return $result; +# +#} + +sub _check_rest_get_item { + + my ($message_prefix,$module,$id,$item_name_field,$get_method,$item_path_method) = @_; + my $item = undef; + $get_method //= 'get_item'; + $item_path_method //= 'get_item_path'; + my $message = ($message_prefix // '') . &{$module . '::' . $item_path_method}($id) . ': '; + return (0,$message . $NOK,$item) unless $id; + eval { + $item = &{$module . '::' . $get_method}($id); + }; + + if (@$ or not defined $item or ('ARRAY' eq ref $item and (scalar @$item) != 1)) { + return (0,$message . $NOK,$item); + } else { + $item = $item->[0] if ('ARRAY' eq ref $item and (scalar @$item) == 1); + return (1,$message . "'" . $item->{$item_name_field} . "' " . $OK,$item); + } + +} + +1; diff --git a/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Settings.pm b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Settings.pm new file mode 100644 index 00000000..b7374a35 --- /dev/null +++ b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/Settings.pm @@ -0,0 +1,134 @@ +package NGCP::BulkProcessor::Projects::Disaster::Acc::Settings; +use strict; + +## no critic + +use NGCP::BulkProcessor::Globals qw( + $enablemultithreading + $cpucount +); +#$working_path +#create_path + +use NGCP::BulkProcessor::Logging qw( + getlogger + scriptinfo + configurationinfo +); + +use NGCP::BulkProcessor::LogError qw( + fileerror + configurationwarn + configurationerror +); + +use NGCP::BulkProcessor::LoadConfig qw( + split_tuple + parse_regexp +); +use NGCP::BulkProcessor::Utils qw(prompt); +#format_number check_ipnet + +require Exporter; +our @ISA = qw(Exporter); +our @EXPORT_OK = qw( + update_settings + check_dry + + $defaultsettings + $defaultconfig + + $dry + $skip_errors + $force + + $process_acc_trash_multithreading + $process_acc_trash_numofthreads + $process_acc_trash_blocksize + + $delete_cdr + + $sleep_secs + $acc_record_limit +); + +our $defaultconfig = 'config.cfg'; +our $defaultsettings = 'settings.cfg'; + +our $force = 0; +our $dry = 0; +our $skip_errors = 0; + +our $process_acc_trash_multithreading = $enablemultithreading; +our $process_acc_trash_numofthreads = $cpucount; +our $process_acc_trash_blocksize = 100; +our $delete_cdr = 1; +our $sleep_secs = 0.5; +our $acc_record_limit = 1000; + +sub update_settings { + + my ($data,$configfile) = @_; + + if (defined $data) { + + my $result = 1; + #my $regexp_result; + + #&$configurationinfocode("testinfomessage",$configlogger); + + #$result &= _prepare_working_paths(1); + + $dry = $data->{dry} if exists $data->{dry}; + $skip_errors = $data->{skip_errors} if exists $data->{skip_errors}; + + $process_acc_trash_multithreading = $data->{process_acc_trash_multithreading} if exists $data->{process_acc_trash_multithreading}; + $process_acc_trash_numofthreads = _get_numofthreads($cpucount,$data,'process_acc_trash_numofthreads'); + + $process_acc_trash_blocksize = $data->{process_acc_trash_blocksize} if exists $data->{process_acc_trash_blocksize}; + $delete_cdr = $data->{delete_cdr} if exists $data->{delete_cdr}; + + $sleep_secs = $data->{sleep_secs} if exists $data->{sleep_secs}; + $acc_record_limit = $data->{acc_record_limit} if exists $data->{acc_record_limit}; + + #if (defined $acc_record_limit and defined ) { + # configurationerror() + #} + + return $result; + + } + return 0; + +} + +sub check_dry { + + if ($dry) { + scriptinfo('running in dry mode - NGCP databases will not be modified',getlogger(__PACKAGE__)); + return 1; + } else { + scriptinfo('NO DRY MODE - NGCP DATABASES WILL BE MODIFIED!',getlogger(__PACKAGE__)); + if (!$force) { + if ('yes' eq lc(prompt("Type 'yes' to proceed: "))) { + return 1; + } else { + return 0; + } + } else { + scriptinfo('force option applied',getlogger(__PACKAGE__)); + return 1; + } + } + +} + +sub _get_numofthreads { + my ($default_value,$data,$key) = @_; + my $_numofthreads = $default_value; + $_numofthreads = $data->{$key} if exists $data->{$key}; + $_numofthreads = $cpucount if $_numofthreads > $cpucount; + return $_numofthreads; +} + +1; diff --git a/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/config.cfg b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/config.cfg new file mode 100644 index 00000000..79ffebdc --- /dev/null +++ b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/config.cfg @@ -0,0 +1,62 @@ +##general settings: +working_path = /var/sipwise +cpucount = 4 +enablemultithreading = 1 + +##gearman/service listener config: +jobservers = 127.0.0.1:4730 + +#provisioning_conf = /etc/ngcp-panel/provisioning.conf + +##NGCP MySQL connectivity - "accounting" db: +accounting_host = 192.168.0.84 +accounting_port = 3306 +accounting_databasename = accounting +accounting_username = root +accounting_password = + +##NGCP MySQL connectivity - "billing" db: +billing_host = 192.168.0.74 +billing_port = 3306 +billing_databasename = billing +billing_username = root +billing_password = + +##NGCP MySQL connectivity - "provisioning" db: +provisioning_host = 192.168.0.74 +provisioning_port = 3306 +provisioning_databasename = provisioning +provisioning_username = root +provisioning_password = + +##NGCP MySQL connectivity - "kamailio" db: +kamailio_host = 192.168.0.84 +kamailio_port = 3306 +kamailio_databasename = kamailio +kamailio_username = root +kamailio_password = + +##NGCP MySQL connectivity - default db for distributed transactions (XA) to connect to: +xa_host = 192.168.0.84 +xa_port = 3306 +xa_databasename = ngcp +xa_username = root +xa_password = + +##NGCP REST-API connectivity: +ngcprestapi_uri = https://127.0.0.1:1443 +ngcprestapi_username = administrator +ngcprestapi_password = administrator +ngcprestapi_realm = api_admin_http + +##sending email: +emailenable = 0 +erroremailrecipient = +warnemailrecipient = +completionemailrecipient = rkrenn@sipwise.com +doneemailrecipient = + +##logging: +fileloglevel = DEBUG +screenloglevel = INFO +emailloglevel = OFF diff --git a/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/process.pl b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/process.pl new file mode 100644 index 00000000..cddfc534 --- /dev/null +++ b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/process.pl @@ -0,0 +1,250 @@ +use strict; + +## no critic + +use File::Basename; +use Cwd; +use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../../'); + +use Getopt::Long qw(GetOptions); +use Fcntl qw(LOCK_EX LOCK_NB); + +use NGCP::BulkProcessor::Globals qw(); +use NGCP::BulkProcessor::Projects::Disaster::Acc::Settings qw( + update_settings + check_dry + + $defaultsettings + $defaultconfig + $dry + $skip_errors + $force + +); + +use NGCP::BulkProcessor::Logging qw( + init_log + getlogger + $attachmentlogfile + scriptinfo + cleanuplogfiles + $currentlogfile +); +use NGCP::BulkProcessor::LogError qw ( + completion + done + scriptwarn + scripterror + filewarn + fileerror +); +use NGCP::BulkProcessor::LoadConfig qw( + load_config + $SIMPLE_CONFIG_TYPE + $YAML_CONFIG_TYPE + $ANY_CONFIG_TYPE +); +use NGCP::BulkProcessor::Array qw(removeduplicates); +use NGCP::BulkProcessor::Utils qw(getscriptpath prompt cleanupdir); +use NGCP::BulkProcessor::Mail qw( + cleanupmsgfiles +); + +use NGCP::BulkProcessor::Dao::Trunk::accounting::cdr qw(); +use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc qw(); +use NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash qw(); + +use NGCP::BulkProcessor::Projects::Disaster::Acc::Check qw( + check_accounting_db_tables + check_kamailio_db_tables +); +#check_rest_get_items + +use NGCP::BulkProcessor::Projects::Disaster::Acc::AccTrash qw( + process_acc_trash +); + +#use NGCP::BulkProcessor::Projects::Disaster::Balances::Api qw( +# set_call_forwards +# set_call_forwards_batch +#); + +use NGCP::BulkProcessor::ConnectorPool qw( + destroy_dbs +); + +scripterror(getscriptpath() . ' already running',getlogger(getscriptpath())) unless flock DATA, LOCK_EX | LOCK_NB; # not tested on windows yet + +my @TASK_OPTS = (); + +my $tasks = []; + +my $check_task_opt = 'check'; +push(@TASK_OPTS,$check_task_opt); + +my $cleanup_task_opt = 'cleanup'; +push(@TASK_OPTS,$cleanup_task_opt); + +my $process_acc_trash_task_opt = 'process_acc_trash'; +push(@TASK_OPTS,$process_acc_trash_task_opt); + +if (init()) { + main(); + exit(0); +} else { + exit(1); +} + +sub init { + + my $configfile = $defaultconfig; + my $settingsfile = $defaultsettings; + + return 0 unless GetOptions( + "config=s" => \$configfile, + "settings=s" => \$settingsfile, + "task=s" => $tasks, + "dry" => \$dry, + "skip-errors" => \$skip_errors, + "force" => \$force, + ); # or scripterror('error in command line arguments',getlogger(getscriptpath())); + + $tasks = removeduplicates($tasks,1); + + my $result = load_config($configfile); + init_log(); + $result &= load_config($settingsfile,\&update_settings,$SIMPLE_CONFIG_TYPE); + return $result; + +} + +sub main() { + + my @messages = (); + my @attachmentfiles = (); + my $result = 1; + my $completion = 0; + + if (defined $tasks and 'ARRAY' eq ref $tasks and (scalar @$tasks) > 0) { + scriptinfo('skip-errors: processing won\'t stop upon errors',getlogger(__PACKAGE__)) if $skip_errors; + foreach my $task (@$tasks) { + + if (lc($check_task_opt) eq lc($task)) { + $result &= check_task(\@messages) if taskinfo($check_task_opt,$result); + } elsif (lc($cleanup_task_opt) eq lc($task)) { + $result &= cleanup_task(\@messages) if taskinfo($cleanup_task_opt,$result); + + } elsif (lc($process_acc_trash_task_opt) eq lc($task)) { + if (taskinfo($process_acc_trash_task_opt,$result)) { + next unless check_dry(); + $result &= process_acc_trash_task(\@messages); + $completion |= 1; + } + + } else { + $result = 0; + scripterror("unknow task option '" . $task . "', must be one of " . join(', ',@TASK_OPTS),getlogger(getscriptpath())); + last; + } + } + } else { + $result = 0; + scripterror('at least one task option is required. supported tasks: ' . join(', ',@TASK_OPTS),getlogger(getscriptpath())); + } + + push(@attachmentfiles,$attachmentlogfile); + if ($completion) { + completion(join("\n\n",@messages),\@attachmentfiles,getlogger(getscriptpath())); + } else { + done(join("\n\n",@messages),\@attachmentfiles,getlogger(getscriptpath())); + } + + return $result; +} + +sub taskinfo { + my ($task,$result) = @_; + scriptinfo($result ? "starting task: '$task'" : "skipping task '$task' due to previous problems",getlogger(getscriptpath())); + return $result; +} + +sub check_task { + my ($messages) = @_; + my @check_messages = (); + my $result = check_kamailio_db_tables(\@check_messages); + #$result &= .. + push(@$messages,join("\n",@check_messages)); + + @check_messages = (); + $result &= check_accounting_db_tables(\@check_messages); + ##$result &= .. + push(@$messages,join("\n",@check_messages)); + + + destroy_dbs(); + return $result; +} + +sub cleanup_task { + my ($messages) = @_; + my $result = 0; + eval { + #cleanupcvsdirs() if $clean_generated; + #cleanupdbfiles() if $clean_generated; + cleanuplogfiles(\&fileerror,\&filewarn,($currentlogfile,$attachmentlogfile)); + cleanupmsgfiles(\&fileerror,\&filewarn); + #cleanupdir($output_path,1,\&filewarn,getlogger(getscriptpath())) if $clean_generated; + #cleanupdir($rollback_path,1,\&filewarn,getlogger(getscriptpath())) if $clean_generated; + $result = 1; + }; + + if ($@ or !$result) { + push(@$messages,'cleanup INCOMPLETE'); + return 0; + } else { + push(@$messages,'cleanup completed'); + return 1; + } +} + + + +sub process_acc_trash_task { + + my ($messages) = @_; + my ($result,$warning_count) = (0,0); + eval { + ($result,$warning_count) = process_acc_trash(1,1); + #if ($batch) { + # ($result,$warning_count) = set_barring_profiles_batch(); + #} else { + # ($result,$warning_count) = set_barring_profiles(); + #} + }; + my $err = $@; + my $stats = ($skip_errors ? ": $warning_count warnings" : ''); + eval { + $stats .= "\n total acc trash records left: " . + NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::count() . ' rows'; + $stats .= "\n total acc trash callids left: " . + NGCP::BulkProcessor::Dao::Trunk::kamailio::acc_trash::count_callids(); + }; + if ($err or !$result) { + push(@$messages,"process acc trash INCOMPLETE$stats"); + } else { + push(@$messages,"process acc trash completed$stats"); + } + destroy_dbs(); #every task should leave with closed connections. + return $result; + +} + +#END { +# # this should not be required explicitly, but prevents Log4Perl's +# # "rootlogger not initialized error upon exit.. +# destroy_all_dbs +#} + +__DATA__ +This exists to allow the locking code at the beginning of the file to work. +DO NOT REMOVE THESE LINES! diff --git a/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/settings.cfg b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/settings.cfg new file mode 100644 index 00000000..fff14993 --- /dev/null +++ b/lib/NGCP/BulkProcessor/Projects/Disaster/Acc/settings.cfg @@ -0,0 +1,12 @@ + +#dry=0 +#skip_errors=0 + +process_acc_trash_blocksize = 300 +process_acc_trash_multithreading = 1 +process_acc_trash_numofthreads = 1 + +delete_cdr = 1 + +sleep_secs = 2.0 +acc_record_limit = 2000 diff --git a/lib/NGCP/BulkProcessor/SqlProcessor.pm b/lib/NGCP/BulkProcessor/SqlProcessor.pm index 8964b14d..799eb108 100644 --- a/lib/NGCP/BulkProcessor/SqlProcessor.pm +++ b/lib/NGCP/BulkProcessor/SqlProcessor.pm @@ -740,6 +740,7 @@ sub transfer_table { $texttable_engine, $fixtable_statements, $multithreading, + $blocksize, $destroy_source_dbs_code, $destroy_target_dbs_code, $selectcount, @@ -755,6 +756,7 @@ sub transfer_table { texttable_engine fixtable_statements multithreading + blocksize destroy_source_dbs_code destroy_target_dbs_code selectcount @@ -813,11 +815,9 @@ sub transfer_table { my $insertstatement = 'INSERT INTO ' . $target_db->tableidentifier($targettablename) . ' (' . join(', ',map { local $_ = $_; $_ = $target_db->columnidentifier($_); $_; } @fieldnames) . ') VALUES (' . $valueplaceholders . ')'; - my $blocksize; - if ($enablemultithreading and $multithreading and $db->multithreading_supported() and $target_db->multithreading_supported() and $cpucount > 1) { # and $multithreaded) { # definitely no multithreading when CSVDB is involved - $blocksize = _calc_blocksize($rowcount,scalar @fieldnames,1,$tabletransfer_threadqueuelength); + $blocksize //= _calc_blocksize($rowcount,scalar @fieldnames,1,$tabletransfer_threadqueuelength); my $reader; my $writer; @@ -895,7 +895,7 @@ sub transfer_table { } else { - $blocksize = _calc_blocksize($rowcount,scalar @fieldnames,0,undef); + $blocksize //= _calc_blocksize($rowcount,scalar @fieldnames,0,undef); #$db->db_disconnect(); #undef $db; @@ -1027,6 +1027,7 @@ sub process_table { $init_process_context_code, $uninit_process_context_code, $multithreading, + $blocksize, $tableprocessing_threads, $destroy_reader_dbs_code, $selectcount, @@ -1039,6 +1040,7 @@ sub process_table { init_process_context_code uninit_process_context_code multithreading + blocksize tableprocessing_threads destroy_reader_dbs_code selectcount @@ -1082,12 +1084,12 @@ sub process_table { } my $errorstate = $RUNNING; - my $blocksize; + #my $blocksize; if ($enablemultithreading and $multithreading and $db->multithreading_supported() and $cpucount > 1) { # and $multithreaded) { # definitely no multithreading when CSVDB is involved $tableprocessing_threads //= $cpucount; - $blocksize = _calc_blocksize($rowcount,scalar @fieldnames,1,$tableprocessing_threadqueuelength); + $blocksize //= _calc_blocksize($rowcount,scalar @fieldnames,1,$tableprocessing_threadqueuelength); my $reader; #my $processor; @@ -1196,7 +1198,7 @@ sub process_table { } else { - $blocksize = _calc_blocksize($rowcount,scalar @fieldnames,0,undef); + $blocksize //= _calc_blocksize($rowcount,scalar @fieldnames,0,undef); #$db->db_disconnect(); #undef $db; #$db = &$get_db($reader_connection_name);