Make job scheduler local to task process #674
Conversation
| LOG = log.getLogger(__name__) | ||
|
|
||
|
|
||
| class FailedTelemetryJob(object): |
There was a problem hiding this comment.
IMO, TelemetryFailedJob is better than FailedTelemetryJob
There was a problem hiding this comment.
done. made it TaskDistributor
|
I think we should remove the word |
Codecov Report
@@ Coverage Diff @@
## perf_coll_fw_enhance #674 +/- ##
========================================================
+ Coverage 70.15% 70.18% +0.02%
========================================================
Files 156 159 +3
Lines 14801 14936 +135
Branches 1822 1822
========================================================
+ Hits 10384 10483 +99
- Misses 3816 3846 +30
- Partials 601 607 +6
|
| '%s' % job['id']) | ||
| self.task_rpcapi.assign_failed_job(self.ctx, job) | ||
|
|
||
| LOG.debug('Assigned failed task for id: ' |
There was a problem hiding this comment.
I think line 58 can be debug and 62 can be info
There was a problem hiding this comment.
done, this code might change according distributor implementation, currently it is a pool based distributor
| return call_context.cast(context, 'remove_job', | ||
| job=job) | ||
|
|
||
| def assign_failed_job(self, context, job): |
There was a problem hiding this comment.
Can we have just 2 apis assign and remove with job names as they differ only by job names
There was a problem hiding this comment.
Actually the handlers for both messages are different , if we make it same we need to pass on more argument to switch between types. since it is distribution over n/w we want to reduce message size, currently we use only task_id as parameter
| 'last_run_time': last_run_time} | ||
| db.task_update(self.ctx, self.task_id, update_task_dict) | ||
| LOG.info('Periodic collection tasks scheduled for for job id: ' | ||
| '%s ' % self.task_id) |
There was a problem hiding this comment.
self.job_ids.add(job_id) need here right?
NajmudheenCT
changed the title
|
leader_election/distributor need a init to make it as a module |
its there ! |
I try to find that, but failed. Only test folder had |
You are right.. missed in this location.. added now |
|
|
||
| def __call__(self): | ||
| """ | ||
| :return: |
There was a problem hiding this comment.
Please remove the useless comments
| def schedule_job(self, task_id): | ||
|
|
||
| if self.stopped: | ||
| """If Job is stopped return immediately""" |
There was a problem hiding this comment.
Single line comment please use #
| # limitations under the License. | ||
| """ | ||
|
|
||
| **periodical task manager for metric collection tasks** |
There was a problem hiding this comment.
Comment style should be same with other files, such as metrics_rpcapi.py
|
|
||
| from delfin import manager | ||
| from delfin.task_manager.scheduler import schedule_manager | ||
|
|
|
LGTM |
* Make job scheduler local to task process (#674) * Make job scheduler local to task process * Notify distributor when a new task added (#678) * Remove db-scan for new task creation (#680) * Use consistent hash to manage the topic (#681) * Remove the periodically call from task distributor (#686) * Start one historic collection immediate when a job is rescheduled (#685) * Start one historic collection immediate when a job is rescheduled * Remove failed task distributor (#687) * Improving Failed job handling and telemetry job removal (#689) Co-authored-by: ThisIsClark <liuyuchibubao@gmail.com> Co-authored-by: Ashit Kumar <akopensrc@gmail.com>
What this PR does / why we need it:
Enhance/Change Performance metrics collection framework to schedule jobs locally in all worker nodes.
Which issue this PR fixes (optional, in
fixes #<issue number>(, fixes #<issue_number>, ...)format, will close that issue when PR gets merged): fixes #670Special notes for your reviewer:
Release note: