src/bin/lttng-relayd/stream-fd.c

   1 /*
   2  * Copyright (C) 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   3  *               2018 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License, version 2 only, as
   7  * published by the Free Software Foundation.
   8  *
   9  * This program is distributed in the hope that it will be useful, but WITHOUT
  10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  12  * more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along with
  15  * this program; if not, write to the Free Software Foundation, Inc., 51
  16  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  17  */
  18
  19 #define _LGPL_SOURCE
  20
  21 #include <urcu/ref.h>
  22 #include <sys/stat.h>
  23 #include <fcntl.h>
  24 #include <common/common.h>
  25 #include <common/fd-tracker/fd-tracker.h>
  26 #include <common/fd-tracker/utils.h>
  27 #include <common/utils.h>
  28
  29 #include "stream-fd.h"
  30 #include "lttng-relayd.h"
  31
  32 struct stream_fd {
  33         bool suspendable;
  34         union {
  35                 /* Suspendable. */
  36                 struct fs_handle *handle;
  37                 /* Unsuspendable. */
  38                 int fd;
  39         } u;
  40         struct urcu_ref ref;
  41 };
  42
  43 static struct stream_fd *_stream_fd_alloc(void)
  44 {
  45         struct stream_fd *sf;
  46
  47         sf = zmalloc(sizeof(*sf));
  48         if (!sf) {
  49                 goto end;
  50         }
  51         urcu_ref_init(&sf->ref);
  52 end:
  53         return sf;
  54 }
  55
  56 static struct stream_fd *stream_fd_suspendable_create(struct fs_handle *handle)
  57 {
  58         struct stream_fd *stream_fd = _stream_fd_alloc();
  59
  60         if (!stream_fd) {
  61                 goto end;
  62         }
  63
  64         stream_fd->suspendable = true;
  65         stream_fd->u.handle = handle;
  66 end:
  67         return stream_fd;
  68 }
  69
  70 static struct stream_fd *stream_fd_unsuspendable_create(int fd)
  71 {
  72         struct stream_fd *stream_fd = _stream_fd_alloc();
  73
  74         if (!stream_fd) {
  75                 goto end;
  76         }
  77
  78         stream_fd->suspendable = false;
  79         stream_fd->u.fd = fd;
  80 end:
  81         return stream_fd;
  82 }
  83
  84 static int open_file(void *data, int *out_fd)
  85 {
  86         int ret;
  87         const char *path = data;
  88
  89         ret = open(path, O_RDONLY);
  90         if (ret < 0) {
  91                 goto end;
  92         }
  93         *out_fd = ret;
  94         ret = 0;
  95 end:
  96         return ret;
  97 }
  98
  99 /*
 100  * Stream files are opened (read-only) on the live end of the relayd.
 101  * In live mode, it is expected that a client is able to consume a
 102  * complete file even if it is replaced (in file rotation mode).
 103  *
 104  * Thus, it is not possible to open those files as suspendable file
 105  * handles. This means that live clients can keep a large number of
 106  * open file descriptors. As a work-around, we could create hard links
 107  * to the files to make the files suspendable. The original file would be
 108  * replaced, but the viewer's hard-link would ensure that the inode is
 109  * still available for restoration.
 110  *
 111  * The main roadblock to this approach is validating that the trace
 112  * directory resides in a filesystem that supports hard-links. Otherwise,
 113  * a cooperative mechanism could allow the viewer end to mark a file as
 114  * being in use and it could be renamed rather than unlinked by the
 115  * receiving end.
 116  */
 117 struct stream_fd *stream_fd_open(const char *path)
 118 {
 119         int ret, fd;
 120         struct stream_fd *stream_fd = NULL;
 121
 122         ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &fd,
 123                         (const char **) &path, 1,
 124                         open_file, (void *) path);
 125         if (ret) {
 126                 goto end;
 127         }
 128
 129         stream_fd = stream_fd_unsuspendable_create(fd);
 130         if (!stream_fd) {
 131                 (void) fd_tracker_close_unsuspendable_fd(the_fd_tracker, &fd, 1,
 132                                 fd_tracker_util_close_fd, NULL);
 133         }
 134 end:
 135         return stream_fd;
 136 }
 137
 138 static
 139 struct fs_handle *create_fs_handle(const char *path)
 140 {
 141         struct fs_handle *handle;
 142         /*
 143          * With the session rotation feature on the relay, we might need to seek
 144          * and truncate a tracefile, so we need read and write access.
 145          */
 146         int flags = O_RDWR | O_CREAT | O_TRUNC;
 147         /* Open with 660 mode */
 148         mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
 149
 150         handle =  fd_tracker_open_fs_handle(the_fd_tracker, path, flags, &mode);
 151         if (!handle) {
 152                 ERR("Failed to open fs handle to %s", path);
 153         }
 154
 155         return handle;
 156 }
 157
 158 /*
 159  * Stream file are created by on the consumerd/data-reception end. Those
 160  * stream fds can be suspended as there is no expectation that the files
 161  * will be unlinked and then need to be appended-to.
 162  *
 163  * Hence, the file descriptors are created as suspendable to allow the
 164  * fd-tracker to reduce the number of active fds..
 165  */
 166 struct stream_fd *stream_fd_create(const char *path_name, const char *file_name,
 167                 uint64_t size, uint64_t count, const char *suffix)
 168 {
 169         struct stream_fd *stream_fd = NULL;
 170         struct fs_handle *handle;
 171         int ret;
 172         char path[PATH_MAX];
 173
 174         ret = utils_stream_file_name(path, path_name, file_name,
 175                         size, count, suffix);
 176         if (ret < 0) {
 177                 goto end;
 178         }
 179
 180         handle = create_fs_handle(path);
 181         if (!handle) {
 182                 goto end;
 183         }
 184
 185         stream_fd = stream_fd_suspendable_create(handle);
 186         if (!stream_fd) {
 187                 (void) fs_handle_close(handle);
 188         }
 189
 190 end:
 191         return stream_fd;
 192 }
 193
 194 int stream_fd_rotate(struct stream_fd *stream_fd, const char *path_name,
 195                 const char *file_name, uint64_t size,
 196                 uint64_t count, uint64_t *new_count)
 197 {
 198         int ret;
 199         bool should_unlink;
 200         char path[PATH_MAX];
 201
 202         assert(stream_fd);
 203         assert(stream_fd->suspendable);
 204
 205         utils_stream_file_rotation_get_new_count(count, new_count,
 206                         &should_unlink);
 207
 208         ret = utils_stream_file_name(path, path_name, file_name,
 209                         size, count, NULL);
 210         if (ret < 0) {
 211                 goto error;
 212         }
 213
 214         ret = fs_handle_close(stream_fd->u.handle);
 215         stream_fd->u.handle = NULL;
 216         if (ret < 0) {
 217                 PERROR("Closing stream tracefile handle");
 218                 goto error;
 219         }
 220
 221         if (should_unlink) {
 222                 unlink(path);
 223                 if (ret < 0 && errno != ENOENT) {
 224                         goto error;
 225                 }
 226         }
 227
 228         ret = utils_stream_file_name(path, path_name, file_name,
 229                         size, new_count ? *new_count : 0, NULL);
 230         if (ret < 0) {
 231                 goto error;
 232         }
 233
 234         stream_fd->u.handle = create_fs_handle(path);
 235         if (!stream_fd->u.handle) {
 236                 ret = -1;
 237                 goto error;
 238         }
 239
 240         ret = 0;
 241
 242 error:
 243         return ret;
 244 }
 245
 246 void stream_fd_get(struct stream_fd *sf)
 247 {
 248         urcu_ref_get(&sf->ref);
 249 }
 250
 251 static void stream_fd_release(struct urcu_ref *ref)
 252 {
 253         struct stream_fd *sf = caa_container_of(ref, struct stream_fd, ref);
 254         int ret;
 255
 256         if (sf->suspendable) {
 257                 ret = fs_handle_close(sf->u.handle);
 258         } else {
 259                 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, &sf->u.fd,
 260                                 1, fd_tracker_util_close_fd, NULL);
 261         }
 262         if (ret) {
 263                 PERROR("Error closing stream handle");
 264         }
 265         free(sf);
 266 }
 267
 268 void stream_fd_put(struct stream_fd *sf)
 269 {
 270         urcu_ref_put(&sf->ref, stream_fd_release);
 271 }
 272
 273 int stream_fd_get_fd(struct stream_fd *sf)
 274 {
 275         return sf->suspendable ? fs_handle_get_fd(sf->u.handle) : sf->u.fd;
 276 }
 277
 278 void stream_fd_put_fd(struct stream_fd *sf)
 279 {
 280         if (sf->suspendable) {
 281                 fs_handle_put_fd(sf->u.handle);
 282         }
 283 }