wip: add support for workflow events (#27)

* wip: add support for workflow events

This will add support for ending the workflow early due
to a count of successes, failures, or job duration metric.
We need to next add ability to grow or shrink (need to think
about how to do that, since we want a cloud agnostic solution)
and then how to handle application specific metrics

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

* feat: add support for minicluster

If we really want to test scale (shrink and grow) of a job
and have it work with the cluster autoscaler, plus collecting
metrics from an HPC app, we can most easily do that with
the flux operator. This feature adds support for specifying
a minicluster property to convert the previous indexed job
into a MiniCluster. The flux operator needs to be installed.

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

* feat: shrink with flux minicluster example working.

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

* save state

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

* feat: support for custom metrics

In this example, the user is allowed to provide a custom script
that will be used against the log, and it needs to return a dictionary
of values (the custom metrics). These are passed back to the manager
from the state machine step and can influence workflow behavior (e.g.,
stop early, grow, or shrink.

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

---------

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

Author: vsoch

api/v1alpha1/statemachine_types.go

@@ -70,7 +70,27 @@ type Workflow struct {
 	// Prefix for jobs (e.g., structure_ for mummi)
 	Prefix string `json:"prefix,omitempty"`
 
-	// TODO add a failure condition?
+	// Custom events  -> actions to take
+	Events []WorkflowEvent `json:"events,omitempty"`
+}
+
+type WorkflowEvent struct {
+
+	// Name of metric, indexed into model lookup (e.g., count.job_b.failed)
+	Metric string `json:"metric"`
+
+	// Conditional to check. If not set, checks if nonzero or nonempty
+	When string `json:"when,omitempty"`
+
+	// Action to take (e.g., finish-workflow)
+	Action string `json:"action"`
+
+	// Backoff and repetitions to respond to event
+	Backoff     int32 `json:"backoff,omitempty"`
+	Repetitions int32 `json:"repetitions,omitempty"`
+	MinCompletions int32 `json:"minCompletions,omitempty"`
+	MaxSize int32 `json:"maxSize,omitempty"`
+	MinSize int32 `json:"minSize,omitempty"`
 }
 
 // A JobSequence is a list of JobSteps
@@ -79,6 +99,12 @@ type Workflow struct {
 // to the job (not knowing the structure in advance)
 type JobSequence []JobStep
 
+type JobEvents struct {
+
+	// Custom parsing script
+	Script string `json:"script,omitempty"`
+}
+
 type JobStep struct {
 
 	// Name is the name of the job (required)
@@ -92,6 +118,10 @@ type JobStep struct {
 	// +optional
 	Registry RegistryConfig `json:"registry,omitempty"`
 
+	// Event for a job
+	// +optional
+	Events JobEvents `json:"events,omitempty"`
+
 	// Architecture (arm64 or amd64)
 	// +kubebuilder:default="amd64"
 	// +default="amd64"
@@ -215,6 +245,10 @@ type Manager struct {
 	// +omitempty
 	NodeSelector string `json:"nodeSelector,omitempty"`
 
+	// Run in more verbose mode
+	// +optional
+	Verbose bool `json:"verbose"`
+
 	// Image pull policy (e.g., Always, Never, etc.)
 	// +kubebuilder:default="IfNotPresent"
 	// +default="IfNotPresent"

config/crd/bases/state-machine.converged-computing.org_statemachines.yaml

@@ -105,6 +105,13 @@ spec:
                           description: Walltime (in string format) for the job
                           type: string
                       type: object
+                    events:
+                      description: Event for a job
+                      properties:
+                        script:
+                          description: Custom parsing script
+                          type: string
+                      type: object
                     image:
                       description: |-
                         Namespace is inherited from StateMachine Spec
@@ -185,6 +192,9 @@ spec:
                     default: r
                     description: Subdomain to use
                     type: string
+                  verbose:
+                    description: Run in more verbose mode
+                    type: boolean
                 type: object
               registry:
                 description: |-
@@ -239,6 +249,42 @@ spec:
                     description: Number of state machine sequences required for completion
                     format: int32
                     type: integer
+                  events:
+                    description: Custom events  -> actions to take
+                    items:
+                      properties:
+                        action:
+                          description: Action to take (e.g., finish-workflow)
+                          type: string
+                        backoff:
+                          description: Backoff and repetitions to respond to event
+                          format: int32
+                          type: integer
+                        maxSize:
+                          format: int32
+                          type: integer
+                        metric:
+                          description: Name of metric, indexed into model lookup (e.g.,
+                            count.job_b.failed)
+                          type: string
+                        minCompletions:
+                          format: int32
+                          type: integer
+                        minSize:
+                          format: int32
+                          type: integer
+                        repetitions:
+                          format: int32
+                          type: integer
+                        when:
+                          description: Conditional to check. If not set, checks if
+                            nonzero or nonempty
+                          type: string
+                      required:
+                      - action
+                      - metric
+                      type: object
+                    type: array
                   prefix:
                     description: Prefix for jobs (e.g., structure_ for mummi)
                     type: string

config/rbac/role.yaml

@@ -10,6 +10,10 @@ rules:
   - events
   verbs:
   - create
+  - delete
+  - get
+  - list
+  - patch
   - update
   - watch
 - apiGroups:
@@ -25,6 +29,42 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - miniclusters
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - miniclusters/finalizers
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - miniclusters/status
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - apps
   resources:
@@ -292,6 +332,54 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - events.k8s.io
+  resources:
+  - events
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - flux-framework.org
+  resources:
+  - miniclusters
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - flux-framework.org
+  resources:
+  - miniclusters/finalizers
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - flux-framework.org
+  resources:
+  - miniclusters/status
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - networking.k8s.io
   resources: