chore: add changelog for 0.4.0 and minor refactor (#80)

Author: kezhenxu94

CHANGELOG.md

@@ -1,5 +1,12 @@
 ## Change Logs
 
+### 0.4.0
+
+- Feature: Support Kafka reporter protocol (#74)
+- BugFix: Move generated packages into `skywalking` namespace to avoid conflicts (#72)
+- BugFix: Agent cannot reconnect after server is down (#79)
+- Test: Mitigate unsafe yaml loading (#76)
+
 ### 0.3.0
 
 - New plugins

skywalking/plugins/__init__.py

@@ -18,6 +18,8 @@
 import logging
 import pkgutil
 import re
+import traceback
+
 import pkg_resources
 
 from packaging import version
@@ -51,7 +53,13 @@ def install():
         if not hasattr(plugin, 'install') or inspect.ismethod(getattr(plugin, 'install')):
             logger.warning('no `install` method in plugin %s, thus the plugin won\'t be installed', modname)
             continue
-        plugin.install()
+
+        # noinspection PyBroadException
+        try:
+            plugin.install()
+        except Exception:
+            logger.warning('failed to install plugin %s', modname)
+            traceback.print_exc() if logger.isEnabledFor(logging.DEBUG) else None
 
 
 _operators = {

skywalking/plugins/sw_django.py

@@ -31,59 +31,56 @@
 
 
 def install():
-    try:
-        from django.core.handlers.base import BaseHandler
-        from django.core.handlers import exception
-
-        _get_response = BaseHandler.get_response
-        _handle_uncaught_exception = exception.handle_uncaught_exception
-
-        def _sw_get_response(this, request):
-            if request is None:
-                resp = _get_response(this, request)
-                return resp
-
-            context = get_context()
-            carrier = Carrier()
-            for item in carrier:
-                # Any HTTP headers in the request are converted to META keys by converting all characters to uppercase,
-                # replacing any hyphens with underscores and adding an HTTP_ prefix to the name.
-                # https://docs.djangoproject.com/en/3.0/ref/request-response/#django.http.HttpRequest.META
-                sw_http_header_key = 'HTTP_%s' % item.key.upper().replace('-', '_')
-                if sw_http_header_key in request.META:
-                    item.val = request.META[sw_http_header_key]
-
-            with context.new_entry_span(op=request.path, carrier=carrier) as span:
-                span.layer = Layer.Http
-                span.component = Component.Django
-                span.peer = '%s:%s' % (request.META.get('REMOTE_ADDR'), request.META.get('REMOTE_PORT') or "80")
-
-                span.tag(Tag(key=tags.HttpMethod, val=request.method))
-                span.tag(Tag(key=tags.HttpUrl, val=request.build_absolute_uri().split("?")[0]))
-
-                # you can get request parameters by `request.GET` even though client are using POST or other methods
-                if config.django_collect_http_params and request.GET:
-                    span.tag(Tag(key=tags.HttpParams,
-                                 val=params_tostring(request.GET)[0:config.http_params_length_threshold]))
-
-                resp = _get_response(this, request)
-                span.tag(Tag(key=tags.HttpStatus, val=resp.status_code))
-                if resp.status_code >= 400:
-                    span.error_occurred = True
-                return resp
-
-        def _sw_handle_uncaught_exception(request, resolver, exc_info):
-            if exc_info is not None:
-                entry_span = get_context().active_span()
-                if entry_span is not None:
-                    entry_span.raised()
-
-            return _handle_uncaught_exception(request, resolver, exc_info)
-
-        BaseHandler.get_response = _sw_get_response
-        exception.handle_uncaught_exception = _sw_handle_uncaught_exception
-    except Exception:
-        logger.warning('failed to install plugin %s', __name__)
+    from django.core.handlers.base import BaseHandler
+    from django.core.handlers import exception
+
+    _get_response = BaseHandler.get_response
+    _handle_uncaught_exception = exception.handle_uncaught_exception
+
+    def _sw_get_response(this, request):
+        if request is None:
+            resp = _get_response(this, request)
+            return resp
+
+        context = get_context()
+        carrier = Carrier()
+        for item in carrier:
+            # Any HTTP headers in the request are converted to META keys by converting all characters to uppercase,
+            # replacing any hyphens with underscores and adding an HTTP_ prefix to the name.
+            # https://docs.djangoproject.com/en/3.0/ref/request-response/#django.http.HttpRequest.META
+            sw_http_header_key = 'HTTP_%s' % item.key.upper().replace('-', '_')
+            if sw_http_header_key in request.META:
+                item.val = request.META[sw_http_header_key]
+
+        with context.new_entry_span(op=request.path, carrier=carrier) as span:
+            span.layer = Layer.Http
+            span.component = Component.Django
+            span.peer = '%s:%s' % (request.META.get('REMOTE_ADDR'), request.META.get('REMOTE_PORT') or "80")
+
+            span.tag(Tag(key=tags.HttpMethod, val=request.method))
+            span.tag(Tag(key=tags.HttpUrl, val=request.build_absolute_uri().split("?")[0]))
+
+            # you can get request parameters by `request.GET` even though client are using POST or other methods
+            if config.django_collect_http_params and request.GET:
+                span.tag(Tag(key=tags.HttpParams,
+                             val=params_tostring(request.GET)[0:config.http_params_length_threshold]))
+
+            resp = _get_response(this, request)
+            span.tag(Tag(key=tags.HttpStatus, val=resp.status_code))
+            if resp.status_code >= 400:
+                span.error_occurred = True
+            return resp
+
+    def _sw_handle_uncaught_exception(request, resolver, exc_info):
+        if exc_info is not None:
+            entry_span = get_context().active_span()
+            if entry_span is not None:
+                entry_span.raised()
+
+        return _handle_uncaught_exception(request, resolver, exc_info)
+
+    BaseHandler.get_response = _sw_get_response
+    exception.handle_uncaught_exception = _sw_handle_uncaught_exception
 
 
 def params_tostring(params):

skywalking/plugins/sw_elasticsearch.py

@@ -25,30 +25,25 @@
 
 
 def install():
-    # noinspection PyBroadException
-    try:
-        from elasticsearch import Transport
-        _perform_request = Transport.perform_request
-
-        def _sw_perform_request(this: Transport, method, url, headers=None, params=None, body=None):
-            context = get_context()
-            peer = ",".join([host["host"] + ":" + str(host["port"]) for host in this.hosts])
-            with context.new_exit_span(op="Elasticsearch/" + method + url, peer=peer) as span:
-                span.layer = Layer.Database
-                span.component = Component.Elasticsearch
-                try:
-                    res = _perform_request(this, method, url, headers=headers, params=params, body=body)
-
-                    span.tag(Tag(key=tags.DbType, val="Elasticsearch"))
-                    if config.elasticsearch_trace_dsl:
-                        span.tag(Tag(key=tags.DbStatement, val="" if body is None else body))
-
-                except BaseException as e:
-                    span.raised()
-                    raise e
-                return res
-
-        Transport.perform_request = _sw_perform_request
-
-    except Exception:
-        logger.warning('failed to install plugin %s', __name__)
+    from elasticsearch import Transport
+    _perform_request = Transport.perform_request
+
+    def _sw_perform_request(this: Transport, method, url, headers=None, params=None, body=None):
+        context = get_context()
+        peer = ",".join([host["host"] + ":" + str(host["port"]) for host in this.hosts])
+        with context.new_exit_span(op="Elasticsearch/" + method + url, peer=peer) as span:
+            span.layer = Layer.Database
+            span.component = Component.Elasticsearch
+            try:
+                res = _perform_request(this, method, url, headers=headers, params=params, body=body)
+
+                span.tag(Tag(key=tags.DbType, val="Elasticsearch"))
+                if config.elasticsearch_trace_dsl:
+                    span.tag(Tag(key=tags.DbStatement, val="" if body is None else body))
+
+            except BaseException as e:
+                span.raised()
+                raise e
+            return res
+
+    Transport.perform_request = _sw_perform_request

skywalking/plugins/sw_flask.py

@@ -27,52 +27,47 @@
 
 
 def install():
-    # noinspection PyBroadException
-    try:
-        from flask import Flask
-        _full_dispatch_request = Flask.full_dispatch_request
+    from flask import Flask
+    _full_dispatch_request = Flask.full_dispatch_request
 
-        _handle_user_exception = Flask.handle_user_exception
+    _handle_user_exception = Flask.handle_user_exception
 
-        def params_tostring(params):
-            return "\n".join([k + '=[' + ",".join(params.getlist(k)) + ']' for k, _ in params.items()])
+    def params_tostring(params):
+        return "\n".join([k + '=[' + ",".join(params.getlist(k)) + ']' for k, _ in params.items()])
 
-        def _sw_full_dispatch_request(this: Flask):
-            import flask
-            req = flask.request
-            context = get_context()
-            carrier = Carrier()
+    def _sw_full_dispatch_request(this: Flask):
+        import flask
+        req = flask.request
+        context = get_context()
+        carrier = Carrier()
 
-            for item in carrier:
-                if item.key.capitalize() in req.headers:
-                    item.val = req.headers[item.key.capitalize()]
-            with context.new_entry_span(op=req.path, carrier=carrier) as span:
-                span.layer = Layer.Http
-                span.component = Component.Flask
-                span.peer = '%s:%s' % (req.environ["REMOTE_ADDR"], req.environ["REMOTE_PORT"])
-                span.tag(Tag(key=tags.HttpMethod, val=req.method))
-                span.tag(Tag(key=tags.HttpUrl, val=req.url.split("?")[0]))
-                if config.flask_collect_http_params and req.values:
-                    span.tag(Tag(key=tags.HttpParams,
+        for item in carrier:
+            if item.key.capitalize() in req.headers:
+                item.val = req.headers[item.key.capitalize()]
+        with context.new_entry_span(op=req.path, carrier=carrier) as span:
+            span.layer = Layer.Http
+            span.component = Component.Flask
+            span.peer = '%s:%s' % (req.environ["REMOTE_ADDR"], req.environ["REMOTE_PORT"])
+            span.tag(Tag(key=tags.HttpMethod, val=req.method))
+            span.tag(Tag(key=tags.HttpUrl, val=req.url.split("?")[0]))
+            if config.flask_collect_http_params and req.values:
+                span.tag(Tag(key=tags.HttpParams,
                              val=params_tostring(req.values)[0:config.http_params_length_threshold]))
-                resp = _full_dispatch_request(this)
+            resp = _full_dispatch_request(this)
 
-                if resp.status_code >= 400:
-                    span.error_occurred = True
+            if resp.status_code >= 400:
+                span.error_occurred = True
 
-                span.tag(Tag(key=tags.HttpStatus, val=resp.status_code))
-                return resp
+            span.tag(Tag(key=tags.HttpStatus, val=resp.status_code))
+            return resp
 
-        def _sw_handle_user_exception(this: Flask, e):
-            if e is not None:
-                entry_span = get_context().active_span()
-                if entry_span is not None and type(entry_span) is not NoopSpan:
-                    entry_span.raised()
+    def _sw_handle_user_exception(this: Flask, e):
+        if e is not None:
+            entry_span = get_context().active_span()
+            if entry_span is not None and type(entry_span) is not NoopSpan:
+                entry_span.raised()
 
-            return _handle_user_exception(this, e)
+        return _handle_user_exception(this, e)
 
-        Flask.full_dispatch_request = _sw_full_dispatch_request
-        Flask.handle_user_exception = _sw_handle_user_exception
-
-    except Exception:
-        logger.warning('failed to install plugin %s', __name__)
+    Flask.full_dispatch_request = _sw_full_dispatch_request
+    Flask.handle_user_exception = _sw_handle_user_exception

skywalking/plugins/sw_http_server.py

@@ -16,7 +16,6 @@
 #
 import inspect
 import logging
-import traceback
 
 from skywalking import Layer, Component
 from skywalking.trace import tags
@@ -28,25 +27,19 @@
 
 
 def install():
-    # noinspection PyBroadException
-    try:
-        from http.server import BaseHTTPRequestHandler
+    from http.server import BaseHTTPRequestHandler
 
-        _handle = BaseHTTPRequestHandler.handle
+    _handle = BaseHTTPRequestHandler.handle
 
-        def _sw_handle(handler: BaseHTTPRequestHandler):
-            clazz = handler.__class__
-            if 'werkzeug.serving.WSGIRequestHandler' == ".".join([clazz.__module__, clazz.__name__]):
-                wrap_werkzeug_request_handler(handler)
-            else:
-                wrap_default_request_handler(handler)
-            _handle(handler)
+    def _sw_handle(handler: BaseHTTPRequestHandler):
+        clazz = handler.__class__
+        if 'werkzeug.serving.WSGIRequestHandler' == ".".join([clazz.__module__, clazz.__name__]):
+            wrap_werkzeug_request_handler(handler)
+        else:
+            wrap_default_request_handler(handler)
+        _handle(handler)
 
-        BaseHTTPRequestHandler.handle = _sw_handle
-
-    except Exception:
-        logger.warning('failed to install plugin %s', __name__)
-        traceback.print_exc()
+    BaseHTTPRequestHandler.handle = _sw_handle
 
 
 def wrap_werkzeug_request_handler(handler):

skywalking/plugins/sw_kafka.py

@@ -27,18 +27,13 @@
 
 
 def install():
-    # noinspection PyBroadException
-    try:
-        from kafka import KafkaProducer
-        from kafka import KafkaConsumer
-
-        _send = KafkaProducer.send
-        __poll_once = KafkaConsumer._poll_once
-        KafkaProducer.send = _sw_send_func(_send)
-        KafkaConsumer._poll_once = _sw__poll_once_func(__poll_once)
-
-    except Exception:
-        logger.warning('failed to install plugin %s', __name__)
+    from kafka import KafkaProducer
+    from kafka import KafkaConsumer
+
+    _send = KafkaProducer.send
+    __poll_once = KafkaConsumer._poll_once
+    KafkaProducer.send = _sw_send_func(_send)
+    KafkaConsumer._poll_once = _sw__poll_once_func(__poll_once)
 
 
 def _sw__poll_once_func(__poll_once):