src/emuvim/api/zerorpc/compute.py

   1 """
   2 Distributed Cloud Emulator (dcemulator)
   3 (c) 2015 by Manuel Peuster <manuel.peuster@upb.de>
   4 """
   5
   6 import logging
   7 import threading
   8 import zerorpc
   9
  10 import paramiko
  11 import ipaddress
  12
  13 logging.basicConfig(level=logging.INFO)
  14
  15
  16 class ZeroRpcApiEndpoint(object):
  17     """
  18     Simple API endpoint that offers a zerorpc-based
  19     interface. This interface will be used by the
  20     default command line client.
  21     It can be used as a reference to implement
  22     REST interfaces providing the same semantics,
  23     like e.g. OpenStack compute API.
  24     """
  25
  26     def __init__(self, listenip, port):
  27         self.dcs = {}
  28         self.ip = listenip
  29         self.port = port
  30         logging.debug("Created API endpoint %s(%s:%d)" % (
  31             self.__class__.__name__, self.ip, self.port))
  32
  33     def connectDatacenter(self, dc):
  34         self.dcs[dc.label] = dc
  35         logging.info("Connected DC(%s) to API endpoint %s(%s:%d)" % (
  36             dc.label, self.__class__.__name__, self.ip, self.port))
  37
  38     def start(self):
  39         thread = threading.Thread(target=self._api_server_thread, args=())
  40         thread.daemon = True
  41         thread.start()
  42         logging.debug("Started API endpoint %s(%s:%d)" % (
  43             self.__class__.__name__, self.ip, self.port))
  44
  45     def _api_server_thread(self):
  46         s = zerorpc.Server(MultiDatacenterApi(self.dcs))
  47         s.bind("tcp://%s:%d" % (self.ip, self.port))
  48         s.run()
  49
  50
  51 class MultiDatacenterApi(object):
  52     """
  53         Just pass through the corresponding request to the
  54         selected data center. Do not implement provisioning
  55         logic here because will will have multiple API
  56         endpoint implementations at the end.
  57     """
  58
  59     def __init__(self, dcs):
  60         self.dcs = dcs
  61
  62     def compute_action_start(self, dc_label, compute_name, image, network, command):
  63         """
  64         Start a new compute instance: A docker container (note: zerorpc does not support keyword arguments)
  65         :param dc_label: name of the DC
  66         :param compute_name: compute container name
  67         :param image: image name
  68         :param command: command to execute
  69         :param network: list of all interface of the vnf, with their parameters (id=id1,ip=x.x.x.x/x),...
  70         :return: networks list({"id":"input","ip": "10.0.0.254/8"}, {"id":"output","ip": "11.0.0.254/24"})
  71         """
  72         # TODO what to return UUID / given name / internal name ?
  73         logging.debug("RPC CALL: compute start")
  74         try:
  75             c = self.dcs.get(dc_label).startCompute(
  76                 compute_name, image=image, command=command, network=network)
  77             #return str(c.name)
  78             # return docker inspect dict
  79             return c.getStatus()
  80         except Exception as ex:
  81             logging.exception("RPC error.")
  82             return ex.message
  83
  84     def compute_action_stop(self, dc_label, compute_name):
  85         logging.debug("RPC CALL: compute stop")
  86         try:
  87             return self.dcs.get(dc_label).stopCompute(compute_name)
  88         except Exception as ex:
  89             logging.exception("RPC error.")
  90             return ex.message
  91
  92     def compute_list(self, dc_label):
  93         logging.debug("RPC CALL: compute list")
  94         try:
  95             if dc_label is None:
  96                 # return list with all compute nodes in all DCs
  97                 all_containers = []
  98                 for dc in self.dcs.itervalues():
  99                     all_containers += dc.listCompute()
 100                 return [(c.name, c.getStatus())
 101                         for c in all_containers]
 102             else:
 103                 # return list of compute nodes for specified DC
 104                 return [(c.name, c.getStatus())
 105                         for c in self.dcs.get(dc_label).listCompute()]
 106         except Exception as ex:
 107             logging.exception("RPC error.")
 108             return ex.message
 109
 110     def compute_status(self, dc_label, compute_name):
 111         logging.debug("RPC CALL: compute status")
 112         try:
 113             return self.dcs.get(
 114                 dc_label).containers.get(compute_name).getStatus()
 115         except Exception as ex:
 116             logging.exception("RPC error.")
 117             return ex.message
 118
 119     def compute_profile(self, dc_label, compute_name, image, kwargs):
 120         # note: zerorpc does not support keyword arguments
 121
 122         ## VIM/dummy gatekeeper's tasks:
 123         # start vnf
 124         vnf_status = self.compute_action_start( dc_label, compute_name, image,
 125                                   kwargs.get('network'),
 126                                   kwargs.get('command'))
 127         # start traffic source (with fixed ip addres, no use for now...)
 128         psrc_status = self.compute_action_start( dc_label, 'psrc', 'profile_source', [{'id':'output'}], None)
 129         # link vnf to traffic source
 130         DCNetwork = self.dcs.get(dc_label).net
 131         DCNetwork.setChain('psrc', compute_name,
 132                            vnf_src_interface='output',
 133                            vnf_dst_interface=kwargs.get('input'),
 134                            cmd='add-flow', weight=None)
 135
 136         ## SSM/SP tasks:
 137         # start traffic generation
 138         for nw in psrc_status.get('network'):
 139             if nw.get('intf_name') == 'output':
 140                 psrc_output_ip = unicode(nw['ip'])
 141                 break
 142         dummy_iperf_server_ip = ipaddress.IPv4Address(psrc_output_ip) + 1
 143         iperf_cmd = 'iperf -c {0} -u -l18 -b10M -t1000 &'.format(dummy_iperf_server_ip)
 144
 145         psrc_mgmt_ip = psrc_status['docker_network']
 146         psrc_user='root'
 147         psrc_passw='root'
 148
 149         # use ssh login when starting command externally
 150         ret = self.dcs.get(dc_label).containers.get('psrc').pexec(iperf_cmd)
 151         logging.info(ret)
 152         self.dcs.get(dc_label).containers.get('psrc').monitor()
 153
 154         #ssh does not work when exectuted via zerorpc command
 155         #psrc_mgmt_ip = '172.17.0.3'
 156         #ssh = paramiko.SSHClient()
 157         #ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 158         #ssh.connect(psrc_mgmt_ip, username='steven', password='test')
 159         #ssh.connect(psrc_mgmt_ip, username='root', password='root')
 160
 161         #iperf_cmd = 'iperf -c {0} -u -l18 -b10M -t1000'.format(dummy_iperf_server_ip)
 162         #stdin, stdout, stderr = ssh.exec_command(iperf_cmd)
 163         # get monitor data and analyze
 164
 165         # create table
 166
 167         ## VIM/dummy gatekeeper's tasks:
 168         # remove vnfs and chain
 169
 170
 171     def datacenter_list(self):
 172         logging.debug("RPC CALL: datacenter list")
 173         try:
 174             return [d.getStatus() for d in self.dcs.itervalues()]
 175         except Exception as ex:
 176             logging.exception("RPC error.")
 177             return ex.message
 178
 179     def datacenter_status(self, dc_label):
 180         logging.debug("RPC CALL: datacenter status")
 181         try:
 182                 return self.dcs.get(dc_label).getStatus()
 183         except Exception as ex:
 184             logging.exception("RPC error.")
 185             return ex.message
 186
 187 '''
 188 if __name__ == "__main__":
 189     test = MultiDatacenterApi({})
 190     test.compute_profile('dc1','vnf1', 'image',network='',command='test',other='other')
 191 '''
 192