From 8bfaabb49a7551737f351fab6667dd51c810056d Mon Sep 17 00:00:00 2001 From: Richard Fuchs Date: Wed, 4 Sep 2019 12:56:40 -0400 Subject: [PATCH] TT#65453 implement intermediate CDR writing Change-Id: Ifa4b33276e2038c6bf075872dd1c5be187652d55 --- cdr.c | 53 +++++++++++++++--- cdr.h | 7 ++- config.c | 9 ++- config.h | 1 + mediator.c | 25 +++++++-- mediator.h | 2 + medmysql.c | 159 ++++++++++++++++++++++++++++++++++++++++++++++++++--- medmysql.h | 3 + medredis.c | 1 + records.c | 4 +- 10 files changed, 241 insertions(+), 23 deletions(-) diff --git a/cdr.c b/cdr.c index 0f473e0..d12e005 100644 --- a/cdr.c +++ b/cdr.c @@ -8,7 +8,7 @@ #include "mediator.h" static int cdr_create_cdrs(med_entry_t *records, uint64_t count, - cdr_entry_t **cdrs, uint64_t *cdr_count, uint8_t *trash); + cdr_entry_t **cdrs, uint64_t *cdr_count, uint8_t *trash, int do_intermediate); static const char* cdr_map_status(const char *sip_status) { @@ -40,11 +40,13 @@ static const char* cdr_map_status(const char *sip_status) return CDR_STATUS_UNKNOWN; } -int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_count, struct medmysql_batches *batches) +int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_count, + struct medmysql_batches *batches, int do_intermediate) { int ret = 0; uint8_t trash = 0; uint64_t i; + int timed_out = 0; uint16_t msg_invites = 0; uint16_t msg_byes = 0; @@ -63,6 +65,10 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun for(i = 0; i < count; ++i) { med_entry_t *e = &(records[i]); + + if (e->timed_out) + timed_out = 1; + if(!e->valid) { ++msg_unknowns; @@ -102,7 +108,7 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun if(msg_invites > 0) { - if(msg_byes > 0 || invite_200 == 0) + if(msg_byes > 0 || do_intermediate || timed_out || invite_200 == 0) { if(/*msg_byes > 2*/ 0) { @@ -112,7 +118,7 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun } else { - if(cdr_create_cdrs(records, count, &cdrs, &cdr_count, &trash) != 0) + if(cdr_create_cdrs(records, count, &cdrs, &cdr_count, &trash, do_intermediate) != 0) goto error; else { @@ -124,11 +130,13 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun /* cdr_log_records(cdrs, cdr_count); */ } - if(medmysql_insert_cdrs(cdrs, cdr_count, batches) != 0) + int insert_ret = medmysql_insert_cdrs(cdrs, cdr_count, batches); + + if(insert_ret < 0) { goto error; } - else + else if (insert_ret == 0) { if (has_redis) { @@ -140,7 +148,16 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun if(medmysql_backup_entries(callid, batches) != 0) goto error; } + if (config_intermediate_interval) + { + if(medmysql_delete_intermediate(cdrs, cdr_count, batches) != 0) + goto error; + } } + else if (insert_ret == 1) + ; // do nothing - intermediate CDR has been created + else + goto error; } else @@ -741,11 +758,12 @@ err: static int cdr_create_cdrs(med_entry_t *records, uint64_t count, - cdr_entry_t **cdrs, uint64_t *cdr_count, uint8_t *trash) + cdr_entry_t **cdrs, uint64_t *cdr_count, uint8_t *trash, int do_intermediate) { uint64_t i = 0, cdr_index = 0; uint32_t invites = 0; size_t cdr_size; + int timed_out = 0; char *endtime = NULL; double unix_endtime = 0, tmp_unix_endtime = 0; @@ -760,6 +778,9 @@ static int cdr_create_cdrs(med_entry_t *records, uint64_t count, for(i = 0; i < count; ++i) { med_entry_t *e = &(records[i]); + if (e->timed_out) + timed_out = 1; + if(e->valid && e->method == MED_INVITE) { ++invites; @@ -810,6 +831,9 @@ static int cdr_create_cdrs(med_entry_t *records, uint64_t count, L_DEBUG("create cdr %lu of %lu in batch\n", i, count); call_status = cdr_map_status(e->sip_code); + if (timed_out) + call_status = CDR_STATUS_FAILED; + if(e->method == MED_INVITE && call_status != NULL) { ++cdr_index; @@ -824,6 +848,16 @@ static int cdr_create_cdrs(med_entry_t *records, uint64_t count, else { tmp_unix_endtime = unix_endtime; + if (!tmp_unix_endtime) { + if (do_intermediate && !timed_out) { + L_DEBUG("CDR %lu is an intermediate record\n", cdr_index); + cdr->intermediate = 1; + } + else { + L_DEBUG("CDR %lu is an expired record\n", cdr_index); + } + tmp_unix_endtime = time(NULL); + } } g_strlcpy(cdr->call_id, e->callid, sizeof(cdr->call_id)); @@ -841,13 +875,13 @@ static int cdr_create_cdrs(med_entry_t *records, uint64_t count, cdr->destination_reseller_cost = 0; cdr->destination_customer_cost = 0; - if(cdr_parse_srcleg(e->src_leg, cdr) < 0) + if(cdr_parse_srcleg(e->src_leg, cdr) < 0 && !cdr->intermediate) { *trash = 1; return 0; } - if(cdr_parse_dstleg(e->dst_leg, cdr) < 0) + if(cdr_parse_dstleg(e->dst_leg, cdr) < 0 && !cdr->intermediate && !timed_out) { *trash = 1; return 0; @@ -859,6 +893,7 @@ static int cdr_create_cdrs(med_entry_t *records, uint64_t count, } cdr->mos = mos_data; + g_strlcpy(cdr->acc_ref, e->acc_ref, sizeof(cdr->acc_ref)); L_DEBUG("Created CDR index %lu\n", cdr_index); } diff --git a/cdr.h b/cdr.h index a1a5119..40175fd 100644 --- a/cdr.h +++ b/cdr.h @@ -12,6 +12,7 @@ #define CDR_STATUS_CANCEL "cancel" #define CDR_STATUS_OFFLINE "offline" #define CDR_STATUS_TIMEOUT "timeout" +#define CDR_STATUS_FAILED "failed" #define CDR_STATUS_UNKNOWN "other" struct medmysql_batches; @@ -86,9 +87,13 @@ typedef struct { uint8_t split; mos_data_t mos; + + char acc_ref[256]; + int intermediate:1; } cdr_entry_t; -int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *cdr_count, struct medmysql_batches *); +int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *cdr_count, struct medmysql_batches *, + int do_intermediate); void cdr_fix_accids(med_entry_t *records, uint64_t count); int cdr_fill_record(cdr_entry_t *cdr); void cdr_set_provider(cdr_entry_t *cdr); diff --git a/config.c b/config.c index 42610bf..719e9e3 100644 --- a/config.c +++ b/config.c @@ -43,6 +43,7 @@ med_stats_period_t config_stats_period = MEDIATOR_DEFAULT_STATSPERIOD; int config_maintenance = 0; int strict_leg_tokens = 0; int config_max_acc_age = 0; +int config_intermediate_interval = 0; med_loglevel_t config_loglevel = MEDIATOR_DEFAULT_LOGLEVEL; @@ -81,9 +82,10 @@ enum config_option { OPT_MAINTENANCE = 'm', OPT_LEG_TOKENS = 's', OPT_MAX_ACC_AGE = 'M', + OPT_INTERMEDIATE_INTERVAL = 'I', }; -static const char options[] = "?a:c:e:D:i:dlL:h:u:p:b:o:H:U:P:B:O:S:t:T:r:R:A:N:Z:z:W:w:X:x:msM:"; +static const char options[] = "?a:c:e:D:i:dlL:h:u:p:b:o:H:U:P:B:O:S:t:T:r:R:A:N:Z:z:W:w:X:x:msM:I:"; struct option long_options[] = { { "configfile", required_argument, NULL, OPT_CONFIGFILE }, @@ -120,6 +122,7 @@ struct option long_options[] = { { "maintenance", no_argument, NULL, OPT_MAINTENANCE }, { "leg-tokens", no_argument, NULL, OPT_LEG_TOKENS }, { "max-acc-age", required_argument, NULL, OPT_MAX_ACC_AGE }, + { "intermediate-interval", required_argument, NULL, OPT_INTERMEDIATE_INTERVAL }, { NULL, 0, NULL, 0 }, }; @@ -164,6 +167,7 @@ static void config_help(const char *self, int rc) " -m, --maintenance\tMaintenance mode (do nothing, just sleep).\n" \ " -s, --leg-tokens\tStrict acc fields (move to trash otherwise).\n" \ " -M, --max-acc-age\tMaximum age of acc records before trashing them (default = disabled).\n" \ +" -I, --intermediate-interval\tHow often to write/update intermediate CDRs (default = disabled).\n" \ " -?, --help\t\tDisplays this message.\n", MEDIATOR_VERSION, self, MEDIATOR_DEFAULT_CONFIG_FILE, MEDIATOR_DEFAULT_PIDPATH, MEDIATOR_DEFAULT_LOGLEVEL, @@ -303,6 +307,9 @@ static void config_set_option(enum config_option option, const char *value) case OPT_MAX_ACC_AGE: config_max_acc_age = atoi(value); break; + case OPT_INTERMEDIATE_INTERVAL: + config_intermediate_interval = atoi(value); + break; } } diff --git a/config.h b/config.h index 2d5b745..b17d047 100644 --- a/config.h +++ b/config.h @@ -44,6 +44,7 @@ extern uint8_t config_dumpcdr; extern int config_maintenance; extern int strict_leg_tokens; extern int config_max_acc_age; +extern int config_intermediate_interval; extern med_loglevel_t config_loglevel; diff --git a/mediator.c b/mediator.c index d310645..3817297 100644 --- a/mediator.c +++ b/mediator.c @@ -23,6 +23,7 @@ sig_atomic_t mediator_shutdown = 0; int mediator_lockfd = -1; uint64_t mediator_count = 0; +static time_t next_intermediate_run = 0; GHashTable *med_peer_ip_table = NULL; GHashTable *med_peer_host_table = NULL; @@ -271,6 +272,18 @@ int main(int argc, char **argv) if (0) mediator_print_maps(); + // process intermediate CDRs this round? + int do_intermediate = 0; + if (config_intermediate_interval > 0) { + if (loop_tv_start.tv_sec >= next_intermediate_run) { + L_DEBUG("Processing intermediate CDRs in this iteration\n"); + do_intermediate = 1; + if (next_intermediate_run == 0) + next_intermediate_run = loop_tv_start.tv_sec; + next_intermediate_run += config_intermediate_interval; + } + } + mysql_id_count = redis_id_count = mysql_rec_count = redis_rec_count = cdr_count = 0; last_count = mediator_count; @@ -326,14 +339,16 @@ int main(int argc, char **argv) records_sort(mysql_records, mysql_rec_count); } - if (!records_complete(mysql_records, mysql_rec_count)) + int are_records_complete = records_complete(mysql_records, mysql_rec_count); + + if (!are_records_complete && !do_intermediate) { L_DEBUG("Found incomplete call with cid '%s', skipping...\n", mysql_callids[i].value); free(mysql_records); continue; } - if(cdr_process_records(mysql_records, mysql_rec_count, &cdr_count, batches) != 0) + if(cdr_process_records(mysql_records, mysql_rec_count, &cdr_count, batches, do_intermediate) != 0) goto out; if(mysql_rec_count > 0) @@ -384,7 +399,9 @@ int main(int argc, char **argv) // always sort records from Redis, regardless of whether records from MySQL were merged records_sort(redis_records, redis_rec_count); - if (!records_complete(redis_records, redis_rec_count)) + int are_records_complete = records_complete(redis_records, redis_rec_count); + + if (!are_records_complete && !do_intermediate) { L_DEBUG("Found incomplete call with cid '%s', skipping...\n", redis_callids[i].value); free(redis_records); @@ -394,7 +411,7 @@ int main(int argc, char **argv) L_DEBUG("process cdr with cid '%s' and %"PRIu64" records\n", redis_callids[i].value, redis_rec_count); if (redis_rec_count) { - if(cdr_process_records(redis_records, redis_rec_count, &cdr_count, batches) != 0) { + if(cdr_process_records(redis_records, redis_rec_count, &cdr_count, batches, do_intermediate) != 0) { free(redis_records); goto out; } diff --git a/mediator.h b/mediator.h index 861a7b6..0eecd44 100644 --- a/mediator.h +++ b/mediator.h @@ -87,6 +87,8 @@ typedef struct { med_method_t method; char sip_method[32]; uint8_t redis; + char acc_ref[256]; + uint8_t timed_out; } med_entry_t; typedef struct { diff --git a/medmysql.c b/medmysql.c index 18e3e22..e754f9d 100644 --- a/medmysql.c +++ b/medmysql.c @@ -13,16 +13,16 @@ " group by a.callid limit 0,200000" #define MED_FETCH_QUERY "(select distinct sip_code, sip_reason, method, callid, time, time_hires, " \ - "src_leg, dst_leg, branch_id " \ + "src_leg, dst_leg, branch_id, id " \ "from acc where method = 'INVITE' and callid = '%s' order by time_hires asc) " \ "union all " \ "(select distinct sip_code, sip_reason, method, callid, time, time_hires, " \ - "src_leg, dst_leg, branch_id " \ + "src_leg, dst_leg, branch_id, id " \ "from acc where method = 'BYE' and callid in ('%s', '%s"PBXSUFFIX"') " \ "order by length(callid) asc, time_hires asc) " \ "union all " \ "(select distinct sip_code, sip_reason, method, callid, time, time_hires, " \ - "src_leg, dst_leg, branch_id " \ + "src_leg, dst_leg, branch_id, id " \ "from acc where method = 'BYE' and callid in ('%s', '%s"XFERSUFFIX"') " \ "order by length(callid) asc, time_hires asc)" @@ -72,6 +72,7 @@ static unsigned long medmysql_tag_direction_source; static unsigned long medmysql_tag_direction_destination; static int medmysql_flush_cdr(struct medmysql_batches *); +static int medmysql_flush_int_cdr(struct medmysql_batches *); static int medmysql_flush_all_med(struct medmysql_batches *); static int medmysql_flush_med_str(struct medmysql_str *); static int medmysql_flush_medlist(struct medmysql_str *); @@ -133,12 +134,94 @@ static const medmysql_batch_definition medmysql_cdr_def = { .min_string_tail_room = 6000, .handler_ptr = &cdr_handler, }; +static const medmysql_batch_definition medmysql_int_cdr_def = { + .sql_init_string = "insert into int_cdr (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, "\ + "destination_user_id, destination_provider_id, destination_external_subscriber_id, "\ + "destination_external_contract_id, destination_account_id, destination_user, destination_domain, " \ + "destination_user_in, destination_domain_in, destination_user_dialed, " \ + "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, " \ + "destination_carrier_cost, destination_reseller_cost, destination_customer_cost, " \ + "split, " \ + "source_gpp0, source_gpp1, source_gpp2, source_gpp3, source_gpp4, " \ + "source_gpp5, source_gpp6, source_gpp7, source_gpp8, source_gpp9, " \ + "destination_gpp0, destination_gpp1, destination_gpp2, destination_gpp3, destination_gpp4, " \ + "destination_gpp5, destination_gpp6, destination_gpp7, destination_gpp8, destination_gpp9, " \ + "source_lnp_prefix, destination_lnp_prefix, " \ + "source_user_out, destination_user_out, " \ + "source_lnp_type, destination_lnp_type, acc_ref" \ + ") values ", + .sql_finish_string = " on duplicate key update " + "update_time = values(update_time), source_user_id = values(source_user_id), " \ + "source_provider_id = values(source_provider_id), source_external_subscriber_id " \ + "= values(source_external_subscriber_id), source_external_contract_id = " \ + "values(source_external_contract_id), source_account_id = " \ + "values(source_account_id), source_user = values(source_user), source_domain = " \ + "values(source_domain), source_cli = values(source_cli), source_clir = " \ + "values(source_clir), source_ip = values(source_ip), destination_user_id = " \ + "values(destination_user_id), destination_provider_id = " \ + "values(destination_provider_id), destination_external_subscriber_id = " \ + "values(destination_external_subscriber_id), destination_external_contract_id = " \ + "values(destination_external_contract_id), destination_account_id = " \ + "values(destination_account_id), destination_user = values(destination_user), " \ + "destination_domain = values(destination_domain), destination_user_in = " \ + "values(destination_user_in), destination_domain_in = " \ + "values(destination_domain_in), destination_user_dialed = " \ + "values(destination_user_dialed), peer_auth_user = values(peer_auth_user), " \ + "peer_auth_realm = values(peer_auth_realm), call_type = values(call_type), " \ + "call_status = values(call_status), call_code = values(call_code), init_time = " \ + "values(init_time), start_time = values(start_time), duration = " \ + "values(duration), call_id = values(call_id), source_carrier_cost = " \ + "values(source_carrier_cost), source_reseller_cost = " \ + "values(source_reseller_cost), source_customer_cost = " \ + "values(source_customer_cost), destination_carrier_cost = " \ + "values(destination_carrier_cost), destination_reseller_cost = " \ + "values(destination_reseller_cost), destination_customer_cost = " \ + "values(destination_customer_cost), split = values(split), source_gpp0 = " \ + "values(source_gpp0), source_gpp1 = values(source_gpp1), source_gpp2 = " \ + "values(source_gpp2), source_gpp3 = values(source_gpp3), source_gpp4 = " \ + "values(source_gpp4), source_gpp5 = values(source_gpp5), source_gpp6 = " \ + "values(source_gpp6), source_gpp7 = values(source_gpp7), source_gpp8 = " \ + "values(source_gpp8), source_gpp9 = values(source_gpp9), destination_gpp0 = " \ + "values(destination_gpp0), destination_gpp1 = values(destination_gpp1), " \ + "destination_gpp2 = values(destination_gpp2), destination_gpp3 = " \ + "values(destination_gpp3), destination_gpp4 = values(destination_gpp4), " \ + "destination_gpp5 = values(destination_gpp5), destination_gpp6 = " \ + "values(destination_gpp6), destination_gpp7 = values(destination_gpp7), " \ + "destination_gpp8 = values(destination_gpp8), destination_gpp9 = " \ + "values(destination_gpp9), source_lnp_prefix = values(source_lnp_prefix), " \ + "destination_lnp_prefix = values(destination_lnp_prefix), source_user_out = " \ + "values(source_user_out), destination_user_out = values(destination_user_out), " \ + "source_lnp_type = values(source_lnp_type), destination_lnp_type = " \ + "values(destination_lnp_type), export_status = 'unexported'", + .full_flush_func = medmysql_flush_int_cdr, + .min_string_tail_room = 9000, + .handler_ptr = &cdr_handler, +}; +static const medmysql_batch_definition medmysql_del_int_cdr_def = { + .sql_init_string = "delete from int_cdr where call_id in (", + .sql_finish_string = ")", + .single_flush_func = medmysql_flush_med_str, + .handler_ptr = &cdr_handler, +}; static const medmysql_batch_definition medmysql_tag_def = { .sql_init_string = "insert into cdr_tag_data (cdr_id, provider_id, direction_id, tag_id, " \ "val, cdr_start_time) values ", .single_flush_func = medmysql_flush_med_str, .handler_ptr = &cdr_handler, }; +static const medmysql_batch_definition medmysql_int_tag_def = { + .sql_init_string = "insert into int_cdr_tag_data (cdr_id, provider_id, direction_id, tag_id, " \ + "val, cdr_start_time) values ", + .sql_finish_string = " on duplicate key update " + "val = values(val)", + .single_flush_func = medmysql_flush_med_str, + .handler_ptr = &cdr_handler, +}; static const medmysql_batch_definition medmysql_mos_def = { .sql_init_string = "insert into cdr_mos_data (" \ "cdr_id, mos_average, mos_average_packetloss, mos_average_jitter, " \ @@ -154,6 +237,13 @@ static const medmysql_batch_definition medmysql_group_def = { .single_flush_func = medmysql_flush_med_str, .handler_ptr = &cdr_handler, }; +static const medmysql_batch_definition medmysql_int_group_def = { + .sql_init_string = "insert ignore into int_cdr_group (" \ + "cdr_id, call_id, cdr_start_time" \ + ") values ", + .single_flush_func = medmysql_flush_med_str, + .handler_ptr = &cdr_handler, +}; static void statement_free(void *stm_p) { @@ -618,6 +708,7 @@ int medmysql_fetch_records(med_callid_t *callid, g_strlcpy(e->src_leg, row[6], sizeof(e->src_leg)); g_strlcpy(e->dst_leg, row[7], sizeof(e->dst_leg)); g_strlcpy(e->branch_id, row[8], sizeof(e->branch_id)); + g_strlcpy(e->acc_ref, row[9], sizeof(e->acc_ref)); e->valid = 1; if (check_shutdown()) @@ -769,12 +860,17 @@ int medmysql_insert_cdrs(cdr_entry_t *entries, uint64_t count, struct medmysql_b uint64_t i; int gpp; gpointer tag_id; + int ret = 1; // default to intermediate for(i = 0; i < count; ++i) { cdr_entry_t *e = &(entries[i]); struct medmysql_cdr_batch *batch = &batches->cdr_batch; + if (e->intermediate) + batch = &batches->int_cdr_batch; + else + ret = 0; // if at least one record is not intermediate, we delete if (medmysql_batch_prepare(&batch->cdrs)) return -1; @@ -939,6 +1035,12 @@ int medmysql_insert_cdrs(cdr_entry_t *entries, uint64_t count, struct medmysql_b CDRPRINT(",NULL"); } + if (batch == &batches->int_cdr_batch) { + CDRPRINT(",'"); + CDRESCAPE(e->acc_ref); + CDRPRINT("'"); + } + CDRPRINT("),"); if(strnlen(e->furnished_charging_info, sizeof(e->furnished_charging_info)) > 0) @@ -993,7 +1095,7 @@ int medmysql_insert_cdrs(cdr_entry_t *entries, uint64_t count, struct medmysql_b return -1; } - if (e->mos.filled) { + if (e->mos.filled && batch->mos.def) { if (medmysql_mos_record(&batch->cdr_mos, batch->num_cdrs, e->mos.avg_score, e->mos.avg_packetloss, e->mos.avg_jitter, e->mos.avg_rtt, e->start_time)) @@ -1015,7 +1117,27 @@ int medmysql_insert_cdrs(cdr_entry_t *entries, uint64_t count, struct medmysql_b /*L_DEBUG("q='%s'", query);*/ + return ret; +} + +/**********************************************************************/ +int medmysql_delete_intermediate(cdr_entry_t *entries, uint64_t count, struct medmysql_batches *batches) +{ + uint64_t i; + + for(i = 0; i < count; ++i) + { + cdr_entry_t *e = &(entries[i]); + char *callid = e->call_id; + char esc_callid[strlen(callid)*2+1]; + + mysql_real_escape_string(cdr_handler->m, esc_callid, callid, strlen(callid)); + if (medmysql_batch_prepare(&batches->int_cdr_delete)) + return -1; + batches->int_cdr_delete.len += sprintf(batches->int_cdr_delete.str + batches->int_cdr_delete.len, + "'%s',", esc_callid); + } return 0; } @@ -1290,9 +1412,20 @@ int medmysql_batch_start(struct medmysql_batches *batches) { batches->cdr_batch.num_cdrs = 0; + medmysql_str_init(&batches->int_cdr_batch.cdrs, &medmysql_int_cdr_def, batches, &batches->int_cdr_batch, NULL); + medmysql_str_init(&batches->int_cdr_batch.tags, &medmysql_int_tag_def, batches, &batches->int_cdr_batch, + &batches->int_cdr_batch.cdr_tags); + medmysql_str_init(&batches->int_cdr_batch.mos, NULL, batches, &batches->int_cdr_batch, + &batches->int_cdr_batch.cdr_mos); + medmysql_str_init(&batches->int_cdr_batch.group, &medmysql_int_group_def, batches, &batches->int_cdr_batch, + &batches->int_cdr_batch.cdr_group); + + batches->int_cdr_batch.num_cdrs = 0; + medmysql_str_init(&batches->acc_backup, &medmysql_backup_def, batches, NULL, NULL); medmysql_str_init(&batches->acc_trash, &medmysql_trash_def, batches, NULL, NULL); medmysql_str_init(&batches->to_delete, &medmysql_delete_def, batches, NULL, NULL); + medmysql_str_init(&batches->int_cdr_delete, &medmysql_del_int_cdr_def, batches, NULL, NULL); return 0; } @@ -1300,6 +1433,8 @@ int medmysql_batch_start(struct medmysql_batches *batches) { static int medmysql_flush_med_str(struct medmysql_str *str) { const medmysql_batch_definition *def = str->def; + if (!def) + return 0; if (str->len == 0) return 0; @@ -1309,12 +1444,12 @@ static int medmysql_flush_med_str(struct medmysql_str *str) { str->len--; str->str[str->len] = '\0'; - L_DEBUG("SQL flush med str\n"); - L_DEBUG("SQL: %.*s\n", str->len, str->str); - if (def->sql_finish_string) str->len += sprintf(str->str + str->len, "%s", def->sql_finish_string); + L_DEBUG("SQL flush med str\n"); + L_DEBUG("SQL: %.*s\n", str->len, str->str); + if (medmysql_query_wrapper_tx(*def->handler_ptr, str->str, str->len) != 0) { str->len = 0; L_CRITICAL("Error executing query: %s", @@ -1333,6 +1468,8 @@ static int medmysql_flush_med_str(struct medmysql_str *str) { static int medmysql_write_tag_records(struct medmysql_str *str, unsigned long long auto_id) { + if (!str->def) + return 0; cdr_tag_record *record; GQueue *q = str->q; while ((record = g_queue_pop_head(q))) { @@ -1405,9 +1542,15 @@ static int medmysql_flush_cdr_batch(struct medmysql_cdr_batch *batch) { } static int medmysql_flush_cdr(struct medmysql_batches *batches) { + if (medmysql_flush_med_str(&batches->int_cdr_delete)) + return -1; return medmysql_flush_cdr_batch(&batches->cdr_batch); } +static int medmysql_flush_int_cdr(struct medmysql_batches *batches) { + return medmysql_flush_cdr_batch(&batches->int_cdr_batch); +} + static int medmysql_flush_medlist(struct medmysql_str *str) { if (medmysql_flush_med_str(str)) { critical("Failed to execute potentially crucial SQL query, check LOG for details"); @@ -1468,6 +1611,8 @@ static int medmysql_flush_call_stat_info() { int medmysql_batch_end(struct medmysql_batches *batches) { if (medmysql_flush_cdr(batches) || check_shutdown()) return -1; + if (medmysql_flush_int_cdr(batches) || check_shutdown()) + return -1; if (medmysql_flush_all_med(batches) || check_shutdown()) return -1; if (medmysql_flush_call_stat_info() || check_shutdown()) diff --git a/medmysql.h b/medmysql.h index 29e226b..34ca8b7 100644 --- a/medmysql.h +++ b/medmysql.h @@ -36,10 +36,12 @@ struct medmysql_cdr_batch { struct medmysql_batches { struct medmysql_cdr_batch cdr_batch; + struct medmysql_cdr_batch int_cdr_batch; struct medmysql_str acc_backup; struct medmysql_str acc_trash; struct medmysql_str to_delete; + struct medmysql_str int_cdr_delete; }; struct medmysql_call_stat_info_t { @@ -56,6 +58,7 @@ int medmysql_trash_entries(const char *callid, struct medmysql_batches *); int medmysql_backup_entries(const char *callid, struct medmysql_batches *); int medmysql_delete_entries(const char *callid, struct medmysql_batches *); int medmysql_insert_cdrs(cdr_entry_t *records, uint64_t count, struct medmysql_batches *); +int medmysql_delete_intermediate(cdr_entry_t *records, uint64_t count, struct medmysql_batches *); int medmysql_load_maps(GHashTable *ip_table, GHashTable *host_table, GHashTable *id_table); int medmysql_load_uuids(GHashTable *uuid_table); int medmysql_load_db_ids(); diff --git a/medredis.c b/medredis.c index fbe7a3f..5bc59ba 100644 --- a/medredis.c +++ b/medredis.c @@ -373,6 +373,7 @@ static med_entry_t *medredis_reply_to_entry(redisReply *reply, const char* cid, } else { medredis_check_reply_string(entry, reply, entry->branch_id, "branch_id", sizeof(entry->branch_id), 8, cid, key); } + g_strlcpy(entry->acc_ref, key, sizeof(entry->acc_ref)); L_DEBUG("Converted record with cid '%s' and method '%s'\n", entry->callid, entry->sip_method); diff --git a/records.c b/records.c index a76dc1f..d7b40db 100644 --- a/records.c +++ b/records.c @@ -35,8 +35,10 @@ int records_complete(med_entry_t *records, uint64_t count) // if our records are old enough, we always consider them complete if (count && config_max_acc_age) { - if (time(NULL) - records[0].unix_timestamp > config_max_acc_age) + if (time(NULL) - records[0].unix_timestamp > config_max_acc_age) { + records[0].timed_out = 1; return 1; + } } for (uint64_t i = 0; i < count; i++)