New Plugin

from skywalking import Layer, Component, config
from skywalking.trace.context import get_context, NoopContext
from skywalking.trace.span import NoopSpan
from skywalking.trace.tags import TagHttpMethod, TagHttpURL, TagHttpStatusCode

link_vector = ['<documentation URL>']
support_matrix = {
    '<pip-package-name>': {
        '>=3.13': ['<major>.*'],  # use .* wildcard for latest patch (e.g., '4.*')
        '>=3.10': ['<older_minor>.*', '<major>.*'],
    }
}
note = """"""

def install():
    from <library> import <Class>
    _original = <Class>.<method>

    def _sw_method(this, *args, **kwargs):
        # Parse URL/peer from args
        span = NoopSpan(NoopContext()) if config.ignore_http_method_check(method) \
            else get_context().new_exit_span(op=path, peer=netloc, component=Component.<Name>)

        with span:
            carrier = span.inject()
            span.layer = Layer.Http
            # Inject carrier into outgoing headers
            for item in carrier:
                headers[item.key] = item.val
            span.tag(TagHttpMethod(method))
            span.tag(TagHttpURL(url))
            res = _original(this, *args, **kwargs)
            span.tag(TagHttpStatusCode(res.status_code))
            if res.status_code >= 400:
                span.error_occurred = True
            return res

    <Class>.<method> = _sw_method

from skywalking import Layer, Component, config
from skywalking.trace.carrier import Carrier
from skywalking.trace.context import get_context, NoopContext
from skywalking.trace.span import NoopSpan
from skywalking.trace.tags import TagHttpMethod, TagHttpURL, TagHttpStatusCode

def install():
    from <framework> import <App>
    _original = <App>.<handler_method>

    def _sw_handler(this, *args, **kwargs):
        req = <get_request>
        carrier = Carrier()
        for item in carrier:
            if item.key.capitalize() in req.headers:
                item.val = req.headers[item.key.capitalize()]

        span = NoopSpan(NoopContext()) if config.ignore_http_method_check(req.method) \
            else get_context().new_entry_span(op=req.path, carrier=carrier, inherit=Component.General)

        with span:
            span.layer = Layer.Http
            span.component = Component.<Name>
            span.peer = f"{remote_addr}:{remote_port}"
            span.tag(TagHttpMethod(req.method))
            span.tag(TagHttpURL(req.url))
            resp = _original(this, *args, **kwargs)
            span.tag(TagHttpStatusCode(resp.status_code))
            if resp.status_code >= 400:
                span.error_occurred = True
            return resp

    <App>.<handler_method> = _sw_handler

from skywalking import Layer, Component
from skywalking.trace.context import get_context
from skywalking.trace.tags import TagDbType, TagDbInstance, TagDbStatement  # or TagCacheType, TagCacheOp, etc.

def install():
    from <library> import <Connection>
    _original = <Connection>.<method>

    def _sw_method(this, *args, **kwargs):
        peer = f'{this.host}:{this.port}'
        with get_context().new_exit_span(op='<DB>/<operation>', peer=peer, component=Component.<Name>) as span:
            span.layer = Layer.Database  # or Layer.Cache
            span.tag(TagDbType('<DB type>'))
            span.tag(TagDbStatement(query))
            res = _original(this, *args, **kwargs)
            return res

    <Connection>.<method> = _sw_method

NewLib = 7020  # Next available ID

from typing import Callable
import pytest
import requests
from skywalking.plugins.sw_<name> import support_matrix
from tests.orchestrator import get_test_vector
from tests.plugin.base import TestPluginBase

@pytest.fixture
def prepare():
    return lambda *_: requests.get('http://0.0.0.0:9090/<endpoint>', timeout=5)

class TestPlugin(TestPluginBase):
    @pytest.mark.parametrize('version', get_test_vector(lib_name='<name>', support_matrix=support_matrix))
    def test_plugin(self, docker_compose, version):
        self.validate()

version: '2.1'

services:
  collector:
    extends:
      service: collector
      file: ../../docker-compose.base.yml

  # Add external service if needed (database, cache, MQ broker)
  # <service_name>:
  #   image: <image>
  #   ports: [<port>:<port>]
  #   healthcheck: ...
  #   networks: [beyond]

  provider:
    extends:
      service: agent
      file: ../../docker-compose.base.yml
    ports:
      - 9091:9091
    volumes:
      - .:/app
    command: ['bash', '-c', 'pip install flask && pip install -r /app/requirements.txt && sw-python run python3 /app/services/provider.py']
    depends_on:
      collector:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/9091"]
      interval: 5s
      timeout: 60s
      retries: 120
    environment:
      SW_AGENT_NAME: provider
      SW_AGENT_LOGGING_LEVEL: DEBUG

  consumer:
    extends:
      service: agent
      file: ../../docker-compose.base.yml
    ports:
      - 9090:9090
    volumes:
      - .:/app
    command: ['bash', '-c', 'pip install flask && pip install -r /app/requirements.txt && sw-python run python3 /app/services/consumer.py']
    depends_on:
      collector:
        condition: service_healthy
      provider:
        condition: service_healthy
    environment:
      SW_AGENT_NAME: consumer
      SW_AGENT_LOGGING_LEVEL: DEBUG

networks:
  beyond: