fix(server) add graceful shutdown timeout and response/connection limits
- commit
- b6d1cf39c37c2d2ea5794ab3095b9c41d2b2c2ab
- author
- bnema · · 2026-02-06 21:29 UTC
- parent
- 7b8d7b1
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +106 -56
M internal/adapters/in/http/proxy/handler.go +3 -1
| @@ -71,6 +71,8 @@ func (h *Handler) Start(ctx context.Context, handler http.Handler) error { | |||
| 71 | 71 | Str(zerowrap.FieldLayer, "adapter"). | |
| 72 | 72 | Str(zerowrap.FieldAdapter, "http"). | |
| 73 | 73 | Msg("proxy server shutting down") | |
| 74 | - | return server.Shutdown(context.Background()) | |
| 74 | + | shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) | |
| 75 | + | defer shutdownCancel() | |
| 76 | + | return server.Shutdown(shutdownCtx) | |
| 75 | 77 | } | |
| 76 | 78 | } | |
| @@ -71,6 +71,8 @@ func (h *Handler) Start(ctx context.Context, handler http.Handler) error { | |||||
| 71 | Str(zerowrap.FieldLayer, "adapter"). | 71 | Str(zerowrap.FieldLayer, "adapter"). | ||
| 72 | Str(zerowrap.FieldAdapter, "http"). | 72 | Str(zerowrap.FieldAdapter, "http"). | ||
| 73 | Msg("proxy server shutting down") | 73 | Msg("proxy server shutting down") | ||
| 74 | - | return server.Shutdown(context.Background()) | 74 | + | shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 75 | + | defer shutdownCancel() | |||
| 76 | + | return server.Shutdown(shutdownCtx) | |||
| 75 | } | 77 | } | ||
| 76 | } | 78 | } | ||
M internal/app/run.go +103 -55
| @@ -64,13 +64,15 @@ import ( | |||
| 64 | 64 | // Config holds the application configuration. | |
| 65 | 65 | type Config struct { | |
| 66 | 66 | Server struct { | |
| 67 | - | Port int `mapstructure:"port"` | |
| 68 | - | RegistryPort int `mapstructure:"registry_port"` | |
| 69 | - | GordonDomain string `mapstructure:"gordon_domain"` | |
| 70 | - | RegistryDomain string `mapstructure:"registry_domain"` // Deprecated: use gordon_domain | |
| 71 | - | DataDir string `mapstructure:"data_dir"` | |
| 72 | - | MaxProxyBodySize string `mapstructure:"max_proxy_body_size"` // e.g., "512MB", "1GB" | |
| 73 | - | MaxBlobChunkSize string `mapstructure:"max_blob_chunk_size"` // e.g., "512MB", "1GB" | |
| 67 | + | Port int `mapstructure:"port"` | |
| 68 | + | RegistryPort int `mapstructure:"registry_port"` | |
| 69 | + | GordonDomain string `mapstructure:"gordon_domain"` | |
| 70 | + | RegistryDomain string `mapstructure:"registry_domain"` // Deprecated: use gordon_domain | |
| 71 | + | DataDir string `mapstructure:"data_dir"` | |
| 72 | + | MaxProxyBodySize string `mapstructure:"max_proxy_body_size"` // e.g., "512MB", "1GB" | |
| 73 | + | MaxBlobChunkSize string `mapstructure:"max_blob_chunk_size"` // e.g., "512MB", "1GB" | |
| 74 | + | MaxProxyResponseSize string `mapstructure:"max_proxy_response_size"` // e.g., "1GB", "0" for no limit | |
| 75 | + | MaxConcurrentConns int `mapstructure:"max_concurrent_connections"` | |
| 74 | 76 | } `mapstructure:"server"` | |
| 75 | 77 | ||
| 76 | 78 | Logging struct { | |
| @@ -300,33 +302,12 @@ func createServices(ctx context.Context, v *viper.Viper, cfg Config, log zerowra | |||
| 300 | 302 | } | |
| 301 | 303 | svc.registrySvc = registrySvc.NewService(svc.blobStorage, svc.manifestStorage, svc.eventBus) | |
| 302 | 304 | ||
| 303 | - | // Parse max_proxy_body_size config (default: 512MB) | |
| 304 | - | maxProxyBodySize := int64(512 << 20) // 512MB default | |
| 305 | - | if cfg.Server.MaxProxyBodySize != "" { | |
| 306 | - | parsedSize, err := bytesize.Parse(cfg.Server.MaxProxyBodySize) | |
| 307 | - | if err != nil { | |
| 308 | - | return nil, log.WrapErrWithFields(err, "invalid server.max_proxy_body_size configuration", map[string]any{"value": cfg.Server.MaxProxyBodySize}) | |
| 309 | - | } | |
| 310 | - | maxProxyBodySize = parsedSize | |
| 311 | - | } | |
| 312 | - | ||
| 313 | - | // Parse max_blob_chunk_size config (default: 95MB) | |
| 314 | - | maxBlobChunkSize := int64(registry.DefaultMaxBlobChunkSize) | |
| 315 | - | if cfg.Server.MaxBlobChunkSize != "" { | |
| 316 | - | parsedSize, err := bytesize.Parse(cfg.Server.MaxBlobChunkSize) | |
| 317 | - | if err != nil { | |
| 318 | - | return nil, log.WrapErrWithFields(err, "invalid server.max_blob_chunk_size configuration", map[string]any{"value": cfg.Server.MaxBlobChunkSize}) | |
| 319 | - | } | |
| 320 | - | maxBlobChunkSize = parsedSize | |
| 305 | + | proxyCfg, err := buildProxyConfig(cfg, log) | |
| 306 | + | if err != nil { | |
| 307 | + | return nil, err | |
| 321 | 308 | } | |
| 322 | - | ||
| 323 | - | svc.maxBlobChunkSize = maxBlobChunkSize | |
| 324 | - | ||
| 325 | - | svc.proxySvc = proxy.NewService(svc.runtime, svc.containerSvc, svc.configSvc, proxy.Config{ | |
| 326 | - | RegistryDomain: cfg.Server.RegistryDomain, | |
| 327 | - | RegistryPort: cfg.Server.RegistryPort, | |
| 328 | - | MaxBodySize: maxProxyBodySize, | |
| 329 | - | }) | |
| 309 | + | svc.maxBlobChunkSize = proxyCfg.maxBlobChunkSize | |
| 310 | + | svc.proxySvc = proxy.NewService(svc.runtime, svc.containerSvc, svc.configSvc, proxyCfg.proxyConfig) | |
| 330 | 311 | ||
| 331 | 312 | // Create token handler for registry token endpoint | |
| 332 | 313 | if svc.authSvc != nil { | |
| @@ -852,6 +833,58 @@ func loadSecret(ctx context.Context, backend domain.SecretsBackend, path, dataDi | |||
| 852 | 833 | } | |
| 853 | 834 | } | |
| 854 | 835 | ||
| 836 | + | // proxyConfigResult holds parsed proxy and blob chunk size config. | |
| 837 | + | type proxyConfigResult struct { | |
| 838 | + | proxyConfig proxy.Config | |
| 839 | + | maxBlobChunkSize int64 | |
| 840 | + | } | |
| 841 | + | ||
| 842 | + | // buildProxyConfig parses size-related config fields and builds the proxy config. | |
| 843 | + | func buildProxyConfig(cfg Config, log zerowrap.Logger) (*proxyConfigResult, error) { | |
| 844 | + | maxProxyBodySize := int64(512 << 20) // 512MB default | |
| 845 | + | if cfg.Server.MaxProxyBodySize != "" { | |
| 846 | + | parsedSize, err := bytesize.Parse(cfg.Server.MaxProxyBodySize) | |
| 847 | + | if err != nil { | |
| 848 | + | return nil, log.WrapErrWithFields(err, "invalid server.max_proxy_body_size configuration", map[string]any{"value": cfg.Server.MaxProxyBodySize}) | |
| 849 | + | } | |
| 850 | + | maxProxyBodySize = parsedSize | |
| 851 | + | } | |
| 852 | + | ||
| 853 | + | maxBlobChunkSize := int64(registry.DefaultMaxBlobChunkSize) | |
| 854 | + | if cfg.Server.MaxBlobChunkSize != "" { | |
| 855 | + | parsedSize, err := bytesize.Parse(cfg.Server.MaxBlobChunkSize) | |
| 856 | + | if err != nil { | |
| 857 | + | return nil, log.WrapErrWithFields(err, "invalid server.max_blob_chunk_size configuration", map[string]any{"value": cfg.Server.MaxBlobChunkSize}) | |
| 858 | + | } | |
| 859 | + | maxBlobChunkSize = parsedSize | |
| 860 | + | } | |
| 861 | + | ||
| 862 | + | maxProxyResponseSize := int64(1 << 30) // 1GB default | |
| 863 | + | if cfg.Server.MaxProxyResponseSize != "" { | |
| 864 | + | parsedSize, err := bytesize.Parse(cfg.Server.MaxProxyResponseSize) | |
| 865 | + | if err != nil { | |
| 866 | + | return nil, log.WrapErrWithFields(err, "invalid server.max_proxy_response_size configuration", map[string]any{"value": cfg.Server.MaxProxyResponseSize}) | |
| 867 | + | } | |
| 868 | + | maxProxyResponseSize = parsedSize | |
| 869 | + | } | |
| 870 | + | ||
| 871 | + | maxConcurrentConns := cfg.Server.MaxConcurrentConns | |
| 872 | + | if maxConcurrentConns == 0 { | |
| 873 | + | maxConcurrentConns = 10000 // default | |
| 874 | + | } | |
| 875 | + | ||
| 876 | + | return &proxyConfigResult{ | |
| 877 | + | proxyConfig: proxy.Config{ | |
| 878 | + | RegistryDomain: cfg.Server.RegistryDomain, | |
| 879 | + | RegistryPort: cfg.Server.RegistryPort, | |
| 880 | + | MaxBodySize: maxProxyBodySize, | |
| 881 | + | MaxResponseSize: maxProxyResponseSize, | |
| 882 | + | MaxConcurrentConns: maxConcurrentConns, | |
| 883 | + | }, | |
| 884 | + | maxBlobChunkSize: maxBlobChunkSize, | |
| 885 | + | }, nil | |
| 886 | + | } | |
| 887 | + | ||
| 855 | 888 | // createContainerService creates the container service with configuration. | |
| 856 | 889 | func createContainerService(ctx context.Context, v *viper.Viper, cfg Config, svc *services, log zerowrap.Logger) (*container.Service, error) { | |
| 857 | 890 | containerConfig := container.Config{ | |
| @@ -1102,8 +1135,8 @@ func runServers(ctx context.Context, cfg Config, registryHandler, proxyHandler h | |||
| 1102 | 1135 | ||
| 1103 | 1136 | errChan := make(chan error, 2) | |
| 1104 | 1137 | ||
| 1105 | - | registryReady := startServer(fmt.Sprintf(":%d", cfg.Server.RegistryPort), registryHandler, "registry", errChan, log) | |
| 1106 | - | proxyReady := startServer(fmt.Sprintf(":%d", cfg.Server.Port), proxyHandler, "proxy", errChan, log) | |
| 1138 | + | registrySrv, registryReady := startServer(fmt.Sprintf(":%d", cfg.Server.RegistryPort), registryHandler, "registry", errChan, log) | |
| 1139 | + | proxySrv, proxyReady := startServer(fmt.Sprintf(":%d", cfg.Server.Port), proxyHandler, "proxy", errChan, log) | |
| 1107 | 1140 | ||
| 1108 | 1141 | // Wait for both servers to bind their ports before auto-starting containers. | |
| 1109 | 1142 | // This prevents the race where auto-start pulls from the registry before it's listening. | |
| @@ -1153,39 +1186,54 @@ func runServers(ctx context.Context, cfg Config, registryHandler, proxyHandler h | |||
| 1153 | 1186 | } | |
| 1154 | 1187 | ||
| 1155 | 1188 | shutdown: | |
| 1189 | + | gracefulShutdown(registrySrv, proxySrv, containerSvc, log) | |
| 1190 | + | return nil | |
| 1191 | + | } | |
| 1192 | + | ||
| 1193 | + | // gracefulShutdown stops HTTP servers with a 30s timeout, then shuts down | |
| 1194 | + | // the container service and cleans up runtime files. | |
| 1195 | + | func gracefulShutdown(registrySrv, proxySrv *http.Server, containerSvc *container.Service, log zerowrap.Logger) { | |
| 1156 | 1196 | log.Info().Msg("shutting down Gordon...") | |
| 1157 | 1197 | ||
| 1158 | - | if err := containerSvc.Shutdown(ctx); err != nil { | |
| 1198 | + | shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) | |
| 1199 | + | defer shutdownCancel() | |
| 1200 | + | ||
| 1201 | + | if err := registrySrv.Shutdown(shutdownCtx); err != nil { | |
| 1202 | + | log.Warn().Err(err).Msg("registry server shutdown error") | |
| 1203 | + | } | |
| 1204 | + | if err := proxySrv.Shutdown(shutdownCtx); err != nil { | |
| 1205 | + | log.Warn().Err(err).Msg("proxy server shutdown error") | |
| 1206 | + | } | |
| 1207 | + | ||
| 1208 | + | if err := containerSvc.Shutdown(context.Background()); err != nil { | |
| 1159 | 1209 | log.Warn().Err(err).Msg("error during container shutdown") | |
| 1160 | 1210 | } | |
| 1161 | 1211 | ||
| 1162 | - | // Clean up internal credentials file | |
| 1163 | 1212 | cleanupInternalCredentials() | |
| 1164 | - | ||
| 1165 | 1213 | log.Info().Msg("Gordon stopped") | |
| 1166 | - | return nil | |
| 1167 | 1214 | } | |
| 1168 | 1215 | ||
| 1169 | - | // startServer starts an HTTP server, returning a channel that closes once the | |
| 1170 | - | // listening socket is bound. This lets callers wait for the port to be ready | |
| 1171 | - | // before taking actions that depend on it (e.g. auto-start pulling from the | |
| 1172 | - | // local registry). | |
| 1173 | - | func startServer(addr string, handler http.Handler, name string, errChan chan<- error, log zerowrap.Logger) <-chan struct{} { | |
| 1216 | + | // startServer starts an HTTP server, returning the server instance and a channel | |
| 1217 | + | // that closes once the listening socket is bound. This lets callers wait for the | |
| 1218 | + | // port to be ready before taking actions that depend on it (e.g. auto-start | |
| 1219 | + | // pulling from the local registry). The returned *http.Server can be used for | |
| 1220 | + | // graceful shutdown. | |
| 1221 | + | func startServer(addr string, handler http.Handler, name string, errChan chan<- error, log zerowrap.Logger) (*http.Server, <-chan struct{}) { | |
| 1174 | 1222 | ready := make(chan struct{}) | |
| 1175 | 1223 | ||
| 1224 | + | server := &http.Server{ | |
| 1225 | + | Addr: addr, | |
| 1226 | + | Handler: handler, | |
| 1227 | + | ReadHeaderTimeout: 10 * time.Second, | |
| 1228 | + | ReadTimeout: 5 * time.Minute, | |
| 1229 | + | WriteTimeout: 5 * time.Minute, | |
| 1230 | + | IdleTimeout: 120 * time.Second, | |
| 1231 | + | MaxHeaderBytes: 1 << 20, | |
| 1232 | + | } | |
| 1233 | + | ||
| 1176 | 1234 | go func() { | |
| 1177 | 1235 | log.Info().Str("address", addr).Msgf("%s server starting", name) | |
| 1178 | 1236 | ||
| 1179 | - | server := &http.Server{ | |
| 1180 | - | Addr: addr, | |
| 1181 | - | Handler: handler, | |
| 1182 | - | ReadHeaderTimeout: 10 * time.Second, | |
| 1183 | - | ReadTimeout: 5 * time.Minute, | |
| 1184 | - | WriteTimeout: 5 * time.Minute, | |
| 1185 | - | IdleTimeout: 120 * time.Second, | |
| 1186 | - | MaxHeaderBytes: 1 << 20, | |
| 1187 | - | } | |
| 1188 | - | ||
| 1189 | 1237 | ln, err := net.Listen("tcp", addr) | |
| 1190 | 1238 | if err != nil { | |
| 1191 | 1239 | errChan <- fmt.Errorf("%s server error: %w", name, err) | |
| @@ -1198,7 +1246,7 @@ func startServer(addr string, handler http.Handler, name string, errChan chan<- | |||
| 1198 | 1246 | } | |
| 1199 | 1247 | }() | |
| 1200 | 1248 | ||
| 1201 | - | return ready | |
| 1249 | + | return server, ready | |
| 1202 | 1250 | } | |
| 1203 | 1251 | ||
| 1204 | 1252 | // SendReloadSignal sends SIGUSR1 to the running Gordon process. | |
| @@ -64,13 +64,15 @@ import ( | |||||
| 64 | // Config holds the application configuration. | 64 | // Config holds the application configuration. | ||
| 65 | type Config struct { | 65 | type Config struct { | ||
| 66 | Server struct { | 66 | Server struct { | ||
| 67 | - | Port int `mapstructure:"port"` | 67 | + | Port int `mapstructure:"port"` |
| 68 | - | RegistryPort int `mapstructure:"registry_port"` | 68 | + | RegistryPort int `mapstructure:"registry_port"` |
| 69 | - | GordonDomain string `mapstructure:"gordon_domain"` | 69 | + | GordonDomain string `mapstructure:"gordon_domain"` |
| 70 | - | RegistryDomain string `mapstructure:"registry_domain"` // Deprecated: use gordon_domain | 70 | + | RegistryDomain string `mapstructure:"registry_domain"` // Deprecated: use gordon_domain |
| 71 | - | DataDir string `mapstructure:"data_dir"` | 71 | + | DataDir string `mapstructure:"data_dir"` |
| 72 | - | MaxProxyBodySize string `mapstructure:"max_proxy_body_size"` // e.g., "512MB", "1GB" | 72 | + | MaxProxyBodySize string `mapstructure:"max_proxy_body_size"` // e.g., "512MB", "1GB" |
| 73 | - | MaxBlobChunkSize string `mapstructure:"max_blob_chunk_size"` // e.g., "512MB", "1GB" | 73 | + | MaxBlobChunkSize string `mapstructure:"max_blob_chunk_size"` // e.g., "512MB", "1GB" |
| 74 | + | MaxProxyResponseSize string `mapstructure:"max_proxy_response_size"` // e.g., "1GB", "0" for no limit | |||
| 75 | + | MaxConcurrentConns int `mapstructure:"max_concurrent_connections"` | |||
| 74 | } `mapstructure:"server"` | 76 | } `mapstructure:"server"` | ||
| 75 | 77 | ||||
| 76 | Logging struct { | 78 | Logging struct { | ||
| @@ -300,33 +302,12 @@ func createServices(ctx context.Context, v *viper.Viper, cfg Config, log zerowra | |||||
| 300 | } | 302 | } | ||
| 301 | svc.registrySvc = registrySvc.NewService(svc.blobStorage, svc.manifestStorage, svc.eventBus) | 303 | svc.registrySvc = registrySvc.NewService(svc.blobStorage, svc.manifestStorage, svc.eventBus) | ||
| 302 | 304 | ||||
| 303 | - | // Parse max_proxy_body_size config (default: 512MB) | 305 | + | proxyCfg, err := buildProxyConfig(cfg, log) |
| 304 | - | maxProxyBodySize := int64(512 << 20) // 512MB default | 306 | + | if err != nil { |
| 305 | - | if cfg.Server.MaxProxyBodySize != "" { | 307 | + | return nil, err |
| 306 | - | parsedSize, err := bytesize.Parse(cfg.Server.MaxProxyBodySize) | |||
| 307 | - | if err != nil { | |||
| 308 | - | return nil, log.WrapErrWithFields(err, "invalid server.max_proxy_body_size configuration", map[string]any{"value": cfg.Server.MaxProxyBodySize}) | |||
| 309 | - | } | |||
| 310 | - | maxProxyBodySize = parsedSize | |||
| 311 | - | } | |||
| 312 | - | ||||
| 313 | - | // Parse max_blob_chunk_size config (default: 95MB) | |||
| 314 | - | maxBlobChunkSize := int64(registry.DefaultMaxBlobChunkSize) | |||
| 315 | - | if cfg.Server.MaxBlobChunkSize != "" { | |||
| 316 | - | parsedSize, err := bytesize.Parse(cfg.Server.MaxBlobChunkSize) | |||
| 317 | - | if err != nil { | |||
| 318 | - | return nil, log.WrapErrWithFields(err, "invalid server.max_blob_chunk_size configuration", map[string]any{"value": cfg.Server.MaxBlobChunkSize}) | |||
| 319 | - | } | |||
| 320 | - | maxBlobChunkSize = parsedSize | |||
| 321 | } | 308 | } | ||
| 322 | - | 309 | + | svc.maxBlobChunkSize = proxyCfg.maxBlobChunkSize | |
| 323 | - | svc.maxBlobChunkSize = maxBlobChunkSize | 310 | + | svc.proxySvc = proxy.NewService(svc.runtime, svc.containerSvc, svc.configSvc, proxyCfg.proxyConfig) |
| 324 | - | ||||
| 325 | - | svc.proxySvc = proxy.NewService(svc.runtime, svc.containerSvc, svc.configSvc, proxy.Config{ | |||
| 326 | - | RegistryDomain: cfg.Server.RegistryDomain, | |||
| 327 | - | RegistryPort: cfg.Server.RegistryPort, | |||
| 328 | - | MaxBodySize: maxProxyBodySize, | |||
| 329 | - | }) | |||
| 330 | 311 | ||||
| 331 | // Create token handler for registry token endpoint | 312 | // Create token handler for registry token endpoint | ||
| 332 | if svc.authSvc != nil { | 313 | if svc.authSvc != nil { | ||
| @@ -852,6 +833,58 @@ func loadSecret(ctx context.Context, backend domain.SecretsBackend, path, dataDi | |||||
| 852 | } | 833 | } | ||
| 853 | } | 834 | } | ||
| 854 | 835 | ||||
| 836 | + | // proxyConfigResult holds parsed proxy and blob chunk size config. | |||
| 837 | + | type proxyConfigResult struct { | |||
| 838 | + | proxyConfig proxy.Config | |||
| 839 | + | maxBlobChunkSize int64 | |||
| 840 | + | } | |||
| 841 | + | ||||
| 842 | + | // buildProxyConfig parses size-related config fields and builds the proxy config. | |||
| 843 | + | func buildProxyConfig(cfg Config, log zerowrap.Logger) (*proxyConfigResult, error) { | |||
| 844 | + | maxProxyBodySize := int64(512 << 20) // 512MB default | |||
| 845 | + | if cfg.Server.MaxProxyBodySize != "" { | |||
| 846 | + | parsedSize, err := bytesize.Parse(cfg.Server.MaxProxyBodySize) | |||
| 847 | + | if err != nil { | |||
| 848 | + | return nil, log.WrapErrWithFields(err, "invalid server.max_proxy_body_size configuration", map[string]any{"value": cfg.Server.MaxProxyBodySize}) | |||
| 849 | + | } | |||
| 850 | + | maxProxyBodySize = parsedSize | |||
| 851 | + | } | |||
| 852 | + | ||||
| 853 | + | maxBlobChunkSize := int64(registry.DefaultMaxBlobChunkSize) | |||
| 854 | + | if cfg.Server.MaxBlobChunkSize != "" { | |||
| 855 | + | parsedSize, err := bytesize.Parse(cfg.Server.MaxBlobChunkSize) | |||
| 856 | + | if err != nil { | |||
| 857 | + | return nil, log.WrapErrWithFields(err, "invalid server.max_blob_chunk_size configuration", map[string]any{"value": cfg.Server.MaxBlobChunkSize}) | |||
| 858 | + | } | |||
| 859 | + | maxBlobChunkSize = parsedSize | |||
| 860 | + | } | |||
| 861 | + | ||||
| 862 | + | maxProxyResponseSize := int64(1 << 30) // 1GB default | |||
| 863 | + | if cfg.Server.MaxProxyResponseSize != "" { | |||
| 864 | + | parsedSize, err := bytesize.Parse(cfg.Server.MaxProxyResponseSize) | |||
| 865 | + | if err != nil { | |||
| 866 | + | return nil, log.WrapErrWithFields(err, "invalid server.max_proxy_response_size configuration", map[string]any{"value": cfg.Server.MaxProxyResponseSize}) | |||
| 867 | + | } | |||
| 868 | + | maxProxyResponseSize = parsedSize | |||
| 869 | + | } | |||
| 870 | + | ||||
| 871 | + | maxConcurrentConns := cfg.Server.MaxConcurrentConns | |||
| 872 | + | if maxConcurrentConns == 0 { | |||
| 873 | + | maxConcurrentConns = 10000 // default | |||
| 874 | + | } | |||
| 875 | + | ||||
| 876 | + | return &proxyConfigResult{ | |||
| 877 | + | proxyConfig: proxy.Config{ | |||
| 878 | + | RegistryDomain: cfg.Server.RegistryDomain, | |||
| 879 | + | RegistryPort: cfg.Server.RegistryPort, | |||
| 880 | + | MaxBodySize: maxProxyBodySize, | |||
| 881 | + | MaxResponseSize: maxProxyResponseSize, | |||
| 882 | + | MaxConcurrentConns: maxConcurrentConns, | |||
| 883 | + | }, | |||
| 884 | + | maxBlobChunkSize: maxBlobChunkSize, | |||
| 885 | + | }, nil | |||
| 886 | + | } | |||
| 887 | + | ||||
| 855 | // createContainerService creates the container service with configuration. | 888 | // createContainerService creates the container service with configuration. | ||
| 856 | func createContainerService(ctx context.Context, v *viper.Viper, cfg Config, svc *services, log zerowrap.Logger) (*container.Service, error) { | 889 | func createContainerService(ctx context.Context, v *viper.Viper, cfg Config, svc *services, log zerowrap.Logger) (*container.Service, error) { | ||
| 857 | containerConfig := container.Config{ | 890 | containerConfig := container.Config{ | ||
| @@ -1102,8 +1135,8 @@ func runServers(ctx context.Context, cfg Config, registryHandler, proxyHandler h | |||||
| 1102 | 1135 | ||||
| 1103 | errChan := make(chan error, 2) | 1136 | errChan := make(chan error, 2) | ||
| 1104 | 1137 | ||||
| 1105 | - | registryReady := startServer(fmt.Sprintf(":%d", cfg.Server.RegistryPort), registryHandler, "registry", errChan, log) | 1138 | + | registrySrv, registryReady := startServer(fmt.Sprintf(":%d", cfg.Server.RegistryPort), registryHandler, "registry", errChan, log) |
| 1106 | - | proxyReady := startServer(fmt.Sprintf(":%d", cfg.Server.Port), proxyHandler, "proxy", errChan, log) | 1139 | + | proxySrv, proxyReady := startServer(fmt.Sprintf(":%d", cfg.Server.Port), proxyHandler, "proxy", errChan, log) |
| 1107 | 1140 | ||||
| 1108 | // Wait for both servers to bind their ports before auto-starting containers. | 1141 | // Wait for both servers to bind their ports before auto-starting containers. | ||
| 1109 | // This prevents the race where auto-start pulls from the registry before it's listening. | 1142 | // This prevents the race where auto-start pulls from the registry before it's listening. | ||
| @@ -1153,39 +1186,54 @@ func runServers(ctx context.Context, cfg Config, registryHandler, proxyHandler h | |||||
| 1153 | } | 1186 | } | ||
| 1154 | 1187 | ||||
| 1155 | shutdown: | 1188 | shutdown: | ||
| 1189 | + | gracefulShutdown(registrySrv, proxySrv, containerSvc, log) | |||
| 1190 | + | return nil | |||
| 1191 | + | } | |||
| 1192 | + | ||||
| 1193 | + | // gracefulShutdown stops HTTP servers with a 30s timeout, then shuts down | |||
| 1194 | + | // the container service and cleans up runtime files. | |||
| 1195 | + | func gracefulShutdown(registrySrv, proxySrv *http.Server, containerSvc *container.Service, log zerowrap.Logger) { | |||
| 1156 | log.Info().Msg("shutting down Gordon...") | 1196 | log.Info().Msg("shutting down Gordon...") | ||
| 1157 | 1197 | ||||
| 1158 | - | if err := containerSvc.Shutdown(ctx); err != nil { | 1198 | + | shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 1199 | + | defer shutdownCancel() | |||
| 1200 | + | ||||
| 1201 | + | if err := registrySrv.Shutdown(shutdownCtx); err != nil { | |||
| 1202 | + | log.Warn().Err(err).Msg("registry server shutdown error") | |||
| 1203 | + | } | |||
| 1204 | + | if err := proxySrv.Shutdown(shutdownCtx); err != nil { | |||
| 1205 | + | log.Warn().Err(err).Msg("proxy server shutdown error") | |||
| 1206 | + | } | |||
| 1207 | + | ||||
| 1208 | + | if err := containerSvc.Shutdown(context.Background()); err != nil { | |||
| 1159 | log.Warn().Err(err).Msg("error during container shutdown") | 1209 | log.Warn().Err(err).Msg("error during container shutdown") | ||
| 1160 | } | 1210 | } | ||
| 1161 | 1211 | ||||
| 1162 | - | // Clean up internal credentials file | |||
| 1163 | cleanupInternalCredentials() | 1212 | cleanupInternalCredentials() | ||
| 1164 | - | ||||
| 1165 | log.Info().Msg("Gordon stopped") | 1213 | log.Info().Msg("Gordon stopped") | ||
| 1166 | - | return nil | |||
| 1167 | } | 1214 | } | ||
| 1168 | 1215 | ||||
| 1169 | - | // startServer starts an HTTP server, returning a channel that closes once the | 1216 | + | // startServer starts an HTTP server, returning the server instance and a channel |
| 1170 | - | // listening socket is bound. This lets callers wait for the port to be ready | 1217 | + | // that closes once the listening socket is bound. This lets callers wait for the |
| 1171 | - | // before taking actions that depend on it (e.g. auto-start pulling from the | 1218 | + | // port to be ready before taking actions that depend on it (e.g. auto-start |
| 1172 | - | // local registry). | 1219 | + | // pulling from the local registry). The returned *http.Server can be used for |
| 1173 | - | func startServer(addr string, handler http.Handler, name string, errChan chan<- error, log zerowrap.Logger) <-chan struct{} { | 1220 | + | // graceful shutdown. |
| 1221 | + | func startServer(addr string, handler http.Handler, name string, errChan chan<- error, log zerowrap.Logger) (*http.Server, <-chan struct{}) { | |||
| 1174 | ready := make(chan struct{}) | 1222 | ready := make(chan struct{}) | ||
| 1175 | 1223 | ||||
| 1224 | + | server := &http.Server{ | |||
| 1225 | + | Addr: addr, | |||
| 1226 | + | Handler: handler, | |||
| 1227 | + | ReadHeaderTimeout: 10 * time.Second, | |||
| 1228 | + | ReadTimeout: 5 * time.Minute, | |||
| 1229 | + | WriteTimeout: 5 * time.Minute, | |||
| 1230 | + | IdleTimeout: 120 * time.Second, | |||
| 1231 | + | MaxHeaderBytes: 1 << 20, | |||
| 1232 | + | } | |||
| 1233 | + | ||||
| 1176 | go func() { | 1234 | go func() { | ||
| 1177 | log.Info().Str("address", addr).Msgf("%s server starting", name) | 1235 | log.Info().Str("address", addr).Msgf("%s server starting", name) | ||
| 1178 | 1236 | ||||
| 1179 | - | server := &http.Server{ | |||
| 1180 | - | Addr: addr, | |||
| 1181 | - | Handler: handler, | |||
| 1182 | - | ReadHeaderTimeout: 10 * time.Second, | |||
| 1183 | - | ReadTimeout: 5 * time.Minute, | |||
| 1184 | - | WriteTimeout: 5 * time.Minute, | |||
| 1185 | - | IdleTimeout: 120 * time.Second, | |||
| 1186 | - | MaxHeaderBytes: 1 << 20, | |||
| 1187 | - | } | |||
| 1188 | - | ||||
| 1189 | ln, err := net.Listen("tcp", addr) | 1237 | ln, err := net.Listen("tcp", addr) | ||
| 1190 | if err != nil { | 1238 | if err != nil { | ||
| 1191 | errChan <- fmt.Errorf("%s server error: %w", name, err) | 1239 | errChan <- fmt.Errorf("%s server error: %w", name, err) | ||
| @@ -1198,7 +1246,7 @@ func startServer(addr string, handler http.Handler, name string, errChan chan<- | |||||
| 1198 | } | 1246 | } | ||
| 1199 | }() | 1247 | }() | ||
| 1200 | 1248 | ||||
| 1201 | - | return ready | 1249 | + | return server, ready |
| 1202 | } | 1250 | } | ||
| 1203 | 1251 | ||||
| 1204 | // SendReloadSignal sends SIGUSR1 to the running Gordon process. | 1252 | // SendReloadSignal sends SIGUSR1 to the running Gordon process. | ||