diff --git a/agent/internal/agent/run.go b/agent/internal/agent/run.go index ac10ee3c..220a2982 100644 --- a/agent/internal/agent/run.go +++ b/agent/internal/agent/run.go @@ -117,13 +117,14 @@ func (a *Agent) RequestStatusReport(reason string) { } func (a *Agent) reportStatus(reason string) { + startedAt := time.Now() report := a.BuildStatusReport(true) reportedDeploymentErrorCount := len(report.DeploymentErrors) completed, active := a.SnapshotWorkStatus() serverlessTransitions := a.SnapshotServerlessTransitions() response, err := a.Client.ReportStatus(report, completed, active, serverlessTransitions) if err != nil { - log.Printf("[status] failed to report (%s): %v", reason, err) + log.Printf("[status] failed to report (%s) latency=%s: %v", reason, time.Since(startedAt).Round(time.Millisecond), err) return } a.ClearReportedDeploymentErrors(reportedDeploymentErrorCount) @@ -131,7 +132,7 @@ func (a *Agent) reportStatus(reason string) { a.AcknowledgeWorkResults(response.AcceptedWorkItemResults, response.RejectedWorkItemResults) a.LogRejectedActiveWorkItems(response.RejectedActiveWorkItems) a.AcceptLeasedWorkItems(response.WorkItems) - log.Printf("[status] reported (%s)", reason) + log.Printf("[status] reported (%s) latency=%s", reason, time.Since(startedAt).Round(time.Millisecond)) } func nextStatusReportDelay() time.Duration { diff --git a/agent/internal/container/runtime.go b/agent/internal/container/runtime.go index 912e694f..660b66b6 100644 --- a/agent/internal/container/runtime.go +++ b/agent/internal/container/runtime.go @@ -91,6 +91,47 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { logFunc("stdout", fmt.Sprintf("Created volume directory: %s", vm.HostPath)) } + args := buildPodmanRunArgs(config, image) + + logFunc("stdout", fmt.Sprintf("Starting container: %s", config.Name)) + + runCmd := exec.Command("podman", args...) + output, err := runCmd.CombinedOutput() + if err != nil { + logFunc("stderr", fmt.Sprintf("Start failed: %s", string(output))) + return nil, fmt.Errorf("failed to run container: %s: %w", string(output), err) + } + + containerID := strings.TrimSpace(string(output)) + logFunc("stdout", fmt.Sprintf("Container started: %s", containerID)) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + logFunc("stdout", "Verifying container is running...") + err = retry.WithBackoff(ctx, retry.DeployBackoff, func() (bool, error) { + running, err := IsContainerRunning(containerID) + if err != nil { + return false, err + } + return running, nil + }) + + if err != nil { + logsCmd := exec.Command("podman", "logs", "--tail", "50", containerID) + logsOutput, _ := logsCmd.CombinedOutput() + logFunc("stderr", fmt.Sprintf("Container failed to stay running. Logs:\n%s", string(logsOutput))) + return nil, fmt.Errorf("container failed to stay running after start: %w", err) + } + + logFunc("stdout", "Container verified running") + + return &DeployResult{ + ContainerID: containerID, + }, nil +} + +func buildPodmanRunArgs(config *DeployConfig, image string) []string { args := []string{ "run", "-d", "--name", config.Name, @@ -117,6 +158,12 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { if config.IPAddress != "" { args = append(args, "--network", NetworkName, "--ip", config.IPAddress) + if config.PublishLocalPorts { + for _, pm := range config.PortMappings { + portMapping := fmt.Sprintf("127.0.0.1:%d:%d", pm.HostPort, pm.ContainerPort) + args = append(args, "-p", portMapping) + } + } } else { for _, pm := range config.PortMappings { portMapping := fmt.Sprintf("%s:%d:%d", config.WireGuardIP, pm.HostPort, pm.ContainerPort) @@ -154,43 +201,7 @@ func Deploy(config *DeployConfig) (*DeployResult, error) { } else { args = append(args, image) } - - logFunc("stdout", fmt.Sprintf("Starting container: %s", config.Name)) - - runCmd := exec.Command("podman", args...) - output, err := runCmd.CombinedOutput() - if err != nil { - logFunc("stderr", fmt.Sprintf("Start failed: %s", string(output))) - return nil, fmt.Errorf("failed to run container: %s: %w", string(output), err) - } - - containerID := strings.TrimSpace(string(output)) - logFunc("stdout", fmt.Sprintf("Container started: %s", containerID)) - - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - - logFunc("stdout", "Verifying container is running...") - err = retry.WithBackoff(ctx, retry.DeployBackoff, func() (bool, error) { - running, err := IsContainerRunning(containerID) - if err != nil { - return false, err - } - return running, nil - }) - - if err != nil { - logsCmd := exec.Command("podman", "logs", "--tail", "50", containerID) - logsOutput, _ := logsCmd.CombinedOutput() - logFunc("stderr", fmt.Sprintf("Container failed to stay running. Logs:\n%s", string(logsOutput))) - return nil, fmt.Errorf("container failed to stay running after start: %w", err) - } - - logFunc("stdout", "Container verified running") - - return &DeployResult{ - ContainerID: containerID, - }, nil + return args } func Stop(containerID string) error { diff --git a/agent/internal/container/runtime_test.go b/agent/internal/container/runtime_test.go new file mode 100644 index 00000000..790d75fd --- /dev/null +++ b/agent/internal/container/runtime_test.go @@ -0,0 +1,45 @@ +package container + +import ( + "slices" + "testing" +) + +func TestBuildPodmanRunArgsPublishesLoopbackPortsWithStaticIP(t *testing.T) { + args := buildPodmanRunArgs(&DeployConfig{ + Name: "svc-dep", + Image: "docker.io/library/nginx:latest", + ServiceID: "svc", + ServiceName: "api", + DeploymentID: "dep", + IPAddress: "10.200.1.2", + PublishLocalPorts: true, + PortMappings: []PortMapping{ + {ContainerPort: 80, HostPort: 30080}, + }, + }, "docker.io/library/nginx:latest") + + for _, want := range []string{"--network", NetworkName, "--ip", "10.200.1.2", "-p", "127.0.0.1:30080:80"} { + if !slices.Contains(args, want) { + t.Fatalf("args missing %q: %+v", want, args) + } + } +} + +func TestBuildPodmanRunArgsDoesNotPublishStaticIPPortsByDefault(t *testing.T) { + args := buildPodmanRunArgs(&DeployConfig{ + Name: "svc-dep", + Image: "docker.io/library/nginx:latest", + ServiceID: "svc", + ServiceName: "api", + DeploymentID: "dep", + IPAddress: "10.200.1.2", + PortMappings: []PortMapping{ + {ContainerPort: 80, HostPort: 30080}, + }, + }, "docker.io/library/nginx:latest") + + if slices.Contains(args, "-p") { + t.Fatalf("args unexpectedly publish ports: %+v", args) + } +} diff --git a/agent/internal/container/types.go b/agent/internal/container/types.go index 53f1a03c..f0571589 100644 --- a/agent/internal/container/types.go +++ b/agent/internal/container/types.go @@ -26,21 +26,22 @@ type VolumeMount struct { type BuildLogFunc func(stream string, message string) type DeployConfig struct { - Name string - Image string - ServiceID string - ServiceName string - DeploymentID string - WireGuardIP string - IPAddress string - PortMappings []PortMapping - HealthCheck *HealthCheck - Env map[string]string - VolumeMounts []VolumeMount - StartCommand string - CPULimit *float64 - MemoryLimitMb *int - LogFunc BuildLogFunc + Name string + Image string + ServiceID string + ServiceName string + DeploymentID string + WireGuardIP string + IPAddress string + PortMappings []PortMapping + PublishLocalPorts bool + HealthCheck *HealthCheck + Env map[string]string + VolumeMounts []VolumeMount + StartCommand string + CPULimit *float64 + MemoryLimitMb *int + LogFunc BuildLogFunc } type DeployResult struct { diff --git a/agent/internal/http/client.go b/agent/internal/http/client.go index 8ec46747..22ee9dd4 100644 --- a/agent/internal/http/client.go +++ b/agent/internal/http/client.go @@ -77,6 +77,7 @@ type ExpectedContainer struct { Image string `json:"image"` IPAddress string `json:"ipAddress"` Ports []PortMapping `json:"ports"` + PublishLocalPorts bool `json:"publishLocalPorts"` Env map[string]string `json:"env"` StartCommand string `json:"startCommand"` HealthCheck *HealthCheck `json:"healthCheck"` diff --git a/agent/internal/reconcile/reconcile.go b/agent/internal/reconcile/reconcile.go index db9e159e..65635afc 100644 --- a/agent/internal/reconcile/reconcile.go +++ b/agent/internal/reconcile/reconcile.go @@ -64,20 +64,21 @@ func (r *Reconciler) Deploy(exp agenthttp.ExpectedContainer) error { } _, err := container.Deploy(&container.DeployConfig{ - Name: exp.Name, - Image: exp.Image, - ServiceID: exp.ServiceID, - ServiceName: exp.ServiceName, - DeploymentID: exp.DeploymentID, - IPAddress: exp.IPAddress, - PortMappings: portMappings, - HealthCheck: healthCheck, - Env: decryptedEnv, - VolumeMounts: volumeMounts, - StartCommand: exp.StartCommand, - CPULimit: exp.ResourceCPULimit, - MemoryLimitMb: exp.ResourceMemoryLimitMb, - LogFunc: func(stream, message string) { log.Printf("[deploy:%s] %s", stream, message) }, + Name: exp.Name, + Image: exp.Image, + ServiceID: exp.ServiceID, + ServiceName: exp.ServiceName, + DeploymentID: exp.DeploymentID, + IPAddress: exp.IPAddress, + PortMappings: portMappings, + PublishLocalPorts: exp.PublishLocalPorts, + HealthCheck: healthCheck, + Env: decryptedEnv, + VolumeMounts: volumeMounts, + StartCommand: exp.StartCommand, + CPULimit: exp.ResourceCPULimit, + MemoryLimitMb: exp.ResourceMemoryLimitMb, + LogFunc: func(stream, message string) { log.Printf("[deploy:%s] %s", stream, message) }, }) return err diff --git a/agent/internal/serverless/gateway.go b/agent/internal/serverless/gateway.go index c7002585..9bc4a7b3 100644 --- a/agent/internal/serverless/gateway.go +++ b/agent/internal/serverless/gateway.go @@ -70,6 +70,30 @@ type activityState struct { sleepTimer *time.Timer } +type upstreamReadiness struct { + ready bool + latency time.Duration + err error +} + +type upstreamResolution struct { + ready []agenthttp.ServerlessUpstream + sleepingLocalIDs []string + waiting []upstreamWaitReason +} + +type upstreamWaitReason struct { + deploymentID string + serviceID string + containerID string + upstreamURL string + reason string + state string + health string + dialLatency time.Duration + err error +} + func NewGateway(runtime Runtime) *Gateway { return &Gateway{ runtime: runtime, @@ -267,9 +291,17 @@ func (g *Gateway) resolveUpstreams(host string) ([]agenthttp.ServerlessUpstream, } func (g *Gateway) readyUpstreams(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState) ([]agenthttp.ServerlessUpstream, []string, error) { + resolution, err := g.inspectUpstreams(route, state) + if err != nil { + return nil, nil, err + } + return resolution.ready, resolution.sleepingLocalIDs, nil +} + +func (g *Gateway) inspectUpstreams(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState) (upstreamResolution, error) { actualContainers, err := g.runtime.ListServerlessContainers() if err != nil { - return nil, nil, fmt.Errorf("failed to list local containers: %w", err) + return upstreamResolution{}, fmt.Errorf("failed to list local containers: %w", err) } expectedByDeploymentID := expectedContainersByDeploymentID(state) @@ -287,24 +319,59 @@ func (g *Gateway) readyUpstreams(route *agenthttp.ServerlessRoute, state *agenth upstreamsByURL[upstream.Url] = upstream } - sleepingLocalIDs := []string{} + resolution := upstreamResolution{} for _, deploymentID := range route.LocalDeploymentIDs { expected, ok := expectedByDeploymentID[deploymentID] if !ok { + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: deploymentID, + serviceID: route.ServiceID, + reason: "missing_expected_state", + }) continue } - actual, isRunning := actualByDeploymentID[deploymentID] - if isRunning && g.isContainerReady(actual, expected) { - if upstream, ok := localUpstream(route, expected); ok { - if checkUpstreamReady(upstream.Url) { + actual, exists := actualByDeploymentID[deploymentID] + if exists && actual.State == "running" { + ready, waitReason := g.containerReady(actual, expected) + if !ready { + resolution.waiting = append(resolution.waiting, waitReason) + continue + } + upstreams := localUpstreamCandidates(route, expected) + if len(upstreams) > 0 { + upstream, waitReasons, upstreamReady := probeLocalUpstreams(upstreams, expected, actual) + if upstreamReady { upstreamsByURL[upstream.Url] = upstream + } else { + resolution.waiting = append(resolution.waiting, waitReasons...) } + } else { + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + reason: "missing_local_upstream", + state: actual.State, + }) } continue } if expected.DesiredState == "stopped" || g.runtime.HasPendingServerlessSleep(deploymentID) { - sleepingLocalIDs = append(sleepingLocalIDs, deploymentID) + resolution.sleepingLocalIDs = append(resolution.sleepingLocalIDs, deploymentID) + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + reason: "sleeping", + state: containerState(actual, exists), + }) + continue } + resolution.waiting = append(resolution.waiting, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + reason: "container_not_running", + state: containerState(actual, exists), + }) } upstreams := make([]agenthttp.ServerlessUpstream, 0, len(upstreamsByURL)) @@ -317,7 +384,8 @@ func (g *Gateway) readyUpstreams(route *agenthttp.ServerlessRoute, state *agenth upstreams = append(upstreams, upstream) } sortUpstreams(upstreams) - return upstreams, sleepingLocalIDs, nil + resolution.ready = upstreams + return resolution, nil } func (g *Gateway) wakeLocalDeployments(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState, deploymentIDs []string) []string { @@ -375,44 +443,58 @@ func (g *Gateway) wakeLocalDeployments(route *agenthttp.ServerlessRoute, state * } func (g *Gateway) waitForReadyUpstreams(route *agenthttp.ServerlessRoute, wakeTimeoutSeconds int, startedAt time.Time, wokenDeploymentIDs []string) ([]agenthttp.ServerlessUpstream, error) { + domain := route.Domain timeout := wakeTimeout(wakeTimeoutSeconds) deadline := startedAt.Add(timeout) + attempt := 0 for { + attempt++ state := g.runtime.ExpectedState() - ready, _, err := g.readyUpstreams(route, state) + currentRoute := findRoute(state, domain) + if currentRoute == nil { + log.Printf( + "[serverless-gateway] wake stale host=%s reason=route_missing attempts=%d latency=%s", + domain, + attempt, + roundDuration(time.Since(startedAt)), + ) + return nil, fmt.Errorf("serverless route changed during wake") + } + route = currentRoute + currentWokenIDs := currentWakeDeploymentIDs(route, state, wokenDeploymentIDs) + resolution, err := g.inspectUpstreams(route, state) if err != nil { return nil, err } + ready := resolution.ready if len(ready) > 0 { - pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) + pendingIDs := pendingWakeDeploymentIDs(currentWokenIDs, ready) if len(pendingIDs) > 0 { go g.waitForWokenDeployments(route, route.WakeTimeoutSeconds, startedAt, pendingIDs) } + logWakeReady(route, ready, attempt, startedAt) + return ready, nil + } + if len(currentWokenIDs) == 0 { log.Printf( - "[serverless-gateway] wake ready host=%s upstreams=%d latency=%s", - route.Domain, - len(ready), + "[serverless-gateway] wake stale host=%s reason=deployments_replaced attempts=%d latency=%s", + domain, + attempt, roundDuration(time.Since(startedAt)), ) - return ready, nil + return nil, fmt.Errorf("serverless route changed during wake") } + waitSummary := summarizeWaitReasons(filterWaitReasons(resolution.waiting, currentWokenIDs)) if time.Now().After(deadline) { - pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) - g.queueWakeTimeouts(route, pendingIDs, startedAt) - if len(ready) > 0 { - log.Printf( - "[serverless-gateway] wake partially ready host=%s upstreams=%d latency=%s", - route.Domain, - len(ready), - roundDuration(time.Since(startedAt)), - ) - return ready, nil - } + pendingIDs := pendingWakeDeploymentIDs(currentWokenIDs, ready) + g.queueWakeTimeouts(route, pendingIDs, startedAt, waitSummary) log.Printf( - "[serverless-gateway] wake timed out host=%s latency=%s", + "[serverless-gateway] wake timed out host=%s attempts=%d latency=%s last_wait=%q", route.Domain, + attempt, roundDuration(time.Since(startedAt)), + waitSummary, ) return nil, fmt.Errorf("timed out waiting for local serverless wake") } @@ -421,28 +503,49 @@ func (g *Gateway) waitForReadyUpstreams(route *agenthttp.ServerlessRoute, wakeTi } func (g *Gateway) waitForWokenDeployments(route *agenthttp.ServerlessRoute, wakeTimeoutSeconds int, startedAt time.Time, wokenDeploymentIDs []string) { + domain := route.Domain timeout := wakeTimeout(wakeTimeoutSeconds) deadline := startedAt.Add(timeout) + attempt := 0 for { + attempt++ state := g.runtime.ExpectedState() - ready, _, err := g.readyUpstreams(route, state) - if err != nil { - log.Printf("[serverless-gateway] wake monitor failed host=%s error=%v", route.Domain, err) + currentRoute := findRoute(state, domain) + if currentRoute == nil { + log.Printf( + "[serverless-gateway] wake monitor stale host=%s reason=route_missing attempts=%d latency=%s", + domain, + attempt, + roundDuration(time.Since(startedAt)), + ) return } - pendingIDs := pendingWakeDeploymentIDs(wokenDeploymentIDs, ready) - if len(pendingIDs) == 0 { + route = currentRoute + currentWokenIDs := currentWakeDeploymentIDs(route, state, wokenDeploymentIDs) + if len(currentWokenIDs) == 0 { log.Printf( - "[serverless-gateway] wake ready host=%s deployments=%d latency=%s", - route.Domain, - len(wokenDeploymentIDs), + "[serverless-gateway] wake monitor stale host=%s reason=deployments_replaced attempts=%d latency=%s", + domain, + attempt, roundDuration(time.Since(startedAt)), ) return } + resolution, err := g.inspectUpstreams(route, state) + if err != nil { + log.Printf("[serverless-gateway] wake monitor failed host=%s error=%v", route.Domain, err) + return + } + ready := resolution.ready + pendingIDs := pendingWakeDeploymentIDs(currentWokenIDs, ready) + if len(pendingIDs) == 0 { + logWakeReadyDeployments(route, ready, len(currentWokenIDs), attempt, startedAt) + return + } + waitSummary := summarizeWaitReasons(filterWaitReasons(resolution.waiting, pendingIDs)) if time.Now().After(deadline) { - g.queueWakeTimeouts(route, pendingIDs, startedAt) + g.queueWakeTimeouts(route, pendingIDs, startedAt, waitSummary) return } time.Sleep(wakePollInterval) @@ -466,15 +569,144 @@ func pendingWakeDeploymentIDs(wokenDeploymentIDs []string, ready []agenthttp.Ser return pendingIDs } -func (g *Gateway) queueWakeTimeouts(route *agenthttp.ServerlessRoute, deploymentIDs []string, startedAt time.Time) { +func filterWaitReasons(reasons []upstreamWaitReason, deploymentIDs []string) []upstreamWaitReason { + if len(reasons) == 0 || len(deploymentIDs) == 0 { + return nil + } + pending := map[string]struct{}{} + for _, deploymentID := range deploymentIDs { + pending[deploymentID] = struct{}{} + } + filtered := []upstreamWaitReason{} + for _, reason := range reasons { + if _, ok := pending[reason.deploymentID]; ok { + filtered = append(filtered, reason) + } + } + return filtered +} + +func currentWakeDeploymentIDs(route *agenthttp.ServerlessRoute, state *agenthttp.ExpectedState, deploymentIDs []string) []string { + if route == nil || len(deploymentIDs) == 0 { + return nil + } + + expectedByDeploymentID := expectedContainersByDeploymentID(state) + routeIDs := map[string]struct{}{} + for _, deploymentID := range route.LocalDeploymentIDs { + routeIDs[deploymentID] = struct{}{} + } + + currentIDs := []string{} + for _, deploymentID := range deploymentIDs { + if _, ok := routeIDs[deploymentID]; !ok { + continue + } + if _, ok := expectedByDeploymentID[deploymentID]; !ok { + continue + } + currentIDs = append(currentIDs, deploymentID) + } + return currentIDs +} + +func logWakeReady(route *agenthttp.ServerlessRoute, ready []agenthttp.ServerlessUpstream, attempt int, startedAt time.Time) { + fields := []string{ + fmt.Sprintf("host=%s", route.Domain), + fmt.Sprintf("upstreams=%d", len(ready)), + } + if localUpstreams := formatLocalUpstreamURLs(ready); localUpstreams != "" { + fields = append(fields, fmt.Sprintf("local_upstreams=%s", localUpstreams)) + } + fields = append(fields, + fmt.Sprintf("attempts=%d", attempt), + fmt.Sprintf("latency=%s", roundDuration(time.Since(startedAt))), + ) + log.Printf("[serverless-gateway] wake ready %s", strings.Join(fields, " ")) +} + +func logWakeReadyDeployments(route *agenthttp.ServerlessRoute, ready []agenthttp.ServerlessUpstream, deployments int, attempt int, startedAt time.Time) { + fields := []string{ + fmt.Sprintf("host=%s", route.Domain), + fmt.Sprintf("deployments=%d", deployments), + } + if localUpstreams := formatLocalUpstreamURLs(ready); localUpstreams != "" { + fields = append(fields, fmt.Sprintf("local_upstreams=%s", localUpstreams)) + } + fields = append(fields, + fmt.Sprintf("attempts=%d", attempt), + fmt.Sprintf("latency=%s", roundDuration(time.Since(startedAt))), + ) + log.Printf("[serverless-gateway] wake ready %s", strings.Join(fields, " ")) +} + +func formatLocalUpstreamURLs(upstreams []agenthttp.ServerlessUpstream) string { + urls := []string{} + for _, upstream := range upstreams { + if upstream.Local { + urls = append(urls, upstream.Url) + } + } + sort.Strings(urls) + return strings.Join(urls, ",") +} + +func summarizeWaitReasons(reasons []upstreamWaitReason) string { + return formatWaitReasons(reasons, true) +} + +func formatWaitReasons(reasons []upstreamWaitReason, includeDialLatency bool) string { + if len(reasons) == 0 { + return "" + } + parts := make([]string, 0, len(reasons)) + for _, reason := range reasons { + part := fmt.Sprintf("deployment=%s reason=%s", reason.deploymentID, reason.reason) + if reason.upstreamURL != "" { + part += fmt.Sprintf(" upstream=%s", reason.upstreamURL) + } + if reason.state != "" { + part += fmt.Sprintf(" state=%s", reason.state) + } + if reason.health != "" { + part += fmt.Sprintf(" health=%s", reason.health) + } + if includeDialLatency && reason.dialLatency > 0 { + part += fmt.Sprintf(" dial_latency=%s", roundDuration(reason.dialLatency)) + } + if reason.err != nil { + part += fmt.Sprintf(" error=%q", compactError(reason.err)) + } + parts = append(parts, part) + } + sort.Strings(parts) + return strings.Join(parts, "; ") +} + +func compactError(err error) string { + if err == nil { + return "" + } + message := strings.TrimSpace(err.Error()) + if len(message) <= 240 { + return message + } + return message[:240] + "..." +} + +func (g *Gateway) queueWakeTimeouts(route *agenthttp.ServerlessRoute, deploymentIDs []string, startedAt time.Time, waitSummary string) { for _, deploymentID := range deploymentIDs { errMessage := fmt.Sprintf("timed out waiting %s for local serverless wake", roundDuration(time.Since(startedAt))) + if waitSummary != "" { + errMessage = fmt.Sprintf("%s; last wait: %s", errMessage, waitSummary) + } log.Printf( - "[serverless-gateway] wake timed out host=%s deployment=%s service=%s latency=%s", + "[serverless-gateway] wake timed out host=%s deployment=%s service=%s latency=%s last_wait=%q", route.Domain, deploymentID, route.ServiceID, roundDuration(time.Since(startedAt)), + waitSummary, ) g.runtime.QueueServerlessTransition(agenthttp.ServerlessTransition{ Type: "wake_failed", @@ -484,24 +716,47 @@ func (g *Gateway) queueWakeTimeouts(route *agenthttp.ServerlessRoute, deployment } } -func (g *Gateway) isContainerReady(actual container.Container, expected agenthttp.ExpectedContainer) bool { +func (g *Gateway) containerReady(actual container.Container, expected agenthttp.ExpectedContainer) (bool, upstreamWaitReason) { if actual.State != "running" { - return false + return false, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + reason: "container_not_running", + state: actual.State, + } } if expected.HealthCheck == nil || expected.HealthCheck.Cmd == "" { - return true + return true, upstreamWaitReason{} } health := g.runtime.GetServerlessContainerHealth(actual.ID) - return health == "healthy" || health == "none" + if health == "healthy" || health == "none" { + return true, upstreamWaitReason{} + } + return false, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + reason: "health_not_ready", + state: actual.State, + health: health, + } } -func tcpUpstreamReady(address string) bool { +func tcpUpstreamReady(address string) upstreamReadiness { + startedAt := time.Now() conn, err := net.DialTimeout("tcp", address, upstreamDialTimeout) if err != nil { - return false + return upstreamReadiness{ + latency: time.Since(startedAt), + err: err, + } } conn.Close() - return true + return upstreamReadiness{ + ready: true, + latency: time.Since(startedAt), + } } func (g *Gateway) cachedUpstreams(host string) ([]agenthttp.ServerlessUpstream, bool) { @@ -725,7 +980,8 @@ func (g *Gateway) sleepService(serviceID string) { ) continue } - if expected.DesiredState != "stopped" && !g.isContainerReady(actual, expected) { + containerReady, _ := g.containerReady(actual, expected) + if expected.DesiredState != "stopped" && !containerReady { log.Printf( "[serverless-gateway] sleep skipped service=%s deployment=%s reason=container_not_ready", serviceID, @@ -888,25 +1144,101 @@ func actualContainersByDeploymentID(containers []container.Container) map[string return containersByDeploymentID } -func localUpstream(route *agenthttp.ServerlessRoute, expected agenthttp.ExpectedContainer) (agenthttp.ServerlessUpstream, bool) { +func containerState(actual container.Container, exists bool) string { + if !exists { + return "missing" + } + if actual.State == "" { + return "unknown" + } + return actual.State +} + +func localUpstreamCandidates(route *agenthttp.ServerlessRoute, expected agenthttp.ExpectedContainer) []agenthttp.ServerlessUpstream { + upstreams := []agenthttp.ServerlessUpstream{} + if expected.PublishLocalPorts { + upstreams = append(upstreams, localLoopbackUpstreams(route, expected)...) + } if expected.IPAddress != "" { - return agenthttp.ServerlessUpstream{ + upstreams = append(upstreams, agenthttp.ServerlessUpstream{ DeploymentID: expected.DeploymentID, Url: fmt.Sprintf("%s:%d", expected.IPAddress, route.Port), Local: true, - }, true + }) } + if expected.IPAddress == "" && !expected.PublishLocalPorts { + upstreams = append(upstreams, localLoopbackUpstreams(route, expected)...) + } + return upstreams +} +func localLoopbackUpstreams(route *agenthttp.ServerlessRoute, expected agenthttp.ExpectedContainer) []agenthttp.ServerlessUpstream { + upstreams := []agenthttp.ServerlessUpstream{} for _, port := range expected.Ports { if port.ContainerPort == route.Port && port.HostPort > 0 { - return agenthttp.ServerlessUpstream{ + upstreams = append(upstreams, agenthttp.ServerlessUpstream{ DeploymentID: expected.DeploymentID, Url: fmt.Sprintf("127.0.0.1:%d", port.HostPort), Local: true, - }, true + }) } } - return agenthttp.ServerlessUpstream{}, false + return upstreams +} + +type upstreamProbeResult struct { + upstream agenthttp.ServerlessUpstream + readiness upstreamReadiness +} + +func probeLocalUpstreams(upstreams []agenthttp.ServerlessUpstream, expected agenthttp.ExpectedContainer, actual container.Container) (agenthttp.ServerlessUpstream, []upstreamWaitReason, bool) { + if len(upstreams) == 0 { + return agenthttp.ServerlessUpstream{}, nil, false + } + + check := checkUpstreamReady + results := make(chan upstreamProbeResult, len(upstreams)) + for _, upstream := range upstreams { + upstream := upstream + go func() { + results <- upstreamProbeResult{ + upstream: upstream, + readiness: check(upstream.Url), + } + }() + } + + waitReasons := make([]upstreamWaitReason, 0, len(upstreams)) + for range upstreams { + result := <-results + if result.readiness.ready { + return result.upstream, nil, true + } + waitReasons = append(waitReasons, upstreamWaitReason{ + deploymentID: expected.DeploymentID, + serviceID: expected.ServiceID, + containerID: actual.ID, + upstreamURL: result.upstream.Url, + reason: "upstream_unreachable", + state: actual.State, + dialLatency: result.readiness.latency, + err: result.readiness.err, + }) + } + sortWaitReasons(waitReasons) + return agenthttp.ServerlessUpstream{}, waitReasons, false +} + +func sortWaitReasons(reasons []upstreamWaitReason) { + sort.Slice(reasons, func(i, j int) bool { + if reasons[i].deploymentID != reasons[j].deploymentID { + return reasons[i].deploymentID < reasons[j].deploymentID + } + if reasons[i].reason != reasons[j].reason { + return reasons[i].reason < reasons[j].reason + } + return reasons[i].upstreamURL < reasons[j].upstreamURL + }) } func hasRunningLocalDeployment(deploymentIDs []string, actualByDeploymentID map[string]container.Container) bool { diff --git a/agent/internal/serverless/gateway_test.go b/agent/internal/serverless/gateway_test.go index ad1e3dec..45dc0e12 100644 --- a/agent/internal/serverless/gateway_test.go +++ b/agent/internal/serverless/gateway_test.go @@ -1,8 +1,12 @@ package serverless import ( + "bytes" "context" "errors" + "log" + "slices" + "strings" "sync" "testing" "time" @@ -12,7 +16,7 @@ import ( ) func init() { - checkUpstreamReady = func(string) bool { return true } + checkUpstreamReady = func(string) upstreamReadiness { return upstreamReadiness{ready: true} } } type fakeRuntime struct { @@ -130,6 +134,36 @@ func (f *fakeRuntime) snapshotContainers() []container.Container { return append([]container.Container(nil), f.containers...) } +func (f *fakeRuntime) setState(state *agenthttp.ExpectedState, containers []container.Container) { + f.mu.Lock() + defer f.mu.Unlock() + f.state = state + f.containers = append([]container.Container(nil), containers...) +} + +func receiveProbe(t *testing.T, probes <-chan string) string { + t.Helper() + select { + case probe := <-probes: + return probe + case <-time.After(100 * time.Millisecond): + t.Fatal("timed out waiting for readiness probe") + return "" + } +} + +func assertProbeSet(t *testing.T, got []string, want ...string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("probes = %+v, want %+v", got, want) + } + for _, probe := range want { + if !slices.Contains(got, probe) { + t.Fatalf("probes = %+v, want %+v", got, want) + } + } +} + func TestGetUpstreamsReturnsWhenFollowerContextIsCancelled(t *testing.T) { state := testExpectedState("stopped") g := NewGateway(&fakeRuntime{state: state}) @@ -241,6 +275,164 @@ func TestSleepingLocalDeploymentWakesAndReturnsLocalUpstream(t *testing.T) { } } +func TestSleepingLocalDeploymentPrefersPublishedLoopbackUpstream(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + releaseStatic := make(chan struct{}) + checkUpstreamReady = func(address string) upstreamReadiness { + if address == "127.0.0.1:31000" { + return upstreamReadiness{ready: true} + } + <-releaseStatic + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + close(releaseStatic) + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("stopped") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 3000, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{state: state} + gateway := NewGateway(runtime) + + upstreams, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err != nil { + t.Fatalf("getUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "127.0.0.1:31000" { + t.Fatalf("upstreams = %+v, want loopback upstream", upstreams) + } +} + +func TestSleepingLocalDeploymentFallsBackToStaticIPWhenLoopbackUnreachable(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(address string) upstreamReadiness { + if address == "10.0.0.10:3000" { + return upstreamReadiness{ready: true} + } + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("stopped") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 3000, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{state: state} + gateway := NewGateway(runtime) + + upstreams, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err != nil { + t.Fatalf("getUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "10.0.0.10:3000" { + t.Fatalf("upstreams = %+v, want static IP fallback", upstreams) + } +} + +func TestPublishedLoopbackAndStaticIPAreProbedConcurrently(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + probes := make(chan string, 2) + release := make(chan struct{}) + var releaseOnce sync.Once + checkUpstreamReady = func(address string) upstreamReadiness { + probes <- address + <-release + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + releaseOnce.Do(func() { close(release) }) + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("running") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 3000, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{ + state: state, + containers: []container.Container{ + {ID: "ctr-local", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + } + gateway := NewGateway(runtime) + + done := make(chan struct{}) + go func() { + _, _ = gateway.inspectUpstreams(&state.Serverless.Routes[0], state) + close(done) + }() + + got := []string{receiveProbe(t, probes)} + select { + case probe := <-probes: + got = append(got, probe) + case <-time.After(100 * time.Millisecond): + t.Fatalf("only saw probes %+v; want loopback and static IP to start concurrently", got) + } + assertProbeSet(t, got, "127.0.0.1:31000", "10.0.0.10:3000") + + releaseOnce.Do(func() { close(release) }) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timed out waiting for concurrent probes to finish") + } +} + +func TestPublishedLocalPortsWithoutMatchingHostPortUsesStaticIP(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(address string) upstreamReadiness { + if address != "10.0.0.10:3000" { + t.Errorf("checked %s, want only static IP", address) + return upstreamReadiness{err: errors.New("unexpected upstream")} + } + return upstreamReadiness{ready: true} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + state := testExpectedState("stopped") + state.Containers[0].PublishLocalPorts = true + state.Containers[0].Ports = []agenthttp.PortMapping{ + {ContainerPort: 8080, HostPort: 31000}, + } + state.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{state: state} + gateway := NewGateway(runtime) + + upstreams, err := gateway.getUpstreams(context.Background(), "app.example.com") + if err != nil { + t.Fatalf("getUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "10.0.0.10:3000" { + t.Fatalf("upstreams = %+v, want static IP upstream", upstreams) + } +} + func TestSleepingLocalDeploymentStartsExistingStoppedContainer(t *testing.T) { previousPoll := wakePollInterval wakePollInterval = time.Millisecond @@ -385,6 +577,154 @@ func TestPendingSleepCanWakeBeforeExpectedStateSettles(t *testing.T) { } } +func TestBlockingWakeRefreshesRouteAfterRedeploy(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(address string) upstreamReadiness { + if address == "10.0.0.11:3000" { + return upstreamReadiness{ready: true} + } + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + oldState := testExpectedStateWithLocalDeployment("running", "dep_local", "10.0.0.10") + oldState.Serverless.Routes[0].Upstreams = nil + newState := testExpectedStateWithLocalDeployment("running", "dep_new", "10.0.0.11") + newState.Serverless.Routes[0].Upstreams = nil + + runtime := &fakeRuntime{ + state: oldState, + containers: []container.Container{ + {ID: "ctr-old", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + } + var swapped sync.Once + runtime.afterList = func() { + swapped.Do(func() { + runtime.setState(newState, []container.Container{ + {ID: "ctr-new", State: "running", DeploymentID: "dep_new", ServiceID: "svc_1"}, + }) + }) + } + gateway := NewGateway(runtime) + + upstreams, err := gateway.waitForReadyUpstreams(&oldState.Serverless.Routes[0], 5, time.Now(), []string{"dep_local"}) + if err != nil { + t.Fatalf("waitForReadyUpstreams failed: %v", err) + } + if len(upstreams) != 1 || upstreams[0].Url != "10.0.0.11:3000" || upstreams[0].DeploymentID != "dep_new" { + t.Fatalf("upstreams = %+v, want redeployed local upstream", upstreams) + } + transitions, _, _ := runtime.snapshot() + if len(transitions) != 0 { + t.Fatalf("transitions = %+v, want no stale wake_failed transition", transitions) + } +} + +func TestBlockingWakeReturnsWhenRouteDisappearsDuringRedeploy(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(string) upstreamReadiness { + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + oldState := testExpectedState("running") + oldState.Serverless.Routes[0].Upstreams = nil + newState := testExpectedState("running") + newState.Serverless.Routes = nil + runtime := &fakeRuntime{ + state: oldState, + containers: []container.Container{ + {ID: "ctr-old", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + } + var swapped sync.Once + runtime.afterList = func() { + swapped.Do(func() { + runtime.setState(newState, nil) + }) + } + gateway := NewGateway(runtime) + + done := make(chan error, 1) + go func() { + _, err := gateway.waitForReadyUpstreams(&oldState.Serverless.Routes[0], 5, time.Now(), []string{"dep_local"}) + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("waitForReadyUpstreams succeeded, want stale route error") + } + case <-time.After(200 * time.Millisecond): + t.Fatal("timed out waiting for stale route to return") + } + transitions, _, _ := runtime.snapshot() + if len(transitions) != 0 { + t.Fatalf("transitions = %+v, want no stale wake_failed transition", transitions) + } +} + +func TestWakeMonitorDropsStaleDeploymentAfterRedeploy(t *testing.T) { + previousPoll := wakePollInterval + previousReady := checkUpstreamReady + wakePollInterval = time.Millisecond + checkUpstreamReady = func(string) upstreamReadiness { + return upstreamReadiness{err: errors.New("connection refused")} + } + t.Cleanup(func() { + wakePollInterval = previousPoll + checkUpstreamReady = previousReady + }) + + oldState := testExpectedStateWithLocalDeployment("running", "dep_local", "10.0.0.10") + oldState.Serverless.Routes[0].Upstreams = nil + newState := testExpectedStateWithLocalDeployment("running", "dep_new", "10.0.0.11") + newState.Serverless.Routes[0].Upstreams = nil + runtime := &fakeRuntime{ + state: oldState, + containers: []container.Container{ + {ID: "ctr-old", State: "running", DeploymentID: "dep_local", ServiceID: "svc_1"}, + }, + } + var swapped sync.Once + runtime.afterList = func() { + swapped.Do(func() { + runtime.setState(newState, []container.Container{ + {ID: "ctr-new", State: "running", DeploymentID: "dep_new", ServiceID: "svc_1"}, + }) + }) + } + gateway := NewGateway(runtime) + + done := make(chan struct{}) + go func() { + gateway.waitForWokenDeployments(&oldState.Serverless.Routes[0], 5, time.Now(), []string{"dep_local"}) + close(done) + }() + + select { + case <-done: + case <-time.After(200 * time.Millisecond): + t.Fatal("timed out waiting for stale wake monitor to return") + } + transitions, _, _ := runtime.snapshot() + if len(transitions) != 0 { + t.Fatalf("transitions = %+v, want no stale wake_failed transition", transitions) + } +} + func TestWakeTimeoutQueuesWakeFailedTransition(t *testing.T) { previousPoll := wakePollInterval wakePollInterval = time.Millisecond @@ -431,12 +771,24 @@ func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { previousPoll := wakePollInterval previousReady := checkUpstreamReady wakePollInterval = time.Millisecond - checkUpstreamReady = func(string) bool { return false } + checkUpstreamReady = func(string) upstreamReadiness { + return upstreamReadiness{err: errors.New("connection refused")} + } t.Cleanup(func() { wakePollInterval = previousPoll checkUpstreamReady = previousReady }) + var logs bytes.Buffer + previousOutput := log.Writer() + previousFlags := log.Flags() + log.SetOutput(&logs) + log.SetFlags(0) + t.Cleanup(func() { + log.SetOutput(previousOutput) + log.SetFlags(previousFlags) + }) + state := testExpectedState("stopped") state.Serverless.Routes[0].Upstreams = nil state.Serverless.Routes[0].WakeTimeoutSeconds = 1 @@ -458,6 +810,9 @@ func TestWakeTimeoutWhenLocalPortNeverOpens(t *testing.T) { if transitions[0].Type != "wake_started" || transitions[1].Type != "wake_failed" { t.Fatalf("transitions = %+v, want wake_started then wake_failed", transitions) } + if strings.Contains(logs.String(), "wake waiting") { + t.Fatalf("log output contains verbose wake waiting logs:\n%s", logs.String()) + } } func testExpectedState(localDesiredState string) *agenthttp.ExpectedState { @@ -494,3 +849,12 @@ func testExpectedState(localDesiredState string) *agenthttp.ExpectedState { } return state } + +func testExpectedStateWithLocalDeployment(localDesiredState string, deploymentID string, ipAddress string) *agenthttp.ExpectedState { + state := testExpectedState(localDesiredState) + state.Containers[0].DeploymentID = deploymentID + state.Containers[0].Name = "svc_1-" + deploymentID + state.Containers[0].IPAddress = ipAddress + state.Serverless.Routes[0].LocalDeploymentIDs = []string{deploymentID} + return state +} diff --git a/compose.dev.yml b/compose.dev.yml index 4419b83a..070d6302 100644 --- a/compose.dev.yml +++ b/compose.dev.yml @@ -19,8 +19,11 @@ services: build: context: ./registry dockerfile: Dockerfile + environment: + REGISTRY_USERNAME: docker + REGISTRY_PASSWORD: docker ports: - - "5000:5000" + - "5002:5000" volumes: - registry-data:/var/lib/registry healthcheck: diff --git a/web/.env.example b/web/.env.example index 212cc136..24da6e3e 100644 --- a/web/.env.example +++ b/web/.env.example @@ -11,6 +11,9 @@ ENCRYPTION_KEY=your-64-character-hex-string # Public URL APP_URL=http://localhost:3000 +# Local dev origins allowed by Next.js, comma-separated (optional) +NEXT_ALLOWED_DEV_ORIGINS=192.168.1.10,dev.local + # Runtime release tag used for control-plane update checks (tip/dev disables prompts) TECHULUS_CLOUD_VERSION=dev diff --git a/web/actions/projects.ts b/web/actions/projects.ts index 1b6e9c82..828a2575 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -47,6 +47,7 @@ import { nameSchema, volumeNameSchema, } from "@/lib/schemas"; +import { MIN_SERVERLESS_SLEEP_AFTER_SECONDS } from "@/lib/service-config"; import type { PortConfig, HealthCheckConfig as ServiceHealthCheckConfig, @@ -1002,7 +1003,11 @@ export async function updateServiceSchedule( const serverlessSettingsSchema = z.object({ enabled: z.boolean(), - sleepAfterSeconds: z.number().int().min(60).max(86_400), + sleepAfterSeconds: z + .number() + .int() + .min(MIN_SERVERLESS_SLEEP_AFTER_SECONDS) + .max(86_400), wakeTimeoutSeconds: z.number().int().min(10).max(900), }); diff --git a/web/app/api/v1/agent/builds/[id]/status/route.ts b/web/app/api/v1/agent/builds/[id]/status/route.ts index da67cf2f..0e53d83e 100644 --- a/web/app/api/v1/agent/builds/[id]/status/route.ts +++ b/web/app/api/v1/agent/builds/[id]/status/route.ts @@ -241,6 +241,7 @@ export async function POST( images: [archImageUri], finalImageUri: baseImageUri, serviceId: shouldDeploy ? build.serviceId : undefined, + buildId, }); await db diff --git a/web/components/service/details/replicas-section.tsx b/web/components/service/details/replicas-section.tsx index acb024cf..82ea6a21 100644 --- a/web/components/service/details/replicas-section.tsx +++ b/web/components/service/details/replicas-section.tsx @@ -19,16 +19,16 @@ import type { ServiceWithDetails as Service, } from "@/db/types"; -type ServerInfo = Pick; +type ServerInfo = Pick; type ServerWithStatus = ServerInfo & { status: string }; const fetcher = async (url: string): Promise => { const res = await fetch(url); const servers: ServerWithStatus[] = await res.json(); - return servers.map(({ id, name, wireguardIp }) => ({ + return servers.map(({ id, name, isProxy }) => ({ id, name, - wireguardIp, + isProxy, })); }; @@ -143,6 +143,8 @@ export const ReplicasSection = memo(function ReplicasSection({ ? 1 : 0 : Object.values(localReplicas).reduce((sum, n) => sum + n, 0); + const formatServerRole = (server: ServerInfo) => + server.isProxy ? "Proxy node" : "Worker node"; if (service.stateful) { return ( @@ -216,13 +218,13 @@ export const ReplicasSection = memo(function ReplicasSection({

{server.name}

- {server.wireguardIp} + {formatServerRole(server)}

@@ -282,8 +284,8 @@ export const ReplicasSection = memo(function ReplicasSection({ >

{server.name}

-

- {server.wireguardIp} +

+ {formatServerRole(server)}

diff --git a/web/components/service/details/serverless-section.tsx b/web/components/service/details/serverless-section.tsx index 17664b35..94814241 100644 --- a/web/components/service/details/serverless-section.tsx +++ b/web/components/service/details/serverless-section.tsx @@ -9,6 +9,7 @@ import { Input } from "@/components/ui/input"; import { Item, ItemContent, ItemMedia, ItemTitle } from "@/components/ui/item"; import { Switch } from "@/components/ui/switch"; import type { ServiceWithDetails as Service } from "@/db/types"; +import { MIN_SERVERLESS_SLEEP_AFTER_SECONDS } from "@/lib/service-config"; export const ServerlessSection = memo(function ServerlessSection({ service, @@ -51,10 +52,10 @@ export const ServerlessSection = memo(function ServerlessSection({ } if ( !Number.isInteger(parsed.sleepAfterSeconds) || - parsed.sleepAfterSeconds < 60 || + parsed.sleepAfterSeconds < MIN_SERVERLESS_SLEEP_AFTER_SECONDS || parsed.sleepAfterSeconds > 86_400 ) { - return "Sleep timeout must be between 60 seconds and 24 hours"; + return `Sleep timeout must be between ${MIN_SERVERLESS_SLEEP_AFTER_SECONDS} seconds and 24 hours`; } if ( !Number.isInteger(parsed.wakeTimeoutSeconds) || @@ -121,7 +122,7 @@ export const ServerlessSection = memo(function ServerlessSection({ {isLoading ? (
- +
) : hasPublicHttp ? (
-

+

{hasMetricData ? formatCompactNumber(stats.totalRequests) : "-"} @@ -212,7 +214,7 @@ function RequestStatsPanel({

-

+

{hasMetricData ? formatRate(getAverageRequestsPerSecond(stats)) : "-"} @@ -224,7 +226,9 @@ function RequestStatsPanel({

) : ( <> -

-

+

+ - +

no public HTTP ingress

@@ -758,6 +762,7 @@ function getServiceStatus( runningDeployments: number, ): ServiceStatus { const latestRollout = service.rollouts?.[0]; + const deployments = service.deployments || []; if (service.migrationStatus) return { label: "Migrating", tone: "progress" }; if ( @@ -774,14 +779,29 @@ function getServiceStatus( } if (runningDeployments > 0) return { label: "Live", tone: "live" }; - for (const deployment of service.deployments || []) { + for (const deployment of deployments) { if (deployment.observedPhase === "failed") { return { label: "Needs attention", tone: "warning" }; } } + for (const deployment of deployments) { + if (deployment.observedPhase === "waking") { + return { label: "Waking", tone: "progress" }; + } + } + for (const deployment of deployments) { + if (isObservedStarting(deployment.observedPhase)) { + return { label: "Starting", tone: "progress" }; + } + } + for (const deployment of deployments) { + if (deployment.observedPhase === "sleeping") { + return { label: "Sleeping", tone: "sleeping" }; + } + } return { - label: service.deployments.length > 0 ? "Stopped" : "Not deployed", + label: deployments.length > 0 ? "Stopped" : "Not deployed", tone: "muted", }; } diff --git a/web/components/ui/canvas-wrapper.tsx b/web/components/ui/canvas-wrapper.tsx index e863f06f..439e7556 100644 --- a/web/components/ui/canvas-wrapper.tsx +++ b/web/components/ui/canvas-wrapper.tsx @@ -40,6 +40,12 @@ const statusColorMap: Record = { dot: "bg-slate-400", text: "text-slate-500", }, + sleeping: { + bg: "bg-cyan-500/5", + border: "border-cyan-500/30", + dot: "bg-cyan-500", + text: "text-cyan-700 dark:text-cyan-400", + }, failed: { bg: "bg-rose-500/5", border: "border-rose-500/30", @@ -79,10 +85,10 @@ export function getStatusColorFromDeployments( d.observedPhase === "waking", ); const hasFailed = deployments.some((d) => d.observedPhase === "failed"); + const hasSleeping = deployments.some((d) => d.observedPhase === "sleeping"); const hasStopped = deployments.some( (d) => d.observedPhase === "stopped" || - d.observedPhase === "sleeping" || d.runtimeDesiredState === "removed", ); const hasUnknown = deployments.some((d) => d.observedPhase === "unknown"); @@ -91,6 +97,7 @@ export function getStatusColorFromDeployments( if (hasPending) return statusColorMap.pending; if (hasFailed) return statusColorMap.failed; if (hasUnknown) return statusColorMap.unknown; + if (hasSleeping) return statusColorMap.sleeping; if (hasStopped) return statusColorMap.stopped; return defaultColors; } diff --git a/web/lib/agent-status.ts b/web/lib/agent-status.ts index dde83bc6..dc4a362b 100644 --- a/web/lib/agent-status.ts +++ b/web/lib/agent-status.ts @@ -66,6 +66,7 @@ export function shouldAttachReportedContainer(observedPhase: ObservedPhase) { export function getStoppedContainerReportUpdate(deployment: { runtimeDesiredState: string; + observedPhase?: string; }) { if (deployment.runtimeDesiredState === "stopped") { return { @@ -74,6 +75,12 @@ export function getStoppedContainerReportUpdate(deployment: { healthStatus: null, }; } + if (deployment.observedPhase === "waking") { + return { + observedPhase: "waking" as const, + healthStatus: null, + }; + } return { observedPhase: "stopped" as const, diff --git a/web/lib/agent/expected-state.ts b/web/lib/agent/expected-state.ts index 7f68931d..74c6f4a5 100644 --- a/web/lib/agent/expected-state.ts +++ b/web/lib/agent/expected-state.ts @@ -51,6 +51,7 @@ export type ExpectedContainer = { image: string; ipAddress: string | null; ports: Array<{ containerPort: number; hostPort: number }>; + publishLocalPorts: boolean; env: Record; startCommand: string | null; healthCheck: { @@ -319,6 +320,7 @@ export function buildExpectedContainersFromRows({ containerPort: p.containerPort, hostPort: p.hostPort, })), + publishLocalPorts: isDeployedServerlessService(service), env: buildEnv(secretsByServiceId.get(dep.serviceId) ?? []), startCommand: service.startCommand || null, healthCheck: buildHealthCheck(service), diff --git a/web/lib/inngest/functions/build-workflow.ts b/web/lib/inngest/functions/build-workflow.ts index d9c1ec43..b66ebb03 100644 --- a/web/lib/inngest/functions/build-workflow.ts +++ b/web/lib/inngest/functions/build-workflow.ts @@ -1,10 +1,46 @@ import { and, eq, isNull } from "drizzle-orm"; import { db } from "@/db"; -import { builds, serviceReplicas, services } from "@/db/schema"; +import { builds, serviceReplicas, services, workQueue } from "@/db/schema"; import { deployServiceInternal } from "@/lib/deploy-service"; import { inngest } from "../client"; import { inngestEvents } from "../events"; +async function hasCompletedManifestWorkItem({ + serviceId, + buildId, + buildGroupId, +}: { + serviceId: string; + buildId?: string; + buildGroupId?: string | null; +}) { + const completedManifestItems = await db + .select({ payload: workQueue.payload }) + .from(workQueue) + .where( + and( + eq(workQueue.type, "create_manifest"), + eq(workQueue.status, "completed"), + ), + ); + + return completedManifestItems.some((item) => { + try { + const payload = JSON.parse(item.payload) as { + serviceId?: string; + buildId?: string; + buildGroupId?: string; + }; + + if (payload.serviceId !== serviceId) return false; + if (buildGroupId) return payload.buildGroupId === buildGroupId; + return !payload.buildGroupId && payload.buildId === buildId; + } catch { + return false; + } + }); +} + export const buildWorkflow = inngest.createFunction( { id: "build-workflow", @@ -42,14 +78,23 @@ export const buildWorkflow = inngest.createFunction( return { status: "failed", reason: result.data.error, buildId }; } - const manifestResult = await step.waitForEvent("wait-manifest", { - event: inngestEvents.manifestCompleted, - timeout: "10m", - if: `async.data.serviceId == "${serviceId}"`, + const manifestAlreadyCompleted = await step.run("check-existing-manifest", async () => { + return hasCompletedManifestWorkItem({ + serviceId, + buildId, + }); }); - if (!manifestResult) { - return { status: "completed_no_manifest", buildId }; + if (!manifestAlreadyCompleted) { + const manifestResult = await step.waitForEvent("wait-manifest", { + event: inngestEvents.manifestCompleted, + timeout: "10m", + if: `async.data.serviceId == "${serviceId}"`, + }); + + if (!manifestResult) { + return { status: "completed_no_manifest", buildId }; + } } const shouldDeploy = await step.run("check-auto-deploy", async () => { @@ -118,14 +163,23 @@ export const buildWorkflow = inngest.createFunction( return { status: "failed", reason: "build_failed", buildGroupId }; } - const manifestResult = await step.waitForEvent("wait-group-manifest", { - event: inngestEvents.manifestCompleted, - timeout: "10m", - if: `async.data.buildGroupId == "${buildGroupId}"`, + const manifestAlreadyCompleted = await step.run("check-existing-group-manifest", async () => { + return hasCompletedManifestWorkItem({ + serviceId, + buildGroupId, + }); }); - if (!manifestResult) { - return { status: "completed_no_manifest", buildGroupId }; + if (!manifestAlreadyCompleted) { + const manifestResult = await step.waitForEvent("wait-group-manifest", { + event: inngestEvents.manifestCompleted, + timeout: "10m", + if: `async.data.buildGroupId == "${buildGroupId}"`, + }); + + if (!manifestResult) { + return { status: "completed_no_manifest", buildGroupId }; + } } const shouldDeploy = await step.run("check-auto-deploy-group", async () => { diff --git a/web/lib/service-config.ts b/web/lib/service-config.ts index 6cf876ff..742aa2aa 100644 --- a/web/lib/service-config.ts +++ b/web/lib/service-config.ts @@ -50,6 +50,10 @@ export type ServerlessConfig = { wakeTimeoutSeconds: number; }; +export const MIN_SERVERLESS_SLEEP_AFTER_SECONDS = 120; +const DEFAULT_SERVERLESS_SLEEP_AFTER_SECONDS = 300; +const DEFAULT_SERVERLESS_WAKE_TIMEOUT_SECONDS = 300; + export type DeployedConfig = { source: SourceConfig; hostname?: string; @@ -142,11 +146,7 @@ export function buildCurrentConfig( protocol: p.protocol ?? "http", tlsPassthrough: p.tlsPassthrough ?? undefined, })), - serverless: { - enabled: service.serverlessEnabled ?? false, - sleepAfterSeconds: service.serverlessSleepAfterSeconds ?? 300, - wakeTimeoutSeconds: service.serverlessWakeTimeoutSeconds ?? 300, - }, + serverless: getCurrentServerlessConfig(service), secrets: (secrets ?? []).map((s) => ({ key: s.key, updatedAt: @@ -553,8 +553,12 @@ export function normalizeServerlessConfig( ): ServerlessConfig { return { enabled: config?.enabled ?? false, - sleepAfterSeconds: config?.sleepAfterSeconds ?? 300, - wakeTimeoutSeconds: config?.wakeTimeoutSeconds ?? 300, + sleepAfterSeconds: Math.max( + config?.sleepAfterSeconds ?? DEFAULT_SERVERLESS_SLEEP_AFTER_SECONDS, + MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + ), + wakeTimeoutSeconds: + config?.wakeTimeoutSeconds ?? DEFAULT_SERVERLESS_WAKE_TIMEOUT_SECONDS, }; } @@ -565,8 +569,14 @@ export function getCurrentServerlessConfig(service: { }): ServerlessConfig { return { enabled: service.serverlessEnabled ?? false, - sleepAfterSeconds: service.serverlessSleepAfterSeconds ?? 300, - wakeTimeoutSeconds: service.serverlessWakeTimeoutSeconds ?? 300, + sleepAfterSeconds: Math.max( + service.serverlessSleepAfterSeconds ?? + DEFAULT_SERVERLESS_SLEEP_AFTER_SECONDS, + MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + ), + wakeTimeoutSeconds: + service.serverlessWakeTimeoutSeconds ?? + DEFAULT_SERVERLESS_WAKE_TIMEOUT_SECONDS, }; } diff --git a/web/next.config.ts b/web/next.config.ts index 398b0d5c..4629536c 100644 --- a/web/next.config.ts +++ b/web/next.config.ts @@ -1,7 +1,12 @@ import type { NextConfig } from "next"; +const allowedDevOrigins = process.env.NEXT_ALLOWED_DEV_ORIGINS?.split(",") + .map((origin) => origin.trim()) + .filter(Boolean); + const nextConfig: NextConfig = { output: "standalone", + ...(allowedDevOrigins?.length ? { allowedDevOrigins } : {}), }; export default nextConfig; diff --git a/web/tests/agent-status.test.ts b/web/tests/agent-status.test.ts index ea56421c..f2c976e1 100644 --- a/web/tests/agent-status.test.ts +++ b/web/tests/agent-status.test.ts @@ -48,6 +48,16 @@ describe("agent status serverless attachment", () => { observedPhase: "stopped", healthStatus: "none", }); + + expect( + getStoppedContainerReportUpdate({ + runtimeDesiredState: "running", + observedPhase: "waking", + }), + ).toEqual({ + observedPhase: "waking", + healthStatus: null, + }); }); it("restores stale stopped serverless observations from live running reports", () => { diff --git a/web/tests/canvas-status-color.test.ts b/web/tests/canvas-status-color.test.ts new file mode 100644 index 00000000..6622a105 --- /dev/null +++ b/web/tests/canvas-status-color.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import { + getStatusColor, + getStatusColorFromDeployments, +} from "@/components/ui/canvas-wrapper"; + +describe("canvas status colors", () => { + it("uses a distinct color for sleeping deployments", () => { + expect(getStatusColorFromDeployments([{ observedPhase: "sleeping" }])).toEqual( + getStatusColor("sleeping"), + ); + expect(getStatusColorFromDeployments([{ observedPhase: "sleeping" }])).not.toEqual( + getStatusColor("stopped"), + ); + }); + + it("keeps active and failed states higher priority than sleeping", () => { + expect( + getStatusColorFromDeployments([ + { observedPhase: "sleeping" }, + { observedPhase: "healthy" }, + ]), + ).toEqual(getStatusColor("running")); + expect( + getStatusColorFromDeployments([ + { observedPhase: "sleeping" }, + { observedPhase: "failed" }, + ]), + ).toEqual(getStatusColor("failed")); + }); +}); diff --git a/web/tests/expected-state.test.ts b/web/tests/expected-state.test.ts index 29c90c87..4848957d 100644 --- a/web/tests/expected-state.test.ts +++ b/web/tests/expected-state.test.ts @@ -98,6 +98,7 @@ describe("expected-state pure builders", () => { { containerPort: 80, hostPort: 80 }, { containerPort: 3000, hostPort: 30001 }, ], + publishLocalPorts: true, env: { ALPHA: "first", ZED: "last" }, volumes: [ { name: "data", containerPath: "/data" }, @@ -214,6 +215,7 @@ describe("expected-state pure builders", () => { expect(containers[0]).toMatchObject({ deploymentId: "dep_stopped", desiredState: "stopped", + publishLocalPorts: false, }); }); @@ -245,6 +247,7 @@ describe("expected-state pure builders", () => { expect(containers[0]).toMatchObject({ deploymentId: "dep_sleeping", desiredState: "stopped", + publishLocalPorts: true, }); }); @@ -277,6 +280,7 @@ describe("expected-state pure builders", () => { expect(containers[0]).toMatchObject({ deploymentId: "dep_sleeping", desiredState: "stopped", + publishLocalPorts: true, }); }); diff --git a/web/tests/service-config.test.ts b/web/tests/service-config.test.ts index 98a1c55b..95444145 100644 --- a/web/tests/service-config.test.ts +++ b/web/tests/service-config.test.ts @@ -3,7 +3,9 @@ import { type DeployedConfig, diffConfigs, getDeployedServerlessConfig, + getCurrentServerlessConfig, isDeployedServerlessService, + MIN_SERVERLESS_SLEEP_AFTER_SECONDS, } from "@/lib/service-config"; function deployedConfig( @@ -42,6 +44,33 @@ describe("service config", () => { expect(isDeployedServerlessService(service)).toBe(true); }); + it("enforces the minimum serverless sleep timeout for legacy config", () => { + expect( + getCurrentServerlessConfig({ + serverlessEnabled: true, + serverlessSleepAfterSeconds: 60, + serverlessWakeTimeoutSeconds: 120, + }), + ).toMatchObject({ + sleepAfterSeconds: MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + }); + expect( + getDeployedServerlessConfig({ + deployedConfig: JSON.stringify( + deployedConfig({ + serverless: { + enabled: true, + sleepAfterSeconds: 60, + wakeTimeoutSeconds: 120, + }, + }), + ), + }), + ).toMatchObject({ + sleepAfterSeconds: MIN_SERVERLESS_SLEEP_AFTER_SECONDS, + }); + }); + it("allows deployed stateful services to be serverless", () => { const service = { serverlessEnabled: true,