sender

Documentation

Index

• Constants
• Variables
• func Demultiplex(items []*cmodel.MetaData) ([]*cmodel.MetaData, []*cmodel.MetaData, []*cmodel.MetaData, []*cmodel.MetaData)
• func DestroyConnPools()
• func KeysOfMap(m map[string]string) []string
• func Push2GraphSendQueue(items []*cmodel.MetaData)
• func Push2InfluxdbSendQueue(items []*cmodel.MetaData)
• func Push2JudgeSendQueue(items []*cmodel.MetaData)
• func Push2NqmIcmpSendQueue(pingItems []*cmodel.MetaData)
• func Push2NqmTcpSendQueue(pingItems []*cmodel.MetaData)
• func Push2NqmTcpconnSendQueue(connItems []*cmodel.MetaData)
• func Push2StagingSendQueue(items []*cmodel.MetricValue)
• func Push2TsdbSendQueue(items []*cmodel.MetaData)
• func SetNodeRings(judgeConfig *g.JudgeConfig, graphConfig *g.GraphConfig)
• func Start()
• type ConsistentHashNodeRing
  • func (this *ConsistentHashNodeRing) GetNode(pk string) (string, error)
  • func (this *ConsistentHashNodeRing) SetNodes(nodes []string)
  • func (this *ConsistentHashNodeRing) SetNumberOfReplicas(num int)

Constants

const (
	DefaultProcCronPeriod = time.Duration(5) * time.Second    //ProcCron的周期,默认1s
	DefaultLogCronPeriod  = time.Duration(3600) * time.Second //LogCron的周期,默认300s
)

const (
	DefaultSendQueueMaxSize = 102400 //10.24w
)

const (
	DefaultSendTaskSleepInterval = time.Millisecond * 50 //默认睡眠间隔为50ms
)

send

Variables

var (
	TsdbQueue       *nlist.SafeListLimited
	JudgeQueues     = make(map[string]*nlist.SafeListLimited)
	GraphQueues     = make(map[string]*nlist.SafeListLimited)
	InfluxdbQueues  = make(map[string]*nlist.SafeListLimited)
	NqmIcmpQueue    *nlist.SafeListLimited
	NqmTcpQueue     *nlist.SafeListLimited
	NqmTcpconnQueue *nlist.SafeListLimited
	StagingQueue    *nlist.SafeListLimited
)

发送缓存队列 node -> queue_of_data

var (
	JudgeConnPools        *cpool.SafeRpcConnPools
	TsdbConnPoolHelper    *cpool.TsdbConnPoolHelper
	GraphConnPools        *cpool.SafeRpcConnPools
	InfluxdbConnPools     *cpool.InfluxdbConnPools
	StagingConnPoolHelper *cpool.StagingConnPoolHelper
)

连接池 node_address -> connection_pool

var (
	MinStep int //最小上报周期,单位sec
)

默认参数

Functions

func Demultiplex

func Demultiplex(items []*cmodel.MetaData) ([]*cmodel.MetaData, []*cmodel.MetaData, []*cmodel.MetaData, []*cmodel.MetaData)

func DestroyConnPools

func DestroyConnPools()

func KeysOfMap

func KeysOfMap(m map[string]string) []string

TODO 考虑放到公共组件库,或utils库

func Push2GraphSendQueue

func Push2GraphSendQueue(items []*cmodel.MetaData)

将数据 打入 某个Graph的发送缓存队列, 具体是哪一个Graph 由一致性哈希 决定

func Push2InfluxdbSendQueue

func Push2InfluxdbSendQueue(items []*cmodel.MetaData)

Push data to 3rd-party database

func Push2JudgeSendQueue

func Push2JudgeSendQueue(items []*cmodel.MetaData)

将数据 打入 某个Judge的发送缓存队列, 具体是哪一个Judge 由一致性哈希 决定

func Push2NqmIcmpSendQueue

func Push2NqmIcmpSendQueue(pingItems []*cmodel.MetaData)

Push metrics from fping to the queue for RESTful API

func Push2NqmTcpSendQueue

func Push2NqmTcpSendQueue(pingItems []*cmodel.MetaData)

Push metrics from tcpping to the queue for RESTful API

func Push2NqmTcpconnSendQueue

func Push2NqmTcpconnSendQueue(connItems []*cmodel.MetaData)

Push metrics from tcpconn to the queue for RESTful API

func Push2StagingSendQueue

func Push2StagingSendQueue(items []*cmodel.MetricValue)

Push data from endpoint in filters to Staging

func Push2TsdbSendQueue

func Push2TsdbSendQueue(items []*cmodel.MetaData)

将原始数据入到tsdb发送缓存队列

func SetNodeRings

func SetNodeRings(
	judgeConfig *g.JudgeConfig,
	graphConfig *g.GraphConfig,
)

func Start

func Start()

初始化数据发送服务, 在main函数中调用

Types

type ConsistentHashNodeRing

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

一致性哈希环,用于管理服务器节点.

var (
	JudgeNodeRing *ConsistentHashNodeRing
	GraphNodeRing *ConsistentHashNodeRing
)

服务节点的一致性哈希环 pk -> node

func (*ConsistentHashNodeRing) GetNode

func (this *ConsistentHashNodeRing) GetNode(pk string) (string, error)

根据pk,获取node节点. chash(pk) -> node

func (*ConsistentHashNodeRing) SetNodes

func (this *ConsistentHashNodeRing) SetNodes(nodes []string)

func (*ConsistentHashNodeRing) SetNumberOfReplicas

func (this *ConsistentHashNodeRing) SetNumberOfReplicas(num int)