forwarding works

network.c

@@ -1,49 +1,182 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <unistd.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/ip.h>
+#include <arpa/inet.h>
 
 #include "configfile.h"
+#include "list.h"
 #include "logging.h"
+#include "network.h"
 #include "plugins.h"
+#include "rrdtool.h"
 
-#define MAX_SIZE	8192
+#define BUFSIZE 1024
 
-static char *pktbuf = NULL;
+static LIST_HEAD(fwd_list);
-static int pos = 0;
 
-void net_submit(char *hostname, struct sammler_plugin *plugin, char *filename, int ds_id, char *data)
+struct fwd_entry {
+	struct list_head list;
+	struct sockaddr_in sa;
+	int sock;
+};
+
+// todo: never freed..
+static char *tx_buf, *rx_buf;
+
+void net_submit(char *hostname, char *plugin, char *filename, int ds_id, char *data)
 {
-	int len = 0;
+	int size = snprintf(tx_buf, BUFSIZE, "%s:%s:%s:%d %s", hostname, plugin, filename, ds_id, data);
-
+	if (size < 0 || size >= BUFSIZE) {
-	if (pktbuf == NULL) {
+		log_print(LOG_ERROR, "net_submit(): arguments too long");
-		pktbuf = malloc(MAX_SIZE);
+		return;
-		pos = 0;
 	}
 
-	if (pos == 0) {
+	struct fwd_entry *entry;
-		len = snprintf(pktbuf, MAX_SIZE - pos, "%s\n", hostname);
+	list_for_each_entry(entry, &fwd_list, list)
-		if (len < 0 || len >= MAX_SIZE - pos)
+		sendto(entry->sock, tx_buf, size +1, 0, (struct sockaddr *)&entry->sa, sizeof(entry->sa));
-			return;
 
-		pos += len;
+//	log_print(LOG_DEBUG, "net_submit: %s", tx_buf);
+}
+
+int net_receive(int socket)
+{
+	char *hostname, *plugin, *filename, *ds_id, *data;
+	int size, num;
+
+	size = recv(socket, rx_buf, BUFSIZE, 0);
+
+	hostname = rx_buf;
+	plugin = strchr(rx_buf, ':');
+	if (plugin == NULL)
+		return 0;
+
+	*plugin++ = '\0';
+	filename = strchr(plugin, ':');
+	if (filename == NULL)
+		return 0;
+
+	*filename++ = '\0';
+	ds_id = strchr(filename, ':');
+	if (ds_id == NULL)
+		return 0;
+
+	*ds_id++ = '\0';
+
+	data = strchr(ds_id, ' ');
+	if (data == NULL)
+		return 0;
+
+	*data++ = '\0';
+
+//	log_print(LOG_DEBUG, "net_receive: %s|%s|%s|%d|%s", hostname, plugin, filename, atoi(ds_id), data);
+	rrd_submit(hostname, plugin, filename, atoi(ds_id), data);
+
+	return 0;
+}
+
+static int parse_address(char *addr, struct sockaddr_in *sa)
+{
+	if (addr == NULL)
+		return -1;
+
+	char *port = strchr(addr, ':');
+	if (port == NULL)
+		return -1;
+
+	*port = '\0';
+
+	sa->sin_family = AF_INET;
+	sa->sin_port = htons(atoi(port +1));
+	inet_aton(addr, &sa->sin_addr);
+
+	*port = ':';
+
+	return 0;
+}
+
+static int net_add_cli(char *addr)
+{
+	struct fwd_entry *entry;
+
+	entry = malloc(sizeof(struct fwd_entry));
+	if (entry == NULL) {
+		log_print(LOG_ERROR, "net_add_cli(): out of memory");
+		return -1;
 	}
 
-	len = snprintf(pktbuf + pos, MAX_SIZE - pos, "%s:%s:%d %s\n",
+	if (parse_address(addr, &entry->sa) == -1) {
-			plugin->name, filename, ds_id, data);
+		free(entry);
+		return -1;
+	}
 
-	if (len < 0 || len >= MAX_SIZE - pos)
+	if ((entry->sock = socket(PF_INET, SOCK_DGRAM, 0)) < 0) {
-		return;
+		log_print(LOG_ERROR, "net_add_cli(): socket()");
+		free(entry);
+		return -1;
+	}
 
-	pos += len;
+	log_print(LOG_INFO, "forwarding to %s:%d", inet_ntoa(entry->sa.sin_addr), ntohs(entry->sa.sin_port));
-	return;
+	list_add_tail(&entry->list, &fwd_list);
+	return 0;
 }
 
-void net_commit()
+int net_init_cli()
 {
-	if (pos == 0)
+	struct conf_section *section;
-		return;
+	struct conf_tupel *tupel;
 
-	log_print(LOG_ERROR, "%s", pktbuf);
+	section = config_get_section("global");
+	if (section == NULL)
+		return 0;
 
-	pos = 0;
+	int retval = 0;
+	list_for_each_entry(tupel, &section->tupel, list) {
+		if (!strcmp(tupel->option, "forward"))
+			retval |= (net_add_cli(tupel->parameter) != -1);
+	}
+
+	if (retval) {
+		tx_buf = malloc(BUFSIZE);
+		if (tx_buf == NULL) {
+			log_print(LOG_ERROR, "net_init_cli(): out of memory");
+			return 0;
+		}
+	}
+
+	return retval;
+}
+
+int net_init_srv()
+{
+	struct sockaddr_in sa_srv;
+	int srv_sock;
+
+	char *listen = config_get_string("global", "listen", NULL);
+	if (parse_address(listen, &sa_srv) == -1)
+		return -1;
+
+	if ((srv_sock = socket(PF_INET, SOCK_DGRAM, 0)) < 0) {
+		log_print(LOG_ERROR, "net_init_src(): socket()");
+		return -1;
+	}
+
+	if (bind(srv_sock, (struct sockaddr *)&sa_srv, sizeof(sa_srv)) < 0) {
+		log_print(LOG_ERROR, "net_init_src(): bind()");
+		close(srv_sock);
+		return -1;
+	}
+
+	rx_buf = malloc(BUFSIZE);
+	if (rx_buf == NULL) {
+		log_print(LOG_ERROR, "net_init_srv(): out of memory");
+		return 0;
+	}
+
+	log_print(LOG_INFO, "listen on %s:%d", inet_ntoa(sa_srv.sin_addr), ntohs(sa_srv.sin_port));
+
+	return srv_sock;
 }

network.h

@@ -1,7 +1,12 @@
 #ifndef _NETWORK_H_
 #define _NETWORK_H_
 
-void net_submit(char *hostname, struct sammler_plugin *plugin, char *filename, int ds_id, char *data);
+#include "plugins.h"
-void net_commit();
+
+int net_init_srv();
+int net_init_cli();
+int net_receive(int sock);
+
+void net_submit(char *hostname, char *plugin, char *filename, int ds_id, char *data);
 
 #endif /* _NETWORK_H_ */

plugins.c

@@ -33,10 +33,12 @@
 #include "network.h"
 #include "rrdtool.h"
 
-#define BUFSIZE 1024
+#define BUFSIZE 512
 
 static LIST_HEAD(plugin_list);
 
+static int plugin_flags;
+
 static void plugin_load(char *filename)
 {
 	struct sammler_plugin *plugin = NULL;
@@ -87,7 +89,7 @@ static void plugin_load(char *filename)
 	return;
 }
 
-void plugin_load_all()
+void plugin_init(int flags)
 {
 	struct conf_section *section;
 	struct conf_tupel *tupel;
@@ -99,6 +101,8 @@ void plugin_load_all()
 	list_for_each_entry(tupel, &section->tupel, list)
 		if (!strcmp(tupel->option, "plugin"))
 			plugin_load(tupel->parameter);
+
+	plugin_flags = flags;
 }
 
 void plugins_probe(void)
@@ -114,8 +118,17 @@ void plugins_probe(void)
 			plugin->lastprobe = now;
 		}
 	}
+}
 
-	net_commit();
+struct sammler_plugin * plugin_lookup(char *name)
+{
+	struct sammler_plugin *plugin;
+	list_for_each_entry(plugin, &plugin_list, list) {
+		if (!strcmp(plugin->name, name))
+			return plugin;
+	}
+
+	return NULL;
 }
 
 void probe_submit(struct sammler_plugin *plugin, char *filename, int ds_id, const char *fmt, ... )
@@ -145,8 +158,11 @@ void probe_submit(struct sammler_plugin *plugin, char *filename, int ds_id, cons
 		return;
 	}
 
-	rrd_submit(hostname, plugin, filename, ds_id, buffer);
+	if (plugin_flags & PLUGIN_RRD)
-	net_submit(hostname, plugin, filename, ds_id, buffer);
+		rrd_submit(hostname, plugin->name, filename, ds_id, buffer);
+
+	if (plugin_flags & PLUGIN_NET)
+		net_submit(hostname, plugin->name, filename, ds_id, buffer);
 
 	free(buffer);
 }

plugins.h

@@ -4,6 +4,9 @@
 #include "list.h"
 #include "logging.h"
 
+#define PLUGIN_RRD	0x01
+#define PLUGIN_NET	0x02
+
 struct sammler_plugin {
 	struct list_head list;
 	char *name;
@@ -13,11 +16,11 @@ struct sammler_plugin {
 	char * (*get_ds) (int ds_id);
 };
 
-void plugin_load_all(void);
+void plugin_init(int flags);
 
 void plugins_probe(void);
 
-char ** plugins_get_ds(char *plugin, int ds_id);
+struct sammler_plugin * plugin_lookup(char *name);
 
 void probe_submit(struct sammler_plugin *plugin, char *filename, int ds_id, const char *fmt, ... );
 

rrdtool.c

@@ -76,6 +76,8 @@ static int do_rrd(int mode, char *cmd)
 	int argc;
 	char *argv[ARGCMAX];
 
+//	log_print(LOG_DEBUG, "do_rrd: %s", cmd);
+
 	argc = strsplit(cmd, argv, ARGCMAX -1);
 	argv[argc] = NULL;
 
@@ -86,7 +88,7 @@ static int do_rrd(int mode, char *cmd)
 		if (rrd_create(argc, argv) == -1) {
 			errno = 0;
 			log_print(LOG_ERROR, "rrd_create failed: %s: %s",
-					argv[2], rrd_get_error());
+					argv[1], rrd_get_error());
 			return -1;
 		}
 
@@ -94,18 +96,25 @@ static int do_rrd(int mode, char *cmd)
 		if (rrd_update(argc, argv) == -1) {
 			errno = 0;
 			log_print(LOG_ERROR, "rrd_update failed: %s: %s",
-					argv[2], rrd_get_error());
+					argv[1], rrd_get_error());
 			return -1;
 		}
 	}
 	return 0;
 }
 
-static int rrd_create_file(char *filename, struct sammler_plugin *plugin, int ds_id)
+static int rrd_create_file(char *filename, char *plugin_name, int ds_id)
 {
+	struct sammler_plugin *plugin;
 	int pos, step, retval;
 	char *ds_def, *buffer;
 
+	plugin = plugin_lookup(plugin_name);
+	if (plugin == NULL) {
+		log_print(LOG_ERROR, "Plugin not found (%s)", plugin_name);
+		return -1;
+	}
+
 	ds_def = plugin->get_ds(ds_id);
 	if (ds_def == NULL) {
 		log_print(LOG_ERROR, "No vaild DS found (%s:%d)", plugin->name, ds_id);
@@ -219,7 +228,7 @@ static int create_parent_dirs(char *filename)
 	return 0;
 }
 
-void rrd_submit(char *hostname, struct sammler_plugin *plugin, char *filename, int ds_id, char *data)
+void rrd_submit(char *hostname, char *plugin, char *filename, int ds_id, char *data)
 {
 	struct stat statbuf;
 	static char *rrd_dir = NULL;

rrdtool.h

@@ -1,6 +1,6 @@
 #ifndef _RRDTOOL_H_
 #define _RRDTOOL_H_
 
-void rrd_submit(char *hostname, struct sammler_plugin *plugin, char *filename, int ds_id, char *data);
+void rrd_submit(char *hostname, char *plugin, char *filename, int ds_id, char *data);
 
 #endif /* _RRDTOOL_H_ */

sammler.c

@@ -29,6 +29,7 @@
 
 #include "configfile.h"
 #include "logging.h"
+#include "network.h"
 #include "plugins.h"
 
 #define DEFAULT_CONFIG "sammler.conf"
@@ -95,19 +96,31 @@ int main(int argc, char *argv[])
 	char *hostname = config_get_string("global", "hostname", "localhost");
 	log_print(LOG_EVERYTIME, "sammler (pid:%d) started on host '%s'", getpid(), hostname);
 
-	plugin_load_all();
-
 	fd_set fdsel, fdcpy;
 	FD_ZERO(&fdsel);
 
+	int srv_sock = net_init_srv();
+	if (srv_sock != -1)
+		FD_SET(srv_sock, &fdsel);
+
+	int probe_flags = 0;
+	if (net_init_cli())
+		probe_flags |= PLUGIN_NET;
+
+	char *fwd_only = config_get_string("global", "forward_only", NULL);
+	if (fwd_only == NULL || strncmp(fwd_only, "true", 4))
+		probe_flags |= PLUGIN_RRD;
+
+	plugin_init(probe_flags);
+
 	struct timeval tv;
-	tv.tv_sec = 1;
+	tv.tv_sec = 0;
 	tv.tv_usec = 0;
 
 	while (1) {
 		memcpy(&fdcpy, &fdsel, sizeof(fdsel));
 
-		int i = select(64, &fdcpy, NULL, NULL, &tv);
+		int i = select(FD_SETSIZE, &fdcpy, NULL, NULL, &tv);
 		if (i == -1) {
 			log_print(LOG_ERROR, "select()");
 
@@ -116,6 +129,9 @@ int main(int argc, char *argv[])
 			plugins_probe();
 			tv.tv_sec = 1;
 			tv.tv_usec = 0;
+
+		} else if (FD_ISSET(srv_sock, &fdsel)) {
+			net_receive(srv_sock);
 		}
 	}
 

sammler.conf

@@ -1,6 +1,10 @@
 [global]
 hostname localhost
 
+#listen 127.0.0.1:5000
+#forward 127.0.0.1:5000
+#forward_only true
+
 logfile sammler.log
 
 rrd_dir rrd