pipeline

Documentation

Index

• Constants
• type Filter
• type FilterType
• type Flavor
  • func (f Flavor) YaString() string
• type Http
• type Instance
• type Kafka
• type Mode
• type Mysql
• type OptionPipeline
  • func WithAddFilter(filter *Filter) OptionPipeline
  • func WithPipeDelete(d bool) OptionPipeline
  • func WithPipeMode(mode Mode) OptionPipeline
  • func WithPipeSafe(uPipe *Pipeline) OptionPipeline
  • func WithPipeStatus(status Status) OptionPipeline
  • func WithUpdateFilter(index int, filter *Filter) OptionPipeline
• type OptionPosition
  • func WithBinlogFile(b string) OptionPosition
  • func WithGTIDSet(g string) OptionPosition
  • func WithPos(p uint32) OptionPosition
• type Output
• type Pipeline
  • func (s *Pipeline) ExpectRun() bool
  • func (s *Pipeline) Key() (key string)
  • func (s *Pipeline) Unmarshal(val []byte) (err error)
  • func (s *Pipeline) Val() (val string)
• type PipelineH
  • func NewPipelineH() *PipelineH
  • func (m *PipelineH) String() string
• type Position
  • func (s *Position) Key() (key string)
  • func (s *Position) Unmarshal(val []byte) (err error)
  • func (s *Position) Val() (val string)
• type Sender
• type Status
• type Stdout

Constants

const SENDER_TYPE_KAFKA = "kafka"

const SNEDER_TYPE_HTTP = "http"

const SNEDER_TYPE_STDOUT = "stdout"

Types

type Filter

type Filter struct {
	Type FilterType `json:"type"`
	Rule string     `json:"rule"`
}

Filter of pipeline

type FilterType

type FilterType string

FilterType types of filter

const (
	// FILTER_WHITE white list
	FILTER_WHITE FilterType = "white"
	// FILTER_BLACK black list
	FILTER_BLACK FilterType = "black"
)

type Flavor

type Flavor string

Flavor mysql or mariaDB

const (
	// FLAVOR_MYSQL MySQL DB
	FLAVOR_MYSQL Flavor = "MySQL"
	// FLAVOR_MARIADB MariaDB
	FLAVOR_MARIADB Flavor = "MariaDB"
)

func (Flavor) YaString

func (f Flavor) YaString() string

YaString convert binlogo flavor string

type Http

type Http struct {
	API     string `json:"api"`
	Retries int    `json:"retries"`
}

type Instance

type Instance struct {
	PipelineName string    `json:"pipeline_name"`
	NodeName     string    `json:"node_name"`
	CreateTime   time.Time `json:"create_time"`
}

Instance store struct

type Kafka

type Kafka struct {
	Brokers      string                   `json:"brokers"`
	Topic        string                   `json:"topic"`
	RequiredAcks *sarama.RequiredAcks     `json:"require_acks"`
	Compression  *sarama.CompressionCodec `json:"compression"`
	Retries      *int                     `json:"retries"`
	Idepotent    *bool                    `json:"idepotent"`
}

type Mode

type Mode string

Mode mysql replication mode

const (
	// MODE_GTID GTID Mode
	MODE_GTID Mode = "gtid"
	// MODE_POSITION common Mode
	MODE_POSITION Mode = "position"
)

type Mysql

type Mysql struct {
	Address  string `json:"address"`
	Port     uint16 `json:"port"`
	User     string `json:"user"`
	Password string `json:"password"`
	ServerId uint32 `json:"server_id"`
	Flavor   Flavor `json:"flavor"`
	Mode     Mode   `json:"mode"`
}

Mysql store struct

type OptionPipeline

type OptionPipeline func(p *Pipeline)

OptionPipeline configure pipeline

func WithAddFilter

func WithAddFilter(filter *Filter) OptionPipeline

func WithPipeDelete

func WithPipeDelete(d bool) OptionPipeline

func WithPipeMode

func WithPipeMode(mode Mode) OptionPipeline

func WithPipeSafe

func WithPipeSafe(uPipe *Pipeline) OptionPipeline

WithPipeSafe sets pipeline

func WithPipeStatus

func WithPipeStatus(status Status) OptionPipeline

WithPipeStatus sets pipeline status

func WithUpdateFilter

func WithUpdateFilter(index int, filter *Filter) OptionPipeline

type OptionPosition

type OptionPosition func(position *Position)

OptionPosition Position options

func WithBinlogFile

func WithBinlogFile(b string) OptionPosition

WithBinlogFile sets binlog file to OptionPosition

func WithGTIDSet

func WithGTIDSet(g string) OptionPosition

WithGTIDSet sets GTIDSet to OptionPosition

func WithPos

func WithPos(p uint32) OptionPosition

WithPos sets binlog position to OptionPosition

type Output

type Output struct {
	Sender *Sender `json:"sender"`
}

type Pipeline

type Pipeline struct {
	Name       string    `json:"name"`
	Status     Status    `json:"status"`
	AliasName  string    `json:"aliasName"`
	Mysql      *Mysql    `json:"mysql"`
	Filters    []*Filter `json:"filters"`
	Output     *Output   `json:"output"`
	Replicas   int       `json:"replicas"`
	CreateTime time.Time `json:"create_time"`
	Remark     string    `json:"remark"`
	IsDelete   bool      `json:"is_delete"`
}

Pipeline pipeline's definition

func (*Pipeline) ExpectRun

func (s *Pipeline) ExpectRun() bool

ExpectRun determine whether the pipeline should run

func (*Pipeline) Key

func (s *Pipeline) Key() (key string)

Key generate etcd key

func (*Pipeline) Unmarshal

func (s *Pipeline) Unmarshal(val []byte) (err error)

Unmarshal generate from json data

func (*Pipeline) Val

func (s *Pipeline) Val() (val string)

Val generate json data

type PipelineH

type PipelineH struct {
	*Pipeline
	*model.Header `json:"header"`
}

func NewPipelineH

func NewPipelineH() *PipelineH

func (*PipelineH) String

func (m *PipelineH) String() string

type Position

type Position struct {
	BinlogFile     string `json:"binlog_file"`
	BinlogPosition uint32 `json:"binlog_position"`
	//GTIDSet        string `json:"gtid_set"`
	GTIDSet      string `json:"gtid_set"`
	PipelineName string `json:"pipeline_name"`
}

Position mysql replication position

func (*Position) Key

func (s *Position) Key() (key string)

Key get etcd key prefix

func (*Position) Unmarshal

func (s *Position) Unmarshal(val []byte) (err error)

Unmarshal unmarshal json data to object

func (*Position) Val

func (s *Position) Val() (val string)

Val get position json data

type Sender

type Sender struct {
	Name   string  `json:"name"`
	Type   string  `json:"type"`
	Kafka  *Kafka  `json:"kafka"`
	Stdout *Stdout `json:"stdout"`
	Http   *Http   `json:"http"`
}

type Status

type Status string

Status of Pipeline

const (
	// STATUS_RUN run
	STATUS_RUN Status = "run"
	// STATUS_STOP stop
	STATUS_STOP Status = "stop"
)

type Stdout

type Stdout struct {
}