diff --git a/.gitignore b/.gitignore index 0763166c..97f7f7de 100644 --- a/.gitignore +++ b/.gitignore @@ -66,7 +66,6 @@ go.work.sum # Build output build dist/ -/pkg/ artifacts/ work/ out/ diff --git a/app/app.go b/app/app.go index 1c621785..7ffc8434 100644 --- a/app/app.go +++ b/app/app.go @@ -5,9 +5,20 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "time" "github.com/gorilla/mux" + "go.opentelemetry.io/otel/sdk/resource" + semconv "go.opentelemetry.io/otel/semconv/v1.26.0" + "go.uber.org/atomic" + + "github.com/webitel/webitel-go-kit/infra/health" + healthhttp "github.com/webitel/webitel-go-kit/infra/health/http" + "github.com/webitel/webitel-go-kit/infra/health/sdnotify" + otelsdk "github.com/webitel/webitel-go-kit/otel/sdk" + "github.com/webitel/wlog" + "github.com/webitel/engine/app/cc" "github.com/webitel/engine/app/flow" "github.com/webitel/engine/call_manager" @@ -20,11 +31,7 @@ import ( "github.com/webitel/engine/pkg/wbt/chat_manager" "github.com/webitel/engine/store" "github.com/webitel/engine/store/sqlstore" - otelsdk "github.com/webitel/webitel-go-kit/otel/sdk" - "github.com/webitel/wlog" - "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.26.0" - "go.uber.org/atomic" + "github.com/webitel/engine/wlogslog" // -------------------- plugin(s) -------------------- // _ "github.com/webitel/webitel-go-kit/otel/sdk/log/otlp" @@ -64,10 +71,11 @@ type App struct { tracer *Tracer otelShutdownFunc otelsdk.ShutdownFunc eventTrigger EventTrigger + health *health.Registry + sdNotify *sdnotify.Notifier } func New(options ...string) (outApp *App, outErr error) { - config, err := loadConfig() if err != nil { return nil, err @@ -126,6 +134,29 @@ func New(options ...string) (outApp *App, outErr error) { wlog.RedirectStdLog(app.Log) wlog.InitGlobalLogger(app.Log) + // Before anything slow: systemd counts TimeoutStartSec from ExecStart. + healthLog := slog.New(wlogslog.NewHandler(app.Log)) + app.health = health.New(health.DefaultConfig(), healthLog) + + // app.ctx: a short-lived context would stop every check. + if err := app.health.Start(app.ctx); err != nil { + return nil, fmt.Errorf("unable to start health registry: %w", err) + } + + // nil when NOTIFY_SOCKET is unset; Start and Stop are both nil-safe. + app.sdNotify = sdnotify.New(app.health, + sdnotify.WithLogger(healthLog), + sdnotify.WithStartTimeout(time.Duration(config.Health.StartTimeout)*time.Second), + ) + if err := app.sdNotify.Start(app.ctx); err != nil { + return nil, fmt.Errorf("unable to start sd_notify: %w", err) + } + + // RootRouter, not the API subrouter: probes answer without a token. + app.Srv.RootRouter.Handle("/livez", healthhttp.LivenessHandler(app.health, healthhttp.WithLogger(healthLog))) + app.Srv.RootRouter.Handle("/readyz", healthhttp.ReadinessHandler(app.health, healthhttp.WithLogger(healthLog))) + app.Srv.RootRouter.Handle("/healthz", healthhttp.HealthHandler(app.health, healthhttp.WithLogger(healthLog))) + if err := app.setupCipher(); err != nil { return nil, err } @@ -164,7 +195,9 @@ func New(options ...string) (outApp *App, outErr error) { } } - app.Store = store.NewLayeredStore(sqlstore.NewSqlSupplier(app.Config().SqlSettings)) + // Concrete handle: store.Store does not expose Ping. + sqlSupplier := sqlstore.NewSqlSupplier(app.Config().SqlSettings) + app.Store = store.NewLayeredStore(sqlSupplier) app.MessageQueue = rabbit.NewRabbitMQ(app.Config().NodeName, &app.Config().MessageQueueSettings) app.MessageQueue.Start() @@ -217,12 +250,29 @@ func New(options ...string) (outApp *App, outErr error) { } } + // Critical is node-local only: a shared one drops the whole fleet at once. + app.health.Critical("grpc", health.ListenerCheck(app.GrpcServer.Listener())) + app.health.Critical("freeswitch", app.callManager.Ready) + app.health.Informational("postgres", sqlSupplier.Ping) + app.health.Informational("rabbitmq", app.MessageQueue.Ping) + return app, outErr } func (app *App) Shutdown() { wlog.Info("stopping Server...") + // First: stop advertising readiness before anything is torn down. + if app.health != nil { + ctx, cancel := context.WithTimeout(context.Background(), + time.Duration(app.config.Health.StopTimeout)*time.Second) + if err := health.Shutdown(ctx, app.health, app.sdNotify); err != nil { + wlog.Error(fmt.Sprintf("health shutdown: %s", err.Error())) + } + + cancel() + } + if app.Hubs != nil { app.Hubs.Clean() } @@ -264,9 +314,23 @@ func (app *App) CallManager() call_manager.CallManager { return app.callManager } +// Ready reports whether this node can take traffic, per the health registry. func (app *App) Ready() (bool, model.AppError) { - //TODO - return true, nil + if app.health == nil { + return false, model.NewInternalError("app.ready.no_registry", "health registry is not initialized") + } + + ok, err := app.health.ReadyFunc()() + if ok { + return true, nil + } + + reason := "not ready" + if err != nil { + reason = err.Error() + } + + return false, model.NewInternalError("app.ready.not_ready", reason) } // DEPRECATED use SendDomainEvent instead @@ -285,18 +349,18 @@ func (a *App) PublishEventContext(ctx context.Context, body []byte, object strin type DomainEventType string const ( - CreateType DomainEventType = "create" + CreateType DomainEventType = "create" DeleteType DomainEventType = "delete" UpdateType DomainEventType = "update" ) type DomainEvent struct { - DomainID int64 - Object string + DomainID int64 + Object string EventType DomainEventType - User int64 - Time time.Time - Body any + User int64 + Time time.Time + Body any } func (d *DomainEvent) Validate() error { @@ -319,11 +383,11 @@ func formatDomainEventKey(event *DomainEvent) (string, error) { if event.Object == "" { return "", errors.New("object required") } - if event.EventType== "" { + + if event.EventType == "" { return "", errors.New("event type required") } return fmt.Sprintf("%s.%s.%d", event.Object, event.EventType, event.DomainID), nil - } func (a *App) SendDomainEvent(ctx context.Context, event *DomainEvent) error { @@ -340,6 +404,6 @@ func (a *App) SendDomainEvent(ctx context.Context, event *DomainEvent) error { if err != nil { return err } - + return a.MessageQueue.Send(ctx, EventExchangeName, routingKey, body) } diff --git a/app/cluster.go b/app/cluster.go index d377a2f2..48ddd9ff 100644 --- a/app/cluster.go +++ b/app/cluster.go @@ -17,9 +17,8 @@ func NewCluster(app *App) *cluster { } func (c *cluster) Start() error { - sd, err := discovery.NewServiceDiscovery(c.app.nodeId, c.app.Config().DiscoverySettings.Url, func() (b bool, appError error) { - return true, nil - }) + sd, err := discovery.NewServiceDiscovery(c.app.nodeId, c.app.Config().DiscoverySettings.Url, + c.app.health.ReadyFunc()) if err != nil { return err } diff --git a/app/grpc_server.go b/app/grpc_server.go index d9b90d5e..01a50549 100644 --- a/app/grpc_server.go +++ b/app/grpc_server.go @@ -41,6 +41,11 @@ type GrpcServer struct { lis net.Listener } +// Listener is the socket actually bound, not the address advertised to Consul. +func (grpc *GrpcServer) Listener() net.Listener { + return grpc.lis +} + func (grpc *GrpcServer) GetPublicInterface() (string, int) { h, p, _ := net.SplitHostPort(grpc.lis.Addr().String()) if h == "::" { diff --git a/call_manager/call_manager.go b/call_manager/call_manager.go index 886fad58..6b79b17e 100644 --- a/call_manager/call_manager.go +++ b/call_manager/call_manager.go @@ -2,6 +2,7 @@ package call_manager import ( "context" + "errors" "fmt" "strings" "sync" @@ -16,6 +17,12 @@ const ( WATCHER_INTERVAL = 1000 * 5 ) +// Plain errors: these travel to a health probe, not a client response. +var ( + ErrNoConnection = errors.New("no freeswitch connection registered") + ErrNotReady = errors.New("no freeswitch connection is ready") +) + type CallManager interface { Start() error Stop() @@ -23,6 +30,7 @@ type CallManager interface { Bridge(legA, legANode, legB, legBNode string) model.AppError CallClient() (CallClient, model.AppError) CallClientById(id string) (CallClient, model.AppError) + Ready(ctx context.Context) error SipWsAddress() string SipRouteUri() string @@ -111,6 +119,23 @@ func (c *callManager) CallClient() (CallClient, model.AppError) { return cli.(CallClient), nil } +// Ready reports whether any FreeSWITCH connection is usable. All, not +// CallClient: CallClient advances the round-robin marker used by real calls. +func (cm *callManager) Ready(context.Context) error { + conns := cm.poolConnections.All() + if len(conns) == 0 { + return ErrNoConnection + } + + for _, conn := range conns { + if conn != nil && conn.Ready() { + return nil + } + } + + return ErrNotReady +} + func (c *callManager) CallClientById(id string) (CallClient, model.AppError) { cli, err := c.poolConnections.GetById(id) if err != nil { diff --git a/call_manager/health_test.go b/call_manager/health_test.go new file mode 100644 index 00000000..0b7a1463 --- /dev/null +++ b/call_manager/health_test.go @@ -0,0 +1,104 @@ +package call_manager + +import ( + "context" + "errors" + "testing" + + "github.com/webitel/engine/pkg/discovery" +) + +type fakeConn struct { + name string + ready bool +} + +func (f *fakeConn) Name() string { return f.name } +func (f *fakeConn) Ready() bool { return f.ready } +func (f *fakeConn) Close() error { return nil } + +func newManager(conns ...discovery.Connection) *callManager { + pool := discovery.NewPoolConnections() + for _, c := range conns { + pool.Append(c) + } + + return &callManager{poolConnections: pool} +} + +func TestReady(t *testing.T) { + tests := []struct { + name string + conns []discovery.Connection + want error + }{ + { + name: "no connection registered", + conns: nil, + want: ErrNoConnection, + }, + { + name: "every connection down", + conns: []discovery.Connection{&fakeConn{name: "a"}, &fakeConn{name: "b"}}, + want: ErrNotReady, + }, + { + name: "one of two up", + conns: []discovery.Connection{&fakeConn{name: "a"}, &fakeConn{name: "b", ready: true}}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := newManager(tt.conns...).Ready(context.Background()); !errors.Is(err, tt.want) { + t.Errorf("Ready() = %v, want %v", err, tt.want) + } + }) + } +} + +// The check runs on a timer, so it must not consume a round-robin slot. +func TestReadyDoesNotAdvanceRoundRobin(t *testing.T) { + pick := func(probe bool) []string { + cm := newManager( + &fakeConn{name: "a", ready: true}, + &fakeConn{name: "b", ready: true}, + &fakeConn{name: "c", ready: true}, + ) + + got := make([]string, 0, 6) + + for range 6 { + if probe { + if err := cm.Ready(context.Background()); err != nil { + t.Fatalf("Ready() = %v, want nil", err) + } + } + + cli, err := cm.poolConnections.Get(discovery.StrategyRoundRobin) + if err != nil { + t.Fatalf("Get() = %v", err) + } + + got = append(got, cli.Name()) + } + + return got + } + + undisturbed, probed := pick(false), pick(true) + for i := range undisturbed { + if undisturbed[i] != probed[i] { + t.Fatalf("probing shifted the round-robin marker: %v, want %v", probed, undisturbed) + } + } +} + +// A typed nil reads as non-nil, and the Consul TTL updater calls .Error() on it. +func TestReadyReturnsNoTypedNil(t *testing.T) { + cm := newManager(&fakeConn{name: "a", ready: true}) + + if err := cm.Ready(context.Background()); err != nil { + t.Errorf("Ready() = %#v, want an untyped nil", err) + } +} diff --git a/deploy/systemd/webitel-engine.service b/deploy/systemd/webitel-engine.service index 00b08ca0..737e7724 100644 --- a/deploy/systemd/webitel-engine.service +++ b/deploy/systemd/webitel-engine.service @@ -5,7 +5,7 @@ StartLimitIntervalSec=60 StartLimitBurst=3 [Service] -Type=simple +Type=notify User=webitel Group=webitel LogsDirectory=webitel @@ -20,7 +20,8 @@ Restart=on-failure RestartSec=5 KillMode=mixed KillSignal=SIGTERM -TimeoutStartSec=0 +# Must exceed HEALTH_START_TIMEOUT, or the fallback READY=1 lands too late. +TimeoutStartSec=90 TimeoutStopSec=30 LimitNOFILE=64000 LimitNPROC=4096 diff --git a/go.mod b/go.mod index 935b8b6e..2b21f39c 100644 --- a/go.mod +++ b/go.mod @@ -27,6 +27,7 @@ require ( github.com/webitel/engine/pkg/wbt v0.0.0-20250801070656-122a5f61b06a github.com/webitel/engine/pkg/werror v0.0.0-20250508121332-6ae1563235d8 github.com/webitel/webitel-go-kit v0.0.13-0.20240908192731-3abe573c0e41 + github.com/webitel/webitel-go-kit/infra/health v0.0.0-20260813065449-ced9c0a18c53 github.com/webitel/wlog v0.0.0-20250325101442-de4f125c1ec7 go.opentelemetry.io/otel v1.36.0 go.opentelemetry.io/otel/sdk v1.36.0 diff --git a/go.sum b/go.sum index ffbae87c..bf2339b1 100644 --- a/go.sum +++ b/go.sum @@ -337,6 +337,8 @@ github.com/webitel/engine/pkg/werror v0.0.0-20250508121332-6ae1563235d8 h1:3++Aq github.com/webitel/engine/pkg/werror v0.0.0-20250508121332-6ae1563235d8/go.mod h1:xLS6bkOYzvYv0dYXUkd5yvYOtujUrXf9lS0gd4qAGO4= github.com/webitel/webitel-go-kit v0.0.13-0.20240908192731-3abe573c0e41 h1:vj6qE8RtTyz8B4syfUDCkZqULLJ/4I+LS0Rw5W7mao0= github.com/webitel/webitel-go-kit v0.0.13-0.20240908192731-3abe573c0e41/go.mod h1:MT93dkScj2kDwVudSdXr0MJdGZ+JvvHbYEsSSb2n+Aw= +github.com/webitel/webitel-go-kit/infra/health v0.0.0-20260813065449-ced9c0a18c53 h1:65vHCDXwJ8QzQvbhk7X74bfDRJcX+abx+dz6MrmXRQ0= +github.com/webitel/webitel-go-kit/infra/health v0.0.0-20260813065449-ced9c0a18c53/go.mod h1:qumOZfpiFfBSIxGd3CMiEyiNKY4xNaVpUh6nlnTgoPk= github.com/webitel/wlog v0.0.0-20250325101442-de4f125c1ec7 h1:Qi/ustGBfh13I6aZ/ljul679SPJmHbZzFn9h43eloLg= github.com/webitel/wlog v0.0.0-20250325101442-de4f125c1ec7/go.mod h1:mXyM8hL9tEBLM4K+Uw8QLuGLo6/eX+5Lvv1ks2Y4us8= github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU= diff --git a/model/config.go b/model/config.go index 2d7b949c..9e13e087 100644 --- a/model/config.go +++ b/model/config.go @@ -55,6 +55,7 @@ type Config struct { PublicHostName *string `json:"public_host" flag:"public_host||Public hostname" default:"" env:"PUBLIC_HOST"` Push PushConfig Log LogSettings `json:"log"` + Health HealthSettings `json:"health"` TriggersSettings TriggersSettings `json:"triggers_settings"` RTCConfiguration string `json:"rtc_configuration" flag:"rtc_configuration||RTCConfiguration" default:"" env:"RTC_CONFIGURATION"` } @@ -102,6 +103,13 @@ type SqlSettings struct { QueryTimeout *int `json:"query_timeout" flag:"sql_query_timeout|10|Sql query timeout seconds" env:"QUERY_TIMEOUT"` } +type HealthSettings struct { + // Must stay under the unit's TimeoutStartSec. + StartTimeout int `json:"start_timeout" flag:"health_start_timeout|60|Seconds before sd_notify reports READY=1 regardless of check state" env:"HEALTH_START_TIMEOUT"` + // Must exceed the package's DrainHold and fit inside TimeoutStopSec. + StopTimeout int `json:"stop_timeout" flag:"health_stop_timeout|12|Seconds budget for the readiness drain on shutdown" env:"HEALTH_STOP_TIMEOUT"` +} + type TriggersSettings struct { Enabled bool `json:"enabled" flag:"trigger_enabled|true|Enable trigger" env:"TRIGGER_ENABLED"` BrokerUrl string `json:"broker_url" flag:"broker_url||Broker for CaseTriggers" default:"" env:"TRIGGER_BROKER_URL"` diff --git a/mq/layered_mq.go b/mq/layered_mq.go index 1eb80e1e..4d07e4e9 100644 --- a/mq/layered_mq.go +++ b/mq/layered_mq.go @@ -34,6 +34,10 @@ func (l *LayeredMQ) Close() { l.MQLayer.Close() } +func (l *LayeredMQ) Ping(ctx context.Context) error { + return l.MQLayer.Ping(ctx) +} + func (l *LayeredMQ) BindCallEvents(domainId, userId int64) error { return l.MQLayer.BindCallEvents(domainId, userId) } diff --git a/mq/mq.go b/mq/mq.go index dd527c06..7bb6d27c 100644 --- a/mq/mq.go +++ b/mq/mq.go @@ -12,6 +12,8 @@ type MQ interface { Start() Close() + // Ping reports whether the broker connection is usable. + Ping(ctx context.Context) error NewDomainQueue(domainId int64, bindings model.GetAllBindings) (DomainQueue, model.AppError) RegisterWebsocket(domainId int64, event *model.RegisterToWebsocketEvent) model.AppError diff --git a/mq/rabbit/client.go b/mq/rabbit/client.go index 7f07a5cf..4d8cb2d2 100644 --- a/mq/rabbit/client.go +++ b/mq/rabbit/client.go @@ -3,6 +3,7 @@ package rabbit import ( "context" "encoding/json" + stderrors "errors" "fmt" "os" "sync" @@ -32,6 +33,12 @@ const ( callServiceHangupData = `{"hangup_by":"service","cause":"SYSTEM_SHUTDOWN","sip":501}` ) +// Stdlib errors: this file's `errors` is pkg/errors, which attaches stack traces. +var ( + errConnectionClosed = stderrors.New("amqp: connection is closed") + errChannelClosed = stderrors.New("amqp: channel is closed") +) + var errMaxRegisterQueueSize = model.NewInternalError("amqp.register_domain.max_queue_size", "") var errMaxUnRegisterQueueSize = model.NewInternalError("amqp.un_register_domain.max_queue_size", "") @@ -88,6 +95,22 @@ func (a *AMQP) Start() { go a.Listen() } +// Ping reads the cached connection's state; it does not dial. +func (a *AMQP) Ping(context.Context) error { + a.mx.Lock() + defer a.mx.Unlock() + + if a.connection == nil || a.connection.IsClosed() { + return errConnectionClosed + } + + if a.channel == nil || a.channel.IsClosed() { + return errChannelClosed + } + + return nil +} + func (a *AMQP) addDomainQueue(id int64, q mq.DomainQueue) { a.mx.Lock() defer a.mx.Unlock() diff --git a/pkg/discovery/consul.go b/pkg/discovery/consul.go index bef1ac06..3fc307ad 100644 --- a/pkg/discovery/consul.go +++ b/pkg/discovery/consul.go @@ -93,6 +93,24 @@ func (c *consul) RegisterService(name, pubHost string, pubPort int, ttl, critica } func (c *consul) register(as *api.AgentServiceRegistration) error { + if err := c.putRegistration(as); err != nil { + return err + } + + c.update(as) + + wlog.Info(fmt.Sprintf("started consul service id: %s", c.id)) + + go c.updateTTL(c.ttl/2, as) + + return nil +} + +// putRegistration registers the service and refreshes the check id. It neither +// updates the TTL nor starts an updater, so recovery can call it from inside +// the updater without leaving a second one behind or recursing back into +// handlePassTTLError. The next tick reports the verdict. +func (c *consul) putRegistration(as *api.AgentServiceRegistration) error { var err error if err = c.agent.ServiceRegister(as); err != nil { return err @@ -113,41 +131,67 @@ func (c *consul) register(as *api.AgentServiceRegistration) error { if serviceCheck == nil { return errors.New("serviceCheck is null") } + c.checkId = serviceCheck.CheckID - c.update(as) - wlog.Info(fmt.Sprintf("started consul service id: %s", c.id)) + return nil +} - go c.updateTTL(c.ttl/2, as) +// ttlVerdict maps a health verdict to a Consul TTL status and output. +// (false, nil) should not occur, but it is engine that crashes if it does. +func ttlVerdict(ok bool, err error) (pass bool, output string) { + if ok { + return true, "ready..." + } - return nil + if err == nil { + return false, "not ready" + } + + return false, err.Error() } func (c *consul) update(as *api.AgentServiceRegistration) { - ok, err := c.check() - if !ok { - if agentErr := c.agent.FailTTL(c.checkId, err.Error()); agentErr != nil { - c.handlePassTTLError(agentErr, as) - } - } else { - if agentErr := c.agent.PassTTL(c.checkId, "ready..."); agentErr != nil { + pass, output := ttlVerdict(c.check()) + + if pass { + if agentErr := c.agent.PassTTL(c.checkId, output); agentErr != nil { c.handlePassTTLError(agentErr, as) } + + return + } + + if agentErr := c.agent.FailTTL(c.checkId, output); agentErr != nil { + c.handlePassTTLError(agentErr, as) } } +// shouldReregister reports whether a TTL update failure means the agent no +// longer holds our registration. 404 is the one that matters now that +// readiness can be false: Consul drops the service after +// DeregisterCriticalServiceAfter, and without this the node stays healthy but +// invisible to discovery forever. +func shouldReregister(err error) bool { + var status api.StatusError + if !errors.As(err, &status) { + return false + } + + return status.Code == http.StatusInternalServerError || status.Code == http.StatusNotFound +} + func (c *consul) handlePassTTLError(err error, as *api.AgentServiceRegistration) { - switch wrapErr := err.(type) { - case api.StatusError: - if wrapErr.Code == http.StatusInternalServerError { - // reconnect ? - wlog.Info(fmt.Sprintf("reconnect consul service id: %s", c.id)) - if e := c.register(as); e != nil { - wlog.Error(fmt.Sprintf("reconnect consul service %s error: %s", c.id, e.Error())) - } - } - default: + if !shouldReregister(err) { wlog.Error(err.Error()) + + return + } + + wlog.Info(fmt.Sprintf("reconnect consul service id: %s", c.id)) + + if e := c.putRegistration(as); e != nil { + wlog.Error(fmt.Sprintf("reconnect consul service %s error: %s", c.id, e.Error())) } } diff --git a/pkg/discovery/consul_test.go b/pkg/discovery/consul_test.go new file mode 100644 index 00000000..ae4f2e6d --- /dev/null +++ b/pkg/discovery/consul_test.go @@ -0,0 +1,137 @@ +package discovery + +import ( + "errors" + "fmt" + "net/http" + "os" + "strings" + "testing" + + "github.com/hashicorp/consul/api" +) + +// (false, nil) is the case that matters: update() used to panic on that branch. +func TestTTLVerdict(t *testing.T) { + tests := []struct { + name string + ok bool + err error + wantPass bool + wantOutput string + }{ + { + name: "ready", + ok: true, + err: nil, + wantPass: true, + wantOutput: "ready...", + }, + { + name: "not ready with reason", + ok: false, + err: errors.New("grpc listener not accepting"), + wantPass: false, + wantOutput: "grpc listener not accepting", + }, + { + name: "not ready without an error must not panic", + ok: false, + err: nil, + wantPass: false, + wantOutput: "not ready", + }, + { + name: "ready wins even if an error is handed along", + ok: true, + err: errors.New("ignored"), + wantPass: true, + wantOutput: "ready...", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pass, output := ttlVerdict(tt.ok, tt.err) + + if pass != tt.wantPass { + t.Errorf("pass = %v, want %v", pass, tt.wantPass) + } + + if output != tt.wantOutput { + t.Errorf("output = %q, want %q", output, tt.wantOutput) + } + }) + } +} + +// Consul answers 404 once DeregisterCriticalServiceAfter has dropped the +// service. Treating that as fatal leaves a recovered node out of discovery. +func TestShouldReregister(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"service was deregistered", api.StatusError{Code: http.StatusNotFound}, true}, + {"agent internal error", api.StatusError{Code: http.StatusInternalServerError}, true}, + {"bad request", api.StatusError{Code: http.StatusBadRequest}, false}, + {"wrapped 404", fmt.Errorf("update ttl: %w", api.StatusError{Code: http.StatusNotFound}), true}, + {"network error", errors.New("connection refused"), false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := shouldReregister(tt.err); got != tt.want { + t.Errorf("shouldReregister(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +// Recovery runs inside the TTL updater, so it must not start another one: +// each extra updater duplicates TTL traffic and races on checkId. +func TestReregisterStartsNoSecondUpdater(t *testing.T) { + src, err := os.ReadFile("consul.go") + if err != nil { + t.Fatal(err) + } + + body := funcBody(t, string(src), "func (c *consul) handlePassTTLError(") + if strings.Contains(body, "c.register(") { + t.Error("handlePassTTLError calls register, which spawns a TTL updater; use putRegistration") + } + + if !strings.Contains(body, "c.putRegistration(") { + t.Error("handlePassTTLError no longer re-registers") + } + + // putRegistration must not call update: update is what routes failures back + // into handlePassTTLError, which would recurse. + if put := funcBody(t, string(src), "func (c *consul) putRegistration("); strings.Contains(put, "c.update(") { + t.Error("putRegistration calls update, which can recurse through handlePassTTLError") + } + + if n := strings.Count(string(src), "go c.updateTTL("); n != 1 { + t.Errorf("found %d updateTTL launches, want exactly 1", n) + } +} + +// funcBody returns the source between a function's signature and its closing brace. +func funcBody(t *testing.T, src, signature string) string { + t.Helper() + + i := strings.Index(src, signature) + if i < 0 { + t.Fatalf("not found: %s", signature) + } + + rest := src[i:] + + end := strings.Index(rest, "\n}") + if end < 0 { + t.Fatalf("no closing brace: %s", signature) + } + + return rest[:end] +} diff --git a/store/sqlstore/supplier.go b/store/sqlstore/supplier.go index 23b6a423..c0b86837 100644 --- a/store/sqlstore/supplier.go +++ b/store/sqlstore/supplier.go @@ -278,6 +278,11 @@ func (ss *SqlSupplier) GetMaster() *gorp.DbMap { return ss.master } +// Ping reports whether the master connection is usable. +func (s *SqlSupplier) Ping(ctx context.Context) error { + return s.master.Db.PingContext(ctx) +} + func (ss *SqlSupplier) GetReplica() *gorp.DbMap { if len(ss.settings.DataSourceReplicas) == 0 || ss.lockedToMaster { return ss.GetMaster() diff --git a/wlogslog/wlogslog.go b/wlogslog/wlogslog.go new file mode 100644 index 00000000..66af21b9 --- /dev/null +++ b/wlogslog/wlogslog.go @@ -0,0 +1,107 @@ +// Package wlogslog bridges log/slog to wlog, for libraries that take a *slog.Logger. +package wlogslog + +import ( + "context" + "log/slog" + + "github.com/webitel/wlog" +) + +type handler struct { + log *wlog.Logger + fields []wlog.Field + // prefix is the accumulated group path, "" or "a.b.". Dotted keys rather + // than wlog.Namespace: a zap namespace stays open, so siblings would nest. + prefix string +} + +// NewHandler returns a slog.Handler that writes through to log. +func NewHandler(log *wlog.Logger) slog.Handler { + return &handler{log: log} +} + +// wlog exposes no level query, so filtering is left to wlog. +func (h *handler) Enabled(context.Context, slog.Level) bool { + return true +} + +func (h *handler) Handle(_ context.Context, rec slog.Record) error { + fields := make([]wlog.Field, 0, len(h.fields)+rec.NumAttrs()) + fields = append(fields, h.fields...) + + rec.Attrs(func(a slog.Attr) bool { + fields = appendAttr(fields, a, h.prefix) + + return true + }) + + switch { + case rec.Level >= slog.LevelError: + h.log.Error(rec.Message, fields...) + case rec.Level >= slog.LevelWarn: + h.log.Warn(rec.Message, fields...) + case rec.Level >= slog.LevelInfo: + h.log.Info(rec.Message, fields...) + default: + h.log.Debug(rec.Message, fields...) + } + + return nil +} + +// Copies rather than mutates: slog handlers must be safe to share. +func (h *handler) WithAttrs(attrs []slog.Attr) slog.Handler { + if len(attrs) == 0 { + return h + } + + fields := make([]wlog.Field, len(h.fields), len(h.fields)+len(attrs)) + copy(fields, h.fields) + + for _, a := range attrs { + fields = appendAttr(fields, a, h.prefix) + } + + return &handler{log: h.log, fields: fields, prefix: h.prefix} +} + +func (h *handler) WithGroup(name string) slog.Handler { + if name == "" { + return h + } + + fields := make([]wlog.Field, len(h.fields)) + copy(fields, h.fields) + + return &handler{log: h.log, fields: fields, prefix: h.prefix + name + "."} +} + +// appendAttr follows the slog contract: resolve LogValuer, drop wholly empty +// attrs and empty groups, inline a group whose key is empty. +func appendAttr(fields []wlog.Field, a slog.Attr, prefix string) []wlog.Field { + a.Value = a.Value.Resolve() + + if a.Equal(slog.Attr{}) { + return fields + } + + if a.Value.Kind() == slog.KindGroup { + group := a.Value.Group() + if len(group) == 0 { + return fields + } + + if a.Key != "" { + prefix += a.Key + "." + } + + for _, ga := range group { + fields = appendAttr(fields, ga, prefix) + } + + return fields + } + + return append(fields, wlog.Any(prefix+a.Key, a.Value.Any())) +} diff --git a/wlogslog/wlogslog_test.go b/wlogslog/wlogslog_test.go new file mode 100644 index 00000000..bba67b20 --- /dev/null +++ b/wlogslog/wlogslog_test.go @@ -0,0 +1,242 @@ +package wlogslog + +import ( + "context" + "encoding/json" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/webitel/wlog" +) + +// newTestLogger returns a wlog logger writing JSON to a temp file, plus a reader. +func newTestLogger(t *testing.T) (*wlog.Logger, func() string) { + t.Helper() + + path := filepath.Join(t.TempDir(), "test.log") + log := wlog.NewLogger(&wlog.LoggerConfiguration{ + EnableFile: true, + FileJson: true, + FileLevel: "debug", + FileLocation: path, + }) + + return log, func() string { + b, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read log: %v", err) + } + + return string(b) + } +} + +func TestWlogHandlerEnabledAlwaysTrue(t *testing.T) { + // wlog exposes no level query, so the handler must not filter. + log, _ := newTestLogger(t) + h := NewHandler(log) + + for _, lvl := range []slog.Level{slog.LevelDebug, slog.LevelInfo, slog.LevelWarn, slog.LevelError} { + if !h.Enabled(context.Background(), lvl) { + t.Errorf("Enabled(%v) = false, want true", lvl) + } + } +} + +func TestWlogHandlerWritesThroughToWlog(t *testing.T) { + log, read := newTestLogger(t) + slog.New(NewHandler(log)).Info("registry started", "checks", 4) + + out := read() + if !strings.Contains(out, "registry started") { + t.Errorf("message missing from wlog output: %s", out) + } + + if !strings.Contains(out, "checks") { + t.Errorf("attribute missing from wlog output: %s", out) + } +} + +func TestWlogHandlerLevelMapping(t *testing.T) { + tests := []struct { + name string + level slog.Level + want string + }{ + {"debug", slog.LevelDebug, "DEBUG"}, + {"info", slog.LevelInfo, "INFO"}, + {"warn", slog.LevelWarn, "WARN"}, + {"error", slog.LevelError, "ERROR"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + log, read := newTestLogger(t) + slog.New(NewHandler(log)).Log(context.Background(), tt.level, "msg-"+tt.name) + + var rec struct { + Level string `json:"level"` + } + + line := strings.TrimSpace(read()) + if line == "" { + t.Fatal("nothing logged") + } + + if err := json.Unmarshal([]byte(line), &rec); err != nil { + t.Fatalf("log line is not JSON: %q", line) + } + + if rec.Level != tt.want { + t.Errorf("level = %q, want %q", rec.Level, tt.want) + } + }) + } +} + +// Handlers must be safe to share, so WithAttrs must copy rather than mutate. +func TestWlogHandlerWithAttrsDoesNotLeak(t *testing.T) { + log, _ := newTestLogger(t) + base := NewHandler(log) + + a := base.WithAttrs([]slog.Attr{slog.String("branch", "a")}) + b := base.WithAttrs([]slog.Attr{slog.String("branch", "b")}) + + if got := len(base.(*handler).fields); got != 0 { + t.Errorf("base handler mutated: has %d fields, want 0", got) + } + + if got := len(a.(*handler).fields); got != 1 { + t.Errorf("branch a: %d fields, want 1", got) + } + + if got := len(b.(*handler).fields); got != 1 { + t.Errorf("branch b: %d fields, want 1", got) + } + + // A second derivation must not disturb the first. + a2 := a.WithAttrs([]slog.Attr{slog.String("extra", "x")}) + if got := len(a.(*handler).fields); got != 1 { + t.Errorf("parent grew to %d fields after deriving a child", got) + } + + if got := len(a2.(*handler).fields); got != 2 { + t.Errorf("child: %d fields, want 2", got) + } +} + +func TestWlogHandlerWithGroupAndEmptyCases(t *testing.T) { + log, _ := newTestLogger(t) + base := NewHandler(log) + + if got := base.WithAttrs(nil); got != base { + t.Error("WithAttrs(nil) should return the receiver unchanged") + } + + if got := base.WithGroup(""); got != base { + t.Error(`WithGroup("") should return the receiver unchanged`) + } + + // WithGroup qualifies later keys via a prefix; it emits no field of its own. + sub := base.WithGroup("sub").(*handler) + if sub.prefix != "sub." { + t.Errorf("prefix = %q, want %q", sub.prefix, "sub.") + } + + if got := len(sub.fields); got != 0 { + t.Errorf("WithGroup should add no field, got %d", got) + } +} + +// decode the single JSON line the logger wrote +func logLine(t *testing.T, read func() string) map[string]any { + t.Helper() + + line := strings.TrimSpace(read()) + if line == "" { + t.Fatal("nothing logged") + } + + var m map[string]any + if err := json.Unmarshal([]byte(line), &m); err != nil { + t.Fatalf("not JSON: %q", line) + } + + return m +} + +type resolvable struct{ v string } + +func (r resolvable) LogValue() slog.Value { return slog.StringValue(r.v) } + +func TestAttrLogValuerIsResolved(t *testing.T) { + log, read := newTestLogger(t) + slog.New(NewHandler(log)).Info("m", "k", resolvable{v: "resolved"}) + + if got := logLine(t, read)["k"]; got != "resolved" { + t.Errorf("k = %#v, want \"resolved\" — LogValuer was not resolved", got) + } +} + +func TestAttrEmptyAndEmptyGroupAreDropped(t *testing.T) { + log, read := newTestLogger(t) + slog.New(NewHandler(log)).Info("m", + slog.Attr{}, + slog.Group("empty"), + slog.String("kept", "yes"), + ) + + m := logLine(t, read) + if _, ok := m["empty"]; ok { + t.Error("an empty group was emitted") + } + + if m["kept"] != "yes" { + t.Errorf("real attr lost: %#v", m) + } +} + +func TestAttrNamedGroupIsQualified(t *testing.T) { + log, read := newTestLogger(t) + slog.New(NewHandler(log)).Info("m", slog.Group("db", slog.String("host", "h"))) + + if got := logLine(t, read)["db.host"]; got != "h" { + t.Errorf("want db.host=h, got %#v", got) + } +} + +// Only a wholly empty Attr is dropped; an empty key carrying a value is not. +func TestAttrEmptyKeyWithValueIsKept(t *testing.T) { + log, read := newTestLogger(t) + slog.New(NewHandler(log)).Info("m", slog.String("", "v")) + + if got := logLine(t, read)[""]; got != "v" { + t.Errorf(`want ""="v", got %#v`, got) + } +} + +// A group with an empty key contributes no qualifier. +func TestAttrEmptyKeyGroupIsInlined(t *testing.T) { + log, read := newTestLogger(t) + slog.New(NewHandler(log)).Info("m", slog.Group("", slog.String("host", "h"))) + + m := logLine(t, read) + if got := m["host"]; got != "h" { + t.Errorf("want host=h inlined, got %#v", m) + } +} + +// Sibling groups must sit beside each other, not nest. +func TestWithGroupSiblingsDoNotNest(t *testing.T) { + log, read := newTestLogger(t) + base := slog.New(NewHandler(log)) + base.WithGroup("a").With("x", 1).WithGroup("b").Info("m", "y", 2) + + m := logLine(t, read) + if m["a.x"] == nil || m["a.b.y"] == nil { + t.Errorf("want a.x and a.b.y, got %#v", m) + } +}