root/lib/pengine/utils.c

/* */

DEFINITIONS

1. pe_rsc_action_details
2. pe_free_rsc_action_details
3. pe_can_fence
4. pe__copy_node
5. node_list_exclude
6. pe__node_list2table
7. sort_node_uname
8. pe__output_node_weights
9. pe__log_node_weights
10. pe__show_node_weights_as
11. sort_rsc_index
12. sort_rsc_priority
13. effective_quorum_policy
14. add_singleton
15. lookup_singleton
16. find_existing_action
17. new_action
18. unpack_action_node_attributes
19. update_action_optional
20. update_resource_action_runnable
21. update_resource_flags_for_action
22. custom_action
23. valid_stop_on_fail
24. unpack_operation_on_fail
25. find_min_interval_mon
26. unpack_start_delay
27. unpack_interval_origin
28. unpack_timeout
29. pe_get_configured_timeout
30. unpack_versioned_meta
31. unpack_operation
32. find_rsc_op_entry_helper
33. find_rsc_op_entry
34. print_str_str
35. pe_free_action
36. find_recurring_actions
37. get_complex_task
38. find_first_action
39. find_actions
40. find_actions_exact
41. pe__resource_actions
42. resource_node_score
43. resource_location
44. sort_op_by_callid
45. get_effective_time
46. get_target_role
47. order_actions
48. get_pseudo_op
49. destroy_ticket
50. ticket_new
51. rsc_printable_id
52. pe__clear_resource_flags_recursive
53. pe__clear_resource_flags_on_all
54. pe__set_resource_flags_recursive
55. find_unfencing_devices
56. node_priority_fencing_delay
57. pe_fence_op
58. trigger_unfencing
59. add_tag_ref
60. pe__action2reason
61. pe_action_set_reason
62. pe__shutdown_requested
63. pe__update_recheck_time
64. pe__unpack_dataset_nvpairs
65. pe__resource_is_disabled
66. pe__clear_resource_history
67. pe__rsc_running_on_any
68. pcmk__rsc_filtered_by_node
69. pe__filter_rsc_list
70. pe__build_node_name_list
71. pe__build_rsc_list
72. pe__failed_probe_for_rsc

   1 /*
   2  * Copyright 2004-2022 the Pacemaker project contributors
   3  *
   4  * The version control history for this file may have further details.
   5  *
   6  * This source code is licensed under the GNU Lesser General Public License
   7  * version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
   8  */
   9 
  10 #include <crm_internal.h>
  11 #include <crm/crm.h>
  12 #include <crm/msg_xml.h>
  13 #include <crm/common/xml.h>
  14 #include <crm/common/xml_internal.h>
  15 #include <crm/common/util.h>
  16 
  17 #include <glib.h>
  18 #include <stdbool.h>
  19 
  20 #include <crm/pengine/rules.h>
  21 #include <crm/pengine/internal.h>
  22 #include "pe_status_private.h"
  23 
  24 extern bool pcmk__is_daemon;
  25 
  26 void print_str_str(gpointer key, gpointer value, gpointer user_data);
  27 gboolean ghash_free_str_str(gpointer key, gpointer value, gpointer user_data);
  28 static void unpack_operation(pe_action_t * action, xmlNode * xml_obj, pe_resource_t * container,
  29                              pe_working_set_t * data_set, guint interval_ms);
  30 static xmlNode *find_rsc_op_entry_helper(pe_resource_t * rsc, const char *key,
  31                                          gboolean include_disabled);
  32 
  33 #if ENABLE_VERSIONED_ATTRS
  34 pe_rsc_action_details_t *
  35 pe_rsc_action_details(pe_action_t *action)
     /*  */
  36 {
  37     pe_rsc_action_details_t *details;
  38 
  39     CRM_CHECK(action != NULL, return NULL);
  40 
  41     if (action->action_details == NULL) {
  42         action->action_details = calloc(1, sizeof(pe_rsc_action_details_t));
  43         CRM_CHECK(action->action_details != NULL, return NULL);
  44     }
  45 
  46     details = (pe_rsc_action_details_t *) action->action_details;
  47     if (details->versioned_parameters == NULL) {
  48         details->versioned_parameters = create_xml_node(NULL,
  49                                                         XML_TAG_OP_VER_ATTRS);
  50     }
  51     if (details->versioned_meta == NULL) {
  52         details->versioned_meta = create_xml_node(NULL, XML_TAG_OP_VER_META);
  53     }
  54     return details;
  55 }
  56 
  57 static void
  58 pe_free_rsc_action_details(pe_action_t *action)
     /*  */
  59 {
  60     pe_rsc_action_details_t *details;
  61 
  62     if ((action == NULL) || (action->action_details == NULL)) {
  63         return;
  64     }
  65 
  66     details = (pe_rsc_action_details_t *) action->action_details;
  67 
  68     if (details->versioned_parameters) {
  69         free_xml(details->versioned_parameters);
  70     }
  71     if (details->versioned_meta) {
  72         free_xml(details->versioned_meta);
  73     }
  74 
  75     action->action_details = NULL;
  76 }
  77 #endif
  78 
  79 /*!
  80  * \internal
  81  * \brief Check whether we can fence a particular node
  82  *
  83  * \param[in] data_set  Working set for cluster
  84  * \param[in] node      Name of node to check
  85  *
  86  * \return true if node can be fenced, false otherwise
  87  */
  88 bool
  89 pe_can_fence(pe_working_set_t *data_set, pe_node_t *node)
     /*  */
  90 {
  91     if (pe__is_guest_node(node)) {
  92         /* Guest nodes are fenced by stopping their container resource. We can
  93          * do that if the container's host is either online or fenceable.
  94          */
  95         pe_resource_t *rsc = node->details->remote_rsc->container;
  96 
  97         for (GList *n = rsc->running_on; n != NULL; n = n->next) {
  98             pe_node_t *container_node = n->data;
  99 
 100             if (!container_node->details->online
 101                 && !pe_can_fence(data_set, container_node)) {
 102                 return false;
 103             }
 104         }
 105         return true;
 106 
 107     } else if (!pcmk_is_set(data_set->flags, pe_flag_stonith_enabled)) {
 108         return false; /* Turned off */
 109 
 110     } else if (!pcmk_is_set(data_set->flags, pe_flag_have_stonith_resource)) {
 111         return false; /* No devices */
 112 
 113     } else if (pcmk_is_set(data_set->flags, pe_flag_have_quorum)) {
 114         return true;
 115 
 116     } else if (data_set->no_quorum_policy == no_quorum_ignore) {
 117         return true;
 118 
 119     } else if(node == NULL) {
 120         return false;
 121 
 122     } else if(node->details->online) {
 123         crm_notice("We can fence %s without quorum because they're in our membership", node->details->uname);
 124         return true;
 125     }
 126 
 127     crm_trace("Cannot fence %s", node->details->uname);
 128     return false;
 129 }
 130 
 131 /*!
 132  * \internal
 133  * \brief Copy a node object
 134  *
 135  * \param[in] this_node  Node object to copy
 136  *
 137  * \return Newly allocated shallow copy of this_node
 138  * \note This function asserts on errors and is guaranteed to return non-NULL.
 139  */
 140 pe_node_t *
 141 pe__copy_node(const pe_node_t *this_node)
     /*  */
 142 {
 143     pe_node_t *new_node = NULL;
 144 
 145     CRM_ASSERT(this_node != NULL);
 146 
 147     new_node = calloc(1, sizeof(pe_node_t));
 148     CRM_ASSERT(new_node != NULL);
 149 
 150     new_node->rsc_discover_mode = this_node->rsc_discover_mode;
 151     new_node->weight = this_node->weight;
 152     new_node->fixed = this_node->fixed;
 153     new_node->details = this_node->details;
 154 
 155     return new_node;
 156 }
 157 
 158 /* any node in list1 or list2 and not in the other gets a score of -INFINITY */
 159 void
 160 node_list_exclude(GHashTable * hash, GList *list, gboolean merge_scores)
     /*  */
 161 {
 162     GHashTable *result = hash;
 163     pe_node_t *other_node = NULL;
 164     GList *gIter = list;
 165 
 166     GHashTableIter iter;
 167     pe_node_t *node = NULL;
 168 
 169     g_hash_table_iter_init(&iter, hash);
 170     while (g_hash_table_iter_next(&iter, NULL, (void **)&node)) {
 171 
 172         other_node = pe_find_node_id(list, node->details->id);
 173         if (other_node == NULL) {
 174             node->weight = -INFINITY;
 175         } else if (merge_scores) {
 176             node->weight = pcmk__add_scores(node->weight, other_node->weight);
 177         }
 178     }
 179 
 180     for (; gIter != NULL; gIter = gIter->next) {
 181         pe_node_t *node = (pe_node_t *) gIter->data;
 182 
 183         other_node = pe_hash_table_lookup(result, node->details->id);
 184 
 185         if (other_node == NULL) {
 186             pe_node_t *new_node = pe__copy_node(node);
 187 
 188             new_node->weight = -INFINITY;
 189             g_hash_table_insert(result, (gpointer) new_node->details->id, new_node);
 190         }
 191     }
 192 }
 193 
 194 /*!
 195  * \internal
 196  * \brief Create a node hash table from a node list
 197  *
 198  * \param[in] list  Node list
 199  *
 200  * \return Hash table equivalent of node list
 201  */
 202 GHashTable *
 203 pe__node_list2table(GList *list)
     /*  */
 204 {
 205     GHashTable *result = NULL;
 206 
 207     result = pcmk__strkey_table(NULL, free);
 208     for (GList *gIter = list; gIter != NULL; gIter = gIter->next) {
 209         pe_node_t *new_node = pe__copy_node((pe_node_t *) gIter->data);
 210 
 211         g_hash_table_insert(result, (gpointer) new_node->details->id, new_node);
 212     }
 213     return result;
 214 }
 215 
 216 gint
 217 sort_node_uname(gconstpointer a, gconstpointer b)
     /*  */
 218 {
 219     return pcmk__numeric_strcasecmp(((const pe_node_t *) a)->details->uname,
 220                                     ((const pe_node_t *) b)->details->uname);
 221 }
 222 
 223 /*!
 224  * \internal
 225  * \brief Output node weights to stdout
 226  *
 227  * \param[in] rsc       Use allowed nodes for this resource
 228  * \param[in] comment   Text description to prefix lines with
 229  * \param[in] nodes     If rsc is not specified, use these nodes
 230  */
 231 static void
 232 pe__output_node_weights(pe_resource_t *rsc, const char *comment,
     /*  */
 233                         GHashTable *nodes, pe_working_set_t *data_set)
 234 {
 235     pcmk__output_t *out = data_set->priv;
 236     char score[128]; // Stack-allocated since this is called frequently
 237 
 238     // Sort the nodes so the output is consistent for regression tests
 239     GList *list = g_list_sort(g_hash_table_get_values(nodes), sort_node_uname);
 240 
 241     for (GList *gIter = list; gIter != NULL; gIter = gIter->next) {
 242         pe_node_t *node = (pe_node_t *) gIter->data;
 243 
 244         score2char_stack(node->weight, score, sizeof(score));
 245         out->message(out, "node-weight", rsc, comment, node->details->uname, score);
 246     }
 247     g_list_free(list);
 248 }
 249 
 250 /*!
 251  * \internal
 252  * \brief Log node weights at trace level
 253  *
 254  * \param[in] file      Caller's filename
 255  * \param[in] function  Caller's function name
 256  * \param[in] line      Caller's line number
 257  * \param[in] rsc       Use allowed nodes for this resource
 258  * \param[in] comment   Text description to prefix lines with
 259  * \param[in] nodes     If rsc is not specified, use these nodes
 260  */
 261 static void
 262 pe__log_node_weights(const char *file, const char *function, int line,
     /*  */
 263                      pe_resource_t *rsc, const char *comment, GHashTable *nodes)
 264 {
 265     GHashTableIter iter;
 266     pe_node_t *node = NULL;
 267     char score[128]; // Stack-allocated since this is called frequently
 268 
 269     // Don't waste time if we're not tracing at this point
 270     pcmk__log_else(LOG_TRACE, return);
 271 
 272     g_hash_table_iter_init(&iter, nodes);
 273     while (g_hash_table_iter_next(&iter, NULL, (void **) &node)) {
 274         score2char_stack(node->weight, score, sizeof(score));
 275         if (rsc) {
 276             qb_log_from_external_source(function, file,
 277                                         "%s: %s allocation score on %s: %s",
 278                                         LOG_TRACE, line, 0,
 279                                         comment, rsc->id,
 280                                         node->details->uname, score);
 281         } else {
 282             qb_log_from_external_source(function, file, "%s: %s = %s",
 283                                         LOG_TRACE, line, 0,
 284                                         comment, node->details->uname,
 285                                         score);
 286         }
 287     }
 288 }
 289 
 290 /*!
 291  * \internal
 292  * \brief Log or output node weights
 293  *
 294  * \param[in] file      Caller's filename
 295  * \param[in] function  Caller's function name
 296  * \param[in] line      Caller's line number
 297  * \param[in] to_log    Log if true, otherwise output
 298  * \param[in] rsc       Use allowed nodes for this resource
 299  * \param[in] comment   Text description to prefix lines with
 300  * \param[in] nodes     Use these nodes
 301  */
 302 void
 303 pe__show_node_weights_as(const char *file, const char *function, int line,
     /*  */
 304                          bool to_log, pe_resource_t *rsc, const char *comment,
 305                          GHashTable *nodes, pe_working_set_t *data_set)
 306 {
 307     if (rsc != NULL && pcmk_is_set(rsc->flags, pe_rsc_orphan)) {
 308         // Don't show allocation scores for orphans
 309         return;
 310     }
 311     if (nodes == NULL) {
 312         // Nothing to show
 313         return;
 314     }
 315 
 316     if (to_log) {
 317         pe__log_node_weights(file, function, line, rsc, comment, nodes);
 318     } else {
 319         pe__output_node_weights(rsc, comment, nodes, data_set);
 320     }
 321 
 322     // If this resource has children, repeat recursively for each
 323     if (rsc && rsc->children) {
 324         for (GList *gIter = rsc->children; gIter != NULL; gIter = gIter->next) {
 325             pe_resource_t *child = (pe_resource_t *) gIter->data;
 326 
 327             pe__show_node_weights_as(file, function, line, to_log, child,
 328                                      comment, child->allowed_nodes, data_set);
 329         }
 330     }
 331 }
 332 
 333 gint
 334 sort_rsc_index(gconstpointer a, gconstpointer b)
     /*  */
 335 {
 336     const pe_resource_t *resource1 = (const pe_resource_t *)a;
 337     const pe_resource_t *resource2 = (const pe_resource_t *)b;
 338 
 339     if (a == NULL && b == NULL) {
 340         return 0;
 341     }
 342     if (a == NULL) {
 343         return 1;
 344     }
 345     if (b == NULL) {
 346         return -1;
 347     }
 348 
 349     if (resource1->sort_index > resource2->sort_index) {
 350         return -1;
 351     }
 352 
 353     if (resource1->sort_index < resource2->sort_index) {
 354         return 1;
 355     }
 356 
 357     return 0;
 358 }
 359 
 360 gint
 361 sort_rsc_priority(gconstpointer a, gconstpointer b)
     /*  */
 362 {
 363     const pe_resource_t *resource1 = (const pe_resource_t *)a;
 364     const pe_resource_t *resource2 = (const pe_resource_t *)b;
 365 
 366     if (a == NULL && b == NULL) {
 367         return 0;
 368     }
 369     if (a == NULL) {
 370         return 1;
 371     }
 372     if (b == NULL) {
 373         return -1;
 374     }
 375 
 376     if (resource1->priority > resource2->priority) {
 377         return -1;
 378     }
 379 
 380     if (resource1->priority < resource2->priority) {
 381         return 1;
 382     }
 383 
 384     return 0;
 385 }
 386 
 387 static enum pe_quorum_policy
 388 effective_quorum_policy(pe_resource_t *rsc, pe_working_set_t *data_set)
     /*  */
 389 {
 390     enum pe_quorum_policy policy = data_set->no_quorum_policy;
 391 
 392     if (pcmk_is_set(data_set->flags, pe_flag_have_quorum)) {
 393         policy = no_quorum_ignore;
 394 
 395     } else if (data_set->no_quorum_policy == no_quorum_demote) {
 396         switch (rsc->role) {
 397             case RSC_ROLE_PROMOTED:
 398             case RSC_ROLE_UNPROMOTED:
 399                 if (rsc->next_role > RSC_ROLE_UNPROMOTED) {
 400                     pe__set_next_role(rsc, RSC_ROLE_UNPROMOTED,
 401                                       "no-quorum-policy=demote");
 402                 }
 403                 policy = no_quorum_ignore;
 404                 break;
 405             default:
 406                 policy = no_quorum_stop;
 407                 break;
 408         }
 409     }
 410     return policy;
 411 }
 412 
 413 static void
 414 add_singleton(pe_working_set_t *data_set, pe_action_t *action)
     /*  */
 415 {
 416     if (data_set->singletons == NULL) {
 417         data_set->singletons = pcmk__strkey_table(NULL, NULL);
 418     }
 419     g_hash_table_insert(data_set->singletons, action->uuid, action);
 420 }
 421 
 422 static pe_action_t *
 423 lookup_singleton(pe_working_set_t *data_set, const char *action_uuid)
     /*  */
 424 {
 425     if (data_set->singletons == NULL) {
 426         return NULL;
 427     }
 428     return g_hash_table_lookup(data_set->singletons, action_uuid);
 429 }
 430 
 431 /*!
 432  * \internal
 433  * \brief Find an existing action that matches arguments
 434  *
 435  * \param[in] key        Action key to match
 436  * \param[in] rsc        Resource to match (if any)
 437  * \param[in] node       Node to match (if any)
 438  * \param[in] data_set   Cluster working set
 439  *
 440  * \return Existing action that matches arguments (or NULL if none)
 441  */
 442 static pe_action_t *
 443 find_existing_action(const char *key, pe_resource_t *rsc, pe_node_t *node,
     /*  */
 444                      pe_working_set_t *data_set)
 445 {
 446     GList *matches = NULL;
 447     pe_action_t *action = NULL;
 448 
 449     /* When rsc is NULL, it would be quicker to check data_set->singletons,
 450      * but checking all data_set->actions takes the node into account.
 451      */
 452     matches = find_actions(((rsc == NULL)? data_set->actions : rsc->actions),
 453                            key, node);
 454     if (matches == NULL) {
 455         return NULL;
 456     }
 457     CRM_LOG_ASSERT(!pcmk__list_of_multiple(matches));
 458 
 459     action = matches->data;
 460     g_list_free(matches);
 461     return action;
 462 }
 463 
 464 /*!
 465  * \internal
 466  * \brief Create a new action object
 467  *
 468  * \param[in] key        Action key
 469  * \param[in] task       Action name
 470  * \param[in] rsc        Resource that action is for (if any)
 471  * \param[in] node       Node that action is on (if any)
 472  * \param[in] optional   Whether action should be considered optional
 473  * \param[in] for_graph  Whether action should be recorded in transition graph
 474  * \param[in] data_set   Cluster working set
 475  *
 476  * \return Newly allocated action
 477  * \note This function takes ownership of \p key. It is the caller's
 478  *       responsibility to free the return value with pe_free_action().
 479  */
 480 static pe_action_t *
 481 new_action(char *key, const char *task, pe_resource_t *rsc, pe_node_t *node,
     /*  */
 482            bool optional, bool for_graph, pe_working_set_t *data_set)
 483 {
 484     pe_action_t *action = calloc(1, sizeof(pe_action_t));
 485 
 486     CRM_ASSERT(action != NULL);
 487 
 488     action->rsc = rsc;
 489     action->task = strdup(task); CRM_ASSERT(action->task != NULL);
 490     action->uuid = key;
 491     action->extra = pcmk__strkey_table(free, free);
 492     action->meta = pcmk__strkey_table(free, free);
 493 
 494     if (node) {
 495         action->node = pe__copy_node(node);
 496     }
 497 
 498     if (pcmk__str_eq(task, CRM_OP_LRM_DELETE, pcmk__str_casei)) {
 499         // Resource history deletion for a node can be done on the DC
 500         pe__set_action_flags(action, pe_action_dc);
 501     }
 502 
 503     pe__set_action_flags(action, pe_action_runnable);
 504     if (optional) {
 505         pe__set_action_flags(action, pe_action_optional);
 506     } else {
 507         pe__clear_action_flags(action, pe_action_optional);
 508     }
 509 
 510     if (rsc != NULL) {
 511         guint interval_ms = 0;
 512 
 513         action->op_entry = find_rsc_op_entry_helper(rsc, key, TRUE);
 514         parse_op_key(key, NULL, NULL, &interval_ms);
 515         unpack_operation(action, action->op_entry, rsc->container, data_set,
 516                          interval_ms);
 517     }
 518 
 519     if (for_graph) {
 520         pe_rsc_trace(rsc, "Created %s action %d (%s): %s for %s on %s",
 521                      (optional? "optional" : "required"),
 522                      data_set->action_id, key, task,
 523                      ((rsc == NULL)? "no resource" : rsc->id),
 524                      ((node == NULL)? "no node" : node->details->uname));
 525         action->id = data_set->action_id++;
 526 
 527         data_set->actions = g_list_prepend(data_set->actions, action);
 528         if (rsc == NULL) {
 529             add_singleton(data_set, action);
 530         } else {
 531             rsc->actions = g_list_prepend(rsc->actions, action);
 532         }
 533     }
 534     return action;
 535 }
 536 
 537 /*!
 538  * \internal
 539  * \brief Evaluate node attribute values for an action
 540  *
 541  * \param[in] action    Action to unpack attributes for
 542  * \param[in] data_set  Cluster working set
 543  */
 544 static void
 545 unpack_action_node_attributes(pe_action_t *action, pe_working_set_t *data_set)
     /*  */
 546 {
 547     if (!pcmk_is_set(action->flags, pe_action_have_node_attrs)
 548         && (action->op_entry != NULL)) {
 549 
 550         pe_rule_eval_data_t rule_data = {
 551             .node_hash = action->node->details->attrs,
 552             .role = RSC_ROLE_UNKNOWN,
 553             .now = data_set->now,
 554             .match_data = NULL,
 555             .rsc_data = NULL,
 556             .op_data = NULL
 557         };
 558 
 559         pe__set_action_flags(action, pe_action_have_node_attrs);
 560         pe__unpack_dataset_nvpairs(action->op_entry, XML_TAG_ATTR_SETS,
 561                                    &rule_data, action->extra, NULL,
 562                                    FALSE, data_set);
 563     }
 564 }
 565 
 566 /*!
 567  * \internal
 568  * \brief Update an action's optional flag
 569  *
 570  * \param[in] action    Action to update
 571  * \param[in] optional  Requested optional status
 572  */
 573 static void
 574 update_action_optional(pe_action_t *action, gboolean optional)
     /*  */
 575 {
 576     // Force a non-recurring action to be optional if its resource is unmanaged
 577     if ((action->rsc != NULL) && (action->node != NULL)
 578         && !pcmk_is_set(action->flags, pe_action_pseudo)
 579         && !pcmk_is_set(action->rsc->flags, pe_rsc_managed)
 580         && (g_hash_table_lookup(action->meta,
 581                                 XML_LRM_ATTR_INTERVAL_MS) == NULL)) {
 582             pe_rsc_debug(action->rsc, "%s on %s is optional (%s is unmanaged)",
 583                          action->uuid, action->node->details->uname,
 584                          action->rsc->id);
 585             pe__set_action_flags(action, pe_action_optional);
 586             // We shouldn't clear runnable here because ... something
 587 
 588     // Otherwise require the action if requested
 589     } else if (!optional) {
 590         pe__clear_action_flags(action, pe_action_optional);
 591     }
 592 }
 593 
 594 /*!
 595  * \internal
 596  * \brief Update a resource action's runnable flag
 597  *
 598  * \param[in] action     Action to update
 599  * \param[in] for_graph  Whether action should be recorded in transition graph
 600  * \param[in] data_set   Cluster working set
 601  *
 602  * \note This may also schedule fencing if a stop is unrunnable.
 603  */
 604 static void
 605 update_resource_action_runnable(pe_action_t *action, bool for_graph,
     /*  */
 606                                 pe_working_set_t *data_set)
 607 {
 608     if (pcmk_is_set(action->flags, pe_action_pseudo)) {
 609         return;
 610     }
 611 
 612     if (action->node == NULL) {
 613         pe_rsc_trace(action->rsc, "%s is unrunnable (unallocated)",
 614                      action->uuid);
 615         pe__clear_action_flags(action, pe_action_runnable);
 616 
 617     } else if (!pcmk_is_set(action->flags, pe_action_dc)
 618                && !(action->node->details->online)
 619                && (!pe__is_guest_node(action->node)
 620                    || action->node->details->remote_requires_reset)) {
 621         pe__clear_action_flags(action, pe_action_runnable);
 622         do_crm_log((for_graph? LOG_WARNING: LOG_TRACE),
 623                    "%s on %s is unrunnable (node is offline)",
 624                    action->uuid, action->node->details->uname);
 625         if (pcmk_is_set(action->rsc->flags, pe_rsc_managed)
 626             && for_graph
 627             && pcmk__str_eq(action->task, CRMD_ACTION_STOP, pcmk__str_casei)
 628             && !(action->node->details->unclean)) {
 629             pe_fence_node(data_set, action->node, "stop is unrunnable", false);
 630         }
 631 
 632     } else if (!pcmk_is_set(action->flags, pe_action_dc)
 633                && action->node->details->pending) {
 634         pe__clear_action_flags(action, pe_action_runnable);
 635         do_crm_log((for_graph? LOG_WARNING: LOG_TRACE),
 636                    "Action %s on %s is unrunnable (node is pending)",
 637                    action->uuid, action->node->details->uname);
 638 
 639     } else if (action->needs == rsc_req_nothing) {
 640         pe_action_set_reason(action, NULL, TRUE);
 641         if (pe__is_guest_node(action->node)
 642             && !pe_can_fence(data_set, action->node)) {
 643             /* An action that requires nothing usually does not require any
 644              * fencing in order to be runnable. However, there is an exception:
 645              * such an action cannot be completed if it is on a guest node whose
 646              * host is unclean and cannot be fenced.
 647              */
 648             pe_rsc_debug(action->rsc, "%s on %s is unrunnable "
 649                          "(node's host cannot be fenced)",
 650                          action->uuid, action->node->details->uname);
 651             pe__clear_action_flags(action, pe_action_runnable);
 652         } else {
 653             pe_rsc_trace(action->rsc,
 654                          "%s on %s does not require fencing or quorum",
 655                          action->uuid, action->node->details->uname);
 656             pe__set_action_flags(action, pe_action_runnable);
 657         }
 658 
 659     } else {
 660         switch (effective_quorum_policy(action->rsc, data_set)) {
 661             case no_quorum_stop:
 662                 pe_rsc_debug(action->rsc, "%s on %s is unrunnable (no quorum)",
 663                              action->uuid, action->node->details->uname);
 664                 pe__clear_action_flags(action, pe_action_runnable);
 665                 pe_action_set_reason(action, "no quorum", true);
 666                 break;
 667 
 668             case no_quorum_freeze:
 669                 if (!action->rsc->fns->active(action->rsc, TRUE)
 670                     || (action->rsc->next_role > action->rsc->role)) {
 671                     pe_rsc_debug(action->rsc,
 672                                  "%s on %s is unrunnable (no quorum)",
 673                                  action->uuid, action->node->details->uname);
 674                     pe__clear_action_flags(action, pe_action_runnable);
 675                     pe_action_set_reason(action, "quorum freeze", true);
 676                 }
 677                 break;
 678 
 679             default:
 680                 //pe_action_set_reason(action, NULL, TRUE);
 681                 pe__set_action_flags(action, pe_action_runnable);
 682                 break;
 683         }
 684     }
 685 }
 686 
 687 /*!
 688  * \internal
 689  * \brief Update a resource object's flags for a new action on it
 690  *
 691  * \param[in] rsc        Resource that action is for (if any)
 692  * \param[in] action     New action
 693  */
 694 static void
 695 update_resource_flags_for_action(pe_resource_t *rsc, pe_action_t *action)
     /*  */
 696 {
 697     /* @COMPAT pe_rsc_starting and pe_rsc_stopping are not actually used
 698      * within Pacemaker, and should be deprecated and eventually removed
 699      */
 700     if (pcmk__str_eq(action->task, CRMD_ACTION_STOP, pcmk__str_casei)) {
 701         pe__set_resource_flags(rsc, pe_rsc_stopping);
 702 
 703     } else if (pcmk__str_eq(action->task, CRMD_ACTION_START, pcmk__str_casei)) {
 704         if (pcmk_is_set(action->flags, pe_action_runnable)) {
 705             pe__set_resource_flags(rsc, pe_rsc_starting);
 706         } else {
 707             pe__clear_resource_flags(rsc, pe_rsc_starting);
 708         }
 709     }
 710 }
 711 
 712 /*!
 713  * \brief Create or update an action object
 714  *
 715  * \param[in] rsc          Resource that action is for (if any)
 716  * \param[in] key          Action key (must be non-NULL)
 717  * \param[in] task         Action name (must be non-NULL)
 718  * \param[in] on_node      Node that action is on (if any)
 719  * \param[in] optional     Whether action should be considered optional
 720  * \param[in] save_action  Whether action should be recorded in transition graph
 721  * \param[in] data_set     Cluster working set
 722  *
 723  * \return Action object corresponding to arguments
 724  * \note This function takes ownership of (and might free) \p key. If
 725  *       \p save_action is true, \p data_set will own the returned action,
 726  *       otherwise it is the caller's responsibility to free the return value
 727  *       with pe_free_action().
 728  */
 729 pe_action_t *
 730 custom_action(pe_resource_t *rsc, char *key, const char *task,
     /*  */
 731               pe_node_t *on_node, gboolean optional, gboolean save_action,
 732               pe_working_set_t *data_set)
 733 {
 734     pe_action_t *action = NULL;
 735 
 736     CRM_ASSERT((key != NULL) && (task != NULL) && (data_set != NULL));
 737 
 738     if (save_action) {
 739         action = find_existing_action(key, rsc, on_node, data_set);
 740     }
 741 
 742     if (action == NULL) {
 743         action = new_action(key, task, rsc, on_node, optional, save_action,
 744                             data_set);
 745     } else {
 746         free(key);
 747     }
 748 
 749     update_action_optional(action, optional);
 750 
 751     if (rsc != NULL) {
 752         if (action->node != NULL) {
 753             unpack_action_node_attributes(action, data_set);
 754         }
 755 
 756         update_resource_action_runnable(action, save_action, data_set);
 757 
 758         if (save_action) {
 759             update_resource_flags_for_action(rsc, action);
 760         }
 761     }
 762 
 763     return action;
 764 }
 765 
 766 static bool
 767 valid_stop_on_fail(const char *value)
     /*  */
 768 {
 769     return !pcmk__strcase_any_of(value, "standby", "demote", "stop", NULL);
 770 }
 771 
 772 static const char *
 773 unpack_operation_on_fail(pe_action_t * action)
     /*  */
 774 {
 775 
 776     const char *name = NULL;
 777     const char *role = NULL;
 778     const char *on_fail = NULL;
 779     const char *interval_spec = NULL;
 780     const char *value = g_hash_table_lookup(action->meta, XML_OP_ATTR_ON_FAIL);
 781 
 782     if (pcmk__str_eq(action->task, CRMD_ACTION_STOP, pcmk__str_casei)
 783         && !valid_stop_on_fail(value)) {
 784 
 785         pcmk__config_err("Resetting '" XML_OP_ATTR_ON_FAIL "' for %s stop "
 786                          "action to default value because '%s' is not "
 787                          "allowed for stop", action->rsc->id, value);
 788         return NULL;
 789 
 790     } else if (pcmk__str_eq(action->task, CRMD_ACTION_DEMOTE, pcmk__str_casei) && !value) {
 791         // demote on_fail defaults to monitor value for promoted role if present
 792         xmlNode *operation = NULL;
 793 
 794         CRM_CHECK(action->rsc != NULL, return NULL);
 795 
 796         for (operation = pcmk__xe_first_child(action->rsc->ops_xml);
 797              (operation != NULL) && (value == NULL);
 798              operation = pcmk__xe_next(operation)) {
 799             bool enabled = false;
 800 
 801             if (!pcmk__str_eq((const char *)operation->name, "op", pcmk__str_none)) {
 802                 continue;
 803             }
 804             name = crm_element_value(operation, "name");
 805             role = crm_element_value(operation, "role");
 806             on_fail = crm_element_value(operation, XML_OP_ATTR_ON_FAIL);
 807             interval_spec = crm_element_value(operation, XML_LRM_ATTR_INTERVAL);
 808             if (!on_fail) {
 809                 continue;
 810             } else if (pcmk__xe_get_bool_attr(operation, "enabled", &enabled) == pcmk_rc_ok && !enabled) {
 811                 continue;
 812             } else if (!pcmk__str_eq(name, "monitor", pcmk__str_casei)
 813                        || !pcmk__strcase_any_of(role, RSC_ROLE_PROMOTED_S,
 814                                                 RSC_ROLE_PROMOTED_LEGACY_S,
 815                                                 NULL)) {
 816                 continue;
 817             } else if (crm_parse_interval_spec(interval_spec) == 0) {
 818                 continue;
 819             } else if (pcmk__str_eq(on_fail, "demote", pcmk__str_casei)) {
 820                 continue;
 821             }
 822 
 823             value = on_fail;
 824         }
 825     } else if (pcmk__str_eq(action->task, CRM_OP_LRM_DELETE, pcmk__str_casei)) {
 826         value = "ignore";
 827 
 828     } else if (pcmk__str_eq(value, "demote", pcmk__str_casei)) {
 829         name = crm_element_value(action->op_entry, "name");
 830         role = crm_element_value(action->op_entry, "role");
 831         interval_spec = crm_element_value(action->op_entry,
 832                                           XML_LRM_ATTR_INTERVAL);
 833 
 834         if (!pcmk__str_eq(name, CRMD_ACTION_PROMOTE, pcmk__str_casei)
 835             && (!pcmk__str_eq(name, CRMD_ACTION_STATUS, pcmk__str_casei)
 836                 || !pcmk__strcase_any_of(role, RSC_ROLE_PROMOTED_S,
 837                                          RSC_ROLE_PROMOTED_LEGACY_S, NULL)
 838                 || (crm_parse_interval_spec(interval_spec) == 0))) {
 839             pcmk__config_err("Resetting '" XML_OP_ATTR_ON_FAIL "' for %s %s "
 840                              "action to default value because 'demote' is not "
 841                              "allowed for it", action->rsc->id, name);
 842             return NULL;
 843         }
 844     }
 845 
 846     return value;
 847 }
 848 
 849 static xmlNode *
 850 find_min_interval_mon(pe_resource_t * rsc, gboolean include_disabled)
     /*  */
 851 {
 852     guint interval_ms = 0;
 853     guint min_interval_ms = G_MAXUINT;
 854     const char *name = NULL;
 855     const char *interval_spec = NULL;
 856     xmlNode *op = NULL;
 857     xmlNode *operation = NULL;
 858 
 859     for (operation = pcmk__xe_first_child(rsc->ops_xml);
 860          operation != NULL;
 861          operation = pcmk__xe_next(operation)) {
 862 
 863         if (pcmk__str_eq((const char *)operation->name, "op", pcmk__str_none)) {
 864             bool enabled = false;
 865 
 866             name = crm_element_value(operation, "name");
 867             interval_spec = crm_element_value(operation, XML_LRM_ATTR_INTERVAL);
 868             if (!include_disabled && pcmk__xe_get_bool_attr(operation, "enabled", &enabled) == pcmk_rc_ok &&
 869                 !enabled) {
 870                 continue;
 871             }
 872 
 873             if (!pcmk__str_eq(name, RSC_STATUS, pcmk__str_casei)) {
 874                 continue;
 875             }
 876 
 877             interval_ms = crm_parse_interval_spec(interval_spec);
 878 
 879             if (interval_ms && (interval_ms < min_interval_ms)) {
 880                 min_interval_ms = interval_ms;
 881                 op = operation;
 882             }
 883         }
 884     }
 885 
 886     return op;
 887 }
 888 
 889 static int
 890 unpack_start_delay(const char *value, GHashTable *meta)
     /*  */
 891 {
 892     int start_delay = 0;
 893 
 894     if (value != NULL) {
 895         start_delay = crm_get_msec(value);
 896 
 897         if (start_delay < 0) {
 898             start_delay = 0;
 899         }
 900 
 901         if (meta) {
 902             g_hash_table_replace(meta, strdup(XML_OP_ATTR_START_DELAY),
 903                                  pcmk__itoa(start_delay));
 904         }
 905     }
 906 
 907     return start_delay;
 908 }
 909 
 910 // true if value contains valid, non-NULL interval origin for recurring op
 911 static bool
 912 unpack_interval_origin(const char *value, xmlNode *xml_obj, guint interval_ms,
     /*  */
 913                        crm_time_t *now, long long *start_delay)
 914 {
 915     long long result = 0;
 916     guint interval_sec = interval_ms / 1000;
 917     crm_time_t *origin = NULL;
 918 
 919     // Ignore unspecified values and non-recurring operations
 920     if ((value == NULL) || (interval_ms == 0) || (now == NULL)) {
 921         return false;
 922     }
 923 
 924     // Parse interval origin from text
 925     origin = crm_time_new(value);
 926     if (origin == NULL) {
 927         pcmk__config_err("Ignoring '" XML_OP_ATTR_ORIGIN "' for operation "
 928                          "'%s' because '%s' is not valid",
 929                          (ID(xml_obj)? ID(xml_obj) : "(missing ID)"), value);
 930         return false;
 931     }
 932 
 933     // Get seconds since origin (negative if origin is in the future)
 934     result = crm_time_get_seconds(now) - crm_time_get_seconds(origin);
 935     crm_time_free(origin);
 936 
 937     // Calculate seconds from closest interval to now
 938     result = result % interval_sec;
 939 
 940     // Calculate seconds remaining until next interval
 941     result = ((result <= 0)? 0 : interval_sec) - result;
 942     crm_info("Calculated a start delay of %llds for operation '%s'",
 943              result,
 944              (ID(xml_obj)? ID(xml_obj) : "(unspecified)"));
 945 
 946     if (start_delay != NULL) {
 947         *start_delay = result * 1000; // milliseconds
 948     }
 949     return true;
 950 }
 951 
 952 static int
 953 unpack_timeout(const char *value)
     /*  */
 954 {
 955     int timeout_ms = crm_get_msec(value);
 956 
 957     if (timeout_ms < 0) {
 958         timeout_ms = crm_get_msec(CRM_DEFAULT_OP_TIMEOUT_S);
 959     }
 960     return timeout_ms;
 961 }
 962 
 963 int
 964 pe_get_configured_timeout(pe_resource_t *rsc, const char *action, pe_working_set_t *data_set)
     /*  */
 965 {
 966     xmlNode *child = NULL;
 967     GHashTable *action_meta = NULL;
 968     const char *timeout_spec = NULL;
 969     int timeout_ms = 0;
 970 
 971     pe_rule_eval_data_t rule_data = {
 972         .node_hash = NULL,
 973         .role = RSC_ROLE_UNKNOWN,
 974         .now = data_set->now,
 975         .match_data = NULL,
 976         .rsc_data = NULL,
 977         .op_data = NULL
 978     };
 979 
 980     for (child = first_named_child(rsc->ops_xml, XML_ATTR_OP);
 981          child != NULL; child = crm_next_same_xml(child)) {
 982         if (pcmk__str_eq(action, crm_element_value(child, XML_NVPAIR_ATTR_NAME),
 983                 pcmk__str_casei)) {
 984             timeout_spec = crm_element_value(child, XML_ATTR_TIMEOUT);
 985             break;
 986         }
 987     }
 988 
 989     if (timeout_spec == NULL && data_set->op_defaults) {
 990         action_meta = pcmk__strkey_table(free, free);
 991         pe__unpack_dataset_nvpairs(data_set->op_defaults, XML_TAG_META_SETS,
 992                                    &rule_data, action_meta, NULL, FALSE, data_set);
 993         timeout_spec = g_hash_table_lookup(action_meta, XML_ATTR_TIMEOUT);
 994     }
 995 
 996     // @TODO check meta-attributes (including versioned meta-attributes)
 997     // @TODO maybe use min-interval monitor timeout as default for monitors
 998 
 999     timeout_ms = crm_get_msec(timeout_spec);
1000     if (timeout_ms < 0) {
1001         timeout_ms = crm_get_msec(CRM_DEFAULT_OP_TIMEOUT_S);
1002     }
1003 
1004     if (action_meta != NULL) {
1005         g_hash_table_destroy(action_meta);
1006     }
1007     return timeout_ms;
1008 }
1009 
1010 #if ENABLE_VERSIONED_ATTRS
1011 static void
1012 unpack_versioned_meta(xmlNode *versioned_meta, xmlNode *xml_obj,
     /*  */
1013                       guint interval_ms, crm_time_t *now)
1014 {
1015     xmlNode *attrs = NULL;
1016     xmlNode *attr = NULL;
1017 
1018     for (attrs = pcmk__xe_first_child(versioned_meta); attrs != NULL;
1019          attrs = pcmk__xe_next(attrs)) {
1020 
1021         for (attr = pcmk__xe_first_child(attrs); attr != NULL;
1022              attr = pcmk__xe_next(attr)) {
1023 
1024             const char *name = crm_element_value(attr, XML_NVPAIR_ATTR_NAME);
1025             const char *value = crm_element_value(attr, XML_NVPAIR_ATTR_VALUE);
1026 
1027             if (pcmk__str_eq(name, XML_OP_ATTR_START_DELAY, pcmk__str_casei)) {
1028                 int start_delay = unpack_start_delay(value, NULL);
1029 
1030                 crm_xml_add_int(attr, XML_NVPAIR_ATTR_VALUE, start_delay);
1031             } else if (pcmk__str_eq(name, XML_OP_ATTR_ORIGIN, pcmk__str_casei)) {
1032                 long long start_delay = 0;
1033 
1034                 if (unpack_interval_origin(value, xml_obj, interval_ms, now,
1035                                            &start_delay)) {
1036                     crm_xml_add(attr, XML_NVPAIR_ATTR_NAME,
1037                                 XML_OP_ATTR_START_DELAY);
1038                     crm_xml_add_ll(attr, XML_NVPAIR_ATTR_VALUE, start_delay);
1039                 }
1040             } else if (pcmk__str_eq(name, XML_ATTR_TIMEOUT, pcmk__str_casei)) {
1041                 int timeout_ms = unpack_timeout(value);
1042 
1043                 crm_xml_add_int(attr, XML_NVPAIR_ATTR_VALUE, timeout_ms);
1044             }
1045         }
1046     }
1047 }
1048 #endif
1049 
1050 /*!
1051  * \brief Unpack operation XML into an action structure
1052  *
1053  * Unpack an operation's meta-attributes (normalizing the interval, timeout,
1054  * and start delay values as integer milliseconds), requirements, and
1055  * failure policy.
1056  *
1057  * \param[in,out] action      Action to unpack into
1058  * \param[in]     xml_obj     Operation XML (or NULL if all defaults)
1059  * \param[in]     container   Resource that contains affected resource, if any
1060  * \param[in]     data_set    Cluster state
1061  * \param[in]     interval_ms How frequently to perform the operation
1062  */
1063 static void
1064 unpack_operation(pe_action_t * action, xmlNode * xml_obj, pe_resource_t * container,
     /*  */
1065                  pe_working_set_t * data_set, guint interval_ms)
1066 {
1067     int timeout_ms = 0;
1068     const char *value = NULL;
1069     bool is_probe = false;
1070 #if ENABLE_VERSIONED_ATTRS
1071     pe_rsc_action_details_t *rsc_details = NULL;
1072 #endif
1073 
1074     pe_rsc_eval_data_t rsc_rule_data = {
1075         .standard = crm_element_value(action->rsc->xml, XML_AGENT_ATTR_CLASS),
1076         .provider = crm_element_value(action->rsc->xml, XML_AGENT_ATTR_PROVIDER),
1077         .agent = crm_element_value(action->rsc->xml, XML_EXPR_ATTR_TYPE)
1078     };
1079 
1080     pe_op_eval_data_t op_rule_data = {
1081         .op_name = action->task,
1082         .interval = interval_ms
1083     };
1084 
1085     pe_rule_eval_data_t rule_data = {
1086         .node_hash = NULL,
1087         .role = RSC_ROLE_UNKNOWN,
1088         .now = data_set->now,
1089         .match_data = NULL,
1090         .rsc_data = &rsc_rule_data,
1091         .op_data = &op_rule_data
1092     };
1093 
1094     CRM_CHECK(action && action->rsc, return);
1095 
1096     is_probe = pcmk_is_probe(action->task, interval_ms);
1097 
1098     // Cluster-wide <op_defaults> <meta_attributes>
1099     pe__unpack_dataset_nvpairs(data_set->op_defaults, XML_TAG_META_SETS, &rule_data,
1100                                action->meta, NULL, FALSE, data_set);
1101 
1102     // Determine probe default timeout differently
1103     if (is_probe) {
1104         xmlNode *min_interval_mon = find_min_interval_mon(action->rsc, FALSE);
1105 
1106         if (min_interval_mon) {
1107             value = crm_element_value(min_interval_mon, XML_ATTR_TIMEOUT);
1108             if (value) {
1109                 crm_trace("\t%s: Setting default timeout to minimum-interval "
1110                           "monitor's timeout '%s'", action->uuid, value);
1111                 g_hash_table_replace(action->meta, strdup(XML_ATTR_TIMEOUT),
1112                                      strdup(value));
1113             }
1114         }
1115     }
1116 
1117     if (xml_obj) {
1118         xmlAttrPtr xIter = NULL;
1119 
1120         // <op> <meta_attributes> take precedence over defaults
1121         pe__unpack_dataset_nvpairs(xml_obj, XML_TAG_META_SETS, &rule_data,
1122                                    action->meta, NULL, TRUE, data_set);
1123 
1124 #if ENABLE_VERSIONED_ATTRS
1125         rsc_details = pe_rsc_action_details(action);
1126 
1127         /* Non-versioned attributes also unpack XML_TAG_ATTR_SETS, but that
1128          * capability is deprecated, so we don't need to extend that support to
1129          * versioned attributes.
1130          */
1131         pe_eval_versioned_attributes(data_set->input, xml_obj,
1132                                      XML_TAG_META_SETS, &rule_data,
1133                                      rsc_details->versioned_meta,
1134                                      NULL);
1135 #endif
1136 
1137         /* Anything set as an <op> XML property has highest precedence.
1138          * This ensures we use the name and interval from the <op> tag.
1139          */
1140         for (xIter = xml_obj->properties; xIter; xIter = xIter->next) {
1141             const char *prop_name = (const char *)xIter->name;
1142             const char *prop_value = crm_element_value(xml_obj, prop_name);
1143 
1144             g_hash_table_replace(action->meta, strdup(prop_name), strdup(prop_value));
1145         }
1146     }
1147 
1148     g_hash_table_remove(action->meta, "id");
1149 
1150     // Normalize interval to milliseconds
1151     if (interval_ms > 0) {
1152         g_hash_table_replace(action->meta, strdup(XML_LRM_ATTR_INTERVAL),
1153                              crm_strdup_printf("%u", interval_ms));
1154     } else {
1155         g_hash_table_remove(action->meta, XML_LRM_ATTR_INTERVAL);
1156     }
1157 
1158     /*
1159      * Timeout order of precedence:
1160      *   1. pcmk_monitor_timeout (if rsc has pcmk_ra_cap_fence_params
1161      *      and task is start or a probe; pcmk_monitor_timeout works
1162      *      by default for a recurring monitor)
1163      *   2. explicit op timeout on the primitive
1164      *   3. default op timeout
1165      *      a. if probe, then min-interval monitor's timeout
1166      *      b. else, in XML_CIB_TAG_OPCONFIG
1167      *   4. CRM_DEFAULT_OP_TIMEOUT_S
1168      *
1169      * #1 overrides general rule of <op> XML property having highest
1170      * precedence.
1171      */
1172     if (pcmk_is_set(pcmk_get_ra_caps(rsc_rule_data.standard),
1173                     pcmk_ra_cap_fence_params)
1174         && (pcmk__str_eq(action->task, RSC_START, pcmk__str_casei)
1175             || is_probe)) {
1176 
1177         GHashTable *params = pe_rsc_params(action->rsc, action->node, data_set);
1178 
1179         value = g_hash_table_lookup(params, "pcmk_monitor_timeout");
1180 
1181         if (value) {
1182             crm_trace("\t%s: Setting timeout to pcmk_monitor_timeout '%s', "
1183                       "overriding default", action->uuid, value);
1184             g_hash_table_replace(action->meta, strdup(XML_ATTR_TIMEOUT),
1185                                  strdup(value));
1186         }
1187     }
1188 
1189     // Normalize timeout to positive milliseconds
1190     value = g_hash_table_lookup(action->meta, XML_ATTR_TIMEOUT);
1191     timeout_ms = unpack_timeout(value);
1192     g_hash_table_replace(action->meta, strdup(XML_ATTR_TIMEOUT),
1193                          pcmk__itoa(timeout_ms));
1194 
1195     if (!pcmk__strcase_any_of(action->task, RSC_START, RSC_PROMOTE, NULL)) {
1196         action->needs = rsc_req_nothing;
1197         value = "nothing (not start or promote)";
1198 
1199     } else if (pcmk_is_set(action->rsc->flags, pe_rsc_needs_fencing)) {
1200         action->needs = rsc_req_stonith;
1201         value = "fencing";
1202 
1203     } else if (pcmk_is_set(action->rsc->flags, pe_rsc_needs_quorum)) {
1204         action->needs = rsc_req_quorum;
1205         value = "quorum";
1206 
1207     } else {
1208         action->needs = rsc_req_nothing;
1209         value = "nothing";
1210     }
1211     pe_rsc_trace(action->rsc, "%s requires %s", action->uuid, value);
1212 
1213     value = unpack_operation_on_fail(action);
1214 
1215     if (value == NULL) {
1216 
1217     } else if (pcmk__str_eq(value, "block", pcmk__str_casei)) {
1218         action->on_fail = action_fail_block;
1219         g_hash_table_insert(action->meta, strdup(XML_OP_ATTR_ON_FAIL), strdup("block"));
1220         value = "block"; // The above could destroy the original string
1221 
1222     } else if (pcmk__str_eq(value, "fence", pcmk__str_casei)) {
1223         action->on_fail = action_fail_fence;
1224         value = "node fencing";
1225 
1226         if (!pcmk_is_set(data_set->flags, pe_flag_stonith_enabled)) {
1227             pcmk__config_err("Resetting '" XML_OP_ATTR_ON_FAIL "' for "
1228                              "operation '%s' to 'stop' because 'fence' is not "
1229                              "valid when fencing is disabled", action->uuid);
1230             action->on_fail = action_fail_stop;
1231             action->fail_role = RSC_ROLE_STOPPED;
1232             value = "stop resource";
1233         }
1234 
1235     } else if (pcmk__str_eq(value, "standby", pcmk__str_casei)) {
1236         action->on_fail = action_fail_standby;
1237         value = "node standby";
1238 
1239     } else if (pcmk__strcase_any_of(value, "ignore", "nothing", NULL)) {
1240         action->on_fail = action_fail_ignore;
1241         value = "ignore";
1242 
1243     } else if (pcmk__str_eq(value, "migrate", pcmk__str_casei)) {
1244         action->on_fail = action_fail_migrate;
1245         value = "force migration";
1246 
1247     } else if (pcmk__str_eq(value, "stop", pcmk__str_casei)) {
1248         action->on_fail = action_fail_stop;
1249         action->fail_role = RSC_ROLE_STOPPED;
1250         value = "stop resource";
1251 
1252     } else if (pcmk__str_eq(value, "restart", pcmk__str_casei)) {
1253         action->on_fail = action_fail_recover;
1254         value = "restart (and possibly migrate)";
1255 
1256     } else if (pcmk__str_eq(value, "restart-container", pcmk__str_casei)) {
1257         if (container) {
1258             action->on_fail = action_fail_restart_container;
1259             value = "restart container (and possibly migrate)";
1260 
1261         } else {
1262             value = NULL;
1263         }
1264 
1265     } else if (pcmk__str_eq(value, "demote", pcmk__str_casei)) {
1266         action->on_fail = action_fail_demote;
1267         value = "demote instance";
1268 
1269     } else {
1270         pe_err("Resource %s: Unknown failure type (%s)", action->rsc->id, value);
1271         value = NULL;
1272     }
1273 
1274     /* defaults */
1275     if (value == NULL && container) {
1276         action->on_fail = action_fail_restart_container;
1277         value = "restart container (and possibly migrate) (default)";
1278 
1279     /* For remote nodes, ensure that any failure that results in dropping an
1280      * active connection to the node results in fencing of the node.
1281      *
1282      * There are only two action failures that don't result in fencing.
1283      * 1. probes - probe failures are expected.
1284      * 2. start - a start failure indicates that an active connection does not already
1285      * exist. The user can set op on-fail=fence if they really want to fence start
1286      * failures. */
1287     } else if (((value == NULL) || !pcmk_is_set(action->rsc->flags, pe_rsc_managed))
1288                && pe__resource_is_remote_conn(action->rsc, data_set)
1289                && !(pcmk__str_eq(action->task, CRMD_ACTION_STATUS, pcmk__str_casei)
1290                     && (interval_ms == 0))
1291                && !pcmk__str_eq(action->task, CRMD_ACTION_START, pcmk__str_casei)) {
1292 
1293         if (!pcmk_is_set(action->rsc->flags, pe_rsc_managed)) {
1294             action->on_fail = action_fail_stop;
1295             action->fail_role = RSC_ROLE_STOPPED;
1296             value = "stop unmanaged remote node (enforcing default)";
1297 
1298         } else {
1299             if (pcmk_is_set(data_set->flags, pe_flag_stonith_enabled)) {
1300                 value = "fence remote node (default)";
1301             } else {
1302                 value = "recover remote node connection (default)";
1303             }
1304 
1305             if (action->rsc->remote_reconnect_ms) {
1306                 action->fail_role = RSC_ROLE_STOPPED;
1307             }
1308             action->on_fail = action_fail_reset_remote;
1309         }
1310 
1311     } else if (value == NULL && pcmk__str_eq(action->task, CRMD_ACTION_STOP, pcmk__str_casei)) {
1312         if (pcmk_is_set(data_set->flags, pe_flag_stonith_enabled)) {
1313             action->on_fail = action_fail_fence;
1314             value = "resource fence (default)";
1315 
1316         } else {
1317             action->on_fail = action_fail_block;
1318             value = "resource block (default)";
1319         }
1320 
1321     } else if (value == NULL) {
1322         action->on_fail = action_fail_recover;
1323         value = "restart (and possibly migrate) (default)";
1324     }
1325 
1326     pe_rsc_trace(action->rsc, "%s failure handling: %s",
1327                  action->uuid, value);
1328 
1329     value = NULL;
1330     if (xml_obj != NULL) {
1331         value = g_hash_table_lookup(action->meta, "role_after_failure");
1332         if (value) {
1333             pe_warn_once(pe_wo_role_after,
1334                         "Support for role_after_failure is deprecated and will be removed in a future release");
1335         }
1336     }
1337     if (value != NULL && action->fail_role == RSC_ROLE_UNKNOWN) {
1338         action->fail_role = text2role(value);
1339     }
1340     /* defaults */
1341     if (action->fail_role == RSC_ROLE_UNKNOWN) {
1342         if (pcmk__str_eq(action->task, CRMD_ACTION_PROMOTE, pcmk__str_casei)) {
1343             action->fail_role = RSC_ROLE_UNPROMOTED;
1344         } else {
1345             action->fail_role = RSC_ROLE_STARTED;
1346         }
1347     }
1348     pe_rsc_trace(action->rsc, "%s failure results in: %s",
1349                  action->uuid, role2text(action->fail_role));
1350 
1351     value = g_hash_table_lookup(action->meta, XML_OP_ATTR_START_DELAY);
1352     if (value) {
1353         unpack_start_delay(value, action->meta);
1354     } else {
1355         long long start_delay = 0;
1356 
1357         value = g_hash_table_lookup(action->meta, XML_OP_ATTR_ORIGIN);
1358         if (unpack_interval_origin(value, xml_obj, interval_ms, data_set->now,
1359                                    &start_delay)) {
1360             g_hash_table_replace(action->meta, strdup(XML_OP_ATTR_START_DELAY),
1361                                  crm_strdup_printf("%lld", start_delay));
1362         }
1363     }
1364 
1365 #if ENABLE_VERSIONED_ATTRS
1366     unpack_versioned_meta(rsc_details->versioned_meta, xml_obj, interval_ms,
1367                           data_set->now);
1368 #endif
1369 }
1370 
1371 static xmlNode *
1372 find_rsc_op_entry_helper(pe_resource_t * rsc, const char *key, gboolean include_disabled)
     /*  */
1373 {
1374     guint interval_ms = 0;
1375     gboolean do_retry = TRUE;
1376     char *local_key = NULL;
1377     const char *name = NULL;
1378     const char *interval_spec = NULL;
1379     char *match_key = NULL;
1380     xmlNode *op = NULL;
1381     xmlNode *operation = NULL;
1382 
1383   retry:
1384     for (operation = pcmk__xe_first_child(rsc->ops_xml); operation != NULL;
1385          operation = pcmk__xe_next(operation)) {
1386 
1387         if (pcmk__str_eq((const char *)operation->name, "op", pcmk__str_none)) {
1388             bool enabled = false;
1389 
1390             name = crm_element_value(operation, "name");
1391             interval_spec = crm_element_value(operation, XML_LRM_ATTR_INTERVAL);
1392             if (!include_disabled && pcmk__xe_get_bool_attr(operation, "enabled", &enabled) == pcmk_rc_ok &&
1393                 !enabled) {
1394                 continue;
1395             }
1396 
1397             interval_ms = crm_parse_interval_spec(interval_spec);
1398             match_key = pcmk__op_key(rsc->id, name, interval_ms);
1399             if (pcmk__str_eq(key, match_key, pcmk__str_casei)) {
1400                 op = operation;
1401             }
1402             free(match_key);
1403 
1404             if (rsc->clone_name) {
1405                 match_key = pcmk__op_key(rsc->clone_name, name, interval_ms);
1406                 if (pcmk__str_eq(key, match_key, pcmk__str_casei)) {
1407                     op = operation;
1408                 }
1409                 free(match_key);
1410             }
1411 
1412             if (op != NULL) {
1413                 free(local_key);
1414                 return op;
1415             }
1416         }
1417     }
1418 
1419     free(local_key);
1420     if (do_retry == FALSE) {
1421         return NULL;
1422     }
1423 
1424     do_retry = FALSE;
1425     if (strstr(key, CRMD_ACTION_MIGRATE) || strstr(key, CRMD_ACTION_MIGRATED)) {
1426         local_key = pcmk__op_key(rsc->id, "migrate", 0);
1427         key = local_key;
1428         goto retry;
1429 
1430     } else if (strstr(key, "_notify_")) {
1431         local_key = pcmk__op_key(rsc->id, "notify", 0);
1432         key = local_key;
1433         goto retry;
1434     }
1435 
1436     return NULL;
1437 }
1438 
1439 xmlNode *
1440 find_rsc_op_entry(pe_resource_t * rsc, const char *key)
     /*  */
1441 {
1442     return find_rsc_op_entry_helper(rsc, key, FALSE);
1443 }
1444 
1445 /*
1446  * Used by the HashTable for-loop
1447  */
1448 void
1449 print_str_str(gpointer key, gpointer value, gpointer user_data)
     /*  */
1450 {
1451     crm_trace("%s%s %s ==> %s",
1452               user_data == NULL ? "" : (char *)user_data,
1453               user_data == NULL ? "" : ": ", (char *)key, (char *)value);
1454 }
1455 
1456 void
1457 pe_free_action(pe_action_t * action)
     /*  */
1458 {
1459     if (action == NULL) {
1460         return;
1461     }
1462     g_list_free_full(action->actions_before, free);     /* pe_action_wrapper_t* */
1463     g_list_free_full(action->actions_after, free);      /* pe_action_wrapper_t* */
1464     if (action->extra) {
1465         g_hash_table_destroy(action->extra);
1466     }
1467     if (action->meta) {
1468         g_hash_table_destroy(action->meta);
1469     }
1470 #if ENABLE_VERSIONED_ATTRS
1471     if (action->rsc) {
1472         pe_free_rsc_action_details(action);
1473     }
1474 #endif
1475     free(action->cancel_task);
1476     free(action->reason);
1477     free(action->task);
1478     free(action->uuid);
1479     free(action->node);
1480     free(action);
1481 }
1482 
1483 GList *
1484 find_recurring_actions(GList *input, pe_node_t * not_on_node)
     /*  */
1485 {
1486     const char *value = NULL;
1487     GList *result = NULL;
1488     GList *gIter = input;
1489 
1490     CRM_CHECK(input != NULL, return NULL);
1491 
1492     for (; gIter != NULL; gIter = gIter->next) {
1493         pe_action_t *action = (pe_action_t *) gIter->data;
1494 
1495         value = g_hash_table_lookup(action->meta, XML_LRM_ATTR_INTERVAL_MS);
1496         if (value == NULL) {
1497             /* skip */
1498         } else if (pcmk__str_eq(value, "0", pcmk__str_casei)) {
1499             /* skip */
1500         } else if (pcmk__str_eq(CRMD_ACTION_CANCEL, action->task, pcmk__str_casei)) {
1501             /* skip */
1502         } else if (not_on_node == NULL) {
1503             crm_trace("(null) Found: %s", action->uuid);
1504             result = g_list_prepend(result, action);
1505 
1506         } else if (action->node == NULL) {
1507             /* skip */
1508         } else if (action->node->details != not_on_node->details) {
1509             crm_trace("Found: %s", action->uuid);
1510             result = g_list_prepend(result, action);
1511         }
1512     }
1513 
1514     return result;
1515 }
1516 
1517 enum action_tasks
1518 get_complex_task(pe_resource_t * rsc, const char *name, gboolean allow_non_atomic)
     /*  */
1519 {
1520     enum action_tasks task = text2task(name);
1521 
1522     if (rsc == NULL) {
1523         return task;
1524 
1525     } else if (allow_non_atomic == FALSE || rsc->variant == pe_native) {
1526         switch (task) {
1527             case stopped_rsc:
1528             case started_rsc:
1529             case action_demoted:
1530             case action_promoted:
1531                 crm_trace("Folding %s back into its atomic counterpart for %s", name, rsc->id);
1532                 return task - 1;
1533             default:
1534                 break;
1535         }
1536     }
1537     return task;
1538 }
1539 
1540 pe_action_t *
1541 find_first_action(GList *input, const char *uuid, const char *task, pe_node_t * on_node)
     /*  */
1542 {
1543     GList *gIter = NULL;
1544 
1545     CRM_CHECK(uuid || task, return NULL);
1546 
1547     for (gIter = input; gIter != NULL; gIter = gIter->next) {
1548         pe_action_t *action = (pe_action_t *) gIter->data;
1549 
1550         if (uuid != NULL && !pcmk__str_eq(uuid, action->uuid, pcmk__str_casei)) {
1551             continue;
1552 
1553         } else if (task != NULL && !pcmk__str_eq(task, action->task, pcmk__str_casei)) {
1554             continue;
1555 
1556         } else if (on_node == NULL) {
1557             return action;
1558 
1559         } else if (action->node == NULL) {
1560             continue;
1561 
1562         } else if (on_node->details == action->node->details) {
1563             return action;
1564         }
1565     }
1566 
1567     return NULL;
1568 }
1569 
1570 GList *
1571 find_actions(GList *input, const char *key, const pe_node_t *on_node)
     /*  */
1572 {
1573     GList *gIter = input;
1574     GList *result = NULL;
1575 
1576     CRM_CHECK(key != NULL, return NULL);
1577 
1578     for (; gIter != NULL; gIter = gIter->next) {
1579         pe_action_t *action = (pe_action_t *) gIter->data;
1580 
1581         if (!pcmk__str_eq(key, action->uuid, pcmk__str_casei)) {
1582             continue;
1583 
1584         } else if (on_node == NULL) {
1585             crm_trace("Action %s matches (ignoring node)", key);
1586             result = g_list_prepend(result, action);
1587 
1588         } else if (action->node == NULL) {
1589             crm_trace("Action %s matches (unallocated, assigning to %s)",
1590                       key, on_node->details->uname);
1591 
1592             action->node = pe__copy_node(on_node);
1593             result = g_list_prepend(result, action);
1594 
1595         } else if (on_node->details == action->node->details) {
1596             crm_trace("Action %s on %s matches", key, on_node->details->uname);
1597             result = g_list_prepend(result, action);
1598         }
1599     }
1600 
1601     return result;
1602 }
1603 
1604 GList *
1605 find_actions_exact(GList *input, const char *key, const pe_node_t *on_node)
     /*  */
1606 {
1607     GList *result = NULL;
1608 
1609     CRM_CHECK(key != NULL, return NULL);
1610 
1611     if (on_node == NULL) {
1612         return NULL;
1613     }
1614 
1615     for (GList *gIter = input; gIter != NULL; gIter = gIter->next) {
1616         pe_action_t *action = (pe_action_t *) gIter->data;
1617 
1618         if ((action->node != NULL)
1619             && pcmk__str_eq(key, action->uuid, pcmk__str_casei)
1620             && pcmk__str_eq(on_node->details->id, action->node->details->id,
1621                             pcmk__str_casei)) {
1622 
1623             crm_trace("Action %s on %s matches", key, on_node->details->uname);
1624             result = g_list_prepend(result, action);
1625         }
1626     }
1627 
1628     return result;
1629 }
1630 
1631 /*!
1632  * \brief Find all actions of given type for a resource
1633  *
1634  * \param[in] rsc           Resource to search
1635  * \param[in] node          Find only actions scheduled on this node
1636  * \param[in] task          Action name to search for
1637  * \param[in] require_node  If TRUE, NULL node or action node will not match
1638  *
1639  * \return List of actions found (or NULL if none)
1640  * \note If node is not NULL and require_node is FALSE, matching actions
1641  *       without a node will be assigned to node.
1642  */
1643 GList *
1644 pe__resource_actions(const pe_resource_t *rsc, const pe_node_t *node,
     /*  */
1645                      const char *task, bool require_node)
1646 {
1647     GList *result = NULL;
1648     char *key = pcmk__op_key(rsc->id, task, 0);
1649 
1650     if (require_node) {
1651         result = find_actions_exact(rsc->actions, key, node);
1652     } else {
1653         result = find_actions(rsc->actions, key, node);
1654     }
1655     free(key);
1656     return result;
1657 }
1658 
1659 static void
1660 resource_node_score(pe_resource_t * rsc, pe_node_t * node, int score, const char *tag)
     /*  */
1661 {
1662     pe_node_t *match = NULL;
1663 
1664     if ((rsc->exclusive_discover || (node->rsc_discover_mode == pe_discover_never))
1665         && pcmk__str_eq(tag, "symmetric_default", pcmk__str_casei)) {
1666         /* This string comparision may be fragile, but exclusive resources and
1667          * exclusive nodes should not have the symmetric_default constraint
1668          * applied to them.
1669          */
1670         return;
1671 
1672     } else if (rsc->children) {
1673         GList *gIter = rsc->children;
1674 
1675         for (; gIter != NULL; gIter = gIter->next) {
1676             pe_resource_t *child_rsc = (pe_resource_t *) gIter->data;
1677 
1678             resource_node_score(child_rsc, node, score, tag);
1679         }
1680     }
1681 
1682     pe_rsc_trace(rsc, "Setting %s for %s on %s: %d", tag, rsc->id, node->details->uname, score);
1683     match = pe_hash_table_lookup(rsc->allowed_nodes, node->details->id);
1684     if (match == NULL) {
1685         match = pe__copy_node(node);
1686         g_hash_table_insert(rsc->allowed_nodes, (gpointer) match->details->id, match);
1687     }
1688     match->weight = pcmk__add_scores(match->weight, score);
1689 }
1690 
1691 void
1692 resource_location(pe_resource_t * rsc, pe_node_t * node, int score, const char *tag,
     /*  */
1693                   pe_working_set_t * data_set)
1694 {
1695     if (node != NULL) {
1696         resource_node_score(rsc, node, score, tag);
1697 
1698     } else if (data_set != NULL) {
1699         GList *gIter = data_set->nodes;
1700 
1701         for (; gIter != NULL; gIter = gIter->next) {
1702             pe_node_t *node_iter = (pe_node_t *) gIter->data;
1703 
1704             resource_node_score(rsc, node_iter, score, tag);
1705         }
1706 
1707     } else {
1708         GHashTableIter iter;
1709         pe_node_t *node_iter = NULL;
1710 
1711         g_hash_table_iter_init(&iter, rsc->allowed_nodes);
1712         while (g_hash_table_iter_next(&iter, NULL, (void **)&node_iter)) {
1713             resource_node_score(rsc, node_iter, score, tag);
1714         }
1715     }
1716 
1717     if (node == NULL && score == -INFINITY) {
1718         if (rsc->allocated_to) {
1719             crm_info("Deallocating %s from %s", rsc->id, rsc->allocated_to->details->uname);
1720             free(rsc->allocated_to);
1721             rsc->allocated_to = NULL;
1722         }
1723     }
1724 }
1725 
1726 #define sort_return(an_int, why) do {                                   \
1727         free(a_uuid);                                           \
1728         free(b_uuid);                                           \
1729         crm_trace("%s (%d) %c %s (%d) : %s",                            \
1730                   a_xml_id, a_call_id, an_int>0?'>':an_int<0?'<':'=',   \
1731                   b_xml_id, b_call_id, why);                            \
1732         return an_int;                                                  \
1733     } while(0)
1734 
1735 gint
1736 sort_op_by_callid(gconstpointer a, gconstpointer b)
     /*  */
1737 {
1738     int a_call_id = -1;
1739     int b_call_id = -1;
1740 
1741     char *a_uuid = NULL;
1742     char *b_uuid = NULL;
1743 
1744     const xmlNode *xml_a = a;
1745     const xmlNode *xml_b = b;
1746 
1747     const char *a_xml_id = crm_element_value(xml_a, XML_ATTR_ID);
1748     const char *b_xml_id = crm_element_value(xml_b, XML_ATTR_ID);
1749 
1750     if (pcmk__str_eq(a_xml_id, b_xml_id, pcmk__str_casei)) {
1751         /* We have duplicate lrm_rsc_op entries in the status
1752          * section which is unlikely to be a good thing
1753          *    - we can handle it easily enough, but we need to get
1754          *    to the bottom of why it's happening.
1755          */
1756         pe_err("Duplicate lrm_rsc_op entries named %s", a_xml_id);
1757         sort_return(0, "duplicate");
1758     }
1759 
1760     crm_element_value_int(xml_a, XML_LRM_ATTR_CALLID, &a_call_id);
1761     crm_element_value_int(xml_b, XML_LRM_ATTR_CALLID, &b_call_id);
1762 
1763     if (a_call_id == -1 && b_call_id == -1) {
1764         /* both are pending ops so it doesn't matter since
1765          *   stops are never pending
1766          */
1767         sort_return(0, "pending");
1768 
1769     } else if (a_call_id >= 0 && a_call_id < b_call_id) {
1770         sort_return(-1, "call id");
1771 
1772     } else if (b_call_id >= 0 && a_call_id > b_call_id) {
1773         sort_return(1, "call id");
1774 
1775     } else if (b_call_id >= 0 && a_call_id == b_call_id) {
1776         /*
1777          * The op and last_failed_op are the same
1778          * Order on last-rc-change
1779          */
1780         time_t last_a = -1;
1781         time_t last_b = -1;
1782 
1783         crm_element_value_epoch(xml_a, XML_RSC_OP_LAST_CHANGE, &last_a);
1784         crm_element_value_epoch(xml_b, XML_RSC_OP_LAST_CHANGE, &last_b);
1785 
1786         crm_trace("rc-change: %lld vs %lld",
1787                   (long long) last_a, (long long) last_b);
1788         if (last_a >= 0 && last_a < last_b) {
1789             sort_return(-1, "rc-change");
1790 
1791         } else if (last_b >= 0 && last_a > last_b) {
1792             sort_return(1, "rc-change");
1793         }
1794         sort_return(0, "rc-change");
1795 
1796     } else {
1797         /* One of the inputs is a pending operation
1798          * Attempt to use XML_ATTR_TRANSITION_MAGIC to determine its age relative to the other
1799          */
1800 
1801         int a_id = -1;
1802         int b_id = -1;
1803 
1804         const char *a_magic = crm_element_value(xml_a, XML_ATTR_TRANSITION_MAGIC);
1805         const char *b_magic = crm_element_value(xml_b, XML_ATTR_TRANSITION_MAGIC);
1806 
1807         CRM_CHECK(a_magic != NULL && b_magic != NULL, sort_return(0, "No magic"));
1808         if (!decode_transition_magic(a_magic, &a_uuid, &a_id, NULL, NULL, NULL,
1809                                      NULL)) {
1810             sort_return(0, "bad magic a");
1811         }
1812         if (!decode_transition_magic(b_magic, &b_uuid, &b_id, NULL, NULL, NULL,
1813                                      NULL)) {
1814             sort_return(0, "bad magic b");
1815         }
1816         /* try to determine the relative age of the operation...
1817          * some pending operations (e.g. a start) may have been superseded
1818          *   by a subsequent stop
1819          *
1820          * [a|b]_id == -1 means it's a shutdown operation and _always_ comes last
1821          */
1822         if (!pcmk__str_eq(a_uuid, b_uuid, pcmk__str_casei) || a_id == b_id) {
1823             /*
1824              * some of the logic in here may be redundant...
1825              *
1826              * if the UUID from the TE doesn't match then one better
1827              *   be a pending operation.
1828              * pending operations don't survive between elections and joins
1829              *   because we query the LRM directly
1830              */
1831 
1832             if (b_call_id == -1) {
1833                 sort_return(-1, "transition + call");
1834 
1835             } else if (a_call_id == -1) {
1836                 sort_return(1, "transition + call");
1837             }
1838 
1839         } else if ((a_id >= 0 && a_id < b_id) || b_id == -1) {
1840             sort_return(-1, "transition");
1841 
1842         } else if ((b_id >= 0 && a_id > b_id) || a_id == -1) {
1843             sort_return(1, "transition");
1844         }
1845     }
1846 
1847     /* we should never end up here */
1848     CRM_CHECK(FALSE, sort_return(0, "default"));
1849 
1850 }
1851 
1852 time_t
1853 get_effective_time(pe_working_set_t * data_set)
     /*  */
1854 {
1855     if(data_set) {
1856         if (data_set->now == NULL) {
1857             crm_trace("Recording a new 'now'");
1858             data_set->now = crm_time_new(NULL);
1859         }
1860         return crm_time_get_seconds_since_epoch(data_set->now);
1861     }
1862 
1863     crm_trace("Defaulting to 'now'");
1864     return time(NULL);
1865 }
1866 
1867 gboolean
1868 get_target_role(pe_resource_t * rsc, enum rsc_role_e * role)
     /*  */
1869 {
1870     enum rsc_role_e local_role = RSC_ROLE_UNKNOWN;
1871     const char *value = g_hash_table_lookup(rsc->meta, XML_RSC_ATTR_TARGET_ROLE);
1872 
1873     CRM_CHECK(role != NULL, return FALSE);
1874 
1875     if (pcmk__str_eq(value, "started", pcmk__str_null_matches | pcmk__str_casei)
1876         || pcmk__str_eq("default", value, pcmk__str_casei)) {
1877         return FALSE;
1878     }
1879 
1880     local_role = text2role(value);
1881     if (local_role == RSC_ROLE_UNKNOWN) {
1882         pcmk__config_err("Ignoring '" XML_RSC_ATTR_TARGET_ROLE "' for %s "
1883                          "because '%s' is not valid", rsc->id, value);
1884         return FALSE;
1885 
1886     } else if (local_role > RSC_ROLE_STARTED) {
1887         if (pcmk_is_set(uber_parent(rsc)->flags, pe_rsc_promotable)) {
1888             if (local_role > RSC_ROLE_UNPROMOTED) {
1889                 /* This is what we'd do anyway, just leave the default to avoid messing up the placement algorithm */
1890                 return FALSE;
1891             }
1892 
1893         } else {
1894             pcmk__config_err("Ignoring '" XML_RSC_ATTR_TARGET_ROLE "' for %s "
1895                              "because '%s' only makes sense for promotable "
1896                              "clones", rsc->id, value);
1897             return FALSE;
1898         }
1899     }
1900 
1901     *role = local_role;
1902     return TRUE;
1903 }
1904 
1905 gboolean
1906 order_actions(pe_action_t * lh_action, pe_action_t * rh_action, enum pe_ordering order)
     /*  */
1907 {
1908     GList *gIter = NULL;
1909     pe_action_wrapper_t *wrapper = NULL;
1910     GList *list = NULL;
1911 
1912     if (order == pe_order_none) {
1913         return FALSE;
1914     }
1915 
1916     if (lh_action == NULL || rh_action == NULL) {
1917         return FALSE;
1918     }
1919 
1920     crm_trace("Creating action wrappers for ordering: %s then %s",
1921               lh_action->uuid, rh_action->uuid);
1922 
1923     /* Ensure we never create a dependency on ourselves... it's happened */
1924     CRM_ASSERT(lh_action != rh_action);
1925 
1926     /* Filter dups, otherwise update_action_states() has too much work to do */
1927     gIter = lh_action->actions_after;
1928     for (; gIter != NULL; gIter = gIter->next) {
1929         pe_action_wrapper_t *after = (pe_action_wrapper_t *) gIter->data;
1930 
1931         if (after->action == rh_action && (after->type & order)) {
1932             return FALSE;
1933         }
1934     }
1935 
1936     wrapper = calloc(1, sizeof(pe_action_wrapper_t));
1937     wrapper->action = rh_action;
1938     wrapper->type = order;
1939     list = lh_action->actions_after;
1940     list = g_list_prepend(list, wrapper);
1941     lh_action->actions_after = list;
1942 
1943     wrapper = calloc(1, sizeof(pe_action_wrapper_t));
1944     wrapper->action = lh_action;
1945     wrapper->type = order;
1946     list = rh_action->actions_before;
1947     list = g_list_prepend(list, wrapper);
1948     rh_action->actions_before = list;
1949     return TRUE;
1950 }
1951 
1952 pe_action_t *
1953 get_pseudo_op(const char *name, pe_working_set_t * data_set)
     /*  */
1954 {
1955     pe_action_t *op = lookup_singleton(data_set, name);
1956 
1957     if (op == NULL) {
1958         op = custom_action(NULL, strdup(name), name, NULL, TRUE, TRUE, data_set);
1959         pe__set_action_flags(op, pe_action_pseudo|pe_action_runnable);
1960     }
1961     return op;
1962 }
1963 
1964 void
1965 destroy_ticket(gpointer data)
     /*  */
1966 {
1967     pe_ticket_t *ticket = data;
1968 
1969     if (ticket->state) {
1970         g_hash_table_destroy(ticket->state);
1971     }
1972     free(ticket->id);
1973     free(ticket);
1974 }
1975 
1976 pe_ticket_t *
1977 ticket_new(const char *ticket_id, pe_working_set_t * data_set)
     /*  */
1978 {
1979     pe_ticket_t *ticket = NULL;
1980 
1981     if (pcmk__str_empty(ticket_id)) {
1982         return NULL;
1983     }
1984 
1985     if (data_set->tickets == NULL) {
1986         data_set->tickets = pcmk__strkey_table(free, destroy_ticket);
1987     }
1988 
1989     ticket = g_hash_table_lookup(data_set->tickets, ticket_id);
1990     if (ticket == NULL) {
1991 
1992         ticket = calloc(1, sizeof(pe_ticket_t));
1993         if (ticket == NULL) {
1994             crm_err("Cannot allocate ticket '%s'", ticket_id);
1995             return NULL;
1996         }
1997 
1998         crm_trace("Creaing ticket entry for %s", ticket_id);
1999 
2000         ticket->id = strdup(ticket_id);
2001         ticket->granted = FALSE;
2002         ticket->last_granted = -1;
2003         ticket->standby = FALSE;
2004         ticket->state = pcmk__strkey_table(free, free);
2005 
2006         g_hash_table_insert(data_set->tickets, strdup(ticket->id), ticket);
2007     }
2008 
2009     return ticket;
2010 }
2011 
2012 const char *rsc_printable_id(pe_resource_t *rsc)
     /*  */
2013 {
2014     if (!pcmk_is_set(rsc->flags, pe_rsc_unique)) {
2015         return ID(rsc->xml);
2016     }
2017     return rsc->id;
2018 }
2019 
2020 void
2021 pe__clear_resource_flags_recursive(pe_resource_t *rsc, uint64_t flags)
     /*  */
2022 {
2023     pe__clear_resource_flags(rsc, flags);
2024     for (GList *gIter = rsc->children; gIter != NULL; gIter = gIter->next) {
2025         pe__clear_resource_flags_recursive((pe_resource_t *) gIter->data, flags);
2026     }
2027 }
2028 
2029 void
2030 pe__clear_resource_flags_on_all(pe_working_set_t *data_set, uint64_t flag)
     /*  */
2031 {
2032     for (GList *lpc = data_set->resources; lpc != NULL; lpc = lpc->next) {
2033         pe_resource_t *r = (pe_resource_t *) lpc->data;
2034         pe__clear_resource_flags_recursive(r, flag);
2035     }
2036 }
2037 
2038 void
2039 pe__set_resource_flags_recursive(pe_resource_t *rsc, uint64_t flags)
     /*  */
2040 {
2041     pe__set_resource_flags(rsc, flags);
2042     for (GList *gIter = rsc->children; gIter != NULL; gIter = gIter->next) {
2043         pe__set_resource_flags_recursive((pe_resource_t *) gIter->data, flags);
2044     }
2045 }
2046 
2047 static GList *
2048 find_unfencing_devices(GList *candidates, GList *matches) 
     /*  */
2049 {
2050     for (GList *gIter = candidates; gIter != NULL; gIter = gIter->next) {
2051         pe_resource_t *candidate = gIter->data;
2052         const char *provides = g_hash_table_lookup(candidate->meta,
2053                                                    PCMK_STONITH_PROVIDES);
2054         const char *requires = g_hash_table_lookup(candidate->meta, XML_RSC_ATTR_REQUIRES);
2055 
2056         if(candidate->children) {
2057             matches = find_unfencing_devices(candidate->children, matches);
2058         } else if (!pcmk_is_set(candidate->flags, pe_rsc_fence_device)) {
2059             continue;
2060 
2061         } else if (pcmk__str_eq(provides, "unfencing", pcmk__str_casei) || pcmk__str_eq(requires, "unfencing", pcmk__str_casei)) {
2062             matches = g_list_prepend(matches, candidate);
2063         }
2064     }
2065     return matches;
2066 }
2067 
2068 static int
2069 node_priority_fencing_delay(pe_node_t * node, pe_working_set_t * data_set)
     /*  */
2070 {
2071     int member_count = 0;
2072     int online_count = 0;
2073     int top_priority = 0;
2074     int lowest_priority = 0;
2075     GList *gIter = NULL;
2076 
2077     // `priority-fencing-delay` is disabled
2078     if (data_set->priority_fencing_delay <= 0) {
2079         return 0;
2080     }
2081 
2082     /* No need to request a delay if the fencing target is not a normal cluster
2083      * member, for example if it's a remote node or a guest node. */
2084     if (node->details->type != node_member) {
2085         return 0;
2086     }
2087 
2088     // No need to request a delay if the fencing target is in our partition
2089     if (node->details->online) {
2090         return 0;
2091     }
2092 
2093     for (gIter = data_set->nodes; gIter != NULL; gIter = gIter->next) {
2094         pe_node_t *n =  gIter->data;
2095 
2096         if (n->details->type != node_member) {
2097             continue;
2098         }
2099 
2100         member_count ++;
2101 
2102         if (n->details->online) {
2103             online_count++;
2104         }
2105 
2106         if (member_count == 1
2107             || n->details->priority > top_priority) {
2108             top_priority = n->details->priority;
2109         }
2110 
2111         if (member_count == 1
2112             || n->details->priority < lowest_priority) {
2113             lowest_priority = n->details->priority;
2114         }
2115     }
2116 
2117     // No need to delay if we have more than half of the cluster members
2118     if (online_count > member_count / 2) {
2119         return 0;
2120     }
2121 
2122     /* All the nodes have equal priority.
2123      * Any configured corresponding `pcmk_delay_base/max` will be applied. */
2124     if (lowest_priority == top_priority) {
2125         return 0;
2126     }
2127 
2128     if (node->details->priority < top_priority) {
2129         return 0;
2130     }
2131 
2132     return data_set->priority_fencing_delay;
2133 }
2134 
2135 pe_action_t *
2136 pe_fence_op(pe_node_t * node, const char *op, bool optional, const char *reason,
     /*  */
2137             bool priority_delay, pe_working_set_t * data_set)
2138 {
2139     char *op_key = NULL;
2140     pe_action_t *stonith_op = NULL;
2141 
2142     if(op == NULL) {
2143         op = data_set->stonith_action;
2144     }
2145 
2146     op_key = crm_strdup_printf("%s-%s-%s", CRM_OP_FENCE, node->details->uname, op);
2147 
2148     stonith_op = lookup_singleton(data_set, op_key);
2149     if(stonith_op == NULL) {
2150         stonith_op = custom_action(NULL, op_key, CRM_OP_FENCE, node, TRUE, TRUE, data_set);
2151 
2152         add_hash_param(stonith_op->meta, XML_LRM_ATTR_TARGET, node->details->uname);
2153         add_hash_param(stonith_op->meta, XML_LRM_ATTR_TARGET_UUID, node->details->id);
2154         add_hash_param(stonith_op->meta, "stonith_action", op);
2155 
2156         if (pe__is_guest_or_remote_node(node)
2157             && pcmk_is_set(data_set->flags, pe_flag_enable_unfencing)) {
2158             /* Extra work to detect device changes on remotes
2159              *
2160              * We may do this for all nodes in the future, but for now
2161              * the pcmk__check_action_config() based stuff works fine.
2162              */
2163             long max = 1024;
2164             long digests_all_offset = 0;
2165             long digests_secure_offset = 0;
2166 
2167             char *digests_all = calloc(max, sizeof(char));
2168             char *digests_secure = calloc(max, sizeof(char));
2169             GList *matches = find_unfencing_devices(data_set->resources, NULL);
2170 
2171             for (GList *gIter = matches; gIter != NULL; gIter = gIter->next) {
2172                 pe_resource_t *match = gIter->data;
2173                 const char *agent = g_hash_table_lookup(match->meta,
2174                                                         XML_ATTR_TYPE);
2175                 op_digest_cache_t *data = NULL;
2176 
2177                 data = pe__compare_fencing_digest(match, agent, node, data_set);
2178                 if(data->rc == RSC_DIGEST_ALL) {
2179                     optional = FALSE;
2180                     crm_notice("Unfencing %s (remote): because the definition of %s changed", node->details->uname, match->id);
2181                     if (!pcmk__is_daemon && data_set->priv != NULL) {
2182                         pcmk__output_t *out = data_set->priv;
2183                         out->info(out, "notice: Unfencing %s (remote): because the definition of %s changed",
2184                                   node->details->uname, match->id);
2185                     }
2186                 }
2187 
2188                 digests_all_offset += snprintf(
2189                     digests_all+digests_all_offset, max-digests_all_offset,
2190                     "%s:%s:%s,", match->id, agent, data->digest_all_calc);
2191 
2192                 digests_secure_offset += snprintf(
2193                     digests_secure+digests_secure_offset, max-digests_secure_offset,
2194                     "%s:%s:%s,", match->id, agent, data->digest_secure_calc);
2195             }
2196             g_hash_table_insert(stonith_op->meta,
2197                                 strdup(XML_OP_ATTR_DIGESTS_ALL),
2198                                 digests_all);
2199             g_hash_table_insert(stonith_op->meta,
2200                                 strdup(XML_OP_ATTR_DIGESTS_SECURE),
2201                                 digests_secure);
2202         }
2203 
2204     } else {
2205         free(op_key);
2206     }
2207 
2208     if (data_set->priority_fencing_delay > 0
2209 
2210             /* It's a suitable case where `priority-fencing-delay` applies.
2211              * At least add `priority-fencing-delay` field as an indicator. */
2212         && (priority_delay
2213 
2214             /* The priority delay needs to be recalculated if this function has
2215              * been called by schedule_fencing_and_shutdowns() after node
2216              * priority has already been calculated by native_add_running().
2217              */
2218             || g_hash_table_lookup(stonith_op->meta,
2219                                    XML_CONFIG_ATTR_PRIORITY_FENCING_DELAY) != NULL)) {
2220 
2221             /* Add `priority-fencing-delay` to the fencing op even if it's 0 for
2222              * the targeting node. So that it takes precedence over any possible
2223              * `pcmk_delay_base/max`.
2224              */
2225             char *delay_s = pcmk__itoa(node_priority_fencing_delay(node, data_set));
2226 
2227             g_hash_table_insert(stonith_op->meta,
2228                                 strdup(XML_CONFIG_ATTR_PRIORITY_FENCING_DELAY),
2229                                 delay_s);
2230     }
2231 
2232     if(optional == FALSE && pe_can_fence(data_set, node)) {
2233         pe__clear_action_flags(stonith_op, pe_action_optional);
2234         pe_action_set_reason(stonith_op, reason, false);
2235 
2236     } else if(reason && stonith_op->reason == NULL) {
2237         stonith_op->reason = strdup(reason);
2238     }
2239 
2240     return stonith_op;
2241 }
2242 
2243 void
2244 trigger_unfencing(
     /*  */
2245     pe_resource_t * rsc, pe_node_t *node, const char *reason, pe_action_t *dependency, pe_working_set_t * data_set) 
2246 {
2247     if (!pcmk_is_set(data_set->flags, pe_flag_enable_unfencing)) {
2248         /* No resources require it */
2249         return;
2250 
2251     } else if ((rsc != NULL)
2252                && !pcmk_is_set(rsc->flags, pe_rsc_fence_device)) {
2253         /* Wasn't a stonith device */
2254         return;
2255 
2256     } else if(node
2257               && node->details->online
2258               && node->details->unclean == FALSE
2259               && node->details->shutdown == FALSE) {
2260         pe_action_t *unfence = pe_fence_op(node, "on", FALSE, reason, FALSE, data_set);
2261 
2262         if(dependency) {
2263             order_actions(unfence, dependency, pe_order_optional);
2264         }
2265 
2266     } else if(rsc) {
2267         GHashTableIter iter;
2268 
2269         g_hash_table_iter_init(&iter, rsc->allowed_nodes);
2270         while (g_hash_table_iter_next(&iter, NULL, (void **)&node)) {
2271             if(node->details->online && node->details->unclean == FALSE && node->details->shutdown == FALSE) {
2272                 trigger_unfencing(rsc, node, reason, dependency, data_set);
2273             }
2274         }
2275     }
2276 }
2277 
2278 gboolean
2279 add_tag_ref(GHashTable * tags, const char * tag_name,  const char * obj_ref)
     /*  */
2280 {
2281     pe_tag_t *tag = NULL;
2282     GList *gIter = NULL;
2283     gboolean is_existing = FALSE;
2284 
2285     CRM_CHECK(tags && tag_name && obj_ref, return FALSE);
2286 
2287     tag = g_hash_table_lookup(tags, tag_name);
2288     if (tag == NULL) {
2289         tag = calloc(1, sizeof(pe_tag_t));
2290         if (tag == NULL) {
2291             return FALSE;
2292         }
2293         tag->id = strdup(tag_name);
2294         tag->refs = NULL;
2295         g_hash_table_insert(tags, strdup(tag_name), tag);
2296     }
2297 
2298     for (gIter = tag->refs; gIter != NULL; gIter = gIter->next) {
2299         const char *existing_ref = (const char *) gIter->data;
2300 
2301         if (pcmk__str_eq(existing_ref, obj_ref, pcmk__str_none)){
2302             is_existing = TRUE;
2303             break;
2304         }
2305     }
2306 
2307     if (is_existing == FALSE) {
2308         tag->refs = g_list_append(tag->refs, strdup(obj_ref));
2309         crm_trace("Added: tag=%s ref=%s", tag->id, obj_ref);
2310     }
2311 
2312     return TRUE;
2313 }
2314 
2315 /*!
2316  * \internal
2317  * \brief Create an action reason string based on the action itself
2318  *
2319  * \param[in] action  Action to create reason string for
2320  * \param[in] flag    Action flag that was cleared
2321  *
2322  * \return Newly allocated string suitable for use as action reason
2323  * \note It is the caller's responsibility to free() the result.
2324  */
2325 char *
2326 pe__action2reason(pe_action_t *action, enum pe_action_flags flag)
     /*  */
2327 {
2328     const char *change = NULL;
2329 
2330     switch (flag) {
2331         case pe_action_runnable:
2332         case pe_action_migrate_runnable:
2333             change = "unrunnable";
2334             break;
2335         case pe_action_optional:
2336             change = "required";
2337             break;
2338         default:
2339             // Bug: caller passed unsupported flag
2340             CRM_CHECK(change != NULL, change = "");
2341             break;
2342     }
2343     return crm_strdup_printf("%s%s%s %s", change,
2344                              (action->rsc == NULL)? "" : " ",
2345                              (action->rsc == NULL)? "" : action->rsc->id,
2346                              action->task);
2347 }
2348 
2349 void pe_action_set_reason(pe_action_t *action, const char *reason, bool overwrite) 
     /*  */
2350 {
2351     if (action->reason != NULL && overwrite) {
2352         pe_rsc_trace(action->rsc, "Changing %s reason from '%s' to '%s'",
2353                      action->uuid, action->reason, crm_str(reason));
2354     } else if (action->reason == NULL) {
2355         pe_rsc_trace(action->rsc, "Set %s reason to '%s'",
2356                      action->uuid, crm_str(reason));
2357     } else {
2358         // crm_assert(action->reason != NULL && !overwrite);
2359         return;
2360     }
2361 
2362     pcmk__str_update(&action->reason, reason);
2363 }
2364 
2365 /*!
2366  * \internal
2367  * \brief Check whether shutdown has been requested for a node
2368  *
2369  * \param[in] node  Node to check
2370  *
2371  * \return TRUE if node has shutdown attribute set and nonzero, FALSE otherwise
2372  * \note This differs from simply using node->details->shutdown in that it can
2373  *       be used before that has been determined (and in fact to determine it),
2374  *       and it can also be used to distinguish requested shutdown from implicit
2375  *       shutdown of remote nodes by virtue of their connection stopping.
2376  */
2377 bool
2378 pe__shutdown_requested(pe_node_t *node)
     /*  */
2379 {
2380     const char *shutdown = pe_node_attribute_raw(node, XML_CIB_ATTR_SHUTDOWN);
2381 
2382     return !pcmk__str_eq(shutdown, "0", pcmk__str_null_matches);
2383 }
2384 
2385 /*!
2386  * \internal
2387  * \brief Update a data set's "recheck by" time
2388  *
2389  * \param[in]     recheck   Epoch time when recheck should happen
2390  * \param[in,out] data_set  Current working set
2391  */
2392 void
2393 pe__update_recheck_time(time_t recheck, pe_working_set_t *data_set)
     /*  */
2394 {
2395     if ((recheck > get_effective_time(data_set))
2396         && ((data_set->recheck_by == 0)
2397             || (data_set->recheck_by > recheck))) {
2398         data_set->recheck_by = recheck;
2399     }
2400 }
2401 
2402 /*!
2403  * \internal
2404  * \brief Wrapper for pe_unpack_nvpairs() using a cluster working set
2405  */
2406 void
2407 pe__unpack_dataset_nvpairs(xmlNode *xml_obj, const char *set_name,
     /*  */
2408                            pe_rule_eval_data_t *rule_data, GHashTable *hash,
2409                            const char *always_first, gboolean overwrite,
2410                            pe_working_set_t *data_set)
2411 {
2412     crm_time_t *next_change = crm_time_new_undefined();
2413 
2414     pe_eval_nvpairs(data_set->input, xml_obj, set_name, rule_data, hash,
2415                     always_first, overwrite, next_change);
2416     if (crm_time_is_defined(next_change)) {
2417         time_t recheck = (time_t) crm_time_get_seconds_since_epoch(next_change);
2418 
2419         pe__update_recheck_time(recheck, data_set);
2420     }
2421     crm_time_free(next_change);
2422 }
2423 
2424 bool
2425 pe__resource_is_disabled(pe_resource_t *rsc)
     /*  */
2426 {
2427     const char *target_role = NULL;
2428 
2429     CRM_CHECK(rsc != NULL, return false);
2430     target_role = g_hash_table_lookup(rsc->meta, XML_RSC_ATTR_TARGET_ROLE);
2431     if (target_role) {
2432         enum rsc_role_e target_role_e = text2role(target_role);
2433 
2434         if ((target_role_e == RSC_ROLE_STOPPED)
2435             || ((target_role_e == RSC_ROLE_UNPROMOTED)
2436                 && pcmk_is_set(uber_parent(rsc)->flags, pe_rsc_promotable))) {
2437             return true;
2438         }
2439     }
2440     return false;
2441 }
2442 
2443 /*!
2444  * \internal
2445  * \brief Create an action to clear a resource's history from CIB
2446  *
2447  * \param[in] rsc   Resource to clear
2448  * \param[in] node  Node to clear history on
2449  *
2450  * \return New action to clear resource history
2451  */
2452 pe_action_t *
2453 pe__clear_resource_history(pe_resource_t *rsc, pe_node_t *node,
     /*  */
2454                            pe_working_set_t *data_set)
2455 {
2456     char *key = NULL;
2457 
2458     CRM_ASSERT(rsc && node);
2459     key = pcmk__op_key(rsc->id, CRM_OP_LRM_DELETE, 0);
2460     return custom_action(rsc, key, CRM_OP_LRM_DELETE, node, FALSE, TRUE,
2461                          data_set);
2462 }
2463 
2464 bool
2465 pe__rsc_running_on_any(pe_resource_t *rsc, GList *node_list)
     /*  */
2466 {
2467     for (GList *ele = rsc->running_on; ele; ele = ele->next) {
2468         pe_node_t *node = (pe_node_t *) ele->data;
2469         if (pcmk__str_in_list(node->details->uname, node_list,
2470                               pcmk__str_star_matches|pcmk__str_casei)) {
2471             return true;
2472         }
2473     }
2474 
2475     return false;
2476 }
2477 
2478 bool
2479 pcmk__rsc_filtered_by_node(pe_resource_t *rsc, GList *only_node)
     /*  */
2480 {
2481     return (rsc->fns->active(rsc, FALSE) && !pe__rsc_running_on_any(rsc, only_node));
2482 }
2483 
2484 GList *
2485 pe__filter_rsc_list(GList *rscs, GList *filter)
     /*  */
2486 {
2487     GList *retval = NULL;
2488 
2489     for (GList *gIter = rscs; gIter; gIter = gIter->next) {
2490         pe_resource_t *rsc = (pe_resource_t *) gIter->data;
2491 
2492         /* I think the second condition is safe here for all callers of this
2493          * function.  If not, it needs to move into pe__node_text.
2494          */
2495         if (pcmk__str_in_list(rsc_printable_id(rsc), filter, pcmk__str_star_matches) ||
2496             (rsc->parent && pcmk__str_in_list(rsc_printable_id(rsc->parent), filter, pcmk__str_star_matches))) {
2497             retval = g_list_prepend(retval, rsc);
2498         }
2499     }
2500 
2501     return retval;
2502 }
2503 
2504 GList *
2505 pe__build_node_name_list(pe_working_set_t *data_set, const char *s) {
     /*  */
2506     GList *nodes = NULL;
2507 
2508     if (pcmk__str_eq(s, "*", pcmk__str_null_matches)) {
2509         /* Nothing was given so return a list of all node names.  Or, '*' was
2510          * given.  This would normally fall into the pe__unames_with_tag branch
2511          * where it will return an empty list.  Catch it here instead.
2512          */
2513         nodes = g_list_prepend(nodes, strdup("*"));
2514     } else {
2515         pe_node_t *node = pe_find_node(data_set->nodes, s);
2516 
2517         if (node) {
2518             /* The given string was a valid uname for a node.  Return a
2519              * singleton list containing just that uname.
2520              */
2521             nodes = g_list_prepend(nodes, strdup(s));
2522         } else {
2523             /* The given string was not a valid uname.  It's either a tag or
2524              * it's a typo or something.  In the first case, we'll return a
2525              * list of all the unames of the nodes with the given tag.  In the
2526              * second case, we'll return a NULL pointer and nothing will
2527              * get displayed.
2528              */
2529             nodes = pe__unames_with_tag(data_set, s);
2530         }
2531     }
2532 
2533     return nodes;
2534 }
2535 
2536 GList *
2537 pe__build_rsc_list(pe_working_set_t *data_set, const char *s) {
     /*  */
2538     GList *resources = NULL;
2539 
2540     if (pcmk__str_eq(s, "*", pcmk__str_null_matches)) {
2541         resources = g_list_prepend(resources, strdup("*"));
2542     } else {
2543         pe_resource_t *rsc = pe_find_resource_with_flags(data_set->resources, s,
2544                                                          pe_find_renamed|pe_find_any);
2545 
2546         if (rsc) {
2547             /* A colon in the name we were given means we're being asked to filter
2548              * on a specific instance of a cloned resource.  Put that exact string
2549              * into the filter list.  Otherwise, use the printable ID of whatever
2550              * resource was found that matches what was asked for.
2551              */
2552             if (strstr(s, ":") != NULL) {
2553                 resources = g_list_prepend(resources, strdup(rsc->id));
2554             } else {
2555                 resources = g_list_prepend(resources, strdup(rsc_printable_id(rsc)));
2556             }
2557         } else {
2558             /* The given string was not a valid resource name.  It's either
2559              * a tag or it's a typo or something.  See build_uname_list for
2560              * more detail.
2561              */
2562             resources = pe__rscs_with_tag(data_set, s);
2563         }
2564     }
2565 
2566     return resources;
2567 }
2568 
2569 xmlNode *
2570 pe__failed_probe_for_rsc(pe_resource_t *rsc, const char *name)
     /*  */
2571 {
2572     pe_resource_t *parent = uber_parent(rsc);
2573     const char *rsc_id = rsc->id;
2574 
2575     if (rsc->variant == pe_clone) {
2576         rsc_id = pe__clone_child_id(rsc);
2577     } else if (parent->variant == pe_clone) {
2578         rsc_id = pe__clone_child_id(parent);
2579     }
2580 
2581     for (xmlNode *xml_op = pcmk__xml_first_child(rsc->cluster->failed); xml_op != NULL;
2582          xml_op = pcmk__xml_next(xml_op)) {
2583         const char *value = NULL;
2584         char *op_id = NULL;
2585 
2586         /* This resource operation is not a failed probe. */
2587         if (!pcmk_xe_mask_probe_failure(xml_op)) {
2588             continue;
2589         }
2590 
2591         /* This resource operation was not run on the given node.  Note that if name is
2592          * NULL, this will always succeed.
2593          */
2594         value = crm_element_value(xml_op, XML_LRM_ATTR_TARGET);
2595         if (value == NULL || !pcmk__str_eq(value, name, pcmk__str_casei|pcmk__str_null_matches)) {
2596             continue;
2597         }
2598 
2599         /* This resource operation has no operation_key. */
2600         value = crm_element_value(xml_op, XML_LRM_ATTR_TASK_KEY);
2601         if (!parse_op_key(value ? value : ID(xml_op), &op_id, NULL, NULL)) {
2602             continue;
2603         }
2604 
2605         /* This resource operation's ID does not match the rsc_id we are looking for. */
2606         if (!pcmk__str_eq(op_id, rsc_id, pcmk__str_none)) {
2607             free(op_id);
2608             continue;
2609         }
2610 
2611         free(op_id);
2612         return xml_op;
2613     }
2614 
2615     return NULL;
2616 }