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

   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 CTF LTTng-live Client Component
   9  */
  10
  11 #include <glib.h>
  12 #include <unistd.h>
  13
  14 #include "common/assert.h"
  15 #include "cpp-common/bt2c/fmt.hpp"
  16 #include "cpp-common/vendor/fmt/format.h"
  17
  18 #include "plugins/common/muxing/muxing.h"
  19 #include "plugins/common/param-validation/param-validation.h"
  20
  21 #include "data-stream.hpp"
  22 #include "lttng-live.hpp"
  23 #include "metadata.hpp"
  24
  25 #define MAX_QUERY_SIZE                     (256 * 1024)
  26 #define URL_PARAM                          "url"
  27 #define INPUTS_PARAM                       "inputs"
  28 #define SESS_NOT_FOUND_ACTION_PARAM        "session-not-found-action"
  29 #define SESS_NOT_FOUND_ACTION_CONTINUE_STR "continue"
  30 #define SESS_NOT_FOUND_ACTION_FAIL_STR     "fail"
  31 #define SESS_NOT_FOUND_ACTION_END_STR      "end"
  32
  33 void lttng_live_stream_iterator_set_state(struct lttng_live_stream_iterator *stream_iter,
  34                                           enum lttng_live_stream_state new_state)
  35 {
  36     BT_CPPLOGD_SPEC(stream_iter->logger,
  37                     "Setting live stream iterator state: viewer-stream-id={}, "
  38                     "old-state={}, new-state={}",
  39                     stream_iter->viewer_stream_id, stream_iter->state, new_state);
  40
  41     stream_iter->state = new_state;
  42 }
  43
  44 #define LTTNG_LIVE_LOGD_STREAM_ITER(live_stream_iter)                                              \
  45     do {                                                                                           \
  46         BT_CPPLOGD_SPEC((live_stream_iter)->logger,                                                \
  47                         "Live stream iterator state={}, "                                          \
  48                         "last-inact-ts-is-set={}, last-inact-ts-value={}, "                        \
  49                         "curr-inact-ts={}",                                                        \
  50                         (live_stream_iter)->state, (live_stream_iter)->last_inactivity_ts.is_set,  \
  51                         (live_stream_iter)->last_inactivity_ts.value,                              \
  52                         (live_stream_iter)->current_inactivity_ts);                                \
  53     } while (0);
  54
  55 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter)
  56 {
  57     bool ret;
  58
  59     if (!msg_iter) {
  60         ret = false;
  61         goto end;
  62     }
  63
  64     ret = bt_self_message_iterator_is_interrupted(msg_iter->self_msg_iter);
  65
  66 end:
  67     return ret;
  68 }
  69
  70 static struct lttng_live_trace *
  71 lttng_live_session_borrow_trace_by_id(struct lttng_live_session *session, uint64_t trace_id)
  72 {
  73     uint64_t trace_idx;
  74     struct lttng_live_trace *ret_trace = NULL;
  75
  76     for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
  77         struct lttng_live_trace *trace =
  78             (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
  79         if (trace->id == trace_id) {
  80             ret_trace = trace;
  81             goto end;
  82         }
  83     }
  84
  85 end:
  86     return ret_trace;
  87 }
  88
  89 static void lttng_live_destroy_trace(struct lttng_live_trace *trace)
  90 {
  91     BT_CPPLOGD_SPEC(trace->logger, "Destroying live trace: trace-id={}", trace->id);
  92
  93     BT_ASSERT(trace->stream_iterators);
  94     g_ptr_array_free(trace->stream_iterators, TRUE);
  95
  96     BT_TRACE_PUT_REF_AND_RESET(trace->trace);
  97     BT_TRACE_CLASS_PUT_REF_AND_RESET(trace->trace_class);
  98
  99     lttng_live_metadata_fini(trace);
 100     delete trace;
 101 }
 102
 103 static struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *session,
 104                                                         uint64_t trace_id)
 105 {
 106     BT_CPPLOGD_SPEC(session->logger, "Creating live trace: session-id={}, trace-id={}", session->id,
 107                     trace_id);
 108
 109     lttng_live_trace *trace = new lttng_live_trace {session->logger};
 110     trace->session = session;
 111     trace->id = trace_id;
 112     trace->trace_class = NULL;
 113     trace->trace = NULL;
 114     trace->stream_iterators =
 115         g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_stream_iterator_destroy);
 116     BT_ASSERT(trace->stream_iterators);
 117     trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
 118     g_ptr_array_add(session->traces, trace);
 119
 120     return trace;
 121 }
 122
 123 struct lttng_live_trace *
 124 lttng_live_session_borrow_or_create_trace_by_id(struct lttng_live_session *session,
 125                                                 uint64_t trace_id)
 126 {
 127     struct lttng_live_trace *trace;
 128
 129     trace = lttng_live_session_borrow_trace_by_id(session, trace_id);
 130     if (trace) {
 131         goto end;
 132     }
 133
 134     /* The session is the owner of the newly created trace. */
 135     trace = lttng_live_create_trace(session, trace_id);
 136
 137 end:
 138     return trace;
 139 }
 140
 141 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter, uint64_t session_id,
 142                            const char *hostname, const char *session_name)
 143 {
 144     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 145                     "Adding live session: "
 146                     "session-id={}, hostname=\"{}\", session-name=\"{}\"",
 147                     session_id, hostname, session_name);
 148
 149     lttng_live_session *session = new lttng_live_session {lttng_live_msg_iter->logger};
 150     session->self_comp = lttng_live_msg_iter->self_comp;
 151     session->id = session_id;
 152     session->traces = g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_destroy_trace);
 153     BT_ASSERT(session->traces);
 154     session->lttng_live_msg_iter = lttng_live_msg_iter;
 155     session->new_streams_needed = true;
 156     session->hostname = g_string_new(hostname);
 157     BT_ASSERT(session->hostname);
 158
 159     session->session_name = g_string_new(session_name);
 160     BT_ASSERT(session->session_name);
 161
 162     g_ptr_array_add(lttng_live_msg_iter->sessions, session);
 163
 164     return 0;
 165 }
 166
 167 static void lttng_live_destroy_session(struct lttng_live_session *session)
 168 {
 169     if (!session) {
 170         goto end;
 171     }
 172
 173     BT_CPPLOGD_SPEC(session->logger,
 174                     "Destroying live session: "
 175                     "session-id={}, session-name=\"{}\"",
 176                     session->id, session->session_name->str);
 177     if (session->id != -1ULL) {
 178         if (lttng_live_session_detach(session)) {
 179             if (!lttng_live_graph_is_canceled(session->lttng_live_msg_iter)) {
 180                 /* Old relayd cannot detach sessions. */
 181                 BT_CPPLOGD_SPEC(session->logger, "Unable to detach lttng live session {}",
 182                                 session->id);
 183             }
 184         }
 185         session->id = -1ULL;
 186     }
 187
 188     if (session->traces) {
 189         g_ptr_array_free(session->traces, TRUE);
 190     }
 191
 192     if (session->hostname) {
 193         g_string_free(session->hostname, TRUE);
 194     }
 195
 196     if (session->session_name) {
 197         g_string_free(session->session_name, TRUE);
 198     }
 199
 200     delete session;
 201
 202 end:
 203     return;
 204 }
 205
 206 static void lttng_live_msg_iter_destroy(struct lttng_live_msg_iter *lttng_live_msg_iter)
 207 {
 208     if (!lttng_live_msg_iter) {
 209         goto end;
 210     }
 211
 212     if (lttng_live_msg_iter->sessions) {
 213         g_ptr_array_free(lttng_live_msg_iter->sessions, TRUE);
 214     }
 215
 216     if (lttng_live_msg_iter->viewer_connection) {
 217         live_viewer_connection_destroy(lttng_live_msg_iter->viewer_connection);
 218     }
 219     BT_ASSERT(lttng_live_msg_iter->lttng_live_comp);
 220     BT_ASSERT(lttng_live_msg_iter->lttng_live_comp->has_msg_iter);
 221
 222     /* All stream iterators must be destroyed at this point. */
 223     BT_ASSERT(lttng_live_msg_iter->active_stream_iter == 0);
 224     lttng_live_msg_iter->lttng_live_comp->has_msg_iter = false;
 225
 226     delete lttng_live_msg_iter;
 227
 228 end:
 229     return;
 230 }
 231
 232 void lttng_live_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
 233 {
 234     struct lttng_live_msg_iter *lttng_live_msg_iter;
 235
 236     BT_ASSERT(self_msg_iter);
 237
 238     lttng_live_msg_iter =
 239         (struct lttng_live_msg_iter *) bt_self_message_iterator_get_data(self_msg_iter);
 240     BT_ASSERT(lttng_live_msg_iter);
 241     lttng_live_msg_iter_destroy(lttng_live_msg_iter);
 242 }
 243
 244 static enum lttng_live_iterator_status
 245 lttng_live_iterator_next_check_stream_state(struct lttng_live_stream_iterator *lttng_live_stream)
 246 {
 247     switch (lttng_live_stream->state) {
 248     case LTTNG_LIVE_STREAM_QUIESCENT:
 249     case LTTNG_LIVE_STREAM_ACTIVE_DATA:
 250         break;
 251     case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
 252         /* Invalid state. */
 253         BT_CPPLOGF_SPEC(lttng_live_stream->logger, "Unexpected stream state \"ACTIVE_NO_DATA\"");
 254         bt_common_abort();
 255     case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
 256         /* Invalid state. */
 257         BT_CPPLOGF_SPEC(lttng_live_stream->logger, "Unexpected stream state \"QUIESCENT_NO_DATA\"");
 258         bt_common_abort();
 259     case LTTNG_LIVE_STREAM_EOF:
 260         break;
 261     }
 262     return LTTNG_LIVE_ITERATOR_STATUS_OK;
 263 }
 264
 265 /*
 266  * For active no data stream, fetch next index. As a result of that it can
 267  * become either:
 268  * - quiescent: won't have events for a bit,
 269  * - have data: need to get that data and produce the event,
 270  * - have no data on this stream at this point: need to retry (AGAIN) or return
 271  *   EOF.
 272  */
 273 static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_no_data_stream(
 274     struct lttng_live_msg_iter *lttng_live_msg_iter,
 275     struct lttng_live_stream_iterator *lttng_live_stream)
 276 {
 277     enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 278     enum lttng_live_stream_state orig_state = lttng_live_stream->state;
 279     struct packet_index index;
 280
 281     if (lttng_live_stream->trace->metadata_stream_state ==
 282         LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
 283         BT_CPPLOGD_SPEC(
 284             lttng_live_msg_iter->logger,
 285             "Need to get an update for the metadata stream before proceeding further with this stream: "
 286             "stream-name=\"{}\"",
 287             lttng_live_stream->name->str);
 288         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 289         goto end;
 290     }
 291
 292     if (lttng_live_stream->trace->session->new_streams_needed) {
 293         BT_CPPLOGD_SPEC(
 294             lttng_live_msg_iter->logger,
 295             "Need to get an update of all streams before proceeding further with this stream: "
 296             "stream-name=\"{}\"",
 297             lttng_live_stream->name->str);
 298         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 299         goto end;
 300     }
 301
 302     if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_NO_DATA &&
 303         lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA) {
 304         goto end;
 305     }
 306     ret = lttng_live_get_next_index(lttng_live_msg_iter, lttng_live_stream, &index);
 307     if (ret != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 308         goto end;
 309     }
 310
 311     BT_ASSERT_DBG(lttng_live_stream->state != LTTNG_LIVE_STREAM_EOF);
 312
 313     if (lttng_live_stream->state == LTTNG_LIVE_STREAM_QUIESCENT) {
 314         uint64_t last_inact_ts = lttng_live_stream->last_inactivity_ts.value,
 315                  curr_inact_ts = lttng_live_stream->current_inactivity_ts;
 316
 317         if (orig_state == LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA && last_inact_ts == curr_inact_ts) {
 318             /*
 319              * Because the stream is in the QUIESCENT_NO_DATA
 320              * state, we can assert that the last_inactivity_ts was
 321              * set and can be safely used in the `if` above.
 322              */
 323             BT_ASSERT(lttng_live_stream->last_inactivity_ts.is_set);
 324
 325             ret = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 326             LTTNG_LIVE_LOGD_STREAM_ITER(lttng_live_stream);
 327         } else {
 328             ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 329         }
 330         goto end;
 331     }
 332
 333     lttng_live_stream->base_offset = index.offset;
 334     lttng_live_stream->offset = index.offset;
 335     lttng_live_stream->len = index.packet_size / CHAR_BIT;
 336
 337     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 338                     "Setting live stream reading info: stream-name=\"{}\", "
 339                     "viewer-stream-id={}, stream-base-offset={}, stream-offset={}, stream-len={}",
 340                     lttng_live_stream->name->str, lttng_live_stream->viewer_stream_id,
 341                     lttng_live_stream->base_offset, lttng_live_stream->offset,
 342                     lttng_live_stream->len);
 343
 344 end:
 345     if (ret == LTTNG_LIVE_ITERATOR_STATUS_OK) {
 346         ret = lttng_live_iterator_next_check_stream_state(lttng_live_stream);
 347     }
 348     return ret;
 349 }
 350
 351 /*
 352  * Creation of the message requires the ctf trace class to be created
 353  * beforehand, but the live protocol gives us all streams (including metadata)
 354  * at once. So we split it in three steps: getting streams, getting metadata
 355  * (which creates the ctf trace class), and then creating the per-stream
 356  * messages.
 357  */
 358 static enum lttng_live_iterator_status
 359 lttng_live_get_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
 360                        struct lttng_live_session *session)
 361 {
 362     enum lttng_live_iterator_status status;
 363     uint64_t trace_idx;
 364
 365     if (!session->attached) {
 366         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger, "Attach to session: session-id={}",
 367                         session->id);
 368         enum lttng_live_viewer_status attach_status =
 369             lttng_live_session_attach(session, lttng_live_msg_iter->self_msg_iter);
 370         if (attach_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 371             if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
 372                 /*
 373                  * Clear any causes appended in
 374                  * `lttng_live_attach_session()` as we want to
 375                  * return gracefully since the graph was
 376                  * cancelled.
 377                  */
 378                 bt_current_thread_clear_error();
 379                 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 380             } else {
 381                 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 382                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 383                                              "Error attaching to LTTng live session");
 384             }
 385             goto end;
 386         }
 387     }
 388
 389     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 390                     "Updating all data streams: "
 391                     "session-id={}, session-name=\"{}\"",
 392                     session->id, session->session_name->str);
 393
 394     status = lttng_live_session_get_new_streams(session, lttng_live_msg_iter->self_msg_iter);
 395     switch (status) {
 396     case LTTNG_LIVE_ITERATOR_STATUS_OK:
 397         break;
 398     case LTTNG_LIVE_ITERATOR_STATUS_END:
 399         /*
 400                  * We received a `_END` from the `_get_new_streams()` function,
 401                  * which means no more data will ever be received from the data
 402                  * streams of this session. But it's possible that the metadata
 403                  * is incomplete.
 404                  * The live protocol guarantees that we receive all the
 405                  * metadata needed before we receive data streams needing it.
 406                  * But it's possible to receive metadata NOT needed by
 407                  * data streams after the session was closed. For example, this
 408                  * could happen if a new event is registered and the session is
 409                  * stopped before any tracepoint for that event is actually
 410                  * fired.
 411                  */
 412         BT_CPPLOGD_SPEC(
 413             lttng_live_msg_iter->logger,
 414             "Updating streams returned _END status. Override status to _OK in order fetch any remaining metadata:"
 415             "session-id={}, session-name=\"{}\"",
 416             session->id, session->session_name->str);
 417         status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 418         break;
 419     default:
 420         goto end;
 421     }
 422
 423     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 424                     "Updating metadata stream for session: "
 425                     "session-id={}, session-name=\"{}\"",
 426                     session->id, session->session_name->str);
 427
 428     trace_idx = 0;
 429     while (trace_idx < session->traces->len) {
 430         struct lttng_live_trace *trace =
 431             (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
 432
 433         status = lttng_live_metadata_update(trace);
 434         switch (status) {
 435         case LTTNG_LIVE_ITERATOR_STATUS_END:
 436         case LTTNG_LIVE_ITERATOR_STATUS_OK:
 437             trace_idx++;
 438             break;
 439         case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
 440         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
 441             goto end;
 442         default:
 443             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 444                                          "Error updating trace metadata: "
 445                                          "stream-iter-status={}, trace-id={}",
 446                                          status, trace->id);
 447             goto end;
 448         }
 449     }
 450
 451     /*
 452      * Now that we have the metadata we can initialize the downstream
 453      * iterator.
 454      */
 455     status = lttng_live_lazy_msg_init(session, lttng_live_msg_iter->self_msg_iter);
 456
 457 end:
 458     return status;
 459 }
 460
 461 static void
 462 lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
 463 {
 464     uint64_t session_idx, trace_idx;
 465
 466     for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
 467         struct lttng_live_session *session =
 468             (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 469         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 470                         "Force marking session as needing new streams: "
 471                         "session-id={}",
 472                         session->id);
 473         session->new_streams_needed = true;
 474         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 475             struct lttng_live_trace *trace =
 476                 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
 477             BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 478                             "Force marking trace metadata state as needing an update: "
 479                             "session-id={}, trace-id={}",
 480                             session->id, trace->id);
 481
 482             BT_ASSERT(trace->metadata_stream_state != LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED);
 483
 484             trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
 485         }
 486     }
 487 }
 488
 489 static enum lttng_live_iterator_status
 490 lttng_live_iterator_handle_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
 491 {
 492     enum lttng_live_iterator_status status;
 493     enum lttng_live_viewer_status viewer_status;
 494     uint64_t session_idx = 0, nr_sessions_opened = 0;
 495     struct lttng_live_session *session;
 496     enum session_not_found_action sess_not_found_act =
 497         lttng_live_msg_iter->lttng_live_comp->params.sess_not_found_act;
 498
 499     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 500                     "Update data and metadata of all sessions: "
 501                     "live-msg-iter-addr={}",
 502                     fmt::ptr(lttng_live_msg_iter));
 503     /*
 504      * In a remotely distant future, we could add a "new
 505      * session" flag to the protocol, which would tell us that we
 506      * need to query for new sessions even though we have sessions
 507      * currently ongoing.
 508      */
 509     if (lttng_live_msg_iter->sessions->len == 0) {
 510         if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
 511             BT_CPPLOGD_SPEC(
 512                 lttng_live_msg_iter->logger,
 513                 "No session found. Exiting in accordance with the `session-not-found-action` parameter");
 514             status = LTTNG_LIVE_ITERATOR_STATUS_END;
 515             goto end;
 516         } else {
 517             BT_CPPLOGD_SPEC(
 518                 lttng_live_msg_iter->logger,
 519                 "No session found. Try creating a new one in accordance with the `session-not-found-action` parameter");
 520             /*
 521              * Retry to create a viewer session for the requested
 522              * session name.
 523              */
 524             viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
 525             if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 526                 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
 527                     status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 528                     BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 529                                                  "Error creating LTTng live viewer session");
 530                 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
 531                     status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
 532                 } else {
 533                     bt_common_abort();
 534                 }
 535                 goto end;
 536             }
 537         }
 538     }
 539
 540     for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
 541         session =
 542             (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 543         status = lttng_live_get_session(lttng_live_msg_iter, session);
 544         switch (status) {
 545         case LTTNG_LIVE_ITERATOR_STATUS_OK:
 546         case LTTNG_LIVE_ITERATOR_STATUS_END:
 547             /*
 548              * A session returned `_END`. Other sessions may still
 549              * be active so we override the status and continue
 550              * looping if needed.
 551              */
 552             break;
 553         default:
 554             goto end;
 555         }
 556         if (!session->closed) {
 557             nr_sessions_opened++;
 558         }
 559     }
 560
 561     if (sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE && nr_sessions_opened == 0) {
 562         status = LTTNG_LIVE_ITERATOR_STATUS_END;
 563     } else {
 564         status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 565     }
 566
 567 end:
 568     return status;
 569 }
 570
 571 static enum lttng_live_iterator_status
 572 emit_inactivity_message(struct lttng_live_msg_iter *lttng_live_msg_iter,
 573                         struct lttng_live_stream_iterator *stream_iter, const bt_message **message,
 574                         uint64_t timestamp)
 575 {
 576     enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 577     bt_message *msg = NULL;
 578
 579     BT_ASSERT(stream_iter->trace->clock_class);
 580
 581     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 582                     "Emitting inactivity message for stream: ctf-stream-id={}, "
 583                     "viewer-stream-id={}, timestamp={}",
 584                     stream_iter->ctf_stream_class_id.value, stream_iter->viewer_stream_id,
 585                     timestamp);
 586
 587     msg = bt_message_message_iterator_inactivity_create(lttng_live_msg_iter->self_msg_iter,
 588                                                         stream_iter->trace->clock_class, timestamp);
 589     if (!msg) {
 590         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 591                                      "Error emitting message iterator inactivity message");
 592         goto error;
 593     }
 594
 595     *message = msg;
 596 end:
 597     return ret;
 598
 599 error:
 600     ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 601     bt_message_put_ref(msg);
 602     goto end;
 603 }
 604
 605 static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_quiescent_stream(
 606     struct lttng_live_msg_iter *lttng_live_msg_iter,
 607     struct lttng_live_stream_iterator *lttng_live_stream, const bt_message **message)
 608 {
 609     enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 610
 611     if (lttng_live_stream->state != LTTNG_LIVE_STREAM_QUIESCENT) {
 612         return LTTNG_LIVE_ITERATOR_STATUS_OK;
 613     }
 614
 615     /*
 616      * Check if we already sent an inactivty message downstream for this
 617      * `current_inactivity_ts` value.
 618      */
 619     if (lttng_live_stream->last_inactivity_ts.is_set &&
 620         lttng_live_stream->current_inactivity_ts == lttng_live_stream->last_inactivity_ts.value) {
 621         lttng_live_stream_iterator_set_state(lttng_live_stream,
 622                                              LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA);
 623
 624         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 625         goto end;
 626     }
 627
 628     ret = emit_inactivity_message(lttng_live_msg_iter, lttng_live_stream, message,
 629                                   lttng_live_stream->current_inactivity_ts);
 630
 631     lttng_live_stream->last_inactivity_ts.value = lttng_live_stream->current_inactivity_ts;
 632     lttng_live_stream->last_inactivity_ts.is_set = true;
 633 end:
 634     return ret;
 635 }
 636
 637 static int live_get_msg_ts_ns(struct lttng_live_msg_iter *lttng_live_msg_iter,
 638                               const bt_message *msg, int64_t last_msg_ts_ns, int64_t *ts_ns)
 639 {
 640     const bt_clock_snapshot *clock_snapshot = NULL;
 641     int ret = 0;
 642
 643     BT_ASSERT_DBG(msg);
 644     BT_ASSERT_DBG(ts_ns);
 645
 646     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 647                     "Getting message's timestamp: iter-data-addr={}, msg-addr={}, "
 648                     "last-msg-ts={}",
 649                     fmt::ptr(lttng_live_msg_iter), fmt::ptr(msg), last_msg_ts_ns);
 650
 651     switch (bt_message_get_type(msg)) {
 652     case BT_MESSAGE_TYPE_EVENT:
 653         clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(msg);
 654         break;
 655     case BT_MESSAGE_TYPE_PACKET_BEGINNING:
 656         clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(msg);
 657         break;
 658     case BT_MESSAGE_TYPE_PACKET_END:
 659         clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(msg);
 660         break;
 661     case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
 662         clock_snapshot =
 663             bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(msg);
 664         break;
 665     case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
 666         clock_snapshot =
 667             bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(msg);
 668         break;
 669     case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
 670         clock_snapshot = bt_message_message_iterator_inactivity_borrow_clock_snapshot_const(msg);
 671         break;
 672     default:
 673         /* All the other messages have a higher priority */
 674         BT_CPPLOGD_STR_SPEC(lttng_live_msg_iter->logger,
 675                             "Message has no timestamp: using the last message timestamp.");
 676         *ts_ns = last_msg_ts_ns;
 677         goto end;
 678     }
 679
 680     ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot, ts_ns);
 681     if (ret) {
 682         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 683                                      "Cannot get nanoseconds from Epoch of clock snapshot: "
 684                                      "clock-snapshot-addr={}",
 685                                      fmt::ptr(clock_snapshot));
 686         goto error;
 687     }
 688
 689     goto end;
 690
 691 error:
 692     ret = -1;
 693
 694 end:
 695     if (ret == 0) {
 696         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 697                         "Found message's timestamp: iter-data-addr={}, msg-addr={}, "
 698                         "last-msg-ts={}, ts={}",
 699                         fmt::ptr(lttng_live_msg_iter), fmt::ptr(msg), last_msg_ts_ns, *ts_ns);
 700     }
 701
 702     return ret;
 703 }
 704
 705 static enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_stream(
 706     struct lttng_live_msg_iter *lttng_live_msg_iter,
 707     struct lttng_live_stream_iterator *lttng_live_stream, const bt_message **message)
 708 {
 709     enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 710     enum ctf_msg_iter_status status;
 711     uint64_t session_idx, trace_idx;
 712
 713     for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
 714         struct lttng_live_session *session =
 715             (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
 716
 717         if (session->new_streams_needed) {
 718             BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 719                             "Need an update for streams: "
 720                             "session-id={}",
 721                             session->id);
 722             ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 723             goto end;
 724         }
 725         for (trace_idx = 0; trace_idx < session->traces->len; trace_idx++) {
 726             struct lttng_live_trace *trace =
 727                 (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
 728             if (trace->metadata_stream_state == LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
 729                 BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 730                                 "Need an update for metadata stream: "
 731                                 "session-id={}, trace-id={}",
 732                                 session->id, trace->id);
 733                 ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 734                 goto end;
 735             }
 736         }
 737     }
 738
 739     if (lttng_live_stream->state != LTTNG_LIVE_STREAM_ACTIVE_DATA) {
 740         ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 741         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 742                                      "Invalid state of live stream iterator"
 743                                      "stream-iter-status={}",
 744                                      lttng_live_stream->state);
 745         goto end;
 746     }
 747
 748     status = ctf_msg_iter_get_next_message(lttng_live_stream->msg_iter, message);
 749     switch (status) {
 750     case CTF_MSG_ITER_STATUS_EOF:
 751         ret = LTTNG_LIVE_ITERATOR_STATUS_END;
 752         break;
 753     case CTF_MSG_ITER_STATUS_OK:
 754         ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 755         break;
 756     case CTF_MSG_ITER_STATUS_AGAIN:
 757         /*
 758          * Continue immediately (end of packet). The next
 759          * get_index may return AGAIN to delay the following
 760          * attempt.
 761          */
 762         ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
 763         break;
 764     case CTF_MSG_ITER_STATUS_ERROR:
 765     default:
 766         ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 767         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 768                                      "CTF message iterator failed to get next message: "
 769                                      "msg-iter={}, msg-iter-status={}",
 770                                      fmt::ptr(lttng_live_stream->msg_iter), status);
 771         break;
 772     }
 773
 774 end:
 775     return ret;
 776 }
 777
 778 static enum lttng_live_iterator_status
 779 lttng_live_iterator_close_stream(struct lttng_live_msg_iter *lttng_live_msg_iter,
 780                                  struct lttng_live_stream_iterator *stream_iter,
 781                                  const bt_message **curr_msg)
 782 {
 783     enum lttng_live_iterator_status live_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 784
 785     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 786                     "Closing live stream iterator: stream-name=\"{}\", "
 787                     "viewer-stream-id={}",
 788                     stream_iter->name->str, stream_iter->viewer_stream_id);
 789
 790     /*
 791      * The viewer has hung up on us so we are closing the stream. The
 792      * `ctf_msg_iter` should simply realize that it needs to close the
 793      * stream properly by emitting the necessary stream end message.
 794      */
 795     enum ctf_msg_iter_status status =
 796         ctf_msg_iter_get_next_message(stream_iter->msg_iter, curr_msg);
 797
 798     if (status == CTF_MSG_ITER_STATUS_ERROR) {
 799         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
 800                                      "Error getting the next message from CTF message iterator");
 801         live_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 802         goto end;
 803     } else if (status == CTF_MSG_ITER_STATUS_EOF) {
 804         BT_CPPLOGI_SPEC(lttng_live_msg_iter->logger,
 805                         "Reached the end of the live stream iterator.");
 806         live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
 807         goto end;
 808     }
 809
 810     BT_ASSERT(status == CTF_MSG_ITER_STATUS_OK);
 811
 812 end:
 813     return live_status;
 814 }
 815
 816 /*
 817  * helper function:
 818  *            handle_no_data_streams()
 819  *              retry:
 820  *                - for each ACTIVE_NO_DATA stream:
 821  *                  - query relayd for stream data, or quiescence info.
 822  *                    - if need metadata, get metadata, goto retry.
 823  *                    - if new stream, get new stream as ACTIVE_NO_DATA, goto retry
 824  *                  - if quiescent, move to QUIESCENT streams
 825  *                  - if fetched data, move to ACTIVE_DATA streams
 826  *                (at this point each stream either has data, or is quiescent)
 827  *
 828  *
 829  * iterator_next:
 830  *            handle_new_streams_and_metadata()
 831  *                  - query relayd for known streams, add them as ACTIVE_NO_DATA
 832  *                  - query relayd for metadata
 833  *
 834  *            call handle_active_no_data_streams()
 835  *
 836  *            handle_quiescent_streams()
 837  *                - if at least one stream is ACTIVE_DATA:
 838  *                  - peek stream event with lowest timestamp -> next_ts
 839  *                  - for each quiescent stream
 840  *                    - if next_ts >= quiescent end
 841  *                      - set state to ACTIVE_NO_DATA
 842  *                - else
 843  *                  - for each quiescent stream
 844  *                      - set state to ACTIVE_NO_DATA
 845  *
 846  *            call handle_active_no_data_streams()
 847  *
 848  *            handle_active_data_streams()
 849  *                - if at least one stream is ACTIVE_DATA:
 850  *                    - get stream event with lowest timestamp from heap
 851  *                    - make that stream event the current message.
 852  *                    - move this stream heap position to its next event
 853  *                      - if we need to fetch data from relayd, move
 854  *                        stream to ACTIVE_NO_DATA.
 855  *                    - return OK
 856  *                - return AGAIN
 857  *
 858  * end criterion: ctrl-c on client. If relayd exits or the session
 859  * closes on the relay daemon side, we keep on waiting for streams.
 860  * Eventually handle --end timestamp (also an end criterion).
 861  *
 862  * When disconnected from relayd: try to re-connect endlessly.
 863  */
 864 static enum lttng_live_iterator_status
 865 lttng_live_iterator_next_msg_on_stream(struct lttng_live_msg_iter *lttng_live_msg_iter,
 866                                        struct lttng_live_stream_iterator *stream_iter,
 867                                        const bt_message **curr_msg)
 868 {
 869     enum lttng_live_iterator_status live_status;
 870
 871     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 872                     "Advancing live stream iterator until next message if possible: "
 873                     "stream-name=\"{}\", viewer-stream-id={}",
 874                     stream_iter->name->str, stream_iter->viewer_stream_id);
 875
 876     if (stream_iter->has_stream_hung_up) {
 877         /*
 878          * The stream has hung up and the stream was properly closed
 879          * during the last call to the current function. Return _END
 880          * status now so that this stream iterator is removed for the
 881          * stream iterator list.
 882          */
 883         live_status = LTTNG_LIVE_ITERATOR_STATUS_END;
 884         goto end;
 885     }
 886
 887 retry:
 888     LTTNG_LIVE_LOGD_STREAM_ITER(stream_iter);
 889
 890     /*
 891      * Make sure we have the most recent metadata and possibly some new
 892      * streams.
 893      */
 894     live_status = lttng_live_iterator_handle_new_streams_and_metadata(lttng_live_msg_iter);
 895     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 896         goto end;
 897     }
 898
 899     live_status =
 900         lttng_live_iterator_next_handle_one_no_data_stream(lttng_live_msg_iter, stream_iter);
 901     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 902         if (live_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
 903             /*
 904              * We overwrite `live_status` since `curr_msg` is
 905              * likely set to a valid message in this function.
 906              */
 907             live_status =
 908                 lttng_live_iterator_close_stream(lttng_live_msg_iter, stream_iter, curr_msg);
 909         }
 910         goto end;
 911     }
 912
 913     live_status = lttng_live_iterator_next_handle_one_quiescent_stream(lttng_live_msg_iter,
 914                                                                        stream_iter, curr_msg);
 915     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 916         BT_ASSERT(!*curr_msg);
 917         goto end;
 918     }
 919     if (*curr_msg) {
 920         goto end;
 921     }
 922     live_status = lttng_live_iterator_next_handle_one_active_data_stream(lttng_live_msg_iter,
 923                                                                          stream_iter, curr_msg);
 924     if (live_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
 925         BT_ASSERT(!*curr_msg);
 926     }
 927
 928 end:
 929     if (live_status == LTTNG_LIVE_ITERATOR_STATUS_CONTINUE) {
 930         BT_CPPLOGD_SPEC(
 931             lttng_live_msg_iter->logger,
 932             "Ask the relay daemon for an updated view of the data and metadata streams");
 933         goto retry;
 934     }
 935
 936     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
 937                     "Returning from advancing live stream iterator: status={}, "
 938                     "stream-name=\"{}\", viewer-stream-id={}",
 939                     live_status, stream_iter->name->str, stream_iter->viewer_stream_id);
 940
 941     return live_status;
 942 }
 943
 944 static bool is_discarded_packet_or_event_message(const bt_message *msg)
 945 {
 946     const enum bt_message_type msg_type = bt_message_get_type(msg);
 947
 948     return msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS ||
 949            msg_type == BT_MESSAGE_TYPE_DISCARDED_PACKETS;
 950 }
 951
 952 static enum lttng_live_iterator_status
 953 adjust_discarded_packets_message(bt_self_message_iterator *iter, const bt_stream *stream,
 954                                  const bt_message *msg_in, bt_message **msg_out,
 955                                  uint64_t new_begin_ts)
 956 {
 957     enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 958     enum bt_property_availability availability;
 959     const bt_clock_snapshot *clock_snapshot;
 960     uint64_t end_ts;
 961     uint64_t count;
 962
 963     clock_snapshot = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(msg_in);
 964     end_ts = bt_clock_snapshot_get_value(clock_snapshot);
 965
 966     availability = bt_message_discarded_packets_get_count(msg_in, &count);
 967     BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
 968
 969     *msg_out = bt_message_discarded_packets_create_with_default_clock_snapshots(
 970         iter, stream, new_begin_ts, end_ts);
 971     if (!*msg_out) {
 972         status = LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
 973         goto end;
 974     }
 975
 976     bt_message_discarded_packets_set_count(*msg_out, count);
 977 end:
 978     return status;
 979 }
 980
 981 static enum lttng_live_iterator_status
 982 adjust_discarded_events_message(bt_self_message_iterator *iter, const bt_stream *stream,
 983                                 const bt_message *msg_in, bt_message **msg_out,
 984                                 uint64_t new_begin_ts)
 985 {
 986     enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
 987     enum bt_property_availability availability;
 988     const bt_clock_snapshot *clock_snapshot;
 989     uint64_t end_ts;
 990     uint64_t count;
 991
 992     clock_snapshot = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(msg_in);
 993     end_ts = bt_clock_snapshot_get_value(clock_snapshot);
 994
 995     availability = bt_message_discarded_events_get_count(msg_in, &count);
 996     BT_ASSERT_DBG(availability == BT_PROPERTY_AVAILABILITY_AVAILABLE);
 997
 998     *msg_out = bt_message_discarded_events_create_with_default_clock_snapshots(
 999         iter, stream, new_begin_ts, end_ts);
1000     if (!*msg_out) {
1001         status = LTTNG_LIVE_ITERATOR_STATUS_NOMEM;
1002         goto end;
1003     }
1004
1005     bt_message_discarded_events_set_count(*msg_out, count);
1006 end:
1007     return status;
1008 }
1009
1010 static enum lttng_live_iterator_status
1011 handle_late_message(struct lttng_live_msg_iter *lttng_live_msg_iter,
1012                     struct lttng_live_stream_iterator *stream_iter, int64_t late_msg_ts_ns,
1013                     const bt_message *late_msg)
1014 {
1015     const bt_clock_class *clock_class;
1016     const bt_stream_class *stream_class;
1017     enum bt_clock_class_cycles_to_ns_from_origin_status ts_ns_status;
1018     int64_t last_inactivity_ts_ns;
1019     enum lttng_live_iterator_status stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1020     enum lttng_live_iterator_status adjust_status;
1021     bt_message *adjusted_message;
1022
1023     /*
1024      * The timestamp of the current message is before the last message sent
1025      * by this component. We CANNOT send it as is.
1026      *
1027      * The only expected scenario in which that could happen is the
1028      * following, everything else is a bug in this component, relay daemon,
1029      * or CTF parser.
1030      *
1031      * Expected scenario: The CTF message iterator emitted discarded
1032      * packets and discarded events with synthesized beginning and end
1033      * timestamps from the bounds of the last known packet and the newly
1034      * decoded packet header. The CTF message iterator is not aware of
1035      * stream inactivity beacons. Hence, we have to adjust the beginning
1036      * timestamp of those types of messages if a stream signalled its
1037      * inactivity up until _after_ the last known packet's beginning
1038      * timestamp.
1039      *
1040      * Otherwise, the monotonicity guarantee of message timestamps would
1041      * not be preserved.
1042      *
1043      * In short, the only scenario in which it's okay and fixable to
1044      * received a late message is when:
1045      *  1. the late message is a discarded packets or discarded events
1046      *  message,
1047      *  2. this stream produced an inactivity message downstream, and
1048      *  3. the timestamp of the late message is within the inactivity
1049      *  timespan we sent downstream through the inactivity message.
1050      */
1051
1052     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1053                     "Handling late message on live stream iterator: "
1054                     "stream-name=\"{}\", viewer-stream-id={}",
1055                     stream_iter->name->str, stream_iter->viewer_stream_id);
1056
1057     if (!stream_iter->last_inactivity_ts.is_set) {
1058         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1059                                      "Invalid live stream state: "
1060                                      "have a late message when no inactivity message "
1061                                      "was ever sent for that stream.");
1062         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1063         goto end;
1064     }
1065
1066     if (!is_discarded_packet_or_event_message(late_msg)) {
1067         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1068                                      "Invalid live stream state: "
1069                                      "have a late message that is not a packet discarded or "
1070                                      "event discarded message: late-msg-type={}",
1071                                      static_cast<bt2::MessageType>(bt_message_get_type(late_msg)));
1072         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1073         goto end;
1074     }
1075
1076     stream_class = bt_stream_borrow_class_const(stream_iter->stream);
1077     clock_class = bt_stream_class_borrow_default_clock_class_const(stream_class);
1078
1079     ts_ns_status = bt_clock_class_cycles_to_ns_from_origin(
1080         clock_class, stream_iter->last_inactivity_ts.value, &last_inactivity_ts_ns);
1081     if (ts_ns_status != BT_CLOCK_CLASS_CYCLES_TO_NS_FROM_ORIGIN_STATUS_OK) {
1082         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1083                                      "Error converting last "
1084                                      "inactivity message timestamp to nanoseconds");
1085         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1086         goto end;
1087     }
1088
1089     if (last_inactivity_ts_ns <= late_msg_ts_ns) {
1090         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1091                                      "Invalid live stream state: "
1092                                      "have a late message that is none included in a stream "
1093                                      "inactivity timespan: last-inactivity-ts-ns={}, "
1094                                      "late-msg-ts-ns={}",
1095                                      last_inactivity_ts_ns, late_msg_ts_ns);
1096         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1097         goto end;
1098     }
1099
1100     /*
1101      * We now know that it's okay for this message to be late, we can now
1102      * adjust its timestamp to ensure monotonicity.
1103      */
1104     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1105                     "Adjusting the timestamp of late message: late-msg-type={}, "
1106                     "msg-new-ts-ns={}",
1107                     static_cast<bt2::MessageType>(bt_message_get_type(late_msg)),
1108                     stream_iter->last_inactivity_ts.value);
1109     switch (bt_message_get_type(late_msg)) {
1110     case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
1111         adjust_status = adjust_discarded_events_message(
1112             lttng_live_msg_iter->self_msg_iter, stream_iter->stream, late_msg, &adjusted_message,
1113             stream_iter->last_inactivity_ts.value);
1114         break;
1115     case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
1116         adjust_status = adjust_discarded_packets_message(
1117             lttng_live_msg_iter->self_msg_iter, stream_iter->stream, late_msg, &adjusted_message,
1118             stream_iter->last_inactivity_ts.value);
1119         break;
1120     default:
1121         bt_common_abort();
1122     }
1123
1124     if (adjust_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1125         stream_iter_status = adjust_status;
1126         goto end;
1127     }
1128
1129     BT_ASSERT_DBG(adjusted_message);
1130     stream_iter->current_msg = adjusted_message;
1131     stream_iter->current_msg_ts_ns = last_inactivity_ts_ns;
1132     bt_message_put_ref(late_msg);
1133
1134 end:
1135     return stream_iter_status;
1136 }
1137
1138 static enum lttng_live_iterator_status
1139 next_stream_iterator_for_trace(struct lttng_live_msg_iter *lttng_live_msg_iter,
1140                                struct lttng_live_trace *live_trace,
1141                                struct lttng_live_stream_iterator **youngest_trace_stream_iter)
1142 {
1143     struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
1144     enum lttng_live_iterator_status stream_iter_status;
1145     int64_t youngest_candidate_msg_ts = INT64_MAX;
1146     uint64_t stream_iter_idx;
1147
1148     BT_ASSERT_DBG(live_trace);
1149     BT_ASSERT_DBG(live_trace->stream_iterators);
1150
1151     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1152                     "Finding the next stream iterator for trace: "
1153                     "trace-id={}",
1154                     live_trace->id);
1155     /*
1156      * Update the current message of every stream iterators of this trace.
1157      * The current msg of every stream must have a timestamp equal or
1158      * larger than the last message returned by this iterator. We must
1159      * ensure monotonicity.
1160      */
1161     stream_iter_idx = 0;
1162     while (stream_iter_idx < live_trace->stream_iterators->len) {
1163         bool stream_iter_is_ended = false;
1164         struct lttng_live_stream_iterator *stream_iter =
1165             (lttng_live_stream_iterator *) g_ptr_array_index(live_trace->stream_iterators,
1166                                                              stream_iter_idx);
1167
1168         /*
1169          * If there is no current message for this stream, go fetch
1170          * one.
1171          */
1172         while (!stream_iter->current_msg) {
1173             const bt_message *msg = NULL;
1174             int64_t curr_msg_ts_ns = INT64_MAX;
1175
1176             stream_iter_status =
1177                 lttng_live_iterator_next_msg_on_stream(lttng_live_msg_iter, stream_iter, &msg);
1178
1179             if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1180                 stream_iter_is_ended = true;
1181                 break;
1182             }
1183
1184             if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1185                 goto end;
1186             }
1187
1188             BT_ASSERT_DBG(msg);
1189
1190             BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1191                             "Live stream iterator returned message: msg-type={}, "
1192                             "stream-name=\"{}\", viewer-stream-id={}",
1193                             static_cast<bt2::MessageType>(bt_message_get_type(msg)),
1194                             stream_iter->name->str, stream_iter->viewer_stream_id);
1195
1196             /*
1197              * Get the timestamp in nanoseconds from origin of this
1198              * message.
1199              */
1200             live_get_msg_ts_ns(lttng_live_msg_iter, msg, lttng_live_msg_iter->last_msg_ts_ns,
1201                                &curr_msg_ts_ns);
1202
1203             /*
1204              * Check if the message of the current live stream
1205              * iterator occurred at the exact same time or after the
1206              * last message returned by this component's message
1207              * iterator. If not, we need to handle it with care.
1208              */
1209             if (curr_msg_ts_ns >= lttng_live_msg_iter->last_msg_ts_ns) {
1210                 stream_iter->current_msg = msg;
1211                 stream_iter->current_msg_ts_ns = curr_msg_ts_ns;
1212             } else {
1213                 /*
1214                  * We received a message from the past. This
1215                  * may be fixable but it can also be an error.
1216                  */
1217                 stream_iter_status =
1218                     handle_late_message(lttng_live_msg_iter, stream_iter, curr_msg_ts_ns, msg);
1219                 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1220                     BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1221                                                  "Late message could not be handled correctly: "
1222                                                  "lttng-live-msg-iter-addr={}, "
1223                                                  "stream-name=\"{}\", "
1224                                                  "curr-msg-ts={}, last-msg-ts={}",
1225                                                  fmt::ptr(lttng_live_msg_iter),
1226                                                  stream_iter->name->str, curr_msg_ts_ns,
1227                                                  lttng_live_msg_iter->last_msg_ts_ns);
1228                     stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1229                     goto end;
1230                 }
1231             }
1232         }
1233
1234         BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter);
1235
1236         if (!stream_iter_is_ended) {
1237             if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1238                 stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1239                 /*
1240                  * Update the current best candidate message
1241                  * for the stream iterator of this live trace
1242                  * to be forwarded downstream.
1243                  */
1244                 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1245                 youngest_candidate_stream_iter = stream_iter;
1246             } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1247                 /*
1248                  * Order the messages in an arbitrary but
1249                  * deterministic way.
1250                  */
1251                 BT_ASSERT_DBG(stream_iter != youngest_candidate_stream_iter);
1252                 int ret = common_muxing_compare_messages(
1253                     stream_iter->current_msg, youngest_candidate_stream_iter->current_msg);
1254                 if (ret < 0) {
1255                     /*
1256                      * The `youngest_candidate_stream_iter->current_msg`
1257                      * should go first. Update the next
1258                      * iterator and the current timestamp.
1259                      */
1260                     youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1261                     youngest_candidate_stream_iter = stream_iter;
1262                 } else if (ret == 0) {
1263                     /*
1264                      * Unable to pick which one should go
1265                      * first.
1266                      */
1267                     BT_CPPLOGW_SPEC(
1268                         lttng_live_msg_iter->logger,
1269                         "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1270                         "stream-iter-addr={}"
1271                         "stream-iter-addr={}",
1272                         fmt::ptr(stream_iter), fmt::ptr(youngest_candidate_stream_iter));
1273                 }
1274             }
1275
1276             stream_iter_idx++;
1277         } else {
1278             /*
1279              * The live stream iterator has ended. That
1280              * iterator is removed from the array, but
1281              * there is no need to increment
1282              * stream_iter_idx as
1283              * g_ptr_array_remove_index_fast replaces the
1284              * removed element with the array's last
1285              * element.
1286              */
1287             g_ptr_array_remove_index_fast(live_trace->stream_iterators, stream_iter_idx);
1288         }
1289     }
1290
1291     if (youngest_candidate_stream_iter) {
1292         *youngest_trace_stream_iter = youngest_candidate_stream_iter;
1293         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1294     } else {
1295         /*
1296          * The only case where we don't have a candidate for this trace
1297          * is if we reached the end of all the iterators.
1298          */
1299         BT_ASSERT(live_trace->stream_iterators->len == 0);
1300         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1301     }
1302
1303 end:
1304     return stream_iter_status;
1305 }
1306
1307 static enum lttng_live_iterator_status
1308 next_stream_iterator_for_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
1309                                  struct lttng_live_session *session,
1310                                  struct lttng_live_stream_iterator **youngest_session_stream_iter)
1311 {
1312     enum lttng_live_iterator_status stream_iter_status;
1313     uint64_t trace_idx = 0;
1314     int64_t youngest_candidate_msg_ts = INT64_MAX;
1315     struct lttng_live_stream_iterator *youngest_candidate_stream_iter = NULL;
1316
1317     BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1318                     "Finding the next stream iterator for session: "
1319                     "session-id={}",
1320                     session->id);
1321     /*
1322      * Make sure we are attached to the session and look for new streams
1323      * and metadata.
1324      */
1325     stream_iter_status = lttng_live_get_session(lttng_live_msg_iter, session);
1326     if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
1327         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_CONTINUE &&
1328         stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_END) {
1329         goto end;
1330     }
1331
1332     BT_ASSERT_DBG(session->traces);
1333
1334     while (trace_idx < session->traces->len) {
1335         bool trace_is_ended = false;
1336         struct lttng_live_stream_iterator *stream_iter;
1337         struct lttng_live_trace *trace =
1338             (lttng_live_trace *) g_ptr_array_index(session->traces, trace_idx);
1339
1340         stream_iter_status =
1341             next_stream_iterator_for_trace(lttng_live_msg_iter, trace, &stream_iter);
1342         if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1343             /*
1344              * All the live stream iterators for this trace are
1345              * ENDed. Remove the trace from this session.
1346              */
1347             trace_is_ended = true;
1348         } else if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1349             goto end;
1350         }
1351
1352         if (!trace_is_ended) {
1353             BT_ASSERT_DBG(stream_iter);
1354
1355             if (G_UNLIKELY(youngest_candidate_stream_iter == NULL) ||
1356                 stream_iter->current_msg_ts_ns < youngest_candidate_msg_ts) {
1357                 youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1358                 youngest_candidate_stream_iter = stream_iter;
1359             } else if (stream_iter->current_msg_ts_ns == youngest_candidate_msg_ts) {
1360                 /*
1361                  * Order the messages in an arbitrary but
1362                  * deterministic way.
1363                  */
1364                 int ret = common_muxing_compare_messages(
1365                     stream_iter->current_msg, youngest_candidate_stream_iter->current_msg);
1366                 if (ret < 0) {
1367                     /*
1368                      * The `youngest_candidate_stream_iter->current_msg`
1369                      * should go first. Update the next iterator
1370                      * and the current timestamp.
1371                      */
1372                     youngest_candidate_msg_ts = stream_iter->current_msg_ts_ns;
1373                     youngest_candidate_stream_iter = stream_iter;
1374                 } else if (ret == 0) {
1375                     /* Unable to pick which one should go first. */
1376                     BT_CPPLOGW_SPEC(
1377                         lttng_live_msg_iter->logger,
1378                         "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1379                         "stream-iter-addr={}"
1380                         "youngest-candidate-stream-iter-addr={}",
1381                         fmt::ptr(stream_iter), fmt::ptr(youngest_candidate_stream_iter));
1382                 }
1383             }
1384             trace_idx++;
1385         } else {
1386             /*
1387              * trace_idx is not incremented since
1388              * g_ptr_array_remove_index_fast replaces the
1389              * element at trace_idx with the array's last element.
1390              */
1391             g_ptr_array_remove_index_fast(session->traces, trace_idx);
1392         }
1393     }
1394     if (youngest_candidate_stream_iter) {
1395         *youngest_session_stream_iter = youngest_candidate_stream_iter;
1396         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1397     } else {
1398         /*
1399          * The only cases where we don't have a candidate for this
1400          * trace is:
1401          *  1. if we reached the end of all the iterators of all the
1402          *  traces of this session,
1403          *  2. if we never had live stream iterator in the first place.
1404          *
1405          * In either cases, we return END.
1406          */
1407         BT_ASSERT(session->traces->len == 0);
1408         stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_END;
1409     }
1410 end:
1411     return stream_iter_status;
1412 }
1413
1414 static inline void put_messages(bt_message_array_const msgs, uint64_t count)
1415 {
1416     uint64_t i;
1417
1418     for (i = 0; i < count; i++) {
1419         BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
1420     }
1421 }
1422
1423 bt_message_iterator_class_next_method_status
1424 lttng_live_msg_iter_next(bt_self_message_iterator *self_msg_it, bt_message_array_const msgs,
1425                          uint64_t capacity, uint64_t *count)
1426 {
1427     try {
1428         bt_message_iterator_class_next_method_status status;
1429         enum lttng_live_viewer_status viewer_status;
1430         struct lttng_live_msg_iter *lttng_live_msg_iter =
1431             (struct lttng_live_msg_iter *) bt_self_message_iterator_get_data(self_msg_it);
1432         struct lttng_live_component *lttng_live = lttng_live_msg_iter->lttng_live_comp;
1433         enum lttng_live_iterator_status stream_iter_status;
1434         uint64_t session_idx;
1435
1436         *count = 0;
1437
1438         BT_ASSERT_DBG(lttng_live_msg_iter);
1439
1440         if (G_UNLIKELY(lttng_live_msg_iter->was_interrupted)) {
1441             /*
1442          * The iterator was interrupted in a previous call to the
1443          * `_next()` method. We currently do not support generating
1444          * messages after such event. The babeltrace2 CLI should never
1445          * be running the graph after being interrupted. So this check
1446          * is to prevent other graph users from using this live
1447          * iterator in an messed up internal state.
1448          */
1449             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1450             BT_CPPLOGE_APPEND_CAUSE_SPEC(
1451                 lttng_live_msg_iter->logger,
1452                 "Message iterator was interrupted during a previous call to the `next()` and currently does not support continuing after such event.");
1453             goto end;
1454         }
1455
1456         /*
1457      * Clear all the invalid message reference that might be left over in
1458      * the output array.
1459      */
1460         memset(msgs, 0, capacity * sizeof(*msgs));
1461
1462         /*
1463      * If no session are exposed on the relay found at the url provided by
1464      * the user, session count will be 0. In this case, we return status
1465      * end to return gracefully.
1466      */
1467         if (lttng_live_msg_iter->sessions->len == 0) {
1468             if (lttng_live->params.sess_not_found_act != SESSION_NOT_FOUND_ACTION_CONTINUE) {
1469                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1470                 goto end;
1471             } else {
1472                 /*
1473              * The are no more active session for this session
1474              * name. Retry to create a viewer session for the
1475              * requested session name.
1476              */
1477                 viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
1478                 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1479                     if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1480                         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1481                         BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1482                                                      "Error creating LTTng live viewer session");
1483                     } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1484                         status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
1485                     } else {
1486                         bt_common_abort();
1487                     }
1488                     goto end;
1489                 }
1490             }
1491         }
1492
1493         if (lttng_live_msg_iter->active_stream_iter == 0) {
1494             lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter);
1495         }
1496
1497         /*
1498      * Here the muxing of message is done.
1499      *
1500      * We need to iterate over all the streams of all the traces of all the
1501      * viewer sessions in order to get the message with the smallest
1502      * timestamp. In this case, a session is a viewer session and there is
1503      * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1504      * kernel). Each viewer session can have multiple traces, for example,
1505      * 64bit UST viewer sessions could have multiple per-pid traces.
1506      *
1507      * We iterate over the streams of each traces to update and see what is
1508      * their next message's timestamp. From those timestamps, we select the
1509      * message with the smallest timestamp as the best candidate message
1510      * for that trace and do the same thing across all the sessions.
1511      *
1512      * We then compare the timestamp of best candidate message of all the
1513      * sessions to pick the message with the smallest timestamp and we
1514      * return it.
1515      */
1516         while (*count < capacity) {
1517             struct lttng_live_stream_iterator *youngest_stream_iter = NULL,
1518                                               *candidate_stream_iter = NULL;
1519             int64_t youngest_msg_ts_ns = INT64_MAX;
1520
1521             BT_ASSERT_DBG(lttng_live_msg_iter->sessions);
1522             session_idx = 0;
1523             while (session_idx < lttng_live_msg_iter->sessions->len) {
1524                 struct lttng_live_session *session = (lttng_live_session *) g_ptr_array_index(
1525                     lttng_live_msg_iter->sessions, session_idx);
1526
1527                 /* Find the best candidate message to send downstream. */
1528                 stream_iter_status = next_stream_iterator_for_session(lttng_live_msg_iter, session,
1529                                                                       &candidate_stream_iter);
1530
1531                 /* If we receive an END status, it means that either:
1532              * - Those traces never had active streams (UST with no
1533              *   data produced yet),
1534              * - All live stream iterators have ENDed.*/
1535                 if (stream_iter_status == LTTNG_LIVE_ITERATOR_STATUS_END) {
1536                     if (session->closed && session->traces->len == 0) {
1537                         /*
1538                      * Remove the session from the list.
1539                      * session_idx is not modified since
1540                      * g_ptr_array_remove_index_fast
1541                      * replaces the the removed element with
1542                      * the array's last element.
1543                      */
1544                         g_ptr_array_remove_index_fast(lttng_live_msg_iter->sessions, session_idx);
1545                     } else {
1546                         session_idx++;
1547                     }
1548                     continue;
1549                 }
1550
1551                 if (stream_iter_status != LTTNG_LIVE_ITERATOR_STATUS_OK) {
1552                     goto return_status;
1553                 }
1554
1555                 if (G_UNLIKELY(youngest_stream_iter == NULL) ||
1556                     candidate_stream_iter->current_msg_ts_ns < youngest_msg_ts_ns) {
1557                     youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1558                     youngest_stream_iter = candidate_stream_iter;
1559                 } else if (candidate_stream_iter->current_msg_ts_ns == youngest_msg_ts_ns) {
1560                     /*
1561                  * The currently selected message to be sent
1562                  * downstream next has the exact same timestamp
1563                  * that of the current candidate message. We
1564                  * must break the tie in a predictable manner.
1565                  */
1566                     BT_CPPLOGD_STR_SPEC(
1567                         lttng_live_msg_iter->logger,
1568                         "Two of the next message candidates have the same timestamps, pick one deterministically.");
1569                     /*
1570                  * Order the messages in an arbitrary but
1571                  * deterministic way.
1572                  */
1573                     int ret = common_muxing_compare_messages(candidate_stream_iter->current_msg,
1574                                                              youngest_stream_iter->current_msg);
1575                     if (ret < 0) {
1576                         /*
1577                      * The `candidate_stream_iter->current_msg`
1578                      * should go first. Update the next
1579                      * iterator and the current timestamp.
1580                      */
1581                         youngest_msg_ts_ns = candidate_stream_iter->current_msg_ts_ns;
1582                         youngest_stream_iter = candidate_stream_iter;
1583                     } else if (ret == 0) {
1584                         /* Unable to pick which one should go first. */
1585                         BT_CPPLOGW_SPEC(
1586                             lttng_live_msg_iter->logger,
1587                             "Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1588                             "next-stream-iter-addr={}"
1589                             "candidate-stream-iter-addr={}",
1590                             fmt::ptr(youngest_stream_iter), fmt::ptr(candidate_stream_iter));
1591                     }
1592                 }
1593
1594                 session_idx++;
1595             }
1596
1597             if (!youngest_stream_iter) {
1598                 stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1599                 goto return_status;
1600             }
1601
1602             BT_ASSERT_DBG(youngest_stream_iter->current_msg);
1603             /* Ensure monotonicity. */
1604             BT_ASSERT_DBG(lttng_live_msg_iter->last_msg_ts_ns <=
1605                           youngest_stream_iter->current_msg_ts_ns);
1606
1607             /*
1608          * Insert the next message to the message batch. This will set
1609          * stream iterator current message to NULL so that next time
1610          * we fetch the next message of that stream iterator
1611          */
1612             BT_MESSAGE_MOVE_REF(msgs[*count], youngest_stream_iter->current_msg);
1613             (*count)++;
1614
1615             /* Update the last timestamp in nanoseconds sent downstream. */
1616             lttng_live_msg_iter->last_msg_ts_ns = youngest_msg_ts_ns;
1617             youngest_stream_iter->current_msg_ts_ns = INT64_MAX;
1618
1619             stream_iter_status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1620         }
1621
1622 return_status:
1623         switch (stream_iter_status) {
1624         case LTTNG_LIVE_ITERATOR_STATUS_OK:
1625         case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
1626             /*
1627          * If we gathered messages, return _OK even if the graph was
1628          * interrupted. This allows for the components downstream to at
1629          * least get the those messages. If the graph was indeed
1630          * interrupted there should not be another _next() call as the
1631          * application will tear down the graph. This component class
1632          * doesn't support restarting after an interruption.
1633          */
1634             if (*count > 0) {
1635                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
1636             } else {
1637                 status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
1638             }
1639             break;
1640         case LTTNG_LIVE_ITERATOR_STATUS_END:
1641             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
1642             break;
1643         case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
1644             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1645                                          "Memory error preparing the next batch of messages: "
1646                                          "live-iter-status={}",
1647                                          stream_iter_status);
1648             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1649             break;
1650         case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
1651         case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
1652         case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
1653             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1654                                          "Error preparing the next batch of messages: "
1655                                          "live-iter-status={}",
1656                                          stream_iter_status);
1657
1658             status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1659             /* Put all existing messages on error. */
1660             put_messages(msgs, *count);
1661             break;
1662         default:
1663             bt_common_abort();
1664         }
1665
1666 end:
1667         return status;
1668     } catch (const std::bad_alloc&) {
1669         return BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
1670     } catch (const bt2::Error&) {
1671         return BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
1672     }
1673 }
1674
1675 static struct lttng_live_msg_iter *
1676 lttng_live_msg_iter_create(struct lttng_live_component *lttng_live_comp,
1677                            bt_self_message_iterator *self_msg_it)
1678 {
1679     lttng_live_msg_iter *msg_iter = new lttng_live_msg_iter {lttng_live_comp->logger};
1680     msg_iter->self_comp = lttng_live_comp->self_comp;
1681     msg_iter->lttng_live_comp = lttng_live_comp;
1682     msg_iter->self_msg_iter = self_msg_it;
1683
1684     msg_iter->active_stream_iter = 0;
1685     msg_iter->last_msg_ts_ns = INT64_MIN;
1686     msg_iter->was_interrupted = false;
1687
1688     msg_iter->sessions =
1689         g_ptr_array_new_with_free_func((GDestroyNotify) lttng_live_destroy_session);
1690     BT_ASSERT(msg_iter->sessions);
1691
1692     return msg_iter;
1693 }
1694
1695 bt_message_iterator_class_initialize_method_status
1696 lttng_live_msg_iter_init(bt_self_message_iterator *self_msg_it,
1697                          bt_self_message_iterator_configuration *, bt_self_component_port_output *)
1698 {
1699     try {
1700         bt_message_iterator_class_initialize_method_status status;
1701         struct lttng_live_component *lttng_live;
1702         struct lttng_live_msg_iter *lttng_live_msg_iter;
1703         enum lttng_live_viewer_status viewer_status;
1704         bt_self_component *self_comp = bt_self_message_iterator_borrow_component(self_msg_it);
1705
1706         lttng_live = (lttng_live_component *) bt_self_component_get_data(self_comp);
1707
1708         /* There can be only one downstream iterator at the same time. */
1709         BT_ASSERT(!lttng_live->has_msg_iter);
1710         lttng_live->has_msg_iter = true;
1711
1712         lttng_live_msg_iter = lttng_live_msg_iter_create(lttng_live, self_msg_it);
1713         if (!lttng_live_msg_iter) {
1714             BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live->logger,
1715                                          "Failed to create lttng_live_msg_iter");
1716             status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1717             goto error;
1718         }
1719
1720         viewer_status = live_viewer_connection_create(
1721             lttng_live->params.url->str, false, lttng_live_msg_iter, lttng_live_msg_iter->logger,
1722             &lttng_live_msg_iter->viewer_connection);
1723         if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1724             if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1725                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1726                                              "Failed to create viewer connection");
1727             } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1728                 /*
1729              * Interruption in the _iter_init() method is not
1730              * supported. Return an error.
1731              */
1732                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1733                                              "Interrupted while creating viewer connection");
1734             }
1735
1736             status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1737             goto error;
1738         }
1739
1740         viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
1741         if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1742             if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1743                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1744                                              "Failed to create viewer session");
1745             } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1746                 /*
1747              * Interruption in the _iter_init() method is not
1748              * supported. Return an error.
1749              */
1750                 BT_CPPLOGE_APPEND_CAUSE_SPEC(lttng_live_msg_iter->logger,
1751                                              "Interrupted when creating viewer session");
1752             }
1753
1754             status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1755             goto error;
1756         }
1757
1758         if (lttng_live_msg_iter->sessions->len == 0) {
1759             switch (lttng_live->params.sess_not_found_act) {
1760             case SESSION_NOT_FOUND_ACTION_CONTINUE:
1761                 BT_CPPLOGI_SPEC(
1762                     lttng_live_msg_iter->logger,
1763                     "Unable to connect to the requested live viewer session. "
1764                     "Keep trying to connect because of {}=\"{}\" component parameter: url=\"{}\"",
1765                     SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_CONTINUE_STR,
1766                     lttng_live->params.url->str);
1767                 break;
1768             case SESSION_NOT_FOUND_ACTION_FAIL:
1769                 BT_CPPLOGE_APPEND_CAUSE_SPEC(
1770                     lttng_live_msg_iter->logger,
1771                     "Unable to connect to the requested live viewer session. "
1772                     "Fail the message iterator initialization because of {}=\"{}\" "
1773                     "component parameter: url =\"{}\"",
1774                     SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_FAIL_STR,
1775                     lttng_live->params.url->str);
1776                 status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1777                 goto error;
1778             case SESSION_NOT_FOUND_ACTION_END:
1779                 BT_CPPLOGI_SPEC(lttng_live_msg_iter->logger,
1780                                 "Unable to connect to the requested live viewer session. "
1781                                 "End gracefully at the first _next() call because of {}=\"{}\""
1782                                 " component parameter: url=\"{}\"",
1783                                 SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_END_STR,
1784                                 lttng_live->params.url->str);
1785                 break;
1786             default:
1787                 bt_common_abort();
1788             }
1789         }
1790
1791         bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
1792         status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
1793         goto end;
1794
1795 error:
1796         lttng_live_msg_iter_destroy(lttng_live_msg_iter);
1797 end:
1798         return status;
1799     } catch (const std::bad_alloc&) {
1800         return BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
1801     } catch (const bt2::Error&) {
1802         return BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
1803     }
1804 }
1805
1806 static struct bt_param_validation_map_value_entry_descr list_sessions_params[] = {
1807     {URL_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY,
1808      bt_param_validation_value_descr::makeString()},
1809     BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END};
1810
1811 static bt_component_class_query_method_status
1812 lttng_live_query_list_sessions(const bt_value *params, const bt_value **result,
1813                                const bt2c::Logger& logger)
1814 {
1815     bt_component_class_query_method_status status;
1816     const bt_value *url_value = NULL;
1817     const char *url;
1818     struct live_viewer_connection *viewer_connection = NULL;
1819     enum lttng_live_viewer_status viewer_status;
1820     enum bt_param_validation_status validation_status;
1821     gchar *validate_error = NULL;
1822
1823     validation_status = bt_param_validation_validate(params, list_sessions_params, &validate_error);
1824     if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
1825         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
1826         goto error;
1827     } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
1828         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1829         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "{}", validate_error);
1830         goto error;
1831     }
1832
1833     url_value = bt_value_map_borrow_entry_value_const(params, URL_PARAM);
1834     url = bt_value_string_get(url_value);
1835
1836     viewer_status = live_viewer_connection_create(url, true, NULL, logger, &viewer_connection);
1837     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1838         if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1839             BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Failed to create viewer connection");
1840             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1841         } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1842             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
1843         } else {
1844             bt_common_abort();
1845         }
1846         goto error;
1847     }
1848
1849     status = live_viewer_connection_list_sessions(viewer_connection, result);
1850     if (status != BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK) {
1851         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Failed to list viewer sessions");
1852         goto error;
1853     }
1854
1855     goto end;
1856
1857 error:
1858     BT_VALUE_PUT_REF_AND_RESET(*result);
1859
1860     if (status >= 0) {
1861         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1862     }
1863
1864 end:
1865     if (viewer_connection) {
1866         live_viewer_connection_destroy(viewer_connection);
1867     }
1868
1869     g_free(validate_error);
1870
1871     return status;
1872 }
1873
1874 static bt_component_class_query_method_status
1875 lttng_live_query_support_info(const bt_value *params, const bt_value **result,
1876                               const bt2c::Logger& logger)
1877 {
1878     bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1879     const bt_value *input_type_value;
1880     const bt_value *input_value;
1881     double weight = 0;
1882     struct bt_common_lttng_live_url_parts parts = {};
1883
1884     /* Used by the logging macros */
1885     __attribute__((unused)) bt_self_component *self_comp = NULL;
1886
1887     *result = NULL;
1888     input_type_value = bt_value_map_borrow_entry_value_const(params, "type");
1889     if (!input_type_value) {
1890         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Missing expected `type` parameter.");
1891         goto error;
1892     }
1893
1894     if (!bt_value_is_string(input_type_value)) {
1895         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "`type` parameter is not a string value.");
1896         goto error;
1897     }
1898
1899     if (strcmp(bt_value_string_get(input_type_value), "string") != 0) {
1900         /* We don't handle file system paths */
1901         goto create_result;
1902     }
1903
1904     input_value = bt_value_map_borrow_entry_value_const(params, "input");
1905     if (!input_value) {
1906         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "Missing expected `input` parameter.");
1907         goto error;
1908     }
1909
1910     if (!bt_value_is_string(input_value)) {
1911         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "`input` parameter is not a string value.");
1912         goto error;
1913     }
1914
1915     parts = bt_common_parse_lttng_live_url(bt_value_string_get(input_value), NULL, 0);
1916     if (parts.session_name) {
1917         /*
1918          * Looks pretty much like an LTTng live URL: we got the
1919          * session name part, which forms a complete URL.
1920          */
1921         weight = .75;
1922     }
1923
1924 create_result:
1925     *result = bt_value_real_create_init(weight);
1926     if (!*result) {
1927         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
1928         goto error;
1929     }
1930
1931     goto end;
1932
1933 error:
1934     if (status >= 0) {
1935         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1936     }
1937
1938     BT_ASSERT(!*result);
1939
1940 end:
1941     bt_common_destroy_lttng_live_url_parts(&parts);
1942     return status;
1943 }
1944
1945 bt_component_class_query_method_status lttng_live_query(bt_self_component_class_source *comp_class,
1946                                                         bt_private_query_executor *priv_query_exec,
1947                                                         const char *object, const bt_value *params,
1948                                                         __attribute__((unused)) void *method_data,
1949                                                         const bt_value **result)
1950 {
1951     try {
1952         bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
1953         bt2c::Logger logger {bt2::SelfComponentClass {comp_class},
1954                              bt2::PrivateQueryExecutor {priv_query_exec},
1955                              "PLUGIN/SRC.CTF.LTTNG-LIVE/QUERY"};
1956
1957         if (strcmp(object, "sessions") == 0) {
1958             status = lttng_live_query_list_sessions(params, result, logger);
1959         } else if (strcmp(object, "babeltrace.support-info") == 0) {
1960             status = lttng_live_query_support_info(params, result, logger);
1961         } else {
1962             BT_CPPLOGI_SPEC(logger, "Unknown query object `{}`", object);
1963             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT;
1964             goto end;
1965         }
1966
1967 end:
1968         return status;
1969     } catch (const std::bad_alloc&) {
1970         return BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
1971     } catch (const bt2::Error&) {
1972         return BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
1973     }
1974 }
1975
1976 static void lttng_live_component_destroy_data(struct lttng_live_component *lttng_live)
1977 {
1978     if (!lttng_live) {
1979         return;
1980     }
1981
1982     if (lttng_live->params.url) {
1983         g_string_free(lttng_live->params.url, TRUE);
1984     }
1985
1986     delete lttng_live;
1987 }
1988
1989 void lttng_live_component_finalize(bt_self_component_source *component)
1990 {
1991     lttng_live_component *data = (lttng_live_component *) bt_self_component_get_data(
1992         bt_self_component_source_as_self_component(component));
1993
1994     if (!data) {
1995         return;
1996     }
1997     lttng_live_component_destroy_data(data);
1998 }
1999
2000 static enum session_not_found_action
2001 parse_session_not_found_action_param(const bt_value *no_session_param)
2002 {
2003     enum session_not_found_action action;
2004     const char *no_session_act_str = bt_value_string_get(no_session_param);
2005
2006     if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_CONTINUE_STR) == 0) {
2007         action = SESSION_NOT_FOUND_ACTION_CONTINUE;
2008     } else if (strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_FAIL_STR) == 0) {
2009         action = SESSION_NOT_FOUND_ACTION_FAIL;
2010     } else {
2011         BT_ASSERT(strcmp(no_session_act_str, SESS_NOT_FOUND_ACTION_END_STR) == 0);
2012         action = SESSION_NOT_FOUND_ACTION_END;
2013     }
2014
2015     return action;
2016 }
2017
2018 static bt_param_validation_value_descr inputs_elem_descr =
2019     bt_param_validation_value_descr::makeString();
2020
2021 static const char *sess_not_found_action_choices[] = {
2022     SESS_NOT_FOUND_ACTION_CONTINUE_STR,
2023     SESS_NOT_FOUND_ACTION_FAIL_STR,
2024     SESS_NOT_FOUND_ACTION_END_STR,
2025 };
2026
2027 static struct bt_param_validation_map_value_entry_descr params_descr[] = {
2028     {INPUTS_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY,
2029      bt_param_validation_value_descr::makeArray(1, 1, inputs_elem_descr)},
2030     {SESS_NOT_FOUND_ACTION_PARAM, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL,
2031      bt_param_validation_value_descr::makeString(sess_not_found_action_choices)},
2032     BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END};
2033
2034 static bt_component_class_initialize_method_status
2035 lttng_live_component_create(const bt_value *params, bt_self_component_source *self_comp,
2036                             struct lttng_live_component **component)
2037 {
2038     struct lttng_live_component *lttng_live = NULL;
2039     const bt_value *inputs_value;
2040     const bt_value *url_value;
2041     const bt_value *value;
2042     const char *url;
2043     enum bt_param_validation_status validation_status;
2044     gchar *validation_error = NULL;
2045     bt_component_class_initialize_method_status status;
2046     bt2c::Logger logger {bt2::SelfSourceComponent {self_comp}, "PLUGIN/SRC.CTF.LTTNG-LIVE/COMP"};
2047
2048     validation_status = bt_param_validation_validate(params, params_descr, &validation_error);
2049     if (validation_status == BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR) {
2050         status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2051         goto error;
2052     } else if (validation_status == BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR) {
2053         BT_CPPLOGE_APPEND_CAUSE_SPEC(logger, "{}", validation_error);
2054         status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
2055         goto error;
2056     }
2057
2058     lttng_live = new lttng_live_component {std::move(logger)};
2059     lttng_live->self_comp = bt_self_component_source_as_self_component(self_comp);
2060     lttng_live->max_query_size = MAX_QUERY_SIZE;
2061     lttng_live->has_msg_iter = false;
2062
2063     inputs_value = bt_value_map_borrow_entry_value_const(params, INPUTS_PARAM);
2064     url_value = bt_value_array_borrow_element_by_index_const(inputs_value, 0);
2065     url = bt_value_string_get(url_value);
2066
2067     lttng_live->params.url = g_string_new(url);
2068     if (!lttng_live->params.url) {
2069         status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2070         goto error;
2071     }
2072
2073     value = bt_value_map_borrow_entry_value_const(params, SESS_NOT_FOUND_ACTION_PARAM);
2074     if (value) {
2075         lttng_live->params.sess_not_found_act = parse_session_not_found_action_param(value);
2076     } else {
2077         BT_CPPLOGI_SPEC(lttng_live->logger,
2078                         "Optional `{}` parameter is missing: defaulting to `{}`.",
2079                         SESS_NOT_FOUND_ACTION_PARAM, SESS_NOT_FOUND_ACTION_CONTINUE_STR);
2080         lttng_live->params.sess_not_found_act = SESSION_NOT_FOUND_ACTION_CONTINUE;
2081     }
2082
2083     status = BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK;
2084     goto end;
2085
2086 error:
2087     lttng_live_component_destroy_data(lttng_live);
2088     lttng_live = NULL;
2089 end:
2090     g_free(validation_error);
2091
2092     *component = lttng_live;
2093     return status;
2094 }
2095
2096 bt_component_class_initialize_method_status
2097 lttng_live_component_init(bt_self_component_source *self_comp_src,
2098                           bt_self_component_source_configuration *, const bt_value *params, void *)
2099 {
2100     try {
2101         struct lttng_live_component *lttng_live;
2102         bt_component_class_initialize_method_status ret;
2103         bt_self_component *self_comp = bt_self_component_source_as_self_component(self_comp_src);
2104         bt_self_component_add_port_status add_port_status;
2105
2106         ret = lttng_live_component_create(params, self_comp_src, &lttng_live);
2107         if (ret != BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK) {
2108             goto error;
2109         }
2110
2111         add_port_status =
2112             bt_self_component_source_add_output_port(self_comp_src, "out", NULL, NULL);
2113         if (add_port_status != BT_SELF_COMPONENT_ADD_PORT_STATUS_OK) {
2114             ret = (bt_component_class_initialize_method_status) add_port_status;
2115             goto end;
2116         }
2117
2118         bt_self_component_set_data(self_comp, lttng_live);
2119         goto end;
2120
2121 error:
2122         lttng_live_component_destroy_data(lttng_live);
2123         lttng_live = NULL;
2124 end:
2125         return ret;
2126     } catch (const std::bad_alloc&) {
2127         return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
2128     } catch (const bt2::Error&) {
2129         return BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
2130     }
2131 }