网络测试中如何使用自动化构造随机路由模型

为什么要仿真随机路由？

路由器测试中，为了最大程度还原现网路由情况，评估路由器在现网环境下稳定工作各项指标，需要对导入路由进行离散仿真，目前路由仿真可分为导入路由与生成路由两种方式，导入路由需要现网路由表导入，本文讨论重点为生成路由方式。

自动化生成路由能解决什么问题？

使用用户界面生成路由时，可根据离散模型生成路由，但生成路由与现网路由相比，只注重路由段离散，未体现AsPath、Community等BGP路由参数离散，使用自动化生成路由可根据定义规则进行生成

如何使用自动化生成随机路由

信而泰Renix平台提供了python API接口,可使用python API进行路由灵活定义。假设路由需求如下：配置Port口1个，包含20个IBGP，个IBGP通告10个路由段、共10wIPv4+10wIPv6路由，要求路由掩码随机选择，AsPath随机选择、Connmity随机选择。

本文选择基础API使用信而泰API（renix_py_api、MiscLibrary），不另做定义，使用时需安装相关环境。代码解析如下：

导入相关库

#!/usr/bin/python

# -*- coding: UTF-8 -*-

import time

from renix_py_api.renix import *

from MiscLibrary.base import *

import logging

import random

import re

2.初始化Python环境及定义参数

if __name__ == '__main__':

# 初始化环境

print('######################初始化环境######################')

print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))

###############################################################

api = MiscAPI()

initialize(log=True, log_level=logging.INFO, log_handle=LogHandle.LOG_FILE)

###############################################################

# 占用测试仪端口

chassis_DY = "10.1.1.7"

port_DY_1 = "//10.1.1.7/3/1"

# 路由起始地址

start_ip1 = "20.0.0.0"

start_ipv61 = "2023::"

#bgp路由参数

RgpSessionCount = 20

BgpRouteBlock = 10

BgpRouteBlockv6 = 10

ipv4routecount = 100000

ipv6routecount = 100000

#ipv4路由掩码

MaskMin = 20

MaskMax = 30

# ipv6路由掩码

PrefixMin = 80

PrefixMax = 120

# bgp as_path&community限制

AsPathMaxLength = 8

CommunityMaxLength = 8

###############################################################

#其它参数

Ipv4RoutePerSession = int(ipv4routecount / RgpSessionCount)

ipv6PrefixPerSession = int(ipv6routecount / RgpSessionCount)

Ipv4CountRandonMax = int(Ipv4RoutePerSession / BgpRouteBlock)

Ipv4CountRandonMin = int(Ipv4CountRandonMax * 0.5)

Ipv6CountRandonMax = int(ipv6PrefixPerSession / BgpRouteBlockv6)

Ipv6CountRandonMin = int(Ipv6CountRandonMax * 0.5)

print('######################初始化完成######################')

print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))

3.创建端口及映射机箱

print('#######################连接机箱#######################')

print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))

sys_entry = get_sys_entry()

sys_entry.edit(ProductType=1)

chassis = ConnectChassisCommand(chassis_DY)

# chassis.execute()

# 占用端口、端口上线

port_1 = Port(upper=sys_entry, Location=port_DY_1, name='port_1')

# BringPortsOnlineCommand(PortList=[port_1.handle,port_2.handle]).execute()

print('######################连接成功！######################')

print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))

print('#######################创建IBGP#######################')

print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))

# 参数生成器定义

GeneratorRouteridv4 = api.address_modifier(Start=r'192.168.0.1', Step=1, Count=1000, Offset=0)

GeneratorRouteridv6 = api.address_modifier(Start=r'192:168::1', Step=1, Count=1000, Offset=0)

GeneratorMacAddress = api.address_modifier(Start=r'00:10:94:00:00:01', Step=1, Count=1000, Offset=0)

GeneratorIPv4Address = api.address_modifier(Start=r'10.0.0.2', Step=1, Count=1000, Offset=8)

GeneratorIPv6Address = api.address_modifier(Start=r'2000::2', Step=1, Count=1000, Offset=80)

#接口生成interface

for x in range(RgpSessionCount):

    #接口参数定义

    Routeridv4 = api.generator_next(GeneratorRouteridv4)

    Routeridv6 = api.generator_next(GeneratorRouteridv6)

    MacAddr = api.generator_next(GeneratorMacAddress)

    IPv4Addr = api.generator_next(GeneratorIPv4Address)

    IPv6Addr = api.generator_next(GeneratorIPv6Address)

    IPv4GWAddr = api.ipv4_address_hopping(IPv4Addr, Mask=32, Type='decrement', Step=1)

    IPv6GWAddr = api.ipv6_address_hopping(IPv6Addr, Mask=128, Type='decrement', Step=1)

    # 创建IPv4接口

    interface = Interface(upper=port_1, RouterId = Routeridv4, Ipv6RouterId = Routeridv6)

    Interface_temp = "Interface_" + str(x+1)

    build_Dual = BuildInterfaceCommand(InterfaceList=Interface_temp, NetworkLayers=['eth', 'vlan'], TopLayers=['ipv4', 'ipv6'])

    build_Dual.execute()

    eth_layer = interface.get_children('EthIILayer')[0]

    eth_layer.edit(Address = MacAddr)

    vlan_layer = interface.get_children('VlanLayer')[0]

    vlan_layer.edit(VlanId = x+1 )

    ipv4_layer = interface.get_children('Ipv4Layer')[0]

    ipv4_layer.edit(Address = IPv4Addr , Gateway=IPv4GWAddr)

    ipv6_layer = interface.get_children('Ipv6Layer')[0]

    ipv6_layer.edit(Address = IPv6Addr , Gateway=IPv6GWAddr)

4.创建BGP协议及路由

     # 创建BGP协议

    BgpSession = BgpProtocolConfig(upper=port_1)

    BgpSession.edit(AsNumber=65000)

    BgpSession.edit(DutAsNumber=65000)

    BgpSession.edit(UseGatewayAsDutIp=False)

    BgpSession.edit(DutIpv4Address=IPv4GWAddr)

    BgpSession.edit(DutIpv6Address=IPv6GWAddr)

    # 绑定Dual接口和协议

    select_interface = SelectInterfaceCommand(ProtocolList=[BgpSession.handle], InterfaceList=[interface.handle])

    select_interface.execute()

    # IPv4路由block创建

    FirstRoute = start_ip1

    Ipv4RouteCount = 0

    for y in range(BgpRouteBlock):

        mask = random.randint(MaskMin, MaskMax)

        if y == BgpRouteBlock-1:

            RouteCount = Ipv4RoutePerSession - Ipv4RouteCount

        else:

            RouteCount = random.randint(Ipv4CountRandonMin, Ipv4CountRandonMax)

        Ipv4RouteCount = Ipv4RouteCount+RouteCount

        offset = 32 - mask

        if x == 0 and y == 0:

            # bgp参数修改

            BgpRoute = BgpIpv4RoutepoolConfig(upper=BgpSession)

            BgpRoute.get()

            BgpRoute.edit(FirstRoute=FirstRoute)

            BgpRoute.edit(PrefixLength=mask)

            BgpRoute.edit(RouteCount=RouteCount)

            GeneratorIPv4Route = api.address_modifier(Start=FirstRoute, Step=RouteCount, Offset=offset, Count=1000)

            IPv4Route = api.generator_next(GeneratorIPv4Route)

            IPv4Route = api.generator_next(GeneratorIPv4Route)

        else:

            BgpRoute = BgpIpv4RoutepoolConfig(upper=BgpSession)

            BgpRoute.get()

            BgpRoute.edit(FirstRoute=IPv4Route)

            BgpRoute.edit(PrefixLength=mask)

            BgpRoute.edit(RouteCount=RouteCount)

            Start = IPv4Route

            GeneratorIPv4Route = api.address_modifier(Start=Start, Step=RouteCount, Offset=offset, Count=1000)

            IPv4Route = api.generator_next(GeneratorIPv4Route)

            IPv4Route = api.generator_next(GeneratorIPv4Route)

            # bgp参数修改

        as_path_length = random.randint(2, AsPathMaxLength)

        community_length = random.randint(2, CommunityMaxLength)

        as_path_list = list()

        community_list = list()

        AsPathIncrement_list = list()

        as_path_tem = str()

        community_tem = str()

        communityIncrement_list = list()

        for z in range(as_path_length):

            as_path = random.randint(300, 64000)

            as_path_list.append(as_path)

        for z in range(community_length):

            community1 = random.randint(1, 65535)

            community2 = random.randint(1, 65535)

            community = str(community1) + ':' + str(community2)

            community_list.append(community)

        for i in range(len(community_list) - 1):

            community = community_list[i]

            community_tem += community

            community_tem += ','

        community_tem += str(community_list[-1])

        AsPathPerBlockCount = int(RouteCount / 6.5)

        for z in range(len(as_path_list)):

            AsPathIncrement_list.append(1)

        for z in range(len(community_list)):

            Temp = str(1) + ':' + str(1)

            communityIncrement_list.append(Temp)

        BgpRoute.edit(AsPath=as_path_list)

        BgpRoute.edit(AsPathIncrement=AsPathIncrement_list)

        BgpRoute.edit(AsPathPerBlockCount=AsPathPerBlockCount)

        BgpRoute.edit(Community=community_tem)

        BgpRoute.edit(CommunityIncrement=communityIncrement_list)

        BgpRoute.edit(CommunityPerBlockCount=AsPathPerBlockCount)

    # IPv6路由block创建

    FirstPrefix = start_ipv61

    Ipv6PrefixCount = 0

    for y in range(BgpRouteBlockv6):

        mask = random.randint(PrefixMin, PrefixMax)

        if y == BgpRouteBlock - 1:

            RouteCount = ipv6PrefixPerSession - Ipv6PrefixCount

        else:

            RouteCount = random.randint(Ipv6CountRandonMin, Ipv6CountRandonMax)

        Ipv6PrefixCount = Ipv6PrefixCount + RouteCount

        offset = 128 - mask

        if x == 0 and y == 0:

            # bgp参数修改

            BgpRoute = BgpIpv6RoutepoolConfig(upper=BgpSession)

            BgpRoute.get()

            BgpRoute.edit(FirstIpv6Route=FirstPrefix)

            BgpRoute.edit(PrefixLength=mask)

            BgpRoute.edit(RouteCount=RouteCount)

            GeneratorIPv6Route = api.address_modifier(Start=FirstPrefix, Step=RouteCount, Offset=offset, Count=1000)

            IPv6Prefix = api.generator_next(GeneratorIPv6Route)

            IPv6Prefix = api.generator_next(GeneratorIPv6Route)

        else:

            BgpRoute = BgpIpv6RoutepoolConfig(upper=BgpSession)

            BgpRoute.get()

            BgpRoute.edit(FirstIpv6Route=IPv6Prefix)

            BgpRoute.edit(PrefixLength=mask)

            BgpRoute.edit(RouteCount=RouteCount)

            Start = IPv6Prefix

            GeneratorIPv6Route = api.address_modifier(Start=Start, Step=RouteCount, Offset=offset, Count=1000)

            IPv6Prefix = api.generator_next(GeneratorIPv6Route)

            IPv6Prefix = api.generator_next(GeneratorIPv6Route)

            # bgp参数修改

        as_path_length = random.randint(2, AsPathMaxLength)

        community_length = random.randint(2, CommunityMaxLength)

        as_path_list = list()

        community_list = list()

        AsPathIncrement_list = list()

        as_path_tem = str()

        community_tem = str()

        communityIncrement_list = list()

        for z in range(as_path_length):

            as_path = random.randint(300, 64000)

            as_path_list.append(as_path)

        for z in range(community_length):

            community1 = random.randint(1, 65535)

            community2 = random.randint(1, 65535)

            community = str(community1) + ':' + str(community2)

            community_list.append(community)

        for i in range(len(community_list) - 1):

            community = community_list[i]

            community_tem += community

            community_tem += ','

        community_tem += str(community_list[-1])

        AsPathPerBlockCount = int(RouteCount / 6.5)

        for z in range(len(as_path_list)):

            AsPathIncrement_list.append(1)

        for z in range(len(community_list)):

            Temp = str(1) + ':' + str(1)

            communityIncrement_list.append(Temp)

        BgpRoute.edit(AsPath=as_path_list)

        BgpRoute.edit(AsPathIncrement=AsPathIncrement_list)

        BgpRoute.edit(AsPathPerBlockCount=AsPathPerBlockCount)

        BgpRoute.edit(Community=community_tem)

        BgpRoute.edit(CommunityIncrement=communityIncrement_list)

        BgpRoute.edit(CommunityPerBlockCount=AsPathPerBlockCount)

# Save the test case to D: as the name of bgp_random.xcfg

save_case = SaveTestCaseCommand(TestCase=''D:/bgp_random.xcfg', ProductType=1)

save_case.execute()

随机路由生成测试

执行代码后，生成配置如下：

接口参数：

BGP Session

IPv4路由

随机变化点：路由数量、路由前缀长度、AsPath、AsPath跳变步长、AsPath数量、Community、Community跳变步长、Community数量

IPv6路由

随机变化点：路由数量、路由前缀长度、AsPath、AsPath跳变步长、AsPath数量、Community、Community跳变步长、Community数量

DarYu-X系列测试仪

DarYu-X系列高性能网络测试仪是信而泰推出的面向高端路由器等高端数通设备的测试产品，具有高性能、高密度、高速率等特点，配置信而泰基于PCT架构的新一代测试软件RENIX和X系列测试模块，可为提供路由哭喊组网测试解决方案，为建立一张高SLA保证、确定性时延、业务感知、灵活业务路径调优的下一代网络保驾护航。