Feature/healthcheck

Dockerfile.auth

@@ -21,13 +21,14 @@ RUN go build -o ./auth/auth ./auth/cmd/auth/main.go
 
 FROM alpine:3.22.1
 
-RUN apk add --no-cache ca-certificates
+RUN apk add --no-cache ca-certificates curl
 
 WORKDIR /app
 
 COPY --from=builder /app/auth/auth .
 COPY --from=builder /app/auth/migrations ./migrations
 
 EXPOSE 50051
+EXPOSE 3000
 
 CMD ["./auth"]

Dockerfile.gateway

@@ -21,13 +21,14 @@ RUN go build -o ./gateway/gateway ./gateway/cmd/gateway/main.go
 
 FROM alpine:3.22.1
 
-RUN apk add --no-cache ca-certificates
+RUN apk add --no-cache ca-certificates curl
 
 WORKDIR /app
 
 COPY --from=builder /app/gateway/gateway .
 COPY protos/gen/swagger ./swagger
 
 EXPOSE 8080
+EXPOSE 3000
 
 CMD ["./gateway"]

Dockerfile.profiles

@@ -21,13 +21,14 @@ RUN go build -o ./profiles/profiles ./profiles/cmd/profiles/main.go
 
 FROM alpine:3.22.1
 
-RUN apk add --no-cache ca-certificates
+RUN apk add --no-cache ca-certificates curl
 
 WORKDIR /app
 
 COPY --from=builder /app/profiles/profiles .
 COPY --from=builder /app/profiles/migrations ./migrations
 
 EXPOSE 50051
+EXPOSE 3000
 
 CMD ["./profiles"]

auth/cmd/auth/main.go

@@ -10,31 +10,41 @@ import (
 	"github.com/alexwatcher/gateofthings/auth/internal/app"
 	"github.com/alexwatcher/gateofthings/auth/internal/config"
 	"github.com/alexwatcher/gateofthings/auth/internal/consts"
+	"github.com/alexwatcher/gateofthings/shared/pkg/healthz"
 	sharedpgsql "github.com/alexwatcher/gateofthings/shared/pkg/migrator/postgresql"
 	"github.com/alexwatcher/gateofthings/shared/pkg/telemetry"
 )
 
 func main() {
-	ctx := context.Background()
+	ctx, stop := context.WithCancel(context.Background())
 	cfg := config.MustLoad()
 
 	res := telemetry.MustCreateResource(consts.ServiceName, consts.ServiceVersion, cfg.Env)
 	telemetry.MustInitLogger(context.Background(), res, cfg.Telemetry.LogsEndpoint)
 	telemetry.MustInitTracer(context.Background(), res, cfg.Telemetry.TraceEndpoint)
 	telemetry.MustInitMeter(context.Background(), res, cfg.Telemetry.MetricsEndpoint)
 
+	application := app.New(ctx, cfg.GRPC, cfg.Database, cfg.TokenSecret, cfg.TokenTTL)
+
+	hc := healthz.New(
+		healthz.WithPort(cfg.HealthPort),
+		healthz.WithLiveProbe(func(ctx context.Context) error { return nil }),
+		healthz.WithReadyProbe(func(ctx context.Context) error { return application.Ready() }),
+	)
+	go hc.MustRun(ctx)
+
 	slog.Info("start migration")
 	sharedpgsql.Migrate(cfg.Database)
 	slog.Info("end migration")
 
 	slog.Info("starting application")
-	application := app.New(ctx, cfg.GRPC, cfg.Database, cfg.TokenSecret, cfg.TokenTTL)
 	go application.MustRun(ctx)
 
-	stop := make(chan os.Signal, 1)
-	signal.Notify(stop, syscall.SIGTERM, syscall.SIGINT)
-	sig := <-stop
+	stopChan := make(chan os.Signal, 1)
+	signal.Notify(stopChan, syscall.SIGTERM, syscall.SIGINT)
+	sig := <-stopChan
 
+	stop()
 	slog.Info("stopping application", "signal", sig)
 	application.Stop(ctx)
 }

auth/internal/app/app.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
+	"sync/atomic"
 	"time"
 
 	grpcauth "github.com/alexwatcher/gateofthings/auth/internal/grpc/auth"
@@ -14,19 +15,21 @@ import (
 	"github.com/alexwatcher/gateofthings/shared/pkg/grpc/interceptors/tracing"
 	"github.com/alexwatcher/gateofthings/shared/pkg/grpc/interceptors/valid"
 	sharedpgsql "github.com/alexwatcher/gateofthings/shared/pkg/repository/postgresql"
+	"github.com/jackc/pgx/v5"
 	"google.golang.org/grpc"
 )
 
 type App struct {
 	gRPCServer *grpc.Server
 	gRPConfig  config.GRPCSrvConfig
+	dbConn     *pgx.Conn
+	isRunning  int32
 }
 
 // New initializes a new instance of the App struct with a gRPC server
 // listening on the specified port. It registers the authentication
 // service with the server and returns the configured App instance.
 func New(ctx context.Context, gRPConfig config.GRPCSrvConfig, dbConfig config.DatabaseConfig, tokenSecret string, tokenTTL time.Duration) *App {
-
 	dbConn, err := sharedpgsql.NewConnection(ctx, dbConfig)
 	if err != nil {
 		slog.Error("failed to connect to database", "error", err)
@@ -46,6 +49,7 @@ func New(ctx context.Context, gRPConfig config.GRPCSrvConfig, dbConfig config.Da
 	return &App{
 		gRPCServer: gRPCServer,
 		gRPConfig:  gRPConfig,
+		dbConn:     dbConn,
 	}
 }
 
@@ -63,6 +67,8 @@ func (a *App) Run(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	atomic.StoreInt32(&a.isRunning, 1)
+	defer atomic.StoreInt32(&a.isRunning, 0)
 	slog.Info("gRPC server started", "port", a.gRPConfig.Port)
 	if err := a.gRPCServer.Serve(lis); err != nil {
 		return fmt.Errorf("app.run: %w", err)
@@ -76,3 +82,17 @@ func (a *App) Stop(ctx context.Context) {
 	slog.Info("stopping gRPC server")
 	a.gRPCServer.GracefulStop()
 }
+
+// Readiness probe
+func (a *App) Ready() error {
+	err := a.dbConn.Ping(context.Background())
+	if err != nil {
+		slog.Warn("app: live probe failed", "error", err)
+		return err
+	}
+	if atomic.LoadInt32(&a.isRunning) == 0 {
+		slog.Warn("app: live probe failed: app is not running")
+		return fmt.Errorf("app: is not running")
+	}
+	return nil
+}

auth/internal/config/config.go

@@ -14,6 +14,7 @@ type Config struct {
 	Telemetry   scfg.TelemetryConfig `envPrefix:"TELEMETRY_"`
 	GRPC        scfg.GRPCSrvConfig   `envPrefix:"GRPC_"`
 	Database    scfg.DatabaseConfig  `envPrefix:"DB_"`
+	HealthPort  uint16               `env:"HEALTH_PORT,required"`
 }
 
 // MustLoad loads configuration from environment variables into a Config instance.

auth/tests/suite/suite.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"net/http"
 	"os/exec"
 	"strconv"
 	"strings"
@@ -26,7 +27,9 @@ var cfg = &config.Config{
 	TokenTTL:    time.Minute * 10,
 	TokenSecret: "test",
 	Telemetry:   scfg.TelemetryConfig{},
-	GRPC:        scfg.GRPCSrvConfig{},
+	GRPC: scfg.GRPCSrvConfig{
+		Port: 50051,
+	},
 	Database: scfg.DatabaseConfig{
 		Host:       "postgres",
 		Port:       5432,
@@ -36,6 +39,7 @@ var cfg = &config.Config{
 		Password:   "pass",
 		Migrations: "./migrations",
 	},
+	HealthPort: 3000,
 }
 
 type Suite struct {
@@ -80,15 +84,20 @@ func New(t *testing.T) (context.Context, *Suite, func()) {
 		}
 	}()
 
-	dbRes := mustSetupPostgres(testName, pool, cfg, networkName)
+	dbRes, err := setupPostgres(testName, pool, cfg, networkName)
+	if err != nil {
+		fmt.Printf("Failed setup PostgreSQL: %v", err)
+		panic(err)
+	}
 	resources = append(resources, dbRes)
 
 	cfg.Database.Host = strings.TrimPrefix(dbRes.Container.Name, "/")
-	authRes, port := mustSetupAuth(testName, pool, cfg, networkName)
+	authRes, port := mustSetupAuth(testName, pool, cfg, networkName, time.Second*20)
 	resources = append(resources, authRes)
 
 	cc, err := grpc.NewClient(fmt.Sprintf("localhost:%d", port), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
+		fmt.Printf("Failed create grpc client: %v", err)
 		panic(err)
 	}
 
@@ -111,7 +120,7 @@ func New(t *testing.T) (context.Context, *Suite, func()) {
 	}
 }
 
-func mustSetupPostgres(testName string, pool *dockertest.Pool, cfg *config.Config, network string) *dockertest.Resource {
+func setupPostgres(testName string, pool *dockertest.Pool, cfg *config.Config, network string) (*dockertest.Resource, error) {
 	res, err := pool.RunWithOptions(&dockertest.RunOptions{
 		Repository: "postgres",
 		Tag:        "17.5",
@@ -124,7 +133,7 @@ func mustSetupPostgres(testName string, pool *dockertest.Pool, cfg *config.Confi
 		},
 	})
 	if err != nil {
-		panic(err)
+		return nil, err
 	}
 
 	err = pool.Retry(func() error {
@@ -138,15 +147,15 @@ func mustSetupPostgres(testName string, pool *dockertest.Pool, cfg *config.Confi
 		return nil
 	})
 	if err != nil {
-		panic(err)
+		return nil, err
 	}
 
-	return res
+	return res, nil
 }
 
-func mustSetupAuth(testName string, pool *dockertest.Pool, cfg *config.Config, network string) (*dockertest.Resource, uint16) {
-	port := 3000
-	exposedPort := fmt.Sprintf("%d/tcp", port)
+func mustSetupAuth(testName string, pool *dockertest.Pool, cfg *config.Config, network string, setupTimeout time.Duration) (*dockertest.Resource, uint16) {
+	exposedPort := fmt.Sprintf("%d/tcp", cfg.GRPC.Port)
+	exposedHealthPort := fmt.Sprintf("%d/tcp", cfg.HealthPort)
 	res, err := pool.RunWithOptions(&dockertest.RunOptions{
 		Repository: "got-auth",
 		Tag:        "latest",
@@ -156,34 +165,67 @@ func mustSetupAuth(testName string, pool *dockertest.Pool, cfg *config.Config, n
 			fmt.Sprintf("ENV=%s", cfg.Env),
 			fmt.Sprintf("TOKEN_TTL=%v", cfg.TokenTTL),
 			fmt.Sprintf("TOKEN_SECRET=%s", cfg.TokenSecret),
-			fmt.Sprintf("GRPC_PORT=%d", port),
+			fmt.Sprintf("GRPC_PORT=%d", cfg.GRPC.Port),
 			fmt.Sprintf("DB_HOST=%s", cfg.Database.Host),
 			fmt.Sprintf("DB_PORT=%d", cfg.Database.Port),
 			fmt.Sprintf("DB_NAME=%s", cfg.Database.Name),
 			fmt.Sprintf("DB_USER=%s", cfg.Database.User),
 			fmt.Sprintf("DB_PASSWORD=%s", cfg.Database.Password),
 			fmt.Sprintf("DB_MIGRATIONS=%s", cfg.Database.Migrations),
+			fmt.Sprintf("HEALTH_PORT=%d", cfg.HealthPort),
 		},
-		ExposedPorts: []string{exposedPort},
+		ExposedPorts: []string{exposedPort, exposedHealthPort},
 	}, func(config *docker.HostConfig) {
 		config.PortBindings = map[docker.Port][]docker.PortBinding{
-			docker.Port(exposedPort): {{HostIP: "0.0.0.0", HostPort: ""}},
+			docker.Port(exposedPort):       {{HostIP: "0.0.0.0", HostPort: ""}},
+			docker.Port(exposedHealthPort): {{HostIP: "0.0.0.0", HostPort: ""}},
 		}
 	})
 	if err != nil {
 		panic(err)
 	}
 
-	hostPort, err := strconv.Atoi(res.GetPort(exposedPort))
-	if err != nil {
+	cleanup := func() {
 		if err := pool.Purge(res); err != nil {
 			log.Printf("Could not purge container: %s", err)
 		}
+	}
+
+	hostPort, err := strconv.Atoi(res.GetPort(exposedPort))
+	if err != nil {
+		cleanup()
+		panic(err)
+	}
+
+	healthHostPort, err := strconv.Atoi(res.GetPort(exposedHealthPort))
+	if err != nil {
+		cleanup()
 		panic(err)
 	}
 
-	// TODO: implement and then use healthcheck to wait availability of auth service
-	time.Sleep(time.Second * 5)
+	err = waitHealthCheck(fmt.Sprintf("http://localhost:%d/readyz", healthHostPort), setupTimeout)
+	if err != nil {
+		cleanup()
+		panic(err)
+	}
 
 	return res, uint16(hostPort)
 }
+
+func waitHealthCheck(address string, timeout time.Duration) error {
+	startTime := time.Now()
+	for {
+		resp, err := http.Get(address)
+		if err != nil {
+			continue
+		}
+		if time.Since(startTime) > timeout {
+			return fmt.Errorf("%s healthcheck timeout", address)
+		}
+		if resp.StatusCode == http.StatusOK {
+			break
+		}
+		time.Sleep(time.Millisecond * 500)
+	}
+	return nil
+}

docker-compose.yaml

@@ -105,8 +105,16 @@ services:
             DB_USER: auth_user
             DB_PASSWORD: auth_pass
             DB_MIGRATIONS: ./migrations
+            HEALTH_PORT: 3000
         ports:
             - 50051:50051
+            - 3001:3000
+        healthcheck:
+            test: ["CMD", "curl", "-f", "http://localhost:3000/readyz"]
+            interval: 3s
+            timeout: 3s
+            retries: 3
+            start_period: 1s
         depends_on:
             postgres:
                 condition: service_healthy
@@ -130,8 +138,16 @@ services:
             DB_USER: profiles_user
             DB_PASSWORD: profiles_pass
             DB_MIGRATIONS: ./migrations
+            HEALTH_PORT: 3000
         ports:
             - 50052:50051
+            - 3002:3000
+        healthcheck:
+            test: ["CMD", "curl", "-f", "http://localhost:3000/readyz"]
+            interval: 3s
+            timeout: 3s
+            retries: 3
+            start_period: 1s
         depends_on:
             postgres:
                 condition: service_healthy
@@ -151,11 +167,23 @@ services:
             TELEMETRY_TRACE: otel-collector:4317
             TELEMETRY_METRICS: otel-collector:4317
             TELEMETRY_LOGS: otel-collector:4317
+            HEALTH_PORT: 3000
         ports:
             - 8080:8080
+            - 3003:3000
+        healthcheck:
+            test: ["CMD", "curl", "-f", "http://localhost:3000/readyz"]
+            interval: 3s
+            timeout: 3s
+            retries: 3
+            start_period: 1s
         depends_on:
             postgres:
                 condition: service_healthy
+            auth:
+                condition: service_healthy
+            profiles:
+                condition: service_healthy
 
     migrator:
         image: got-migrator:latest