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-internal.h>
20 #include <lttng/action/notify.h>
21 #include <lttng/action/rotate-session.h>
22 #include <lttng/action/snapshot-session.h>
23 #include <lttng/action/start-session.h>
24 #include <lttng/action/stop-session.h>
25 #include <lttng/condition/evaluation.h>
26 #include <lttng/condition/on-event-internal.h>
27 #include <lttng/lttng-error.h>
28 #include <lttng/trigger/trigger-internal.h>
32 #include <urcu/list.h>
34 #define THREAD_NAME "Action Executor"
35 #define MAX_QUEUED_WORK_COUNT 8192
37 struct action_work_item
{
39 struct lttng_trigger
*trigger
;
40 struct lttng_evaluation
*evaluation
;
41 struct notification_client_list
*client_list
;
42 LTTNG_OPTIONAL(struct lttng_credentials
) object_creds
;
43 struct cds_list_head list_node
;
46 struct action_executor
{
47 struct lttng_thread
*thread
;
48 struct notification_thread_handle
*notification_thread_handle
;
50 uint64_t pending_count
;
51 struct cds_list_head list
;
56 uint64_t next_work_item_id
;
60 * Only return non-zero on a fatal error that should shut down the action
63 typedef int (*action_executor_handler
)(struct action_executor
*executor
,
64 const struct action_work_item
*,
65 const struct lttng_action
*action
);
67 static int action_executor_notify_handler(struct action_executor
*executor
,
68 const struct action_work_item
*,
69 const struct lttng_action
*);
70 static int action_executor_incr_value_handler(struct action_executor
*executor
,
71 const struct action_work_item
*,
72 const struct lttng_action
*);
73 static int action_executor_start_session_handler(struct action_executor
*executor
,
74 const struct action_work_item
*,
75 const struct lttng_action
*);
76 static int action_executor_stop_session_handler(struct action_executor
*executor
,
77 const struct action_work_item
*,
78 const struct lttng_action
*);
79 static int action_executor_rotate_session_handler(struct action_executor
*executor
,
80 const struct action_work_item
*,
81 const struct lttng_action
*);
82 static int action_executor_snapshot_session_handler(struct action_executor
*executor
,
83 const struct action_work_item
*,
84 const struct lttng_action
*);
85 static int action_executor_group_handler(struct action_executor
*executor
,
86 const struct action_work_item
*,
87 const struct lttng_action
*);
88 static int action_executor_generic_handler(struct action_executor
*executor
,
89 const struct action_work_item
*,
90 const struct lttng_action
*);
92 static const action_executor_handler action_executors
[] = {
93 [LTTNG_ACTION_TYPE_NOTIFY
] = action_executor_notify_handler
,
94 [LTTNG_ACTION_TYPE_INCREMENT_VALUE
] = action_executor_incr_value_handler
,
95 [LTTNG_ACTION_TYPE_START_SESSION
] = action_executor_start_session_handler
,
96 [LTTNG_ACTION_TYPE_STOP_SESSION
] = action_executor_stop_session_handler
,
97 [LTTNG_ACTION_TYPE_ROTATE_SESSION
] = action_executor_rotate_session_handler
,
98 [LTTNG_ACTION_TYPE_SNAPSHOT_SESSION
] = action_executor_snapshot_session_handler
,
99 [LTTNG_ACTION_TYPE_GROUP
] = action_executor_group_handler
,
103 static const char *get_action_name(const struct lttng_action
*action
)
105 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
107 assert(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
109 return lttng_action_type_string(action_type
);
112 /* Check if this trigger allowed to interect with a given session. */
113 static bool is_trigger_allowed_for_session(const struct lttng_trigger
*trigger
,
114 struct ltt_session
*session
)
116 bool is_allowed
= false;
117 const struct lttng_credentials session_creds
= {
118 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
->uid
),
119 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
->gid
),
121 /* Can never be NULL. */
122 const struct lttng_credentials
*trigger_creds
=
123 lttng_trigger_get_credentials(trigger
);
125 is_allowed
= (lttng_credentials_is_equal_uid(trigger_creds
, &session_creds
)) ||
126 (lttng_credentials_get_uid(trigger_creds
) == 0);
128 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
130 (long int) session
->uid
,
131 (long int) session
->gid
,
132 (long int) lttng_credentials_get_uid(trigger_creds
));
138 static const char *get_trigger_name(const struct lttng_trigger
*trigger
)
140 const char *trigger_name
;
141 enum lttng_trigger_status trigger_status
;
143 trigger_status
= lttng_trigger_get_name(trigger
, &trigger_name
);
144 assert(trigger_status
== LTTNG_TRIGGER_STATUS_OK
);
149 static int client_handle_transmission_status(
150 struct notification_client
*client
,
151 enum client_transmission_status status
,
155 struct action_executor
*executor
= user_data
;
156 bool update_communication
= true;
159 case CLIENT_TRANSMISSION_STATUS_COMPLETE
:
160 DBG("Successfully sent full notification to client, client_id = %" PRIu64
,
162 update_communication
= false;
164 case CLIENT_TRANSMISSION_STATUS_QUEUED
:
165 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64
,
168 case CLIENT_TRANSMISSION_STATUS_FAIL
:
169 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64
,
173 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64
,
179 if (!update_communication
) {
183 /* Safe to read client's id without locking as it is immutable. */
184 ret
= notification_thread_client_communication_update(
185 executor
->notification_thread_handle
, client
->id
,
191 static int action_executor_notify_handler(struct action_executor
*executor
,
192 const struct action_work_item
*work_item
,
193 const struct lttng_action
*action
)
195 return notification_client_list_send_evaluation(work_item
->client_list
,
196 lttng_trigger_get_const_condition(work_item
->trigger
),
197 work_item
->evaluation
,
198 lttng_trigger_get_credentials(work_item
->trigger
),
199 work_item
->object_creds
.is_set
?
200 &(work_item
->object_creds
.value
) :
202 client_handle_transmission_status
, executor
);
205 static int action_executor_incr_value_handler(struct action_executor
*executor
,
206 const struct action_work_item
*work_item
,
207 const struct lttng_action
*action
)
209 /* This action is executed by the tracer. */
213 static int action_executor_start_session_handler(struct action_executor
*executor
,
214 const struct action_work_item
*work_item
,
215 const struct lttng_action
*action
)
218 const char *session_name
;
219 enum lttng_action_status action_status
;
220 struct ltt_session
*session
;
221 enum lttng_error_code cmd_ret
;
223 action_status
= lttng_action_start_session_get_session_name(
224 action
, &session_name
);
225 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
226 ERR("Failed to get session name from `%s` action",
227 get_action_name(action
));
233 session
= session_find_by_name(session_name
);
235 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
236 session_name
, get_action_name(action
),
237 get_trigger_name(work_item
->trigger
));
238 goto error_unlock_list
;
241 session_lock(session
);
242 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
243 goto error_dispose_session
;
245 cmd_ret
= cmd_start_trace(session
);
248 DBG("Successfully started session `%s` on behalf of trigger `%s`",
249 session_name
, get_trigger_name(work_item
->trigger
));
251 case LTTNG_ERR_TRACE_ALREADY_STARTED
:
252 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
253 session_name
, get_trigger_name(work_item
->trigger
));
256 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
257 session_name
, get_trigger_name(work_item
->trigger
),
258 lttng_strerror(-cmd_ret
));
262 error_dispose_session
:
263 session_unlock(session
);
264 session_put(session
);
266 session_unlock_list();
271 static int action_executor_stop_session_handler(struct action_executor
*executor
,
272 const struct action_work_item
*work_item
,
273 const struct lttng_action
*action
)
276 const char *session_name
;
277 enum lttng_action_status action_status
;
278 struct ltt_session
*session
;
279 enum lttng_error_code cmd_ret
;
281 action_status
= lttng_action_stop_session_get_session_name(
282 action
, &session_name
);
283 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
284 ERR("Failed to get session name from `%s` action",
285 get_action_name(action
));
291 session
= session_find_by_name(session_name
);
293 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
294 session_name
, get_action_name(action
),
295 get_trigger_name(work_item
->trigger
));
296 goto error_unlock_list
;
299 session_lock(session
);
300 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
301 goto error_dispose_session
;
304 cmd_ret
= cmd_stop_trace(session
);
307 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
308 session_name
, get_trigger_name(work_item
->trigger
));
310 case LTTNG_ERR_TRACE_ALREADY_STOPPED
:
311 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
312 session_name
, get_trigger_name(work_item
->trigger
));
315 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
316 session_name
, get_trigger_name(work_item
->trigger
),
317 lttng_strerror(-cmd_ret
));
321 error_dispose_session
:
322 session_unlock(session
);
323 session_put(session
);
325 session_unlock_list();
330 static int action_executor_rotate_session_handler(struct action_executor
*executor
,
331 const struct action_work_item
*work_item
,
332 const struct lttng_action
*action
)
335 const char *session_name
;
336 enum lttng_action_status action_status
;
337 struct ltt_session
*session
;
338 enum lttng_error_code cmd_ret
;
340 action_status
= lttng_action_rotate_session_get_session_name(
341 action
, &session_name
);
342 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
343 ERR("Failed to get session name from `%s` action",
344 get_action_name(action
));
350 session
= session_find_by_name(session_name
);
352 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
353 session_name
, get_action_name(action
),
354 get_trigger_name(work_item
->trigger
));
355 goto error_unlock_list
;
358 session_lock(session
);
359 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
360 goto error_dispose_session
;
363 cmd_ret
= cmd_rotate_session(session
, NULL
, false,
364 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
367 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
368 session_name
, get_trigger_name(work_item
->trigger
));
370 case LTTNG_ERR_ROTATION_PENDING
:
371 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
372 session_name
, get_trigger_name(work_item
->trigger
));
374 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
375 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
376 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",
377 session_name
, get_trigger_name(work_item
->trigger
));
380 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
381 session_name
, get_trigger_name(work_item
->trigger
),
382 lttng_strerror(-cmd_ret
));
386 error_dispose_session
:
387 session_unlock(session
);
388 session_put(session
);
390 session_unlock_list();
395 static int action_executor_snapshot_session_handler(struct action_executor
*executor
,
396 const struct action_work_item
*work_item
,
397 const struct lttng_action
*action
)
400 const char *session_name
;
401 enum lttng_action_status action_status
;
402 struct ltt_session
*session
;
403 const struct lttng_snapshot_output default_snapshot_output
= {
404 .max_size
= UINT64_MAX
,
406 const struct lttng_snapshot_output
*snapshot_output
=
407 &default_snapshot_output
;
408 enum lttng_error_code cmd_ret
;
410 action_status
= lttng_action_snapshot_session_get_session_name(
411 action
, &session_name
);
412 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
413 ERR("Failed to get session name from `%s` action",
414 get_action_name(action
));
419 action_status
= lttng_action_snapshot_session_get_output(
420 action
, &snapshot_output
);
421 if (action_status
!= LTTNG_ACTION_STATUS_OK
&&
422 action_status
!= LTTNG_ACTION_STATUS_UNSET
) {
423 ERR("Failed to get output from `%s` action",
424 get_action_name(action
));
430 session
= session_find_by_name(session_name
);
432 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
433 session_name
, get_action_name(action
),
434 get_trigger_name(work_item
->trigger
));
435 goto error_unlock_list
;
439 session_lock(session
);
440 if (!is_trigger_allowed_for_session(work_item
->trigger
, session
)) {
441 goto error_dispose_session
;
444 cmd_ret
= cmd_snapshot_record(session
, snapshot_output
, 0);
447 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
448 session_name
, get_trigger_name(work_item
->trigger
));
451 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
452 session_name
, get_trigger_name(work_item
->trigger
),
453 lttng_strerror(-cmd_ret
));
457 error_dispose_session
:
458 session_unlock(session
);
459 session_put(session
);
461 session_unlock_list();
466 static int action_executor_group_handler(struct action_executor
*executor
,
467 const struct action_work_item
*work_item
,
468 const struct lttng_action
*action_group
)
471 unsigned int i
, count
;
472 enum lttng_action_status action_status
;
474 action_status
= lttng_action_group_get_count(action_group
, &count
);
475 if (action_status
!= LTTNG_ACTION_STATUS_OK
) {
477 ERR("Failed to get count of action in action group");
482 DBG("Action group has %u action%s", count
, count
!= 1 ? "s" : "");
483 for (i
= 0; i
< count
; i
++) {
484 const struct lttng_action
*action
=
485 lttng_action_group_get_at_index(
488 ret
= action_executor_generic_handler(
489 executor
, work_item
, action
);
491 ERR("Stopping the execution of the action group of trigger `%s` following a fatal error",
492 get_trigger_name(work_item
->trigger
));
500 static int action_executor_generic_handler(struct action_executor
*executor
,
501 const struct action_work_item
*work_item
,
502 const struct lttng_action
*action
)
504 const enum lttng_action_type action_type
= lttng_action_get_type(action
);
506 assert(action_type
!= LTTNG_ACTION_TYPE_UNKNOWN
);
508 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64
,
509 get_action_name(action
),
510 get_trigger_name(work_item
->trigger
),
513 return action_executors
[action_type
](
514 executor
, work_item
, action
);
517 static int action_work_item_execute(struct action_executor
*executor
,
518 struct action_work_item
*work_item
)
521 const struct lttng_action
*action
=
522 lttng_trigger_get_const_action(work_item
->trigger
);
524 DBG("Starting execution of action work item %" PRIu64
" of trigger `%s`",
525 work_item
->id
, get_trigger_name(work_item
->trigger
));
526 ret
= action_executor_generic_handler(executor
, work_item
, action
);
527 DBG("Completed execution of action work item %" PRIu64
" of trigger `%s`",
528 work_item
->id
, get_trigger_name(work_item
->trigger
));
532 static void action_work_item_destroy(struct action_work_item
*work_item
)
534 lttng_trigger_put(work_item
->trigger
);
535 lttng_evaluation_destroy(work_item
->evaluation
);
536 notification_client_list_put(work_item
->client_list
);
540 static void *action_executor_thread(void *_data
)
542 struct action_executor
*executor
= _data
;
546 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR
);
548 rcu_register_thread();
551 DBG("Entering work execution loop");
552 pthread_mutex_lock(&executor
->work
.lock
);
553 while (!executor
->should_quit
) {
555 struct action_work_item
*work_item
;
557 health_code_update();
558 if (executor
->work
.pending_count
== 0) {
560 DBG("No work items enqueued, entering wait");
561 pthread_cond_wait(&executor
->work
.cond
,
562 &executor
->work
.lock
);
563 DBG("Woke-up from wait");
568 /* Pop item from front of the list with work lock held. */
569 work_item
= cds_list_first_entry(&executor
->work
.list
,
570 struct action_work_item
, list_node
);
571 cds_list_del(&work_item
->list_node
);
572 executor
->work
.pending_count
--;
575 * Work can be performed without holding the work lock,
576 * allowing new items to be queued.
578 pthread_mutex_unlock(&executor
->work
.lock
);
579 ret
= action_work_item_execute(executor
, work_item
);
580 action_work_item_destroy(work_item
);
586 health_code_update();
587 pthread_mutex_lock(&executor
->work
.lock
);
590 if (executor
->should_quit
) {
591 pthread_mutex_unlock(&executor
->work
.lock
);
593 DBG("Left work execution loop");
595 health_code_update();
597 rcu_thread_offline();
598 rcu_unregister_thread();
599 health_unregister(health_sessiond
);
604 static bool shutdown_action_executor_thread(void *_data
)
606 struct action_executor
*executor
= _data
;
608 pthread_mutex_lock(&executor
->work
.lock
);
609 executor
->should_quit
= true;
610 pthread_cond_signal(&executor
->work
.cond
);
611 pthread_mutex_unlock(&executor
->work
.lock
);
615 static void clean_up_action_executor_thread(void *_data
)
617 struct action_executor
*executor
= _data
;
619 assert(cds_list_empty(&executor
->work
.list
));
621 pthread_mutex_destroy(&executor
->work
.lock
);
622 pthread_cond_destroy(&executor
->work
.cond
);
626 struct action_executor
*action_executor_create(
627 struct notification_thread_handle
*handle
)
629 struct action_executor
*executor
= zmalloc(sizeof(*executor
));
635 CDS_INIT_LIST_HEAD(&executor
->work
.list
);
636 pthread_cond_init(&executor
->work
.cond
, NULL
);
637 pthread_mutex_init(&executor
->work
.lock
, NULL
);
638 executor
->notification_thread_handle
= handle
;
640 executor
->thread
= lttng_thread_create(THREAD_NAME
,
641 action_executor_thread
, shutdown_action_executor_thread
,
642 clean_up_action_executor_thread
, executor
);
647 void action_executor_destroy(struct action_executor
*executor
)
649 struct action_work_item
*work_item
, *tmp
;
651 /* TODO Wait for work list to drain? */
652 lttng_thread_shutdown(executor
->thread
);
653 pthread_mutex_lock(&executor
->work
.lock
);
654 if (executor
->work
.pending_count
!= 0) {
656 " trigger action%s still queued for execution and will be discarded",
657 executor
->work
.pending_count
,
658 executor
->work
.pending_count
== 1 ? " is" :
662 cds_list_for_each_entry_safe (
663 work_item
, tmp
, &executor
->work
.list
, list_node
) {
664 WARN("Discarding action work item %" PRIu64
665 " associated to trigger `%s`",
666 work_item
->id
, get_trigger_name(work_item
->trigger
));
667 cds_list_del(&work_item
->list_node
);
668 action_work_item_destroy(work_item
);
670 pthread_mutex_unlock(&executor
->work
.lock
);
671 lttng_thread_put(executor
->thread
);
674 /* RCU read-lock must be held by the caller. */
675 enum action_executor_status
action_executor_enqueue(
676 struct action_executor
*executor
,
677 struct lttng_trigger
*trigger
,
678 struct lttng_evaluation
*evaluation
,
679 const struct lttng_credentials
*object_creds
,
680 struct notification_client_list
*client_list
)
682 enum action_executor_status executor_status
= ACTION_EXECUTOR_STATUS_OK
;
683 const uint64_t work_item_id
= executor
->next_work_item_id
++;
684 struct action_work_item
*work_item
;
687 pthread_mutex_lock(&executor
->work
.lock
);
688 /* Check for queue overflow. */
689 if (executor
->work
.pending_count
>= MAX_QUEUED_WORK_COUNT
) {
690 /* Most likely spammy, remove if it is the case. */
691 DBG("Refusing to enqueue action for trigger `%s` as work item %" PRIu64
692 " (overflow)", get_trigger_name(trigger
), work_item_id
);
693 executor_status
= ACTION_EXECUTOR_STATUS_OVERFLOW
;
697 work_item
= zmalloc(sizeof(*work_item
));
699 PERROR("Failed to allocate action executor work item on behalf of trigger `%s`",
700 get_trigger_name(trigger
));
701 executor_status
= ACTION_EXECUTOR_STATUS_ERROR
;
705 lttng_trigger_get(trigger
);
707 const bool reference_acquired
=
708 notification_client_list_get(client_list
);
710 assert(reference_acquired
);
713 *work_item
= (typeof(*work_item
)){
716 /* Ownership transferred to the work item. */
717 .evaluation
= evaluation
,
719 .is_set
= !!object_creds
,
720 .value
= object_creds
? *object_creds
:
721 (typeof(work_item
->object_creds
.value
)) {},
723 .client_list
= client_list
,
724 .list_node
= CDS_LIST_HEAD_INIT(work_item
->list_node
),
728 cds_list_add_tail(&work_item
->list_node
, &executor
->work
.list
);
729 executor
->work
.pending_count
++;
730 DBG("Enqueued action for trigger `%s` as work item #%" PRIu64
,
731 get_trigger_name(trigger
), work_item_id
);
736 pthread_cond_signal(&executor
->work
.cond
);
738 pthread_mutex_unlock(&executor
->work
.lock
);
740 lttng_evaluation_destroy(evaluation
);
741 return executor_status
;