- After a WorkerReport has been sent, ask for new assignments immediately. So, the Worker does not have to wait for hours for the Controller to check for duplicate files in the DB, retrieve and upload the full-texts and insert the records to the DB.
- Special care is taken to delete the delivered full-texts as soon as possible. - Write the workerReport to a json-file, in case something goes wrong, and keep it until the Controller notifies the Worker that the processing was successful.
This commit is contained in:
parent
9cb43b3d94
commit
903032f454
|
|
@ -41,6 +41,9 @@ dependencies {
|
|||
|
||||
implementation group: 'com.google.guava', name: 'guava', version: '31.1-jre'
|
||||
|
||||
// https://mvnrepository.com/artifact/com.google.code.gson/gson
|
||||
implementation 'com.google.code.gson:gson:2.10.1'
|
||||
|
||||
implementation 'org.apache.commons:commons-compress:1.23.0'
|
||||
implementation 'com.github.luben:zstd-jni:1.5.5-3' // Even though this is part of the above dependency, it is needed separately as well, specifically here, in the Worker.
|
||||
|
||||
|
|
|
|||
|
|
@ -91,10 +91,11 @@ public class UrlsWorkerApplication {
|
|||
}
|
||||
}
|
||||
|
||||
FullTextsController.deleteDirectory(-1);
|
||||
FullTextsController.deleteAssignmentsDirectory(-1, null);
|
||||
logger.info("Exiting..");
|
||||
}
|
||||
|
||||
|
||||
@Bean
|
||||
public CorsConfigurationSource corsConfigurationSource() {
|
||||
CorsConfiguration configuration = new CorsConfiguration();
|
||||
|
|
|
|||
|
|
@ -7,14 +7,15 @@ import eu.openaire.publications_retriever.util.url.GenericUtils;
|
|||
import eu.openaire.publications_retriever.util.url.UrlUtils;
|
||||
import eu.openaire.urls_worker.UrlsWorkerApplication;
|
||||
import eu.openaire.urls_worker.components.plugins.PublicationsRetrieverPlugin;
|
||||
import eu.openaire.urls_worker.controllers.FullTextsController;
|
||||
import eu.openaire.urls_worker.controllers.GeneralController;
|
||||
import eu.openaire.urls_worker.models.Assignment;
|
||||
import eu.openaire.urls_worker.models.UrlReport;
|
||||
import eu.openaire.urls_worker.payloads.requests.AssignmentsRequest;
|
||||
import eu.openaire.urls_worker.payloads.responces.WorkerReport;
|
||||
import eu.openaire.urls_worker.util.FilesCompressor;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.boot.web.client.RestTemplateBuilder;
|
||||
import org.springframework.http.HttpStatus;
|
||||
|
|
@ -24,9 +25,14 @@ import org.springframework.web.client.HttpServerErrorException;
|
|||
import org.springframework.web.client.RestClientException;
|
||||
import org.springframework.web.client.RestTemplate;
|
||||
|
||||
import java.io.BufferedWriter;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Paths;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
|
||||
@Component
|
||||
|
|
@ -34,6 +40,8 @@ public class AssignmentsHandler {
|
|||
|
||||
private static final Logger logger = LoggerFactory.getLogger(AssignmentsHandler.class);
|
||||
|
||||
@Autowired
|
||||
PublicationsRetrieverPlugin publicationsRetrieverPlugin;
|
||||
|
||||
private final String workerId;
|
||||
private final String controllerBaseUrl;
|
||||
|
|
@ -65,10 +73,13 @@ public class AssignmentsHandler {
|
|||
public static final long idUrlsToHandleBeforeClearingDuplicateUrlsData = 200_000;
|
||||
public static long timesClearingDuplicateUrlsData = 0;
|
||||
|
||||
public String workerReportsDirPath;
|
||||
|
||||
|
||||
public AssignmentsHandler(@Value("${info.workerId}") String workerId, @Value("${info.maxAssignmentsLimitPerBatch}") int maxAssignmentsLimitPerBatch,
|
||||
@Value("${info.maxAssignmentsBatchesToHandleBeforeShutdown}") int maxAssignmentsBatchesToHandleBeforeShutdown,
|
||||
@Value("${info.controllerBaseUrl}") String controllerBaseUrl)
|
||||
@Value("${info.controllerBaseUrl}") String controllerBaseUrl,
|
||||
@Value("${workerReportsDirPath}") String workerReportsDirPath)
|
||||
{
|
||||
this.workerId = workerId;
|
||||
this.maxAssignmentsLimitPerBatch = maxAssignmentsLimitPerBatch;
|
||||
|
|
@ -85,6 +96,18 @@ public class AssignmentsHandler {
|
|||
// X hours. Time to wait for the data to get transferred over the network. Many workers may try to get assignments from the Worker, so each worker might have to wait some 10s of minutes for work.
|
||||
// When giving the assignments, the Controller has to retrieve the data from the database, then prepare them in memory, insert them in the "assignment"-table and, finally, return them to the worker.
|
||||
// When receiving the Worker-Report, the Controller has to check for existing fulltext files (by previous runs), request and get thousands of file from the Worker (in batches), upload them to S3, prepare and import the payload and the attempt records in the database and return to the Worker.
|
||||
|
||||
if ( !workerReportsDirPath.endsWith("/") )
|
||||
workerReportsDirPath += "/";
|
||||
|
||||
this.workerReportsDirPath = workerReportsDirPath;
|
||||
try {
|
||||
Files.createDirectories(Paths.get(this.workerReportsDirPath)); // No-op if it already exists.
|
||||
} catch (Exception e) {
|
||||
String errorMsg = "Could not create the \"workerReportsDirPath\": " + this.workerReportsDirPath;
|
||||
logger.error(errorMsg, e);
|
||||
throw new RuntimeException(errorMsg);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -107,6 +130,9 @@ public class AssignmentsHandler {
|
|||
}
|
||||
|
||||
|
||||
public static boolean shouldNotRequestMore = false;
|
||||
|
||||
|
||||
public void handleAssignments()
|
||||
{
|
||||
AssignmentsRequest assignmentsRequest = requestAssignments();
|
||||
|
|
@ -157,7 +183,7 @@ public class AssignmentsHandler {
|
|||
|
||||
// For now, let's just run all tasks in the generic plugin.
|
||||
try {
|
||||
PublicationsRetrieverPlugin.processAssignments(assignmentRequestCounter, assignmentsForPlugins.values());
|
||||
publicationsRetrieverPlugin.processAssignments(assignmentRequestCounter, assignmentsForPlugins.values());
|
||||
} catch (Exception e) {
|
||||
logger.error("Exception when processing the assignments_" + assignmentRequestCounter, e);
|
||||
} // In this case, we will either have an empty WorkerReport or a half-filled one. Either way, we want to report back to the Controller.
|
||||
|
|
@ -197,12 +223,12 @@ public class AssignmentsHandler {
|
|||
if ( GeneralController.shouldShutdownWorker
|
||||
|| (AssignmentsHandler.numHandledAssignmentsBatches == this.maxAssignmentsBatchesToHandleBeforeShutdown) )
|
||||
{
|
||||
logger.info("The worker will now shutdown, as " + (GeneralController.shouldShutdownWorker
|
||||
logger.info("The worker will shutdown, after the full-texts are delivered to the Controller, as " + (GeneralController.shouldShutdownWorker
|
||||
? "it received a \"shutdownWorker\" request!"
|
||||
: "the maximum assignments-batches (" + this.maxAssignmentsBatchesToHandleBeforeShutdown + ") to be handled was reached!"));
|
||||
|
||||
// TODO - Should this worker inform the Controller that will shutdown?
|
||||
UrlsWorkerApplication.gentleAppShutdown(); // This will exit the app.
|
||||
// Here, just specify that we do not want to request for more assignments. A scheduling job will check if the fulltexts were delivered to the Controller and then shutdown the Worker.
|
||||
shouldNotRequestMore = true;
|
||||
}
|
||||
|
||||
// Note: Cannot call this method, here, retrospectively, as if it runs 100s of times, the memory-stack may break..
|
||||
|
|
@ -222,20 +248,20 @@ public class AssignmentsHandler {
|
|||
{
|
||||
String postUrl = this.controllerBaseUrl + "urls/addWorkerReport";
|
||||
logger.info("Going to post the WorkerReport of assignments_" + assignmentRequestCounter + " to the controller-server: " + postUrl);
|
||||
WorkerReport workerReport = new WorkerReport(this.workerId, assignmentRequestCounter, urlReports);
|
||||
|
||||
// Create the report file. It may be useful later, in case something goes wrong when sending the report to the Controller or the Controller cannot process it.
|
||||
// The report-file is deleted, along with the full-texts) when the Controller posts that the processing of this report was successful.
|
||||
writeToFile(this.workerReportsDirPath + this.workerId + "_assignments_" + assignmentRequestCounter + "_report.json", workerReport.getJsonReport(), false);
|
||||
|
||||
// The worker sends this "WorkerReport" to the Controller, which after some checks, it adds a job to a background thread and responds to the Worker with HTTP-200-OK.
|
||||
try {
|
||||
ResponseEntity<String> responseEntity = restTemplateForReport.postForEntity(postUrl, new WorkerReport(this.workerId, assignmentRequestCounter, urlReports), String.class);
|
||||
|
||||
// The worker sends the "WorkerReport" and before this "POST"-request returns here, the Controller, after analyzing the report, opens new request to the Worker in order to receive the full-texts.
|
||||
// After the report and the full-texts are received and uploaded to the database/S3, the Controller returns a response to the Worker.
|
||||
|
||||
ResponseEntity<String> responseEntity = restTemplateForReport.postForEntity(postUrl, workerReport, String.class);
|
||||
int responseCode = responseEntity.getStatusCodeValue();
|
||||
if ( responseCode == HttpStatus.OK.value() ) {
|
||||
logger.info("The submission of the WorkerReport of assignments_" + assignmentRequestCounter + " to the Controller, and the full-text delivering, were successful.");
|
||||
logger.info("The submission of the WorkerReport of assignments_" + assignmentRequestCounter + " to the Controller, was successful.");
|
||||
assignmentsNumsHandled.add(assignmentRequestCounter);
|
||||
return true;
|
||||
} else if ( responseCode == HttpStatus.MULTI_STATUS.value() ) {
|
||||
logger.warn("The submission of the WorkerReport of assignments_" + assignmentRequestCounter + " to the Controller was successful, but the full-texts' delivering failed!");
|
||||
return true;
|
||||
} else { // This does not include HTTP-5XX errors. For them an "HttpServerErrorException" is thrown.
|
||||
logger.error("HTTP-Connection problem with the submission of the WorkerReport of assignments_" + assignmentRequestCounter + " to the Controller! Error-code was: " + responseCode);
|
||||
return false;
|
||||
|
|
@ -249,11 +275,7 @@ public class AssignmentsHandler {
|
|||
} finally {
|
||||
urlReports.clear(); // Reset, without de-allocating.
|
||||
assignmentsForPlugins.clear();
|
||||
|
||||
FullTextsController.deleteDirectory(assignmentRequestCounter);
|
||||
// Even though we delete the full-texts batch-by-batch, some files may not have been previously deleted, since they may be duplicates of others found by previous assignments-batches
|
||||
// and thus, they may have not been requested by the Controller (and thus not deleted after transferring the batches).
|
||||
// Also, the ".tar.zstd" file of last batch will be deleted here, as well as the whole directory itself.
|
||||
// The full-text files will be deleted after being transferred to the Controller.
|
||||
}
|
||||
|
||||
// Note: It is possible that one or more full-texts-batches, are not sent to the Controller, or that the Controller failed to process them.
|
||||
|
|
@ -355,6 +377,28 @@ public class AssignmentsHandler {
|
|||
}
|
||||
|
||||
|
||||
Lock fileWriteLock = new ReentrantLock(true);
|
||||
|
||||
public String writeToFile(String fileFullPath, String stringToWrite, boolean shouldLockThreads)
|
||||
{
|
||||
if ( shouldLockThreads )
|
||||
fileWriteLock.lock();
|
||||
|
||||
try ( BufferedWriter bufferedWriter = new BufferedWriter(Files.newBufferedWriter(Paths.get(fileFullPath)), FilesCompressor.bufferSize) )
|
||||
{
|
||||
bufferedWriter.write(stringToWrite); // This will overwrite the file. If the new string is smaller, then it does not matter.
|
||||
} catch (Exception e) {
|
||||
String errorMsg = "Failed to create or acquire the file \"" + fileFullPath + "\"!";
|
||||
logger.error(errorMsg, e);
|
||||
return errorMsg;
|
||||
} finally {
|
||||
if ( shouldLockThreads )
|
||||
fileWriteLock.unlock();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
public static void countDatasourcesAndRecords(int assignmentsSize)
|
||||
{
|
||||
Set<String> datasources = assignmentsForPlugins.keySet();
|
||||
|
|
|
|||
|
|
@ -20,11 +20,17 @@ public class ScheduledTasks {
|
|||
|
||||
@Scheduled(fixedDelay = 1) // Request the next batch immediately after the last one finishes.
|
||||
public void handleNewAssignments() {
|
||||
if ( AssignmentsHandler.shouldNotRequestMore ) {
|
||||
// Here we will be right after the Worker has posted its last report. It is guaranteed that the Controller will not have processed it and have not requested the full-text files.
|
||||
// We do not want to shut down the controller.
|
||||
return;
|
||||
}
|
||||
|
||||
if ( AssignmentsHandler.hadConnectionErrorOnRequest ) {
|
||||
if ( GeneralController.shouldShutdownWorker ) // Make sure the worker shuts-down, in case the user sends the relevant request, while the worker is stuck in a data-request error-loop.
|
||||
// TODO - Should this worker inform the Controller that will shutdown?
|
||||
// TODO - Yes, apart from this knowledge, it may be the case that we make the Controller to request the Workers to shutdown.
|
||||
UrlsWorkerApplication.gentleAppShutdown();
|
||||
if ( GeneralController.shouldShutdownWorker ) { // Make sure the worker shuts-down, in case the user sends the relevant request, while the worker is stuck in a data-request error-loop.
|
||||
AssignmentsHandler.shouldNotRequestMore = true;
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(900_000); // Sleep for 15 mins to stall the scheduler from retrying right away, thus giving time to the Controller to recover.
|
||||
} catch (InterruptedException ie) {
|
||||
|
|
|
|||
|
|
@ -39,11 +39,14 @@ public class PublicationsRetrieverPlugin {
|
|||
|
||||
public static String assignmentsBasePath;
|
||||
|
||||
private static String workerId;
|
||||
|
||||
private static CookieStore cookieStore = null;
|
||||
|
||||
|
||||
public PublicationsRetrieverPlugin(@Value("${info.maxAssignmentsLimitPerBatch}") int maxAssignmentsLimitPerBatch, FileStorageService fileStorageService) {
|
||||
public PublicationsRetrieverPlugin(@Value("${info.workerId}")String workerId, @Value("${info.maxAssignmentsLimitPerBatch}") int maxAssignmentsLimitPerBatch, FileStorageService fileStorageService) {
|
||||
// Specify some configurations
|
||||
this.workerId = workerId;
|
||||
LoaderAndChecker.retrieveDocuments = true;
|
||||
LoaderAndChecker.retrieveDatasets = false;
|
||||
ConnSupportUtils.setKnownMimeTypes();
|
||||
|
|
@ -73,7 +76,7 @@ public class PublicationsRetrieverPlugin {
|
|||
|
||||
private static final List<Callable<Boolean>> callableTasks = new ArrayList<>(FileUtils.jsonBatchSize);
|
||||
|
||||
public static void processAssignments(Long assignmentRequestCounter, Collection<Assignment> assignments) throws RuntimeException
|
||||
public void processAssignments(Long assignmentRequestCounter, Collection<Assignment> assignments) throws RuntimeException
|
||||
{
|
||||
// At this point, the "assignmentsBasePath"-directory has already been successfully created.
|
||||
|
||||
|
|
|
|||
|
|
@ -68,10 +68,10 @@ public class FullTextsController {
|
|||
|
||||
File zstdFile = FilesCompressor.compressMultipleFilesIntoOne(assignmentsCounter, batchCounter, fileNamesWithExtensions, currentAssignmentsBaseFullTextsPath);
|
||||
if ( zstdFile == null ) {
|
||||
// The failed files (including the ".tar"), have already been deleted.
|
||||
String errorMsg = "Failed to create the zstd file for \"batchCounter\"-" + batchCounter;
|
||||
logger.error(errorMsg);
|
||||
return ResponseEntity.internalServerError().body(errorMsg);
|
||||
// The related files will be deleted later, upon completing the Worker-report process, in "AssignmentsHandler.postWorkerReport()".
|
||||
}
|
||||
|
||||
if ( batchCounter == totalBatches )
|
||||
|
|
@ -89,14 +89,11 @@ public class FullTextsController {
|
|||
logger.error(errorMsg, e);
|
||||
return ResponseEntity.internalServerError().body(errorMsg);
|
||||
} finally {
|
||||
// In some cases, the full-texts might be too large and their total number too,
|
||||
// so if we leave them be, and wait for all batches to finish, we may get a "java.io.IOException: No space left on device" error.
|
||||
deleteFulltextBatchFiles(currentAssignmentsBaseFullTextsPath, assignmentsCounter, batchCounter, fileNamesWithExtensions);
|
||||
// The ".tar.zstd" file of this batch, for which we pass a steam to the Controller, will be deleted after the next batch, or after all batches are transferred and handles by the Controller.
|
||||
// The ".tar.zstd" file of this batch, for which we pass a steam to the Controller, will be deleted by the next batch or in the end of these assignments.
|
||||
// Now we will delete the zstd file of the previous assignments.
|
||||
if ( batchCounter >= 2 )
|
||||
deleteFile(currentAssignmentsBaseFullTextsPath + "assignments_" + assignmentsCounter + "_full-texts_" + (batchCounter -1) + ".tar.zstd");
|
||||
}
|
||||
|
||||
// The related fulltext and (zstd-)tar files will be deleted in "AssignmentsHandler.postWorkerReport()", after the Controller has finished transferring them. They will be deleted even in case of a Controller-error.
|
||||
// In case of an error and file-deletion, the related id-url records will just be re-processed in the future by some (maybe different) Worker.
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -124,48 +121,33 @@ public class FullTextsController {
|
|||
}
|
||||
|
||||
|
||||
public static boolean deleteDirectory(long curAssignments)
|
||||
public static boolean deleteAssignmentsDirectory(long curAssignments, File dir)
|
||||
{
|
||||
String directoryPath = PublicationsRetrieverPlugin.assignmentsBasePath;
|
||||
|
||||
if ( curAssignments != -1 ) {
|
||||
directoryPath += "assignments_" + curAssignments + "_fullTexts" + File.separator;
|
||||
logger.debug("Going to delete the files inside the directory of assignments_" + curAssignments);
|
||||
} else
|
||||
logger.debug("Going to delete the parent directory: " + directoryPath);
|
||||
|
||||
try {
|
||||
FileUtils.deleteDirectory(new File(directoryPath));
|
||||
return true;
|
||||
} catch (IOException e) {
|
||||
logger.error("The following directory could not be deleted: " + directoryPath, e);
|
||||
return false;
|
||||
} catch (IllegalArgumentException iae) {
|
||||
logger.error("This assignments-dir does not exist: " + directoryPath);
|
||||
return false;
|
||||
if ( dir == null ) {
|
||||
String directoryPath = PublicationsRetrieverPlugin.assignmentsBasePath;
|
||||
if ( curAssignments != -1 ) {
|
||||
directoryPath += "assignments_" + curAssignments + "_fullTexts";
|
||||
logger.debug("Going to delete the files inside the directory of assignments_" + curAssignments);
|
||||
} else
|
||||
logger.debug("Going to delete the parent directory: " + directoryPath);
|
||||
dir = new File(directoryPath);
|
||||
}
|
||||
return deleteDirectory(dir);
|
||||
}
|
||||
|
||||
|
||||
public static void deleteFulltextBatchFiles(String assignmentsBatchDir, long assignmentsCounter, long fulltextsBatch, List<String> filenames)
|
||||
public static boolean deleteDirectory(File directory)
|
||||
{
|
||||
// We will delete all the files related to the given fulltexts-batch, along with the created tar and zstd files.
|
||||
|
||||
for ( String fileName : filenames )
|
||||
deleteFile(assignmentsBatchDir + fileName);
|
||||
|
||||
// Now let's delete the ".tar" and ".tar.zstd" files as well.
|
||||
String partialNonBatchFileName = assignmentsBatchDir + "assignments_" + assignmentsCounter + "_full-texts_";
|
||||
deleteFile(partialNonBatchFileName + fulltextsBatch + ".tar");
|
||||
|
||||
// The ".tar.zstd" file of this batch will be deleted by the next batch or in the end of these assignments.
|
||||
// Now we will delete the zstd file of the previous assignments.
|
||||
if ( fulltextsBatch >= 2 )
|
||||
deleteFile(partialNonBatchFileName + (fulltextsBatch -1) + ".tar.zstd");
|
||||
|
||||
// We do not use a fulltexts-batch directory, since even if it makes the deletion faster, it will make the full-texts delivery to the controller slower,
|
||||
// since we will need to move the requested full-texts to that directory before continuing with tarring and compressing the files and sending them over to the Controller.
|
||||
// Also, we cannot pre-create such directories, since it will add complexity in the download process and also some of the full-texts may not be requested by the Controller (because of duplicates).
|
||||
try {
|
||||
FileUtils.deleteDirectory(directory);
|
||||
return true;
|
||||
} catch (IOException e) {
|
||||
logger.error("The following directory could not be deleted: " + directory.getPath(), e);
|
||||
return false;
|
||||
} catch (IllegalArgumentException iae) {
|
||||
logger.error("This directory does not exist: " + directory.getPath());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,15 +1,14 @@
|
|||
package eu.openaire.urls_worker.controllers;
|
||||
|
||||
import eu.openaire.urls_worker.components.AssignmentsHandler;
|
||||
import eu.openaire.urls_worker.components.plugins.PublicationsRetrieverPlugin;
|
||||
import eu.openaire.urls_worker.util.UriBuilder;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.web.bind.annotation.GetMapping;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
|
||||
import javax.servlet.http.HttpServletRequest;
|
||||
import java.io.File;
|
||||
|
|
@ -24,13 +23,13 @@ public class GeneralController {
|
|||
private static final Logger logger = LoggerFactory.getLogger(GeneralController.class);
|
||||
|
||||
private final String controllerIp;
|
||||
//private final String workerReportsDirPath;
|
||||
private final String workerReportsDirPath;
|
||||
private final String workerId;
|
||||
|
||||
|
||||
private static final Pattern DOMAIN_DETECTOR = Pattern.compile("^.*[a-zA-Z].*$");
|
||||
|
||||
public GeneralController(@Value("${info.controllerIp}") String controllerIp, /*@Value("${workerReportsDirPath}") String workerReportsDirPath,*/ @Value("${info.workerId}") String workerId)
|
||||
public GeneralController(@Value("${info.controllerIp}") String controllerIp, @Value("${workerReportsDirPath}") String workerReportsDirPath, @Value("${info.workerId}") String workerId)
|
||||
{
|
||||
if ( DOMAIN_DETECTOR.matcher(controllerIp).matches() ) {
|
||||
try {
|
||||
|
|
@ -43,7 +42,7 @@ public class GeneralController {
|
|||
} else
|
||||
this.controllerIp = controllerIp;
|
||||
|
||||
//this.workerReportsDirPath = workerReportsDirPath;
|
||||
this.workerReportsDirPath = workerReportsDirPath;
|
||||
this.workerId = workerId;
|
||||
}
|
||||
|
||||
|
|
@ -99,6 +98,27 @@ public class GeneralController {
|
|||
}
|
||||
|
||||
|
||||
@PostMapping("addReportResultToWorker/{assignmentsCounter}")
|
||||
public ResponseEntity<?> addReportResultToWorker(@PathVariable long assignmentsCounter, @RequestBody(required=false) String errorMsg)
|
||||
{
|
||||
String directoryPath = PublicationsRetrieverPlugin.assignmentsBasePath + "assignments_" + assignmentsCounter + "_fullTexts";
|
||||
File dir = new File(directoryPath);
|
||||
if ( ! dir.isDirectory() ) {
|
||||
logger.error("The \"addRReportResultToWorker\"-endpoint was called for an unknown \"assignmentsCounter\": " + assignmentsCounter);
|
||||
return ResponseEntity.notFound().build();
|
||||
}
|
||||
|
||||
if ( errorMsg == null ) {
|
||||
logger.info("The Controller successfully handled the WorkerReport, for assignments_" + assignmentsCounter + ". The worker-report and all full-text files associated with it, will be deleted.");
|
||||
FullTextsController.deleteAssignmentsDirectory(assignmentsCounter, dir);
|
||||
FullTextsController.deleteFile(this.workerReportsDirPath + this.workerId + "_assignments_" + assignmentsCounter + "_report.json");
|
||||
} else
|
||||
logger.error("The Controller failed to handle the WorkerReport, for assignments_" + assignmentsCounter + ". The error is:\n" + errorMsg);
|
||||
|
||||
return ResponseEntity.ok().build();
|
||||
}
|
||||
|
||||
|
||||
public ResponseEntity<?> passSecurityChecks(HttpServletRequest request, String initMsg)
|
||||
{
|
||||
if ( request == null ) {
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package eu.openaire.urls_worker.payloads.responces;
|
|||
import com.fasterxml.jackson.annotation.JsonInclude;
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
|
||||
import com.google.gson.Gson;
|
||||
import eu.openaire.urls_worker.models.UrlReport;
|
||||
|
||||
import java.util.List;
|
||||
|
|
@ -16,6 +17,8 @@ import java.util.List;
|
|||
})
|
||||
public class WorkerReport {
|
||||
|
||||
private static final Gson gson = new Gson(); // This is "transient" be default. It won't be included in any json object.
|
||||
|
||||
@JsonProperty("workerId")
|
||||
private String workerId;
|
||||
|
||||
|
|
@ -55,6 +58,11 @@ public class WorkerReport {
|
|||
this.urlReports = urlReports;
|
||||
}
|
||||
|
||||
public String getJsonReport() {
|
||||
return gson.toJson(this);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "WorkerReport{" +
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
package eu.openaire.urls_worker.util;
|
||||
|
||||
import eu.openaire.urls_worker.controllers.FullTextsController;
|
||||
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
|
||||
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
|
||||
import org.apache.commons.compress.compressors.zstandard.ZstdCompressorOutputStream;
|
||||
|
|
@ -25,10 +26,17 @@ public class FilesCompressor {
|
|||
public static File compressMultipleFilesIntoOne(long assignmentsCounter, int tarBatchCounter, List<String> filesToCompress, String baseDirectory)
|
||||
{
|
||||
// For example: assignments_2_full-texts_4.tar.zstd | where < 4 > is referred to the 4th batch of files requested by the Controller.
|
||||
|
||||
File tarFile = getTarArchiveWithFullTexts(filesToCompress, baseDirectory, assignmentsCounter, tarBatchCounter);
|
||||
if ( tarFile == null )
|
||||
return null; // The error-cause is already logged.
|
||||
File tarFile;
|
||||
try {
|
||||
tarFile = getTarArchiveWithFullTexts(filesToCompress, baseDirectory, assignmentsCounter, tarBatchCounter);
|
||||
} catch (Exception e) {
|
||||
logger.error("Exception when creating the tar-file for assignments_" + assignmentsCounter, e);
|
||||
return null;
|
||||
} finally {
|
||||
// Delete the files of this failed batch immediately. These files will not be requested again. The urls leading to these file will be reprocessed in the future.
|
||||
for ( String fileName : filesToCompress )
|
||||
FullTextsController.deleteFile(baseDirectory + fileName);
|
||||
}
|
||||
|
||||
// The "TAR" archive is not compressed, but it helps deliver multiple full-texts with a single Stream.
|
||||
// Then, we compress the archive, using Facebook's "ZStandard" algorithm, which delivers both high compression-rate and compression and decompression efficiency.
|
||||
|
|
@ -47,6 +55,8 @@ public class FilesCompressor {
|
|||
} catch (Exception e) {
|
||||
logger.error("Exception when compressing the tar-archive: " + tarFilePath, e);
|
||||
return null;
|
||||
} finally {
|
||||
FullTextsController.deleteFile(tarFilePath);
|
||||
}
|
||||
|
||||
logger.debug("Finished archiving and compressing the full-texts of assignments_" + assignmentsCounter + ", batch_" + tarBatchCounter);
|
||||
|
|
@ -57,8 +67,8 @@ public class FilesCompressor {
|
|||
/**
|
||||
* This method adds the requested full-text file into a TAR archive, which later will be compressed.
|
||||
* */
|
||||
private static File getTarArchiveWithFullTexts(List<String> filesToTar, String baseDir, long assignmentsCounter, int tarBatchCounter) {
|
||||
|
||||
private static File getTarArchiveWithFullTexts(List<String> filesToTar, String baseDir, long assignmentsCounter, int tarBatchCounter) throws Exception
|
||||
{
|
||||
String tarFileFullPath = baseDir + "assignments_" + assignmentsCounter + "_full-texts_" + tarBatchCounter + ".tar";
|
||||
// For example: assignments_2_full-texts_4.tar.zstd | where < 4 > is referred to the 4th batch of files requested by the Controller.
|
||||
|
||||
|
|
@ -73,9 +83,6 @@ public class FilesCompressor {
|
|||
if ( addTarEntry(taos, fileName, baseDir) )
|
||||
numTarredFiles ++;
|
||||
}
|
||||
} catch (Exception e) {
|
||||
logger.error("Exception when creating the tar-file: " + tarFileFullPath, e);
|
||||
return null;
|
||||
}
|
||||
|
||||
if ( numTarredFiles != filesToTar.size() )
|
||||
|
|
|
|||
|
|
@ -29,6 +29,9 @@ info.controllerPort = XX
|
|||
info.controllerBaseUrl = http://${info.controllerIp}:${info.controllerPort}/api/
|
||||
|
||||
|
||||
workerReportsDirPath: ${HOME}/workerReports/
|
||||
|
||||
|
||||
# LOGGING LEVELS
|
||||
logging.config=classpath:logback-spring.xml
|
||||
logging.level.root=INFO
|
||||
|
|
|
|||
Loading…
Reference in New Issue