datarouter-prov/src/main/java/com/att/research/datarouter/provisioning/utils/LogfileLoader.java

   1 /*******************************************************************************\r
   2  * ============LICENSE_START==================================================\r
   3  * * org.onap.dmaap\r
   4  * * ===========================================================================\r
   5  * * Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
   6  * * ===========================================================================\r
   7  * * Licensed under the Apache License, Version 2.0 (the "License");\r
   8  * * you may not use this file except in compliance with the License.\r
   9  * * You may obtain a copy of the License at\r
  10  * * \r
  11  *  *      http://www.apache.org/licenses/LICENSE-2.0\r
  12  * * \r
  13  *  * Unless required by applicable law or agreed to in writing, software\r
  14  * * distributed under the License is distributed on an "AS IS" BASIS,\r
  15  * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  16  * * See the License for the specific language governing permissions and\r
  17  * * limitations under the License.\r
  18  * * ============LICENSE_END====================================================\r
  19  * *\r
  20  * * ECOMP is a trademark and service mark of AT&T Intellectual Property.\r
  21  * *\r
  22  ******************************************************************************/\r
  23 \r
  24 \r
  25 package com.att.research.datarouter.provisioning.utils;\r
  26 \r
  27 import java.io.File;\r
  28 import java.io.FileInputStream;\r
  29 import java.io.FileNotFoundException;\r
  30 import java.io.FileReader;\r
  31 import java.io.FilenameFilter;\r
  32 import java.io.IOException;\r
  33 import java.io.InputStreamReader;\r
  34 import java.io.LineNumberReader;\r
  35 import java.io.Reader;\r
  36 import java.sql.Connection;\r
  37 import java.sql.PreparedStatement;\r
  38 import java.sql.ResultSet;\r
  39 import java.sql.SQLException;\r
  40 import java.sql.Statement;\r
  41 import java.text.ParseException;\r
  42 import java.util.Date;\r
  43 import java.util.HashMap;\r
  44 import java.util.Iterator;\r
  45 import java.util.Map;\r
  46 import java.util.TreeSet;\r
  47 import java.util.zip.GZIPInputStream;\r
  48 \r
  49 import org.apache.log4j.Logger;\r
  50 \r
  51 import com.att.research.datarouter.provisioning.BaseServlet;\r
  52 import com.att.research.datarouter.provisioning.beans.DeliveryExtraRecord;\r
  53 import com.att.research.datarouter.provisioning.beans.DeliveryRecord;\r
  54 import com.att.research.datarouter.provisioning.beans.ExpiryRecord;\r
  55 import com.att.research.datarouter.provisioning.beans.Loadable;\r
  56 import com.att.research.datarouter.provisioning.beans.LogRecord;\r
  57 import com.att.research.datarouter.provisioning.beans.Parameters;\r
  58 import com.att.research.datarouter.provisioning.beans.PubFailRecord;\r
  59 import com.att.research.datarouter.provisioning.beans.PublishRecord;\r
  60 \r
  61 /**\r
  62  * This class provides methods that run in a separate thread, in order to process logfiles uploaded into the spooldir.\r
  63  * These logfiles are loaded into the MySQL LOG_RECORDS table. In a running provisioning server, there should only be\r
  64  * two places where records can be loaded into this table; here, and in the method DB.retroFit4() which may be run at\r
  65  * startup to load the old (1.0) style log tables into LOG_RECORDS;\r
  66  * <p>This method maintains an {@link RLEBitSet} which can be used to easily see what records are presently in the\r
  67  * database.\r
  68  * This bit set is used to synchronize between provisioning servers.</p>\r
  69  *\r
  70  * @author Robert Eby\r
  71  * @version $Id: LogfileLoader.java,v 1.22 2014/03/12 19:45:41 eby Exp $\r
  72  */\r
  73 public class LogfileLoader extends Thread {\r
  74         /** Default number of log records to keep when pruning.  Keep 10M by default. */\r
  75         public static final long DEFAULT_LOG_RETENTION = 10000000L;\r
  76         /** NOT USED: Percentage of free space required before old records are removed. */\r
  77         public static final int REQUIRED_FREE_PCT = 20;\r
  78 \r
  79         /** This is a singleton -- there is only one LogfileLoader object in the server */\r
  80         private static LogfileLoader p;\r
  81 \r
  82         /**\r
  83          * Get the singleton LogfileLoader object, and start it if it is not running.\r
  84          * @return the LogfileLoader\r
  85          */\r
  86         public static synchronized LogfileLoader getLoader() {\r
  87                 if (p == null)\r
  88                         p = new LogfileLoader();\r
  89                 if (!p.isAlive())\r
  90                         p.start();\r
  91                 return p;\r
  92         }\r
  93 \r
  94         /** The PreparedStatement which is loaded by a <i>Loadable</i>. */\r
  95         public static final String INSERT_SQL = "insert into LOG_RECORDS values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";\r
  96         /** Each server can assign this many IDs */\r
  97         private static final long SET_SIZE = (1L << 56);\r
  98 \r
  99         private final Logger logger;\r
 100         private final DB db;\r
 101         private final String spooldir;\r
 102         private final long set_start;\r
 103         private final long set_end;\r
 104         private RLEBitSet seq_set;\r
 105         private long nextid;\r
 106         private boolean idle;\r
 107 \r
 108         private LogfileLoader() {\r
 109                 this.logger   = Logger.getLogger("com.att.research.datarouter.provisioning.internal");\r
 110                 this.db       = new DB();\r
 111                 this.spooldir = db.getProperties().getProperty("com.att.research.datarouter.provserver.spooldir");\r
 112                 this.set_start = getIdRange();\r
 113                 this.set_end   = set_start + SET_SIZE - 1;\r
 114                 this.seq_set  = new RLEBitSet();\r
 115                 this.nextid   = 0;\r
 116                 this.idle     = false;\r
 117 \r
 118                 // This is a potentially lengthy operation, so has been moved to run()\r
 119                 //initializeNextid();\r
 120                 this.setDaemon(true);\r
 121                 this.setName("LogfileLoader");\r
 122         }\r
 123 \r
 124         private long getIdRange() {\r
 125                 long n;\r
 126                 if (BaseServlet.isInitialActivePOD())\r
 127                         n = 0;\r
 128                 else if (BaseServlet.isInitialStandbyPOD())\r
 129                         n = SET_SIZE;\r
 130                 else\r
 131                         n = SET_SIZE * 2;\r
 132                 String r = String.format("[%X .. %X]", n, n+SET_SIZE-1);\r
 133                 logger.debug("This server shall assign RECORD_IDs in the range "+r);\r
 134                 return n;\r
 135         }\r
 136         /**\r
 137          * Return the bit set representing the record ID's that are loaded in this database.\r
 138          * @return the bit set\r
 139          */\r
 140         public RLEBitSet getBitSet() {\r
 141                 return seq_set;\r
 142         }\r
 143         /**\r
 144          * True if the LogfileLoader is currently waiting for work.\r
 145          * @return true if idle\r
 146          */\r
 147         public boolean isIdle() {\r
 148                 return idle;\r
 149         }\r
 150         /**\r
 151          * Run continuously to look for new logfiles in the spool directory and import them into the DB.\r
 152          * The spool is checked once per second.  If free space on the MySQL filesystem falls below\r
 153          * REQUIRED_FREE_PCT (normally 20%) then the oldest logfile entries are removed and the LOG_RECORDS\r
 154          * table is compacted until free space rises above the threshold.\r
 155          */\r
 156         @Override\r
 157         public void run() {\r
 158                 initializeNextid();     // moved from the constructor\r
 159                 while (true) {\r
 160                         try {\r
 161                                 File dirfile = new File(spooldir);\r
 162                                 while (true) {\r
 163                                         // process IN files\r
 164                                         File[] infiles = dirfile.listFiles(new FilenameFilter() {\r
 165                                                 @Override\r
 166                                                 public boolean accept(File dir, String name) {\r
 167                                                         return name.startsWith("IN.");\r
 168                                                 }\r
 169                                         });\r
 170 \r
 171                                         if (infiles.length == 0) {\r
 172                                                 idle = true;\r
 173                                                 try {\r
 174                                                         Thread.sleep(1000L);\r
 175                                                 } catch (InterruptedException e) {\r
 176                                                 }\r
 177                                                 idle = false;\r
 178                                         } else {\r
 179                                                 // Remove old rows\r
 180                                                 if (pruneRecords()) {\r
 181                                                         // Removed at least some entries, recompute the bit map\r
 182                                                         initializeNextid();\r
 183                                                 }\r
 184 \r
 185                                                 // Process incoming logfiles\r
 186                                                 for (File f : infiles) {\r
 187                                                         if (logger.isDebugEnabled())\r
 188                                                                 logger.debug("PROV8001 Starting " + f + " ...");\r
 189                                                         long time = System.currentTimeMillis();\r
 190                                                         int[] n = process(f);\r
 191                                                         time = System.currentTimeMillis() - time;\r
 192                                                         logger.info(String\r
 193                                                                         .format("PROV8000 Processed %s in %d ms; %d of %d records.",\r
 194                                                                                         f.toString(), time, n[0], n[1]));\r
 195                                                         f.delete();\r
 196                                                 }\r
 197                                         }\r
 198                                 }\r
 199                         } catch (Exception e) {\r
 200                                 logger.warn("PROV0020: Caught exception in LogfileLoader: " + e);\r
 201                                 e.printStackTrace();\r
 202                         }\r
 203                 }\r
 204         }\r
 205         private boolean pruneRecords() {\r
 206                 boolean did1 = false;\r
 207                 long count = countRecords();\r
 208                 long threshold = DEFAULT_LOG_RETENTION;\r
 209                 Parameters param = Parameters.getParameter(Parameters.PROV_LOG_RETENTION);\r
 210                 if (param != null) {\r
 211                         try {\r
 212                                 long n = Long.parseLong(param.getValue());\r
 213                                 // This check is to prevent inadvertent errors from wiping the table out\r
 214                                 if (n > 1000000L)\r
 215                                         threshold = n;\r
 216                         } catch (NumberFormatException e) {\r
 217                                 // ignore\r
 218                         }\r
 219                 }\r
 220                 logger.debug("Pruning LOG_RECORD table: records in DB="+count+", threshold="+threshold);\r
 221                 if (count > threshold) {\r
 222                         count -= threshold;                                             // we need to remove this many records;\r
 223                         Map<Long,Long> hist = getHistogram();   // histogram of records per day\r
 224                         // Determine the cutoff point to remove the needed number of records\r
 225                         long sum = 0;\r
 226                         long cutoff = 0;\r
 227                         for (Long day : new TreeSet<Long>(hist.keySet())) {\r
 228                                 sum += hist.get(day);\r
 229                                 cutoff = day;\r
 230                                 if (sum >= count)\r
 231                                         break;\r
 232                         }\r
 233                         cutoff++;\r
 234                         cutoff *= 86400000L;            // convert day to ms\r
 235                         logger.debug("  Pruning records older than="+(cutoff/86400000L)+" ("+new Date(cutoff)+")");\r
 236 \r
 237                         Connection conn = null;\r
 238                         try {\r
 239                                 // Limit to a million at a time to avoid typing up the DB for too long.\r
 240                                 conn = db.getConnection();\r
 241                                 PreparedStatement ps = conn.prepareStatement("DELETE from LOG_RECORDS where EVENT_TIME < ? limit 1000000");\r
 242                                 ps.setLong(1, cutoff);\r
 243                                 while (count > 0) {\r
 244                                         if (!ps.execute()) {\r
 245                                                 int dcount = ps.getUpdateCount();\r
 246                                                 count -= dcount;\r
 247                                                 logger.debug("  "+dcount+" rows deleted.");\r
 248                                                 did1 |= (dcount!=0);\r
 249                                                 if (dcount == 0)\r
 250                                                         count = 0;      // prevent inf. loops\r
 251                                         } else {\r
 252                                                 count = 0;      // shouldn't happen!\r
 253                                         }\r
 254                                 }\r
 255                                 ps.close();\r
 256                                 Statement stmt = conn.createStatement();\r
 257                                 stmt.execute("OPTIMIZE TABLE LOG_RECORDS");\r
 258                                 stmt.close();\r
 259                         } catch (SQLException e) {\r
 260                                 System.err.println(e);\r
 261                                 e.printStackTrace();\r
 262                         } finally {\r
 263                                 db.release(conn);\r
 264                         }\r
 265                 }\r
 266                 return did1;\r
 267         }\r
 268         private long countRecords() {\r
 269                 long count = 0;\r
 270                 Connection conn = null;\r
 271                 try {\r
 272                         conn = db.getConnection();\r
 273                         Statement stmt = conn.createStatement();\r
 274                         ResultSet rs = stmt.executeQuery("SELECT COUNT(*) as COUNT from LOG_RECORDS");\r
 275                         if (rs.next()) {\r
 276                                 count = rs.getLong("COUNT");\r
 277                         }\r
 278                         rs.close();\r
 279                         stmt.close();\r
 280                 } catch (SQLException e) {\r
 281                         System.err.println(e);\r
 282                         e.printStackTrace();\r
 283                 } finally {\r
 284                         db.release(conn);\r
 285                 }\r
 286                 return count;\r
 287         }\r
 288         private Map<Long,Long> getHistogram() {\r
 289                 Map<Long,Long> map = new HashMap<Long,Long>();\r
 290                 Connection conn = null;\r
 291                 try {\r
 292                         logger.debug("  LOG_RECORD table histogram...");\r
 293                         conn = db.getConnection();\r
 294                         Statement stmt = conn.createStatement();\r
 295                         ResultSet rs = stmt.executeQuery("SELECT FLOOR(EVENT_TIME/86400000) AS DAY, COUNT(*) AS COUNT FROM LOG_RECORDS GROUP BY DAY");\r
 296                         while (rs.next()) {\r
 297                                 long day = rs.getLong("DAY");\r
 298                                 long cnt = rs.getLong("COUNT");\r
 299                                 map.put(day, cnt);\r
 300                                 logger.debug("  "+day + "  "+cnt);\r
 301                         }\r
 302                         rs.close();\r
 303                         stmt.close();\r
 304                 } catch (SQLException e) {\r
 305                         System.err.println(e);\r
 306                         e.printStackTrace();\r
 307                 } finally {\r
 308                         db.release(conn);\r
 309                 }\r
 310                 return map;\r
 311         }\r
 312         private void initializeNextid() {\r
 313                 Connection conn = null;\r
 314                 try {\r
 315                         conn = db.getConnection();\r
 316                         Statement stmt = conn.createStatement();\r
 317                         // Build a bitset of all records in the LOG_RECORDS table\r
 318                         // We need to run this SELECT in stages, because otherwise we run out of memory!\r
 319                         RLEBitSet nbs = new RLEBitSet();\r
 320                         final long stepsize = 6000000L;\r
 321                         boolean go_again = true;\r
 322                         for (long i = 0; go_again; i += stepsize) {\r
 323                                 String sql = String.format("select RECORD_ID from LOG_RECORDS LIMIT %d,%d", i, stepsize);\r
 324                                 ResultSet rs = stmt.executeQuery(sql);\r
 325                                 go_again = false;\r
 326                                 while (rs.next()) {\r
 327                                         long n = rs.getLong("RECORD_ID");\r
 328                                         nbs.set(n);\r
 329                                         go_again = true;\r
 330                                 }\r
 331                                 rs.close();\r
 332                         }\r
 333                         stmt.close();\r
 334                         seq_set = nbs;\r
 335 \r
 336                         // Compare with the range for this server\r
 337                         // Determine the next ID for this set of record IDs\r
 338                         RLEBitSet tbs = (RLEBitSet) nbs.clone();\r
 339                         RLEBitSet idset = new RLEBitSet();\r
 340                         idset.set(set_start, set_start+SET_SIZE);\r
 341                         tbs.and(idset);\r
 342                         long t = tbs.length();\r
 343                         nextid = (t == 0) ? set_start : (t - 1);\r
 344                         if (nextid >= set_start+SET_SIZE) {\r
 345                                 // Handle wraparound, when the IDs reach the end of our "range"\r
 346                                 Long[] last = null;\r
 347                                 Iterator<Long[]> li = tbs.getRangeIterator();\r
 348                                 while (li.hasNext()) {\r
 349                                         last = li.next();\r
 350                                 }\r
 351                                 if (last != null) {\r
 352                                         tbs.clear(last[0], last[1]+1);\r
 353                                         t = tbs.length();\r
 354                                         nextid = (t == 0) ? set_start : (t - 1);\r
 355                                 }\r
 356                         }\r
 357                         logger.debug(String.format("initializeNextid, next ID is %d (%x)", nextid, nextid));\r
 358                 } catch (SQLException e) {\r
 359                         System.err.println(e);\r
 360                         e.printStackTrace();\r
 361                 } finally {\r
 362                         db.release(conn);\r
 363                 }\r
 364         }\r
 365 // OLD CODE - commented here for historical purposes\r
 366 //\r
 367 //      private boolean pruneRecordsOldAlgorithm() {\r
 368 //              // Determine space available -- available space must be at least 20% under /opt/app/mysql\r
 369 //              int pct = getFreePercentage();\r
 370 //              boolean did1 = false;\r
 371 //              while (pct < REQUIRED_FREE_PCT) {\r
 372 //                      logger.info("PROV8008: Free space is " + pct + "% - removing old log entries");\r
 373 //                      boolean didit = removeOldestEntries();\r
 374 //                      pct = didit ? getFreePercentage() : 100; // don't loop endlessly\r
 375 //                      did1 |= didit;\r
 376 //              }\r
 377 //              return did1;\r
 378 //      }\r
 379 //      private int getFreePercentage() {\r
 380 //              FileSystem fs = (Paths.get("/opt/app/mysql")).getFileSystem();\r
 381 //              long total = 0;\r
 382 //              long avail = 0;\r
 383 //              try {\r
 384 //                      for (FileStore store : fs.getFileStores()) {\r
 385 //                              total += store.getTotalSpace();\r
 386 //                              avail += store.getUsableSpace();\r
 387 //                      }\r
 388 //              } catch (IOException e) {\r
 389 //              }\r
 390 //              try { fs.close(); } catch (Exception e) { }\r
 391 //              return (int)((avail * 100) / total);\r
 392 //      }\r
 393 //      private boolean removeOldestEntries() {\r
 394 //              // Remove the last days worth of entries\r
 395 //              Connection conn = null;\r
 396 //              try {\r
 397 //                      conn = db.getConnection();\r
 398 //                      Statement stmt = conn.createStatement();\r
 399 //                      ResultSet rs = stmt.executeQuery("select min(event_time) as MIN from LOG_RECORDS");\r
 400 //                      if (rs != null) {\r
 401 //                              if (rs.next()) {\r
 402 //                                      // Compute the end of the first day of logs\r
 403 //                                      long first = rs.getLong("MIN");\r
 404 //                                      Calendar cal = new GregorianCalendar();\r
 405 //                                      cal.setTime(new Date(first));\r
 406 //                                      cal.add(Calendar.DAY_OF_YEAR, 1);\r
 407 //                                      cal.set(Calendar.HOUR_OF_DAY, 0);\r
 408 //                                      cal.set(Calendar.MINUTE, 0);\r
 409 //                                      cal.set(Calendar.SECOND, 0);\r
 410 //                                      cal.set(Calendar.MILLISECOND, 0);\r
 411 //                                      if (!stmt.execute("delete from LOG_RECORDS where event_time < " + cal.getTimeInMillis())) {\r
 412 //                                              int count = stmt.getUpdateCount();\r
 413 //                                              logger.info("PROV0009: Removed "+count+" old log entries.");\r
 414 //                                              stmt.execute("OPTIMIZE TABLE LOG_RECORDS");\r
 415 //                                      }\r
 416 //                                      rs.close();\r
 417 //                                      stmt.close();\r
 418 //                                      return true;\r
 419 //                              }\r
 420 //                              rs.close();\r
 421 //                      }\r
 422 //                      stmt.close();\r
 423 //              } catch (SQLException e) {\r
 424 //                      System.err.println(e);\r
 425 //                      e.printStackTrace();\r
 426 //              } finally {\r
 427 //                      db.release(conn);\r
 428 //              }\r
 429 //              return false;\r
 430 //      }\r
 431         @SuppressWarnings("resource")\r
 432         private int[] process(File f) {\r
 433                 int ok = 0, total = 0;\r
 434                 try {\r
 435                         Connection conn = db.getConnection();\r
 436                         PreparedStatement ps = conn.prepareStatement(INSERT_SQL);\r
 437                         Reader r = f.getPath().endsWith(".gz")\r
 438                                 ? new InputStreamReader(new GZIPInputStream(new FileInputStream(f)))\r
 439                                 : new FileReader(f);\r
 440                         LineNumberReader in = new LineNumberReader(r);\r
 441                         String line;\r
 442                         while ((line = in.readLine()) != null) {\r
 443                                 try {\r
 444                                         for (Loadable rec : buildRecords(line)) {\r
 445                                                 rec.load(ps);\r
 446                                                 if (rec instanceof LogRecord) {\r
 447                                                         LogRecord lr = ((LogRecord)rec);\r
 448                                                         if (!seq_set.get(lr.getRecordId())) {\r
 449                                                                 ps.executeUpdate();\r
 450                                                                 seq_set.set(lr.getRecordId());\r
 451                                                         } else\r
 452                                                                 logger.debug("Duplicate record ignored: "+lr.getRecordId());\r
 453                                                 } else {\r
 454                                                         if (++nextid > set_end)\r
 455                                                                 nextid = set_start;\r
 456                                                         ps.setLong(18, nextid);\r
 457                                                         ps.executeUpdate();\r
 458                                                         seq_set.set(nextid);\r
 459                                                 }\r
 460                                                 ps.clearParameters();\r
 461                                                 ok++;\r
 462                                         }\r
 463                                 } catch (SQLException e) {\r
 464                                         logger.warn("PROV8003 Invalid value in record: "+line);\r
 465                                         logger.debug(e);\r
 466                                         e.printStackTrace();\r
 467                                 } catch (NumberFormatException e) {\r
 468                                         logger.warn("PROV8004 Invalid number in record: "+line);\r
 469                                         logger.debug(e);\r
 470                                         e.printStackTrace();\r
 471                                 } catch (ParseException e) {\r
 472                                         logger.warn("PROV8005 Invalid date in record: "+line);\r
 473                                         logger.debug(e);\r
 474                                         e.printStackTrace();\r
 475                                 } catch (Exception e) {\r
 476                                         logger.warn("PROV8006 Invalid pattern in record: "+line);\r
 477                                         logger.debug(e);\r
 478                                         e.printStackTrace();\r
 479                                 }\r
 480                                 total++;\r
 481                         }\r
 482                         in.close();\r
 483                         ps.close();\r
 484                         db.release(conn);\r
 485                         conn = null;\r
 486                 } catch (FileNotFoundException e) {\r
 487                         logger.warn("PROV8007 Exception reading "+f+": "+e);\r
 488                 } catch (IOException e) {\r
 489                         logger.warn("PROV8007 Exception reading "+f+": "+e);\r
 490                 } catch (SQLException e) {\r
 491                         logger.warn("PROV8007 Exception reading "+f+": "+e);\r
 492                 }\r
 493                 return new int[] { ok, total };\r
 494         }\r
 495         private Loadable[] buildRecords(String line) throws ParseException {\r
 496                 String[] pp = line.split("\\|");\r
 497                 if (pp != null && pp.length >= 7) {\r
 498                         String rtype = pp[1].toUpperCase();\r
 499                         if (rtype.equals("PUB") && pp.length == 11) {\r
 500                                 // Fields are: date|PUB|pubid|feedid|requrl|method|ctype|clen|srcip|user|status\r
 501                                 return new Loadable[] { new PublishRecord(pp) };\r
 502                         }\r
 503                         if (rtype.equals("DEL") && pp.length == 12) {\r
 504                                 // Fields are: date|DEL|pubid|feedid|subid|requrl|method|ctype|clen|user|status|xpubid\r
 505                                 String[] subs = pp[4].split("\\s+");\r
 506                                 if (subs != null) {\r
 507                                         Loadable[] rv = new Loadable[subs.length];\r
 508                                         for (int i = 0; i < subs.length; i++) {\r
 509                                                 // create a new record for each individual sub\r
 510                                                 pp[4] = subs[i];\r
 511                                                 rv[i] = new DeliveryRecord(pp);\r
 512                                         }\r
 513                                         return rv;\r
 514                                 }\r
 515                         }\r
 516                         if (rtype.equals("EXP") && pp.length == 11) {\r
 517                                 // Fields are: date|EXP|pubid|feedid|subid|requrl|method|ctype|clen|reason|attempts\r
 518                                 ExpiryRecord e = new ExpiryRecord(pp);\r
 519                                 if (e.getReason().equals("other"))\r
 520                                         logger.info("Invalid reason '"+pp[9]+"' changed to 'other' for record: "+e.getPublishId());\r
 521                                 return new Loadable[] { e };\r
 522                         }\r
 523                         if (rtype.equals("PBF") && pp.length == 12) {\r
 524                                 // Fields are: date|PBF|pubid|feedid|requrl|method|ctype|clen-expected|clen-received|srcip|user|error\r
 525                                 return new Loadable[] { new PubFailRecord(pp) };\r
 526                         }\r
 527                         if (rtype.equals("DLX") && pp.length == 7) {\r
 528                                 // Fields are: date|DLX|pubid|feedid|subid|clen-tosend|clen-sent\r
 529                                 return new Loadable[] { new DeliveryExtraRecord(pp) };\r
 530                         }\r
 531                         if (rtype.equals("LOG") && (pp.length == 19 || pp.length == 20)) {\r
 532                                 // Fields are: date|LOG|pubid|feedid|requrl|method|ctype|clen|type|feedFileid|remoteAddr|user|status|subid|fileid|result|attempts|reason|record_id\r
 533                                 return new Loadable[] { new LogRecord(pp) };\r
 534                         }\r
 535                 }\r
 536                 logger.warn("PROV8002 bad record: "+line);\r
 537                 return new Loadable[0];\r
 538         }\r
 539 \r
 540         /**\r
 541          * The LogfileLoader can be run stand-alone by invoking the main() method of this class.\r
 542          * @param a ignored\r
 543          * @throws InterruptedException\r
 544          */\r
 545         public static void main(String[] a) throws InterruptedException {\r
 546                 LogfileLoader.getLoader();\r
 547                 Thread.sleep(200000L);\r
 548         }\r
 549 }\r