Monasca Events REST API
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

346 lines
13KB

  1. # Copyright 2015 Hewlett-Packard
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  4. # not use this file except in compliance with the License. You may obtain
  5. # a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12. # License for the specific language governing permissions and limitations
  13. # under the License.
  14. import uuid
  15. import MySQLdb
  16. from oslo_log import log
  17. from oslo_utils import timeutils
  18. from monasca_events_api.common.repositories import constants
  19. from monasca_events_api.common.repositories import exceptions
  20. from monasca_events_api.common.repositories.mysql import mysql_repository
  21. from monasca_events_api.common.repositories import streams_repository as sdr
  22. LOG = log.getLogger(__name__)
  23. class StreamsRepository(mysql_repository.MySQLRepository,
  24. sdr.StreamsRepository):
  25. base_query = """
  26. select sd.id, sd.tenant_id, sd.name, sd.description,
  27. sd.select_by, sd.group_by, sd.fire_criteria, sd.expiration,
  28. sd.actions_enabled, sd.created_at,
  29. sd.updated_at, sd.deleted_at,
  30. saf.fire_actions, sae.expire_actions
  31. from stream_definition as sd
  32. left join (select stream_definition_id,
  33. group_concat(action_id) as fire_actions
  34. from stream_actions
  35. where action_type = 'FIRE'
  36. group by stream_definition_id) as saf
  37. on saf.stream_definition_id = sd.id
  38. left join (select stream_definition_id,
  39. group_concat(action_id) as expire_actions
  40. from stream_actions
  41. where action_type = 'EXPIRE'
  42. group by stream_definition_id) as sae
  43. on sae.stream_definition_id = sd.id
  44. """
  45. def __init__(self):
  46. super(StreamsRepository, self).__init__()
  47. @mysql_repository.mysql_try_catch_block
  48. def get_stream_definition(self, tenant_id, stream_definition_id):
  49. parms = [tenant_id, stream_definition_id]
  50. where_clause = """ where sd.tenant_id = %s
  51. and sd.id = %s
  52. and deleted_at is NULL """
  53. query = StreamsRepository.base_query + where_clause
  54. rows = self._execute_query(query, parms)
  55. if rows:
  56. return rows[0]
  57. else:
  58. raise exceptions.DoesNotExistException
  59. @mysql_repository.mysql_try_catch_block
  60. def get_stream_definitions(self, tenant_id, name, offset=None, limit=None):
  61. parms = [tenant_id]
  62. select_clause = StreamsRepository.base_query
  63. where_clause = " where sd.tenant_id = %s and deleted_at is NULL "
  64. if name:
  65. where_clause += " and sd.name = %s "
  66. parms.append(name.encode('utf8'))
  67. if offset is not None:
  68. order_by_clause = " order by sd.id, sd.created_at "
  69. where_clause += " and sd.id > %s "
  70. parms.append(offset.encode('utf8'))
  71. limit_clause = " limit %s "
  72. parms.append(constants.PAGE_LIMIT)
  73. else:
  74. order_by_clause = " order by sd.created_at "
  75. limit_clause = ""
  76. if limit:
  77. limit_clause = " limit %s"
  78. parms.append(int(limit))
  79. query = select_clause + where_clause + order_by_clause + limit_clause
  80. return self._execute_query(query, parms)
  81. @mysql_repository.mysql_try_catch_block
  82. def get_all_stream_definitions(self, offset=None, limit=None):
  83. parms = []
  84. select_clause = StreamsRepository.base_query
  85. where_clause = " where deleted_at is NULL "
  86. if offset is not None:
  87. order_by_clause = " order by sd.id, sd.created_at "
  88. where_clause += " and sd.id > %s "
  89. parms.append(offset.encode('utf8'))
  90. limit_clause = " limit %s "
  91. if limit is not None:
  92. parms.append(limit)
  93. else:
  94. parms.append(constants.PAGE_LIMIT)
  95. else:
  96. order_by_clause = " order by sd.created_at "
  97. limit_clause = ""
  98. query = select_clause + where_clause + order_by_clause + limit_clause
  99. return self._execute_query(query, parms)
  100. @mysql_repository.mysql_try_catch_block
  101. def delete_stream_definition(self, tenant_id, stream_definition_id):
  102. """Delete the stream definition.
  103. :param tenant_id:
  104. :param stream_definition_id:
  105. :returns True: -- if stream definition exists and was deleted.
  106. :returns False: -- if the stream definition does not exists.
  107. :raises RepositoryException:
  108. """
  109. cnxn, cursor = self._get_cnxn_cursor_tuple()
  110. with cnxn:
  111. cursor.execute("""delete from stream_definition
  112. where tenant_id = %s and id = %s""",
  113. [tenant_id, stream_definition_id])
  114. if cursor.rowcount < 1:
  115. return False
  116. return True
  117. @mysql_repository.mysql_try_catch_block
  118. def create_stream_definition(self,
  119. tenant_id,
  120. name,
  121. description,
  122. select,
  123. group_by,
  124. fire_criteria,
  125. expiration,
  126. fire_actions,
  127. expire_actions):
  128. cnxn, cursor = self._get_cnxn_cursor_tuple()
  129. with cnxn:
  130. now = timeutils.utcnow()
  131. stream_definition_id = str(uuid.uuid1())
  132. try:
  133. cursor.execute("""insert into stream_definition(
  134. id,
  135. tenant_id,
  136. name,
  137. description,
  138. select_by,
  139. group_by,
  140. fire_criteria,
  141. expiration,
  142. created_at,
  143. updated_at)
  144. values (%s, %s, %s, %s, %s, %s, %s, %s, %s,
  145. %s)""", (
  146. stream_definition_id, tenant_id, name.encode('utf8'),
  147. description.encode('utf8'), select.encode('utf8'),
  148. group_by.encode('utf8'), fire_criteria.encode('utf8'),
  149. expiration, now, now))
  150. except MySQLdb.IntegrityError as e:
  151. code, msg = e
  152. if code == 1062:
  153. raise exceptions.AlreadyExistsException(
  154. 'Stream Definition already '
  155. 'exists for tenant_id: {0} name: {1}'.format(
  156. tenant_id, name.encode('utf8')))
  157. else:
  158. raise e
  159. self._insert_into_stream_actions(cursor, stream_definition_id,
  160. fire_actions, u"FIRE")
  161. self._insert_into_stream_actions(cursor, stream_definition_id,
  162. expire_actions,
  163. u"EXPIRE")
  164. return stream_definition_id
  165. @mysql_repository.mysql_try_catch_block
  166. def patch_stream_definition(self, tenant_id, stream_definition_id, name, description, select, group_by,
  167. fire_criteria, expiration, fire_actions, expire_actions):
  168. cnxn, cursor = self._get_cnxn_cursor_tuple()
  169. with cnxn:
  170. # Get the original alarm definition from the DB
  171. parms = [tenant_id, stream_definition_id]
  172. where_clause = """ where sd.tenant_id = %s
  173. and sd.id = %s"""
  174. query = StreamsRepository.base_query + where_clause
  175. cursor.execute(query, parms)
  176. if cursor.rowcount < 1:
  177. raise exceptions.DoesNotExistException
  178. original_definition = cursor.fetchall()[0]
  179. # Update that stream definition in the database
  180. patch_query = """
  181. update stream_definition
  182. set name = %s,
  183. description = %s,
  184. select_by = %s,
  185. group_by = %s,
  186. fire_criteria = %s,
  187. expiration = %s,
  188. updated_at = %s
  189. where tenant_id = %s and id = %s"""
  190. if name is None:
  191. name = original_definition['name']
  192. if description is None:
  193. description = original_definition['description']
  194. if select is None:
  195. select = original_definition['select_by']
  196. if select != original_definition['select_by']:
  197. msg = "select_by must not change".encode('utf8')
  198. raise exceptions.InvalidUpdateException(msg)
  199. if group_by is None:
  200. group_by = original_definition['group_by']
  201. if group_by != original_definition['group_by']:
  202. msg = "group_by must not change".encode('utf8')
  203. raise exceptions.InvalidUpdateException(msg)
  204. if fire_criteria is None:
  205. fire_criteria = original_definition['fire_criteria']
  206. if expiration is None:
  207. expiration = original_definition['expiration']
  208. now = timeutils.utcnow()
  209. update_parms = [
  210. name,
  211. description,
  212. select,
  213. group_by,
  214. fire_criteria,
  215. expiration,
  216. now,
  217. tenant_id,
  218. stream_definition_id]
  219. cursor.execute(patch_query, update_parms)
  220. # Update the fire and expire actions in the database if defined
  221. if fire_actions is not None:
  222. self._delete_stream_actions(cursor, stream_definition_id,
  223. u'FIRE')
  224. if expire_actions is not None:
  225. self._delete_stream_actions(cursor, stream_definition_id,
  226. u'EXPIRE')
  227. self._insert_into_stream_actions(cursor, stream_definition_id,
  228. fire_actions,
  229. u"FIRE")
  230. self._insert_into_stream_actions(cursor, stream_definition_id,
  231. expire_actions,
  232. u"EXPIRE")
  233. # Get updated entry from mysql
  234. cursor.execute(query, parms)
  235. return cursor.fetchall()[0]
  236. def _delete_stream_actions(self, cursor, stream_definition_id, action_type):
  237. query = """
  238. delete
  239. from stream_actions
  240. where stream_definition_id = %s and action_type = %s
  241. """
  242. parms = [stream_definition_id, action_type.encode('utf8')]
  243. cursor.execute(query, parms)
  244. def _insert_into_stream_actions(self, cursor, stream_definition_id,
  245. actions, action_type):
  246. if actions is None:
  247. return
  248. for action in actions:
  249. cursor.execute(
  250. "select id,type from notification_method where id = %s",
  251. (action.encode('utf8'),))
  252. row = cursor.fetchone()
  253. if not row:
  254. raise exceptions.InvalidUpdateException(
  255. "Non-existent notification id {} submitted for {} "
  256. "notification action".format(action.encode('utf8'),
  257. action_type.encode('utf8')))
  258. else:
  259. if row['type'] == 'PAGERDUTY':
  260. raise exceptions.InvalidUpdateException(
  261. "PAGERDUTY action not supported for "
  262. "notification id {} submitted for {} "
  263. "notification action".format(
  264. action.encode('utf8'),
  265. action_type.encode('utf8')))
  266. cursor.execute("""insert into stream_actions(
  267. stream_definition_id,
  268. action_type,
  269. action_id)
  270. values(%s,%s,%s)""", (
  271. stream_definition_id, action_type.encode('utf8'),
  272. action.encode('utf8')))