2021-01-28 09:51:17 +01:00
|
|
|
|
2021-02-12 12:31:02 +01:00
|
|
|
package eu.dnetlib.dhp.collection;
|
2021-01-28 09:51:17 +01:00
|
|
|
|
2021-02-08 12:19:38 +01:00
|
|
|
import static eu.dnetlib.dhp.common.Constants.*;
|
2021-02-06 20:12:00 +01:00
|
|
|
import static eu.dnetlib.dhp.utils.DHPUtils.*;
|
2021-02-02 12:28:21 +01:00
|
|
|
|
2021-02-04 14:06:02 +01:00
|
|
|
import java.io.IOException;
|
2021-02-05 19:18:05 +01:00
|
|
|
import java.util.Optional;
|
2021-02-04 14:06:02 +01:00
|
|
|
|
|
|
|
import org.apache.commons.cli.ParseException;
|
2021-01-28 09:51:17 +01:00
|
|
|
import org.apache.commons.io.IOUtils;
|
2021-02-05 19:18:05 +01:00
|
|
|
import org.apache.hadoop.fs.FileSystem;
|
2021-01-28 09:51:17 +01:00
|
|
|
import org.slf4j.Logger;
|
|
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
|
2021-02-17 10:28:01 +01:00
|
|
|
import eu.dnetlib.dhp.aggregation.common.AggregatorReport;
|
2021-01-28 09:51:17 +01:00
|
|
|
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
|
2021-02-05 19:18:05 +01:00
|
|
|
import eu.dnetlib.dhp.message.MessageSender;
|
2021-05-10 14:32:05 +02:00
|
|
|
import eu.dnetlib.dhp.schema.mdstore.MDStoreVersion;
|
2021-01-28 09:51:17 +01:00
|
|
|
|
|
|
|
/**
|
2021-02-03 17:58:22 +01:00
|
|
|
* CollectorWorkerApplication is the main class responsible to start the metadata collection process, storing the outcomes
|
|
|
|
* into HDFS. This application will be executed on the hadoop cluster, where invoked in the context of the metadata collection
|
|
|
|
* oozie workflow, it will receive all the input parameters necessary to instantiate the specific collection plugin and the
|
|
|
|
* relative specific configurations
|
2021-01-28 09:51:17 +01:00
|
|
|
*
|
2021-02-03 17:58:22 +01:00
|
|
|
* @author Sandro La Bruzzo, Claudio Atzori
|
2021-01-28 09:51:17 +01:00
|
|
|
*/
|
|
|
|
public class CollectorWorkerApplication {
|
|
|
|
|
|
|
|
private static final Logger log = LoggerFactory.getLogger(CollectorWorkerApplication.class);
|
|
|
|
|
2021-05-14 10:58:12 +02:00
|
|
|
private final FileSystem fileSystem;
|
2021-02-12 12:31:02 +01:00
|
|
|
|
|
|
|
public CollectorWorkerApplication(FileSystem fileSystem) {
|
|
|
|
this.fileSystem = fileSystem;
|
|
|
|
}
|
|
|
|
|
2021-01-28 09:51:17 +01:00
|
|
|
/**
|
|
|
|
* @param args
|
|
|
|
*/
|
2021-02-05 19:18:05 +01:00
|
|
|
public static void main(final String[] args)
|
|
|
|
throws ParseException, IOException, UnknownCollectorPluginException, CollectorException {
|
2021-01-28 09:51:17 +01:00
|
|
|
|
|
|
|
final ArgumentApplicationParser argumentParser = new ArgumentApplicationParser(
|
|
|
|
IOUtils
|
|
|
|
.toString(
|
2021-02-08 18:07:23 +01:00
|
|
|
CollectorWorkerApplication.class
|
2021-01-28 09:51:17 +01:00
|
|
|
.getResourceAsStream(
|
2021-02-05 19:18:05 +01:00
|
|
|
"/eu/dnetlib/dhp/collection/collector_worker_input_parameter.json")));
|
2021-01-28 09:51:17 +01:00
|
|
|
argumentParser.parseArgument(args);
|
|
|
|
|
2021-02-08 18:07:23 +01:00
|
|
|
log.info("Java Xmx: {}m", Runtime.getRuntime().maxMemory() / (1024 * 1024));
|
|
|
|
|
2021-01-28 09:51:17 +01:00
|
|
|
final String hdfsuri = argumentParser.get("namenode");
|
|
|
|
log.info("hdfsURI is {}", hdfsuri);
|
2021-01-29 16:42:41 +01:00
|
|
|
|
2021-01-28 09:51:17 +01:00
|
|
|
final String apiDescriptor = argumentParser.get("apidescriptor");
|
2021-01-29 16:42:41 +01:00
|
|
|
log.info("apiDescriptor is {}", apiDescriptor);
|
|
|
|
|
|
|
|
final String mdStoreVersion = argumentParser.get("mdStoreVersion");
|
|
|
|
log.info("mdStoreVersion is {}", mdStoreVersion);
|
2021-01-28 09:51:17 +01:00
|
|
|
|
2021-02-08 12:19:38 +01:00
|
|
|
final String dnetMessageManagerURL = argumentParser.get(DNET_MESSAGE_MGR_URL);
|
2021-02-04 15:51:15 +01:00
|
|
|
log.info("dnetMessageManagerURL is {}", dnetMessageManagerURL);
|
|
|
|
|
|
|
|
final String workflowId = argumentParser.get("workflowId");
|
|
|
|
log.info("workflowId is {}", workflowId);
|
|
|
|
|
2021-02-12 12:31:02 +01:00
|
|
|
final HttpClientParams clientParams = getClientParams(argumentParser);
|
|
|
|
|
|
|
|
final ApiDescriptor api = MAPPER.readValue(apiDescriptor, ApiDescriptor.class);
|
|
|
|
final FileSystem fileSystem = FileSystem.get(getHadoopConfiguration(hdfsuri));
|
|
|
|
|
|
|
|
new CollectorWorkerApplication(fileSystem)
|
|
|
|
.run(mdStoreVersion, clientParams, api, dnetMessageManagerURL, workflowId);
|
|
|
|
}
|
|
|
|
|
|
|
|
protected void run(String mdStoreVersion, HttpClientParams clientParams, ApiDescriptor api,
|
2021-02-15 15:08:59 +01:00
|
|
|
String dnetMessageManagerURL, String workflowId)
|
|
|
|
throws IOException, CollectorException, UnknownCollectorPluginException {
|
2021-02-12 12:31:02 +01:00
|
|
|
|
2021-02-03 12:33:41 +01:00
|
|
|
final MDStoreVersion currentVersion = MAPPER.readValue(mdStoreVersion, MDStoreVersion.class);
|
2021-02-17 10:28:01 +01:00
|
|
|
final MessageSender ms = new MessageSender(dnetMessageManagerURL, workflowId);
|
2021-02-03 12:33:41 +01:00
|
|
|
|
2021-02-17 10:28:01 +01:00
|
|
|
try (AggregatorReport report = new AggregatorReport(ms)) {
|
2021-02-15 15:08:59 +01:00
|
|
|
new CollectorWorker(api, fileSystem, currentVersion, clientParams, report).collect();
|
2021-02-05 19:18:05 +01:00
|
|
|
}
|
2021-01-28 09:51:17 +01:00
|
|
|
}
|
2021-01-29 16:42:41 +01:00
|
|
|
|
2021-02-05 19:18:05 +01:00
|
|
|
private static HttpClientParams getClientParams(ArgumentApplicationParser argumentParser) {
|
|
|
|
final HttpClientParams clientParams = new HttpClientParams();
|
|
|
|
clientParams
|
|
|
|
.setMaxNumberOfRetry(
|
|
|
|
Optional
|
2021-02-08 12:19:38 +01:00
|
|
|
.ofNullable(argumentParser.get(MAX_NUMBER_OF_RETRY))
|
2021-02-05 19:18:05 +01:00
|
|
|
.map(Integer::parseInt)
|
|
|
|
.orElse(HttpClientParams._maxNumberOfRetry));
|
|
|
|
log.info("maxNumberOfRetry is {}", clientParams.getMaxNumberOfRetry());
|
|
|
|
|
2021-02-08 12:19:38 +01:00
|
|
|
clientParams
|
|
|
|
.setRequestDelay(
|
|
|
|
Optional
|
|
|
|
.ofNullable(argumentParser.get(REQUEST_DELAY))
|
|
|
|
.map(Integer::parseInt)
|
|
|
|
.orElse(HttpClientParams._requestDelay));
|
|
|
|
log.info("requestDelay is {}", clientParams.getRequestDelay());
|
|
|
|
|
2021-02-05 19:18:05 +01:00
|
|
|
clientParams
|
|
|
|
.setRetryDelay(
|
|
|
|
Optional
|
2021-02-08 12:19:38 +01:00
|
|
|
.ofNullable(argumentParser.get(RETRY_DELAY))
|
2021-02-05 19:18:05 +01:00
|
|
|
.map(Integer::parseInt)
|
|
|
|
.orElse(HttpClientParams._retryDelay));
|
|
|
|
log.info("retryDelay is {}", clientParams.getRetryDelay());
|
|
|
|
|
|
|
|
clientParams
|
|
|
|
.setConnectTimeOut(
|
|
|
|
Optional
|
2021-02-08 12:19:38 +01:00
|
|
|
.ofNullable(argumentParser.get(CONNECT_TIMEOUT))
|
2021-02-05 19:18:05 +01:00
|
|
|
.map(Integer::parseInt)
|
|
|
|
.orElse(HttpClientParams._connectTimeOut));
|
|
|
|
log.info("connectTimeOut is {}", clientParams.getConnectTimeOut());
|
|
|
|
|
|
|
|
clientParams
|
|
|
|
.setReadTimeOut(
|
|
|
|
Optional
|
2021-02-08 12:19:38 +01:00
|
|
|
.ofNullable(argumentParser.get(READ_TIMEOUT))
|
2021-02-05 19:18:05 +01:00
|
|
|
.map(Integer::parseInt)
|
|
|
|
.orElse(HttpClientParams._readTimeOut));
|
|
|
|
log.info("readTimeOut is {}", clientParams.getReadTimeOut());
|
|
|
|
return clientParams;
|
|
|
|
}
|
2021-02-04 15:51:15 +01:00
|
|
|
|
2021-01-28 09:51:17 +01:00
|
|
|
}
|