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         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,
 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,
 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,
 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(struct action_executor *executor,
 637                 const struct action_work_item *work_item,
 638                 struct action_work_subitem *item)
 639 {
 640         ERR("Execution of a list action by the action executor should never occur");
 641         abort();
 642 }
 643
 644 static int action_executor_generic_handler(struct action_executor *executor,
 645                 const struct action_work_item *work_item,
 646                 struct action_work_subitem *item)
 647 {
 648         int ret;
 649         struct lttng_action *action = item->action;
 650         const enum lttng_action_type action_type = lttng_action_get_type(action);
 651
 652         LTTNG_ASSERT(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
 653
 654         lttng_action_increase_execution_request_count(action);
 655         if (!lttng_action_should_execute(action)) {
 656                 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64,
 657                                 get_action_name(action),
 658                                 get_trigger_name(work_item->trigger),
 659                                 work_item->id);
 660                 ret = 0;
 661                 goto end;
 662         }
 663
 664         lttng_action_increase_execution_count(action);
 665         DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64,
 666                         get_action_name(action),
 667                         get_trigger_name(work_item->trigger),
 668                         work_item->id);
 669         ret = action_executors[action_type](executor, work_item, item);
 670 end:
 671         return ret;
 672 }
 673
 674 static int action_work_item_execute(struct action_executor *executor,
 675                 struct action_work_item *work_item)
 676 {
 677         int ret;
 678         size_t count, i;
 679
 680         DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
 681                         work_item->id, get_trigger_name(work_item->trigger));
 682
 683         count = lttng_dynamic_array_get_count(&work_item->subitems);
 684         for (i = 0; i < count; i++) {
 685                 struct action_work_subitem *item;
 686
 687                 item = (action_work_subitem *) lttng_dynamic_array_get_element(&work_item->subitems, i);
 688                 ret = action_executor_generic_handler(
 689                                 executor, work_item, item);
 690                 if (ret) {
 691                         goto end;
 692                 }
 693         }
 694 end:
 695         DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
 696                         work_item->id, get_trigger_name(work_item->trigger));
 697         return ret;
 698 }
 699
 700 static void action_work_item_destroy(struct action_work_item *work_item)
 701 {
 702         lttng_trigger_put(work_item->trigger);
 703         lttng_evaluation_destroy(work_item->evaluation);
 704         notification_client_list_put(work_item->client_list);
 705         lttng_dynamic_array_reset(&work_item->subitems);
 706         free(work_item);
 707 }
 708
 709 static void *action_executor_thread(void *_data)
 710 {
 711         struct action_executor *executor = (action_executor *) _data;
 712
 713         LTTNG_ASSERT(executor);
 714
 715         health_register(the_health_sessiond,
 716                         HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
 717
 718         rcu_register_thread();
 719         rcu_thread_online();
 720
 721         DBG("Entering work execution loop");
 722         pthread_mutex_lock(&executor->work.lock);
 723         while (!executor->should_quit) {
 724                 int ret = 0;
 725                 struct action_work_item *work_item;
 726
 727                 health_code_update();
 728                 if (executor->work.pending_count == 0) {
 729                         health_poll_entry();
 730                         DBG("No work items enqueued, entering wait");
 731                         pthread_cond_wait(&executor->work.cond,
 732                                         &executor->work.lock);
 733                         DBG("Woke-up from wait");
 734                         health_poll_exit();
 735                         continue;
 736                 }
 737
 738                 /* Pop item from front of the list with work lock held. */
 739                 work_item = cds_list_first_entry(&executor->work.list,
 740                                 struct action_work_item, list_node);
 741                 cds_list_del(&work_item->list_node);
 742                 executor->work.pending_count--;
 743
 744                 /*
 745                  * Work can be performed without holding the work lock,
 746                  * allowing new items to be queued.
 747                  */
 748                 pthread_mutex_unlock(&executor->work.lock);
 749
 750                 /* Execute item only if a trigger is registered. */
 751                 lttng_trigger_lock(work_item->trigger);
 752                 if (!lttng_trigger_is_registered(work_item->trigger)) {
 753                         const char *trigger_name = NULL;
 754                         uid_t trigger_owner_uid;
 755                         enum lttng_trigger_status trigger_status;
 756
 757                         trigger_name = get_trigger_name(work_item->trigger);
 758
 759                         trigger_status = lttng_trigger_get_owner_uid(
 760                                         work_item->trigger, &trigger_owner_uid);
 761                         LTTNG_ASSERT(trigger_status == LTTNG_TRIGGER_STATUS_OK);
 762
 763                         DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64 ", trigger name = `%s`, trigger owner uid = %d",
 764                                         work_item->id, trigger_name,
 765                                         (int) trigger_owner_uid);
 766                         ret = 0;
 767                         goto skip_execute;
 768                 }
 769
 770                 ret = action_work_item_execute(executor, work_item);
 771
 772         skip_execute:
 773                 lttng_trigger_unlock(work_item->trigger);
 774                 action_work_item_destroy(work_item);
 775                 if (ret) {
 776                         /* Fatal error. */
 777                         break;
 778                 }
 779
 780                 health_code_update();
 781                 pthread_mutex_lock(&executor->work.lock);
 782         }
 783
 784         if (executor->should_quit) {
 785                 pthread_mutex_unlock(&executor->work.lock);
 786         }
 787         DBG("Left work execution loop");
 788
 789         health_code_update();
 790
 791         rcu_thread_offline();
 792         rcu_unregister_thread();
 793         health_unregister(the_health_sessiond);
 794
 795         return NULL;
 796 }
 797
 798 static bool shutdown_action_executor_thread(void *_data)
 799 {
 800         struct action_executor *executor = (action_executor *) _data;
 801
 802         pthread_mutex_lock(&executor->work.lock);
 803         executor->should_quit = true;
 804         pthread_cond_signal(&executor->work.cond);
 805         pthread_mutex_unlock(&executor->work.lock);
 806         return true;
 807 }
 808
 809 static void clean_up_action_executor_thread(void *_data)
 810 {
 811         struct action_executor *executor = (action_executor *) _data;
 812
 813         LTTNG_ASSERT(cds_list_empty(&executor->work.list));
 814
 815         pthread_mutex_destroy(&executor->work.lock);
 816         pthread_cond_destroy(&executor->work.cond);
 817         free(executor);
 818 }
 819
 820 struct action_executor *action_executor_create(
 821                 struct notification_thread_handle *handle)
 822 {
 823         struct action_executor *executor = (action_executor *) zmalloc(sizeof(*executor));
 824
 825         if (!executor) {
 826                 goto end;
 827         }
 828
 829         CDS_INIT_LIST_HEAD(&executor->work.list);
 830         pthread_cond_init(&executor->work.cond, NULL);
 831         pthread_mutex_init(&executor->work.lock, NULL);
 832         executor->notification_thread_handle = handle;
 833
 834         executor->thread = lttng_thread_create(THREAD_NAME,
 835                         action_executor_thread, shutdown_action_executor_thread,
 836                         clean_up_action_executor_thread, executor);
 837 end:
 838         return executor;
 839 }
 840
 841 void action_executor_destroy(struct action_executor *executor)
 842 {
 843         struct action_work_item *work_item, *tmp;
 844
 845         /* TODO Wait for work list to drain? */
 846         lttng_thread_shutdown(executor->thread);
 847         pthread_mutex_lock(&executor->work.lock);
 848         if (executor->work.pending_count != 0) {
 849                 WARN("%" PRIu64
 850                         " trigger action%s still queued for execution and will be discarded",
 851                                 executor->work.pending_count,
 852                                 executor->work.pending_count == 1 ? " is" :
 853                                                                     "s are");
 854         }
 855
 856         cds_list_for_each_entry_safe (
 857                         work_item, tmp, &executor->work.list, list_node) {
 858                 WARN("Discarding action work item %" PRIu64
 859                                 " associated to trigger `%s`",
 860                                 work_item->id, get_trigger_name(work_item->trigger));
 861                 cds_list_del(&work_item->list_node);
 862                 action_work_item_destroy(work_item);
 863         }
 864         pthread_mutex_unlock(&executor->work.lock);
 865         lttng_thread_put(executor->thread);
 866 }
 867
 868 /* RCU read-lock must be held by the caller. */
 869 enum action_executor_status action_executor_enqueue_trigger(
 870                 struct action_executor *executor,
 871                 struct lttng_trigger *trigger,
 872                 struct lttng_evaluation *evaluation,
 873                 const struct lttng_credentials *object_creds,
 874                 struct notification_client_list *client_list)
 875 {
 876         int ret;
 877         enum action_executor_status executor_status = ACTION_EXECUTOR_STATUS_OK;
 878         const uint64_t work_item_id = executor->next_work_item_id++;
 879         struct action_work_item *work_item;
 880         bool signal = false;
 881
 882         LTTNG_ASSERT(trigger);
 883         ASSERT_RCU_READ_LOCKED();
 884
 885         pthread_mutex_lock(&executor->work.lock);
 886         /* Check for queue overflow. */
 887         if (executor->work.pending_count >= MAX_QUEUED_WORK_COUNT) {
 888                 /* Most likely spammy, remove if it is the case. */
 889                 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64,
 890                                 get_trigger_name(trigger), work_item_id);
 891                 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
 892                 goto error_unlock;
 893         }
 894
 895         work_item = (action_work_item *) zmalloc(sizeof(*work_item));
 896         if (!work_item) {
 897                 PERROR("Failed to allocate action executor work item: trigger name = `%s`",
 898                                 get_trigger_name(trigger));
 899                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 900                 goto error_unlock;
 901         }
 902
 903         lttng_trigger_get(trigger);
 904         if (client_list) {
 905                 const bool reference_acquired =
 906                                 notification_client_list_get(client_list);
 907
 908                 LTTNG_ASSERT(reference_acquired);
 909         }
 910
 911         work_item->id = work_item_id;
 912         work_item->trigger = trigger;
 913
 914         /* Ownership transferred to the work item. */
 915         work_item->evaluation = evaluation;
 916         evaluation = NULL;
 917
 918         work_item->client_list = client_list;
 919         work_item->object_creds.is_set = !!object_creds;
 920         if (object_creds) {
 921                 work_item->object_creds.value = *object_creds;
 922         }
 923
 924         CDS_INIT_LIST_HEAD(&work_item->list_node);
 925
 926         /* Build the array of action work subitems for the passed trigger. */
 927         lttng_dynamic_array_init(&work_item->subitems,
 928                         sizeof(struct action_work_subitem),
 929                         action_work_subitem_destructor);
 930
 931         ret = populate_subitem_array_from_trigger(
 932                         trigger, &work_item->subitems);
 933         if (ret) {
 934                 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
 935                                 get_trigger_name(trigger));
 936                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 937                 goto error_unlock;
 938         }
 939
 940         cds_list_add_tail(&work_item->list_node, &executor->work.list);
 941         executor->work.pending_count++;
 942         DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64,
 943                         get_trigger_name(trigger), work_item_id);
 944         signal = true;
 945
 946 error_unlock:
 947         if (signal) {
 948                 pthread_cond_signal(&executor->work.cond);
 949         }
 950
 951         pthread_mutex_unlock(&executor->work.lock);
 952         lttng_evaluation_destroy(evaluation);
 953         return executor_status;
 954 }
 955
 956 static int add_action_to_subitem_array(struct lttng_action *action,
 957                 struct lttng_dynamic_array *subitems)
 958 {
 959         int ret = 0;
 960         enum lttng_action_type type = lttng_action_get_type(action);
 961         const char *session_name = NULL;
 962         enum lttng_action_status status;
 963         struct action_work_subitem subitem = {
 964                 .action = NULL,
 965                 .context = {
 966                         .session_id = LTTNG_OPTIONAL_INIT_UNSET,
 967                 },
 968         };
 969
 970         LTTNG_ASSERT(action);
 971         LTTNG_ASSERT(subitems);
 972
 973         if (type == LTTNG_ACTION_TYPE_LIST) {
 974                 unsigned int count, i;
 975
 976                 status = lttng_action_list_get_count(action, &count);
 977                 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
 978
 979                 for (i = 0; i < count; i++) {
 980                         struct lttng_action *inner_action = NULL;
 981
 982                         inner_action = lttng_action_list_borrow_mutable_at_index(
 983                                         action, i);
 984                         LTTNG_ASSERT(inner_action);
 985                         ret = add_action_to_subitem_array(
 986                                         inner_action, subitems);
 987                         if (ret) {
 988                                 goto end;
 989                         }
 990                 }
 991
 992                 /*
 993                  * Go directly to the end since there is no need to add the
 994                  * list action by itself to the subitems array.
 995                  */
 996                 goto end;
 997         }
 998
 999         /* Gather execution context. */
1000         switch (type) {
1001         case LTTNG_ACTION_TYPE_NOTIFY:
1002                 break;
1003         case LTTNG_ACTION_TYPE_START_SESSION:
1004                 status = lttng_action_start_session_get_session_name(
1005                                 action, &session_name);
1006                 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1007                 break;
1008         case LTTNG_ACTION_TYPE_STOP_SESSION:
1009                 status = lttng_action_stop_session_get_session_name(
1010                                 action, &session_name);
1011                 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1012                 break;
1013         case LTTNG_ACTION_TYPE_ROTATE_SESSION:
1014                 status = lttng_action_rotate_session_get_session_name(
1015                                 action, &session_name);
1016                 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1017                 break;
1018         case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION:
1019                 status = lttng_action_snapshot_session_get_session_name(
1020                                 action, &session_name);
1021                 LTTNG_ASSERT(status == LTTNG_ACTION_STATUS_OK);
1022                 break;
1023         case LTTNG_ACTION_TYPE_LIST:
1024         case LTTNG_ACTION_TYPE_UNKNOWN:
1025                 /* Fallthrough */
1026         default:
1027                 abort();
1028                 break;
1029         }
1030
1031         /*
1032          * Fetch the session execution context info as needed.
1033          * Note that we could decide to not add an action for which we know the
1034          * execution will not happen (i.e no session exists for that name). For
1035          * now we leave the decision to skip to the action executor for sake of
1036          * simplicity and consistency.
1037          */
1038         if (session_name != NULL) {
1039                 uint64_t session_id;
1040
1041                 /*
1042                  * Instantaneous sampling of the session id if present.
1043                  *
1044                  * This method is preferred over `sessiond_find_by_name` then
1045                  * fetching the session'd id since `sessiond_find_by_name`
1046                  * requires the session list lock to be taken.
1047                  *
1048                  * Taking the session list lock can lead to a deadlock
1049                  * between the action executor and the notification thread
1050                  * (caller of add_action_to_subitem_array). It is okay if the
1051                  * session state changes between the enqueuing time and the
1052                  * execution time. The execution context is validated at
1053                  * execution time.
1054                  */
1055                 if (sample_session_id_by_name(session_name, &session_id)) {
1056                         LTTNG_OPTIONAL_SET(&subitem.context.session_id,
1057                                         session_id);
1058                 }
1059         }
1060
1061         /* Get a reference to the action. */
1062         lttng_action_get(action);
1063         subitem.action = action;
1064
1065         ret = lttng_dynamic_array_add_element(subitems, &subitem);
1066         if (ret) {
1067                 ERR("Failed to add work subitem to the subitem array");
1068                 lttng_action_put(action);
1069                 ret = -1;
1070                 goto end;
1071         }
1072
1073 end:
1074         return ret;
1075 }
1076
1077 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
1078                 struct lttng_dynamic_array *subitems)
1079 {
1080         struct lttng_action *action;
1081
1082         action = lttng_trigger_get_action(trigger);
1083         LTTNG_ASSERT(action);
1084
1085         return add_action_to_subitem_array(action, subitems);
1086 }