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 <stdbool.h>
  15 #include <stdint.h>
  16
  17 #include <glib.h>
  18
  19 #include <babeltrace2/babeltrace.h>
  20
  21 #include "common/macros.h"
  22 #include "cpp-common/bt2/message.hpp"
  23 #include "../common/src/metadata/tsdl/decoder.hpp"
  24 #include "../common/src/msg-iter/msg-iter.hpp"
  25 #include "viewer-connection.hpp"
  26
  27 struct lttng_live_component;
  28 struct lttng_live_session;
  29 struct lttng_live_msg_iter;
  30
  31 /*
  32  * bt_common_lttng_live_url_parts is defined in common code, and is also used
  33  * by C code, so it can't be C++-ified yet.  Use this separate deleter object
  34  * in the mean time.
  35  */
  36 struct bt_common_lttng_live_url_parts_deleter
  37 {
  38     bt_common_lttng_live_url_parts_deleter(bt_common_lttng_live_url_parts& obj) : _mObj {obj}
  39     {
  40     }
  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 /* Iterator over a live stream. */
  72 struct lttng_live_stream_iterator
  73 {
  74     using UP = std::unique_ptr<lttng_live_stream_iterator>;
  75
  76     explicit lttng_live_stream_iterator(const bt2_common::LogCfg& logCfgParam) noexcept :
  77         logCfg {logCfgParam}
  78     {
  79     }
  80
  81     ~lttng_live_stream_iterator();
  82
  83     const bt2_common::LogCfg logCfg;
  84
  85     /* Owned by this. */
  86     nonstd::optional<bt2::Stream::Shared> stream;
  87
  88     /* Weak reference. */
  89     struct lttng_live_trace *trace = nullptr;
  90
  91     /*
  92      * Since only a single iterator per viewer connection, we have
  93      * only a single message iterator per stream.
  94      */
  95     ctf_msg_iter_up msg_iter;
  96
  97     uint64_t viewer_stream_id = 0;
  98
  99     struct
 100     {
 101         bool is_set = false;
 102         uint64_t value = 0;
 103     } ctf_stream_class_id;
 104
 105     /* base offset in current index. */
 106     uint64_t base_offset = 0;
 107     /* len to read in current index. */
 108     uint64_t len = 0;
 109     /* offset in current index. */
 110     uint64_t offset = 0;
 111
 112     /*
 113      * Clock Snapshot value of the last message iterator inactivity message
 114      * sent downstream.
 115      */
 116     struct
 117     {
 118         bool is_set = false;
 119         uint64_t value = 0;
 120     } last_inactivity_ts;
 121
 122     /*
 123      * Clock Snapshot value of the current message iterator inactivity
 124      * message we might want to send downstream.
 125      */
 126     uint64_t current_inactivity_ts = 0;
 127
 128     enum lttng_live_stream_state state = LTTNG_LIVE_STREAM_QUIESCENT;
 129
 130     /*
 131      * The current message produced by this live stream iterator. Owned by
 132      * this.
 133      */
 134     nonstd::optional<bt2::ConstMessage::Shared> current_msg;
 135
 136     /* Timestamp in nanoseconds of the current message (current_msg). */
 137     int64_t current_msg_ts_ns = 0;
 138
 139     std::vector<uint8_t> buf;
 140
 141     std::string name;
 142
 143     bool has_stream_hung_up = false;
 144 };
 145
 146 struct lttng_live_metadata
 147 {
 148     using UP = std::unique_ptr<lttng_live_metadata>;
 149
 150     explicit lttng_live_metadata(const bt2_common::LogCfg& logCfgParam) noexcept :
 151         logCfg {logCfgParam}
 152     {
 153     }
 154
 155     const bt2_common::LogCfg logCfg;
 156
 157     uint64_t stream_id = 0;
 158
 159     /* Weak reference. */
 160     ctf_metadata_decoder_up decoder;
 161 };
 162
 163 enum lttng_live_metadata_stream_state
 164 {
 165     /*
 166      * The metadata needs to be updated. This is either because we just
 167      * created the trace and haven't asked yet, or the relay specifically
 168      * told us that new metadata is available.
 169      */
 170     LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED,
 171     /*
 172      * The metadata was updated and the relay has not told us we need to
 173      * update it yet.
 174      */
 175     LTTNG_LIVE_METADATA_STREAM_STATE_NOT_NEEDED,
 176     /*
 177      * The relay has closed this metadata stream. We set this in reaction
 178      * to a LTTNG_VIEWER_METADATA_ERR reply to a LTTNG_VIEWER_GET_METADATA
 179      * command to the relay. If this field is set, we have received all the
 180      * metadata that we are ever going to get for that metadata stream.
 181      */
 182     LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED,
 183 };
 184
 185 struct lttng_live_trace
 186 {
 187     using UP = std::unique_ptr<lttng_live_trace>;
 188
 189     explicit lttng_live_trace(const bt2_common::LogCfg& logCfgParam) noexcept : logCfg {logCfgParam}
 190     {
 191     }
 192
 193     const bt2_common::LogCfg logCfg;
 194
 195     /* Back reference to session. */
 196     struct lttng_live_session *session = nullptr;
 197
 198     /* ctf trace ID within the session. */
 199     uint64_t id = 0;
 200
 201     /* Owned by this. */
 202     nonstd::optional<bt2::Trace::Shared> trace;
 203
 204     nonstd::optional<bt2::TraceClass::Shared> trace_class;
 205
 206     lttng_live_metadata::UP metadata;
 207
 208     const bt_clock_class *clock_class = nullptr;
 209
 210     std::vector<lttng_live_stream_iterator::UP> stream_iterators;
 211
 212     enum lttng_live_metadata_stream_state metadata_stream_state =
 213         LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
 214 };
 215
 216 struct lttng_live_session
 217 {
 218     using UP = std::unique_ptr<lttng_live_session>;
 219
 220     explicit lttng_live_session(const bt2_common::LogCfg& logCfgParam) noexcept :
 221         logCfg {logCfgParam}
 222     {
 223     }
 224
 225     ~lttng_live_session();
 226
 227     const bt2_common::LogCfg logCfg;
 228
 229     bt_self_component *self_comp = nullptr;
 230
 231     /* Weak reference. */
 232     struct lttng_live_msg_iter *lttng_live_msg_iter = nullptr;
 233
 234     std::string hostname;
 235
 236     std::string session_name;
 237
 238     uint64_t id = 0;
 239
 240     std::vector<lttng_live_trace::UP> traces;
 241
 242     bool attached = false;
 243     bool new_streams_needed = false;
 244     bool lazy_stream_msg_init = false;
 245     bool closed = false;
 246 };
 247
 248 enum session_not_found_action
 249 {
 250     SESSION_NOT_FOUND_ACTION_CONTINUE,
 251     SESSION_NOT_FOUND_ACTION_FAIL,
 252     SESSION_NOT_FOUND_ACTION_END,
 253 };
 254
 255 /*
 256  * A component instance is an iterator on a single session.
 257  */
 258 struct lttng_live_component
 259 {
 260     using UP = std::unique_ptr<lttng_live_component>;
 261
 262     explicit lttng_live_component(const bt2_common::LogCfg& logCfgParam) noexcept :
 263         logCfg {logCfgParam}
 264     {
 265     }
 266
 267     const bt2_common::LogCfg logCfg;
 268
 269     /* Weak reference. */
 270     bt_self_component *self_comp = nullptr;
 271
 272     struct
 273     {
 274         std::string url;
 275         enum session_not_found_action sess_not_found_act = SESSION_NOT_FOUND_ACTION_CONTINUE;
 276     } params;
 277
 278     size_t max_query_size = 0;
 279
 280     /*
 281      * Keeps track of whether the downstream component already has a
 282      * message iterator on this component.
 283      */
 284     bool has_msg_iter = false;
 285 };
 286
 287 struct lttng_live_msg_iter
 288 {
 289     using UP = std::unique_ptr<lttng_live_msg_iter>;
 290
 291     explicit lttng_live_msg_iter(const bt2_common::LogCfg& logCfgParam) noexcept :
 292         logCfg {logCfgParam}
 293     {
 294     }
 295
 296     ~lttng_live_msg_iter();
 297
 298     const bt2_common::LogCfg logCfg;
 299
 300     bt_self_component *self_comp = nullptr;
 301
 302     /* Weak reference. */
 303     struct lttng_live_component *lttng_live_comp = nullptr;
 304
 305     /* Weak reference. */
 306     bt_self_message_iterator *self_msg_iter = nullptr;
 307
 308     live_viewer_connection::UP viewer_connection;
 309
 310     std::vector<lttng_live_session::UP> sessions;
 311
 312     /* Number of live stream iterator this message iterator has.*/
 313     uint64_t active_stream_iter = 0;
 314
 315     /* Timestamp in nanosecond of the last message sent downstream. */
 316     int64_t last_msg_ts_ns = 0;
 317
 318     /* True if the iterator was interrupted. */
 319     bool was_interrupted = false;
 320 };
 321
 322 enum lttng_live_iterator_status
 323 {
 324     /** Iterator state has progressed. Continue iteration immediately. */
 325     LTTNG_LIVE_ITERATOR_STATUS_CONTINUE = 3,
 326     /** No message available for now. Try again later. */
 327     LTTNG_LIVE_ITERATOR_STATUS_AGAIN = 2,
 328     /** No more CTF_LTTNG_LIVEs to be delivered. */
 329     LTTNG_LIVE_ITERATOR_STATUS_END = 1,
 330     /** No error, okay. */
 331     LTTNG_LIVE_ITERATOR_STATUS_OK = 0,
 332     /** Invalid arguments. */
 333     LTTNG_LIVE_ITERATOR_STATUS_INVAL = -1,
 334     /** General error. */
 335     LTTNG_LIVE_ITERATOR_STATUS_ERROR = -2,
 336     /** Out of memory. */
 337     LTTNG_LIVE_ITERATOR_STATUS_NOMEM = -3,
 338     /** Unsupported iterator feature. */
 339     LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED = -4,
 340 };
 341
 342 bt_component_class_initialize_method_status
 343 lttng_live_component_init(bt_self_component_source *self_comp,
 344                           bt_self_component_source_configuration *config, const bt_value *params,
 345                           void *init_method_data);
 346
 347 bt_component_class_query_method_status lttng_live_query(bt_self_component_class_source *comp_class,
 348                                                         bt_private_query_executor *priv_query_exec,
 349                                                         const char *object, const bt_value *params,
 350                                                         void *method_data, const bt_value **result);
 351
 352 void lttng_live_component_finalize(bt_self_component_source *component);
 353
 354 bt_message_iterator_class_next_method_status
 355 lttng_live_msg_iter_next(bt_self_message_iterator *iterator, bt_message_array_const msgs,
 356                          uint64_t capacity, uint64_t *count);
 357
 358 bt_message_iterator_class_initialize_method_status
 359 lttng_live_msg_iter_init(bt_self_message_iterator *self_msg_it,
 360                          bt_self_message_iterator_configuration *config,
 361                          bt_self_component_port_output *self_port);
 362
 363 void lttng_live_msg_iter_finalize(bt_self_message_iterator *it);
 364
 365 enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session,
 366                                                         bt_self_message_iterator *self_msg_iter);
 367
 368 enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_session *session);
 369
 370 enum lttng_live_iterator_status
 371 lttng_live_session_get_new_streams(struct lttng_live_session *session,
 372                                    bt_self_message_iterator *self_msg_iter);
 373
 374 struct lttng_live_trace *
 375 lttng_live_session_borrow_or_create_trace_by_id(struct lttng_live_session *session,
 376                                                 uint64_t trace_id);
 377
 378 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, uint64_t session_id,
 379                            const char *hostname, const char *session_name);
 380
 381 /*
 382  * lttng_live_get_one_metadata_packet() asks the Relay Daemon for new metadata.
 383  * If new metadata is received, the function writes it to the provided file
 384  * handle and updates the reply_len output parameter. This function should be
 385  * called in loop until _END status is received to ensure all metadata is
 386  * written to the file.
 387  */
 388 enum lttng_live_get_one_metadata_status
 389 lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<char>& buf);
 390
 391 enum lttng_live_iterator_status
 392 lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
 393                           struct lttng_live_stream_iterator *stream, struct packet_index *index);
 394
 395 enum ctf_msg_iter_medium_status
 396 lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
 397                             struct lttng_live_stream_iterator *stream, uint8_t *buf,
 398                             uint64_t offset, uint64_t req_len, uint64_t *recv_len);
 399
 400 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter);
 401
 402 BT_HIDDEN
 403 void lttng_live_stream_iterator_set_state(struct lttng_live_stream_iterator *stream_iter,
 404                                           enum lttng_live_stream_state new_state);
 405
 406 #endif /* BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H */