mirror of https://github.com/veops/cmdb.git
parent
f788adc8cf
commit
3f3eda8b3c
|
@ -15,6 +15,7 @@ Flask-SQLAlchemy = "==2.5.0"
|
|||
SQLAlchemy = "==1.4.49"
|
||||
PyMySQL = "==1.1.0"
|
||||
redis = "==4.6.0"
|
||||
python-redis-lock = "==4.0.0"
|
||||
# Migrations
|
||||
Flask-Migrate = "==2.5.2"
|
||||
# Deployment
|
||||
|
|
|
@ -6,6 +6,7 @@ import datetime
|
|||
import json
|
||||
import threading
|
||||
|
||||
import redis_lock
|
||||
from flask import abort
|
||||
from flask import current_app
|
||||
from flask_login import current_user
|
||||
|
@ -45,7 +46,6 @@ from api.lib.perm.acl.acl import is_app_admin
|
|||
from api.lib.perm.acl.acl import validate_permission
|
||||
from api.lib.secrets.inner import InnerCrypt
|
||||
from api.lib.secrets.vault import VaultClient
|
||||
from api.lib.utils import Lock
|
||||
from api.lib.utils import handle_arg_list
|
||||
from api.lib.webhook import webhook_request
|
||||
from api.models.cmdb import AttributeHistory
|
||||
|
@ -61,7 +61,6 @@ from api.tasks.cmdb import ci_relation_add
|
|||
from api.tasks.cmdb import ci_relation_cache
|
||||
from api.tasks.cmdb import ci_relation_delete
|
||||
|
||||
PRIVILEGED_USERS = {"worker", "cmdb_agent", "agent"}
|
||||
PASSWORD_DEFAULT_SHOW = "******"
|
||||
|
||||
|
||||
|
@ -278,10 +277,10 @@ class CIManager(object):
|
|||
|
||||
@staticmethod
|
||||
def _auto_inc_id(attr):
|
||||
db.session.remove()
|
||||
db.session.commit()
|
||||
|
||||
value_table = TableMap(attr_name=attr.name).table
|
||||
with Lock("auto_inc_id_{}".format(attr.name), need_lock=True):
|
||||
with redis_lock.Lock(rd.r, "auto_inc_id_{}".format(attr.name)):
|
||||
max_v = value_table.get_by(attr_id=attr.id, only_query=True).order_by(
|
||||
getattr(value_table, 'value').desc()).first()
|
||||
if max_v is not None:
|
||||
|
@ -312,10 +311,10 @@ class CIManager(object):
|
|||
|
||||
unique_key = AttributeCache.get(ci_type.unique_id) or abort(
|
||||
400, ErrFormat.unique_value_not_found.format("unique_id={}".format(ci_type.unique_id)))
|
||||
if (unique_key.default and unique_key.default.get('default') == AttributeDefaultValueEnum.AUTO_INC_ID and
|
||||
not ci_dict.get(unique_key.name)):
|
||||
ci_dict[unique_key.name] = cls._auto_inc_id(unique_key)
|
||||
|
||||
unique_value = None
|
||||
if not (unique_key.default and unique_key.default.get('default') == AttributeDefaultValueEnum.AUTO_INC_ID and
|
||||
not ci_dict.get(unique_key.name)): # primary key is not auto inc id
|
||||
unique_value = ci_dict.get(unique_key.name) or ci_dict.get(unique_key.alias) or ci_dict.get(unique_key.id)
|
||||
unique_value = unique_value or abort(400, ErrFormat.unique_key_required.format(unique_key.name))
|
||||
|
||||
|
@ -327,8 +326,15 @@ class CIManager(object):
|
|||
ci = None
|
||||
record_id = None
|
||||
password_dict = {}
|
||||
need_lock = current_user.username not in current_app.config.get('PRIVILEGED_USERS', PRIVILEGED_USERS)
|
||||
with Lock(ci_type_name, need_lock=need_lock):
|
||||
with redis_lock.Lock(rd.r, ci_type.name):
|
||||
db.session.commit()
|
||||
|
||||
if (unique_key.default and unique_key.default.get('default') == AttributeDefaultValueEnum.AUTO_INC_ID and
|
||||
not ci_dict.get(unique_key.name)):
|
||||
ci_dict[unique_key.name] = cls._auto_inc_id(unique_key)
|
||||
current_app.logger.info(ci_dict[unique_key.name])
|
||||
unique_value = ci_dict[unique_key.name]
|
||||
|
||||
existed = cls.ci_is_exist(unique_key, unique_value, ci_type.id)
|
||||
if existed is not None:
|
||||
if exist_policy == ExistPolicy.REJECT:
|
||||
|
@ -463,8 +469,9 @@ class CIManager(object):
|
|||
limit_attrs = self._valid_ci_for_no_read(ci) if not _is_admin else {}
|
||||
|
||||
record_id = None
|
||||
need_lock = current_user.username not in current_app.config.get('PRIVILEGED_USERS', PRIVILEGED_USERS)
|
||||
with Lock(ci.ci_type.name, need_lock=need_lock):
|
||||
with redis_lock.Lock(rd.r, ci.ci_type.name):
|
||||
db.session.commit()
|
||||
|
||||
self._valid_unique_constraint(ci.type_id, ci_dict, ci_id)
|
||||
|
||||
ci_dict = {k: v for k, v in ci_dict.items() if k in ci_type_attrs_name}
|
||||
|
@ -912,7 +919,7 @@ class CIRelationManager(object):
|
|||
|
||||
@staticmethod
|
||||
def _check_constraint(first_ci_id, first_type_id, second_ci_id, second_type_id, type_relation):
|
||||
db.session.remove()
|
||||
db.session.commit()
|
||||
if type_relation.constraint == ConstraintEnum.Many2Many:
|
||||
return
|
||||
|
||||
|
@ -972,7 +979,7 @@ class CIRelationManager(object):
|
|||
else:
|
||||
type_relation = CITypeRelation.get_by_id(relation_type_id)
|
||||
|
||||
with Lock("ci_relation_add_{}_{}".format(first_ci.type_id, second_ci.type_id), need_lock=True):
|
||||
with redis_lock.Lock(rd.r, "ci_relation_add_{}_{}".format(first_ci.type_id, second_ci.type_id)):
|
||||
|
||||
cls._check_constraint(first_ci_id, first_ci.type_id, second_ci_id, second_ci.type_id, type_relation)
|
||||
|
||||
|
@ -1062,7 +1069,7 @@ class CIRelationManager(object):
|
|||
class CITriggerManager(object):
|
||||
@staticmethod
|
||||
def get(type_id):
|
||||
db.session.remove()
|
||||
db.session.commit()
|
||||
return CITypeTrigger.get_by(type_id=type_id, to_dict=True)
|
||||
|
||||
@staticmethod
|
||||
|
|
|
@ -644,10 +644,15 @@ class CITypeAttributeManager(object):
|
|||
existed.soft_delete()
|
||||
|
||||
for ci in CI.get_by(type_id=type_id, to_dict=False):
|
||||
AttributeValueManager.delete_attr_value(attr_id, ci.id)
|
||||
AttributeValueManager.delete_attr_value(attr_id, ci.id, commit=False)
|
||||
|
||||
ci_cache.apply_async(args=(ci.id, None, None), queue=CMDB_QUEUE)
|
||||
|
||||
for item in PreferenceShowAttributes.get_by(type_id=type_id, attr_id=attr_id, to_dict=False):
|
||||
item.soft_delete(commit=False)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
CITypeAttributeCache.clean(type_id, attr_id)
|
||||
|
||||
CITypeHistoryManager.add(CITypeOperateType.DELETE_ATTRIBUTE, type_id, attr_id=attr.id,
|
||||
|
|
|
@ -302,9 +302,9 @@ class AttributeValueManager(object):
|
|||
return self.write_change2(changed)
|
||||
|
||||
@staticmethod
|
||||
def delete_attr_value(attr_id, ci_id):
|
||||
def delete_attr_value(attr_id, ci_id, commit=True):
|
||||
attr = AttributeCache.get(attr_id)
|
||||
if attr is not None:
|
||||
value_table = TableMap(attr=attr).table
|
||||
for item in value_table.get_by(attr_id=attr.id, ci_id=ci_id, to_dict=False):
|
||||
item.delete()
|
||||
item.delete(commit=commit)
|
||||
|
|
|
@ -2,10 +2,11 @@
|
|||
|
||||
|
||||
import msgpack
|
||||
import redis_lock
|
||||
|
||||
from api.extensions import cache
|
||||
from api.extensions import rd
|
||||
from api.lib.decorator import flush_db
|
||||
from api.lib.utils import Lock
|
||||
from api.models.acl import App
|
||||
from api.models.acl import Permission
|
||||
from api.models.acl import Resource
|
||||
|
@ -136,14 +137,14 @@ class HasResourceRoleCache(object):
|
|||
|
||||
@classmethod
|
||||
def add(cls, rid, app_id):
|
||||
with Lock('HasResourceRoleCache'):
|
||||
with redis_lock.Lock(rd.r, 'HasResourceRoleCache'):
|
||||
c = cls.get(app_id)
|
||||
c[rid] = 1
|
||||
cache.set(cls.PREFIX_KEY.format(app_id), c, timeout=0)
|
||||
|
||||
@classmethod
|
||||
def remove(cls, rid, app_id):
|
||||
with Lock('HasResourceRoleCache'):
|
||||
with redis_lock.Lock(rd.r, 'HasResourceRoleCache'):
|
||||
c = cls.get(app_id)
|
||||
c.pop(rid, None)
|
||||
cache.set(cls.PREFIX_KEY.format(app_id), c, timeout=0)
|
||||
|
|
|
@ -1,8 +1,6 @@
|
|||
# -*- coding:utf-8 -*-
|
||||
|
||||
import base64
|
||||
import sys
|
||||
import time
|
||||
from typing import Set
|
||||
|
||||
import elasticsearch
|
||||
|
@ -213,52 +211,6 @@ class ESHandler(object):
|
|||
return 0, [], {}
|
||||
|
||||
|
||||
class Lock(object):
|
||||
def __init__(self, name, timeout=10, app=None, need_lock=True):
|
||||
self.lock_key = name
|
||||
self.need_lock = need_lock
|
||||
self.timeout = timeout
|
||||
if not app:
|
||||
app = current_app
|
||||
self.app = app
|
||||
try:
|
||||
self.redis = redis.Redis(host=self.app.config.get('CACHE_REDIS_HOST'),
|
||||
port=self.app.config.get('CACHE_REDIS_PORT'),
|
||||
password=self.app.config.get('CACHE_REDIS_PASSWORD'))
|
||||
except:
|
||||
self.app.logger.error("cannot connect redis")
|
||||
raise Exception("cannot connect redis")
|
||||
|
||||
def lock(self, timeout=None):
|
||||
if not timeout:
|
||||
timeout = self.timeout
|
||||
retry = 0
|
||||
while retry < 100:
|
||||
timestamp = time.time() + timeout + 1
|
||||
_lock = self.redis.setnx(self.lock_key, timestamp)
|
||||
if _lock == 1 or (
|
||||
time.time() > float(self.redis.get(self.lock_key) or sys.maxsize) and
|
||||
time.time() > float(self.redis.getset(self.lock_key, timestamp) or sys.maxsize)):
|
||||
break
|
||||
else:
|
||||
retry += 1
|
||||
time.sleep(0.6)
|
||||
if retry >= 100:
|
||||
raise Exception("get lock failed...")
|
||||
|
||||
def release(self):
|
||||
if time.time() < float(self.redis.get(self.lock_key)):
|
||||
self.redis.delete(self.lock_key)
|
||||
|
||||
def __enter__(self):
|
||||
if self.need_lock:
|
||||
self.lock()
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
if self.need_lock:
|
||||
self.release()
|
||||
|
||||
|
||||
class AESCrypto(object):
|
||||
BLOCK_SIZE = 16 # Bytes
|
||||
pad = lambda s: s + ((AESCrypto.BLOCK_SIZE - len(s) % AESCrypto.BLOCK_SIZE) *
|
||||
|
|
|
@ -4,6 +4,7 @@
|
|||
import json
|
||||
import time
|
||||
|
||||
import redis_lock
|
||||
from flask import current_app
|
||||
from flask_login import login_user
|
||||
|
||||
|
@ -20,7 +21,6 @@ from api.lib.cmdb.const import REDIS_PREFIX_CI_RELATION2
|
|||
from api.lib.decorator import flush_db
|
||||
from api.lib.decorator import reconnect_db
|
||||
from api.lib.perm.acl.cache import UserCache
|
||||
from api.lib.utils import Lock
|
||||
from api.lib.utils import handle_arg_list
|
||||
from api.models.cmdb import CI
|
||||
from api.models.cmdb import CIRelation
|
||||
|
@ -99,7 +99,7 @@ def ci_delete_trigger(trigger, operate_type, ci_dict):
|
|||
@flush_db
|
||||
@reconnect_db
|
||||
def ci_relation_cache(parent_id, child_id, ancestor_ids):
|
||||
with Lock("CIRelation_{}".format(parent_id)):
|
||||
with redis_lock.Lock(rd.r, "CIRelation_{}".format(parent_id)):
|
||||
if ancestor_ids is None:
|
||||
children = rd.get([parent_id], REDIS_PREFIX_CI_RELATION)[0]
|
||||
children = json.loads(children) if children is not None else {}
|
||||
|
@ -177,7 +177,7 @@ def ci_relation_add(parent_dict, child_id, uid):
|
|||
@celery.task(name="cmdb.ci_relation_delete", queue=CMDB_QUEUE)
|
||||
@reconnect_db
|
||||
def ci_relation_delete(parent_id, child_id, ancestor_ids):
|
||||
with Lock("CIRelation_{}".format(parent_id)):
|
||||
with redis_lock.Lock(rd.r, "CIRelation_{}".format(parent_id)):
|
||||
if ancestor_ids is None:
|
||||
children = rd.get([parent_id], REDIS_PREFIX_CI_RELATION)[0]
|
||||
children = json.loads(children) if children is not None else {}
|
||||
|
|
|
@ -37,6 +37,7 @@ PyMySQL==1.1.0
|
|||
ldap3==2.9.1
|
||||
PyYAML==6.0.1
|
||||
redis==4.6.0
|
||||
python-redis-lock==4.0.0
|
||||
requests==2.31.0
|
||||
requests_oauthlib==1.3.1
|
||||
markdownify==0.11.6
|
||||
|
|
Loading…
Reference in New Issue