0.10/crawler/src/main/java/org/apache/oodt/cas/crawl/ProductCrawler.java - oodt - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.oodt.cas.crawl;

 //OODT imports
 import org.apache.oodt.cas.crawl.action.CrawlerAction;
 import org.apache.oodt.cas.crawl.action.CrawlerActionRepo;
 import org.apache.oodt.cas.crawl.config.ProductCrawlerBean;
 import org.apache.oodt.cas.crawl.status.IngestStatus;
 import org.apache.oodt.cas.filemgr.ingest.Ingester;
 import org.apache.oodt.cas.filemgr.ingest.StdIngester;
 import org.apache.oodt.cas.filemgr.metadata.CoreMetKeys;
 import org.apache.oodt.cas.metadata.Metadata;

 import com.google.common.annotations.VisibleForTesting;

 //JDK imports
 import java.io.File;
 import java.io.FileFilter;
 import java.net.URL;
 import java.util.Collections;
 import java.util.List;
 import java.util.Stack;
 import java.util.Vector;
 import java.util.logging.Level;
 import java.util.logging.Logger;

 /**
  * An abstract base class for Product Crawling. This class provides methods to
  * communicate with the file manager and parse met files that show how to ingest
  * a particular Product into the File Manager.
  *
  * @author mattmann (Chris Mattmann)
  * @author bfoster (Brian Foster)
  */
 public abstract class ProductCrawler extends ProductCrawlerBean {

    /* our log stream */
    protected static Logger LOG = Logger.getLogger(ProductCrawler.class
          .getName());

    // filter to only find directories when doing a listFiles
    protected static FileFilter DIR_FILTER = new FileFilter() {
       public boolean accept(File file) {
          return file.isDirectory();
       }
    };

    // filter to only find product files, not met files
    protected static FileFilter FILE_FILTER = new FileFilter() {
       public boolean accept(File file) {
          return file.isFile();
       }
    };

    protected List<IngestStatus> ingestStatus = new Vector<IngestStatus>();
    protected CrawlerActionRepo actionRepo;
    protected Ingester ingester;

    public void crawl() {
       crawl(new File(getProductPath()));
    }

    public void crawl(File dirRoot) {
       // Reset ingest status.
       ingestStatus.clear();

       // Load actions.
       loadAndValidateActions();

       // Create Ingester.
       setupIngester();

       // Verify valid crawl directory.
       if (dirRoot == null || !dirRoot.exists()) {
          throw new IllegalArgumentException("dir root is null or non existant!");
       }

       // Start crawling.
       Stack<File> stack = new Stack<File>();
       stack.push(dirRoot.isDirectory() ? dirRoot : dirRoot.getParentFile());
       while (!stack.isEmpty()) {
          File dir = (File) stack.pop();
          LOG.log(Level.INFO, "Crawling " + dir);

          File[] productFiles = null;
          if (isCrawlForDirs()) {
             productFiles = dir.listFiles(DIR_FILTER);
          } else {
             productFiles = dir.listFiles(FILE_FILTER);
          }

          for (int j = 0; j < productFiles.length; j++) {
             ingestStatus.add(handleFile(productFiles[j]));
          }

          if (!isNoRecur()) {
             File[] subdirs = dir.listFiles(DIR_FILTER);
             if (subdirs != null) {
                for (int j = 0; j < subdirs.length; j++) {
                   stack.push(subdirs[j]);
                }
             }
          }
       }
    }

    public IngestStatus handleFile(File product) {
       LOG.log(Level.INFO, "Handling file " + product);

       // Check preconditions.
       if (!passesPreconditions(product)) {
          LOG.log(Level.WARNING,
                "Failed to pass preconditions for ingest of product: ["
                      + product.getAbsolutePath() + "]");
          return createIngestStatus(product,
                IngestStatus.Result.PRECONDS_FAILED,
                "Failed to pass preconditions");
       }

       // Generate Metadata for product.
       Metadata productMetadata = new Metadata();
       productMetadata.addMetadata(getGlobalMetadata());
       try {
          productMetadata.replaceMetadata(getMetadataForProduct(product));
       } catch (Exception e) {
          LOG.log(Level.SEVERE,
                "Failed to get metadata for product : " + e.getMessage(), e);
          performPostIngestOnFailActions(product, productMetadata);
          return createIngestStatus(product,
                IngestStatus.Result.FAILURE,
                "Failed to get metadata for product : " + e.getMessage());
       }

       // Rename the product.
       try {
          product = renameProduct(product, productMetadata);
       } catch (Exception e) {
          LOG.log(Level.SEVERE,
                "Failed to rename product : " + e.getMessage(), e);
          performPostIngestOnFailActions(product, productMetadata);
          return createIngestStatus(product, IngestStatus.Result.FAILURE,
                "Failed to rename product : " + e.getMessage());
       }

       // Set known metadata if not already specified.
       addKnownMetadata(product, productMetadata);

       // Check that metadata contains required metadata.
       if (!containsRequiredMetadata(productMetadata)) {
          LOG.log(Level.SEVERE, "Missing required metadata for product '"
                + product + "'");
          performPostIngestOnFailActions(product, productMetadata);
          return createIngestStatus(product, IngestStatus.Result.FAILURE,
                "Missing required metadata");
       }

       // Run preIngest actions.
       if (!performPreIngestActions(product, productMetadata)) {
          performPostIngestOnFailActions(product, productMetadata);
          return createIngestStatus(product, IngestStatus.Result.FAILURE,
             "PreIngest actions failed to complete");
       }

       // Check if ingest has been turned off.
       if (isSkipIngest()) {
          LOG.log(Level.INFO, "Skipping ingest of product: ["
                + product.getAbsolutePath() + "]");
          return createIngestStatus(product, IngestStatus.Result.SKIPPED,
                "Crawler ingest turned OFF");
       }

       // Ingest product.
       boolean ingestSuccess = ingest(product, productMetadata);

       // On Successful Ingest.
       if (ingestSuccess) {
          LOG.log(Level.INFO, "Successful ingest of product: ["
                + product.getAbsolutePath() + "]");
          performPostIngestOnSuccessActions(product, productMetadata);
          return createIngestStatus(product,
                IngestStatus.Result.SUCCESS, "Ingest was successful");

       // On Failed Ingest.
       } else {
          LOG.log(Level.WARNING, "Failed to ingest product: ["
                + product.getAbsolutePath()
                + "]: performing postIngestFail actions");
          performPostIngestOnFailActions(product, productMetadata);
          return createIngestStatus(product, IngestStatus.Result.FAILURE,
                "Failed to ingest product");
       }
    }

    public List<IngestStatus> getIngestStatus() {
       return Collections.unmodifiableList(ingestStatus);
    }

    protected abstract boolean passesPreconditions(File product);

    protected abstract Metadata getMetadataForProduct(File product)
          throws Exception;

    protected abstract File renameProduct(File product, Metadata productMetadata)
          throws Exception;

    @VisibleForTesting void setupIngester() {
       ingester = new StdIngester(getClientTransferer());
    }

    @VisibleForTesting void loadAndValidateActions() {
       if (actionRepo == null && getApplicationContext() != null) {
          actionRepo = new CrawlerActionRepo();
          actionRepo.loadActionsFromBeanFactory(
                getApplicationContext(), getActionIds());
          validateActions();
       }
    }

    @VisibleForTesting void validateActions() {
       StringBuffer actionErrors = new StringBuffer("");
       for (CrawlerAction action : actionRepo.getActions()) {
          try {
             action.validate();
          } catch (Exception e) {
             actionErrors.append(" " + action.getId() + ": " + e.getMessage()
                   + "\n");
          }
       }
       if (actionErrors.length() > 0) {
          throw new RuntimeException("Actions failed validation:\n"
                + actionErrors);
       }
    }

    @VisibleForTesting synchronized boolean containsRequiredMetadata(
          Metadata productMetadata) {
       for (String reqMetKey : getRequiredMetadata()) {
          if (!productMetadata.containsKey(reqMetKey)) {
             LOG.log(Level.WARNING, "Missing required metadata field "
                   + reqMetKey);
             return false;
          }
       }
       return true;
    }

    @VisibleForTesting void addKnownMetadata(File product,
          Metadata productMetadata) {
       // Add ProductName if not specified.
       if (!productMetadata.containsKey(PRODUCT_NAME)) {
          productMetadata.addMetadata(PRODUCT_NAME, product.getName());
       }
       // Add Filename if not specified.
       if (!productMetadata.containsKey(FILENAME)) {
          productMetadata.addMetadata(FILENAME, product.getName());
       }
       // Add FileLocation if not specified.
       if (!productMetadata.containsKey(FILE_LOCATION)) {
          productMetadata.addMetadata(FILE_LOCATION, product
                .getAbsoluteFile().getParentFile().getAbsolutePath());
       }
       // Add FileSize if not specified
       if (!productMetadata.containsKey(FILE_SIZE)) {
          productMetadata.addMetadata(FILE_SIZE,
                Long.toString(product.length()));
       }
    }

    @VisibleForTesting IngestStatus createIngestStatus(final File product,
          final IngestStatus.Result result, final String message) {
       return new IngestStatus() {
          public File getProduct() {
             return product;
          }
          public Result getResult() {
             return result;
          }
          public String getMessage() {
             return message;
          }
       };
    }

    @VisibleForTesting boolean ingest(File product, Metadata productMetdata) {
       try {
          LOG.log(Level.INFO, "ProductCrawler: Ready to ingest product: ["
                + product + "]: ProductType: ["
                + productMetdata.getMetadata(PRODUCT_TYPE) + "]");
          String productId = ingester.ingest(new URL(getFilemgrUrl()),
                product, productMetdata);
          LOG.log(Level.INFO, "Successfully ingested product: [" + product
                + "]: product id: " + productId);
       } catch (Exception e) {
          LOG.log(Level.WARNING,
                "ProductCrawler: Exception ingesting product: [" + product
                      + "]: Message: " + e.getMessage()
                      + ": attempting to continue crawling", e);
          return false;
       }
       return true;
    }

    @VisibleForTesting boolean performPreIngestActions(File product,
          Metadata productMetadata) {
       if (actionRepo != null) {
          return performProductCrawlerActions(
                actionRepo.getPreIngestActions(), product, productMetadata);
       } else {
          return true;
       }
    }

    @VisibleForTesting boolean performPostIngestOnSuccessActions(File product,
          Metadata productMetadata) {
       if (actionRepo != null) {
          return performProductCrawlerActions(
                actionRepo.getPostIngestOnSuccessActions(), product,
                productMetadata);
       } else {
          return true;
       }
    }

    @VisibleForTesting boolean performPostIngestOnFailActions(File product,
          Metadata productMetadata) {
       if (actionRepo != null) {
          return performProductCrawlerActions(
                actionRepo.getPostIngestOnFailActions(), product,
                productMetadata);
       } else {
          return true;
       }
    }

    @VisibleForTesting boolean performProductCrawlerActions(
          List<CrawlerAction> actions, File product, Metadata productMetadata) {
       boolean allSucceeded = true;
       for (CrawlerAction action : actions) {
          try {
             LOG.log(Level.INFO, "Performing action (id = " + action.getId()
                   + " : description = " + action.getDescription() + ")");
             if (!action.performAction(product, productMetadata)) {
                throw new Exception("Action (id = " + action.getId()
                      + " : description = " + action.getDescription()
                      + ") returned false");
             }
          } catch (Exception e) {
             allSucceeded = false;
             LOG.log(Level.WARNING,
                   "Failed to perform crawler action : " + e.getMessage(), e);
          }
       }
       return allSucceeded;
    }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.oodt.cas.crawl;

	//OODT imports
	import org.apache.oodt.cas.crawl.action.CrawlerAction;
	import org.apache.oodt.cas.crawl.action.CrawlerActionRepo;
	import org.apache.oodt.cas.crawl.config.ProductCrawlerBean;
	import org.apache.oodt.cas.crawl.status.IngestStatus;
	import org.apache.oodt.cas.filemgr.ingest.Ingester;
	import org.apache.oodt.cas.filemgr.ingest.StdIngester;
	import org.apache.oodt.cas.filemgr.metadata.CoreMetKeys;
	import org.apache.oodt.cas.metadata.Metadata;

	import com.google.common.annotations.VisibleForTesting;

	//JDK imports
	import java.io.File;
	import java.io.FileFilter;
	import java.net.URL;
	import java.util.Collections;
	import java.util.List;
	import java.util.Stack;
	import java.util.Vector;
	import java.util.logging.Level;
	import java.util.logging.Logger;

	/**
	* An abstract base class for Product Crawling. This class provides methods to
	* communicate with the file manager and parse met files that show how to ingest
	* a particular Product into the File Manager.
	*
	* @author mattmann (Chris Mattmann)
	* @author bfoster (Brian Foster)
	*/
	public abstract class ProductCrawler extends ProductCrawlerBean {

	/* our log stream */
	protected static Logger LOG = Logger.getLogger(ProductCrawler.class
	.getName());

	// filter to only find directories when doing a listFiles
	protected static FileFilter DIR_FILTER = new FileFilter() {
	public boolean accept(File file) {
	return file.isDirectory();
	}
	};

	// filter to only find product files, not met files
	protected static FileFilter FILE_FILTER = new FileFilter() {
	public boolean accept(File file) {
	return file.isFile();
	}
	};

	protected List<IngestStatus> ingestStatus = new Vector<IngestStatus>();
	protected CrawlerActionRepo actionRepo;
	protected Ingester ingester;

	public void crawl() {
	crawl(new File(getProductPath()));
	}

	public void crawl(File dirRoot) {
	// Reset ingest status.
	ingestStatus.clear();

	// Load actions.
	loadAndValidateActions();

	// Create Ingester.
	setupIngester();

	// Verify valid crawl directory.
	if (dirRoot == null \|\| !dirRoot.exists()) {
	throw new IllegalArgumentException("dir root is null or non existant!");
	}

	// Start crawling.
	Stack<File> stack = new Stack<File>();
	stack.push(dirRoot.isDirectory() ? dirRoot : dirRoot.getParentFile());
	while (!stack.isEmpty()) {
	File dir = (File) stack.pop();
	LOG.log(Level.INFO, "Crawling " + dir);

	File[] productFiles = null;
	if (isCrawlForDirs()) {
	productFiles = dir.listFiles(DIR_FILTER);
	} else {
	productFiles = dir.listFiles(FILE_FILTER);
	}

	for (int j = 0; j < productFiles.length; j++) {
	ingestStatus.add(handleFile(productFiles[j]));
	}

	if (!isNoRecur()) {
	File[] subdirs = dir.listFiles(DIR_FILTER);
	if (subdirs != null) {
	for (int j = 0; j < subdirs.length; j++) {
	stack.push(subdirs[j]);
	}
	}
	}
	}
	}

	public IngestStatus handleFile(File product) {
	LOG.log(Level.INFO, "Handling file " + product);

	// Check preconditions.
	if (!passesPreconditions(product)) {
	LOG.log(Level.WARNING,
	"Failed to pass preconditions for ingest of product: ["
	+ product.getAbsolutePath() + "]");
	return createIngestStatus(product,
	IngestStatus.Result.PRECONDS_FAILED,
	"Failed to pass preconditions");
	}

	// Generate Metadata for product.
	Metadata productMetadata = new Metadata();
	productMetadata.addMetadata(getGlobalMetadata());
	try {
	productMetadata.replaceMetadata(getMetadataForProduct(product));
	} catch (Exception e) {
	LOG.log(Level.SEVERE,
	"Failed to get metadata for product : " + e.getMessage(), e);
	performPostIngestOnFailActions(product, productMetadata);
	return createIngestStatus(product,
	IngestStatus.Result.FAILURE,
	"Failed to get metadata for product : " + e.getMessage());
	}

	// Rename the product.
	try {
	product = renameProduct(product, productMetadata);
	} catch (Exception e) {
	LOG.log(Level.SEVERE,
	"Failed to rename product : " + e.getMessage(), e);
	performPostIngestOnFailActions(product, productMetadata);
	return createIngestStatus(product, IngestStatus.Result.FAILURE,
	"Failed to rename product : " + e.getMessage());
	}

	// Set known metadata if not already specified.
	addKnownMetadata(product, productMetadata);

	// Check that metadata contains required metadata.
	if (!containsRequiredMetadata(productMetadata)) {
	LOG.log(Level.SEVERE, "Missing required metadata for product '"
	+ product + "'");
	performPostIngestOnFailActions(product, productMetadata);
	return createIngestStatus(product, IngestStatus.Result.FAILURE,
	"Missing required metadata");
	}

	// Run preIngest actions.
	if (!performPreIngestActions(product, productMetadata)) {
	performPostIngestOnFailActions(product, productMetadata);
	return createIngestStatus(product, IngestStatus.Result.FAILURE,
	"PreIngest actions failed to complete");
	}

	// Check if ingest has been turned off.
	if (isSkipIngest()) {
	LOG.log(Level.INFO, "Skipping ingest of product: ["
	+ product.getAbsolutePath() + "]");
	return createIngestStatus(product, IngestStatus.Result.SKIPPED,
	"Crawler ingest turned OFF");
	}

	// Ingest product.
	boolean ingestSuccess = ingest(product, productMetadata);

	// On Successful Ingest.
	if (ingestSuccess) {
	LOG.log(Level.INFO, "Successful ingest of product: ["
	+ product.getAbsolutePath() + "]");
	performPostIngestOnSuccessActions(product, productMetadata);
	return createIngestStatus(product,
	IngestStatus.Result.SUCCESS, "Ingest was successful");

	// On Failed Ingest.
	} else {
	LOG.log(Level.WARNING, "Failed to ingest product: ["
	+ product.getAbsolutePath()
	+ "]: performing postIngestFail actions");
	performPostIngestOnFailActions(product, productMetadata);
	return createIngestStatus(product, IngestStatus.Result.FAILURE,
	"Failed to ingest product");
	}
	}

	public List<IngestStatus> getIngestStatus() {
	return Collections.unmodifiableList(ingestStatus);
	}

	protected abstract boolean passesPreconditions(File product);

	protected abstract Metadata getMetadataForProduct(File product)
	throws Exception;

	protected abstract File renameProduct(File product, Metadata productMetadata)
	throws Exception;

	@VisibleForTesting void setupIngester() {
	ingester = new StdIngester(getClientTransferer());
	}

	@VisibleForTesting void loadAndValidateActions() {
	if (actionRepo == null && getApplicationContext() != null) {
	actionRepo = new CrawlerActionRepo();
	actionRepo.loadActionsFromBeanFactory(
	getApplicationContext(), getActionIds());
	validateActions();
	}
	}

	@VisibleForTesting void validateActions() {
	StringBuffer actionErrors = new StringBuffer("");
	for (CrawlerAction action : actionRepo.getActions()) {
	try {
	action.validate();
	} catch (Exception e) {
	actionErrors.append(" " + action.getId() + ": " + e.getMessage()
	+ "\n");
	}
	}
	if (actionErrors.length() > 0) {
	throw new RuntimeException("Actions failed validation:\n"
	+ actionErrors);
	}
	}

	@VisibleForTesting synchronized boolean containsRequiredMetadata(
	Metadata productMetadata) {
	for (String reqMetKey : getRequiredMetadata()) {
	if (!productMetadata.containsKey(reqMetKey)) {
	LOG.log(Level.WARNING, "Missing required metadata field "
	+ reqMetKey);
	return false;
	}
	}
	return true;
	}

	@VisibleForTesting void addKnownMetadata(File product,
	Metadata productMetadata) {
	// Add ProductName if not specified.
	if (!productMetadata.containsKey(PRODUCT_NAME)) {
	productMetadata.addMetadata(PRODUCT_NAME, product.getName());
	}
	// Add Filename if not specified.
	if (!productMetadata.containsKey(FILENAME)) {
	productMetadata.addMetadata(FILENAME, product.getName());
	}
	// Add FileLocation if not specified.
	if (!productMetadata.containsKey(FILE_LOCATION)) {
	productMetadata.addMetadata(FILE_LOCATION, product
	.getAbsoluteFile().getParentFile().getAbsolutePath());
	}
	// Add FileSize if not specified
	if (!productMetadata.containsKey(FILE_SIZE)) {
	productMetadata.addMetadata(FILE_SIZE,
	Long.toString(product.length()));
	}
	}

	@VisibleForTesting IngestStatus createIngestStatus(final File product,
	final IngestStatus.Result result, final String message) {
	return new IngestStatus() {
	public File getProduct() {
	return product;
	}
	public Result getResult() {
	return result;
	}
	public String getMessage() {
	return message;
	}
	};
	}

	@VisibleForTesting boolean ingest(File product, Metadata productMetdata) {
	try {
	LOG.log(Level.INFO, "ProductCrawler: Ready to ingest product: ["
	+ product + "]: ProductType: ["
	+ productMetdata.getMetadata(PRODUCT_TYPE) + "]");
	String productId = ingester.ingest(new URL(getFilemgrUrl()),
	product, productMetdata);
	LOG.log(Level.INFO, "Successfully ingested product: [" + product
	+ "]: product id: " + productId);
	} catch (Exception e) {
	LOG.log(Level.WARNING,
	"ProductCrawler: Exception ingesting product: [" + product
	+ "]: Message: " + e.getMessage()
	+ ": attempting to continue crawling", e);
	return false;
	}
	return true;
	}

	@VisibleForTesting boolean performPreIngestActions(File product,
	Metadata productMetadata) {
	if (actionRepo != null) {
	return performProductCrawlerActions(
	actionRepo.getPreIngestActions(), product, productMetadata);
	} else {
	return true;
	}
	}

	@VisibleForTesting boolean performPostIngestOnSuccessActions(File product,
	Metadata productMetadata) {
	if (actionRepo != null) {
	return performProductCrawlerActions(
	actionRepo.getPostIngestOnSuccessActions(), product,
	productMetadata);
	} else {
	return true;
	}
	}

	@VisibleForTesting boolean performPostIngestOnFailActions(File product,
	Metadata productMetadata) {
	if (actionRepo != null) {
	return performProductCrawlerActions(
	actionRepo.getPostIngestOnFailActions(), product,
	productMetadata);
	} else {
	return true;
	}
	}

	@VisibleForTesting boolean performProductCrawlerActions(
	List<CrawlerAction> actions, File product, Metadata productMetadata) {
	boolean allSucceeded = true;
	for (CrawlerAction action : actions) {
	try {
	LOG.log(Level.INFO, "Performing action (id = " + action.getId()
	+ " : description = " + action.getDescription() + ")");
	if (!action.performAction(product, productMetadata)) {
	throw new Exception("Action (id = " + action.getId()
	+ " : description = " + action.getDescription()
	+ ") returned false");
	}
	} catch (Exception e) {
	allSucceeded = false;
	LOG.log(Level.WARNING,
	"Failed to perform crawler action : " + e.getMessage(), e);
	}
	}
	return allSucceeded;
	}
	}