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