Fix: action executor: ref count imbalance for session object
[lttng-tools.git] / src / bin / lttng-sessiond / action-executor.cpp
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 */
be65f802 80 struct lttng_dynamic_array subitems;
72365501
JR
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 *);
7c2fae7c 138static int action_executor_list_handler(struct action_executor *executor,
f2b3ef9f 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[] = {
7966af57
SM
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,
f2b3ef9f
JG
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{
7966af57 163 struct action_work_subitem *subitem = (action_work_subitem *) element;
72365501
JR
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
a0377dfe 172 LTTNG_ASSERT(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
0e43bcbf 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);
0efb2ad7
JG
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 }
34f87583
JR
219
220 return trigger_name;
221}
222
f2b3ef9f
JG
223static 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;
7966af57 229 struct action_executor *executor = (action_executor *) user_data;
f2b3ef9f
JG
230 bool update_communication = true;
231
f2b3ef9f
JG
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);
f2b3ef9f
JG
245 break;
246 default:
247 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64,
248 client->id);
f2b3ef9f
JG
249 ret = -1;
250 goto end;
251 }
252
253 if (!update_communication) {
254 goto end;
255 }
256
6c24d3fd 257 /* Safe to read client's id without locking as it is immutable. */
f2b3ef9f
JG
258 ret = notification_thread_client_communication_update(
259 executor->notification_thread_handle, client->id,
260 status);
261end:
262 return ret;
263}
264
265static int action_executor_notify_handler(struct action_executor *executor,
266 const struct action_work_item *work_item,
72365501 267 struct action_work_subitem *item)
f2b3ef9f
JG
268{
269 return notification_client_list_send_evaluation(work_item->client_list,
52d55cf9 270 work_item->trigger,
f2b3ef9f 271 work_item->evaluation,
c203f058
JR
272 work_item->object_creds.is_set ?
273 &(work_item->object_creds.value) :
274 NULL,
64eafdf6 275 client_handle_transmission_status, executor);
f2b3ef9f
JG
276}
277
2d57482c
JR
278static int action_executor_start_session_handler(
279 struct action_executor *executor,
f2b3ef9f 280 const struct action_work_item *work_item,
72365501 281 struct action_work_subitem *item)
f2b3ef9f
JG
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;
72365501 288 struct lttng_action *action = item->action;
f2b3ef9f
JG
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
72365501
JR
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
f2b3ef9f
JG
312 session_lock_list();
313 session = session_find_by_name(session_name);
314 if (!session) {
34f87583 315 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 316 session_name, get_action_name(action),
34f87583 317 get_trigger_name(work_item->trigger));
f2b3ef9f
JG
318 goto error_unlock_list;
319 }
320
72365501
JR
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;
635e0160 334 goto error_put_session;
72365501
JR
335 }
336
f2b3ef9f
JG
337 session_lock(session);
338 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
635e0160 339 goto error_unlock_session;
f2b3ef9f
JG
340 }
341
7966af57 342 cmd_ret = (lttng_error_code) cmd_start_trace(session);
f2b3ef9f
JG
343 switch (cmd_ret) {
344 case LTTNG_OK:
34f87583
JR
345 DBG("Successfully started session `%s` on behalf of trigger `%s`",
346 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
347 break;
348 case LTTNG_ERR_TRACE_ALREADY_STARTED:
34f87583
JR
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));
f2b3ef9f
JG
351 break;
352 default:
34f87583
JR
353 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
354 session_name, get_trigger_name(work_item->trigger),
f2b3ef9f 355 lttng_strerror(-cmd_ret));
2d57482c 356 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
357 break;
358 }
359
635e0160 360error_unlock_session:
f2b3ef9f 361 session_unlock(session);
635e0160 362error_put_session:
f2b3ef9f
JG
363 session_put(session);
364error_unlock_list:
365 session_unlock_list();
366end:
367 return ret;
368}
369
2d57482c
JR
370static int action_executor_stop_session_handler(
371 struct action_executor *executor,
f2b3ef9f 372 const struct action_work_item *work_item,
72365501 373 struct action_work_subitem *item)
f2b3ef9f
JG
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;
72365501 380 struct lttng_action *action = item->action;
f2b3ef9f
JG
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
72365501
JR
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
f2b3ef9f
JG
404 session_lock_list();
405 session = session_find_by_name(session_name);
406 if (!session) {
34f87583 407 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 408 session_name, get_action_name(action),
34f87583 409 get_trigger_name(work_item->trigger));
2d57482c 410 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
411 goto error_unlock_list;
412 }
413
72365501
JR
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;
635e0160 427 goto error_put_session;
72365501
JR
428 }
429
f2b3ef9f
JG
430 session_lock(session);
431 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
635e0160 432 goto error_unlock_session;
f2b3ef9f
JG
433 }
434
7966af57 435 cmd_ret = (lttng_error_code) cmd_stop_trace(session);
f2b3ef9f
JG
436 switch (cmd_ret) {
437 case LTTNG_OK:
34f87583
JR
438 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
439 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
440 break;
441 case LTTNG_ERR_TRACE_ALREADY_STOPPED:
34f87583
JR
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));
f2b3ef9f
JG
444 break;
445 default:
34f87583
JR
446 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
447 session_name, get_trigger_name(work_item->trigger),
f2b3ef9f 448 lttng_strerror(-cmd_ret));
2d57482c 449 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
450 break;
451 }
452
635e0160 453error_unlock_session:
f2b3ef9f 454 session_unlock(session);
635e0160 455error_put_session:
f2b3ef9f
JG
456 session_put(session);
457error_unlock_list:
458 session_unlock_list();
459end:
460 return ret;
461}
462
2d57482c
JR
463static int action_executor_rotate_session_handler(
464 struct action_executor *executor,
f2b3ef9f 465 const struct action_work_item *work_item,
72365501 466 struct action_work_subitem *item)
f2b3ef9f
JG
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;
72365501 473 struct lttng_action *action = item->action;
f2b3ef9f
JG
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
72365501
JR
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
f2b3ef9f
JG
497 session_lock_list();
498 session = session_find_by_name(session_name);
499 if (!session) {
34f87583 500 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 501 session_name, get_action_name(action),
34f87583 502 get_trigger_name(work_item->trigger));
2d57482c 503 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
504 goto error_unlock_list;
505 }
506
72365501
JR
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;
635e0160 520 goto error_put_session;
72365501
JR
521 }
522
f2b3ef9f
JG
523 session_lock(session);
524 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
635e0160 525 goto error_unlock_session;
f2b3ef9f
JG
526 }
527
7966af57 528 cmd_ret = (lttng_error_code) cmd_rotate_session(session, NULL, false,
f2b3ef9f
JG
529 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
530 switch (cmd_ret) {
531 case LTTNG_OK:
34f87583
JR
532 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
533 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
534 break;
535 case LTTNG_ERR_ROTATION_PENDING:
34f87583
JR
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));
2d57482c 538 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
539 break;
540 case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
541 case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
34f87583
JR
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));
f2b3ef9f
JG
544 break;
545 default:
34f87583
JR
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),
f2b3ef9f 548 lttng_strerror(-cmd_ret));
2d57482c 549 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
550 break;
551 }
552
635e0160 553error_unlock_session:
f2b3ef9f 554 session_unlock(session);
635e0160 555error_put_session:
f2b3ef9f
JG
556 session_put(session);
557error_unlock_list:
558 session_unlock_list();
559end:
560 return ret;
561}
562
2d57482c
JR
563static int action_executor_snapshot_session_handler(
564 struct action_executor *executor,
f2b3ef9f 565 const struct action_work_item *work_item,
72365501 566 struct action_work_subitem *item)
f2b3ef9f
JG
567{
568 int ret = 0;
569 const char *session_name;
570 enum lttng_action_status action_status;
571 struct ltt_session *session;
7966af57 572 lttng_snapshot_output default_snapshot_output;
f2b3ef9f
JG
573 const struct lttng_snapshot_output *snapshot_output =
574 &default_snapshot_output;
575 enum lttng_error_code cmd_ret;
72365501
JR
576 struct lttng_action *action = item->action;
577
7966af57
SM
578 default_snapshot_output.max_size = UINT64_MAX;
579
72365501
JR
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) {
42ef691e
JG
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),
72365501
JR
587 get_trigger_name(work_item->trigger));
588 lttng_action_increase_execution_failure_count(action);
589 ret = 0;
590 goto end;
591 }
f2b3ef9f
JG
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) {
ca46af4e 615 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
f2b3ef9f 616 session_name, get_action_name(action),
ca46af4e 617 get_trigger_name(work_item->trigger));
2d57482c 618 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
619 goto error_unlock_list;
620 }
621
72365501
JR
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;
635e0160 635 goto error_put_session;
72365501 636 }
f2b3ef9f
JG
637
638 session_lock(session);
639 if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
635e0160 640 goto error_unlock_session;
f2b3ef9f
JG
641 }
642
7966af57 643 cmd_ret = (lttng_error_code) cmd_snapshot_record(session, snapshot_output, 0);
f2b3ef9f
JG
644 switch (cmd_ret) {
645 case LTTNG_OK:
34f87583
JR
646 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
647 session_name, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
648 break;
649 default:
34f87583
JR
650 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
651 session_name, get_trigger_name(work_item->trigger),
f2b3ef9f 652 lttng_strerror(-cmd_ret));
2d57482c 653 lttng_action_increase_execution_failure_count(action);
f2b3ef9f
JG
654 break;
655 }
656
635e0160 657error_unlock_session:
f2b3ef9f 658 session_unlock(session);
635e0160 659error_put_session:
f2b3ef9f
JG
660 session_put(session);
661error_unlock_list:
662 session_unlock_list();
663end:
664 return ret;
665}
666
7c2fae7c 667static int action_executor_list_handler(struct action_executor *executor,
f2b3ef9f 668 const struct action_work_item *work_item,
72365501 669 struct action_work_subitem *item)
f2b3ef9f 670{
7c2fae7c 671 ERR("Execution of a list action by the action executor should never occur");
72365501 672 abort();
f2b3ef9f
JG
673}
674
675static int action_executor_generic_handler(struct action_executor *executor,
676 const struct action_work_item *work_item,
72365501 677 struct action_work_subitem *item)
f2b3ef9f 678{
2d57482c 679 int ret;
72365501 680 struct lttng_action *action = item->action;
0e43bcbf
JG
681 const enum lttng_action_type action_type = lttng_action_get_type(action);
682
a0377dfe 683 LTTNG_ASSERT(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
0e43bcbf 684
2d57482c
JR
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);
2516f2d8 696 DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64,
f2b3ef9f 697 get_action_name(action),
34f87583 698 get_trigger_name(work_item->trigger),
f2b3ef9f 699 work_item->id);
72365501 700 ret = action_executors[action_type](executor, work_item, item);
2d57482c
JR
701end:
702 return ret;
f2b3ef9f
JG
703}
704
705static int action_work_item_execute(struct action_executor *executor,
706 struct action_work_item *work_item)
707{
708 int ret;
72365501 709 size_t count, i;
f2b3ef9f 710
34f87583
JR
711 DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
712 work_item->id, get_trigger_name(work_item->trigger));
72365501 713
be65f802 714 count = lttng_dynamic_array_get_count(&work_item->subitems);
72365501
JR
715 for (i = 0; i < count; i++) {
716 struct action_work_subitem *item;
717
7966af57 718 item = (action_work_subitem *) lttng_dynamic_array_get_element(&work_item->subitems, i);
72365501
JR
719 ret = action_executor_generic_handler(
720 executor, work_item, item);
721 if (ret) {
722 goto end;
723 }
724 }
725end:
34f87583
JR
726 DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
727 work_item->id, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
728 return ret;
729}
730
731static 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);
be65f802 736 lttng_dynamic_array_reset(&work_item->subitems);
f2b3ef9f
JG
737 free(work_item);
738}
739
740static void *action_executor_thread(void *_data)
741{
7966af57 742 struct action_executor *executor = (action_executor *) _data;
f2b3ef9f 743
a0377dfe 744 LTTNG_ASSERT(executor);
f2b3ef9f 745
412d7227
SM
746 health_register(the_health_sessiond,
747 HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
f2b3ef9f
JG
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) {
da247508 755 int ret = 0;
f2b3ef9f
JG
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
0db0f8e0 769 /* Pop item from front of the list with work lock held. */
f2b3ef9f
JG
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);
d2a28b27
JR
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
0efb2ad7 788 trigger_name = get_trigger_name(work_item->trigger);
d2a28b27
JR
789
790 trigger_status = lttng_trigger_get_owner_uid(
791 work_item->trigger, &trigger_owner_uid);
a0377dfe 792 LTTNG_ASSERT(trigger_status == LTTNG_TRIGGER_STATUS_OK);
d2a28b27
JR
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
f2b3ef9f 801 ret = action_work_item_execute(executor, work_item);
d2a28b27
JR
802
803 skip_execute:
804 lttng_trigger_unlock(work_item->trigger);
f2b3ef9f
JG
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
f5f5c54d
JG
815 if (executor->should_quit) {
816 pthread_mutex_unlock(&executor->work.lock);
817 }
f2b3ef9f
JG
818 DBG("Left work execution loop");
819
820 health_code_update();
821
822 rcu_thread_offline();
823 rcu_unregister_thread();
412d7227 824 health_unregister(the_health_sessiond);
f2b3ef9f
JG
825
826 return NULL;
827}
828
829static bool shutdown_action_executor_thread(void *_data)
830{
7966af57 831 struct action_executor *executor = (action_executor *) _data;
f2b3ef9f 832
8db3acaf 833 pthread_mutex_lock(&executor->work.lock);
f2b3ef9f
JG
834 executor->should_quit = true;
835 pthread_cond_signal(&executor->work.cond);
8db3acaf 836 pthread_mutex_unlock(&executor->work.lock);
f2b3ef9f
JG
837 return true;
838}
839
840static void clean_up_action_executor_thread(void *_data)
841{
7966af57 842 struct action_executor *executor = (action_executor *) _data;
f2b3ef9f 843
a0377dfe 844 LTTNG_ASSERT(cds_list_empty(&executor->work.list));
f2b3ef9f
JG
845
846 pthread_mutex_destroy(&executor->work.lock);
847 pthread_cond_destroy(&executor->work.cond);
848 free(executor);
849}
850
851struct action_executor *action_executor_create(
852 struct notification_thread_handle *handle)
853{
7966af57 854 struct action_executor *executor = (action_executor *) zmalloc(sizeof(*executor));
f2b3ef9f
JG
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);
868end:
869 return executor;
870}
871
872void 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
34f87583
JR
890 " associated to trigger `%s`",
891 work_item->id, get_trigger_name(work_item->trigger));
f2b3ef9f
JG
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. */
72365501 900enum action_executor_status action_executor_enqueue_trigger(
f2b3ef9f
JG
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{
72365501 907 int ret;
f2b3ef9f
JG
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;
72365501 912
a0377dfe 913 LTTNG_ASSERT(trigger);
72365501 914
f2b3ef9f
JG
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. */
72365501
JR
919 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64,
920 get_trigger_name(trigger), work_item_id);
f2b3ef9f
JG
921 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
922 goto error_unlock;
923 }
924
7966af57 925 work_item = (action_work_item *) zmalloc(sizeof(*work_item));
f2b3ef9f 926 if (!work_item) {
72365501 927 PERROR("Failed to allocate action executor work item: trigger name = '%s'",
34f87583 928 get_trigger_name(trigger));
f2b3ef9f
JG
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
a0377dfe 938 LTTNG_ASSERT(reference_acquired);
f2b3ef9f
JG
939 }
940
7966af57
SM
941 work_item->id = work_item_id;
942 work_item->trigger = trigger;
f2b3ef9f 943
7966af57
SM
944 /* Ownership transferred to the work item. */
945 work_item->evaluation = evaluation;
f2b3ef9f 946 evaluation = NULL;
be65f802 947
7966af57
SM
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
be65f802
JG
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
f2b3ef9f
JG
970 cds_list_add_tail(&work_item->list_node, &executor->work.list);
971 executor->work.pending_count++;
72365501 972 DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64,
34f87583 973 get_trigger_name(trigger), work_item_id);
f2b3ef9f
JG
974 signal = true;
975
976error_unlock:
f2b3ef9f
JG
977 if (signal) {
978 pthread_cond_signal(&executor->work.cond);
979 }
980
be65f802 981 pthread_mutex_unlock(&executor->work.lock);
f2b3ef9f
JG
982 lttng_evaluation_destroy(evaluation);
983 return executor_status;
984}
72365501
JR
985
986static int add_action_to_subitem_array(struct lttng_action *action,
987 struct lttng_dynamic_array *subitems)
988{
da247508 989 int ret = 0;
72365501
JR
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
a0377dfe
FD
1000 LTTNG_ASSERT(action);
1001 LTTNG_ASSERT(subitems);
72365501 1002
7c2fae7c 1003 if (type == LTTNG_ACTION_TYPE_LIST) {
72365501
JR
1004 unsigned int count, i;
1005
702f26c8 1006 status = lttng_action_list_get_count(action, &count);
a0377dfe 1007 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
72365501
JR
1008
1009 for (i = 0; i < count; i++) {
1010 struct lttng_action *inner_action = NULL;
1011
702f26c8 1012 inner_action = lttng_action_list_borrow_mutable_at_index(
72365501 1013 action, i);
a0377dfe 1014 LTTNG_ASSERT(inner_action);
72365501
JR
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
7c2fae7c 1024 * list action by itself to the subitems array.
72365501
JR
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);
a0377dfe 1036 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
72365501
JR
1037 break;
1038 case LTTNG_ACTION_TYPE_STOP_SESSION:
1039 status = lttng_action_stop_session_get_session_name(
1040 action, &session_name);
a0377dfe 1041 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
72365501
JR
1042 break;
1043 case LTTNG_ACTION_TYPE_ROTATE_SESSION:
1044 status = lttng_action_rotate_session_get_session_name(
1045 action, &session_name);
a0377dfe 1046 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
72365501
JR
1047 break;
1048 case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION:
1049 status = lttng_action_snapshot_session_get_session_name(
1050 action, &session_name);
a0377dfe 1051 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
72365501 1052 break;
7c2fae7c 1053 case LTTNG_ACTION_TYPE_LIST:
72365501
JR
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) {
e1bbf989 1069 uint64_t session_id;
72365501 1070
e1bbf989
JR
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)) {
72365501 1086 LTTNG_OPTIONAL_SET(&subitem.context.session_id,
e1bbf989 1087 session_id);
72365501 1088 }
72365501
JR
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
1103end:
1104 return ret;
1105}
1106
1107static 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);
a0377dfe 1113 LTTNG_ASSERT(action);
72365501
JR
1114
1115 return add_action_to_subitem_array(action, subitems);
1116}
This page took 0.084749 seconds and 4 git commands to generate.