#include "fieldstat.h"
#include "fieldstat_internal.h"
#include "threadsafe_counter.h"
#include "hdr_histogram.h"
#include "cJSON.h"
#define HTTPSERVER_IMPL
#include "httpserver.h"

#include <sys/socket.h>//socket
#include <sys/types.h>//socket
#include <netinet/in.h>
#include <arpa/inet.h>
#include <string.h> //strerror
#include <errno.h>//strerror
#include <fcntl.h>//fcntl
#include <unistd.h>//fcntl
#include <net/if.h>//fcntl
#include <sys/ioctl.h>//ioctl
#include <stdio.h>
#include <time.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#include <assert.h>
#include <sys/time.h>


int FIELD_STAT_VERSION_2_8_20200805_fix_outOfBound=0;

//Automatically generate the version number
#ifdef __cplusplus
extern "C"
{
#endif

#define GIT_VERSION_CATTER(v) __attribute__((__used__)) const char * GIT_VERSION_##v = NULL
#define GIT_VERSION_EXPEND(v) GIT_VERSION_CATTER(v)

/* VERSION TAG */
#ifdef GIT_VERSION
GIT_VERSION_EXPEND(GIT_VERSION);
#else
static __attribute__((__used__)) const char * GIT_VERSION_UNKNOWN = NULL;
#endif
#undef GIT_VERSION_CATTER
#undef GIT_VERSION_EXPEND

#ifdef __cplusplus
}
#endif
//endof Automatically generate the version number

struct prometheus_endpoint_instance g_prometheus_endpoint_instance = 
{
	9273,
	NULL,
	0,
	0,
	NULL,
	0,
	NULL,
	REALLOC_SCALE_SIZE,
};

const char* draw_line="________________________________________________________________________________________________________________________________________________";
const char* draw_boundary="============================================================";

static char* __str_dup(const char* str)
{
	char* dup=NULL;
	dup=(char*)calloc(sizeof(char),strlen(str)+1);
	memcpy(dup, str, strlen(str));
	return dup;
}

static int is_valid_field_name(const char* name)
{
	const char* reserverd="|:\n\r. \t<>[]#!@";
	unsigned int i=0,j=0;
	for(i=0;i<strlen(name);i++)
	{
		for(j=0;j<strlen(reserverd);j++)
		if(name[i]==reserverd[j])
		{
			return 0;
		}
	}
	return 1;
}

static char* str_unescape(char* s, char *d, int d_len)
{
	int i=0,j=0;
	int len=strlen(s);
	for(i=0; i<len && j<d_len; i++)
	{
		if(s[i]=='(' || s[i]==')'  || s[i]=='{' || s[i]=='}' || 
		   s[i]=='/' || s[i]=='\\' || s[i]=='%' || s[i]=='*' || 
		   s[i]=='$' || s[i]=='-'  || s[i]==',' || s[i]==';')
		{
			if(i==0)
			{
				continue;
			}

			d[j++]='_';
		}
		else
		{
			d[j++]=s[i];
		}
	}

	if(d[j-1]=='_')
	{
		j-=1;
	}

	d[j]='\0';
	
	return 0;
}


struct metric_t * get_metric(struct fieldstat_instance *instance, int metric_id)
{
	int block_index = 0;
	int block_in_index = 0;
	struct metric_t ** metrics_array = NULL;
	struct metric_t * metric = NULL;

	if(instance == NULL)
	{
		return NULL;
	}

	block_index = metric_id / NUM_INIT_METRICS;
	block_in_index = metric_id % NUM_INIT_METRICS;
	metrics_array = instance->metric_block_list[block_index];
	metric = metrics_array[block_in_index];

	return metric;
}


struct metric_t* metric_new(enum field_type type, const char *field_name, const char *tag_key[], const char *tag_value[], size_t n_tag)
{
	int i = 0;
	struct metric_t* metric=(struct metric_t*)calloc(sizeof(struct metric_t),1);
	metric->field_name =__str_dup(field_name);
	metric->field_type = type;
	metric->is_ratio = 0;
	metric->output_scaling = 1;
	metric->n_tag = n_tag;

	for(i = 0; i < (int)n_tag; i++)
	{
		metric->tag_key[i]   = __str_dup(tag_key[i]);
		metric->tag_value[i] = __str_dup(tag_value[i]);
	}
	return metric;
}

void metric_free(struct metric_t* metric)
{
	int i = 0;

	free(metric->field_name);
	metric->field_name = NULL;

	for(i = 0; i < (int)metric->n_tag; i++)
	{
		free(metric->tag_key[i]);
		metric->tag_key[i] = NULL;

		free(metric->tag_value[i]);
		metric->tag_value[i] = NULL;
	}
	metric->n_tag = 0;

	free(metric);

	return;
}

static struct metric_t ** read_metric_slot(struct fieldstat_instance *instance, int metric_id)
{
	int block_index = 0;
	int in_block_index = 0;
	struct metric_t ** metrics_block = NULL;

	if(instance == NULL)
	{
		return NULL;
	}
	block_index = metric_id / NUM_INIT_METRICS;
	in_block_index = metric_id % NUM_INIT_METRICS;
	if(in_block_index == 0)
	{
		assert(instance->metric_block_list[block_index] == NULL);
		instance->metric_block_list[block_index] = (struct metric_t **)calloc(sizeof(struct metric *), NUM_INIT_METRICS);	
	}
	else
	{
		while (instance->metric_block_list[block_index] == NULL);
	}
	metrics_block = (struct metric_t **)instance->metric_block_list[block_index];

	return (struct metric_t **)&(metrics_block[in_block_index]);
}


struct table_line * read_table_line(struct table_metric *table, int line_id)
{
	int block_index = 0;
	int block_in_index = 0;
	struct table_line **line_array = NULL;
	struct table_line *line = NULL;

	block_index = line_id / NUM_INIT_METRICS;
	block_in_index = line_id % NUM_INIT_METRICS;
	line_array = table->line_block[block_index];
	line = line_array[block_in_index];

	return line;
}


static int startup_udp()
{
	int sd_udp=-1;
	int flags;
	int opt=1;
 	if(-1==(sd_udp = socket(AF_INET, SOCK_DGRAM, 0)))
	{
		printf("FS2: socket error: %d %s, restart socket.", errno, strerror(errno));
		sd_udp=-1;
	}
	flags=fcntl(sd_udp,F_GETFL);
	flags|=O_NONBLOCK;
	if(fcntl(sd_udp,F_SETFL,flags)==-1)
	{
		printf("FS2: socket error: %d %s, restart socket.", errno, strerror(errno));
		sd_udp=-1;
	}
	if (setsockopt (sd_udp, SOL_SOCKET, SO_REUSEADDR, (char *)&opt, sizeof(opt) ) < 0) 
	{
		printf("FS2:setsockopt error: %d %s, restart socket.", errno, strerror(errno));
		close (sd_udp);
		sd_udp=-1;
		return sd_udp;
	}

	return sd_udp;
}

static int send_udp(int sd, unsigned int dest_ip, unsigned short dest_port, const char * data, int len)
{
	int to_send_len=len;
	int already_sended_len=0,this_sended_len=0;
	struct sockaddr_in addr;
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr =dest_ip;
	addr.sin_port = htons(dest_port);

	while(to_send_len>already_sended_len)
	{
		this_sended_len=sendto(sd,(void*)(data+already_sended_len),
								to_send_len-already_sended_len,
								0,
								(struct sockaddr *)&(addr),
								sizeof(addr));
		if(this_sended_len==-1)
		{
			if((EAGAIN == errno)||( EINTR == errno )|| (EWOULDBLOCK==errno))
			{
				continue;
			}
			else
			{
				printf("FS2: at send,socket error: %d %s", errno, strerror(errno));
				return -1;
			}
		}
		already_sended_len=+this_sended_len;			
	}
	
	return 0;
}

int fieldstat_set_output_interval(struct fieldstat_instance *instance, int seconds)
{
	if(instance->running == 1 || seconds <= 0 )
	{
		return -1;
	}
	instance->output_interval_s = seconds;
	return 0;
}

int fieldstat_background_thead_disable(struct fieldstat_instance *instance)
{
	if(instance->running == 1)
	{
		return -1;
	}
	instance->background_thread_disable = 1;
	return 0;
}


int fieldstat_set_local_output(struct fieldstat_instance *instance, const char *filename, const char *format)
{
	int len_filename = strlen(filename);
	int len_format   = strlen(format);

	if(instance->running == 1)
	{
		return -1;
	}
	if(strcmp(format,"default") != 0 && strcmp(format,"json") != 0)
	{
		return -1;
	}
	if(len_filename <= 0 || len_filename >= LEN_PATH_MAX)
	{
		return -1;
	}
	if(len_format <= 0 || len_format >= LEN_FORMAT_MAX)
	{
		return -1;
	}
	strncpy(instance->local_output_filename, (char *)filename, len_filename);
	strncpy(instance->local_output_format, (char *)format, len_format);
	instance->local_output_enable  = 1;

	return 0;
}

int fieldstat_set_line_protocol_server(struct fieldstat_instance *instance, const char *ip, unsigned short port)
{
	if(instance->running == 1)
	{
		return -1;
	}

	if(1 != inet_pton(AF_INET, ip, (void *)&(instance->line_protocol_server_ip)))
	{
		return -1;
	}
	instance->line_protocol_socket = startup_udp();
	instance->line_protocol_server_port   = port;
	instance->line_protocol_output_enable = 1;

	return 0;
}

int fieldstat_set_statsd_server(struct fieldstat_instance *instance, const char *ip, unsigned short port)
{
	if(instance->running == 1)
	{
		return -1;
	}

	if(1 != inet_pton(AF_INET, ip, (void *)&(instance->statsd_server_ip)))
	{
		return -1;
	}

	instance->statsd_server_port   = port;
	instance->statsd_output_enable = 1;
	return 0;
}

int fieldstat_register(struct fieldstat_instance *instance, enum field_type type, const char *field_name, const char *tag_key[], const char *tag_value[], size_t n_tag)
{
	int metric_id = 0;
	struct metric_t * metric = NULL;
	struct metric_t **metric_slot = NULL;
	if(!is_valid_field_name(field_name))
	{
		return -1;
	}
	if(n_tag > N_TAG_MAX)
	{
		return -1;
	}

	metric_id = atomic_inc(&instance->metric_cnt) - 1;
	metric_slot = read_metric_slot(instance, metric_id);
	metric = metric_new(type,field_name,tag_key,tag_value,n_tag);
	*metric_slot = metric;

	switch(type)
	{
		case FIELD_TYPE_COUNTER:
			memset(&(metric->counter), 0, sizeof(metric->counter));
			break;
		case FIELD_TYPE_GAUGE:
			memset(&(metric->gauge), 0, sizeof(metric->gauge));
			break;
		default:
			assert(0);
	}
	return metric_id;
}

long long get_metric_unit_val(struct metric_t *metric,enum field_calc_algo calc_type,int is_refer)
{
	stat_unit_t* target = NULL;
	long long value = 0;
	switch(metric->field_type)
	{
		case FIELD_TYPE_COUNTER:
			target = &(metric->counter);
			break;
		case FIELD_TYPE_GAUGE:
			target = &(metric->gauge);
			break;
		default:
			break;
	}
	value = threadsafe_counter_read(&(target->changing));
	//value= threadsafe_counter_read(&(target->changing));
	if(is_refer == 0)
	{
		target->previous_changed = value;
		target->accumulated += value;
		threadsafe_counter_set(&(target->changing), 0);
	}
	switch(calc_type)
	{
		case	FS_CALC_CURRENT:
			value=target->accumulated;
			break;
		case	FS_CALC_SPEED:
			value=target->previous_changed;
			break;
		default:
			assert(0);
			break;
	}
	return value;
}


void flush_line_protocol_metric(struct fieldstat_instance *instance)
{
	if(instance->line_protocol_send_buff_offset == 0)
	{
		return;
	}

	if(instance->line_protocol_server_ip > 0 && instance->line_protocol_server_port > 0)
	{
		send_udp(instance->line_protocol_socket, instance->line_protocol_server_ip,
				(unsigned short)instance->line_protocol_server_port, 
				instance->line_protocol_send_buff, 
				instance->line_protocol_send_buff_offset
			);
	}
	instance->line_protocol_send_buff_offset = 0;
	memset(instance->line_protocol_send_buff, 0, sizeof(instance->line_protocol_send_buff));	
	return;
}

void append_line_protocol_line(struct fieldstat_instance *instance, const char* measurement, char *tag_set, char *field_set)
{
	if(field_set==NULL)
	{
		return;
	}
	if(UDP_PAYLOAD_SIZE - (unsigned int)instance->line_protocol_send_buff_offset < strlen(measurement) + strlen(field_set) + strlen(tag_set) + 2)
	{
		flush_line_protocol_metric(instance);
	}
	printf("Line_protocol metric: %s%s %s\n",measurement,tag_set,field_set);
	instance->line_protocol_send_buff_offset += snprintf(instance->line_protocol_send_buff + instance->line_protocol_send_buff_offset,
														sizeof(instance->line_protocol_send_buff) - instance->line_protocol_send_buff_offset,
														"%s%s %s\n",
														measurement, tag_set, field_set
													);
	return;
}

static int output_line_protocol_tag_set_buf(char *tag_key[], char *tag_value[], int n_tag, char *tag_set_buff, unsigned int size)
{
	int i = 0;
	char *tag_pos = tag_set_buff;
	for(i = 0; i < n_tag; i++)
	{
		tag_pos += snprintf(tag_pos, 
							size - (tag_pos - tag_set_buff), 
							",%s=%s", 
							tag_key[i],
							tag_value[i]
						);
	}
	return tag_pos - tag_set_buff;
}


static void output_line_protocol_table(struct fieldstat_instance *instance)
{
	int i = 0, j = 0, k = 0;
	metric_t *metric = NULL;
	long long value = 0;
	//double ratio = 0.0;
	char field_set_buff[UDP_PAYLOAD_SIZE];
	char tag_set_buff[UDP_PAYLOAD_SIZE];

	memset(field_set_buff, 0, sizeof(field_set_buff));
	memset(tag_set_buff,  0, sizeof(tag_set_buff));
	char *tag_pos = tag_set_buff;
	char *field_pos = field_set_buff;

	struct table_metric *table = NULL;
	struct table_line *line = NULL;

	for(i = 0; i < instance->table_num; i++)
	{
		table = instance->table_metrics[i];
		for(j = 0; j < table->line_cnt; j++)
		{
			line = read_table_line(table, j);
			tag_pos += snprintf(tag_pos, 
						sizeof(tag_set_buff) - (tag_pos - tag_set_buff), 
						",app_name=%s,table_name=%s",
						instance->name,
						table->name
					);
			
			tag_pos += output_line_protocol_tag_set_buf(line->tag_key, line->tag_value, line->n_tag, tag_pos, 
														sizeof(tag_set_buff) - (tag_pos - tag_set_buff));

			for(k = 0; k < table->column_cnt; k ++)
			{
				metric = get_metric(instance, line->metric_id_belong_to_line[k]);

				value = metric->field_type == FIELD_TYPE_GAUGE ? 
					get_metric_unit_val(metric, FS_CALC_CURRENT, 1):
					get_metric_unit_val(metric, FS_CALC_SPEED, 1);

					field_pos += snprintf(field_pos, 
											sizeof(field_set_buff) - (field_set_buff - field_set_buff), 
											"%s=%lld,", 
											metric->table_column_name, 
											value
										);

			}
			if(field_pos - field_set_buff > 0)
			{
				*(field_pos - 1) = '\0';
			}
			// measurement,tag_set field_set
			append_line_protocol_line(instance, metric->field_name, tag_set_buff, field_set_buff);
			tag_pos = tag_set_buff;
			field_pos = field_set_buff;

		}

	}
}

int line_protocol_output(struct fieldstat_instance *instance)
{
	metric_t *metric = NULL;
	long long value=0;
	int i=0;
	char field_set_buff[UDP_PAYLOAD_SIZE];
	char tag_set_buff[UDP_PAYLOAD_SIZE];

	memset(field_set_buff, 0, sizeof(field_set_buff));
	memset(tag_set_buff,  0, sizeof(tag_set_buff));
	char *tag_pos = tag_set_buff;

	for(i = 0; i < instance->metric_cnt; i++)
	{
		metric = get_metric(instance, i);
		if(metric->is_ratio == 1)
		{
			continue;
		}

		if(metric->belong_to_table == 1)
		{
			continue;
		}

		switch(metric->field_type)
		{
			case FIELD_TYPE_GAUGE:
				value = get_metric_unit_val(metric, FS_CALC_CURRENT, 1);
				if(value != 0)
				{
					snprintf(field_set_buff, UDP_PAYLOAD_SIZE, "%s=%lld", metric->field_name, value);
					tag_pos += snprintf(tag_pos, 
										sizeof(tag_set_buff) - (tag_pos - tag_set_buff), 
										",app_name=%s", 
										instance->name
									);
					output_line_protocol_tag_set_buf(metric->tag_key, metric->tag_value, metric->n_tag, tag_pos, sizeof(tag_set_buff) - (tag_pos - tag_set_buff));
					append_line_protocol_line(instance, metric->field_name, tag_set_buff, field_set_buff);
					tag_pos = tag_set_buff;

				}
				break;
			case FIELD_TYPE_COUNTER:
				value = get_metric_unit_val(metric, FS_CALC_SPEED, 1);
				if(value != 0)
				{
					snprintf(field_set_buff, UDP_PAYLOAD_SIZE, "%s=%lld", metric->field_name, value);
					tag_pos += snprintf(tag_pos, 
											sizeof(tag_set_buff) - (tag_pos - tag_set_buff), 
											",app_name=%s", 
											instance->name
										);
					output_line_protocol_tag_set_buf(metric->tag_key, metric->tag_value, metric->n_tag, tag_pos, sizeof(tag_set_buff) - (tag_pos - tag_set_buff));
					append_line_protocol_line(instance, metric->field_name, tag_set_buff, field_set_buff);
					tag_pos = tag_set_buff;
				}
				break;
			default:
				break;

		}
	}

	output_line_protocol_table(instance);
	flush_line_protocol_metric(instance);

	return 0;
}


static int print_buf_tag_append_position(char *tag_key[], char *tag_value[], size_t n_tag, char *print_buf_tags, unsigned int size)
{
	int i = 0;
	char *append_pos = print_buf_tags;

	if(n_tag <= 0)
	{
		return 0;
	}

	append_pos += snprintf(append_pos, 
							size - (append_pos - print_buf_tags),
							"{"
						);

	for(i = 0; i < (int)n_tag; i++)
	{
		append_pos += snprintf(append_pos,
								size - (append_pos - print_buf_tags),
								"%s=\"%s\",", 
								tag_key[i],
								tag_value[i]
							);
	}

	if(append_pos - print_buf_tags > 0)
	{
		append_pos--;
	}

	append_pos += snprintf(append_pos, 
							size - (append_pos - print_buf_tags),
							"}\t"
						);

	return append_pos - print_buf_tags;
}


static int output_file_format_default_type_gauge(struct fieldstat_instance *instance,long long interval_ms,char *print_buf, unsigned int size)
{
	int i = 0, j = 0;
	//display_manifest_t* p = NULL;
	metric_t *metric = NULL;
	long long value = 0;
	//double ratio = 0.0;
	//char* pos=print_buf;
	char *append_pos = print_buf;
	char print_buf_tags[1024];

	for(i = 0; i < instance->metric_cnt; i++)
	{
		//metric = instance->metric[i];
		metric = get_metric(instance, i);
		if(metric->field_type != FIELD_TYPE_GAUGE)
		{
			continue;
		}
		if(metric->belong_to_table == 1)
		{
			continue;
		}
		if(metric->is_invisible == 1)
		{
			value = get_metric_unit_val(metric, FS_CALC_SPEED, 0);
			continue;
		}
		/*
		if(metric->is_ratio==1)
		{
			ratio=get_stat_ratio(_handle->display[p->numerator_id],  _handle->display[p->denominator_id], NULL,
								p->output_scaling, p->calc_type);
			pos+=snprintf(pos,size-(pos-print_buf),"%s: %10.2e\t",p->name,ratio);
		}
		*/
		value = get_metric_unit_val(metric, FS_CALC_CURRENT, 0);
		//value=value * metric->output_scaling * 1000 / interval_ms;
		memset(print_buf_tags,0, sizeof(print_buf_tags));
		print_buf_tag_append_position(metric->tag_key,metric->tag_value, metric->n_tag, print_buf_tags, sizeof(print_buf_tags));
		append_pos += snprintf(append_pos, 
								size - (append_pos - print_buf), 
								"%s %s: %-10lld\t", 
								metric->field_name, 
								print_buf_tags, 
								value
							);
		j++;
		if(j == STATUS_PER_LINE)
		{
			append_pos += snprintf(append_pos, size - (append_pos - print_buf),"\n");
			j=0;
		}
	}

	if(append_pos - print_buf > 0)
	{
		if(*(append_pos - 1) == '\n')
		{
			append_pos --;
		}
		append_pos += snprintf(append_pos, size - (append_pos-print_buf),"\n%s\n",draw_line);
	}
	return append_pos - print_buf;
}


static int output_file_format_default_type_counter(struct fieldstat_instance *instance,long long interval_ms,char*print_buf, unsigned int size)
{
	int i=0,j=0;
	//display_manifest_t* p=NULL;
	metric_t *metric = NULL;
	long long value = 0;
	//double ratio = 0.0;
	char* append_pos = print_buf;
	int metric_id[INIT_STAT_FIELD_NUM] = {0};
	int metric_cnt = 0;
	char print_buf_tags[1024];

	for(i = 0;i < instance->metric_cnt; i++)
	{
		//p=_handle->display[i];
		//metric = instance->metric[i];
		metric = get_metric(instance, i);
		if(metric->field_type != FIELD_TYPE_COUNTER)
		{
			continue;
		}
		if(metric->belong_to_table == 1)
		{
			continue;
		}
		if(metric->is_invisible == 1)
		{
			get_metric_unit_val(metric,FS_CALC_CURRENT,0);
			continue;
		}
		metric_id[metric_cnt] = i;
		metric_cnt++;
	}

	for(i = 0; i < metric_cnt; i++)
	{
		append_pos += snprintf(append_pos, size - (append_pos - print_buf),"\t");

		for(j = 0; j < FIELD_PER_LINE && i+j < metric_cnt; j++)
		{
			//metric = instance->metric[metric_id[i+j]];
			metric = get_metric(instance, metric_id[i+j]);
			memset(print_buf_tags,0, sizeof(print_buf_tags));
			print_buf_tag_append_position(metric->tag_key,metric->tag_value, metric->n_tag, print_buf_tags, sizeof(print_buf_tags));
			append_pos += snprintf(append_pos, 
									size - (append_pos - print_buf), 
									"%10s %s\t", 
									metric->field_name, 
									print_buf_tags
								);
		}
		append_pos += snprintf(append_pos, size - (append_pos-print_buf), "\nsum\t");

		for(j=0; j < FIELD_PER_LINE && i+j < metric_cnt; j++)
		{
			//metric = instance->metric[metric_id[i+j]];
			metric = get_metric(instance, metric_id[i+j]);
			value  = get_metric_unit_val(metric,FS_CALC_CURRENT, 1);
			append_pos += snprintf(append_pos, 
									sizeof(print_buf) - (append_pos - print_buf), 
									"%10lld\t", 
									value
								);
		}
		append_pos += snprintf(append_pos, 
								sizeof(print_buf) - (append_pos - print_buf), 
								"\nspeed/s\t"
							);

		for(j=0;j<FIELD_PER_LINE&&i+j<metric_cnt;j++)
		{
			//metric = instance->metric[metric_id[i+j]];
			metric = get_metric(instance, metric_id[i+j]);
			value  = get_metric_unit_val(metric,FS_CALC_SPEED, 0);
			append_pos += snprintf(append_pos, 
									size - (append_pos - print_buf), 
									"%10lld\t", 
									value*1000/interval_ms
								);
		}
		i += (j-1);
		append_pos += snprintf(append_pos, 
								size - (append_pos - print_buf), 
								"\n"
							);
	}

	if(append_pos - print_buf > 0)
	{
		if(*(append_pos - 1)=='\n')
		{
			append_pos--;
		}
		append_pos += snprintf(append_pos, 
								size - (append_pos - print_buf),
								"\n%s\n", 
								draw_line
							);
	}

	return append_pos - print_buf;
}

static int output_file_format_default_table(struct fieldstat_instance *instance,long long interval_ms,char*print_buf, unsigned int size)
{
	int i = 0, j = 0, k = 0;
	struct table_metric *table =  NULL;
	struct table_line  *line = NULL;
	int metric_id = 0;
	struct metric_t *metric = NULL;
	long long value = 0;
	char* append_pos = print_buf;

	for(i = 0; i < instance->table_num; i++)   //per table
	{
		table = instance->table_metrics[i];
		append_pos += snprintf(append_pos, size - (append_pos - print_buf),"%-20s\t\t",table->name);
		for(j = 0; j < table->column_cnt; j ++)
		{
			//print table column
			append_pos += snprintf(append_pos,
									size - (append_pos - print_buf), 
									"\t%10s", 
									table->column_name[j]
								);
		}
		for(j = 0; j < table->line_cnt; j++)  //per line
		{
			line = read_table_line(table,j);
			//print table line name + tag
			append_pos += snprintf(append_pos,
									size - (append_pos - print_buf), 
									"\n%s ", 
									line->name
								);

			append_pos += print_buf_tag_append_position(line->tag_key, line->tag_value, line->n_tag, 
														append_pos, size - (append_pos - print_buf));

			
			for(k = 0; k < table->column_cnt; k++) //per metric
			{
				//print table metric value
				metric_id = line->metric_id_belong_to_line[k];
				metric = get_metric(instance,metric_id);
				value = get_metric_unit_val(metric, FS_CALC_CURRENT, 0);
				append_pos += snprintf(append_pos, 
										size - (append_pos - print_buf), 
										"%10lld\t", 
										value
									);

			}

		} 

		if(append_pos - print_buf > 0)
		{
			if(*(append_pos - 1) == '\n')
			{
				append_pos--;
			}
			append_pos += snprintf(append_pos, 
									size - (append_pos - print_buf), 
									"\n%s\n", 
									draw_line
								);
		}

	}
	return append_pos - print_buf;
}

static int output_file_print_hdr_head(struct metric_t *metric, char *print_buf, size_t size)
{
	char * pos = print_buf;
	char bin_format[STR_LEN_256], str_format[STR_LEN_256];
	const char* extra[]={"MAX", "MIN", "AVG", "STDDEV", "CNT"};
	char buff[STR_LEN_32];
	int i=0;
	double * bins = metric->histogram.bins;
	int bins_num  = metric->histogram.bins_num;
	if(metric->field_type == FIELD_TYPE_SUMMARY)
	{
		snprintf(bin_format, sizeof(bin_format), "%%%d.2lf%%%%", HISTOGRAM_WIDTH-1);
	}
	if(metric->field_type == FILED_TYPE_HISTOGRAM)
	{
		snprintf(bin_format, sizeof(bin_format), "le=%%0.0f");
	}
	snprintf(str_format, sizeof(str_format), "%%%ds", HISTOGRAM_WIDTH);
	if(metric->field_type == FIELD_TYPE_SUMMARY)
	{
		pos+=snprintf(pos,size-(pos-print_buf),"%-8s\t","summary");
	}
	if(metric->field_type == FILED_TYPE_HISTOGRAM)
	{
		pos+=snprintf(pos,size-(pos-print_buf),"%-8s\t","histogram");
	}
	for(i = 0;i < bins_num; i++)
	{
		if(metric->field_type == FIELD_TYPE_SUMMARY)
		{
			snprintf(buff,sizeof(buff),bin_format,bins[i]*100);
		}
		if(metric->field_type == FILED_TYPE_HISTOGRAM)
		{
			snprintf(buff,sizeof(buff),bin_format,bins[i]);
		}
		pos+=snprintf(pos,size-(pos-print_buf),str_format, buff);
	}
	for(i=0;(unsigned int)i<sizeof(extra)/sizeof(extra[0]);i++)
	{
		pos+=snprintf(pos,size-(pos-print_buf),str_format, extra[i]);
	}
	pos+=snprintf(pos,size-(pos-print_buf),"\n");
	return pos-print_buf;
}


long long hdr_count_le_value(const struct hdr_histogram* h, long long value)
{
	struct hdr_iter iter;
	long long count = 0;
	hdr_iter_recorded_init(&iter, h);
	while (hdr_iter_next(&iter))
	{
		if((long long)iter.value <= value)
			count ++;
	}
	return count;
}
static int output_file_print_hdr_unit(struct metric_t *metric, char*print_buf, size_t size)
{
	double * bins = metric->histogram.bins;
	int bins_num  = metric->histogram.bins_num;
	char* pos=print_buf;
	long long value=0;
	int i=0;
	struct histogram_t* h=&(metric->histogram);
	struct hdr_histogram* h_out=NULL, *h_tmp=NULL;
	char int_format[STR_LEN_256], double_format[STR_LEN_256];
	snprintf(int_format, sizeof(int_format), "%%%dlld",HISTOGRAM_WIDTH);
	snprintf(double_format, sizeof(double_format), "%%%d.2lf",HISTOGRAM_WIDTH);

	hdr_init(h->lowest_trackable_value,  h->highest_trackable_value, h->significant_figures, &(h_tmp));
	if(h->previous_changed!=NULL)	hdr_close(h->previous_changed);

	h->previous_changed=atomic_read(&(h->changing));
	h_tmp=atomic_set(&(h->changing), h_tmp);// left h_tmp is used to avoid warining [-Wunused-value]

	hdr_add(h->accumulated, h->previous_changed);
	h_out=h->accumulated; //TODO
/*
	if(metric->calc_type==FS_CALC_SPEED)
	{
		h_out=h->previous_changed;
	}
	else
	{
		h_out=h->accumulated;
	}
*/
	pos+=snprintf(pos,size-(pos-print_buf),"%-10s\t", metric->field_name);

	for(i=0;i<bins_num;i++)
	{
		if(metric->field_type == FIELD_TYPE_SUMMARY)
		{
			value=(long long)hdr_value_at_percentile(h_out, bins[i]*100);
		}
		if(metric->field_type == FILED_TYPE_HISTOGRAM)
		{
			value= hdr_count_le_value(h_out, (long long)bins[i]);
		}
		pos+=snprintf(pos,size-(pos-print_buf),int_format, value);
	}
	pos+=snprintf(pos,size-(pos-print_buf),int_format,h_out->total_count==0?0:(long long)hdr_max(h_out));
	pos+=snprintf(pos,size-(pos-print_buf),int_format,h_out->total_count==0?0:(long long)hdr_min(h_out));
	pos+=snprintf(pos,size-(pos-print_buf),double_format,h_out->total_count==0?0:hdr_mean(h_out));
	pos+=snprintf(pos,size-(pos-print_buf),double_format,h_out->total_count==0?0:hdr_stddev(h_out));
	pos+=snprintf(pos,size-(pos-print_buf),int_format,(long long)h_out->total_count);

	pos+=snprintf(pos,size-(pos-print_buf),"\n");

	h_tmp=NULL;

	return pos-print_buf;
}

static int output_file_format_default_type_histogram_and_summary(struct fieldstat_instance *instance, long long interval_ms, char*print_buf, size_t size)
{
	int i = 0, j = 0, metric_num = 0;
	char *pos = print_buf;
	//display_manifest_t* p=NULL;
	struct metric_t *metric = NULL;
	struct metric_t  *metric_array[INIT_STAT_FIELD_NUM] = {NULL};
	int metric_is_print[INIT_STAT_FIELD_NUM] = {0};

	if(instance->histogram_cnt == 0 
		&& instance->summary_cnt == 0)
	{
		return 0;
	}

	for(i = 0; i < instance->metric_cnt; i ++)
	{
		metric = get_metric(instance, i);
		if(metric->field_type != FIELD_TYPE_SUMMARY 
			&& metric->field_type != FILED_TYPE_HISTOGRAM)
		{
			continue;
		}
		metric_array[metric_num++] = metric;	
	}

	for(i = 0; i < metric_num; i++)
	{
		if(metric_is_print[i] == 1)
		{
			continue;
		}
		pos += output_file_print_hdr_head(metric_array[i], pos, size-(pos-print_buf));
		for(j = i; j < metric_num; j ++)
		{
			if(metric_array[j] == NULL)
			{
				continue;
			}
			if(metric_array[i]->histogram.bins_num != metric_array[j]->histogram.bins_num)
			{
				continue;
			}
			if(memcmp(metric_array[i]->histogram.bins, metric_array[j]->histogram.bins, metric_array[i]->histogram.bins_num))
			{
				continue;
			}
			pos += output_file_print_hdr_unit(metric_array[j], pos, size-(pos-print_buf));
			metric_is_print[j] = 1;
		}
		if(pos-print_buf>0)
		{
			if(*(pos-1)=='\n')
			{
				pos--;
			}
			pos+=snprintf(pos,size-(pos-print_buf),"\n%s\n", draw_line);
		}
	}
	return pos-print_buf;	
}


int fieldstat_output_file(struct fieldstat_instance *instance,long long interval_ms)
{
	size_t print_buf_sz = instance->metric_cnt*1024;
	char *print_buf = NULL;
	char *append_pos = NULL;
	time_t current = 0;
	char ctime_buff[STR_LEN_32]={0};

	if(instance->fp == NULL)
	{
		instance->fp = fopen(instance->local_output_filename, "w");
		if(instance->fp == NULL)
		{
			printf("Field Stat: open %s failed.\n",instance->local_output_filename);
			assert(0);
			return -1;
		}
	}

	if(!strcmp(instance->local_output_format, "default"))
	{
		time(&current);
		ctime_r(&current, ctime_buff);
		print_buf = (char*)calloc(sizeof(char), print_buf_sz);
		append_pos = print_buf;
		append_pos += snprintf(append_pos, print_buf_sz - (append_pos - print_buf), "%s%s", draw_boundary, ctime_buff);
		append_pos --;//jump '\n' generate by ctime()
		append_pos += snprintf(append_pos, print_buf_sz - (append_pos - print_buf),"%s\n",draw_boundary);

		//pthread_mutex_lock(&(_handle->reg_lock)); //TODO
		append_pos += output_file_format_default_type_gauge(instance, interval_ms, append_pos, print_buf_sz - (append_pos - print_buf));
		append_pos += output_file_format_default_type_counter(instance, interval_ms, append_pos, print_buf_sz - (append_pos - print_buf));
		append_pos += output_file_format_default_table(instance, interval_ms, append_pos, print_buf_sz - (append_pos - print_buf));
		append_pos += output_file_format_default_type_histogram_and_summary(instance, interval_ms, append_pos, print_buf_sz - (append_pos - print_buf));
		//TODO  output table,output histogram,output summary 
		//pthread_mutex_unlock(&(_handle->reg_lock));//TODO
	}

    if(!strcmp(instance->local_output_format, "json"))
    {
		//TODO from json output
    }

	fseek(instance->fp,0,SEEK_SET);
	fwrite(print_buf,append_pos - print_buf,1,instance->fp);

	fflush(instance->fp);

	if(print_buf)
	{
		free(print_buf);
		print_buf = NULL;
	}
	return 0;
}


void fieldstat_passive_output(struct fieldstat_instance *instance)
{
	struct timespec this_output_time;
	long long interval_ms = 0;
	int ret = 0;

	if(instance->running == 0)
	{
		return;
	}

	clock_gettime(CLOCK_MONOTONIC ,&this_output_time);
	interval_ms = (this_output_time.tv_sec - instance->last_output_time.tv_sec) * 1000 + (this_output_time.tv_nsec - instance->last_output_time.tv_nsec) / 1000000;
	if(interval_ms < 1)
	{
		printf("Passive return\n");
		return;
	}

	if(instance->local_output_enable)
	{
		ret = fieldstat_output_file(instance, interval_ms);
	}
	if(instance->line_protocol_output_enable)
	{
		ret = line_protocol_output(instance);
	}
	if(ret == -1)
	{
		return;
	}
	memcpy(&(instance->last_output_time),&this_output_time, sizeof(this_output_time));
}

void *fieldstat_thread_schema_output(void *arg)
{
	struct fieldstat_instance *instance=(struct fieldstat_instance *)arg;
	while(instance->background_thread_disable == 0)
	{
		fieldstat_passive_output(instance);
		sleep(instance->output_interval_s);
	}
	return NULL;
}


void fieldstat_instance_start(struct fieldstat_instance *instance)
{
	instance->running = 1;
	clock_gettime(CLOCK_MONOTONIC,&(instance->last_output_time));
	if(instance->background_thread_disable == 0)
	{
		pthread_create(&(instance->cfg_mon_t), NULL, fieldstat_thread_schema_output, (void*)instance);
	}
	//append instance to prometheus output
}

struct fieldstat_instance * fieldstat_instance_create(const char *name)
{
	struct fieldstat_instance *instance = (struct fieldstat_instance *)calloc(sizeof(struct fieldstat_instance),1);

	if(strlen(name) >= INSTANCE_NAME_LEN)
	{
		return NULL;
	}

	strcpy(instance->name, name);
	instance->running = 0;
	instance->output_interval_s = 2;
	instance->background_thread_disable = 0;

	return instance;
}

struct table_metric* table_metric_new(const char *name, const char *column_name[], enum field_type column_type[], size_t n_column)
{
	int i = 0;
	struct table_metric *table_metric = (struct table_metric *)calloc(sizeof(struct table_metric), 1);

	table_metric->column_cnt = (int)n_column;
	table_metric->name = __str_dup(name);

	for(i = 0; i < (int)n_column; i++)
	{
		table_metric->column_name[i] = __str_dup(column_name[i]);
		table_metric->column_type[i] = column_type[i];
	}
	return table_metric;
}

 int fieldstat_register_table(struct fieldstat_instance *instance, const char *name, const char *column_name[], enum field_type column_type[], size_t n_column)
 {
	int table_id = 0;

	struct table_metric *table_metric = NULL;

	if(n_column <= 0 || n_column > TABLE_COLUMN_SIZE)
	{
		return -1;
	}

	if(instance->table_num > TABLE_MAX_NUM)
	{
		return -1;
	}

	table_id = atomic_inc(&instance->table_num) - 1;
	table_metric = table_metric_new(name, column_name, column_type, n_column);
	instance->table_metrics[table_id] = table_metric;

	return table_id;
 }

static struct table_line ** read_table_line_slot(struct table_metric *table, int line_id)
{
	int block_index = 0;
	int in_block_index = 0;
	struct table_line **line = NULL;

	block_index = line_id / NUM_INIT_METRICS;
	in_block_index = line_id % NUM_INIT_METRICS;
	if(in_block_index == 0)
	{
		assert(table->line_block[in_block_index] == NULL);
		table->line_block[block_index] = (struct table_line **)calloc(sizeof(struct table_line *), NUM_INIT_METRICS); 	
	}
	else
	{
		while (table->line_block[block_index] == NULL);
	}
	line = table->line_block[block_index];

	return (struct table_line **)&(line[in_block_index]);
}

static struct table_line *table_line_new(const char *name, const char *tag_key[], const char *tag_value[],size_t n_tag)
{
	int i = 0;
	struct table_line *table_line = (struct table_line *)calloc(sizeof(struct table_line), 1);

	table_line->name = __str_dup(name);
	table_line->n_tag = n_tag;

	for(i = 0; i < (int)n_tag; i++)
	{
		table_line->tag_key[i] = __str_dup(tag_key[i]);
		table_line->tag_value[i] = __str_dup(tag_value[i]);
	}
	return table_line;
}


struct metric_id_list fieldstat_register_table_metrics(struct fieldstat_instance * instance, int table_id, const char *line_name, const char *tag_key[], const char *tag_value[],size_t n_tag)
{
	int metric_id = 0;
	struct metric_t *metric = NULL;

	int line_id = 0;
	int i = 0;

	struct table_metric *table = NULL;
	struct table_line **line_slot = NULL;
	struct table_line *table_line = NULL;
	struct metric_id_list ret_metric_id_list;
	memset(&ret_metric_id_list, 0, sizeof(struct metric_id_list));
	

	if(table_id < 0 || table_id >= TABLE_MAX_NUM)
	{
		return ret_metric_id_list;
	}
	if(!is_valid_field_name(line_name) || n_tag > N_TAG_MAX)
	{
		return ret_metric_id_list;
	}

	table = instance->table_metrics[table_id];
	line_id = atomic_inc(&(table->line_cnt)) - 1;
	line_slot = read_table_line_slot(table,line_id);
	table_line = table_line_new(line_name, tag_key, tag_value, n_tag);
	*line_slot = table_line;

	for(i = 0; i < table->column_cnt; i++)
	{
		metric_id = fieldstat_register(instance, table->column_type[i], line_name,tag_key, tag_value, n_tag);

		table_line->metric_id_belong_to_line[i] = metric_id;

		metric = get_metric(instance, metric_id);
		metric->table_id = table_id;
		metric->table_column_name = __str_dup(table->column_name[i]); 
		metric->table_name = __str_dup(table->name);
		metric->belong_to_table = 1;
		ret_metric_id_list.id[ret_metric_id_list.count ++] = metric_id;

	}

	return ret_metric_id_list;
}

/*
* ret = -1, output not match fieldstat instance
	output:
		/metrics
		/sapp
		/tfe
		...
* ret >=0 && ret < instance_cnt  output sepecify fieldstat instance 
	output:
		http://127.0.0.1:9273/sapp content
* ret = instance_cnt output all fieldstat instance
   	output:  
		http://127.0.0.1:9273/metrics content
*/
static int prometheus_get_fs_instance_id(struct prometheus_endpoint_instance *global_prometheus_output, char *uri, int uri_len)
{
	int i = 0;
	char instance_name_url[INSTANCE_NAME_LEN + 1] = {0};
	int  instance_name_url_len = 0;

	if(uri_len == (int)strlen(global_prometheus_output->url_path) 
		  && 0 == memcmp(uri, global_prometheus_output->url_path, strlen(global_prometheus_output->url_path)))
	{		
		return global_prometheus_output->fs_instance_cnt;
	}
	
	for(i = 0; i < global_prometheus_output->fs_instance_cnt; i++)
	{
		memset(instance_name_url, 0, sizeof(instance_name_url));
		instance_name_url_len = snprintf(instance_name_url, sizeof(instance_name_url),"/%s",global_prometheus_output->fs_instance[i]->name);

		if(uri_len == instance_name_url_len  
			  && 0 == memcmp( uri, instance_name_url, instance_name_url_len))
		{
			return i;
		}
	}
	return -1;
}

static void prometheus_output_uri_list(struct prometheus_endpoint_instance *prometheus_output,struct http_request_s* request)
{
	int i = 0;
	int payload_len = 0;
	char *payload = NULL;
	char *payload_append_position = NULL;
	struct fieldstat_instance **fs_instance = NULL;
	struct http_response_s* response = NULL;

	fs_instance = prometheus_output->fs_instance;
	

	payload_len = prometheus_output->fs_instance_cnt * 128; //TODO using marco, len?
	payload_append_position = payload = (char *)calloc(1,payload_len);
	
	payload_append_position += snprintf(payload_append_position, payload_len - (payload_append_position - payload),"url_path:\n\t%s\n", prometheus_output->url_path);

	for(i = 0; i < prometheus_output->fs_instance_cnt; i++)
	{
		payload_append_position += snprintf(payload_append_position, payload_len - (payload_append_position - payload),"\t/%s\n", fs_instance[i]->name);
	}

	response = http_response_init();
	http_response_status(response, 404);
	http_response_header(response, "Content-Type", "text/plain; charset=utf-8");
	http_response_body(response, payload, strlen(payload));
	http_respond(request, response);
	
	free(payload);
	payload=NULL;
	return;
}

static int prometheus_output_read_metric_tags(struct metric_t *metric, char *instance_name, char *tags_buff, unsigned int size)
{
	int i = 0;//used_len = 0;
	char unescape[STR_LEN_256] = {0};
	char *append_pos = tags_buff;
	
	//used_len += snprint(tags_buff, size - used_len, "app_name=\"%s\"", instance_name);

	append_pos += snprintf(append_pos, size - (append_pos - tags_buff), "app_name=\"%s\"", instance_name);

	if(metric->belong_to_table == 1)
	{
		append_pos += snprintf(append_pos, size - (append_pos - tags_buff), ",line_name=\"%s\"", metric->field_name);
	}

	for(i = 0; i < (int)metric->n_tag; i++)
	{
		memset(unescape, 0, sizeof(unescape));
		str_unescape(metric->tag_key[i], unescape, sizeof(unescape));
		append_pos += snprintf(append_pos, size - (append_pos - tags_buff), ",%s=\"%s\"", unescape, metric->tag_value[i]);
	}
	return append_pos - tags_buff;
}


static int prometheus_output_read_metric_name(struct metric_t *metric, char *instance_app_name, char *name_buff, unsigned int size)
{
	char unescape[256] = {0};
	char *append_pos = name_buff;
	if(metric->belong_to_table == 1)
	{
		str_unescape(metric->table_column_name, unescape, sizeof(unescape));
	}
	else
	{
		str_unescape(metric->field_name, unescape, sizeof(unescape));
	}
	append_pos += snprintf(append_pos, size - (append_pos - name_buff), "%s_%s", instance_app_name, unescape);

	return append_pos - name_buff;
}


static int prometheus_output_histogram_and_summary(struct metric_t *metric, char *payload, int payload_len, char *instance_app_name, char *metric_name, char *metric_tags)
{
	long long value=0;
	long long sum = 0;
	int i=0,used_len=0;

	struct hdr_iter iter;
	char *output_format = NULL;

	//struct histogram_t* h=&(p->histogram);
	struct histogram_t *h = &(metric->histogram);
	struct hdr_histogram* h_out=NULL, *h_tmp=NULL;
	int bin_num = metric->histogram.bins_num;
	double *bins = metric->histogram.bins;

	if(metric->field_type == FILED_TYPE_HISTOGRAM)
	{
		output_format = (char *)"%s_bucket{%s,le=\"%0.2f\"} %llu\n";
	}

	if(metric->field_type == FIELD_TYPE_SUMMARY)
	{
		output_format = (char *)"%s{%s,quantile=\"%0.2f\"} %llu\n";
	}

	hdr_init(h->lowest_trackable_value,  h->highest_trackable_value, h->significant_figures, &(h_tmp));

	hdr_add(h_tmp, h->accumulated);
	hdr_add(h_tmp, h->changing);
	h_out=h_tmp;

	for(i=0;i<bin_num;i++)
	{
		if(metric->field_type == FILED_TYPE_HISTOGRAM)
		{
			value = (long long)hdr_count_le_value(h_out, (long long)bins[i]);
		}
		if(metric->field_type == FIELD_TYPE_SUMMARY)
		{
			value=(long long)hdr_value_at_percentile(h_out, bins[i]);
		}
	
		used_len+=snprintf(payload+used_len,
							payload_len-used_len,
							output_format,
							metric_name,
							metric_tags,
							bins[i],
							value
					);
	}
	used_len+=snprintf(payload+used_len,
							payload_len-used_len,
							"%s_count{%s} %llu\n",
							metric_name,
							metric_tags,
							(long long)h_out->total_count
					);

	hdr_iter_recorded_init(&iter, h_out);
	while (hdr_iter_next(&iter))
	{
		sum+=(long long)iter.value;
	}
	used_len+=snprintf(payload+used_len,
			payload_len-used_len,
			"%s_sum{%s} %llu\n",
			metric_name,
			metric_tags,
			sum
			);

	hdr_close(h_tmp);
	h_tmp=NULL;

	return used_len;
}



static int prometheus_get_instance_metric_playload(struct fieldstat_instance *instance, char **payload, int *payload_size, int offset)
{
	int i = 0;
	struct metric_t *metric = NULL;
	long long value = 0;
	char metric_name[256] = {0};     //match the regex [a-zA-Z_:][a-zA-Z0-9_:]*
	char instance_name[256] = {0};
	char metric_tags[1024] = {0};    //label name match the regex [a-zA-Z_:][a-zA-Z0-9_:]* 
	int append_offset = offset;
	char *new_payload = NULL;
	int new_payload_size = 0;	

	if(instance->running != 1)
	{
		return -1;
	}

	if(payload == NULL)
	{
		return -1;
	}

	new_payload = *payload;
	new_payload_size = *payload_size;

	str_unescape(instance->name, instance_name, sizeof(instance_name));

	for(i = 0; i < instance->metric_cnt; i++)
	{
		metric = get_metric(instance, i);

		if(metric->is_ratio == 1)
		{
			continue;
		}
		if(metric->is_invisible == 1)
		{
			continue;
		}

		if( new_payload_size - append_offset <= LEFT_MIN_BUFF_LEN)
		{
			new_payload_size += REALLOC_SCALE_SIZE;
			new_payload = (char *)realloc(new_payload, new_payload_size);
		}

		prometheus_output_read_metric_name(metric, instance_name, metric_name, sizeof(metric_name));
		prometheus_output_read_metric_tags(metric, instance_name, metric_tags, sizeof(metric_tags));

		switch(metric->field_type)
		{
			case FIELD_TYPE_COUNTER:
			case FIELD_TYPE_GAUGE:
				value = get_metric_unit_val(metric, FS_CALC_CURRENT, 1);
				append_offset += snprintf(new_payload + append_offset, 
											new_payload_size - append_offset,
											"%s{%s} %llu\n", 
											metric_name, 
											metric_tags, 
											value
										);
				break;
			case FIELD_TYPE_SUMMARY:
			case FILED_TYPE_HISTOGRAM:
				append_offset += prometheus_output_histogram_and_summary(metric,new_payload + append_offset,
								new_payload_size - append_offset, instance_name, metric_name, metric_tags);
				break;
			default:
				break;
		}
	}

	*payload = new_payload;
	*payload_size = new_payload_size;

	return append_offset;

}

static void prometheus_output_instance_metric(struct prometheus_endpoint_instance *prometheus_output, struct http_request_s* request, int fs_instance_idx)
{
	int i = 0;
	int payload_size = 0;
	int payload_offset = 0;
	char *payload = NULL;
	struct fieldstat_instance *instance = NULL;
	struct http_response_s* response = NULL;

	if(fs_instance_idx == prometheus_output->fs_instance_cnt)
	{
		for(i = 0; i < prometheus_output->fs_instance_cnt; i++)
		{
			instance = prometheus_output->fs_instance[i];
			payload_offset = prometheus_get_instance_metric_playload(instance, &payload, &payload_size, payload_offset);				
		}
	}
	else
	{
		instance = prometheus_output->fs_instance[fs_instance_idx];
		payload_offset = prometheus_get_instance_metric_playload(instance, &payload, &payload_size, payload_offset);	
	}

	if(payload != NULL)
	{	
		response = http_response_init();
		http_response_status(response, 200);
		http_response_header(response, "Content-Type", "text/plain; charset=utf-8");
		http_response_body(response, payload, strlen(payload));
		http_respond(request, response);
	
		free(payload);
		payload=NULL;
	}

}


static void prometheus_endpoint_instance_output(struct http_request_s* request)
{
	int fs_instance_idx = -1;
	struct http_string_s uri;
	struct prometheus_endpoint_instance *prometheus_endpoint_instance = &g_prometheus_endpoint_instance;;

	uri = http_request_target(request);
	fs_instance_idx = prometheus_get_fs_instance_id(prometheus_endpoint_instance, (char *)uri.buf, uri.len);

	fs_instance_idx == -1 ? 
		prometheus_output_uri_list(prometheus_endpoint_instance, request) 
		:prometheus_output_instance_metric(prometheus_endpoint_instance, request, fs_instance_idx);

	return;
}


static void *prometheus_endpoint_listen_thread_entry(void *arg)
{
	struct prometheus_endpoint_instance *global_prometheus = (struct prometheus_endpoint_instance *)arg;
	http_server_listen(global_prometheus->server_handle);
	return NULL;
} 

int fieldstat_global_enable_prometheus_endpoint(unsigned short listen_port, const char *url)
{
	g_prometheus_endpoint_instance.server_handle = http_server_init(listen_port, prometheus_endpoint_instance_output);
	if(g_prometheus_endpoint_instance.server_handle == NULL)
	{
		return -1;
	}

	g_prometheus_endpoint_instance.url_path = url == NULL ? strdup(PROMETHEUS_ENDPOINT_DEFAULT_URL):strdup(url);
	g_prometheus_endpoint_instance.port = listen_port;
	g_prometheus_endpoint_instance.running = 1;
	g_prometheus_endpoint_instance.fs_instance = (struct fieldstat_instance **)calloc( sizeof(struct fieldstat_instance *), g_prometheus_endpoint_instance.fs_instance_size);

	pthread_create(&g_prometheus_endpoint_instance.tid, NULL, prometheus_endpoint_listen_thread_entry, (void *)&g_prometheus_endpoint_instance);

	return 0;
}


int fieldstat_set_prometheus_output(struct fieldstat_instance *instance)
{
	int fs_instance_id = 0;

	if(g_prometheus_endpoint_instance.running != 1)
	{
		return -1;
	}

	if(g_prometheus_endpoint_instance.fs_instance_cnt >= g_prometheus_endpoint_instance.fs_instance_size)
	{
		g_prometheus_endpoint_instance.fs_instance_size += REALLOC_SCALE_SIZE;
		g_prometheus_endpoint_instance.fs_instance = (struct fieldstat_instance **)realloc(g_prometheus_endpoint_instance.fs_instance, g_prometheus_endpoint_instance.fs_instance_size);
	}

	fs_instance_id = g_prometheus_endpoint_instance.fs_instance_cnt++;
	g_prometheus_endpoint_instance.fs_instance[fs_instance_id] = instance;
	return 0;
}



static int fieldstat_value_operate(struct fieldstat_instance *instance, int field_id, enum field_op op, long long value)
{
	struct metric_t * metric = NULL;
	struct stat_unit_t *target = NULL;

	if(field_id >= instance->metric_cnt)
	{
		return -1;
	}

	metric = get_metric(instance, field_id);

	switch(metric->field_type)
	{
		case FIELD_TYPE_COUNTER:
			target = &(metric->counter);
			break;
		case FIELD_TYPE_GAUGE:
			target = &(metric->gauge);
			break;
		case FIELD_TYPE_SUMMARY:
		case FILED_TYPE_HISTOGRAM: 
			hdr_record_value(metric->histogram.changing, (int64_t) value);
			return 0;
		default:
			break;
	}

	switch(op)
	{
		case	FS_OP_ADD:
			threadsafe_counter_add(&(target->changing), value);
			break;
		case	FS_OP_SET:
			threadsafe_counter_set(&(target->changing), value-target->accumulated);
			break;
		case	FS_OP_SUB:
			threadsafe_counter_sub(&(target->changing), value);
			break;
		default:
			assert(0);
			break;
	}

	return 0;
}


int fieldstat_value_incrby(struct fieldstat_instance *instance, int field_id, long long increment)
{
	int ret = 0;
	ret = fieldstat_value_operate(instance, field_id, FS_OP_ADD, increment);
	return ret;
}

int fieldstat_value_set(struct fieldstat_instance *instance, int field_id, long long retain)
{
	int ret = 0;
	ret = fieldstat_value_operate(instance, field_id, FS_OP_SET, retain);
	return ret;
}

int fieldstat_value_decrby(struct fieldstat_instance *instance, int field_id, long long decrment)
{
	int ret = 0;
	ret = fieldstat_value_operate(instance, field_id, FS_OP_SUB, decrment);
	return ret;
}

static int parse_histogram_bin_format(const char* format, double **output_bins, enum field_type type)
{
	char *token=NULL,*sub_token=NULL,*saveptr;
	size_t i=0;
	int comma_num=0,ret=0;
	double *bins;
	char* dup_format=__str_dup(format);
	for(i=0;i<strlen(dup_format);i++)
	{
		if(dup_format[i]==',')
		{
			comma_num++;
		}
	}
	bins=(double*)calloc(sizeof(double),comma_num+1);
	for (token = dup_format,i=0; ; token= NULL, i++)
	{
		sub_token= strtok_r(token,",", &saveptr);
		if (sub_token == NULL)
			break;
		ret=sscanf(sub_token,"%lf",bins+i);
		if(type == FIELD_TYPE_SUMMARY)
		{
			if(bins[i]>1.0)
			{
				goto error_out;
			}
		}
		if(ret!=1)
		{
			goto error_out;
		}
	}
	free(dup_format);
	*output_bins=bins;
	return i;
error_out:
	free(dup_format);
	free(bins);
	return -1;
}


static int fieldstat_register_histogram_and_summary(struct fieldstat_instance *instance, enum field_type type, const char *field_name, const char *tag_key[], const char *tag_value[], size_t n_tag, 
		const char * bins, long long lowest_trackable_value,long long highest_trackable_value,int significant_figures)
{
	struct metric_t *metric = NULL;
	struct metric_t **metric_slot = NULL;
	int metric_id = -1;
	int ret;

	if(!is_valid_field_name(field_name))
	{
		return -1;
	}
	if(n_tag > N_TAG_MAX)
	{
		return -1;
	}

	if (lowest_trackable_value < 1 || significant_figures < 1 || 5 < significant_figures)
	{
		return -1;
	}

	if (lowest_trackable_value * 2 > highest_trackable_value)
	{
		return -1;
	}

	metric_id = atomic_inc(&instance->metric_cnt) - 1;
	metric_slot = read_metric_slot(instance, metric_id);
	metric = metric_new(type,field_name,tag_key,tag_value,n_tag);
	*metric_slot = metric;

	metric->histogram.highest_trackable_value = (int64_t)highest_trackable_value;
	metric->histogram.lowest_trackable_value  = (int64_t)lowest_trackable_value;
	metric->histogram.significant_figures     = significant_figures;
	metric->histogram.bins_num = parse_histogram_bin_format((const char*)bins, &(metric->histogram.bins), type);

	ret = hdr_init((int64_t)lowest_trackable_value, (int64_t)highest_trackable_value, significant_figures, &(metric->histogram.changing));
	assert(ret==0);	
	ret = hdr_init((int64_t)lowest_trackable_value, (int64_t)highest_trackable_value, significant_figures, &(metric->histogram.accumulated));
	assert(ret==0);

	switch(type)
	{
		case FIELD_TYPE_SUMMARY:
			atomic_inc(&instance->summary_cnt);
			break;
		case FILED_TYPE_HISTOGRAM:
			atomic_inc(&instance->histogram_cnt);
			break;
		default:
			break;
	}
	return metric_id;
}

int fieldstat_register_histogram(struct fieldstat_instance *instance, const char *field_name, const char *tag_key[], const char *tag_value[], size_t n_tag, 
		const char * bins, long long lowest_trackable_value,long long highest_trackable_value,int significant_figures)
{
	return fieldstat_register_histogram_and_summary(instance, FILED_TYPE_HISTOGRAM, field_name, tag_key, tag_value, n_tag, 
		bins, lowest_trackable_value, highest_trackable_value, significant_figures);
}

int fieldstat_register_summary(struct fieldstat_instance *instance, const char *field_name, const char *tag_key[], const char *tag_value[], size_t n_tag, 
		const char * bins,long long lowest_trackable_value,long long highest_trackable_value,int significant_figures)
{
	return fieldstat_register_histogram_and_summary(instance, FIELD_TYPE_SUMMARY, field_name, tag_key, tag_value, n_tag, 
		bins, lowest_trackable_value, highest_trackable_value, significant_figures);
}