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 10KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292
  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. class ConductorEndpoint(object):
  35. def __init__(self, ragent):
  36. transport = oslo_messaging.get_transport(cfg.CONF)
  37. self.target = oslo_messaging.Target()
  38. self.wamp_agent_client = oslo_messaging.RPCClient(transport,
  39. self.target)
  40. self.wamp_agent_client.prepare(timeout=10)
  41. self.ragent = ragent
  42. def echo(self, ctx, data):
  43. LOG.info("ECHO: %s" % data)
  44. return data
  45. def registration(self, ctx, code, session_num):
  46. LOG.debug('Received registration from %s with session %s',
  47. code, session_num)
  48. try:
  49. board = objects.Board.get_by_code(ctx, code)
  50. except Exception as exc:
  51. msg = exc.message % {'board': code}
  52. LOG.error(msg)
  53. return wm.WampError(msg).serialize()
  54. if not board.status == states.REGISTERED:
  55. msg = "board with code %(board)s cannot " \
  56. "be registered again." % {'board': code}
  57. LOG.error(msg)
  58. return wm.WampError(msg).serialize()
  59. try:
  60. old_ses = objects.SessionWP(ctx)
  61. old_ses = old_ses.get_session_by_board_uuid(ctx, board.uuid,
  62. valid=True)
  63. old_ses.valid = False
  64. old_ses.save()
  65. except Exception:
  66. LOG.debug('valid session for %s not found', board.uuid)
  67. session_data = {'board_id': board.id,
  68. 'board_uuid': board.uuid,
  69. 'session_id': session_num}
  70. session = objects.SessionWP(ctx, **session_data)
  71. session.create()
  72. board.agent = get_best_agent(ctx)
  73. agent = objects.WampAgent.get_by_hostname(ctx, board.agent)
  74. prov = Provisioner(board)
  75. prov.conf_registration_agent(self.ragent.wsurl)
  76. prov.conf_main_agent(agent.wsurl)
  77. loc = objects.Location.list_by_board_uuid(ctx, board.uuid)[0]
  78. prov.conf_location(loc)
  79. board.config = prov.get_config()
  80. board.status = states.OFFLINE
  81. board.save()
  82. LOG.debug('sending this conf %s', board.config)
  83. wmessage = wm.WampSuccess(board.config)
  84. return wmessage.serialize()
  85. def destroy_board(self, ctx, board_id):
  86. LOG.info('Destroying board with id %s',
  87. board_id)
  88. board = objects.Board.get_by_uuid(ctx, board_id)
  89. result = None
  90. if board.is_online():
  91. prov = Provisioner()
  92. prov.conf_clean()
  93. p = prov.get_config()
  94. LOG.debug('sending this conf %s', p)
  95. try:
  96. result = self.execute_on_board(ctx,
  97. board_id,
  98. 'destroyBoard',
  99. (p,))
  100. except exception:
  101. return exception
  102. board.destroy()
  103. if result:
  104. LOG.debug(result)
  105. return result
  106. return
  107. def update_board(self, ctx, board_obj):
  108. board = serializer.deserialize_entity(ctx, board_obj)
  109. LOG.debug('Updating board %s', board.name)
  110. board.save()
  111. return serializer.serialize_entity(ctx, board)
  112. def create_board(self, ctx, board_obj, location_obj):
  113. new_board = serializer.deserialize_entity(ctx, board_obj)
  114. LOG.debug('Creating board %s',
  115. new_board.name)
  116. new_location = serializer.deserialize_entity(ctx, location_obj)
  117. new_board.create()
  118. new_location.board_id = new_board.id
  119. new_location.create()
  120. return serializer.serialize_entity(ctx, new_board)
  121. def execute_on_board(self, ctx, board_uuid, wamp_rpc_call, wamp_rpc_args):
  122. LOG.debug('Executing \"%s\" on the board: %s',
  123. wamp_rpc_call, board_uuid)
  124. board = objects.Board.get_by_uuid(ctx, board_uuid)
  125. s4t_topic = 's4t_invoke_wamp'
  126. full_topic = board.agent + '.' + s4t_topic
  127. self.target.topic = full_topic
  128. full_wamp_call = 'iotronic.' + board.uuid + "." + wamp_rpc_call
  129. # check the session; it rise an excpetion if session miss
  130. if not board.is_online():
  131. raise exception.BoardNotConnected(board=board.uuid)
  132. res = self.wamp_agent_client.call(ctx, full_topic,
  133. wamp_rpc_call=full_wamp_call,
  134. data=wamp_rpc_args)
  135. res = wm.deserialize(res)
  136. if res.result == wm.SUCCESS:
  137. return res.message
  138. elif res.result == wm.WARNING:
  139. LOG.warning('Warning in the execution of %s on %s', wamp_rpc_call,
  140. board_uuid)
  141. return res.message
  142. elif res.result == wm.ERROR:
  143. LOG.error('Error in the execution of %s on %s: %s', wamp_rpc_call,
  144. board_uuid, res.message)
  145. raise exception.ErrorExecutionOnBoard(call=wamp_rpc_call,
  146. board=board.uuid,
  147. error=res.message)
  148. def destroy_plugin(self, ctx, plugin_id):
  149. LOG.info('Destroying plugin with id %s',
  150. plugin_id)
  151. plugin = objects.Plugin.get_by_uuid(ctx, plugin_id)
  152. plugin.destroy()
  153. return
  154. def update_plugin(self, ctx, plugin_obj):
  155. plugin = serializer.deserialize_entity(ctx, plugin_obj)
  156. LOG.debug('Updating plugin %s', plugin.name)
  157. plugin.save()
  158. return serializer.serialize_entity(ctx, plugin)
  159. def create_plugin(self, ctx, plugin_obj):
  160. new_plugin = serializer.deserialize_entity(ctx, plugin_obj)
  161. LOG.debug('Creating plugin %s',
  162. new_plugin.name)
  163. new_plugin.code = cpickle.dumps(new_plugin.code, 0)
  164. new_plugin.create()
  165. return serializer.serialize_entity(ctx, new_plugin)
  166. def inject_plugin(self, ctx, plugin_uuid, board_uuid, onboot):
  167. LOG.info('Injecting plugin with id %s into the board %s',
  168. plugin_uuid, board_uuid)
  169. plugin = objects.Plugin.get(ctx, plugin_uuid)
  170. try:
  171. result = self.execute_on_board(ctx,
  172. board_uuid,
  173. 'PluginInject',
  174. (plugin, onboot))
  175. except exception:
  176. return exception
  177. injection = None
  178. try:
  179. injection = objects.InjectionPlugin.get(ctx,
  180. board_uuid,
  181. plugin_uuid)
  182. except Exception:
  183. pass
  184. if injection:
  185. injection.status = 'updated'
  186. injection.save()
  187. else:
  188. inj_data = {
  189. 'board_uuid': board_uuid,
  190. 'plugin_uuid': plugin_uuid,
  191. 'onboot': onboot,
  192. 'status': 'injected'
  193. }
  194. injection = objects.InjectionPlugin(ctx, **inj_data)
  195. injection.create()
  196. LOG.debug(result)
  197. return result
  198. def remove_plugin(self, ctx, plugin_uuid, board_uuid):
  199. LOG.info('Removing plugin with id %s into the board %s',
  200. plugin_uuid, board_uuid)
  201. plugin = objects.Plugin.get_by_uuid(ctx, plugin_uuid)
  202. injection = objects.InjectionPlugin.get(ctx, board_uuid, plugin_uuid)
  203. try:
  204. result = self.execute_on_board(ctx, board_uuid, 'PluginRemove',
  205. (plugin.uuid,))
  206. except exception:
  207. return exception
  208. LOG.debug(result)
  209. injection.destroy()
  210. return result
  211. def action_plugin(self, ctx, plugin_uuid, board_uuid, action, params):
  212. LOG.info('Calling plugin with id %s into the board %s with params %s',
  213. plugin_uuid, board_uuid, params)
  214. plugin = objects.Plugin.get(ctx, plugin_uuid)
  215. objects.plugin.is_valid_action(action)
  216. try:
  217. if objects.plugin.want_params(action):
  218. result = self.execute_on_board(ctx, board_uuid, action,
  219. (plugin.uuid, params))
  220. else:
  221. result = self.execute_on_board(ctx, board_uuid, action,
  222. (plugin.uuid,))
  223. except exception:
  224. return exception
  225. LOG.debug(result)
  226. return result
  227. def create_service(self, ctx, service_obj):
  228. new_service = serializer.deserialize_entity(ctx, service_obj)
  229. LOG.debug('Creating service %s',
  230. new_service.name)
  231. new_service.create()
  232. return serializer.serialize_entity(ctx, new_service)
  233. def destroy_service(self, ctx, service_id):
  234. LOG.info('Destroying service with id %s',
  235. service_id)
  236. service = objects.Service.get_by_uuid(ctx, service_id)
  237. service.destroy()
  238. return
  239. def update_service(self, ctx, service_obj):
  240. service = serializer.deserialize_entity(ctx, service_obj)
  241. LOG.debug('Updating service %s', service.name)
  242. service.save()
  243. return serializer.serialize_entity(ctx, service)