src/logger/log_buffer.c

   1 #include "log_buffer.h"
   2 #include "log_compressed_storage.h"
   3 #include "logger_internal.h"
   4 #include "subreader_dlogutil.h"
   5 #include "reader_memory.h"
   6 #include <ptrs_list.h>
   7
   8 /**
   9  * @brief Append to buffer
  10  * @details Appends an entry to the buffer
  11  * @param[in] s The logger server
  12  * @param[in] entry The entry to append
  13  * @param[in] b The buffer whither to append
  14  */
  15 int buffer_append(const dlogutil_entry_s *entry, struct log_buffer *b)
  16 {
  17         if (!log_storage_add_new_entry(b->log_storage_ptr, entry))
  18                 return -ENOMEM;
  19         return 0;
  20 }
  21
  22 /**
  23  * @brief Service pipe log data
  24  * @details Handle log messages incoming through a pipe
  25  * @param[in] server The logger server
  26  * @param[in] wr The writer who sent the logs
  27  * @param[in] event The event associated with the data
  28  * @return 0 on success, else -errno
  29  */
  30 static int service_writer_pipe(struct logger *server, struct writer *wr, struct epoll_event *event)
  31 {
  32         if (~event->events & EPOLLIN) {
  33                 return event->events & EPOLLHUP
  34                         ? -EBADF
  35                         : 0
  36                 ;
  37         }
  38
  39         int r = read(wr->fd_entity.fd, wr->buffer + wr->readed, sizeof wr->buffer - wr->readed);
  40         if (r == -1 && errno == EAGAIN)
  41                 return 0;
  42         else if ((r == 0 || r == -1) && event->events & EPOLLHUP)
  43                 return -EINVAL;
  44         else if (r == 0)
  45                 return -EBADF;
  46
  47         wr->readed += r;
  48
  49         struct pipe_logger_entry *const ple = (struct pipe_logger_entry *const)wr->buffer;
  50         while ((wr->readed >= sizeof(ple->len)) && (ple->len <= wr->readed)) {
  51                 const int payload_size = ple->len - sizeof *ple;
  52                 if (payload_size <= 0 || payload_size > LOG_MAX_PAYLOAD_SIZE)
  53                         return -EINVAL;
  54
  55                 struct dlogutil_entry_with_msg lem;
  56                 parse_pipe_message(ple, &lem.header, ple->len);
  57
  58                 struct now_t now;
  59                 r = get_now(&now);
  60                 if (r < 0)
  61                         return r;
  62                 add_recv_timestamp(&lem.header, now);
  63
  64                 fixup_pipe_msg(&lem, payload_size);
  65                 if (server->qos)
  66                         qos_add_log(server->qos, &lem.header);
  67                 r = buffer_append(&lem.header, wr->buf_ptr);
  68                 wr->readed -= ple->len;
  69                 memmove(wr->buffer, wr->buffer + ple->len, sizeof wr->buffer - ple->len);
  70
  71                 if (r)
  72                         return r;
  73         }
  74
  75         return 0;
  76 }
  77
  78 static int service_writer_stdout(struct logger *server, struct writer *wr, struct epoll_event *event)
  79 {
  80         assert(server);
  81         assert(wr);
  82         assert(wr->stdout_data);
  83         assert(event);
  84
  85         int ret = 0;
  86         char *eol = NULL;
  87         char *terminal = NULL;
  88
  89         if (event->events & EPOLLIN) {
  90                 // The - 1 is important, since we want to add a zero at the end.
  91                 int r = read(wr->fd_entity.fd, wr->buffer + wr->readed, sizeof wr->buffer - wr->readed - 1);
  92
  93                 if (r == -1 && errno == EAGAIN)
  94                         return 0;
  95                 else if ((r == 0 || r == -1) && event->events & EPOLLHUP)
  96                         ret = -EINVAL;
  97                 else if (r == 0)
  98                         ret = -EBADF;
  99                 else if (!wr->buf_ptr->accept_stdout)
 100                         return 0;
 101                 else {
 102                         terminal = wr->buffer + wr->readed + r;
 103                         *terminal = '\0';
 104                         eol = strchrnul(wr->buffer + wr->readed, '\n');
 105                         wr->readed += r;
 106                 }
 107         } else if (event->events & EPOLLHUP)
 108                 ret = -EBADF;
 109
 110         // If we will return something that's not zero, we need to flush the entire buffer.
 111         // We do this by pretending there is an extra \0 at the end of the input.
 112         // Worth noting that if we got here, there can't be a \0 or \n in the input already
 113         // (because if there were, the previous iteration would take care of it).
 114         if (ret != 0) {
 115                 if (wr->readed > 0) {
 116                         eol = wr->buffer + wr->readed;
 117                         *eol = '\0';
 118                         wr->readed += 1;
 119
 120                         assert(wr->readed <= sizeof(wr->buffer) - 1);
 121                         terminal = wr->buffer + wr->readed;
 122                         *terminal = '\0';
 123                 }
 124                 else
 125                         return ret;
 126         }
 127
 128         assert(eol);
 129         assert(terminal);
 130
 131         // We need to handle the case in which the buffer is full (and again, - 1 is important).
 132         while (eol < terminal || wr->readed == sizeof(wr->buffer) - 1) {
 133                 *eol = '\0';
 134
 135                 struct dlogutil_entry_with_msg lem = {
 136                         .header = {
 137                                 .len = sizeof(lem.header),
 138                                 .priority = wr->stdout_data->prio,
 139                                 .pid = wr->stdout_data->pid,
 140                                 .tid = wr->stdout_data->pid,
 141                         },
 142                 };
 143
 144                 struct now_t now;
 145                 int r = get_now(&now);
 146                 if (r < 0)
 147                         return r;
 148                 add_recv_timestamp(&lem.header, now);
 149
 150                 /* HACK: We copy recv to sent, because we assume that those timestamps are available
 151                  * and we do not want to special case them. TODO: Is there a better solution? */
 152                 lem.header.sec_sent_mono = lem.header.sec_recv_mono;
 153                 lem.header.sec_sent_real = lem.header.sec_recv_real;
 154                 lem.header.nsec_sent_mono = lem.header.nsec_recv_mono;
 155                 lem.header.nsec_sent_real = lem.header.nsec_recv_real;
 156
 157                 lem.header.tag_len = strlen(strncpy((char *)&lem + lem.header.len, wr->stdout_data->tag, sizeof(lem) - lem.header.len - 1));
 158                 lem.header.len += lem.header.tag_len + 1;
 159                 assert(lem.header.len <= sizeof(lem));
 160                 assert(((char *)&lem)[lem.header.len - 1] == '\0');
 161                 int cut;
 162                 if (sizeof(lem) - lem.header.len - 1 >= 1) {
 163                         cut = strlen(strncpy((char *)&lem + lem.header.len, wr->buffer, sizeof(lem) - lem.header.len - 1));
 164                         lem.header.len += cut + 1;
 165                 }
 166                 else
 167                         return -EINVAL;
 168                 assert(lem.header.len <= sizeof(lem));
 169                 assert(((char *)&lem)[lem.header.len - 1] == '\0');
 170
 171                 if (server->qos)
 172                         qos_add_log(server->qos, &lem.header);
 173
 174                 r = buffer_append(&lem.header, wr->buf_ptr);
 175                 if (r != 0)
 176                         return r;
 177
 178                 if (wr->buffer[cut] == '\0')
 179                         cut += 1;
 180                 if (cut == sizeof(wr->buffer)) {
 181                         // This will happen if the buffer was full.
 182                         wr->readed = 0;
 183                         wr->buffer[0] = '\0';
 184                         break;
 185                 }
 186                 wr->readed -= cut;
 187                 terminal -= cut;
 188                 assert(terminal == wr->buffer + wr->readed);
 189                 memmove(wr->buffer, wr->buffer + cut, sizeof(wr->buffer) - cut);
 190                 assert(*terminal == '\0');
 191
 192                 eol = strchrnul(wr->buffer, '\n');
 193         }
 194
 195         return ret;
 196 }
 197
 198 static bool is_control_msg_valid_for_util_req(const struct dlog_control_msg *msg)
 199 {
 200         return msg->length >  sizeof *msg
 201             && msg->length <= sizeof *msg + MAX_LOGGER_REQUEST_LEN
 202             && msg->data[msg->length - sizeof *msg] == '\0'
 203         ;
 204 }
 205
 206 static int parse_req_for_dlogutil_params(struct dlogutil_line_params *params, const struct dlog_control_msg *msg)
 207 {
 208         if (!initialize_dlogutil_line_params(params, (struct buf_params) { }))
 209                 return -ENOMEM;
 210
 211         /* Note that in this case, the format doesn't matter
 212          * since we're going to send binary data anyway.
 213          * Therefore, we just pass OFF as default. */
 214         const int r = get_dlogutil_line_params(msg->data, FORMAT_OFF, params);
 215         if (r < 0)
 216                 return r;
 217
 218         /* Do not trust writer-based readers (only config-based).
 219          * The control socket's privilege checks are fairly lenient
 220          * so this prevents people from asking us to overwrite
 221          * some potentially important files at logger privilege.
 222          *
 223          * At some point it would be good to be able to skip the
 224          * middleman and become able to write to a file directly
 225          * though. The daemon should become able to receive an
 226          * opened file descriptor from a writer. */
 227         if (params->file_path)
 228                 return -EPERM;
 229
 230         return 0;
 231 }
 232
 233 static int sent_req_reply(int fd, signed char request)
 234 {
 235         const int ret = send_dlog_reply(fd, request, DLOG_REQ_RESULT_ERR, NULL, 0);
 236         if (ret < 0)
 237                 printf("ERROR: both the request handling and send_dlog_reply() failed\n");
 238
 239         return ret;
 240 }
 241
 242 static int distribute_pipes(struct logger *server, struct reader_common *reader, struct dlogutil_line_params *params, int writer_fd)
 243 {
 244         int pipe_fd[2] = { -1, -1 };
 245         if (create_fifo_fds(server, pipe_fd, 0, params->is_dumping))
 246                 return false;
 247
 248         if (reader_add_subreader_dlogutil(reader, params->filter, pipe_fd[1]) != 0)
 249                 return false;
 250
 251         /* FIXME: ideally the FD entity would belong to the sub,
 252          * it only works because there is only one sub at a time.
 253          * Changing this requires some design thought around how
 254          * to handle multiple subs with varying sink throughput. */
 255         set_write_fd_entity(&reader->fd_entity_sink, pipe_fd[1]);
 256
 257         const int rs = send_pipe(writer_fd, pipe_fd[0]);
 258         if (pipe_fd[0] > 0)
 259                 close(pipe_fd[0]);
 260         if (rs)
 261                 return false;
 262
 263         const int ra = add_reader_to_server(reader, server);
 264         if (ra < 0)
 265                 return false;
 266
 267         return true;
 268 }
 269
 270 int req_init_reader_pipe(struct logger *server, struct writer *wr, void *reader, struct dlogutil_line_params *params)
 271 {
 272         return reader_pipe_init((struct reader_pipe **)reader, params->enabled_buffers, server, params->monitor, params->is_dumping);
 273 }
 274
 275 int req_init_reader_memory(struct logger *server, struct writer *wr, void *reader, struct dlogutil_line_params *params)
 276 {
 277         return reader_memory_init_with_writer((struct reader_memory **)reader, wr, server, params->compression, params->monitor, params->is_dumping);
 278 }
 279
 280 /**
 281  * @brief Service util request
 282  * @details Handle a request from util
 283  * @param[in] server The logger server
 284  * @param[in] wr The writer who sent the request
 285  * @param[in] msg The message containing the request
 286  * @return 0 on success, else -errno
 287  */
 288 int service_writer_handle_req_generic_util(struct logger *server, struct writer *wr, struct dlog_control_msg *msg, bool wanted_compression,
 289         int (*make_reader_func)(struct logger *, struct writer *, void *, struct dlogutil_line_params *), signed char request)
 290 {
 291         assert(server);
 292         assert(wr);
 293         assert(msg);
 294         assert(msg->request == request);
 295
 296         if (!is_control_msg_valid_for_util_req(msg))
 297                 return -EINVAL;
 298
 299         __attribute__((cleanup(free_dlogutil_line_params))) struct dlogutil_line_params params;
 300         int r = parse_req_for_dlogutil_params(&params, msg);
 301         if (r < 0)
 302                 goto cleanup;
 303
 304         /* Enabling compression is only available from the text
 305          * config. libdlogutil clients have a separate request
 306          * type to obtain data. Eventually it would be good to
 307          * merge them together, but that requires some thought
 308          * put into it. */
 309         if (!!params.compression != wanted_compression)
 310                 goto cleanup;
 311
 312         __attribute__((cleanup(reader_free_ptr))) struct reader_common *reader = NULL;
 313         r = make_reader_func(server, wr, &reader, &params);
 314         if (r != 0)
 315                 goto cleanup;
 316
 317         if (!distribute_pipes(server, reader, &params, wr->fd_entity.fd))
 318                 goto cleanup;
 319
 320         reader = NULL;
 321         return 0;
 322
 323 cleanup:
 324         /* NB: reply success means that stuff is still stable enough that the client
 325          * can probably get a second chance, so return a success from the whole func
 326          * so as not to drop the client */
 327         return sent_req_reply(wr->fd_entity.fd, request);
 328 }
 329
 330 static int service_writer_handle_req_util(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 331 {
 332         return service_writer_handle_req_generic_util(server, wr, msg,
 333                 false, req_init_reader_pipe, DLOG_REQ_HANDLE_LOGUTIL);
 334 }
 335
 336 static int service_writer_handle_req_compressed_memory_util(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 337 {
 338         return service_writer_handle_req_generic_util(server, wr, msg,
 339                 true, req_init_reader_memory, DLOG_REQ_HANDLE_COMPRESSED_LOGUTIL);
 340 }
 341
 342 static int service_writer_handle_req_get_usage(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 343 {
 344         assert(server);
 345         assert(wr);
 346         assert(msg);
 347         assert(msg->request == DLOG_REQ_GET_USAGE);
 348
 349         uint32_t buf_size = log_storage_get_usage(wr->buf_ptr->log_storage_ptr);
 350
 351         return send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_GET_USAGE, DLOG_REQ_RESULT_OK, &buf_size, sizeof buf_size);
 352 }
 353
 354 static int service_writer_handle_req_get_capacity(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 355 {
 356         assert(server);
 357         assert(wr);
 358         assert(msg);
 359         assert(msg->request == DLOG_REQ_GET_CAPACITY);
 360
 361         uint32_t buf_size = log_storage_get_capacity(wr->buf_ptr->log_storage_ptr);
 362
 363         return send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_GET_CAPACITY, DLOG_REQ_RESULT_OK, &buf_size, sizeof buf_size);
 364 }
 365
 366 /**
 367  * @brief Service clear request
 368  * @details Handle a clear-buffer request
 369  * @param[in] server The logger server
 370  * @param[in] wr The writer who sent the request
 371  * @param[in] msg The message containing the request
 372  * @return 0 on success, else -errno
 373  */
 374 static int service_writer_handle_req_clear(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 375 {
 376         (void) server;
 377         assert(msg);
 378
 379         assert(msg->request == DLOG_REQ_CLEAR);
 380
 381         if (msg->length != (sizeof(struct dlog_control_msg)))
 382                 return -EINVAL;
 383
 384         if (!wr || !wr->buf_ptr)
 385                 return -EINVAL;
 386
 387         log_storage_clear(wr->buf_ptr->log_storage_ptr);
 388
 389         return 0;
 390 }
 391
 392 static int service_writer_handle_req_global_enable_disable_stdout(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 393 {
 394         assert(server);
 395
 396         assert(msg);
 397         assert(msg->request == DLOG_REQ_GLOBAL_ENABLE_DISABLE_STDOUT);
 398
 399         assert(wr);
 400         assert(wr->buf_ptr);
 401
 402         if (msg->length != sizeof(struct dlog_control_msg) + sizeof(char))
 403                 return -EINVAL;
 404
 405         wr->buf_ptr->accept_stdout = msg->data[0] != '\0';
 406         return 0;
 407 }
 408
 409 static int service_writer_handle_req_get_stdout(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 410 {
 411         assert(server);
 412
 413         assert(msg);
 414         assert(msg->request == DLOG_REQ_GET_STDOUT);
 415
 416         assert(wr);
 417         assert(wr->buf_ptr);
 418
 419         if (msg->length != sizeof(struct dlog_control_msg))
 420                 return -EINVAL;
 421
 422         return send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_GET_STDOUT, DLOG_REQ_RESULT_OK, &(char){ wr->buf_ptr->accept_stdout }, sizeof(char));
 423 }
 424
 425 bool cmp_names(void *element, void *userdata)
 426 {
 427         return strcmp(log_compressed_storage_get_name((log_compressed_storage *) element), (char *) userdata) == 0;
 428 }
 429
 430 static int service_writer_handle_req_compression_resize(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 431 {
 432         assert(server);
 433         assert(wr);
 434         assert(wr->buf_ptr);
 435         assert(msg);
 436         assert(msg->request == DLOG_REQ_CHANGE_COMPRESSION_SIZE);
 437
 438         if (msg->length <= sizeof(struct dlog_control_msg) ||
 439                 msg->length > sizeof(struct dlog_control_msg) + MAX_LOGGER_REQUEST_LEN)
 440                 return -EINVAL;
 441
 442         if (msg->data[msg->length - sizeof(struct dlog_control_msg)] != 0)
 443                 return -EINVAL;
 444
 445         __attribute__((cleanup(free_ptr))) char *storage_name = NULL;
 446         __attribute__((cleanup(free_ptr))) char *data = calloc(1, msg->length);
 447         if (data == NULL)
 448                 return -ENOMEM;
 449         memcpy(data, msg->data, msg->length);
 450
 451         unsigned int capacity = *(unsigned int *)data;
 452         if (capacity == 0)
 453                 return -EINVAL;
 454
 455         size_t storage_name_len = msg->length - sizeof(unsigned int);
 456         storage_name = calloc(1, storage_name_len);
 457         if (storage_name == NULL)
 458                 return -ENOMEM;
 459         memcpy(storage_name, (data + sizeof(unsigned int)), storage_name_len);
 460
 461         log_compressed_storage *storage = list_find_if(server->compressed_memories, storage_name, cmp_names);
 462         if (storage == NULL)
 463                 return -ENOENT;
 464
 465         log_compressed_storage_resize(storage, capacity);
 466         return 0;
 467 }
 468
 469 /**
 470  * @brief Service control request
 471  * @details Handle a clear-buffer or util request in respect to msg request type
 472  * @param[in] server The logger server
 473  * @param[in] wr The writer who sent the request
 474  * @param[in] msg The message containing the request
 475  * @return 0 on success, else -errno
 476  */
 477 static int service_writer_handle_req_ctrl(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 478 {
 479         assert(server);
 480         assert(wr);
 481         assert(msg);
 482
 483         int ret;
 484         switch (msg->request) {
 485                 case DLOG_REQ_CLEAR:                         ret = service_writer_handle_req_clear                       (server, wr, msg); break;
 486                 case DLOG_REQ_HANDLE_LOGUTIL:                ret = service_writer_handle_req_util                        (server, wr, msg); break;
 487                 case DLOG_REQ_HANDLE_COMPRESSED_LOGUTIL:     ret = service_writer_handle_req_compressed_memory_util      (server, wr, msg); break;
 488                 case DLOG_REQ_GET_CAPACITY:                  ret = service_writer_handle_req_get_capacity                (server, wr, msg); break;
 489                 case DLOG_REQ_GET_USAGE:                     ret = service_writer_handle_req_get_usage                   (server, wr, msg); break;
 490                 case DLOG_REQ_GLOBAL_ENABLE_DISABLE_STDOUT:  ret = service_writer_handle_req_global_enable_disable_stdout(server, wr, msg); break;
 491                 case DLOG_REQ_GET_STDOUT:                    ret = service_writer_handle_req_get_stdout                  (server, wr, msg); break;
 492                 case DLOG_REQ_CHANGE_COMPRESSION_SIZE:       ret = service_writer_handle_req_compression_resize          (server, wr, msg); break;
 493
 494                 default: ret = -EINVAL;
 495         }
 496
 497         if (wr->readed > msg->length) {
 498                 wr->readed -= msg->length;
 499                 memmove(wr->buffer, wr->buffer + msg->length, wr->readed);
 500         } else
 501                 wr->readed = 0;
 502
 503         return ret;
 504 }
 505
 506 /**
 507  * @brief Service a pipe acquisition request
 508  * @details Handle a pipe request
 509  * @param[in] server The logger server
 510  * @param[in] wr The writer who sent the request
 511  * @param[in] msg The message containing the request
 512  * @return 0 on success, else -errno
 513  */
 514 static int service_writer_handle_req_pipe(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 515 {
 516         int r;
 517
 518         assert(msg);
 519
 520         assert(msg->request == DLOG_REQ_PIPE);
 521
 522         if (msg->length != sizeof(struct dlog_control_msg))
 523                 return -EINVAL;
 524
 525         /* We get named fifo fds here instead of regular pipe(2) to ensure Smack
 526          * labels are set correctly - see comments in get_nonblocking_fifo().
 527          */
 528         int pipe_fd[2] = { -1, -1 };
 529         int ret = get_nonblocking_fifo(pipe_fd, O_CLOEXEC);
 530         if (ret < 0) {
 531                 check_if_fd_limit_reached(server, errno);
 532                 return ret;
 533         }
 534
 535         if (fcntl(pipe_fd[1], F_SETPIPE_SZ, PIPE_REQUESTED_SIZE) < 0) {
 536                 /* Ignore failures. This call is just a performance optimisation
 537                  * and doesn't affect functionality; we can't do anything about
 538                  * an error anyway. */
 539         }
 540
 541         assert(server);
 542         assert(wr);
 543
 544         struct fd_entity pipe_entity = wr->fd_entity;
 545         set_read_fd_entity(&pipe_entity, pipe_fd[0]);
 546         r = add_fd_entity(&server->epoll_common, &pipe_entity);
 547         if (r < 0)
 548                 goto err_close;
 549
 550         r = send_pipe(wr->fd_entity.fd, pipe_fd[1]);
 551         if (r < 0)
 552                 goto err_remove;
 553         close(pipe_fd[1]);
 554
 555         writer_close_fd(server, wr);
 556         wr->service_writer = service_writer_pipe;
 557         wr->fd_entity = pipe_entity;
 558         wr->readed = 0;
 559         return 0;
 560
 561 err_remove:
 562         remove_fd_entity(&server->epoll_common, &pipe_entity);
 563
 564 err_close:
 565         close(pipe_fd[0]);
 566         close(pipe_fd[1]);
 567
 568         return r;
 569 }
 570
 571 static int service_writer_handle_req_stdout(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 572 {
 573         int r;
 574
 575         assert(msg);
 576
 577         assert(msg->request == DLOG_REQ_STDOUT);
 578
 579         if (msg->length <= sizeof(struct dlog_control_msg) + sizeof(struct dlog_control_msg_stdout))
 580                 return -EINVAL;
 581         struct dlog_control_msg_stdout *pre_data = (void *)msg->data;
 582         if (pre_data->prio <= DLOG_DEFAULT || pre_data->prio >= DLOG_PRIO_MAX) // TODO: ugly
 583                 return -EINVAL;
 584
 585         int data_len = msg->length - sizeof(struct dlog_control_msg);
 586         struct dlog_control_msg_stdout *data = malloc(data_len + 1);
 587         if (!data)
 588                 return -ENOMEM;
 589         memcpy(data, pre_data, data_len);
 590         // NB: There might have been a zero earlier. We don't really care, as nothing wrong will happen.
 591         ((char *)data)[data_len] = '\0';
 592
 593         // Open connection to logger, do not set O_CLOEXEC as stdout might get inherited
 594         int pipe_fd[2] = { -1, -1 };
 595         r = get_nonblocking_fifo(pipe_fd, 0);
 596         if (r < 0) {
 597                 check_if_fd_limit_reached(server, errno);
 598                 goto err_free;
 599         }
 600
 601         if (fcntl(pipe_fd[1], F_SETPIPE_SZ, PIPE_REQUESTED_SIZE) < 0) {
 602                 /* Ignore failures. This call is just a performance optimisation
 603                  * and doesn't affect functionality; we can't do anything about
 604                  * an error anyway. */
 605         }
 606
 607         assert(server);
 608         assert(wr);
 609
 610         struct fd_entity pipe_entity = wr->fd_entity;
 611         set_read_fd_entity(&pipe_entity, pipe_fd[0]);
 612         r = add_fd_entity(&server->epoll_common, &pipe_entity);
 613         if (r < 0)
 614                 goto err_close;
 615
 616         r = send_pipe(wr->fd_entity.fd, pipe_fd[1]);
 617         if (r < 0)
 618                 goto err_remove;
 619         close(pipe_fd[1]);
 620
 621         writer_close_fd(server, wr);
 622         wr->service_writer = service_writer_stdout;
 623         wr->fd_entity = pipe_entity;
 624         wr->readed = 0;
 625         wr->stdout_data = data;
 626         return 0;
 627
 628 err_remove:
 629         remove_fd_entity(&server->epoll_common, &pipe_entity);
 630
 631 err_close:
 632         close(pipe_fd[0]);
 633         close(pipe_fd[1]);
 634
 635 err_free:
 636         free(data);
 637
 638         return r;
 639 }
 640
 641 static int service_writer_handle_req_write(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 642 {
 643         assert(server);
 644         assert(wr);
 645         assert(msg);
 646
 647         switch (msg->request) {
 648                 case DLOG_REQ_PIPE: return service_writer_handle_req_pipe(server, wr, msg);
 649                 case DLOG_REQ_STDOUT:
 650                         return service_writer_handle_req_stdout(server, wr, msg);
 651
 652                 default: return -EINVAL;
 653         }
 654
 655         /* NB: unlike for the control socket, resetting `wr->readed`
 656          * is the responsibility of the specific request handlers,
 657          * since they may obsolete the need by modifying the connection,
 658          * in particular, reset it via morph to a pipe */
 659 }
 660
 661 /**
 662  * @brief Create buffer
 663  * @details Allocate a buffer structure
 664  * @param[out] lb The newly-allocated buffer
 665  * @param[in] buf_id The buffer ID
 666  * @param[in] data Buffer config data
 667  * @return 0 on success, -errno on failure
 668  */
 669 int buffer_create(struct log_buffer **log_buffer, log_id_t buf_id, struct buffer_config_data *data, struct logger *server)
 670 {
 671         assert(data);
 672         struct log_buffer *lb = (struct log_buffer *) calloc(1, sizeof(*lb));
 673
 674         if (!lb)
 675                 return -ENOMEM;
 676
 677         lb->id = buf_id;
 678         lb->log_storage_ptr = log_storage_create(data->size, server->sort_by);
 679         if (!lb->log_storage_ptr) {
 680                 free(lb);
 681                 return -ENOMEM;
 682         }
 683
 684         int r;
 685         r = socket_initialize(&lb->sock_ctl, lb, service_writer_handle_req_ctrl, &data->ctl_socket);
 686         if (r < 0) {
 687                 log_storage_free(lb->log_storage_ptr);
 688                 free(lb);
 689                 return r;
 690         }
 691
 692         r = socket_initialize(&lb->sock_conn, lb, service_writer_handle_req_write, &data->conn_socket);
 693         if (r < 0) {
 694                 socket_close(&lb->sock_ctl);
 695                 log_storage_free(lb->log_storage_ptr);
 696                 free(lb);
 697                 return r;
 698         }
 699
 700         lb->accept_stdout = true;
 701
 702         *log_buffer = lb;
 703         return 0;
 704 }
 705
 706 /**
 707  * @brief Free buffer
 708  * @details Deallocate a buffer
 709  * @param[in] buffer The buffer to deallocate
 710  */
 711 void buffer_free(struct log_buffer *buffer)
 712 {
 713         assert(buffer);
 714
 715         socket_close(&buffer->sock_ctl);
 716         socket_close(&buffer->sock_conn);
 717
 718         log_storage_free(buffer->log_storage_ptr);
 719
 720         free(buffer);
 721 }