<commit_before># -*- coding: utf-8 -*-
"""
 Copyright 2016 Telefonica Investigacion y Desarrollo, S.A.U

 This file is part of Orion Context Broker.

 Orion Context Broker is free software: you can redistribute it and/or
 modify it under the terms of the GNU Affero General Public License as
 published by the Free Software Foundation, either version 3 of the
 License, or (at your option) any later version.

 Orion Context Broker 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 Affero
 General Public License for more details.

 You should have received a copy of the GNU Affero General Public License
 along with Orion Context Broker. If not, see http://www.gnu.org/licenses/.

 For those usages not covered by this license please contact with
 iot_support at tid dot es
"""
__author__ = '<NAME> (ivan dot ariasleon at telefonica dot com)'

import sys
import requests
import random
import time
import datetime
import pymongo
import rpyc
import json
import logging

from requests.exceptions import ConnectionError

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s | %(levelname)s | %(message)s',
                    filename='connections_stress_tests.log',
                    filemode='w')
# disable log messages from the Requests library
logging.getLogger("requests").setLevel(logging.WARNING)


class Stablished_Connections:
    """
    Verify that ContextBroker works properly with a large number of stablished connections
    Tests procedure:
    - drop the db in mongo
    - create 5000 subscription with subject.entities.idPattern: .*
    - verify/modify the notification listener with a delay of 10 minutes before answering.
    - modify the ContextBroker config with: -httpTimeout 600000 -notificationMode threadpool:60000:5000 and restart it.
    - launch an entity update, that it triggers all subscriptions.
    - launch for a given time (in minutes) a "/version" request per second and:
         - reports that its response is correct.
         - reports the number of ESTABLISHED and CLOSE_WAIT connections and the sum of both (if -noConnectionInfo param is used this informations is not provided)
         - reports the notification queue size into ContextBroker (if `-noQueueSize` param is used this column is ignored)
    """
    # variables
    max_subscription_created = 5000
    service = u'stablished_connections'
    service_path = u'/test'
    host = u'localhost'
    port = u'1026'
    cb_endpoint = u'http://%s:%s' % (host, port)
    notif_url = 'http://localhost:8090/notify'
    mongo_host = "localhost"
    mongo_port = u'27017'
    verbose = False
    no_connections_info_flag = False
    no_queue_size_flag = False
    duration = 60       # time in minutes
    version_delay = 1   # time in seconds
    queue_size = u'N/A'
    stat_error = False
    subsc_error = False

    @staticmethod
    def __usage():
        """
        usage message
        """
        print " *****************************************************************************************************************"
        print " * This script verifies that CB works properly with a large number of stablished connections:                    *"
        print " *                                                                                                               *"
        print " *  Parameters:                                                                                                  *"
        print " *     -host=<host>         : CB host (OPTIONAL) (default: localhost).                                           *"
        print " *     -u                   : show this usage (OPTIONAL).                                                        *"
        print " *     -v                   : verbose with all responses (OPTIONAL) (default: False).                            *"
        print " *     -noConnectionInfo    : is used to ignore ESTABLISHED and CLOSE_WAIT connections (OPTIONAL)(default: False)*"
        print " *     -noQueueSize         : is used to ignore the Notification Queue Size (OPTIONAL) (default: False).         *"
        print " *     -service=<value>     : service header (OPTIONAL) (default: stablished_connections).                       *"
        print " *     -service_path=<value>: service path header (OPTIONAL) (default: /test)                                    *"
        print " *     -notif_url=<value>   : url used to notifications (OPTIONAL) (default: http://localhost:8090/notify)       *"
        print " *     -mongo=<value>       : mongo host used to clean de bd (OPTIONAL) (default: localhost)                     *"
        print " *     -duration=<value>    : test duration, value is in minutes (OPTIONAL) (default: 60 minutes)                *"
        print " *                                                                                                               *"
        print " *  Examples:                                                                                                    *"
        print " *   python connections_stress_tests.py -host=10.10.10.10 -notif_url=http://10.0.0.1:8090/notify duration=100 -v *"
        print " *                                                                                                               *"
        print " *  Note:                                                                                                        *"
        print " *    - the version delay is a second                                                                            *"
        print " *    - the number of subscriptions is 5000                                                                      *"
        print " *                                                                                                               *"
        print " *****************************************************************************************************************"
        exit(0)

    @staticmethod
    def __convert_timestamp_to_zulu(value):
        """
        convert a timestamp to a zulu date
        :param value: timestamp to convert
        :return: string (zulu date)
        """
        return str(datetime.datetime.fromtimestamp(value).strftime("%Y-%m-%dT%H:%M:%S.%fZ"))

    def __print_by_console(self, resp):
        """
        print by console http response
        resp: http response
        """
        if self.verbose:
            logging.debug("HTTP Code: %s" % str(resp.status_code))
            logging.debug("Headers:\n %s" % str(resp.headers))
            logging.debug("Body:\n %s" % resp.text)

    def __get_queue_size(self):
        """
        return the notification queue size
        :return: string (size)
        """
        try:
            resp = requests.get("%s/statistics" % self.cb_endpoint)
            self.__print_by_console(resp)
            resp_dict = json.loads(resp.text)
            self.stat_error = False
            self.queue_size = str(resp_dict["notifQueue"]["size"])
        except Exception, e:
            if not self.stat_error:
                self.stat_error = True
                logging.warn("Statistics operation failed. This is typical in the failng case scenario so noQueueSize parameter is recommended. Check documentation for details")
                self.queue_size = u'NOK'

    def __get_connection_info(self):
        """
        return the number of ESTABLISHED and CLOSE_WAIT connections per ContextBroker using "prpyc" and "psutil" libraries.
        Note: the starswith() method is used instead of just ==. This way, the process
             can have a name like "contextBroker-1.4.3..." and the logic will still work
        :return tuple (established_conn, close_wait_conn)
        """
        try:
            conn = rpyc.classic.connect(self.host)
            connections_est = """def get_number_of_connections():
                   import psutil
                   process_name = "contextBroker"
                   pid = 0
                   e_c = 0
                   cw_c = 0
                   for proc in psutil.process_iter():
                        pinfo = proc.as_dict(attrs=['pid', 'name'])
                        if pinfo["name"].startswith(process_name):
                            pid = pinfo["pid"]
                            break
                   p = psutil.Process(pid)
                   connections = p.get_connections()
                   for c in connections:
                        if c.status == "ESTABLISHED":
                            e_c += 1
                        elif c.status == "CLOSE_WAIT":
                            cw_c += 1
                   return str(e_c), str(cw_c)"""

            conn.execute(connections_est)
            remote_exec = conn.namespace['get_number_of_connections']<commit_msg>[FIX] ordering within the method<commit_after># -*- coding: utf-8 -*-
"""
 Copyright 2016 Telefonica Investigacion y Desarrollo, S.A.U

 This file is part of Orion Context Broker.

 Orion Context Broker is free software: you can redistribute it and/or
 modify it under the terms of the GNU Affero General Public License as
 published by the Free Software Foundation, either version 3 of the
 License, or (at your option) any later version.

 Orion Context Broker 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 Affero
 General Public License for more details.

 You should have received a copy of the GNU Affero General Public License
 along with Orion Context Broker. If not, see http://www.gnu.org/licenses/.

 For those usages not covered by this license please contact with
 iot_support at tid dot es
"""
__author__ = '<NAME> (ivan dot ariasleon at telefonica dot com)'

import sys
import requests
import random
import time
import datetime
import pymongo
import rpyc
import json
import logging

from requests.exceptions import ConnectionError

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s | %(levelname)s | %(message)s',
                    filename='connections_stress_tests.log',
                    filemode='w')
# disable log messages from the Requests library
logging.getLogger("requests").setLevel(logging.WARNING)


class Stablished_Connections:
    """
    Verify that ContextBroker works properly with a large number of stablished connections
    Tests procedure:
    - drop the db in mongo
    - create 5000 subscription with subject.entities.idPattern: .*
    - verify/modify the notification listener with a delay of 10 minutes before answering.
    - modify the ContextBroker config with: -httpTimeout 600000 -notificationMode threadpool:60000:5000 and restart it.
    - launch an entity update, that it triggers all subscriptions.
    - launch for a given time (in minutes) a "/version" request per second and:
         - reports that its response is correct.
         - reports the number of ESTABLISHED and CLOSE_WAIT connections and the sum of both (if -noConnectionInfo param is used this informations is not provided)
         - reports the notification queue size into ContextBroker (if `-noQueueSize` param is used this column is ignored)
    """
    # variables
    max_subscription_created = 5000
    service = u'stablished_connections'
    service_path = u'/test'
    host = u'localhost'
    port = u'1026'
    cb_endpoint = u'http://%s:%s' % (host, port)
    notif_url = 'http://localhost:8090/notify'
    mongo_host = "localhost"
    mongo_port = u'27017'
    verbose = False
    no_connections_info_flag = False
    no_queue_size_flag = False
    duration = 60       # time in minutes
    version_delay = 1   # time in seconds
    queue_size = u'N/A'
    stat_error = False
    subsc_error = False

    @staticmethod
    def __usage():
        """
        usage message
        """
        print " *****************************************************************************************************************"
        print " * This script verifies that CB works properly with a large number of stablished connections:                    *"
        print " *                                                                                                               *"
        print " *  Parameters:                                                                                                  *"
        print " *     -host=<host>         : CB host (OPTIONAL) (default: localhost).                                           *"
        print " *     -u                   : show this usage (OPTIONAL).                                                        *"
        print " *     -v                   : verbose with all responses (OPTIONAL) (default: False).                            *"
        print " *     -noConnectionInfo    : is used to ignore ESTABLISHED and CLOSE_WAIT connections (OPTIONAL)(default: False)*"
        print " *     -noQueueSize         : is used to ignore the Notification Queue Size (OPTIONAL) (default: False).         *"
        print " *     -service=<value>     : service header (OPTIONAL) (default: stablished_connections).                       *"
        print " *     -service_path=<value>: service path header (OPTIONAL) (default: /test)                                    *"
        print " *     -notif_url=<value>   : url used to notifications (OPTIONAL) (default: http://localhost:8090/notify)       *"
        print " *     -mongo=<value>       : mongo host used to clean de bd (OPTIONAL) (default: localhost)                     *"
        print " *     -duration=<value>    : test duration, value is in minutes (OPTIONAL) (default: 60 minutes)                *"
        print " *                                                                                                               *"
        print " *  Examples:                                                                                                    *"
        print " *   python connections_stress_tests.py -host=10.10.10.10 -notif_url=http://10.0.0.1:8090/notify duration=100 -v *"
        print " *                                                                                                               *"
        print " *  Note:                                                                                                        *"
        print " *    - the version delay is a second                                                                            *"
        print " *    - the number of subscriptions is 5000                                                                      *"
        print " *                                                                                                               *"
        print " *****************************************************************************************************************"
        exit(0)

    @staticmethod
    def __convert_timestamp_to_zulu(value):
        """
        convert a timestamp to a zulu date
        :param value: timestamp to convert
        :return: string (zulu date)
        """
        return str(datetime.datetime.fromtimestamp(value).strftime("%Y-%m-%dT%H:%M:%S.%fZ"))

    def __print_by_console(self, resp):
        """
        print by console http response
        resp: http response
        """
        if self.verbose:
            logging.debug("HTTP Code: %s" % str(resp.status_code))
            logging.debug("Headers:\n %s" % str(resp.headers))
            logging.debug("Body:\n %s" % resp.text)

    def __get_queue_size(self):
        """
        return the notification queue size
        :return: string (size)
        """
        try:
            resp = requests.get("%s/statistics" % self.cb_endpoint)
            self.__print_by_console(resp)
            resp_dict = json.loads(resp.text)
            self.stat_error = False
            self.queue_size = str(resp_dict["notifQueue"]["size"])
        except Exception, e:
            if not self.stat_error:
                self.stat_error = True
                logging.warn("Statistics operation failed. This is typical in the failng case scenario so noQueueSize parameter is recommended. Check documentation for details")
                self.queue_size = u'NOK'

    def __get_connection_info(self):
        """
        return the number of ESTABLISHED and CLOSE_WAIT connections per ContextBroker using "prpyc" and "psutil" libraries.
        Note: the starswith() method is used instead of just ==. This way, the process
             can have a name like "contextBroker-1.4.3..." and the logic will still work
        :return tuple (established_conn, close_wait_conn)
        """
        connections_est = """def get_number_of_connections():
                   import psutil
                   process_name = "contextBroker"
                   pid = 0
                   e_c = 0
                   cw_c = 0
                   for proc in psutil.process_iter():
                        pinfo = proc.as_dict(attrs=['pid', 'name'])
                        if pinfo["name"].startswith(process_name):
                            pid = pinfo["pid"]
                            break
                   p = psutil.Process(pid)
                   connections = p.get_connections()
                   for c in connections:
                        if c.status == "ESTABLISHED":
                            e_c += 1
                        elif c.status == "CLOSE_WAIT":
                            cw_c += 1
                   return str(e_c), str(cw_c)"""
        try:
            conn = rpyc.classic.connect(self.host)
            conn.execute(connections_est)
            remote_exec = conn.namespace['get_number_of_connections']
{'fraction_non_alphanumeric': 0.10209336169260047, 'fraction_numerical': 0.00962722489645136, 'mean_word_length': 1.5444317858160068, 'pattern_counts': {'":': 4, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 86, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4906700', 'n_tokens_mistral': 4306, 'n_tokens_neox': 4071, 'n_words': 1375}