Add HTCondor-CE support

gip/lib/python/gip_batch.py

@@ -2,40 +2,88 @@
 Common functions for GIP batch system providers and plugins.
 """
 
-from gip_common import cp_getBoolean, cp_get, cp_getList, cp_getInt
+from gip_common import cp_getBoolean, cp_get, cp_getList, cp_getInt, vdtDir, getHTCondorCEPort
 from gip_cluster import getOSGVersion
 from gip_sections import ce
+from gip_testing import runCommand
+import os
+import re
+import logging
+import subprocess
 
 __author__ = "Burt Holzman"
 
 def buildCEUniqueID(cp, ce_name, batch, queue):
     ce_prefix = 'jobmanager'
     if cp_getBoolean(cp, 'cream', 'enabled', False):
         ce_prefix = 'cream'
+    if cp_getBoolean(cp, 'htcondorce', 'enabled', False):
+        ce_prefix = 'htcondorce'
 
     port = getPort(cp)
     ce_unique_id = '%s:%d/%s-%s-%s' % (ce_name, port, ce_prefix, batch, queue)
     return ce_unique_id
 
 def getGramVersion(cp):
     gramVersion = '\n' + 'GlueCEInfoGRAMVersion: 5.0'
-    if cp_getBoolean(cp, 'cream', 'enabled', False):    
+    if cp_getBoolean(cp, 'cream', 'enabled', False):
+        gramVersion = ''
+    if cp_getBoolean(cp, 'htcondorce', 'enabled', False):
         gramVersion = ''
 
     return gramVersion
-
+
+def getHTCondorCEVersion(cp):
+    """
+    Returns the running version of the HTCondor CE
+    Copied from getOSGVersion() in gip_cluster.py
+    """
+    log = logging.getLogger()
+    htcondorce_ver_backup = cp_get(cp, "ce", "htcondorce_version", "1.8")
+    htcondorce_version_script = cp_get(cp, "gip", "htcondorce_version_script",
+        "")
+    htcondorce_ver = ''
+
+    if len(htcondorce_version_script) == 0:
+        htcondorce_version_script = vdtDir('$VDT_LOCATION/condor_ce_config_val',
+                                    '/usr/bin/condor_ce_config_val')
+
+        htcondorce_version_script = os.path.expandvars(htcondorce_version_script)
+
+        if not os.path.exists(htcondorce_version_script):
+            htcondorce_version_script = os.path.expandvars("$VDT_LOCATION/osg/bin/" \
+                "osg-version")
+
+    if os.path.exists(htcondorce_version_script):
+        try:
+            htcondorce_version_script += " HTCondorCEVersion"
+            htcondorce_ver = runCommand(htcondorce_version_script).read().strip()
+            htcondorce_ver = htcondorce_ver.replace('"','')
+        except Exception, e:
+            log.exception(e)
+
+    if len(htcondorce_ver) == 0:
+        htcondorce_ver = htcondorce_ver_backup
+    return htcondorce_ver
+
+
 def getCEImpl(cp):
     ceImpl = 'Globus'
     ceImplVersion = cp_get(cp, ce, 'globus_version', '4.0.6')    
     if cp_getBoolean(cp, 'cream', 'enabled', False):
         ceImpl = 'CREAM'
         ceImplVersion = getOSGVersion(cp)
+    if cp_getBoolean(cp, 'htcondorce', 'enabled', False):
+        ceImpl = 'HTCondorCE'
+        ceImplVersion = getHTCondorCEVersion(cp)
     return (ceImpl, ceImplVersion)
 
 def getPort(cp):
     port = 2119
     if cp_getBoolean(cp, 'cream', 'enabled', False):
         port = 8443
+    if cp_getBoolean(cp, 'htcondorce', 'enabled', False):
+        port = getHTCondorCEPort()
     return port
 
 def buildContactString(cp, batch, queue, ce_unique_id, log):
@@ -102,3 +150,4 @@ def getHTPCInfo(cp, batch, queue, log):
     return ('HTPCrsl: %s' % queueRSL, htpcMaxSlots)
 
 
+

gip/lib/python/gip_cese_bind.py

@@ -5,7 +5,7 @@
 """
 import re
 
-from gip_common import cp_get, cp_getBoolean
+from gip_common import cp_get, cp_getBoolean, getHTCondorCEPort
 from gip_storage import getPath
 from pbs_common import getQueueList as getPBSQueueList
 from lsf_common import getQueueList as getLSFQueueList
@@ -34,6 +34,9 @@ def getCEList(cp, extraCEs=[]):
     if cp_getBoolean(cp, 'cream', 'enabled', False):
         prefix = 'cream'
         port = 8443
+    if cp_getBoolean(cp, 'htcondorce', 'enabled', False):
+        prefix = 'htcondorce'
+        port = getHTCondorCEPort()
     ce_names = ['%s:%d/%s-%s-%%s' % (hostname, port, prefix, jobman) for hostname in hostnames]
 
     ce_list = []

gip/lib/python/gip_common.py

@@ -21,6 +21,7 @@
 import urllib
 import tempfile
 import glob
+import subprocess
 
 from gip_ldap import read_bdii
 
@@ -979,3 +980,51 @@ class _Constants: #pylint: disable-msg=C0103
     def __init__(self):
         self.CR = '\r'
         self.LF = '\n'
+
+
+def getHTCondorCEPort():
+    """Get what port condor-ce is running on based on its configuration"""
+
+    # If COLLECTOR_HOST is defined and has a port number, use that.
+    # Else, if COLLECTOR_PORT is defined, use that.
+    # Else, use 9619.
+
+    collector_host = getCondorCEConfigVal('COLLECTOR_HOST')
+    if collector_host:
+        if collector_host.count(':') > 1:
+            # ipv6 address, must be bracketed if it has a port at the end, i.e. [ADDR]:PORT
+            match = re.search(r'\]:(\d+)$', collector_host)
+            if match:
+                return int(match.group(1))
+        else:
+            # at most 1 colon -> hostname or ipv4 address with or without port
+            match = re.search(r':(\d+)$', collector_host)
+            if match:
+                return int(match.group(1))
+
+    try:
+        return int(getCondorCEConfigVal('COLLECTOR_PORT'))
+    except (TypeError, ValueError):
+        return 9619
+
+
+def getCondorCEConfigVal(variable):
+    """Use condor_ce_config_val to return the expanded value of a variable.
+    Returns: the stripped output of condor_ce_config_val, or None if the
+    variable is undefined or there is an error.
+    """
+    log = logging.getLogger()
+    try:
+        process = subprocess.Popen(['condor_ce_config_val', '-expand', variable],
+                                   stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        output, error = process.communicate()
+        if error and not error.startswith('Not defined:'):
+            log.warning('condor_ce_config_val on %s reported error: %s' % (variable, error))
+        if process.returncode != 0:
+          return None
+        return output.strip()
+    except OSError:
+        return None
+
+
+

gip/lib/python/gip_osg.py

@@ -9,7 +9,7 @@
 import ConfigParser
 
 from gip_sections import ce, site, pbs, condor, sge, lsf, se, subcluster, \
-    cluster, cesebind, cream, slurm
+    cluster, cesebind, cream, slurm, htcondorce
 from gip_common import getLogger, py23, vdtDir, get_file_list
 
 log = getLogger("GIP")
@@ -25,13 +25,10 @@
 lsf_sec = 'LSF'
 cream_sec = 'CREAM'
 slurm_sec = 'SLURM'
+gateway_sec = 'Gateway'
 
-default_osg_ress_servers = \
-    "https://osg-ress-1.fnal.gov:8443/ig/services/CEInfoCollector[OLD_CLASSAD]"
 default_osg_bdii_servers = \
     "http://is1.grid.iu.edu:14001[RAW], http://is2.grid.iu.edu:14001[RAW]"
-default_itb_ress_servers = \
-    "https://osg-ress-4.fnal.gov:8443/ig/services/CEInfoCollector[OLD_CLASSAD]"
 # We have a way to distinguish ITB from OSG sites so they don't need to go
 # to a separate server (SOFTWARE-1406):
 default_itb_bdii_servers = default_osg_bdii_servers
@@ -289,6 +286,10 @@ def __write_config_value(section, option, new_val):
     if cp2.has_section(cream_sec) and cp2.has_option(cream_sec, 'enabled'):
         __write_config(cream_sec, 'enabled', cream, 'enabled')
 
+    # [Gateway]
+    if cp2.has_section(gateway_sec) and cp2.has_option(gateway_sec, 'htcondor_gateway_enabled'):
+        __write_config(gateway_sec, 'htcondor_gateway_enabled', htcondorce, 'enabled')
+
     # [Misc Services]
     glexec_enabled = False
     if cp2.has_section(misc_sec) and cp2.has_option(misc_sec, 'glexec_location'):
@@ -346,7 +347,6 @@ def __write_config_value(section, option, new_val):
     __write_config(gip_sec, "advertise_gums", site, "advertise_gums")
     __write_config(gip_sec, "other_ces", cluster, "other_ces")
     __write_config(gip_sec, "bdii_endpoints", "gip", "bdii_endpoints")
-    __write_config(gip_sec, "ress_endpoints", "gip", "ress_endpoints")
 
     cluster_name = cp_get(cp2, gip_sec, "cluster_name", "")
     if len(cluster_name) > 0:
@@ -660,8 +660,8 @@ def config_info(ocp, gcp):
     """
     Configure the information services.  Right now, this means that we look at
     and configure the CEMon or Info Services section from config.ini to
-    determine the BDII and ReSS endpoints.  We then save this to the [GIP]
-    configuration section in the bdii_endpoints and ress_endpoints attributes.
+    determine the BDII endpoints.  We then save this to the [GIP]
+    configuration section in the bdii_endpoints attribute.
 
     If all else fails, we default to the OSG servers
     """
@@ -677,7 +677,6 @@ def config_info(ocp, gcp):
         override = False
 
 
-    ress_endpoints = []
     bdii_endpoints = []
 
     # In OSG 3.1, the osg config section is [Cemon]; in 3.2 it's [Info Services]
@@ -703,25 +702,11 @@ def get_endpoints(cp, name, default):
         return parse_endpoints(name_str)
 
     # These are the default endpoints
-    osg_ress_servers = get_endpoints(ocp, "osg-ress-servers",
-                                     default_osg_ress_servers)
     osg_bdii_servers = get_endpoints(ocp, "osg-bdii-servers",
                                      default_osg_bdii_servers)
-    itb_ress_servers = get_endpoints(ocp, "itb-ress-servers",
-                                     default_itb_ress_servers)
     itb_bdii_servers = get_endpoints(ocp, "itb-bdii-servers",
                                      default_itb_bdii_servers)
 
-    # See if the admins set something by hand; if not, go to the correct
-    # endpoint depending on the grid.
-    ress_servers = cp_get(ocp, info_section, "ress_servers", "UNAVAILABLE")
-    ress_servers = parse_endpoints(ress_servers)
-    if not ress_servers:
-        if is_osg:
-            ress_servers = osg_ress_servers
-        else:
-            ress_servers = itb_ress_servers
-
     bdii_servers = cp_get(ocp, info_section, "bdii_servers", "UNAVAILABLE")
     bdii_servers = parse_endpoints(bdii_servers)
     if not bdii_servers:
@@ -742,14 +727,6 @@ def get_endpoints(cp, name, default):
         log.info("Previously configured BDII endpoints: %s." % \
             ", ".join(gip_bdii_servers))
 
-    gip_ress_servers = cp_get(gcp, "gip", "ress_endpoints", None)
-    if (ress_servers and override) or (ress_servers and not gip_ress_servers):
-        gcp.set("gip", "ress_endpoints", ", ".join(ress_servers))
-        log.info("Configured ReSS endpoints: %s." % ", ".join(ress_servers))
-    else:
-        log.info("Previously configured ReSS endpoints: %s." % \
-            ", ".join(gip_ress_servers))
-
 def getSiteName(cp):
     siteName = cp_get(cp, site_sec, "resource_group", "UNKNOWN")
     if siteName == "UNKNOWN":

gip/lib/python/gip_sections.py

@@ -10,4 +10,5 @@
 sge = 'sge'
 lsf = 'lsf'
 cream = 'cream'
+htcondorce = 'htcondorce'
 slurm = 'slurm'