/*
** Zabbix
** Copyright (C) 2001-2014 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 "common.h"

#include "cfg.h"
#include "pid.h"
#include "db.h"
#include "dbcache.h"
#include "log.h"
#include "zbxgetopt.h"
#include "mutexs.h"

#include "sysinfo.h"
#include "zbxserver.h"

#include "daemon.h"
#include "zbxself.h"

#include "alerter/alerter.h"
#include "dbsyncer/dbsyncer.h"
#include "dbconfig/dbconfig.h"
#include "discoverer/discoverer.h"
#include "httppoller/httppoller.h"
#include "housekeeper/housekeeper.h"
#include "pinger/pinger.h"
#include "poller/poller.h"
#include "poller/checks_ipmi.h"
#include "timer/timer.h"
#include "trapper/trapper.h"
#include "snmptrapper/snmptrapper.h"
#include "nodewatcher/nodewatcher.h"
#include "watchdog/watchdog.h"
#include "utils/nodechange.h"
#include "escalator/escalator.h"
#include "proxypoller/proxypoller.h"
#include "selfmon/selfmon.h"
#include <unistd.h>
#include <librdkafka/rdkafka.h>
#include <zookeeper/zookeeper.h>
#include <zookeeper/zookeeper.jute.h>
#include <jansson.h>

#define INIT_SERVER(type, count)								\
	process_type = type;									\
	process_num = server_num - server_count + count;					\
	zabbix_log(LOG_LEVEL_INFORMATION, "server #%d started [%s #%d]",			\
			server_num, get_process_type_string(process_type), process_num)

const char	*progname = NULL;
const char	title_message[] = "Zabbix server";
const char	syslog_app_name[] = "zabbix_server";
const char	usage_message[] = "[-hV] [-c <file>] [-n <nodeid>] [-R <option>]";

const char	*help_message[] = {
	"Options:",
	"  -c --config <file>              Absolute path to the configuration file",
	"  -n --new-nodeid <nodeid>        Convert database data to new nodeid",
	"  -R --runtime-control <option>   Perform administrative functions",
	"",
	"Runtime control options:",
	"  " ZBX_CONFIG_CACHE_RELOAD "             Reload configuration cache",
	"",
	"Other options:",
	"  -h --help                       Give this help",
	"  -V --version                    Display version number",
	NULL	/* end of text */
};

/* COMMAND LINE OPTIONS */

/* long options */
static struct zbx_option	longopts[] =
{
	{"config",		1,	NULL,	'c'},
	{"new-nodeid",		1,	NULL,	'n'},
	{"runtime-control",	1,	NULL,	'R'},
	{"help",		0,	NULL,	'h'},
	{"version",		0,	NULL,	'V'},
	{NULL}
};

/* short options */
static char	shortopts[] = "c:n:hVR:";

/* end of COMMAND LINE OPTIONS */

int	threads_num = 0;
pid_t	*threads = NULL;

unsigned char	daemon_type		= ZBX_DAEMON_TYPE_SERVER;

int		process_num		= 0;
unsigned char	process_type		= ZBX_PROCESS_TYPE_UNKNOWN;

int	CONFIG_ALERTER_FORKS		= 1;
int	CONFIG_DISCOVERER_FORKS		= 1;
int	CONFIG_HOUSEKEEPER_FORKS	= 1;
int	CONFIG_NODEWATCHER_FORKS	= 1;
int	CONFIG_PINGER_FORKS		= 1;
int	CONFIG_POLLER_FORKS		= 5;
int	CONFIG_UNREACHABLE_POLLER_FORKS	= 1;
int	CONFIG_HTTPPOLLER_FORKS		= 1;
int	CONFIG_IPMIPOLLER_FORKS		= 0;
int	CONFIG_TIMER_FORKS		= 1;
int	CONFIG_TRAPPER_FORKS		= 5;
int	CONFIG_SNMPTRAPPER_FORKS	= 0;
int	CONFIG_JAVAPOLLER_FORKS		= 0;
int	CONFIG_ESCALATOR_FORKS		= 1;
int	CONFIG_SELFMON_FORKS		= 1;
int	CONFIG_WATCHDOG_FORKS		= 1;
int	CONFIG_DATASENDER_FORKS		= 0;
int	CONFIG_HEARTBEAT_FORKS		= 0;

int	CONFIG_LISTEN_PORT		= ZBX_DEFAULT_SERVER_PORT;
char	*CONFIG_LISTEN_IP		= NULL;
char	*CONFIG_SOURCE_IP		= NULL;
int	CONFIG_TRAPPER_TIMEOUT		= 300;

int	CONFIG_HOUSEKEEPING_FREQUENCY	= 1;
int	CONFIG_MAX_HOUSEKEEPER_DELETE	= 500;		/* applies for every separate field value */
int	CONFIG_SENDER_FREQUENCY		= 30;
int	CONFIG_HISTSYNCER_FORKS		= 4;
int	CONFIG_HISTSYNCER_FREQUENCY	= 5;
int	CONFIG_CONFSYNCER_FORKS		= 1;
int	CONFIG_CONFSYNCER_FREQUENCY	= 60;
int	CONFIG_CONF_CACHE_SIZE		= 8 * ZBX_MEBIBYTE;
int	CONFIG_HISTORY_CACHE_SIZE	= 8 * ZBX_MEBIBYTE;
int	CONFIG_TRENDS_CACHE_SIZE	= 4 * ZBX_MEBIBYTE;
int	CONFIG_TEXT_CACHE_SIZE		= 16 * ZBX_MEBIBYTE;
int	CONFIG_DISABLE_HOUSEKEEPING	= 0;
int	CONFIG_UNREACHABLE_PERIOD	= 45;
int	CONFIG_UNREACHABLE_DELAY	= 15;
int	CONFIG_UNAVAILABLE_DELAY	= 60;
int	CONFIG_LOG_LEVEL		= LOG_LEVEL_WARNING;
char	*CONFIG_ALERT_SCRIPTS_PATH	= NULL;
char	*CONFIG_EXTERNALSCRIPTS		= NULL;
char	*CONFIG_TMPDIR			= NULL;
char	*CONFIG_FPING_LOCATION		= NULL;
#ifdef HAVE_IPV6
char	*CONFIG_FPING6_LOCATION		= NULL;
#endif
char	*CONFIG_DBHOST			= NULL;
char	*CONFIG_DBNAME			= NULL;
char	*CONFIG_DBSCHEMA		= NULL;
char	*CONFIG_DBUSER			= NULL;
char	*CONFIG_DBPASSWORD		= NULL;
char	*CONFIG_DBSOCKET		= NULL;
int	CONFIG_DBPORT			= 0;
int	CONFIG_ENABLE_REMOTE_COMMANDS	= 0;
int	CONFIG_LOG_REMOTE_COMMANDS	= 0;
int	CONFIG_UNSAFE_USER_PARAMETERS	= 0;

int	CONFIG_NODEID			= 0;
int	CONFIG_MASTER_NODEID		= 0;
int	CONFIG_NODE_NOEVENTS		= 0;
int	CONFIG_NODE_NOHISTORY		= 0;

char	*CONFIG_SNMPTRAP_FILE		= NULL;

char	*CONFIG_JAVA_GATEWAY		= NULL;
int	CONFIG_JAVA_GATEWAY_PORT	= ZBX_DEFAULT_GATEWAY_PORT;

char	*CONFIG_SSH_KEY_LOCATION	= NULL;

int	CONFIG_LOG_SLOW_QUERIES		= 0;	/* ms; 0 - disable */

int	CONFIG_SERVER_STARTUP_TIME	= 0;	/* zabbix server startup time */

int	CONFIG_PROXYPOLLER_FORKS	= 1;	/* parameters for passive proxies */

/* how often zabbix server sends configuration data to proxy, in seconds */
int	CONFIG_PROXYCONFIG_FREQUENCY	= 3600;	/* 1h */
int	CONFIG_PROXYDATA_FREQUENCY	= 1;	/* 1s */

// where zabbix logs specific content to
char*	CONFIG_EMAILLOG_FILE			= NULL;
char*	CONFIG_ITEM_VALUE_LOG_FILE		= NULL;
char*	CONFIG_ALERT_LOG_FILE			= NULL;
char*   CONFIG_UPDATE_ITEM_LOG_FILE     	= NULL;
char*	CONFIG_UPDATE_HOST_LOG_FILE		= NULL;
char*   CONFIG_UPDATE_TRIGGER_LOG_FILE  	= NULL;
char*   CONFIG_UPDATE_HOST_STATUS_LOG_FILE	= NULL;
char*	CONFIG_HOST_NOT_MONITOR_LOG_FILE	= NULL;
char*   CONFIG_ITEM_NOT_MONITOR_LOG_FILE        = NULL;
char*	CONFIG_UPDATE_FUNCTION_LOG_FILE		= NULL;
char*   CONFIG_SMTP_USER                = NULL;
char*   CONFIG_SMTP_PASSWORD            = NULL;

// location to store host status via proxy
char*   CONFIG_HOST_STATUS_PATH         = NULL;

int	CONFIG_ZABBIX_SERVER_ID		= 0;
int CONFIG_SMTP_PORT            = 587;

// base host id
int	CONFIG_BASE_HOST_ID		= 10000;

#define MAX_HOST_COUNT 			100000
#define MAX_FILE_NAME_LENGTH		100
#define MAX_LOG_STRING_LENGTH		1024
#define MAX_ITEM_LOG_STRING_LENGTH	8192
#define MAX_KAFKA_ITEM_LOG_STRING_LENGTH 8192

int proxy_host_status[MAX_HOST_COUNT];

/* mutex for node syncs */
ZBX_MUTEX	node_sync_access;

/* Enable & Disable Write Log */
#define VNG_WRITE_LOG_ENABLE 1
#define VNG_WRITE_LOG_DISABLE 0

/* Config write log enable */
int CONFIG_VNG_WRITE_LOG_ENABLE = 1;

// where kafka log
char* CONFIG_KAFKALOG_FILE	= NULL;

/* instance kafka & brokers & topic*/
#define MAX_BROKER_BUFFER_SIZE	1024
#define MAX_BROKER_PATH_SIZE	255
#define MAX_BROKER_CONFIG_SIZE  1024

/* Connection Status */
#define KAFKA_OK 0
#define KAFKA_FAIL -1

/* Enable & Disable Kafka */
#define KAFKA_ENABLE 1
#define KAFKA_DISABLE 0

/* zookeeper handle & kafka instance */
static rd_kafka_t *rk = NULL;
static zhandle_t *zh = NULL;

/* json topic instance */
static json_t *jConfTopic = NULL;


/* Config kafka enable */
int CONFIG_KAFKA_ENABLE = 0;

/* where connect to zookeeper server & topic */
char*	CONFIG_ZOOKEEPER_SERVER		= NULL;
char*	CONFIG_BROKER_PATH = NULL;
char*	CONFIG_TOPIC_HISTORY_ITEMS	= NULL;
char*	CONFIG_TOPIC_HISTORY_ITEMS_DEFAULT = NULL;

/* config kafka performance */
char*   CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MESSAGES = NULL;
char*	CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MS = NULL;
char*	CONFIG_KAFKA_MESSAGE_SEND_MAX_RETRIES = NULL;
char*	CONFIG_KAFKA_RETRY_BACKOFF_MS = NULL;

void logger (const rd_kafka_t *rk, int level,
		    const char *fac, const char *buf) 
{
	const char* __function_name = "logger";
	struct timeval tv;
	gettimeofday(&tv, NULL);
	VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_ERR, "In %s(), %u.%03u RDKAFKA-%i-%s: %s: %s\n",
                                        __function_name, (int)tv.tv_sec, (int)(tv.tv_usec / 1000),
											level, fac, rd_kafka_name(rk), buf);
}


void msg_delivered (rd_kafka_t *rk,
			   void *payload, size_t len,
			   int error_code,
			   void *opaque, void *msg_opaque) {
	const char* __function_name = "msg_delivered";

	if (error_code)
	{
		VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_ERR, "In %s(), %% Message delivery failed: %s\n",
                                        __function_name, rd_kafka_err2str(error_code));

	}
	else{
		VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_DEBUG, "In %s(), %% Message delivered (%zd bytes)\n",
                                        __function_name, len);
	}
}

void set_brokerlist_from_zookeeper(zhandle_t *zzh, char *brokers)
{
	const char* __function_name = "set_brokerlist_from_zookeeper";
	
	if (zzh)
	{
		struct String_vector brokerlist;
		if (zoo_get_children(zzh, CONFIG_BROKER_PATH, 1, &brokerlist) != ZOK)
		{
			VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_ERR, "In %s(), No brokers found on path %s\n",
                                        __function_name, CONFIG_BROKER_PATH);
			return;
		}

		int i;
		char *brokerptr = brokers;
		for (i = 0; i < brokerlist.count; i++)
		{
			char path[MAX_BROKER_PATH_SIZE], cfg[MAX_BROKER_CONFIG_SIZE];
			zbx_snprintf(path, MAX_BROKER_PATH_SIZE, "/brokers/ids/%s", brokerlist.data[i]);
			int len = sizeof(cfg);
			zoo_get(zzh, path, 0, cfg, &len, NULL);

			if (len > 0)
			{
				cfg[len] = '\0';
				json_error_t jerror;
				json_t *jobj = json_loads(cfg, 0, &jerror);
				if (jobj)
				{
					json_t *jhost = json_object_get(jobj, "host");
					json_t *jport = json_object_get(jobj, "port");

					if (jhost && jport)
					{
						const char *host = json_string_value(jhost);
						const int   port = json_integer_value(jport);
						zbx_snprintf(brokerptr, MAX_BROKER_BUFFER_SIZE, "%s:%d", host, port);

						brokerptr += strlen(brokerptr);
						if (i < brokerlist.count - 1)
						{
							*brokerptr++ = ',';
						}
					}
					json_decref(jobj);
				}
			}
		}
		deallocate_String_vector(&brokerlist);
	}
}

void watcher(zhandle_t *zh, int type, int state, const char *path, void *watcherCtx)
{
	char brokers[1024];
	if (type == ZOO_CHILD_EVENT && strncmp(path, CONFIG_BROKER_PATH, sizeof(CONFIG_BROKER_PATH) - 1) == 0)
	{
		brokers[0] = '\0';
		set_brokerlist_from_zookeeper(zh, brokers);
		if (brokers[0] != '\0' && rk != NULL)
		{
			rd_kafka_brokers_add(rk, brokers);
			rd_kafka_poll(rk, 10);
		}
	}
}

static zhandle_t* initialize_zookeeper(const char * zookeeper, const int debug)
{
	const char* __function_name = "initialize_zookeeper";
	zhandle_t *zh;
	
	if (debug)
	{
		zoo_set_debug_level(ZOO_LOG_LEVEL_DEBUG);
	}
	zh = zookeeper_init(zookeeper, watcher, 10000, 0, 0, 0);
	if (zh == NULL)
	{
		VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_ERR, "In %s(), Zookeeper connection not established.",
                                        __function_name);
	}
	return zh;
}

void set_kafka_config(rd_kafka_conf_t *conf)
{
	rd_kafka_conf_set(conf, "queue.buffering.max.messages", CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MESSAGES, NULL, 0);
	rd_kafka_conf_set(conf, "queue.buffering.max.ms", CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MS, NULL, 0);
	rd_kafka_conf_set(conf, "message.send.max.retries", CONFIG_KAFKA_MESSAGE_SEND_MAX_RETRIES, NULL, 0);
	rd_kafka_conf_set(conf, "retry.backoff.ms", CONFIG_KAFKA_RETRY_BACKOFF_MS, NULL, 0);
}

int VNG_Kafkaconnect()
{
	int iKafkaResult = KAFKA_OK;

	const char* __function_name = "VNG_Kafkaconnect";
	zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __function_name);
	
	rd_kafka_conf_t *conf;

	char *zookeeper = CONFIG_ZOOKEEPER_SERVER;
	char brokers[MAX_BROKER_BUFFER_SIZE];
	int partition = RD_KAFKA_PARTITION_UA;

	char errstr[MAX_LOG_STRING_LENGTH];
	const char *debug = NULL;

	memset(brokers, 0, sizeof(brokers));
	
	/* Kafka configuration */
	conf = rd_kafka_conf_new();
	
	/* Set kafka Config*/
	set_kafka_config(conf);

	/** Initialize zookeeper */
	zh = initialize_zookeeper(zookeeper, debug != NULL);

	/* Add brokers */
	set_brokerlist_from_zookeeper(zh, brokers);

	/*
	 * Producer
	 */
	/* Set up a message delivery report callback.
	 * It will be called once for each message, either on successful
	 * delivery to broker, or upon failure to deliver to broker. */
	rd_kafka_conf_set_dr_cb(conf, msg_delivered);

	/* Create Kafka handle */
	if (!(rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf,
				errstr, sizeof(errstr)))) {
		VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_ERR, "In %s(), %s",
                                        __function_name, errstr);
		iKafkaResult = KAFKA_FAIL;
	}
	else
	{
		/* Set logger */
		rd_kafka_set_logger(rk, logger);
		rd_kafka_set_log_level(rk, LOG_DEBUG);

		/* Add brokers */
		if (rd_kafka_brokers_add(rk, brokers) == 0) {
			VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_ERR, "In %s(), %% No valid brokers specified\n",
                                        __function_name);
			iKafkaResult = KAFKA_FAIL;
		}
	}
	zabbix_log(LOG_LEVEL_DEBUG, "End of %s():%d", __function_name, iKafkaResult);
	return iKafkaResult;
}

void VNG_Kafkaclose()
{
	const char* __function_name = "VNG_Kafkaclose";
	zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __function_name);
	/* Destroy the handle */
	if (NULL != rk)
		rd_kafka_destroy(rk);

	/* Close Connection Zookeeper */
	if (NULL != zh)
		zookeeper_close(zh);
	zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __function_name);
}


void VNG_KafkaProducer(const char* topic, const char* strContent, const char* strErrorMsg)
{
	const char* __function_name = "VNG_KafkaProducer";
	zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __function_name);
	
	rd_kafka_topic_t *rkt;
	rd_kafka_topic_conf_t *topic_conf;
	
	int partition = RD_KAFKA_PARTITION_UA;
	
	/* Topic configuration */
	topic_conf = rd_kafka_topic_conf_new();
	
	/* Create topic */
	rkt = rd_kafka_topic_new(rk, topic, topic_conf);

	if (NULL == rkt)
	{
		VNG_Kafkaclose();
		VNG_Kafkaconnect();
		
		if (NULL == (rkt = rd_kafka_topic_new(rk, topic, topic_conf)))
		{
			VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_WARNING, "Kafka Cluster is down.");
		}
	}

	size_t len = strlen(strContent);
	char buf[len];
	zbx_snprintf(buf, len, "%s", strContent);
	/* Send/Produce message. */
	if (rd_kafka_produce(rkt, partition,
				 RD_KAFKA_MSG_F_COPY,
				 /* Payload and length */
				 buf, len,
				 /* Optional key and its length */
				 NULL, 0,
				 /* Message opaque, provided in
				  * delivery report callback as
				  * msg_opaque. */
				 NULL) == -1) {
		 VNG_KafkaWriteLog(CONFIG_KAFKALOG_FILE, LOG_LEVEL_ERR, "In %s(), %% Failed to produce to topic %s ",
							"partition %i: %s\n", __function_name, rd_kafka_topic_name(rkt), partition,
																	rd_kafka_err2str(
																		rd_kafka_errno2err(errno)));
		/* Poll to handle delivery reports */
		rd_kafka_poll(rk, 0);	
	}
	else{
		zabbix_log(LOG_LEVEL_DEBUG, "In %s(), Send/Produce message:%s",
											__function_name, strContent);
		/* Poll to handle delivery reports */
		rd_kafka_poll(rk, 0);
	}

	/* Wait for messages to be delivered */
	if(rd_kafka_outq_len(rk) > 0)
		rd_kafka_poll(rk, 0);
	
	/* Destroy topic */
	rd_kafka_topic_destroy(rkt);

	zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __function_name);
}

void VNG_LoadConfigTopic()
{
	const char* __function_name = "VNG_LoadConfigTopic";
	zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __function_name);
	json_error_t jerror;
	
	jConfTopic = json_loads(CONFIG_TOPIC_HISTORY_ITEMS, 0, &jerror);
	if (!jConfTopic)
	{
		zabbix_log(LOG_LEVEL_ERR, "In %s(), %s incorrect format %s", __function_name, CONFIG_TOPIC_HISTORY_ITEMS, jerror.text);
	}

	zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __function_name);
}

void VNG_DestroyConfigTopic()
{
	if (jConfTopic)
		json_decref(jConfTopic);
}

void VNG_DetectTopic(const char* strKey, char* strTopic)
{
	const char* __function_name = "VNG_DetectTopic";
	zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __function_name);
	
	if (jConfTopic)
	{
		const char* key;
		const char* strValue;
		
		json_t	*jvalue;
		json_object_foreach(jConfTopic, key, jvalue){
			if (strstr(strKey, key) != NULL)
			{
				strValue = json_string_value(jvalue);
				zbx_snprintf(strTopic, MAX_LOG_STRING_LENGTH, "%s", strValue);
			}
		}
	}
	zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __function_name);
}

void VNG_KafkaItemValue(zbx_uint64_t hostid, const char* strHostName, 
			const char* strIfAddr, unsigned char maintenance_status, 
			zbx_uint64_t itemid, const char* strKey, 
			unsigned char value_type, const char* strValue)
{
	const char*	__function_name = "VNG_KafkaItemValue";
	zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __function_name);

	char strTopic[MAX_LOG_STRING_LENGTH];
	zbx_snprintf(strTopic, MAX_LOG_STRING_LENGTH, "%s", CONFIG_TOPIC_HISTORY_ITEMS_DEFAULT);

	VNG_DetectTopic(strKey, strTopic);
	
	if (0 != strcmp(strTopic, ""))
	{
		//++ Add Define Json
		json_t *root = json_object();
		json_object_set_new( root, "clock", json_integer( (unsigned)time(NULL) ) );
		json_object_set_new( root, "location", json_integer( CONFIG_ZABBIX_SERVER_ID ) );
		json_object_set_new( root, "hostid", json_integer( hostid ) );
		json_object_set_new( root, "host", json_string( strHostName ) );
		json_object_set_new( root, "zb_ip", json_string( strIfAddr ) );
		json_object_set_new( root, "maintenance_status", json_integer( maintenance_status ) );
		json_object_set_new( root, "itemid", json_integer( itemid ) );
		json_object_set_new( root, "key_", json_string( strKey ) );
		json_object_set_new( root, "value_type", json_integer( value_type ) );
		json_object_set_new( root, "value", json_string( strValue ) );
		
		char strJsonLog[MAX_KAFKA_ITEM_LOG_STRING_LENGTH];
		zbx_snprintf(strJsonLog, MAX_KAFKA_ITEM_LOG_STRING_LENGTH, "%s\n", json_dumps(root, 0));
		// Add new line when log buffer cannot affort
		strJsonLog[MAX_KAFKA_ITEM_LOG_STRING_LENGTH - 2] = '\n';
		json_decref( root );

		//-- Add Define Json
		VNG_KafkaProducer(strTopic, strJsonLog, "Could not log update item in group");
	}
	zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __function_name);
}
void VNG_WriteLog(const char* strFileName, const char* strContent, const char* strErrorMsg)
{
	const char*	__function_name = "VNG_WriteLog";
        FILE* pFile = NULL;
        size_t lenLog = (size_t)strlen(strContent);

        pFile = fopen(strFileName, "a+");
        if (NULL != pFile)
        {
                if (!fwrite(strContent, lenLog, (size_t)1, pFile))
                {
                        zabbix_log(LOG_LEVEL_DEBUG, "In %s(), %s",
                                        __function_name, strErrorMsg);
                }

                fclose(pFile);
        }
}

void VNG_ServerLogAlert(zbx_timespec_t timespec, zbx_uint64_t eventid, 
			unsigned char status, zbx_uint64_t triggerid, zbx_uint64_t hostid, 
            zbx_uint64_t itemid, unsigned char value_changed)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

		// Log value to file
		char strLogFile[MAX_FILE_NAME_LENGTH];
		char strLog[MAX_LOG_STRING_LENGTH];

		// Get current date
		time_t now = time(NULL);
		struct tm *t = localtime(&now);

		// Format file name
		zbx_snprintf(strLogFile, MAX_FILE_NAME_LENGTH, "%s%d%02d%02d", CONFIG_ALERT_LOG_FILE, t->tm_year+1900,
    			t->tm_mon+1, t->tm_mday);

		// Format log content
		zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%d %d %d %d %d %d %d %d\n", timespec,
    			CONFIG_ZABBIX_SERVER_ID, eventid, status, triggerid,
						hostid, itemid, value_changed);

		VNG_WriteLog(strLogFile, strLog, "Could not log alert");
}

void VNG_ServerLogUpdateItem(zbx_uint64_t hostid, zbx_uint64_t itemid, 	
				unsigned char status, const char* strKey, 
				unsigned char value_type, const char* strUnits,
				const char* strName)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

		// Log value to file
		char strLog[MAX_LOG_STRING_LENGTH];

		// Format log content
		zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%u %d %d %d %d %s {%s} %d %s\n", 
			(unsigned)time(NULL), CONFIG_ZABBIX_SERVER_ID, hostid, itemid,
			status, strKey, strName, value_type, strUnits);

		VNG_WriteLog(CONFIG_UPDATE_ITEM_LOG_FILE, strLog, "Could not log update item");
}

void VNG_ServerLogItemNotMonitor(zbx_uint64_t itemid, unsigned char status)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

		// Log value to file
		char strLogFile[MAX_FILE_NAME_LENGTH];
		char strLog[MAX_LOG_STRING_LENGTH];

		// Get current date
		time_t now = time(NULL);
		struct tm *t = localtime(&now);

		// Format file name
		zbx_snprintf(strLogFile, MAX_FILE_NAME_LENGTH, "%s%d%02d", CONFIG_ITEM_NOT_MONITOR_LOG_FILE,
						t->tm_year+1900, t->tm_mon+1);

		// Format log content
		zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%d %d %d\n", CONFIG_ZABBIX_SERVER_ID, 
			itemid, status);

		VNG_WriteLog(strLogFile, strLog, "Could not log item not monitor");
}

void VNG_ServerLogHostNotMonitor(zbx_uint64_t hostid)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

		// Log value to file
		char strLogFile[MAX_FILE_NAME_LENGTH];
		char strLog[MAX_LOG_STRING_LENGTH];

		// Get current date
		time_t now = time(NULL);
		struct tm *t = localtime(&now);

		// Format file name
		zbx_snprintf(strLogFile, MAX_FILE_NAME_LENGTH, "%s%d%02d%02d", CONFIG_HOST_NOT_MONITOR_LOG_FILE,
			t->tm_year+1900, t->tm_mon+1, t->tm_mday);

		// Format log content
		zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%d %d\n", CONFIG_ZABBIX_SERVER_ID, hostid);

		VNG_WriteLog(strLogFile, strLog, "Could not log host not monitor");
}

void VNG_ServerLogUpdateHost(zbx_uint64_t hostid, const char* strHost,
                                const char* strName, unsigned char status,
                                unsigned char available, unsigned char maintenance_status,
				int maintenance_from)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

		// Log value to file
		char strLog[MAX_LOG_STRING_LENGTH];

		// Format log content
		zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%d %d {%s} {%s} %d %d %d %d\n", 
			CONFIG_ZABBIX_SERVER_ID, hostid, strHost, strName, status, 
			available, maintenance_status, maintenance_from);

		VNG_WriteLog(CONFIG_UPDATE_HOST_LOG_FILE, strLog, "Could not log update host");
}

void VNG_ServerLogUpdateHostStatus(zbx_uint64_t hostid, unsigned char status)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

		// Log value to file
		char strLogFile[MAX_FILE_NAME_LENGTH];
		char strLog[MAX_LOG_STRING_LENGTH];

		// Get current date
		time_t now = time(NULL);
		struct tm *t = localtime(&now);

		// Format file name
		zbx_snprintf(strLogFile, MAX_FILE_NAME_LENGTH, "%s%d%02d", CONFIG_UPDATE_HOST_STATUS_LOG_FILE,
			t->tm_year+1900, t->tm_mon+1);


		// Format log content
		zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%d %d %d %u\n", CONFIG_ZABBIX_SERVER_ID,
						hostid, status, (unsigned)now);

		VNG_WriteLog(strLogFile, strLog, "Could not log update host status");
}

void VNG_ServerLogUpdateTrigger(zbx_uint64_t triggerid, const char* strExpression,
                                const char* strDescription, unsigned char status,
                                unsigned char value, unsigned char priority,
                                unsigned int clock)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

		// Log value to file
		char strLog[MAX_LOG_STRING_LENGTH];

		// Format log content
		zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%d %d %s {%s} %d %d %d %d\n", 
			CONFIG_ZABBIX_SERVER_ID, triggerid, strExpression, strDescription, 
			status, value, priority, clock);

		VNG_WriteLog(CONFIG_UPDATE_TRIGGER_LOG_FILE, strLog, "Could not log update trigger");
}

void VNG_LogItemValue(zbx_uint64_t hostid, const char* strHostName, 
			const char* strIfAddr, unsigned char maintenance_status, 
			zbx_uint64_t itemid, const char* strKey, 
			unsigned char value_type, const char* strValue, 
			pid_t process_id)
{
		if (CONFIG_KAFKA_ENABLE == KAFKA_ENABLE){
			VNG_KafkaItemValue(hostid, strHostName, strIfAddr, maintenance_status, itemid, strKey, value_type, strValue);
		}

		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;


		const char*      __function_name = "VNG_LogItemValue";

		// Log value to file
		char strLogFile[MAX_FILE_NAME_LENGTH];
		char strLog[MAX_ITEM_LOG_STRING_LENGTH];
		int nMin;
		
		// Replace newline character with '|'
		strValue = string_replace(strValue,"\n","");

		// Get current date
		time_t now = time(NULL);
		struct tm *t = localtime(&now);

		// Calculate minute to format file name
		nMin = t->tm_min - (t->tm_min % 5);

		// Format file name
		zbx_snprintf(strLogFile, MAX_FILE_NAME_LENGTH, "%s%d%02d%02d_%02d%02d_%d", CONFIG_ITEM_VALUE_LOG_FILE,
			t->tm_year+1900, t->tm_mon+1, t->tm_mday, t->tm_hour, nMin, process_id);

		// Format log content
		zbx_snprintf(strLog, MAX_ITEM_LOG_STRING_LENGTH, "%u %d %d {%s} {%s} %d %d %s %d %s\n",
			(unsigned)time(NULL), CONFIG_ZABBIX_SERVER_ID, hostid, strHostName,
						strIfAddr, maintenance_status, itemid, strKey, value_type, strValue);

		// Add new line when log buffer cannot affort
		strLog[MAX_ITEM_LOG_STRING_LENGTH - 2] = '\n';
		        
		VNG_WriteLog(strLogFile, strLog, "Could not log item value");		
}

void VNG_ServerLogUpdateFunction(zbx_uint64_t functionid, zbx_uint64_t triggerid,
				zbx_uint64_t itemid, const char* strFunction,
				const char* strParameter)
{
		if (CONFIG_VNG_WRITE_LOG_ENABLE == VNG_WRITE_LOG_DISABLE)
			return;

        // Log value to file
        char strLogFile[MAX_FILE_NAME_LENGTH];
        char strLog[MAX_LOG_STRING_LENGTH];

        // Get current date
        time_t now = time(NULL);
        struct tm *t = localtime(&now);

        // Format file name
        zbx_snprintf(strLogFile, MAX_FILE_NAME_LENGTH, "%s%d%02d%02d", CONFIG_UPDATE_FUNCTION_LOG_FILE,
                        t->tm_year+1900, t->tm_mon+1, t->tm_mday);


        // Format log content
        zbx_snprintf(strLog, MAX_LOG_STRING_LENGTH, "%d %d %d %d %s {%s}\n", 
			CONFIG_ZABBIX_SERVER_ID, functionid, triggerid,
                        itemid, strFunction, strParameter);

        VNG_WriteLog(strLogFile, strLog, "Could not log update function");
}

/******************************************************************************
 *                                                                            *
 * Function: zbx_set_defaults                                                 *
 *                                                                            *
 * Purpose: set configuration defaults                                        *
 *                                                                            *
 * Author: Vladimir Levijev                                                   *
 *                                                                            *
 ******************************************************************************/
static void	zbx_set_defaults()
{
	CONFIG_SERVER_STARTUP_TIME = time(NULL);

	if (NULL == CONFIG_DBHOST)
		CONFIG_DBHOST = zbx_strdup(CONFIG_DBHOST, "localhost");

	if (NULL == CONFIG_SNMPTRAP_FILE)
		CONFIG_SNMPTRAP_FILE = zbx_strdup(CONFIG_SNMPTRAP_FILE, "/tmp/zabbix_traps.tmp");

	if (NULL == CONFIG_PID_FILE)
		CONFIG_PID_FILE = zbx_strdup(CONFIG_PID_FILE, "/tmp/zabbix_server.pid");

	if (NULL == CONFIG_ALERT_SCRIPTS_PATH)
		CONFIG_ALERT_SCRIPTS_PATH = zbx_strdup(CONFIG_ALERT_SCRIPTS_PATH, DATADIR "/zabbix/alertscripts");

	if (NULL == CONFIG_TMPDIR)
		CONFIG_TMPDIR = zbx_strdup(CONFIG_TMPDIR, "/tmp");

	if (NULL == CONFIG_FPING_LOCATION)
		CONFIG_FPING_LOCATION = zbx_strdup(CONFIG_FPING_LOCATION, "/usr/sbin/fping");

#ifdef HAVE_IPV6
	if (NULL == CONFIG_FPING6_LOCATION)
		CONFIG_FPING6_LOCATION = zbx_strdup(CONFIG_FPING6_LOCATION, "/usr/sbin/fping6");
#endif

	if (NULL == CONFIG_EXTERNALSCRIPTS)
		CONFIG_EXTERNALSCRIPTS = zbx_strdup(CONFIG_EXTERNALSCRIPTS, DATADIR "/zabbix/externalscripts");

	if (0 == CONFIG_NODEID)
		CONFIG_NODEWATCHER_FORKS = 0;

#ifdef HAVE_SQLITE3
	CONFIG_MAX_HOUSEKEEPER_DELETE = 0;
#endif

	if (1 == CONFIG_DISABLE_HOUSEKEEPING)
		CONFIG_HOUSEKEEPER_FORKS = 0;
		
	if (NULL == CONFIG_EMAILLOG_FILE)
	{
		CONFIG_EMAILLOG_FILE = zbx_strdup(CONFIG_EMAILLOG_FILE, "/var/log/zabbix_emaillog");
	}
	if (NULL == CONFIG_ITEM_VALUE_LOG_FILE)
        {
                CONFIG_ITEM_VALUE_LOG_FILE = zbx_strdup(CONFIG_ITEM_VALUE_LOG_FILE, "/var/log/zabbix_item/item_value_log_");
        }
	if (NULL == CONFIG_ALERT_LOG_FILE)
        {
                CONFIG_ALERT_LOG_FILE = zbx_strdup(CONFIG_ALERT_LOG_FILE, "/var/log/zabbix_alert/alert_log_");
        }
	if (NULL == CONFIG_HOST_STATUS_PATH)
        {
                CONFIG_HOST_STATUS_PATH = zbx_strdup(CONFIG_HOST_STATUS_PATH, "/tmp");
        }
	if (NULL == CONFIG_UPDATE_ITEM_LOG_FILE)
        {
                CONFIG_UPDATE_ITEM_LOG_FILE = zbx_strdup(CONFIG_UPDATE_ITEM_LOG_FILE, "/tmp/update_item");
        }
	if (NULL == CONFIG_UPDATE_HOST_LOG_FILE)
        {
                CONFIG_UPDATE_HOST_LOG_FILE = zbx_strdup(CONFIG_UPDATE_HOST_LOG_FILE, "/tmp/update_host");
        }
	if (NULL == CONFIG_UPDATE_TRIGGER_LOG_FILE)
        {
                CONFIG_UPDATE_TRIGGER_LOG_FILE = zbx_strdup(CONFIG_UPDATE_TRIGGER_LOG_FILE, "/tmp/update_trigger");
        }
	if (NULL == CONFIG_UPDATE_HOST_STATUS_LOG_FILE)
        {
                CONFIG_UPDATE_HOST_STATUS_LOG_FILE = zbx_strdup(CONFIG_UPDATE_HOST_STATUS_LOG_FILE, "/tmp/update_host_status_");
        }
	if (NULL == CONFIG_HOST_NOT_MONITOR_LOG_FILE)
        {
                CONFIG_HOST_NOT_MONITOR_LOG_FILE = zbx_strdup(CONFIG_HOST_NOT_MONITOR_LOG_FILE, "/tmp/update_host_not_monitor_");
        }
	if (NULL == CONFIG_ITEM_NOT_MONITOR_LOG_FILE)
        {
                CONFIG_ITEM_NOT_MONITOR_LOG_FILE = zbx_strdup(CONFIG_ITEM_NOT_MONITOR_LOG_FILE, "/tmp/update_item_not_monitor_");
        }
	if (NULL == CONFIG_UPDATE_FUNCTION_LOG_FILE)
        {
                CONFIG_UPDATE_FUNCTION_LOG_FILE = zbx_strdup(CONFIG_UPDATE_FUNCTION_LOG_FILE, "/tmp/update_function_");
        }
    if (NULL == CONFIG_SMTP_USER)
    {
        CONFIG_SMTP_USER = zbx_strdup(CONFIG_SMTP_USER, "");
    }
    if (NULL == CONFIG_SMTP_PASSWORD)
    {
        CONFIG_SMTP_PASSWORD = zbx_strdup(CONFIG_SMTP_PASSWORD, "");
    }
	if (NULL == CONFIG_KAFKALOG_FILE)
		{
			CONFIG_KAFKALOG_FILE = zbx_strdup(CONFIG_KAFKALOG_FILE, "/tmp/zabbix_kafka.log");
		}
	if (NULL == CONFIG_ZOOKEEPER_SERVER)
		{
			CONFIG_ZOOKEEPER_SERVER = zbx_strdup(CONFIG_ZOOKEEPER_SERVER, "localhost:2181");
		}
	if (NULL == CONFIG_TOPIC_HISTORY_ITEMS)
		{
			CONFIG_TOPIC_HISTORY_ITEMS = zbx_strdup(CONFIG_TOPIC_HISTORY_ITEMS, "");
		}
	if (NULL == CONFIG_TOPIC_HISTORY_ITEMS_DEFAULT)
		{
			CONFIG_TOPIC_HISTORY_ITEMS_DEFAULT = zbx_strdup(CONFIG_TOPIC_HISTORY_ITEMS_DEFAULT, "");
		}
	if (NULL == CONFIG_BROKER_PATH)
		{
			CONFIG_BROKER_PATH = zbx_strdup(CONFIG_BROKER_PATH, "/brokers/ids");
		}
	if (NULL == CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MESSAGES)
		{
			CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MESSAGES = zbx_strdup(CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MESSAGES, "500000");
		}
	if (NULL == CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MS)
		{
			CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MS = zbx_strdup(CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MS, "1000");
		}
	if (NULL == CONFIG_KAFKA_MESSAGE_SEND_MAX_RETRIES)
		{
			CONFIG_KAFKA_MESSAGE_SEND_MAX_RETRIES = zbx_strdup(CONFIG_KAFKA_MESSAGE_SEND_MAX_RETRIES, "3");
		}
	if (NULL == CONFIG_KAFKA_RETRY_BACKOFF_MS)
		{
			CONFIG_KAFKA_RETRY_BACKOFF_MS = zbx_strdup(CONFIG_KAFKA_RETRY_BACKOFF_MS, "500");
		}
	// Initialize host status via proxy
	memset(&proxy_host_status, -1, MAX_HOST_COUNT);
}

/******************************************************************************
 *                                                                            *
 * Function: zbx_validate_config                                              *
 *                                                                            *
 * Purpose: validate configuration parameters                                 *
 *                                                                            *
 * Author: Vladimir Levijev                                                   *
 *                                                                            *
 ******************************************************************************/
static void	zbx_validate_config()
{
	if ((NULL == CONFIG_JAVA_GATEWAY || '\0' == *CONFIG_JAVA_GATEWAY) && CONFIG_JAVAPOLLER_FORKS > 0)
	{
		zabbix_log(LOG_LEVEL_CRIT, "JavaGateway not specified in config file or empty");
		exit(1);
	}
}

/******************************************************************************
 *                                                                            *
 * Function: zbx_load_config                                                  *
 *                                                                            *
 * Purpose: parse config file and update configuration parameters             *
 *                                                                            *
 * Parameters:                                                                *
 *                                                                            *
 * Return value:                                                              *
 *                                                                            *
 * Author: Alexei Vladishev                                                   *
 *                                                                            *
 * Comments: will terminate process if parsing fails                          *
 *                                                                            *
 ******************************************************************************/
static void	zbx_load_config()
{
	static struct cfg_line	cfg[] =
	{
		/* PARAMETER,			VAR,					TYPE,
			MANDATORY,	MIN,			MAX */
		{"StartDBSyncers",		&CONFIG_HISTSYNCER_FORKS,		TYPE_INT,
			PARM_OPT,	1,			100},
		{"StartDiscoverers",		&CONFIG_DISCOVERER_FORKS,		TYPE_INT,
			PARM_OPT,	0,			250},
		{"StartHTTPPollers",		&CONFIG_HTTPPOLLER_FORKS,		TYPE_INT,
			PARM_OPT,	0,			1000},
		{"StartPingers",		&CONFIG_PINGER_FORKS,			TYPE_INT,
			PARM_OPT,	0,			1000},
		{"StartPollers",		&CONFIG_POLLER_FORKS,			TYPE_INT,
			PARM_OPT,	0,			1000},
		{"StartPollersUnreachable",	&CONFIG_UNREACHABLE_POLLER_FORKS,	TYPE_INT,
			PARM_OPT,	0,			1000},
		{"StartIPMIPollers",		&CONFIG_IPMIPOLLER_FORKS,		TYPE_INT,
			PARM_OPT,	0,			1000},
		{"StartTrappers",		&CONFIG_TRAPPER_FORKS,			TYPE_INT,
			PARM_OPT,	0,			1000},
		{"StartJavaPollers",		&CONFIG_JAVAPOLLER_FORKS,		TYPE_INT,
			PARM_OPT,	0,			1000},
		{"JavaGateway",			&CONFIG_JAVA_GATEWAY,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"JavaGatewayPort",		&CONFIG_JAVA_GATEWAY_PORT,		TYPE_INT,
			PARM_OPT,	1024,			32767},
		{"SNMPTrapperFile",		&CONFIG_SNMPTRAP_FILE,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"StartSNMPTrapper",		&CONFIG_SNMPTRAPPER_FORKS,		TYPE_INT,
			PARM_OPT,	0,			1},
		{"CacheSize",			&CONFIG_CONF_CACHE_SIZE,		TYPE_INT,
			PARM_OPT,	128 * ZBX_KIBIBYTE,	0x7fffffff},	/* just below 2GB */
		{"HistoryCacheSize",		&CONFIG_HISTORY_CACHE_SIZE,		TYPE_INT,
			PARM_OPT,	128 * ZBX_KIBIBYTE,	0x7fffffff},	/* just below 2GB */
		{"TrendCacheSize",		&CONFIG_TRENDS_CACHE_SIZE,		TYPE_INT,
			PARM_OPT,	128 * ZBX_KIBIBYTE,	0x7fffffff},	/* just below 2GB */
		{"HistoryTextCacheSize",	&CONFIG_TEXT_CACHE_SIZE,		TYPE_INT,
			PARM_OPT,	128 * ZBX_KIBIBYTE,	0x7fffffff},	/* just below 2GB */
		{"CacheUpdateFrequency",	&CONFIG_CONFSYNCER_FREQUENCY,		TYPE_INT,
			PARM_OPT,	1,			SEC_PER_HOUR},
		{"HousekeepingFrequency",	&CONFIG_HOUSEKEEPING_FREQUENCY,		TYPE_INT,
			PARM_OPT,	1,			24},
		{"MaxHousekeeperDelete",	&CONFIG_MAX_HOUSEKEEPER_DELETE,		TYPE_INT,
			PARM_OPT,	0,			1000000},
		{"SenderFrequency",		&CONFIG_SENDER_FREQUENCY,		TYPE_INT,
			PARM_OPT,	5,			SEC_PER_HOUR},
		{"TmpDir",			&CONFIG_TMPDIR,				TYPE_STRING,
			PARM_OPT,	0,			0},
		{"FpingLocation",		&CONFIG_FPING_LOCATION,			TYPE_STRING,
			PARM_OPT,	0,			0},
#ifdef HAVE_IPV6
		{"Fping6Location",		&CONFIG_FPING6_LOCATION,		TYPE_STRING,
			PARM_OPT,	0,			0},
#endif
		{"Timeout",			&CONFIG_TIMEOUT,			TYPE_INT,
			PARM_OPT,	1,			30},
		{"TrapperTimeout",		&CONFIG_TRAPPER_TIMEOUT,		TYPE_INT,
			PARM_OPT,	1,			300},
		{"UnreachablePeriod",		&CONFIG_UNREACHABLE_PERIOD,		TYPE_INT,
			PARM_OPT,	1,			SEC_PER_HOUR},
		{"UnreachableDelay",		&CONFIG_UNREACHABLE_DELAY,		TYPE_INT,
			PARM_OPT,	1,			SEC_PER_HOUR},
		{"UnavailableDelay",		&CONFIG_UNAVAILABLE_DELAY,		TYPE_INT,
			PARM_OPT,	1,			SEC_PER_HOUR},
		{"ListenIP",			&CONFIG_LISTEN_IP,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"ListenPort",			&CONFIG_LISTEN_PORT,			TYPE_INT,
			PARM_OPT,	1024,			32767},
		{"SourceIP",			&CONFIG_SOURCE_IP,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"DisableHousekeeping",		&CONFIG_DISABLE_HOUSEKEEPING,		TYPE_INT,
			PARM_OPT,	0,			1},
		{"DebugLevel",			&CONFIG_LOG_LEVEL,			TYPE_INT,
			PARM_OPT,	0,			4},
		{"PidFile",			&CONFIG_PID_FILE,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"LogFile",			&CONFIG_LOG_FILE,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"LogFileSize",			&CONFIG_LOG_FILE_SIZE,			TYPE_INT,
			PARM_OPT,	0,			1024},
		{"AlertScriptsPath",		&CONFIG_ALERT_SCRIPTS_PATH,		TYPE_STRING,
			PARM_OPT,	0,			0},
		{"ExternalScripts",		&CONFIG_EXTERNALSCRIPTS,		TYPE_STRING,
			PARM_OPT,	0,			0},
		{"DBHost",			&CONFIG_DBHOST,				TYPE_STRING,
			PARM_OPT,	0,			0},
		{"DBName",			&CONFIG_DBNAME,				TYPE_STRING,
			PARM_MAND,	0,			0},
		{"DBSchema",			&CONFIG_DBSCHEMA,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"DBUser",			&CONFIG_DBUSER,				TYPE_STRING,
			PARM_OPT,	0,			0},
		{"DBPassword",			&CONFIG_DBPASSWORD,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"DBSocket",			&CONFIG_DBSOCKET,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"DBPort",			&CONFIG_DBPORT,				TYPE_INT,
			PARM_OPT,	1024,			65535},
		{"NodeID",			&CONFIG_NODEID,				TYPE_INT,
			PARM_OPT,	0,			999},
		{"NodeNoEvents",		&CONFIG_NODE_NOEVENTS,			TYPE_INT,
			PARM_OPT,	0,			1},
		{"NodeNoHistory",		&CONFIG_NODE_NOHISTORY,			TYPE_INT,
			PARM_OPT,	0,			1},
		{"SSHKeyLocation",		&CONFIG_SSH_KEY_LOCATION,		TYPE_STRING,
			PARM_OPT,	0,			0},
		{"LogSlowQueries",		&CONFIG_LOG_SLOW_QUERIES,		TYPE_INT,
			PARM_OPT,	0,			3600000},
		{"StartProxyPollers",		&CONFIG_PROXYPOLLER_FORKS,		TYPE_INT,
			PARM_OPT,	0,			250},
		{"ProxyConfigFrequency",	&CONFIG_PROXYCONFIG_FREQUENCY,		TYPE_INT,
			PARM_OPT,	1,			SEC_PER_WEEK},
		{"ProxyDataFrequency",		&CONFIG_PROXYDATA_FREQUENCY,		TYPE_INT,
			PARM_OPT,	1,			SEC_PER_HOUR},
		{"EmailLogFile",		&CONFIG_EMAILLOG_FILE,			TYPE_STRING,
			PARM_OPT,	0,			0},
		{"ItemValueLogFile",            &CONFIG_ITEM_VALUE_LOG_FILE,            TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"AlertLogFile",                &CONFIG_ALERT_LOG_FILE,                 TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"HostStatusPath",             	&CONFIG_HOST_STATUS_PATH,               TYPE_STRING,
                        PARM_OPT,       0,                      0},
        {"UpdateItemLogFile",           &CONFIG_UPDATE_ITEM_LOG_FILE,           TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"UpdateHostLogFile",           &CONFIG_UPDATE_HOST_LOG_FILE,           TYPE_STRING,
                        PARM_OPT,       0,			0},
		{"UpdateHostStatusLogFile",     &CONFIG_UPDATE_HOST_STATUS_LOG_FILE,    TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"UpdateTriggerLogFile",        &CONFIG_UPDATE_TRIGGER_LOG_FILE,        TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"UpdateHostNotMonitorLogFile", &CONFIG_HOST_NOT_MONITOR_LOG_FILE,      TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"UpdateItemNotMonitorLogFile", &CONFIG_ITEM_NOT_MONITOR_LOG_FILE,      TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"UpdateFunctionLogFile",     	&CONFIG_UPDATE_FUNCTION_LOG_FILE,    	TYPE_STRING,
                        PARM_OPT,       0,                      0},
        {"SMTP_USER",                   &CONFIG_SMTP_USER,                      TYPE_STRING,                        
                        PARM_OPT,       0,                      0},
        {"SMTP_PASSWORD",               &CONFIG_SMTP_PASSWORD,                  TYPE_STRING,
                        PARM_OPT,       0,                      0},
		{"ZabbixServerID",		&CONFIG_ZABBIX_SERVER_ID,		TYPE_INT,
			PARM_OPT,	1,			10},
		{"BaseHostID",              	&CONFIG_BASE_HOST_ID,               	TYPE_INT,
                        PARM_OPT,       0,                      100000},
        {"SMTP_PORT",                   &CONFIG_SMTP_PORT,                      TYPE_INT,
                        PARM_OPT,       0,                      1024},                   
		{"VNGWriteLogEnable",			&CONFIG_VNG_WRITE_LOG_ENABLE,			TYPE_INT,
						PARM_OPT,		0,			1},
		{"KafkaEnable",					&CONFIG_KAFKA_ENABLE,					TYPE_INT,
						PARM_OPT,		0,			1},
		{"KafkaLogFile",				&CONFIG_KAFKALOG_FILE,					TYPE_STRING,
						PARM_OPT,		0,						0},
		{"ZookeeperServer",             &CONFIG_ZOOKEEPER_SERVER,               TYPE_STRING,                        
                        PARM_OPT,       0,                      0},
		{"TOPIC_HISTORY_ITEMS",         &CONFIG_TOPIC_HISTORY_ITEMS,			TYPE_STRING,                        
                        PARM_OPT,       0,                      0},
		{"TOPIC_HISTORY_ITEMS_DEFAULT", &CONFIG_TOPIC_HISTORY_ITEMS_DEFAULT,	TYPE_STRING,                        
                        PARM_OPT,       0,                      0},
		{"BrokerPath",					&CONFIG_BROKER_PATH,					TYPE_STRING,
						PARM_OPT,		0,						0},
		{"KafkaQueueBufferingMaxMessages", &CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MESSAGES,	TYPE_STRING,
						PARM_OPT,		0,						0},
		{"KafkaQueueBufferingMaxms",	&CONFIG_KAFKA_QUEUE_BUFFERING_MAX_MS,			TYPE_STRING,
						PARM_OPT,		0,						0},
		{"KafkaMessageSendMaxRetries",	&CONFIG_KAFKA_MESSAGE_SEND_MAX_RETRIES,			TYPE_STRING,
						PARM_OPT,		0,						0},
		{"KafkaRetryBackoffms",			&CONFIG_KAFKA_RETRY_BACKOFF_MS,					TYPE_STRING,
						PARM_OPT,		0,						0},
		{NULL}
	};

	parse_cfg_file(CONFIG_FILE, cfg, ZBX_CFG_FILE_REQUIRED, ZBX_CFG_STRICT);

	zbx_set_defaults();

	zbx_validate_config();
}

#ifdef HAVE_SIGQUEUE
void	zbx_sigusr_handler(zbx_task_t task)
{
	switch (task)
	{
		case ZBX_TASK_CONFIG_CACHE_RELOAD:
			if (ZBX_PROCESS_TYPE_CONFSYNCER == process_type)
			{
				zabbix_log(LOG_LEVEL_WARNING, "forced reloading of the configuration cache");
				zbx_wakeup();
			}
			break;
		default:
			break;
	}
}
#endif

/******************************************************************************
 *                                                                            *
 * Function: main                                                             *
 *                                                                            *
 * Purpose: executes server processes                                         *
 *                                                                            *
 * Author: Eugene Grigorjev                                                   *
 *                                                                            *
 ******************************************************************************/
int	main(int argc, char **argv)
{
	zbx_task_t	task = ZBX_TASK_START;
	char		ch = '\0';
	int		nodeid = 0;

	progname = get_program_name(argv[0]);

	/* parse the command-line */
	while ((char)EOF != (ch = (char)zbx_getopt_long(argc, argv, shortopts, longopts, NULL)))
	{
		switch (ch)
		{
			case 'c':
				CONFIG_FILE = zbx_strdup(CONFIG_FILE, zbx_optarg);
				break;
			case 'R':
				if (0 == strcmp(zbx_optarg, ZBX_CONFIG_CACHE_RELOAD))
					task = ZBX_TASK_CONFIG_CACHE_RELOAD;
				else
				{
					printf("invalid runtime control option: %s\n", zbx_optarg);
					exit(EXIT_FAILURE);
				}
				break;
			case 'h':
				help();
				exit(-1);
				break;
			case 'n':
				nodeid = (NULL == zbx_optarg ? 0 : atoi(zbx_optarg));
				task = ZBX_TASK_CHANGE_NODEID;
				break;
			case 'V':
				version();
				exit(-1);
				break;
			default:
				usage();
				exit(-1);
				break;
		}
	}

	if (NULL == CONFIG_FILE)
		CONFIG_FILE = zbx_strdup(CONFIG_FILE, SYSCONFDIR "/zabbix_server.conf");

	/* required for simple checks */
	init_metrics();

	zbx_load_config();

	if (ZBX_TASK_CONFIG_CACHE_RELOAD == task)
		exit(SUCCEED == zbx_sigusr_send(ZBX_TASK_CONFIG_CACHE_RELOAD) ? EXIT_SUCCESS : EXIT_FAILURE);

#ifdef HAVE_OPENIPMI
	init_ipmi_handler();
#endif

	switch (task)
	{
		case ZBX_TASK_CHANGE_NODEID:
			exit(SUCCEED == change_nodeid(0, nodeid) ? EXIT_SUCCESS : EXIT_FAILURE);
			break;
		default:
			break;
	}

	return daemon_start(CONFIG_ALLOW_ROOT);
}

int	MAIN_ZABBIX_ENTRY()
{
	DB_RESULT	result;
	DB_ROW		row;
	pid_t		pid;
	zbx_sock_t	listen_sock;
	int		i, server_num = 0, server_count = 0;

	if (NULL == CONFIG_LOG_FILE || '\0' == *CONFIG_LOG_FILE)
		zabbix_open_log(LOG_TYPE_SYSLOG, CONFIG_LOG_LEVEL, NULL);
	else
		zabbix_open_log(LOG_TYPE_FILE, CONFIG_LOG_LEVEL, CONFIG_LOG_FILE);

#ifdef	HAVE_SNMP
#	define SNMP_FEATURE_STATUS	"YES"
#else
#	define SNMP_FEATURE_STATUS	" NO"
#endif
#ifdef	HAVE_OPENIPMI
#	define IPMI_FEATURE_STATUS	"YES"
#else
#	define IPMI_FEATURE_STATUS	" NO"
#endif
#ifdef	HAVE_LIBCURL
#	define LIBCURL_FEATURE_STATUS	"YES"
#else
#	define LIBCURL_FEATURE_STATUS	" NO"
#endif
#ifdef	HAVE_JABBER
#	define JABBER_FEATURE_STATUS	"YES"
#else
#	define JABBER_FEATURE_STATUS	" NO"
#endif
#ifdef	HAVE_ODBC
#	define ODBC_FEATURE_STATUS	"YES"
#else
#	define ODBC_FEATURE_STATUS	" NO"
#endif
#ifdef	HAVE_SSH2
#	define SSH2_FEATURE_STATUS	"YES"
#else
#	define SSH2_FEATURE_STATUS	" NO"
#endif
#ifdef	HAVE_IPV6
#	define IPV6_FEATURE_STATUS	"YES"
#else
#	define IPV6_FEATURE_STATUS	" NO"
#endif

	zabbix_log(LOG_LEVEL_INFORMATION, "Starting Zabbix Server. Zabbix %s (revision %s).",
			ZABBIX_VERSION, ZABBIX_REVISION);

	zabbix_log(LOG_LEVEL_INFORMATION, "****** Enabled features ******");
	zabbix_log(LOG_LEVEL_INFORMATION, "SNMP monitoring:           " SNMP_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "IPMI monitoring:           " IPMI_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "WEB monitoring:            " LIBCURL_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "Jabber notifications:      " JABBER_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "Ez Texting notifications:  " LIBCURL_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "ODBC:                      " ODBC_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "SSH2 support:              " SSH2_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "IPv6 support:              " IPV6_FEATURE_STATUS);
	zabbix_log(LOG_LEVEL_INFORMATION, "******************************");

	if (0 != CONFIG_NODEID)
	{
		zabbix_log(LOG_LEVEL_INFORMATION, "NodeID:                    %3d", CONFIG_NODEID);
		zabbix_log(LOG_LEVEL_INFORMATION, "******************************");
	}

#ifdef	HAVE_SQLITE3
	zbx_create_sqlite3_mutex(CONFIG_DBNAME);
#endif

	DBconnect(ZBX_DB_CONNECT_NORMAL);

	if (0 != CONFIG_NODEID)
	{
		result = DBselect("select masterid from nodes where nodeid=%d", CONFIG_NODEID);

		if (NULL != (row = DBfetch(result)) && SUCCEED != DBis_null(row[0]))
			CONFIG_MASTER_NODEID = atoi(row[0]);
		DBfree_result(result);
	}

	init_database_cache();
	init_configuration_cache();
	init_selfmon_collector();

	zbx_create_itservices_lock();

	DCload_config();

	/* need to set trigger status to UNKNOWN since last run */
	DBupdate_triggers_status_after_restart();

	/* make initial configuration sync before worker processes are forked */
	DCsync_configuration();

	DBclose();

	if (ZBX_MUTEX_ERROR == zbx_mutex_create_force(&node_sync_access, ZBX_MUTEX_NODE_SYNC))
	{
		zbx_error("Unable to create mutex for node syncs");
		exit(FAIL);
	}

	threads_num = CONFIG_CONFSYNCER_FORKS + CONFIG_WATCHDOG_FORKS + CONFIG_POLLER_FORKS
			+ CONFIG_UNREACHABLE_POLLER_FORKS + CONFIG_TRAPPER_FORKS + CONFIG_PINGER_FORKS
			+ CONFIG_ALERTER_FORKS + CONFIG_HOUSEKEEPER_FORKS + CONFIG_TIMER_FORKS
			+ CONFIG_NODEWATCHER_FORKS + CONFIG_HTTPPOLLER_FORKS + CONFIG_DISCOVERER_FORKS
			+ CONFIG_HISTSYNCER_FORKS + CONFIG_ESCALATOR_FORKS + CONFIG_IPMIPOLLER_FORKS
			+ CONFIG_JAVAPOLLER_FORKS + CONFIG_SNMPTRAPPER_FORKS + CONFIG_PROXYPOLLER_FORKS
			+ CONFIG_SELFMON_FORKS;
	threads = zbx_calloc(threads, threads_num, sizeof(pid_t));

	if (0 < CONFIG_TRAPPER_FORKS)
	{
		if (FAIL == zbx_tcp_listen(&listen_sock, CONFIG_LISTEN_IP, (unsigned short)CONFIG_LISTEN_PORT))
		{
			zabbix_log(LOG_LEVEL_CRIT, "listener failed: %s", zbx_tcp_strerror());
			exit(1);
		}
	}

	for (i = 0; i < threads_num; i++)
	{
		if (0 == (pid = zbx_child_fork()))
		{
			server_num = i + 1;	/* child processes are numbered starting from 1 */
			break;
		}
		else
			threads[i] = pid;
	}

	if (0 == server_num)
	{
		zabbix_log(LOG_LEVEL_INFORMATION, "server #0 started [main process]");

		while (-1 == wait(&i))	/* wait for any child to exit */
		{
			if (EINTR != errno)
			{
				zabbix_log(LOG_LEVEL_ERR, "failed to wait on child processes: %s", zbx_strerror(errno));
				break;
			}
		}

		/* all exiting child processes should be caught by signal handlers */
		THIS_SHOULD_NEVER_HAPPEN;

		zbx_on_exit();
	}
	else if (server_num <= (server_count += CONFIG_CONFSYNCER_FORKS))
	{
		/* !!! configuration syncer must be server #1 - child_signal_handler() uses threads[0] !!! */

		INIT_SERVER(ZBX_PROCESS_TYPE_CONFSYNCER, CONFIG_CONFSYNCER_FORKS);

		main_dbconfig_loop();
	}
	else if (server_num <= (server_count += CONFIG_WATCHDOG_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_WATCHDOG, CONFIG_WATCHDOG_FORKS);

		main_watchdog_loop();
	}
	else if (server_num <= (server_count += CONFIG_POLLER_FORKS))
	{
#ifdef HAVE_SNMP
		init_snmp("zabbix_server");
#endif

		INIT_SERVER(ZBX_PROCESS_TYPE_POLLER, CONFIG_POLLER_FORKS);
		if (CONFIG_KAFKA_ENABLE == KAFKA_ENABLE){
			VNG_Kafkaconnect();
			if (!jConfTopic)
				VNG_LoadConfigTopic();
		}

		main_poller_loop(ZBX_POLLER_TYPE_NORMAL);
	}
	else if (server_num <= (server_count += CONFIG_UNREACHABLE_POLLER_FORKS))
	{
#ifdef HAVE_SNMP
		init_snmp("zabbix_server");
#endif

		INIT_SERVER(ZBX_PROCESS_TYPE_UNREACHABLE, CONFIG_UNREACHABLE_POLLER_FORKS);
		if (CONFIG_KAFKA_ENABLE == KAFKA_ENABLE){
			VNG_Kafkaconnect();
			if (!jConfTopic)
				VNG_LoadConfigTopic();
		}

		main_poller_loop(ZBX_POLLER_TYPE_UNREACHABLE);
	}
	else if (server_num <= (server_count += CONFIG_TRAPPER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_TRAPPER, CONFIG_TRAPPER_FORKS);

		main_trapper_loop(&listen_sock);
	}
	else if (server_num <= (server_count += CONFIG_PINGER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_PINGER, CONFIG_PINGER_FORKS);

		main_pinger_loop();
	}
	else if (server_num <= (server_count += CONFIG_ALERTER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_ALERTER, CONFIG_ALERTER_FORKS);

		main_alerter_loop();
	}
	else if (server_num <= (server_count += CONFIG_HOUSEKEEPER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_HOUSEKEEPER, CONFIG_HOUSEKEEPER_FORKS);

		main_housekeeper_loop();
	}
	else if (server_num <= (server_count += CONFIG_TIMER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_TIMER, CONFIG_TIMER_FORKS);

		main_timer_loop();
	}
	else if (server_num <= (server_count += CONFIG_NODEWATCHER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_NODEWATCHER, CONFIG_NODEWATCHER_FORKS);

		main_nodewatcher_loop();
	}
	else if (server_num <= (server_count += CONFIG_HTTPPOLLER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_HTTPPOLLER, CONFIG_HTTPPOLLER_FORKS);

		main_httppoller_loop();
	}
	else if (server_num <= (server_count += CONFIG_DISCOVERER_FORKS))
	{
#ifdef HAVE_SNMP
		init_snmp("zabbix_server");
#endif

		INIT_SERVER(ZBX_PROCESS_TYPE_DISCOVERER, CONFIG_DISCOVERER_FORKS);

		main_discoverer_loop();
	}
	else if (server_num <= (server_count += CONFIG_HISTSYNCER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_HISTSYNCER, CONFIG_HISTSYNCER_FORKS);

		main_dbsyncer_loop();
	}
	else if (server_num <= (server_count += CONFIG_ESCALATOR_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_ESCALATOR, CONFIG_ESCALATOR_FORKS);

		main_escalator_loop();
	}
	else if (server_num <= (server_count += CONFIG_IPMIPOLLER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_IPMIPOLLER, CONFIG_IPMIPOLLER_FORKS);

		main_poller_loop(ZBX_POLLER_TYPE_IPMI);
	}
	else if (server_num <= (server_count += CONFIG_JAVAPOLLER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_JAVAPOLLER, CONFIG_JAVAPOLLER_FORKS);

		main_poller_loop(ZBX_POLLER_TYPE_JAVA);
	}
	else if (server_num <= (server_count += CONFIG_SNMPTRAPPER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_SNMPTRAPPER, CONFIG_SNMPTRAPPER_FORKS);

		main_snmptrapper_loop();
	}
	else if (server_num <= (server_count += CONFIG_PROXYPOLLER_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_PROXYPOLLER, CONFIG_PROXYPOLLER_FORKS);

		main_proxypoller_loop();
	}
	else if (server_num <= (server_count += CONFIG_SELFMON_FORKS))
	{
		INIT_SERVER(ZBX_PROCESS_TYPE_SELFMON, CONFIG_SELFMON_FORKS);

		main_selfmon_loop();
	}

	return SUCCEED;
}

void	zbx_on_exit()
{
	zabbix_log(LOG_LEVEL_DEBUG, "zbx_on_exit() called");

	if (SUCCEED == DBtxn_ongoing())
		DBrollback();

	if (NULL != threads)
	{
		int		i;
		sigset_t	set;

		/* ignore SIGCHLD signals in order for zbx_sleep() to work */
		sigemptyset(&set);
		sigaddset(&set, SIGCHLD);
		sigprocmask(SIG_BLOCK, &set, NULL);

		for (i = 0; i < threads_num; i++)
		{
			if (threads[i])
			{
				kill(threads[i], SIGTERM);
				threads[i] = ZBX_THREAD_HANDLE_NULL;
			}
		}

		zbx_free(threads);
	}

	free_metrics();

	zbx_sleep(2);	/* wait for all child processes to exit */

	DBconnect(ZBX_DB_CONNECT_EXIT);
	free_database_cache();
	free_configuration_cache();

	zbx_destroy_itservices_lock();

	DBclose();

	zbx_mutex_destroy(&node_sync_access);

#ifdef HAVE_OPENIPMI
	free_ipmi_handler();
#endif

#ifdef HAVE_SQLITE3
	zbx_remove_sqlite3_mutex();
#endif

	free_selfmon_collector();

	zabbix_log(LOG_LEVEL_INFORMATION, "Zabbix Server stopped. Zabbix %s (revision %s).",
			ZABBIX_VERSION, ZABBIX_REVISION);

	zabbix_close_log();

	if (CONFIG_KAFKA_ENABLE == KAFKA_ENABLE){
		VNG_Kafkaclose();
		VNG_DestroyConfigTopic();
	}
	exit(SUCCEED);
}