package worker import ( "os" "sync" "sync/atomic" "time" ) // baseProvider is the base mixin of providers type baseProvider struct { sync.Mutex ctx *Context name string interval time.Duration retry int timeout time.Duration isMaster bool cmd *cmdJob logFileFd *os.File isRunning atomic.Value cgroup *cgroupHook zfs *zfsHook docker *dockerHook hooks []jobHook } func (p *baseProvider) Name() string { return p.name } func (p *baseProvider) EnterContext() *Context { p.ctx = p.ctx.Enter() return p.ctx } func (p *baseProvider) ExitContext() *Context { p.ctx, _ = p.ctx.Exit() return p.ctx } func (p *baseProvider) Context() *Context { return p.ctx } func (p *baseProvider) Interval() time.Duration { // logger.Debug("interval for %s: %v", p.Name(), p.interval) return p.interval } func (p *baseProvider) Retry() int { return p.retry } func (p *baseProvider) Timeout() time.Duration { return p.timeout } func (p *baseProvider) IsMaster() bool { return p.isMaster } func (p *baseProvider) WorkingDir() string { if v, ok := p.ctx.Get(_WorkingDirKey); ok { if s, ok := v.(string); ok { return s } } panic("working dir is impossible to be non-exist") } func (p *baseProvider) LogDir() string { if v, ok := p.ctx.Get(_LogDirKey); ok { if s, ok := v.(string); ok { return s } } panic("log dir is impossible to be unavailable") } func (p *baseProvider) LogFile() string { if v, ok := p.ctx.Get(_LogFileKey); ok { if s, ok := v.(string); ok { return s } } panic("log file is impossible to be unavailable") } func (p *baseProvider) AddHook(hook jobHook) { switch v := hook.(type) { case *cgroupHook: p.cgroup = v case *zfsHook: p.zfs = v case *dockerHook: p.docker = v } p.hooks = append(p.hooks, hook) } func (p *baseProvider) Hooks() []jobHook { return p.hooks } func (p *baseProvider) Cgroup() *cgroupHook { return p.cgroup } func (p *baseProvider) ZFS() *zfsHook { return p.zfs } func (p *baseProvider) Docker() *dockerHook { return p.docker } func (p *baseProvider) prepareLogFile(append bool) error { if p.LogFile() == "/dev/null" { p.cmd.SetLogFile(nil) return nil } appendMode := 0 if append { appendMode = os.O_APPEND } logFile, err := os.OpenFile(p.LogFile(), os.O_WRONLY|os.O_CREATE|appendMode, 0644) if err != nil { logger.Errorf("Error opening logfile %s: %s", p.LogFile(), err.Error()) return err } p.logFileFd = logFile p.cmd.SetLogFile(logFile) return nil } func (p *baseProvider) closeLogFile() (err error) { if p.logFileFd != nil { err = p.logFileFd.Close() p.logFileFd = nil } return } func (p *baseProvider) Run(started chan empty) error { panic("Not Implemented") } func (p *baseProvider) Start() error { panic("Not Implemented") } func (p *baseProvider) IsRunning() bool { isRunning, _ := p.isRunning.Load().(bool) return isRunning } func (p *baseProvider) Wait() error { defer func() { logger.Debugf("set isRunning to false: %s", p.Name()) p.isRunning.Store(false) }() logger.Debugf("calling Wait: %s", p.Name()) return p.cmd.Wait() } func (p *baseProvider) Terminate() error { p.Lock() defer p.Unlock() logger.Debugf("terminating provider: %s", p.Name()) if !p.IsRunning() { logger.Warningf("Terminate() called while IsRunning is false: %s", p.Name()) return nil } err := p.cmd.Terminate() return err } func (p *baseProvider) DataSize() string { return "" }