dbwriter

Documentation

Index

• Constants
• func ReleaseExtMetrics(m *ExtMetrics)
• type ClusterNode
• type Counter
• type ExtMetrics
  • func AcquireExtMetrics() *ExtMetrics
  • func (m *ExtMetrics) Columns() []*ckdb.Column
  • func (m *ExtMetrics) DatabaseName() string
  • func (m *ExtMetrics) GenCKTable(cluster, storagePolicy, ckdbType string, ttl int, ...) *ckdb.Table
  • func (m *ExtMetrics) GenerateNewFlowTags(cache *flow_tag.FlowTagCache)
  • func (m *ExtMetrics) IsValid() bool
  • func (m *ExtMetrics) OrgID() uint16
  • func (m *ExtMetrics) Release()
  • func (m *ExtMetrics) TableName() string
  • func (m *ExtMetrics) VirtualTableName() string
  • func (m *ExtMetrics) WriteBlock(block *ckdb.Block)
• type ExtMetricsWriter
  • func NewExtMetricsWriter(decoderIndex int, msgType datatype.MessageType, flowTagTablePrefix string, ...) (*ExtMetricsWriter, error)
  • func (w *ExtMetricsWriter) GetCounter() interface{}
  • func (w *ExtMetricsWriter) Write(m *ExtMetrics)
  • func (w *ExtMetricsWriter) WriteBatch(batch []interface{})
• type WriterDBID
  • func (t WriterDBID) String() string

Constants

const (
	QUEUE_BATCH_SIZE  = 1024
	EXT_METRICS_DB    = "ext_metrics"
	EXT_METRICS_TABLE = "metrics"

	DEEPFLOW_ADMIN_DB           = "deepflow_admin"
	DEEPFLOW_ADMIN_SERVER_TABLE = "deepflow_server"

	DEEPFLOW_TENANT_DB              = "deepflow_tenant"
	DEEPFLOW_TENANT_COLLECTOR_TABLE = "deepflow_collector"
)

const (
	DefaultPartition = ckdb.TimeFuncTwoHour
)

Functions

func ReleaseExtMetrics

func ReleaseExtMetrics(m *ExtMetrics)

Types

type ClusterNode

type ClusterNode struct {
	Addr string
	Port uint16
}

type Counter

type Counter struct {
	MetricsCount int64 `statsd:"metrics-count"`
	WriteErr     int64 `statsd:"write-err"`
}

type ExtMetrics

type ExtMetrics struct {
	Timestamp uint32 // s
	MsgType   datatype.MessageType

	UniversalTag flow_metrics.UniversalTag

	VTableName string

	AgentID uint16

	// Not stored, only determines which database to store in.
	// When Orgid is 0 or 1, it is stored in database '<DatabaseName()>', otherwise stored in '<OrgId>_<DatabaseName()>'.
	OrgId, RawOrgId uint16 // RawOrgId is read from server-stats message, only used to distinguish which database data is written to
	TeamID          uint16

	TagNames  []string
	TagValues []string

	MetricsFloatNames  []string
	MetricsFloatValues []float64
}

func AcquireExtMetrics

func AcquireExtMetrics() *ExtMetrics

func (*ExtMetrics) Columns

func (m *ExtMetrics) Columns() []*ckdb.Column

Note: The order of append() must be consistent with the order of Write() in WriteBlock.

func (*ExtMetrics) DatabaseName

func (m *ExtMetrics) DatabaseName() string

func (*ExtMetrics) GenCKTable

func (m *ExtMetrics) GenCKTable(cluster, storagePolicy, ckdbType string, ttl int, coldStorage *ckdb.ColdStorage) *ckdb.Table

func (*ExtMetrics) GenerateNewFlowTags

func (m *ExtMetrics) GenerateNewFlowTags(cache *flow_tag.FlowTagCache)

Check if there is a TagName/TagValue/MetricsName not in fieldCache or fieldValueCache, and store the newly appeared item in cache.

func (*ExtMetrics) IsValid

func (m *ExtMetrics) IsValid() bool

func (*ExtMetrics) OrgID

func (m *ExtMetrics) OrgID() uint16

func (*ExtMetrics) Release

func (m *ExtMetrics) Release()

func (*ExtMetrics) TableName

func (m *ExtMetrics) TableName() string

func (*ExtMetrics) VirtualTableName

func (m *ExtMetrics) VirtualTableName() string

func (*ExtMetrics) WriteBlock

func (m *ExtMetrics) WriteBlock(block *ckdb.Block)

Note: The order of Write() must be consistent with the order of append() in Columns.

type ExtMetricsWriter

type ExtMetricsWriter struct {
	utils.Closable
	// contains filtered or unexported fields
}

func NewExtMetricsWriter

func NewExtMetricsWriter(
	decoderIndex int,
	msgType datatype.MessageType,
	flowTagTablePrefix string,
	config *config.Config) (*ExtMetricsWriter, error)

func (*ExtMetricsWriter) GetCounter

func (w *ExtMetricsWriter) GetCounter() interface{}

func (*ExtMetricsWriter) Write

func (w *ExtMetricsWriter) Write(m *ExtMetrics)

func (*ExtMetricsWriter) WriteBatch

func (w *ExtMetricsWriter) WriteBatch(batch []interface{})

This function can be called when the FlowTags in the batch are the same (e.g. Prometheus metrics).

type WriterDBID

type WriterDBID uint8

const (
	EXT_METRICS_DB_ID WriterDBID = iota
	DEEPFLOW_TENANT_DB_ID
	DEEPFLOW_ADMIN_DB_ID
	MAX_DB_ID
)

func (WriterDBID) String

func (t WriterDBID) String() string