blob: a77d2d24229969853334e0dd16978a8e5c76b1e7 (
plain)
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
|
;;;; ============LICENSE_START==========================================
;;;; ===================================================================
;;;; Copyright (c) 2017 AT&T
;;;;
;;;; Licensed under the Apache License, Version 2.0 (the "License");
;;;; you may not use this file except in compliance with the License.
;;;; You may obtain a copy of the License at
;;;;
;;;; http://www.apache.org/licenses/LICENSE-2.0
;;;;
;;;; Unless required by applicable law or agreed to in writing, software
;;;; distributed under the License is distributed on an "AS IS" BASIS,
;;;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;;;; See the License for the specific language governing permissions and
;;;; limitations under the License.
;;;;============LICENSE_END============================================
(where (service #"{{service_selector}}")
#(info "got event: " %)
(where (not (expired? event))
(moving-time-window {{moving_window_size}}
(fn [events]
(let [
hostmap (atom {})
hostcnt (atom {})
]
(do
(doseq [m events]
(if (nil? (@hostmap (m :host)))
(do
(swap! hostmap assoc (m :host) (m :metric))
(swap! hostcnt assoc (m :host) 1)
)
(do
(swap! hostmap assoc (m :host) (+ (m :metric) (@hostmap (m :host))))
(swap! hostcnt assoc (m :host) (inc (@hostcnt (m :host))))
)
)
)
(doseq [entry @hostmap]
(swap! hostmap assoc (key entry) (/ (val entry) (@hostcnt (key entry))))
)
(let
[ hostcnt (count @hostmap)
conns (/ (apply + (map (fn [a] (val a)) @hostmap)) hostcnt)
cooling (not (nil? (riemann.index/lookup index "scaling" "suspended")))]
(do
(info "cooling=" cooling " scale_direction={{scale_direction}} hostcnt=" hostcnt " scale_threshold={{scale_threshold}} conns=" conns)
(if (and (not cooling) ({{scale_direction}} hostcnt {{scale_limit}}) ({{scale_direction}} {{scale_threshold}} conns))
(do
(info "=== SCALE ===" "{{scale_direction}}")
(process-policy-triggers {})
(riemann.index/update index {:host "scaling" :service "suspended" :time (unix-time) :description "cooldown flag" :metric 0 :ttl {{cooldown_time}} :state "ok"})
)
)
)
)
)
)
)
)
)
)
|
