4 * Babeltrace CTF LTTng-live Client Component
6 * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
7 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
8 * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
10 * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
12 * Permission is hereby granted, free of charge, to any person obtaining a copy
13 * of this software and associated documentation files (the "Software"), to deal
14 * in the Software without restriction, including without limitation the rights
15 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
16 * copies of the Software, and to permit persons to whom the Software is
17 * furnished to do so, subject to the following conditions:
19 * The above copyright notice and this permission notice shall be included in
20 * all copies or substantial portions of the Software.
22 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
23 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
24 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
25 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
26 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
27 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
31 #define BT_COMP_LOG_SELF_COMP self_comp
32 #define BT_LOG_OUTPUT_LEVEL log_level
33 #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE"
34 #include "logging/comp-logging.h"
42 #include "common/assert.h"
43 #include <babeltrace2/babeltrace.h>
44 #include "compat/compiler.h"
45 #include <babeltrace2/types.h>
47 #include "plugins/common/muxing/muxing.h"
48 #include "plugins/common/param-validation/param-validation.h"
50 #include "data-stream.h"
52 #include "lttng-live.h"
54 #define MAX_QUERY_SIZE (256*1024)
55 #define URL_PARAM "url"
56 #define INPUTS_PARAM "inputs"
57 #define SESS_NOT_FOUND_ACTION_PARAM "session-not-found-action"
58 #define SESS_NOT_FOUND_ACTION_CONTINUE_STR "continue"
59 #define SESS_NOT_FOUND_ACTION_FAIL_STR "fail"
60 #define SESS_NOT_FOUND_ACTION_END_STR "end"
62 #define print_dbg(fmt, ...) BT_COMP_LOGD(fmt, ## __VA_ARGS__)
65 const char *lttng_live_iterator_status_string(
66 enum lttng_live_iterator_status status
)
69 case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
:
70 return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
71 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN
:
72 return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
73 case LTTNG_LIVE_ITERATOR_STATUS_END
:
74 return "LTTNG_LIVE_ITERATOR_STATUS_END";
75 case LTTNG_LIVE_ITERATOR_STATUS_OK
:
76 return "LTTNG_LIVE_ITERATOR_STATUS_OK";
77 case LTTNG_LIVE_ITERATOR_STATUS_INVAL
:
78 return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
79 case LTTNG_LIVE_ITERATOR_STATUS_ERROR
:
80 return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
81 case LTTNG_LIVE_ITERATOR_STATUS_NOMEM
:
82 return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
83 case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED
:
84 return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
91 const char *lttng_live_stream_state_string(enum lttng_live_stream_state state
)
94 case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA
:
95 return "ACTIVE_NO_DATA";
96 case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA
:
97 return "QUIESCENT_NO_DATA";
98 case LTTNG_LIVE_STREAM_QUIESCENT
:
100 case LTTNG_LIVE_STREAM_ACTIVE_DATA
:
101 return "ACTIVE_DATA";
102 case LTTNG_LIVE_STREAM_EOF
:
109 #define LTTNG_LIVE_LOGD_STREAM_ITER(live_stream_iter) \
111 BT_COMP_LOGD("Live stream iterator state=%s, last-inact-ts=%" PRId64 \
112 ", curr-inact-ts %" PRId64, \
113 lttng_live_stream_state_string(live_stream_iter->state), \
114 live_stream_iter->last_inactivity_ts, \
115 live_stream_iter->current_inactivity_ts); \
119 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter
*msg_iter
)
128 ret
= bt_self_message_iterator_is_interrupted(
129 msg_iter
->self_msg_iter
);
136 struct lttng_live_trace
*lttng_live_find_trace(struct lttng_live_session
*session
,
140 struct lttng_live_trace
*ret_trace
= NULL
;
142 for (trace_idx
= 0; trace_idx
< session
->traces
->len
; trace_idx
++) {
143 struct lttng_live_trace
*trace
=
144 g_ptr_array_index(session
->traces
, trace_idx
);
145 if (trace
->id
== trace_id
) {
156 void lttng_live_destroy_trace(struct lttng_live_trace
*trace
)
158 bt_logging_level log_level
= trace
->log_level
;
159 bt_self_component
*self_comp
= trace
->self_comp
;
161 BT_COMP_LOGD("Destroying live trace: trace-id=%"PRIu64
, trace
->id
);
163 BT_ASSERT(trace
->stream_iterators
);
164 g_ptr_array_free(trace
->stream_iterators
, TRUE
);
166 BT_TRACE_PUT_REF_AND_RESET(trace
->trace
);
167 BT_TRACE_CLASS_PUT_REF_AND_RESET(trace
->trace_class
);
169 lttng_live_metadata_fini(trace
);
174 struct lttng_live_trace
*lttng_live_create_trace(struct lttng_live_session
*session
,
177 struct lttng_live_trace
*trace
= NULL
;
178 bt_logging_level log_level
= session
->log_level
;
179 bt_self_component
*self_comp
= session
->self_comp
;
181 BT_COMP_LOGD("Creating live trace: "
182 "session-id=%"PRIu64
", trace-id=%"PRIu64
,
183 session
->id
, trace_id
);
184 trace
= g_new0(struct lttng_live_trace
, 1);
186 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
187 "Failed to allocate live trace");
190 trace
->log_level
= session
->log_level
;
191 trace
->self_comp
= session
->self_comp
;
192 trace
->session
= session
;
193 trace
->id
= trace_id
;
194 trace
->trace_class
= NULL
;
196 trace
->stream_iterators
= g_ptr_array_new_with_free_func(
197 (GDestroyNotify
) lttng_live_stream_iterator_destroy
);
198 BT_ASSERT(trace
->stream_iterators
);
199 trace
->metadata_stream_state
= LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED
;
200 g_ptr_array_add(session
->traces
, trace
);
211 struct lttng_live_trace
*lttng_live_borrow_trace(
212 struct lttng_live_session
*session
, uint64_t trace_id
)
214 struct lttng_live_trace
*trace
;
216 trace
= lttng_live_find_trace(session
, trace_id
);
221 /* The session is the owner of the newly created trace. */
222 trace
= lttng_live_create_trace(session
, trace_id
);
229 int lttng_live_add_session(struct lttng_live_msg_iter
*lttng_live_msg_iter
,
230 uint64_t session_id
, const char *hostname
,
231 const char *session_name
)
234 struct lttng_live_session
*session
;
235 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
236 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
238 BT_COMP_LOGD("Adding live session: "
239 "session-id=%" PRIu64
", hostname=\"%s\" session-name=\"%s\"",
240 session_id
, hostname
, session_name
);
242 session
= g_new0(struct lttng_live_session
, 1);
244 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
245 "Failed to allocate live session");
249 session
->log_level
= lttng_live_msg_iter
->log_level
;
250 session
->self_comp
= lttng_live_msg_iter
->self_comp
;
251 session
->id
= session_id
;
252 session
->traces
= g_ptr_array_new_with_free_func(
253 (GDestroyNotify
) lttng_live_destroy_trace
);
254 BT_ASSERT(session
->traces
);
255 session
->lttng_live_msg_iter
= lttng_live_msg_iter
;
256 session
->new_streams_needed
= true;
257 session
->hostname
= g_string_new(hostname
);
258 BT_ASSERT(session
->hostname
);
260 session
->session_name
= g_string_new(session_name
);
261 BT_ASSERT(session
->session_name
);
263 g_ptr_array_add(lttng_live_msg_iter
->sessions
, session
);
273 void lttng_live_destroy_session(struct lttng_live_session
*session
)
275 bt_logging_level log_level
;
276 bt_self_component
*self_comp
;
282 log_level
= session
->log_level
;
283 self_comp
= session
->self_comp
;
284 BT_COMP_LOGD("Destroying live session: "
285 "session-id=%"PRIu64
", session-name=\"%s\"",
286 session
->id
, session
->session_name
->str
);
287 if (session
->id
!= -1ULL) {
288 if (lttng_live_detach_session(session
)) {
289 if (!lttng_live_graph_is_canceled(
290 session
->lttng_live_msg_iter
)) {
291 /* Old relayd cannot detach sessions. */
292 BT_COMP_LOGD("Unable to detach lttng live session %" PRIu64
,
299 if (session
->traces
) {
300 g_ptr_array_free(session
->traces
, TRUE
);
303 if (session
->hostname
) {
304 g_string_free(session
->hostname
, TRUE
);
306 if (session
->session_name
) {
307 g_string_free(session
->session_name
, TRUE
);
316 void lttng_live_msg_iter_destroy(struct lttng_live_msg_iter
*lttng_live_msg_iter
)
318 if (!lttng_live_msg_iter
) {
322 if (lttng_live_msg_iter
->sessions
) {
323 g_ptr_array_free(lttng_live_msg_iter
->sessions
, TRUE
);
326 if (lttng_live_msg_iter
->viewer_connection
) {
327 live_viewer_connection_destroy(lttng_live_msg_iter
->viewer_connection
);
329 BT_ASSERT(lttng_live_msg_iter
->lttng_live_comp
);
330 BT_ASSERT(lttng_live_msg_iter
->lttng_live_comp
->has_msg_iter
);
332 /* All stream iterators must be destroyed at this point. */
333 BT_ASSERT(lttng_live_msg_iter
->active_stream_iter
== 0);
334 lttng_live_msg_iter
->lttng_live_comp
->has_msg_iter
= false;
336 g_free(lttng_live_msg_iter
);
343 void lttng_live_msg_iter_finalize(bt_self_message_iterator
*self_msg_iter
)
345 struct lttng_live_msg_iter
*lttng_live_msg_iter
;
347 BT_ASSERT(self_msg_iter
);
349 lttng_live_msg_iter
= bt_self_message_iterator_get_data(self_msg_iter
);
350 BT_ASSERT(lttng_live_msg_iter
);
351 lttng_live_msg_iter_destroy(lttng_live_msg_iter
);
355 enum lttng_live_iterator_status
lttng_live_iterator_next_check_stream_state(
356 struct lttng_live_stream_iterator
*lttng_live_stream
)
358 bt_logging_level log_level
= lttng_live_stream
->log_level
;
359 bt_self_component
*self_comp
= lttng_live_stream
->self_comp
;
361 switch (lttng_live_stream
->state
) {
362 case LTTNG_LIVE_STREAM_QUIESCENT
:
363 case LTTNG_LIVE_STREAM_ACTIVE_DATA
:
365 case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA
:
367 BT_COMP_LOGF("Unexpected stream state \"ACTIVE_NO_DATA\"");
369 case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA
:
371 BT_COMP_LOGF("Unexpected stream state \"QUIESCENT_NO_DATA\"");
373 case LTTNG_LIVE_STREAM_EOF
:
376 return LTTNG_LIVE_ITERATOR_STATUS_OK
;
380 * For active no data stream, fetch next data. It can be either:
381 * - quiescent: need to put it in the prio heap at quiescent end
383 * - have data: need to wire up first event into the prio heap,
384 * - have no data on this stream at this point: need to retry (AGAIN) or
388 enum lttng_live_iterator_status
lttng_live_iterator_next_handle_one_no_data_stream(
389 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
390 struct lttng_live_stream_iterator
*lttng_live_stream
)
392 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
393 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
394 enum lttng_live_iterator_status ret
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
395 enum lttng_live_stream_state orig_state
= lttng_live_stream
->state
;
396 struct packet_index index
;
398 if (lttng_live_stream
->trace
->metadata_stream_state
==
399 LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED
) {
400 ret
= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
;
403 if (lttng_live_stream
->trace
->session
->new_streams_needed
) {
404 ret
= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
;
407 if (lttng_live_stream
->state
!= LTTNG_LIVE_STREAM_ACTIVE_NO_DATA
&&
408 lttng_live_stream
->state
!= LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA
) {
411 ret
= lttng_live_get_next_index(lttng_live_msg_iter
, lttng_live_stream
,
413 if (ret
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
416 BT_ASSERT_DBG(lttng_live_stream
->state
!= LTTNG_LIVE_STREAM_EOF
);
417 if (lttng_live_stream
->state
== LTTNG_LIVE_STREAM_QUIESCENT
) {
418 uint64_t last_inact_ts
= lttng_live_stream
->last_inactivity_ts
,
419 curr_inact_ts
= lttng_live_stream
->current_inactivity_ts
;
421 if (orig_state
== LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA
&&
422 last_inact_ts
== curr_inact_ts
) {
423 ret
= LTTNG_LIVE_ITERATOR_STATUS_AGAIN
;
424 LTTNG_LIVE_LOGD_STREAM_ITER(lttng_live_stream
);
426 ret
= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
;
430 lttng_live_stream
->base_offset
= index
.offset
;
431 lttng_live_stream
->offset
= index
.offset
;
432 lttng_live_stream
->len
= index
.packet_size
/ CHAR_BIT
;
434 if (ret
== LTTNG_LIVE_ITERATOR_STATUS_OK
) {
435 ret
= lttng_live_iterator_next_check_stream_state(lttng_live_stream
);
441 * Creation of the message requires the ctf trace class to be created
442 * beforehand, but the live protocol gives us all streams (including metadata)
443 * at once. So we split it in three steps: getting streams, getting metadata
444 * (which creates the ctf trace class), and then creating the per-stream
448 enum lttng_live_iterator_status
lttng_live_get_session(
449 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
450 struct lttng_live_session
*session
)
452 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
453 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
454 enum lttng_live_iterator_status status
;
457 BT_COMP_LOGD("Updating all streams for session: "
458 "session-id=%"PRIu64
", session-name=\"%s\"",
459 session
->id
, session
->session_name
->str
);
461 if (!session
->attached
) {
462 enum lttng_live_viewer_status attach_status
=
463 lttng_live_attach_session(session
,
464 lttng_live_msg_iter
->self_msg_iter
);
465 if (attach_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
466 if (lttng_live_graph_is_canceled(lttng_live_msg_iter
)) {
468 * Clear any causes appended in
469 * `lttng_live_attach_session()` as we want to
470 * return gracefully since the graph was
473 bt_current_thread_clear_error();
474 status
= LTTNG_LIVE_ITERATOR_STATUS_AGAIN
;
476 status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
477 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
478 "Error attaching to LTTng live session");
484 status
= lttng_live_get_new_streams(session
,
485 lttng_live_msg_iter
->self_msg_iter
);
486 if (status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
&&
487 status
!= LTTNG_LIVE_ITERATOR_STATUS_END
) {
491 while (trace_idx
< session
->traces
->len
) {
492 struct lttng_live_trace
*trace
=
493 g_ptr_array_index(session
->traces
, trace_idx
);
495 status
= lttng_live_metadata_update(trace
);
497 case LTTNG_LIVE_ITERATOR_STATUS_END
:
498 case LTTNG_LIVE_ITERATOR_STATUS_OK
:
501 case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
:
502 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN
:
505 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
506 "Error updating trace metadata: "
507 "stream-iter-status=%s, trace-id=%"PRIu64
,
508 lttng_live_iterator_status_string(status
),
513 status
= lttng_live_lazy_msg_init(session
,
514 lttng_live_msg_iter
->self_msg_iter
);
521 void lttng_live_need_new_streams(struct lttng_live_msg_iter
*lttng_live_msg_iter
)
523 uint64_t session_idx
;
525 for (session_idx
= 0; session_idx
< lttng_live_msg_iter
->sessions
->len
;
527 struct lttng_live_session
*session
=
528 g_ptr_array_index(lttng_live_msg_iter
->sessions
, session_idx
);
529 session
->new_streams_needed
= true;
534 void lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter
*lttng_live_msg_iter
)
536 uint64_t session_idx
, trace_idx
;
538 for (session_idx
= 0; session_idx
< lttng_live_msg_iter
->sessions
->len
;
540 struct lttng_live_session
*session
=
541 g_ptr_array_index(lttng_live_msg_iter
->sessions
, session_idx
);
542 session
->new_streams_needed
= true;
543 for (trace_idx
= 0; trace_idx
< session
->traces
->len
;
545 struct lttng_live_trace
*trace
=
546 g_ptr_array_index(session
->traces
, trace_idx
);
548 BT_ASSERT(trace
->metadata_stream_state
!=
549 LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED
);
551 trace
->metadata_stream_state
= LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED
;
557 enum lttng_live_iterator_status
558 lttng_live_iterator_handle_new_streams_and_metadata(
559 struct lttng_live_msg_iter
*lttng_live_msg_iter
)
561 enum lttng_live_iterator_status status
;
562 enum lttng_live_viewer_status viewer_status
;
563 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
564 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
565 uint64_t session_idx
= 0, nr_sessions_opened
= 0;
566 struct lttng_live_session
*session
;
567 enum session_not_found_action sess_not_found_act
=
568 lttng_live_msg_iter
->lttng_live_comp
->params
.sess_not_found_act
;
570 BT_COMP_LOGD("Update data and metadata of all sessions"
571 "live-msg-iter-addr=%p", lttng_live_msg_iter
);
573 * In a remotely distant future, we could add a "new
574 * session" flag to the protocol, which would tell us that we
575 * need to query for new sessions even though we have sessions
578 if (lttng_live_msg_iter
->sessions
->len
== 0) {
579 if (sess_not_found_act
!= SESSION_NOT_FOUND_ACTION_CONTINUE
) {
580 BT_COMP_LOGD("No session found. Exiting in accordance with the `session-not-found-action` parameter");
581 status
= LTTNG_LIVE_ITERATOR_STATUS_END
;
584 BT_COMP_LOGD("No session found. Try creating a new one in accordance with the `session-not-found-action` parameter");
586 * Retry to create a viewer session for the requested
589 viewer_status
= lttng_live_create_viewer_session(lttng_live_msg_iter
);
590 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
591 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
592 status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
593 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
594 "Error creating LTTng live viewer session");
595 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
596 status
= LTTNG_LIVE_ITERATOR_STATUS_AGAIN
;
605 for (session_idx
= 0; session_idx
< lttng_live_msg_iter
->sessions
->len
;
607 session
= g_ptr_array_index(lttng_live_msg_iter
->sessions
,
609 status
= lttng_live_get_session(lttng_live_msg_iter
, session
);
611 case LTTNG_LIVE_ITERATOR_STATUS_OK
:
613 case LTTNG_LIVE_ITERATOR_STATUS_END
:
614 status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
619 if (!session
->closed
) {
620 nr_sessions_opened
++;
624 if (sess_not_found_act
!= SESSION_NOT_FOUND_ACTION_CONTINUE
&&
625 nr_sessions_opened
== 0) {
626 status
= LTTNG_LIVE_ITERATOR_STATUS_END
;
628 status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
636 enum lttng_live_iterator_status
emit_inactivity_message(
637 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
638 struct lttng_live_stream_iterator
*stream_iter
,
639 bt_message
**message
, uint64_t timestamp
)
641 enum lttng_live_iterator_status ret
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
642 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
643 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
644 bt_message
*msg
= NULL
;
646 BT_ASSERT(stream_iter
->trace
->clock_class
);
648 msg
= bt_message_message_iterator_inactivity_create(
649 lttng_live_msg_iter
->self_msg_iter
,
650 stream_iter
->trace
->clock_class
, timestamp
);
652 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
653 "Error emitting message iterator inactivity message");
662 ret
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
663 bt_message_put_ref(msg
);
668 enum lttng_live_iterator_status
lttng_live_iterator_next_handle_one_quiescent_stream(
669 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
670 struct lttng_live_stream_iterator
*lttng_live_stream
,
671 bt_message
**message
)
673 enum lttng_live_iterator_status ret
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
675 if (lttng_live_stream
->state
!= LTTNG_LIVE_STREAM_QUIESCENT
) {
676 return LTTNG_LIVE_ITERATOR_STATUS_OK
;
679 if (lttng_live_stream
->current_inactivity_ts
==
680 lttng_live_stream
->last_inactivity_ts
) {
681 lttng_live_stream
->state
= LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA
;
682 ret
= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
;
686 ret
= emit_inactivity_message(lttng_live_msg_iter
, lttng_live_stream
,
687 message
, lttng_live_stream
->current_inactivity_ts
);
689 lttng_live_stream
->last_inactivity_ts
=
690 lttng_live_stream
->current_inactivity_ts
;
696 int live_get_msg_ts_ns(struct lttng_live_stream_iterator
*stream_iter
,
697 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
698 const bt_message
*msg
, int64_t last_msg_ts_ns
,
701 const bt_clock_class
*clock_class
= NULL
;
702 const bt_clock_snapshot
*clock_snapshot
= NULL
;
704 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
705 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
708 BT_ASSERT_DBG(ts_ns
);
710 BT_COMP_LOGD("Getting message's timestamp: iter-data-addr=%p, msg-addr=%p, "
711 "last-msg-ts=%" PRId64
, lttng_live_msg_iter
, msg
,
714 switch (bt_message_get_type(msg
)) {
715 case BT_MESSAGE_TYPE_EVENT
:
716 clock_class
= bt_message_event_borrow_stream_class_default_clock_class_const(
718 BT_ASSERT_DBG(clock_class
);
720 clock_snapshot
= bt_message_event_borrow_default_clock_snapshot_const(
723 case BT_MESSAGE_TYPE_PACKET_BEGINNING
:
724 clock_class
= bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
726 BT_ASSERT(clock_class
);
728 clock_snapshot
= bt_message_packet_beginning_borrow_default_clock_snapshot_const(
731 case BT_MESSAGE_TYPE_PACKET_END
:
732 clock_class
= bt_message_packet_end_borrow_stream_class_default_clock_class_const(
734 BT_ASSERT(clock_class
);
736 clock_snapshot
= bt_message_packet_end_borrow_default_clock_snapshot_const(
739 case BT_MESSAGE_TYPE_DISCARDED_EVENTS
:
740 clock_class
= bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
742 BT_ASSERT(clock_class
);
744 clock_snapshot
= bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
747 case BT_MESSAGE_TYPE_DISCARDED_PACKETS
:
748 clock_class
= bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
750 BT_ASSERT(clock_class
);
752 clock_snapshot
= bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
755 case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY
:
756 clock_snapshot
= bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
760 /* All the other messages have a higher priority */
761 BT_COMP_LOGD_STR("Message has no timestamp: using the last message timestamp.");
762 *ts_ns
= last_msg_ts_ns
;
766 clock_class
= bt_clock_snapshot_borrow_clock_class_const(clock_snapshot
);
767 BT_ASSERT_DBG(clock_class
);
769 ret
= bt_clock_snapshot_get_ns_from_origin(clock_snapshot
, ts_ns
);
771 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
772 "Cannot get nanoseconds from Epoch of clock snapshot: "
773 "clock-snapshot-addr=%p", clock_snapshot
);
784 BT_COMP_LOGD("Found message's timestamp: "
785 "iter-data-addr=%p, msg-addr=%p, "
786 "last-msg-ts=%" PRId64
", ts=%" PRId64
,
787 lttng_live_msg_iter
, msg
, last_msg_ts_ns
, *ts_ns
);
794 enum lttng_live_iterator_status
lttng_live_iterator_next_handle_one_active_data_stream(
795 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
796 struct lttng_live_stream_iterator
*lttng_live_stream
,
797 bt_message
**message
)
799 enum lttng_live_iterator_status ret
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
800 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
801 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
802 enum ctf_msg_iter_status status
;
803 uint64_t session_idx
, trace_idx
;
805 for (session_idx
= 0; session_idx
< lttng_live_msg_iter
->sessions
->len
;
807 struct lttng_live_session
*session
=
808 g_ptr_array_index(lttng_live_msg_iter
->sessions
, session_idx
);
810 if (session
->new_streams_needed
) {
811 ret
= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
;
814 for (trace_idx
= 0; trace_idx
< session
->traces
->len
;
816 struct lttng_live_trace
*trace
=
817 g_ptr_array_index(session
->traces
, trace_idx
);
818 if (trace
->metadata_stream_state
== LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED
) {
819 ret
= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
;
825 if (lttng_live_stream
->state
!= LTTNG_LIVE_STREAM_ACTIVE_DATA
) {
826 ret
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
827 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
828 "Invalid state of live stream iterator"
829 "stream-iter-status=%s",
830 lttng_live_stream_state_string(lttng_live_stream
->state
));
834 status
= ctf_msg_iter_get_next_message(
835 lttng_live_stream
->msg_iter
, message
);
837 case CTF_MSG_ITER_STATUS_EOF
:
838 ret
= LTTNG_LIVE_ITERATOR_STATUS_END
;
840 case CTF_MSG_ITER_STATUS_OK
:
841 ret
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
843 case CTF_MSG_ITER_STATUS_AGAIN
:
845 * Continue immediately (end of packet). The next
846 * get_index may return AGAIN to delay the following
849 ret
= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
;
851 case CTF_MSG_ITER_STATUS_INVAL
:
852 /* No argument provided by the user, so don't return INVAL. */
853 case CTF_MSG_ITER_STATUS_ERROR
:
855 ret
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
856 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
857 "CTF message iterator failed to get next message: "
858 "msg-iter=%p, msg-iter-status=%s",
859 lttng_live_stream
->msg_iter
,
860 ctf_msg_iter_status_string(status
));
869 enum lttng_live_iterator_status
lttng_live_iterator_close_stream(
870 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
871 struct lttng_live_stream_iterator
*stream_iter
,
872 bt_message
**curr_msg
)
874 enum lttng_live_iterator_status live_status
=
875 LTTNG_LIVE_ITERATOR_STATUS_OK
;
876 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
877 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
879 * The viewer has hung up on us so we are closing the stream. The
880 * `ctf_msg_iter` should simply realize that it needs to close the
881 * stream properly by emitting the necessary stream end message.
883 enum ctf_msg_iter_status status
= ctf_msg_iter_get_next_message(
884 stream_iter
->msg_iter
, curr_msg
);
886 if (status
== CTF_MSG_ITER_STATUS_ERROR
) {
887 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
888 "Error getting the next message from CTF message iterator");
889 live_status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
893 BT_ASSERT(status
== CTF_MSG_ITER_STATUS_OK
);
901 * handle_no_data_streams()
903 * - for each ACTIVE_NO_DATA stream:
904 * - query relayd for stream data, or quiescence info.
905 * - if need metadata, get metadata, goto retry.
906 * - if new stream, get new stream as ACTIVE_NO_DATA, goto retry
907 * - if quiescent, move to QUIESCENT streams
908 * - if fetched data, move to ACTIVE_DATA streams
909 * (at this point each stream either has data, or is quiescent)
913 * handle_new_streams_and_metadata()
914 * - query relayd for known streams, add them as ACTIVE_NO_DATA
915 * - query relayd for metadata
917 * call handle_active_no_data_streams()
919 * handle_quiescent_streams()
920 * - if at least one stream is ACTIVE_DATA:
921 * - peek stream event with lowest timestamp -> next_ts
922 * - for each quiescent stream
923 * - if next_ts >= quiescent end
924 * - set state to ACTIVE_NO_DATA
926 * - for each quiescent stream
927 * - set state to ACTIVE_NO_DATA
929 * call handle_active_no_data_streams()
931 * handle_active_data_streams()
932 * - if at least one stream is ACTIVE_DATA:
933 * - get stream event with lowest timestamp from heap
934 * - make that stream event the current message.
935 * - move this stream heap position to its next event
936 * - if we need to fetch data from relayd, move
937 * stream to ACTIVE_NO_DATA.
941 * end criterion: ctrl-c on client. If relayd exits or the session
942 * closes on the relay daemon side, we keep on waiting for streams.
943 * Eventually handle --end timestamp (also an end criterion).
945 * When disconnected from relayd: try to re-connect endlessly.
948 enum lttng_live_iterator_status
lttng_live_iterator_next_msg_on_stream(
949 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
950 struct lttng_live_stream_iterator
*stream_iter
,
951 bt_message
**curr_msg
)
953 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
954 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
955 enum lttng_live_iterator_status live_status
;
957 BT_COMP_LOGD("Finding the next message for stream iterator: "
958 "stream-name=\"%s\"", stream_iter
->name
->str
);
960 if (stream_iter
->has_stream_hung_up
) {
962 * The stream has hung up and the stream was properly closed
963 * during the last call to the current function. Return _END
964 * status now so that this stream iterator is removed for the
965 * stream iterator list.
967 live_status
= LTTNG_LIVE_ITERATOR_STATUS_END
;
972 LTTNG_LIVE_LOGD_STREAM_ITER(stream_iter
);
973 live_status
= lttng_live_iterator_handle_new_streams_and_metadata(
974 lttng_live_msg_iter
);
975 if (live_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
978 live_status
= lttng_live_iterator_next_handle_one_no_data_stream(
979 lttng_live_msg_iter
, stream_iter
);
981 if (live_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
982 if (live_status
== LTTNG_LIVE_ITERATOR_STATUS_END
) {
984 * We overwrite `live_status` since `curr_msg` is
985 * likely set to a valid message in this function.
987 live_status
= lttng_live_iterator_close_stream(
988 lttng_live_msg_iter
, stream_iter
, curr_msg
);
992 live_status
= lttng_live_iterator_next_handle_one_quiescent_stream(
993 lttng_live_msg_iter
, stream_iter
, curr_msg
);
994 if (live_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
995 BT_ASSERT(!*curr_msg
);
1001 live_status
= lttng_live_iterator_next_handle_one_active_data_stream(
1002 lttng_live_msg_iter
, stream_iter
, curr_msg
);
1003 if (live_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
1004 BT_ASSERT(!*curr_msg
);
1008 if (live_status
== LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
) {
1016 enum lttng_live_iterator_status
next_stream_iterator_for_trace(
1017 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
1018 struct lttng_live_trace
*live_trace
,
1019 struct lttng_live_stream_iterator
**youngest_trace_stream_iter
)
1021 struct lttng_live_stream_iterator
*youngest_candidate_stream_iter
= NULL
;
1022 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
1023 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
1024 enum lttng_live_iterator_status stream_iter_status
;;
1025 int64_t youngest_candidate_msg_ts
= INT64_MAX
;
1026 uint64_t stream_iter_idx
;
1028 BT_ASSERT_DBG(live_trace
);
1029 BT_ASSERT_DBG(live_trace
->stream_iterators
);
1031 BT_COMP_LOGD("Finding the next stream iterator for trace: "
1032 "trace-id=%"PRIu64
, live_trace
->id
);
1034 * Update the current message of every stream iterators of this trace.
1035 * The current msg of every stream must have a timestamp equal or
1036 * larger than the last message returned by this iterator. We must
1037 * ensure monotonicity.
1039 stream_iter_idx
= 0;
1040 while (stream_iter_idx
< live_trace
->stream_iterators
->len
) {
1041 bool stream_iter_is_ended
= false;
1042 struct lttng_live_stream_iterator
*stream_iter
=
1043 g_ptr_array_index(live_trace
->stream_iterators
,
1047 * Find if there is are now current message for this stream
1050 while (!stream_iter
->current_msg
) {
1051 bt_message
*msg
= NULL
;
1052 int64_t curr_msg_ts_ns
= INT64_MAX
;
1053 stream_iter_status
= lttng_live_iterator_next_msg_on_stream(
1054 lttng_live_msg_iter
, stream_iter
, &msg
);
1056 BT_COMP_LOGD("live stream iterator returned status :%s",
1057 lttng_live_iterator_status_string(stream_iter_status
));
1058 if (stream_iter_status
== LTTNG_LIVE_ITERATOR_STATUS_END
) {
1059 stream_iter_is_ended
= true;
1063 if (stream_iter_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
1070 * Get the timestamp in nanoseconds from origin of this
1073 live_get_msg_ts_ns(stream_iter
, lttng_live_msg_iter
,
1074 msg
, lttng_live_msg_iter
->last_msg_ts_ns
,
1078 * Check if the message of the current live stream
1079 * iterator occured at the exact same time or after the
1080 * last message returned by this component's message
1081 * iterator. If not, we return an error.
1083 if (curr_msg_ts_ns
>= lttng_live_msg_iter
->last_msg_ts_ns
) {
1084 stream_iter
->current_msg
= msg
;
1085 stream_iter
->current_msg_ts_ns
= curr_msg_ts_ns
;
1088 * We received a message in the past. To ensure
1089 * monotonicity, we can't send it forward.
1091 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1092 "Message's timestamp is less than "
1093 "lttng-live's message iterator's last "
1094 "returned timestamp: "
1095 "lttng-live-msg-iter-addr=%p, ts=%" PRId64
", "
1096 "last-msg-ts=%" PRId64
,
1097 lttng_live_msg_iter
, curr_msg_ts_ns
,
1098 lttng_live_msg_iter
->last_msg_ts_ns
);
1099 stream_iter_status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
1104 BT_ASSERT_DBG(stream_iter
!= youngest_candidate_stream_iter
);
1106 if (!stream_iter_is_ended
) {
1107 if (G_UNLIKELY(youngest_candidate_stream_iter
== NULL
) ||
1108 stream_iter
->current_msg_ts_ns
< youngest_candidate_msg_ts
) {
1110 * Update the current best candidate message
1111 * for the stream iterator of this live trace
1112 * to be forwarded downstream.
1114 youngest_candidate_msg_ts
= stream_iter
->current_msg_ts_ns
;
1115 youngest_candidate_stream_iter
= stream_iter
;
1116 } else if (stream_iter
->current_msg_ts_ns
== youngest_candidate_msg_ts
) {
1118 * Order the messages in an arbitrary but
1119 * deterministic way.
1121 BT_ASSERT_DBG(stream_iter
!= youngest_candidate_stream_iter
);
1122 int ret
= common_muxing_compare_messages(
1123 stream_iter
->current_msg
,
1124 youngest_candidate_stream_iter
->current_msg
);
1127 * The `youngest_candidate_stream_iter->current_msg`
1128 * should go first. Update the next
1129 * iterator and the current timestamp.
1131 youngest_candidate_msg_ts
= stream_iter
->current_msg_ts_ns
;
1132 youngest_candidate_stream_iter
= stream_iter
;
1133 } else if (ret
== 0) {
1135 * Unable to pick which one should go
1138 BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1139 "stream-iter-addr=%p"
1140 "stream-iter-addr=%p",
1142 youngest_candidate_stream_iter
);
1149 * The live stream iterator has ended. That
1150 * iterator is removed from the array, but
1151 * there is no need to increment
1152 * stream_iter_idx as
1153 * g_ptr_array_remove_index_fast replaces the
1154 * removed element with the array's last
1157 g_ptr_array_remove_index_fast(
1158 live_trace
->stream_iterators
,
1163 if (youngest_candidate_stream_iter
) {
1164 *youngest_trace_stream_iter
= youngest_candidate_stream_iter
;
1165 stream_iter_status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
1168 * The only case where we don't have a candidate for this trace
1169 * is if we reached the end of all the iterators.
1171 BT_ASSERT(live_trace
->stream_iterators
->len
== 0);
1172 stream_iter_status
= LTTNG_LIVE_ITERATOR_STATUS_END
;
1176 return stream_iter_status
;
1180 enum lttng_live_iterator_status
next_stream_iterator_for_session(
1181 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
1182 struct lttng_live_session
*session
,
1183 struct lttng_live_stream_iterator
**youngest_session_stream_iter
)
1185 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
1186 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
1187 enum lttng_live_iterator_status stream_iter_status
;
1188 uint64_t trace_idx
= 0;
1189 int64_t youngest_candidate_msg_ts
= INT64_MAX
;
1190 struct lttng_live_stream_iterator
*youngest_candidate_stream_iter
= NULL
;
1192 BT_COMP_LOGD("Finding the next stream iterator for session: "
1193 "session-id=%"PRIu64
, session
->id
);
1195 * Make sure we are attached to the session and look for new streams
1198 stream_iter_status
= lttng_live_get_session(lttng_live_msg_iter
, session
);
1199 if (stream_iter_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
&&
1200 stream_iter_status
!= LTTNG_LIVE_ITERATOR_STATUS_CONTINUE
&&
1201 stream_iter_status
!= LTTNG_LIVE_ITERATOR_STATUS_END
) {
1205 BT_ASSERT_DBG(session
->traces
);
1207 while (trace_idx
< session
->traces
->len
) {
1208 bool trace_is_ended
= false;
1209 struct lttng_live_stream_iterator
*stream_iter
;
1210 struct lttng_live_trace
*trace
=
1211 g_ptr_array_index(session
->traces
, trace_idx
);
1213 stream_iter_status
= next_stream_iterator_for_trace(
1214 lttng_live_msg_iter
, trace
, &stream_iter
);
1215 if (stream_iter_status
== LTTNG_LIVE_ITERATOR_STATUS_END
) {
1217 * All the live stream iterators for this trace are
1218 * ENDed. Remove the trace from this session.
1220 trace_is_ended
= true;
1221 } else if (stream_iter_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
1225 if (!trace_is_ended
) {
1226 BT_ASSERT_DBG(stream_iter
);
1228 if (G_UNLIKELY(youngest_candidate_stream_iter
== NULL
) ||
1229 stream_iter
->current_msg_ts_ns
< youngest_candidate_msg_ts
) {
1230 youngest_candidate_msg_ts
= stream_iter
->current_msg_ts_ns
;
1231 youngest_candidate_stream_iter
= stream_iter
;
1232 } else if (stream_iter
->current_msg_ts_ns
== youngest_candidate_msg_ts
) {
1234 * Order the messages in an arbitrary but
1235 * deterministic way.
1237 int ret
= common_muxing_compare_messages(
1238 stream_iter
->current_msg
,
1239 youngest_candidate_stream_iter
->current_msg
);
1242 * The `youngest_candidate_stream_iter->current_msg`
1243 * should go first. Update the next iterator
1244 * and the current timestamp.
1246 youngest_candidate_msg_ts
= stream_iter
->current_msg_ts_ns
;
1247 youngest_candidate_stream_iter
= stream_iter
;
1248 } else if (ret
== 0) {
1249 /* Unable to pick which one should go first. */
1250 BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1251 "stream-iter-addr=%p" "stream-iter-addr=%p",
1252 stream_iter
, youngest_candidate_stream_iter
);
1258 * trace_idx is not incremented since
1259 * g_ptr_array_remove_index_fast replaces the
1260 * element at trace_idx with the array's last element.
1262 g_ptr_array_remove_index_fast(session
->traces
,
1266 if (youngest_candidate_stream_iter
) {
1267 *youngest_session_stream_iter
= youngest_candidate_stream_iter
;
1268 stream_iter_status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
1271 * The only cases where we don't have a candidate for this
1273 * 1. if we reached the end of all the iterators of all the
1274 * traces of this session,
1275 * 2. if we never had live stream iterator in the first place.
1277 * In either cases, we return END.
1279 BT_ASSERT(session
->traces
->len
== 0);
1280 stream_iter_status
= LTTNG_LIVE_ITERATOR_STATUS_END
;
1283 return stream_iter_status
;
1287 void put_messages(bt_message_array_const msgs
, uint64_t count
)
1291 for (i
= 0; i
< count
; i
++) {
1292 BT_MESSAGE_PUT_REF_AND_RESET(msgs
[i
]);
1297 bt_component_class_message_iterator_next_method_status
lttng_live_msg_iter_next(
1298 bt_self_message_iterator
*self_msg_it
,
1299 bt_message_array_const msgs
, uint64_t capacity
,
1302 bt_component_class_message_iterator_next_method_status status
;
1303 enum lttng_live_viewer_status viewer_status
;
1304 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
1305 bt_self_message_iterator_get_data(self_msg_it
);
1306 struct lttng_live_component
*lttng_live
=
1307 lttng_live_msg_iter
->lttng_live_comp
;
1308 bt_self_component
*self_comp
= lttng_live_msg_iter
->self_comp
;
1309 bt_logging_level log_level
= lttng_live_msg_iter
->log_level
;
1310 enum lttng_live_iterator_status stream_iter_status
;
1311 uint64_t session_idx
;
1315 BT_ASSERT_DBG(lttng_live_msg_iter
);
1317 if (G_UNLIKELY(lttng_live_msg_iter
->was_interrupted
)) {
1319 * The iterator was interrupted in a previous call to the
1320 * `_next()` method. We currently do not support generating
1321 * messages after such event. The babeltrace2 CLI should never
1322 * be running the graph after being interrupted. So this check
1323 * is to prevent other graph users from using this live
1324 * iterator in an messed up internal state.
1326 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1327 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1328 "Message iterator was interrupted during a previous call to the `next()` and currently does not support continuing after such event.");
1333 * Clear all the invalid message reference that might be left over in
1336 memset(msgs
, 0, capacity
* sizeof(*msgs
));
1339 * If no session are exposed on the relay found at the url provided by
1340 * the user, session count will be 0. In this case, we return status
1341 * end to return gracefully.
1343 if (lttng_live_msg_iter
->sessions
->len
== 0) {
1344 if (lttng_live
->params
.sess_not_found_act
!=
1345 SESSION_NOT_FOUND_ACTION_CONTINUE
) {
1346 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1350 * The are no more active session for this session
1351 * name. Retry to create a viewer session for the
1352 * requested session name.
1354 viewer_status
= lttng_live_create_viewer_session(lttng_live_msg_iter
);
1355 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1356 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
1357 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1358 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1359 "Error creating LTTng live viewer session");
1360 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
1361 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN
;
1370 if (lttng_live_msg_iter
->active_stream_iter
== 0) {
1371 lttng_live_force_new_streams_and_metadata(lttng_live_msg_iter
);
1375 * Here the muxing of message is done.
1377 * We need to iterate over all the streams of all the traces of all the
1378 * viewer sessions in order to get the message with the smallest
1379 * timestamp. In this case, a session is a viewer session and there is
1380 * one viewer session per consumer daemon. (UST 32bit, UST 64bit and/or
1381 * kernel). Each viewer session can have multiple traces, for example,
1382 * 64bit UST viewer sessions could have multiple per-pid traces.
1384 * We iterate over the streams of each traces to update and see what is
1385 * their next message's timestamp. From those timestamps, we select the
1386 * message with the smallest timestamp as the best candidate message
1387 * for that trace and do the same thing across all the sessions.
1389 * We then compare the timestamp of best candidate message of all the
1390 * sessions to pick the message with the smallest timestamp and we
1393 while (*count
< capacity
) {
1394 struct lttng_live_stream_iterator
*youngest_stream_iter
= NULL
,
1395 *candidate_stream_iter
= NULL
;
1396 int64_t youngest_msg_ts_ns
= INT64_MAX
;
1398 BT_ASSERT_DBG(lttng_live_msg_iter
->sessions
);
1400 while (session_idx
< lttng_live_msg_iter
->sessions
->len
) {
1401 struct lttng_live_session
*session
=
1402 g_ptr_array_index(lttng_live_msg_iter
->sessions
,
1405 /* Find the best candidate message to send downstream. */
1406 stream_iter_status
= next_stream_iterator_for_session(
1407 lttng_live_msg_iter
, session
,
1408 &candidate_stream_iter
);
1410 /* If we receive an END status, it means that either:
1411 * - Those traces never had active streams (UST with no
1412 * data produced yet),
1413 * - All live stream iterators have ENDed.*/
1414 if (stream_iter_status
== LTTNG_LIVE_ITERATOR_STATUS_END
) {
1415 if (session
->closed
&& session
->traces
->len
== 0) {
1417 * Remove the session from the list.
1418 * session_idx is not modified since
1419 * g_ptr_array_remove_index_fast
1420 * replaces the the removed element with
1421 * the array's last element.
1423 g_ptr_array_remove_index_fast(
1424 lttng_live_msg_iter
->sessions
,
1432 if (stream_iter_status
!= LTTNG_LIVE_ITERATOR_STATUS_OK
) {
1436 if (G_UNLIKELY(youngest_stream_iter
== NULL
) ||
1437 candidate_stream_iter
->current_msg_ts_ns
< youngest_msg_ts_ns
) {
1438 youngest_msg_ts_ns
= candidate_stream_iter
->current_msg_ts_ns
;
1439 youngest_stream_iter
= candidate_stream_iter
;
1440 } else if (candidate_stream_iter
->current_msg_ts_ns
== youngest_msg_ts_ns
) {
1442 * The currently selected message to be sent
1443 * downstream next has the exact same timestamp
1444 * that of the current candidate message. We
1445 * must break the tie in a predictable manner.
1447 BT_COMP_LOGD_STR("Two of the next message candidates have the same timestamps, pick one deterministically.");
1449 * Order the messages in an arbitrary but
1450 * deterministic way.
1452 int ret
= common_muxing_compare_messages(
1453 candidate_stream_iter
->current_msg
,
1454 youngest_stream_iter
->current_msg
);
1457 * The `candidate_stream_iter->current_msg`
1458 * should go first. Update the next
1459 * iterator and the current timestamp.
1461 youngest_msg_ts_ns
= candidate_stream_iter
->current_msg_ts_ns
;
1462 youngest_stream_iter
= candidate_stream_iter
;
1463 } else if (ret
== 0) {
1464 /* Unable to pick which one should go first. */
1465 BT_COMP_LOGW("Cannot deterministically pick next live stream message iterator because they have identical next messages: "
1466 "next-stream-iter-addr=%p" "candidate-stream-iter-addr=%p",
1467 youngest_stream_iter
, candidate_stream_iter
);
1474 if (!youngest_stream_iter
) {
1475 stream_iter_status
= LTTNG_LIVE_ITERATOR_STATUS_AGAIN
;
1479 BT_ASSERT_DBG(youngest_stream_iter
->current_msg
);
1480 /* Ensure monotonicity. */
1481 BT_ASSERT_DBG(lttng_live_msg_iter
->last_msg_ts_ns
<=
1482 youngest_stream_iter
->current_msg_ts_ns
);
1485 * Insert the next message to the message batch. This will set
1486 * stream iterator current messsage to NULL so that next time
1487 * we fetch the next message of that stream iterator
1489 BT_MESSAGE_MOVE_REF(msgs
[*count
], youngest_stream_iter
->current_msg
);
1492 /* Update the last timestamp in nanoseconds sent downstream. */
1493 lttng_live_msg_iter
->last_msg_ts_ns
= youngest_msg_ts_ns
;
1494 youngest_stream_iter
->current_msg_ts_ns
= INT64_MAX
;
1496 stream_iter_status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
1500 switch (stream_iter_status
) {
1501 case LTTNG_LIVE_ITERATOR_STATUS_OK
:
1502 case LTTNG_LIVE_ITERATOR_STATUS_AGAIN
:
1504 * If we gathered messages, return _OK even if the graph was
1505 * interrupted. This allows for the components downstream to at
1506 * least get the thoses messages. If the graph was indeed
1507 * interrupted there should not be another _next() call as the
1508 * application will tear down the graph. This component class
1509 * doesn't support restarting after an interruption.
1512 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK
;
1514 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN
;
1517 case LTTNG_LIVE_ITERATOR_STATUS_END
:
1518 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END
;
1520 case LTTNG_LIVE_ITERATOR_STATUS_NOMEM
:
1521 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1522 "Memory error preparing the next batch of messages: "
1523 "live-iter-status=%s",
1524 lttng_live_iterator_status_string(stream_iter_status
));
1525 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR
;
1527 case LTTNG_LIVE_ITERATOR_STATUS_ERROR
:
1528 case LTTNG_LIVE_ITERATOR_STATUS_INVAL
:
1529 case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED
:
1530 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1531 "Error preparing the next batch of messages: "
1532 "live-iter-status=%s",
1533 lttng_live_iterator_status_string(stream_iter_status
));
1535 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR
;
1536 /* Put all existing messages on error. */
1537 put_messages(msgs
, *count
);
1548 struct lttng_live_msg_iter
*lttng_live_msg_iter_create(
1549 struct lttng_live_component
*lttng_live_comp
,
1550 bt_self_message_iterator
*self_msg_it
)
1552 bt_self_component
*self_comp
= lttng_live_comp
->self_comp
;
1553 bt_logging_level log_level
= lttng_live_comp
->log_level
;
1555 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
1556 g_new0(struct lttng_live_msg_iter
, 1);
1557 if (!lttng_live_msg_iter
) {
1558 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1559 "Failed to allocate lttng_live_msg_iter");
1563 lttng_live_msg_iter
->log_level
= lttng_live_comp
->log_level
;
1564 lttng_live_msg_iter
->self_comp
= lttng_live_comp
->self_comp
;
1565 lttng_live_msg_iter
->lttng_live_comp
= lttng_live_comp
;
1566 lttng_live_msg_iter
->self_msg_iter
= self_msg_it
;
1568 lttng_live_msg_iter
->active_stream_iter
= 0;
1569 lttng_live_msg_iter
->last_msg_ts_ns
= INT64_MIN
;
1570 lttng_live_msg_iter
->was_interrupted
= false;
1572 lttng_live_msg_iter
->sessions
= g_ptr_array_new_with_free_func(
1573 (GDestroyNotify
) lttng_live_destroy_session
);
1574 BT_ASSERT(lttng_live_msg_iter
->sessions
);
1577 return lttng_live_msg_iter
;
1582 bt_component_class_message_iterator_initialize_method_status
lttng_live_msg_iter_init(
1583 bt_self_message_iterator
*self_msg_it
,
1584 bt_self_message_iterator_configuration
*config
,
1585 bt_self_component_source
*self_comp_src
,
1586 bt_self_component_port_output
*self_port
)
1588 bt_component_class_message_iterator_initialize_method_status status
;
1589 bt_self_component
*self_comp
=
1590 bt_self_component_source_as_self_component(self_comp_src
);
1591 struct lttng_live_component
*lttng_live
;
1592 struct lttng_live_msg_iter
*lttng_live_msg_iter
;
1593 enum lttng_live_viewer_status viewer_status
;
1594 bt_logging_level log_level
;
1596 BT_ASSERT(self_msg_it
);
1598 lttng_live
= bt_self_component_get_data(self_comp
);
1599 log_level
= lttng_live
->log_level
;
1600 self_comp
= lttng_live
->self_comp
;
1603 /* There can be only one downstream iterator at the same time. */
1604 BT_ASSERT(!lttng_live
->has_msg_iter
);
1605 lttng_live
->has_msg_iter
= true;
1607 lttng_live_msg_iter
= lttng_live_msg_iter_create(lttng_live
,
1609 if (!lttng_live_msg_iter
) {
1610 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1611 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1612 "Failed to create lttng_live_msg_iter");
1616 viewer_status
= live_viewer_connection_create(self_comp
, NULL
,
1617 log_level
, lttng_live
->params
.url
->str
, false,
1618 lttng_live_msg_iter
, <tng_live_msg_iter
->viewer_connection
);
1619 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1620 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
1621 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1622 "Failed to create viewer connection");
1623 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
1625 * Interruption in the _iter_init() method is not
1626 * supported. Return an error.
1628 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1629 "Interrupted while creating viewer connection");
1634 viewer_status
= lttng_live_create_viewer_session(lttng_live_msg_iter
);
1635 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1636 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
1637 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1638 "Failed to create viewer session");
1639 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
1641 * Interruption in the _iter_init() method is not
1642 * supported. Return an error.
1644 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1645 "Interrupted when creating viewer session");
1650 if (lttng_live_msg_iter
->sessions
->len
== 0) {
1651 switch (lttng_live
->params
.sess_not_found_act
) {
1652 case SESSION_NOT_FOUND_ACTION_CONTINUE
:
1653 BT_COMP_LOGI("Unable to connect to the requested live viewer session. Keep trying to connect because of "
1654 "%s=\"%s\" component parameter: url=\"%s\"",
1655 SESS_NOT_FOUND_ACTION_PARAM
,
1656 SESS_NOT_FOUND_ACTION_CONTINUE_STR
,
1657 lttng_live
->params
.url
->str
);
1659 case SESSION_NOT_FOUND_ACTION_FAIL
:
1660 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1661 "Unable to connect to the requested live viewer session. Fail the message iterator initialization because of %s=\"%s\" "
1662 "component parameter: url =\"%s\"",
1663 SESS_NOT_FOUND_ACTION_PARAM
,
1664 SESS_NOT_FOUND_ACTION_FAIL_STR
,
1665 lttng_live
->params
.url
->str
);
1667 case SESSION_NOT_FOUND_ACTION_END
:
1668 BT_COMP_LOGI("Unable to connect to the requested live viewer session. End gracefully at the first _next() "
1669 "call because of %s=\"%s\" component parameter: "
1670 "url=\"%s\"", SESS_NOT_FOUND_ACTION_PARAM
,
1671 SESS_NOT_FOUND_ACTION_END_STR
,
1672 lttng_live
->params
.url
->str
);
1679 bt_self_message_iterator_set_data(self_msg_it
, lttng_live_msg_iter
);
1680 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK
;
1684 status
= BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR
;
1685 lttng_live_msg_iter_destroy(lttng_live_msg_iter
);
1690 static struct bt_param_validation_map_value_entry_descr list_sessions_params
[] = {
1691 { URL_PARAM
, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY
, { .type
= BT_VALUE_TYPE_STRING
} },
1692 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
1696 bt_component_class_query_method_status
lttng_live_query_list_sessions(
1697 const bt_value
*params
, const bt_value
**result
,
1698 bt_self_component_class
*self_comp_class
,
1699 bt_logging_level log_level
)
1701 bt_component_class_query_method_status status
;
1702 const bt_value
*url_value
= NULL
;
1704 struct live_viewer_connection
*viewer_connection
= NULL
;
1705 enum lttng_live_viewer_status viewer_status
;
1706 enum bt_param_validation_status validation_status
;
1707 gchar
*validate_error
= NULL
;
1709 validation_status
= bt_param_validation_validate(params
,
1710 list_sessions_params
, &validate_error
);
1711 if (validation_status
== BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR
) {
1712 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR
;
1714 } else if (validation_status
== BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR
) {
1715 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
1716 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
, "%s",
1721 url_value
= bt_value_map_borrow_entry_value_const(params
, URL_PARAM
);
1722 url
= bt_value_string_get(url_value
);
1724 viewer_status
= live_viewer_connection_create(NULL
, self_comp_class
,
1725 log_level
, url
, true, NULL
, &viewer_connection
);
1726 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1727 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
1728 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
1729 "Failed to create viewer connection");
1730 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
1731 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
1732 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN
;
1739 status
= live_viewer_connection_list_sessions(viewer_connection
,
1741 if (status
!= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK
) {
1742 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
1743 "Failed to list viewer sessions");
1750 BT_VALUE_PUT_REF_AND_RESET(*result
);
1753 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
1757 if (viewer_connection
) {
1758 live_viewer_connection_destroy(viewer_connection
);
1761 g_free(validate_error
);
1767 bt_component_class_query_method_status
lttng_live_query_support_info(
1768 const bt_value
*params
, const bt_value
**result
,
1769 bt_self_component_class
*self_comp_class
,
1770 bt_logging_level log_level
)
1772 bt_component_class_query_method_status status
=
1773 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK
;
1774 const bt_value
*input_type_value
;
1775 const bt_value
*input_value
;
1777 struct bt_common_lttng_live_url_parts parts
= { 0 };
1779 /* Used by the logging macros */
1780 __attribute__((unused
)) bt_self_component
*self_comp
= NULL
;
1783 input_type_value
= bt_value_map_borrow_entry_value_const(params
,
1785 if (!input_type_value
) {
1786 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
1787 "Missing expected `type` parameter.");
1791 if (!bt_value_is_string(input_type_value
)) {
1792 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
1793 "`type` parameter is not a string value.");
1797 if (strcmp(bt_value_string_get(input_type_value
), "string") != 0) {
1798 /* We don't handle file system paths */
1802 input_value
= bt_value_map_borrow_entry_value_const(params
, "input");
1804 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
1805 "Missing expected `input` parameter.");
1809 if (!bt_value_is_string(input_value
)) {
1810 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
1811 "`input` parameter is not a string value.");
1815 parts
= bt_common_parse_lttng_live_url(bt_value_string_get(input_value
),
1817 if (parts
.session_name
) {
1819 * Looks pretty much like an LTTng live URL: we got the
1820 * session name part, which forms a complete URL.
1826 *result
= bt_value_real_create_init(weight
);
1828 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR
;
1836 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
1839 BT_ASSERT(!*result
);
1842 bt_common_destroy_lttng_live_url_parts(&parts
);
1847 bt_component_class_query_method_status
lttng_live_query(
1848 bt_self_component_class_source
*comp_class
,
1849 bt_private_query_executor
*priv_query_exec
,
1850 const char *object
, const bt_value
*params
,
1851 __attribute__((unused
)) void *method_data
,
1852 const bt_value
**result
)
1854 bt_component_class_query_method_status status
=
1855 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK
;
1856 bt_self_component
*self_comp
= NULL
;
1857 bt_self_component_class
*self_comp_class
=
1858 bt_self_component_class_source_as_self_component_class(comp_class
);
1859 bt_logging_level log_level
= bt_query_executor_get_logging_level(
1860 bt_private_query_executor_as_query_executor_const(
1863 if (strcmp(object
, "sessions") == 0) {
1864 status
= lttng_live_query_list_sessions(params
, result
,
1865 self_comp_class
, log_level
);
1866 } else if (strcmp(object
, "babeltrace.support-info") == 0) {
1867 status
= lttng_live_query_support_info(params
, result
,
1868 self_comp_class
, log_level
);
1870 BT_COMP_LOGI("Unknown query object `%s`", object
);
1871 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_UNKNOWN_OBJECT
;
1880 void lttng_live_component_destroy_data(struct lttng_live_component
*lttng_live
)
1885 if (lttng_live
->params
.url
) {
1886 g_string_free(lttng_live
->params
.url
, TRUE
);
1892 void lttng_live_component_finalize(bt_self_component_source
*component
)
1894 void *data
= bt_self_component_get_data(
1895 bt_self_component_source_as_self_component(component
));
1900 lttng_live_component_destroy_data(data
);
1904 enum session_not_found_action
parse_session_not_found_action_param(
1905 const bt_value
*no_session_param
)
1907 enum session_not_found_action action
;
1908 const char *no_session_act_str
= bt_value_string_get(no_session_param
);
1910 if (strcmp(no_session_act_str
, SESS_NOT_FOUND_ACTION_CONTINUE_STR
) == 0) {
1911 action
= SESSION_NOT_FOUND_ACTION_CONTINUE
;
1912 } else if (strcmp(no_session_act_str
, SESS_NOT_FOUND_ACTION_FAIL_STR
) == 0) {
1913 action
= SESSION_NOT_FOUND_ACTION_FAIL
;
1915 BT_ASSERT(strcmp(no_session_act_str
, SESS_NOT_FOUND_ACTION_END_STR
) == 0);
1916 action
= SESSION_NOT_FOUND_ACTION_END
;
1922 static struct bt_param_validation_value_descr inputs_elem_descr
= {
1923 .type
= BT_VALUE_TYPE_STRING
,
1926 static const char *sess_not_found_action_choices
[] = {
1927 SESS_NOT_FOUND_ACTION_CONTINUE_STR
,
1928 SESS_NOT_FOUND_ACTION_FAIL_STR
,
1929 SESS_NOT_FOUND_ACTION_END_STR
,
1932 static struct bt_param_validation_map_value_entry_descr params_descr
[] = {
1933 { INPUTS_PARAM
, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_MANDATORY
, { BT_VALUE_TYPE_ARRAY
, .array
= {
1936 .element_type
= &inputs_elem_descr
,
1938 { SESS_NOT_FOUND_ACTION_PARAM
, BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_OPTIONAL
, { BT_VALUE_TYPE_STRING
, .string
= {
1939 .choices
= sess_not_found_action_choices
,
1941 BT_PARAM_VALIDATION_MAP_VALUE_ENTRY_END
1945 bt_component_class_initialize_method_status
lttng_live_component_create(
1946 const bt_value
*params
,
1947 bt_logging_level log_level
,
1948 bt_self_component
*self_comp
,
1949 struct lttng_live_component
**component
)
1951 struct lttng_live_component
*lttng_live
= NULL
;
1952 const bt_value
*inputs_value
;
1953 const bt_value
*url_value
;
1954 const bt_value
*value
;
1956 enum bt_param_validation_status validation_status
;
1957 gchar
*validation_error
= NULL
;
1958 bt_component_class_initialize_method_status status
;
1960 validation_status
= bt_param_validation_validate(params
, params_descr
,
1962 if (validation_status
== BT_PARAM_VALIDATION_STATUS_MEMORY_ERROR
) {
1963 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1965 } else if (validation_status
== BT_PARAM_VALIDATION_STATUS_VALIDATION_ERROR
) {
1966 BT_COMP_LOGE_APPEND_CAUSE(self_comp
, "%s", validation_error
);
1967 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_ERROR
;
1971 lttng_live
= g_new0(struct lttng_live_component
, 1);
1973 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1976 lttng_live
->log_level
= log_level
;
1977 lttng_live
->self_comp
= self_comp
;
1978 lttng_live
->max_query_size
= MAX_QUERY_SIZE
;
1979 lttng_live
->has_msg_iter
= false;
1982 bt_value_map_borrow_entry_value_const(params
, INPUTS_PARAM
);
1984 bt_value_array_borrow_element_by_index_const(inputs_value
, 0);
1985 url
= bt_value_string_get(url_value
);
1987 lttng_live
->params
.url
= g_string_new(url
);
1988 if (!lttng_live
->params
.url
) {
1989 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR
;
1993 value
= bt_value_map_borrow_entry_value_const(params
,
1994 SESS_NOT_FOUND_ACTION_PARAM
);
1996 lttng_live
->params
.sess_not_found_act
=
1997 parse_session_not_found_action_param(value
);
1999 BT_COMP_LOGI("Optional `%s` parameter is missing: "
2000 "defaulting to `%s`.",
2001 SESS_NOT_FOUND_ACTION_PARAM
,
2002 SESS_NOT_FOUND_ACTION_CONTINUE_STR
);
2003 lttng_live
->params
.sess_not_found_act
=
2004 SESSION_NOT_FOUND_ACTION_CONTINUE
;
2007 status
= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
;
2011 lttng_live_component_destroy_data(lttng_live
);
2014 g_free(validation_error
);
2016 *component
= lttng_live
;
2021 bt_component_class_initialize_method_status
lttng_live_component_init(
2022 bt_self_component_source
*self_comp_src
,
2023 bt_self_component_source_configuration
*config
,
2024 const bt_value
*params
,
2025 __attribute__((unused
)) void *init_method_data
)
2027 struct lttng_live_component
*lttng_live
;
2028 bt_component_class_initialize_method_status ret
;
2029 bt_self_component
*self_comp
=
2030 bt_self_component_source_as_self_component(self_comp_src
);
2031 bt_logging_level log_level
= bt_component_get_logging_level(
2032 bt_self_component_as_component(self_comp
));
2033 bt_self_component_add_port_status add_port_status
;
2035 ret
= lttng_live_component_create(params
, log_level
, self_comp
, <tng_live
);
2036 if (ret
!= BT_COMPONENT_CLASS_INITIALIZE_METHOD_STATUS_OK
) {
2040 add_port_status
= bt_self_component_source_add_output_port(
2041 self_comp_src
, "out", NULL
, NULL
);
2042 if (add_port_status
!= BT_SELF_COMPONENT_ADD_PORT_STATUS_OK
) {
2043 ret
= (int) add_port_status
;
2047 bt_self_component_set_data(self_comp
, lttng_live
);
2051 lttng_live_component_destroy_data(lttng_live
);