micro/service.go

package micro

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"net"
	hp "net/http"
	"net/http/pprof"
	"os"
	"os/signal"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"git.nspix.com/golang/micro/gateway"
	"git.nspix.com/golang/micro/gateway/http"
	"git.nspix.com/golang/micro/gateway/rpc"
	"git.nspix.com/golang/micro/log"
	"git.nspix.com/golang/micro/registry"
	"git.nspix.com/golang/micro/utils/docker"
	"git.nspix.com/golang/micro/utils/metric"
	"git.nspix.com/golang/micro/utils/net/ip"
	"git.nspix.com/golang/micro/utils/unsafestr"
)

var (
	upMetricProcessFlag int32
)

type Service struct {
	opts        *Options
	ctx         context.Context
	cancelFunc  context.CancelFunc
	registry    registry.Registry
	node        *registry.ServiceNode
	listener    net.Listener
	gateway     *gateway.Gateway
	wg          sync.WaitGroup
	httpSvr     *http.Server
	rpcSvr      *rpc.Server
	upTime      time.Time
	client      *Client
	environment string
}

func (svr *Service) wrapSync(f func()) {
	svr.wg.Add(1)
	go func() {
		f()
		svr.wg.Done()
	}()
}

func (svr *Service) eventLoop() {
	var (
		err            error
		registryTicker *time.Ticker
		upMetricTicker *time.Ticker
	)
	registryTicker = time.NewTicker(time.Second * 20)
	upMetricTicker = time.NewTicker(time.Second * 5)
	defer func() {
		registryTicker.Stop()
	}()
	for {
		select {
		case <-registryTicker.C:
			if !svr.opts.DisableRegister {
				if err = svr.registry.Register(svr.node); err != nil {
					log.Warnf("registry service %s error: %s", svr.opts.Name, err.Error())
				}
			}
		case <-upMetricTicker.C:
			if svr.opts.TSDBUrl != "" {
				if atomic.CompareAndSwapInt32(&upMetricProcessFlag, 0, 1) {
					go svr.upMetrics()
				}
			}
		case <-svr.ctx.Done():
			return
		}
	}
}

func (svr *Service) upMetrics() {
	ms := make(metric.Values, 0)
	tags := make(map[string]string)
	tags["host"] = ip.InternalIP()
	tags["name"] = svr.opts.ShortName()
	tags["version"] = svr.opts.Version
	memStats := &runtime.MemStats{}
	runtime.ReadMemStats(memStats)
	ms = append(ms, metric.New("sys.go.goroutine", float64(runtime.NumGoroutine())).SetTags(tags))
	ms = append(ms, metric.New("sys.go.ccall", float64(runtime.NumCgoCall())).SetTags(tags))
	ms = append(ms, metric.New("sys.go.cpus", float64(runtime.NumCPU())).SetTags(tags))
	ms = append(ms, metric.New("sys.go.gc", float64(memStats.NumGC)).SetTags(tags))
	ms = append(ms, metric.New("sys.go.next_gc", float64(memStats.NextGC)).SetTags(tags))
	ms = append(ms, metric.New("sys.go.heap_alloc", float64(memStats.HeapAlloc)).SetTags(tags))
	ms = append(ms, metric.New("sys.go.heap_inuse", float64(memStats.HeapInuse)).SetTags(tags))
	ms = append(ms, metric.New("sys.go.memory_allow", float64(memStats.Alloc)).SetTags(tags))
	ms = append(ms, metric.New("sys.go.memory_frees", float64(memStats.Frees)).SetTags(tags))

	if svr.opts.MetricCallback != nil {
		vs := svr.opts.MetricCallback()
		for _, v := range vs {
			v.SetTags(tags)
			ms = append(ms, v)
		}
	}
	metric.Push(svr.ctx, svr.opts.TSDBUrl, ms)
}

func (svr *Service) Handle(method string, cb HandleFunc, opts ...HandleOption) {
	opt := &HandleOptions{HttpMethod: "POST"}
	for _, f := range opts {
		f(opt)
	}
	//HTTP处理
	if svr.opts.EnableHttp && !opt.DisableHttp {
		if opt.HttpPath == "" {
			opt.HttpPath = strings.ReplaceAll(method, ".", "/")
		}
		if opt.HttpPath[0] != '/' {
			opt.HttpPath = "/" + opt.HttpPath
		}
		svr.httpSvr.Handle(opt.HttpMethod, opt.HttpPath, func(ctx *http.Context) (err error) {
			return cb(ctx)
		})
	}
	//启动RPC功能
	if svr.opts.EnableRPC && !opt.DisableRpc {
		svr.rpcSvr.Handle(method, func(ctx *rpc.Context) error {
			return cb(ctx)
		})
	}
	return
}

func (svr *Service) NewRequest(name, method string, body interface{}) (req *Request, err error) {
	return &Request{
		ServiceName: name,
		Method:      method,
		Body:        body,
		client:      svr.client,
	}, nil
}

func (svr *Service) PeekService(name string) ([]*registry.ServiceNode, error) {
	return svr.registry.Get(name)
}

func (svr *Service) HttpServe() *http.Server {
	return svr.httpSvr
}

func (svr *Service) RPCServe() *rpc.Server {
	return svr.rpcSvr
}

func (svr *Service) Node() *registry.ServiceNode {
	return svr.node
}

func (svr *Service) Environment() string {
	return svr.environment
}

func (svr *Service) instance() *registry.ServiceNode {
	var (
		err      error
		id       string
		dockerID string
		tcpAddr  *net.TCPAddr
		ipLocal  string
		node     *registry.ServiceNode
	)
	if id, err = docker.SelfContainerID(); err != nil {
		//生成唯一ID
		e5 := md5.New()
		e5.Write(unsafestr.StringToBytes(svr.opts.Name))
		e5.Write(unsafestr.StringToBytes(svr.opts.Version))
		id = hex.EncodeToString(e5.Sum(nil))
	} else {
		dockerID = id
		svr.environment = EnvironmentDocker
	}
	node = &registry.ServiceNode{
		ID:        id,
		Name:      svr.opts.Name,
		Version:   svr.opts.Version,
		Metadata:  map[string]string{},
		Addresses: make(map[string]registry.Addr),
	}
	if svr.opts.Address == "" {
		ipLocal = ip.InternalIP()
	} else {
		ipLocal = svr.opts.Address
	}
	node.Address = ipLocal
	if svr.listener != nil {
		if tcpAddr, err = net.ResolveTCPAddr("tcp", svr.listener.Addr().String()); err == nil {
			node.Port = tcpAddr.Port
		}
	} else {
		node.Port = svr.opts.Port
	}
	node.Metadata["docker-id"] = dockerID
	if svr.opts.EnableHttp {
		node.Metadata["enable-http"] = "true"
	}
	if svr.opts.EnableRPC {
		node.Metadata["enable-rpc"] = "true"
	}
	return node
}

func (svr *Service) startHttpServe() (err error) {
	l := gateway.NewListener(svr.listener.Addr())
	if err = svr.gateway.Attaches([][]byte{[]byte("GET"), []byte("POST"), []byte("PUT"), []byte("DELETE"), []byte("OPTIONS")}, l); err == nil {
		svr.wrapSync(func() {
			if err = svr.httpSvr.Serve(l); err != nil {
				log.Warnf("http serve error: %s", err.Error())
			}
		})
		svr.httpSvr.Handle("GET", "/healthy", func(ctx *http.Context) (err error) {
			return ctx.Success(map[string]interface{}{
				"id":      svr.node.ID,
				"healthy": "healthy",
				"uptime":  time.Now().Sub(svr.upTime).String(),
			})
		})
		if svr.opts.EnableHttpPProf {
			svr.httpSvr.Handler("GET", "/debug/pprof/", hp.HandlerFunc(pprof.Index))
			svr.httpSvr.Handler("GET", "/debug/pprof/goroutine", hp.HandlerFunc(pprof.Index))
			svr.httpSvr.Handler("GET", "/debug/pprof/heap", hp.HandlerFunc(pprof.Index))
			svr.httpSvr.Handler("GET", "/debug/pprof/mutex", hp.HandlerFunc(pprof.Index))
			svr.httpSvr.Handler("GET", "/debug/pprof/threadcreate", hp.HandlerFunc(pprof.Index))
			svr.httpSvr.Handler("GET", "/debug/pprof/cmdline", hp.HandlerFunc(pprof.Cmdline))
			svr.httpSvr.Handler("GET", "/debug/pprof/profile", hp.HandlerFunc(pprof.Profile))
			svr.httpSvr.Handler("GET", "/debug/pprof/symbol", hp.HandlerFunc(pprof.Symbol))
			svr.httpSvr.Handler("GET", "/debug/pprof/trace", hp.HandlerFunc(pprof.Trace))
		}
		log.Infof("attach http server success")
	} else {
		log.Warnf("attach http listener error: %s", err.Error())
	}
	return
}

func (svr *Service) startRpcServe() (err error) {
	l := gateway.NewListener(svr.listener.Addr())
	if err = svr.gateway.Attach([]byte("RPC"), l); err == nil {
		svr.wrapSync(func() {
			if err = svr.rpcSvr.Serve(l); err != nil {
				log.Warnf("rpc serve error: %s", err.Error())
			}
		})
		log.Infof("attach rpc server success")
	} else {
		log.Warnf("attach rpc listener error: %s", err.Error())
	}
	return
}

func (svr *Service) prepare() (err error) {
	log.Prefix(svr.opts.Name)
	svr.ctx = WithContext(svr.ctx, svr)
	if svr.opts.EnableInternalListener {
		var tcpAddr *net.TCPAddr
		//绑定指定的端口
		if svr.opts.Port != 0 {
			tcpAddr = &net.TCPAddr{
				Port: svr.opts.Port,
			}
		}
		//绑定指定的IP
		if svr.opts.Address != "" {
			if tcpAddr == nil {
				tcpAddr = &net.TCPAddr{
					IP: net.ParseIP(svr.opts.Address),
				}
			} else {
				tcpAddr.IP = net.ParseIP(svr.opts.Address)
			}
		}
		if svr.listener, err = net.ListenTCP("tcp", tcpAddr); err != nil {
			return
		}
		log.Infof("listener on: %s", svr.listener.Addr())
		svr.gateway = gateway.New(svr.listener)
		svr.wrapSync(func() {
			svr.gateway.Run(svr.ctx)
		})

		//start http serve
		if svr.opts.EnableHttp {
			err = svr.startHttpServe()
		}

		//start rpc serve
		if svr.opts.EnableRPC {
			err = svr.startRpcServe()
		}
	}
	svr.node = svr.instance()
	svr.wrapSync(func() {
		svr.eventLoop()
	})
	if !svr.opts.DisableRegister {
		_ = svr.registry.Register(svr.node)
	}
	return
}

func (svr *Service) destroy() (err error) {
	log.Infof("service stopping")
	svr.cancelFunc()
	if !svr.opts.DisableRegister {
		if err = svr.registry.Deregister(svr.node); err != nil {
			log.Warnf("deregister service %s error: %s", svr.opts.Name, err.Error())
		} else {
			log.Infof("deregister service %s successful", svr.opts.Name)
		}
	}
	if svr.listener != nil {
		if err = svr.listener.Close(); err != nil {
			log.Warnf(err.Error())
		}
	}
	if err = svr.client.Close(); err != nil {
		log.Warnf(err.Error())
	}
	log.Infof("service stopped")
	return
}

func (svr *Service) Run() (err error) {
	log.Infof("service starting")
	if err = svr.prepare(); err != nil {
		return
	}
	//start server
	if svr.opts.Server != nil {
		if err = svr.opts.Server.Start(svr.ctx); err != nil {
			return
		}
	}
	log.Infof("service ready")
	//waiting
	ch := make(chan os.Signal, 1)
	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGKILL)
	select {
	case <-ch:
	case <-svr.ctx.Done():
	}
	//stop server
	if svr.opts.Server != nil {
		err = svr.opts.Server.Stop()
	}
	return svr.destroy()
}

func New(opts ...Option) *Service {
	o := NewOptions()
	for _, opt := range opts {
		opt(o)
	}
	svr := &Service{
		opts:        o,
		upTime:      time.Now(),
		httpSvr:     http.New(),
		rpcSvr:      rpc.NewServer(),
		registry:    o.registry,
		client:      NewClient(o.registry),
		environment: EnvironmentHost,
	}
	svr.ctx, svr.cancelFunc = context.WithCancel(o.Context)
	return svr
}