diff --git a/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/exceptions/BatchExecutionFailedException.java b/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/exceptions/BatchExecutionFailedException.java new file mode 100644 index 00000000..3a36e240 --- /dev/null +++ b/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/exceptions/BatchExecutionFailedException.java @@ -0,0 +1,7 @@ +package fr.insee.kraftwerk.api.exceptions; + +public class BatchExecutionFailedException extends RuntimeException { + public BatchExecutionFailedException(Throwable cause) { + super(cause); + } +} diff --git a/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/BatchExportService.java b/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/BatchExportService.java index 12b1f966..8c06e719 100644 --- a/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/BatchExportService.java +++ b/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/BatchExportService.java @@ -5,6 +5,7 @@ import fr.insee.kraftwerk.api.configuration.MinioConfig; import fr.insee.kraftwerk.api.dto.BatchResponseDto; import fr.insee.kraftwerk.api.dto.ExportCheckResultDto; +import fr.insee.kraftwerk.api.exceptions.BatchExecutionFailedException; import fr.insee.kraftwerk.api.process.MainProcessing; import fr.insee.kraftwerk.api.process.MainProcessingGenesisNew; import fr.insee.kraftwerk.api.services.async.InMemoryExportJobStore; @@ -41,6 +42,7 @@ public class BatchExportService extends KraftwerkService { private final InMemoryExportJobStore exportJobStore; private final OutputZipService outputZipService; + private static final String BATCH_RESPONSE_LOG = "Batch response: {}"; private MinioClient minioClient; private boolean useMinio; @@ -68,146 +70,63 @@ public BatchExportService( } } - - private String normalizePath(Path path) { - return path.toString().replace("\\", "/"); - } - - FileUtilsInterface getFileUtilsInterface() { - if (useMinio) { - return new MinioImpl(minioClient, minioConfig.getBucketName()); - } - return new FileSystemImpl(configProperties.getDefaultDirectory()); - } - - private MainProcessing getMainProcessing( + //LEGACY + public BatchResponseDto mainServiceBatch( String inDirectoryParam, - boolean fileByFile, - boolean withDDI, + boolean archiveAtEnd, boolean withEncryption, - FileUtilsInterface fileUtilsInterface, boolean addStates ) { - KraftwerkExecutionContext kraftwerkExecutionContext = new KraftwerkExecutionContext( + return runLegacyBatchExport( inDirectoryParam, - fileByFile, - withDDI, + archiveAtEnd, withEncryption, - limitSize, - addStates - ); - - return new MainProcessing(kraftwerkExecutionContext, defaultDirectory, fileUtilsInterface); + addStates, + false, + true); } - MainProcessingGenesisNew getMainProcessingGenesisByQuestionnaire( - boolean withDDI, + public BatchResponseDto mainFileByFileBatch( + String inDirectoryParam, + boolean archiveAtEnd, boolean withEncryption, - FileUtilsInterface fileUtilsInterface, boolean addStates ) { - KraftwerkExecutionContext kraftwerkExecutionContext = new KraftwerkExecutionContext( - null, - false, - withDDI, + return runLegacyBatchExport( + inDirectoryParam, + archiveAtEnd, withEncryption, - limitSize, - addStates - ); - - return new MainProcessingGenesisNew( - configProperties, - new GenesisClient(new RestTemplateBuilder(), configProperties), - fileUtilsInterface, - kraftwerkExecutionContext - ); - } - - private Path buildBatchOutDirectoryForMain(String inDirectoryParam, LocalDateTime executionDateTime) { - return FileUtilsInterface.transformToOut(Paths.get(inDirectoryParam), executionDateTime); - } - - private Path buildBatchOutDirectoryForGenesis(String collectionInstrumentId, LocalDateTime executionDateTime) { - return Paths.get( - "out", - collectionInstrumentId, - executionDateTime.format(DateTimeFormatter.ofPattern(Constants.OUTPUT_FOLDER_DATETIME_PATTERN)) + addStates, + true, + true ); } - public BatchResponseDto mainServiceBatch( + public BatchResponseDto mainLunaticOnlyBatch( String inDirectoryParam, boolean archiveAtEnd, boolean withEncryption, boolean addStates ) { - boolean fileByFile = false; - boolean withDDI = true; - String jobId = UUID.randomUUID().toString(); - - FileUtilsInterface fileUtilsInterface = getFileUtilsInterface(); - MainProcessing mp = getMainProcessing( + return runLegacyBatchExport( inDirectoryParam, - fileByFile, - withDDI, + archiveAtEnd, withEncryption, - fileUtilsInterface, - addStates + addStates, + false, + false ); - - LocalDateTime executionDateTime = - mp.getKraftwerkExecutionContext().getExecutionDateTime(); - - Path outputPath = buildBatchOutDirectoryForMain(inDirectoryParam, executionDateTime); - exportJobStore.start(jobId); - - try { - mp.runMain(); - - outputZipService.encryptAndArchiveOutputs( - mp.getKraftwerkExecutionContext(), - fileUtilsInterface - ); - - if (archiveAtEnd) { - archive(inDirectoryParam, fileUtilsInterface); - } - - List errors = new ArrayList<>(); - - if (mp.getKraftwerkExecutionContext().getErrors() != null) { - mp.getKraftwerkExecutionContext() - .getErrors() - .forEach(error -> errors.add(error.toString())); - } - - ExportCheckResultDto result = new ExportCheckResultDto( - inDirectoryParam, - 0L - ); - - exportJobStore.complete(jobId, result, errors); - - - BatchResponseDto response = new BatchResponseDto(jobId,normalizePath(outputPath)); - log.info("Batch response: {}", response); - return response; - - } catch (KraftwerkException e) { - log.error("Batch main export failed for input directory {}", inDirectoryParam, e); - exportJobStore.fail(jobId, e); - throw new RuntimeException(e); - } } - public BatchResponseDto mainFileByFileBatch( + + private BatchResponseDto runLegacyBatchExport( String inDirectoryParam, boolean archiveAtEnd, boolean withEncryption, - boolean addStates + boolean addStates, + boolean fileByFile, + boolean withDDI ) { - boolean fileByFile = true; - boolean withDDI = true; String jobId = UUID.randomUUID().toString(); FileUtilsInterface fileUtilsInterface = getFileUtilsInterface(); @@ -254,80 +173,42 @@ public BatchResponseDto mainFileByFileBatch( } BatchResponseDto response = new BatchResponseDto(jobId,normalizePath(outputPath)); - log.info("Batch response: {}", response); + log.info(BATCH_RESPONSE_LOG, response); return response; } catch (KraftwerkException e) { - log.error("Batch file-by-file export failed for input directory {}", inDirectoryParam, e); + log.error("Batch legacy export failed for input directory {}", inDirectoryParam, e); exportJobStore.fail(jobId, e); - throw new RuntimeException(e); + throw new BatchExecutionFailedException(e); } } - public BatchResponseDto mainLunaticOnlyBatch( + private MainProcessing getMainProcessing( String inDirectoryParam, - boolean archiveAtEnd, + boolean fileByFile, + boolean withDDI, boolean withEncryption, + FileUtilsInterface fileUtilsInterface, boolean addStates ) { - boolean fileByFile = false; - boolean withDDI = false; - String jobId = UUID.randomUUID().toString(); - - FileUtilsInterface fileUtilsInterface = getFileUtilsInterface(); - MainProcessing mp = getMainProcessing( + KraftwerkExecutionContext kraftwerkExecutionContext = new KraftwerkExecutionContext( inDirectoryParam, fileByFile, withDDI, withEncryption, - fileUtilsInterface, + limitSize, addStates ); - LocalDateTime executionDateTime = - mp.getKraftwerkExecutionContext().getExecutionDateTime(); - - Path outputPath = buildBatchOutDirectoryForMain(inDirectoryParam, executionDateTime); - exportJobStore.start(jobId); - - try { - mp.runMain(); - - outputZipService.encryptAndArchiveOutputs( - mp.getKraftwerkExecutionContext(), - fileUtilsInterface - ); - - List errors = new ArrayList<>(); - - if (mp.getKraftwerkExecutionContext().getErrors() != null) { - mp.getKraftwerkExecutionContext() - .getErrors() - .forEach(error -> errors.add(error.toString())); - } - - ExportCheckResultDto result = new ExportCheckResultDto( - inDirectoryParam, - 0L - ); - - exportJobStore.complete(jobId, result, errors); - - if (archiveAtEnd) { - archive(inDirectoryParam, fileUtilsInterface); - } - - BatchResponseDto response = new BatchResponseDto(jobId,normalizePath(outputPath)); - log.info("Batch response: {}", response); - return response; + return new MainProcessing(kraftwerkExecutionContext, defaultDirectory, fileUtilsInterface); + } - } catch (KraftwerkException e) { - log.error("Batch export failed for input directory {}", inDirectoryParam, e); - exportJobStore.fail(jobId, e); - throw new RuntimeException(e); - } + private Path buildBatchOutDirectoryForMain(String inDirectoryParam, LocalDateTime executionDateTime) { + return FileUtilsInterface.transformToOut(Paths.get(inDirectoryParam), executionDateTime); } + //GENESIS + public BatchResponseDto mainGenesisByQuestionnaireIdBatch( String questionnaireModelId, Mode dataMode, @@ -335,65 +216,34 @@ public BatchResponseDto mainGenesisByQuestionnaireIdBatch( boolean withEncryption, boolean addStates ) { - boolean withDDI = true; - FileUtilsInterface fileUtilsInterface = getFileUtilsInterface(); - String jobId = UUID.randomUUID().toString(); + return runGenesisBatchExport(questionnaireModelId, dataMode, batchSize, withEncryption, addStates, true); + } - MainProcessingGenesisNew mpGenesis = getMainProcessingGenesisByQuestionnaire( - withDDI, + public BatchResponseDto mainGenesisLunaticOnlyByQuestionnaireBatch( + String questionnaireModelId, + Mode dataMode, + int batchSize, + boolean withEncryption, + boolean addStates + ) { + return runGenesisBatchExport( + questionnaireModelId, + dataMode, + batchSize, withEncryption, - fileUtilsInterface, - addStates + addStates, + false ); - - LocalDateTime executionDateTime = - mpGenesis.getKraftwerkExecutionContext().getExecutionDateTime(); - - Path outputPath = buildBatchOutDirectoryForGenesis(questionnaireModelId, executionDateTime); - - exportJobStore.start(jobId); - - try { - mpGenesis.runMain(questionnaireModelId, batchSize, dataMode); - - outputZipService.encryptAndArchiveOutputs( - mpGenesis.getKraftwerkExecutionContext(), - fileUtilsInterface - ); - - List errors = new ArrayList<>(); - - if (mpGenesis.getKraftwerkExecutionContext().getErrors() != null) { - mpGenesis.getKraftwerkExecutionContext() - .getErrors() - .forEach(error -> errors.add(error.toString())); - } - - ExportCheckResultDto result = new ExportCheckResultDto( - questionnaireModelId, - 0L - ); - - exportJobStore.complete(jobId, result, errors); - - BatchResponseDto response = new BatchResponseDto(jobId, normalizePath(outputPath)); - log.info("Batch response: {}", response); - return response; - - } catch (KraftwerkException | IOException e) { - exportJobStore.fail(jobId, e); - throw new RuntimeException(e); - } } - public BatchResponseDto mainGenesisLunaticOnlyByQuestionnaireBatch( + private BatchResponseDto runGenesisBatchExport( String questionnaireModelId, Mode dataMode, int batchSize, boolean withEncryption, - boolean addStates + boolean addStates, + boolean withDDI ) { - boolean withDDI = false; FileUtilsInterface fileUtilsInterface = getFileUtilsInterface(); String jobId = UUID.randomUUID().toString(); @@ -408,6 +258,7 @@ public BatchResponseDto mainGenesisLunaticOnlyByQuestionnaireBatch( mpGenesis.getKraftwerkExecutionContext().getExecutionDateTime(); Path outputPath = buildBatchOutDirectoryForGenesis(questionnaireModelId, executionDateTime); + exportJobStore.start(jobId); try { @@ -417,7 +268,9 @@ public BatchResponseDto mainGenesisLunaticOnlyByQuestionnaireBatch( mpGenesis.getKraftwerkExecutionContext(), fileUtilsInterface ); + List errors = new ArrayList<>(); + if (mpGenesis.getKraftwerkExecutionContext().getErrors() != null) { mpGenesis.getKraftwerkExecutionContext() .getErrors() @@ -431,15 +284,13 @@ public BatchResponseDto mainGenesisLunaticOnlyByQuestionnaireBatch( exportJobStore.complete(jobId, result, errors); - - BatchResponseDto response = new BatchResponseDto(jobId,normalizePath(outputPath)); - log.info("Batch response: {}", response); + BatchResponseDto response = new BatchResponseDto(jobId, normalizePath(outputPath)); + log.info(BATCH_RESPONSE_LOG, response); return response; } catch (KraftwerkException | IOException e) { - log.error("Batch export failed for questionnaireModelId {}", questionnaireModelId, e); exportJobStore.fail(jobId, e); - throw new RuntimeException(e); + throw new BatchExecutionFailedException(e); } } @@ -467,7 +318,7 @@ public ResponseEntity jsonExtractionBatch( try { mpGenesis.runMainJson(collectionInstrumentId, batchSize, dataMode, since); BatchResponseDto response = new BatchResponseDto("",normalizePath(outputPath)); - log.info("Batch response: {}", response); + log.info(BATCH_RESPONSE_LOG, response); return ResponseEntity.ok(response); } catch (KraftwerkException e) { return ResponseEntity.status(e.getStatus()).body(e.getMessage()); @@ -475,4 +326,47 @@ public ResponseEntity jsonExtractionBatch( return ResponseEntity.internalServerError().body(e.getMessage()); } } + + MainProcessingGenesisNew getMainProcessingGenesisByQuestionnaire( + boolean withDDI, + boolean withEncryption, + FileUtilsInterface fileUtilsInterface, + boolean addStates + ) { + KraftwerkExecutionContext kraftwerkExecutionContext = new KraftwerkExecutionContext( + null, + false, + withDDI, + withEncryption, + limitSize, + addStates + ); + + return new MainProcessingGenesisNew( + configProperties, + new GenesisClient(new RestTemplateBuilder(), configProperties), + fileUtilsInterface, + kraftwerkExecutionContext + ); + } + + private Path buildBatchOutDirectoryForGenesis(String collectionInstrumentId, LocalDateTime executionDateTime) { + return Paths.get( + "out", + collectionInstrumentId, + executionDateTime.format(DateTimeFormatter.ofPattern(Constants.OUTPUT_FOLDER_DATETIME_PATTERN)) + ); + } + + //UTILS + private String normalizePath(Path path) { + return path.toString().replace("\\", "/"); + } + + FileUtilsInterface getFileUtilsInterface() { + if (useMinio) { + return new MinioImpl(minioClient, minioConfig.getBucketName()); + } + return new FileSystemImpl(configProperties.getDefaultDirectory()); + } } diff --git a/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/OutputZipService.java b/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/OutputZipService.java index 9ef6c5e6..e45aacf9 100644 --- a/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/OutputZipService.java +++ b/kraftwerk-api/src/main/java/fr/insee/kraftwerk/api/services/OutputZipService.java @@ -235,8 +235,13 @@ private void deleteWithRetry(Path path) { try { Files.deleteIfExists(path); return; - } catch (IOException e) { - try { Thread.sleep(80L); } catch (InterruptedException ignored) {} + } catch (IOException _) { + try { + Thread.sleep(80L); + } + catch (InterruptedException _) { + Thread.currentThread().interrupt(); + } } } path.toFile().deleteOnExit();