Commit 219a7995 authored by Nigel Kukard's avatar Nigel Kukard

Initial commit

parents
This diff is collapsed.
[build-system]
requires = ["setuptools", "pytest", "pytest-runner", "pytest-cov", "pylama", "mccabe"]
# Copyright (c) 2019, AllWorldIT
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
[aliases]
test = pytest
[tool:pytest]
addopts = --pylama --cov=src/ tests/ src/
[coverage:run]
branch = true
parallel = true
[coverage:report]
show_missing = true
exclude_lines =
pragma: no cover
def __repr__
if self.debug:
raise AssertionError
raise NotImplementedError
if __name__ == ['"]__main__['"]:$
[pylama]
linters = pep257,pycodestyle,pyflakes,pylint,mccabe,mypy,isort
# D202: No blank lines allowed after function docstring
# D203: 1 blank line required before class docstring
# D213: Multi-line docstring summary should start at the second line
# R0201: Method could be a function
# R0903: Too few public methods
ignore = D202,D203,D213,R0201,R0903
[pylama:pycodestyle]
max_line_length = 132
[pylama:pylint]
max_line_length = 132
[mypy]
ignore_missing_imports = true
# Copyright (c) 2019, AllWorldIT
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Open Traffic Shaper."""
from setuptools import find_packages, setup
NAME = 'OpenTrafficShaper'
VERSION = '0.0.1'
setup(
name=NAME,
version=VERSION,
packages=find_packages('src', exclude=['tests']),
package_dir={'': 'src'},
entry_points={
'console_scripts': [
'awit-interface-shaper=opentrafficshaper.interface:main',
],
},
)
# Copyright (c) 2019, AllWorldIT
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Open Traffic Shaper package."""
# Copyright (c) 2019, AllWorldIT
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Interface traffic shaping."""
import argparse
import subprocess
from typing import List, Optional
__version__ = '0.0.1'
def run_tc(*args: List[str], ignore_ret=False):
"""Run tc."""
tc_args = ['tc']
tc_args.extend(args)
try:
subprocess.check_output(tc_args, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as exception:
# Check if we ignoring the result
if not ignore_ret:
print('ERROR: %s' % exception.output.decode('utf-8'))
exit(1)
def set_rate_out(interface: str, rate_out: int):
"""Set outbound rate limit."""
speed_low = float(rate_out) * 0.95
speed_high = float(rate_out) * 0.98
# Set packet limits to a sane value
limit_packets = int(float(rate_out) * 50)
if limit_packets < 100:
limit_packets = 100
elif limit_packets > 10000:
limit_packets = 10000
# Set flows based on the speed
flows = int(float(limit_packets) / 10)
# Set flow depth
depth = flows * 2
# Set hash size
hash_size = 2**(limit_packets - 1).bit_length()
if hash_size < 1024:
hash_size = 1024
# Flow depth * half packet size * 30% queue size
redflowlimit = int(depth * 750)
print(f'Rate out........: {rate_out}Mbit')
print(f'Speed low.......: {speed_low}Mbit')
print(f'Speed high......: {speed_high}Mbit')
print(f'Limit packets...: {limit_packets}')
print(f'Hash size.......: {hash_size}')
print(f'Flows...........: {flows}')
print(f'Flow depth......: {depth}')
print(f'Flow threshold..: {redflowlimit}')
# Clear root qdisc
run_tc('qdisc', 'del', 'dev', interface, 'root', ignore_ret=True)
# Add root qdisc
run_tc('qdisc', 'add', 'dev', interface, 'root', 'handle', '1:', 'hfsc', 'default', '1')
# Add class
run_tc('class', 'add', 'dev', interface, 'parent', '1:', 'classid', '1:1',
'hfsc', 'sc', 'rate', f'{speed_low}mbit', 'ul', 'rate', f'{speed_high}mbit')
# Add SFQ qdisc
run_tc('qdisc', 'add', 'dev', interface, 'parent', '1:1', 'handle', '10:',
'sfq', 'perturb', '10',
'divisor', f'{hash_size}',
'limit', f'{limit_packets}',
'depth', f'{depth}',
'flows', f'{flows}',
'redflowlimit', f'{redflowlimit}',
'ecn')
def set_rate_in(interface: str, rate_in: Optional[int]):
"""Set inbound rate limit."""
run_tc('qdisc', 'del', 'dev', interface, 'ingress', ignore_ret=True)
if rate_in:
burst = int((float(rate_in) * 1024) / 8 / 2)
print(f'Rate in.........: {rate_in}Mbit')
print(f'Burst size......: {burst}k')
# Add qdisc ingress
run_tc('qdisc', 'add', 'dev', interface, 'ingress')
# Add filter for inbound policing
run_tc('filter', 'add', 'dev', interface, 'parent', 'ffff:',
'u32', 'match', 'u32', '0', '0',
'police', 'rate', f'{rate_in}mbit', 'burst', f'{burst}k', 'mtu', '64kb', 'drop')
def main():
"""Entry point for execution from the commandline."""
print(f'AllWorldIT Interface Shaper v{__version__} - Copyright © 2019, AllWorldIT.\n')
# Start argument parser
argparser = argparse.ArgumentParser(add_help=False)
# Create argument group for rate limit options
rate_group = argparser.add_argument_group('Rate limit options')
rate_group.add_argument('interface', metavar='IFACE', nargs=1,
help='Interface name')
rate_group.add_argument('rate_out', metavar='RATE-OUT', nargs=1,
help='Outbound traffic rate limit')
rate_group.add_argument('rate_in', metavar='RATE-IN', nargs='?',
help='Inbound traffic rate limit')
# Create argument group for optionals
optional_group = argparser.add_argument_group('Optional arguments')
optional_group.add_argument('-h', '--help', action="help", help="Show this help message and exit")
# Parse args
args = argparser.parse_args()
# Grab interfac
interface = args.interface[0]
# Default rate_in to None
rate_in = None
if args.rate_in:
rate_in = args.rate_in[0]
# Set rate out and rate in
print(f'Interface.......: {interface}\n')
set_rate_out(args.interface[0], args.rate_out[0])
print('')
set_rate_in(args.interface[0], rate_in)
# Copyright (c) 2019, AllWorldIT
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Dummy test."""
from typing import List
class TestDummy():
"""Dummy test."""
callback_lines: List[str]
callback_results: List[str]
def test_dummy(self):
"""Dummy test."""
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment