Gecko [ sachasauda.akalacademy.ac.in ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	0 B	-rw-r--r--
agent.py	96.8 KB	-rw-r--r--
agent_exceptions.py	110 B	-rw-r--r--
agent_util.py	8.58 KB	-rw-r--r--
aggregator.py	14.89 KB	-rw-r--r--
anomaly.py	2.19 KB	-rw-r--r--
blacklister.py	809 B	-rw-r--r--
container_discovery.py	3.3 KB	-rw-r--r--
display.py	2.06 KB	-rw-r--r--
forticlient_helper.py	2.59 KB	-rw-r--r--
inspector.py	15.7 KB	-rw-r--r--
iperf3.py	2.12 KB	-rw-r--r--
log_matcher.py	4.27 KB	-rw-r--r--
maintenance.py	3.61 KB	-rw-r--r--
pickle_database.py	1.28 KB	-rw-r--r--
plugin_driver.py	4.78 KB	-rw-r--r--
plugin_manager.py	11.04 KB	-rw-r--r--
process_manager.py	851 B	-rw-r--r--
progress_printer.py	837 B	-rw-r--r--
result_queue.py	1.99 KB	-rw-r--r--
schedule.py	3.19 KB	-rw-r--r--
threshold.py	1.5 KB	-rw-r--r--

Code Editor : iperf3.py

import json
import logging
import os
import subprocess
import requests

class Iperf3Runner:
    default_start_url = 'https://speed.fortisase.com/start_iperf'
    cached_start_info = None
    def __init__(self, iperf3_bin, start_url=None):
        if start_url is None:
            start_url = self.default_start_url
        self.start_url = start_url
        self.iperf_bin = iperf3_bin
        stat = os.stat(self.iperf_bin)
        perms = oct(stat.st_mode)[-3:]
        if '744' != perms:
            os.chmod(iperf3_bin, 0o744)
        self.log = logging.getLogger('Agent')

def _fetch_info(self):
        if self.cached_start_info is not None:
            return self.cached_start_info
        r = requests.get(
            self.start_url,
            timeout=10
        )
        if 200 != r.status_code:
            raise Exception(f'Start_perf failure: {r.status_code}')
        self.cached_start_info = r.json()
        return self.cached_start_info

def _run_test(self, iperf_args, response_key):
        p = subprocess.run(
            iperf_args,
            shell=True,
            capture_output=True,
            check=True,
            encoding='utf-8'
        )
        response = json.loads(p.stdout)
        bps = response['end']['streams'][0][response_key]['bits_per_second']
        metric_name = 'download speed'
        if response_key == 'sender':
            metric_name = 'upload speed'
        self.log.info('Metric {} args {} -> {} bits/s'.format(
            metric_name,
            iperf_args,
            bps
        ))
        return bps
            
    def get_download_speed(self):
        info = self._fetch_info()
        args = '{} -c {} -p {} -t 10 -R --json'.format(
            self.iperf_bin,
            info['ip'],
            info['upload_port']
        )
        return self._run_test(iperf_args=args, response_key='receiver')

def get_upload_speed(self):
        info = self._fetch_info()
        args = '{} -c {} -p {} -t 10 --json'.format(
            self.iperf_bin,
            info['ip'],
            info['download_port']
        )
        return self._run_test(iperf_args=args, response_key='sender')

${this.title}

Code Editor : iperf3.py