|
| 1 | +# BSD 3-Clause License |
| 2 | +# |
| 3 | +# Copyright (c) 2019, Elasticsearch BV |
| 4 | +# All rights reserved. |
| 5 | +# |
| 6 | +# Redistribution and use in source and binary forms, with or without |
| 7 | +# modification, are permitted provided that the following conditions are met: |
| 8 | +# |
| 9 | +# * Redistributions of source code must retain the above copyright notice, this |
| 10 | +# list of conditions and the following disclaimer. |
| 11 | +# |
| 12 | +# * Redistributions in binary form must reproduce the above copyright notice, |
| 13 | +# this list of conditions and the following disclaimer in the documentation |
| 14 | +# and/or other materials provided with the distribution. |
| 15 | +# |
| 16 | +# * Neither the name of the copyright holder nor the names of its |
| 17 | +# contributors may be used to endorse or promote products derived from |
| 18 | +# this software without specific prior written permission. |
| 19 | +# |
| 20 | +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" |
| 21 | +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE |
| 22 | +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE |
| 23 | +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE |
| 24 | +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL |
| 25 | +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR |
| 26 | +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER |
| 27 | +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, |
| 28 | +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
| 29 | +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| 30 | + |
| 31 | +import time |
| 32 | +from typing import Optional |
| 33 | + |
| 34 | +import elasticapm |
| 35 | +from elasticapm import get_client |
| 36 | +from elasticapm.conf import constants |
| 37 | +from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule |
| 38 | +from elasticapm.traces import DroppedSpan, capture_span, execution_context |
| 39 | +from elasticapm.utils.disttracing import TraceParent |
| 40 | + |
| 41 | + |
| 42 | +class KafkaInstrumentation(AbstractInstrumentedModule): |
| 43 | + |
| 44 | + instrument_list = [ |
| 45 | + ("kafka", "KafkaProducer.send"), |
| 46 | + ("kafka", "KafkaConsumer.poll"), |
| 47 | + ("kafka", "KafkaConsumer.__next__"), |
| 48 | + ] |
| 49 | + provider_name = "kafka" |
| 50 | + name = "kafka" |
| 51 | + |
| 52 | + def _trace_send(self, instance, wrapped, *args, destination_info=None, **kwargs): |
| 53 | + topic = args[0] if args else kwargs["topic"] |
| 54 | + headers = args[4] if len(args) > 4 else kwargs.get("headers", None) |
| 55 | + |
| 56 | + span_name = f"Kafka SEND to {topic}" |
| 57 | + destination_info["service"]["resource"] += topic |
| 58 | + with capture_span( |
| 59 | + name=span_name, |
| 60 | + span_type="messaging", |
| 61 | + span_subtype=self.provider_name, |
| 62 | + span_action="send", |
| 63 | + leaf=True, |
| 64 | + extra={ |
| 65 | + "message": {"queue": {"name": topic}}, |
| 66 | + "destination": destination_info, |
| 67 | + }, |
| 68 | + ) as span: |
| 69 | + transaction = execution_context.get_transaction() |
| 70 | + if transaction: |
| 71 | + tp = transaction.trace_parent.copy_from(span_id=span.id) |
| 72 | + if headers: |
| 73 | + headers.append((constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary())) |
| 74 | + else: |
| 75 | + headers = [(constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary())] |
| 76 | + if len(args) > 4: |
| 77 | + args = list(args) |
| 78 | + args[4] = headers |
| 79 | + else: |
| 80 | + kwargs["headers"] = headers |
| 81 | + result = wrapped(*args, **kwargs) |
| 82 | + if instance and instance._metadata.controller and not isinstance(span, DroppedSpan): |
| 83 | + address = instance._metadata.controller[1] |
| 84 | + port = instance._metadata.controller[2] |
| 85 | + span.context["destination"]["address"] = address |
| 86 | + span.context["destination"]["port"] = port |
| 87 | + return result |
| 88 | + |
| 89 | + def call_if_sampling(self, module, method, wrapped, instance, args, kwargs): |
| 90 | + # Contrasting to the superclass implementation, we *always* want to |
| 91 | + # return a proxied connection, even if there is no ongoing elasticapm |
| 92 | + # transaction yet. This ensures that we instrument the cursor once |
| 93 | + # the transaction started. |
| 94 | + return self.call(module, method, wrapped, instance, args, kwargs) |
| 95 | + |
| 96 | + def call(self, module, method, wrapped, instance, args, kwargs): |
| 97 | + client = get_client() |
| 98 | + destination_info = { |
| 99 | + "service": {"name": "kafka", "resource": "kafka/", "type": "messaging"}, |
| 100 | + } |
| 101 | + |
| 102 | + if method == "KafkaProducer.send": |
| 103 | + topic = args[0] if args else kwargs["topic"] |
| 104 | + if client.should_ignore_topic(topic) or not execution_context.get_transaction(): |
| 105 | + return wrapped(*args, **kwargs) |
| 106 | + return self._trace_send(instance, wrapped, destination_info=destination_info, *args, **kwargs) |
| 107 | + |
| 108 | + elif method == "KafkaConsumer.poll": |
| 109 | + transaction = execution_context.get_transaction() |
| 110 | + if transaction: |
| 111 | + with capture_span( |
| 112 | + name="Kafka POLL", |
| 113 | + span_type="messaging", |
| 114 | + span_subtype=self.provider_name, |
| 115 | + span_action="poll", |
| 116 | + leaf=True, |
| 117 | + extra={ |
| 118 | + "destination": destination_info, |
| 119 | + }, |
| 120 | + ) as span: |
| 121 | + if not isinstance(span, DroppedSpan) and instance._subscription.subscription: |
| 122 | + span.name += " from " + ", ".join(sorted(instance._subscription.subscription)) |
| 123 | + results = wrapped(*args, **kwargs) |
| 124 | + return results |
| 125 | + else: |
| 126 | + return wrapped(*args, **kwargs) |
| 127 | + |
| 128 | + elif method == "KafkaConsumer.__next__": |
| 129 | + transaction = execution_context.get_transaction() |
| 130 | + if transaction and transaction.transaction_type != "messaging": |
| 131 | + # somebody started a transaction outside of the consumer, |
| 132 | + # so we capture it as a span, and record the causal trace as a link |
| 133 | + with capture_span( |
| 134 | + name="consumer", |
| 135 | + span_type="messaging", |
| 136 | + span_subtype=self.provider_name, |
| 137 | + span_action="receive", |
| 138 | + leaf=True, |
| 139 | + extra={ |
| 140 | + "message": {"queue": {"name": ""}}, |
| 141 | + "destination": destination_info, |
| 142 | + }, |
| 143 | + ) as span: |
| 144 | + try: |
| 145 | + result = wrapped(*args, **kwargs) |
| 146 | + except StopIteration: |
| 147 | + span.cancel() |
| 148 | + raise |
| 149 | + if not isinstance(span, DroppedSpan): |
| 150 | + topic = result[0] |
| 151 | + if client.should_ignore_topic(topic): |
| 152 | + span.cancel() |
| 153 | + return result |
| 154 | + trace_parent = self.get_traceparent_from_result(result) |
| 155 | + if trace_parent: |
| 156 | + span.add_link(trace_parent) |
| 157 | + destination_info["service"]["resource"] += topic |
| 158 | + span.context["message"]["queue"]["name"] = topic |
| 159 | + span.name = "Kafka RECEIVE from " + topic |
| 160 | + return result |
| 161 | + else: |
| 162 | + # No transaction running, or this is a transaction started by us, |
| 163 | + # so let's end it and start the next, |
| 164 | + # unless a StopIteration is raised, at which point we do nothing. |
| 165 | + if transaction: |
| 166 | + client.end_transaction() |
| 167 | + result = wrapped(*args, **kwargs) |
| 168 | + topic = result[0] |
| 169 | + if client.should_ignore_topic(topic): |
| 170 | + return result |
| 171 | + trace_parent = self.get_traceparent_from_result(result) |
| 172 | + transaction = client.begin_transaction("messaging", trace_parent=trace_parent) |
| 173 | + if result.timestamp_type == 0: |
| 174 | + current_time_millis = int(round(time.time() * 1000)) |
| 175 | + age = current_time_millis - result.timestamp |
| 176 | + transaction.context = { |
| 177 | + "message": {"age": {"ms": age}, "queue": {"name": topic}}, |
| 178 | + "service": {"framework": {"name": "Kafka"}}, |
| 179 | + } |
| 180 | + transaction_name = "Kafka RECEIVE from " + topic |
| 181 | + elasticapm.set_transaction_name(transaction_name, override=True) |
| 182 | + res = constants.OUTCOME.SUCCESS |
| 183 | + elasticapm.set_transaction_result(res, override=False) |
| 184 | + return result |
| 185 | + |
| 186 | + def get_traceparent_from_result(self, result) -> Optional[TraceParent]: |
| 187 | + for k, v in result.headers: |
| 188 | + if k == constants.TRACEPARENT_BINARY_HEADER_NAME: |
| 189 | + return TraceParent.from_binary(v) |
0 commit comments