微服务troubleshooting利器——调用链

发表于2018-02-08由daniel

微服务troubleshooting利器——调用链

Docker、k8s以及「微服务」是过去两年（2016~2017）对互联网系统架构影响最深远的技术和理念。自己非常幸运，全称参与了公司新架构的设计、研发与迁移，目前公司整个架构已经完成了微服务1.0的建设，正在朝着微服务2.0的方向继续前进（关于微服务1.0与2.0的关系和差异，后面有空了单独在写）。期间有非常多有意思的事情值得与大家分享。今天先聊一聊微服务中的调用链系统。

微服务的喜与悲

从以前的巨服务完成微服务化后，服务的数量大幅增加，系统进化为完全的分布式系统。对运维、上线的自动化的要求自不必说，对研发troubleshooting也有了更高的要求。troubleshooting面临的挑战主要体现在：

服务日志分散化。以前在一个巨服务里面大多时候可以顺利的将一次调用的所有日志查完，但是现在日志分散在各个服务之中。
服务之间的日志需要某种手段进行关联查询。
由于日志量巨大，直接从原始日志进行关联查询没有普适性。因此，需要对这种关联数据进行索引单独存储。

这是一个有意思的挑战。更有意思的是Google早在2010年发表的论文Dapper, a Large-Scale Distributed Systems Tracing Infrastructure中就已经系统的阐述了如何在分布式系统中使用调用链来解决该问题，更更有意思的是，国内几乎所有介绍调用链系统的文章都会提到这篇论文，也会引用下面这张图?‍♀️……

这个问题虽然在国内已经被谈及了多次，但是我们发现大部分文章都是介绍如何基于open-trace, zipkin建立一个跟踪系统，即使有自建系统的分享，也相对缺乏对构建适合自己的调用链系统时的考量和分析。因此，这里主要谈谈我们在做调用链系统时的设计理念和取舍。

cdapper调用链系统

我们的调用链系统取名cdapper, 显然基本思想来自Google Dapper论文，对此我们毫不避讳。字母c则是公司首字母。

我们没有采用开源的dapper系统，而是完全自己实现。主要有以下原因：

开源的dapper系统为了顾及通用行和普适性，代码都非常臃肿。这与我们一贯主张的「大系统小做」原则不符。
dapper的思想并不复杂，甚至可以说非常简单，因此自己实现这套系统并不是「没问题创造问题去解决」。事实上，我们的dapper数据采集核心代码（golang实现）也就400行以内。
我们的基础服务组件有很多定制功能，直接使用开源dapper系统不可避免设计相关系统改造，而我们自己实现系统，可以从设计之初就与已有基础组件整合。

在我们的系统中，各微服务主要通过http API进行交互，下面以API调用为例进行说明：

流量通过gateway服务进入系统，gateway会在每一个http API请求的header中生成一个trace_id，然后将该请求分发给各个微服务。
微服务接收到请求后，会生成一个子节点span_id。以service_0为例，生成子节点span_id_0, service 0调用service 1, 则继续生成子节点span_id_1. trace_id, span_id_0, span_id_1,...形成链式关系。
节点的粒度粗细无须局限在http API级别，各个微服务可以根据自身业务特点，进一步将子节点细化到每一个redis调用、每一次mysql查询等。
我们使用自己深度定制的gin作为http 微服务的基础框架，通过中间件的方式来集成调用链。大部分时候，添加一行代码即可完成调用链集成。此外，我也提供redis, mysql, rpc等其他基础组件的低侵入集成方式。
调用链数据统一输出到我们的data bus的基础组件。data bus是一个架构上的逻辑概念，目前实现层面主要通过文件日志和kafka承载。我们不推荐使用tcp网络传输调用链数据。主要有以下考虑：
1. 我们使用的是阿里云，而阿里云内网抖动几乎是家常便饭。使用网络传输调用链数据稳定性差。
2. 由于业务性质，流量存在突发性，我们需要一个高吞吐的的消息队列来缓冲这种尖峰流量。在这方面，我们非常信赖kafka.
data bus是我们整个分布式系统的数据总线，我们在这个总线上安插了很多子系统，其中一个子系统是log sink日志数据汇聚系统（其他子系统后面有空了介绍）。该系统对调用链数据进行收集，然后存储在cassandra中。在这方面，我们非常信赖cassandra, 尤其是其优异的写入性能。如果你对我们如何存储调用链数据感兴趣，可以看上一篇文章LSM Tree/MemTable/SSTable基本原理了解细节。
每一个http API请求的响应中都会在header中返回trace_id。研发通过trace query进行调用链查询：

小结

微服务不是银弹，在带来很多好处的同时，也对整个系统架构的其他基础提出了更高的要求。微服务系统要先运作起来，一个简单、高效的系统诊断工具对系统微服务化的落地有着非常重要的促进作用。调用链就是一把这样的瑞士军刀。当然，调用链其实已经是8年前的理念了，如果你关心这个选题的最新发展方向，你还需要持续保持对世界上最大的分布式系统公司——Google的关注?！

使用Golang reflect 对 gin handler 进行简单封装

发表于2017-03-23由daniel

使用Golang reflect 对 gin handler 进行简单封装

项目中大量使用 gin 作为service API的 http framework. 大部分时候我们的代码结构类似这样：

func main() {

    engine := gin.New()
    engine.GET("/hi", hiHandler)
    engine.Run(":8083")
}

func hiHandler(c *gin.Context) { 
    rsp, err := businessLogicProcess()
    if err != nil {
        c.AbortWithStatus(http.StatusBadRequest)
    } else {
        c.JSON(http.StatusOK, rsp)
    }
 }

func businessLogicProcess() {
    ...
}

func main() {

engine := gin.New()

engine.GET("/hi", hiHandler)

engine.Run(":8083")

}

func hiHandler(c *gin.Context) {

rsp, err := businessLogicProcess()

if err != nil {

c.AbortWithStatus(http.StatusBadRequest)

} else {

c.JSON(http.StatusOK, rsp)

}

func businessLogicProcess() {

...

}

数据流：hiHandler -> businessLogicProcess -> hiHandler. 这本身没有什么严重的问题，但是当你注册的API越来越多的时候，你的项目中会出现大量重复且类似hiHandler 结构的胶水层handler: hiHandler只做了一个 http request 数据与 businessLogicProcess 的粘合，再将返回数据塞回 http response.

从数据流来看，这个胶水层是无法避免的。大量重复的 hiHandler 并不符合 write reusable code 原则。因此，我们可以尝试对该层统一抽象进行封装：

func main() {

    engine := gin.New()
    engine.GET("/time", WrapHandler(GetTime))
    engine.Run(":8083")
}


type GetTimeReq struct {
    Nation string `form:"nation" binding:"required" json:"nation"`
}

func GetTime(req *GetTimeReq) (*OKRsp, *ErrRsp) {
    ...
}


// WrapHandler wrap gin handler, the handler signature should be
// func(req *ReqStruct) (*OKRspStruct, *ErrorRspStruct) or
// func(c *gin.Context, req *ReqStruct) (*OKRspStruct, *ErrorRspStruct)
func WrapHandler(f interface{}) gin.HandlerFunc {
    t := reflect.TypeOf(f)
    if t.Kind() != reflect.Func {
        panic("handdler should be function type")
    }

    fnumIn := t.NumIn()
    if fnumIn == 0 || fnumIn &gt; 2 {
        panic("handler function require 1 or 2 input parameters")
    }

    if fnumIn == 2 {
        tc := reflect.TypeOf(&amp;gin.Context{})
        if t.In(0) != tc {
            panic("handler function first paramter should by type of *gin.Context if you have 2 input parameters")
        }
    }

    if t.NumOut() != 2 {
        panic("handler function return values should contain response data &amp; error")
    }

    // errorInterface := reflect.TypeOf((*error)(nil)).Elem()
    // if !t.Out(1).Implements(errorInterface) {
    //  panic("handler function second return value should by type of error")

    // }

    return func(c *gin.Context) {
        var req interface{}
        if fnumIn == 1 {
            req = newReqInstance(t.In(0))
        } else {
            req = newReqInstance(t.In(1))
        }

        if !c.Bind(req) {
            err := c.LastError()
            log.Warning("bind parameter error:%v", err)
            GinErrRsp(c, ErrCodeParamInvalid, err.Error())
            return
        }

        var inValues []reflect.Value
        if fnumIn == 1 {
            inValues = []reflect.Value{reflect.ValueOf(req)}
        } else {
            inValues = []reflect.Value{reflect.ValueOf(c), reflect.ValueOf(req)}
        }

        ret := reflect.ValueOf(f).Call(inValues)
        if ret[1].IsNil() {
            GinRsp(c, http.StatusOK, ret[0].Interface())
        } else {
            GinRsp(c, http.StatusOK, ret[1].Interface())
        }
    }
}

func newReqInstance(t reflect.Type) interface{} {
    switch t.Kind() {
    case reflect.Ptr, reflect.Interface:
        return newReqInstance(t.Elem())
    default:
        return reflect.New(t).Interface()
    }
}

func main() {

engine := gin.New()

engine.GET("/time", WrapHandler(GetTime))

engine.Run(":8083")

}

type GetTimeReq struct {

Nation string `form:"nation" binding:"required" json:"nation"`

}

func GetTime(req *GetTimeReq) (*OKRsp, *ErrRsp) {

...

}

// WrapHandler wrap gin handler, the handler signature should be

// func(req *ReqStruct) (*OKRspStruct, *ErrorRspStruct) or

// func(c *gin.Context, req *ReqStruct) (*OKRspStruct, *ErrorRspStruct)

func WrapHandler(f interface{}) gin.HandlerFunc {

t := reflect.TypeOf(f)

if t.Kind() != reflect.Func {

panic("handdler should be function type")

}

fnumIn := t.NumIn()

if fnumIn == 0 || fnumIn > 2 {

panic("handler function require 1 or 2 input parameters")

}

if fnumIn == 2 {

tc := reflect.TypeOf(&gin.Context{})

if t.In(0) != tc {

panic("handler function first paramter should by type of *gin.Context if you have 2 input parameters")

}

if t.NumOut() != 2 {

panic("handler function return values should contain response data & error")

}

// errorInterface := reflect.TypeOf((*error)(nil)).Elem()

// if !t.Out(1).Implements(errorInterface) {

// panic("handler function second return value should by type of error")

// }

return func(c *gin.Context) {

var req interface{}

if fnumIn == 1 {

req = newReqInstance(t.In(0))

} else {

req = newReqInstance(t.In(1))

}

if !c.Bind(req) {

err := c.LastError()

log.Warning("bind parameter error:%v", err)

GinErrRsp(c, ErrCodeParamInvalid, err.Error())

return

}

var inValues []reflect.Value

if fnumIn == 1 {

inValues = []reflect.Value{reflect.ValueOf(req)}

} else {

inValues = []reflect.Value{reflect.ValueOf(c), reflect.ValueOf(req)}

}

ret := reflect.ValueOf(f).Call(inValues)

if ret[1].IsNil() {

GinRsp(c, http.StatusOK, ret[0].Interface())

} else {

GinRsp(c, http.StatusOK, ret[1].Interface())

}

func newReqInstance(t reflect.Type) interface{} {

switch t.Kind() {

case reflect.Ptr, reflect.Interface:

return newReqInstance(t.Elem())

default:

return reflect.New(t).Interface()

}

这样做还有一个额外的好处：实现了业务处理函数 (GetTime) 与gin的解耦，使得业务处理函数复用性更强。

golang中的错误处理模式

发表于2015-04-14由daniel

年前拜读了Rob Pike的Errors are values，不禁感叹——原来错误可以这样处理。这里简单做一个笔记。

模式1：闭包记录异常

var err error
write := func(buf []byte) {
    if err != nil {
        return
    }
    _, err = w.Write(buf)
}
write(p0[a:b])
write(p1[c:d])
write(p2[e:f])
// and so on
if err != nil {
    return err
}

var err error

write := func(buf []byte) {

if err != nil {

return

}

_, err = w.Write(buf)

}

write(p0[a:b])

write(p1[c:d])

write(p2[e:f])

// and so on

if err != nil {

return err

}

这种方式适用于多次相同类型调用的错误处理。缺点就是，每个类似的地方都需要引入一个闭包函数，显得冗余重复。

模式2：组合error struct

type errWriter struct {
    w   io.Writer
    err error
}

func (ew *errWriter) write(buf []byte) {
    if ew.err != nil {
        return
    }
    _, ew.err = ew.w.Write(buf)
}

...

ew := &errWriter{w: fd}
ew.write(p0[a:b])
ew.write(p1[c:d])
ew.write(p2[e:f])
// and so on
if ew.err != nil {
    return ew.err
}

type errWriter struct {

w io.Writer

err error

}

func (ew *errWriter) write(buf []byte) {

if ew.err != nil {

return

}

_, ew.err = ew.w.Write(buf)

}

...

ew := &errWriter{w: fd}

ew.write(p0[a:b])

ew.write(p1[c:d])

ew.write(p2[e:f])

// and so on

if ew.err != nil {

return ew.err

}

代码比第一种方式更加干净。但是与第一种方式有一样的问题，就是无法具体的知道是在哪个位置发生的错误以及错误的具体信息。

模式3：error转化为panic

来自知乎唐生的回答：

func ce(msg string, err error) {                                                                                                                                           
 if err != nil {                                                                                                                                                           
  panic(fmt.Errorf("%s error: %v", msg, err))                                                                                                                              
 }                                                                                                                                                                         
}                                                                                                                                                                          
                                                                                                                                                                           
func catch(err *error) {                                                                                                                                                   
 if p := recover(); p != nil {                                                                                                                                             
  *err = p.(error)                                                                                                                                                         
 }                                                                                                                                                                         
}                                                                                                                                                                          
                                                                                                                                                                           
func foo() (err error) {                                                                                                                                                   
 defer catch(&err)                                                                                                                                                         
                                                                                                                                                                           
 // ...                                                                                                                                                                    
 ce("duang", err)                                                                                                                                                          
 // ...                                                                                                                                                                    
 ce("duang", err)                                                                                                                                                          
 // ...                                                                                                                                                                    
 ce("duang", err)                                                                                                                                                          
                                                                                                                                                                           
 return                                                                                                                                                                    
}

func ce(msg string, err error) {

if err != nil {

panic(fmt.Errorf("%s error: %v", msg, err))

}

func catch(err *error) {

if p := recover(); p != nil {

*err = p.(error)

}

func foo() (err error) {

defer catch(&err)

// ...

ce("duang", err)

// ...

ce("duang", err)

// ...

ce("duang", err)

return

}

这种方式的通用性较好。但是把error主动转化为panic会导致goroutine的stack unwind。这会带来一定的系统开销，甚至是系统性能抖动。

参考：

行思錄 | Travel Coder

Arch, Coding, Life

分类目录归档：Design Pattern

微服务troubleshooting利器——调用链

微服务troubleshooting利器——调用链

微服务的喜与悲

cdapper调用链系统

小结

使用Golang reflect 对 gin handler 进行简单封装

使用Golang reflect 对 gin handler 进行简单封装

golang中的错误处理模式

模式1：闭包记录异常

模式2：组合error struct

模式3：error转化为panic