support python async api (#694)

* support python async api

* add thread/process support

* [enh] add asyncio plugins

- standalone-py
- py-photonix testapps
- fastapi example
- pinpointpy plugins and testcase

* Update setup-php-sdk action

* add more doc

* fix unittest ci

Author: eeliu

.github/workflows/main.yml

@@ -141,11 +141,12 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - id: setup-php-sdk
-        uses: php/setup-php-sdk@v0.8
+        uses: php/setup-php-sdk@v0.10
         with:
           version: ${{ matrix.php-versions }}
           arch: ${{ matrix.arch }}
           ts: ${{ matrix.ts }}
+          cache: true
       - uses: ilammy/msvc-dev-cmd@v1
         with:
           arch: ${{ matrix.arch }}
@@ -308,4 +309,4 @@ jobs:
           go install google.golang.org/protobuf/cmd/[email protected]
           go install google.golang.org/grpc/cmd/[email protected]
           # export PATH="$PATH:$(go env GOPATH)/bin"
-          cd collector-agent && go mod tidy && go test ./... -v
+          cd collector-agent && go mod tidy && go test ./... -v

DOC/PY/python..excalidraw.png

@@ -154,8 +154,7 @@ int pinpoint_trace_is_root(NodeID);
  * @return int 0 : means oK
  *             -1: exception found, check the log
  */
-DEPRECATED(
-    "use pinpoint_end_trace. if you need no span missing, set pinpoint_set_agent with `timeout_ms`")
+DEPRECATED("use pinpoint_end_trace. if you need all span, set pinpoint_set_agent with `timeout_ms`")
 int pinpoint_force_end_trace(NodeID, int32_t timeout);
 
 /**

common/include/common.h.in

@@ -154,8 +154,7 @@ int pinpoint_trace_is_root(NodeID);
  * @return int 0 : means oK
  *             -1: exception found, check the log
  */
-DEPRECATED(
-    "use pinpoint_end_trace. if you need no span missing, set pinpoint_set_agent with `timeout_ms`")
+DEPRECATED("use pinpoint_end_trace. if you need all span, set pinpoint_set_agent with `timeout_ms`")
 int pinpoint_force_end_trace(NodeID, int32_t timeout);
 
 /**

common/include/pinpoint/common.h

@@ -23,6 +23,20 @@
 from pinpointPy.TraceContext import get_trace_context
 from functools import wraps
 
+import warnings
+import functools
+
+
+def deprecated(reason: str):
+    def decorator(func):
+        @functools.wraps(func)
+        def deprecated_func(*args, **kwargs):
+            warnings.warn(f"{func.__name__} is deprecated. Reason: {reason}",
+                          category=DeprecationWarning, stacklevel=2)
+            return func(*args, **kwargs)
+        return deprecated_func
+    return decorator
+
 
 class Trace:
     def __init__(self, name):
@@ -58,6 +72,9 @@ class PinTrace:
     def __init__(self, name):
         self.name = name
 
+    def setCurrentTraceNodeId(self, traceId):
+        get_trace_context().set_parent_id(traceId)
+
     def onBefore(self, parentId: int, *args, **kwargs):
         traceId = pinpoint.with_trace(parentId)
         get_trace_context().set_parent_id(traceId)
@@ -111,6 +128,32 @@ def getUniqueName(self):
         return self.name
 
 
+class AsyncPinTrace(PinTrace):
+
+    def __call__(self, func):
+        self.func_name = func.__name__
+
+        @wraps(func)
+        async def pinpointTrace(*args, **kwargs):
+            ret = None
+            # avoiding variable missing
+            # use and return
+            sampled, parentId, nArgs, nKwargs = self._isSample(*args, **kwargs)
+            if not sampled:
+                return await func(*nArgs, **nKwargs)
+            traceId, nArgs, nKwargs = self.onBefore(
+                parentId, *nArgs, **nKwargs)
+            try:
+                ret = await func(*nArgs, **nKwargs)
+                return ret
+            except Exception as e:
+                self.onException(traceId, e)
+                raise e
+            finally:
+                self.onEnd(traceId, ret)
+        return pinpointTrace
+
+
 class TraceIdObject:
     def __init__(self, id: int) -> None:
         self.traceId = id
@@ -179,7 +222,7 @@ class PinTransaction(PinTrace):
     def __init__(self, name: str, userGenHeaderCb: GenPinHeader):
         """pinpointPy user entry point
 
-        Example: 
+        Example:
 
             ```
             from pinpointPy.Common import GenPinHeader, PinHeader, PinTransaction
@@ -195,7 +238,7 @@ def run(msg):
             ```
         Args:
             name (str): entry points name(showing pinpoint)
-            userGenHeaderCb (GenPinHeader): This helps getting header from current function 
+            userGenHeaderCb (GenPinHeader): This helps getting header from current function
         """
         super().__init__(name)
         self.name: str = name
@@ -215,12 +258,16 @@ def onBefore(self, parentId: int, *args, **kwargs):
 
         pinpoint.add_trace_header(
             Defines.PP_INTERCEPTOR_NAME, self.name, traceId)
+
         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_REQ_URI, header.Url, traceId)
         pinpoint.add_trace_header(Defines.PP_REQ_SERVER, header.Host, traceId)
@@ -264,3 +311,13 @@ def onEnd(self, traceId, ret):
     def onException(self, traceId, e):
         pinpoint.mark_as_error(str(e), "", 0, traceId)
         raise e
+
+
+def enable_experiment_plugins(async_plugins: bool = True):
+    if async_plugins:
+        from pinpointPy.libs._threading import monkey_patch as thread_patch
+        thread_patch()
+        from pinpointPy.libs._process import monkey_patch as process_patch
+        process_patch()
+        from pinpointPy.libs._asyncio import monkey_patch as asyncio_patch
+        asyncio_patch()

plugins/PY/pinpointPy/Common.py

@@ -16,6 +16,7 @@
 
 
 from pinpointPy import Common, Defines, pinpoint
+import random
 
 
 class PinpointCommonPlugin(Common.PinTrace):
@@ -36,3 +37,92 @@ def onEnd(self, traceId, ret):
     def onException(self, traceId, e):
         pinpoint.add_exception(str(e), traceId)
         raise e
+
+
+class AsyncCommonPlugin(Common.AsyncPinTrace):
+
+    # -> tuple[int, Any, dict[str, Any]]:
+    def onBefore(self, parentId, *args, **kwargs):
+        traceId, args, kwargs = super().onBefore(parentId, *args, **kwargs)
+        pinpoint.add_trace_header(
+            Defines.PP_INTERCEPTOR_NAME, self.getUniqueName(), traceId)
+        pinpoint.add_trace_header(
+            Defines.PP_SERVER_TYPE, Defines.PP_METHOD_CALL, traceId)
+        return traceId, args, kwargs
+
+    def onEnd(self, traceId, ret):
+        super().onEnd(traceId, ret)
+
+    def onException(self, traceId, e):
+        pinpoint.add_exception(str(e), traceId)
+
+
+class HookTargetPlugins(Common.PinTrace):
+    def onBefore(self, parentId: int,  *args, **kwargs):
+
+        traceId, args, kwargs = super().onBefore(parentId, *args, **kwargs)
+        pinpoint.add_trace_header(
+            Defines.PP_INTERCEPTOR_NAME, self.getUniqueName(), traceId)
+        pinpoint.add_trace_header(
+            Defines.PP_SERVER_TYPE, Defines.P_INVOCATION_CALL_TYPE, traceId)
+
+        async_id = random.randint(0, 9999)
+        pinpoint.add_trace_header(
+            Defines.PP_ASYNC_CALL_ID, f'{async_id}', traceId)
+
+        sequence_id = pinpoint.get_sequence_id(traceId)
+
+        tid = pinpoint.get_context(Defines.PP_TRANSCATION_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)
+
+        if 'target' in kwargs:
+            origin_target = kwargs['target']
+
+            def pp_new_entry_func(*args, **kwargs):
+                # start trace
+                thread_trace_id = pinpoint.with_trace(0)
+                self.setCurrentTraceNodeId(thread_trace_id)
+                pinpoint.add_trace_header(
+                    Defines.PP_APP_NAME, app_name, thread_trace_id)
+                pinpoint.add_context(
+                    Defines.PP_APP_NAME, app_name, thread_trace_id)
+
+                pinpoint.add_trace_header(
+                    Defines.PP_APP_ID, app_id, thread_trace_id)
+                pinpoint.add_context(
+                    Defines.PP_APP_ID, app_id, thread_trace_id)
+
+                pinpoint.add_trace_header(
+                    Defines.PP_SPAN_ID, seq_id, thread_trace_id)
+                pinpoint.add_context(
+                    Defines.PP_SPAN_ID, seq_id, thread_trace_id)
+
+                pinpoint.add_trace_header(
+                    Defines.PP_TRANSCATION_ID, tid, thread_trace_id)
+                pinpoint.add_context(
+                    Defines.PP_TRANSCATION_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)
+
+                    call_origin_target(*args, **kwargs)
+
+                pinpoint.end_trace(thread_trace_id)
+
+            kwargs['target'] = pp_new_entry_func
+
+        return traceId, args, kwargs
+
+    def onEnd(self, traceId, ret):
+        super().onEnd(traceId, ret)
+        return ret

plugins/PY/pinpointPy/CommonPlugin.py

@@ -76,6 +76,9 @@
 PP_CELERY = '1702'
 
 PP_REMOTE_METHOD = '9900'
+P_INVOCATION_CALL_TYPE = '100'
+
+PP_ASYNC_CALL_ID = 'asyId'
 
 PP_HTTP_URL = '40'
 PP_HTTP_PARAM = '41'

plugins/PY/pinpointPy/Defines.py

@@ -20,20 +20,24 @@
 from pinpointPy.TraceContext import TraceContext
 from pinpointPy.Common import PinTrace
 
+
 class AsyncTraceContext (TraceContext):
     def get_parent_id(self):
-        id = context.get('_pinpoint_id_', 0)
-        if id == 0:
+        try:
+            id = context.get('_pinpoint_id_', 0)
+            if id == 0:
+                return False, -1
+            else:
+                return True, id
+        except Exception:
             return False, -1
-        else:
-            return True, id
 
     def set_parent_id(self, id: int):
         context['_pinpoint_id_'] = id
 
 
 class AsyncPinTrace(PinTrace):
-    
+
     def __call__(self, func):
         self.func_name = func.__name__
 

plugins/PY/pinpointPy/Fastapi/AsyCommon.py

@@ -17,8 +17,10 @@
 
 from pinpointPy.Fastapi.AsyCommon import AsyncPinTrace
 from pinpointPy import Defines, pinpoint
+from pinpointPy.Common import deprecated
 
 
+@deprecated("please use AsyncCommonPlugin in pinpointPy.CommonPlugin eg: `from pinpointPy.CommonPlugin import AsyncCommonPlugin`")
 class CommonPlugin(AsyncPinTrace):
 
     # -> tuple[int, Any, dict[str, Any]]:

plugins/PY/pinpointPy/Fastapi/AsyCommonPlugin.py

@@ -34,6 +34,7 @@ def onBefore(self, parentId, *args, **kwargs):
         pinpoint.add_trace_header(
             Defines.PP_APP_ID, pinpoint.app_id(), traceId)
         pinpoint.add_context(Defines.PP_APP_NAME, pinpoint.app_name(), traceId)
+        pinpoint.add_context(Defines.PP_APP_ID, pinpoint.app_id(), traceId)
         ###############################################################
         request = args[0]