Logging und refactoring verbesserungen
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
package de.galabau.dateieingang.pipeline;
|
||||
|
||||
import de.galabau.dateieingang.config.SftpConfig;
|
||||
import de.galabau.dateieingang.exception.OciException;
|
||||
import de.galabau.dateieingang.exception.OrdsException;
|
||||
import de.galabau.dateieingang.exception.SftpException;
|
||||
@@ -19,6 +20,8 @@ import org.slf4j.MDC;
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.time.Duration;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
@@ -44,6 +47,9 @@ public class FileProcessingPipeline {
|
||||
@Inject
|
||||
OrdsNotificationService ordsNotificationService;
|
||||
|
||||
@Inject
|
||||
SftpConfig sftpConfig;
|
||||
|
||||
@Inject
|
||||
ManagedExecutor executor;
|
||||
|
||||
@@ -70,9 +76,11 @@ public class FileProcessingPipeline {
|
||||
return true;
|
||||
}
|
||||
|
||||
void processAll() {
|
||||
private void processAll() {
|
||||
Log.info("Pipeline-Lauf gestartet");
|
||||
|
||||
preProcessingCleanup();
|
||||
|
||||
List<String> zipFiles;
|
||||
try {
|
||||
zipFiles = sftpService.listZipFiles();
|
||||
@@ -99,6 +107,7 @@ public class FileProcessingPipeline {
|
||||
private void processZip(String zipFilename) {
|
||||
ProcessingContext context = new ProcessingContext(UUID.randomUUID(), zipFilename);
|
||||
MDC.put("runId", context.runId.toString());
|
||||
Log.infof("Starte Verarbeitung von '%s' [runId=%s]", zipFilename, context.runId);
|
||||
|
||||
try {
|
||||
// --- Download ---
|
||||
@@ -115,18 +124,20 @@ public class FileProcessingPipeline {
|
||||
|
||||
// --- OCI Upload ---
|
||||
MDC.put("step", "oci-upload");
|
||||
context.status = ProcessingStatus.PARTIALLY_UPLOADED;
|
||||
ociUploadService.upload(context);
|
||||
context.status = ProcessingStatus.MARKER_UPLOADED;
|
||||
|
||||
// --- SFTP Rename → .processed ---
|
||||
MDC.put("step", "sftp-rename");
|
||||
sftpService.renameRemote(zipFilename, zipFilename + ".processed");
|
||||
Log.infof("SFTP Rename: '%s' → '%s.processed'", zipFilename, zipFilename);
|
||||
sftpService.renameFile(zipFilename, zipFilename + ".processed");
|
||||
|
||||
// --- ORDS Notify ---
|
||||
MDC.put("step", "ords-notify");
|
||||
ordsNotificationService.notify(context);
|
||||
ordsNotificationService.triggerDbProcessing(context);
|
||||
|
||||
context.status = ProcessingStatus.ORDS_NOTIFIED;
|
||||
Log.infof("Verarbeitung erfolgreich abgeschlossen: '%s'", zipFilename);
|
||||
|
||||
} catch (SftpException | ZipException | OciException | OrdsException e) {
|
||||
Log.errorf(e, "Verarbeitung von '%s' fehlgeschlagen: %s", zipFilename, e.getMessage());
|
||||
@@ -137,7 +148,10 @@ public class FileProcessingPipeline {
|
||||
context.status = ProcessingStatus.FAILED;
|
||||
tryRenameToError(zipFilename);
|
||||
} finally {
|
||||
cleanup(context);
|
||||
postProcessingCleanup(context);
|
||||
long durationSeconds = Duration.between(context.startTime, LocalDateTime.now()).toSeconds();
|
||||
Log.infof("Lauf %s abgeschlossen — Status: %s, Dauer: %ds",
|
||||
context.runId, context.status, durationSeconds);
|
||||
MDC.clear();
|
||||
}
|
||||
}
|
||||
@@ -145,24 +159,73 @@ public class FileProcessingPipeline {
|
||||
private void tryRenameToError(String zipFilename) {
|
||||
try {
|
||||
MDC.put("step", "sftp-rename");
|
||||
sftpService.renameRemote(zipFilename, zipFilename + ".error");
|
||||
Log.infof("SFTP Rename: '%s' → '%s.error'", zipFilename, zipFilename);
|
||||
sftpService.renameFile(zipFilename, zipFilename + ".error");
|
||||
} catch (SftpException e) {
|
||||
Log.warnf(e, "Umbenennen zu .error fehlgeschlagen für '%s' — Datei bleibt auf SFTP zur manuellen Prüfung",
|
||||
zipFilename);
|
||||
}
|
||||
}
|
||||
|
||||
private void cleanup(ProcessingContext context) {
|
||||
MDC.put("step", "cleanup");
|
||||
/**
|
||||
* Bereinigt verwaiste lokale Arbeitsdateien aus fehlgeschlagenen Vorläufen.
|
||||
*
|
||||
* <p>Wird einmal pro Pipeline-Lauf <em>vor</em> dem SFTP-Listing aufgerufen.
|
||||
* Notwendig weil {@link #postProcessingCleanup} zwar im {@code finally}-Block läuft,
|
||||
* aber bei I/O-Fehlern selbst fehlschlagen kann — in diesem Fall bleiben ZIP-Dateien
|
||||
* und Entpack-Verzeichnisse im Arbeitsverzeichnis zurück. Ohne diesen Schritt würden
|
||||
* sich diese Reste akkumulieren und das Arbeitsverzeichnis über Zeit vollschreiben.
|
||||
*
|
||||
* <p>Ein Zeitstempel-Schwellwert ist nicht nötig: der {@code AtomicBoolean}-Guard in
|
||||
* {@link #tryProcessAllAsync} stellt sicher dass nie zwei Läufe gleichzeitig aktiv sind.
|
||||
* Alles was beim Start eines Laufs im Arbeitsverzeichnis liegt, ist daher garantiert
|
||||
* ein Überrest eines abgeschlossenen oder abgebrochenen Vorlaufs.
|
||||
*/
|
||||
private void preProcessingCleanup() {
|
||||
MDC.put("step", "pre-cleanup");
|
||||
Path workDir = Path.of(sftpConfig.localWorkDir());
|
||||
if (!Files.exists(workDir)) {
|
||||
return;
|
||||
}
|
||||
try (Stream<Path> entries = Files.list(workDir)) {
|
||||
entries.forEach(path -> {
|
||||
try {
|
||||
if (Files.isDirectory(path)) {
|
||||
deleteLocalDirectory(path);
|
||||
Log.warnf("Verwaistes Entpack-Verzeichnis gelöscht: %s", path);
|
||||
} else {
|
||||
Files.delete(path);
|
||||
Log.warnf("Verwaiste Datei gelöscht: %s", path);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
Log.warnf(e, "Pre-Cleanup fehlgeschlagen für: %s", path);
|
||||
}
|
||||
});
|
||||
} catch (IOException e) {
|
||||
Log.warnf(e, "Pre-Cleanup: Arbeitsverzeichnis konnte nicht gelesen werden: %s", workDir);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Bereinigt die lokalen Arbeitsdateien eines abgeschlossenen Laufs (ZIP + Entpack-Verzeichnis).
|
||||
*
|
||||
* <p>Wird im {@code finally}-Block von {@link #processZip} aufgerufen, also sowohl nach
|
||||
* erfolgreichem Abschluss als auch nach Fehlern. Schlägt dieser Cleanup bei I/O-Problemen
|
||||
* fehl, verbleiben die Dateien im Arbeitsverzeichnis — sie werden dann beim nächsten
|
||||
* Pipeline-Lauf durch {@link #preProcessingCleanup} entfernt.
|
||||
*
|
||||
* @param context enthält die Pfade der zu löschenden lokalen ZIP und des Entpack-Verzeichnisses
|
||||
*/
|
||||
private void postProcessingCleanup(ProcessingContext context) {
|
||||
MDC.put("step", "post-cleanup");
|
||||
Log.infof("Cleanup gestartet für Lauf %s", context.runId);
|
||||
try {
|
||||
if (context.localZipPath != null) {
|
||||
Files.deleteIfExists(context.localZipPath);
|
||||
Log.debugf("Lokale ZIP gelöscht: %s", context.localZipPath);
|
||||
Log.infof("Lokale ZIP gelöscht: %s", context.localZipPath);
|
||||
}
|
||||
if (context.localExtractDir != null) {
|
||||
deleteLocalDirectory(context.localExtractDir);
|
||||
Log.debugf("Lokales Entpack-Verzeichnis gelöscht: %s", context.localExtractDir);
|
||||
Log.infof("Lokales Entpack-Verzeichnis gelöscht: %s", context.localExtractDir);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
Log.warnf(e, "Cleanup für Lauf %s fehlgeschlagen — lokale Dateien verbleiben ggf. in %s",
|
||||
|
||||
Reference in New Issue
Block a user