From 4ed23ca0e757f1e4b33fbd3357e45230fbb61f9f Mon Sep 17 00:00:00 2001
From: Willy Tarreau <w@1wt.eu>
Date: Fri, 23 Aug 2019 15:47:49 +0200
Subject: [PATCH] MINOR: sink: add support for ring buffers

This now provides sink_new_buf() which allocates a ring buffer. One such
ring ("buf0") of 1 MB is created already, and may be used by sink_write().
The sink's creation should probably be moved somewhere else later.
---
 include/types/sink.h |  5 +++-
 src/sink.c           | 59 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 63 insertions(+), 1 deletion(-)

diff --git a/include/types/sink.h b/include/types/sink.h
index 32c056712..5465ca63d 100644
--- a/include/types/sink.h
+++ b/include/types/sink.h
@@ -27,12 +27,14 @@
 #include <common/config.h>
 #include <common/ist.h>
 
-/* A sink may be of several types. For now the following types are supported:
+/* A sink may be of 4 distinct types :
  *   - file descriptor (such as stdout)
+ *   - ring buffer, readable from CLI
  */
 enum sink_type {
 	SINK_TYPE_NEW,      // not yet initialized
 	SINK_TYPE_FD,       // events sent to a file descriptor
+	SINK_TYPE_BUFFER,   // events sent to a ring buffer
 };
 
 /* This indicates the default event format, which is the destination's
@@ -58,6 +60,7 @@ struct sink {
 	uint8_t syslog_minlvl;     // used by syslog & short formats
 	uint32_t maxlen;           // max message length (truncated above)
 	struct {
+		struct ring *ring;    // used by ring buffer and STRM sender
 		unsigned int dropped; // dropped events since last one.
 		__decl_hathreads(HA_RWLOCK_T lock); // used by some types
 		int fd;               // fd num for FD type sink
diff --git a/src/sink.c b/src/sink.c
index 18b5c07b5..3295cca02 100644
--- a/src/sink.c
+++ b/src/sink.c
@@ -25,6 +25,7 @@
 #include <common/ist.h>
 #include <common/mini-clist.h>
 #include <proto/log.h>
+#include <proto/ring.h>
 #include <proto/sink.h>
 
 struct list sink_list = LIST_HEAD_INIT(sink_list);
@@ -101,6 +102,45 @@ struct sink *sink_new_fd(const char *name, const char *desc, enum sink_fmt fmt,
 	return sink;
 }
 
+/* creates a sink called <name> of type BUF of size <size>, format <fmt>,
+ * and description <desc>. Returns NULL on allocation failure or conflict.
+ * Perfect duplicates are merged (same type and name). If sizes differ, the
+ * largest one is kept.
+ */
+struct sink *sink_new_buf(const char *name, const char *desc, enum sink_fmt fmt, size_t size)
+{
+	struct sink *sink;
+
+	sink = __sink_new(name, desc, fmt);
+	if (!sink)
+		goto fail;
+
+	if (sink->type == SINK_TYPE_BUFFER) {
+		/* such a buffer already exists, we may have to resize it */
+		if (!ring_resize(sink->ctx.ring, size))
+			goto fail;
+		goto end;
+	}
+
+	if (sink->type != SINK_TYPE_NEW) {
+		/* already exists of another type */
+		goto fail;
+	}
+
+	sink->ctx.ring = ring_new(size);
+	if (!sink->ctx.ring) {
+		LIST_DEL(&sink->sink_list);
+		free(sink);
+		goto fail;
+	}
+
+	sink->type = SINK_TYPE_BUFFER;
+ end:
+	return sink;
+ fail:
+	return NULL;
+}
+
 /* 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.
@@ -124,6 +164,11 @@ void sink_write(struct sink *sink, const struct ist msg[], size_t nmsg)
 
 	if (sink->type == SINK_TYPE_FD) {
 		sent = fd_write_frag_line(sink->ctx.fd, sink->maxlen, pfx, npfx, msg, nmsg, 1);
+		/* sent > 0 if the message was delivered */
+	}
+	else if (sink->type == SINK_TYPE_BUFFER) {
+		sent = ring_write(sink->ctx.ring, sink->maxlen, pfx, npfx, msg, nmsg);
+		/* sent > 0 if the message was delivered */
 	}
 
 	/* account for errors now */
@@ -135,9 +180,23 @@ static void sink_init()
 {
 	sink_new_fd("stdout", "standard output (fd#1)", SINK_FMT_RAW, 1);
 	sink_new_fd("stderr", "standard output (fd#2)", SINK_FMT_RAW, 2);
+	sink_new_buf("buf0",  "in-memory ring buffer", SINK_FMT_RAW, 1048576);
+}
+
+static void sink_deinit()
+{
+	struct sink *sink, *sb;
+
+	list_for_each_entry_safe(sink, sb, &sink_list, sink_list) {
+		if (sink->type == SINK_TYPE_BUFFER)
+			ring_free(sink->ctx.ring);
+		LIST_DEL(&sink->sink_list);
+		free(sink);
+	}
 }
 
 INITCALL0(STG_REGISTER, sink_init);
+REGISTER_POST_DEINIT(sink_deinit);
 
 /*
  * Local variables: