How to use the instana.singletons.tracer.recorder function in instana

To help you get started, we’ve selected a few instana examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github instana / python-sensor / tests / clients / test_couchbase.py View on Github external
def setup_class(self):
        """ Clear all spans before a test run """
        self.recorder = tracer.recorder
        self.cluster = Cluster('couchbase://%s' % testenv['couchdb_host'])
        self.bucket = Bucket('couchbase://%s/travel-sample' % testenv['couchdb_host'],
                             username=testenv['couchdb_username'], password=testenv['couchdb_password'])
github instana / python-sensor / tests / clients / test_logging.py View on Github external
def setUp(self):
        """ Clear all spans before a test run """
        self.recorder = tracer.recorder
        self.recorder.clear_spans()
        self.logger = logging.getLogger('unit test')
github instana / python-sensor / tests / frameworks / test_grpcio.py View on Github external
def setUp(self):
        """ Clear all spans before a test run """
        self.recorder = tracer.recorder
        self.recorder.clear_spans()
        self.channel = grpc.insecure_channel(testenv["grpc_server"])
        self.server_stub = stan_pb2_grpc.StanStub(self.channel)
        # The grpc client apparently needs a second to connect and initialize
        time.sleep(1)
github instana / python-sensor / tests / clients / test_pymongo.py View on Github external
def setUp(self):
        logger.warning("Connecting to MongoDB mongo://%s:@%s:%s",
                    testenv['mongodb_user'], testenv['mongodb_host'], testenv['mongodb_port'])

        self.conn = pymongo.MongoClient(host=testenv['mongodb_host'], port=int(testenv['mongodb_port']),
                                        username=testenv['mongodb_user'], password=testenv['mongodb_pw'])
        self.conn.test.records.delete_many(filter={})

        self.recorder = tracer.recorder
        self.recorder.clear_spans()
github instana / python-sensor / tests / frameworks / test_flask.py View on Github external
def setUp(self):
        """ Clear all spans before a test run """
        self.http = urllib3.PoolManager()
        self.recorder = tracer.recorder
        self.recorder.clear_spans()
github instana / python-sensor / tests / clients / test_sqlalchemy.py View on Github external
def setUp(self):
        """ Clear all spans before a test run """
        self.recorder = tracer.recorder
        self.recorder.clear_spans()
        self.session = Session()
github instana / python-sensor / tests / frameworks / test_celery.py View on Github external
def test_error_reporting(celery_app, celery_worker):
    result = None
    with tracer.start_active_span('test'):
        result = will_raise_error.apply_async()

    # Wait for jobs to finish
    time.sleep(0.5)

    spans = filter_out_ping_tasks(tracer.recorder.queued_spans())
    assert len(spans) == 4

    filter = lambda span: span.n == "sdk"
    test_span = get_first_span_by_filter(spans, filter)
    assert(test_span)

    filter = lambda span: span.n == "celery-client"
    client_span = get_first_span_by_filter(spans, filter)
    assert(client_span)

    filter = lambda span: span.n == "log"
    log_span = get_first_span_by_filter(spans, filter)
    assert(log_span)

    filter = lambda span: span.n == "celery-worker"
    worker_span = get_first_span_by_filter(spans, filter)
github instana / python-sensor / tests / frameworks / test_django.py View on Github external
def setUp(self):
        """ Clear all spans before a test run """
        self.recorder = tracer.recorder
        self.recorder.clear_spans()
        self.http = urllib3.PoolManager()
github instana / python-sensor / tests / clients / test_mysqlclient.py View on Github external
def setUp(self):
        logger.info("MySQL connecting: %s:@%s:3306/%s", testenv['mysql_user'], testenv['mysql_host'], testenv['mysql_db'])
        self.db = MySQLdb.connect(host=testenv['mysql_host'], port=testenv['mysql_port'],
                                  user=testenv['mysql_user'], passwd=testenv['mysql_pw'],
                                  db=testenv['mysql_db'])
        self.cursor = self.db.cursor()
        self.recorder = tracer.recorder
        self.recorder.clear_spans()
        tracer.cur_ctx = None
github instana / python-sensor / instana / agent / host.py View on Github external
def start(self, _):
        """
        Starts the agent and required threads

        This method is called after a successful announce.  See fsm.py
        """
        logger.debug("Spawning metric & span reporting threads")
        self.should_threads_shutdown.clear()
        self.sensor.start()
        instana.singletons.tracer.recorder.start()