datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java

   1 /*******************************************************************************
   2  * ============LICENSE_START==================================================
   3  * * org.onap.dmaap
   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
  10  * *
  11  *  *      http://www.apache.org/licenses/LICENSE-2.0
  12  * *
  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====================================================
  19  * *
  20  * * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  21  * *
  22  ******************************************************************************/
  23
  24
  25 package org.onap.dmaap.datarouter.node;
  26
  27 import java.io.*;
  28 import java.util.*;
  29
  30 /**
  31  * Mechanism for monitoring and controlling delivery of files to a destination.
  32  * <p>
  33  * The DeliveryQueue class maintains lists of DeliveryTasks for a single
  34  * destination (a subscription or another data router node) and assigns
  35  * delivery threads to try to deliver them.  It also maintains a delivery
  36  * status that causes it to back off on delivery attempts after a failure.
  37  * <p>
  38  * If the most recent delivery result was a failure, then no more attempts
  39  * will be made for a period of time.  Initially, and on the first failure
  40  * following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer() (milliseconds).
  41  * If, after this delay, additional failures occur, each failure will
  42  * multiply the delay by DeliveryQueueHelper.getFailureBackoff() up to a
  43  * maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer().
  44  * Note that this behavior applies to the delivery queue as a whole and not
  45  * to individual files in the queue.  If multiple files are being
  46  * delivered and one fails, the delay will be started.  If a second
  47  * delivery fails while the delay was active, it will not change the delay
  48  * or change the duration of any subsequent delay.
  49  * If, however, it succeeds, it will cancel the delay.
  50  * <p>
  51  * The queue maintains 3 collections of files to deliver: A todo list of
  52  * files that will be attempted, a working set of files that are being
  53  * attempted, and a retry set of files that were attempted and failed.
  54  * Whenever the todo list is empty and needs to be refilled, a scan of the
  55  * spool directory is made and the file names sorted.  Any files in the working set are ignored.
  56  * If a DeliveryTask for the file is in the retry set, then that delivery
  57  * task is placed on the todo list.  Otherwise, a new DeliveryTask for the
  58  * file is created and placed on the todo list.
  59  * If, when a DeliveryTask is about to be removed from the todo list, its
  60  * age exceeds DeliveryQueueHelper.getExpirationTimer(), then it is instead
  61  * marked as expired.
  62  * <p>
  63  * A delivery queue also maintains a skip flag.  This flag is true if the
  64  * failure timer is active or if no files are found in a directory scan.
  65  */
  66 public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
  67     private DeliveryQueueHelper deliveryQueueHelper;
  68     private DestInfo destinationInfo;
  69     private Hashtable<String, DeliveryTask> working = new Hashtable<>();
  70     private Hashtable<String, DeliveryTask> retry = new Hashtable<>();
  71     private int todoindex;
  72     private boolean failed;
  73     private long failduration;
  74     private long resumetime;
  75     private File dir;
  76     private Vector<DeliveryTask> todo = new Vector<>();
  77
  78     /**
  79      * Try to cancel a delivery task.
  80      *
  81      * @return The length of the task in bytes or 0 if the task cannot be cancelled.
  82      */
  83     synchronized long cancelTask(String pubid) {
  84         if (working.get(pubid) != null) {
  85             return (0);
  86         }
  87         DeliveryTask dt = retry.get(pubid);
  88         if (dt == null) {
  89             for (int i = todoindex; i < todo.size(); i++) {
  90                 DeliveryTask xdt = todo.get(i);
  91                 if (xdt.getPublishId().equals(pubid)) {
  92                     dt = xdt;
  93                     break;
  94                 }
  95             }
  96         }
  97         if (dt == null) {
  98             dt = new DeliveryTask(this, pubid);
  99             if (dt.getFileId() == null) {
 100                 return (0);
 101             }
 102         }
 103         if (dt.isCleaned()) {
 104             return (0);
 105         }
 106         StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(), dt.getLength(), "diskFull", dt.getAttempts());
 107         dt.clean();
 108         return (dt.getLength());
 109     }
 110
 111     /**
 112      * Mark that a delivery task has succeeded.
 113      */
 114     private synchronized void markSuccess(DeliveryTask task) {
 115         working.remove(task.getPublishId());
 116         task.clean();
 117         failed = false;
 118         failduration = 0;
 119     }
 120
 121     /**
 122      * Mark that a delivery task has expired.
 123      */
 124     private synchronized void markExpired(DeliveryTask task) {
 125         task.clean();
 126     }
 127
 128     /**
 129      * Mark that a delivery task has failed permanently.
 130      */
 131     private synchronized void markFailNoRetry(DeliveryTask task) {
 132         working.remove(task.getPublishId());
 133         task.clean();
 134         failed = false;
 135         failduration = 0;
 136     }
 137
 138     private void fdupdate() {
 139         if (!failed) {
 140             failed = true;
 141             if (failduration == 0) {
 142                 if (destinationInfo.isPrivilegedSubscriber()) {
 143                     failduration = deliveryQueueHelper.getWaitForFileProcessFailureTimer();
 144                 } else{
 145                     failduration = deliveryQueueHelper.getInitFailureTimer();
 146                 }
 147             }
 148             resumetime = System.currentTimeMillis() + failduration;
 149             long maxdur = deliveryQueueHelper.getMaxFailureTimer();
 150             failduration = (long) (failduration * deliveryQueueHelper.getFailureBackoff());
 151             if (failduration > maxdur) {
 152                 failduration = maxdur;
 153             }
 154         }
 155     }
 156
 157     /**
 158      * Mark that a delivery task has been redirected.
 159      */
 160     private synchronized void markRedirect(DeliveryTask task) {
 161         working.remove(task.getPublishId());
 162         retry.put(task.getPublishId(), task);
 163     }
 164
 165     /**
 166      * Mark that a delivery task has temporarily failed.
 167      */
 168     private synchronized void markFailWithRetry(DeliveryTask task) {
 169         working.remove(task.getPublishId());
 170         retry.put(task.getPublishId(), task);
 171         fdupdate();
 172     }
 173
 174     /**
 175      * Get the next task.
 176      */
 177     synchronized DeliveryTask getNext() {
 178         DeliveryTask ret = peekNext();
 179         if (ret != null) {
 180             todoindex++;
 181             working.put(ret.getPublishId(), ret);
 182         }
 183         return (ret);
 184     }
 185
 186     /**
 187      * Peek at the next task.
 188      */
 189     synchronized DeliveryTask peekNext() {
 190         long now = System.currentTimeMillis();
 191         long mindate = now - deliveryQueueHelper.getExpirationTimer();
 192         if (failed) {
 193             if (now > resumetime) {
 194                 failed = false;
 195             } else {
 196                 return (null);
 197             }
 198         }
 199         while (true) {
 200             if (todoindex >= todo.size()) {
 201                 todoindex = 0;
 202                 todo = new Vector<>();
 203                 String[] files = dir.list();
 204                 Arrays.sort(files);
 205                 for (String fname : files) {
 206                     if (!fname.endsWith(".M")) {
 207                         continue;
 208                     }
 209                     String fname2 = fname.substring(0, fname.length() - 2);
 210                     long pidtime = 0;
 211                     int dot = fname2.indexOf('.');
 212                     if (dot < 1) {
 213                         continue;
 214                     }
 215                     try {
 216                         pidtime = Long.parseLong(fname2.substring(0, dot));
 217                     } catch (Exception e) {
 218                     }
 219                     if (pidtime < 1000000000000L) {
 220                         continue;
 221                     }
 222                     if (working.get(fname2) != null) {
 223                         continue;
 224                     }
 225                     DeliveryTask dt = retry.get(fname2);
 226                     if (dt == null) {
 227                         dt = new DeliveryTask(this, fname2);
 228                     }
 229                     todo.add(dt);
 230                 }
 231                 retry = new Hashtable<>();
 232             }
 233             if (todoindex < todo.size()) {
 234                 DeliveryTask dt = todo.get(todoindex);
 235                 if (dt.isCleaned()) {
 236                     todoindex++;
 237                     continue;
 238                 }
 239                 if (destinationInfo.isPrivilegedSubscriber() && dt.getResumeTime() > System.currentTimeMillis()) {
 240                     retry.put(dt.getPublishId(), dt);
 241                     todoindex++;
 242                     continue;
 243                 }
 244                 if (dt.getDate() >= mindate) {
 245                     return (dt);
 246                 }
 247                 todoindex++;
 248                 reportExpiry(dt);
 249                 continue;
 250             }
 251             return (null);
 252         }
 253     }
 254
 255     /**
 256      * Create a delivery queue for a given destination info
 257      */
 258     DeliveryQueue(DeliveryQueueHelper deliveryQueueHelper, DestInfo destinationInfo) {
 259         this.deliveryQueueHelper = deliveryQueueHelper;
 260         this.destinationInfo = destinationInfo;
 261         dir = new File(destinationInfo.getSpool());
 262         dir.mkdirs();
 263     }
 264
 265     /**
 266      * Update the destination info for this delivery queue
 267      */
 268     public void config(DestInfo destinationInfo) {
 269         this.destinationInfo = destinationInfo;
 270     }
 271
 272     /**
 273      * Get the dest info
 274      */
 275     public DestInfo getDestinationInfo() {
 276         return (destinationInfo);
 277     }
 278
 279     /**
 280      * Get the config manager
 281      */
 282     public DeliveryQueueHelper getConfig() {
 283         return (deliveryQueueHelper);
 284     }
 285
 286     /**
 287      * Exceptional condition occurred during delivery
 288      */
 289     public void reportDeliveryExtra(DeliveryTask task, long sent) {
 290         StatusLog.logDelExtra(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getLength(), sent);
 291     }
 292
 293     /**
 294      * Message too old to deliver
 295      */
 296     void reportExpiry(DeliveryTask task) {
 297         StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts());
 298         markExpired(task);
 299     }
 300
 301     /**
 302      * Completed a delivery attempt
 303      */
 304     public void reportStatus(DeliveryTask task, int status, String xpubid, String location) {
 305         if (status < 300) {
 306             StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid);
 307             if (destinationInfo.isPrivilegedSubscriber()) {
 308                 task.setResumeTime(System.currentTimeMillis() + deliveryQueueHelper.getWaitForFileProcessFailureTimer());
 309                 markFailWithRetry(task);
 310             } else {
 311                 markSuccess(task);
 312             }
 313         } else if (status < 400 && deliveryQueueHelper.isFollowRedirects()) {
 314             StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
 315             if (deliveryQueueHelper.handleRedirection(destinationInfo, location, task.getFileId())) {
 316                 markRedirect(task);
 317             } else {
 318                 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
 319                 markFailNoRetry(task);
 320             }
 321         } 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.
 322             StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
 323             StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
 324             markFailNoRetry(task);
 325         } else {
 326             StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
 327             markFailWithRetry(task);
 328         }
 329     }
 330
 331     /**
 332      * Delivery failed by reason of an exception
 333      */
 334     public void reportException(DeliveryTask task, Exception exception) {
 335         StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), -1, exception.toString());
 336         deliveryQueueHelper.handleUnreachable(destinationInfo);
 337         markFailWithRetry(task);
 338     }
 339
 340     /**
 341      * Get the feed ID for a subscription
 342      *
 343      * @param subid The subscription ID
 344      * @return The feed ID
 345      */
 346     public String getFeedId(String subid) {
 347         return (deliveryQueueHelper.getFeedId(subid));
 348     }
 349
 350     /**
 351      * Get the URL to deliver a message to given the file ID
 352      */
 353     public String getDestURL(String fileid) {
 354         return (deliveryQueueHelper.getDestURL(destinationInfo, fileid));
 355     }
 356
 357     /**
 358      * Deliver files until there's a failure or there are no more
 359      * files to deliver
 360      */
 361     public void run() {
 362         DeliveryTask t;
 363         long endtime = System.currentTimeMillis() + deliveryQueueHelper.getFairTimeLimit();
 364         int filestogo = deliveryQueueHelper.getFairFileLimit();
 365         while ((t = getNext()) != null) {
 366             t.run();
 367             if (--filestogo <= 0 || System.currentTimeMillis() > endtime) {
 368                 break;
 369             }
 370         }
 371     }
 372
 373     /**
 374      * Is there no work to do for this queue right now?
 375      */
 376     synchronized boolean isSkipSet() {
 377         return (peekNext() == null);
 378     }
 379
 380     /**
 381      * Reset the retry timer
 382      */
 383     void resetQueue() {
 384         resumetime = System.currentTimeMillis();
 385     }
 386
 387     /**
 388      * Get task if in queue and mark as success
 389      */
 390     boolean markTaskSuccess(String pubId) {
 391         DeliveryTask task = working.get(pubId);
 392         if (task != null) {
 393             markSuccess(task);
 394             return true;
 395         }
 396         task = retry.get(pubId);
 397         if (task != null) {
 398             retry.remove(pubId);
 399             task.clean();
 400             resetQueue();
 401             failduration = 0;
 402             return true;
 403         }
 404         return false;
 405     }
 406 }