2 * Copyright (C) 2020 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 * SPDX-License-Identifier: GPL-2.0-only
8 #include "action-executor.h"
10 #include "health-sessiond.h"
11 #include "lttng-sessiond.h"
12 #include "notification-thread-internal.h"
15 #include <common/dynamic-array.h>
16 #include <common/macros.h>
17 #include <common/optional.h>
18 #include <lttng/action/action-internal.h>
19 #include <lttng/action/list-internal.h>
20 #include <lttng/action/list.h>
21 #include <lttng/action/notify-internal.h>
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.h>
29 #include <lttng/lttng-error.h>
30 #include <lttng/trigger/trigger-internal.h>
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 [LTTNG_ACTION_TYPE_NOTIFY
] = action_executor_notify_handler
,
147 [LTTNG_ACTION_TYPE_START_SESSION
] = action_executor_start_session_handler
,
148 [LTTNG_ACTION_TYPE_STOP_SESSION
] = action_executor_stop_session_handler
,
149 [LTTNG_ACTION_TYPE_ROTATE_SESSION
] = action_executor_rotate_session_handler
,
150 [LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
] = action_executor_snapshot_session_handler
,
151 [LTTNG_ACTION_TYPE_LIST
] = 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
= 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 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
= 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
)
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
,
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_name(session_name
);
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 goto error_unlock_list
;
331 * Check if the session id is the same as when the work item was
334 if (session
->id
!= LTTNG_OPTIONAL_GET(item
->context
.session_id
)) {
335 DBG("Session id for session `%s` (id: %" PRIu64
336 " is not the same that was sampled (id: %" PRIu64
337 " at the moment the work item was enqueued for `%s` action of trigger `%s`",
338 session_name
, session
->id
,
339 LTTNG_OPTIONAL_GET(item
->context
.session_id
),
340 get_action_name(action
),
341 get_trigger_name(work_item
->trigger
));
343 goto error_put_session
;
346 session_lock(session
);
347 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
348 goto error_unlock_session
;
351 cmd_ret
= cmd_start_trace(session
);
354 DBG("Successfully started session `%s` on behalf of trigger `%s`",
355 session_name
, get_trigger_name(work_item
->trigger
));
357 case LTTNG_ERR_TRACE_ALREADY_STARTED
:
358 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
359 session_name
, get_trigger_name(work_item
->trigger
));
362 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
363 session_name
, get_trigger_name(work_item
->trigger
),
364 lttng_strerror(-cmd_ret
));
365 lttng_action_increase_execution_failure_count(action
);
369 error_unlock_session
:
370 session_unlock(session
);
372 session_put(session
);
374 session_unlock_list();
379 static int action_executor_stop_session_handler(
380 struct action_executor
*executor
,
381 const struct action_work_item
*work_item
,
382 struct action_work_subitem
*item
)
385 const char *session_name
;
386 enum lttng_action_status action_status
;
387 struct ltt_session
*session
;
388 enum lttng_error_code cmd_ret
;
389 struct lttng_action
*action
= item
->action
;
391 action_status
= lttng_action_stop_session_get_session_name(
392 action
, &session_name
);
393 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
394 ERR("Failed to get session name from `%s` action",
395 get_action_name(action
));
401 * Validate if, at the moment the action was queued, the target session
402 * existed. If not, skip the action altogether.
404 if (!item
->context
.session_id
.is_set
) {
405 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
406 session_name
, get_action_name(action
),
407 get_trigger_name(work_item
->trigger
));
408 lttng_action_increase_execution_failure_count(action
);
414 session
= session_find_by_name(session_name
);
416 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
417 session_name
, get_action_name(action
),
418 get_trigger_name(work_item
->trigger
));
419 lttng_action_increase_execution_failure_count(action
);
420 goto error_unlock_list
;
424 * Check if the session id is the same as when the work item was
427 if (session
->id
!= LTTNG_OPTIONAL_GET(item
->context
.session_id
)) {
428 DBG("Session id for session `%s` (id: %" PRIu64
429 " is not the same that was sampled (id: %" PRIu64
430 " at the moment the work item was enqueued for `%s` action of trigger `%s`",
431 session_name
, session
->id
,
432 LTTNG_OPTIONAL_GET(item
->context
.session_id
),
433 get_action_name(action
),
434 get_trigger_name(work_item
->trigger
));
436 goto error_put_session
;
439 session_lock(session
);
440 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
441 goto error_unlock_session
;
444 cmd_ret
= cmd_stop_trace(session
);
447 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
448 session_name
, get_trigger_name(work_item
->trigger
));
450 case LTTNG_ERR_TRACE_ALREADY_STOPPED
:
451 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
452 session_name
, get_trigger_name(work_item
->trigger
));
455 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
456 session_name
, get_trigger_name(work_item
->trigger
),
457 lttng_strerror(-cmd_ret
));
458 lttng_action_increase_execution_failure_count(action
);
462 error_unlock_session
:
463 session_unlock(session
);
465 session_put(session
);
467 session_unlock_list();
472 static int action_executor_rotate_session_handler(
473 struct action_executor
*executor
,
474 const struct action_work_item
*work_item
,
475 struct action_work_subitem
*item
)
478 const char *session_name
;
479 enum lttng_action_status action_status
;
480 struct ltt_session
*session
;
481 enum lttng_error_code cmd_ret
;
482 struct lttng_action
*action
= item
->action
;
484 action_status
= lttng_action_rotate_session_get_session_name(
485 action
, &session_name
);
486 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
487 ERR("Failed to get session name from `%s` action",
488 get_action_name(action
));
494 * Validate if, at the moment the action was queued, the target session
495 * existed. If not, skip the action altogether.
497 if (!item
->context
.session_id
.is_set
) {
498 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
499 session_name
, get_action_name(action
),
500 get_trigger_name(work_item
->trigger
));
501 lttng_action_increase_execution_failure_count(action
);
507 session
= session_find_by_name(session_name
);
509 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
510 session_name
, get_action_name(action
),
511 get_trigger_name(work_item
->trigger
));
512 lttng_action_increase_execution_failure_count(action
);
513 goto error_unlock_list
;
517 * Check if the session id is the same as when the work item was
520 if (session
->id
!= LTTNG_OPTIONAL_GET(item
->context
.session_id
)) {
521 DBG("Session id for session `%s` (id: %" PRIu64
522 " is not the same that was sampled (id: %" PRIu64
523 " at the moment the work item was enqueued for `%s` action of trigger `%s`",
524 session_name
, session
->id
,
525 LTTNG_OPTIONAL_GET(item
->context
.session_id
),
526 get_action_name(action
),
527 get_trigger_name(work_item
->trigger
));
529 goto error_put_session
;
532 session_lock(session
);
533 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
534 goto error_unlock_session
;
537 cmd_ret
= cmd_rotate_session(session
, NULL
, false,
538 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
541 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
542 session_name
, get_trigger_name(work_item
->trigger
));
544 case LTTNG_ERR_ROTATION_PENDING
:
545 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
546 session_name
, get_trigger_name(work_item
->trigger
));
547 lttng_action_increase_execution_failure_count(action
);
549 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
550 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
551 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",
552 session_name
, get_trigger_name(work_item
->trigger
));
555 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
556 session_name
, get_trigger_name(work_item
->trigger
),
557 lttng_strerror(-cmd_ret
));
558 lttng_action_increase_execution_failure_count(action
);
562 error_unlock_session
:
563 session_unlock(session
);
565 session_put(session
);
567 session_unlock_list();
572 static int action_executor_snapshot_session_handler(
573 struct action_executor
*executor
,
574 const struct action_work_item
*work_item
,
575 struct action_work_subitem
*item
)
578 const char *session_name
;
579 enum lttng_action_status action_status
;
580 struct ltt_session
*session
;
581 const struct lttng_snapshot_output default_snapshot_output
= {
582 .max_size
= UINT64_MAX
,
584 const struct lttng_snapshot_output
*snapshot_output
=
585 &default_snapshot_output
;
586 enum lttng_error_code cmd_ret
;
587 struct lttng_action
*action
= item
->action
;
590 * Validate if, at the moment the action was queued, the target session
591 * existed. If not, skip the action altogether.
593 if (!item
->context
.session_id
.is_set
) {
594 DBG("Session was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
595 get_action_name(action
),
596 get_trigger_name(work_item
->trigger
));
597 lttng_action_increase_execution_failure_count(action
);
602 action_status
= lttng_action_snapshot_session_get_session_name(
603 action
, &session_name
);
604 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
605 ERR("Failed to get session name from `%s` action",
606 get_action_name(action
));
611 action_status
= lttng_action_snapshot_session_get_output(
612 action
, &snapshot_output
);
613 if (action_status
!= LTTNG_ACTION_STATUS_OK
&&
614 action_status
!= LTTNG_ACTION_STATUS_UNSET
) {
615 ERR("Failed to get output from `%s` action",
616 get_action_name(action
));
622 session
= session_find_by_name(session_name
);
624 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
625 session_name
, get_action_name(action
),
626 get_trigger_name(work_item
->trigger
));
627 lttng_action_increase_execution_failure_count(action
);
628 goto error_unlock_list
;
632 * Check if the session id is the same as when the work item was
635 if (session
->id
!= LTTNG_OPTIONAL_GET(item
->context
.session_id
)) {
636 DBG("Session id for session `%s` (id: %" PRIu64
637 " is not the same that was sampled (id: %" PRIu64
638 " at the moment the work item was enqueued for `%s` action of trigger `%s`",
639 session_name
, session
->id
,
640 LTTNG_OPTIONAL_GET(item
->context
.session_id
),
641 get_action_name(action
),
642 get_trigger_name(work_item
->trigger
));
644 goto error_put_session
;
647 session_lock(session
);
648 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
649 goto error_unlock_session
;
652 cmd_ret
= cmd_snapshot_record(session
, snapshot_output
, 0);
655 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
656 session_name
, get_trigger_name(work_item
->trigger
));
659 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
660 session_name
, get_trigger_name(work_item
->trigger
),
661 lttng_strerror(-cmd_ret
));
662 lttng_action_increase_execution_failure_count(action
);
666 error_unlock_session
:
667 session_unlock(session
);
669 session_put(session
);
671 session_unlock_list();
676 static int action_executor_list_handler(struct action_executor
*executor
,
677 const struct action_work_item
*work_item
,
678 struct action_work_subitem
*item
)
680 ERR("Execution of a list action by the action executor should never occur");
684 static int action_executor_generic_handler(struct action_executor
*executor
,
685 const struct action_work_item
*work_item
,
686 struct action_work_subitem
*item
)
689 struct lttng_action
*action
= item
->action
;
690 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
692 assert(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
694 lttng_action_increase_execution_request_count(action
);
695 if (!lttng_action_should_execute(action
)) {
696 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64
,
697 get_action_name(action
),
698 get_trigger_name(work_item
->trigger
),
704 lttng_action_increase_execution_count(action
);
705 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64
,
706 get_action_name(action
),
707 get_trigger_name(work_item
->trigger
),
709 ret
= action_executors
[action_type
](executor
, work_item
, item
);
714 static int action_work_item_execute(struct action_executor
*executor
,
715 struct action_work_item
*work_item
)
720 DBG("Starting execution of action work item %" PRIu64
" of trigger `%s`",
721 work_item
->id
, get_trigger_name(work_item
->trigger
));
723 count
= lttng_dynamic_array_get_count(&work_item
->subitems
);
724 for (i
= 0; i
< count
; i
++) {
725 struct action_work_subitem
*item
;
727 item
= lttng_dynamic_array_get_element(&work_item
->subitems
, i
);
728 ret
= action_executor_generic_handler(
729 executor
, work_item
, item
);
735 DBG("Completed execution of action work item %" PRIu64
" of trigger `%s`",
736 work_item
->id
, get_trigger_name(work_item
->trigger
));
740 static void action_work_item_destroy(struct action_work_item
*work_item
)
742 lttng_trigger_put(work_item
->trigger
);
743 lttng_evaluation_destroy(work_item
->evaluation
);
744 notification_client_list_put(work_item
->client_list
);
745 lttng_dynamic_array_reset(&work_item
->subitems
);
749 static void *action_executor_thread(void *_data
)
751 struct action_executor
*executor
= _data
;
755 health_register(the_health_sessiond
,
756 HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR
);
758 rcu_register_thread();
761 DBG("Entering work execution loop");
762 pthread_mutex_lock(&executor
->work
.lock
);
763 while (!executor
->should_quit
) {
765 struct action_work_item
*work_item
;
767 health_code_update();
768 if (executor
->work
.pending_count
== 0) {
770 DBG("No work items enqueued, entering wait");
771 pthread_cond_wait(&executor
->work
.cond
,
772 &executor
->work
.lock
);
773 DBG("Woke-up from wait");
778 /* Pop item from front of the list with work lock held. */
779 work_item
= cds_list_first_entry(&executor
->work
.list
,
780 struct action_work_item
, list_node
);
781 cds_list_del(&work_item
->list_node
);
782 executor
->work
.pending_count
--;
785 * Work can be performed without holding the work lock,
786 * allowing new items to be queued.
788 pthread_mutex_unlock(&executor
->work
.lock
);
790 /* Execute item only if a trigger is registered. */
791 lttng_trigger_lock(work_item
->trigger
);
792 if (!lttng_trigger_is_registered(work_item
->trigger
)) {
793 const char *trigger_name
= NULL
;
794 uid_t trigger_owner_uid
;
795 enum lttng_trigger_status trigger_status
;
797 trigger_name
= get_trigger_name(work_item
->trigger
);
799 trigger_status
= lttng_trigger_get_owner_uid(
800 work_item
->trigger
, &trigger_owner_uid
);
801 assert(trigger_status
== LTTNG_TRIGGER_STATUS_OK
);
803 DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64
", trigger name = `%s`, trigger owner uid = %d",
804 work_item
->id
, trigger_name
,
805 (int) trigger_owner_uid
);
810 ret
= action_work_item_execute(executor
, work_item
);
813 lttng_trigger_unlock(work_item
->trigger
);
814 action_work_item_destroy(work_item
);
820 health_code_update();
821 pthread_mutex_lock(&executor
->work
.lock
);
824 if (executor
->should_quit
) {
825 pthread_mutex_unlock(&executor
->work
.lock
);
827 DBG("Left work execution loop");
829 health_code_update();
831 rcu_thread_offline();
832 rcu_unregister_thread();
833 health_unregister(the_health_sessiond
);
838 static bool shutdown_action_executor_thread(void *_data
)
840 struct action_executor
*executor
= _data
;
842 pthread_mutex_lock(&executor
->work
.lock
);
843 executor
->should_quit
= true;
844 pthread_cond_signal(&executor
->work
.cond
);
845 pthread_mutex_unlock(&executor
->work
.lock
);
849 static void clean_up_action_executor_thread(void *_data
)
851 struct action_executor
*executor
= _data
;
853 assert(cds_list_empty(&executor
->work
.list
));
855 pthread_mutex_destroy(&executor
->work
.lock
);
856 pthread_cond_destroy(&executor
->work
.cond
);
860 struct action_executor
*action_executor_create(
861 struct notification_thread_handle
*handle
)
863 struct action_executor
*executor
= zmalloc(sizeof(*executor
));
869 CDS_INIT_LIST_HEAD(&executor
->work
.list
);
870 pthread_cond_init(&executor
->work
.cond
, NULL
);
871 pthread_mutex_init(&executor
->work
.lock
, NULL
);
872 executor
->notification_thread_handle
= handle
;
874 executor
->thread
= lttng_thread_create(THREAD_NAME
,
875 action_executor_thread
, shutdown_action_executor_thread
,
876 clean_up_action_executor_thread
, executor
);
881 void action_executor_destroy(struct action_executor
*executor
)
883 struct action_work_item
*work_item
, *tmp
;
885 /* TODO Wait for work list to drain? */
886 lttng_thread_shutdown(executor
->thread
);
887 pthread_mutex_lock(&executor
->work
.lock
);
888 if (executor
->work
.pending_count
!= 0) {
890 " trigger action%s still queued for execution and will be discarded",
891 executor
->work
.pending_count
,
892 executor
->work
.pending_count
== 1 ? " is" :
896 cds_list_for_each_entry_safe (
897 work_item
, tmp
, &executor
->work
.list
, list_node
) {
898 WARN("Discarding action work item %" PRIu64
899 " associated to trigger `%s`",
900 work_item
->id
, get_trigger_name(work_item
->trigger
));
901 cds_list_del(&work_item
->list_node
);
902 action_work_item_destroy(work_item
);
904 pthread_mutex_unlock(&executor
->work
.lock
);
905 lttng_thread_put(executor
->thread
);
908 /* RCU read-lock must be held by the caller. */
909 enum action_executor_status
action_executor_enqueue_trigger(
910 struct action_executor
*executor
,
911 struct lttng_trigger
*trigger
,
912 struct lttng_evaluation
*evaluation
,
913 const struct lttng_credentials
*object_creds
,
914 struct notification_client_list
*client_list
)
917 enum action_executor_status executor_status
= ACTION_EXECUTOR_STATUS_OK
;
918 const uint64_t work_item_id
= executor
->next_work_item_id
++;
919 struct action_work_item
*work_item
;
924 pthread_mutex_lock(&executor
->work
.lock
);
925 /* Check for queue overflow. */
926 if (executor
->work
.pending_count
>= MAX_QUEUED_WORK_COUNT
) {
927 /* Most likely spammy, remove if it is the case. */
928 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64
,
929 get_trigger_name(trigger
), work_item_id
);
930 executor_status
= ACTION_EXECUTOR_STATUS_OVERFLOW
;
934 work_item
= zmalloc(sizeof(*work_item
));
936 PERROR("Failed to allocate action executor work item: trigger name = `%s`",
937 get_trigger_name(trigger
));
938 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
942 lttng_trigger_get(trigger
);
944 const bool reference_acquired
=
945 notification_client_list_get(client_list
);
947 assert(reference_acquired
);
950 *work_item
= (typeof(*work_item
)){
953 /* Ownership transferred to the work item. */
954 .evaluation
= evaluation
,
956 .is_set
= !!object_creds
,
957 .value
= object_creds
? *object_creds
:
958 (typeof(work_item
->object_creds
.value
)) {},
960 .client_list
= client_list
,
961 .list_node
= CDS_LIST_HEAD_INIT(work_item
->list_node
),
966 /* Build the array of action work subitems for the passed trigger. */
967 lttng_dynamic_array_init(&work_item
->subitems
,
968 sizeof(struct action_work_subitem
),
969 action_work_subitem_destructor
);
971 ret
= populate_subitem_array_from_trigger(
972 trigger
, &work_item
->subitems
);
974 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
975 get_trigger_name(trigger
));
976 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
980 cds_list_add_tail(&work_item
->list_node
, &executor
->work
.list
);
981 executor
->work
.pending_count
++;
982 DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64
,
983 get_trigger_name(trigger
), work_item_id
);
988 pthread_cond_signal(&executor
->work
.cond
);
991 pthread_mutex_unlock(&executor
->work
.lock
);
992 lttng_evaluation_destroy(evaluation
);
993 return executor_status
;
996 static int add_action_to_subitem_array(struct lttng_action
*action
,
997 struct lttng_dynamic_array
*subitems
)
1000 enum lttng_action_type type
= lttng_action_get_type(action
);
1001 const char *session_name
= NULL
;
1002 enum lttng_action_status status
;
1003 struct action_work_subitem subitem
= {
1006 .session_id
= LTTNG_OPTIONAL_INIT_UNSET
,
1013 if (type
== LTTNG_ACTION_TYPE_LIST
) {
1014 unsigned int count
, i
;
1016 status
= lttng_action_list_get_count(action
, &count
);
1017 assert(status
== LTTNG_ACTION_STATUS_OK
);
1019 for (i
= 0; i
< count
; i
++) {
1020 struct lttng_action
*inner_action
= NULL
;
1022 inner_action
= lttng_action_list_borrow_mutable_at_index(
1024 assert(inner_action
);
1025 ret
= add_action_to_subitem_array(
1026 inner_action
, subitems
);
1033 * Go directly to the end since there is no need to add the
1034 * list action by itself to the subitems array.
1039 /* Gather execution context. */
1041 case LTTNG_ACTION_TYPE_NOTIFY
:
1043 case LTTNG_ACTION_TYPE_START_SESSION
:
1044 status
= lttng_action_start_session_get_session_name(
1045 action
, &session_name
);
1046 assert(status
== LTTNG_ACTION_STATUS_OK
);
1048 case LTTNG_ACTION_TYPE_STOP_SESSION
:
1049 status
= lttng_action_stop_session_get_session_name(
1050 action
, &session_name
);
1051 assert(status
== LTTNG_ACTION_STATUS_OK
);
1053 case LTTNG_ACTION_TYPE_ROTATE_SESSION
:
1054 status
= lttng_action_rotate_session_get_session_name(
1055 action
, &session_name
);
1056 assert(status
== LTTNG_ACTION_STATUS_OK
);
1058 case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
:
1059 status
= lttng_action_snapshot_session_get_session_name(
1060 action
, &session_name
);
1061 assert(status
== LTTNG_ACTION_STATUS_OK
);
1063 case LTTNG_ACTION_TYPE_LIST
:
1064 case LTTNG_ACTION_TYPE_UNKNOWN
:
1072 * Fetch the session execution context info as needed.
1073 * Note that we could decide to not add an action for which we know the
1074 * execution will not happen (i.e no session exists for that name). For
1075 * now we leave the decision to skip to the action executor for sake of
1076 * simplicity and consistency.
1078 if (session_name
!= NULL
) {
1079 uint64_t session_id
;
1082 * Instantaneous sampling of the session id if present.
1084 * This method is preferred over `sessiond_find_by_name` then
1085 * fetching the session'd id since `sessiond_find_by_name`
1086 * requires the session list lock to be taken.
1088 * Taking the session list lock can lead to a deadlock
1089 * between the action executor and the notification thread
1090 * (caller of add_action_to_subitem_array). It is okay if the
1091 * session state changes between the enqueuing time and the
1092 * execution time. The execution context is validated at
1095 if (sample_session_id_by_name(session_name
, &session_id
)) {
1096 LTTNG_OPTIONAL_SET(&subitem
.context
.session_id
,
1101 /* Get a reference to the action. */
1102 lttng_action_get(action
);
1103 subitem
.action
= action
;
1105 ret
= lttng_dynamic_array_add_element(subitems
, &subitem
);
1107 ERR("Failed to add work subitem to the subitem array");
1108 lttng_action_put(action
);
1117 static int populate_subitem_array_from_trigger(struct lttng_trigger
*trigger
,
1118 struct lttng_dynamic_array
*subitems
)
1120 struct lttng_action
*action
;
1122 action
= lttng_trigger_get_action(trigger
);
1125 return add_action_to_subitem_array(action
, subitems
);