blob: 0f016890dfc71f00730284fc02c54d6ad0df45c8 [file] [log] [blame]
Willy Tarreau67b5a162019-08-11 16:38:56 +02001/*
2 * Event sink management
3 *
4 * Copyright (C) 2000-2019 Willy Tarreau - w@1wt.eu
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation, version 2.1
9 * exclusively.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, write to the Free Software
18 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
19 */
20
Willy Tarreau36979d92020-06-05 17:27:29 +020021#include <import/ist.h>
Willy Tarreau4c7e4b72020-05-27 12:58:42 +020022#include <haproxy/api.h>
Willy Tarreau6be78492020-06-05 00:00:29 +020023#include <haproxy/cfgparse.h>
Willy Tarreau83487a82020-06-04 20:19:54 +020024#include <haproxy/cli.h>
Willy Tarreau36979d92020-06-05 17:27:29 +020025#include <haproxy/errors.h>
Willy Tarreau853b2972020-05-27 18:01:47 +020026#include <haproxy/list.h>
Willy Tarreauaeed4a82020-06-04 22:01:04 +020027#include <haproxy/log.h>
Willy Tarreau817538e2021-05-08 20:20:21 +020028#include <haproxy/proxy.h>
Willy Tarreaud2ad57c2020-06-03 19:43:35 +020029#include <haproxy/ring.h>
Willy Tarreau3727a8a2020-06-04 17:37:26 +020030#include <haproxy/signal.h>
Willy Tarreauba2f73d2020-06-03 20:02:28 +020031#include <haproxy/sink.h>
Willy Tarreau5e539c92020-06-04 20:45:39 +020032#include <haproxy/stream_interface.h>
Willy Tarreaub2551052020-06-09 09:07:15 +020033#include <haproxy/time.h>
Willy Tarreau4bad5e22021-05-08 13:05:30 +020034#include <haproxy/tools.h>
Willy Tarreau67b5a162019-08-11 16:38:56 +020035
36struct list sink_list = LIST_HEAD_INIT(sink_list);
37
Emeric Brun99c453d2020-05-25 15:01:04 +020038struct sink *cfg_sink;
39
Willy Tarreau67b5a162019-08-11 16:38:56 +020040struct sink *sink_find(const char *name)
41{
42 struct sink *sink;
43
44 list_for_each_entry(sink, &sink_list, sink_list)
45 if (strcmp(sink->name, name) == 0)
46 return sink;
47 return NULL;
48}
49
50/* creates a new sink and adds it to the list, it's still generic and not fully
51 * initialized. Returns NULL on allocation failure. If another one already
52 * exists with the same name, it will be returned. The caller can detect it as
53 * a newly created one has type SINK_TYPE_NEW.
54 */
Emeric Brun54648852020-07-06 15:54:06 +020055static struct sink *__sink_new(const char *name, const char *desc, int fmt)
Willy Tarreau67b5a162019-08-11 16:38:56 +020056{
57 struct sink *sink;
58
59 sink = sink_find(name);
60 if (sink)
61 goto end;
62
Emeric Brun494c5052020-05-28 11:13:15 +020063 sink = calloc(1, sizeof(*sink));
Willy Tarreau67b5a162019-08-11 16:38:56 +020064 if (!sink)
65 goto end;
66
Emeric Brun99c453d2020-05-25 15:01:04 +020067 sink->name = strdup(name);
Tim Duesterhusa7ebffe2021-01-03 19:54:11 +010068 if (!sink->name)
69 goto err;
70
Emeric Brun99c453d2020-05-25 15:01:04 +020071 sink->desc = strdup(desc);
Tim Duesterhusa7ebffe2021-01-03 19:54:11 +010072 if (!sink->desc)
73 goto err;
74
Willy Tarreau67b5a162019-08-11 16:38:56 +020075 sink->fmt = fmt;
76 sink->type = SINK_TYPE_NEW;
Christopher Fauleta63a5c22019-11-15 15:10:12 +010077 sink->maxlen = BUFSIZE;
Willy Tarreau67b5a162019-08-11 16:38:56 +020078 /* address will be filled by the caller if needed */
Willy Tarreau973e6622019-08-20 11:57:52 +020079 sink->ctx.fd = -1;
Willy Tarreau67b5a162019-08-11 16:38:56 +020080 sink->ctx.dropped = 0;
81 HA_RWLOCK_INIT(&sink->ctx.lock);
Willy Tarreau2b718102021-04-21 07:32:39 +020082 LIST_APPEND(&sink_list, &sink->sink_list);
Willy Tarreau67b5a162019-08-11 16:38:56 +020083 end:
84 return sink;
Tim Duesterhusa7ebffe2021-01-03 19:54:11 +010085
86 err:
Willy Tarreau61cfdf42021-02-20 10:46:51 +010087 ha_free(&sink->name);
88 ha_free(&sink->desc);
89 ha_free(&sink);
Tim Duesterhusa7ebffe2021-01-03 19:54:11 +010090
91 return NULL;
Willy Tarreau67b5a162019-08-11 16:38:56 +020092}
93
Willy Tarreau973e6622019-08-20 11:57:52 +020094/* creates a sink called <name> of type FD associated to fd <fd>, format <fmt>,
95 * and description <desc>. Returns NULL on allocation failure or conflict.
96 * Perfect duplicates are merged (same type, fd, and name).
97 */
Emeric Brun54648852020-07-06 15:54:06 +020098struct sink *sink_new_fd(const char *name, const char *desc, enum log_fmt fmt, int fd)
Willy Tarreau973e6622019-08-20 11:57:52 +020099{
100 struct sink *sink;
101
102 sink = __sink_new(name, desc, fmt);
103 if (!sink || (sink->type == SINK_TYPE_FD && sink->ctx.fd == fd))
104 goto end;
105
106 if (sink->type != SINK_TYPE_NEW) {
107 sink = NULL;
108 goto end;
109 }
110
111 sink->type = SINK_TYPE_FD;
112 sink->ctx.fd = fd;
113 end:
114 return sink;
115}
116
Willy Tarreau4ed23ca2019-08-23 15:47:49 +0200117/* creates a sink called <name> of type BUF of size <size>, format <fmt>,
118 * and description <desc>. Returns NULL on allocation failure or conflict.
119 * Perfect duplicates are merged (same type and name). If sizes differ, the
120 * largest one is kept.
121 */
Emeric Brun54648852020-07-06 15:54:06 +0200122struct sink *sink_new_buf(const char *name, const char *desc, enum log_fmt fmt, size_t size)
Willy Tarreau4ed23ca2019-08-23 15:47:49 +0200123{
124 struct sink *sink;
125
126 sink = __sink_new(name, desc, fmt);
127 if (!sink)
128 goto fail;
129
130 if (sink->type == SINK_TYPE_BUFFER) {
131 /* such a buffer already exists, we may have to resize it */
132 if (!ring_resize(sink->ctx.ring, size))
133 goto fail;
134 goto end;
135 }
136
137 if (sink->type != SINK_TYPE_NEW) {
138 /* already exists of another type */
139 goto fail;
140 }
141
142 sink->ctx.ring = ring_new(size);
143 if (!sink->ctx.ring) {
Willy Tarreau2b718102021-04-21 07:32:39 +0200144 LIST_DELETE(&sink->sink_list);
Emeric Brun99c453d2020-05-25 15:01:04 +0200145 free(sink->name);
146 free(sink->desc);
Willy Tarreau4ed23ca2019-08-23 15:47:49 +0200147 free(sink);
148 goto fail;
149 }
150
151 sink->type = SINK_TYPE_BUFFER;
152 end:
153 return sink;
154 fail:
155 return NULL;
156}
157
Willy Tarreau67b5a162019-08-11 16:38:56 +0200158/* tries to send <nmsg> message parts (up to 8, ignored above) from message
Ilya Shipitsin46a030c2020-07-05 16:36:08 +0500159 * array <msg> to sink <sink>. Formatting according to the sink's preference is
Willy Tarreau8f240232019-08-27 16:41:06 +0200160 * done here. Lost messages are NOT accounted for. It is preferable to call
161 * sink_write() instead which will also try to emit the number of dropped
162 * messages when there are any. It returns >0 if it could write anything,
163 * <=0 otherwise.
Willy Tarreau67b5a162019-08-11 16:38:56 +0200164 */
Emeric Brun54648852020-07-06 15:54:06 +0200165 ssize_t __sink_write(struct sink *sink, const struct ist msg[], size_t nmsg,
166 int level, int facility, struct ist *metadata)
167 {
168 struct ist *pfx = NULL;
Willy Tarreaua1426de2019-08-27 14:21:02 +0200169 size_t npfx = 0;
Emeric Brunbd163812020-05-06 14:33:46 +0200170
Emeric Brun54648852020-07-06 15:54:06 +0200171 if (sink->fmt == LOG_FORMAT_RAW)
Emeric Brunbd163812020-05-06 14:33:46 +0200172 goto send;
Willy Tarreau67b5a162019-08-11 16:38:56 +0200173
Emeric Brun54648852020-07-06 15:54:06 +0200174 pfx = build_log_header(sink->fmt, level, facility, metadata, &npfx);
Emeric Brunbd163812020-05-06 14:33:46 +0200175
176send:
Willy Tarreau973e6622019-08-20 11:57:52 +0200177 if (sink->type == SINK_TYPE_FD) {
Willy Tarreau8f240232019-08-27 16:41:06 +0200178 return fd_write_frag_line(sink->ctx.fd, sink->maxlen, pfx, npfx, msg, nmsg, 1);
Willy Tarreau4ed23ca2019-08-23 15:47:49 +0200179 }
180 else if (sink->type == SINK_TYPE_BUFFER) {
Willy Tarreau8f240232019-08-27 16:41:06 +0200181 return ring_write(sink->ctx.ring, sink->maxlen, pfx, npfx, msg, nmsg);
Willy Tarreau973e6622019-08-20 11:57:52 +0200182 }
Willy Tarreau8f240232019-08-27 16:41:06 +0200183 return 0;
184}
Willy Tarreau67b5a162019-08-11 16:38:56 +0200185
Willy Tarreau8f240232019-08-27 16:41:06 +0200186/* Tries to emit a message indicating the number of dropped events. In case of
187 * success, the amount of drops is reduced by as much. It's supposed to be
188 * called under an exclusive lock on the sink to avoid multiple produces doing
189 * the same. On success, >0 is returned, otherwise <=0 on failure.
190 */
Emeric Brun54648852020-07-06 15:54:06 +0200191int sink_announce_dropped(struct sink *sink, int facility)
Willy Tarreau8f240232019-08-27 16:41:06 +0200192{
Emeric Brun54648852020-07-06 15:54:06 +0200193 static THREAD_LOCAL struct ist metadata[LOG_META_FIELDS];
194 static THREAD_LOCAL pid_t curr_pid;
195 static THREAD_LOCAL char pidstr[16];
Willy Tarreau8f240232019-08-27 16:41:06 +0200196 unsigned int dropped;
197 struct buffer msg;
198 struct ist msgvec[1];
199 char logbuf[64];
200
201 while (unlikely((dropped = sink->ctx.dropped) > 0)) {
202 chunk_init(&msg, logbuf, sizeof(logbuf));
203 chunk_printf(&msg, "%u event%s dropped", dropped, dropped > 1 ? "s" : "");
204 msgvec[0] = ist2(msg.area, msg.data);
Emeric Brunbd163812020-05-06 14:33:46 +0200205
Emeric Brun54648852020-07-06 15:54:06 +0200206 if (!metadata[LOG_META_HOST].len) {
207 if (global.log_send_hostname)
Tim Duesterhus77508502022-03-15 13:11:06 +0100208 metadata[LOG_META_HOST] = ist(global.log_send_hostname);
Emeric Brun54648852020-07-06 15:54:06 +0200209 }
210
211 if (!metadata[LOG_META_TAG].len)
212 metadata[LOG_META_TAG] = ist2(global.log_tag.area, global.log_tag.data);
213
214 if (unlikely(curr_pid != getpid()))
215 metadata[LOG_META_PID].len = 0;
216
217 if (!metadata[LOG_META_PID].len) {
218 curr_pid = getpid();
219 ltoa_o(curr_pid, pidstr, sizeof(pidstr));
220 metadata[LOG_META_PID] = ist2(pidstr, strlen(pidstr));
221 }
222
223 if (__sink_write(sink, msgvec, 1, LOG_NOTICE, facility, metadata) <= 0)
Willy Tarreau8f240232019-08-27 16:41:06 +0200224 return 0;
225 /* success! */
226 HA_ATOMIC_SUB(&sink->ctx.dropped, dropped);
227 }
228 return 1;
Willy Tarreau67b5a162019-08-11 16:38:56 +0200229}
230
Willy Tarreau9f830d72019-08-26 18:17:04 +0200231/* parse the "show events" command, returns 1 if a message is returned, otherwise zero */
232static int cli_parse_show_events(char **args, char *payload, struct appctx *appctx, void *private)
233{
234 struct sink *sink;
Willy Tarreau1d181e42019-08-30 11:17:01 +0200235 int arg;
Willy Tarreau9f830d72019-08-26 18:17:04 +0200236
237 args++; // make args[1] the 1st arg
238
239 if (!*args[1]) {
240 /* no arg => report the list of supported sink */
Willy Tarreau1d181e42019-08-30 11:17:01 +0200241 chunk_printf(&trash, "Supported events sinks are listed below. Add -w(wait), -n(new). Any key to stop\n");
Willy Tarreau9f830d72019-08-26 18:17:04 +0200242 list_for_each_entry(sink, &sink_list, sink_list) {
243 chunk_appendf(&trash, " %-10s : type=%s, %u dropped, %s\n",
244 sink->name,
245 sink->type == SINK_TYPE_NEW ? "init" :
246 sink->type == SINK_TYPE_FD ? "fd" :
247 sink->type == SINK_TYPE_BUFFER ? "buffer" : "?",
248 sink->ctx.dropped, sink->desc);
249 }
250
251 trash.area[trash.data] = 0;
252 return cli_msg(appctx, LOG_WARNING, trash.area);
253 }
254
255 if (!cli_has_level(appctx, ACCESS_LVL_OPER))
256 return 1;
257
258 sink = sink_find(args[1]);
259 if (!sink)
260 return cli_err(appctx, "No such event sink");
261
262 if (sink->type != SINK_TYPE_BUFFER)
263 return cli_msg(appctx, LOG_NOTICE, "Nothing to report for this sink");
264
Willy Tarreau1d181e42019-08-30 11:17:01 +0200265 for (arg = 2; *args[arg]; arg++) {
266 if (strcmp(args[arg], "-w") == 0)
267 appctx->ctx.cli.i0 |= 1; // wait mode
268 else if (strcmp(args[arg], "-n") == 0)
269 appctx->ctx.cli.i0 |= 2; // seek to new
270 else if (strcmp(args[arg], "-nw") == 0 || strcmp(args[arg], "-wn") == 0)
271 appctx->ctx.cli.i0 |= 3; // seek to new + wait
272 else
273 return cli_err(appctx, "unknown option");
274 }
Willy Tarreau9f830d72019-08-26 18:17:04 +0200275 return ring_attach_cli(sink->ctx.ring, appctx);
276}
277
Ilya Shipitsin47d17182020-06-21 21:42:57 +0500278/* Pre-configures a ring proxy to emit connections */
Emeric Brun494c5052020-05-28 11:13:15 +0200279void sink_setup_proxy(struct proxy *px)
280{
281 px->last_change = now.tv_sec;
282 px->cap = PR_CAP_FE | PR_CAP_BE;
283 px->maxconn = 0;
284 px->conn_retries = 1;
285 px->timeout.server = TICK_ETERNITY;
286 px->timeout.client = TICK_ETERNITY;
287 px->timeout.connect = TICK_ETERNITY;
288 px->accept = NULL;
289 px->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Emeric Brun494c5052020-05-28 11:13:15 +0200290}
291
292/*
293 * IO Handler to handle message push to syslog tcp server
294 */
295static void sink_forward_io_handler(struct appctx *appctx)
296{
Christopher Faulet53fa7872022-03-16 10:01:26 +0100297 struct stream_interface *si = cs_si(appctx->owner);
Emeric Brun494c5052020-05-28 11:13:15 +0200298 struct stream *s = si_strm(si);
299 struct sink *sink = strm_fe(s)->parent;
300 struct sink_forward_target *sft = appctx->ctx.sft.ptr;
301 struct ring *ring = sink->ctx.ring;
302 struct buffer *buf = &ring->buf;
303 uint64_t msg_len;
304 size_t len, cnt, ofs;
305 int ret = 0;
306
Ilya Shipitsin47d17182020-06-21 21:42:57 +0500307 /* if stopping was requested, close immediately */
Emeric Brun494c5052020-05-28 11:13:15 +0200308 if (unlikely(stopping))
309 goto close;
310
311 /* for rex because it seems reset to timeout
312 * and we don't want expire on this case
313 * with a syslog server
314 */
315 si_oc(si)->rex = TICK_ETERNITY;
316 /* rto should not change but it seems the case */
317 si_oc(si)->rto = TICK_ETERNITY;
318
319 /* an error was detected */
320 if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
321 goto close;
322
323 /* con closed by server side */
324 if ((si_oc(si)->flags & CF_SHUTW))
325 goto close;
326
327 /* if the connection is not established, inform the stream that we want
328 * to be notified whenever the connection completes.
329 */
330 if (si_opposite(si)->state < SI_ST_EST) {
331 si_cant_get(si);
332 si_rx_conn_blk(si);
333 si_rx_endp_more(si);
334 return;
335 }
336
337 HA_SPIN_LOCK(SFT_LOCK, &sft->lock);
338 if (appctx != sft->appctx) {
339 HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
340 goto close;
341 }
342 ofs = sft->ofs;
343
344 HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
345 LIST_DEL_INIT(&appctx->wait_entry);
346 HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &ring->lock);
347
348 HA_RWLOCK_RDLOCK(LOGSRV_LOCK, &ring->lock);
349
350 /* explanation for the initialization below: it would be better to do
351 * this in the parsing function but this would occasionally result in
352 * dropped events because we'd take a reference on the oldest message
353 * and keep it while being scheduled. Thus instead let's take it the
354 * first time we enter here so that we have a chance to pass many
355 * existing messages before grabbing a reference to a location. This
356 * value cannot be produced after initialization.
357 */
358 if (unlikely(ofs == ~0)) {
359 ofs = 0;
360
Willy Tarreau4781b152021-04-06 13:53:36 +0200361 HA_ATOMIC_INC(b_peek(buf, ofs));
Emeric Brun494c5052020-05-28 11:13:15 +0200362 ofs += ring->ofs;
363 }
364
Emeric Brun494c5052020-05-28 11:13:15 +0200365 /* in this loop, ofs always points to the counter byte that precedes
366 * the message so that we can take our reference there if we have to
367 * stop before the end (ret=0).
368 */
369 if (si_opposite(si)->state == SI_ST_EST) {
Emeric Brunfdabf492020-12-02 17:02:09 +0100370 /* we were already there, adjust the offset to be relative to
371 * the buffer's head and remove us from the counter.
372 */
373 ofs -= ring->ofs;
374 BUG_ON(ofs >= buf->size);
Willy Tarreau4781b152021-04-06 13:53:36 +0200375 HA_ATOMIC_DEC(b_peek(buf, ofs));
Emeric Brunfdabf492020-12-02 17:02:09 +0100376
Emeric Brun494c5052020-05-28 11:13:15 +0200377 ret = 1;
378 while (ofs + 1 < b_data(buf)) {
379 cnt = 1;
380 len = b_peek_varint(buf, ofs + cnt, &msg_len);
381 if (!len)
382 break;
383 cnt += len;
384 BUG_ON(msg_len + ofs + cnt + 1 > b_data(buf));
385
386 if (unlikely(msg_len + 1 > b_size(&trash))) {
387 /* too large a message to ever fit, let's skip it */
388 ofs += cnt + msg_len;
389 continue;
390 }
391
392 chunk_reset(&trash);
393 len = b_getblk(buf, trash.area, msg_len, ofs + cnt);
394 trash.data += len;
395 trash.area[trash.data++] = '\n';
396
397 if (ci_putchk(si_ic(si), &trash) == -1) {
398 si_rx_room_blk(si);
399 ret = 0;
400 break;
401 }
402 ofs += cnt + msg_len;
403 }
404
Willy Tarreau4781b152021-04-06 13:53:36 +0200405 HA_ATOMIC_INC(b_peek(buf, ofs));
Emeric Brun494c5052020-05-28 11:13:15 +0200406 ofs += ring->ofs;
407 sft->ofs = ofs;
408 }
409 HA_RWLOCK_RDUNLOCK(LOGSRV_LOCK, &ring->lock);
410
411 if (ret) {
412 /* let's be woken up once new data arrive */
413 HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
Willy Tarreau2b718102021-04-21 07:32:39 +0200414 LIST_APPEND(&ring->waiters, &appctx->wait_entry);
Emeric Brun494c5052020-05-28 11:13:15 +0200415 HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &ring->lock);
416 si_rx_endp_done(si);
417 }
418 HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
419
420 /* always drain data from server */
421 co_skip(si_oc(si), si_oc(si)->output);
422 return;
423
424close:
425 si_shutw(si);
426 si_shutr(si);
427 si_ic(si)->flags |= CF_READ_NULL;
428}
429
Emeric Brun97556472020-05-30 01:42:45 +0200430/*
431 * IO Handler to handle message push to syslog tcp server
432 * using octet counting frames
433 */
434static void sink_forward_oc_io_handler(struct appctx *appctx)
435{
Christopher Faulet53fa7872022-03-16 10:01:26 +0100436 struct stream_interface *si = cs_si(appctx->owner);
Emeric Brun97556472020-05-30 01:42:45 +0200437 struct stream *s = si_strm(si);
438 struct sink *sink = strm_fe(s)->parent;
439 struct sink_forward_target *sft = appctx->ctx.sft.ptr;
440 struct ring *ring = sink->ctx.ring;
441 struct buffer *buf = &ring->buf;
442 uint64_t msg_len;
443 size_t len, cnt, ofs;
444 int ret = 0;
445 char *p;
446
Ilya Shipitsin47d17182020-06-21 21:42:57 +0500447 /* if stopping was requested, close immediately */
Emeric Brun97556472020-05-30 01:42:45 +0200448 if (unlikely(stopping))
449 goto close;
450
451 /* for rex because it seems reset to timeout
452 * and we don't want expire on this case
453 * with a syslog server
454 */
455 si_oc(si)->rex = TICK_ETERNITY;
456 /* rto should not change but it seems the case */
457 si_oc(si)->rto = TICK_ETERNITY;
458
459 /* an error was detected */
460 if (unlikely(si_ic(si)->flags & (CF_WRITE_ERROR|CF_SHUTW)))
461 goto close;
462
463 /* con closed by server side */
464 if ((si_oc(si)->flags & CF_SHUTW))
465 goto close;
466
467 /* if the connection is not established, inform the stream that we want
468 * to be notified whenever the connection completes.
469 */
470 if (si_opposite(si)->state < SI_ST_EST) {
471 si_cant_get(si);
472 si_rx_conn_blk(si);
473 si_rx_endp_more(si);
474 return;
475 }
476
477 HA_SPIN_LOCK(SFT_LOCK, &sft->lock);
478 if (appctx != sft->appctx) {
479 HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
480 goto close;
481 }
482 ofs = sft->ofs;
483
484 HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
485 LIST_DEL_INIT(&appctx->wait_entry);
486 HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &ring->lock);
487
488 HA_RWLOCK_RDLOCK(LOGSRV_LOCK, &ring->lock);
489
490 /* explanation for the initialization below: it would be better to do
491 * this in the parsing function but this would occasionally result in
492 * dropped events because we'd take a reference on the oldest message
493 * and keep it while being scheduled. Thus instead let's take it the
494 * first time we enter here so that we have a chance to pass many
495 * existing messages before grabbing a reference to a location. This
496 * value cannot be produced after initialization.
497 */
498 if (unlikely(ofs == ~0)) {
499 ofs = 0;
500
Willy Tarreau4781b152021-04-06 13:53:36 +0200501 HA_ATOMIC_INC(b_peek(buf, ofs));
Emeric Brun97556472020-05-30 01:42:45 +0200502 ofs += ring->ofs;
503 }
504
Emeric Brun97556472020-05-30 01:42:45 +0200505 /* in this loop, ofs always points to the counter byte that precedes
506 * the message so that we can take our reference there if we have to
507 * stop before the end (ret=0).
508 */
509 if (si_opposite(si)->state == SI_ST_EST) {
Emeric Brunfdabf492020-12-02 17:02:09 +0100510 /* we were already there, adjust the offset to be relative to
511 * the buffer's head and remove us from the counter.
512 */
513 ofs -= ring->ofs;
514 BUG_ON(ofs >= buf->size);
Willy Tarreau4781b152021-04-06 13:53:36 +0200515 HA_ATOMIC_DEC(b_peek(buf, ofs));
Emeric Brunfdabf492020-12-02 17:02:09 +0100516
Emeric Brun97556472020-05-30 01:42:45 +0200517 ret = 1;
518 while (ofs + 1 < b_data(buf)) {
519 cnt = 1;
520 len = b_peek_varint(buf, ofs + cnt, &msg_len);
521 if (!len)
522 break;
523 cnt += len;
524 BUG_ON(msg_len + ofs + cnt + 1 > b_data(buf));
525
526 chunk_reset(&trash);
527 p = ulltoa(msg_len, trash.area, b_size(&trash));
528 if (p) {
529 trash.data = (p - trash.area) + 1;
530 *p = ' ';
531 }
532
533 if (!p || (trash.data + msg_len > b_size(&trash))) {
534 /* too large a message to ever fit, let's skip it */
535 ofs += cnt + msg_len;
536 continue;
537 }
538
539 trash.data += b_getblk(buf, p + 1, msg_len, ofs + cnt);
540
541 if (ci_putchk(si_ic(si), &trash) == -1) {
542 si_rx_room_blk(si);
543 ret = 0;
544 break;
545 }
546 ofs += cnt + msg_len;
547 }
548
Willy Tarreau4781b152021-04-06 13:53:36 +0200549 HA_ATOMIC_INC(b_peek(buf, ofs));
Emeric Brun97556472020-05-30 01:42:45 +0200550 ofs += ring->ofs;
551 sft->ofs = ofs;
552 }
553 HA_RWLOCK_RDUNLOCK(LOGSRV_LOCK, &ring->lock);
554
555 if (ret) {
556 /* let's be woken up once new data arrive */
557 HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &ring->lock);
Willy Tarreau2b718102021-04-21 07:32:39 +0200558 LIST_APPEND(&ring->waiters, &appctx->wait_entry);
Emeric Brun97556472020-05-30 01:42:45 +0200559 HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &ring->lock);
560 si_rx_endp_done(si);
561 }
562 HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
563
564 /* always drain data from server */
565 co_skip(si_oc(si), si_oc(si)->output);
566 return;
567
568close:
569 si_shutw(si);
570 si_shutr(si);
571 si_ic(si)->flags |= CF_READ_NULL;
572}
573
Emeric Brun494c5052020-05-28 11:13:15 +0200574void __sink_forward_session_deinit(struct sink_forward_target *sft)
575{
576 struct stream_interface *si;
577 struct stream *s;
578 struct sink *sink;
579
580 if (!sft->appctx)
581 return;
582
Christopher Faulet53fa7872022-03-16 10:01:26 +0100583 si = cs_si(sft->appctx->owner);
Emeric Brun494c5052020-05-28 11:13:15 +0200584 if (!si)
585 return;
586
587 s = si_strm(si);
588 if (!s)
589 return;
590
591 sink = strm_fe(s)->parent;
592 if (!sink)
593 return;
594
595 HA_RWLOCK_WRLOCK(LOGSRV_LOCK, &sink->ctx.ring->lock);
596 LIST_DEL_INIT(&sft->appctx->wait_entry);
597 HA_RWLOCK_WRUNLOCK(LOGSRV_LOCK, &sink->ctx.ring->lock);
598
599 sft->appctx = NULL;
600 task_wakeup(sink->forward_task, TASK_WOKEN_MSG);
601}
602
603
604static void sink_forward_session_release(struct appctx *appctx)
605{
Christopher Fauletdd0b1442022-01-14 15:03:22 +0100606 struct sink_forward_target *sft = appctx->ctx.sft.ptr;
Emeric Brun494c5052020-05-28 11:13:15 +0200607
608 if (!sft)
609 return;
610
611 HA_SPIN_LOCK(SFT_LOCK, &sft->lock);
612 if (sft->appctx == appctx)
613 __sink_forward_session_deinit(sft);
614 HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
615}
616
617static struct applet sink_forward_applet = {
618 .obj_type = OBJ_TYPE_APPLET,
619 .name = "<SINKFWD>", /* used for logging */
620 .fct = sink_forward_io_handler,
621 .release = sink_forward_session_release,
622};
623
Emeric Brun97556472020-05-30 01:42:45 +0200624static struct applet sink_forward_oc_applet = {
625 .obj_type = OBJ_TYPE_APPLET,
626 .name = "<SINKFWDOC>", /* used for logging */
627 .fct = sink_forward_oc_io_handler,
628 .release = sink_forward_session_release,
629};
630
Emeric Brun494c5052020-05-28 11:13:15 +0200631/*
632 * Create a new peer session in assigned state (connect will start automatically)
633 */
634static struct appctx *sink_forward_session_create(struct sink *sink, struct sink_forward_target *sft)
635{
636 struct proxy *p = sink->forward_px;
637 struct appctx *appctx;
638 struct session *sess;
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100639 struct cs_endpoint *endp;
Christopher Faulet13a35e52021-12-20 15:34:16 +0100640 struct conn_stream *cs;
Emeric Brun494c5052020-05-28 11:13:15 +0200641 struct stream *s;
Emeric Brun97556472020-05-30 01:42:45 +0200642 struct applet *applet = &sink_forward_applet;
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100643 struct sockaddr_storage *addr = NULL;
Emeric Brun97556472020-05-30 01:42:45 +0200644
645 if (sft->srv->log_proto == SRV_LOG_PROTO_OCTET_COUNTING)
646 applet = &sink_forward_oc_applet;
Emeric Brun494c5052020-05-28 11:13:15 +0200647
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100648 appctx = appctx_new(applet);
Christopher Faulet2479e5f2022-01-19 14:50:11 +0100649 if (!appctx)
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100650 goto out_close;
Emeric Brun494c5052020-05-28 11:13:15 +0200651
652 appctx->ctx.sft.ptr = (void *)sft;
653
654 sess = session_new(p, NULL, &appctx->obj_type);
655 if (!sess) {
Christopher Faulet13a35e52021-12-20 15:34:16 +0100656 ha_alert("out of memory in sink_forward_session_create().\n");
Emeric Brun494c5052020-05-28 11:13:15 +0200657 goto out_free_appctx;
658 }
659
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100660 if (!sockaddr_alloc(&addr, &sft->srv->addr, sizeof(sft->srv->addr)))
Christopher Faulet2479e5f2022-01-19 14:50:11 +0100661 goto out_free_sess;
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100662
663 endp = cs_endpoint_new();
664 if (!endp)
665 goto out_free_addr;
666 endp->target = appctx;
667 endp->ctx = appctx;
668 endp->flags |= CS_EP_T_APPLET;
669
670 cs = cs_new_from_applet(endp, sess, &BUF_NULL);
671 if (!cs) {
672 ha_alert("Failed to initialize stream in sink_forward_session_create().\n");
673 cs_endpoint_free(endp);
674 goto out_free_addr;
Christopher Faulet13a35e52021-12-20 15:34:16 +0100675 }
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100676 s = DISGUISE(cs_strm(cs));
Christopher Faulet13a35e52021-12-20 15:34:16 +0100677
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100678 cs_si(s->csb)->dst = addr;
679 cs_si(s->csb)->flags |= SI_FL_NOLINGER;
Emeric Brun494c5052020-05-28 11:13:15 +0200680
681 s->target = &sft->srv->obj_type;
Emeric Brun494c5052020-05-28 11:13:15 +0200682 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun494c5052020-05-28 11:13:15 +0200683
684 s->do_log = NULL;
685 s->uniq_id = 0;
686
687 s->res.flags |= CF_READ_DONTWAIT;
688 /* for rto and rex to eternity to not expire on idle recv:
689 * We are using a syslog server.
690 */
691 s->res.rto = TICK_ETERNITY;
692 s->res.rex = TICK_ETERNITY;
693 sft->appctx = appctx;
Emeric Brun494c5052020-05-28 11:13:15 +0200694 return appctx;
695
696 /* Error unrolling */
Christopher Fauleta9e8b392022-03-23 11:01:09 +0100697 out_free_addr:
698 sockaddr_free(&addr);
Emeric Brun494c5052020-05-28 11:13:15 +0200699 out_free_sess:
700 session_free(sess);
701 out_free_appctx:
702 appctx_free(appctx);
703 out_close:
704 return NULL;
705}
706
707/*
708 * Task to handle connctions to forward servers
709 */
Willy Tarreau144f84a2021-03-02 16:09:26 +0100710static struct task *process_sink_forward(struct task * task, void *context, unsigned int state)
Emeric Brun494c5052020-05-28 11:13:15 +0200711{
712 struct sink *sink = (struct sink *)context;
713 struct sink_forward_target *sft = sink->sft;
714
715 task->expire = TICK_ETERNITY;
716
717 if (!stopping) {
718 while (sft) {
719 HA_SPIN_LOCK(SFT_LOCK, &sft->lock);
720 /* if appctx is NULL, start a new session */
721 if (!sft->appctx)
722 sft->appctx = sink_forward_session_create(sink, sft);
723 HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
724 sft = sft->next;
725 }
726 }
727 else {
728 while (sft) {
729 HA_SPIN_LOCK(SFT_LOCK, &sft->lock);
730 /* awake applet to perform a clean close */
731 if (sft->appctx)
732 appctx_wakeup(sft->appctx);
733 HA_SPIN_UNLOCK(SFT_LOCK, &sft->lock);
734 sft = sft->next;
735 }
736 }
737
738 return task;
739}
740/*
741 * Init task to manage connctions to forward servers
742 *
743 * returns 0 in case of error.
744 */
745int sink_init_forward(struct sink *sink)
746{
Willy Tarreaubeeabf52021-10-01 18:23:30 +0200747 sink->forward_task = task_new_anywhere();
Emeric Brun494c5052020-05-28 11:13:15 +0200748 if (!sink->forward_task)
749 return 0;
750
751 sink->forward_task->process = process_sink_forward;
752 sink->forward_task->context = (void *)sink;
753 sink->forward_sighandler = signal_register_task(0, sink->forward_task, 0);
754 task_wakeup(sink->forward_task, TASK_WOKEN_INIT);
755 return 1;
756}
Emeric Brun99c453d2020-05-25 15:01:04 +0200757/*
758 * Parse "ring" section and create corresponding sink buffer.
759 *
760 * The function returns 0 in success case, otherwise, it returns error
761 * flags.
762 */
763int cfg_parse_ring(const char *file, int linenum, char **args, int kwm)
764{
765 int err_code = 0;
766 const char *inv;
767 size_t size = BUFSIZE;
Emeric Brun494c5052020-05-28 11:13:15 +0200768 struct proxy *p;
Emeric Brun99c453d2020-05-25 15:01:04 +0200769
770 if (strcmp(args[0], "ring") == 0) { /* new peers section */
771 if (!*args[1]) {
772 ha_alert("parsing [%s:%d] : missing ring name.\n", file, linenum);
773 err_code |= ERR_ALERT | ERR_FATAL;
774 goto err;
775 }
776
777 inv = invalid_char(args[1]);
778 if (inv) {
779 ha_alert("parsing [%s:%d] : invalid ring name '%s' (character '%c' is not permitted).\n", file, linenum, args[1], *inv);
780 err_code |= ERR_ALERT | ERR_FATAL;
781 goto err;
782 }
783
784 if (sink_find(args[1])) {
785 ha_alert("parsing [%s:%d] : sink named '%s' already exists.\n", file, linenum, args[1]);
786 err_code |= ERR_ALERT | ERR_FATAL;
787 goto err;
788 }
789
Emeric Brun54648852020-07-06 15:54:06 +0200790 cfg_sink = sink_new_buf(args[1], args[1], LOG_FORMAT_RAW, size);
Emeric Brun99c453d2020-05-25 15:01:04 +0200791 if (!cfg_sink || cfg_sink->type != SINK_TYPE_BUFFER) {
792 ha_alert("parsing [%s:%d] : unable to create a new sink buffer for ring '%s'.\n", file, linenum, args[1]);
793 err_code |= ERR_ALERT | ERR_FATAL;
794 goto err;
795 }
Emeric Brun494c5052020-05-28 11:13:15 +0200796
797 /* allocate new proxy to handle forwards */
798 p = calloc(1, sizeof *p);
799 if (!p) {
800 ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
801 err_code |= ERR_ALERT | ERR_FATAL;
802 goto err;
803 }
804
805 init_new_proxy(p);
806 sink_setup_proxy(p);
807 p->parent = cfg_sink;
808 p->id = strdup(args[1]);
809 p->conf.args.file = p->conf.file = strdup(file);
810 p->conf.args.line = p->conf.line = linenum;
811 cfg_sink->forward_px = p;
Emeric Brun99c453d2020-05-25 15:01:04 +0200812 }
813 else if (strcmp(args[0], "size") == 0) {
814 size = atol(args[1]);
815 if (!size) {
816 ha_alert("parsing [%s:%d] : invalid size '%s' for new sink buffer.\n", file, linenum, args[1]);
817 err_code |= ERR_ALERT | ERR_FATAL;
818 goto err;
819 }
820
821 if (!cfg_sink || (cfg_sink->type != SINK_TYPE_BUFFER)
822 || !ring_resize(cfg_sink->ctx.ring, size)) {
823 ha_alert("parsing [%s:%d] : fail to set sink buffer size '%s'.\n", file, linenum, args[1]);
824 err_code |= ERR_ALERT | ERR_FATAL;
825 goto err;
826 }
827 }
Emeric Brun494c5052020-05-28 11:13:15 +0200828 else if (strcmp(args[0],"server") == 0) {
Amaury Denoyelle30c05372021-03-08 16:36:46 +0100829 err_code |= parse_server(file, linenum, args, cfg_sink->forward_px, NULL,
830 SRV_PARSE_PARSE_ADDR|SRV_PARSE_INITIAL_RESOLVE);
Emeric Brun494c5052020-05-28 11:13:15 +0200831 }
832 else if (strcmp(args[0],"timeout") == 0) {
833 if (!cfg_sink || !cfg_sink->forward_px) {
834 ha_alert("parsing [%s:%d] : unable to set timeout '%s'.\n", file, linenum, args[1]);
835 err_code |= ERR_ALERT | ERR_FATAL;
836 goto err;
837 }
838
839 if (strcmp(args[1], "connect") == 0 ||
840 strcmp(args[1], "server") == 0) {
841 const char *res;
842 unsigned int tout;
843
844 if (!*args[2]) {
845 ha_alert("parsing [%s:%d] : '%s %s' expects <time> as argument.\n",
846 file, linenum, args[0], args[1]);
847 err_code |= ERR_ALERT | ERR_FATAL;
848 goto err;
849 }
850 res = parse_time_err(args[2], &tout, TIME_UNIT_MS);
851 if (res == PARSE_TIME_OVER) {
852 ha_alert("parsing [%s:%d]: timer overflow in argument <%s> to <%s %s>, maximum value is 2147483647 ms (~24.8 days).\n",
853 file, linenum, args[2], args[0], args[1]);
854 err_code |= ERR_ALERT | ERR_FATAL;
855 goto err;
856 }
857 else if (res == PARSE_TIME_UNDER) {
858 ha_alert("parsing [%s:%d]: timer underflow in argument <%s> to <%s %s>, minimum non-null value is 1 ms.\n",
859 file, linenum, args[2], args[0], args[1]);
860 err_code |= ERR_ALERT | ERR_FATAL;
861 goto err;
862 }
863 else if (res) {
864 ha_alert("parsing [%s:%d]: unexpected character '%c' in argument to <%s %s>.\n",
865 file, linenum, *res, args[0], args[1]);
866 err_code |= ERR_ALERT | ERR_FATAL;
867 goto err;
868 }
869 if (args[1][2] == 'c')
870 cfg_sink->forward_px->timeout.connect = tout;
871 else
872 cfg_sink->forward_px->timeout.server = tout;
873 }
874 }
Emeric Brun99c453d2020-05-25 15:01:04 +0200875 else if (strcmp(args[0],"format") == 0) {
876 if (!cfg_sink) {
877 ha_alert("parsing [%s:%d] : unable to set format '%s'.\n", file, linenum, args[1]);
878 err_code |= ERR_ALERT | ERR_FATAL;
879 goto err;
880 }
881
Emeric Brun54648852020-07-06 15:54:06 +0200882 cfg_sink->fmt = get_log_format(args[1]);
883 if (cfg_sink->fmt == LOG_FORMAT_UNSPEC) {
Emeric Brun99c453d2020-05-25 15:01:04 +0200884 ha_alert("parsing [%s:%d] : unknown format '%s'.\n", file, linenum, args[1]);
885 err_code |= ERR_ALERT | ERR_FATAL;
886 goto err;
887 }
888 }
889 else if (strcmp(args[0],"maxlen") == 0) {
890 if (!cfg_sink) {
891 ha_alert("parsing [%s:%d] : unable to set event max length '%s'.\n", file, linenum, args[1]);
892 err_code |= ERR_ALERT | ERR_FATAL;
893 goto err;
894 }
895
896 cfg_sink->maxlen = atol(args[1]);
897 if (!cfg_sink->maxlen) {
898 ha_alert("parsing [%s:%d] : invalid size '%s' for new sink buffer.\n", file, linenum, args[1]);
899 err_code |= ERR_ALERT | ERR_FATAL;
900 goto err;
901 }
902 }
903 else if (strcmp(args[0],"description") == 0) {
904 if (!cfg_sink) {
905 ha_alert("parsing [%s:%d] : unable to set description '%s'.\n", file, linenum, args[1]);
906 err_code |= ERR_ALERT | ERR_FATAL;
907 goto err;
908 }
909
910 if (!*args[1]) {
911 ha_alert("parsing [%s:%d] : missing ring description text.\n", file, linenum);
912 err_code |= ERR_ALERT | ERR_FATAL;
913 goto err;
914 }
915
916 free(cfg_sink->desc);
917
918 cfg_sink->desc = strdup(args[1]);
919 if (!cfg_sink->desc) {
920 ha_alert("parsing [%s:%d] : fail to set description '%s'.\n", file, linenum, args[1]);
921 err_code |= ERR_ALERT | ERR_FATAL;
922 goto err;
923 }
924 }
Emeric Brun9f2ff3a2020-05-29 15:47:52 +0200925 else {
926 ha_alert("parsing [%s:%d] : unknown statement '%s'.\n", file, linenum, args[0]);
927 err_code |= ERR_ALERT | ERR_FATAL;
928 goto err;
929 }
Emeric Brun99c453d2020-05-25 15:01:04 +0200930
931err:
932 return err_code;
933}
934
Emeric Brun94aab062021-04-02 10:41:36 +0200935/* Creates an new sink buffer from a log server.
936 *
937 * It uses the logsrvaddress to declare a forward
938 * server for this buffer. And it initializes the
939 * forwarding.
940 *
941 * The function returns a pointer on the
942 * allocated struct sink if allocate
943 * and initialize succeed, else if it fails
944 * it returns NULL.
945 *
946 * Note: the sink is created using the name
Ilya Shipitsinb2be9a12021-04-24 13:25:42 +0500947 * specified into logsrv->ring_name
Emeric Brun94aab062021-04-02 10:41:36 +0200948 */
949struct sink *sink_new_from_logsrv(struct logsrv *logsrv)
950{
951 struct proxy *p = NULL;
952 struct sink *sink = NULL;
953 struct server *srv = NULL;
954 struct sink_forward_target *sft = NULL;
Emeric Brun94aab062021-04-02 10:41:36 +0200955
956 /* allocate new proxy to handle
957 * forward to a stream server
958 */
959 p = calloc(1, sizeof *p);
960 if (!p) {
961 goto error;
962 }
963
964 init_new_proxy(p);
965 sink_setup_proxy(p);
966 p->id = strdup(logsrv->ring_name);
967 p->conf.args.file = p->conf.file = strdup(logsrv->conf.file);
968 p->conf.args.line = p->conf.line = logsrv->conf.line;
969
970 /* allocate a new server to forward messages
971 * from ring buffer
972 */
973 srv = new_server(p);
974 if (!srv)
975 goto error;
976
977 /* init server */
978 srv->id = strdup(logsrv->ring_name);
979 srv->conf.file = strdup(logsrv->conf.file);
980 srv->conf.line = logsrv->conf.line;
981 srv->addr = logsrv->addr;
982 srv->svc_port = get_host_port(&logsrv->addr);
983 HA_SPIN_INIT(&srv->lock);
984
985 /* process per thread init */
Miroslav Zagorac8a8f2702021-06-15 15:33:20 +0200986 if (srv_init_per_thr(srv) == -1)
Emeric Brun94aab062021-04-02 10:41:36 +0200987 goto error;
988
Emeric Brun94aab062021-04-02 10:41:36 +0200989 /* the servers are linked backwards
990 * first into proxy
991 */
992 p->srv = srv;
993 srv->next = p->srv;
994
995 /* allocate sink_forward_target descriptor */
996 sft = calloc(1, sizeof(*sft));
997 if (!sft)
998 goto error;
999
1000 /* init sink_forward_target offset */
1001 sft->srv = srv;
1002 sft->appctx = NULL;
1003 sft->ofs = ~0;
1004 HA_SPIN_INIT(&sft->lock);
1005
Ilya Shipitsinb2be9a12021-04-24 13:25:42 +05001006 /* prepare description for the sink */
Emeric Brun94aab062021-04-02 10:41:36 +02001007 chunk_reset(&trash);
1008 chunk_printf(&trash, "created from logserver declared into '%s' at line %d", logsrv->conf.file, logsrv->conf.line);
1009
1010 /* allocate a new sink buffer */
1011 sink = sink_new_buf(logsrv->ring_name, trash.area, logsrv->format, BUFSIZE);
1012 if (!sink || sink->type != SINK_TYPE_BUFFER) {
1013 goto error;
1014 }
1015
1016 /* link sink_forward_target to proxy */
1017 sink->forward_px = p;
1018 p->parent = sink;
1019
1020 /* insert into sink_forward_targets
1021 * list into sink
1022 */
1023 sft->next = sink->sft;
1024 sink->sft = sft;
1025
1026 /* mark server as an attached reader to the ring */
1027 if (!ring_attach(sink->ctx.ring)) {
1028 /* should never fail since there is
1029 * only one reader
1030 */
1031 goto error;
1032 }
1033
1034 /* initialize sink buffer forwarding */
1035 if (!sink_init_forward(sink))
1036 goto error;
1037
1038 /* reset familyt of logsrv to consider the ring buffer target */
1039 logsrv->addr.ss_family = AF_UNSPEC;
1040
1041 return sink;
1042error:
1043 if (p) {
1044 if (p->id)
1045 free(p->id);
1046 if (p->conf.file)
1047 free(p->conf.file);
1048
1049 free(p);
1050 }
1051
1052 if (srv) {
1053 if (srv->id)
1054 free(srv->id);
1055 if (srv->conf.file)
1056 free((void *)srv->conf.file);
1057 if (srv->per_thr)
1058 free(srv->per_thr);
1059 free(srv);
1060 }
1061
1062 if (sft)
1063 free(sft);
1064
1065 if (sink) {
1066 if (sink->ctx.ring)
1067 ring_free(sink->ctx.ring);
1068
Willy Tarreau2b718102021-04-21 07:32:39 +02001069 LIST_DELETE(&sink->sink_list);
Emeric Brun94aab062021-04-02 10:41:36 +02001070 free(sink->name);
1071 free(sink->desc);
1072 free(sink);
1073 }
1074
1075 return NULL;
1076}
1077
Emeric Brun99c453d2020-05-25 15:01:04 +02001078/*
1079 * Post parsing "ring" section.
1080 *
1081 * The function returns 0 in success case, otherwise, it returns error
1082 * flags.
1083 */
1084int cfg_post_parse_ring()
1085{
1086 int err_code = 0;
Emeric Brun494c5052020-05-28 11:13:15 +02001087 struct server *srv;
Emeric Brun99c453d2020-05-25 15:01:04 +02001088
1089 if (cfg_sink && (cfg_sink->type == SINK_TYPE_BUFFER)) {
1090 if (cfg_sink->maxlen > b_size(&cfg_sink->ctx.ring->buf)) {
1091 ha_warning("ring '%s' event max length '%u' exceeds size, forced to size '%lu'.\n",
Willy Tarreau570a22b2020-06-02 12:00:46 +02001092 cfg_sink->name, cfg_sink->maxlen, (unsigned long)b_size(&cfg_sink->ctx.ring->buf));
Emeric Brun99c453d2020-05-25 15:01:04 +02001093 cfg_sink->maxlen = b_size(&cfg_sink->ctx.ring->buf);
1094 err_code |= ERR_ALERT;
1095 }
Emeric Brun494c5052020-05-28 11:13:15 +02001096
1097 /* prepare forward server descriptors */
1098 if (cfg_sink->forward_px) {
1099 srv = cfg_sink->forward_px->srv;
1100 while (srv) {
1101 struct sink_forward_target *sft;
1102 /* init ssl if needed */
1103 if (srv->use_ssl == 1 && xprt_get(XPRT_SSL) && xprt_get(XPRT_SSL)->prepare_srv) {
1104 if (xprt_get(XPRT_SSL)->prepare_srv(srv)) {
1105 ha_alert("unable to prepare SSL for server '%s' in ring '%s'.\n", srv->id, cfg_sink->name);
1106 err_code |= ERR_ALERT | ERR_FATAL;
1107 }
1108 }
Emeric Brun99c453d2020-05-25 15:01:04 +02001109
Emeric Brun494c5052020-05-28 11:13:15 +02001110 /* allocate sink_forward_target descriptor */
1111 sft = calloc(1, sizeof(*sft));
1112 if (!sft) {
1113 ha_alert("memory allocation error initializing server '%s' in ring '%s'.\n",srv->id, cfg_sink->name);
1114 err_code |= ERR_ALERT | ERR_FATAL;
1115 break;
1116 }
1117 sft->srv = srv;
1118 sft->appctx = NULL;
1119 sft->ofs = ~0; /* init ring offset */
1120 sft->next = cfg_sink->sft;
1121 HA_SPIN_INIT(&sft->lock);
1122
1123 /* mark server attached to the ring */
1124 if (!ring_attach(cfg_sink->ctx.ring)) {
1125 ha_alert("server '%s' sets too many watchers > 255 on ring '%s'.\n", srv->id, cfg_sink->name);
1126 err_code |= ERR_ALERT | ERR_FATAL;
1127 }
1128 cfg_sink->sft = sft;
1129 srv = srv->next;
1130 }
1131 sink_init_forward(cfg_sink);
1132 }
1133 }
Emeric Brun99c453d2020-05-25 15:01:04 +02001134 cfg_sink = NULL;
1135
1136 return err_code;
1137}
1138
1139/* resolve sink names at end of config. Returns 0 on success otherwise error
1140 * flags.
1141*/
1142int post_sink_resolve()
1143{
Christopher Fauletfc633b62020-11-06 15:24:23 +01001144 int err_code = ERR_NONE;
Emeric Brun99c453d2020-05-25 15:01:04 +02001145 struct logsrv *logsrv, *logb;
1146 struct sink *sink;
1147 struct proxy *px;
1148
1149 list_for_each_entry_safe(logsrv, logb, &global.logsrvs, list) {
1150 if (logsrv->type == LOG_TARGET_BUFFER) {
1151 sink = sink_find(logsrv->ring_name);
Emeric Brun94aab062021-04-02 10:41:36 +02001152 if (!sink) {
1153 /* LOG_TARGET_BUFFER but !AF_UNSPEC
1154 * means we must allocate a sink
1155 * buffer to send messages to this logsrv
1156 */
1157 if (logsrv->addr.ss_family != AF_UNSPEC) {
1158 sink = sink_new_from_logsrv(logsrv);
1159 if (!sink) {
1160 ha_alert("global stream log server declared in file '%s' at line %d cannot be initialized'.\n",
1161 logsrv->conf.file, logsrv->conf.line);
1162 err_code |= ERR_ALERT | ERR_FATAL;
1163 }
1164 }
1165 else {
1166 ha_alert("global log server declared in file '%s' at line %d uses unknown ring named '%s'.\n",
1167 logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
1168 err_code |= ERR_ALERT | ERR_FATAL;
1169 }
1170 }
1171 else if (sink->type != SINK_TYPE_BUFFER) {
1172 ha_alert("global log server declared in file '%s' at line %d uses incompatible ring '%s'.\n",
1173 logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
Emeric Brun99c453d2020-05-25 15:01:04 +02001174 err_code |= ERR_ALERT | ERR_FATAL;
1175 }
1176 logsrv->sink = sink;
1177 }
Emeric Brun94aab062021-04-02 10:41:36 +02001178
Emeric Brun99c453d2020-05-25 15:01:04 +02001179 }
1180
1181 for (px = proxies_list; px; px = px->next) {
1182 list_for_each_entry_safe(logsrv, logb, &px->logsrvs, list) {
1183 if (logsrv->type == LOG_TARGET_BUFFER) {
1184 sink = sink_find(logsrv->ring_name);
Emeric Brun94aab062021-04-02 10:41:36 +02001185 if (!sink) {
1186 /* LOG_TARGET_BUFFER but !AF_UNSPEC
1187 * means we must allocate a sink
1188 * buffer to send messages to this logsrv
1189 */
1190 if (logsrv->addr.ss_family != AF_UNSPEC) {
1191 sink = sink_new_from_logsrv(logsrv);
1192 if (!sink) {
1193 ha_alert("log server declared in proxy section '%s' file '%s' at line %d cannot be initialized'.\n",
1194 px->id, logsrv->conf.file, logsrv->conf.line);
1195 err_code |= ERR_ALERT | ERR_FATAL;
1196 }
1197 }
1198 else {
1199 ha_alert("log server declared in proxy section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n",
1200 px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
1201 err_code |= ERR_ALERT | ERR_FATAL;
1202 }
1203 }
1204 else if (sink->type != SINK_TYPE_BUFFER) {
1205 ha_alert("log server declared in proxy section '%s' in file '%s' at line %d uses incomatible ring named '%s'.\n",
1206 px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
Emeric Brun99c453d2020-05-25 15:01:04 +02001207 err_code |= ERR_ALERT | ERR_FATAL;
1208 }
1209 logsrv->sink = sink;
1210 }
1211 }
1212 }
Emeric Brun12941c82020-07-07 14:19:42 +02001213
1214 for (px = cfg_log_forward; px; px = px->next) {
1215 list_for_each_entry_safe(logsrv, logb, &px->logsrvs, list) {
1216 if (logsrv->type == LOG_TARGET_BUFFER) {
1217 sink = sink_find(logsrv->ring_name);
Emeric Brun94aab062021-04-02 10:41:36 +02001218 if (!sink) {
1219 /* LOG_TARGET_BUFFER but !AF_UNSPEC
1220 * means we must allocate a sink
1221 * buffer to send messages to this logsrv
1222 */
1223 if (logsrv->addr.ss_family != AF_UNSPEC) {
1224 sink = sink_new_from_logsrv(logsrv);
1225 if (!sink) {
1226 ha_alert("log server declared in log-forward section '%s' file '%s' at line %d cannot be initialized'.\n",
1227 px->id, logsrv->conf.file, logsrv->conf.line);
1228 err_code |= ERR_ALERT | ERR_FATAL;
1229 }
1230 }
1231 else {
1232 ha_alert("log server declared in log-forward section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n",
1233 px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
1234 err_code |= ERR_ALERT | ERR_FATAL;
1235 }
1236 }
1237 else if (sink->type != SINK_TYPE_BUFFER) {
1238 ha_alert("log server declared in log-forward section '%s' in file '%s' at line %d uses unknown ring named '%s'.\n",
1239 px->id, logsrv->conf.file, logsrv->conf.line, logsrv->ring_name);
Emeric Brun12941c82020-07-07 14:19:42 +02001240 err_code |= ERR_ALERT | ERR_FATAL;
1241 }
1242 logsrv->sink = sink;
1243 }
1244 }
1245 }
Emeric Brun99c453d2020-05-25 15:01:04 +02001246 return err_code;
1247}
1248
1249
Willy Tarreau973e6622019-08-20 11:57:52 +02001250static void sink_init()
1251{
Emeric Brun54648852020-07-06 15:54:06 +02001252 sink_new_fd("stdout", "standard output (fd#1)", LOG_FORMAT_RAW, 1);
1253 sink_new_fd("stderr", "standard output (fd#2)", LOG_FORMAT_RAW, 2);
1254 sink_new_buf("buf0", "in-memory ring buffer", LOG_FORMAT_TIMED, 1048576);
Willy Tarreau4ed23ca2019-08-23 15:47:49 +02001255}
1256
1257static void sink_deinit()
1258{
1259 struct sink *sink, *sb;
1260
1261 list_for_each_entry_safe(sink, sb, &sink_list, sink_list) {
1262 if (sink->type == SINK_TYPE_BUFFER)
1263 ring_free(sink->ctx.ring);
Willy Tarreau2b718102021-04-21 07:32:39 +02001264 LIST_DELETE(&sink->sink_list);
Emeric Brun99c453d2020-05-25 15:01:04 +02001265 free(sink->name);
1266 free(sink->desc);
Willy Tarreau4ed23ca2019-08-23 15:47:49 +02001267 free(sink);
1268 }
Willy Tarreau973e6622019-08-20 11:57:52 +02001269}
1270
1271INITCALL0(STG_REGISTER, sink_init);
Willy Tarreau4ed23ca2019-08-23 15:47:49 +02001272REGISTER_POST_DEINIT(sink_deinit);
Willy Tarreau973e6622019-08-20 11:57:52 +02001273
Willy Tarreau9f830d72019-08-26 18:17:04 +02001274static struct cli_kw_list cli_kws = {{ },{
Willy Tarreaub205bfd2021-05-07 11:38:37 +02001275 { { "show", "events", NULL }, "show events [<sink>] [-w] [-n] : show event sink state", cli_parse_show_events, NULL, NULL },
Willy Tarreau9f830d72019-08-26 18:17:04 +02001276 {{},}
1277}};
1278
1279INITCALL1(STG_REGISTER, cli_register_kw, &cli_kws);
1280
Emeric Brun99c453d2020-05-25 15:01:04 +02001281/* config parsers for this section */
1282REGISTER_CONFIG_SECTION("ring", cfg_parse_ring, cfg_post_parse_ring);
1283REGISTER_POST_CHECK(post_sink_resolve);
1284
Willy Tarreau67b5a162019-08-11 16:38:56 +02001285/*
1286 * Local variables:
1287 * c-indent-level: 8
1288 * c-basic-offset: 8
1289 * End:
1290 */