Metacat

/**

 *  '$RCSfile$'

 *  Copyright: 2000-2011 Regents of the University of California and the

 *              National Center for Ecological Analysis and Synthesis

 *

 *   '$Author:  $'

 *     '$Date:  $'

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat.dataone;

import java.io.IOException;

import java.io.InputStream;

import java.security.NoSuchAlgorithmException;

import java.sql.SQLException;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.List;

import java.util.Timer;

import org.apache.commons.io.IOUtils;

import org.apache.log4j.Logger;

import org.dataone.client.D1Client;

import org.dataone.client.MNode;

import org.dataone.service.util.Constants;

import org.dataone.service.exceptions.IdentifierNotUnique;

import org.dataone.service.exceptions.InsufficientResources;

import org.dataone.service.exceptions.InvalidRequest;

import org.dataone.service.exceptions.InvalidSystemMetadata;

import org.dataone.service.exceptions.InvalidToken;

import org.dataone.service.exceptions.NotAuthorized;

import org.dataone.service.exceptions.NotFound;

import org.dataone.service.exceptions.NotImplemented;

import org.dataone.service.exceptions.ServiceFailure;

import org.dataone.service.exceptions.SynchronizationFailed;

import org.dataone.service.exceptions.UnsupportedType;

import org.dataone.service.mn.tier1.v1.MNCore;

import org.dataone.service.mn.tier1.v1.MNRead;

import org.dataone.service.mn.tier2.v1.MNAuthorization;

import org.dataone.service.mn.tier3.v1.MNStorage;

import org.dataone.service.mn.tier4.v1.MNReplication;

import org.dataone.service.types.v1.Checksum;

import org.dataone.service.types.v1.DescribeResponse;

import org.dataone.service.types.v1.Event;

import org.dataone.service.types.v1.Group;

import org.dataone.service.types.v1.Identifier;

import org.dataone.service.types.v1.Log;

import org.dataone.service.types.v1.LogEntry;

import org.dataone.service.types.v1.MonitorInfo;

import org.dataone.service.types.v1.MonitorList;

import org.dataone.service.types.v1.Node;

import org.dataone.service.types.v1.NodeReference;

import org.dataone.service.types.v1.NodeState;

import org.dataone.service.types.v1.NodeType;

import org.dataone.service.types.v1.ObjectFormatIdentifier;

import org.dataone.service.types.v1.ObjectList;

import org.dataone.service.types.v1.Permission;

import org.dataone.service.types.v1.Ping;

import org.dataone.service.types.v1.Schedule;

import org.dataone.service.types.v1.Service;

import org.dataone.service.types.v1.Services;

import org.dataone.service.types.v1.Session;

import org.dataone.service.types.v1.Subject;

import org.dataone.service.types.v1.Synchronization;

import org.dataone.service.types.v1.SystemMetadata;

import org.dataone.service.types.v1.util.ChecksumUtil;

import edu.ucsb.nceas.metacat.DocumentImpl;

import edu.ucsb.nceas.metacat.EventLog;

import edu.ucsb.nceas.metacat.IdentifierManager;

import edu.ucsb.nceas.metacat.McdbDocNotFoundException;

import edu.ucsb.nceas.metacat.MetacatHandler;

import edu.ucsb.nceas.metacat.client.InsufficientKarmaException;

import edu.ucsb.nceas.metacat.database.DBConnection;

import edu.ucsb.nceas.metacat.database.DBConnectionPool;

import edu.ucsb.nceas.metacat.properties.PropertyService;

import edu.ucsb.nceas.metacat.util.SystemUtil;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

/**

 * Represents Metacat's implementation of the DataONE Member Node 

 * service API. Methods implement the various MN* interfaces, and methods common

 * to both Member Node and Coordinating Node interfaces are found in the

 * D1NodeService base class.

 * 

 * Implements:

 * MNCore.ping()

 * MNCore.getLogRecords()

 * MNCore.getObjectStatistics()

 * MNCore.getOperationStatistics()

 * MNCore.getStatus()

 * MNCore.getCapabilities()

 * MNRead.get()

 * MNRead.getSystemMetadata()

 * MNRead.describe()

 * MNRead.getChecksum()

 * MNRead.listObjects()

 * MNRead.synchronizationFailed()

 * MNAuthorization.isAuthorized()

 * MNAuthorization.setAccessPolicy()

 * MNStorage.create()

 * MNStorage.update()

 * MNStorage.delete()

 * MNReplication.replicate()

 * 

 */

public class MNodeService extends D1NodeService implements MNAuthorization,

  MNCore, MNRead, MNReplication, MNStorage {

  /* the instance of the MNodeService object */

  private static MNodeService instance = null;

  /* the logger instance */

  private Logger logMetacat = null;

  /**

   * Singleton accessor to get an instance of MNodeService.

   * 

   * @return instance - the instance of MNodeService

   */

  public static MNodeService getInstance() {

    if (instance == null) {

      instance = new MNodeService();

    }

    return instance;

  }

  /**

   * Constructor, private for singleton access

   */

  private MNodeService() {

    super();

    logMetacat = Logger.getLogger(MNodeService.class);

  }

  /**

   * Deletes an object from the Member Node, where the object is either a 

   * data object or a science metadata object.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The object identifier to be deleted

   * 

   * @return pid - the identifier of the object used for the deletion

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  //@Override

  public Identifier delete(Session session, Identifier pid) 

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, 

    NotImplemented, InvalidRequest {

    String localId = null;

    boolean allowed = false;

    String username = Constants.PUBLIC_SUBJECT;

    String[] groupnames = null;

    if (session != null ) {

    	username = session.getSubject().getValue();

    	if (session.getSubjectList() != null) {

    		List<Group> groupList = session.getSubjectList().getGroupList();

    		if (groupList != null) {

    			groupnames = new String[groupList.size()];

    			for (int i = 0; i > groupList.size(); i++ ) {

    				groupnames[i] = groupList.get(i).getGroupName();

    			}

    		}

    	}

    }

    // do we have a valid pid?

    if ( pid == null || pid.getValue().trim().equals("") ) {

      throw new InvalidRequest("1322", "The provided identifier was invalid.");

    }

    // check for the existing identifier

    try {

      localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

    } catch (McdbDocNotFoundException e) {

      throw new InvalidRequest("1322", "The object with the provided " +

        "identifier was not found.");

    }

    // does the subject have DELETE (a D1 CHANGE_PERMISSION level) priveleges on the pid?

    allowed = isAuthorized(session, pid, Permission.CHANGE_PERMISSION);

    if ( allowed ) {

      try {

        // delete the document

        DocumentImpl.delete(localId, username, groupnames, null);

        EventLog.getInstance().log(metacatUrl, username, localId, Event.DELETE.xmlValue());

      } catch (McdbDocNotFoundException e) {

        throw new InvalidRequest("1322", "The provided identifier was invalid.");

      } catch (SQLException e) {

        throw new ServiceFailure("1350", "There was a problem deleting the object." +

          "The error message was: " + e.getMessage());

      } catch (InsufficientKarmaException e) {

        throw new NotAuthorized("1320", "The provided identity does not have " +

        "permission to DELETE objects on the Member Node.");

      } catch (Exception e) { // for some reason DocumentImpl throws a general Exception

        throw new ServiceFailure("1350", "There was a problem deleting the object." +

            "The error message was: " + e.getMessage());

      }

    } else {

      throw new NotAuthorized("1320", "The provided identity does not have " +

      "permission to DELETE objects on the Member Node.");

    }

    return pid;

  }

  /**

   * Updates an existing object by creating a new object identified by 

   * newPid on the Member Node which explicitly obsoletes the object 

   * identified by pid through appropriate changes to the SystemMetadata 

   * of pid and newPid

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The identifier of the object to be updated

   * @param object - the new object bytes

   * @param sysmeta - the new system metadata describing the object

   * 

   * @return newPid - the identifier of the new object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws IdentifierNotUnique

   * @throws UnsupportedType

   * @throws InsufficientResources

   * @throws InvalidSystemMetadata

   * @throws InvalidRequest

   */

  //@Override

  public Identifier update(Session session, Identifier pid, InputStream object,

    Identifier newPid, SystemMetadata sysmeta) 

    throws InvalidToken, ServiceFailure, NotAuthorized, IdentifierNotUnique, 

    UnsupportedType, InsufficientResources, NotFound, InvalidSystemMetadata, 

    NotImplemented, InvalidRequest {

	// check if the pid has been reserved

	try {

		boolean hasReservation = D1Client.getCN().hasReservation(session, pid);

		if (!hasReservation) {

			throw new NotAuthorized("", "No reservation for pid: " + pid.getValue());

		}

	} catch (NotFound e) {

		// okay to continue

	}

    String localId = null;

    boolean allowed = false;

    boolean isScienceMetadata = false;

    Subject subject = session.getSubject();

    // do we have a valid pid?

    if ( pid == null || pid.getValue().trim().equals("") ) {

      throw new InvalidRequest("1202", "The provided identifier was invalid.");

    }

    // check for the existing identifier

    try {

      localId = IdentifierManager.getInstance().getLocalId(pid.getValue());

    } catch (McdbDocNotFoundException e) {

      throw new InvalidRequest("1202", "The object with the provided " +

        "identifier was not found.");

    }

    // does the subject have WRITE ( == update) priveleges on the pid?

    allowed = isAuthorized(session, pid, Permission.WRITE);

    if ( allowed ) {

      // get the existing system metadata for the object

      SystemMetadata existingSysMeta = getSystemMetadata(session, pid);

      // add the newPid to the obsoletedBy list for the existing sysmeta

      existingSysMeta.setObsoletedBy(newPid);

      // then update the existing system metadata

      updateSystemMetadata(existingSysMeta);

      // prep the new system metadata, add pid to the affected lists

      sysmeta.setObsoletes(pid);

      //sysmeta.addDerivedFrom(pid);

      isScienceMetadata = isScienceMetadata(sysmeta);

      // do we have XML metadata or a data object?

      if ( isScienceMetadata ) {

        // update the science metadata XML document

        // TODO: handle non-XML metadata/data documents (like netCDF)

        // TODO: don't put objects into memory using stream to string

        String objectAsXML = "";

        try {

          objectAsXML = IOUtils.toString(object, "UTF-8");

          localId = insertOrUpdateDocument(objectAsXML, newPid, session, "update");

          // register the newPid and the generated localId

          if ( newPid != null ) {

        	  IdentifierManager.getInstance().createMapping(newPid.getValue(), localId);

          }

        } catch (IOException e) {

          String msg = "The Node is unable to create the object. " +

          "There was a problem converting the object to XML";

          logMetacat.info(msg);

          throw new ServiceFailure("1310", msg + ": " + e.getMessage());

        }

      } else {

        // update the data object

        localId = insertDataObject(object, newPid, session);

      }

      // and insert the new system metadata

      insertSystemMetadata(sysmeta);

      // log the update event

      EventLog.getInstance().log(metacatUrl, subject.getValue(), localId, "update");

    } else {

      throw new NotAuthorized("1200", "The provided identity does not have " +

      "permission to UPDATE the object identified by " +

      pid.getValue() + " on the Member Node.");

    }

    return newPid;

  }

  public Identifier create(Session session, Identifier pid,

			InputStream object, SystemMetadata sysmeta) throws InvalidToken,

			ServiceFailure, NotAuthorized, IdentifierNotUnique,

			UnsupportedType, InsufficientResources, InvalidSystemMetadata,

			NotImplemented, InvalidRequest {

		// check if the pid has been reserved

		try {

			boolean hasReservation = D1Client.getCN().hasReservation(session, pid);

			if (!hasReservation) {

				throw new NotAuthorized("", "No reservation for pid: " + pid.getValue());

			}

		} catch (NotFound e) {

			// okay to continue

		}

		return super.create(session, pid, object, sysmeta);

	}

  /**

   * Called by a Coordinating Node to request that the Member Node create a 

   * copy of the specified object by retrieving it from another Member 

   * Node and storing it locally so that it can be made accessible to 

   * the DataONE system.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param sysmeta - Copy of the CN held system metadata for the object

   * @param sourceNode - A reference to node from which the content should be 

   *                     retrieved. The reference should be resolved by 

   *                     checking the CN node registry.

   * 

   * @return true if the replication succeeds

   * 

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotImplemented

   * @throws UnsupportedType

   * @throws InsufficientResources

   * @throws InvalidRequest

   */

  public boolean replicate(Session session, SystemMetadata sysmeta, 

    NodeReference sourceNode)

    throws NotImplemented, ServiceFailure, NotAuthorized, InvalidRequest,

    InsufficientResources, UnsupportedType {

    boolean result = false;

    // TODO: check credentials

    // get the referenced object

    Identifier pid = sysmeta.getIdentifier();

    // get from the membernode

    // TODO: switch credentials for the server retrieval?

    MNode mn = D1Client.getMN(sourceNode);

    InputStream object = null;

	try {

		object = mn.get(session, pid);

	} catch (InvalidToken e) {

		e.printStackTrace();

		throw new ServiceFailure("2151", "Could not retrieve object to replicate (InvalidToken): " + e.getMessage());

	} catch (NotFound e) {

		e.printStackTrace();

		throw new ServiceFailure("2151", "Could not retrieve object to replicate (NotFound): " + e.getMessage());

	}

    // add it to local store

	Identifier retPid;

	try {

		retPid = create(session, pid, object, sysmeta);

		result = (retPid.getValue().equals(pid.getValue()));

	} catch (InvalidToken e) {

		e.printStackTrace();

		throw new ServiceFailure("2151", "Could not save object to local store (InvalidToken): " + e.getMessage());

	} catch (IdentifierNotUnique e) {

		e.printStackTrace();

		throw new ServiceFailure("2151", "Could not save object to local store (IdentifierNotUnique): " + e.getMessage());

	} catch (InvalidSystemMetadata e) {

		e.printStackTrace();

		throw new ServiceFailure("2151", "Could not save object to local store (InvalidSystemMetadata): " + e.getMessage());

	}

    return result;

  }

  /**

   * This method provides a lighter weight mechanism than 

   * MN_read.getSystemMetadata() for a client to determine basic 

   * properties of the referenced object.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the identifier of the object to be described

   * 

   * @return describeResponse - A set of values providing a basic description 

   *                            of the object.

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  public DescribeResponse describe(Session session, Identifier pid)

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound,

    NotImplemented, InvalidRequest {

    if(session == null) {

      throw new InvalidToken("1370", "The session object is null");

    }

    if(pid == null || pid.getValue().trim().equals(""))

    {

      throw new InvalidRequest("1362", "The object identifier is null. " +

        "A valid identifier is required.");

    }

    SystemMetadata sysmeta = getSystemMetadata(session, pid);

    DescribeResponse describeResponse = 

      new DescribeResponse(sysmeta.getFmtid(), 

      sysmeta.getSize(), sysmeta.getDateSysMetadataModified(), sysmeta.getChecksum());

    return describeResponse;

  }

  /**

   * Return the object identified by the given object identifier

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * 

   * @return inputStream - the input stream of the given object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  @Override

  public InputStream get(Session session, Identifier pid) 

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, 

    NotImplemented, InvalidRequest {

    return super.get(session, pid);

  }

  /**

   * Returns a Checksum for the specified object using an accepted hashing algorithm

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * @param algorithm -  the name of an algorithm that will be used to compute 

   *                     a checksum of the bytes of the object

   * 

   * @return checksum - the checksum of the given object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public Checksum getChecksum(Session session, Identifier pid, String algorithm)

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound,

    InvalidRequest, NotImplemented {

    Checksum checksum = null;

    InputStream inputStream = get(session, pid);

    try {

      checksum = 

    	  ChecksumUtil.checksum(inputStream, algorithm);

    } catch (NoSuchAlgorithmException e) {

      throw new ServiceFailure("1410", "The checksum for the object specified by " + 

        pid.getValue() +

        "could not be returned due to an internal error: " +

        e.getMessage());

    } catch (IOException e) {

      throw new ServiceFailure("1410", "The checksum for the object specified by " + 

        pid.getValue() +

        "could not be returned due to an internal error: " +

        e.getMessage());

    }

    if ( checksum == null ) {

      throw new ServiceFailure("1410", "The checksum for the object specified by " + 

        pid.getValue() +

        "could not be returned.");

    }

    return checksum;

  }

  /**

   * Return the system metadata for a given object

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * 

   * @return inputStream - the input stream of the given system metadata object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  @Override

  public SystemMetadata getSystemMetadata(Session session, Identifier pid)

      throws InvalidToken, ServiceFailure, NotAuthorized, NotFound,

      InvalidRequest, NotImplemented {

    return super.getSystemMetadata(session, pid);

  }

  /**

   * Retrieve the list of objects present on the MN that match the calling parameters

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param startTime - Specifies the beginning of the time range from which 

   *                    to return object (>=)

   * @param endTime - Specifies the beginning of the time range from which 

   *                  to return object (>=)

   * @param objectFormat - Restrict results to the specified object format

   * @param replicaStatus - Indicates if replicated objects should be returned in the list

   * @param start - The zero-based index of the first value, relative to the 

   *                first record of the resultset that matches the parameters.

   * @param count - The maximum number of entries that should be returned in 

   *                the response. The Member Node may return less entries 

   *                than specified in this value.

   * 

   * @return objectList - the list of objects matching the criteria

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public ObjectList listObjects(Session session, Date startTime, Date endTime,

    ObjectFormatIdentifier objectFormatId, Boolean replicaStatus, Integer start, Integer count)

    throws NotAuthorized, InvalidRequest, NotImplemented, ServiceFailure,

    InvalidToken {

    ObjectList objectList = null;

    try {

	    objectList = IdentifierManager.getInstance().querySystemMetadata(startTime, endTime,

	        objectFormatId, replicaStatus, start, count);

    } catch (Exception e) {

		throw new ServiceFailure("1580", "Error querying system metadata: " + e.getMessage());

	}

    return objectList;

  }

  /**

   * Retrieve the list of objects present on the MN that match the calling parameters

   * 

   * @return node - the technical capabilities of the Member Node

   * 

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public Node getCapabilities() throws NotImplemented, NotAuthorized,

      ServiceFailure, InvalidRequest {

  	String nodeName = null;

    String nodeId = null;

    String nodeUrl = null;

    String nodeDesc = null;

    String nodeType = null;

    String mnCoreServiceVersion = null;

    String mnReadServiceVersion = null;

    String mnAuthorizationServiceVersion = null;

    String mnStorageServiceVersion = null;

    String mnReplicationServiceVersion = null;

    boolean nodeSynchronize = false;

    boolean nodeReplicate = false;

    boolean mnCoreServiceAvailable = false;

    boolean mnReadServiceAvailable = false;

    boolean mnAuthorizationServiceAvailable = false;

    boolean mnStorageServiceAvailable = false;

    boolean mnReplicationServiceAvailable = false;

    try

    {

    	// get the properties of the node based on configuration information

      nodeId = PropertyService.getProperty("dataone.memberNodeId");

      nodeName = PropertyService.getProperty("dataone.nodeName");

      nodeUrl = SystemUtil.getContextURL() + "/d1/";

      nodeDesc = PropertyService.getProperty("dataone.nodeDescription");

      nodeType = PropertyService.getProperty("dataone.nodeType");

      nodeSynchronize = 

      	new Boolean(PropertyService.getProperty(

      		"dataone.nodeSynchronize")).booleanValue();

      nodeReplicate = 

      	new Boolean(PropertyService.getProperty(

      		"dataone.nodeReplicate")).booleanValue();

      mnCoreServiceVersion = 

      	PropertyService.getProperty("dataone.mnCore.serviceVersion");

      mnReadServiceVersion = 

      	PropertyService.getProperty("dataone.mnRead.serviceVersion");

      mnAuthorizationServiceVersion = 

      	PropertyService.getProperty("dataone.mnAuthorization.serviceVersion");

      mnStorageServiceVersion = 

      	PropertyService.getProperty("dataone.mnStorage.serviceVersion");

      mnReplicationServiceVersion = 

      	PropertyService.getProperty("dataone.mnReplication.serviceVersion");

      mnCoreServiceAvailable = new Boolean(

      	PropertyService.getProperty("dataone.mnCore.serviceAvailable")).booleanValue();

      mnReadServiceAvailable =  new Boolean(

      	PropertyService.getProperty(

      		"dataone.mnRead.serviceAvailable")).booleanValue();

      mnAuthorizationServiceAvailable =  new Boolean(

      	PropertyService.getProperty(

      		"dataone.mnAuthorization.serviceAvailable")).booleanValue();

      mnStorageServiceAvailable =  new Boolean(

      	PropertyService.getProperty(

      	  "dataone.mnStorage.serviceAvailable")).booleanValue();

      mnReplicationServiceAvailable =  new Boolean(

      	PropertyService.getProperty(

      	  "dataone.mnReplication.serviceAvailable")).booleanValue();

    } catch(PropertyNotFoundException pnfe) {

        logMetacat.error("MNodeService.getCapabilities(): " +

          "property not found: " + pnfe.getMessage());

    }

  	// Set the properties of the node based on configuration information and

    // calls to current status methods

	  Node node = new Node();

	  node.setBaseURL(metacatUrl + "/" + nodeType);

	  node.setDescription(nodeDesc);

	  // set the node's health information

	  NodeState state = NodeState.UP;

	  node.setState(state);

	  // set the ping response to the current value

	  Ping canPing = new Ping();

	  canPing.setSuccess(false);

	  try {

	    canPing.setSuccess(ping());

    } catch (InsufficientResources e) {

	    e.printStackTrace();

    } catch (UnsupportedType e) {

	    e.printStackTrace();

    }

	  node.setPing(canPing);

	  NodeReference identifier = new NodeReference();

	  identifier.setValue(nodeId);

	  node.setIdentifier(identifier);

	  node.setName(nodeName + " -- WAR version WARVERSION");

	  node.setReplicate(new Boolean(nodeReplicate).booleanValue());

	  node.setSynchronize(new Boolean(nodeSynchronize).booleanValue());

	  // services: MNAuthorization, MNCore, MNRead, MNReplication, MNStorage

	  Services services = new Services();

	  Service sMNCore = new Service();

	  sMNCore.setName("MNCore");

	  sMNCore.setVersion(mnCoreServiceVersion);

	  sMNCore.setAvailable(mnCoreServiceAvailable);

	  Service sMNRead = new Service();

	  sMNRead.setName("MNRead");

	  sMNRead.setVersion(mnReadServiceVersion);

	  sMNRead.setAvailable(mnReadServiceAvailable);

	  Service sMNAuthorization = new Service();

	  sMNAuthorization.setName("MNAuthorization");

	  sMNAuthorization.setVersion(mnAuthorizationServiceVersion);

	  sMNAuthorization.setAvailable(mnAuthorizationServiceAvailable);

	  Service sMNStorage = new Service();

	  sMNStorage.setName("MNStorage");

	  sMNStorage.setVersion(mnStorageServiceVersion);

	  sMNStorage.setAvailable(mnStorageServiceAvailable);

	  Service sMNReplication = new Service();

	  sMNReplication.setName("MNReplication");

	  sMNReplication.setVersion(mnReplicationServiceVersion);

	  sMNReplication.setAvailable(mnReplicationServiceAvailable);

	  services.addService(sMNRead);

	  services.addService(sMNCore);

	  services.addService(sMNAuthorization);

	  services.addService(sMNStorage);

	  services.addService(sMNReplication);

	  node.setServices(services);

	  // TODO: Determine the synchronization info without mock values

	  Synchronization synchronization = new Synchronization();

	  Schedule schedule = new Schedule();

	  Date now = new Date();

	  schedule.setYear(new SimpleDateFormat("yyyy").format(now));

	  schedule.setMon(new SimpleDateFormat("MM").format(now));

	  schedule.setMday(new SimpleDateFormat("dd").format(now));

	  schedule.setWday(new SimpleDateFormat("dd").format(now));

	  schedule.setHour(new SimpleDateFormat("HH").format(now));

	  schedule.setMin(new SimpleDateFormat("mm").format(now));

	  schedule.setSec(new SimpleDateFormat("ss").format(now));

	  synchronization.setSchedule(schedule);

	  synchronization.setLastHarvested(now);

	  synchronization.setLastCompleteHarvest(now);

	  node.setSynchronization(synchronization);

	  node.setSynchronize(false);

	  node.setType(NodeType.MN);

	  return node;

  }

  /**

   * Returns the number of operations that have been serviced by the node 

   * over time periods of one and 24 hours.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param period - An ISO8601 compatible DateTime range specifying the time 

   *                 range for which to return operation statistics.

   * @param requestor - Limit to operations performed by given requestor identity.

   * @param event -  Enumerated value indicating the type of event being examined

   * @param format - Limit to events involving objects of the specified format

   * 

   * @return the desired log records

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public MonitorList getOperationStatistics(Session session, Date startTime,

  		Date endTime, Subject requestor, Event event, ObjectFormatIdentifier formatId) 

    throws NotImplemented, ServiceFailure, NotAuthorized, InvalidRequest, 

    InsufficientResources, UnsupportedType {

	  MonitorList monitorList = new MonitorList();

	  try {

		  // get log records first

		  Log logs = getLogRecords(session, startTime, endTime, event, 0, null);

		  // TODO: aggregate by day or hour -- needs clarification

		  int count = 1;

		  for (LogEntry logEntry: logs.getLogEntryList()) {

			  Identifier pid = logEntry.getIdentifier();

			  Date logDate = logEntry.getDateLogged();

			  // if we are filtering by format

			  if (formatId != null) {

				  SystemMetadata sysmeta = IdentifierManager.getInstance().getSystemMetadata(pid.getValue());

				  if (!sysmeta.getFmtid().getValue().equals(formatId.getValue())) {

					  // does not match

					  continue;

				  }

			  }

			  MonitorInfo item = new MonitorInfo();

			  item.setCount(count);

			  item.setDate(new java.sql.Date(logDate.getTime()));

			  monitorList.addMonitorInfo(item);

		  }

	} catch (Exception e) {

		e.printStackTrace();

		throw new ServiceFailure("2081", "Could not retrieve statistics: " + e.getMessage());

	}

	return monitorList;

  }

  /**

   * Low level “are you alive” operation. A valid ping response is 

   * indicated by a HTTP status of 200.

   * 

   * @return true if the service is alive

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public boolean ping() 

    throws NotImplemented, ServiceFailure, NotAuthorized, InvalidRequest, 

    InsufficientResources, UnsupportedType {

    // test if we can get a database connection

    boolean alive = false;

    int serialNumber = -1;

    DBConnection dbConn = null;

    try {

      dbConn = DBConnectionPool

      .getDBConnection("MNodeService.ping");

      serialNumber = dbConn.getCheckOutSerialNumber();

      alive = true;

    } catch (SQLException e) {

      return alive;

    } finally {

      // Return the database connection

      DBConnectionPool.returnDBConnection(dbConn, serialNumber);

    }

    return alive;

  }

  /**

   * A callback method used by a CN to indicate to a MN that it cannot 

   * complete synchronization of the science metadata identified by pid.  Log

   * the event in the metacat event log.

   * 

   * @param session

   * @param syncFailed

   * 

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  public void synchronizationFailed(Session session, SynchronizationFailed syncFailed)

      throws NotImplemented, ServiceFailure, NotAuthorized, InvalidRequest {

    String localId;

    try {

      localId = IdentifierManager.getInstance().getLocalId(syncFailed.getPid());

    } catch (McdbDocNotFoundException e) {

      throw new InvalidRequest("2163", "The identifier specified by " +

          syncFailed.getPid() + 

          " was not found on this node.");

    }

    // TODO: update the CN URL below when the CNRead.SynchronizationFailed

    // method is changed to include the URL as a parameter

    logMetacat.debug("Synchronization for the object identified by " +

      syncFailed.getPid() + 

      " failed from " +

      syncFailed.getNodeId() +

      " Logging the event to the Metacat EventLog as a 'syncFailed' event.");

    // TODO: use the event type enum when the SYNCHRONIZATION_FAILED event is added

    EventLog.getInstance().log(syncFailed.getNodeId(), 

      session.getSubject().getValue(), localId, "synchronization_failed");

    //EventLog.getInstance().log("CN URL WILL GO HERE", 

    //  session.getSubject().getValue(), localId, Event.SYNCHRONIZATION_FAILED);

  }

  /**

   * Essentially a get() but with different logging behavior and different access control.

   */

  public InputStream getReplica(Session session, Identifier pid)

    throws InvalidRequest, InvalidToken, NotAuthorized, NotImplemented,

    ServiceFailure, NotFound {

		InputStream inputStream = null; // bytes to be returned

	    handler = new MetacatHandler(new Timer());

	    boolean allowed = false;