mgr

Documentation

Index

• Constants
• Variables
• func NewMetric(tp string) (metric.Collector, error)
• type CleanInfo
• type LogExportRunner
  • func NewLogExportRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal, ...) (runner *LogExportRunner, err error)
  • func NewLogExportRunnerWithService(info RunnerInfo, reader reader.Reader, cleaner *cleaner.Cleaner, ...) (runner *LogExportRunner, err error)
  • func (r *LogExportRunner) Cleaner() CleanInfo
  • func (r *LogExportRunner) LagStats() (rl RunnerLag, err error)
  • func (r *LogExportRunner) Name() string
  • func (r *LogExportRunner) Run()
  • func (r *LogExportRunner) Status() RunnerStatus
  • func (r *LogExportRunner) Stop()
• type Manager
  • func NewCustomManager(conf ManagerConfig, pr *parser.ParserRegistry, sr *sender.SenderRegistry) (*Manager, error)
  • func NewManager(conf ManagerConfig) (*Manager, error)
  • func (m *Manager) Add(confPath string)
  • func (m *Manager) Remove(confPath string)
  • func (m *Manager) Status() (rss map[string]RunnerStatus)
  • func (m *Manager) Stop() error
  • func (m *Manager) Watch(confsPath []string) (err error)
• type ManagerConfig
• type MetricRunner
  • func NewMetricRunner(rc RunnerConfig, sr *sender.SenderRegistry) (runner *MetricRunner, err error)
  • func (_ *MetricRunner) Cleaner() CleanInfo
  • func (mr *MetricRunner) Name() string
  • func (r *MetricRunner) Run()
  • func (mr *MetricRunner) Status() RunnerStatus
  • func (mr *MetricRunner) Stop()
• type RestService
  • func NewRestService(mgr *Manager) *RestService
  • func (rs *RestService) GetStatus(rw http.ResponseWriter, req *http.Request)
  • func (rs *RestService) Stop()
• type Runner
  • func NewCustomRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal, ...) (runner Runner, err error)
  • func NewRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal) (runner Runner, err error)
  • func NewRunnerWithService(info RunnerInfo, reader reader.Reader, cleaner *cleaner.Cleaner, ...) (runner Runner, err error)
• type RunnerConfig
  • func Compatible(rc RunnerConfig) RunnerConfig
• type RunnerInfo
• type RunnerLag
• type RunnerStatus

Constants

const (
	StatsShell = "stats"
	PREFIX     = "/logkit"
)

const (
	KeyMetricType = "type"
)

Variables

var DEFAULT_PORT = 4000

var DIR_NOT_EXIST_SLEEP_TIME = 300 //300 s

Functions

func NewMetric

func NewMetric(tp string) (metric.Collector, error)

Types

type CleanInfo

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

type LogExportRunner

type LogExportRunner struct {
	RunnerInfo
	// contains filtered or unexported fields
}

func NewLogExportRunner

func NewLogExportRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal, ps *parser.ParserRegistry, sr *sender.SenderRegistry) (runner *LogExportRunner, err error)

func NewLogExportRunnerWithService

func NewLogExportRunnerWithService(info RunnerInfo, reader reader.Reader, cleaner *cleaner.Cleaner, parser parser.LogParser, senders []sender.Sender, meta *reader.Meta) (runner *LogExportRunner, err error)

func (*LogExportRunner) Cleaner

func (r *LogExportRunner) Cleaner() CleanInfo

func (*LogExportRunner) LagStats

func (r *LogExportRunner) LagStats() (rl RunnerLag, err error)

func (*LogExportRunner) Name

func (r *LogExportRunner) Name() string

func (*LogExportRunner) Run

func (r *LogExportRunner) Run()

func (*LogExportRunner) Status

func (r *LogExportRunner) Status() RunnerStatus

func (*LogExportRunner) Stop

func (r *LogExportRunner) Stop()

type Manager

type Manager struct {
	ManagerConfig
	// contains filtered or unexported fields
}

func NewCustomManager

func NewCustomManager(conf ManagerConfig, pr *parser.ParserRegistry, sr *sender.SenderRegistry) (*Manager, error)

func NewManager

func NewManager(conf ManagerConfig) (*Manager, error)

func (*Manager) Add

func (m *Manager) Add(confPath string)

func (*Manager) Remove

func (m *Manager) Remove(confPath string)

func (*Manager) Status

func (m *Manager) Status() (rss map[string]RunnerStatus)

func (*Manager) Stop

func (m *Manager) Stop() error

func (*Manager) Watch

func (m *Manager) Watch(confsPath []string) (err error)

type ManagerConfig

type ManagerConfig struct {
	BindHost string `json:"bind_host"`
	Idc      string `json:"idc"`
	Zone     string `json:"zone"`
}

type MetricRunner

type MetricRunner struct {
	RunnerName string `json:"name"`
	// contains filtered or unexported fields
}

func NewMetricRunner

func NewMetricRunner(rc RunnerConfig, sr *sender.SenderRegistry) (runner *MetricRunner, err error)

func (*MetricRunner) Cleaner

func (_ *MetricRunner) Cleaner() CleanInfo

func (*MetricRunner) Name

func (mr *MetricRunner) Name() string

func (*MetricRunner) Run

func (r *MetricRunner) Run()

func (*MetricRunner) Status

func (mr *MetricRunner) Status() RunnerStatus

func (*MetricRunner) Stop

func (mr *MetricRunner) Stop()

type RestService

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

func NewRestService

func NewRestService(mgr *Manager) *RestService

func (*RestService) GetStatus

func (rs *RestService) GetStatus(rw http.ResponseWriter, req *http.Request)

get /logkit/status

func (*RestService) Stop

func (rs *RestService) Stop()

Stop will stop RestService

type Runner

type Runner interface {
	Name() string
	Run()
	Stop()
	Cleaner() CleanInfo
	Status() RunnerStatus
}

func NewCustomRunner

func NewCustomRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal, ps *parser.ParserRegistry, sr *sender.SenderRegistry) (runner Runner, err error)

func NewRunner

func NewRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal) (runner Runner, err error)

NewRunner 创建Runner

func NewRunnerWithService

func NewRunnerWithService(info RunnerInfo, reader reader.Reader, cleaner *cleaner.Cleaner, parser parser.LogParser, senders []sender.Sender, meta *reader.Meta) (runner Runner, err error)

type RunnerConfig

type RunnerConfig struct {
	RunnerInfo
	Metric        []conf.MapConf `json:"metric"`
	ReaderConfig  conf.MapConf   `json:"reader"`
	CleanerConfig conf.MapConf   `json:"cleaner"`
	ParserConf    conf.MapConf   `json:"parser"`
	SenderConfig  []conf.MapConf `json:"senders"`
}

RunnerConfig 从多数据源读取，经过解析后，发往多个数据目的地

func Compatible

func Compatible(rc RunnerConfig) RunnerConfig

Compatible 用于新老配置的兼容

type RunnerInfo

type RunnerInfo struct {
	RunnerName       string `json:"name"`
	CollectInterval  string `json:"collect_interval"` // metric runner收集的频率
	MaxBatchLen      int    `json:"batch_len"`        // 每个read batch的行数
	MaxBatchSize     int    `json:"batch_size"`       // 每个read batch的字节数
	MaxBatchInteval  int    `json:"batch_interval"`   // 最大发送时间间隔
	MaxBatchTryTimes int    `json:"batch_try_times"`  // 最大发送次数，小于等于0代表无限重试
}

type RunnerLag

type RunnerLag struct {
	Size   int64 `json:"size"`
	Files  int64 `json:"files"`
	Ftlags int64 `json:"ftlags"`
}

type RunnerStatus

type RunnerStatus struct {
	Name        string                     `json:"name"`
	Logpath     string                     `json:"logpath"`
	Lag         RunnerLag                  `json:"lag,omitempty"`
	ParserStats utils.StatsInfo            `json:"parserStats,omitempty"`
	SenderStats map[string]utils.StatsInfo `json:"senderStats,omitempty"`
	Error       error                      `json:"error,omitempty"`
}