example: heartbeat example

This updates the heartbeat so it is entirely derived from
the config. This can happen explicitly if the user sets
logging->heartbeat to a non zero value, but it will also
happen if there is a grow or shrink action used. If the user
defines a grow/shrink and sets the heartbeat to 0 it will
still be set to the default, 60, because grow/shrink will
not work as expected without it.

Signed-off-by: vsoch <[email protected]>

Author: vsoch

README.md

@@ -32,6 +32,27 @@ This design will be translated into more consolidated design documentation. For
 - **Plugins**: A plugin is a collection of custom actions that are typically associated with a particular application. For example, a plugin for LAMMPS might know how to check LAMMPS output and act on a specific parsing of a result. Plugins are used equivalently to custom functions, and can accept arguments.
 - **Metrics** are summary metrics collected for groups of jobs, for customized algorithms. To support this, we use online (or streaming) ML algorithms for things like mean, IQR, etc. While there could be a named entity called an algorithm, since it's just a set of rules (triggers and actions) that means we don't need to explicitly define them (yet). I can see at some point creating "packaged" rule sets that are called that.
 
+
+#### Logging
+
+By default, we have a pretty printed set of the "main" events (triggers and submissions). However, to see ALL events from Flux (good for debugging) you can turn them on:
+
+```yaml
+logging:
+  debug: true
+```
+
+To set the event heartbeat to fire at some increment, set it:
+
+```yaml
+logging:
+  debug: true
+  heartbeat: 60
+```
+
+Note that by default it is turned off (set to 0 seconds) unless you include a grow or shrink action. In that case, it turns on and defaults to 60, unless you've specified another interval.
+If you have grow/shrink and explicitly turn it off, it will still default to 60 seconds, because grow/shrink won't work as expected without the heartbeat.
+
 #### Rules
 
 A rule defines a trigger and action to take. The library is event driven, meaning that the queue is expected to send events, and we don't do any polling.
@@ -79,9 +100,11 @@ The design of a rule is to have an action, and the action is something your ense
 For the scale operations, since this brings in the issue of resource contention between different ensembles, we have to assume to start that a request to scale:
 
 1. Should only happen once. If it's granted, great, if not, we aren't going to ask again.
-2. Does not need to consider another level of scheduler (e.g., fair shaire)
+2. Does not need to consider another level of scheduler (e.g., fair share)
 
 I started thinking about putting a second scheduler (or fair share algorithm) in the grpc service, but realized this was another level of complexity that although we might work on it later, is not warranted yet.
+Also note that since scale operation triggers might not be linked to job events (e.g., if we want to trigger when a job group has been in the queue for too long) we added support for a heartbeat. The heartbeat
+isn't a trigger in and of itself, but when it runs, it will run through rules that are relevant to queue metrics.
 We see "submit" as two examples in the above, which is a common thing you'd want to do! For each action, you should minimally define the "name" and a "label" that typically corresponds to a job group.
 You can also optionally define "repetitions," which are the number of times the action should be run before expiring. If you want a backoff period between repetitions, set "backoff" to a non zero value.
 By default, when no repetitions or backoff are set, the action is assumed to have a repetition of 1. It will be run once! Let's now look at a custom action. Here is what your function should look like in your `ensemble.yaml`
@@ -177,6 +200,10 @@ ensemble run examples/custom-action-example.yaml
 
 # This shows termination, which is necessary for when you want an exit
 ensemble run examples/terminate-example.yaml
+
+# Run a heartbeat every 3 seconds.
+# This will trigger a check to see if actions need to be performed
+ensemble run examples/heartbeat-example.yaml
 ```
 
 Right now, this will run any rules with "start" triggers, which for this hello world example includes a few hello world jobs! You'll then be able to watch and see flux events coming in!

ensemble/config/config.py

@@ -5,6 +5,7 @@
 
 import jsonschema
 
+import ensemble.defaults as defaults
 import ensemble.utils as utils
 from ensemble import schema
 from ensemble.config.types import Rule
@@ -14,6 +15,9 @@
 script_template = """from ensemble.config.types import Action, Rule
 """
 
+# These are the actions that warrant the heartbeat
+heartbeat_actions = {"grow", "shrink"}
+
 
 def load_config(config_path):
     """
@@ -34,6 +38,9 @@ def __init__(self, cfg):
         self._cfg = cfg
         self.jobs = {}
         self.rules = {}
+
+        # By default, we don't require a heartbeat
+        self.require_heartbeat = False
         self.parse()
 
         # Cache of action names
@@ -43,6 +50,20 @@ def __init__(self, cfg):
     def debug_logging(self):
         return self._cfg.get("logging", {}).get("debug") is True
 
+    @property
+    def heartbeat(self):
+        """
+        Get the heartbeat seconds.
+
+        If heartbeat actions are defined and no heartbeat is set, we require
+        it and default to 60. Otherwise, we allow it unset (0) or a user
+        specified value.
+        """
+        heartbeat = self._cfg.get("logging", {}).get("heartbeat") or 0
+        if not heartbeat and self.require_heartbeat:
+            heartbeat = defaults.heartbeat_seconds
+        return heartbeat
+
     def pretty_job(self, name):
         """
         Pretty print a job (for the logger) across a single line
@@ -105,6 +126,10 @@ def parse(self):
         for rule in self._cfg["rules"]:
             rule = Rule(rule, self.custom)
 
+            # If the rule action is in the heartbeat set, we require heartbeat
+            if rule.action.name in heartbeat_actions:
+                self.require_heartbeat = True
+
             # Group rules with common trigger together
             if rule.trigger not in self.rules:
                 self.rules[rule.trigger] = []

ensemble/defaults.py

@@ -2,6 +2,7 @@
 port = 50051
 
 valid_actions = ["submit", "custom", "terminate"]
+heartbeat_seconds = 60
 
 job_events = [
     "job-depend",

ensemble/members/flux/queue.py

@@ -63,7 +63,6 @@ def __init__(
         include_inactive=False,
         refresh_interval=0,
         summary_frequency=10,
-        heartbeat_seconds=60,
         **kwargs,
     ):
         """
@@ -73,7 +72,6 @@ def __init__(
         job_filters (dict)     : key value pairs for attributes or states to filter jobs to
         include_inactive (bool): consider and include inactive jobs when doing initial parse
         refresh_interval (int) : number of seconds to refresh all metrics (0 indicates no refresh)
-        heartbeat_seconds (int): number of seconds to heartbeat (for grow/shrink). This defaults
         to 60 seconds. If you set to 0, it will not be set.
         """
         # Filter jobs to any attributes, states, or similar.
@@ -85,7 +83,6 @@ def __init__(
         # How often on job completions to summarize?
         self.summary_freqency = summary_frequency
         self.completion_counter = 0
-        self.heartbeat_seconds = heartbeat_seconds
 
         # The flux sentinel tells us when we have finished with the backlog
         # https://github.com/flux-framework/flux-core/blob/master/src/modules/job-manager/journal.c#L28-L33
@@ -104,7 +101,7 @@ def terminate(self):
         Custom termination function for flux.
         """
         self.handle.reactor_stop()
-        if self.heartbeat_seconds:
+        if self.cfg.heartbeat:
             self.heartbeat.stop()
 
     def record_metrics(self, record):
@@ -283,15 +280,15 @@ def setup_heartbeat(self):
         """
         Setup the heartbeat - a threading.Thread
         """
-        if not self.heartbeat_seconds:
+        if not self.cfg.heartbeat:
             return
 
         def heartbeat_callback(cls):
             print("💗 HEARTBEAT")
             print(cls)
 
         # Instead we are using threading, which works!
-        self.heartbeat = QueueHeartbeat(self.heartbeat_seconds, heartbeat_callback, cls=self)
+        self.heartbeat = QueueHeartbeat(self.cfg.heartbeat, heartbeat_callback, cls=self)
         self.heartbeat.start()
 
     def cron_heartbeat(self):
@@ -305,14 +302,14 @@ def heartbeat_callback(response):
 
         # Create a cron heartbeat every N seconds, only if we have a heartbeat set
         # This is intended for grow/shrink actions that might need a regular check
-        print(f"  💗 Creating flux heartbeat every {self.heartbeat_seconds} seconds")
+        print(f"  💗 Creating flux heartbeat every {self.cfg.heartbeat} seconds")
         heartbeat = self.handle.rpc(
             "cron.create",
             {
                 "type": "interval",
                 "name": "heartbeat",
                 "command": "sleep 0",
-                "args": {"interval": self.heartbeat_seconds},
+                "args": {"interval": self.cfg.heartbeat},
             },
             flux.constants.FLUX_NODEID_ANY,
             flags=flux.constants.FLUX_RPC_STREAMING,

ensemble/schema.py

@@ -19,7 +19,10 @@
         "custom": {"type": "string"},
         "logging": {
             "type": "object",
-            "properties": {"debug": {"type": "boolean", "default": False}},
+            "properties": {
+                "debug": {"type": "boolean", "default": False},
+                "heartbeat": {"type": "number"},
+            },
             "additionalProperties": False,
         },
         "jobs": {

examples/heartbeat-example.yaml

@@ -0,0 +1,33 @@
+# No verbose logging, but turn on the heartbeat every 3 seconds
+logging:
+  debug: false
+  heartbeat: 3
+
+# I made these the size of the cluster so we trigger scaling at least once
+# We currently assume these are uniform (so no order enforced beyond the listing here)
+jobs:
+  - name: echo
+    command: echo hello world
+    count: 5
+    nodes: 1
+  - name: sleep
+    command: sleep 10
+    count: 5
+    nodes: 1
+
+rules:
+  # 1. This rule says to submit the sleep jobs when we start
+  - trigger: start
+    action:
+      name: submit
+      label: sleep
+
+  # This says to submit the echo hello world jobs when we have 3
+  # successfully completed sleep jobs. This dot notation is a path into the
+  # models data structure.
+  - trigger: metric
+    name: count.sleep.success
+    when: 3
+    action:
+      name: submit
+      label: echo