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