如何在采用Ryu控制器的OpenFlow 1.3 SDN网络中计算吞吐量

Hey there! Calculating throughput in an Ryu-based SDN using OpenFlow 1.3 is totally manageable—let’s break down the core approaches, how to implement them in Ryu, and key things to keep in mind.

Throughput (usually measured in Mbps or Gbps) relies on tracking byte/packet counts over a time interval. OpenFlow 1.3 provides two main stats sources for this:

Flow-Level Statistics #

Ideal if you need throughput for specific traffic flows (e.g., between a pair of IPs or ports). OpenFlow 1.3’s OFPFlowStats includes:

• byte_count: Total bytes transmitted by the flow
• packet_count: Total packets transmitted by the flow
• duration_sec/duration_nsec: How long the flow has been active

Throughput formula for a flow:

Throughput (bps) = (Current byte_count - Previous byte_count) * 8 / Time interval

(Multiply by 8 to convert bytes to bits)

Port-Level Statistics #

Great for measuring overall throughput of a switch port (inbound or outbound). OpenFlow 1.3’s OFPPortStats includes:

• rx_bytes: Total bytes received on the port
• tx_bytes: Total bytes sent from the port

Throughput formula for a port:

RX Throughput (bps) = (Current rx_bytes - Previous rx_bytes) * 8 / Time interval
TX Throughput (bps) = (Current tx_bytes - Previous tx_bytes) * 8 / Time interval

Let’s walk through code snippets to implement both methods in Ryu.

Example 1: Flow-Level Throughput Monitor #

This app will periodically request flow stats from connected switches, track historical counts, and calculate throughput:

from ryu.base import app_manager
from ryu.controller import ofp_event
from ryu.controller.handler import MAIN_DISPATCHER, DEAD_DISPATCHER
from ryu.controller.handler import set_ev_cls
from ryu.ofproto import ofproto_v1_3
import time
import threading

class FlowThroughputMonitor(app_manager.RyuApp):
    OFP_VERSIONS = [ofproto_v1_3.OFP_VERSION]

    def __init__(self, *args, **kwargs):
        super(FlowThroughputMonitor, self).__init__(*args, **kwargs)
        self.datapaths = {}
        self.flow_cache = {}  # Key: (datapath_id, flow_match), Value: (timestamp, byte_count)
        self.monitor_interval = 5  # Check every 5 seconds

    @set_ev_cls(ofp_event.EventOFPStateChange, [MAIN_DISPATCHER, DEAD_DISPATCHER])
    def _state_change_handler(self, ev):
        datapath = ev.datapath
        if ev.state == MAIN_DISPATCHER:
            if datapath.id not in self.datapaths:
                self.logger.info(f"Registered switch: {datapath.id:016x}")
                self.datapaths[datapath.id] = datapath
                self.start_flow_monitoring(datapath)
        elif ev.state == DEAD_DISPATCHER:
            if datapath.id in self.datapaths:
                self.logger.info(f"Unregistered switch: {datapath.id:016x}")
                del self.datapaths[datapath.id]

    def start_flow_monitoring(self, datapath):
        # Request flow stats now
        self._request_flow_stats(datapath)
        # Schedule next request
        threading.Timer(self.monitor_interval, self.start_flow_monitoring, args=[datapath]).start()

    def _request_flow_stats(self, datapath):
        parser = datapath.ofproto_parser
        # Request stats for all flows (adjust match to target specific flows)
        req = parser.OFPFlowStatsRequest(
            datapath, 0, datapath.ofproto.OFPTT_ALL,
            datapath.ofproto.OFPP_ANY, datapath.ofproto.OFPG_ANY,
            0, 0, None
        )
        datapath.send_msg(req)

    @set_ev_cls(ofp_event.EventOFPFlowStatsReply, MAIN_DISPATCHER)
    def _flow_stats_reply_handler(self, ev):
        current_time = time.time()
        datapath_id = ev.msg.datapath.id

        for stat in ev.msg.body:
            # Use flow match to uniquely identify the flow
            flow_key = (datapath_id, stat.match.to_jsondict())
            
            if flow_key in self.flow_cache:
                prev_time, prev_bytes = self.flow_cache[flow_key]
                time_diff = current_time - prev_time
                
                if time_diff > 0:
                    throughput_bps = (stat.byte_count - prev_bytes) * 8 / time_diff
                    self.logger.info(
                        f"Flow {stat.match.to_jsondict()} on switch {datapath_id:016x}: "
                        f"Throughput = {throughput_bps/1e6:.2f} Mbps"
                    )
            
            # Update cache with latest stats
            self.flow_cache[flow_key] = (current_time, stat.byte_count)

Example 2: Port-Level Throughput Monitor #

Add this to the same app (or create a new one) to track port throughput:

# Add this to the FlowThroughputMonitor class
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    # ... existing init code ...
    self.port_cache = {}  # Key: (datapath_id, port_no), Value: (timestamp, rx_bytes, tx_bytes)

def start_port_monitoring(self, datapath):
    self._request_port_stats(datapath)
    threading.Timer(self.monitor_interval, self.start_port_monitoring, args=[datapath]).start()

def _request_port_stats(self, datapath):
    parser = datapath.ofproto_parser
    req = parser.OFPPortStatsRequest(datapath, 0, datapath.ofproto.OFPP_ANY)
    datapath.send_msg(req)

@set_ev_cls(ofp_event.EventOFPPortStatsReply, MAIN_DISPATCHER)
def _port_stats_reply_handler(self, ev):
    current_time = time.time()
    datapath_id = ev.msg.datapath.id

    for stat in ev.msg.body:
        # Skip the local switch port (OFPP_LOCAL)
        if stat.port_no == ev.msg.datapath.ofproto.OFPP_LOCAL:
            continue
            
        port_key = (datapath_id, stat.port_no)
        if port_key in self.port_cache:
            prev_time, prev_rx, prev_tx = self.port_cache[port_key]
            time_diff = current_time - prev_time
            
            if time_diff > 0:
                rx_mbps = (stat.rx_bytes - prev_rx) * 8 / time_diff / 1e6
                tx_mbps = (stat.tx_bytes - prev_tx) * 8 / time_diff / 1e6
                self.logger.info(
                    f"Port {stat.port_no} on switch {datapath_id:016x}: "
                    f"RX = {rx_mbps:.2f} Mbps | TX = {tx_mbps:.2f} Mbps"
                )
        
        self.port_cache[port_key] = (current_time, stat.rx_bytes, stat.tx_bytes)

• Time Sync: Ensure your Ryu controller and switches have synchronized system time (use NTP) to avoid skewed time intervals.
• Monitor Interval: 5-10 seconds is a sweet spot—too short and you’ll get noisy data; too long and you’ll miss short-term throughput spikes.
• Targeted Flows: To track specific traffic, modify the match parameter in OFPFlowStatsRequest (e.g., match source/dest IPs, TCP ports).
• Cache Cleanup: Periodically remove stale entries from your cache (e.g., flows that haven’t updated in 30+ seconds) to avoid memory bloat.

内容的提问来源于stack exchange，提问作者Chit Su