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"
15 #include <common/dynamic-array.hpp>
16 #include <common/macros.hpp>
17 #include <common/optional.hpp>
18 #include <lttng/action/action-internal.hpp>
19 #include <lttng/action/list-internal.hpp>
20 #include <lttng/action/list.h>
21 #include <lttng/action/notify-internal.hpp>
22 #include <lttng/action/notify.h>
23 #include <lttng/action/rotate-session.h>
24 #include <lttng/action/snapshot-session.h>
25 #include <lttng/action/start-session.h>
26 #include <lttng/action/stop-session.h>
27 #include <lttng/condition/evaluation.h>
28 #include <lttng/condition/event-rule-matches-internal.hpp>
29 #include <lttng/lttng-error.h>
30 #include <lttng/trigger/trigger-internal.hpp>
34 #include <urcu/list.h>
36 #define THREAD_NAME "Action Executor"
37 #define MAX_QUEUED_WORK_COUNT 8192
40 * A work item is composed of a dynamic array of sub-items which
41 * represent a flattened, and augmented, version of a trigger's actions.
43 * We cannot rely solely on the trigger's actions since each action can have an
44 * execution context we need to comply with.
46 * The notion of execution context is required since for some actions the
47 * associated object are referenced by name and not by id. This can lead to
48 * a number of ambiguities when executing an action work item.
50 * For example, let's take a simple trigger such as:
51 * - condition: ust event a
52 * - action: start session S
54 * At time T, session S exists.
55 * At T + 1, the event A is hit.
56 * At T + 2, the tracer event notification is received and the work item is
57 * queued. Here session S have an id of 1.
58 * At T + 3, the session S is destroyed and a new session S is created, with a
59 * resulting id of 200.
60 * At T +4, the work item is popped from the queue and begin execution and will
61 * start session S with an id of 200 instead of the session S id 1 that was
62 * present at the queuing phase.
64 * The context to be respected is the one when the work item is queued. If the
65 * execution context is not the same at the moment of execution, we skip the
66 * execution of that sub-item.
68 * It is the same policy in regards to the validity of the associated
69 * trigger object at the moment of execution, if the trigger is found to be
70 * unregistered, the execution is skipped.
73 struct action_work_item
{
77 * The actions to be executed with their respective execution context.
78 * See struct `action_work_subitem`.
80 struct lttng_dynamic_array subitems
;
82 /* Execution context data */
83 struct lttng_trigger
*trigger
;
84 struct lttng_evaluation
*evaluation
;
85 struct notification_client_list
*client_list
;
86 LTTNG_OPTIONAL(struct lttng_credentials
) object_creds
;
87 struct cds_list_head list_node
;
90 struct action_work_subitem
{
91 struct lttng_action
*action
;
93 /* Used by actions targeting a session. */
94 LTTNG_OPTIONAL(uint64_t) session_id
;
98 struct action_executor
{
99 struct lttng_thread
*thread
;
100 struct notification_thread_handle
*notification_thread_handle
;
102 uint64_t pending_count
;
103 struct cds_list_head list
;
105 pthread_mutex_t lock
;
108 uint64_t next_work_item_id
;
112 * Only return non-zero on a fatal error that should shut down the action
115 typedef int (*action_executor_handler
)(struct action_executor
*executor
,
116 const struct action_work_item
*,
117 struct action_work_subitem
*item
);
119 static int action_executor_notify_handler(struct action_executor
*executor
,
120 const struct action_work_item
*,
121 struct action_work_subitem
*);
122 static int action_executor_start_session_handler(
123 struct action_executor
*executor
,
124 const struct action_work_item
*,
125 struct action_work_subitem
*);
126 static int action_executor_stop_session_handler(
127 struct action_executor
*executor
,
128 const struct action_work_item
*,
129 struct action_work_subitem
*);
130 static int action_executor_rotate_session_handler(
131 struct action_executor
*executor
,
132 const struct action_work_item
*,
133 struct action_work_subitem
*);
134 static int action_executor_snapshot_session_handler(
135 struct action_executor
*executor
,
136 const struct action_work_item
*,
137 struct action_work_subitem
*);
138 static int action_executor_list_handler(struct action_executor
*executor
,
139 const struct action_work_item
*,
140 struct action_work_subitem
*);
141 static int action_executor_generic_handler(struct action_executor
*executor
,
142 const struct action_work_item
*,
143 struct action_work_subitem
*);
145 static const action_executor_handler action_executors
[] = {
146 action_executor_notify_handler
,
147 action_executor_start_session_handler
,
148 action_executor_stop_session_handler
,
149 action_executor_rotate_session_handler
,
150 action_executor_snapshot_session_handler
,
151 action_executor_list_handler
,
154 /* Forward declaration */
155 static int add_action_to_subitem_array(struct lttng_action
*action
,
156 struct lttng_dynamic_array
*subitems
);
158 static int populate_subitem_array_from_trigger(struct lttng_trigger
*trigger
,
159 struct lttng_dynamic_array
*subitems
);
161 static void action_work_subitem_destructor(void *element
)
163 struct action_work_subitem
*subitem
= (action_work_subitem
*) element
;
165 lttng_action_put(subitem
->action
);
168 static const char *get_action_name(const struct lttng_action
*action
)
170 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
172 LTTNG_ASSERT(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
174 return lttng_action_type_string(action_type
);
177 /* Check if this trigger allowed to interect with a given session. */
178 static bool is_trigger_allowed_for_session(const struct lttng_trigger
*trigger
,
179 struct ltt_session
*session
)
181 bool is_allowed
= false;
182 const struct lttng_credentials session_creds
= {
183 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
->uid
),
184 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
->gid
),
186 /* Can never be NULL. */
187 const struct lttng_credentials
*trigger_creds
=
188 lttng_trigger_get_credentials(trigger
);
190 is_allowed
= (lttng_credentials_is_equal_uid(trigger_creds
, &session_creds
)) ||
191 (lttng_credentials_get_uid(trigger_creds
) == 0);
193 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
195 (long int) session
->uid
,
196 (long int) session
->gid
,
197 (long int) lttng_credentials_get_uid(trigger_creds
));
203 static const char *get_trigger_name(const struct lttng_trigger
*trigger
)
205 const char *trigger_name
;
206 enum lttng_trigger_status trigger_status
;
208 trigger_status
= lttng_trigger_get_name(trigger
, &trigger_name
);
209 switch (trigger_status
) {
210 case LTTNG_TRIGGER_STATUS_OK
:
212 case LTTNG_TRIGGER_STATUS_UNSET
:
213 trigger_name
= "(anonymous)";
216 trigger_name
= "(failed to get name)";
223 static int client_handle_transmission_status(
224 struct notification_client
*client
,
225 enum client_transmission_status status
,
229 struct action_executor
*executor
= (action_executor
*) user_data
;
230 bool update_communication
= true;
233 case CLIENT_TRANSMISSION_STATUS_COMPLETE
:
234 DBG("Successfully sent full notification to client, client_id = %" PRIu64
,
237 * There is no need to wake the (e)poll thread. If it was waiting for
238 * "out" events on the client's socket, it will see that no payload
239 * in queued and will unsubscribe from that event.
241 * In the other cases, we have to wake the the (e)poll thread to either
242 * handle the error on the client or to get it to monitor the client "out"
245 update_communication
= false;
247 case CLIENT_TRANSMISSION_STATUS_QUEUED
:
248 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64
,
251 case CLIENT_TRANSMISSION_STATUS_FAIL
:
252 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64
,
256 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64
,
262 if (!update_communication
) {
266 /* Safe to read client's id without locking as it is immutable. */
267 ret
= notification_thread_client_communication_update(
268 executor
->notification_thread_handle
, client
->id
,
274 static int action_executor_notify_handler(struct action_executor
*executor
,
275 const struct action_work_item
*work_item
,
276 struct action_work_subitem
*item
__attribute__((unused
)))
278 return notification_client_list_send_evaluation(work_item
->client_list
,
280 work_item
->evaluation
,
281 work_item
->object_creds
.is_set
?
282 &(work_item
->object_creds
.value
) :
284 client_handle_transmission_status
, executor
);
287 static int action_executor_start_session_handler(
288 struct action_executor
*executor
__attribute__((unused
)),
289 const struct action_work_item
*work_item
,
290 struct action_work_subitem
*item
)
293 const char *session_name
;
294 enum lttng_action_status action_status
;
295 struct ltt_session
*session
;
296 enum lttng_error_code cmd_ret
;
297 struct lttng_action
*action
= item
->action
;
299 action_status
= lttng_action_start_session_get_session_name(
300 action
, &session_name
);
301 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
302 ERR("Failed to get session name from `%s` action",
303 get_action_name(action
));
309 * Validate if at the moment of the action was queued the session
310 * existed. If not skip the action altogether.
312 if (!item
->context
.session_id
.is_set
) {
313 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
314 session_name
, get_action_name(action
),
315 get_trigger_name(work_item
->trigger
));
316 lttng_action_increase_execution_failure_count(action
);
322 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
324 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
325 session_name
, get_action_name(action
),
326 get_trigger_name(work_item
->trigger
));
327 lttng_action_increase_execution_failure_count(action
);
328 goto error_unlock_list
;
331 session_lock(session
);
332 if (session
->destroyed
) {
333 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
334 session
->name
, session
->id
,
335 get_action_name(action
),
336 get_trigger_name(work_item
->trigger
));
337 goto error_unlock_session
;
340 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
341 goto error_unlock_session
;
344 cmd_ret
= (lttng_error_code
) cmd_start_trace(session
);
347 DBG("Successfully started session `%s` on behalf of trigger `%s`",
348 session_name
, get_trigger_name(work_item
->trigger
));
350 case LTTNG_ERR_TRACE_ALREADY_STARTED
:
351 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
352 session_name
, get_trigger_name(work_item
->trigger
));
355 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
356 session_name
, get_trigger_name(work_item
->trigger
),
357 lttng_strerror(-cmd_ret
));
358 lttng_action_increase_execution_failure_count(action
);
362 error_unlock_session
:
363 session_unlock(session
);
364 session_put(session
);
367 session_unlock_list();
372 static int action_executor_stop_session_handler(
373 struct action_executor
*executor
__attribute__((unused
)),
374 const struct action_work_item
*work_item
,
375 struct action_work_subitem
*item
)
378 const char *session_name
;
379 enum lttng_action_status action_status
;
380 struct ltt_session
*session
;
381 enum lttng_error_code cmd_ret
;
382 struct lttng_action
*action
= item
->action
;
384 action_status
= lttng_action_stop_session_get_session_name(
385 action
, &session_name
);
386 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
387 ERR("Failed to get session name from `%s` action",
388 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`",
399 session_name
, get_action_name(action
),
400 get_trigger_name(work_item
->trigger
));
401 lttng_action_increase_execution_failure_count(action
);
407 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
409 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
410 session_name
, get_action_name(action
),
411 get_trigger_name(work_item
->trigger
));
412 lttng_action_increase_execution_failure_count(action
);
413 goto error_unlock_list
;
416 session_lock(session
);
417 if (session
->destroyed
) {
418 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
419 session
->name
, session
->id
,
420 get_action_name(action
),
421 get_trigger_name(work_item
->trigger
));
422 goto error_unlock_session
;
425 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
426 goto error_unlock_session
;
429 cmd_ret
= (lttng_error_code
) cmd_stop_trace(session
);
432 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
433 session_name
, get_trigger_name(work_item
->trigger
));
435 case LTTNG_ERR_TRACE_ALREADY_STOPPED
:
436 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
437 session_name
, get_trigger_name(work_item
->trigger
));
440 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
441 session_name
, get_trigger_name(work_item
->trigger
),
442 lttng_strerror(-cmd_ret
));
443 lttng_action_increase_execution_failure_count(action
);
447 error_unlock_session
:
448 session_unlock(session
);
449 session_put(session
);
452 session_unlock_list();
457 static int action_executor_rotate_session_handler(
458 struct action_executor
*executor
__attribute__((unused
)),
459 const struct action_work_item
*work_item
,
460 struct action_work_subitem
*item
)
463 const char *session_name
;
464 enum lttng_action_status action_status
;
465 struct ltt_session
*session
;
466 enum lttng_error_code cmd_ret
;
467 struct lttng_action
*action
= item
->action
;
469 action_status
= lttng_action_rotate_session_get_session_name(
470 action
, &session_name
);
471 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
472 ERR("Failed to get session name from `%s` action",
473 get_action_name(action
));
479 * Validate if, at the moment the action was queued, the target session
480 * existed. If not, skip the action altogether.
482 if (!item
->context
.session_id
.is_set
) {
483 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
484 session_name
, get_action_name(action
),
485 get_trigger_name(work_item
->trigger
));
486 lttng_action_increase_execution_failure_count(action
);
492 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
494 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
495 session_name
, get_action_name(action
),
496 get_trigger_name(work_item
->trigger
));
497 lttng_action_increase_execution_failure_count(action
);
498 goto error_unlock_list
;
501 session_lock(session
);
502 if (session
->destroyed
) {
503 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
504 session
->name
, session
->id
,
505 get_action_name(action
),
506 get_trigger_name(work_item
->trigger
));
507 goto error_unlock_session
;
510 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
511 goto error_unlock_session
;
514 cmd_ret
= (lttng_error_code
) cmd_rotate_session(session
, NULL
, false,
515 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
518 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
519 session_name
, get_trigger_name(work_item
->trigger
));
521 case LTTNG_ERR_ROTATION_PENDING
:
522 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
523 session_name
, get_trigger_name(work_item
->trigger
));
524 lttng_action_increase_execution_failure_count(action
);
526 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
527 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
528 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",
529 session_name
, get_trigger_name(work_item
->trigger
));
532 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
533 session_name
, get_trigger_name(work_item
->trigger
),
534 lttng_strerror(-cmd_ret
));
535 lttng_action_increase_execution_failure_count(action
);
539 error_unlock_session
:
540 session_unlock(session
);
541 session_put(session
);
544 session_unlock_list();
549 static int action_executor_snapshot_session_handler(
550 struct action_executor
*executor
__attribute__((unused
)),
551 const struct action_work_item
*work_item
,
552 struct action_work_subitem
*item
)
555 const char *session_name
;
556 enum lttng_action_status action_status
;
557 struct ltt_session
*session
;
558 lttng_snapshot_output default_snapshot_output
;
559 const struct lttng_snapshot_output
*snapshot_output
=
560 &default_snapshot_output
;
561 enum lttng_error_code cmd_ret
;
562 struct lttng_action
*action
= item
->action
;
564 default_snapshot_output
.max_size
= UINT64_MAX
;
567 * Validate if, at the moment the action was queued, the target session
568 * existed. If not, skip the action altogether.
570 if (!item
->context
.session_id
.is_set
) {
571 DBG("Session was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
572 get_action_name(action
),
573 get_trigger_name(work_item
->trigger
));
574 lttng_action_increase_execution_failure_count(action
);
578 action_status
= lttng_action_snapshot_session_get_session_name(
579 action
, &session_name
);
580 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
581 ERR("Failed to get session name from `%s` action",
582 get_action_name(action
));
587 action_status
= lttng_action_snapshot_session_get_output(
588 action
, &snapshot_output
);
589 if (action_status
!= LTTNG_ACTION_STATUS_OK
&&
590 action_status
!= LTTNG_ACTION_STATUS_UNSET
) {
591 ERR("Failed to get output from `%s` action",
592 get_action_name(action
));
599 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
601 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
602 session_name
, get_action_name(action
),
603 get_trigger_name(work_item
->trigger
));
604 lttng_action_increase_execution_failure_count(action
);
605 goto error_unlock_list
;
608 session_lock(session
);
609 if (session
->destroyed
) {
610 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
611 session
->name
, session
->id
,
612 get_action_name(action
),
613 get_trigger_name(work_item
->trigger
));
614 goto error_unlock_session
;
617 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
618 goto error_unlock_session
;
621 cmd_ret
= (lttng_error_code
) cmd_snapshot_record(session
, snapshot_output
, 0);
624 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
625 session_name
, get_trigger_name(work_item
->trigger
));
628 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
629 session_name
, get_trigger_name(work_item
->trigger
),
630 lttng_strerror(-cmd_ret
));
631 lttng_action_increase_execution_failure_count(action
);
635 error_unlock_session
:
636 session_unlock(session
);
637 session_put(session
);
640 session_unlock_list();
645 static int action_executor_list_handler(
646 struct action_executor
*executor
__attribute__((unused
)),
647 const struct action_work_item
*work_item
__attribute__((unused
)),
648 struct action_work_subitem
*item
__attribute__((unused
)))
650 ERR("Execution of a list action by the action executor should never occur");
654 static int action_executor_generic_handler(struct action_executor
*executor
,
655 const struct action_work_item
*work_item
,
656 struct action_work_subitem
*item
)
659 struct lttng_action
*action
= item
->action
;
660 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
662 LTTNG_ASSERT(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
664 lttng_action_increase_execution_request_count(action
);
665 if (!lttng_action_should_execute(action
)) {
666 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64
,
667 get_action_name(action
),
668 get_trigger_name(work_item
->trigger
),
674 lttng_action_increase_execution_count(action
);
675 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64
,
676 get_action_name(action
),
677 get_trigger_name(work_item
->trigger
),
679 ret
= action_executors
[action_type
](executor
, work_item
, item
);
684 static int action_work_item_execute(struct action_executor
*executor
,
685 struct action_work_item
*work_item
)
690 DBG("Starting execution of action work item %" PRIu64
" of trigger `%s`",
691 work_item
->id
, get_trigger_name(work_item
->trigger
));
693 count
= lttng_dynamic_array_get_count(&work_item
->subitems
);
694 for (i
= 0; i
< count
; i
++) {
695 struct action_work_subitem
*item
;
697 item
= (action_work_subitem
*) lttng_dynamic_array_get_element(&work_item
->subitems
, i
);
698 ret
= action_executor_generic_handler(
699 executor
, work_item
, item
);
705 DBG("Completed execution of action work item %" PRIu64
" of trigger `%s`",
706 work_item
->id
, get_trigger_name(work_item
->trigger
));
710 static void action_work_item_destroy(struct action_work_item
*work_item
)
712 lttng_trigger_put(work_item
->trigger
);
713 lttng_evaluation_destroy(work_item
->evaluation
);
714 notification_client_list_put(work_item
->client_list
);
715 lttng_dynamic_array_reset(&work_item
->subitems
);
719 static void *action_executor_thread(void *_data
)
721 struct action_executor
*executor
= (action_executor
*) _data
;
723 LTTNG_ASSERT(executor
);
725 health_register(the_health_sessiond
,
726 HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR
);
728 rcu_register_thread();
731 DBG("Entering work execution loop");
732 pthread_mutex_lock(&executor
->work
.lock
);
733 while (!executor
->should_quit
) {
735 struct action_work_item
*work_item
;
737 health_code_update();
738 if (executor
->work
.pending_count
== 0) {
740 DBG("No work items enqueued, entering wait");
741 pthread_cond_wait(&executor
->work
.cond
,
742 &executor
->work
.lock
);
743 DBG("Woke-up from wait");
748 /* Pop item from front of the list with work lock held. */
749 work_item
= cds_list_first_entry(&executor
->work
.list
,
750 struct action_work_item
, list_node
);
751 cds_list_del(&work_item
->list_node
);
752 executor
->work
.pending_count
--;
755 * Work can be performed without holding the work lock,
756 * allowing new items to be queued.
758 pthread_mutex_unlock(&executor
->work
.lock
);
760 /* Execute item only if a trigger is registered. */
761 lttng_trigger_lock(work_item
->trigger
);
762 if (!lttng_trigger_is_registered(work_item
->trigger
)) {
763 const char *trigger_name
= NULL
;
764 uid_t trigger_owner_uid
;
765 enum lttng_trigger_status trigger_status
;
767 trigger_name
= get_trigger_name(work_item
->trigger
);
769 trigger_status
= lttng_trigger_get_owner_uid(
770 work_item
->trigger
, &trigger_owner_uid
);
771 LTTNG_ASSERT(trigger_status
== LTTNG_TRIGGER_STATUS_OK
);
773 DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64
", trigger name = `%s`, trigger owner uid = %d",
774 work_item
->id
, trigger_name
,
775 (int) trigger_owner_uid
);
780 ret
= action_work_item_execute(executor
, work_item
);
783 lttng_trigger_unlock(work_item
->trigger
);
784 action_work_item_destroy(work_item
);
790 health_code_update();
791 pthread_mutex_lock(&executor
->work
.lock
);
794 if (executor
->should_quit
) {
795 pthread_mutex_unlock(&executor
->work
.lock
);
797 DBG("Left work execution loop");
799 health_code_update();
801 rcu_thread_offline();
802 rcu_unregister_thread();
803 health_unregister(the_health_sessiond
);
808 static bool shutdown_action_executor_thread(void *_data
)
810 struct action_executor
*executor
= (action_executor
*) _data
;
812 pthread_mutex_lock(&executor
->work
.lock
);
813 executor
->should_quit
= true;
814 pthread_cond_signal(&executor
->work
.cond
);
815 pthread_mutex_unlock(&executor
->work
.lock
);
819 static void clean_up_action_executor_thread(void *_data
)
821 struct action_executor
*executor
= (action_executor
*) _data
;
823 LTTNG_ASSERT(cds_list_empty(&executor
->work
.list
));
825 pthread_mutex_destroy(&executor
->work
.lock
);
826 pthread_cond_destroy(&executor
->work
.cond
);
830 struct action_executor
*action_executor_create(
831 struct notification_thread_handle
*handle
)
833 struct action_executor
*executor
= zmalloc
<action_executor
>();
839 CDS_INIT_LIST_HEAD(&executor
->work
.list
);
840 pthread_cond_init(&executor
->work
.cond
, NULL
);
841 pthread_mutex_init(&executor
->work
.lock
, NULL
);
842 executor
->notification_thread_handle
= handle
;
844 executor
->thread
= lttng_thread_create(THREAD_NAME
,
845 action_executor_thread
, shutdown_action_executor_thread
,
846 clean_up_action_executor_thread
, executor
);
851 void action_executor_destroy(struct action_executor
*executor
)
853 struct action_work_item
*work_item
, *tmp
;
855 /* TODO Wait for work list to drain? */
856 lttng_thread_shutdown(executor
->thread
);
857 pthread_mutex_lock(&executor
->work
.lock
);
858 if (executor
->work
.pending_count
!= 0) {
860 " trigger action%s still queued for execution and will be discarded",
861 executor
->work
.pending_count
,
862 executor
->work
.pending_count
== 1 ? " is" :
866 cds_list_for_each_entry_safe (
867 work_item
, tmp
, &executor
->work
.list
, list_node
) {
868 WARN("Discarding action work item %" PRIu64
869 " associated to trigger `%s`",
870 work_item
->id
, get_trigger_name(work_item
->trigger
));
871 cds_list_del(&work_item
->list_node
);
872 action_work_item_destroy(work_item
);
874 pthread_mutex_unlock(&executor
->work
.lock
);
875 lttng_thread_put(executor
->thread
);
878 /* RCU read-lock must be held by the caller. */
879 enum action_executor_status
action_executor_enqueue_trigger(
880 struct action_executor
*executor
,
881 struct lttng_trigger
*trigger
,
882 struct lttng_evaluation
*evaluation
,
883 const struct lttng_credentials
*object_creds
,
884 struct notification_client_list
*client_list
)
887 enum action_executor_status executor_status
= ACTION_EXECUTOR_STATUS_OK
;
888 const uint64_t work_item_id
= executor
->next_work_item_id
++;
889 struct action_work_item
*work_item
;
892 LTTNG_ASSERT(trigger
);
893 ASSERT_RCU_READ_LOCKED();
895 pthread_mutex_lock(&executor
->work
.lock
);
896 /* Check for queue overflow. */
897 if (executor
->work
.pending_count
>= MAX_QUEUED_WORK_COUNT
) {
898 /* Most likely spammy, remove if it is the case. */
899 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64
,
900 get_trigger_name(trigger
), work_item_id
);
901 executor_status
= ACTION_EXECUTOR_STATUS_OVERFLOW
;
905 work_item
= zmalloc
<action_work_item
>();
907 PERROR("Failed to allocate action executor work item: trigger name = `%s`",
908 get_trigger_name(trigger
));
909 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
913 lttng_trigger_get(trigger
);
915 const bool reference_acquired
=
916 notification_client_list_get(client_list
);
918 LTTNG_ASSERT(reference_acquired
);
921 work_item
->id
= work_item_id
;
922 work_item
->trigger
= trigger
;
924 /* Ownership transferred to the work item. */
925 work_item
->evaluation
= evaluation
;
928 work_item
->client_list
= client_list
;
929 work_item
->object_creds
.is_set
= !!object_creds
;
931 work_item
->object_creds
.value
= *object_creds
;
934 CDS_INIT_LIST_HEAD(&work_item
->list_node
);
936 /* Build the array of action work subitems for the passed trigger. */
937 lttng_dynamic_array_init(&work_item
->subitems
,
938 sizeof(struct action_work_subitem
),
939 action_work_subitem_destructor
);
941 ret
= populate_subitem_array_from_trigger(
942 trigger
, &work_item
->subitems
);
944 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
945 get_trigger_name(trigger
));
946 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
950 cds_list_add_tail(&work_item
->list_node
, &executor
->work
.list
);
951 executor
->work
.pending_count
++;
952 DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64
,
953 get_trigger_name(trigger
), work_item_id
);
958 pthread_cond_signal(&executor
->work
.cond
);
961 pthread_mutex_unlock(&executor
->work
.lock
);
962 lttng_evaluation_destroy(evaluation
);
963 return executor_status
;
966 static int add_action_to_subitem_array(struct lttng_action
*action
,
967 struct lttng_dynamic_array
*subitems
)
970 enum lttng_action_type type
= lttng_action_get_type(action
);
971 const char *session_name
= NULL
;
972 enum lttng_action_status status
;
973 struct action_work_subitem subitem
= {
976 .session_id
= LTTNG_OPTIONAL_INIT_UNSET
,
980 LTTNG_ASSERT(action
);
981 LTTNG_ASSERT(subitems
);
983 if (type
== LTTNG_ACTION_TYPE_LIST
) {
984 unsigned int count
, i
;
986 status
= lttng_action_list_get_count(action
, &count
);
987 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
989 for (i
= 0; i
< count
; i
++) {
990 struct lttng_action
*inner_action
= NULL
;
992 inner_action
= lttng_action_list_borrow_mutable_at_index(
994 LTTNG_ASSERT(inner_action
);
995 ret
= add_action_to_subitem_array(
996 inner_action
, subitems
);
1003 * Go directly to the end since there is no need to add the
1004 * list action by itself to the subitems array.
1009 /* Gather execution context. */
1011 case LTTNG_ACTION_TYPE_NOTIFY
:
1013 case LTTNG_ACTION_TYPE_START_SESSION
:
1014 status
= lttng_action_start_session_get_session_name(
1015 action
, &session_name
);
1016 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1018 case LTTNG_ACTION_TYPE_STOP_SESSION
:
1019 status
= lttng_action_stop_session_get_session_name(
1020 action
, &session_name
);
1021 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1023 case LTTNG_ACTION_TYPE_ROTATE_SESSION
:
1024 status
= lttng_action_rotate_session_get_session_name(
1025 action
, &session_name
);
1026 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1028 case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
:
1029 status
= lttng_action_snapshot_session_get_session_name(
1030 action
, &session_name
);
1031 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1033 case LTTNG_ACTION_TYPE_LIST
:
1034 case LTTNG_ACTION_TYPE_UNKNOWN
:
1042 * Fetch the session execution context info as needed.
1043 * Note that we could decide to not add an action for which we know the
1044 * execution will not happen (i.e no session exists for that name). For
1045 * now we leave the decision to skip to the action executor for sake of
1046 * simplicity and consistency.
1048 if (session_name
!= NULL
) {
1049 uint64_t session_id
;
1052 * Instantaneous sampling of the session id if present.
1054 * This method is preferred over `sessiond_find_by_name` then
1055 * fetching the session'd id since `sessiond_find_by_name`
1056 * requires the session list lock to be taken.
1058 * Taking the session list lock can lead to a deadlock
1059 * between the action executor and the notification thread
1060 * (caller of add_action_to_subitem_array). It is okay if the
1061 * session state changes between the enqueuing time and the
1062 * execution time. The execution context is validated at
1065 if (sample_session_id_by_name(session_name
, &session_id
)) {
1066 LTTNG_OPTIONAL_SET(&subitem
.context
.session_id
,
1071 /* Get a reference to the action. */
1072 lttng_action_get(action
);
1073 subitem
.action
= action
;
1075 ret
= lttng_dynamic_array_add_element(subitems
, &subitem
);
1077 ERR("Failed to add work subitem to the subitem array");
1078 lttng_action_put(action
);
1087 static int populate_subitem_array_from_trigger(struct lttng_trigger
*trigger
,
1088 struct lttng_dynamic_array
*subitems
)
1090 struct lttng_action
*action
;
1092 action
= lttng_trigger_get_action(trigger
);
1093 LTTNG_ASSERT(action
);
1095 return add_action_to_subitem_array(action
, subitems
);