model

Documentation

Index

• Variables
• type AttrDataExtra
• type AttrDataFields
• type AttrDataSQL
• type AttrDatabus
• type AttrIndex
• type AttrOther
• type AttrTable
• type Attrs
• type Bsn
• type BsnAppInfo
• type ES
• type JSONTime
  • func (p JSONTime) MarshalJSON() ([]byte, error)
  • func (p JSONTime) String() string
  • func (p *JSONTime) UnmarshalJSON(data []byte) error
• type LoopOffset
  • func (lo *LoopOffset) SetLoop(isLoop bool)
  • func (lo *LoopOffset) SetOffset(id int64, t string)
  • func (lo *LoopOffset) SetRecoverOffset(recoverID int64, recoverTime string)
  • func (lo *LoopOffset) SetRecoverTempOffset(recoverID int64, recoverTime string)
  • func (lo *LoopOffset) SetReview(rid int64, rtime int64)
  • func (lo *LoopOffset) SetTempOffset(id int64, time string)
• type LoopOffsets
  • func (los LoopOffsets) SetLoops(i int, isLoop bool)
  • func (los LoopOffsets) SetOffsets(i int, id int64, time string)
  • func (los LoopOffsets) SetRecoverOffsets(i int, recoverID int64, recoverTime string)
  • func (los LoopOffsets) SetRecoverTempOffsets(i int, recoverID int64, recoverTime string)
  • func (los LoopOffsets) SetTempOffsets(i int, id int64, time string)
• type MapData
  • func (m MapData) Index(attrs *Attrs) (indexAliasName, indexEntityName string)
  • func (m MapData) InetNtoA(ip int64) string
  • func (m MapData) PrimaryID() int64
  • func (m MapData) StrCTime() string
  • func (m MapData) StrID(attrs *Attrs) (string, bool)
  • func (m MapData) StrMTime() string
  • func (m MapData) TransData(attr *Attrs)
  • func (m MapData) TransDtb(attr *Attrs)
• type Message
• type Offset
  • func (o *Offset) OffsetID() int64
  • func (o *Offset) OffsetTime() string
• type SQLAsset
• type SQLAttrs
• type SQLBusiness
• type Stat

Variables

var DatabusPool = []string{"dm", "dmreport_new"}

DatabusPool poll from db.

var (
	ExistsAction = map[string]bool{"repair": true, "all": true}
)

var .

Types

type AttrDataExtra

type AttrDataExtra struct {
	Type         string            `json:"type"`
	Tag          string            `json:"tag"`
	Condition    map[string]string `json:"condition"`
	SliceField   string            `json:"slice_field"` // 逗号分隔，支持多个字段
	DBName       string            `json:"dbname"`
	Table        string            `json:"table"`
	TableFormat  string            `json:"table_format"`
	InField      string            `json:"in_field"`
	FieldsStr    string            `json:"fields_str"`
	Fields       []string          `json:"fields"`
	RemoveFields []string          `json:"remove_fields"`
	SQL          string            `json:"sql"`
}

AttrDataExtra .

type AttrDataFields

type AttrDataFields struct {
	ESField string `json:"es"`
	Field   string `json:"field"`
	SQL     string `json:"sql"`
	Expect  string `json:"expect"`
	Stored  string `json:"stored"`
	InDtb   string `json:"in_dtb"`
}

AttrDataFields .

type AttrDataSQL

type AttrDataSQL struct {
	DataIndexSuffix       string //索引数据归属
	DataFields            string
	DataFieldsV2          map[string]AttrDataFields //�存放json转换得到的data_fields字段信息, 替换老的DataFields
	DataIndexFields       []string                  //来自DataFields左数第一位
	DataIndexRemoveFields []string                  //ES需要移除的字段
	DataIndexFormatFields map[string]string         //ES每个字段的格式化，如int64,time,int等
	DataDtbFields         map[string][]string       //databus的字段对应的es字段, TODO 改成 map[string]map[string]bool 或 map[string][]string，应对一个数据库字段用在多个es字段
	DataExtraInfo         string
	SQLFields             string //来自DataFields左数第二位，含表名和字段的alias以及mysql函数等其他表达式
	SQLByID               string //因为有left join的缘故，顾提供完整sql（抛除字段部分，下同）
	SQLByMTime            string
	SQLByIDMTime          string
}

AttrDataSQL .

type AttrDatabus

type AttrDatabus struct {
	DatabusInfo string
	Ticker      int    // 定时时间(毫秒)
	AggCount    int    // 聚合数量
	Databus     string // databus Map key
	PrimaryID   string // 主表索引id
	RelatedID   string // 关联表索引id
}

AttrDatabus .

type AttrIndex

type AttrIndex struct {
	IndexAliasPrefix  string //写和读数据时，指向的索引。是别名索引
	IndexEntityPrefix string //创建索引时，指向的索引。是实体索引名
	IndexFormat       string
	IndexSplit        string
	IndexFrom         int
	IndexTo           int
	IndexType         string
	IndexID           string
	IndexMapping      string
	IndexZero         string
	IndexFixed        bool
}

AttrIndex .

type AttrOther

type AttrOther struct {
	ReviewNum  int64
	ReviewTime int64
	Sleep      float64
	Size       int
}

AttrOther .

type AttrTable

type AttrTable struct {
	TablePrefix string
	TableFormat string
	TableSplit  string
	TableFrom   int
	TableTo     int
	TableZero   string
	TableFixed  bool
}

AttrTable .

type Attrs

type Attrs struct {
	Business   string
	AppID      string
	DBName     string
	ESName     string
	DtbName    string
	Table      *AttrTable
	Index      *AttrIndex
	DataSQL    *AttrDataSQL
	DataExtras []AttrDataExtra //appID需要关联其他库的数据
	Databus    *AttrDatabus
	Other      *AttrOther
}

Attrs parse AppAttrs

type Bsn

type Bsn struct {
	Business string
	AppInfo  []BsnAppInfo
	AssetDB  map[string]*sql.Config
	AssetES  []string
}

Bsn single table offset

type BsnAppInfo

type BsnAppInfo struct {
	AppID       string `json:"appid"`
	IncrWay     string `json:"incr_way"`
	IncrOpen    bool   `json:"incr_open"`
	RecoverLock bool
}

BsnAppInfo .

type ES

type ES struct {
	Addr string
}

ES .

type JSONTime

type JSONTime time.Time

JSONTime .

func (JSONTime) MarshalJSON

func (p JSONTime) MarshalJSON() ([]byte, error)

MarshalJSON .

func (JSONTime) String

func (p JSONTime) String() string

String .

func (*JSONTime) UnmarshalJSON

func (p *JSONTime) UnmarshalJSON(data []byte) error

UnmarshalJSON .

type LoopOffset

type LoopOffset struct {
	IsLoop          bool
	OffsetID        int64
	OffsetTime      string
	TempOffsetID    int64
	TempOffsetTime  string
	RecoverID       int64
	RecoverTime     string
	TempRecoverID   int64
	TempRecoverTime string
	ReviewID        int64
	ReviewTime      int64
}

LoopOffset single table offset

func (*LoopOffset) SetLoop

func (lo *LoopOffset) SetLoop(isLoop bool)

SetLoop .

func (*LoopOffset) SetOffset

func (lo *LoopOffset) SetOffset(id int64, t string)

SetOffset .

func (*LoopOffset) SetRecoverOffset

func (lo *LoopOffset) SetRecoverOffset(recoverID int64, recoverTime string)

SetRecoverOffset .

func (*LoopOffset) SetRecoverTempOffset

func (lo *LoopOffset) SetRecoverTempOffset(recoverID int64, recoverTime string)

SetRecoverTempOffset .

func (*LoopOffset) SetReview

func (lo *LoopOffset) SetReview(rid int64, rtime int64)

SetReview .

func (*LoopOffset) SetTempOffset

func (lo *LoopOffset) SetTempOffset(id int64, time string)

SetTempOffset .

type LoopOffsets

type LoopOffsets map[int]*LoopOffset

LoopOffsets more tables offset

func (LoopOffsets) SetLoops

func (los LoopOffsets) SetLoops(i int, isLoop bool)

SetLoops .

func (LoopOffsets) SetOffsets

func (los LoopOffsets) SetOffsets(i int, id int64, time string)

SetOffsets .

func (LoopOffsets) SetRecoverOffsets

func (los LoopOffsets) SetRecoverOffsets(i int, recoverID int64, recoverTime string)

SetRecoverOffsets .

func (LoopOffsets) SetRecoverTempOffsets

func (los LoopOffsets) SetRecoverTempOffsets(i int, recoverID int64, recoverTime string)

SetRecoverTempOffsets .

func (LoopOffsets) SetTempOffsets

func (los LoopOffsets) SetTempOffsets(i int, id int64, time string)

SetTempOffsets .

type MapData

type MapData map[string]interface{}

MapData .

func (MapData) Index

func (m MapData) Index(attrs *Attrs) (indexAliasName, indexEntityName string)

Index .

func (MapData) InetNtoA

func (m MapData) InetNtoA(ip int64) string

InetNtoA int64 to string ip.

func (MapData) PrimaryID

func (m MapData) PrimaryID() int64

PrimaryID .

func (MapData) StrCTime

func (m MapData) StrCTime() string

StrCTime .

func (MapData) StrID

func (m MapData) StrID(attrs *Attrs) (string, bool)

StrID .

func (MapData) StrMTime

func (m MapData) StrMTime() string

StrMTime .

func (MapData) TransData

func (m MapData) TransData(attr *Attrs)

TransData transfer address into value

func (MapData) TransDtb

func (m MapData) TransDtb(attr *Attrs)

TransDtb transfer databus fields into es fields

type Message

type Message struct {
	Action string          `json:"action"`
	Table  string          `json:"table"`
	New    json.RawMessage `json:"new"`
	Old    json.RawMessage `json:"old"`
}

Message canal binlog message.

type Offset

type Offset struct {
	OffID      int64
	OffTime    xtime.Time
	ReviewID   int64
	ReviewTime int64
}

Offset .

func (*Offset) OffsetID

func (o *Offset) OffsetID() int64

OffsetID .

func (*Offset) OffsetTime

func (o *Offset) OffsetTime() string

OffsetTime .

type SQLAsset

type SQLAsset struct {
	ID   int64
	Name string
	Type int
	Src  string
}

SQLAsset .

type SQLAttrs

type SQLAttrs struct {
	AppID            string
	DBName           string
	ESName           string
	DtbName          string
	TablePrefix      string
	TableFormat      string
	IndexAliasPrefix string //写和读数据时，指向的索引。是别名索引
	IndexVersion     string //创建索引时，指向的索引。这里是实体索引版本号
	IndexFormat      string
	IndexType        string
	IndexID          string
	IndexMapping     string
	DataIndexSuffix  string
	DataExtraInfo    string
	ReviewNum        int64
	ReviewTime       int64
	Sleep            float64
	Size             int
	Business         string
	DataFields       string
	SQLByID          string
	SQLByMTime       string
	SQLByIDMTime     string
	DatabusInfo      string
	DatabusIndexID   string
}

SQLAttrs get attrs from db

type SQLBusiness

type SQLBusiness struct {
	Business string
	AppIds   string
	AssetDB  string
	AssetES  string
	AssetDtb string
}

SQLBusiness single table offset

type Stat

type Stat struct {
	Counts int `json:"counts"`
}

Stat all data statistics