tsdb

Documentation

Index

• func GetCounter(metric, tag string, tagMap map[string]string) (counter string, err error)
• func GetIndex()
• func GetIndexLoop()
• func PostIndex(url string, calltimeout int64, recv interface{}, resp interface{}) error
• func RebuildConsistentHashRing(hashRing *ConsistentHashRing, nodes []string, replicas int)
• type ClusterNode
• type ConsistentHashRing
  • func NewConsistentHashRing(replicas int32, nodes []string) *ConsistentHashRing
  • func (c *ConsistentHashRing) GetNode(pk string) (string, error)
  • func (c *ConsistentHashRing) GetRing() *consistent.Consistent
  • func (c *ConsistentHashRing) Set(r *consistent.Consistent)
• type IndexAddrs
  • func (i *IndexAddrs) Get() []string
  • func (i *IndexAddrs) Set(addrs []string)
• type IndexByFullTagsResp
• type IndexCludeResp
• type IndexMetricsResp
• type IndexTagPairsResp
• type Pool
• type TsdbDataSource
  • func (tsdb *TsdbDataSource) GetInstance(metric, endpoint string, tags map[string]string) []string
  • func (tsdb *TsdbDataSource) Init()
  • func (tsdb *TsdbDataSource) Push2Queue(items []*dataobj.MetricValue)
  • func (tsdb *TsdbDataSource) QueryData(inputs []dataobj.QueryData) []*dataobj.TsdbQueryResponse
  • func (tsdb *TsdbDataSource) QueryDataForUI(input dataobj.QueryDataForUI) []*dataobj.TsdbQueryResponse
  • func (tsdb *TsdbDataSource) QueryIndexByClude(recv []dataobj.CludeRecv) []dataobj.XcludeResp
  • func (tsdb *TsdbDataSource) QueryIndexByFullTags(recv []dataobj.IndexByFullTagsRecv) ([]dataobj.IndexByFullTagsResp, int)
  • func (tsdb *TsdbDataSource) QueryMetrics(recv dataobj.EndpointsRecv) *dataobj.MetricResp
  • func (tsdb *TsdbDataSource) QueryOne(para dataobj.TsdbQueryParam) (resp *dataobj.TsdbQueryResponse, err error)
  • func (tsdb *TsdbDataSource) QueryTagPairs(recv dataobj.EndpointMetricRecv) []dataobj.IndexTagkvResp
  • func (tsdb *TsdbDataSource) SelectPoolByPK(pk string) ([]Pool, error)
  • func (tsdb *TsdbDataSource) Send2TsdbTask(Q *list.SafeListLimited, node, addr string, concurrent int)
• type TsdbSection

Functions

func GetCounter

func GetCounter(metric, tag string, tagMap map[string]string) (counter string, err error)

func GetIndex

func GetIndex()

func GetIndexLoop

func GetIndexLoop()

func PostIndex

func PostIndex(url string, calltimeout int64, recv interface{}, resp interface{}) error

func RebuildConsistentHashRing

func RebuildConsistentHashRing(hashRing *ConsistentHashRing, nodes []string, replicas int)

Types

type ClusterNode

type ClusterNode struct {
	Addrs []string `json:"addrs"`
}

type ConsistentHashRing

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

func NewConsistentHashRing

func NewConsistentHashRing(replicas int32, nodes []string) *ConsistentHashRing

func (*ConsistentHashRing) GetNode

func (c *ConsistentHashRing) GetNode(pk string) (string, error)

func (*ConsistentHashRing) GetRing

func (c *ConsistentHashRing) GetRing() *consistent.Consistent

func (*ConsistentHashRing) Set

func (c *ConsistentHashRing) Set(r *consistent.Consistent)

type IndexAddrs

type IndexAddrs struct {
	sync.RWMutex
	Data []string
}

var IndexList IndexAddrs

func (*IndexAddrs) Get

func (i *IndexAddrs) Get() []string

func (*IndexAddrs) Set

func (i *IndexAddrs) Set(addrs []string)

type IndexByFullTagsResp

type IndexByFullTagsResp struct {
	Data []dataobj.IndexByFullTagsResp `json:"dat"`
	Err  string                        `json:"err"`
}

type IndexCludeResp

type IndexCludeResp struct {
	Data []dataobj.XcludeResp `json:"dat"`
	Err  string               `json:"err"`
}

type IndexMetricsResp

type IndexMetricsResp struct {
	Data *dataobj.MetricResp `json:"dat"`
	Err  string              `json:"err"`
}

type IndexTagPairsResp

type IndexTagPairsResp struct {
	Data []dataobj.IndexTagkvResp `json:"dat"`
	Err  string                   `json:"err"`
}

type Pool

type Pool struct {
	Pool *pool.ConnPool
	Addr string
}

type TsdbDataSource

type TsdbDataSource struct {
	//config
	Section               TsdbSection
	SendQueueMaxSize      int
	SendTaskSleepInterval time.Duration

	// 服务节点的一致性哈希环 pk -> node
	TsdbNodeRing *ConsistentHashRing

	// 发送缓存队列 node -> queue_of_data
	TsdbQueues map[string]*list.SafeListLimited

	// 连接池 node_address -> connection_pool
	TsdbConnPools *pools.ConnPools
}

func (*TsdbDataSource) GetInstance

func (tsdb *TsdbDataSource) GetInstance(metric, endpoint string, tags map[string]string) []string

func (*TsdbDataSource) Init

func (tsdb *TsdbDataSource) Init()

func (*TsdbDataSource) Push2Queue

func (tsdb *TsdbDataSource) Push2Queue(items []*dataobj.MetricValue)

Push2TsdbSendQueue pushes data to a TSDB instance which depends on the consistent ring.

func (*TsdbDataSource) QueryData

func (tsdb *TsdbDataSource) QueryData(inputs []dataobj.QueryData) []*dataobj.TsdbQueryResponse

func (*TsdbDataSource) QueryDataForUI

func (tsdb *TsdbDataSource) QueryDataForUI(input dataobj.QueryDataForUI) []*dataobj.TsdbQueryResponse

func (*TsdbDataSource) QueryIndexByClude

func (tsdb *TsdbDataSource) QueryIndexByClude(recv []dataobj.CludeRecv) []dataobj.XcludeResp

func (*TsdbDataSource) QueryIndexByFullTags

func (tsdb *TsdbDataSource) QueryIndexByFullTags(recv []dataobj.IndexByFullTagsRecv) ([]dataobj.IndexByFullTagsResp, int)

deprecated

func (*TsdbDataSource) QueryMetrics

func (tsdb *TsdbDataSource) QueryMetrics(recv dataobj.EndpointsRecv) *dataobj.MetricResp

func (*TsdbDataSource) QueryOne

func (tsdb *TsdbDataSource) QueryOne(para dataobj.TsdbQueryParam) (resp *dataobj.TsdbQueryResponse, err error)

func (*TsdbDataSource) QueryTagPairs

func (tsdb *TsdbDataSource) QueryTagPairs(recv dataobj.EndpointMetricRecv) []dataobj.IndexTagkvResp

func (*TsdbDataSource) SelectPoolByPK

func (tsdb *TsdbDataSource) SelectPoolByPK(pk string) ([]Pool, error)

func (*TsdbDataSource) Send2TsdbTask

func (tsdb *TsdbDataSource) Send2TsdbTask(Q *list.SafeListLimited, node, addr string, concurrent int)

type TsdbSection

type TsdbSection struct {
	Enabled      bool   `yaml:"enabled"`
	Name         string `yaml:"name"`
	Batch        int    `yaml:"batch"`
	ConnTimeout  int    `yaml:"connTimeout"`
	CallTimeout  int    `yaml:"callTimeout"`
	WorkerNum    int    `yaml:"workerNum"`
	MaxConns     int    `yaml:"maxConns"`
	MaxIdle      int    `yaml:"maxIdle"`
	IndexTimeout int    `yaml:"indexTimeout"`

	Replicas    int                     `yaml:"replicas"`
	Cluster     map[string]string       `yaml:"cluster"`
	ClusterList map[string]*ClusterNode `json:"clusterList"`
}