afonso/core-extra
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge branch 'enhancement/distributed-flask' of https://github.com/coreemu/core into enhancement/distributed-flask

Browse source
This commit is contained in:
bharnden 2019-10-17 12:13:26 -07:00
commit 8a3183c8b3
31 changed files with 692 additions and 1965 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
daemon/core/__init__.py
View file

@ -2,3 +2,6 @@ import logging.config
# setup default null handler
logging.getLogger(__name__).addHandler(logging.NullHandler())
# disable paramiko logging
logging.getLogger("paramiko").setLevel(logging.WARNING)

1147
daemon/core/api/tlv/broker.py
View file

File diff suppressed because it is too large Load diff

18
daemon/core/api/tlv/coreapi.py
View file

@ -15,7 +15,6 @@ from core.api.tlv import structutils
from core.emulator.enumerations import (
ConfigTlvs,
EventTlvs,
EventTypes,
ExceptionTlvs,
ExecuteTlvs,
FileTlvs,
@ -1017,20 +1016,3 @@ def str_to_list(value):
return None
return value.split("|")
def state_name(value):
"""
Helper to convert state number into state name using event types.
:param int value: state value to derive name from
:return: state name
:rtype: str
"""
try:
value = EventTypes(value).name
except ValueError:
value = "unknown"
return value

74
daemon/core/api/tlv/corehandlers.py
View file

@ -86,6 +86,7 @@ class CoreHandler(socketserver.BaseRequestHandler):
self.master = False
self.session = None
self.session_clients = {}
# core emulator
self.coreemu = server.coreemu
@ -138,8 +139,9 @@ class CoreHandler(socketserver.BaseRequestHandler):
if self.session:
# remove client from session broker and shutdown if there are no clients
self.remove_session_handlers()
self.session.broker.session_clients.remove(self)
if not self.session.broker.session_clients and not self.session.is_active():
clients = self.session_clients[self.session.id]
clients.remove(self)
if not clients and not self.session.is_active():
logging.info(
"no session clients left and not active, initiating shutdown"
)
@ -407,9 +409,7 @@ class CoreHandler(socketserver.BaseRequestHandler):
tlv_data += coreapi.CoreRegisterTlv.pack(
RegisterTlvs.EMULATION_SERVER.value, "core-daemon"
)
tlv_data += coreapi.CoreRegisterTlv.pack(
self.session.broker.config_type, self.session.broker.name
)
tlv_data += coreapi.CoreRegisterTlv.pack(RegisterTlvs.UTILITY.value, "broker")
tlv_data += coreapi.CoreRegisterTlv.pack(
self.session.location.config_type, self.session.location.name
)
@ -533,10 +533,6 @@ class CoreHandler(socketserver.BaseRequestHandler):
:param message: message to handle
:return: nothing
"""
if self.session and self.session.broker.handle_message(message):
logging.debug("message not being handled locally")
return
logging.debug(
"%s handling message:\n%s", threading.currentThread().getName(), message
)
@ -606,12 +602,11 @@ class CoreHandler(socketserver.BaseRequestHandler):
self.session = self.coreemu.create_session(port, master=False)
logging.debug("created new session for client: %s", self.session.id)
# TODO: hack to associate this handler with this sessions broker for broadcasting
# TODO: broker needs to be pulled out of session to the server/handler level
if self.master:
logging.debug("session set to master")
self.session.master = True
self.session.broker.session_clients.append(self)
clients = self.session_clients.setdefault(self.session.id, [])
clients.append(self)
# add handlers for various data
self.add_session_handlers()
@ -643,7 +638,8 @@ class CoreHandler(socketserver.BaseRequestHandler):
]:
continue
for client in self.session.broker.session_clients:
clients = self.session_clients[self.session.id]
for client in clients:
if client == self:
continue
@ -734,6 +730,7 @@ class CoreHandler(socketserver.BaseRequestHandler):
node_options.icon = message.get_tlv(NodeTlvs.ICON.value)
node_options.canvas = message.get_tlv(NodeTlvs.CANVAS.value)
node_options.opaque = message.get_tlv(NodeTlvs.OPAQUE.value)
node_options.emulation_server = message.get_tlv(NodeTlvs.EMULATION_SERVER.value)
services = message.get_tlv(NodeTlvs.SERVICES.value)
if services:
@ -1027,8 +1024,9 @@ class CoreHandler(socketserver.BaseRequestHandler):
# find the session containing this client and set the session to master
for _id in self.coreemu.sessions:
session = self.coreemu.sessions[_id]
if self in session.broker.session_clients:
clients = self.session_clients[_id]
if self in clients:
session = self.coreemu.sessions[_id]
logging.debug("setting session to master: %s", session.id)
session.master = True
break
@ -1077,7 +1075,7 @@ class CoreHandler(socketserver.BaseRequestHandler):
self.handle_config_location(message_type, config_data)
elif config_data.object == self.session.metadata.name:
replies = self.handle_config_metadata(message_type, config_data)
elif config_data.object == self.session.broker.name:
elif config_data.object == "broker":
self.handle_config_broker(message_type, config_data)
elif config_data.object == self.session.services.name:
replies = self.handle_config_services(message_type, config_data)
@ -1182,7 +1180,6 @@ class CoreHandler(socketserver.BaseRequestHandler):
def handle_config_broker(self, message_type, config_data):
if message_type not in [ConfigFlags.REQUEST, ConfigFlags.RESET]:
session_id = config_data.session
if not config_data.data_values:
logging.info("emulation server data missing")
else:
@ -1194,29 +1191,10 @@ class CoreHandler(socketserver.BaseRequestHandler):
for server in server_list:
server_items = server.split(":")
name, host, port = server_items[:3]
if host == "":
host = None
if port == "":
port = None
else:
port = int(port)
if session_id is not None:
# receive session ID and my IP from master
self.session.broker.session_id_master = int(
session_id.split("|")[0]
)
self.session.broker.myip = host
host = None
port = None
# this connects to the server immediately; maybe we should wait
# or spin off a new "client" thread here
self.session.broker.addserver(name, host, port)
self.session.broker.setupserver(name)
name, host, _ = server_items[:3]
self.session.distributed.add_server(name, host)
elif message_type == ConfigFlags.RESET:
self.session.distributed.shutdown()
def handle_config_services(self, message_type, config_data):
replies = []
@ -1842,11 +1820,9 @@ class CoreHandler(socketserver.BaseRequestHandler):
# remove client from session broker and shutdown if needed
self.remove_session_handlers()
self.session.broker.session_clients.remove(self)
if (
not self.session.broker.session_clients
and not self.session.is_active()
):
clients = self.session_clients[self.session.id]
clients.remove(self)
if not clients and not self.session.is_active():
self.coreemu.delete_session(self.session.id)
# set session to join
@ -1855,7 +1831,8 @@ class CoreHandler(socketserver.BaseRequestHandler):
# add client to session broker and set master if needed
if self.master:
self.session.master = True
self.session.broker.session_clients.append(self)
clients = self.session_clients.setdefault(self.session.id, [])
clients.append(self)
# add broadcast handlers
logging.info("adding session broadcast handlers")
@ -2106,6 +2083,7 @@ class CoreUdpHandler(CoreHandler):
logging.debug("session handling message: %s", session.session_id)
self.session = session
self.handle_message(message)
self.session.sdt.handle_distributed(message)
self.broadcast(message)
else:
logging.error(
@ -2130,6 +2108,7 @@ class CoreUdpHandler(CoreHandler):
if session or message.message_type == MessageTypes.REGISTER.value:
self.session = session
self.handle_message(message)
self.session.sdt.handle_distributed(message)
self.broadcast(message)
else:
logging.error(
@ -2140,7 +2119,8 @@ class CoreUdpHandler(CoreHandler):
if not isinstance(message, (coreapi.CoreNodeMessage, coreapi.CoreLinkMessage)):
return
for client in self.session.broker.session_clients:
clients = self.session_clients[self.session.id]
for client in clients:
try:
client.sendall(message.raw_message)
except IOError:

189
daemon/core/emane/emanemanager.py
View file

@ -2,14 +2,12 @@
emane.py: definition of an Emane class for implementing configuration control of an EMANE emulation.
"""
import copy
import logging
import os
import threading
from core import utils
from core.api.tlv import coreapi, dataconversion
from core.config import ConfigGroup, ConfigShim, Configuration, ModelManager
from core.config import ConfigGroup, Configuration, ModelManager
from core.emane import emanemanifest
from core.emane.bypass import EmaneBypassModel
from core.emane.commeffect import EmaneCommEffectModel
@ -18,15 +16,7 @@ from core.emane.ieee80211abg import EmaneIeee80211abgModel
from core.emane.nodes import EmaneNet
from core.emane.rfpipe import EmaneRfPipeModel
from core.emane.tdma import EmaneTdmaModel
from core.emulator import distributed
from core.emulator.enumerations import (
ConfigDataTypes,
ConfigFlags,
ConfigTlvs,
MessageFlags,
MessageTypes,
RegisterTlvs,
)
from core.emulator.enumerations import ConfigDataTypes, RegisterTlvs
from core.errors import CoreCommandError, CoreError
from core.xml import emanexml
@ -76,8 +66,6 @@ class EmaneManager(ModelManager):
self.session = session
self._emane_nets = {}
self._emane_node_lock = threading.Lock()
self._ifccounts = {}
self._ifccountslock = threading.Lock()
# port numbers are allocated from these counters
self.platformport = self.session.options.get_config_int(
"emane_platform_port", 8100
@ -92,7 +80,6 @@ class EmaneManager(ModelManager):
self.emane_config = EmaneGlobalModel(session)
self.set_configs(self.emane_config.default_values())
session.broker.handlers.add(self.handledistributed)
self.service = None
self.event_device = None
self.emane_check()
@ -155,9 +142,7 @@ class EmaneManager(ModelManager):
args = "emane --version"
emane_version = utils.check_cmd(args)
logging.info("using EMANE: %s", emane_version)
for server in self.session.servers:
conn = self.session.servers[server]
distributed.remote_cmd(conn, args)
self.session.distributed.execute(lambda x: x.remote_cmd(args))
# load default emane models
self.load_models(EMANE_MODELS)
@ -283,7 +268,6 @@ class EmaneManager(ModelManager):
return EmaneManager.NOT_NEEDED
# control network bridge required for EMANE 0.9.2
# - needs to be configured before checkdistributed() for distributed
# - needs to exist when eventservice binds to it (initeventservice)
if self.session.master:
otadev = self.get_config("otamanagerdevice")
@ -298,10 +282,9 @@ class EmaneManager(ModelManager):
)
return EmaneManager.NOT_READY
ctrlnet = self.session.add_remove_control_net(
self.session.add_remove_control_net(
net_index=netidx, remove=False, conf_required=False
)
self.distributedctrlnet(ctrlnet)
eventdev = self.get_config("eventservicedevice")
logging.debug("emane event service device: eventdev(%s)", eventdev)
if eventdev != otadev:
@ -314,18 +297,9 @@ class EmaneManager(ModelManager):
)
return EmaneManager.NOT_READY
ctrlnet = self.session.add_remove_control_net(
self.session.add_remove_control_net(
net_index=netidx, remove=False, conf_required=False
)
self.distributedctrlnet(ctrlnet)
if self.checkdistributed():
# we are slave, but haven't received a platformid yet
platform_id_start = "platform_id_start"
default_values = self.emane_config.default_values()
value = self.get_config(platform_id_start)
if value == default_values[platform_id_start]:
return EmaneManager.NOT_READY
self.check_node_models()
return EmaneManager.SUCCESS
@ -414,9 +388,6 @@ class EmaneManager(ModelManager):
"""
stop all EMANE daemons
"""
with self._ifccountslock:
self._ifccounts.clear()
with self._emane_node_lock:
if not self._emane_nets:
return
@ -425,92 +396,6 @@ class EmaneManager(ModelManager):
self.stopdaemons()
self.stopeventmonitor()
def handledistributed(self, message):
"""
Broker handler for processing CORE API messages as they are
received. This is used to snoop the Link add messages to get NEM
counts of NEMs that exist on other servers.
"""
if (
message.message_type == MessageTypes.LINK.value
and message.flags & MessageFlags.ADD.value
):
nn = message.node_numbers()
# first node is always link layer node in Link add message
if nn[0] in self.session.broker.network_nodes:
serverlist = self.session.broker.getserversbynode(nn[1])
for server in serverlist:
with self._ifccountslock:
if server not in self._ifccounts:
self._ifccounts[server] = 1
else:
self._ifccounts[server] += 1
def checkdistributed(self):
"""
Check for EMANE nodes that exist on multiple emulation servers and
coordinate the NEM id and port number space.
If we are the master EMANE node, return False so initialization will
proceed as normal; otherwise slaves return True here and
initialization is deferred.
"""
# check with the session if we are the "master" Emane object?
master = False
with self._emane_node_lock:
if self._emane_nets:
master = self.session.master
logging.info("emane check distributed as master: %s.", master)
# we are not the master Emane object, wait for nem id and ports
if not master:
return True
nemcount = 0
with self._emane_node_lock:
for key in self._emane_nets:
emane_node = self._emane_nets[key]
nemcount += emane_node.numnetif()
nemid = int(self.get_config("nem_id_start"))
nemid += nemcount
platformid = int(self.get_config("platform_id_start"))
# build an ordered list of servers so platform ID is deterministic
servers = []
for key in sorted(self._emane_nets):
for server in self.session.broker.getserversbynode(key):
if server not in servers:
servers.append(server)
servers.sort(key=lambda x: x.name)
for server in servers:
if server.name == "localhost":
continue
if server.sock is None:
continue
platformid += 1
# create temporary config for updating distributed nodes
typeflags = ConfigFlags.UPDATE.value
config = copy.deepcopy(self.get_configs())
config["platform_id_start"] = str(platformid)
config["nem_id_start"] = str(nemid)
config_data = ConfigShim.config_data(
0, None, typeflags, self.emane_config, config
)
message = dataconversion.convert_config(config_data)
server.sock.send(message)
# increment nemid for next server by number of interfaces
with self._ifccountslock:
if server in self._ifccounts:
nemid += self._ifccounts[server]
return False
def buildxml(self):
"""
Build XML files required to run EMANE on each node.
@ -527,52 +412,6 @@ class EmaneManager(ModelManager):
self.buildnemxml()
self.buildeventservicexml()
# TODO: remove need for tlv messaging
def distributedctrlnet(self, ctrlnet):
"""
Distributed EMANE requires multiple control network prefixes to
be configured. This generates configuration for slave control nets
using the default list of prefixes.
"""
# slave server
session = self.session
if not session.master:
return
# not distributed
servers = session.broker.getservernames()
if len(servers) < 2:
return
# normal Config messaging will distribute controlnets
prefix = session.options.get_config("controlnet", default="")
prefixes = prefix.split()
if len(prefixes) < len(servers):
logging.info(
"setting up default controlnet prefixes for distributed (%d configured)",
len(prefixes),
)
prefix = ctrlnet.DEFAULT_PREFIX_LIST[0]
prefixes = prefix.split()
servers.remove("localhost")
servers.insert(0, "localhost")
prefix = " ".join("%s:%s" % (s, prefixes[i]) for i, s in enumerate(servers))
# this generates a config message having controlnet prefix assignments
logging.info("setting up controlnet prefixes for distributed: %s", prefix)
vals = "controlnet=%s" % prefix
tlvdata = b""
tlvdata += coreapi.CoreConfigTlv.pack(ConfigTlvs.OBJECT.value, "session")
tlvdata += coreapi.CoreConfigTlv.pack(ConfigTlvs.TYPE.value, 0)
tlvdata += coreapi.CoreConfigTlv.pack(ConfigTlvs.VALUES.value, vals)
rawmsg = coreapi.CoreConfMessage.pack(0, tlvdata)
msghdr = rawmsg[: coreapi.CoreMessage.header_len]
msg = coreapi.CoreConfMessage(
flags=0, hdr=msghdr, data=rawmsg[coreapi.CoreMessage.header_len :]
)
logging.debug("sending controlnet message:\n%s", msg)
self.session.broker.handle_message(msg)
def check_node_models(self):
"""
Associate EMANE model classes with EMANE network nodes.
@ -677,11 +516,11 @@ class EmaneManager(ModelManager):
dev = self.get_config("eventservicedevice")
emanexml.create_event_service_xml(group, port, dev, self.session.session_dir)
for server in self.session.servers:
conn = self.session.servers[server]
emanexml.create_event_service_xml(
group, port, dev, self.session.session_dir, conn
self.session.distributed.execute(
lambda x: emanexml.create_event_service_xml(
group, port, dev, self.session.session_dir, x
)
)
def startdaemons(self):
"""
@ -757,9 +596,7 @@ class EmaneManager(ModelManager):
emanecmd += " -f %s" % os.path.join(path, "emane.log")
emanecmd += " %s" % os.path.join(path, "platform.xml")
utils.check_cmd(emanecmd, cwd=path)
for server in self.session.servers:
conn = self.session.servers[server]
distributed.remote_cmd(conn, emanecmd, cwd=path)
self.session.distributed.execute(lambda x: x.remote_cmd(emanecmd, cwd=path))
logging.info("host emane daemon running: %s", emanecmd)
def stopdaemons(self):
@ -784,10 +621,8 @@ class EmaneManager(ModelManager):
try:
utils.check_cmd(kill_emaned)
utils.check_cmd(kill_transortd)
for server in self.session.servers:
conn = self.session[server]
distributed.remote_cmd(conn, kill_emaned)
distributed.remote_cmd(conn, kill_transortd)
self.session.distributed.execute(lambda x: x.remote_cmd(kill_emaned))
self.session.distributed.execute(lambda x: x.remote_cmd(kill_transortd))
except CoreCommandError:
logging.exception("error shutting down emane daemons")

262
daemon/core/emulator/distributed.py
View file

@ -1,61 +1,247 @@
"""
Defines distributed server functionality.
"""
import logging
import os
import threading
from collections import OrderedDict
from tempfile import NamedTemporaryFile
from fabric import Connection
from invoke import UnexpectedExit
from core import utils
from core.errors import CoreCommandError
from core.nodes.interface import GreTap
from core.nodes.ipaddress import IpAddress
from core.nodes.network import CoreNetwork, CtrlNet
LOCK = threading.Lock()
def remote_cmd(server, cmd, env=None, cwd=None, wait=True):
class DistributedServer(object):
"""
Run command remotely using server connection.
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param str cmd: command to run
:param dict env: environment for remote command, default is None
:param str cwd: directory to run command in, defaults to None, which is the user's
home directory
:param bool wait: True to wait for status, False to background process
:return: stdout when success
:rtype: str
:raises CoreCommandError: when a non-zero exit status occurs
Provides distributed server interactions.
"""
replace_env = env is not None
if not wait:
cmd += " &"
logging.info(
"remote cmd server(%s) cwd(%s) wait(%s): %s", server.host, cwd, wait, cmd
)
try:
with LOCK:
def __init__(self, name, host):
"""
Create a DistributedServer instance.
:param str name: convenience name to associate with host
:param str host: host to connect to
"""
self.name = name
self.host = host
self.conn = Connection(host, user="root")
self.lock = threading.Lock()
def remote_cmd(self, cmd, env=None, cwd=None, wait=True):
"""
Run command remotely using server connection.
:param str cmd: command to run
:param dict env: environment for remote command, default is None
:param str cwd: directory to run command in, defaults to None, which is the
user's home directory
:param bool wait: True to wait for status, False to background process
:return: stdout when success
:rtype: str
:raises CoreCommandError: when a non-zero exit status occurs
"""
replace_env = env is not None
if not wait:
cmd += " &"
logging.info(
"remote cmd server(%s) cwd(%s) wait(%s): %s", self.host, cwd, wait, cmd
)
try:
if cwd is None:
result = server.run(cmd, hide=False, env=env, replace_env=replace_env)
result = self.conn.run(
cmd, hide=False, env=env, replace_env=replace_env
)
else:
with server.cd(cwd):
result = server.run(
with self.conn.cd(cwd):
result = self.conn.run(
cmd, hide=False, env=env, replace_env=replace_env
)
return result.stdout.strip()
except UnexpectedExit as e:
stdout, stderr = e.streams_for_display()
raise CoreCommandError(e.result.exited, cmd, stdout, stderr)
return result.stdout.strip()
except UnexpectedExit as e:
stdout, stderr = e.streams_for_display()
raise CoreCommandError(e.result.exited, cmd, stdout, stderr)
def remote_put(self, source, destination):
"""
Push file to remote server.
:param str source: source file to push
:param str destination: destination file location
:return: nothing
"""
with self.lock:
self.conn.put(source, destination)
def remote_put_temp(self, destination, data):
"""
Remote push file contents to a remote server, using a temp file as an
intermediate step.
:param str destination: file destination for data
:param str data: data to store in remote file
:return: nothing
"""
with self.lock:
temp = NamedTemporaryFile(delete=False)
temp.write(data.encode("utf-8"))
temp.close()
self.conn.put(temp.name, destination)
os.unlink(temp.name)
def remote_put(server, source, destination):
with LOCK:
server.put(source, destination)
class DistributedController(object):
def __init__(self, session):
"""
Create
:param session:
"""
self.session = session
self.servers = OrderedDict()
self.tunnels = {}
self.address = self.session.options.get_config(
"distributed_address", default=None
)
def add_server(self, name, host):
"""
Add distributed server configuration.
:param str name: distributed server name
:param str host: distributed server host address
:return: nothing
"""
server = DistributedServer(name, host)
self.servers[name] = server
cmd = "mkdir -p %s" % self.session.session_dir
server.remote_cmd(cmd)
def execute(self, func):
"""
Convenience for executing logic against all distributed servers.
:param func: function to run, that takes a DistributedServer as a parameter
:return: nothing
"""
for name in self.servers:
server = self.servers[name]
func(server)
def shutdown(self):
"""
Shutdown logic for dealing with distributed tunnels and server session
directories.
:return: nothing
"""
# shutdown all tunnels
for key in self.tunnels:
tunnels = self.tunnels[key]
for tunnel in tunnels:
tunnel.shutdown()
# remove all remote session directories
for name in self.servers:
server = self.servers[name]
cmd = "rm -rf %s" % self.session.session_dir
server.remote_cmd(cmd)
# clear tunnels
self.tunnels.clear()
def start(self):
"""
Start distributed network tunnels.
:return: nothing
"""
for node_id in self.session.nodes:
node = self.session.nodes[node_id]
if not isinstance(node, CoreNetwork):
continue
if isinstance(node, CtrlNet) and node.serverintf is not None:
continue
for name in self.servers:
server = self.servers[name]
self.create_gre_tunnel(node, server)
def create_gre_tunnel(self, node, server):
"""
Create gre tunnel using a pair of gre taps between the local and remote server.
def remote_put_temp(server, destination, data):
with LOCK:
temp = NamedTemporaryFile(delete=False)
temp.write(data.encode("utf-8"))
temp.close()
server.put(temp.name, destination)
os.unlink(temp.name)
:param core.nodes.network.CoreNetwork node: node to create gre tunnel for
:param core.emulator.distributed.DistributedServer server: server to create
tunnel for
:return: local and remote gre taps created for tunnel
:rtype: tuple
"""
host = server.host
key = self.tunnel_key(node.id, IpAddress.to_int(host))
tunnel = self.tunnels.get(key)
if tunnel is not None:
return tunnel
# local to server
logging.info(
"local tunnel node(%s) to remote(%s) key(%s)", node.name, host, key
)
local_tap = GreTap(session=self.session, remoteip=host, key=key)
local_tap.net_client.create_interface(node.brname, local_tap.localname)
# server to local
logging.info(
"remote tunnel node(%s) to local(%s) key(%s)", node.name, self.address, key
)
remote_tap = GreTap(
session=self.session, remoteip=self.address, key=key, server=server
)
remote_tap.net_client.create_interface(node.brname, remote_tap.localname)
# save tunnels for shutdown
tunnel = (local_tap, remote_tap)
self.tunnels[key] = tunnel
return tunnel
def tunnel_key(self, n1_id, n2_id):
"""
Compute a 32-bit key used to uniquely identify a GRE tunnel.
The hash(n1num), hash(n2num) values are used, so node numbers may be
None or string values (used for e.g. "ctrlnet").
:param int n1_id: node one id
:param int n2_id: node two id
:return: tunnel key for the node pair
:rtype: int
"""
logging.debug("creating tunnel key for: %s, %s", n1_id, n2_id)
key = (
(self.session.id << 16) ^ utils.hashkey(n1_id) ^ (utils.hashkey(n2_id) << 8)
)
return key & 0xFFFFFFFF
def get_tunnel(self, n1_id, n2_id):
"""
Return the GreTap between two nodes if it exists.
:param int n1_id: node one id
:param int n2_id: node two id
:return: gre tap between nodes or None
"""
key = self.tunnel_key(n1_id, n2_id)
logging.debug("checking for tunnel key(%s) in: %s", key, self.tunnels)
return self.tunnels.get(key)

168
daemon/core/emulator/session.py
View file

@ -14,14 +14,11 @@ import threading
import time
from multiprocessing.pool import ThreadPool
from fabric import Connection
from core import constants, utils
from core.api.tlv import coreapi
from core.api.tlv.broker import CoreBroker
from core.emane.emanemanager import EmaneManager
from core.emane.nodes import EmaneNet
from core.emulator.data import EventData, ExceptionData, NodeData
from core.emulator.distributed import DistributedController
from core.emulator.emudata import (
IdGen,
LinkOptions,
@ -37,11 +34,9 @@ from core.location.event import EventLoop
from core.location.mobility import MobilityManager
from core.nodes.base import CoreNetworkBase, CoreNode, CoreNodeBase
from core.nodes.docker import DockerNode
from core.nodes.interface import GreTap
from core.nodes.ipaddress import IpAddress, MacAddress
from core.nodes.ipaddress import MacAddress
from core.nodes.lxd import LxcNode
from core.nodes.network import (
CoreNetwork,
CtrlNet,
GreTapBridge,
HubNode,
@ -140,13 +135,10 @@ class Session(object):
self.options.set_config(key, value)
self.metadata = SessionMetaData()
# distributed servers
self.servers = {}
self.tunnels = {}
self.address = None
# distributed support and logic
self.distributed = DistributedController(self)
# initialize session feature helpers
self.broker = CoreBroker(session=self)
self.location = CoreLocation()
self.mobility = MobilityManager(session=self)
self.services = CoreServices(session=self)
@ -157,89 +149,11 @@ class Session(object):
self.services.default_services = {
"mdr": ("zebra", "OSPFv3MDR", "IPForward"),
"PC": ("DefaultRoute",),
"prouter": ("zebra", "OSPFv2", "OSPFv3", "IPForward"),
"prouter": (),
"router": ("zebra", "OSPFv2", "OSPFv3", "IPForward"),
"host": ("DefaultRoute", "SSH"),
}
def add_distributed(self, server):
conn = Connection(server, user="root")
self.servers[server] = conn
cmd = "mkdir -p %s" % self.session_dir
conn.run(cmd, hide=False)
def shutdown_distributed(self):
# shutdown all tunnels
for key in self.tunnels:
tunnels = self.tunnels[key]
for tunnel in tunnels:
tunnel.shutdown()
# remove all remote session directories
for server in self.servers:
conn = self.servers[server]
cmd = "rm -rf %s" % self.session_dir
conn.run(cmd, hide=False)
# clear tunnels
self.tunnels.clear()
def initialize_distributed(self):
for node_id in self.nodes:
node = self.nodes[node_id]
if not isinstance(node, CoreNetwork):
continue
if isinstance(node, CtrlNet) and node.serverintf is not None:
continue
for server in self.servers:
conn = self.servers[server]
key = self.tunnelkey(node_id, IpAddress.to_int(server))
# local to server
logging.info(
"local tunnel node(%s) to remote(%s) key(%s)",
node.name,
server,
key,
)
local_tap = GreTap(session=self, remoteip=server, key=key)
local_tap.net_client.create_interface(node.brname, local_tap.localname)
# server to local
logging.info(
"remote tunnel node(%s) to local(%s) key(%s)",
node.name,
self.address,
key,
)
remote_tap = GreTap(
session=self, remoteip=self.address, key=key, server=conn
)
remote_tap.net_client.create_interface(
node.brname, remote_tap.localname
)
# save tunnels for shutdown
self.tunnels[key] = [local_tap, remote_tap]
def tunnelkey(self, n1num, n2num):
"""
Compute a 32-bit key used to uniquely identify a GRE tunnel.
The hash(n1num), hash(n2num) values are used, so node numbers may be
None or string values (used for e.g. "ctrlnet").
:param int n1num: node one id
:param int n2num: node two id
:return: tunnel key for the node pair
:rtype: int
"""
logging.debug("creating tunnel key for: %s, %s", n1num, n2num)
key = (self.id << 16) ^ utils.hashkey(n1num) ^ (utils.hashkey(n2num) << 8)
return key & 0xFFFFFFFF
@classmethod
def get_node_class(cls, _type):
"""
@ -289,7 +203,7 @@ class Session(object):
node_two = self.get_node(node_two_id)
# both node ids are provided
tunnel = self.broker.gettunnel(node_one_id, node_two_id)
tunnel = self.distributed.get_tunnel(node_one_id, node_two_id)
logging.debug("tunnel between nodes: %s", tunnel)
if isinstance(tunnel, GreTapBridge):
net_one = tunnel
@ -754,7 +668,7 @@ class Session(object):
name = "%s%s" % (node_class.__name__, _id)
# verify distributed server
server = self.servers.get(node_options.emulation_server)
server = self.distributed.servers.get(node_options.emulation_server)
if node_options.emulation_server is not None and server is None:
raise CoreError(
"invalid distributed server: %s" % node_options.emulation_server
@ -775,6 +689,7 @@ class Session(object):
name=name,
start=start,
image=node_options.image,
server=server,
)
else:
node = self.create_node(
@ -962,13 +877,13 @@ class Session(object):
def clear(self):
"""
Clear all CORE session data. (objects, hooks, broker)
Clear all CORE session data. (nodes, hooks, etc)
:return: nothing
"""
self.delete_nodes()
self.distributed.shutdown()
self.del_hooks()
self.broker.reset()
self.emane.reset()
def start_events(self):
@ -1042,17 +957,16 @@ class Session(object):
# shutdown/cleanup feature helpers
self.emane.shutdown()
self.broker.shutdown()
self.sdt.shutdown()
# delete all current nodes
# remove and shutdown all nodes and tunnels
self.delete_nodes()
self.distributed.shutdown()
# remove this sessions working directory
preserve = self.options.get_config("preservedir") == "1"
if not preserve:
shutil.rmtree(self.session_dir, ignore_errors=True)
self.shutdown_distributed()
# call session shutdown handlers
for handler in self.shutdown_handlers:
@ -1164,7 +1078,7 @@ class Session(object):
"""
try:
state_file = open(self._state_file, "w")
state_file.write("%d %s\n" % (state, coreapi.state_name(state)))
state_file.write("%d %s\n" % (state, EventTypes(self.state).name))
state_file.close()
except IOError:
logging.exception("error writing state file: %s", state)
@ -1282,7 +1196,7 @@ class Session(object):
hook(state)
except Exception:
message = "exception occured when running %s state hook: %s" % (
coreapi.state_name(state),
EventTypes(self.state).name,
hook,
)
logging.exception(message)
@ -1553,14 +1467,13 @@ class Session(object):
# write current nodes out to session directory file
self.write_nodes()
# create control net interfaces and broker network tunnels
# create control net interfaces and network tunnels
# which need to exist for emane to sync on location events
# in distributed scenarios
self.add_remove_control_interface(node=None, remove=False)
self.broker.startup()
# initialize distributed tunnels
self.initialize_distributed()
self.distributed.start()
# instantiate will be invoked again upon Emane configure
if self.emane.startup() == self.emane.NOT_READY:
@ -1570,9 +1483,6 @@ class Session(object):
self.boot_nodes()
self.mobility.startup()
# set broker local instantiation to complete
self.broker.local_instantiation_complete()
# notify listeners that instantiation is complete
event = EventData(event_type=EventTypes.INSTANTIATION_COMPLETE.value)
self.broadcast_event(event)
@ -1610,21 +1520,16 @@ class Session(object):
have entered runtime (time=0).
"""
# this is called from instantiate() after receiving an event message
# for the instantiation state, and from the broker when distributed
# nodes have been started
# for the instantiation state
logging.debug(
"session(%s) checking if not in runtime state, current state: %s",
self.id,
coreapi.state_name(self.state),
EventTypes(self.state).name,
)
if self.state == EventTypes.RUNTIME_STATE.value:
logging.info("valid runtime state found, returning")
return
# check to verify that all nodes and networks are running
if not self.broker.instantiation_complete():
return
# start event loop and set to runtime
self.event_loop.run()
self.set_state(EventTypes.RUNTIME_STATE, send_event=True)
@ -1834,37 +1739,11 @@ class Session(object):
except IndexError:
# no server name. possibly only one server
prefix = prefixes[0]
else:
# slave servers have their name and localhost in the serverlist
servers = self.broker.getservernames()
servers.remove("localhost")
prefix = None
for server_prefix in prefixes:
try:
# split each entry into server and prefix
server, p = server_prefix.split(":")
except ValueError:
server = ""
p = None
if server == servers[0]:
# the server name in the list matches this server
prefix = p
break
if not prefix:
logging.error(
"control network prefix not found for server: %s", servers[0]
)
assign_address = False
try:
prefix = prefixes[0].split(":", 1)[1]
except IndexError:
prefix = prefixes[0]
# len(prefixes) == 1
else:
# TODO: can we get the server name from the servers.conf or from the node assignments?
# TODO: can we get the server name from the servers.conf or from the node
# assignments?o
# with one prefix, only master gets a ctrlnet address
assign_address = self.master
prefix = prefixes[0]
@ -1886,13 +1765,6 @@ class Session(object):
serverintf=server_interface,
)
# tunnels between controlnets will be built with Broker.addnettunnels()
# TODO: potentially remove documentation saying node ids are ints
# TODO: need to move broker code out of the session object
self.broker.addnet(_id)
for server in self.broker.getservers():
self.broker.addnodemap(server, _id)
return control_net
def add_remove_control_interface(

93
daemon/core/location/mobility.py
View file

@ -19,13 +19,9 @@ from core.emulator.enumerations import (
EventTypes,
LinkTypes,
MessageFlags,
MessageTypes,
NodeTlvs,
RegisterTlvs,
)
from core.errors import CoreError
from core.nodes.base import CoreNodeBase
from core.nodes.ipaddress import IpAddress
class MobilityManager(ModelManager):
@ -48,11 +44,6 @@ class MobilityManager(ModelManager):
self.models[BasicRangeModel.name] = BasicRangeModel
self.models[Ns2ScriptedMobility.name] = Ns2ScriptedMobility
# dummy node objects for tracking position of nodes on other servers
self.phys = {}
self.physnets = {}
self.session.broker.handlers.add(self.physnodehandlelink)
def reset(self):
"""
Clear out all current configurations.
@ -93,9 +84,6 @@ class MobilityManager(ModelManager):
model_class = self.models[model_name]
self.set_model(node, model_class, config)
if self.session.master:
self.installphysnodes(node)
if node.mobility:
self.session.event_loop.add_event(0.0, node.mobility.startup)
@ -209,87 +197,6 @@ class MobilityManager(ModelManager):
if node.model:
node.model.update(moved, moved_netifs)
def addphys(self, netnum, node):
"""
Keep track of PhysicalNodes and which network they belong to.
:param int netnum: network number
:param core.coreobj.PyCoreNode node: node to add physical network to
:return: nothing
"""
node_id = node.id
self.phys[node_id] = node
if netnum not in self.physnets:
self.physnets[netnum] = [node_id]
else:
self.physnets[netnum].append(node_id)
# TODO: remove need for handling old style message
def physnodehandlelink(self, message):
"""
Broker handler. Snoop Link add messages to get
node numbers of PhyiscalNodes and their nets.
Physical nodes exist only on other servers, but a shadow object is
created here for tracking node position.
:param message: link message to handle
:return: nothing
"""
if (
message.message_type == MessageTypes.LINK.value
and message.flags & MessageFlags.ADD.value
):
nn = message.node_numbers()
# first node is always link layer node in Link add message
if nn[0] not in self.session.broker.network_nodes:
return
if nn[1] in self.session.broker.physical_nodes:
# record the fact that this PhysicalNode is linked to a net
dummy = CoreNodeBase(
session=self.session, _id=nn[1], name="n%d" % nn[1], start=False
)
self.addphys(nn[0], dummy)
# TODO: remove need to handling old style messages
def physnodeupdateposition(self, message):
"""
Snoop node messages belonging to physical nodes. The dummy object
in self.phys[] records the node position.
:param message: message to handle
:return: nothing
"""
nodenum = message.node_numbers()[0]
try:
dummy = self.phys[nodenum]
nodexpos = message.get_tlv(NodeTlvs.X_POSITION.value)
nodeypos = message.get_tlv(NodeTlvs.Y_POSITION.value)
dummy.setposition(nodexpos, nodeypos, None)
except KeyError:
logging.exception("error retrieving physical node: %s", nodenum)
def installphysnodes(self, net):
"""
After installing a mobility model on a net, include any physical
nodes that we have recorded. Use the GreTap tunnel to the physical node
as the node's interface.
:param net: network to install
:return: nothing
"""
node_ids = self.physnets.get(net.id, [])
for node_id in node_ids:
node = self.phys[node_id]
# TODO: fix this bad logic, relating to depending on a break to get a valid server
for server in self.session.broker.getserversbynode(node_id):
break
netif = self.session.broker.gettunnel(net.id, IpAddress.to_int(server.host))
node.addnetif(netif, 0)
netif.node = node
x, y, z = netif.node.position.get()
netif.poshook(netif, x, y, z)
class WirelessModel(ConfigurableOptions):
"""

111
daemon/core/nodes/base.py
View file

@ -4,23 +4,19 @@ Defines the base logic for nodes used within core.
import logging
import os
import random
import shutil
import socket
import string
import threading
from builtins import range
from socket import AF_INET, AF_INET6
from core import utils
from core.constants import MOUNT_BIN, VNODED_BIN
from core.emulator import distributed
from core.emulator.data import LinkData, NodeData
from core.emulator.enumerations import LinkTypes, NodeTypes
from core.errors import CoreCommandError
from core.nodes import client, ipaddress
from core.nodes.interface import CoreInterface, TunTap, Veth
from core.nodes.netclient import LinuxNetClient, OvsNetClient
from core.nodes.interface import TunTap, Veth
from core.nodes.netclient import get_net_client
_DEFAULT_MTU = 1500
@ -41,8 +37,8 @@ class NodeBase(object):
:param int _id: id
:param str name: object name
:param bool start: start value
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
"""
self.session = session
@ -64,10 +60,8 @@ class NodeBase(object):
self.opaque = None
self.position = Position()
if session.options.get_config("ovs") == "True":
self.net_client = OvsNetClient(self.net_cmd)
else:
self.net_client = LinuxNetClient(self.net_cmd)
use_ovs = session.options.get_config("ovs") == "True"
self.net_client = get_net_client(use_ovs, self.net_cmd)
def startup(self):
"""
@ -101,7 +95,7 @@ class NodeBase(object):
if self.server is None:
return utils.check_cmd(args, env, cwd, wait)
else:
return distributed.remote_cmd(self.server, args, env, cwd, wait)
return self.server.remote_cmd(args, env, cwd, wait)
def setposition(self, x=None, y=None, z=None):
"""
@ -200,7 +194,9 @@ class NodeBase(object):
x, y, _ = self.getposition()
model = self.type
emulation_server = self.server
emulation_server = None
if self.server is not None:
emulation_server = self.server.host
services = self.services
if services is not None:
@ -253,8 +249,8 @@ class CoreNodeBase(NodeBase):
:param int _id: object id
:param str name: object name
:param bool start: boolean for starting
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
"""
super(CoreNodeBase, self).__init__(session, _id, name, start, server)
self.services = []
@ -437,8 +433,8 @@ class CoreNode(CoreNodeBase):
:param str nodedir: node directory
:param str bootsh: boot shell to use
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
"""
super(CoreNode, self).__init__(session, _id, name, start, server)
self.nodedir = nodedir
@ -452,14 +448,22 @@ class CoreNode(CoreNodeBase):
self._mounts = []
self.bootsh = bootsh
if session.options.get_config("ovs") == "True":
self.node_net_client = OvsNetClient(self.node_net_cmd)
else:
self.node_net_client = LinuxNetClient(self.node_net_cmd)
use_ovs = session.options.get_config("ovs") == "True"
self.node_net_client = self.create_node_net_client(use_ovs)
if start:
self.startup()
def create_node_net_client(self, use_ovs):
"""
Create node network client for running network commands within the nodes
container.
:param bool use_ovs: True for OVS bridges, False for Linux bridges
:return:node network client
"""
return get_net_client(use_ovs, self.node_net_cmd)
def alive(self):
"""
Check if the node is alive.
@ -575,7 +579,7 @@ class CoreNode(CoreNodeBase):
return self.client.check_cmd(args, wait=wait)
else:
args = self.client.create_cmd(args)
return distributed.remote_cmd(self.server, args, wait=wait)
return self.server.remote_cmd(args, wait=wait)
def termcmdstring(self, sh="/bin/sh"):
"""
@ -584,7 +588,13 @@ class CoreNode(CoreNodeBase):
:param str sh: shell to execute command in
:return: str
"""
return self.client.termcmdstring(sh)
terminal = self.client.create_cmd(sh)
if self.server is None:
return terminal
else:
return "ssh -X -f {host} xterm -e {terminal}".format(
host=self.server.host, terminal=terminal
)
def privatedir(self, path):
"""
@ -658,11 +668,7 @@ class CoreNode(CoreNodeBase):
raise ValueError("interface name (%s) too long" % name)
veth = Veth(
node=self,
name=name,
localname=localname,
start=self.up,
server=self.server,
self.session, self, name, localname, start=self.up, server=self.server
)
if self.up:
@ -715,7 +721,7 @@ class CoreNode(CoreNodeBase):
sessionid = self.session.short_session_id()
localname = "tap%s.%s.%s" % (self.id, ifindex, sessionid)
name = ifname
tuntap = TunTap(node=self, name=name, localname=localname, start=self.up)
tuntap = TunTap(self.session, self, name, localname, start=self.up)
try:
self.addnetif(tuntap, ifindex)
@ -832,35 +838,6 @@ class CoreNode(CoreNodeBase):
self.ifup(ifindex)
return ifindex
def connectnode(self, ifname, othernode, otherifname):
"""
Connect a node.
:param str ifname: name of interface to connect
:param core.nodes.base.CoreNode othernode: node to connect to
:param str otherifname: interface name to connect to
:return: nothing
"""
tmplen = 8
tmp1 = "tmp." + "".join(
[random.choice(string.ascii_lowercase) for _ in range(tmplen)]
)
tmp2 = "tmp." + "".join(
[random.choice(string.ascii_lowercase) for _ in range(tmplen)]
)
self.net_client.create_veth(tmp1, tmp2)
self.net_client.device_ns(tmp1, str(self.pid))
self.node_net_client.device_name(tmp1, ifname)
interface = CoreInterface(node=self, name=ifname, mtu=_DEFAULT_MTU)
self.addnetif(interface, self.newifindex())
self.net_client.device_ns(tmp2, str(othernode.pid))
othernode.node_net_client.device_name(tmp2, otherifname)
other_interface = CoreInterface(
node=othernode, name=otherifname, mtu=_DEFAULT_MTU
)
othernode.addnetif(other_interface, othernode.newifindex())
def addfile(self, srcname, filename):
"""
Add a file.
@ -878,7 +855,7 @@ class CoreNode(CoreNodeBase):
self.client.check_cmd("sync")
else:
self.net_cmd("mkdir -p %s" % directory)
distributed.remote_put(self.server, srcname, filename)
self.server.remote_put(srcname, filename)
def hostfilename(self, filename):
"""
@ -915,7 +892,7 @@ class CoreNode(CoreNodeBase):
os.chmod(open_file.name, mode)
else:
self.net_cmd("mkdir -m %o -p %s" % (0o755, dirname))
distributed.remote_put_temp(self.server, hostfilename, contents)
self.server.remote_put_temp(hostfilename, contents)
self.net_cmd("chmod %o %s" % (mode, hostfilename))
logging.debug(
"node(%s) added file: %s; mode: 0%o", self.name, hostfilename, mode
@ -934,12 +911,10 @@ class CoreNode(CoreNodeBase):
hostfilename = self.hostfilename(filename)
if self.server is None:
shutil.copy2(srcfilename, hostfilename)
if mode is not None:
os.chmod(hostfilename, mode)
else:
distributed.remote_put(self.server, srcfilename, hostfilename)
if mode is not None:
self.net_cmd("chmod %o %s" % (mode, hostfilename))
self.server.remote_put(srcfilename, hostfilename)
if mode is not None:
self.net_cmd("chmod %o %s" % (mode, hostfilename))
logging.info(
"node(%s) copied file: %s; mode: %s", self.name, hostfilename, mode
)
@ -961,8 +936,8 @@ class CoreNetworkBase(NodeBase):
:param int _id: object id
:param str name: object name
:param bool start: should object start
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
"""
super(CoreNetworkBase, self).__init__(session, _id, name, start, server)
self._linked = {}

11
daemon/core/nodes/client.py
View file

@ -4,7 +4,7 @@ over a control channel to the vnoded process running in a network namespace.
The control channel can be accessed via calls using the vcmd shell.
"""
from core import constants, utils
from core import utils
from core.constants import VCMD_BIN
@ -66,12 +66,3 @@ class VnodeClient(object):
self._verify_connection()
args = self.create_cmd(args)
return utils.check_cmd(args, wait=wait)
def termcmdstring(self, sh="/bin/sh"):
"""
Create a terminal command string.
:param str sh: shell to execute command in
:return: str
"""
return "%s -c %s -- %s" % (constants.VCMD_BIN, self.ctrlchnlname, sh)

110
daemon/core/nodes/docker.py
View file

@ -1,21 +1,24 @@
import json
import logging
import os
from tempfile import NamedTemporaryFile
from core import utils
from core.emulator.enumerations import NodeTypes
from core.errors import CoreCommandError
from core.nodes.base import CoreNode
from core.nodes.netclient import get_net_client
class DockerClient(object):
def __init__(self, name, image):
def __init__(self, name, image, run):
self.name = name
self.image = image
self.run = run
self.pid = None
def create_container(self):
utils.check_cmd(
self.run(
"docker run -td --init --net=none --hostname {name} --name {name} "
"--sysctl net.ipv6.conf.all.disable_ipv6=0 "
"{image} /bin/bash".format(
@ -27,7 +30,7 @@ class DockerClient(object):
def get_info(self):
args = "docker inspect {name}".format(name=self.name)
output = utils.check_cmd(args)
output = self.run(args)
data = json.loads(output)
if not data:
raise CoreCommandError(
@ -43,22 +46,24 @@ class DockerClient(object):
return False
def stop_container(self):
utils.check_cmd("docker rm -f {name}".format(
self.run("docker rm -f {name}".format(
name=self.name
))
def check_cmd(self, cmd):
if isinstance(cmd, list):
cmd = " ".join(cmd)
logging.info("docker cmd output: %s", cmd)
return utils.check_cmd("docker exec {name} {cmd}".format(
name=self.name,
cmd=cmd
))
def create_ns_cmd(self, cmd):
return "nsenter -t {pid} -u -i -p -n {cmd}".format(
pid=self.pid,
cmd=cmd
)
def ns_cmd(self, cmd, wait):
if isinstance(cmd, list):
cmd = " ".join(cmd)
args = "nsenter -t {pid} -u -i -p -n {cmd}".format(
pid=self.pid,
cmd=cmd
@ -67,7 +72,7 @@ class DockerClient(object):
def get_pid(self):
args = "docker inspect -f '{{{{.State.Pid}}}}' {name}".format(name=self.name)
output = utils.check_cmd(args)
output = self.run(args)
self.pid = output
logging.debug("node(%s) pid: %s", self.name, self.pid)
return output
@ -78,13 +83,23 @@ class DockerClient(object):
name=self.name,
destination=destination
)
return utils.check_cmd(args)
return self.run(args)
class DockerNode(CoreNode):
apitype = NodeTypes.DOCKER.value
def __init__(self, session, _id=None, name=None, nodedir=None, bootsh="boot.sh", start=True, image=None):
def __init__(
self,
session,
_id=None,
name=None,
nodedir=None,
bootsh="boot.sh",
start=True,
server=None,
image=None
):
"""
Create a DockerNode instance.
@ -94,12 +109,26 @@ class DockerNode(CoreNode):
:param str nodedir: node directory
:param str bootsh: boot shell to use
:param bool start: start flag
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:param str image: image to start container with
"""
if image is None:
image = "ubuntu"
self.image = image
super(DockerNode, self).__init__(session, _id, name, nodedir, bootsh, start)
super(DockerNode, self).__init__(
session, _id, name, nodedir, bootsh, start, server
)
def create_node_net_client(self, use_ovs):
"""
Create node network client for running network commands within the nodes
container.
:param bool use_ovs: True for OVS bridges, False for Linux bridges
:return:node network client
"""
return get_net_client(use_ovs, self.nsenter_cmd)
def alive(self):
"""
@ -122,7 +151,7 @@ class DockerNode(CoreNode):
if self.up:
raise ValueError("starting a node that is already up")
self.makenodedir()
self.client = DockerClient(self.name, self.image)
self.client = DockerClient(self.name, self.image, self.net_cmd)
self.pid = self.client.create_container()
self.up = True
@ -141,12 +170,13 @@ class DockerNode(CoreNode):
self.client.stop_container()
self.up = False
def node_net_cmd(self, args, wait=True):
if not self.up:
logging.debug("node down, not running network command: %s", args)
return ""
return self.client.ns_cmd(args, wait)
def nsenter_cmd(self, args, wait=True):
if self.server is None:
args = self.client.create_ns_cmd(args)
return utils.check_cmd(args, wait=wait)
else:
args = self.client.create_ns_cmd(args)
return self.server.remote_cmd(args, wait=wait)
def termcmdstring(self, sh="/bin/sh"):
"""
@ -166,7 +196,7 @@ class DockerNode(CoreNode):
"""
logging.debug("creating node dir: %s", path)
args = "mkdir -p {path}".format(path=path)
self.client.check_cmd(args)
self.node_net_cmd(args)
def mount(self, source, target):
"""
@ -189,13 +219,24 @@ class DockerNode(CoreNode):
:param int mode: mode for file
:return: nothing
"""
logging.debug("node dir(%s) ctrlchannel(%s)", self.nodedir, self.ctrlchnlname)
logging.debug("nodefile filename(%s) mode(%s)", filename, mode)
file_path = os.path.join(self.nodedir, filename)
with open(file_path, "w") as f:
os.chmod(f.name, mode)
f.write(contents)
self.client.copy_file(file_path, filename)
directory = os.path.dirname(filename)
temp = NamedTemporaryFile(delete=False)
temp.write(contents.encode("utf-8"))
temp.close()
if directory:
self.node_net_cmd("mkdir -m %o -p %s" % (0o755, directory))
if self.server is not None:
self.server.remote_put(temp.name, temp.name)
self.client.copy_file(temp.name, filename)
self.node_net_cmd("chmod %o %s" % (mode, filename))
if self.server is not None:
self.net_cmd("rm -f %s" % temp.name)
os.unlink(temp.name)
logging.debug(
"node(%s) added file: %s; mode: 0%o", self.name, filename, mode
)
def nodefilecopy(self, filename, srcfilename, mode=None):
"""
@ -207,5 +248,18 @@ class DockerNode(CoreNode):
:param int mode: mode to copy to
:return: nothing
"""
logging.info("node file copy file(%s) source(%s) mode(%s)", filename, srcfilename, mode)
raise Exception("not supported")
logging.info(
"node file copy file(%s) source(%s) mode(%s)", filename, srcfilename, mode
)
directory = os.path.dirname(filename)
self.node_net_cmd("mkdir -p %s" % directory)
if self.server is None:
source = srcfilename
else:
temp = NamedTemporaryFile(delete=False)
source = temp.name
self.server.remote_put(source, temp.name)
self.client.copy_file(source, filename)
self.node_net_cmd("chmod %o %s" % (mode, filename))

66
daemon/core/nodes/interface.py
View file

@ -7,9 +7,8 @@ import time
from builtins import int, range
from core import utils
from core.emulator import distributed
from core.errors import CoreCommandError
from core.nodes.netclient import LinuxNetClient
from core.nodes.netclient import get_net_client
class CoreInterface(object):
@ -17,17 +16,18 @@ class CoreInterface(object):
Base class for network interfaces.
"""
def __init__(self, node, name, mtu, server=None):
def __init__(self, session, node, name, mtu, server=None):
"""
Creates a PyCoreNetIf instance.
:param core.emulator.session.Session session: core session instance
:param core.nodes.base.CoreNode node: node for interface
:param str name: interface name
:param mtu: mtu value
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param int mtu: mtu value
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
"""
self.session = session
self.node = node
self.name = name
if not isinstance(mtu, int):
@ -46,7 +46,8 @@ class CoreInterface(object):
# index used to find flow data
self.flow_id = None
self.server = server
self.net_client = LinuxNetClient(self.net_cmd)
use_ovs = session.options.get_config("ovs") == "True"
self.net_client = get_net_client(use_ovs, self.net_cmd)
def net_cmd(self, args, env=None, cwd=None, wait=True):
"""
@ -63,7 +64,7 @@ class CoreInterface(object):
if self.server is None:
return utils.check_cmd(args, env, cwd, wait)
else:
return distributed.remote_cmd(self.server, args, env, cwd, wait)
return self.server.remote_cmd(args, env, cwd, wait)
def startup(self):
"""
@ -212,21 +213,24 @@ class Veth(CoreInterface):
Provides virtual ethernet functionality for core nodes.
"""
def __init__(self, node, name, localname, mtu=1500, server=None, start=True):
def __init__(
self, session, node, name, localname, mtu=1500, server=None, start=True
):
"""
Creates a VEth instance.
:param core.emulator.session.Session session: core session instance
:param core.nodes.base.CoreNode node: related core node
:param str name: interface name
:param str localname: interface local name
:param mtu: interface mtu
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param int mtu: interface mtu
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:param bool start: start flag
:raises CoreCommandError: when there is a command exception
"""
# note that net arg is ignored
CoreInterface.__init__(self, node, name, mtu, server)
CoreInterface.__init__(self, session, node, name, mtu, server)
self.localname = localname
self.up = False
if start:
@ -272,19 +276,22 @@ class TunTap(CoreInterface):
TUN/TAP virtual device in TAP mode
"""
def __init__(self, node, name, localname, mtu=1500, server=None, start=True):
def __init__(
self, session, node, name, localname, mtu=1500, server=None, start=True
):
"""
Create a TunTap instance.
:param core.emulator.session.Session session: core session instance
:param core.nodes.base.CoreNode node: related core node
:param str name: interface name
:param str localname: local interface name
:param mtu: interface mtu
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param int mtu: interface mtu
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:param bool start: start flag
"""
CoreInterface.__init__(self, node, name, mtu, server)
CoreInterface.__init__(self, session, node, name, mtu, server)
self.localname = localname
self.up = False
self.transport_type = "virtual"
@ -456,19 +463,18 @@ class GreTap(CoreInterface):
:param core.nodes.base.CoreNode node: related core node
:param str name: interface name
:param core.emulator.session.Session session: core session instance
:param mtu: interface mtu
:param int mtu: interface mtu
:param str remoteip: remote address
:param int _id: object id
:param str localip: local address
:param ttl: ttl value
:param key: gre tap key
:param int ttl: ttl value
:param int key: gre tap key
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:raises CoreCommandError: when there is a command exception
"""
CoreInterface.__init__(self, node, name, mtu, server)
self.session = session
CoreInterface.__init__(self, session, node, name, mtu, server)
if _id is None:
# from PyCoreObj
_id = ((id(self) >> 16) ^ (id(self) & 0xFFFF)) & 0xFFFF
@ -484,13 +490,7 @@ class GreTap(CoreInterface):
if remoteip is None:
raise ValueError("missing remote IP required for GRE TAP device")
if localip is not None:
localip = str(localip)
if ttl is not None:
ttl = str(ttl)
if key is not None:
key = str(key)
self.net_client.create_gretap(self.localname, str(remoteip), localip, ttl, key)
self.net_client.create_gretap(self.localname, remoteip, localip, ttl, key)
self.net_client.device_up(self.localname)
self.up = True

76
daemon/core/nodes/lxd.py
View file

@ -2,6 +2,7 @@ import json
import logging
import os
import time
from tempfile import NamedTemporaryFile
from core import utils
from core.emulator.enumerations import NodeTypes
@ -10,22 +11,21 @@ from core.nodes.base import CoreNode
class LxdClient(object):
def __init__(self, name, image):
def __init__(self, name, image, run):
self.name = name
self.image = image
self.run = run
self.pid = None
def create_container(self):
utils.check_cmd(
"lxc launch {image} {name}".format(name=self.name, image=self.image)
)
self.run("lxc launch {image} {name}".format(name=self.name, image=self.image))
data = self.get_info()
self.pid = data["state"]["pid"]
return self.pid
def get_info(self):
args = "lxc list {name} --format json".format(name=self.name)
output = utils.check_cmd(args)
output = self.run(args)
data = json.loads(output)
if not data:
raise CoreCommandError(
@ -41,20 +41,16 @@ class LxdClient(object):
return False
def stop_container(self):
utils.check_cmd("lxc delete --force {name}".format(name=self.name))
self.run("lxc delete --force {name}".format(name=self.name))
def create_cmd(self, cmd):
return "lxc exec -nT {name} -- {cmd}".format(name=self.name, cmd=cmd)
def check_cmd(self, cmd, wait=True):
args = self.create_cmd(cmd)
return utils.check_cmd(args, wait=wait)
def create_ns_cmd(self, cmd):
return "nsenter -t {pid} -m -u -i -p -n {cmd}".format(pid=self.pid, cmd=cmd)
def ns_check_cmd(self, cmd, wait=True):
args = self.create_ns_cmd(cmd)
def check_cmd(self, cmd, wait=True):
args = self.create_cmd(cmd)
return utils.check_cmd(args, wait=wait)
def copy_file(self, source, destination):
@ -64,7 +60,7 @@ class LxdClient(object):
args = "lxc file push {source} {name}/{destination}".format(
source=source, name=self.name, destination=destination
)
utils.check_cmd(args)
self.run(args)
class LxcNode(CoreNode):
@ -78,6 +74,7 @@ class LxcNode(CoreNode):
nodedir=None,
bootsh="boot.sh",
start=True,
server=None,
image=None,
):
"""
@ -89,12 +86,16 @@ class LxcNode(CoreNode):
:param str nodedir: node directory
:param str bootsh: boot shell to use
:param bool start: start flag
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:param str image: image to start container with
"""
if image is None:
image = "ubuntu"
self.image = image
super(LxcNode, self).__init__(session, _id, name, nodedir, bootsh, start)
super(LxcNode, self).__init__(
session, _id, name, nodedir, bootsh, start, server
)
def alive(self):
"""
@ -115,7 +116,7 @@ class LxcNode(CoreNode):
if self.up:
raise ValueError("starting a node that is already up")
self.makenodedir()
self.client = LxdClient(self.name, self.image)
self.client = LxdClient(self.name, self.image, self.net_cmd)
self.pid = self.client.create_container()
self.up = True
@ -134,12 +135,6 @@ class LxcNode(CoreNode):
self.client.stop_container()
self.up = False
def node_net_cmd(self, args, wait=True):
if not self.up:
logging.debug("node down, not running network command: %s", args)
return ""
return self.client.check_cmd(args, wait)
def termcmdstring(self, sh="/bin/sh"):
"""
Create a terminal command string.
@ -147,7 +142,7 @@ class LxcNode(CoreNode):
:param str sh: shell to execute command in
:return: str
"""
return "lxc exec {name} -- bash".format(name=self.name)
return "lxc exec {name} -- {sh}".format(name=self.name, sh=sh)
def privatedir(self, path):
"""
@ -158,7 +153,7 @@ class LxcNode(CoreNode):
"""
logging.info("creating node dir: %s", path)
args = "mkdir -p {path}".format(path=path)
return self.client.check_cmd(args)
return self.node_net_cmd(args)
def mount(self, source, target):
"""
@ -181,13 +176,23 @@ class LxcNode(CoreNode):
:param int mode: mode for file
:return: nothing
"""
logging.debug("node dir(%s) ctrlchannel(%s)", self.nodedir, self.ctrlchnlname)
logging.debug("nodefile filename(%s) mode(%s)", filename, mode)
file_path = os.path.join(self.nodedir, filename)
with open(file_path, "w") as f:
os.chmod(f.name, mode)
f.write(contents)
self.client.copy_file(file_path, filename)
directory = os.path.dirname(filename)
temp = NamedTemporaryFile(delete=False)
temp.write(contents.encode("utf-8"))
temp.close()
if directory:
self.node_net_cmd("mkdir -m %o -p %s" % (0o755, directory))
if self.server is not None:
self.server.remote_put(temp.name, temp.name)
self.client.copy_file(temp.name, filename)
self.node_net_cmd("chmod %o %s" % (mode, filename))
if self.server is not None:
self.net_cmd("rm -f %s" % temp.name)
os.unlink(temp.name)
logging.debug("node(%s) added file: %s; mode: 0%o", self.name, filename, mode)
def nodefilecopy(self, filename, srcfilename, mode=None):
"""
@ -202,7 +207,18 @@ class LxcNode(CoreNode):
logging.info(
"node file copy file(%s) source(%s) mode(%s)", filename, srcfilename, mode
)
raise Exception("not supported")
directory = os.path.dirname(filename)
self.node_net_cmd("mkdir -p %s" % directory)
if self.server is None:
source = srcfilename
else:
temp = NamedTemporaryFile(delete=False)
source = temp.name
self.server.remote_put(source, temp.name)
self.client.copy_file(source, filename)
self.node_net_cmd("chmod %o %s" % (mode, filename))
def addnetif(self, netif, ifindex):
super(LxcNode, self).addnetif(netif, ifindex)

18
daemon/core/nodes/netclient.py
View file

@ -7,6 +7,20 @@ import os
from core.constants import BRCTL_BIN, ETHTOOL_BIN, IP_BIN, OVS_BIN, TC_BIN
def get_net_client(use_ovs, run):
"""
Retrieve desired net client for running network commands.
:param bool use_ovs: True for OVS bridges, False for Linux bridges
:param func run: function used to run net client commands
:return: net client class
"""
if use_ovs:
return OvsNetClient(run)
else:
return LinuxNetClient(run)
class LinuxNetClient(object):
"""
Client for creating Linux bridges and ip interfaces for nodes.
@ -177,8 +191,8 @@ class LinuxNetClient(object):
:param str device: device to add tap to
:param str address: address to add tap for
:param str local: local address to tie to
:param str ttl: time to live value
:param str key: key for tap
:param int ttl: time to live value
:param int key: key for tap
:return: nothing
"""
cmd = "%s link add %s type gretap remote %s" % (IP_BIN, device, address)

61
daemon/core/nodes/network.py
View file

@ -10,13 +10,13 @@ from socket import AF_INET, AF_INET6
from core import utils
from core.constants import EBTABLES_BIN, TC_BIN
from core.emulator import distributed
from core.emulator.data import LinkData
from core.emulator.enumerations import LinkTypes, NodeTypes, RegisterTlvs
from core.errors import CoreCommandError, CoreError
from core.nodes import ipaddress
from core.nodes.base import CoreNetworkBase
from core.nodes.interface import GreTap, Veth
from core.nodes.netclient import get_net_client
ebtables_lock = threading.Lock()
@ -257,8 +257,8 @@ class CoreNetwork(CoreNetworkBase):
:param int _id: object id
:param str name: object name
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:param policy: network policy
"""
CoreNetworkBase.__init__(self, session, _id, name, start, server)
@ -289,9 +289,7 @@ class CoreNetwork(CoreNetworkBase):
"""
logging.info("network node(%s) cmd", self.name)
output = utils.check_cmd(args, env, cwd, wait)
for server in self.session.servers:
conn = self.session.servers[server]
distributed.remote_cmd(conn, args, env, cwd, wait)
self.session.distributed.execute(lambda x: x.remote_cmd(args, env, cwd, wait))
return output
def startup(self):
@ -559,7 +557,7 @@ class CoreNetwork(CoreNetworkBase):
if len(name) >= 16:
raise ValueError("interface name %s too long" % name)
netif = Veth(node=None, name=name, localname=localname, mtu=1500, start=self.up)
netif = Veth(self.session, None, name, localname, start=self.up)
self.attach(netif)
if net.up:
# this is similar to net.attach() but uses netif.name instead of localname
@ -632,8 +630,8 @@ class GreTapBridge(CoreNetwork):
:param ttl: ttl value
:param key: gre tap key
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
"""
CoreNetwork.__init__(self, session, _id, name, False, server, policy)
self.grekey = key
@ -753,8 +751,8 @@ class CtrlNet(CoreNetwork):
:param prefix: control network ipv4 prefix
:param hostid: host id
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:param str assign_address: assigned address
:param str updown_script: updown script
:param serverintf: server interface
@ -767,6 +765,25 @@ class CtrlNet(CoreNetwork):
self.serverintf = serverintf
CoreNetwork.__init__(self, session, _id, name, start, server)
def add_addresses(self, address):
"""
Add addresses used for created control networks,
:param core.nodes.interfaces.IpAddress address: starting address to use
:return:
"""
use_ovs = self.session.options.get_config("ovs") == "True"
current = "%s/%s" % (address, self.prefix.prefixlen)
net_client = get_net_client(use_ovs, utils.check_cmd)
net_client.create_address(self.brname, current)
servers = self.session.distributed.servers
for name in servers:
server = servers[name]
address -= 1
current = "%s/%s" % (address, self.prefix.prefixlen)
net_client = get_net_client(use_ovs, server.remote_cmd)
net_client.create_address(self.brname, current)
def startup(self):
"""
Startup functionality for the control network.
@ -779,16 +796,14 @@ class CtrlNet(CoreNetwork):
CoreNetwork.startup(self)
if self.hostid:
addr = self.prefix.addr(self.hostid)
else:
addr = self.prefix.max_addr()
logging.info("added control network bridge: %s %s", self.brname, self.prefix)
if self.assign_address:
addrlist = ["%s/%s" % (addr, self.prefix.prefixlen)]
self.addrconfig(addrlist=addrlist)
if self.hostid and self.assign_address:
address = self.prefix.addr(self.hostid)
self.add_addresses(address)
elif self.assign_address:
address = self.prefix.max_addr()
self.add_addresses(address)
if self.updown_script:
logging.info(
@ -1008,8 +1023,8 @@ class HubNode(CoreNetwork):
:param int _id: node id
:param str name: node namee
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:raises CoreCommandError: when there is a command exception
"""
CoreNetwork.__init__(self, session, _id, name, start, server)
@ -1039,8 +1054,8 @@ class WlanNode(CoreNetwork):
:param int _id: node id
:param str name: node name
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:param policy: wlan policy
"""
CoreNetwork.__init__(self, session, _id, name, start, server, policy)

48
daemon/core/nodes/physical.py
View file

@ -9,15 +9,19 @@ import threading
from core import utils
from core.constants import MOUNT_BIN, UMOUNT_BIN
from core.emulator.enumerations import NodeTypes
from core.errors import CoreCommandError
from core.errors import CoreCommandError, CoreError
from core.nodes.base import CoreNodeBase
from core.nodes.interface import CoreInterface
from core.nodes.network import CoreNetwork, GreTap
class PhysicalNode(CoreNodeBase):
def __init__(self, session, _id=None, name=None, nodedir=None, start=True):
CoreNodeBase.__init__(self, session, _id, name, start=start)
def __init__(
self, session, _id=None, name=None, nodedir=None, start=True, server=None
):
CoreNodeBase.__init__(self, session, _id, name, start, server)
if not self.server:
raise CoreError("physical nodes must be assigned to a remote server")
self.nodedir = nodedir
self.up = start
self.lock = threading.RLock()
@ -86,7 +90,6 @@ class PhysicalNode(CoreNodeBase):
def adoptnetif(self, netif, ifindex, hwaddr, addrlist):
"""
The broker builds a GreTap tunnel device to this physical node.
When a link message is received linking this node to another part of
the emulation, no new interface is created; instead, adopt the
GreTap netif as the node interface.
@ -157,26 +160,20 @@ class PhysicalNode(CoreNodeBase):
if ifindex is None:
ifindex = self.newifindex()
if self.up:
# this is reached when this node is linked to a network node
# tunnel to net not built yet, so build it now and adopt it
gt = self.session.broker.addnettunnel(net.id)
if gt is None or len(gt) != 1:
raise ValueError(
"error building tunnel from adding a new network interface: %s" % gt
)
gt = gt[0]
net.detach(gt)
self.adoptnetif(gt, ifindex, hwaddr, addrlist)
return ifindex
# this is reached when configuring services (self.up=False)
if ifname is None:
ifname = "gt%d" % ifindex
netif = GreTap(node=self, name=ifname, session=self.session, start=False)
self.adoptnetif(netif, ifindex, hwaddr, addrlist)
return ifindex
if self.up:
# this is reached when this node is linked to a network node
# tunnel to net not built yet, so build it now and adopt it
_, remote_tap = self.session.distributed.create_gre_tunnel(net, self.server)
self.adoptnetif(remote_tap, ifindex, hwaddr, addrlist)
return ifindex
else:
# this is reached when configuring services (self.up=False)
netif = GreTap(node=self, name=ifname, session=self.session, start=False)
self.adoptnetif(netif, ifindex, hwaddr, addrlist)
return ifindex
def privatedir(self, path):
if path[0] != "/":
@ -223,6 +220,9 @@ class PhysicalNode(CoreNodeBase):
os.chmod(node_file.name, mode)
logging.info("created nodefile: '%s'; mode: 0%o", node_file.name, mode)
def node_net_cmd(self, args, wait=True):
return self.net_cmd(args, wait=wait)
class Rj45Node(CoreNodeBase, CoreInterface):
"""
@ -242,11 +242,11 @@ class Rj45Node(CoreNodeBase, CoreInterface):
:param str name: node name
:param mtu: rj45 mtu
:param bool start: start flag
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
"""
CoreNodeBase.__init__(self, session, _id, name, start, server)
CoreInterface.__init__(self, node=self, name=name, mtu=mtu)
CoreInterface.__init__(self, session, self, name, mtu, server)
self.up = False
self.lock = threading.RLock()
self.ifindex = None

1
daemon/core/plugins/sdt.py
View file

@ -76,7 +76,6 @@ class Sdt(object):
# node information for remote nodes not in session._objs
# local nodes also appear here since their obj may not exist yet
self.remotes = {}
session.broker.handlers.add(self.handle_distributed)
# add handler for node updates
self.session.node_handlers.append(self.handle_node_update)

4
daemon/core/xml/corexmldeployment.py
View file

@ -107,10 +107,6 @@ class CoreXmlDeployment(object):
def add_deployment(self):
physical_host = self.add_physical_host(socket.gethostname())
# TODO: handle other servers
# servers = self.session.broker.getservernames()
# servers.remove("localhost")
for node_id in self.session.nodes:
node = self.session.nodes[node_id]
if isinstance(node, CoreNodeBase):

47
daemon/core/xml/emanexml.py
View file

@ -5,7 +5,6 @@ from tempfile import NamedTemporaryFile
from lxml import etree
from core import utils
from core.emulator import distributed
from core.nodes.ipaddress import MacAddress
from core.xml import corexml
@ -53,8 +52,8 @@ def create_file(xml_element, doc_name, file_path, server=None):
:param lxml.etree.Element xml_element: root element to write to file
:param str doc_name: name to use in the emane doctype
:param str file_path: file path to write xml file to
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:return: nothing
"""
doctype = (
@ -65,7 +64,7 @@ def create_file(xml_element, doc_name, file_path, server=None):
temp = NamedTemporaryFile(delete=False)
create_file(xml_element, doc_name, temp.name)
temp.close()
distributed.remote_put(server, temp.name, file_path)
server.remote_put(temp.name, file_path)
os.unlink(temp.name)
else:
corexml.write_xml_file(xml_element, file_path, doctype=doctype)
@ -315,9 +314,9 @@ def build_transport_xml(emane_manager, node, transport_type):
file_name = transport_file_name(node.id, transport_type)
file_path = os.path.join(emane_manager.session.session_dir, file_name)
create_file(transport_element, doc_name, file_path)
for server in emane_manager.session.servers:
conn = emane_manager.session.servers[server]
create_file(transport_element, doc_name, file_path, conn)
emane_manager.session.distributed.execute(
lambda x: create_file(transport_element, doc_name, file_path, x)
)
def create_phy_xml(emane_model, config, file_path, server):
@ -327,8 +326,8 @@ def create_phy_xml(emane_model, config, file_path, server):
:param core.emane.emanemodel.EmaneModel emane_model: emane model to create xml
:param dict config: all current configuration values
:param str file_path: path to write file to
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:return: nothing
"""
phy_element = etree.Element("phy", name="%s PHY" % emane_model.name)
@ -343,9 +342,9 @@ def create_phy_xml(emane_model, config, file_path, server):
create_file(phy_element, "phy", file_path, server)
else:
create_file(phy_element, "phy", file_path)
for server in emane_model.session.servers:
conn = emane_model.session.servers[server]
create_file(phy_element, "phy", file_path, conn)
emane_model.session.distributed.execute(
lambda x: create_file(phy_element, "phy", file_path, x)
)
def create_mac_xml(emane_model, config, file_path, server):
@ -355,8 +354,8 @@ def create_mac_xml(emane_model, config, file_path, server):
:param core.emane.emanemodel.EmaneModel emane_model: emane model to create xml
:param dict config: all current configuration values
:param str file_path: path to write file to
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:return: nothing
"""
if not emane_model.mac_library:
@ -373,9 +372,9 @@ def create_mac_xml(emane_model, config, file_path, server):
create_file(mac_element, "mac", file_path, server)
else:
create_file(mac_element, "mac", file_path)
for server in emane_model.session.servers:
conn = emane_model.session.servers[server]
create_file(mac_element, "mac", file_path, conn)
emane_model.session.distributed.execute(
lambda x: create_file(mac_element, "mac", file_path, x)
)
def create_nem_xml(
@ -396,8 +395,8 @@ def create_nem_xml(
:param str transport_definition: transport file definition path
:param str mac_definition: mac file definition path
:param str phy_definition: phy file definition path
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:return: nothing
"""
nem_element = etree.Element("nem", name="%s NEM" % emane_model.name)
@ -411,9 +410,9 @@ def create_nem_xml(
create_file(nem_element, "nem", nem_file, server)
else:
create_file(nem_element, "nem", nem_file)
for server in emane_model.session.servers:
conn = emane_model.session.servers[server]
create_file(nem_element, "nem", nem_file, conn)
emane_model.session.distributed.execute(
lambda x: create_file(nem_element, "nem", nem_file, x)
)
def create_event_service_xml(group, port, device, file_directory, server=None):
@ -424,8 +423,8 @@ def create_event_service_xml(group, port, device, file_directory, server=None):
:param str port: event port
:param str device: event device
:param str file_directory: directory to create file in
:param fabric.connection.Connection server: remote server node will run on,
default is None for localhost
:param core.emulator.distributed.DistributedServer server: remote server node
will run on, default is None for localhost
:return: nothing
"""
event_element = etree.Element("emaneeventmsgsvc")

4
daemon/examples/lxd/lxd2lxd.py
View file

@ -5,7 +5,7 @@ from core.emulator.emudata import IpPrefixes, NodeOptions
from core.emulator.enumerations import EventTypes, NodeTypes
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
logging.basicConfig(level=logging.INFO)
coreemu = CoreEmu()
session = coreemu.create_session()
@ -14,7 +14,7 @@ if __name__ == "__main__":
# create nodes and interfaces
try:
prefixes = IpPrefixes(ip4_prefix="10.83.0.0/16")
options = NodeOptions(image="ubuntu")
options = NodeOptions(image="ubuntu:18.04")
# create node one
node_one = session.add_node(_type=NodeTypes.LXC, node_options=options)

16
daemon/examples/python/distributed.py
View file

@ -8,21 +8,19 @@ from core.emulator.enumerations import EventTypes, NodeTypes
def main():
address = sys.argv[1]
remote = sys.argv[2]
# ip generator for example
prefixes = IpPrefixes(ip4_prefix="10.83.0.0/16")
# create emulator instance for creating sessions and utility methods
coreemu = CoreEmu()
coreemu = CoreEmu({"controlnet": "172.16.0.0/24", "distributed_address": address})
session = coreemu.create_session()
# set controlnet
session.options.set_config("controlnet", "172.16.0.0/24")
# initialize distributed
address = sys.argv[1]
remote = sys.argv[2]
session.address = address
session.add_distributed(remote)
server_name = "core2"
session.distributed.add_server(server_name, remote)
# must be in configuration state for nodes to start, when using "node_add" below
session.set_state(EventTypes.CONFIGURATION_STATE)
@ -31,7 +29,7 @@ def main():
node_one = session.add_node()
switch = session.add_node(_type=NodeTypes.SWITCH)
options = NodeOptions()
options.emulation_server = remote
options.emulation_server = server_name
node_two = session.add_node(node_options=options)
# create node interfaces and link

26
daemon/examples/python/distributed_emane.py
View file

@ -9,25 +9,25 @@ from core.emulator.enumerations import EventTypes, NodeTypes
def main():
address = sys.argv[1]
remote = sys.argv[2]
# ip generator for example
prefixes = IpPrefixes(ip4_prefix="10.83.0.0/16")
# create emulator instance for creating sessions and utility methods
coreemu = CoreEmu()
coreemu = CoreEmu(
{
"controlnet": "core1:172.16.1.0/24 core2:172.16.2.0/24 core3:172.16.3.0/24 "
"core4:172.16.4.0/24 core5:172.16.5.0/24",
"distributed_address": address,
}
)
session = coreemu.create_session()
# set controlnet
session.options.set_config(
"controlnet",
"core1:172.16.1.0/24 core2:172.16.2.0/24 core3:172.16.3.0/24 "
"core4:172.16.4.0/24 core5:172.16.5.0/24",
)
# initialize distributed
address = sys.argv[1]
remote = sys.argv[2]
session.address = address
session.add_distributed(remote)
server_name = "core2"
session.distributed.add_server(server_name, remote)
# must be in configuration state for nodes to start, when using "node_add" below
session.set_state(EventTypes.CONFIGURATION_STATE)
@ -38,7 +38,7 @@ def main():
node_one = session.add_node(node_options=options)
emane_net = session.add_node(_type=NodeTypes.EMANE)
session.emane.set_model(emane_net, EmaneIeee80211abgModel)
options.emulation_server = remote
options.emulation_server = server_name
node_two = session.add_node(node_options=options)
# create node interfaces and link

51
daemon/examples/python/distributed_lxd.py Normal file
View file

@ -0,0 +1,51 @@
import logging
import pdb
import sys
from core.emulator.coreemu import CoreEmu
from core.emulator.emudata import IpPrefixes, NodeOptions
from core.emulator.enumerations import EventTypes, NodeTypes
def main():
address = sys.argv[1]
remote = sys.argv[2]
# ip generator for example
prefixes = IpPrefixes(ip4_prefix="10.83.0.0/16")
# create emulator instance for creating sessions and utility methods
coreemu = CoreEmu({"distributed_address": address})
session = coreemu.create_session()
# initialize distributed
server_name = "core2"
session.distributed.add_server(server_name, remote)
# must be in configuration state for nodes to start, when using "node_add" below
session.set_state(EventTypes.CONFIGURATION_STATE)
# create local node, switch, and remote nodes
options = NodeOptions(image="ubuntu:18.04")
node_one = session.add_node(_type=NodeTypes.LXC, node_options=options)
options.emulation_server = server_name
node_two = session.add_node(_type=NodeTypes.LXC, node_options=options)
# create node interfaces and link
interface_one = prefixes.create_interface(node_one)
interface_two = prefixes.create_interface(node_two)
session.add_link(node_one.id, node_two.id, interface_one, interface_two)
# instantiate session
session.instantiate()
# pause script for verification
pdb.set_trace()
# shutdown session
coreemu.shutdown()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
main()

13
daemon/examples/python/distributed_ptp.py
View file

@ -8,18 +8,19 @@ from core.emulator.enumerations import EventTypes
def main():
address = sys.argv[1]
remote = sys.argv[2]
# ip generator for example
prefixes = IpPrefixes(ip4_prefix="10.83.0.0/16")
# create emulator instance for creating sessions and utility methods
coreemu = CoreEmu()
coreemu = CoreEmu({"distributed_address": address})
session = coreemu.create_session()
# initialize distributed
address = sys.argv[1]
remote = sys.argv[2]
session.address = address
session.add_distributed(remote)
server_name = "core2"
session.distributed.add_server(server_name, remote)
# must be in configuration state for nodes to start, when using "node_add" below
session.set_state(EventTypes.CONFIGURATION_STATE)
@ -27,7 +28,7 @@ def main():
# create local node, switch, and remote nodes
options = NodeOptions()
node_one = session.add_node(node_options=options)
options.emulation_server = remote
options.emulation_server = server_name
node_two = session.add_node(node_options=options)
# create node interfaces and link

11
daemon/examples/python/distributed_switches.py
View file

@ -7,15 +7,16 @@ from core.emulator.enumerations import EventTypes, NodeTypes
def main():
address = sys.argv[1]
remote = sys.argv[2]
# create emulator instance for creating sessions and utility methods
coreemu = CoreEmu()
coreemu = CoreEmu({"distributed_address": address})
session = coreemu.create_session()
# initialize distributed
address = sys.argv[1]
remote = sys.argv[2]
session.address = address
session.add_distributed(remote)
server_name = "core2"
session.distributed.add_server(server_name, remote)
# must be in configuration state for nodes to start, when using "node_add" below
session.set_state(EventTypes.CONFIGURATION_STATE)

16
daemon/examples/python/distributed_wlan.py
View file

@ -9,21 +9,19 @@ from core.location.mobility import BasicRangeModel
def main():
address = sys.argv[1]
remote = sys.argv[2]
# ip generator for example
prefixes = IpPrefixes(ip4_prefix="10.83.0.0/16")
# create emulator instance for creating sessions and utility methods
coreemu = CoreEmu()
coreemu = CoreEmu({"distributed_address": address})
session = coreemu.create_session()
# set controlnet
# session.options.set_config("controlnet", "172.16.0.0/24")
# initialize distributed
address = sys.argv[1]
remote = sys.argv[2]
session.address = address
session.add_distributed(remote)
server_name = "core2"
session.distributed.add_server(server_name, remote)
# must be in configuration state for nodes to start, when using "node_add" below
session.set_state(EventTypes.CONFIGURATION_STATE)
@ -31,7 +29,7 @@ def main():
# create local node, switch, and remote nodes
options = NodeOptions()
options.set_position(0, 0)
options.emulation_server = remote
options.emulation_server = server_name
node_one = session.add_node(node_options=options)
wlan = session.add_node(_type=NodeTypes.WIRELESS_LAN)
session.mobility.set_model(wlan, BasicRangeModel)

2
daemon/requirements.txt
View file

@ -1,7 +1,9 @@
configparser==4.0.2
fabric==2.5.0
future==0.17.1
grpcio==1.23.0
grpcio-tools==1.21.1
invoke==1.3.0
lxml==4.4.1
protobuf==3.9.1
six==1.12.0

2
daemon/setup.py.in
View file

@ -35,8 +35,10 @@ setup(
packages=find_packages(),
install_requires=[
"configparser",
"fabric",
"future",
"grpcio",
"invoke",
"lxml",
"protobuf",
],

1
daemon/tests/conftest.py
View file

@ -58,7 +58,6 @@ class CoreServerTest(object):
self.request_handler = CoreHandler(request_mock, "", self.server)
self.request_handler.session = self.session
self.request_handler.add_session_handlers()
self.session.broker.session_clients.append(self.request_handler)
# have broker handle a configuration state change
self.session.set_state(EventTypes.DEFINITION_STATE)

6
daemon/tests/test_gui.py
View file

@ -763,13 +763,11 @@ class TestGui:
(ConfigTlvs.VALUES, "%s:%s:%s" % (server, host, port)),
],
)
coreserver.session.broker.addserver = mock.MagicMock()
coreserver.session.broker.setupserver = mock.MagicMock()
coreserver.session.distributed.add_server = mock.MagicMock()
coreserver.request_handler.handle_message(message)
coreserver.session.broker.addserver.assert_called_once_with(server, host, port)
coreserver.session.broker.setupserver.assert_called_once_with(server)
coreserver.session.distributed.add_server.assert_called_once_with(server, host)
def test_config_services_request_all(self, coreserver):
message = coreapi.CoreConfMessage.create(

2
gui/nodes.tcl
View file

@ -19,7 +19,7 @@ array set g_node_types_default {
4 {mdr mdr.gif mdr.gif {zebra OSPFv3MDR IPForward} \
netns {built-in type for wireless routers}}
5 {prouter router_green.gif router_green.gif \
{zebra OSPFv2 OSPFv3 IPForward} \
{} \
physical {built-in type for physical nodes}}
}