(d *Driver)
| 125 | } |
| 126 | |
| 127 | func (h *taskHandle) run(d *Driver) { |
| 128 | h.stateLock.Lock() |
| 129 | if h.exitResult == nil { |
| 130 | h.exitResult = &drivers.ExitResult{} |
| 131 | } |
| 132 | h.procState = drivers.TaskStateRunning |
| 133 | h.stateLock.Unlock() |
| 134 | |
| 135 | consulJobInfo, err := d.storeManager.GetJobInfo(h.taskConfig.JobName) |
| 136 | if nil != err { |
| 137 | h.onError(errors.Wrap(err, "get pause status from consul failed")) |
| 138 | return |
| 139 | } |
| 140 | |
| 141 | if consulJobInfo.JobStatus == common.DtleJobStatusPaused { |
| 142 | h.logger.Info("job is paused. not starting the runner") |
| 143 | } else { |
| 144 | err = h.resumeTask(d) |
| 145 | if err != nil { |
| 146 | h.onError(err) |
| 147 | return |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | go func() { |
| 152 | duration := time.Duration(d.config.StatsCollectionInterval) * time.Second |
| 153 | t := time.NewTimer(0) |
| 154 | for { |
| 155 | select { |
| 156 | case <-h.ctx.Done(): |
| 157 | if !t.Stop() { |
| 158 | <-t.C |
| 159 | } |
| 160 | return |
| 161 | case <-t.C: |
| 162 | if h.runner != nil { |
| 163 | s, err := h.runner.Stats() |
| 164 | if err != nil { |
| 165 | // ignore |
| 166 | } else { |
| 167 | h.stats = s |
| 168 | if d.config.PublishMetrics { |
| 169 | h.logger.Trace("emitStats") |
| 170 | h.emitStats(s) |
| 171 | } |
| 172 | } |
| 173 | } |
| 174 | t.Reset(duration) |
| 175 | } |
| 176 | } |
| 177 | }() |
| 178 | } |
| 179 | |
| 180 | func (h *taskHandle) NewRunner(d *Driver) (runner DriverHandle, err error) { |
| 181 | ctx := &common.ExecContext{ |
no test coverage detected