1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273

"""
SDN Server Load Balancing
Demonstrates VIP-based load balancing with health checks
"""
 
from dataclasses import dataclass
from typing import Dict, List, Optional
from enum import Enum
import hashlib
import time
 
class ServerState(Enum):
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"
    DRAINING = "draining"  # Finishing existing connections
 
@dataclass
class Server:
    ip: str
    port: int
    weight: int = 1
    state: ServerState = ServerState.HEALTHY
    active_connections: int = 0
    last_health_check: float = 0
 
@dataclass
class VirtualService:
    vip: str
    vip_port: int
    name: str
    servers: List[Server]
    algorithm: str = "weighted_round_robin"
    health_check_interval: int = 5  # seconds
    session_persistence: bool = True
 
class SDNServerLoadBalancer:
    """
    SDN-based server load balancing application.
    Distributes traffic across server pools using programmable switches.
    """
 
    def __init__(self, controller, health_checker):
        self.controller = controller
        self.health_checker = health_checker
        self.services: Dict[str, VirtualService] = {}
        self.round_robin_counters: Dict[str, int] = {}
 
    def register_service(self, service: VirtualService):
        """Register a virtual service for load balancing."""
        self.services[f"{service.vip}:{service.vip_port}"] = service
        self.round_robin_counters[service.name] = 0
        
        # Start health checking
        self.health_checker.add_targets(
            [s.ip for s in service.servers],
            callback=lambda ip, healthy: self._on_health_change(
                service.name, ip, healthy
            )
        )
        
        # Install initial flow rules
        self._install_lb_rules(service)
 
    def _install_lb_rules(self, service: VirtualService):
        """Install OpenFlow rules for load balancing."""
        
        healthy_servers = [
            s for s in service.servers 
            if s.state == ServerState.HEALTHY
        ]
        
        if not healthy_servers:
            # No healthy servers - install drop rule with alert
            self._install_service_unavailable(service)
            return
 
        # Calculate bucket weights
        total_weight = sum(s.weight for s in healthy_servers)
        
        buckets = []
        for server in healthy_servers:
            bucket_weight = int((server.weight / total_weight) * 100)
            buckets.append({
                "weight": bucket_weight,
                "actions": [
                    {"type": "SET_FIELD", "field": "ip_dst", 
                     "value": server.ip},
                    {"type": "SET_FIELD", "field": "tcp_dst", 
                     "value": server.port},
                    {"type": "OUTPUT", "port": self._get_server_port(server)}
                ]
            })
 
        # Create or update SELECT group
        group_id = self._get_group_id(service)
        
        for switch_id in self._get_ingress_switches():
            self.controller.install_group(
                switch_id=switch_id,
                group_id=group_id,
                group_type="SELECT",
                buckets=buckets
            )
            
            # Install flow rule using the group
            self.controller.install_flow(
                switch_id=switch_id,
                priority=1000,
                match={
                    "ip_dst": service.vip,
                    "tcp_dst": service.vip_port,
                },
                actions=[{"type": "GROUP", "group_id": group_id}]
            )
 
            # Install reverse NAT for responses
            for server in healthy_servers:
                self.controller.install_flow(
                    switch_id=switch_id,
                    priority=1000,
                    match={
                        "ip_src": server.ip,
                        "tcp_src": server.port,
                    },
                    actions=[
                        {"type": "SET_FIELD", "field": "ip_src", 
                         "value": service.vip},
                        {"type": "SET_FIELD", "field": "tcp_src", 
                         "value": service.vip_port},
                        {"type": "OUTPUT", "port": "NORMAL"}
                    ]
                )
 
    def _on_health_change(self, service_name: str, server_ip: str, healthy: bool):
        """Handle server health state change."""
        service_key = None
        for key, service in self.services.items():
            if service.name == service_name:
                service_key = key
                break
        
        if not service_key:
            return
            
        service = self.services[service_key]
        
        for server in service.servers:
            if server.ip == server_ip:
                old_state = server.state
                server.state = (ServerState.HEALTHY if healthy 
                               else ServerState.UNHEALTHY)
                
                if old_state != server.state:
                    print(f"Server {server_ip} state: {old_state} -> {server.state}")
                    
                    # Reinstall rules with updated server list
                    self._install_lb_rules(service)
                break
 
    def select_server_reactive(
        self, 
        service: VirtualService,
        client_ip: str,
        flow_tuple: tuple
    ) -> Optional[Server]:
        """
        Select server for reactive (per-flow) load balancing.
        Called when first packet triggers Packet-In.
        """
        healthy_servers = [
            s for s in service.servers 
            if s.state == ServerState.HEALTHY
        ]
        
        if not healthy_servers:
            return None
 
        if service.algorithm == "round_robin":
            return self._round_robin(service.name, healthy_servers)
        elif service.algorithm == "weighted_round_robin":
            return self._weighted_round_robin(service.name, healthy_servers)
        elif service.algorithm == "least_connections":
            return self._least_connections(healthy_servers)
        elif service.algorithm == "ip_hash":
            return self._ip_hash(client_ip, healthy_servers)
        else:
            return healthy_servers[0]
 
    def _round_robin(
        self, 
        service_name: str, 
        servers: List[Server]
    ) -> Server:
        """Simple round-robin selection."""
        counter = self.round_robin_counters.get(service_name, 0)
        selected = servers[counter % len(servers)]
        self.round_robin_counters[service_name] = counter + 1
        return selected
 
    def _weighted_round_robin(
        self, 
        service_name: str, 
        servers: List[Server]
    ) -> Server:
        """Weighted round-robin based on server weights."""
        # Expand server list by weights
        weighted_list = []
        for server in servers:
            weighted_list.extend([server] * server.weight)
        
        counter = self.round_robin_counters.get(service_name, 0)
        selected = weighted_list[counter % len(weighted_list)]
        self.round_robin_counters[service_name] = counter + 1
        return selected
 
    def _least_connections(self, servers: List[Server]) -> Server:
        """Select server with fewest active connections."""
        return min(servers, key=lambda s: s.active_connections)
 
    def _ip_hash(self, client_ip: str, servers: List[Server]) -> Server:
        """Consistent hashing based on client IP."""
        hash_value = int(hashlib.md5(client_ip.encode()).hexdigest(), 16)
        return servers[hash_value % len(servers)]
 
    def drain_server(self, service_name: str, server_ip: str):
        """
        Gracefully remove server from pool.
        Existing connections continue; new connections go elsewhere.
        """
        for service in self.services.values():
            if service.name == service_name:
                for server in service.servers:
                    if server.ip == server_ip:
                        server.state = ServerState.DRAINING
                        
                        # Update LB rules to exclude this server
                        self._install_lb_rules(service)
                        
                        print(f"Server {server_ip} draining - "
                              f"no new connections")
                        return
 
    def _install_service_unavailable(self, service: VirtualService):
        """Install rules when no servers are available."""
        for switch_id in self._get_ingress_switches():
            # Option 1: Send ICMP unreachable
            # Option 2: Redirect to maintenance page
            # Option 3: Simply drop (worst UX)
            
            self.controller.install_flow(
                switch_id=switch_id,
                priority=1000,
                match={
                    "ip_dst": service.vip,
                    "tcp_dst": service.vip_port,
                },
                actions=[
                    # Send to controller for custom response
                    {"type": "OUTPUT", "port": "CONTROLLER"}
                ]
            )
 
    def _get_group_id(self, service: VirtualService) -> int:
        """Generate unique group ID for service."""
        return hash(f"{service.vip}:{service.vip_port}") & 0xFFFFFFFF
 
    def _get_server_port(self, server: Server) -> int:
        """Get switch port for server."""
        return self.controller.get_host_port(server.ip)
 
    def _get_ingress_switches(self) -> List[str]:
        """Get switches where LB rules should be installed."""
        return self.controller.get_all_switches()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258

"""
SDN Path Load Balancing
Distributes traffic across multiple network paths based on utilization
"""
 
from dataclasses import dataclass
from typing import Dict, List, Tuple
import heapq
 
@dataclass
class NetworkPath:
    path_id: str
    hops: List[str]  # List of switch IDs
    capacity_gbps: float
    current_utilization: float = 0.0
 
    @property
    def available_bandwidth(self) -> float:
        return self.capacity_gbps * (1 - self.current_utilization)
 
class PathLoadBalancer:
    """
    SDN path load balancing application.
    Distributes flows across multiple paths based on capacity and utilization.
    """
 
    def __init__(self, controller, topology_manager):
        self.controller = controller
        self.topology = topology_manager
        self.path_flows: Dict[str, List[dict]] = {}  # path_id -> flows
 
    def compute_paths(
        self, 
        source: str, 
        destination: str,
        k: int = 3
    ) -> List[NetworkPath]:
        """
        Compute K diverse paths between source and destination.
        """
        all_paths = self.topology.get_k_shortest_paths(source, destination, k)
        
        return [
            NetworkPath(
                path_id=f"{source}-{destination}-{i}",
                hops=path,
                capacity_gbps=self._get_path_capacity(path),
                current_utilization=self._get_path_utilization(path)
            )
            for i, path in enumerate(all_paths)
        ]
 
    def distribute_flow(
        self, 
        source: str, 
        destination: str,
        flow_demand_gbps: float,
        flow_id: str
    ) -> List[Tuple[NetworkPath, float]]:
        """
        Distribute a flow across paths based on available capacity.
        Returns list of (path, allocated_bandwidth) tuples.
        """
        paths = self.compute_paths(source, destination)
        
        # For small flows, use single best path
        if flow_demand_gbps < 1.0:
            best_path = max(paths, key=lambda p: p.available_bandwidth)
            self._install_single_path(flow_id, best_path)
            return [(best_path, flow_demand_gbps)]
 
        # For large flows, split across paths
        return self._split_flow_across_paths(flow_id, flow_demand_gbps, paths)
 
    def _split_flow_across_paths(
        self,
        flow_id: str,
        demand: float,
        paths: List[NetworkPath]
    ) -> List[Tuple[NetworkPath, float]]:
        """
        Split large flow across multiple paths proportionally.
        """
        allocations = []
        remaining_demand = demand
        
        # Sort paths by available bandwidth
        sorted_paths = sorted(
            paths, 
            key=lambda p: p.available_bandwidth, 
            reverse=True
        )
        
        # Allocate proportionally to available bandwidth
        total_available = sum(p.available_bandwidth for p in sorted_paths)
        
        if total_available < demand:
            # Not enough capacity - allocate what we can
            print(f"Warning: Insufficient capacity for flow {flow_id}")
            
        for path in sorted_paths:
            if remaining_demand <= 0:
                break
                
            allocation = min(
                path.available_bandwidth,
                remaining_demand * (path.available_bandwidth / total_available)
            )
            
            if allocation > 0:
                allocations.append((path, allocation))
                remaining_demand -= allocation
 
        # Install split flow rules
        self._install_split_flow(flow_id, allocations)
        
        return allocations
 
    def _install_single_path(self, flow_id: str, path: NetworkPath):
        """Install flow rules for single-path routing."""
        
        hops = path.hops
        for i, switch_id in enumerate(hops[:-1]):
            next_hop = hops[i + 1]
            out_port = self.topology.get_port_to_neighbor(switch_id, next_hop)
            
            self.controller.install_flow(
                switch_id=switch_id,
                priority=500,
                match=self._get_flow_match(flow_id),
                actions=[{"type": "OUTPUT", "port": out_port}],
                cookie=hash(f"{flow_id}-{path.path_id}")
            )
 
    def _install_split_flow(
        self, 
        flow_id: str, 
        allocations: List[Tuple[NetworkPath, float]]
    ):
        """
        Install flow rules to split traffic across paths.
        Uses weighted group buckets.
        """
        # Get first switch (ingress)
        first_hops = set(alloc[0].hops[0] for alloc in allocations)
        
        for ingress_switch in first_hops:
            # Create weighted group for splitting
            buckets = []
            total_allocation = sum(a[1] for a in allocations)
            
            for path, bandwidth in allocations:
                if path.hops[0] != ingress_switch:
                    continue
                    
                weight = int((bandwidth / total_allocation) * 100)
                next_hop = path.hops[1]
                out_port = self.topology.get_port_to_neighbor(
                    ingress_switch, next_hop
                )
                
                buckets.append({
                    "weight": weight,
                    "actions": [
                        # Optionally set path identifier in DSCP/MPLS
                        {"type": "OUTPUT", "port": out_port}
                    ]
                })
 
            group_id = hash(f"{flow_id}-split") & 0xFFFFFFFF
            
            self.controller.install_group(
                switch_id=ingress_switch,
                group_id=group_id,
                group_type="SELECT",
                buckets=buckets
            )
            
            self.controller.install_flow(
                switch_id=ingress_switch,
                priority=500,
                match=self._get_flow_match(flow_id),
                actions=[{"type": "GROUP", "group_id": group_id}]
            )
 
        # Install path rules for subsequent hops
        for path, _ in allocations:
            for i, switch_id in enumerate(path.hops[1:-1], start=1):
                next_hop = path.hops[i + 1]
                out_port = self.topology.get_port_to_neighbor(
                    switch_id, next_hop
                )
                
                self.controller.install_flow(
                    switch_id=switch_id,
                    priority=500,
                    match=self._get_flow_match(flow_id),
                    actions=[{"type": "OUTPUT", "port": out_port}]
                )
 
    def rebalance_paths(self):
        """
        Periodically rebalance flows across paths based on current utilization.
        Move flows from congested to available paths.
        """
        for (src, dst), flows in self._get_active_flows().items():
            paths = self.compute_paths(src, dst)
            
            # Check for imbalance
            utilizations = [p.current_utilization for p in paths]
            max_util = max(utilizations)
            min_util = min(utilizations)
            
            if max_util - min_util > 0.3:  # 30% imbalance threshold
                # Identify flows on congested path
                congested_path = max(paths, key=lambda p: p.current_utilization)
                flows_on_congested = self.path_flows.get(
                    congested_path.path_id, []
                )
                
                if flows_on_congested:
                    # Move smallest flow to least-loaded path
                    flow_to_move = min(
                        flows_on_congested, 
                        key=lambda f: f['demand']
                    )
                    new_path = min(paths, key=lambda p: p.current_utilization)
                    
                    self._move_flow(flow_to_move['id'], new_path)
 
    def _get_path_capacity(self, path: List[str]) -> float:
        """Get minimum link capacity along path (bottleneck)."""
        min_capacity = float('inf')
        for i in range(len(path) - 1):
            link_capacity = self.topology.get_link_capacity(path[i], path[i+1])
            min_capacity = min(min_capacity, link_capacity)
        return min_capacity
 
    def _get_path_utilization(self, path: List[str]) -> float:
        """Get maximum link utilization along path (bottleneck)."""
        max_util = 0.0
        for i in range(len(path) - 1):
            link_util = self.topology.get_link_utilization(path[i], path[i+1])
            max_util = max(max_util, link_util)
        return max_util
 
    def _get_flow_match(self, flow_id: str) -> Dict:
        """Get OpenFlow match for flow."""
        # In practice, look up flow definition
        return {"cookie": hash(flow_id)}
 
    def _get_active_flows(self) -> Dict:
        """Get currently active flows by source-destination pair."""
        return {}
 
    def _move_flow(self, flow_id: str, new_path: NetworkPath):
        """Move flow to new path."""
        pass