2 * ============LICENSE_START======================================================================
3 * Copyright (C) 2018-2023 Nordix Foundation. All rights reserved.
4 * Copyright (C) 2020-2022 Nokia. All rights reserved.
5 * ===============================================================================================
6 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
7 * in compliance with the License. You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software distributed under the License
12 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
13 * or implied. See the License for the specific language governing permissions and limitations under
15 * ============LICENSE_END========================================================================
18 package org.oran.datafile.tasks;
20 import java.nio.file.Files;
21 import java.nio.file.Path;
22 import java.nio.file.Paths;
23 import java.time.Duration;
24 import java.util.Optional;
26 import org.oran.datafile.commons.FileCollectClient;
27 import org.oran.datafile.configuration.AppConfig;
28 import org.oran.datafile.configuration.CertificateConfig;
29 import org.oran.datafile.exceptions.DatafileTaskException;
30 import org.oran.datafile.exceptions.NonRetryableDatafileTaskException;
31 import org.oran.datafile.ftp.FtpesClient;
32 import org.oran.datafile.ftp.SftpClient;
33 import org.oran.datafile.ftp.SftpClientSettings;
34 import org.oran.datafile.http.DfcHttpClient;
35 import org.oran.datafile.http.DfcHttpsClient;
36 import org.oran.datafile.http.HttpsClientConnectionManagerUtil;
37 import org.oran.datafile.model.Counters;
38 import org.oran.datafile.model.FileData;
39 import org.oran.datafile.model.FilePublishInformation;
40 import org.oran.datafile.model.FileReadyMessage;
41 import org.slf4j.Logger;
42 import org.slf4j.LoggerFactory;
44 import reactor.core.publisher.Mono;
45 import reactor.util.retry.Retry;
48 * Collects a file from a PNF.
50 public class FileCollector {
52 private static final Logger logger = LoggerFactory.getLogger(FileCollector.class);
53 private final AppConfig appConfig;
54 private final Counters counters;
59 * @param appConfig application configuration
61 public FileCollector(AppConfig appConfig, Counters counters) {
62 this.appConfig = appConfig;
63 this.counters = counters;
67 * Collects a file from the PNF and stores it in the local file system.
69 * @param fileData data about the file to collect.
70 * @param numRetries the number of retries if the publishing fails
71 * @param firstBackoff the time to delay the first retry
72 * @param contextMap context for logging.
73 * @return the data needed to publish the file.
75 public Mono<FilePublishInformation> collectFile(FileData fileData, long numRetries, Duration firstBackoff) {
77 logger.trace("Entering collectFile with {}", fileData);
79 return Mono.just(fileData) //
81 .flatMap(fd -> tryCollectFile(fileData)) //
82 .retryWhen(Retry.backoff(numRetries, firstBackoff)) //
83 .flatMap(FileCollector::checkCollectedFile);
86 private static Mono<FilePublishInformation> checkCollectedFile(Optional<FilePublishInformation> info) {
87 if (info.isPresent()) {
88 return Mono.just(info.get());
90 // If there is no info, the file is not retrievable
91 return Mono.error(new DatafileTaskException("Non retryable file transfer failure"));
95 private Mono<Optional<FilePublishInformation>> tryCollectFile(FileData fileData) {
96 logger.trace("starting to collectFile {}", fileData.fileInfo.name);
98 final String remoteFile = fileData.remoteFilePath();
99 final Path localFile = fileData.getLocalFilePath(this.appConfig);
101 try (FileCollectClient currentClient = createClient(fileData)) {
102 currentClient.open();
103 Files.createDirectories(localFile.getParent());
104 currentClient.collectFile(remoteFile, localFile);
105 counters.incNoOfCollectedFiles();
106 return Mono.just(Optional.of(createFilePublishInformation(fileData)));
107 } catch (NonRetryableDatafileTaskException nre) {
108 logger.warn("Failed to download file, not retryable: {} {}, reason: {}", fileData.sourceName(),
109 fileData.fileInfo.name, nre.getMessage());
110 incFailedAttemptsCounter(fileData);
111 return Mono.just(Optional.empty()); // Give up
112 } catch (DatafileTaskException e) {
113 logger.warn("Failed to download file: {} {}, reason: {}", fileData.sourceName(), fileData.fileInfo.name,
115 incFailedAttemptsCounter(fileData);
116 return Mono.error(e);
117 } catch (Exception throwable) {
118 logger.warn("Failed to close client: {} {}, reason: {}", fileData.sourceName(), fileData.fileInfo.name,
119 throwable.getMessage(), throwable);
120 return Mono.just(Optional.of(createFilePublishInformation(fileData)));
124 private void incFailedAttemptsCounter(FileData fileData) {
125 if (FileData.Scheme.isFtpScheme(fileData.scheme())) {
126 counters.incNoOfFailedFtpAttempts();
128 counters.incNoOfFailedHttpAttempts();
132 private FileCollectClient createClient(FileData fileData) throws DatafileTaskException {
133 switch (fileData.scheme()) {
135 return createSftpClient(fileData);
137 return createFtpesClient(fileData);
139 return createHttpClient(fileData);
141 return createHttpsClient(fileData);
143 throw new DatafileTaskException("Unhandled protocol: " + fileData.scheme());
147 public FilePublishInformation createFilePublishInformation(FileData fileData) {
148 FileReadyMessage.MessageMetaData metaData = fileData.messageMetaData;
149 return FilePublishInformation.builder() //
150 .productName(metaData.productName()) //
151 .vendorName(metaData.vendorName()) //
152 .lastEpochMicrosec(metaData.lastEpochMicrosec) //
153 .sourceName(metaData.sourceName) //
154 .startEpochMicrosec(metaData.startEpochMicrosec) //
155 .timeZoneOffset(metaData.timeZoneOffset) //
156 .name(metaData.sourceName + "/" + fileData.fileInfo.name) //
157 .compression(fileData.fileInfo.hashMap.compression) //
158 .fileFormatType(fileData.fileInfo.hashMap.fileFormatType) //
159 .fileFormatVersion(fileData.fileInfo.hashMap.fileFormatVersion) //
160 .changeIdentifier(fileData.messageMetaData.changeIdentifier) //
161 .objectStoreBucket(this.appConfig.isS3Enabled() ? this.appConfig.getS3Bucket() : null) //
165 protected SftpClient createSftpClient(FileData fileData) {
166 return new SftpClient(fileData.fileServerData(), new SftpClientSettings(appConfig.getSftpConfiguration()));
169 protected FtpesClient createFtpesClient(FileData fileData) throws DatafileTaskException {
170 CertificateConfig config = appConfig.getCertificateConfiguration();
171 Path trustedCa = config.trustedCa.isEmpty() ? null : Paths.get(config.trustedCa);
173 return new FtpesClient(fileData.fileServerData(), Paths.get(config.keyCert), config.keyPasswordPath, trustedCa,
174 config.trustedCaPasswordPath);
177 protected FileCollectClient createHttpClient(FileData fileData) {
178 return new DfcHttpClient(fileData.fileServerData());
181 protected FileCollectClient createHttpsClient(FileData fileData) throws DatafileTaskException {
182 return new DfcHttpsClient(fileData.fileServerData(), HttpsClientConnectionManagerUtil.instance());