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