statsd_exporter/pkg/event/event.go

138 lines
3.4 KiB
Go
Raw Normal View History

// 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 (c *GaugeEvent) Labels() map[string]string { return c.GLabels }
func (c *GaugeEvent) MetricType() mapper.MetricType { return mapper.MetricTypeGauge }
type TimerEvent struct {
TMetricName string
TValue float64
TLabels map[string]string
}
func (t *TimerEvent) MetricName() string { return t.TMetricName }
func (t *TimerEvent) Value() float64 { return t.TValue }
func (c *TimerEvent) Labels() map[string]string { return c.TLabels }
func (c *TimerEvent) MetricType() mapper.MetricType { return mapper.MetricTypeTimer }
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,
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
}