fix(1c): harden live ingestion path

ansible/deploy_file_1c_analytics.yml

@@ -148,6 +148,7 @@
 
           archive_dir: {{ aw_file_1c_root }}/archive
           delete_after_load: false
+          min_file_age_seconds: 180
 
     - name: Создать symlink на активный root file-1C analytics
       ansible.builtin.file:

clickhouse-1c/etl/load_1c_exports.py

@@ -6,7 +6,7 @@ import csv
 import json
 import shutil
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -38,6 +38,7 @@ class Config:
     formats: dict[str, str]
     archive_dir: str | None
     delete_after_load: bool
+    min_file_age_seconds: int
 
 
 def parse_args() -> argparse.Namespace:
@@ -55,6 +56,7 @@ def load_config(path: str) -> Config:
         formats=raw.get("formats", {"default": "jsonl"}),
         archive_dir=raw.get("archive_dir"),
         delete_after_load=bool(raw.get("delete_after_load", False)),
+        min_file_age_seconds=int(raw.get("min_file_age_seconds", 180)),
     )
 
 
@@ -201,6 +203,10 @@ def main() -> int:
         if not landing.exists():
             continue
         for path in sorted(p for p in landing.iterdir() if p.is_file()):
+            age_seconds = max(0, int((datetime.now(UTC) - datetime.fromtimestamp(path.stat().st_mtime, UTC)).total_seconds()))
+            if age_seconds < conf.min_file_age_seconds:
+                print(f"skip {dataset}: {path.name} age={age_seconds}s < min_file_age_seconds={conf.min_file_age_seconds}")
+                continue
             rows = iter_rows(path, fmt)
             if not rows:
                 archive_or_delete(conf, dataset, path)

clickhouse-1c/ops/run_ingest_cycle.sh

@@ -6,6 +6,7 @@ ENV_FILE="${ROOT}/.env"
 VENV="${ROOT}/.venv"
 CONFIG="${ROOT}/etl/config.yml"
 CH_CONTAINER="${AW_1C_CLICKHOUSE_CONTAINER:-aw-rus-1c-clickhouse}"
+LOCK_FILE="${ROOT}/.ingest.lock"
 
 if [[ ! -f "${ENV_FILE}" ]]; then
   echo "missing env file: ${ENV_FILE}" >&2
@@ -27,6 +28,12 @@ if ! docker ps --format '{{.Names}}' | grep -qx "${CH_CONTAINER}"; then
   exit 1
 fi
 
+exec 9>"${LOCK_FILE}"
+if ! flock -n 9; then
+  echo "ingest cycle already running" >&2
+  exit 0
+fi
+
 # shellcheck disable=SC1090
 . "${ENV_FILE}"
 

windows/export-upload-file-1c-telemetry.ps1

@@ -106,6 +106,30 @@ function Write-JsonLines {
         Set-Content -LiteralPath $Path -Encoding UTF8
 }
 
+function Invoke-SshUploadWithRetry {
+    param(
+        [Parameter(Mandatory = $true)]
+        [string]$KeyPath,
+        [Parameter(Mandatory = $true)]
+        [string]$SourcePath,
+        [Parameter(Mandatory = $true)]
+        [string]$Destination,
+        [int]$Attempts = 3,
+        [int]$DelaySeconds = 5
+    )
+
+    for ($attempt = 1; $attempt -le $Attempts; $attempt++) {
+        & scp.exe -q -i $KeyPath -o LogLevel=ERROR -o StrictHostKeyChecking=no -o UserKnownHostsFile=NUL $SourcePath $Destination
+        if ($LASTEXITCODE -eq 0) {
+            return
+        }
+        if ($attempt -ge $Attempts) {
+            throw "scp upload failed after $Attempts attempts for $SourcePath with rc=$LASTEXITCODE"
+        }
+        Start-Sleep -Seconds $DelaySeconds
+    }
+}
+
 if (-not (Test-Path -LiteralPath $RemoteKeyPath)) {
     throw "SSH private key not found: $RemoteKeyPath"
 }
@@ -252,10 +276,7 @@ $effectiveKeyPath = New-TemporarySshKeyCopy -SourceKeyPath $RemoteKeyPath
 
 try {
     foreach ($dataset in 'documents', 'reglog', 'audit', 'host') {
-        & scp.exe -q -i $effectiveKeyPath -o LogLevel=ERROR -o StrictHostKeyChecking=no -o UserKnownHostsFile=NUL $files[$dataset] "$AnalyticsUser@$AnalyticsHost`:$RemoteRoot/$dataset/"
+        Invoke-SshUploadWithRetry -KeyPath $effectiveKeyPath -SourcePath ([string]$files[$dataset]) -Destination "$AnalyticsUser@$AnalyticsHost`:$RemoteRoot/$dataset/"
-        if ($LASTEXITCODE -ne 0) {
-            throw "scp upload failed for dataset $dataset with rc=$LASTEXITCODE"
-        }
     }
 }
 finally {