[pve-cluster.git] / data / src / server.c

/*
  Copyright (C) 2010 Proxmox Server Solutions GmbH

  This program is free software: you can redistribute it and/or modify
  it under the terms of the GNU Affero General Public License as published by
  the Free Software Foundation, either version 3 of the License, or
  (at your option) any later version.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU Affero General Public License for more details.

  You should have received a copy of the GNU Affero General Public License
  along with this program.  If not, see <http://www.gnu.org/licenses/>.

  Author: Dietmar Maurer <dietmar@proxmox.com>

*/

#define G_LOG_DOMAIN "ipcs"

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif /* HAVE_CONFIG_H */

#include <stdint.h>
#include <errno.h>
#include <string.h>
#include <sys/syslog.h>
#include <sys/uio.h>

#include <qb/qbdefs.h>
#include <qb/qbutil.h>
#include <qb/qbloop.h>
#include <qb/qbipcs.h>

#include <glib.h>

#include "cfs-utils.h"
#include "cfs-ipc-ops.h"
#include "status.h"
#include "memdb.h"
#include "logger.h"

static GThread *worker;
static qb_loop_t *loop;
static qb_ipcs_service_t* s1;
static GString *outbuf;
static memdb_t *memdb;

static int server_started = 0;   /* protect with server_started_mutex */
static int terminate_server = 0; /* protect with server_started_mutex */
static GCond server_started_cond;
static GCond server_stopped_cond;
static GMutex server_started_mutex;


typedef struct {
	struct qb_ipc_request_header req_header;
	char name[256];
} cfs_status_update_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	char name[256];
	char nodename[256];
} cfs_status_get_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	uint8_t priority;
	uint8_t ident_len;
	uint8_t tag_len;
	char data[];
} cfs_log_msg_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	uint32_t max_entries;
	uint32_t res1;
	uint32_t res2;
	uint32_t res3;
} cfs_log_get_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	uint32_t vmid;
	char property[];
} cfs_guest_config_propery_get_request_header_t;

struct s1_context {
	int32_t client_pid;
	uid_t uid;
	gid_t gid;	
	gboolean read_only;
};
 
static int32_t s1_connection_accept_fn(
	qb_ipcs_connection_t *c, 
	uid_t uid, 
	gid_t gid)
{
	if ((uid == 0 && gid == 0) || (gid == cfs.gid)) {
		cfs_debug("authenticated connection %d/%d", uid, gid);
		struct s1_context *ctx = g_new0(struct s1_context, 1);
		ctx->uid = uid;
		ctx->gid = gid;
		ctx->read_only = (gid == cfs.gid);

		struct qb_ipcs_connection_stats stats;
		qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
		ctx->client_pid = stats.client_pid;

		qb_ipcs_context_set(c, ctx);
		return 0;
	}
	cfs_critical("connection from bad user %d! - rejected", uid);
	return 1;
}

static void s1_connection_created_fn(
	qb_ipcs_connection_t *c)
{
	struct qb_ipcs_stats srv_stats;

	qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE);

	cfs_debug("Connection created > active:%d > closed:%d",
		    srv_stats.active_connections,
		    srv_stats.closed_connections);
}

static void s1_connection_destroyed_fn(
	qb_ipcs_connection_t *c)
{
	cfs_debug("connection about to be freed");
	
	gpointer ctx;
	if ((ctx = qb_ipcs_context_get(c)))
		g_free(ctx);

}

static int32_t s1_connection_closed_fn(
	qb_ipcs_connection_t *c)
{
	struct qb_ipcs_connection_stats stats;

	qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);

	cfs_debug("Connection to pid:%d destroyed", stats.client_pid);

	return 0;
}

static int32_t s1_msg_process_fn(
	qb_ipcs_connection_t *c,
	void *data,
	size_t size)
{
	struct qb_ipc_request_header *req_pt = 
		(struct qb_ipc_request_header *)data;

	struct s1_context *ctx = (struct s1_context *)qb_ipcs_context_get(c);

	if (!ctx) {
		cfs_critical("qb_ipcs_context_get failed");
		qb_ipcs_disconnect(c);
		return 0;
	}

	int32_t request_id __attribute__ ((aligned(8))) = req_pt->id;
	int32_t request_size __attribute__ ((aligned(8))) = req_pt->size;
	cfs_debug("process msg:%d, size:%d", request_id, request_size);

	char *resp = NULL;

	g_string_truncate(outbuf, 0);

	int32_t result = -ECHRNG;
	if (request_id == CFS_IPC_GET_FS_VERSION) {

		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			result = cfs_create_version_msg(outbuf);
		}

	} else if (request_id == CFS_IPC_GET_CLUSTER_INFO) {

		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			result = cfs_create_memberlist_msg(outbuf);
		}

	} else if (request_id == CFS_IPC_GET_GUEST_LIST) {
		
		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			result = cfs_create_vmlist_msg(outbuf);
		}
	} else if (request_id == CFS_IPC_SET_STATUS) {

		cfs_status_update_request_header_t *rh = 
			(cfs_status_update_request_header_t *)data;

		int datasize = request_size - sizeof(cfs_status_update_request_header_t);

		if (ctx->read_only) {
			result = -EPERM;
		} else if (datasize < 0) {
			result = -EINVAL;
		} else {	
			/* make sure name is 0 terminated */
			rh->name[sizeof(rh->name) - 1] = 0;

			char *dataptr = (char*) data + sizeof(cfs_status_update_request_header_t);

			result = cfs_status_set(rh->name, dataptr, datasize);
		}
	} else if (request_id == CFS_IPC_GET_STATUS) {

		cfs_status_get_request_header_t *rh =
			(cfs_status_get_request_header_t *)data;

		int datasize = request_size - sizeof(cfs_status_get_request_header_t);

		if (datasize < 0) {
			result = -EINVAL;
		} else {	
			/* make sure all names are 0 terminated */
			rh->name[sizeof(rh->name) - 1] = 0;
			rh->nodename[sizeof(rh->nodename) - 1] = 0;

			result = cfs_create_status_msg(outbuf, rh->nodename, rh->name);
		}
	} else if (request_id == CFS_IPC_GET_CONFIG) {

		int pathlen = request_size - sizeof(struct qb_ipc_request_header);

		if (pathlen <= 0) {
			result = -EINVAL;
		} else {
			/* make sure path is 0 terminated */
			((char *)data)[request_size] = 0;
			char *path = (char*) data + sizeof(struct qb_ipc_request_header);

			if (ctx->read_only &&  path_is_private(path)) {
				result = -EPERM;
			} else {
				gpointer tmp = NULL;
				result = memdb_read(memdb, path, &tmp);
				if (result > 0) {
					g_string_append_len(outbuf, tmp, result);
					g_free(tmp);
				}
			}
		}			
	} else if (request_id == CFS_IPC_LOG_CLUSTER_MSG) {

		cfs_log_msg_request_header_t *rh = 
			(cfs_log_msg_request_header_t *)data;

		int datasize = request_size - G_STRUCT_OFFSET(cfs_log_msg_request_header_t, data);
		int msg_len = datasize - rh->ident_len - rh->tag_len;

		if (ctx->read_only) {
			result = -EPERM;
		} else if (msg_len  < 1) {
			result = -EINVAL;
		} else {
			char *msg = rh->data;
			if ((msg[rh->ident_len - 1] == 0) &&
			    (msg[rh->ident_len + rh->tag_len - 1] == 0) &&
			    (((char *)data)[request_size] == 0)) {

				char *ident = msg;
				char *tag = msg + rh->ident_len;
				msg = msg + rh->ident_len + rh->tag_len;

				time_t ctime = time(NULL);
				clog_entry_t *entry = (clog_entry_t *)alloca(CLOG_MAX_ENTRY_SIZE);
				if (clog_pack(entry, cfs.nodename, ident, tag, ctx->client_pid,
					      ctime, rh->priority, msg)) {
					cfs_cluster_log(entry);
				}

				result = 0;

			} else {
				result = -EINVAL;
			}
		}
	} else if (request_id == CFS_IPC_GET_CLUSTER_LOG) {

		cfs_log_get_request_header_t *rh = 
			(cfs_log_get_request_header_t *)data;

		int userlen = request_size - sizeof(cfs_log_get_request_header_t);

		if (userlen <= 0) {
			result = -EINVAL;
		} else {
			/* make sure user string is 0 terminated */
			((char *)data)[request_size] = 0;
			char *user = (char*) data + sizeof(cfs_log_get_request_header_t);

			uint32_t max = rh->max_entries ?  rh->max_entries : 50;
			cfs_cluster_log_dump(outbuf, user, max);
			result = 0;
		}
	} else if (request_id == CFS_IPC_GET_RRD_DUMP) {
	
		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			cfs_rrd_dump(outbuf);
			result = 0;
		}
	} else if (request_id == CFS_IPC_GET_GUEST_CONFIG_PROPERTY) {

		cfs_guest_config_propery_get_request_header_t *rh =
			(cfs_guest_config_propery_get_request_header_t *) data;

		int proplen = request_size - G_STRUCT_OFFSET(cfs_guest_config_propery_get_request_header_t, property);

		result = 0;
		if (rh->vmid < 100 && rh->vmid != 0) {
			cfs_debug("vmid out of range %u", rh->vmid);
			result = -EINVAL;
		} else if (rh->vmid >= 100 && !vmlist_vm_exists(rh->vmid)) {
			result = -ENOENT;
		} else if (proplen <= 0) {
			cfs_debug("proplen <= 0, %d", proplen);
			result = -EINVAL;
		} else {
			((char *)data)[request_size] = 0; // ensure property is 0 terminated

			cfs_debug("cfs_get_guest_config_property: basic valid checked, do request");

			result = cfs_create_guest_conf_property_msg(outbuf, memdb, rh->property, rh->vmid);
		}
	}

	cfs_debug("process result %d", result);

	if (result >= 0) {
		resp = outbuf->str;
		result = 0;
	}

	int iov_len = 2;
	struct iovec iov[iov_len];
	struct qb_ipc_response_header res_header;

	int resp_data_len = resp ? outbuf->len : 0;

	res_header.id = request_id;
	res_header.size = sizeof(res_header) + resp_data_len;
	res_header.error = result;

	iov[0].iov_base = (char *)&res_header;
	iov[0].iov_len = sizeof(res_header);
	iov[1].iov_base = resp;
	iov[1].iov_len = resp_data_len;

	ssize_t res = qb_ipcs_response_sendv(c, iov, iov_len);
	if (res < 0) {
		cfs_critical("qb_ipcs_response_send: %s", strerror(errno));
		qb_ipcs_disconnect(c);
	}

	return 0;
}

static int32_t my_job_add(
	enum qb_loop_priority p, 
	void *data, 
	qb_loop_job_dispatch_fn fn)
{
	return qb_loop_job_add(loop, p, data, fn);
}

static int32_t my_dispatch_add(
	enum qb_loop_priority p, 
	int32_t fd, 
	int32_t evts,
	void *data, 
	qb_ipcs_dispatch_fn_t fn)
{
	return qb_loop_poll_add(loop, p, fd, evts, data, fn);
}

static int32_t my_dispatch_mod(
	enum qb_loop_priority p, 
	int32_t fd, 
	int32_t evts,
	void *data, 
	qb_ipcs_dispatch_fn_t fn)
{
	return qb_loop_poll_mod(loop, p, fd, evts, data, fn);
}

static int32_t my_dispatch_del(
	int32_t fd)
{
	return qb_loop_poll_del(loop, fd);
}

static struct qb_ipcs_service_handlers service_handlers = {
	.connection_accept = s1_connection_accept_fn,
	.connection_created = s1_connection_created_fn,
	.msg_process = s1_msg_process_fn,
	.connection_destroyed = s1_connection_destroyed_fn,
	.connection_closed = s1_connection_closed_fn,
};

static struct qb_ipcs_poll_handlers poll_handlers = {
	.job_add = my_job_add,
	.dispatch_add = my_dispatch_add,
	.dispatch_mod = my_dispatch_mod,
	.dispatch_del = my_dispatch_del,
};

static void timer_job(void *data)
{
	gboolean terminate = FALSE;

	g_mutex_lock (&server_started_mutex);

	if (terminate_server) {
		cfs_debug ("got terminate request");

		if (loop)
			qb_loop_stop (loop);
		
		if (s1) {
			qb_ipcs_destroy (s1);
			s1 = 0;
		}
		server_started = 0;

		g_cond_signal (&server_stopped_cond);
		
		terminate = TRUE;
	} else if (!server_started) {
		server_started = 1;
		g_cond_signal (&server_started_cond);
	}
	
	g_mutex_unlock (&server_started_mutex);

	if (terminate)
		return;
			       
	qb_loop_timer_handle th;
	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000000000, NULL, timer_job, &th);
}

static gpointer worker_thread(gpointer data)
{
	g_return_val_if_fail(loop != NULL, NULL);

	cfs_debug("start event loop");

	qb_ipcs_run(s1);

	qb_loop_timer_handle th;
	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000, NULL, timer_job, &th);

	qb_loop_run(loop);

	cfs_debug("event loop finished - exit worker thread");
	
	return NULL;
}

gboolean server_start(memdb_t *db)
{
	g_return_val_if_fail(loop == NULL, FALSE);
	g_return_val_if_fail(worker == NULL, FALSE);
	g_return_val_if_fail(db != NULL, FALSE);

	terminate_server = 0;
	server_started = 0;
	
	memdb = db;

	outbuf = g_string_sized_new(8192*8);

	if (!(loop = qb_loop_create())) {
		cfs_critical("cant create event loop");
		return FALSE;
	}
	
	s1 = qb_ipcs_create("pve2", 1, QB_IPC_SHM, &service_handlers);
	if (s1 == 0) {
		cfs_critical("qb_ipcs_create failed: %s", strerror(errno));
		return FALSE;
	}
	qb_ipcs_poll_handlers_set(s1, &poll_handlers);

	worker = g_thread_new ("server", worker_thread, NULL);

	g_mutex_lock (&server_started_mutex);
	while (!server_started)
		g_cond_wait (&server_started_cond, &server_started_mutex);
	g_mutex_unlock (&server_started_mutex);
	
	cfs_debug("server started");
	
	return TRUE;
}

void server_stop(void)
{
	cfs_debug("server stop");

	g_mutex_lock (&server_started_mutex);
	terminate_server = 1;
	while (server_started)
		g_cond_wait (&server_stopped_cond, &server_started_mutex);
	g_mutex_unlock (&server_started_mutex);

	if (worker) {
		g_thread_join(worker);
		worker = NULL;
	}
	
	cfs_debug("worker thread finished");

	if (loop) {
		qb_loop_destroy(loop);

		loop = NULL;
	}

	if (outbuf) {
		g_string_free(outbuf, TRUE);
		outbuf = NULL;
	}
}
Commit	Line	Data
fe000966 DM	1	/*
	2	Copyright (C) 2010 Proxmox Server Solutions GmbH
	3
	4	This program is free software: you can redistribute it and/or modify
	5	it under the terms of the GNU Affero General Public License as published by
	6	the Free Software Foundation, either version 3 of the License, or
	7	(at your option) any later version.
	8
	9	This program is distributed in the hope that it will be useful,
	10	but WITHOUT ANY WARRANTY; without even the implied warranty of
	11	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	12	GNU Affero General Public License for more details.
	13
	14	You should have received a copy of the GNU Affero General Public License
	15	along with this program. If not, see <http://www.gnu.org/licenses/>.
	16
	17	Author: Dietmar Maurer <dietmar@proxmox.com>
	18
	19	*/
	20
	21	#define G_LOG_DOMAIN "ipcs"
	22
	23	#ifdef HAVE_CONFIG_H
	24	#include <config.h>
	25	#endif /* HAVE_CONFIG_H */
	26
	27	#include <stdint.h>
	28	#include <errno.h>
	29	#include <string.h>
	30	#include <sys/syslog.h>
	31	#include <sys/uio.h>
	32
	33	#include <qb/qbdefs.h>
	34	#include <qb/qbutil.h>
	35	#include <qb/qbloop.h>
	36	#include <qb/qbipcs.h>
	37
	38	#include <glib.h>
	39
	40	#include "cfs-utils.h"
95c44445	41	#include "cfs-ipc-ops.h"
fe000966 DM	42	#include "status.h"
	43	#include "memdb.h"
	44	#include "logger.h"
	45
	46	static GThread *worker;
	47	static qb_loop_t *loop;
	48	static qb_ipcs_service_t* s1;
	49	static GString *outbuf;
	50	static memdb_t *memdb;
26754d72 DM	51
	52	static int server_started = 0; /* protect with server_started_mutex */
	53	static int terminate_server = 0; /* protect with server_started_mutex */
	54	static GCond server_started_cond;
	55	static GCond server_stopped_cond;
	56	static GMutex server_started_mutex;
	57
	58
fe000966 DM	59	typedef struct {
	60	struct qb_ipc_request_header req_header;
	61	char name[256];
	62	} cfs_status_update_request_header_t;
	63
	64	typedef struct {
	65	struct qb_ipc_request_header req_header;
	66	char name[256];
	67	char nodename[256];
	68	} cfs_status_get_request_header_t;
	69
	70	typedef struct {
	71	struct qb_ipc_request_header req_header;
	72	uint8_t priority;
	73	uint8_t ident_len;
	74	uint8_t tag_len;
	75	char data[];
	76	} cfs_log_msg_request_header_t;
	77
	78	typedef struct {
	79	struct qb_ipc_request_header req_header;
	80	uint32_t max_entries;
	81	uint32_t res1;
	82	uint32_t res2;
	83	uint32_t res3;
	84	} cfs_log_get_request_header_t;
	85
cf1b19d9 TL	86	typedef struct {
	87	struct qb_ipc_request_header req_header;
	88	uint32_t vmid;
	89	char property[];
	90	} cfs_guest_config_propery_get_request_header_t;
fe000966 DM	91
	92	struct s1_context {
	93	int32_t client_pid;
	94	uid_t uid;
	95	gid_t gid;
	96	gboolean read_only;
	97	};
	98
	99	static int32_t s1_connection_accept_fn(
	100	qb_ipcs_connection_t *c,
	101	uid_t uid,
	102	gid_t gid)
	103	{
	104	if ((uid == 0 && gid == 0) \|\| (gid == cfs.gid)) {
	105	cfs_debug("authenticated connection %d/%d", uid, gid);
	106	struct s1_context *ctx = g_new0(struct s1_context, 1);
	107	ctx->uid = uid;
	108	ctx->gid = gid;
	109	ctx->read_only = (gid == cfs.gid);
	110
	111	struct qb_ipcs_connection_stats stats;
	112	qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
	113	ctx->client_pid = stats.client_pid;
	114
	115	qb_ipcs_context_set(c, ctx);
	116	return 0;
	117	}
	118	cfs_critical("connection from bad user %d! - rejected", uid);
	119	return 1;
	120	}
	121
	122	static void s1_connection_created_fn(
	123	qb_ipcs_connection_t *c)
	124	{
	125	struct qb_ipcs_stats srv_stats;
	126
	127	qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE);
	128
	129	cfs_debug("Connection created > active:%d > closed:%d",
	130	srv_stats.active_connections,
	131	srv_stats.closed_connections);
	132	}
	133
	134	static void s1_connection_destroyed_fn(
	135	qb_ipcs_connection_t *c)
	136	{
	137	cfs_debug("connection about to be freed");
	138
	139	gpointer ctx;
	140	if ((ctx = qb_ipcs_context_get(c)))
	141	g_free(ctx);
	142
	143	}
	144
	145	static int32_t s1_connection_closed_fn(
	146	qb_ipcs_connection_t *c)
	147	{
	148	struct qb_ipcs_connection_stats stats;
	149
	150	qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
	151
	152	cfs_debug("Connection to pid:%d destroyed", stats.client_pid);
	153
	154	return 0;
155	}
156
157	static int32_t s1_msg_process_fn(
158	qb_ipcs_connection_t *c,
bd087c23	159	void *data,
fe000966 DM	160	size_t size)
	161	{
	162	struct qb_ipc_request_header *req_pt =
	163	(struct qb_ipc_request_header *)data;
	164
	165	struct s1_context ctx = (struct s1_context )qb_ipcs_context_get(c);
	166
	167	if (!ctx) {
	168	cfs_critical("qb_ipcs_context_get failed");
	169	qb_ipcs_disconnect(c);
	170	return 0;
	171	}
	172
8dbade24 TL	173	int32_t request_id __attribute__ ((aligned(8))) = req_pt->id;
	174	int32_t request_size __attribute__ ((aligned(8))) = req_pt->size;
	175	cfs_debug("process msg:%d, size:%d", request_id, request_size);
fe000966 DM	176
	177	char *resp = NULL;
	178
	179	g_string_truncate(outbuf, 0);
	180
	181	int32_t result = -ECHRNG;
8dbade24	182	if (request_id == CFS_IPC_GET_FS_VERSION) {
fe000966	183
8dbade24	184	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	185	result = -EINVAL;
	186	} else {
	187	result = cfs_create_version_msg(outbuf);
	188	}
	189
8dbade24	190	} else if (request_id == CFS_IPC_GET_CLUSTER_INFO) {
fe000966	191
8dbade24	192	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	193	result = -EINVAL;
	194	} else {
	195	result = cfs_create_memberlist_msg(outbuf);
	196	}
	197
8dbade24	198	} else if (request_id == CFS_IPC_GET_GUEST_LIST) {
fe000966	199
8dbade24	200	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	201	result = -EINVAL;
	202	} else {
	203	result = cfs_create_vmlist_msg(outbuf);
	204	}
8dbade24	205	} else if (request_id == CFS_IPC_SET_STATUS) {
fe000966 DM	206
	207	cfs_status_update_request_header_t *rh =
	208	(cfs_status_update_request_header_t *)data;
	209
8dbade24	210	int datasize = request_size - sizeof(cfs_status_update_request_header_t);
fe000966 DM	211
	212	if (ctx->read_only) {
	213	result = -EPERM;
	214	} else if (datasize < 0) {
	215	result = -EINVAL;
	216	} else {
	217	/* make sure name is 0 terminated */
	218	rh->name[sizeof(rh->name) - 1] = 0;
	219
bd087c23	220	char dataptr = (char) data + sizeof(cfs_status_update_request_header_t);
fe000966 DM	221
	222	result = cfs_status_set(rh->name, dataptr, datasize);
	223	}
8dbade24	224	} else if (request_id == CFS_IPC_GET_STATUS) {
fe000966 DM	225
	226	cfs_status_get_request_header_t *rh =
	227	(cfs_status_get_request_header_t *)data;
	228
8dbade24	229	int datasize = request_size - sizeof(cfs_status_get_request_header_t);
fe000966 DM	230
	231	if (datasize < 0) {
	232	result = -EINVAL;
	233	} else {
	234	/* make sure all names are 0 terminated */
	235	rh->name[sizeof(rh->name) - 1] = 0;
	236	rh->nodename[sizeof(rh->nodename) - 1] = 0;
	237
	238	result = cfs_create_status_msg(outbuf, rh->nodename, rh->name);
	239	}
8dbade24	240	} else if (request_id == CFS_IPC_GET_CONFIG) {
fe000966	241
8dbade24	242	int pathlen = request_size - sizeof(struct qb_ipc_request_header);
fe000966 DM	243
	244	if (pathlen <= 0) {
	245	result = -EINVAL;
	246	} else {
	247	/* make sure path is 0 terminated */
8dbade24	248	((char *)data)[request_size] = 0;
bd087c23	249	char path = (char) data + sizeof(struct qb_ipc_request_header);
fe000966 DM	250
	251	if (ctx->read_only && path_is_private(path)) {
	252	result = -EPERM;
	253	} else {
	254	gpointer tmp = NULL;
	255	result = memdb_read(memdb, path, &tmp);
4a9c52e8	256	if (result > 0) {
fe000966 DM	257	g_string_append_len(outbuf, tmp, result);
	258	g_free(tmp);
	259	}
	260	}
	261	}
8dbade24	262	} else if (request_id == CFS_IPC_LOG_CLUSTER_MSG) {
fe000966 DM	263
	264	cfs_log_msg_request_header_t *rh =
	265	(cfs_log_msg_request_header_t *)data;
	266
8dbade24	267	int datasize = request_size - G_STRUCT_OFFSET(cfs_log_msg_request_header_t, data);
fe000966 DM	268	int msg_len = datasize - rh->ident_len - rh->tag_len;
	269
	270	if (ctx->read_only) {
	271	result = -EPERM;
	272	} else if (msg_len < 1) {
	273	result = -EINVAL;
	274	} else {
	275	char *msg = rh->data;
	276	if ((msg[rh->ident_len - 1] == 0) &&
	277	(msg[rh->ident_len + rh->tag_len - 1] == 0) &&
8dbade24	278	(((char *)data)[request_size] == 0)) {
fe000966 DM	279
	280	char *ident = msg;
	281	char *tag = msg + rh->ident_len;
	282	msg = msg + rh->ident_len + rh->tag_len;
	283
	284	time_t ctime = time(NULL);
	285	clog_entry_t entry = (clog_entry_t )alloca(CLOG_MAX_ENTRY_SIZE);
	286	if (clog_pack(entry, cfs.nodename, ident, tag, ctx->client_pid,
	287	ctime, rh->priority, msg)) {
	288	cfs_cluster_log(entry);
	289	}
	290
	291	result = 0;
	292
	293	} else {
	294	result = -EINVAL;
	295	}
	296	}
8dbade24	297	} else if (request_id == CFS_IPC_GET_CLUSTER_LOG) {
fe000966 DM	298
	299	cfs_log_get_request_header_t *rh =
	300	(cfs_log_get_request_header_t *)data;
	301
8dbade24	302	int userlen = request_size - sizeof(cfs_log_get_request_header_t);
fe000966 DM	303
	304	if (userlen <= 0) {
	305	result = -EINVAL;
	306	} else {
	307	/* make sure user string is 0 terminated */
8dbade24	308	((char *)data)[request_size] = 0;
bd087c23	309	char user = (char) data + sizeof(cfs_log_get_request_header_t);
fe000966 DM	310
	311	uint32_t max = rh->max_entries ? rh->max_entries : 50;
	312	cfs_cluster_log_dump(outbuf, user, max);
	313	result = 0;
	314	}
8dbade24	315	} else if (request_id == CFS_IPC_GET_RRD_DUMP) {
fe000966	316
8dbade24	317	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	318	result = -EINVAL;
	319	} else {
	320	cfs_rrd_dump(outbuf);
	321	result = 0;
	322	}
8dbade24	323	} else if (request_id == CFS_IPC_GET_GUEST_CONFIG_PROPERTY) {
cf1b19d9 TL	324
	325	cfs_guest_config_propery_get_request_header_t *rh =
	326	(cfs_guest_config_propery_get_request_header_t *) data;
	327
8dbade24	328	int proplen = request_size - G_STRUCT_OFFSET(cfs_guest_config_propery_get_request_header_t, property);
cf1b19d9 TL	329
	330	result = 0;
	331	if (rh->vmid < 100 && rh->vmid != 0) {
	332	cfs_debug("vmid out of range %u", rh->vmid);
	333	result = -EINVAL;
	334	} else if (rh->vmid >= 100 && !vmlist_vm_exists(rh->vmid)) {
	335	result = -ENOENT;
	336	} else if (proplen <= 0) {
	337	cfs_debug("proplen <= 0, %d", proplen);
	338	result = -EINVAL;
	339	} else {
8dbade24	340	((char *)data)[request_size] = 0; // ensure property is 0 terminated
cf1b19d9 TL	341
	342	cfs_debug("cfs_get_guest_config_property: basic valid checked, do request");
	343
	344	result = cfs_create_guest_conf_property_msg(outbuf, memdb, rh->property, rh->vmid);
	345	}
fe000966 DM	346	}
	347
	348	cfs_debug("process result %d", result);
	349
	350	if (result >= 0) {
	351	resp = outbuf->str;
	352	result = 0;
	353	}
	354
	355	int iov_len = 2;
	356	struct iovec iov[iov_len];
	357	struct qb_ipc_response_header res_header;
	358
	359	int resp_data_len = resp ? outbuf->len : 0;
	360
8dbade24	361	res_header.id = request_id;
fe000966 DM	362	res_header.size = sizeof(res_header) + resp_data_len;
	363	res_header.error = result;
	364
	365	iov[0].iov_base = (char *)&res_header;
	366	iov[0].iov_len = sizeof(res_header);
	367	iov[1].iov_base = resp;
	368	iov[1].iov_len = resp_data_len;
	369
	370	ssize_t res = qb_ipcs_response_sendv(c, iov, iov_len);
	371	if (res < 0) {
	372	cfs_critical("qb_ipcs_response_send: %s", strerror(errno));
	373	qb_ipcs_disconnect(c);
	374	}
	375
	376	return 0;
	377	}
	378
	379	static int32_t my_job_add(
	380	enum qb_loop_priority p,
	381	void *data,
	382	qb_loop_job_dispatch_fn fn)
	383	{
	384	return qb_loop_job_add(loop, p, data, fn);
	385	}
	386
	387	static int32_t my_dispatch_add(
	388	enum qb_loop_priority p,
	389	int32_t fd,
	390	int32_t evts,
	391	void *data,
	392	qb_ipcs_dispatch_fn_t fn)
	393	{
	394	return qb_loop_poll_add(loop, p, fd, evts, data, fn);
	395	}
	396
	397	static int32_t my_dispatch_mod(
	398	enum qb_loop_priority p,
	399	int32_t fd,
	400	int32_t evts,
	401	void *data,
	402	qb_ipcs_dispatch_fn_t fn)
	403	{
	404	return qb_loop_poll_mod(loop, p, fd, evts, data, fn);
	405	}
	406
	407	static int32_t my_dispatch_del(
	408	int32_t fd)
	409	{
	410	return qb_loop_poll_del(loop, fd);
	411	}
	412
	413	static struct qb_ipcs_service_handlers service_handlers = {
	414	.connection_accept = s1_connection_accept_fn,
	415	.connection_created = s1_connection_created_fn,
	416	.msg_process = s1_msg_process_fn,
	417	.connection_destroyed = s1_connection_destroyed_fn,
	418	.connection_closed = s1_connection_closed_fn,
	419	};
	420
	421	static struct qb_ipcs_poll_handlers poll_handlers = {
	422	.job_add = my_job_add,
	423	.dispatch_add = my_dispatch_add,
	424	.dispatch_mod = my_dispatch_mod,
	425	.dispatch_del = my_dispatch_del,
426	};
427
428	static void timer_job(void *data)
429	{
26754d72 DM	430	gboolean terminate = FALSE;
	431
	432	g_mutex_lock (&server_started_mutex);
	433
	434	if (terminate_server) {
	435	cfs_debug ("got terminate request");
	436
	437	if (loop)
	438	qb_loop_stop (loop);
	439
	440	if (s1) {
	441	qb_ipcs_destroy (s1);
	442	s1 = 0;
	443	}
	444	server_started = 0;
	445
	446	g_cond_signal (&server_stopped_cond);
	447
	448	terminate = TRUE;
	449	} else if (!server_started) {
	450	server_started = 1;
	451	g_cond_signal (&server_started_cond);
	452	}
	453
	454	g_mutex_unlock (&server_started_mutex);
fe000966	455
26754d72 DM	456	if (terminate)
	457	return;
	458
fe000966 DM	459	qb_loop_timer_handle th;
	460	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000000000, NULL, timer_job, &th);
	461	}
	462
	463	static gpointer worker_thread(gpointer data)
	464	{
	465	g_return_val_if_fail(loop != NULL, NULL);
	466
	467	cfs_debug("start event loop");
	468
	469	qb_ipcs_run(s1);
	470
	471	qb_loop_timer_handle th;
	472	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000, NULL, timer_job, &th);
	473
	474	qb_loop_run(loop);
	475
	476	cfs_debug("event loop finished - exit worker thread");
	477
	478	return NULL;
	479	}
	480
	481	gboolean server_start(memdb_t *db)
	482	{
	483	g_return_val_if_fail(loop == NULL, FALSE);
	484	g_return_val_if_fail(worker == NULL, FALSE);
	485	g_return_val_if_fail(db != NULL, FALSE);
	486
26754d72 DM	487	terminate_server = 0;
	488	server_started = 0;
	489
fe000966 DM	490	memdb = db;
	491
	492	outbuf = g_string_sized_new(8192*8);
	493
	494	if (!(loop = qb_loop_create())) {
	495	cfs_critical("cant create event loop");
	496	return FALSE;
	497	}
	498
	499	s1 = qb_ipcs_create("pve2", 1, QB_IPC_SHM, &service_handlers);
	500	if (s1 == 0) {
	501	cfs_critical("qb_ipcs_create failed: %s", strerror(errno));
	502	return FALSE;
	503	}
	504	qb_ipcs_poll_handlers_set(s1, &poll_handlers);
26754d72	505
89fde9ac	506	worker = g_thread_new ("server", worker_thread, NULL);
fe000966	507
26754d72 DM	508	g_mutex_lock (&server_started_mutex);
	509	while (!server_started)
	510	g_cond_wait (&server_started_cond, &server_started_mutex);
	511	g_mutex_unlock (&server_started_mutex);
	512
	513	cfs_debug("server started");
	514
fe000966 DM	515	return TRUE;
	516	}
	517
	518	void server_stop(void)
	519	{
	520	cfs_debug("server stop");
	521
26754d72 DM	522	g_mutex_lock (&server_started_mutex);
	523	terminate_server = 1;
	524	while (server_started)
	525	g_cond_wait (&server_stopped_cond, &server_started_mutex);
	526	g_mutex_unlock (&server_started_mutex);
fe000966 DM	527
	528	if (worker) {
	529	g_thread_join(worker);
	530	worker = NULL;
	531	}
	532
	533	cfs_debug("worker thread finished");
	534
	535	if (loop) {
	536	qb_loop_destroy(loop);
	537
	538	loop = NULL;
	539	}
	540
	541	if (outbuf) {
	542	g_string_free(outbuf, TRUE);
	543	outbuf = NULL;
	544	}
	545	}