Internet of Things resource management service for OpenStack clouds.
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.

endpoints.py 16KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437
  1. # Copyright 2017 MDSLAB - University of Messina
  2. # All Rights Reserved.
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  5. # not use this file except in compliance with the License. You may obtain
  6. # a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13. # License for the specific language governing permissions and limitations
  14. # under the License.
  15. import _pickle as cpickle
  16. from iotronic.common import exception
  17. from iotronic.common import states
  18. from iotronic.conductor.provisioner import Provisioner
  19. from iotronic import objects
  20. from iotronic.objects import base as objects_base
  21. from iotronic.wamp import wampmessage as wm
  22. from oslo_config import cfg
  23. from oslo_log import log as logging
  24. import oslo_messaging
  25. import random
  26. LOG = logging.getLogger(__name__)
  27. serializer = objects_base.IotronicObjectSerializer()
  28. def get_best_agent(ctx):
  29. agents = objects.WampAgent.list(ctx, filters={'online': True})
  30. LOG.debug('found %d Agent(s).', len(agents))
  31. agent = random.choice(agents)
  32. LOG.debug('Selected agent: %s', agent.hostname)
  33. return agent.hostname
  34. def random_public_port():
  35. return random.randint(6000, 7000)
  36. def manage_result(res, wamp_rpc_call, board_uuid):
  37. if res.result == wm.SUCCESS:
  38. return res.message
  39. elif res.result == wm.WARNING:
  40. LOG.warning('Warning in the execution of %s on %s', wamp_rpc_call,
  41. board_uuid)
  42. return res.message
  43. elif res.result == wm.ERROR:
  44. LOG.error('Error in the execution of %s on %s: %s', wamp_rpc_call,
  45. board_uuid, res.message)
  46. raise exception.ErrorExecutionOnBoard(call=wamp_rpc_call,
  47. board=board_uuid,
  48. error=res.message)
  49. return res.message
  50. class ConductorEndpoint(object):
  51. def __init__(self, ragent):
  52. transport = oslo_messaging.get_transport(cfg.CONF)
  53. self.target = oslo_messaging.Target()
  54. self.wamp_agent_client = oslo_messaging.RPCClient(transport,
  55. self.target)
  56. self.wamp_agent_client.prepare(timeout=10)
  57. self.ragent = ragent
  58. def echo(self, ctx, data):
  59. LOG.info("ECHO: %s" % data)
  60. return data
  61. def registration(self, ctx, code, session_num):
  62. LOG.debug('Received registration from %s with session %s',
  63. code, session_num)
  64. try:
  65. board = objects.Board.get_by_code(ctx, code)
  66. except Exception as exc:
  67. msg = exc.message % {'board': code}
  68. LOG.error(msg)
  69. return wm.WampError(msg).serialize()
  70. if not board.status == states.REGISTERED:
  71. msg = "board with code %(board)s cannot " \
  72. "be registered again." % {'board': code}
  73. LOG.error(msg)
  74. return wm.WampError(msg).serialize()
  75. try:
  76. old_ses = objects.SessionWP(ctx)
  77. old_ses = old_ses.get_session_by_board_uuid(ctx, board.uuid,
  78. valid=True)
  79. old_ses.valid = False
  80. old_ses.save()
  81. except Exception:
  82. LOG.debug('valid session for %s not found', board.uuid)
  83. session_data = {'board_id': board.id,
  84. 'board_uuid': board.uuid,
  85. 'session_id': session_num}
  86. session = objects.SessionWP(ctx, **session_data)
  87. session.create()
  88. board.agent = get_best_agent(ctx)
  89. agent = objects.WampAgent.get_by_hostname(ctx, board.agent)
  90. prov = Provisioner(board)
  91. prov.conf_registration_agent(self.ragent.wsurl)
  92. prov.conf_main_agent(agent.wsurl)
  93. loc = objects.Location.list_by_board_uuid(ctx, board.uuid)[0]
  94. prov.conf_location(loc)
  95. board.config = prov.get_config()
  96. board.status = states.OFFLINE
  97. board.save()
  98. LOG.debug('sending this conf %s', board.config)
  99. wmessage = wm.WampSuccess(board.config)
  100. return wmessage.serialize()
  101. def destroy_board(self, ctx, board_id):
  102. LOG.info('Destroying board with id %s',
  103. board_id)
  104. board = objects.Board.get_by_uuid(ctx, board_id)
  105. result = None
  106. if board.is_online():
  107. prov = Provisioner()
  108. prov.conf_clean()
  109. p = prov.get_config()
  110. LOG.debug('sending this conf %s', p)
  111. try:
  112. result = self.execute_on_board(ctx,
  113. board_id,
  114. 'destroyBoard',
  115. (p,))
  116. except exception:
  117. return exception
  118. board.destroy()
  119. if result:
  120. result = manage_result(result, 'destroyBoard', board_id)
  121. LOG.debug(result)
  122. return result
  123. return
  124. def update_board(self, ctx, board_obj):
  125. board = serializer.deserialize_entity(ctx, board_obj)
  126. LOG.debug('Updating board %s', board.name)
  127. board.save()
  128. return serializer.serialize_entity(ctx, board)
  129. def create_board(self, ctx, board_obj, location_obj):
  130. new_board = serializer.deserialize_entity(ctx, board_obj)
  131. LOG.debug('Creating board %s',
  132. new_board.name)
  133. new_location = serializer.deserialize_entity(ctx, location_obj)
  134. new_board.create()
  135. new_location.board_id = new_board.id
  136. new_location.create()
  137. return serializer.serialize_entity(ctx, new_board)
  138. def execute_on_board(self, ctx, board_uuid, wamp_rpc_call, wamp_rpc_args):
  139. LOG.debug('Executing \"%s\" on the board: %s',
  140. wamp_rpc_call, board_uuid)
  141. board = objects.Board.get_by_uuid(ctx, board_uuid)
  142. s4t_topic = 's4t_invoke_wamp'
  143. full_topic = board.agent + '.' + s4t_topic
  144. self.target.topic = full_topic
  145. full_wamp_call = 'iotronic.' + board.uuid + "." + wamp_rpc_call
  146. # check the session; it rise an excpetion if session miss
  147. if not board.is_online():
  148. raise exception.BoardNotConnected(board=board.uuid)
  149. res = self.wamp_agent_client.call(ctx, full_topic,
  150. wamp_rpc_call=full_wamp_call,
  151. data=wamp_rpc_args)
  152. res = wm.deserialize(res)
  153. return res
  154. def destroy_plugin(self, ctx, plugin_id):
  155. LOG.info('Destroying plugin with id %s',
  156. plugin_id)
  157. plugin = objects.Plugin.get_by_uuid(ctx, plugin_id)
  158. plugin.destroy()
  159. return
  160. def update_plugin(self, ctx, plugin_obj):
  161. plugin = serializer.deserialize_entity(ctx, plugin_obj)
  162. LOG.debug('Updating plugin %s', plugin.name)
  163. plugin.save()
  164. return serializer.serialize_entity(ctx, plugin)
  165. def create_plugin(self, ctx, plugin_obj):
  166. new_plugin = serializer.deserialize_entity(ctx, plugin_obj)
  167. LOG.debug('Creating plugin %s',
  168. new_plugin.name)
  169. new_plugin.code = cpickle.dumps(new_plugin.code, 0)
  170. new_plugin.create()
  171. return serializer.serialize_entity(ctx, new_plugin)
  172. def inject_plugin(self, ctx, plugin_uuid, board_uuid, onboot):
  173. LOG.info('Injecting plugin with id %s into the board %s',
  174. plugin_uuid, board_uuid)
  175. plugin = objects.Plugin.get(ctx, plugin_uuid)
  176. try:
  177. result = self.execute_on_board(ctx,
  178. board_uuid,
  179. 'PluginInject',
  180. (plugin, onboot))
  181. except exception:
  182. return exception
  183. injection = None
  184. try:
  185. injection = objects.InjectionPlugin.get(ctx,
  186. board_uuid,
  187. plugin_uuid)
  188. except Exception:
  189. pass
  190. if injection:
  191. injection.status = 'updated'
  192. injection.save()
  193. else:
  194. inj_data = {
  195. 'board_uuid': board_uuid,
  196. 'plugin_uuid': plugin_uuid,
  197. 'onboot': onboot,
  198. 'status': 'injected'
  199. }
  200. injection = objects.InjectionPlugin(ctx, **inj_data)
  201. injection.create()
  202. result = manage_result(result, 'PluginInject', board_uuid)
  203. LOG.debug(result)
  204. return result
  205. def remove_plugin(self, ctx, plugin_uuid, board_uuid):
  206. LOG.info('Removing plugin with id %s into the board %s',
  207. plugin_uuid, board_uuid)
  208. plugin = objects.Plugin.get_by_uuid(ctx, plugin_uuid)
  209. injection = objects.InjectionPlugin.get(ctx, board_uuid, plugin_uuid)
  210. try:
  211. result = self.execute_on_board(ctx, board_uuid, 'PluginRemove',
  212. (plugin.uuid,))
  213. except exception:
  214. return exception
  215. result = manage_result(result, 'PluginRemove', board_uuid)
  216. LOG.debug(result)
  217. injection.destroy()
  218. return result
  219. def action_plugin(self, ctx, plugin_uuid, board_uuid, action, params):
  220. LOG.info('Calling plugin with id %s into the board %s with params %s',
  221. plugin_uuid, board_uuid, params)
  222. plugin = objects.Plugin.get(ctx, plugin_uuid)
  223. objects.plugin.is_valid_action(action)
  224. try:
  225. if objects.plugin.want_params(action):
  226. result = self.execute_on_board(ctx, board_uuid, action,
  227. (plugin.uuid, params))
  228. else:
  229. result = self.execute_on_board(ctx, board_uuid, action,
  230. (plugin.uuid,))
  231. except exception:
  232. return exception
  233. result = manage_result(result, action, board_uuid)
  234. LOG.debug(result)
  235. return result
  236. def create_service(self, ctx, service_obj):
  237. new_service = serializer.deserialize_entity(ctx, service_obj)
  238. LOG.debug('Creating service %s',
  239. new_service.name)
  240. new_service.create()
  241. return serializer.serialize_entity(ctx, new_service)
  242. def destroy_service(self, ctx, service_id):
  243. LOG.info('Destroying service with id %s',
  244. service_id)
  245. service = objects.Service.get_by_uuid(ctx, service_id)
  246. service.destroy()
  247. return
  248. def update_service(self, ctx, service_obj):
  249. service = serializer.deserialize_entity(ctx, service_obj)
  250. LOG.debug('Updating service %s', service.name)
  251. service.save()
  252. return serializer.serialize_entity(ctx, service)
  253. def action_service(self, ctx, service_uuid, board_uuid, action):
  254. service = objects.Service.get(ctx, service_uuid)
  255. objects.service.is_valid_action(action)
  256. if action == "ServiceEnable":
  257. LOG.info('Enabling service with id %s into the board %s',
  258. service_uuid, board_uuid)
  259. try:
  260. objects.ExposedService.get(ctx,
  261. board_uuid,
  262. service_uuid)
  263. return exception.ServiceAlreadyExposed(uuid=service_uuid)
  264. except Exception:
  265. name = service.name
  266. public_port = random_public_port()
  267. port = service.port
  268. res = self.execute_on_board(ctx, board_uuid, action,
  269. (name, public_port, port))
  270. if res.result == wm.SUCCESS:
  271. pid = res.message[0]
  272. exp_data = {
  273. 'board_uuid': board_uuid,
  274. 'service_uuid': service_uuid,
  275. 'public_port': public_port,
  276. 'pid': pid,
  277. }
  278. exposed = objects.ExposedService(ctx, **exp_data)
  279. exposed.create()
  280. res.message = res.message[1]
  281. elif res.result == wm.ERROR:
  282. LOG.error('Error in the execution of %s on %s: %s',
  283. action,
  284. board_uuid, res.message)
  285. raise exception.ErrorExecutionOnBoard(call=action,
  286. board=board_uuid,
  287. error=res.message)
  288. LOG.debug(res.message)
  289. return res.message
  290. elif action == "ServiceDisable":
  291. LOG.info('Disabling service with id %s into the board %s',
  292. service_uuid, board_uuid)
  293. exposed = objects.ExposedService.get(ctx,
  294. board_uuid,
  295. service_uuid)
  296. res = self.execute_on_board(ctx, board_uuid, action,
  297. (service.name, exposed.pid))
  298. result = manage_result(res, action, board_uuid)
  299. LOG.debug(res.message)
  300. exposed.destroy()
  301. return result
  302. elif action == "ServiceRestore":
  303. LOG.info('Restoring service with id %s into the board %s',
  304. service_uuid, board_uuid)
  305. exposed = objects.ExposedService.get(ctx, board_uuid,
  306. service_uuid)
  307. res = self.execute_on_board(ctx, board_uuid, action,
  308. (service.name, exposed.public_port,
  309. service.port, exposed.pid))
  310. if res.result == wm.SUCCESS:
  311. pid = res.message[0]
  312. exp_data = {
  313. 'id': exposed.id,
  314. 'board_uuid': board_uuid,
  315. 'service_uuid': service_uuid,
  316. 'public_port': exposed.public_port,
  317. 'pid': pid,
  318. }
  319. exposed = objects.ExposedService(ctx, **exp_data)
  320. exposed.save()
  321. res.message = res.message[1]
  322. elif res.result == wm.ERROR:
  323. LOG.error('Error in the execution of %s on %s: %s',
  324. action,
  325. board_uuid, res.message)
  326. raise exception.ErrorExecutionOnBoard(call=action,
  327. board=board_uuid,
  328. error=res.message)
  329. LOG.debug(res.message)
  330. return res.message
  331. def restore_services_on_board(self, ctx, board_uuid):
  332. LOG.info('Restoring the services into the board %s',
  333. board_uuid)
  334. exposed_list = objects.ExposedService.get_by_board_uuid(ctx,
  335. board_uuid)
  336. # response = []
  337. for exposed in exposed_list:
  338. service = objects.Service.get_by_uuid(ctx, exposed.service_uuid)
  339. res = self.execute_on_board(ctx, board_uuid, "ServiceRestore",
  340. (service.name, exposed.public_port,
  341. service.port, exposed.pid))
  342. if res.result == wm.SUCCESS:
  343. pid = res.message[0]
  344. exp_data = {
  345. 'id': exposed.id,
  346. 'board_uuid': exposed.board_uuid,
  347. 'service_uuid': exposed.service_uuid,
  348. 'public_port': exposed.public_port,
  349. 'pid': pid,
  350. }
  351. exposed = objects.ExposedService(ctx, **exp_data)
  352. exposed.save()
  353. # response.append(exposed)
  354. elif res.result == wm.ERROR:
  355. LOG.error('Error in restoring %s on %s: %s',
  356. service.name,
  357. board_uuid, res.message)
  358. raise exception.ErrorExecutionOnBoard(call="ServiceRestore",
  359. board=board_uuid,
  360. error=res.message)
  361. return 0