ci(py): py plugins (#555)

- add--exit-code-from python-plugins
- bug from _urlib
- fix get_logger

.github/workflows/main.yml

@@ -120,13 +120,18 @@ jobs:
       - uses: actions/checkout@v2
         with:
           submodules: 'true'
+      - name: update docker compose
+        run: |
+          wget "https://github.com/docker/compose/releases/latest/download/docker-compose-linux-x86_64"
+          chmod +x docker-compose-linux-x86_64
+          ./docker-compose-linux-x86_64 version
       # - name: build python-plugins
         # run: docker-compose  -f "testapps/compose.yaml" build python-plugins
       - name: start test environment
-        run: docker-compose -f "testapps/compose.yaml" up python-plugins
+        run: ./docker-compose-linux-x86_64 -f "testapps/compose.yaml" up python-plugins --exit-code-from python-plugins
       - name: Stop containers
-        if: always()
-        run: docker-compose -f "testapps/compose.yaml" down  
+        # if: always()
+        run: ./docker-compose-linux-x86_64 -f "testapps/compose.yaml" down python-plugins
   Collector-agent:
     strategy:
       matrix:

plugins/PY/pinpointPy/Fastapi/__init__.py

@@ -33,9 +33,9 @@ def __monkey_patch(*args, **kwargs):
             module = importlib.import_module('pinpointPy.Fastapi.' + key)
             monkey_patch = getattr(module, 'monkey_patch')
             if callable(monkey_patch):
-                monkey_patch()
                 get_logger().debug(
                     f"try to install pinpointPy.Fastapi.{key} module")
+                monkey_patch()
 
 
 def async_monkey_patch_for_pinpoint(AioRedis=True, MotorMongo=True, httpx=True):

plugins/PY/pinpointPy/Helper.py

@@ -40,7 +40,7 @@ def generatePinpointHeader(host, headers, traceId=-1):
     nextSeqId = pinpoint.gen_sid()
     pinpoint.add_context(Defines.PP_NEXT_SPAN_ID, nextSeqId, traceId)
     headers[Defines.PP_HEADER_PINPOINT_SPANID] = nextSeqId
-    get_logger().debug(f'headers:{headers}')
+    get_logger().debug(f'append PinpointHeader header:{headers}')
 
 
 def startPinpointByEnviron(environ, trace_id: int):

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

@@ -22,15 +22,11 @@
 def monkey_patch():
     try:
         from mysql.connector.cursor import MySQLCursor, MySQLCursorPrepared
-        from mysql.connector.cursor_cext import CMySQLCursor, CMySQLCursorPrepared
         from .MysqlPlugin import MysqlPlugin
-        from .CMysqlPlugin import CMysqlPlugin
 
         Interceptors = [
             Interceptor(MySQLCursor, 'execute', MysqlPlugin),
             Interceptor(MySQLCursorPrepared, 'execute', MysqlPlugin),
-            Interceptor(CMySQLCursor, 'execute', CMysqlPlugin),
-            Interceptor(CMySQLCursorPrepared, 'execute', CMysqlPlugin),
         ]
         for interceptor in Interceptors:
             interceptor.enable()
@@ -41,6 +37,22 @@ def monkey_patch():
         if 'unittest' in sys.modules.keys():
             raise e
 
+    try:
+        from mysql.connector.cursor_cext import CMySQLCursor, CMySQLCursorPrepared
+        from .CMysqlPlugin import CMysqlPlugin
+
+        Interceptors = [
+            Interceptor(CMySQLCursor, 'execute', CMysqlPlugin),
+            Interceptor(CMySQLCursorPrepared, 'execute', CMysqlPlugin),
+        ]
+        for interceptor in Interceptors:
+            interceptor.enable()
+    except ImportError as e:
+        get_logger().warning(f'import _mysql_connector {e}')
+        import sys
+        if 'unittest' in sys.modules.keys():
+            raise e
+
 
 __all__ = ['monkey_patch']
 __version__ = '0.0.3'

plugins/PY/pinpointPy/libs/_urllib/RequestPlugin.py

@@ -24,6 +24,16 @@
 
 class RequestPlugin(Common.PinTrace):
 
+    @staticmethod
+    def get_url(*args, **kwargs):
+        url = "/"
+        if 'url' in kwargs:
+            url = kwargs['url']
+        else:
+            if len(args) >= 2:
+                url = args[1]
+        return url
+
     def __init__(self, name):
         super().__init__(name)
 
@@ -41,7 +51,7 @@ def isSample(*args, **kwargs):
             return False, parentId, args, kwargs
         # pull out headers
         if sampled:
-            url = args[1]
+            url = RequestPlugin.get_url(*args, **kwargs)
             target = urlparse(url).netloc
             if pinpoint.get_context(Defines.PP_HEADER_PINPOINT_SAMPLED, parentId) == "s1":
                 Helper.generatePinpointHeader(
@@ -60,7 +70,7 @@ def isSample(*args, **kwargs):
 
     def onBefore(self, parentId, *args, **kwargs):
         traceId, args, kwargs = super().onBefore(parentId, *args, **kwargs)
-        url = args[1]
+        url = RequestPlugin.get_url(*args, **kwargs)
         target = urlparse(url).netloc
 
         ###############################################################

plugins/PY/pinpointPy/pinpoint.py

@@ -28,7 +28,7 @@
 __logger__ = None
 
 
-def get_logger(level=logging.INFO) -> logging.Logger:
+def get_logger() -> logging.Logger:
     global __logger__
     if __logger__:
         return __logger__
@@ -42,22 +42,21 @@ def get_logger(level=logging.INFO) -> logging.Logger:
         file_handler = logging.FileHandler(filepath)
         print(filepath)
         file_handler.setFormatter(formatter)
-        file_handler.setLevel(logging.DEBUG)
+        file_handler.setLevel(level=logging.DEBUG)
         logger.addHandler(file_handler)
-    elif level == logging.DEBUG:
-        ch = logging.StreamHandler(sys.stdout)
-        ch.setLevel(logging.DEBUG)
-        ch.setFormatter(formatter)
-        logger.addHandler(ch)
     else:
         ch = logging.StreamHandler(sys.stdout)
-        ch.setLevel(logging.INFO)
+        ch.setLevel(logging.DEBUG)
         ch.setFormatter(formatter)
         logger.addHandler(ch)
     __logger__ = logger
     return __logger__
 
 
+def _set_logger_level(level=logging.INFO):
+    get_logger().setLevel(level)
+
+
 def app_id():
     global __app_id
     return __app_id
@@ -122,13 +121,10 @@ def set_agent(app_id_str: str, app_name_str: str, collect_agent_host: str,  trac
     __app_id = app_id_str
     __app_name = app_name_str
     _pinpointPy.set_agent(collect_agent_host, trace_limit)
-    get_logger().setLevel(log_level)
     if log_level == logging.DEBUG:
         def debug_func(msg: str):
             get_logger().debug(msg=msg)
         _pinpointPy.enable_debug(debug_func)
-    global __logger__
-    __logger__ = None
-    get_logger(log_level)
+    _set_logger_level(log_level)
     get_logger().debug(
         f"appid:{app_id_str} appname:{app_name_str} collector_agent:{collect_agent_host} trace_limit:{trace_limit} log_level:{log_level}")

plugins/PY/requirements.txt

@@ -2,7 +2,7 @@ bottle==0.12.25
 fastapi==0.104.1
 flask==3.0.0
 httpx==0.25.1
-mysql-connector-python==8.2.0
+mysql-connector-python==8.0.31
 pymongo==4.6.0
 PyMySQL==1.1.0
 redis==5.0.1

testapps/backend/main.py

@@ -1,7 +1,7 @@
 from fastapi import FastAPI, Request, Response, Depends, HTTPException
 from starlette.middleware import Middleware
-from pinpointPy.Fastapi import PinPointMiddleWare, async_monkey_patch_for_pinpoint
-from pinpointPy import set_agent
+from pinpointPy.Fastapi import PinPointMiddleWare, async_monkey_patch_for_pinpoint, use_starlette_context
+from pinpointPy import set_agent, monkey_patch_for_pinpoint
 from sqlalchemy.orm import Session
 from starlette_context.middleware import ContextMiddleware
 import aioredis
@@ -33,9 +33,11 @@ async def dispatch(self, request: Request, call_next):
     Middleware(UserMiddleWare)
 ]
 
+use_starlette_context()
+monkey_patch_for_pinpoint()
 async_monkey_patch_for_pinpoint()
 
-set_agent("cd.dev.test.py.backend", "cd.dev.test.py.backend", 'tcp:dev-collector:10000')
+set_agent("cd.dev.test.py.backends", "cd.dev.test.backends", 'tcp:dev-collector:10000')
 
 
 @asynccontextmanager
@@ -67,16 +69,48 @@ async def db_session_middleware(request: Request, call_next):
     return response
 
 
-def get_db(request: Request):
-    return request.state.db
-
-
 @app.get("/")
 async def root():
     return {"message": "Hello World"}
 
 
-@app.get("/test-redis/set/{uid}", tags=["redis"])
+@app.get("/test-requests", tags=['sync-libraries'])
+async def test_requests(request: Request, url='http://www.example.com'):
+    import requests
+    x = requests.get(url)
+    return {"response": x.status_code}
+
+
+@app.get("/test-mysql-connector", tags=['sync-libraries'])
+async def test_mysql_connector(request: Request):
+    import datetime
+    import mysql.connector
+
+    cnx = mysql.connector.connect(
+        user='root',
+        password='password',
+        host='dev-mysql',
+        database='employees')
+    cursor = cnx.cursor()
+
+    query = ("SELECT first_name, last_name, hire_date FROM employees "
+             "WHERE hire_date BETWEEN %s AND %s")
+
+    hire_start = datetime.date(1999, 1, 1)
+    hire_end = datetime.date(1999, 12, 31)
+
+    cursor.execute(query, (hire_start, hire_end))
+
+    for (first_name, last_name, hire_date) in cursor:
+        print("{}, {} was hired on {:%d %b %Y}".format(
+            last_name, first_name, hire_date))
+
+    cursor.close()
+    cnx.close()
+    return {"response": 200}
+
+
+@app.get("/test-redis/set/{uid}", tags=["aioredis"])
 async def test_redis(uid: str = "default"):
     await redis.set(uid, "50fdf310-7d3b-11ee-b962-0242ac120002", ex=1)
     in_value = await redis.get(uid)
@@ -91,45 +125,53 @@ async def test_httpx(request: Request, url='http://www.example.com/'):
     return {"response": response.status_code}
 
 
+@app.get("/httpx/backend", tags=["httpx"])
+async def test_httpx_backend(request: Request, url='http://backend:8000/'):
+    requests_client = request.app.requests_client
+    print(request.headers)
+    response = await requests_client.get(url)
+    return {"response": response.status_code}
+
+
 @app.get("/httpx-self/", tags=["httpx"])
-async def test_httpx(request: Request):
+async def test_httpx_self(request: Request):
     requests_client = request.app.requests_client
     response = await requests_client.get('http://127.0.0.1:8000/httpx/example')
     return {"response": response.status_code}
 
 # thanks guide from https://fastapi.tiangolo.com/tutorial/sql-databases/
 
 
-@app.post("/users/", response_model=schemas.User, tags=["mysql"])
+@app.post("/users/", response_model=schemas.User, tags=["sqlalchemy"])
 def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
     db_user = crud.get_user_by_email(db, email=user.email)
     if db_user:
         raise HTTPException(status_code=400, detail="Email already registered")
     return crud.create_user(db=db, user=user)
 
 
-@app.get("/users/", response_model=List[schemas.User], tags=["mysql"])
+@app.get("/users/", response_model=List[schemas.User], tags=["sqlalchemy"])
 def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
     users = crud.get_users(db, skip=skip, limit=limit)
     return users
 
 
-@app.get("/users/{user_id}", response_model=schemas.User, tags=["mysql"])
+@app.get("/users/{user_id}", response_model=schemas.User, tags=["sqlalchemy"])
 def read_user(user_id: int, db: Session = Depends(get_db)):
     db_user = crud.get_user(db, user_id=user_id)
     if db_user is None:
         raise HTTPException(status_code=404, detail="User not found")
     return db_user
 
 
-@app.post("/users/{user_id}/items/", response_model=schemas.Item, tags=["mysql"])
+@app.post("/users/{user_id}/items/", response_model=schemas.Item, tags=["sqlalchemy"])
 def create_item_for_user(
     user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
 ):
     return crud.create_user_item(db=db, item=item, user_id=user_id)
 
 
-@app.get("/items/", response_model=List[schemas.Item], tags=["mysql"])
+@app.get("/items/", response_model=List[schemas.Item], tags=["sqlalchemy"])
 def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
     items = crud.get_items(db, skip=skip, limit=limit)
     return items

testapps/backend/requirements.txt

@@ -6,4 +6,6 @@ SQLAlchemy==2.0.23
 starlette
 starlette_context
 uvicorn
-pymysql
+pymysql
+requests
+mysql-connector-python==8.0.31

testapps/fastapi/main.py

@@ -69,10 +69,6 @@ async def db_session_middleware(request: Request, call_next):
     return response
 
 
-def get_db(request: Request):
-    return request.state.db
-
-
 @app.get("/")
 async def root():
     return {"message": "Hello World"}

testapps/fastapi/requirements.txt

@@ -6,4 +6,6 @@ SQLAlchemy==2.0.23
 starlette
 starlette_context
 uvicorn
-pymysql
+pymysql
+requests
+mysql-connector-python==8.0.31