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
39 struct action_executor
{
40 struct lttng_thread
*thread
;
41 struct notification_thread_handle
*notification_thread_handle
;
43 uint64_t pending_count
;
44 struct cds_list_head list
;
49 uint64_t next_work_item_id
;
54 * A work item is composed of a dynamic array of sub-items which
55 * represent a flattened, and augmented, version of a trigger's actions.
57 * We cannot rely solely on the trigger's actions since each action can have an
58 * execution context we need to comply with.
60 * The notion of execution context is required since for some actions the
61 * associated object are referenced by name and not by id. This can lead to
62 * a number of ambiguities when executing an action work item.
64 * For example, let's take a simple trigger such as:
65 * - condition: ust event a
66 * - action: start session S
68 * At time T, session S exists.
69 * At T + 1, the event A is hit.
70 * At T + 2, the tracer event notification is received and the work item is
71 * queued. Here session S have an id of 1.
72 * At T + 3, the session S is destroyed and a new session S is created, with a
73 * resulting id of 200.
74 * At T +4, the work item is popped from the queue and begin execution and will
75 * start session S with an id of 200 instead of the session S id 1 that was
76 * present at the queuing phase.
78 * The context to be respected is the one when the work item is queued. If the
79 * execution context is not the same at the moment of execution, we skip the
80 * execution of that sub-item.
82 * It is the same policy in regards to the validity of the associated
83 * trigger object at the moment of execution, if the trigger is found to be
84 * unregistered, the execution is skipped.
86 struct action_work_item
{
90 * The actions to be executed with their respective execution context.
91 * See struct `action_work_subitem`.
93 struct lttng_dynamic_array subitems
;
95 /* Execution context data */
96 struct lttng_trigger
*trigger
;
97 struct lttng_evaluation
*evaluation
;
98 struct notification_client_list
*client_list
;
99 LTTNG_OPTIONAL(struct lttng_credentials
) object_creds
;
100 struct cds_list_head list_node
;
103 struct action_work_subitem
{
104 struct lttng_action
*action
;
106 /* Used by actions targeting a session. */
107 LTTNG_OPTIONAL(uint64_t) session_id
;
114 * Only return non-zero on a fatal error that should shut down the action
117 typedef int (*action_executor_handler
)(struct action_executor
*executor
,
118 const struct action_work_item
*,
119 struct action_work_subitem
*item
);
121 static int action_executor_notify_handler(struct action_executor
*executor
,
122 const struct action_work_item
*,
123 struct action_work_subitem
*);
124 static int action_executor_start_session_handler(
125 struct action_executor
*executor
,
126 const struct action_work_item
*,
127 struct action_work_subitem
*);
128 static int action_executor_stop_session_handler(
129 struct action_executor
*executor
,
130 const struct action_work_item
*,
131 struct action_work_subitem
*);
132 static int action_executor_rotate_session_handler(
133 struct action_executor
*executor
,
134 const struct action_work_item
*,
135 struct action_work_subitem
*);
136 static int action_executor_snapshot_session_handler(
137 struct action_executor
*executor
,
138 const struct action_work_item
*,
139 struct action_work_subitem
*);
140 static int action_executor_list_handler(struct action_executor
*executor
,
141 const struct action_work_item
*,
142 struct action_work_subitem
*);
143 static int action_executor_generic_handler(struct action_executor
*executor
,
144 const struct action_work_item
*,
145 struct action_work_subitem
*);
147 static const action_executor_handler action_executors
[] = {
148 action_executor_notify_handler
,
149 action_executor_start_session_handler
,
150 action_executor_stop_session_handler
,
151 action_executor_rotate_session_handler
,
152 action_executor_snapshot_session_handler
,
153 action_executor_list_handler
,
156 /* Forward declaration */
157 static int add_action_to_subitem_array(struct lttng_action
*action
,
158 struct lttng_dynamic_array
*subitems
);
160 static int populate_subitem_array_from_trigger(struct lttng_trigger
*trigger
,
161 struct lttng_dynamic_array
*subitems
);
163 static void action_work_subitem_destructor(void *element
)
165 struct action_work_subitem
*subitem
= (action_work_subitem
*) element
;
167 lttng_action_put(subitem
->action
);
170 static const char *get_action_name(const struct lttng_action
*action
)
172 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
174 LTTNG_ASSERT(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
176 return lttng_action_type_string(action_type
);
179 /* Check if this trigger allowed to interect with a given session. */
180 static bool is_trigger_allowed_for_session(const struct lttng_trigger
*trigger
,
181 struct ltt_session
*session
)
183 bool is_allowed
= false;
184 const struct lttng_credentials session_creds
= {
185 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
->uid
),
186 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
->gid
),
188 /* Can never be NULL. */
189 const struct lttng_credentials
*trigger_creds
=
190 lttng_trigger_get_credentials(trigger
);
192 is_allowed
= (lttng_credentials_is_equal_uid(trigger_creds
, &session_creds
)) ||
193 (lttng_credentials_get_uid(trigger_creds
) == 0);
195 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
197 (long int) session
->uid
,
198 (long int) session
->gid
,
199 (long int) lttng_credentials_get_uid(trigger_creds
));
205 static const char *get_trigger_name(const struct lttng_trigger
*trigger
)
207 const char *trigger_name
;
208 enum lttng_trigger_status trigger_status
;
210 trigger_status
= lttng_trigger_get_name(trigger
, &trigger_name
);
211 switch (trigger_status
) {
212 case LTTNG_TRIGGER_STATUS_OK
:
214 case LTTNG_TRIGGER_STATUS_UNSET
:
215 trigger_name
= "(anonymous)";
218 trigger_name
= "(failed to get name)";
225 static int client_handle_transmission_status(
226 struct notification_client
*client
,
227 enum client_transmission_status status
,
231 struct action_executor
*executor
= (action_executor
*) user_data
;
232 bool update_communication
= true;
235 case CLIENT_TRANSMISSION_STATUS_COMPLETE
:
236 DBG("Successfully sent full notification to client, client_id = %" PRIu64
,
239 * There is no need to wake the (e)poll thread. If it was waiting for
240 * "out" events on the client's socket, it will see that no payload
241 * in queued and will unsubscribe from that event.
243 * In the other cases, we have to wake the the (e)poll thread to either
244 * handle the error on the client or to get it to monitor the client "out"
247 update_communication
= false;
249 case CLIENT_TRANSMISSION_STATUS_QUEUED
:
250 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64
,
253 case CLIENT_TRANSMISSION_STATUS_FAIL
:
254 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64
,
258 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64
,
264 if (!update_communication
) {
268 /* Safe to read client's id without locking as it is immutable. */
269 ret
= notification_thread_client_communication_update(
270 executor
->notification_thread_handle
, client
->id
,
276 static int action_executor_notify_handler(struct action_executor
*executor
,
277 const struct action_work_item
*work_item
,
278 struct action_work_subitem
*item
__attribute__((unused
)))
280 return notification_client_list_send_evaluation(work_item
->client_list
,
282 work_item
->evaluation
,
283 work_item
->object_creds
.is_set
?
284 &(work_item
->object_creds
.value
) :
286 client_handle_transmission_status
, executor
);
289 static int action_executor_start_session_handler(
290 struct action_executor
*executor
__attribute__((unused
)),
291 const struct action_work_item
*work_item
,
292 struct action_work_subitem
*item
)
295 const char *session_name
;
296 enum lttng_action_status action_status
;
297 struct ltt_session
*session
;
298 enum lttng_error_code cmd_ret
;
299 struct lttng_action
*action
= item
->action
;
301 action_status
= lttng_action_start_session_get_session_name(
302 action
, &session_name
);
303 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
304 ERR("Failed to get session name from `%s` action",
305 get_action_name(action
));
311 * Validate if at the moment of the action was queued the session
312 * existed. If not skip the action altogether.
314 if (!item
->context
.session_id
.is_set
) {
315 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
316 session_name
, get_action_name(action
),
317 get_trigger_name(work_item
->trigger
));
318 lttng_action_increase_execution_failure_count(action
);
324 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
326 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
327 session_name
, get_action_name(action
),
328 get_trigger_name(work_item
->trigger
));
329 lttng_action_increase_execution_failure_count(action
);
330 goto error_unlock_list
;
333 session_lock(session
);
334 if (session
->destroyed
) {
335 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
336 session
->name
, session
->id
,
337 get_action_name(action
),
338 get_trigger_name(work_item
->trigger
));
339 goto error_unlock_session
;
342 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
343 goto error_unlock_session
;
346 cmd_ret
= (lttng_error_code
) cmd_start_trace(session
);
349 DBG("Successfully started session `%s` on behalf of trigger `%s`",
350 session_name
, get_trigger_name(work_item
->trigger
));
352 case LTTNG_ERR_TRACE_ALREADY_STARTED
:
353 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
354 session_name
, get_trigger_name(work_item
->trigger
));
357 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
358 session_name
, get_trigger_name(work_item
->trigger
),
359 lttng_strerror(-cmd_ret
));
360 lttng_action_increase_execution_failure_count(action
);
364 error_unlock_session
:
365 session_unlock(session
);
366 session_put(session
);
369 session_unlock_list();
374 static int action_executor_stop_session_handler(
375 struct action_executor
*executor
__attribute__((unused
)),
376 const struct action_work_item
*work_item
,
377 struct action_work_subitem
*item
)
380 const char *session_name
;
381 enum lttng_action_status action_status
;
382 struct ltt_session
*session
;
383 enum lttng_error_code cmd_ret
;
384 struct lttng_action
*action
= item
->action
;
386 action_status
= lttng_action_stop_session_get_session_name(
387 action
, &session_name
);
388 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
389 ERR("Failed to get session name from `%s` action",
390 get_action_name(action
));
396 * Validate if, at the moment the action was queued, the target session
397 * existed. If not, skip the action altogether.
399 if (!item
->context
.session_id
.is_set
) {
400 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
401 session_name
, get_action_name(action
),
402 get_trigger_name(work_item
->trigger
));
403 lttng_action_increase_execution_failure_count(action
);
409 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
411 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
412 session_name
, get_action_name(action
),
413 get_trigger_name(work_item
->trigger
));
414 lttng_action_increase_execution_failure_count(action
);
415 goto error_unlock_list
;
418 session_lock(session
);
419 if (session
->destroyed
) {
420 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
421 session
->name
, session
->id
,
422 get_action_name(action
),
423 get_trigger_name(work_item
->trigger
));
424 goto error_unlock_session
;
427 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
428 goto error_unlock_session
;
431 cmd_ret
= (lttng_error_code
) cmd_stop_trace(session
);
434 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
435 session_name
, get_trigger_name(work_item
->trigger
));
437 case LTTNG_ERR_TRACE_ALREADY_STOPPED
:
438 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
439 session_name
, get_trigger_name(work_item
->trigger
));
442 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
443 session_name
, get_trigger_name(work_item
->trigger
),
444 lttng_strerror(-cmd_ret
));
445 lttng_action_increase_execution_failure_count(action
);
449 error_unlock_session
:
450 session_unlock(session
);
451 session_put(session
);
454 session_unlock_list();
459 static int action_executor_rotate_session_handler(
460 struct action_executor
*executor
__attribute__((unused
)),
461 const struct action_work_item
*work_item
,
462 struct action_work_subitem
*item
)
465 const char *session_name
;
466 enum lttng_action_status action_status
;
467 struct ltt_session
*session
;
468 enum lttng_error_code cmd_ret
;
469 struct lttng_action
*action
= item
->action
;
471 action_status
= lttng_action_rotate_session_get_session_name(
472 action
, &session_name
);
473 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
474 ERR("Failed to get session name from `%s` action",
475 get_action_name(action
));
481 * Validate if, at the moment the action was queued, the target session
482 * existed. If not, skip the action altogether.
484 if (!item
->context
.session_id
.is_set
) {
485 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
486 session_name
, get_action_name(action
),
487 get_trigger_name(work_item
->trigger
));
488 lttng_action_increase_execution_failure_count(action
);
494 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
496 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
497 session_name
, get_action_name(action
),
498 get_trigger_name(work_item
->trigger
));
499 lttng_action_increase_execution_failure_count(action
);
500 goto error_unlock_list
;
503 session_lock(session
);
504 if (session
->destroyed
) {
505 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
506 session
->name
, session
->id
,
507 get_action_name(action
),
508 get_trigger_name(work_item
->trigger
));
509 goto error_unlock_session
;
512 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
513 goto error_unlock_session
;
516 cmd_ret
= (lttng_error_code
) cmd_rotate_session(session
, NULL
, false,
517 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
520 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
521 session_name
, get_trigger_name(work_item
->trigger
));
523 case LTTNG_ERR_ROTATION_PENDING
:
524 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
525 session_name
, get_trigger_name(work_item
->trigger
));
526 lttng_action_increase_execution_failure_count(action
);
528 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
529 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
530 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",
531 session_name
, get_trigger_name(work_item
->trigger
));
534 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
535 session_name
, get_trigger_name(work_item
->trigger
),
536 lttng_strerror(-cmd_ret
));
537 lttng_action_increase_execution_failure_count(action
);
541 error_unlock_session
:
542 session_unlock(session
);
543 session_put(session
);
546 session_unlock_list();
551 static int action_executor_snapshot_session_handler(
552 struct action_executor
*executor
__attribute__((unused
)),
553 const struct action_work_item
*work_item
,
554 struct action_work_subitem
*item
)
557 const char *session_name
;
558 enum lttng_action_status action_status
;
559 struct ltt_session
*session
;
560 lttng_snapshot_output default_snapshot_output
;
561 const struct lttng_snapshot_output
*snapshot_output
=
562 &default_snapshot_output
;
563 enum lttng_error_code cmd_ret
;
564 struct lttng_action
*action
= item
->action
;
566 default_snapshot_output
.max_size
= UINT64_MAX
;
569 * Validate if, at the moment the action was queued, the target session
570 * existed. If not, skip the action altogether.
572 if (!item
->context
.session_id
.is_set
) {
573 DBG("Session was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
574 get_action_name(action
),
575 get_trigger_name(work_item
->trigger
));
576 lttng_action_increase_execution_failure_count(action
);
580 action_status
= lttng_action_snapshot_session_get_session_name(
581 action
, &session_name
);
582 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
583 ERR("Failed to get session name from `%s` action",
584 get_action_name(action
));
589 action_status
= lttng_action_snapshot_session_get_output(
590 action
, &snapshot_output
);
591 if (action_status
!= LTTNG_ACTION_STATUS_OK
&&
592 action_status
!= LTTNG_ACTION_STATUS_UNSET
) {
593 ERR("Failed to get output from `%s` action",
594 get_action_name(action
));
601 session
= session_find_by_id(LTTNG_OPTIONAL_GET(item
->context
.session_id
));
603 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
604 session_name
, get_action_name(action
),
605 get_trigger_name(work_item
->trigger
));
606 lttng_action_increase_execution_failure_count(action
);
607 goto error_unlock_list
;
610 session_lock(session
);
611 if (session
->destroyed
) {
612 DBG("Session `%s` with id = %" PRIu64
" is flagged as destroyed. Skipping: action = `%s`, trigger = `%s`",
613 session
->name
, session
->id
,
614 get_action_name(action
),
615 get_trigger_name(work_item
->trigger
));
616 goto error_unlock_session
;
619 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
620 goto error_unlock_session
;
623 cmd_ret
= (lttng_error_code
) cmd_snapshot_record(session
, snapshot_output
, 0);
626 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
627 session_name
, get_trigger_name(work_item
->trigger
));
630 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
631 session_name
, get_trigger_name(work_item
->trigger
),
632 lttng_strerror(-cmd_ret
));
633 lttng_action_increase_execution_failure_count(action
);
637 error_unlock_session
:
638 session_unlock(session
);
639 session_put(session
);
642 session_unlock_list();
647 static int action_executor_list_handler(
648 struct action_executor
*executor
__attribute__((unused
)),
649 const struct action_work_item
*work_item
__attribute__((unused
)),
650 struct action_work_subitem
*item
__attribute__((unused
)))
652 ERR("Execution of a list action by the action executor should never occur");
656 static int action_executor_generic_handler(struct action_executor
*executor
,
657 const struct action_work_item
*work_item
,
658 struct action_work_subitem
*item
)
661 struct lttng_action
*action
= item
->action
;
662 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
664 LTTNG_ASSERT(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
666 lttng_action_increase_execution_request_count(action
);
667 if (!lttng_action_should_execute(action
)) {
668 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64
,
669 get_action_name(action
),
670 get_trigger_name(work_item
->trigger
),
676 lttng_action_increase_execution_count(action
);
677 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64
,
678 get_action_name(action
),
679 get_trigger_name(work_item
->trigger
),
681 ret
= action_executors
[action_type
](executor
, work_item
, item
);
686 static int action_work_item_execute(struct action_executor
*executor
,
687 struct action_work_item
*work_item
)
692 DBG("Starting execution of action work item %" PRIu64
" of trigger `%s`",
693 work_item
->id
, get_trigger_name(work_item
->trigger
));
695 count
= lttng_dynamic_array_get_count(&work_item
->subitems
);
696 for (i
= 0; i
< count
; i
++) {
697 struct action_work_subitem
*item
;
699 item
= (action_work_subitem
*) lttng_dynamic_array_get_element(&work_item
->subitems
, i
);
700 ret
= action_executor_generic_handler(
701 executor
, work_item
, item
);
707 DBG("Completed execution of action work item %" PRIu64
" of trigger `%s`",
708 work_item
->id
, get_trigger_name(work_item
->trigger
));
712 static void action_work_item_destroy(struct action_work_item
*work_item
)
714 lttng_trigger_put(work_item
->trigger
);
715 lttng_evaluation_destroy(work_item
->evaluation
);
716 notification_client_list_put(work_item
->client_list
);
717 lttng_dynamic_array_reset(&work_item
->subitems
);
721 static void *action_executor_thread(void *_data
)
723 struct action_executor
*executor
= (action_executor
*) _data
;
725 LTTNG_ASSERT(executor
);
727 health_register(the_health_sessiond
,
728 HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR
);
730 rcu_register_thread();
733 DBG("Entering work execution loop");
734 pthread_mutex_lock(&executor
->work
.lock
);
735 while (!executor
->should_quit
) {
737 struct action_work_item
*work_item
;
739 health_code_update();
740 if (executor
->work
.pending_count
== 0) {
742 DBG("No work items enqueued, entering wait");
743 pthread_cond_wait(&executor
->work
.cond
,
744 &executor
->work
.lock
);
745 DBG("Woke-up from wait");
750 /* Pop item from front of the list with work lock held. */
751 work_item
= cds_list_first_entry(&executor
->work
.list
,
752 struct action_work_item
, list_node
);
753 cds_list_del(&work_item
->list_node
);
754 executor
->work
.pending_count
--;
757 * Work can be performed without holding the work lock,
758 * allowing new items to be queued.
760 pthread_mutex_unlock(&executor
->work
.lock
);
762 /* Execute item only if a trigger is registered. */
763 lttng_trigger_lock(work_item
->trigger
);
764 if (!lttng_trigger_is_registered(work_item
->trigger
)) {
765 const char *trigger_name
= NULL
;
766 uid_t trigger_owner_uid
;
767 enum lttng_trigger_status trigger_status
;
769 trigger_name
= get_trigger_name(work_item
->trigger
);
771 trigger_status
= lttng_trigger_get_owner_uid(
772 work_item
->trigger
, &trigger_owner_uid
);
773 LTTNG_ASSERT(trigger_status
== LTTNG_TRIGGER_STATUS_OK
);
775 DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64
", trigger name = `%s`, trigger owner uid = %d",
776 work_item
->id
, trigger_name
,
777 (int) trigger_owner_uid
);
782 ret
= action_work_item_execute(executor
, work_item
);
785 lttng_trigger_unlock(work_item
->trigger
);
786 action_work_item_destroy(work_item
);
792 health_code_update();
793 pthread_mutex_lock(&executor
->work
.lock
);
796 if (executor
->should_quit
) {
797 pthread_mutex_unlock(&executor
->work
.lock
);
799 DBG("Left work execution loop");
801 health_code_update();
803 rcu_thread_offline();
804 rcu_unregister_thread();
805 health_unregister(the_health_sessiond
);
810 static bool shutdown_action_executor_thread(void *_data
)
812 struct action_executor
*executor
= (action_executor
*) _data
;
814 pthread_mutex_lock(&executor
->work
.lock
);
815 executor
->should_quit
= true;
816 pthread_cond_signal(&executor
->work
.cond
);
817 pthread_mutex_unlock(&executor
->work
.lock
);
821 static void clean_up_action_executor_thread(void *_data
)
823 struct action_executor
*executor
= (action_executor
*) _data
;
825 LTTNG_ASSERT(cds_list_empty(&executor
->work
.list
));
827 pthread_mutex_destroy(&executor
->work
.lock
);
828 pthread_cond_destroy(&executor
->work
.cond
);
832 struct action_executor
*action_executor_create(
833 struct notification_thread_handle
*handle
)
835 struct action_executor
*executor
= zmalloc
<action_executor
>();
841 CDS_INIT_LIST_HEAD(&executor
->work
.list
);
842 pthread_cond_init(&executor
->work
.cond
, NULL
);
843 pthread_mutex_init(&executor
->work
.lock
, NULL
);
844 executor
->notification_thread_handle
= handle
;
846 executor
->thread
= lttng_thread_create(THREAD_NAME
,
847 action_executor_thread
, shutdown_action_executor_thread
,
848 clean_up_action_executor_thread
, executor
);
853 void action_executor_destroy(struct action_executor
*executor
)
855 struct action_work_item
*work_item
, *tmp
;
857 /* TODO Wait for work list to drain? */
858 lttng_thread_shutdown(executor
->thread
);
859 pthread_mutex_lock(&executor
->work
.lock
);
860 if (executor
->work
.pending_count
!= 0) {
862 " trigger action%s still queued for execution and will be discarded",
863 executor
->work
.pending_count
,
864 executor
->work
.pending_count
== 1 ? " is" :
868 cds_list_for_each_entry_safe (
869 work_item
, tmp
, &executor
->work
.list
, list_node
) {
870 WARN("Discarding action work item %" PRIu64
871 " associated to trigger `%s`",
872 work_item
->id
, get_trigger_name(work_item
->trigger
));
873 cds_list_del(&work_item
->list_node
);
874 action_work_item_destroy(work_item
);
876 pthread_mutex_unlock(&executor
->work
.lock
);
877 lttng_thread_put(executor
->thread
);
880 /* RCU read-lock must be held by the caller. */
881 enum action_executor_status
action_executor_enqueue_trigger(
882 struct action_executor
*executor
,
883 struct lttng_trigger
*trigger
,
884 struct lttng_evaluation
*evaluation
,
885 const struct lttng_credentials
*object_creds
,
886 struct notification_client_list
*client_list
)
889 enum action_executor_status executor_status
= ACTION_EXECUTOR_STATUS_OK
;
890 const uint64_t work_item_id
= executor
->next_work_item_id
++;
891 struct action_work_item
*work_item
;
894 LTTNG_ASSERT(trigger
);
895 ASSERT_RCU_READ_LOCKED();
897 pthread_mutex_lock(&executor
->work
.lock
);
898 /* Check for queue overflow. */
899 if (executor
->work
.pending_count
>= MAX_QUEUED_WORK_COUNT
) {
900 /* Most likely spammy, remove if it is the case. */
901 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64
,
902 get_trigger_name(trigger
), work_item_id
);
903 executor_status
= ACTION_EXECUTOR_STATUS_OVERFLOW
;
907 work_item
= zmalloc
<action_work_item
>();
909 PERROR("Failed to allocate action executor work item: trigger name = `%s`",
910 get_trigger_name(trigger
));
911 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
915 lttng_trigger_get(trigger
);
917 const bool reference_acquired
=
918 notification_client_list_get(client_list
);
920 LTTNG_ASSERT(reference_acquired
);
923 work_item
->id
= work_item_id
;
924 work_item
->trigger
= trigger
;
926 /* Ownership transferred to the work item. */
927 work_item
->evaluation
= evaluation
;
930 work_item
->client_list
= client_list
;
931 work_item
->object_creds
.is_set
= !!object_creds
;
933 work_item
->object_creds
.value
= *object_creds
;
936 CDS_INIT_LIST_HEAD(&work_item
->list_node
);
938 /* Build the array of action work subitems for the passed trigger. */
939 lttng_dynamic_array_init(&work_item
->subitems
,
940 sizeof(struct action_work_subitem
),
941 action_work_subitem_destructor
);
943 ret
= populate_subitem_array_from_trigger(
944 trigger
, &work_item
->subitems
);
946 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
947 get_trigger_name(trigger
));
948 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
952 cds_list_add_tail(&work_item
->list_node
, &executor
->work
.list
);
953 executor
->work
.pending_count
++;
954 DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64
,
955 get_trigger_name(trigger
), work_item_id
);
960 pthread_cond_signal(&executor
->work
.cond
);
963 pthread_mutex_unlock(&executor
->work
.lock
);
964 lttng_evaluation_destroy(evaluation
);
965 return executor_status
;
968 static int add_action_to_subitem_array(struct lttng_action
*action
,
969 struct lttng_dynamic_array
*subitems
)
972 enum lttng_action_type type
= lttng_action_get_type(action
);
973 const char *session_name
= NULL
;
974 enum lttng_action_status status
;
975 struct action_work_subitem subitem
= {
978 .session_id
= LTTNG_OPTIONAL_INIT_UNSET
,
982 LTTNG_ASSERT(action
);
983 LTTNG_ASSERT(subitems
);
985 if (type
== LTTNG_ACTION_TYPE_LIST
) {
986 unsigned int count
, i
;
988 status
= lttng_action_list_get_count(action
, &count
);
989 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
991 for (i
= 0; i
< count
; i
++) {
992 struct lttng_action
*inner_action
= NULL
;
994 inner_action
= lttng_action_list_borrow_mutable_at_index(
996 LTTNG_ASSERT(inner_action
);
997 ret
= add_action_to_subitem_array(
998 inner_action
, subitems
);
1005 * Go directly to the end since there is no need to add the
1006 * list action by itself to the subitems array.
1011 /* Gather execution context. */
1013 case LTTNG_ACTION_TYPE_NOTIFY
:
1015 case LTTNG_ACTION_TYPE_START_SESSION
:
1016 status
= lttng_action_start_session_get_session_name(
1017 action
, &session_name
);
1018 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1020 case LTTNG_ACTION_TYPE_STOP_SESSION
:
1021 status
= lttng_action_stop_session_get_session_name(
1022 action
, &session_name
);
1023 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1025 case LTTNG_ACTION_TYPE_ROTATE_SESSION
:
1026 status
= lttng_action_rotate_session_get_session_name(
1027 action
, &session_name
);
1028 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1030 case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
:
1031 status
= lttng_action_snapshot_session_get_session_name(
1032 action
, &session_name
);
1033 LTTNG_ASSERT(status
== LTTNG_ACTION_STATUS_OK
);
1035 case LTTNG_ACTION_TYPE_LIST
:
1036 case LTTNG_ACTION_TYPE_UNKNOWN
:
1044 * Fetch the session execution context info as needed.
1045 * Note that we could decide to not add an action for which we know the
1046 * execution will not happen (i.e no session exists for that name). For
1047 * now we leave the decision to skip to the action executor for sake of
1048 * simplicity and consistency.
1050 if (session_name
!= NULL
) {
1051 uint64_t session_id
;
1054 * Instantaneous sampling of the session id if present.
1056 * This method is preferred over `sessiond_find_by_name` then
1057 * fetching the session'd id since `sessiond_find_by_name`
1058 * requires the session list lock to be taken.
1060 * Taking the session list lock can lead to a deadlock
1061 * between the action executor and the notification thread
1062 * (caller of add_action_to_subitem_array). It is okay if the
1063 * session state changes between the enqueuing time and the
1064 * execution time. The execution context is validated at
1067 if (sample_session_id_by_name(session_name
, &session_id
)) {
1068 LTTNG_OPTIONAL_SET(&subitem
.context
.session_id
,
1073 /* Get a reference to the action. */
1074 lttng_action_get(action
);
1075 subitem
.action
= action
;
1077 ret
= lttng_dynamic_array_add_element(subitems
, &subitem
);
1079 ERR("Failed to add work subitem to the subitem array");
1080 lttng_action_put(action
);
1089 static int populate_subitem_array_from_trigger(struct lttng_trigger
*trigger
,
1090 struct lttng_dynamic_array
*subitems
)
1092 struct lttng_action
*action
;
1094 action
= lttng_trigger_get_action(trigger
);
1095 LTTNG_ASSERT(action
);
1097 return add_action_to_subitem_array(action
, subitems
);