Fix: action executor: ref count imbalance for session object
[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_put_session;
335 }
336
337 session_lock(session);
338 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
339 goto error_unlock_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_unlock_session:
361 session_unlock(session);
362 error_put_session:
363 session_put(session);
364 error_unlock_list:
365 session_unlock_list();
366 end:
367 return ret;
368 }
369
370 static int action_executor_stop_session_handler(
371 struct action_executor *executor,
372 const struct action_work_item *work_item,
373 struct action_work_subitem *item)
374 {
375 int ret = 0;
376 const char *session_name;
377 enum lttng_action_status action_status;
378 struct ltt_session *session;
379 enum lttng_error_code cmd_ret;
380 struct lttng_action *action = item->action;
381
382 action_status = lttng_action_stop_session_get_session_name(
383 action, &session_name);
384 if (action_status != LTTNG_ACTION_STATUS_OK) {
385 ERR("Failed to get session name from `%s` action",
386 get_action_name(action));
387 ret = -1;
388 goto end;
389 }
390
391 /*
392 * Validate if, at the moment the action was queued, the target session
393 * existed. If not, skip the action altogether.
394 */
395 if (!item->context.session_id.is_set) {
396 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
397 session_name, get_action_name(action),
398 get_trigger_name(work_item->trigger));
399 lttng_action_increase_execution_failure_count(action);
400 ret = 0;
401 goto end;
402 }
403
404 session_lock_list();
405 session = session_find_by_name(session_name);
406 if (!session) {
407 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
408 session_name, get_action_name(action),
409 get_trigger_name(work_item->trigger));
410 lttng_action_increase_execution_failure_count(action);
411 goto error_unlock_list;
412 }
413
414 /*
415 * Check if the session id is the same as when the work item was
416 * enqueued
417 */
418 if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
419 DBG("Session id for session `%s` (id: %" PRIu64
420 " is not the same that was sampled (id: %" PRIu64
421 " at the moment the work item was enqueued for %s` action of trigger `%s`",
422 session_name, session->id,
423 LTTNG_OPTIONAL_GET(item->context.session_id),
424 get_action_name(action),
425 get_trigger_name(work_item->trigger));
426 ret = 0;
427 goto error_put_session;
428 }
429
430 session_lock(session);
431 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
432 goto error_unlock_session;
433 }
434
435 cmd_ret = (lttng_error_code) cmd_stop_trace(session);
436 switch (cmd_ret) {
437 case LTTNG_OK:
438 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
439 session_name, get_trigger_name(work_item->trigger));
440 break;
441 case LTTNG_ERR_TRACE_ALREADY_STOPPED:
442 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
443 session_name, get_trigger_name(work_item->trigger));
444 break;
445 default:
446 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
447 session_name, get_trigger_name(work_item->trigger),
448 lttng_strerror(-cmd_ret));
449 lttng_action_increase_execution_failure_count(action);
450 break;
451 }
452
453 error_unlock_session:
454 session_unlock(session);
455 error_put_session:
456 session_put(session);
457 error_unlock_list:
458 session_unlock_list();
459 end:
460 return ret;
461 }
462
463 static int action_executor_rotate_session_handler(
464 struct action_executor *executor,
465 const struct action_work_item *work_item,
466 struct action_work_subitem *item)
467 {
468 int ret = 0;
469 const char *session_name;
470 enum lttng_action_status action_status;
471 struct ltt_session *session;
472 enum lttng_error_code cmd_ret;
473 struct lttng_action *action = item->action;
474
475 action_status = lttng_action_rotate_session_get_session_name(
476 action, &session_name);
477 if (action_status != LTTNG_ACTION_STATUS_OK) {
478 ERR("Failed to get session name from `%s` action",
479 get_action_name(action));
480 ret = -1;
481 goto end;
482 }
483
484 /*
485 * Validate if, at the moment the action was queued, the target session
486 * existed. If not, skip the action altogether.
487 */
488 if (!item->context.session_id.is_set) {
489 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
490 session_name, get_action_name(action),
491 get_trigger_name(work_item->trigger));
492 lttng_action_increase_execution_failure_count(action);
493 ret = 0;
494 goto end;
495 }
496
497 session_lock_list();
498 session = session_find_by_name(session_name);
499 if (!session) {
500 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
501 session_name, get_action_name(action),
502 get_trigger_name(work_item->trigger));
503 lttng_action_increase_execution_failure_count(action);
504 goto error_unlock_list;
505 }
506
507 /*
508 * Check if the session id is the same as when the work item was
509 * enqueued.
510 */
511 if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
512 DBG("Session id for session `%s` (id: %" PRIu64
513 " is not the same that was sampled (id: %" PRIu64
514 " at the moment the work item was enqueued for %s` action of trigger `%s`",
515 session_name, session->id,
516 LTTNG_OPTIONAL_GET(item->context.session_id),
517 get_action_name(action),
518 get_trigger_name(work_item->trigger));
519 ret = 0;
520 goto error_put_session;
521 }
522
523 session_lock(session);
524 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
525 goto error_unlock_session;
526 }
527
528 cmd_ret = (lttng_error_code) cmd_rotate_session(session, NULL, false,
529 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
530 switch (cmd_ret) {
531 case LTTNG_OK:
532 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
533 session_name, get_trigger_name(work_item->trigger));
534 break;
535 case LTTNG_ERR_ROTATION_PENDING:
536 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
537 session_name, get_trigger_name(work_item->trigger));
538 lttng_action_increase_execution_failure_count(action);
539 break;
540 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
541 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
542 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",
543 session_name, get_trigger_name(work_item->trigger));
544 break;
545 default:
546 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
547 session_name, get_trigger_name(work_item->trigger),
548 lttng_strerror(-cmd_ret));
549 lttng_action_increase_execution_failure_count(action);
550 break;
551 }
552
553 error_unlock_session:
554 session_unlock(session);
555 error_put_session:
556 session_put(session);
557 error_unlock_list:
558 session_unlock_list();
559 end:
560 return ret;
561 }
562
563 static int action_executor_snapshot_session_handler(
564 struct action_executor *executor,
565 const struct action_work_item *work_item,
566 struct action_work_subitem *item)
567 {
568 int ret = 0;
569 const char *session_name;
570 enum lttng_action_status action_status;
571 struct ltt_session *session;
572 lttng_snapshot_output default_snapshot_output;
573 const struct lttng_snapshot_output *snapshot_output =
574 &default_snapshot_output;
575 enum lttng_error_code cmd_ret;
576 struct lttng_action *action = item->action;
577
578 default_snapshot_output.max_size = UINT64_MAX;
579
580 /*
581 * Validate if, at the moment the action was queued, the target session
582 * existed. If not, skip the action altogether.
583 */
584 if (!item->context.session_id.is_set) {
585 DBG("Session was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
586 get_action_name(action),
587 get_trigger_name(work_item->trigger));
588 lttng_action_increase_execution_failure_count(action);
589 ret = 0;
590 goto end;
591 }
592
593 action_status = lttng_action_snapshot_session_get_session_name(
594 action, &session_name);
595 if (action_status != LTTNG_ACTION_STATUS_OK) {
596 ERR("Failed to get session name from `%s` action",
597 get_action_name(action));
598 ret = -1;
599 goto end;
600 }
601
602 action_status = lttng_action_snapshot_session_get_output(
603 action, &snapshot_output);
604 if (action_status != LTTNG_ACTION_STATUS_OK &&
605 action_status != LTTNG_ACTION_STATUS_UNSET) {
606 ERR("Failed to get output from `%s` action",
607 get_action_name(action));
608 ret = -1;
609 goto end;
610 }
611
612 session_lock_list();
613 session = session_find_by_name(session_name);
614 if (!session) {
615 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
616 session_name, get_action_name(action),
617 get_trigger_name(work_item->trigger));
618 lttng_action_increase_execution_failure_count(action);
619 goto error_unlock_list;
620 }
621
622 /*
623 * Check if the session id is the same as when the work item was
624 * enqueued.
625 */
626 if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
627 DBG("Session id for session `%s` (id: %" PRIu64
628 " is not the same that was sampled (id: %" PRIu64
629 " at the moment the work item was enqueued for %s` action of trigger `%s`",
630 session_name, session->id,
631 LTTNG_OPTIONAL_GET(item->context.session_id),
632 get_action_name(action),
633 get_trigger_name(work_item->trigger));
634 ret = 0;
635 goto error_put_session;
636 }
637
638 session_lock(session);
639 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
640 goto error_unlock_session;
641 }
642
643 cmd_ret = (lttng_error_code) cmd_snapshot_record(session, snapshot_output, 0);
644 switch (cmd_ret) {
645 case LTTNG_OK:
646 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
647 session_name, get_trigger_name(work_item->trigger));
648 break;
649 default:
650 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
651 session_name, get_trigger_name(work_item->trigger),
652 lttng_strerror(-cmd_ret));
653 lttng_action_increase_execution_failure_count(action);
654 break;
655 }
656
657 error_unlock_session:
658 session_unlock(session);
659 error_put_session:
660 session_put(session);
661 error_unlock_list:
662 session_unlock_list();
663 end:
664 return ret;
665 }
666
667 static int action_executor_list_handler(struct action_executor *executor,
668 const struct action_work_item *work_item,
669 struct action_work_subitem *item)
670 {
671 ERR("Execution of a list action by the action executor should never occur");
672 abort();
673 }
674
675 static int action_executor_generic_handler(struct action_executor *executor,
676 const struct action_work_item *work_item,
677 struct action_work_subitem *item)
678 {
679 int ret;
680 struct lttng_action *action = item->action;
681 const enum lttng_action_type action_type = lttng_action_get_type(action);
682
683 LTTNG_ASSERT(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
684
685 lttng_action_increase_execution_request_count(action);
686 if (!lttng_action_should_execute(action)) {
687 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64,
688 get_action_name(action),
689 get_trigger_name(work_item->trigger),
690 work_item->id);
691 ret = 0;
692 goto end;
693 }
694
695 lttng_action_increase_execution_count(action);
696 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64,
697 get_action_name(action),
698 get_trigger_name(work_item->trigger),
699 work_item->id);
700 ret = action_executors[action_type](executor, work_item, item);
701 end:
702 return ret;
703 }
704
705 static int action_work_item_execute(struct action_executor *executor,
706 struct action_work_item *work_item)
707 {
708 int ret;
709 size_t count, i;
710
711 DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
712 work_item->id, get_trigger_name(work_item->trigger));
713
714 count = lttng_dynamic_array_get_count(&work_item->subitems);
715 for (i = 0; i < count; i++) {
716 struct action_work_subitem *item;
717
718 item = (action_work_subitem *) lttng_dynamic_array_get_element(&work_item->subitems, i);
719 ret = action_executor_generic_handler(
720 executor, work_item, item);
721 if (ret) {
722 goto end;
723 }
724 }
725 end:
726 DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
727 work_item->id, get_trigger_name(work_item->trigger));
728 return ret;
729 }
730
731 static void action_work_item_destroy(struct action_work_item *work_item)
732 {
733 lttng_trigger_put(work_item->trigger);
734 lttng_evaluation_destroy(work_item->evaluation);
735 notification_client_list_put(work_item->client_list);
736 lttng_dynamic_array_reset(&work_item->subitems);
737 free(work_item);
738 }
739
740 static void *action_executor_thread(void *_data)
741 {
742 struct action_executor *executor = (action_executor *) _data;
743
744 LTTNG_ASSERT(executor);
745
746 health_register(the_health_sessiond,
747 HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
748
749 rcu_register_thread();
750 rcu_thread_online();
751
752 DBG("Entering work execution loop");
753 pthread_mutex_lock(&executor->work.lock);
754 while (!executor->should_quit) {
755 int ret = 0;
756 struct action_work_item *work_item;
757
758 health_code_update();
759 if (executor->work.pending_count == 0) {
760 health_poll_entry();
761 DBG("No work items enqueued, entering wait");
762 pthread_cond_wait(&executor->work.cond,
763 &executor->work.lock);
764 DBG("Woke-up from wait");
765 health_poll_exit();
766 continue;
767 }
768
769 /* Pop item from front of the list with work lock held. */
770 work_item = cds_list_first_entry(&executor->work.list,
771 struct action_work_item, list_node);
772 cds_list_del(&work_item->list_node);
773 executor->work.pending_count--;
774
775 /*
776 * Work can be performed without holding the work lock,
777 * allowing new items to be queued.
778 */
779 pthread_mutex_unlock(&executor->work.lock);
780
781 /* Execute item only if a trigger is registered. */
782 lttng_trigger_lock(work_item->trigger);
783 if (!lttng_trigger_is_registered(work_item->trigger)) {
784 const char *trigger_name = NULL;
785 uid_t trigger_owner_uid;
786 enum lttng_trigger_status trigger_status;
787
788 trigger_name = get_trigger_name(work_item->trigger);
789
790 trigger_status = lttng_trigger_get_owner_uid(
791 work_item->trigger, &trigger_owner_uid);
792 LTTNG_ASSERT(trigger_status == LTTNG_TRIGGER_STATUS_OK);
793
794 DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64 ", trigger name = '%s', trigger owner uid = %d",
795 work_item->id, trigger_name,
796 (int) trigger_owner_uid);
797 ret = 0;
798 goto skip_execute;
799 }
800
801 ret = action_work_item_execute(executor, work_item);
802
803 skip_execute:
804 lttng_trigger_unlock(work_item->trigger);
805 action_work_item_destroy(work_item);
806 if (ret) {
807 /* Fatal error. */
808 break;
809 }
810
811 health_code_update();
812 pthread_mutex_lock(&executor->work.lock);
813 }
814
815 if (executor->should_quit) {
816 pthread_mutex_unlock(&executor->work.lock);
817 }
818 DBG("Left work execution loop");
819
820 health_code_update();
821
822 rcu_thread_offline();
823 rcu_unregister_thread();
824 health_unregister(the_health_sessiond);
825
826 return NULL;
827 }
828
829 static bool shutdown_action_executor_thread(void *_data)
830 {
831 struct action_executor *executor = (action_executor *) _data;
832
833 pthread_mutex_lock(&executor->work.lock);
834 executor->should_quit = true;
835 pthread_cond_signal(&executor->work.cond);
836 pthread_mutex_unlock(&executor->work.lock);
837 return true;
838 }
839
840 static void clean_up_action_executor_thread(void *_data)
841 {
842 struct action_executor *executor = (action_executor *) _data;
843
844 LTTNG_ASSERT(cds_list_empty(&executor->work.list));
845
846 pthread_mutex_destroy(&executor->work.lock);
847 pthread_cond_destroy(&executor->work.cond);
848 free(executor);
849 }
850
851 struct action_executor *action_executor_create(
852 struct notification_thread_handle *handle)
853 {
854 struct action_executor *executor = (action_executor *) zmalloc(sizeof(*executor));
855
856 if (!executor) {
857 goto end;
858 }
859
860 CDS_INIT_LIST_HEAD(&executor->work.list);
861 pthread_cond_init(&executor->work.cond, NULL);
862 pthread_mutex_init(&executor->work.lock, NULL);
863 executor->notification_thread_handle = handle;
864
865 executor->thread = lttng_thread_create(THREAD_NAME,
866 action_executor_thread, shutdown_action_executor_thread,
867 clean_up_action_executor_thread, executor);
868 end:
869 return executor;
870 }
871
872 void action_executor_destroy(struct action_executor *executor)
873 {
874 struct action_work_item *work_item, *tmp;
875
876 /* TODO Wait for work list to drain? */
877 lttng_thread_shutdown(executor->thread);
878 pthread_mutex_lock(&executor->work.lock);
879 if (executor->work.pending_count != 0) {
880 WARN("%" PRIu64
881 " trigger action%s still queued for execution and will be discarded",
882 executor->work.pending_count,
883 executor->work.pending_count == 1 ? " is" :
884 "s are");
885 }
886
887 cds_list_for_each_entry_safe (
888 work_item, tmp, &executor->work.list, list_node) {
889 WARN("Discarding action work item %" PRIu64
890 " associated to trigger `%s`",
891 work_item->id, get_trigger_name(work_item->trigger));
892 cds_list_del(&work_item->list_node);
893 action_work_item_destroy(work_item);
894 }
895 pthread_mutex_unlock(&executor->work.lock);
896 lttng_thread_put(executor->thread);
897 }
898
899 /* RCU read-lock must be held by the caller. */
900 enum action_executor_status action_executor_enqueue_trigger(
901 struct action_executor *executor,
902 struct lttng_trigger *trigger,
903 struct lttng_evaluation *evaluation,
904 const struct lttng_credentials *object_creds,
905 struct notification_client_list *client_list)
906 {
907 int ret;
908 enum action_executor_status executor_status = ACTION_EXECUTOR_STATUS_OK;
909 const uint64_t work_item_id = executor->next_work_item_id++;
910 struct action_work_item *work_item;
911 bool signal = false;
912
913 LTTNG_ASSERT(trigger);
914
915 pthread_mutex_lock(&executor->work.lock);
916 /* Check for queue overflow. */
917 if (executor->work.pending_count >= MAX_QUEUED_WORK_COUNT) {
918 /* Most likely spammy, remove if it is the case. */
919 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64,
920 get_trigger_name(trigger), work_item_id);
921 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
922 goto error_unlock;
923 }
924
925 work_item = (action_work_item *) zmalloc(sizeof(*work_item));
926 if (!work_item) {
927 PERROR("Failed to allocate action executor work item: trigger name = '%s'",
928 get_trigger_name(trigger));
929 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
930 goto error_unlock;
931 }
932
933 lttng_trigger_get(trigger);
934 if (client_list) {
935 const bool reference_acquired =
936 notification_client_list_get(client_list);
937
938 LTTNG_ASSERT(reference_acquired);
939 }
940
941 work_item->id = work_item_id;
942 work_item->trigger = trigger;
943
944 /* Ownership transferred to the work item. */
945 work_item->evaluation = evaluation;
946 evaluation = NULL;
947
948 work_item->client_list = client_list;
949 work_item->object_creds.is_set = !!object_creds;
950 if (object_creds) {
951 work_item->object_creds.value = *object_creds;
952 }
953
954 CDS_INIT_LIST_HEAD(&work_item->list_node);
955
956 /* Build the array of action work subitems for the passed trigger. */
957 lttng_dynamic_array_init(&work_item->subitems,
958 sizeof(struct action_work_subitem),
959 action_work_subitem_destructor);
960
961 ret = populate_subitem_array_from_trigger(
962 trigger, &work_item->subitems);
963 if (ret) {
964 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
965 get_trigger_name(trigger));
966 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
967 goto error_unlock;
968 }
969
970 cds_list_add_tail(&work_item->list_node, &executor->work.list);
971 executor->work.pending_count++;
972 DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64,
973 get_trigger_name(trigger), work_item_id);
974 signal = true;
975
976 error_unlock:
977 if (signal) {
978 pthread_cond_signal(&executor->work.cond);
979 }
980
981 pthread_mutex_unlock(&executor->work.lock);
982 lttng_evaluation_destroy(evaluation);
983 return executor_status;
984 }
985
986 static int add_action_to_subitem_array(struct lttng_action *action,
987 struct lttng_dynamic_array *subitems)
988 {
989 int ret = 0;
990 enum lttng_action_type type = lttng_action_get_type(action);
991 const char *session_name = NULL;
992 enum lttng_action_status status;
993 struct action_work_subitem subitem = {
994 .action = NULL,
995 .context = {
996 .session_id = LTTNG_OPTIONAL_INIT_UNSET,
997 },
998 };
999
1000 LTTNG_ASSERT(action);
1001 LTTNG_ASSERT(subitems);
1002
1003 if (type == LTTNG_ACTION_TYPE_LIST) {
1004 unsigned int count, i;
1005
1006 status = lttng_action_list_get_count(action, &count);
1007 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1008
1009 for (i = 0; i < count; i++) {
1010 struct lttng_action *inner_action = NULL;
1011
1012 inner_action = lttng_action_list_borrow_mutable_at_index(
1013 action, i);
1014 LTTNG_ASSERT(inner_action);
1015 ret = add_action_to_subitem_array(
1016 inner_action, subitems);
1017 if (ret) {
1018 goto end;
1019 }
1020 }
1021
1022 /*
1023 * Go directly to the end since there is no need to add the
1024 * list action by itself to the subitems array.
1025 */
1026 goto end;
1027 }
1028
1029 /* Gather execution context. */
1030 switch (type) {
1031 case LTTNG_ACTION_TYPE_NOTIFY:
1032 break;
1033 case LTTNG_ACTION_TYPE_START_SESSION:
1034 status = lttng_action_start_session_get_session_name(
1035 action, &session_name);
1036 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1037 break;
1038 case LTTNG_ACTION_TYPE_STOP_SESSION:
1039 status = lttng_action_stop_session_get_session_name(
1040 action, &session_name);
1041 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1042 break;
1043 case LTTNG_ACTION_TYPE_ROTATE_SESSION:
1044 status = lttng_action_rotate_session_get_session_name(
1045 action, &session_name);
1046 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1047 break;
1048 case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION:
1049 status = lttng_action_snapshot_session_get_session_name(
1050 action, &session_name);
1051 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1052 break;
1053 case LTTNG_ACTION_TYPE_LIST:
1054 case LTTNG_ACTION_TYPE_UNKNOWN:
1055 /* Fallthrough */
1056 default:
1057 abort();
1058 break;
1059 }
1060
1061 /*
1062 * Fetch the session execution context info as needed.
1063 * Note that we could decide to not add an action for which we know the
1064 * execution will not happen (i.e no session exists for that name). For
1065 * now we leave the decision to skip to the action executor for sake of
1066 * simplicity and consistency.
1067 */
1068 if (session_name != NULL) {
1069 uint64_t session_id;
1070
1071 /*
1072 * Instantaneous sampling of the session id if present.
1073 *
1074 * This method is preferred over `sessiond_find_by_name` then
1075 * fetching the session'd id since `sessiond_find_by_name`
1076 * requires the session list lock to be taken.
1077 *
1078 * Taking the session list lock can lead to a deadlock
1079 * between the action executor and the notification thread
1080 * (caller of add_action_to_subitem_array). It is okay if the
1081 * session state changes between the enqueuing time and the
1082 * execution time. The execution context is validated at
1083 * execution time.
1084 */
1085 if (sample_session_id_by_name(session_name, &session_id)) {
1086 LTTNG_OPTIONAL_SET(&subitem.context.session_id,
1087 session_id);
1088 }
1089 }
1090
1091 /* Get a reference to the action. */
1092 lttng_action_get(action);
1093 subitem.action = action;
1094
1095 ret = lttng_dynamic_array_add_element(subitems, &subitem);
1096 if (ret) {
1097 ERR("Failed to add work subitem to the subitem array");
1098 lttng_action_put(action);
1099 ret = -1;
1100 goto end;
1101 }
1102
1103 end:
1104 return ret;
1105 }
1106
1107 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
1108 struct lttng_dynamic_array *subitems)
1109 {
1110 struct lttng_action *action;
1111
1112 action = lttng_trigger_get_action(trigger);
1113 LTTNG_ASSERT(action);
1114
1115 return add_action_to_subitem_array(action, subitems);
1116 }
This page took 0.087017 seconds and 4 git commands to generate.