Fix: RabbitMQ check proper span type (#309)

* Fix: RabbitMQ check proper span type
Closes #301

Author: pdimitra

.circleci/config.yml

@@ -76,6 +76,30 @@ jobs:
             . venv/bin/activate
             pytest -v
 
+  python37:
+    docker:
+      - image: circleci/python:3.7.9
+      - image: circleci/postgres:9.6.5-alpine-ram
+      - image: circleci/mariadb:10-ram
+      - image: circleci/redis:5.0.4
+      - image: rabbitmq:3.5.4
+      - image: circleci/mongo:4.2.3-ram
+      - image: singularities/pubsub-emulator
+        environment:
+          PUBSUB_PROJECT_ID: "project-test"
+          PUBSUB_LISTEN_ADDRESS: "0.0.0.0:8432"
+    working_directory: ~/repo
+    steps:
+      - checkout
+      - pip-install-deps
+      - run:
+          name: run tests
+          environment:
+            INSTANA_TEST: "true"
+          command: |
+            . venv/bin/activate
+            pytest -v
+
   python38:
     docker:
       - image: circleci/python:3.8.6
@@ -227,6 +251,7 @@ workflows:
   build:
     jobs:
       - python27
+      - python37
       - python38
       - python39
       - py27cassandra

instana/span.py

@@ -26,7 +26,7 @@ class InstanaSpan(BasicSpan):
     stack = None
     synthetic = False
 
-    def mark_as_errored(self, tags = None):
+    def mark_as_errored(self, tags=None):
         """
         Mark this span as errored.
 
@@ -90,6 +90,7 @@ def log_exception(self, exc):
             logger.debug("span.log_exception", exc_info=True)
             raise
 
+
 class BaseSpan(object):
     sy = None
 
@@ -149,7 +150,7 @@ def _validate_tag(self, key, value):
         try:
             # Tag keys must be some type of text or string type
             if isinstance(key, (six.text_type, six.string_types)):
-                validated_key = key[0:1024] # Max key length of 1024 characters
+                validated_key = key[0:1024]  # Max key length of 1024 characters
 
                 if isinstance(value, (bool, float, int, list, dict, six.text_type, six.string_types)):
                     validated_value = value
@@ -169,7 +170,7 @@ def _convert_tag_value(self, value):
             final_value = repr(value)
         except Exception:
             final_value = "(non-fatal) span.set_tag: values must be one of these types: bool, float, int, list, " \
-                            "set, str or alternatively support 'repr'. tag discarded"
+                          "set, str or alternatively support 'repr'. tag discarded"
             logger.debug(final_value, exc_info=True)
             return None
         return final_value
@@ -247,24 +248,25 @@ def __init__(self, span, source, service_name, **kwargs):
         super(RegisteredSpan, self).__init__(span, source, service_name, **kwargs)
         self.n = span.operation_name
         self.k = 1
+
         if span.operation_name in self.ENTRY_SPANS:
             # entry
             self._populate_entry_span_data(span)
             self.data["service"] = service_name
         elif span.operation_name in self.EXIT_SPANS:
-            self.k = 2 # exit
+            self.k = 2  # exit
             self._populate_exit_span_data(span)
         elif span.operation_name in self.LOCAL_SPANS:
-            self.k = 3 # intermediate span
+            self.k = 3  # intermediate span
             self._populate_local_span_data(span)
 
-        if "rabbitmq" in self.data and self.data["rabbitmq"]["sort"] == "consume":
-            self.k = 1  # entry
+        if "rabbitmq" in self.data and self.data["rabbitmq"]["sort"] == "publish":
+            self.k = 2  # exit
 
         # unify the span operation_name for gcps-producer and gcps-consumer
         if "gcps" in span.operation_name:
             self.n = 'gcps'
-        
+
         # Store any leftover tags in the custom section
         if len(span.tags) > 0:
             self.data["custom"]["tags"] = self._validate_tags(span.tags)

tests/clients/test_asynqp.py

@@ -24,7 +24,8 @@
     rabbitmq_host = "localhost"
 
 is_unsupported_version = LooseVersion(sys.version) < LooseVersion('3.5.3') \
-    or LooseVersion(sys.version) >= LooseVersion('3.8.0')
+                         or LooseVersion(sys.version) >= LooseVersion('3.8.0')
+
 
 @pytest.mark.skipif(is_unsupported_version, reason="Asynqp supports >=3.5.3;<3.8.0")
 class TestAsynqp(unittest.TestCase):
@@ -96,6 +97,9 @@ def test():
         self.assertIsNone(test_span.ec)
         self.assertIsNone(rabbitmq_span.ec)
 
+        # Span type
+        self.assertEqual(rabbitmq_span.k, 2)  # exit
+
         # Rabbitmq
         self.assertEqual('test.exchange', rabbitmq_span.data["rabbitmq"]["exchange"])
         self.assertEqual('publish', rabbitmq_span.data["rabbitmq"]["sort"])
@@ -132,6 +136,9 @@ def test():
         self.assertIsNone(test_span.ec)
         self.assertIsNone(rabbitmq_span.ec)
 
+        # Span type
+        self.assertEqual(rabbitmq_span.k, 2)  # exit
+
         # Rabbitmq
         self.assertEqual('test.exchange', rabbitmq_span.data["rabbitmq"]["exchange"])
         self.assertEqual('publish', rabbitmq_span.data["rabbitmq"]["sort"])
@@ -196,6 +203,10 @@ def publish():
         self.assertEqual(publish_span.p, test_span.s)
         self.assertEqual(get_span.p, test_span.s)
 
+        # Span type
+        self.assertEqual(publish_span.k, 2)  # exit
+        self.assertEqual(get_span.k, 1)  # entry
+
         # Error logging
         self.assertIsNone(test_span.ec)
         self.assertIsNone(publish_span.ec)
@@ -250,6 +261,10 @@ def test():
         self.assertEqual(publish_span.p, test_span.s)
         self.assertEqual(consume_span.p, publish_span.s)
 
+        # Span type
+        self.assertEqual(publish_span.k, 2)  # exit
+        self.assertEqual(consume_span.k, 1)  # entry
+
         # publish
         self.assertEqual('test.exchange', publish_span.data["rabbitmq"]["exchange"])
         self.assertEqual('publish', publish_span.data["rabbitmq"]["sort"])
@@ -311,6 +326,11 @@ def test():
         self.assertEqual(consume1_span.p, publish1_span.s)
         self.assertEqual(publish2_span.p, consume1_span.s)
 
+        # Span type
+        self.assertEqual(publish1_span.k, 2)  # exit
+        self.assertEqual(consume1_span.k, 1)  # entry
+        self.assertEqual(publish2_span.k, 2)  # exit
+
         # publish
         self.assertEqual('test.exchange', publish1_span.data["rabbitmq"]["exchange"])
         self.assertEqual('publish', publish1_span.data["rabbitmq"]["sort"])
@@ -402,6 +422,10 @@ def test():
         self.assertEqual(run_later_span.p, consume_span.s)
         self.assertEqual(wsgi_span.p, aioclient_span.s)
 
+        # Span type
+        self.assertEqual(publish_span.k, 2)  # exit
+        self.assertEqual(consume_span.k, 1)  # entry
+
         # publish
         self.assertEqual('test.exchange', publish_span.data["rabbitmq"]["exchange"])
         self.assertEqual('publish', publish_span.data["rabbitmq"]["sort"])