model

Documentation

Overview

Copyright [2019] housepower

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Constants
• func GetValueByType(metric Metric, cwt *ColumnWithType) (val interface{})
• func PutRow(r *Row)
• func PutRows(rs *Rows)
• func WhichType(typ string) (dataType int, nullable bool)
• type Batch
  • func NewBatch() (b *Batch)
  • func (b *Batch) Commit() error
  • func (b *Batch) Size() int
• type BatchGroup
• type BatchSys
  • func NewBatchSys(taskCfg *config.TaskConfig, fnCommit func(partition int, offset int64) error) *BatchSys
  • func (bs *BatchSys) CreateBatchGroupMulti(batches []*Batch, offsets map[int]int64)
  • func (bs *BatchSys) CreateBatchGroupSingle(batch *Batch, partition int, offset int64)
  • func (bs *BatchSys) TryCommit() error
• type ColumnWithType
• type DimMetrics
• type InputMessage
• type Metric
• type MsgRow
• type Row
  • func GetRow() *Row
  • func MetricToRow(metric Metric, msg InputMessage, dims []*ColumnWithType) (row *Row)
• type Rows
  • func GetRows() (rs *Rows)
• type TypeInfo

Constants

const (
	TypeUnknown = iota
	Int
	Float
	String
	DateTime
	ElasticDateTime
	IntArray
	FloatArray
	StringArray
	DateTimeArray
)

Functions

func GetValueByType

func GetValueByType(metric Metric, cwt *ColumnWithType) (val interface{})

There are only three cases for the value type of metric, (float64, string, map [string] interface {})

func PutRow

func PutRow(r *Row)

func PutRows

func PutRows(rs *Rows)

func WhichType

func WhichType(typ string) (dataType int, nullable bool)

Types

type Batch

type Batch struct {
	Rows     *Rows
	BatchIdx int64
	RealSize int
	Group    *BatchGroup
}

func NewBatch

func NewBatch() (b *Batch)

func (*Batch) Commit

func (b *Batch) Commit() error

Commit is not retry-able!

func (*Batch) Size

func (b *Batch) Size() int

type BatchGroup

type BatchGroup struct {
	Batchs    []*Batch
	Offsets   map[int]int64
	Sys       *BatchSys
	PendWrite int32 //how many batches in this group are pending to wirte to ClickHouse
}

BatchGroup consists of multiple batches. The `before` relationship could be impossible if messages of a partition are distributed to multiple batches. So those batches need to be committed after ALL of them have been written to clickhouse.

type BatchSys

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

func NewBatchSys

func NewBatchSys(taskCfg *config.TaskConfig, fnCommit func(partition int, offset int64) error) *BatchSys

func (*BatchSys) CreateBatchGroupMulti

func (bs *BatchSys) CreateBatchGroupMulti(batches []*Batch, offsets map[int]int64)

func (*BatchSys) CreateBatchGroupSingle

func (bs *BatchSys) CreateBatchGroupSingle(batch *Batch, partition int, offset int64)

func (*BatchSys) TryCommit

func (bs *BatchSys) TryCommit() error

type ColumnWithType

type ColumnWithType struct {
	Name       string
	Type       int
	Nullable   bool
	SourceName string
}

ColumnWithType

type DimMetrics

type DimMetrics struct {
	Dims   []*ColumnWithType
	Fields []*ColumnWithType
}

DimMetrics

type InputMessage

type InputMessage struct {
	Topic     string
	Partition int
	Key       []byte
	Value     []byte
	Offset    int64
	Timestamp *time.Time
}

MsgWithMeta abstract messages We are not using interface because virtual call. See https://syslog.ravelin.com/go-interfaces-but-at-what-cost-961e0f58a07b?gi=58f6761d1d70

type Metric

type Metric interface {
	GetInt(key string, nullable bool) (val interface{})
	GetFloat(key string, nullable bool) (val interface{})
	GetString(key string, nullable bool) (val interface{})
	GetDateTime(key string, nullable bool) (val interface{})
	GetElasticDateTime(key string, nullable bool) (val interface{})
	GetArray(key string, t int) (val interface{})
	GetNewKeys(knownKeys *sync.Map, newKeys *sync.Map) bool
}

Metric interface for metric collection

type MsgRow

type MsgRow struct {
	Msg   *InputMessage
	Row   *Row
	Shard int
}

type Row

type Row []interface{}

var (
	FakedRow Row = make([]interface{}, 0)
)

func GetRow

func GetRow() *Row

func MetricToRow

func MetricToRow(metric Metric, msg InputMessage, dims []*ColumnWithType) (row *Row)

type Rows

type Rows []*Row

func GetRows

func GetRows() (rs *Rows)

type TypeInfo

type TypeInfo struct {
	Type     int
	Nullable bool
}