cc7834f2745ca6e58d8299dd7c649fe68ecc0aaa
[lttng-tools.git] / src / bin / lttng-sessiond / action-executor.cpp
1 /*
2 * Copyright (C) 2020 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 *
4 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8 #include "action-executor.h"
9 #include "cmd.h"
10 #include "health-sessiond.h"
11 #include "lttng-sessiond.h"
12 #include "notification-thread-internal.h"
13 #include "session.h"
14 #include "thread.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>
31 #include <pthread.h>
32 #include <stdbool.h>
33 #include <stddef.h>
34 #include <urcu/list.h>
35
36 #define THREAD_NAME "Action Executor"
37 #define MAX_QUEUED_WORK_COUNT 8192
38
39 /*
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.
42 *
43 * We cannot rely solely on the trigger's actions since each action can have an
44 * execution context we need to comply with.
45 *
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.
49 *
50 * For example, let's take a simple trigger such as:
51 * - condition: ust event a
52 * - action: start session S
53 *
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.
63 *
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.
67 *
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.
71 */
72
73 struct action_work_item {
74 uint64_t id;
75
76 /*
77 * The actions to be executed with their respective execution context.
78 * See struct `action_work_subitem`.
79 */
80 struct lttng_dynamic_array subitems;
81
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;
88 };
89
90 struct action_work_subitem {
91 struct lttng_action *action;
92 struct {
93 /* Used by actions targeting a session. */
94 LTTNG_OPTIONAL(uint64_t) session_id;
95 } context;
96 };
97
98 struct action_executor {
99 struct lttng_thread *thread;
100 struct notification_thread_handle *notification_thread_handle;
101 struct {
102 uint64_t pending_count;
103 struct cds_list_head list;
104 pthread_cond_t cond;
105 pthread_mutex_t lock;
106 } work;
107 bool should_quit;
108 uint64_t next_work_item_id;
109 };
110
111 /*
112 * Only return non-zero on a fatal error that should shut down the action
113 * executor.
114 */
115 typedef int (*action_executor_handler)(struct action_executor *executor,
116 const struct action_work_item *,
117 struct action_work_subitem *item);
118
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 *);
144
145 static const action_executor_handler action_executors[] = {
146 action_executor_notify_handler,
147 action_executor_start_session_handler,
148 action_executor_stop_session_handler,
149 action_executor_rotate_session_handler,
150 action_executor_snapshot_session_handler,
151 action_executor_list_handler,
152 };
153
154 /* Forward declaration */
155 static int add_action_to_subitem_array(struct lttng_action *action,
156 struct lttng_dynamic_array *subitems);
157
158 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
159 struct lttng_dynamic_array *subitems);
160
161 static void action_work_subitem_destructor(void *element)
162 {
163 struct action_work_subitem *subitem = (action_work_subitem *) element;
164
165 lttng_action_put(subitem->action);
166 }
167
168 static const char *get_action_name(const struct lttng_action *action)
169 {
170 const enum lttng_action_type action_type = lttng_action_get_type(action);
171
172 LTTNG_ASSERT(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
173
174 return lttng_action_type_string(action_type);
175 }
176
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)
180 {
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),
185 };
186 /* Can never be NULL. */
187 const struct lttng_credentials *trigger_creds =
188 lttng_trigger_get_credentials(trigger);
189
190 is_allowed = (lttng_credentials_is_equal_uid(trigger_creds, &session_creds)) ||
191 (lttng_credentials_get_uid(trigger_creds) == 0);
192 if (!is_allowed) {
193 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
194 session->name,
195 (long int) session->uid,
196 (long int) session->gid,
197 (long int) lttng_credentials_get_uid(trigger_creds));
198 }
199
200 return is_allowed;
201 }
202
203 static const char *get_trigger_name(const struct lttng_trigger *trigger)
204 {
205 const char *trigger_name;
206 enum lttng_trigger_status trigger_status;
207
208 trigger_status = lttng_trigger_get_name(trigger, &trigger_name);
209 switch (trigger_status) {
210 case LTTNG_TRIGGER_STATUS_OK:
211 break;
212 case LTTNG_TRIGGER_STATUS_UNSET:
213 trigger_name = "(anonymous)";
214 break;
215 default:
216 trigger_name = "(failed to get name)";
217 break;
218 }
219
220 return trigger_name;
221 }
222
223 static int client_handle_transmission_status(
224 struct notification_client *client,
225 enum client_transmission_status status,
226 void *user_data)
227 {
228 int ret = 0;
229 struct action_executor *executor = (action_executor *) user_data;
230 bool update_communication = true;
231
232 switch (status) {
233 case CLIENT_TRANSMISSION_STATUS_COMPLETE:
234 DBG("Successfully sent full notification to client, client_id = %" PRIu64,
235 client->id);
236 update_communication = false;
237 break;
238 case CLIENT_TRANSMISSION_STATUS_QUEUED:
239 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64,
240 client->id);
241 break;
242 case CLIENT_TRANSMISSION_STATUS_FAIL:
243 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64,
244 client->id);
245 break;
246 default:
247 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64,
248 client->id);
249 ret = -1;
250 goto end;
251 }
252
253 if (!update_communication) {
254 goto end;
255 }
256
257 /* Safe to read client's id without locking as it is immutable. */
258 ret = notification_thread_client_communication_update(
259 executor->notification_thread_handle, client->id,
260 status);
261 end:
262 return ret;
263 }
264
265 static int action_executor_notify_handler(struct action_executor *executor,
266 const struct action_work_item *work_item,
267 struct action_work_subitem *item)
268 {
269 return notification_client_list_send_evaluation(work_item->client_list,
270 work_item->trigger,
271 work_item->evaluation,
272 work_item->object_creds.is_set ?
273 &(work_item->object_creds.value) :
274 NULL,
275 client_handle_transmission_status, executor);
276 }
277
278 static int action_executor_start_session_handler(
279 struct action_executor *executor,
280 const struct action_work_item *work_item,
281 struct action_work_subitem *item)
282 {
283 int ret = 0;
284 const char *session_name;
285 enum lttng_action_status action_status;
286 struct ltt_session *session;
287 enum lttng_error_code cmd_ret;
288 struct lttng_action *action = item->action;
289
290 action_status = lttng_action_start_session_get_session_name(
291 action, &session_name);
292 if (action_status != LTTNG_ACTION_STATUS_OK) {
293 ERR("Failed to get session name from `%s` action",
294 get_action_name(action));
295 ret = -1;
296 goto end;
297 }
298
299 /*
300 * Validate if at the moment of the action was queued the session
301 * existed. If not skip the action altogether.
302 */
303 if (!item->context.session_id.is_set) {
304 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
305 session_name, get_action_name(action),
306 get_trigger_name(work_item->trigger));
307 lttng_action_increase_execution_failure_count(action);
308 ret = 0;
309 goto end;
310 }
311
312 session_lock_list();
313 session = session_find_by_name(session_name);
314 if (!session) {
315 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
316 session_name, get_action_name(action),
317 get_trigger_name(work_item->trigger));
318 goto error_unlock_list;
319 }
320
321 /*
322 * Check if the session id is the same as when the work item was
323 * enqueued.
324 */
325 if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
326 DBG("Session id for session `%s` (id: %" PRIu64
327 " is not the same that was sampled (id: %" PRIu64
328 " at the moment the work item was enqueued for %s` action of trigger `%s`",
329 session_name, session->id,
330 LTTNG_OPTIONAL_GET(item->context.session_id),
331 get_action_name(action),
332 get_trigger_name(work_item->trigger));
333 ret = 0;
334 goto error_unlock_list;
335 }
336
337 session_lock(session);
338 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
339 goto error_dispose_session;
340 }
341
342 cmd_ret = (lttng_error_code) cmd_start_trace(session);
343 switch (cmd_ret) {
344 case LTTNG_OK:
345 DBG("Successfully started session `%s` on behalf of trigger `%s`",
346 session_name, get_trigger_name(work_item->trigger));
347 break;
348 case LTTNG_ERR_TRACE_ALREADY_STARTED:
349 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
350 session_name, get_trigger_name(work_item->trigger));
351 break;
352 default:
353 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
354 session_name, get_trigger_name(work_item->trigger),
355 lttng_strerror(-cmd_ret));
356 lttng_action_increase_execution_failure_count(action);
357 break;
358 }
359
360 error_dispose_session:
361 session_unlock(session);
362 session_put(session);
363 error_unlock_list:
364 session_unlock_list();
365 end:
366 return ret;
367 }
368
369 static int action_executor_stop_session_handler(
370 struct action_executor *executor,
371 const struct action_work_item *work_item,
372 struct action_work_subitem *item)
373 {
374 int ret = 0;
375 const char *session_name;
376 enum lttng_action_status action_status;
377 struct ltt_session *session;
378 enum lttng_error_code cmd_ret;
379 struct lttng_action *action = item->action;
380
381 action_status = lttng_action_stop_session_get_session_name(
382 action, &session_name);
383 if (action_status != LTTNG_ACTION_STATUS_OK) {
384 ERR("Failed to get session name from `%s` action",
385 get_action_name(action));
386 ret = -1;
387 goto end;
388 }
389
390 /*
391 * Validate if, at the moment the action was queued, the target session
392 * existed. If not, skip the action altogether.
393 */
394 if (!item->context.session_id.is_set) {
395 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
396 session_name, get_action_name(action),
397 get_trigger_name(work_item->trigger));
398 lttng_action_increase_execution_failure_count(action);
399 ret = 0;
400 goto end;
401 }
402
403 session_lock_list();
404 session = session_find_by_name(session_name);
405 if (!session) {
406 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
407 session_name, get_action_name(action),
408 get_trigger_name(work_item->trigger));
409 lttng_action_increase_execution_failure_count(action);
410 goto error_unlock_list;
411 }
412
413 /*
414 * Check if the session id is the same as when the work item was
415 * enqueued
416 */
417 if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
418 DBG("Session id for session `%s` (id: %" PRIu64
419 " is not the same that was sampled (id: %" PRIu64
420 " at the moment the work item was enqueued for %s` action of trigger `%s`",
421 session_name, session->id,
422 LTTNG_OPTIONAL_GET(item->context.session_id),
423 get_action_name(action),
424 get_trigger_name(work_item->trigger));
425 ret = 0;
426 goto error_unlock_list;
427 }
428
429 session_lock(session);
430 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
431 goto error_dispose_session;
432 }
433
434 cmd_ret = (lttng_error_code) cmd_stop_trace(session);
435 switch (cmd_ret) {
436 case LTTNG_OK:
437 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
438 session_name, get_trigger_name(work_item->trigger));
439 break;
440 case LTTNG_ERR_TRACE_ALREADY_STOPPED:
441 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
442 session_name, get_trigger_name(work_item->trigger));
443 break;
444 default:
445 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
446 session_name, get_trigger_name(work_item->trigger),
447 lttng_strerror(-cmd_ret));
448 lttng_action_increase_execution_failure_count(action);
449 break;
450 }
451
452 error_dispose_session:
453 session_unlock(session);
454 session_put(session);
455 error_unlock_list:
456 session_unlock_list();
457 end:
458 return ret;
459 }
460
461 static int action_executor_rotate_session_handler(
462 struct action_executor *executor,
463 const struct action_work_item *work_item,
464 struct action_work_subitem *item)
465 {
466 int ret = 0;
467 const char *session_name;
468 enum lttng_action_status action_status;
469 struct ltt_session *session;
470 enum lttng_error_code cmd_ret;
471 struct lttng_action *action = item->action;
472
473 action_status = lttng_action_rotate_session_get_session_name(
474 action, &session_name);
475 if (action_status != LTTNG_ACTION_STATUS_OK) {
476 ERR("Failed to get session name from `%s` action",
477 get_action_name(action));
478 ret = -1;
479 goto end;
480 }
481
482 /*
483 * Validate if, at the moment the action was queued, the target session
484 * existed. If not, skip the action altogether.
485 */
486 if (!item->context.session_id.is_set) {
487 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
488 session_name, get_action_name(action),
489 get_trigger_name(work_item->trigger));
490 lttng_action_increase_execution_failure_count(action);
491 ret = 0;
492 goto end;
493 }
494
495 session_lock_list();
496 session = session_find_by_name(session_name);
497 if (!session) {
498 DBG("Failed to find session `%s` by name while executing `%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);
502 goto error_unlock_list;
503 }
504
505 /*
506 * Check if the session id is the same as when the work item was
507 * enqueued.
508 */
509 if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
510 DBG("Session id for session `%s` (id: %" PRIu64
511 " is not the same that was sampled (id: %" PRIu64
512 " at the moment the work item was enqueued for %s` action of trigger `%s`",
513 session_name, session->id,
514 LTTNG_OPTIONAL_GET(item->context.session_id),
515 get_action_name(action),
516 get_trigger_name(work_item->trigger));
517 ret = 0;
518 goto error_unlock_list;
519 }
520
521 session_lock(session);
522 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
523 goto error_dispose_session;
524 }
525
526 cmd_ret = (lttng_error_code) cmd_rotate_session(session, NULL, false,
527 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
528 switch (cmd_ret) {
529 case LTTNG_OK:
530 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
531 session_name, get_trigger_name(work_item->trigger));
532 break;
533 case LTTNG_ERR_ROTATION_PENDING:
534 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
535 session_name, get_trigger_name(work_item->trigger));
536 lttng_action_increase_execution_failure_count(action);
537 break;
538 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
539 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
540 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",
541 session_name, get_trigger_name(work_item->trigger));
542 break;
543 default:
544 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
545 session_name, get_trigger_name(work_item->trigger),
546 lttng_strerror(-cmd_ret));
547 lttng_action_increase_execution_failure_count(action);
548 break;
549 }
550
551 error_dispose_session:
552 session_unlock(session);
553 session_put(session);
554 error_unlock_list:
555 session_unlock_list();
556 end:
557 return ret;
558 }
559
560 static int action_executor_snapshot_session_handler(
561 struct action_executor *executor,
562 const struct action_work_item *work_item,
563 struct action_work_subitem *item)
564 {
565 int ret = 0;
566 const char *session_name;
567 enum lttng_action_status action_status;
568 struct ltt_session *session;
569 lttng_snapshot_output default_snapshot_output;
570 const struct lttng_snapshot_output *snapshot_output =
571 &default_snapshot_output;
572 enum lttng_error_code cmd_ret;
573 struct lttng_action *action = item->action;
574
575 default_snapshot_output.max_size = UINT64_MAX;
576
577 /*
578 * Validate if, at the moment the action was queued, the target session
579 * existed. If not, skip the action altogether.
580 */
581 if (!item->context.session_id.is_set) {
582 DBG("Session was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
583 get_action_name(action),
584 get_trigger_name(work_item->trigger));
585 lttng_action_increase_execution_failure_count(action);
586 ret = 0;
587 goto end;
588 }
589
590 action_status = lttng_action_snapshot_session_get_session_name(
591 action, &session_name);
592 if (action_status != LTTNG_ACTION_STATUS_OK) {
593 ERR("Failed to get session name from `%s` action",
594 get_action_name(action));
595 ret = -1;
596 goto end;
597 }
598
599 action_status = lttng_action_snapshot_session_get_output(
600 action, &snapshot_output);
601 if (action_status != LTTNG_ACTION_STATUS_OK &&
602 action_status != LTTNG_ACTION_STATUS_UNSET) {
603 ERR("Failed to get output from `%s` action",
604 get_action_name(action));
605 ret = -1;
606 goto end;
607 }
608
609 session_lock_list();
610 session = session_find_by_name(session_name);
611 if (!session) {
612 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
613 session_name, get_action_name(action),
614 get_trigger_name(work_item->trigger));
615 lttng_action_increase_execution_failure_count(action);
616 goto error_unlock_list;
617 }
618
619 /*
620 * Check if the session id is the same as when the work item was
621 * enqueued.
622 */
623 if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
624 DBG("Session id for session `%s` (id: %" PRIu64
625 " is not the same that was sampled (id: %" PRIu64
626 " at the moment the work item was enqueued for %s` action of trigger `%s`",
627 session_name, session->id,
628 LTTNG_OPTIONAL_GET(item->context.session_id),
629 get_action_name(action),
630 get_trigger_name(work_item->trigger));
631 ret = 0;
632 goto error_unlock_list;
633 }
634
635 session_lock(session);
636 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
637 goto error_dispose_session;
638 }
639
640 cmd_ret = (lttng_error_code) cmd_snapshot_record(session, snapshot_output, 0);
641 switch (cmd_ret) {
642 case LTTNG_OK:
643 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
644 session_name, get_trigger_name(work_item->trigger));
645 break;
646 default:
647 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
648 session_name, get_trigger_name(work_item->trigger),
649 lttng_strerror(-cmd_ret));
650 lttng_action_increase_execution_failure_count(action);
651 break;
652 }
653
654 error_dispose_session:
655 session_unlock(session);
656 session_put(session);
657 error_unlock_list:
658 session_unlock_list();
659 end:
660 return ret;
661 }
662
663 static int action_executor_list_handler(struct action_executor *executor,
664 const struct action_work_item *work_item,
665 struct action_work_subitem *item)
666 {
667 ERR("Execution of a list action by the action executor should never occur");
668 abort();
669 }
670
671 static int action_executor_generic_handler(struct action_executor *executor,
672 const struct action_work_item *work_item,
673 struct action_work_subitem *item)
674 {
675 int ret;
676 struct lttng_action *action = item->action;
677 const enum lttng_action_type action_type = lttng_action_get_type(action);
678
679 LTTNG_ASSERT(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
680
681 lttng_action_increase_execution_request_count(action);
682 if (!lttng_action_should_execute(action)) {
683 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64,
684 get_action_name(action),
685 get_trigger_name(work_item->trigger),
686 work_item->id);
687 ret = 0;
688 goto end;
689 }
690
691 lttng_action_increase_execution_count(action);
692 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64,
693 get_action_name(action),
694 get_trigger_name(work_item->trigger),
695 work_item->id);
696 ret = action_executors[action_type](executor, work_item, item);
697 end:
698 return ret;
699 }
700
701 static int action_work_item_execute(struct action_executor *executor,
702 struct action_work_item *work_item)
703 {
704 int ret;
705 size_t count, i;
706
707 DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
708 work_item->id, get_trigger_name(work_item->trigger));
709
710 count = lttng_dynamic_array_get_count(&work_item->subitems);
711 for (i = 0; i < count; i++) {
712 struct action_work_subitem *item;
713
714 item = (action_work_subitem *) lttng_dynamic_array_get_element(&work_item->subitems, i);
715 ret = action_executor_generic_handler(
716 executor, work_item, item);
717 if (ret) {
718 goto end;
719 }
720 }
721 end:
722 DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
723 work_item->id, get_trigger_name(work_item->trigger));
724 return ret;
725 }
726
727 static void action_work_item_destroy(struct action_work_item *work_item)
728 {
729 lttng_trigger_put(work_item->trigger);
730 lttng_evaluation_destroy(work_item->evaluation);
731 notification_client_list_put(work_item->client_list);
732 lttng_dynamic_array_reset(&work_item->subitems);
733 free(work_item);
734 }
735
736 static void *action_executor_thread(void *_data)
737 {
738 struct action_executor *executor = (action_executor *) _data;
739
740 LTTNG_ASSERT(executor);
741
742 health_register(the_health_sessiond,
743 HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
744
745 rcu_register_thread();
746 rcu_thread_online();
747
748 DBG("Entering work execution loop");
749 pthread_mutex_lock(&executor->work.lock);
750 while (!executor->should_quit) {
751 int ret = 0;
752 struct action_work_item *work_item;
753
754 health_code_update();
755 if (executor->work.pending_count == 0) {
756 health_poll_entry();
757 DBG("No work items enqueued, entering wait");
758 pthread_cond_wait(&executor->work.cond,
759 &executor->work.lock);
760 DBG("Woke-up from wait");
761 health_poll_exit();
762 continue;
763 }
764
765 /* Pop item from front of the list with work lock held. */
766 work_item = cds_list_first_entry(&executor->work.list,
767 struct action_work_item, list_node);
768 cds_list_del(&work_item->list_node);
769 executor->work.pending_count--;
770
771 /*
772 * Work can be performed without holding the work lock,
773 * allowing new items to be queued.
774 */
775 pthread_mutex_unlock(&executor->work.lock);
776
777 /* Execute item only if a trigger is registered. */
778 lttng_trigger_lock(work_item->trigger);
779 if (!lttng_trigger_is_registered(work_item->trigger)) {
780 const char *trigger_name = NULL;
781 uid_t trigger_owner_uid;
782 enum lttng_trigger_status trigger_status;
783
784 trigger_name = get_trigger_name(work_item->trigger);
785
786 trigger_status = lttng_trigger_get_owner_uid(
787 work_item->trigger, &trigger_owner_uid);
788 LTTNG_ASSERT(trigger_status == LTTNG_TRIGGER_STATUS_OK);
789
790 DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64 ", trigger name = '%s', trigger owner uid = %d",
791 work_item->id, trigger_name,
792 (int) trigger_owner_uid);
793 ret = 0;
794 goto skip_execute;
795 }
796
797 ret = action_work_item_execute(executor, work_item);
798
799 skip_execute:
800 lttng_trigger_unlock(work_item->trigger);
801 action_work_item_destroy(work_item);
802 if (ret) {
803 /* Fatal error. */
804 break;
805 }
806
807 health_code_update();
808 pthread_mutex_lock(&executor->work.lock);
809 }
810
811 if (executor->should_quit) {
812 pthread_mutex_unlock(&executor->work.lock);
813 }
814 DBG("Left work execution loop");
815
816 health_code_update();
817
818 rcu_thread_offline();
819 rcu_unregister_thread();
820 health_unregister(the_health_sessiond);
821
822 return NULL;
823 }
824
825 static bool shutdown_action_executor_thread(void *_data)
826 {
827 struct action_executor *executor = (action_executor *) _data;
828
829 pthread_mutex_lock(&executor->work.lock);
830 executor->should_quit = true;
831 pthread_cond_signal(&executor->work.cond);
832 pthread_mutex_unlock(&executor->work.lock);
833 return true;
834 }
835
836 static void clean_up_action_executor_thread(void *_data)
837 {
838 struct action_executor *executor = (action_executor *) _data;
839
840 LTTNG_ASSERT(cds_list_empty(&executor->work.list));
841
842 pthread_mutex_destroy(&executor->work.lock);
843 pthread_cond_destroy(&executor->work.cond);
844 free(executor);
845 }
846
847 struct action_executor *action_executor_create(
848 struct notification_thread_handle *handle)
849 {
850 struct action_executor *executor = (action_executor *) zmalloc(sizeof(*executor));
851
852 if (!executor) {
853 goto end;
854 }
855
856 CDS_INIT_LIST_HEAD(&executor->work.list);
857 pthread_cond_init(&executor->work.cond, NULL);
858 pthread_mutex_init(&executor->work.lock, NULL);
859 executor->notification_thread_handle = handle;
860
861 executor->thread = lttng_thread_create(THREAD_NAME,
862 action_executor_thread, shutdown_action_executor_thread,
863 clean_up_action_executor_thread, executor);
864 end:
865 return executor;
866 }
867
868 void action_executor_destroy(struct action_executor *executor)
869 {
870 struct action_work_item *work_item, *tmp;
871
872 /* TODO Wait for work list to drain? */
873 lttng_thread_shutdown(executor->thread);
874 pthread_mutex_lock(&executor->work.lock);
875 if (executor->work.pending_count != 0) {
876 WARN("%" PRIu64
877 " trigger action%s still queued for execution and will be discarded",
878 executor->work.pending_count,
879 executor->work.pending_count == 1 ? " is" :
880 "s are");
881 }
882
883 cds_list_for_each_entry_safe (
884 work_item, tmp, &executor->work.list, list_node) {
885 WARN("Discarding action work item %" PRIu64
886 " associated to trigger `%s`",
887 work_item->id, get_trigger_name(work_item->trigger));
888 cds_list_del(&work_item->list_node);
889 action_work_item_destroy(work_item);
890 }
891 pthread_mutex_unlock(&executor->work.lock);
892 lttng_thread_put(executor->thread);
893 }
894
895 /* RCU read-lock must be held by the caller. */
896 enum action_executor_status action_executor_enqueue_trigger(
897 struct action_executor *executor,
898 struct lttng_trigger *trigger,
899 struct lttng_evaluation *evaluation,
900 const struct lttng_credentials *object_creds,
901 struct notification_client_list *client_list)
902 {
903 int ret;
904 enum action_executor_status executor_status = ACTION_EXECUTOR_STATUS_OK;
905 const uint64_t work_item_id = executor->next_work_item_id++;
906 struct action_work_item *work_item;
907 bool signal = false;
908
909 LTTNG_ASSERT(trigger);
910
911 pthread_mutex_lock(&executor->work.lock);
912 /* Check for queue overflow. */
913 if (executor->work.pending_count >= MAX_QUEUED_WORK_COUNT) {
914 /* Most likely spammy, remove if it is the case. */
915 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64,
916 get_trigger_name(trigger), work_item_id);
917 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
918 goto error_unlock;
919 }
920
921 work_item = (action_work_item *) zmalloc(sizeof(*work_item));
922 if (!work_item) {
923 PERROR("Failed to allocate action executor work item: trigger name = '%s'",
924 get_trigger_name(trigger));
925 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
926 goto error_unlock;
927 }
928
929 lttng_trigger_get(trigger);
930 if (client_list) {
931 const bool reference_acquired =
932 notification_client_list_get(client_list);
933
934 LTTNG_ASSERT(reference_acquired);
935 }
936
937 work_item->id = work_item_id;
938 work_item->trigger = trigger;
939
940 /* Ownership transferred to the work item. */
941 work_item->evaluation = evaluation;
942 evaluation = NULL;
943
944 work_item->client_list = client_list;
945 work_item->object_creds.is_set = !!object_creds;
946 if (object_creds) {
947 work_item->object_creds.value = *object_creds;
948 }
949
950 CDS_INIT_LIST_HEAD(&work_item->list_node);
951
952 /* Build the array of action work subitems for the passed trigger. */
953 lttng_dynamic_array_init(&work_item->subitems,
954 sizeof(struct action_work_subitem),
955 action_work_subitem_destructor);
956
957 ret = populate_subitem_array_from_trigger(
958 trigger, &work_item->subitems);
959 if (ret) {
960 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
961 get_trigger_name(trigger));
962 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
963 goto error_unlock;
964 }
965
966 cds_list_add_tail(&work_item->list_node, &executor->work.list);
967 executor->work.pending_count++;
968 DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64,
969 get_trigger_name(trigger), work_item_id);
970 signal = true;
971
972 error_unlock:
973 if (signal) {
974 pthread_cond_signal(&executor->work.cond);
975 }
976
977 pthread_mutex_unlock(&executor->work.lock);
978 lttng_evaluation_destroy(evaluation);
979 return executor_status;
980 }
981
982 static int add_action_to_subitem_array(struct lttng_action *action,
983 struct lttng_dynamic_array *subitems)
984 {
985 int ret = 0;
986 enum lttng_action_type type = lttng_action_get_type(action);
987 const char *session_name = NULL;
988 enum lttng_action_status status;
989 struct action_work_subitem subitem = {
990 .action = NULL,
991 .context = {
992 .session_id = LTTNG_OPTIONAL_INIT_UNSET,
993 },
994 };
995
996 LTTNG_ASSERT(action);
997 LTTNG_ASSERT(subitems);
998
999 if (type == LTTNG_ACTION_TYPE_LIST) {
1000 unsigned int count, i;
1001
1002 status = lttng_action_list_get_count(action, &count);
1003 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1004
1005 for (i = 0; i < count; i++) {
1006 struct lttng_action *inner_action = NULL;
1007
1008 inner_action = lttng_action_list_borrow_mutable_at_index(
1009 action, i);
1010 LTTNG_ASSERT(inner_action);
1011 ret = add_action_to_subitem_array(
1012 inner_action, subitems);
1013 if (ret) {
1014 goto end;
1015 }
1016 }
1017
1018 /*
1019 * Go directly to the end since there is no need to add the
1020 * list action by itself to the subitems array.
1021 */
1022 goto end;
1023 }
1024
1025 /* Gather execution context. */
1026 switch (type) {
1027 case LTTNG_ACTION_TYPE_NOTIFY:
1028 break;
1029 case LTTNG_ACTION_TYPE_START_SESSION:
1030 status = lttng_action_start_session_get_session_name(
1031 action, &session_name);
1032 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1033 break;
1034 case LTTNG_ACTION_TYPE_STOP_SESSION:
1035 status = lttng_action_stop_session_get_session_name(
1036 action, &session_name);
1037 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1038 break;
1039 case LTTNG_ACTION_TYPE_ROTATE_SESSION:
1040 status = lttng_action_rotate_session_get_session_name(
1041 action, &session_name);
1042 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1043 break;
1044 case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION:
1045 status = lttng_action_snapshot_session_get_session_name(
1046 action, &session_name);
1047 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1048 break;
1049 case LTTNG_ACTION_TYPE_LIST:
1050 case LTTNG_ACTION_TYPE_UNKNOWN:
1051 /* Fallthrough */
1052 default:
1053 abort();
1054 break;
1055 }
1056
1057 /*
1058 * Fetch the session execution context info as needed.
1059 * Note that we could decide to not add an action for which we know the
1060 * execution will not happen (i.e no session exists for that name). For
1061 * now we leave the decision to skip to the action executor for sake of
1062 * simplicity and consistency.
1063 */
1064 if (session_name != NULL) {
1065 uint64_t session_id;
1066
1067 /*
1068 * Instantaneous sampling of the session id if present.
1069 *
1070 * This method is preferred over `sessiond_find_by_name` then
1071 * fetching the session'd id since `sessiond_find_by_name`
1072 * requires the session list lock to be taken.
1073 *
1074 * Taking the session list lock can lead to a deadlock
1075 * between the action executor and the notification thread
1076 * (caller of add_action_to_subitem_array). It is okay if the
1077 * session state changes between the enqueuing time and the
1078 * execution time. The execution context is validated at
1079 * execution time.
1080 */
1081 if (sample_session_id_by_name(session_name, &session_id)) {
1082 LTTNG_OPTIONAL_SET(&subitem.context.session_id,
1083 session_id);
1084 }
1085 }
1086
1087 /* Get a reference to the action. */
1088 lttng_action_get(action);
1089 subitem.action = action;
1090
1091 ret = lttng_dynamic_array_add_element(subitems, &subitem);
1092 if (ret) {
1093 ERR("Failed to add work subitem to the subitem array");
1094 lttng_action_put(action);
1095 ret = -1;
1096 goto end;
1097 }
1098
1099 end:
1100 return ret;
1101 }
1102
1103 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
1104 struct lttng_dynamic_array *subitems)
1105 {
1106 struct lttng_action *action;
1107
1108 action = lttng_trigger_get_action(trigger);
1109 LTTNG_ASSERT(action);
1110
1111 return add_action_to_subitem_array(action, subitems);
1112 }
This page took 0.088899 seconds and 3 git commands to generate.