Feat celery plugins

plugins/PY/pinpointPy/Common.py

@@ -19,7 +19,8 @@
 
 # Created by eeliu at 3/5/20
 
-from pinpointPy import Defines, pinpoint, get_logger
+from pinpointPy import Defines, pinpoint
+from pinpointPy.pinpoint import get_logger
 from pinpointPy.TraceContext import get_trace_context
 from functools import wraps
 
@@ -69,7 +70,7 @@ def pinpointTrace(*args, **kwargs):
 
 class PinTrace:
 
-    def __init__(self, name):
+    def __init__(self, name=''):
         self.name = name
 
     def setCurrentTraceNodeId(self, traceId):
@@ -125,7 +126,10 @@ def pinpointTrace(*args, **kwargs):
         return pinpointTrace
 
     def getUniqueName(self):
-        return self.name
+        if self.name:
+            return self.name
+        else:
+            return self.func_name
 
 
 class AsyncPinTrace(PinTrace):
@@ -189,8 +193,10 @@ def __init__(self) -> None:
         self.ParentName = ''
         # REMOTE_ADDRESS field (the same as RemoteAddr)
         self.ParentHost = ''
-        # create a new one, or use an exist parents Tid, this Tid used to generate call-tree
+        # create a new one, or use an exist parents Tid, this Tid is used to generate call-tree
         self.ParentTid = ''
+        # create a new one, or use an exist parent sequence id
+        self.ParentSid = ''
         # if find a error, just fill here. it could rise an error(read mark) in pinpoint-web
         self.Error = ''
 
@@ -292,13 +298,13 @@ def onBefore(self, parentId: int, *args, **kwargs):
         tid = ''
         if header.ParentTid != '':
             tid = header.ParentTid
-            pinpoint.add_trace_header(Defines.PP_PARENT_SPAN_ID, tid, traceId)
+            # pinpoint.add_trace_header(Defines.PP_PARENT_SPAN_ID, tid, traceId)
         else:
             tid = pinpoint.gen_tid()
 
-        pinpoint.add_trace_header(Defines.PP_TRANSCATION_ID, tid, traceId)
+        pinpoint.add_trace_header(Defines.PP_TRANSACTION_ID, tid, traceId)
 
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, tid, traceId)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, tid, traceId)
 
         if header.Error:
             pinpoint.mark_as_error(header.Error, header.Error, 0, traceId)

plugins/PY/pinpointPy/CommonPlugin.py

@@ -72,7 +72,7 @@ def onBefore(self, parentId: int,  *args, **kwargs):
 
         sequence_id = pinpoint.get_sequence_id(traceId)
 
-        tid = pinpoint.get_context(Defines.PP_TRANSCATION_ID, traceId)
+        tid = pinpoint.get_context(Defines.PP_TRANSACTION_ID, traceId)
         seq_id = pinpoint.get_context(Defines.PP_SPAN_ID, traceId)
         app_name = pinpoint.get_context(Defines.PP_APP_NAME, traceId)
         app_id = pinpoint.get_context(Defines.PP_APP_ID, traceId)
@@ -100,17 +100,16 @@ def pp_new_entry_func(*args, **kwargs):
                     Defines.PP_SPAN_ID, seq_id, thread_trace_id)
 
                 pinpoint.add_trace_header(
-                    Defines.PP_TRANSCATION_ID, tid, thread_trace_id)
+                    Defines.PP_TRANSACTION_ID, tid, thread_trace_id)
                 pinpoint.add_context(
-                    Defines.PP_TRANSCATION_ID, tid, thread_trace_id)
+                    Defines.PP_TRANSACTION_ID, tid, thread_trace_id)
 
                 pinpoint.add_trace_header(
                     Defines.PP_SERVER_TYPE, Defines.PYTHON, thread_trace_id)
                 pinpoint.set_async_context(
                     thread_trace_id, async_id, sequence_id)
 
                 if callable(origin_target):
-                    # todo add
                     @PinpointCommonPlugin(origin_target.__name__)
                     def call_origin_target(*args, **kwargs):
                         origin_target(*args, **kwargs)

plugins/PY/pinpointPy/Defines.py

@@ -60,7 +60,7 @@
 
 PP_NGINX_PROXY = 'NP'
 PP_APACHE_PROXY = 'AP'
-PP_TRANSCATION_ID = 'tid'
+PP_TRANSACTION_ID = 'tid'
 PP_SPAN_ID = 'sid'
 PP_NOT_SAMPLED = 's0'
 PP_SAMPLED = 's1'
@@ -74,6 +74,7 @@
 PYTHON = '1700'
 PP_METHOD_CALL = '1701'
 PP_CELERY = '1702'
+PP_CELERY_WORKER = '1703'
 
 PP_REMOTE_METHOD = '9900'
 P_INVOCATION_CALL_TYPE = '100'

plugins/PY/pinpointPy/Django/BaseDjangoRequestPlugins.py

@@ -75,7 +75,7 @@ def onBefore(self, parentId, *args, **kwargs):
             self.tid = headers[Defines.PP_HEADER_PINPOINT_TRACEID]
         else:
             self.tid = pinpoint.gen_tid()
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, self.tid, trace_id)
 
         if Defines.PP_HTTP_PINPOINT_PAPPNAME in headers:
             self.pname = headers[Defines.PP_HTTP_PINPOINT_PAPPNAME]
@@ -133,9 +133,9 @@ def onBefore(self, parentId, *args, **kwargs):
                 Defines.PP_HEADER_PINPOINT_SAMPLED, "s0", trace_id)
 
         pinpoint.add_trace_header(
-            Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+            Defines.PP_TRANSACTION_ID, self.tid, trace_id)
         pinpoint.add_trace_header(Defines.PP_SPAN_ID, self.sid, trace_id)
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, self.tid, trace_id)
         pinpoint.add_context(Defines.PP_SPAN_ID, self.sid, trace_id)
         pinpoint.add_trace_header_v2(
             Defines.PP_HTTP_METHOD, headers["REQUEST_METHOD"], trace_id)

plugins/PY/pinpointPy/Fastapi/AsyRequestPlugin.py

@@ -108,8 +108,8 @@ def onBefore(self, parentId, *args, **kwargs):
             pinpoint.drop_trace(traceId)
             pinpoint.add_context("Pinpoint-Sampled", "s0", traceId)
 
-        pinpoint.add_trace_header(Defines.PP_TRANSCATION_ID, tid, traceId)
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, tid, traceId)
+        pinpoint.add_trace_header(Defines.PP_TRANSACTION_ID, tid, traceId)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, tid, traceId)
         pinpoint.add_trace_header(Defines.PP_SPAN_ID, sid, traceId)
         pinpoint.add_context(Defines.PP_SPAN_ID, sid, traceId)
         return traceId, args, kwargs

plugins/PY/pinpointPy/Fastapi/PinTranscation.py

@@ -108,8 +108,8 @@ def onBefore(self, parentId, *args, **kwargs):
         else:
             tid = pinpoint.gen_tid()
 
-        pinpoint.add_trace_header(Defines.PP_TRANSCATION_ID, tid, traceId)
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, tid, traceId)
+        pinpoint.add_trace_header(Defines.PP_TRANSACTION_ID, tid, traceId)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, tid, traceId)
 
         if header.Error:
             pinpoint.mark_as_error(header.Error, header.Error, 0, traceId)

plugins/PY/pinpointPy/Flask/FlaskPlugins.py

@@ -38,8 +38,11 @@ def onBefore(self, parentId, *args, **kwargs):
         pinpoint.add_trace_header(
             Defines.PP_APP_NAME, pinpoint.app_name(), traceId)
         pinpoint.add_context(Defines.PP_APP_NAME, pinpoint.app_name(), traceId)
+
         pinpoint.add_trace_header(
             Defines.PP_APP_ID, pinpoint.app_id(), traceId)
+        pinpoint.add_context(Defines.PP_APP_ID, pinpoint.app_id(), traceId)
+
         ###############################################################
         pinpoint.add_trace_header(
             Defines.PP_INTERCEPTOR_NAME, 'BaseFlaskrequest', traceId)
@@ -125,8 +128,8 @@ def onBefore(self, parentId, *args, **kwargs):
             pinpoint.add_context(
                 Defines.PP_HEADER_PINPOINT_SAMPLED, "s0", traceId)
 
-        pinpoint.add_trace_header(Defines.PP_TRANSCATION_ID, self.tid, traceId)
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, self.tid, traceId)
+        pinpoint.add_trace_header(Defines.PP_TRANSACTION_ID, self.tid, traceId)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, self.tid, traceId)
         pinpoint.add_trace_header(Defines.PP_SPAN_ID, self.sid, traceId)
         pinpoint.add_context(Defines.PP_SPAN_ID, self.sid, traceId)
         pinpoint.add_trace_header_v2(
@@ -138,6 +141,6 @@ def onEnd(self, traceId, ret):
         super().onEnd(traceId, ret)
         return ret
 
-    def onException(self, e):
-        pinpoint.mark_as_error(str(e), "", 0)
+    def onException(self, traceId, e):
+        pinpoint.mark_as_error(str(e), "", 0, traceId)
         raise e

plugins/PY/pinpointPy/Helper.py

@@ -34,7 +34,7 @@ def generatePinpointHeader(host, headers, traceId=-1):
     headers['Pinpoint-Flags'] = "0"
     headers[Defines.PP_HEADER_PINPOINT_HOST] = host
     headers[Defines.PP_HEADER_PINPOINT_TRACEID] = pinpoint.get_context(
-        Defines.PP_TRANSCATION_ID, traceId)
+        Defines.PP_TRANSACTION_ID, traceId)
     headers[Defines.PP_HEADER_PINPOINT_PSPANID] = pinpoint.get_context(
         Defines.PP_SPAN_ID, traceId)
     nextSeqId = pinpoint.gen_sid()
@@ -128,8 +128,8 @@ def startPinpointByEnviron(environ, trace_id: int):
         pinpoint.add_context(
             Defines.PP_HEADER_PINPOINT_SAMPLED, "s0", trace_id)
 
-    pinpoint.add_trace_header(Defines.PP_TRANSCATION_ID, tid, trace_id)
-    pinpoint.add_context(Defines.PP_TRANSCATION_ID, tid, trace_id)
+    pinpoint.add_trace_header(Defines.PP_TRANSACTION_ID, tid, trace_id)
+    pinpoint.add_context(Defines.PP_TRANSACTION_ID, tid, trace_id)
 
     pinpoint.add_trace_header(Defines.PP_SPAN_ID, sid, trace_id)
     pinpoint.add_context(Defines.PP_SPAN_ID, sid, trace_id)

plugins/PY/pinpointPy/RequestPlugins.py

@@ -72,7 +72,7 @@ def onBefore(self, parentId, *args, **kwargs):
             self.tid = request.headers[Defines.PP_HEADER_PINPOINT_TRACEID]
         else:
             self.tid = pinpoint.gen_tid()
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, self.tid, trace_id)
 
         if Defines.PP_HTTP_PINPOINT_PAPPNAME in request.headers:
             self.pname = request.headers[Defines.PP_HTTP_PINPOINT_PAPPNAME]
@@ -131,9 +131,9 @@ def onBefore(self, parentId, *args, **kwargs):
                 Defines.PP_HEADER_PINPOINT_SAMPLED, "s0", trace_id)
 
         pinpoint.add_trace_header(
-            Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+            Defines.PP_TRANSACTION_ID, self.tid, trace_id)
         pinpoint.add_trace_header(Defines.PP_SPAN_ID, self.sid, trace_id)
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, self.tid, trace_id)
         pinpoint.add_context(Defines.PP_SPAN_ID, self.sid, trace_id)
         return trace_id, args, kwargs
 

plugins/PY/pinpointPy/grpc_/GrpcRequestPlugins.py

@@ -51,11 +51,11 @@ def onBefore(self, parentId, *args, **kwargs):
         self.sid = pinpoint.gen_sid()
         pinpoint.add_context(Defines.PP_SPAN_ID, self.sid, trace_id)
         self.tid = pinpoint.gen_tid()
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, self.tid, trace_id)
         pinpoint.add_trace_header(
-            Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+            Defines.PP_TRANSACTION_ID, self.tid, trace_id)
         pinpoint.add_trace_header(Defines.PP_SPAN_ID, self.sid, trace_id)
-        pinpoint.add_context(Defines.PP_TRANSCATION_ID, self.tid, trace_id)
+        pinpoint.add_context(Defines.PP_TRANSACTION_ID, self.tid, trace_id)
         pinpoint.add_context(Defines.PP_SPAN_ID, self.sid, trace_id)
         return trace_id, args, kwargs
 

plugins/PY/pinpointPy/libs/__init__.py

@@ -15,6 +15,7 @@
 # ------------------------------------------------------------------------------
 import importlib
 from pinpointPy.pinpoint import get_logger
+from pinpointPy.libs._celery import CeleryWorkerPlugin, CeleryCallerPlugin
 
 
 def __monkey_patch(*args, **kwargs):
@@ -37,11 +38,18 @@ def monkey_patch_for_pinpoint(pymongo=True,
                               sqlalchemy=True,
                               MySQLdb=True,
                               MysqlConnector=True,
-                              pyscopg2=True):
+                              pyscopg2=True,
+                              rabbitmq=True,
+                              kombu=True):
     __monkey_patch(_pymongo=pymongo, _MySQLdb=MySQLdb, _PyMysql=PyMysql, _pyRedis=pyRedis, _requests=requests,
-                   _urllib=urllib, _sqlalchemy=sqlalchemy,   _MysqlConnector=MysqlConnector, _psycopg2=pyscopg2)
+                   _urllib=urllib, _sqlalchemy=sqlalchemy,   _MysqlConnector=MysqlConnector, _psycopg2=pyscopg2,
+                   _rabbitmq=rabbitmq, _kombu=kombu)
 
 
-__all__ = ['monkey_patch_for_pinpoint']
-__version__ = '0.0.4'
+__all__ = ['monkey_patch_for_pinpoint',
+           'CeleryWorkerPlugin', 'CeleryCallerPlugin']
+__version__ = '0.0.5'
 __author__ = '[email protected]'
+
+# 0.0.5
+# add rabbitmq

plugins/PY/pinpointPy/libs/_asyncio/__init__.py

@@ -44,7 +44,7 @@ def onBefore(self, parentId: int,  *args, **kwargs):
 
         sequence_id = pinpoint.get_sequence_id(traceId)
 
-        tid = pinpoint.get_context(Defines.PP_TRANSCATION_ID, traceId)
+        tid = pinpoint.get_context(Defines.PP_TRANSACTION_ID, traceId)
         seq_id = pinpoint.get_context(Defines.PP_SPAN_ID, traceId)
         app_name = pinpoint.get_context(Defines.PP_APP_NAME, traceId)
         app_id = pinpoint.get_context(Defines.PP_APP_ID, traceId)
@@ -73,10 +73,10 @@ async def pp_new_main_coroutine():
                 Defines.PP_SPAN_ID, seq_id, thread_trace_id)
 
             pinpoint.add_trace_header(
-                Defines.PP_TRANSCATION_ID, tid, thread_trace_id)
+                Defines.PP_TRANSACTION_ID, tid, thread_trace_id)
 
             pinpoint.add_context(
-                Defines.PP_TRANSCATION_ID, tid, thread_trace_id)
+                Defines.PP_TRANSACTION_ID, tid, thread_trace_id)
 
             pinpoint.add_trace_header(
                 Defines.PP_SERVER_TYPE, Defines.PYTHON, thread_trace_id)