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

   1 /*
   2  * Copyright (C) 2020 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * SPDX-License-Identifier: GPL-2.0-only
   5  *
   6  */
   7
   8 #include "action-executor.h"
   9 #include "cmd.h"
  10 #include "health-sessiond.h"
  11 #include "lttng-sessiond.h"
  12 #include "notification-thread-internal.h"
  13 #include "session.h"
  14 #include "thread.h"
  15 #include <common/dynamic-array.h>
  16 #include <common/macros.h>
  17 #include <common/optional.h>
  18 #include <lttng/action/action-internal.h>
  19 #include <lttng/action/group-internal.h>
  20 #include <lttng/action/group.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/on-event-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_group_handler(struct action_executor *executor,
 139                 const struct action_work_item *,
 140                 struct action_work_subitem *);
 141 static int action_executor_generic_handler(struct action_executor *executor,
 142                 const struct action_work_item *,
 143                 struct action_work_subitem *);
 144
 145 static const action_executor_handler action_executors[] = {
 146         [LTTNG_ACTION_TYPE_NOTIFY] = action_executor_notify_handler,
 147         [LTTNG_ACTION_TYPE_START_SESSION] = action_executor_start_session_handler,
 148         [LTTNG_ACTION_TYPE_STOP_SESSION] = action_executor_stop_session_handler,
 149         [LTTNG_ACTION_TYPE_ROTATE_SESSION] = action_executor_rotate_session_handler,
 150         [LTTNG_ACTION_TYPE_SNAPSHOT_SESSION] = action_executor_snapshot_session_handler,
 151         [LTTNG_ACTION_TYPE_GROUP] = action_executor_group_handler,
 152 };
 153
 154 /* Forward declaration */
 155 static int add_action_to_subitem_array(struct lttng_action *action,
 156                 struct lttng_dynamic_array *subitems);
 157
 158 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
 159                 struct lttng_dynamic_array *subitems);
 160
 161 static void action_work_subitem_destructor(void *element)
 162 {
 163         struct action_work_subitem *subitem = element;
 164
 165         lttng_action_put(subitem->action);
 166 }
 167
 168 static const char *get_action_name(const struct lttng_action *action)
 169 {
 170         const enum lttng_action_type action_type = lttng_action_get_type(action);
 171
 172         assert(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
 173
 174         return lttng_action_type_string(action_type);
 175 }
 176
 177 /* Check if this trigger allowed to interect with a given session. */
 178 static bool is_trigger_allowed_for_session(const struct lttng_trigger *trigger,
 179                 struct ltt_session *session)
 180 {
 181         bool is_allowed = false;
 182         const struct lttng_credentials session_creds = {
 183                 .uid = LTTNG_OPTIONAL_INIT_VALUE(session->uid),
 184                 .gid = LTTNG_OPTIONAL_INIT_VALUE(session->gid),
 185         };
 186         /* Can never be NULL. */
 187         const struct lttng_credentials *trigger_creds =
 188                         lttng_trigger_get_credentials(trigger);
 189
 190         is_allowed = (lttng_credentials_is_equal_uid(trigger_creds, &session_creds)) ||
 191                         (lttng_credentials_get_uid(trigger_creds) == 0);
 192         if (!is_allowed) {
 193                 WARN("Trigger is not allowed to interact with session `%s`: session uid = %ld, session gid = %ld, trigger uid = %ld",
 194                                 session->name,
 195                                 (long int) session->uid,
 196                                 (long int) session->gid,
 197                                 (long int) lttng_credentials_get_uid(trigger_creds));
 198         }
 199
 200         return is_allowed;
 201 }
 202
 203 static const char *get_trigger_name(const struct lttng_trigger *trigger)
 204 {
 205         const char *trigger_name;
 206         enum lttng_trigger_status trigger_status;
 207
 208         trigger_status = lttng_trigger_get_name(trigger, &trigger_name);
 209         assert(trigger_status == LTTNG_TRIGGER_STATUS_OK);
 210
 211         return trigger_name;
 212 }
 213
 214 static int client_handle_transmission_status(
 215                 struct notification_client *client,
 216                 enum client_transmission_status status,
 217                 void *user_data)
 218 {
 219         int ret = 0;
 220         struct action_executor *executor = user_data;
 221         bool update_communication = true;
 222
 223         switch (status) {
 224         case CLIENT_TRANSMISSION_STATUS_COMPLETE:
 225                 DBG("Successfully sent full notification to client, client_id = %" PRIu64,
 226                                 client->id);
 227                 update_communication = false;
 228                 break;
 229         case CLIENT_TRANSMISSION_STATUS_QUEUED:
 230                 DBG("Queued notification in client outgoing buffer, client_id = %" PRIu64,
 231                                 client->id);
 232                 break;
 233         case CLIENT_TRANSMISSION_STATUS_FAIL:
 234                 DBG("Communication error occurred while sending notification to client, client_id = %" PRIu64,
 235                                 client->id);
 236                 break;
 237         default:
 238                 ERR("Fatal error encoutered while sending notification to client, client_id = %" PRIu64,
 239                                 client->id);
 240                 ret = -1;
 241                 goto end;
 242         }
 243
 244         if (!update_communication) {
 245                 goto end;
 246         }
 247
 248         /* Safe to read client's id without locking as it is immutable. */
 249         ret = notification_thread_client_communication_update(
 250                         executor->notification_thread_handle, client->id,
 251                         status);
 252 end:
 253         return ret;
 254 }
 255
 256 static int action_executor_notify_handler(struct action_executor *executor,
 257                 const struct action_work_item *work_item,
 258                 struct action_work_subitem *item)
 259 {
 260         return notification_client_list_send_evaluation(work_item->client_list,
 261                         work_item->trigger,
 262                         work_item->evaluation,
 263                         work_item->object_creds.is_set ?
 264                                         &(work_item->object_creds.value) :
 265                                         NULL,
 266                         client_handle_transmission_status, executor);
 267 }
 268
 269 static int action_executor_start_session_handler(
 270                 struct action_executor *executor,
 271                 const struct action_work_item *work_item,
 272                 struct action_work_subitem *item)
 273 {
 274         int ret = 0;
 275         const char *session_name;
 276         enum lttng_action_status action_status;
 277         struct ltt_session *session;
 278         enum lttng_error_code cmd_ret;
 279         struct lttng_action *action = item->action;
 280
 281         action_status = lttng_action_start_session_get_session_name(
 282                         action, &session_name);
 283         if (action_status != LTTNG_ACTION_STATUS_OK) {
 284                 ERR("Failed to get session name from `%s` action",
 285                                 get_action_name(action));
 286                 ret = -1;
 287                 goto end;
 288         }
 289
 290         /*
 291          * Validate if at the moment of the action was queued the session
 292          * existed. If not skip the action altogether.
 293          */
 294         if (!item->context.session_id.is_set) {
 295                 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
 296                                 session_name, get_action_name(action),
 297                                 get_trigger_name(work_item->trigger));
 298                 lttng_action_increase_execution_failure_count(action);
 299                 ret = 0;
 300                 goto end;
 301         }
 302
 303         session_lock_list();
 304         session = session_find_by_name(session_name);
 305         if (!session) {
 306                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 307                                 session_name, get_action_name(action),
 308                                 get_trigger_name(work_item->trigger));
 309                 goto error_unlock_list;
 310         }
 311
 312         /*
 313          * Check if the session id is the same as when the work item was
 314          * enqueued.
 315          */
 316         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 317                 DBG("Session id for session `%s` (id: %" PRIu64
 318                                 " is not the same that was sampled (id: %" PRIu64
 319                                 " at the moment the work item was enqueued for %s` action of trigger `%s`",
 320                                 session_name, session->id,
 321                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 322                                 get_action_name(action),
 323                                 get_trigger_name(work_item->trigger));
 324                 ret = 0;
 325                 goto error_unlock_list;
 326         }
 327
 328         session_lock(session);
 329         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 330                 goto error_dispose_session;
 331         }
 332
 333         cmd_ret = cmd_start_trace(session);
 334         switch (cmd_ret) {
 335         case LTTNG_OK:
 336                 DBG("Successfully started session `%s` on behalf of trigger `%s`",
 337                                 session_name, get_trigger_name(work_item->trigger));
 338                 break;
 339         case LTTNG_ERR_TRACE_ALREADY_STARTED:
 340                 DBG("Attempted to start session `%s` on behalf of trigger `%s` but it was already started",
 341                                 session_name, get_trigger_name(work_item->trigger));
 342                 break;
 343         default:
 344                 WARN("Failed to start session `%s` on behalf of trigger `%s`: %s",
 345                                 session_name, get_trigger_name(work_item->trigger),
 346                                 lttng_strerror(-cmd_ret));
 347                 lttng_action_increase_execution_failure_count(action);
 348                 break;
 349         }
 350
 351 error_dispose_session:
 352         session_unlock(session);
 353         session_put(session);
 354 error_unlock_list:
 355         session_unlock_list();
 356 end:
 357         return ret;
 358 }
 359
 360 static int action_executor_stop_session_handler(
 361                 struct action_executor *executor,
 362                 const struct action_work_item *work_item,
 363                 struct action_work_subitem *item)
 364 {
 365         int ret = 0;
 366         const char *session_name;
 367         enum lttng_action_status action_status;
 368         struct ltt_session *session;
 369         enum lttng_error_code cmd_ret;
 370         struct lttng_action *action = item->action;
 371
 372         action_status = lttng_action_stop_session_get_session_name(
 373                         action, &session_name);
 374         if (action_status != LTTNG_ACTION_STATUS_OK) {
 375                 ERR("Failed to get session name from `%s` action",
 376                                 get_action_name(action));
 377                 ret = -1;
 378                 goto end;
 379         }
 380
 381         /*
 382          * Validate if, at the moment the action was queued, the target session
 383          * existed. If not, skip the action altogether.
 384          */
 385         if (!item->context.session_id.is_set) {
 386                 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
 387                                 session_name, get_action_name(action),
 388                                 get_trigger_name(work_item->trigger));
 389                 lttng_action_increase_execution_failure_count(action);
 390                 ret = 0;
 391                 goto end;
 392         }
 393
 394         session_lock_list();
 395         session = session_find_by_name(session_name);
 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         /*
 405          * Check if the session id is the same as when the work item was
 406          * enqueued
 407          */
 408         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 409                 DBG("Session id for session `%s` (id: %" PRIu64
 410                                 " is not the same that was sampled (id: %" PRIu64
 411                                 " at the moment the work item was enqueued for %s` action of trigger `%s`",
 412                                 session_name, session->id,
 413                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 414                                 get_action_name(action),
 415                                 get_trigger_name(work_item->trigger));
 416                 ret = 0;
 417                 goto error_unlock_list;
 418         }
 419
 420         session_lock(session);
 421         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 422                 goto error_dispose_session;
 423         }
 424
 425         cmd_ret = cmd_stop_trace(session);
 426         switch (cmd_ret) {
 427         case LTTNG_OK:
 428                 DBG("Successfully stopped session `%s` on behalf of trigger `%s`",
 429                                 session_name, get_trigger_name(work_item->trigger));
 430                 break;
 431         case LTTNG_ERR_TRACE_ALREADY_STOPPED:
 432                 DBG("Attempted to stop session `%s` on behalf of trigger `%s` but it was already stopped",
 433                                 session_name, get_trigger_name(work_item->trigger));
 434                 break;
 435         default:
 436                 WARN("Failed to stop session `%s` on behalf of trigger `%s`: %s",
 437                                 session_name, get_trigger_name(work_item->trigger),
 438                                 lttng_strerror(-cmd_ret));
 439                 lttng_action_increase_execution_failure_count(action);
 440                 break;
 441         }
 442
 443 error_dispose_session:
 444         session_unlock(session);
 445         session_put(session);
 446 error_unlock_list:
 447         session_unlock_list();
 448 end:
 449         return ret;
 450 }
 451
 452 static int action_executor_rotate_session_handler(
 453                 struct action_executor *executor,
 454                 const struct action_work_item *work_item,
 455                 struct action_work_subitem *item)
 456 {
 457         int ret = 0;
 458         const char *session_name;
 459         enum lttng_action_status action_status;
 460         struct ltt_session *session;
 461         enum lttng_error_code cmd_ret;
 462         struct lttng_action *action = item->action;
 463
 464         action_status = lttng_action_rotate_session_get_session_name(
 465                         action, &session_name);
 466         if (action_status != LTTNG_ACTION_STATUS_OK) {
 467                 ERR("Failed to get session name from `%s` action",
 468                                 get_action_name(action));
 469                 ret = -1;
 470                 goto end;
 471         }
 472
 473         /*
 474          * Validate if, at the moment the action was queued, the target session
 475          * existed. If not, skip the action altogether.
 476          */
 477         if (!item->context.session_id.is_set) {
 478                 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
 479                                 session_name, get_action_name(action),
 480                                 get_trigger_name(work_item->trigger));
 481                 lttng_action_increase_execution_failure_count(action);
 482                 ret = 0;
 483                 goto end;
 484         }
 485
 486         session_lock_list();
 487         session = session_find_by_name(session_name);
 488         if (!session) {
 489                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 490                                 session_name, get_action_name(action),
 491                                 get_trigger_name(work_item->trigger));
 492                 lttng_action_increase_execution_failure_count(action);
 493                 goto error_unlock_list;
 494         }
 495
 496         /*
 497          * Check if the session id is the same as when the work item was
 498          * enqueued.
 499          */
 500         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 501                 DBG("Session id for session `%s` (id: %" PRIu64
 502                     " is not the same that was sampled (id: %" PRIu64
 503                     " at the moment the work item was enqueued for %s` action of trigger `%s`",
 504                                 session_name, session->id,
 505                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 506                                 get_action_name(action),
 507                                 get_trigger_name(work_item->trigger));
 508                 ret = 0;
 509                 goto error_unlock_list;
 510         }
 511
 512         session_lock(session);
 513         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 514                 goto error_dispose_session;
 515         }
 516
 517         cmd_ret = cmd_rotate_session(session, NULL, false,
 518                         LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 519         switch (cmd_ret) {
 520         case LTTNG_OK:
 521                 DBG("Successfully started rotation of session `%s` on behalf of trigger `%s`",
 522                                 session_name, get_trigger_name(work_item->trigger));
 523                 break;
 524         case LTTNG_ERR_ROTATION_PENDING:
 525                 DBG("Attempted to start a rotation of session `%s` on behalf of trigger `%s` but a rotation is already ongoing",
 526                                 session_name, get_trigger_name(work_item->trigger));
 527                 lttng_action_increase_execution_failure_count(action);
 528                 break;
 529         case LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
 530         case LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
 531                 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",
 532                                 session_name, get_trigger_name(work_item->trigger));
 533                 break;
 534         default:
 535                 WARN("Failed to start a rotation of session `%s` on behalf of trigger `%s`: %s",
 536                                 session_name, get_trigger_name(work_item->trigger),
 537                                 lttng_strerror(-cmd_ret));
 538                 lttng_action_increase_execution_failure_count(action);
 539                 break;
 540         }
 541
 542 error_dispose_session:
 543         session_unlock(session);
 544         session_put(session);
 545 error_unlock_list:
 546         session_unlock_list();
 547 end:
 548         return ret;
 549 }
 550
 551 static int action_executor_snapshot_session_handler(
 552                 struct action_executor *executor,
 553                 const struct action_work_item *work_item,
 554                 struct action_work_subitem *item)
 555 {
 556         int ret = 0;
 557         const char *session_name;
 558         enum lttng_action_status action_status;
 559         struct ltt_session *session;
 560         const struct lttng_snapshot_output default_snapshot_output = {
 561                 .max_size = UINT64_MAX,
 562         };
 563         const struct lttng_snapshot_output *snapshot_output =
 564                         &default_snapshot_output;
 565         enum lttng_error_code cmd_ret;
 566         struct lttng_action *action = item->action;
 567
 568         /*
 569          * Validate if, at the moment the action was queued, the target session
 570          * existed. If not, skip the action altogether.
 571          */
 572         if (!item->context.session_id.is_set) {
 573                 DBG("Session `%s` was not present at the moment the work item was enqueued for %s` action of trigger `%s`",
 574                                 session_name, get_action_name(action),
 575                                 get_trigger_name(work_item->trigger));
 576                 lttng_action_increase_execution_failure_count(action);
 577                 ret = 0;
 578                 goto end;
 579         }
 580
 581         action_status = lttng_action_snapshot_session_get_session_name(
 582                         action, &session_name);
 583         if (action_status != LTTNG_ACTION_STATUS_OK) {
 584                 ERR("Failed to get session name from `%s` action",
 585                                 get_action_name(action));
 586                 ret = -1;
 587                 goto end;
 588         }
 589
 590         action_status = lttng_action_snapshot_session_get_output(
 591                         action, &snapshot_output);
 592         if (action_status != LTTNG_ACTION_STATUS_OK &&
 593                         action_status != LTTNG_ACTION_STATUS_UNSET) {
 594                 ERR("Failed to get output from `%s` action",
 595                                 get_action_name(action));
 596                 ret = -1;
 597                 goto end;
 598         }
 599
 600         session_lock_list();
 601         session = session_find_by_name(session_name);
 602         if (!session) {
 603                 DBG("Failed to find session `%s` by name while executing `%s` action of trigger `%s`",
 604                                 session_name, get_action_name(action),
 605                                 get_trigger_name(work_item->trigger));
 606                 lttng_action_increase_execution_failure_count(action);
 607                 goto error_unlock_list;
 608         }
 609
 610         /*
 611          * Check if the session id is the same as when the work item was
 612          * enqueued.
 613          */
 614         if (session->id != LTTNG_OPTIONAL_GET(item->context.session_id)) {
 615                 DBG("Session id for session `%s` (id: %" PRIu64
 616                     " is not the same that was sampled (id: %" PRIu64
 617                     " at the moment the work item was enqueued for %s` action of trigger `%s`",
 618                                 session_name, session->id,
 619                                 LTTNG_OPTIONAL_GET(item->context.session_id),
 620                                 get_action_name(action),
 621                                 get_trigger_name(work_item->trigger));
 622                 ret = 0;
 623                 goto error_unlock_list;
 624         }
 625
 626         session_lock(session);
 627         if (!is_trigger_allowed_for_session(work_item->trigger, session)) {
 628                 goto error_dispose_session;
 629         }
 630
 631         cmd_ret = cmd_snapshot_record(session, snapshot_output, 0);
 632         switch (cmd_ret) {
 633         case LTTNG_OK:
 634                 DBG("Successfully recorded snapshot of session `%s` on behalf of trigger `%s`",
 635                                 session_name, get_trigger_name(work_item->trigger));
 636                 break;
 637         default:
 638                 WARN("Failed to record snapshot of session `%s` on behalf of trigger `%s`: %s",
 639                                 session_name, get_trigger_name(work_item->trigger),
 640                                 lttng_strerror(-cmd_ret));
 641                 lttng_action_increase_execution_failure_count(action);
 642                 break;
 643         }
 644
 645 error_dispose_session:
 646         session_unlock(session);
 647         session_put(session);
 648 error_unlock_list:
 649         session_unlock_list();
 650 end:
 651         return ret;
 652 }
 653
 654 static int action_executor_group_handler(struct action_executor *executor,
 655                 const struct action_work_item *work_item,
 656                 struct action_work_subitem *item)
 657 {
 658         ERR("Execution of a group action by the action executor should never occur");
 659         abort();
 660 }
 661
 662 static int action_executor_generic_handler(struct action_executor *executor,
 663                 const struct action_work_item *work_item,
 664                 struct action_work_subitem *item)
 665 {
 666         int ret;
 667         struct lttng_action *action = item->action;
 668         const enum lttng_action_type action_type = lttng_action_get_type(action);
 669
 670         assert(action_type != LTTNG_ACTION_TYPE_UNKNOWN);
 671
 672         lttng_action_increase_execution_request_count(action);
 673         if (!lttng_action_should_execute(action)) {
 674                 DBG("Policy prevented execution of action `%s` of trigger `%s` action work item %" PRIu64,
 675                                 get_action_name(action),
 676                                 get_trigger_name(work_item->trigger),
 677                                 work_item->id);
 678                 ret = 0;
 679                 goto end;
 680         }
 681
 682         lttng_action_increase_execution_count(action);
 683         DBG("Executing action `%s` of trigger `%s` action work item %" PRIu64,
 684                         get_action_name(action),
 685                         get_trigger_name(work_item->trigger),
 686                         work_item->id);
 687         ret = action_executors[action_type](executor, work_item, item);
 688 end:
 689         return ret;
 690 }
 691
 692 static int action_work_item_execute(struct action_executor *executor,
 693                 struct action_work_item *work_item)
 694 {
 695         int ret;
 696         size_t count, i;
 697
 698         DBG("Starting execution of action work item %" PRIu64 " of trigger `%s`",
 699                         work_item->id, get_trigger_name(work_item->trigger));
 700
 701         count = lttng_dynamic_array_get_count(work_item->subitems);
 702         for (i = 0; i < count; i++) {
 703                 struct action_work_subitem *item;
 704
 705                 item = lttng_dynamic_array_get_element(work_item->subitems, i);
 706                 ret = action_executor_generic_handler(
 707                                 executor, work_item, item);
 708                 if (ret) {
 709                         goto end;
 710                 }
 711         }
 712 end:
 713         DBG("Completed execution of action work item %" PRIu64 " of trigger `%s`",
 714                         work_item->id, get_trigger_name(work_item->trigger));
 715         return ret;
 716 }
 717
 718 static void action_work_item_destroy(struct action_work_item *work_item)
 719 {
 720         lttng_trigger_put(work_item->trigger);
 721         lttng_evaluation_destroy(work_item->evaluation);
 722         notification_client_list_put(work_item->client_list);
 723         lttng_dynamic_array_reset(work_item->subitems);
 724         free(work_item);
 725 }
 726
 727 static void *action_executor_thread(void *_data)
 728 {
 729         struct action_executor *executor = _data;
 730
 731         assert(executor);
 732
 733         health_register(the_health_sessiond,
 734                         HEALTH_SESSIOND_TYPE_ACTION_EXECUTOR);
 735
 736         rcu_register_thread();
 737         rcu_thread_online();
 738
 739         DBG("Entering work execution loop");
 740         pthread_mutex_lock(&executor->work.lock);
 741         while (!executor->should_quit) {
 742                 int ret;
 743                 struct action_work_item *work_item;
 744
 745                 health_code_update();
 746                 if (executor->work.pending_count == 0) {
 747                         health_poll_entry();
 748                         DBG("No work items enqueued, entering wait");
 749                         pthread_cond_wait(&executor->work.cond,
 750                                         &executor->work.lock);
 751                         DBG("Woke-up from wait");
 752                         health_poll_exit();
 753                         continue;
 754                 }
 755
 756                 /* Pop item from front of the list with work lock held. */
 757                 work_item = cds_list_first_entry(&executor->work.list,
 758                                 struct action_work_item, list_node);
 759                 cds_list_del(&work_item->list_node);
 760                 executor->work.pending_count--;
 761
 762                 /*
 763                  * Work can be performed without holding the work lock,
 764                  * allowing new items to be queued.
 765                  */
 766                 pthread_mutex_unlock(&executor->work.lock);
 767
 768                 /* Execute item only if a trigger is registered. */
 769                 lttng_trigger_lock(work_item->trigger);
 770                 if (!lttng_trigger_is_registered(work_item->trigger)) {
 771                         const char *trigger_name = NULL;
 772                         uid_t trigger_owner_uid;
 773                         enum lttng_trigger_status trigger_status;
 774
 775                         trigger_status = lttng_trigger_get_name(
 776                                         work_item->trigger, &trigger_name);
 777                         switch (trigger_status) {
 778                         case LTTNG_TRIGGER_STATUS_OK:
 779                                 break;
 780                         case LTTNG_TRIGGER_STATUS_UNSET:
 781                                 trigger_name = "(unset)";
 782                                 break;
 783                         default:
 784                                 abort();
 785                         }
 786
 787                         trigger_status = lttng_trigger_get_owner_uid(
 788                                         work_item->trigger, &trigger_owner_uid);
 789                         assert(trigger_status == LTTNG_TRIGGER_STATUS_OK);
 790
 791                         DBG("Work item skipped since the associated trigger is no longer registered: work item id = %" PRIu64 ", trigger name = '%s', trigger owner uid = %d",
 792                                         work_item->id, trigger_name,
 793                                         (int) trigger_owner_uid);
 794                         ret = 0;
 795                         goto skip_execute;
 796                 }
 797
 798                 ret = action_work_item_execute(executor, work_item);
 799
 800         skip_execute:
 801                 lttng_trigger_unlock(work_item->trigger);
 802                 action_work_item_destroy(work_item);
 803                 if (ret) {
 804                         /* Fatal error. */
 805                         break;
 806                 }
 807
 808                 health_code_update();
 809                 pthread_mutex_lock(&executor->work.lock);
 810         }
 811
 812         if (executor->should_quit) {
 813                 pthread_mutex_unlock(&executor->work.lock);
 814         }
 815         DBG("Left work execution loop");
 816
 817         health_code_update();
 818
 819         rcu_thread_offline();
 820         rcu_unregister_thread();
 821         health_unregister(the_health_sessiond);
 822
 823         return NULL;
 824 }
 825
 826 static bool shutdown_action_executor_thread(void *_data)
 827 {
 828         struct action_executor *executor = _data;
 829
 830         pthread_mutex_lock(&executor->work.lock);
 831         executor->should_quit = true;
 832         pthread_cond_signal(&executor->work.cond);
 833         pthread_mutex_unlock(&executor->work.lock);
 834         return true;
 835 }
 836
 837 static void clean_up_action_executor_thread(void *_data)
 838 {
 839         struct action_executor *executor = _data;
 840
 841         assert(cds_list_empty(&executor->work.list));
 842
 843         pthread_mutex_destroy(&executor->work.lock);
 844         pthread_cond_destroy(&executor->work.cond);
 845         free(executor);
 846 }
 847
 848 struct action_executor *action_executor_create(
 849                 struct notification_thread_handle *handle)
 850 {
 851         struct action_executor *executor = zmalloc(sizeof(*executor));
 852
 853         if (!executor) {
 854                 goto end;
 855         }
 856
 857         CDS_INIT_LIST_HEAD(&executor->work.list);
 858         pthread_cond_init(&executor->work.cond, NULL);
 859         pthread_mutex_init(&executor->work.lock, NULL);
 860         executor->notification_thread_handle = handle;
 861
 862         executor->thread = lttng_thread_create(THREAD_NAME,
 863                         action_executor_thread, shutdown_action_executor_thread,
 864                         clean_up_action_executor_thread, executor);
 865 end:
 866         return executor;
 867 }
 868
 869 void action_executor_destroy(struct action_executor *executor)
 870 {
 871         struct action_work_item *work_item, *tmp;
 872
 873         /* TODO Wait for work list to drain? */
 874         lttng_thread_shutdown(executor->thread);
 875         pthread_mutex_lock(&executor->work.lock);
 876         if (executor->work.pending_count != 0) {
 877                 WARN("%" PRIu64
 878                         " trigger action%s still queued for execution and will be discarded",
 879                                 executor->work.pending_count,
 880                                 executor->work.pending_count == 1 ? " is" :
 881                                                                     "s are");
 882         }
 883
 884         cds_list_for_each_entry_safe (
 885                         work_item, tmp, &executor->work.list, list_node) {
 886                 WARN("Discarding action work item %" PRIu64
 887                                 " associated to trigger `%s`",
 888                                 work_item->id, get_trigger_name(work_item->trigger));
 889                 cds_list_del(&work_item->list_node);
 890                 action_work_item_destroy(work_item);
 891         }
 892         pthread_mutex_unlock(&executor->work.lock);
 893         lttng_thread_put(executor->thread);
 894 }
 895
 896 /* RCU read-lock must be held by the caller. */
 897 enum action_executor_status action_executor_enqueue_trigger(
 898                 struct action_executor *executor,
 899                 struct lttng_trigger *trigger,
 900                 struct lttng_evaluation *evaluation,
 901                 const struct lttng_credentials *object_creds,
 902                 struct notification_client_list *client_list)
 903 {
 904         int ret;
 905         enum action_executor_status executor_status = ACTION_EXECUTOR_STATUS_OK;
 906         const uint64_t work_item_id = executor->next_work_item_id++;
 907         struct action_work_item *work_item;
 908         bool signal = false;
 909         struct lttng_dynamic_array *subitems = NULL;
 910
 911         assert(trigger);
 912
 913         /* Build the array of action work subitems for the passed trigger. */
 914         subitems = zmalloc(sizeof(*subitems));
 915         if (!subitems) {
 916                 PERROR("Failed to allocate action executor subitems array: trigger name = `%s`",
 917                                 get_trigger_name(trigger));
 918                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 919                 goto error_unlock;
 920         }
 921
 922         lttng_dynamic_array_init(subitems, sizeof(struct action_work_subitem),
 923                         action_work_subitem_destructor);
 924
 925         ret = populate_subitem_array_from_trigger(trigger, subitems);
 926         if (ret) {
 927                 ERR("Failed to populate work item sub items on behalf of trigger: trigger name = `%s`",
 928                                 get_trigger_name(trigger));
 929                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 930                 goto error_unlock;
 931         }
 932
 933         pthread_mutex_lock(&executor->work.lock);
 934         /* Check for queue overflow. */
 935         if (executor->work.pending_count >= MAX_QUEUED_WORK_COUNT) {
 936                 /* Most likely spammy, remove if it is the case. */
 937                 DBG("Refusing to enqueue action for trigger (overflow): trigger name = `%s`, work item id = %" PRIu64,
 938                                 get_trigger_name(trigger), work_item_id);
 939                 executor_status = ACTION_EXECUTOR_STATUS_OVERFLOW;
 940                 goto error_unlock;
 941         }
 942
 943         work_item = zmalloc(sizeof(*work_item));
 944         if (!work_item) {
 945                 PERROR("Failed to allocate action executor work item: trigger name = '%s'",
 946                                 get_trigger_name(trigger));
 947                 executor_status = ACTION_EXECUTOR_STATUS_ERROR;
 948                 goto error_unlock;
 949         }
 950
 951         lttng_trigger_get(trigger);
 952         if (client_list) {
 953                 const bool reference_acquired =
 954                                 notification_client_list_get(client_list);
 955
 956                 assert(reference_acquired);
 957         }
 958
 959         *work_item = (typeof(*work_item)){
 960                         .id = work_item_id,
 961                         /* Ownership transferred to the work item. */
 962                         .subitems = subitems,
 963                         .trigger = trigger,
 964                         /* Ownership transferred to the work item. */
 965                         .evaluation = evaluation,
 966                         .object_creds = {
 967                                 .is_set = !!object_creds,
 968                                 .value = object_creds ? *object_creds :
 969                                         (typeof(work_item->object_creds.value)) {},
 970                         },
 971                         .client_list = client_list,
 972                         .list_node = CDS_LIST_HEAD_INIT(work_item->list_node),
 973         };
 974
 975         evaluation = NULL;
 976         subitems = NULL;
 977         cds_list_add_tail(&work_item->list_node, &executor->work.list);
 978         executor->work.pending_count++;
 979         DBG("Enqueued action for trigger: trigger name = `%s`, work item id = %" PRIu64,
 980                         get_trigger_name(trigger), work_item_id);
 981         signal = true;
 982
 983 error_unlock:
 984         if (signal) {
 985                 pthread_cond_signal(&executor->work.cond);
 986         }
 987         pthread_mutex_unlock(&executor->work.lock);
 988
 989         lttng_evaluation_destroy(evaluation);
 990         if (subitems) {
 991                 lttng_dynamic_array_reset(subitems);
 992                 free(subitems);
 993         }
 994         return executor_status;
 995 }
 996
 997 static int add_action_to_subitem_array(struct lttng_action *action,
 998                 struct lttng_dynamic_array *subitems)
 999 {
1000         int ret;
1001         enum lttng_action_type type = lttng_action_get_type(action);
1002         const char *session_name = NULL;
1003         enum lttng_action_status status;
1004         struct action_work_subitem subitem = {
1005                 .action = NULL,
1006                 .context = {
1007                         .session_id = LTTNG_OPTIONAL_INIT_UNSET,
1008                 },
1009         };
1010
1011         assert(action);
1012         assert(subitems);
1013
1014         if (type == LTTNG_ACTION_TYPE_GROUP) {
1015                 unsigned int count, i;
1016
1017                 status = lttng_action_group_get_count(action, &count);
1018                 assert(status == LTTNG_ACTION_STATUS_OK);
1019
1020                 for (i = 0; i < count; i++) {
1021                         struct lttng_action *inner_action = NULL;
1022
1023                         inner_action = lttng_action_group_borrow_mutable_at_index(
1024                                         action, i);
1025                         assert(inner_action);
1026                         ret = add_action_to_subitem_array(
1027                                         inner_action, subitems);
1028                         if (ret) {
1029                                 goto end;
1030                         }
1031                 }
1032
1033                 /*
1034                  * Go directly to the end since there is no need to add the
1035                  * group action by itself to the subitems array.
1036                  */
1037                 goto end;
1038         }
1039
1040         /* Gather execution context. */
1041         switch (type) {
1042         case LTTNG_ACTION_TYPE_NOTIFY:
1043                 break;
1044         case LTTNG_ACTION_TYPE_START_SESSION:
1045                 status = lttng_action_start_session_get_session_name(
1046                                 action, &session_name);
1047                 assert(status == LTTNG_ACTION_STATUS_OK);
1048                 break;
1049         case LTTNG_ACTION_TYPE_STOP_SESSION:
1050                 status = lttng_action_stop_session_get_session_name(
1051                                 action, &session_name);
1052                 assert(status == LTTNG_ACTION_STATUS_OK);
1053                 break;
1054         case LTTNG_ACTION_TYPE_ROTATE_SESSION:
1055                 status = lttng_action_rotate_session_get_session_name(
1056                                 action, &session_name);
1057                 assert(status == LTTNG_ACTION_STATUS_OK);
1058                 break;
1059         case LTTNG_ACTION_TYPE_SNAPSHOT_SESSION:
1060                 status = lttng_action_snapshot_session_get_session_name(
1061                                 action, &session_name);
1062                 assert(status == LTTNG_ACTION_STATUS_OK);
1063                 break;
1064         case LTTNG_ACTION_TYPE_GROUP:
1065         case LTTNG_ACTION_TYPE_UNKNOWN:
1066                 /* Fallthrough */
1067         default:
1068                 abort();
1069                 break;
1070         }
1071
1072         /*
1073          * Fetch the session execution context info as needed.
1074          * Note that we could decide to not add an action for which we know the
1075          * execution will not happen (i.e no session exists for that name). For
1076          * now we leave the decision to skip to the action executor for sake of
1077          * simplicity and consistency.
1078          */
1079         if (session_name != NULL) {
1080                 struct ltt_session *session = NULL;
1081
1082                 session_lock_list();
1083                 session = session_find_by_name(session_name);
1084                 if (session) {
1085                         LTTNG_OPTIONAL_SET(&subitem.context.session_id,
1086                                         session->id);
1087                         session_put(session);
1088                 }
1089
1090                 session_unlock_list();
1091         }
1092
1093         /* Get a reference to the action. */
1094         lttng_action_get(action);
1095         subitem.action = action;
1096
1097         ret = lttng_dynamic_array_add_element(subitems, &subitem);
1098         if (ret) {
1099                 ERR("Failed to add work subitem to the subitem array");
1100                 lttng_action_put(action);
1101                 ret = -1;
1102                 goto end;
1103         }
1104
1105 end:
1106         return ret;
1107 }
1108
1109 static int populate_subitem_array_from_trigger(struct lttng_trigger *trigger,
1110                 struct lttng_dynamic_array *subitems)
1111 {
1112         struct lttng_action *action;
1113
1114         action = lttng_trigger_get_action(trigger);
1115         assert(action);
1116
1117         return add_action_to_subitem_array(action, subitems);
1118 }