datafilecollector/src/main/java/org/oran/datafile/tasks/FileCollector.java

   1 /*-
   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
   8  *
   9  * http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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
  14  * the License.
  15  * ============LICENSE_END========================================================================
  16  */
  17
  18 package org.oran.datafile.tasks;
  19
  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;
  25
  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;
  43
  44 import reactor.core.publisher.Mono;
  45 import reactor.util.retry.Retry;
  46
  47 /**
  48  * Collects a file from a PNF.
  49  */
  50 public class FileCollector {
  51
  52     private static final Logger logger = LoggerFactory.getLogger(FileCollector.class);
  53     private final AppConfig appConfig;
  54     private final Counters counters;
  55
  56     /**
  57      * Constructor.
  58      *
  59      * @param appConfig application configuration
  60      */
  61     public FileCollector(AppConfig appConfig, Counters counters) {
  62         this.appConfig = appConfig;
  63         this.counters = counters;
  64     }
  65
  66     /**
  67      * Collects a file from the PNF and stores it in the local file system.
  68      *
  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.
  74      */
  75     public Mono<FilePublishInformation> collectFile(FileData fileData, long numRetries, Duration firstBackoff) {
  76
  77         logger.trace("Entering collectFile with {}", fileData);
  78
  79         return Mono.just(fileData) //
  80             .cache() //
  81             .flatMap(fd -> tryCollectFile(fileData)) //
  82             .retryWhen(Retry.backoff(numRetries, firstBackoff)) //
  83             .flatMap(FileCollector::checkCollectedFile);
  84     }
  85
  86     private static Mono<FilePublishInformation> checkCollectedFile(Optional<FilePublishInformation> info) {
  87         if (info.isPresent()) {
  88             return Mono.just(info.get());
  89         } else {
  90             // If there is no info, the file is not retrievable
  91             return Mono.error(new DatafileTaskException("Non retryable file transfer failure"));
  92         }
  93     }
  94
  95     private Mono<Optional<FilePublishInformation>> tryCollectFile(FileData fileData) {
  96         logger.trace("starting to collectFile {}", fileData.fileInfo.name);
  97
  98         final String remoteFile = fileData.remoteFilePath();
  99         final Path localFile = fileData.getLocalFilePath(this.appConfig);
 100
 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,
 114                 e.getMessage());
 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)));
 121         }
 122     }
 123
 124     private void incFailedAttemptsCounter(FileData fileData) {
 125         if (FileData.Scheme.isFtpScheme(fileData.scheme())) {
 126             counters.incNoOfFailedFtpAttempts();
 127         } else {
 128             counters.incNoOfFailedHttpAttempts();
 129         }
 130     }
 131
 132     private FileCollectClient createClient(FileData fileData) throws DatafileTaskException {
 133         switch (fileData.scheme()) {
 134             case SFTP:
 135                 return createSftpClient(fileData);
 136             case FTPES:
 137                 return createFtpesClient(fileData);
 138             case HTTP:
 139                 return createHttpClient(fileData);
 140             case HTTPS:
 141                 return createHttpsClient(fileData);
 142             default:
 143                 throw new DatafileTaskException("Unhandled protocol: " + fileData.scheme());
 144         }
 145     }
 146
 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) //
 162             .build();
 163     }
 164
 165     protected SftpClient createSftpClient(FileData fileData) {
 166         return new SftpClient(fileData.fileServerData(), new SftpClientSettings(appConfig.getSftpConfiguration()));
 167     }
 168
 169     protected FtpesClient createFtpesClient(FileData fileData) throws DatafileTaskException {
 170         CertificateConfig config = appConfig.getCertificateConfiguration();
 171         Path trustedCa = config.trustedCa.isEmpty() ? null : Paths.get(config.trustedCa);
 172
 173         return new FtpesClient(fileData.fileServerData(), Paths.get(config.keyCert), config.keyPasswordPath, trustedCa,
 174             config.trustedCaPasswordPath);
 175     }
 176
 177     protected FileCollectClient createHttpClient(FileData fileData) {
 178         return new DfcHttpClient(fileData.fileServerData());
 179     }
 180
 181     protected FileCollectClient createHttpsClient(FileData fileData) throws DatafileTaskException {
 182         return new DfcHttpsClient(fileData.fileServerData(), HttpsClientConnectionManagerUtil.instance());
 183     }
 184 }