1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
from __future__ import print_function
from tornado import gen, ioloop, queues
import opentracing
from opentracing.ext import tags
from opentracing.mocktracer import MockTracer
from opentracing.scope_managers.tornado import TornadoScopeManager, \
tracer_stack_context
from ..testcase import OpenTracingTestCase
from ..utils import get_logger, get_one_by_tag, stop_loop_when
logger = get_logger(__name__)
class Server(object):
def __init__(self, *args, **kwargs):
tracer = kwargs.pop('tracer')
queue = kwargs.pop('queue')
super(Server, self).__init__(*args, **kwargs)
self.tracer = tracer
self.queue = queue
@gen.coroutine
def run(self):
value = yield self.queue.get()
self.process(value)
def process(self, message):
logger.info('Processing message in server')
ctx = self.tracer.extract(opentracing.Format.TEXT_MAP, message)
with self.tracer.start_active_span('receive',
child_of=ctx) as scope:
scope.span.set_tag(tags.SPAN_KIND, tags.SPAN_KIND_RPC_SERVER)
class Client(object):
def __init__(self, tracer, queue):
self.tracer = tracer
self.queue = queue
@gen.coroutine
def send(self):
with self.tracer.start_active_span('send') as scope:
scope.span.set_tag(tags.SPAN_KIND, tags.SPAN_KIND_RPC_CLIENT)
message = {}
self.tracer.inject(scope.span.context,
opentracing.Format.TEXT_MAP,
message)
yield self.queue.put(message)
logger.info('Sent message from client')
class TestTornado(OpenTracingTestCase):
def setUp(self):
self.tracer = MockTracer(TornadoScopeManager())
self.queue = queues.Queue()
self.loop = ioloop.IOLoop.current()
self.server = Server(tracer=self.tracer, queue=self.queue)
def test(self):
client = Client(self.tracer, self.queue)
with tracer_stack_context():
self.loop.add_callback(self.server.run)
self.loop.add_callback(client.send)
stop_loop_when(self.loop,
lambda: len(self.tracer.finished_spans()) >= 2)
self.loop.start()
spans = self.tracer.finished_spans()
self.assertIsNotNone(get_one_by_tag(spans,
tags.SPAN_KIND,
tags.SPAN_KIND_RPC_SERVER))
self.assertIsNotNone(get_one_by_tag(spans,
tags.SPAN_KIND,
tags.SPAN_KIND_RPC_CLIENT))
|
