metrics.go 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299
  1. package metrics
  2. import (
  3. "runtime"
  4. "strings"
  5. "time"
  6. iradix "github.com/hashicorp/go-immutable-radix"
  7. )
  8. type Label struct {
  9. Name string
  10. Value string
  11. }
  12. func (m *Metrics) SetGauge(key []string, val float32) {
  13. m.SetGaugeWithLabels(key, val, nil)
  14. }
  15. func (m *Metrics) SetGaugeWithLabels(key []string, val float32, labels []Label) {
  16. if m.HostName != "" {
  17. if m.EnableHostnameLabel {
  18. labels = append(labels, Label{"host", m.HostName})
  19. } else if m.EnableHostname {
  20. key = insert(0, m.HostName, key)
  21. }
  22. }
  23. if m.EnableTypePrefix {
  24. key = insert(0, "gauge", key)
  25. }
  26. if m.ServiceName != "" {
  27. if m.EnableServiceLabel {
  28. labels = append(labels, Label{"service", m.ServiceName})
  29. } else {
  30. key = insert(0, m.ServiceName, key)
  31. }
  32. }
  33. allowed, labelsFiltered := m.allowMetric(key, labels)
  34. if !allowed {
  35. return
  36. }
  37. m.sink.SetGaugeWithLabels(key, val, labelsFiltered)
  38. }
  39. func (m *Metrics) EmitKey(key []string, val float32) {
  40. if m.EnableTypePrefix {
  41. key = insert(0, "kv", key)
  42. }
  43. if m.ServiceName != "" {
  44. key = insert(0, m.ServiceName, key)
  45. }
  46. allowed, _ := m.allowMetric(key, nil)
  47. if !allowed {
  48. return
  49. }
  50. m.sink.EmitKey(key, val)
  51. }
  52. func (m *Metrics) IncrCounter(key []string, val float32) {
  53. m.IncrCounterWithLabels(key, val, nil)
  54. }
  55. func (m *Metrics) IncrCounterWithLabels(key []string, val float32, labels []Label) {
  56. if m.HostName != "" && m.EnableHostnameLabel {
  57. labels = append(labels, Label{"host", m.HostName})
  58. }
  59. if m.EnableTypePrefix {
  60. key = insert(0, "counter", key)
  61. }
  62. if m.ServiceName != "" {
  63. if m.EnableServiceLabel {
  64. labels = append(labels, Label{"service", m.ServiceName})
  65. } else {
  66. key = insert(0, m.ServiceName, key)
  67. }
  68. }
  69. allowed, labelsFiltered := m.allowMetric(key, labels)
  70. if !allowed {
  71. return
  72. }
  73. m.sink.IncrCounterWithLabels(key, val, labelsFiltered)
  74. }
  75. func (m *Metrics) AddSample(key []string, val float32) {
  76. m.AddSampleWithLabels(key, val, nil)
  77. }
  78. func (m *Metrics) AddSampleWithLabels(key []string, val float32, labels []Label) {
  79. if m.HostName != "" && m.EnableHostnameLabel {
  80. labels = append(labels, Label{"host", m.HostName})
  81. }
  82. if m.EnableTypePrefix {
  83. key = insert(0, "sample", key)
  84. }
  85. if m.ServiceName != "" {
  86. if m.EnableServiceLabel {
  87. labels = append(labels, Label{"service", m.ServiceName})
  88. } else {
  89. key = insert(0, m.ServiceName, key)
  90. }
  91. }
  92. allowed, labelsFiltered := m.allowMetric(key, labels)
  93. if !allowed {
  94. return
  95. }
  96. m.sink.AddSampleWithLabels(key, val, labelsFiltered)
  97. }
  98. func (m *Metrics) MeasureSince(key []string, start time.Time) {
  99. m.MeasureSinceWithLabels(key, start, nil)
  100. }
  101. func (m *Metrics) MeasureSinceWithLabels(key []string, start time.Time, labels []Label) {
  102. if m.HostName != "" && m.EnableHostnameLabel {
  103. labels = append(labels, Label{"host", m.HostName})
  104. }
  105. if m.EnableTypePrefix {
  106. key = insert(0, "timer", key)
  107. }
  108. if m.ServiceName != "" {
  109. if m.EnableServiceLabel {
  110. labels = append(labels, Label{"service", m.ServiceName})
  111. } else {
  112. key = insert(0, m.ServiceName, key)
  113. }
  114. }
  115. allowed, labelsFiltered := m.allowMetric(key, labels)
  116. if !allowed {
  117. return
  118. }
  119. now := time.Now()
  120. elapsed := now.Sub(start)
  121. msec := float32(elapsed.Nanoseconds()) / float32(m.TimerGranularity)
  122. m.sink.AddSampleWithLabels(key, msec, labelsFiltered)
  123. }
  124. // UpdateFilter overwrites the existing filter with the given rules.
  125. func (m *Metrics) UpdateFilter(allow, block []string) {
  126. m.UpdateFilterAndLabels(allow, block, m.AllowedLabels, m.BlockedLabels)
  127. }
  128. // UpdateFilterAndLabels overwrites the existing filter with the given rules.
  129. func (m *Metrics) UpdateFilterAndLabels(allow, block, allowedLabels, blockedLabels []string) {
  130. m.filterLock.Lock()
  131. defer m.filterLock.Unlock()
  132. m.AllowedPrefixes = allow
  133. m.BlockedPrefixes = block
  134. if allowedLabels == nil {
  135. // Having a white list means we take only elements from it
  136. m.allowedLabels = nil
  137. } else {
  138. m.allowedLabels = make(map[string]bool)
  139. for _, v := range allowedLabels {
  140. m.allowedLabels[v] = true
  141. }
  142. }
  143. m.blockedLabels = make(map[string]bool)
  144. for _, v := range blockedLabels {
  145. m.blockedLabels[v] = true
  146. }
  147. m.AllowedLabels = allowedLabels
  148. m.BlockedLabels = blockedLabels
  149. m.filter = iradix.New()
  150. for _, prefix := range m.AllowedPrefixes {
  151. m.filter, _, _ = m.filter.Insert([]byte(prefix), true)
  152. }
  153. for _, prefix := range m.BlockedPrefixes {
  154. m.filter, _, _ = m.filter.Insert([]byte(prefix), false)
  155. }
  156. }
  157. func (m *Metrics) Shutdown() {
  158. if ss, ok := m.sink.(ShutdownSink); ok {
  159. ss.Shutdown()
  160. }
  161. }
  162. // labelIsAllowed return true if a should be included in metric
  163. // the caller should lock m.filterLock while calling this method
  164. func (m *Metrics) labelIsAllowed(label *Label) bool {
  165. labelName := (*label).Name
  166. if m.blockedLabels != nil {
  167. _, ok := m.blockedLabels[labelName]
  168. if ok {
  169. // If present, let's remove this label
  170. return false
  171. }
  172. }
  173. if m.allowedLabels != nil {
  174. _, ok := m.allowedLabels[labelName]
  175. return ok
  176. }
  177. // Allow by default
  178. return true
  179. }
  180. // filterLabels return only allowed labels
  181. // the caller should lock m.filterLock while calling this method
  182. func (m *Metrics) filterLabels(labels []Label) []Label {
  183. if labels == nil {
  184. return nil
  185. }
  186. toReturn := []Label{}
  187. for _, label := range labels {
  188. if m.labelIsAllowed(&label) {
  189. toReturn = append(toReturn, label)
  190. }
  191. }
  192. return toReturn
  193. }
  194. // Returns whether the metric should be allowed based on configured prefix filters
  195. // Also return the applicable labels
  196. func (m *Metrics) allowMetric(key []string, labels []Label) (bool, []Label) {
  197. m.filterLock.RLock()
  198. defer m.filterLock.RUnlock()
  199. if m.filter == nil || m.filter.Len() == 0 {
  200. return m.Config.FilterDefault, m.filterLabels(labels)
  201. }
  202. _, allowed, ok := m.filter.Root().LongestPrefix([]byte(strings.Join(key, ".")))
  203. if !ok {
  204. return m.Config.FilterDefault, m.filterLabels(labels)
  205. }
  206. return allowed.(bool), m.filterLabels(labels)
  207. }
  208. // Periodically collects runtime stats to publish
  209. func (m *Metrics) collectStats() {
  210. for {
  211. time.Sleep(m.ProfileInterval)
  212. m.EmitRuntimeStats()
  213. }
  214. }
  215. // Emits various runtime statsitics
  216. func (m *Metrics) EmitRuntimeStats() {
  217. // Export number of Goroutines
  218. numRoutines := runtime.NumGoroutine()
  219. m.SetGauge([]string{"runtime", "num_goroutines"}, float32(numRoutines))
  220. // Export memory stats
  221. var stats runtime.MemStats
  222. runtime.ReadMemStats(&stats)
  223. m.SetGauge([]string{"runtime", "alloc_bytes"}, float32(stats.Alloc))
  224. m.SetGauge([]string{"runtime", "sys_bytes"}, float32(stats.Sys))
  225. m.SetGauge([]string{"runtime", "malloc_count"}, float32(stats.Mallocs))
  226. m.SetGauge([]string{"runtime", "free_count"}, float32(stats.Frees))
  227. m.SetGauge([]string{"runtime", "heap_objects"}, float32(stats.HeapObjects))
  228. m.SetGauge([]string{"runtime", "total_gc_pause_ns"}, float32(stats.PauseTotalNs))
  229. m.SetGauge([]string{"runtime", "total_gc_runs"}, float32(stats.NumGC))
  230. // Export info about the last few GC runs
  231. num := stats.NumGC
  232. // Handle wrap around
  233. if num < m.lastNumGC {
  234. m.lastNumGC = 0
  235. }
  236. // Ensure we don't scan more than 256
  237. if num-m.lastNumGC >= 256 {
  238. m.lastNumGC = num - 255
  239. }
  240. for i := m.lastNumGC; i < num; i++ {
  241. pause := stats.PauseNs[i%256]
  242. m.AddSample([]string{"runtime", "gc_pause_ns"}, float32(pause))
  243. }
  244. m.lastNumGC = num
  245. }
  246. // Creates a new slice with the provided string value as the first element
  247. // and the provided slice values as the remaining values.
  248. // Ordering of the values in the provided input slice is kept in tact in the output slice.
  249. func insert(i int, v string, s []string) []string {
  250. // Allocate new slice to avoid modifying the input slice
  251. newS := make([]string, len(s)+1)
  252. // Copy s[0, i-1] into newS
  253. for j := 0; j < i; j++ {
  254. newS[j] = s[j]
  255. }
  256. // Insert provided element at index i
  257. newS[i] = v
  258. // Copy s[i, len(s)-1] into newS starting at newS[i+1]
  259. for j := i; j < len(s); j++ {
  260. newS[j+1] = s[j]
  261. }
  262. return newS
  263. }