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 /**
 199  * @brief Service util request
 200  * @details Handle a standard (not compressed memory) request from util
 201  * @param[in] server The logger server
 202  * @param[in] wr The writer who sent the request
 203  * @param[in] msg The message containing the request
 204  * @return 0 on success, else -errno
 205  */
 206 static int service_writer_handle_req_util(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 207 {
 208         assert(server);
 209         assert(wr);
 210         assert(msg);
 211
 212         assert(msg->request == DLOG_REQ_HANDLE_LOGUTIL);
 213
 214         if (msg->length <= sizeof(struct dlog_control_msg) ||
 215             msg->length > sizeof(struct dlog_control_msg) + MAX_LOGGER_REQUEST_LEN)
 216                 return -EINVAL;
 217
 218         if (msg->data[msg->length - sizeof(struct dlog_control_msg)] != 0)
 219                 return -EINVAL;
 220
 221         __attribute__((cleanup(reader_free_ptr))) struct reader_pipe *reader = NULL;
 222
 223         int retval;
 224         __attribute__((cleanup(free_dlogutil_line_params))) struct dlogutil_line_params params;
 225         if (!initialize_dlogutil_line_params(&params, (struct buf_params) { })) {
 226                 /* TODO: cleanup discards this value, so there isn't much
 227                  * point setting it. Ideally it would be attached to the
 228                  * reply but that's a protocol change so not worth it atm */
 229                 // retval = -ENOMEM;
 230                 goto cleanup;
 231         }
 232
 233         /* Note that in this case, the format doesn't matter.
 234          * Therefore, we just pass OFF as default. */
 235         retval = get_dlogutil_line_params(msg->data, FORMAT_OFF, &params);
 236         if (retval < 0) {
 237                 // retval = -ENOMEM; // see above
 238                 goto cleanup;
 239         }
 240
 241         if (params.compression) {
 242                 /* Memory compression is only available from the text
 243                  * config. libdlogutil clients have a separate request
 244                  * type to obtain data. Eventually it would be good to
 245                  * merge them together, but that requires some thought
 246                  * put into it. */
 247                 goto cleanup;
 248         }
 249
 250         if (params.file_path) {
 251                 /* Do not trust writer-based readers (only config-based).
 252                  * The control socket's privilege checks are fairly lenient
 253                  * so this prevents people from asking us to overwrite
 254                  * some potentially important files at logger privilege.
 255                  *
 256                  * At some point it would be good to be able to skip the
 257                  * middleman and become able to write to a file directly
 258                  * though. The daemon should become able to receive an
 259                  * opened file descriptor from a writer. */
 260                 // retval = -EPERM; // see above
 261                 goto cleanup;
 262         }
 263
 264         retval = reader_pipe_init(&reader, params.enabled_buffers, server, params.monitor, params.is_dumping);
 265         if (retval != 0)
 266                 goto cleanup;
 267
 268         int pipe_fd[2] = { -1, -1 };
 269         retval = create_fifo_fds(server, pipe_fd, 0, params.is_dumping);
 270         if (retval < 0)
 271                 goto cleanup;
 272
 273         retval = reader_add_subreader_dlogutil(&reader->common, params.filter, pipe_fd[1]);
 274         if (retval != 0)
 275                 goto cleanup;
 276
 277         set_write_fd_entity(&reader->common.fd_entity_sink, pipe_fd[1]);
 278         retval = send_pipe(wr->fd_entity.fd, pipe_fd[0]);
 279         if (pipe_fd[0] > 0)
 280                 close(pipe_fd[0]);
 281         if (retval)
 282                 goto cleanup;
 283
 284         retval = add_reader_to_server(&reader->common, server);
 285         if (retval < 0)
 286                 goto cleanup;
 287
 288         reader = NULL;
 289         return 0;
 290
 291 cleanup:
 292         /* NB: reply success means that stuff is still stable enough that the client
 293          * can probably get a second chance, so return a success from the whole func
 294          * so as not to drop the client */
 295         retval = send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_HANDLE_LOGUTIL, DLOG_REQ_RESULT_ERR, NULL, 0);
 296         if (retval < 0)
 297                 printf("ERROR: both create_reader_from_dlogutil_line() and send_dlog_reply() failed\n");
 298
 299         return retval;
 300 }
 301
 302 static int service_writer_handle_req_compressed_memory_util(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 303 {
 304         assert(server);
 305         assert(wr);
 306         assert(msg);
 307
 308         assert(msg->request == DLOG_REQ_HANDLE_COMPRESSED_LOGUTIL);
 309
 310         if (msg->length <= sizeof(struct dlog_control_msg) ||
 311             msg->length > sizeof(struct dlog_control_msg) + MAX_LOGGER_REQUEST_LEN)
 312                 return -EINVAL;
 313
 314         if (msg->data[msg->length - sizeof(struct dlog_control_msg)] != 0)
 315                 return -EINVAL;
 316
 317         __attribute__((cleanup(reader_free_ptr))) struct reader_memory *reader = NULL;
 318
 319         int retval;
 320         __attribute__((cleanup(free_dlogutil_line_params))) struct dlogutil_line_params params;
 321         if (!initialize_dlogutil_line_params(&params, (struct buf_params) { })) {
 322                 /* TODO: cleanup discards this value, so there isn't much
 323                  * point setting it. Ideally it would be attached to the
 324                  * reply but that's a protocol change so not worth it atm */
 325                 // retval = -ENOMEM;
 326                 goto cleanup;
 327         }
 328
 329         /* Note that in this case, the format doesn't matter.
 330          * Therefore, we just pass OFF as default. */
 331         retval = get_dlogutil_line_params(msg->data, FORMAT_OFF, &params);
 332         if (retval < 0) {
 333                 // retval = -ENOMEM; // see above
 334                 goto cleanup;
 335         }
 336
 337         if (params.file_path) {
 338                 /* Do not trust writer-based readers (only config-based).
 339                  * The control socket's privilege checks are fairly lenient
 340                  * so this prevents people from asking us to overwrite
 341                  * some potentially important files at logger privilege.
 342                  *
 343                  * At some point it would be good to be able to skip the
 344                  * middleman and become able to write to a file directly
 345                  * though. The daemon should become able to receive an
 346                  * opened file descriptor from a writer. */
 347                 // retval = -EPERM; // see above
 348                 goto cleanup;
 349         }
 350
 351         if (!params.compression)
 352                 goto cleanup;
 353
 354         retval = reader_memory_init_with_writer(&reader, wr, server, params.compression, params.monitor, params.is_dumping);
 355         if (retval != 0)
 356                 goto cleanup;
 357
 358         int pipe_fd[2] = { -1, -1 };
 359         retval = create_fifo_fds(server, pipe_fd, 0, reader->is_dumping);
 360         if (retval < 0)
 361                 goto cleanup;
 362
 363         retval = reader_add_subreader_dlogutil(&reader->common, params.filter, pipe_fd[1]);
 364         if (retval != 0)
 365                 goto cleanup;
 366
 367         /* FIXME: ideally the FD entity would belong to the sub,
 368          * it only works because there is only one sub at a time.
 369          * Changing this requires some design thought around how
 370          * to handle multiple subs with varying sink throughput. */
 371         set_write_fd_entity(&reader->common.fd_entity_sink, pipe_fd[1]);
 372
 373         retval = send_pipe(wr->fd_entity.fd, pipe_fd[0]);
 374         if (pipe_fd[0] > 0)
 375                 close(pipe_fd[0]);
 376         if (retval)
 377                 goto cleanup;
 378
 379         retval = add_reader_to_server(&reader->common, server);
 380         if (retval < 0)
 381                 goto cleanup;
 382
 383         reader = NULL;
 384         return 0;
 385
 386 cleanup:
 387         /* NB: reply success means that stuff is still stable enough that the client
 388          * can probably get a second chance, so return a success from the whole func
 389          * so as not to drop the client */
 390         retval = send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_HANDLE_COMPRESSED_LOGUTIL, DLOG_REQ_RESULT_ERR, NULL, 0);
 391         if (retval < 0)
 392                 printf("ERROR: both create_reader_from_dlogutil_line() and send_dlog_reply() failed\n");
 393
 394         return retval;
 395 }
 396
 397 static int service_writer_handle_req_get_usage(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 398 {
 399         assert(server);
 400         assert(wr);
 401         assert(msg);
 402         assert(msg->request == DLOG_REQ_GET_USAGE);
 403
 404         uint32_t buf_size = log_storage_get_usage(wr->buf_ptr->log_storage_ptr);
 405
 406         return send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_GET_USAGE, DLOG_REQ_RESULT_OK, &buf_size, sizeof buf_size);
 407 }
 408
 409 static int service_writer_handle_req_get_capacity(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 410 {
 411         assert(server);
 412         assert(wr);
 413         assert(msg);
 414         assert(msg->request == DLOG_REQ_GET_CAPACITY);
 415
 416         uint32_t buf_size = log_storage_get_capacity(wr->buf_ptr->log_storage_ptr);
 417
 418         return send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_GET_CAPACITY, DLOG_REQ_RESULT_OK, &buf_size, sizeof buf_size);
 419 }
 420
 421 /**
 422  * @brief Service clear request
 423  * @details Handle a clear-buffer request
 424  * @param[in] server The logger server
 425  * @param[in] wr The writer who sent the request
 426  * @param[in] msg The message containing the request
 427  * @return 0 on success, else -errno
 428  */
 429 static int service_writer_handle_req_clear(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 430 {
 431         (void) server;
 432         assert(msg);
 433
 434         assert(msg->request == DLOG_REQ_CLEAR);
 435
 436         if (msg->length != (sizeof(struct dlog_control_msg)))
 437                 return -EINVAL;
 438
 439         if (!wr || !wr->buf_ptr)
 440                 return -EINVAL;
 441
 442         log_storage_clear(wr->buf_ptr->log_storage_ptr);
 443
 444         return 0;
 445 }
 446
 447 static int service_writer_handle_req_global_enable_disable_stdout(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 448 {
 449         assert(server);
 450
 451         assert(msg);
 452         assert(msg->request == DLOG_REQ_GLOBAL_ENABLE_DISABLE_STDOUT);
 453
 454         assert(wr);
 455         assert(wr->buf_ptr);
 456
 457         if (msg->length != sizeof(struct dlog_control_msg) + sizeof(char))
 458                 return -EINVAL;
 459
 460         wr->buf_ptr->accept_stdout = msg->data[0] != '\0';
 461         return 0;
 462 }
 463
 464 static int service_writer_handle_req_get_stdout(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 465 {
 466         assert(server);
 467
 468         assert(msg);
 469         assert(msg->request == DLOG_REQ_GET_STDOUT);
 470
 471         assert(wr);
 472         assert(wr->buf_ptr);
 473
 474         if (msg->length != sizeof(struct dlog_control_msg))
 475                 return -EINVAL;
 476
 477         return send_dlog_reply(wr->fd_entity.fd, DLOG_REQ_GET_STDOUT, DLOG_REQ_RESULT_OK, &(char){ wr->buf_ptr->accept_stdout }, sizeof(char));
 478 }
 479
 480 bool cmp_names(void *element, void *userdata)
 481 {
 482         return strcmp(log_compressed_storage_get_name((log_compressed_storage *) element), (char *) userdata) == 0;
 483 }
 484
 485 static int service_writer_handle_req_compression_resize(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 486 {
 487         assert(server);
 488         assert(wr);
 489         assert(wr->buf_ptr);
 490         assert(msg);
 491         assert(msg->request == DLOG_REQ_CHANGE_COMPRESSION_SIZE);
 492
 493         if (msg->length <= sizeof(struct dlog_control_msg) ||
 494                 msg->length > sizeof(struct dlog_control_msg) + MAX_LOGGER_REQUEST_LEN)
 495                 return -EINVAL;
 496
 497         if (msg->data[msg->length - sizeof(struct dlog_control_msg)] != 0)
 498                 return -EINVAL;
 499
 500         __attribute__((cleanup(free_ptr))) char *storage_name = NULL;
 501         __attribute__((cleanup(free_ptr))) char *data = calloc(1, msg->length);
 502         if (data == NULL)
 503                 return -ENOMEM;
 504         memcpy(data, msg->data, msg->length);
 505
 506         unsigned int capacity = *(unsigned int *)data;
 507         if (capacity == 0)
 508                 return -EINVAL;
 509
 510         size_t storage_name_len = msg->length - sizeof(unsigned int);
 511         storage_name = calloc(1, storage_name_len);
 512         if (storage_name == NULL)
 513                 return -ENOMEM;
 514         memcpy(storage_name, (data + sizeof(unsigned int)), storage_name_len);
 515
 516         log_compressed_storage *storage = list_find_if(server->compressed_memories, storage_name, cmp_names);
 517         if (storage == NULL)
 518                 return -ENOENT;
 519
 520         log_compressed_storage_resize(storage, capacity);
 521         return 0;
 522 }
 523
 524 /**
 525  * @brief Service control request
 526  * @details Handle a clear-buffer or util request in respect to msg request type
 527  * @param[in] server The logger server
 528  * @param[in] wr The writer who sent the request
 529  * @param[in] msg The message containing the request
 530  * @return 0 on success, else -errno
 531  */
 532 static int service_writer_handle_req_ctrl(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 533 {
 534         assert(server);
 535         assert(wr);
 536         assert(msg);
 537
 538         int ret;
 539         switch (msg->request) {
 540                 case DLOG_REQ_CLEAR:                         ret = service_writer_handle_req_clear                       (server, wr, msg); break;
 541                 case DLOG_REQ_HANDLE_LOGUTIL:                ret = service_writer_handle_req_util                        (server, wr, msg); break;
 542                 case DLOG_REQ_HANDLE_COMPRESSED_LOGUTIL:     ret = service_writer_handle_req_compressed_memory_util      (server, wr, msg); break;
 543                 case DLOG_REQ_GET_CAPACITY:                  ret = service_writer_handle_req_get_capacity                (server, wr, msg); break;
 544                 case DLOG_REQ_GET_USAGE:                     ret = service_writer_handle_req_get_usage                   (server, wr, msg); break;
 545                 case DLOG_REQ_GLOBAL_ENABLE_DISABLE_STDOUT:  ret = service_writer_handle_req_global_enable_disable_stdout(server, wr, msg); break;
 546                 case DLOG_REQ_GET_STDOUT:                    ret = service_writer_handle_req_get_stdout                  (server, wr, msg); break;
 547                 case DLOG_REQ_CHANGE_COMPRESSION_SIZE:       ret = service_writer_handle_req_compression_resize          (server, wr, msg); break;
 548
 549                 default: ret = -EINVAL;
 550         }
 551
 552         if (wr->readed > msg->length) {
 553                 wr->readed -= msg->length;
 554                 memmove(wr->buffer, wr->buffer + msg->length, wr->readed);
 555         } else
 556                 wr->readed = 0;
 557
 558         return ret;
 559 }
 560
 561 /**
 562  * @brief Service a pipe acquisition request
 563  * @details Handle a pipe request
 564  * @param[in] server The logger server
 565  * @param[in] wr The writer who sent the request
 566  * @param[in] msg The message containing the request
 567  * @return 0 on success, else -errno
 568  */
 569 static int service_writer_handle_req_pipe(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 570 {
 571         int r;
 572
 573         assert(msg);
 574
 575         assert(msg->request == DLOG_REQ_PIPE);
 576
 577         if (msg->length != sizeof(struct dlog_control_msg))
 578                 return -EINVAL;
 579
 580         /* We get named fifo fds here instead of regular pipe(2) to ensure Smack
 581          * labels are set correctly - see comments in get_nonblocking_fifo().
 582          */
 583         int pipe_fd[2] = { -1, -1 };
 584         int ret = get_nonblocking_fifo(pipe_fd, O_CLOEXEC);
 585         if (ret < 0) {
 586                 check_if_fd_limit_reached(server, errno);
 587                 return ret;
 588         }
 589
 590         if (fcntl(pipe_fd[1], F_SETPIPE_SZ, PIPE_REQUESTED_SIZE) < 0) {
 591                 /* Ignore failures. This call is just a performance optimisation
 592                  * and doesn't affect functionality; we can't do anything about
 593                  * an error anyway. */
 594         }
 595
 596         assert(server);
 597         assert(wr);
 598
 599         struct fd_entity pipe_entity = wr->fd_entity;
 600         set_read_fd_entity(&pipe_entity, pipe_fd[0]);
 601         r = add_fd_entity(&server->epoll_common, &pipe_entity);
 602         if (r < 0)
 603                 goto err_close;
 604
 605         r = send_pipe(wr->fd_entity.fd, pipe_fd[1]);
 606         if (r < 0)
 607                 goto err_remove;
 608         close(pipe_fd[1]);
 609
 610         writer_close_fd(server, wr);
 611         wr->service_writer = service_writer_pipe;
 612         wr->fd_entity = pipe_entity;
 613         wr->readed = 0;
 614         return 0;
 615
 616 err_remove:
 617         remove_fd_entity(&server->epoll_common, &pipe_entity);
 618
 619 err_close:
 620         close(pipe_fd[0]);
 621         close(pipe_fd[1]);
 622
 623         return r;
 624 }
 625
 626 static int service_writer_handle_req_stdout(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 627 {
 628         int r;
 629
 630         assert(msg);
 631
 632         assert(msg->request == DLOG_REQ_STDOUT);
 633
 634         if (msg->length <= sizeof(struct dlog_control_msg) + sizeof(struct dlog_control_msg_stdout))
 635                 return -EINVAL;
 636         struct dlog_control_msg_stdout *pre_data = (void *)msg->data;
 637         if (pre_data->prio <= DLOG_DEFAULT || pre_data->prio >= DLOG_PRIO_MAX) // TODO: ugly
 638                 return -EINVAL;
 639
 640         int data_len = msg->length - sizeof(struct dlog_control_msg);
 641         struct dlog_control_msg_stdout *data = malloc(data_len + 1);
 642         if (!data)
 643                 return -ENOMEM;
 644         memcpy(data, pre_data, data_len);
 645         // NB: There might have been a zero earlier. We don't really care, as nothing wrong will happen.
 646         ((char *)data)[data_len] = '\0';
 647
 648         // Open connection to logger, do not set O_CLOEXEC as stdout might get inherited
 649         int pipe_fd[2] = { -1, -1 };
 650         r = get_nonblocking_fifo(pipe_fd, 0);
 651         if (r < 0) {
 652                 check_if_fd_limit_reached(server, errno);
 653                 goto err_free;
 654         }
 655
 656         if (fcntl(pipe_fd[1], F_SETPIPE_SZ, PIPE_REQUESTED_SIZE) < 0) {
 657                 /* Ignore failures. This call is just a performance optimisation
 658                  * and doesn't affect functionality; we can't do anything about
 659                  * an error anyway. */
 660         }
 661
 662         assert(server);
 663         assert(wr);
 664
 665         struct fd_entity pipe_entity = wr->fd_entity;
 666         set_read_fd_entity(&pipe_entity, pipe_fd[0]);
 667         r = add_fd_entity(&server->epoll_common, &pipe_entity);
 668         if (r < 0)
 669                 goto err_close;
 670
 671         r = send_pipe(wr->fd_entity.fd, pipe_fd[1]);
 672         if (r < 0)
 673                 goto err_remove;
 674         close(pipe_fd[1]);
 675
 676         writer_close_fd(server, wr);
 677         wr->service_writer = service_writer_stdout;
 678         wr->fd_entity = pipe_entity;
 679         wr->readed = 0;
 680         wr->stdout_data = data;
 681         return 0;
 682
 683 err_remove:
 684         remove_fd_entity(&server->epoll_common, &pipe_entity);
 685
 686 err_close:
 687         close(pipe_fd[0]);
 688         close(pipe_fd[1]);
 689
 690 err_free:
 691         free(data);
 692
 693         return r;
 694 }
 695
 696 static int service_writer_handle_req_write(struct logger *server, struct writer *wr, struct dlog_control_msg *msg)
 697 {
 698         assert(server);
 699         assert(wr);
 700         assert(msg);
 701
 702         switch (msg->request) {
 703                 case DLOG_REQ_PIPE: return service_writer_handle_req_pipe(server, wr, msg);
 704                 case DLOG_REQ_STDOUT:
 705                         return service_writer_handle_req_stdout(server, wr, msg);
 706
 707                 default: return -EINVAL;
 708         }
 709
 710         /* NB: unlike for the control socket, resetting `wr->readed`
 711          * is the responsibility of the specific request handlers,
 712          * since they may obsolete the need by modifying the connection,
 713          * in particular, reset it via morph to a pipe */
 714 }
 715
 716 /**
 717  * @brief Create buffer
 718  * @details Allocate a buffer structure
 719  * @param[out] lb The newly-allocated buffer
 720  * @param[in] buf_id The buffer ID
 721  * @param[in] data Buffer config data
 722  * @return 0 on success, -errno on failure
 723  */
 724 int buffer_create(struct log_buffer **log_buffer, log_id_t buf_id, struct buffer_config_data *data, struct logger *server)
 725 {
 726         assert(data);
 727         struct log_buffer *lb = (struct log_buffer *) calloc(1, sizeof(*lb));
 728
 729         if (!lb)
 730                 return -ENOMEM;
 731
 732         lb->id = buf_id;
 733         lb->log_storage_ptr = log_storage_create(data->size, server->sort_by);
 734         if (!lb->log_storage_ptr) {
 735                 free(lb);
 736                 return -ENOMEM;
 737         }
 738
 739         int r;
 740         r = socket_initialize(&lb->sock_ctl, lb, service_writer_handle_req_ctrl, &data->ctl_socket);
 741         if (r < 0) {
 742                 log_storage_free(lb->log_storage_ptr);
 743                 free(lb);
 744                 return r;
 745         }
 746
 747         r = socket_initialize(&lb->sock_conn, lb, service_writer_handle_req_write, &data->conn_socket);
 748         if (r < 0) {
 749                 socket_close(&lb->sock_ctl);
 750                 log_storage_free(lb->log_storage_ptr);
 751                 free(lb);
 752                 return r;
 753         }
 754
 755         lb->accept_stdout = true;
 756
 757         *log_buffer = lb;
 758         return 0;
 759 }
 760
 761 /**
 762  * @brief Free buffer
 763  * @details Deallocate a buffer
 764  * @param[in] buffer The buffer to deallocate
 765  */
 766 void buffer_free(struct log_buffer *buffer)
 767 {
 768         assert(buffer);
 769
 770         socket_close(&buffer->sock_ctl);
 771         socket_close(&buffer->sock_conn);
 772
 773         log_storage_free(buffer->log_storage_ptr);
 774
 775         free(buffer);
 776 }