!399 delete lineage cache when parse failed

Merge pull request !399 from luopengting/fix_lineage_update

mindinsight/datavisual/data_transform/data_manager.py

@@ -149,10 +149,13 @@ class CachedTrainJob:
         """Set value to cache."""
         self._content[key] = value
 
-    def delete(self, key):
+    def delete(self, key, raise_exception=True):
         """Delete key in cache."""
-        if key in self._content:
+        try:
             self._content.pop(key)
+        except KeyError:
+            if raise_exception:
+                raise ParamValueError("Delete failed. Invalid cache key({}).".format(key))
 
     def get(self, key, raise_exception=True):
         """

mindinsight/lineagemgr/cache_item_updater.py

@@ -17,7 +17,7 @@ import os
 
 from mindinsight.datavisual.data_transform.data_manager import BaseCacheItemUpdater, CachedTrainJob
 from mindinsight.lineagemgr.common.log import logger
-from mindinsight.lineagemgr.common.exceptions.exceptions import LineageFileNotFoundError
+from mindinsight.lineagemgr.common.exceptions.exceptions import MindInsightException
 from mindinsight.lineagemgr.common.validator.validate import validate_train_id, validate_added_info
 from mindinsight.lineagemgr.lineage_parser import LineageParser, LINEAGE
 from mindinsight.utils.exceptions import ParamValueError
@@ -59,14 +59,19 @@ class LineageCacheItemUpdater(BaseCacheItemUpdater):
 
         try:
             lineage_parser = self._lineage_parsing(cache_item)
-        except LineageFileNotFoundError:
+        except MindInsightException as err:
             with cache_item.lock_key(LINEAGE):
-                cache_item.delete(key=LINEAGE)
+                try:
+                    cache_item.delete(key=LINEAGE)
+                    logger.info("Parse failed, delete the tran job %s. Detail: %s.", relative_path, str(err))
+                except ParamValueError:
+                    logger.debug("Parse failed, no need to delete the train job %s. "
+                                 "Detail: %s.", relative_path, str(err))
             return
 
         super_lineage_obj = lineage_parser.super_lineage_obj
         if super_lineage_obj is None:
-            logger.info("There is no lineage to update in tran job %s.", relative_path)
+            logger.debug("There is no lineage to update in tran job %s.", relative_path)
             return
 
         cache_item.set(key=LINEAGE, value=lineage_parser)

mindinsight/lineagemgr/lineage_parser.py

@@ -123,17 +123,9 @@ class LineageParser:
             bool, `True` if parse summary log success, else `False`.
         """
         file_path = os.path.realpath(os.path.join(self._summary_dir, self._latest_filename))
-        try:
-            lineage_info = LineageSummaryAnalyzer.get_summary_infos(file_path)
-            user_defined_info = LineageSummaryAnalyzer.get_user_defined_info(file_path)
-            self._update_lineage_obj(lineage_info, user_defined_info)
-        except LineageSummaryAnalyzeException:
-            logger.warning("Parse file failed under summary_dir %s.", file_path)
-        except (LineageEventNotExistException, LineageEventFieldNotExistException) as error:
-            logger.warning("Parse file failed under summary_dir %s. Detail: %s.", file_path, str(error))
-        except MindInsightException as error:
-            logger.exception(error)
-            logger.warning("Parse file failed under summary_dir %s.", file_path)
+        lineage_info = LineageSummaryAnalyzer.get_summary_infos(file_path)
+        user_defined_info = LineageSummaryAnalyzer.get_user_defined_info(file_path)
+        self._update_lineage_obj(lineage_info, user_defined_info)
 
     def _update_lineage_obj(self, lineage_info, user_defined_info):
         """Update lineage object."""
@@ -197,6 +189,13 @@ class LineageOrganizer:
                     self._super_lineage_objs.update({abs_summary_dir: super_lineage_obj})
             except LineageFileNotFoundError:
                 no_lineage_count += 1
+            except (LineageSummaryAnalyzeException,
+                    LineageEventNotExistException,
+                    LineageEventFieldNotExistException) as error:
+                logger.warning("Parse file failed under summary_dir %s. Detail: %s.", relative_dir, str(error))
+            except MindInsightException as error:
+                logger.exception(error)
+                logger.warning("Parse file failed under summary_dir %s.", relative_dir)
 
         if no_lineage_count == len(relative_dirs):
             logger.info('There is no summary log file under summary_base_dir.')
@@ -215,7 +214,7 @@ class LineageOrganizer:
                 super_lineage_obj = cache_train_job.get("lineage").super_lineage_obj
                 self._super_lineage_objs.update({relative_dir: super_lineage_obj})
             except ParamValueError:
-                logger.info("This is no lineage info in train job %s.", relative_dir)
+                logger.debug("This is no lineage info in train job %s.", relative_dir)
 
     @property
     def super_lineage_objs(self):