AwsBatchIndexer.java

  1. package gov.usgs.earthquake.aws;

  3. import java.io.BufferedReader;
  4. import java.io.InputStreamReader;
  5. import java.io.StringReader;
  6. import java.net.URL;
  7. import java.nio.charset.StandardCharsets;
  8. import java.sql.Connection;
  9. import java.sql.PreparedStatement;
  10. import java.sql.ResultSet;
  11. import java.util.Date;
  12. import java.util.concurrent.Executors;
  13. import java.util.concurrent.ThreadPoolExecutor;
  14. import java.util.logging.Level;
  15. import java.util.logging.Logger;

  17. import javax.json.Json;
  18. import javax.json.JsonObject;
  19. import javax.json.JsonReader;

  21. import gov.usgs.earthquake.distribution.Bootstrappable;
  22. import gov.usgs.earthquake.indexer.Indexer;
  23. import gov.usgs.earthquake.product.Product;
  24. import gov.usgs.earthquake.product.ProductId;
  25. import gov.usgs.earthquake.util.JDBCConnection;
  26. import gov.usgs.util.Config;
  27. import gov.usgs.util.StreamUtils;
  28. import gov.usgs.util.XmlUtils;

  30. /**
  31.  * Class to index a batch of products that have already been sent to the AWS hub.
  32.  *
  33.  * Reads a list of products to be indexed.
  34.  * Reads indexer from configuration file.
  35.  *
  36.  * For each product, fetch product information from the get_product AWS endpoint
  37.  * and call indexer.onProduct.
  38.  */
  39. public class AwsBatchIndexer implements Bootstrappable {
  40.   /** Force reindex argument */
  41.   public static final String FORCE_REINDEX_ARGUMENT = "--force";
  42.   /** Get product URL argument */
  43.   public static final String GET_PRODUCT_URL_ARGUMENT = "--getProductUrl=";
  44.   /** Argument for indexer configuration name */
  45.   public static final String INDEXER_CONFIG_NAME_ARGUMENT="--indexerConfigName=";
  46.   /** Default indexer configuration name */
  47.   public static final String INDEXER_CONFIG_NAME_DEFAULT = "indexer";

  49.   /** Argument for database driver */
  50.   public static final String DATABASE_DRIVER_ARGUMENT = "--databaseDriver=";
  51.   /** Argument for database URL */
  52.   public static final String DATABASE_URL_ARGUMENT = "--databaseUrl=";
  53.   /** Argument for indexer database */
  54.   public static final String INDEXER_DATABASE_ARGUMENT = "--indexerDatabase=";
  55.   /** Default database for indexer */
  56.   public static final String INDEXER_DATABASE_DEFAULT = "indexer";

  58.   /** Logging object. */
  59.   private static final Logger LOGGER = Logger.getLogger(AwsBatchIndexer.class.getName());

  61.   /** Executor where indexing runs. */
  62.   private ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);

  64.   /** Whether to force indexing. */
  65.   private boolean force = false;

  67.   /** AWS URL for get_product endpoint, with placeholders. */
  68.   private String getProductUrlTemplate = "https://earthquake.usgs.gov/pdl/west"
  69.       + "/get_product?source={source}&type={type}&code={code}&updateTime={updateTime}";

  71.   /** Indexer to process products. */
  72.   private Indexer indexer;


  75.   @Override
  76.   public void run(String[] args) throws Exception {
  77.     // parse arguments
  78.     String databaseDriver = "com.mysql.jdbc.Driver";
  79.     String databaseUrl = null;
  80.     String indexerConfigName = INDEXER_CONFIG_NAME_DEFAULT;
  81.     for (final String arg : args) {
  82.       if (arg.startsWith(DATABASE_DRIVER_ARGUMENT)) {
  83.         databaseDriver = arg.replace(DATABASE_DRIVER_ARGUMENT, "");
  84.       } else if (arg.startsWith(DATABASE_URL_ARGUMENT)) {
  85.         databaseUrl = arg.replace(DATABASE_URL_ARGUMENT, "");
  86.       } else if (arg.equals(FORCE_REINDEX_ARGUMENT)) {
  87.         force = true;
  88.       } else if (arg.startsWith(GET_PRODUCT_URL_ARGUMENT)) {
  89.         getProductUrlTemplate = arg.replace(GET_PRODUCT_URL_ARGUMENT, "");
  90.       } else if (arg.startsWith(INDEXER_CONFIG_NAME_ARGUMENT)) {
  91.         indexerConfigName = arg.replace(INDEXER_CONFIG_NAME_ARGUMENT, "");
  92.       }
  93.     }

  95.     // load indexer from configuration
  96.     indexer = (Indexer) Config.getConfig().getObject(indexerConfigName);
  97.     indexer.startup();

  99.     try {
  100.       if (databaseUrl != null) {
  101.         LOGGER.info("Reading product ids from database");
  102.         readProductIdsFromDatabase(databaseDriver, databaseUrl);
  103.       } else {
  104.         LOGGER.info("Reading product ids from stdin");
  105.         readProductIdsFromStdin();
  106.       }
  107.     } finally {
  108.       indexer.shutdown();
  109.     }
  110.   }

  112.   /**
  113.    * Use getProductUrl template to generate URL.
  114.    *
  115.    * Replace "{source}", "{type}", "{code}", and "{updateTime}" placeholders.
  116.    *
  117.    * @param id
  118.    *     which product.
  119.    * @return URL with placeholders replaced.
  120.    * @throws Exception Exception
  121.    */
  122.   public URL getProductUrl(final ProductId id) throws Exception {
  123.     String url = getProductUrlTemplate;
  124.     url = url.replace("{source}", id.getSource());
  125.     url = url.replace("{type}", id.getType());
  126.     url = url.replace("{code}", id.getCode());
  127.     url = url.replace("{updateTime}", XmlUtils.formatDate(id.getUpdateTime()));
  128.     return new URL(url);
  129.   }

  131.   /**
  132.    * Get Product from endpoint.
  133.    *
  134.    * @param id
  135.    *     which product.
  136.    * @return Product object.
  137.    * @throws Exception Exception
  138.    */
  139.   public Product getProduct(final ProductId id) throws Exception {
  140.     final URL url = getProductUrl(id);
  141.     byte[] bytes = StreamUtils.readStream(url);
  142.     try (
  143.         final JsonReader reader = Json.createReader(new StringReader(
  144.             new String(bytes, StandardCharsets.UTF_8)))
  145.     ) {
  146.       // parse message
  147.       final JsonObject json = reader.readObject();
  148.       final JsonNotification notification = new JsonNotification(json);
  149.       return notification.product;
  150.     }
  151.   }

  153.   /**
  154.    * Fetch and Index a product.
  155.    *
  156.    * Called from executor service to process product ids.
  157.    *
  158.    * @param id
  159.    *     which product
  160.    */
  161.   public void processProductId(final ProductId id) {
  162.     long start = new Date().getTime();
  163.     try {
  164.       final Product product = getProduct(id);
  165.       long afterGetProduct = new Date().getTime();
  166.       LOGGER.fine("Loaded product " + id.toString() + " in "
  167.           + (afterGetProduct - start) + " ms");

  169.       indexer.onProduct(product, force);
  170.       LOGGER.info("Indexed " + id.toString()
  171.           + " in " + (new Date().getTime() - afterGetProduct) + " ms");
  172.     } catch (Exception e) {
  173.       LOGGER.log(
  174.           Level.WARNING,
  175.           "Error indexing " + id.toString()
  176.               + " in " + (new Date().getTime() - start) + "ms",
  177.           e);
  178.     }
  179.   }

  181.   /**
  182.    * Read product ids (as urns) from database and submit to executor for processing.
  183.    *
  184.    * @param driver database driver
  185.    * @param url database url
  186.    *
  187.    * @throws Exception exception
  188.    */
  189.   public void readProductIdsFromDatabase(
  190.       final String driver,
  191.       final String url) throws Exception {
  192.     try (
  193.       final JDBCConnection jdbcConnection = new JDBCConnection()
  194.     ) {
  195.       jdbcConnection.setDriver(driver);
  196.       jdbcConnection.setUrl(url);
  197.       jdbcConnection.startup();

  199.       final String sql = "SELECT id, source, type, code, updatetime"
  200.           + " FROM pdl.product h"
  201.           + " WHERE id > ?"
  202.           + " AND NOT EXISTS ("
  203.           + "  SELECT * FROM indexer.productSummary i"
  204.           + "  WHERE h.source=i.source"
  205.           + "  AND h.type=i.type"
  206.           + "  AND h.code=i.code"
  207.           + "  AND h.updatetime=i.updateTime"
  208.           + " )"
  209.           + " ORDER BY id"
  210.           + " LIMIT 500";

  212.       // start at the beginning
  213.       long lastId = -1;
  214.       while (true) {
  215.         try (
  216.           final Connection conn = jdbcConnection.verifyConnection();
  217.           final PreparedStatement statement = conn.prepareStatement(sql);
  218.         ) {
  219.           // load next batch of products
  220.           statement.setLong(1, lastId);
  221.           try (
  222.             final ResultSet rs = statement.executeQuery();
  223.           ) {
  224.             int count = 0;
  225.             while (rs.next()) {
  226.               lastId = rs.getLong("id");
  227.               final ProductId id = new ProductId(
  228.                   rs.getString("source"),
  229.                   rs.getString("type"),
  230.                   rs.getString("code"),
  231.                   new Date(rs.getLong("updatetime")));
  232.               submitProductId(id);
  233.               count++;
  234.             }

  236.             // exit once all products processed
  237.             if (count == 0) {
  238.               LOGGER.info("No more rows returned, exiting");
  239.               break;
  240.             }
  241.           }
  242.         }
  243.       }
  244.     }
  245.   }

  247.   /**
  248.    * Read product ids (as urns) from stdin and submit to executor for processing.
  249.    *
  250.    * @throws Exception Exception
  251.    */
  252.   public void readProductIdsFromStdin() throws Exception {
  253.     // read product ids from stdin
  254.     BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
  255.     String line = null;
  256.     while ((line = br.readLine()) != null) {
  257.       if (line.equals("")) {
  258.         continue;
  259.       }
  260.       // parse product id
  261.       final ProductId id;
  262.       try {
  263.         id = ProductId.parse(line);
  264.       } catch (Exception e) {
  265.         LOGGER.warning("Error parsing product id '" + line + "'");
  266.         continue;
  267.       }
  268.       submitProductId(id);
  269.     }
  270.   }

  272.   /**
  273.    * Submit a product id to the executor service for processing.
  274.    *
  275.    * If queue is too large (500 ids), blocks until queue is smaller (100 ids).
  276.    *
  277.    * @param id
  278.    *     which product
  279.    * @throws InterruptedException InterruptedException
  280.    */
  281.   public void submitProductId(final ProductId id) throws InterruptedException {
  282.     // queue for processing
  283.     executor.submit(() -> processProductId(id));

  285.     // keep queue size smallish
  286.     if (executor.getQueue().size() > 500) {
  287.       while (executor.getQueue().size() > 100) {
  288.         LOGGER.info("Queue size " + executor.getQueue().size());
  289.         Thread.sleep(5000L);
  290.       }
  291.     }
  292.   }
  293. }
Created with JaCoCo 0.8.8.202204050719