sammler/network.c

#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 "helper.h"
#include "list.h"
#include "logging.h"
#include "network.h"
#include "plugins.h"
#include "rrdtool.h"

#define BUFSIZE 8192

static struct sockaddr_in fwd_sa;
static int fwd_sock;

// todo: never freed..
static char *tx_buf, *rx_buf;
static int tx_pos;

int net_submit(char *hostname, char *plugin, char *filename, int ds_id, char *data)
{
	int size = snprintf(tx_buf + tx_pos, BUFSIZE - tx_pos, "%s:%s:%s:%d %s\n",
				hostname, plugin, filename, ds_id, data);

	if (size < 0 || size >= BUFSIZE - tx_pos) {
		log_print(LOG_ERROR, "net_submit(): arguments too long");
		// TODO: retry with flushed buffer?
		return -1;
	}

	tx_pos += size;
	return 0;
}

int net_submit_flush()
{
	if (tx_pos != 0)
		sendto(fwd_sock, tx_buf, tx_pos, 0, (struct sockaddr *)&fwd_sa, sizeof(fwd_sa));

	tx_pos = 0;
	return 0;
}

int net_receive(int socket)
{
	int size = recv(socket, rx_buf, BUFSIZE, 0);
	int rx_pos = 0;

	char *delim;
	while ((delim = memchr(rx_buf + rx_pos, '\n', size - rx_pos))) {
		*delim = '\0';

		char *data[2], *part[4];
		int ret = strsplit(rx_buf + rx_pos, " ", data, 2);
		if (ret != 2) {
			log_print(LOG_ERROR, "net_receive() abort data-split");
			continue;
		}

		ret = strsplit(data[0], ":", part, 4);
		if (ret != 4) {
			log_print(LOG_ERROR, "net_receive() abort header-split");
			continue;
		}
		rrd_submit(part[0], part[1], part[2], atoi(part[3]), data[1]);

		rx_pos = (delim - rx_buf) +1;
	}
	return 0;
}

int net_init_cli()
{
	int ret = config_get_saddr("global", "forward", &fwd_sa);
	if (ret < 0)
		return -1;

	tx_buf = malloc(BUFSIZE);
	if (tx_buf == NULL) {
		log_print(LOG_ERROR, "net_init_cli(): out of memory");
		return -1;
	}

	tx_pos = 0;

	fwd_sock = socket(PF_INET, SOCK_DGRAM, 0);
	if (fwd_sock < 0) {
		log_print(LOG_ERROR, "net_init_cli(): socket()");
		return -1;
	}

	log_print(LOG_INFO, "forwarding to %s:%d", inet_ntoa(fwd_sa.sin_addr), ntohs(fwd_sa.sin_port));
	return 0;
}

int net_init_srv()
{
	struct sockaddr_in sa_srv;

	int ret = config_get_saddr("global", "listen", &sa_srv);
	if (ret < 0)
		return -1;

	int srv_sock = socket(PF_INET, SOCK_DGRAM, 0);
	if (srv_sock < 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");
		close(srv_sock);
		return -1;
	}

	log_print(LOG_INFO, "listen on %s:%d", inet_ntoa(sa_srv.sin_addr), ntohs(sa_srv.sin_port));
	return srv_sock;
}