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

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