path: root/resources/traffic_scripts/honeycomb/read_vpp_version.py

# Copyright (c) 2017 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import socket
import multiprocessing
import argparse
from time import time


class Counter(object):
    """Counter used for stats collection."""
    def __init__(self, start=0):
        """Initializer."""
        self.lock = multiprocessing.Lock()
        self.value = start

    def increment(self, value=1):
        """Increment counter and return the new value."""
        self.lock.acquire()
        val = self.value
        try:
            self.value += value
        finally:
            self.lock.release()
        return val


class timer(object):
    """Timer used used during test execution."""
    def __init__(self, verbose=False):
        self.verbose = verbose

    def __enter__(self):
        """Start the timer."""
        self.start = time()
        return self

    def __exit__(self, *args):
        """Stop the timer and save current value."""
        self.end = time()
        self.secs = self.end - self.start
        self.msecs = self.secs * 1000  # millisecs
        if self.verbose:
            print("elapsed time: {0} ms".format(self.msecs))


class ConfigBlaster(object):
    """Generates Netconf requests, receives replies and collects statistics."""

    TIMEOUT = 10

    # Hello message with capabilities list for Netconf sessions.
    hello = u"""<hello xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"
    message-id="m-0">
    <capabilities>
    <capability>urn:ietf:params:netconf:base:1.0</capability>
    </capabilities>
    </hello>
    ]]>]]>"""

    # RPC to retrieve VPP version (minimal processing in VPP)
    request_template = u"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"
    message-id="m-1">
    <get>
    <filter xmlns:ns0="urn:ietf:params:xml:ns:netconf:base:1.0"
    ns0:type="subtree">
    <vpp-state xmlns="urn:opendaylight:params:xml:ns:yang:vpp:management">
    <version/>
    </vpp-state>
    </filter>
    </get>
    </rpc>
    ]]>]]>"""

    class Stats(object):
        """Stores and further processes statistics collected by worker
        threads during their execution.
        """

        def __init__(self):
            """Initializer."""
            self.ok_rqst_rate = Counter(0)
            self.total_rqst_rate = Counter(0)
            self.ok_rqsts = Counter(0)
            self.total_rqsts = Counter(0)

        def process_stats(self, rqst_stats, elapsed_time):
            """Calculates the stats for request/reply throughput, and aggregates
            statistics across all threads.

            :param rqst_stats: Request statistics dictionary.
            :param elapsed_time: Elapsed time for the test.
            :type rqst_stats: dict
            :type elapsed_time: int
            :returns: Rates (requests/sec) for successfully finished requests
                     and the total number of requests.
            :rtype: tuple
            """
            ok_rqsts = rqst_stats["OK"]
            total_rqsts = sum(rqst_stats.values())

            ok_rqst_rate = ok_rqsts / elapsed_time
            total_rqst_rate = total_rqsts / elapsed_time

            self.ok_rqsts.increment(ok_rqsts)
            self.total_rqsts.increment(total_rqsts)

            self.ok_rqst_rate.increment(ok_rqst_rate)
            self.total_rqst_rate.increment(total_rqst_rate)

            return ok_rqst_rate, total_rqst_rate

        @property
        def get_ok_rqst_rate(self):
            return self.ok_rqst_rate.value

        @property
        def get_total_rqst_rate(self):
            return self.total_rqst_rate.value

        @property
        def get_ok_rqsts(self):
            return self.ok_rqsts.value

        @property
        def get_total_rqsts(self):
            return self.total_rqsts.value

    def __init__(self, host, port, ncycles, nthreads, nrequests):
        """Initializer.

        :param host: Target IP address.
        :param port: Target port.
        :param ncycles: Number of test cycles.
        :param nthreads: Number of threads for packet generation.
        :param nrequests: Number of requests to send per thread.
        :type host: str
        :type port: int
        :type ncycles: int
        :type nthreads: int
        :type nrequests: int
        """

        self.host = host
        self.port = port
        self.ncycles = ncycles
        self.nthreads = nthreads
        self.nrequests = nrequests

        self.stats = self.Stats()
        self.total_ok_rqsts = 0

        self.print_lock = multiprocessing.Lock()
        self.cond = multiprocessing.Condition()
        self.threads_done = 0

        self.recv_buf = 8192

    def send_request(self, sock):
        """Send Netconf request and receive the reply.

        :param sock: Socket object to use for transfer.
        :type sock: socket object
        :returns: Response to request or error message.
        :rtype: str
        """

        sock.send(self.request_template)
        try:
            return sock.recv(self.recv_buf)
        except socket.timeout:
            return "timeout"
        except socket.error:
            return "error"

    def send_requests(self, tid, stats):
        """Read entries from the Honeycomb operational data store. This function
        is executed by a worker thread.

        :param tid: Thread ID - used to id the Blaster thread when
        statistics for the thread are printed out.
        :param stats: Synchronized queue object for returning execution stats.
        :type tid: int
        :type stats: multiprocessing.Queue
        """

        rqst_stats = {"OK": 0, "Error": 0, "Timeout": 0}

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        # Initiate connection
        sock.connect((self.host, self.port))
        # Send hello message
        sock.send(self.hello)
        # Receive hello message
        sock.recv(self.recv_buf)
        # Determine length of expected responses
        self.recv_buf = len(self.send_request(sock))

        with self.print_lock:
            print("\n    Thread {0}:\n"
                  "        Sending {1} requests".format(tid,
                                                        self.nrequests))

        replies = [None]*self.nrequests
        with timer() as t:
            for x in range(self.nrequests):
                sts = self.send_request(sock)
                replies[x] = sts

        for reply in replies:
            if reply == "timeout":
                rqst_stats["Timeout"] += 1
            elif "error" in reply:
                rqst_stats["Error"] += 1
            else:
                rqst_stats["OK"] += 1

        ok_rps, total_rps = self.stats.process_stats(
            rqst_stats, t.secs)

        with self.print_lock:
            print("\n    Thread {0} results (READ): ".format(tid))
            print("        Elapsed time: {0:.2f}s,".format(t.secs))
            print("        Requests/s: {0:.2f} OK, {1:.2f} Total".format(
                ok_rps, total_rps))
            print("        Stats ({Requests}, {entries}): "),
            print(rqst_stats)
            self.threads_done += 1

        sock.close()

        stats.put({"stats": rqst_stats, "time": t.secs})

        with self.cond:
            self.cond.notify_all()

    def run_cycle(self, function):
        """Runs a test cycle. Each test consists of <cycles> test cycles, where
        <threads> worker threads are started in each test cycle. Each thread
        reads <requests> entries using Netconf RPCs.

        :param function: Function to be executed in each thread.
        :type function: function
        :return: None
        """

        self.total_ok_rqsts = 0
        stats_queue = multiprocessing.Queue()

        for c in range(self.ncycles):
            self.stats = self.Stats()
            with self.print_lock:
                print "\nCycle {0}:".format(c)

            threads = []
            thread_stats = []
            for i in range(self.nthreads):
                t = multiprocessing.Process(target=function,
                                            args=(i, stats_queue))
                threads.append(t)
                t.start()

            # Wait for all threads to finish and measure the execution time
            with timer() as t:
                for _ in threads:
                    thread_stats.append(stats_queue.get())
                for thread in threads:
                    thread.join()

            for item in thread_stats:
                self.stats.process_stats(item["stats"], item["time"])

            with self.print_lock:
                print("\n*** Test summary:")
                print("    Elapsed time:    {0:.2f}s".format(t.secs))
                print(
                    "    Peak requests/s: {0:.2f} OK, {1:.2f} Total".format(
                        self.stats.get_ok_rqst_rate,
                        self.stats.get_total_rqst_rate))
                print(
                    "    Avg. requests/s: {0:.2f} OK, {1:.2f} Total ({2:.2f} "
                    "of peak total)".format(
                        self.stats.get_ok_rqsts / t.secs,
                        self.stats.get_total_rqsts / t.secs,
                        (self.stats.get_total_rqsts / t.secs * 100) /
                        self.stats.get_total_rqst_rate))

            self.total_ok_rqsts += self.stats.get_ok_rqsts

            self.threads_done = 0

    def add_blaster(self):
        """Run the test."""
        self.run_cycle(self.send_requests)

    @property
    def get_ok_rqsts(self):
        return self.total_ok_rqsts


def create_arguments_parser():
    """Creates argument parser for test script.
    Shorthand to arg parser on library level in order to access and
    eventually enhance in ancestors.

    :returns: argument parser supporting arguments and parameters
    :rtype: argparse.ArgumentParser
    """
    my_parser = argparse.ArgumentParser(
        description="entry reading performance test: Reads entries from "
                    "the config tree, as specified by optional parameters.")

    my_parser.add_argument(
        "--host", default="127.0.0.1",
        help="Host where odl controller is running (default is 127.0.0.1).")
    my_parser.add_argument(
        "--port", default=7777,
        help="Port on which Honeycomb's Netconf is listening"
             " (default is 7777 for TCP)")
    my_parser.add_argument(
        "--cycles", type=int, default=1,
        help="Number of entry read cycles; default 1. <THREADS> worker threads "
             "are started in each cycle and the cycle ends when all threads "
             "finish. Another cycle is started when the previous cycle "
             "is finished.")
    my_parser.add_argument(
        "--threads", type=int, default=1,
        help="Number of request worker threads to start in each cycle; "
             "default=1. Each thread will read <entries> entries.")
    my_parser.add_argument(
        "--requests", type=int, default=10,
        help="Number of requests that will be made by each worker thread "
             "in each cycle; default 10")

    return my_parser

if __name__ == "__main__":

    parser = create_arguments_parser()
    in_args = parser.parse_args()

    fct = ConfigBlaster(in_args.host, in_args.port, in_args.cycles,
                        in_args.threads, in_args.requests)

    # Run through <cycles>, where <threads> are started in each cycle and
    # <entries> are added from each thread
    fct.add_blaster()

    print "    Successful reads:  {0}\n".format(fct.get_ok_rqsts)