[mirror_frr.git] / lib / zlog_live.c

// SPDX-License-Identifier: ISC
/*
 * Copyright (c) 2019-22  David Lamparter, for NetDEF, Inc.
 */

#include "zebra.h"

#include "zlog_live.h"

#include "memory.h"
#include "frrcu.h"
#include "zlog.h"
#include "printfrr.h"
#include "network.h"

DEFINE_MTYPE_STATIC(LOG, LOG_LIVE, "log vtysh live target");

enum {
	STATE_NORMAL = 0,
	STATE_FD_DEAD,
	STATE_DISOWNED,
};

struct zlt_live {
	struct zlog_target zt;

	atomic_uint_fast32_t fd;
	struct rcu_head_close head_close;
	struct rcu_head head_self;

	atomic_uint_fast32_t state;
	atomic_uint_fast32_t lost_msgs;
};

static void zlog_live(struct zlog_target *zt, struct zlog_msg *msgs[],
		      size_t nmsgs)
{
	struct zlt_live *zte = container_of(zt, struct zlt_live, zt);
	struct zlog_live_hdr hdrs[nmsgs], *hdr = hdrs;
	struct mmsghdr mmhs[nmsgs], *mmh = mmhs;
	struct iovec iovs[nmsgs * 3], *iov = iovs;
	struct timespec ts;
	size_t i, textlen;
	int fd;
	uint_fast32_t state;

	fd = atomic_load_explicit(&zte->fd, memory_order_relaxed);

	if (fd < 0)
		return;

	memset(mmhs, 0, sizeof(mmhs));
	memset(hdrs, 0, sizeof(hdrs));

	for (i = 0; i < nmsgs; i++) {
		const struct fmt_outpos *argpos;
		size_t n_argpos, texthdrlen;
		struct zlog_msg *msg = msgs[i];
		int prio = zlog_msg_prio(msg);
		const struct xref_logmsg *xref;
		intmax_t pid, tid;

		if (prio > zt->prio_min)
			continue;

		zlog_msg_args(msg, &texthdrlen, &n_argpos, &argpos);

		mmh->msg_hdr.msg_iov = iov;

		iov->iov_base = hdr;
		iov->iov_len = sizeof(*hdr);
		iov++;

		if (n_argpos) {
			iov->iov_base = (char *)argpos;
			iov->iov_len = sizeof(*argpos) * n_argpos;
			iov++;
		}

		iov->iov_base = (char *)zlog_msg_text(msg, &textlen);
		iov->iov_len = textlen;
		iov++;

		zlog_msg_tsraw(msg, &ts);
		zlog_msg_pid(msg, &pid, &tid);
		xref = zlog_msg_xref(msg);

		hdr->ts_sec = ts.tv_sec;
		hdr->ts_nsec = ts.tv_nsec;
		hdr->pid = pid;
		hdr->tid = tid;
		hdr->lost_msgs = atomic_load_explicit(&zte->lost_msgs,
						      memory_order_relaxed);
		hdr->prio = prio;
		hdr->flags = 0;
		hdr->textlen = textlen;
		hdr->texthdrlen = texthdrlen;
		hdr->n_argpos = n_argpos;
		if (xref) {
			memcpy(hdr->uid, xref->xref.xrefdata->uid,
			       sizeof(hdr->uid));
			hdr->ec = xref->ec;
		} else {
			memset(hdr->uid, 0, sizeof(hdr->uid));
			hdr->ec = 0;
		}
		hdr->hdrlen = sizeof(*hdr) + sizeof(*argpos) * n_argpos;

		mmh->msg_hdr.msg_iovlen = iov - mmh->msg_hdr.msg_iov;
		mmh++;
		hdr++;
	}

	size_t msgtotal = mmh - mmhs;
	ssize_t sent;

	for (size_t msgpos = 0; msgpos < msgtotal; msgpos += sent) {
		sent = sendmmsg(fd, mmhs + msgpos, msgtotal - msgpos, 0);

		if (sent <= 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
			atomic_fetch_add_explicit(&zte->lost_msgs,
						  msgtotal - msgpos,
						  memory_order_relaxed);
			break;
		}
		if (sent <= 0)
			goto out_err;
	}
	return;

out_err:
	fd = atomic_exchange_explicit(&zte->fd, -1, memory_order_relaxed);
	if (fd < 0)
		return;

	rcu_close(&zte->head_close, fd);
	zlog_target_replace(zt, NULL);

	state = STATE_NORMAL;
	atomic_compare_exchange_strong_explicit(
		&zte->state, &state, STATE_FD_DEAD, memory_order_relaxed,
		memory_order_relaxed);
	if (state == STATE_DISOWNED)
		rcu_free(MTYPE_LOG_LIVE, zte, head_self);
}

static void zlog_live_sigsafe(struct zlog_target *zt, const char *text,
			      size_t len)
{
	struct zlt_live *zte = container_of(zt, struct zlt_live, zt);
	struct zlog_live_hdr hdr[1] = {};
	struct iovec iovs[2], *iov = iovs;
	struct timespec ts;
	int fd;

	fd = atomic_load_explicit(&zte->fd, memory_order_relaxed);
	if (fd < 0)
		return;

	clock_gettime(CLOCK_REALTIME, &ts);

	hdr->ts_sec = ts.tv_sec;
	hdr->ts_nsec = ts.tv_nsec;
	hdr->prio = LOG_CRIT;
	hdr->textlen = len;

	iov->iov_base = (char *)hdr;
	iov->iov_len = sizeof(hdr);
	iov++;

	iov->iov_base = (char *)text;
	iov->iov_len = len;
	iov++;

	writev(fd, iovs, iov - iovs);
}

void zlog_live_open(struct zlog_live_cfg *cfg, int prio_min, int *other_fd)
{
	int sockets[2];

	if (cfg->target)
		zlog_live_close(cfg);

	*other_fd = -1;
	if (prio_min == ZLOG_DISABLED)
		return;

	/* the only reason for SEQPACKET here is getting close notifications.
	 * otherwise if you open a bunch of vtysh connections with live logs
	 * and close them all, the fds will stick around until we get an error
	 * when trying to log something to them at some later point -- which
	 * eats up fds and might be *much* later for some daemons.
	 */
	if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, sockets) < 0) {
		if (socketpair(AF_UNIX, SOCK_DGRAM, 0, sockets) < 0) {
			zlog_warn("%% could not open socket pair: %m");
			return;
		}
	} else
		/* SEQPACKET only: try to zap read direction */
		shutdown(sockets[0], SHUT_RD);

	*other_fd = sockets[1];
	zlog_live_open_fd(cfg, prio_min, sockets[0]);
}

void zlog_live_open_fd(struct zlog_live_cfg *cfg, int prio_min, int fd)
{
	struct zlt_live *zte;
	struct zlog_target *zt;

	if (cfg->target)
		zlog_live_close(cfg);

	zt = zlog_target_clone(MTYPE_LOG_LIVE, NULL, sizeof(*zte));
	zte = container_of(zt, struct zlt_live, zt);
	cfg->target = zte;

	set_nonblocking(fd);
	zte->fd = fd;
	zte->zt.prio_min = prio_min;
	zte->zt.logfn = zlog_live;
	zte->zt.logfn_sigsafe = zlog_live_sigsafe;

	zlog_target_replace(NULL, zt);
}

void zlog_live_close(struct zlog_live_cfg *cfg)
{
	struct zlt_live *zte;
	int fd;

	if (!cfg->target)
		return;

	zte = cfg->target;
	cfg->target = NULL;

	fd = atomic_exchange_explicit(&zte->fd, -1, memory_order_relaxed);

	if (fd >= 0) {
		rcu_close(&zte->head_close, fd);
		zlog_target_replace(&zte->zt, NULL);
	}
	rcu_free(MTYPE_LOG_LIVE, zte, head_self);
}

void zlog_live_disown(struct zlog_live_cfg *cfg)
{
	struct zlt_live *zte;
	uint_fast32_t state;

	if (!cfg->target)
		return;

	zte = cfg->target;
	cfg->target = NULL;

	state = STATE_NORMAL;
	atomic_compare_exchange_strong_explicit(
		&zte->state, &state, STATE_DISOWNED, memory_order_relaxed,
		memory_order_relaxed);
	if (state == STATE_FD_DEAD)
		rcu_free(MTYPE_LOG_LIVE, zte, head_self);
}
Commit	Line	Data
acddc0ed	1	// SPDX-License-Identifier: ISC
0798d276 DL	2	/*
0798d276 DL	3	* Copyright (c) 2019-22 David Lamparter, for NetDEF, Inc.
0798d276 DL	4	*/
	5
	6	#include "zebra.h"
	7
	8	#include "zlog_live.h"
	9
	10	#include "memory.h"
	11	#include "frrcu.h"
	12	#include "zlog.h"
	13	#include "printfrr.h"
2eda953a	14	#include "network.h"
0798d276 DL	15
	16	DEFINE_MTYPE_STATIC(LOG, LOG_LIVE, "log vtysh live target");
	17
	18	enum {
	19	STATE_NORMAL = 0,
	20	STATE_FD_DEAD,
	21	STATE_DISOWNED,
	22	};
	23
	24	struct zlt_live {
	25	struct zlog_target zt;
	26
	27	atomic_uint_fast32_t fd;
	28	struct rcu_head_close head_close;
	29	struct rcu_head head_self;
	30
	31	atomic_uint_fast32_t state;
a4af82ee	32	atomic_uint_fast32_t lost_msgs;
0798d276 DL	33	};
	34
	35	static void zlog_live(struct zlog_target zt, struct zlog_msg msgs[],
	36	size_t nmsgs)
	37	{
	38	struct zlt_live *zte = container_of(zt, struct zlt_live, zt);
	39	struct zlog_live_hdr hdrs[nmsgs], *hdr = hdrs;
	40	struct mmsghdr mmhs[nmsgs], *mmh = mmhs;
	41	struct iovec iovs[nmsgs * 3], *iov = iovs;
	42	struct timespec ts;
	43	size_t i, textlen;
	44	int fd;
	45	uint_fast32_t state;
	46
	47	fd = atomic_load_explicit(&zte->fd, memory_order_relaxed);
	48
	49	if (fd < 0)
	50	return;
	51
	52	memset(mmhs, 0, sizeof(mmhs));
	53	memset(hdrs, 0, sizeof(hdrs));
	54
	55	for (i = 0; i < nmsgs; i++) {
	56	const struct fmt_outpos *argpos;
834585bd	57	size_t n_argpos, texthdrlen;
0798d276 DL	58	struct zlog_msg *msg = msgs[i];
0798d276 DL	59	int prio = zlog_msg_prio(msg);
834585bd DL	60	const struct xref_logmsg *xref;
834585bd DL	61	intmax_t pid, tid;
0798d276 DL	62
	63	if (prio > zt->prio_min)
	64	continue;
	65
834585bd	66	zlog_msg_args(msg, &texthdrlen, &n_argpos, &argpos);
0798d276 DL	67
	68	mmh->msg_hdr.msg_iov = iov;
	69
	70	iov->iov_base = hdr;
	71	iov->iov_len = sizeof(*hdr);
	72	iov++;
	73
	74	if (n_argpos) {
	75	iov->iov_base = (char *)argpos;
	76	iov->iov_len = sizeof(argpos) n_argpos;
	77	iov++;
	78	}
	79
	80	iov->iov_base = (char *)zlog_msg_text(msg, &textlen);
	81	iov->iov_len = textlen;
	82	iov++;
	83
	84	zlog_msg_tsraw(msg, &ts);
834585bd DL	85	zlog_msg_pid(msg, &pid, &tid);
834585bd DL	86	xref = zlog_msg_xref(msg);
0798d276 DL	87
	88	hdr->ts_sec = ts.tv_sec;
	89	hdr->ts_nsec = ts.tv_nsec;
834585bd DL	90	hdr->pid = pid;
834585bd DL	91	hdr->tid = tid;
a4af82ee DL	92	hdr->lost_msgs = atomic_load_explicit(&zte->lost_msgs,
a4af82ee DL	93	memory_order_relaxed);
834585bd	94	hdr->prio = prio;
0798d276 DL	95	hdr->flags = 0;
0798d276 DL	96	hdr->textlen = textlen;
834585bd	97	hdr->texthdrlen = texthdrlen;
0798d276	98	hdr->n_argpos = n_argpos;
834585bd DL	99	if (xref) {
	100	memcpy(hdr->uid, xref->xref.xrefdata->uid,
	101	sizeof(hdr->uid));
	102	hdr->ec = xref->ec;
	103	} else {
	104	memset(hdr->uid, 0, sizeof(hdr->uid));
	105	hdr->ec = 0;
	106	}
	107	hdr->hdrlen = sizeof(hdr) + sizeof(argpos) * n_argpos;
0798d276 DL	108
	109	mmh->msg_hdr.msg_iovlen = iov - mmh->msg_hdr.msg_iov;
	110	mmh++;
	111	hdr++;
	112	}
	113
	114	size_t msgtotal = mmh - mmhs;
	115	ssize_t sent;
	116
	117	for (size_t msgpos = 0; msgpos < msgtotal; msgpos += sent) {
	118	sent = sendmmsg(fd, mmhs + msgpos, msgtotal - msgpos, 0);
	119
a4af82ee DL	120	if (sent <= 0 && (errno == EAGAIN \|\| errno == EWOULDBLOCK)) {
	121	atomic_fetch_add_explicit(&zte->lost_msgs,
	122	msgtotal - msgpos,
	123	memory_order_relaxed);
2eda953a	124	break;
a4af82ee	125	}
0798d276 DL	126	if (sent <= 0)
	127	goto out_err;
	128	}
	129	return;
	130
	131	out_err:
	132	fd = atomic_exchange_explicit(&zte->fd, -1, memory_order_relaxed);
	133	if (fd < 0)
	134	return;
	135
	136	rcu_close(&zte->head_close, fd);
	137	zlog_target_replace(zt, NULL);
	138
	139	state = STATE_NORMAL;
	140	atomic_compare_exchange_strong_explicit(
	141	&zte->state, &state, STATE_FD_DEAD, memory_order_relaxed,
	142	memory_order_relaxed);
	143	if (state == STATE_DISOWNED)
	144	rcu_free(MTYPE_LOG_LIVE, zte, head_self);
	145	}
	146
	147	static void zlog_live_sigsafe(struct zlog_target zt, const char text,
	148	size_t len)
	149	{
	150	struct zlt_live *zte = container_of(zt, struct zlt_live, zt);
1609a9d6	151	struct zlog_live_hdr hdr[1] = {};
0798d276 DL	152	struct iovec iovs[2], *iov = iovs;
	153	struct timespec ts;
	154	int fd;
	155
	156	fd = atomic_load_explicit(&zte->fd, memory_order_relaxed);
	157	if (fd < 0)
	158	return;
	159
1609a9d6	160	clock_gettime(CLOCK_REALTIME, &ts);
0798d276 DL	161
	162	hdr->ts_sec = ts.tv_sec;
	163	hdr->ts_nsec = ts.tv_nsec;
	164	hdr->prio = LOG_CRIT;
0798d276	165	hdr->textlen = len;
0798d276 DL	166
	167	iov->iov_base = (char *)hdr;
	168	iov->iov_len = sizeof(hdr);
	169	iov++;
	170
	171	iov->iov_base = (char *)text;
	172	iov->iov_len = len;
	173	iov++;
	174
	175	writev(fd, iovs, iov - iovs);
	176	}
	177
	178	void zlog_live_open(struct zlog_live_cfg cfg, int prio_min, int other_fd)
	179	{
	180	int sockets[2];
0798d276 DL	181
	182	if (cfg->target)
	183	zlog_live_close(cfg);
	184
	185	*other_fd = -1;
	186	if (prio_min == ZLOG_DISABLED)
	187	return;
	188
	189	/* the only reason for SEQPACKET here is getting close notifications.
	190	* otherwise if you open a bunch of vtysh connections with live logs
	191	* and close them all, the fds will stick around until we get an error
	192	* when trying to log something to them at some later point -- which
	193	* eats up fds and might be much later for some daemons.
	194	*/
	195	if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, sockets) < 0) {
	196	if (socketpair(AF_UNIX, SOCK_DGRAM, 0, sockets) < 0) {
	197	zlog_warn("%% could not open socket pair: %m");
	198	return;
	199	}
	200	} else
	201	/* SEQPACKET only: try to zap read direction */
	202	shutdown(sockets[0], SHUT_RD);
	203
	204	*other_fd = sockets[1];
3bcdae10 DL	205	zlog_live_open_fd(cfg, prio_min, sockets[0]);
	206	}
	207
	208	void zlog_live_open_fd(struct zlog_live_cfg *cfg, int prio_min, int fd)
	209	{
	210	struct zlt_live *zte;
	211	struct zlog_target *zt;
	212
	213	if (cfg->target)
	214	zlog_live_close(cfg);
0798d276 DL	215
	216	zt = zlog_target_clone(MTYPE_LOG_LIVE, NULL, sizeof(*zte));
	217	zte = container_of(zt, struct zlt_live, zt);
	218	cfg->target = zte;
	219
3bcdae10 DL	220	set_nonblocking(fd);
3bcdae10 DL	221	zte->fd = fd;
0798d276 DL	222	zte->zt.prio_min = prio_min;
	223	zte->zt.logfn = zlog_live;
	224	zte->zt.logfn_sigsafe = zlog_live_sigsafe;
	225
	226	zlog_target_replace(NULL, zt);
	227	}
	228
	229	void zlog_live_close(struct zlog_live_cfg *cfg)
	230	{
	231	struct zlt_live *zte;
	232	int fd;
	233
	234	if (!cfg->target)
	235	return;
	236
	237	zte = cfg->target;
	238	cfg->target = NULL;
	239
	240	fd = atomic_exchange_explicit(&zte->fd, -1, memory_order_relaxed);
	241
	242	if (fd >= 0) {
	243	rcu_close(&zte->head_close, fd);
	244	zlog_target_replace(&zte->zt, NULL);
	245	}
	246	rcu_free(MTYPE_LOG_LIVE, zte, head_self);
	247	}
	248
	249	void zlog_live_disown(struct zlog_live_cfg *cfg)
	250	{
	251	struct zlt_live *zte;
	252	uint_fast32_t state;
	253
	254	if (!cfg->target)
	255	return;
	256
	257	zte = cfg->target;
	258	cfg->target = NULL;
	259
	260	state = STATE_NORMAL;
	261	atomic_compare_exchange_strong_explicit(
	262	&zte->state, &state, STATE_DISOWNED, memory_order_relaxed,
	263	memory_order_relaxed);
	264	if (state == STATE_FD_DEAD)
	265	rcu_free(MTYPE_LOG_LIVE, zte, head_self);
	266	}