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