MINOR: sink: create definitions a minimal code for event sinks

The principle will be to be able to dispatch events to various destinations
called "sinks". This is already done in part in logs where log servers can
be either a UDP socket or a file descriptor. This will be needed with the
new trace subsystem where we may also want to add ring buffers. And it turns
out that all such destinations make sense at all places. Logs may need to be
sent to a TCP server via a ring buffer, or consulted from the CLI. Trace
events may need to be sent to stdout/stderr as well as to remote log servers.

This patch creates a new structure "sink" aiming at addressing these similar
needs. The goal is to merge together what is common to all of them, such as
the output format, the dropped events count, etc, and also keep separately
the target identification (network address, file descriptor). Provisions
were made to have a "waiter" on the sink. For a TCP log server it will be
the task to wake up after writing to the log buffer. For a ring buffer, it
could be the list of watchers on the CLI running a "tail" operation and
waiting for new events. A lock was also placed in the struct since many
operations will require some locking, including the FD ones. The output
formats covers those in use by logs and two extra ones prepending the ISO
time in front of the message (convenient for stdio/buffer).

For now only the generic infrastructure is present, no type-specific
output is implemented. There's the sink_write() function which prepares
and formats a message to be sent, trying hard to avoid copies and only
using pointer manipulation, where the type-specific code just has to be
added. Dropped messages are already counted (for now 100% drop). The
message is put into an iovec array as it will be trivial to use with
file descriptors and sockets.

Makefile

@@ -764,7 +764,7 @@ OBJS = src/http_ana.o src/cfgparse-listen.o src/stream.o                      \
        src/sample.o src/stream_interface.o src/proto_tcp.o src/listener.o     \
        src/h1.o src/cfgparse-global.o src/cache.o src/http_rules.o            \
        src/http_act.o src/tcp_rules.o src/filters.o src/connection.o          \
-       src/session.o src/acl.o src/vars.o src/raw_sock.o src/map.o            \
+       src/session.o src/acl.o src/vars.o src/raw_sock.o src/map.o src/sink.o \
        src/proto_uxst.o src/payload.o src/fd.o src/queue.o src/flt_trace.o    \
        src/task.o src/lb_chash.o src/frontend.o src/applet.o src/mux_pt.o     \
        src/signal.o src/ev_select.o src/proto_sockpair.o src/compression.o    \

include/proto/sink.h

@@ -0,0 +1,40 @@
+/*
+ * include/proto/sink.h
+ * This file provides declarations for event sinks management
+ *
+ * Copyright (C) 2000-2019 Willy Tarreau - w@1wt.eu
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation, version 2.1
+ * exclusively.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#ifndef _PROTO_SINK_H
+#define _PROTO_SINK_H
+
+#include <common/mini-clist.h>
+#include <types/sink.h>
+
+extern struct list sink_list;
+
+struct sink *sink_find(const char *name);
+void sink_write(struct sink *sink, const struct ist msg[], size_t nmsg);
+
+#endif /* _PROTO_SINK_H */
+
+/*
+ * Local variables:
+ *  c-indent-level: 8
+ *  c-basic-offset: 8
+ * End:
+ */

include/types/sink.h

@@ -0,0 +1,72 @@
+/*
+ * include/types/sink.h
+ * This file provides definitions for event sinks
+ *
+ * Copyright (C) 2000-2019 Willy Tarreau - w@1wt.eu
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation, version 2.1
+ * exclusively.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#ifndef _TYPES_SINK_H
+#define _TYPES_SINK_H
+
+#include <common/buffer.h>
+#include <common/compat.h>
+#include <common/config.h>
+#include <common/ist.h>
+
+/* A sink may be of several types. For now the following types are supported:
+ * (none yet)
+ */
+enum sink_type {
+	SINK_TYPE_NEW,      // not yet initialized
+};
+
+/* This indicates the default event format, which is the destination's
+ * preferred format, but may be overridden by the source.
+ */
+enum sink_fmt {
+	SINK_FMT_RAW,       // raw text sent as-is
+	SINK_FMT_SHORT,     // raw text prefixed with a syslog level
+	SINK_FMT_ISO,       // raw text prefixed with ISO time
+	SINK_FMT_TIMED,     // syslog level then ISO
+	SINK_FMT_RFC3164,   // regular syslog
+	SINK_FMT_RFC5424,   // extended syslog
+};
+
+/* describes the configuration and current state of an event sink */
+struct sink {
+	struct list sink_list;     // position in the sink list
+	const char *name;          // sink name
+	const char *desc;          // sink description
+	enum sink_fmt fmt;         // format expected by the sink
+	enum sink_type type;       // type of storage
+	uint8_t syslog_facility;   // used by syslog format
+	uint8_t syslog_minlvl;     // used by syslog & short formats
+	uint32_t maxlen;           // max message length (truncated above)
+	struct {
+		unsigned int dropped; // dropped events since last one.
+		__decl_hathreads(HA_RWLOCK_T lock); // used by some types
+	} ctx;
+};
+
+#endif /* _TYPES_SINK_H */
+
+/*
+ * Local variables:
+ *  c-indent-level: 8
+ *  c-basic-offset: 8
+ * End:
+ */

src/sink.c

@@ -0,0 +1,124 @@
+/*
+ * Event sink management
+ *
+ * Copyright (C) 2000-2019 Willy Tarreau - w@1wt.eu
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation, version 2.1
+ * exclusively.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#include <sys/uio.h>
+#include <common/compat.h>
+#include <common/config.h>
+#include <common/ist.h>
+#include <common/mini-clist.h>
+#include <proto/log.h>
+#include <proto/sink.h>
+
+struct list sink_list = LIST_HEAD_INIT(sink_list);
+
+struct sink *sink_find(const char *name)
+{
+	struct sink *sink;
+
+	list_for_each_entry(sink, &sink_list, sink_list)
+		if (strcmp(sink->name, name) == 0)
+			return sink;
+	return NULL;
+}
+
+/* creates a new sink and adds it to the list, it's still generic and not fully
+ * initialized. Returns NULL on allocation failure. If another one already
+ * exists with the same name, it will be returned. The caller can detect it as
+ * a newly created one has type SINK_TYPE_NEW.
+ */
+static __maybe_unused struct sink *__sink_new(const char *name, const char *desc, enum sink_fmt fmt)
+{
+	struct sink *sink;
+
+	sink = sink_find(name);
+	if (sink)
+		goto end;
+
+	sink = malloc(sizeof(*sink));
+	if (!sink)
+		goto end;
+
+	sink->name = name;
+	sink->desc = desc;
+	sink->fmt  = fmt;
+	sink->type = SINK_TYPE_NEW;
+	/* set defaults for syslog ones */
+	sink->syslog_facility = 0;
+	sink->syslog_minlvl   = 0;
+	sink->maxlen = MAX_SYSLOG_LEN;
+	/* address will be filled by the caller if needed */
+	sink->ctx.dropped = 0;
+	HA_RWLOCK_INIT(&sink->ctx.lock);
+	LIST_ADDQ(&sink_list, &sink->sink_list);
+ end:
+	return sink;
+}
+
+/* tries to send <nmsg> message parts (up to 8, ignored above) from message
+ * array <msg> to sink <sink>. Formating according to the sink's preference is
+ * done here. Lost messages are accounted for in the sink's counter.
+ */
+void sink_write(struct sink *sink, const struct ist msg[], size_t nmsg)
+{
+	struct iovec iovec[10];
+	char short_hdr[4];
+	size_t maxlen = sink->maxlen ? sink->maxlen : ~0;
+	size_t sent = 0;
+	int vec = 0;
+
+	/* keep one char for a possible trailing '\n' in any case */
+	maxlen--;
+
+	if (sink->fmt == SINK_FMT_SHORT) {
+		short_hdr[0] = '<';
+		short_hdr[1] = '0' + sink->syslog_minlvl;
+		short_hdr[2] = '>';
+
+		iovec[vec].iov_base = short_hdr;
+		iovec[vec].iov_len  = MIN(maxlen, 3);
+		maxlen -= iovec[vec].iov_len;
+		vec++;
+	}
+
+	/* copy the remaining entries from the original message. Skip empty fields and
+	 * truncate the whole message to maxlen.
+	 */
+	while (nmsg && vec < (sizeof(iovec) / sizeof(iovec[0]) - 1)) {
+		iovec[vec].iov_base = msg->ptr;
+		iovec[vec].iov_len  = MIN(maxlen, msg->len);
+		maxlen -= iovec[vec].iov_len;
+		if (iovec[vec].iov_len)
+			vec++;
+		msg++; nmsg--;
+	}
+
+	/* now deal with the various sink types here */
+
+	/* account for errors now */
+	if (sent <= 0)
+		HA_ATOMIC_ADD(&sink->ctx.dropped, 1);
+}
+
+/*
+ * Local variables:
+ *  c-indent-level: 8
+ *  c-basic-offset: 8
+ * End:
+ */