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;
30 import java.util.Arrays;
31 import java.util.Hashtable;
32 import java.util.Vector;
33 import org.jetbrains.annotations.Nullable;
36 * Mechanism for monitoring and controlling delivery of files to a destination.
38 * <p>The DeliveryQueue class maintains lists of DeliveryTasks for a single destination (a subscription or another data
39 * router node) and assigns delivery threads to try to deliver them. It also maintains a delivery status that causes it
40 * to back off on delivery attempts after a failure.
42 * <p>If the most recent delivery result was a failure, then no more attempts will be made for a period of time.
43 * Initially, and on the first failure following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer()
44 * (milliseconds). If, after this delay, additional failures occur, each failure will multiply the delay by
45 * DeliveryQueueHelper.getFailureBackoff() up to a maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer().
46 * Note that this behavior applies to the delivery queue as a whole and not to individual files in the queue. If
47 * multiple files are being delivered and one fails, the delay will be started. If a second delivery fails while the
48 * delay was active, it will not change the delay or change the duration of any subsequent delay. If, however, it
49 * succeeds, it will cancel the delay.
51 * <p>The queue maintains 3 collections of files to deliver: A to do list of files that will be attempted, a working
52 * set of files that are being attempted, and a retry set of files that were attempted and failed. Whenever the to do
53 * list is empty and needs to be refilled, a scan of the spool directory is made and the file names sorted. Any files
54 * in the working set are ignored. If a DeliveryTask for the file is in the retry set, then that delivery task is placed
55 * on the to do list. Otherwise, a new DeliveryTask for the file is created and placed on the to do list. If, when a
56 * DeliveryTask is about to be removed from the to do list, its age exceeds DeliveryQueueHelper.getExpirationTimer(),
57 * then it is instead marked as expired.
59 * <p>A delivery queue also maintains a skip flag. This flag is true if the failure timer is active or if no files are
60 * found in a directory scan.
62 public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
64 private static EELFLogger logger = EELFManager.getInstance().getLogger(DeliveryQueue.class);
65 private DeliveryQueueHelper deliveryQueueHelper;
66 private DestInfo destinationInfo;
67 private Hashtable<String, DeliveryTask> working = new Hashtable<>();
68 private Hashtable<String, DeliveryTask> retry = new Hashtable<>();
69 private int todoindex;
70 private boolean failed;
71 private long failduration;
72 private long resumetime;
74 private Vector<DeliveryTask> todo = new Vector<>();
77 * Create a delivery queue for a given destination info.
79 DeliveryQueue(DeliveryQueueHelper deliveryQueueHelper, DestInfo destinationInfo) {
80 this.deliveryQueueHelper = deliveryQueueHelper;
81 this.destinationInfo = destinationInfo;
82 dir = new File(destinationInfo.getSpool());
87 * Try to cancel a delivery task.
89 * @return The length of the task in bytes or 0 if the task cannot be cancelled.
91 synchronized long cancelTask(String pubid) {
92 if (working.get(pubid) != null) {
95 DeliveryTask dt = retry.get(pubid);
97 for (int i = todoindex; i < todo.size(); i++) {
98 DeliveryTask xdt = todo.get(i);
99 if (xdt.getPublishId().equals(pubid)) {
106 dt = new DeliveryTask(this, pubid);
107 if (dt.getFileId() == null) {
111 if (dt.isCleaned()) {
114 StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(),
115 dt.getLength(), "diskFull", dt.getAttempts());
117 return (dt.getLength());
121 * Mark that a delivery task has succeeded.
123 private synchronized void markSuccess(DeliveryTask task) {
124 working.remove(task.getPublishId());
131 * Mark that a delivery task has expired.
133 private synchronized void markExpired(DeliveryTask task) {
138 * Mark that a delivery task has failed permanently.
140 private synchronized void markFailNoRetry(DeliveryTask task) {
141 working.remove(task.getPublishId());
147 private void fdupdate() {
150 if (failduration == 0) {
151 if (destinationInfo.isPrivilegedSubscriber()) {
152 failduration = deliveryQueueHelper.getWaitForFileProcessFailureTimer();
154 failduration = deliveryQueueHelper.getInitFailureTimer();
157 resumetime = System.currentTimeMillis() + failduration;
158 long maxdur = deliveryQueueHelper.getMaxFailureTimer();
159 failduration = (long) (failduration * deliveryQueueHelper.getFailureBackoff());
160 if (failduration > maxdur) {
161 failduration = maxdur;
167 * Mark that a delivery task has been redirected.
169 private synchronized void markRedirect(DeliveryTask task) {
170 working.remove(task.getPublishId());
171 retry.put(task.getPublishId(), task);
175 * Mark that a delivery task has temporarily failed.
177 private synchronized void markFailWithRetry(DeliveryTask task) {
178 working.remove(task.getPublishId());
179 retry.put(task.getPublishId(), task);
186 synchronized DeliveryTask getNext() {
187 DeliveryTask ret = peekNext();
190 working.put(ret.getPublishId(), ret);
196 * Peek at the next task.
198 synchronized DeliveryTask peekNext() {
199 long now = System.currentTimeMillis();
200 long mindate = now - deliveryQueueHelper.getExpirationTimer();
202 if (now > resumetime) {
209 if (todoindex >= todo.size()) {
211 todo = new Vector<>();
212 String[] files = dir.list();
214 scanForNextTask(files);
215 retry = new Hashtable<>();
217 DeliveryTask dt = getDeliveryTask(mindate);
226 * Update the destination info for this delivery queue.
228 public void config(DestInfo destinationInfo) {
229 this.destinationInfo = destinationInfo;
235 public DestInfo getDestinationInfo() {
236 return (destinationInfo);
240 * Get the config manager.
242 public DeliveryQueueHelper getConfig() {
243 return (deliveryQueueHelper);
247 * Exceptional condition occurred during delivery.
249 public void reportDeliveryExtra(DeliveryTask task, long sent) {
250 StatusLog.logDelExtra(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getLength(), sent);
254 * Message too old to deliver.
256 void reportExpiry(DeliveryTask task) {
257 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
258 task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts());
263 * Completed a delivery attempt.
265 public void reportStatus(DeliveryTask task, int status, String xpubid, String location) {
267 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
268 task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid);
269 if (destinationInfo.isPrivilegedSubscriber()) {
271 System.currentTimeMillis() + deliveryQueueHelper.getWaitForFileProcessFailureTimer());
272 markFailWithRetry(task);
276 } else if (status < 400 && deliveryQueueHelper.isFollowRedirects()) {
277 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
278 task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
279 if (deliveryQueueHelper.handleRedirection(destinationInfo, location, task.getFileId())) {
283 .logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
284 task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
285 markFailNoRetry(task);
287 } else if (status < 500 && status != 429) {
288 // Status 429 is the standard response for Too Many Requests and indicates
289 // that a file needs to be delivered again at a later time.
290 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
291 task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
292 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
293 task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
294 markFailNoRetry(task);
296 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
297 task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
298 markFailWithRetry(task);
303 * Delivery failed by reason of an exception.
305 public void reportException(DeliveryTask task, Exception exception) {
306 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
307 task.getCType(), task.getLength(), destinationInfo.getAuthUser(), -1, exception.toString());
308 deliveryQueueHelper.handleUnreachable(destinationInfo);
309 markFailWithRetry(task);
313 * Get the feed ID for a subscription.
315 * @param subid The subscription ID
316 * @return The feed ID
318 public String getFeedId(String subid) {
319 return (deliveryQueueHelper.getFeedId(subid));
323 * Get the URL to deliver a message to given the file ID.
325 public String getDestURL(String fileid) {
326 return (deliveryQueueHelper.getDestURL(destinationInfo, fileid));
330 * Deliver files until there's a failure or there are no more files to deliver.
334 long endtime = System.currentTimeMillis() + deliveryQueueHelper.getFairTimeLimit();
335 int filestogo = deliveryQueueHelper.getFairFileLimit();
336 while ((task = getNext()) != null) {
338 if (--filestogo <= 0 || System.currentTimeMillis() > endtime) {
345 * Is there no work to do for this queue right now.
347 synchronized boolean isSkipSet() {
348 return (peekNext() == null);
352 * Reset the retry timer.
355 resumetime = System.currentTimeMillis();
359 * Get task if in queue and mark as success.
361 boolean markTaskSuccess(String pubId) {
362 DeliveryTask task = working.get(pubId);
367 task = retry.get(pubId);
378 private void scanForNextTask(String[] files) {
379 for (String fname : files) {
380 String pubId = getPubId(fname);
384 DeliveryTask dt = retry.get(pubId);
386 dt = new DeliveryTask(this, pubId);
393 private DeliveryTask getDeliveryTask(long mindate) {
394 if (todoindex < todo.size()) {
395 DeliveryTask dt = todo.get(todoindex);
396 if (dt.isCleaned()) {
399 if (destinationInfo.isPrivilegedSubscriber() && dt.getResumeTime() > System.currentTimeMillis()) {
400 retry.put(dt.getPublishId(), dt);
403 if (dt.getDate() >= mindate) {
413 private String getPubId(String fname) {
414 if (!fname.endsWith(".M")) {
417 String fname2 = fname.substring(0, fname.length() - 2);
419 int dot = fname2.indexOf('.');
424 pidtime = Long.parseLong(fname2.substring(0, dot));
425 } catch (Exception e) {
426 logger.error("Exception", e);
428 if (pidtime < 1000000000000L) {
431 if (working.get(fname2) != null) {