Repository: riga/law (language: python)

Issue:
Support for CRAB workflows
Resources: https://twiki.cern.ch/twiki/bin/view/CMSPublic/SWGuideCrab


Discussion / hints:
Note: this would also require a workflow implementation that supports a batched submission. The HTCondorWorkflow could greatly profit from this as well.

Fix patch:
diff --git a/README.md b/README.md
index 4efea232..e7882aa0 100644
--- a/README.md
+++ b/README.md
@@ -17,7 +17,7 @@
 
 
 **Note**: This project is currently under development.
-Version 1.0.0 will be the first, fully documented beta release, targetted for mid 2023.
+Version 1.0.0 will be the first, fully documented beta release, targetted for fall 2023.
 
 Use law to build complex and large-scale task workflows.
 It is build on top of [luigi](https://github.com/spotify/luigi) and adds abstractions for **run locations**, **storage locations** and **software environments**.
@@ -29,7 +29,7 @@ Key features:
 - Remote targets with automatic retries and local caching
   - WebDAV, HTTP, Dropbox, SFTP, all WLCG protocols (srm, xrootd, dcap, gsiftp, webdav, ...)
 - Automatic submission to batch systems from within tasks
-  - HTCondor, LSF, gLite, ARC, Slurm
+  - HTCondor, LSF, gLite, ARC, Slurm, CMS-CRAB
 - Environment sandboxing, configurable on task level
   - Docker, Singularity, Sub-Shells, Virutal envs
 
@@ -141,6 +141,7 @@ docker run -ti riga/law:example <example_name>
 - [wlcg_targets](https://github.com/riga/law/tree/master/examples/wlcg_targets): Working with targets that are stored on WLCG storage elements (dCache, EOS, ...). TODO.
 - [htcondor_at_vispa](https://github.com/riga/law/tree/master/examples/htcondor_at_vispa): HTCondor workflows at the [VISPA service](https://vispa.physik.rwth-aachen.de).
 - [htcondor_at_cern](https://github.com/riga/law/tree/master/examples/htcondor_at_cern): HTCondor workflows at the CERN batch infrastructure.
+- [CMS Crab at CERN](https://github.com/riga/law_example_CMSCrabWorkflows): CMS Crab workflows executed from lxplus at CERN.
 - [sequential_htcondor_at_cern](https://github.com/riga/law/tree/master/examples/sequential_htcondor_at_cern): Continuation of the [htcondor_at_cern](https://github.com/riga/law/tree/master/examples/htcondor_at_cern) example, showing sequential jobs that eagerly start once jobs running previous requirements succeeded.
 - [htcondor_at_naf](https://github.com/riga/law/tree/master/examples/htcondor_at_naf): HTCondor workflows at German [National Analysis Facility (NAF)](https://confluence.desy.de/display/IS/NAF+-+National+Analysis+Facility).
 - [slurm_at_maxwell](https://github.com/riga/law/tree/master/examples/slurm_at_maxwell): Slurm workflows at the [Desy Maxwell cluster](https://confluence.desy.de/display/MXW/Maxwell+Cluster).
diff --git a/examples/htcondor_at_cern/analysis/bootstrap.sh b/examples/htcondor_at_cern/analysis/bootstrap.sh
index b2057f86..5ef50ad1 100644
--- a/examples/htcondor_at_cern/analysis/bootstrap.sh
+++ b/examples/htcondor_at_cern/analysis/bootstrap.sh
@@ -7,6 +7,6 @@
 # base tasks in analysis/framework.py.
 
 action() {
-    source "{{analysis_path}}/setup.sh"
+    source "{{analysis_path}}/setup.sh" "$@"
 }
-action
+action "$@"
diff --git a/law.cfg.example b/law.cfg.example
index e3b47b83..30e4bb9f 100644
--- a/law.cfg.example
+++ b/law.cfg.example
@@ -33,6 +33,8 @@
 ;   - [singularity_sandbox]
 ;   - [singularity_sandbox_env]
 ;   - [singularity_sandbox_volumes]
+;   - [cmssw_sandbox]
+;   - [cmssw_sandbox_env]
 ;
 ; Please note that configuration options of law contrib packages might also appear in sections of
 ; the general law configuration.
@@ -905,6 +907,36 @@
 ; Type: integer
 ; Default: 25
 
+; crab_job_file_dir
+; crab_job_file_dir_mkdtemp
+; crab_job_file_dir_cleanup
+; Description: These three options are identical to the ones above without the "crab" prefix, but
+; only apply to the "law.cms.CrabJobFileFactory". When "None" or not existing, the values above are
+; used.
+
+; crab_work_area
+; Description: The directory in which the "law.cms.CrabJobManager" will create crab project
+; directories upon submission. Defaults to "crab_job_file_dir".
+; Type: string
+; Default: job.crab_job_file_dir
+
+; crab_sandbox_name
+; Description: The name of the "cmssw" sandbox to use which provides the crab executable. In its
+; simplest form, this is just the version string of the CMSSW release to use. However, additional
+; settings can be appended with the "::" delimiter in the format
+; "cmssw_version::setting=value::setting=value" for more configurablity. Supported settings are
+; "setup", an additional setup script that is executed inside the src directory during installation,
+; "dir", a custom install directory, "arch", a custom scram architecture, and "cores", the number of
+; CPU cores to use for installation.
+; Type: string
+; Default: CMSSW_10_6_30
+
+; crab_password_file
+; Description: A file containing the X509 certificate password for automatic proxy delegations by
+; the "law.crab.CrabJobManager".
+; Type: string
+; Default: None
+
 
 ; --- notifications section ------------------------------------------------------------------------
 
@@ -1238,3 +1270,53 @@
 ; represent host and container directory, respectively.
 ; The volumes defined in this section are applied to all singularity sandboxes. To configure volumes
 ; per image, create a section "singularity_sandbox_volumes_<image_name>" with the desired values.
+
+
+; --- cmssw_sandbox section ------------------------------------------------------------------------
+
+[cmssw_sandbox]
+
+; Note:
+; This section defines the default options for all cmssw sandboxes. To configure options per cmssw
+; version, create a section "cmssw_sandbox_<cmssw_version>" with the desired options.
+
+
+; stagein_dir_name
+; Description: Name of a directory which is placed automatically inside a temporary directory and to
+; which input files are staged-in outside of the sandbox and provided to the sandboxed task. When
+; "None", no stage-in is performed and the task is assumed to be able to access inputs directly from
+; within the sandbox.
+; Type: string
+; Default: "stagein"
+
+; stageout_dir_name
+; Description: Name of a directory which is placed automatically inside a temporary directory and
+; provided to sandboxed tasks to store outputs and from which those files are staged-out outside the
+; sandbox. When "None", no stage-out is performed and the task is assumed to be able to store
+; outputs directly from within the sandbox.
+; Type: string
+; Default: "stageout"
+
+; law_executable
+; Description: The law executable to use within sandboxes, e.g. "law" or "python -m law".
+; Type: string
+; Default: "law"
+
+; login
+; Description: A boolean flag that decides whether the bash beneath the sandbox should be invoked as
+; a login shell.
+; Type: boolean
+; Default: False
+
+
+; --- cmssw_sandbox_env section --------------------------------------------------------------------
+
+[cmssw_sandbox_env]
+
+; Note:
+; Here you can define environment variables via key-value pairs that are accessible in a cmssw
+; sandbox. When an option has no value, i.e., when only a key is given, the value of the variable of
+; the current environment is used.
+; The environment variables defined in this section are applied to all cmssw sandboxes. To
+; configure variables per cmssw version, create a section "cmssw_sandbox_env_<cmssw_version>" with
+; the desired values.
diff --git a/law/config.py b/law/config.py
index 9ffa4f7e..692b2f90 100644
--- a/law/config.py
+++ b/law/config.py
@@ -246,6 +246,10 @@ def __str__(self):
             "slurm_job_file_dir_cleanup": False,
             "slurm_chunk_size_cancel": 25,
             "slurm_chunk_size_query": 25,
+            "crab_job_file_dir": None,
+            "crab_job_file_dir_cleanup": False,
+            "crab_sandbox_name": "CMSSW_10_6_30",
+            "crab_password_file": None,
         },
         "notifications": {
             "mail_recipient": None,
@@ -299,6 +303,13 @@ def __str__(self):
         },
         "singularity_sandbox_env": {},
         "singularity_sandbox_volumes": {},
+        "cmssw_sandbox": {
+            "stagein_dir_name": "stagein",
+            "stageout_dir_name": "stageout",
+            "law_executable": "law",
+            "login": False,
+        },
+        "cmssw_sandbox_env": {},
     }
 
     _config_files = ["$LAW_CONFIG_FILE", "law.cfg", law_home_path("config"), "etc/law/config"]
diff --git a/law/contrib/arc/job.py b/law/contrib/arc/job.py
index f74089fb..59e46925 100644
--- a/law/contrib/arc/job.py
+++ b/law/contrib/arc/job.py
@@ -10,7 +10,6 @@
 
 import os
 import stat
-import sys
 import time
 import re
 import random
@@ -81,7 +80,7 @@ def submit(self, job_file, job_list=None, ce=None, retries=0, retry_delay=3, sil
             # run the command
             logger.debug("submit arc job(s) with command '{}'".format(cmd))
             code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-                stdout=subprocess.PIPE, stderr=sys.stderr, cwd=job_file_dir)
+                stdout=subprocess.PIPE, cwd=job_file_dir)
 
             # in some cases, the return code is 0 but the ce did not respond valid job ids
             job_ids = []
@@ -135,7 +134,7 @@ def cancel(self, job_id, job_list=None, silent=False):
         # run it
         logger.debug("cancel arc job(s) with command '{}'".format(cmd))
         code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-            stdout=subprocess.PIPE, stderr=sys.stderr)
+            stdout=subprocess.PIPE)
 
         # check success
         if code != 0 and not silent:
@@ -163,7 +162,7 @@ def cleanup(self, job_id, job_list=None, silent=False):
         # run it
         logger.debug("cleanup arc job(s) with command '{}'".format(cmd))
         code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-            stdout=subprocess.PIPE, stderr=sys.stderr)
+            stdout=subprocess.PIPE)
 
         # check success
         if code != 0 and not silent:
diff --git a/law/contrib/arc/workflow.py b/law/contrib/arc/workflow.py
index 9e380352..d987b60b 100644
--- a/law/contrib/arc/workflow.py
+++ b/law/contrib/arc/workflow.py
@@ -139,9 +139,13 @@ def create_job_file(self, job_num, branches):
         return {"job": job_file, "log": abs_log_file}
 
     def destination_info(self):
-        info = ["ce: {}".format(",".join(self.task.arc_ce))]
+        info = super(ARCWorkflowProxy, self).destination_info()
+
+        info["ce"] = "ce: {}".format(",".join(self.task.arc_ce))
+
         info = self.task.arc_destination_info(info)
-        return ", ".join(map(str, info))
+
+        return info
 
 
 class ARCWorkflow(BaseRemoteWorkflow):
diff --git a/law/contrib/cms/__init__.py b/law/contrib/cms/__init__.py
index 23e644f5..d7b06fd8 100644
--- a/law/contrib/cms/__init__.py
+++ b/law/contrib/cms/__init__.py
@@ -4,10 +4,18 @@
 CMS-related contrib package. https://home.cern/about/experiments/cms
 """
 
-__all__ = ["CMSJobDashboard", "BundleCMSSW", "Site", "lfn_to_pfn"]
+__all__ = [
+    "CMSSWSandbox",
+    "CrabJobManager", "CrabJobFileFactory", "CMSJobDashboard",
+    "CrabWorkflow",
+    "BundleCMSSW",
+    "Site", "lfn_to_pfn", "renew_vomsproxy", "delegate_myproxy",
+]
 
 
 # provisioning imports
-from law.contrib.cms.job import CMSJobDashboard
+from law.contrib.cms.sandbox import CMSSWSandbox
+from law.contrib.cms.job import CrabJobManager, CrabJobFileFactory, CMSJobDashboard
+from law.contrib.cms.workflow import CrabWorkflow
 from law.contrib.cms.tasks import BundleCMSSW
-from law.contrib.cms.util import Site, lfn_to_pfn
+from law.contrib.cms.util import Site, lfn_to_pfn, renew_vomsproxy, delegate_myproxy
diff --git a/law/contrib/cms/crab/PSet.py b/law/contrib/cms/crab/PSet.py
new file mode 100644
index 00000000..d89f9c82
--- /dev/null
+++ b/law/contrib/cms/crab/PSet.py
@@ -0,0 +1,31 @@
+# coding: utf-8
+
+"""
+Minimal valid configuration.
+"""
+
+import FWCore.ParameterSet.Config as cms
+
+
+process = cms.Process("LAW")
+
+process.source = cms.Source(
+    "PoolSource",
+    fileNames=cms.untracked.vstring([""]),
+)
+
+process.output = cms.OutputModule(
+    "PoolOutputModule",
+    fileName=cms.untracked.string("out.root"),
+)
+
+process.maxEvents = cms.untracked.PSet(
+    input=cms.untracked.int32(1),
+)
+
+process.options = cms.untracked.PSet(
+    allowUnscheduled=cms.untracked.bool(True),
+    wantSummary=cms.untracked.bool(False),
+)
+
+process.out = cms.EndPath(process.output)
diff --git a/law/contrib/cms/crab/crab_wrapper.sh b/law/contrib/cms/crab/crab_wrapper.sh
new file mode 100644
index 00000000..31d62b03
--- /dev/null
+++ b/law/contrib/cms/crab/crab_wrapper.sh
@@ -0,0 +1,221 @@
+#!/usr/bin/env bash
+
+# Wrapper script that is to be configured as crab's scriptExe file and that
+# - sets up files and objects needed for crab to consider this a standard job,
+# - renders variables in other job input files, and
+# - runs the actual job file with arguments that depend on the crab job number.
+
+# disable exiting at first error and command-echoing
+set +e
+set +v
+
+action() {
+    #
+    # detect variables
+    #
+
+    local shell_is_zsh="$( [ -z "${ZSH_VERSION}" ] && echo "false" || echo "true" )"
+    local this_file="$( ${shell_is_zsh} && echo "${(%):-%x}" || echo "${BASH_SOURCE[0]}" )"
+    local this_file_base="$( basename "${this_file}" )"
+
+    export LAW_CRAB_JOB_NUMBER="$( ls -1 | grep -Po "jobReport\.json\.\K\d+" | head -n 1 )"
+    echo "running ${this_file_base} for job number ${LAW_CRAB_JOB_NUMBER}"
+
+
+    #
+    # patch edmProvDump
+    #
+
+    MD5SUM="$( cat "${this_file}" | md5sum | awk '{print $1}' )"
+
+    cat <<EOF > "${CMSSW_BASE}/bin/${SCRAM_ARCH}/edmProvDump"
+#!/bin/sh
+echo "Processing History:"
+echo "  USER '' '\"${CMSSW_VERSION}\"' [1]  (${MD5SUM})"
+EOF
+
+    chmod +x "${CMSSW_BASE}/bin/${SCRAM_ARCH}/edmProvDump"
+
+
+    #
+    # create dummy outputs
+    #
+
+    # output file as defined in the job report below
+    # (not required if disableAutomaticOutputCollection is set, but kept here as a good default)
+    touch out.root
+
+    # job report
+cat <<EOF > FrameworkJobReport.xml
+<FrameworkJobReport>
+  <InputFile>
+    <LFN></LFN>
+    <PFN></PFN>
+    <Catalog></Catalog>
+    <InputType>primaryFiles</InputType>
+    <ModuleLabel>source</ModuleLabel>
+    <GUID></GUID>
+    <InputSourceClass>PoolSource</InputSourceClass>
+    <EventsRead>1</EventsRead>
+  </InputFile>
+
+  <File>
+    <LFN></LFN>
+    <PFN>out.root</PFN>
+    <Catalog></Catalog>
+    <ModuleLabel>USER</ModuleLabel>
+    <GUID></GUID>
+    <OutputModuleClass>PoolOutputModule</OutputModuleClass>
+    <TotalEvents>1</TotalEvents>
+    <BranchHash>806a51af4d0c43b79de23e695823bf38</BranchHash>
+  </File>
+
+  <ReadBranches>
+  </ReadBranches>
+
+  <GeneratorInfo>
+  </GeneratorInfo>
+
+  <PerformanceReport>
+    <PerformanceSummary Metric="StorageStatistics">
+      <Metric Name="Parameter-untracked-bool-enabled" Value="true"/>
+      <Metric Name="Parameter-untracked-bool-stats" Value="true"/>
+      <Metric Name="Parameter-untracked-string-cacheHint" Value="application-only"/>
+      <Metric Name="Parameter-untracked-string-readHint" Value="auto-detect"/>
+      <Metric Name="ROOT-tfile-read-totalMegabytes" Value="0"/>
+      <Metric Name="ROOT-tfile-write-totalMegabytes" Value="0"/>
+    </PerformanceSummary>
+  </PerformanceReport>
+
+</FrameworkJobReport>
+EOF
+
+
+    #
+    # job argument definitons, depending on LAW_CRAB_JOB_NUMBER
+    #
+
+    # definition
+    local crab_job_arguments_map
+    declare -A crab_job_arguments_map
+    crab_job_arguments_map=(
+        {{crab_job_arguments_map}}
+    )
+
+    # pick
+    local crab_job_arguments="${crab_job_arguments_map[${LAW_CRAB_JOB_NUMBER}]}"
+    if [ -z "${crab_job_arguments}" ]; then
+        >&2 echo "empty crab job arguments for LAW_CRAB_JOB_NUMBER ${LAW_CRAB_JOB_NUMBER}"
+        return "1"
+    fi
+
+
+    #
+    # variable rendering
+    #
+
+    # check variables
+    local render_variables="{{render_variables}}"
+    if [ -z "${render_variables}" ]; then
+        >&2 echo "empty render variables"
+        return "2"
+    fi
+
+    # decode
+    render_variables="$( echo "${render_variables}" | base64 --decode )"
+
+    # check files to render
+    local input_files_render=( {{input_files_render}} )
+    if [ "${#input_files_render[@]}" == "0" ]; then
+        >&2 echo "received empty input files for rendering for LAW_CRAB_JOB_NUMBER ${LAW_CRAB_JOB_NUMBER}"
+        return "3"
+    fi
+
+    # render files
+    local input_file_render
+    for input_file_render in ${input_files_render[@]}; do
+        # skip if the file refers to _this_ one
+        local input_file_render_base="$( basename "${input_file_render}" )"
+        [ "${input_file_render_base}" = "${this_file_base}" ] && continue
+        # render
+        echo "render ${input_file_render}"
+        python -c "\
+import re;\
+repl = ${render_variables};\
+repl['input_files_render'] = '';\
+content = open('${input_file_render}', 'r').read();\
+content = re.sub(r'\{\{(\w+)\}\}', lambda m: repl.get(m.group(1), ''), content);\
+open('${input_file_render_base}', 'w').write(content);\
+"
+        local render_ret="$?"
+        # handle rendering errors
+        if [ "${render_ret}" != "0" ]; then
+            >&2 echo "input file rendering failed with code ${render_ret}"
+            return "4"
+        fi
+    done
+
+
+    #
+    # run the actual job file
+    #
+
+    # check the job file
+    local job_file="{{job_file}}"
+    if [ ! -f "${job_file}" ]; then
+        >&2 echo "job file '${job_file}' does not exist"
+        return "5"
+    fi
+
+    # debugging: print its contents
+    # echo "=== content of job file '${job_file}'"
+    # echo
+    # cat "${job_file}"
+    # echo
+    # echo "=== end of job file content"
+
+    # debugging: pretty print path variables
+    # log_path_var() {
+    #     python2 -c "import os; print('$1:\n  ' + '\n  '.join(os.getenv('$1', '').split(':')))"
+    # }
+    # log_path_var PATH
+    # log_path_var PYTHONPATH
+    # log_path_var LD_LIBRARY_PATH
+
+    # helper to remove fragments of a path variable
+    filter_path_var() {
+        # get arguments
+        local old_val="$1"
+        shift
+        local regexps=${@}
+
+        # loop through paths and set the new variable if no expression matched
+        local new_val=""
+        printf '%s:\0' "${old_val}" | while IFS=: read -d: -r p; do
+            local matched="false"
+            local regexp
+            for regexp in ${regexps[@]}; do
+                if echo "${p}" | grep -Po "${regexp}" &> /dev/null; then
+                    matched="true"
+                    break
+                fi
+            done
+            if ! ${matched}; then
+                [ ! -z "${new_val}" ] && new_val="${new_val}:"
+                new_val="${new_val}${p}"
+                echo "${new_val}"
+            fi
+        done | tail -n 1
+    }
+
+    # remove cmssw related variables from paths
+    local NEW_PATH="$( filter_path_var "${PATH}" "^/cvmfs/cms\.cern\.ch/" "^${CMSSW_BASE:-NOT_SET}" )"
+    local NEW_PYTHONPATH="$( filter_path_var "${PYTHONPATH}" "^/cvmfs/cms\.cern\.ch/" "python2\.7/site-packages" )"
+    local NEW_LD_LIBRARY_PATH="$( filter_path_var "${LD_LIBRARY_PATH}" "^/cvmfs/cms\.cern\.ch/" "^${CMSSW_BASE:-NOT_SET}" )"
+
+    # run it
+    echo "run actual job file"
+    PATH="${NEW_PATH}" PYTHONPATH="${NEW_PYTHONPATH}" LD_LIBRARY_PATH="${NEW_LD_LIBRARY_PATH}" bash "${job_file}" ${crab_job_arguments}
+}
+
+action "$@"
diff --git a/law/contrib/cms/job.py b/law/contrib/cms/job.py
index 2bbaf146..2139fa94 100644
--- a/law/contrib/cms/job.py
+++ b/law/contrib/cms/job.py
@@ -1,21 +1,809 @@
 # coding: utf-8
 
 """
-CMS-related job helpers.
+Crab job manager and CMS-related job helpers.
 """
 
-__all__ = ["CMSJobDashboard"]
+__all__ = ["CrabJobManager", "CrabJobFileFactory", "CMSJobDashboard"]
 
 
+import os
+import stat
 import time
 import socket
 import threading
+import re
+import json
+import subprocess
+import shutil
+from collections import OrderedDict, namedtuple
 
 import six
 
 import law
-from law.job.base import BaseJobManager
+from law.config import Config
+from law.sandbox.base import Sandbox
+from law.job.base import BaseJobManager, BaseJobFileFactory, JobInputFile, DeprecatedInputFiles
 from law.job.dashboard import BaseJobDashboard
+from law.util import (
+    DotDict, interruptable_popen, make_list, make_unique, quote_cmd, no_value, rel_path,
+)
+from law.logger import get_logger
+
+import law.contrib.cms.sandbox
+
+
+law.contrib.load("wlcg")
+
+logger = get_logger(__name__)
+
+
+class CrabJobManager(BaseJobManager):
+
+    submission_task_name_cre = re.compile(r"^Task\s+name\s*\:\s+([^\s]+)\s*$")
+    submission_log_file_cre = re.compile(r"^Log\s+file\s+is\s+([^\s]+\.log)\s*$")
+    query_server_status_cre = re.compile(r"^Status\s+on\s+the\s+CRAB\s+server\s*\:\s+([^\s].*)$")
+    query_user_cre = re.compile(r"^Task\s+name\s*:\s+\d+_\d+\:([^_]+)_.+$")
+    query_scheduler_cre = re.compile(r"^Grid\s+scheduler\s+-\s+Task\s+Worker\s*\:\s+([^\s]+).+$")
+    query_scheduler_id_cre = re.compile(r"^Grid\s+scheduler\s+-\s+Task\s+Worker\s*\:\s+crab.*\@.+[^\d](\d+)\..+$")  # noqa
+    query_scheduler_status_cre = re.compile(r"^Status\s+on\s+the\s+scheduler\s*\:\s+([^\s].*)$")
+    query_monitoring_url_cre = re.compile(r"^Dashboard\s+monitoring\s+URL\s*\:\s+([^\s].*)$")
+    query_json_line_cre = re.compile(r"^\s*(\{.+\})\s*$")
+    log_n_jobs_cre = re.compile(r"^config\.Data\.totalUnits\s+\=\s+(\d+)\s*$")
+    log_task_name_cre = re.compile(r"^.+\s+Task\s+name\s*\:\s+([^\s]+)\s*$")
+
+    log_file_pattern = "https://cmsweb.cern.ch:8443/scheddmon/{scheduler_id}/{user}/{task_name}/job_out.{crab_num}.{attempt}.txt"  # noqa
+
+    job_grouping = True
+
+    JobId = namedtuple("JobId", ["crab_num", "task_name", "proj_dir"])
+
+    def __init__(self, sandbox_name=None, proxy=None, instance=None, threads=1):
+        super(CrabJobManager, self).__init__()
+
+        # default sandbox name
+        if sandbox_name is None:
+            cfg = Config.instance()
+            sandbox_name = cfg.get_expanded("job", "crab_sandbox_name")
+
+        # create the cmssw sandbox
+        self.cmssw_sandbox = Sandbox.new("cmssw::{}".format(sandbox_name))
+
+        # store attributes
+        self.proxy = proxy
+        self.instance = instance
+        self.threads = threads
+
+    @classmethod
+    def cast_job_id(cls, job_id):
+        """
+        Converts a *job_id*, for instance after json deserialization, into a :py:class:`JobId`
+        object.
+        """
+        return cls.JobId(*job_id) if isinstance(job_id, (list, tuple)) else job_id
+
+    @property
+    def cmssw_env(self):
+        return self.cmssw_sandbox.env
+
+    def group_job_ids(self, job_ids):
+        groups = OrderedDict()
+
+        # group by project directory
+        for job_id in job_ids:
+            if job_id.proj_dir not in groups:
+                groups[job_id.proj_dir] = []
+            groups[job_id.proj_dir].append(job_id)
+
+        return groups
+
+    def _apply_group(self, func, result_type, group_func, job_objs, *args, **kwargs):
+        # when job_objs is a string or a sequence of strings, interpret them as project dirs, read
+        # their log files to extract task names, build actual job ids and forward them
+        if func != self.submit:
+            job_ids = []
+            for i, job_id in enumerate(make_list(job_objs)):
+                if not isinstance(job_id, six.string_types):
+                    job_ids.append(job_id)
+                    continue
+
+                # get n_jobs and task_name from log file
+                proj_dir = job_id
+                log_file = os.path.join(proj_dir, "crab.log")
+                log_data = self._parse_log_file(log_file)
+                if not log_data or "n_jobs" not in log_data or "task_name" not in log_data:
+                    job_ids.append(job_id)
+                    continue
+
+                # expand ids
+                for crab_num in range(1, int(log_data["n_jobs"]) + 1):
+                    job_ids.append(self.JobId(crab_num, log_data["task_name"], proj_dir))
+            job_objs = job_ids
+
+        return super(CrabJobManager, self)._apply_group(
+            func,
+            result_type,
+            group_func,
+            job_objs,
+            *args,
+            **kwargs  # noqa
+        )
+
+    def submit(self, job_file, job_files=None, proxy=None, instance=None, myproxy_username=None,
+            retries=0, retry_delay=3, silent=False):
+        # default arguments
+        if proxy is None:
+            proxy = self.proxy
+        if instance is None:
+            instance = self.instance
+
+        # get the job file location as the submission command is run it the same directory
+        job_file_dir, job_file_name = os.path.split(os.path.abspath(job_file))
+
+        # define the actual submission in a loop to simplify retries
+        while True:
+            # build the command
+            cmd = ["crab", "submit", "--config", job_file_name]
+            if proxy:
+                cmd += ["--proxy", proxy]
+            if instance:
+                cmd += ["--instance", instance]
+            cmd = quote_cmd(cmd)
+
+            # run the command
+            # crab prints everything to stdout
+            logger.debug("submit crab jobs with command '{}'".format(cmd))
+            code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
+                stdout=subprocess.PIPE, cwd=job_file_dir, env=self.cmssw_env)
+
+            # handle errors
+            if code != 0:
+                logger.debug("submission of crab job '{}' failed with code {}:\n{}".format(
+                    job_file, code, out))
+
+                # remove the project directory
+                proj_dir = self._proj_dir_from_job_file(job_file, self.cmssw_env)
+                if proj_dir and os.path.isdir(proj_dir):
+                    logger.debug("removing crab project '{}' from previous attempt".format(
+                        proj_dir))
+                    shutil.rmtree(proj_dir)
+
+                if retries > 0:
+                    retries -= 1
+                    time.sleep(retry_delay)
+                    continue
+
+                if silent:
+                    return None
+
+                raise Exception("submission of crab job '{}' failed with code {}:\n{}".format(
+                    job_file, code, out))
+
+            # parse outputs
+            task_name, log_file = None, None
+            for line in out.replace("\r", "").split("\n"):
+                if not task_name:
+                    m_task_name = self.submission_task_name_cre.match(line)
+                    if m_task_name:
+                        task_name = m_task_name.group(1)
+
+                if not log_file:
+                    m_log_file = self.submission_log_file_cre.match(line)
+                    if m_log_file:
+                        log_file = m_log_file.group(1)
+
+                if task_name and log_file:
+                    break
+
+            if not task_name:
+                raise Exception("no valid task name found in submission output:\n\n{}".format(out))
+            if not log_file:
+                raise Exception("no valid log file found in submission output:\n\n{}".format(out))
+
+            # create job ids with log data
+            proj_dir = os.path.dirname(log_file)
+            job_ids = self._job_ids_from_proj_dir(proj_dir)
+
+            # checks
+            if not job_ids:
+                raise Exception("number of jobs not extractable from log file {}".format(log_file))
+            if job_files is not None and len(job_files) != len(job_ids):
+                raise Exception(
+                    "number of submited jobs ({}) does not match number of job files ({})".format(
+                        len(job_ids), len(job_file)),
+                )
+
+            return job_ids
+
+    def cancel(self, proj_dir, job_ids=None, proxy=None, instance=None, myproxy_username=None,
+            silent=False):
+        if job_ids is None:
+            job_ids = self._job_ids_from_proj_dir(proj_dir)
+
+        # build the command
+        cmd = ["crab", "kill", "--dir", proj_dir]
+        if proxy:
+            cmd += ["--proxy", proxy]
+        if instance:
+            cmd += ["--instance", instance]
+        cmd = quote_cmd(cmd)
+
+        # run it
+        logger.debug("cancel crab job(s) with command '{}'".format(cmd))
+        code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
+            stdout=subprocess.PIPE, env=self.cmssw_env)
+
+        # check success
+        if code != 0 and not silent:
+            # crab prints everything to stdout
+            raise Exception(
+                "cancellation of crab jobs from project '{}' failed with code {}:\n{}".format(
+                    proj_dir, code, out),
+            )
+
+        return {job_id: None for job_id in job_ids}
+
+    def cleanup(self, proj_dir, job_ids=None, proxy=None, instance=None, myproxy_username=None,
+            silent=False):
+        if job_ids is None:
+            job_ids = self._job_ids_from_proj_dir(proj_dir)
+
+        # just delete the project directory
+        if os.path.isdir(proj_dir):
+            shutil.rmtree(proj_dir)
+
+        return {job_id: None for job_id in job_ids}
+
+    def query(self, proj_dir, job_ids=None, proxy=None, instance=None, myproxy_username=None,
+            silent=False):
+        if job_ids is None:
+            job_ids = self._job_ids_from_proj_dir(proj_dir)
+
+        # build the command
+        cmd = ["crab", "status", "--dir", proj_dir, "--json"]
+        if proxy:
+            cmd += ["--proxy", proxy]
+        if instance:
+            cmd += ["--instance", instance]
+        cmd = quote_cmd(cmd)
+
+        # run it
+        logger.debug("query crab job(s) with command '{}'".format(cmd))
+        code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
+            stdout=subprocess.PIPE, env=self.cmssw_env)
+
+        # handle errors
+        if code != 0:
+            if silent:
+                return None
+            # crab prints everything to stdout
+            raise Exception(
+                "status query of crab jobs from project '{}' failed with code {}:\n{}".format(
+                    proj_dir, code, out),
+            )
+
+        # parse the output and extract the status per job
+        query_data = self.parse_query_output(out, proj_dir, job_ids)
+
+        # compare to the requested job ids and perform some checks
+        for job_id in job_ids:
+            if job_id not in query_data:
+                query_data[job_id] = self.job_status_dict(
+                    job_id=job_id,
+                    status=self.FAILED,
+                    error="job not found in query response",
+                )
+
+        return query_data
+
+    @classmethod
+    def parse_query_output(cls, out, proj_dir, job_ids):
+        # parse values using compiled regexps
+        cres = [
+            cls.query_user_cre,
+            cls.query_server_status_cre,
+            cls.query_scheduler_id_cre,
+            cls.query_scheduler_status_cre,
+            cls.query_json_line_cre,
+            cls.query_monitoring_url_cre,
+        ]
+        values = len(cres) * [None]
+        for line in out.replace("\r", "").split("\n"):
+            for i, (cre, value) in enumerate(zip(cres, values)):
+                if value:
+                    continue
+                m = cre.match(line)
+                if m:
+                    values[i] = m.group(1)
+            if all(values):
+                break
+
+        # unpack
+        username, server_status, scheduler_id, scheduler_status, json_line, monitoring_url = values
+
+        # helper to build extra info
+        def extra(job_id, job_data=None):
+            extra = {}
+            if username and scheduler_id and job_data:
+                extra["log_file"] = cls.log_file_pattern.format(
+                    scheduler_id=scheduler_id,
+                    user=username,
+                    task_name=job_id.task_name,
+                    crab_num=job_id.crab_num,
+                    attempt=job_data.get("Retries", 0),
+                )
+            if monitoring_url:
+                extra["tracking_url"] = monitoring_url
+            return extra
+
+        # in case scheduler status or the json line is missing, the submission could be too new
+        if not scheduler_status or not json_line:
+            accepted_server_states = [
+                "HOLDING on command SUBMIT",
+                "NEW on command SUBMIT",
+                "QUEUED on command SUBMIT",
+                "SUBMITTED",
+            ]
+            if server_status not in accepted_server_states:
+                s = ",".join(map("'{}'".format, accepted_server_states))
+                raise Exception(
+                    "no per-job information available (yet?), which is only accepted if the crab " +
+                    "server status is any of {}, but got '{}'".format(s, server_status),
+                )
+            # interpret all jobs as pending
+            return {
+                job_id: cls.job_status_dict(job_id=job_id, status=cls.PENDING, extra=extra(job_id))
+                for job_id in job_ids
+            }
+
+        # parse json data
+        if not json_line:
+            raise Exception(
+                "no per-job information available in status response, crab server " +
+                "status '{}', scheduler status '{}'".format(server_status, scheduler_status),
+            )
+
+        # map of crab job numbers to full ids for faster lookup
+        num_to_id_map = {job_id.crab_num: job_id for job_id in job_ids}
+
+        # build query data
+        query_data = {}
+        for crab_num_str, data in json.loads(json_line).items():
+            crab_num = int(crab_num_str)
+            if crab_num not in num_to_id_map:
+                continue
+            job_id = num_to_id_map[crab_num]
+
+            # parse error info
+            code = None
+            error = data.get("Error")
+            if isinstance(error, list) and len(error) >= 2:
+                code = error[0]
+                error = str(error[1]).strip()
+
+            # extra info
+            _extra = extra(job_id, data) or {}
+            _extra["site_history"] = data.get("SiteHistory")
+
+            # fill query data
+            query_data[job_id] = cls.job_status_dict(
+                job_id=job_id,
+                status=cls.map_status(data["State"]),
+                code=code,
+                error=error,
+                extra=_extra,
+            )
+
+        return query_data
+
+    @classmethod
+    def _proj_dir_from_job_file(cls, job_file, cmssw_env):
+        work_area = None
+        request_name = None
+
+        with open(job_file, "r") as f:
+            # fast approach: parse the job file
+            for line in f.readlines():
+                if work_area and request_name:
+                    break
+                if not work_area:
+                    m = re.match(r"^.+[^\w]workArea\s*=\s(\'|\")(.+)(\'|\").*$", line.strip())
+                    if m:
+                        work_area = m.group(2)
+                        continue
+                if not request_name:
+                    m = re.match(r"^.+[^\w]requestName\s*=\s(\'|\")(.+)(\'|\").*$", line.strip())
+                    if m:
+                        request_name = m.group(2)
+                        continue
+
+            # when the combination is correct, return
+            if work_area and request_name and 0:
+                path = os.path.join(work_area, "crab_{}".format(request_name))
+                path = os.path.expandvars(os.path.expanduser(path))
+                if os.path.isdir(path):
+                    return path
+
+            # long approach: read the file in the cmssw env and manually print
+            m = re.match(r"^CMSSW(_.+|)_(\d)+_\d+_\d+.*$", cmssw_env["CMSSW_VERSION"])
+            cmssw_major = int(m.group(2)) if m else None
+            py_exec = "python3" if cmssw_major is None or cmssw_major >= 11 else "python"
+            cmd = """{} -c '
+from os.path import join
+with open("{}", "r") as f:
+    mod = dict()
+    exec(f.read(), mod)
+    cfg = mod["cfg"]
+print(join(cfg.General.workArea, "crab_" + cfg.General.requestName))'""".format(py_exec, job_file)
+            code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
+                stdout=subprocess.PIPE, env=cmssw_env)
+            if code == 0:
+                path = out.strip().replace("\r\n", "\n").split("\n")[-1]
+                path = os.path.expandvars(os.path.expanduser(path))
+                if os.path.isdir(path):
+                    return path
+
+        return None
+
+    @classmethod
+    def _parse_log_file(cls, log_file):
+        log_file = os.path.expandvars(os.path.expanduser(log_file))
+        if not os.path.exists(log_file):
+            return None
+
+        cres = [cls.log_n_jobs_cre, cls.log_task_name_cre]
+        names = ["n_jobs", "task_name"]
+        values = len(cres) * [None]
+
+        with open(log_file, "r") as f:
+            for line in f.readlines():
+                for i, (cre, value) in enumerate(zip(cres, values)):
+                    if value:
+                        continue
+                    m = cre.match(line)
+                    if m:
+                        values[i] = m.group(1)
+                if all(values):
+                    break
+
+        return dict(zip(names, values))
+
+    @classmethod
+    def _job_ids_from_proj_dir(cls, proj_dir):
+        # read log data
+        log_data = cls._parse_log_file(os.path.join(proj_dir, "crab.log"))
+        if not log_data or "n_jobs" not in log_data or "task_name" not in log_data:
+            return None
+
+        # build and return ids
+        return [
+            cls.JobId(crab_num, log_data["task_name"], proj_dir)
+            for crab_num in range(1, int(log_data["n_jobs"]) + 1)
+        ]
+
+    @classmethod
+    def map_status(cls, status):
+        # see https://twiki.cern.ch/twiki/bin/view/CMSPublic/Crab3HtcondorStates
+        if status in ("cooloff", "unsubmitted", "idle"):
+            return cls.PENDING
+        if status in ("running", "transferring", "transferred"):
+            return cls.RUNNING
+        if status in ("finished",):
+            return cls.FINISHED
+        if status in ("killing", "failed", "held"):
+            return cls.FAILED
+        return cls.FAILED
+
+
+class CrabJobFileFactory(BaseJobFileFactory):
+
+    config_attrs = BaseJobFileFactory.config_attrs + [
+        "file_name", "executable", "arguments", "work_area", "request_name", "input_files",
+        "output_files", "storage_site", "output_lfn_base", "vo_group", "vo_role", "crab",
+        "custom_content", "absolute_paths",
+    ]
+
+    def __init__(self, file_name="crab_job.py", executable=None, arguments=None, work_area=None,
+            request_name=None, input_files=None, output_files=None, storage_site=None,
+            output_lfn_base=None, vo_group=None, vo_role=None, custom_content=None,
+            absolute_paths=False, **kwargs):
+        # get some default kwargs from the config
+        cfg = Config.instance()
+        default_dir = cfg.get_expanded(
+            "job",
+            cfg.find_option("job", "crab_job_file_dir", "job_file_dir"),
+        )
+        if kwargs.get("dir") is None:
+            kwargs["dir"] = cfg.get_expanded(
+                "job",
+                cfg.find_option("job", "crab_job_file_dir", "job_file_dir"),
+            )
+        if kwargs.get("cleanup") is None:
+            kwargs["cleanup"] = cfg.get_expanded_bool(
+                "job",
+                cfg.find_option("job", "crab_job_file_dir_cleanup", "job_file_dir_cleanup"),
+            )
+        if kwargs.get("mkdtemp") is None:
+            kwargs["mkdtemp"] = cfg.get_expanded_bool(
+                "job",
+                cfg.find_option("job", "crab_job_file_dir_mkdtemp", "job_file_dir_mkdtemp"),
+            )
+
+        super(CrabJobFileFactory, self).__init__(**kwargs)
+
+        self.file_name = file_name
+        self.executable = executable
+        self.arguments = arguments
+        self.work_area = work_area
+        self.request_name = request_name
+        self.input_files = DeprecatedInputFiles(input_files or {})
+        self.output_files = output_files or []
+        self.storage_site = storage_site
+        self.output_lfn_base = output_lfn_base
+        self.vo_group = vo_group
+        self.vo_role = vo_role
+        self.custom_content = custom_content
+        self.absolute_paths = absolute_paths
+
+        # defaults
+        if not self.work_area:
+            self.work_area = default_dir
+
+        # crab config
+        # "no_value" marks required settings, None marks optional settings
+        self.crab = DotDict([
+            ("General", DotDict([
+                ("requestName", no_value),
+                ("workArea", no_value),
+                ("transferLogs", False),
+                ("transferOutputs", no_value),
+            ])),
+            ("JobType", DotDict([
+                ("pluginName", "Analysis"),
+                ("psetName", rel_path(__file__, "crab", "PSet.py")),
+                ("scriptExe", no_value),
+                ("maxMemoryMB", 2048),
+                ("allowUndistributedCMSSW", True),
+                ("sendPythonFolder", False),
+                ("disableAutomaticOutputCollection", True),
+                ("inputFiles", no_value),
+                ("outputFiles", no_value),
+            ])),
+            ("Data", DotDict([
+                ("inputDBS", "global"),
+                ("splitting", "FileBased"),
+                ("unitsPerJob", 1),
+                ("totalUnits", no_value),
+                ("allowNonValidInputDataset", True),
+                ("outLFNDirBase", no_value),
+                ("publication", False),
+                ("ignoreLocality", False),
+            ])),
+            ("Site", DotDict([
+                ("storageSite", no_value),
+            ])),
+            ("User", DotDict([
+                ("vo_group", None),
+                ("vo_role", None),
+            ])),
+        ])
+
+    def create(self, **kwargs):
+        # merge kwargs and instance attributes
+        c = self.get_config(**kwargs)
+
+        # some sanity checks
+        if not c.file_name:
+            raise ValueError("file_name must not be empty")
+        if not c.executable:
+            raise ValueError("executable must not be empty")
+        if not c.request_name:
+            raise ValueError("request_name must not be empty")
+        if len(c.request_name) > 100:
+            raise ValueError(
+                "request_name must be less then 100 characters long, got {}: {}".format(
+                    len(c.request_name), c.request_name),
+            )
+        if not c.output_lfn_base:
+            raise ValueError("output_lfn_base must not be empty")
+        if not c.storage_site:
+            raise ValueError("storage_site must not be empty")
+        if not isinstance(c.arguments, (list, tuple)):
+            raise ValueError("arguments must be a list, got '{}'".format(c.arguments))
+        if "job_file" not in c.input_files:
+            raise ValueError("an input file with key 'job_file' is required")
+
+        # ensure that all log files are output files
+        for attr in ["custom_log_file"]:
+            if c[attr] and c[attr] not in c.output_files:
+                c.output_files.append(c[attr])
+
+        # ensure that all input files are JobInputFile's
+        c.input_files = {
+            key: JobInputFile(f)
+            for key, f in c.input_files.items()
+        }
+
+        # ensure that the executable is an input file, remember the key to access it
+        if c.executable:
+            executable_keys = [k for k, v in c.input_files.items() if v == c.executable]
+            if executable_keys:
+                executable_key = executable_keys[0]
+            else:
+                executable_key = "executable_file"
+                c.input_files[executable_key] = JobInputFile(c.executable)
+
+        # add the wrapper file to the inputs
+        c.input_files["crab_wrapper"] = JobInputFile(
+            path=rel_path(__file__, "crab", "crab_wrapper.sh"),
+            copy=True,
+            render=True,
+        )
+
+        # prepare input files
+        def prepare_input(f):
+            # when not copied, just return the absolute, original path
+            abs_path = os.path.abspath(f.path)
+            if f.copy:
+                # copy the file and apply other transformations
+                abs_path = self.provide_input(
+                    src=abs_path,
+                    dir=c.dir,
+                    skip_existing=f.share,
+                )
+            return abs_path
+
+        # absolute input paths
+        for key, f in c.input_files.items():
+            f.path_sub_abs = prepare_input(f)
+
+        # input paths relative to the submission dir
+        for key, f in c.input_files.items():
+            f.path_sub_rel = (
+                os.path.basename(f.path_sub_abs)
+                if f.copy and not c.absolute_paths
+                else f.path_sub_abs
+            )
+
+        # input paths as seen by the job, before and after potential job-side rendering
+        for key, f in c.input_files.items():
+            f.path_job_pre_render = os.path.basename(f.path_sub_abs)
+            f.path_job_post_render = f.path_job_pre_render
+
+        # update files in render variables with that after potential rendering
+        c.render_variables.update({
+            key: f.path_job_post_render
+            for key, f in c.input_files.items()
+        })
+
+        # add space separated input files before potential rendering to render variables
+        c.render_variables["input_files"] = " ".join(
+            f.path_job_pre_render
+            for f in c.input_files.values()
+        )
+
+        # add space separated list of input files for rendering
+        c.render_variables["input_files_render"] = " ".join(
+            f.path_job_pre_render
+            for f in c.input_files.values()
+            if f.render_job
+        )
+
+        # add the custom log file to render variables
+        if c.custom_log_file:
+            c.render_variables["log_file"] = c.custom_log_file
+
+        # inject arguments into the crab wrapper via render variables
+        c.render_variables["crab_job_arguments_map"] = ("\n" + 8 * " ").join(
+            "['{}']=\"{}\"".format(i + 1, str(args))
+            for i, args in enumerate(c.arguments)
+        )
+
+        # linearize render variables
+        render_variables = self.linearize_render_variables(c.render_variables)
+
+        # prepare the job file
+        job_file = self.postfix_input_file(os.path.join(c.dir, c.file_name))
+
+        # render copied input files
+        for key, f in c.input_files.items():
+            if not f.copy or not f.render_local:
+                continue
+            self.render_file(
+                f.path_sub_abs,
+                f.path_sub_abs,
+                render_variables,
+            )
+
+        # prepare the executable when given
+        if c.executable:
+            c.executable = c.input_files[executable_key].path_job_post_render
+            # make the file executable for the user and group
+            path = os.path.join(c.dir, os.path.basename(c.executable))
+            if os.path.exists(path):
+                os.chmod(path, os.stat(path).st_mode | stat.S_IXUSR | stat.S_IXGRP)
+
+        # resolve work_area relative to self.dir
+        if not c.work_area:
+            c.work_area = self.dir
+        else:
+            c.work_area = os.path.join(self.dir, os.path.expandvars(os.path.expanduser(c.work_area)))
+
+        # General
+        c.crab.General.requestName = c.request_name
+        c.crab.General.workArea = c.work_area
+        c.crab.General.transferOutputs = bool(c.output_files)
+
+        # JobType
+        c.crab.JobType.scriptExe = c.input_files["crab_wrapper"].path_sub_rel
+        c.crab.JobType.inputFiles = make_unique([
+            f.path_sub_rel
+            for f in c.input_files.values()
+        ])
+        c.crab.JobType.outputFiles = make_unique(c.output_files) or None
+
+        # Data
+        c.crab.Data.totalUnits = len(c.arguments)
+        c.crab.Data.outLFNDirBase = c.output_lfn_base
+        c.crab.Data.userInputFiles = [
+            "input_{}.root".format(i + 1)
+            for i in range(c.crab.Data.totalUnits)
+        ]
+
+        # Site
+        c.crab.Site.storageSite = c.storage_site
+
+        # User
+        if c.vo_group:
+            c.crab.User.voGroup = c.vo_group
+        if c.vo_role:
+            c.crab.role.voRole = c.vo_role
+
+        # write the job file
+        self.write_crab_config(job_file, c.crab, custom_content=c.custom_content)
+
+        logger.debug("created crab job file at '{}'".format(job_file))
+
+        return job_file, c
+
+    @classmethod
+    def write_crab_config(cls, job_file, crab_config, custom_content=None):
+        fmt_flat = lambda s: "\"{}\"".format(s) if isinstance(s, six.string_types) else str(s)
+
+        with open(job_file, "w") as f:
+            # header
+            f.write("# coding: utf-8\n")
+            f.write("\n")
+            f.write("from CRABClient.UserUtilities import config\n")
+            f.write("\n")
+            f.write("cfg = config()\n")
+            f.write("\n")
+
+            # sections
+            for section, cfg in crab_config.items():
+                f.write("cfg.section_(\"{}\")\n".format(section))
+                # options
+                for option, value in cfg.items():
+                    if value == no_value:
+                        raise Exception(
+                            "cannot assign {} to crab config {}.{}".format(value, section, option),
+                        )
+                    if value is None:
+                        continue
+                    value_str = (
+                        "[\n{}\n]".format("\n".join("    {},".format(fmt_flat(v)) for v in value))
+                        if isinstance(value, (list, tuple))
+                        else fmt_flat(value)
+                    )
+                    f.write("cfg.{}.{} = {}\n".format(section, option, value_str))
+                f.write("\n")
+
+            # custom content
+            if isinstance(custom_content, six.string_types):
+                f.write(custom_content + "\n")
+            elif isinstance(custom_content, (list, tuple)):
+                for line in custom_content:
+                    f.write(str(line) + "\n")
 
 
 class CMSJobDashboard(BaseJobDashboard):
diff --git a/law/contrib/cms/sandbox.py b/law/contrib/cms/sandbox.py
new file mode 100644
index 00000000..9e048103
--- /dev/null
+++ b/law/contrib/cms/sandbox.py
@@ -0,0 +1,207 @@
+# coding: utf-8
+
+"""
+CMS related sandbox implementations.
+"""
+
+__all__ = ["CMSSWSandbox"]
+
+
+import os
+import collections
+
+import six
+
+from law.sandbox.base import _current_sandbox
+from law.sandbox.bash import BashSandbox
+from law.util import (
+    tmp_file, interruptable_popen, quote_cmd, flatten, makedirs, rel_path, law_home_path,
+    create_hash,
+)
+
+
+class CMSSWSandbox(BashSandbox):
+
+    sandbox_type = "cmssw"
+
+    # type for sandbox variables
+    # (names corresond to variables used in setup_cmssw.sh script)
+    Variables = collections.namedtuple(
+        "Variables",
+        ["version", "setup", "dir", "arch", "cores"],
+    )
+
+    @classmethod
+    def create_variables(cls, s):
+        # input format: <cmssw_version>[::<other_var=value>[::...]]
+        if not s:
+            raise ValueError("cannot create {} variables from input '{}'".format(cls.__name__, s))
+
+        # split values
+        values = {}
+        for i, part in enumerate(s.split(cls.delimiter)):
+            if i == 0:
+                values["version"] = part
+                continue
+            if "=" not in part:
+                raise ValueError(
+                    "wrong format, part '{}' at index {} does not contain a '='".format(part, i),
+                )
+            field, value = part.split("=", 1)
+            if field not in cls.Variables._fields:
+                raise KeyError("unknown variable name '{}' at index {}".format(field, i))
+            values[field] = value
+
+        # special treatments
+        expand = lambda p: os.path.abspath(os.path.expandvars(os.path.expanduser(p)))
+        if "setup" in values:
+            values["setup"] = expand(values["setup"])
+        if "dir" in values:
+            values["dir"] = expand(values["dir"])
+        else:
+            h = create_hash((cls.sandbox_type, values["version"], values.get("setup")))
+            values["dir"] = law_home_path("cms", "cmssw", "{}_{}".format(values["version"], h))
+
+        return cls.Variables(*[values.get(field, "") for field in cls.Variables._fields])
+
+    def __init__(self, *args, **kwargs):
+        super(CMSSWSandbox, self).__init__(*args, **kwargs)
+
+        # parse name into variables
+        self.variables = self.create_variables(self.name)
+
+        # when no env cache path was given, set it to a deterministic path in LAW_HOME
+        if not self.env_cache_path:
+            h = create_hash((self.sandbox_type, self.env_cache_key))
+            self.env_cache_path = law_home_path(
+                "cms",
+                "{}_cache".format(self.sandbox_type),
+                "{}_{}.pkl".format(self.variables.version, h),
+            )
+
+    def is_active(self):
+        # check if any current sandbox matches the version, setup and dir of this one
+        for key in _current_sandbox:
+            _type, name = self.split_key(key)
+            if _type != self.sandbox_type:
+                continue
+            variables = self.create_variables(name)
+            if variables[:3] == self.variables[:3]:
+                return True
+
+        return False
+
+    def get_custom_config_section_postfix(self):
+        return self.variables.version
+
+    @property
+    def env_cache_key(self):
+        return self.variables[:3]
+
+    @property
+    def script(self):
+        return rel_path(__file__, "scripts", "setup_cmssw.sh")
+
+    def create_env(self):
+        # strategy: create a tempfile, let python dump its full env in a subprocess and load the
+        # env file again afterwards
+
+        # helper to write the env
+        def write_env(path):
+            # get the bash command
+            bash_cmd = self._bash_cmd()
+
+            # build commands to setup the environment
+            setup_cmds = self._build_setup_cmds(self._get_env())
+
+            # build script variable exports
+            export_cmds = self._build_setup_cmds(collections.OrderedDict(
+                ("LAW_CMSSW_{}".format(attr.upper()), value)
+                for attr, value in zip(self.variables._fields, self.variables)
+            ))
+
+            # build the python command that dumps the environment
+            py_cmd = "import os,pickle;" \
+                + "pickle.dump(dict(os.environ),open('{}','wb'),protocol=2)".format(path)
+
+            # build the full command
+            cmd = quote_cmd(bash_cmd + ["-c", " && ".join(flatten(
+                export_cmds,
+                "source \"{}\" \"\"".format(self.script),
+                setup_cmds,
+                quote_cmd(["python", "-c", py_cmd]),
+            ))])
+
+            # run it
+            returncode = interruptable_popen(cmd, shell=True, executable="/bin/bash")[0]
+            if returncode != 0:
+                raise Exception("bash sandbox env loading failed with exit code {}".format(
+                    returncode))
+
+        # helper to load the env
+        def load_env(path):
+            pickle_kwargs = {"encoding": "utf-8"} if six.PY3 else {}
+            with open(path, "rb") as f:
+                try:
+                    return collections.OrderedDict(six.moves.cPickle.load(f, **pickle_kwargs))
+                except Exception as e:
+                    raise Exception(
+                        "env deserialization of sandbox {} failed: {}".format(self, e),
+                    )
+
+        # use the cache path if set
+        if self.env_cache_path:
+            # write it if it does not exist yet
+            if not os.path.exists(self.env_cache_path):
+                makedirs(os.path.dirname(self.env_cache_path))
+                write_env(self.env_cache_path)
+
+            # load it
+            env = load_env(self.env_cache_path)
+
+        else:
+            # use a temp file
+            with tmp_file() as tmp:
+                tmp_path = os.path.realpath(tmp[1])
+
+                # write and load it
+                write_env(tmp_path)
+                env = load_env(tmp_path)
+
+        return env
+
+    def cmd(self, proxy_cmd):
+        # environment variables to set
+        env = self._get_env()
+
+        # add staging directories
+        if self.stagein_info:
+            env["LAW_SANDBOX_STAGEIN_DIR"] = self.stagein_info.stage_dir.path
+        if self.stageout_info:
+            env["LAW_SANDBOX_STAGEOUT_DIR"] = self.stageout_info.stage_dir.path
+
+        # get the bash command
+        bash_cmd = self._bash_cmd()
+
+        # build commands to setup the environment
+        setup_cmds = self._build_setup_cmds(env)
+
+        # build script variable exports
+        export_cmds = self._build_setup_cmds(collections.OrderedDict(
+            ("LAW_CMSSW_{}".format(attr.upper()), value)
+            for attr, value in zip(self.variables._fields, self.variables)
+        ))
+
+        # handle local scheduling within the container
+        if self.force_local_scheduler():
+            proxy_cmd.add_arg("--local-scheduler", "True", overwrite=True)
+
+        # build the final command
+        cmd = quote_cmd(bash_cmd + ["-c", " && ".join(flatten(
+            export_cmds,
+            "source \"{}\" \"\"".format(self.script),
+            setup_cmds,
+            proxy_cmd.build(),
+        ))])
+
+        return cmd
diff --git a/law/contrib/cms/scripts/law_cms_delegate_myproxy b/law/contrib/cms/scripts/law_cms_delegate_myproxy
new file mode 100755
index 00000000..260e595a
--- /dev/null
+++ b/law/contrib/cms/scripts/law_cms_delegate_myproxy
@@ -0,0 +1,100 @@
+#!/usr/bin/env python
+
+"""
+Script to trigger a myproxy delegation, e.g. for crab submission.
+"""
+
+
+def delegate(renew, endpoint, username, password_file, vo, voms_proxy):
+    import law
+
+    law.contrib.load("cms", "wlcg")
+
+    # when not renewing, check if a previous delegation exists
+    if not renew:
+        info = law.wlcg.get_myproxy_info(
+            endpoint=endpoint,
+            username=username,
+            silent=True,
+        )
+        if info:
+            print("existing myproxy delegation found for username {}".format(info["username"]))
+            return
+
+    if voms_proxy and (renew or not law.wlcg.check_vomsproxy_validity()):
+        print("\nrenewing voms-proxy")
+        law.cms.renew_vomsproxy(vo=vo, password_file=password_file)
+
+    # create a new delegation
+    print("\ndelegating to {}".format(endpoint))
+    law.cms.delegate_myproxy(
+        endpoint=endpoint,
+        username=username,
+        password_file=password_file,
+        vo=vo,
+    )
+
+
+def main():
+    from argparse import ArgumentParser
+
+    import law
+
+    default_pf = law.config.get_expanded("job", "crab_password_file")
+
+    parser = ArgumentParser(
+        prog="law_cms_delegate_myproxy",
+        description="delegates an X509 proxy to a myproxy server",
+    )
+    parser.add_argument(
+        "--renew",
+        "-r",
+        action="store_true",
+        help="renew the delegation even if an existing one was found",
+    )
+    parser.add_argument(
+        "--endpoint",
+        "-e",
+        default="myproxy.cern.ch",
+        help="the server endpoint; default: myproxy.cern.ch",
+    )
+    parser.add_argument(
+        "--password-file",
+        "-p",
+        help="a file containing the certificate password" + (
+            "; default: {}".format(default_pf) if default_pf else ""
+        ),
+        default=default_pf,
+    )
+    parser.add_argument(
+        "--username",
+        "-u",
+        help="a custom user name",
+    )
+    parser.add_argument(
+        "--vo",
+        "-m",
+        default="cms",
+        help="virtual organization to use; default: cms",
+    )
+    parser.add_argument(
+        "--voms-proxy",
+        "-v",
+        action="store_true",
+        help="create a voms-proxy prior to the delegation",
+    )
+    args = parser.parse_args()
+
+    delegate(
+        args.renew,
+        args.endpoint,
+        args.username,
+        args.password_file,
+        args.vo,
+        args.voms_proxy,
+    )
+
+
+# entry hook
+if __name__ == "__main__":
+    main()
diff --git a/law/contrib/cms/scripts/law_cms_delegate_myproxy3 b/law/contrib/cms/scripts/law_cms_delegate_myproxy3
new file mode 100755
index 00000000..eca4fd89
--- /dev/null
+++ b/law/contrib/cms/scripts/law_cms_delegate_myproxy3
@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+
+action() {
+    local shell_is_zsh="$( [ -z "${ZSH_VERSION}" ] && echo "false" || echo "true" )"
+    local this_file="$( ${shell_is_zsh} && echo "${(%):-%x}" || echo "${BASH_SOURCE[0]}" )"
+    local this_dir="$( cd "$( dirname "${this_file}" )" && pwd )"
+
+    python3 "${this_dir}/law_cms_delegate_myproxy" "$@"
+}
+action "$@"
diff --git a/law/contrib/cms/scripts/setup_cmssw.sh b/law/contrib/cms/scripts/setup_cmssw.sh
new file mode 100644
index 00000000..85101255
--- /dev/null
+++ b/law/contrib/cms/scripts/setup_cmssw.sh
@@ -0,0 +1,193 @@
+#!/usr/bin/env bash
+
+# Script that installs, removes and / or sources a CMSSW environment. Distinctions are made
+# depending on whether a valid CMSSW environment is already present and sourced. If so, this scipt
+# only makes sure that crab is available. If not, a new CMSSW environment is created. It can be
+# configured through a set of variables:
+#
+#   - LAW_CMSSW_VERSION:
+#       The CMSSW version to install. Required.
+#   - LAW_CMSSW_ARCH:
+#       The scram architecture to use. Optional.
+#   - LAW_CMSSW_SETUP:
+#       A path to a script that is called inside the src directory upon installation. Optional.
+#   - LAW_CMSSW_DIR:
+#       The directory in which CMSSW is installed. Defaults to $LAW_HOME/cms/cmssw.
+#   - LAW_CMSSW_CORES:
+#       The number of cores to use for compilation. Defaults to 1.
+
+setup_cmssw() {
+    local shell_is_zsh="$( [ -z "${ZSH_VERSION}" ] && echo "false" || echo "true" )"
+    local this_file="$( ${shell_is_zsh} && echo "${(%):-%x}" || echo "${BASH_SOURCE[0]}" )"
+    local this_dir="$( cd "$( dirname "${this_file}" )" && pwd )"
+    local orig_dir="$( pwd )"
+
+
+    #
+    # helpers
+    #
+
+    source_crab() {
+        # check the python package and executable
+        type crab &> /dev/null && python3 -c "import CRABClient" &> /dev/null && return "0"
+
+        local crab_source="/cvmfs/cms.cern.ch/crab3/crab.sh"
+        if ! source "${crab_source}" ""; then
+            >&2 echo "error while sourcing crab setup file ${crab_source}"
+            return "1"
+        fi
+    }
+
+    source_cmssw() {
+        type scram &> /dev/null && return "0"
+
+        local cmssw_source="/cvmfs/cms.cern.ch/cmsset_default.sh"
+        if ! source "${cmssw_source}" ""; then
+            >&2 echo "error while sourcing cmssw setup file ${cmssw_source}"
+            return "2"
+        fi
+    }
+
+
+    #
+    # detect and setup existing cmssw
+    #
+
+    # detect
+    local outer_cmssw_exists="false"
+    if [ ! -z "${CMSSW_VERSION}" ] && [ -d "${CMSSW_BASE}/src" ] && [ -d "${CMSSW_RELEASE_BASE}" ]; then
+        outer_cmssw_exists="true"
+    fi
+
+    # setup
+    if ${outer_cmssw_exists}; then
+        # just make sure crab is available
+        source_crab || return "$?"
+
+        # setup is done
+        return "0"
+    fi
+
+
+    #
+    # check arguments / required env variables for custom setup
+    #
+
+    # read variables
+    local custom_cmssw_version="${LAW_CMSSW_VERSION}"
+    local custom_scram_arch="${LAW_CMSSW_ARCH}"
+    local custom_setup_script="${LAW_CMSSW_SETUP}"
+    local custom_install_dir="${LAW_CMSSW_DIR}"
+    local custom_install_cores="${LAW_CMSSW_CORES:-1}"
+
+    # checks
+    if [ -z "${custom_cmssw_version}" ]; then
+        >&2 echo "no custom cmssw version defined, please set LAW_CMSSW_VERSION"
+        return "3"
+    fi
+    if [ ! -z "${custom_setup_script}" ] && [ ! -f "${custom_setup_script}" ]; then
+        >&2 echo "custom setup script ${custom_setup_script} (LAW_CMSSW_SETUP) does not exist"
+        return "4"
+    fi
+    if [ -z "${custom_install_dir}" ]; then
+        if [ -z "${LAW_HOME}" ]; then
+            >&2 echo "cannot determine custom install dir, when LAW_CMSSW_DIR is empty, LAW_HOME needs to be set"
+            return "5"
+        fi
+        custom_install_dir="${LAW_HOME}/cms/cmssw"
+    fi
+
+
+    #
+    # setup new cmssw
+    #
+
+    # source cmssw and crab setup
+    source_cmssw || return "$?"
+    source_crab || return "$?"
+
+    # define variables
+    local custom_cmssw_dir="${custom_install_dir}/${custom_cmssw_version}"
+    local custom_src_dir="${custom_cmssw_dir}/src"
+    local pending_flag_file="${custom_install_dir}/pending_${custom_cmssw_version}"
+
+    # from here onwards, files and directories could be created and in order to prevent race
+    # conditions from multiple processes, guard the setup with a pending_flag_file and
+    # sleep for a random amount of seconds between 0 and 5 to further reduce the chance of
+    # simultaneously starting processes reaching this point at the same time
+    sleep "$( python3 -c 'import random;print(random.random() * 5)' )"
+    # if an existing flag file is older than 25 minutes, consider it a dangling leftover from a
+    # previously failed installation attempt and delete it
+    if [ -f "${pending_flag_file}" ]; then
+        local flag_file_age="$(( $( date +%s ) - $( date +%s -r "${pending_flag_file}" ) ))"
+        [ "${flag_file_age}" -ge "1500" ] && rm -f "${pending_flag_file}"
+    fi
+    # start the sleep loop
+    local sleep_counter="0"
+    while [ -f "${pending_flag_file}" ]; do
+        # wait at most 20 minutes
+        sleep_counter="$(( ${sleep_counter} + 1 ))"
+        if [ "${sleep_counter}" -ge 120 ]; then
+            >&2 echo "installation of ${custom_cmssw_dir} is done in different process, but number of sleeps exceeded"
+            return "6"
+        fi
+        echo "installation of ${custom_cmssw_dir} is done in different process, sleep ${sleep_counter} / 120"
+        sleep 10
+    done
+
+    # install it if still not existing
+    if [ ! -d "${custom_src_dir}" ]; then
+        mkdir -p "${custom_install_dir}" || return "$?"
+        touch "${pending_flag_file}"
+
+        echo "installing ${custom_cmssw_version} in ${custom_install_dir}"
+        (
+            cd "${custom_install_dir}"
+            [ ! -z "${custom_scram_arch}" ] && export SCRAM_ARCH="${custom_scram_arch}"
+            scramv1 project CMSSW "${custom_cmssw_version}" && \
+                cd "${custom_cmssw_version}/src" && \
+                eval "$( scramv1 runtime -sh )" && \
+                scram b -j "${custom_install_cores}"
+        )
+
+        local setup_ret="$?"
+        if [ "${setup_ret}" != "0" ]; then
+            >&2 echo "cmssw setup failed"
+            return "7"
+        fi
+
+        # call the custom setup on top
+        if [ -f "${custom_setup_script}" ]; then
+            echo "running custom setup script ${custom_setup_script}"
+            (
+                cd "${custom_src_dir}"
+                eval "$( scramv1 runtime -sh )" && \
+                    bash "${custom_setup_script}" && \
+                    scram b -j "${custom_install_cores}"
+            )
+        fi
+
+        local custom_setup_ret="$?"
+        if [ "${custom_setup_ret}" != "0" ]; then
+            >&2 echo "custom cmssw setup script failed"
+            return "8"
+        fi
+
+        rm -f "${pending_flag_file}"
+    fi
+
+    # source it
+    cd "${custom_src_dir}"
+    eval "$( scramv1 runtime -sh )"
+    local cmsenv_ret="$?"
+    cd "${orig_dir}"
+
+    if [ "${cmsenv_ret}" != "0" ]; then
+        >&2 echo "cmsenv failed in ${custom_src_dir}"
+        return "9"
+    fi
+
+    # setup done
+    return "0"
+}
+setup_cmssw "$@"
diff --git a/law/contrib/cms/util.py b/law/contrib/cms/util.py
index a0858d6e..3fb56b97 100644
--- a/law/contrib/cms/util.py
+++ b/law/contrib/cms/util.py
@@ -4,11 +4,15 @@
 CMS-related utilities.
 """
 
-__all__ = ["Site", "lfn_to_pfn"]
+__all__ = ["Site", "lfn_to_pfn", "renew_vomsproxy", "delegate_myproxy"]
 
 
 import os
 
+import law
+
+law.contrib.load("wlcg")
+
 
 class Site(object):
     """
@@ -18,7 +22,7 @@ class Site(object):
 
     .. code-block:: python
 
-        site = Site() # executed on T2_DE_RWTH
+        site = Site()  # executed on T2_DE_RWTH
         print(site.name)        # "T2_DE_RWTH"
         print(site.country)     # "DE"
         print(site.redirector)  # "xrootd-cms.infn.it"
@@ -112,3 +116,26 @@ def lfn_to_pfn(lfn, redirector="global"):
         raise ValueError("unknown redirector: {}".format(redirector))
 
     return "root://{}/{}".format(Site.redirectors[redirector], lfn)
+
+
+def renew_vomsproxy(**kwargs):
+    """
+    Renews a VOMS proxy in the exact same way that :py:func:`law.wlcg.renew_vomsproxy` does, but
+    with the *vo* attribute set to ``"cms"`` by default.
+    """
+    kwargs.setdefault("vo", "cms")
+    return law.wlcg.renew_vomsproxy(**kwargs)
+
+
+def delegate_myproxy(**kwargs):
+    """
+    Delegates a X509 proxy to a myproxy server in the exact same way that
+    :py:func:`law.wlcg.delegate_myproxy` does, but with the *retrievers* argument set to a value
+    that is usually expected for crab submissions and the vo set to "cms".
+    """
+    kwargs.setdefault(
+        "retrievers",
+        "/DC=ch/DC=cern/OU=computers/CN=crab-(preprod|prod|dev)-tw(01|02|03).cern.ch|/DC=ch/DC=cern/OU=computers/CN=stefanov(m|m2).cern.ch|/DC=ch/DC=cern/OU=computers/CN=dciangot-tw.cern.ch",  # noqa
+    )
+    kwargs.setdefault("vo", "cms")
+    return law.wlcg.delegate_myproxy(**kwargs)
diff --git a/law/contrib/cms/workflow.py b/law/contrib/cms/workflow.py
new file mode 100644
index 00000000..ff210b97
--- /dev/null
+++ b/law/contrib/cms/workflow.py
@@ -0,0 +1,352 @@
+# coding: utf-8
+
+"""
+CMS CRAB remote workflow implementation. See
+https://twiki.cern.ch/twiki/bin/view/CMSPublic/SWGuideCrab.
+"""
+
+__all__ = ["CrabWorkflow"]
+
+
+import uuid
+from abc import abstractmethod
+from collections import OrderedDict
+
+import six
+
+import law
+from law.config import Config
+from law.workflow.remote import BaseRemoteWorkflow, BaseRemoteWorkflowProxy
+from law.job.base import JobArguments, JobInputFile
+from law.target.file import get_path
+from law.target.local import LocalDirectoryTarget
+from law.task.proxy import ProxyCommand
+from law.util import no_value, law_src_path, merge_dicts, DotDict, human_duration
+from law.logger import get_logger
+
+from law.contrib.cms.job import CrabJobManager, CrabJobFileFactory
+from law.contrib.cms.util import renew_vomsproxy, delegate_myproxy
+
+
+logger = get_logger(__name__)
+
+
+class CrabWorkflowProxy(BaseRemoteWorkflowProxy):
+
+    workflow_type = "crab"
+
+    # job script error codes are not transferred, so disable them
+    job_error_messages = {}
+
+    def create_job_manager(self, **kwargs):
+        return self.task.crab_create_job_manager(**kwargs)
+
+    def setup_job_manager(self):
+        cfg = Config.instance()
+        password_file = cfg.get_expanded("job", "crab_password_file")
+
+        # ensure a VOMS proxy exists
+        if not law.wlcg.check_vomsproxy_validity():
+            print("renew voms-proxy")
+            renew_vomsproxy(password_file=password_file)
+
+        # ensure that it has been delegated to the myproxy server
+        info = law.wlcg.get_myproxy_info(silent=True)
+        delegate = False
+        if not info:
+            delegate = True
+        elif "username" not in info:
+            logger.warning("field 'username' not in myproxy info")
+            delegate = True
+        elif "timeleft" not in info:
+            logger.warning("field 'timeleft' not in myproxy info")
+            delegate = True
+        elif info["timeleft"] < 86400:
+            logger.warning("myproxy lifetime below 24h ({})".format(
+                human_duration(seconds=info["timeleft"]),
+            ))
+            delegate = True
+
+        # actual delegation
+        if delegate:
+            print("delegate to myproxy server")
+            myproxy_username = delegate_myproxy(password_file=password_file)
+        else:
+            myproxy_username = info["username"]
+
+        return {"myproxy_username": myproxy_username}
+
+    def create_job_file_factory(self, **kwargs):
+        return self.task.crab_create_job_file_factory(**kwargs)
+
+    def create_job_file(self, submit_jobs):
+        task = self.task
+
+        # create the config
+        c = self.job_file_factory.get_config()
+        c.input_files = {}
+        c.output_files = []
+        c.render_variables = {}
+        c.custom_content = []
+
+        # get remote job file, force remote rendering
+        law_job_file = JobInputFile(task.crab_job_file())
+        law_job_file = JobInputFile(law_job_file.path, copy=False, render_job=True)
+        c.executable = law_job_file
+        c.input_files["job_file"] = c.executable
+
+        # collect task parameters
+        exclude_args = (
+            task.exclude_params_branch |
+            task.exclude_params_workflow |
+            task.exclude_params_remote_workflow |
+            task.exclude_params_crab_workflow |
+            {"workflow"}
+        )
+        proxy_cmd = ProxyCommand(
+            task.as_branch(),
+            exclude_task_args=exclude_args,
+            exclude_global_args=["workers"],
+        )
+        proxy_cmd.add_arg("--local-scheduler", "True", overwrite=True)
+        for key, value in OrderedDict(task.crab_cmdline_args()).items():
+            proxy_cmd.add_arg(key, value, overwrite=True)
+
+        # job script arguments per job number
+        c.arguments = []
+        for job_num, branches in submit_jobs.items():
+            job_args = JobArguments(
+                task_cls=task.__class__,
+                task_params=proxy_cmd.build(skip_run=True),
+                branches=branches,
+                workers=task.job_workers,
+                auto_retry=False,
+                dashboard_data=self.dashboard.remote_hook_data(
+                    job_num, self.job_data.attempts.get(job_num, 0)),
+            )
+            c.arguments.append(job_args.join())
+
+        # add the work area
+        c.work_area = get_path(task.crab_work_area())
+
+        # add the request name
+        c.request_name = task.crab_request_name(submit_jobs)
+
+        # add the storage site and output base
+        stageout_location = task.crab_stageout_location()
+        if not isinstance(stageout_location, (list, tuple)) or len(stageout_location) != 2:
+            raise ValueError(
+                "the return value of crab_stageout_location() is expected to be a 2-tuple, got "
+                "'{}'".format(stageout_location),
+            )
+        c.storage_site, c.output_lfn_base = stageout_location
+
+        # add the bootstrap file
+        bootstrap_file = task.crab_bootstrap_file()
+        if bootstrap_file:
+            c.input_files["bootstrap_file"] = bootstrap_file
+
+        # add the stageout file
+        stageout_file = task.crab_stageout_file()
+        if stageout_file:
+            c.input_files["stageout_file"] = stageout_file
+
+        # does the dashboard have a hook file?
+        dashboard_file = self.dashboard.remote_hook_file()
+        if dashboard_file:
+            c.input_files["dashboard_file"] = dashboard_file
+
+        # log file
+        if task.transfer_logs:
+            log_file = "stdall.txt"
+            c.custom_log_file = log_file
+
+        # task hook
+        c = task.crab_job_config(c, submit_jobs)
+
+        # build the job file and get the sanitized config
+        job_file, c = self.job_file_factory(**c.__dict__)
+
+        # return job and log file entry
+        # (the latter is None but will be synced from query data)
+        return {"job": job_file, "log": None}
+
+    def _status_error_pairs(self, job_num, job_data):
+        pairs = super(CrabWorkflowProxy, self)._status_error_pairs(job_num, job_data)
+
+        # add site history
+        pairs.insert_before("log", "site history", job_data["extra"].get("site_history", no_value))
+
+        return pairs
+
+    def destination_info(self):
+        info = super(CrabWorkflowProxy, self).destination_info()
+
+        info = self.task.crab_destination_info(info)
+
+        return info
+
+
+class CrabWorkflow(BaseRemoteWorkflow):
+
+    workflow_proxy_cls = CrabWorkflowProxy
+
+    crab_workflow_run_decorators = None
+    crab_job_manager_defaults = None
+    crab_job_file_factory_defaults = None
+
+    crab_job_kwargs = []
+    crab_job_kwargs_submit = None
+    crab_job_kwargs_cancel = None
+    crab_job_kwargs_cleanup = None
+    crab_job_kwargs_query = None
+
+    exclude_params_branch = set()
+    exclude_params_crab_workflow = set()
+
+    exclude_index = True
+
+    @abstractmethod
+    def crab_stageout_location(self):
+        """
+        Hook to define both the "Site.storageSite" and "Data.outLFNDirBase" settings in a 2-tuple,
+        i.e., the name of the storage site to use and the base directory for crab's own output
+        staging. An example would be ``("T2_DE_DESY", "/store/user/...")``.
+
+        In case this is not used, the choice of the output base has no affect, but is still required
+        for crab's job submission to work.
+        """
+        return
+
+    @abstractmethod
+    def crab_output_directory(self):
+        """
+        Hook to define the location of submission output files, such as the json files containing
+        job data. This method should return a :py:class:`FileSystemDirectoryTarget`.
+        """
+        return
+
+    def crab_request_name(self, submit_jobs):
+        """
+        Returns a random name for a request, i.e., the project directory inside the crab job working
+        area.
+        """
+        return "{}_{}".format(self.live_task_id, str(uuid.uuid4())[:8])
+
+    def crab_work_area(self):
+        """
+        Returns the location of the crab working area, defaulting to the value of
+        :py:meth:`crab_output_directory` in case it refers to a local directory. When *None*, the
+        value of the "job.crab_work_area" configuration options is used.
+        """
+        # when job files are cleaned, try to use the output directory when local
+        if self.workflow_proxy.job_file_factory and self.workflow_proxy.job_file_factory.cleanup:
+            out_dir = self.crab_output_directory()
+            if isinstance(out_dir, six.string_types) or isinstance(out_dir, LocalDirectoryTarget):
+                return out_dir
+
+        # relative to the job file directory
+        return ""
+
+    def crab_job_file(self):
+        """
+        Hook to return the location of the job file that is executed on job nodes.
+        """
+        return JobInputFile(law_src_path("job", "law_job.sh"))
+
+    def crab_bootstrap_file(self):
+        """
+        Hook to define the location of an optional, so-called bootstrap file that is sent alongside
+        jobs and called prior to the actual job payload. It is meant to run a custom setup routine
+        in order for the payload to run successfully (e.g. software setup, data retrieval).
+        """
+        return None
+
+    def crab_stageout_file(self):
+        """
+        Hook to define the location of an optional, so-called stageout file that is sent alongside
+        jobs and called after to the actual job payload. It is meant to run a custom output stageout
+        routine if required so by your workflow or target storage element.
+        """
+        return None
+
+    def crab_workflow_requires(self):
+        """
+        Hook to define requirements for the workflow itself and that need to be resolved before any
+        submission can happen.
+        """
+        return DotDict()
+
+    def crab_output_postfix(self):
+        """
+        Hook to define the postfix of outputs, for instance such that workflows with different
+        parameters do not write their intermediate job status information into the same json file.
+        """
+        return ""
+
+    def crab_output_uri(self):
+        """
+        Hook to return the URI of the remote crab output directory.
+        """
+        return self.crab_output_directory().uri()
+
+    def crab_job_manager_cls(self):
+        """
+        Hook to define a custom job managet class to use.
+        """
+        return CrabJobManager
+
+    def crab_create_job_manager(self, **kwargs):
+        """
+        Hook to configure how the underlying job manager is instantiated and configured.
+        """
+        kwargs = merge_dicts(self.crab_job_manager_defaults, kwargs)
+        return self.crab_job_manager_cls()(**kwargs)
+
+    def crab_job_file_factory_cls(self):
+        """
+        Hook to define a custom job file factory class to use.
+        """
+        return CrabJobFileFactory
+
+    def crab_create_job_file_factory(self, **kwargs):
+        """
+        Hook to configure how the underlying job file factory is instantiated and configured.
+        """
+        # job file fectory config priority: kwargs > class defaults
+        kwargs = merge_dicts({}, self.crab_job_file_factory_defaults, kwargs)
+        return self.crab_job_file_factory_cls()(**kwargs)
+
+    def crab_job_config(self, config, submit_jobs):
+        """
+        Hook to inject custom settings into the job *config*, which is an instance of the
+        :py:attr:`Config` class defined inside the job manager.
+        """
+        return config
+
+    def crab_check_job_completeness(self):
+        """
+        Hook to define whether
+        """
+        return False
+
+    def crab_check_job_completeness_delay(self):
+        """
+        Grace period before :py:meth:`crab_check_job_completeness` is called to ensure that output
+        files are accessible. Especially useful on distributed file systems with possibly
+        asynchronous behavior.
+        """
+        return 0.0
+
+    def crab_cmdline_args(self):
+        """
+        Hook to add additional cli parameters to "law run" commands executed on job nodes.
+        """
+        return {}
+
+    def crab_destination_info(self, info):
+        """
+        Hook to add additional information behind each job status query line by extending an *info*
+        dictionary whose values will be shown separated by comma.
+        """
+        return info
diff --git a/law/contrib/git/scripts/repository_checksum.sh b/law/contrib/git/scripts/repository_checksum.sh
index 747812be..71e8e163 100755
--- a/law/contrib/git/scripts/repository_checksum.sh
+++ b/law/contrib/git/scripts/repository_checksum.sh
@@ -35,7 +35,7 @@ action() {
         cd "${repo_path}" && \
         git rev-parse HEAD && \
         git diff && \
-        git ls-files --others --exclude-standard | xargs cat; \
+        git ls-files --others --exclude-standard | xargs cat 2>&1; \
         [ "${recursive}" = "1" ] && git submodule foreach --recursive --quiet "\
             git rev-parse HEAD && \
             git diff && \
diff --git a/law/contrib/glite/job.py b/law/contrib/glite/job.py
index 12b85ff5..14def639 100644
--- a/law/contrib/glite/job.py
+++ b/law/contrib/glite/job.py
@@ -9,7 +9,6 @@
 
 import os
 import stat
-import sys
 import time
 import re
 import random
@@ -80,7 +79,7 @@ def submit(self, job_file, ce=None, delegation_id=None, retries=0, retry_delay=3
             # glite prints everything to stdout
             logger.debug("submit glite job with command '{}'".format(cmd))
             code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-                stdout=subprocess.PIPE, stderr=sys.stderr, cwd=job_file_dir)
+                stdout=subprocess.PIPE, cwd=job_file_dir)
 
             # in some cases, the return code is 0 but the ce did not respond with a valid id
             if code == 0:
@@ -117,7 +116,7 @@ def cancel(self, job_id, silent=False):
         # run it
         logger.debug("cancel glite job(s) with command '{}'".format(cmd))
         code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-            stdout=subprocess.PIPE, stderr=sys.stderr)
+            stdout=subprocess.PIPE)
 
         # check success
         if code != 0 and not silent:
@@ -138,7 +137,7 @@ def cleanup(self, job_id, silent=False):
         # run it
         logger.debug("cleanup glite job(s) with command '{}'".format(cmd))
         code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-            stdout=subprocess.PIPE, stderr=sys.stderr)
+            stdout=subprocess.PIPE)
 
         # check success
         if code != 0 and not silent:
@@ -159,7 +158,7 @@ def query(self, job_id, silent=False):
         # run it
         logger.debug("query glite job(s) with command '{}'".format(cmd))
         code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-            stdout=subprocess.PIPE, stderr=sys.stderr)
+            stdout=subprocess.PIPE)
 
         # handle errors
         if code != 0:
diff --git a/law/contrib/glite/workflow.py b/law/contrib/glite/workflow.py
index 2b2f264c..cf0e7149 100644
--- a/law/contrib/glite/workflow.py
+++ b/law/contrib/glite/workflow.py
@@ -13,13 +13,13 @@
 from abc import abstractmethod
 from collections import OrderedDict
 
+import law
 from law.workflow.remote import BaseRemoteWorkflow, BaseRemoteWorkflowProxy
 from law.job.base import JobArguments, JobInputFile, DeprecatedInputFiles
 from law.task.proxy import ProxyCommand
 from law.target.file import get_path
 from law.parameter import CSVParameter
 from law.util import law_src_path, merge_dicts, DotDict
-from law.contrib.wlcg import delegate_voms_proxy_glite, get_ce_endpoint
 from law.logger import get_logger
 
 from law.contrib.glite.job import GLiteJobManager, GLiteJobFileFactory
@@ -44,6 +44,19 @@ def __init__(self, *args, **kwargs):
     def create_job_manager(self, **kwargs):
         return self.task.glite_create_job_manager(**kwargs)
 
+    def setup_job_mananger(self):
+        kwargs = {}
+
+        # delegate the voms proxy to all endpoints
+        if callable(self.task.glite_delegate_proxy):
+            delegation_ids = []
+            for ce in self.task.glite_ce:
+                endpoint = law.wlcg.get_ce_endpoint(ce)
+                delegation_ids.append(self.task.glite_delegate_proxy(endpoint))
+            kwargs["delegation_id"] = delegation_ids
+
+        return kwargs
+
     def create_job_file_factory(self, **kwargs):
         return self.task.glite_create_job_file_factory(**kwargs)
 
@@ -141,23 +154,14 @@ def create_job_file(self, job_num, branches):
         # return job and log files
         return {"job": job_file, "log": abs_log_file}
 
-    def _submit(self, *args, **kwargs):
-        task = self.task
-
-        # delegate the voms proxy to all endpoints
-        if self.delegation_ids is None and callable(task.glite_delegate_proxy):
-            self.delegation_ids = []
-            for ce in task.glite_ce:
-                endpoint = get_ce_endpoint(ce)
-                self.delegation_ids.append(task.glite_delegate_proxy(endpoint))
-        kwargs["delegation_id"] = self.delegation_ids
+    def destination_info(self):
+        info = super(GLiteWorkflowProxy, self).destination_info()
 
-        return super(GLiteWorkflowProxy, self)._submit(*args, **kwargs)
+        info["ce"] = "ce: {}".format(",".join(self.task.glite_ce))
 
-    def destination_info(self):
-        info = ["ce: {}".format(",".join(self.task.glite_ce))]
         info = self.task.glite_destination_info(info)
-        return ", ".join(map(str, info))
+
+        return info
 
 
 class GLiteWorkflow(BaseRemoteWorkflow):
@@ -213,7 +217,7 @@ def glite_output_uri(self):
         return self.glite_output_directory().uri()
 
     def glite_delegate_proxy(self, endpoint):
-        return delegate_voms_proxy_glite(endpoint, stdout=sys.stdout, stderr=sys.stderr,
+        return law.wlcg.delegate_vomsproxy_glite(endpoint, stdout=sys.stdout, stderr=sys.stderr,
             cache=True)
 
     def glite_job_manager_cls(self):
diff --git a/law/contrib/htcondor/workflow.py b/law/contrib/htcondor/workflow.py
index cf38c38c..4b63e181 100644
--- a/law/contrib/htcondor/workflow.py
+++ b/law/contrib/htcondor/workflow.py
@@ -145,13 +145,17 @@ def create_job_file(self, job_num, branches):
         return {"job": job_file, "log": abs_log_file}
 
     def destination_info(self):
-        info = []
+        info = super(HTCondorWorkflowProxy, self).destination_info()
+
         if self.task.htcondor_pool and self.task.htcondor_pool != NO_STR:
-            info.append("pool: {}".format(self.task.htcondor_pool))
+            info["pool"] = "pool: {}".format(self.task.htcondor_pool)
+
         if self.task.htcondor_scheduler and self.task.htcondor_scheduler != NO_STR:
-            info.append("scheduler: {}".format(self.task.htcondor_scheduler))
+            info["scheduler"] = "scheduler: {}".format(self.task.htcondor_scheduler)
+
         info = self.task.htcondor_destination_info(info)
-        return ", ".join(map(str, info))
+
+        return info
 
 
 class HTCondorWorkflow(BaseRemoteWorkflow):
diff --git a/law/contrib/lsf/workflow.py b/law/contrib/lsf/workflow.py
index 4ddcf1af..a25a0218 100644
--- a/law/contrib/lsf/workflow.py
+++ b/law/contrib/lsf/workflow.py
@@ -147,11 +147,14 @@ def create_job_file(self, job_num, branches):
         return {"job": job_file, "log": abs_log_file}
 
     def destination_info(self):
-        info = []
+        info = super(LSFWorkflowProxy, self).destination_info()
+
         if self.task.lsf_queue != NO_STR:
-            info.append("queue: {}".format(self.task.lsf_queue))
+            info["queue"] = "queue: {}".format(self.task.lsf_queue)
+
         info = self.task.lsf_destination_info(info)
-        return ", ".join(map(str, info))
+
+        return info
 
 
 class LSFWorkflow(BaseRemoteWorkflow):
diff --git a/law/contrib/slurm/workflow.py b/law/contrib/slurm/workflow.py
index 27643c0e..dccc8fb3 100644
--- a/law/contrib/slurm/workflow.py
+++ b/law/contrib/slurm/workflow.py
@@ -146,9 +146,11 @@ def create_job_file(self, job_num, branches):
         return {"job": job_file, "log": abs_log_file}
 
     def destination_info(self):
-        info = []
-        info.extend(self.task.slurm_destination_info(info))
-        return ", ".join(map(str, info))
+        info = super(SlurmWorkflowProxy, self).destination_info()
+
+        info = self.task.slurm_destination_info(info)
+
+        return info
 
 
 class SlurmWorkflow(BaseRemoteWorkflow):
diff --git a/law/contrib/wlcg/__init__.py b/law/contrib/wlcg/__init__.py
index 6e831d42..0622c1fa 100644
--- a/law/contrib/wlcg/__init__.py
+++ b/law/contrib/wlcg/__init__.py
@@ -6,17 +6,22 @@
 """
 
 __all__ = [
-    "get_voms_proxy_file", "get_voms_proxy_user", "get_voms_proxy_lifetime", "get_voms_proxy_vo",
-    "check_voms_proxy_validity", "renew_voms_proxy", "delegate_voms_proxy_glite", "get_ce_endpoint",
+    "get_userkey", "get_usercert", "get_usercert_subject", "get_vomsproxy_file",
+    "get_vomsproxy_identity", "get_vomsproxy_lifetime", "get_vomsproxy_vo",
+    "check_vomsproxy_validity", "renew_vomsproxy", "delegate_vomsproxy_glite",
+    "delegate_myproxy", "get_myproxy_info", "get_ce_endpoint",
     "WLCGFileSystem", "WLCGTarget", "WLCGFileTarget", "WLCGDirectoryTarget",
-    "ensure_voms_proxy",
+    "ensure_vomsproxy",
 ]
 
 
 # provisioning imports
 from law.contrib.wlcg.util import (
-    get_voms_proxy_file, get_voms_proxy_user, get_voms_proxy_lifetime, get_voms_proxy_vo,
-    check_voms_proxy_validity, renew_voms_proxy, delegate_voms_proxy_glite, get_ce_endpoint,
+    get_userkey, get_usercert, get_usercert_subject,
+    get_vomsproxy_file, get_vomsproxy_identity, get_vomsproxy_lifetime, get_vomsproxy_vo,
+    check_vomsproxy_validity, renew_vomsproxy, delegate_vomsproxy_glite,
+    delegate_myproxy, get_myproxy_info,
+    get_ce_endpoint,
 )
 from law.contrib.wlcg.target import WLCGFileSystem, WLCGTarget, WLCGFileTarget, WLCGDirectoryTarget
-from law.contrib.wlcg.decorator import ensure_voms_proxy
+from law.contrib.wlcg.decorator import ensure_vomsproxy
diff --git a/law/contrib/wlcg/decorator.py b/law/contrib/wlcg/decorator.py
index 2b8b18ba..554d61c7 100644
--- a/law/contrib/wlcg/decorator.py
+++ b/law/contrib/wlcg/decorator.py
@@ -4,23 +4,23 @@
 Decorators for task methods for convenient working with the WLCG.
 """
 
-__all__ = ["ensure_voms_proxy"]
+__all__ = ["ensure_vomsproxy"]
 
 
 from law.decorator import factory
-from law.contrib.wlcg import check_voms_proxy_validity
+from law.contrib.wlcg import check_vomsproxy_validity
 
 
 @factory(accept_generator=True)
-def ensure_voms_proxy(fn, opts, task, *args, **kwargs):
-    """ ensure_voms_proxy()
+def ensure_vomsproxy(fn, opts, task, *args, **kwargs):
+    """ ensure_vomsproxy()
     Decorator for law task methods that checks the validity of the voms proxy and throws an
     exception in case it is invalid. This can prevent late errors on remote worker notes that except
     voms proxies to be present. Accepts generator functions.
     """
     def before_call():
         # check the proxy validity
-        if not check_voms_proxy_validity():
+        if not check_vomsproxy_validity():
             raise Exception("voms proxy not valid")
 
         return None
diff --git a/law/contrib/wlcg/util.py b/law/contrib/wlcg/util.py
index 2de7fdaf..7aa3e770 100644
--- a/law/contrib/wlcg/util.py
+++ b/law/contrib/wlcg/util.py
@@ -5,8 +5,11 @@
 """
 
 __all__ = [
-    "get_voms_proxy_file", "get_voms_proxy_user", "get_voms_proxy_lifetime", "get_voms_proxy_vo",
-    "check_voms_proxy_validity", "renew_voms_proxy", "delegate_voms_proxy_glite", "get_ce_endpoint",
+    "get_userkey", "get_usercert", "get_usercert_subject",
+    "get_vomsproxy_file", "get_vomsproxy_identity", "get_vomsproxy_lifetime", "get_vomsproxy_vo",
+    "check_vomsproxy_validity", "renew_vomsproxy", "delegate_vomsproxy_glite",
+    "delegate_myproxy", "get_myproxy_info",
+    "get_ce_endpoint",
 ]
 
 
@@ -15,11 +18,14 @@
 import subprocess
 import uuid
 import json
+import hashlib
+import getpass
+import functools
 
 import six
 
 from law.util import (
-    interruptable_popen, tmp_file, create_hash, human_duration, parse_duration, quote_cmd,
+    interruptable_popen, create_hash, human_duration, parse_duration, quote_cmd,
 )
 from law.logger import get_logger
 
@@ -27,7 +33,46 @@
 logger = get_logger(__name__)
 
 
-def get_voms_proxy_file():
+def get_userkey():
+    """
+    Returns the expanded path to the globus user key, reading from "$X509_USER_KEY" and defaulting
+    to "$HOME/.globus/userkey.pem".
+    """
+    path = os.getenv("X509_USER_KEY", "$HOME/.globus/userkey.pem")
+    return os.path.expandvars(os.path.expanduser(path))
+
+
+def get_usercert():
+    """
+    Returns the expanded path to the globus user certificate, reading from "$X509_USER_CERT" and
+    defaulting to "$HOME/.globus/usercert.pem".
+    """
+    path = os.getenv("X509_USER_CERT", "$HOME/.globus/usercert.pem")
+    return os.path.expandvars(os.path.expanduser(path))
+
+
+def get_usercert_subject(usercert=None):
+    """
+    Returns the user "subject" string of the certificate at *usercert*, which defaults to the
+    return value of :py:func:`get_usercert`.
+    """
+    # get the user certificate file
+    if usercert is None:
+        usercert = get_usercert()
+    if not os.path.exists(usercert):
+        raise Exception("usercert does not exist at '{}'".format(usercert))
+
+    # extract the subject via openssl
+    cmd = ["openssl", "x509", "-in", usercert, "-noout", "-subject"]
+    code, out, err = interruptable_popen(quote_cmd(cmd), shell=True, executable="/bin/bash",
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if code != 0:
+        raise Exception("subject extraction from usercert failed: {}".format(err))
+
+    return re.sub(r"^subject\s*=\s*", "", out.strip())
+
+
+def get_vomsproxy_file():
     """
     Returns the path to the voms proxy file.
     """
@@ -37,13 +82,13 @@ def get_voms_proxy_file():
     return "/tmp/x509up_u{}".format(os.getuid())
 
 
-def _voms_proxy_info(args=None, proxy_file=None, silent=False):
+def _vomsproxy_info(args=None, proxy_file=None, silent=False):
     cmd = ["voms-proxy-info"] + (args or [])
 
     # when proxy_file is None, get the default
     # when empty string, don't add a --file argument
     if proxy_file is None:
-        proxy_file = get_voms_proxy_file()
+        proxy_file = get_vomsproxy_file()
     if proxy_file:
         proxy_file = os.path.expandvars(os.path.expanduser(proxy_file))
         cmd.extend(["--file", proxy_file])
@@ -57,65 +102,63 @@ def _voms_proxy_info(args=None, proxy_file=None, silent=False):
     return code, out, err
 
 
-def get_voms_proxy_user(proxy_file=None):
+def get_vomsproxy_identity(proxy_file=None, silent=False):
     """
-    Returns the owner of the voms proxy. When *proxy_file* is *None*, it defaults to the result of
-    :py:func:`get_voms_proxy_file`. Otherwise, when it evaluates to *False*, ``voms-proxy-info`` is
-    queried without a custom proxy file.
+    Returns the identity information of the voms proxy. When *proxy_file* is *None*, it defaults to
+    the result of :py:func:`get_vomsproxy_file`. Otherwise, when it evaluates to *False*,
+    ``voms-proxy-info`` is queried without a custom proxy file.
     """
-    out = _voms_proxy_info(args=["--identity"], proxy_file=proxy_file)[1].strip()
-
-    cns = re.findall(r"/CN=[a-z]+", out.replace(" ", ""))
-    if not cns:
-        raise Exception("no valid identity found in voms proxy: {}".format(out))
-
-    # extract actual names
-    names = [cn[4:] for cn in cns]
+    code, out, _ = _vomsproxy_info(args=["--identity"], proxy_file=proxy_file, silent=silent)
+    return out.strip() if code == 0 else None
 
-    # return the shortest name
-    return sorted(names, key=len)[0]
 
-
-def get_voms_proxy_lifetime(proxy_file=None):
+def get_vomsproxy_lifetime(proxy_file=None, silent=False):
     """
     Returns the remaining lifetime of the voms proxy in seconds. When *proxy_file* is *None*, it
-    defaults to the result of :py:func:`get_voms_proxy_file`. Otherwise, when it evaluates to
+    defaults to the result of :py:func:`get_vomsproxy_file`. Otherwise, when it evaluates to
     *False*, ``voms-proxy-info`` is queried without a custom proxy file.
     """
-    out = _voms_proxy_info(args=["--timeleft"], proxy_file=proxy_file)[1].strip()
+    code, out, _ = _vomsproxy_info(args=["--timeleft"], proxy_file=proxy_file, silent=silent)
+
+    if code != 0:
+        return None
+
     try:
         return int(out)
     except:
+        if silent:
+            return None
         raise Exception("no valid lifetime found in voms proxy: {}".format(out))
 
 
-def get_voms_proxy_vo(proxy_file=None):
+def get_vomsproxy_vo(proxy_file=None, silent=False):
     """
     Returns the virtual organization name of the voms proxy. When *proxy_file* is *None*, it
-    defaults to the result of :py:func:`get_voms_proxy_file`. Otherwise, when it evaluates to
+    defaults to the result of :py:func:`get_vomsproxy_file`. Otherwise, when it evaluates to
     *False*, ``voms-proxy-info`` is queried without a custom proxy file.
     """
-    return _voms_proxy_info(args=["--vo"], proxy_file=proxy_file)[1].strip()
+    code, out, _ = _vomsproxy_info(args=["--vo"], proxy_file=proxy_file, silent=silent)
+    return out.strip() if code == 0 else None
 
 
-def check_voms_proxy_validity(return_rfc=False, proxy_file=None):
+def check_vomsproxy_validity(return_rfc=False, proxy_file=None):
     """
     Returns *True* if a valid voms proxy exists (positive lifetime), and *False* otherwise. When
     *return_rfc* is *True*, The return value will be a 2-tuple, containing also whether the proxy
     is RFC3820 compliant.
 
-     When *proxy_file* is *None*, it defaults to the result of :py:func:`get_voms_proxy_file`.
-     Otherwise, when it evaluates to *False*, ``voms-proxy-info`` is ueried without a custom proxy
-     file.
+    When *proxy_file* is *None*, it defaults to the result of :py:func:`get_vomsproxy_file`.
+    Otherwise, when it evaluates to *False*, ``voms-proxy-info`` is ueried without a custom proxy
+    file.
     """
-    code, _, err = _voms_proxy_info(args=["--exists"], proxy_file=proxy_file, silent=True)
+    code, _, err = _vomsproxy_info(args=["--exists"], proxy_file=proxy_file, silent=True)
 
     rfc = False
     if code == 0:
-        valid = get_voms_proxy_lifetime(proxy_file=proxy_file) > 0
+        valid = get_vomsproxy_lifetime(proxy_file=proxy_file) > 0
 
         if valid and return_rfc:
-            out = _voms_proxy_info(args=["--type"], proxy_file=proxy_file, silent=True)[1]
+            out = _vomsproxy_info(args=["--type"], proxy_file=proxy_file, silent=True)[1]
             rfc = out.strip().lower().startswith("rfc3820 compliant")
 
     elif err.strip().lower().startswith("proxy not found"):
@@ -127,14 +170,26 @@ def check_voms_proxy_validity(return_rfc=False, proxy_file=None):
     return (valid, rfc) if return_rfc else valid
 
 
-def renew_voms_proxy(password="", vo=None, lifetime="8 days", proxy_file=None):
+def renew_vomsproxy(
+    proxy_file=None,
+    vo=None,
+    rfc=True,
+    lifetime="8 days",
+    password=None,
+    password_file=None,
+    silent=False,
+):
     """
-    Renews the voms proxy using a password *password*, an optional virtual organization name *vo*,
-    and a default *lifetime* of 8 days, which is internally parsed by
-    :py:func:`law.util.parse_duration` where the default input unit is hours. To ensure that the
-    *password* is not visible in any process listing, it is written to a temporary file first and
-    piped into the ``voms-proxy-init`` command. When *proxy_file* is *None*, it defaults to the
-    result of :py:func:`get_voms_proxy_file`.
+    Renews a voms proxy at *proxy_file* using an optional virtual organization name *vo*, and a
+    default *lifetime* of 8 days, which is internally parsed by :py:func:`law.util.parse_duration`
+    where the default input unit is hours. When *proxy_file* is *None*, it defaults to the result of
+    :py:func:`get_vomsproxy_file`.
+
+    When *rfc* is *True*, the created proxy will be RFC compliant.
+
+    By default, this function will prompt for input to securely receive the password. However, if
+    *password* or *password_file* is given, the password is extracted from that variable or file
+    instead.
     """
     # parse and format the lifetime
     lifetime_seconds = max(parse_duration(lifetime, input_unit="h", unit="s"), 60.0)
@@ -144,33 +199,48 @@ def renew_voms_proxy(password="", vo=None, lifetime="8 days", proxy_file=None):
     lifetime = ":".join(normalized.rsplit(":", 3)[-3:-1])
 
     # when proxy_file is None, get the default
-    # when empty string, don't add a --out argument
     if proxy_file is None:
-        proxy_file = get_voms_proxy_file()
-
-    with tmp_file() as (_, tmp):
-        with open(tmp, "w") as f:
-            f.write(password)
+        proxy_file = get_vomsproxy_file()
+
+    # build the command
+    cmd = [
+        "voms-proxy-init",
+        "--valid", lifetime,
+        "--out", proxy_file,
+    ]
+    if vo:
+        cmd.extend(["-voms", vo])
+    if rfc:
+        cmd.append("--rfc")
+
+    # run it, depending on whether a password file is given
+    silent_pipe = subprocess.PIPE if silent else None
+    if password_file:
+        password_file = os.path.expandvars(os.path.expanduser(password_file))
+        cmd = "cat \"{}\" | {}".format(password_file, quote_cmd(cmd))
+        code = interruptable_popen(cmd, shell=True, executable="/bin/bash", stdout=silent_pipe,
+            stderr=silent_pipe)[0]
+    else:
+        cmd = quote_cmd(cmd)
+        stdin_callback = (lambda: password) if password else functools.partial(getpass.getpass, "")
+        code = interruptable_popen(cmd, shell=True, executable="/bin/bash", stdout=silent_pipe,
+            stderr=silent_pipe, stdin=subprocess.PIPE, stdin_callback=stdin_callback,
+            stdin_delay=0.2)[0]
 
-        cmd = "cat '{}' | voms-proxy-init --valid '{}'".format(tmp, lifetime)
-        if vo:
-            cmd += " -voms '{}'".format(vo)
-        if proxy_file:
-            proxy_file = os.path.expandvars(os.path.expanduser(proxy_file))
-            cmd += " --out '{}'".format(proxy_file)
+    if code == 0:
+        return proxy_file
 
-        code, out, _ = interruptable_popen(cmd, shell=True, executable="/bin/bash",
-            stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    if silent:
+        return None
 
-        if code != 0:
-            raise Exception("voms-proxy-init failed: {}".format(out))
+    raise Exception("voms-proxy-init failed with code {}".format(code))
 
 
-def delegate_voms_proxy_glite(endpoint, proxy_file=None, stdout=None, stderr=None, cache=True):
+def delegate_vomsproxy_glite(endpoint, proxy_file=None, stdout=None, stderr=None, cache=True):
     """
     Delegates the voms proxy via gLite to an *endpoint*, e.g.
     ``grid-ce.physik.rwth-aachen.de:8443``. When *proxy_file* is *None*, it defaults to the result
-    of :py:func:`get_voms_proxy_file`. *stdout* and *stderr* are passed to the *Popen* constructor
+    of :py:func:`get_vomsproxy_file`. *stdout* and *stderr* are passed to the *Popen* constructor
     for executing the ``glite-ce-delegate-proxy`` command. When *cache* is *True*, a json file is
     created alongside the proxy file, which stores the delegation ids per endpoint. The next time
     the exact same proxy should be delegated to the same endpoint, the cached delegation id is
@@ -178,7 +248,7 @@ def delegate_voms_proxy_glite(endpoint, proxy_file=None, stdout=None, stderr=Non
     """
     # get the proxy file
     if not proxy_file:
-        proxy_file = get_voms_proxy_file()
+        proxy_file = get_vomsproxy_file()
     proxy_file = os.path.expandvars(os.path.expanduser(proxy_file))
     if not os.path.exists(proxy_file):
         raise Exception("proxy file '{}' does not exist".format(proxy_file))
@@ -236,6 +306,141 @@ def remove_cache():
     return delegation_id
 
 
+def delegate_myproxy(
+    endpoint="myproxy.cern.ch",
+    userkey=None,
+    usercert=None,
+    username=None,
+    encode_username=True,
+    cred_lifetime=720,
+    proxy_lifetime=168,
+    retrievers=None,
+    rfc=True,
+    vo=None,
+    create_local=False,
+    password=None,
+    password_file=None,
+    silent=False,
+):
+    """
+    Delegates an X509 proxy to a myproxy server *endpoint*.
+
+    *userkey* and *usercert* default to the return values of :py:func:`get_userkey` and
+    :py:func:`get_usercert`, respectively. When *username* is *None*, the identity field of an
+    existing voms proxy is used, or the subject string of the certificate otherwise. If
+    *encode_username* is set, *username* is the sha1 encoded.
+
+    The credential and proxy lifetimes can be defined in hours by *cred_lifetime* and
+    *proxy_lifetime*. When *retrievers* is given, it is passed as both ``--renewable_by`` and
+    ``--retrievable_by_cert`` to the underlying ``myproxy-init`` command.
+
+    When *rfc* is *True*, the delegated proxy will be RFC compliant. To pass VOMS attributes to the
+    ``myproxy-init`` command, *vo* can be defined. When *create_local* is *True*, the delegation
+    also creates a local proxy file (usually at $X509_USER_PROXY).
+
+    If no *password* is given, the user is prompted for the password of the user certificate.
+    However, if a *password_file* is present, the password is extracted from this file.
+
+    The user name is returned upon success. Otherwise an exception is raised unless *silent* is
+    *True* in which case *None* is returned.
+    """
+    # prepare arguments
+    if not userkey:
+        userkey = get_userkey()
+    if not usercert:
+        usercert = get_usercert()
+    if not username:
+        username = get_vomsproxy_identity(silent=True) or get_usercert_subject()
+    if encode_username:
+        username = hashlib.sha1(username.encode("utf-8")).hexdigest()
+
+    # build the command
+    cmd = [
+        "myproxy-init",
+        "-s", endpoint,
+        "-y", userkey,
+        "-C", usercert,
+        "-l", username,
+        "-t", str(proxy_lifetime),
+        "-c", str(cred_lifetime),
+    ]
+    if retrievers:
+        cmd.extend([
+            "-x", "-R", retrievers,
+            "-x", "-Z", retrievers,
+        ])
+    if vo:
+        cmd.extend(["-m", vo])
+    if create_local:
+        cmd.append("-L")
+    rfc_export = "GT_PROXY_MODE=rfc " if rfc else ""
+
+    # run it, depending on whether a password file is given
+    silent_pipe = subprocess.PIPE if silent else None
+    if password_file:
+        password_file = os.path.expandvars(os.path.expanduser(password_file))
+        cmd = "{}cat \"{}\" | {} -S".format(rfc_export, password_file, quote_cmd(cmd))
+        code = interruptable_popen(cmd, shell=True, executable="/bin/bash", stdout=silent_pipe,
+            stderr=silent_pipe)[0]
+    else:
+        cmd = "{}{}".format(rfc_export, quote_cmd(cmd))
+        stdin_callback = (lambda: password) if password else functools.partial(getpass.getpass, "")
+        code = interruptable_popen(cmd, shell=True, executable="/bin/bash", stdout=silent_pipe,
+            stderr=silent_pipe, stdin=subprocess.PIPE, stdin_callback=stdin_callback,
+            stdin_delay=0.2)[0]
+
+    if code == 0:
+        return username
+
+    if silent:
+        return None
+
+    raise Exception("myproxy-init failed with code {}".format(code))
+
+
+def get_myproxy_info(endpoint="myproxy.cern.ch", username=None, encode_username=True, silent=False):
+    """
+    Returns information about a previous myproxy delegation to a server *endpoint*. When *username*
+    is *None*, the subject string of the certificate is used instead, and sha1 encoded if
+    *encode_username* is *True*.
+
+    The returned dictionary contains the fields ``username``, ``subject`` and ``timeleft``.
+
+    An exception is raised if the underlying ``myproxy-info`` command fails, unless *silent* is
+    *True* in which case *None* is returned.
+    """
+    # prepare arguments
+    if not username:
+        username = get_usercert_subject()
+    if encode_username:
+        username = hashlib.sha1(username.encode("utf-8")).hexdigest()
+
+    # build and run the command
+    cmd = ["myproxy-info", "-s", endpoint, "-l", username]
+    code, out, _ = interruptable_popen(quote_cmd(cmd), shell=True, executable="/bin/bash",
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE if silent else None)
+    if code != 0:
+        if silent:
+            return
+        raise Exception("myproxy-info failed with code {}".format(code))
+
+    # parse the output
+    info = {}
+    for line in out.strip().replace("\r\n", "\n").split("\n"):
+        line = line.strip()
+        if line.startswith("username: "):
+            info["username"] = line[len("username: "):]
+        elif line.startswith("owner: "):
+            info["subject"] = line[len("owner: "):]
+        elif line.startswith("timeleft: "):
+            m = re.match(r"^timeleft:\s+(\d+):(\d+):(\d+).*$", line)
+            if m:
+                times = list(map(int, m.groups()))
+                info["timeleft"] = times[0] * 3600 + times[1] * 60 + times[2]
+
+    return info
+
+
 def get_ce_endpoint(ce):
     """
     Extracts the endpoint from a computing element *ce* and returns it. Example:
diff --git a/law/job/base.py b/law/job/base.py
index 826758ee..85b662a4 100644
--- a/law/job/base.py
+++ b/law/job/base.py
@@ -25,7 +25,9 @@
 
 from law.config import Config
 from law.target.file import get_scheme
-from law.util import colored, make_list, iter_chunks, makedirs, create_hash, empty_context
+from law.util import (
+    colored, make_list, make_tuple, iter_chunks, makedirs, create_hash, empty_context,
+)
 from law.logger import get_logger
 
 
@@ -235,8 +237,10 @@ def query(self):
     def group_job_ids(self, job_ids):
         """
         Hook that needs to be implemented if the job mananger supports grouping of jobs, i.e., when
-        :py:attr:`job_grouping` is *True*. If so, it should take a sequence of *job_ids and return a
-        list of groups (again lists) of ids, with an arbitrary grouping mechanism.
+        :py:attr:`job_grouping` is *True*, and potentially used during status queries, job
+        cancellation and removal. If so, it should take a sequence of *job_ids* and return a
+        dictionary mapping ids of group jobs (used for queries etc) to the corresponding lists of
+        original job ids, with an arbitrary grouping mechanism.
         """
         raise NotImplementedError(
             "internal error, {}.group_job_ids not implemented".format(self.__class__.__name__),
@@ -423,7 +427,7 @@ def _apply_group(
         self,
         func,
         result_type,
-        group_ids,
+        group_func,
         job_objs,
         threads=None,
         callback=None,
@@ -433,56 +437,39 @@ def _apply_group(
         threads = max(threads or self.threads or 1, 1)
 
         # group objects
-        job_objs = make_list(job_objs)
-        if group_ids:
-            job_objs = self.group_job_ids(job_objs)
-            if isinstance(job_objs, dict):
-                job_objs = list(job_objs.values())
+        job_objs = group_func(make_list(job_objs))
 
         # factory to call the passed callback for each job file even when chunking
         def cb_factory(i):
             if not callable(callback):
                 return None
 
-            if group_ids:
-                def wrapper(result_data):
-                    offset = sum(map(len, job_objs[:i]))
-                    for j in range(len(job_objs[i])):
-                        data = result_data if isinstance(result_data, Exception) else result_data[j]
-                        callback(offset + j, data)
-            else:
-                def wrapper(data):
-                    callback(i, data)
+            def wrapper(result_data):
+                offset = sum(map(len, list(job_objs.values())[:i]))
+                for j in range(len(list(job_objs.values())[i])):
+                    data = result_data if isinstance(result_data, Exception) else result_data[j]
+                    callback(offset + j, data)
 
             return wrapper
 
         # threaded processing
         pool = ThreadPool(threads)
         results = [
-            pool.apply_async(func, (arg,), kwargs, callback=cb_factory(i))
-            for i, arg in enumerate(job_objs)
+            pool.apply_async(func, make_tuple(arg), kwargs, callback=cb_factory(i))
+            for i, arg in enumerate(job_objs.items())
         ]
         pool.close()
         pool.join()
 
         # store result data or an exception
         result_data = result_type()
-        if group_ids:
-            for _job_objs, res in six.moves.zip(job_objs, results):
-                data = get_async_result_silent(res)
-                for job_obj in _job_objs:
-                    _data = data if isinstance(data, Exception) else data[job_obj]
-                    if isinstance(result_data, list):
-                        result_data.append(_data)
-                    else:
-                        result_data[job_obj] = _data
-        else:
-            for job_obj, res in six.moves.zip(job_objs, results):
-                data = get_async_result_silent(res)
+        for _job_objs, res in six.moves.zip(job_objs.values(), results):
+            data = get_async_result_silent(res)
+            for i, job_obj in enumerate(_job_objs):
                 if isinstance(result_data, list):
-                    result_data.append(data)
+                    result_data.append(data if isinstance(data, Exception) else data[i])
                 else:
-                    result_data[job_obj] = data
+                    result_data[job_obj] = data if isinstance(data, Exception) else data[job_obj]
 
         return result_data
 
@@ -501,10 +488,17 @@ def submit_group(self, job_files, threads=None, callback=None, **kwargs):
         file properly expanded. When an exception was raised during a submission, this exception is
         added to the returned list.
         """
+        # in order to use the generic grouping mechanism in _apply_group create a trivial group_func
+        def group_func(job_files):
+            groups = defaultdict(list)
+            for job_file in job_files:
+                groups[job_file].append(job_file)
+            return groups
+
         return self._apply_group(
             func=self.submit,
             result_type=list,
-            group_ids=False,
+            group_func=group_func,
             job_objs=job_files,
             threads=threads,
             callback=callback,
@@ -527,7 +521,7 @@ def cancel_group(self, job_ids, threads=None, callback=None, **kwargs):
         results = self._apply_group(
             func=self.cancel,
             result_type=dict,
-            group_ids=True,
+            group_func=self.group_job_ids,
             job_objs=job_ids,
             threads=threads,
             callback=callback,
@@ -553,7 +547,7 @@ def cleanup_group(self, job_ids, threads=None, callback=None, **kwargs):
         results = self._apply_group(
             func=self.cleanup,
             result_type=dict,
-            group_ids=True,
+            group_func=self.group_job_ids,
             job_objs=job_ids,
             threads=threads,
             callback=callback,
@@ -579,7 +573,7 @@ def query_group(self, job_ids, threads=None, callback=None, **kwargs):
         return self._apply_group(
             func=self.query,
             result_type=dict,
-            group_ids=True,
+            group_func=self.group_job_ids,
             job_objs=job_ids,
             threads=threads,
             callback=callback,
diff --git a/law/job/law_job.sh b/law/job/law_job.sh
index 64992ad4..67419610 100755
--- a/law/job/law_job.sh
+++ b/law/job/law_job.sh
@@ -168,7 +168,7 @@ law_job() {
 
         # function existing?
         if command -v "${name}" &> /dev/null; then
-            eval "${name}" "$@"
+            eval "${name}" "${args}"
         else
             >&2 echo "function '${name}' does not exist, skip"
             return "2"
@@ -448,7 +448,7 @@ law_job() {
     # handle input file rendering
     local render_ret
     render_variables="$( echo "${render_variables}" | base64 --decode )"
-    if [ ! -z "${input_files_render}" ] && [ ! -z "${render_variables}" ] && [ "${render_variables}" != "-" ]; then
+    if [ "${#input_files_render[@]}" != "0" ] && [ ! -z "${render_variables}" ] && [ "${render_variables}" != "-" ]; then
         echo
         _law_job_subsection "render input files"
 
diff --git a/law/util.py b/law/util.py
index e20458c8..8e3174f2 100644
--- a/law/util.py
+++ b/law/util.py
@@ -1264,11 +1264,14 @@ def perf_counter():
 
 
 def interruptable_popen(*args, **kwargs):
-    """ interruptable_popen(*args, interrupt_callback=None, kill_timeout=None, **kwargs)
+    """ interruptable_popen(*args, stdin_callback=None, stdin_delay=0, interrupt_callback=None, kill_timeout=None, **kwargs)  # noqa
     Shorthand to :py:class:`Popen` followed by :py:meth:`Popen.communicate` which can be interrupted
     by *KeyboardInterrupt*. The return code, standard output and standard error are returned in a
     3-tuple.
 
+    *stdin_callback* can be a function accepting no arguments and whose return value is passed to
+    ``communicate`` after a delay of *stdin_delay* to feed data input to the subprocess.
+
     *interrupt_callback* can be a function, accepting the process instance as an argument, that is
     called immediately after a *KeyboardInterrupt* occurs. After that, a SIGTERM signal is send to
     the subprocess to allow it to gracefully shutdown.
@@ -1279,6 +1282,8 @@ def interruptable_popen(*args, **kwargs):
     All other *args* and *kwargs* are forwarded to the :py:class:`Popen` constructor.
     """
     # get kwargs not being passed to Popen
+    stdin_callback = kwargs.pop("stdin_callback", None)
+    stdin_delay = kwargs.pop("stdin_delay", 0)
     interrupt_callback = kwargs.pop("interrupt_callback", None)
     kill_timeout = kwargs.pop("kill_timeout", None)
 
@@ -1286,9 +1291,18 @@ def interruptable_popen(*args, **kwargs):
     kwargs["preexec_fn"] = os.setsid
     p = subprocess.Popen(*args, **kwargs)
 
+    # get stdin
+    stdin_data = None
+    if callable(stdin_callback):
+        if stdin_delay > 0:
+            time.sleep(stdin_delay)
+        stdin_data = stdin_callback()
+        if isinstance(stdin_data, six.string_types):
+            stdin_data = (stdin_data + "\n").encode("utf-8")
+
     # handle interrupts
     try:
-        out, err = p.communicate()
+        out, err = p.communicate(stdin_data)
     except KeyboardInterrupt:
         # allow the interrupt_callback to perform a custom process termination
         if callable(interrupt_callback):
diff --git a/law/workflow/remote.py b/law/workflow/remote.py
index 47cb046f..ca142ef2 100644
--- a/law/workflow/remote.py
+++ b/law/workflow/remote.py
@@ -23,7 +23,8 @@
 from law.job.dashboard import NoJobDashboard
 from law.parameter import NO_FLOAT, NO_INT, get_param, DurationParameter
 from law.util import (
-    is_number, colored, iter_chunks, merge_dicts, human_duration, DotDict, ShorthandDict,
+    no_value, is_number, colored, iter_chunks, merge_dicts, human_duration, DotDict, ShorthandDict,
+    InsertableDict,
 )
 from law.logger import get_logger
 
@@ -56,14 +57,14 @@ class JobData(ShorthandDict):
 
     @classmethod
     def job_data(cls, job_id=dummy_job_id, branches=None, status=None, code=None, error=None,
-            log_file=None, **kwargs):
+            extra=None, **kwargs):
         """
         Returns a dictionary containing default job submission information such as the *job_id*,
         task *branches* covered by the job, a job *status* string, a job return code, an *error*
-        message, and a *log_file*. Additional *kwargs* are accepted but _not_ stored.
+        message, and *extra* data. Additional *kwargs* are accepted but _not_ stored.
         """
         return dict(job_id=job_id, branches=branches or [], status=status, code=code, error=error,
-            log_file=log_file)
+            extra=extra or {})
 
     def __len__(self):
         return len(self.jobs) + len(self.unsubmitted_jobs)
@@ -202,6 +203,7 @@ def __init__(self, *args, **kwargs):
         # setup the job mananger
         self.job_manager = self.create_job_manager(threads=task.submission_threads)
         self.job_manager.status_diff_styles["unsubmitted"] = ({"color": "green"}, {}, {})
+        self._job_manager_setup_kwargs = no_value
 
         # boolean per job num denoting if a job should be / was skipped
         self.skip_jobs = {}
@@ -221,15 +223,18 @@ def __init__(self, *args, **kwargs):
         # flag denoting if jobs were cancelled or cleaned up (i.e. controlled)
         self._controlled_jobs = False
 
+        # tracking url
+        self._tracking_url = None
+
+        # flag denoting whether the first log file was already printed
+        self._printed_first_log = False
+
         # lock to protect the dumping of submission data
         self._dump_lock = threading.Lock()
 
         # intially, set the number of parallel jobs which might change at some piont
         self._set_parallel_jobs(task.parallel_jobs)
 
-        # tracking url
-        self._tracking_url = None
-
     @property
     def job_data_cls(self):
         return JobData
@@ -243,10 +248,24 @@ def create_job_manager(self, **kwargs):
         """
         return
 
+    def setup_job_manager(self):
+        """
+        Hook invoked externally to further setup the job mananger or perform batch system related
+        preparations, e.g. before jobs can be submitted. The returned keyword arguments will be
+        forwarded to the submit, cancel, cleanup and query methods of the job mananger.
+        """
+        return {}
+
+    def _setup_job_manager(self):
+        if self._job_manager_setup_kwargs == no_value:
+            self._job_manager_setup_kwargs = self.setup_job_manager()
+
+        return self._job_manager_setup_kwargs or {}
+
     @abstractmethod
     def create_job_file_factory(self, **kwargs):
         """
-        Hook to instantiate and return a class derived  of
+        Hook to instantiate and return a class derived of
         :py:class:`law.job.base.BaseJobFileFactory`. This method must be implemented by inheriting
         classes and should update and forward all *kwargs* to the constructor of the respective job
         file factory.
@@ -254,11 +273,16 @@ def create_job_file_factory(self, **kwargs):
         return
 
     @abstractmethod
-    def create_job_file(self, job_num, branches):
+    def create_job_file(self, *args, **kwargs):
         """
-        Creates a job file using the :py:attr:`job_file_factory` given the job number *job_num* and
-        the list of branch numbers *branches* covered by the job. The path of the job file is
-        returned. This method must be implemented by inheriting classes.
+        Creates a job file using the :py:attr:`job_file_factory`. The expected arguments depend on
+        wether the job manager supports job grouping (:py:attr:`BaseJobManager.job_grouping`). If it
+        does, two arguments containing the job number (*job_num*) and the list of branch numbers
+        (*branches*) covered by the job. If job grouping is supported, a single dictionary mapping
+        job numbers to covered branch values must be passed. In any case, the path(s) of job files
+        are returned.
+
+        This method must be implemented by inheriting classes.
         """
         return
 
@@ -267,19 +291,34 @@ def destination_info(self):
         Hook that can return a string containing information on the location that jobs are submitted
         to. The string is appended to submission and status messages.
         """
-        return ""
+        return InsertableDict()
 
     def _destination_info_postfix(self):
         """
         Returns the destination info ready to be appended to a string.
         """
         dst_info = self.destination_info()
+        if isinstance(dst_info, dict):
+            dst_info = list(dst_info.values())
         if isinstance(dst_info, (list, tuple)):
             dst_info = ", ".join(map(str, dst_info))
         if dst_info != "":
             dst_info = ", {}".format(dst_info)
         return dst_info
 
+    @property
+    def tracking_url(self):
+        return self._tracking_url
+
+    @tracking_url.setter
+    def tracking_url(self, tracking_url):
+        old_url = self.tracking_url
+        self._tracking_url = tracking_url
+        if tracking_url and self.task:
+            self.task.set_tracking_url(tracking_url)
+            if tracking_url != old_url:
+                self.task.publish_message("tracking url: {}".format(tracking_url))
+
     @property
     def _cancel_jobs(self):
         """
@@ -357,6 +396,18 @@ def _set_parallel_jobs(self, n_parallel):
 
         return n_parallel
 
+    def _status_error_pairs(self, job_num, job_data):
+        return InsertableDict([
+            ("job", job_num),
+            ("branches", job_data["branches"]),
+            ("id", job_data["job_id"]),
+            ("status", job_data["status"]),
+            ("code", job_data["code"]),
+            ("error", job_data.get("error")),
+            ("job script error", self.job_error_messages.get(job_data["code"], no_value)),
+            ("log", job_data["extra"].get("log_file", no_value)),
+        ])
+
     def _print_status_errors(self, failed_jobs):
         print("{} in task {}:".format(
             colored("{} failed job(s)".format(len(failed_jobs)), color="red", style="bright"),
@@ -368,17 +419,19 @@ def _print_status_errors(self, failed_jobs):
         show_summary = threshold and (isinstance(threshold, bool) or len(failed_jobs) >= threshold)
 
         # show the first n errors
-        tmpl = "    job: {job_num}, branch(es): {branches}, id: {job_id}, " \
-            "status: {status}, code: {code}, error: {error}{ext}"
         for i, (job_num, data) in enumerate(six.iteritems(failed_jobs)):
-            ext = ""
-            if data["code"] in self.job_error_messages:
-                law_err = self.job_error_messages[data["code"]]
-                ext += ", job script error: {}".format(law_err)
-            if data["log_file"]:
-                ext += ", log: {}".format(data["log_file"])
-
-            print(tmpl.format(job_num=job_num, ext=ext, **data))
+            # get status pairs
+            status_pairs = self._status_error_pairs(job_num, data)
+            if isinstance(status_pairs, dict):
+                status_pairs = list(status_pairs.items())
+
+            # print the status line
+            status_line = ", ".join([
+                "{} {}".format(colored("{}:".format(key), style="bright"), value)
+                for key, value in status_pairs
+                if value != no_value
+            ])
+            print("    {}".format(status_line))
 
             if i >= self.show_errors and len(failed_jobs) > self.show_errors + 1:
                 remaining = len(failed_jobs) - self.show_errors
@@ -391,35 +444,42 @@ def _print_status_errors(self, failed_jobs):
 
         # additional error summary
         if show_summary:
-            # group by error code, status, and the first 40 characters of the error message
+            # group by error code and status
             groups = OrderedDict()
-            n_error_chars = 40
             for job_num, data in six.iteritems(failed_jobs):
-                error_trunc = (
-                    data["error"]
-                    if len(data["error"]) <= n_error_chars else
-                    data["error"][:n_error_chars - 3] + "..."
-                )
-                key = (data["code"], data["status"], error_trunc)
+                key = (data["code"], data["status"])
                 if key not in groups:
-                    groups[key] = {"n_jobs": 0, "n_branches": 0, "log_file": None}
+                    groups[key] = {"n_jobs": 0, "n_branches": 0, "log_file": None, "error": None}
                 groups[key]["n_jobs"] += 1
                 groups[key]["n_branches"] += len(data["branches"])
+                if not groups[key]["error"]:
+                    groups[key]["error"] = data["error"]
                 if not groups[key]["log_file"]:
-                    groups[key]["log_file"] = data["log_file"]
+                    groups[key]["log_file"] = data["extra"].get("log_file")
 
             # show the summary
             print(colored("error summary:", color="red", style="bright"))
-            tmpl = "    {n_jobs} jobs ({n_branches} branches) with status: {status}, " \
-                "code: {code}, error: {error}{ext}"
-            for (code, status, error_trunc), stats in six.iteritems(groups):
-                # add status messsages of known error codes
+            for (code, status), stats in six.iteritems(groups):
+                # status messsages of known error codes
+                code_str = ""
                 if code in self.job_error_messages:
-                    code = "{} ({})".format(code, self.job_error_messages[code])
+                    code_str = " ({})".format(self.job_error_messages[code])
+                # pairs for printing
+                summary_pairs = [
+                    ("status", status),
+                    ("code", str(code) + code_str),
+                    ("example error", stats.get("error")),
+                ]
                 # add an example log file
-                ext = ", example log: {log_file}".format(**stats) if stats["log_file"] else ""
+                if stats["log_file"]:
+                    summary_pairs.append(("example log", stats["log_file"]))
                 # print the line
-                print(tmpl.format(status=status, code=code, error=error_trunc, ext=ext, **stats))
+                summary_line = ", ".join([
+                    "{} {}".format(colored("{}:".format(key), style="bright"), value)
+                    for key, value in summary_pairs
+                ])
+                print("    {n_jobs} jobs ({n_branches} branches) with {summary_line}".format(
+                    summary_line=summary_line, **stats))
 
     def complete(self):
         if self.task.is_controlling_remote_jobs():
@@ -532,9 +592,7 @@ def run(self):
         # from here on, submit and/or wait while polling
 
         # maybe set a tracking url
-        self._tracking_url = self.dashboard.create_tracking_url()
-        if self._tracking_url:
-            task.set_tracking_url(self._tracking_url)
+        self.tracking_url = self.dashboard.create_tracking_url()
 
         # ensure the output directory exists
         if not self._submitted:
@@ -587,8 +645,11 @@ def cancel(self):
         if not job_ids:
             return
 
+        # setup the job manager
+        job_man_kwargs = self._setup_job_manager()
+
         # get job kwargs for cancelling
-        cancel_kwargs = self._get_job_kwargs("cancel")
+        cancel_kwargs = merge_dicts(job_man_kwargs, self._get_job_kwargs("cancel"))
 
         # cancel jobs
         task.publish_message("going to cancel {} jobs".format(len(job_ids)))
@@ -629,8 +690,11 @@ def cleanup(self):
         if not job_ids:
             return
 
+        # setup the job manager
+        job_man_kwargs = self._setup_job_manager()
+
         # get job kwargs for cleanup
-        cleanup_kwargs = self._get_job_kwargs("cleanup")
+        cleanup_kwargs = merge_dicts(job_man_kwargs, self._get_job_kwargs("cleanup"))
 
         # cleanup jobs
         task.publish_message("going to cleanup {} jobs".format(len(job_ids)))
@@ -743,7 +807,7 @@ def submit(self, retry_jobs=None):
             # set the job id in the job data
             job_data = self.job_data.jobs[job_num]
             job_data["job_id"] = job_id
-            job_data["log_file"] = files.get("log")
+            job_data["extra"]["log_file"] = files.get("log")
             new_submission_data[job_num] = copy.deepcopy(job_data)
 
             # inform the dashboard
@@ -765,7 +829,9 @@ def submit(self, retry_jobs=None):
                         print("    ... and {} more".format(remaining))
                     break
         else:
-            task.publish_message("submitted {} job(s)".format(len(submit_jobs)) + dst_info)
+            task.publish_message(
+                "submitted {} {} job(s){}".format(len(submit_jobs), self.workflow_type, dst_info),
+            )
 
         return new_submission_data
 
@@ -787,9 +853,11 @@ def _submit_batch(self, submit_jobs, **kwargs):
         if dump_freq and not is_number(dump_freq):
             dump_freq = 50
 
+        # setup the job manager
+        job_man_kwargs = self._setup_job_manager()
+
         # get job kwargs for submission and merge with passed kwargs
-        submit_kwargs = self._get_job_kwargs("submit")
-        submit_kwargs = merge_dicts(submit_kwargs, kwargs)
+        submit_kwargs = merge_dicts(job_man_kwargs, self._get_job_kwargs("submit"), kwargs)
 
         # progress callback to inform the scheduler
         def progress_callback(i, job_id):
@@ -828,13 +896,15 @@ def _submit_group(self, submit_jobs, **kwargs):
         # create the single multi submission file, passing the job_num -> branches dict
         job_file = self.create_job_file(submit_jobs)
 
+        # setup the job manager
+        job_man_kwargs = self._setup_job_manager()
+
         # get job kwargs for submission and merge with passed kwargs
-        submit_kwargs = self._get_job_kwargs("submit")
-        submit_kwargs = merge_dicts(submit_kwargs, kwargs)
+        submit_kwargs = merge_dicts(job_man_kwargs, self._get_job_kwargs("submit"), kwargs)
 
         # submission
         job_ids = self.job_manager.submit_group(
-            job_file["job"],
+            [job_file["job"]] * len(submit_jobs),
             retries=3,
             threads=task.submission_threads,
             **submit_kwargs  # noqa
@@ -844,9 +914,6 @@ def _submit_group(self, submit_jobs, **kwargs):
         for job_num, job_id in zip(submit_jobs, job_ids):
             self.job_data.jobs[job_num]["job_id"] = job_id
 
-        # dump job data
-        self.dump_job_data()
-
         return (
             job_ids,
             {job_num: job_file for job_num in submit_jobs},
@@ -870,8 +937,11 @@ def poll(self):
         n_poll_fails = 0
         start_time = time.time()
 
+        # setup the job manager
+        job_man_kwargs = self._setup_job_manager()
+
         # get job kwargs for status querying
-        query_kwargs = self._get_job_kwargs("query")
+        query_kwargs = merge_dicts(job_man_kwargs, self._get_job_kwargs("query"))
 
         # start the poll loop
         i = -1
@@ -933,16 +1003,25 @@ def poll(self):
             for job_num, (job_id, state_or_error) in zip(active_jobs, six.iteritems(query_data)):
                 if isinstance(state_or_error, Exception):
                     errors.append(state_or_error)
-                else:
-                    states_by_id[job_id] = state_or_error
-                    # sync extra info if available
-                    extra = state_or_error.get("extra")
-                    if isinstance(extra, dict):
-                        if not self._tracking_url and "tracking_url" in extra:
-                            self._tracking_url = extra["tracking_url"]
-                            task.set_tracking_url(self._tracking_url)
-                        if not self.job_data.jobs[job_num]["log_file"] and "log_file" in extra:
-                            self.job_data.jobs[job_num]["log_file"] = extra["log_file"]
+                    continue
+
+                # set the data
+                states_by_id[job_id] = state_or_error
+
+                # sync extra info if available
+                extra = state_or_error.get("extra")
+                if not isinstance(extra, dict):
+                    continue
+                self.job_data.jobs[job_num]["extra"] = extra
+
+                # set tracking url on the workflow proxy
+                if not self.tracking_url and "tracking_url" in extra:
+                    self.tracking_url = extra["tracking_url"]
+
+                # print the first log file
+                if not self._printed_first_log and extra.get("log_file"):
+                    task.publish_message("first log file: {}".format(extra["log_file"]))
+                    self._printed_first_log = True
             del query_data
 
             # print the first couple errors


Test patch:
diff --git a/.github/workflows/lint_and_test.yml b/.github/workflows/lint_and_test.yml
index 57973134..be729efa 100644
--- a/.github/workflows/lint_and_test.yml
+++ b/.github/workflows/lint_and_test.yml
@@ -48,7 +48,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python: ["27", "36", "37", "38", "39", "310"]
+        python: ["27", "36", "37", "38", "39", "310", "311"]
     name: test (python ${{ matrix.python }})
     steps:
       - name: Checkout 🛎️