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