writer

Documentation

Index

• Constants
• Variables
• func Process(labels []prompb.Label, cfgs ...*pconf.RelabelConfig) []prompb.Label
• type IdentQueue
• type LabelBuilder
• type SafeList
  • func NewSafeList() *SafeList
  • func (sl *SafeList) Len() int
  • func (sl *SafeList) PopBack(max int) []prompb.TimeSeries
  • func (sl *SafeList) PushFront(v interface{}) *list.Element
  • func (sl *SafeList) PushFrontBatch(vs []interface{})
  • func (sl *SafeList) RemoveAll()
• type SafeListLimited
  • func NewSafeListLimited(maxSize int) *SafeListLimited
  • func (sll *SafeListLimited) Len() int
  • func (sll *SafeListLimited) PopBack(max int) []prompb.TimeSeries
  • func (sll *SafeListLimited) PushFront(v interface{}) bool
  • func (sll *SafeListLimited) PushFrontBatch(vs []interface{}) bool
  • func (sll *SafeListLimited) RemoveAll()
• type WriterType
  • func (w WriterType) Post(req []byte, headers ...map[string]string) error
  • func (w WriterType) Write(key string, items []prompb.TimeSeries, headers ...map[string]string)
• type WritersType
  • func NewWriters(pushgwConfig pconf.Pushgw) *WritersType
  • func (ws *WritersType) CleanExpQueue()
  • func (ws *WritersType) Init() error
  • func (ws *WritersType) PushSample(ident string, v interface{})
  • func (ws *WritersType) Put(name string, writer WriterType)
  • func (ws *WritersType) ReportQueueStats(ident string, identQueue *IdentQueue) (interface{}, bool)
  • func (ws *WritersType) StartConsumer(identQueue *IdentQueue)

Constants

const (
	Replace   string = "replace"
	Keep      string = "keep"
	Drop      string = "drop"
	HashMod   string = "hashmod"
	LabelMap  string = "labelmap"
	LabelDrop string = "labeldrop"
	LabelKeep string = "labelkeep"
	Lowercase string = "lowercase"
	Uppercase string = "uppercase"
)

Variables

var (
	// 发往后端TSDB，延迟如何
	ForwardDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Namespace: namespace,
			Subsystem: subsystem,
			Buckets:   []float64{.1, 1, 10},
			Name:      "forward_duration_seconds",
			Help:      "Forward samples to TSDB. latencies in seconds.",
		}, []string{"url"},
	)

	GaugeSampleQueueSize = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Namespace: namespace,
			Subsystem: subsystem,
			Name:      "sample_queue_size",
			Help:      "The size of sample queue.",
		}, []string{"host_ident"},
	)

	CounterWirteTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: namespace,
		Subsystem: subsystem,
		Name:      "write_total",
		Help:      "Number of write.",
	}, []string{"url"})

	CounterWirteErrorTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: namespace,
		Subsystem: subsystem,
		Name:      "write_error_total",
		Help:      "Number of write error.",
	}, []string{"url"})

	CounterPushQueueErrorTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: namespace,
		Subsystem: subsystem,
		Name:      "push_queue_error_total",
		Help:      "Number of push queue error.",
	}, []string{"host_ident"})
)

Functions

func Process

func Process(labels []prompb.Label, cfgs ...*pconf.RelabelConfig) []prompb.Label

Types

type IdentQueue

type IdentQueue struct {
	// contains filtered or unexported fields
}

type LabelBuilder

type LabelBuilder struct {
	LabelSet map[string]string
}

type SafeList

type SafeList struct {
	sync.RWMutex
	L *list.List
}

func NewSafeList

func NewSafeList() *SafeList

func (*SafeList) Len

func (sl *SafeList) Len() int

func (*SafeList) PopBack

func (sl *SafeList) PopBack(max int) []prompb.TimeSeries

func (*SafeList) PushFront

func (sl *SafeList) PushFront(v interface{}) *list.Element

func (*SafeList) PushFrontBatch

func (sl *SafeList) PushFrontBatch(vs []interface{})

func (*SafeList) RemoveAll

func (sl *SafeList) RemoveAll()

type SafeListLimited

type SafeListLimited struct {
	SL *SafeList
	// contains filtered or unexported fields
}

SafeList with Limited Size

func NewSafeListLimited

func NewSafeListLimited(maxSize int) *SafeListLimited

func (*SafeListLimited) Len

func (sll *SafeListLimited) Len() int

func (*SafeListLimited) PopBack

func (sll *SafeListLimited) PopBack(max int) []prompb.TimeSeries

func (*SafeListLimited) PushFront

func (sll *SafeListLimited) PushFront(v interface{}) bool

func (*SafeListLimited) PushFrontBatch

func (sll *SafeListLimited) PushFrontBatch(vs []interface{}) bool

func (*SafeListLimited) RemoveAll

func (sll *SafeListLimited) RemoveAll()

type WriterType

type WriterType struct {
	Opts             pconf.WriterOptions
	ForceUseServerTS bool
	Client           api.Client
}

func (WriterType) Post

func (w WriterType) Post(req []byte, headers ...map[string]string) error

func (WriterType) Write

func (w WriterType) Write(key string, items []prompb.TimeSeries, headers ...map[string]string)

type WritersType

type WritersType struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewWriters

func NewWriters(pushgwConfig pconf.Pushgw) *WritersType

func (*WritersType) CleanExpQueue

func (ws *WritersType) CleanExpQueue()

func (*WritersType) Init

func (ws *WritersType) Init() error

func (*WritersType) PushSample

func (ws *WritersType) PushSample(ident string, v interface{})

func (*WritersType) Put

func (ws *WritersType) Put(name string, writer WriterType)

func (*WritersType) ReportQueueStats

func (ws *WritersType) ReportQueueStats(ident string, identQueue *IdentQueue) (interface{}, bool)

func (*WritersType) StartConsumer

func (ws *WritersType) StartConsumer(identQueue *IdentQueue)