1 /*******************************************************************************
2 * ============LICENSE_START==================================================
4 * * ===========================================================================
5 * * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
6 * * ===========================================================================
7 * * Licensed under the Apache License, Version 2.0 (the "License");
8 * * you may not use this file except in compliance with the License.
9 * * You may obtain a copy of the License at
11 * * http://www.apache.org/licenses/LICENSE-2.0
13 * * Unless required by applicable law or agreed to in writing, software
14 * * distributed under the License is distributed on an "AS IS" BASIS,
15 * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * * See the License for the specific language governing permissions and
17 * * limitations under the License.
18 * * ============LICENSE_END====================================================
20 * * ECOMP is a trademark and service mark of AT&T Intellectual Property.
22 ******************************************************************************/
25 package org.onap.dmaap.datarouter.node;
27 import com.att.eelf.configuration.EELFLogger;
28 import com.att.eelf.configuration.EELFManager;
33 * Mechanism for monitoring and controlling delivery of files to a destination.
35 * The DeliveryQueue class maintains lists of DeliveryTasks for a single
36 * destination (a subscription or another data router node) and assigns
37 * delivery threads to try to deliver them. It also maintains a delivery
38 * status that causes it to back off on delivery attempts after a failure.
40 * If the most recent delivery result was a failure, then no more attempts
41 * will be made for a period of time. Initially, and on the first failure
42 * following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer() (milliseconds).
43 * If, after this delay, additional failures occur, each failure will
44 * multiply the delay by DeliveryQueueHelper.getFailureBackoff() up to a
45 * maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer().
46 * Note that this behavior applies to the delivery queue as a whole and not
47 * to individual files in the queue. If multiple files are being
48 * delivered and one fails, the delay will be started. If a second
49 * delivery fails while the delay was active, it will not change the delay
50 * or change the duration of any subsequent delay.
51 * If, however, it succeeds, it will cancel the delay.
53 * The queue maintains 3 collections of files to deliver: A todo list of
54 * files that will be attempted, a working set of files that are being
55 * attempted, and a retry set of files that were attempted and failed.
56 * Whenever the todo list is empty and needs to be refilled, a scan of the
57 * spool directory is made and the file names sorted. Any files in the working set are ignored.
58 * If a DeliveryTask for the file is in the retry set, then that delivery
59 * task is placed on the todo list. Otherwise, a new DeliveryTask for the
60 * file is created and placed on the todo list.
61 * If, when a DeliveryTask is about to be removed from the todo list, its
62 * age exceeds DeliveryQueueHelper.getExpirationTimer(), then it is instead
65 * A delivery queue also maintains a skip flag. This flag is true if the
66 * failure timer is active or if no files are found in a directory scan.
68 public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
69 private static EELFLogger logger = EELFManager.getInstance().getLogger(DeliveryQueue.class);
70 private DeliveryQueueHelper deliveryQueueHelper;
71 private DestInfo destinationInfo;
72 private Hashtable<String, DeliveryTask> working = new Hashtable<>();
73 private Hashtable<String, DeliveryTask> retry = new Hashtable<>();
74 private int todoindex;
75 private boolean failed;
76 private long failduration;
77 private long resumetime;
79 private Vector<DeliveryTask> todo = new Vector<>();
82 * Try to cancel a delivery task.
84 * @return The length of the task in bytes or 0 if the task cannot be cancelled.
86 synchronized long cancelTask(String pubid) {
87 if (working.get(pubid) != null) {
90 DeliveryTask dt = retry.get(pubid);
92 for (int i = todoindex; i < todo.size(); i++) {
93 DeliveryTask xdt = todo.get(i);
94 if (xdt.getPublishId().equals(pubid)) {
101 dt = new DeliveryTask(this, pubid);
102 if (dt.getFileId() == null) {
106 if (dt.isCleaned()) {
109 StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(), dt.getLength(), "diskFull", dt.getAttempts());
111 return (dt.getLength());
115 * Mark that a delivery task has succeeded.
117 private synchronized void markSuccess(DeliveryTask task) {
118 working.remove(task.getPublishId());
125 * Mark that a delivery task has expired.
127 private synchronized void markExpired(DeliveryTask task) {
132 * Mark that a delivery task has failed permanently.
134 private synchronized void markFailNoRetry(DeliveryTask task) {
135 working.remove(task.getPublishId());
141 private void fdupdate() {
144 if (failduration == 0) {
145 if (destinationInfo.isPrivilegedSubscriber()) {
146 failduration = deliveryQueueHelper.getWaitForFileProcessFailureTimer();
148 failduration = deliveryQueueHelper.getInitFailureTimer();
151 resumetime = System.currentTimeMillis() + failduration;
152 long maxdur = deliveryQueueHelper.getMaxFailureTimer();
153 failduration = (long) (failduration * deliveryQueueHelper.getFailureBackoff());
154 if (failduration > maxdur) {
155 failduration = maxdur;
161 * Mark that a delivery task has been redirected.
163 private synchronized void markRedirect(DeliveryTask task) {
164 working.remove(task.getPublishId());
165 retry.put(task.getPublishId(), task);
169 * Mark that a delivery task has temporarily failed.
171 private synchronized void markFailWithRetry(DeliveryTask task) {
172 working.remove(task.getPublishId());
173 retry.put(task.getPublishId(), task);
180 synchronized DeliveryTask getNext() {
181 DeliveryTask ret = peekNext();
184 working.put(ret.getPublishId(), ret);
190 * Peek at the next task.
192 synchronized DeliveryTask peekNext() {
193 long now = System.currentTimeMillis();
194 long mindate = now - deliveryQueueHelper.getExpirationTimer();
196 if (now > resumetime) {
203 if (todoindex >= todo.size()) {
205 todo = new Vector<>();
206 String[] files = dir.list();
208 for (String fname : files) {
209 if (!fname.endsWith(".M")) {
212 String fname2 = fname.substring(0, fname.length() - 2);
214 int dot = fname2.indexOf('.');
219 pidtime = Long.parseLong(fname2.substring(0, dot));
220 } catch (Exception e) {
221 logger.error("Exception", e);
223 if (pidtime < 1000000000000L) {
226 if (working.get(fname2) != null) {
229 DeliveryTask dt = retry.get(fname2);
231 dt = new DeliveryTask(this, fname2);
235 retry = new Hashtable<>();
237 if (todoindex < todo.size()) {
238 DeliveryTask dt = todo.get(todoindex);
239 if (dt.isCleaned()) {
243 if (destinationInfo.isPrivilegedSubscriber() && dt.getResumeTime() > System.currentTimeMillis()) {
244 retry.put(dt.getPublishId(), dt);
248 if (dt.getDate() >= mindate) {
260 * Create a delivery queue for a given destination info
262 DeliveryQueue(DeliveryQueueHelper deliveryQueueHelper, DestInfo destinationInfo) {
263 this.deliveryQueueHelper = deliveryQueueHelper;
264 this.destinationInfo = destinationInfo;
265 dir = new File(destinationInfo.getSpool());
270 * Update the destination info for this delivery queue
272 public void config(DestInfo destinationInfo) {
273 this.destinationInfo = destinationInfo;
279 public DestInfo getDestinationInfo() {
280 return (destinationInfo);
284 * Get the config manager
286 public DeliveryQueueHelper getConfig() {
287 return (deliveryQueueHelper);
291 * Exceptional condition occurred during delivery
293 public void reportDeliveryExtra(DeliveryTask task, long sent) {
294 StatusLog.logDelExtra(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getLength(), sent);
298 * Message too old to deliver
300 void reportExpiry(DeliveryTask task) {
301 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts());
306 * Completed a delivery attempt
308 public void reportStatus(DeliveryTask task, int status, String xpubid, String location) {
310 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid);
311 if (destinationInfo.isPrivilegedSubscriber()) {
312 task.setResumeTime(System.currentTimeMillis() + deliveryQueueHelper.getWaitForFileProcessFailureTimer());
313 markFailWithRetry(task);
317 } else if (status < 400 && deliveryQueueHelper.isFollowRedirects()) {
318 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
319 if (deliveryQueueHelper.handleRedirection(destinationInfo, location, task.getFileId())) {
322 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
323 markFailNoRetry(task);
325 } 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.
326 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
327 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
328 markFailNoRetry(task);
330 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
331 markFailWithRetry(task);
336 * Delivery failed by reason of an exception
338 public void reportException(DeliveryTask task, Exception exception) {
339 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), -1, exception.toString());
340 deliveryQueueHelper.handleUnreachable(destinationInfo);
341 markFailWithRetry(task);
345 * Get the feed ID for a subscription
347 * @param subid The subscription ID
348 * @return The feed ID
350 public String getFeedId(String subid) {
351 return (deliveryQueueHelper.getFeedId(subid));
355 * Get the URL to deliver a message to given the file ID
357 public String getDestURL(String fileid) {
358 return (deliveryQueueHelper.getDestURL(destinationInfo, fileid));
362 * Deliver files until there's a failure or there are no more
367 long endtime = System.currentTimeMillis() + deliveryQueueHelper.getFairTimeLimit();
368 int filestogo = deliveryQueueHelper.getFairFileLimit();
369 while ((t = getNext()) != null) {
371 if (--filestogo <= 0 || System.currentTimeMillis() > endtime) {
378 * Is there no work to do for this queue right now?
380 synchronized boolean isSkipSet() {
381 return (peekNext() == null);
385 * Reset the retry timer
388 resumetime = System.currentTimeMillis();
392 * Get task if in queue and mark as success
394 boolean markTaskSuccess(String pubId) {
395 DeliveryTask task = working.get(pubId);
400 task = retry.get(pubId);