WTEL-10090: Integrate infra/health probes

.gitignore

@@ -66,7 +66,6 @@ go.work.sum
 # Build output
 build
 dist/
-/pkg/
 artifacts/
 work/
 out/

app/app.go

@@ -5,9 +5,20 @@
 	"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 @@
 	"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"
@@ -60,14 +67,15 @@
 	cc               cc.CCManager
 	cipher           presign.PreSign
 	audit            *logger.Audit
 	ctx              context.Context
 	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 @@
 	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 @@
 		}
 	}
 
-	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 @@
 		}
 	}
 
+	// 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 @@
 	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 @@
 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 @@
 	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 @@
 	if err != nil {
 		return err
 	}
-		
+
 	return a.MessageQueue.Send(ctx, EventExchangeName, routingKey, body)
 }

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
 	}

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 == "::" {

call_manager/call_manager.go

@@ -2,6 +2,7 @@ package call_manager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"sync"
@@ -16,13 +17,20 @@ 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()
 	MakeOutboundCall(req *model.CallRequest) (string, model.AppError)
 	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 {

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)
+	}
+}

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