From 12f81945b628e57a41b033c701bccebbcd6fd621 Mon Sep 17 00:00:00 2001 From: vitaliikucherov <312534721+vitaliikucherov@users.noreply.github.com> Date: Thu, 13 Aug 2026 10:06:13 +0300 Subject: [PATCH 1/5] WTEL-10090: Integrate infra/health probes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real readiness replaces the always-ready stub in cluster.go, /livez /readyz /healthz hang off RootRouter, and sd_notify reports to systemd. gRPC and FreeSWITCH are critical: both node-local. Postgres and RabbitMQ are shared, so informational — critical would evacuate the whole fleet at once. Drain runs first in Shutdown so Consul sees the node leave before its dependencies go. That makes the (false, nil) path in consul.go reachable, where err.Error() was unguarded — extracted to ttlVerdict and tested. .gitignore: drop the stale GOPATH-era /pkg/ rule, which was hiding source. --- .gitignore | 1 - app/app.go | 100 +++++++++++++++--- app/cluster.go | 5 +- app/grpc_server.go | 6 ++ app/health.go | 26 +++++ app/health_test.go | 69 +++++++++++++ deploy/systemd/webitel-engine.service | 11 +- go.mod | 1 + go.sum | 2 + mq/rabbit/client.go | 26 +++++ pkg/discovery/consul.go | 32 ++++-- pkg/discovery/consul_test.go | 67 +++++++++++++ wlogslog/wlogslog.go | 74 ++++++++++++++ wlogslog/wlogslog_test.go | 139 ++++++++++++++++++++++++++ 14 files changed, 534 insertions(+), 25 deletions(-) create mode 100644 app/health.go create mode 100644 app/health_test.go create mode 100644 pkg/discovery/consul_test.go create mode 100644 wlogslog/wlogslog.go create mode 100644 wlogslog/wlogslog_test.go 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..de58477e 100644 --- a/app/app.go +++ b/app/app.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "time" "github.com/gorilla/mux" @@ -20,6 +21,10 @@ import ( "github.com/webitel/engine/pkg/wbt/chat_manager" "github.com/webitel/engine/store" "github.com/webitel/engine/store/sqlstore" + "github.com/webitel/engine/wlogslog" + "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" "go.opentelemetry.io/otel/sdk/resource" @@ -64,6 +69,8 @@ type App struct { tracer *Tracer otelShutdownFunc otelsdk.ShutdownFunc eventTrigger EventTrigger + health *health.Registry + sdNotify *sdnotify.Notifier } func New(options ...string) (outApp *App, outErr error) { @@ -164,7 +171,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 GetMaster. + sqlSupplier := sqlstore.NewSqlSupplier(app.Config().SqlSettings) + app.Store = store.NewLayeredStore(sqlSupplier) app.MessageQueue = rabbit.NewRabbitMQ(app.Config().NodeName, &app.Config().MessageQueueSettings) app.MessageQueue.Start() @@ -175,6 +184,16 @@ func New(options ...string) (outApp *App, outErr error) { app.GrpcServer = NewGrpcServer(app, app.Config().ServerSettings) + // Must exist before cluster.Start, which hands the verdict to Consul. + // Checks are registered at the end of New, once every manager exists. + healthLog := slog.New(wlogslog.NewHandler(app.Log)) + app.health = health.New(health.DefaultConfig(), healthLog) + + // 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 outErr = app.cluster.Start(); outErr != nil { return nil, outErr } @@ -217,12 +236,55 @@ func New(options ...string) (outApp *App, outErr error) { } } + // Critical is for node-local faults only: a shared dependency marked + // critical would take the whole fleet out of rotation at once. Consul is + // deliberately unchecked — the verdict travels through it. + app.health.Critical("grpc", health.ListenerCheck(app.GrpcServer.Listener())) + app.health.Critical("freeswitch", freeswitchCheck(app.callManager)) + app.health.Informational("postgres", func(ctx context.Context) error { + return sqlSupplier.GetMaster().Db.PingContext(ctx) + }) + if p, ok := app.MessageQueue.(interface { + Ping(context.Context) error + }); ok { + app.health.Informational("rabbitmq", p.Ping) + } + + // app.ctx: Start's context is the scheduler's lifetime, so a short-lived + // one would silently 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(90*time.Second), + ) + if err := app.sdNotify.Start(app.ctx); err != nil { + return nil, fmt.Errorf("unable to start sd_notify: %w", err) + } + return app, outErr } func (app *App) Shutdown() { wlog.Info("stopping Server...") + // Drain before anything is torn down, so the node stops advertising + // readiness while its dependencies are still up. The DrainHold wait happens + // inside Stop: 12s clears the 10s hold and fits TimeoutStopSec=30. Stop also + // halts the scheduler before MessageQueue.Close, so the rabbitmq check + // cannot race a closing connection. + if app.health != nil { + ctx, cancel := context.WithTimeout(context.Background(), 12*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 +326,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 initialised") + } + + 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 +361,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 +395,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 +416,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..6303d84e 100644 --- a/app/grpc_server.go +++ b/app/grpc_server.go @@ -41,6 +41,12 @@ type GrpcServer struct { lis net.Listener } +// Listener is the socket actually bound, unlike GetPublicInterface which +// reports 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/app/health.go b/app/health.go new file mode 100644 index 00000000..04919839 --- /dev/null +++ b/app/health.go @@ -0,0 +1,26 @@ +package app + +import ( + "context" + "errors" + "fmt" + + "github.com/webitel/engine/call_manager" + "github.com/webitel/webitel-go-kit/infra/health" +) + +// freeswitchCheck reports whether this node's FreeSWITCH is usable. +func freeswitchCheck(cm call_manager.CallManager) health.Check { + return func(context.Context) error { + cli, appErr := cm.CallClient() + if appErr != nil { + return fmt.Errorf("freeswitch client unavailable: %w", appErr) + } + + if !cli.Ready() { + return errors.New("freeswitch client not ready") + } + + return nil + } +} diff --git a/app/health_test.go b/app/health_test.go new file mode 100644 index 00000000..af654d5f --- /dev/null +++ b/app/health_test.go @@ -0,0 +1,69 @@ +package app + +import ( + "context" + "testing" + + "github.com/webitel/engine/call_manager" + "github.com/webitel/engine/model" +) + +type fakeCallClient struct { + call_manager.CallClient + ready bool +} + +func (f *fakeCallClient) Ready() bool { return f.ready } + +type fakeCallManager struct { + call_manager.CallManager + cli call_manager.CallClient + err model.AppError +} + +func (f *fakeCallManager) CallClient() (call_manager.CallClient, model.AppError) { + return f.cli, f.err +} + +func TestFreeswitchCheck(t *testing.T) { + tests := []struct { + name string + cm call_manager.CallManager + wantErr bool + }{ + { + name: "client ready", + cm: &fakeCallManager{cli: &fakeCallClient{ready: true}}, + }, + { + name: "client not ready", + cm: &fakeCallManager{cli: &fakeCallClient{ready: false}}, + wantErr: true, + }, + { + name: "no client available", + cm: &fakeCallManager{err: model.NewInternalError("test.no_client", "none")}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := freeswitchCheck(tt.cm)(context.Background()) + if (err != nil) != tt.wantErr { + t.Errorf("err = %v, wantErr = %v", err, tt.wantErr) + } + }) + } +} + +// A health check must never hand back a nil-valued error interface: the Consul +// TTL updater treats a non-nil error as safe to call .Error() on. +func TestFreeswitchCheckNoTypedNilError(t *testing.T) { + var nilAppErr model.AppError // nil interface value of a concrete-free type + cm := &fakeCallManager{cli: &fakeCallClient{ready: true}, err: nilAppErr} + + if err := freeswitchCheck(cm)(context.Background()); err != nil { + t.Errorf("a nil AppError must not surface as a non-nil error, got %v", err) + } +} diff --git a/deploy/systemd/webitel-engine.service b/deploy/systemd/webitel-engine.service index 00b08ca0..89dcaeeb 100644 --- a/deploy/systemd/webitel-engine.service +++ b/deploy/systemd/webitel-engine.service @@ -5,7 +5,11 @@ StartLimitIntervalSec=60 StartLimitBurst=3 [Service] -Type=simple +# Type=notify: the health registry reports READY=1 once every critical check is +# green. WatchdogSec is deliberately absent — it stays off until it can be +# turned on together with the restart policy, or a wedged node either loops on +# restarts or parks in failed. +Type=notify User=webitel Group=webitel LogsDirectory=webitel @@ -20,7 +24,10 @@ Restart=on-failure RestartSec=5 KillMode=mixed KillSignal=SIGTERM -TimeoutStartSec=0 +# Finite, because under Type=notify a node whose critical check never goes +# green would otherwise sit in activating forever. sdnotify.WithStartTimeout +# matches this and sends READY=1 with STATUS=starting degraded instead. +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/mq/rabbit/client.go b/mq/rabbit/client.go index 7f07a5cf..45821e5b 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,13 @@ const ( callServiceHangupData = `{"hangup_by":"service","cause":"SYSTEM_SHUTDOWN","sip":501}` ) +// Stdlib errors: this file's `errors` is github.com/pkg/errors, whose New +// attaches a stack trace — too noisy for a check that runs every few seconds. +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 +96,24 @@ func (a *AMQP) Start() { go a.Listen() } +// Ping reports whether the broker connection is usable, read off the cached +// connection rather than dialing. Reached by an anonymous interface assertion +// so mq.MQ need not grow a method. +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..9204cc35 100644 --- a/pkg/discovery/consul.go +++ b/pkg/discovery/consul.go @@ -123,16 +123,34 @@ func (c *consul) register(as *api.AgentServiceRegistration) error { return nil } +// 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 — +// this branch previously called err.Error() unconditionally. +func ttlVerdict(ok bool, err error) (pass bool, output string) { + if ok { + return true, "ready..." + } + + 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) } } diff --git a/pkg/discovery/consul_test.go b/pkg/discovery/consul_test.go new file mode 100644 index 00000000..68177b63 --- /dev/null +++ b/pkg/discovery/consul_test.go @@ -0,0 +1,67 @@ +package discovery + +import ( + "errors" + "testing" +) + +// TestTTLVerdict covers the three shapes a health verdict can take on its way +// to Consul's TTL check. +// +// The case that matters is (false, nil). The health package promises a false +// verdict always carries a non-nil error, but engine is what panics if that +// promise is ever broken — update() used to call err.Error() unconditionally on +// the not-ok branch. This test fails against that code, and it is hermetic: no +// Consul agent, no network. +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) + } + }) + } +} diff --git a/wlogslog/wlogslog.go b/wlogslog/wlogslog.go new file mode 100644 index 00000000..395cbddf --- /dev/null +++ b/wlogslog/wlogslog.go @@ -0,0 +1,74 @@ +// Package wlogslog bridges log/slog to wlog, for libraries that take a +// *slog.Logger. It has no engine dependencies. +package wlogslog + +import ( + "context" + "log/slog" + + "github.com/webitel/wlog" +) + +type handler struct { + log *wlog.Logger + fields []wlog.Field +} + +// 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 = append(fields, wlog.Any(a.Key, a.Value.Any())) + 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 = append(fields, wlog.Any(a.Key, a.Value.Any())) + } + + return &handler{log: h.log, fields: fields} +} + +// wlog.Namespace is zap.Namespace, which already has slog's group semantics. +func (h *handler) WithGroup(name string) slog.Handler { + if name == "" { + return h + } + + fields := make([]wlog.Field, len(h.fields), len(h.fields)+1) + copy(fields, h.fields) + + return &handler{log: h.log, fields: append(fields, wlog.Namespace(name))} +} diff --git a/wlogslog/wlogslog_test.go b/wlogslog/wlogslog_test.go new file mode 100644 index 00000000..0a50f832 --- /dev/null +++ b/wlogslog/wlogslog_test.go @@ -0,0 +1,139 @@ +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 for whatever it wrote. +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 — otherwise + // it would silently drop records wlog would have kept. + 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) + } + }) + } +} + +// slog's contract says a handler must be safe to share, so WithAttrs must copy. +// Mutating in place would leak attributes between unrelated loggers. +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`) + } + if got := len(base.WithGroup("sub").(*handler).fields); got != 1 { + t.Errorf("WithGroup: %d fields, want 1", got) + } +} From a57b568f3e256f5625d3c994e3e41626a34323e5 Mon Sep 17 00:00:00 2001 From: vitaliikucherov <312534721+vitaliikucherov@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:00:31 +0300 Subject: [PATCH 2/5] WTEL-10090: Address review on #465 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Health and sd_notify now start right after the logger. systemd counts TimeoutStartSec from ExecStart, so starting the notifier after the managers meant its fallback READY=1 could land after systemd had already given up — with an unreachable database it heard nothing at all. Fallback is 60s, under the unit's 90. wlogslog follows the slog.Attr contract: resolve LogValuer, drop empty attrs and groups, inline empty-key groups, qualify named ones. Groups are dotted prefixes, not wlog.Namespace, which stays open and nests later siblings. --- app/app.go | 79 ++++++++++++++++-------------- app/health.go | 3 +- app/health_test.go | 3 ++ wlogslog/wlogslog.go | 53 +++++++++++++++++--- wlogslog/wlogslog_test.go | 100 +++++++++++++++++++++++++++++++++++++- 5 files changed, 193 insertions(+), 45 deletions(-) diff --git a/app/app.go b/app/app.go index de58477e..29375690 100644 --- a/app/app.go +++ b/app/app.go @@ -9,6 +9,16 @@ import ( "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" @@ -22,14 +32,6 @@ import ( "github.com/webitel/engine/store" "github.com/webitel/engine/store/sqlstore" "github.com/webitel/engine/wlogslog" - "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" - "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.26.0" - "go.uber.org/atomic" // -------------------- plugin(s) -------------------- // _ "github.com/webitel/webitel-go-kit/otel/sdk/log/otlp" @@ -74,7 +76,6 @@ type App struct { } func New(options ...string) (outApp *App, outErr error) { - config, err := loadConfig() if err != nil { return nil, err @@ -133,6 +134,36 @@ func New(options ...string) (outApp *App, outErr error) { wlog.RedirectStdLog(app.Log) wlog.InitGlobalLogger(app.Log) + // Health starts here, before anything slow. systemd counts + // TimeoutStartSec from ExecStart, so WithStartTimeout has to be measured + // from about the same moment — start the notifier after the managers and + // its fallback READY=1 can land after systemd has already given up. The + // registry reports not-ready until checks are registered further down, + // which is what a booting node should say. + healthLog := slog.New(wlogslog.NewHandler(app.Log)) + app.health = health.New(health.DefaultConfig(), healthLog) + + // app.ctx: Start's context is the scheduler's lifetime, so a short-lived + // one would silently stop every check. + if err := app.health.Start(app.ctx); err != nil { + return nil, fmt.Errorf("unable to start health registry: %w", err) + } + + // 60s leaves 30s of margin under the unit's TimeoutStartSec=90. + // nil when NOTIFY_SOCKET is unset; Start and Stop are both nil-safe. + app.sdNotify = sdnotify.New(app.health, + sdnotify.WithLogger(healthLog), + sdnotify.WithStartTimeout(60*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 } @@ -184,16 +215,6 @@ func New(options ...string) (outApp *App, outErr error) { app.GrpcServer = NewGrpcServer(app, app.Config().ServerSettings) - // Must exist before cluster.Start, which hands the verdict to Consul. - // Checks are registered at the end of New, once every manager exists. - healthLog := slog.New(wlogslog.NewHandler(app.Log)) - app.health = health.New(health.DefaultConfig(), healthLog) - - // 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 outErr = app.cluster.Start(); outErr != nil { return nil, outErr } @@ -244,27 +265,13 @@ func New(options ...string) (outApp *App, outErr error) { app.health.Informational("postgres", func(ctx context.Context) error { return sqlSupplier.GetMaster().Db.PingContext(ctx) }) + if p, ok := app.MessageQueue.(interface { Ping(context.Context) error }); ok { app.health.Informational("rabbitmq", p.Ping) } - // app.ctx: Start's context is the scheduler's lifetime, so a short-lived - // one would silently 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(90*time.Second), - ) - if err := app.sdNotify.Start(app.ctx); err != nil { - return nil, fmt.Errorf("unable to start sd_notify: %w", err) - } - return app, outErr } @@ -329,7 +336,7 @@ func (app *App) CallManager() call_manager.CallManager { // Ready reports whether this node can take traffic, per the health registry. func (app *App) Ready() (bool, model.AppError) { if app.health == nil { - return false, model.NewInternalError("app.ready.no_registry", "health registry is not initialised") + return false, model.NewInternalError("app.ready.no_registry", "health registry is not initialized") } ok, err := app.health.ReadyFunc()() @@ -395,11 +402,11 @@ func formatDomainEventKey(event *DomainEvent) (string, error) { if event.Object == "" { return "", errors.New("object required") } + 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 { diff --git a/app/health.go b/app/health.go index 04919839..bc81692d 100644 --- a/app/health.go +++ b/app/health.go @@ -5,8 +5,9 @@ import ( "errors" "fmt" - "github.com/webitel/engine/call_manager" "github.com/webitel/webitel-go-kit/infra/health" + + "github.com/webitel/engine/call_manager" ) // freeswitchCheck reports whether this node's FreeSWITCH is usable. diff --git a/app/health_test.go b/app/health_test.go index af654d5f..b8d49276 100644 --- a/app/health_test.go +++ b/app/health_test.go @@ -10,6 +10,7 @@ import ( type fakeCallClient struct { call_manager.CallClient + ready bool } @@ -17,6 +18,7 @@ func (f *fakeCallClient) Ready() bool { return f.ready } type fakeCallManager struct { call_manager.CallManager + cli call_manager.CallClient err model.AppError } @@ -61,6 +63,7 @@ func TestFreeswitchCheck(t *testing.T) { // TTL updater treats a non-nil error as safe to call .Error() on. func TestFreeswitchCheckNoTypedNilError(t *testing.T) { var nilAppErr model.AppError // nil interface value of a concrete-free type + cm := &fakeCallManager{cli: &fakeCallClient{ready: true}, err: nilAppErr} if err := freeswitchCheck(cm)(context.Background()); err != nil { diff --git a/wlogslog/wlogslog.go b/wlogslog/wlogslog.go index 395cbddf..0a533844 100644 --- a/wlogslog/wlogslog.go +++ b/wlogslog/wlogslog.go @@ -12,6 +12,11 @@ import ( type handler struct { log *wlog.Logger fields []wlog.Field + // prefix is the accumulated group path, "" or "a.b." with a trailing dot. + // Groups are flattened into dotted keys rather than wlog.Namespace: a zap + // namespace stays open, so a later sibling group would nest inside the + // earlier one instead of beside it. + prefix string } // NewHandler returns a slog.Handler that writes through to log. @@ -27,8 +32,10 @@ func (h *handler) Enabled(context.Context, slog.Level) bool { 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 = append(fields, wlog.Any(a.Key, a.Value.Any())) + fields = appendAttr(fields, a, h.prefix) + return true }) @@ -54,21 +61,55 @@ func (h *handler) WithAttrs(attrs []slog.Attr) slog.Handler { fields := make([]wlog.Field, len(h.fields), len(h.fields)+len(attrs)) copy(fields, h.fields) + for _, a := range attrs { - fields = append(fields, wlog.Any(a.Key, a.Value.Any())) + fields = appendAttr(fields, a, h.prefix) } - return &handler{log: h.log, fields: fields} + return &handler{log: h.log, fields: fields, prefix: h.prefix} } -// wlog.Namespace is zap.Namespace, which already has slog's group semantics. func (h *handler) WithGroup(name string) slog.Handler { if name == "" { return h } - fields := make([]wlog.Field, len(h.fields), len(h.fields)+1) + fields := make([]wlog.Field, len(h.fields)) copy(fields, h.fields) - return &handler{log: h.log, fields: append(fields, wlog.Namespace(name))} + return &handler{log: h.log, fields: fields, prefix: h.prefix + name + "."} +} + +// appendAttr converts one slog.Attr, following the slog contract: resolve +// LogValuer, drop empty attrs, drop empty groups, and inline a group with an +// empty key rather than qualifying its children. +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 + } + + if a.Key == "" { + 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 index 0a50f832..336ab4d1 100644 --- a/wlogslog/wlogslog_test.go +++ b/wlogslog/wlogslog_test.go @@ -30,6 +30,7 @@ func newTestLogger(t *testing.T) (*wlog.Logger, func() string) { if err != nil { t.Fatalf("read log: %v", err) } + return string(b) } } @@ -55,6 +56,7 @@ func TestWlogHandlerWritesThroughToWlog(t *testing.T) { 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) } @@ -80,13 +82,16 @@ func TestWlogHandlerLevelMapping(t *testing.T) { 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) } @@ -106,9 +111,11 @@ func TestWlogHandlerWithAttrsDoesNotLeak(t *testing.T) { 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) } @@ -118,6 +125,7 @@ func TestWlogHandlerWithAttrsDoesNotLeak(t *testing.T) { 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) } @@ -130,10 +138,98 @@ func TestWlogHandlerWithGroupAndEmptyCases(t *testing.T) { 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`) } - if got := len(base.WithGroup("sub").(*handler).fields); got != 1 { - t.Errorf("WithGroup: %d fields, want 1", got) + + // 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) + } +} + +// 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) } } From 3c691ad04939bc6459a388a645e4d5d495861165 Mon Sep 17 00:00:00 2001 From: vitaliikucherov <312534721+vitaliikucherov@users.noreply.github.com> Date: Fri, 14 Aug 2026 08:55:31 +0300 Subject: [PATCH 3/5] WTEL-10090: Move health checks onto their owning types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each check is now a method on the type that holds the connection: callManager.Ready, SqlSupplier.Ping, mq.MQ.Ping. Drops app/health.go and the anonymous interface cast — which was already hiding LayeredMQ. callManager.Ready reads the pool with All, not CallClient: CallClient advances the round-robin marker, so probing skewed real call routing. Start and stop timeouts move to HEALTH_START_TIMEOUT / HEALTH_STOP_TIMEOUT. Also keeps empty-key slog attrs, and trims comments. --- app/app.go | 41 +++------- app/grpc_server.go | 3 +- app/health.go | 27 ------- app/health_test.go | 72 ------------------ call_manager/call_manager.go | 25 +++++++ call_manager/health_test.go | 104 ++++++++++++++++++++++++++ deploy/systemd/webitel-engine.service | 8 +- model/config.go | 8 ++ mq/layered_mq.go | 4 + mq/mq.go | 2 + mq/rabbit/client.go | 7 +- store/sqlstore/supplier.go | 5 ++ wlogslog/wlogslog.go | 18 ++--- wlogslog/wlogslog_test.go | 19 +++-- 14 files changed, 181 insertions(+), 162 deletions(-) delete mode 100644 app/health.go delete mode 100644 app/health_test.go create mode 100644 call_manager/health_test.go diff --git a/app/app.go b/app/app.go index 29375690..7ffc8434 100644 --- a/app/app.go +++ b/app/app.go @@ -134,26 +134,19 @@ func New(options ...string) (outApp *App, outErr error) { wlog.RedirectStdLog(app.Log) wlog.InitGlobalLogger(app.Log) - // Health starts here, before anything slow. systemd counts - // TimeoutStartSec from ExecStart, so WithStartTimeout has to be measured - // from about the same moment — start the notifier after the managers and - // its fallback READY=1 can land after systemd has already given up. The - // registry reports not-ready until checks are registered further down, - // which is what a booting node should say. + // Before anything slow: systemd counts TimeoutStartSec from ExecStart. healthLog := slog.New(wlogslog.NewHandler(app.Log)) app.health = health.New(health.DefaultConfig(), healthLog) - // app.ctx: Start's context is the scheduler's lifetime, so a short-lived - // one would silently stop every check. + // 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) } - // 60s leaves 30s of margin under the unit's TimeoutStartSec=90. // nil when NOTIFY_SOCKET is unset; Start and Stop are both nil-safe. app.sdNotify = sdnotify.New(app.health, sdnotify.WithLogger(healthLog), - sdnotify.WithStartTimeout(60*time.Second), + 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) @@ -202,7 +195,7 @@ func New(options ...string) (outApp *App, outErr error) { } } - // Concrete handle: store.Store does not expose GetMaster. + // Concrete handle: store.Store does not expose Ping. sqlSupplier := sqlstore.NewSqlSupplier(app.Config().SqlSettings) app.Store = store.NewLayeredStore(sqlSupplier) @@ -257,20 +250,11 @@ func New(options ...string) (outApp *App, outErr error) { } } - // Critical is for node-local faults only: a shared dependency marked - // critical would take the whole fleet out of rotation at once. Consul is - // deliberately unchecked — the verdict travels through it. + // 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", freeswitchCheck(app.callManager)) - app.health.Informational("postgres", func(ctx context.Context) error { - return sqlSupplier.GetMaster().Db.PingContext(ctx) - }) - - if p, ok := app.MessageQueue.(interface { - Ping(context.Context) error - }); ok { - app.health.Informational("rabbitmq", p.Ping) - } + app.health.Critical("freeswitch", app.callManager.Ready) + app.health.Informational("postgres", sqlSupplier.Ping) + app.health.Informational("rabbitmq", app.MessageQueue.Ping) return app, outErr } @@ -278,13 +262,10 @@ func New(options ...string) (outApp *App, outErr error) { func (app *App) Shutdown() { wlog.Info("stopping Server...") - // Drain before anything is torn down, so the node stops advertising - // readiness while its dependencies are still up. The DrainHold wait happens - // inside Stop: 12s clears the 10s hold and fits TimeoutStopSec=30. Stop also - // halts the scheduler before MessageQueue.Close, so the rabbitmq check - // cannot race a closing connection. + // First: stop advertising readiness before anything is torn down. if app.health != nil { - ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second) + 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())) } diff --git a/app/grpc_server.go b/app/grpc_server.go index 6303d84e..01a50549 100644 --- a/app/grpc_server.go +++ b/app/grpc_server.go @@ -41,8 +41,7 @@ type GrpcServer struct { lis net.Listener } -// Listener is the socket actually bound, unlike GetPublicInterface which -// reports the address advertised to Consul. +// Listener is the socket actually bound, not the address advertised to Consul. func (grpc *GrpcServer) Listener() net.Listener { return grpc.lis } diff --git a/app/health.go b/app/health.go deleted file mode 100644 index bc81692d..00000000 --- a/app/health.go +++ /dev/null @@ -1,27 +0,0 @@ -package app - -import ( - "context" - "errors" - "fmt" - - "github.com/webitel/webitel-go-kit/infra/health" - - "github.com/webitel/engine/call_manager" -) - -// freeswitchCheck reports whether this node's FreeSWITCH is usable. -func freeswitchCheck(cm call_manager.CallManager) health.Check { - return func(context.Context) error { - cli, appErr := cm.CallClient() - if appErr != nil { - return fmt.Errorf("freeswitch client unavailable: %w", appErr) - } - - if !cli.Ready() { - return errors.New("freeswitch client not ready") - } - - return nil - } -} diff --git a/app/health_test.go b/app/health_test.go deleted file mode 100644 index b8d49276..00000000 --- a/app/health_test.go +++ /dev/null @@ -1,72 +0,0 @@ -package app - -import ( - "context" - "testing" - - "github.com/webitel/engine/call_manager" - "github.com/webitel/engine/model" -) - -type fakeCallClient struct { - call_manager.CallClient - - ready bool -} - -func (f *fakeCallClient) Ready() bool { return f.ready } - -type fakeCallManager struct { - call_manager.CallManager - - cli call_manager.CallClient - err model.AppError -} - -func (f *fakeCallManager) CallClient() (call_manager.CallClient, model.AppError) { - return f.cli, f.err -} - -func TestFreeswitchCheck(t *testing.T) { - tests := []struct { - name string - cm call_manager.CallManager - wantErr bool - }{ - { - name: "client ready", - cm: &fakeCallManager{cli: &fakeCallClient{ready: true}}, - }, - { - name: "client not ready", - cm: &fakeCallManager{cli: &fakeCallClient{ready: false}}, - wantErr: true, - }, - { - name: "no client available", - cm: &fakeCallManager{err: model.NewInternalError("test.no_client", "none")}, - wantErr: true, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - err := freeswitchCheck(tt.cm)(context.Background()) - if (err != nil) != tt.wantErr { - t.Errorf("err = %v, wantErr = %v", err, tt.wantErr) - } - }) - } -} - -// A health check must never hand back a nil-valued error interface: the Consul -// TTL updater treats a non-nil error as safe to call .Error() on. -func TestFreeswitchCheckNoTypedNilError(t *testing.T) { - var nilAppErr model.AppError // nil interface value of a concrete-free type - - cm := &fakeCallManager{cli: &fakeCallClient{ready: true}, err: nilAppErr} - - if err := freeswitchCheck(cm)(context.Background()); err != nil { - t.Errorf("a nil AppError must not surface as a non-nil error, got %v", err) - } -} 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 89dcaeeb..737e7724 100644 --- a/deploy/systemd/webitel-engine.service +++ b/deploy/systemd/webitel-engine.service @@ -5,10 +5,6 @@ StartLimitIntervalSec=60 StartLimitBurst=3 [Service] -# Type=notify: the health registry reports READY=1 once every critical check is -# green. WatchdogSec is deliberately absent — it stays off until it can be -# turned on together with the restart policy, or a wedged node either loops on -# restarts or parks in failed. Type=notify User=webitel Group=webitel @@ -24,9 +20,7 @@ Restart=on-failure RestartSec=5 KillMode=mixed KillSignal=SIGTERM -# Finite, because under Type=notify a node whose critical check never goes -# green would otherwise sit in activating forever. sdnotify.WithStartTimeout -# matches this and sends READY=1 with STATUS=starting degraded instead. +# Must exceed HEALTH_START_TIMEOUT, or the fallback READY=1 lands too late. TimeoutStartSec=90 TimeoutStopSec=30 LimitNOFILE=64000 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 45821e5b..4d8cb2d2 100644 --- a/mq/rabbit/client.go +++ b/mq/rabbit/client.go @@ -33,8 +33,7 @@ const ( callServiceHangupData = `{"hangup_by":"service","cause":"SYSTEM_SHUTDOWN","sip":501}` ) -// Stdlib errors: this file's `errors` is github.com/pkg/errors, whose New -// attaches a stack trace — too noisy for a check that runs every few seconds. +// 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") @@ -96,9 +95,7 @@ func (a *AMQP) Start() { go a.Listen() } -// Ping reports whether the broker connection is usable, read off the cached -// connection rather than dialing. Reached by an anonymous interface assertion -// so mq.MQ need not grow a method. +// 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() 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 index 0a533844..66af21b9 100644 --- a/wlogslog/wlogslog.go +++ b/wlogslog/wlogslog.go @@ -1,5 +1,4 @@ -// Package wlogslog bridges log/slog to wlog, for libraries that take a -// *slog.Logger. It has no engine dependencies. +// Package wlogslog bridges log/slog to wlog, for libraries that take a *slog.Logger. package wlogslog import ( @@ -12,10 +11,8 @@ import ( type handler struct { log *wlog.Logger fields []wlog.Field - // prefix is the accumulated group path, "" or "a.b." with a trailing dot. - // Groups are flattened into dotted keys rather than wlog.Namespace: a zap - // namespace stays open, so a later sibling group would nest inside the - // earlier one instead of beside it. + // 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 } @@ -80,9 +77,8 @@ func (h *handler) WithGroup(name string) slog.Handler { return &handler{log: h.log, fields: fields, prefix: h.prefix + name + "."} } -// appendAttr converts one slog.Attr, following the slog contract: resolve -// LogValuer, drop empty attrs, drop empty groups, and inline a group with an -// empty key rather than qualifying its children. +// 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() @@ -107,9 +103,5 @@ func appendAttr(fields []wlog.Field, a slog.Attr, prefix string) []wlog.Field { return fields } - if a.Key == "" { - 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 index 336ab4d1..bba67b20 100644 --- a/wlogslog/wlogslog_test.go +++ b/wlogslog/wlogslog_test.go @@ -12,8 +12,7 @@ import ( "github.com/webitel/wlog" ) -// newTestLogger returns a wlog logger writing JSON to a temp file, plus a -// reader for whatever it wrote. +// 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() @@ -36,8 +35,7 @@ func newTestLogger(t *testing.T) (*wlog.Logger, func() string) { } func TestWlogHandlerEnabledAlwaysTrue(t *testing.T) { - // wlog exposes no level query, so the handler must not filter — otherwise - // it would silently drop records wlog would have kept. + // wlog exposes no level query, so the handler must not filter. log, _ := newTestLogger(t) h := NewHandler(log) @@ -99,8 +97,7 @@ func TestWlogHandlerLevelMapping(t *testing.T) { } } -// slog's contract says a handler must be safe to share, so WithAttrs must copy. -// Mutating in place would leak attributes between unrelated loggers. +// Handlers must be safe to share, so WithAttrs must copy rather than mutate. func TestWlogHandlerWithAttrsDoesNotLeak(t *testing.T) { log, _ := newTestLogger(t) base := NewHandler(log) @@ -211,6 +208,16 @@ func TestAttrNamedGroupIsQualified(t *testing.T) { } } +// 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) From 34dcee6298e812e28005bcb6ba8f0c9d8a388709 Mon Sep 17 00:00:00 2001 From: vitaliikucherov <312534721+vitaliikucherov@users.noreply.github.com> Date: Fri, 14 Aug 2026 08:55:40 +0300 Subject: [PATCH 4/5] WTEL-10090: Re-register with Consul after a critical deregister Consul drops a service once its TTL check has been critical for DeregisterCriticalServiceAfter, then answers 404 to further updates. handlePassTTLError only re-registered on 500, so a 404 fell through silently: the node stayed healthy but invisible to discovery forever. Unreachable before this branch, since readiness was hardcoded true. Found by running a node with its critical dependency down for 60s. --- pkg/discovery/consul.go | 36 +++++++++++++++++++++++------------ pkg/discovery/consul_test.go | 37 ++++++++++++++++++++++++++++-------- 2 files changed, 53 insertions(+), 20 deletions(-) diff --git a/pkg/discovery/consul.go b/pkg/discovery/consul.go index 9204cc35..f8757737 100644 --- a/pkg/discovery/consul.go +++ b/pkg/discovery/consul.go @@ -124,8 +124,7 @@ func (c *consul) register(as *api.AgentServiceRegistration) error { } // 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 — -// this branch previously called err.Error() unconditionally. +// (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..." @@ -154,18 +153,31 @@ func (c *consul) update(as *api.AgentServiceRegistration) { } } +// 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.register(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 index 68177b63..4d83ab21 100644 --- a/pkg/discovery/consul_test.go +++ b/pkg/discovery/consul_test.go @@ -2,17 +2,14 @@ package discovery import ( "errors" + "fmt" + "net/http" "testing" + + "github.com/hashicorp/consul/api" ) -// TestTTLVerdict covers the three shapes a health verdict can take on its way -// to Consul's TTL check. -// -// The case that matters is (false, nil). The health package promises a false -// verdict always carries a non-nil error, but engine is what panics if that -// promise is ever broken — update() used to call err.Error() unconditionally on -// the not-ok branch. This test fails against that code, and it is hermetic: no -// Consul agent, no network. +// (false, nil) is the case that matters: update() used to panic on that branch. func TestTTLVerdict(t *testing.T) { tests := []struct { name string @@ -65,3 +62,27 @@ func TestTTLVerdict(t *testing.T) { }) } } + +// 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) + } + }) + } +} From 806988ec041a13f36e76d66ecafe6bd9f368184d Mon Sep 17 00:00:00 2001 From: vitaliikucherov <312534721+vitaliikucherov@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:15:11 +0300 Subject: [PATCH 5/5] WTEL-10090: Keep one Consul TTL updater across reconnects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recovery ran inside the updater goroutine and called register, which starts another one. Every reconnect left an extra updater behind, each duplicating TTL traffic and racing on checkId — and since each one 404s independently, they multiply rather than add. Split registration from updater startup: putRegistration refreshes the service and check id only, and does not update the TTL, so it cannot recurse back through handlePassTTLError. --- pkg/discovery/consul.go | 26 ++++++++++++++----- pkg/discovery/consul_test.go | 49 ++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 6 deletions(-) diff --git a/pkg/discovery/consul.go b/pkg/discovery/consul.go index f8757737..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,12 +131,8 @@ 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)) - - go c.updateTTL(c.ttl/2, as) + c.checkId = serviceCheck.CheckID return nil } @@ -176,7 +190,7 @@ func (c *consul) handlePassTTLError(err error, as *api.AgentServiceRegistration) wlog.Info(fmt.Sprintf("reconnect consul service id: %s", c.id)) - if e := c.register(as); e != nil { + 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 index 4d83ab21..ae4f2e6d 100644 --- a/pkg/discovery/consul_test.go +++ b/pkg/discovery/consul_test.go @@ -4,6 +4,8 @@ import ( "errors" "fmt" "net/http" + "os" + "strings" "testing" "github.com/hashicorp/consul/api" @@ -86,3 +88,50 @@ func TestShouldReregister(t *testing.T) { }) } } + +// 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] +}