/* ** Zabbix ** Copyright (C) 2001-2022 Zabbix SIA ** ** 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 2 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, write to the Free Software ** Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ #include "report_manager.h" #include "zbxself.h" #include "daemon.h" #include "base64.h" #include "zbxreport.h" #include "../../libs/zbxcrypto/hmac_sha256.h" #include "sha256crypt.h" #include "../../libs/zbxalgo/vectorimpl.h" #include "zbxalert.h" #include "zbxserver.h" #include "report_protocol.h" #define ZBX_REPORT_INCLUDE_USER 0 #define ZBX_REPORT_EXCLUDE_USER 1 #define ZBX_REPORT_UPDATE_LASTSENT 0x0001 #define ZBX_REPORT_UPDATE_STATE 0x0002 #define ZBX_REPORT_UPDATE_ERROR 0x0004 #define ZBX_REPORT_UPDATE (ZBX_REPORT_UPDATE_LASTSENT | ZBX_REPORT_UPDATE_STATE | ZBX_REPORT_UPDATE_ERROR) #define ZBX_REPORT_STATE_SUCCESS 1 #define ZBX_REPORT_STATE_ERROR 2 #define ZBX_REPORT_STATE_SUCCESS_INFO 3 extern ZBX_THREAD_LOCAL unsigned char process_type; extern unsigned char program_type; extern ZBX_THREAD_LOCAL int server_num, process_num; extern int CONFIG_REPORTWRITER_FORKS; /* report manager data */ typedef struct { /* config.url, config.session_key fields synced from database together with reports */ char *zabbix_url; char *session_key; /* the IPC service */ zbx_ipc_service_t ipc; /* the next writer index to be assigned to new IPC service clients */ int next_writer_index; /* identifier of the last batchid (autogenerated) */ zbx_uint64_t last_batchid; /* report writer vector, created during manager initialization */ zbx_vector_ptr_t writers; zbx_queue_ptr_t free_writers; zbx_hashset_t sessions; zbx_hashset_t reports; zbx_hashset_t batches; zbx_vector_uint64_t flush_queue; zbx_binary_heap_t report_queue; zbx_list_t job_queue; } zbx_rm_t; typedef struct { zbx_uint64_t id; zbx_uint64_t access_userid; } zbx_rm_recipient_t; ZBX_VECTOR_DECL(recipient, zbx_rm_recipient_t) ZBX_VECTOR_IMPL(recipient, zbx_rm_recipient_t) typedef struct { zbx_uint64_t reportid; zbx_uint64_t userid; zbx_uint64_t dashboardid; char *name; char *timezone; char *error; unsigned char period; unsigned char cycle; unsigned char weekdays; unsigned char status; int start_time; int state; zbx_uint32_t flags; int nextcheck; int active_since; int active_till; int lastsent; zbx_vector_ptr_pair_t params; zbx_vector_recipient_t usergroups; zbx_vector_recipient_t users; zbx_vector_uint64_t users_excl; } zbx_rm_report_t; typedef struct { zbx_uint64_t access_userid; zbx_uint64_t batchid; int report_width; int report_height; char *url; char *cookie; char *report_name; zbx_vector_uint64_t userids; zbx_vector_ptr_pair_t params; zbx_ipc_client_t *client; } zbx_rm_job_t; typedef struct { zbx_uint64_t batchid; zbx_uint64_t reportid; int error_num; int sent_num; int total_num; char *info; size_t info_alloc; size_t info_offset; zbx_vector_ptr_t jobs; } zbx_rm_batch_t; /* user session, cached to generate authentication cookies */ typedef struct { zbx_uint64_t userid; char *sid; char *cookie; int db_lastaccess; int lastaccess; } zbx_rm_session_t; typedef struct { /* the connected report writer client */ zbx_ipc_client_t *client; zbx_rm_job_t *job; } zbx_rm_writer_t; /****************************************************************************** * * * Purpose: return writer with the specified client * * * ******************************************************************************/ static zbx_rm_writer_t *rm_get_writer(zbx_rm_t *manager, const zbx_ipc_client_t *client) { int i; for (i = 0; i < manager->writers.values_num; i++) { zbx_rm_writer_t *writer = (zbx_rm_writer_t *)manager->writers.values[i]; if (writer->client == client) return writer; } return NULL; } static void rm_writer_free(zbx_rm_writer_t *writer) { zbx_ipc_client_close(writer->client); zbx_free(writer); } static int rm_report_compare_nextcheck(const void *d1, const void *d2) { const zbx_binary_heap_elem_t *e1 = (const zbx_binary_heap_elem_t *)d1; const zbx_binary_heap_elem_t *e2 = (const zbx_binary_heap_elem_t *)d2; return ((zbx_rm_report_t *)e1->data)->nextcheck - ((zbx_rm_report_t *)e2->data)->nextcheck; } static void rm_report_clean(zbx_rm_report_t *report) { zbx_free(report->name); zbx_free(report->timezone); zbx_free(report->error); report_destroy_params(&report->params); zbx_vector_recipient_destroy(&report->usergroups); zbx_vector_recipient_destroy(&report->users); zbx_vector_uint64_destroy(&report->users_excl); } static void rm_job_free(zbx_rm_job_t *job) { if (NULL != job->client) zbx_ipc_client_release(job->client); zbx_free(job->report_name); zbx_free(job->url); zbx_free(job->cookie); zbx_vector_uint64_destroy(&job->userids); report_destroy_params(&job->params); zbx_free(job); } static void rm_batch_clean(zbx_rm_batch_t *batch) { zbx_vector_ptr_clear_ext(&batch->jobs, (zbx_ptr_free_func_t)rm_job_free); zbx_vector_ptr_destroy(&batch->jobs); zbx_free(batch->info); } /****************************************************************************** * * * Purpose: initializes report manager * * * * Parameters: manager - [IN] the manager to initialize * * * ******************************************************************************/ static int rm_init(zbx_rm_t *manager, char **error) { int i, ret; zbx_rm_writer_t *writer; zabbix_log(LOG_LEVEL_DEBUG, "In %s() writers:%d", __func__, CONFIG_REPORTWRITER_FORKS); if (FAIL == (ret = zbx_ipc_service_start(&manager->ipc, ZBX_IPC_SERVICE_REPORTER, error))) goto out; zbx_vector_ptr_create(&manager->writers); zbx_queue_ptr_create(&manager->free_writers); zbx_hashset_create(&manager->sessions, 0, ZBX_DEFAULT_UINT64_HASH_FUNC, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_hashset_create(&manager->reports, 0, ZBX_DEFAULT_UINT64_HASH_FUNC, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_hashset_create(&manager->batches, 0, ZBX_DEFAULT_UINT64_HASH_FUNC, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_binary_heap_create(&manager->report_queue, rm_report_compare_nextcheck, ZBX_BINARY_HEAP_OPTION_DIRECT); zbx_vector_uint64_create(&manager->flush_queue); zbx_list_create(&manager->job_queue); manager->next_writer_index = 0; manager->session_key = NULL; manager->zabbix_url = NULL; manager->last_batchid = 0; for (i = 0; i < CONFIG_REPORTWRITER_FORKS; i++) { writer = (zbx_rm_writer_t *)zbx_malloc(NULL, sizeof(zbx_rm_writer_t)); writer->client = NULL; zbx_vector_ptr_append(&manager->writers, writer); } out: zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); return ret; } /****************************************************************************** * * * Purpose: destroys report manager * * * * Parameters: manager - [IN] the manager to destroy * * * ******************************************************************************/ static void rm_destroy(zbx_rm_t *manager) { zbx_hashset_iter_t iter; zbx_rm_session_t *session; zbx_rm_report_t *report; zbx_rm_job_t *job; zbx_rm_batch_t *batch; while (SUCCEED == zbx_list_pop(&manager->job_queue, (void **)&job)) rm_job_free(job); zbx_hashset_iter_reset(&manager->sessions, &iter); while (NULL != (session = (zbx_rm_session_t *)zbx_hashset_iter_next(&iter))) zbx_free(session->sid); zbx_hashset_destroy(&manager->sessions); zbx_hashset_iter_reset(&manager->reports, &iter); while (NULL != (report = (zbx_rm_report_t *)zbx_hashset_iter_next(&iter))) rm_report_clean(report); zbx_hashset_destroy(&manager->reports); zbx_hashset_iter_reset(&manager->batches, &iter); while (NULL != (batch = (zbx_rm_batch_t *)zbx_hashset_iter_next(&iter))) rm_batch_clean(batch); zbx_hashset_destroy(&manager->batches); zbx_vector_uint64_destroy(&manager->flush_queue); zbx_queue_ptr_destroy(&manager->free_writers); zbx_vector_ptr_clear_ext(&manager->writers, (zbx_mem_free_func_t)rm_writer_free); zbx_vector_ptr_destroy(&manager->writers); } /****************************************************************************** * * * Purpose: registers report writer * * * * Parameters: manager - [IN] * * client - [IN] the connected writer * * message - [IN] the received message * * * ******************************************************************************/ static void rm_register_writer(zbx_rm_t *manager, zbx_ipc_client_t *client, zbx_ipc_message_t *message) { zbx_rm_writer_t *writer = NULL; pid_t ppid; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); memcpy(&ppid, message->data, sizeof(ppid)); if (ppid != getppid()) { zbx_ipc_client_close(client); zabbix_log(LOG_LEVEL_DEBUG, "refusing connection from foreign process"); } else { if (manager->next_writer_index == manager->writers.values_num) { THIS_SHOULD_NEVER_HAPPEN; exit(EXIT_FAILURE); } writer = (zbx_rm_writer_t *)manager->writers.values[manager->next_writer_index++]; writer->client = client; zbx_queue_ptr_push(&manager->free_writers, writer); } zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: convert timestamp to range format used in URL query fields * * * * Parameters: tm - [IN] the timestamp * * * * Return value: formatted time to be used in URL query fields * * * ******************************************************************************/ static char *rm_time_to_urlfield(const struct tm *tm) { static char buf[26]; zbx_snprintf(buf, sizeof(buf), "%02d-%02d-%02d%%20%02d%%3A%02d%%3A%02d", tm->tm_year + 1900, tm->tm_mon + 1, tm->tm_mday, tm->tm_hour, tm->tm_min, tm->tm_sec); return buf; } /****************************************************************************** * * * Purpose: create zbx_session cookie for frontend authentication * * * * Parameters: manager - [IN] * * sessionid - [IN] * * * * Return value: zbx_session cookie * * * ******************************************************************************/ static char *report_create_cookie(zbx_rm_t *manager, const char *sessionid) { struct zbx_json j; char *cookie = NULL, *out_str_raw = NULL; size_t i; char out_str[ZBX_SHA256_DIGEST_SIZE * 2 + 1]; uint8_t out[ZBX_SHA256_DIGEST_SIZE]; zbx_json_init(&j, 512); zbx_json_addstring(&j, ZBX_PROTO_TAG_SESSIONID, sessionid, ZBX_JSON_TYPE_STRING); hmac_sha256(manager->session_key, strlen(manager->session_key), j.buffer, j.buffer_size, &out, sizeof(out)); memset(&out_str, 0, sizeof(out_str)); for (i = 0; i < sizeof(out); i++) zbx_snprintf(&out_str[i*2], 3, "%02x", out[i]); out_str_raw = zbx_dsprintf(NULL, "\"%s\"", out_str); zbx_json_addraw(&j, ZBX_PROTO_TAG_SIGN, out_str_raw); str_base64_encode_dyn(j.buffer, &cookie, j.buffer_size); zbx_json_clean(&j); zbx_free(out_str_raw); return cookie; } /****************************************************************************** * * * Purpose: get specified user session, creating one if necessary * * * * Parameters: manager - [IN] * * userid - [IN] * * * * Return value: session * * * * Comments: When returning new session it's cached and also stored in * * database. * * When returning cached session database is checked if the session * * is not removed. In that case a new session is created. * * * ******************************************************************************/ static zbx_rm_session_t *rm_get_session(zbx_rm_t *manager, zbx_uint64_t userid) { zbx_rm_session_t *session; int now; now = (int)time(NULL); if (NULL != (session = (zbx_rm_session_t *)zbx_hashset_search(&manager->sessions, &userid))) { DB_RESULT result; result = DBselect("select NULL from sessions where sessionid='%s'", session->sid); if (NULL == DBfetch(result)) { zbx_hashset_remove_direct(&manager->sessions, session); session = NULL; } DBfree_result(result); } if (NULL == session) { zbx_rm_session_t session_local; zbx_db_insert_t db_insert; session_local.userid = userid; session = (zbx_rm_session_t *)zbx_hashset_insert(&manager->sessions, &session_local, sizeof(session_local)); session->sid = zbx_create_token(0); session->cookie = report_create_cookie(manager, session->sid); session->db_lastaccess = now; zbx_db_insert_prepare(&db_insert, "sessions", "sessionid", "userid", "lastaccess", "status", NULL); zbx_db_insert_add_values(&db_insert, session->sid, userid, now, ZBX_SESSION_ACTIVE); zbx_db_insert_execute(&db_insert); zbx_db_insert_clean(&db_insert); } session->lastaccess = now; return session; } /****************************************************************************** * * * Purpose: flushes session lastaccess changes to database * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_db_flush_sessions(zbx_rm_t *manager) { zbx_hashset_iter_t iter; zbx_rm_session_t *session; char *sql = NULL; size_t sql_alloc = 0, sql_offset = 0; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); DBbegin(); DBbegin_multiple_update(&sql, &sql_alloc, &sql_offset); zbx_hashset_iter_reset(&manager->sessions, &iter); while (NULL != (session = (zbx_rm_session_t *)zbx_hashset_iter_next(&iter))) { if (session->lastaccess == session->db_lastaccess) continue; zbx_snprintf_alloc(&sql, &sql_alloc, &sql_offset, "update sessions set lastaccess=%d" " where sessionid='%s';\n", session->lastaccess, session->sid); DBexecute_overflowed_sql(&sql, &sql_alloc, &sql_offset); session->db_lastaccess = session->lastaccess; } DBend_multiple_update(&sql, &sql_alloc, &sql_offset); if (16 < sql_offset) DBexecute("%s", sql); DBcommit(); zbx_free(sql); zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: flushes report state, lastaccess and error fields * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_db_flush_reports(zbx_rm_t *manager) { int i; char *sql = NULL; size_t sql_alloc = 0, sql_offset = 0; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); if (0 == manager->flush_queue.values_num) goto out; zbx_vector_uint64_sort(&manager->flush_queue, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_vector_uint64_uniq(&manager->flush_queue, ZBX_DEFAULT_UINT64_COMPARE_FUNC); DBbegin(); DBbegin_multiple_update(&sql, &sql_alloc, &sql_offset); for (i = 0; i < manager->flush_queue.values_num; i++) { zbx_rm_report_t *report; char delim = ' '; if (NULL == (report = (zbx_rm_report_t *)zbx_hashset_search(&manager->reports, &manager->flush_queue.values[i]))) { /* report was removed */ continue; } zbx_strcpy_alloc(&sql, &sql_alloc, &sql_offset, "update report set"); if (0 != (report->flags & ZBX_REPORT_UPDATE_LASTSENT)) { zbx_snprintf_alloc(&sql, &sql_alloc, &sql_offset, "%clastsent=%d", delim, (int)time(NULL)); delim = ','; } if (0 != (report->flags & ZBX_REPORT_UPDATE_STATE)) { zbx_snprintf_alloc(&sql, &sql_alloc, &sql_offset, "%cstate=%d", delim, report->state); delim = ','; } if (0 != (report->flags & ZBX_REPORT_UPDATE_ERROR)) { char *esc, *empty = ""; if (NULL != report->error) esc = DBdyn_escape_string_len(report->error, REPORT_ERROR_LEN); else esc = empty; zbx_snprintf_alloc(&sql, &sql_alloc, &sql_offset, "%cinfo='%s'", delim, esc); if (esc != empty) zbx_free(esc); } zbx_snprintf_alloc(&sql, &sql_alloc, &sql_offset, " where reportid=" ZBX_FS_UI64 ";\n", report->reportid); DBexecute_overflowed_sql(&sql, &sql_alloc, &sql_offset); report->flags = 0; } DBend_multiple_update(&sql, &sql_alloc, &sql_offset); if (16 < sql_offset) /* in ORACLE always present begin..end; */ DBexecute("%s", sql); DBcommit(); /* recreate flush queue to release memory */ zbx_vector_uint64_destroy(&manager->flush_queue); zbx_vector_uint64_create(&manager->flush_queue); zbx_free(sql); out: zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: calculate report range from report time and period * * * * Parameters: report_time - [IN] the report writing time * * period - [IN] the dashboard period * * from - [OUT] the report start time * * to - [OUT] the report end time * * * * Return value: SUCCEED - the report range was calculated successfully * * FAIL - otherwise * * * ******************************************************************************/ static int rm_get_report_range(int report_time, unsigned char period, struct tm *from, struct tm *to) { struct tm *tm; time_t from_time = report_time; zbx_time_unit_t period2unit[] = {ZBX_TIME_UNIT_DAY, ZBX_TIME_UNIT_WEEK, ZBX_TIME_UNIT_MONTH, ZBX_TIME_UNIT_YEAR}; if (ARRSIZE(period2unit) <= period || NULL == (tm = localtime(&from_time))) return FAIL; *to = *tm; zbx_tm_round_down(to, period2unit[period]); *from = *to; zbx_tm_sub(from, 1, period2unit[period]); return SUCCEED; } /****************************************************************************** * * * Purpose: make report attachment name based on report name and timestamp * * * * Parameters: name - [IN] the report name * * report_time - [IN] * * * * Return value: The report attachment name * * * ******************************************************************************/ static char *rm_get_report_name(const char *name, int report_time) { time_t rtime = report_time; struct tm *tm; char *name_esc, *ptr, *name_full; name_esc = zbx_strdup(NULL, name); for (ptr = name_esc; '\0' != *name; ptr++, name++) { switch (*name) { case ' ': case '\t': case ':': case '/': case '\\': *ptr = '_'; break; default: *ptr = *name; break; } } if (NULL == (tm = localtime(&rtime))) name_full = zbx_dsprintf(NULL, "%s.pdf", name_esc); else name_full = zbx_dsprintf(NULL, "%s_%04d-%02d-%02d_%02d-%02d.pdf", name_esc, tm->tm_year + 1900, tm->tm_mon + 1, tm->tm_mday, tm->tm_hour, tm->tm_min); zbx_free(name_esc); return name_full; } /****************************************************************************** * * * Purpose: create new job to be processed by report writers * * * * Parameters: manager - [IN] * * report_name - [IN] * * dashboardid - [IN] the dashboard to view * * access_userid - [IN] the user accessing the dashboard * * report_time - [IN] * * period - [IN] the report period * * userids - [IN] the recipient user identifiers * * userids_num - [IN] the number of recipients * * report_width - [IN] * * report_height - [IN] * * params - [IN] the viewing and processing parameters * * error - [OUT] * * * ******************************************************************************/ static zbx_rm_job_t *rm_create_job(zbx_rm_t *manager, const char *report_name, zbx_uint64_t dashboardid, zbx_uint64_t access_userid, int report_time, unsigned char period, zbx_uint64_t *userids, int userids_num, int report_width, int report_height, const zbx_vector_ptr_pair_t *params, char **error) { zbx_rm_job_t *job; size_t url_alloc = 0, url_offset = 0; zbx_rm_session_t *session; struct tm from, to; int i; if ('\0' == *manager->zabbix_url) { *error = zbx_strdup(NULL, "The Frontend URL has not been configured"); return NULL; } if (SUCCEED != rm_get_report_range(report_time, period, &from, &to)) { *error = zbx_strdup(NULL, "invalid report time or period"); return NULL; } job = (zbx_rm_job_t *)zbx_malloc(NULL, sizeof(zbx_rm_job_t)); memset(job, 0, sizeof(zbx_rm_job_t)); job->report_name = rm_get_report_name(report_name, report_time); zbx_vector_ptr_pair_create(&job->params); for (i = 0; i < params->values_num; i++) { zbx_ptr_pair_t pair; pair.first = zbx_strdup(NULL, (const char *)params->values[i].first); pair.second = zbx_strdup(NULL, (const char *)params->values[i].second); zbx_vector_ptr_pair_append(&job->params, pair); } zbx_vector_uint64_create(&job->userids); zbx_vector_uint64_append_array(&job->userids, userids, userids_num); zbx_snprintf_alloc(&job->url, &url_alloc, &url_offset, "%s/zabbix.php?action=dashboard.print&dashboardid=" ZBX_FS_UI64, manager->zabbix_url, dashboardid); zbx_snprintf_alloc(&job->url, &url_alloc, &url_offset, "&from=%s", rm_time_to_urlfield(&from)); zbx_snprintf_alloc(&job->url, &url_alloc, &url_offset, "&to=%s", rm_time_to_urlfield(&to)); session = rm_get_session(manager, access_userid); job->cookie = zbx_strdup(NULL, session->cookie); job->access_userid = access_userid; job->report_width = report_width; job->report_height = report_height; return job; } /****************************************************************************** * * * Purpose: update report state, lastsent, error in cache * * * * Parameters: manager - [IN] the report manager * * report - [IN] the report to process * * state - [IN] the new report status * * info - [IN] the new report error message * * * ******************************************************************************/ static void rm_update_report(zbx_rm_t *manager, zbx_rm_report_t *report, int state, const char *info) { zbx_uint32_t flags = 0; zabbix_log(LOG_LEVEL_DEBUG, "In %s() reportid:" ZBX_FS_UI64 ", state:%d info:%s", __func__, report->reportid, state, ZBX_NULL2EMPTY_STR(info)); if (ZBX_REPORT_STATE_ERROR != state) flags |= ZBX_REPORT_UPDATE_LASTSENT; if (report->state != state) { report->state = state; flags |= ZBX_REPORT_UPDATE_STATE; } if (NULL == info) { if (NULL != report->error && '\0' != *report->error) { flags |= ZBX_REPORT_UPDATE_ERROR; zbx_free(report->error); } } else if (NULL == report->error || 0 != strcmp(report->error, info)) { report->error = zbx_strdup(report->error, info); flags |= ZBX_REPORT_UPDATE_ERROR; } if (0 != flags) { report->flags |= flags; zbx_vector_uint64_append(&manager->flush_queue, report->reportid); } zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: calculate time when report must be generated * * * * Parameters: report - [IN] * * now - [IN] the current time * * error - [OUT] * * * ******************************************************************************/ static int rm_report_calc_nextcheck(const zbx_rm_report_t *report, int now, char **error) { int nextcheck; zabbix_log(LOG_LEVEL_DEBUG, "In %s() now:%s %s", __func__, zbx_date2str(now, NULL), zbx_time2str(now, NULL)); if (ZBX_REPORT_CYCLE_WEEKLY == report->cycle && 0 == report->weekdays) { *error = zbx_strdup(NULL, "Cannot calculate report start time: weekdays must be set for weekly cycle"); nextcheck = -1; } else { if (-1 == (nextcheck = zbx_get_report_nextcheck(now, report->cycle, report->weekdays, report->start_time, report->timezone))) { *error = zbx_dsprintf(NULL, "Cannot calculate report start time: %s", zbx_strerror(errno)); } } zabbix_log(LOG_LEVEL_DEBUG, "End of %s() nextcheck:%s %s, error:%s", __func__, zbx_date2str(nextcheck, NULL), zbx_time2str(nextcheck, NULL), ZBX_NULL2EMPTY_STR(*error)); return nextcheck; } /****************************************************************************** * * * Purpose: update report parameters * * * * Parameters: report - [IN] * * params - [IN] the report parameters * * * ******************************************************************************/ static void rm_report_update_params(zbx_rm_report_t *report, zbx_vector_ptr_pair_t *params) { zbx_vector_ptr_pair_t old_params; int i, j; zbx_vector_ptr_pair_create(&old_params); zbx_vector_ptr_pair_append_array(&old_params, report->params.values, report->params.values_num); zbx_vector_ptr_pair_clear(&report->params); for (i = 0; i < params->values_num; i++) { zbx_ptr_pair_t pair = {0}; zbx_ptr_pair_t *new_param = ¶ms->values[i]; for (j = 0; j < old_params.values_num; j++) { zbx_ptr_pair_t *old_param = &old_params.values[j]; if (0 == strcmp(new_param->first, old_param->first)) { pair.first = old_param->first; old_param->first = NULL; if (0 == strcmp(new_param->second, old_param->second)) { pair.second = old_param->second; old_param->second = NULL; } else pair.second = zbx_strdup(old_param->second, new_param->second); zbx_vector_ptr_pair_remove_noorder(&old_params, j); break; } } if (NULL == pair.first) { pair.first = zbx_strdup(NULL, new_param->first); pair.second = zbx_strdup(NULL, new_param->second); } zbx_vector_ptr_pair_append(&report->params, pair); } report_destroy_params(&old_params); } /****************************************************************************** * * * Purpose: update report recipient users * * * * Parameters: report - [IN] * * users - [IN] the recipient users * * users_excl - [IN] the excluded user ids * * * ******************************************************************************/ static void rm_report_update_users(zbx_rm_report_t *report, const zbx_vector_recipient_t *users, const zbx_vector_uint64_t *users_excl) { zbx_vector_recipient_clear(&report->users); zbx_vector_recipient_append_array(&report->users, users->values, users->values_num); zbx_vector_uint64_clear(&report->users_excl); zbx_vector_uint64_append_array(&report->users_excl, users_excl->values, users_excl->values_num); zbx_vector_uint64_sort(&report->users_excl, ZBX_DEFAULT_UINT64_COMPARE_FUNC); } /****************************************************************************** * * * Purpose: update report recipient user groups * * * * Parameters: report - [IN] * * usergroups - [IN] the recipient user groups * * * ******************************************************************************/ static void rm_report_update_usergroups(zbx_rm_report_t *report, const zbx_vector_recipient_t *usergroups) { zbx_vector_recipient_clear(&report->usergroups); zbx_vector_recipient_append_array(&report->usergroups, usergroups->values, usergroups->values_num); } /****************************************************************************** * * * Purpose: update general settings cache * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_update_cache_settings(zbx_rm_t *manager) { DB_RESULT result; DB_ROW row; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); result = DBselect("select session_key,url from config"); if (NULL != (row = DBfetch(result))) { manager->session_key = zbx_strdup(manager->session_key, row[0]); manager->zabbix_url = zbx_strdup(manager->zabbix_url, row[1]); } else { manager->session_key = zbx_strdup(manager->session_key, ""); manager->zabbix_url = zbx_strdup(manager->zabbix_url, ""); } DBfree_result(result); zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: check if the report is active based on the specified time * * * * Parameters: report - [IN] * * now - [IN] the current time * * * * Return value: SUCCEED - the report is active * * FAIL - otherwise * * * ******************************************************************************/ static int rm_is_report_active(const zbx_rm_report_t *report, int now) { if (0 != report->active_since && now < report->active_since) return FAIL; if (0 != report->active_till && now >= report->active_till) return FAIL; return SUCCEED; } /****************************************************************************** * * * Purpose: remove report from queue if it was queued * * * * Parameters: manager - [IN] * * report - [IN] * * * ******************************************************************************/ static void rm_dequeue_report(zbx_rm_t *manager, zbx_rm_report_t *report) { if (0 != report->nextcheck) { zbx_binary_heap_remove_direct(&manager->report_queue, report->reportid); report->nextcheck = 0; } } /****************************************************************************** * * * Purpose: update reports cache * * * * Parameters: manager - [IN] * * now - [IN] the current time * * * ******************************************************************************/ static void rm_update_cache_reports(zbx_rm_t *manager, int now) { DB_RESULT result; DB_ROW row; zbx_vector_uint64_t reportids; zbx_hashset_iter_t iter; zbx_rm_report_t *report, report_local; zbx_config_t cfg; const char *tz; char *error = NULL; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); zbx_config_get(&cfg, ZBX_CONFIG_FLAGS_DEFAULT_TIMEZONE); zbx_vector_uint64_create(&reportids); result = DBselect("select r.reportid,r.userid,r.name,r.dashboardid,r.period,r.cycle,r.weekdays,r.start_time," "r.active_since,r.active_till,u.timezone,r.state,r.info,r.lastsent,r.status" " from report r,users u" " where r.userid=u.userid"); while (NULL != (row = DBfetch(result))) { zbx_uint64_t reportid; int nextcheck, start_time, active_since, active_till, reschedule = 0; unsigned char period, cycle, weekdays; ZBX_STR2UINT64(reportid, row[0]); zbx_vector_uint64_append(&reportids, reportid); tz = row[10]; if (0 == strcmp(tz, ZBX_TIMEZONE_DEFAULT_VALUE)) tz = cfg.default_timezone; ZBX_STR2UCHAR(period, row[4]); ZBX_STR2UCHAR(cycle, row[5]); ZBX_STR2UCHAR(weekdays, row[6]); start_time = atoi(row[7]); active_since = atoi(row[8]); active_till = atoi(row[9]); if (NULL == (report = (zbx_rm_report_t *)zbx_hashset_search(&manager->reports, &reportid))) { report_local.reportid = reportid; report = (zbx_rm_report_t *)zbx_hashset_insert(&manager->reports, &report_local, sizeof(report_local)); zbx_vector_ptr_pair_create(&report->params); zbx_vector_recipient_create(&report->usergroups); zbx_vector_recipient_create(&report->users); zbx_vector_uint64_create(&report->users_excl); report->name = zbx_strdup(NULL, row[2]); report->timezone = zbx_strdup(NULL, tz); report->nextcheck = 0; ZBX_STR2UCHAR(report->state, row[11]); report->error = zbx_strdup(NULL, row[12]); report->lastsent = atoi(row[13]); report->flags = 0; reschedule = 1; } else { if (report->period != period || report->cycle != cycle || report->weekdays != weekdays || report->start_time != start_time || report->active_since != active_since || report->active_till != active_till) { reschedule = 1; } if (0 != strcmp(report->name, row[2])) report->name = zbx_strdup(report->name, row[2]); if (0 != strcmp(report->timezone, tz)) { report->timezone = zbx_strdup(report->timezone, tz); reschedule = 1; } } ZBX_STR2UINT64(report->userid, row[1]); ZBX_STR2UINT64(report->dashboardid, row[3]); ZBX_STR2UCHAR(report->period, row[4]); ZBX_STR2UCHAR(report->cycle, row[5]); ZBX_STR2UCHAR(report->weekdays, row[6]); report->start_time = atoi(row[7]); report->active_since = atoi(row[8]); report->active_till = atoi(row[9]); ZBX_STR2UCHAR(report->status, row[14]); if (ZBX_REPORT_STATUS_DISABLED == report->status) { rm_dequeue_report(manager, report); continue; } if (0 == reschedule) continue; if (-1 != (nextcheck = rm_report_calc_nextcheck(report, now, &error))) { if (nextcheck != report->nextcheck) { if (SUCCEED == rm_is_report_active(report, now)) { zbx_binary_heap_elem_t elem = {report->reportid, (void *)report}; int nextcheck_old = report->nextcheck; report->nextcheck = nextcheck; if (0 != nextcheck_old) zbx_binary_heap_update_direct(&manager->report_queue, &elem); else zbx_binary_heap_insert(&manager->report_queue, &elem); } else rm_dequeue_report(manager, report); } } else { rm_update_report(manager, report, ZBX_REPORT_STATE_ERROR, error); rm_dequeue_report(manager, report); zbx_free(error); } } DBfree_result(result); /* remove deleted reports from cache */ zbx_vector_uint64_sort(&reportids, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_hashset_iter_reset(&manager->reports, &iter); while (NULL != (report = (zbx_rm_report_t *)zbx_hashset_iter_next(&iter))) { if (FAIL == zbx_vector_uint64_bsearch(&reportids, report->reportid, ZBX_DEFAULT_UINT64_COMPARE_FUNC)) { rm_dequeue_report(manager, report); rm_report_clean(report); zbx_hashset_iter_remove(&iter); } } zbx_vector_uint64_destroy(&reportids); zbx_config_clean(&cfg); zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: update cached report parameters * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_update_cache_reports_params(zbx_rm_t *manager) { DB_RESULT result; DB_ROW row; zbx_rm_report_t *report = NULL; zbx_vector_ptr_pair_t params; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); zbx_vector_ptr_pair_create(¶ms); result = DBselect("select rp.reportid,rp.name,rp.value" " from report_param rp,report r" " where rp.reportid=r.reportid" " and r.status=%d" " order by r.reportid", ZBX_REPORT_STATUS_ENABLED); while (NULL != (row = DBfetch(result))) { zbx_uint64_t reportid; zbx_ptr_pair_t pair; ZBX_STR2UINT64(reportid, row[0]); if (NULL != report) { if (reportid != report->reportid) { rm_report_update_params(report, ¶ms); report_clear_params(¶ms); } report = NULL; } if (NULL == report) { if (NULL == (report = (zbx_rm_report_t *)zbx_hashset_search(&manager->reports, &reportid))) continue; } pair.first = zbx_strdup(NULL, row[1]); pair.second = zbx_strdup(NULL, row[2]); zbx_vector_ptr_pair_append(¶ms, pair); } DBfree_result(result); if (0 != params.values_num) rm_report_update_params(report, ¶ms); report_destroy_params(¶ms); zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: update cached report recipient users * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_update_cache_reports_users(zbx_rm_t *manager) { DB_RESULT result; DB_ROW row; zbx_rm_report_t *report = NULL; zbx_vector_recipient_t users; zbx_vector_uint64_t users_excl; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); zbx_vector_recipient_create(&users); zbx_vector_uint64_create(&users_excl); result = DBselect("select ru.reportid,ru.userid,ru.exclude,ru.access_userid" " from report_user ru,report r" " where ru.reportid=r.reportid" " and r.status=%d" " order by r.reportid", ZBX_REPORT_STATUS_ENABLED); while (NULL != (row = DBfetch(result))) { zbx_uint64_t reportid, userid; ZBX_STR2UINT64(reportid, row[0]); if (NULL != report) { if (reportid != report->reportid) { rm_report_update_users(report, &users, &users_excl); zbx_vector_recipient_clear(&users); zbx_vector_uint64_clear(&users_excl); } report = NULL; } if (NULL == report) { if (NULL == (report = (zbx_rm_report_t *)zbx_hashset_search(&manager->reports, &reportid))) continue; } ZBX_STR2UINT64(userid, row[1]); if (ZBX_REPORT_INCLUDE_USER == atoi(row[2])) { zbx_rm_recipient_t user; user.id = userid; ZBX_DBROW2UINT64(user.access_userid, row[3]); zbx_vector_recipient_append(&users, user); } else zbx_vector_uint64_append(&users_excl, userid); } DBfree_result(result); if (0 != users.values_num || 0 != users_excl.values_num) rm_report_update_users(report, &users, &users_excl); zbx_vector_uint64_destroy(&users_excl); zbx_vector_recipient_destroy(&users); zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: update cached report recipient user groups * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_update_cache_reports_usergroups(zbx_rm_t *manager) { DB_RESULT result; DB_ROW row; zbx_rm_report_t *report = NULL; zbx_vector_recipient_t usergroups; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); zbx_vector_recipient_create(&usergroups); result = DBselect("select rg.reportid,rg.usrgrpid,rg.access_userid" " from report_usrgrp rg,report r" " where rg.reportid=r.reportid" " and r.status=%d" " order by r.reportid", ZBX_REPORT_STATUS_ENABLED); while (NULL != (row = DBfetch(result))) { zbx_uint64_t reportid; zbx_rm_recipient_t usergroup; ZBX_STR2UINT64(reportid, row[0]); if (NULL != report) { if (reportid != report->reportid) { rm_report_update_usergroups(report, &usergroups); zbx_vector_recipient_clear(&usergroups); } report = NULL; } if (NULL == report) { if (NULL == (report = (zbx_rm_report_t *)zbx_hashset_search(&manager->reports, &reportid))) continue; } ZBX_STR2UINT64(usergroup.id, row[1]); ZBX_DBROW2UINT64(usergroup.access_userid, row[2]); zbx_vector_recipient_append(&usergroups, usergroup); } DBfree_result(result); if (0 != usergroups.values_num) rm_report_update_usergroups(report, &usergroups); zbx_vector_recipient_destroy(&usergroups); zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: dump cached reports into log * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_dump_cache(zbx_rm_t *manager) { zbx_hashset_iter_t iter; zbx_rm_report_t *report; char *str = NULL; size_t str_alloc = 0, str_offset; int i; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); zbx_hashset_iter_reset(&manager->reports, &iter); while (NULL != (report = (zbx_rm_report_t *)zbx_hashset_iter_next(&iter))) { str_offset = 0; zabbix_log(LOG_LEVEL_TRACE, "reportid:" ZBX_FS_UI64 ", name:%s, userid:" ZBX_FS_UI64 ", dashboardid:" ZBX_FS_UI64 ", period:%d, cycle:%d, weekdays:0x%x", report->reportid, report->name, report->userid, report->dashboardid, report->period, report->cycle, report->weekdays); zbx_strcpy_alloc(&str, &str_alloc, &str_offset, "active:"); if (0 != report->active_since) { zbx_snprintf_alloc(&str, &str_alloc, &str_offset, "%s %s", zbx_date2str(report->active_since, NULL), zbx_time2str(report->active_since, NULL)); } zbx_strcpy_alloc(&str, &str_alloc, &str_offset, " - "); if (0 != report->active_till) { zbx_snprintf_alloc(&str, &str_alloc, &str_offset, "%s %s", zbx_date2str(report->active_till, NULL), zbx_time2str(report->active_till, NULL)); } zbx_snprintf_alloc(&str, &str_alloc, &str_offset, ", start_time:%d:%02d:%02d, timezone:%s", report->start_time / SEC_PER_HOUR, report->start_time % SEC_PER_HOUR / SEC_PER_MIN, report->start_time % SEC_PER_MIN, report->timezone); zbx_snprintf_alloc(&str, &str_alloc, &str_offset, ", nextcheck:%s %s", zbx_date2str(report->nextcheck, NULL), zbx_time2str(report->nextcheck, NULL)); zabbix_log(LOG_LEVEL_TRACE, " %s", str); zabbix_log(LOG_LEVEL_TRACE, " params:"); for (i = 0; i < report->params.values_num; i++) { zabbix_log(LOG_LEVEL_TRACE, " %s:%s", (char *)report->params.values[i].first, (char *)report->params.values[i].second); } zabbix_log(LOG_LEVEL_TRACE, " users:"); for (i = 0; i < report->users.values_num; i++) { zbx_rm_recipient_t *user = (zbx_rm_recipient_t *)&report->users.values[i]; zabbix_log(LOG_LEVEL_TRACE, " userid:" ZBX_FS_UI64 ", acess_userid:" ZBX_FS_UI64, user->id, user->access_userid); } zabbix_log(LOG_LEVEL_TRACE, " usergroups:"); for (i = 0; i < report->usergroups.values_num; i++) { zbx_rm_recipient_t *usergroup = (zbx_rm_recipient_t *)&report->usergroups.values[i]; zabbix_log(LOG_LEVEL_TRACE, " usrgrpid:" ZBX_FS_UI64 ", acess_userid:" ZBX_FS_UI64, usergroup->id, usergroup->access_userid); } zabbix_log(LOG_LEVEL_TRACE, " exclude:"); for (i = 0; i < report->users_excl.values_num; i++) { zabbix_log(LOG_LEVEL_TRACE, " userid:" ZBX_FS_UI64, report->users_excl.values[i]); } } zbx_free(str); zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: update configuration and report cache * * * * Parameters: manager - [IN] * * * ******************************************************************************/ static void rm_update_cache(zbx_rm_t *manager) { int now; now = (int)time(NULL); rm_update_cache_settings(manager); rm_update_cache_reports(manager, now); rm_update_cache_reports_params(manager); rm_update_cache_reports_users(manager); rm_update_cache_reports_usergroups(manager); if (SUCCEED == ZBX_CHECK_LOG_LEVEL(LOG_LEVEL_TRACE)) rm_dump_cache(manager); } typedef struct { zbx_uint64_t mediatypeid; char *recipient; } zbx_report_dst_t; static void zbx_report_dst_free(zbx_report_dst_t *dst) { zbx_free(dst->recipient); zbx_free(dst); } #define ZBX_REPORT_DEFAULT_WIDTH 1920 #define ZBX_REPORT_DEFAULT_HEIGHT 1080 #define ZBX_REPORT_ROW_HEIGHT 70 #define ZBX_REPORT_BOTTOM_MARGIN 12 /****************************************************************************** * * * Purpose: calculate report dimensions based on dashboard contents * * * * Parameters: dashboardid - [IN] * * width - [OUT] the report width in pixels * * height - [OUT] the report height in pixels * * * ******************************************************************************/ static void rm_get_report_dimensions(zbx_uint64_t dashboardid, int *width, int *height) { DB_RESULT result; DB_ROW row; int y_max = 0; zabbix_log(LOG_LEVEL_DEBUG, "In %s() dashboardid:" ZBX_FS_UI64, __func__, dashboardid); result = DBselect("select w.y,w.height" " from widget w,dashboard_page p" " where w.dashboard_pageid=p.dashboard_pageid" " and p.dashboardid=" ZBX_FS_UI64 " and p.sortorder=0", dashboardid); while (NULL != (row = DBfetch(result))) { int bottom; bottom = atoi(row[0]) + atoi(row[1]); if (bottom > y_max) y_max = bottom; } DBfree_result(result); if (0 != y_max) *height = y_max * ZBX_REPORT_ROW_HEIGHT + ZBX_REPORT_BOTTOM_MARGIN; else *height = ZBX_REPORT_DEFAULT_HEIGHT; *width = ZBX_REPORT_DEFAULT_WIDTH; zabbix_log(LOG_LEVEL_DEBUG, "End of %s() width:%d height:%d", __func__, *width, *height); } /****************************************************************************** * * * Purpose: process job by sending it to writer * * * * Parameters: writer - [IN] * * job - [IN] the view to process * * error - [OUT] the error message * * * ******************************************************************************/ static int rm_writer_process_job(zbx_rm_writer_t *writer, zbx_rm_job_t *job, char **error) { unsigned char *data = NULL; zbx_uint32_t size; int ret = FAIL, rc; char *sql = NULL; size_t sql_alloc = 0, sql_offset = 0; zbx_vector_uint64_t mediatypeids; zbx_vector_ptr_t dsts; DB_RESULT result; DB_ROW row; zbx_report_dst_t *dst; zabbix_log(LOG_LEVEL_DEBUG, "In %s() url:%s", __func__, job->url); zbx_vector_uint64_create(&mediatypeids); zbx_vector_ptr_create(&dsts); zbx_strcpy_alloc(&sql, &sql_alloc, &sql_offset, "select m.sendto,mt.mediatypeid" " from media m,media_type mt" " where"); DBadd_condition_alloc(&sql, &sql_alloc, &sql_offset, "m.userid", job->userids.values, job->userids.values_num); zbx_snprintf_alloc(&sql, &sql_alloc, &sql_offset, " and m.active=%d" " and m.mediatypeid=mt.mediatypeid" " and mt.type=%d" " and mt.status=%d", MEDIA_STATUS_ACTIVE, MEDIA_TYPE_EMAIL, MEDIA_TYPE_STATUS_ACTIVE); result = DBselect("%s", sql); while (NULL != (row = DBfetch(result))) { dst = (zbx_report_dst_t *)zbx_malloc(NULL, sizeof(zbx_report_dst_t)); ZBX_STR2UINT64(dst->mediatypeid, row[1]); dst->recipient = zbx_strdup(NULL, row[0]); zbx_vector_ptr_append(&dsts, dst); zbx_vector_uint64_append(&mediatypeids, dst->mediatypeid); } DBfree_result(result); if (0 == dsts.values_num) { *error = zbx_dsprintf(NULL, "No media configured for the report recipients"); goto out; } size = report_serialize_begin_report(&data, job->report_name, job->url, job->cookie, job->report_width, job->report_height, &job->params); if (SUCCEED != zbx_ipc_client_send(writer->client, ZBX_IPC_REPORTER_BEGIN_REPORT, data, size)) { THIS_SHOULD_NEVER_HAPPEN; *error = zbx_dsprintf(NULL, "Cannot send message to report writer"); goto out; } zbx_free(data); ret = SUCCEED; if (0 != dsts.values_num) { zbx_vector_str_t recipients; int index = 0; zbx_vector_str_create(&recipients); zbx_vector_ptr_sort(&dsts, ZBX_DEFAULT_UINT64_PTR_COMPARE_FUNC); zbx_vector_uint64_sort(&mediatypeids, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_vector_uint64_uniq(&mediatypeids, ZBX_DEFAULT_UINT64_COMPARE_FUNC); sql_offset = 0; zbx_strcpy_alloc(&sql, &sql_alloc, &sql_offset, "select mediatypeid,type,smtp_server,smtp_helo,smtp_email,exec_path,gsm_modem,username," "passwd,smtp_port,smtp_security,smtp_verify_peer,smtp_verify_host," "smtp_authentication,exec_params,maxsessions,maxattempts,attempt_interval," "content_type,script,timeout" " from media_type" " where"); DBadd_condition_alloc(&sql, &sql_alloc, &sql_offset, "mediatypeid", mediatypeids.values, mediatypeids.values_num); result = DBselect("%s", sql); while (NULL != (row = DBfetch(result)) && SUCCEED == ret) { DB_MEDIATYPE mt; ZBX_STR2UINT64(mt.mediatypeid, row[0]); mt.type = atoi(row[1]); mt.smtp_server = zbx_strdup(NULL, row[2]); mt.smtp_helo = zbx_strdup(NULL, row[3]); mt.smtp_email = zbx_strdup(NULL, row[4]); mt.exec_path = zbx_strdup(NULL, row[5]); mt.gsm_modem = zbx_strdup(NULL, row[6]); mt.username = zbx_strdup(NULL, row[7]); mt.passwd = zbx_strdup(NULL, row[8]); mt.smtp_port = (unsigned short)atoi(row[9]); ZBX_STR2UCHAR(mt.smtp_security, row[10]); ZBX_STR2UCHAR(mt.smtp_verify_peer, row[11]); ZBX_STR2UCHAR(mt.smtp_verify_host, row[12]); ZBX_STR2UCHAR(mt.smtp_authentication, row[13]); mt.exec_params = zbx_strdup(NULL, row[14]); mt.maxsessions = atoi(row[15]); mt.maxattempts = atoi(row[16]); mt.attempt_interval = zbx_strdup(NULL, row[17]); ZBX_STR2UCHAR(mt.content_type, row[18]); mt.script = zbx_strdup(NULL, row[19]); mt.timeout = zbx_strdup(NULL, row[20]); for (; index < dsts.values_num; index++) { dst = (zbx_report_dst_t *)dsts.values[index]; if (dst->mediatypeid != mt.mediatypeid) break; zbx_vector_str_append(&recipients, dst->recipient); } if (0 != recipients.values_num) { size = report_serialize_send_report(&data, &mt, &recipients); ret = zbx_ipc_client_send(writer->client, ZBX_IPC_REPORTER_SEND_REPORT, data, size); zbx_free(data); } else THIS_SHOULD_NEVER_HAPPEN; zbx_vector_str_clear(&recipients); zbx_db_mediatype_clean(&mt); } DBfree_result(result); zbx_vector_str_destroy(&recipients); } /* attempt to send finish request even if last sending failed */ rc = zbx_ipc_client_send(writer->client, ZBX_IPC_REPORTER_END_REPORT, NULL, 0); if (SUCCEED == ret) ret = rc; out: zbx_free(sql); zbx_free(data); zbx_vector_ptr_clear_ext(&dsts, (zbx_ptr_free_func_t)zbx_report_dst_free); zbx_vector_ptr_destroy(&dsts); zbx_vector_uint64_destroy(&mediatypeids); zabbix_log(LOG_LEVEL_DEBUG, "End of %s():%s", __func__, zbx_result_string(ret)); return ret; } /****************************************************************************** * * * Purpose: create jobs to process the report * * * * Parameters: manager - [IN] * * report - [IN] the report to process * * userid - [IN] the recipient user id * * access_userid - [IN] the user id used to create the report * * now - [IN] the current time * * params - [IN] the report parameters * * width - [IN] the report width * * height - [IN] the report height * * jobs - [IN/OUT] the created jobs * * error - [OUT] the error message * * * * Return value: SUCCEED - the user was added to existing job or a new was * * successfully created. * * FAIL - failed to create a new job for the user. * * * ******************************************************************************/ static int rm_jobs_add_user(zbx_rm_t *manager, zbx_rm_report_t *report, zbx_uint64_t userid, zbx_uint64_t access_userid, int now, const zbx_vector_ptr_pair_t *params, int width, int height, zbx_vector_ptr_t *jobs, char **error) { int i; zbx_rm_job_t *job; if (FAIL != zbx_vector_uint64_search(&report->users_excl, userid, ZBX_DEFAULT_UINT64_COMPARE_FUNC)) return SUCCEED; for (i = 0; i < jobs->values_num; i++) { job = (zbx_rm_job_t *)jobs->values[i]; if (job->access_userid == access_userid) break; } if (i == jobs->values_num) { if (NULL == (job = rm_create_job(manager, report->name, report->dashboardid, access_userid, now, report->period, &userid, 1, width, height, params, error))) { return FAIL; } zbx_vector_ptr_append(jobs, job); } zbx_vector_uint64_append(&job->userids, userid); return SUCCEED; } /****************************************************************************** * * * Purpose: create user group based jobs * * * * Parameters: manager - [IN] * * report - [IN] the report to process * * now - [IN] the current time * * params - [IN] the report parameters * * width - [IN] the report width * * height - [IN] the report height * * jobs - [IN/OUT] the created jobs * * error - [OUT] the error message * * * * Return value: SUCCEED - jobs were created successfully * * FAIL - otherwise * * * ******************************************************************************/ static int rm_report_create_usergroup_jobs(zbx_rm_t *manager, zbx_rm_report_t *report, int now, const zbx_vector_ptr_pair_t *params, int width, int height, zbx_vector_ptr_t *jobs, char **error) { DB_ROW row; DB_RESULT result; zbx_vector_uint64_t ids; int i, ret = FAIL; char *sql = NULL; size_t sql_alloc = 0, sql_offset = 0; zbx_uint64_t userid, usrgrpid, access_userid; zbx_vector_uint64_create(&ids); for (i = 0; i < report->usergroups.values_num; i++) zbx_vector_uint64_append(&ids, report->usergroups.values[i].id); zbx_vector_uint64_sort(&ids, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_vector_uint64_uniq(&ids, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_strcpy_alloc(&sql, &sql_alloc, &sql_offset, "select userid,usrgrpid from users_groups where"); DBadd_condition_alloc(&sql, &sql_alloc, &sql_offset, "usrgrpid", ids.values, ids.values_num); result = DBselect("%s", sql); while (NULL != (row = DBfetch(result))) { access_userid = 0; ZBX_STR2UINT64(userid, row[0]); ZBX_STR2UINT64(usrgrpid, row[1]); for (i = 0; i < report->usergroups.values_num; i++) { if (report->usergroups.values[i].id == usrgrpid) { access_userid = report->usergroups.values[i].access_userid; break; } } if (0 == access_userid) access_userid = userid; if (SUCCEED != rm_jobs_add_user(manager, report, userid, access_userid, now, params, width, height, jobs, error)) { goto out; } } ret = SUCCEED; out: DBfree_result(result); zbx_free(sql); zbx_vector_uint64_destroy(&ids); return ret; } /****************************************************************************** * * * Purpose: create jobs to process the report * * * * Parameters: manager - [IN] * * report - [IN] the report to process * * now - [IN] the current time * * error - [OUT] the error message * * * * Return value: SUCCEED - jobs were created successfully * * FAIL - otherwise * * * ******************************************************************************/ static int rm_report_create_jobs(zbx_rm_t *manager, zbx_rm_report_t *report, int now, char **error) { zbx_vector_ptr_t jobs; int i, ret = FAIL, jobs_num, width, height; zbx_uint64_t access_userid; zbx_rm_batch_t *batch, batch_local; zbx_vector_ptr_pair_t params; zabbix_log(LOG_LEVEL_DEBUG, "In %s() reportid:" ZBX_FS_UI64 , __func__, report->reportid); rm_get_report_dimensions(report->dashboardid, &width, &height); zbx_vector_ptr_create(&jobs); zbx_vector_ptr_pair_create(¶ms); for (i = 0; i < report->params.values_num; i++) { zbx_ptr_pair_t pair; pair.first = zbx_strdup(NULL, report->params.values[i].first); pair.second = zbx_strdup(NULL, report->params.values[i].second); if (0 == strcmp(pair.first, ZBX_REPORT_PARAM_BODY) || 0 == strcmp(pair.first, ZBX_REPORT_PARAM_SUBJECT)) { substitute_simple_macros(NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, (char **)&pair.second, MACRO_TYPE_REPORT, NULL, 0); } zbx_vector_ptr_pair_append(¶ms, pair); } DBbegin(); for (i = 0; i < report->users.values_num; i++) { if (0 == (access_userid = report->users.values[i].access_userid)) access_userid = report->users.values[i].id; if (SUCCEED != rm_jobs_add_user(manager, report, report->users.values[i].id, access_userid, now, ¶ms, width, height, &jobs, error)) { goto out; } } if (0 != report->usergroups.values_num) { if (SUCCEED != rm_report_create_usergroup_jobs(manager, report, now, ¶ms, width, height, &jobs, error)) { goto out; } } /* create job batch for result tracking */ batch_local.batchid = ++manager->last_batchid; batch = (zbx_rm_batch_t *)zbx_hashset_insert(&manager->batches, &batch_local, sizeof(batch_local)); batch->reportid = report->reportid; batch->error_num = 0; batch->sent_num = 0; batch->total_num = 0; batch->info = NULL; batch->info_alloc = 0; batch->info_offset = 0; zbx_vector_ptr_create(&batch->jobs); zbx_vector_ptr_append_array(&batch->jobs, jobs.values, jobs.values_num); /* queue jobs */ for (i = 0; i < jobs.values_num; i++) { zbx_rm_job_t *job = (zbx_rm_job_t *)jobs.values[i]; zbx_vector_uint64_sort(&job->userids, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_vector_uint64_uniq(&job->userids, ZBX_DEFAULT_UINT64_COMPARE_FUNC); zbx_list_append(&manager->job_queue, job, NULL); job->batchid = batch->batchid; } ret = SUCCEED; out: if (SUCCEED == ret) { DBcommit(); jobs_num = jobs.values_num; } else { DBrollback(); jobs_num = 0; } report_destroy_params(¶ms); zbx_vector_ptr_destroy(&jobs); zabbix_log(LOG_LEVEL_DEBUG, "End of %s():%s jobs:%d %s", __func__, zbx_result_string(ret), jobs_num, ZBX_NULL2EMPTY_STR(*error)); return ret; } /****************************************************************************** * * * Purpose: process queue * * * * Parameters: manager - [IN] * * now - [IN] the current time * * * * Return value: The number of scheduled jobs. * * * ******************************************************************************/ static int rm_schedule_jobs(zbx_rm_t *manager, int now) { zbx_rm_report_t *report; zbx_binary_heap_elem_t *elem; int nextcheck, ret, jobs_num = 0; char *error = NULL; zabbix_log(LOG_LEVEL_DEBUG, "In %s() queue:%d", __func__, manager->report_queue.elems_num); while (SUCCEED != zbx_binary_heap_empty(&manager->report_queue)) { elem = zbx_binary_heap_find_min(&manager->report_queue); report = (zbx_rm_report_t *)elem->data; if (now < report->nextcheck) break; zbx_binary_heap_remove_min(&manager->report_queue); report->nextcheck = 0; if (SUCCEED == (ret = rm_report_create_jobs(manager, report, now, &error))) { if (-1 != (nextcheck = rm_report_calc_nextcheck(report, now, &error))) { if (SUCCEED == rm_is_report_active(report, now)) { zbx_binary_heap_elem_t elem_new = {report->reportid, report}; report->nextcheck = nextcheck; zbx_binary_heap_insert(&manager->report_queue, &elem_new); jobs_num++; } } else ret = FAIL; } if (FAIL == ret) { rm_update_report(manager, report, ZBX_REPORT_STATE_ERROR, error); zabbix_log(LOG_LEVEL_DEBUG, "Cannot process report: %s", error); zbx_free(error); } } zabbix_log(LOG_LEVEL_DEBUG, "End of %s() jobs:%d", __func__, jobs_num); return jobs_num; } /****************************************************************************** * * * Purpose: finish job * * * * Parameters: manager - [IN] * * job - [IN] * * status - [IN] the job status * * error - * * sent_num - * * total_num - * * * ******************************************************************************/ static void rm_finish_job(zbx_rm_t *manager, zbx_rm_job_t *job, int status, const char *error, int sent_num, int total_num) { zbx_rm_batch_t *batch; int i; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); if (NULL == (batch = (zbx_rm_batch_t *)zbx_hashset_search(&manager->batches, &job->batchid))) { THIS_SHOULD_NEVER_HAPPEN; return; } if (SUCCEED != status) { size_t offset = batch->info_offset; batch->error_num++; zbx_strcpy_alloc(&batch->info, &batch->info_alloc, &batch->info_offset, error); batch->info[offset] = toupper(batch->info[offset]); zbx_strcpy_alloc(&batch->info, &batch->info_alloc, &batch->info_offset, ".\n"); } else { batch->sent_num += sent_num; batch->total_num += total_num; } for (i = 0; i < batch->jobs.values_num; i++) { if (batch->jobs.values[i] == job) { rm_job_free(job); zbx_vector_ptr_remove_noorder(&batch->jobs, i); break; } } if (0 == batch->jobs.values_num) { zbx_rm_report_t *report; zabbix_log(LOG_LEVEL_DEBUG, "%s() batch finished with %d failed jobs", __func__, batch->error_num); if (NULL != (report = (zbx_rm_report_t *)zbx_hashset_search(&manager->reports, &batch->reportid))) { char *info = NULL; size_t info_alloc = 0, info_offset = 0; status = ZBX_REPORT_STATE_SUCCESS; if (batch->sent_num != batch->total_num) { zbx_snprintf_alloc(&info, &info_alloc, &info_offset, "Failed to sent %d report(s) from %d.\n", batch->total_num - batch->sent_num, batch->total_num); status = ZBX_REPORT_STATE_SUCCESS_INFO; } if (0 != batch->error_num) { zbx_snprintf_alloc(&info, &info_alloc, &info_offset, "Failed to create %d report(s):\n%s", batch->error_num, batch->info); status = ZBX_REPORT_STATE_ERROR; } rm_update_report(manager, report, status, info); zbx_free(info); } rm_batch_clean(batch); zbx_hashset_remove_direct(&manager->batches, batch); } zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__); } /****************************************************************************** * * * Purpose: send error result in response to test request * * * * Parameters: client - [IN] the connected trapper * * error - [IN] the error message * * * ******************************************************************************/ static void rm_send_test_error_result(zbx_ipc_client_t *client, const char *error) { unsigned char *data; zbx_uint32_t size; size = report_serialize_response(&data, FAIL, error, NULL); zbx_ipc_client_send(client, ZBX_IPC_REPORTER_TEST_RESULT, data, size); zbx_free(data); } /****************************************************************************** * * * Purpose: process queue * * * * Parameters: manager - [IN] * * * * Return value: The number of started jobs. * * * ******************************************************************************/ static int rm_process_jobs(zbx_rm_t *manager) { zbx_rm_writer_t *writer; zbx_rm_job_t *job; char *error = NULL; int jobs_num = 0; zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__); while (SUCCEED != zbx_queue_ptr_empty(&manager->free_writers)) { if (SUCCEED != zbx_list_pop(&manager->job_queue, (void **)&job)) break; writer = zbx_queue_ptr_pop(&manager->free_writers); if (SUCCEED != rm_writer_process_job(writer, job, &error)) { if (NULL != job->client) { rm_send_test_error_result(job->client, error); rm_job_free(job); } else rm_finish_job(manager, job, FAIL, error, 0, 0); zbx_queue_ptr_push(&manager->free_writers, writer); zbx_free(error); } else writer->job = job; jobs_num++; } zabbix_log(LOG_LEVEL_DEBUG, "End of %s() jobs:%d", __func__, jobs_num); return jobs_num; } /****************************************************************************** * * * Purpose: test report * * * * Parameters: manager - [IN] * * client - [IN] the connected writer * * message - [IN] the received message * * error - [IN] the error message * * * * Return value: SUCCEED - the test report job was created successfully * * FAIL - otherwise * * * ******************************************************************************/ static int rm_test_report(zbx_rm_t *manager, zbx_ipc_client_t *client, zbx_ipc_message_t *message, char **error) { zbx_uint64_t dashboardid, userid, access_userid; zbx_vector_ptr_pair_t params; int report_time, ret, width, height; unsigned char period; zbx_rm_job_t *job; char *name; zbx_vector_ptr_pair_create(¶ms); report_deserialize_test_report(message->data, &name, &dashboardid, &userid, &access_userid, &report_time, &period, ¶ms); rm_get_report_dimensions(dashboardid, &width, &height); if (NULL != (job = rm_create_job(manager, name, dashboardid, access_userid, report_time, period, &userid, 1, width, height, ¶ms, error))) { zbx_ipc_client_addref(client); job->client = client; zbx_list_append(&manager->job_queue, job, NULL); ret = SUCCEED; } else ret = FAIL; zbx_free(name); report_destroy_params(¶ms); return ret; } /****************************************************************************** * * * Purpose: process report result message * * * * Parameters: manager - [IN] * * client - [IN] the connected writer * * message - [IN] the received message * * * ******************************************************************************/ static void rm_process_result(zbx_rm_t *manager, zbx_ipc_client_t *client, zbx_ipc_message_t *message) { zbx_rm_writer_t *writer; if (NULL == (writer = rm_get_writer(manager, client))) { THIS_SHOULD_NEVER_HAPPEN; return; } if (NULL != writer->job->client) { /* external test request - forward the response to the requester */ if (SUCCEED == zbx_ipc_client_connected(writer->job->client)) { zbx_ipc_client_send(writer->job->client, ZBX_IPC_REPORTER_TEST_RESULT, message->data, message->size); } rm_job_free(writer->job); } else { zbx_vector_ptr_t results; int status, i, total_num = 0, sent_num = 0; zbx_alerter_dispatch_result_t *result; char *error; zbx_vector_ptr_create(&results); report_deserialize_response(message->data, &status, &error, &results); for (i = 0; i < results.values_num; i++) { result = (zbx_alerter_dispatch_result_t *)results.values[i]; if (SUCCEED == result->status) { sent_num++; } else { zabbix_log(LOG_LEVEL_DEBUG, "failed to send report to \"%s\": %s", result->recipient, result->info); } total_num++; } rm_finish_job(manager, writer->job, status, error, sent_num, total_num); zbx_free(error); zbx_vector_ptr_clear_ext(&results, (zbx_clean_func_t)zbx_alerter_dispatch_result_free); zbx_vector_ptr_destroy(&results); } writer->job = NULL; zbx_queue_ptr_push(&manager->free_writers, writer); } ZBX_THREAD_ENTRY(report_manager_thread, args) { #define ZBX_STAT_INTERVAL 5 /* if a process is busy and does not sleep then update status not faster than */ /* once in STAT_INTERVAL seconds */ #define ZBX_SYNC_INTERVAL 60 /* report configuration refresh interval */ #define ZBX_FLUSH_INTERVAL 10 char *error = NULL; zbx_ipc_client_t *client; zbx_ipc_message_t *message; double time_stat, time_idle = 0, time_now, sec, time_sync, time_flush_sessions, time_flush; int ret, processed_num = 0, created_num = 0; zbx_rm_t manager; zbx_timespec_t timeout; process_type = ((zbx_thread_args_t *)args)->process_type; server_num = ((zbx_thread_args_t *)args)->server_num; process_num = ((zbx_thread_args_t *)args)->process_num; zbx_setproctitle("%s #%d starting", get_process_type_string(process_type), process_num); zabbix_log(LOG_LEVEL_INFORMATION, "%s #%d started [%s #%d]", get_program_type_string(program_type), server_num, get_process_type_string(process_type), process_num); update_selfmon_counter(ZBX_PROCESS_STATE_BUSY); if (FAIL == rm_init(&manager, &error)) { zabbix_log(LOG_LEVEL_CRIT, "cannot initialize alert manager: %s", error); zbx_free(error); exit(EXIT_FAILURE); } DBconnect(ZBX_DB_CONNECT_NORMAL); /* initialize statistics */ time_stat = zbx_time(); time_sync = 0; time_flush_sessions = time_stat; time_flush = time_stat; zbx_setproctitle("%s #%d started", get_process_type_string(process_type), process_num); while (ZBX_IS_RUNNING()) { time_now = zbx_time(); if (ZBX_STAT_INTERVAL < time_now - time_stat) { zbx_setproctitle("%s #%d [jobs created %d, processed %d, idle " ZBX_FS_DBL " sec during " ZBX_FS_DBL " sec]", get_process_type_string(process_type), process_num, created_num, processed_num, time_idle, time_now - time_stat); time_stat = time_now; time_idle = 0; created_num = 0; processed_num = 0; } if (SEC_PER_HOUR < time_now - time_flush_sessions) { rm_db_flush_sessions(&manager); time_flush_sessions = time_now; } if (ZBX_FLUSH_INTERVAL < time_now - time_flush) { rm_db_flush_reports(&manager); time_flush = time_now; } if (time_now - time_sync >= ZBX_SYNC_INTERVAL) { rm_update_cache(&manager); time_sync = time_now; } created_num += rm_schedule_jobs(&manager, (int)time(NULL)); processed_num += rm_process_jobs(&manager); sec = zbx_time(); if (sec - time_now >= 1) { timeout.sec = 0; timeout.ns = 0; } else { double delay = 1 - (sec - time_now); timeout.sec = (int)delay; timeout.ns = (int)(delay * 1000000000) % 1000000000; } time_now = sec; update_selfmon_counter(ZBX_PROCESS_STATE_IDLE); ret = zbx_ipc_service_recv(&manager.ipc, &timeout, &client, &message); update_selfmon_counter(ZBX_PROCESS_STATE_BUSY); sec = zbx_time(); zbx_update_env(sec); if (ZBX_IPC_RECV_IMMEDIATE != ret) time_idle += sec - time_now; if (NULL != message) { switch (message->code) { case ZBX_IPC_REPORTER_REGISTER: rm_register_writer(&manager, client, message); break; case ZBX_IPC_REPORTER_TEST: if (FAIL == rm_test_report(&manager, client, message, &error)) { rm_send_test_error_result(client, error); zbx_free(error); } break; case ZBX_IPC_REPORTER_RESULT: rm_process_result(&manager, client, message); break; } zbx_ipc_message_free(message); } if (NULL != client) zbx_ipc_client_release(client); } zbx_setproctitle("%s #%d [terminated]", get_process_type_string(process_type), process_num); while (1) zbx_sleep(SEC_PER_MIN); zbx_ipc_service_close(&manager.ipc); rm_destroy(&manager); }