P1-9: readiness-проба /ready отражает состояние bridge/consumer
/health остаётся liveness (процесс жив); /ready возвращает 503, если bridge или consumer не в строю (горутина упала/не поднялась). Добавлен api.Readiness с atomic-флагами, NewRouter принимает *Readiness, main поднимает/опускает флаги вокруг фоновых горутин.
This commit is contained in:
@@ -42,10 +42,17 @@
|
|||||||
`store.Flush(ctx)`, и только при успехе — `DeleteMessage`. При ошибке flush
|
`store.Flush(ctx)`, и только при успехе — `DeleteMessage`. При ошибке flush
|
||||||
сообщения НЕ удаляются (вернутся после visibility timeout).
|
сообщения НЕ удаляются (вернутся после visibility timeout).
|
||||||
|
|
||||||
|
### P1-9 — `/ready` отражает состояние подсистем
|
||||||
|
- `internal/service/api/readiness.go` — тип `Readiness` (atomic-флаги bridge/consumer).
|
||||||
|
- `internal/service/api/router.go` — `NewRouter` принимает `*Readiness`; добавлен
|
||||||
|
`/ready` (200 если обе подсистемы в строю, иначе 503). `/health` остался
|
||||||
|
liveness-пробой.
|
||||||
|
- `cmd/iot-service/main.go` — bridge/consumer поднимают/опускают свои флаги;
|
||||||
|
если горутина завершилась (ошибка/падение), `/ready` вернёт 503.
|
||||||
|
|
||||||
## Осталось (не сделано)
|
## Осталось (не сделано)
|
||||||
- P0-3 часть 1 (хэш паролей + EMQX), P0-8 (идемпотентность), P1-9 (`/health`
|
- P0-3 часть 1 (хэш паролей + EMQX), P0-8 (идемпотентность), P1-11/P1-12
|
||||||
отражает состояние подсистем), P1-11/P1-12 (тесты/CI), P2-14 (вычистить
|
(тесты/CI), P2-14 (вычистить легаси `internal/api/`).
|
||||||
легаси `internal/api/`).
|
|
||||||
- Платформенные проблемы (wss ~150с, MSS/MTU) — вне кода.
|
- Платформенные проблемы (wss ~150с, MSS/MTU) — вне кода.
|
||||||
|
|
||||||
## Проверки
|
## Проверки
|
||||||
|
|||||||
@@ -89,7 +89,9 @@ func main() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// HTTP API — в основном потоке.
|
// HTTP API — в основном потоке.
|
||||||
router := api.NewRouter(h, log, cfg.AuthTestMode, cfg.JwtHMACSecret, version)
|
// Readiness: /ready отдаёт 503, пока bridge/consumer не подняты.
|
||||||
|
ready := &api.Readiness{}
|
||||||
|
router := api.NewRouter(h, log, cfg.AuthTestMode, cfg.JwtHMACSecret, version, ready)
|
||||||
server := &http.Server{
|
server := &http.Server{
|
||||||
Addr: ":" + cfg.APIPort,
|
Addr: ":" + cfg.APIPort,
|
||||||
Handler: router,
|
Handler: router,
|
||||||
@@ -102,6 +104,8 @@ func main() {
|
|||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
|
ready.SetBridge(true)
|
||||||
|
defer ready.SetBridge(false)
|
||||||
if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) {
|
if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) {
|
||||||
log.Error("bridge stopped", "err", err)
|
log.Error("bridge stopped", "err", err)
|
||||||
}
|
}
|
||||||
@@ -109,6 +113,8 @@ func main() {
|
|||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
|
ready.SetConsumer(true)
|
||||||
|
defer ready.SetConsumer(false)
|
||||||
if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) {
|
if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) {
|
||||||
log.Error("consumer stopped", "err", err)
|
log.Error("consumer stopped", "err", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
// readiness.go — состояние подсистем для readiness-пробы /ready.
|
||||||
|
package api
|
||||||
|
|
||||||
|
import "sync/atomic"
|
||||||
|
|
||||||
|
// Readiness отслеживает, работают ли фоновые подсистемы (bridge, consumer).
|
||||||
|
// /health — liveness (процесс жив), /ready — readiness (подсистемы в строю).
|
||||||
|
type Readiness struct {
|
||||||
|
bridgeUp atomic.Bool
|
||||||
|
consumerUp atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetBridge — отметить состояние bridge.
|
||||||
|
func (r *Readiness) SetBridge(up bool) { r.bridgeUp.Store(up) }
|
||||||
|
|
||||||
|
// SetConsumer — отметить состояние consumer.
|
||||||
|
func (r *Readiness) SetConsumer(up bool) { r.consumerUp.Store(up) }
|
||||||
|
|
||||||
|
// Ready — true, когда обе подсистемы работают.
|
||||||
|
func (r *Readiness) Ready() bool {
|
||||||
|
return r.bridgeUp.Load() && r.consumerUp.Load()
|
||||||
|
}
|
||||||
@@ -26,13 +26,16 @@ func corsMiddleware(next http.Handler) http.Handler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// NewRouter собирает все маршруты сервиса.
|
// NewRouter собирает все маршруты сервиса.
|
||||||
func NewRouter(h *handler.Handler, log *slog.Logger, authTestMode bool, jwtHMACSecret string, version string) http.Handler {
|
func NewRouter(h *handler.Handler, log *slog.Logger, authTestMode bool, jwtHMACSecret string, version string, ready *Readiness) http.Handler {
|
||||||
r := mux.NewRouter()
|
r := mux.NewRouter()
|
||||||
|
|
||||||
// Health — для платформенных проверок контейнера.
|
// Health — для платформенных проверок контейнера (liveness).
|
||||||
r.HandleFunc("/health", healthHandler(version)).Methods(http.MethodGet)
|
r.HandleFunc("/health", healthHandler(version)).Methods(http.MethodGet)
|
||||||
r.HandleFunc("/healthz", healthHandler(version)).Methods(http.MethodGet)
|
r.HandleFunc("/healthz", healthHandler(version)).Methods(http.MethodGet)
|
||||||
|
|
||||||
|
// Ready — readiness: 503, если фоновые подсистемы (bridge/consumer) не в строю.
|
||||||
|
r.HandleFunc("/ready", readyHandler(ready)).Methods(http.MethodGet)
|
||||||
|
|
||||||
// Лендинг "/" — инструкция (дизайн Nubes, как у shared-sqs) + статика.
|
// Лендинг "/" — инструкция (дизайн Nubes, как у shared-sqs) + статика.
|
||||||
r.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
|
r.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
|
||||||
ServeLanding(w, req, version)
|
ServeLanding(w, req, version)
|
||||||
@@ -70,3 +73,18 @@ func healthHandler(version string) http.HandlerFunc {
|
|||||||
_, _ = w.Write([]byte(`{"status":"ok","version":"` + version + `"}`))
|
_, _ = w.Write([]byte(`{"status":"ok","version":"` + version + `"}`))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// readyHandler отдаёт 200 если все подсистемы в строю, иначе 503.
|
||||||
|
// Отделяет readiness от liveness: процесс может быть жив, но pipeline стоять.
|
||||||
|
func readyHandler(ready *Readiness) http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
if ready != nil && !ready.Ready() {
|
||||||
|
w.WriteHeader(http.StatusServiceUnavailable)
|
||||||
|
_, _ = w.Write([]byte(`{"status":"not ready"}`))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
_, _ = w.Write([]byte(`{"status":"ready"}`))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user