在前面的《通过项目学习Go语言之...》系列文章中,我们对Go语言开发环境的配置以及开发项目最基础的功能之go mod、log做了一些入门的讲解。同时,也对项目gatekeeper中使用的核心组件gin做了一个稍微详细的说明。
本篇,将通过分析gatekeeper项目结构,理清通过代理到转发到后端realserver以及将代理结果最终响应请求的全流程。
下面先看一张getekeeper的http网关代理的请求响应全流程图:
接下来,我们对整个服务的启动和重要的流程节点进行源码的分析。
启动
一般服务启动过程中主要是进行初始化、回调钩子和监听注册等基础性工作。gatekeeper也不例外,再启动过程中,进行了配置加载、数据库访问初始化、http和tcp检测注册、系统信号监听等一系列工作。
//main.go
func main() {
conf = flag.String("config", "./conf/dev/", "input config file like ./conf/dev/")
flag.Parse()
//初始化mysql redis配置
lib.InitModule(*conf,[]string{"base","mysql","redis",})
defer lib.Destroy()
public.InitMysql()
public.InitConf()
//初始化配置、配置管理设置实时监控配置变化并刷新
service.SysConfMgr = service.NewSysConfigManage()
service.SysConfMgr.InitConfig()
service.SysConfMgr.MonitorConfig()
//注册请求前验证request方法,进行请求处理前的验证,只允许授权方访问
//目前支持的是固定的配置模式,可以根据需求进行自定义
service.RegisterBeforeRequestAuthFunc(service.AuthAppToken)
//注册请求后更改response方法
service.RegisterModifyResponseFunc(service.FilterCityData([]string{"/gatekeeper/tester_filter/goods_list"}))
//启动http、tcp监听
router.HTTPServerRun()
router.TCPServerRun()
//注册系统信号监听
quit := make(chan os.Signal)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
//注册服务优雅关闭
router.TCPServerStop()
router.HTTPServerStop()
signal.Stop(quit)
}
Http服务监听是处理来自外部的全部请求,下面看一下HTTPServerRun,getekeeper的http服务是基于gin来实现的。
//httpserver.go
//HTTPServerRun 服务启动
func HTTPServerRun() {
//设置运行模式 debug
gin.SetMode(lib.ConfBase.DebugMode)
//初始化路由
r := InitRouter()
//设置http server运行参数
HTTPSrvHandler = &http.Server{
Addr: lib.GetStringConf("base.http.addr"),
Handler: r,
ReadTimeout: time.Duration(lib.GetIntConf("base.http.read_timeout")) * time.Second,
WriteTimeout: time.Duration(lib.GetIntConf("base.http.write_timeout")) * time.Second,
MaxHeaderBytes: 1 << uint(lib.GetIntConf("base.http.max_header_bytes")),
}
//设置recover 以实现遇到panic时服务不中断
//启动http服务
go func() {
defer func() {
if err := recover(); err != nil {
public.SysLogger.Error("HttpServerRun_recover:%v", err)
}
}()
log.Printf(" [INFO] HttpServer %s listening\n",lib.GetStringConf("base.http.addr"))
if err := HTTPSrvHandler.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatalf(" [ERROR] HttpServer %s err:%v\n", lib.GetStringConf("base.http.addr"), err)
}
}()
}
路由注册
在上面启动http服务的代码中,调用了InitRouter()对相关的路由进行的设置,分别注册了四组路由:admin(管理后台)、gateway(服务探活ping)、cluster(集群管理,reload处理配置变更刷新)、gatekeeper(网关代理服务);另外还启动了一个静态服务器/assets,用于管理后台资源的服务器。下面我们看一下详细的代码:
//httprouter.go
//InitRouter 声明http路由
func InitRouter() *gin.Engine {
//创建路由r(engine),注册中间件
router := gin.New()
//加载gin内置中间件,recovery所有panic是服务长久运行不中断
router.Use(middleware.Recovery())
//admin
admin := router.Group("/admin")
admin.Use(middleware.RequestTraceLog())
{
controller.AdminRegister(admin)
}
//assets
router.Static("/assets", "./tmpl/green/assets")
//gateway
//提供探活
gateway := controller.Gateway{}
router.GET("/ping", gateway.Ping)
//cluster
//提供服务配置刷新
csr:=router.Group("/")
csr.Use(middleware.ClusterAuth())
csr.GET("/reload", gateway.Reload)
//注册gatekeeper路由组,代理所有网关请求
gw:=router.Group(lib.GetStringConf("base.http.route_prefix"))
//注册一系列中间件
gw.Use(
middleware.RequestTraceLog(),
middleware.MatchRule(),
middleware.AccessControl(),
middleware.HTTPLimit(),
//todo 拓展中间件
//核心中间件部分
middleware.LoadBalance())
{
gw.GET("/*action", gateway.Index)
gw.POST("/*action", gateway.Index)
gw.DELETE("/*action", gateway.Index)
gw.OPTIONS("/*action", gateway.Index)
}
return router
}
中间件注册
在上面的代码中,可以看到默认注册了RequestTraceLog(),MatchRule(),AccessControl(),HTTPLimit(),LoadBalance()五个getekeeper项目自定义的中间件,分配用于请求treace日志记录、规则匹配、访问控制、限流和负载均衡。
- RequestTraceLog
通过trace可以快速定位请求链,在进行微服务构建时,使用teaceid机制能够很方便的绘制请求的整个链路,以方便问题的排查和各段服务的耗时记录,优化服务。
//trace_log.go
//RequestTraceLog trace中间件
func RequestTraceLog() gin.HandlerFunc {
return func(c *gin.Context) {
RequestInLog(c)
defer RequestOutLog(c)
c.Next()
}
}
//RequestInLog 请求进入日志
func RequestInLog(c *gin.Context) {
//初始化trance 默认分配一个traceid
traceContext := lib.NewTrace()
//如果能在header中获取到traceid,则使用已有的id
if traceID := c.Request.Header.Get("didi-header-rid"); traceID != "" {
traceContext.TraceId = traceID
}
if spanID := c.Request.Header.Get("didi-header-spanid"); spanID != "" {
traceContext.SpanId = spanID
}
...
//将trace相关的追踪信息保存进上下文request
c.Set("startExecTime", time.Now())
c.Set("trace", traceContext)
c.Request.Header.Set("didi-header-rid", traceContext.TraceId)
c.Request = c.Request.WithContext(context.WithValue(c.Request.Context(), public.ContextKey("trace"), traceContext))
c.Request = c.Request.WithContext(context.WithValue(c.Request.Context(), public.ContextKey("request_url"), c.Request.URL.Path))
c.Request.Body = ioutil.NopCloser(bytes.NewBuffer(bodyBytes))
}
- MatchRule
用于规则的匹配,基于分解URI实现和管理配置后台相对应的转发逻辑的匹配,
//match_rule.go
//MatchRule 匹配模块中间件
func MatchRule() gin.HandlerFunc {
return func(c *gin.Context) {
//初始化geteway service
gws := service.NewGateWayService(c.Writer, c.Request)
//进行规则匹配 选出匹配的module
if err := gws.MatchRule(); err != nil {
public.ResponseError(c, http.StatusBadRequest, err)
return
}
c.Set(MiddlewareServiceKey,gws)
}
}
- AccessControl
权限控制中间件,用于对外部过来的请求就行访问控制过滤,目前默认支持的包括IP黑白名单、主机白名单以及请求注册函数过滤
//gate_service.go
//AccessControl 权限验证
func (o *GateWayService) AccessControl() error {
if o.currentModule.AccessControl == nil {
return nil
}
ctx := public.NewContext(o.w, o.req)
var errmsg string
switch {
case !AuthModuleOpened(o, ctx):
public.ContextNotice(o.req.Context(), DLTagAccessControlSuccess, map[string]interface{}{
"msg": "access_control_not_open",
})
return nil
case AuthInBlackIPList(o, ctx):
public.ContextNotice(o.req.Context(), DLTagAccessControlFailure, map[string]interface{}{
"msg": "AuthInBlackIPList",
})
return errors.New("msg:AuthInBlackIPList")
case AuthInWhiteIPList(o, ctx):
public.ContextNotice(o.req.Context(), DLTagAccessControlSuccess, map[string]interface{}{
"msg": "AuthWhiteIPList_success",
})
return nil
case AuthInWhiteHostList(o, ctx):
public.ContextNotice(o.req.Context(), DLTagAccessControlSuccess, map[string]interface{}{
"msg": "AuthWhiteHostList_success",
})
return nil
case AuthRegisterFunc(o, &errmsg):
public.ContextNotice(o.req.Context(), DLTagAccessControlSuccess, map[string]interface{}{
"msg": "AuthRegisterFunc_success",
})
return nil
}
if errmsg==""{
errmsg="auth_failure"
}
public.ContextWarning(o.req.Context(), DLTagAccessControlFailure, map[string]interface{}{
"msg": errmsg,
})
return errors.New(errmsg)
}
- HTTPLimit
限流控制中间件,以保护后端的realserver服务器以免过载导致服务异常不可用。
//HTTPLimit http限流中间件
func HTTPLimit() gin.HandlerFunc {
return func(c *gin.Context) {
//获取上游服务
gws, ok := c.MustGet(MiddlewareServiceKey).(*service.GateWayService)
if !ok {
public.ResponseError(c, http.StatusBadRequest, errors.New("gateway_service not valid"))
return
}
//入口流量统计
currentModule := gws.CurrentModule()
counter := public.FlowCounterHandler.GetRequestCounter(currentModule.Base.Name)
counter.Increase(c.Request.Context(), c.Request.RemoteAddr)
//客户端ip限流
remoteIP := public.Substr(c.Request.RemoteAddr, 0, int64(strings.Index(c.Request.RemoteAddr, ":")))
if currentModule.AccessControl.ClientFlowLimit > 0 {
limiter := public.FlowLimiterHandler.GetModuleIPVisitor(currentModule.Base.Name+"_"+remoteIP, currentModule.AccessControl.ClientFlowLimit)
if limiter.Allow() == false {
errmsg := fmt.Sprintf("moduleName:%s remoteIP:%s, QPS limit : %d, %d", currentModule.Base.Name, remoteIP, int64(limiter.Limit()), limiter.Burst())
public.ContextWarning(c.Request.Context(), service.DLTagAccessControlFailure, map[string]interface{}{
"msg": errmsg,
"ip": remoteIP,
"moduleName": currentModule.Base.Name,
})
public.ResponseError(c, http.StatusBadRequest, errors.New(errmsg))
}
}
//todo
c.Next()
}
}
- LoadBalance
该中间件是最核心的部门,它负责整个负责均衡的处理,负责根据匹配规则创建正确的proxy,进行正常服务的处理。目前提供的负载均衡策略是rr,同时也实现了后端realserver的探活自动剔除策略。
//load_balance.go
//LoadBalance 负载均衡中间件
func LoadBalance() gin.HandlerFunc {
return func(c *gin.Context) {
gws,ok:=c.MustGet(MiddlewareServiceKey).(*service.GateWayService)
if !ok{
public.ResponseError(c, http.StatusBadRequest, errors.New("gateway_service not valid"))
return
}
//进入核心负载算法,选出正常的服务
proxy, err := gws.LoadBalance()
if err != nil {
public.ResponseError(c, http.StatusProxyAuthRequired, err)
return
}
requestBody,ok:=c.MustGet(MiddlewareRequestBodyKey).([]byte)
if !ok{
public.ResponseError(c, http.StatusBadRequest, errors.New("request_body not valid"))
return
}
c.Request.Body = ioutil.NopCloser(bytes.NewBuffer(requestBody))
proxy.ServeHTTP(c.Writer, c.Request)
c.Abort()
}
}
//LoadBalance 请求负载
func (o *GateWayService) LoadBalance() (*httputil.ReverseProxy, error) {
ipList, err := SysConfMgr.GetModuleIPList(o.currentModule.Base.Name)
if err != nil {
public.ContextWarning(o.req.Context(), DLTagLoadBalanceFailure, map[string]interface{}{
"msg": err,
"modulename": o.currentModule.Base.Name,
"availableIpList": SysConfMgr.GetAvaliableIPList(o.currentModule.Base.Name),
})
return nil, errors.New("get_iplist_error")
}
if len(ipList) == 0 {
public.ContextWarning(o.req.Context(), DLTagLoadBalanceFailure, map[string]interface{}{
"msg": "empty_iplist_error",
"modulename": o.currentModule.Base.Name,
"availableIpList": SysConfMgr.GetAvaliableIPList(o.currentModule.Base.Name),
})
return nil, errors.New("empty_iplist_error")
}
//正常proxy的遴选
proxy, err := o.GetModuleHTTPProxy()
if err != nil {
public.ContextWarning(o.req.Context(), DLTagLoadBalanceFailure, map[string]interface{}{
"msg": err,
"module": o.currentModule.Base.Name,
})
return nil, err
}
return proxy, nil
}
//GetModuleHTTPProxy 获取模块的代理
func (o *GateWayService) GetModuleHTTPProxy() (*httputil.ReverseProxy, error) {
proxy,err:=SysConfMgr.GetModuleHTTPProxy(o.currentModule.Base.Name)
if err != nil {
public.ContextWarning(o.req.Context(), DLTagLoadBalanceFailure, map[string]interface{}{
"err": err,
"module": o.currentModule.Base.Name,
})
return &httputil.ReverseProxy{}, err
}
return proxy,nil
}
//GetModuleHTTPProxy 获取http代理方法
func (s *SysConfigManage) GetModuleHTTPProxy(moduleName string) (*httputil.ReverseProxy, error) {
//基于rr的module选择
rr, err := s.GetModuleRR(moduleName)
if err != nil {
return nil, err
}
s.moduleProxyFuncMapLocker.RLock()
defer s.moduleProxyFuncMapLocker.RUnlock()
proxyFunc, ok := s.moduleProxyFuncMap[moduleName]
if ok {
return proxyFunc(rr), nil
}
return nil, errors.New("module proxy empty")
}
以上,我们对getekeeper的启动以及核心的组件加载和请求流程进行了分析,从源码上看,getekeeper的代码思路还是很清晰明了的。
本节完。