7x31/cmdb
1
0
Fork 0
mirror of https://github.com/veops/cmdb.git synced 2025-08-07 18:36:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: ci triggers

Browse Source
This commit is contained in:
pycook
2023-09-26 21:18:34 +08:00
parent 983cb4041c
commit 373346e71e
13 changed files with 550 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
cmdb-api/Pipfile
View File

@@ -44,11 +44,12 @@ treelib = "==1.6.1"
flasgger = "==0.9.5"
Pillow = "==9.3.0"
# other
six = "==1.12.0"
six = "==1.16.0"
bs4 = ">=0.0.1"
toposort = ">=1.5"
requests = ">=2.22.0"
requests_oauthlib = "==1.3.1"
markdownify = "==0.11.6"
PyJWT = "==2.4.0"
elasticsearch = "==7.17.9"
future = "==0.18.3"

65
cmdb-api/api/commands/click_cmdb.py
View File

@@ -15,7 +15,6 @@ import api.lib.cmdb.ci
from api.extensions import db
from api.extensions import rd
from api.lib.cmdb.cache import AttributeCache
from api.lib.cmdb.ci_type import CITypeTriggerManager
from api.lib.cmdb.const import PermEnum
from api.lib.cmdb.const import REDIS_PREFIX_CI
from api.lib.cmdb.const import REDIS_PREFIX_CI_RELATION
@@ -24,8 +23,8 @@ from api.lib.cmdb.const import RoleEnum
from api.lib.cmdb.const import ValueTypeEnum
from api.lib.exception import AbortException
from api.lib.perm.acl.acl import ACLManager
from api.lib.perm.acl.acl import UserCache
from api.lib.perm.acl.cache import AppCache
from api.lib.perm.acl.cache import UserCache
from api.lib.perm.acl.resource import ResourceCRUD
from api.lib.perm.acl.resource import ResourceTypeCRUD
from api.lib.perm.acl.role import RoleCRUD
@@ -227,50 +226,60 @@ def cmdb_counter():
@with_appcontext
def cmdb_trigger():
"""
Trigger execution
Trigger execution for date attribute
"""
from api.lib.cmdb.ci import CITriggerManager
current_day = datetime.datetime.today().strftime("%Y-%m-%d")
trigger2cis = dict()
trigger2completed = dict()
i = 0
while True:
db.session.remove()
if datetime.datetime.today().strftime("%Y-%m-%d") != current_day:
trigger2cis = dict()
trigger2completed = dict()
current_day = datetime.datetime.today().strftime("%Y-%m-%d")
try:
db.session.remove()
if i == 360 or i == 0:
i = 0
try:
triggers = CITypeTrigger.get_by(to_dict=False)
if datetime.datetime.today().strftime("%Y-%m-%d") != current_day:
trigger2cis = dict()
trigger2completed = dict()
current_day = datetime.datetime.today().strftime("%Y-%m-%d")
if i == 3 or i == 0:
i = 0
triggers = CITypeTrigger.get_by(to_dict=False, __func_isnot__key_attr_id=None)
for trigger in triggers:
ready_cis = CITypeTriggerManager.waiting_cis(trigger)
try:
ready_cis = CITriggerManager.waiting_cis(trigger)
except Exception as e:
print(e)
continue
if trigger.id not in trigger2cis:
trigger2cis[trigger.id] = (trigger, ready_cis)
else:
cur = trigger2cis[trigger.id]
cur_ci_ids = {i.ci_id for i in cur[1]}
trigger2cis[trigger.id] = (trigger, cur[1] + [i for i in ready_cis if i.ci_id not in cur_ci_ids
and i.ci_id not in trigger2completed[trigger.id]])
trigger2cis[trigger.id] = (
trigger, cur[1] + [i for i in ready_cis if i.ci_id not in cur_ci_ids
and i.ci_id not in trigger2completed.get(trigger.id, {})])
except Exception as e:
print(e)
for tid in trigger2cis:
trigger, cis = trigger2cis[tid]
for ci in copy.deepcopy(cis):
if CITriggerManager.trigger_notify(trigger, ci):
trigger2completed.setdefault(trigger.id, set()).add(ci.ci_id)
for tid in trigger2cis:
trigger, cis = trigger2cis[tid]
for ci in copy.deepcopy(cis):
if CITypeTriggerManager.trigger_notify(trigger, ci):
trigger2completed.setdefault(trigger.id, set()).add(ci.ci_id)
for _ci in cis:
if _ci.ci_id == ci.ci_id:
cis.remove(_ci)
for _ci in cis:
if _ci.ci_id == ci.ci_id:
cis.remove(_ci)
i += 1
time.sleep(10)
i += 1
time.sleep(10)
except Exception as e:
import traceback
print(traceback.format_exc())
current_app.logger.error("cmdb trigger exception: {}".format(e))
time.sleep(60)
@click.command()

174
cmdb-api/api/lib/cmdb/ci.py
View File

@@ -54,6 +54,7 @@ from api.models.cmdb import CITypeRelation
from api.models.cmdb import CITypeTrigger
from api.tasks.cmdb import ci_cache
from api.tasks.cmdb import ci_delete
from api.tasks.cmdb import ci_delete_trigger
from api.tasks.cmdb import ci_relation_add
from api.tasks.cmdb import ci_relation_cache
from api.tasks.cmdb import ci_relation_delete
@@ -464,6 +465,17 @@ class CIManager(object):
ci_dict = cls.get_cis_by_ids([ci_id])
ci_dict = ci_dict and ci_dict[0]
triggers = CITriggerManager.get(ci_dict['_type'])
for trigger in triggers:
option = trigger['option']
if not option.get('enable') or option.get('action') != OperateType.DELETE:
continue
if option.get('filter') and not CITriggerManager.ci_filter(ci_dict.get('_id'), option['filter']):
continue
ci_delete_trigger.apply_async(args=(trigger, OperateType.DELETE, ci_dict), queue=CMDB_QUEUE)
attrs = CITypeAttribute.get_by(type_id=ci.type_id, to_dict=False)
attr_names = set([AttributeCache.get(attr.attr_id).name for attr in attrs])
for attr_name in attr_names:
@@ -486,9 +498,9 @@ class CIManager(object):
db.session.commit()
record_id = AttributeHistoryManger.add(None, ci_id, [(None, OperateType.DELETE, ci_dict, None)], ci.type_id)
AttributeHistoryManger.add(None, ci_id, [(None, OperateType.DELETE, ci_dict, None)], ci.type_id)
ci_delete.apply_async(args=(ci_dict, OperateType.DELETE, record_id), queue=CMDB_QUEUE)
ci_delete.apply_async(args=(ci_id,), queue=CMDB_QUEUE)
return ci_id
@@ -910,56 +922,89 @@ class CIRelationManager(object):
class CITriggerManager(object):
@staticmethod
def get(type_id):
return CITypeTrigger.get_by(type_id=type_id, to_dict=False)
db.session.remove()
return CITypeTrigger.get_by(type_id=type_id, to_dict=True)
@staticmethod
def _exec_webhook(operate_type, webhook, ci_dict, trigger_id, record_id):
try:
response = webhook_request(webhook, ci_dict).text
def _update_old_attr_value(record_id, ci_dict):
attr_history = AttributeHistory.get_by(record_id=record_id, to_dict=False)
attr_dict = dict()
for attr_h in attr_history:
attr_dict['old_{}'.format(AttributeCache.get(attr_h.attr_id).name)] = attr_h.old
ci_dict.update({'old_{}'.format(k): ci_dict[k] for k in ci_dict})
ci_dict.update(attr_dict)
@classmethod
def _exec_webhook(cls, operate_type, webhook, ci_dict, trigger_id, trigger_name, record_id, ci_id=None, app=None):
app = app or current_app
with app.app_context():
if operate_type == OperateType.UPDATE:
cls._update_old_attr_value(record_id, ci_dict)
if ci_id is not None:
ci_dict = CIManager().get_ci_by_id_from_db(ci_id, need_children=False, use_master=False)
try:
response = webhook_request(webhook, ci_dict).text
is_ok = True
except Exception as e:
current_app.logger.warning("exec webhook failed: {}".format(e))
response = e
is_ok = False
CITriggerHistoryManager.add(operate_type,
record_id,
ci_dict.get('_id'),
trigger_id,
trigger_name,
is_ok=is_ok,
webhook=response)
return is_ok
@classmethod
def _exec_notify(cls, operate_type, notify, ci_dict, trigger_id, trigger_name, record_id, ci_id=None, app=None):
app = app or current_app
with app.app_context():
if ci_id is not None:
ci_dict = CIManager().get_ci_by_id_from_db(ci_id, need_children=False, use_master=False)
if operate_type == OperateType.UPDATE:
cls._update_old_attr_value(record_id, ci_dict)
is_ok = True
except Exception as e:
current_app.logger.warning("exec webhook failed: {}".format(e))
response = e
is_ok = False
response = ''
for method in (notify.get('method') or []):
try:
res = notify_send(notify.get('subject'), notify.get('body'), [method],
notify.get('tos'), ci_dict)
response = "{}\n{}".format(response, res)
except Exception as e:
current_app.logger.warning("send notify failed: {}".format(e))
response = "{}\n{}".format(response, e)
is_ok = False
CITriggerHistoryManager.add(operate_type,
record_id,
ci_dict.get('_id'),
trigger_id,
is_ok=is_ok,
webhook=response)
CITriggerHistoryManager.add(operate_type,
record_id,
ci_dict.get('_id'),
trigger_id,
trigger_name,
is_ok=is_ok,
notify=response.strip())
return is_ok
@staticmethod
def _exec_notify(operate_type, notify, ci_dict, trigger_id, record_id, ci_id=None):
if ci_id is not None:
ci_dict = CIManager().get_ci_by_id_from_db(ci_id, need_children=False, use_master=False)
try:
response = notify_send(notify.get('subject'), notify.get('body'), notify.get('tos'), ci_dict)
is_ok = True
except Exception as e:
current_app.logger.warning("send notify failed: {}".format(e))
response = e
is_ok = False
CITriggerHistoryManager.add(operate_type,
record_id,
ci_dict.get('_id'),
trigger_id,
is_ok=is_ok,
notify=response)
return is_ok
return is_ok
@staticmethod
def ci_filter(ci_id, other_filter):
from api.lib.cmdb.search import SearchError
from api.lib.cmdb.search.ci import search
query = "_id:{},{}".format(ci_id, other_filter)
query = "{},_id:{}".format(other_filter, ci_id)
try:
_, _, _, _, numfound, _ = search(query).search()
@@ -973,28 +1018,40 @@ class CITriggerManager(object):
triggers = cls.get(type_id) or []
for trigger in triggers:
if not trigger.option.get('enable'):
option = trigger['option']
if not option.get('enable'):
continue
if trigger.option.get('filter') and not cls.ci_filter(ci_dict.get('_id'), trigger.option['filter']):
if option.get('filter') and not cls.ci_filter(ci_dict.get('_id'), option['filter']):
continue
if trigger.option.get('attr_ids') and isinstance(trigger.option['attr_ids'], list):
if not (set(trigger.option['attr_ids']) &
if option.get('attr_ids') and isinstance(option['attr_ids'], list):
if not (set(option['attr_ids']) &
set([i.attr_id for i in AttributeHistory.get_by(record_id=record_id, to_dict=False)])):
continue
if trigger.option.get('action') == operate_type:
if trigger.option.get('webhooks'):
cls._exec_webhook(operate_type, trigger.option['webhooks'], ci_dict, trigger.id, record_id)
elif trigger.option.get('notifies'):
cls._exec_notify(operate_type, trigger.option['notifies'], ci_dict, trigger.id, record_id)
if option.get('action') == operate_type:
cls.fire_by_trigger(trigger, operate_type, ci_dict, record_id)
@classmethod
def fire_by_trigger(cls, trigger, operate_type, ci_dict, record_id=None):
option = trigger['option']
if option.get('webhooks'):
cls._exec_webhook(operate_type, option['webhooks'], ci_dict, trigger['id'],
option.get('name'), record_id)
elif option.get('notifies'):
cls._exec_notify(operate_type, option['notifies'], ci_dict, trigger['id'],
option.get('name'), record_id)
@classmethod
def waiting_cis(cls, trigger):
now = datetime.datetime.today()
delta_time = datetime.timedelta(days=(trigger.option.get('before_days', 0) or 0))
config = trigger.option.get('notifies') or {}
delta_time = datetime.timedelta(days=(config.get('before_days', 0) or 0))
attr = AttributeCache.get(trigger.attr_id)
@@ -1022,10 +1079,17 @@ class CITriggerManager(object):
:param ci:
:return:
"""
if (trigger.notify.get('notify_at') == datetime.datetime.now().strftime("%H:%M") or
not trigger.option.get('notify_at')):
threading.Thread(target=cls._exec_notify, args=(
None, trigger.option['notifies'], None, trigger.id, None, ci.id)).start()
if (trigger.option.get('notifies', {}).get('notify_at') == datetime.datetime.now().strftime("%H:%M") or
not trigger.option.get('notifies', {}).get('notify_at')):
if trigger.option.get('webhooks'):
threading.Thread(target=cls._exec_webhook, args=(
None, trigger.option['webhooks'], None, trigger.id, trigger.option.get('name'), None, ci.ci_id,
current_app._get_current_object())).start()
elif trigger.option.get('notifies'):
threading.Thread(target=cls._exec_notify, args=(
None, trigger.option['notifies'], None, trigger.id, trigger.option.get('name'), None, ci.ci_id,
current_app._get_current_object())).start()
return True

4
cmdb-api/api/lib/cmdb/ci_type.py
View File

@@ -1187,12 +1187,12 @@ class CITypeTriggerManager(object):
return trigger.to_dict()
@staticmethod
def update(_id, option):
def update(_id, attr_id, option):
existed = (CITypeTrigger.get_by_id(_id) or
abort(404, ErrFormat.ci_type_trigger_not_found.format("id={}".format(_id))))
existed2 = existed.to_dict()
new = existed.update(option=option)
new = existed.update(attr_id=attr_id or None, option=option, filter_none=False)
CITypeHistoryManager.add(CITypeOperateType.UPDATE_TRIGGER,
existed.type_id,

19
cmdb-api/api/lib/cmdb/history.py
View File

@@ -293,13 +293,13 @@ class CITriggerHistoryManager(object):
@staticmethod
def get(page, page_size, type_id=None, trigger_id=None, operate_type=None):
query = CITriggerHistory.get_by(only_query=True)
if type_id is not None:
if type_id:
query = query.filter(CITriggerHistory.type_id == type_id)
if trigger_id:
query = query.filter(CITriggerHistory.trigger_id == trigger_id)
if operate_type is not None:
if operate_type:
query = query.filter(CITriggerHistory.operate_type == operate_type)
numfound = query.count()
@@ -316,22 +316,22 @@ class CITriggerHistoryManager(object):
@staticmethod
def get_by_ci_id(ci_id):
res = db.session.query(CITriggerHistory, CITypeTrigger, OperationRecord).join(
CITypeTrigger, CITypeTrigger.id == CITriggerHistory.trigger_id).join(
OperationRecord, OperationRecord.id == CITriggerHistory.record_id).filter(
res = db.session.query(CITriggerHistory, CITypeTrigger).join(
CITypeTrigger, CITypeTrigger.id == CITriggerHistory.trigger_id).filter(
CITriggerHistory.ci_id == ci_id).order_by(CITriggerHistory.id.desc())
result = []
id2trigger = dict()
for i in res:
hist = i.CITriggerHistory
record = i.OperationRecord
item = dict(is_ok=hist.is_ok,
operate_type=hist.operate_type,
notify=hist.notify,
trigger_id=hist.trigger_id,
trigger_name=hist.trigger_name,
webhook=hist.webhook,
created_at=record.created_at.strftime('%Y-%m-%d %H:%M:%S'),
record_id=record.id,
created_at=hist.created_at.strftime('%Y-%m-%d %H:%M:%S'),
record_id=hist.record_id,
hid=hist.id
)
if i.CITypeTrigger.id not in id2trigger:
@@ -342,12 +342,13 @@ class CITriggerHistoryManager(object):
return dict(items=result, id2trigger=id2trigger)
@staticmethod
def add(operate_type, record_id, ci_id, trigger_id, is_ok=False, notify=None, webhook=None):
def add(operate_type, record_id, ci_id, trigger_id, trigger_name, is_ok=False, notify=None, webhook=None):
CITriggerHistory.create(operate_type=operate_type,
record_id=record_id,
ci_id=ci_id,
trigger_id=trigger_id,
trigger_name=trigger_name,
is_ok=is_ok,
notify=notify,
webhook=webhook)

20
cmdb-api/api/lib/notify.py
View File

@@ -5,23 +5,32 @@ import json
import requests
from flask import current_app
from jinja2 import Template
from markdownify import markdownify as md
from api.lib.mail import send_mail
def _request_messenger(subject, body, tos, sender):
def _request_messenger(subject, body, tos, sender, payload):
params = dict(sender=sender, title=subject,
tos=[to[sender] for to in tos if to.get(sender)])
if not params['tos']:
raise Exception("no receivers")
params['tos'] = [Template(i).render(payload) for i in params['tos'] if i.strip()]
if sender == "email":
params['msgtype'] = 'text/html'
params['content'] = body
else:
params['msgtype'] = 'text'
params['content'] = json.dumps(dict(content=subject or body))
params['msgtype'] = 'markdown'
try:
content = md("{}\n{}".format(subject or '', body or ''))
except Exception as e:
current_app.logger.warning("html2markdown failed: {}".format(e))
content = "{}\n{}".format(subject or '', body or '')
params['content'] = json.dumps(dict(content=content))
resp = requests.post(current_app.config.get('MESSENGER_URL'), json=params)
if resp.status_code != 200:
@@ -32,14 +41,15 @@ def _request_messenger(subject, body, tos, sender):
def notify_send(subject, body, methods, tos, payload=None):
payload = payload or {}
payload = {k: v or '' for k, v in payload.items()}
subject = Template(subject).render(payload)
body = Template(body).render(payload)
res = ''
for method in methods:
if method == "email" and not current_app.config.get('USE_MESSENGER', True):
send_mail(None, [to.get('email') for to in tos], subject, body)
send_mail(None, [Template(to.get('email')).render(payload) for to in tos], subject, body)
res += _request_messenger(subject, body, tos, method) + "\n"
res += (_request_messenger(subject, body, tos, method, payload) + "\n")
return res

6
cmdb-api/api/lib/webhook.py
View File

@@ -82,12 +82,16 @@ def webhook_request(webhook, payload):
"""
assert webhook.get('url') is not None
payload = {k: v or '' for k, v in payload.items()}
url = Template(webhook['url']).render(payload)
params = webhook.get('parameters') or None
if isinstance(params, dict):
params = json.loads(Template(json.dumps(params)).render(payload))
headers = json.loads(Template(json.dumps(webhook.get('headers') or {})).render(payload))
data = Template(json.dumps(webhook.get('body', ''))).render(payload)
auth = _wrap_auth(**webhook.get('authorization', {}))
@@ -99,7 +103,7 @@ def webhook_request(webhook, payload):
return request(
url,
params=params,
headers=webhook.get('headers') or None,
headers=headers or None,
data=data,
auth=auth
)

1
cmdb-api/api/models/cmdb.py
View File

@@ -139,6 +139,7 @@ class CITriggerHistory(Model):
record_id = db.Column(db.Integer, db.ForeignKey("c_records.id"))
ci_id = db.Column(db.Integer, index=True, nullable=False)
trigger_id = db.Column(db.Integer, db.ForeignKey("c_c_t_t.id"))
trigger_name = db.Column(db.String(64))
is_ok = db.Column(db.Boolean, default=False)
notify = db.Column(db.Text)
webhook = db.Column(db.Text)

13
cmdb-api/api/tasks/cmdb.py
View File

@@ -65,10 +65,7 @@ def batch_ci_cache(ci_ids, ): # only for attribute change index
@celery.task(name="cmdb.ci_delete", queue=CMDB_QUEUE)
def ci_delete(ci_dict, operate_type, record_id):
from api.lib.cmdb.ci import CITriggerManager
ci_id = ci_dict.get('_id')
def ci_delete(ci_id):
current_app.logger.info(ci_id)
if current_app.config.get("USE_ES"):
@@ -78,10 +75,16 @@ def ci_delete(ci_dict, operate_type, record_id):
current_app.logger.info("{0} delete..........".format(ci_id))
@celery.task(name="cmdb.ci_delete_trigger", queue=CMDB_QUEUE)
def ci_delete_trigger(trigger, operate_type, ci_dict):
current_app.logger.info('delete ci {} trigger'.format(ci_dict['_id']))
from api.lib.cmdb.ci import CITriggerManager
current_app.test_request_context().push()
login_user(UserCache.get('worker'))
CITriggerManager.fire(operate_type, ci_dict, record_id)
CITriggerManager.fire_by_trigger(trigger, operate_type, ci_dict)
@celery.task(name="cmdb.ci_relation_cache", queue=CMDB_QUEUE)

5
cmdb-api/requirements.txt
View File

@@ -37,11 +37,12 @@ PyYAML==6.0
redis==4.6.0
requests==2.31.0
requests_oauthlib==1.3.1
six==1.12.0
markdownify==0.11.6
six==1.16.0
SQLAlchemy==1.4.49
supervisor==4.0.3
timeout-decorator==0.5.0
toposort==1.10
treelib==1.6.1
Werkzeug==2.3.6
WTForms==3.0.0
WTForms==3.0.0

4
cmdb-api/settings.example.py
View File

@@ -94,3 +94,7 @@ ES_HOST = '127.0.0.1'
USE_ES = False
BOOL_TRUE = ['true', 'TRUE', 'True', True, '1', 1, "Yes", "YES", "yes", 'Y', 'y']
# # messenger
USE_MESSENGER = True
MESSENGER_URL = "http://{messenger_url}/v1/message"