Previously files was read into a buffer for publishing.
This does not work when files are bigger than the available memory.
After the fix , files are streamed instead.
Implemented a new REST primitive for exposing status and
statistics. To be used for test and trouble shooting.
Change-Id: Iab5a1ee9ffcbf6836fcf709d115bf25ab0391732
Issue-ID: DCAEGEN2-1532
Signed-off-by: PatrikBuhr <patrik.buhr@est.tech>
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;
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")
@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;
+ }
+
+
+
}
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;
- }
}
--- /dev/null
+/*-
+ * ============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);
+ }
+
+}
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;
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);
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;
} 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");
}
} 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);
+ }
}
}
--- /dev/null
+/*-
+ * ============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);
+ }
+}
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;
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;
}
/**
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);
}
}
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,
}
}
- 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 {
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;
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;
}
/**
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(),
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;
/**
- * 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 {
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.
@Autowired
public ScheduledTasks(AppConfig applicationConfiguration) {
this.applicationConfiguration = applicationConfiguration;
+ this.currentNumberOfTasks = counters.getCurrentNumberOfTasks();
+ this.currentNumberOfSubscriptions = counters.getCurrentNumberOfSubscriptions();
}
/**
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()) //
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) {
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());
deleteFile(localFilePath, fileData.context);
publishedFilesCache.remove(localFilePath);
currentNumberOfTasks.decrementAndGet();
+ counters.incNoOfFailedFtp();
return Mono.empty();
}
deleteFile(internalFileName, publishInfo.getContext());
publishedFilesCache.remove(internalFileName);
currentNumberOfTasks.decrementAndGet();
+ counters.incNoOfFailedPublish();
return Mono.empty();
}
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import ch.qos.logback.core.read.ListAppender;
import org.apache.commons.lang3.StringUtils;
import org.junit.jupiter.api.Test;
-import org.onap.dcaegen2.collectors.datafile.controllers.HeartbeatController;
+import org.onap.dcaegen2.collectors.datafile.controllers.StatusController;
+import org.onap.dcaegen2.collectors.datafile.model.Counters;
import org.onap.dcaegen2.collectors.datafile.tasks.ScheduledTasks;
import org.onap.dcaegen2.collectors.datafile.utils.LoggingUtils;
import org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables;
import org.springframework.http.ResponseEntity;
import reactor.core.publisher.Mono;
-public class HeartbeatControllerTest {
+public class StatusControllerTest {
@Test
public void heartbeat_success() {
ScheduledTasks scheduledTasksMock = mock(ScheduledTasks.class);
- when(scheduledTasksMock.getCurrentNumberOfTasks()).thenReturn(10);
- when(scheduledTasksMock.publishedFilesCacheSize()).thenReturn(20);
HttpHeaders httpHeaders = new HttpHeaders();
- HeartbeatController controllerUnderTest = new HeartbeatController(scheduledTasksMock);
+ StatusController controllerUnderTest = new StatusController(scheduledTasksMock);
- ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(HeartbeatController.class);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(StatusController.class);
Mono<ResponseEntity<String>> result = controllerUnderTest.heartbeat(httpHeaders);
validateLogging(logAppender);
String body = result.block().getBody();
- assertTrue(body.startsWith("I'm living! Status: "));
- assertTrue(body.contains("numberOfFileCollectionTasks=10"));
- assertTrue(body.contains("fileCacheSize=20"));
+ assertTrue(body.startsWith("I'm living!"));
assertFalse(StringUtils.isBlank(MDC.get(MdcVariables.REQUEST_ID)));
assertFalse(StringUtils.isBlank(MDC.get(MdcVariables.INVOCATION_ID)));
}
+
+ @Test
+ public void status() {
+ ScheduledTasks scheduledTasksMock = mock(ScheduledTasks.class);
+ Counters counters = new Counters();
+ doReturn(counters).when(scheduledTasksMock).getCounters();
+
+ HttpHeaders httpHeaders = new HttpHeaders();
+
+ StatusController controllerUnderTest = new StatusController(scheduledTasksMock);
+
+ Mono<ResponseEntity<String>> result = controllerUnderTest.status(httpHeaders);
+
+ String body = result.block().getBody();
+ System.out.println(body);
+
+ assertFalse(StringUtils.isBlank(MDC.get(MdcVariables.REQUEST_ID)));
+ assertFalse(StringUtils.isBlank(MDC.get(MdcVariables.INVOCATION_ID)));
+ }
+
+
private void validateLogging(ListAppender<ILoggingEvent> logAppender) {
assertEquals(logAppender.list.get(0).getMarker().getName(), "ENTRY");
assertNotNull(logAppender.list.get(0).getMDCPropertyMap().get("InvocationID"));
doReturn(false).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock);
assertThatThrownBy(() -> clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH))
- .hasMessage("Could not retrieve file /dir/sample.txt");
+ .hasMessageContaining(REMOTE_FILE_PATH)
+ .hasMessageContaining("No retry");
verifyFtpsClientMock_openOk();
verify(ftpsClientMock, times(1)).retrieveFile(ArgumentMatchers.eq(REMOTE_FILE_PATH), any());
doReturn(jschMock).when(sftpClientSpy).createJsch();
when(jschMock.getSession(anyString(), anyString(), anyInt())).thenThrow(new JSchException("Failed"));
- assertThatThrownBy(() -> sftpClientSpy.open())
- .hasMessageStartingWith("Could not open Sftp client. com.jcraft.jsch.JSchException: Failed");
+ assertThatThrownBy(() -> sftpClientSpy.open()).hasMessageStartingWith("Could not open Sftp client.");
}
@SuppressWarnings("resource")
assertThatThrownBy(() -> sftpClient.collectFile("remoteFile", Paths.get("localFile")))
.isInstanceOf(DatafileTaskException.class)
- .hasMessageStartingWith("Unable to get file from xNF. Data: FileServerData{serverAddress=" + HOST
- + ", " + "userId=" + USERNAME + ", password=####, port=" + SFTP_PORT);
+ .hasMessageStartingWith("Unable to get file from xNF. No retry attempts will be done")
+ .hasMessageContaining("Data: FileServerData{serverAddress=" + HOST + ", " + "userId=" + USERNAME
+ + ", password=####, port=" + SFTP_PORT);
}
}
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
+import java.io.File;
+
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
-import java.io.ByteArrayInputStream;
-import java.io.InputStream;
import java.net.URI;
import java.nio.file.Path;
import java.nio.file.Paths;
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.ImmutableFilePublishInformation;
import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerHttpClient;
private static final String APPLICATION_OCTET_STREAM_CONTENT_TYPE = "application/octet-stream";
private static final String PUBLISH_TOPIC = "publish";
private static final String FEED_ID = "1";
- private static final String FILE_CONTENT = "Just a string.";
private static FilePublishInformation filePublishInformation;
private static DmaapProducerHttpClient httpClientMock;
.changeIdentifier(CHANGE_IDENTIFIER) //
.build(); //
appConfig = mock(AppConfig.class);
- publisherTaskUnderTestSpy = spy(new DataRouterPublisher(appConfig));
+ publisherTaskUnderTestSpy = spy(new DataRouterPublisher(appConfig, new Counters()));
}
@Test
when(httpResponseMock.getStatusLine()).thenReturn(statusLineMock);
when(statusLineMock.getStatusCode()).thenReturn(firstResponse, nextHttpResponses);
- InputStream fileStream = new ByteArrayInputStream(FILE_CONTENT.getBytes());
- doReturn(fileStream).when(publisherTaskUnderTestSpy).createInputStream(Paths.get("target", PM_FILE_NAME));
+ File file = File.createTempFile("DFC", "tmp");
+ doReturn(file).when(publisherTaskUnderTestSpy).createInputFile(Paths.get("target", PM_FILE_NAME));
}
private Map<String, String> getMetaDataAsMap(Header[] metaHeaders) {
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.FtpsClient;
import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
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.ImmutableFileData;
private SftpClient sftpClientMock = mock(SftpClient.class);
private final Map<String, String> contextMap = new HashMap<>();
+ private final Counters counters = new Counters();
private MessageMetaData createMessageMetaData() {
return ImmutableMessageMetaData.builder() //
.compression(GZIP_COMPRESSION) //
.fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
.fileFormatVersion(FILE_FORMAT_VERSION) //
- .context(new HashMap<String,String>()) //
+ .context(new HashMap<String, String>()) //
.changeIdentifier(CHANGE_IDENTIFIER) //
.build();
}
@Test
public void whenFtpesFile_returnCorrectResponse() throws Exception {
- FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+ FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock, counters));
doReturn(ftpsClientMock).when(collectorUndetTest).createFtpsClient(any());
FileData fileData = createFileData(FTPES_LOCATION_NO_PORT, Scheme.FTPS);
@Test
public void whenSftpFile_returnCorrectResponse() throws Exception {
- FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+ FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock, counters));
doReturn(sftpClientMock).when(collectorUndetTest).createSftpClient(any());
@Test
public void whenFtpesFileAlwaysFail_retryAndFail() throws Exception {
- FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+ FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock, counters));
doReturn(ftpsClientMock).when(collectorUndetTest).createFtpsClient(any());
FileData fileData = createFileData(FTPES_LOCATION, Scheme.FTPS);
@Test
public void whenFtpesFileAlwaysFail_failWithoutRetry() throws Exception {
- FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+ FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock, new Counters()));
doReturn(ftpsClientMock).when(collectorUndetTest).createFtpsClient(any());
- final boolean retry = false;
FileData fileData = createFileData(FTPES_LOCATION, Scheme.FTPS);
- doThrow(new DatafileTaskException("Unable to collect file.", retry)).when(ftpsClientMock)
+ doThrow(new NonRetryableDatafileTaskException("Unable to collect file.")).when(ftpsClientMock)
.collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
StepVerifier.create(collectorUndetTest.collectFile(fileData, 3, Duration.ofSeconds(0), contextMap))
@Test
public void whenFtpesFileFailOnce_retryAndReturnCorrectResponse() throws Exception {
- FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock));
+ FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock, counters));
doReturn(ftpsClientMock).when(collectorUndetTest).createFtpsClient(any());
doThrow(new DatafileTaskException("Unable to collect file.")).doNothing().when(ftpsClientMock)
.collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION);
Code Review / dcaegen2 / collectors / datafile.git / commitdiff