fix: address project review findings
ci / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 48s
ci / api (push) Successful in 51s
ci / recommender (push) Canceled after 2s
ci / ts (packages + web) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (apps/web/Dockerfile, web, ) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (services/api/Dockerfile, api, api) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (services/api/Dockerfile, storage-agent, storage-agent) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (services/recommender/Dockerfile, recommender, ) (push) Canceled after 0s
ci / mobile (Metro-бандл Android) (push) Canceled after 0s
ci / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 48s
ci / api (push) Successful in 51s
ci / recommender (push) Canceled after 2s
ci / ts (packages + web) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (apps/web/Dockerfile, web, ) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (services/api/Dockerfile, api, api) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (services/api/Dockerfile, storage-agent, storage-agent) (push) Canceled after 0s
ci / docker (образы собираются — их сервер собирает сам при обновлении) (services/recommender/Dockerfile, recommender, ) (push) Canceled after 0s
ci / mobile (Metro-бандл Android) (push) Canceled after 0s
Security and robustness: - storage-agent rejects an empty INTERNAL_TOKEN (config notEmpty + middleware) - api takes the client IP only from X-Real-IP (chi RealIP trusted True-Client-IP) - /readyz reports ok/fail and logs the cause; checks run in parallel with a timeout - storage-agent shuts down gracefully; shared httpx.NewServer/Serve with IdleTimeout - goose migrations take a Postgres advisory lock - recommender readyz: per-check timeout, redis socket timeouts Clients: - Android release allows cleartext HTTP for LAN servers (expo-build-properties) - checkServer times out after 10s; connect button is disabled while checking Deploy and CI: - install.sh hands cache/uploads/updates/models to the non-root service users - update.sh refuses downgrades unless PIPU_ALLOW_DOWNGRADE=1 - recommender/analyzer get only the env they need; required secrets fail fast - nginx security headers and CSP; strip True-Client-IP - dev compose binds ports to 127.0.0.1; postgres/redis images pinned - version.mjs fails on a malformed explicit tag instead of releasing 0.0.0-dev - contract check catches untracked generated files - final releases fail without a release keystore; changelog skips rc tags Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1324cad2af
commit
4fa1add80d
37 files changed
+503
-199
No files matched your search
@@ -3,13 +3,10 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
@@ -63,37 +60,12 @@ func run() error {
|
||||
Version: buildinfo.Version,
|
||||
SupportedLocales: cfg.SupportedLocales,
|
||||
MinAndroidVersionCode: cfg.MinAndroidVersionCode,
|
||||
}, map[string]system.Checker{
|
||||
}, log, map[string]system.Checker{
|
||||
"postgres": system.CheckerFunc(pool.Ping),
|
||||
"redis": system.CheckerFunc(func(ctx context.Context) error { return rdb.Ping(ctx).Err() }),
|
||||
})
|
||||
|
||||
srv := &http.Server{
|
||||
Addr: cfg.HTTPAddr,
|
||||
Handler: server.NewRouter(server.Deps{Log: log, System: sys}),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
}
|
||||
return serve(ctx, log, srv, cfg.ShutdownTimeout)
|
||||
}
|
||||
|
||||
func serve(ctx context.Context, log *slog.Logger, srv *http.Server, shutdownTimeout time.Duration) error {
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
log.Info("listening", "addr", srv.Addr, "version", buildinfo.Version)
|
||||
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
errCh <- err
|
||||
}
|
||||
close(errCh)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
}
|
||||
|
||||
log.Info("shutting down")
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
|
||||
defer cancel()
|
||||
return srv.Shutdown(shutdownCtx)
|
||||
srv := httpx.NewServer(cfg.HTTPAddr, server.NewRouter(server.Deps{Log: log, System: sys}))
|
||||
log.Info("starting", "version", buildinfo.Version)
|
||||
return httpx.Serve(ctx, log, srv, cfg.ShutdownTimeout)
|
||||
}
|
||||
@@ -2,10 +2,11 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"pipu/api/internal/buildinfo"
|
||||
"pipu/api/internal/platform/config"
|
||||
@@ -14,21 +15,24 @@ import (
|
||||
)
|
||||
|
||||
func main() {
|
||||
cfg, err := config.Load[config.StorageAgent]()
|
||||
if err != nil {
|
||||
slog.Error("storage-agent: config", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
log := httpx.NewLogger(cfg.LogLevel)
|
||||
|
||||
srv := &http.Server{
|
||||
Addr: cfg.HTTPAddr,
|
||||
Handler: storageagent.NewRouter(cfg.InternalToken, cfg.LibraryRoot),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
}
|
||||
log.Info("listening", "addr", srv.Addr, "version", buildinfo.Version)
|
||||
if err := srv.ListenAndServe(); err != nil {
|
||||
log.Error("storage-agent stopped", "err", err)
|
||||
if err := run(); err != nil {
|
||||
slog.Error("storage-agent stopped", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func run() error {
|
||||
cfg, err := config.Load[config.StorageAgent]()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log := httpx.NewLogger(cfg.LogLevel)
|
||||
slog.SetDefault(log)
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
srv := httpx.NewServer(cfg.HTTPAddr, storageagent.NewRouter(log, cfg.InternalToken, cfg.LibraryRoot))
|
||||
log.Info("starting", "version", buildinfo.Version)
|
||||
return httpx.Serve(ctx, log, srv, cfg.ShutdownTimeout)
|
||||
}
|
||||
@@ -474,28 +474,28 @@ func (sh *strictHandler) GetReadyz(w http.ResponseWriter, r *http.Request) {
|
||||
// Base64 encoded, gzipped, json marshaled Swagger object
|
||||
var swaggerSpec = []string{
|
||||
|
||||
"H4sIAAAAAAAC/7xWX28bRRD/KquFh1hcbIdSQf1WVUD7gBQliJc2SjZ322Sb+8feXkSILMUJapFSGqkg",
|
||||
"xANq6Tdwgt0cdn35CrNfgU+CZu/OvnMuITyUJ/t2Z3fm95vfzM4+tQMvDHzuq4h29qnkURj4ETcfyzLY",
|
||||
"dLmHf+3AV9xX+JeFoStspkTgt8LM4qMnUeDjXmRvc4/hvw8lf0w79IPW7P5Wthu1inu73a5FHR7ZUoR4",
|
||||
"He1QeKV/hAROYQR9srDyxT1y55PbnzYoWubH8fb7nLlqe1UxFZvvUAYhl0pkcUfTde7HHu08pMEOXbOo",
|
||||
"2gs57dBISeFvmSsl/zYWkjtok5+a2QWbT7itaNcqE1H1ZAcOx985EL9D38CYwEQfw18ERpDCgEA6BZcQ",
|
||||
"GMBYnxCx9JlvEZhAHy70ASTwDob6gGywWG03hb/LXOGs25I73FeCudEGvYQDOVRMuDWBvIZUn+geTGAI",
|
||||
"E0hgWLiFMaTwJ6RwRhZwd7p+ASlyD+fQzz7G+jmcoyH09SEMYaxfNOpiCJlkniGFOY7AAJi7XCJLyZjX",
|
||||
"UDvLVb4lfMW3uMS9bAVBPWaxq2iHss0gVp1Nl/k79Ib5tLIk1aV1hTNn7yoR2dvc3rkGzzzXG8HOBoEE",
|
||||
"xpAQQ9RI9/RhJef1qask7A8Ywrk+Qtb1ITJOjCyQ/qE+yISD6XkLA0hRWefQhzNIdM9IJ0WfZUd1RJeK",
|
||||
"Av1vSeZw58b1YRXM1DG6yuUulw/8x8FlQlko1ne5jETWKvh3zAtdPL67VEeMJ/x15jsyEE5xbP3KckuM",
|
||||
"ut9B38g1K7r80L3A4eRudtFiXmMo/rdZSegTy5SnPoTUED3MZD+AgWH8LSQofRjqQ6J7ZsmkglozBEvt",
|
||||
"9p07Vo2AfebxKtZQhPGiF0fCrsMcxWEYSMWddTewmZvxNj38kMoY3fpIvVDcK5fN7JJ8gUnJ9vC7xPkc",
|
||||
"by8RCCpHn1SwYetd5d43XDbKMGm7udRs/2vdGdAzt1Yl8XUYr8n1ZY2hN5Hra77Z6SM4xfrTz3IJrHy+",
|
||||
"+jW5u/yALIswJl8h7U0Cv5kiSdEKEhgRrB6sGjxU9MF5OgIWikUMaYv7jUd+VoZj7KgwwcuwjwYh99EM",
|
||||
"I84CazQJvDGNdmxehFxx5O+DX4gT2FGLhaLpOc1HSIwSyrA8ixVDLzFZJKBr0dwV7dBbzXbzFsX2q7aN",
|
||||
"HvDO1u5SKzKluFhwtcVVnQL0MZzr40LhuodPwCn2IAJ9fWT6GOJL9A9IFfT1U0gg6RBkBk4JJEVpVfqU",
|
||||
"PtEvsHWdTQX2Yo5R65Gve8Yc37uM/rx96edVbvvopb4m4Z0+hiHRJwYENscF/ROM0AFMijMjfLL0s4L8",
|
||||
"RsY2diYzwzxwaId+yVWpdVnVMejjdvuaEei/jT4lL3XTzxvD62AqFEirvA2zJyN/DutdTWMvzVlYdp7H",
|
||||
"5N6NnEyHgTmJo0jZVmQeg71IcY+u4dWtbTONfV+S2SV27+cm75HaykxYR+5rFKh+avTWI5mK5rn51Uhr",
|
||||
"qsOL8hHoX0WAxEHiWvwrmcV7hF8eZurQv8QcXzEvYMIzzEdwgW0QZXa7fet/C+4VDHAELuYnnEh1z0zN",
|
||||
"+J6PIK0PfIgT0MSIeS78alZ/zt53OIPJLLcpnOoejPVR0U+y5fN8Eke7Y7JwQ8IatdLAKEzB4/o+jaVL",
|
||||
"O7RFu2tT4/18RigOdde6/wQAAP//nG+aXpMNAAA=",
|
||||
"H4sIAAAAAAAC/7xW3W4bRRR+ldHARSw2tkOpoL6rKqC9QIoSxE0bJZPdSTLN/jE7GxEiS3GCWqSURioI",
|
||||
"cYFa+gabYDdLHG9e4cwr8CTozO7aa2cTwkW5snf3zPn5vu+cOXvUDrww8LmvItrZo5JHYeBH3DwsymDd",
|
||||
"5R7+tQNfcV/hXxaGrrCZEoHfCnOLj55GgY/fInuLewz/fSj5Bu3QD1oT/638a9Qq/Xa7XYs6PLKlCNEd",
|
||||
"7VB4rX+EFE7gHBIyt/TFA3Lvk7ufNihaFsfR+0POXLW1rJiKzXMog5BLJfK8o/F77sce7TymwTZdsaja",
|
||||
"DTnt0EhJ4W8al5J/GwvJHbQpTk3sgvWn3Fa0a1WBmI5kBw7H35kifofElDGCkT6CvwicQwZ9Atm4uJRA",
|
||||
"H4b6mIiFz3yLwAgSuNT7kMIFDPQ+WWOx2moKf4e5wlm1JXe4rwRzozV6pQ7EUDHh1iTyBjJ9rHswggGM",
|
||||
"IIVBGRaGkMGfkMEpmcOv4/eXkCH2cAZJ/jDUL+AMDSHRBzCAoX7ZqMshZJJ5BhTmOAITYO5iBSwlY14D",
|
||||
"7YSr4pPwFd/kEr/lb7CoDRa7inYoWw9i1Vl3mb9Nb8mnlZNUR+sSZ87udSKyt7i9fUM9s1ivBdtrBFIY",
|
||||
"QkrWNphw18hcTql+bqSQkL/3fyH6oMD0HDICpyURA6J7SDycQqp7kDTqaZ4i9w8YwJk+RG/6ANkhJh5S",
|
||||
"hZ6MyJDKd9CHDFV4BknhH2WW6Z4+gJTeSEqlgTD+pmQOd27dS1aJYh36y1zucPnI3wiugs9CsbrDZSTy",
|
||||
"scK/Y17o4vGdhTpgPOGvMt+RgXDKY6vXtmZqOuECEkND3qDFoQeBw8n93NF80Y/Iz7u8ffSxZVoZOTRA",
|
||||
"D/IW6UPfIP4OUmwTGOiDCZ0DvU+tSQUL7fa9e1aN2H3m8elaQxHG814cCbuu5igOw0Aq7qy6gc3cHLfx",
|
||||
"4cdUxhjWR+iF4l61xSZOihdMSraLzxXMZ3B7hYWgcvTxVG04ppe59w2XjWqZtN1caLb/tUdN0ZOw1hTx",
|
||||
"dTXewPVVjWE0UehrdjDqQzjBXtXPCwksfb78Nbm/+IgsijAmXyHsTQK/mSbJ0ApSOCfYPdg1eKicmbNw",
|
||||
"BCwU85jSJvcbT/y8DYc4fWFk2v+UzAUh99EMM84TazQJvDWzYGhuj0JxZmg4gR21WCiantN8gsAooQzK",
|
||||
"k1wx9QqSJQFdixahaIfeababdyiOarVl9IA+WzsLrci04nyJ1SZXdQrQR3Cmj0qF6x5eFyc4gwgk+tBc",
|
||||
"Dlhfqn9AqCDRzyCFtEMQGTghkJatNTWn9LF+iaPrdCywlzOIWk983TPmeDfm8BfjS7+YxjbBKPU9CRf6",
|
||||
"CMfssSkCh+Oc/gnOMQCMyjPneL3p5yX4jRxtnExm33nk0A79kqvK6LKmV6aP2+0b1qX/tiZVotRtSm8N",
|
||||
"rv2xUCCbxm2QXxnF1Vkfapx7ZSfDtvM8JndvFWS8OMxIHEXKNiNzGexGint0BV23tszm9n1FZlfQfViY",
|
||||
"vEdop/bHOnDfoED1M6O3HslVNIvNr0ZaYx1eVo9Ach0AEpeOG+tfyi3eY/nVxaeu+lfI8TX7AhKe13wI",
|
||||
"lzgGUWZ323f+t+ReQ99sU8Wuhdur7pkNe7xW1SU+wA1oZMQ8k/40qz/n9zucwmjCbQYnugdDfVjOk/z1",
|
||||
"WbG1o90RmbslYI1aaWAWpuHx/R6NpUs7tEW7K2PjvWJHKA91V7r/BAAA//8arOezvw0AAA==",
|
||||
}
|
||||
|
||||
// GetSwagger returns the content of the embedded swagger specification file
|
||||
|
||||
@@ -23,10 +23,11 @@ type API struct {
|
||||
|
||||
// StorageAgent — конфигурация внутреннего сервиса storage-agent.
|
||||
type StorageAgent struct {
|
||||
HTTPAddr string `env:"HTTP_ADDR" envDefault:":8090"`
|
||||
InternalToken string `env:"INTERNAL_TOKEN,required"`
|
||||
LibraryRoot string `env:"LIBRARY_ROOT" envDefault:"/library"`
|
||||
LogLevel string `env:"LOG_LEVEL" envDefault:"info"`
|
||||
HTTPAddr string `env:"HTTP_ADDR" envDefault:":8090"`
|
||||
InternalToken string `env:"INTERNAL_TOKEN,required,notEmpty"`
|
||||
LibraryRoot string `env:"LIBRARY_ROOT" envDefault:"/library"`
|
||||
LogLevel string `env:"LOG_LEVEL" envDefault:"info"`
|
||||
ShutdownTimeout time.Duration `env:"SHUTDOWN_TIMEOUT" envDefault:"15s"`
|
||||
}
|
||||
|
||||
// Load заполняет структуру конфигурации T из окружения.
|
||||
|
||||
@@ -24,3 +24,10 @@ func TestLoadAPIRequiresDatabase(t *testing.T) {
|
||||
_, err := Load[API]()
|
||||
assert.ErrorContains(t, err, "DATABASE_URL")
|
||||
}
|
||||
|
||||
func TestLoadStorageAgentRejectsEmptyToken(t *testing.T) {
|
||||
t.Setenv("INTERNAL_TOKEN", "")
|
||||
|
||||
_, err := Load[StorageAgent]()
|
||||
assert.ErrorContains(t, err, "INTERNAL_TOKEN")
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/jackc/pgx/v5/stdlib"
|
||||
"github.com/pressly/goose/v3"
|
||||
"github.com/pressly/goose/v3/lock"
|
||||
|
||||
"pipu/api/migrations"
|
||||
)
|
||||
@@ -26,12 +27,17 @@ func Connect(ctx context.Context, url string) (*pgxpool.Pool, error) {
|
||||
return pool, nil
|
||||
}
|
||||
|
||||
// Migrate применяет встроенные миграции goose.
|
||||
// Migrate применяет встроенные миграции goose. Advisory-lock в PostgreSQL не даёт
|
||||
// нескольким экземплярам api применять миграции одновременно.
|
||||
func Migrate(ctx context.Context, pool *pgxpool.Pool) error {
|
||||
sqlDB := stdlib.OpenDBFromPool(pool)
|
||||
defer func(d *sql.DB) { _ = d.Close() }(sqlDB)
|
||||
|
||||
provider, err := goose.NewProvider(goose.DialectPostgres, sqlDB, migrations.FS)
|
||||
locker, err := lock.NewPostgresSessionLocker()
|
||||
if err != nil {
|
||||
return fmt.Errorf("db: migrations lock: %w", err)
|
||||
}
|
||||
provider, err := goose.NewProvider(goose.DialectPostgres, sqlDB, migrations.FS, goose.WithSessionLocker(locker))
|
||||
if err != nil {
|
||||
return fmt.Errorf("db: migrations: %w", err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
package httpx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// NewServer создаёт http.Server с общими таймаутами. WriteTimeout не задаётся:
|
||||
// стриминг аудио и загрузки держат ответ открытым дольше любого разумного предела.
|
||||
func NewServer(addr string, h http.Handler) *http.Server {
|
||||
return &http.Server{
|
||||
Addr: addr,
|
||||
Handler: h,
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
IdleTimeout: 120 * time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
// Serve запускает сервер и корректно останавливает его при отмене ctx.
|
||||
func Serve(ctx context.Context, log *slog.Logger, srv *http.Server, shutdownTimeout time.Duration) error {
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
log.Info("listening", "addr", srv.Addr)
|
||||
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
errCh <- err
|
||||
}
|
||||
close(errCh)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
}
|
||||
|
||||
log.Info("shutting down")
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
|
||||
defer cancel()
|
||||
return srv.Shutdown(shutdownCtx)
|
||||
}
|
||||
|
||||
// RealIP подставляет в RemoteAddr адрес клиента из X-Real-IP, который выставляет nginx.
|
||||
// В отличие от middleware.RealIP из chi, не доверяет True-Client-IP и X-Forwarded-For:
|
||||
// их клиент может прислать сам, а nginx пропускает их дальше без изменений.
|
||||
func RealIP(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if ip := strings.TrimSpace(r.Header.Get("X-Real-IP")); net.ParseIP(ip) != nil {
|
||||
r.RemoteAddr = ip
|
||||
}
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package httpx
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestRealIP(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
headers map[string]string
|
||||
want string
|
||||
}{
|
||||
{"no headers", nil, "192.0.2.1:1234"},
|
||||
{"x-real-ip", map[string]string{"X-Real-IP": "203.0.113.7"}, "203.0.113.7"},
|
||||
{"invalid x-real-ip", map[string]string{"X-Real-IP": "evil"}, "192.0.2.1:1234"},
|
||||
{"spoofed true-client-ip", map[string]string{"True-Client-IP": "1.1.1.1"}, "192.0.2.1:1234"},
|
||||
{"spoofed x-forwarded-for", map[string]string{"X-Forwarded-For": "1.1.1.1"}, "192.0.2.1:1234"},
|
||||
{
|
||||
"x-real-ip wins over spoofed headers",
|
||||
map[string]string{"X-Real-IP": "203.0.113.7", "True-Client-IP": "1.1.1.1"},
|
||||
"203.0.113.7",
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var got string
|
||||
h := RealIP(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { got = r.RemoteAddr }))
|
||||
req := httptest.NewRequest(http.MethodGet, "/", nil)
|
||||
req.RemoteAddr = "192.0.2.1:1234"
|
||||
for k, v := range tc.headers {
|
||||
req.Header.Set(k, v)
|
||||
}
|
||||
h.ServeHTTP(httptest.NewRecorder(), req)
|
||||
assert.Equal(t, tc.want, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -35,7 +35,7 @@ func NewRouter(d Deps) http.Handler {
|
||||
}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Use(middleware.RequestID, middleware.RealIP, httpx.AccessLog(log), middleware.Recoverer)
|
||||
r.Use(middleware.RequestID, httpx.RealIP, httpx.AccessLog(log), middleware.Recoverer)
|
||||
r.NotFound(httpx.NotFound)
|
||||
r.MethodNotAllowed(httpx.MethodNotAllowed)
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ package storageagent
|
||||
import (
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
@@ -21,8 +22,12 @@ type Mount struct {
|
||||
Path string `json:"path"`
|
||||
}
|
||||
|
||||
// NewRouter возвращает роутер агента. token — общий секрет INTERNAL_TOKEN.
|
||||
func NewRouter(token, libraryRoot string) http.Handler {
|
||||
// NewRouter возвращает роутер агента. token — общий секрет INTERNAL_TOKEN;
|
||||
// при пустом token все запросы к /internal отклоняются.
|
||||
func NewRouter(log *slog.Logger, token, libraryRoot string) http.Handler {
|
||||
if log == nil {
|
||||
log = slog.New(slog.DiscardHandler)
|
||||
}
|
||||
r := chi.NewRouter()
|
||||
r.NotFound(httpx.NotFound)
|
||||
r.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) {
|
||||
@@ -30,10 +35,11 @@ func NewRouter(token, libraryRoot string) http.Handler {
|
||||
})
|
||||
r.Route("/internal", func(r chi.Router) {
|
||||
r.Use(requireToken(token))
|
||||
r.Get("/mounts", func(w http.ResponseWriter, _ *http.Request) {
|
||||
r.Get("/mounts", func(w http.ResponseWriter, r *http.Request) {
|
||||
mounts, err := listMounts(libraryRoot)
|
||||
if err != nil {
|
||||
httpx.WriteProblem(w, http.StatusInternalServerError, "storage.list_failed", err.Error())
|
||||
log.ErrorContext(r.Context(), "list mounts", "err", err)
|
||||
httpx.WriteProblem(w, http.StatusInternalServerError, "storage.list_failed", "cannot list mounts")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"mounts": mounts})
|
||||
@@ -46,7 +52,7 @@ func requireToken(token string) func(http.Handler) http.Handler {
|
||||
return func(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
got, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ")
|
||||
if !ok || subtle.ConstantTimeCompare([]byte(got), []byte(token)) != 1 {
|
||||
if !ok || token == "" || subtle.ConstantTimeCompare([]byte(got), []byte(token)) != 1 {
|
||||
httpx.WriteProblem(w, http.StatusUnauthorized, "auth.invalid_internal_token", "invalid internal token")
|
||||
return
|
||||
}
|
||||
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
)
|
||||
|
||||
func TestInternalRequiresToken(t *testing.T) {
|
||||
h := NewRouter("secret", t.TempDir())
|
||||
h := NewRouter(nil, "secret", t.TempDir())
|
||||
|
||||
for _, auth := range []string{"", "Bearer wrong", "secret"} {
|
||||
rec := httptest.NewRecorder()
|
||||
@@ -26,6 +26,18 @@ func TestInternalRequiresToken(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestInternalRejectsEverythingWithEmptyToken(t *testing.T) {
|
||||
h := NewRouter(nil, "", t.TempDir())
|
||||
|
||||
for _, auth := range []string{"", "Bearer ", "Bearer x"} {
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/internal/mounts", nil)
|
||||
req.Header.Set("Authorization", auth)
|
||||
h.ServeHTTP(rec, req)
|
||||
assert.Equal(t, http.StatusUnauthorized, rec.Code, "auth=%q", auth)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListMounts(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
require.NoError(t, os.Mkdir(filepath.Join(root, "src-1"), 0o755))
|
||||
@@ -34,7 +46,7 @@ func TestListMounts(t *testing.T) {
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/internal/mounts", nil)
|
||||
req.Header.Set("Authorization", "Bearer secret")
|
||||
NewRouter("secret", root).ServeHTTP(rec, req)
|
||||
NewRouter(nil, "secret", root).ServeHTTP(rec, req)
|
||||
|
||||
require.Equal(t, http.StatusOK, rec.Code)
|
||||
var body struct{ Mounts []Mount }
|
||||
|
||||
@@ -3,6 +3,8 @@ package system
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"pipu/api/internal/api/gen"
|
||||
@@ -29,13 +31,17 @@ type Info struct {
|
||||
// Handler реализует системную часть gen.StrictServerInterface.
|
||||
type Handler struct {
|
||||
info Info
|
||||
log *slog.Logger
|
||||
checks map[string]Checker
|
||||
checkTimeout time.Duration
|
||||
}
|
||||
|
||||
// NewHandler создаёт обработчик. checks — именованные проверки для /readyz.
|
||||
func NewHandler(info Info, checks map[string]Checker) *Handler {
|
||||
return &Handler{info: info, checks: checks, checkTimeout: 2 * time.Second}
|
||||
func NewHandler(info Info, log *slog.Logger, checks map[string]Checker) *Handler {
|
||||
if log == nil {
|
||||
log = slog.New(slog.DiscardHandler)
|
||||
}
|
||||
return &Handler{info: info, log: log, checks: checks, checkTimeout: 2 * time.Second}
|
||||
}
|
||||
|
||||
// GetHealthz сообщает, что процесс жив; зависимости не проверяет.
|
||||
@@ -43,21 +49,35 @@ func (h *Handler) GetHealthz(context.Context, gen.GetHealthzRequestObject) (gen.
|
||||
return gen.GetHealthz200JSONResponse{Status: gen.HealthStatusStatusOk}, nil
|
||||
}
|
||||
|
||||
// GetReadyz проверяет все зависимости; 503, если хотя бы одна недоступна.
|
||||
// GetReadyz параллельно проверяет все зависимости; 503, если хотя бы одна недоступна.
|
||||
// /readyz доступен снаружи, поэтому текст ошибки пишется только в лог, а в ответ — "fail".
|
||||
func (h *Handler) GetReadyz(ctx context.Context, _ gen.GetReadyzRequestObject) (gen.GetReadyzResponseObject, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, h.checkTimeout)
|
||||
defer cancel()
|
||||
|
||||
results := make(map[string]string, len(h.checks))
|
||||
healthy := true
|
||||
var (
|
||||
mu sync.Mutex
|
||||
wg sync.WaitGroup
|
||||
results = make(map[string]string, len(h.checks))
|
||||
healthy = true
|
||||
)
|
||||
for name, c := range h.checks {
|
||||
if err := c.Check(ctx); err != nil {
|
||||
results[name] = err.Error()
|
||||
healthy = false
|
||||
continue
|
||||
}
|
||||
results[name] = "ok"
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
err := c.Check(ctx)
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if err != nil {
|
||||
h.log.WarnContext(ctx, "readiness check failed", "check", name, "err", err)
|
||||
results[name] = "fail"
|
||||
healthy = false
|
||||
return
|
||||
}
|
||||
results[name] = "ok"
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if !healthy {
|
||||
return gen.GetReadyz503JSONResponse{Status: gen.ReadyStatusStatusDegraded, Checks: results}, nil
|
||||
|
||||
@@ -20,7 +20,7 @@ func newServer(checks map[string]system.Checker) http.Handler {
|
||||
Version: "1.2.3",
|
||||
SupportedLocales: []string{"ru", "en"},
|
||||
MinAndroidVersionCode: 10203,
|
||||
}, checks)
|
||||
}, nil, checks)
|
||||
return server.NewRouter(server.Deps{System: h})
|
||||
}
|
||||
|
||||
@@ -68,7 +68,7 @@ func TestReadyz(t *testing.T) {
|
||||
|
||||
assert.Equal(t, http.StatusServiceUnavailable, rec.Code)
|
||||
assert.Equal(t, "degraded", body["status"])
|
||||
assert.Equal(t, map[string]any{"postgres": "ok", "redis": "connection refused"}, body["checks"])
|
||||
assert.Equal(t, map[string]any{"postgres": "ok", "redis": "fail"}, body["checks"])
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
"""FastAPI-приложение внутреннего API рекомендаций."""
|
||||
|
||||
import asyncio
|
||||
import hmac
|
||||
import logging
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Annotated
|
||||
|
||||
@@ -11,6 +13,10 @@ from pipu_rec.settings import Settings
|
||||
|
||||
Check = Callable[[], Awaitable[None]]
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
CHECK_TIMEOUT_S = 2.0
|
||||
|
||||
|
||||
def create_app(settings: Settings, checks: dict[str, Check] | None = None) -> FastAPI:
|
||||
app = FastAPI(title="pipu-recommender", version=settings.version, docs_url=None, redoc_url=None)
|
||||
@@ -26,15 +32,20 @@ def create_app(settings: Settings, checks: dict[str, Check] | None = None) -> Fa
|
||||
async def healthz() -> dict[str, str]:
|
||||
return {"status": "ok"}
|
||||
|
||||
async def run_check(name: str, check: Check) -> str:
|
||||
try:
|
||||
await asyncio.wait_for(check(), CHECK_TIMEOUT_S)
|
||||
except Exception as exc:
|
||||
log.warning("readiness check %s failed: %r", name, exc)
|
||||
return "fail"
|
||||
return "ok"
|
||||
|
||||
@app.get("/readyz")
|
||||
async def readyz() -> JSONResponse:
|
||||
results: dict[str, str] = {}
|
||||
for name, check in checks.items():
|
||||
try:
|
||||
await check()
|
||||
results[name] = "ok"
|
||||
except Exception as exc:
|
||||
results[name] = str(exc) or type(exc).__name__
|
||||
# Проверки параллельные, каждая ограничена таймаутом; причину ошибки пишем только в лог.
|
||||
names = list(checks)
|
||||
outcomes = await asyncio.gather(*(run_check(n, checks[n]) for n in names))
|
||||
results = dict(zip(names, outcomes, strict=True))
|
||||
healthy = all(v == "ok" for v in results.values())
|
||||
return JSONResponse(
|
||||
{"status": "ok" if healthy else "degraded", "checks": results},
|
||||
|
||||
@@ -15,7 +15,7 @@ log = logging.getLogger("pipu_rec")
|
||||
|
||||
|
||||
def _serve(settings: Settings) -> None:
|
||||
rdb = aioredis.Redis.from_url(settings.redis_url)
|
||||
rdb = aioredis.Redis.from_url(settings.redis_url, socket_connect_timeout=2, socket_timeout=2)
|
||||
|
||||
async def check_postgres() -> None:
|
||||
async with await psycopg.AsyncConnection.connect(settings.database_url, connect_timeout=2) as conn:
|
||||
|
||||
@@ -32,7 +32,21 @@ def test_readyz_all_up() -> None:
|
||||
def test_readyz_dependency_down() -> None:
|
||||
r = client({"postgres": ok, "redis": down}).get("/readyz")
|
||||
assert r.status_code == 503
|
||||
assert r.json()["checks"]["redis"] == "connection refused"
|
||||
assert r.json()["checks"]["redis"] == "fail"
|
||||
|
||||
|
||||
def test_readyz_dependency_hangs(monkeypatch) -> None: # type: ignore[no-untyped-def]
|
||||
import asyncio
|
||||
|
||||
from pipu_rec.api import app as app_module
|
||||
|
||||
async def hang() -> None:
|
||||
await asyncio.sleep(10)
|
||||
|
||||
monkeypatch.setattr(app_module, "CHECK_TIMEOUT_S", 0.05)
|
||||
r = client({"postgres": ok, "redis": hang}).get("/readyz")
|
||||
assert r.status_code == 503
|
||||
assert r.json()["checks"] == {"postgres": "ok", "redis": "fail"}
|
||||
|
||||
|
||||
def test_internal_status_requires_token() -> None:
|
||||
|
||||
Reference in new issue
Block a user