gsnake

README

gsnake

The name "gsnake" is an acronym for "gluttonous snake". It is a data processing framework that helps you read and process logs from a variety of data format, such as plain text file, gz text file, pcap data file.

TODO

• Support priority path reading

Documentation

Index

• func GetAbsPath(p string) string
• func IsDir(dir string) bool
• func IsExist(path string) bool
• func LookupFiles(dir string, pattern string) ([]string, error)
• type DirReader
  • func NewPathReader(dir string) (*DirReader, error)
  • func (r *DirReader) GetPendingFileCount() int
  • func (r *DirReader) OnFileCreated(file string) (err error)
  • func (r *DirReader) OnFileModified(file string) (err error)
  • func (r *DirReader) Read() (err error)
  • func (r *DirReader) Stop()
  • func (r *DirReader) Wait() int
• type Dispatcher
  • func New() (*Dispatcher, error)
  • func (d *Dispatcher) Close()
  • func (d *Dispatcher) Register(m TextModule)
  • func (d *Dispatcher) Run()
  • func (d *Dispatcher) Stop()
• type FileProcessingTime
• type FileReader
• type FilesHandler
  • func NewFilesHandler(dir string) (h *FilesHandler, err error)
  • func (h *FilesHandler) OnFileCreated(file string)
  • func (h *FilesHandler) OnFileModified(file string)
  • func (h *FilesHandler) RecordPos() (err error)
  • func (h *FilesHandler) Run()
  • func (h *FilesHandler) Stop()
• type GzipFileReader
  • func NewGzipFileReader() *GzipFileReader
  • func (r *GzipFileReader) LoadFile(filepath string, fp *os.File) (err error)
  • func (r *GzipFileReader) ReadLine() (line []byte, err error)
• type PTailFileReader
  • func NewPTailFileReader() *PTailFileReader
  • func (r *PTailFileReader) LoadFile(filepath string, fp *os.File) (err error)
  • func (r *PTailFileReader) ReadLine() (line []byte, err error)
• type ProcessStatus
  • func NewProcessStatus(statusFile string) (ps *ProcessStatus, err error)
  • func (ps *ProcessStatus) Close()
  • func (ps *ProcessStatus) GetProcessedFiles() map[string]FileProcessingTime
  • func (ps *ProcessStatus) IsProcessed(file string) bool
  • func (ps *ProcessStatus) OnFileDeleted(path string)
  • func (ps *ProcessStatus) OnFileProcessingFinished(path string, startProcessing time.Time)
• type StringArray
  • func (ss StringArray) Len() int
  • func (ss StringArray) Less(i, j int) bool
  • func (ss StringArray) Swap(i, j int)
• type TextFileReader
• type TextFileTailReader
  • func NewTextFileTailReader(dr *DirReader) *TextFileTailReader
  • func (r *TextFileTailReader) ReadFile(file string, offset int) (err error)
• type TextModule

Functions

func GetAbsPath

func GetAbsPath(p string) string

GetAbsPath gets the absolute path of the giving path p

func IsDir

func IsDir(dir string) bool

IsDir returns true if given path is a directory, or returns false when it's a file or does not exist.

func IsExist

func IsExist(path string) bool

IsExist checks whether a file or directory exists. It returns false when the file or directory does not exist.

func LookupFiles

func LookupFiles(dir string, pattern string) ([]string, error)

Types

type DirReader

type DirReader struct {
	Running bool
	// contains filtered or unexported fields
}

func NewPathReader

func NewPathReader(dir string) (*DirReader, error)

func (*DirReader) GetPendingFileCount

func (r *DirReader) GetPendingFileCount() int

func (*DirReader) OnFileCreated

func (r *DirReader) OnFileCreated(file string) (err error)

func (*DirReader) OnFileModified

func (r *DirReader) OnFileModified(file string) (err error)

func (*DirReader) Read

func (r *DirReader) Read() (err error)

func (*DirReader) Stop

func (r *DirReader) Stop()

func (*DirReader) Wait

func (r *DirReader) Wait() int

type Dispatcher

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

func New

func New() (*Dispatcher, error)

func (*Dispatcher) Close

func (d *Dispatcher) Close()

func (*Dispatcher) Register

func (d *Dispatcher) Register(m TextModule)

func (*Dispatcher) Run

func (d *Dispatcher) Run()

func (*Dispatcher) Stop

func (d *Dispatcher) Stop()

type FileProcessingTime

type FileProcessingTime struct {
	Start time.Time
	End   time.Time
}

type FileReader

type FileReader interface {
	ReadFile(file string, offset int) (err error)
}

type FilesHandler

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

func NewFilesHandler

func NewFilesHandler(dir string) (h *FilesHandler, err error)

func (*FilesHandler) OnFileCreated

func (h *FilesHandler) OnFileCreated(file string)

func (*FilesHandler) OnFileModified

func (h *FilesHandler) OnFileModified(file string)

func (*FilesHandler) RecordPos

func (h *FilesHandler) RecordPos() (err error)

func (*FilesHandler) Run

func (h *FilesHandler) Run()

func (*FilesHandler) Stop

func (h *FilesHandler) Stop()

type GzipFileReader

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

func NewGzipFileReader

func NewGzipFileReader() *GzipFileReader

func (*GzipFileReader) LoadFile

func (r *GzipFileReader) LoadFile(filepath string, fp *os.File) (err error)

func (*GzipFileReader) ReadLine

func (r *GzipFileReader) ReadLine() (line []byte, err error)

type PTailFileReader

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

func NewPTailFileReader

func NewPTailFileReader() *PTailFileReader

func (*PTailFileReader) LoadFile

func (r *PTailFileReader) LoadFile(filepath string, fp *os.File) (err error)

func (*PTailFileReader) ReadLine

func (r *PTailFileReader) ReadLine() (line []byte, err error)

type ProcessStatus

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

func NewProcessStatus

func NewProcessStatus(statusFile string) (ps *ProcessStatus, err error)

func (*ProcessStatus) Close

func (ps *ProcessStatus) Close()

func (*ProcessStatus) GetProcessedFiles

func (ps *ProcessStatus) GetProcessedFiles() map[string]FileProcessingTime

func (*ProcessStatus) IsProcessed

func (ps *ProcessStatus) IsProcessed(file string) bool

func (*ProcessStatus) OnFileDeleted

func (ps *ProcessStatus) OnFileDeleted(path string)

func (*ProcessStatus) OnFileProcessingFinished

func (ps *ProcessStatus) OnFileProcessingFinished(path string, startProcessing time.Time)

type StringArray

type StringArray []string

func (StringArray) Len

func (ss StringArray) Len() int

func (StringArray) Less

func (ss StringArray) Less(i, j int) bool

func (StringArray) Swap

func (ss StringArray) Swap(i, j int)

type TextFileReader

type TextFileReader interface {
	LoadFile(filepath string, fp *os.File) error
	ReadLine() ([]byte, error)
}

type TextFileTailReader

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

func NewTextFileTailReader

func NewTextFileTailReader(dr *DirReader) *TextFileTailReader

func (*TextFileTailReader) ReadFile

func (r *TextFileTailReader) ReadFile(file string, offset int) (err error)

type TextModule

type TextModule interface {
	OnRecord([]byte)
}