persistence

Documentation

Index

• Constants
• func SetConfig(dbConfig DbConfig)
• func StopDbHandler()
• type DbConfig
  • func GetConfig() DbConfig
  • func (d *DbConfig) SetDefaults()
• type DbHandler
  • func GetOrCreateShareDbHandler() *DbHandler
  • func NewDbHandler(config DbConfig) *DbHandler
  • func (d *DbHandler) FindAll() []reg.Registry
  • func (d *DbHandler) FindBy(jobUid string, sourceName string, pipelineName string) reg.Registry
  • func (d *DbHandler) HandleOpt(opt DbOpt)
  • func (d *DbHandler) Stop()
  • func (d *DbHandler) String() string
• type DbOpt
• type DbOptType
• type State
  • func (s *State) AppendTags(tag string)

Constants

const (
	DriverSqlite = "sqlite"
	DriverBadger = "badger"

	TimeFormatPattern = "2006-01-02 15:04:05.999"

	DeleteByIdOpt               = DbOptType(1)
	DeleteByJobUidOpt           = DbOptType(2)
	UpsertOffsetByJobWatchIdOpt = DbOptType(3)
	UpdateNameByJobWatchIdOpt   = DbOptType(4)
)

Functions

func SetConfig

func SetConfig(dbConfig DbConfig)

func StopDbHandler

func StopDbHandler()

Types

type DbConfig

type DbConfig struct {
	File                 string        `yaml:"file,omitempty"`
	FlushTimeout         time.Duration `yaml:"flushTimeout,omitempty" default:"2s"`
	BufferSize           int           `yaml:"bufferSize,omitempty" default:"2048"`
	CleanInactiveTimeout time.Duration `yaml:"cleanInactiveTimeout,omitempty" default:"504h"` // default records not updated in 21 days will be deleted
	CleanScanInterval    time.Duration `yaml:"cleanScanInterval,omitempty" default:"1h"`
}

func GetConfig

func GetConfig() DbConfig

func (*DbConfig) SetDefaults

func (d *DbConfig) SetDefaults()

type DbHandler

type DbHandler struct {
	State chan *State
	// contains filtered or unexported fields
}

func GetOrCreateShareDbHandler

func GetOrCreateShareDbHandler() *DbHandler

func NewDbHandler

func NewDbHandler(config DbConfig) *DbHandler

func (*DbHandler) FindAll

func (d *DbHandler) FindAll() []reg.Registry

func (*DbHandler) FindBy

func (d *DbHandler) FindBy(jobUid string, sourceName string, pipelineName string) reg.Registry

func (*DbHandler) HandleOpt

func (d *DbHandler) HandleOpt(opt DbOpt)

func (*DbHandler) Stop

func (d *DbHandler) Stop()

func (*DbHandler) String

func (d *DbHandler) String() string

type DbOpt

type DbOpt struct {
	R           reg.Registry
	OptType     DbOptType
	Immediately bool
}

type DbOptType

type DbOptType int

type State

type State struct {
	Epoch        *pipeline.Epoch `json:"-"`
	PipelineName string          `json:"-"`
	SourceName   string          `json:"-"`
	Offset       int64           `json:"offset"`
	NextOffset   int64           `json:"nextOffset"`
	Filename     string          `json:"filename,omitempty"`
	CollectTime  time.Time       `json:"collectTime,omitempty"`
	ContentBytes int64           `json:"contentBytes"`
	JobUid       string          `json:"jobUid,omitempty"`
	JobIndex     uint32          `json:"-"`
	EventUid     string          `json:"-"`
	LineNumber   int64           `json:"lineNumber,omitempty"`
	Tags         string          `json:"tags,omitempty"`

	// for cache
	WatchUid string

	// JobFields from job
	JobFields map[string]interface{}
}

func (*State) AppendTags

func (s *State) AppendTags(tag string)