diff options
Diffstat (limited to 'datafile-app-server/src/main')
9 files changed, 250 insertions, 69 deletions
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/HeartbeatController.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/StatusController.java index c6d56c42..40f9d99b 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/HeartbeatController.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/controllers/StatusController.java @@ -19,6 +19,8 @@ package org.onap.dcaegen2.collectors.datafile.controllers; import static org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext.ENTRY; import static org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext.EXIT; +import org.onap.dcaegen2.collectors.datafile.model.Counters; + import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiResponse; @@ -37,28 +39,25 @@ import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; /** - * Controller to check the heartbeat of DFC. - * - * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/19/18 - * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a> + * REST Controller to check the heart beat and status of the DFC. */ @RestController -@Api(value = "HeartbeatController") -public class HeartbeatController { +@Api(value = "StatusController") +public class StatusController { - private static final Logger logger = LoggerFactory.getLogger(HeartbeatController.class); + private static final Logger logger = LoggerFactory.getLogger(StatusController.class); private final ScheduledTasks scheduledTasks; @Autowired - public HeartbeatController(ScheduledTasks scheduledTasks) { + public StatusController(ScheduledTasks scheduledTasks) { this.scheduledTasks = scheduledTasks; } /** - * Checks the heartbeat of DFC. + * Checks the heart beat of DFC. * - * @return the heartbeat status of DFC. + * @return the heart beat status of DFC. */ @GetMapping("/heartbeat") @ApiOperation(value = "Returns liveness of DATAFILE service") @@ -66,18 +65,41 @@ public class HeartbeatController { @ApiResponse(code = 200, message = "DATAFILE service is living"), @ApiResponse(code = 401, message = "You are not authorized to view the resource"), @ApiResponse(code = 403, message = "Accessing the resource you were trying to reach is forbidden"), - @ApiResponse(code = 404, message = "The resource you were trying to reach is not found") }) + @ApiResponse(code = 404, message = "The resource you were trying to reach is not found")}) public Mono<ResponseEntity<String>> heartbeat(@RequestHeader HttpHeaders headers) { MappedDiagnosticContext.initializeTraceContext(headers); logger.info(ENTRY, "Heartbeat request"); - StringBuilder statusString = new StringBuilder("I'm living! Status: "); - statusString.append("numberOfFileCollectionTasks=").append(scheduledTasks.getCurrentNumberOfTasks()) - .append(","); - statusString.append("fileCacheSize=").append(scheduledTasks.publishedFilesCacheSize()); + String statusString = "I'm living!"; - Mono<ResponseEntity<String>> response = Mono.just(new ResponseEntity<>(statusString.toString(), HttpStatus.OK)); + Mono<ResponseEntity<String>> response = Mono.just(new ResponseEntity<>(statusString, HttpStatus.OK)); logger.info(EXIT, "Heartbeat request"); return response; } + + /** + * Returns diagnostics and statistics information. It is intended for testing and trouble + * shooting. + * + * @return information. + */ + @GetMapping("/status") + @ApiOperation(value = "Returns status and statistics of DATAFILE service") + @ApiResponses(value = { // + @ApiResponse(code = 200, message = "DATAFILE service is living"), + @ApiResponse(code = 401, message = "You are not authorized to view the resource"), + @ApiResponse(code = 403, message = "Accessing the resource you were trying to reach is forbidden"), + @ApiResponse(code = 404, message = "The resource you were trying to reach is not found")}) + public Mono<ResponseEntity<String>> status(@RequestHeader HttpHeaders headers) { + MappedDiagnosticContext.initializeTraceContext(headers); + logger.info(ENTRY, "Status request"); + + Counters counters = scheduledTasks.getCounters(); + Mono<ResponseEntity<String>> response = Mono.just(new ResponseEntity<>(counters.toString(), HttpStatus.OK)); + logger.info(EXIT, "Status request"); + return response; + } + + + } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java index 38f74ed1..6aa76157 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DatafileTaskException.java @@ -21,29 +21,13 @@ package org.onap.dcaegen2.collectors.datafile.exceptions; public class DatafileTaskException extends Exception { private static final long serialVersionUID = 1L; - private final boolean isRetryable; public DatafileTaskException(String message) { super(message); - isRetryable = true; - } - - public DatafileTaskException(String message, boolean retry) { - super(message); - isRetryable = retry; } public DatafileTaskException(String message, Exception originalException) { super(message, originalException); - isRetryable = true; } - public DatafileTaskException(String message, Exception originalException, boolean retry) { - super(message, originalException); - isRetryable = retry; - } - - public boolean isRetryable() { - return this.isRetryable; - } } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/NonRetryableDatafileTaskException.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/NonRetryableDatafileTaskException.java new file mode 100644 index 00000000..a4bdd667 --- /dev/null +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/NonRetryableDatafileTaskException.java @@ -0,0 +1,34 @@ +/*- + * ============LICENSE_START====================================================================== + * Copyright (C) 2019 Nordix Foundation. All rights reserved. + * =============================================================================================== + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END======================================================================== + */ + +package org.onap.dcaegen2.collectors.datafile.exceptions; + + +public class NonRetryableDatafileTaskException extends DatafileTaskException { + + private static final long serialVersionUID = 1L; + + public NonRetryableDatafileTaskException(String message) { + super(message); + } + + public NonRetryableDatafileTaskException(String message, Exception originalException) { + super(message, originalException); + } + +} diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java index bb3016ce..18288603 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java @@ -38,6 +38,7 @@ import org.apache.commons.net.ftp.FTPReply; import org.apache.commons.net.ftp.FTPSClient; import org.apache.commons.net.util.KeyManagerUtils; import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException; +import org.onap.dcaegen2.collectors.datafile.exceptions.NonRetryableDatafileTaskException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.core.io.FileSystemResource; @@ -119,8 +120,7 @@ public class FtpsClient implements FileCollectClient { try (OutputStream output = createOutputStream(localFileName)) { logger.trace("begin to retrieve from xNF."); if (!realFtpsClient.retrieveFile(remoteFileName, output)) { - final boolean retry = false; // Skip retrying for all problems except IOException - throw new DatafileTaskException("Could not retrieve file " + remoteFileName, retry); + throw new NonRetryableDatafileTaskException("Could not retrieve file. No retry attempts will be done, file :" + remoteFileName); } } catch (IOException e) { throw new DatafileTaskException("Could not fetch file: " + e, e); diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java index ec523354..f5d10d34 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java @@ -25,6 +25,7 @@ import com.jcraft.jsch.SftpException; import java.nio.file.Path; import java.util.Optional; import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException; +import org.onap.dcaegen2.collectors.datafile.exceptions.NonRetryableDatafileTaskException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -57,7 +58,12 @@ public class SftpClient implements FileCollectClient { } catch (SftpException e) { boolean retry = e.id != ChannelSftp.SSH_FX_NO_SUCH_FILE && e.id != ChannelSftp.SSH_FX_PERMISSION_DENIED && e.id != ChannelSftp.SSH_FX_OP_UNSUPPORTED; - throw new DatafileTaskException("Unable to get file from xNF. Data: " + fileServerData, e, retry); + if (retry) { + throw new DatafileTaskException("Unable to get file from xNF. Data: " + fileServerData, e); + } else { + throw new NonRetryableDatafileTaskException( + "Unable to get file from xNF. No retry attempts will be done. Data: " + fileServerData, e); + } } logger.trace("collectFile OK"); @@ -85,7 +91,12 @@ public class SftpClient implements FileCollectClient { } } catch (JSchException e) { boolean retry = !e.getMessage().contains("Auth fail"); - throw new DatafileTaskException("Could not open Sftp client. " + e, e, retry); + if (retry) { + throw new DatafileTaskException("Could not open Sftp client. " + e); + } else { + throw new NonRetryableDatafileTaskException( + "Could not open Sftp client, no retry attempts will be done " + e); + } } } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/Counters.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/Counters.java new file mode 100644 index 00000000..5efbe37a --- /dev/null +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/Counters.java @@ -0,0 +1,108 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 Nordix Foundation. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.dcaegen2.collectors.datafile.model; + +import java.time.Instant; +import java.util.concurrent.atomic.AtomicInteger; + + +/** + * + * Various counters that can be shown via a REST API. + * + */ +public class Counters { + + private final AtomicInteger numberOfTasks = new AtomicInteger(); + private final AtomicInteger numberOfSubscriptions = new AtomicInteger(); + private int noOfCollectedFiles = 0; + private int noOfFailedFtpAttempts = 0; + private int noOfFailedFtp = 0; + private int noOfFailedPublishAttempts = 0; + private int totalPublishedFiles = 0; + private int noOfFailedPublish = 0; + private Instant lastPublishedTime = Instant.MIN; + private int totalReceivedEvents = 0; + private Instant lastEventTime = Instant.MIN; + + public AtomicInteger getCurrentNumberOfTasks() { + return numberOfTasks; + } + + public AtomicInteger getCurrentNumberOfSubscriptions() { + return numberOfSubscriptions; + } + + public synchronized void incNoOfReceivedEvents() { + totalReceivedEvents++; + lastEventTime = Instant.now(); + } + + public synchronized void incNoOfCollectedFiles() { + noOfCollectedFiles++; + } + + public synchronized void incNoOfFailedFtpAttempts() { + noOfFailedFtpAttempts++; + } + + public synchronized void incNoOfFailedFtp() { + noOfFailedFtp++; + } + + public synchronized void incNoOfFailedPublishAttempts() { + noOfFailedPublishAttempts++; + } + + public synchronized void incTotalPublishedFiles() { + totalPublishedFiles++; + lastPublishedTime = Instant.now(); + } + + public synchronized void incNoOfFailedPublish() { + noOfFailedPublish++; + } + + public synchronized String toString() { + StringBuilder str = new StringBuilder(); + str.append(format("totalReceivedEvents", totalReceivedEvents)); + str.append(format("lastEventTime", lastEventTime)); + str.append(format("numberOfTasks", numberOfTasks)); + str.append(format("numberOfSubscriptions", numberOfSubscriptions)); + str.append("\n"); + str.append(format("collectedFiles", noOfCollectedFiles)); + str.append(format("failedFtpAttempts", noOfFailedFtpAttempts)); + str.append(format("failedFtp", noOfFailedFtp)); + str.append("\n"); + str.append(format("totalPublishedFiles", totalPublishedFiles)); + str.append(format("lastPublishedTime", lastPublishedTime)); + + str.append(format("failedPublishAttempts", noOfFailedPublishAttempts)); + str.append(format("noOfFailedPublish", noOfFailedPublish)); + + return str.toString(); + } + + private String format(String name, Object value) { + String header = name + ":"; + return String.format("%-24s%-22s\n", header, value); + } +} diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java index d9efe802..02e153cf 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java @@ -22,19 +22,20 @@ package org.onap.dcaegen2.collectors.datafile.tasks; import com.google.gson.JsonElement; import com.google.gson.JsonParser; +import java.io.File; import java.io.IOException; -import java.io.InputStream; import java.net.MalformedURLException; import java.net.URI; import java.nio.file.Path; import java.time.Duration; -import org.apache.commons.io.IOUtils; import org.apache.http.HttpResponse; import org.apache.http.client.methods.HttpPut; -import org.apache.http.entity.ByteArrayEntity; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.FileEntity; import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig; import org.onap.dcaegen2.collectors.datafile.configuration.PublisherConfiguration; import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException; +import org.onap.dcaegen2.collectors.datafile.model.Counters; import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation; import org.onap.dcaegen2.collectors.datafile.model.JsonSerializer; import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext; @@ -62,9 +63,11 @@ public class DataRouterPublisher { private static final Logger logger = LoggerFactory.getLogger(DataRouterPublisher.class); private final AppConfig datafileAppConfig; + private final Counters counters; - public DataRouterPublisher(AppConfig datafileAppConfig) { + public DataRouterPublisher(AppConfig datafileAppConfig, Counters counters) { this.datafileAppConfig = datafileAppConfig; + this.counters = counters; } /** @@ -98,8 +101,10 @@ public class DataRouterPublisher { HttpResponse response = dmaapProducerHttpClient.getDmaapProducerResponseWithRedirect(put, publishInfo.getContext()); logger.trace("{}", response); + counters.incTotalPublishedFiles(); return Mono.just(HttpStatus.valueOf(response.getStatusLine().getStatusCode())); } catch (Exception e) { + counters.incNoOfFailedPublishAttempts(); logger.warn("Publishing file {} to DR unsuccessful.", publishInfo.getName(), e); return Mono.error(e); } @@ -121,10 +126,9 @@ public class DataRouterPublisher { } private void prepareBody(FilePublishInformation publishInfo, HttpPut put) throws IOException { - Path fileLocation = publishInfo.getInternalLocation(); - try (InputStream fileInputStream = createInputStream(fileLocation)) { - put.setEntity(new ByteArrayEntity(IOUtils.toByteArray(fileInputStream))); - } + File file = createInputFile(publishInfo.getInternalLocation()); + FileEntity entity = new FileEntity(file, ContentType.DEFAULT_BINARY); + put.setEntity(entity); } private static Mono<FilePublishInformation> handleHttpResponse(HttpStatus response, @@ -140,9 +144,9 @@ public class DataRouterPublisher { } } - InputStream createInputStream(Path filePath) throws IOException { + File createInputFile(Path filePath) throws IOException { FileSystemResource realResource = new FileSystemResource(filePath); - return realResource.getInputStream(); + return realResource.getFile(); } PublisherConfiguration resolveConfiguration(String changeIdentifer) throws DatafileTaskException { diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java index aeacaffc..311f752d 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java @@ -24,9 +24,11 @@ import java.util.Optional; import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig; import org.onap.dcaegen2.collectors.datafile.configuration.FtpesConfig; import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException; +import org.onap.dcaegen2.collectors.datafile.exceptions.NonRetryableDatafileTaskException; import org.onap.dcaegen2.collectors.datafile.ftp.FileCollectClient; import org.onap.dcaegen2.collectors.datafile.ftp.FtpsClient; import org.onap.dcaegen2.collectors.datafile.ftp.SftpClient; +import org.onap.dcaegen2.collectors.datafile.model.Counters; import org.onap.dcaegen2.collectors.datafile.model.FileData; import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation; import org.onap.dcaegen2.collectors.datafile.model.ImmutableFilePublishInformation; @@ -45,14 +47,16 @@ public class FileCollector { private static final Logger logger = LoggerFactory.getLogger(FileCollector.class); private final AppConfig datafileAppConfig; + private final Counters counters; /** * Constructor. * * @param datafileAppConfig application configuration */ - public FileCollector(AppConfig datafileAppConfig) { + public FileCollector(AppConfig datafileAppConfig, Counters counters) { this.datafileAppConfig = datafileAppConfig; + this.counters = counters; } /** @@ -97,14 +101,16 @@ public class FileCollector { currentClient.open(); localFile.getParent().toFile().mkdir(); // Create parent directories currentClient.collectFile(remoteFile, localFile); + counters.incNoOfCollectedFiles(); return Mono.just(Optional.of(getFilePublishInformation(fileData, localFile, context))); } catch (DatafileTaskException e) { logger.warn("Failed to download file: {} {}, reason: {}", fileData.sourceName(), fileData.name(), e.toString()); - if (e.isRetryable()) { - return Mono.error(e); - } else { + counters.incNoOfFailedFtpAttempts(); + if (e instanceof NonRetryableDatafileTaskException) { return Mono.just(Optional.empty()); // Give up + } else { + return Mono.error(e); } } catch (Exception throwable) { logger.warn("Failed to close ftp client: {} {}, reason: {}", fileData.sourceName(), fileData.name(), diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java index 99b2d918..300ca601 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig; import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException; +import org.onap.dcaegen2.collectors.datafile.model.Counters; import org.onap.dcaegen2.collectors.datafile.model.FileData; import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation; import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage; @@ -41,8 +42,8 @@ import reactor.core.scheduler.Schedulers; /** - * This implements the main flow of the data file collector. Fetch file ready events from the message router, fetch new - * files from the PNF publish these in the data router. + * This implements the main flow of the data file collector. Fetch file ready events from the + * message router, fetch new files from the PNF publish these in the data router. */ @Component public class ScheduledTasks { @@ -57,11 +58,12 @@ public class ScheduledTasks { private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class); private final AppConfig applicationConfiguration; - private final AtomicInteger currentNumberOfTasks = new AtomicInteger(); + private final AtomicInteger currentNumberOfTasks; private final AtomicInteger threadPoolQueueSize = new AtomicInteger(); - private final AtomicInteger currentNumberOfSubscriptions = new AtomicInteger(); + private final AtomicInteger currentNumberOfSubscriptions; private final Scheduler scheduler = Schedulers.newParallel("FileCollectorWorker", NUMBER_OF_WORKER_THREADS); PublishedFileCache publishedFilesCache = new PublishedFileCache(); + private Counters counters = new Counters(); /** * Constructor for task registration in Datafile Workflow. @@ -71,6 +73,8 @@ public class ScheduledTasks { @Autowired public ScheduledTasks(AppConfig applicationConfiguration) { this.applicationConfiguration = applicationConfiguration; + this.currentNumberOfTasks = counters.getCurrentNumberOfTasks(); + this.currentNumberOfSubscriptions = counters.getCurrentNumberOfSubscriptions(); } /** @@ -112,6 +116,7 @@ public class ScheduledTasks { Flux<FilePublishInformation> createMainTask(Map<String, String> context) { return fetchMoreFileReadyMessages() // .doOnNext(fileReadyMessage -> threadPoolQueueSize.incrementAndGet()) // + .doOnNext(fileReadyMessage -> counters.incNoOfReceivedEvents()) // .parallel(NUMBER_OF_WORKER_THREADS) // Each FileReadyMessage in a separate thread .runOn(scheduler) // .doOnNext(fileReadyMessage -> threadPoolQueueSize.decrementAndGet()) // @@ -148,32 +153,21 @@ public class ScheduledTasks { return new PublishedChecker(applicationConfiguration); } - public int getCurrentNumberOfTasks() { - return currentNumberOfTasks.get(); + public Counters getCounters() { + return this.counters; } - public int publishedFilesCacheSize() { - return publishedFilesCache.size(); - } - - public int getCurrentNumberOfSubscriptions() { - return currentNumberOfSubscriptions.get(); - } - - public int getThreadPoolQueueSize() { - return this.threadPoolQueueSize.get(); - } protected DMaaPMessageConsumer createConsumerTask() throws DatafileTaskException { return new DMaaPMessageConsumer(this.applicationConfiguration); } protected FileCollector createFileCollector() { - return new FileCollector(applicationConfiguration); + return new FileCollector(applicationConfiguration, counters); } protected DataRouterPublisher createDataRouterPublisher() { - return new DataRouterPublisher(applicationConfiguration); + return new DataRouterPublisher(applicationConfiguration, counters); } private static void onComplete(Map<String, String> contextMap) { @@ -181,6 +175,22 @@ public class ScheduledTasks { logger.trace("Datafile tasks have been completed"); } + int publishedFilesCacheSize() { + return publishedFilesCache.size(); + } + + int getCurrentNumberOfTasks() { + return this.currentNumberOfTasks.get(); + } + + int getCurrentNumberOfSubscriptions() { + return this.currentNumberOfSubscriptions.get(); + } + + int getThreadPoolQueueSize() { + return this.threadPoolQueueSize.get(); + } + private static synchronized void onSuccess(FilePublishInformation publishInfo) { MDC.setContextMap(publishInfo.getContext()); logger.info("Datafile file published {}", publishInfo.getInternalLocation()); @@ -239,6 +249,7 @@ public class ScheduledTasks { deleteFile(localFilePath, fileData.context); publishedFilesCache.remove(localFilePath); currentNumberOfTasks.decrementAndGet(); + counters.incNoOfFailedFtp(); return Mono.empty(); } @@ -257,6 +268,7 @@ public class ScheduledTasks { deleteFile(internalFileName, publishInfo.getContext()); publishedFilesCache.remove(internalFileName); currentNumberOfTasks.decrementAndGet(); + counters.incNoOfFailedPublish(); return Mono.empty(); } |