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.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.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 org.dataone.service.util.Constants;

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.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 {

        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.");

        }

        // set the submitter to match the certificate

        sysmeta.setSubmitter(subject);

        // 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 {

        // set the submitter to match the certificate

        sysmeta.setSubmitter(session.getSubject());

        // set the originating node

        NodeReference originMemberNode = this.getCapabilities().getIdentifier();

        sysmeta.setOriginMemberNode(originMemberNode);

        // call the shared impl

        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

     */

    @Override

    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

     */

    @Override

    public DescribeResponse describe(Session session, Identifier pid) throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, NotImplemented,

            InvalidRequest {

    	// get system metadata and construct the describe response

        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

     */

    @Override

    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

     */

    @Override

    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;

    }

    /**

     * Return a description of the node's capabilities and services.

     * 

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

     * 

     * @throws ServiceFailure

     * @throws NotAuthorized

     * @throws InvalidRequest

     * @throws NotImplemented

     */

    @Override

    public Node getCapabilities() throws NotImplemented, NotAuthorized, ServiceFailure, InvalidRequest {

        String nodeName = null;

        String nodeId = null;

        String subject = null;

        String nodeDesc = null;

        String nodeTypeString = null;

        NodeType 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

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

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

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

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

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

            nodeType = NodeType.convert(nodeTypeString);

            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();

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

            // calls to current status methods

            Node node = new Node();

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

            node.setDescription(nodeDesc);

            // set the node's health information

            node.setState(NodeState.UP);

            // 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);

            Subject s = new Subject();

            s.setValue(subject);

            node.addSubject(s);

            node.setName(nodeName);

            node.setReplicate(nodeReplicate);

            node.setSynchronize(nodeSynchronize);

            // 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: Allow the metacat admin to determine the schedule

            // Set the schedule for synchronization

            Synchronization synchronization = new Synchronization();

            Schedule schedule = new Schedule();

            Date now = new Date();

            schedule.setYear("*");

            schedule.setMon("*");

            schedule.setMday("*");

            schedule.setWday("?");

            schedule.setHour("*");

            schedule.setMin("0/3");

            schedule.setSec("10");

            synchronization.setSchedule(schedule);

            synchronization.setLastHarvested(now);

            synchronization.setLastCompleteHarvest(now);

            node.setSynchronization(synchronization);

            node.setType(nodeType);

            return node;

        } catch (PropertyNotFoundException pnfe) {

            String msg = "MNodeService.getCapabilities(): " + "property not found: " + pnfe.getMessage();

            logMetacat.error(msg);

            throw new ServiceFailure("2162", msg);

        }

    }

    /**

     * 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

     */

    @Override

    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

     */

    @Override

    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

     */

    @Override

    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

        String principal = Constants.PUBLIC_SUBJECT;

        if (session != null && session.getSubject() != null) {

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

        }

        try {

        	EventLog.getInstance().log(syncFailed.getNodeId(), principal, localId, "synchronization_failed");

        } catch (Exception e) {

            throw new ServiceFailure("2161", "Could not log the error for: " + syncFailed.getPid());

		}

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

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

    }

    /**

     * Essentially a get() but with different logging behavior

     */

    @Override

    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;

        String localId; // the metacat docid for the pid

        // get the local docid from Metacat

        try {

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

        } catch (McdbDocNotFoundException e) {

            throw new NotFound("1020", "The object specified by " + pid.getValue() + " does not exist at this node.");

        }

        Node node = this.getCapabilities();

        Subject targetNodeSubject = node.getSubject(0);

        // check for authorization to replicate

        allowed = D1Client.getCN().isNodeAuthorized(session, targetNodeSubject, pid, Permission.REPLICATE);

        // if the person is authorized, perform the read

        if (allowed) {

            try {

                inputStream = handler.read(localId);

            } catch (Exception e) {

                throw new ServiceFailure("1020", "The object specified by " + pid.getValue() + "could not be returned due to error: " + e.getMessage());

            }

        }

        // if we fail to set the input stream

        if (inputStream == null) {

            throw new NotFound("1020", "The object specified by " + pid.getValue() + "does not exist at this node.");

        }

        // log the replica event

        String principal = null;

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

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

        }

        EventLog.getInstance().log(null, principal, localId, "getreplica");

        return inputStream;

    }

}