Merge pull request #599 from perfsonar/single-ended

Add --single-ended flag to iperf2 and iperf3  #456

pscheduler-test-throughput/tests/cli-to-spec_test.py

@@ -27,6 +27,8 @@ def test_all_options(self):
             "congestion": {"val": "reno" },
             "client-cpu-affinity": {"val": "1" },
             "server-cpu-affinity": {"val": "0" },
+            "single-ended": {},
+            "single-ended-port": { "val": "5000" },
             "reverse": {}
         })
 

pscheduler-test-throughput/tests/spec-is-valid_test.py

@@ -40,6 +40,8 @@ def test_basic(self):
             "flow-label": 1,
             "client-cpu-affinity": 1,
             "server-cpu-affinity": 1,
+            "single-ended": True,
+            "single-ended-port": 5000,
             "reverse": True
             }
 

pscheduler-test-throughput/throughput/cli-to-spec

@@ -159,6 +159,16 @@ opt_parser.add_option("--server-cpu-affinity",
                       action="store", type="int",
                       dest="server_cpu_affinity")
 
+opt_parser.add_option("--single-ended",
+                      help="Run a test directly to a host without pscheduler.",
+                      action="store_true",
+                      dest="single_ended")
+
+opt_parser.add_option("--single-ended-port",
+                      help="Run a test directly to a given port.",
+                      action="store", type="int",
+                      dest="single_ended_port")
+
 opt_parser.add_option("--reverse",
                       help="Reverses the direction of the test.",
                       action="store_true",
@@ -174,7 +184,8 @@ if len(remaining_args) != 0:
    pscheduler.fail("Unusable arguments: %s" % " ".join(remaining_args))
 
 
-result = { 'schema': 1 }
+result = { }
+schema = pscheduler.HighInteger(1)
 
 if options.source is not None:
    result['source'] = options.source
@@ -272,9 +283,19 @@ if options.client_cpu_affinity != None:
 if options.server_cpu_affinity != None:
    result["server-cpu-affinity"] = options.server_cpu_affinity
 
+if options.single_ended:
+   result["single-ended"] = True
+   schema.set(2)
+
+if options.single_ended_port:
+   result["single-ended-port"] = options.single_ended_port
+   schema.set(2)
+
 if options.reverse:
    result["reverse"] = True
 
+result["schema"] = schema.value()
+
 logger.debug("cli-to-spec -> %s" % result)
 
 pscheduler.succeed_json(result)

pscheduler-test-throughput/throughput/participants

@@ -13,7 +13,7 @@ import sys
 
 logger = pscheduler.Log(prefix='test-throughput', quiet=True)
 
-json = pscheduler.json_load(exit_on_error=True, max_schema=1)
+json = pscheduler.json_load(exit_on_error=True, max_schema=2)
 
 null_reason = None
 
@@ -42,11 +42,14 @@ has = pscheduler.api_has_services([source, destination], timeout=3)
 
 src_has_psc = has[source]["pscheduler"]
 dst_has_psc = has[destination]["pscheduler"]
+single_ended = json.get('single-ended', False)
 
 
 logger.debug("src_has_psc = %s, dst_has_psc = %s" % (src_has_psc, dst_has_psc))
 
-if src_has_psc and dst_has_psc:
+if src_has_psc and single_ended:
+    participants = [ source ]
+elif src_has_psc and dst_has_psc:
     participants = [ source, destination ]
 elif src_has_psc:
     if not has[destination]["bwctl"]:

pscheduler-test-throughput/throughput/spec-format

@@ -12,7 +12,7 @@ try:
 except IndexError:
    format = 'text/plain'
 
-json = pscheduler.json_load(exit_on_error=True, max_schema=1)
+json = pscheduler.json_load(exit_on_error=True, max_schema=2)
 
 if format == 'text/plain':
 

pscheduler-test-throughput/throughput/spec-is-valid

@@ -9,8 +9,8 @@ from validate import spec_is_valid
 
 
 try:
-    json = pscheduler.json_load(max_schema=1)
-    pscheduler.json_check_schema(json, 1)
+    json = pscheduler.json_load(max_schema=2)
+    pscheduler.json_check_schema(json, 2)
 except ValueError as ex:
     pscheduler.succeed_json({
         "valid": False,

pscheduler-test-throughput/throughput/spec-to-cli

@@ -5,7 +5,7 @@
 import pscheduler
 import validate
 
-spec = pscheduler.json_load(exit_on_error=True, max_schema=1)
+spec = pscheduler.json_load(exit_on_error=True, max_schema=2)
 
 if type(spec) != dict:
     pscheduler.fail("Invalid JSON for this operation")
@@ -17,7 +17,10 @@ if not valid:
 strings = []
 bools = []
 
-spec_properties = validate.SPEC_SCHEMA['properties']
+if spec['schema'] == 2:
+    spec_properties = validate.SPEC_SCHEMA['local']['throughput_v2']['properties']
+else:
+    spec_properties = validate.SPEC_SCHEMA['local']['throughput_v1']['properties']
 
 for item in spec_properties.keys():
     if item == 'schema': continue

pscheduler-test-throughput/throughput/validate.py

@@ -5,42 +5,89 @@
 from pscheduler import json_validate
 import json
 
-SPEC_SCHEMA = {        
-    "title": "pScheduler Throughput Specification Schema",
-    "type": "object",
-    "additionalProperties": False,
+SPEC_SCHEMA = {
     "local": {
+        "congestion": {
+            "type": "string",
+            "enum": ["reno", "cubic", "bic", "htcp", "vegas", "westwood",
+                     "yeah", "bbr"]
         },
-    "required": ["dest"],
-    "properties": {           
-        "schema":      { "$ref": "#/pScheduler/Cardinal" },
-        "source":      { "$ref": "#/pScheduler/Host" },
-        "source-node": { "$ref": "#/pScheduler/URLHostPort" },
-        "dest":        { "$ref": "#/pScheduler/Host" },
-        "dest-node":   { "$ref": "#/pScheduler/URLHostPort" },
-        "duration":    { "$ref": "#/pScheduler/Duration" },
-        "interval":    { "$ref": "#/pScheduler/Duration" },
-        "parallel":    { "$ref": "#/pScheduler/Cardinal" },
-        "udp":         { "$ref": "#/pScheduler/Boolean" },
-        "bandwidth":   { "$ref": "#/pScheduler/Cardinal" },           
-        "window-size": { "$ref": "#/pScheduler/Cardinal" },
-        "mss":         { "$ref": "#/pScheduler/Cardinal" },
-        "buffer-length": { "$ref": "#/pScheduler/Cardinal" },
-        "ip-tos":        { "$ref": "#/pScheduler/IPTOS" },
-        "ip-version":    { "$ref": "#/pScheduler/ip-version" },
-        "local-address": { "$ref": "#/pScheduler/Host" },
-        "omit":          { "$ref": "#/pScheduler/Duration" },
-        "no-delay":    { "$ref": "#/pScheduler/Boolean" },
-        "congestion":    { "enum": ["reno", "cubic", "bic", "htcp", "vegas",
-                                    "westwood", "yeah", "bbr"]
-                           },
-        "zero-copy":    { "$ref": "#/pScheduler/Boolean" },
-        "flow-label":    { "$ref": "#/pScheduler/Cardinal" },
-        "client-cpu-affinity":    { "$ref": "#/pScheduler/Integer" },
-        "server-cpu-affinity":    { "$ref": "#/pScheduler/Integer" },
-        "reverse": { "$ref": "#/pScheduler/Boolean" }
+        "throughput_v1" : {
+            "title": "pScheduler Throughput Specification Schema",
+            "type": "object",
+            "properties": {
+                "schema":      { "type": "integer", "enum": [ 1 ] },
+                "source":      { "$ref": "#/pScheduler/Host" },
+                "source-node": { "$ref": "#/pScheduler/URLHostPort" },
+                "dest":        { "$ref": "#/pScheduler/Host" },
+                "dest-node":   { "$ref": "#/pScheduler/URLHostPort" },
+                "duration":    { "$ref": "#/pScheduler/Duration" },
+                "interval":    { "$ref": "#/pScheduler/Duration" },
+                "parallel":    { "$ref": "#/pScheduler/Cardinal" },
+                "udp":         { "$ref": "#/pScheduler/Boolean" },
+                "bandwidth":   { "$ref": "#/pScheduler/Cardinal" },
+                "window-size": { "$ref": "#/pScheduler/Cardinal" },
+                "mss":         { "$ref": "#/pScheduler/Cardinal" },
+                "buffer-length": { "$ref": "#/pScheduler/Cardinal" },
+                "ip-tos":        { "$ref": "#/pScheduler/IPTOS" },
+                "ip-version":    { "$ref": "#/pScheduler/ip-version" },
+                "local-address": { "$ref": "#/pScheduler/Host" },
+                "omit":          { "$ref": "#/pScheduler/Duration" },
+                "no-delay":    { "$ref": "#/pScheduler/Boolean" },
+                "congestion":    { "$ref": "#/local/congestion" },
+                "zero-copy":    { "$ref": "#/pScheduler/Boolean" },
+                "flow-label":    { "$ref": "#/pScheduler/Cardinal" },
+                "client-cpu-affinity":    { "$ref": "#/pScheduler/Integer" },
+                "server-cpu-affinity":    { "$ref": "#/pScheduler/Integer" },
+                "reverse": { "$ref": "#/pScheduler/Boolean" }
+            },
+            "additionalProperties": False,
+            "required": [ "dest" ]
+        },
+        "throughput_v2" : {
+            "title": "pScheduler Throughput Specification Schema",
+            "type": "object",
+            "properties": {
+                "schema":      { "type": "integer", "enum": [ 2 ] },
+                "source":      { "$ref": "#/pScheduler/Host" },
+                "source-node": { "$ref": "#/pScheduler/URLHostPort" },
+                "dest":        { "$ref": "#/pScheduler/Host" },
+                "dest-node":   { "$ref": "#/pScheduler/URLHostPort" },
+                "duration":    { "$ref": "#/pScheduler/Duration" },
+                "interval":    { "$ref": "#/pScheduler/Duration" },
+                "parallel":    { "$ref": "#/pScheduler/Cardinal" },
+                "udp":         { "$ref": "#/pScheduler/Boolean" },
+                "bandwidth":   { "$ref": "#/pScheduler/Cardinal" },
+                "window-size": { "$ref": "#/pScheduler/Cardinal" },
+                "mss":         { "$ref": "#/pScheduler/Cardinal" },
+                "buffer-length": { "$ref": "#/pScheduler/Cardinal" },
+                "ip-tos":        { "$ref": "#/pScheduler/IPTOS" },
+                "ip-version":    { "$ref": "#/pScheduler/ip-version" },
+                "local-address": { "$ref": "#/pScheduler/Host" },
+                "omit":          { "$ref": "#/pScheduler/Duration" },
+                "no-delay":    { "$ref": "#/pScheduler/Boolean" },
+                "congestion":    { "$ref": "#/local/congestion" },
+                "zero-copy":    { "$ref": "#/pScheduler/Boolean" },
+                "flow-label":    { "$ref": "#/pScheduler/Cardinal" },
+                "client-cpu-affinity":    { "$ref": "#/pScheduler/Integer" },
+                "server-cpu-affinity":    { "$ref": "#/pScheduler/Integer" },
+                "single-ended": { "$ref": "#/pScheduler/Boolean" },
+                "single-ended-port": { "$ref": "#/pScheduler/Integer" },
+                "reverse": { "$ref": "#/pScheduler/Boolean" }
+            },
+            "additionalProperties": False,
+            "required": [ "dest" ]
+        },
+        "throughput": {
+            "anyOf" : [
+                { "$ref": "#/local/throughput_v1" },
+                { "$ref": "#/local/throughput_v2" }
+            ]
         }
-    }
+    },
+
+    "$ref": "#/local/throughput"
+}
 
 RESULT_SCHEMA = {        
     "title": "pScheduler Throughput Response Schema",

pscheduler-tool-iperf2/iperf2/can-run

@@ -32,7 +32,7 @@ if not json.get('spec'):
 
 try:
     spec = json["spec"]
-    pscheduler.json_check_schema(spec, 1)
+    pscheduler.json_check_schema(spec, 2)
 except KeyError:
     pscheduler.succeed_json({
         "can-run": False,
@@ -55,7 +55,7 @@ supported_options = ["schema",
                      "ip-version", "duration", "interval",
                      "parallel", "window-size", "mss", "bandwidth",
                      "buffer-length", "local-address", "congestion",
-                     "udp"
+                     "udp", "single-ended", "single-ended-port"
                      ]
 
 for option in spec.keys():

pscheduler-tool-iperf2/iperf2/run

@@ -40,7 +40,9 @@ except KeyError as e:
 except:
     pscheduler.fail("Error parsing run input: %s" % sys.exc_info()[0])
 
-if len(participant_data) != 2:
+single_ended = test_spec.get('single-ended', False)
+participants = len(participant_data)
+if not(participants == 2 or (participants == 1 and single_ended)):
     pscheduler.fail("iperf2 requires exactly 2 participants, got %s" % (len(participant_data)))
 
 
@@ -50,7 +52,10 @@ config = iperf2_utils.get_config()
 iperf2_cmd  = config["iperf2_cmd"]
 
 # grab the server port from the test spec if there, otherwise use default
-server_port = participant_data[1].get('server_port', DEFAULT_SERVER_PORT)
+if single_ended:
+    server_port = test_spec.get('single-ended-port', DEFAULT_SERVER_PORT)
+else:
+    server_port = participant_data[1].get('server_port', DEFAULT_SERVER_PORT)
 
 
 def run_client():    

pscheduler-tool-iperf3/iperf3/can-run

@@ -29,7 +29,7 @@ except KeyError:
 
 try:
     spec = json["spec"]
-    pscheduler.json_check_schema(spec, 1)
+    pscheduler.json_check_schema(spec, 2)
 except KeyError:
     pscheduler.succeed_json({
         "can-run": False,
@@ -47,7 +47,7 @@ errors = []
 supported_options = ["schema",
                      "source", "source-node",
                      "dest", "dest-node",
-                     "reverse", "omit",
+                     "reverse", "omit", "single-ended", "single-ended-port",
                      "duration", "interval", "parallel", "window-size",
                      "mss", "bandwidth", "udp", "buffer-length",
                      "ip-tos", "local-address", "ip-version", "congestion",

pscheduler-tool-iperf3/iperf3/merged-results

@@ -21,17 +21,20 @@ except KeyError as e:
     logger.error("merged-result error %s" % e)
     pscheduler.fail("Missing required key in merged-result input: %s" % e)
 
-if len(result_list) != 2:
+single_ended = input['test']['spec'].get('single-ended')
+result_len = len(result_list)
+if not(result_len == 2 or (result_len == 1 and single_ended)):
     pscheduler.fail("Expected 2 results in merged-results, got %s" % len(result_list))
 
 source_results = result_list[0]
-dest_results   = result_list[1]
+if not single_ended:
+    dest_results = result_list[1]
 
 final_results = source_results
 
 # if this was a reverse test, we want to look at the destination's
 # output instead of the source's
-if input['test']['spec'].get('reverse'):
+if input['test']['spec'].get('reverse') and not single_ended:
     final_results = dest_results
 
 # it's possible one could come back as null, ensure we have the right type
@@ -43,7 +46,7 @@ final_diag = ""
 if source_results and source_results.get('diags'):
     final_diag += "Participant 0:\n%s\n" % source_results.get('diags')
 
-if dest_results and dest_results.get('diags'):
+if not single_ended and dest_results and dest_results.get('diags'):
     final_diag += "Participant 1:\n%s\n" % dest_results.get('diags')
 
 final_results['diags'] = final_diag

pscheduler-tool-iperf3/iperf3/run

@@ -38,7 +38,9 @@ except KeyError as e:
 except Exception:
     pscheduler.fail("Error parsing run input: %s" % sys.exc_info()[0])
 
-if len(participant_data) != 2:
+single_ended = test_spec.get('single-ended', False)
+participants = len(participant_data)
+if not(participants == 2 or (participants == 1 and single_ended)):
     pscheduler.fail("iperf3 requires exactly 2 participants, got %s" % (len(participant_data)))
 
 config = iperf3_utils.get_config()
@@ -47,7 +49,10 @@ config = iperf3_utils.get_config()
 iperf3_cmd  = config["iperf3_cmd"]
 
 # grab the server port from the test spec
-server_port = participant_data[1]['server_port']   
+if single_ended:
+    server_port = test_spec.get('single-ended-port', DEFAULT_SERVER_PORT)
+else:
+    server_port = participant_data[1]['server_port']
 
 # need this to pad out the wait timeout,
 # convert to seconds