DataModel | Fabric API | Dataflow API

ONEx API and Data Models

This ONEx repository produces OpenAPI artifacts that describe APIs and Data Models neccessary for creating open network experiments.

Fabric example

Here's a simple fabric example, creating a clos fabric with 1 spine, 2 pods and 1 ToR in each pod:

Click on language/format node to expand the sample!

Json

{
"choice": "spine_pod_rack",
"spine_pod_rack": {
"spines": [
{
"count": 1
}
],
"pods": [
{
"count": 2,
"pod_profile_name": [ "Pod Profile 1" ]
}
],
"pod_profiles": [
{
"name": "Pod Profile 1",
"pod_switch": {
"count": 1
},
"rack": {
"count": 2,
"rack_profile_names": [ "Rack Profile 1" ]
}
}
],
"rack_profiles": [
{
"name": "Rack Profile 1",
"tor_to_pod_oversubscription": "2:1"
}
]
}
}

Yaml

choice: spine_pod_rackspine_pod_rack:
spines:
- count: 1pods:
- count: 2pod_profile_name:
- Pod Profile 1pod_profiles:
- name: Pod Profile 1pod_switch:
count: 1rack:
count: 2rack_profile_names:
- Rack Profile 1rack_profiles:
- name: Rack Profile 1tor_to_pod_oversubscription: '2:1'

Python

deffabric_sample():
config=onex.api().config()
config.fabric.spine_pod_rack.spines.add(count=1)
config.fabric.spine_pod_rack.pods.add(
count=2,
pod_profile_name=["Pod Profile 1"]
)
pod_profile=config.fabric.spine_pod_rack.pod_profiles.add(name="Pod Profile 1")
pod_profile.pod_switch.count=1rack_profile=config.fabric.spine_pod_rack.rack_profiles.add(
name="Rack Profile 1",
tor_to_pod_oversubscription="2:1"
)
pod_profile.rack.rack_profile_names= [ rack_profile.name ]
pod_profile.rack.count=2

Dataflow example

Below is a simple scatter-gather dataflow example:

Json

{
"dataflow": {
"workload": [
{
"name": "Scatter",
"choice": "scatter",
"scatter": {
"sources": [
"Aggregator"
],
"destinations": [
"Compute 1",
"Compute 2"
],
"flow_profile_name": "data transfer"
}
},
{
"name": "Gather",
"choice": "gather",
"gather": {
"sources": [
"Compute 1",
"Compute 2"
],
"destinations": [
"Aggregator"
],
"flow_profile_name": "data transfer"
}
}
]
},
"hosts": [
{
"name": "Aggregator",
"address": "1.1.1.1"
},
{
"name": "Compute 1",
"address": "3.3.3.3"
},
{
"name": "Compute 2",
"address": "4.4.4.4"
}
],
"flow_profiles": [
{
"name": "data transfer",
"data_size": 1073741824
}
]
}

Yaml

dataflow:
flow_profiles:
- name: data transferdata_size: 1073741824workload:
- name: Scatterchoice: scatterscatter:
destinations:
- Compute 1
- Compute 2flow_profile_name: data transfersources:
- Aggregator
- name: Gatherchoice: gathergather:
destinations:
- Aggregatorflow_profile_name: data transfersources:
- Compute 1
- Compute 2hosts:
- name: Aggregatoraddress: 1.1.1.1
- name: Compute 1address: 3.3.3.3
- name: Compute 2address: 4.4.4.4

Python

defdataflow_sample():
api=onex.api()
config=api.config()
aggregator=config.hosts.add(name="Aggregator", address="1.1.1.1") compute1=config.hosts.add(name="Compute 1", address="3.3.3.3")
compute2=config.hosts.add(name="Compute 2", address="4.4.4.4")
data_transfer=config.dataflow.flow_profiles.add(name='data transfer', data_size=1*1024*1024*1024)
scatter=config.dataflow.workload.add(name="Scatter").scatterscatter.sources= [ aggregator.name ]
scatter.destinations= [ compute1.name, compute2.name ]
scatter.flow_profile_name=data_transfer.namegather=config.dataflow.workload.add(name="Gather").gathergather.sources= [ compute1.name, compute2.name ]
gather.destinations= [ aggregator.name ]
gather.flow_profile_name=data_transfer.nameapi.set_config(config)
api.run_experiment(api.experiment_request())
jct=api.get_metrics(api.metrics_request()).jctprint (f"Experiment complete, JCT: {jct}")

System Experiment example

This example showcase running an ML training job over a simple fabric with different fabric buffer and transport settings while also running an background traffic to put pressure on links up to the spine

Python

defconfigure_fabric(buffer):
# Objective: Configure a fabric with 1 spine, 2 pods, 2 ToRs pe pod and set a buffer in each port int he pod switchapi=onex.api()
config=api.config()
# Create a qos profile with the buffer settingsqos_profile=config.fabric.qos_profiles.add(name='restricted ingress admission')
qos_profile.ingress_admission.shared_buffer_bytes=0qos_profile.ingress_admission.reserved_buffer_bytes=buffer# Create the topology and assign the qos profile to pod switchesconfig.fabric.spine_pod_rack.spines.add(count=1)
config.fabric.spine_pod_rack.pods.add(
count=2,
pod_profile_name=["Pod Profile 1"]
)
pod_profile=config.fabric.spine_pod_rack.pod_profiles.add(name="Pod Profile 1")
pod_profile.pod_switch.count=1rack_profile=config.fabric.spine_pod_rack.rack_profiles.add(
name="Rack Profile 1",
tor_to_pod_oversubscription="2:1"
)
pod_profile.rack.rack_profile_names= [ rack_profile.name ]
pod_profile.rack.count=2pod_profile.pod_switch.qos_profile_name=qos_profile.name# Apply the fabric configapi.set_config(config)
defapply_impairments(spine_link_load):
# Objective: Inject a background traffic in the spine links to create congestion while running traffic from external hostsapi=onex.api()
config=api.get_config()
# Create the flow and injecting in the pod switch of the 1st podflow1=config.chaos.background_traffic.flows.add(name="Flow 1")
flow1.fabric_entry_point.switch_reference.pod.pod_index=1flow1.fabric_entry_point.switch_reference.pod.switch_index=1stateless_flow=flow1.stateless.add(name='Load Spine')
stateless_flow.rate=spine_link_loadstateless_flow.rate_unit='Gbps'# Update fabric config with background trafficapi.set_config(config)
defrun_workfload(mtu):
# Objective: Create a ML Training data flow, run and print the Job Completion Timeapi=onex.api()
config=onex.api().config()
storage_host=config.hosts.add(name="Data Storage 1", address="1.1.1.1")
compute1=config.hosts.add(name="Compute 1", address="3.3.3.3")
compute2=config.hosts.add(name="Compute 2", address="4.4.4.4")
hyperparameters=config.dataflow.flow_profiles.add(name='hyperparameters', data_size=10000)
image_data=config.dataflow.flow_profiles.add(name='image data', data_size=10000000)
gradients_exchange=config.dataflow.flow_profiles.add(name='receive and update gradients', data_size=1000000)
init_scatter=config.dataflow.workload.add(name="transfer hyperparameters").scatterinit_scatter.sources= [ storage_host.name ]
init_scatter.destinations= [ compute1.name, compute2.name ]
init_scatter.flow_profile_name=hyperparameters.nameepoch_loop=config.dataflow.workload.add(name="Epoch loop").loopepoch_loop.iterations=10batch_scatter=epoch_loop.children.add(name='Transfer images').scatterbatch_scatter.sources= [ storage_host.name ]
batch_scatter.destinations= [ compute1.name, compute2.name ]
batch_scatter.flow_profile_name=image_data.namebatch_compute=epoch_loop.children.add(name='Calculate gradients').computebatch_compute.nodes= [ compute1.name, compute2.name ]
batch_compute.simulated.duration=10batch_all_reduce=epoch_loop.children.add(name='Exchange gradients').all_reducebatch_all_reduce.nodes= [ compute1.name, compute2.name ]
batch_all_reduce.flow_profile_name=gradients_exchange.namebatch_all_reduce.type=batch_all_reduce.RINGback_compute_optimizer=epoch_loop.children.add(name='Compute optimizer function + update model').computeback_compute_optimizer.nodes= [ compute1.name, compute2.name ]
back_compute_optimizer.simulated.duration=10# Set the MTUhyperparameters.ethernet.mtu=mtuimage_data.ethernet.mtu=mtugradients_exchange.ethernet.mtu=mtu# Apply data flow configapi.set_config(config)
# Run the workfload and print out Job Completion Timeapi.run_experiment(api.experiment_request())
jct=api.get_metrics(api.metrics_request()).jctprint (f"Experiment complete, JCT: {jct}")
defrun_experiments():
forfabric_switch_port_bufferin [0, 10000, 1000000]:
configure_fabric(fabric_switch_port_buffer)
forspine_link_loadin [0, 10, 20]:
apply_impairments(spine_link_load)
formtuin [1500, 9000]:
run_workfload(mtu)

Contributing

The open-network-experiment organization welcomes new members to join this open source community project and contribute to its development.