aboutsummaryrefslogtreecommitdiff
path: root/rumba
diff options
context:
space:
mode:
Diffstat (limited to 'rumba')
-rw-r--r--rumba/log.py7
-rw-r--r--rumba/model.py220
-rw-r--r--rumba/recpoisson.py65
-rw-r--r--rumba/ssh_support.py12
-rw-r--r--rumba/testbeds/emulab.py2
-rw-r--r--rumba/testbeds/jfed.py14
-rw-r--r--rumba/testbeds/qemu.py2
7 files changed, 285 insertions, 37 deletions
diff --git a/rumba/log.py b/rumba/log.py
index bed0170..c509532 100644
--- a/rumba/log.py
+++ b/rumba/log.py
@@ -24,6 +24,13 @@ import sys
import multiprocessing
+DEBUG = logging.DEBUG
+INFO = logging.INFO
+WARNING = logging.WARNING
+ERROR = logging.ERROR
+CRITICAL = logging.CRITICAL
+
+
loggers_set = set()
diff --git a/rumba/model.py b/rumba/model.py
index f4f98d1..ffb67cc 100644
--- a/rumba/model.py
+++ b/rumba/model.py
@@ -20,13 +20,29 @@
# MA 02110-1301 USA
import abc
-
-import rumba.log as log
import os
+import random
import stat
+import time
+
+import rumba.log as log
+from rumba import ssh_support
logger = log.get_logger(__name__)
+try:
+ from numpy.random import poisson
+ from numpy.random import exponential
+ logger.debug("Using numpy for faster and better random variables.")
+except ImportError:
+ from rumba.recpoisson import poisson
+
+ def exponential(mean_duration):
+ return random.expovariate(1.0 / mean_duration)
+
+ logger.debug("Falling back to simple implementations.")
+ # PROBLEM! These logs will almost never be printed... But we might not care
+
tmp_dir = '/tmp/rumba'
try:
os.mkdir(tmp_dir)
@@ -42,6 +58,7 @@ except OSError:
# Already there, nothing to do
pass
+
# Represents generic testbed info
#
# @username [string] user name
@@ -164,10 +181,18 @@ class NormalDIF(DIF):
# SSH Configuration
#
class SSHConfig:
- def __init__(self, hostname, port=22, proxycommand=None):
+ def __init__(self, hostname, port=22, proxy_command=None):
+ self.username = None
+ self.password = None
self.hostname = hostname
self.port = port
- self.proxycommand = proxycommand
+ self.proxy_command = proxy_command
+
+ def set_username(self, username):
+ self.username = username
+
+ def set_password(self, password):
+ self.password = password
# A node in the experiment
@@ -282,6 +307,74 @@ class Node:
def get_policy(self, dif):
return self.policies[dif]
+ def execute_commands(self, commands, time_out=3, use_proxy=False):
+ # Ssh_config is used twice since it doubles as testbed info
+ # (it holds fields username and password)
+ if use_proxy:
+ return ssh_support.execute_proxy_commands(
+ self.ssh_config,
+ self.ssh_config,
+ commands,
+ time_out
+ )
+ # else:
+ return ssh_support.execute_commands(
+ self.ssh_config,
+ self.ssh_config,
+ commands,
+ time_out
+ )
+
+ def execute_command(self, command, time_out=3, use_proxy=False):
+ # Ssh_config is used twice since it doubles as testbed info
+ # (it holds fields username and password)
+ if use_proxy:
+ return ssh_support.execute_proxy_command(
+ self.ssh_config,
+ self.ssh_config,
+ command,
+ time_out
+ )
+ # else:
+ return ssh_support.execute_command(
+ self.ssh_config,
+ self.ssh_config,
+ command,
+ time_out
+ )
+
+ def copy_file_to_testbed(self, text, file_name):
+ ssh_support.copy_file_to_testbed(
+ self.ssh_config,
+ self.ssh_config,
+ text,
+ file_name
+ )
+
+ def copy_path_to_testbed(self, path, destination):
+ ssh_support.copy_path_to_testbed(
+ self.ssh_config,
+ self.ssh_config,
+ path,
+ destination
+ )
+
+ def copy_paths_to_testbed(self, paths, destination):
+ ssh_support.copy_paths_to_testbed(
+ self.ssh_config,
+ self.ssh_config,
+ paths,
+ destination
+ )
+
+ def setup_vlan(self, vlan_id, int_name):
+ ssh_support.setup_vlan(
+ self.ssh_config,
+ self.ssh_config,
+ vlan_id,
+ int_name
+ )
+
# Base class representing an IPC Process to be created in the experiment
#
@@ -647,7 +740,7 @@ class Experiment:
self.testbed.username,
node.ssh_config.hostname,
node.ssh_config.port,
- node.ssh_config.proxycommand))
+ node.ssh_config.proxy_command))
f.close()
# Examine the nodes and DIFs, compute the registration and enrollment
@@ -692,38 +785,57 @@ class Client(object):
self.ap = ap
self.options = options
- def start_process(self, node, duration, start_time):
- return ClientProcess(self.ap, node, duration, start_time, self.options)
+ def start_process(self, duration):
+ return ClientProcess(self.ap, duration, self.options)
# Base class for client processes
#
# @ap: Application Process binary
-# @node: The node on which this process should run
# @duration: The time (in seconds) this process should run
# @start_time: The time at which this process is started.
# @options: Options to pass to the binary
#
class ClientProcess(Client):
- def __init__(self, ap, node, duration, start_time, options=None):
+ def __init__(self, ap, duration, options=None):
super(ClientProcess, self).__init__(ap, options=options)
- self.node = node
self.duration = duration
- self.start_time = start_time
- self.run()
- self.running = True
+ self.start_time = None
+ self.running = False
+ self.node = None
+ self.pid = None
- def run(self):
- pass # TODO to be implemented
+ def run(self, node):
+ self.node = node
+ self.start_time = time.time()
- def stop(self):
- pass # TODO to be implemented
+ logger.debug(
+ 'Starting client app %s on node %s with duration %s.',
+ self.ap, self.node.name, self.duration
+ )
+
+ opt_str = self.options if self.options is not None else ""
+ cmd = "./startup.sh %s %s" % (self.ap, opt_str)
+ self.running = True
+ self.pid = self.node.execute_command(cmd)
- def check(self, now):
+ def stop(self):
+ logger.debug(
+ 'Killing client %s on node %s.',
+ self.ap, self.node.name
+ )
+ self.node.execute_command("kill %s" % self.pid)
+
+ def check(self):
+ """Check if the process should keep running, stop it if not,
+ and return true if and only if it is still running."""
+ now = time.time()
if not self.running:
- return
+ return False
if now - self.start_time >= self.duration:
self.stop()
+ return False
+ return True
# Base class for server programs
@@ -738,7 +850,7 @@ class ClientProcess(Client):
#
class Server:
def __init__(self, ap, arrival_rate, mean_duration,
- options=None, max_clients=None,
+ options=None, max_clients=float('inf'),
clients=None, nodes=None):
self.ap = ap
self.options = options
@@ -749,6 +861,7 @@ class Server:
self.nodes = nodes
self.arrival_rate = arrival_rate # mean requests/s
self.mean_duration = mean_duration # in seconds
+ self.pids = {}
def add_client(self, client):
self.clients.append(client)
@@ -770,28 +883,63 @@ class Server:
interval seconds.
Hence, the average size should be interval * arrival_rate.
"""
- pass
+ # WARNING! using numpy. To be discussed.
+ number = poisson(self.arrival_rate * interval)
+ number = int(min(number, self.max_clients))
+ l = [self.make_client_process() for _ in range(number)]
+ return l
def make_client_process(self):
"""Returns a client of this server"""
if len(self.clients) == 0:
- raise Exception("Server %s has empty client list," % (self,))
- pass # TODO should return a ClientProcess
+ raise Exception("Server %s has empty client list." % (self,))
+ duration = exponential(self.mean_duration)
+ return random.choice(self.clients)\
+ .start_process(duration=duration)
+
+ def run(self):
+ for node in self.nodes:
+ opt_str = self.options if self.options is not None else ""
+ logfile = "%s.log" % self.ap
+ script = r'nohup "$@" > %s & echo "$!"' % logfile
+ cmds = ["echo '%s' > startup.sh && chmod a+x startup.sh"
+ % (script,),
+ "./startup.sh %s %s" % (self.ap, opt_str)]
+ logger.debug(
+ 'Starting server %s on node %s with logfile %s.',
+ self.ap, node.name, logfile
+ )
+ self.pids[node] = (node.execute_commands(cmds))
+
+ def stop(self):
+ for node, pid in self.pids.items():
+ logger.debug(
+ 'Killing server %s on node %s.',
+ self.ap, node.name
+ )
+ node.execute_command("kill %s" % pid)
# Base class for ARCFIRE storyboards
#
# @experiment: Experiment to use as input
# @duration: Duration of the whole storyboard
+# @testbed: The testbed the experiment is run on.
# @servers: App servers available in the network
#
class StoryBoard:
+
+ DEFAULT_INTERVAL = 2.5 # in seconds (may be a float)
+
def __init__(self, experiment, duration, servers=None):
self.experiment = experiment
self.duration = duration
if servers is None:
servers = list()
self.servers = servers
+ self.client_nodes = [c for c in experiment.nodes if c.client]
+ self.active_clients = []
+ self.start_time = None
def add_server(self, server):
self.servers.append(server)
@@ -800,4 +948,28 @@ class StoryBoard:
self.servers.remove(server)
def start(self):
- pass
+ self.start_time = time.time()
+ script = r'nohup "$@" > /dev/null & echo "$!"'
+ for node in self.client_nodes:
+ logger.debug("Writing utility startup script on client nodes.")
+ node.execute_command(
+ "echo '%s' > startup.sh && chmod a+x startup.sh" % (script,)
+ )
+ try:
+ for server in self.servers:
+ server.run()
+ while time.time() - self.start_time < self.duration:
+ for server in self.servers:
+ clients = server.get_new_clients(self.DEFAULT_INTERVAL)
+ for new_client in clients:
+ client_node = random.choice(self.client_nodes)
+ new_client.run(client_node)
+ self.active_clients.append(new_client)
+ self.active_clients = \
+ [x for x in self.active_clients if x.check()]
+ time.sleep(self.DEFAULT_INTERVAL)
+ finally:
+ for client in self.active_clients:
+ client.stop()
+ for server in self.servers:
+ server.stop()
diff --git a/rumba/recpoisson.py b/rumba/recpoisson.py
new file mode 100644
index 0000000..3c1e6fe
--- /dev/null
+++ b/rumba/recpoisson.py
@@ -0,0 +1,65 @@
+import math
+import random
+
+import sys
+
+if sys.version_info < (3, 2):
+ from repoze.lru import lru_cache
+ # from functools32 import lru_cache
+else:
+ from functools import lru_cache
+
+
+@lru_cache(1000)
+def _get_poisson_var(parameter):
+ return Poisson(parameter)
+
+
+class Poisson(object):
+
+ def __init__(self, parameter):
+ self.parameter = parameter
+
+ def c_p(k):
+ """Compute the Poisson CDF for k iteratively."""
+ if k == 0:
+ return self._p(0)
+ else:
+ return self._compute_poisson_cdf(k - 1) + self._p(k)
+ self._compute_poisson_cdf = lru_cache(int(2.5*self.parameter) + 1)(c_p)
+
+ @staticmethod
+ def _get_random():
+ return random.random()
+
+ def _p(self, k):
+ # l^k * e^-l / k!
+ # Computed as exp(klog(l) - l - log(k!))
+ l = self.parameter
+ l_to_the_k = k * math.log(l)
+ k_fact = sum([math.log(i + 1) for i in range(k)])
+ return math.exp(l_to_the_k - l - k_fact)
+
+ def sample(self):
+ # The idea is:
+ # take a sample from U(0,1) and call it f.
+ # Let x be s.t. x = min_N F(x) > f
+ # where F is the cumulative distribution function of Poisson(parameter)
+ # return x
+ f = self._get_random()
+
+ # We compute x iteratively by computing
+ # \sum_k(P=k)
+ # where P ~ Poisson(parameter) and stopping as soon as
+ # it is greater than f.
+ # We use the cache to store results.
+ current_cdf = -1
+ current_x = -1
+ while current_cdf < f:
+ current_x += 1
+ current_cdf = self._compute_poisson_cdf(current_x)
+ return current_x
+
+
+def poisson(parameter):
+ return _get_poisson_var(parameter).sample()
diff --git a/rumba/ssh_support.py b/rumba/ssh_support.py
index b0970e1..e785f33 100644
--- a/rumba/ssh_support.py
+++ b/rumba/ssh_support.py
@@ -102,8 +102,8 @@ def execute_commands(testbed, ssh_config, commands, time_out=3):
"""
ssh_client = get_ssh_client()
- if ssh_config.proxycommand is not None:
- proxy = paramiko.ProxyCommand(ssh_config.proxycommand)
+ if ssh_config.proxy_command is not None:
+ proxy = paramiko.ProxyCommand(ssh_config.proxy_command)
else:
proxy = None
@@ -162,8 +162,8 @@ def copy_file_to_testbed(testbed, ssh_config, text, file_name):
"""
ssh_client = get_ssh_client()
- if ssh_config.proxycommand is not None:
- proxy = paramiko.ProxyCommand(ssh_config.proxycommand)
+ if ssh_config.proxy_command is not None:
+ proxy = paramiko.ProxyCommand(ssh_config.proxy_command)
else:
proxy = None
@@ -205,8 +205,8 @@ def copy_paths_to_testbed(testbed, ssh_config, paths, destination):
"""
ssh_client = get_ssh_client()
- if ssh_config.proxycommand is not None:
- proxy = paramiko.ProxyCommand(ssh_config.proxycommand)
+ if ssh_config.proxy_command is not None:
+ proxy = paramiko.ProxyCommand(ssh_config.proxy_command)
else:
proxy = None
diff --git a/rumba/testbeds/emulab.py b/rumba/testbeds/emulab.py
index e7458bc..9b90e68 100644
--- a/rumba/testbeds/emulab.py
+++ b/rumba/testbeds/emulab.py
@@ -210,6 +210,8 @@ class Testbed(mod.Testbed):
for node in experiment.nodes:
node.ssh_config.hostname = self.full_name(node.name)
+ node.ssh_config.set_username(self.username)
+ node.ssh_config.set_password(self.password)
cmd = 'cat /var/emulab/boot/topomap'
topomap = ssh.execute_command(self, experiment.nodes[0].ssh_config, cmd)
diff --git a/rumba/testbeds/jfed.py b/rumba/testbeds/jfed.py
index 9c72ca7..8867dc6 100644
--- a/rumba/testbeds/jfed.py
+++ b/rumba/testbeds/jfed.py
@@ -155,11 +155,13 @@ class Testbed(mod.Testbed):
node.ssh_config.hostname = \
node.name + "." + self.exp_name + "." + \
auth_name_r + "." + self.auth_name
- node.ssh_config.proxycommand = "ssh -i '" + self.cert_file + \
- "' -o StrictHostKeyChecking=no " + \
- self.username + \
- "@bastion.test.iminds.be nc " + \
- node.ssh_config.hostname + " 22"
+ node.ssh_config.proxy_command = "ssh -i '" + self.cert_file + \
+ "' -o StrictHostKeyChecking=no " + \
+ self.username + \
+ "@bastion.test.iminds.be nc " + \
+ node.ssh_config.hostname + " 22"
+ node.ssh_config.username = self.username
+ node.ssh_config.password = self.password
subprocess.call(["java", "-jar", self.jfed_jar, "create", "-S",
self.proj_name, "--rspec",
@@ -174,8 +176,6 @@ class Testbed(mod.Testbed):
rspec = xml.parse(self.manifest)
xml_nodes = rspec.getElementsByTagName("node")
- dir_path = os.path.dirname(os.path.abspath(__file__))
-
# Complete details of the nodes after swapin
logger.info("Sleeping for two seconds to avoid contacting jfed nodes "
"too soon.")
diff --git a/rumba/testbeds/qemu.py b/rumba/testbeds/qemu.py
index f0b73a8..3d30ce2 100644
--- a/rumba/testbeds/qemu.py
+++ b/rumba/testbeds/qemu.py
@@ -254,6 +254,8 @@ class Testbed(mod.Testbed):
vm['id'] = vmid
node.ssh_config.hostname = "localhost"
node.ssh_config.port = fwdp
+ node.ssh_config.username = self.username
+ node.ssh_config.password = self.password
log_file = os.path.join(mod.tmp_dir, name + '.log')
vars_dict = {'fwdp': fwdp, 'id': vmid, 'mac': mac,
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-03 06:21:02 +0000