【导读】本文介绍了分布式开源分布式tracing框架Skywalking的使用。

简介

SkyWalking 是一个开源的 APM 系统，包括对 Cloud Native 架构中分布式系统的监控、跟踪、诊断能力。核心功能如下。

• 服务、服务实例、端点指标分析
• 根本原因分析。在运行时分析代码
• 服务拓扑图分析
• 服务、服务实例和端点依赖分析
• 检测到缓慢的服务和端点
• 性能优化
• 分布式跟踪和上下文传播
• 数据库访问指标。检测慢速数据库访问语句（包括 SQL 语句）
• 警报
• 浏览器性能监控
• 基础设施（VM、网络、磁盘等）监控
• 跨指标、跟踪和日志的协作

skywalking-arch

快速安装 skywalking

skywalking 支持多种安装方式，二进制安装，docker 安装以及 helm、k8s 安装等。

这里直接使用 docker-compose 进行快速部署测试

docker-compose.yaml

version: '3.3'
services:
  # storage
  elasticsearch:
    image: elasticsearch:6.8.16
    container_name: elasticsearch
    restart: always
    ports:
      - 9200:9200
    environment:
      discovery.type: single-node
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes: 
      - ./elasticsearch/logs:/usr/share/elasticsearch/logs
      - ./elasticsearch/data:/usr/share/elasticsearch/data
      - /etc/localtime:/etc/localtime
  # server
  oap:
    image: apache/skywalking-oap-server:8.6.0-es6
    container_name: oap
    depends_on:
      - elasticsearch
    links:
      - elasticsearch
    restart: always
    ports:
      - 11800:11800
      - 12800:12800
    environment:
      SW_STORAGE: elasticsearch # 默认为 es6，es7 为 elasticsearch7
      SW_STORAGE_ES_CLUSTER_NODES: elasticsearch:9200
    volumes: 
      - /etc/localtime:/etc/localtime
  # dashboard
  ui:
    image: apache/skywalking-ui:8.6.0
    container_name: ui
    depends_on:
      - oap
    links:
      - oap
    restart: always
    ports:
      - 8080:8080
    environment:
      SW_OAP_ADDRESS: oap:12800
    volumes: 
      - /etc/localtime:/etc/localtime

运行

1docker-compose up -d

访问控制台

浏览器访问 http://localhost:8080

gin 接入 skywalking

这里使用第三方库 go2sky 以及 go2sky 的 gin 中间件

1go get -u github.com/SkyAPM/go2sky
2go get github.com/SkyAPM/go2sky-plugins/gin/v3

代码示例

两个 gin 服务 demo-server1 和 demo-server2,demo-server1 调用 demo-server2 demo-server2 提供 POST /user/info 接口 demo-server1 提供 GET /tracer 接口

demo-server2 代码如下

package main

import (
  "fmt"
  "time"

  "github.com/SkyAPM/go2sky"
  "github.com/SkyAPM/go2sky/reporter"
  "github.com/gin-gonic/gin"

  v3 "github.com/SkyAPM/go2sky-plugins/gin/v3"
)

const (
  serverName = "demo-server2"
  serverPort = 8082
)

var skyAddr = "localhost:11800"

type Params struct {
  Name string
}

func panicErr(err error) {
  if err != nil {
    panic(err)
  }
}

func main() {
  r := gin.Default()
  // skyAddr 是 skywaling 的 grpc 地址，默认是 localhost:11800， 默认心跳检测时间是 1s
  rp, err := reporter.NewGRPCReporter(skyAddr, reporter.WithCheckInterval(5*time.Second))
  panicErr(err)
  // 初始化一个 tracer，一个服务只需要一个 tracer，其含义是这个服务名称
  tracer, err := go2sky.NewTracer(serverName, go2sky.WithReporter(rp))
  panicErr(err)
  // gin 使用 sky 自带的 middleware
  r.Use(v3.Middleware(r, tracer))

  // 自定义一个接口
  r.POST("/user/info", func(context *gin.Context) {
    // LocalSpan 可以理解为本地日志的 tracer，一般用户当前应用
    span, ctx, err := tracer.CreateLocalSpan(context.Request.Context())
    panicErr(err)
    // 每一个 span 都有一个名字去标实操作的名称！
    span.SetOperationName("UserInfo")
    // 记住重新设置一个 ctx，再其次这个 ctx 不是 gin 的 ctx，而是 http request 的 ctx
    context.Request = context.Request.WithContext(ctx)

    params := new(Params)
    err = context.BindJSON(params)
    panicErr(err)
    // 记录日志信息
    span.Log(time.Now(), "[UserInfo]", fmt.Sprintf(serverName+" satrt, req : %+v", params))
    local := gin.H{
      "msg": fmt.Sprintf(serverName+" time : %s", time.Now().Format("15:04:05.9999")),
    }
    context.JSON(200, local)
    span.Log(time.Now(), "[UserInfo]", fmt.Sprintf(serverName+" end, resp : %s", local))
    // 切记最后要设置 span - end，不然就是一个非闭环的
    span.End()
  })

  r.Run(fmt.Sprintf(":%d", serverPort))
}

demo-server1 代码如下

package main

import (
  "bytes"
  "encoding/json"
  "fmt"
  "io/ioutil"
  "log"
  "net/http"
  "time"

  "github.com/SkyAPM/go2sky"
  "github.com/SkyAPM/go2sky/reporter"
  "github.com/gin-gonic/gin"

  v3 "github.com/SkyAPM/go2sky-plugins/gin/v3"
  agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
)

const (
  serverName       = "demo-server1"
  serverPort       = 8081
  remoteServerName = "demo-server2"
  remoteServerAddr = "localhost:8082"
  remotePath       = "/user/info"
)

var skyAddr = "localhost:11800"

func panicErr(err error) {
  if err != nil {
    log.Fatal(err.Error())
  }
}

type Params struct {
  Name string
}

var tracer *go2sky.Tracer

func skyMiddleware(r *gin.Engine) {
  var err error
  rp, err := reporter.NewGRPCReporter(skyAddr, reporter.WithCheckInterval(5*time.Second))
  panicErr(err)
  tracer, err = go2sky.NewTracer(serverName, go2sky.WithReporter(rp))
  panicErr(err)
  r.Use(v3.Middleware(r, tracer))
}

func trace(context *gin.Context) {
  span, ctx, err := tracer.CreateLocalSpan(context.Request.Context())
  panicErr(err)
  span.SetOperationName("Trace")

  context.Request = context.Request.WithContext(ctx)
  span.Log(time.Now(), "[Trace]", fmt.Sprintf(serverName+" satrt, params : %s", time.Now().Format("15:04:05.9999")))

  result := make([]map[string]interface{}, 0)

  //1、请求一次
  {
    url := fmt.Sprintf("http://%s%s", remoteServerAddr, remotePath)

    params := Params{
      Name: serverName + time.Now().Format("15:04:05.9999"),
    }
    buffer := &bytes.Buffer{}
    _ = json.NewEncoder(buffer).Encode(params)
    req, err := http.NewRequest(http.MethodPost, url, buffer)
    panicErr(err)

    // op_name 是每一个操作的名称
    reqSpan, err := tracer.CreateExitSpan(context.Request.Context(), "invoke - "+remoteServerName, "localhost:8082/user/info", func(headerKey, headerValue string) error {
      req.Header.Set(headerKey, headerValue)
      return nil
    })
    panicErr(err)
    reqSpan.SetComponent(2)
    reqSpan.SetSpanLayer(agentv3.SpanLayer_RPCFramework) // rpc 调用

    resp, err := http.DefaultClient.Do(req)
    panicErr(err)
    defer resp.Body.Close()

    reqSpan.Log(time.Now(), "[HttpRequest]", fmt.Sprintf("开始请求，请求服务：%s, 请求地址：%s, 请求参数：%+v", remoteServerName, url, params))
    body, err := ioutil.ReadAll(resp.Body)
    panicErr(err)
    fmt.Printf("接受到消息： %s\n", body)
    reqSpan.Tag(go2sky.TagHTTPMethod, http.MethodPost)
    reqSpan.Tag(go2sky.TagURL, url)
    reqSpan.Log(time.Now(), "[HttpRequest]", fmt.Sprintf("结束请求，响应结果：%s", body))
    reqSpan.End()
    res := map[string]interface{}{}
    err = json.Unmarshal(body, &res)
    panicErr(err)
    result = append(result, res)
  }

  //2 、再请求一次
  {
    url := fmt.Sprintf("http://%s%s", remoteServerAddr, remotePath)

    params := Params{
      Name: serverName + time.Now().Format("15:04:05.9999"),
    }
    buffer := &bytes.Buffer{}
    _ = json.NewEncoder(buffer).Encode(params)
    req, err := http.NewRequest(http.MethodPost, url, buffer)
    panicErr(err)

    // 出去必须用这个携带 header
    reqSpan, err := tracer.CreateExitSpan(context.Request.Context(), "invoke - "+remoteServerName, "localhost:8082/user/info", func(headerKey, headerValue string) error {
      req.Header.Set(headerKey, headerValue)
      return nil
    })
    panicErr(err)
    reqSpan.SetComponent(2)
    reqSpan.SetSpanLayer(agentv3.SpanLayer_RPCFramework) // rpc 调用

    resp, err := http.DefaultClient.Do(req)
    panicErr(err)
    defer resp.Body.Close()

    reqSpan.Log(time.Now(), "[HttpRequest]", fmt.Sprintf("开始请求，请求服务：%s, 请求地址：%s, 请求参数：%+v", remoteServerName, url, params))
    body, err := ioutil.ReadAll(resp.Body)
    panicErr(err)
    fmt.Printf("接受到消息： %s\n", body)

    reqSpan.Tag(go2sky.TagHTTPMethod, http.MethodPost)
    reqSpan.Tag(go2sky.TagURL, url)
    reqSpan.Log(time.Now(), "[HttpRequest]", fmt.Sprintf("结束请求，响应结果：%s", body))
    reqSpan.End()
    res := map[string]interface{}{}
    err = json.Unmarshal(body, &res)
    panicErr(err)
    result = append(result, res)
  }

  // 设置响应结果
  local := gin.H{
    "msg": result,
  }
  context.JSON(200, local)
  span.Log(time.Now(), "[Trace]", fmt.Sprintf(serverName+" end, resp : %s", local))
  span.End()
  {
    span, ctx, err := tracer.CreateEntrySpan(context.Request.Context(), "Send", func(s string) (string, error) {
      return "", nil
    })
    context.Request = context.Request.WithContext(ctx)
    panicErr(err)
    span.SetOperationName("Send")
    span.Log(time.Now(), "[Info]", "send resp")
    span.End()
  }
}

func main() {

  // 这些都一样
  r := gin.Default()
  // 使用 go2sky gin 中间件
  skyMiddleware(r)

  // 调用接口
  r.GET("/trace", trace)

  r.Run(fmt.Sprintf(":%d", serverPort))
}

调用测试

1for i in $(seq 100); do curl -s http://localhost:8081/trace >/dev/null; sleep 0.1 ; done

demo-server2

demo-server1

dashboard

拓扑

追踪

总结

1. gin 接入 skywalking 总体还是比较简单的，正式使用的时候还需要再次封装，简化操作；
2. 应用程序和 skywalking 之前不是强依赖关系，skywalking 关闭的情况下能够正常响应请求。

来源：

huangzhongde.cn/post/Golang/%E4%BD%BF%E7%94%A8skywalking%E7%9B%91%E6%8E%A7gin_web%E6%9C%8D%E5%8A%A1/