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