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