feature:提供对table row所用值操作的接口,提高dynamic metric写的性能v3.0.8

4 files changed, 212 insertions, 7 deletions

diff --git a/src/fieldstat.cpp b/src/fieldstat.cpp
index 2126d82..df773dd 100644
--- a/src/fieldstat.cpp
+++ b/src/fieldstat.cpp
@@ -116,6 +116,29 @@ void escaping_special_chars(char* str)
 
 }
 
+int escaping_special_chars_cpoy(char *dest, char *src, size_t n)
+{
+	size_t i;
+	int len = 0;
+
+	for(i = 0; i < n - 1 && src[i] != '\0'; i++)
+	{
+		if (src[i] == '.')
+		{
+			dest[i] = '_';
+		}
+		else
+		{
+			dest[i] = src[i];
+		}
+		len ++;
+	}
+	for ( ; i < n; i++)
+	{
+		dest[i] = '\0';
+	}
+	return len;
+}
 
 void get_current_table_line_cnt(struct fieldstat_instance *instance, int n_table, int *tables_line_cnt)
 {
diff --git a/src/fieldstat_dynamic.cpp b/src/fieldstat_dynamic.cpp
index 1b5cc54..59faba5 100644
--- a/src/fieldstat_dynamic.cpp
+++ b/src/fieldstat_dynamic.cpp
@@ -244,7 +244,7 @@ static int build_dynamic_metric_key(int table_id, const char *field_name, const
     /* part1: field name */
     unsigned int field_name_len = strlen(field_name) + 1;
     memcpy(out_key + used_len, field_name, field_name_len);
-    escaping_special_chars(out_key);
+    //escaping_special_chars(out_key);
     used_len += field_name_len;
 
     /* part2: table id */
@@ -505,3 +505,173 @@ long long fieldstat_dynamic_table_metric_value_get(struct fieldstat_dynamic_inst
 	value = dynamic_metric_value_read(instance, table_id, column_id, row_name, tags, n_tags, thread_id);
 	return value;
 }
+
+static struct metric **read_dynamic_row_metrics(
+    struct fieldstat_dynamic_instance *instance, 
+    char *metric_key, 
+    size_t metric_keylen,
+    int thread_id)
+{
+    struct dynamic_metric **head = NULL; 
+    struct dynamic_metric *find = NULL;
+
+    head = &instance->n_thread_dynamic_metric[thread_id];
+
+    HASH_FIND(hh, *head, metric_key, metric_keylen, find);
+    if(find == NULL)
+    {
+        return NULL;
+    }
+
+    return find->metrics;
+}
+
+static struct metric **create_dynamic_table_row_metrics(
+    struct fieldstat_dynamic_instance *instance, 
+    int table_id,
+    const char *row_name, 
+    const struct fieldstat_tag tags[], 
+    size_t n_tags,
+    int thread_id, 
+    char *metric_key,
+    unsigned metric_keylen)
+{
+	int i = 0;
+	struct dynamic_metric **head = NULL;
+	struct dynamic_metric *value = NULL;
+	struct table_metric *table = NULL;
+	struct metric *metric = NULL;
+
+	head = &instance->n_thread_dynamic_metric[thread_id];
+
+    if(!is_valid_field_name(row_name))
+    {
+        return NULL;
+    }
+    if(0 == is_valid_tags(tags, n_tags))
+    {
+        return NULL;
+    }
+
+	value = (struct dynamic_metric *)calloc(1, sizeof(struct dynamic_metric));
+
+    value->metric_keylen = metric_keylen;
+    memcpy(value->metric_key, metric_key, metric_keylen);
+
+	table = instance->table_metrics[table_id];
+	value->metrics = (struct metric **)calloc(table->column_cnt, 
+                                              sizeof(struct metric *));
+	for(i = 0; i < table->column_cnt; i ++)
+	{
+		metric = metric_new(table->column_type[i], row_name, tags, n_tags);
+
+		switch(table->column_type[i])
+		{
+			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);
+				break;
+		}
+		metric->table = table;
+		metric->table_column_id = i;
+		value->metrics[i] = metric;
+	}
+
+    HASH_ADD_KEYPTR(hh, *head, value->metric_key, value->metric_keylen, value);
+	return value->metrics;
+}
+
+static int table_row_metric_values_operate(
+    struct fieldstat_dynamic_instance *instance,
+    int table_id,
+    const char *row_name,
+    long long values[],
+    size_t n_values,
+    const struct fieldstat_tag tags[],
+    size_t n_tags,
+    int thread_id,
+    enum field_op op)
+{
+    struct metric **metrics = NULL;
+    char metric_key[512];
+    unsigned int metric_keylen = 0;
+
+    metric_keylen = build_dynamic_metric_key(table_id, row_name, tags, n_tags, 
+                                             sizeof(metric_key), metric_key);
+
+    metrics = read_dynamic_row_metrics(instance, metric_key, metric_keylen, 
+                                       thread_id);
+    if(metrics == NULL)
+    {
+        metrics = create_dynamic_table_row_metrics(instance, table_id, row_name, 
+                                                   tags, n_tags, thread_id, 
+                                                   metric_key, metric_keylen);
+    }
+    if(metrics == NULL)
+    {
+        return -1;
+    }
+    for(int i = 0; i <(int)n_values; i++)
+    {
+        metric_value_operate(metrics[i], op, values[i]);   
+    }
+
+    //metric_value_operate(metric, FS_OP_ADD, value);
+    return 0;
+}
+
+int fieldstat_dynamic_table_row_metric_values_incrby(
+    struct fieldstat_dynamic_instance *instance,
+    int table_id,
+    const char *row_name,
+    long long values[],
+    size_t n_values,
+    const struct fieldstat_tag tags[],
+    size_t n_tags,
+    int thread_id)
+{
+    int ret = 0;
+    ret = table_row_metric_values_operate(instance, table_id, row_name, 
+                                          values, n_values, tags, n_tags,
+                                          thread_id, FS_OP_ADD);
+    return ret;
+}
+
+int fieldstat_dynamic_table_row_metric_values_decrby(
+    struct fieldstat_dynamic_instance *instance,
+    int table_id,
+    const char *row_name,
+    long long values[],
+    size_t n_values,
+    const struct fieldstat_tag tags[],
+    size_t n_tags,
+    int thread_id)
+{
+    int ret = 0;
+    ret = table_row_metric_values_operate(instance, table_id, row_name, 
+                                          values, n_values, tags, n_tags,
+                                          thread_id, FS_OP_SUB);
+    return ret;
+}
+
+int fieldstat_dynamic_table_row_metric_values_set(
+    struct fieldstat_dynamic_instance *instance,
+    int table_id,
+    const char *row_name,
+    long long values[],
+    size_t n_values,
+    const struct fieldstat_tag tags[],
+    size_t n_tags,
+    int thread_id)
+{
+    int ret = 0;
+    ret = table_row_metric_values_operate(instance, table_id, row_name, 
+                                          values, n_values, tags, n_tags,
+                                          thread_id, FS_OP_SET);
+    return ret;
+}
diff --git a/src/fieldstat_internal.h b/src/fieldstat_internal.h
index 80cd3f3..08ecd2a 100644
--- a/src/fieldstat_internal.h
+++ b/src/fieldstat_internal.h
@@ -277,4 +277,6 @@ void fieldstat_global_disable_prometheus_endpoint();
 int enable_line_protocol_output(struct line_protocol_output *line_protocol_output, const char *ip, unsigned short port);
 void disable_line_protocol_output(struct line_protocol_output *line_protocol_output);
 
-void escaping_special_chars(char* str);
\ No newline at end of file
+void escaping_special_chars(char* str);
+
+int escaping_special_chars_cpoy(char *dest, char *src, size_t n);
diff --git a/src/line_protocol_output.cpp b/src/line_protocol_output.cpp
index bbc6974..af28b6e 100644
--- a/src/line_protocol_output.cpp
+++ b/src/line_protocol_output.cpp
@@ -179,10 +179,12 @@ static int add_hdr_field_set(struct metric *metric, struct hdr_histogram *h_out,
 static int add_table_row_field_set(char *column_name[], long long *row_value, int n_column, char *line_buf, int line_buf_size)
 {
 	int used_len = 0;
+	char unescape[256] = {0};
 
 	for(int i = 0; i < n_column; i++)
 	{
-		used_len += add_field_set(column_name[i], row_value[i], line_buf + used_len, line_buf_size - used_len);
+		escaping_special_chars_cpoy(unescape, column_name[i], sizeof(unescape));
+		used_len += add_field_set(unescape, row_value[i], line_buf + used_len, line_buf_size - used_len);
 		used_len += snprintf(line_buf + used_len, line_buf_size - used_len, ",");		
 	}
 
@@ -201,6 +203,7 @@ static int build_single_metric_line_buf(char *instance_name, int output_type, st
 {
 	int used_len = 0;
 	long long value = 0;
+	char unescape[256] = {0};
 
 	value = read_single_metric_value(metric, output_type);
 
@@ -209,7 +212,9 @@ static int build_single_metric_line_buf(char *instance_name, int output_type, st
 		return 0;
 	}
 
-	used_len += add_measurement(metric->field_name, line_buf, line_buf_size);
+	escaping_special_chars_cpoy(unescape, metric->field_name, sizeof(unescape));
+
+	used_len += add_measurement(unescape, line_buf, line_buf_size);
 
 	used_len += add_default_tag_set(instance_name, NULL, line_buf + used_len, line_buf_size - used_len);
 
@@ -217,7 +222,7 @@ static int build_single_metric_line_buf(char *instance_name, int output_type, st
 
 	used_len += snprintf(line_buf + used_len, line_buf_size - used_len, " ");
 
-	used_len += add_field_set(metric->field_name, value, line_buf + used_len, line_buf_size - used_len);
+	used_len += add_field_set(unescape, value, line_buf + used_len, line_buf_size - used_len);
 
 	used_len += snprintf(line_buf + used_len, line_buf_size - used_len, "\n");
 
@@ -252,6 +257,7 @@ static int build_hdr_metric_line_buf(char *instance_name, int output_type, struc
 {
 	int used_len = 0;
 	struct hdr_histogram *h_out = NULL;
+	char unescape[256] = {0};
 
 	h_out = read_hdr_metric_value(metric, output_type);
 
@@ -260,7 +266,9 @@ static int build_hdr_metric_line_buf(char *instance_name, int output_type, struc
 		return 0;
 	}
 
-	used_len += add_measurement(metric->field_name, line_buf, line_buf_size);
+	escaping_special_chars_cpoy(unescape, metric->field_name, sizeof(unescape));
+
+	used_len += add_measurement(unescape, line_buf, line_buf_size);
 
 	used_len += add_default_tag_set(instance_name, NULL, line_buf + used_len, 
 								    line_buf_size - used_len);
@@ -310,6 +318,7 @@ static int build_table_row_line_buf(char *instance_name, int output_type, struct
 	int used_len = 0;
 	struct metric *metric = NULL;
 	int n_send = 0;
+	char unescape[256] = {0};
 
 	if(table->column_cnt <= 0)
 	{
@@ -331,7 +340,8 @@ static int build_table_row_line_buf(char *instance_name, int output_type, struct
 
 	metric = row_metric[0];
 
-	used_len += add_measurement(metric->field_name, line_buf, line_buf_size);
+	escaping_special_chars_cpoy(unescape, metric->field_name, sizeof(unescape));
+	used_len += add_measurement(unescape, line_buf, line_buf_size);
 
 	used_len += add_default_tag_set(instance_name, table->name, line_buf + used_len, line_buf_size - used_len);