test worker failure

dlrover/examples/deepctr_auto_scale_job.yaml

@@ -64,4 +64,5 @@ spec:
  restartPolicy: Never
  containers:
  - name: main
+ image: registry.cn-hangzhou.aliyuncs.com/intell-ai/dlrover:test 
  imagePullPolicy: Always

dlrover/examples/deepctr_manual_scale_job.yaml

@@ -11,7 +11,7 @@ spec:
  replicaSpecs:
  ps:
  autoScale: False
- replicas: 3
+ replicas: 2
  template:
  spec:
  restartPolicy: Never
@@ -30,8 +30,7 @@ spec:
  command:
  - /bin/bash
  - -c
- - "pip install pyhocon \
- && cd /home/model_zoo/tf_estimator/criteo_deeprec \
+ - "cd /home/model_zoo/tf_estimator/criteo_deeprec \
  && python -m dlrover.trainer.entry.local_entry \
  --platform=Kubernetes --conf=train_conf.TrainConf \
  --enable_auto_scaling=True"
@@ -44,7 +43,7 @@ spec:
  claimName: pvc-nas
  worker:
  autoScale: False
- replicas: 3
+ replicas: 2
  template:
  spec:
  restartPolicy: Never
@@ -63,8 +62,7 @@ spec:
  command:
  - /bin/bash
  - -c
- - "pip install pyhocon \
- && cd /home/model_zoo/tf_estimator/criteo_deeprec \
+ - "cd /home/model_zoo/tf_estimator/criteo_deeprec \
  && python -m dlrover.trainer.entry.local_entry \
  --platform=Kubernetes --conf=train_conf.TrainConf \
  --enable_auto_scaling=True"
@@ -81,6 +79,5 @@ spec:
  restartPolicy: Never
  containers:
  - name: main
- imagePullPolicy: Always
- # yamllint disable-line rule:line-length
- image: registry.cn-hangzhou.aliyuncs.com/dlrover_deeprec/deeprec:v11
+ image: registry.cn-hangzhou.aliyuncs.com/intell-ai/dlrover:test 
+ imagePullPolicy: Always

dlrover/examples/scale_plan.yaml

@@ -1,12 +1,12 @@
 apiVersion: elastic.iml.github.io/v1alpha1
 kind: ScalePlan
 metadata:
- name: deepctr-auto-scaling-job-i22
+ name: deepctr-auto-scaling-job-01
  labels:
  elasticjob-name: deepctr-auto-scale
  scale-type: manual
 spec:
- ownerJob: deepctr-auto-scaling-job
+ ownerJob: deepctr-auto-scale
  replicaResourceSpecs:
- worker:
- replicas: 4
+ ps:
+ replicas: 3

dlrover/trainer/constants/tf_constants.py

@@ -73,7 +73,6 @@ class TFConstants(object):
  KeepCheckpointMax = Constant("keep_checkpoint_max", 5)
  DataShardClient = Constant("data_shard_client", None)
  ExitRecoverableSession = Constant("exit_recoverable_session", None)
- DataShardCheckpoint = Constant("data_shard_checkpoint", "data_shard_checkpoint.json")
-
-
-
+ DataShardCheckpoint = Constant(
+ "data_shard_checkpoint", "data_shard_checkpoint.json"
+ )

dlrover/trainer/tensorflow/executor/estimator_executor.py

@@ -209,6 +209,7 @@ def _prepare_estimator_config_and_params(self):
  def _prepare_train_dataset(self):
  """prepare_train_dataset"""
  train_set = self._task_conf.get(TFConstants.TrainSet.name)
+ logger.info("Prepare training dataset with {}".format(train_set))
  self.train_dataset = DatasetUtil.create(train_set)
 
  def _prepare_eval_dataset(self):

dlrover/trainer/tensorflow/failover/tensorflow_failover.py

@@ -74,6 +74,7 @@ def _start_failover_monitor(self):
  def monitor_fun():
  logger.info("Successfully to start failover monitor!")
  while True:
+ logger.info("querying master for ps cluster info")
  ps_address_changed, change_type = self.ps_addresses_changed()
  if ps_address_changed:
  self.refresh_env()

dlrover/trainer/tensorflow/reader/file_reader.py

@@ -17,20 +17,23 @@
 
 class FileReader(ElasticReader):
  def __init__(self, path=None, skip_header=True):
+ print("FileReader is initiating path is {}".format(path))
  self._skip_header = skip_header
  self._file_handler = open(path, "r")
+ self.data = self._file_handler.readlines()
  self._file_name = path
  super().__init__(
  path=path,
  )
+ self._data_nums = None
 
  def count_data(self):
- self.data = self._file_handler.readlines()
- if self._skip_header:
- self._data_nums = len(self.data) - 1
- self.data = self.data[1:]
- else:
- self._data_nums = len(self.data)
+ if self._data_nums is None:
+  if self._skip_header:
+  self._data_nums = len(self.data) - 1
+  self.data = self.data[1:]
+  else:
+  self._data_nums = len(self.data)
 
  def read_data_by_index_range(self, start_index, end_index):
  for i in range(start_index, end_index):