《游戏服务端编程实践》20.5 部署与监控全流程

// file: ch20/deploy/deploy.go
// tier: L1
// verify: go test ./ch20/deploy/

package deploy

import (
	"errors"
	"sync"
	"time"
)

var (
	ErrInvalidConfig  = errors.New("deploy: invalid probe configuration")
	ErrAlreadyStarted = errors.New("deploy: server already started")
	ErrNotStarted     = errors.New("deploy: server has not started")
	ErrNotReady       = errors.New("deploy: server is not ready")
	ErrDraining       = errors.New("deploy: server is draining")
	ErrStopped        = errors.New("deploy: server has stopped")
	ErrDrainRequired  = errors.New("deploy: drain before stop")
	ErrInflight       = errors.New("deploy: requests still in flight")
	ErrNoInflight     = errors.New("deploy: no request to complete")
	ErrStartupTimeout = errors.New("deploy: startup probe timed out")
)

// State 是进程生命周期状态，顺序与发布阶段一致。
type State int

const (
	StateNew State = iota
	StateStarting
	StateReady
	StateDraining
	StateStopped
)

// Server 用显式状态机描述“启动中、就绪、摘流、停止”四个阶段。
// 所有时间都由调用方传入，测试不需要 sleep，线上由时钟驱动。
type Server struct {
	mu             sync.Mutex
	state          State
	startedAt      time.Time
	startupTimeout time.Duration
	failureLimit   int
	failures       int
	inflight       int
	restart        chan struct{}
}

func New(startupTimeout time.Duration, failureLimit int) (*Server, error) {
	if startupTimeout <= 0 || failureLimit <= 0 {
		return nil, ErrInvalidConfig
	}
	return &Server{
		state:          StateNew,
		startupTimeout: startupTimeout,
		failureLimit:   failureLimit,
		restart:        make(chan struct{}, 1),
	}, nil
}

func (s *Server) Start(now time.Time) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.state != StateNew {
		return ErrAlreadyStarted
	}
	s.state = StateStarting
	s.startedAt = now
	return nil
}

// Startup 是启动探针：初始化未完成时返回 ErrNotReady；超过启动宽限期
// 仍未就绪，则返回 ErrStartupTimeout 并发出重启信号。
func (s *Server) Startup(now time.Time) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	switch s.state {
	case StateNew:
		return ErrNotStarted
	case StateReady, StateDraining, StateStopped:
		return nil
	}
	if now.Sub(s.startedAt) > s.startupTimeout {
		s.signalRestartLocked()
		return ErrStartupTimeout
	}
	return ErrNotReady
}

// Liveness 只回答“进程是否还活着”，不判断能否接流量。
func (s *Server) Liveness() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	switch s.state {
	case StateNew:
		return ErrNotStarted
	case StateStopped:
		return ErrStopped
	default:
		return nil
	}
}

// Readiness 回答“现在能否接新流量”。摘流开始后立即失败。
func (s *Server) Readiness() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.readinessLocked()
}

func (s *Server) readinessLocked() error {
	switch s.state {
	case StateNew:
		return ErrNotStarted
	case StateStarting:
		return ErrNotReady
	case StateReady:
		return nil
	case StateDraining:
		return ErrDraining
	default:
		return ErrStopped
	}
}

// MarkReady 由业务初始化完成后调用；重复调用是幂等的。
func (s *Server) MarkReady() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	switch s.state {
	case StateNew:
		return ErrNotStarted
	case StateStarting:
		s.state = StateReady
		return nil
	case StateReady:
		return nil
	case StateDraining:
		return ErrDraining
	default:
		return ErrStopped
	}
}

// BeginDrain 先让就绪探针失败，负载均衡停止转发新请求；已有请求继续执行。
func (s *Server) BeginDrain() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	switch s.state {
	case StateNew:
		return ErrNotStarted
	case StateStopped:
		return ErrStopped
	case StateDraining:
		return nil
	default:
		s.state = StateDraining
		return nil
	}
}

// BeginRequest 只在就绪状态计数，摘流期间拒绝新请求。
func (s *Server) BeginRequest() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.readinessLocked(); err != nil {
		return err
	}
	s.inflight++
	return nil
}

func (s *Server) EndRequest() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.inflight == 0 {
		return ErrNoInflight
	}
	s.inflight--
	return nil
}

func (s *Server) Inflight() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.inflight
}

// Stop 要求先摘流，并确认没有在途请求；顺序错误会返回明确错误，
// 而不是在还有请求时直接退出。
func (s *Server) Stop() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	switch s.state {
	case StateNew:
		return ErrNotStarted
	case StateStopped:
		return ErrStopped
	case StateStarting, StateReady:
		return ErrDrainRequired
	}
	if s.inflight > 0 {
		return ErrInflight
	}
	s.state = StateStopped
	return nil
}

// ReportProbe 记录连续探测失败。达到阈值时发出一次非阻塞重启信号，
// 并把计数清零，避免同一故障持续灌满通道。
func (s *Server) ReportProbe(success bool) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if success {
		s.failures = 0
		return false
	}
	s.failures++
	if s.failures < s.failureLimit {
		return false
	}
	s.failures = 0
	s.signalRestartLocked()
	return true
}

func (s *Server) RestartSignal() <-chan struct{} {
	return s.restart
}

func (s *Server) signalRestartLocked() {
	select {
	case s.restart <- struct{}{}:
	default:
	}
}

cd .local/checks/gs-practice-lab && gofmt -l ch20/deploy && go vet ./ch20/deploy/ && go test ./ch20/deploy/ -count=1

ok  	gspractice/ch20/deploy	0.287s