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 string, ttl int, coldStorage *ckdb.ColdStorage) *ckdb.Table
  • func (m *ExtMetrics) GenerateNewFlowTags(cache *flow_tag.FlowTagCache)
  • 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, db string, ...) (*ExtMetricsWriter, error)
  • func (w *ExtMetricsWriter) GetCounter() interface{}
  • func (w *ExtMetricsWriter) InitDatabase() error
  • func (w *ExtMetricsWriter) Write(m *ExtMetrics)
  • func (w *ExtMetricsWriter) WriteBatch(batch []interface{})

Constants

const (
	QUEUE_BATCH_SIZE          = 1024
	EXT_METRICS_DB            = "ext_metrics"
	EXT_METRICS_TABLE         = "metrics"
	OBSERVATRACE_SYSTEM_DB    = "observatrace_system"
	OBSERVATRACE_SYSTEM_TABLE = "observatrace_system"
)

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 zerodoc.UniversalTag

	// in observatrace_system: table name
	// in ext_metrids: virtual_table_name
	VTableName string

	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 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) 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,
	db string,
	config *config.Config) (*ExtMetricsWriter, error)

func (*ExtMetricsWriter) GetCounter

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

func (*ExtMetricsWriter) InitDatabase

func (w *ExtMetricsWriter) InitDatabase() error

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).