fix(deploy): add signal-based readiness and drain with documented timeouts
- commit
- 17b7bd9e42485c09a72945b2ca94bc22c36c785b
- author
- bnema · · 2026-02-07 21:11 UTC
- parent
- 58a3ffa
- signature
- unsigned
10 files · +357 -15
M docs/config/deploy.md +55 -0
| @@ -7,7 +7,11 @@ Controls how Gordon pulls images during deployment. | |||
| 7 | 7 | ```toml | |
| 8 | 8 | [deploy] | |
| 9 | 9 | pull_policy = "if-tag-changed" | |
| 10 | + | readiness_mode = "auto" | |
| 11 | + | health_timeout = "90s" | |
| 10 | 12 | readiness_delay = "5s" | |
| 13 | + | drain_mode = "auto" | |
| 14 | + | drain_timeout = "30s" | |
| 11 | 15 | drain_delay = "2s" | |
| 12 | 16 | ``` | |
| 13 | 17 | ||
| @@ -34,6 +38,29 @@ considered ready for traffic. | |||
| 34 | 38 | ||
| 35 | 39 | Default: `"5s"` | |
| 36 | 40 | ||
| 41 | + | ### `deploy.readiness_mode` | |
| 42 | + | ||
| 43 | + | How Gordon determines that a newly started container is ready for traffic. | |
| 44 | + | ||
| 45 | + | - `auto`: use Docker health status if the container has a healthcheck; otherwise | |
| 46 | + | fall back to `deploy.readiness_delay`. | |
| 47 | + | - `docker-health`: require a Docker healthcheck and wait for `healthy`. | |
| 48 | + | Deploy fails with `no healthcheck detected` if absent. | |
| 49 | + | - `delay`: always use `deploy.readiness_delay`. | |
| 50 | + | ||
| 51 | + | Default: `"auto"` | |
| 52 | + | ||
| 53 | + | ### `deploy.health_timeout` | |
| 54 | + | ||
| 55 | + | Maximum time Gordon waits for health-based readiness in `auto`/`docker-health` | |
| 56 | + | mode before failing the deploy. | |
| 57 | + | ||
| 58 | + | - Uses Go duration format (examples: `"30s"`, `"90s"`, `"2m"`). | |
| 59 | + | - Default `"90s"` balances typical app startup time with fast failure feedback. | |
| 60 | + | - Increase for slow cold-starts, migrations, or heavy warmup workloads. | |
| 61 | + | ||
| 62 | + | Default: `"90s"` | |
| 63 | + | ||
| 37 | 64 | ### `deploy.drain_delay` | |
| 38 | 65 | ||
| 39 | 66 | How long Gordon waits after synchronous proxy cache invalidation before stopping | |
| @@ -45,8 +72,36 @@ the previous container during zero-downtime replacement. | |||
| 45 | 72 | ||
| 46 | 73 | Default: `"2s"` | |
| 47 | 74 | ||
| 75 | + | ### `deploy.drain_mode` | |
| 76 | + | ||
| 77 | + | How Gordon decides when it is safe to stop the previous container after routing | |
| 78 | + | traffic to the new one. | |
| 79 | + | ||
| 80 | + | - `auto`: wait for in-flight proxy requests to drain when available; otherwise | |
| 81 | + | fall back to `deploy.drain_delay`. | |
| 82 | + | - `inflight`: wait for in-flight proxy requests on the previous container to | |
| 83 | + | reach zero (bounded by `deploy.drain_timeout`). | |
| 84 | + | - `delay`: always use `deploy.drain_delay`. | |
| 85 | + | ||
| 86 | + | Default: `"auto"` | |
| 87 | + | ||
| 88 | + | ### `deploy.drain_timeout` | |
| 89 | + | ||
| 90 | + | Maximum time Gordon waits for in-flight request drain before continuing with old | |
| 91 | + | container shutdown. | |
| 92 | + | ||
| 93 | + | - Uses Go duration format (examples: `"10s"`, `"30s"`, `"1m"`). | |
| 94 | + | - Default `"30s"` is a practical upper bound for most HTTP workloads. | |
| 95 | + | - Increase for long-lived requests (large uploads, SSE, long polling). | |
| 96 | + | ||
| 97 | + | Default: `"30s"` | |
| 98 | + | ||
| 48 | 99 | ## Notes | |
| 49 | 100 | ||
| 50 | 101 | - Changes to `deploy.pull_policy` require a restart to take effect. | |
| 102 | + | - Changes to `deploy.readiness_mode` require a restart to take effect. | |
| 103 | + | - Changes to `deploy.health_timeout` require a restart to take effect. | |
| 51 | 104 | - Changes to `deploy.readiness_delay` require a restart to take effect. | |
| 105 | + | - Changes to `deploy.drain_mode` require a restart to take effect. | |
| 106 | + | - Changes to `deploy.drain_timeout` require a restart to take effect. | |
| 52 | 107 | - Changes to `deploy.drain_delay` require a restart to take effect. | |
| @@ -7,7 +7,11 @@ Controls how Gordon pulls images during deployment. | |||||
| 7 | ```toml | 7 | ```toml | ||
| 8 | [deploy] | 8 | [deploy] | ||
| 9 | pull_policy = "if-tag-changed" | 9 | pull_policy = "if-tag-changed" | ||
| 10 | + | readiness_mode = "auto" | |||
| 11 | + | health_timeout = "90s" | |||
| 10 | readiness_delay = "5s" | 12 | readiness_delay = "5s" | ||
| 13 | + | drain_mode = "auto" | |||
| 14 | + | drain_timeout = "30s" | |||
| 11 | drain_delay = "2s" | 15 | drain_delay = "2s" | ||
| 12 | ``` | 16 | ``` | ||
| 13 | 17 | ||||
| @@ -34,6 +38,29 @@ considered ready for traffic. | |||||
| 34 | 38 | ||||
| 35 | Default: `"5s"` | 39 | Default: `"5s"` | ||
| 36 | 40 | ||||
| 41 | + | ### `deploy.readiness_mode` | |||
| 42 | + | ||||
| 43 | + | How Gordon determines that a newly started container is ready for traffic. | |||
| 44 | + | ||||
| 45 | + | - `auto`: use Docker health status if the container has a healthcheck; otherwise | |||
| 46 | + | fall back to `deploy.readiness_delay`. | |||
| 47 | + | - `docker-health`: require a Docker healthcheck and wait for `healthy`. | |||
| 48 | + | Deploy fails with `no healthcheck detected` if absent. | |||
| 49 | + | - `delay`: always use `deploy.readiness_delay`. | |||
| 50 | + | ||||
| 51 | + | Default: `"auto"` | |||
| 52 | + | ||||
| 53 | + | ### `deploy.health_timeout` | |||
| 54 | + | ||||
| 55 | + | Maximum time Gordon waits for health-based readiness in `auto`/`docker-health` | |||
| 56 | + | mode before failing the deploy. | |||
| 57 | + | ||||
| 58 | + | - Uses Go duration format (examples: `"30s"`, `"90s"`, `"2m"`). | |||
| 59 | + | - Default `"90s"` balances typical app startup time with fast failure feedback. | |||
| 60 | + | - Increase for slow cold-starts, migrations, or heavy warmup workloads. | |||
| 61 | + | ||||
| 62 | + | Default: `"90s"` | |||
| 63 | + | ||||
| 37 | ### `deploy.drain_delay` | 64 | ### `deploy.drain_delay` | ||
| 38 | 65 | ||||
| 39 | How long Gordon waits after synchronous proxy cache invalidation before stopping | 66 | How long Gordon waits after synchronous proxy cache invalidation before stopping | ||
| @@ -45,8 +72,36 @@ the previous container during zero-downtime replacement. | |||||
| 45 | 72 | ||||
| 46 | Default: `"2s"` | 73 | Default: `"2s"` | ||
| 47 | 74 | ||||
| 75 | + | ### `deploy.drain_mode` | |||
| 76 | + | ||||
| 77 | + | How Gordon decides when it is safe to stop the previous container after routing | |||
| 78 | + | traffic to the new one. | |||
| 79 | + | ||||
| 80 | + | - `auto`: wait for in-flight proxy requests to drain when available; otherwise | |||
| 81 | + | fall back to `deploy.drain_delay`. | |||
| 82 | + | - `inflight`: wait for in-flight proxy requests on the previous container to | |||
| 83 | + | reach zero (bounded by `deploy.drain_timeout`). | |||
| 84 | + | - `delay`: always use `deploy.drain_delay`. | |||
| 85 | + | ||||
| 86 | + | Default: `"auto"` | |||
| 87 | + | ||||
| 88 | + | ### `deploy.drain_timeout` | |||
| 89 | + | ||||
| 90 | + | Maximum time Gordon waits for in-flight request drain before continuing with old | |||
| 91 | + | container shutdown. | |||
| 92 | + | ||||
| 93 | + | - Uses Go duration format (examples: `"10s"`, `"30s"`, `"1m"`). | |||
| 94 | + | - Default `"30s"` is a practical upper bound for most HTTP workloads. | |||
| 95 | + | - Increase for long-lived requests (large uploads, SSE, long polling). | |||
| 96 | + | ||||
| 97 | + | Default: `"30s"` | |||
| 98 | + | ||||
| 48 | ## Notes | 99 | ## Notes | ||
| 49 | 100 | ||||
| 50 | - Changes to `deploy.pull_policy` require a restart to take effect. | 101 | - Changes to `deploy.pull_policy` require a restart to take effect. | ||
| 102 | + | - Changes to `deploy.readiness_mode` require a restart to take effect. | |||
| 103 | + | - Changes to `deploy.health_timeout` require a restart to take effect. | |||
| 51 | - Changes to `deploy.readiness_delay` require a restart to take effect. | 104 | - Changes to `deploy.readiness_delay` require a restart to take effect. | ||
| 105 | + | - Changes to `deploy.drain_mode` require a restart to take effect. | |||
| 106 | + | - Changes to `deploy.drain_timeout` require a restart to take effect. | |||
| 52 | - Changes to `deploy.drain_delay` require a restart to take effect. | 107 | - Changes to `deploy.drain_delay` require a restart to take effect. | ||
M docs/config/index.md +12 -0
| @@ -56,7 +56,11 @@ trusted_proxies = [] # IPs/CIDRs trusted for X-Forwarded-For | |||
| 56 | 56 | # Deploy behavior | |
| 57 | 57 | [deploy] | |
| 58 | 58 | pull_policy = "if-tag-changed" # always, if-not-present, if-tag-changed | |
| 59 | + | readiness_mode = "auto" # auto, docker-health, delay | |
| 60 | + | health_timeout = "90s" # Max wait for health-based readiness | |
| 59 | 61 | readiness_delay = "5s" # Wait after running before ready | |
| 62 | + | drain_mode = "auto" # auto, inflight, delay | |
| 63 | + | drain_timeout = "30s" # Max wait for in-flight drain | |
| 60 | 64 | drain_delay = "2s" # Wait after proxy invalidation before old stop | |
| 61 | 65 | ||
| 62 | 66 | # Logging | |
| @@ -159,7 +163,11 @@ monthly = 12 | |||
| 159 | 163 | | `api.rate_limit.burst` | `100` | | |
| 160 | 164 | | `api.rate_limit.trusted_proxies` | `[]` | | |
| 161 | 165 | | `deploy.pull_policy` | `"if-tag-changed"` | | |
| 166 | + | | `deploy.readiness_mode` | `"auto"` | | |
| 167 | + | | `deploy.health_timeout` | `"90s"` | | |
| 162 | 168 | | `deploy.readiness_delay` | `"5s"` | | |
| 169 | + | | `deploy.drain_mode` | `"auto"` | | |
| 170 | + | | `deploy.drain_timeout` | `"30s"` | | |
| 163 | 171 | | `deploy.drain_delay` | `"2s"` | | |
| 164 | 172 | | `logging.level` | `"info"` | | |
| 165 | 173 | | `logging.format` | `"console"` | | |
| @@ -190,7 +198,11 @@ The following settings require a restart to take effect: | |||
| 190 | 198 | - `server.data_dir` | |
| 191 | 199 | - `auth` settings | |
| 192 | 200 | - `deploy.pull_policy` | |
| 201 | + | - `deploy.readiness_mode` | |
| 202 | + | - `deploy.health_timeout` | |
| 193 | 203 | - `deploy.readiness_delay` | |
| 204 | + | - `deploy.drain_mode` | |
| 205 | + | - `deploy.drain_timeout` | |
| 194 | 206 | - `deploy.drain_delay` | |
| 195 | 207 | ||
| 196 | 208 | ## Environment Variable Override | |
| @@ -56,7 +56,11 @@ trusted_proxies = [] # IPs/CIDRs trusted for X-Forwarded-For | |||||
| 56 | # Deploy behavior | 56 | # Deploy behavior | ||
| 57 | [deploy] | 57 | [deploy] | ||
| 58 | pull_policy = "if-tag-changed" # always, if-not-present, if-tag-changed | 58 | pull_policy = "if-tag-changed" # always, if-not-present, if-tag-changed | ||
| 59 | + | readiness_mode = "auto" # auto, docker-health, delay | |||
| 60 | + | health_timeout = "90s" # Max wait for health-based readiness | |||
| 59 | readiness_delay = "5s" # Wait after running before ready | 61 | readiness_delay = "5s" # Wait after running before ready | ||
| 62 | + | drain_mode = "auto" # auto, inflight, delay | |||
| 63 | + | drain_timeout = "30s" # Max wait for in-flight drain | |||
| 60 | drain_delay = "2s" # Wait after proxy invalidation before old stop | 64 | drain_delay = "2s" # Wait after proxy invalidation before old stop | ||
| 61 | 65 | ||||
| 62 | # Logging | 66 | # Logging | ||
| @@ -159,7 +163,11 @@ monthly = 12 | |||||
| 159 | | `api.rate_limit.burst` | `100` | | 163 | | `api.rate_limit.burst` | `100` | | ||
| 160 | | `api.rate_limit.trusted_proxies` | `[]` | | 164 | | `api.rate_limit.trusted_proxies` | `[]` | | ||
| 161 | | `deploy.pull_policy` | `"if-tag-changed"` | | 165 | | `deploy.pull_policy` | `"if-tag-changed"` | | ||
| 166 | + | | `deploy.readiness_mode` | `"auto"` | | |||
| 167 | + | | `deploy.health_timeout` | `"90s"` | | |||
| 162 | | `deploy.readiness_delay` | `"5s"` | | 168 | | `deploy.readiness_delay` | `"5s"` | | ||
| 169 | + | | `deploy.drain_mode` | `"auto"` | | |||
| 170 | + | | `deploy.drain_timeout` | `"30s"` | | |||
| 163 | | `deploy.drain_delay` | `"2s"` | | 171 | | `deploy.drain_delay` | `"2s"` | | ||
| 164 | | `logging.level` | `"info"` | | 172 | | `logging.level` | `"info"` | | ||
| 165 | | `logging.format` | `"console"` | | 173 | | `logging.format` | `"console"` | | ||
| @@ -190,7 +198,11 @@ The following settings require a restart to take effect: | |||||
| 190 | - `server.data_dir` | 198 | - `server.data_dir` | ||
| 191 | - `auth` settings | 199 | - `auth` settings | ||
| 192 | - `deploy.pull_policy` | 200 | - `deploy.pull_policy` | ||
| 201 | + | - `deploy.readiness_mode` | |||
| 202 | + | - `deploy.health_timeout` | |||
| 193 | - `deploy.readiness_delay` | 203 | - `deploy.readiness_delay` | ||
| 204 | + | - `deploy.drain_mode` | |||
| 205 | + | - `deploy.drain_timeout` | |||
| 194 | - `deploy.drain_delay` | 206 | - `deploy.drain_delay` | ||
| 195 | 207 | ||||
| 196 | ## Environment Variable Override | 208 | ## Environment Variable Override | ||
M docs/config/reference.md +8 -0
| @@ -74,7 +74,11 @@ dir = "" # Env files directory (default: {da | |||
| 74 | 74 | # ============================================================================= | |
| 75 | 75 | [deploy] | |
| 76 | 76 | pull_policy = "if-tag-changed" # "always", "if-tag-changed", "never" | |
| 77 | + | readiness_mode = "auto" # "auto", "docker-health", "delay" | |
| 78 | + | health_timeout = "90s" # Max wait for health-based readiness | |
| 77 | 79 | readiness_delay = "5s" # Wait after running before considered ready | |
| 80 | + | drain_mode = "auto" # "auto", "inflight", "delay" | |
| 81 | + | drain_timeout = "30s" # Max wait for in-flight request drain | |
| 78 | 82 | drain_delay = "2s" # Wait after cache invalidation before old stop | |
| 79 | 83 | ||
| 80 | 84 | # ============================================================================= | |
| @@ -174,7 +178,11 @@ monthly = 0 # Keep N monthly backups per DB | |||
| 174 | 178 | | `logging.container_logs.max_backups` | `3` | Keep 3 old files | | |
| 175 | 179 | | `logging.container_logs.max_age` | `28` | 28 days | | |
| 176 | 180 | | `deploy.pull_policy` | `"if-tag-changed"` | Pull on tag change | | |
| 181 | + | | `deploy.readiness_mode` | `"auto"` | Readiness strategy (`auto`, `docker-health`, `delay`) | | |
| 182 | + | | `deploy.health_timeout` | `"90s"` | Max wait for health-based readiness before deploy fails | | |
| 177 | 183 | | `deploy.readiness_delay` | `"5s"` | Delay before container is considered ready | | |
| 184 | + | | `deploy.drain_mode` | `"auto"` | Drain strategy (`auto`, `inflight`, `delay`) | | |
| 185 | + | | `deploy.drain_timeout` | `"30s"` | Max wait for in-flight request drain before old stop | | |
| 178 | 186 | | `deploy.drain_delay` | `"2s"` | Delay before stopping previous container after cache invalidation | | |
| 179 | 187 | | `auto_route.enabled` | `false` | Auto-route disabled | | |
| 180 | 188 | | `network_isolation.enabled` | `false` | Isolation disabled | | |
| @@ -74,7 +74,11 @@ dir = "" # Env files directory (default: {da | |||||
| 74 | # ============================================================================= | 74 | # ============================================================================= | ||
| 75 | [deploy] | 75 | [deploy] | ||
| 76 | pull_policy = "if-tag-changed" # "always", "if-tag-changed", "never" | 76 | pull_policy = "if-tag-changed" # "always", "if-tag-changed", "never" | ||
| 77 | + | readiness_mode = "auto" # "auto", "docker-health", "delay" | |||
| 78 | + | health_timeout = "90s" # Max wait for health-based readiness | |||
| 77 | readiness_delay = "5s" # Wait after running before considered ready | 79 | readiness_delay = "5s" # Wait after running before considered ready | ||
| 80 | + | drain_mode = "auto" # "auto", "inflight", "delay" | |||
| 81 | + | drain_timeout = "30s" # Max wait for in-flight request drain | |||
| 78 | drain_delay = "2s" # Wait after cache invalidation before old stop | 82 | drain_delay = "2s" # Wait after cache invalidation before old stop | ||
| 79 | 83 | ||||
| 80 | # ============================================================================= | 84 | # ============================================================================= | ||
| @@ -174,7 +178,11 @@ monthly = 0 # Keep N monthly backups per DB | |||||
| 174 | | `logging.container_logs.max_backups` | `3` | Keep 3 old files | | 178 | | `logging.container_logs.max_backups` | `3` | Keep 3 old files | | ||
| 175 | | `logging.container_logs.max_age` | `28` | 28 days | | 179 | | `logging.container_logs.max_age` | `28` | 28 days | | ||
| 176 | | `deploy.pull_policy` | `"if-tag-changed"` | Pull on tag change | | 180 | | `deploy.pull_policy` | `"if-tag-changed"` | Pull on tag change | | ||
| 181 | + | | `deploy.readiness_mode` | `"auto"` | Readiness strategy (`auto`, `docker-health`, `delay`) | | |||
| 182 | + | | `deploy.health_timeout` | `"90s"` | Max wait for health-based readiness before deploy fails | | |||
| 177 | | `deploy.readiness_delay` | `"5s"` | Delay before container is considered ready | | 183 | | `deploy.readiness_delay` | `"5s"` | Delay before container is considered ready | | ||
| 184 | + | | `deploy.drain_mode` | `"auto"` | Drain strategy (`auto`, `inflight`, `delay`) | | |||
| 185 | + | | `deploy.drain_timeout` | `"30s"` | Max wait for in-flight request drain before old stop | | |||
| 178 | | `deploy.drain_delay` | `"2s"` | Delay before stopping previous container after cache invalidation | | 186 | | `deploy.drain_delay` | `"2s"` | Delay before stopping previous container after cache invalidation | | ||
| 179 | | `auto_route.enabled` | `false` | Auto-route disabled | | 187 | | `auto_route.enabled` | `false` | Auto-route disabled | | ||
| 180 | | `network_isolation.enabled` | `false` | Isolation disabled | | 188 | | `network_isolation.enabled` | `false` | Isolation disabled | | ||
M internal/adapters/out/docker/runtime.go +21 -0
| @@ -596,6 +596,27 @@ func (r *Runtime) IsContainerRunning(ctx context.Context, containerID string) (b | |||
| 596 | 596 | return ctr.Status == string(domain.ContainerStatusRunning), nil | |
| 597 | 597 | } | |
| 598 | 598 | ||
| 599 | + | // GetContainerHealthStatus returns Docker healthcheck status when available. | |
| 600 | + | func (r *Runtime) GetContainerHealthStatus(ctx context.Context, containerID string) (string, bool, error) { | |
| 601 | + | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | |
| 602 | + | zerowrap.FieldLayer: "adapter", | |
| 603 | + | zerowrap.FieldAdapter: "docker", | |
| 604 | + | zerowrap.FieldAction: "GetContainerHealthStatus", | |
| 605 | + | zerowrap.FieldEntityID: containerID, | |
| 606 | + | }) | |
| 607 | + | log := zerowrap.FromCtx(ctx) | |
| 608 | + | ||
| 609 | + | resp, err := r.client.ContainerInspect(ctx, containerID) | |
| 610 | + | if err != nil { | |
| 611 | + | return "", false, log.WrapErr(err, "failed to inspect container") | |
| 612 | + | } | |
| 613 | + | if resp.State == nil || resp.State.Health == nil { | |
| 614 | + | return "", false, nil | |
| 615 | + | } | |
| 616 | + | ||
| 617 | + | return resp.State.Health.Status, true, nil | |
| 618 | + | } | |
| 619 | + | ||
| 599 | 620 | // GetContainerPort gets the host port for a container's internal port. | |
| 600 | 621 | func (r *Runtime) GetContainerPort(ctx context.Context, containerID string, internalPort int) (int, error) { | |
| 601 | 622 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | |
| @@ -596,6 +596,27 @@ func (r *Runtime) IsContainerRunning(ctx context.Context, containerID string) (b | |||||
| 596 | return ctr.Status == string(domain.ContainerStatusRunning), nil | 596 | return ctr.Status == string(domain.ContainerStatusRunning), nil | ||
| 597 | } | 597 | } | ||
| 598 | 598 | ||||
| 599 | + | // GetContainerHealthStatus returns Docker healthcheck status when available. | |||
| 600 | + | func (r *Runtime) GetContainerHealthStatus(ctx context.Context, containerID string) (string, bool, error) { | |||
| 601 | + | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | |||
| 602 | + | zerowrap.FieldLayer: "adapter", | |||
| 603 | + | zerowrap.FieldAdapter: "docker", | |||
| 604 | + | zerowrap.FieldAction: "GetContainerHealthStatus", | |||
| 605 | + | zerowrap.FieldEntityID: containerID, | |||
| 606 | + | }) | |||
| 607 | + | log := zerowrap.FromCtx(ctx) | |||
| 608 | + | ||||
| 609 | + | resp, err := r.client.ContainerInspect(ctx, containerID) | |||
| 610 | + | if err != nil { | |||
| 611 | + | return "", false, log.WrapErr(err, "failed to inspect container") | |||
| 612 | + | } | |||
| 613 | + | if resp.State == nil || resp.State.Health == nil { | |||
| 614 | + | return "", false, nil | |||
| 615 | + | } | |||
| 616 | + | ||||
| 617 | + | return resp.State.Health.Status, true, nil | |||
| 618 | + | } | |||
| 619 | + | ||||
| 599 | // GetContainerPort gets the host port for a container's internal port. | 620 | // GetContainerPort gets the host port for a container's internal port. | ||
| 600 | func (r *Runtime) GetContainerPort(ctx context.Context, containerID string, internalPort int) (int, error) { | 621 | func (r *Runtime) GetContainerPort(ctx context.Context, containerID string, internalPort int) (int, error) { | ||
| 601 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | 622 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | ||
M internal/app/run.go +9 -0
| @@ -463,6 +463,7 @@ func createServices(ctx context.Context, v *viper.Viper, cfg Config, log zerowra | |||
| 463 | 463 | // Wire synchronous proxy cache invalidation for zero-downtime deployments. | |
| 464 | 464 | // The proxy service implements out.ProxyCacheInvalidator via InvalidateTarget(). | |
| 465 | 465 | svc.containerSvc.SetProxyCacheInvalidator(svc.proxySvc) | |
| 466 | + | svc.containerSvc.SetProxyDrainWaiter(svc.proxySvc) | |
| 466 | 467 | ||
| 467 | 468 | // Create token handler for registry token endpoint | |
| 468 | 469 | if svc.authSvc != nil { | |
| @@ -1059,7 +1060,11 @@ func createContainerService(ctx context.Context, v *viper.Viper, cfg Config, svc | |||
| 1059 | 1060 | NetworkGroups: svc.configSvc.GetNetworkGroups(), | |
| 1060 | 1061 | Attachments: svc.configSvc.GetAttachments(), | |
| 1061 | 1062 | ReadinessDelay: v.GetDuration("deploy.readiness_delay"), | |
| 1063 | + | ReadinessMode: v.GetString("deploy.readiness_mode"), | |
| 1064 | + | HealthTimeout: v.GetDuration("deploy.health_timeout"), | |
| 1062 | 1065 | DrainDelay: v.GetDuration("deploy.drain_delay"), | |
| 1066 | + | DrainMode: v.GetString("deploy.drain_mode"), | |
| 1067 | + | DrainTimeout: v.GetDuration("deploy.drain_timeout"), | |
| 1063 | 1068 | } | |
| 1064 | 1069 | ||
| 1065 | 1070 | if cfg.Auth.Enabled && svc.authSvc != nil { | |
| @@ -1855,7 +1860,11 @@ func loadConfig(v *viper.Viper, configPath string) error { | |||
| 1855 | 1860 | v.SetDefault("server.max_concurrent_connections", -1) // -1 = use default (10000), 0 = no limit | |
| 1856 | 1861 | v.SetDefault("server.registry_allowed_ips", []string{}) | |
| 1857 | 1862 | v.SetDefault("deploy.readiness_delay", "5s") | |
| 1863 | + | v.SetDefault("deploy.readiness_mode", "auto") | |
| 1864 | + | v.SetDefault("deploy.health_timeout", "90s") | |
| 1858 | 1865 | v.SetDefault("deploy.drain_delay", "2s") | |
| 1866 | + | v.SetDefault("deploy.drain_mode", "auto") | |
| 1867 | + | v.SetDefault("deploy.drain_timeout", "30s") | |
| 1859 | 1868 | ||
| 1860 | 1869 | ConfigureViper(v, configPath) | |
| 1861 | 1870 | ||
| @@ -463,6 +463,7 @@ func createServices(ctx context.Context, v *viper.Viper, cfg Config, log zerowra | |||||
| 463 | // Wire synchronous proxy cache invalidation for zero-downtime deployments. | 463 | // Wire synchronous proxy cache invalidation for zero-downtime deployments. | ||
| 464 | // The proxy service implements out.ProxyCacheInvalidator via InvalidateTarget(). | 464 | // The proxy service implements out.ProxyCacheInvalidator via InvalidateTarget(). | ||
| 465 | svc.containerSvc.SetProxyCacheInvalidator(svc.proxySvc) | 465 | svc.containerSvc.SetProxyCacheInvalidator(svc.proxySvc) | ||
| 466 | + | svc.containerSvc.SetProxyDrainWaiter(svc.proxySvc) | |||
| 466 | 467 | ||||
| 467 | // Create token handler for registry token endpoint | 468 | // Create token handler for registry token endpoint | ||
| 468 | if svc.authSvc != nil { | 469 | if svc.authSvc != nil { | ||
| @@ -1059,7 +1060,11 @@ func createContainerService(ctx context.Context, v *viper.Viper, cfg Config, svc | |||||
| 1059 | NetworkGroups: svc.configSvc.GetNetworkGroups(), | 1060 | NetworkGroups: svc.configSvc.GetNetworkGroups(), | ||
| 1060 | Attachments: svc.configSvc.GetAttachments(), | 1061 | Attachments: svc.configSvc.GetAttachments(), | ||
| 1061 | ReadinessDelay: v.GetDuration("deploy.readiness_delay"), | 1062 | ReadinessDelay: v.GetDuration("deploy.readiness_delay"), | ||
| 1063 | + | ReadinessMode: v.GetString("deploy.readiness_mode"), | |||
| 1064 | + | HealthTimeout: v.GetDuration("deploy.health_timeout"), | |||
| 1062 | DrainDelay: v.GetDuration("deploy.drain_delay"), | 1065 | DrainDelay: v.GetDuration("deploy.drain_delay"), | ||
| 1066 | + | DrainMode: v.GetString("deploy.drain_mode"), | |||
| 1067 | + | DrainTimeout: v.GetDuration("deploy.drain_timeout"), | |||
| 1063 | } | 1068 | } | ||
| 1064 | 1069 | ||||
| 1065 | if cfg.Auth.Enabled && svc.authSvc != nil { | 1070 | if cfg.Auth.Enabled && svc.authSvc != nil { | ||
| @@ -1855,7 +1860,11 @@ func loadConfig(v *viper.Viper, configPath string) error { | |||||
| 1855 | v.SetDefault("server.max_concurrent_connections", -1) // -1 = use default (10000), 0 = no limit | 1860 | v.SetDefault("server.max_concurrent_connections", -1) // -1 = use default (10000), 0 = no limit | ||
| 1856 | v.SetDefault("server.registry_allowed_ips", []string{}) | 1861 | v.SetDefault("server.registry_allowed_ips", []string{}) | ||
| 1857 | v.SetDefault("deploy.readiness_delay", "5s") | 1862 | v.SetDefault("deploy.readiness_delay", "5s") | ||
| 1863 | + | v.SetDefault("deploy.readiness_mode", "auto") | |||
| 1864 | + | v.SetDefault("deploy.health_timeout", "90s") | |||
| 1858 | v.SetDefault("deploy.drain_delay", "2s") | 1865 | v.SetDefault("deploy.drain_delay", "2s") | ||
| 1866 | + | v.SetDefault("deploy.drain_mode", "auto") | |||
| 1867 | + | v.SetDefault("deploy.drain_timeout", "30s") | |||
| 1859 | 1868 | ||||
| 1860 | ConfigureViper(v, configPath) | 1869 | ConfigureViper(v, configPath) | ||
| 1861 | 1870 | ||||
M internal/boundaries/out/mocks/mock_container_runtime.go +72 -0
| @@ -570,6 +570,78 @@ func (_c *MockContainerRuntime_GetContainerExposedPorts_Call) RunAndReturn(run f | |||
| 570 | 570 | return _c | |
| 571 | 571 | } | |
| 572 | 572 | ||
| 573 | + | // GetContainerHealthStatus provides a mock function for the type MockContainerRuntime | |
| 574 | + | func (_mock *MockContainerRuntime) GetContainerHealthStatus(ctx context.Context, containerID string) (string, bool, error) { | |
| 575 | + | ret := _mock.Called(ctx, containerID) | |
| 576 | + | ||
| 577 | + | if len(ret) == 0 { | |
| 578 | + | panic("no return value specified for GetContainerHealthStatus") | |
| 579 | + | } | |
| 580 | + | ||
| 581 | + | var r0 string | |
| 582 | + | var r1 bool | |
| 583 | + | var r2 error | |
| 584 | + | if returnFunc, ok := ret.Get(0).(func(context.Context, string) (string, bool, error)); ok { | |
| 585 | + | return returnFunc(ctx, containerID) | |
| 586 | + | } | |
| 587 | + | if returnFunc, ok := ret.Get(0).(func(context.Context, string) string); ok { | |
| 588 | + | r0 = returnFunc(ctx, containerID) | |
| 589 | + | } else { | |
| 590 | + | r0 = ret.Get(0).(string) | |
| 591 | + | } | |
| 592 | + | if returnFunc, ok := ret.Get(1).(func(context.Context, string) bool); ok { | |
| 593 | + | r1 = returnFunc(ctx, containerID) | |
| 594 | + | } else { | |
| 595 | + | r1 = ret.Get(1).(bool) | |
| 596 | + | } | |
| 597 | + | if returnFunc, ok := ret.Get(2).(func(context.Context, string) error); ok { | |
| 598 | + | r2 = returnFunc(ctx, containerID) | |
| 599 | + | } else { | |
| 600 | + | r2 = ret.Error(2) | |
| 601 | + | } | |
| 602 | + | return r0, r1, r2 | |
| 603 | + | } | |
| 604 | + | ||
| 605 | + | // MockContainerRuntime_GetContainerHealthStatus_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetContainerHealthStatus' | |
| 606 | + | type MockContainerRuntime_GetContainerHealthStatus_Call struct { | |
| 607 | + | *mock.Call | |
| 608 | + | } | |
| 609 | + | ||
| 610 | + | // GetContainerHealthStatus is a helper method to define mock.On call | |
| 611 | + | // - ctx context.Context | |
| 612 | + | // - containerID string | |
| 613 | + | func (_e *MockContainerRuntime_Expecter) GetContainerHealthStatus(ctx interface{}, containerID interface{}) *MockContainerRuntime_GetContainerHealthStatus_Call { | |
| 614 | + | return &MockContainerRuntime_GetContainerHealthStatus_Call{Call: _e.mock.On("GetContainerHealthStatus", ctx, containerID)} | |
| 615 | + | } | |
| 616 | + | ||
| 617 | + | func (_c *MockContainerRuntime_GetContainerHealthStatus_Call) Run(run func(ctx context.Context, containerID string)) *MockContainerRuntime_GetContainerHealthStatus_Call { | |
| 618 | + | _c.Call.Run(func(args mock.Arguments) { | |
| 619 | + | var arg0 context.Context | |
| 620 | + | if args[0] != nil { | |
| 621 | + | arg0 = args[0].(context.Context) | |
| 622 | + | } | |
| 623 | + | var arg1 string | |
| 624 | + | if args[1] != nil { | |
| 625 | + | arg1 = args[1].(string) | |
| 626 | + | } | |
| 627 | + | run( | |
| 628 | + | arg0, | |
| 629 | + | arg1, | |
| 630 | + | ) | |
| 631 | + | }) | |
| 632 | + | return _c | |
| 633 | + | } | |
| 634 | + | ||
| 635 | + | func (_c *MockContainerRuntime_GetContainerHealthStatus_Call) Return(status string, hasHealthcheck bool, err error) *MockContainerRuntime_GetContainerHealthStatus_Call { | |
| 636 | + | _c.Call.Return(status, hasHealthcheck, err) | |
| 637 | + | return _c | |
| 638 | + | } | |
| 639 | + | ||
| 640 | + | func (_c *MockContainerRuntime_GetContainerHealthStatus_Call) RunAndReturn(run func(ctx context.Context, containerID string) (string, bool, error)) *MockContainerRuntime_GetContainerHealthStatus_Call { | |
| 641 | + | _c.Call.Return(run) | |
| 642 | + | return _c | |
| 643 | + | } | |
| 644 | + | ||
| 573 | 645 | // GetContainerLogs provides a mock function for the type MockContainerRuntime | |
| 574 | 646 | func (_mock *MockContainerRuntime) GetContainerLogs(ctx context.Context, containerID string, follow bool) (io.ReadCloser, error) { | |
| 575 | 647 | ret := _mock.Called(ctx, containerID, follow) | |
| @@ -570,6 +570,78 @@ func (_c *MockContainerRuntime_GetContainerExposedPorts_Call) RunAndReturn(run f | |||||
| 570 | return _c | 570 | return _c | ||
| 571 | } | 571 | } | ||
| 572 | 572 | ||||
| 573 | + | // GetContainerHealthStatus provides a mock function for the type MockContainerRuntime | |||
| 574 | + | func (_mock *MockContainerRuntime) GetContainerHealthStatus(ctx context.Context, containerID string) (string, bool, error) { | |||
| 575 | + | ret := _mock.Called(ctx, containerID) | |||
| 576 | + | ||||
| 577 | + | if len(ret) == 0 { | |||
| 578 | + | panic("no return value specified for GetContainerHealthStatus") | |||
| 579 | + | } | |||
| 580 | + | ||||
| 581 | + | var r0 string | |||
| 582 | + | var r1 bool | |||
| 583 | + | var r2 error | |||
| 584 | + | if returnFunc, ok := ret.Get(0).(func(context.Context, string) (string, bool, error)); ok { | |||
| 585 | + | return returnFunc(ctx, containerID) | |||
| 586 | + | } | |||
| 587 | + | if returnFunc, ok := ret.Get(0).(func(context.Context, string) string); ok { | |||
| 588 | + | r0 = returnFunc(ctx, containerID) | |||
| 589 | + | } else { | |||
| 590 | + | r0 = ret.Get(0).(string) | |||
| 591 | + | } | |||
| 592 | + | if returnFunc, ok := ret.Get(1).(func(context.Context, string) bool); ok { | |||
| 593 | + | r1 = returnFunc(ctx, containerID) | |||
| 594 | + | } else { | |||
| 595 | + | r1 = ret.Get(1).(bool) | |||
| 596 | + | } | |||
| 597 | + | if returnFunc, ok := ret.Get(2).(func(context.Context, string) error); ok { | |||
| 598 | + | r2 = returnFunc(ctx, containerID) | |||
| 599 | + | } else { | |||
| 600 | + | r2 = ret.Error(2) | |||
| 601 | + | } | |||
| 602 | + | return r0, r1, r2 | |||
| 603 | + | } | |||
| 604 | + | ||||
| 605 | + | // MockContainerRuntime_GetContainerHealthStatus_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetContainerHealthStatus' | |||
| 606 | + | type MockContainerRuntime_GetContainerHealthStatus_Call struct { | |||
| 607 | + | *mock.Call | |||
| 608 | + | } | |||
| 609 | + | ||||
| 610 | + | // GetContainerHealthStatus is a helper method to define mock.On call | |||
| 611 | + | // - ctx context.Context | |||
| 612 | + | // - containerID string | |||
| 613 | + | func (_e *MockContainerRuntime_Expecter) GetContainerHealthStatus(ctx interface{}, containerID interface{}) *MockContainerRuntime_GetContainerHealthStatus_Call { | |||
| 614 | + | return &MockContainerRuntime_GetContainerHealthStatus_Call{Call: _e.mock.On("GetContainerHealthStatus", ctx, containerID)} | |||
| 615 | + | } | |||
| 616 | + | ||||
| 617 | + | func (_c *MockContainerRuntime_GetContainerHealthStatus_Call) Run(run func(ctx context.Context, containerID string)) *MockContainerRuntime_GetContainerHealthStatus_Call { | |||
| 618 | + | _c.Call.Run(func(args mock.Arguments) { | |||
| 619 | + | var arg0 context.Context | |||
| 620 | + | if args[0] != nil { | |||
| 621 | + | arg0 = args[0].(context.Context) | |||
| 622 | + | } | |||
| 623 | + | var arg1 string | |||
| 624 | + | if args[1] != nil { | |||
| 625 | + | arg1 = args[1].(string) | |||
| 626 | + | } | |||
| 627 | + | run( | |||
| 628 | + | arg0, | |||
| 629 | + | arg1, | |||
| 630 | + | ) | |||
| 631 | + | }) | |||
| 632 | + | return _c | |||
| 633 | + | } | |||
| 634 | + | ||||
| 635 | + | func (_c *MockContainerRuntime_GetContainerHealthStatus_Call) Return(status string, hasHealthcheck bool, err error) *MockContainerRuntime_GetContainerHealthStatus_Call { | |||
| 636 | + | _c.Call.Return(status, hasHealthcheck, err) | |||
| 637 | + | return _c | |||
| 638 | + | } | |||
| 639 | + | ||||
| 640 | + | func (_c *MockContainerRuntime_GetContainerHealthStatus_Call) RunAndReturn(run func(ctx context.Context, containerID string) (string, bool, error)) *MockContainerRuntime_GetContainerHealthStatus_Call { | |||
| 641 | + | _c.Call.Return(run) | |||
| 642 | + | return _c | |||
| 643 | + | } | |||
| 644 | + | ||||
| 573 | // GetContainerLogs provides a mock function for the type MockContainerRuntime | 645 | // GetContainerLogs provides a mock function for the type MockContainerRuntime | ||
| 574 | func (_mock *MockContainerRuntime) GetContainerLogs(ctx context.Context, containerID string, follow bool) (io.ReadCloser, error) { | 646 | func (_mock *MockContainerRuntime) GetContainerLogs(ctx context.Context, containerID string, follow bool) (io.ReadCloser, error) { | ||
| 575 | ret := _mock.Called(ctx, containerID, follow) | 647 | ret := _mock.Called(ctx, containerID, follow) | ||
M internal/boundaries/out/proxy_cache.go +10 -1
| @@ -1,6 +1,9 @@ | |||
| 1 | 1 | package out | |
| 2 | 2 | ||
| 3 | - | import "context" | |
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "time" | |
| 6 | + | ) | |
| 4 | 7 | ||
| 5 | 8 | // ProxyCacheInvalidator defines the contract for synchronously invalidating | |
| 6 | 9 | // proxy target cache entries. This is used during zero-downtime deployments | |
| @@ -8,3 +11,9 @@ import "context" | |||
| 8 | 11 | type ProxyCacheInvalidator interface { | |
| 9 | 12 | InvalidateTarget(ctx context.Context, domainName string) | |
| 10 | 13 | } | |
| 14 | + | ||
| 15 | + | // ProxyDrainWaiter defines the contract for waiting until no in-flight | |
| 16 | + | // requests remain for a container. | |
| 17 | + | type ProxyDrainWaiter interface { | |
| 18 | + | WaitForNoInFlight(ctx context.Context, containerID string, timeout time.Duration) bool | |
| 19 | + | } | |
| @@ -1,6 +1,9 @@ | |||||
| 1 | package out | 1 | package out | ||
| 2 | 2 | ||||
| 3 | - | import "context" | 3 | + | import ( |
| 4 | + | "context" | |||
| 5 | + | "time" | |||
| 6 | + | ) | |||
| 4 | 7 | ||||
| 5 | // ProxyCacheInvalidator defines the contract for synchronously invalidating | 8 | // ProxyCacheInvalidator defines the contract for synchronously invalidating | ||
| 6 | // proxy target cache entries. This is used during zero-downtime deployments | 9 | // proxy target cache entries. This is used during zero-downtime deployments | ||
| @@ -8,3 +11,9 @@ import "context" | |||||
| 8 | type ProxyCacheInvalidator interface { | 11 | type ProxyCacheInvalidator interface { | ||
| 9 | InvalidateTarget(ctx context.Context, domainName string) | 12 | InvalidateTarget(ctx context.Context, domainName string) | ||
| 10 | } | 13 | } | ||
| 14 | + | ||||
| 15 | + | // ProxyDrainWaiter defines the contract for waiting until no in-flight | |||
| 16 | + | // requests remain for a container. | |||
| 17 | + | type ProxyDrainWaiter interface { | |||
| 18 | + | WaitForNoInFlight(ctx context.Context, containerID string, timeout time.Duration) bool | |||
| 19 | + | } | |||
M internal/boundaries/out/runtime.go +1 -0
| @@ -40,6 +40,7 @@ type ContainerRuntime interface { | |||
| 40 | 40 | ||
| 41 | 41 | // Health and status | |
| 42 | 42 | IsContainerRunning(ctx context.Context, containerID string) (bool, error) | |
| 43 | + | GetContainerHealthStatus(ctx context.Context, containerID string) (status string, hasHealthcheck bool, err error) | |
| 43 | 44 | GetContainerPort(ctx context.Context, containerID string, internalPort int) (int, error) | |
| 44 | 45 | ||
| 45 | 46 | // Image and port inspection | |
| @@ -40,6 +40,7 @@ type ContainerRuntime interface { | |||||
| 40 | 40 | ||||
| 41 | // Health and status | 41 | // Health and status | ||
| 42 | IsContainerRunning(ctx context.Context, containerID string) (bool, error) | 42 | IsContainerRunning(ctx context.Context, containerID string) (bool, error) | ||
| 43 | + | GetContainerHealthStatus(ctx context.Context, containerID string) (status string, hasHealthcheck bool, err error) | |||
| 43 | GetContainerPort(ctx context.Context, containerID string, internalPort int) (int, error) | 44 | GetContainerPort(ctx context.Context, containerID string, internalPort int) (int, error) | ||
| 44 | 45 | ||||
| 45 | // Image and port inspection | 46 | // Image and port inspection | ||
M internal/usecase/container/service.go +113 -14
| @@ -36,7 +36,11 @@ type Config struct { | |||
| 36 | 36 | NetworkGroups map[string][]string | |
| 37 | 37 | Attachments map[string][]string | |
| 38 | 38 | ReadinessDelay time.Duration // Delay after container starts before considering it ready | |
| 39 | + | ReadinessMode string // Readiness strategy: auto, docker-health, delay | |
| 40 | + | HealthTimeout time.Duration // Max wait for health-based readiness | |
| 39 | 41 | DrainDelay time.Duration // Grace period after cache invalidation before stopping old container | |
| 42 | + | DrainMode string // Drain strategy: auto, inflight, delay | |
| 43 | + | DrainTimeout time.Duration // Max wait for in-flight requests to drain | |
| 40 | 44 | } | |
| 41 | 45 | ||
| 42 | 46 | const ( | |
| @@ -58,6 +62,7 @@ type Service struct { | |||
| 58 | 62 | eventBus out.EventPublisher | |
| 59 | 63 | logWriter out.ContainerLogWriter | |
| 60 | 64 | cacheInvalidator out.ProxyCacheInvalidator | |
| 65 | + | drainWaiter out.ProxyDrainWaiter | |
| 61 | 66 | config Config | |
| 62 | 67 | containers map[string]*domain.Container | |
| 63 | 68 | attachments map[string][]string // ownerDomain → []containerIDs | |
| @@ -138,6 +143,14 @@ func (s *Service) SetProxyCacheInvalidator(inv out.ProxyCacheInvalidator) { | |||
| 138 | 143 | s.mu.Unlock() | |
| 139 | 144 | } | |
| 140 | 145 | ||
| 146 | + | // SetProxyDrainWaiter sets the proxy in-flight drain waiter used during | |
| 147 | + | // zero-downtime replacement before stopping the old container. | |
| 148 | + | func (s *Service) SetProxyDrainWaiter(waiter out.ProxyDrainWaiter) { | |
| 149 | + | s.mu.Lock() | |
| 150 | + | s.drainWaiter = waiter | |
| 151 | + | s.mu.Unlock() | |
| 152 | + | } | |
| 153 | + | ||
| 141 | 154 | // domainDeployMu returns a per-domain lock for serializing deploy operations. | |
| 142 | 155 | func (s *Service) domainDeployMu(domain string) *domainDeployLock { | |
| 143 | 156 | v, _ := s.deployMu.LoadOrStore(domain, newDomainDeployLock()) | |
| @@ -371,16 +384,7 @@ func (s *Service) finalizePreviousContainer(ctx context.Context, domainName stri | |||
| 371 | 384 | log := zerowrap.FromCtx(ctx) | |
| 372 | 385 | ||
| 373 | 386 | if invalidated { | |
| 374 | - | s.mu.RLock() | |
| 375 | - | drainDelay := s.config.DrainDelay | |
| 376 | - | s.mu.RUnlock() | |
| 377 | - | if drainDelay == 0 { | |
| 378 | - | drainDelay = 2 * time.Second | |
| 379 | - | } | |
| 380 | - | select { | |
| 381 | - | case <-time.After(drainDelay): | |
| 382 | - | case <-ctx.Done(): | |
| 383 | - | } | |
| 387 | + | s.waitForDrain(ctx, existing.ID) | |
| 384 | 388 | } | |
| 385 | 389 | ||
| 386 | 390 | if err := ctx.Err(); err != nil { | |
| @@ -391,6 +395,37 @@ func (s *Service) finalizePreviousContainer(ctx context.Context, domainName stri | |||
| 391 | 395 | s.cleanupOldContainer(ctx, existing, newContainerID, domainName) | |
| 392 | 396 | } | |
| 393 | 397 | ||
| 398 | + | func (s *Service) waitForDrain(ctx context.Context, oldContainerID string) { | |
| 399 | + | s.mu.RLock() | |
| 400 | + | cfg := s.config | |
| 401 | + | waiter := s.drainWaiter | |
| 402 | + | s.mu.RUnlock() | |
| 403 | + | ||
| 404 | + | drainMode := cfg.DrainMode | |
| 405 | + | if drainMode == "" { | |
| 406 | + | drainMode = "delay" | |
| 407 | + | } | |
| 408 | + | ||
| 409 | + | shouldUseInFlight := (drainMode == "inflight" || drainMode == "auto") && waiter != nil | |
| 410 | + | if shouldUseInFlight { | |
| 411 | + | timeout := cfg.DrainTimeout | |
| 412 | + | if timeout == 0 { | |
| 413 | + | timeout = 30 * time.Second | |
| 414 | + | } | |
| 415 | + | waiter.WaitForNoInFlight(ctx, oldContainerID, timeout) | |
| 416 | + | return | |
| 417 | + | } | |
| 418 | + | ||
| 419 | + | drainDelay := cfg.DrainDelay | |
| 420 | + | if drainDelay == 0 { | |
| 421 | + | drainDelay = 2 * time.Second | |
| 422 | + | } | |
| 423 | + | select { | |
| 424 | + | case <-time.After(drainDelay): | |
| 425 | + | case <-ctx.Done(): | |
| 426 | + | } | |
| 427 | + | } | |
| 428 | + | ||
| 394 | 429 | // Restart restarts a running container for the given domain. | |
| 395 | 430 | func (s *Service) Restart(ctx context.Context, domainName string, withAttachments bool) error { | |
| 396 | 431 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | |
| @@ -1844,11 +1879,7 @@ func (s *Service) findContainerByName(ctx context.Context, name string) *domain. | |||
| 1844 | 1879 | return nil | |
| 1845 | 1880 | } | |
| 1846 | 1881 | ||
| 1847 | - | // waitForReady waits for a container to be ready. | |
| 1848 | - | // Uses simple "running + delay" approach for universal compatibility. | |
| 1849 | 1882 | func (s *Service) waitForReady(ctx context.Context, containerID string) error { | |
| 1850 | - | log := zerowrap.FromCtx(ctx) | |
| 1851 | - | ||
| 1852 | 1883 | if err := s.pollContainerRunning(ctx, containerID); err != nil { | |
| 1853 | 1884 | return err | |
| 1854 | 1885 | } | |
| @@ -1857,6 +1888,43 @@ func (s *Service) waitForReady(ctx context.Context, containerID string) error { | |||
| 1857 | 1888 | cfg := s.config | |
| 1858 | 1889 | s.mu.RUnlock() | |
| 1859 | 1890 | ||
| 1891 | + | readinessMode := cfg.ReadinessMode | |
| 1892 | + | if readinessMode == "" { | |
| 1893 | + | readinessMode = "delay" | |
| 1894 | + | } | |
| 1895 | + | ||
| 1896 | + | switch readinessMode { | |
| 1897 | + | case "delay": | |
| 1898 | + | return s.waitForReadyByDelay(ctx, containerID) | |
| 1899 | + | case "docker-health": | |
| 1900 | + | _, hasHealthcheck, err := s.runtime.GetContainerHealthStatus(ctx, containerID) | |
| 1901 | + | if err != nil { | |
| 1902 | + | return err | |
| 1903 | + | } | |
| 1904 | + | if !hasHealthcheck { | |
| 1905 | + | return errors.New("no healthcheck detected") | |
| 1906 | + | } | |
| 1907 | + | return s.waitForHealthy(ctx, containerID, cfg.HealthTimeout) | |
| 1908 | + | default: // auto | |
| 1909 | + | _, hasHealthcheck, err := s.runtime.GetContainerHealthStatus(ctx, containerID) | |
| 1910 | + | if err != nil { | |
| 1911 | + | return err | |
| 1912 | + | } | |
| 1913 | + | if hasHealthcheck { | |
| 1914 | + | return s.waitForHealthy(ctx, containerID, cfg.HealthTimeout) | |
| 1915 | + | } | |
| 1916 | + | return s.waitForReadyByDelay(ctx, containerID) | |
| 1917 | + | } | |
| 1918 | + | } | |
| 1919 | + | ||
| 1920 | + | // waitForReadyByDelay waits using the legacy running+delay strategy. | |
| 1921 | + | func (s *Service) waitForReadyByDelay(ctx context.Context, containerID string) error { | |
| 1922 | + | log := zerowrap.FromCtx(ctx) | |
| 1923 | + | ||
| 1924 | + | s.mu.RLock() | |
| 1925 | + | cfg := s.config | |
| 1926 | + | s.mu.RUnlock() | |
| 1927 | + | ||
| 1860 | 1928 | delay := cfg.ReadinessDelay | |
| 1861 | 1929 | if delay == 0 { | |
| 1862 | 1930 | delay = 5 * time.Second | |
| @@ -1881,6 +1949,37 @@ func (s *Service) waitForReady(ctx context.Context, containerID string) error { | |||
| 1881 | 1949 | return s.waitForRecovery(ctx, containerID) | |
| 1882 | 1950 | } | |
| 1883 | 1951 | ||
| 1952 | + | func (s *Service) waitForHealthy(ctx context.Context, containerID string, timeout time.Duration) error { | |
| 1953 | + | if timeout == 0 { | |
| 1954 | + | timeout = 90 * time.Second | |
| 1955 | + | } | |
| 1956 | + | waitCtx, cancel := context.WithTimeout(ctx, timeout) | |
| 1957 | + | defer cancel() | |
| 1958 | + | ||
| 1959 | + | lastStatus := "starting" | |
| 1960 | + | for { | |
| 1961 | + | status, hasHealthcheck, err := s.runtime.GetContainerHealthStatus(waitCtx, containerID) | |
| 1962 | + | if err != nil { | |
| 1963 | + | return err | |
| 1964 | + | } | |
| 1965 | + | if !hasHealthcheck { | |
| 1966 | + | return errors.New("no healthcheck detected") | |
| 1967 | + | } | |
| 1968 | + | if status == "healthy" { | |
| 1969 | + | return nil | |
| 1970 | + | } | |
| 1971 | + | if status != "" { | |
| 1972 | + | lastStatus = status | |
| 1973 | + | } | |
| 1974 | + | ||
| 1975 | + | select { | |
| 1976 | + | case <-time.After(time.Second): | |
| 1977 | + | case <-waitCtx.Done(): | |
| 1978 | + | return fmt.Errorf("container healthcheck timeout after %s (last status: %s)", timeout, lastStatus) | |
| 1979 | + | } | |
| 1980 | + | } | |
| 1981 | + | } | |
| 1982 | + | ||
| 1884 | 1983 | // pollContainerRunning polls until the container is running (max 30 seconds). | |
| 1885 | 1984 | func (s *Service) pollContainerRunning(ctx context.Context, containerID string) error { | |
| 1886 | 1985 | for i := 0; i < 30; i++ { | |
| @@ -36,7 +36,11 @@ type Config struct { | |||||
| 36 | NetworkGroups map[string][]string | 36 | NetworkGroups map[string][]string | ||
| 37 | Attachments map[string][]string | 37 | Attachments map[string][]string | ||
| 38 | ReadinessDelay time.Duration // Delay after container starts before considering it ready | 38 | ReadinessDelay time.Duration // Delay after container starts before considering it ready | ||
| 39 | + | ReadinessMode string // Readiness strategy: auto, docker-health, delay | |||
| 40 | + | HealthTimeout time.Duration // Max wait for health-based readiness | |||
| 39 | DrainDelay time.Duration // Grace period after cache invalidation before stopping old container | 41 | DrainDelay time.Duration // Grace period after cache invalidation before stopping old container | ||
| 42 | + | DrainMode string // Drain strategy: auto, inflight, delay | |||
| 43 | + | DrainTimeout time.Duration // Max wait for in-flight requests to drain | |||
| 40 | } | 44 | } | ||
| 41 | 45 | ||||
| 42 | const ( | 46 | const ( | ||
| @@ -58,6 +62,7 @@ type Service struct { | |||||
| 58 | eventBus out.EventPublisher | 62 | eventBus out.EventPublisher | ||
| 59 | logWriter out.ContainerLogWriter | 63 | logWriter out.ContainerLogWriter | ||
| 60 | cacheInvalidator out.ProxyCacheInvalidator | 64 | cacheInvalidator out.ProxyCacheInvalidator | ||
| 65 | + | drainWaiter out.ProxyDrainWaiter | |||
| 61 | config Config | 66 | config Config | ||
| 62 | containers map[string]*domain.Container | 67 | containers map[string]*domain.Container | ||
| 63 | attachments map[string][]string // ownerDomain → []containerIDs | 68 | attachments map[string][]string // ownerDomain → []containerIDs | ||
| @@ -138,6 +143,14 @@ func (s *Service) SetProxyCacheInvalidator(inv out.ProxyCacheInvalidator) { | |||||
| 138 | s.mu.Unlock() | 143 | s.mu.Unlock() | ||
| 139 | } | 144 | } | ||
| 140 | 145 | ||||
| 146 | + | // SetProxyDrainWaiter sets the proxy in-flight drain waiter used during | |||
| 147 | + | // zero-downtime replacement before stopping the old container. | |||
| 148 | + | func (s *Service) SetProxyDrainWaiter(waiter out.ProxyDrainWaiter) { | |||
| 149 | + | s.mu.Lock() | |||
| 150 | + | s.drainWaiter = waiter | |||
| 151 | + | s.mu.Unlock() | |||
| 152 | + | } | |||
| 153 | + | ||||
| 141 | // domainDeployMu returns a per-domain lock for serializing deploy operations. | 154 | // domainDeployMu returns a per-domain lock for serializing deploy operations. | ||
| 142 | func (s *Service) domainDeployMu(domain string) *domainDeployLock { | 155 | func (s *Service) domainDeployMu(domain string) *domainDeployLock { | ||
| 143 | v, _ := s.deployMu.LoadOrStore(domain, newDomainDeployLock()) | 156 | v, _ := s.deployMu.LoadOrStore(domain, newDomainDeployLock()) | ||
| @@ -371,16 +384,7 @@ func (s *Service) finalizePreviousContainer(ctx context.Context, domainName stri | |||||
| 371 | log := zerowrap.FromCtx(ctx) | 384 | log := zerowrap.FromCtx(ctx) | ||
| 372 | 385 | ||||
| 373 | if invalidated { | 386 | if invalidated { | ||
| 374 | - | s.mu.RLock() | 387 | + | s.waitForDrain(ctx, existing.ID) |
| 375 | - | drainDelay := s.config.DrainDelay | |||
| 376 | - | s.mu.RUnlock() | |||
| 377 | - | if drainDelay == 0 { | |||
| 378 | - | drainDelay = 2 * time.Second | |||
| 379 | - | } | |||
| 380 | - | select { | |||
| 381 | - | case <-time.After(drainDelay): | |||
| 382 | - | case <-ctx.Done(): | |||
| 383 | - | } | |||
| 384 | } | 388 | } | ||
| 385 | 389 | ||||
| 386 | if err := ctx.Err(); err != nil { | 390 | if err := ctx.Err(); err != nil { | ||
| @@ -391,6 +395,37 @@ func (s *Service) finalizePreviousContainer(ctx context.Context, domainName stri | |||||
| 391 | s.cleanupOldContainer(ctx, existing, newContainerID, domainName) | 395 | s.cleanupOldContainer(ctx, existing, newContainerID, domainName) | ||
| 392 | } | 396 | } | ||
| 393 | 397 | ||||
| 398 | + | func (s *Service) waitForDrain(ctx context.Context, oldContainerID string) { | |||
| 399 | + | s.mu.RLock() | |||
| 400 | + | cfg := s.config | |||
| 401 | + | waiter := s.drainWaiter | |||
| 402 | + | s.mu.RUnlock() | |||
| 403 | + | ||||
| 404 | + | drainMode := cfg.DrainMode | |||
| 405 | + | if drainMode == "" { | |||
| 406 | + | drainMode = "delay" | |||
| 407 | + | } | |||
| 408 | + | ||||
| 409 | + | shouldUseInFlight := (drainMode == "inflight" || drainMode == "auto") && waiter != nil | |||
| 410 | + | if shouldUseInFlight { | |||
| 411 | + | timeout := cfg.DrainTimeout | |||
| 412 | + | if timeout == 0 { | |||
| 413 | + | timeout = 30 * time.Second | |||
| 414 | + | } | |||
| 415 | + | waiter.WaitForNoInFlight(ctx, oldContainerID, timeout) | |||
| 416 | + | return | |||
| 417 | + | } | |||
| 418 | + | ||||
| 419 | + | drainDelay := cfg.DrainDelay | |||
| 420 | + | if drainDelay == 0 { | |||
| 421 | + | drainDelay = 2 * time.Second | |||
| 422 | + | } | |||
| 423 | + | select { | |||
| 424 | + | case <-time.After(drainDelay): | |||
| 425 | + | case <-ctx.Done(): | |||
| 426 | + | } | |||
| 427 | + | } | |||
| 428 | + | ||||
| 394 | // Restart restarts a running container for the given domain. | 429 | // Restart restarts a running container for the given domain. | ||
| 395 | func (s *Service) Restart(ctx context.Context, domainName string, withAttachments bool) error { | 430 | func (s *Service) Restart(ctx context.Context, domainName string, withAttachments bool) error { | ||
| 396 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | 431 | ctx = zerowrap.CtxWithFields(ctx, map[string]any{ | ||
| @@ -1844,11 +1879,7 @@ func (s *Service) findContainerByName(ctx context.Context, name string) *domain. | |||||
| 1844 | return nil | 1879 | return nil | ||
| 1845 | } | 1880 | } | ||
| 1846 | 1881 | ||||
| 1847 | - | // waitForReady waits for a container to be ready. | |||
| 1848 | - | // Uses simple "running + delay" approach for universal compatibility. | |||
| 1849 | func (s *Service) waitForReady(ctx context.Context, containerID string) error { | 1882 | func (s *Service) waitForReady(ctx context.Context, containerID string) error { | ||
| 1850 | - | log := zerowrap.FromCtx(ctx) | |||
| 1851 | - | ||||
| 1852 | if err := s.pollContainerRunning(ctx, containerID); err != nil { | 1883 | if err := s.pollContainerRunning(ctx, containerID); err != nil { | ||
| 1853 | return err | 1884 | return err | ||
| 1854 | } | 1885 | } | ||
| @@ -1857,6 +1888,43 @@ func (s *Service) waitForReady(ctx context.Context, containerID string) error { | |||||
| 1857 | cfg := s.config | 1888 | cfg := s.config | ||
| 1858 | s.mu.RUnlock() | 1889 | s.mu.RUnlock() | ||
| 1859 | 1890 | ||||
| 1891 | + | readinessMode := cfg.ReadinessMode | |||
| 1892 | + | if readinessMode == "" { | |||
| 1893 | + | readinessMode = "delay" | |||
| 1894 | + | } | |||
| 1895 | + | ||||
| 1896 | + | switch readinessMode { | |||
| 1897 | + | case "delay": | |||
| 1898 | + | return s.waitForReadyByDelay(ctx, containerID) | |||
| 1899 | + | case "docker-health": | |||
| 1900 | + | _, hasHealthcheck, err := s.runtime.GetContainerHealthStatus(ctx, containerID) | |||
| 1901 | + | if err != nil { | |||
| 1902 | + | return err | |||
| 1903 | + | } | |||
| 1904 | + | if !hasHealthcheck { | |||
| 1905 | + | return errors.New("no healthcheck detected") | |||
| 1906 | + | } | |||
| 1907 | + | return s.waitForHealthy(ctx, containerID, cfg.HealthTimeout) | |||
| 1908 | + | default: // auto | |||
| 1909 | + | _, hasHealthcheck, err := s.runtime.GetContainerHealthStatus(ctx, containerID) | |||
| 1910 | + | if err != nil { | |||
| 1911 | + | return err | |||
| 1912 | + | } | |||
| 1913 | + | if hasHealthcheck { | |||
| 1914 | + | return s.waitForHealthy(ctx, containerID, cfg.HealthTimeout) | |||
| 1915 | + | } | |||
| 1916 | + | return s.waitForReadyByDelay(ctx, containerID) | |||
| 1917 | + | } | |||
| 1918 | + | } | |||
| 1919 | + | ||||
| 1920 | + | // waitForReadyByDelay waits using the legacy running+delay strategy. | |||
| 1921 | + | func (s *Service) waitForReadyByDelay(ctx context.Context, containerID string) error { | |||
| 1922 | + | log := zerowrap.FromCtx(ctx) | |||
| 1923 | + | ||||
| 1924 | + | s.mu.RLock() | |||
| 1925 | + | cfg := s.config | |||
| 1926 | + | s.mu.RUnlock() | |||
| 1927 | + | ||||
| 1860 | delay := cfg.ReadinessDelay | 1928 | delay := cfg.ReadinessDelay | ||
| 1861 | if delay == 0 { | 1929 | if delay == 0 { | ||
| 1862 | delay = 5 * time.Second | 1930 | delay = 5 * time.Second | ||
| @@ -1881,6 +1949,37 @@ func (s *Service) waitForReady(ctx context.Context, containerID string) error { | |||||
| 1881 | return s.waitForRecovery(ctx, containerID) | 1949 | return s.waitForRecovery(ctx, containerID) | ||
| 1882 | } | 1950 | } | ||
| 1883 | 1951 | ||||
| 1952 | + | func (s *Service) waitForHealthy(ctx context.Context, containerID string, timeout time.Duration) error { | |||
| 1953 | + | if timeout == 0 { | |||
| 1954 | + | timeout = 90 * time.Second | |||
| 1955 | + | } | |||
| 1956 | + | waitCtx, cancel := context.WithTimeout(ctx, timeout) | |||
| 1957 | + | defer cancel() | |||
| 1958 | + | ||||
| 1959 | + | lastStatus := "starting" | |||
| 1960 | + | for { | |||
| 1961 | + | status, hasHealthcheck, err := s.runtime.GetContainerHealthStatus(waitCtx, containerID) | |||
| 1962 | + | if err != nil { | |||
| 1963 | + | return err | |||
| 1964 | + | } | |||
| 1965 | + | if !hasHealthcheck { | |||
| 1966 | + | return errors.New("no healthcheck detected") | |||
| 1967 | + | } | |||
| 1968 | + | if status == "healthy" { | |||
| 1969 | + | return nil | |||
| 1970 | + | } | |||
| 1971 | + | if status != "" { | |||
| 1972 | + | lastStatus = status | |||
| 1973 | + | } | |||
| 1974 | + | ||||
| 1975 | + | select { | |||
| 1976 | + | case <-time.After(time.Second): | |||
| 1977 | + | case <-waitCtx.Done(): | |||
| 1978 | + | return fmt.Errorf("container healthcheck timeout after %s (last status: %s)", timeout, lastStatus) | |||
| 1979 | + | } | |||
| 1980 | + | } | |||
| 1981 | + | } | |||
| 1982 | + | ||||
| 1884 | // pollContainerRunning polls until the container is running (max 30 seconds). | 1983 | // pollContainerRunning polls until the container is running (max 30 seconds). | ||
| 1885 | func (s *Service) pollContainerRunning(ctx context.Context, containerID string) error { | 1984 | func (s *Service) pollContainerRunning(ctx context.Context, containerID string) error { | ||
| 1886 | for i := 0; i < 30; i++ { | 1985 | for i := 0; i < 30; i++ { | ||
M internal/usecase/proxy/service.go +56 -0
| @@ -56,6 +56,8 @@ type Service struct { | |||
| 56 | 56 | targets map[string]*domain.ProxyTarget | |
| 57 | 57 | mu sync.RWMutex | |
| 58 | 58 | activeConns atomic.Int64 // Atomic counter for concurrent connection limiting | |
| 59 | + | inFlight map[string]int | |
| 60 | + | inFlightMu sync.Mutex | |
| 59 | 61 | } | |
| 60 | 62 | ||
| 61 | 63 | // NewService creates a new proxy service. | |
| @@ -71,6 +73,7 @@ func NewService( | |||
| 71 | 73 | configSvc: configSvc, | |
| 72 | 74 | config: config, | |
| 73 | 75 | targets: make(map[string]*domain.ProxyTarget), | |
| 76 | + | inFlight: make(map[string]int), | |
| 74 | 77 | } | |
| 75 | 78 | } | |
| 76 | 79 | ||
| @@ -304,6 +307,37 @@ func (s *Service) InvalidateTarget(_ context.Context, domainName string) { | |||
| 304 | 307 | delete(s.targets, domainName) | |
| 305 | 308 | } | |
| 306 | 309 | ||
| 310 | + | // WaitForNoInFlight waits until no requests are currently proxied to the | |
| 311 | + | // given container, or until timeout/context cancellation. | |
| 312 | + | func (s *Service) WaitForNoInFlight(ctx context.Context, containerID string, timeout time.Duration) bool { | |
| 313 | + | if containerID == "" { | |
| 314 | + | return true | |
| 315 | + | } | |
| 316 | + | if timeout <= 0 { | |
| 317 | + | timeout = 30 * time.Second | |
| 318 | + | } | |
| 319 | + | ||
| 320 | + | deadline := time.Now().Add(timeout) | |
| 321 | + | for { | |
| 322 | + | s.inFlightMu.Lock() | |
| 323 | + | count := s.inFlight[containerID] | |
| 324 | + | s.inFlightMu.Unlock() | |
| 325 | + | if count <= 0 { | |
| 326 | + | return true | |
| 327 | + | } | |
| 328 | + | ||
| 329 | + | if time.Now().After(deadline) { | |
| 330 | + | return false | |
| 331 | + | } | |
| 332 | + | ||
| 333 | + | select { | |
| 334 | + | case <-time.After(100 * time.Millisecond): | |
| 335 | + | case <-ctx.Done(): | |
| 336 | + | return false | |
| 337 | + | } | |
| 338 | + | } | |
| 339 | + | } | |
| 340 | + | ||
| 307 | 341 | // RefreshTargets refreshes all proxy targets from container state. | |
| 308 | 342 | func (s *Service) RefreshTargets(ctx context.Context) error { | |
| 309 | 343 | s.mu.Lock() | |
| @@ -459,6 +493,8 @@ func (s *Service) proxyToTarget(w http.ResponseWriter, r *http.Request, target * | |||
| 459 | 493 | // If DNS pinning resolved hostname to IP, preserve the original hostname | |
| 460 | 494 | // for the Host header so virtual-hosted upstreams work correctly. | |
| 461 | 495 | originalHost := target.OriginalHost | |
| 496 | + | releaseInFlight := s.trackInFlight(target.ContainerID) | |
| 497 | + | defer releaseInFlight() | |
| 462 | 498 | ||
| 463 | 499 | proxy := newReverseProxy(targetURL, originalHost, | |
| 464 | 500 | func(w http.ResponseWriter, _ *http.Request, err error) { | |
| @@ -482,6 +518,26 @@ func (s *Service) proxyToTarget(w http.ResponseWriter, r *http.Request, target * | |||
| 482 | 518 | proxy.ServeHTTP(w, r) | |
| 483 | 519 | } | |
| 484 | 520 | ||
| 521 | + | func (s *Service) trackInFlight(containerID string) func() { | |
| 522 | + | if containerID == "" { | |
| 523 | + | return func() {} | |
| 524 | + | } | |
| 525 | + | ||
| 526 | + | s.inFlightMu.Lock() | |
| 527 | + | s.inFlight[containerID]++ | |
| 528 | + | s.inFlightMu.Unlock() | |
| 529 | + | ||
| 530 | + | return func() { | |
| 531 | + | s.inFlightMu.Lock() | |
| 532 | + | if s.inFlight[containerID] > 1 { | |
| 533 | + | s.inFlight[containerID]-- | |
| 534 | + | } else { | |
| 535 | + | delete(s.inFlight, containerID) | |
| 536 | + | } | |
| 537 | + | s.inFlightMu.Unlock() | |
| 538 | + | } | |
| 539 | + | } | |
| 540 | + | ||
| 485 | 541 | // proxyToRegistry forwards requests to the local registry HTTP server. | |
| 486 | 542 | // SECURITY: This uses http://localhost:{port} which is safe because the registry | |
| 487 | 543 | // runs on the same host and traffic never leaves the loopback interface. If Gordon | |
| @@ -56,6 +56,8 @@ type Service struct { | |||||
| 56 | targets map[string]*domain.ProxyTarget | 56 | targets map[string]*domain.ProxyTarget | ||
| 57 | mu sync.RWMutex | 57 | mu sync.RWMutex | ||
| 58 | activeConns atomic.Int64 // Atomic counter for concurrent connection limiting | 58 | activeConns atomic.Int64 // Atomic counter for concurrent connection limiting | ||
| 59 | + | inFlight map[string]int | |||
| 60 | + | inFlightMu sync.Mutex | |||
| 59 | } | 61 | } | ||
| 60 | 62 | ||||
| 61 | // NewService creates a new proxy service. | 63 | // NewService creates a new proxy service. | ||
| @@ -71,6 +73,7 @@ func NewService( | |||||
| 71 | configSvc: configSvc, | 73 | configSvc: configSvc, | ||
| 72 | config: config, | 74 | config: config, | ||
| 73 | targets: make(map[string]*domain.ProxyTarget), | 75 | targets: make(map[string]*domain.ProxyTarget), | ||
| 76 | + | inFlight: make(map[string]int), | |||
| 74 | } | 77 | } | ||
| 75 | } | 78 | } | ||
| 76 | 79 | ||||
| @@ -304,6 +307,37 @@ func (s *Service) InvalidateTarget(_ context.Context, domainName string) { | |||||
| 304 | delete(s.targets, domainName) | 307 | delete(s.targets, domainName) | ||
| 305 | } | 308 | } | ||
| 306 | 309 | ||||
| 310 | + | // WaitForNoInFlight waits until no requests are currently proxied to the | |||
| 311 | + | // given container, or until timeout/context cancellation. | |||
| 312 | + | func (s *Service) WaitForNoInFlight(ctx context.Context, containerID string, timeout time.Duration) bool { | |||
| 313 | + | if containerID == "" { | |||
| 314 | + | return true | |||
| 315 | + | } | |||
| 316 | + | if timeout <= 0 { | |||
| 317 | + | timeout = 30 * time.Second | |||
| 318 | + | } | |||
| 319 | + | ||||
| 320 | + | deadline := time.Now().Add(timeout) | |||
| 321 | + | for { | |||
| 322 | + | s.inFlightMu.Lock() | |||
| 323 | + | count := s.inFlight[containerID] | |||
| 324 | + | s.inFlightMu.Unlock() | |||
| 325 | + | if count <= 0 { | |||
| 326 | + | return true | |||
| 327 | + | } | |||
| 328 | + | ||||
| 329 | + | if time.Now().After(deadline) { | |||
| 330 | + | return false | |||
| 331 | + | } | |||
| 332 | + | ||||
| 333 | + | select { | |||
| 334 | + | case <-time.After(100 * time.Millisecond): | |||
| 335 | + | case <-ctx.Done(): | |||
| 336 | + | return false | |||
| 337 | + | } | |||
| 338 | + | } | |||
| 339 | + | } | |||
| 340 | + | ||||
| 307 | // RefreshTargets refreshes all proxy targets from container state. | 341 | // RefreshTargets refreshes all proxy targets from container state. | ||
| 308 | func (s *Service) RefreshTargets(ctx context.Context) error { | 342 | func (s *Service) RefreshTargets(ctx context.Context) error { | ||
| 309 | s.mu.Lock() | 343 | s.mu.Lock() | ||
| @@ -459,6 +493,8 @@ func (s *Service) proxyToTarget(w http.ResponseWriter, r *http.Request, target * | |||||
| 459 | // If DNS pinning resolved hostname to IP, preserve the original hostname | 493 | // If DNS pinning resolved hostname to IP, preserve the original hostname | ||
| 460 | // for the Host header so virtual-hosted upstreams work correctly. | 494 | // for the Host header so virtual-hosted upstreams work correctly. | ||
| 461 | originalHost := target.OriginalHost | 495 | originalHost := target.OriginalHost | ||
| 496 | + | releaseInFlight := s.trackInFlight(target.ContainerID) | |||
| 497 | + | defer releaseInFlight() | |||
| 462 | 498 | ||||
| 463 | proxy := newReverseProxy(targetURL, originalHost, | 499 | proxy := newReverseProxy(targetURL, originalHost, | ||
| 464 | func(w http.ResponseWriter, _ *http.Request, err error) { | 500 | func(w http.ResponseWriter, _ *http.Request, err error) { | ||
| @@ -482,6 +518,26 @@ func (s *Service) proxyToTarget(w http.ResponseWriter, r *http.Request, target * | |||||
| 482 | proxy.ServeHTTP(w, r) | 518 | proxy.ServeHTTP(w, r) | ||
| 483 | } | 519 | } | ||
| 484 | 520 | ||||
| 521 | + | func (s *Service) trackInFlight(containerID string) func() { | |||
| 522 | + | if containerID == "" { | |||
| 523 | + | return func() {} | |||
| 524 | + | } | |||
| 525 | + | ||||
| 526 | + | s.inFlightMu.Lock() | |||
| 527 | + | s.inFlight[containerID]++ | |||
| 528 | + | s.inFlightMu.Unlock() | |||
| 529 | + | ||||
| 530 | + | return func() { | |||
| 531 | + | s.inFlightMu.Lock() | |||
| 532 | + | if s.inFlight[containerID] > 1 { | |||
| 533 | + | s.inFlight[containerID]-- | |||
| 534 | + | } else { | |||
| 535 | + | delete(s.inFlight, containerID) | |||
| 536 | + | } | |||
| 537 | + | s.inFlightMu.Unlock() | |||
| 538 | + | } | |||
| 539 | + | } | |||
| 540 | + | ||||
| 485 | // proxyToRegistry forwards requests to the local registry HTTP server. | 541 | // proxyToRegistry forwards requests to the local registry HTTP server. | ||
| 486 | // SECURITY: This uses http://localhost:{port} which is safe because the registry | 542 | // SECURITY: This uses http://localhost:{port} which is safe because the registry | ||
| 487 | // runs on the same host and traffic never leaves the loopback interface. If Gordon | 543 | // runs on the same host and traffic never leaves the loopback interface. If Gordon | ||