1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
package policytrigger
import (
"context"
"errors"
"fmt"
"log/slog"
"sync"
policyv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/policy/v1alpha1"
ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
pecontroller "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/controller"
)
const (
Name = "policytrigger"
Export = "policy"
)
type ActionsFunc func(workspace string) (map[string]pecontroller.ActionHandler, error)
type Options struct {
Actions ActionsFunc
Log *slog.Logger
}
type Controller struct {
actions ActionsFunc
log *slog.Logger
mu sync.Mutex
handlers map[string]map[string]pecontroller.ActionHandler
triggers map[string]map[string]struct{}
runs map[string]map[string]*policyv1alpha1.PolicyWorkflowRun
}
func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
log := opts.Log
if log == nil {
log = slog.New(slog.DiscardHandler)
}
c := &Controller{actions: opts.Actions, log: log.With("component", "policytrigger-controller"),
handlers: map[string]map[string]pecontroller.ActionHandler{}, triggers: map[string]map[string]struct{}{},
runs: map[string]map[string]*policyv1alpha1.PolicyWorkflowRun{}}
if err := ctlruntime.For[*policyv1alpha1.PolicyTrigger](m, Name, c, ctlruntime.Export(Export),
ctlruntime.Watches[*policyv1alpha1.PolicyWorkflowRun](c.runChanged)); err != nil {
return nil, fmt.Errorf("policytrigger: registering the PolicyTrigger controller: %w", err)
}
return c, nil
}
func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, trigger *policyv1alpha1.PolicyTrigger) (ctlruntime.Result, error) {
c.track(req.Cluster, trigger.Name, trigger.DeletionTimestamp.IsZero())
handlers, err := c.handlersFor(req)
if err != nil {
return ctlruntime.Result{}, err
}
reconciler := &pecontroller.TriggerReconciler{Handlers: handlers}
state, err := reconciler.ReconcilePolicyTrigger(ctx, trigger)
if err != nil {
return ctlruntime.Result{}, err
}
state.LastRun = trigger.Status.LastRun
state.NextRun = trigger.Status.NextRun
if !pecontroller.TriggerStatusEqual(trigger.Status, state) {
if err := c.apply(ctx, req, trigger, state); err != nil {
return ctlruntime.Result{}, err
}
}
runs := c.finished(req.Cluster)
if len(runs) == 0 {
return ctlruntime.Result{}, nil
}
dispatchErr := reconciler.DispatchRuns(ctx, trigger, runs, func(state pecontroller.TriggerReconcileState) error {
return c.apply(ctx, req, trigger, state)
})
if dispatchErr != nil {
c.log.Warn("dispatching a policy trigger", "trigger", trigger.Name, "workspace", req.Workspace, "err", dispatchErr.Error())
}
return ctlruntime.Result{}, dispatchErr
}
func (c *Controller) apply(ctx context.Context, req ctlruntime.Request, trigger *policyv1alpha1.PolicyTrigger, state pecontroller.TriggerReconcileState) error {
object := trigger.DeepCopy()
object.Status = policyv1alpha1.PolicyTriggerStatus{LastRun: state.LastRun, NextRun: state.NextRun, Active: state.Active, Conditions: state.Conditions}
if err := req.ApplyStatus(ctx, object); err != nil {
return err
}
trigger.Status = object.Status
return nil
}
func (c *Controller) handlersFor(req ctlruntime.Request) (map[string]pecontroller.ActionHandler, error) {
c.mu.Lock()
handlers, ok := c.handlers[req.Cluster]
c.mu.Unlock()
if ok {
return handlers, nil
}
if c.actions == nil {
return nil, nil
}
handlers, err := c.actions(req.Workspace)
if err != nil {
return nil, errors.Join(errors.New("policytrigger: the trigger action handlers"), err)
}
c.mu.Lock()
c.handlers[req.Cluster] = handlers
c.mu.Unlock()
return handlers, nil
}
func (c *Controller) track(cluster, name string, live bool) {
c.mu.Lock()
defer c.mu.Unlock()
if !live {
delete(c.triggers[cluster], name)
return
}
if c.triggers[cluster] == nil {
c.triggers[cluster] = map[string]struct{}{}
}
c.triggers[cluster][name] = struct{}{}
}
func (c *Controller) finished(cluster string) []*policyv1alpha1.PolicyWorkflowRun {
c.mu.Lock()
defer c.mu.Unlock()
out := make([]*policyv1alpha1.PolicyWorkflowRun, 0, len(c.runs[cluster]))
for _, run := range c.runs[cluster] {
out = append(out, run)
}
return out
}
func (c *Controller) runChanged(_ context.Context, src ctlruntime.Request, run *policyv1alpha1.PolicyWorkflowRun) []ctlruntime.Request {
c.mu.Lock()
defer c.mu.Unlock()
if !run.DeletionTimestamp.IsZero() || !pecontroller.Finished(run) {
delete(c.runs[src.Cluster], run.Name)
return nil
}
if c.runs[src.Cluster] == nil {
c.runs[src.Cluster] = map[string]*policyv1alpha1.PolicyWorkflowRun{}
}
c.runs[src.Cluster][run.Name] = run.DeepCopy()
out := make([]ctlruntime.Request, 0, len(c.triggers[src.Cluster]))
for name := range c.triggers[src.Cluster] {
out = append(out, ctlruntime.Request{Cluster: src.Cluster, Name: name})
}
return out
}
