<commit_before>// ----------------------------------------------------------------------
// This code is developed as part of the Java CoG Kit project
// The terms of the license can be found at http://www.cogkit.org/license
// This message may not be removed or altered.
// ----------------------------------------------------------------------

package org.globus.cog.abstraction.impl.execution.gt4_0_0;

import java.net.URL;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.axis.components.uuid.UUIDGenFactory;
import org.apache.axis.message.MessageElement;
import org.apache.axis.message.addressing.EndpointReferenceType;
import org.apache.axis.types.NonNegativeInteger;
import org.apache.axis.types.PositiveInteger;
import org.apache.log4j.Logger;
import org.globus.cog.abstraction.impl.common.AbstractDelegatedTaskHandler;
import org.globus.cog.abstraction.impl.common.AbstractionProperties;
import org.globus.cog.abstraction.impl.common.execution.WallTime;
import org.globus.cog.abstraction.impl.common.task.IllegalSpecException;
import org.globus.cog.abstraction.impl.common.task.InvalidProviderException;
import org.globus.cog.abstraction.impl.common.task.InvalidSecurityContextException;
import org.globus.cog.abstraction.impl.common.task.InvalidServiceContactException;
import org.globus.cog.abstraction.impl.common.task.TaskSubmissionException;
import org.globus.cog.abstraction.interfaces.Delegation;
import org.globus.cog.abstraction.interfaces.ExecutionService;
import org.globus.cog.abstraction.interfaces.FileLocation;
import org.globus.cog.abstraction.interfaces.JobSpecification;
import org.globus.cog.abstraction.interfaces.Service;
import org.globus.cog.abstraction.interfaces.ServiceContact;
import org.globus.cog.abstraction.interfaces.Status;
import org.globus.cog.abstraction.interfaces.Task;
import org.globus.exec.client.GramJob;
import org.globus.exec.client.GramJobListener;
import org.globus.exec.generated.ExtensionsType;
import org.globus.exec.generated.FaultType;
import org.globus.exec.generated.JobDescriptionType;
import org.globus.exec.generated.JobTypeEnumeration;
import org.globus.exec.generated.NameValuePairType;
import org.globus.exec.generated.StateEnumeration;
import org.globus.exec.utils.ManagedJobFactoryConstants;
import org.globus.exec.utils.client.ManagedJobFactoryClientHelper;
import org.globus.exec.utils.rsl.RSLParseException;
import org.globus.wsrf.impl.security.authentication.Constants;
import org.globus.wsrf.impl.security.authorization.Authorization;
import org.globus.wsrf.impl.security.authorization.HostAuthorization;
import org.globus.wsrf.impl.security.authorization.SelfAuthorization;
import org.ietf.jgss.GSSCredential;

/**
 * @author CoG Team
 * @author <NAME>
 */
public class JobSubmissionTaskHandler extends AbstractDelegatedTaskHandler implements 
        GramJobListener {
    static Logger logger = Logger.getLogger(JobSubmissionTaskHandler.class);
    
    private GramJob gramJob;
    private boolean canceling;
    private String jobManager;

    private static Map jmMappings;

    static {
        jmMappings = new HashMap();
        jmMappings.put("fork", "Fork");
        jmMappings.put("pbs", "PBS");
        jmMappings.put("lsf", "LSF");
        jmMappings.put("condor", "Condor");
        jmMappings.put("sge", "SGE");
        jmMappings.put("loadleveler", "Loadleveler");
    }

    public void submit(final Task task) throws IllegalSpecException,
            InvalidSecurityContextException, InvalidServiceContactException,
            TaskSubmissionException {
        
        checkAndSetTask(task);
        task.setStatus(Status.SUBMITTING);

        JobDescriptionType rsl;
        JobSpecification spec;
        try {
            spec = (JobSpecification) task.getSpecification();
        }
        catch (Exception e) {
            throw new IllegalSpecException(
                    "Exception while retreiving Job Specification", e);
        }

        Service service = task.getService(0);
        ServiceContact serviceContact = service.getServiceContact();
        String server = serviceContact.getContact();

        String factoryType = ManagedJobFactoryConstants.DEFAULT_FACTORY_TYPE;
        if (service instanceof ExecutionService) {
            jobManager = ((ExecutionService) service)
                    .getJobManager();
            if (logger.isDebugEnabled()) {
                logger.debug("Requested job manager: " + jobManager);
            }
            if (jobManager != null) {
                String lc = jobManager.toLowerCase();
                if (lc.startsWith("jobmanager-")) {
                    lc = lc.substring(11);
                }
                if (jmMappings.containsKey(lc)) {
                    jobManager = (String) jmMappings.get(lc);
                }
                factoryType = jobManager;
            }
            if (logger.isDebugEnabled()) {
                logger.debug("Actual job manager: " + factoryType);
            }
        }

        EndpointReferenceType factoryEndpoint;
        URL factoryURL;

        try {
            factoryURL = ManagedJobFactoryClientHelper
                    .getServiceURL(server).getURL();

            // Fix (hack) for the above returning the wrong default port
            if (server.indexOf(':') == -1) {
                factoryURL = new URL(factoryURL.getProtocol(), factoryURL
                        .getHost(), 8443, factoryURL.getFile());
            }

            factoryEndpoint = ManagedJobFactoryClientHelper
                    .getFactoryEndpoint(factoryURL, factoryType);
        }
        catch (Exception e) {
            throw new IllegalSpecException("Invalid service factory", e);
        }

        if (spec.getSpecification() != null) {
            try {
                this.gramJob = new GramJob(spec.getSpecification());
            }
            catch (RSLParseException e) {
                throw new IllegalSpecException(e.getMessage(), e);
            }
        }
        else {
            rsl = prepareSpecification(spec, server);
            this.gramJob = new GramJob(rsl);
        }

        GlobusSecurityContextImpl securityContext = (GlobusSecurityContextImpl) task
                .getService(0).getSecurityContext();

        try {
       	    this.gramJob.setCredentials((GSSCredential) securityContext
        	            .getCredentials());
        }
        catch (IllegalArgumentException iae) {
            throw new InvalidSecurityContextException(
                    "Cannot set the SecurityContext twice", iae);
        }

        Authorization authorization = getAuthorization(securityContext);
        this.gramJob.setAuthorization(authorization);

        switch (securityContext.getXMLSec()) {
            case GlobusSecurityContextImpl.XML_ENCRYPTION:
                this.gramJob.setMessageProtectionType(Constants.ENCRYPTION);
                break;
            default:
                this.gramJob.setMessageProtectionType(Constants.SIGNATURE);
                break;
        }

        if (!spec.isBatchJob()) {
            this.gramJob.addListener(this);
        }

        setMiscJobParams(spec, this.gramJob);

        try {
            if (logger.isInfoEnabled()) {
                logger.info("Submitting task: " + task);
                if (logger.isDebugEnabled()) {
                    logger.debug("Rsl is " + this.gramJob.toString());
                }
            }
            this.gramJob.submit(factoryEndpoint, spec.isBatchJob(), spec
                    .getDelegation() != Delegation.NO_DELEGATION,
                    "uuid:" + UUIDGenFactory.getUUIDGen().nextUUID());
            logger.info("Task submitted: " + task);
            if (logger.isInfoEnabled()) {
            }
            if (spec.isBatchJob()) {
                task.setStatus(Status.COMPLETED);
            }
        }
        catch (Exception e) {
            // No need for cleanup. Reportedly no resource has been created
            // if an exception is thrown
            gramJob.removeListener(this);
            throw new TaskSubmissionException("Cannot submit job: "
                    + e.getMessage(), e);
        }
    }

    protected void setMiscJobParams(JobSpecification spec, GramJob job) {
        job
                .setDelegationEnabled(spec.getDelegation() != Delegation.NO_DELEGATION);<commit_msg>specifying full delegation caused limited delegation to be performed; fixed<commit_after>// ----------------------------------------------------------------------
// This code is developed as part of the Java CoG Kit project
// The terms of the license can be found at http://www.cogkit.org/license
// This message may not be removed or altered.
// ----------------------------------------------------------------------

package org.globus.cog.abstraction.impl.execution.gt4_0_0;

import java.net.URL;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.axis.components.uuid.UUIDGenFactory;
import org.apache.axis.message.MessageElement;
import org.apache.axis.message.addressing.EndpointReferenceType;
import org.apache.axis.types.NonNegativeInteger;
import org.apache.axis.types.PositiveInteger;
import org.apache.log4j.Logger;
import org.globus.cog.abstraction.impl.common.AbstractDelegatedTaskHandler;
import org.globus.cog.abstraction.impl.common.AbstractionProperties;
import org.globus.cog.abstraction.impl.common.execution.WallTime;
import org.globus.cog.abstraction.impl.common.task.IllegalSpecException;
import org.globus.cog.abstraction.impl.common.task.InvalidProviderException;
import org.globus.cog.abstraction.impl.common.task.InvalidSecurityContextException;
import org.globus.cog.abstraction.impl.common.task.InvalidServiceContactException;
import org.globus.cog.abstraction.impl.common.task.TaskSubmissionException;
import org.globus.cog.abstraction.interfaces.Delegation;
import org.globus.cog.abstraction.interfaces.ExecutionService;
import org.globus.cog.abstraction.interfaces.FileLocation;
import org.globus.cog.abstraction.interfaces.JobSpecification;
import org.globus.cog.abstraction.interfaces.Service;
import org.globus.cog.abstraction.interfaces.ServiceContact;
import org.globus.cog.abstraction.interfaces.Status;
import org.globus.cog.abstraction.interfaces.Task;
import org.globus.exec.client.GramJob;
import org.globus.exec.client.GramJobListener;
import org.globus.exec.generated.ExtensionsType;
import org.globus.exec.generated.FaultType;
import org.globus.exec.generated.JobDescriptionType;
import org.globus.exec.generated.JobTypeEnumeration;
import org.globus.exec.generated.NameValuePairType;
import org.globus.exec.generated.StateEnumeration;
import org.globus.exec.utils.ManagedJobFactoryConstants;
import org.globus.exec.utils.client.ManagedJobFactoryClientHelper;
import org.globus.exec.utils.rsl.RSLParseException;
import org.globus.wsrf.impl.security.authentication.Constants;
import org.globus.wsrf.impl.security.authorization.Authorization;
import org.globus.wsrf.impl.security.authorization.HostAuthorization;
import org.globus.wsrf.impl.security.authorization.SelfAuthorization;
import org.ietf.jgss.GSSCredential;

/**
 * @author CoG Team
 * @author <NAME>
 */
public class JobSubmissionTaskHandler extends AbstractDelegatedTaskHandler implements 
        GramJobListener {
    static Logger logger = Logger.getLogger(JobSubmissionTaskHandler.class);
    
    private GramJob gramJob;
    private boolean canceling;
    private String jobManager;

    private static Map jmMappings;

    static {
        jmMappings = new HashMap();
        jmMappings.put("fork", "Fork");
        jmMappings.put("pbs", "PBS");
        jmMappings.put("lsf", "LSF");
        jmMappings.put("condor", "Condor");
        jmMappings.put("sge", "SGE");
        jmMappings.put("loadleveler", "Loadleveler");
    }

    public void submit(final Task task) throws IllegalSpecException,
            InvalidSecurityContextException, InvalidServiceContactException,
            TaskSubmissionException {
        
        checkAndSetTask(task);
        task.setStatus(Status.SUBMITTING);

        JobDescriptionType rsl;
        JobSpecification spec;
        try {
            spec = (JobSpecification) task.getSpecification();
        }
        catch (Exception e) {
            throw new IllegalSpecException(
                    "Exception while retreiving Job Specification", e);
        }

        Service service = task.getService(0);
        ServiceContact serviceContact = service.getServiceContact();
        String server = serviceContact.getContact();

        String factoryType = ManagedJobFactoryConstants.DEFAULT_FACTORY_TYPE;
        if (service instanceof ExecutionService) {
            jobManager = ((ExecutionService) service)
                    .getJobManager();
            if (logger.isDebugEnabled()) {
                logger.debug("Requested job manager: " + jobManager);
            }
            if (jobManager != null) {
                String lc = jobManager.toLowerCase();
                if (lc.startsWith("jobmanager-")) {
                    lc = lc.substring(11);
                }
                if (jmMappings.containsKey(lc)) {
                    jobManager = (String) jmMappings.get(lc);
                }
                factoryType = jobManager;
            }
            if (logger.isDebugEnabled()) {
                logger.debug("Actual job manager: " + factoryType);
            }
        }

        EndpointReferenceType factoryEndpoint;
        URL factoryURL;

        try {
            factoryURL = ManagedJobFactoryClientHelper
                    .getServiceURL(server).getURL();

            // Fix (hack) for the above returning the wrong default port
            if (server.indexOf(':') == -1) {
                factoryURL = new URL(factoryURL.getProtocol(), factoryURL
                        .getHost(), 8443, factoryURL.getFile());
            }

            factoryEndpoint = ManagedJobFactoryClientHelper
                    .getFactoryEndpoint(factoryURL, factoryType);
        }
        catch (Exception e) {
            throw new IllegalSpecException("Invalid service factory", e);
        }

        if (spec.getSpecification() != null) {
            try {
                this.gramJob = new GramJob(spec.getSpecification());
            }
            catch (RSLParseException e) {
                throw new IllegalSpecException(e.getMessage(), e);
            }
        }
        else {
            rsl = prepareSpecification(spec, server);
            this.gramJob = new GramJob(rsl);
        }

        GlobusSecurityContextImpl securityContext = (GlobusSecurityContextImpl) task
                .getService(0).getSecurityContext();

        try {
       	    this.gramJob.setCredentials((GSSCredential) securityContext
        	            .getCredentials());
        }
        catch (IllegalArgumentException iae) {
            throw new InvalidSecurityContextException(
                    "Cannot set the SecurityContext twice", iae);
        }

        Authorization authorization = getAuthorization(securityContext);
        this.gramJob.setAuthorization(authorization);

        switch (securityContext.getXMLSec()) {
            case GlobusSecurityContextImpl.XML_ENCRYPTION:
                this.gramJob.setMessageProtectionType(Constants.ENCRYPTION);
                break;
            default:
                this.gramJob.setMessageProtectionType(Constants.SIGNATURE);
                break;
        }

        if (!spec.isBatchJob()) {
            this.gramJob.addListener(this);
        }

        setMiscJobParams(spec, this.gramJob);

        try {
            if (logger.isInfoEnabled()) {
                logger.info("Submitting task: " + task);
                if (logger.isDebugEnabled()) {
                    logger.debug("Rsl is " + this.gramJob.toString());
                }
            }
            this.gramJob.submit(factoryEndpoint, spec.isBatchJob(), spec
                    .getDelegation() != Delegation.FULL_DELEGATION,
                    "uuid:" + UUIDGenFactory.getUUIDGen().nextUUID());
            logger.info("Task submitted: " + task);
            if (logger.isInfoEnabled()) {
            }
            if (spec.isBatchJob()) {
                task.setStatus(Status.COMPLETED);
            }
        }
        catch (Exception e) {
            // No need for cleanup. Reportedly no resource has been created
            // if an exception is thrown
            gramJob.removeListener(this);
            throw new TaskSubmissionException("Cannot submit job: "
                    + e.getMessage(), e);
        }
    }

    protected void setMiscJobParams(JobSpecification spec, GramJob job) {
        job
                .setDelegationEnabled(spec.getDelegation() != Delegation.NO_DELEGATION);
{'fraction_non_alphanumeric': 0.1118265595463138, 'fraction_numerical': 0.0015359168241965973, 'mean_word_length': 2.3509501187648456, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2684970', 'n_tokens_mistral': 4612, 'n_tokens_neox': 4481, 'n_words': 824}