From 6251be7e3d33700fe0366a3427d4c1ff338894e1 Mon Sep 17 00:00:00 2001 From: Firmlyzhu Date: Wed, 8 May 2019 18:19:08 +0800 Subject: [PATCH] Make releasemgr work --- src/master/httprest.py | 2 +- src/master/releasemgr.py | 43 ++++++++++++++++++++++++++++----------- src/master/taskmgr.py | 2 +- src/master/vclustermgr.py | 1 + src/utils/model.py | 7 +++++-- user/user.py | 15 ++++++++++++++ 6 files changed, 54 insertions(+), 16 deletions(-) diff --git a/src/master/httprest.py b/src/master/httprest.py index 51f47df..907fdbc 100755 --- a/src/master/httprest.py +++ b/src/master/httprest.py @@ -1162,7 +1162,7 @@ if __name__ == '__main__': G_imagemgr = imagemgr.ImageMgr() logger.info("imagemgr started") - G_releasemgr = releasemgr.ReleaseMgr(G_vclustermgr,G_ulockmgr,10) + G_releasemgr = releasemgr.ReleaseMgr(G_vclustermgr,G_ulockmgr) G_releasemgr.start() logger.info("releasemgr started") diff --git a/src/master/releasemgr.py b/src/master/releasemgr.py index 2d276fd..afa4915 100644 --- a/src/master/releasemgr.py +++ b/src/master/releasemgr.py @@ -1,4 +1,4 @@ -import threading, time, requests, json +import threading, time, requests, json, traceback from utils import env from utils.log import logger from utils.model import db, VCluster, Container @@ -26,15 +26,17 @@ class ReleaseMgr(threading.Thread): self.release_days = int(env.getenv("RELEASE_DAYS")) if self.release_days <= self.warning_days: self.release_days = self.warning_days+1 + logger.info("[ReleaseMgr] start withe warning_days=%d release_days=%d"%(self.warning_days, self.release_days)) - def _send_email(to_address, username, vcluster, days, is_released=True): + def _send_email(self, to_address, username, vcluster, days, is_released=True): email_from_address = settings.get('EMAIL_FROM_ADDRESS') if (email_from_address in ['\'\'', '\"\"', '']): return text = '

Dear '+ username + ':

' + st_str = vcluster.stop_time.strftime("%Y-%m-%d %H:%M:%S") text += '''

      Your workspace/vcluster(name:%s id:%d) in %s - has been stopped more than %d days now.

- ''' % (vc.clustername, vc.clusterid, env.getenv("PORTAL_URL"), env.getenv("PORTAL_URL"), days) + has been stopped more than %d days now(stopped at:%s).

+ ''' % (vcluster.clustername, vcluster.clusterid, env.getenv("PORTAL_URL"), env.getenv("PORTAL_URL"), days, st_str) if is_released: text += '''

      Therefore, the workspace/vcluster has been released now.

      And the data in it couldn't be recoverd unless you save it.

@@ -43,7 +45,7 @@ class ReleaseMgr(threading.Thread): else: #day_d = self.release_days - (datetime.datetime.now() - vcluster.stop_time).days release_date = vcluster.stop_time + datetime.timedelta(days=self.release_days) - day_d = (release_date - vcluster.stop_time).days + day_d = (release_date - datetime.datetime.now()).days rd_str = release_date.strftime("%Y-%m-%d %H:%M:%S") text += '''

      It will be released after %s(in about %d days).

      And the data in it couldn't be recoverd after releasing.

@@ -81,6 +83,7 @@ class ReleaseMgr(threading.Thread): quotas[group['name']] = group['quotas'] vcs = VCluster.query.filter_by(status='stopped').all() + #logger.info(str(vcs)) for vc in vcs: if vc.stop_time is None: continue @@ -89,28 +92,44 @@ class ReleaseMgr(threading.Thread): if days >= self.release_days: logger.info("[ReleaseMgr] VCluster(id:%d,user:%s) has been stopped(%s) for more than %d days, it will be released." % (vc.clusterid, vc.ownername, vc.stop_time.strftime("%Y-%m-%d %H:%M:%S"), self.release_days)) - rc_info = post_to_user("/master/user/recoverinfo/", {'username':username,'auth_key':auth_key}) + rc_info = post_to_user("/master/user/recoverinfo/", {'username':vc.ownername,'auth_key':auth_key}) logger.info("[ReleaseMgr] %s"%str(rc_info)) groupname = rc_info['groupname'] - user_info = {"data":{"id":rc_info['uid'],"groupinfo":quotas[groupname]}} + user_info = {"data":{"id":rc_info['uid'],"group":groupname,"groupinfo":quotas[groupname]}} self.ulockmgr.acquire(vc.ownername) try: - success, msg = self.vclustermgr.delete_cluster(vc.clustername, vc.ownername, user_info) + [status, usage_info] = self.vclustermgr.get_clustersetting(vc.clustername, vc.ownername, "all", True) + success, msg = self.vclustermgr.delete_cluster(vc.clustername, vc.ownername, json.dumps(user_info)) if not success: logger.error("[ReleaseMgr] Can't release VCluster(id:%d,user:%s) for %s"%(vc.clusterid, vc.ownername, msg)) else: - self._send_email(rc_info['email'], vc.ownername, vc, days) + if status: + logger.info("[ReleaseMgr] Release Quota.") + post_to_user("/master/user/usageRelease/", {'auth_key':auth_key,'username':vc.ownername, + 'cpu':usage_info['cpu'], 'memory':usage_info['memory'],'disk':usage_info['disk']}) + self._send_email(rc_info['email'], vc.ownername, vc, self.release_days) + logger.info("[ReleaseMgr] Succeed to releasing VCluster(id:%d,user:%s) for %s. Send mail to info."%(vc.clusterid, vc.ownername, msg)) except Exception as err: - logger.error(err) + logger.error(traceback.format_exc()) finally: self.ulockmgr.release(vc.ownername) elif days >= self.warning_days: logger.info("[ReleaseMgr] VCluster(id:%d,user:%s) has been stopped(%s) for more than %d days. A warning email will be sent to the user." % (vc.clusterid, vc.ownername, vc.stop_time.strftime("%Y-%m-%d %H:%M:%S"), self.warning_days)) - rc_info = post_to_user("/master/user/recoverinfo/", {'username':username,'auth_key':auth_key}) + if vc.is_warned: + logger.info("[ReleaseMgr] VCluster(id:%d,user:%s) has been warned before. Skip it."% (vc.clusterid, vc.ownername)) + continue + rc_info = post_to_user("/master/user/recoverinfo/", {'username':vc.ownername,'auth_key':auth_key}) logger.info("[ReleaseMgr] %s"%str(rc_info)) - self._send_email(rc_info['email'], vc.ownername, vc, days, False) + self._send_email(rc_info['email'], vc.ownername, vc, self.warning_days, False) + vc.is_warned = True + + try: + db.session.commit() + except Exception as err: + db.session.rollback() + logger.warning(traceback.format_exc()) time.sleep(self.check_interval) def stop(self): diff --git a/src/master/taskmgr.py b/src/master/taskmgr.py index 0bed1a6..65a93b9 100644 --- a/src/master/taskmgr.py +++ b/src/master/taskmgr.py @@ -179,7 +179,7 @@ class TaskMgr(threading.Thread): self.free_nets = [] for i in range(0, (1 << (32-self.batch_cidr)) - 1, (1 << self.task_cidr)): self.free_nets.append(i) - self.logger.info("Free nets addresses pool %s" % str(self.free_nets)) + #self.logger.info("Free nets addresses pool %s" % str(self.free_nets)) self.logger.info("Each Batch Net CIDR:%s"%(str(self.task_cidr))) def data_lock(lockname): diff --git a/src/master/vclustermgr.py b/src/master/vclustermgr.py index e264871..26f5980 100755 --- a/src/master/vclustermgr.py +++ b/src/master/vclustermgr.py @@ -624,6 +624,7 @@ class VclusterMgr(object): [status,vcluster] = self.get_vcluster(clustername,username) vcluster.status ='running' vcluster.start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + vcluster.is_warned = False if not db_commit(): return [False, "Commit Errror"] return [True, "start cluster"] diff --git a/src/utils/model.py b/src/utils/model.py index 9437a51..fa04df1 100755 --- a/src/utils/model.py +++ b/src/utils/model.py @@ -160,7 +160,7 @@ class UserUsage(db.Model): self.disk = '0' def __repr__(self): - return '' % self.name + return 'cpu:%s memory:%s disk:%s' % (self.username,self.cpu,self.memory,self.disk) class Notification(db.Model): id = db.Column(db.Integer, primary_key=True) @@ -379,6 +379,7 @@ class VCluster(db.Model): create_time = db.Column(db.DateTime) start_time = db.Column(db.String(20)) stop_time = db.Column(db.DateTime) + is_warned = db.Column(db.Boolean) proxy_server_ip = db.Column(db.String(20)) proxy_public_ip = db.Column(db.String(20)) port_mapping = db.relationship('PortMapping', backref='v_cluster', lazy='dynamic') @@ -398,7 +399,8 @@ class VCluster(db.Model): self.billing_history = [] self.create_time = datetime.now() self.start_time = "------" - self.stop_time = None + self.stop_time = datetime.now() + self.is_warned = False def __repr__(self): info = {} @@ -416,6 +418,7 @@ class VCluster(db.Model): info['stop_time'] = "------" else: info['stop_time'] = self.stop_time.strftime("%Y-%m-%d %H:%M:%S") + info["is_warned"] = self.is_warned info["containers"] = [dict(eval(str(con))) for con in self.containers] info["port_mapping"] = [dict(eval(str(pm))) for pm in self.port_mapping] info["billing_history"] = [dict(eval(str(bh))) for bh in self.billing_history] diff --git a/user/user.py b/user/user.py index abfbdbc..ec9f9b7 100755 --- a/user/user.py +++ b/user/user.py @@ -325,6 +325,21 @@ def get_master_groupinfo(): groupfile.close() return json.dumps({'success':'true', 'groups':json.dumps(groups)}) +@app.route("/master/user/usageRelease/", methods=['POST']) +@auth_key_required +def usageRelease_master(): + global G_usermgr + logger.info("handle request: /master/user/usageRelease/") + form = request.form + user = form.get("username",None) + cur_user = User.query.filter_by(username=user).first() + if user is None or cur_user is None: + return json.dumps({'success':'false', 'message':'Null username field or user does not exist.'}) + G_lockmgr.acquire('__usage_'+str(user)) + result = G_usermgr.usageRelease(cur_user = cur_user, cpu = form.get('cpu'), memory = form.get('memory'), disk = form.get('disk')) + G_lockmgr.release('__usage_'+str(user)) + return json.dumps(result) + @app.route("/user/selfModify/", methods=['POST']) @login_required def selfModify_user(cur_user, user, form):