MEDIUM: log: support tcp or stream addresses on log lines.

An explicit stream address prefix such as "tcp6@" "tcp4@"
"stream+ipv6@" "stream+ipv4@" or "stream+unix@" will
allocate an implicit ring buffer with a forward server
targeting the given address.

This is usefull to simply send logs to a log server in tcp
and It doesn't need to declare a ring section in configuration.

doc/configuration.txt

@@ -7415,6 +7415,10 @@ no log
                  when used as a complement this can help troubleshooting by
                  having the logs instantly available.
 
+               - An explicit stream address prefix such as "tcp@","tcp6@",
+                 "tcp4@" or "uxst@" will allocate an implicit ring buffer with
+                 a stream forward server targeting the given address.
+
                You may want to reference some environment variables in the
                address parameter, see section 2.3 about environment variables.
 
@@ -7522,7 +7526,8 @@ no log
     log stdout format raw daemon            # send everything to stdout
     log stderr format raw daemon notice     # send important events to stderr
     log 127.0.0.1:514 local0 notice         # only send important events
-    log 127.0.0.1:514 local0 notice notice  # same but limit output level
+    log tcp@127.0.0.1:514 local0 notice notice  # same but limit output
+                                                # level and send in tcp
     log "${LOCAL_SYSLOG}:514" local0 notice   # send to local server
 
 

src/log.c

@@ -812,6 +812,7 @@ int parse_logsrv(char **args, struct list *logsrvs, int do_del, const char *file
 {
 	struct smp_log_range *smp_rgs = NULL;
 	struct sockaddr_storage *sk;
+	struct protocol *proto;
 	struct logsrv *logsrv = NULL;
 	int port1, port2;
 	int cur_arg;
@@ -1035,8 +1036,9 @@ int parse_logsrv(char **args, struct list *logsrvs, int do_del, const char *file
 		goto done;
 	}
 
-	sk = str2sa_range(args[1], NULL, &port1, &port2, &fd, NULL,
+	sk = str2sa_range(args[1], NULL, &port1, &port2, &fd, &proto,
-	                  err, NULL, NULL, PA_O_RESOLVE | PA_O_PORT_OK | PA_O_RAW_FD | PA_O_DGRAM);
+	                  err, NULL, NULL,
+	                  PA_O_RESOLVE | PA_O_PORT_OK | PA_O_RAW_FD | PA_O_DGRAM | PA_O_STREAM | PA_O_DEFAULT_DGRAM);
 	if (!sk)
 		goto error;
 
@@ -1049,6 +1051,18 @@ int parse_logsrv(char **args, struct list *logsrvs, int do_del, const char *file
 			set_host_port(&logsrv->addr, SYSLOG_PORT);
 	}
 
+	if (proto->ctrl_type == SOCK_STREAM) {
+		static unsigned long ring_ids;
+
+		/* Implicit sink buffer will be
+		 * initialized in post_check
+		 */
+		logsrv->type = LOG_TARGET_BUFFER;
+		logsrv->sink = NULL;
+		/* compute uniq name for the ring */
+		memprintf(&logsrv->ring_name, "ring#%lu", ++ring_ids);
+	}
+
  done:
 	LIST_ADDQ(logsrvs, &logsrv->list);
 	return 1;

src/sink.c

@@ -918,6 +918,158 @@ int cfg_parse_ring(const char *file, int linenum, char **args, int kwm)
 	return err_code;
 }
 
+/* Creates an new sink buffer from a log server.
+ *
+ * It uses the logsrvaddress to declare a forward
+ * server for this buffer. And it initializes the
+ * forwarding.
+ *
+ * The function returns a pointer on the
+ * allocated struct sink if allocate
+ * and initialize succeed, else if it fails
+ * it returns NULL.
+ *
+ * Note: the sink is created using the name
+ *       specified inot logsrv->ring_name
+ */
+struct sink *sink_new_from_logsrv(struct logsrv *logsrv)
+{
+	struct proxy *p = NULL;
+	struct sink *sink = NULL;
+	struct server *srv = NULL;
+	struct sink_forward_target *sft = NULL;
+	int i;
+
+	/* allocate new proxy to handle
+	 * forward to a stream server
+	 */
+	p = calloc(1, sizeof *p);
+	if (!p) {
+		goto error;
+        }
+
+	init_new_proxy(p);
+	sink_setup_proxy(p);
+	p->id = strdup(logsrv->ring_name);
+	p->conf.args.file = p->conf.file = strdup(logsrv->conf.file);
+	p->conf.args.line = p->conf.line = logsrv->conf.line;
+
+	/* allocate a new server to forward messages
+	 * from ring buffer
+	 */
+	srv = new_server(p);
+	if (!srv)
+		goto error;
+
+	/* init server */
+	srv->id = strdup(logsrv->ring_name);
+	srv->conf.file = strdup(logsrv->conf.file);
+	srv->conf.line = logsrv->conf.line;
+	srv->addr = logsrv->addr;
+        srv->svc_port = get_host_port(&logsrv->addr);
+	HA_SPIN_INIT(&srv->lock);
+
+	/* process per thread init */
+	srv->per_thr = calloc(global.nbthread, sizeof(*srv->per_thr));
+	if (!srv->per_thr)
+		goto error;
+
+	for (i = 0; i < global.nbthread; i++) {
+		srv->per_thr[i].idle_conns = EB_ROOT;
+		srv->per_thr[i].safe_conns = EB_ROOT;
+		srv->per_thr[i].avail_conns = EB_ROOT;
+		MT_LIST_INIT(&srv->per_thr[i].streams);
+	}
+
+	/* the servers are linked backwards
+	 * first into proxy
+	 */
+	p->srv = srv;
+	srv->next = p->srv;
+
+	/* allocate sink_forward_target descriptor */
+	sft = calloc(1, sizeof(*sft));
+	if (!sft)
+		goto error;
+
+	/* init sink_forward_target offset */
+	sft->srv = srv;
+	sft->appctx = NULL;
+	sft->ofs = ~0;
+	HA_SPIN_INIT(&sft->lock);
+
+	/* prepare descrition for sink */
+	chunk_reset(&trash);
+	chunk_printf(&trash, "created from logserver declared into '%s' at line %d", logsrv->conf.file, logsrv->conf.line);
+
+	/* allocate a new sink buffer */
+	sink = sink_new_buf(logsrv->ring_name, trash.area, logsrv->format, BUFSIZE);
+	if (!sink || sink->type != SINK_TYPE_BUFFER) {
+		goto error;
+	}
+
+	/* link sink_forward_target to proxy */
+	sink->forward_px = p;
+	p->parent = sink;
+
+	/* insert into sink_forward_targets
+	 * list into sink
+	 */
+	sft->next = sink->sft;
+	sink->sft = sft;
+
+	/* mark server as an attached reader to the ring */
+	if (!ring_attach(sink->ctx.ring)) {
+		/* should never fail since there is
+		 * only one reader
+		 */
+		goto error;
+	}
+
+	/* initialize sink buffer forwarding */
+	if (!sink_init_forward(sink))
+		goto error;
+
+	/* reset familyt of logsrv to consider the ring buffer target */
+	logsrv->addr.ss_family = AF_UNSPEC;
+
+	return sink;
+error:
+	if (p) {
+		if (p->id)
+			free(p->id);
+		if (p->conf.file)
+			free(p->conf.file);
+
+		free(p);
+	}
+
+	if (srv) {
+		if (srv->id)
+			free(srv->id);
+		if (srv->conf.file)
+			free((void *)srv->conf.file);
+		if (srv->per_thr)
+		       free(srv->per_thr);
+		free(srv);
+	}
+
+	if (sft)
+		free(sft);
+
+	if (sink) {
+		if (sink->ctx.ring)
+			ring_free(sink->ctx.ring);
+
+		LIST_DEL(&sink->sink_list);
+		free(sink->name);
+		free(sink->desc);
+		free(sink);
+	}
+
+	return NULL;
+}
+
 /*
  * Post parsing "ring" section.
  *
@@ -992,20 +1144,61 @@ int post_sink_resolve()
 	list_for_each_entry_safe(logsrv, logb, &global.logsrvs, list) {
 		if (logsrv->type == LOG_TARGET_BUFFER) {
 			sink = sink_find(logsrv->ring_name);
-			if (!sink || sink->type != SINK_TYPE_BUFFER) {
+			if (!sink) {
-				ha_alert("global log server declared in file '%s' at line %d uses unknown ring named '%s'.\n", logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
+				/* LOG_TARGET_BUFFER but !AF_UNSPEC
+				 * means we must allocate a sink
+				 * buffer to send messages to this logsrv
+				 */
+				if (logsrv->addr.ss_family != AF_UNSPEC) {
+					sink = sink_new_from_logsrv(logsrv);
+					if (!sink) {
+						ha_alert("global stream log server declared in file '%s' at line %d cannot be initialized'.\n",
+						         logsrv->conf.file, logsrv->conf.line);
+						err_code |= ERR_ALERT | ERR_FATAL;
+					}
+				}
+				else {
+					ha_alert("global log server declared in file '%s' at line %d uses unknown ring named '%s'.\n",
+					         logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
+					err_code |= ERR_ALERT | ERR_FATAL;
+				}
+			}
+			else if (sink->type != SINK_TYPE_BUFFER) {
+				ha_alert("global log server declared in file '%s' at line %d uses incompatible ring '%s'.\n",
+				         logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
 				err_code |= ERR_ALERT | ERR_FATAL;
 			}
 			logsrv->sink = sink;
 		}
+
 	}
 
 	for (px = proxies_list; px; px = px->next) {
 		list_for_each_entry_safe(logsrv, logb, &px->logsrvs, list) {
 			if (logsrv->type == LOG_TARGET_BUFFER) {
 				sink = sink_find(logsrv->ring_name);
-				if (!sink || sink->type != SINK_TYPE_BUFFER) {
+				if (!sink) {
-					ha_alert("log server declared in proxy section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n", px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
+					/* LOG_TARGET_BUFFER but !AF_UNSPEC
+					 * means we must allocate a sink
+					 * buffer to send messages to this logsrv
+					 */
+					if (logsrv->addr.ss_family != AF_UNSPEC) {
+						sink = sink_new_from_logsrv(logsrv);
+						if (!sink) {
+							ha_alert("log server declared in proxy section '%s' file '%s' at line %d cannot be initialized'.\n",
+							         px->id, logsrv->conf.file, logsrv->conf.line);
+							err_code |= ERR_ALERT | ERR_FATAL;
+						}
+					}
+					else {
+						ha_alert("log server declared in proxy section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n",
+						         px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
+						err_code |= ERR_ALERT | ERR_FATAL;
+					}
+				}
+				else if (sink->type != SINK_TYPE_BUFFER) {
+					ha_alert("log server declared in proxy section '%s' in file '%s' at line %d uses incomatible ring named '%s'.\n",
+					         px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
 					err_code |= ERR_ALERT | ERR_FATAL;
 				}
 				logsrv->sink = sink;
@@ -1017,8 +1210,28 @@ int post_sink_resolve()
 		list_for_each_entry_safe(logsrv, logb, &px->logsrvs, list) {
 			if (logsrv->type == LOG_TARGET_BUFFER) {
 				sink = sink_find(logsrv->ring_name);
-				if (!sink || sink->type != SINK_TYPE_BUFFER) {
+				if (!sink) {
-					ha_alert("log server declared in log-forward section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n", px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
+					/* LOG_TARGET_BUFFER but !AF_UNSPEC
+					 * means we must allocate a sink
+					 * buffer to send messages to this logsrv
+					 */
+					if (logsrv->addr.ss_family != AF_UNSPEC) {
+						sink = sink_new_from_logsrv(logsrv);
+						if (!sink) {
+							ha_alert("log server declared in log-forward section '%s' file '%s' at line %d cannot be initialized'.\n",
+							         px->id, logsrv->conf.file, logsrv->conf.line);
+							err_code |= ERR_ALERT | ERR_FATAL;
+						}
+					}
+					else {
+						ha_alert("log server declared in log-forward section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n",
+							 px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
+						err_code |= ERR_ALERT | ERR_FATAL;
+					}
+				}
+				else if (sink->type != SINK_TYPE_BUFFER) {
+					ha_alert("log server declared in log-forward section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n",
+						 px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
 					err_code |= ERR_ALERT | ERR_FATAL;
 				}
 				logsrv->sink = sink;