adjust pipeline

Signed-off-by: mgqa34 <mgq3374541@163.com>

python/fate_client/fate_client/pipeline/conf/env_config.py

@@ -18,7 +18,7 @@
 from pathlib import Path
 
 
-__all__ = ["StatusCode", "LogPath", "LogFormat", "FlowConfig", "StandaloneConfig", "SiteInfo"]
+__all__ = ["StatusCode", "FlowConfig", "StandaloneConfig", "SiteInfo"]
 
 
 with Path(__file__).parent.parent.joinpath("pipeline_config.yaml").resolve().open("r") as fin:
@@ -124,37 +124,11 @@ class StandaloneConfig(object):
     OUTPUT_DATA_DIR = job_dir.joinpath("data").as_uri()
     OUTPUT_MODEL_DIR = job_dir.joinpath("model").as_uri()
     OUTPUT_METRIC_DIR = job_dir.joinpath("metric").as_uri()
+    OUTPUT_LOG_DIR = job_dir.joinpath("logs")
 
     MLMD = MLMD(conf)
     LOGGER = LOGGER(conf)
     DEVICE = Device(conf)
     COMPUTING_ENGINE = ComputingEngine(conf)
     FEDERATION_ENGINE = FederationEngine(conf)
 
-
-class LogPath(object):
-    @classmethod
-    def log_directory(cls):
-        conf = get_default_config()
-        log_directory = conf.get("log_directory")
-        if log_directory:
-            log_directory = Path(log_directory).resolve()
-        else:
-            log_directory = Path(__file__).parent.parent.joinpath("logs")
-        try:
-            log_directory.mkdir(parents=True, exist_ok=True)
-        except Exception as e:
-            raise RuntimeError(f"can't create log directory for pipeline: {log_directory}") from e
-        if not Path(log_directory).resolve().is_dir():
-            raise NotADirectoryError(f"provided log directory {log_directory} is not a directory.")
-        return log_directory
-
-    DEBUG = 'DEBUG.log'
-    INFO = 'INFO.log'
-    ERROR = 'ERROR.log'
-
-
-class LogFormat(object):
-    SIMPLE = '<green>[{time:HH:mm:ss}]</green><level>{message}</level>'
-    NORMAL = '<green>{time:YYYY-MM-DD HH:mm:ss}</green> | ' \
-             '<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>'

python/fate_client/fate_client/pipeline/executor/task_executor.py

@@ -14,7 +14,7 @@
 #  limitations under the License.
 from pathlib import Path
 from typing import Dict
-from ..conf.env_config import LogPath
+from ..conf.env_config import StandaloneConfig
 from ..utils.standalone.id_gen import gen_job_id
 from ..utils.standalone.job_process import process_task
 from ..entity.dag_structures import DAGSchema
@@ -62,7 +62,7 @@ def predict(self,
 
     def _run(self, fit_model_info: StandaloneModelInfo = None):
         self._job_id = gen_job_id()
-        self._log_dir_prefix = Path(LogPath.log_directory()).joinpath(self._job_id)
+        self._log_dir_prefix = StandaloneConfig.OUTPUT_LOG_DIR.joinpath(self._job_id)
         print(f"log prefix {self._log_dir_prefix}")
 
         runtime_constructor_dict = dict()
@@ -73,9 +73,7 @@ def _run(self, fit_model_info: StandaloneModelInfo = None):
             stage = task_node.stage
             runtime_parties = task_node.runtime_parties
             runtime_parameters = task_node.runtime_parameters
-            component_spec = task_node.component_spec
             upstream_inputs = task_node.upstream_inputs
-            # output_definitions = task_node.output_definitions
 
             runtime_constructor = RuntimeConstructor(runtime_parties=runtime_parties,
                                                      job_id=self._job_id,