Pegasus

The abstract workflow after parsing is optionally handed over to the Data Reuse Module. The Data Reuse Algorithm in Pegasus attempts to prune all the nodes in the abstract workflow for which the output files exist in the Replica Catalog. It also attempts to cascade the deletion to the parents of the deleted node for e.g if the output files for the leaf nodes are specified, Pegasus will prune out all the workflow as the output files in which a user is interested in already exist in the Replica Catalog.

The Data Reuse Algorithm works in two passes

First Pass - Determine all the jobs whose output files exist in the Replica Catalog. An output file with the transfer flag set to false is treated equivalent to the file existing in the Replica Catalog , if the output file is not an input to any of the children of the job X.

Second Pass - The algorithm removes the job whose output files exist in the Replica Catalog and tries to cascade the deletion upwards to the parent jobs. We start the breadth first traversal of the workflow bottom up.

( It is already marked for deletion in Pass 1
     OR
      ( ALL of it's children have been marked for deletion
        AND
        Node's output files have transfer flags set to false
       )
 )

Figure 5.2. Workflow Data Reuse

The abstract workflow is then handed over to the Site Selector module where the abstract jobs in the pruned workflow are mapped to the various sites passed by a user. The target sites for planning are specified on the command line using the --sites option to pegasus-plan. If not specified, then Pegasus picks up all the sites in the Site Catalog as candidate sites. Pegasus will map a compute job to a site only if Pegasus can

Figure 5.3. Workflow Site Selection

After site selection, the workflow is optionally handed for to the job clustering module, which clusters jobs that are scheduled to the same site. Clustering is usually done on short running jobs in order to reduce the remote execution overheads associated with a job. Clustering is described in detail in the Reference Manualchapter.

After job clustering, the workflow is handed to the Data Transfer module that adds data stage-in , inter site and stage-out nodes to the workflow. Data Stage-in Nodes transfer input data required by the workflow from the locations specified in the Replica Catalog to a directory on the execution site where the job executes. In case, multiple locations are specified for the same input file, the location from where to stage the data is selected using a Replica Selector . Replica Selection is described in detail in the Replica Selection section of the Reference Manual.

The process of adding the data stage-in and data stage-out nodes is handled by Transfer Refiners. All data transfer jobs in Pegasus are executed using pegasus-transfer . The pegasus-transfer client is a python based wrapper around various transfer clients like globus-url-copy, lcg-copy, wget, cp, ln . It looks at source and destination url and figures out automatically which underlying client to use. pegasus-transfer is distributed with the PEGASUS and can be found in the bin subdirectory . Pegasus Transfer Refiners are are described in the detail in the Transfers section of the Reference Manual. The default transfer refiner that is used in Pegasus is the Bundle Transfer Refiner, that bundles data stage-in nodes and data stage-out nodes on the basis of certain pegasus profile keys associated with the workflow.

Figure 5.4. Addition of Data Transfer Nodes to the Workflow

Data Registration Nodes may also be added to the final executable workflow to register the location of the output files on the final output site back in the Replica Catalog . An output file is registered in the Replica Catalog if the register flag for the file is set to true in the DAX.

Figure 5.5. Addition of Data Registration Nodes to the Workflow

The data staged-in and staged-out from a directory that is created on the head node by a create dir job in the workflow. In the vanilla case, the directory is visible to all the worker nodes and compute jobs are launched in this directory on the shared filesystem. In the case where there is no shared filesystem, users can turn on worker node execution, where the data is staged from the head node directory to a directory on the worker node filesystem. This feature will be refined further for Pegasus 3.1. To use it with Pegasus 3.0 send email to pegasus-support at isi.edu.

After the data transfer nodes have been added to the workflow, Pegasus adds a create dir jobs to the workflow. Pegasus usually , creates one workflow specific directory per compute site , that is on the shared filesystem of compute site. This directory is visible to all the worker nodes and that is where the data is staged-in by the data stage-in jobs.

After addition of the create dir jobs, the workflow is optionally handed to the cleanup module. The cleanup module adds cleanup nodes to the workflow that remove data from the directory on the shared filesystem when it is no longer required by the workflow. This is useful in reducing the peak storage requirements of the workflow.

Figure 5.6. Addition of Directory Creation and File Removal Jobs

The last step of refinement process, is the code generation where Pegasus writes out the executable workflow in a form understandable by the underlying workflow executor. At present Pegasus supports the following code generators

Figure 5.7. Final Executable Workflow

The property pegasus.home cannot be set in the property file. This property is automatically set up by the pegasus clients internally by determining the installation directory of pegasus. Knowledge about this property is important for developers who want to invoke PEGASUS JAVA classes without the shell wrappers.

The rest of this chapter will focus on the vanilla grid execution environment.

Figure 5.8. Grid Sample Site Layout

The vanilla grid environment shown in the figure above. We will work from the left to the right top, then the right bottom.

On the left side, you have a submit machine where Pegasus runs, Condor schedules jobs, and workflows are executed. We call it the submit host (SH), though its functionality can be assumed by a virtual machine image. In order to properly communicate over secured channels, it is important that the submit machine has a proper notion of time, i.e. runs an NTP daemon to keep accurate time. To be able to connect to remote clusters and receive connections from the remote clusters, the submit host has a public IP address to facilitate this communication.

In order to send a job request to the remote cluster, Condor wraps the job into Globus calls via Condor-G. Globus uses GRAM to manage jobs on remote sites. In terms of a software stack, Pegasus wraps the job into Condor. Condor wraps the job into Globus. Globus transports the job to the remote site, and unwraps the Globus component, sending it to the remote site's resource manager (RM).

To be able to communicate using the Globus security infrastructure (GSI), the submit machine needs to have the certificate authority (CA) certificates configured, requires a host certificate in certain circumstances, and the user a user certificate that is enabled on the remote site. On the remote end, the remote gatekeeper node requires a host certificate, all signing CA certificate chains and policy files, and a goot time source.

In a grid environment, there are one or more clusters accessible via grid middleware like the Globus Toolkit. In case of Globus, there is the Globus gatekeeper listening on TCP port 2119 of the remote cluster. The port is opened to a single machine called head node (HN).The head-node is typically located in a de-militarized zone (DMZ) of the firewall setup, as it requires limited outside connectivity and a public IP address so that it can be contacted. Additionally, once the gatekeeper accepted a job, it passes it on to a jobmanager. Often, these jobmanagers require a limited port range, in the example TCP ports 40000-41000, to call back to the submit machine.

For the user to be able to run jobs on the remote site, the user must have some form of an account on the remtoe site. The user's grid identity is passed from the submit host. An entity called grid mapfile on the gatekeeper maps the user's grid identity into a remote account. While most sites do not permit account sharing, it is possible to map multiple user certificates to the same account.

The gatekeeper is the interface through which jobs are submitted to the remote cluster's resource manager. A resource manager is a scheduling system like PBS, Maui, LSF, FBSNG or Condor that queues tasks and allocates worker nodes. The worker nodes (WN) in the remote cluster might not have outside connectivity and often use all private IP addresses. The Globus toolkit requires a shared filesystem to properly stage files between the head node and worker nodes.

To stage data between external sites for the job, it is recommended to enable a GridFTP server. If a shared networked filesystem is involved, the GridFTP server should be located as close to the file-server as possible. The GridFTP server requires TCP port 2811 for the control channel, and a limited port range for data channels, here as an example the TPC ports from 40000 to 41000. The GridFTP server requires a host certificate, the signing CA chain and policy files, a stable time source, and a gridmap file that maps between a user's grid identify and the user's account on the remote site.

The GridFTP server is often installed on the head node, the same as the gatekeeper, so that they can share the grid mapfile, CA certificate chains and other setups. However, for performance purposes it is recommended that the GridFTP server has its own machine.

Figure 5.9. Cloud Sample Site Layout

The pevious figure shows a sample layout for sky computing (as in: multiple clouds) as supported by Pegasus. At this point, it is up to the user to provision the remote resources with a proper VM image that includes a Condor startd and proper Condor configuration to report back to a Condor collector that the Condor schedd has access to.

In this discussion, the submit host (SH) is located logically external to the cloud provider(s). The SH is the point where a user submits Pegasus workflows for execution. This site typically runs a Condor collector to gather resource announcements, or is part of a larger Condor pool that collects these announcement. Condor makes the remote resources available to the submit host’s Condor installation.

The figure above shows the way Pegasus WMS is deployed in cloud computing resources, ignoring how these resources were provisioned. The provisioning request shows multiple resources per provisioning request.

The provisioning broker -- Nimbus, Eucalyptus or EC2 -- at the remote site is responsible to allocate and set up the resources. For a multi-node request, the worker nodes often require access to a form of shared data storage. Concretely, either a POSIX-compliant shared file system (e.g. NFS, PVFS) is available to the nodes, or can be brought up for the lifetime of the application workflow. The task steps of the application workflow facilitate shared file systems to exchange intermediary results between tasks on the same cloud site. Pegasus also supports an S3 data mode for the application workflow data staging.

The initial stage-in and final stage-out of application data into and out of the node set is part of any Pegasus-planned workflow. Several configuration options exist in Pegasus to deal with the dynamics of push and pull of data, and when to stage data. In many use-cases, some form of external access to or from the shared file system that is visible to the application workflow is required to facilitate successful data staging. However, Pegasus is prepared to deal with a set of boundary cases.

The data server in the figure is shown at the submit host. This is not a strict requirement. The data server for consumed data and data products may both be different and external to the submit host.

Once resources begin appearing in the pool managed by the submit machine’s Condor collector, the application workflow can be submitted to Condor. A Condor DAGMan will manage the application workflow execution. Pegasus run-time tools obtain timing-, performance and provenance information as the application workflow is executed. At this point, it is the user's responsibility to de-provision the allocated resources.

In the figure, the cloud resources on the right side are assumed to have uninhibited outside connectivity. This enables the Condor I/O to communicate with the resources. The right side includes a setup where the worker nodes use all private IP, but have out-going connectivity and a NAT router to talk to the internet. The Condor connection broker (CCB) facilitates this setup almost effortlessly.

The left side shows a more difficult setup where the connectivity is fully firewalled without any connectivity except to in-site nodes. In this case, a proxy server process, the generic connection broker (GCB), needs to be set up in the DMZ of the cloud site to facilitate Condor I/O between the submit host and worker nodes.

If the cloud supports data storage servers, Pegasus is starting to support workflows that require staging in two steps: Consumed data is first staged to a data server in the remote site's DMZ, and then a second staging task moves the data from the data server to the worker node where the job runs. For staging out, data needs to be first staged from the job's worker node to the site's data server, and possibly from there to another data server external to the site. Pegasus is capable to plan both steps: Normal staging to the site's data server, and the worker-node staging from and to the site's data server as part of the job. We are working on expanding the current code to support a more generic set by Pegasus 3.1.

In this configuration , Pegasus schedules the jobs to run locally on the submit host. This is achieved by executing the workflow on site local ( --sites local option to pegasus-plan ). The site "local" is a reserved site in Pegasus and results in the jobs to run on the submit host in condor universe local.

In this configuration, it is assumed that the target execution system consists of one or more Grid resources. These resources may be geographically distributed and under various administrative domains. Each resource might be a single desktop computer or a network of workstations (NOW) or a cluster of dedicated machines. However, each resource must provide a Globus GRAM interface which allows the users to submit jobs remotely. In case the Grid resource consists of multiple compute nodes, e.g. a cluster or a network of workstations, there is a central entity called the head node that acts as the single point of job submissions to the resource. It is generally possible to specify whether the submitted job should run on the head node of the resource or a worker node in the cluster or NOW. In the latter case, the head node is responsible for submitting the job to a local resource management system (PBS, LSF, Condor etc) which controls all the machines in the resource. Since, the head node is the central point of job submissions to the resource it should not be used for job execution since that can overload the head node delaying further job submissions. Pegasus does not make any assumptions about the configuration of the remote resource; rather it provides the mechanisms by which such distinctions can be made.

Figure 5.10. Resource Configuration using GRAM

In this configuration, Condor-G is used for submitting jobs to these resources. Condor-G is an extension to Condor that allows the jobs to be described in a Condor submit file and when the job is submitted to Condor for execution, it uses the Globus GRAM interface to submit the job to the remote resource and monitor its execution. The distinction is made in the Condor submit files by specify the universe as Grid and the grid_resource or globusscheduler attribute is used to indicate the location of the head node for the remote resource. Thus, Condor DAGMan is used for maintaining the dependencies between the jobs and Condor-G is used to launch the jobs on the remote resources using GRAM. The implicit assumption in this case is that all the worker nodes on a remote resource have access to a shared file system that can be used for data transfer between the tasks mapped on that resource. This data transfer is done using files.

Below is an example of a site configured to use Globus GRAM2| GRAM5

 <site  handle="isi" arch="x86" os="LINUX" osrelease="" osversion="" glibc="">
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="auxillary"/>
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="compute"/>
       <head-fs>
               <scratch>
                   <shared>
                         <file-server protocol="gsiftp" url="gsiftp://skynet-data.isi.edu" \
                                                     mount-point="/nfs/scratch01"/>
                         <internal-mount-point mount-point="/nfs/scratch01"/>
                    </shared>
               </scratch>
               <storage>
                   <shared>
                         <file-server protocol="gsiftp" url="gsiftp://skynet-data.isi.edu" \
                                                     mount-point="/exports/storage01"/>
                         <internal-mount-point mount-point="/exports/storage01"/>
                    </shared>
               </storage>
        </head-fs>
        <replica-catalog  type="LRC" url="rlsn://smarty.isi.edu" />
  </site>

A Condor pool is a set of machines that use Condor for resource management. A Condor pool can be a cluster of dedicated machines or a set of distributively owned machines. Pegasus can generate concrete workflows that can be executed on a Condor pool.

Figure 5.11. The distributed resources appear to be part of a Condor pool.

The workflow is submitted using DAGMan from one of the job submission machines in the Condor pool. It is the responsibility of the Central Manager of the pool to match the task in the workflow submitted by DAGMan to the execution machines in the pool. This matching process can be guided by including Condor specific attributes in the submit files of the tasks. If the user wants to execute the workflow on the execution machines (worker nodes) in a Condor pool, there should be a resource defined in the site catalog which represents these execution machines. The universe attribute of the resource should be vanilla. There can be multiple resources associated with a single Condor pool, where each resource identifies a subset of machine (worker nodes) in the pool. Pegasus currently uses the FileSystemDomain classad[] attribute to restrict the set of machines that make up a single resource. To clarify this point, suppose there are certain execution machines in the Condor pool whose FileSystemDomain is set to “viz.isi.edu”. If the user wants to execute the workflow on these machines, then there should be a site, say “isi_viz” defined in the site catalog and the FileSystemDomain and universe attributes for this resource should be defined as “viz.isi.edu” and “vanilla” respectively. When invoking Pegasus, the user should select isi_viz as the compute resource.

 <site  handle="isi_viz" arch="x86" os="LINUX" osrelease="" osversion="" glibc="">
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="auxillary"/>
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="compute"/>
       <head-fs>
              <scratch>
                    <shared>
                           <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                              mount-point="/nfs/scratch01"/>
                           <internal-mount-point mount-point="/nfs/scratch01"/>
                    </shared>
              </scratch>
              <storage>
                    <shared>
                           <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                           mount-point="/exports/storage01"/>
                           <internal-mount-point mount-point="/exports/storage01"/>
                           </shared>
               </storage>
       </head-fs>
       <replica-catalog  type="LRC" url="rlsn://smarty.isi.edu"/>
       <profile namespace="condor" key="universe">vanilla</profile> 
       <profile namespace="condor" key="FileSystemDomain">viz.isi.edu</profile>
 </site>

Pegasus WMS can execute workflows over Condor pool. This pool can contain machines managed by a single institution or department and belonging to a single administrative domain. This is the case for most of the Condor pools. In this section we describe how machines from different administrative domains and supercomputing centers can be dynamically added to a Condor pool for certain timeframe. These machines join the Condor pool temporarily and can be used to execute jobs in a non preemptive manner. This functionality is achieved using a Condor feature called Glide-in http://cs.wisc.edu/condor/glidein that uses Globus GRAM interface for migrating machines from different domains to a Condor pool. The number of machines and the duration for which they are required can be specified

In this case, we use the abstraction of a local Condor pool to execute the jobs in the workflow over remote resources that have joined the pool for certain timeframe. Details about the use of this feature can be found in the condor manual (http://cs.wisc.edu/condor/manual/).

A basic step to migrate in a job to a local condor pool is described below.

$condor_glidein -count 10 gatekeeper.site.edu/jobmanager-pbs

GlideIn of Remote Globus Resources

The above step glides in 10 nodes to the user’s local condor pool, from the remote pbs scheduler running at gatekeeper.site.edu. By default, the glide in binaries are installed in the users home directory on the remote host.

It is possible that the Condor pool can contain resources from multiple Grid sites. It is normally the case that the resources from a particular site share the same file system and thus use the same FileSystemDomain attribute while advertising their presence to the Central Manager of the pool. If the user wants to run his jobs on machines from a particular Grid site, he has to specify the FileSystemDomain attribute in the requirements classad expression in the submit files with a value matching the FileSystemDomain of the machines from that site. For example, the user migrates nodes from the ISI cluster (with FileSystemDomain viz.isi.org) into a Condor pool and specifies FileSystemDomain == “viz.isi.edu”. Condor would then schedule the jobs only on the nodes from the ISI VIZ cluster in the local condor pool. The FileSystemDomain can be specified for an execution site in the site catalog with condor profile namespace as follows

 <site  handle="isi_viz" arch="x86" os="LINUX" osrelease="" osversion="" glibc="">
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="auxillary"/>
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="compute"/>
       <head-fs>
              <scratch>
                    <shared>
                         <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                                 mount-point="/nfs/scratch01"/>
                         <internal-mount-point mount-point="/nfs/scratch01"/>
                    </shared>
              </scratch>
              <storage>
                     <shared>
                          <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                                      mount-point="/exports/storage01"/>
                          <internal-mount-point mount-point="/exports/storage01"/>
                     </shared>
              </storage>
       </head-fs>
       <replica-catalog  type="LRC" url="rlsn://smarty.isi.edu"/>
       <profile namespace="pegasus" key="style">glidein</profile>
       <profile namespace="condor" key="universe">vanilla</profile> 
       <profile namespace="condor" key="FileSystemDomain">viz.isi.edu</profile>
 </site>

Specifying the FileSystemDomain key in condor namespace for a site, triggers Pegasus into generating the requirements classad expression in the submit file for all the jobs scheduled on that particular site. For example, in the above case all jobs scheduled on site isi_condor would have the following expression in the submit file.

requirements = FileSystemDomain == “viz.isi.edu”

glideinWMS is a glidein system widely used on Open Science Grid. Pegasus has a convenience style named glideinWMS to make running workflows on top of glideinWMS easy. This style is similar to the Condor style, but provides some more defaults set up to work well with glideinWMS. All you have to do is specify the style profile in the site catalog:

 <site  handle="isi_viz" arch="x86" os="LINUX" osrelease="" osversion="" glibc="">
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="auxillary"/>
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="compute"/>
       <head-fs>
              <scratch>
                    <shared>
                         <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                                 mount-point="/nfs/scratch01"/>
                         <internal-mount-point mount-point="/nfs/scratch01"/>
                    </shared>
              </scratch>
              <storage>
                     <shared>
                          <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                                      mount-point="/exports/storage01"/>
                          <internal-mount-point mount-point="/exports/storage01"/>
                     </shared>
              </storage>
       </head-fs>
       <replica-catalog  type="LRC" url="rlsn://smarty.isi.edu"/>
       <profile namespace="pegasus" key="style">glideinwms</profile>
 </site>

Planned jobs will have universe, requirements and rank set:

universe = vanilla
rank = DaemonStartTime
requirements = ((GLIDEIN_Entry_Name == "isi_iz") || (TARGET.Pegasus_Site == "isi_viz")) \
               && (IS_MONITOR_VM == False) && (Arch != "") && (OpSys != "") \
               && (Disk != -42) && (Memory > 1) && (FileSystemDomain != "")

This section describes the various changes required in the site catalog for Pegasus to generate an executable workflow that uses gLite blahp to directly submit to PBS on the local machine. This mode of submission should only be used when the condor on the submit host can directly talk to scheduler running on the cluster. It is recommended that the cluster that gLite talks to is designated as a separate compute site in the Pegasus site catalog. To tag a site as a gLite site the following two profiles need to be specified for the site in the site catalog

An example site catalog entry for a glite site looks as follows in the site catalog

 <site  handle="isi_viz_glite" arch="x86" os="LINUX" osrelease="" osversion="" glibc="">
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="auxillary"/>
       <grid  type="gt2" contact="smarty.isi.edu/jobmanager-pbs" scheduler="PBS" jobtype="compute"/>
       <head-fs>
             <scratch>
                   <shared>
                         <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                       mount-point="/nfs/scratch01"\>
                         <internal-mount-point mount-point="/nfs/scratch01"/>
                   </shared>
             </scratch>
             <storage>
                   <shared>
                         <file-server protocol="gsiftp" url="gsiftp://viz-login.isi.edu" \
                                                mount-point="/exports/storage01"\>
                         <internal-mount-point mount-point="/exports/storage01"/>
                    </shared>
             </storage>
       </head-fs>
       <replica-catalog  type="LRC" url="rlsn://smarty.isi.edu" />
               
       <!-- following profiles reqd for glite grid style-->
       <profile namespace="pegasus" key="style">glite</profile> 
       <profile namespace="condor" key="grid_resource">pbs</profile>
 </site>

+remote_cerequirements = "PROCS==18 && NODES==1 && PRIORITY==10 && WALLTIME==3600 \
   && PASSENV==1 && JOBNAME==\"TEST JOB\" && MYENV ==\"JAVA_HOME=/bin/java,APP_HOME=/bin/app\""

It is possible to run on a condor pool without the worker nodes sharing a filesystem and relying on Condor to do the file transfers. However, for the Condor file to transfer the input data for a job, all the input data

To set up Pegasus to plan workflows for Amazon you need to create an 'local-condor' entry your site catalog. The site configuration is similar to what you would create for running on a local Condor pool (See previous section).

 <site  handle="local-condor" arch="x86" os="LINUX">
        <grid  type="gt2" contact="localhost/jobmanager-fork" scheduler="Fork" jobtype="auxillary"/>
        <grid  type="gt2" contact="localhost/jobmanager-condor" scheduler="unknown" jobtype="compute"/>
        <head-fs>

          <!-- the paths for scratch filesystem are the paths on local site as we execute create dir job
               on local site. Improvements planned for 3.1 release.-->
            <scratch>
                <shared>
                    <file-server protocol="file" url="file:///" mount-point="/submit-host/scratch"/>
                    <internal-mount-point mount-point="/submit-host/scratch"/>
                </shared>
            </scratch>
            <storage>
                <shared>
                    <file-server protocol="file" url="file:///" mount-point="/glusterfs/scratch"/>
                    <internal-mount-point mount-point="/glusterfs/scratch"/>
                </shared>
            </storage>
        </head-fs>
        <replica-catalog  type="LRC" url="rlsn://dummyValue.url.edu" />
        <profile namespace="env" key="PEGASUS_HOME" >/cluster-software/pegasus/2.4.1</profile>
        <profile namespace="env" key="GLOBUS_LOCATION" >/cluster-software/globus/5.0.1</profile>

        <!-- profies for site to be treated as condor pool -->
        <profile namespace="pegasus" key="style" >condor</profile>
        <profile namespace="condor" key="universe" >vanilla</profile>

        <!-- required for Condor File IO and transferring stdout and stderr back -->
        <profile namespace="condor" key="should_transfer_files">YES</profile>
        <profile namespace="condor" key="transfer_output">true</profile>
        <profile namespace="condor" key="transfer_error">true</profile>
        <profile namespace="condor" key="WhenToTransferOutput">ON_EXIT</profile>
        
        <!-- to enable kickstart staging from local site-->
        <profile namespace="condor" key="transfer_executable">true</profile>


    </site>

Next you need need to update pegasus.properties to tell Pegasus to turn on Condor File IO and execute jobs on the worker nodes filesystem.

pegasus.data.configuration = Condor

By default, Pegasus creates a workflow specific execution directory that is associated with the compute site ( in this case local-condor). However, in this scenario, the workflow specific execution directory needs to be created on the submit host, as native Condor File I/O only works against local directories. To ensure that the create dir job for the workflow for the local-condor site, you need to do the following