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