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                 update_communication = false;
 237                 break;
 238         case CLIENT_TRANSMISSION_STATUS_QUEUED:
 239                 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64,
 240                                 client->id);
 241                 break;
 242         case CLIENT_TRANSMISSION_STATUS_FAIL:
 243                 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64,
 244                                 client->id);
 245                 break;
 246         default:
 247                 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64,
 248                                 client->id);
 249                 ret = -1;
 250                 goto end;
 251         }
 252
 253         if (!update_communication) {
 254                 goto end;
 255         }
 256
 257         /* Safe to read client's id without locking as it is immutable. */
 258         ret = notification_thread_client_communication_update(
 259                         executor->notification_thread_handle, client->id,
 260                         status);
 261 end:
 262         return ret;
 263 }
 264
 265 static int action_executor_notify_handler(struct action_executor *executor,
 266                 const struct action_work_item *work_item,
 267                 struct action_work_subitem *item)
 268 {
 269         return notification_client_list_send_evaluation(work_item->client_list,
 270                         work_item->trigger,
 271                         work_item->evaluation,
 272                         work_item->object_creds.is_set ?
 273                                         &(work_item->object_creds.value) :
 274                                         NULL,
 275                         client_handle_transmission_status, executor);
 276 }
 277
 278 static int action_executor_start_session_handler(
 279                 struct action_executor *executor,
 280                 const struct action_work_item *work_item,
 281                 struct action_work_subitem *item)
 282 {
 283         int ret = 0;
 284         const char *session_name;
 285         enum lttng_action_status action_status;
 286         struct ltt_session *session;
 287         enum lttng_error_code cmd_ret;
 288         struct lttng_action *action = item->action;
 289
 290         action_status = lttng_action_start_session_get_session_name(
 291                         action, &session_name);
 292         if (action_status != LTTNG_ACTION_STATUS_OK) {
 293                 ERR("Failed to get session name from `%s` action",
 294                                 get_action_name(action));
 295                 ret = -1;
 296                 goto end;
 297         }
 298
 299         /*
 300          * Validate if at the moment of the action was queued the session
 301          * existed. If not skip the action altogether.
 302          */
 303         if (!item->context.session_id.is_set) {
 304                 DBG("Session `%s` was not present at the moment the work item was enqueued for `%s` action of trigger `%s`",
 305                                 session_name, get_action_name(action),
 306                                 get_trigger_name(work_item->trigger));
 307                 lttng_action_increase_execution_failure_count(action);
 308                 ret = 0;
 309                 goto end;
 310         }
 311
 312         session_lock_list();
 313         session = session_find_by_name(session_name);
 314         if (!session) {
 315                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 316                                 session_name, get_action_name(action),
 317                                 get_trigger_name(work_item->trigger));
 318                 goto error_unlock_list;
 319         }
 320
 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;
 334                 goto error_put_session;
 335         }
 336
 337         session_lock(session);
 338         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 339                 goto error_unlock_session;
 340         }
 341
 342         cmd_ret = cmd_start_trace(session);
 343         switch (cmd_ret) {
 344         case LTTNG_OK:
 345                 DBG("Successfully started session `%s` on behalf of trigger `%s`",
 346                                 session_name, get_trigger_name(work_item->trigger));
 347                 break;
 348         case LTTNG_ERR_TRACE_ALREADY_STARTED:
 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));
 351                 break;
 352         default:
 353                 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
 354                                 session_name, get_trigger_name(work_item->trigger),
 355                                 lttng_strerror(-cmd_ret));
 356                 lttng_action_increase_execution_failure_count(action);
 357                 break;
 358         }
 359
 360 error_unlock_session:
 361         session_unlock(session);
 362 error_put_session:
 363         session_put(session);
 364 error_unlock_list:
 365         session_unlock_list();
 366 end:
 367         return ret;
 368 }
 369
 370 static int action_executor_stop_session_handler(
 371                 struct action_executor *executor,
 372                 const struct action_work_item *work_item,
 373                 struct action_work_subitem *item)
 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;
 380         struct lttng_action *action = item->action;
 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
 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
 404         session_lock_list();
 405         session = session_find_by_name(session_name);
 406         if (!session) {
 407                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 408                                 session_name, get_action_name(action),
 409                                 get_trigger_name(work_item->trigger));
 410                 lttng_action_increase_execution_failure_count(action);
 411                 goto error_unlock_list;
 412         }
 413
 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;
 427                 goto error_put_session;
 428         }
 429
 430         session_lock(session);
 431         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 432                 goto error_unlock_session;
 433         }
 434
 435         cmd_ret = cmd_stop_trace(session);
 436         switch (cmd_ret) {
 437         case LTTNG_OK:
 438                 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
 439                                 session_name, get_trigger_name(work_item->trigger));
 440                 break;
 441         case LTTNG_ERR_TRACE_ALREADY_STOPPED:
 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));
 444                 break;
 445         default:
 446                 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
 447                                 session_name, get_trigger_name(work_item->trigger),
 448                                 lttng_strerror(-cmd_ret));
 449                 lttng_action_increase_execution_failure_count(action);
 450                 break;
 451         }
 452
 453 error_unlock_session:
 454         session_unlock(session);
 455 error_put_session:
 456         session_put(session);
 457 error_unlock_list:
 458         session_unlock_list();
 459 end:
 460         return ret;
 461 }
 462
 463 static int action_executor_rotate_session_handler(
 464                 struct action_executor *executor,
 465                 const struct action_work_item *work_item,
 466                 struct action_work_subitem *item)
 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;
 473         struct lttng_action *action = item->action;
 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
 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
 497         session_lock_list();
 498         session = session_find_by_name(session_name);
 499         if (!session) {
 500                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 501                                 session_name, get_action_name(action),
 502                                 get_trigger_name(work_item->trigger));
 503                 lttng_action_increase_execution_failure_count(action);
 504                 goto error_unlock_list;
 505         }
 506
 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;
 520                 goto error_put_session;
 521         }
 522
 523         session_lock(session);
 524         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 525                 goto error_unlock_session;
 526         }
 527
 528         cmd_ret = cmd_rotate_session(session, NULL, false,
 529                         LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 530         switch (cmd_ret) {
 531         case LTTNG_OK:
 532                 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
 533                                 session_name, get_trigger_name(work_item->trigger));
 534                 break;
 535         case LTTNG_ERR_ROTATION_PENDING:
 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));
 538                 lttng_action_increase_execution_failure_count(action);
 539                 break;
 540         case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
 541         case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
 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));
 544                 break;
 545         default:
 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),
 548                                 lttng_strerror(-cmd_ret));
 549                 lttng_action_increase_execution_failure_count(action);
 550                 break;
 551         }
 552
 553 error_unlock_session:
 554         session_unlock(session);
 555 error_put_session:
 556         session_put(session);
 557 error_unlock_list:
 558         session_unlock_list();
 559 end:
 560         return ret;
 561 }
 562
 563 static int action_executor_snapshot_session_handler(
 564                 struct action_executor *executor,
 565                 const struct action_work_item *work_item,
 566                 struct action_work_subitem *item)
 567 {
 568         int ret = 0;
 569         const char *session_name;
 570         enum lttng_action_status action_status;
 571         struct ltt_session *session;
 572         const struct lttng_snapshot_output default_snapshot_output = {
 573                 .max_size = UINT64_MAX,
 574         };
 575         const struct lttng_snapshot_output *snapshot_output =
 576                         &default_snapshot_output;
 577         enum lttng_error_code cmd_ret;
 578         struct lttng_action *action = item->action;
 579
 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) {
 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),
 587                                 get_trigger_name(work_item->trigger));
 588                 lttng_action_increase_execution_failure_count(action);
 589                 ret = 0;
 590                 goto end;
 591         }
 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) {
 615                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 616                                 session_name, get_action_name(action),
 617                                 get_trigger_name(work_item->trigger));
 618                 lttng_action_increase_execution_failure_count(action);
 619                 goto error_unlock_list;
 620         }
 621
 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;
 635                 goto error_put_session;
 636         }
 637
 638         session_lock(session);
 639         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 640                 goto error_unlock_session;
 641         }
 642
 643         cmd_ret = cmd_snapshot_record(session, snapshot_output, 0);
 644         switch (cmd_ret) {
 645         case LTTNG_OK:
 646                 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
 647                                 session_name, get_trigger_name(work_item->trigger));
 648                 break;
 649         default:
 650                 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
 651                                 session_name, get_trigger_name(work_item->trigger),
 652                                 lttng_strerror(-cmd_ret));
 653                 lttng_action_increase_execution_failure_count(action);
 654                 break;
 655         }
 656
 657 error_unlock_session:
 658         session_unlock(session);
 659 error_put_session:
 660         session_put(session);
 661 error_unlock_list:
 662         session_unlock_list();
 663 end:
 664         return ret;
 665 }
 666
 667 static int action_executor_list_handler(struct action_executor *executor,
 668                 const struct action_work_item *work_item,
 669                 struct action_work_subitem *item)
 670 {
 671         ERR("Execution of a list action by the action executor should never occur");
 672         abort();
 673 }
 674
 675 static int action_executor_generic_handler(struct action_executor *executor,
 676                 const struct action_work_item *work_item,
 677                 struct action_work_subitem *item)
 678 {
 679         int ret;
 680         struct lttng_action *action = item->action;
 681         const enum lttng_action_type action_type = lttng_action_get_type(action);
 682
 683         assert(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
 684
 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);
 696         DBG("Executing 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 = action_executors[action_type](executor, work_item, item);
 701 end:
 702         return ret;
 703 }
 704
 705 static int action_work_item_execute(struct action_executor *executor,
 706                 struct action_work_item *work_item)
 707 {
 708         int ret;
 709         size_t count, i;
 710
 711         DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
 712                         work_item->id, get_trigger_name(work_item->trigger));
 713
 714         count = lttng_dynamic_array_get_count(&work_item->subitems);
 715         for (i = 0; i < count; i++) {
 716                 struct action_work_subitem *item;
 717
 718                 item = lttng_dynamic_array_get_element(&work_item->subitems, i);
 719                 ret = action_executor_generic_handler(
 720                                 executor, work_item, item);
 721                 if (ret) {
 722                         goto end;
 723                 }
 724         }
 725 end:
 726         DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
 727                         work_item->id, get_trigger_name(work_item->trigger));
 728         return ret;
 729 }
 730
 731 static 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);
 736         lttng_dynamic_array_reset(&work_item->subitems);
 737         free(work_item);
 738 }
 739
 740 static void *action_executor_thread(void *_data)
 741 {
 742         struct action_executor *executor = _data;
 743
 744         assert(executor);
 745
 746         health_register(the_health_sessiond,
 747                         HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
 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) {
 755                 int ret = 0;
 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
 769                 /* Pop item from front of the list with work lock held. */
 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);
 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
 788                         trigger_name = get_trigger_name(work_item->trigger);
 789
 790                         trigger_status = lttng_trigger_get_owner_uid(
 791                                         work_item->trigger, &trigger_owner_uid);
 792                         assert(trigger_status == LTTNG_TRIGGER_STATUS_OK);
 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
 801                 ret = action_work_item_execute(executor, work_item);
 802
 803         skip_execute:
 804                 lttng_trigger_unlock(work_item->trigger);
 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
 815         if (executor->should_quit) {
 816                 pthread_mutex_unlock(&executor->work.lock);
 817         }
 818         DBG("Left work execution loop");
 819
 820         health_code_update();
 821
 822         rcu_thread_offline();
 823         rcu_unregister_thread();
 824         health_unregister(the_health_sessiond);
 825
 826         return NULL;
 827 }
 828
 829 static bool shutdown_action_executor_thread(void *_data)
 830 {
 831         struct action_executor *executor = _data;
 832
 833         pthread_mutex_lock(&executor->work.lock);
 834         executor->should_quit = true;
 835         pthread_cond_signal(&executor->work.cond);
 836         pthread_mutex_unlock(&executor->work.lock);
 837         return true;
 838 }
 839
 840 static void clean_up_action_executor_thread(void *_data)
 841 {
 842         struct action_executor *executor = _data;
 843
 844         assert(cds_list_empty(&executor->work.list));
 845
 846         pthread_mutex_destroy(&executor->work.lock);
 847         pthread_cond_destroy(&executor->work.cond);
 848         free(executor);
 849 }
 850
 851 struct action_executor *action_executor_create(
 852                 struct notification_thread_handle *handle)
 853 {
 854         struct action_executor *executor = zmalloc(sizeof(*executor));
 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);
 868 end:
 869         return executor;
 870 }
 871
 872 void 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
 890                                 " associated to trigger `%s`",
 891                                 work_item->id, get_trigger_name(work_item->trigger));
 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. */
 900 enum action_executor_status action_executor_enqueue_trigger(
 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 {
 907         int ret;
 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;
 912
 913         assert(trigger);
 914
 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. */
 919                 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64,
 920                                 get_trigger_name(trigger), work_item_id);
 921                 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
 922                 goto error_unlock;
 923         }
 924
 925         work_item = zmalloc(sizeof(*work_item));
 926         if (!work_item) {
 927                 PERROR("Failed to allocate action executor work item: trigger name = `%s`",
 928                                 get_trigger_name(trigger));
 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
 938                 assert(reference_acquired);
 939         }
 940
 941         *work_item = (typeof(*work_item)){
 942                         .id = work_item_id,
 943                         .trigger = trigger,
 944                         /* Ownership transferred to the work item. */
 945                         .evaluation = evaluation,
 946                         .object_creds = {
 947                                 .is_set = !!object_creds,
 948                                 .value = object_creds ? *object_creds :
 949                                         (typeof(work_item->object_creds.value)) {},
 950                         },
 951                         .client_list = client_list,
 952                         .list_node = CDS_LIST_HEAD_INIT(work_item->list_node),
 953         };
 954
 955         evaluation = NULL;
 956
 957         /* Build the array of action work subitems for the passed trigger. */
 958         lttng_dynamic_array_init(&work_item->subitems,
 959                         sizeof(struct action_work_subitem),
 960                         action_work_subitem_destructor);
 961
 962         ret = populate_subitem_array_from_trigger(
 963                         trigger, &work_item->subitems);
 964         if (ret) {
 965                 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
 966                                 get_trigger_name(trigger));
 967                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 968                 goto error_unlock;
 969         }
 970
 971         cds_list_add_tail(&work_item->list_node, &executor->work.list);
 972         executor->work.pending_count++;
 973         DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64,
 974                         get_trigger_name(trigger), work_item_id);
 975         signal = true;
 976
 977 error_unlock:
 978         if (signal) {
 979                 pthread_cond_signal(&executor->work.cond);
 980         }
 981
 982         pthread_mutex_unlock(&executor->work.lock);
 983         lttng_evaluation_destroy(evaluation);
 984         return executor_status;
 985 }
 986
 987 static int add_action_to_subitem_array(struct lttng_action *action,
 988                 struct lttng_dynamic_array *subitems)
 989 {
 990         int ret = 0;
 991         enum lttng_action_type type = lttng_action_get_type(action);
 992         const char *session_name = NULL;
 993         enum lttng_action_status status;
 994         struct action_work_subitem subitem = {
 995                 .action = NULL,
 996                 .context = {
 997                         .session_id = LTTNG_OPTIONAL_INIT_UNSET,
 998                 },
 999         };
1000
1001         assert(action);
1002         assert(subitems);
1003
1004         if (type == LTTNG_ACTION_TYPE_LIST) {
1005                 unsigned int count, i;
1006
1007                 status = lttng_action_list_get_count(action, &count);
1008                 assert(status == LTTNG_ACTION_STATUS_OK);
1009
1010                 for (i = 0; i < count; i++) {
1011                         struct lttng_action *inner_action = NULL;
1012
1013                         inner_action = lttng_action_list_borrow_mutable_at_index(
1014                                         action, i);
1015                         assert(inner_action);
1016                         ret = add_action_to_subitem_array(
1017                                         inner_action, subitems);
1018                         if (ret) {
1019                                 goto end;
1020                         }
1021                 }
1022
1023                 /*
1024                  * Go directly to the end since there is no need to add the
1025                  * list action by itself to the subitems array.
1026                  */
1027                 goto end;
1028         }
1029
1030         /* Gather execution context. */
1031         switch (type) {
1032         case LTTNG_ACTION_TYPE_NOTIFY:
1033                 break;
1034         case LTTNG_ACTION_TYPE_START_SESSION:
1035                 status = lttng_action_start_session_get_session_name(
1036                                 action, &session_name);
1037                 assert(status == LTTNG_ACTION_STATUS_OK);
1038                 break;
1039         case LTTNG_ACTION_TYPE_STOP_SESSION:
1040                 status = lttng_action_stop_session_get_session_name(
1041                                 action, &session_name);
1042                 assert(status == LTTNG_ACTION_STATUS_OK);
1043                 break;
1044         case LTTNG_ACTION_TYPE_ROTATE_SESSION:
1045                 status = lttng_action_rotate_session_get_session_name(
1046                                 action, &session_name);
1047                 assert(status == LTTNG_ACTION_STATUS_OK);
1048                 break;
1049         case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION:
1050                 status = lttng_action_snapshot_session_get_session_name(
1051                                 action, &session_name);
1052                 assert(status == LTTNG_ACTION_STATUS_OK);
1053                 break;
1054         case LTTNG_ACTION_TYPE_LIST:
1055         case LTTNG_ACTION_TYPE_UNKNOWN:
1056                 /* Fallthrough */
1057         default:
1058                 abort();
1059                 break;
1060         }
1061
1062         /*
1063          * Fetch the session execution context info as needed.
1064          * Note that we could decide to not add an action for which we know the
1065          * execution will not happen (i.e no session exists for that name). For
1066          * now we leave the decision to skip to the action executor for sake of
1067          * simplicity and consistency.
1068          */
1069         if (session_name != NULL) {
1070                 uint64_t session_id;
1071
1072                 /*
1073                  * Instantaneous sampling of the session id if present.
1074                  *
1075                  * This method is preferred over `sessiond_find_by_name` then
1076                  * fetching the session'd id since `sessiond_find_by_name`
1077                  * requires the session list lock to be taken.
1078                  *
1079                  * Taking the session list lock can lead to a deadlock
1080                  * between the action executor and the notification thread
1081                  * (caller of add_action_to_subitem_array). It is okay if the
1082                  * session state changes between the enqueuing time and the
1083                  * execution time. The execution context is validated at
1084                  * execution time.
1085                  */
1086                 if (sample_session_id_by_name(session_name, &session_id)) {
1087                         LTTNG_OPTIONAL_SET(&subitem.context.session_id,
1088                                         session_id);
1089                 }
1090         }
1091
1092         /* Get a reference to the action. */
1093         lttng_action_get(action);
1094         subitem.action = action;
1095
1096         ret = lttng_dynamic_array_add_element(subitems, &subitem);
1097         if (ret) {
1098                 ERR("Failed to add work subitem to the subitem array");
1099                 lttng_action_put(action);
1100                 ret = -1;
1101                 goto end;
1102         }
1103
1104 end:
1105         return ret;
1106 }
1107
1108 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
1109                 struct lttng_dynamic_array *subitems)
1110 {
1111         struct lttng_action *action;
1112
1113         action = lttng_trigger_get_action(trigger);
1114         assert(action);
1115
1116         return add_action_to_subitem_array(action, subitems);
1117 }