WIP

10 files changed, 277 insertions, 67 deletions

diff --git a/v3/glougloud/glougloud.c b/v3/glougloud/glougloud.c
index b1e582d..07628fb 100644
--- a/v3/glougloud/glougloud.c
+++ b/v3/glougloud/glougloud.c
@@ -40,41 +40,42 @@ main(int argc, char **argv)
 	struct glougloud *ggd;
 	int op;
 
-	ggd = xcalloc(1, struct glougloud);
+	ggd = xcalloc(1, sizeof(struct glougloud));
 
-	addr_aton("127.0.0.1", &ggd->sp_ip);
-	addr_aton("127.0.0.1", &ggd->sv_ip);
-	ggd->sp_port = GLOUGLOU_PROBE_DEFAULT_PORT;
-	ggd->sv_port = GLOUGLOU_VIZ_DEFAULT_PORT;
+	addr_aton("127.0.0.1", &ggd->probes.serv_ip);
+	addr_aton("127.0.0.1", &ggd->viz.serv_ip);
+	ggd->probes.serv_port = GLOUGLOU_PROBE_DEFAULT_PORT;
+	ggd->viz.serv_port = GLOUGLOU_VIZ_DEFAULT_PORT;
 	ggd->daemonize = 1;
 	ggd->logfile = GLOUGLOUD_LOGFILE;
 	ggd->loglevel = LOG_WARN;
+	ggd->redis.socket = "/tmp/glougloud_redis.sock";
 
 	while ((op = getopt(argc, argv, "Dhl:L:p:P:v")) != -1) {
 		switch (op) {
 			case 'D':
-				logfile = NULL;
-				daemonize = 0;
+				ggd->logfile = NULL;
+				ggd->daemonize = 0;
 				break;
 			case 'h':
 				usage();
 				/* NOTREACHED */
 			case 'l':
-				if (addr_aton(optarg, &sp_ip) < 0)
+				if (addr_aton(optarg, &ggd->probes.serv_ip) < 0)
 					err(1, "invalid probes server ip");
 				break;
 			case 'L':
-				if (addr_aton(optarg, &sv_ip) < 0)
+				if (addr_aton(optarg, &ggd->viz.serv_ip) < 0)
 					err(1, "invalid vizualisation server ip");
 				break;
 			case 'p':
-				sp_port = atoi(optarg);
+				ggd->probes.serv_port = atoi(optarg);
 				break;
 			case 'P':
-				sv_port = atoi(optarg);
+				ggd->viz.serv_port = atoi(optarg);
 				break;
 			case 'v':
-				loglevel++;
+				ggd->loglevel++;
 				break;
 			default:
 				usage();
@@ -84,7 +85,7 @@ main(int argc, char **argv)
 
 	if (geteuid() != 0)
 		errx(1, "must be root");
-	log_init(logfile, loglevel);
+	log_init(ggd->logfile, ggd->loglevel);
 	log_warn("glougloud startup");
 
 	ev_base = event_base_new();
@@ -100,15 +101,15 @@ main(int argc, char **argv)
 
 	if (redis_init(ggd) < 0)
 		log_fatal("init redis failed");
-	if (probes_init(ggd, &sp_ip, sp_port) < 0)
+	if (probes_init(ggd) < 0)
 		log_fatal("init probes failed");
-	if (viz_init(ggd, &sv_ip, sv_port) < 0)
+	if (viz_init(ggd) < 0)
 		log_fatal("init viz failed");
 
-	if (daemonize) {
-		pid = fork();
-		log_info("daemonized, pid %d", pid);
-		if (pid > 0)
+	if (ggd->daemonize) {
+		ggd->pid = fork();
+		log_info("daemonized, pid %d", ggd->pid);
+		if (ggd->pid > 0)
 			return 0;
 	}
 
diff --git a/v3/glougloud/glougloud.h b/v3/glougloud/glougloud.h
index 2fc7c0d..b588d75 100644
--- a/v3/glougloud/glougloud.h
+++ b/v3/glougloud/glougloud.h
@@ -2,40 +2,42 @@
 
 #include <dnet.h>
 
-#define GLOUGLOUD_USER_PROBES = "_glougloud_probe"
-#define GLOUGLOUD_USER_VIZ = "_glougloud_viz"
+#define GLOUGLOUD_USER_PROBES "_glougloud_probe"
+#define GLOUGLOUD_USER_VIZ "_glougloud_viz"
 #define GLOUGLOUD_LOGFILE "/var/log/glougloud.log"
 #define GLOUGLOUD_MOD_PATH "/lib/glougloud/modules/"
 
 struct glougloud {
-	struct addr sp_ip, sv_ip;
-	int sp_port, sv_port;
 	int daemonize;
 	char *logfile;
 	int loglevel;
 	int pid;
-	struct redis {
-		int pid;
-	};
-	struct probes {
-		
-	};
-	struct viz {
-		
-	};
+	struct {
+		char *socket;
+	} redis;
+	struct {
+		struct addr serv_ip;
+		int serv_port;
+	} probes;
+	struct {
+		struct addr serv_ip;
+		int serv_port;
+	} viz;
 };
 
 /* redis.c */
 
-int  redis_init(struct glougloud *ggd);
-void redis_shutdown();
+int  redis_init(struct glougloud *);
+void redis_shutdown(void);
+
+int  redis_connect(struct event_base *);
 
 /* probes.c */
 
-int  probes_init(struct glougloud *ggd, struct addr *ip, int port);
-void probes_shutdown();
+int  probes_init(struct glougloud *);
+void probes_shutdown(void);
 
 /* viz.c */
 
-int  viz_init(struct glougloud *ggd, struct addr *ip, int port);
-void viz_shutdown();
+int  viz_init(struct glougloud *);
+void viz_shutdown(void);
diff --git a/v3/glougloud/probes.c b/v3/glougloud/probes.c
index f8704b1..9853a8f 100644
--- a/v3/glougloud/probes.c
+++ b/v3/glougloud/probes.c
@@ -1,11 +1,13 @@
 #include <dnet.h>
 
+#include "glougloud.h"
+
 int
-probes_init(struct addr *ip, int port) {
+probes_init(void) {
 	return -1;
 }
 
 void
-probes_shutdown() {
+probes_shutdown(void) {
 }
 
diff --git a/v3/glougloud/redis.c b/v3/glougloud/redis.c
index e694185..4d0f430 100644
--- a/v3/glougloud/redis.c
+++ b/v3/glougloud/redis.c
@@ -2,11 +2,21 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include <string.h>
-#include <signal.h>
 #include <hiredis/hiredis.h>
 #include <hiredis/async.h>
 #include <hiredis/adapters/libevent.h>
 
+#include <libglouglou.h>
+
+#include "glougloud.h"
+
+struct glougloud_redis {
+	int pid;
+};
+
+struct glougloud *_ggd = NULL;
+struct glougloud_redis *_redis;
+
 int
 redis_init(struct glougloud *ggd) {
 	char redis_conf[4096];
@@ -15,28 +25,31 @@ redis_init(struct glougloud *ggd) {
 	char *redis_args[] = {"redis-server", "-", NULL};
 	char *redis_env[] = {NULL};
 
-	droppriv(GLOUGLOUD_USER_PROBES, 1);
-	ggd->redis.pid = fork();
-	if (ggd->redis.pid > 0)
+	_ggd = ggd;
+	_redis = xcalloc(1, sizeof(struct glougloud_redis));
+	_redis->pid = fork();
+	if (_redis->pid > 0)
 		return 0;
+	droppriv(GLOUGLOUD_USER_PROBES, 1, NULL);
 
 	snprintf(redis_conf, sizeof(redis_conf),
 		"daemonize no\n"
 		"pidfile /var/run/glougloud/redis.pid\n"
 		"port 0\n"
-		"unixsocket /tmp/glougloud_redis.sock\n"
-		"unixsocketperm 750\n",
-		"timeout 0\n",
-		"loglevel notice\n",
-		"databases 16\n",
-		"save 900 1\n",
-		"save 300 10\n",
-		"save 60 10000\n",
-		"rdbcompression yes\n",
-		"dbfilename glougloud_dump.rdb\n",
-		"dir /var/lib/glougloud/\n",
-		"slowlog-log-slower-than 10000\n",
-		"slowlog-max-len 1024\n");
+		"unixsocket %s\n"
+		"unixsocketperm 750\n"
+		"timeout 0\n"
+		"loglevel notice\n"
+		"databases 16\n"
+		"save 900 1\n"
+		"save 300 10\n"
+		"save 60 10000\n"
+		"rdbcompression yes\n"
+		"dbfilename glougloud_dump.rdb\n"
+		"dir /var/lib/glougloud/\n"
+		"slowlog-log-slower-than 10000\n"
+		"slowlog-max-len 1024\n",
+		_ggd->redis.socket);
 	exec_pipe("echo", echo_args, echo_env,
 		"redis-server", redis_args, redis_env);
 
@@ -45,5 +58,27 @@ redis_init(struct glougloud *ggd) {
 }
 
 void
-redis_shutdown() {
+redis_shutdown(void) {
+	log_debug("redis shutdown ...");
+	kill_wait(_redis->pid, 10);
+	log_info("redis shutdown ok");
+}
+
+redisContext *
+redis_connect(struct event_base *evb,
+	void (*cb_connect)(const redisAsyncContext *, int),
+	void (*cb_disconnect)(const redisAsyncContext *, int))
+{
+	redisAsyncContext *rc;
+
+	rc = redisAsyncConnectUnix(_redis->socket);
+	if (rc->err) {
+		log_warn("redis connect: %s\n", rc->errstr);
+		return -1;
+	}
+	redisLibeventAttach(rc, evb);
+	redisAsyncSetConnectCallback(c, cb_connect);
+	redisAsyncSetDisconnectCallback(c, cb_disconnect);
+
+	return rc;
 }
diff --git a/v3/glougloud/viz.c b/v3/glougloud/viz.c
index fc12ba5..f9c6844 100644
--- a/v3/glougloud/viz.c
+++ b/v3/glougloud/viz.c
@@ -1,20 +1,69 @@
 #include <dnet.h>
 
+#include <libglouglou.h>
+
 #include "glougloud.h"
 
+struct glougloud_viz {
+	int pid;
+	redisAsyncContext *rc;
+};
+
+struct glougloud *_ggd;
+struct glougloud_viz *_ggdviz;
+
+void cb_notification(redisAsyncContext *c, void *r, void *privdata)
+{
+	redisReply *reply = r;
+	if (!reply == NULL)
+		return;
+	/* XXX notify modules */
+	log_debug("redis cb_notification: %s\n", reply->str);
+}
+
+void cb_connect(const redisAsyncContext *c, int status)
+{
+	if (status != REDIS_OK) {
+		log_warn("redis error: %s", c->errstr);
+		return;
+	}
+	log_info("redis connected\n");
+}
+
+void cb_disconnect(const redisAsyncContext *c, int status)
+{
+	if (status != REDIS_OK) {
+		log_warn("redis error: %s", c->errstr);
+		return;
+	}
+	log_info("redis disconnected");
+}
+
 int
-viz_init(struct addr *ip, int port) {
+viz_init(struct glougloud *ggd) {
+	struct event_base *evb;
 	struct modules *mods;
-	droppriv(); // XXX
-	mods = modules_load(GLOUGLOUD_MOD_PATH);
-	
-	redis_notification_cb; // XXX
-	socket(); // XXX
 
-	return -1;
+	_ggd = ggd;
+	_ggdviz = xcalloc(1, sizeof(struct glougloud_viz));
+	_ggdviz->pid = fork();
+	if (_ggdviz->pid > 0)
+		return 0;
+	droppriv(GLOUGLOUD_USER_VIZ, 1, NULL);
+
+	mods = modules_load(GLOUGLOUD_MOD_PATH, NULL);
+
+	rc = redis_connect(evb, cb_connect, cb_disconnect);
+	redisAsyncCommand(c, cb_notification, "event",
+		"SUBSCRIBE __keyevent@ggd__:*");
+
+	ev = tcp_server_create(evb, _ggd->viz.serv_ip,
+		_ggd->viz.serv_port);
+
+	return 0;
 }
 
 void
-viz_shutdown() {
+viz_shutdown(void) {
 }
 
diff --git a/v3/libglouglou/Makefile b/v3/libglouglou/Makefile
index 520700e..4c2a90d 100644
--- a/v3/libglouglou/Makefile
+++ b/v3/libglouglou/Makefile
@@ -23,5 +23,10 @@ install:
 	install -m 0644 $(TARGET) $(LIBDIR)
 	install -m 0644 $(HEADERS) $(INCLUDEDIR)
 
+tests:
+	$(MAKE) -C tests run
+
 clean:
 	rm -f $(TARGET) $(OBJECTS)
+
+.PHONY: tests
diff --git a/v3/libglouglou/libglouglou.h b/v3/libglouglou/libglouglou.h
index 687a34c..034c232 100644
--- a/v3/libglouglou/libglouglou.h
+++ b/v3/libglouglou/libglouglou.h
@@ -82,5 +82,8 @@ int	 addrcmp(struct sockaddr_in *, struct sockaddr_in *);
 void	 droppriv(char *, int, char *);
 struct modules *modules_load(char *, char *);
 int	exec_pipe(char *, char **, char **, char *, char **, char **);
+void	kill_wait(pid_t, int);
+struct event *tcp_server_create(struct event_base *, struct addr *, int, event_callback_fn, void *);
+struct event *udp_server_create(struct event_base *, struct addr *, int, event_callback_fn, void *);
 
 #endif /* _LIBGLOUGLOU_H_ */
diff --git a/v3/libglouglou/tests/Makefile b/v3/libglouglou/tests/Makefile
index 75f195c..42f8109 100644
--- a/v3/libglouglou/tests/Makefile
+++ b/v3/libglouglou/tests/Makefile
@@ -1,5 +1,5 @@
 CFLAGS += -Wall -g
-LDFLAGS=-L../ -levent -ldl -lglouglou
+LDFLAGS=-L../ -levent -ldnet -ldl -lglouglou
 
 SOURCES = $(shell echo *.c)
 OBJECTS = $(SOURCES:.c=.o)
diff --git a/v3/libglouglou/tests/test_tcp_server.c b/v3/libglouglou/tests/test_tcp_server.c
new file mode 100644
index 0000000..9e1180c
--- /dev/null
+++ b/v3/libglouglou/tests/test_tcp_server.c
@@ -0,0 +1,25 @@
+#include "../libglouglou.h"
+
+void
+cb_conn(evutil_socket_t listener, short event, void *arg)
+{
+	
+}
+
+int
+main(void)
+{
+	struct event_base *evb;
+	struct event *ev;
+	struct addr ip;
+	int port;
+
+	addr_aton("127.0.0.1", &ip);
+	port = 12345;
+	evb = event_base_new();
+
+	ev = tcp_server_create(evb, &ip, port, cb_conn, NULL);
+	if (ev)
+		return 0;
+	return 1;
+}
diff --git a/v3/libglouglou/utils.c b/v3/libglouglou/utils.c
index 4d22e88..3a936ad 100644
--- a/v3/libglouglou/utils.c
+++ b/v3/libglouglou/utils.c
@@ -17,7 +17,9 @@
 #include <string.h>
 #include <dlfcn.h>
 #include <dirent.h>
+#include <signal.h>
 #include <sys/wait.h>
+#include <sys/socket.h>
 
 #include "libglouglou.h"
 
@@ -169,3 +171,89 @@ exec_pipe(char *cmd1, char **cmd1_args, char **cmd1_env,
 	perror("execve");
 	return -1;
 }
+
+void
+kill_wait(pid_t pid, int seconds) {
+	int i;
+
+	kill(pid, SIGTERM);
+	for(i=0; i<10; i++) {
+		if (waitpid(pid, NULL, WNOHANG) == -1)
+			return;
+		sleep(1);
+	}
+	log_warn("timeout SIGTERM %d, sending SIGKILL", pid);
+	kill(pid, SIGKILL);
+	sleep(1);
+}
+
+struct event *
+tcp_server_create(struct event_base *evb, struct addr *ip, int port, event_callback_fn cb_conn, void *cb_conn_data)
+{
+	int s = -1;
+	struct sockaddr_in sock_addr;
+	struct event *ev = NULL;
+	int sock_on = 1;
+
+	s = socket(AF_INET, SOCK_STREAM, 0);
+	if (s < 0)
+		goto err;
+	setsockopt(s, SOL_SOCKET, SO_REUSEADDR,
+			&sock_on, sizeof(sock_on));
+	fd_nonblock(s);
+
+	bzero(&sock_addr, sizeof(sock_addr));
+	addr_ntos(ip, (struct sockaddr *)&sock_addr);
+	sock_addr.sin_port = htons(port);
+
+	if (bind(s, (struct sockaddr *)&sock_addr,
+				sizeof(sock_addr)) != 0)
+		goto err;
+
+	ev = event_new(evb, s, EV_READ|EV_PERSIST, cb_conn, cb_conn_data);
+	event_add(ev, NULL);
+	return ev;
+
+err:
+	if (s != -1)
+		close(s);
+	if (ev)
+		event_del(ev);
+	return NULL;
+}
+
+struct event *
+udp_server_create(struct event_base *evb, struct addr *ip, int port, event_callback_fn cb_srv, void *cb_srv_data)
+{
+	int s = -1;
+	struct sockaddr_in sock_addr;
+	struct event *ev = NULL;
+	int sock_on = 1;
+
+	s = socket(AF_INET, SOCK_STREAM, 0);
+	if (s < 0)
+		goto err;
+	setsockopt(s, SOL_SOCKET, SO_REUSEADDR,
+			&sock_on, sizeof(sock_on));
+	fd_nonblock(s);
+
+	bzero(&sock_addr, sizeof(sock_addr));
+	addr_ntos(ip, (struct sockaddr *)&sock_addr);
+	sock_addr.sin_port = htons(port);
+
+	if (bind(s, (struct sockaddr *)&sock_addr,
+				sizeof(sock_addr)) != 0)
+		goto err;
+
+	ev = event_new(evb, s, EV_READ|EV_PERSIST, cb_srv, cb_srv_data);
+	event_add(ev, NULL);
+	return ev;
+
+err:
+	if (s != -1)
+		close(s);
+	if (ev)
+		event_del(ev);
+	return NULL;
+}
+