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/macros.h>
16 #include <common/optional.h>
17 #include <lttng/action/action-internal.h>
18 #include <lttng/action/group.h>
19 #include <lttng/action/notify.h>
20 #include <lttng/action/rotate-session.h>
21 #include <lttng/action/snapshot-session.h>
22 #include <lttng/action/start-session.h>
23 #include <lttng/action/stop-session.h>
24 #include <lttng/condition/evaluation.h>
25 #include <lttng/lttng-error.h>
26 #include <lttng/trigger/trigger-internal.h>
30 #include <urcu/list.h>
32 #define THREAD_NAME "Action Executor"
33 #define MAX_QUEUED_WORK_COUNT 8192
35 struct action_work_item
{
37 struct lttng_trigger
*trigger
;
38 struct lttng_evaluation
*evaluation
;
39 struct notification_client_list
*client_list
;
40 LTTNG_OPTIONAL(struct lttng_credentials
) object_creds
;
41 struct cds_list_head list_node
;
44 struct action_executor
{
45 struct lttng_thread
*thread
;
46 struct notification_thread_handle
*notification_thread_handle
;
48 uint64_t pending_count
;
49 struct cds_list_head list
;
54 uint64_t next_work_item_id
;
58 * Only return non-zero on a fatal error that should shut down the action
61 typedef int (*action_executor_handler
)(struct action_executor
*executor
,
62 const struct action_work_item
*,
63 const struct lttng_action
*action
);
65 static int action_executor_notify_handler(struct action_executor
*executor
,
66 const struct action_work_item
*,
67 const struct lttng_action
*);
68 static int action_executor_start_session_handler(struct action_executor
*executor
,
69 const struct action_work_item
*,
70 const struct lttng_action
*);
71 static int action_executor_stop_session_handler(struct action_executor
*executor
,
72 const struct action_work_item
*,
73 const struct lttng_action
*);
74 static int action_executor_rotate_session_handler(struct action_executor
*executor
,
75 const struct action_work_item
*,
76 const struct lttng_action
*);
77 static int action_executor_snapshot_session_handler(struct action_executor
*executor
,
78 const struct action_work_item
*,
79 const struct lttng_action
*);
80 static int action_executor_group_handler(struct action_executor
*executor
,
81 const struct action_work_item
*,
82 const struct lttng_action
*);
83 static int action_executor_generic_handler(struct action_executor
*executor
,
84 const struct action_work_item
*,
85 const struct lttng_action
*);
87 static const action_executor_handler action_executors
[] = {
88 [LTTNG_ACTION_TYPE_NOTIFY
] = action_executor_notify_handler
,
89 [LTTNG_ACTION_TYPE_START_SESSION
] = action_executor_start_session_handler
,
90 [LTTNG_ACTION_TYPE_STOP_SESSION
] = action_executor_stop_session_handler
,
91 [LTTNG_ACTION_TYPE_ROTATE_SESSION
] = action_executor_rotate_session_handler
,
92 [LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
] = action_executor_snapshot_session_handler
,
93 [LTTNG_ACTION_TYPE_GROUP
] = action_executor_group_handler
,
96 static const char *action_type_names
[] = {
97 [LTTNG_ACTION_TYPE_NOTIFY
] = "Notify",
98 [LTTNG_ACTION_TYPE_START_SESSION
] = "Start session",
99 [LTTNG_ACTION_TYPE_STOP_SESSION
] = "Stop session",
100 [LTTNG_ACTION_TYPE_ROTATE_SESSION
] = "Rotate session",
101 [LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
] = "Snapshot session",
102 [LTTNG_ACTION_TYPE_GROUP
] = "Group",
105 static const char *get_action_name(const struct lttng_action
*action
)
107 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
109 assert(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
111 return action_type_names
[action_type
];
114 /* Check if this trigger allowed to interect with a given session. */
115 static bool is_trigger_allowed_for_session(const struct lttng_trigger
*trigger
,
116 struct ltt_session
*session
)
118 bool is_allowed
= false;
119 const struct lttng_credentials session_creds
= {
120 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
->uid
),
121 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
->gid
),
123 /* Can never be NULL. */
124 const struct lttng_credentials
*trigger_creds
=
125 lttng_trigger_get_credentials(trigger
);
127 is_allowed
= (lttng_credentials_is_equal_uid(trigger_creds
, &session_creds
)) ||
128 (lttng_credentials_get_uid(trigger_creds
) == 0);
130 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
132 (long int) session
->uid
,
133 (long int) session
->gid
,
134 (long int) lttng_credentials_get_uid(trigger_creds
));
140 static const char *get_trigger_name(const struct lttng_trigger
*trigger
)
142 const char *trigger_name
;
143 enum lttng_trigger_status trigger_status
;
145 trigger_status
= lttng_trigger_get_name(trigger
, &trigger_name
);
146 assert(trigger_status
== LTTNG_TRIGGER_STATUS_OK
);
151 static int client_handle_transmission_status(
152 struct notification_client
*client
,
153 enum client_transmission_status status
,
157 struct action_executor
*executor
= user_data
;
158 bool update_communication
= true;
161 case CLIENT_TRANSMISSION_STATUS_COMPLETE
:
162 DBG("Successfully sent full notification to client, client_id = %" PRIu64
,
164 update_communication
= false;
166 case CLIENT_TRANSMISSION_STATUS_QUEUED
:
167 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64
,
170 case CLIENT_TRANSMISSION_STATUS_FAIL
:
171 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64
,
175 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64
,
181 if (!update_communication
) {
185 /* Safe to read client's id without locking as it is immutable. */
186 ret
= notification_thread_client_communication_update(
187 executor
->notification_thread_handle
, client
->id
,
193 static int action_executor_notify_handler(struct action_executor
*executor
,
194 const struct action_work_item
*work_item
,
195 const struct lttng_action
*action
)
197 return notification_client_list_send_evaluation(work_item
->client_list
,
198 lttng_trigger_get_const_condition(work_item
->trigger
),
199 work_item
->evaluation
,
200 lttng_trigger_get_credentials(work_item
->trigger
),
201 work_item
->object_creds
.is_set
?
202 &(work_item
->object_creds
.value
) :
204 client_handle_transmission_status
, executor
);
207 static int action_executor_start_session_handler(struct action_executor
*executor
,
208 const struct action_work_item
*work_item
,
209 const struct lttng_action
*action
)
212 const char *session_name
;
213 enum lttng_action_status action_status
;
214 struct ltt_session
*session
;
215 enum lttng_error_code cmd_ret
;
217 action_status
= lttng_action_start_session_get_session_name(
218 action
, &session_name
);
219 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
220 ERR("Failed to get session name from `%s` action",
221 get_action_name(action
));
227 session
= session_find_by_name(session_name
);
229 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
230 session_name
, get_action_name(action
),
231 get_trigger_name(work_item
->trigger
));
232 goto error_unlock_list
;
235 session_lock(session
);
236 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
237 goto error_dispose_session
;
240 cmd_ret
= cmd_start_trace(session
);
243 DBG("Successfully started session `%s` on behalf of trigger `%s`",
244 session_name
, get_trigger_name(work_item
->trigger
));
246 case LTTNG_ERR_TRACE_ALREADY_STARTED
:
247 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
248 session_name
, get_trigger_name(work_item
->trigger
));
251 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
252 session_name
, get_trigger_name(work_item
->trigger
),
253 lttng_strerror(-cmd_ret
));
257 error_dispose_session
:
258 session_unlock(session
);
259 session_put(session
);
261 session_unlock_list();
266 static int action_executor_stop_session_handler(struct action_executor
*executor
,
267 const struct action_work_item
*work_item
,
268 const struct lttng_action
*action
)
271 const char *session_name
;
272 enum lttng_action_status action_status
;
273 struct ltt_session
*session
;
274 enum lttng_error_code cmd_ret
;
276 action_status
= lttng_action_stop_session_get_session_name(
277 action
, &session_name
);
278 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
279 ERR("Failed to get session name from `%s` action",
280 get_action_name(action
));
286 session
= session_find_by_name(session_name
);
288 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
289 session_name
, get_action_name(action
),
290 get_trigger_name(work_item
->trigger
));
291 goto error_unlock_list
;
294 session_lock(session
);
295 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
296 goto error_dispose_session
;
299 cmd_ret
= cmd_stop_trace(session
);
302 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
303 session_name
, get_trigger_name(work_item
->trigger
));
305 case LTTNG_ERR_TRACE_ALREADY_STOPPED
:
306 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
307 session_name
, get_trigger_name(work_item
->trigger
));
310 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
311 session_name
, get_trigger_name(work_item
->trigger
),
312 lttng_strerror(-cmd_ret
));
316 error_dispose_session
:
317 session_unlock(session
);
318 session_put(session
);
320 session_unlock_list();
325 static int action_executor_rotate_session_handler(struct action_executor
*executor
,
326 const struct action_work_item
*work_item
,
327 const struct lttng_action
*action
)
330 const char *session_name
;
331 enum lttng_action_status action_status
;
332 struct ltt_session
*session
;
333 enum lttng_error_code cmd_ret
;
335 action_status
= lttng_action_rotate_session_get_session_name(
336 action
, &session_name
);
337 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
338 ERR("Failed to get session name from `%s` action",
339 get_action_name(action
));
345 session
= session_find_by_name(session_name
);
347 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
348 session_name
, get_action_name(action
),
349 get_trigger_name(work_item
->trigger
));
350 goto error_unlock_list
;
353 session_lock(session
);
354 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
355 goto error_dispose_session
;
358 cmd_ret
= cmd_rotate_session(session
, NULL
, false,
359 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
362 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
363 session_name
, get_trigger_name(work_item
->trigger
));
365 case LTTNG_ERR_ROTATION_PENDING
:
366 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
367 session_name
, get_trigger_name(work_item
->trigger
));
369 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
370 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
371 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",
372 session_name
, get_trigger_name(work_item
->trigger
));
375 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
376 session_name
, get_trigger_name(work_item
->trigger
),
377 lttng_strerror(-cmd_ret
));
381 error_dispose_session
:
382 session_unlock(session
);
383 session_put(session
);
385 session_unlock_list();
390 static int action_executor_snapshot_session_handler(struct action_executor
*executor
,
391 const struct action_work_item
*work_item
,
392 const struct lttng_action
*action
)
395 const char *session_name
;
396 enum lttng_action_status action_status
;
397 struct ltt_session
*session
;
398 const struct lttng_snapshot_output default_snapshot_output
= {
399 .max_size
= UINT64_MAX
,
401 const struct lttng_snapshot_output
*snapshot_output
=
402 &default_snapshot_output
;
403 enum lttng_error_code cmd_ret
;
405 action_status
= lttng_action_snapshot_session_get_session_name(
406 action
, &session_name
);
407 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
408 ERR("Failed to get session name from `%s` action",
409 get_action_name(action
));
414 action_status
= lttng_action_snapshot_session_get_output(
415 action
, &snapshot_output
);
416 if (action_status
!= LTTNG_ACTION_STATUS_OK
&&
417 action_status
!= LTTNG_ACTION_STATUS_UNSET
) {
418 ERR("Failed to get output from `%s` action",
419 get_action_name(action
));
425 session
= session_find_by_name(session_name
);
427 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%p`",
428 session_name
, get_action_name(action
),
430 goto error_unlock_list
;
434 session_lock(session
);
435 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
436 goto error_dispose_session
;
439 cmd_ret
= cmd_snapshot_record(session
, snapshot_output
, 0);
442 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
443 session_name
, get_trigger_name(work_item
->trigger
));
446 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
447 session_name
, get_trigger_name(work_item
->trigger
),
448 lttng_strerror(-cmd_ret
));
452 error_dispose_session
:
453 session_unlock(session
);
454 session_put(session
);
456 session_unlock_list();
461 static int action_executor_group_handler(struct action_executor
*executor
,
462 const struct action_work_item
*work_item
,
463 const struct lttng_action
*action_group
)
466 unsigned int i
, count
;
467 enum lttng_action_status action_status
;
469 action_status
= lttng_action_group_get_count(action_group
, &count
);
470 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
472 ERR("Failed to get count of action in action group");
477 DBG("Action group has %u action%s", count
, count
!= 1 ? "s" : "");
478 for (i
= 0; i
< count
; i
++) {
479 const struct lttng_action
*action
=
480 lttng_action_group_get_at_index(
483 ret
= action_executor_generic_handler(
484 executor
, work_item
, action
);
486 ERR("Stopping the execution of the action group of trigger `%s` following a fatal error",
487 get_trigger_name(work_item
->trigger
));
495 static int action_executor_generic_handler(struct action_executor
*executor
,
496 const struct action_work_item
*work_item
,
497 const struct lttng_action
*action
)
499 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
501 assert(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
503 DBG("Executing action `%s` of trigger `%p` action work item %" PRIu64
,
504 get_action_name(action
),
505 get_trigger_name(work_item
->trigger
),
508 return action_executors
[action_type
](
509 executor
, work_item
, action
);
512 static int action_work_item_execute(struct action_executor
*executor
,
513 struct action_work_item
*work_item
)
516 const struct lttng_action
*action
=
517 lttng_trigger_get_const_action(work_item
->trigger
);
519 DBG("Starting execution of action work item %" PRIu64
" of trigger `%s`",
520 work_item
->id
, get_trigger_name(work_item
->trigger
));
521 ret
= action_executor_generic_handler(executor
, work_item
, action
);
522 DBG("Completed execution of action work item %" PRIu64
" of trigger `%s`",
523 work_item
->id
, get_trigger_name(work_item
->trigger
));
527 static void action_work_item_destroy(struct action_work_item
*work_item
)
529 lttng_trigger_put(work_item
->trigger
);
530 lttng_evaluation_destroy(work_item
->evaluation
);
531 notification_client_list_put(work_item
->client_list
);
535 static void *action_executor_thread(void *_data
)
537 struct action_executor
*executor
= _data
;
541 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR
);
543 rcu_register_thread();
546 DBG("Entering work execution loop");
547 pthread_mutex_lock(&executor
->work
.lock
);
548 while (!executor
->should_quit
) {
550 struct action_work_item
*work_item
;
552 health_code_update();
553 if (executor
->work
.pending_count
== 0) {
555 DBG("No work items enqueued, entering wait");
556 pthread_cond_wait(&executor
->work
.cond
,
557 &executor
->work
.lock
);
558 DBG("Woke-up from wait");
563 /* Pop item from front of the list with work lock held. */
564 work_item
= cds_list_first_entry(&executor
->work
.list
,
565 struct action_work_item
, list_node
);
566 cds_list_del(&work_item
->list_node
);
567 executor
->work
.pending_count
--;
570 * Work can be performed without holding the work lock,
571 * allowing new items to be queued.
573 pthread_mutex_unlock(&executor
->work
.lock
);
574 ret
= action_work_item_execute(executor
, work_item
);
575 action_work_item_destroy(work_item
);
581 health_code_update();
582 pthread_mutex_lock(&executor
->work
.lock
);
585 if (executor
->should_quit
) {
586 pthread_mutex_unlock(&executor
->work
.lock
);
588 DBG("Left work execution loop");
590 health_code_update();
592 rcu_thread_offline();
593 rcu_unregister_thread();
594 health_unregister(health_sessiond
);
599 static bool shutdown_action_executor_thread(void *_data
)
601 struct action_executor
*executor
= _data
;
603 executor
->should_quit
= true;
604 pthread_cond_signal(&executor
->work
.cond
);
608 static void clean_up_action_executor_thread(void *_data
)
610 struct action_executor
*executor
= _data
;
612 assert(cds_list_empty(&executor
->work
.list
));
614 pthread_mutex_destroy(&executor
->work
.lock
);
615 pthread_cond_destroy(&executor
->work
.cond
);
619 struct action_executor
*action_executor_create(
620 struct notification_thread_handle
*handle
)
622 struct action_executor
*executor
= zmalloc(sizeof(*executor
));
628 CDS_INIT_LIST_HEAD(&executor
->work
.list
);
629 pthread_cond_init(&executor
->work
.cond
, NULL
);
630 pthread_mutex_init(&executor
->work
.lock
, NULL
);
631 executor
->notification_thread_handle
= handle
;
633 executor
->thread
= lttng_thread_create(THREAD_NAME
,
634 action_executor_thread
, shutdown_action_executor_thread
,
635 clean_up_action_executor_thread
, executor
);
640 void action_executor_destroy(struct action_executor
*executor
)
642 struct action_work_item
*work_item
, *tmp
;
644 /* TODO Wait for work list to drain? */
645 lttng_thread_shutdown(executor
->thread
);
646 pthread_mutex_lock(&executor
->work
.lock
);
647 if (executor
->work
.pending_count
!= 0) {
649 " trigger action%s still queued for execution and will be discarded",
650 executor
->work
.pending_count
,
651 executor
->work
.pending_count
== 1 ? " is" :
655 cds_list_for_each_entry_safe (
656 work_item
, tmp
, &executor
->work
.list
, list_node
) {
657 WARN("Discarding action work item %" PRIu64
658 " associated to trigger `%s`",
659 work_item
->id
, get_trigger_name(work_item
->trigger
));
660 cds_list_del(&work_item
->list_node
);
661 action_work_item_destroy(work_item
);
663 pthread_mutex_unlock(&executor
->work
.lock
);
664 lttng_thread_put(executor
->thread
);
667 /* RCU read-lock must be held by the caller. */
668 enum action_executor_status
action_executor_enqueue(
669 struct action_executor
*executor
,
670 struct lttng_trigger
*trigger
,
671 struct lttng_evaluation
*evaluation
,
672 const struct lttng_credentials
*object_creds
,
673 struct notification_client_list
*client_list
)
675 enum action_executor_status executor_status
= ACTION_EXECUTOR_STATUS_OK
;
676 const uint64_t work_item_id
= executor
->next_work_item_id
++;
677 struct action_work_item
*work_item
;
680 pthread_mutex_lock(&executor
->work
.lock
);
681 /* Check for queue overflow. */
682 if (executor
->work
.pending_count
>= MAX_QUEUED_WORK_COUNT
) {
683 /* Most likely spammy, remove if it is the case. */
684 DBG("Refusing to enqueue action for trigger `%s` as work item %" PRIu64
685 " (overflow)", get_trigger_name(trigger
), work_item_id
);
686 executor_status
= ACTION_EXECUTOR_STATUS_OVERFLOW
;
690 work_item
= zmalloc(sizeof(*work_item
));
692 PERROR("Failed to allocate action executor work item on behalf of trigger `%p`",
693 get_trigger_name(trigger
));
694 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
698 lttng_trigger_get(trigger
);
700 const bool reference_acquired
=
701 notification_client_list_get(client_list
);
703 assert(reference_acquired
);
706 *work_item
= (typeof(*work_item
)){
709 /* Ownership transferred to the work item. */
710 .evaluation
= evaluation
,
712 .is_set
= !!object_creds
,
713 .value
= object_creds
? *object_creds
:
714 (typeof(work_item
->object_creds
.value
)) {},
716 .client_list
= client_list
,
717 .list_node
= CDS_LIST_HEAD_INIT(work_item
->list_node
),
721 cds_list_add_tail(&work_item
->list_node
, &executor
->work
.list
);
722 executor
->work
.pending_count
++;
723 DBG("Enqueued action for trigger `%p` as work item %" PRIu64
,
724 get_trigger_name(trigger
), work_item_id
);
728 pthread_mutex_unlock(&executor
->work
.lock
);
730 pthread_cond_signal(&executor
->work
.cond
);
733 lttng_evaluation_destroy(evaluation
);
734 return executor_status
;