forked from riemann/riemann
-
Notifications
You must be signed in to change notification settings - Fork 0
/
deps.clj
80 lines (64 loc) · 2.32 KB
/
deps.clj
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
(ns riemann.deps
"Riemann's dependency resolution system expresses stateful relationships
between events. Dependencies are expressed as Rules; a Rule is a statement
about the relationship between a particular event and the current state of
the index.
Maps are rules which specify that their keys and values should be present in
some event in the index. {} will match any non-empty index. {:service \"a\"
:state \"ok\"} will match an index which has {:service \"a\" :state \"ok\"
:metric 123}, and so on.
(all & rules) matches only if all rules match.
(any & rules) matches if any of the rules match.
(localhost & rules) states that all child rules must have the same host as
the event of interest.
(depends a & bs) means that if a matches the current event (and only the
current event, not the full index), b must match the current event and index.
"
(:require [riemann.streams :as streams]))
(defprotocol Rule
(match [this context event]))
(extend-protocol Rule
clojure.lang.IPersistentMap
(match [this index _]
(some (fn [e] (= this (select-keys e (keys this))))
index)
))
(defrecord All [rules]
Rule
(match [this index event]
; (prn "Matching all" rules)
; (prn "index are" index)
; (prn "event is" event)
(every? #(match % index event) rules)))
(defn all [& rules]
(All. rules))
(defrecord Any [rules]
Rule
(match [this index event]
(some #(match % index event) rules)))
(defn any [& rules]
(Any. rules))
(defrecord Localhost [rule]
Rule
(match [this index event]
(match rule
(filter (fn [e] (= (:host event) (:host e))) index)
event)))
(defn localhost [& rules]
(Localhost. (apply all rules)))
(defrecord Depends [a b]
Rule
(match [this index event]
(if (match a [event] event)
(match b index event)
true)))
(defn depends [a & bs]
(Depends. a (All. bs)))
(defn deps-tag [index rule & children]
"Returns a stream which accepts events, checks whether they satisfy the given
rule, and associates those which have their dependencies satisfied with
{:deps-satisfied true}, and false for those which are satisfied."
(fn [event]
(streams/call-rescue
(assoc event :deps-satisfied? (match rule index event))
children)))