2 * Copyright (C) 2020 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 * SPDX-License-Identifier: GPL-2.0-only
8 #include "action-executor.hpp"
10 #include "health-sessiond.hpp"
11 #include "lttng-sessiond.hpp"
12 #include "notification-thread-internal.hpp"
13 #include "session.hpp"
16 #include <common/dynamic-array.hpp>
17 #include <common/macros.hpp>
18 #include <common/optional.hpp>
19 #include <common/urcu.hpp>
21 #include <lttng/action/action-internal.hpp>
22 #include <lttng/action/list-internal.hpp>
23 #include <lttng/action/list.h>
24 #include <lttng/action/notify-internal.hpp>
25 #include <lttng/action/notify.h>
26 #include <lttng/action/rotate-session.h>
27 #include <lttng/action/snapshot-session.h>
28 #include <lttng/action/start-session.h>
29 #include <lttng/action/stop-session.h>
30 #include <lttng/condition/evaluation.h>
31 #include <lttng/condition/event-rule-matches-internal.hpp>
32 #include <lttng/lttng-error.h>
33 #include <lttng/trigger/trigger-internal.hpp>
38 #include <urcu/list.h>
40 #define THREAD_NAME "Action Executor"
41 #define MAX_QUEUED_WORK_COUNT 8192
43 struct action_executor
{
44 struct lttng_thread
*thread
;
45 struct notification_thread_handle
*notification_thread_handle
;
47 uint64_t pending_count
;
48 struct cds_list_head list
;
53 uint64_t next_work_item_id
;
58 * A work item is composed of a dynamic array of sub-items which
59 * represent a flattened, and augmented, version of a trigger's actions.
61 * We cannot rely solely on the trigger's actions since each action can have an
62 * execution context we need to comply with.
64 * The notion of execution context is required since for some actions the
65 * associated object are referenced by name and not by id. This can lead to
66 * a number of ambiguities when executing an action work item.
68 * For example, let's take a simple trigger such as:
69 * - condition: ust event a
70 * - action: start session S
72 * At time T, session S exists.
73 * At T + 1, the event A is hit.
74 * At T + 2, the tracer event notification is received and the work item is
75 * queued. Here session S have an id of 1.
76 * At T + 3, the session S is destroyed and a new session S is created, with a
77 * resulting id of 200.
78 * At T +4, the work item is popped from the queue and begin execution and will
79 * start session S with an id of 200 instead of the session S id 1 that was
80 * present at the queuing phase.
82 * The context to be respected is the one when the work item is queued. If the
83 * execution context is not the same at the moment of execution, we skip the
84 * execution of that sub-item.
86 * It is the same policy in regards to the validity of the associated
87 * trigger object at the moment of execution, if the trigger is found to be
88 * unregistered, the execution is skipped.
90 struct action_work_item
{
94 * The actions to be executed with their respective execution context.
95 * See struct `action_work_subitem`.
97 struct lttng_dynamic_array subitems
;
99 /* Execution context data */
100 struct lttng_trigger
*trigger
;
101 struct lttng_evaluation
*evaluation
;
102 struct notification_client_list
*client_list
;
103 LTTNG_OPTIONAL(struct lttng_credentials
) object_creds
;
104 struct cds_list_head list_node
;
107 struct action_work_subitem
{
108 struct lttng_action
*action
;
110 /* Used by actions targeting a session. */
111 LTTNG_OPTIONAL(uint64_t) session_id
;
117 * Only return non-zero on a fatal error that should shut down the action
120 using action_executor_handler
= int (*)(struct action_executor
*,
121 const struct action_work_item
*,
122 struct action_work_subitem
*);
124 static int action_executor_notify_handler(struct action_executor
*executor
,
125 const struct action_work_item
*,
126 struct action_work_subitem
*);
127 static int action_executor_start_session_handler(struct action_executor
*executor
,
128 const struct action_work_item
*,
129 struct action_work_subitem
*);
130 static int action_executor_stop_session_handler(struct action_executor
*executor
,
131 const struct action_work_item
*,
132 struct action_work_subitem
*);
133 static int action_executor_rotate_session_handler(struct action_executor
*executor
,
134 const struct action_work_item
*,
135 struct action_work_subitem
*);
136 static int action_executor_snapshot_session_handler(struct action_executor
*executor
,
137 const struct action_work_item
*,
138 struct action_work_subitem
*);
139 static int action_executor_list_handler(struct action_executor
*executor
,
140 const struct action_work_item
*,
141 struct action_work_subitem
*);
142 static int action_executor_generic_handler(struct action_executor
*executor
,
143 const struct action_work_item
*,
144 struct action_work_subitem
*);
146 static const action_executor_handler action_executors
[] = {
147 action_executor_notify_handler
, action_executor_start_session_handler
,
148 action_executor_stop_session_handler
, action_executor_rotate_session_handler
,
149 action_executor_snapshot_session_handler
, action_executor_list_handler
,
152 /* Forward declaration */
153 static int add_action_to_subitem_array(struct lttng_action
*action
,
154 struct lttng_dynamic_array
*subitems
);
156 static int populate_subitem_array_from_trigger(struct lttng_trigger
*trigger
,
157 struct lttng_dynamic_array
*subitems
);
159 static void action_work_subitem_destructor(void *element
)
161 struct action_work_subitem
*subitem
= (action_work_subitem
*) element
;
163 lttng_action_put(subitem
->action
);
166 static const char *get_action_name(const struct lttng_action
*action
)
168 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
170 LTTNG_ASSERT(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
172 return lttng_action_type_string(action_type
);
175 /* Check if this trigger allowed to interect with a given session. */
176 static bool is_trigger_allowed_for_session(const struct lttng_trigger
*trigger
,
177 struct ltt_session
*session
)
179 bool is_allowed
= false;
180 const struct lttng_credentials session_creds
= {
181 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
->uid
),
182 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
->gid
),
184 /* Can never be NULL. */
185 const struct lttng_credentials
*trigger_creds
= lttng_trigger_get_credentials(trigger
);
187 is_allowed
= (lttng_credentials_is_equal_uid(trigger_creds
, &session_creds
)) ||
188 (lttng_credentials_get_uid(trigger_creds
) == 0);
190 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
192 (long int) session
->uid
,
193 (long int) session
->gid
,
194 (long int) lttng_credentials_get_uid(trigger_creds
));
200 static const char *get_trigger_name(const struct lttng_trigger
*trigger
)
202 const char *trigger_name
;
203 enum lttng_trigger_status trigger_status
;
205 trigger_status
= lttng_trigger_get_name(trigger
, &trigger_name
);
206 switch (trigger_status
) {
207 case LTTNG_TRIGGER_STATUS_OK
:
209 case LTTNG_TRIGGER_STATUS_UNSET
:
210 trigger_name
= "(anonymous)";
213 trigger_name
= "(failed to get name)";
220 static int client_handle_transmission_status(struct notification_client
*client
,
221 enum client_transmission_status status
,
225 struct action_executor
*executor
= (action_executor
*) user_data
;
226 bool update_communication
= true;
229 case CLIENT_TRANSMISSION_STATUS_COMPLETE
:
230 DBG("Successfully sent full notification to client, client_id = %" PRIu64
,
233 * There is no need to wake the (e)poll thread. If it was waiting for
234 * "out" events on the client's socket, it will see that no payload
235 * in queued and will unsubscribe from that event.
237 * In the other cases, we have to wake the the (e)poll thread to either
238 * handle the error on the client or to get it to monitor the client "out"
241 update_communication
= false;
243 case CLIENT_TRANSMISSION_STATUS_QUEUED
:
244 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64
,
247 case CLIENT_TRANSMISSION_STATUS_FAIL
:
248 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64
,
252 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64
,
258 if (!update_communication
) {
262 /* Safe to read client's id without locking as it is immutable. */
263 ret
= notification_thread_client_communication_update(
264 executor
->notification_thread_handle
, client
->id
, status
);
269 static int action_executor_notify_handler(struct action_executor
*executor
,
270 const struct action_work_item
*work_item
,
271 struct action_work_subitem
*item
__attribute__((unused
)))
273 return notification_client_list_send_evaluation(
274 work_item
->client_list
,
276 work_item
->evaluation
,
277 work_item
->object_creds
.is_set
? &(work_item
->object_creds
.value
) : nullptr,
278 client_handle_transmission_status
,
282 static int action_executor_start_session_handler(struct action_executor
*executor
283 __attribute__((unused
)),
284 const struct action_work_item
*work_item
,
285 struct action_work_subitem
*item
)
287 const char *session_name
;
288 enum lttng_action_status action_status
;
289 enum lttng_error_code cmd_ret
;
290 struct lttng_action
*action
= item
->action
;
292 lttng::urcu::read_lock_guard read_lock
;
294 action_status
= lttng_action_start_session_get_session_name(action
, &session_name
);
295 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
296 ERR("Failed to get session name from `%s` action", get_action_name(action
));
301 * Validate if at the moment of the action was queued the session
302 * existed. If not skip the action altogether.
304 if (!item
->context
.session_id
.is_set
) {
305 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
307 get_action_name(action
),
308 get_trigger_name(work_item
->trigger
));
309 lttng_action_increase_execution_failure_count(action
);
314 * Mind the order of the declaration of list_lock vs target_session:
315 * the session list lock must always be released _after_ the release of
316 * a session's reference (the destruction of a ref/locked_ref) to ensure
317 * since the reference's release may unpublish the session from the list of
320 const auto list_lock
= lttng::sessiond::lock_session_list();
321 ltt_session::locked_ref session
;
324 session
= ltt_session::find_locked_session(
325 LTTNG_OPTIONAL_GET(item
->context
.session_id
));
326 } catch (const lttng::sessiond::exceptions::session_not_found_error
& ex
) {
327 DBG_FMT("Failed to execution trigger action: {}, action=`{}`, trigger_name=`{}`, location='{}'",
330 get_action_name(action
),
331 get_trigger_name(work_item
->trigger
),
333 lttng_action_increase_execution_failure_count(action
);
337 if (session
->destroyed
) {
338 DBG("Session `%s` with id = %" PRIu64
339 " is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
342 get_action_name(action
),
343 get_trigger_name(work_item
->trigger
));
347 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
.get())) {
351 cmd_ret
= (lttng_error_code
) cmd_start_trace(session
.get());
354 DBG("Successfully started session `%s` on behalf of trigger `%s`",
356 get_trigger_name(work_item
->trigger
));
358 case LTTNG_ERR_TRACE_ALREADY_STARTED
:
359 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
361 get_trigger_name(work_item
->trigger
));
364 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
366 get_trigger_name(work_item
->trigger
),
367 lttng_strerror(-cmd_ret
));
368 lttng_action_increase_execution_failure_count(action
);
375 static int action_executor_stop_session_handler(struct action_executor
*executor
376 __attribute__((unused
)),
377 const struct action_work_item
*work_item
,
378 struct action_work_subitem
*item
)
380 const char *session_name
;
381 enum lttng_action_status action_status
;
382 enum lttng_error_code cmd_ret
;
383 struct lttng_action
*action
= item
->action
;
385 lttng::urcu::read_lock_guard read_lock
;
387 action_status
= lttng_action_stop_session_get_session_name(action
, &session_name
);
388 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
389 ERR("Failed to get session name from `%s` action", get_action_name(action
));
394 * Validate if, at the moment the action was queued, the target session
395 * existed. If not, skip the action altogether.
397 if (!item
->context
.session_id
.is_set
) {
398 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
400 get_action_name(action
),
401 get_trigger_name(work_item
->trigger
));
402 lttng_action_increase_execution_failure_count(action
);
407 * Mind the order of the declaration of list_lock vs target_session:
408 * the session list lock must always be released _after_ the release of
409 * a session's reference (the destruction of a ref/locked_ref) to ensure
410 * since the reference's release may unpublish the session from the list of
413 const auto list_lock
= lttng::sessiond::lock_session_list();
414 ltt_session::locked_ref session
;
417 session
= ltt_session::find_locked_session(
418 LTTNG_OPTIONAL_GET(item
->context
.session_id
));
419 } catch (const lttng::sessiond::exceptions::session_not_found_error
& ex
) {
420 DBG_FMT("Failed to execution trigger action: {}, action=`{}`, trigger_name=`{}`, location='{}'",
423 get_action_name(action
),
424 get_trigger_name(work_item
->trigger
),
426 lttng_action_increase_execution_failure_count(action
);
430 if (session
->destroyed
) {
431 DBG("Session `%s` with id = %" PRIu64
432 " is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
435 get_action_name(action
),
436 get_trigger_name(work_item
->trigger
));
440 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
.get())) {
444 cmd_ret
= (lttng_error_code
) cmd_stop_trace(session
.get());
447 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
449 get_trigger_name(work_item
->trigger
));
451 case LTTNG_ERR_TRACE_ALREADY_STOPPED
:
452 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
454 get_trigger_name(work_item
->trigger
));
457 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
459 get_trigger_name(work_item
->trigger
),
460 lttng_strerror(-cmd_ret
));
461 lttng_action_increase_execution_failure_count(action
);
468 static int action_executor_rotate_session_handler(struct action_executor
*executor
469 __attribute__((unused
)),
470 const struct action_work_item
*work_item
,
471 struct action_work_subitem
*item
)
473 const char *session_name
;
474 enum lttng_action_status action_status
;
475 enum lttng_error_code cmd_ret
;
476 struct lttng_action
*action
= item
->action
;
478 lttng::urcu::read_lock_guard read_lock
;
480 action_status
= lttng_action_rotate_session_get_session_name(action
, &session_name
);
481 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
482 ERR("Failed to get session name from `%s` action", get_action_name(action
));
487 * Validate if, at the moment the action was queued, the target session
488 * existed. If not, skip the action altogether.
490 if (!item
->context
.session_id
.is_set
) {
491 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
493 get_action_name(action
),
494 get_trigger_name(work_item
->trigger
));
495 lttng_action_increase_execution_failure_count(action
);
500 * Mind the order of the declaration of list_lock vs target_session:
501 * the session list lock must always be released _after_ the release of
502 * a session's reference (the destruction of a ref/locked_ref) to ensure
503 * since the reference's release may unpublish the session from the list of
506 const auto list_lock
= lttng::sessiond::lock_session_list();
507 ltt_session::locked_ref session
;
510 session
= ltt_session::find_locked_session(
511 LTTNG_OPTIONAL_GET(item
->context
.session_id
));
512 } catch (const lttng::sessiond::exceptions::session_not_found_error
& ex
) {
513 DBG_FMT("Failed to execution trigger action: {}, action=`{}`, trigger_name=`{}`, location='{}'",
516 get_action_name(action
),
517 get_trigger_name(work_item
->trigger
),
519 lttng_action_increase_execution_failure_count(action
);
523 if (session
->destroyed
) {
524 DBG("Session `%s` with id = %" PRIu64
525 " is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
528 get_action_name(action
),
529 get_trigger_name(work_item
->trigger
));
533 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
.get())) {
537 cmd_ret
= (lttng_error_code
) cmd_rotate_session(
538 session
.get(), nullptr, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
541 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
543 get_trigger_name(work_item
->trigger
));
545 case LTTNG_ERR_ROTATION_PENDING
:
546 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
548 get_trigger_name(work_item
->trigger
));
549 lttng_action_increase_execution_failure_count(action
);
551 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
552 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
553 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation has already been completed since the last stop or clear",
555 get_trigger_name(work_item
->trigger
));
558 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
560 get_trigger_name(work_item
->trigger
),
561 lttng_strerror(-cmd_ret
));
562 lttng_action_increase_execution_failure_count(action
);
569 static int action_executor_snapshot_session_handler(struct action_executor
*executor
570 __attribute__((unused
)),
571 const struct action_work_item
*work_item
,
572 struct action_work_subitem
*item
)
574 const char *session_name
;
575 enum lttng_action_status action_status
;
576 lttng_snapshot_output default_snapshot_output
;
577 const struct lttng_snapshot_output
*snapshot_output
= &default_snapshot_output
;
578 enum lttng_error_code cmd_ret
;
579 struct lttng_action
*action
= item
->action
;
581 default_snapshot_output
.max_size
= UINT64_MAX
;
583 lttng::urcu::read_lock_guard read_lock
;
586 * Validate if, at the moment the action was queued, the target session
587 * existed. If not, skip the action altogether.
589 if (!item
->context
.session_id
.is_set
) {
590 DBG("Session was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
591 get_action_name(action
),
592 get_trigger_name(work_item
->trigger
));
593 lttng_action_increase_execution_failure_count(action
);
597 action_status
= lttng_action_snapshot_session_get_session_name(action
, &session_name
);
598 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
599 ERR("Failed to get session name from `%s` action", get_action_name(action
));
603 action_status
= lttng_action_snapshot_session_get_output(action
, &snapshot_output
);
604 if (action_status
!= LTTNG_ACTION_STATUS_OK
&& action_status
!= LTTNG_ACTION_STATUS_UNSET
) {
605 ERR("Failed to get output from `%s` action", get_action_name(action
));
610 * Mind the order of the declaration of list_lock vs session:
611 * the session list lock must always be released _after_ the release of
612 * a session's reference (the destruction of a ref/locked_ref) to ensure
613 * since the reference's release may unpublish the session from the list of
616 const auto list_lock
= lttng::sessiond::lock_session_list();
617 ltt_session::locked_ref session
;
620 session
= ltt_session::find_locked_session(
621 LTTNG_OPTIONAL_GET(item
->context
.session_id
));
622 } catch (const lttng::sessiond::exceptions::session_not_found_error
& ex
) {
623 DBG_FMT("Failed to execution trigger action: {}, action=`{}`, trigger_name=`{}`, location='{}'",
626 get_action_name(action
),
627 get_trigger_name(work_item
->trigger
),
629 lttng_action_increase_execution_failure_count(action
);
633 if (session
->destroyed
) {
634 DBG("Session `%s` with id = %" PRIu64
635 " is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
638 get_action_name(action
),
639 get_trigger_name(work_item
->trigger
));
643 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
.get())) {
647 cmd_ret
= (lttng_error_code
) cmd_snapshot_record(session
.get(), snapshot_output
, 0);
650 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
652 get_trigger_name(work_item
->trigger
));
655 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
657 get_trigger_name(work_item
->trigger
),
658 lttng_strerror(-cmd_ret
));
659 lttng_action_increase_execution_failure_count(action
);
666 static int action_executor_list_handler(struct action_executor
*executor
__attribute__((unused
)),
667 const struct action_work_item
*work_item
668 __attribute__((unused
)),
669 struct action_work_subitem
*item
__attribute__((unused
)))
671 ERR("Execution of a list action by the action executor should never occur");
675 static int action_executor_generic_handler(struct action_executor
*executor
,
676 const struct action_work_item
*work_item
,
677 struct action_work_subitem
*item
)
680 struct lttng_action
*action
= item
->action
;
681 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
683 LTTNG_ASSERT(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
685 lttng_action_increase_execution_request_count(action
);
686 if (!lttng_action_should_execute(action
)) {
687 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64
,
688 get_action_name(action
),
689 get_trigger_name(work_item
->trigger
),
695 lttng_action_increase_execution_count(action
);
696 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64
,
697 get_action_name(action
),
698 get_trigger_name(work_item
->trigger
),
700 ret
= action_executors
[action_type
](executor
, work_item
, item
);
705 static int action_work_item_execute(struct action_executor
*executor
,
706 struct action_work_item
*work_item
)
711 DBG("Starting execution of action work item %" PRIu64
" of trigger `%s`",
713 get_trigger_name(work_item
->trigger
));
715 count
= lttng_dynamic_array_get_count(&work_item
->subitems
);
716 for (i
= 0; i
< count
; i
++) {
717 struct action_work_subitem
*item
;
719 item
= (action_work_subitem
*) lttng_dynamic_array_get_element(&work_item
->subitems
,
721 ret
= action_executor_generic_handler(executor
, work_item
, item
);
727 DBG("Completed execution of action work item %" PRIu64
" of trigger `%s`",
729 get_trigger_name(work_item
->trigger
));
733 static void action_work_item_destroy(struct action_work_item
*work_item
)
735 lttng_trigger_put(work_item
->trigger
);
736 lttng_evaluation_destroy(work_item
->evaluation
);
737 notification_client_list_put(work_item
->client_list
);
738 lttng_dynamic_array_reset(&work_item
->subitems
);
742 static void *action_executor_thread(void *_data
)
744 struct action_executor
*executor
= (action_executor
*) _data
;
746 LTTNG_ASSERT(executor
);
748 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR
);
750 rcu_register_thread();
753 DBG("Entering work execution loop");
754 pthread_mutex_lock(&executor
->work
.lock
);
755 while (!executor
->should_quit
) {
757 struct action_work_item
*work_item
;
759 health_code_update();
760 if (executor
->work
.pending_count
== 0) {
762 DBG("No work items enqueued, entering wait");
763 pthread_cond_wait(&executor
->work
.cond
, &executor
->work
.lock
);
764 DBG("Woke-up from wait");
769 /* Pop item from front of the list with work lock held. */
770 work_item
= cds_list_first_entry(
771 &executor
->work
.list
, struct action_work_item
, list_node
);
772 cds_list_del(&work_item
->list_node
);
773 executor
->work
.pending_count
--;
776 * Work can be performed without holding the work lock,
777 * allowing new items to be queued.
779 pthread_mutex_unlock(&executor
->work
.lock
);
781 /* Execute item only if a trigger is registered. */
782 lttng_trigger_lock(work_item
->trigger
);
783 if (!lttng_trigger_is_registered(work_item
->trigger
)) {
784 const char *trigger_name
= nullptr;
785 uid_t trigger_owner_uid
;
786 enum lttng_trigger_status trigger_status
;
788 trigger_name
= get_trigger_name(work_item
->trigger
);
791 lttng_trigger_get_owner_uid(work_item
->trigger
, &trigger_owner_uid
);
792 LTTNG_ASSERT(trigger_status
== LTTNG_TRIGGER_STATUS_OK
);
794 DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64
795 ", trigger name = `%s`, trigger owner uid = %d",
798 (int) trigger_owner_uid
);
803 ret
= action_work_item_execute(executor
, work_item
);
806 lttng_trigger_unlock(work_item
->trigger
);
807 action_work_item_destroy(work_item
);
813 health_code_update();
814 pthread_mutex_lock(&executor
->work
.lock
);
817 if (executor
->should_quit
) {
818 pthread_mutex_unlock(&executor
->work
.lock
);
820 DBG("Left work execution loop");
822 health_code_update();
824 rcu_thread_offline();
825 rcu_unregister_thread();
826 health_unregister(the_health_sessiond
);
831 static bool shutdown_action_executor_thread(void *_data
)
833 struct action_executor
*executor
= (action_executor
*) _data
;
835 pthread_mutex_lock(&executor
->work
.lock
);
836 executor
->should_quit
= true;
837 pthread_cond_signal(&executor
->work
.cond
);
838 pthread_mutex_unlock(&executor
->work
.lock
);
842 static void clean_up_action_executor_thread(void *_data
)
844 struct action_executor
*executor
= (action_executor
*) _data
;
846 LTTNG_ASSERT(cds_list_empty(&executor
->work
.list
));
848 pthread_mutex_destroy(&executor
->work
.lock
);
849 pthread_cond_destroy(&executor
->work
.cond
);
853 struct action_executor
*action_executor_create(struct notification_thread_handle
*handle
)
855 struct action_executor
*executor
= zmalloc
<action_executor
>();
861 CDS_INIT_LIST_HEAD(&executor
->work
.list
);
862 pthread_cond_init(&executor
->work
.cond
, nullptr);
863 pthread_mutex_init(&executor
->work
.lock
, nullptr);
864 executor
->notification_thread_handle
= handle
;
866 executor
->thread
= lttng_thread_create(THREAD_NAME
,
867 action_executor_thread
,
868 shutdown_action_executor_thread
,
869 clean_up_action_executor_thread
,
875 void action_executor_destroy(struct action_executor
*executor
)
877 struct action_work_item
*work_item
, *tmp
;
879 /* TODO Wait for work list to drain? */
880 lttng_thread_shutdown(executor
->thread
);
881 pthread_mutex_lock(&executor
->work
.lock
);
882 if (executor
->work
.pending_count
!= 0) {
884 " trigger action%s still queued for execution and will be discarded",
885 executor
->work
.pending_count
,
886 executor
->work
.pending_count
== 1 ? " is" : "s are");
889 cds_list_for_each_entry_safe (work_item
, tmp
, &executor
->work
.list
, list_node
) {
890 WARN("Discarding action work item %" PRIu64
" associated to trigger `%s`",
892 get_trigger_name(work_item
->trigger
));
893 cds_list_del(&work_item
->list_node
);
894 action_work_item_destroy(work_item
);
896 pthread_mutex_unlock(&executor
->work
.lock
);
897 lttng_thread_put(executor
->thread
);
900 /* RCU read-lock must be held by the caller. */
901 enum action_executor_status
902 action_executor_enqueue_trigger(struct action_executor
*executor
,
903 struct lttng_trigger
*trigger
,
904 struct lttng_evaluation
*evaluation
,
905 const struct lttng_credentials
*object_creds
,
906 struct notification_client_list
*client_list
)
909 enum action_executor_status executor_status
= ACTION_EXECUTOR_STATUS_OK
;
910 const uint64_t work_item_id
= executor
->next_work_item_id
++;
911 struct action_work_item
*work_item
;
914 LTTNG_ASSERT(trigger
);
915 ASSERT_RCU_READ_LOCKED();
917 pthread_mutex_lock(&executor
->work
.lock
);
918 /* Check for queue overflow. */
919 if (executor
->work
.pending_count
>= MAX_QUEUED_WORK_COUNT
) {
920 /* Most likely spammy, remove if it is the case. */
921 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64
,
922 get_trigger_name(trigger
),
924 executor_status
= ACTION_EXECUTOR_STATUS_OVERFLOW
;
928 work_item
= zmalloc
<action_work_item
>();
930 PERROR("Failed to allocate action executor work item: trigger name = `%s`",
931 get_trigger_name(trigger
));
932 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
936 lttng_trigger_get(trigger
);
938 const bool reference_acquired
= notification_client_list_get(client_list
);
940 LTTNG_ASSERT(reference_acquired
);
943 work_item
->id
= work_item_id
;
944 work_item
->trigger
= trigger
;
946 /* Ownership transferred to the work item. */
947 work_item
->evaluation
= evaluation
;
948 evaluation
= nullptr;
950 work_item
->client_list
= client_list
;
951 work_item
->object_creds
.is_set
= !!object_creds
;
953 work_item
->object_creds
.value
= *object_creds
;
956 CDS_INIT_LIST_HEAD(&work_item
->list_node
);
958 /* Build the array of action work subitems for the passed trigger. */
959 lttng_dynamic_array_init(&work_item
->subitems
,
960 sizeof(struct action_work_subitem
),
961 action_work_subitem_destructor
);
963 ret
= populate_subitem_array_from_trigger(trigger
, &work_item
->subitems
);
965 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
966 get_trigger_name(trigger
));
967 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
971 cds_list_add_tail(&work_item
->list_node
, &executor
->work
.list
);
972 executor
->work
.pending_count
++;
973 DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64
,
974 get_trigger_name(trigger
),
980 pthread_cond_signal(&executor
->work
.cond
);
983 pthread_mutex_unlock(&executor
->work
.lock
);
984 lttng_evaluation_destroy(evaluation
);
985 return executor_status
;
988 static int add_action_to_subitem_array(struct lttng_action
*action
,
989 struct lttng_dynamic_array
*subitems
)
992 enum lttng_action_type type
= lttng_action_get_type(action
);
993 const char *session_name
= nullptr;
994 enum lttng_action_status status
;
995 struct action_work_subitem subitem
= {
998 .session_id
= LTTNG_OPTIONAL_INIT_UNSET
,
1002 LTTNG_ASSERT(action
);
1003 LTTNG_ASSERT(subitems
);
1005 if (type
== LTTNG_ACTION_TYPE_LIST
) {
1006 for (auto inner_action
: lttng::ctl::action_list_view(action
)) {
1007 LTTNG_ASSERT(inner_action
);
1009 ret
= add_action_to_subitem_array(inner_action
, subitems
);
1016 * Go directly to the end since there is no need to add the
1017 * list action by itself to the subitems array.
1022 /* Gather execution context. */
1024 case LTTNG_ACTION_TYPE_NOTIFY
:
1026 case LTTNG_ACTION_TYPE_START_SESSION
:
1027 status
= lttng_action_start_session_get_session_name(action
, &session_name
);
1028 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1030 case LTTNG_ACTION_TYPE_STOP_SESSION
:
1031 status
= lttng_action_stop_session_get_session_name(action
, &session_name
);
1032 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1034 case LTTNG_ACTION_TYPE_ROTATE_SESSION
:
1035 status
= lttng_action_rotate_session_get_session_name(action
, &session_name
);
1036 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1038 case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
:
1039 status
= lttng_action_snapshot_session_get_session_name(action
, &session_name
);
1040 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1042 case LTTNG_ACTION_TYPE_LIST
:
1043 case LTTNG_ACTION_TYPE_UNKNOWN
:
1051 * Fetch the session execution context info as needed.
1052 * Note that we could decide to not add an action for which we know the
1053 * execution will not happen (i.e no session exists for that name). For
1054 * now we leave the decision to skip to the action executor for sake of
1055 * simplicity and consistency.
1057 if (session_name
!= nullptr) {
1058 uint64_t session_id
;
1061 * Instantaneous sampling of the session id if present.
1063 * This method is preferred over `sessiond_find_by_name` then
1064 * fetching the session'd id since `sessiond_find_by_name`
1065 * requires the session list lock to be taken.
1067 * Taking the session list lock can lead to a deadlock
1068 * between the action executor and the notification thread
1069 * (caller of add_action_to_subitem_array). It is okay if the
1070 * session state changes between the enqueuing time and the
1071 * execution time. The execution context is validated at
1074 if (sample_session_id_by_name(session_name
, &session_id
)) {
1075 LTTNG_OPTIONAL_SET(&subitem
.context
.session_id
, session_id
);
1079 /* Get a reference to the action. */
1080 lttng_action_get(action
);
1081 subitem
.action
= action
;
1083 ret
= lttng_dynamic_array_add_element(subitems
, &subitem
);
1085 ERR("Failed to add work subitem to the subitem array");
1086 lttng_action_put(action
);
1095 static int populate_subitem_array_from_trigger(struct lttng_trigger
*trigger
,
1096 struct lttng_dynamic_array
*subitems
)
1098 struct lttng_action
*action
;
1100 action
= lttng_trigger_get_action(trigger
);
1101 LTTNG_ASSERT(action
);
1103 return add_action_to_subitem_array(action
, subitems
);