HH-442: isolate runtime processes and harden shutdown (#90)
* HH-442: isolate runtime processes and harden shutdown * HH-442: harden worker shutdown races * HH-442: gate dependency shutdown on active handlers --------- Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
Rogee
@@ -4,12 +4,26 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/gochat/gochat/internal/config"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestPrintUsage_Cov1(t *testing.T) {
|
||||
printUsage()
|
||||
}
|
||||
|
||||
func TestParseServeArgs(t *testing.T) {
|
||||
workerOnly, err := parseServeArgs(nil)
|
||||
require.NoError(t, err)
|
||||
require.False(t, workerOnly)
|
||||
|
||||
workerOnly, err = parseServeArgs([]string{"--worker-only"})
|
||||
require.NoError(t, err)
|
||||
require.True(t, workerOnly)
|
||||
|
||||
_, err = parseServeArgs([]string{"--unknown"})
|
||||
require.EqualError(t, err, `unknown serve option "--unknown"`)
|
||||
}
|
||||
|
||||
func TestShouldRunSeedMigrations_True_Cov1(t *testing.T) {
|
||||
cfg := &config.Config{Database: config.DatabaseConfig{RunMigrations: true}}
|
||||
result := shouldRunSeedMigrations(cfg)
|
||||
|
||||
@@ -35,7 +35,9 @@ func main() {
|
||||
var err error
|
||||
switch cmd {
|
||||
case "serve", "server", "run":
|
||||
err = serve()
|
||||
err = serve(os.Args[2:])
|
||||
case "worker":
|
||||
err = serve([]string{"--worker-only"})
|
||||
case "seed":
|
||||
err = seed()
|
||||
case "init":
|
||||
@@ -53,13 +55,18 @@ func main() {
|
||||
}
|
||||
|
||||
func printUsage() {
|
||||
fmt.Println("Usage: gochat [serve|seed|init]")
|
||||
fmt.Println(" serve Start the GoChat HTTP server")
|
||||
fmt.Println("Usage: gochat [serve [--worker-only]|worker|seed|init]")
|
||||
fmt.Println(" serve Start the GoChat HTTP server")
|
||||
fmt.Println(" worker Start background workers without HTTP")
|
||||
fmt.Println(" seed Create deterministic development/smoke data")
|
||||
fmt.Println(" init Initialize super admin account (interactive or via flags)")
|
||||
}
|
||||
|
||||
func serve() error {
|
||||
func serve(args []string) error {
|
||||
workerOnly, err := parseServeArgs(args)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
env := os.Getenv("GOCHAT_ENV")
|
||||
if env == "" {
|
||||
env = "development"
|
||||
@@ -68,9 +75,23 @@ func serve() error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if workerOnly {
|
||||
return application.RunWorker()
|
||||
}
|
||||
return application.Run()
|
||||
}
|
||||
|
||||
func parseServeArgs(args []string) (bool, error) {
|
||||
workerOnly := false
|
||||
for _, arg := range args {
|
||||
if arg != "--worker-only" {
|
||||
return false, fmt.Errorf("unknown serve option %q", arg)
|
||||
}
|
||||
workerOnly = true
|
||||
}
|
||||
return workerOnly, nil
|
||||
}
|
||||
|
||||
func seed() error {
|
||||
env := os.Getenv("GOCHAT_ENV")
|
||||
if env == "" {
|
||||
|
||||
@@ -2,6 +2,12 @@ server:
|
||||
host: "0.0.0.0"
|
||||
port: 3000
|
||||
mode: "debug" # debug, release, test
|
||||
read_header_timeout_seconds: 5
|
||||
read_timeout_seconds: 30
|
||||
write_timeout_seconds: 30
|
||||
idle_timeout_seconds: 120
|
||||
shutdown_timeout_seconds: 30
|
||||
max_header_bytes: 1048576
|
||||
cors:
|
||||
allowed_origins: [] # empty = Allow-Origin:* in debug mode; production must list exact origins
|
||||
# Examples:
|
||||
|
||||
@@ -17,7 +17,7 @@ groups:
|
||||
|
||||
# High error rate
|
||||
- alert: GoChatHighErrorRate
|
||||
expr: rate(http_requests_total{job="gochat", status=~"5.."}[5m]) / rate(http_requests_total{job="gochat"}[5m]) > 0.05
|
||||
expr: sum by (job, instance) (rate(http_requests_total{job="gochat", status=~"5.."}[5m])) / sum by (job, instance) (rate(http_requests_total{job="gochat"}[5m])) > 0.05
|
||||
for: 5m
|
||||
labels:
|
||||
severity: warning
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
rule_files:
|
||||
- prometheus_alerts.yml
|
||||
|
||||
evaluation_interval: 1m
|
||||
|
||||
tests:
|
||||
- interval: 1m
|
||||
input_series:
|
||||
- series: 'http_requests_total{job="gochat",instance="below-threshold",method="GET",route="/ok",status="200"}'
|
||||
values: '0+96x12'
|
||||
- series: 'http_requests_total{job="gochat",instance="below-threshold",method="GET",route="/error",status="500"}'
|
||||
values: '0+4x12'
|
||||
- series: 'http_requests_total{job="gochat",instance="above-threshold",method="GET",route="/ok",status="200"}'
|
||||
values: '0+94x12'
|
||||
- series: 'http_requests_total{job="gochat",instance="above-threshold",method="GET",route="/error",status="500"}'
|
||||
values: '0+6x12'
|
||||
alert_rule_test:
|
||||
- eval_time: 10m
|
||||
alertname: GoChatHighErrorRate
|
||||
exp_alerts:
|
||||
- exp_labels:
|
||||
instance: above-threshold
|
||||
job: gochat
|
||||
severity: warning
|
||||
exp_annotations:
|
||||
summary: GoChat error rate above 5%
|
||||
description: Error rate is 6% over the last 5 minutes.
|
||||
+121
-17
@@ -2,9 +2,14 @@ package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"os/signal"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
@@ -12,6 +17,7 @@ import (
|
||||
"github.com/gochat/gochat/internal/canned"
|
||||
"github.com/gochat/gochat/internal/config"
|
||||
ws "github.com/gochat/gochat/internal/handler/ws"
|
||||
"github.com/gochat/gochat/internal/lifecycle"
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
channelmodel "github.com/gochat/gochat/internal/model/channel"
|
||||
"github.com/gochat/gochat/internal/pubsub"
|
||||
@@ -35,6 +41,10 @@ type App struct {
|
||||
wsRelay *wspkg.BroadcastRelay
|
||||
notificationDeliverySvc *service.NotificationDeliveryService
|
||||
workerPool *worker.WorkerPool
|
||||
ready *atomic.Bool
|
||||
notificationRunning atomic.Bool
|
||||
handlerGroupOnce sync.Once
|
||||
handlerGroup *lifecycle.HandlerGroup
|
||||
}
|
||||
|
||||
// New creates and initializes the application.
|
||||
@@ -82,38 +92,115 @@ func New(cfg *config.Config) (*App, error) {
|
||||
return application, nil
|
||||
}
|
||||
|
||||
// Run starts the notification delivery pipeline and the HTTP server.
|
||||
// Run starts only the web process. Background jobs are consumed by RunWorker.
|
||||
func (a *App) Run() error {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
return a.runWeb(ctx)
|
||||
}
|
||||
|
||||
func (a *App) runWeb(ctx context.Context) error {
|
||||
lifecycle, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if a.wsHub != nil {
|
||||
go a.wsHub.Run(lifecycle)
|
||||
}
|
||||
if a.wsRelay != nil {
|
||||
if err := a.wsRelay.Start(ctx); err != nil {
|
||||
if err := a.wsRelay.Start(lifecycle); err != nil {
|
||||
return fmt.Errorf("failed to start WebSocket relay: %w", err)
|
||||
}
|
||||
defer a.wsRelay.Stop()
|
||||
}
|
||||
|
||||
server := a.HTTPServer()
|
||||
serveErr := make(chan error, 1)
|
||||
go func() {
|
||||
log.Printf("Starting gochat web server on %s", server.Addr)
|
||||
serveErr <- server.ListenAndServe()
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-serveErr:
|
||||
cancel()
|
||||
_ = a.Shutdown(a.shutdownTimeout())
|
||||
if errors.Is(err, http.ErrServerClosed) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
}
|
||||
|
||||
if a.ready != nil {
|
||||
a.ready.Store(false)
|
||||
}
|
||||
timeout := a.shutdownTimeout()
|
||||
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer shutdownCancel()
|
||||
httpErr := server.Shutdown(shutdownCtx)
|
||||
cancel()
|
||||
return errors.Join(httpErr, a.shutdown(shutdownCtx))
|
||||
}
|
||||
|
||||
// RunWorker consumes background jobs and notifications without opening an HTTP listener.
|
||||
func (a *App) RunWorker() error {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
return a.runWorker(ctx)
|
||||
}
|
||||
|
||||
func (a *App) runWorker(ctx context.Context) error {
|
||||
if a.workerPool != nil {
|
||||
if err := a.workerPool.Start(); err != nil {
|
||||
return fmt.Errorf("failed to start background worker: %w", err)
|
||||
}
|
||||
applogger.L().Info("Background worker started")
|
||||
}
|
||||
|
||||
// Start notification delivery service (Watermill router) in background
|
||||
notificationErr := make(chan error, 1)
|
||||
if a.notificationDeliverySvc != nil {
|
||||
go func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
if err := a.notificationDeliverySvc.Start(ctx); err != nil {
|
||||
applogger.L().Errorf("Notification delivery service start error: %v", err)
|
||||
}
|
||||
}()
|
||||
a.notificationRunning.Store(true)
|
||||
go func() { notificationErr <- a.notificationDeliverySvc.Start(a.handlers().Context()) }()
|
||||
applogger.L().Info("Notification delivery pipeline started")
|
||||
}
|
||||
|
||||
addr := fmt.Sprintf("%s:%d", a.config.Server.Host, a.config.Server.Port)
|
||||
log.Printf("Starting gochat server on %s", addr)
|
||||
return a.engine.Run(addr)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case err := <-notificationErr:
|
||||
if err != nil {
|
||||
return errors.Join(err, a.Shutdown(a.shutdownTimeout()))
|
||||
}
|
||||
}
|
||||
return a.Shutdown(a.shutdownTimeout())
|
||||
}
|
||||
|
||||
// HTTPServer applies the production request-boundary limits from configuration.
|
||||
func (a *App) HTTPServer() *http.Server {
|
||||
cfg := a.config.Server
|
||||
return &http.Server{
|
||||
Addr: a.Address(),
|
||||
Handler: a.Handler(),
|
||||
ReadHeaderTimeout: positiveDuration(cfg.ReadHeaderTimeoutS, 5*time.Second),
|
||||
ReadTimeout: positiveDuration(cfg.ReadTimeoutS, 30*time.Second),
|
||||
WriteTimeout: positiveDuration(cfg.WriteTimeoutS, 30*time.Second),
|
||||
IdleTimeout: positiveDuration(cfg.IdleTimeoutS, 120*time.Second),
|
||||
MaxHeaderBytes: positiveInt(cfg.MaxHeaderBytes, 1<<20),
|
||||
}
|
||||
}
|
||||
|
||||
func (a *App) shutdownTimeout() time.Duration {
|
||||
return positiveDuration(a.config.Server.ShutdownTimeoutS, 30*time.Second)
|
||||
}
|
||||
|
||||
func positiveDuration(seconds int, fallback time.Duration) time.Duration {
|
||||
if seconds <= 0 {
|
||||
return fallback
|
||||
}
|
||||
return time.Duration(seconds) * time.Second
|
||||
}
|
||||
|
||||
func positiveInt(value, fallback int) int {
|
||||
if value <= 0 {
|
||||
return fallback
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
// Address returns the server listen address.
|
||||
@@ -123,7 +210,24 @@ func (a *App) Address() string {
|
||||
|
||||
// Handler returns the gin engine as http.Handler for use with http.Server.
|
||||
func (a *App) Handler() http.Handler {
|
||||
return a.engine
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
handlers := a.handlers()
|
||||
if !handlers.Begin() {
|
||||
http.Error(w, http.StatusText(http.StatusServiceUnavailable), http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
defer handlers.Done()
|
||||
a.engine.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
func (a *App) handlers() *lifecycle.HandlerGroup {
|
||||
a.handlerGroupOnce.Do(func() {
|
||||
if a.handlerGroup == nil {
|
||||
a.handlerGroup = lifecycle.NewHandlerGroup()
|
||||
}
|
||||
})
|
||||
return a.handlerGroup
|
||||
}
|
||||
|
||||
// Config returns the application config.
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
@@ -28,6 +29,7 @@ import (
|
||||
whatsappchannel "github.com/gochat/gochat/internal/channel/whatsapp"
|
||||
"github.com/gochat/gochat/internal/config"
|
||||
"github.com/gochat/gochat/internal/database"
|
||||
basehandler "github.com/gochat/gochat/internal/handler"
|
||||
v1 "github.com/gochat/gochat/internal/handler/api/v1"
|
||||
webhook "github.com/gochat/gochat/internal/handler/webhook"
|
||||
widget "github.com/gochat/gochat/internal/handler/widget"
|
||||
@@ -944,10 +946,16 @@ func Bootstrap(env string) (*App, error) {
|
||||
// (ref: Chatwoot Rails middleware stack in config/application.rb)
|
||||
gin.SetMode(cfg.Server.Mode)
|
||||
engine := gin.New()
|
||||
startedAt := time.Now()
|
||||
httpMetrics := basehandler.NewHTTPMetrics(startedAt)
|
||||
ready := &atomic.Bool{}
|
||||
ready.Store(true)
|
||||
readinessChecks := newReadinessChecks(cfg, db, rdb, ready)
|
||||
|
||||
// Global middleware — applies to ALL routes
|
||||
corsMiddleware := middleware.CORS(middleware.CORSConfigFromAppConfig(cfg))
|
||||
engine.Use(middleware.Recovery()) // panic recovery
|
||||
engine.Use(middleware.Recovery()) // panic recovery
|
||||
engine.Use(httpMetrics.Middleware())
|
||||
engine.Use(middleware.RequestLogger()) // structured request logging
|
||||
engine.Use(middleware.RateLimit(rdb)) // rate limiting (ref: Chatwoot rack-attack)
|
||||
engine.Use(corsMiddleware) // CORS with configurable whitelist
|
||||
@@ -967,6 +975,10 @@ func Bootstrap(env string) (*App, error) {
|
||||
sessionMwCfg := middleware.SessionMiddlewareConfigFromAppConfig(cfg)
|
||||
engine.Use(middleware.SessionMiddleware(sessionStore, sessionMwCfg))
|
||||
}
|
||||
engine.GET("/health", basehandler.HealthHandler(db, startedAt, config.Version, readinessChecks...))
|
||||
engine.GET("/ready", basehandler.ReadyHandler(db, readinessChecks...))
|
||||
engine.GET("/live", basehandler.LiveHandler())
|
||||
engine.GET("/metrics", httpMetrics.Handler())
|
||||
|
||||
// Step 11: WebSocket hub already created above (before handler wiring)
|
||||
// so handlers can reference it via the hubTypingAdapter.
|
||||
@@ -994,7 +1006,7 @@ func Bootstrap(env string) (*App, error) {
|
||||
|
||||
applogger.L().Info("All dependencies wired successfully")
|
||||
|
||||
return &App{
|
||||
application := &App{
|
||||
config: cfg,
|
||||
reloader: reloader,
|
||||
db: db,
|
||||
@@ -1004,7 +1016,10 @@ func Bootstrap(env string) (*App, error) {
|
||||
wsRelay: wsRelay,
|
||||
notificationDeliverySvc: notificationDeliverySvc,
|
||||
workerPool: workerPool,
|
||||
}, nil
|
||||
ready: ready,
|
||||
}
|
||||
notificationDeliverySvc.SetHandlerGroup(application.handlers())
|
||||
return application, nil
|
||||
}
|
||||
|
||||
func validateStartupMigrations(env string, cfg *config.Config) error {
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/gochat/gochat/internal/config"
|
||||
"github.com/gochat/gochat/internal/database"
|
||||
basehandler "github.com/gochat/gochat/internal/handler"
|
||||
"github.com/gochat/gochat/internal/search"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func newReadinessChecks(cfg *config.Config, db *gorm.DB, rdb redis.UniversalClient, ready *atomic.Bool) []basehandler.DependencyCheck {
|
||||
latestMigration, latestMigrationErr := database.LatestVersion(cfg.Database.GetMigrationsPath())
|
||||
checks := []basehandler.DependencyCheck{
|
||||
{Name: "redis", Check: func(ctx context.Context) error { return rdb.Ping(ctx).Err() }},
|
||||
{Name: "migrations", Check: func(ctx context.Context) error {
|
||||
if latestMigrationErr != nil {
|
||||
return latestMigrationErr
|
||||
}
|
||||
return database.CheckVersion(ctx, db, latestMigration)
|
||||
}},
|
||||
{Name: "draining", Check: func(context.Context) error {
|
||||
if !ready.Load() {
|
||||
return fmt.Errorf("shutdown in progress")
|
||||
}
|
||||
return nil
|
||||
}},
|
||||
}
|
||||
if !strings.EqualFold(cfg.Search.Engine, search.EngineMeilisearch) {
|
||||
return checks
|
||||
}
|
||||
searchClient := &http.Client{Timeout: time.Duration(cfg.Search.TimeoutSeconds) * time.Second}
|
||||
return append(checks, basehandler.DependencyCheck{Name: "search", Check: func(ctx context.Context) error {
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(cfg.Search.Host, "/")+"/health", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
response, err := searchClient.Do(request)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
_, _ = io.Copy(io.Discard, response.Body)
|
||||
if response.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("search health returned %s", response.Status)
|
||||
}
|
||||
return nil
|
||||
}})
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/gochat/gochat/internal/config"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func TestReadinessChecksRequiredDependenciesAndDrainState(t *testing.T) {
|
||||
migrations := t.TempDir()
|
||||
require.NoError(t, os.WriteFile(filepath.Join(migrations, "000001_initial.up.sql"), []byte("SELECT 1;"), 0o600))
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.Exec("CREATE TABLE schema_migrations (version INTEGER NOT NULL, dirty BOOLEAN NOT NULL)").Error)
|
||||
require.NoError(t, db.Exec("INSERT INTO schema_migrations(version, dirty) VALUES (1, false)").Error)
|
||||
mini := miniredis.RunT(t)
|
||||
rdb := redis.NewClient(&redis.Options{Addr: mini.Addr()})
|
||||
t.Cleanup(func() { _ = rdb.Close() })
|
||||
search := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }))
|
||||
defer search.Close()
|
||||
ready := &atomic.Bool{}
|
||||
ready.Store(true)
|
||||
cfg := &config.Config{
|
||||
Database: config.DatabaseConfig{MigrationsPath: migrations},
|
||||
Search: config.SearchConfig{Engine: "meilisearch", Host: search.URL, TimeoutSeconds: 1},
|
||||
}
|
||||
|
||||
checks := newReadinessChecks(cfg, db, rdb, ready)
|
||||
require.Len(t, checks, 4)
|
||||
byName := map[string]func(context.Context) error{}
|
||||
for _, check := range checks {
|
||||
byName[check.Name] = check.Check
|
||||
require.NoError(t, check.Check(context.Background()), check.Name)
|
||||
}
|
||||
ready.Store(false)
|
||||
require.ErrorContains(t, byName["draining"](context.Background()), "shutdown")
|
||||
mini.Close()
|
||||
require.Error(t, byName["redis"](context.Background()))
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/gochat/gochat/internal/config"
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/pubsub"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type shutdownOrderPubSub struct {
|
||||
db *gorm.DB
|
||||
jobID uint
|
||||
status string
|
||||
err error
|
||||
}
|
||||
|
||||
type shutdownRedisPubSub struct{ client *redis.Client }
|
||||
|
||||
func (p *shutdownRedisPubSub) Publish(ctx context.Context, topic string, _ pubsub.Event) error {
|
||||
return p.client.Set(ctx, topic, "available", 0).Err()
|
||||
}
|
||||
func (*shutdownRedisPubSub) Subscribe(context.Context, string, pubsub.EventHandler) error {
|
||||
return nil
|
||||
}
|
||||
func (*shutdownRedisPubSub) Unsubscribe(context.Context, string) error { return nil }
|
||||
func (p *shutdownRedisPubSub) Close() error { return p.client.Close() }
|
||||
|
||||
func (*shutdownOrderPubSub) Publish(context.Context, string, pubsub.Event) error { return nil }
|
||||
func (*shutdownOrderPubSub) Subscribe(context.Context, string, pubsub.EventHandler) error {
|
||||
return nil
|
||||
}
|
||||
func (*shutdownOrderPubSub) Unsubscribe(context.Context, string) error { return nil }
|
||||
func (p *shutdownOrderPubSub) Close() error {
|
||||
var job model.BackgroundJob
|
||||
p.err = p.db.First(&job, p.jobID).Error
|
||||
p.status = job.Status
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestWebProcessDoesNotConsumeJobs(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.AutoMigrate(&model.BackgroundJob{}))
|
||||
pool := worker.NewWorkerPool(db)
|
||||
var handled atomic.Int32
|
||||
pool.Register("web_must_not_consume", func(context.Context, *model.BackgroundJob) error {
|
||||
handled.Add(1)
|
||||
return nil
|
||||
})
|
||||
_, err = pool.Enqueue(context.Background(), "web_must_not_consume", nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
engine := gin.New()
|
||||
engine.GET("/live", func(c *gin.Context) { c.Status(http.StatusOK) })
|
||||
application := &App{
|
||||
config: &config.Config{Server: config.ServerConfig{
|
||||
Host: "127.0.0.1", Port: 0, ReadHeaderTimeoutS: 1, ReadTimeoutS: 1,
|
||||
WriteTimeoutS: 1, IdleTimeoutS: 1, ShutdownTimeoutS: 1, MaxHeaderBytes: 1024,
|
||||
}},
|
||||
db: db, engine: engine, workerPool: pool,
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
cancel()
|
||||
}()
|
||||
require.NoError(t, application.runWeb(ctx))
|
||||
require.Zero(t, handled.Load())
|
||||
}
|
||||
|
||||
func TestWorkerProcessConsumesJobsWithoutHTTP(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.AutoMigrate(&model.BackgroundJob{}))
|
||||
pool := worker.NewWorkerPoolWithOptions(db, worker.WithPollInterval(5*time.Millisecond))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
pool.Register("worker_only", func(context.Context, *model.BackgroundJob) error {
|
||||
cancel()
|
||||
return nil
|
||||
})
|
||||
_, err = pool.Enqueue(context.Background(), "worker_only", nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
occupied, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
defer occupied.Close()
|
||||
port := occupied.Addr().(*net.TCPAddr).Port
|
||||
application := &App{
|
||||
config: &config.Config{Server: config.ServerConfig{Host: "127.0.0.1", Port: port, ShutdownTimeoutS: 1}},
|
||||
db: db, engine: gin.New(), workerPool: pool,
|
||||
}
|
||||
require.NoError(t, application.runWorker(ctx))
|
||||
}
|
||||
|
||||
func TestShutdownWaitsForCancelledJobPersistenceBeforeClosingDependencies(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open("file:shutdown-order?mode=memory&cache=shared"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.AutoMigrate(&model.BackgroundJob{}))
|
||||
pool := worker.NewWorkerPoolWithOptions(db,
|
||||
worker.WithPollInterval(5*time.Millisecond),
|
||||
worker.WithBackoff(func(int) time.Duration { return 0 }),
|
||||
)
|
||||
started := make(chan struct{})
|
||||
pool.Register("shutdown_order", func(ctx context.Context, _ *model.BackgroundJob) error {
|
||||
close(started)
|
||||
<-ctx.Done()
|
||||
return ctx.Err()
|
||||
})
|
||||
job, err := pool.Enqueue(context.Background(), "shutdown_order", nil)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, pool.Start())
|
||||
<-started
|
||||
|
||||
ps := &shutdownOrderPubSub{db: db, jobID: job.ID}
|
||||
application := &App{db: db, pubsub: ps, workerPool: pool}
|
||||
require.ErrorIs(t, application.Shutdown(20*time.Millisecond), context.DeadlineExceeded)
|
||||
require.NoError(t, ps.err)
|
||||
require.Equal(t, model.BackgroundJobStatusRetrying, ps.status)
|
||||
}
|
||||
|
||||
func TestHTTPShutdownWaitsForSlowHandlerBeforeClosingDependencies(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open("file:http-shutdown-order?mode=memory&cache=shared"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
redisServer := miniredis.RunT(t)
|
||||
redisClient := redis.NewClient(&redis.Options{Addr: redisServer.Addr()})
|
||||
ps := &shutdownRedisPubSub{client: redisClient}
|
||||
|
||||
started := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
handlerErr := make(chan error, 1)
|
||||
engine := gin.New()
|
||||
application := &App{db: db, pubsub: ps, engine: engine}
|
||||
engine.GET("/slow", func(c *gin.Context) {
|
||||
close(started)
|
||||
<-release
|
||||
var one int
|
||||
dbErr := db.Raw("SELECT 1").Scan(&one).Error
|
||||
redisErr := ps.Publish(c.Request.Context(), "shutdown-order", pubsub.Event{Type: "still-open"})
|
||||
handlerErr <- errors.Join(dbErr, redisErr)
|
||||
c.Status(http.StatusNoContent)
|
||||
})
|
||||
|
||||
server := httptest.NewServer(application.Handler())
|
||||
defer server.Close()
|
||||
requestDone := make(chan error, 1)
|
||||
go func() {
|
||||
response, requestErr := http.Get(server.URL + "/slow")
|
||||
if requestErr == nil {
|
||||
_ = response.Body.Close()
|
||||
}
|
||||
requestDone <- requestErr
|
||||
}()
|
||||
<-started
|
||||
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
|
||||
defer cancel()
|
||||
require.ErrorIs(t, server.Config.Shutdown(shutdownCtx), context.DeadlineExceeded)
|
||||
shutdownDone := make(chan error, 1)
|
||||
go func() { shutdownDone <- application.shutdown(shutdownCtx) }()
|
||||
select {
|
||||
case err := <-shutdownDone:
|
||||
t.Fatalf("dependencies closed before slow handler completed: %v", err)
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
|
||||
close(release)
|
||||
require.NoError(t, <-handlerErr)
|
||||
require.NoError(t, <-requestDone)
|
||||
require.NoError(t, <-shutdownDone)
|
||||
|
||||
sqlDB, err := db.DB()
|
||||
require.NoError(t, err)
|
||||
require.Error(t, sqlDB.Ping())
|
||||
require.Error(t, ps.Publish(context.Background(), "shutdown-order", pubsub.Event{Type: "closed"}))
|
||||
}
|
||||
|
||||
func TestHTTPServerLimits(t *testing.T) {
|
||||
engine := gin.New()
|
||||
engine.GET("/", func(c *gin.Context) { c.Status(http.StatusOK) })
|
||||
application := &App{config: &config.Config{Server: config.ServerConfig{
|
||||
Host: "127.0.0.1", Port: 0, ReadHeaderTimeoutS: 2, ReadTimeoutS: 3,
|
||||
WriteTimeoutS: 4, IdleTimeoutS: 5, MaxHeaderBytes: 1024,
|
||||
}}, engine: engine}
|
||||
server := application.HTTPServer()
|
||||
require.Equal(t, 2*time.Second, server.ReadHeaderTimeout)
|
||||
require.Equal(t, 3*time.Second, server.ReadTimeout)
|
||||
require.Equal(t, 4*time.Second, server.WriteTimeout)
|
||||
require.Equal(t, 5*time.Second, server.IdleTimeout)
|
||||
require.Equal(t, 1024, server.MaxHeaderBytes)
|
||||
|
||||
server.ReadHeaderTimeout = 50 * time.Millisecond
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
serveDone := make(chan error, 1)
|
||||
go func() { serveDone <- server.Serve(listener) }()
|
||||
|
||||
slow, err := net.Dial("tcp", listener.Addr().String())
|
||||
require.NoError(t, err)
|
||||
_, err = slow.Write([]byte("GET / HTTP/1.1\r\nHost: localhost\r\nX-Slow:"))
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, slow.SetReadDeadline(time.Now().Add(500*time.Millisecond)))
|
||||
started := time.Now()
|
||||
line, readErr := bufio.NewReader(slow).ReadString('\n')
|
||||
require.Less(t, time.Since(started), 300*time.Millisecond)
|
||||
require.True(t, readErr != nil || strings.Contains(line, "400"), "expected timeout rejection, got %q (%v)", line, readErr)
|
||||
require.NoError(t, slow.Close())
|
||||
|
||||
large, err := net.Dial("tcp", listener.Addr().String())
|
||||
require.NoError(t, err)
|
||||
_, err = large.Write([]byte("GET / HTTP/1.1\r\nHost: localhost\r\nX-Large: " + strings.Repeat("a", 10_000) + "\r\n\r\n"))
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, large.SetReadDeadline(time.Now().Add(time.Second)))
|
||||
status, err := bufio.NewReader(large).ReadString('\n')
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, status, "431")
|
||||
require.NoError(t, large.Close())
|
||||
|
||||
require.NoError(t, server.Shutdown(context.Background()))
|
||||
require.ErrorIs(t, <-serveDone, http.ErrServerClosed)
|
||||
}
|
||||
@@ -2,9 +2,8 @@ package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
applogger "github.com/gochat/gochat/pkg/logger"
|
||||
@@ -14,15 +13,27 @@ import (
|
||||
// Pattern follows Chatwoot's Puma graceful shutdown (config/puma.rb)
|
||||
// and Rails signal handling (SIGTERM -> graceful stop).
|
||||
// Order: 1. Stop accepting new connections (done by http.Server.Shutdown in main)
|
||||
// 2. Stop config hot-reloader
|
||||
// 3. Close notification delivery service (Watermill router + subscriber)
|
||||
// 4. Close WebSocket Hub (disconnect all clients)
|
||||
// 5. Close PubSub (stop message publishing/consuming)
|
||||
// 6. Close database connection pool
|
||||
// 7. Flush logger buffers
|
||||
// 2. Stop config hot-reloader
|
||||
// 3. Close notification delivery service (Watermill router + subscriber)
|
||||
// 4. Close WebSocket Hub (disconnect all clients)
|
||||
// 5. Close PubSub (stop message publishing/consuming)
|
||||
// 6. Close database connection pool
|
||||
// 7. Flush logger buffers
|
||||
func (a *App) Shutdown(timeout time.Duration) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
return a.shutdown(ctx)
|
||||
}
|
||||
|
||||
func (a *App) shutdown(ctx context.Context) error {
|
||||
start := time.Now()
|
||||
applogger.L().Info("Shutting down GoChat application...")
|
||||
if a.ready != nil {
|
||||
a.ready.Store(false)
|
||||
}
|
||||
var shutdownErrs []error
|
||||
handlers := a.handlers()
|
||||
handlers.Stop()
|
||||
|
||||
// Step 0: Stop config hot-reloader
|
||||
if a.reloader != nil {
|
||||
@@ -30,29 +41,44 @@ func (a *App) Shutdown(timeout time.Duration) error {
|
||||
applogger.L().Info("Config hot-reloader stopped")
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
// Stop claiming new jobs first and let active handlers drain.
|
||||
if a.workerPool != nil {
|
||||
if err := a.workerPool.Shutdown(ctx); err != nil {
|
||||
shutdownErrs = append(shutdownErrs, fmt.Errorf("worker shutdown: %w", err))
|
||||
}
|
||||
}
|
||||
|
||||
// Step 0.5: Close notification delivery service (Watermill router + subscriber)
|
||||
if a.notificationDeliverySvc != nil {
|
||||
if a.notificationDeliverySvc != nil && a.notificationRunning.Swap(false) {
|
||||
if closeErr := a.notificationDeliverySvc.Close(); closeErr != nil {
|
||||
applogger.L().Errorf("Notification delivery service close error: %v", closeErr)
|
||||
shutdownErrs = append(shutdownErrs, closeErr)
|
||||
} else {
|
||||
applogger.L().Info("Notification delivery service closed")
|
||||
}
|
||||
}
|
||||
|
||||
// Router.Close may time out while HTTP or notification handlers still use
|
||||
// shared dependencies. Keep those dependencies alive until every handler exits.
|
||||
handlers.Wait()
|
||||
|
||||
// Step 1: Close WebSocket Hub — disconnect all connected clients
|
||||
if a.wsHub != nil {
|
||||
a.wsHub.Shutdown(ctx)
|
||||
applogger.L().Info("WebSocket hub closed")
|
||||
}
|
||||
if a.wsRelay != nil {
|
||||
if err := a.wsRelay.Stop(); err != nil {
|
||||
shutdownErrs = append(shutdownErrs, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Step 2: Close PubSub — stop event publishing and consuming
|
||||
if a.pubsub != nil {
|
||||
if closer, ok := a.pubsub.(interface{ Close() error }); ok {
|
||||
if closeErr := closer.Close(); closeErr != nil {
|
||||
applogger.L().Errorf("PubSub close error: %v", closeErr)
|
||||
shutdownErrs = append(shutdownErrs, closeErr)
|
||||
} else {
|
||||
applogger.L().Info("PubSub closed")
|
||||
}
|
||||
@@ -65,6 +91,7 @@ func (a *App) Shutdown(timeout time.Duration) error {
|
||||
if err == nil {
|
||||
if closeErr := sqlDB.Close(); closeErr != nil {
|
||||
applogger.L().Errorf("Database close error: %v", closeErr)
|
||||
shutdownErrs = append(shutdownErrs, closeErr)
|
||||
} else {
|
||||
applogger.L().Info("Database connection pool closed")
|
||||
}
|
||||
@@ -77,13 +104,5 @@ func (a *App) Shutdown(timeout time.Duration) error {
|
||||
duration := time.Since(start)
|
||||
applogger.L().Infof("GoChat shutdown complete (took %v)", duration)
|
||||
|
||||
return nil
|
||||
return errors.Join(shutdownErrs...)
|
||||
}
|
||||
|
||||
// WaitForShutdownSignal blocks until a termination signal is received.
|
||||
func WaitForShutdownSignal() {
|
||||
quit := make(chan os.Signal, 1)
|
||||
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
||||
<-quit
|
||||
applogger.L().Info("Received shutdown signal")
|
||||
}
|
||||
@@ -84,10 +84,16 @@ type OAuthConfig struct {
|
||||
}
|
||||
|
||||
type ServerConfig struct {
|
||||
Host string `mapstructure:"host"`
|
||||
Port int `mapstructure:"port"`
|
||||
Mode string `mapstructure:"mode"` // debug, release, test
|
||||
CORS CORSConfig `mapstructure:"cors"`
|
||||
Host string `mapstructure:"host"`
|
||||
Port int `mapstructure:"port"`
|
||||
Mode string `mapstructure:"mode"` // debug, release, test
|
||||
ReadHeaderTimeoutS int `mapstructure:"read_header_timeout_seconds"`
|
||||
ReadTimeoutS int `mapstructure:"read_timeout_seconds"`
|
||||
WriteTimeoutS int `mapstructure:"write_timeout_seconds"`
|
||||
IdleTimeoutS int `mapstructure:"idle_timeout_seconds"`
|
||||
ShutdownTimeoutS int `mapstructure:"shutdown_timeout_seconds"`
|
||||
MaxHeaderBytes int `mapstructure:"max_header_bytes"`
|
||||
CORS CORSConfig `mapstructure:"cors"`
|
||||
}
|
||||
|
||||
// CORSConfig holds CORS middleware configuration.
|
||||
@@ -474,40 +480,46 @@ func LoadWithEnv(env string) (*Config, error) {
|
||||
// Bind specific env keys that viper can't auto-infer for nested structs
|
||||
// These are common overrides that users set via environment variables
|
||||
envBindings := map[string]string{
|
||||
"GOCHAT_SERVER_HOST": "server.host",
|
||||
"GOCHAT_SERVER_PORT": "server.port",
|
||||
"GOCHAT_SERVER_MODE": "server.mode",
|
||||
"GOCHAT_SERVER_CORS_ALLOWED_ORIGINS": "server.cors.allowed_origins",
|
||||
"GOCHAT_DATABASE_DSN": "database.dsn",
|
||||
"GOCHAT_DATABASE_MAX_IDLE_CONNS": "database.max_idle_conns",
|
||||
"GOCHAT_DATABASE_MAX_OPEN_CONNS": "database.max_open_conns",
|
||||
"GOCHAT_DATABASE_CONN_MAX_LIFETIME": "database.conn_max_lifetime",
|
||||
"GOCHAT_DATABASE_RUN_MIGRATIONS": "database.run_migrations",
|
||||
"GOCHAT_DATABASE_MIGRATIONS_PATH": "database.migrations_path",
|
||||
"GOCHAT_REDIS_DSN": "redis.dsn",
|
||||
"GOCHAT_REDIS_POOL_SIZE": "redis.pool_size",
|
||||
"GOCHAT_JWT_SECRET": "jwt.secret",
|
||||
"JWT_SECRET": "jwt.secret", // Alias for compatibility (no prefix)
|
||||
"GOCHAT_JWT_PREVIOUS_SECRETS": "jwt.previous_secrets",
|
||||
"GOCHAT_JWT_ALLOW_INSECURE_HEADER_AUTH": "jwt.allow_insecure_header_auth",
|
||||
"GOCHAT_JWT_EXPIRY_HOURS": "jwt.expiry_hours",
|
||||
"GOCHAT_JWT_ACCESS_EXPIRY_MINUTES": "jwt.access_expiry_minutes",
|
||||
"GOCHAT_JWT_REFRESH_EXPIRY_HOURS": "jwt.refresh_expiry_hours",
|
||||
"GOCHAT_LOG_LEVEL": "log.level",
|
||||
"GOCHAT_LOG_FORMAT": "log.format",
|
||||
"GOCHAT_WORKER_CONCURRENCY": "worker.concurrency",
|
||||
"GOCHAT_WORKER_REDIS_STREAM_PREFIX": "worker.redis_stream_prefix",
|
||||
"GOCHAT_WORKER_REDIS_CONSUMER_GROUP": "worker.redis_consumer_group",
|
||||
"GOCHAT_WORKER_REDIS_BLOCK_TIMEOUT_S": "worker.redis_block_timeout_s",
|
||||
"GOCHAT_WORKER_REDIS_SWEEP_INTERVAL_S": "worker.redis_sweep_interval_s",
|
||||
"GOCHAT_SEARCH_ENGINE": "search.engine",
|
||||
"GOCHAT_SEARCH_HOST": "search.host",
|
||||
"GOCHAT_SEARCH_API_KEY": "search.api_key",
|
||||
"GOCHAT_SEARCH_INDEX_PREFIX": "search.index_prefix",
|
||||
"GOCHAT_SEARCH_TIMEOUT_SECONDS": "search.timeout_seconds",
|
||||
"GOCHAT_STORAGE_PROVIDER": "storage.provider",
|
||||
"GOCHAT_STORAGE_LOCAL_PATH": "storage.local_path",
|
||||
"GOCHAT_STORAGE_MAX_FILE_SIZE": "storage.max_file_size",
|
||||
"GOCHAT_SERVER_HOST": "server.host",
|
||||
"GOCHAT_SERVER_PORT": "server.port",
|
||||
"GOCHAT_SERVER_MODE": "server.mode",
|
||||
"GOCHAT_SERVER_READ_HEADER_TIMEOUT_SECONDS": "server.read_header_timeout_seconds",
|
||||
"GOCHAT_SERVER_READ_TIMEOUT_SECONDS": "server.read_timeout_seconds",
|
||||
"GOCHAT_SERVER_WRITE_TIMEOUT_SECONDS": "server.write_timeout_seconds",
|
||||
"GOCHAT_SERVER_IDLE_TIMEOUT_SECONDS": "server.idle_timeout_seconds",
|
||||
"GOCHAT_SERVER_SHUTDOWN_TIMEOUT_SECONDS": "server.shutdown_timeout_seconds",
|
||||
"GOCHAT_SERVER_MAX_HEADER_BYTES": "server.max_header_bytes",
|
||||
"GOCHAT_SERVER_CORS_ALLOWED_ORIGINS": "server.cors.allowed_origins",
|
||||
"GOCHAT_DATABASE_DSN": "database.dsn",
|
||||
"GOCHAT_DATABASE_MAX_IDLE_CONNS": "database.max_idle_conns",
|
||||
"GOCHAT_DATABASE_MAX_OPEN_CONNS": "database.max_open_conns",
|
||||
"GOCHAT_DATABASE_CONN_MAX_LIFETIME": "database.conn_max_lifetime",
|
||||
"GOCHAT_DATABASE_RUN_MIGRATIONS": "database.run_migrations",
|
||||
"GOCHAT_DATABASE_MIGRATIONS_PATH": "database.migrations_path",
|
||||
"GOCHAT_REDIS_DSN": "redis.dsn",
|
||||
"GOCHAT_REDIS_POOL_SIZE": "redis.pool_size",
|
||||
"GOCHAT_JWT_SECRET": "jwt.secret",
|
||||
"JWT_SECRET": "jwt.secret", // Alias for compatibility (no prefix)
|
||||
"GOCHAT_JWT_PREVIOUS_SECRETS": "jwt.previous_secrets",
|
||||
"GOCHAT_JWT_ALLOW_INSECURE_HEADER_AUTH": "jwt.allow_insecure_header_auth",
|
||||
"GOCHAT_JWT_EXPIRY_HOURS": "jwt.expiry_hours",
|
||||
"GOCHAT_JWT_ACCESS_EXPIRY_MINUTES": "jwt.access_expiry_minutes",
|
||||
"GOCHAT_JWT_REFRESH_EXPIRY_HOURS": "jwt.refresh_expiry_hours",
|
||||
"GOCHAT_LOG_LEVEL": "log.level",
|
||||
"GOCHAT_LOG_FORMAT": "log.format",
|
||||
"GOCHAT_WORKER_CONCURRENCY": "worker.concurrency",
|
||||
"GOCHAT_WORKER_REDIS_STREAM_PREFIX": "worker.redis_stream_prefix",
|
||||
"GOCHAT_WORKER_REDIS_CONSUMER_GROUP": "worker.redis_consumer_group",
|
||||
"GOCHAT_WORKER_REDIS_BLOCK_TIMEOUT_S": "worker.redis_block_timeout_s",
|
||||
"GOCHAT_WORKER_REDIS_SWEEP_INTERVAL_S": "worker.redis_sweep_interval_s",
|
||||
"GOCHAT_SEARCH_ENGINE": "search.engine",
|
||||
"GOCHAT_SEARCH_HOST": "search.host",
|
||||
"GOCHAT_SEARCH_API_KEY": "search.api_key",
|
||||
"GOCHAT_SEARCH_INDEX_PREFIX": "search.index_prefix",
|
||||
"GOCHAT_SEARCH_TIMEOUT_SECONDS": "search.timeout_seconds",
|
||||
"GOCHAT_STORAGE_PROVIDER": "storage.provider",
|
||||
"GOCHAT_STORAGE_LOCAL_PATH": "storage.local_path",
|
||||
"GOCHAT_STORAGE_MAX_FILE_SIZE": "storage.max_file_size",
|
||||
// G10: OAuth config for new channel integrations (Twitter, Microsoft, Google)
|
||||
"GOCHAT_OAUTH_TWITTER_CLIENT_ID": "oauth.twitter.client_id",
|
||||
"GOCHAT_OAUTH_TWITTER_CLIENT_SECRET": "oauth.twitter.client_secret",
|
||||
@@ -698,6 +710,12 @@ func setDefaults(v *viper.Viper) {
|
||||
v.SetDefault("server.host", "0.0.0.0")
|
||||
v.SetDefault("server.port", 3000)
|
||||
v.SetDefault("server.mode", "debug")
|
||||
v.SetDefault("server.read_header_timeout_seconds", 5)
|
||||
v.SetDefault("server.read_timeout_seconds", 30)
|
||||
v.SetDefault("server.write_timeout_seconds", 30)
|
||||
v.SetDefault("server.idle_timeout_seconds", 120)
|
||||
v.SetDefault("server.shutdown_timeout_seconds", 30)
|
||||
v.SetDefault("server.max_header_bytes", 1<<20)
|
||||
|
||||
v.SetDefault("database.max_idle_conns", 10)
|
||||
v.SetDefault("database.max_open_conns", 100)
|
||||
|
||||
@@ -1,15 +1,19 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/golang-migrate/migrate/v4"
|
||||
_ "github.com/golang-migrate/migrate/v4/database/postgres"
|
||||
_ "github.com/golang-migrate/migrate/v4/database/sqlite3"
|
||||
_ "github.com/golang-migrate/migrate/v4/source/file"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
var unsupportedPQEnvironmentKeys = []string{"PGSERVICE", "PGSERVICEFILE", "PGREALM"}
|
||||
@@ -164,6 +168,54 @@ func CurrentVersion(dbURL string, migrationsPath string) (uint, bool, error) {
|
||||
return version, dirty, nil
|
||||
}
|
||||
|
||||
// LatestVersion returns the highest numbered up migration on disk.
|
||||
func LatestVersion(migrationsPath string) (uint, error) {
|
||||
entries, err := os.ReadDir(migrationsPath)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("read migrations: %w", err)
|
||||
}
|
||||
var latest uint64
|
||||
for _, entry := range entries {
|
||||
name := entry.Name()
|
||||
if entry.IsDir() || !strings.HasSuffix(name, ".up.sql") {
|
||||
continue
|
||||
}
|
||||
prefix := strings.SplitN(filepath.Base(name), "_", 2)[0]
|
||||
version, parseErr := strconv.ParseUint(prefix, 10, 64)
|
||||
if parseErr != nil {
|
||||
continue
|
||||
}
|
||||
if version > latest {
|
||||
latest = version
|
||||
}
|
||||
}
|
||||
if latest == 0 {
|
||||
return 0, errors.New("no numbered up migrations found")
|
||||
}
|
||||
return uint(latest), nil
|
||||
}
|
||||
|
||||
// CheckVersion verifies that golang-migrate reached the expected clean version.
|
||||
func CheckVersion(ctx context.Context, db *gorm.DB, expected uint) error {
|
||||
if db == nil {
|
||||
return errors.New("database is not configured")
|
||||
}
|
||||
var state struct {
|
||||
Version uint
|
||||
Dirty bool
|
||||
}
|
||||
if err := db.WithContext(ctx).Table("schema_migrations").Select("version, dirty").Take(&state).Error; err != nil {
|
||||
return fmt.Errorf("read schema migration state: %w", err)
|
||||
}
|
||||
if state.Dirty {
|
||||
return fmt.Errorf("schema migration %d is dirty", state.Version)
|
||||
}
|
||||
if state.Version != expected {
|
||||
return fmt.Errorf("schema migration version %d, expected %d", state.Version, expected)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func newMigrate(dbURL, migrationsPath string) (*migrate.Migrate, error) {
|
||||
if strings.TrimSpace(migrationsPath) == "" {
|
||||
return nil, errors.New("migration path must not be empty")
|
||||
|
||||
@@ -1,10 +1,14 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func TestMigrationCommandsRejectBlankPath(t *testing.T) {
|
||||
@@ -56,3 +60,23 @@ func TestSanitizePostgresEnvironment(t *testing.T) {
|
||||
require.False(t, exists, "%s should be removed at application startup", key)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLatestVersionAndCheckVersion(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "000001_first.up.sql"), []byte("SELECT 1;"), 0o600))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "000012_latest.up.sql"), []byte("SELECT 1;"), 0o600))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "README"), nil, 0o600))
|
||||
|
||||
latest, err := LatestVersion(dir)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint(12), latest)
|
||||
|
||||
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, db.Exec("CREATE TABLE schema_migrations (version INTEGER NOT NULL, dirty BOOLEAN NOT NULL)").Error)
|
||||
require.NoError(t, db.Exec("INSERT INTO schema_migrations(version, dirty) VALUES (12, false)").Error)
|
||||
require.NoError(t, CheckVersion(context.Background(), db, latest))
|
||||
require.Error(t, CheckVersion(context.Background(), db, 13))
|
||||
require.NoError(t, db.Exec("UPDATE schema_migrations SET dirty = true").Error)
|
||||
require.ErrorContains(t, CheckVersion(context.Background(), db, latest), "dirty")
|
||||
}
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
@@ -93,6 +95,49 @@ func TestReadyHandler_Ready(t *testing.T) {
|
||||
assert.True(t, resp["ready"].(bool))
|
||||
}
|
||||
|
||||
func TestReadyHandler_RequiredDependencyFailure(t *testing.T) {
|
||||
db := newTestDB(t)
|
||||
router := gin.New()
|
||||
router.GET("/ready", ReadyHandler(db, DependencyCheck{Name: "redis", Check: func(context.Context) error {
|
||||
return errors.New("connection refused")
|
||||
}}))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/ready", nil))
|
||||
assert.Equal(t, http.StatusServiceUnavailable, w.Code)
|
||||
assert.JSONEq(t, `{"ready":false,"checks":{"database":"healthy","redis":"unhealthy: connection refused"}}`, w.Body.String())
|
||||
}
|
||||
|
||||
func TestDatabaseOutageFailsReadinessButNotLiveness(t *testing.T) {
|
||||
db := newTestDB(t)
|
||||
sqlDB, err := db.DB()
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, sqlDB.Close())
|
||||
router := gin.New()
|
||||
router.GET("/ready", ReadyHandler(db))
|
||||
router.GET("/live", LiveHandler())
|
||||
|
||||
ready := httptest.NewRecorder()
|
||||
router.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/ready", nil))
|
||||
assert.Equal(t, http.StatusServiceUnavailable, ready.Code)
|
||||
live := httptest.NewRecorder()
|
||||
router.ServeHTTP(live, httptest.NewRequest(http.MethodGet, "/live", nil))
|
||||
assert.Equal(t, http.StatusOK, live.Code)
|
||||
}
|
||||
|
||||
func TestHealthHandler_SelectedDependency(t *testing.T) {
|
||||
db := newTestDB(t)
|
||||
router := gin.New()
|
||||
router.GET("/health", HealthHandler(db, time.Now(), "test", DependencyCheck{Name: "redis", Check: func(context.Context) error { return nil }}))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/health?check=redis", nil))
|
||||
assert.Equal(t, http.StatusOK, w.Code)
|
||||
var response HealthResponse
|
||||
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &response))
|
||||
assert.Equal(t, map[string]string{"redis": "healthy"}, response.Checks)
|
||||
}
|
||||
|
||||
func TestReadyHandler_NilDB_NotReady(t *testing.T) {
|
||||
// ReadyHandler does not handle nil DB (panics on db.DB()).
|
||||
t.Skip("ReadyHandler panics on nil DB; nil DB is not a valid runtime state")
|
||||
@@ -144,6 +189,23 @@ func TestPrometheusHandler(t *testing.T) {
|
||||
assert.Contains(t, body, "# TYPE")
|
||||
}
|
||||
|
||||
func TestHTTPMetricsRecordsRequestsErrorsAndLatency(t *testing.T) {
|
||||
metrics := NewHTTPMetrics(time.Now())
|
||||
router := gin.New()
|
||||
router.Use(metrics.Middleware())
|
||||
router.GET("/items/:id", func(c *gin.Context) { c.Status(http.StatusInternalServerError) })
|
||||
router.GET("/metrics", metrics.Handler())
|
||||
|
||||
router.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/items/42", nil))
|
||||
w := httptest.NewRecorder()
|
||||
router.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
|
||||
body := w.Body.String()
|
||||
assert.Contains(t, body, `http_requests_total{method="GET",route="/items/:id",status="500"} 1`)
|
||||
assert.Contains(t, body, `http_request_errors_total{method="GET",route="/items/:id",status="500"} 1`)
|
||||
assert.Contains(t, body, `http_request_duration_seconds_count{method="GET",route="/items/:id",status="500"} 1`)
|
||||
}
|
||||
|
||||
func TestFormatGauge(t *testing.T) {
|
||||
result := formatGauge("test_metric", 42)
|
||||
assert.Equal(t, "test_metric 42", result)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"runtime"
|
||||
"strconv"
|
||||
@@ -10,6 +12,11 @@ import (
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type DependencyCheck struct {
|
||||
Name string
|
||||
Check func(context.Context) error
|
||||
}
|
||||
|
||||
// HealthResponse is the structured health check response.
|
||||
type HealthResponse struct {
|
||||
Status string `json:"status"`
|
||||
@@ -19,52 +26,26 @@ type HealthResponse struct {
|
||||
Checks map[string]string `json:"checks"`
|
||||
}
|
||||
|
||||
// HealthHandler returns application health status.
|
||||
// Reference: Chatwoot uses /health for monitoring in docker-compose.production.yaml
|
||||
func HealthHandler(db *gorm.DB, startTime time.Time, version string) gin.HandlerFunc {
|
||||
// HealthHandler reports dependency state and optional runtime diagnostics.
|
||||
func HealthHandler(db *gorm.DB, startTime time.Time, version string, dependencies ...DependencyCheck) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
checks := make(map[string]string)
|
||||
overall := "healthy"
|
||||
|
||||
// Database check
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
checks["database"] = "unhealthy: " + err.Error()
|
||||
overall = "unhealthy"
|
||||
} else if err := sqlDB.Ping(); err != nil {
|
||||
checks["database"] = "unhealthy: " + err.Error()
|
||||
overall = "unhealthy"
|
||||
} else {
|
||||
checks["database"] = "healthy"
|
||||
}
|
||||
|
||||
// Memory check
|
||||
var m runtime.MemStats
|
||||
runtime.ReadMemStats(&m)
|
||||
memMB := m.Alloc / 1024 / 1024
|
||||
checks["memory_alloc_mb"] = strconv.FormatUint(memMB, 10)
|
||||
if memMB > 500 {
|
||||
checks["memory_warning"] = "high memory usage"
|
||||
}
|
||||
|
||||
// Goroutine check
|
||||
goroutines := runtime.NumGoroutine()
|
||||
checks["goroutines"] = strconv.Itoa(goroutines)
|
||||
if goroutines > 1000 {
|
||||
checks["goroutine_warning"] = "high goroutine count"
|
||||
}
|
||||
|
||||
// Uptime
|
||||
selected := c.Query("check")
|
||||
checks, healthy := runDependencyChecks(c.Request.Context(), db, selected, dependencies)
|
||||
uptime := time.Since(startTime)
|
||||
checks["uptime_seconds"] = strconv.FormatUint(uint64(uptime.Seconds()), 10)
|
||||
|
||||
statusCode := http.StatusOK
|
||||
if overall == "unhealthy" {
|
||||
statusCode = http.StatusServiceUnavailable
|
||||
if selected == "" {
|
||||
var memory runtime.MemStats
|
||||
runtime.ReadMemStats(&memory)
|
||||
checks["memory_alloc_mb"] = strconv.FormatUint(memory.Alloc/1024/1024, 10)
|
||||
checks["goroutines"] = strconv.Itoa(runtime.NumGoroutine())
|
||||
checks["uptime_seconds"] = strconv.FormatUint(uint64(uptime.Seconds()), 10)
|
||||
}
|
||||
|
||||
status, statusCode := "healthy", http.StatusOK
|
||||
if !healthy {
|
||||
status, statusCode = "unhealthy", http.StatusServiceUnavailable
|
||||
}
|
||||
c.JSON(statusCode, HealthResponse{
|
||||
Status: overall,
|
||||
Status: status,
|
||||
Timestamp: time.Now().UTC().Format(time.RFC3339),
|
||||
Version: version,
|
||||
Uptime: uptime.String(),
|
||||
@@ -73,27 +54,64 @@ func HealthHandler(db *gorm.DB, startTime time.Time, version string) gin.Handler
|
||||
}
|
||||
}
|
||||
|
||||
// ReadyHandler returns whether the app is ready to accept traffic.
|
||||
// Used by K8s readiness probes — returns 503 if not ready.
|
||||
func ReadyHandler(db *gorm.DB) gin.HandlerFunc {
|
||||
// ReadyHandler returns 503 while draining or when a required dependency fails.
|
||||
func ReadyHandler(db *gorm.DB, dependencies ...DependencyCheck) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
c.JSON(http.StatusServiceUnavailable, gin.H{"ready": false, "reason": "db error"})
|
||||
return
|
||||
checks, ready := runDependencyChecks(c.Request.Context(), db, "", dependencies)
|
||||
statusCode := http.StatusOK
|
||||
if !ready {
|
||||
statusCode = http.StatusServiceUnavailable
|
||||
}
|
||||
if err := sqlDB.Ping(); err != nil {
|
||||
c.JSON(http.StatusServiceUnavailable, gin.H{"ready": false, "reason": "db unreachable"})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ready": true})
|
||||
c.JSON(statusCode, gin.H{"ready": ready, "checks": checks})
|
||||
}
|
||||
}
|
||||
|
||||
// LiveHandler returns whether the app process is alive.
|
||||
// Used by K8s liveness probes — simplest possible check.
|
||||
// LiveHandler only proves that the process can serve HTTP; dependency failures
|
||||
// belong to readiness so an outage does not trigger a restart loop.
|
||||
func LiveHandler() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"alive": true})
|
||||
}
|
||||
}
|
||||
|
||||
func runDependencyChecks(ctx context.Context, db *gorm.DB, selected string, dependencies []DependencyCheck) (map[string]string, bool) {
|
||||
checks := make(map[string]string, len(dependencies)+1)
|
||||
healthy := true
|
||||
all := append([]DependencyCheck{{Name: "database", Check: databasePing(db)}}, dependencies...)
|
||||
found := selected == ""
|
||||
for _, dependency := range all {
|
||||
if selected != "" && dependency.Name != selected {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if dependency.Check == nil {
|
||||
checks[dependency.Name] = "unhealthy: check is not configured"
|
||||
healthy = false
|
||||
continue
|
||||
}
|
||||
if err := dependency.Check(ctx); err != nil {
|
||||
checks[dependency.Name] = "unhealthy: " + err.Error()
|
||||
healthy = false
|
||||
} else {
|
||||
checks[dependency.Name] = "healthy"
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
checks[selected] = "unhealthy: unknown check"
|
||||
healthy = false
|
||||
}
|
||||
return checks, healthy
|
||||
}
|
||||
|
||||
func databasePing(db *gorm.DB) func(context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
if db == nil {
|
||||
return errors.New("database is not configured")
|
||||
}
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return sqlDB.PingContext(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,79 +4,147 @@ import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// MetricsHandler exposes Prometheus-compatible metrics in text exposition format.
|
||||
// Reference: Chatwoot uses Prometheus exporter for Sidekiq, Rails metrics
|
||||
// This provides Go runtime + application metrics on a dedicated port.
|
||||
var durationBuckets = [...]float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5}
|
||||
|
||||
// PrometheusHandler returns metrics in Prometheus text format.
|
||||
func PrometheusHandler(startTime time.Time) gin.HandlerFunc {
|
||||
type requestMetricKey struct {
|
||||
Method string
|
||||
Route string
|
||||
Status int
|
||||
}
|
||||
|
||||
type requestMetric struct {
|
||||
Count uint64
|
||||
Errors uint64
|
||||
Sum float64
|
||||
Buckets [len(durationBuckets)]uint64
|
||||
}
|
||||
|
||||
// HTTPMetrics records bounded route-template labels without a Prometheus client dependency.
|
||||
type HTTPMetrics struct {
|
||||
startTime time.Time
|
||||
mu sync.RWMutex
|
||||
requests map[requestMetricKey]requestMetric
|
||||
}
|
||||
|
||||
func NewHTTPMetrics(startTime time.Time) *HTTPMetrics {
|
||||
return &HTTPMetrics{startTime: startTime, requests: make(map[requestMetricKey]requestMetric)}
|
||||
}
|
||||
|
||||
func (m *HTTPMetrics) Middleware() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
var m runtime.MemStats
|
||||
runtime.ReadMemStats(&m)
|
||||
|
||||
uptime := time.Since(startTime).Seconds()
|
||||
|
||||
metrics := []string{
|
||||
// Go runtime metrics
|
||||
formatGauge("gochat_go_goroutines", uint64(runtime.NumGoroutine())),
|
||||
formatGauge("gochat_go_memory_alloc_bytes", m.Alloc),
|
||||
formatGauge("gochat_go_memory_sys_bytes", m.Sys),
|
||||
formatGauge("gochat_go_memory_total_alloc_bytes", m.TotalAlloc),
|
||||
formatGauge("gochat_go_gc_pause_total_ns", m.PauseTotalNs),
|
||||
formatCounter("gochat_go_gc_count", uint64(m.NumGC)),
|
||||
// Application metrics
|
||||
formatGauge("gochat_uptime_seconds", uint64(uptime)),
|
||||
formatGauge("gochat_threads_count", uint64(runtime.NumCPU())),
|
||||
started := time.Now()
|
||||
c.Next()
|
||||
route := c.FullPath()
|
||||
if route == "" {
|
||||
route = "unmatched"
|
||||
}
|
||||
|
||||
// HELP and TYPE annotations
|
||||
help := []string{
|
||||
"# HELP gochat_go_goroutines Number of goroutines currently running",
|
||||
"# TYPE gochat_go_goroutines gauge",
|
||||
"# HELP gochat_go_memory_alloc_bytes Bytes of allocated heap objects",
|
||||
"# TYPE gochat_go_memory_alloc_bytes gauge",
|
||||
"# HELP gochat_go_memory_sys_bytes Bytes obtained from system",
|
||||
"# TYPE gochat_go_memory_sys_bytes gauge",
|
||||
"# HELP gochat_uptime_seconds Application uptime in seconds",
|
||||
"# TYPE gochat_uptime_seconds gauge",
|
||||
}
|
||||
|
||||
output := ""
|
||||
for _, h := range help {
|
||||
output += h + "\n"
|
||||
}
|
||||
for _, m := range metrics {
|
||||
output += m + "\n"
|
||||
}
|
||||
|
||||
c.Header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
||||
c.String(http.StatusOK, output)
|
||||
m.observe(requestMetricKey{Method: c.Request.Method, Route: route, Status: c.Writer.Status()}, time.Since(started).Seconds())
|
||||
}
|
||||
}
|
||||
|
||||
func formatGauge(name string, value uint64) string {
|
||||
return name + " " + formatValue(value)
|
||||
func (m *HTTPMetrics) observe(key requestMetricKey, seconds float64) {
|
||||
m.mu.Lock()
|
||||
metric := m.requests[key]
|
||||
metric.Count++
|
||||
metric.Sum += seconds
|
||||
if key.Status >= http.StatusInternalServerError {
|
||||
metric.Errors++
|
||||
}
|
||||
for i, boundary := range durationBuckets {
|
||||
if seconds <= boundary {
|
||||
metric.Buckets[i]++
|
||||
}
|
||||
}
|
||||
m.requests[key] = metric
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
func formatCounter(name string, value uint64) string {
|
||||
return name + " " + formatValue(value)
|
||||
func (m *HTTPMetrics) Handler() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
var memory runtime.MemStats
|
||||
runtime.ReadMemStats(&memory)
|
||||
keys, snapshot := m.snapshot()
|
||||
|
||||
var output strings.Builder
|
||||
output.WriteString("# HELP http_requests_total Total HTTP requests\n# TYPE http_requests_total counter\n")
|
||||
output.WriteString("# HELP http_request_errors_total Total HTTP 5xx responses\n# TYPE http_request_errors_total counter\n")
|
||||
output.WriteString("# HELP http_request_duration_seconds HTTP request duration\n# TYPE http_request_duration_seconds histogram\n")
|
||||
for _, key := range keys {
|
||||
metric := snapshot[key]
|
||||
labels := requestLabels(key)
|
||||
fmt.Fprintf(&output, "http_requests_total{%s} %d\n", labels, metric.Count)
|
||||
fmt.Fprintf(&output, "http_request_errors_total{%s} %d\n", labels, metric.Errors)
|
||||
for i, boundary := range durationBuckets {
|
||||
fmt.Fprintf(&output, "http_request_duration_seconds_bucket{%s,le=%q} %d\n", labels, strconv.FormatFloat(boundary, 'g', -1, 64), metric.Buckets[i])
|
||||
}
|
||||
fmt.Fprintf(&output, "http_request_duration_seconds_bucket{%s,le=\"+Inf\"} %d\n", labels, metric.Count)
|
||||
fmt.Fprintf(&output, "http_request_duration_seconds_sum{%s} %s\n", labels, strconv.FormatFloat(metric.Sum, 'g', -1, 64))
|
||||
fmt.Fprintf(&output, "http_request_duration_seconds_count{%s} %d\n", labels, metric.Count)
|
||||
}
|
||||
|
||||
output.WriteString("# HELP gochat_go_goroutines Number of goroutines currently running\n# TYPE gochat_go_goroutines gauge\n")
|
||||
output.WriteString("# HELP gochat_go_memory_alloc_bytes Bytes of allocated heap objects\n# TYPE gochat_go_memory_alloc_bytes gauge\n")
|
||||
output.WriteString("# HELP gochat_go_memory_sys_bytes Bytes obtained from system\n# TYPE gochat_go_memory_sys_bytes gauge\n")
|
||||
output.WriteString("# HELP gochat_uptime_seconds Application uptime in seconds\n# TYPE gochat_uptime_seconds gauge\n")
|
||||
fmt.Fprintf(&output, "gochat_go_goroutines %d\n", runtime.NumGoroutine())
|
||||
fmt.Fprintf(&output, "gochat_go_memory_alloc_bytes %d\n", memory.Alloc)
|
||||
fmt.Fprintf(&output, "gochat_go_memory_sys_bytes %d\n", memory.Sys)
|
||||
fmt.Fprintf(&output, "gochat_go_memory_total_alloc_bytes %d\n", memory.TotalAlloc)
|
||||
fmt.Fprintf(&output, "gochat_go_gc_pause_total_ns %d\n", memory.PauseTotalNs)
|
||||
fmt.Fprintf(&output, "gochat_go_gc_count %d\n", memory.NumGC)
|
||||
fmt.Fprintf(&output, "gochat_uptime_seconds %s\n", strconv.FormatFloat(time.Since(m.startTime).Seconds(), 'f', 3, 64))
|
||||
fmt.Fprintf(&output, "gochat_threads_count %d\n", runtime.NumCPU())
|
||||
|
||||
c.Data(http.StatusOK, "text/plain; version=0.0.4; charset=utf-8", []byte(output.String()))
|
||||
}
|
||||
}
|
||||
|
||||
func formatValue(v uint64) string {
|
||||
// Simple uint64 formatting without strconv dependency
|
||||
if v == 0 {
|
||||
return "0"
|
||||
func (m *HTTPMetrics) snapshot() ([]requestMetricKey, map[requestMetricKey]requestMetric) {
|
||||
m.mu.RLock()
|
||||
snapshot := make(map[requestMetricKey]requestMetric, len(m.requests))
|
||||
keys := make([]requestMetricKey, 0, len(m.requests))
|
||||
for key, metric := range m.requests {
|
||||
keys = append(keys, key)
|
||||
snapshot[key] = metric
|
||||
}
|
||||
result := ""
|
||||
for v > 0 {
|
||||
digit := v % 10
|
||||
result = fmt.Sprintf("%d%s", digit, result)
|
||||
v /= 10
|
||||
}
|
||||
return result
|
||||
m.mu.RUnlock()
|
||||
sort.Slice(keys, func(i, j int) bool {
|
||||
left, right := keys[i], keys[j]
|
||||
if left.Route != right.Route {
|
||||
return left.Route < right.Route
|
||||
}
|
||||
if left.Method != right.Method {
|
||||
return left.Method < right.Method
|
||||
}
|
||||
return left.Status < right.Status
|
||||
})
|
||||
return keys, snapshot
|
||||
}
|
||||
|
||||
func requestLabels(key requestMetricKey) string {
|
||||
return fmt.Sprintf(`method="%s",route="%s",status="%s"`, escapeLabel(key.Method), escapeLabel(key.Route), strconv.Itoa(key.Status))
|
||||
}
|
||||
|
||||
func escapeLabel(value string) string {
|
||||
value = strings.ReplaceAll(value, `\`, `\\`)
|
||||
value = strings.ReplaceAll(value, "\n", `\n`)
|
||||
return strings.ReplaceAll(value, `"`, `\"`)
|
||||
}
|
||||
|
||||
// PrometheusHandler is retained for callers that only need runtime metrics.
|
||||
func PrometheusHandler(startTime time.Time) gin.HandlerFunc {
|
||||
return NewHTTPMetrics(startTime).Handler()
|
||||
}
|
||||
|
||||
func formatGauge(name string, value uint64) string { return name + " " + formatValue(value) }
|
||||
func formatCounter(name string, value uint64) string { return name + " " + formatValue(value) }
|
||||
func formatValue(value uint64) string { return strconv.FormatUint(value, 10) }
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// HandlerGroup stops new handlers, cancels their shared lifecycle context,
|
||||
// and waits for handlers already using application dependencies.
|
||||
type HandlerGroup struct {
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
|
||||
mu sync.Mutex
|
||||
done *sync.Cond
|
||||
active int
|
||||
stopped bool
|
||||
}
|
||||
|
||||
func NewHandlerGroup() *HandlerGroup {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
g := &HandlerGroup{ctx: ctx, cancel: cancel}
|
||||
g.done = sync.NewCond(&g.mu)
|
||||
return g
|
||||
}
|
||||
|
||||
func (g *HandlerGroup) Context() context.Context { return g.ctx }
|
||||
|
||||
func (g *HandlerGroup) Begin() bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.stopped {
|
||||
return false
|
||||
}
|
||||
g.active++
|
||||
return true
|
||||
}
|
||||
|
||||
func (g *HandlerGroup) Done() {
|
||||
g.mu.Lock()
|
||||
g.active--
|
||||
if g.active == 0 {
|
||||
g.done.Broadcast()
|
||||
}
|
||||
g.mu.Unlock()
|
||||
}
|
||||
|
||||
func (g *HandlerGroup) Stop() {
|
||||
g.mu.Lock()
|
||||
if !g.stopped {
|
||||
g.stopped = true
|
||||
g.cancel()
|
||||
}
|
||||
g.mu.Unlock()
|
||||
}
|
||||
|
||||
func (g *HandlerGroup) Wait() {
|
||||
g.mu.Lock()
|
||||
for g.active != 0 {
|
||||
g.done.Wait()
|
||||
}
|
||||
g.mu.Unlock()
|
||||
}
|
||||
@@ -1859,6 +1859,11 @@ func TestIsRateLimitExemptPath_Metrics_Cov7(t *testing.T) {
|
||||
assert.True(t, isRateLimitExemptPath("/metrics"))
|
||||
}
|
||||
|
||||
func TestIsRateLimitExemptPath_Probes_Cov7(t *testing.T) {
|
||||
assert.True(t, isRateLimitExemptPath("/ready"))
|
||||
assert.True(t, isRateLimitExemptPath("/live"))
|
||||
}
|
||||
|
||||
func TestIsRateLimitExemptPath_Other_Cov7(t *testing.T) {
|
||||
assert.False(t, isRateLimitExemptPath("/api/v1/conversations"))
|
||||
}
|
||||
|
||||
@@ -305,7 +305,7 @@ func RateLimit(rdb *redis.Client) gin.HandlerFunc {
|
||||
|
||||
func isRateLimitExemptPath(path string) bool {
|
||||
switch path {
|
||||
case "/health", "/metrics":
|
||||
case "/health", "/ready", "/live", "/metrics":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
package router
|
||||
|
||||
import (
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func TestHealthCheck_Cov1(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("GET", "/health", nil)
|
||||
healthCheck(c)
|
||||
}
|
||||
@@ -11,7 +11,6 @@ import (
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
swaggerFiles "github.com/swaggo/files"
|
||||
@@ -33,9 +32,6 @@ import (
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// startTime records when the application process launched, used for uptime in /health.
|
||||
var startTime = time.Now()
|
||||
|
||||
// Handlers holds all instantiated handler structs for route registration.
|
||||
// Passed from bootstrap to avoid global state and keep dependency wiring explicit.
|
||||
type Handlers struct {
|
||||
@@ -201,9 +197,6 @@ func RegisterRoutes(
|
||||
corsCfg middleware.CORSConfig,
|
||||
db *gorm.DB,
|
||||
) {
|
||||
// Health check endpoint (ref: Chatwoot health_check route)
|
||||
engine.GET("/health", healthCheck)
|
||||
|
||||
// Swagger UI — interactive API documentation
|
||||
engine.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
|
||||
|
||||
@@ -2238,19 +2231,6 @@ func webhookProviderUnavailable(c *gin.Context) {
|
||||
})
|
||||
}
|
||||
|
||||
func healthCheck(c *gin.Context) {
|
||||
uptime := time.Since(startTime)
|
||||
c.JSON(200, gin.H{
|
||||
"status": "ok",
|
||||
"service": "gochat",
|
||||
"version": config.Version,
|
||||
"commit": config.CommitSHA,
|
||||
"buildDate": config.BuildDate,
|
||||
"uptime": uptime.String(),
|
||||
"uptimeSeconds": uint64(uptime.Seconds()),
|
||||
})
|
||||
}
|
||||
|
||||
func dashboardIndex(c *gin.Context) {
|
||||
if dashboardWantsJSON(c) {
|
||||
c.JSON(http.StatusNotAcceptable, gin.H{"error": "Please use API routes instead of dashboard routes for JSON requests"})
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ThreeDotsLabs/watermill"
|
||||
"github.com/ThreeDotsLabs/watermill/message"
|
||||
"github.com/gochat/gochat/internal/lifecycle"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type notificationLifecycleSubscriber struct {
|
||||
ready chan struct{}
|
||||
messages chan *message.Message
|
||||
once sync.Once
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s *notificationLifecycleSubscriber) Subscribe(ctx context.Context, _ string) (<-chan *message.Message, error) {
|
||||
s.ctx = ctx
|
||||
s.once.Do(func() { close(s.ready) })
|
||||
return s.messages, nil
|
||||
}
|
||||
|
||||
func (*notificationLifecycleSubscriber) Close() error { return nil }
|
||||
|
||||
func (s *notificationLifecycleSubscriber) Send(msg *message.Message) {
|
||||
msg.SetContext(s.ctx)
|
||||
s.messages <- msg
|
||||
}
|
||||
|
||||
func TestNotificationCloseTimeoutWaitsForHandlerBeforeDependenciesClose(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open("file:notification-shutdown-order?mode=memory&cache=shared"), &gorm.Config{})
|
||||
require.NoError(t, err)
|
||||
router, err := message.NewRouter(message.RouterConfig{CloseTimeout: 10 * time.Millisecond}, watermill.NopLogger{})
|
||||
require.NoError(t, err)
|
||||
subscriber := ¬ificationLifecycleSubscriber{
|
||||
ready: make(chan struct{}),
|
||||
messages: make(chan *message.Message),
|
||||
}
|
||||
handlers := lifecycle.NewHandlerGroup()
|
||||
service := &NotificationDeliveryService{router: router, subscriber: subscriber, handlers: handlers}
|
||||
|
||||
started := make(chan struct{})
|
||||
cancelled := make(chan struct{})
|
||||
probeDependency := make(chan struct{})
|
||||
handlerDBErr := make(chan error, 1)
|
||||
router.AddNoPublisherHandler("blocked", "blocked", subscriber, service.track(func(msg *message.Message) error {
|
||||
close(started)
|
||||
<-msg.Context().Done()
|
||||
close(cancelled)
|
||||
<-probeDependency
|
||||
var one int
|
||||
handlerDBErr <- db.Raw("SELECT 1").Scan(&one).Error
|
||||
return nil
|
||||
}))
|
||||
|
||||
runDone := make(chan error, 1)
|
||||
go func() { runDone <- service.Start(handlers.Context()) }()
|
||||
<-subscriber.ready
|
||||
subscriber.Send(message.NewMessage("blocked", nil))
|
||||
<-started
|
||||
|
||||
handlers.Stop()
|
||||
<-cancelled
|
||||
require.ErrorContains(t, service.Close(), "router close timeout")
|
||||
|
||||
waitDone := make(chan struct{})
|
||||
go func() {
|
||||
handlers.Wait()
|
||||
close(waitDone)
|
||||
}()
|
||||
select {
|
||||
case <-waitDone:
|
||||
t.Fatal("handler gate opened before the blocked handler completed")
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
require.NoError(t, db.Exec("SELECT 1").Error)
|
||||
|
||||
close(probeDependency)
|
||||
require.NoError(t, <-handlerDBErr)
|
||||
<-waitDone
|
||||
sqlDB, err := db.DB()
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, sqlDB.Close())
|
||||
require.Error(t, sqlDB.Ping())
|
||||
require.NoError(t, <-runDone)
|
||||
}
|
||||
@@ -3,6 +3,7 @@ package service
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/ThreeDotsLabs/watermill"
|
||||
@@ -12,6 +13,7 @@ import (
|
||||
|
||||
applogger "github.com/gochat/gochat/pkg/logger"
|
||||
|
||||
"github.com/gochat/gochat/internal/lifecycle"
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/pubsub"
|
||||
"github.com/gochat/gochat/internal/repository"
|
||||
@@ -22,24 +24,26 @@ import (
|
||||
// creation and multi-channel delivery (push, email, webhook).
|
||||
//
|
||||
// Architecture mapping:
|
||||
// Chatwoot Wisper → Sidekiq workers → NotificationDeliveryService
|
||||
// Each event (message_created, conversation_assigned, etc.) triggers:
|
||||
// 1. Create a Notification record in the DB
|
||||
// 2. Check NotificationPreference for push/email enabled
|
||||
// 3. Deliver push via PushDeliveryService
|
||||
// 4. Deliver webhook via WebhookDeliveryService (HTTP POST with HMAC signing)
|
||||
//
|
||||
// Chatwoot Wisper → Sidekiq workers → NotificationDeliveryService
|
||||
// Each event (message_created, conversation_assigned, etc.) triggers:
|
||||
// 1. Create a Notification record in the DB
|
||||
// 2. Check NotificationPreference for push/email enabled
|
||||
// 3. Deliver push via PushDeliveryService
|
||||
// 4. Deliver webhook via WebhookDeliveryService (HTTP POST with HMAC signing)
|
||||
//
|
||||
// Reference: Chatwoot notification_service.rb + P2B M8 spec
|
||||
type NotificationDeliveryService struct {
|
||||
notificationService *NotificationService
|
||||
pushDeliveryService *PushDeliveryService
|
||||
webhookDeliverySvc *WebhookDeliveryService
|
||||
webhookSignatureSvc *security.WebhookSignatureService
|
||||
notifPrefRepo *repository.NotificationPreferenceRepo
|
||||
pushTokenRepo *repository.PushTokenRepo
|
||||
webhookSubRepo *repository.WebhookSubscriptionRepo
|
||||
router *message.Router
|
||||
subscriber *redisstream.Subscriber
|
||||
notificationService *NotificationService
|
||||
pushDeliveryService *PushDeliveryService
|
||||
webhookDeliverySvc *WebhookDeliveryService
|
||||
webhookSignatureSvc *security.WebhookSignatureService
|
||||
notifPrefRepo *repository.NotificationPreferenceRepo
|
||||
pushTokenRepo *repository.PushTokenRepo
|
||||
webhookSubRepo *repository.WebhookSubscriptionRepo
|
||||
router *message.Router
|
||||
subscriber message.Subscriber
|
||||
handlers *lifecycle.HandlerGroup
|
||||
}
|
||||
|
||||
// NewNotificationDeliveryService creates a delivery service and registers Watermill handlers.
|
||||
@@ -72,15 +76,15 @@ func NewNotificationDeliveryService(
|
||||
}
|
||||
|
||||
s := &NotificationDeliveryService{
|
||||
notificationService: notificationService,
|
||||
pushDeliveryService: pushDeliveryService,
|
||||
webhookDeliverySvc: webhookDeliverySvc,
|
||||
webhookSignatureSvc: webhookSignatureSvc,
|
||||
notifPrefRepo: notifPrefRepo,
|
||||
pushTokenRepo: pushTokenRepo,
|
||||
webhookSubRepo: webhookSubRepo,
|
||||
router: router,
|
||||
subscriber: subscriber,
|
||||
notificationService: notificationService,
|
||||
pushDeliveryService: pushDeliveryService,
|
||||
webhookDeliverySvc: webhookDeliverySvc,
|
||||
webhookSignatureSvc: webhookSignatureSvc,
|
||||
notifPrefRepo: notifPrefRepo,
|
||||
pushTokenRepo: pushTokenRepo,
|
||||
webhookSubRepo: webhookSubRepo,
|
||||
router: router,
|
||||
subscriber: subscriber,
|
||||
}
|
||||
|
||||
s.registerHandlers()
|
||||
@@ -116,7 +120,7 @@ func (s *NotificationDeliveryService) registerHandlers() {
|
||||
handlerName,
|
||||
topic,
|
||||
s.subscriber,
|
||||
s.handleNotificationEvent(notifType),
|
||||
s.track(s.handleNotificationEvent(notifType)),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -125,10 +129,26 @@ func (s *NotificationDeliveryService) registerHandlers() {
|
||||
"notif-delivery-system-notification-handler",
|
||||
pubsub.TopicSystemNotification,
|
||||
s.subscriber,
|
||||
s.handleSystemNotification(),
|
||||
s.track(s.handleSystemNotification()),
|
||||
)
|
||||
}
|
||||
|
||||
func (s *NotificationDeliveryService) SetHandlerGroup(handlers *lifecycle.HandlerGroup) {
|
||||
s.handlers = handlers
|
||||
}
|
||||
|
||||
func (s *NotificationDeliveryService) track(handler message.NoPublishHandlerFunc) message.NoPublishHandlerFunc {
|
||||
return func(msg *message.Message) error {
|
||||
if s.handlers != nil {
|
||||
if !s.handlers.Begin() {
|
||||
return context.Canceled
|
||||
}
|
||||
defer s.handlers.Done()
|
||||
}
|
||||
return handler(msg)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Event payload structure ---
|
||||
// The Watermill message payload is expected to be JSON with these fields.
|
||||
|
||||
@@ -159,7 +179,7 @@ func (s *NotificationDeliveryService) handleNotificationEvent(notifType string)
|
||||
return nil
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
ctx := msg.Context()
|
||||
|
||||
// Step 1: Create notification record
|
||||
notif := &model.Notification{
|
||||
@@ -209,7 +229,7 @@ func (s *NotificationDeliveryService) handleSystemNotification() func(msg *messa
|
||||
return nil
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
ctx := msg.Context()
|
||||
|
||||
notif := &model.Notification{
|
||||
AccountID: nilIfZero(payload.AccountID),
|
||||
@@ -313,10 +333,11 @@ func (s *NotificationDeliveryService) Start(ctx context.Context) error {
|
||||
|
||||
// Close shuts down the delivery router and subscriber.
|
||||
func (s *NotificationDeliveryService) Close() error {
|
||||
if err := s.router.Close(); err != nil {
|
||||
applogger.L().Errorf("notif-delivery: failed to close router: %v", err)
|
||||
routerErr := s.router.Close()
|
||||
if routerErr != nil {
|
||||
applogger.L().Errorf("notif-delivery: failed to close router: %v", routerErr)
|
||||
}
|
||||
return s.subscriber.Close()
|
||||
return errors.Join(routerErr, s.subscriber.Close())
|
||||
}
|
||||
|
||||
// --- Watermill logger adapter for delivery service ---
|
||||
@@ -340,4 +361,4 @@ func (a *deliveryWatermillAdapter) Trace(msg string, fields watermill.LogFields)
|
||||
|
||||
func (a *deliveryWatermillAdapter) With(fields watermill.LogFields) watermill.LoggerAdapter {
|
||||
return a
|
||||
}
|
||||
}
|
||||
|
||||
@@ -58,10 +58,13 @@ type WorkerPool struct {
|
||||
blockTimeout time.Duration // XREADGROUP block duration, default 5s
|
||||
sweepInterval time.Duration // compensation sweep interval, default 30s
|
||||
|
||||
mu sync.RWMutex
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
mu sync.RWMutex
|
||||
claimMu sync.RWMutex // serializes lifecycle cancellation with claim transitions
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
jobCtx context.Context
|
||||
jobCancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
type Option func(*WorkerPool)
|
||||
@@ -343,11 +346,21 @@ func (wp *WorkerPool) Start() error {
|
||||
return nil
|
||||
}
|
||||
wp.ctx, wp.cancel = context.WithCancel(context.Background())
|
||||
ctx := wp.ctx
|
||||
wp.jobCtx, wp.jobCancel = context.WithCancel(context.Background())
|
||||
ctx, jobCtx := wp.ctx, wp.jobCtx
|
||||
cancel, jobCancel := wp.cancel, wp.jobCancel
|
||||
workerCount := wp.workerCount
|
||||
wp.mu.Unlock()
|
||||
|
||||
if _, err := wp.RequeueStaleJobs(ctx); err != nil {
|
||||
wp.mu.Lock()
|
||||
wp.cancel = nil
|
||||
wp.ctx = nil
|
||||
wp.jobCancel = nil
|
||||
wp.jobCtx = nil
|
||||
wp.mu.Unlock()
|
||||
cancel()
|
||||
jobCancel()
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -358,7 +371,7 @@ func (wp *WorkerPool) Start() error {
|
||||
|
||||
for i := 0; i < workerCount; i++ {
|
||||
wp.wg.Add(1)
|
||||
go wp.run(ctx, i)
|
||||
go wp.run(ctx, jobCtx, i)
|
||||
}
|
||||
|
||||
// Start the sweep goroutine for delayed-job delivery and stale-job recovery.
|
||||
@@ -373,31 +386,98 @@ func (wp *WorkerPool) Start() error {
|
||||
func (wp *WorkerPool) Stop() error {
|
||||
wp.mu.Lock()
|
||||
cancel := wp.cancel
|
||||
jobCancel := wp.jobCancel
|
||||
wp.cancel = nil
|
||||
wp.ctx = nil
|
||||
wp.jobCancel = nil
|
||||
wp.jobCtx = nil
|
||||
wp.mu.Unlock()
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
if jobCancel != nil {
|
||||
jobCancel()
|
||||
}
|
||||
wp.claimMu.Lock()
|
||||
wp.claimMu.Unlock()
|
||||
wp.wg.Wait()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Shutdown stops claiming new jobs, lets active handlers finish, and only
|
||||
// cancels them if the drain deadline expires.
|
||||
func (wp *WorkerPool) Shutdown(ctx context.Context) error {
|
||||
wp.mu.Lock()
|
||||
cancel := wp.cancel
|
||||
jobCancel := wp.jobCancel
|
||||
wp.cancel = nil
|
||||
wp.ctx = nil
|
||||
wp.jobCancel = nil
|
||||
wp.jobCtx = nil
|
||||
wp.mu.Unlock()
|
||||
if cancel == nil {
|
||||
return nil
|
||||
}
|
||||
cancel()
|
||||
claimsDone := make(chan struct{})
|
||||
go func() {
|
||||
wp.claimMu.Lock()
|
||||
wp.claimMu.Unlock()
|
||||
close(claimsDone)
|
||||
}()
|
||||
timedOut := false
|
||||
select {
|
||||
case <-claimsDone:
|
||||
case <-ctx.Done():
|
||||
timedOut = true
|
||||
if jobCancel != nil {
|
||||
jobCancel()
|
||||
}
|
||||
<-claimsDone
|
||||
}
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
wp.wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
if timedOut {
|
||||
<-done
|
||||
return ctx.Err()
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
if jobCancel != nil {
|
||||
jobCancel()
|
||||
}
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
if jobCancel != nil {
|
||||
jobCancel()
|
||||
}
|
||||
<-done
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (wp *WorkerPool) ProcessOne(ctx context.Context) (bool, error) {
|
||||
return wp.processOne(nil, ctx)
|
||||
}
|
||||
|
||||
func (wp *WorkerPool) processOne(lifecycleCtx, jobCtx context.Context) (bool, error) {
|
||||
if wp.db == nil {
|
||||
return false, ErrWorkerDatabaseRequired
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
if err := jobCtx.Err(); err != nil {
|
||||
return false, err
|
||||
}
|
||||
job, err := wp.claimNext(ctx)
|
||||
job, err := wp.claimNext(lifecycleCtx, jobCtx)
|
||||
if err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return true, wp.perform(ctx, job)
|
||||
return true, wp.perform(jobCtx, job)
|
||||
}
|
||||
|
||||
func (wp *WorkerPool) RequeueStaleJobs(ctx context.Context) (int64, error) {
|
||||
@@ -418,7 +498,7 @@ func (wp *WorkerPool) RequeueStaleJobs(ctx context.Context) (int64, error) {
|
||||
|
||||
// run is the per-goroutine consume loop. When Redis is configured it uses
|
||||
// XREADGROUP BLOCK; otherwise it falls back to DB polling.
|
||||
func (wp *WorkerPool) run(ctx context.Context, index int) {
|
||||
func (wp *WorkerPool) run(ctx, jobCtx context.Context, index int) {
|
||||
defer wp.wg.Done()
|
||||
|
||||
// Each goroutine gets a unique Redis consumer name so XINFO CONSUMERS
|
||||
@@ -426,7 +506,7 @@ func (wp *WorkerPool) run(ctx context.Context, index int) {
|
||||
consumerID := fmt.Sprintf("%s-%d", wp.workerID, index)
|
||||
|
||||
if wp.rdb == nil {
|
||||
wp.runDBPollLoop(ctx)
|
||||
wp.runDBPollLoop(ctx, jobCtx)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -460,7 +540,10 @@ func (wp *WorkerPool) run(ctx context.Context, index int) {
|
||||
|
||||
for _, xstream := range results {
|
||||
for _, msg := range xstream.Messages {
|
||||
wp.processRedisMessage(ctx, xstream.Stream, msg)
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
wp.processRedisMessage(ctx, jobCtx, xstream.Stream, msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -468,11 +551,16 @@ func (wp *WorkerPool) run(ctx context.Context, index int) {
|
||||
|
||||
// runDBPollLoop is the legacy DB-polling loop, used when Redis is unavailable
|
||||
// (e.g. SQLite test mode) or not configured.
|
||||
func (wp *WorkerPool) runDBPollLoop(ctx context.Context) {
|
||||
func (wp *WorkerPool) runDBPollLoop(ctx, jobCtx context.Context) {
|
||||
ticker := time.NewTicker(wp.pollInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
processed, _ := wp.ProcessOne(ctx)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
processed, _ := wp.processOne(ctx, jobCtx)
|
||||
if processed {
|
||||
continue
|
||||
}
|
||||
@@ -488,24 +576,27 @@ func (wp *WorkerPool) runDBPollLoop(ctx context.Context) {
|
||||
// the handler, and acknowledges the Redis message. The DB claim check
|
||||
// guarantees at-most-once execution even if the same job is XADD'd multiple
|
||||
// times (e.g. by both Enqueue and sweep).
|
||||
func (wp *WorkerPool) processRedisMessage(ctx context.Context, stream string, msg redis.XMessage) {
|
||||
func (wp *WorkerPool) processRedisMessage(lifecycleCtx, jobCtx context.Context, stream string, msg redis.XMessage) {
|
||||
if lifecycleCtx.Err() != nil {
|
||||
return
|
||||
}
|
||||
jobIDStr, ok := msg.Values["job_id"]
|
||||
if !ok {
|
||||
wp.ackRedis(ctx, stream, msg.ID)
|
||||
wp.ackRedis(jobCtx, stream, msg.ID)
|
||||
return
|
||||
}
|
||||
|
||||
jobID, err := strconv.ParseUint(fmt.Sprintf("%v", jobIDStr), 10, 64)
|
||||
if err != nil {
|
||||
wp.ackRedis(ctx, stream, msg.ID)
|
||||
wp.ackRedis(jobCtx, stream, msg.ID)
|
||||
return
|
||||
}
|
||||
|
||||
var job model.BackgroundJob
|
||||
if err := wp.db.WithContext(ctx).First(&job, jobID).Error; err != nil {
|
||||
if err := wp.db.WithContext(jobCtx).First(&job, jobID).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
// Job was deleted; ACK and drop.
|
||||
wp.ackRedis(ctx, stream, msg.ID)
|
||||
wp.ackRedis(jobCtx, stream, msg.ID)
|
||||
return
|
||||
}
|
||||
// DB error — do not ACK so Redis can redeliver to another consumer.
|
||||
@@ -516,8 +607,13 @@ func (wp *WorkerPool) processRedisMessage(ctx context.Context, stream string, ms
|
||||
// Claim check: atomically transition queued/retrying → running.
|
||||
// RowsAffected == 0 means another consumer already claimed it or the
|
||||
// job is not yet due; ACK to avoid redelivery loops.
|
||||
wp.claimMu.RLock()
|
||||
if lifecycleCtx.Err() != nil {
|
||||
wp.claimMu.RUnlock()
|
||||
return
|
||||
}
|
||||
now := wp.now()
|
||||
result := wp.db.WithContext(ctx).Model(&model.BackgroundJob{}).
|
||||
result := wp.db.WithContext(jobCtx).Model(&model.BackgroundJob{}).
|
||||
Where("id = ? AND status IN ? AND scheduled_at <= ?",
|
||||
jobID,
|
||||
[]string{model.BackgroundJobStatusQueued, model.BackgroundJobStatusRetrying},
|
||||
@@ -529,25 +625,26 @@ func (wp *WorkerPool) processRedisMessage(ctx context.Context, stream string, ms
|
||||
"locked_by": wp.workerID,
|
||||
"attempts": gorm.Expr("attempts + 1"),
|
||||
})
|
||||
wp.claimMu.RUnlock()
|
||||
if result.Error != nil {
|
||||
applogger.L().Errorf("claim job %d failed: %v", jobID, result.Error)
|
||||
return
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
wp.ackRedis(ctx, stream, msg.ID)
|
||||
wp.ackRedis(jobCtx, stream, msg.ID)
|
||||
return
|
||||
}
|
||||
|
||||
// Reload the job with updated attempts/locked_by fields.
|
||||
wp.db.WithContext(ctx).First(&job, jobID)
|
||||
wp.db.WithContext(jobCtx).First(&job, jobID)
|
||||
|
||||
if err := wp.perform(ctx, &job); err != nil {
|
||||
if failErr := wp.fail(ctx, &job, err); failErr != nil {
|
||||
if err := wp.perform(jobCtx, &job); err != nil {
|
||||
if failErr := wp.fail(jobCtx, &job, err); failErr != nil {
|
||||
applogger.L().Errorf("record job %d failure: %v", job.ID, failErr)
|
||||
}
|
||||
}
|
||||
|
||||
wp.ackRedis(ctx, stream, msg.ID)
|
||||
wp.ackRedis(jobCtx, stream, msg.ID)
|
||||
}
|
||||
|
||||
func (wp *WorkerPool) ackRedis(ctx context.Context, stream, msgID string) {
|
||||
@@ -660,9 +757,16 @@ func (wp *WorkerPool) sweepDueJobs(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (wp *WorkerPool) claimNext(ctx context.Context) (*model.BackgroundJob, error) {
|
||||
func (wp *WorkerPool) claimNext(lifecycleCtx, jobCtx context.Context) (*model.BackgroundJob, error) {
|
||||
wp.claimMu.RLock()
|
||||
defer wp.claimMu.RUnlock()
|
||||
if lifecycleCtx != nil {
|
||||
if err := lifecycleCtx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
var job model.BackgroundJob
|
||||
err := wp.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
err := wp.db.WithContext(jobCtx).Transaction(func(tx *gorm.DB) error {
|
||||
query := tx.Where("status IN ? AND scheduled_at <= ?", []string{model.BackgroundJobStatusQueued, model.BackgroundJobStatusRetrying}, wp.now())
|
||||
if len(wp.queues) > 0 {
|
||||
query = query.Where("queue IN ?", wp.queues)
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -277,6 +278,80 @@ func TestWorkerPoolStopPersistsCancelledJobRetry(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorkerPoolShutdownDrainsActiveJob(t *testing.T) {
|
||||
db := newWorkerTestDB(t)
|
||||
wp := NewWorkerPoolWithOptions(db, WithPollInterval(5*time.Millisecond))
|
||||
started := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
wp.Register("drain", func(ctx context.Context, job *model.BackgroundJob) error {
|
||||
close(started)
|
||||
select {
|
||||
case <-release:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
})
|
||||
job, err := wp.Enqueue(context.Background(), "drain", nil)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, wp.Start())
|
||||
select {
|
||||
case <-started:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("worker did not start job")
|
||||
}
|
||||
|
||||
done := make(chan error, 1)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
go func() { done <- wp.Shutdown(ctx) }()
|
||||
select {
|
||||
case err := <-done:
|
||||
t.Fatalf("shutdown returned before active job drained: %v", err)
|
||||
case <-time.After(30 * time.Millisecond):
|
||||
}
|
||||
close(release)
|
||||
require.NoError(t, <-done)
|
||||
require.Equal(t, model.BackgroundJobStatusCompleted, loadJob(t, db, job.ID).Status)
|
||||
}
|
||||
|
||||
func TestWorkerPoolShutdownDeadlineCancelsActiveJob(t *testing.T) {
|
||||
db := newWorkerTestDB(t)
|
||||
wp := NewWorkerPoolWithOptions(db, WithPollInterval(5*time.Millisecond), WithBackoff(func(int) time.Duration { return 0 }))
|
||||
started := make(chan struct{})
|
||||
cancelled := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
wp.Register("deadline", func(ctx context.Context, job *model.BackgroundJob) error {
|
||||
close(started)
|
||||
<-ctx.Done()
|
||||
close(cancelled)
|
||||
<-release
|
||||
return ctx.Err()
|
||||
})
|
||||
job, err := wp.Enqueue(context.Background(), "deadline", nil)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, wp.Start())
|
||||
select {
|
||||
case <-started:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("worker did not start job")
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- wp.Shutdown(ctx) }()
|
||||
<-cancelled
|
||||
select {
|
||||
case err := <-done:
|
||||
t.Fatalf("shutdown returned before cancelled handler released: %v", err)
|
||||
default:
|
||||
}
|
||||
close(release)
|
||||
require.ErrorIs(t, <-done, context.DeadlineExceeded)
|
||||
require.Equal(t, model.BackgroundJobStatusRetrying, loadJob(t, db, job.ID).Status)
|
||||
}
|
||||
|
||||
// --- Redis Stream path tests ---
|
||||
|
||||
func newMiniRedis(t *testing.T) (*miniredis.Miniredis, *redis.Client) {
|
||||
@@ -300,6 +375,53 @@ func newRedisWorkerPool(t *testing.T, db *gorm.DB, rdb redis.UniversalClient, op
|
||||
return wp
|
||||
}
|
||||
|
||||
func TestRedisMessageRejectedWhenShutdownFollowsBatchCheck(t *testing.T) {
|
||||
db := newWorkerTestDB(t)
|
||||
_, rdb := newMiniRedis(t)
|
||||
wp := newRedisWorkerPool(t, db, rdb)
|
||||
var handled atomic.Int32
|
||||
wp.Register("shutdown_race", func(context.Context, *model.BackgroundJob) error {
|
||||
handled.Add(1)
|
||||
return nil
|
||||
})
|
||||
job, err := wp.Enqueue(context.Background(), "shutdown_race", nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
lifecycleCtx, cancel := context.WithCancel(context.Background())
|
||||
require.NoError(t, lifecycleCtx.Err())
|
||||
cancel() // SIGTERM after the batch-level lifecycle check.
|
||||
wp.processRedisMessage(lifecycleCtx, context.Background(), wp.streamKeyFor(job.Queue), redis.XMessage{
|
||||
ID: "1-0", Values: map[string]any{"job_id": job.ID},
|
||||
})
|
||||
|
||||
require.Zero(t, handled.Load())
|
||||
require.Equal(t, model.BackgroundJobStatusQueued, loadJob(t, db, job.ID).Status)
|
||||
}
|
||||
|
||||
func TestRedisClaimCanceledWhenShutdownRacesClaim(t *testing.T) {
|
||||
db := newWorkerTestDB(t)
|
||||
_, rdb := newMiniRedis(t)
|
||||
wp := newRedisWorkerPool(t, db, rdb, WithBlockTimeout(10*time.Millisecond))
|
||||
require.NoError(t, wp.Start())
|
||||
|
||||
claimStarted := make(chan struct{})
|
||||
var once sync.Once
|
||||
require.NoError(t, db.Callback().Update().Before("gorm:update").Register("test:block_claim", func(tx *gorm.DB) {
|
||||
once.Do(func() { close(claimStarted) })
|
||||
<-tx.Statement.Context.Done()
|
||||
}))
|
||||
t.Cleanup(func() { db.Callback().Update().Remove("test:block_claim") })
|
||||
job, err := wp.Enqueue(context.Background(), "shutdown_claim_race", nil)
|
||||
require.NoError(t, err)
|
||||
<-claimStarted
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
||||
defer cancel()
|
||||
require.ErrorIs(t, wp.Shutdown(ctx), context.DeadlineExceeded)
|
||||
|
||||
require.Equal(t, model.BackgroundJobStatusQueued, loadJob(t, db, job.ID).Status)
|
||||
}
|
||||
|
||||
// TestRedisEnqueueAndProcessEndToEnd verifies the full Redis path: Enqueue
|
||||
// XADDs to the stream, XREADGROUP picks it up, DB claim succeeds, handler
|
||||
// runs, and the job reaches "completed" status.
|
||||
|
||||
Regular → Executable
+78
-150
@@ -1,175 +1,103 @@
|
||||
#!/bin/bash
|
||||
# GoChat Health Check Script
|
||||
# Reference: Chatwoot health_check endpoint pattern
|
||||
# Checks: HTTP server, database, Redis, worker queues, WebSocket hub
|
||||
#
|
||||
# Usage:
|
||||
# ./scripts/health_check.sh [--full] [--json] [--timeout SECONDS]
|
||||
#
|
||||
# Exit codes:
|
||||
# 0 = healthy
|
||||
# 1 = degraded (some checks failed but core is OK)
|
||||
# 2 = unhealthy (critical checks failed)
|
||||
# Exit codes: 0 healthy, 1 degraded, 2 unhealthy/invalid invocation.
|
||||
|
||||
set -euo pipefail
|
||||
set -uo pipefail
|
||||
|
||||
# Configuration
|
||||
GOCHAT_HOST="${GOCHAT_HOST:-localhost}"
|
||||
GOCHAT_PORT="${GOCHAT_PORT:-3000}"
|
||||
GOCHAT_METRICS_PORT="${GOCHAT_METRICS_PORT:-9090}"
|
||||
GOCHAT_METRICS_PORT="${GOCHAT_METRICS_PORT:-${GOCHAT_PORT}}"
|
||||
TIMEOUT="${TIMEOUT:-5}"
|
||||
FULL_CHECK=false
|
||||
JSON_OUTPUT=false
|
||||
RESULTS=()
|
||||
CRITICAL_FAIL=0
|
||||
DEGRADED_FAIL=0
|
||||
|
||||
for arg in "$@"; do
|
||||
[[ "${arg}" == "--json" ]] && JSON_OUTPUT=true
|
||||
done
|
||||
|
||||
usage_error() {
|
||||
if [[ "${JSON_OUTPUT}" == "true" ]]; then
|
||||
printf '{"error":"%s","checks":[],"summary":{"critical_failures":1,"degraded_failures":0}}\n' "$1"
|
||||
else
|
||||
printf 'health_check: %s\n' "$1" >&2
|
||||
fi
|
||||
exit 2
|
||||
}
|
||||
|
||||
# Parse arguments
|
||||
while [[ $# -gt 0 ]]; do
|
||||
case "$1" in
|
||||
--full) FULL_CHECK=true; shift ;;
|
||||
--json) JSON_OUTPUT=true; shift ;;
|
||||
--timeout) TIMEOUT="$2"; shift 2 ;;
|
||||
--help) echo "Usage: $0 [--full] [--json] [--timeout SECONDS]"; exit 0 ;;
|
||||
*) echo "Unknown option: $1"; exit 1 ;;
|
||||
esac
|
||||
case "$1" in
|
||||
--full) FULL_CHECK=true; shift ;;
|
||||
--json) JSON_OUTPUT=true; shift ;;
|
||||
--timeout)
|
||||
[[ $# -ge 2 && "$2" =~ ^[1-9][0-9]*$ ]] || usage_error "--timeout requires a positive integer"
|
||||
TIMEOUT="$2"
|
||||
shift 2
|
||||
;;
|
||||
--help)
|
||||
if [[ "${JSON_OUTPUT}" == "true" ]]; then
|
||||
printf '{"usage":"health_check.sh [--full] [--json] [--timeout SECONDS]"}\n'
|
||||
else
|
||||
printf 'Usage: %s [--full] [--json] [--timeout SECONDS]\n' "$0"
|
||||
fi
|
||||
exit 0
|
||||
;;
|
||||
*) usage_error "unknown option" ;;
|
||||
esac
|
||||
done
|
||||
|
||||
BASE_URL="http://${GOCHAT_HOST}:${GOCHAT_PORT}"
|
||||
METRICS_URL="http://${GOCHAT_HOST}:${GOCHAT_METRICS_PORT}"
|
||||
|
||||
# Result tracking
|
||||
RESULTS=()
|
||||
CRITICAL_FAIL=0
|
||||
DEGRADED_FAIL=0
|
||||
|
||||
check_result() {
|
||||
local name="$1" status="$2" detail="$3"
|
||||
RESULTS+=("${name}|${status}|${detail}")
|
||||
if [[ "$status" == "CRITICAL_FAIL" ]]; then
|
||||
CRITICAL_FAIL=$((CRITICAL_FAIL + 1))
|
||||
elif [[ "$status" == "DEGRADED" ]]; then
|
||||
DEGRADED_FAIL=$((DEGRADED_FAIL + 1))
|
||||
fi
|
||||
local name="$1" status="$2" detail="$3"
|
||||
RESULTS+=("${name}|${status}|${detail}")
|
||||
[[ "${status}" == "CRITICAL_FAIL" ]] && CRITICAL_FAIL=$((CRITICAL_FAIL + 1))
|
||||
[[ "${status}" == "DEGRADED" ]] && DEGRADED_FAIL=$((DEGRADED_FAIL + 1))
|
||||
}
|
||||
|
||||
# ---- Check 1: HTTP liveness endpoint ----
|
||||
check_liveness() {
|
||||
local response
|
||||
response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/live" 2>&1) && {
|
||||
check_result "liveness" "OK" "${response}"
|
||||
} || {
|
||||
check_result "liveness" "CRITICAL_FAIL" "HTTP /live endpoint unreachable"
|
||||
}
|
||||
check_url() {
|
||||
local name="$1" severity="$2" url="$3" ok_detail="$4" fail_detail="$5"
|
||||
if curl -fsS --max-time "${TIMEOUT}" -o /dev/null "${url}"; then
|
||||
check_result "${name}" "OK" "${ok_detail}"
|
||||
else
|
||||
check_result "${name}" "${severity}" "${fail_detail}"
|
||||
fi
|
||||
}
|
||||
|
||||
# ---- Check 2: HTTP readiness endpoint ----
|
||||
check_readiness() {
|
||||
local response
|
||||
response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/ready" 2>&1) && {
|
||||
check_result "readiness" "OK" "${response}"
|
||||
} || {
|
||||
check_result "readiness" "CRITICAL_FAIL" "HTTP /ready endpoint unreachable"
|
||||
}
|
||||
}
|
||||
check_url "liveness" "CRITICAL_FAIL" "${BASE_URL}/live" "process is alive" "liveness endpoint unreachable"
|
||||
check_url "readiness" "CRITICAL_FAIL" "${BASE_URL}/ready" "dependencies are ready" "readiness endpoint failed"
|
||||
|
||||
# ---- Check 3: Database connectivity ----
|
||||
check_database() {
|
||||
if [[ "${FULL_CHECK}" == "true" ]]; then
|
||||
local response
|
||||
response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/health?check=database" 2>&1) && {
|
||||
check_result "database" "OK" "${response}"
|
||||
} || {
|
||||
check_result "database" "CRITICAL_FAIL" "Database connectivity failed"
|
||||
}
|
||||
else
|
||||
check_result "database" "SKIPPED" "Not in full mode"
|
||||
fi
|
||||
}
|
||||
|
||||
# ---- Check 4: Redis connectivity ----
|
||||
check_redis() {
|
||||
if [[ "${FULL_CHECK}" == "true" ]]; then
|
||||
local response
|
||||
response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/health?check=redis" 2>&1) && {
|
||||
check_result "redis" "OK" "${response}"
|
||||
} || {
|
||||
check_result "redis" "DEGRADED" "Redis connectivity failed (non-critical)"
|
||||
}
|
||||
else
|
||||
check_result "redis" "SKIPPED" "Not in full mode"
|
||||
fi
|
||||
}
|
||||
|
||||
# ---- Check 5: Metrics endpoint ----
|
||||
check_metrics() {
|
||||
local response
|
||||
response=$(curl -sf --max-time "${TIMEOUT}" "${METRICS_URL}/metrics" 2>&1) && {
|
||||
check_result "metrics" "OK" "Prometheus metrics endpoint responding"
|
||||
} || {
|
||||
check_result "metrics" "DEGRADED" "Metrics endpoint unreachable (non-critical)"
|
||||
}
|
||||
}
|
||||
|
||||
# ---- Check 6: WebSocket hub (full mode) ----
|
||||
check_websocket() {
|
||||
if [[ "${FULL_CHECK}" == "true" ]]; then
|
||||
# WebSocket health check via HTTP status endpoint
|
||||
local response
|
||||
response=$(curl -sf --max-time "${TIMEOUT}" "${BASE_URL}/health?check=websocket" 2>&1) && {
|
||||
check_result "websocket" "OK" "${response}"
|
||||
} || {
|
||||
check_result "websocket" "DEGRADED" "WebSocket hub unreachable"
|
||||
}
|
||||
else
|
||||
check_result "websocket" "SKIPPED" "Not in full mode"
|
||||
fi
|
||||
}
|
||||
|
||||
# ---- Run all checks ----
|
||||
check_liveness
|
||||
check_readiness
|
||||
check_database
|
||||
check_redis
|
||||
check_metrics
|
||||
check_websocket
|
||||
|
||||
# ---- Output results ----
|
||||
if [[ "${JSON_OUTPUT}" == "true" ]]; then
|
||||
# JSON output for programmatic consumption
|
||||
echo '{'
|
||||
echo ' "checks": ['
|
||||
for i in "${!RESULTS[@]}"; do
|
||||
IFS='|' read -r name status detail <<< "${RESULTS[$i]}"
|
||||
comma=""
|
||||
[[ $i -gt 0 ]] && comma=","
|
||||
echo " ${comma}{\"name\": \"${name}\", \"status\": \"${status}\", \"detail\": \"${detail}\"}"
|
||||
done
|
||||
echo ' ],'
|
||||
echo ' "summary": {'
|
||||
echo ' "critical_failures": ${CRITICAL_FAIL},'
|
||||
echo ' "degraded_failures": ${DEGRADED_FAIL}'
|
||||
echo ' }'
|
||||
echo '}'
|
||||
if [[ "${FULL_CHECK}" == "true" ]]; then
|
||||
check_url "database" "CRITICAL_FAIL" "${BASE_URL}/health?check=database" "database is healthy" "database check failed"
|
||||
check_url "redis" "CRITICAL_FAIL" "${BASE_URL}/health?check=redis" "redis is healthy" "redis check failed"
|
||||
else
|
||||
# Human-readable output
|
||||
echo "=== GoChat Health Check Report ==="
|
||||
for result in "${RESULTS[@]}"; do
|
||||
IFS='|' read -r name status detail <<< "${result}"
|
||||
case "$status" in
|
||||
OK) icon="✅" ;;
|
||||
CRITICAL_FAIL) icon="❌" ;;
|
||||
DEGRADED) icon="⚠️" ;;
|
||||
SKIPPED) icon="⏭️" ;;
|
||||
esac
|
||||
echo " ${icon} ${name}: ${status} — ${detail}"
|
||||
done
|
||||
echo ""
|
||||
echo "Critical failures: ${CRITICAL_FAIL} | Degraded: ${DEGRADED_FAIL}"
|
||||
check_result "database" "SKIPPED" "not in full mode"
|
||||
check_result "redis" "SKIPPED" "not in full mode"
|
||||
fi
|
||||
|
||||
# ---- Determine exit code ----
|
||||
if [[ ${CRITICAL_FAIL} -gt 0 ]]; then
|
||||
exit 2
|
||||
elif [[ ${DEGRADED_FAIL} -gt 0 ]]; then
|
||||
exit 1
|
||||
check_url "metrics" "DEGRADED" "${METRICS_URL}/metrics" "metrics endpoint is responding" "metrics endpoint unreachable"
|
||||
|
||||
if [[ "${JSON_OUTPUT}" == "true" ]]; then
|
||||
printf '{"checks":['
|
||||
for i in "${!RESULTS[@]}"; do
|
||||
IFS='|' read -r name status detail <<< "${RESULTS[$i]}"
|
||||
[[ $i -gt 0 ]] && printf ','
|
||||
printf '{"name":"%s","status":"%s","detail":"%s"}' "${name}" "${status}" "${detail}"
|
||||
done
|
||||
printf '],"summary":{"critical_failures":%d,"degraded_failures":%d}}\n' "${CRITICAL_FAIL}" "${DEGRADED_FAIL}"
|
||||
else
|
||||
exit 0
|
||||
fi
|
||||
printf '=== GoChat Health Check Report ===\n'
|
||||
for result in "${RESULTS[@]}"; do
|
||||
IFS='|' read -r name status detail <<< "${result}"
|
||||
printf ' %s: %s — %s\n' "${name}" "${status}" "${detail}"
|
||||
done
|
||||
printf 'Critical failures: %d | Degraded: %d\n' "${CRITICAL_FAIL}" "${DEGRADED_FAIL}"
|
||||
fi
|
||||
|
||||
if [[ ${CRITICAL_FAIL} -gt 0 ]]; then
|
||||
exit 2
|
||||
elif [[ ${DEGRADED_FAIL} -gt 0 ]]; then
|
||||
exit 1
|
||||
fi
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
package scripts
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/exec"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestHealthCheckJSONExitCodes(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
status func(string) int
|
||||
exitCode int
|
||||
}{
|
||||
{name: "healthy", status: func(string) int { return http.StatusOK }, exitCode: 0},
|
||||
{name: "degraded", status: func(path string) int {
|
||||
if path == "/metrics" {
|
||||
return http.StatusServiceUnavailable
|
||||
}
|
||||
return http.StatusOK
|
||||
}, exitCode: 1},
|
||||
{name: "unhealthy", status: func(path string) int {
|
||||
if path == "/live" || path == "/ready" {
|
||||
return http.StatusServiceUnavailable
|
||||
}
|
||||
return http.StatusOK
|
||||
}, exitCode: 2},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(test.status(r.URL.Path))
|
||||
}))
|
||||
defer server.Close()
|
||||
output, exitCode := runHealthScript(t, server.URL, "--json", "--full")
|
||||
require.Equal(t, test.exitCode, exitCode)
|
||||
var payload map[string]any
|
||||
require.NoError(t, json.Unmarshal(output, &payload), string(output))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHealthCheckJSONInvalidInvocation(t *testing.T) {
|
||||
command := exec.Command("./health_check.sh", "--json", "--timeout")
|
||||
output, err := command.Output()
|
||||
var exitErr *exec.ExitError
|
||||
require.True(t, errors.As(err, &exitErr))
|
||||
require.Equal(t, 2, exitErr.ExitCode())
|
||||
require.JSONEq(t, `{"error":"--timeout requires a positive integer","checks":[],"summary":{"critical_failures":1,"degraded_failures":0}}`, string(output))
|
||||
}
|
||||
|
||||
func runHealthScript(t *testing.T, rawURL string, args ...string) ([]byte, int) {
|
||||
t.Helper()
|
||||
parsed, err := url.Parse(rawURL)
|
||||
require.NoError(t, err)
|
||||
host, port, err := net.SplitHostPort(parsed.Host)
|
||||
require.NoError(t, err)
|
||||
command := exec.Command("./health_check.sh", args...)
|
||||
command.Env = append(os.Environ(), "GOCHAT_HOST="+host, "GOCHAT_PORT="+port, "GOCHAT_METRICS_PORT="+port)
|
||||
output, err := command.Output()
|
||||
if err == nil {
|
||||
return output, 0
|
||||
}
|
||||
var exitErr *exec.ExitError
|
||||
require.True(t, errors.As(err, &exitErr), "health script failed to execute: %v", err)
|
||||
return output, exitErr.ExitCode()
|
||||
}
|
||||
Reference in New Issue
Block a user