diff --git a/README.md b/README.md index bf2bdb4..90c3719 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ The `viam:sensor-bundle` module provides these models: 1. **`viam:sensor-bundle:stateful-sensor`** - A sensor component that holds a value you set via `DoCommand`, serves it through the Readings API, and persists it to a file on disk so it survives restarts. -2. **`viam:sensor-bundle:sensor-monitor`** - A sensor component that watches another sensor's readings against numeric trigger rules and sends a notification via a generic service's `DoCommand` when a threshold is crossed. +2. **`viam:sensor-bundle:sensor-monitor`** - A sensor component that watches another sensor's readings against numeric trigger rules and fires `DoCommand` actions on other resources when a rule triggers or resolves. --- @@ -69,22 +69,26 @@ Returns: **API:** `rdk:component:sensor` -Monitors the readings of another sensor and fires notifications when a numeric reading crosses a configured threshold. It takes two dependencies — the sensor to watch and a generic service to notify — and evaluates a set of numeric trigger rules on every poll. +Monitors the readings of another sensor and, when a numeric reading crosses a configured threshold, fires `DoCommand` **actions** on other resources. It evaluates a set of numeric trigger rules on every poll; the only dependency is the sensor to watch (plus whatever resources the actions target). -When a rule fires, the monitor calls `DoCommand` on the notifier with the payload: +**Actions.** Each rule has an `on_trigger` and an `on_resolve` list. Every action names a `resource` (any component or service on the machine, by name) and a `command` (the `DoCommand` payload sent to it). -```json -{"command": "send", "text": ""} -``` +- `on_trigger` fires when the rule transitions from not-triggered to triggered, and again at most once per `cooldown_seconds` while it stays triggered (with the default `cooldown_seconds: 0`, that's edge-only). +- `on_resolve` fires once when the reading clears the threshold. +- Actions are best-effort: a failure is logged and never affects monitoring. Each named resource is automatically added as a dependency. + +**References and captures.** String values in a `command` may embed `${...}` references, resolved just before the command is sent: -Notifications are **edge-triggered**: a message is sent when a rule transitions from not-triggered to triggered. While the rule stays triggered no further message is sent, unless `cooldown_seconds` is set, in which case the message repeats at most once per cooldown window. When the reading clears the threshold and crosses it again, a new notification is sent. +- Rule/reading context: `${value}` (the current reading), `${key}`, `${threshold}`, `${operator}`. +- Captured responses: set `"capture": ""` on an action to store its `DoCommand` response, then reference its fields from later actions (including `on_resolve`) as `${.}`. + +A value that is exactly one reference keeps the referenced value's type; a reference embedded in a larger string is substituted as text. If a reference can't be resolved (e.g. nothing was captured), that action is skipped. A capture is overwritten on each cooldown re-fire (so resolve references the most recent response) and cleared once the rule resolves. ### Configuration ```json { "sensor": "", - "notifier": "", "poll_interval_seconds": , "cooldown_seconds": , "rules": [ @@ -92,7 +96,8 @@ Notifications are **edge-triggered**: a message is sent when a rule transitions "key": "", "operator": "", "threshold": , - "message": "" + "on_trigger": [{ "resource": "", "command": { }, "capture": "" }], + "on_resolve": [{ "resource": "", "command": { } }] } ] } @@ -101,10 +106,9 @@ Notifications are **edge-triggered**: a message is sent when a rule transitions | Name | Type | Required | Description | | ----------------------- | ------ | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | | `sensor` | string | **Yes** | Name of the sensor dependency whose readings are monitored. | -| `notifier` | string | **Yes** | Name of the generic service dependency that receives notification `DoCommand`s. | | `rules` | array | **Yes** | One or more numeric trigger rules (see below). At least one rule is required. | | `poll_interval_seconds` | number | No | How often the sensor is polled, in seconds. Defaults to `10`. | -| `cooldown_seconds` | number | No | Minimum time between repeat notifications while a rule stays triggered. `0` (default) means no repeat until the rule clears and fires again. | +| `cooldown_seconds` | number | No | Minimum time between repeat `on_trigger` firings while a rule stays triggered. `0` (default) means fire only on the edge. | Each entry in `rules`: @@ -113,27 +117,43 @@ Each entry in `rules`: | `key` | string | **Yes** | The reading key to watch, e.g. `"temperature"`. Non-numeric or missing values are skipped. | | `operator` | string | **Yes** | Comparison applied between the reading and `threshold`. One of `>`, `>=`, `<`, `<=`, `==`, `!=` (aliases: `gt`, `gte`, `lt`, `lte`, `eq`, `ne`). | | `threshold` | number | **Yes** | The value the reading is compared against. | -| `message` | string | No | Notification template. Supports placeholders `{key}`, `{value}`, `{threshold}`, `{operator}`. If omitted, a message like `temperature is 95 (> 90)` is used. | +| `on_trigger` | array | No | Actions to fire when the rule triggers (and on each cooldown window). Each is `{ "resource": , "command": , "capture": }`. | +| `on_resolve` | array | No | Actions to fire when the rule clears. Same shape as `on_trigger`. | ### Example Configuration +A low-bean alert that posts to Slack on trigger and adds a ✅ reaction to that same message on resolve, plus a fan toggled directly: + ```json { - "sensor": "outdoor-temp", - "notifier": "slack-notifier", + "sensor": "usage-sensor", "poll_interval_seconds": 30, "cooldown_seconds": 3600, "rules": [ + { + "key": "regular_grinds", + "operator": ">=", + "threshold": 20, + "on_trigger": [ + { + "resource": "slack-notifier", + "command": { "command": "send", "text": "☕ Regular beans are low (${value}), please refill." }, + "capture": "msg" + } + ], + "on_resolve": [ + { + "resource": "slack-notifier", + "command": { "command": "react", "name": "white_check_mark", "ts": "${msg.ts}", "channel": "${msg.channel}" } + } + ] + }, { "key": "temperature", "operator": ">", "threshold": 90, - "message": "High temperature alert: {key} is {value} (limit {threshold})" - }, - { - "key": "humidity", - "operator": "<", - "threshold": 20 + "on_trigger": [{ "resource": "cooling-fan", "command": { "set": { "power": 1 } } }], + "on_resolve": [{ "resource": "cooling-fan", "command": { "set": { "power": 0 } } }] } ] } @@ -154,7 +174,7 @@ Returns the most recent readings from the monitored sensor, plus a `_trigge ### DoCommand -**`check`** - Force an immediate poll of the sensor (instead of waiting for the next interval), evaluating all rules and sending any notifications. +**`check`** - Force an immediate poll of the sensor (instead of waiting for the next interval), evaluating all rules and firing any actions. ```json {"check": true} diff --git a/resources/sensormonitor/sensormonitor.go b/resources/sensormonitor/sensormonitor.go index e91f601..fe49f31 100644 --- a/resources/sensormonitor/sensormonitor.go +++ b/resources/sensormonitor/sensormonitor.go @@ -1,12 +1,14 @@ // Package sensormonitor implements the viam:sensor-bundle:sensor-monitor model: a // sensor that watches another sensor's readings against numeric trigger rules and -// sends a notification via a generic service's DoCommand when a threshold is crossed. +// fires configurable DoCommand actions on other resources when a rule triggers or +// resolves. package sensormonitor import ( "context" "encoding/json" "fmt" + "regexp" "strconv" "strings" "sync" @@ -15,12 +17,11 @@ import ( sensor "go.viam.com/rdk/components/sensor" "go.viam.com/rdk/logging" "go.viam.com/rdk/resource" - generic "go.viam.com/rdk/services/generic" ) // Model is the sensor-monitor model triplet. It watches the readings of a sensor -// and fires notifications through a generic service when a numeric reading crosses -// a configured threshold. +// and fires DoCommand actions on other resources when a numeric reading crosses a +// configured threshold. var Model = resource.NewModel("viam", "sensor-bundle", "sensor-monitor") func init() { @@ -34,6 +35,18 @@ func init() { // defaultPollInterval is used when poll_interval_seconds is not set. const defaultPollInterval = 10 * time.Second +// Action is a DoCommand to fire on a resource when a rule changes state. +type Action struct { + // Resource is the name of the resource to call. + Resource string `json:"resource"` + // Command is the DoCommand payload. String values may contain ${...} + // references that are resolved before the command is sent — see resolveValue. + Command map[string]interface{} `json:"command"` + // Capture, when set, stores this action's response under this name so later + // actions can reference its fields as ${name.field}. + Capture string `json:"capture,omitempty"` +} + // Rule describes a single numeric trigger on one reading key. type Rule struct { // Key is the reading key to watch, e.g. "temperature". @@ -43,25 +56,24 @@ type Rule struct { Operator string `json:"operator"` // Threshold is the value the reading is compared against. Threshold float64 `json:"threshold"` - // Message is an optional notification template. It supports the placeholders - // {key}, {value}, {threshold} and {operator}. If empty, a message is generated. - Message string `json:"message,omitempty"` + // OnTrigger lists actions to fire when the rule transitions to triggered, and + // again on each cooldown window while it stays triggered. + OnTrigger []Action `json:"on_trigger,omitempty"` + // OnResolve lists actions to fire when the rule clears (its reading returns to + // the non-triggered side of the threshold). + OnResolve []Action `json:"on_resolve,omitempty"` } // Config is the configuration for the sensor-monitor model. type Config struct { // Sensor is the name of the sensor dependency whose readings are monitored. Sensor string `json:"sensor"` - // Notifier is the name of the generic service dependency that receives - // notification DoCommands of the form {"command": "send", "text": }. - Notifier string `json:"notifier"` // Rules is the set of numeric trigger rules. At least one is required. Rules []Rule `json:"rules"` // PollIntervalSec is how often the sensor is polled, in seconds. Defaults to 10. PollIntervalSec float64 `json:"poll_interval_seconds,omitempty"` - // CooldownSec is the minimum time between repeat notifications while a rule - // stays triggered, in seconds. 0 (default) means do not re-notify until the - // rule clears and fires again. + // CooldownSec is the minimum time between repeat on_trigger firings while a + // rule stays triggered, in seconds. 0 (default) means fire only on the edge. CooldownSec float64 `json:"cooldown_seconds,omitempty"` } @@ -70,12 +82,19 @@ func (cfg *Config) Validate(path string) ([]string, []string, error) { if cfg.Sensor == "" { return nil, nil, fmt.Errorf("%s: missing required field 'sensor'", path) } - if cfg.Notifier == "" { - return nil, nil, fmt.Errorf("%s: missing required field 'notifier'", path) - } if len(cfg.Rules) == 0 { return nil, nil, fmt.Errorf("%s: at least one rule is required", path) } + + deps := []string{cfg.Sensor} + seen := map[string]bool{cfg.Sensor: true} + addDep := func(name string) { + if !seen[name] { + seen[name] = true + deps = append(deps, name) + } + } + for i, r := range cfg.Rules { if r.Key == "" { return nil, nil, fmt.Errorf("%s: rules[%d] missing required field 'key'", path, i) @@ -83,15 +102,41 @@ func (cfg *Config) Validate(path string) ([]string, []string, error) { if _, err := parseOperator(r.Operator); err != nil { return nil, nil, fmt.Errorf("%s: rules[%d] %w", path, i, err) } + for j, a := range r.OnTrigger { + if err := validateAction(a); err != nil { + return nil, nil, fmt.Errorf("%s: rules[%d].on_trigger[%d] %w", path, i, j, err) + } + addDep(a.Resource) + } + for j, a := range r.OnResolve { + if err := validateAction(a); err != nil { + return nil, nil, fmt.Errorf("%s: rules[%d].on_resolve[%d] %w", path, i, j, err) + } + addDep(a.Resource) + } } - return []string{cfg.Sensor, cfg.Notifier}, nil, nil + return deps, nil, nil +} + +// validateAction checks a single action's required fields. +func validateAction(a Action) error { + if a.Resource == "" { + return fmt.Errorf("missing required field 'resource'") + } + if a.Command == nil { + return fmt.Errorf("missing required field 'command'") + } + return nil } // ruleState tracks the runtime state of a single rule across polls. type ruleState struct { - triggered bool - lastNotified time.Time - lastValue float64 + triggered bool + lastFired time.Time + lastValue float64 + // vars holds the responses of actions that set "capture", keyed by capture + // name, so later actions can reference values produced. Reset when the rule resolves. + vars map[string]interface{} } type sensorMonitor struct { @@ -101,8 +146,10 @@ type sensorMonitor struct { logger logging.Logger cfg *Config - sensorDep sensor.Sensor - notifierDep generic.Service + sensorDep sensor.Sensor + // actionResources holds every resource named by a rule action, resolved once + // at construction and keyed by name. + actionResources map[string]resource.Resource pollInterval time.Duration cooldown time.Duration @@ -145,9 +192,31 @@ func newMonitor(deps resource.Dependencies, name resource.Name, conf *Config, lo if err != nil { return nil, fmt.Errorf("failed to get sensor dependency %q: %w", conf.Sensor, err) } - notifierDep, err := generic.FromProvider(deps, conf.Notifier) - if err != nil { - return nil, fmt.Errorf("failed to get notifier dependency %q: %w", conf.Notifier, err) + + // Resolve every resource named by an action up front so firing is a simple map lookup + DoCommand. + actionResources := map[string]resource.Resource{} + resolveAction := func(a Action) error { + if _, ok := actionResources[a.Resource]; ok { + return nil + } + res, err := lookupResource(deps, a.Resource) + if err != nil { + return fmt.Errorf("action resource %q: %w", a.Resource, err) + } + actionResources[a.Resource] = res + return nil + } + for _, r := range conf.Rules { + for _, a := range r.OnTrigger { + if err := resolveAction(a); err != nil { + return nil, err + } + } + for _, a := range r.OnResolve { + if err := resolveAction(a); err != nil { + return nil, err + } + } } pollInterval := defaultPollInterval @@ -159,20 +228,40 @@ func newMonitor(deps resource.Dependencies, name resource.Name, conf *Config, lo cancelCtx, cancelFunc := context.WithCancel(context.Background()) return &sensorMonitor{ - Named: name.AsNamed(), - logger: logger, - cfg: conf, - sensorDep: sensorDep, - notifierDep: notifierDep, - pollInterval: pollInterval, - cooldown: cooldown, - cancelCtx: cancelCtx, - cancelFunc: cancelFunc, - lastReadings: map[string]interface{}{}, - ruleStates: make([]ruleState, len(conf.Rules)), + Named: name.AsNamed(), + logger: logger, + cfg: conf, + sensorDep: sensorDep, + actionResources: actionResources, + pollInterval: pollInterval, + cooldown: cooldown, + cancelCtx: cancelCtx, + cancelFunc: cancelFunc, + lastReadings: map[string]interface{}{}, + ruleStates: make([]ruleState, len(conf.Rules)), }, nil } +// lookupResource finds a dependency by its short name regardless of API, so an +// action can target any resource type. Errors if no dependency — or more than +// one — matches the name. +func lookupResource(deps resource.Dependencies, shortName string) (resource.Resource, error) { + var found resource.Resource + for n, r := range deps { + if n.Name != shortName { + continue + } + if found != nil { + return nil, fmt.Errorf("matches multiple dependencies") + } + found = r + } + if found == nil { + return nil, fmt.Errorf("not found in dependencies") + } + return found, nil +} + // run is the background polling loop. It exits when the resource is closed. func (m *sensorMonitor) run() { defer m.wg.Done() @@ -193,7 +282,7 @@ func (m *sensorMonitor) run() { } } -// poll reads the sensor once, evaluates every rule, and sends notifications. +// poll reads the sensor once, evaluates every rule, and fires the rule's actions. func (m *sensorMonitor) poll(ctx context.Context) { readings, err := m.sensorDep.Readings(ctx, nil) if err != nil { @@ -224,41 +313,101 @@ func (m *sensorMonitor) poll(ctx context.Context) { cmp, _ := parseOperator(rule.Operator) fired := cmp(value, rule.Threshold) - notify := false + fireTrigger := false + fireResolve := false m.mu.Lock() st := &m.ruleStates[i] st.lastValue = value switch { case fired && !st.triggered: st.triggered = true - st.lastNotified = now - notify = true - case fired && st.triggered && m.cooldown > 0 && now.Sub(st.lastNotified) >= m.cooldown: - st.lastNotified = now - notify = true + st.lastFired = now + fireTrigger = true + case fired && st.triggered && m.cooldown > 0 && now.Sub(st.lastFired) >= m.cooldown: + st.lastFired = now + fireTrigger = true case !fired: + fireResolve = st.triggered st.triggered = false } m.mu.Unlock() - if notify { - m.sendNotification(ctx, rule, value) + if fireTrigger { + m.runActions(ctx, i, rule, rule.OnTrigger, value) + } + if fireResolve { + m.runActions(ctx, i, rule, rule.OnResolve, value) + // The episode is over; drop captured vars so the next one starts fresh. + m.mu.Lock() + m.ruleStates[i].vars = nil + m.mu.Unlock() } } } -// sendNotification renders the rule's message and calls DoCommand on the notifier. -func (m *sensorMonitor) sendNotification(ctx context.Context, rule Rule, value float64) { - msg := renderMessage(rule, value) - cmd := map[string]interface{}{ - "command": "send", - "text": msg, - } - if _, err := m.notifierDep.DoCommand(ctx, cmd); err != nil { - m.logger.Errorf("failed to notify via %q: %v", m.cfg.Notifier, err) +// runActions fires each action's DoCommand on its resolved resource. Before each +// call it resolves ${...} references in the command against the rule/reading +// context and any captured responses; after a successful call it stores the +// response under the action's capture name (if set) for later actions to +// reference. Best-effort: an unresolved reference, missing resource, or DoCommand +// error logs a warning and skips that action without affecting monitoring. +func (m *sensorMonitor) runActions(ctx context.Context, ruleIdx int, rule Rule, actions []Action, value float64) { + if len(actions) == 0 { return } - m.logger.Infof("notification sent: %s", msg) + + // Substitution context: the rule/reading values plus any previously captured + // responses for this rule. + subCtx := map[string]interface{}{ + "value": value, + "key": rule.Key, + "threshold": rule.Threshold, + "operator": rule.Operator, + } + m.mu.RLock() + for k, v := range m.ruleStates[ruleIdx].vars { + subCtx[k] = v + } + m.mu.RUnlock() + + captured := map[string]interface{}{} + for _, a := range actions { + res, ok := m.actionResources[a.Resource] + if !ok { + // Every action resource is resolved at construction, so this is unexpected. + m.logger.Warnf("action resource %q not resolved; skipping", a.Resource) + continue + } + resolved, err := resolveValue(a.Command, subCtx) + if err != nil { + m.logger.Warnf("skipping action on %q: %v", a.Resource, err) + continue + } + cmd, _ := resolved.(map[string]interface{}) + resp, err := res.DoCommand(ctx, cmd) + if err != nil { + m.logger.Warnf("action DoCommand on %q failed: %v", a.Resource, err) + continue + } + if a.Capture != "" { + // Available to later actions in this batch and (after the loop) persisted + // for on_resolve. + subCtx[a.Capture] = resp + captured[a.Capture] = resp + } + m.logger.Infof("action fired on %q", a.Resource) + } + + if len(captured) > 0 { + m.mu.Lock() + if m.ruleStates[ruleIdx].vars == nil { + m.ruleStates[ruleIdx].vars = map[string]interface{}{} + } + for k, v := range captured { + m.ruleStates[ruleIdx].vars[k] = v + } + m.mu.Unlock() + } } // Readings returns the most recent sensor readings plus, per rule, a @@ -336,26 +485,88 @@ func toFloat64(v interface{}) (float64, bool) { } } -// renderMessage builds the notification text for a fired rule. -func renderMessage(rule Rule, value float64) string { - if rule.Message == "" { - return fmt.Sprintf("%s is %s (%s %s)", - rule.Key, - formatFloat(value), - rule.Operator, - formatFloat(rule.Threshold), - ) +// refExact matches a string that is exactly one reference, e.g. "${msg.ts}". +// refAny matches every reference embedded anywhere in a string. +var ( + refExact = regexp.MustCompile(`^\$\{([^}]+)\}$`) + refAny = regexp.MustCompile(`\$\{([^}]+)\}`) +) + +// resolveValue walks a command value and substitutes ${name} / ${name.path} +// references against ctx. A value that is exactly one reference keeps the +// referenced value's type (so a captured map or number passes through intact); a +// reference embedded in a larger string is substituted textually. An unresolved +// reference returns an error so the caller can skip the action. +func resolveValue(v interface{}, ctx map[string]interface{}) (interface{}, error) { + switch x := v.(type) { + case string: + if mm := refExact.FindStringSubmatch(x); mm != nil { + val, ok := lookupPath(ctx, mm[1]) + if !ok { + return nil, fmt.Errorf("unresolved reference ${%s}", mm[1]) + } + return val, nil + } + var refErr error + out := refAny.ReplaceAllStringFunc(x, func(match string) string { + path := refAny.FindStringSubmatch(match)[1] + val, ok := lookupPath(ctx, path) + if !ok { + refErr = fmt.Errorf("unresolved reference ${%s}", path) + return match + } + return stringify(val) + }) + if refErr != nil { + return nil, refErr + } + return out, nil + case map[string]interface{}: + out := make(map[string]interface{}, len(x)) + for k, val := range x { + rv, err := resolveValue(val, ctx) + if err != nil { + return nil, err + } + out[k] = rv + } + return out, nil + case []interface{}: + out := make([]interface{}, len(x)) + for i, e := range x { + rv, err := resolveValue(e, ctx) + if err != nil { + return nil, err + } + out[i] = rv + } + return out, nil + default: + return v, nil } - r := strings.NewReplacer( - "{key}", rule.Key, - "{value}", formatFloat(value), - "{threshold}", formatFloat(rule.Threshold), - "{operator}", rule.Operator, - ) - return r.Replace(rule.Message) } -// formatFloat renders a float without a trailing ".0" for whole numbers. -func formatFloat(f float64) string { - return strconv.FormatFloat(f, 'f', -1, 64) +// lookupPath walks a dot-separated path into a context map. +func lookupPath(ctx map[string]interface{}, path string) (interface{}, bool) { + var cur interface{} = ctx + for _, part := range strings.Split(path, ".") { + m, ok := cur.(map[string]interface{}) + if !ok { + return nil, false + } + cur, ok = m[part] + if !ok { + return nil, false + } + } + return cur, true +} + +// stringify renders a referenced value for embedding in a larger string, +// trimming trailing zeros from floats. +func stringify(v interface{}) string { + if f, ok := v.(float64); ok { + return strconv.FormatFloat(f, 'f', -1, 64) + } + return fmt.Sprint(v) } diff --git a/resources/sensormonitor/sensormonitor_test.go b/resources/sensormonitor/sensormonitor_test.go index d4d279f..ff6b36b 100644 --- a/resources/sensormonitor/sensormonitor_test.go +++ b/resources/sensormonitor/sensormonitor_test.go @@ -4,6 +4,7 @@ import ( "context" "sync" "testing" + "time" sensor "go.viam.com/rdk/components/sensor" "go.viam.com/rdk/logging" @@ -45,51 +46,50 @@ func (f *fakeSensor) Readings(ctx context.Context, extra map[string]interface{}) func (f *fakeSensor) Close(context.Context) error { return nil } -// fakeNotifier is a generic.Service that records the DoCommands it receives. -type fakeNotifier struct { +// fakeTarget is a generic resource standing in for an action target. It records +// the DoCommands it receives and returns a configurable response, so tests can +// assert what the monitor fired and exercise the capture/reference mechanism. +type fakeTarget struct { resource.Named resource.AlwaysRebuild mu sync.Mutex received []map[string]interface{} + resp map[string]interface{} } -func newFakeNotifier(name string) *fakeNotifier { - return &fakeNotifier{Named: generic.Named(name).AsNamed()} +func newFakeTarget(name string) *fakeTarget { + return &fakeTarget{Named: generic.Named(name).AsNamed()} } -func (f *fakeNotifier) DoCommand(ctx context.Context, cmd map[string]interface{}) (map[string]interface{}, error) { +func (f *fakeTarget) DoCommand(ctx context.Context, cmd map[string]interface{}) (map[string]interface{}, error) { f.mu.Lock() defer f.mu.Unlock() f.received = append(f.received, cmd) + if f.resp != nil { + return f.resp, nil + } return map[string]interface{}{}, nil } -func (f *fakeNotifier) Close(context.Context) error { return nil } +func (f *fakeTarget) Close(context.Context) error { return nil } -// texts returns the notification message strings received so far. -func (f *fakeNotifier) texts() []string { +func (f *fakeTarget) commands() []map[string]interface{} { f.mu.Lock() defer f.mu.Unlock() - out := make([]string, 0, len(f.received)) - for _, cmd := range f.received { - if cmd["command"] != "send" { - continue - } - if text, ok := cmd["text"].(string); ok { - out = append(out, text) - } - } - return out + return append([]map[string]interface{}{}, f.received...) } -// newTestMonitor builds a monitor wired to the given fakes WITHOUT starting the -// background polling loop, so tests drive poll() deterministically. -func newTestMonitor(t *testing.T, cfg *Config, src *fakeSensor, notifier *fakeNotifier) *sensorMonitor { +// newTestMonitor builds a monitor wired to the given sensor and action targets +// WITHOUT starting the background polling loop, so tests drive poll() +// deterministically. +func newTestMonitor(t *testing.T, cfg *Config, src *fakeSensor, targets map[string]*fakeTarget) *sensorMonitor { t.Helper() deps := resource.Dependencies{ - sensor.Named(cfg.Sensor): src, - generic.Named(cfg.Notifier): notifier, + sensor.Named(cfg.Sensor): src, + } + for name, tgt := range targets { + deps[generic.Named(name)] = tgt } name := resource.NewName(sensor.API, "monitor") m, err := newMonitor(deps, name, cfg, logging.NewTestLogger(t)) @@ -109,32 +109,49 @@ func TestValidate(t *testing.T) { }{ { name: "valid", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, - reqDeps: []string{"s", "n"}, + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, + reqDeps: []string{"s"}, }, { name: "missing sensor", - cfg: Config{Notifier: "n", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, - wantErr: true, - }, - { - name: "missing notifier", - cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, + cfg: Config{Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1}}}, wantErr: true, }, { name: "no rules", - cfg: Config{Sensor: "s", Notifier: "n"}, + cfg: Config{Sensor: "s"}, wantErr: true, }, { name: "rule missing key", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Operator: ">", Threshold: 1}}}, + cfg: Config{Sensor: "s", Rules: []Rule{{Operator: ">", Threshold: 1}}}, wantErr: true, }, { name: "rule bad operator", - cfg: Config{Sensor: "s", Notifier: "n", Rules: []Rule{{Key: "t", Operator: "=>", Threshold: 1}}}, + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: "=>", Threshold: 1}}}, + wantErr: true, + }, + { + name: "with actions", + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + OnTrigger: []Action{{Resource: "r1", Command: map[string]interface{}{"x": 1}}}, + OnResolve: []Action{{Resource: "r2", Command: map[string]interface{}{"x": 0}}}, + }}}, + reqDeps: []string{"s", "r1", "r2"}, + }, + { + name: "action missing resource", + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + OnTrigger: []Action{{Command: map[string]interface{}{"x": 1}}}, + }}}, + wantErr: true, + }, + { + name: "action missing command", + cfg: Config{Sensor: "s", Rules: []Rule{{Key: "t", Operator: ">", Threshold: 1, + OnResolve: []Action{{Resource: "r1"}}, + }}}, wantErr: true, }, } @@ -222,70 +239,231 @@ func TestToFloat64(t *testing.T) { } } -func TestRenderMessage(t *testing.T) { - value := 95.0 - def := renderMessage(Rule{Key: "temperature", Operator: ">", Threshold: 90}, value) - if def != "temperature is 95 (> 90)" { - t.Fatalf("default message = %q", def) - } - custom := renderMessage(Rule{Key: "temperature", Operator: ">", Threshold: 90, Message: "ALERT {key}={value} over {threshold}"}, value) - if custom != "ALERT temperature=95 over 90" { - t.Fatalf("custom message = %q", custom) +func TestResolveValue(t *testing.T) { + ctx := map[string]interface{}{ + "value": 95.0, + "threshold": 90.0, + "key": "temperature", + "msg": map[string]interface{}{"ts": "100.1", "channel": "C1"}, } + + t.Run("exact reference keeps type", func(t *testing.T) { + got, err := resolveValue("${value}", ctx) + if err != nil { + t.Fatal(err) + } + if f, ok := got.(float64); !ok || f != 95.0 { + t.Fatalf("expected float64 95, got %T %v", got, got) + } + }) + + t.Run("exact reference into capture", func(t *testing.T) { + got, err := resolveValue("${msg.ts}", ctx) + if err != nil { + t.Fatal(err) + } + if got != "100.1" { + t.Fatalf("expected \"100.1\", got %v", got) + } + }) + + t.Run("embedded references are stringified", func(t *testing.T) { + got, err := resolveValue("temp ${value} over ${threshold}", ctx) + if err != nil { + t.Fatal(err) + } + if got != "temp 95 over 90" { + t.Fatalf("got %q", got) + } + }) + + t.Run("recurses into maps and slices", func(t *testing.T) { + got, err := resolveValue(map[string]interface{}{ + "a": "${value}", + "b": []interface{}{"${key}"}, + }, ctx) + if err != nil { + t.Fatal(err) + } + m := got.(map[string]interface{}) + if m["a"] != 95.0 { + t.Fatalf("nested map not resolved: %v", m["a"]) + } + if m["b"].([]interface{})[0] != "temperature" { + t.Fatalf("nested slice not resolved: %v", m["b"]) + } + }) + + t.Run("unresolved reference errors", func(t *testing.T) { + if _, err := resolveValue("${nope.field}", ctx); err == nil { + t.Fatal("expected error for unresolved reference") + } + }) } -func TestEdgeTriggeredNotifies(t *testing.T) { +func TestActionsFireOnTriggerAndResolve(t *testing.T) { ctx := context.Background() src := newFakeSensor("src") - src.set(map[string]interface{}{"temperature": 50.0}) - notifier := newFakeNotifier("notify") - + relay := newFakeTarget("relay") m := newTestMonitor(t, &Config{ - Sensor: "src", - Notifier: "notify", - Rules: []Rule{{Key: "temperature", Operator: ">", Threshold: 90}}, - }, src, notifier) + Sensor: "src", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "relay", Command: map[string]interface{}{"set": true}}}, + OnResolve: []Action{{Resource: "relay", Command: map[string]interface{}{"set": false}}}, + }}, + }, src, map[string]*fakeTarget{"relay": relay}) - // Below threshold: no notification. - src.set(map[string]interface{}{"temperature": 50.0}) + // Below threshold: nothing fired. + src.set(map[string]interface{}{"usage": 5.0}) m.poll(ctx) - if got := len(notifier.texts()); got != 0 { - t.Fatalf("expected 0 notifications below threshold, got %d", got) + if got := len(relay.commands()); got != 0 { + t.Fatalf("expected no actions below threshold, got %d", got) } - // Cross above threshold: exactly one notification. - src.set(map[string]interface{}{"temperature": 95.0}) + // Cross above: on_trigger fires once. + src.set(map[string]interface{}{"usage": 20.0}) + m.poll(ctx) + cmds := relay.commands() + if len(cmds) != 1 || cmds[0]["set"] != true { + t.Fatalf("expected one on_trigger command {set:true}, got %v", cmds) + } + + // Still triggered, cooldown 0: no refire. m.poll(ctx) - if got := notifier.texts(); len(got) != 1 { - t.Fatalf("expected 1 notification on edge, got %d: %v", len(got), got) + if got := len(relay.commands()); got != 1 { + t.Fatalf("expected no refire while triggered, got %d", got) } - // Still above threshold, cooldown 0: no repeat. + // Reading returns below threshold: on_resolve fires once. + src.set(map[string]interface{}{"usage": 0.0}) m.poll(ctx) + cmds = relay.commands() + if len(cmds) != 2 || cmds[1]["set"] != false { + t.Fatalf("expected on_resolve command {set:false}, got %v", cmds) + } + + // Stays resolved: no refire. m.poll(ctx) - if got := len(notifier.texts()); got != 1 { - t.Fatalf("expected no repeat while triggered, got %d", got) + if got := len(relay.commands()); got != 2 { + t.Fatalf("expected no refire while resolved, got %d", got) } +} + +func TestOnTriggerRepeatsOnCooldown(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + tgt := newFakeTarget("tgt") + m := newTestMonitor(t, &Config{ + Sensor: "src", + CooldownSec: 60, // long; we backdate lastFired to force a repeat + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "tgt", Command: map[string]interface{}{"ping": 1}}}, + }}, + }, src, map[string]*fakeTarget{"tgt": tgt}) + + src.set(map[string]interface{}{"usage": 20.0}) + m.poll(ctx) // edge fire + if got := len(tgt.commands()); got != 1 { + t.Fatalf("expected 1 fire on edge, got %d", got) + } + + // Backdate so the cooldown has elapsed; next poll re-fires. + m.ruleStates[0].lastFired = m.ruleStates[0].lastFired.Add(-time.Hour) + m.poll(ctx) + if got := len(tgt.commands()); got != 2 { + t.Fatalf("expected a cooldown repeat, got %d", got) + } +} + +func TestCaptureCarriesAcrossTriggerAndResolve(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + // The notifier returns a message identity on send; the monitor must carry it + // to the react command on resolve. + notifier := newFakeTarget("notifier") + notifier.resp = map[string]interface{}{"ok": true, "ts": "100.1", "channel": "C1"} + m := newTestMonitor(t, &Config{ + Sensor: "src", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{ + Resource: "notifier", + Command: map[string]interface{}{"command": "send", "text": "low (${value})"}, + Capture: "msg", + }}, + OnResolve: []Action{{ + Resource: "notifier", + Command: map[string]interface{}{ + "command": "react", "name": "white_check_mark", + "ts": "${msg.ts}", "channel": "${msg.channel}", + }, + }}, + }}, + }, src, map[string]*fakeTarget{"notifier": notifier}) - // Clear then re-fire: a second notification. - src.set(map[string]interface{}{"temperature": 50.0}) + src.set(map[string]interface{}{"usage": 20.0}) m.poll(ctx) - src.set(map[string]interface{}{"temperature": 99.0}) + src.set(map[string]interface{}{"usage": 0.0}) m.poll(ctx) - if got := len(notifier.texts()); got != 2 { - t.Fatalf("expected 2 notifications after re-fire, got %d", got) + + cmds := notifier.commands() + if len(cmds) != 2 { + t.Fatalf("expected send then react, got %v", cmds) + } + if cmds[0]["command"] != "send" || cmds[0]["text"] != "low (20)" { + t.Fatalf("send command not as expected: %v", cmds[0]) + } + react := cmds[1] + if react["command"] != "react" || react["name"] != "white_check_mark" || + react["ts"] != "100.1" || react["channel"] != "C1" { + t.Fatalf("react command did not carry captured message: %v", react) + } +} + +func TestResolveActionSkippedWhenCaptureMissing(t *testing.T) { + ctx := context.Background() + src := newFakeSensor("src") + notifier := newFakeTarget("notifier") + // on_trigger does NOT capture, so the on_resolve reference can't resolve. + m := newTestMonitor(t, &Config{ + Sensor: "src", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "notifier", Command: map[string]interface{}{"command": "send"}}}, + OnResolve: []Action{{Resource: "notifier", Command: map[string]interface{}{"command": "react", "ts": "${msg.ts}"}}}, + }}, + }, src, map[string]*fakeTarget{"notifier": notifier}) + + src.set(map[string]interface{}{"usage": 20.0}) + m.poll(ctx) + src.set(map[string]interface{}{"usage": 0.0}) + m.poll(ctx) + + cmds := notifier.commands() + if len(cmds) != 1 || cmds[0]["command"] != "send" { + t.Fatalf("expected only the send (react skipped, unresolved ref), got %v", cmds) + } +} + +func TestActionResourceMissingFromDeps(t *testing.T) { + src := newFakeSensor("src") + deps := resource.Dependencies{sensor.Named("src"): src} + cfg := &Config{ + Sensor: "src", + Rules: []Rule{{Key: "usage", Operator: ">=", Threshold: 15, + OnTrigger: []Action{{Resource: "ghost", Command: map[string]interface{}{"x": 1}}}, + }}, + } + if _, err := newMonitor(deps, resource.NewName(sensor.API, "monitor"), cfg, logging.NewTestLogger(t)); err == nil { + t.Fatal("expected construction to fail when an action resource is not a dependency") } } func TestReadingsExposesTriggerState(t *testing.T) { ctx := context.Background() src := newFakeSensor("src") - notifier := newFakeNotifier("notify") m := newTestMonitor(t, &Config{ - Sensor: "src", - Notifier: "notify", - Rules: []Rule{{Key: "humidity", Operator: "<", Threshold: 30}}, - }, src, notifier) + Sensor: "src", + Rules: []Rule{{Key: "humidity", Operator: "<", Threshold: 30}}, + }, src, nil) src.set(map[string]interface{}{"humidity": 20.0}) m.poll(ctx) @@ -306,18 +484,19 @@ func TestDoCommandCheckForcesPoll(t *testing.T) { ctx := context.Background() src := newFakeSensor("src") src.set(map[string]interface{}{"temperature": 95.0}) - notifier := newFakeNotifier("notify") + tgt := newFakeTarget("tgt") m := newTestMonitor(t, &Config{ - Sensor: "src", - Notifier: "notify", - Rules: []Rule{{Key: "temperature", Operator: ">", Threshold: 90}}, - }, src, notifier) + Sensor: "src", + Rules: []Rule{{Key: "temperature", Operator: ">", Threshold: 90, + OnTrigger: []Action{{Resource: "tgt", Command: map[string]interface{}{"go": 1}}}, + }}, + }, src, map[string]*fakeTarget{"tgt": tgt}) if _, err := m.DoCommand(ctx, map[string]interface{}{"check": true}); err != nil { t.Fatalf("DoCommand check: %v", err) } - if got := len(notifier.texts()); got == 0 { - t.Fatal("expected a notification after forced check") + if got := len(tgt.commands()); got == 0 { + t.Fatal("expected an action to fire after forced check") } if _, err := m.DoCommand(ctx, map[string]interface{}{"bogus": 1}); err == nil {