From 0605d6d3df64caa9fafa3dc2311551ce15292abd Mon Sep 17 00:00:00 2001
From: Simon Crowle <sgc@it-innovation.soton.ac.uk>
Date: Tue, 27 Mar 2018 09:54:25 +0100
Subject: [PATCH] Refactors simulator based fixtures in conftest and embeds MC
config state into simulation boot sequence
---
clmctest/monitoring/LineProtocolGenerator.py | 45 +++++-
clmctest/monitoring/StreamingSim.py | 143 +++++++++----------
clmctest/monitoring/conftest.py | 36 +++++
clmctest/monitoring/test_MC_Config.py | 38 -----
clmctest/monitoring/test_simresults.py | 8 +-
5 files changed, 152 insertions(+), 118 deletions(-)
delete mode 100644 clmctest/monitoring/test_MC_Config.py
diff --git a/clmctest/monitoring/LineProtocolGenerator.py b/clmctest/monitoring/LineProtocolGenerator.py
index fb54cb2..cb2cacf 100644
--- a/clmctest/monitoring/LineProtocolGenerator.py
+++ b/clmctest/monitoring/LineProtocolGenerator.py
@@ -94,18 +94,51 @@ def _getNSTime(time):
return timestamp
-def generate_mc_serviceConfig( mcMeasurement, sStop, asStop, sStart, asStart, time ):
+def generate_mc_service_config( mcMeasurement, stateTimeStats, time ):
+
+ stateTimeStats = validate_state_time_stats( stateTimeStats )
+
result = [{ "measurement" : mcMeasurement,
- "fields" :
- { "serviceStopped" : sStop,
- "avgServiceStopped" : asStop,
- "serviceStarted" : sStart,
- "avgServiceStarted" : asStart
+ "fields" :
+ { "stopped" : stateTimeStats['stopped'],
+ "avg_stopped" : stateTimeStats['avg_stopped'],
+ "starting" : stateTimeStats['starting'],
+ "avg_starting" : stateTimeStats['avg_starting'],
+ "running" : stateTimeStats['running'],
+ "avg_running" : stateTimeStats['avg_running'],
+ "stopping" : stateTimeStats['stopping'],
+ "avg_stopping" : stateTimeStats['avg_stopping']
},
"time" : _getNSTime(time) }]
return result
+def validate_state_time_stats( stateTimeStats ):
+
+ if ( not 'stopped' in stateTimeStats ):
+ stateTimeStats['stopped'] = 0
+
+ if ( not 'avg_stopped' in stateTimeStats ):
+ stateTimeStats['avg_stopped'] = 0
+
+ if ( not 'starting' in stateTimeStats ):
+ stateTimeStats['starting'] = 0
+
+ if ( not 'avg_starting' in stateTimeStats ):
+ stateTimeStats['avg_starting'] = 0
+
+ if ( not 'running' in stateTimeStats ):
+ stateTimeStats['running'] = 0
+
+ if ( not 'stopping' in stateTimeStats ):
+ stateTimeStats['stopping'] = 0
+
+ if ( not 'avg_stopping' in stateTimeStats ):
+ stateTimeStats['avg_stopping'] = 0
+
+ return stateTimeStats
+
+
# DEPRECATED
# ____________________________________________________________________________
diff --git a/clmctest/monitoring/StreamingSim.py b/clmctest/monitoring/StreamingSim.py
index 29a7acf..5a995cc 100644
--- a/clmctest/monitoring/StreamingSim.py
+++ b/clmctest/monitoring/StreamingSim.py
@@ -5,7 +5,7 @@ import time
import urllib.parse
import pytest
import random
-import sys
+import sys, getopt
from influxdb import InfluxDBClient
# Simulation parameters
@@ -23,6 +23,8 @@ AGENT2_URL = 'http://172.23.1.22:8186'
class Sim(object):
"""
+
+
Simulator for services
"""
@@ -75,6 +77,10 @@ class Sim(object):
# endpoint state->mu, sigma, secs normal distribution
config_delay_dist = {"placing": [10, 0.68], "booting": [10, 0.68], "connecting": [10, 0.68]}
+ # Simulation configuration of the media component (MC) state changes
+ # "MC state", [average (sec), stddev]
+ mc_config_delay_dist = { "starting": [5, 0.68], "stopping": [2, 0.68]}
+
print("\nSimulation started. Generating data...")
# Place endpoints
@@ -99,6 +105,19 @@ class Sim(object):
max_delay = max(delay_time, max_delay)
sim_time += max_delay
+ # move mpegdash_service endpoints state through from 'starting' to 'running'
+ max_delay = 0
+ for ip_endpoint in ip_endpoints:
+ agent_url = urllib.parse.urlparse(ip_endpoint["agent_url"])
+ agent_db_client = InfluxDBClient(host=agent_url.hostname, port=agent_url.port, database=self.influx_db_name, timeout=10)
+ delay_avg = mc_config_delay_dist['starting'][0]
+ delay_std = delay_avg * mc_config_delay_dist['starting'][1]
+
+ delay_time = self._changeMCState(agent_db_client, sim_time, "mpegdash_service_config", delay_avg, delay_std, 0.7, 'starting', 'running')
+ max_delay = max(delay_time, max_delay)
+
+ sim_time += max_delay
+
# Connect endpoints
max_delay = 0
for ip_endpoint in ip_endpoints:
@@ -174,6 +193,8 @@ class Sim(object):
# remove requests processed off the queue
ip_endpoint['request_queue'] -= int(requests_processed)
+ # update media component state
+
sim_time += TICK_TIME
end_time = sim_time
print("Simulation Finished. Start time {0}. End time {1}. Total time {2}".format(start_time, end_time,
@@ -223,78 +244,35 @@ class Sim(object):
return delay_time
-## PYTEST FIXTURES START
-## ------------------------------------------------------------------------------------------------------
-
-@pytest.fixture(scope='module')
-def run_simulation_fixture(streaming_sim_config):
- """
- A fixture, which checks if the the DB has been created, if not it runs the simulator with a 10 seconds timeout after that
- """
-
- influx_db_url = "http://" + streaming_sim_config['hosts'][0]['ip_address'] + ":8086"
- agent1_url = "http://" + streaming_sim_config['hosts'][1]['ip_address'] + ":8186"
- agent2_url = "http://" + streaming_sim_config['hosts'][2]['ip_address'] + ":8186"
-
- global INFLUX_DB_URL
- global INFLUX_DB_NAME
- global SIMULATION_TIME_SEC
- global AGENT1_URL
- global AGENT2_URL
-
- simulator = Sim(influx_db_url, INFLUX_DB_NAME, agent1_url, agent2_url)
- dbs = simulator.db_client.get_list_database()
- dbs = [db.get("name") for db in dbs]
-
- # This check needed to be disabled as the CLMCMetrics database is always created when
- # the test starts, irrespective of whether this is the 1st time or not
-# if INFLUX_DB_NAME not in dbs:
- simulator.reset()
- simulator.run(SIMULATION_TIME_SEC)
-
- print("10 seconds timeout is given so that the data could properly be inserted into the database.")
- import time
- time.sleep(10)
-
-### Media Component Configuration fixtures
-"""
-Line Protocol report format:
-
-mediaComponentConfig <global tags>,<mediaComp> <configState1=milliseconds>,<configState2=milliseconds> time
-"""
-
-@pytest.fixture(scope='module')
-def reportMC_ServiceState(streaming_sim_config):
- """
- Report to determine whether the simulated streaming servers on the endpoints is running.
- """
-
- # IPEndpoint 1 scenario
- # 10 second period: Stopped --> started --> stopped --> started
- # Stopped state: 1000 + 500 = 1500 [avg = 750]
- # Started state: 3000 + 5500 = 8500 [avg = 4250]
-
- epURL = streaming_sim_config['hosts'][1]['ip_address']
- idbc = InfluxDBClient( host=epURL, port=8186, database=INFLUX_DB_NAME, timeout=10 )
+ @staticmethod
+ def _changeMCState(agent_db_client, sim_time, mc_measurement, mu, sigma, trans_ratio, transition_state, next_state):
+ """
+ Send INFLUX data indicating the time taken to transition to a new state
- idbc.write_points( lp.generate_mc_serviceConfig("apache",1500, 750, 8500, 4250, 0) )
+ Returns the total time delay for the state change
+ """
- # IPEndpoint 2 scenario
- # 10 second period: Stopped --> started --> stopped --> started
- # Stopped state: 250 + 250 = 500 [avg = 250]
- # Started state: 3500 + 6000 = 9500 [avg = 4750]
+ # Calculate a randomized total time for the transition (and calculate relative ratios of time in transition and next state)
+ total_delay_time = random.normalvariate(mu, sigma)
+ transition_time = total_delay_time * trans_ratio
+ next_state_time = total_delay_time - transition_time
- epURL = streaming_sim_config['hosts'][2]['ip_address']
- idbc = InfluxDBClient( host=epURL, port=8186, database=INFLUX_DB_NAME, timeout=10 )
+ mc_states = {}
+
+ # Report time in transition (and add the same as average)
+ mc_states[transition_state] = transition_time
+ mc_states["avg_" +transition_state] = transition_time
- idbc.write_points( lp.generate_mc_serviceConfig("apache",250, 250, 9500, 4750, 0) )
+ # Report time remaining in the next state (adding the same as the average)
+ mc_states[next_state] = next_state_time
+ mc_states["avg_" +next_state] = next_state_time
+ agent_db_client.write_points(lp.generate_mc_service_config(mc_measurement, mc_states, sim_time ))
-## ------------------------------------------------------------------------------------------------------
-## PYTEST FIXTURES END
+ return total_delay_time
-def run_simulation(generate=True):
+def run_simulation(generate=True, sTime=3600):
"""
A method which runs the data generation simulator
:param generate: True for generating data, False for deleting the DB (optional argument, if not given, default value True is used)
@@ -306,6 +284,8 @@ def run_simulation(generate=True):
global AGENT1_URL
global AGENT2_URL
+ SIMULATION_TIME_SEC = sTime
+
simulator = Sim(INFLUX_DB_URL, INFLUX_DB_NAME, AGENT1_URL, AGENT2_URL)
if generate:
@@ -320,11 +300,30 @@ if __name__ == "__main__":
but not when it's imported in another module
"""
- # check if there are any command line arguments given when executing the module
- if len(sys.argv) > 1:
- # if CLI argument '-c' is set when executing the script, the influx db will be deleted instead of generating data
- option = str(sys.argv[1]) != "-c"
- run_simulation(generate=option)
+ # Default options for simulation (generate data and simulation time of 3600 seconds)
+ genOpt = True
+ simTime = 60 * 60
+
+ # Try get some options
+ try:
+ opts, args = getopt.getopt( sys.argv[1:], "c:t:", ['clear','time='])
+
+ except getopt.GetoptError:
+ print( 'StreamingSim.py -c -t <seconds>' )
+ sys.exit(2)
+
+ # Apply options, if any
+ for opt, arg in opts:
+ if opt in ( '-c','--clear' ):
+ genOpt = False
+
+ elif opt in ('-t','--time'):
+ simTime = arg
+
+ if ( genOpt == True ):
+ print( "Running simulation to generate data" )
+ print( "Time period for this simulation: " + str(simTime) + " seconds" )
else:
- # no argument is given to the function call, hence the default value True is used
- run_simulation()
+ print( "Clearing simulation data" )
+
+ run_simulation( genOpt, simTime )
diff --git a/clmctest/monitoring/conftest.py b/clmctest/monitoring/conftest.py
index 828b68c..ad4d96b 100644
--- a/clmctest/monitoring/conftest.py
+++ b/clmctest/monitoring/conftest.py
@@ -4,6 +4,7 @@ import pytest
import yaml
import pkg_resources
from influxdb import InfluxDBClient
+from clmctest.monitoring.StreamingSim import Sim
@pytest.fixture(scope="module")
@@ -21,6 +22,22 @@ def streaming_sim_config():
data_loaded = yaml.load(stream)
return data_loaded
+@pytest.fixture(scope="module")
+def streaming_sim_params(streaming_sim_config):
+ """
+ Uses attributes from the local streaming_sim_config and creates a dictionary of simulation parameters
+ """
+
+ sim_params = {}
+ sim_params["INFLUX_DB_URL"] = "http://" + streaming_sim_config['hosts'][0]['ip_address'] + ":8086"
+ sim_params["INFLUX_DB_NAME"] = streaming_sim_config['hosts'][1]['database_name'] # Note: could this be specified in the clmc-service instead?
+ sim_params["AGENT1_URL"] = "http://" + streaming_sim_config['hosts'][1]['ip_address'] + ":8186"
+ sim_params["AGENT2_URL"] = "http://" + streaming_sim_config['hosts'][2]['ip_address'] + ":8186"
+
+ sim_params["SIMULATION_TIME_SEC"] = 60 * 60
+
+ return sim_params
+
@pytest.fixture(params=[{'database': 'CLMCMetrics'}], scope='module')
def get_db_client(streaming_sim_config, request):
@@ -34,3 +51,22 @@ def get_db_client(streaming_sim_config, request):
return InfluxDBClient(host=streaming_sim_config['hosts'][0]['ip_address'], port=8086, database=request.param['database'], timeout=10)
+@pytest.fixture(scope='module')
+def run_simulation_fixture(streaming_sim_params):
+ """
+ A fixture, which checks if the the DB has been created, if not it runs the simulator with a 10 seconds timeout after that
+ """
+
+ simulator = Sim( streaming_sim_params['INFLUX_DB_URL'], streaming_sim_params['INFLUX_DB_NAME'], streaming_sim_params['AGENT1_URL'], streaming_sim_params['AGENT2_URL'])
+ dbs = simulator.db_client.get_list_database()
+ dbs = [db.get("name") for db in dbs]
+
+ # This check needed to be disabled as the CLMCMetrics database is always created when
+ # the test starts, irrespective of whether this is the 1st time or not
+ # if INFLUX_DB_NAME not in dbs:
+ simulator.reset()
+ simulator.run(streaming_sim_params['SIMULATION_TIME_SEC'])
+
+ print("10 seconds timeout is given so that the data could properly be inserted into the database.")
+ import time
+ time.sleep(10)
diff --git a/clmctest/monitoring/test_MC_Config.py b/clmctest/monitoring/test_MC_Config.py
deleted file mode 100644
index 37ee3bf..0000000
--- a/clmctest/monitoring/test_MC_Config.py
+++ /dev/null
@@ -1,38 +0,0 @@
-#!/usr/bin/python3
-
-import pytest
-from clmctest.monitoring.StreamingSim import reportMC_ServiceState
-
-class TestMediaComponentConfig( object ):
- """
- Test class to check the reported output of a media component's configuration test
- as it is reported to the CLMC
- """
-
- # For data sent to INFLUX, see reportMC_ServiceState function in StreamingSim.py
-
- @pytest.mark.parametrize( "query, expectedResult", [
- ('SELECT "serviceStopped", "avgServiceStopped", "serviceStarted", "avgServiceStarted" FROM "CLMCMetrics"."autogen"."apache" WHERE ipendpoint=\'adaptive_streaming_I1_apache1\'',
- {"time" : "1970-01-01T00:00:00Z", "serviceStopped" : 1500, "avgServiceStopped" : 750, "serviceStarted" : 8500, "avgServiceStarted" : 4250}),
- ('SELECT "serviceStopped", "avgServiceStopped", "serviceStarted", "avgServiceStarted" FROM "CLMCMetrics"."autogen"."apache" WHERE ipendpoint=\'adaptive_streaming_I1_apache2\'',
- {"time" : "1970-01-01T00:00:00Z", "serviceStopped" : 250, "avgServiceStopped" : 250, "serviceStarted" : 9500, "avgServiceStarted" : 4750})
- ])
-
- def test_serviceStateReport( self, query, expectedResult, get_db_client, reportMC_ServiceState ):
- """
- :param query: the LineProtocol query to search for MC service state report
- :param expectedResult: the JSON result obtained from the query
- :param get_db_client: fixture from conftest.py returning INFLUX query client
- :param reportMC_ServiceState: the fixture that makes the service state report
- """
- print( "\n" ) # White space for output
-
- # Query for result and report problems with query if necessary
- queryResult = get_db_client.query( query, raise_errors = False )
- assert queryResult.error is None, "An error occurred executing query {0}."
-
- # Get result and evaluate
- actualResult = next( queryResult.get_points() )
-
- assert expectedResult == actualResult, "FAIL" #self.outputComparison( expectedResult, actualResult )
- print ("Media Component service state test succeeded: {0}".format( query ))
diff --git a/clmctest/monitoring/test_simresults.py b/clmctest/monitoring/test_simresults.py
index 7940fc3..7468594 100644
--- a/clmctest/monitoring/test_simresults.py
+++ b/clmctest/monitoring/test_simresults.py
@@ -1,7 +1,6 @@
#!/usr/bin/python3
import pytest
-from clmctest.monitoring.StreamingSim import run_simulation_fixture
class TestSimulation(object):
@@ -19,7 +18,12 @@ class TestSimulation(object):
('SELECT count(*) FROM "CLMCMetrics"."autogen"."net_port_io"',
{"time": "1970-01-01T00:00:00Z", "count_RX_BYTES_PORT_M": 7200, "count_TX_BYTES_PORT_M": 7200}),
('SELECT count(*) FROM "CLMCMetrics"."autogen"."vm_res_alloc"',
- {"time": "1970-01-01T00:00:00Z", "count_cpu": 12, "count_memory": 12, "count_storage": 12})
+ {"time": "1970-01-01T00:00:00Z", "count_cpu": 12, "count_memory": 12, "count_storage": 12}),
+
+ ('SELECT count(*) FROM "CLMCMetrics"."autogen"."mpegdash_service_config" WHERE ipendpoint=\'adaptive_streaming_I1_apache1\'',
+ {"time" : "1970-01-01T00:00:00Z", "count_avg_running" : 1, "count_avg_starting" : 1, "count_avg_stopped" : 1, "count_avg_stopping" : 1, "count_running" : 1, "count_starting" : 1, "count_stopped" : 1, "count_stopping" : 1}),
+ ('SELECT count(*) FROM "CLMCMetrics"."autogen"."mpegdash_service_config" WHERE ipendpoint=\'adaptive_streaming_I1_apache2\'',
+ {"time" : "1970-01-01T00:00:00Z", "count_avg_running" : 1, "count_avg_starting" : 1, "count_avg_stopped" : 1, "count_avg_stopping" : 1, "count_running" : 1, "count_starting" : 1, "count_stopped" : 1, "count_stopping" : 1}),
])
def test_simulation(self, query, expected_result, get_db_client, run_simulation_fixture):
"""
--
GitLab