Privileged Docker
-
+
@@ -1072,7 +1091,7 @@
diff --git a/src/WebUI/dotnet/WebPortal/Views/Home/ViewCluster.cshtml b/src/WebUI/dotnet/WebPortal/Views/Home/ViewCluster.cshtml
index 2951cc326..328a1ef54 100755
--- a/src/WebUI/dotnet/WebPortal/Views/Home/ViewCluster.cshtml
+++ b/src/WebUI/dotnet/WebPortal/Views/Home/ViewCluster.cshtml
@@ -166,7 +166,7 @@
-
Cluster Status:
+
Team Virtual Cluster Status:
+
+ Team VC User Status:
+
+
+
+
+
+ | User Name |
+ Used GPU |
+
+
+
+
-
Cluster Usage:
+
+
Physical Cluster Usage:
@@ -205,22 +219,10 @@
-
- User Status:
-
-
-
-
- | User Name |
- Used GPU |
-
-
-
-
- Node Status:
+ Physical Cluster Node Status:
diff --git a/src/WebUI/dotnet/WebPortal/Views/Shared/_LoginPartial.cshtml b/src/WebUI/dotnet/WebPortal/Views/Shared/_LoginPartial.cshtml
index 4ef74eb47..c8b2d082b 100755
--- a/src/WebUI/dotnet/WebPortal/Views/Shared/_LoginPartial.cshtml
+++ b/src/WebUI/dotnet/WebPortal/Views/Shared/_LoginPartial.cshtml
@@ -8,18 +8,22 @@
{
- @if (ViewData["isAuthorized"] != null && (bool)ViewData["isAuthorized"])
+ @if (Context.Session.GetString("Team") != null)
{
- -
- @Context.Session.GetString("Team")
-
-
+ try
+ {
+ -
+ @Context.Session.GetString("Team")
+
+
+ }
+ catch { /* ignored */ }
}
- Hello, @ViewData["Username"]
@@ -30,12 +34,12 @@
var querystring = location.search.replace(/^\?/, '');
var queries = querystring.split('&');
queries = queries.filter(function (query) {
- return !/^\s*$/.test(query) && !/^team=/i.test(query);
+ return !/^\s*$/.test(query) && !/^current-team=/i.test(query);
});
if (queries.length > 0) {
- querystring = '?' + queries.join('&') + '&team=';
+ querystring = '?' + queries.join('&') + '¤t-team=';
} else {
- querystring = '?team='
+ querystring = '?current-team=';
}
$('#teams-dropdown .dropdown-menu a').each(function () {
diff --git a/src/docker-images/RestfulAPI/Dockerfile b/src/docker-images/RestfulAPI/Dockerfile
index b84c01ed9..8d7ada285 100755
--- a/src/docker-images/RestfulAPI/Dockerfile
+++ b/src/docker-images/RestfulAPI/Dockerfile
@@ -1,30 +1,39 @@
-FROM dlws/restfulapi:v1.5
-MAINTAINER Hongzhi Li
-
-COPY kubectl /usr/local/bin/kubectl
-RUN chmod +x /usr/local/bin/kubectl
-#COPY gittoken /root/.ssh/id_rsa
-#RUN chmod 400 /root/.ssh/id_rsa
-
-RUN rm /etc/apache2/mods-enabled/mpm_*
-COPY mpm_prefork.conf /etc/apache2/mods-available/mpm_prefork.conf
-COPY 000-default.conf /etc/apache2/sites-available/000-default.conf
-COPY ports.conf /etc/apache2/ports.conf
-RUN ln -s /etc/apache2/mods-available/mpm_prefork.conf /etc/apache2/mods-enabled/mpm_prefork.conf
-RUN ln -s /etc/apache2/mods-available/mpm_prefork.load /etc/apache2/mods-enabled/mpm_prefork.load
-
-COPY dlws-restfulapi.wsgi /wsgi/dlws-restfulapi.wsgi
-
-
-COPY runScheduler.sh /
-RUN chmod +x /runScheduler.sh
-COPY pullsrc.sh /
-RUN chmod +x /pullsrc.sh
-COPY run.sh /
-RUN chmod +x /run.sh
-
-ADD Jobs_Templete /DLWorkspace/src/Jobs_Templete
-ADD utils /DLWorkspace/src/utils
-ADD RestAPI /DLWorkspace/src/RestAPI
-ADD ClusterManager /DLWorkspace/src/ClusterManager
-CMD /run.sh
+FROM dlws/restfulapi:v1.5
+MAINTAINER Hongzhi Li
+
+COPY kubectl /usr/local/bin/kubectl
+RUN chmod +x /usr/local/bin/kubectl
+#COPY gittoken /root/.ssh/id_rsa
+#RUN chmod 400 /root/.ssh/id_rsa
+
+RUN rm /etc/apache2/mods-enabled/mpm_*
+COPY mpm_prefork.conf /etc/apache2/mods-available/mpm_prefork.conf
+COPY 000-default.conf /etc/apache2/sites-available/000-default.conf
+COPY ports.conf /etc/apache2/ports.conf
+RUN ln -s /etc/apache2/mods-available/mpm_prefork.conf /etc/apache2/mods-enabled/mpm_prefork.conf
+RUN ln -s /etc/apache2/mods-available/mpm_prefork.load /etc/apache2/mods-enabled/mpm_prefork.load
+
+COPY dlws-restfulapi.wsgi /wsgi/dlws-restfulapi.wsgi
+
+
+COPY runScheduler.sh /
+RUN chmod +x /runScheduler.sh
+COPY pullsrc.sh /
+RUN chmod +x /pullsrc.sh
+COPY run.sh /
+RUN chmod +x /run.sh
+
+COPY ClusterManager/requirements.txt /
+# TODO refine later
+# install requirements
+RUN rm -rf /usr/lib/python2.7/dist-packages/yaml
+RUN rm -rf /usr/lib/python2.7/dist-packages/PyYAML-*
+RUN pip install -r /requirements.txt
+
+ADD Jobs_Templete /DLWorkspace/src/Jobs_Templete
+ADD utils /DLWorkspace/src/utils
+ADD RestAPI /DLWorkspace/src/RestAPI
+ADD ClusterManager /DLWorkspace/src/ClusterManager
+
+
+CMD /run.sh
diff --git a/src/docker-images/reaper/Dockerfile b/src/docker-images/reaper/Dockerfile
new file mode 100644
index 000000000..7ff9dd641
--- /dev/null
+++ b/src/docker-images/reaper/Dockerfile
@@ -0,0 +1,7 @@
+FROM python:3.7
+
+RUN pip3 install requests flask
+
+WORKDIR /reaper
+
+COPY * /reaper/
diff --git a/src/docker-images/reaper/main.py b/src/docker-images/reaper/main.py
new file mode 100644
index 000000000..6b6460414
--- /dev/null
+++ b/src/docker-images/reaper/main.py
@@ -0,0 +1,81 @@
+#!/usr/bin/python
+
+import urllib.parse
+import argparse
+import requests
+import logging
+import faulthandler
+import signal
+import json
+
+import flask
+from flask import Flask
+from flask import request
+
+logger = logging.getLogger(__name__)
+
+app = Flask(__name__)
+
+@app.route("/kill", methods=["POST"])
+def kill():
+ args = request.args
+ auth = request.headers.get("Authorization")
+ if auth != "Bearer shinigami":
+ logger.warning("get unauthorized call")
+ return "Unauthorized", 401
+ try:
+ body = json.loads(request.data.decode("utf-8"))
+
+ for alert in body["alerts"]:
+ if alert.get("status") == "resolved":
+ continue
+ logger.info("processing alert of %s", alert)
+ if not dry_run:
+ job_name = alert["labels"]["job_name"]
+ username = alert["labels"]["user_email"]
+ params = {"jobId": job_name, "userName": username}
+ args = urllib.parse.urlencode(params)
+ url = restful_url + "/KillJob?" + args
+
+ response = requests.get(url, timeout=10)
+ response.raise_for_status()
+ result = response.json().get("result")
+ if result is not None and result.startswith("Success"):
+ logger.info("killing %s success", params)
+ else:
+ logger.warning("killing %s failed", params)
+ else:
+ logger.info("reaper in dry_run mode, will not kill %s", alert)
+ return "Ok", 200
+ except Exception as e:
+ logger.exception("caught exception while processing kill, data is %s",
+ requests.data)
+ raise e
+
+def register_stack_trace_dump():
+ faulthandler.register(signal.SIGTRAP, all_threads=True, chain=False)
+
+def main(args):
+ app.run(host="0.0.0.0", port=args.port, debug=False, use_reloader=False)
+
+if __name__ == "__main__":
+ register_stack_trace_dump()
+ logging.basicConfig(format="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)s@%(thread)d - %(message)s",
+ level=logging.INFO)
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--port", "-p", default=9500, type=int,
+ help="port to listen, default 9500")
+ parser.add_argument("--restful_url", "-r", required=True,
+ help="restful api url, e.g. http://localhost:5000")
+ parser.add_argument("--dry_run", "-d", action="store_true",
+ help="if dry_run, the reaper will do nothing")
+ args = parser.parse_args()
+
+ global dry_run
+ global restful_url
+
+ dry_run = args.dry_run
+ restful_url = args.restful_url
+
+ main(args)
diff --git a/src/utils/JobRestAPIUtils.py b/src/utils/JobRestAPIUtils.py
index 3f20d58b0..850d0bd2c 100755
--- a/src/utils/JobRestAPIUtils.py
+++ b/src/utils/JobRestAPIUtils.py
@@ -17,7 +17,6 @@
import re
from config import global_vars
-from MyLogger import MyLogger
from authorization import ResourceType, Permission, AuthorizationManager, IdentityManager
import authorization
from cache import CacheManager
@@ -25,8 +24,9 @@
from ResourceInfo import ResourceInfo
import copy
+import logging
-logger = MyLogger()
+logger = logging.getLogger(__name__)
def LoadJobParams(jobParamsJsonStr):
return json.loads(jobParamsJsonStr)
@@ -237,8 +237,8 @@ def GetJobList(userName, vcName, jobOwner, num=None):
dataHandler.Close()
return jobs
except Exception as e:
- logger.error('Exception: '+ str(e))
- logger.warn("Fail to get job list for user %s, return empty list" % userName)
+ logger.error('Exception: %s', str(e))
+ logger.warn("Fail to get job list for user %s, return empty list", userName)
return []
@@ -498,7 +498,7 @@ def GetVC(userName, vcName):
clusterStatus, dummy = DataManager.GetClusterStatus()
clusterTotalRes = ResourceInfo(clusterStatus["gpu_capacity"])
- clusterReservedRes = ResourceInfo(clusterStatus["gpu_unschedulable"])
+ clusterReservedRes = ResourceInfo(clusterStatus["gpu_reserved"])
user_status = {}
diff --git a/src/utils/MyLogger.py b/src/utils/MyLogger.py
deleted file mode 100755
index ebcd48226..000000000
--- a/src/utils/MyLogger.py
+++ /dev/null
@@ -1,52 +0,0 @@
-from config import global_vars
-import logging
-from logging.config import dictConfig
-import thread
-
-
-class MyLogger:
-
- def init(self):
- if self.logger is None and "logger" in global_vars and global_vars["logger"] is not None:
- self.logger = global_vars["logger"]
-
- def __init__(self):
- self.logger = None
- self.init()
-
- def info(self, msg):
- self.init()
- txt = str(thread.get_ident()) + " : " + msg
- #print txt
- if self.logger is not None:
- self.logger.info(txt)
-
- def error(self, msg):
- self.init()
- txt = str(thread.get_ident()) + " : " + msg
- #print msg
-
- if self.logger is not None:
- self.logger.error(txt)
-
- def warn(self, msg):
- self.init()
- txt = str(thread.get_ident()) + " : " + msg
- #print msg
-
- if self.logger is not None:
- self.logger.warn(txt)
-
- def debug(self, msg):
- self.init()
- print msg
-
- if self.logger is not None:
- self.logger.debug(msg)
-
- def exception(self, msg):
- self.init()
- print msg
-
- if self.logger is not None:
- self.logger.exception(msg)
diff --git a/src/utils/MySQLDataHandler.py b/src/utils/MySQLDataHandler.py
index 03de3baa3..2496fc571 100755
--- a/src/utils/MySQLDataHandler.py
+++ b/src/utils/MySQLDataHandler.py
@@ -1,8 +1,9 @@
-# from config import config
import mysql.connector
import json
import base64
import os
+import logging
+import functools
import timeit
@@ -10,14 +11,36 @@
from config import config
from config import global_vars
-from MyLogger import MyLogger
-logger = MyLogger()
+from prometheus_client import Histogram
+logger = logging.getLogger(__name__)
-class DataHandler:
+data_handler_fn_histogram = Histogram("datahandler_fn_latency_seconds",
+ "latency for executing data handler function (seconds)",
+ buckets=(.05, .075, .1, .25, .5, .75, 1.0, 2.5, 5.0,
+ 7.5, 10.0, 12.5, 15.0, 17.5, 20.0, float("inf")),
+ labelnames=("fn_name",))
+db_connect_histogram = Histogram("db_connect_latency_seconds",
+ "latency for connecting to db (seconds)",
+ buckets=(.05, .075, .1, .25, .5, .75, 1.0, 2.5, 5.0, 7.5, float("inf")))
+
+def record(fn):
+ @functools.wraps(fn)
+ def wrapped(*args, **kwargs):
+ start = timeit.default_timer()
+ try:
+ return fn(*args, **kwargs)
+ finally:
+ elapsed = timeit.default_timer() - start
+ logger.info("DataHandler: %s, time elapsed %.2fs", fn.__name__, elapsed)
+ data_handler_fn_histogram.labels(fn.__name__).observe(elapsed)
+ return wrapped
+
+
+class DataHandler(object):
def __init__(self):
start_time = timeit.default_timer()
self.database = "DLWSCluster-%s" % config["clusterId"]
@@ -32,18 +55,17 @@ def __init__(self):
username = config["mysql"]["username"]
password = config["mysql"]["password"]
+ self.CreateDatabase()
- self.conn = mysql.connector.connect(user=username, password=password,
- host=server, database=self.database)
+ with db_connect_histogram.time():
+ self.conn = mysql.connector.connect(user=username, password=password,
+ host=server, database=self.database)
- self.CreateDatabase()
self.CreateTable()
elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler initialization, time elapsed %f s" % elapsed)
-
-
+ logger.info("DataHandler initialization, time elapsed %f s", elapsed)
def CreateDatabase(self):
if "initSQLDB" not in global_vars or not global_vars["initSQLDB"]:
@@ -218,41 +240,34 @@ def CreateTable(self):
self.conn.commit()
cursor.close()
-
+ @record
def AddStorage(self, vcName, url, storageType, metadata, defaultMountPath):
try:
- start_time = timeit.default_timer()
sql = "INSERT INTO `"+self.storagetablename+"` (storageType, url, metadata, vcName, defaultMountPath) VALUES (%s,%s,%s,%s,%s)"
cursor = self.conn.cursor()
cursor.execute(sql, (storageType, url, metadata, vcName, defaultMountPath))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: AddStorage to DB: url : %s, vcName: %s , time elapsed %f s" % (url, vcName, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
-
+ @record
def DeleteStorage(self, vcName, url):
try:
- start_time = timeit.default_timer()
sql = "DELETE FROM `%s` WHERE url = '%s' and vcName = '%s'" % (self.storagetablename, url, vcName)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteStorage: url:%s, vcName:%s, time elapsed %f s" % (url, vcName, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
-
+ @record
def ListStorages(self, vcName):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `storageType`,`url`,`metadata`,`vcName`,`defaultMountPath` FROM `%s` WHERE vcName = '%s' " % (self.storagetablename, vcName)
ret = []
@@ -267,49 +282,43 @@ def ListStorages(self, vcName):
record["defaultMountPath"] = defaultMountPath
ret.append(record)
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: ListStorages time elapsed %f s" % (elapsed))
return ret
+ @record
def UpdateStorage(self, vcName, url, storageType, metadata, defaultMountPath):
try:
- start_time = timeit.default_timer()
sql = """update `%s` set storageType = '%s', metadata = '%s', defaultMountPath = '%s' where vcName = '%s' and url = '%s' """ % (self.storagetablename, storageType, metadata, defaultMountPath, vcName, url)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateStorage: vcName: %s, url: %s, time elapsed %f s" % (vcName, url, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def AddVC(self, vcName, quota, metadata):
try:
- start_time = timeit.default_timer()
sql = "INSERT INTO `"+self.vctablename+"` (vcName, quota, metadata) VALUES (%s,%s,%s)"
cursor = self.conn.cursor()
cursor.execute(sql, (vcName, quota, metadata))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: AddVC to DB: vcName: %s , time elapsed %f s" % (vcName, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def ListVCs(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `vcName`,`quota`,`metadata` FROM `%s`" % (self.vctablename)
ret = []
@@ -322,49 +331,43 @@ def ListVCs(self):
record["metadata"] = metadata
ret.append(record)
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: ListVCs time elapsed %f s" % (elapsed))
return ret
+ @record
def DeleteVC(self, vcName):
try:
- start_time = timeit.default_timer()
sql = "DELETE FROM `%s` WHERE vcName = '%s'" % (self.vctablename, vcName)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteVC: vcName: %s , time elapsed %f s" % (vcName, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def UpdateVC(self, vcName, quota, metadata):
try:
- start_time = timeit.default_timer()
sql = """update `%s` set quota = '%s', metadata = '%s' where vcName = '%s' """ % (self.vctablename, quota, metadata, vcName)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateVC: vcName: %s , time elapsed %f s" % (vcName, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def GetIdentityInfo(self, identityName):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `identityName`,`uid`,`gid`,`groups` FROM `%s` where `identityName` = '%s'" % (self.identitytablename, identityName)
ret = []
@@ -378,23 +381,20 @@ def GetIdentityInfo(self, identityName):
record["groups"] = json.loads(groups)
ret.append(record)
except Exception as e:
- logger.error('GetIdentityInfo Exception: '+ str(e))
- pass
+ logger.error('GetIdentityInfo Exception: %s', str(e))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetIdentityInfo time elapsed %f s" % (elapsed))
return ret
+ @record
def UpdateIdentityInfo(self, identityName, uid, gid, groups):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
if (isinstance(groups, list)):
groups = json.dumps(groups)
-
+
if len(self.GetIdentityInfo(identityName)) == 0:
sql = "INSERT INTO `"+self.identitytablename+"` (identityName,uid,gid,groups) VALUES (%s,%s,%s,%s)"
cursor.execute(sql, (identityName, uid, gid, groups))
@@ -404,16 +404,14 @@ def UpdateIdentityInfo(self, identityName, uid, gid, groups):
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateIdentityInfo %s to database , time elapsed %f s" % (identityName, elapsed))
return True
except Exception as e:
- logger.error('UpdateIdentityInfo Exception: '+ str(e))
+ logger.error('UpdateIdentityInfo Exception: %s', str(e))
return False
+ @record
def GetAceCount(self, identityName, resource):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT count(ALL id) as c FROM `%s` where `identityName` = '%s' and `resource` = '%s'" % (self.acltablename,identityName, resource)
cursor.execute(query)
@@ -422,14 +420,12 @@ def GetAceCount(self, identityName, resource):
ret = c[0]
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetAceCount time elapsed %f s" % ( elapsed))
return ret
+ @record
def UpdateAce(self, identityName, identityId, resource, permissions, isDeny):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
existingAceCount = self.GetAceCount(identityName, resource)
logger.info(existingAceCount)
@@ -443,34 +439,30 @@ def UpdateAce(self, identityName, identityId, resource, permissions, isDeny):
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateAce %s - %s to database , time elapsed %f s" % (identityName, resource, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def UpdateAclIdentityId(self, identityName, identityId):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
sql = """update `%s` set identityId = '%s' where `identityName` = '%s' """ % (self.acltablename, identityId, identityName)
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateAclIdentityId %s - %s to database , time elapsed %f s" % (identityName, identityId, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def DeleteResourceAcl(self, resource):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
sql = "DELETE FROM `%s` WHERE `resource` = '%s'" % (self.acltablename, resource)
@@ -478,17 +470,15 @@ def DeleteResourceAcl(self, resource):
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteResourceAcl %s, time elapsed %f s" % (resource, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def DeleteAce(self, identityName, resource):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
sql = "DELETE FROM `%s` WHERE `identityName` = '%s' and `resource` = '%s'" % (self.acltablename, identityName, resource)
@@ -496,16 +486,14 @@ def DeleteAce(self, identityName, resource):
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteAce %s : %s time elapsed %f s" % (resource, identityName, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def GetAcl(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `identityName`,`identityId`,`resource`,`permissions`,`isDeny` FROM `%s`" % (self.acltablename)
ret = []
@@ -520,17 +508,15 @@ def GetAcl(self):
record["isDeny"] = isDeny
ret.append(record)
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetAcl time elapsed %f s" % ( elapsed))
return ret
+ @record
def GetResourceAcl(self, resource):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `identityName`,`identityId`,`resource`,`permissions`,`isDeny` FROM `%s` where `resource` = '%s'" % (self.acltablename, resource)
ret = []
@@ -545,34 +531,29 @@ def GetResourceAcl(self, resource):
record["isDeny"] = isDeny
ret.append(record)
except Exception as e:
- logger.error('Exception: '+ str(e))
- pass
+ logger.error('Exception: %s', str(e))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetResourceAcl time elapsed %f s" % ( elapsed))
return ret
+ @record
def AddJob(self, jobParams):
try:
- start_time = timeit.default_timer()
sql = "INSERT INTO `"+self.jobtablename+"` (jobId, familyToken, isParent, jobName, userName, vcName, jobType,jobParams ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)"
cursor = self.conn.cursor()
jobParam = base64.b64encode(json.dumps(jobParams))
cursor.execute(sql, (jobParams["jobId"], jobParams["familyToken"], jobParams["isParent"], jobParams["jobName"], jobParams["userName"], jobParams["vcName"], jobParams["jobType"],jobParam))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: added job %s to database, time elapsed %f s" % (jobParams["jobId"], elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
- return False
+ logger.error('Exception: %s', str(e))
+ return False
+ @record
def GetJobList(self, userName, vcName, num = None, status = None, op = ("=","or")):
- start_time = timeit.default_timer()
ret = []
cursor = self.conn.cursor()
try:
@@ -592,13 +573,12 @@ def GetJobList(self, userName, vcName, num = None, status = None, op = ("=","or"
if num is not None:
query += " limit %s " % str(num)
- start_time1 = timeit.default_timer()
cursor.execute(query)
- elapsed1 = timeit.default_timer() - start_time1
- start_time2 = timeit.default_timer()
+
+ fetch_start_time = timeit.default_timer()
data = cursor.fetchall()
- elapsed2 = timeit.default_timer() - start_time2
- logger.info ("(fetchall time: %f)" % (elapsed2))
+ fetch_elapsed = timeit.default_timer() - fetch_start_time
+ logger.info("(fetchall time: %f)", fetch_elapsed)
for (jobId,jobName,userName, vcName, jobStatus,jobStatusDetail, jobType, jobDescriptionPath, jobDescription, jobTime, endpoints, jobParams,errorMsg, jobMeta) in data:
record = {}
record["jobId"] = jobId
@@ -617,16 +597,13 @@ def GetJobList(self, userName, vcName, num = None, status = None, op = ("=","or"
record["jobMeta"] = jobMeta
ret.append(record)
except Exception as e:
- logger.error('Exception: '+ str(e))
- pass
+ logger.error('Exception: %s', str(e))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get job list for user %s , time elapsed %f s (SQL query time: %f)" % (userName, elapsed, elapsed1))
return ret
+ @record
def GetJob(self, **kwargs):
- start_time = timeit.default_timer()
valid_keys = ["jobId", "familyToken", "isParent", "jobName", "userName", "vcName", "jobStatus", "jobType", "jobTime"]
if len(kwargs) != 1: return []
key, expected = kwargs.popitem()
@@ -640,27 +617,23 @@ def GetJob(self, **kwargs):
ret = [dict(zip(columns, row)) for row in cursor.fetchall()]
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get job details with query %s=%s , time elapsed %f s" % (key, expected, elapsed))
return ret
+ @record
def AddCommand(self, jobId, command):
try:
- start_time = timeit.default_timer()
sql = "INSERT INTO `"+self.commandtablename+"` (jobId, command) VALUES (%s,%s)"
cursor = self.conn.cursor()
cursor.execute(sql, (jobId, command))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: add command to database, jobId: %s , time elapsed %f s" % (jobId, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def GetPendingCommands(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `id`, `jobId`, `command` FROM `%s` WHERE `status` = 'pending' order by `time`" % (self.commandtablename)
cursor.execute(query)
@@ -673,27 +646,23 @@ def GetPendingCommands(self):
ret.append(record)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get pending command , time elapsed %f s" % (elapsed))
return ret
+ @record
def FinishCommand(self, commandId):
try:
- start_time = timeit.default_timer()
sql = """update `%s` set status = 'run' where `id` = '%s' """ % (self.commandtablename, commandId)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: set command %s as finished , time elapsed %f s" % (commandId, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def GetCommands(self, jobId):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `time`, `command`, `status`, `output` FROM `%s` WHERE `jobId` = '%s' order by `time`" % (self.commandtablename, jobId)
cursor.execute(query)
@@ -707,8 +676,6 @@ def GetCommands(self, jobId):
ret.append(record)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get command list for job %s , time elapsed %f s" % (jobId, elapsed))
return ret
def load_json(self, raw_str):
@@ -719,9 +686,9 @@ def load_json(self, raw_str):
except:
return {}
+ @record
def GetPendingEndpoints(self):
try:
- start_time = timeit.default_timer()
jobs = self.GetJob(jobStatus="running")
# [ {endpoint1:{},endpoint2:{}}, {endpoint3:{}, ... }, ... ]
@@ -730,19 +697,33 @@ def GetPendingEndpoints(self):
# endpoint["status"] == "pending"
pendingEndpoints = {k: v for d in endpoints for k, v in d.items() if v["status"] == "pending"}
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get pending endpoints %d, time elapsed %f s" % (len(pendingEndpoints), elapsed))
return pendingEndpoints
except Exception as e:
logger.exception("Query pending endpoints failed!")
return {}
+ @record
+ def GetJobEndpoints(self, job_id):
+ try:
+ jobs = self.GetJob(jobId=job_id)
+
+ # [ {endpoint1:{},endpoint2:{}}, {endpoint3:{}, ... }, ... ]
+ endpoints = map(lambda job: self.load_json(job["endpoints"]), jobs)
+ # {endpoint1: {}, endpoint2: {}, ... }
+ # endpoint["status"] == "pending"
+ endpoints = {k: v for d in endpoints for k, v in d.items()}
+
+ return endpoints
+ except Exception as e:
+ logger.warning("Query job endpoints failed! Job {}".format(job_id), exc_info=True)
+ return {}
+
+ @record
def GetDeadEndpoints(self):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
# TODO we need job["lastUpdated"] for filtering
- query = "SELECT `endpoints` FROM jobs WHERE `jobStatus` <> 'running' order by `jobTime` DESC"
+ query = "SELECT `endpoints` FROM jobs WHERE `jobStatus` <> 'running' and `jobStatus` <> 'pending' and `jobStatus` <> 'queued' and `jobStatus` <> 'scheduling' order by `jobTime` DESC"
cursor.execute(query)
dead_endpoints = {}
for [endpoints] in cursor:
@@ -750,18 +731,14 @@ def GetDeadEndpoints(self):
dead_endpoints.update(endpoint_list)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get dead endpoints %d , time elapsed %f s" % (len(dead_endpoints), elapsed))
return dead_endpoints
except Exception as e:
- import traceback
- traceback.print_exc()
logger.exception("Query dead endpoints failed!")
return {}
+ @record
def UpdateEndpoint(self, endpoint):
try:
- start_time = timeit.default_timer()
job_id = endpoint["jobId"]
job = self.GetJob(jobId=job_id)[0]
job_endpoints = self.load_json(job["endpoints"])
@@ -774,15 +751,13 @@ def UpdateEndpoint(self, endpoint):
cursor.execute(sql, (json.dumps(job_endpoints), job_id))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: update endpoints to database, endpointId: %s , time elapsed %f s" % (endpoint["id"], elapsed))
return True
except Exception as e:
- logger.exception("Update endpoints failed!")
+ logger.exception("Update endpoints failed! Endpoints: {}".format(endpoint))
return False
+ @record
def GetPendingJobs(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `jobId`,`jobName`,`userName`, `vcName`, `jobStatus`, `jobType`, `jobDescriptionPath`, `jobDescription`, `jobTime`, `endpoints`, `jobParams`,`errorMsg` ,`jobMeta` FROM `%s` where `jobStatus` <> 'error' and `jobStatus` <> 'failed' and `jobStatus` <> 'finished' and `jobStatus` <> 'killed' order by `jobTime` DESC" % (self.jobtablename)
cursor.execute(query)
@@ -805,44 +780,37 @@ def GetPendingJobs(self):
ret.append(record)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get pending jobs %d, time elapsed %f s" % (len(ret), elapsed))
return ret
-
+ @record
def SetJobError(self, jobId, errorMsg):
try:
- start_time = timeit.default_timer()
sql = """update `%s` set jobStatus = 'error', `errorMsg` = '%s' where `jobId` = '%s' """ % (self.jobtablename, errorMsg, jobId)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: set job %s error status in database, time elapsed %f s" % (jobId, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def UpdateJobTextField(self, jobId, field, value):
try:
- start_time = timeit.default_timer()
sql = "update `%s` set `%s` = '%s' where `jobId` = '%s' " % (self.jobtablename, field, value, jobId)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: update job %s, field %s to %s, time elapsed %f s" % (jobId, field, value, elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def GetJobTextField(self, jobId, field):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `jobId`, `%s` FROM `%s` where `jobId` = '%s' " % (field, self.jobtablename, jobId)
ret = None
@@ -851,16 +819,14 @@ def GetJobTextField(self, jobId, field):
for (jobId, value) in cursor:
ret = value
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get filed %s of job %s , time elapsed %f s" % (field, jobId, elapsed))
return ret
+ @record
def AddandGetJobRetries(self, jobId):
- start_time = timeit.default_timer()
sql = """update `%s` set `retries` = `retries` + 1 where `jobId` = '%s' """ % (self.jobtablename, jobId)
cursor = self.conn.cursor()
cursor.execute(sql)
@@ -876,29 +842,25 @@ def AddandGetJobRetries(self, jobId):
ret = value
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get and update retries for job %s , time elapsed %f s" % (jobId, elapsed))
return ret
+ @record
def UpdateClusterStatus(self, clusterStatus):
try:
status = base64.b64encode(json.dumps(clusterStatus))
- start_time = timeit.default_timer()
sql = "INSERT INTO `%s` (status) VALUES ('%s')" % (self.clusterstatustablename, status)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: update cluster status, time elapsed %f s" % (elapsed))
return True
except Exception as e:
- logger.error('Exception: '+ str(e))
+ logger.error('Exception: %s', str(e))
return False
+ @record
def GetClusterStatus(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `time`, `status` FROM `%s` order by `time` DESC limit 1" % (self.clusterstatustablename)
ret = None
@@ -909,17 +871,14 @@ def GetClusterStatus(self):
ret = json.loads(base64.b64decode(value))
time = t
except Exception as e:
- logger.error('Exception: '+ str(e))
- pass
+ logger.error('Exception: %s', str(e))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get cluster status , time elapsed %f s" % (elapsed))
return ret, time
+ @record
def GetUsers(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT `identityName`,`uid` FROM `%s`" % (self.identitytablename)
ret = []
@@ -928,14 +887,12 @@ def GetUsers(self):
for (identityName,uid) in cursor:
ret.append((identityName,uid))
except Exception as e:
- logger.error('Exception: '+ str(e))
- pass
+ logger.error('Exception: %s', str(e))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info("DataHandler: get users, time elapsed %f s" % (elapsed))
return ret
+ @record
def GetActiveJobsCount(self):
cursor = self.conn.cursor()
query = "SELECT count(ALL id) as c FROM `%s` where `jobStatus` <> 'error' and `jobStatus` <> 'failed' and `jobStatus` <> 'finished' and `jobStatus` <> 'killed' " % (self.jobtablename)
@@ -948,6 +905,7 @@ def GetActiveJobsCount(self):
return ret
+ @record
def GetALLJobsCount(self):
cursor = self.conn.cursor()
query = "SELECT count(ALL id) as c FROM `%s`" % (self.jobtablename)
diff --git a/src/utils/SQLDataHandler.py b/src/utils/SQLDataHandler.py
index ca7e31581..f352ebea4 100755
--- a/src/utils/SQLDataHandler.py
+++ b/src/utils/SQLDataHandler.py
@@ -3,6 +3,7 @@
import json
import base64
import os
+import logging
import timeit
@@ -11,16 +12,16 @@
from config import config
from config import global_vars
-from MyLogger import MyLogger
+from MySQLDataHandler import record
-logger = MyLogger()
+logger = logging.getLogger(__name__)
### set to a larger number if flask is running on multithreading
sql_max_connect_num = 35
sql_live_connect_num = 25
-class SQLConnManager:
+class SQLConnManager(object):
@staticmethod
def Connect():
@@ -142,7 +143,7 @@ def ReturnConnection(conn):
global_vars["sql_lock"].release()
return None
-class DataHandler:
+class DataHandler(object):
def __init__(self):
start_time = timeit.default_timer()
self.CreateDatabase()
@@ -150,7 +151,7 @@ def __init__(self):
logger.debug ("********************** created a new Data Handler *******************")
self.conn = SQLConnManager.GetConnection()
logger.debug ("Get database connection %s" % str(self.conn))
-
+
#print "Connecting to server ..."
self.jobtablename = "jobs-%s" % config["clusterId"]
self.acltablename = "acl-%s" % config["clusterId"]
@@ -164,8 +165,6 @@ def __init__(self):
elapsed = timeit.default_timer() - start_time
logger.debug ("DataHandler initialization, time elapsed %f s" % elapsed)
-
-
def CreateDatabase(self):
if "initSQLDB" not in global_vars or not global_vars["initSQLDB"]:
logger.info("===========init SQL database===============")
@@ -327,7 +326,7 @@ def CreateTable(self):
self.conn.commit()
cursor.close()
-
+
sql = """
if not exists (select * from sysobjects where name='%s' and xtype='U')
CREATE TABLE [dbo].[%s]
@@ -349,41 +348,34 @@ def CreateTable(self):
self.conn.commit()
cursor.close()
-
+ @record
def AddStorage(self, vcName, url, storageType, metadata, defaultMountPath):
try:
- start_time = timeit.default_timer()
sql = "INSERT INTO [%s] (storageType, url, metadata, vcName, defaultMountPath) VALUES (?,?,?,?,?)""" % self.storagetablename
cursor = self.conn.cursor()
cursor.execute(sql, (storageType, url, metadata, vcName, defaultMountPath))
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: AddStorage to DB: url : %s, vcName: %s , time elapsed %f s" % (url, vcName, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def DeleteStorage(self, vcName, url):
try:
- start_time = timeit.default_timer()
sql = "DELETE FROM [%s] WHERE [url] = '%s' and [vcName] = '%s'" % (self.storagetablename, url, vcName)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteStorage: url:%s, vcName:%s, time elapsed %f s" % (url, vcName, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def ListStorages(self, vcName):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [vcName],[url],[storageType],[metadata],[defaultMountPath] FROM [%s] WHERE [vcName] = '%s' " % (self.storagetablename, vcName)
ret = []
@@ -402,45 +394,36 @@ def ListStorages(self, vcName):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: ListStorages time elapsed %f s" % (elapsed))
return ret
-
+ @record
def UpdateStorage(self, vcName, url, storageType, metadata, defaultMountPath):
try:
- start_time = timeit.default_timer()
sql = """update [%s] set storageType = '%s', metadata = '%s', defaultMountPath = '%s' where [vcName] = '%s' and [url] = '%s' """ % (self.storagetablename, storageType, metadata, defaultMountPath, vcName, url)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateStorage: vcName: %s, url: %s, time elapsed %f s" % (vcName, url, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def AddVC(self, vcName, quota, metadata):
try:
- start_time = timeit.default_timer()
sql = """INSERT INTO [%s] (vcName, quota, metadata) VALUES (?,?,?)""" % self.vctablename
cursor = self.conn.cursor()
cursor.execute(sql, vcName, quota, metadata)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: AddVC to DB: vcName: %s , time elapsed %f s" % (vcName, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def ListVCs(self, vcName):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [vcName],[quota],[metadata] FROM [%s]" % (self.vctablename)
ret = []
@@ -457,45 +440,36 @@ def ListVCs(self, vcName):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: ListVCs time elapsed %f s" % ( elapsed))
- return ret
-
+ return ret
+ @record
def DeleteVC(self, vcName):
try:
- start_time = timeit.default_timer()
sql = "DELETE FROM [%s] WHERE [vcName] = '%s'" % (self.vctablename, vcName)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteVC: vcName: %s , time elapsed %f s" % (vcName, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def UpdateVC(self, vcName, quota, metadata):
try:
- start_time = timeit.default_timer()
sql = """update [%s] set quota = '%s', metadata = '%s' where [vcName] = '%s'""" % (self.vctablename, quota, metadata, vcName)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateVC: vcName: %s , time elapsed %f s" % (vcName, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetIdentityInfo(self, identityName):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [identityName],[uid],[gid],[groups] FROM [%s] where [identityName] = '%s'" % (self.identitytablename, identityName)
ret = []
@@ -513,35 +487,29 @@ def GetIdentityInfo(self, identityName):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetIdentityInfo time elapsed %f s" % (elapsed))
- return ret
-
+ return ret
+ @record
def UpdateIdentityInfo(self, identityName, uid, gid, groups):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
-
+
if len(self.GetIdentityInfo(identityName)) == 0:
sql = """INSERT INTO [%s] (identityName,uid,gid,groups) VALUES (?,?,?,?)""" % self.identitytablename
cursor.execute(sql, identityName, uid, gid, json.dumps(groups))
else:
sql = """update [%s] set uid = '%s', gid = '%s', groups = '%s' where [identityName] = '%s' """ % (self.identitytablename, uid, gid, groups, identityName)
cursor.execute(sql)
-
+
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateIdentityInfo %s to database , time elapsed %f s" % (identityName, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetAceCount(self, identityId, resource):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT count(ALL id) as c FROM [%s] where [identityId] = '%s' and [resource] = '%s'" % (self.acltablename,identityId, resource)
cursor.execute(query)
@@ -550,88 +518,73 @@ def GetAceCount(self, identityId, resource):
ret = c[0]
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetAceCount time elapsed %f s" % ( elapsed))
- return ret
-
+ return ret
+ @record
def UpdateAce(self, identityName, identityId, resource, permissions, isDeny):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
-
+
if self.GetAceCount(identityId, resource) == 0:
sql = """INSERT INTO [%s] (identityName,identityId,resource,permissions,isDeny) VALUES (?,?,?,?,?)""" % self.acltablename
cursor.execute(sql, identityName, identityId, resource, permissions, isDeny)
else:
sql = """update [%s] set permissions = '%s' where [identityName] = '%s' and [resource] = '%s' """ % (self.acltablename, permissions, identityName, resource)
cursor.execute(sql)
-
+
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateAce %s - %s to database , time elapsed %f s" % (identityName, resource, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def UpdateAclIdentityId(self, identityName, identityId):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
sql = """update [%s] set identityName = '%s' where [identityName] = '%s' """ % (self.acltablename, identityId, identityName)
cursor.execute(sql)
-
+
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: UpdateAclIdentityId %s - %s to database , time elapsed %f s" % (identityName, identityId, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def DeleteResourceAcl(self, resource):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
-
+
sql = "DELETE FROM [%s] WHERE [resource] = '%s'" % (self.acltablename, resource)
cursor = self.conn.cursor()
-
+
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteResourceAcl %s, time elapsed %f s" % (resource, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def DeleteAce(self, identityName, resource):
try:
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
-
+
sql = "DELETE FROM [%s] WHERE [identityName] = '%s' and [resource] = '%s'" % (self.acltablename, identityName, resource)
cursor = self.conn.cursor()
-
+
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: DeleteAce %s : %s, time elapsed %f s" % (resource, identityName, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetAcl(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [identityName],[identityId],[resource],[permissions],[isDeny] FROM [%s]" % (self.acltablename)
ret = []
@@ -650,13 +603,10 @@ def GetAcl(self):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetAcl time elapsed %f s" % ( elapsed))
- return ret
-
+ return ret
+ @record
def GetResourceAcl(self, resource):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [identityName],[identityId],[resource],[permissions],[isDeny] FROM [%s] where [resource] = '%s'" % (self.acltablename, resource)
ret = []
@@ -675,30 +625,24 @@ def GetResourceAcl(self, resource):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: GetResourceAcl time elapsed %f s" % ( elapsed))
- return ret
-
+ return ret
+ @record
def AddJob(self, jobParams):
try:
- start_time = timeit.default_timer()
sql = """INSERT INTO [%s] (jobId, familyToken, isParent, jobName, userName, vcName, jobType,jobParams ) VALUES (?,?,?,?,?,?,?)""" % self.jobtablename
cursor = self.conn.cursor()
jobParam = base64.b64encode(json.dumps(jobParams))
cursor.execute(sql, jobParams["jobId"], jobParams["familyToken"], jobParams["isParent"], jobParams["jobName"], jobParams["userName"], jobParams["vcName"], jobParams["jobType"],jobParam)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: added job %s to database, time elapsed %f s" % (jobParams["jobId"],elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetJobList(self, userName, vcName, num = None, status = None, op = ("=","or")):
- start_time = timeit.default_timer()
ret = []
cursor = self.conn.cursor()
try:
@@ -715,16 +659,14 @@ def GetJobList(self, userName, vcName, num = None, status = None, op = ("=","or"
else:
status_list = [ " [jobStatus] %s '%s' " % (op[0],s) for s in status.split(',')]
status_statement = (" "+op[1]+" ").join(status_list)
- query += " and ( %s ) " % status_statement
+ query += " and ( %s ) " % status_statement
query += " order by [jobTime] Desc"
- start_time1 = timeit.default_timer()
cursor.execute(query)
- elapsed1 = timeit.default_timer() - start_time1
- start_time2 = timeit.default_timer()
+ fetch_start = timeit.default_timer()
data = cursor.fetchall()
- elapsed2 = timeit.default_timer() - start_time2
- logger.info ("(fetchall time: %f)" % (elapsed2))
+ fetch_time = timeit.default_timer() - fetch_start
+ logger.info ("(fetchall time: %f)" % (fetch_time))
for (jobId,jobName,userName, vcName,jobStatus,jobStatusDetail, jobType, jobDescriptionPath, jobDescription, jobTime, endpoints, jobParams,errorMsg, jobMeta) in data:
record = {}
record["jobId"] = jobId
@@ -744,16 +686,12 @@ def GetJobList(self, userName, vcName, num = None, status = None, op = ("=","or"
ret.append(record)
except Exception as e:
logger.error('Exception: '+ str(e))
- pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get job list for user %s , time elapsed %f s (SQL query time: %f)" % (userName, elapsed, elapsed1))
return ret
-
+ @record
def GetJob(self, **kwargs):
- start_time = timeit.default_timer()
valid_keys = ["jobId", "familyToken", "isParent", "jobName", "userName", "vcName", "jobStatus", "jobType", "jobTime"]
if len(kwargs) != 1: return []
key, expected = kwargs.popitem()
@@ -767,29 +705,23 @@ def GetJob(self, **kwargs):
ret = [dict(zip(columns, row)) for row in cursor.fetchall()]
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get job details with query %s=%s , time elapsed %f s" % (key, expected, elapsed))
return ret
-
+ @record
def AddCommand(self,jobId,command):
try:
- start_time = timeit.default_timer()
sql = """INSERT INTO [%s] (jobId, command) VALUES (?,?)""" % self.commandtablename
cursor = self.conn.cursor()
cursor.execute(sql, jobId, command)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: add command to database, jobId: %s , time elapsed %f s" % (jobId, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetPendingCommands(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [id], [jobId], [command] FROM [%s] WHERE [status] = 'pending' order by [time]" % (self.commandtablename)
cursor.execute(query)
@@ -802,29 +734,23 @@ def GetPendingCommands(self):
ret.append(record)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get pending command , time elapsed %f s" % (elapsed))
- return ret
-
+ return ret
+ @record
def FinishCommand(self,commandId):
try:
- start_time = timeit.default_timer()
sql = """update [%s] set status = 'run' where [id] = '%s' """ % (self.commandtablename, commandId)
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: set command %s as finished , time elapsed %f s" % (commandId, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetCommands(self, jobId):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [time], [command], [status], [output] FROM [%s] WHERE [jobId] = '%s' order by [time]" % (self.commandtablename, jobId)
cursor.execute(query)
@@ -838,13 +764,10 @@ def GetCommands(self, jobId):
ret.append(record)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get command list for job %s , time elapsed %f s" % (jobId, elapsed))
- return ret
-
+ return ret
+ @record
def GetPendingJobs(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [jobId],[jobName],[userName], [vcName], [jobStatus], [jobType], [jobDescriptionPath], [jobDescription], [jobTime], [endpoints], [jobParams],[errorMsg] ,[jobMeta] FROM [%s] where [jobStatus] <> 'error' and [jobStatus] <> 'failed' and [jobStatus] <> 'finished' and [jobStatus] <> 'killed' order by [jobTime] DESC" % (self.jobtablename)
cursor.execute(query)
@@ -867,45 +790,36 @@ def GetPendingJobs(self):
ret.append(record)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get pending jobs , time elapsed %f s" % (elapsed))
- return ret
-
+ return ret
+ @record
def SetJobError(self,jobId,errorMsg):
try:
- start_time = timeit.default_timer()
sql = """update [%s] set jobStatus = 'error', [errorMsg] = ? where [jobId] = '%s' """ % (self.jobtablename,jobId)
cursor = self.conn.cursor()
cursor.execute(sql,errorMsg)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: set job %s error status in database, time elapsed %f s" % (jobId, elapsed))
return True
except Exception as e:
logger.error('Exception: '+ str(e))
- return False
-
+ return False
+ @record
def UpdateJobTextField(self,jobId,field,value):
try:
- start_time = timeit.default_timer()
sql = """update [%s] set [%s] = ? where [jobId] = '%s' """ % (self.jobtablename,field, jobId)
cursor = self.conn.cursor()
cursor.execute(sql,value)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: update job %s, field %s , time elapsed %f s" % (jobId, field, elapsed))
return True
except Exception, e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetJobTextField(self,jobId,field):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [jobId], [%s] FROM [%s] where [jobId] = '%s' " % (field, self.jobtablename,jobId)
ret = None
@@ -918,12 +832,10 @@ def GetJobTextField(self,jobId,field):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get filed %s of job %s , time elapsed %f s" % (field, jobId, elapsed))
return ret
+ @record
def AddandGetJobRetries(self,jobId):
- start_time = timeit.default_timer()
sql = """update [%s] set [retries] = [retries] + 1 where [jobId] = '%s' """ % (self.jobtablename, jobId)
cursor = self.conn.cursor()
cursor.execute(sql)
@@ -939,30 +851,24 @@ def AddandGetJobRetries(self,jobId):
ret = value
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get and update retries for job %s , time elapsed %f s" % (jobId, elapsed))
return ret
-
+ @record
def UpdateClusterStatus(self,clusterStatus):
try:
- start_time = timeit.default_timer()
sql = """INSERT INTO [%s] (status) VALUES (?)""" % self.clusterstatustablename
cursor = self.conn.cursor()
status = base64.b64encode(json.dumps(clusterStatus))
cursor.execute(sql,status)
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: update cluster status, time elapsed %f s" % (elapsed))
return True
except Exception, e:
logger.error('Exception: '+ str(e))
return False
-
+ @record
def GetClusterStatus(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT TOP 1 [time], [status] FROM [%s] order by [time] DESC" % (self.clusterstatustablename)
ret = None
@@ -977,13 +883,10 @@ def GetClusterStatus(self):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get cluster status , time elapsed %f s" % (elapsed))
return ret, time
-
+ @record
def GetUsers(self):
- start_time = timeit.default_timer()
cursor = self.conn.cursor()
query = "SELECT [identityName],[uid] FROM [%s]" % (self.identitytablename)
ret = []
@@ -996,11 +899,9 @@ def GetUsers(self):
pass
self.conn.commit()
cursor.close()
- elapsed = timeit.default_timer() - start_time
- logger.info ("DataHandler: get users, time elapsed %f s" % ( elapsed))
return ret
-
+ @record
def GetActiveJobsCount(self):
cursor = self.conn.cursor()
query = "SELECT count(ALL id) as c FROM [%s] where [jobStatus] <> 'error' and [jobStatus] <> 'failed' and [jobStatus] <> 'finished' and [jobStatus] <> 'killed' " % (self.jobtablename)
@@ -1011,8 +912,9 @@ def GetActiveJobsCount(self):
self.conn.commit()
cursor.close()
- return ret
+ return ret
+ @record
def GetALLJobsCount(self):
cursor = self.conn.cursor()
query = "SELECT count(ALL id) as c FROM [%s]" % (self.jobtablename)
@@ -1023,7 +925,7 @@ def GetALLJobsCount(self):
self.conn.commit()
cursor.close()
- return ret
+ return ret
def __del__(self):
logger.debug("********************** deleted a DataHandler instance *******************")
@@ -1039,7 +941,7 @@ def Close(self):
CREATE_TABLE = False
CREATE_DB = True
dataHandler = DataHandler()
-
+
if TEST_INSERT_JOB:
jobParams = {}
jobParams["id"] = "dist-tf-00001"
@@ -1047,9 +949,8 @@ def Close(self):
jobParams["user-id"] = "hongzl"
jobParams["job-meta-path"] = "/dlws/jobfiles/***"
jobParams["job-meta"] = "ADSCASDcAE!EDASCASDFD"
-
+
dataHandler.AddJob(jobParams)
-
if CREATE_TABLE:
dataHandler.CreateTable()
diff --git a/src/utils/authorization.py b/src/utils/authorization.py
index 112f45c4e..4145102fa 100755
--- a/src/utils/authorization.py
+++ b/src/utils/authorization.py
@@ -1,5 +1,5 @@
from DataHandler import DataHandler, DataManager
-from MyLogger import MyLogger
+import logging
import json
import requests
import random
@@ -7,7 +7,7 @@
import timeit
from cache import fcache
-logger = MyLogger()
+logger = logging.getLogger(__name__)
def enum(*sequential, **named):
enums = dict(zip(sequential, range(len(sequential))), **named)
diff --git a/src/utils/cache.py b/src/utils/cache.py
index d3f51b538..a3a174447 100644
--- a/src/utils/cache.py
+++ b/src/utils/cache.py
@@ -1,94 +1,97 @@
-from functools import wraps
-import threading
-from datetime import datetime
-from datetime import timedelta
-import time
-import Queue
-import copy
-
-# decorator (with different TTL for each function)
-# No removal of entries (designed for small number of entries to be always kept in memory)
-# option to invalidate specific entries
-# non-blocking (updates done by background thread; stale data returned while fetching in progress) : avoids thundering herd for data source
-def fcache(TTLInSec=30):
- def fcache_decorator(func):
- @wraps(func)
- def wrapped_function(*args, **kwargs):
- val = CacheManager.GetValue(func, TTLInSec, args)
- if (None == val):
- return func(*args)
- return copy.deepcopy(val[0])
- return wrapped_function
- return fcache_decorator
-
-
-
-class CacheManager(object):
- data = {}
- taskQueue = Queue.Queue()
- pendingTasks = set()
-
- @staticmethod
- def Invalidate(funcName, *args):
- key = CacheManager._GetKey(funcName, args)
- if key in CacheManager.data:
- val = CacheManager.data[key]
- CacheManager.data[key] = [val[0], datetime.now()]
- print("Cache invalidated " + key)
-
- @staticmethod
- def GetValue(func, ttl, args):
- val = None
- key = CacheManager._GetKey(func.__name__, args)
- needUpdate = False
- if key not in CacheManager.data:
- print("Cache miss " + key)
- needUpdate = True
- else:
- val = CacheManager.data[key]
- print("Cache hit " + key + " " + str(val[1]) + " " + str(len(CacheManager.data)) + " " + str(CacheManager.taskQueue.qsize()))
- if CacheManager._Invalid(val):
- needUpdate = True
-
- if needUpdate and key not in CacheManager.pendingTasks:
- CacheManager.taskQueue.put((func, ttl, args))
- CacheManager.pendingTasks.add(key)
-
- return val
-
-
- @staticmethod
- def _GetKey(funcName, args):
- key = funcName
- for arg in args:
- key += "__"
- key += str(arg)
- return key
-
- @staticmethod
- def _Invalid(value):
- if value[1] < datetime.now():
- return True
- return False
-
- @staticmethod
- def _WorkerThreadFunc():
- while (True):
- try:
- while not CacheManager.taskQueue.empty():
- task = CacheManager.taskQueue.get()
- key = CacheManager._GetKey(task[0].__name__, task[2])
- if key in CacheManager.pendingTasks:
- if key not in CacheManager.data or CacheManager._Invalid(CacheManager.data[key]):
- result = task[0](*(task[2]))
- CacheManager.data[key] = [result, datetime.now() + timedelta(seconds=int(task[1]))]
- print("Cache inserted " + key)
- CacheManager.pendingTasks.remove(key)
- time.sleep(0.001)
- except Exception as e:
- print('cache exception: '+ str(e))
-
-workerThread = threading.Thread(target=CacheManager._WorkerThreadFunc, args=())
-workerThread.daemon = True
-workerThread.start()
-
+from functools import wraps
+import threading
+from datetime import datetime
+from datetime import timedelta
+import time
+import Queue
+import copy
+import logging
+
+logger = logging.getLogger(__name__)
+
+# decorator (with different TTL for each function)
+# No removal of entries (designed for small number of entries to be always kept in memory)
+# option to invalidate specific entries
+# non-blocking (updates done by background thread; stale data returned while fetching in progress) : avoids thundering herd for data source
+def fcache(TTLInSec=30):
+ def fcache_decorator(func):
+ @wraps(func)
+ def wrapped_function(*args, **kwargs):
+ val = CacheManager.GetValue(func, TTLInSec, args)
+ if (None == val):
+ return func(*args)
+ return copy.deepcopy(val[0])
+ return wrapped_function
+ return fcache_decorator
+
+
+
+class CacheManager(object):
+ data = {}
+ taskQueue = Queue.Queue()
+ pendingTasks = set()
+
+ @staticmethod
+ def Invalidate(funcName, *args):
+ key = CacheManager._GetKey(funcName, args)
+ if key in CacheManager.data:
+ val = CacheManager.data[key]
+ CacheManager.data[key] = [val[0], datetime.now()]
+ logger.info("Cache invalidated %s", key)
+
+ @staticmethod
+ def GetValue(func, ttl, args):
+ val = None
+ key = CacheManager._GetKey(func.__name__, args)
+ needUpdate = False
+ if key not in CacheManager.data:
+ logger.info("Cache miss %s", key)
+ needUpdate = True
+ else:
+ val = CacheManager.data[key]
+ logger.info("Cache hit %s %s %s %s", key, str(val[1]), str(len(CacheManager.data)), str(CacheManager.taskQueue.qsize()))
+ if CacheManager._Invalid(val):
+ needUpdate = True
+
+ if needUpdate and key not in CacheManager.pendingTasks:
+ CacheManager.taskQueue.put((func, ttl, args))
+ CacheManager.pendingTasks.add(key)
+
+ return val
+
+
+ @staticmethod
+ def _GetKey(funcName, args):
+ key = funcName
+ for arg in args:
+ key += "__"
+ key += str(arg)
+ return key
+
+ @staticmethod
+ def _Invalid(value):
+ if value[1] < datetime.now():
+ return True
+ return False
+
+ @staticmethod
+ def _WorkerThreadFunc():
+ while (True):
+ try:
+ while not CacheManager.taskQueue.empty():
+ task = CacheManager.taskQueue.get()
+ key = CacheManager._GetKey(task[0].__name__, task[2])
+ if key in CacheManager.pendingTasks:
+ if key not in CacheManager.data or CacheManager._Invalid(CacheManager.data[key]):
+ result = task[0](*(task[2]))
+ CacheManager.data[key] = [result, datetime.now() + timedelta(seconds=int(task[1]))]
+ logger.info("Cache inserted %s", key)
+ CacheManager.pendingTasks.remove(key)
+ time.sleep(0.001)
+ except Exception as e:
+ logger.warning('cache exception: %s', str(e))
+
+workerThread = threading.Thread(target=CacheManager._WorkerThreadFunc, args=())
+workerThread.daemon = True
+workerThread.start()
+
diff --git a/src/utils/config.py b/src/utils/config.py
index 7c19c84ad..25d6e572e 100755
--- a/src/utils/config.py
+++ b/src/utils/config.py
@@ -1,11 +1,11 @@
import yaml
import os
-from Queue import Queue
+from Queue import Queue
import threading
try:
- f = open(os.path.join(os.path.dirname(os.path.realpath(__file__)),"config.yaml"))
- config = yaml.load(f)
+ f = open(os.path.join(os.path.dirname(os.path.realpath(__file__)), "config.yaml"))
+ config = yaml.full_load(f)
except Exception:
config = {}
()
diff --git a/src/utils/config.yaml b/src/utils/config.yaml
new file mode 100644
index 000000000..7f3e1964e
--- /dev/null
+++ b/src/utils/config.yaml
@@ -0,0 +1,36 @@
+# It's a dummy config, for unit test only!!!
+
+# database :
+# hostname :
+# username :
+# password :
+# database :
+# mysql :
+# hostname : hao-dev-infra01.eastus.cloudapp.azure.com
+# port : 3306
+# username : root
+# password : MSFT1234!
+# kubelet-path : /usr/local/bin/kubectl
+storage-mount-path : /dlwsdata
+# root-path : /DLWorkspace/src/
+root-path : ".." # for unittest only
+# nvidiaDriverPath : /opt/nvidia-driver/current
+# clusterId : b6815f6d-612c-4feb-a8f3-bc7ceaa43993
+# domain : eastus.cloudapp.azure.com
+# apiserver: https://hao-dev-infra01.eastus.cloudapp.azure.com:1443
+# certificate-authority: /etc/kubernetes/ssl/ca.pem
+# client-certificate: /etc/kubernetes/ssl/apiserver.pem
+# client-key: /etc/kubernetes/ssl/apiserver-key.pem
+# pod_ip_range: 10.2.0.0/16
+# per_user_gpu_limit: -1
+rest-api: http://faked.uri/
+# usefreeflow: False
+# mountdescription : {'hdfs': 'Hadoop file system (replicated distribute storage).', 'azurefileshare': 'Azure file storage', 'localHDD': 'Local HDD. ', 'glusterfs': 'GlusterFS (replicated distributed storage)', 'nfs': 'NFS (remote file share)', 'emptyDir': 'Kubernetes emptyDir (folder will be erased after job termination).', 'local': 'Local SSD. '}
+# mountpoints : {'rootshare': {'curphysicalmountpoint': '/mntdlws/nfs', 'mountpoints': '', 'server': '192.168.255.1', 'type': 'nfs', 'filesharename': '/data/share'}}
+# mounthomefolder : yes
+# deploymounts : []
+# default-storage-folders : ['jobfiles', 'storage', 'work', 'namenodeshare']
+# webportal_node: hao-dev-infra01.eastus.cloudapp.azure.com
+# datasource : MySQL
+# kube_custom_scheduler: False
+# WinbindServers: []
diff --git a/src/utils/k8sUtils.py b/src/utils/k8sUtils.py
index 7f6943d65..e90713f3b 100755
--- a/src/utils/k8sUtils.py
+++ b/src/utils/k8sUtils.py
@@ -9,6 +9,7 @@
from tzlocal import get_localzone
import pytz
+import logging
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)),"../storage"))
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)),"../utils"))
@@ -28,6 +29,7 @@
import pycurl
from StringIO import StringIO
+logger = logging.getLogger(__name__)
def curl_get(url):
curl = pycurl.Curl()
@@ -53,7 +55,7 @@ def kubectl_create(jobfile, EXEC=True):
try:
output = subprocess32.check_output(["bash", "-c", config["kubelet-path"] + " create -f " + jobfile])
except Exception as e:
- print e
+ logger.exception("kubectl create")
output = ""
else:
output = "Job " + jobfile + " is not submitted to kubernetes cluster"
@@ -64,10 +66,10 @@ def kubectl_delete(jobfile, EXEC=True):
if EXEC:
try:
cmd = "bash -c '" + config["kubelet-path"] + " delete -f " + jobfile + "'"
- print cmd
+ logger.info("executing %s", cmd)
output = os.system(cmd)
except Exception as e:
- print e
+ logger.exception("kubectl delete")
output = -1
else:
output = -1
@@ -83,7 +85,7 @@ def kubectl_exec(params, timeout=None):
# TODO set the timeout
output = subprocess32.check_output(["bash", "-c", config["kubelet-path"] + " " + params], timeout=timeout)
except Exception as e:
- print "EXCEPTION: " + str(e)
+ logger.exception("kubectl exec")
output = ""
return output
@@ -166,7 +168,7 @@ def GetPod(selector):
output = kubectl_exec(" get pod -o yaml --show-all -l " + selector)
podInfo = yaml.load(output)
except Exception as e:
- print e
+ logger.exception("kubectl get pod")
podInfo = None
return podInfo
@@ -355,19 +357,6 @@ def GetJobStatus(jobId):
return output, detail
-def all_pod_ready(job_id):
- pods = GetPod("run=" + job_id)
- print("\n\n\n--------------------------------------------\n\n")
- print("=======%s" % pods)
- if pods is None:
- return False
- if "items" in pods:
- pod_status = [check_pod_status(pod) for pod in pods["items"]]
- if any([status != "Running" for status in pod_status]):
- return False
- return True
-
-
def get_node_labels(key):
url = "%s/api/v1/nodes" % (config["apiserver"])
responseStr = curl_get(url)
diff --git a/src/utils/notify.py b/src/utils/notify.py
new file mode 100644
index 000000000..584f99c7f
--- /dev/null
+++ b/src/utils/notify.py
@@ -0,0 +1,166 @@
+#!/usr/bin/env python
+
+import threading
+import time
+import logging
+import urlparse
+import json
+import smtplib
+
+from Queue import Queue
+from Queue import Empty
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+class NotifyMsg(object):
+ def __init__(self, email, alert_name):
+ self.email = email
+ self.alert_name = alert_name
+
+ def labels(self):
+ raise NotImplementedError()
+
+ def subject(self):
+ raise NotImplementedError()
+
+ def body(self):
+ return self.subject()
+
+
+class JobStateChangedMsg(NotifyMsg):
+ def __init__(self, email, alert_name, job_name, job_state):
+ super(JobStateChangedMsg, self).__init__(email, alert_name)
+ self.job_name = job_name
+ self.job_state = job_state
+
+ def labels(self):
+ return {"job_name": self.job_name, "job_state": self.job_state}
+
+ def subject(self):
+ return "Your job %s has changed to state of %s" % (self.job_name, self.job_state)
+
+
+def new_job_state_change_message(email, job_name, state):
+ return JobStateChangedMsg(email, "job-state-changed", job_name, state)
+
+
+class Notifier(object):
+ def __init__(self, config):
+ self.queue = Queue()
+ self.running = False
+ self.thread = None
+
+ self.cluster = None
+ self.alert_manager_url = None
+ self.smtp_url = self.smtp_from = self.smtp_auth_name = self.smtp_auth_pass = None
+
+ if config is not None and "notifier" in config:
+ notifier_config = config["notifier"]
+
+ self.cluster = notifier_config.get("cluster")
+ self.smtp_url = notifier_config.get("smtp-url")
+ self.smtp_from = notifier_config.get("smtp-from")
+ self.smtp_auth_name = notifier_config.get("smtp-auth-username")
+ self.smtp_auth_pass = notifier_config.get("smtp-auth-password")
+
+ alert_manager_url = notifier_config.get("alert-manager-url")
+ if alert_manager_url is not None and len(alert_manager_url) > 0:
+ if alert_manager_url[-1] == "/":
+ self.alert_manager_url = alert_manager_url + "api/v1/alerts"
+ else:
+ self.alert_manager_url = alert_manager_url + "/api/v1/alerts"
+
+ if self.cluster is None or \
+ self.alert_manager_url is None and (
+ self.smtp_url is None or \
+ self.smtp_from is None or \
+ self.smtp_auth_name is None or \
+ self.smtp_auth_pass is None):
+ logger.warning("Notifier not configured")
+
+ def start(self):
+ if not self.running:
+ self.running = True
+ self.thread = threading.Thread(target=self.process, name="notifier")
+ self.thread.start()
+
+ def stop(self):
+ if self.running:
+ self.running = False
+ self.thread.join()
+ self.thread = None
+
+ def notify(self, msg):
+ self.queue.put(msg)
+
+ def process(self):
+ while self.running:
+ try:
+ msg = self.queue.get(block=True, timeout=1) # 1s timeout
+ except Empty:
+ continue
+
+ retry_count = 0
+ sent = False
+
+ while retry_count < 3:
+ if self.send(msg):
+ sent = True
+ break
+ time.sleep(0.2)
+ retry_count += 1
+
+ if not sent:
+ logger.error("failed to send out, discard msg: %s", msg)
+
+ def send(self, msg):
+ subject = msg.subject()
+
+ try:
+ if self.alert_manager_url is not None:
+ labels = msg.labels()
+ labels.update({
+ "alertname": msg.alert_name,
+ "type": "user_alert",
+ "cluster": self.cluster,
+ "user_email": msg.email,
+ "subject": subject,
+ })
+
+ resp = requests.post(self.alert_manager_url, timeout=5,
+ data=json.dumps([{"labels": labels}]))
+ resp.raise_for_status()
+ return True
+ elif self.smtp_url is not None and \
+ self.smtp_from is not None and \
+ self.smtp_auth_name is not None and \
+ self.smtp_auth_pass is not None:
+ smtp_send_email(self.smtp_url, self.smtp_from,
+ self.smtp_auth_name, self.smtp_auth_pass,
+ msg.email, subject, msg.body())
+ return True
+ else:
+ # not configured, discard message
+ return True
+ except Exception as e:
+ logger.exception("sending email failed")
+ return False
+
+
+def smtp_send_email(smtp_url, smtp_from, smtp_auth_name, smtp_auth_pass, to, subject, body):
+ msg = "From: %s\r\nTo: %s\r\nSubject: %s\r\n\r\n%s" % (smtp_from, to, subject, body)
+ conn = smtplib.SMTP(smtp_url)
+ conn.starttls()
+ conn.login(smtp_auth_name, smtp_auth_pass)
+ conn.sendmail(smtp_from, to, msg)
+
+
+if __name__ == "__main__":
+ logging.basicConfig(format='%(asctime)s : %(levelname)s : %(message)s', level=logging.INFO)
+ notifier = Notifier({"notifier": {"cluster": "local", "alert-manager-url": "http://localhost:9093/alert-manager"}})
+ notifier.start()
+
+ notifier.notify(new_job_state_change_message("dixu@microsoft.com", "job-id", "stopped"))
+ notifier.stop()