Merge pull request #18 from zeerayne/release/v2024.04.28

.flake8

@@ -18,4 +18,7 @@ per-file-ignores =
     settings.tpl.py: E266
     # module imported but unused
     surrogate/tests/test_surrogate.py: F401
+    # local variable 'succeeded' is assigned to but never used
+    conftest.py: F841
 inline-quotes = "
+pytest-fixture-no-parentheses = True

.github/workflows/flake8.yaml

@@ -1,7 +1,9 @@
 # This workflow will check modified files with flake8
 
 name: flake8
-on: [push]
+on:   
+  pull_request:
+    types: [opened, reopened, synchronize]
 jobs:
   flake8:
     runs-on: ubuntu-latest
@@ -11,7 +13,7 @@ jobs:
       - name: Install Python
         uses: actions/setup-python@v4
         with:
-          python-version: "3.10"
+          python-version: "3.12"
       - name: Install Poetry
         uses: abatilo/[email protected]
       - name: Install dependencies

.github/workflows/pytest.yaml

@@ -1,12 +1,14 @@
 # This workflow will run tests with all supported python versions
 
 name: pytest
-on: [push]
+on:
+  pull_request:
+    types: [opened, reopened, synchronize]
 jobs:
   test:
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
     steps:

README.md

@@ -78,7 +78,7 @@ git clone https://github.com/zeerayne/1cv8-mgmt-tool.git
 
 ```powershell
 cd 1cv8-mgmt-tool
-poetry install --no-root --no-dev
+poetry install --no-root --only main
 ```
 
 # Настройка
@@ -134,15 +134,16 @@ TIP: для работы с путями в шаблонном файле нас
 
 |Параметр|Описание|
 |-------:|:-------|
-|`BACKUP_CONCURRENCY`      |Параллелизм: сколько резервных копий может создаваться одновременно|
-|`BACKUP_PATH`             |Путь к каталогу, куда будут помещены файлы резервных копий|
-|`BACKUP_PG`               |Включает или отключает функцию создания резервных копий средствами PostgreSQL для совместимых информационных баз (базы данных которых размещены на СУБД PostgreSQL), принимает значения `True` или `False`|
-|`BACKUP_RETENTION_DAYS`   |Копии старше, чем количество дней в этой настройке будут удалсяться из каталога резервных копий при работе резервного копирования|
-|`BACKUP_REPLICATION`      |Включает или отключает функцию копирования резервных копий в дополнительные локации (например на сетевой диск), принимает значения `True` или `False`|
-|`BACKUP_REPLICATION_PATHS`|Список путей, куда резервные копии будут реплицированы|
-|`BACKUP_RETRIES_V8`       |Количество повторных попыток создания резервной копии средствами 1С Предприятие в случае возникновении ошибки. Если установлено значение 0, повторные попытки предприниматься не будут|
-|`BACKUP_RETRIES_PG`       |Количество повторных попыток загрузки резервной копии средствами PostgreSQL (см. секцию PostgreSQL). Если установлено значение 0, повторные попытки предприниматься не будут|
-|`BACKUP_TIMEOUT_V8`       |Таймаут в секундах, по истечению которого резервное копирование информационной базы считается неуспешным и принудительно завершается|
+|`BACKUP_CONCURRENCY`            |Параллелизм: сколько резервных копий может создаваться одновременно|
+|`BACKUP_PATH`                   |Путь к каталогу, куда будут помещены файлы резервных копий|
+|`BACKUP_PG`                     |Включает или отключает функцию создания резервных копий средствами PostgreSQL для совместимых информационных баз (базы данных которых размещены на СУБД PostgreSQL), принимает значения `True` или `False`|
+|`BACKUP_RETENTION_DAYS`         |Копии старше, чем количество дней в этой настройке будут удалсяться из каталога резервных копий при работе резервного копирования|
+|`BACKUP_REPLICATION`            |Включает или отключает функцию копирования резервных копий в дополнительные локации (например на сетевой диск), принимает значения `True` или `False`|
+|`BACKUP_REPLICATION_CONCURRENCY`|Параллелизм: сколько резервных копий может копироваться в места репликации одновременно|
+|`BACKUP_REPLICATION_PATHS`      |Список путей, куда резервные копии будут реплицированы|
+|`BACKUP_RETRIES_V8`             |Количество повторных попыток создания резервной копии средствами 1С Предприятие в случае возникновении ошибки. Если установлено значение 0, повторные попытки предприниматься не будут|
+|`BACKUP_RETRIES_PG`             |Количество повторных попыток загрузки резервной копии средствами PostgreSQL (см. секцию PostgreSQL). Если установлено значение 0, повторные попытки предприниматься не будут|
+|`BACKUP_TIMEOUT_V8`             |Таймаут в секундах, по истечению которого резервное копирование информационной базы считается неуспешным и принудительно завершается|
 
 ### Update
 

backup.py

@@ -2,21 +2,20 @@
 import logging
 import os
 import pathlib
-import sys
 from datetime import datetime
 from typing import List
 
 import aioshutil
 
 import core.models as core_models
 from conf import settings
-from core import utils
+from core import aws, utils
 from core.analyze import analyze_backup_result, analyze_s3_result
-from core.aws import upload_infobase_to_s3
 from core.cluster import utils as cluster_utils
 from core.exceptions import SubprocessException, V8Exception
 from core.process import execute_subprocess_command, execute_v8_command
 from utils import postgres
+from utils.asyncio import initialize_event_loop, initialize_semaphore
 from utils.log import configure_logging
 from utils.notification import make_html_table, send_notification
 
@@ -114,7 +113,8 @@ async def _backup_pgdump(
     Выполняет резервное копирование ИБ средствами СУБД PostgreSQL при помощи утилиты pg_dump
     1. Проверяет, использует ли ИБ СУБД PostgreSQL
     2. Проверяет, есть ли подходящие учетные данные для подключения к базе данных
-    3. Создаёт резервную копию средствами pg_dump
+    3. Подключается к СУБД, чтобы узнать версию
+    4. Создаёт резервную копию средствами pg_dump
     :param ib_name:
     :return:
     """
@@ -124,24 +124,24 @@ async def _backup_pgdump(
     except (ValueError, KeyError) as e:
         log.error(f"<{ib_name}> {str(e)}")
         return core_models.InfoBaseBackupTaskResult(ib_name, False)
+
+    try:
+        pg_major_version = (await postgres.get_postgres_version(db_host, db_port, db_name, db_user, db_pwd)).major
+        blobs = "large-objects" if pg_major_version >= 16 else "blobs"
+    except ConnectionRefusedError as e:
+        log.error(f"<{ib_name}> {str(e)}")
+        return core_models.InfoBaseBackupTaskResult(ib_name, False)
+
     ib_and_time_str = utils.get_ib_and_time_string(ib_name)
     backup_filename = os.path.join(
         settings.BACKUP_PATH, utils.append_file_extension_to_string(ib_and_time_str, "pgdump")
     )
     log_filename = os.path.join(settings.LOG_PATH, utils.append_file_extension_to_string(ib_and_time_str, "log"))
     pg_dump_path = os.path.join(settings.PG_BIN_PATH, "pg_dump.exe")
-    # --blobs
-    # Include large objects in the dump.
-    # This is the default behavior except when --schema, --table, or --schema-only is specified.
-    #
-    # --format=custom
-    # Output a custom-format archive suitable for input into pg_restore.
-    # Together with the directory output format, this is the most flexible output format in that it allows
-    # manual selection and reordering of archived items during restore. This format is also compressed by default.
     pgdump_command = (
         rf'"{pg_dump_path}" '
         rf"--host={db_host} --port={db_port} --username={db_user} "
-        rf"--format=custom --blobs --verbose "
+        rf"--format=custom --{blobs} --verbose "
         rf"--file={backup_filename} --dbname={db_name} > {log_filename} 2>&1"
     )
     pgdump_env = os.environ.copy()
@@ -185,12 +185,6 @@ async def backup_info_base(ib_name: str, semaphore: asyncio.Semaphore) -> core_m
         except Exception:
             log.exception(f"<{ib_name}> Unknown exception occurred in `_backup_info_base` coroutine")
             return core_models.InfoBaseBackupTaskResult(ib_name, False)
-        try:
-            # Если включена репликация и результат бэкапа успешен
-            if settings.BACKUP_REPLICATION and result.succeeded:
-                await replicate_backup(result.backup_filename, settings.BACKUP_REPLICATION_PATHS)
-        except Exception:
-            log.exception(f"<{ib_name}> Unknown exception occurred in `replicate_backup` coroutine")
         try:
             # Ротация бэкапов, удаляет старые
             await rotate_backups(ib_name)
@@ -199,6 +193,37 @@ async def backup_info_base(ib_name: str, semaphore: asyncio.Semaphore) -> core_m
         return result
 
 
+async def replicate_info_base(backup_result: core_models.InfoBaseBackupTaskResult, semaphore: asyncio.Semaphore):
+    async with semaphore:
+        try:
+            if settings.BACKUP_REPLICATION and backup_result.succeeded:
+                await replicate_backup(backup_result.backup_filename, settings.BACKUP_REPLICATION_PATHS)
+        except Exception:
+            log.exception(f"<{backup_result.infobase_name}> Unknown exception occurred in `replicate_backup` coroutine")
+
+
+def create_aws_upload_task(
+    backup_result: core_models.InfoBaseBackupTaskResult,
+    aws_semaphore: asyncio.Semaphore,
+):
+    if settings.AWS_ENABLED and backup_result.succeeded:
+        return asyncio.create_task(
+            aws.upload_infobase_to_s3(backup_result.infobase_name, backup_result.backup_filename, aws_semaphore),
+            name=f"Task :: Upload {backup_result.infobase_name} to S3",
+        )
+
+
+def create_backup_replication_task(
+    backup_result: core_models.InfoBaseBackupTaskResult,
+    backup_replication_semaphore: asyncio.Semaphore,
+):
+    if settings.BACKUP_REPLICATION and backup_result.succeeded:
+        return asyncio.create_task(
+            replicate_info_base(backup_result, backup_replication_semaphore),
+            name=f"Task :: Replicate backup {backup_result.infobase_name}",
+        )
+
+
 def analyze_results(
     infobases: List[str],
     backup_result: List[core_models.InfoBaseBackupTaskResult],
@@ -227,46 +252,46 @@ def send_email_notification(
 
 async def main():
     try:
-        # Если скрипт используется через планировщик задач windows, лучше всего логгировать консольный вывод в файл
-        # Например: backup.py >> D:\backup\log\1cv8-mgmt-backup-system.log 2>&1
-        cci = cluster_utils.get_cluster_controller_class()()
-        info_bases = cci.get_info_bases()
-        backup_concurrency = settings.BACKUP_CONCURRENCY
-        aws_concurrency = settings.AWS_CONCURRENCY
+        info_bases = utils.get_info_bases()
+        backup_semaphore = initialize_semaphore(settings.BACKUP_CONCURRENCY, log_prefix, "backup")
+        aws_semaphore = (
+            initialize_semaphore(settings.AWS_CONCURRENCY, log_prefix, "AWS") if settings.AWS_ENABLED else None
+        )
+        backup_replication_semaphore = (
+            initialize_semaphore(settings.BACKUP_REPLICATION_CONCURRENCY, log_prefix, "backup replication")
+            if settings.BACKUP_REPLICATION
+            else None
+        )
+
         backup_results = []
         aws_results = []
 
-        backup_semaphore = asyncio.Semaphore(backup_concurrency)
-        aws_semaphore = asyncio.Semaphore(aws_concurrency)
-        log.info(
-            f"<{log_prefix}> Asyncio semaphores initialized: {backup_concurrency} backup concurrency, {aws_concurrency} AWS concurrency"
-        )
         backup_coroutines = [backup_info_base(ib_name, backup_semaphore) for ib_name in info_bases]
         backup_datetime_start = datetime.now()
         aws_tasks = []
         aws_datetime_start = None
+        backup_replication_tasks = []
         for backup_coro in asyncio.as_completed(backup_coroutines):
             backup_result = await backup_coro
             backup_results.append(backup_result)
-            backup_datetime_finish = datetime.now()
-            # Только резервные копии, созданные без ошибок нужно загрузить на S3
-            if backup_result.succeeded and settings.AWS_ENABLED:
-                if aws_datetime_start is None:
-                    aws_datetime_start = datetime.now()
-                aws_tasks.append(
-                    asyncio.create_task(
-                        upload_infobase_to_s3(
-                            backup_result.infobase_name, backup_result.backup_filename, aws_semaphore
-                        ),
-                        name=f"Task :: Upload {backup_result.infobase_name} to S3",
-                    )
-                )
+            if aws_datetime_start is None:
+                aws_datetime_start = datetime.now()
+            aws_upload_task = create_aws_upload_task(backup_result, aws_semaphore)
+            if aws_upload_task:
+                aws_tasks.append(aws_upload_task)
+            backup_replication_task = create_backup_replication_task(backup_result, backup_replication_semaphore)
+            if backup_replication_task:
+                backup_replication_tasks.append(backup_replication_task)
+        backup_datetime_finish = datetime.now()
 
         if aws_tasks:
             await asyncio.wait(aws_tasks)
             aws_results = [task.result() for task in aws_tasks]
         aws_datetime_finish = datetime.now()
 
+        if backup_replication_tasks:
+            await asyncio.wait(backup_replication_tasks)
+
         analyze_results(
             info_bases,
             backup_results,
@@ -286,10 +311,4 @@ async def main():
 
 if __name__ == "__main__":
     configure_logging(settings.LOG_LEVEL)
-    if sys.version_info < (3, 10):
-        # Использование asyncio.run() в windows бросает исключение
-        # `RuntimeError: Event loop is closed` при завершении run.
-        # WindowsSelectorEventLoopPolicy не работает с подпроцессами полноценно в python 3.8
-        asyncio.get_event_loop().run_until_complete(main())
-    else:
-        asyncio.run(main())
+    initialize_event_loop(main())

conf/default_settings.py

@@ -27,6 +27,7 @@
 BACKUP_PG = False
 BACKUP_RETENTION_DAYS = 30
 BACKUP_REPLICATION = False
+BACKUP_REPLICATION_CONCURRENCY = 3
 BACKUP_REPLICATION_PATHS = [
     join("\\\\192.168.1.2", "backup", "1cv8"),
 ]