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_PLUGINS_CTF_LTTNG_LIVE_LTTNG_LIVE_HPP
  12 #define BABELTRACE_PLUGINS_CTF_LTTNG_LIVE_LTTNG_LIVE_HPP
  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/metadata-stream-parser-utils.hpp"
  23 #include "../common/src/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     bt2s::optional<ctf::src::MsgIter> 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     /*
 127      * Clock Snapshot value of the last message iterator inactivity message
 128      * sent downstream.
 129      */
 130     struct
 131     {
 132         bool is_set = false;
 133         uint64_t value = 0;
 134     } last_inactivity_ts;
 135
 136     /*
 137      * Clock Snapshot value of the current message iterator inactivity
 138      * message we might want to send downstream.
 139      */
 140     uint64_t current_inactivity_ts = 0;
 141
 142     enum lttng_live_stream_state state = LTTNG_LIVE_STREAM_QUIESCENT;
 143
 144     /* The current message produced by this live stream iterator. */
 145     bt2::ConstMessage::Shared current_msg;
 146
 147     /* Timestamp in nanoseconds of the current message (current_msg). */
 148     int64_t current_msg_ts_ns = 0;
 149
 150     std::string name;
 151
 152     bool has_stream_hung_up = false;
 153
 154     struct CurPktInfo
 155     {
 156         bt2c::DataLen offsetInRelay;
 157         bt2c::DataLen len;
 158     };
 159
 160     bt2s::optional<CurPktInfo> curPktInfo;
 161 };
 162
 163 struct lttng_live_metadata
 164 {
 165     using UP = std::unique_ptr<lttng_live_metadata>;
 166
 167     explicit lttng_live_metadata(const bt2::SelfComponent selfComp,
 168                                  const bt2c::Logger& parentLogger) :
 169         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/METADATA"},
 170         _mSelfComp {selfComp}
 171
 172     {
 173     }
 174
 175     const ctf::src::TraceCls *traceCls() const
 176     {
 177         return _mMetadataStreamParser->traceCls();
 178     }
 179
 180     const bt2s::optional<bt2c::Uuid>& metadataStreamUuid() const noexcept
 181     {
 182         return _mMetadataStreamParser->metadataStreamUuid();
 183     }
 184
 185     void parseSection(const bt2c::ConstBytes data)
 186     {
 187         if (!_mMetadataStreamParser) {
 188             _mMetadataStreamParser =
 189                 ctf::src::createMetadataStreamParser(data, _mSelfComp, {}, logger);
 190         }
 191
 192         _mMetadataStreamParser->parseSection(data);
 193     }
 194
 195     bt2::SelfComponent selfComp() const noexcept
 196     {
 197         return _mSelfComp;
 198     }
 199
 200     bt2c::Logger logger;
 201
 202     uint64_t stream_id = 0;
 203
 204 private:
 205     bt2::SelfComponent _mSelfComp;
 206     ctf::src::MetadataStreamParser::UP _mMetadataStreamParser;
 207 };
 208
 209 enum lttng_live_metadata_stream_state
 210 {
 211     /*
 212      * The metadata needs to be updated. This is either because we just
 213      * created the trace and haven't asked yet, or the relay specifically
 214      * told us that new metadata is available.
 215      */
 216     LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED,
 217     /*
 218      * The metadata was updated and the relay has not told us we need to
 219      * update it yet.
 220      */
 221     LTTNG_LIVE_METADATA_STREAM_STATE_NOT_NEEDED,
 222     /*
 223      * The relay has closed this metadata stream. We set this in reaction
 224      * to a LTTNG_VIEWER_METADATA_ERR reply to a LTTNG_VIEWER_GET_METADATA
 225      * command to the relay. If this field is set, we have received all the
 226      * metadata that we are ever going to get for that metadata stream.
 227      */
 228     LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED,
 229 };
 230
 231 struct lttng_live_trace
 232 {
 233     using UP = std::unique_ptr<lttng_live_trace>;
 234
 235     explicit lttng_live_trace(const bt2c::Logger& parentLogger) :
 236         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/TRACE"}
 237     {
 238     }
 239
 240     bt2c::Logger logger;
 241
 242     /* Back reference to session. */
 243     struct lttng_live_session *session = nullptr;
 244
 245     /* ctf trace ID within the session. */
 246     uint64_t id = 0;
 247
 248     bt2::Trace::Shared trace;
 249
 250     lttng_live_metadata::UP metadata;
 251
 252     bt2::OptionalBorrowedObject<bt2::ConstClockClass> clock_class;
 253
 254     std::vector<lttng_live_stream_iterator::UP> stream_iterators;
 255
 256     enum lttng_live_metadata_stream_state metadata_stream_state =
 257         LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
 258 };
 259
 260 struct lttng_live_session
 261 {
 262     using UP = std::unique_ptr<lttng_live_session>;
 263
 264     explicit lttng_live_session(const bt2c::Logger& parentLogger,
 265                                 const bt2::SelfComponent selfCompParam) :
 266         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/SESSION"},
 267         selfComp {selfCompParam}
 268     {
 269     }
 270
 271     ~lttng_live_session();
 272
 273     bt2c::Logger logger;
 274
 275     bt2::SelfComponent selfComp;
 276
 277     /* Weak reference. */
 278     struct lttng_live_msg_iter *lttng_live_msg_iter = nullptr;
 279
 280     std::string hostname;
 281
 282     std::string session_name;
 283
 284     uint64_t id = 0;
 285
 286     std::vector<lttng_live_trace::UP> traces;
 287
 288     bool attached = false;
 289     bool new_streams_needed = false;
 290     bool lazy_stream_msg_init = false;
 291     bool closed = false;
 292 };
 293
 294 enum session_not_found_action
 295 {
 296     SESSION_NOT_FOUND_ACTION_CONTINUE,
 297     SESSION_NOT_FOUND_ACTION_FAIL,
 298     SESSION_NOT_FOUND_ACTION_END,
 299 };
 300
 301 /*
 302  * A component instance is an iterator on a single session.
 303  */
 304 struct lttng_live_component
 305 {
 306     using UP = std::unique_ptr<lttng_live_component>;
 307
 308     explicit lttng_live_component(bt2c::Logger loggerParam,
 309                                   const bt2::SelfComponent selfCompParam) noexcept :
 310         logger {std::move(loggerParam)},
 311         selfComp {selfCompParam}
 312     {
 313     }
 314
 315     bt2c::Logger logger;
 316
 317     bt2::SelfComponent selfComp;
 318
 319     struct
 320     {
 321         std::string url;
 322         enum session_not_found_action sess_not_found_act = SESSION_NOT_FOUND_ACTION_CONTINUE;
 323     } params;
 324
 325     size_t max_query_size = 0;
 326
 327     /*
 328      * Keeps track of whether the downstream component already has a
 329      * message iterator on this component.
 330      */
 331     bool has_msg_iter = false;
 332 };
 333
 334 struct lttng_live_msg_iter
 335 {
 336     using UP = std::unique_ptr<lttng_live_msg_iter>;
 337
 338     explicit lttng_live_msg_iter(const bt2c::Logger& parentLogger,
 339                                  const bt2::SelfComponent selfCompParam,
 340                                  const bt2::SelfMessageIterator selfMsgIter) :
 341         logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/MSG-ITER"},
 342         selfComp {selfCompParam}, selfMsgIter {selfMsgIter}
 343     {
 344     }
 345
 346     ~lttng_live_msg_iter();
 347
 348     bt2c::Logger logger;
 349
 350     bt2::SelfComponent selfComp;
 351
 352     /* Weak reference. */
 353     struct lttng_live_component *lttng_live_comp = nullptr;
 354
 355     bt2::SelfMessageIterator selfMsgIter;
 356
 357     live_viewer_connection::UP viewer_connection;
 358
 359     std::vector<lttng_live_session::UP> sessions;
 360
 361     /* Number of live stream iterator this message iterator has.*/
 362     uint64_t active_stream_iter = 0;
 363
 364     /* Timestamp in nanosecond of the last message sent downstream. */
 365     int64_t last_msg_ts_ns = 0;
 366
 367     /* True if the iterator was interrupted. */
 368     bool was_interrupted = false;
 369 };
 370
 371 enum lttng_live_iterator_status
 372 {
 373     /** Iterator state has progressed. Continue iteration immediately. */
 374     LTTNG_LIVE_ITERATOR_STATUS_CONTINUE = 3,
 375     /** No message available for now. Try again later. */
 376     LTTNG_LIVE_ITERATOR_STATUS_AGAIN = 2,
 377     /** No more CTF_LTTNG_LIVEs to be delivered. */
 378     LTTNG_LIVE_ITERATOR_STATUS_END = 1,
 379     /** No error, okay. */
 380     LTTNG_LIVE_ITERATOR_STATUS_OK = 0,
 381     /** Invalid arguments. */
 382     LTTNG_LIVE_ITERATOR_STATUS_INVAL = -1,
 383     /** General error. */
 384     LTTNG_LIVE_ITERATOR_STATUS_ERROR = -2,
 385     /** Out of memory. */
 386     LTTNG_LIVE_ITERATOR_STATUS_NOMEM = -3,
 387     /** Unsupported iterator feature. */
 388     LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED = -4,
 389 };
 390
 391 inline const char *format_as(const lttng_live_iterator_status status) noexcept
 392 {
 393     switch (status) {
 394     case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
 395         return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
 396
 397     case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
 398         return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
 399
 400     case LTTNG_LIVE_ITERATOR_STATUS_END:
 401         return "LTTNG_LIVE_ITERATOR_STATUS_END";
 402
 403     case LTTNG_LIVE_ITERATOR_STATUS_OK:
 404         return "LTTNG_LIVE_ITERATOR_STATUS_OK";
 405
 406     case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
 407         return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
 408
 409     case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
 410         return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
 411
 412     case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
 413         return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
 414
 415     case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
 416         return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
 417     }
 418
 419     bt_common_abort();
 420 }
 421
 422 bt_component_class_initialize_method_status
 423 lttng_live_component_init(bt_self_component_source *self_comp,
 424                           bt_self_component_source_configuration *config, const bt_value *params,
 425                           void *init_method_data);
 426
 427 bt_component_class_query_method_status lttng_live_query(bt_self_component_class_source *comp_class,
 428                                                         bt_private_query_executor *priv_query_exec,
 429                                                         const char *object, const bt_value *params,
 430                                                         void *method_data, const bt_value **result);
 431
 432 void lttng_live_component_finalize(bt_self_component_source *component);
 433
 434 bt_message_iterator_class_next_method_status
 435 lttng_live_msg_iter_next(bt_self_message_iterator *self_msg_it, bt_message_array_const msgs,
 436                          uint64_t capacity, uint64_t *count);
 437
 438 bt_message_iterator_class_initialize_method_status
 439 lttng_live_msg_iter_init(bt_self_message_iterator *self_msg_it,
 440                          bt_self_message_iterator_configuration *config,
 441                          bt_self_component_port_output *self_port);
 442
 443 void lttng_live_msg_iter_finalize(bt_self_message_iterator *it);
 444
 445 enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session);
 446
 447 enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_session *session);
 448
 449 enum lttng_live_iterator_status
 450 lttng_live_session_get_new_streams(struct lttng_live_session *session);
 451
 452 struct lttng_live_trace *
 453 lttng_live_session_borrow_or_create_trace_by_id(struct lttng_live_session *session,
 454                                                 uint64_t trace_id);
 455
 456 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, uint64_t session_id,
 457                            const char *hostname, const char *session_name);
 458
 459 /*
 460  * lttng_live_get_one_metadata_packet() asks the Relay Daemon for new metadata.
 461  * If new metadata is received, the function writes it to the provided file
 462  * handle and updates the reply_len output parameter. This function should be
 463  * called in loop until _END status is received to ensure all metadata is
 464  * written to the file.
 465  */
 466 enum lttng_live_get_one_metadata_status
 467 lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<uint8_t>& buf);
 468
 469 enum lttng_live_iterator_status
 470 lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
 471                           struct lttng_live_stream_iterator *stream, struct packet_index *index);
 472
 473 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter);
 474
 475 void lttng_live_stream_iterator_set_state(struct lttng_live_stream_iterator *stream_iter,
 476                                           enum lttng_live_stream_state new_state);
 477
 478 void lttng_live_stream_iterator_set_stream_class(lttng_live_stream_iterator *streamIter,
 479                                                  uint64_t ctfStreamClsId);
 480
 481 #endif /* BABELTRACE_PLUGINS_CTF_LTTNG_LIVE_LTTNG_LIVE_HPP */