python实现skywalking的trace模块过滤和报警(实例代码)

下面为大家详细讲解如何实现Python的Skywalking Trace模块的过滤和报警，并提供两条示例说明。

什么是Skywalking Trace模块

Skywalking是由Apache基金会发布的一款开源APM（应用程序性能管理）系统，用于帮助我们深入了解和优化分布式系统。Trace模块是Skywalking中的核心模块，用于跨越各种分布式环境，从应用程序级别开始跟踪和监控各种业务流程，并使用分布式跟踪和服务映射技术来收集和分析数据。

如何实现Trace模块的过滤和报警

Skywalking Trace模块的过滤和报警可以通过编写Python插件来实现。下面是一个简单的示例，以帮助你开始编写自己的Trace插件。

步骤1：安装Skywalking Trace插件包

首先，需要在你的Python环境中安装Skywalking Trace插件包。运行以下命令：

pip install skywalking

步骤2：编写Trace插件代码

然后，创建一个新的Python文件，例如filter_plugin.py，并编写以下插件代码：

import logging
from skywalking import plugin, Layer, Component, config
from skywalking.trace import Tag


logger = logging.getLogger(__name__)


class CustomTraceFilterPlugin(plugin.FilterPlugin):
    def __init__(self):
        super().__init__(__name__)

    def install(self):
        pass

    def filter(self, span):
        # Replace the condition with your own filtering logic
        if span.operation_name.startswith("/health-check"):
            logger.warning("Health check endpoint observed.")
            return False

        return True

    def register(self, pipeline):
        pipeline.add_filter(self)


plugin.register(FilterPluginComponent())

该插件会检查跨度实例的操作名称是否以/health-check开头。如果是，它会发出警告并丢弃该跨度。否则，它将继续允许跨度的传播。

步骤3：注册Trace插件

最后，你需要在Skywalking的Trace配置文件中注册你的节点。将以下配置添加到你的配置文件中，并替换filter_plugin.py为你的插件文件名。

agent:
  plugins:
    filter_plugin:
      module: filter_plugin
      component: filter_plugin
      enabled: true
      sitewide: true

示例说明

1. 过滤错误状态码的跨度

下面的示例演示如何在过滤器插件中实现仅记录成功状态码的跨度：

import logging
from skywalking import plugin, Layer, Component, config
from skywalking.trace import Tag, ErrorLog


logger = logging.getLogger(__name__)


class SuccessStatusCodeFilterPlugin(plugin.FilterPlugin):
    def __init__(self):
        super().__init__(__name__)

    def install(self):
        pass

    def filter(self, span):
        http_status_code = span.get_tag("http.status_code")
        if http_status_code and 200 <= int(http_status_code) < 300:
            return True

        span.log(ErrorLog("Unexpected HTTP status code: {}".format(http_status_code)))
        return False

    def register(self, pipeline):
        pipeline.add_filter(self)


plugin.register(SuccessStatusCodeFilterPlugin())

在上面的示例中，我们将插件命名为SuccessStatusCodeFilterPlugin。此插件检查跨度实例的http.status_code标签是否在正确的范围内（200 ~ 299）。如果标签不存在或值不在该范围内，则会在跨度上记录错误日志并拒绝传播。

1. 报警超时的跨度

以下示例演示如何在超时的情况下发出警告：

import logging
from skywalking import plugin, Layer, Component, config
from skywalking.trace import Tag, ErrorLog


logger = logging.getLogger(__name__)


class TimeoutFilterPlugin(plugin.FilterPlugin):
    def __init__(self):
        super().__init__(__name__)

    def install(self):
        pass

    def filter(self, span):
        if span.is_entry and span.detect_error().is_timeout():
            logger.warning("Detected a timeout for span {}".format(span.operation_name))
            return False

        return True

    def register(self, pipeline):
        pipeline.add_filter(self)


plugin.register(TimeoutFilterPlugin())

在这个例子中，我们将插件命名为TimeoutFilterPlugin。它检查跨度是否超时，并在读取或写入超时时发出警告。它还检查是否为请求链的最后一个跨度。如果跨度满足这些条件，则拒绝传递。

这两个示例只是Skywalking Trace插件的开始。您可以编写自己的过滤器和报警器，并将它们添加到Skywalking中，以更好地监控和管理您的应用程序。