<gh_stars>1-10
import time
import requests

from performance.driver.core.classes import Observer
from performance.driver.core.events import Event, TeardownEvent, StartEvent, ParameterUpdateEvent
from performance.driver.core.reflection import subscribesToHint, publishesHint
from threading import Thread


class HTTPTimingResultEvent(Event):
  """
  The results of a timing event, initiated by a ``HTTPTimingObserver``
  """

  def __init__(self, url, verb, statusCode, requestTime, responseTime,
               totalTime, contentLength, *args, **kwargs):
    super().__init__(*args, **kwargs)

    #: The URL requested
    self.url = url

    #: The HTTP verb used to request this resource
    self.verb = verb

    #: The HTTP response code
    self.statusCode = statusCode

    #: The time the HTTP request took to complete
    self.requestTime = requestTime

    #: The time the HTTP response took to complete
    self.responseTime = responseTime

    #: The overall time from the beginning of the request, till the end of the
    #: response
    self.totalTime = totalTime

    #: The length of the response body
    self.contentLength = contentLength


class HTTPTimingObserver(Observer):
  """
  The *HTTP Timing Observer* is performing HTTP requests to the given endpoint
  and is measuring the request and response times.

  ::

    observers:
      - class: observer.HTTPTimingObserver

        # The URL to send the requests at
        url: http://127.0.0.1:8080/v2/apps

        # [Optional] The interval of the reqeusts (seconds)
        interval: 1

        # [Optional] The body of the HTTP request
        body: |
          {
            "cmd": "sleep 1200",
            "cpus": 0.1,
            "mem": 64,
            "disk": 0,
            "instances": {{instances}},
            "id": "/scale-instances/{{uuid()}}",
            "backoffFactor": 1.0,
            "backoffSeconds": 0
          }

        # [Optional] The HTTP Verb to use (Defaults to 'GET')
        verb: POST

        # [Optional] The HTTP headers to send
        headers:
          Accept: text/plain

  This observer is publishing a ``HTTPTimingResultEvent`` every time a sample
  is taken. Refer to the event documentation for more details.

  """

  @subscribesToHint(TeardownEvent, StartEvent, ParameterUpdateEvent)
  def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.url = self.getConfig('url')
    self.interval = float(self.getConfig('interval', '1'))
    self.clockThread = None
    self.active = False
    self.traceids = None

    # Register to the Start / Teardown events
    self.eventbus.subscribe(self.handleTeardownEvent, events=(TeardownEvent, ))
    self.eventbus.subscribe(self.handleStartEvent, events=(StartEvent, ))
    self.eventbus.subscribe(
        self.handleParameterUpdate, events=(ParameterUpdateEvent, ))

  def handleParameterUpdate(self, event):
    """
    Update trace IDs
    """
    self.traceids = event.traceids

  def handleStartEvent(self, event):
    """
    Start polling timer
    """
    self.logger.debug('Starting polling timer')
    self.active = True
    self.clockThread = Thread(target=self.pollThreadHandler, name="http-timing")
    self.clockThread.start()

  def handleTeardownEvent(self, event):
    """
    Interrupt polling timer
    """
    self.logger.debug('Stopping polling timer')
    self.active = False
    self.clockThread.join()
    self.clockThread = None

  @publishesHint(HTTPTimingResultEvent)
  def pollThreadHandler(self):
    """
    A thread that keeps polling the given url until it responds
    """

    # Render config and definitions
    config = self.getRenderedConfig()
    definitions = self.getDefinitions()

    # If we are missing an `Authorization` header but we have a
    # `dcos_auth_token` definition, allocate an `Authorization` header now
    if not 'headers' in config:
      config['headers'] = {}
    if not 'Authorization' in config['headers'] \
       and 'dcos_auth_token' in definitions:
      config['headers']['Authorization'] = 'token={}'.format(
          definitions['dcos_auth_token'])

    # Extract useful info
    url = config['url']
    body = config.get('body', None)
    headers = config['headers']
    verb = config.get('verb', 'get')

    # While running, start
    while self.active:
      self.logger.debug('Checking the endpoint')
      try:

        # Reset timer values
        times = [0, 0, 0]

        # Acknowledge response
        def ack_response(request, *args, **kwargs):
          times[1] = time.time()

        # Send request (and catch errors)
        times[0] = time.time()
        self.logger.debug('Performing HTTP {} to {}'.format(verb, url))
        res = requests.request(
            verb,
            url,
            verify=False,
            data=body,
            headers=headers,
            hooks=dict(response=ack_response))
        times[2] = time.time()

        # Log error status codes
        self.logger.debug('Completed with HTTP {}'.format(res.status_code))
        if res.status_code != 200:
          self.logger.warn('Endpoint at {} responded with HTTP {}'.format(
              url, res.status_code))

        # Broadcast status
        self.logger.debug(
            'Measurement completed: request={}, response={}, total={}'.format(
                times[1] - times[0], times[2] - times[1], times[2] - times[0]))
        self.eventbus.publish(
            HTTPTimingResultEvent(
                url,
                verb,
                res.status_code,
                times[1] - times[0],
                times[2] - times[1],
                times[2] - times[0],
                len(res.text),
                traceid=self.traceids))

      except requests.exceptions.ConnectionError as e:
        self.logger.error('Unable to connect to {}'.format(url))

      except Exception as e:
        self.logger.error(
            'An unhandled urllib exception occurred: {}'.format(e))

      # Wait for next tick
      time.sleep(self.interval)

import time
import requests

from performance.driver.core.classes import Observer
from performance.driver.core.events import Event, TeardownEvent, StartEvent, ParameterUpdateEvent
from performance.driver.core.reflection import subscribesToHint, publishesHint
from threading import Thread


class HTTPTimingResultEvent(Event):
  """
  The results of a timing event, initiated by a ``HTTPTimingObserver``
  """

  def __init__(self, url, verb, statusCode, requestTime, responseTime,
               totalTime, contentLength, *args, **kwargs):
    super().__init__(*args, **kwargs)

    #: The URL requested
    self.url = url

    #: The HTTP verb used to request this resource
    self.verb = verb

    #: The HTTP response code
    self.statusCode = statusCode

    #: The time the HTTP request took to complete
    self.requestTime = requestTime

    #: The time the HTTP response took to complete
    self.responseTime = responseTime

    #: The overall time from the beginning of the request, till the end of the
    #: response
    self.totalTime = totalTime

    #: The length of the response body
    self.contentLength = contentLength


class HTTPTimingObserver(Observer):
  """
  The *HTTP Timing Observer* is performing HTTP requests to the given endpoint
  and is measuring the request and response times.

  ::

    observers:
      - class: observer.HTTPTimingObserver

        # The URL to send the requests at
        url: http://127.0.0.1:8080/v2/apps

        # [Optional] The interval of the reqeusts (seconds)
        interval: 1

        # [Optional] The body of the HTTP request
        body: |
          {
            "cmd": "sleep 1200",
            "cpus": 0.1,
            "mem": 64,
            "disk": 0,
            "instances": {{instances}},
            "id": "/scale-instances/{{uuid()}}",
            "backoffFactor": 1.0,
            "backoffSeconds": 0
          }

        # [Optional] The HTTP Verb to use (Defaults to 'GET')
        verb: POST

        # [Optional] The HTTP headers to send
        headers:
          Accept: text/plain

  This observer is publishing a ``HTTPTimingResultEvent`` every time a sample
  is taken. Refer to the event documentation for more details.

  """

  @subscribesToHint(TeardownEvent, StartEvent, ParameterUpdateEvent)
  def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.url = self.getConfig('url')
    self.interval = float(self.getConfig('interval', '1'))
    self.clockThread = None
    self.active = False
    self.traceids = None

    # Register to the Start / Teardown events
    self.eventbus.subscribe(self.handleTeardownEvent, events=(TeardownEvent, ))
    self.eventbus.subscribe(self.handleStartEvent, events=(StartEvent, ))
    self.eventbus.subscribe(
        self.handleParameterUpdate, events=(ParameterUpdateEvent, ))

  def handleParameterUpdate(self, event):
    """
    Update trace IDs
    """
    self.traceids = event.traceids

  def handleStartEvent(self, event):
    """
    Start polling timer
    """
    self.logger.debug('Starting polling timer')
    self.active = True
    self.clockThread = Thread(target=self.pollThreadHandler, name="http-timing")
    self.clockThread.start()

  def handleTeardownEvent(self, event):
    """
    Interrupt polling timer
    """
    self.logger.debug('Stopping polling timer')
    self.active = False
    self.clockThread.join()
    self.clockThread = None

  @publishesHint(HTTPTimingResultEvent)
  def pollThreadHandler(self):
    """
    A thread that keeps polling the given url until it responds
    """

    # Render config and definitions
    config = self.getRenderedConfig()
    definitions = self.getDefinitions()

    # If we are missing an `Authorization` header but we have a
    # `dcos_auth_token` definition, allocate an `Authorization` header now
    if not 'headers' in config:
      config['headers'] = {}
    if not 'Authorization' in config['headers'] \
       and 'dcos_auth_token' in definitions:
      config['headers']['Authorization'] = 'token={}'.format(
          definitions['dcos_auth_token'])

    # Extract useful info
    url = config['url']
    body = config.get('body', None)
    headers = config['headers']
    verb = config.get('verb', 'get')

    # While running, start
    while self.active:
      self.logger.debug('Checking the endpoint')
      try:

        # Reset timer values
        times = [0, 0, 0]

        # Acknowledge response
        def ack_response(request, *args, **kwargs):
          times[1] = time.time()

        # Send request (and catch errors)
        times[0] = time.time()
        self.logger.debug('Performing HTTP {} to {}'.format(verb, url))
        res = requests.request(
            verb,
            url,
            verify=False,
            data=body,
            headers=headers,
            hooks=dict(response=ack_response))
        times[2] = time.time()

        # Log error status codes
        self.logger.debug('Completed with HTTP {}'.format(res.status_code))
        if res.status_code != 200:
          self.logger.warn('Endpoint at {} responded with HTTP {}'.format(
              url, res.status_code))

        # Broadcast status
        self.logger.debug(
            'Measurement completed: request={}, response={}, total={}'.format(
                times[1] - times[0], times[2] - times[1], times[2] - times[0]))
        self.eventbus.publish(
            HTTPTimingResultEvent(
                url,
                verb,
                res.status_code,
                times[1] - times[0],
                times[2] - times[1],
                times[2] - times[0],
                len(res.text),
                traceid=self.traceids))

      except requests.exceptions.ConnectionError as e:
        self.logger.error('Unable to connect to {}'.format(url))

      except Exception as e:
        self.logger.error(
            'An unhandled urllib exception occurred: {}'.format(e))

      # Wait for next tick
      time.sleep(self.interval)
The results of a timing event, initiated by a ``HTTPTimingObserver`` #: The URL requested #: The HTTP verb used to request this resource #: The HTTP response code #: The time the HTTP request took to complete #: The time the HTTP response took to complete #: The overall time from the beginning of the request, till the end of the #: response #: The length of the response body The *HTTP Timing Observer* is performing HTTP requests to the given endpoint   and is measuring the request and response times.    ::      observers:       - class: observer.HTTPTimingObserver          # The URL to send the requests at         url: http://127.0.0.1:8080/v2/apps          # [Optional] The interval of the reqeusts (seconds)         interval: 1          # [Optional] The body of the HTTP request         body: |           {             "cmd": "sleep 1200",             "cpus": 0.1,             "mem": 64,             "disk": 0,             "instances": {{instances}},             "id": "/scale-instances/{{uuid()}}",             "backoffFactor": 1.0,             "backoffSeconds": 0           }          # [Optional] The HTTP Verb to use (Defaults to 'GET')         verb: POST          # [Optional] The HTTP headers to send         headers:           Accept: text/plain    This observer is publishing a ``HTTPTimingResultEvent`` every time a sample   is taken. Refer to the event documentation for more details. # Register to the Start / Teardown events Update trace IDs Start polling timer Interrupt polling timer A thread that keeps polling the given url until it responds # Render config and definitions # If we are missing an `Authorization` header but we have a # `dcos_auth_token` definition, allocate an `Authorization` header now # Extract useful info # While running, start # Reset timer values # Acknowledge response # Send request (and catch errors) # Log error status codes # Broadcast status # Wait for next tick