m26508134
/
docklet
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

update

This commit is contained in:
Gallen 2018-08-08 21:51:27 +08:00
parent c7ae571bae
commit 0a4507b3a5
2 changed files with 25 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
src/master/taskmgr.py
View File

@ -5,9 +5,9 @@ import random
import json import json
# must import logger after initlogging, ugly # must import logger after initlogging, ugly
from utils.log import initlogging # from utils.log import initlogging
initlogging("docklet-taskmgr") # initlogging("docklet-taskmgr")
from utils.log import logger # from utils.log import logger
# grpc # grpc
from concurrent import futures from concurrent import futures
@ -49,10 +49,7 @@ class TaskMgr(threading.Thread):
self.task_queue = [] self.task_queue = []
self.scheduler_interval = scheduler_interval self.scheduler_interval = scheduler_interval
if external_logger is None: self.logger = external_logger
self.logger = logger
else:
self.logger = external_logger
self.master_port = env.getenv('BATCH_MASTER_PORT') self.master_port = env.getenv('BATCH_MASTER_PORT')
self.worker_port = env.getenv('BATCH_WORKER_PORT') self.worker_port = env.getenv('BATCH_WORKER_PORT')
@ -209,7 +206,7 @@ class TaskMgr(threading.Thread):
return task, index, worker return task, index, worker
# find timeout instance # find timeout instance
elif instance['status'] == RUNNING: elif instance['status'] == RUNNING:
if not is_alive(instance['worker']): if not self.is_alive(instance['worker']):
instance['status'] = FAILED instance['status'] = FAILED
instance['token'] = '' instance['token'] = ''
self.cpu_usage[instance['worker']] -= task.info.cluster.instance.cpu self.cpu_usage[instance['worker']] -= task.info.cluster.instance.cpu
@ -297,28 +294,29 @@ class TaskMgr(threading.Thread):
task = Task(TaskInfo( task = Task(TaskInfo(
id = taskid, id = taskid,
username = username, username = username,
instanceCount = json_task['instanceCount'], instanceCount = int(json_task['instCount']),
maxRetryCount = json_task['maxRetryCount'], maxRetryCount = int(json_task['retryCount']),
timeout = json_task['timeout'], timeout = int(json_task['expTime']),
parameters = Parameters( parameters = Parameters(
command = Command( command = Command(
commandLine = json_task['parameters']['command']['commandLine'], commandLine = json_task['command'],
packagePath = json_task['parameters']['command']['packagePath'], packagePath = json_task['srcAddr'],
envVars = json_task['parameters']['command']['envVars']), envVars = {}),
stderrRedirectPath = json_task['parameters']['stderrRedirectPath'], stderrRedirectPath = json_task['stdErrRedPth'],
stdoutRedirectPath = json_task['parameters']['stdoutRedirectPath']), stdoutRedirectPath = json_task['stdOutRedPth']),
cluster = Cluster( cluster = Cluster(
image = Image( image = Image(
name = json_task['cluster']['image']['name'], name = 'base', #json_task['cluster']['image']['name'],
type = json_task['cluster']['image']['type'], type = Image.BASE, #json_task['cluster']['image']['type'],
owner = json_task['cluster']['image']['owner']), owner = 'base'), #json_task['cluster']['image']['owner']),
instance = Instance( instance = Instance(
cpu = json_task['cluster']['instance']['cpu'], cpu = int(json_task['cpuSetting']),
memory = json_task['cluster']['instance']['memory'], memory = int(json_task['memorySetting']),
disk = json_task['cluster']['instance']['disk'], disk = int(json_task['diskSetting']),
gpu = json_task['cluster']['instance']['gpu'])))) gpu = int(json_task['gpuSetting'])))))
task.info.cluster.mount.extend([Mount(localPath=mount['localPath'], remotePath=mount['remotePath']) task.info.cluster.mount.extend([Mount(localPath=json_task['mapping'][mapping_key]['mappingLocalDir'],
for mount in json_task['cluster']['mount']]) remotePath=json_task['mapping'][mapping_key]['mappingRemotefDir'])
for mapping_key in json_task['mapping']])
self.task_queue.append(task) self.task_queue.append(task)

4
src/master/testTaskMgr.py
View File

@ -94,7 +94,7 @@ class SimulatedJobMgr(threading.Thread):
task['timeout'] = timeout task['timeout'] = timeout
task['parameters'] = {} task['parameters'] = {}
task['parameters']['command'] = {} task['parameters']['command'] = {}
task['parameters']['command']['commandLine'] = '' task['parameters']['command']['commandLine'] = 'ls'
task['parameters']['command']['packagePath'] = '' task['parameters']['command']['packagePath'] = ''
task['parameters']['command']['envVars'] = {'a':'1'} task['parameters']['command']['envVars'] = {'a':'1'}
task['parameters']['stderrRedirectPath'] = '' task['parameters']['stderrRedirectPath'] = ''
@ -111,7 +111,7 @@ class SimulatedJobMgr(threading.Thread):
task['cluster']['instance']['gpu'] = 0 task['cluster']['instance']['gpu'] = 0
task['cluster']['mount'] = [{'remotePath':'', 'localPath':''}] task['cluster']['mount'] = [{'remotePath':'', 'localPath':''}]
taskmgr.add_task('user', taskid, json.dumps(task)) taskmgr.add_task('root', taskid, json.dumps(task))
class SimulatedLogger(): class SimulatedLogger():