src/plugins/ctf/lttng-live/lttng-live.hpp

   1 /*
   2  * SPDX-License-Identifier: MIT
   3  *
   4  * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
   5  * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   6  * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   7  *
   8  * BabelTrace - LTTng-live client Component
   9  */
  10
  11 #ifndef BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H
  12 #define BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H
  13
  14 #include <glib.h>
  15 #include <stdint.h>
  16
  17 #include <babeltrace2/babeltrace.h>
  18
  19 #include "cpp-common/bt2/message.hpp"
  20 #include "cpp-common/vendor/fmt/format.h" /* IWYU pragma: keep */
  21
  22 #include "../common/src/metadata/tsdl/decoder.hpp"
  23 #include "../common/src/msg-iter/msg-iter.hpp"
  24 #include "viewer-connection.hpp"
  25
  26 /*
  27  * bt_common_lttng_live_url_parts is defined in common code, and is also used
  28  * by C code, so it can't be C++-ified yet.  Use this separate deleter object
  29  * in the mean time.
  30  */
  31 struct bt_common_lttng_live_url_parts_deleter
  32 {
  33     explicit bt_common_lttng_live_url_parts_deleter(bt_common_lttng_live_url_parts& obj) noexcept :
  34         _mObj {&obj}
  35     {
  36     }
  37
  38     bt_common_lttng_live_url_parts_deleter(const bt_common_lttng_live_url_parts_deleter&) = delete;
  39     bt_common_lttng_live_url_parts&
  40     operator=(const bt_common_lttng_live_url_parts_deleter&) = delete;
  41
  42     ~bt_common_lttng_live_url_parts_deleter()
  43     {
  44         bt_common_destroy_lttng_live_url_parts(_mObj);
  45     }
  46
  47 private:
  48     bt_common_lttng_live_url_parts *_mObj;
  49 };
  50
  51 enum lttng_live_stream_state
  52 {
  53     /* This stream won't have data until some known time in the future. */
  54     LTTNG_LIVE_STREAM_QUIESCENT,
  55     /*
  56      * This stream won't have data until some known time in the future and
  57      * the message iterator inactivity message was already sent downstream.
  58      */
  59     LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA, /* */
  60     /* This stream has data ready to be consumed. */
  61     LTTNG_LIVE_STREAM_ACTIVE_DATA,
  62     /*
  63      * This stream has no data left to consume. We should asked the relay
  64      * for more.
  65      */
  66     LTTNG_LIVE_STREAM_ACTIVE_NO_DATA,
  67     /* This stream won't have anymore data, ever. */
  68     LTTNG_LIVE_STREAM_EOF,
  69 };
  70
  71 inline const char *format_as(const lttng_live_stream_state state) noexcept
  72 {
  73     switch (state) {
  74     case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
  75         return "ACTIVE_NO_DATA";
  76
  77     case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
  78         return "QUIESCENT_NO_DATA";
  79
  80     case LTTNG_LIVE_STREAM_QUIESCENT:
  81         return "QUIESCENT";
  82
  83     case LTTNG_LIVE_STREAM_ACTIVE_DATA:
  84         return "ACTIVE_DATA";
  85
  86     case LTTNG_LIVE_STREAM_EOF:
  87         return "EOF";
  88     }
  89
  90     bt_common_abort();
  91 }
  92
  93 /* Iterator over a live stream. */
  94 struct lttng_live_stream_iterator
  95 {
  96     using UP = std::unique_ptr<lttng_live_stream_iterator>;
  97
  98     explicit lttng_live_stream_iterator(const bt2c::Logger& parentLogger) :
  99         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/STREAM-ITER"}
 100     {
 101     }
 102
 103     ~lttng_live_stream_iterator();
 104
 105     bt2c::Logger logger;
 106
 107     bt2::Stream::Shared stream;
 108
 109     /* Weak reference. */
 110     struct lttng_live_trace *trace = nullptr;
 111
 112     /*
 113      * Since only a single iterator per viewer connection, we have
 114      * only a single message iterator per stream.
 115      */
 116     ctf_msg_iter_up msg_iter;
 117
 118     uint64_t viewer_stream_id = 0;
 119
 120     struct
 121     {
 122         bool is_set = false;
 123         uint64_t value = 0;
 124     } ctf_stream_class_id;
 125
 126     /* base offset in current index. */
 127     uint64_t base_offset = 0;
 128     /* len to read in current index. */
 129     uint64_t len = 0;
 130     /* offset in current index. */
 131     uint64_t offset = 0;
 132
 133     /*
 134      * Clock Snapshot value of the last message iterator inactivity message
 135      * sent downstream.
 136      */
 137     struct
 138     {
 139         bool is_set = false;
 140         uint64_t value = 0;
 141     } last_inactivity_ts;
 142
 143     /*
 144      * Clock Snapshot value of the current message iterator inactivity
 145      * message we might want to send downstream.
 146      */
 147     uint64_t current_inactivity_ts = 0;
 148
 149     enum lttng_live_stream_state state = LTTNG_LIVE_STREAM_QUIESCENT;
 150
 151     /* The current message produced by this live stream iterator. */
 152     bt2::ConstMessage::Shared current_msg;
 153
 154     /* Timestamp in nanoseconds of the current message (current_msg). */
 155     int64_t current_msg_ts_ns = 0;
 156
 157     std::vector<uint8_t> buf;
 158
 159     std::string name;
 160
 161     bool has_stream_hung_up = false;
 162 };
 163
 164 struct lttng_live_metadata
 165 {
 166     using UP = std::unique_ptr<lttng_live_metadata>;
 167
 168     explicit lttng_live_metadata(const bt2c::Logger& parentLogger) :
 169         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/METADATA"}
 170     {
 171     }
 172
 173     bt2c::Logger logger;
 174
 175     uint64_t stream_id = 0;
 176
 177     /* Weak reference. */
 178     ctf_metadata_decoder_up decoder;
 179 };
 180
 181 enum lttng_live_metadata_stream_state
 182 {
 183     /*
 184      * The metadata needs to be updated. This is either because we just
 185      * created the trace and haven't asked yet, or the relay specifically
 186      * told us that new metadata is available.
 187      */
 188     LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED,
 189     /*
 190      * The metadata was updated and the relay has not told us we need to
 191      * update it yet.
 192      */
 193     LTTNG_LIVE_METADATA_STREAM_STATE_NOT_NEEDED,
 194     /*
 195      * The relay has closed this metadata stream. We set this in reaction
 196      * to a LTTNG_VIEWER_METADATA_ERR reply to a LTTNG_VIEWER_GET_METADATA
 197      * command to the relay. If this field is set, we have received all the
 198      * metadata that we are ever going to get for that metadata stream.
 199      */
 200     LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED,
 201 };
 202
 203 struct lttng_live_trace
 204 {
 205     using UP = std::unique_ptr<lttng_live_trace>;
 206
 207     explicit lttng_live_trace(const bt2c::Logger& parentLogger) :
 208         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/TRACE"}
 209     {
 210     }
 211
 212     bt2c::Logger logger;
 213
 214     /* Back reference to session. */
 215     struct lttng_live_session *session = nullptr;
 216
 217     /* ctf trace ID within the session. */
 218     uint64_t id = 0;
 219
 220     bt2::Trace::Shared trace;
 221
 222     bt2::TraceClass::Shared trace_class;
 223
 224     lttng_live_metadata::UP metadata;
 225
 226     const bt_clock_class *clock_class = nullptr;
 227
 228     std::vector<lttng_live_stream_iterator::UP> stream_iterators;
 229
 230     enum lttng_live_metadata_stream_state metadata_stream_state =
 231         LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
 232 };
 233
 234 struct lttng_live_session
 235 {
 236     using UP = std::unique_ptr<lttng_live_session>;
 237
 238     explicit lttng_live_session(const bt2c::Logger& parentLogger) :
 239         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/SESSION"}
 240     {
 241     }
 242
 243     ~lttng_live_session();
 244
 245     bt2c::Logger logger;
 246
 247     bt_self_component *self_comp = nullptr;
 248
 249     /* Weak reference. */
 250     struct lttng_live_msg_iter *lttng_live_msg_iter = nullptr;
 251
 252     std::string hostname;
 253
 254     std::string session_name;
 255
 256     uint64_t id = 0;
 257
 258     std::vector<lttng_live_trace::UP> traces;
 259
 260     bool attached = false;
 261     bool new_streams_needed = false;
 262     bool lazy_stream_msg_init = false;
 263     bool closed = false;
 264 };
 265
 266 enum session_not_found_action
 267 {
 268     SESSION_NOT_FOUND_ACTION_CONTINUE,
 269     SESSION_NOT_FOUND_ACTION_FAIL,
 270     SESSION_NOT_FOUND_ACTION_END,
 271 };
 272
 273 /*
 274  * A component instance is an iterator on a single session.
 275  */
 276 struct lttng_live_component
 277 {
 278     using UP = std::unique_ptr<lttng_live_component>;
 279
 280     explicit lttng_live_component(bt2c::Logger loggerParam) noexcept :
 281         logger {std::move(loggerParam)}
 282     {
 283     }
 284
 285     bt2c::Logger logger;
 286
 287     /* Weak reference. */
 288     bt_self_component *self_comp = nullptr;
 289
 290     struct
 291     {
 292         std::string url;
 293         enum session_not_found_action sess_not_found_act = SESSION_NOT_FOUND_ACTION_CONTINUE;
 294     } params;
 295
 296     size_t max_query_size = 0;
 297
 298     /*
 299      * Keeps track of whether the downstream component already has a
 300      * message iterator on this component.
 301      */
 302     bool has_msg_iter = false;
 303 };
 304
 305 struct lttng_live_msg_iter
 306 {
 307     using UP = std::unique_ptr<lttng_live_msg_iter>;
 308
 309     explicit lttng_live_msg_iter(const bt2c::Logger& parentLogger) :
 310         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/MSG-ITER"}
 311     {
 312     }
 313
 314     ~lttng_live_msg_iter();
 315
 316     bt2c::Logger logger;
 317
 318     bt_self_component *self_comp = nullptr;
 319
 320     /* Weak reference. */
 321     struct lttng_live_component *lttng_live_comp = nullptr;
 322
 323     /* Weak reference. */
 324     bt_self_message_iterator *self_msg_iter = nullptr;
 325
 326     live_viewer_connection::UP viewer_connection;
 327
 328     std::vector<lttng_live_session::UP> sessions;
 329
 330     /* Number of live stream iterator this message iterator has.*/
 331     uint64_t active_stream_iter = 0;
 332
 333     /* Timestamp in nanosecond of the last message sent downstream. */
 334     int64_t last_msg_ts_ns = 0;
 335
 336     /* True if the iterator was interrupted. */
 337     bool was_interrupted = false;
 338 };
 339
 340 enum lttng_live_iterator_status
 341 {
 342     /** Iterator state has progressed. Continue iteration immediately. */
 343     LTTNG_LIVE_ITERATOR_STATUS_CONTINUE = 3,
 344     /** No message available for now. Try again later. */
 345     LTTNG_LIVE_ITERATOR_STATUS_AGAIN = 2,
 346     /** No more CTF_LTTNG_LIVEs to be delivered. */
 347     LTTNG_LIVE_ITERATOR_STATUS_END = 1,
 348     /** No error, okay. */
 349     LTTNG_LIVE_ITERATOR_STATUS_OK = 0,
 350     /** Invalid arguments. */
 351     LTTNG_LIVE_ITERATOR_STATUS_INVAL = -1,
 352     /** General error. */
 353     LTTNG_LIVE_ITERATOR_STATUS_ERROR = -2,
 354     /** Out of memory. */
 355     LTTNG_LIVE_ITERATOR_STATUS_NOMEM = -3,
 356     /** Unsupported iterator feature. */
 357     LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED = -4,
 358 };
 359
 360 inline const char *format_as(const lttng_live_iterator_status status) noexcept
 361 {
 362     switch (status) {
 363     case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
 364         return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
 365
 366     case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
 367         return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
 368
 369     case LTTNG_LIVE_ITERATOR_STATUS_END:
 370         return "LTTNG_LIVE_ITERATOR_STATUS_END";
 371
 372     case LTTNG_LIVE_ITERATOR_STATUS_OK:
 373         return "LTTNG_LIVE_ITERATOR_STATUS_OK";
 374
 375     case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
 376         return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
 377
 378     case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
 379         return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
 380
 381     case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
 382         return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
 383
 384     case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
 385         return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
 386     }
 387
 388     bt_common_abort();
 389 }
 390
 391 bt_component_class_initialize_method_status
 392 lttng_live_component_init(bt_self_component_source *self_comp,
 393                           bt_self_component_source_configuration *config, const bt_value *params,
 394                           void *init_method_data);
 395
 396 bt_component_class_query_method_status lttng_live_query(bt_self_component_class_source *comp_class,
 397                                                         bt_private_query_executor *priv_query_exec,
 398                                                         const char *object, const bt_value *params,
 399                                                         void *method_data, const bt_value **result);
 400
 401 void lttng_live_component_finalize(bt_self_component_source *component);
 402
 403 bt_message_iterator_class_next_method_status
 404 lttng_live_msg_iter_next(bt_self_message_iterator *iterator, bt_message_array_const msgs,
 405                          uint64_t capacity, uint64_t *count);
 406
 407 bt_message_iterator_class_initialize_method_status
 408 lttng_live_msg_iter_init(bt_self_message_iterator *self_msg_it,
 409                          bt_self_message_iterator_configuration *config,
 410                          bt_self_component_port_output *self_port);
 411
 412 void lttng_live_msg_iter_finalize(bt_self_message_iterator *it);
 413
 414 enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session,
 415                                                         bt_self_message_iterator *self_msg_iter);
 416
 417 enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_session *session);
 418
 419 enum lttng_live_iterator_status
 420 lttng_live_session_get_new_streams(struct lttng_live_session *session,
 421                                    bt_self_message_iterator *self_msg_iter);
 422
 423 struct lttng_live_trace *
 424 lttng_live_session_borrow_or_create_trace_by_id(struct lttng_live_session *session,
 425                                                 uint64_t trace_id);
 426
 427 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, uint64_t session_id,
 428                            const char *hostname, const char *session_name);
 429
 430 /*
 431  * lttng_live_get_one_metadata_packet() asks the Relay Daemon for new metadata.
 432  * If new metadata is received, the function writes it to the provided file
 433  * handle and updates the reply_len output parameter. This function should be
 434  * called in loop until _END status is received to ensure all metadata is
 435  * written to the file.
 436  */
 437 enum lttng_live_get_one_metadata_status
 438 lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<char>& buf);
 439
 440 enum lttng_live_iterator_status
 441 lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
 442                           struct lttng_live_stream_iterator *stream, struct packet_index *index);
 443
 444 enum ctf_msg_iter_medium_status
 445 lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
 446                             struct lttng_live_stream_iterator *stream, uint8_t *buf,
 447                             uint64_t offset, uint64_t req_len, uint64_t *recv_len);
 448
 449 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter);
 450
 451 void lttng_live_stream_iterator_set_state(struct lttng_live_stream_iterator *stream_iter,
 452                                           enum lttng_live_stream_state new_state);
 453
 454 #endif /* BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H */