bugfix:将line protocol发送方式改为先收集metric然后集体发送v3.0.13

3 files changed, 49 insertions, 11 deletions

diff --git a/src/fieldstat.cpp b/src/fieldstat.cpp
index 5696345..2b4a04e 100644
--- a/src/fieldstat.cpp
+++ b/src/fieldstat.cpp
@@ -341,6 +341,24 @@ int startup_udp()
 		return sd_udp;
 	}
 
+	// uint64_t rcvbuf_size = 32 * 1024 * 1024;
+	// if (setsockopt (sd_udp, SOL_SOCKET, SO_RCVBUF, &rcvbuf_size, sizeof(rcvbuf_size) ) < 0) 
+	// {
+	// 	printf("FS2:setsockopt error: %d %s, restart socket.", errno, strerror(errno));
+	// 	close (sd_udp);
+	// 	sd_udp=-1;
+	// 	return sd_udp;
+	// }
+
+	// uint64_t sndbuf_size = 32 * 1024 * 1024;
+	// if (setsockopt (sd_udp, SOL_SOCKET, SO_RCVBUF, &sndbuf_size, sizeof(sndbuf_size) ) < 0) 
+	// {
+	// 	printf("FS2:setsockopt error: %d %s, restart socket.", errno, strerror(errno));
+	// 	close (sd_udp);
+	// 	sd_udp=-1;
+	// 	return sd_udp;
+	// }
+
 	return sd_udp;
 }
 
diff --git a/src/fieldstat_internal.h b/src/fieldstat_internal.h
index 47da22d..e471116 100644
--- a/src/fieldstat_internal.h
+++ b/src/fieldstat_internal.h
@@ -27,7 +27,6 @@
 
 #define	INIT_STAT_FIELD_NUM		1024
 #define MAX_PATH_LEN			256
-#define	UDP_PAYLOAD_SIZE		1460
 
 #define	STATUS_PER_LINE				6
 #define	FIELD_PER_LINE				8
@@ -52,7 +51,7 @@
 #define LEN_PATH_MAX		256
 #define	TABLE_MAX_NUM		64
 #define TABLE_COLUMN_SIZE	64
-#define	UDP_PAYLOAD_SIZE	1460
+#define	UDP_PAYLOAD_SIZE	512
 #define	LEFT_MIN_BUFF_LEN	1024
 #define	REALLOC_SCALE_SIZE	2048
 #define STR_LEN_32			32
diff --git a/src/line_protocol_output.cpp b/src/line_protocol_output.cpp
index b1f9389..c5ef544 100644
--- a/src/line_protocol_output.cpp
+++ b/src/line_protocol_output.cpp
@@ -1,5 +1,8 @@
 #include "fieldstat_internal.h"
+#include <string>
+#include <vector>
 
+#define  LINE_BUF_SIZE 1460
 
 static void flush_send_buf(struct line_protocol_output *line_protocol_output)
 {
@@ -21,13 +24,20 @@ static void flush_send_buf(struct line_protocol_output *line_protocol_output)
 }
 
 
-static void send_line_buf(struct line_protocol_output *line_protocol_output, char *line_buf, unsigned int line_buf_len)
+void send_line_buf(struct line_protocol_output *line_protocol_output, const char *line_buf, unsigned int line_buf_len)
 {
 	if(line_protocol_output == NULL || line_buf == NULL || line_buf_len == 0)
 	{
 		return;
 	}
 
+	if(line_buf_len > UDP_PAYLOAD_SIZE - 1)
+	{
+		send_udp(line_protocol_output->send_socket, line_protocol_output->server_ip,
+				line_protocol_output->server_port, line_buf, line_buf_len);
+		return;
+	}
+
 	if(UDP_PAYLOAD_SIZE - line_protocol_output->send_buf_offset - 1 < line_buf_len)
 	{
 		flush_send_buf(line_protocol_output);
@@ -363,7 +373,7 @@ static void output_line_protocol_single_metric(struct fieldstat_instance *instan
 	int used_len = 0;
 
 	struct metric *metric = NULL;
-	char line_buf[UDP_PAYLOAD_SIZE];
+	char line_buf[LINE_BUF_SIZE];
 
 	for(i = 0; i < n_cur_metric; i++)
 	{
@@ -406,7 +416,7 @@ static void output_line_protocol_table_row(struct fieldstat_instance *instance,
 	int i = 0, j = 0, k = 0;
 	int used_len = 0;
 
-	char line_buf[UDP_PAYLOAD_SIZE];
+	char line_buf[LINE_BUF_SIZE];
 	struct table_metric *table = NULL;
 	struct table_line *row = NULL;
 	struct metric *row_metrics[TABLE_COLUMN_SIZE];
@@ -465,32 +475,43 @@ int line_protocol_dynamic_metric_output(struct fieldstat_dynamic_instance *insta
 	struct metric **metrics = NULL;
 	struct metric *metric = NULL;
 
-	char line_buf[UDP_PAYLOAD_SIZE];
-	int used_len = 0;
+	//std::string line_buf_to_send;
+	//std::vector<std::string> line_buf_to_send;
 
 	for(int i = 0; i < instance->n_thread; i++)
 	{
+		std::vector<std::string> line_buf_to_send;
 		pthread_spin_lock(instance->uthash_locks + i);
 		head = &instance->n_thread_dynamic_metric[i];
 		HASH_ITER(hh, *head, dyn_metric, tmp_dyn_metric)
 		{
+			char line_buf[LINE_BUF_SIZE] = "\0";
+
 			metrics = dyn_metric->metrics;
 			metric = metrics[0];
-			used_len = 0;
 			memset(line_buf, 0, sizeof(line_buf));
 			if(metric->table)
 			{
-				used_len = build_table_row_line_buf(instance->name, instance->output_type, metric->table, metrics, line_buf, sizeof(line_buf));
+				build_table_row_line_buf(instance->name, instance->output_type, metric->table, metrics, line_buf, sizeof(line_buf));
 			}
 			else
 			{
-				used_len = build_single_metric_line_buf(instance->name, instance->output_type, metric, line_buf, sizeof(line_buf));
+				build_single_metric_line_buf(instance->name, instance->output_type, metric, line_buf, sizeof(line_buf));
 			}
-			send_line_buf(&instance->line_protocol_output, line_buf, used_len);
+
+			/* copy the line_buf as str to vector line_buf_to_send */
+			line_buf_to_send.push_back(std::string(line_buf));  
 		}
 		pthread_spin_unlock(instance->uthash_locks + i);
 
+		for (auto it = line_buf_to_send.begin(); it != line_buf_to_send.end(); ++it)
+		{
+			const std::string& str = *it;
+			send_line_buf(&instance->line_protocol_output, str.c_str(), str.length());
+		}
 	}
+
 	flush_send_buf(&instance->line_protocol_output);
+
 	return 0;
 }