src/bin/lttng-sessiond/action-executor.c

   1 /*
   2  * Copyright (C) 2020 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * SPDX-License-Identifier: GPL-2.0-only
   5  *
   6  */
   7
   8 #include "action-executor.h"
   9 #include "cmd.h"
  10 #include "health-sessiond.h"
  11 #include "lttng-sessiond.h"
  12 #include "notification-thread-internal.h"
  13 #include "session.h"
  14 #include "thread.h"
  15 #include <common/dynamic-array.h>
  16 #include <common/macros.h>
  17 #include <common/optional.h>
  18 #include <lttng/action/action-internal.h>
  19 #include <lttng/action/list-internal.h>
  20 #include <lttng/action/list.h>
  21 #include <lttng/action/notify-internal.h>
  22 #include <lttng/action/notify.h>
  23 #include <lttng/action/rotate-session.h>
  24 #include <lttng/action/snapshot-session.h>
  25 #include <lttng/action/start-session.h>
  26 #include <lttng/action/stop-session.h>
  27 #include <lttng/condition/evaluation.h>
  28 #include <lttng/condition/event-rule-matches-internal.h>
  29 #include <lttng/lttng-error.h>
  30 #include <lttng/trigger/trigger-internal.h>
  31 #include <pthread.h>
  32 #include <stdbool.h>
  33 #include <stddef.h>
  34 #include <urcu/list.h>
  35
  36 #define THREAD_NAME "Action Executor"
  37 #define MAX_QUEUED_WORK_COUNT 8192
  38
  39 /*
  40  * A work item is composed of a dynamic array of sub-items which
  41  * represent a flattened, and augmented, version of a trigger's actions.
  42  *
  43  * We cannot rely solely on the trigger's actions since each action can have an
  44  * execution context we need to comply with.
  45  *
  46  * The notion of execution context is required since for some actions the
  47  * associated object are referenced by name and not by id. This can lead to
  48  * a number of ambiguities when executing an action work item.
  49  *
  50  * For example, let's take a simple trigger such as:
  51  *   - condition: ust event a
  52  *   - action: start session S
  53  *
  54  * At time T, session S exists.
  55  * At T + 1, the event A is hit.
  56  * At T + 2, the tracer event notification is received and the work item is
  57  * queued. Here session S have an id of 1.
  58  * At T + 3, the session S is destroyed and a new session S is created, with a
  59  * resulting id of 200.
  60  * At T +4, the work item is popped from the queue and begin execution and will
  61  * start session S with an id of 200 instead of the session S id 1 that was
  62  * present at the queuing phase.
  63  *
  64  * The context to be respected is the one when the work item is queued. If the
  65  * execution context is not the same at the moment of execution, we skip the
  66  * execution of that sub-item.
  67  *
  68  * It is the same policy in regards to the validity of the associated
  69  * trigger object at the moment of execution, if the trigger is found to be
  70  * unregistered, the execution is skipped.
  71  */
  72
  73 struct action_work_item {
  74         uint64_t id;
  75
  76         /*
  77          * The actions to be executed with their respective execution context.
  78          * See struct `action_work_subitem`.
  79          */
  80         struct lttng_dynamic_array subitems;
  81
  82         /* Execution context data */
  83         struct lttng_trigger *trigger;
  84         struct lttng_evaluation *evaluation;
  85         struct notification_client_list *client_list;
  86         LTTNG_OPTIONAL(struct lttng_credentials) object_creds;
  87         struct cds_list_head list_node;
  88 };
  89
  90 struct action_work_subitem {
  91         struct lttng_action *action;
  92         struct {
  93                 /* Used by actions targeting a session. */
  94                 LTTNG_OPTIONAL(uint64_t) session_id;
  95         } context;
  96 };
  97
  98 struct action_executor {
  99         struct lttng_thread *thread;
 100         struct notification_thread_handle *notification_thread_handle;
 101         struct {
 102                 uint64_t pending_count;
 103                 struct cds_list_head list;
 104                 pthread_cond_t cond;
 105                 pthread_mutex_t lock;
 106         } work;
 107         bool should_quit;
 108         uint64_t next_work_item_id;
 109 };
 110
 111 /*
 112  * Only return non-zero on a fatal error that should shut down the action
 113  * executor.
 114  */
 115 typedef int (*action_executor_handler)(struct action_executor *executor,
 116                 const struct action_work_item *,
 117                 struct action_work_subitem *item);
 118
 119 static int action_executor_notify_handler(struct action_executor *executor,
 120                 const struct action_work_item *,
 121                 struct action_work_subitem *);
 122 static int action_executor_start_session_handler(
 123                 struct action_executor *executor,
 124                 const struct action_work_item *,
 125                 struct action_work_subitem *);
 126 static int action_executor_stop_session_handler(
 127                 struct action_executor *executor,
 128                 const struct action_work_item *,
 129                 struct action_work_subitem *);
 130 static int action_executor_rotate_session_handler(
 131                 struct action_executor *executor,
 132                 const struct action_work_item *,
 133                 struct action_work_subitem *);
 134 static int action_executor_snapshot_session_handler(
 135                 struct action_executor *executor,
 136                 const struct action_work_item *,
 137                 struct action_work_subitem *);
 138 static int action_executor_list_handler(struct action_executor *executor,
 139                 const struct action_work_item *,
 140                 struct action_work_subitem *);
 141 static int action_executor_generic_handler(struct action_executor *executor,
 142                 const struct action_work_item *,
 143                 struct action_work_subitem *);
 144
 145 static const action_executor_handler action_executors[] = {
 146         [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_LIST] = action_executor_list_handler,
 152 };
 153
 154 /* Forward declaration */
 155 static int add_action_to_subitem_array(struct lttng_action *action,
 156                 struct lttng_dynamic_array *subitems);
 157
 158 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
 159                 struct lttng_dynamic_array *subitems);
 160
 161 static void action_work_subitem_destructor(void *element)
 162 {
 163         struct action_work_subitem *subitem = element;
 164
 165         lttng_action_put(subitem->action);
 166 }
 167
 168 static const char *get_action_name(const struct lttng_action *action)
 169 {
 170         const enum lttng_action_type action_type = lttng_action_get_type(action);
 171
 172         assert(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
 173
 174         return lttng_action_type_string(action_type);
 175 }
 176
 177 /* Check if this trigger allowed to interect with a given session. */
 178 static bool is_trigger_allowed_for_session(const struct lttng_trigger *trigger,
 179                 struct ltt_session *session)
 180 {
 181         bool is_allowed = false;
 182         const struct lttng_credentials session_creds = {
 183                 .uid = LTTNG_OPTIONAL_INIT_VALUE(session->uid),
 184                 .gid = LTTNG_OPTIONAL_INIT_VALUE(session->gid),
 185         };
 186         /* Can never be NULL. */
 187         const struct lttng_credentials *trigger_creds =
 188                         lttng_trigger_get_credentials(trigger);
 189
 190         is_allowed = (lttng_credentials_is_equal_uid(trigger_creds, &session_creds)) ||
 191                         (lttng_credentials_get_uid(trigger_creds) == 0);
 192         if (!is_allowed) {
 193                 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
 194                                 session->name,
 195                                 (long int) session->uid,
 196                                 (long int) session->gid,
 197                                 (long int) lttng_credentials_get_uid(trigger_creds));
 198         }
 199
 200         return is_allowed;
 201 }
 202
 203 static const char *get_trigger_name(const struct lttng_trigger *trigger)
 204 {
 205         const char *trigger_name;
 206         enum lttng_trigger_status trigger_status;
 207
 208         trigger_status = lttng_trigger_get_name(trigger, &trigger_name);
 209         switch (trigger_status) {
 210         case LTTNG_TRIGGER_STATUS_OK:
 211                 break;
 212         case LTTNG_TRIGGER_STATUS_UNSET:
 213                 trigger_name = "(anonymous)";
 214                 break;
 215         default:
 216                 trigger_name = "(failed to get name)";
 217                 break;
 218         }
 219
 220         return trigger_name;
 221 }
 222
 223 static int client_handle_transmission_status(
 224                 struct notification_client *client,
 225                 enum client_transmission_status status,
 226                 void *user_data)
 227 {
 228         int ret = 0;
 229         struct action_executor *executor = 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);
 270 end:
 271         return ret;
 272 }
 273
 274 static int action_executor_notify_handler(struct action_executor *executor,
 275                 const struct action_work_item *work_item,
 276                 struct action_work_subitem *item)
 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
 287 static int action_executor_start_session_handler(
 288                 struct action_executor *executor,
 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                 ret = 0;
 318                 goto end;
 319         }
 320
 321         session_lock_list();
 322         session = session_find_by_name(session_name);
 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                 goto error_unlock_list;
 328         }
 329
 330         /*
 331          * Check if the session id is the same as when the work item was
 332          * enqueued.
 333          */
 334         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 335                 DBG("Session id for session `%s` (id: %" PRIu64
 336                                 " is not the same that was sampled (id: %" PRIu64
 337                                 " at the moment the work item was enqueued for `%s` action of trigger `%s`",
 338                                 session_name, session->id,
 339                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 340                                 get_action_name(action),
 341                                 get_trigger_name(work_item->trigger));
 342                 ret = 0;
 343                 goto error_put_session;
 344         }
 345
 346         session_lock(session);
 347         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 348                 goto error_unlock_session;
 349         }
 350
 351         cmd_ret = cmd_start_trace(session);
 352         switch (cmd_ret) {
 353         case LTTNG_OK:
 354                 DBG("Successfully started session `%s` on behalf of trigger `%s`",
 355                                 session_name, get_trigger_name(work_item->trigger));
 356                 break;
 357         case LTTNG_ERR_TRACE_ALREADY_STARTED:
 358                 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
 359                                 session_name, get_trigger_name(work_item->trigger));
 360                 break;
 361         default:
 362                 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
 363                                 session_name, get_trigger_name(work_item->trigger),
 364                                 lttng_strerror(-cmd_ret));
 365                 lttng_action_increase_execution_failure_count(action);
 366                 break;
 367         }
 368
 369 error_unlock_session:
 370         session_unlock(session);
 371 error_put_session:
 372         session_put(session);
 373 error_unlock_list:
 374         session_unlock_list();
 375 end:
 376         return ret;
 377 }
 378
 379 static int action_executor_stop_session_handler(
 380                 struct action_executor *executor,
 381                 const struct action_work_item *work_item,
 382                 struct action_work_subitem *item)
 383 {
 384         int ret = 0;
 385         const char *session_name;
 386         enum lttng_action_status action_status;
 387         struct ltt_session *session;
 388         enum lttng_error_code cmd_ret;
 389         struct lttng_action *action = item->action;
 390
 391         action_status = lttng_action_stop_session_get_session_name(
 392                         action, &session_name);
 393         if (action_status != LTTNG_ACTION_STATUS_OK) {
 394                 ERR("Failed to get session name from `%s` action",
 395                                 get_action_name(action));
 396                 ret = -1;
 397                 goto end;
 398         }
 399
 400         /*
 401          * Validate if, at the moment the action was queued, the target session
 402          * existed. If not, skip the action altogether.
 403          */
 404         if (!item->context.session_id.is_set) {
 405                 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
 406                                 session_name, get_action_name(action),
 407                                 get_trigger_name(work_item->trigger));
 408                 lttng_action_increase_execution_failure_count(action);
 409                 ret = 0;
 410                 goto end;
 411         }
 412
 413         session_lock_list();
 414         session = session_find_by_name(session_name);
 415         if (!session) {
 416                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 417                                 session_name, get_action_name(action),
 418                                 get_trigger_name(work_item->trigger));
 419                 lttng_action_increase_execution_failure_count(action);
 420                 goto error_unlock_list;
 421         }
 422
 423         /*
 424          * Check if the session id is the same as when the work item was
 425          * enqueued
 426          */
 427         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 428                 DBG("Session id for session `%s` (id: %" PRIu64
 429                                 " is not the same that was sampled (id: %" PRIu64
 430                                 " at the moment the work item was enqueued for `%s` action of trigger `%s`",
 431                                 session_name, session->id,
 432                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 433                                 get_action_name(action),
 434                                 get_trigger_name(work_item->trigger));
 435                 ret = 0;
 436                 goto error_put_session;
 437         }
 438
 439         session_lock(session);
 440         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 441                 goto error_unlock_session;
 442         }
 443
 444         cmd_ret = cmd_stop_trace(session);
 445         switch (cmd_ret) {
 446         case LTTNG_OK:
 447                 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
 448                                 session_name, get_trigger_name(work_item->trigger));
 449                 break;
 450         case LTTNG_ERR_TRACE_ALREADY_STOPPED:
 451                 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
 452                                 session_name, get_trigger_name(work_item->trigger));
 453                 break;
 454         default:
 455                 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
 456                                 session_name, get_trigger_name(work_item->trigger),
 457                                 lttng_strerror(-cmd_ret));
 458                 lttng_action_increase_execution_failure_count(action);
 459                 break;
 460         }
 461
 462 error_unlock_session:
 463         session_unlock(session);
 464 error_put_session:
 465         session_put(session);
 466 error_unlock_list:
 467         session_unlock_list();
 468 end:
 469         return ret;
 470 }
 471
 472 static int action_executor_rotate_session_handler(
 473                 struct action_executor *executor,
 474                 const struct action_work_item *work_item,
 475                 struct action_work_subitem *item)
 476 {
 477         int ret = 0;
 478         const char *session_name;
 479         enum lttng_action_status action_status;
 480         struct ltt_session *session;
 481         enum lttng_error_code cmd_ret;
 482         struct lttng_action *action = item->action;
 483
 484         action_status = lttng_action_rotate_session_get_session_name(
 485                         action, &session_name);
 486         if (action_status != LTTNG_ACTION_STATUS_OK) {
 487                 ERR("Failed to get session name from `%s` action",
 488                                 get_action_name(action));
 489                 ret = -1;
 490                 goto end;
 491         }
 492
 493         /*
 494          * Validate if, at the moment the action was queued, the target session
 495          * existed. If not, skip the action altogether.
 496          */
 497         if (!item->context.session_id.is_set) {
 498                 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
 499                                 session_name, get_action_name(action),
 500                                 get_trigger_name(work_item->trigger));
 501                 lttng_action_increase_execution_failure_count(action);
 502                 ret = 0;
 503                 goto end;
 504         }
 505
 506         session_lock_list();
 507         session = session_find_by_name(session_name);
 508         if (!session) {
 509                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 510                                 session_name, get_action_name(action),
 511                                 get_trigger_name(work_item->trigger));
 512                 lttng_action_increase_execution_failure_count(action);
 513                 goto error_unlock_list;
 514         }
 515
 516         /*
 517          * Check if the session id is the same as when the work item was
 518          * enqueued.
 519          */
 520         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 521                 DBG("Session id for session `%s` (id: %" PRIu64
 522                     " is not the same that was sampled (id: %" PRIu64
 523                     " at the moment the work item was enqueued for `%s` action of trigger `%s`",
 524                                 session_name, session->id,
 525                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 526                                 get_action_name(action),
 527                                 get_trigger_name(work_item->trigger));
 528                 ret = 0;
 529                 goto error_put_session;
 530         }
 531
 532         session_lock(session);
 533         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 534                 goto error_unlock_session;
 535         }
 536
 537         cmd_ret = cmd_rotate_session(session, NULL, false,
 538                         LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 539         switch (cmd_ret) {
 540         case LTTNG_OK:
 541                 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
 542                                 session_name, get_trigger_name(work_item->trigger));
 543                 break;
 544         case LTTNG_ERR_ROTATION_PENDING:
 545                 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
 546                                 session_name, get_trigger_name(work_item->trigger));
 547                 lttng_action_increase_execution_failure_count(action);
 548                 break;
 549         case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
 550         case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
 551                 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",
 552                                 session_name, get_trigger_name(work_item->trigger));
 553                 break;
 554         default:
 555                 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
 556                                 session_name, get_trigger_name(work_item->trigger),
 557                                 lttng_strerror(-cmd_ret));
 558                 lttng_action_increase_execution_failure_count(action);
 559                 break;
 560         }
 561
 562 error_unlock_session:
 563         session_unlock(session);
 564 error_put_session:
 565         session_put(session);
 566 error_unlock_list:
 567         session_unlock_list();
 568 end:
 569         return ret;
 570 }
 571
 572 static int action_executor_snapshot_session_handler(
 573                 struct action_executor *executor,
 574                 const struct action_work_item *work_item,
 575                 struct action_work_subitem *item)
 576 {
 577         int ret = 0;
 578         const char *session_name;
 579         enum lttng_action_status action_status;
 580         struct ltt_session *session;
 581         const struct lttng_snapshot_output default_snapshot_output = {
 582                 .max_size = UINT64_MAX,
 583         };
 584         const struct lttng_snapshot_output *snapshot_output =
 585                         &default_snapshot_output;
 586         enum lttng_error_code cmd_ret;
 587         struct lttng_action *action = item->action;
 588
 589         /*
 590          * Validate if, at the moment the action was queued, the target session
 591          * existed. If not, skip the action altogether.
 592          */
 593         if (!item->context.session_id.is_set) {
 594                 DBG("Session was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
 595                                 get_action_name(action),
 596                                 get_trigger_name(work_item->trigger));
 597                 lttng_action_increase_execution_failure_count(action);
 598                 ret = 0;
 599                 goto end;
 600         }
 601
 602         action_status = lttng_action_snapshot_session_get_session_name(
 603                         action, &session_name);
 604         if (action_status != LTTNG_ACTION_STATUS_OK) {
 605                 ERR("Failed to get session name from `%s` action",
 606                                 get_action_name(action));
 607                 ret = -1;
 608                 goto end;
 609         }
 610
 611         action_status = lttng_action_snapshot_session_get_output(
 612                         action, &snapshot_output);
 613         if (action_status != LTTNG_ACTION_STATUS_OK &&
 614                         action_status != LTTNG_ACTION_STATUS_UNSET) {
 615                 ERR("Failed to get output from `%s` action",
 616                                 get_action_name(action));
 617                 ret = -1;
 618                 goto end;
 619         }
 620
 621         session_lock_list();
 622         session = session_find_by_name(session_name);
 623         if (!session) {
 624                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 625                                 session_name, get_action_name(action),
 626                                 get_trigger_name(work_item->trigger));
 627                 lttng_action_increase_execution_failure_count(action);
 628                 goto error_unlock_list;
 629         }
 630
 631         /*
 632          * Check if the session id is the same as when the work item was
 633          * enqueued.
 634          */
 635         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 636                 DBG("Session id for session `%s` (id: %" PRIu64
 637                     " is not the same that was sampled (id: %" PRIu64
 638                     " at the moment the work item was enqueued for `%s` action of trigger `%s`",
 639                                 session_name, session->id,
 640                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 641                                 get_action_name(action),
 642                                 get_trigger_name(work_item->trigger));
 643                 ret = 0;
 644                 goto error_put_session;
 645         }
 646
 647         session_lock(session);
 648         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 649                 goto error_unlock_session;
 650         }
 651
 652         cmd_ret = cmd_snapshot_record(session, snapshot_output, 0);
 653         switch (cmd_ret) {
 654         case LTTNG_OK:
 655                 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
 656                                 session_name, get_trigger_name(work_item->trigger));
 657                 break;
 658         default:
 659                 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
 660                                 session_name, get_trigger_name(work_item->trigger),
 661                                 lttng_strerror(-cmd_ret));
 662                 lttng_action_increase_execution_failure_count(action);
 663                 break;
 664         }
 665
 666 error_unlock_session:
 667         session_unlock(session);
 668 error_put_session:
 669         session_put(session);
 670 error_unlock_list:
 671         session_unlock_list();
 672 end:
 673         return ret;
 674 }
 675
 676 static int action_executor_list_handler(struct action_executor *executor,
 677                 const struct action_work_item *work_item,
 678                 struct action_work_subitem *item)
 679 {
 680         ERR("Execution of a list action by the action executor should never occur");
 681         abort();
 682 }
 683
 684 static int action_executor_generic_handler(struct action_executor *executor,
 685                 const struct action_work_item *work_item,
 686                 struct action_work_subitem *item)
 687 {
 688         int ret;
 689         struct lttng_action *action = item->action;
 690         const enum lttng_action_type action_type = lttng_action_get_type(action);
 691
 692         assert(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
 693
 694         lttng_action_increase_execution_request_count(action);
 695         if (!lttng_action_should_execute(action)) {
 696                 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64,
 697                                 get_action_name(action),
 698                                 get_trigger_name(work_item->trigger),
 699                                 work_item->id);
 700                 ret = 0;
 701                 goto end;
 702         }
 703
 704         lttng_action_increase_execution_count(action);
 705         DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64,
 706                         get_action_name(action),
 707                         get_trigger_name(work_item->trigger),
 708                         work_item->id);
 709         ret = action_executors[action_type](executor, work_item, item);
 710 end:
 711         return ret;
 712 }
 713
 714 static int action_work_item_execute(struct action_executor *executor,
 715                 struct action_work_item *work_item)
 716 {
 717         int ret;
 718         size_t count, i;
 719
 720         DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
 721                         work_item->id, get_trigger_name(work_item->trigger));
 722
 723         count = lttng_dynamic_array_get_count(&work_item->subitems);
 724         for (i = 0; i < count; i++) {
 725                 struct action_work_subitem *item;
 726
 727                 item = lttng_dynamic_array_get_element(&work_item->subitems, i);
 728                 ret = action_executor_generic_handler(
 729                                 executor, work_item, item);
 730                 if (ret) {
 731                         goto end;
 732                 }
 733         }
 734 end:
 735         DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
 736                         work_item->id, get_trigger_name(work_item->trigger));
 737         return ret;
 738 }
 739
 740 static void action_work_item_destroy(struct action_work_item *work_item)
 741 {
 742         lttng_trigger_put(work_item->trigger);
 743         lttng_evaluation_destroy(work_item->evaluation);
 744         notification_client_list_put(work_item->client_list);
 745         lttng_dynamic_array_reset(&work_item->subitems);
 746         free(work_item);
 747 }
 748
 749 static void *action_executor_thread(void *_data)
 750 {
 751         struct action_executor *executor = _data;
 752
 753         assert(executor);
 754
 755         health_register(the_health_sessiond,
 756                         HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
 757
 758         rcu_register_thread();
 759         rcu_thread_online();
 760
 761         DBG("Entering work execution loop");
 762         pthread_mutex_lock(&executor->work.lock);
 763         while (!executor->should_quit) {
 764                 int ret = 0;
 765                 struct action_work_item *work_item;
 766
 767                 health_code_update();
 768                 if (executor->work.pending_count == 0) {
 769                         health_poll_entry();
 770                         DBG("No work items enqueued, entering wait");
 771                         pthread_cond_wait(&executor->work.cond,
 772                                         &executor->work.lock);
 773                         DBG("Woke-up from wait");
 774                         health_poll_exit();
 775                         continue;
 776                 }
 777
 778                 /* Pop item from front of the list with work lock held. */
 779                 work_item = cds_list_first_entry(&executor->work.list,
 780                                 struct action_work_item, list_node);
 781                 cds_list_del(&work_item->list_node);
 782                 executor->work.pending_count--;
 783
 784                 /*
 785                  * Work can be performed without holding the work lock,
 786                  * allowing new items to be queued.
 787                  */
 788                 pthread_mutex_unlock(&executor->work.lock);
 789
 790                 /* Execute item only if a trigger is registered. */
 791                 lttng_trigger_lock(work_item->trigger);
 792                 if (!lttng_trigger_is_registered(work_item->trigger)) {
 793                         const char *trigger_name = NULL;
 794                         uid_t trigger_owner_uid;
 795                         enum lttng_trigger_status trigger_status;
 796
 797                         trigger_name = get_trigger_name(work_item->trigger);
 798
 799                         trigger_status = lttng_trigger_get_owner_uid(
 800                                         work_item->trigger, &trigger_owner_uid);
 801                         assert(trigger_status == LTTNG_TRIGGER_STATUS_OK);
 802
 803                         DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64 ", trigger name = `%s`, trigger owner uid = %d",
 804                                         work_item->id, trigger_name,
 805                                         (int) trigger_owner_uid);
 806                         ret = 0;
 807                         goto skip_execute;
 808                 }
 809
 810                 ret = action_work_item_execute(executor, work_item);
 811
 812         skip_execute:
 813                 lttng_trigger_unlock(work_item->trigger);
 814                 action_work_item_destroy(work_item);
 815                 if (ret) {
 816                         /* Fatal error. */
 817                         break;
 818                 }
 819
 820                 health_code_update();
 821                 pthread_mutex_lock(&executor->work.lock);
 822         }
 823
 824         if (executor->should_quit) {
 825                 pthread_mutex_unlock(&executor->work.lock);
 826         }
 827         DBG("Left work execution loop");
 828
 829         health_code_update();
 830
 831         rcu_thread_offline();
 832         rcu_unregister_thread();
 833         health_unregister(the_health_sessiond);
 834
 835         return NULL;
 836 }
 837
 838 static bool shutdown_action_executor_thread(void *_data)
 839 {
 840         struct action_executor *executor = _data;
 841
 842         pthread_mutex_lock(&executor->work.lock);
 843         executor->should_quit = true;
 844         pthread_cond_signal(&executor->work.cond);
 845         pthread_mutex_unlock(&executor->work.lock);
 846         return true;
 847 }
 848
 849 static void clean_up_action_executor_thread(void *_data)
 850 {
 851         struct action_executor *executor = _data;
 852
 853         assert(cds_list_empty(&executor->work.list));
 854
 855         pthread_mutex_destroy(&executor->work.lock);
 856         pthread_cond_destroy(&executor->work.cond);
 857         free(executor);
 858 }
 859
 860 struct action_executor *action_executor_create(
 861                 struct notification_thread_handle *handle)
 862 {
 863         struct action_executor *executor = zmalloc(sizeof(*executor));
 864
 865         if (!executor) {
 866                 goto end;
 867         }
 868
 869         CDS_INIT_LIST_HEAD(&executor->work.list);
 870         pthread_cond_init(&executor->work.cond, NULL);
 871         pthread_mutex_init(&executor->work.lock, NULL);
 872         executor->notification_thread_handle = handle;
 873
 874         executor->thread = lttng_thread_create(THREAD_NAME,
 875                         action_executor_thread, shutdown_action_executor_thread,
 876                         clean_up_action_executor_thread, executor);
 877 end:
 878         return executor;
 879 }
 880
 881 void action_executor_destroy(struct action_executor *executor)
 882 {
 883         struct action_work_item *work_item, *tmp;
 884
 885         /* TODO Wait for work list to drain? */
 886         lttng_thread_shutdown(executor->thread);
 887         pthread_mutex_lock(&executor->work.lock);
 888         if (executor->work.pending_count != 0) {
 889                 WARN("%" PRIu64
 890                         " trigger action%s still queued for execution and will be discarded",
 891                                 executor->work.pending_count,
 892                                 executor->work.pending_count == 1 ? " is" :
 893                                                                     "s are");
 894         }
 895
 896         cds_list_for_each_entry_safe (
 897                         work_item, tmp, &executor->work.list, list_node) {
 898                 WARN("Discarding action work item %" PRIu64
 899                                 " associated to trigger `%s`",
 900                                 work_item->id, get_trigger_name(work_item->trigger));
 901                 cds_list_del(&work_item->list_node);
 902                 action_work_item_destroy(work_item);
 903         }
 904         pthread_mutex_unlock(&executor->work.lock);
 905         lttng_thread_put(executor->thread);
 906 }
 907
 908 /* RCU read-lock must be held by the caller. */
 909 enum action_executor_status action_executor_enqueue_trigger(
 910                 struct action_executor *executor,
 911                 struct lttng_trigger *trigger,
 912                 struct lttng_evaluation *evaluation,
 913                 const struct lttng_credentials *object_creds,
 914                 struct notification_client_list *client_list)
 915 {
 916         int ret;
 917         enum action_executor_status executor_status = ACTION_EXECUTOR_STATUS_OK;
 918         const uint64_t work_item_id = executor->next_work_item_id++;
 919         struct action_work_item *work_item;
 920         bool signal = false;
 921
 922         assert(trigger);
 923
 924         pthread_mutex_lock(&executor->work.lock);
 925         /* Check for queue overflow. */
 926         if (executor->work.pending_count >= MAX_QUEUED_WORK_COUNT) {
 927                 /* Most likely spammy, remove if it is the case. */
 928                 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64,
 929                                 get_trigger_name(trigger), work_item_id);
 930                 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
 931                 goto error_unlock;
 932         }
 933
 934         work_item = zmalloc(sizeof(*work_item));
 935         if (!work_item) {
 936                 PERROR("Failed to allocate action executor work item: trigger name = `%s`",
 937                                 get_trigger_name(trigger));
 938                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 939                 goto error_unlock;
 940         }
 941
 942         lttng_trigger_get(trigger);
 943         if (client_list) {
 944                 const bool reference_acquired =
 945                                 notification_client_list_get(client_list);
 946
 947                 assert(reference_acquired);
 948         }
 949
 950         *work_item = (typeof(*work_item)){
 951                         .id = work_item_id,
 952                         .trigger = trigger,
 953                         /* Ownership transferred to the work item. */
 954                         .evaluation = evaluation,
 955                         .object_creds = {
 956                                 .is_set = !!object_creds,
 957                                 .value = object_creds ? *object_creds :
 958                                         (typeof(work_item->object_creds.value)) {},
 959                         },
 960                         .client_list = client_list,
 961                         .list_node = CDS_LIST_HEAD_INIT(work_item->list_node),
 962         };
 963
 964         evaluation = NULL;
 965
 966         /* Build the array of action work subitems for the passed trigger. */
 967         lttng_dynamic_array_init(&work_item->subitems,
 968                         sizeof(struct action_work_subitem),
 969                         action_work_subitem_destructor);
 970
 971         ret = populate_subitem_array_from_trigger(
 972                         trigger, &work_item->subitems);
 973         if (ret) {
 974                 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
 975                                 get_trigger_name(trigger));
 976                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 977                 goto error_unlock;
 978         }
 979
 980         cds_list_add_tail(&work_item->list_node, &executor->work.list);
 981         executor->work.pending_count++;
 982         DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64,
 983                         get_trigger_name(trigger), work_item_id);
 984         signal = true;
 985
 986 error_unlock:
 987         if (signal) {
 988                 pthread_cond_signal(&executor->work.cond);
 989         }
 990
 991         pthread_mutex_unlock(&executor->work.lock);
 992         lttng_evaluation_destroy(evaluation);
 993         return executor_status;
 994 }
 995
 996 static int add_action_to_subitem_array(struct lttng_action *action,
 997                 struct lttng_dynamic_array *subitems)
 998 {
 999         int ret = 0;
1000         enum lttng_action_type type = lttng_action_get_type(action);
1001         const char *session_name = NULL;
1002         enum lttng_action_status status;
1003         struct action_work_subitem subitem = {
1004                 .action = NULL,
1005                 .context = {
1006                         .session_id = LTTNG_OPTIONAL_INIT_UNSET,
1007                 },
1008         };
1009
1010         assert(action);
1011         assert(subitems);
1012
1013         if (type == LTTNG_ACTION_TYPE_LIST) {
1014                 unsigned int count, i;
1015
1016                 status = lttng_action_list_get_count(action, &count);
1017                 assert(status == LTTNG_ACTION_STATUS_OK);
1018
1019                 for (i = 0; i < count; i++) {
1020                         struct lttng_action *inner_action = NULL;
1021
1022                         inner_action = lttng_action_list_borrow_mutable_at_index(
1023                                         action, i);
1024                         assert(inner_action);
1025                         ret = add_action_to_subitem_array(
1026                                         inner_action, subitems);
1027                         if (ret) {
1028                                 goto end;
1029                         }
1030                 }
1031
1032                 /*
1033                  * Go directly to the end since there is no need to add the
1034                  * list action by itself to the subitems array.
1035                  */
1036                 goto end;
1037         }
1038
1039         /* Gather execution context. */
1040         switch (type) {
1041         case LTTNG_ACTION_TYPE_NOTIFY:
1042                 break;
1043         case LTTNG_ACTION_TYPE_START_SESSION:
1044                 status = lttng_action_start_session_get_session_name(
1045                                 action, &session_name);
1046                 assert(status == LTTNG_ACTION_STATUS_OK);
1047                 break;
1048         case LTTNG_ACTION_TYPE_STOP_SESSION:
1049                 status = lttng_action_stop_session_get_session_name(
1050                                 action, &session_name);
1051                 assert(status == LTTNG_ACTION_STATUS_OK);
1052                 break;
1053         case LTTNG_ACTION_TYPE_ROTATE_SESSION:
1054                 status = lttng_action_rotate_session_get_session_name(
1055                                 action, &session_name);
1056                 assert(status == LTTNG_ACTION_STATUS_OK);
1057                 break;
1058         case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION:
1059                 status = lttng_action_snapshot_session_get_session_name(
1060                                 action, &session_name);
1061                 assert(status == LTTNG_ACTION_STATUS_OK);
1062                 break;
1063         case LTTNG_ACTION_TYPE_LIST:
1064         case LTTNG_ACTION_TYPE_UNKNOWN:
1065                 /* Fallthrough */
1066         default:
1067                 abort();
1068                 break;
1069         }
1070
1071         /*
1072          * Fetch the session execution context info as needed.
1073          * Note that we could decide to not add an action for which we know the
1074          * execution will not happen (i.e no session exists for that name). For
1075          * now we leave the decision to skip to the action executor for sake of
1076          * simplicity and consistency.
1077          */
1078         if (session_name != NULL) {
1079                 uint64_t session_id;
1080
1081                 /*
1082                  * Instantaneous sampling of the session id if present.
1083                  *
1084                  * This method is preferred over `sessiond_find_by_name` then
1085                  * fetching the session'd id since `sessiond_find_by_name`
1086                  * requires the session list lock to be taken.
1087                  *
1088                  * Taking the session list lock can lead to a deadlock
1089                  * between the action executor and the notification thread
1090                  * (caller of add_action_to_subitem_array). It is okay if the
1091                  * session state changes between the enqueuing time and the
1092                  * execution time. The execution context is validated at
1093                  * execution time.
1094                  */
1095                 if (sample_session_id_by_name(session_name, &session_id)) {
1096                         LTTNG_OPTIONAL_SET(&subitem.context.session_id,
1097                                         session_id);
1098                 }
1099         }
1100
1101         /* Get a reference to the action. */
1102         lttng_action_get(action);
1103         subitem.action = action;
1104
1105         ret = lttng_dynamic_array_add_element(subitems, &subitem);
1106         if (ret) {
1107                 ERR("Failed to add work subitem to the subitem array");
1108                 lttng_action_put(action);
1109                 ret = -1;
1110                 goto end;
1111         }
1112
1113 end:
1114         return ret;
1115 }
1116
1117 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
1118                 struct lttng_dynamic_array *subitems)
1119 {
1120         struct lttng_action *action;
1121
1122         action = lttng_trigger_get_action(trigger);
1123         assert(action);
1124
1125         return add_action_to_subitem_array(action, subitems);
1126 }