migration: saving to file migration error

Added --migration-error-path option, this in case of exception will save the error to file

Author: danielblasina-tech

README.md

@@ -70,6 +70,8 @@ optional arguments:
                         Max total size of databases to be migrated, ignored by default
   --output-meta-file OUTPUT_META_FILE
                         Output file which includes metadata such as dump GTIDs (for replication method only) in JSON format.
+  --output_error_file OUTPUT_ERROR_FILE
+                        Save migration error to file in JSON format.
   --allow-source-without-dbs
                         Allow migrating from a source that has no migratable databases
 ```

aiven_mysql_migrate/main.py

@@ -62,6 +62,13 @@ def main(args: Sequence[str] | None = None, *, app: str = "mysql_migrate") -> Op
         default=None,
         help="Output file which includes metadata such as dump GTIDs (for replication method only) in JSON format.",
     )
+    parser.add_argument(
+        "--output-error-file",
+        type=Path,
+        required=False,
+        default=None,
+        help="Save migration error to file in JSON format.",
+    )
     parser.add_argument(
         "--allow-source-without-dbs",
         action="store_true",
@@ -89,6 +96,7 @@ def main(args: Sequence[str] | None = None, *, app: str = "mysql_migrate") -> Op
         privilege_check_user=parsed_args.privilege_check_user,
         output_meta_file=parsed_args.output_meta_file,
         dump_tool=parsed_args.dump_tool,
+        output_error_file=parsed_args.output_error_file,
     )
     migration.setup_signal_handlers()
 

aiven_mysql_migrate/migration.py

@@ -1,4 +1,6 @@
 # Copyright (c) 2020 Aiven, Helsinki, Finland. https://aiven.io/
+import datetime
+
 from aiven_mysql_migrate import config
 from aiven_mysql_migrate.dump_tools import MySQLMigrationToolBase, get_dump_tool
 from aiven_mysql_migrate.enums import MySQLMigrateTool, MySQLMigrateMethod
@@ -8,6 +10,7 @@
     SSLNotSupportedException, TooManyDatabasesException, UnsupportedBinLogFormatException, UnsupportedMySQLEngineException,
     UnsupportedMySQLVersionException, WrongMigrationConfigurationException
 )
+from aiven_mysql_migrate.migration_error import MysqlMigrationError
 from aiven_mysql_migrate.utils import MySQLConnectionInfo, PrivilegeCheckUser, select_global_var
 from looseversion import LooseVersion
 from pathlib import Path
@@ -41,6 +44,7 @@ def __init__(
         privilege_check_user: Optional[str] = None,
         output_meta_file: Optional[Path] = None,
         dump_tool: MySQLMigrateTool = MySQLMigrateTool.mysqldump,
+        output_error_file: Optional[Path] = None,
     ):
         self.dump_tool_name = dump_tool
         self.dump_tool: Optional[MySQLMigrationToolBase] = None
@@ -61,6 +65,7 @@ def __init__(
         if privilege_check_user:
             self.privilege_check_user = PrivilegeCheckUser.parse(privilege_check_user)
         self.output_meta_file = output_meta_file
+        self.output_error_file = output_error_file
 
     def setup_signal_handlers(self):
         signal.signal(signal.SIGINT, self._stop_migration)
@@ -380,6 +385,24 @@ def _wait_for_replication(self, *, seconds_behind_master: int = 0, check_interva
 
     def start(self, *,
               migration_method: MySQLMigrateMethod, seconds_behind_master: int, stop_replication: bool = False) -> None:
+        try:
+            self.start_migration(migration_method=migration_method,
+                                 seconds_behind_master=seconds_behind_master,
+                                 stop_replication=stop_replication)
+        except Exception as e:
+            if self.output_error_file is not None:
+                with open(self.output_error_file, "w", encoding='utf-8') as f:
+                    error = MysqlMigrationError(error_type=e.__class__.__module__ + "." + e.__class__.__name__,
+                                                error_msg=str(e),
+                                                error_date=datetime.datetime.now(datetime.timezone.utc))
+                    f.write(json.dumps(error.__dict__, default=str))
+            raise
+
+    def start_migration(self,
+                        *,
+                        migration_method: MySQLMigrateMethod,
+                        seconds_behind_master: int,
+                        stop_replication: bool = False) -> None:
         LOGGER.info("Start migration of the following databases:")
         for db in self.databases:
             LOGGER.info("\t%s", db)

aiven_mysql_migrate/migration_error.py

@@ -0,0 +1,15 @@
+import datetime
+
+
+class MysqlMigrationError:
+    error_type: str
+    error_msg: str
+    error_date: datetime.datetime
+
+    def __init__(self, error_type: str, error_msg: str, error_date: str | datetime.datetime):
+        self.error_type = error_type
+        self.error_msg = error_msg
+        if isinstance(error_date, datetime.datetime):
+            self.error_date = error_date
+        else:
+            self.error_date = datetime.datetime.strptime(error_date, "%Y-%m-%d %H:%M:%S.%f%z")

test/unit/test_utils.py

@@ -1,5 +1,14 @@
 # Copyright (c) 2025 Aiven, Helsinki, Finland. https://aiven.io/
+import datetime
+import json
+
+from pymysql import OperationalError
+from aiven_mysql_migrate.enums import MySQLMigrateMethod
+
+from aiven_mysql_migrate.migration import MySQLMigration
+
 from aiven_mysql_migrate.exceptions import WrongMigrationConfigurationException
+from aiven_mysql_migrate.migration_error import MysqlMigrationError
 from aiven_mysql_migrate.utils import MySQLConnectionInfo, MySQLDumpProcessor, MydumperDumpProcessor
 from pathlib import Path
 from pytest import mark, raises
@@ -292,3 +301,30 @@ def test_mydumper_dump_processor_ignores_database_files_from_metadata_database()
         result4 = processor.process_line("-- metadata 0")
         assert result4 == "-- metadata 0"
         assert processor.gtid == _GTID
+
+
+def test_failed_migration_generates_file():
+    migration_error_path = Path("/", "tmp", "error.json")
+    server_uri = "mysql://user:password@invalid:3306"
+    migration = MySQLMigration(
+        source_uri=server_uri,
+        target_uri=server_uri,
+        target_master_uri=server_uri,
+        privilege_check_user="root@%",
+        output_error_file=migration_error_path
+    )
+    with raises(OperationalError):
+        migration.start(migration_method=MySQLMigrateMethod.dump, seconds_behind_master=0)
+
+    with open(migration_error_path, encoding='utf-8') as file:
+        error: MysqlMigrationError = json.loads(file.read(), object_hook=lambda d: MysqlMigrationError(**d))
+        assert error.error_type == "pymysql.err.OperationalError"
+        assert error.error_msg == ("(2003, "
+                                   "\"Can't connect to MySQL server on 'invalid' "
+                                   "([Errno -2] Name or service not known)\")")
+        assert error.error_date < datetime.datetime.now(datetime.timezone.utc)
+
+
+def test_migration_error_fails_on_invalid_date():
+    with raises(ValueError):
+        MysqlMigrationError(error_type="pymysql.err.OperationalError", error_msg="message", error_date="invalid")