regularly update the task status value by the memory usage
(store *StoreManager, shutdownCh chan struct{}, jobId string)
| 160 | |
| 161 | // regularly update the task status value by the memory usage |
| 162 | func RegularlyUpdateJobStatus(store *StoreManager, shutdownCh chan struct{}, jobId string) { |
| 163 | ticker := time.NewTicker(time.Second * 3) |
| 164 | defer ticker.Stop() |
| 165 | for { |
| 166 | select { |
| 167 | case <-shutdownCh: |
| 168 | return |
| 169 | case <-ticker.C: |
| 170 | lowMemoryStatus := g.GetLowMemoryStatus() |
| 171 | jobInfo, err := store.GetJobInfo(jobId) |
| 172 | if err != nil { |
| 173 | store.logger.Error("get job info err", "jobId", jobId, "err", err) |
| 174 | continue |
| 175 | } |
| 176 | if jobInfo.JobStatus == DtleJobStatusNonPaused && lowMemoryStatus { |
| 177 | jobInfo.JobStatus = DtleJobStatusStop |
| 178 | } else if jobInfo.JobStatus == DtleJobStatusStop && !lowMemoryStatus { |
| 179 | jobInfo.JobStatus = DtleJobStatusNonPaused |
| 180 | } else { |
| 181 | continue |
| 182 | } |
| 183 | store.logger.Info("update job status", "jobId", jobId, "jobStatus", jobInfo.JobStatus) |
| 184 | if err = store.SaveJobInfo(*jobInfo); err != nil { |
| 185 | store.logger.Error("get job info err", "jobId", jobId, "err", err) |
| 186 | continue |
| 187 | } |
| 188 | } |
| 189 | } |
| 190 | } |
no test coverage detected