MEDIUM: sink: build header in sink_write for log formats

This patch extends the sink_write prototype and code to
handle the rfc5424 and rfc3164 header.

It uses header building tools from log.c. Doing this some
functions/vars have been externalized.

facility and minlevel have been removed from the struct sink
and passed to args at sink_write because they depends of the log
and not of the sink (they remained unused by rest of the code
until now).

include/proto/log.h

@@ -183,6 +183,14 @@ char *lf_ip(char *dst, const struct sockaddr *sockaddr, size_t size, const struc
 char *lf_port(char *dst, const struct sockaddr *sockaddr, size_t size, const struct logformat_node *node);
 
 
+/*
+ * Function to handle log header building (exported for sinks)
+ */
+char *update_log_hdr_rfc5424(const time_t time);
+char *update_log_hdr(const time_t time);
+char * get_format_pid_sep1(int format, size_t *len);
+char * get_format_pid_sep2(int format, size_t *len);
+
 #endif /* _PROTO_LOG_H */
 
 /*

include/proto/sink.h

@@ -29,8 +29,10 @@ extern struct list sink_list;
 
 struct sink *sink_find(const char *name);
 struct sink *sink_new_fd(const char *name, const char *desc, enum sink_fmt fmt, int fd);
-ssize_t __sink_write(struct sink *sink, const struct ist msg[], size_t nmsg);
+ssize_t __sink_write(struct sink *sink, const struct ist msg[], size_t nmsg,
-int sink_announce_dropped(struct sink *sink);
+                     int level, int facility, struct ist * tag,
+                     struct ist *pid, struct ist *sd);
+int sink_announce_dropped(struct sink *sink, int facility, struct ist *pid);
 
 
 /* tries to send <nmsg> message parts (up to 8, ignored above) from message
@@ -38,7 +40,9 @@ int sink_announce_dropped(struct sink *sink);
  * done here. Lost messages are accounted for in the sink's counter. If there
  * were lost messages, an attempt is first made to indicate it.
  */
-static inline void sink_write(struct sink *sink, const struct ist msg[], size_t nmsg)
+static inline void sink_write(struct sink *sink, const struct ist msg[], size_t nmsg,
+                              int level, int facility, struct ist * tag,
+                              struct ist *pid, struct ist *sd)
 {
 	ssize_t sent;
 
@@ -50,7 +54,7 @@ static inline void sink_write(struct sink *sink, const struct ist msg[], size_t
 		 * position.
 		 */
 		HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &sink->ctx.lock);
-		sent = sink_announce_dropped(sink);
+		sent = sink_announce_dropped(sink, facility, pid);
 		HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &sink->ctx.lock);
 
 		if (!sent) {
@@ -62,7 +66,7 @@ static inline void sink_write(struct sink *sink, const struct ist msg[], size_t
 	}
 
 	HA_RWLOCK_RDLOCK(LOGSRV_LOCK, &sink->ctx.lock);
-	sent = __sink_write(sink, msg, nmsg);
+	sent = __sink_write(sink, msg, nmsg, level, facility, tag, pid, sd);
 	HA_RWLOCK_RDUNLOCK(LOGSRV_LOCK, &sink->ctx.lock);
 
  fail:

include/types/sink.h

@@ -56,8 +56,6 @@ struct sink {
 	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 {
 		__decl_hathreads(HA_RWLOCK_T lock); // shared/excl for dropped

src/log.c

@@ -86,6 +86,18 @@ static const struct log_fmt log_formats[LOG_FORMATS] = {
 	},
 };
 
+char *get_format_pid_sep1(int format, size_t *len)
+{
+	*len = log_formats[format].pid.sep1.data;
+	return log_formats[format].pid.sep1.area;
+}
+
+char *get_format_pid_sep2(int format, size_t *len)
+{
+	*len = log_formats[format].pid.sep2.data;
+	return log_formats[format].pid.sep2.area;
+}
+
 /*
  * This map is used with all the FD_* macros to check whether a particular bit
  * is set or not. Each bit represents an ACSII code. ha_bit_set() sets those
@@ -1429,7 +1441,7 @@ char *lf_port(char *dst, const struct sockaddr *sockaddr, size_t size, const str
  * the beginning of logheader once a second and return the pointer to the
  * first character after it.
  */
-static char *update_log_hdr(const time_t time)
+char *update_log_hdr(const time_t time)
 {
 	static THREAD_LOCAL long tvsec;
 	static THREAD_LOCAL struct buffer host = { };
@@ -1473,7 +1485,7 @@ static char *update_log_hdr(const time_t time)
  * the beginning of logheader_rfc5424 once a second and return the pointer
  * to the first character after it.
  */
-static char *update_log_hdr_rfc5424(const time_t time)
+char *update_log_hdr_rfc5424(const time_t time)
 {
 	static THREAD_LOCAL long tvsec;
 	const char *gmt_offset;

src/sample.c

@@ -1478,7 +1478,7 @@ static int sample_conv_debug(const struct arg *arg_p, struct sample *smp, void *
 
  done:
 	line = ist2(buf->area, buf->data);
-	sink_write(sink, &line, 1);
+	sink_write(sink, &line, 1, 0, 0, NULL, NULL, NULL);
  end:
 	free_trash_chunk(buf);
 	return 1;

src/sink.c

@@ -62,9 +62,6 @@ static struct sink *__sink_new(const char *name, const char *desc, enum sink_fmt
 	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 = BUFSIZE;
 	/* address will be filled by the caller if needed */
 	sink->ctx.fd = -1;
@@ -144,28 +141,89 @@ struct sink *sink_new_buf(const char *name, const char *desc, enum sink_fmt fmt,
  * messages when there are any. It returns >0 if it could write anything,
  * <=0 otherwise.
  */
-ssize_t __sink_write(struct sink *sink, const struct ist msg[], size_t nmsg)
+ssize_t __sink_write(struct sink *sink, const struct ist msg[], size_t nmsg,
+	             int level, int facility, struct ist *tag,
+		     struct ist *pid, struct ist *sd)
 {
+	int log_format;
 	char short_hdr[4];
-	struct ist pfx[4];
+	struct ist pfx[6];
 	size_t npfx = 0;
+	char *hdr_ptr;
+	int fac_level;
+
+	if (sink->fmt == SINK_FMT_RAW)
+		goto send;
 
 	if (sink->fmt == SINK_FMT_SHORT || sink->fmt == SINK_FMT_TIMED) {
 		short_hdr[0] = '<';
-		short_hdr[1] = '0' + sink->syslog_minlvl;
+		short_hdr[1] = '0' + level;
 		short_hdr[2] = '>';
 
 		pfx[npfx].ptr = short_hdr;
 		pfx[npfx].len = 3;
 		npfx++;
+		if (sink->fmt == SINK_FMT_SHORT)
+			goto send;
         }
 
+
 	if (sink->fmt == SINK_FMT_ISO || sink->fmt == SINK_FMT_TIMED) {
 		pfx[npfx].ptr = timeofday_as_iso_us(1);
 		pfx[npfx].len = 27;
 		npfx++;
+		goto send;
+        }
+	else if (sink->fmt == SINK_FMT_RFC5424) {
+		pfx[npfx].ptr = logheader_rfc5424;
+                pfx[npfx].len = update_log_hdr_rfc5424(date.tv_sec) - pfx[npfx].ptr;
+		log_format = LOG_FORMAT_RFC5424;
+	}
+	else {
+		pfx[npfx].ptr = logheader;
+                pfx[npfx].len = update_log_hdr(date.tv_sec) - pfx[npfx].ptr;
+		log_format = LOG_FORMAT_RFC3164;
+		sd = NULL;
 	}
 
+	fac_level = (facility << 3) + level;
+	hdr_ptr = pfx[npfx].ptr + 3; /* last digit of the log level */
+        do {
+		*hdr_ptr = '0' + fac_level % 10;
+		fac_level /= 10;
+                hdr_ptr--;
+	} while (fac_level && hdr_ptr > pfx[npfx].ptr);
+	*hdr_ptr = '<';
+	pfx[npfx].len -= hdr_ptr - pfx[npfx].ptr;
+	pfx[npfx].ptr = hdr_ptr;
+	npfx++;
+
+	if (tag && tag->len) {
+		pfx[npfx].ptr = tag->ptr;
+		pfx[npfx].len = tag->len;
+		npfx++;
+	}
+	pfx[npfx].ptr = get_format_pid_sep1(log_format, &pfx[npfx].len);
+	if (pfx[npfx].len)
+		npfx++;
+
+	if (pid && pid->len) {
+		pfx[npfx].ptr = pid->ptr;
+		pfx[npfx].len = pid->len;
+		npfx++;
+	}
+
+	pfx[npfx].ptr = get_format_pid_sep2(log_format, &pfx[npfx].len);
+	if (pfx[npfx].len)
+		npfx++;
+
+	if (sd && sd->len) {
+		pfx[npfx].ptr = sd->ptr;
+		pfx[npfx].len = sd->len;
+		npfx++;
+	}
+
+send:
 	if (sink->type == SINK_TYPE_FD) {
 		return fd_write_frag_line(sink->ctx.fd, sink->maxlen, pfx, npfx, msg, nmsg, 1);
 	}
@@ -180,18 +238,25 @@ ssize_t __sink_write(struct sink *sink, const struct ist msg[], size_t nmsg)
  * called under an exclusive lock on the sink to avoid multiple produces doing
  * the same. On success, >0 is returned, otherwise <=0 on failure.
  */
-int sink_announce_dropped(struct sink *sink)
+int sink_announce_dropped(struct sink *sink, int facility, struct ist *pid)
 {
 	unsigned int dropped;
 	struct buffer msg;
 	struct ist msgvec[1];
 	char logbuf[64];
+	struct ist sd;
+	struct ist tag;
 
 	while (unlikely((dropped = sink->ctx.dropped) > 0)) {
 		chunk_init(&msg, logbuf, sizeof(logbuf));
 		chunk_printf(&msg, "%u event%s dropped", dropped, dropped > 1 ? "s" : "");
 		msgvec[0] = ist2(msg.area, msg.data);
-		if (__sink_write(sink, msgvec, 1) <= 0)
+
+		sd.ptr = default_rfc5424_sd_log_format;
+		sd.len = 2;
+		tag.ptr = global.log_tag.area;
+		tag.len = global.log_tag.data;
+		if (__sink_write(sink, msgvec, 1, LOG_NOTICE, facility, &tag, pid, &sd) <= 0)
 			return 0;
 		/* success! */
 		HA_ATOMIC_SUB(&sink->ctx.dropped, dropped);

src/trace.c

@@ -229,7 +229,7 @@ void __trace(enum trace_level level, uint64_t mask, struct trace_source *src,
 	}
 
 	if (src->sink)
-		sink_write(src->sink, line, words);
+		sink_write(src->sink, line, words, 0, 0, NULL, NULL, NULL);
 
  end:
 	/* check if we need to stop the trace now */