<gh_stars>0
# Copyright (c) Microsoft. All rights reserved.
# Licensed under the MIT license. See LICENSE file in the project root for
# full license information.
import logging
import os
import time
import uuid
import json
import datetime
import threading
import pprint
import sys
import collections
import platform
import random
from executor import BetterThreadPoolExecutor, reset_watchdog, dump_active_stacks
import dps
import queue
import faulthandler
import gc
from azure.iot.device import Message, MethodResponse
import azure.iot.device.constant
from measurement import ThreadSafeCounter
from system_health_telemetry import SystemHealthTelemetry
import azure_monitor
from azure_monitor_metrics import MetricsReporter
from thief_constants import (
    Const,
    Fields,
    Commands,
    Events,
    Metrics,
    Settings,
    CustomDimensions,
    MethodNames,
    RunStates,
    SystemProperties,
)
from operation_tickets import OperationTicketList
import thief_secrets

faulthandler.enable()

provisioning_host = thief_secrets.DEVICE_PROVISIONING_HOST
id_scope = thief_secrets.DEVICE_ID_SCOPE
group_symmetric_key = thief_secrets.DEVICE_GROUP_SYMMETRIC_KEY
registration_id = thief_secrets.DEVICE_ID
requested_service_pool = thief_secrets.REQUESTED_SERVICE_POOL

# run_reason can be an environment variable or is can be passed on the command line.
run_reason = os.getenv("RUN_REASON")
# run_id can be an environment variable or it can be automatically generated
run_id = str(uuid.uuid4())

# set default logging which will only go to the console

logging.basicConfig(level=logging.WARNING)
# logging.getLogger("paho").setLevel(level=logging.DEBUG)
logging.getLogger("thief").setLevel(level=logging.INFO)
# logging.getLogger("azure.iot").setLevel(level=logging.INFO)
logger = logging.getLogger("thief.{}".format(__name__))

# configure our traces and events to go to Azure Monitor
azure_monitor.add_logging_properties(
    client_type="device",
    run_id=run_id,
    sdk_version=azure.iot.device.constant.VERSION,
    transport="mqtt",
    pool_id=requested_service_pool,
)
event_logger = azure_monitor.get_event_logger()
azure_monitor.log_to_azure_monitor("thief")


TestMethod = collections.namedtuple(
    "TestMethod", "method_name expected_status_code include_payload"
)


class ThiefFatalException(Exception):
    pass


def _get_os_release_based_on_user_agent_standard():
    return "({python_runtime};{os_type} {os_release};{architecture})".format(
        python_runtime=platform.python_version(),
        os_type=platform.system(),
        os_release=platform.version(),
        architecture=platform.machine(),
    )


def custom_props(extra_props):
    """
    helper function for adding customDimensions to logger calls at execution time
    """
    return {"custom_dimensions": extra_props}


class DeviceRunMetrics(object):
    """
    Object we use internally to keep track of how a the entire test is performing.
    """

    def __init__(self):
        self.run_start_utc = None
        self.run_state = RunStates.WAITING
        self.exit_reason = None

        self.exception_count = ThreadSafeCounter()

        self.send_message_count_queued = ThreadSafeCounter()
        self.send_message_count_sent = ThreadSafeCounter()
        self.send_message_count_verified = ThreadSafeCounter()
        self.send_message_count_timed_out = ThreadSafeCounter()
        self.receive_c2d_count_received = ThreadSafeCounter()
        self.receive_c2d_count_timed_out = ThreadSafeCounter()

        self.reported_properties_count_added = ThreadSafeCounter()
        self.reported_properties_count_removed = ThreadSafeCounter()
        self.reported_properties_count_timed_out = ThreadSafeCounter()

        self.get_twin_count_succeeded = ThreadSafeCounter()
        self.get_twin_count_timed_out = ThreadSafeCounter()

        self.desired_property_patch_count_received = ThreadSafeCounter()
        self.desired_property_patch_count_timed_out = ThreadSafeCounter()

        self.method_invoke_count_request_received = ThreadSafeCounter()
        self.method_invoke_count_request_timed_out = ThreadSafeCounter()


"""
Object we use internally to keep track of how the entire test is configured.
Currently hardcoded. Later, this will come from desired properties.
"""
device_run_config = {
    Settings.MAX_RUN_DURATION_IN_SECONDS: 12 * 60 * 60,
    Settings.ALLOWED_EXCEPTION_COUNT: 10,
    Settings.INTER_TEST_DELAY_INTERVAL_IN_SECONDS: 1,
    Settings.OPERATION_TIMEOUT_IN_SECONDS: 300,
    Settings.OPERATION_TIMEOUT_ALLOWED_FAILURE_COUNT: 10,
    Settings.PAIRING_REQUEST_TIMEOUT_INTERVAL_IN_SECONDS: 900,
    Settings.PAIRING_REQUEST_SEND_INTERVAL_IN_SECONDS: 30,
    Settings.SEND_MESSAGE_OPERATIONS_PER_SECOND: 1,
    Settings.MQTT_KEEP_ALIVE_INTERVAL: 20,
    Settings.SAS_TOKEN_RENEWAL_INTERVAL: 600,
    Settings.MAX_TEST_SEND_THREADS: 128,
    Settings.MAX_TEST_RECEIVE_THREADS: 10,
    Settings.DO_SEND_MESSAGE_FLOODS: True,
    Settings.SEND_MESSAGE_FLOOD_INTERVAL_IN_SECONDS: 600,
    Settings.SEND_MESSAGE_FLOOD_MAX_MESSAGE_COUNT: 500,
}


def return_if_already_running(func):
    """
    Decorator function to make sure that the wrapped function cannot be called in parallel.
    The wrapped funciton can be invoked once.  If is called again before the first invocation
    is complete, it immediately returns without running.  Once the first invocation completes,
    then it an be invoked again.
    """
    already_running = threading.BoundedSemaphore()

    def inner(*args, **kwargs):
        if already_running.acquire(blocking=False):
            try:
                return func(*args, **kwargs)
            finally:
                already_running.release()

    return inner


class DeviceApp(object):
    """
    Main application object
    """

    def __init__(self):
        global device_run_config
        super(DeviceApp, self).__init__()

        self.config = device_run_config
        self.outgoing_executor = BetterThreadPoolExecutor(
            max_workers=self.config[Settings.MAX_TEST_SEND_THREADS]
        )
        self.incoming_executor = BetterThreadPoolExecutor(
            max_workers=self.config[Settings.MAX_TEST_RECEIVE_THREADS]
        )
        self.done = threading.Event()
        self.client = None
        self.hub = None
        self.device_id = None
        self.metrics = DeviceRunMetrics()
        self.service_instance_id = None
        self.system_health_telemetry = SystemHealthTelemetry()
        # for running operations
        self.operation_ticket_list = OperationTicketList()
        # for metrics
        self.reporter_lock = threading.Lock()
        self.reporter = MetricsReporter()
        self._configure_azure_monitor_metrics()
        # for pairing
        self.incoming_pairing_message_queue = queue.Queue()
        # for twin
        self.reported_property_index = 1
        self.incoming_desired_property_patch_queue = queue.Queue()

    def _configure_azure_monitor_metrics(self):
        # ---------------------
        # System Health metrics
        # ---------------------
        self.reporter.add_float_measurement(
            Metrics.PROCESS_CPU_PERCENT,
            "CPU use for the device app, as a percentage of all cores",
            "percentage",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_WORKING_SET_BYTES,
            "Working set for the device app, includes shared and private, read-only and writeable memory",
            "bytes",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_WORKING_SET_PRIVATE_BYTES,
            "Amount of private data used by the process",
            "bytes",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_GARBAGE_COLLECTION_OBJECTS,
            "Number of active objects being managed by the garbage collector",
            "objects",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_SDK_THREADS,
            "number of threads being used by the current SDK",
            "threads",
        )
        self.reporter.add_integer_measurement(
            Metrics.TEST_APP_ACTIVE_RECEIVE_THREADS,
            "number of active threads in the test app receive threadpool",
            "threads",
        )
        self.reporter.add_integer_measurement(
            Metrics.TEST_APP_ACTIVE_SEND_THREADS,
            "number of active threads in the test app send threadpool",
            "threads",
        )

        # ----------------
        # test app metrics
        # ----------------
        self.reporter.add_integer_measurement(
            Metrics.EXCEPTION_COUNT,
            "Number of (non-fatal) exceptions raised by the client library or test code",
            "exception count",
        )

        # --------------------
        # SendMesssage metrics
        # --------------------
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_QUEUED,
            "Number of telemetry messages queued for sending",
            "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_SENT, "Number of telemetry messages sent", "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_VERIFIED,
            "Number of telemetry messages sent and verified by the service",
            "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_TIMED_OUT,
            "Number of telemetry messages that timed out with no response from the service",
            "messages",
        )

        # -------------------
        # Receive c2d metrics
        # -------------------
        self.reporter.add_integer_measurement(
            Metrics.RECEIVE_C2D_COUNT_RECEIVED, "Number of c2d messages received", "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.RECEIVE_C2D_COUNT_TIMED_OUT,
            "Number of c2d messages not received in time",
            "messages",
        )

        # -------------------------
        # Reported property metrics
        # -------------------------
        self.reporter.add_integer_measurement(
            Metrics.REPORTED_PROPERTIES_COUNT_ADDED,
            "Number of reported properties which have been added",
            "patches",
        )
        self.reporter.add_integer_measurement(
            Metrics.REPORTED_PROPERTIES_COUNT_REMOVED,
            "Number of reported properties which have been removed",
            "patches",
        )
        self.reporter.add_integer_measurement(
            Metrics.REPORTED_PROPERTIES_COUNT_TIMED_OUT,
            "Number of reported property add & remove operations that timed out",
            "patches",
        )

        # ----------------
        # Get-twin metrics
        # ----------------
        self.reporter.add_integer_measurement(
            Metrics.GET_TWIN_COUNT_SUCCEEDED,
            "Number of times get_twin successfully verified a property update",
            "calls",
        )
        self.reporter.add_integer_measurement(
            Metrics.GET_TWIN_COUNT_TIMED_OUT,
            "Number of times get_twin was unable to verify a property update",
            "calls",
        )

        # ------------------------------
        # desired property patch metrics
        # ------------------------------
        self.reporter.add_integer_measurement(
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_RECEIVED,
            "Count of desired property patches that were successfully received",
            "patches",
        )
        self.reporter.add_integer_measurement(
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_TIMED_OUT,
            "Count of desired property patches that were not received",
            "patches",
        )

        # ---------------------
        # method invoke metrics
        # ---------------------
        self.reporter.add_integer_measurement(
            Metrics.METHOD_INVOKE_COUNT_REQUEST_RECEIVED,
            "Number of method invoke requests received",
            "invokes",
        )
        self.reporter.add_integer_measurement(
            Metrics.METHOD_INVOKE_COUNT_REQUEST_TIMED_OUT,
            "Number of method invoke requests timed out",
            "invokes",
        )

        # ---------------
        # Latency metrics
        # ---------------
        self.reporter.add_float_measurement(
            Metrics.LATENCY_QUEUE_MESSAGE_TO_SEND,
            "Number of milliseconds between queueing a telemetry message and actually sending it",
            "milliseconds",
        )

    def get_session_metrics(self):
        """
        Return metrics which describe the session the tests are running in
        """
        now = datetime.datetime.now(datetime.timezone.utc)
        elapsed_time = now - self.metrics.run_start_utc

        props = {
            Fields.RUN_START_UTC: self.metrics.run_start_utc.isoformat(),
            Fields.LATEST_UPDATE_TIME_UTC: now.isoformat(),
            Fields.ELAPSED_TIME: str(elapsed_time),
            Fields.RUN_STATE: str(self.metrics.run_state),
            Fields.EXIT_REASON: self.metrics.exit_reason,
        }
        return props

    def get_test_metrics(self):
        """
        Return metrics which describe the progress of the  different features being tested
        """

        props = {
            Metrics.EXCEPTION_COUNT: self.metrics.exception_count.get_count(),
            Metrics.SEND_MESSAGE_COUNT_QUEUED: self.metrics.send_message_count_queued.get_count(),
            Metrics.SEND_MESSAGE_COUNT_SENT: self.metrics.send_message_count_sent.get_count(),
            Metrics.SEND_MESSAGE_COUNT_VERIFIED: self.metrics.send_message_count_verified.get_count(),
            Metrics.SEND_MESSAGE_COUNT_TIMED_OUT: self.metrics.send_message_count_timed_out.get_count(),
            Metrics.RECEIVE_C2D_COUNT_RECEIVED: self.metrics.receive_c2d_count_received.get_count(),
            Metrics.RECEIVE_C2D_COUNT_TIMED_OUT: self.metrics.receive_c2d_count_timed_out.get_count(),
            Metrics.REPORTED_PROPERTIES_COUNT_ADDED: self.metrics.reported_properties_count_added.get_count(),
            Metrics.REPORTED_PROPERTIES_COUNT_REMOVED: self.metrics.reported_properties_count_removed.get_count(),
            Metrics.REPORTED_PROPERTIES_COUNT_TIMED_OUT: self.metrics.reported_properties_count_timed_out.get_count(),
            Metrics.GET_TWIN_COUNT_SUCCEEDED: self.metrics.get_twin_count_succeeded.get_count(),
            Metrics.GET_TWIN_COUNT_TIMED_OUT: self.metrics.get_twin_count_timed_out.get_count(),
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_RECEIVED: self.metrics.desired_property_patch_count_received.get_count(),
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_TIMED_OUT: self.metrics.desired_property_patch_count_timed_out.get_count(),
            Metrics.METHOD_INVOKE_COUNT_REQUEST_RECEIVED: self.metrics.method_invoke_count_request_received.get_count(),
            Metrics.METHOD_INVOKE_COUNT_REQUEST_TIMED_OUT: self.metrics.method_invoke_count_request_timed_out.get_count(),
        }
        return props

    def check_failure_counts(self):
        """
        Check all failure counts and raise an exception if they exceeded the allowed count
        """
        if self.metrics.exception_count.get_count() > self.config[Settings.ALLOWED_EXCEPTION_COUNT]:
            raise Exception(
                "Exception count ({}) too high.".format(self.metrics.exception_count.get_count())
            )

        timeout_count = (
            self.metrics.send_message_count_timed_out.get_count()
            + self.metrics.receive_c2d_count_timed_out.get_count()
            + self.metrics.reported_properties_count_timed_out.get_count()
            + self.metrics.get_twin_count_timed_out.get_count()
            + self.metrics.desired_property_patch_count_timed_out.get_count()
            + self.metrics.method_invoke_count_request_timed_out.get_count()
        )
        if timeout_count > self.config[Settings.OPERATION_TIMEOUT_ALLOWED_FAILURE_COUNT]:
            raise Exception("Timeout count ({}) too high.".format(timeout_count))

    def update_initial_reported_properties(self):
        """
        Update reported properties at the start of a run
        """

        # First remove all old props (in case of schema change).  Also clears out old results.
        props = {}
        for key in self.client.get_twin()["reported"]:
            if not key.startswith("$"):
                props[key] = None
        self.client.patch_twin_reported_properties(props)

        props = {
            Fields.SYSTEM_PROPERTIES: self.get_system_properties(),
            Fields.SESSION_METRICS: self.get_session_metrics(),
            Fields.TEST_METRICS: self.get_test_metrics(),
            Fields.CONFIG: self.config,
        }
        self.client.patch_twin_reported_properties(props)

    def get_system_properties(self):
        return {
            SystemProperties.LANGUAGE: "python",
            SystemProperties.LANGUAGE_VERSION: platform.python_version(),
            SystemProperties.SDK_VERSION: azure.iot.device.constant.VERSION,
            SystemProperties.SDK_GITHUB_REPO: os.getenv("SDK_GIT_REPO"),
            SystemProperties.SDK_GITHUB_BRANCH: os.getenv("SDK_GIT_BRANCH"),
            SystemProperties.SDK_GITHUB_COMMIT: os.getenv("SDK_GIT_COMMIT"),
            SystemProperties.OS_TYPE: platform.system(),
            SystemProperties.OS_RELEASE: _get_os_release_based_on_user_agent_standard(),
        }

    def get_system_health_telemetry(self):
        props = {
            Metrics.PROCESS_CPU_PERCENT: self.system_health_telemetry.process_cpu_percent,
            Metrics.PROCESS_WORKING_SET_BYTES: self.system_health_telemetry.process_working_set_bytes,
            Metrics.PROCESS_WORKING_SET_PRIVATE_BYTES: self.system_health_telemetry.process_working_set_private_bytes,
            Metrics.PROCESS_GARBAGE_COLLECTION_OBJECTS: len(gc.get_objects()),
            Metrics.PROCESS_SDK_THREADS: threading.active_count()
            - len(self.outgoing_executor.all_threads)
            - len(self.incoming_executor.all_threads),
            Metrics.TEST_APP_ACTIVE_RECEIVE_THREADS: len(
                list(self.incoming_executor.active_threads)
            ),
            Metrics.TEST_APP_ACTIVE_SEND_THREADS: len(list(self.outgoing_executor.active_threads)),
        }
        return props

    def create_message_from_dict(self, payload):
        """
        helper function to create a message from a dict object
        """

        # Note: we're changing the dictionary that the user passed in.
        # This isn't the best idea, but it works and it saves us from deep copies
        if self.service_instance_id:
            payload[Fields.SERVICE_INSTANCE_ID] = self.service_instance_id
        payload[Fields.RUN_ID] = run_id

        # This function only creates the message.  The caller needs to queue it up for sending.
        msg = Message(json.dumps(payload))
        msg.content_type = Const.JSON_CONTENT_TYPE
        msg.content_encoding = Const.JSON_CONTENT_ENCODING

        return msg

    def pair_with_service(self):
        """
        "pair" with a service app. This is necessary because we can have a single
        service app responsible for multiple device apps and we want to make sure
        the service app is actually running before we continue.
        """

        pairing_start_time = time.time()

        while (time.time() - pairing_start_time) < self.config[
            Settings.PAIRING_REQUEST_TIMEOUT_INTERVAL_IN_SECONDS
        ]:
            operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()
            pairing_payload = {
                Fields.CMD: Commands.PAIR_WITH_SERVICE_APP,
                Fields.OPERATION_ID: operation_ticket.id,
                Fields.REQUESTED_SERVICE_POOL: requested_service_pool,
            }
            msg = self.create_message_from_dict(pairing_payload)
            self.client.send_message(msg)
            logger.info("Sending pairing request: {}".format(pprint.pformat(pairing_payload)))
            event_logger.info(Events.SENDING_PAIRING_REQUEST)

            operation_ticket.event.wait(
                timeout=self.config[Settings.PAIRING_REQUEST_SEND_INTERVAL_IN_SECONDS]
            )
            if operation_ticket.event.isSet():
                body = json.loads(operation_ticket.result_message.data.decode())

                logger.info("Received pairing response: {}".format(pprint.pformat(body)))
                event_logger.info(Events.RECEIVED_PAIRING_RESPONSE)

                self.service_instance_id = body[Fields.SERVICE_INSTANCE_ID]

                event_logger.info(Events.PAIRING_COMPLETE)
                return

        raise Exception("Pairing timed out")

    def send_metrics_to_azure_monitor(self, props):
        """
        Send metrics to azure monitor, based on the reported properties that we probably just
        sent to the hub
        """
        # We don't record session_metrics to Azure Monitor because the session metrics are
        # recording things like "start time" and "elapsed time" which are already available
        # in Azure Monitor in other forms.
        with self.reporter_lock:
            self.reporter.set_metrics_from_dict(props[Fields.SYSTEM_HEALTH_METRICS])
            self.reporter.set_metrics_from_dict(props[Fields.TEST_METRICS])
            self.reporter.record()

    def send_message_with_metrics(self, message, operation_id, message_type="send_message"):
        """
        Wrapper around send_message which sends a message to iothub and keeps track of all
        necessary metric.
        """
        queue_time = time.time()
        self.metrics.send_message_count_queued.increment()

        self.client.send_message(message)
        self.metrics.send_message_count_sent.increment()

        logger.info("{} op {} sent to service".format(message_type, operation_id))

        send_time = time.time()

        with self.reporter_lock:
            self.reporter.set_metrics_from_dict(
                {Metrics.LATENCY_QUEUE_MESSAGE_TO_SEND: (send_time - queue_time) * 1000}
            )
            self.reporter.record()

    def test_send_message_thread(self):
        """
        Thread to continuously send d2c messages throughout the longhaul run.  This thread doesn't
        actually send messages because send_message is blocking and we want to overlap our send
        operations.  Instead, this thread creates new transient threads that actually send the
        message.
        """

        def send_message():
            """
            Helper function to create and send send a single message to iothub.  Intended
            to run in it's own thread.  Returns after the service acknowledges receipt of
            the message.
            """
            operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()

            payload = {
                Fields.CMD: Commands.SEND_OPERATION_RESPONSE,
                Fields.OPERATION_ID: operation_ticket.id,
                Fields.SESSION_METRICS: self.get_session_metrics(),
                Fields.TEST_METRICS: self.get_test_metrics(),
                Fields.SYSTEM_HEALTH_METRICS: self.get_system_health_telemetry(),
            }

            msg = self.create_message_from_dict(payload)
            self.send_message_with_metrics(msg, operation_ticket.id)

            if operation_ticket.event.wait(
                timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]
            ):
                logger.info("Telemetry op {} arrived at service".format(operation_ticket.id))
                self.metrics.send_message_count_verified.increment()

            else:
                logger.warning(
                    "Telemetry op {} did not arrive at service".format(operation_ticket.id)
                )
                self.metrics.send_message_count_timed_out.increment()

        while not self.done.isSet():
            reset_watchdog()
            self.outgoing_executor.submit(send_message)
            # sleep until we need to send again
            self.done.wait(1 / self.config[Settings.SEND_MESSAGE_OPERATIONS_PER_SECOND])

    def sleep_with_watchdog(self, sleep_time):
        """
        Execute a `time.sleep` operation which will not get interrupted by a watchdog timeout.
        """
        sleep_end = time.time() + sleep_time
        while not self.done.isSet() and time.time() <= sleep_end:
            reset_watchdog()
            time_left = max(0, sleep_end - time.time())
            self.done.wait(
                timeout=min(time_left, self.outgoing_executor.watchdog_interval - 10)
            )  # -10 for headroom
        reset_watchdog()

    def test_flood_thread(self):
        """
        A thread which runs occasional message floods on some regular interval.  A message
        flood is a large number of send_message calls in a short period of time.
        """

        def send_flood_message(operation_id):
            """
            Helper function to create and send send a single message to iothub.  Intended
            to run in it's own thread.  Returns when the PUBACK returns.
            """
            payload = {
                Fields.CMD: Commands.SEND_OPERATION_RESPONSE,
                Fields.OPERATION_ID: operation_id,
                "extraPayload": self.make_random_payload(),
            }

            msg = self.create_message_from_dict(payload)
            self.send_message_with_metrics(msg, operation_id, "Flood")

        def do_flood():
            """
            Function to run send a flood of messages to iothub.  Intended to run in its own
            thread.  Waits for the server to acknowledge all messages before returning.
            """

            # Make RunningOperation objects for all of our messages.
            all_ops = [
                self.operation_ticket_list.make_event_based_operation_ticket()
                for x in range(
                    random.randint(1, self.config[Settings.SEND_MESSAGE_FLOOD_MAX_MESSAGE_COUNT])
                )
            ]
            logger.info("flooding with {} send_message operations".format(len(all_ops)))

            # Queue up sends for all these messages.  The number of sends which can be outstanding
            # on the wire is limited by the number of threads in the outgoing_executor thread pool.
            for op in all_ops:
                self.outgoing_executor.submit(send_flood_message, op.id)

            # Wait for all of the sends to complete.  We do this by waiting for some arbitrary
            # operation to complete, then we prune the list of remaining operations and wait
            # for another arbitrary operation until the list of remaining operations is empty.
            start_time = time.time()
            timeout_time = time.time() + self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]
            while len(all_ops) and time.time() < timeout_time:
                time_left = timeout_time - time.time()
                all_ops[0].event.wait(timeout=time_left)
                remaining_ops = [x for x in all_ops if not x.event.isSet()]
                count_remaining = len(remaining_ops)
                count_verified = len(all_ops) - len(remaining_ops)
                logger.info(
                    "Flood check - after {} seconds, {} operations verified, {} remaining".format(
                        time.time() - start_time, count_verified, count_remaining,
                    )
                )
                self.metrics.send_message_count_verified.add(count_verified)
                all_ops = remaining_ops

            if len(all_ops):
                logger.warning("Flood complete.  {} opperations timed out".format(len(all_ops)))
                self.metrics.send_message_count_timed_out.add(len(all_ops))
            else:
                logger.info("Flood complete.  All operations verified")

        # Finally the main loop of the flood thread.  This just sleeps for a while, then queues
        # up the flood in a different thread.  This means that floods can theoretically overlap.
        while not self.done.isSet():
            self.sleep_with_watchdog(self.config[Settings.SEND_MESSAGE_FLOOD_INTERVAL_IN_SECONDS])
            if self.done.isSet():
                return

            self.outgoing_executor.submit(do_flood, long_run=True)

    def handle_desired_property_patch_received(self, patch):
        """
        callback for desired property patch reception
        """
        # props that have the pairing structure go to `incoming_pairing_message_queue`
        if patch.get(Fields.PAIRING, {}):
            self.incoming_pairing_message_queue.put(patch)

        # other props go into incoming_deisred_property_patch_queue
        else:
            self.incoming_desired_property_patch_queue.put(patch)

    def handle_message_received(self, msg):
        """
        Callback for receiving c2d messages.
        """
        body = json.loads(msg.data.decode())

        if body and body[Fields.RUN_ID] == run_id:
            if body[Fields.CMD] in [
                Commands.PAIR_RESPONSE,
                Commands.OPERATION_RESPONSE,
                Commands.METHOD_RESPONSE,
                Commands.C2D_RESPONSE,
            ]:
                self.incoming_executor.submit(self.handle_operation_response, msg)

            else:
                logger.warning("Unknown command received: {}".format(body))

        else:
            logger.warning("C2D received, but it's not for us: {}".format(body))

    def handle_operation_response(self, msg):
        """
        Code to handle incoming operationResponse messages.  This is where we go through the list
        of `operationId` values that we received and call the appropriate callbacks to indicate that
        the service has responded.
        """

        body = json.loads(msg.data.decode())

        operation_ids = body.get(Fields.OPERATION_IDS, [])
        if not operation_ids:
            operation_ids = [
                body.get(Fields.OPERATION_ID),
            ]

        logger.info("Received {} message with {}".format(body[Fields.CMD], operation_ids))

        for operation_id in operation_ids:
            operation_ticket = self.operation_ticket_list.get(operation_id)
            if operation_ticket:
                operation_ticket.result_message = msg
                operation_ticket.complete()
            else:
                logger.warning("Received unknown operationId: {}:".format(operation_id))

    def operation_test_thread(self):
        """
        Thread which loops through various hub features and tests them one at a time.
        """

        tests = [
            (self.test_reported_properties, (), {}),
            (self.test_desired_properties, (), {}),
            (self.test_method_invoke, [TestMethod(MethodNames.FAIL_WITH_404, 404, False)], {}),
            (self.test_method_invoke, [TestMethod(MethodNames.ECHO_REQUEST, 200, True)], {}),
            (
                self.test_method_invoke,
                [TestMethod(MethodNames.UNDEFINED_METHOD_NAME, 500, False)],
                {},
            ),
            (self.test_c2d, (), {}),
        ]

        while not self.done.isSet():
            for (function, args, kwargs) in tests:
                reset_watchdog()
                self.outgoing_executor.submit(function, *args, **kwargs)

                time.sleep(self.config[Settings.INTER_TEST_DELAY_INTERVAL_IN_SECONDS])
                if self.done.isSet():
                    break

    def test_reported_properties(self):
        """
        test function to send a single reported property and then clear it after the
        server verifies it. It does this by setting properties inside
        `properties/reported/testContent/reportedPropertyTest`.  Each property has
        a `addOperationId` value and a `removeOperationId` value.  When  the service sees the
        property added, it sends the `addOperationId` to the device.  When the service sees the
        property removed, it sends the `removeOperationId` to the device. This way the device
        can add a property, verify that it was added, then remove it and verify that it was removed.
        """

        # Get metrics.  We report them to Azure monitor and also push them as part of the twin
        metrics = {
            Fields.TEST_METRICS: self.get_test_metrics(),
            Fields.SYSTEM_HEALTH_METRICS: self.get_system_health_telemetry(),
        }
        self.send_metrics_to_azure_monitor(metrics)

        # session metrics goes into reported propertes, but not azure monitor
        # They're either never-changing (like start time) or redundant (like elapsed time)
        metrics[Fields.SESSION_METRICS] = self.get_session_metrics()

        def make_reported_prop(property_name, val):
            return {Fields.TEST_CONTENT: {Fields.REPORTED_PROPERTY_TEST: {property_name: val}}}

        add_operation = self.operation_ticket_list.make_event_based_operation_ticket()
        remove_operation = self.operation_ticket_list.make_event_based_operation_ticket()

        prop_name = "prop_{}".format(self.reported_property_index)
        self.reported_property_index += 1

        prop_value = {
            Fields.ADD_OPERATION_ID: add_operation.id,
            Fields.REMOVE_OPERATION_ID: remove_operation.id,
        }

        logger.info("Adding test property {}".format(prop_name))
        props = make_reported_prop(prop_name, prop_value)
        props.update(metrics)
        self.client.patch_twin_reported_properties(props)

        if add_operation.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info("Add of reported property {} verified by service".format(prop_name))
            self.metrics.reported_properties_count_added.increment()
        else:
            logger.info("Add of reported property {} verification timeout".format(prop_name))
            self.metrics.reported_properties_count_timed_out.increment()

        logger.info("Removing test property {}".format(prop_name))
        self.client.patch_twin_reported_properties(make_reported_prop(prop_name, None))

        if remove_operation.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info("Remove of reported property {} verified by service".format(prop_name))
            self.metrics.reported_properties_count_removed.increment()
        else:
            logger.info("Remove of reported property {} verification timeout".format(prop_name))
            self.metrics.reported_properties_count_timed_out.increment()

    @return_if_already_running
    def test_desired_properties(self):
        """
        Test function to set a desired property, wait for the patch to arrive at the client,
        then get the twin to verify that it shows up in the twin.
        """
        twin_guid = str(uuid.uuid4())

        payload_set_desired_props = {
            Fields.CMD: Commands.SET_DESIRED_PROPS,
            Fields.DESIRED_PROPERTIES: {Fields.TEST_CONTENT: {Fields.TWIN_GUID: twin_guid}},
        }
        msg = self.create_message_from_dict(payload_set_desired_props)
        logger.info("Sending message to update desired getTwin property to {}".format(twin_guid))
        self.client.send_message(msg)

        start_time = time.time()
        end_time = start_time + self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]

        actual_twin_guid = None
        while time.time() <= end_time:
            try:
                patch = self.incoming_desired_property_patch_queue.get(
                    max(0, end_time - time.time())
                )
            except queue.Empty:
                break
            else:
                test_content = patch.get(Fields.TEST_CONTENT, {})
                actual_twin_guid = test_content.get(Fields.TWIN_GUID)
                if actual_twin_guid == twin_guid:
                    logger.info("received expected patch: {} succeeded".format(twin_guid))
                    self.metrics.desired_property_patch_count_received.increment()
                    break
                else:
                    logger.info(
                        "Did not receive expected patch (yet). actual = {}.  expected = {}".format(
                            actual_twin_guid, twin_guid
                        )
                    )

        if actual_twin_guid != twin_guid:
            logger.warning(
                "wait_for_desired_property_patch did not retrieve expected properties within expected time.  expected={}".format(
                    twin_guid
                )
            )
            self.metrics.desired_property_patch_count_timed_out.increment()

        # Now test the twin.  Since we already received the patch, our property should
        # be in the twin.
        twin = self.client.get_twin()
        logger.info("Got twin: {}".format(twin))

        desired = twin.get(Fields.DESIRED, {})
        test_content = desired.get(Fields.TEST_CONTENT, {})
        actual_twin_guid = test_content.get(Fields.TWIN_GUID)

        if actual_twin_guid == twin_guid:
            logger.info("Got desired_twin: {} succeeded".format(twin_guid))
            self.metrics.get_twin_count_succeeded.increment()
        else:
            logger.info(
                "get_twin property did not match: actual = {} expected = {}".format(
                    actual_twin_guid, twin_guid
                )
            )
            self.metrics.get_twin_count_timed_out_out.increment()

    def handle_method_received(self, method_request):
        def handle(method_request):
            logger.info("Received method request {}".format(method_request.name))
            self.metrics.method_invoke_count_request_received.increment()
            if method_request.name == MethodNames.FAIL_WITH_404:
                logger.info("sending response to fail with 404")
                self.client.send_method_response(
                    MethodResponse.create_from_method_request(method_request, 404)
                )
            elif method_request.name == MethodNames.ECHO_REQUEST:
                self.client.send_method_response(
                    MethodResponse.create_from_method_request(
                        method_request, 200, method_request.payload,
                    )
                )
            else:
                logger.info("sending response to fail with 500")
                self.client.send_method_response(
                    MethodResponse.create_from_method_request(method_request, 500)
                )

        self.incoming_executor.submit(handle, method_request)

    def make_random_payload(self):
        return {
            "random_guid": str(uuid.uuid4()),
            "sub_object": {
                "string_value": str(uuid.uuid4()),
                "bool_value": random.random() > 0.5,
                "int_value": random.randint(-65535, 65535),
            },
        }

    def test_method_invoke(self, method):
        operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()

        if method.include_payload:
            payload = self.make_random_payload()
        else:
            payload = None

        command_payload = {
            Fields.CMD: Commands.INVOKE_METHOD,
            Fields.OPERATION_ID: operation_ticket.id,
            Fields.METHOD_NAME: method.method_name,
            Fields.METHOD_INVOKE_PAYLOAD: payload,
        }
        msg = self.create_message_from_dict(command_payload)
        logger.info("sending method invoke with guid={}".format(operation_ticket.id))
        self.client.send_message(msg)

        if operation_ticket.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info(
                "method with guid {} returned {}".format(
                    operation_ticket.id, operation_ticket.result_message
                )
            )
            body = json.loads(operation_ticket.result_message.data.decode())
            fail = False
            if body[Fields.METHOD_RESPONSE_STATUS_CODE] != method.expected_status_code:
                logger.error(
                    "Unexpected method status: id={}, received {} expected {}".format(
                        operation_ticket.id,
                        body[Fields.METHOD_RESPONSE_STATUS_CODE],
                        method.expected_status_code,
                    )
                )
                fail = True
            if body[Fields.METHOD_RESPONSE_PAYLOAD] != payload:
                logger.error(
                    "Unexpected payload: id={}, received {} expected {}".format(
                        operation_ticket.id, body[Fields.METHOD_RESPONSE_PAYLOAD], payload
                    )
                )
                fail = True

            if fail:
                raise ThiefFatalException("Content mismatch on method invoke")
            else:
                logger.info("Method call check succeeded")

        else:
            logger.error("method with guid {} never completed".format(operation_ticket.id))
            self.metrics.method_invoke_count_request_timed_out = ThreadSafeCounter()

    def test_c2d(self):

        operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()

        sent_test_payload = self.make_random_payload()

        command_payload = {
            Fields.CMD: Commands.SEND_C2D,
            Fields.OPERATION_ID: operation_ticket.id,
            Fields.TEST_C2D_PAYLOAD: sent_test_payload,
        }

        logger.info("Requesting C2D for operationId {}".format(operation_ticket.id))
        self.client.send_message(self.create_message_from_dict(command_payload))

        if operation_ticket.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info("C2D received for operationId {}".format(operation_ticket.id))
            self.metrics.receive_c2d_count_received.increment()

            body = json.loads(operation_ticket.result_message.data.decode())
            if body[Fields.TEST_C2D_PAYLOAD] != sent_test_payload:
                logger.warning(
                    "C2D payload for operationId {}  does not match.  Expected={}, Received={}".format(
                        operation_ticket.id, sent_test_payload, body[Fields.TEST_C2D_PAYLOAD]
                    )
                )
                raise ThiefFatalException("C2D payload mismatch")
        else:
            logger.info("C2D timed out for operationId {}".format(operation_ticket.id))
            self.metrics.receive_c2d_count_timed_out.increment()

    def wrap_handler(self, func):
        """
        Function to take a handler function and wrap it inside a try/catch block
        which increments exception count.  This is done because iothub callbacks
        will catch and swallow any exceptions that get raised in handlers.
        """

        def newfunc(*args, **kwargs):
            try:
                func(*args, **kwargs)
            except Exception as e:
                logger.error("exception in handler: {}".format(str(e) or type(e)), exc_info=True)
                self.metrics.exception_count.increment()

        return newfunc

    def main(self):

        self.metrics.run_start_utc = datetime.datetime.now(datetime.timezone.utc)
        self.metrics.run_state = RunStates.RUNNING
        logger.info("Starting at {}".format(self.metrics.run_start_utc))

        # Create our client and push initial properties
        self.client, self.hub, self.device_id = dps.create_device_client_using_dps_group_key(
            provisioning_host=provisioning_host,
            registration_id=registration_id,
            id_scope=id_scope,
            group_symmetric_key=group_symmetric_key,
            keep_alive=self.config[Settings.MQTT_KEEP_ALIVE_INTERVAL],
            sastoken_ttl=self.config[Settings.SAS_TOKEN_RENEWAL_INTERVAL],
        )
        hub_host_name = self.hub[: self.hub.find(".")]  # keep everything before the first dot
        azure_monitor.add_logging_properties(hub=hub_host_name, device_id=self.device_id)
        self.update_initial_reported_properties()

        event_logger.info(
            Events.STARTING_RUN, extra=custom_props({CustomDimensions.RUN_REASON: run_reason})
        )

        # Set handlers for incoming hub messages
        self.client.on_message_received = self.wrap_handler(self.handle_message_received)
        self.client.on_method_request_received = self.wrap_handler(self.handle_method_received)
        self.client.on_twin_desired_properties_patch_received = self.wrap_handler(
            self.handle_desired_property_patch_received
        )

        # Pair
        self.pair_with_service()

        # Start testing threads
        self.outgoing_executor.submit(self.test_send_message_thread, critical=True)
        self.outgoing_executor.submit(self.operation_test_thread, critical=True)
        if self.config[Settings.DO_SEND_MESSAGE_FLOODS]:
            self.outgoing_executor.submit(self.test_flood_thread, critical=True)

        self.metrics.exit_reason = ""
        start_time = time.time()
        planned_end_time = start_time + self.config[Settings.MAX_RUN_DURATION_IN_SECONDS]
        try:
            while time.time() < planned_end_time:
                # TODO: limit sleep length to check failure counts more often
                self.outgoing_executor.wait_for_thread_death_event(planned_end_time - time.time())
                self.outgoing_executor.check_watchdogs()
                self.incoming_executor.check_watchdogs()
                self.check_failure_counts()

                non_fatal_errors = 0
                fatal_error = None

                def count_failure(e):
                    nonlocal non_fatal_errors, fatal_error
                    if isinstance(e, ThiefFatalException):
                        logger.info("FATAL ERROR: {}".format(str(e) or type(e)))
                        fatal_error = e
                    else:
                        non_fatal_errors += 1

                self.outgoing_executor.check_for_failures(count_failure)
                self.incoming_executor.check_for_failures(count_failure)
                if non_fatal_errors:
                    logger.info("Adding {} failures to count".format(non_fatal_errors))
                    self.metrics.exception_count.add(non_fatal_errors)
                if fatal_error:
                    raise (fatal_error)
            self.metrics.run_state = RunStates.COMPLETE
        except KeyboardInterrupt:
            self.metrics.run_state = RunStates.INTERRUPTED
            self.metrics.exit_reason = "KeyboardInterrupt"
            raise
        except BaseException as e:
            self.metrics.run_state = RunStates.FAILED
            self.metrics.exit_reason = str(e) or type(e)
            logger.error("Run failed: {}".format(self.metrics.exit_reason), exc_info=True)
            raise
        finally:
            logger.info("-------------------------------------------------------------")
            logger.info("Setting done flag: {}".format(self.metrics.exit_reason))
            logger.info("-------------------------------------------------------------")
            self.done.set()
            active_threads = 0
            for executor in [self.outgoing_executor, self.incoming_executor]:
                done, not_done = executor.wait(
                    timeout=2 * self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]
                )
                active_threads += len(not_done)
            if active_threads:
                logger.error("{} threads not stopped after ending run".format(active_threads))
                dump_active_stacks(logger.info)

            logger.info("Exiting main at {}".format(datetime.datetime.utcnow()))
            event_logger.info(
                Events.ENDING_RUN,
                extra=custom_props({CustomDimensions.EXIT_REASON: self.metrics.exit_reason}),
            )

            # Update one last time.  This is required because the service app relies
            # on RUN_STATE to know when we're dead
            props = {
                Fields.SESSION_METRICS: self.get_session_metrics(),
                Fields.TEST_METRICS: self.get_test_metrics(),
                Fields.SYSTEM_HEALTH_METRICS: self.get_system_health_telemetry(),
            }
            logger.info("Results: {}".format(pprint.pformat(props)))
            self.client.patch_twin_reported_properties(props)

            logger.info("Disconnecting")
            self.client.disconnect()
            logger.info("Done disconnecting")


if __name__ == "__main__":
    try:
        if len(sys.argv) > 1:
            run_reason = " ".join(sys.argv[1:])
        DeviceApp().main()
    except BaseException as e:
        logger.critical("App shutdown exception: {}".format(str(e) or type(e)), exc_info=True)
        raise
    finally:
        # Flush azure monitor telemetry
        logging.shutdown()

# Copyright (c) Microsoft. All rights reserved.
# Licensed under the MIT license. See LICENSE file in the project root for
# full license information.
import logging
import os
import time
import uuid
import json
import datetime
import threading
import pprint
import sys
import collections
import platform
import random
from executor import BetterThreadPoolExecutor, reset_watchdog, dump_active_stacks
import dps
import queue
import faulthandler
import gc
from azure.iot.device import Message, MethodResponse
import azure.iot.device.constant
from measurement import ThreadSafeCounter
from system_health_telemetry import SystemHealthTelemetry
import azure_monitor
from azure_monitor_metrics import MetricsReporter
from thief_constants import (
    Const,
    Fields,
    Commands,
    Events,
    Metrics,
    Settings,
    CustomDimensions,
    MethodNames,
    RunStates,
    SystemProperties,
)
from operation_tickets import OperationTicketList
import thief_secrets

faulthandler.enable()

provisioning_host = thief_secrets.DEVICE_PROVISIONING_HOST
id_scope = thief_secrets.DEVICE_ID_SCOPE
group_symmetric_key = thief_secrets.DEVICE_GROUP_SYMMETRIC_KEY
registration_id = thief_secrets.DEVICE_ID
requested_service_pool = thief_secrets.REQUESTED_SERVICE_POOL

# run_reason can be an environment variable or is can be passed on the command line.
run_reason = os.getenv("RUN_REASON")
# run_id can be an environment variable or it can be automatically generated
run_id = str(uuid.uuid4())

# set default logging which will only go to the console

logging.basicConfig(level=logging.WARNING)
# logging.getLogger("paho").setLevel(level=logging.DEBUG)
logging.getLogger("thief").setLevel(level=logging.INFO)
# logging.getLogger("azure.iot").setLevel(level=logging.INFO)
logger = logging.getLogger("thief.{}".format(__name__))

# configure our traces and events to go to Azure Monitor
azure_monitor.add_logging_properties(
    client_type="device",
    run_id=run_id,
    sdk_version=azure.iot.device.constant.VERSION,
    transport="mqtt",
    pool_id=requested_service_pool,
)
event_logger = azure_monitor.get_event_logger()
azure_monitor.log_to_azure_monitor("thief")


TestMethod = collections.namedtuple(
    "TestMethod", "method_name expected_status_code include_payload"
)


class ThiefFatalException(Exception):
    pass


def _get_os_release_based_on_user_agent_standard():
    return "({python_runtime};{os_type} {os_release};{architecture})".format(
        python_runtime=platform.python_version(),
        os_type=platform.system(),
        os_release=platform.version(),
        architecture=platform.machine(),
    )


def custom_props(extra_props):
    """
    helper function for adding customDimensions to logger calls at execution time
    """
    return {"custom_dimensions": extra_props}


class DeviceRunMetrics(object):
    """
    Object we use internally to keep track of how a the entire test is performing.
    """

    def __init__(self):
        self.run_start_utc = None
        self.run_state = RunStates.WAITING
        self.exit_reason = None

        self.exception_count = ThreadSafeCounter()

        self.send_message_count_queued = ThreadSafeCounter()
        self.send_message_count_sent = ThreadSafeCounter()
        self.send_message_count_verified = ThreadSafeCounter()
        self.send_message_count_timed_out = ThreadSafeCounter()
        self.receive_c2d_count_received = ThreadSafeCounter()
        self.receive_c2d_count_timed_out = ThreadSafeCounter()

        self.reported_properties_count_added = ThreadSafeCounter()
        self.reported_properties_count_removed = ThreadSafeCounter()
        self.reported_properties_count_timed_out = ThreadSafeCounter()

        self.get_twin_count_succeeded = ThreadSafeCounter()
        self.get_twin_count_timed_out = ThreadSafeCounter()

        self.desired_property_patch_count_received = ThreadSafeCounter()
        self.desired_property_patch_count_timed_out = ThreadSafeCounter()

        self.method_invoke_count_request_received = ThreadSafeCounter()
        self.method_invoke_count_request_timed_out = ThreadSafeCounter()


"""
Object we use internally to keep track of how the entire test is configured.
Currently hardcoded. Later, this will come from desired properties.
"""
device_run_config = {
    Settings.MAX_RUN_DURATION_IN_SECONDS: 12 * 60 * 60,
    Settings.ALLOWED_EXCEPTION_COUNT: 10,
    Settings.INTER_TEST_DELAY_INTERVAL_IN_SECONDS: 1,
    Settings.OPERATION_TIMEOUT_IN_SECONDS: 300,
    Settings.OPERATION_TIMEOUT_ALLOWED_FAILURE_COUNT: 10,
    Settings.PAIRING_REQUEST_TIMEOUT_INTERVAL_IN_SECONDS: 900,
    Settings.PAIRING_REQUEST_SEND_INTERVAL_IN_SECONDS: 30,
    Settings.SEND_MESSAGE_OPERATIONS_PER_SECOND: 1,
    Settings.MQTT_KEEP_ALIVE_INTERVAL: 20,
    Settings.SAS_TOKEN_RENEWAL_INTERVAL: 600,
    Settings.MAX_TEST_SEND_THREADS: 128,
    Settings.MAX_TEST_RECEIVE_THREADS: 10,
    Settings.DO_SEND_MESSAGE_FLOODS: True,
    Settings.SEND_MESSAGE_FLOOD_INTERVAL_IN_SECONDS: 600,
    Settings.SEND_MESSAGE_FLOOD_MAX_MESSAGE_COUNT: 500,
}


def return_if_already_running(func):
    """
    Decorator function to make sure that the wrapped function cannot be called in parallel.
    The wrapped funciton can be invoked once.  If is called again before the first invocation
    is complete, it immediately returns without running.  Once the first invocation completes,
    then it an be invoked again.
    """
    already_running = threading.BoundedSemaphore()

    def inner(*args, **kwargs):
        if already_running.acquire(blocking=False):
            try:
                return func(*args, **kwargs)
            finally:
                already_running.release()

    return inner


class DeviceApp(object):
    """
    Main application object
    """

    def __init__(self):
        global device_run_config
        super(DeviceApp, self).__init__()

        self.config = device_run_config
        self.outgoing_executor = BetterThreadPoolExecutor(
            max_workers=self.config[Settings.MAX_TEST_SEND_THREADS]
        )
        self.incoming_executor = BetterThreadPoolExecutor(
            max_workers=self.config[Settings.MAX_TEST_RECEIVE_THREADS]
        )
        self.done = threading.Event()
        self.client = None
        self.hub = None
        self.device_id = None
        self.metrics = DeviceRunMetrics()
        self.service_instance_id = None
        self.system_health_telemetry = SystemHealthTelemetry()
        # for running operations
        self.operation_ticket_list = OperationTicketList()
        # for metrics
        self.reporter_lock = threading.Lock()
        self.reporter = MetricsReporter()
        self._configure_azure_monitor_metrics()
        # for pairing
        self.incoming_pairing_message_queue = queue.Queue()
        # for twin
        self.reported_property_index = 1
        self.incoming_desired_property_patch_queue = queue.Queue()

    def _configure_azure_monitor_metrics(self):
        # ---------------------
        # System Health metrics
        # ---------------------
        self.reporter.add_float_measurement(
            Metrics.PROCESS_CPU_PERCENT,
            "CPU use for the device app, as a percentage of all cores",
            "percentage",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_WORKING_SET_BYTES,
            "Working set for the device app, includes shared and private, read-only and writeable memory",
            "bytes",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_WORKING_SET_PRIVATE_BYTES,
            "Amount of private data used by the process",
            "bytes",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_GARBAGE_COLLECTION_OBJECTS,
            "Number of active objects being managed by the garbage collector",
            "objects",
        )
        self.reporter.add_integer_measurement(
            Metrics.PROCESS_SDK_THREADS,
            "number of threads being used by the current SDK",
            "threads",
        )
        self.reporter.add_integer_measurement(
            Metrics.TEST_APP_ACTIVE_RECEIVE_THREADS,
            "number of active threads in the test app receive threadpool",
            "threads",
        )
        self.reporter.add_integer_measurement(
            Metrics.TEST_APP_ACTIVE_SEND_THREADS,
            "number of active threads in the test app send threadpool",
            "threads",
        )

        # ----------------
        # test app metrics
        # ----------------
        self.reporter.add_integer_measurement(
            Metrics.EXCEPTION_COUNT,
            "Number of (non-fatal) exceptions raised by the client library or test code",
            "exception count",
        )

        # --------------------
        # SendMesssage metrics
        # --------------------
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_QUEUED,
            "Number of telemetry messages queued for sending",
            "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_SENT, "Number of telemetry messages sent", "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_VERIFIED,
            "Number of telemetry messages sent and verified by the service",
            "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.SEND_MESSAGE_COUNT_TIMED_OUT,
            "Number of telemetry messages that timed out with no response from the service",
            "messages",
        )

        # -------------------
        # Receive c2d metrics
        # -------------------
        self.reporter.add_integer_measurement(
            Metrics.RECEIVE_C2D_COUNT_RECEIVED, "Number of c2d messages received", "messages",
        )
        self.reporter.add_integer_measurement(
            Metrics.RECEIVE_C2D_COUNT_TIMED_OUT,
            "Number of c2d messages not received in time",
            "messages",
        )

        # -------------------------
        # Reported property metrics
        # -------------------------
        self.reporter.add_integer_measurement(
            Metrics.REPORTED_PROPERTIES_COUNT_ADDED,
            "Number of reported properties which have been added",
            "patches",
        )
        self.reporter.add_integer_measurement(
            Metrics.REPORTED_PROPERTIES_COUNT_REMOVED,
            "Number of reported properties which have been removed",
            "patches",
        )
        self.reporter.add_integer_measurement(
            Metrics.REPORTED_PROPERTIES_COUNT_TIMED_OUT,
            "Number of reported property add & remove operations that timed out",
            "patches",
        )

        # ----------------
        # Get-twin metrics
        # ----------------
        self.reporter.add_integer_measurement(
            Metrics.GET_TWIN_COUNT_SUCCEEDED,
            "Number of times get_twin successfully verified a property update",
            "calls",
        )
        self.reporter.add_integer_measurement(
            Metrics.GET_TWIN_COUNT_TIMED_OUT,
            "Number of times get_twin was unable to verify a property update",
            "calls",
        )

        # ------------------------------
        # desired property patch metrics
        # ------------------------------
        self.reporter.add_integer_measurement(
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_RECEIVED,
            "Count of desired property patches that were successfully received",
            "patches",
        )
        self.reporter.add_integer_measurement(
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_TIMED_OUT,
            "Count of desired property patches that were not received",
            "patches",
        )

        # ---------------------
        # method invoke metrics
        # ---------------------
        self.reporter.add_integer_measurement(
            Metrics.METHOD_INVOKE_COUNT_REQUEST_RECEIVED,
            "Number of method invoke requests received",
            "invokes",
        )
        self.reporter.add_integer_measurement(
            Metrics.METHOD_INVOKE_COUNT_REQUEST_TIMED_OUT,
            "Number of method invoke requests timed out",
            "invokes",
        )

        # ---------------
        # Latency metrics
        # ---------------
        self.reporter.add_float_measurement(
            Metrics.LATENCY_QUEUE_MESSAGE_TO_SEND,
            "Number of milliseconds between queueing a telemetry message and actually sending it",
            "milliseconds",
        )

    def get_session_metrics(self):
        """
        Return metrics which describe the session the tests are running in
        """
        now = datetime.datetime.now(datetime.timezone.utc)
        elapsed_time = now - self.metrics.run_start_utc

        props = {
            Fields.RUN_START_UTC: self.metrics.run_start_utc.isoformat(),
            Fields.LATEST_UPDATE_TIME_UTC: now.isoformat(),
            Fields.ELAPSED_TIME: str(elapsed_time),
            Fields.RUN_STATE: str(self.metrics.run_state),
            Fields.EXIT_REASON: self.metrics.exit_reason,
        }
        return props

    def get_test_metrics(self):
        """
        Return metrics which describe the progress of the  different features being tested
        """

        props = {
            Metrics.EXCEPTION_COUNT: self.metrics.exception_count.get_count(),
            Metrics.SEND_MESSAGE_COUNT_QUEUED: self.metrics.send_message_count_queued.get_count(),
            Metrics.SEND_MESSAGE_COUNT_SENT: self.metrics.send_message_count_sent.get_count(),
            Metrics.SEND_MESSAGE_COUNT_VERIFIED: self.metrics.send_message_count_verified.get_count(),
            Metrics.SEND_MESSAGE_COUNT_TIMED_OUT: self.metrics.send_message_count_timed_out.get_count(),
            Metrics.RECEIVE_C2D_COUNT_RECEIVED: self.metrics.receive_c2d_count_received.get_count(),
            Metrics.RECEIVE_C2D_COUNT_TIMED_OUT: self.metrics.receive_c2d_count_timed_out.get_count(),
            Metrics.REPORTED_PROPERTIES_COUNT_ADDED: self.metrics.reported_properties_count_added.get_count(),
            Metrics.REPORTED_PROPERTIES_COUNT_REMOVED: self.metrics.reported_properties_count_removed.get_count(),
            Metrics.REPORTED_PROPERTIES_COUNT_TIMED_OUT: self.metrics.reported_properties_count_timed_out.get_count(),
            Metrics.GET_TWIN_COUNT_SUCCEEDED: self.metrics.get_twin_count_succeeded.get_count(),
            Metrics.GET_TWIN_COUNT_TIMED_OUT: self.metrics.get_twin_count_timed_out.get_count(),
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_RECEIVED: self.metrics.desired_property_patch_count_received.get_count(),
            Metrics.DESIRED_PROPERTY_PATCH_COUNT_TIMED_OUT: self.metrics.desired_property_patch_count_timed_out.get_count(),
            Metrics.METHOD_INVOKE_COUNT_REQUEST_RECEIVED: self.metrics.method_invoke_count_request_received.get_count(),
            Metrics.METHOD_INVOKE_COUNT_REQUEST_TIMED_OUT: self.metrics.method_invoke_count_request_timed_out.get_count(),
        }
        return props

    def check_failure_counts(self):
        """
        Check all failure counts and raise an exception if they exceeded the allowed count
        """
        if self.metrics.exception_count.get_count() > self.config[Settings.ALLOWED_EXCEPTION_COUNT]:
            raise Exception(
                "Exception count ({}) too high.".format(self.metrics.exception_count.get_count())
            )

        timeout_count = (
            self.metrics.send_message_count_timed_out.get_count()
            + self.metrics.receive_c2d_count_timed_out.get_count()
            + self.metrics.reported_properties_count_timed_out.get_count()
            + self.metrics.get_twin_count_timed_out.get_count()
            + self.metrics.desired_property_patch_count_timed_out.get_count()
            + self.metrics.method_invoke_count_request_timed_out.get_count()
        )
        if timeout_count > self.config[Settings.OPERATION_TIMEOUT_ALLOWED_FAILURE_COUNT]:
            raise Exception("Timeout count ({}) too high.".format(timeout_count))

    def update_initial_reported_properties(self):
        """
        Update reported properties at the start of a run
        """

        # First remove all old props (in case of schema change).  Also clears out old results.
        props = {}
        for key in self.client.get_twin()["reported"]:
            if not key.startswith("$"):
                props[key] = None
        self.client.patch_twin_reported_properties(props)

        props = {
            Fields.SYSTEM_PROPERTIES: self.get_system_properties(),
            Fields.SESSION_METRICS: self.get_session_metrics(),
            Fields.TEST_METRICS: self.get_test_metrics(),
            Fields.CONFIG: self.config,
        }
        self.client.patch_twin_reported_properties(props)

    def get_system_properties(self):
        return {
            SystemProperties.LANGUAGE: "python",
            SystemProperties.LANGUAGE_VERSION: platform.python_version(),
            SystemProperties.SDK_VERSION: azure.iot.device.constant.VERSION,
            SystemProperties.SDK_GITHUB_REPO: os.getenv("SDK_GIT_REPO"),
            SystemProperties.SDK_GITHUB_BRANCH: os.getenv("SDK_GIT_BRANCH"),
            SystemProperties.SDK_GITHUB_COMMIT: os.getenv("SDK_GIT_COMMIT"),
            SystemProperties.OS_TYPE: platform.system(),
            SystemProperties.OS_RELEASE: _get_os_release_based_on_user_agent_standard(),
        }

    def get_system_health_telemetry(self):
        props = {
            Metrics.PROCESS_CPU_PERCENT: self.system_health_telemetry.process_cpu_percent,
            Metrics.PROCESS_WORKING_SET_BYTES: self.system_health_telemetry.process_working_set_bytes,
            Metrics.PROCESS_WORKING_SET_PRIVATE_BYTES: self.system_health_telemetry.process_working_set_private_bytes,
            Metrics.PROCESS_GARBAGE_COLLECTION_OBJECTS: len(gc.get_objects()),
            Metrics.PROCESS_SDK_THREADS: threading.active_count()
            - len(self.outgoing_executor.all_threads)
            - len(self.incoming_executor.all_threads),
            Metrics.TEST_APP_ACTIVE_RECEIVE_THREADS: len(
                list(self.incoming_executor.active_threads)
            ),
            Metrics.TEST_APP_ACTIVE_SEND_THREADS: len(list(self.outgoing_executor.active_threads)),
        }
        return props

    def create_message_from_dict(self, payload):
        """
        helper function to create a message from a dict object
        """

        # Note: we're changing the dictionary that the user passed in.
        # This isn't the best idea, but it works and it saves us from deep copies
        if self.service_instance_id:
            payload[Fields.SERVICE_INSTANCE_ID] = self.service_instance_id
        payload[Fields.RUN_ID] = run_id

        # This function only creates the message.  The caller needs to queue it up for sending.
        msg = Message(json.dumps(payload))
        msg.content_type = Const.JSON_CONTENT_TYPE
        msg.content_encoding = Const.JSON_CONTENT_ENCODING

        return msg

    def pair_with_service(self):
        """
        "pair" with a service app. This is necessary because we can have a single
        service app responsible for multiple device apps and we want to make sure
        the service app is actually running before we continue.
        """

        pairing_start_time = time.time()

        while (time.time() - pairing_start_time) < self.config[
            Settings.PAIRING_REQUEST_TIMEOUT_INTERVAL_IN_SECONDS
        ]:
            operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()
            pairing_payload = {
                Fields.CMD: Commands.PAIR_WITH_SERVICE_APP,
                Fields.OPERATION_ID: operation_ticket.id,
                Fields.REQUESTED_SERVICE_POOL: requested_service_pool,
            }
            msg = self.create_message_from_dict(pairing_payload)
            self.client.send_message(msg)
            logger.info("Sending pairing request: {}".format(pprint.pformat(pairing_payload)))
            event_logger.info(Events.SENDING_PAIRING_REQUEST)

            operation_ticket.event.wait(
                timeout=self.config[Settings.PAIRING_REQUEST_SEND_INTERVAL_IN_SECONDS]
            )
            if operation_ticket.event.isSet():
                body = json.loads(operation_ticket.result_message.data.decode())

                logger.info("Received pairing response: {}".format(pprint.pformat(body)))
                event_logger.info(Events.RECEIVED_PAIRING_RESPONSE)

                self.service_instance_id = body[Fields.SERVICE_INSTANCE_ID]

                event_logger.info(Events.PAIRING_COMPLETE)
                return

        raise Exception("Pairing timed out")

    def send_metrics_to_azure_monitor(self, props):
        """
        Send metrics to azure monitor, based on the reported properties that we probably just
        sent to the hub
        """
        # We don't record session_metrics to Azure Monitor because the session metrics are
        # recording things like "start time" and "elapsed time" which are already available
        # in Azure Monitor in other forms.
        with self.reporter_lock:
            self.reporter.set_metrics_from_dict(props[Fields.SYSTEM_HEALTH_METRICS])
            self.reporter.set_metrics_from_dict(props[Fields.TEST_METRICS])
            self.reporter.record()

    def send_message_with_metrics(self, message, operation_id, message_type="send_message"):
        """
        Wrapper around send_message which sends a message to iothub and keeps track of all
        necessary metric.
        """
        queue_time = time.time()
        self.metrics.send_message_count_queued.increment()

        self.client.send_message(message)
        self.metrics.send_message_count_sent.increment()

        logger.info("{} op {} sent to service".format(message_type, operation_id))

        send_time = time.time()

        with self.reporter_lock:
            self.reporter.set_metrics_from_dict(
                {Metrics.LATENCY_QUEUE_MESSAGE_TO_SEND: (send_time - queue_time) * 1000}
            )
            self.reporter.record()

    def test_send_message_thread(self):
        """
        Thread to continuously send d2c messages throughout the longhaul run.  This thread doesn't
        actually send messages because send_message is blocking and we want to overlap our send
        operations.  Instead, this thread creates new transient threads that actually send the
        message.
        """

        def send_message():
            """
            Helper function to create and send send a single message to iothub.  Intended
            to run in it's own thread.  Returns after the service acknowledges receipt of
            the message.
            """
            operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()

            payload = {
                Fields.CMD: Commands.SEND_OPERATION_RESPONSE,
                Fields.OPERATION_ID: operation_ticket.id,
                Fields.SESSION_METRICS: self.get_session_metrics(),
                Fields.TEST_METRICS: self.get_test_metrics(),
                Fields.SYSTEM_HEALTH_METRICS: self.get_system_health_telemetry(),
            }

            msg = self.create_message_from_dict(payload)
            self.send_message_with_metrics(msg, operation_ticket.id)

            if operation_ticket.event.wait(
                timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]
            ):
                logger.info("Telemetry op {} arrived at service".format(operation_ticket.id))
                self.metrics.send_message_count_verified.increment()

            else:
                logger.warning(
                    "Telemetry op {} did not arrive at service".format(operation_ticket.id)
                )
                self.metrics.send_message_count_timed_out.increment()

        while not self.done.isSet():
            reset_watchdog()
            self.outgoing_executor.submit(send_message)
            # sleep until we need to send again
            self.done.wait(1 / self.config[Settings.SEND_MESSAGE_OPERATIONS_PER_SECOND])

    def sleep_with_watchdog(self, sleep_time):
        """
        Execute a `time.sleep` operation which will not get interrupted by a watchdog timeout.
        """
        sleep_end = time.time() + sleep_time
        while not self.done.isSet() and time.time() <= sleep_end:
            reset_watchdog()
            time_left = max(0, sleep_end - time.time())
            self.done.wait(
                timeout=min(time_left, self.outgoing_executor.watchdog_interval - 10)
            )  # -10 for headroom
        reset_watchdog()

    def test_flood_thread(self):
        """
        A thread which runs occasional message floods on some regular interval.  A message
        flood is a large number of send_message calls in a short period of time.
        """

        def send_flood_message(operation_id):
            """
            Helper function to create and send send a single message to iothub.  Intended
            to run in it's own thread.  Returns when the PUBACK returns.
            """
            payload = {
                Fields.CMD: Commands.SEND_OPERATION_RESPONSE,
                Fields.OPERATION_ID: operation_id,
                "extraPayload": self.make_random_payload(),
            }

            msg = self.create_message_from_dict(payload)
            self.send_message_with_metrics(msg, operation_id, "Flood")

        def do_flood():
            """
            Function to run send a flood of messages to iothub.  Intended to run in its own
            thread.  Waits for the server to acknowledge all messages before returning.
            """

            # Make RunningOperation objects for all of our messages.
            all_ops = [
                self.operation_ticket_list.make_event_based_operation_ticket()
                for x in range(
                    random.randint(1, self.config[Settings.SEND_MESSAGE_FLOOD_MAX_MESSAGE_COUNT])
                )
            ]
            logger.info("flooding with {} send_message operations".format(len(all_ops)))

            # Queue up sends for all these messages.  The number of sends which can be outstanding
            # on the wire is limited by the number of threads in the outgoing_executor thread pool.
            for op in all_ops:
                self.outgoing_executor.submit(send_flood_message, op.id)

            # Wait for all of the sends to complete.  We do this by waiting for some arbitrary
            # operation to complete, then we prune the list of remaining operations and wait
            # for another arbitrary operation until the list of remaining operations is empty.
            start_time = time.time()
            timeout_time = time.time() + self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]
            while len(all_ops) and time.time() < timeout_time:
                time_left = timeout_time - time.time()
                all_ops[0].event.wait(timeout=time_left)
                remaining_ops = [x for x in all_ops if not x.event.isSet()]
                count_remaining = len(remaining_ops)
                count_verified = len(all_ops) - len(remaining_ops)
                logger.info(
                    "Flood check - after {} seconds, {} operations verified, {} remaining".format(
                        time.time() - start_time, count_verified, count_remaining,
                    )
                )
                self.metrics.send_message_count_verified.add(count_verified)
                all_ops = remaining_ops

            if len(all_ops):
                logger.warning("Flood complete.  {} opperations timed out".format(len(all_ops)))
                self.metrics.send_message_count_timed_out.add(len(all_ops))
            else:
                logger.info("Flood complete.  All operations verified")

        # Finally the main loop of the flood thread.  This just sleeps for a while, then queues
        # up the flood in a different thread.  This means that floods can theoretically overlap.
        while not self.done.isSet():
            self.sleep_with_watchdog(self.config[Settings.SEND_MESSAGE_FLOOD_INTERVAL_IN_SECONDS])
            if self.done.isSet():
                return

            self.outgoing_executor.submit(do_flood, long_run=True)

    def handle_desired_property_patch_received(self, patch):
        """
        callback for desired property patch reception
        """
        # props that have the pairing structure go to `incoming_pairing_message_queue`
        if patch.get(Fields.PAIRING, {}):
            self.incoming_pairing_message_queue.put(patch)

        # other props go into incoming_deisred_property_patch_queue
        else:
            self.incoming_desired_property_patch_queue.put(patch)

    def handle_message_received(self, msg):
        """
        Callback for receiving c2d messages.
        """
        body = json.loads(msg.data.decode())

        if body and body[Fields.RUN_ID] == run_id:
            if body[Fields.CMD] in [
                Commands.PAIR_RESPONSE,
                Commands.OPERATION_RESPONSE,
                Commands.METHOD_RESPONSE,
                Commands.C2D_RESPONSE,
            ]:
                self.incoming_executor.submit(self.handle_operation_response, msg)

            else:
                logger.warning("Unknown command received: {}".format(body))

        else:
            logger.warning("C2D received, but it's not for us: {}".format(body))

    def handle_operation_response(self, msg):
        """
        Code to handle incoming operationResponse messages.  This is where we go through the list
        of `operationId` values that we received and call the appropriate callbacks to indicate that
        the service has responded.
        """

        body = json.loads(msg.data.decode())

        operation_ids = body.get(Fields.OPERATION_IDS, [])
        if not operation_ids:
            operation_ids = [
                body.get(Fields.OPERATION_ID),
            ]

        logger.info("Received {} message with {}".format(body[Fields.CMD], operation_ids))

        for operation_id in operation_ids:
            operation_ticket = self.operation_ticket_list.get(operation_id)
            if operation_ticket:
                operation_ticket.result_message = msg
                operation_ticket.complete()
            else:
                logger.warning("Received unknown operationId: {}:".format(operation_id))

    def operation_test_thread(self):
        """
        Thread which loops through various hub features and tests them one at a time.
        """

        tests = [
            (self.test_reported_properties, (), {}),
            (self.test_desired_properties, (), {}),
            (self.test_method_invoke, [TestMethod(MethodNames.FAIL_WITH_404, 404, False)], {}),
            (self.test_method_invoke, [TestMethod(MethodNames.ECHO_REQUEST, 200, True)], {}),
            (
                self.test_method_invoke,
                [TestMethod(MethodNames.UNDEFINED_METHOD_NAME, 500, False)],
                {},
            ),
            (self.test_c2d, (), {}),
        ]

        while not self.done.isSet():
            for (function, args, kwargs) in tests:
                reset_watchdog()
                self.outgoing_executor.submit(function, *args, **kwargs)

                time.sleep(self.config[Settings.INTER_TEST_DELAY_INTERVAL_IN_SECONDS])
                if self.done.isSet():
                    break

    def test_reported_properties(self):
        """
        test function to send a single reported property and then clear it after the
        server verifies it. It does this by setting properties inside
        `properties/reported/testContent/reportedPropertyTest`.  Each property has
        a `addOperationId` value and a `removeOperationId` value.  When  the service sees the
        property added, it sends the `addOperationId` to the device.  When the service sees the
        property removed, it sends the `removeOperationId` to the device. This way the device
        can add a property, verify that it was added, then remove it and verify that it was removed.
        """

        # Get metrics.  We report them to Azure monitor and also push them as part of the twin
        metrics = {
            Fields.TEST_METRICS: self.get_test_metrics(),
            Fields.SYSTEM_HEALTH_METRICS: self.get_system_health_telemetry(),
        }
        self.send_metrics_to_azure_monitor(metrics)

        # session metrics goes into reported propertes, but not azure monitor
        # They're either never-changing (like start time) or redundant (like elapsed time)
        metrics[Fields.SESSION_METRICS] = self.get_session_metrics()

        def make_reported_prop(property_name, val):
            return {Fields.TEST_CONTENT: {Fields.REPORTED_PROPERTY_TEST: {property_name: val}}}

        add_operation = self.operation_ticket_list.make_event_based_operation_ticket()
        remove_operation = self.operation_ticket_list.make_event_based_operation_ticket()

        prop_name = "prop_{}".format(self.reported_property_index)
        self.reported_property_index += 1

        prop_value = {
            Fields.ADD_OPERATION_ID: add_operation.id,
            Fields.REMOVE_OPERATION_ID: remove_operation.id,
        }

        logger.info("Adding test property {}".format(prop_name))
        props = make_reported_prop(prop_name, prop_value)
        props.update(metrics)
        self.client.patch_twin_reported_properties(props)

        if add_operation.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info("Add of reported property {} verified by service".format(prop_name))
            self.metrics.reported_properties_count_added.increment()
        else:
            logger.info("Add of reported property {} verification timeout".format(prop_name))
            self.metrics.reported_properties_count_timed_out.increment()

        logger.info("Removing test property {}".format(prop_name))
        self.client.patch_twin_reported_properties(make_reported_prop(prop_name, None))

        if remove_operation.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info("Remove of reported property {} verified by service".format(prop_name))
            self.metrics.reported_properties_count_removed.increment()
        else:
            logger.info("Remove of reported property {} verification timeout".format(prop_name))
            self.metrics.reported_properties_count_timed_out.increment()

    @return_if_already_running
    def test_desired_properties(self):
        """
        Test function to set a desired property, wait for the patch to arrive at the client,
        then get the twin to verify that it shows up in the twin.
        """
        twin_guid = str(uuid.uuid4())

        payload_set_desired_props = {
            Fields.CMD: Commands.SET_DESIRED_PROPS,
            Fields.DESIRED_PROPERTIES: {Fields.TEST_CONTENT: {Fields.TWIN_GUID: twin_guid}},
        }
        msg = self.create_message_from_dict(payload_set_desired_props)
        logger.info("Sending message to update desired getTwin property to {}".format(twin_guid))
        self.client.send_message(msg)

        start_time = time.time()
        end_time = start_time + self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]

        actual_twin_guid = None
        while time.time() <= end_time:
            try:
                patch = self.incoming_desired_property_patch_queue.get(
                    max(0, end_time - time.time())
                )
            except queue.Empty:
                break
            else:
                test_content = patch.get(Fields.TEST_CONTENT, {})
                actual_twin_guid = test_content.get(Fields.TWIN_GUID)
                if actual_twin_guid == twin_guid:
                    logger.info("received expected patch: {} succeeded".format(twin_guid))
                    self.metrics.desired_property_patch_count_received.increment()
                    break
                else:
                    logger.info(
                        "Did not receive expected patch (yet). actual = {}.  expected = {}".format(
                            actual_twin_guid, twin_guid
                        )
                    )

        if actual_twin_guid != twin_guid:
            logger.warning(
                "wait_for_desired_property_patch did not retrieve expected properties within expected time.  expected={}".format(
                    twin_guid
                )
            )
            self.metrics.desired_property_patch_count_timed_out.increment()

        # Now test the twin.  Since we already received the patch, our property should
        # be in the twin.
        twin = self.client.get_twin()
        logger.info("Got twin: {}".format(twin))

        desired = twin.get(Fields.DESIRED, {})
        test_content = desired.get(Fields.TEST_CONTENT, {})
        actual_twin_guid = test_content.get(Fields.TWIN_GUID)

        if actual_twin_guid == twin_guid:
            logger.info("Got desired_twin: {} succeeded".format(twin_guid))
            self.metrics.get_twin_count_succeeded.increment()
        else:
            logger.info(
                "get_twin property did not match: actual = {} expected = {}".format(
                    actual_twin_guid, twin_guid
                )
            )
            self.metrics.get_twin_count_timed_out_out.increment()

    def handle_method_received(self, method_request):
        def handle(method_request):
            logger.info("Received method request {}".format(method_request.name))
            self.metrics.method_invoke_count_request_received.increment()
            if method_request.name == MethodNames.FAIL_WITH_404:
                logger.info("sending response to fail with 404")
                self.client.send_method_response(
                    MethodResponse.create_from_method_request(method_request, 404)
                )
            elif method_request.name == MethodNames.ECHO_REQUEST:
                self.client.send_method_response(
                    MethodResponse.create_from_method_request(
                        method_request, 200, method_request.payload,
                    )
                )
            else:
                logger.info("sending response to fail with 500")
                self.client.send_method_response(
                    MethodResponse.create_from_method_request(method_request, 500)
                )

        self.incoming_executor.submit(handle, method_request)

    def make_random_payload(self):
        return {
            "random_guid": str(uuid.uuid4()),
            "sub_object": {
                "string_value": str(uuid.uuid4()),
                "bool_value": random.random() > 0.5,
                "int_value": random.randint(-65535, 65535),
            },
        }

    def test_method_invoke(self, method):
        operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()

        if method.include_payload:
            payload = self.make_random_payload()
        else:
            payload = None

        command_payload = {
            Fields.CMD: Commands.INVOKE_METHOD,
            Fields.OPERATION_ID: operation_ticket.id,
            Fields.METHOD_NAME: method.method_name,
            Fields.METHOD_INVOKE_PAYLOAD: payload,
        }
        msg = self.create_message_from_dict(command_payload)
        logger.info("sending method invoke with guid={}".format(operation_ticket.id))
        self.client.send_message(msg)

        if operation_ticket.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info(
                "method with guid {} returned {}".format(
                    operation_ticket.id, operation_ticket.result_message
                )
            )
            body = json.loads(operation_ticket.result_message.data.decode())
            fail = False
            if body[Fields.METHOD_RESPONSE_STATUS_CODE] != method.expected_status_code:
                logger.error(
                    "Unexpected method status: id={}, received {} expected {}".format(
                        operation_ticket.id,
                        body[Fields.METHOD_RESPONSE_STATUS_CODE],
                        method.expected_status_code,
                    )
                )
                fail = True
            if body[Fields.METHOD_RESPONSE_PAYLOAD] != payload:
                logger.error(
                    "Unexpected payload: id={}, received {} expected {}".format(
                        operation_ticket.id, body[Fields.METHOD_RESPONSE_PAYLOAD], payload
                    )
                )
                fail = True

            if fail:
                raise ThiefFatalException("Content mismatch on method invoke")
            else:
                logger.info("Method call check succeeded")

        else:
            logger.error("method with guid {} never completed".format(operation_ticket.id))
            self.metrics.method_invoke_count_request_timed_out = ThreadSafeCounter()

    def test_c2d(self):

        operation_ticket = self.operation_ticket_list.make_event_based_operation_ticket()

        sent_test_payload = self.make_random_payload()

        command_payload = {
            Fields.CMD: Commands.SEND_C2D,
            Fields.OPERATION_ID: operation_ticket.id,
            Fields.TEST_C2D_PAYLOAD: sent_test_payload,
        }

        logger.info("Requesting C2D for operationId {}".format(operation_ticket.id))
        self.client.send_message(self.create_message_from_dict(command_payload))

        if operation_ticket.event.wait(timeout=self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]):
            logger.info("C2D received for operationId {}".format(operation_ticket.id))
            self.metrics.receive_c2d_count_received.increment()

            body = json.loads(operation_ticket.result_message.data.decode())
            if body[Fields.TEST_C2D_PAYLOAD] != sent_test_payload:
                logger.warning(
                    "C2D payload for operationId {}  does not match.  Expected={}, Received={}".format(
                        operation_ticket.id, sent_test_payload, body[Fields.TEST_C2D_PAYLOAD]
                    )
                )
                raise ThiefFatalException("C2D payload mismatch")
        else:
            logger.info("C2D timed out for operationId {}".format(operation_ticket.id))
            self.metrics.receive_c2d_count_timed_out.increment()

    def wrap_handler(self, func):
        """
        Function to take a handler function and wrap it inside a try/catch block
        which increments exception count.  This is done because iothub callbacks
        will catch and swallow any exceptions that get raised in handlers.
        """

        def newfunc(*args, **kwargs):
            try:
                func(*args, **kwargs)
            except Exception as e:
                logger.error("exception in handler: {}".format(str(e) or type(e)), exc_info=True)
                self.metrics.exception_count.increment()

        return newfunc

    def main(self):

        self.metrics.run_start_utc = datetime.datetime.now(datetime.timezone.utc)
        self.metrics.run_state = RunStates.RUNNING
        logger.info("Starting at {}".format(self.metrics.run_start_utc))

        # Create our client and push initial properties
        self.client, self.hub, self.device_id = dps.create_device_client_using_dps_group_key(
            provisioning_host=provisioning_host,
            registration_id=registration_id,
            id_scope=id_scope,
            group_symmetric_key=group_symmetric_key,
            keep_alive=self.config[Settings.MQTT_KEEP_ALIVE_INTERVAL],
            sastoken_ttl=self.config[Settings.SAS_TOKEN_RENEWAL_INTERVAL],
        )
        hub_host_name = self.hub[: self.hub.find(".")]  # keep everything before the first dot
        azure_monitor.add_logging_properties(hub=hub_host_name, device_id=self.device_id)
        self.update_initial_reported_properties()

        event_logger.info(
            Events.STARTING_RUN, extra=custom_props({CustomDimensions.RUN_REASON: run_reason})
        )

        # Set handlers for incoming hub messages
        self.client.on_message_received = self.wrap_handler(self.handle_message_received)
        self.client.on_method_request_received = self.wrap_handler(self.handle_method_received)
        self.client.on_twin_desired_properties_patch_received = self.wrap_handler(
            self.handle_desired_property_patch_received
        )

        # Pair
        self.pair_with_service()

        # Start testing threads
        self.outgoing_executor.submit(self.test_send_message_thread, critical=True)
        self.outgoing_executor.submit(self.operation_test_thread, critical=True)
        if self.config[Settings.DO_SEND_MESSAGE_FLOODS]:
            self.outgoing_executor.submit(self.test_flood_thread, critical=True)

        self.metrics.exit_reason = ""
        start_time = time.time()
        planned_end_time = start_time + self.config[Settings.MAX_RUN_DURATION_IN_SECONDS]
        try:
            while time.time() < planned_end_time:
                # TODO: limit sleep length to check failure counts more often
                self.outgoing_executor.wait_for_thread_death_event(planned_end_time - time.time())
                self.outgoing_executor.check_watchdogs()
                self.incoming_executor.check_watchdogs()
                self.check_failure_counts()

                non_fatal_errors = 0
                fatal_error = None

                def count_failure(e):
                    nonlocal non_fatal_errors, fatal_error
                    if isinstance(e, ThiefFatalException):
                        logger.info("FATAL ERROR: {}".format(str(e) or type(e)))
                        fatal_error = e
                    else:
                        non_fatal_errors += 1

                self.outgoing_executor.check_for_failures(count_failure)
                self.incoming_executor.check_for_failures(count_failure)
                if non_fatal_errors:
                    logger.info("Adding {} failures to count".format(non_fatal_errors))
                    self.metrics.exception_count.add(non_fatal_errors)
                if fatal_error:
                    raise (fatal_error)
            self.metrics.run_state = RunStates.COMPLETE
        except KeyboardInterrupt:
            self.metrics.run_state = RunStates.INTERRUPTED
            self.metrics.exit_reason = "KeyboardInterrupt"
            raise
        except BaseException as e:
            self.metrics.run_state = RunStates.FAILED
            self.metrics.exit_reason = str(e) or type(e)
            logger.error("Run failed: {}".format(self.metrics.exit_reason), exc_info=True)
            raise
        finally:
            logger.info("-------------------------------------------------------------")
            logger.info("Setting done flag: {}".format(self.metrics.exit_reason))
            logger.info("-------------------------------------------------------------")
            self.done.set()
            active_threads = 0
            for executor in [self.outgoing_executor, self.incoming_executor]:
                done, not_done = executor.wait(
                    timeout=2 * self.config[Settings.OPERATION_TIMEOUT_IN_SECONDS]
                )
                active_threads += len(not_done)
            if active_threads:
                logger.error("{} threads not stopped after ending run".format(active_threads))
                dump_active_stacks(logger.info)

            logger.info("Exiting main at {}".format(datetime.datetime.utcnow()))
            event_logger.info(
                Events.ENDING_RUN,
                extra=custom_props({CustomDimensions.EXIT_REASON: self.metrics.exit_reason}),
            )

            # Update one last time.  This is required because the service app relies
            # on RUN_STATE to know when we're dead
            props = {
                Fields.SESSION_METRICS: self.get_session_metrics(),
                Fields.TEST_METRICS: self.get_test_metrics(),
                Fields.SYSTEM_HEALTH_METRICS: self.get_system_health_telemetry(),
            }
            logger.info("Results: {}".format(pprint.pformat(props)))
            self.client.patch_twin_reported_properties(props)

            logger.info("Disconnecting")
            self.client.disconnect()
            logger.info("Done disconnecting")


if __name__ == "__main__":
    try:
        if len(sys.argv) > 1:
            run_reason = " ".join(sys.argv[1:])
        DeviceApp().main()
    except BaseException as e:
        logger.critical("App shutdown exception: {}".format(str(e) or type(e)), exc_info=True)
        raise
    finally:
        # Flush azure monitor telemetry
        logging.shutdown()
# Copyright (c) Microsoft. All rights reserved. # Licensed under the MIT license. See LICENSE file in the project root for # full license information. # run_reason can be an environment variable or is can be passed on the command line. # run_id can be an environment variable or it can be automatically generated # set default logging which will only go to the console # logging.getLogger("paho").setLevel(level=logging.DEBUG) # logging.getLogger("azure.iot").setLevel(level=logging.INFO) # configure our traces and events to go to Azure Monitor helper function for adding customDimensions to logger calls at execution time Object we use internally to keep track of how a the entire test is performing. Object we use internally to keep track of how the entire test is configured. Currently hardcoded. Later, this will come from desired properties. Decorator function to make sure that the wrapped function cannot be called in parallel.     The wrapped funciton can be invoked once.  If is called again before the first invocation     is complete, it immediately returns without running.  Once the first invocation completes,     then it an be invoked again. Main application object # for running operations # for metrics # for pairing # for twin # --------------------- # System Health metrics # --------------------- # ---------------- # test app metrics # ---------------- # -------------------- # SendMesssage metrics # -------------------- # ------------------- # Receive c2d metrics # ------------------- # ------------------------- # Reported property metrics # ------------------------- # ---------------- # Get-twin metrics # ---------------- # ------------------------------ # desired property patch metrics # ------------------------------ # --------------------- # method invoke metrics # --------------------- # --------------- # Latency metrics # --------------- Return metrics which describe the session the tests are running in Return metrics which describe the progress of the  different features being tested Check all failure counts and raise an exception if they exceeded the allowed count Update reported properties at the start of a run # First remove all old props (in case of schema change).  Also clears out old results. helper function to create a message from a dict object # Note: we're changing the dictionary that the user passed in. # This isn't the best idea, but it works and it saves us from deep copies # This function only creates the message.  The caller needs to queue it up for sending. "pair" with a service app. This is necessary because we can have a single         service app responsible for multiple device apps and we want to make sure         the service app is actually running before we continue. Send metrics to azure monitor, based on the reported properties that we probably just         sent to the hub # We don't record session_metrics to Azure Monitor because the session metrics are # recording things like "start time" and "elapsed time" which are already available # in Azure Monitor in other forms. Wrapper around send_message which sends a message to iothub and keeps track of all         necessary metric. Thread to continuously send d2c messages throughout the longhaul run.  This thread doesn't         actually send messages because send_message is blocking and we want to overlap our send         operations.  Instead, this thread creates new transient threads that actually send the         message. Helper function to create and send send a single message to iothub.  Intended             to run in it's own thread.  Returns after the service acknowledges receipt of             the message. # sleep until we need to send again Execute a `time.sleep` operation which will not get interrupted by a watchdog timeout. # -10 for headroom A thread which runs occasional message floods on some regular interval.  A message         flood is a large number of send_message calls in a short period of time. Helper function to create and send send a single message to iothub.  Intended             to run in it's own thread.  Returns when the PUBACK returns. Function to run send a flood of messages to iothub.  Intended to run in its own             thread.  Waits for the server to acknowledge all messages before returning. # Make RunningOperation objects for all of our messages. # Queue up sends for all these messages.  The number of sends which can be outstanding # on the wire is limited by the number of threads in the outgoing_executor thread pool. # Wait for all of the sends to complete.  We do this by waiting for some arbitrary # operation to complete, then we prune the list of remaining operations and wait # for another arbitrary operation until the list of remaining operations is empty. # Finally the main loop of the flood thread.  This just sleeps for a while, then queues # up the flood in a different thread.  This means that floods can theoretically overlap. callback for desired property patch reception # props that have the pairing structure go to `incoming_pairing_message_queue` # other props go into incoming_deisred_property_patch_queue Callback for receiving c2d messages. Code to handle incoming operationResponse messages.  This is where we go through the list         of `operationId` values that we received and call the appropriate callbacks to indicate that         the service has responded. Thread which loops through various hub features and tests them one at a time. test function to send a single reported property and then clear it after the         server verifies it. It does this by setting properties inside         `properties/reported/testContent/reportedPropertyTest`.  Each property has         a `addOperationId` value and a `removeOperationId` value.  When  the service sees the         property added, it sends the `addOperationId` to the device.  When the service sees the         property removed, it sends the `removeOperationId` to the device. This way the device         can add a property, verify that it was added, then remove it and verify that it was removed. # Get metrics.  We report them to Azure monitor and also push them as part of the twin # session metrics goes into reported propertes, but not azure monitor # They're either never-changing (like start time) or redundant (like elapsed time) Test function to set a desired property, wait for the patch to arrive at the client,         then get the twin to verify that it shows up in the twin. # Now test the twin.  Since we already received the patch, our property should # be in the twin. Function to take a handler function and wrap it inside a try/catch block         which increments exception count.  This is done because iothub callbacks         will catch and swallow any exceptions that get raised in handlers. # Create our client and push initial properties # keep everything before the first dot # Set handlers for incoming hub messages # Pair # Start testing threads # TODO: limit sleep length to check failure counts more often # Update one last time.  This is required because the service app relies # on RUN_STATE to know when we're dead # Flush azure monitor telemetry