forked from mirrors/statsd_exporter
138 lines
3.5 KiB
Go
138 lines
3.5 KiB
Go
// Copyright 2013 The Prometheus Authors
|
|
// 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.
|
|
|
|
package event
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/prometheus/statsd_exporter/pkg/clock"
|
|
"github.com/prometheus/statsd_exporter/pkg/mapper"
|
|
)
|
|
|
|
type Event interface {
|
|
MetricName() string
|
|
Value() float64
|
|
Labels() map[string]string
|
|
MetricType() mapper.MetricType
|
|
}
|
|
|
|
type CounterEvent struct {
|
|
CMetricName string
|
|
CValue float64
|
|
CLabels map[string]string
|
|
}
|
|
|
|
func (c *CounterEvent) MetricName() string { return c.CMetricName }
|
|
func (c *CounterEvent) Value() float64 { return c.CValue }
|
|
func (c *CounterEvent) Labels() map[string]string { return c.CLabels }
|
|
func (c *CounterEvent) MetricType() mapper.MetricType { return mapper.MetricTypeCounter }
|
|
|
|
type GaugeEvent struct {
|
|
GMetricName string
|
|
GValue float64
|
|
GRelative bool
|
|
GLabels map[string]string
|
|
}
|
|
|
|
func (g *GaugeEvent) MetricName() string { return g.GMetricName }
|
|
func (g *GaugeEvent) Value() float64 { return g.GValue }
|
|
func (g *GaugeEvent) Labels() map[string]string { return g.GLabels }
|
|
func (g *GaugeEvent) MetricType() mapper.MetricType { return mapper.MetricTypeGauge }
|
|
|
|
type ObserverEvent struct {
|
|
OMetricName string
|
|
OValue float64
|
|
OLabels map[string]string
|
|
}
|
|
|
|
func (o *ObserverEvent) MetricName() string { return o.OMetricName }
|
|
func (o *ObserverEvent) Value() float64 { return o.OValue }
|
|
func (o *ObserverEvent) Labels() map[string]string { return o.OLabels }
|
|
func (o *ObserverEvent) MetricType() mapper.MetricType { return mapper.MetricTypeObserver }
|
|
|
|
type Events []Event
|
|
|
|
type EventQueue struct {
|
|
C chan Events
|
|
q Events
|
|
m sync.Mutex
|
|
flushTicker *time.Ticker
|
|
flushThreshold int
|
|
flushInterval time.Duration
|
|
eventsFlushed prometheus.Counter
|
|
}
|
|
|
|
type EventHandler interface {
|
|
Queue(event Events)
|
|
}
|
|
|
|
func NewEventQueue(c chan Events, flushThreshold int, flushInterval time.Duration, eventsFlushed prometheus.Counter) *EventQueue {
|
|
ticker := clock.NewTicker(flushInterval)
|
|
eq := &EventQueue{
|
|
C: c,
|
|
flushThreshold: flushThreshold,
|
|
flushInterval: flushInterval,
|
|
flushTicker: ticker,
|
|
q: make([]Event, 0, flushThreshold),
|
|
eventsFlushed: eventsFlushed,
|
|
}
|
|
go func() {
|
|
for {
|
|
<-ticker.C
|
|
eq.Flush()
|
|
}
|
|
}()
|
|
return eq
|
|
}
|
|
|
|
func (eq *EventQueue) Queue(events Events) {
|
|
eq.m.Lock()
|
|
defer eq.m.Unlock()
|
|
|
|
for _, e := range events {
|
|
eq.q = append(eq.q, e)
|
|
if len(eq.q) >= eq.flushThreshold {
|
|
eq.FlushUnlocked()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (eq *EventQueue) Flush() {
|
|
eq.m.Lock()
|
|
defer eq.m.Unlock()
|
|
eq.FlushUnlocked()
|
|
}
|
|
|
|
func (eq *EventQueue) FlushUnlocked() {
|
|
eq.C <- eq.q
|
|
eq.q = make([]Event, 0, cap(eq.q))
|
|
eq.eventsFlushed.Inc()
|
|
}
|
|
|
|
func (eq *EventQueue) Len() int {
|
|
eq.m.Lock()
|
|
defer eq.m.Unlock()
|
|
|
|
return len(eq.q)
|
|
}
|
|
|
|
type UnbufferedEventHandler struct {
|
|
C chan Events
|
|
}
|
|
|
|
func (ueh *UnbufferedEventHandler) Queue(events Events) {
|
|
ueh.C <- events
|
|
}
|