reader

package

Go to main page

Versions in this module

v1

v1.5.5

May 18, 2020

v1.5.4

Aug 20, 2019

Changes in this version

+ func AndCondition(f1, f2 func(os.FileInfo) bool) func(os.FileInfo) bool

+ func CompressedFile(path string) bool

+ func GetLogPathAbs(conf conf.MapConf) (logpath string, err error)

+ func GetMaxFile(logdir string, condition func(os.FileInfo) bool, ...) (chosen os.FileInfo, err error)

+ func GetMetaOption(conf conf.MapConf) (string, string, string, error)

+ func GetMinFile(logdir string, condition func(os.FileInfo) bool, ...) (os.FileInfo, error)

+ func IgnoreFileSuffixes(file string, suffixes []string) bool

+ func IgnoreHidden(file string, ignoreHidden bool) bool

+ func InRunTime(hour, minute int, runTime RunTime) bool

+ func JoinFileInode(filename, inode string) string

+ func NoCondition(f os.FileInfo) bool

+ func NotCondition(f1 func(os.FileInfo) bool) func(os.FileInfo) bool

+ func OrCondition(f1, f2 func(os.FileInfo) bool) func(os.FileInfo) bool

+ func ParseNumber(str string) (number int, err error)

+ func ParseTime(timeStr string) (hour, minute int, err error)

+ func ValidFileRegex(file, validFilePattern string) bool

type Meta

+ LastKey string

+ func NewMetaWithRunnerName(runnerName, metadir, filedonedir, logpath, mode, tagfile string, ...) (m *Meta, err error)

+ func (m *Meta) Delete() error

+ func (m *Meta) GetEncodeTag() string

+ func (m *Meta) IsStatisticFileExist() bool

+ func (m *Meta) IsStatisticFileNotExist() bool

+ type NewSourceRecorder interface

+ NewSourceIndex func() []SourceIndex

+ type OnceReader interface

+ ReadDone func() bool

+ type RunTime struct

+ EndHour int

+ EndMin int

+ StartHour int

+ StartMin int

+ func ParseRunTime(runTimeStr string) (runTime RunTime, err error)

+ func ParseRunTimeWithMode(mode string, v interface{}) (runTime RunTime, err error)

+ func (r RunTime) Equal() bool

+ func (r RunTime) GreaterThanStart(hour, minute int) bool

+ func (r RunTime) LessThanEnd(hour, minute int) bool

+ type RunTimeReader interface

+ SetRunTime func(mode string, v interface{}) error

type Statistic

+ TransCnt map[string][2]int64

v1.5.3

Sep 4, 2018

Changes in this version

+ const DataTypeHash

+ const DataTypeSortedSet

+ const FtSaveLogPath

+ const KeyMssqlSchema

+ const KeyPGsqlSchema

+ const KeySocketRule

+ const KeySocketSplitByLine

+ const KeySubmetaExpire

+ const SocketRuleJson

+ const SocketRuleLine

+ const SocketRulePacket

+ var OptionKeySubmetaExpire = Option

type Meta

+ func (m *Meta) CheckExpiredSubMetas(expire time.Duration)

+ func (m *Meta) CleanExpiredSubMetas(expire time.Duration)

v1.5.2

Jul 30, 2018

Changes in this version

+ const KeySnmpTableInitHost

+ const ModeDirx

+ var DefaultIgnoreFileSuffixes = []string

+ var OptionKeyIgnoreFileSuffix = Option

+ var OptionKeyIgnoreHiddenFile = Option

+ var OptionKeyMaxOpenFiles = Option

+ var OptionKeyStatInterval = Option

+ type DaemonReader interface

+ Start func() error

type Statistic

+ ParseErrors ErrorQueue

+ ReadErrors ErrorQueue

+ SendErrors map[string]ErrorQueue

+ TransformErrors map[string]ErrorQueue

v1.5.1

Jul 13, 2018

Changes in this version

+ const KeyMysqlEncoding

+ var ModeToolTips = []KeyValue

type Reader

+ func NewReader(conf conf.MapConf, errDirectReturn bool) (reader Reader, err error)

v1.5.0

Jun 11, 2018

Changes in this version

+ const DefaultBufSize

+ const DefaultHTTPServiceAddress

+ const DefaultHTTPServicePath

+ const DefautFileRetention

+ const DoneFileName

+ const KeyHTTPServiceAddress

+ const KeyHTTPServicePath

+ const KeyMysqlHistoryAll

+ const KyeMysqlTable

+ const ModeHTTP

+ const ModeMSSQL

+ const ModeMySQL

+ const ModePostgreSQL

+ func ParseLoopDuration(cronSched string) (dur time.Duration, err error)

+ func RegisterConstructor(typ string, c Constructor)

type BufReader

+ Meta *Meta

+ func (b *BufReader) FormMutiLine() []byte

+ type Constructor func(*Meta, conf.MapConf) (Reader, error)

+ type DataReader interface

+ ReadData func() (Data, int64, error)

type Meta

+ Dir string

+ Readlimit int

+ TagFile string

+ func (m *Meta) AddSubMeta(key string, meta *Meta) error

+ func (m *Meta) ReadRecordsFile(recordsFile string) ([]string, error)

+ func (m *Meta) RemoveSubMeta(key string)

+ type Registry struct

+ func NewRegistry() *Registry

+ func (reg *Registry) NewReader(conf conf.MapConf, errDirectReturn bool) (reader Reader, err error)

+ func (reg *Registry) NewReaderWithMeta(conf conf.MapConf, meta *Meta, errDirectReturn bool) (reader Reader, err error)

+ func (reg *Registry) RegisterReader(readerType string, constructor Constructor) error

type SeqFile

+ SkipFileFirstLine bool

v1.4.9

May 24, 2018

v1.4.8

May 23, 2018

Changes in this version

+ const DefaultMsSQL

+ const DefaultMySQL

+ const DefaultPGSQL

+ const KeyMagicLagDuration

+ const KeySkipFileFirstLine

+ var OptionKeySkipFileFirstLine = Option

+ var OptionMagicLagDuration = Option

+ func GetDefualtMetaStore(bucket, prefix, region, ak, sk string) string

+ func GetDefualtSyncDir(bucket, prefix, region, ak, sk string) string

+ func GetS3UserInfo(conf conf.MapConf) (bucket, prefix, region, ak, sk string, err error)

+ func WriteDBDoneFile(doneFilePath, database, content string) (err error)

type CloudTrailReader

+ func (ctr *CloudTrailReader) Reset() (err error)

+ type LineSkipper interface

+ IsNewOpen func() bool

+ SetSkipped func()

type Meta

+ func (m *Meta) AppendDoneFileInode(path string, inode uint64) (err error)

+ func (m *Meta) GetDoneFileContent() ([]string, error)

+ func (m *Meta) GetDoneFileInode() map[uint64]bool

+ func (m *Meta) ReadDBDoneFile(database string) (content []string, err error)

+ type NewLineBytesRecorder interface

+ NewLineBytesIndex func() []SourceIndex

type SeqFile

+ func (sf *SeqFile) IsNewOpen() bool

+ func (sf *SeqFile) NewLineBytesIndex() []SourceIndex

+ func (sf *SeqFile) SetSkipped()

+ type SourceIndex struct

+ Index int

+ Source string

v1.4.7

Apr 17, 2018

Changes in this version

+ const DefaultSyncDirectory

+ const DefaultSyncMetaStore

+ const KeyAWSAccessKey

+ const KeyAWSProfile

+ const KeyAWSSecretKey

+ const KeyAWSToken

+ const KeyCacheTTL

+ const KeyCollectInterval

+ const KeyDelay

+ const KeyDimension

+ const KeyErrDirectReturn

+ const KeyMetrics

+ const KeyNamespace

+ const KeyPeriod

+ const KeyRateLimit

+ const KeyRegion

+ const KeyRoleArn

+ const KeyS3AccessKey

+ const KeyS3Bucket

+ const KeyS3Prefix

+ const KeyS3Region

+ const KeyS3SecretKey

+ const KeySharedCredentialFile

+ const KeySyncConcurrent

+ const KeySyncDirectory

+ const KeySyncInterval

+ const KeySyncMetastore

+ const ModeCloudTrail

+ const ModeCloudWatch

+ func SelectMetrics(c *CloudWatch) ([]*cloudwatch.Metric, error)

+ type CloudTrailReader struct

+ func (ctr *CloudTrailReader) Close() error

+ type CloudWatch struct

+ CacheTTL time.Duration

+ CollectInterval time.Duration

+ DataChan chan models.Data

+ Delay time.Duration

+ Metrics []*Metric

+ Namespace string

+ Period time.Duration

+ RateLimit int64

+ Region string

+ StopChan chan struct{}

+ func (c *CloudWatch) Close() error

+ func (c *CloudWatch) Gather() error

+ func (c *CloudWatch) Name() string

+ func (c *CloudWatch) ReadLine() (line string, err error)

+ func (c *CloudWatch) SetMode(mode string, v interface{}) error

+ func (c *CloudWatch) Source() string

+ func (c *CloudWatch) Start() error

+ func (c *CloudWatch) SyncMeta()

+ type CredentialConfig struct

+ AccessKey string

+ Filename string

+ Profile string

+ Region string

+ RoleARN string

+ SecretKey string

+ Token string

+ func (c *CredentialConfig) Credentials() (client.ConfigProvider, error)

+ type Dimension struct

+ Name string

+ Value string

+ type Metric struct

+ Dimensions []*Dimension

+ MetricNames []string

+ type MetricCache struct

+ Fetched time.Time

+ Metrics []*cloudwatch.Metric

+ TTL time.Duration

+ func (c *MetricCache) IsValid() bool

type Reader

+ func NewCloudTrailReader(meta *Meta, conf conf.MapConf) (Reader, error)

+ func NewCloudWatchReader(meta *Meta, conf conf.MapConf) (c Reader, err error)

+ func NewFileDirReader(meta *Meta, conf conf.MapConf) (reader Reader, err error)

+ func NewSingleFileReader(meta *Meta, conf conf.MapConf) (reader Reader, err error)

+ type ReaderRegistry struct

+ func NewReaderRegistry() *ReaderRegistry

+ func (r *ReaderRegistry) NewReader(conf conf.MapConf, errDirectReturn bool) (reader Reader, err error)

+ func (r *ReaderRegistry) NewReaderWithMeta(conf conf.MapConf, meta *Meta, errDirectReturn bool) (reader Reader, err error)

+ func (registry *ReaderRegistry) RegisterReader(readerType string, constructor func(*Meta, conf.MapConf) (Reader, error)) error

v1.4.6

Mar 28, 2018

Changes in this version

type Meta

+ func (m *Meta) ExtraInfo() map[string]string

v1.4.5

Mar 14, 2018

Changes in this version

+ const KeyNewFileNewLine

+ var OptionKeyNewFileNewLine = Option

+ var OptionKeyValidFilePattern = Option

v1.4.4

Mar 5, 2018

Changes in this version

type ActiveReader

+ func (ar *ActiveReader) Lag() (rl *models.LagInfo, err error)

type BufReader

+ func (b *BufReader) Lag() (rl *models.LagInfo, err error)

type KafkaReader

+ func (kr *KafkaReader) Lag() (rl *models.LagInfo, err error)

+ type LagReader interface

+ Lag func() (*models.LagInfo, error)

type MultiReader

+ func (mr *MultiReader) Lag() (rl *models.LagInfo, err error)

type SeqFile

+ func (sf *SeqFile) Lag() (rl *models.LagInfo, err error)

type SingleFile

+ func (sf *SingleFile) Lag() (rl *models.LagInfo, err error)

type SnmpReader

+ Status int32

+ StopChan chan struct{}

+ func (s *SnmpReader) Start() error

v1.4.3

Feb 6, 2018

Changes in this version

+ const ChannelTooShortErr

+ const KeyExecInterpreter

+ const KeyKafkaZookeeperChroot

+ const KeyScriptCron

+ const KeyScriptExecOnStart

+ const KeySnmpReaderAgents

+ const KeySnmpReaderAuthPassword

+ const KeySnmpReaderAuthProtocol

+ const KeySnmpReaderCommunity

+ const KeySnmpReaderContextName

+ const KeySnmpReaderEngineBoots

+ const KeySnmpReaderEngineID

+ const KeySnmpReaderEngineTime

+ const KeySnmpReaderFields

+ const KeySnmpReaderInterval

+ const KeySnmpReaderMaxRepetitions

+ const KeySnmpReaderName

+ const KeySnmpReaderPrivPassword

+ const KeySnmpReaderPrivProtocol

+ const KeySnmpReaderRetries

+ const KeySnmpReaderSecLevel

+ const KeySnmpReaderSecName

+ const KeySnmpReaderTables

+ const KeySnmpReaderTimeOut

+ const KeySnmpReaderVersion

+ const KeySnmpTableName

+ const KeyTagFile

+ const KeyTimestamp

+ const MaxChannelLength

+ const ModeScript

+ const ModeSnmp

+ func Errorf(err error, msg string, format ...interface{}) error

+ type Field struct

+ Conversion string

+ IsTag bool

+ Name string

+ Oid string

+ OidIndexSuffix string

type Meta

+ func (m *Meta) GetTagFile() string

+ func (m *Meta) GetTags() map[string]interface{}

type MultiReader

+ func (mr *MultiReader) Reset() (err error)

+ type NestedError struct

+ Err error

+ NestedErr error

+ func (ne NestedError) Error() string

+ type RTable struct

+ Name string

+ Rows []RTableRow

+ Time string

+ type RTableRow struct

+ Fields map[string]interface{}

+ Tags map[string]string

+ type ScriptReader struct

+ Cron *cron.Cron

+ func NewScriptReader(meta *Meta, conf conf.MapConf) (sr *ScriptReader, err error)

+ func (sr *ScriptReader) Close() (err error)

+ func (sr *ScriptReader) LoopRun()

+ func (sr *ScriptReader) Name() string

+ func (sr *ScriptReader) ReadLine() (data string, err error)

+ func (sr *ScriptReader) SetMode(mode string, v interface{}) error

+ func (sr *ScriptReader) Source() string

+ func (sr *ScriptReader) Start()

+ func (sr *ScriptReader) SyncMeta()

+ type SnmpReader struct

+ Agents []string

+ AuthPassword string

+ AuthProtocol string

+ Community string

+ ConnectionCache []snmpConnection

+ ContextName string

+ DataChan chan interface{}

+ EngineBoots uint32

+ EngineID string

+ EngineTime uint32

+ Fields []Field

+ Interval time.Duration

+ LastGather time.Time

+ MaxRepetitions uint8

+ Meta *Meta

+ PrivPassword string

+ PrivProtocol string

+ Retries int

+ SecLevel string

+ SecName string

+ SnmpName string

+ Tables []Table

+ Timeout time.Duration

+ Version uint8

+ func NewSnmpReader(meta *Meta, c conf.MapConf) (s *SnmpReader, err error)

+ func (s *SnmpReader) Close() error

+ func (s *SnmpReader) Gather() (err error)

+ func (s *SnmpReader) Name() string

+ func (s *SnmpReader) ReadLine() (line string, err error)

+ func (s *SnmpReader) SetMode(mode string, v interface{}) error

+ func (s *SnmpReader) Source() string

+ func (s *SnmpReader) StoreData(data []map[string]interface{}) (err error)

+ func (s *SnmpReader) SyncMeta()

+ type Table struct

+ Fields []Field

+ IndexAsTag bool

+ InheritTags []string

+ Name string

+ Oid string

+ func (t Table) Build(gs snmpConnection, walk bool) (*RTable, error)

v1.4.2

Jan 4, 2018

Changes in this version

+ const ContentEncodingHeader

+ const ContentTypeHeader

+ const DefaultHttpServiceAddress

+ const DefaultHttpServicePath

+ const DefaultMaxBodySize

+ const DefaultMaxBytesPerFile

+ const DefaultSyncEvery

+ const DefaultWriteSpeedLimit

+ const KeyHttpServiceAddress

+ const KeyHttpServicePath

+ const ModeHttp

+ const StatusStopping

+ type HttpReader struct

+ func NewHttpReader(meta *Meta, conf conf.MapConf) (*HttpReader, error)

+ func (h *HttpReader) Close() error

+ func (h *HttpReader) Name() string

+ func (h *HttpReader) ReadLine() (data string, err error)

+ func (h *HttpReader) SetMode(mode string, v interface{}) error

+ func (h *HttpReader) Source() string

+ func (h *HttpReader) Start() error

+ func (h *HttpReader) SyncMeta()

v1.4.1

Dec 7, 2017

Changes in this version

+ const DataTypeSet

+ const DateTypeHash

+ const DateTypeSortedSet

+ const KeyPGsqlCron

+ const KeyPGsqlDataBase

+ const KeyPGsqlDataSource

+ const KeyPGsqlExecOnStart

+ const KeyPGsqlOffsetKey

+ const KeyPGsqlReadBatch

+ const KeyPGsqlSQL

+ const KeyRedisHashArea

+ const ModeFileAuto

+ const ModePG

+ var ElasticVersion3 = "3.x"

+ var ElasticVersion6 = "6.x"

+ var OptionBuffSize = utils.Option

+ var OptionEncoding = utils.Option

+ var OptionHeadPattern = utils.Option

+ var OptionMetaPath = utils.Option

+ var OptionReadIoLimit = utils.Option

+ var OptionSQLSchema = utils.Option

+ var OptionWhence = utils.Option

type Reader

+ func NewFileAutoReader(conf conf.MapConf, meta *Meta, isFromWeb bool, bufSize int, whence string, ...) (reader Reader, err error)

v1.4.0

Nov 27, 2017

v1.3.6

Nov 22, 2017

Changes in this version

+ const DataTypeString

+ const KeySocketKeepAlivePeriod

+ const KeySocketMaxConnections

+ const KeySocketReadBufferSize

+ const KeySocketReadTimeout

+ const KeySocketServiceAddress

+ const ModeMetrics

+ const ModeSocket

+ type SocketReader struct

+ Closer io.Closer

+ KeepAlivePeriod time.Duration

+ MaxConnections int

+ ReadBufferSize int

+ ReadChan chan string

+ ReadTimeout time.Duration

+ ServiceAddress string

+ func NewSocketReader(meta *Meta, conf conf.MapConf) (*SocketReader, error)

+ func (sr *SocketReader) Close() (err error)

+ func (sr *SocketReader) Name() string

+ func (sr *SocketReader) ReadLine() (data string, err error)

+ func (sr *SocketReader) SetMode(mode string, v interface{}) error

+ func (sr *SocketReader) Source() string

+ func (sr *SocketReader) Start() error

+ func (sr *SocketReader) SyncMeta()

v1.3.5

Oct 31, 2017

Changes in this version

+ const KeyKafkaZookeeperTimeout

type KafkaReader

+ ZookeeperTimeout time.Duration

type Meta

+ func (m *Meta) FtSaveLogPath() string

+ func (m *Meta) ReadStatistic() (stat Statistic, err error)

+ func (m *Meta) StatisticFile() string

+ func (m *Meta) WriteStatistic(stat *Statistic) error

+ type Statistic struct

+ ParserCnt [2]int64

+ ReaderCnt int64

+ SenderCnt map[string][2]int64

v1.3.4

Oct 17, 2017

Changes in this version

+ const Loop

v1.3.3

Oct 10, 2017

Changes in this version

+ var OptionDataSourceTag = utils.Option

type MongoReader

+ func (mr *MongoReader) LoopRun()

type SqlReader

+ func (mr *SqlReader) LoopRun()

v1.3.2

Sep 27, 2017

v1.3.1

Sep 13, 2017

Changes in this version

type ActiveReader

+ func (ar *ActiveReader) Status() utils.StatsInfo

type BufReader

+ func (b *BufReader) Status() utils.StatsInfo

type ElasticReader

+ func (er *ElasticReader) Status() utils.StatsInfo

type KafkaReader

+ func (kr *KafkaReader) Status() utils.StatsInfo

type Meta

+ func (m *Meta) IsFileMode() bool

type MongoReader

+ func (mr *MongoReader) Status() utils.StatsInfo

type MultiReader

+ func (mr *MultiReader) Status() utils.StatsInfo

type RedisReader

+ func (rr *RedisReader) Status() utils.StatsInfo

type SqlReader

+ func (mr *SqlReader) Status() utils.StatsInfo

+ type StatsReader interface

+ Name func() string

+ Status func() utils.StatsInfo

v1.3.0

Sep 5, 2017

v1.2.4

Aug 29, 2017

Changes in this version

type Meta

+ func (b *Meta) Reset() error

v1.2.3

Aug 16, 2017

Changes in this version

+ var ModeKeyOptions = map[string][]utils.Option

+ var ModeUsages = []utils.KeyValue

v1.2.2

Aug 1, 2017

Changes in this version

+ type Result struct

v1.2.1

Jul 18, 2017

Changes in this version

+ const DataTypeChannel

+ const DataTypeList

+ const DataTypePatterChannel

+ const KeyESVersion

+ const KeyRedisAddress

+ const KeyRedisDB

+ const KeyRedisDataType

+ const KeyRedisKey

+ const KeyRedisPassword

+ const KeySQLSchema

+ const KeyTimeoutDuration

+ const ModeRedis

+ var ElasticVersion2 = "2.x"

+ var ElasticVersion5 = "5.x"

+ type RedisOptionn struct

+ type RedisReader struct

+ func NewRedisReader(meta *Meta, conf conf.MapConf) (rr *RedisReader, err error)

+ func (rr *RedisReader) Close() (err error)

+ func (rr *RedisReader) Name() string

+ func (rr *RedisReader) ReadLine() (data string, err error)

+ func (rr *RedisReader) SetMode(mode string, v interface{}) error

+ func (rr *RedisReader) Source() string

+ func (rr *RedisReader) Start()

+ func (rr *RedisReader) SyncMeta()

+ type ServerReader interface

+ Close func() error

+ Name func() string

+ ReadLine func() (string, error)

+ Source func() string

+ Start func()

+ SyncMeta func()

v1.1.0

Jun 27, 2017

Changes in this version

+ const DirMode

+ const FileMode

+ const KeyBufSize

+ const KeyDataSourceTag

+ const KeyESHost

+ const KeyESIndex

+ const KeyESKeepAlive

+ const KeyESReadBatch

+ const KeyESType

+ const KeyEncoding

+ const KeyExpire

+ const KeyFileDone

+ const KeyHeadPattern

+ const KeyIgnoreFileSuffix

+ const KeyIgnoreHiddenFile

+ const KeyKafkaGroupID

+ const KeyKafkaTopic

+ const KeyKafkaZookeeper

+ const KeyLogPath

+ const KeyMaxOpenFiles

+ const KeyMetaPath

+ const KeyMode

+ const KeyMongoCert

+ const KeyMongoCollection

+ const KeyMongoCron

+ const KeyMongoDatabase

+ const KeyMongoExecOnstart

+ const KeyMongoFilters

+ const KeyMongoHost

+ const KeyMongoOffsetKey

+ const KeyMongoReadBatch

+ const KeyMssqlCron

+ const KeyMssqlDataBase

+ const KeyMssqlDataSource

+ const KeyMssqlExecOnStart

+ const KeyMssqlOffsetKey

+ const KeyMssqlReadBatch

+ const KeyMssqlSQL

+ const KeyMysqlCron

+ const KeyMysqlDataBase

+ const KeyMysqlDataSource

+ const KeyMysqlExecOnStart

+ const KeyMysqlOffsetKey

+ const KeyMysqlReadBatch

+ const KeyMysqlSQL

+ const KeyReadIOLimit

+ const KeyRunnerName

+ const KeyStatInterval

+ const KeyValidFilePattern

+ const KeyWhence

+ const MaxHeadPatternBufferSize

+ const ModeDir

+ const ModeElastic

+ const ModeFile

+ const ModeKafka

+ const ModeMongo

+ const ModeMssql

+ const ModeMysql

+ const ModeTailx

+ const MongoDefaultOffsetKey

+ const ReadModeHeadPatternRegexp

+ const ReadModeHeadPatternString

+ const SQL_SPLITER

+ const StatusInit

+ const StatusRunning

+ const StatusStoping

+ const StatusStopped

+ const WhenceNewest

+ const WhenceOldest

+ var ErrBufferFull = errors.New("bufio: buffer full")

+ var ErrFileNotDir = errors.New("file is not directory")

+ var ErrFileNotRegular = errors.New("file is not regular")

+ var ErrInvalidUnreadByte = errors.New("bufio: invalid use of UnreadByte")

+ var ErrInvalidUnreadRune = errors.New("bufio: invalid use of UnreadRune")

+ var ErrMetaFileRead = errors.New("cannot read meta file")

+ var ErrNegativeCount = errors.New("bufio: negative count")

+ var ErrNoFileChosen = errors.New("no files found")

+ var ErrStopped = errors.New("runner stopped")

+ var WaitNoSuchFile = time.Second

+ func HeadPatternMode(mode string, v interface{}) (reg *regexp.Regexp, err error)

+ type ActiveReader struct

+ func NewActiveReader(logPath, whence string, meta *Meta) (ar *ActiveReader, err error)

+ func (ar *ActiveReader) Close() error

+ func (ar *ActiveReader) Run()

+ func (ar *ActiveReader) SyncMeta()

+ type BufReader struct

+ func NewReaderSize(rd FileReader, meta *Meta, size int) (*BufReader, error)

+ func (b *BufReader) Close() error

+ func (b *BufReader) Name() string

+ func (b *BufReader) ReadLine() (ret string, err error)

+ func (b *BufReader) ReadPattern() (string, error)

+ func (b *BufReader) ReadString(delim byte) (ret string, err error)

+ func (b *BufReader) SetMode(mode string, v interface{}) (err error)

+ func (b *BufReader) Source() string

+ func (b *BufReader) SyncMeta()

+ type CollectionFilter map[string]interface

+ type ElasticReader struct

+ func NewESReader(meta *Meta, readBatch int, estype, esindex, eshost, keepAlive string) (er *ElasticReader, err error)

+ func (er *ElasticReader) Close() (err error)

+ func (er *ElasticReader) Name() string

+ func (er *ElasticReader) ReadLine() (data string, err error)

+ func (er *ElasticReader) SetMode(mode string, v interface{}) error

+ func (er *ElasticReader) Source() string

+ func (er *ElasticReader) Start()

+ func (er *ElasticReader) SyncMeta()

+ type FileReader interface

+ Close func() error

+ Name func() string

+ Read func(p []byte) (n int, err error)

+ Source func() string

+ SyncMeta func() error

+ type KafkaReader struct

+ Consumer *consumergroup.ConsumerGroup

+ ConsumerGroup string

+ Topics []string

+ Whence string

+ ZookeeperChroot string

+ ZookeeperPeers []string

+ func NewKafkaReader(meta *Meta, consumerGroup string, topics []string, zookeeper []string, ...) (kr *KafkaReader, err error)

+ func (kr *KafkaReader) Close() (err error)

+ func (kr *KafkaReader) Name() string

+ func (kr *KafkaReader) ReadLine() (data string, err error)

+ func (kr *KafkaReader) SetMode(mode string, v interface{}) error

+ func (kr *KafkaReader) Source() string

+ func (kr *KafkaReader) Start()

+ func (kr *KafkaReader) SyncMeta()

+ type LastSync struct

+ type Meta struct

+ RunnerName string

+ func NewMeta(metadir, filedonedir, logpath, mode string, donefileRetention int) (m *Meta, err error)

+ func NewMetaWithConf(conf conf.MapConf) (meta *Meta, err error)

+ func (m *Meta) AppendDeleteFile(path string) (err error)

+ func (m *Meta) AppendDoneFile(path string) (err error)

+ func (m *Meta) BufFile() string

+ func (m *Meta) BufMetaFile() string

+ func (m *Meta) CacheLineFile() string

+ func (m *Meta) Clear() error

+ func (m *Meta) DeleteDoneFile(path string) error

+ func (m *Meta) DeleteFile() string

+ func (m *Meta) DoneFile() string

+ func (m *Meta) DoneFilePath() string

+ func (m *Meta) GetDataSourceTag() string

+ func (m *Meta) GetDoneFiles() (doneFiles []utils.File, err error)

+ func (m *Meta) GetEncodingWay() (e string)

+ func (m *Meta) GetMode() string

+ func (m *Meta) IsDoneFile(file string) bool

+ func (m *Meta) IsExist() bool

+ func (m *Meta) IsNotExist() bool

+ func (m *Meta) IsNotValid() bool

+ func (m *Meta) IsValid() bool

+ func (m *Meta) LogPath() string

+ func (m *Meta) MetaFile() string

+ func (m *Meta) ReadBuf(buf []byte) (n int, err error)

+ func (m *Meta) ReadBufMeta() (r, w, bufsize int, err error)

+ func (m *Meta) ReadCacheLine() ([]byte, error)

+ func (m *Meta) ReadOffset() (currFile string, offset int64, err error)

+ func (m *Meta) SetEncodingWay(e string)

+ func (m *Meta) WriteBuf(buf []byte, r, w, bufsize int) (err error)

+ func (m *Meta) WriteCacheLine(lines string) error

+ func (m *Meta) WriteOffset(currFile string, offset int64) (err error)

+ type MongoReader struct

+ Cron *cron.Cron

+ func NewMongoReader(meta *Meta, readBatch int, ...) (mr *MongoReader, err error)

+ func (mr *MongoReader) Close() (err error)

+ func (mr *MongoReader) Name() string

+ func (mr *MongoReader) ReadLine() (data string, err error)

+ func (mr *MongoReader) SetMode(mode string, v interface{}) error

+ func (mr *MongoReader) Source() string

+ func (mr *MongoReader) Start()

+ func (mr *MongoReader) SyncMeta()

+ type MultiReader struct

+ func NewMultiReader(meta *Meta, logPathPattern, whence, expireDur, statIntervalDur string, ...) (mr *MultiReader, err error)

+ func (mr *MultiReader) Close() (err error)

+ func (mr *MultiReader) Expire()

+ func (mr *MultiReader) Name() string

+ func (mr *MultiReader) ReadLine() (data string, err error)

+ func (mr *MultiReader) SetMode(mode string, value interface{}) (err error)

+ func (mr *MultiReader) Source() string

+ func (mr *MultiReader) Start()

+ func (mr *MultiReader) StatLogPath()

+ func (mr *MultiReader) SyncMeta()

+ type Reader interface

+ Close func() error

+ Name func() string

+ ReadLine func() (string, error)

+ SetMode func(mode string, v interface{}) error

+ Source func() string

+ SyncMeta func()

+ func NewFileBufReader(conf conf.MapConf) (reader Reader, err error)

+ func NewFileBufReaderWithMeta(conf conf.MapConf, meta *Meta) (reader Reader, err error)

+ type SeqFile struct

+ func NewSeqFile(meta *Meta, path string, ignoreHidden bool, suffixes []string, ...) (sf *SeqFile, err error)

+ func (sf *SeqFile) Close() (err error)

+ func (sf *SeqFile) Name() string

+ func (sf *SeqFile) Read(p []byte) (n int, err error)

+ func (sf *SeqFile) Source() string

+ func (sf *SeqFile) SyncMeta() (err error)

+ type SingleFile struct

+ func NewSingleFile(meta *Meta, path, whence string) (sf *SingleFile, err error)

+ func (sf *SingleFile) Close() (err error)

+ func (sf *SingleFile) Name() string

+ func (sf *SingleFile) Read(p []byte) (n int, err error)

+ func (sf *SingleFile) Reopen() (err error)

+ func (sf *SingleFile) Source() string

+ func (sf *SingleFile) SyncMeta() error

+ type SqlReader struct

+ Cron *cron.Cron

+ func NewSQLReader(meta *Meta, readBatch int, ...) (mr *SqlReader, err error)

+ func (mr *SqlReader) Close() (err error)

+ func (mr *SqlReader) Name() string

+ func (mr *SqlReader) ReadLine() (data string, err error)

+ func (mr *SqlReader) SetMode(mode string, v interface{}) error

+ func (mr *SqlReader) Source() string

+ func (mr *SqlReader) Start()

+ func (mr *SqlReader) SyncMeta()

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL