diff options
Diffstat (limited to 'datarouter-node')
4 files changed, 100 insertions, 85 deletions
diff --git a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/Delivery.java b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/Delivery.java index e4013240..150d2aa2 100644 --- a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/Delivery.java +++ b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/Delivery.java @@ -134,7 +134,7 @@ public class Delivery { cur = spoolfile.getUsableSpace(); if (cur >= stop) { logger.info("NODE0503 Free disk space at or above yellow threshold. current=" + cur + YELLOW + stop - + TOTAL + tspace); + + TOTAL + tspace); return; } logger.warn( @@ -195,7 +195,7 @@ public class Delivery { cleardirs(); while (curthreads < threads) { curthreads++; - (new Thread("Delivery Thread") { + (new Thread("del-thread-" + curthreads) { @Override public void run() { dodelivery(); @@ -249,7 +249,7 @@ public class Delivery { for (DelItem item : items) { long amount = dqs.get(item.getSpool()).cancelTask(item.getPublishId()); logger.info("NODE0502 Attempting to discard " + item.getSpool() + "/" + item.getPublishId() - + " to free up disk"); + + " to free up disk"); if (amount > 0) { cur += amount; if (cur >= stop) { @@ -302,7 +302,7 @@ public class Delivery { } DelItem delItem = (DelItem) object; return Objects.equals(pubid, delItem.pubid) - && Objects.equals(getSpool(), delItem.getSpool()); + && Objects.equals(getSpool(), delItem.getSpool()); } @Override diff --git a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java index 8cdafd6f..0ba9ecfd 100644 --- a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java +++ b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java @@ -26,41 +26,47 @@ package org.onap.dmaap.datarouter.node; import com.att.eelf.configuration.EELFLogger; import com.att.eelf.configuration.EELFManager; -import java.io.File; -import java.util.Arrays; -import java.util.Hashtable; -import java.util.Vector; +import java.io.*; +import java.util.*; import org.jetbrains.annotations.Nullable; /** * Mechanism for monitoring and controlling delivery of files to a destination. - * - * <p>The DeliveryQueue class maintains lists of DeliveryTasks for a single destination (a subscription or another data - * router node) and assigns delivery threads to try to deliver them. It also maintains a delivery status that causes it - * to back off on delivery attempts after a failure. - * - * <p>If the most recent delivery result was a failure, then no more attempts will be made for a period of time. - * Initially, and on the first failure following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer() - * (milliseconds). If, after this delay, additional failures occur, each failure will multiply the delay by - * DeliveryQueueHelper.getFailureBackoff() up to a maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer(). - * Note that this behavior applies to the delivery queue as a whole and not to individual files in the queue. If - * multiple files are being delivered and one fails, the delay will be started. If a second delivery fails while the - * delay was active, it will not change the delay or change the duration of any subsequent delay. If, however, it - * succeeds, it will cancel the delay. - * - * <p>The queue maintains 3 collections of files to deliver: A to do list of files that will be attempted, a working - * set of files that are being attempted, and a retry set of files that were attempted and failed. Whenever the to do - * list is empty and needs to be refilled, a scan of the spool directory is made and the file names sorted. Any files - * in the working set are ignored. If a DeliveryTask for the file is in the retry set, then that delivery task is placed - * on the to do list. Otherwise, a new DeliveryTask for the file is created and placed on the to do list. If, when a - * DeliveryTask is about to be removed from the to do list, its age exceeds DeliveryQueueHelper.getExpirationTimer(), - * then it is instead marked as expired. - * - * <p>A delivery queue also maintains a skip flag. This flag is true if the failure timer is active or if no files are - * found in a directory scan. + * <p> + * The DeliveryQueue class maintains lists of DeliveryTasks for a single + * destination (a subscription or another data router node) and assigns + * delivery threads to try to deliver them. It also maintains a delivery + * status that causes it to back off on delivery attempts after a failure. + * <p> + * If the most recent delivery result was a failure, then no more attempts + * will be made for a period of time. Initially, and on the first failure + * following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer() (milliseconds). + * If, after this delay, additional failures occur, each failure will + * multiply the delay by DeliveryQueueHelper.getFailureBackoff() up to a + * maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer(). + * Note that this behavior applies to the delivery queue as a whole and not + * to individual files in the queue. If multiple files are being + * delivered and one fails, the delay will be started. If a second + * delivery fails while the delay was active, it will not change the delay + * or change the duration of any subsequent delay. + * If, however, it succeeds, it will cancel the delay. + * <p> + * The queue maintains 3 collections of files to deliver: A todo list of + * files that will be attempted, a working set of files that are being + * attempted, and a retry set of files that were attempted and failed. + * Whenever the todo list is empty and needs to be refilled, a scan of the + * spool directory is made and the file names sorted. Any files in the working set are ignored. + * If a DeliveryTask for the file is in the retry set, then that delivery + * task is placed on the todo list. Otherwise, a new DeliveryTask for the + * file is created and placed on the todo list. + * If, when a DeliveryTask is about to be removed from the todo list, its + * age exceeds DeliveryQueueHelper.getExpirationTimer(), then it is instead + * marked as expired. + * <p> + * A delivery queue also maintains a skip flag. This flag is true if the + * failure timer is active or if no files are found in a directory scan. */ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { - private static EELFLogger logger = EELFManager.getInstance().getLogger(DeliveryQueue.class); private DeliveryQueueHelper deliveryQueueHelper; private DestInfo destinationInfo; @@ -74,16 +80,6 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { private Vector<DeliveryTask> todo = new Vector<>(); /** - * Create a delivery queue for a given destination info. - */ - DeliveryQueue(DeliveryQueueHelper deliveryQueueHelper, DestInfo destinationInfo) { - this.deliveryQueueHelper = deliveryQueueHelper; - this.destinationInfo = destinationInfo; - dir = new File(destinationInfo.getSpool()); - dir.mkdirs(); - } - - /** * Try to cancel a delivery task. * * @return The length of the task in bytes or 0 if the task cannot be cancelled. @@ -111,8 +107,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { if (dt.isCleaned()) { return (0); } - StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(), - dt.getLength(), "diskFull", dt.getAttempts()); + StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(), dt.getLength(), "diskFull", dt.getAttempts()); dt.clean(); return (dt.getLength()); } @@ -122,6 +117,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { */ private synchronized void markSuccess(DeliveryTask task) { working.remove(task.getPublishId()); + logger.debug(task.getPublishId() + " marked as success."); task.clean(); failed = false; failduration = 0; @@ -131,6 +127,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { * Mark that a delivery task has expired. */ private synchronized void markExpired(DeliveryTask task) { + logger.debug(task.getPublishId() + " marked as expired."); task.clean(); } @@ -139,6 +136,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { */ private synchronized void markFailNoRetry(DeliveryTask task) { working.remove(task.getPublishId()); + logger.debug(task.getPublishId() + " marked as failed permanently"); task.clean(); failed = false; failduration = 0; @@ -150,7 +148,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { if (failduration == 0) { if (destinationInfo.isPrivilegedSubscriber()) { failduration = deliveryQueueHelper.getWaitForFileProcessFailureTimer(); - } else { + } else{ failduration = deliveryQueueHelper.getInitFailureTimer(); } } @@ -168,6 +166,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { */ private synchronized void markRedirect(DeliveryTask task) { working.remove(task.getPublishId()); + logger.debug(task.getPublishId() + " marked as redirected."); retry.put(task.getPublishId(), task); } @@ -176,6 +175,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { */ private synchronized void markFailWithRetry(DeliveryTask task) { working.remove(task.getPublishId()); + logger.debug(task.getPublishId() + " marked as temporarily failed."); retry.put(task.getPublishId(), task); fdupdate(); } @@ -219,98 +219,97 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { return dt; } return null; + } } /** - * Update the destination info for this delivery queue. + * Create a delivery queue for a given destination info + */ + DeliveryQueue(DeliveryQueueHelper deliveryQueueHelper, DestInfo destinationInfo) { + this.deliveryQueueHelper = deliveryQueueHelper; + this.destinationInfo = destinationInfo; + dir = new File(destinationInfo.getSpool()); + dir.mkdirs(); + } + + /** + * Update the destination info for this delivery queue */ public void config(DestInfo destinationInfo) { this.destinationInfo = destinationInfo; } /** - * Get the dest info. + * Get the dest info */ public DestInfo getDestinationInfo() { return (destinationInfo); } /** - * Get the config manager. + * Get the config manager */ public DeliveryQueueHelper getConfig() { return (deliveryQueueHelper); } /** - * Exceptional condition occurred during delivery. + * Exceptional condition occurred during delivery */ public void reportDeliveryExtra(DeliveryTask task, long sent) { StatusLog.logDelExtra(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getLength(), sent); } /** - * Message too old to deliver. + * Message too old to deliver */ void reportExpiry(DeliveryTask task) { - StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts()); + StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts()); markExpired(task); } /** - * Completed a delivery attempt. + * Completed a delivery attempt */ public void reportStatus(DeliveryTask task, int status, String xpubid, String location) { if (status < 300) { - StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid); + StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid); if (destinationInfo.isPrivilegedSubscriber()) { - task.setResumeTime( - System.currentTimeMillis() + deliveryQueueHelper.getWaitForFileProcessFailureTimer()); + task.setResumeTime(System.currentTimeMillis() + deliveryQueueHelper.getWaitForFileProcessFailureTimer()); markFailWithRetry(task); } else { markSuccess(task); } } else if (status < 400 && deliveryQueueHelper.isFollowRedirects()) { - StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location); + StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location); if (deliveryQueueHelper.handleRedirection(destinationInfo, location, task.getFileId())) { markRedirect(task); } else { - StatusLog - .logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), "notRetryable", task.getAttempts()); + StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts()); markFailNoRetry(task); } - } else if (status < 500 && status != 429) { - // Status 429 is the standard response for Too Many Requests and indicates - // that a file needs to be delivered again at a later time. - StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location); - StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), "notRetryable", task.getAttempts()); + } else if (status < 500 && status != 429) { // Status 429 is the standard response for Too Many Requests and indicates that a file needs to be delivered again at a later time. + StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location); + StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts()); markFailNoRetry(task); } else { - StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location); + StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location); markFailWithRetry(task); } } /** - * Delivery failed by reason of an exception. + * Delivery failed by reason of an exception */ public void reportException(DeliveryTask task, Exception exception) { - StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), - task.getCType(), task.getLength(), destinationInfo.getAuthUser(), -1, exception.toString()); + StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), -1, exception.toString()); deliveryQueueHelper.handleUnreachable(destinationInfo); markFailWithRetry(task); } /** - * Get the feed ID for a subscription. + * Get the feed ID for a subscription * * @param subid The subscription ID * @return The feed ID @@ -320,20 +319,22 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { } /** - * Get the URL to deliver a message to given the file ID. + * Get the URL to deliver a message to given the file ID */ public String getDestURL(String fileid) { return (deliveryQueueHelper.getDestURL(destinationInfo, fileid)); } /** - * Deliver files until there's a failure or there are no more files to deliver. + * Deliver files until there's a failure or there are no more + * files to deliver */ public void run() { DeliveryTask task; long endtime = System.currentTimeMillis() + deliveryQueueHelper.getFairTimeLimit(); int filestogo = deliveryQueueHelper.getFairFileLimit(); while ((task = getNext()) != null) { + logger.debug("Processing file: " + task.getPublishId()); task.run(); if (--filestogo <= 0 || System.currentTimeMillis() > endtime) { break; @@ -342,21 +343,21 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { } /** - * Is there no work to do for this queue right now. + * Is there no work to do for this queue right now? */ synchronized boolean isSkipSet() { return (peekNext() == null); } /** - * Reset the retry timer. + * Reset the retry timer */ void resetQueue() { resumetime = System.currentTimeMillis(); } /** - * Get task if in queue and mark as success. + * Get task if in queue and mark as success */ boolean markTaskSuccess(String pubId) { DeliveryTask task = working.get(pubId); @@ -374,7 +375,6 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper { } return false; } - private void scanForNextTask(String[] files) { for (String fname : files) { String pubId = getPubId(fname); diff --git a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryTask.java b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryTask.java index bbd13912..91bd716b 100644 --- a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryTask.java +++ b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryTask.java @@ -39,6 +39,7 @@ import java.io.OutputStream; import java.net.HttpURLConnection; import java.net.ProtocolException; import java.net.URL; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; import java.util.UUID; @@ -355,13 +356,27 @@ public class DeliveryTask implements Runnable, Comparable<DeliveryTask> { * Remove meta and data files. */ void clean() { - datafile.delete(); - metafile.delete(); + deleteWithRetry(datafile); + deleteWithRetry(metafile); eelfLogger.info(EelfMsgs.INVOKE, newInvocationId); eelfLogger.info(EelfMsgs.EXIT); hdrs = null; } + private void deleteWithRetry(File file) { + int maxTries = 3; + int tryCount = 1; + while (tryCount <= maxTries) { + try { + Files.deleteIfExists(file.toPath()); + break; + } catch (IOException e) { + eelfLogger.error("IOException : Failed to delete file :" + file.getName() + " on attempt " + tryCount, e); + } + tryCount++; + } + } + /** * Get the resume time for a delivery task. */ diff --git a/datarouter-node/src/main/resources/logback.xml b/datarouter-node/src/main/resources/logback.xml index 2ce2050c..cc2ff4e2 100644 --- a/datarouter-node/src/main/resources/logback.xml +++ b/datarouter-node/src/main/resources/logback.xml @@ -365,7 +365,7 @@ - <root level="TRACE"> + <root level="INFO"> <appender-ref ref="asyncEELF" /> <appender-ref ref="asyncEELFError" /> <appender-ref ref="asyncEELFjettyAndNodelog" /> |