diff --git a/Dockerfile b/Dockerfile index a896948..15e4982 100644 --- a/Dockerfile +++ b/Dockerfile @@ -30,13 +30,16 @@ RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build -trimpat FROM nginx:1.27-alpine -RUN apk add --no-cache dumb-init openssl +# nginx-module-njs provides ngx_http_js_module.so, loaded by nginx.conf to run +# the desktop access-token verification (deployments/nginx/njs/desktop_auth.js). +RUN apk add --no-cache dumb-init openssl nginx-module-njs WORKDIR /app COPY --from=backend-builder /out/clawreef-server /usr/local/bin/clawreef-server COPY --from=frontend-builder /app/frontend/dist /usr/share/nginx/html COPY deployments/nginx/nginx.conf /etc/nginx/nginx.conf +COPY deployments/nginx/njs/desktop_auth.js /etc/nginx/njs/desktop_auth.js COPY deployments/container/start.sh /app/start.sh RUN chmod +x /app/start.sh \ diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index 8abb0de..8ff6b03 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -7,6 +7,7 @@ import ( "os" "os/signal" "strings" + "sync" "syscall" "time" @@ -19,6 +20,7 @@ import ( "clawreef/internal/repository" "clawreef/internal/services" "clawreef/internal/services/k8s" + "clawreef/internal/services/leader" "github.com/gin-gonic/gin" ) @@ -177,18 +179,21 @@ func main() { services.StartRuntimeAdminEventBridge(bridgeCtx, runtimeEvents, wsHub) } - // Start sync service to keep instance status in sync with K8s + // Control-plane singleton background loops. The HTTP API and the in-pod + // nginx desktop data plane run on every replica, but these loops must run + // on exactly one replica. With leader election enabled they only run on the + // elected leader and migrate on failover; with it disabled (single-replica + // deployments) they run directly. syncService := services.NewSyncService(instanceRepo, instanceRuntimeStatusService) - syncService.Start() - teamService.Start() var runtimeSchedulerCancel context.CancelFunc + var runtimeSchedulerMu sync.Mutex var runtimeScheduler *services.RuntimeScheduler if cfg.Runtime.SchedulerEnabled { k8sClient := k8s.GetClient() if k8sClient == nil || k8sClient.Clientset == nil { log.Printf("runtime scheduler disabled: k8s client is unavailable") } else { - leader := services.NewRuntimeLeaderService(k8sClient.Clientset, cfg.Runtime.Namespace, cfg.Runtime.BackendReplicaID) + runtimeLeader := services.NewRuntimeLeaderService(k8sClient.Clientset, cfg.Runtime.Namespace, cfg.Runtime.BackendReplicaID) runtimeDeployments := k8s.NewRuntimeDeploymentService(k8sClient.Clientset) runtimeSchedulerOptions := []services.RuntimeSchedulerOption{ services.WithRuntimeSchedulerWorkspaceRoot(cfg.Runtime.WorkspaceRoot), @@ -209,21 +214,65 @@ func main() { rolloutRepo, runtimeAgentClient, runtimeEvents, - leader, + runtimeLeader, runtimeDeployments, cfg.Runtime.SchedulerTick, runtimeSchedulerOptions..., ) - var schedulerCtx context.Context - schedulerCtx, runtimeSchedulerCancel = context.WithCancel(context.Background()) - runtimeScheduler.Start(schedulerCtx) - log.Printf("runtime scheduler started") + log.Printf("runtime scheduler initialized") } } else { log.Printf("runtime scheduler disabled by configuration") } runtimePoolHandler := handlers.NewRuntimePoolHandler(runtimePodRepo, bindingRepo, rolloutRepo, runtimeScheduler, runtimeEvents) + leaderCtx, leaderCancel := context.WithCancel(context.Background()) + defer leaderCancel() + + startBackground := func(ctx context.Context) { + log.Printf("Starting leader-only background loops (identity=%s)", cfg.LeaderElection.Identity) + syncService.Start() + teamService.StartBackground(ctx) + if runtimeScheduler != nil { + runtimeSchedulerMu.Lock() + if runtimeSchedulerCancel == nil { + var schedulerCtx context.Context + schedulerCtx, runtimeSchedulerCancel = context.WithCancel(ctx) + runtimeScheduler.Start(schedulerCtx) + log.Printf("runtime scheduler started") + } + runtimeSchedulerMu.Unlock() + } + } + stopBackground := func() { + log.Printf("Stopping leader-only background loops (identity=%s)", cfg.LeaderElection.Identity) + runtimeSchedulerMu.Lock() + if runtimeSchedulerCancel != nil { + runtimeSchedulerCancel() + runtimeSchedulerCancel = nil + } + runtimeSchedulerMu.Unlock() + teamService.StopBackground() + syncService.Stop() + } + + if cfg.LeaderElection.Enabled && k8s.GetClient() != nil && k8s.GetClient().Clientset != nil { + go leader.Run(leaderCtx, k8s.GetClient().Clientset, leader.Config{ + Namespace: cfg.LeaderElection.Namespace, + LeaseName: cfg.LeaderElection.LeaseName, + Identity: cfg.LeaderElection.Identity, + LeaseDuration: time.Duration(cfg.LeaderElection.LeaseDuration) * time.Second, + RenewDeadline: time.Duration(cfg.LeaderElection.RenewDeadline) * time.Second, + RetryPeriod: time.Duration(cfg.LeaderElection.RetryPeriod) * time.Second, + }, leader.Callbacks{ + OnStartedLeading: startBackground, + OnStoppedLeading: stopBackground, + }) + } else { + log.Println("Leader election disabled or K8s unavailable; running control-plane background loops directly") + startBackground(leaderCtx) + } + // Setup router r := gin.Default() @@ -540,15 +589,14 @@ func main() { log.Printf("HTTP server forced to shutdown: %v", err) } - // Stop background services - if runtimeSchedulerCancel != nil { - runtimeSchedulerCancel() - } + // Stop background services. Cancelling leaderCtx releases the lease (and, + // if we were leader, triggers stopBackground); the explicit stopBackground + // call is idempotent and covers the leader-election-disabled path. + leaderCancel() + stopBackground() if runtimeAdminEventBridgeCancel != nil { runtimeAdminEventBridgeCancel() } - syncService.Stop() - teamService.Stop() wsHub.Stop() instanceHandler.Shutdown() diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 84bc033..3ba9f2f 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -12,13 +12,29 @@ import ( // Config holds all application configuration type Config struct { - Server ServerConfig `yaml:"server"` - Database DatabaseConfig `yaml:"database"` - JWT JWTConfig `yaml:"jwt"` - Kubernetes KubernetesConfig `yaml:"kubernetes"` - Runtime RuntimePoolConfig `yaml:"runtime"` - ObjectStorage ObjectStorageConfig `yaml:"objectStorage"` - SkillScanner SkillScannerConfig `yaml:"skillScanner"` + Server ServerConfig `yaml:"server"` + Database DatabaseConfig `yaml:"database"` + JWT JWTConfig `yaml:"jwt"` + Kubernetes KubernetesConfig `yaml:"kubernetes"` + Runtime RuntimePoolConfig `yaml:"runtime"` + ObjectStorage ObjectStorageConfig `yaml:"objectStorage"` + SkillScanner SkillScannerConfig `yaml:"skillScanner"` + LeaderElection LeaderElectionConfig `yaml:"leaderElection"` +} + +// LeaderElectionConfig controls the control-plane leader election that gates +// singleton background loops (SyncService + Team event consumers + stale-task +// monitor) so they run on exactly one replica when clawmanager-app is scaled +// horizontally. The HTTP API and the in-pod nginx data plane run on every +// replica regardless. +type LeaderElectionConfig struct { + Enabled bool `yaml:"enabled"` + Namespace string `yaml:"namespace"` + LeaseName string `yaml:"leaseName"` + Identity string `yaml:"identity"` + LeaseDuration int `yaml:"leaseDuration"` // seconds + RenewDeadline int `yaml:"renewDeadline"` // seconds + RetryPeriod int `yaml:"retryPeriod"` // seconds } // ServerConfig holds server-related configuration @@ -258,6 +274,15 @@ func Load() (*Config, error) { TimeoutSeconds: 30, Enabled: strings.EqualFold(getEnv("SKILL_SCANNER_ENABLED", "false"), "true"), }, + LeaderElection: LeaderElectionConfig{ + Enabled: strings.EqualFold(getEnv("CLAWMANAGER_LEADER_ELECTION", "true"), "true"), + Namespace: getEnv("POD_NAMESPACE", "clawmanager-system"), + LeaseName: getEnv("CLAWMANAGER_LEADER_LEASE_NAME", "clawmanager-controlplane-leader"), + Identity: getEnv("POD_NAME", ""), + LeaseDuration: 15, + RenewDeadline: 10, + RetryPeriod: 2, + }, } // Try to load from k8s config file @@ -372,6 +397,13 @@ func applyEnvOverrides(config *Config) { config.Runtime.MaxGatewaysPerPod = getEnvInt("RUNTIME_MAX_GATEWAYS_PER_POD", config.Runtime.MaxGatewaysPerPod) config.Runtime.GatewayPortStart = getEnvInt("RUNTIME_GATEWAY_PORT_START", config.Runtime.GatewayPortStart) config.Runtime.GatewayPortEnd = getEnvInt("RUNTIME_GATEWAY_PORT_END", config.Runtime.GatewayPortEnd) + config.LeaderElection.Enabled = getEnvBool("CLAWMANAGER_LEADER_ELECTION", config.LeaderElection.Enabled) + config.LeaderElection.Namespace = getEnv("POD_NAMESPACE", config.LeaderElection.Namespace) + config.LeaderElection.LeaseName = getEnv("CLAWMANAGER_LEADER_LEASE_NAME", config.LeaderElection.LeaseName) + config.LeaderElection.Identity = getEnv("POD_NAME", config.LeaderElection.Identity) + config.LeaderElection.LeaseDuration = getEnvInt("CLAWMANAGER_LEADER_LEASE_DURATION", config.LeaderElection.LeaseDuration) + config.LeaderElection.RenewDeadline = getEnvInt("CLAWMANAGER_LEADER_RENEW_DEADLINE", config.LeaderElection.RenewDeadline) + config.LeaderElection.RetryPeriod = getEnvInt("CLAWMANAGER_LEADER_RETRY_PERIOD", config.LeaderElection.RetryPeriod) if endpoint := os.Getenv("OBJECT_STORAGE_ENDPOINT"); endpoint != "" { config.ObjectStorage.Endpoint = endpoint diff --git a/backend/internal/handlers/instance_handler.go b/backend/internal/handlers/instance_handler.go index 98dc1bf..7781bb6 100644 --- a/backend/internal/handlers/instance_handler.go +++ b/backend/internal/handlers/instance_handler.go @@ -30,6 +30,31 @@ const ( workspaceArchiveMaxMiBEnv = "CLAWMANAGER_WORKSPACE_ARCHIVE_MAX_MIB" ) +// desktopDirectProxyEnv toggles embedding the instance Service "host:port" into +// the access token so the edge gateway can proxy desktop traffic directly to the +// instance instead of relaying it through this control-plane process. +const desktopDirectProxyEnv = "CLAWMANAGER_DESKTOP_DIRECT_PROXY" + +// desktopDirectProxyEnabled reports whether direct desktop proxying is enabled. +func desktopDirectProxyEnabled() bool { + switch strings.ToLower(strings.TrimSpace(os.Getenv(desktopDirectProxyEnv))) { + case "1", "true", "yes", "on": + return true + default: + return false + } +} + +func desktopProxyMode(directEnabled bool, upstream string) string { + if !directEnabled { + return "control-plane" + } + if strings.TrimSpace(upstream) == "" { + return "fallback" + } + return "direct" +} + func workspaceArchiveMaxMiB() int64 { value := strings.TrimSpace(os.Getenv(workspaceArchiveMaxMiBEnv)) if value == "" { @@ -112,10 +137,11 @@ type ExternalAccessRequest struct { type CreateInstanceRequest struct { Name string `json:"name" binding:"required,min=3,max=50"` Description *string `json:"description,omitempty"` - Type string `json:"type" binding:"required,oneof=openclaw hermes"` + Type string `json:"type" binding:"required,oneof=openclaw ubuntu debian centos custom webtop hermes"` Mode string `json:"mode" binding:"omitempty,oneof=lite pro"` InstanceMode string `json:"instance_mode" binding:"omitempty,oneof=lite pro"` RuntimeType string `json:"runtime_type" binding:"omitempty,oneof=gateway desktop shell"` + DesktopStreamProfile string `json:"desktop_stream_profile,omitempty" binding:"omitempty,oneof=low standard high"` CPUCores float64 `json:"cpu_cores" binding:"required,min=0.1,max=32"` MemoryGB int `json:"memory_gb" binding:"required,min=1,max=128"` DiskGB int `json:"disk_gb" binding:"required,min=10,max=1000"` @@ -133,8 +159,9 @@ type CreateInstanceRequest struct { // UpdateInstanceRequest represents an update instance request type UpdateInstanceRequest struct { - Name *string `json:"name,omitempty" binding:"omitempty,min=3,max=50"` - Description *string `json:"description,omitempty"` + Name *string `json:"name,omitempty" binding:"omitempty,min=3,max=50"` + Description *string `json:"description,omitempty"` + DesktopStreamProfile *string `json:"desktop_stream_profile,omitempty" binding:"omitempty,oneof=low standard high"` } // ListInstancesRequest represents a list instances request @@ -225,6 +252,7 @@ func (h *InstanceHandler) CreateInstance(c *gin.Context) { Mode: req.Mode, InstanceMode: req.InstanceMode, RuntimeType: req.RuntimeType, + DesktopStreamProfile: req.DesktopStreamProfile, CPUCores: req.CPUCores, MemoryGB: req.MemoryGB, DiskGB: req.DiskGB, @@ -369,8 +397,9 @@ func (h *InstanceHandler) UpdateInstance(c *gin.Context) { } updateReq := services.UpdateInstanceRequest{ - Name: req.Name, - Description: req.Description, + Name: req.Name, + Description: req.Description, + DesktopStreamProfile: req.DesktopStreamProfile, } if err := h.instanceService.Update(id, updateReq); err != nil { @@ -845,6 +874,39 @@ func (h *InstanceHandler) GenerateAccessToken(c *gin.Context) { return } + targetPort := h.proxyService.GetTargetPortForInstance(instance) + + // When direct desktop proxying is enabled, embed the instance Service + // "host:port" into the token so the edge gateway can dial the instance + // directly. On any failure we fall back to an empty upstream, which keeps + // the request flowing through the in-process control-plane proxy. + upstream := "" + directProxyEnabled := desktopDirectProxyEnabled() + if directProxyEnabled { + if h.proxyService.IsWebtopInstanceType(instance.Type) { + resolved, resolveErr := h.proxyService.ResolveUpstreamHostPort( + c.Request.Context(), + instance.UserID, + instance.ID, + targetPort, + ) + if resolveErr != nil { + fmt.Printf("Desktop direct proxy fallback: failed to resolve upstream instance=%d user=%d type=%s target_port=%d error=%v\n", + instance.ID, instance.UserID, instance.Type, targetPort, resolveErr) + } else if strings.TrimSpace(resolved) == "" { + fmt.Printf("Desktop direct proxy fallback: resolved empty upstream instance=%d user=%d type=%s target_port=%d\n", + instance.ID, instance.UserID, instance.Type, targetPort) + } else { + upstream = resolved + fmt.Printf("Desktop direct proxy resolved: instance=%d user=%d type=%s target_port=%d upstream=%s\n", + instance.ID, instance.UserID, instance.Type, targetPort, upstream) + } + } else { + fmt.Printf("Desktop direct proxy fallback: unsupported desktop instance type instance=%d user=%d type=%s target_port=%d\n", + instance.ID, instance.UserID, instance.Type, targetPort) + } + } + // Generate access token (valid for 1 hour) maxAgeSeconds := int(time.Hour.Seconds()) token, err := h.accessService.GenerateToken( @@ -852,7 +914,8 @@ func (h *InstanceHandler) GenerateAccessToken(c *gin.Context) { instance.ID, instance.Type, accessURL, - h.proxyService.GetTargetPortForInstance(instance), + upstream, + targetPort, 1*time.Hour, ) if err != nil { @@ -874,10 +937,12 @@ func (h *InstanceHandler) GenerateAccessToken(c *gin.Context) { // Return token and URLs response := map[string]interface{}{ - "token": token.Token, - "access_url": accessURL, - "proxy_url": h.proxyService.GetProxyURLForInstance(instance, token.Token), - "expires_at": token.ExpiresAt, + "token": token.Token, + "access_url": accessURL, + "proxy_url": h.proxyService.GetProxyURLForInstance(instance, token.Token), + "expires_at": token.ExpiresAt, + "desktop_proxy_mode": desktopProxyMode(directProxyEnabled, upstream), + "desktop_upstream_present": upstream != "", } utils.Success(c, http.StatusOK, "Access token generated successfully", response) diff --git a/backend/internal/models/instance.go b/backend/internal/models/instance.go index 0de5d32..5b9f62b 100644 --- a/backend/internal/models/instance.go +++ b/backend/internal/models/instance.go @@ -25,6 +25,7 @@ type Instance struct { ImageRegistry *string `db:"image_registry" json:"image_registry,omitempty"` ImageTag *string `db:"image_tag" json:"image_tag,omitempty"` EnvironmentOverridesJSON *string `db:"environment_overrides_json" json:"-"` + DesktopStreamProfile string `db:"-" json:"desktop_stream_profile,omitempty"` StorageClass string `db:"storage_class" json:"storage_class"` MountPath string `db:"mount_path" json:"mount_path"` WorkspacePath *string `db:"workspace_path" json:"workspace_path,omitempty"` diff --git a/backend/internal/services/desktop_stream_profile.go b/backend/internal/services/desktop_stream_profile.go new file mode 100644 index 0000000..cdb01f5 --- /dev/null +++ b/backend/internal/services/desktop_stream_profile.go @@ -0,0 +1,90 @@ +package services + +import "strings" + +const ( + DesktopStreamProfileLow = "low" + DesktopStreamProfileStandard = "standard" + DesktopStreamProfileHigh = "high" + + desktopStreamProfileEnvKey = "CLAWMANAGER_DESKTOP_STREAM_PROFILE" +) + +var selkiesDesktopStreamEnvKeys = []string{ + desktopStreamProfileEnvKey, + "SELKIES_ENCODER", + "SELKIES_FRAMERATE", + "SELKIES_H264_CRF", + "SELKIES_AUDIO_ENABLED", + "SELKIES_SECOND_SCREEN", +} + +func normalizeDesktopStreamProfile(profile string) (string, bool) { + switch strings.ToLower(strings.TrimSpace(profile)) { + case "": + return "", true + case DesktopStreamProfileLow: + return DesktopStreamProfileLow, true + case DesktopStreamProfileStandard: + return DesktopStreamProfileStandard, true + case DesktopStreamProfileHigh: + return DesktopStreamProfileHigh, true + default: + return "", false + } +} + +func applyDesktopStreamProfileEnv(overrides map[string]string, profile string) map[string]string { + normalized, ok := normalizeDesktopStreamProfile(profile) + if !ok || normalized == "" { + return overrides + } + + if overrides == nil { + overrides = map[string]string{} + } + + for _, key := range selkiesDesktopStreamEnvKeys { + delete(overrides, key) + } + + overrides[desktopStreamProfileEnvKey] = normalized + overrides["SELKIES_ENCODER"] = "x264enc" + overrides["SELKIES_SECOND_SCREEN"] = "false" + overrides["SELKIES_AUDIO_ENABLED"] = "false" + + switch normalized { + case DesktopStreamProfileLow: + overrides["SELKIES_FRAMERATE"] = "30" + overrides["SELKIES_H264_CRF"] = "42" + case DesktopStreamProfileHigh: + overrides["SELKIES_FRAMERATE"] = "40" + overrides["SELKIES_H264_CRF"] = "24" + default: + overrides["SELKIES_FRAMERATE"] = "35" + overrides["SELKIES_H264_CRF"] = "34" + } + + return overrides +} + +func desktopStreamProfileFromEnv(overrides map[string]string) string { + if profile, ok := overrides[desktopStreamProfileEnvKey]; ok { + if normalized, valid := normalizeDesktopStreamProfile(profile); valid && normalized != "" { + return normalized + } + } + + framerate := strings.TrimSpace(overrides["SELKIES_FRAMERATE"]) + crf := strings.TrimSpace(overrides["SELKIES_H264_CRF"]) + switch { + case framerate == "30" && crf == "42": + return DesktopStreamProfileLow + case framerate == "40" && crf == "24": + return DesktopStreamProfileHigh + case framerate == "35" && crf == "34": + return DesktopStreamProfileStandard + default: + return "" + } +} diff --git a/backend/internal/services/instance_access_service.go b/backend/internal/services/instance_access_service.go index c58e200..230dd11 100644 --- a/backend/internal/services/instance_access_service.go +++ b/backend/internal/services/instance_access_service.go @@ -18,6 +18,7 @@ type AccessToken struct { InstanceType string `json:"instance_type"` TargetPort int32 `json:"target_port"` AccessURL string `json:"access_url"` + Upstream string `json:"upstream"` ExpiresAt time.Time `json:"expires_at"` CreatedAt time.Time `json:"created_at"` } @@ -36,6 +37,7 @@ type instanceAccessClaims struct { InstanceType string `json:"instance_type"` TargetPort int32 `json:"target_port"` AccessURL string `json:"access_url"` + Upstream string `json:"upstream"` TokenType string `json:"token_type"` jwt.RegisteredClaims } @@ -55,7 +57,7 @@ func NewInstanceAccessService() *InstanceAccessService { } // GenerateToken generates a new access token for an instance -func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceType string, accessURL string, targetPort int32, duration time.Duration) (*AccessToken, error) { +func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceType string, accessURL string, upstream string, targetPort int32, duration time.Duration) (*AccessToken, error) { now := time.Now() expiresAt := now.Add(duration) @@ -65,6 +67,7 @@ func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceTy InstanceType: instanceType, TargetPort: targetPort, AccessURL: accessURL, + Upstream: upstream, TokenType: "instance_access", RegisteredClaims: jwt.RegisteredClaims{ IssuedAt: jwt.NewNumericDate(now), @@ -85,6 +88,7 @@ func (s *InstanceAccessService) GenerateToken(userID, instanceID int, instanceTy InstanceType: instanceType, TargetPort: targetPort, AccessURL: accessURL, + Upstream: upstream, ExpiresAt: expiresAt, CreatedAt: now, } @@ -146,6 +150,7 @@ func (s *InstanceAccessService) validateSignedToken(token string) (*AccessToken, InstanceType: claims.InstanceType, TargetPort: claims.TargetPort, AccessURL: claims.AccessURL, + Upstream: claims.Upstream, ExpiresAt: expiresAt, CreatedAt: createdAt, }, nil diff --git a/backend/internal/services/instance_access_service_test.go b/backend/internal/services/instance_access_service_test.go index a13aee8..9b35794 100644 --- a/backend/internal/services/instance_access_service_test.go +++ b/backend/internal/services/instance_access_service_test.go @@ -11,7 +11,8 @@ func TestInstanceAccessServiceValidatesTokenAcrossServiceInstances(t *testing.T) issuer := NewInstanceAccessService() validator := NewInstanceAccessService() - token, err := issuer.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", 3001, 5*time.Minute) + wantUpstream := "clawreef-42-demo-svc.clawmanager-user-7.svc.cluster.local:3001" + token, err := issuer.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", wantUpstream, 3001, 5*time.Minute) if err != nil { t.Fatalf("GenerateToken() error = %v", err) } @@ -30,13 +31,16 @@ func TestInstanceAccessServiceValidatesTokenAcrossServiceInstances(t *testing.T) if validated.InstanceType != "openclaw" { t.Fatalf("validated.InstanceType = %q, want openclaw", validated.InstanceType) } + if validated.Upstream != wantUpstream { + t.Fatalf("validated.Upstream = %q, want %q", validated.Upstream, wantUpstream) + } } func TestInstanceAccessServiceRejectsExpiredSignedToken(t *testing.T) { t.Setenv("INSTANCE_ACCESS_TOKEN_SECRET", "cluster-shared-secret") service := NewInstanceAccessService() - token, err := service.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", 3001, -time.Second) + token, err := service.GenerateToken(7, 42, "openclaw", "/api/v1/instances/42/proxy/", "", 3001, -time.Second) if err != nil { t.Fatalf("GenerateToken() error = %v", err) } @@ -81,7 +85,7 @@ func TestInstanceAccessServiceStopTerminatesCleanup(t *testing.T) { // After Stop, the service should still be usable for token operations // (only the background cleanup goroutine is stopped). - token, err := service.GenerateToken(1, 1, "openclaw", "/proxy", 3001, time.Minute) + token, err := service.GenerateToken(1, 1, "openclaw", "/proxy", "", 3001, time.Minute) if err != nil { t.Fatalf("GenerateToken after Stop() error = %v", err) } diff --git a/backend/internal/services/instance_proxy_service.go b/backend/internal/services/instance_proxy_service.go index 9143255..cad6a81 100644 --- a/backend/internal/services/instance_proxy_service.go +++ b/backend/internal/services/instance_proxy_service.go @@ -659,6 +659,18 @@ func (s *InstanceProxyService) GetTargetPortForInstance(instance *models.Instanc return buildRuntimeConfig(instance.Type, instance.OSType, instance.OSVersion, instance.ImageRegistry, instance.ImageTag).Port } +// ResolveUpstreamHostPort ensures the instance Service exists and returns its +// cluster-internal "host:port" target so the edge gateway can proxy directly to +// the instance without routing pixel traffic through this control-plane process. +func (s *InstanceProxyService) ResolveUpstreamHostPort(ctx context.Context, userID, instanceID int, targetPort int32) (string, error) { + serviceInfo, err := s.getOrCreateService(ctx, userID, instanceID, targetPort) + if err != nil { + return "", fmt.Errorf("failed to resolve upstream service: %w", err) + } + + return fmt.Sprintf("%s.%s.svc.cluster.local:%d", serviceInfo.Name, serviceInfo.Namespace, serviceInfo.TargetPort), nil +} + func (s *InstanceProxyService) resolveTargetPort(instanceType string, defaultPort int32, targetPath string) int32 { if usesWebtopImage(instanceType) { if defaultPort == 0 { @@ -734,6 +746,13 @@ func (s *InstanceProxyService) shouldRewriteHTMLForProxy(instanceID int, instanc return s.shouldRewriteHTML(instanceType) } +// IsWebtopInstanceType reports whether the instance type is served by a +// Webtop/KasmVNC desktop image (and therefore eligible for direct gateway +// proxying via SUBFOLDER-prefixed paths). +func (s *InstanceProxyService) IsWebtopInstanceType(instanceType string) bool { + return usesWebtopImage(instanceType) +} + func (s *InstanceProxyService) getCachedService(key serviceCacheKey) *k8s.ServiceInfo { s.cacheMu.RLock() entry, ok := s.serviceCache[key] diff --git a/backend/internal/services/instance_service.go b/backend/internal/services/instance_service.go index c85a2bb..77eec8c 100644 --- a/backend/internal/services/instance_service.go +++ b/backend/internal/services/instance_service.go @@ -52,6 +52,9 @@ func (s *instanceService) ValidateCreateRequests(userID int, requests []CreateIn if _, err := marshalEnvironmentOverrides(environmentOverrides); err != nil { return err } + if _, ok := normalizeDesktopStreamProfile(requests[idx].DesktopStreamProfile); !ok { + return fmt.Errorf("invalid desktop stream profile") + } } quota, err := s.quotaRepo.GetByUserID(userID) @@ -136,10 +139,11 @@ func (s *instanceService) ValidateCreateRequests(userID int, requests []CreateIn type CreateInstanceRequest struct { Name string `json:"name" validate:"required,min=3,max=50"` Description *string `json:"description,omitempty"` - Type string `json:"type" validate:"required,oneof=openclaw hermes"` + Type string `json:"type" validate:"required,oneof=openclaw ubuntu debian centos custom webtop hermes"` Mode string `json:"mode" validate:"omitempty,oneof=lite pro"` InstanceMode string `json:"instance_mode" validate:"omitempty,oneof=lite pro"` RuntimeType string `json:"runtime_type" validate:"omitempty,oneof=gateway desktop shell"` + DesktopStreamProfile string `json:"desktop_stream_profile,omitempty" validate:"omitempty,oneof=low standard high"` CPUCores float64 `json:"cpu_cores" validate:"required,min=0.1,max=32"` MemoryGB int `json:"memory_gb" validate:"required,min=1,max=128"` DiskGB int `json:"disk_gb" validate:"required,min=10,max=1000"` @@ -177,8 +181,9 @@ type instanceModeLimitConfig struct { // UpdateInstanceRequest holds data for updating an instance type UpdateInstanceRequest struct { - Name *string `json:"name,omitempty" validate:"omitempty,min=3,max=50"` - Description *string `json:"description,omitempty"` + Name *string `json:"name,omitempty" validate:"omitempty,min=3,max=50"` + Description *string `json:"description,omitempty"` + DesktopStreamProfile *string `json:"desktop_stream_profile,omitempty" validate:"omitempty,oneof=low standard high"` } // InstanceStatus holds the status of an instance @@ -269,6 +274,11 @@ func (s *instanceService) Create(userID int, req CreateInstanceRequest) (*models if err != nil { return nil, err } + if profile, ok := normalizeDesktopStreamProfile(req.DesktopStreamProfile); !ok { + return nil, fmt.Errorf("invalid desktop stream profile") + } else if profile != "" { + environmentOverrides = applyDesktopStreamProfileEnv(environmentOverrides, profile) + } environmentOverridesJSON, err := marshalEnvironmentOverrides(environmentOverrides) if err != nil { return nil, err @@ -666,6 +676,7 @@ func (s *instanceService) Create(userID int, req CreateInstanceRequest) (*models // Broadcast initial creating status via WebSocket. Sync service will mark it // running only after the pod becomes Ready. + hydrateInstanceDesktopStreamProfile(instance) GetHub().BroadcastInstanceStatus(userID, instance) fmt.Printf("Instance %d status broadcast complete\n", instance.ID) @@ -722,7 +733,12 @@ func (s *instanceService) createV2Instance(ctx context.Context, userID int, req // GetByID gets an instance by ID func (s *instanceService) GetByID(id int) (*models.Instance, error) { - return s.instanceRepo.GetByID(id) + instance, err := s.instanceRepo.GetByID(id) + if err != nil { + return nil, err + } + hydrateInstanceDesktopStreamProfile(instance) + return instance, nil } // GetByUserID gets instances by user ID with pagination @@ -731,6 +747,7 @@ func (s *instanceService) GetByUserID(userID int, offset, limit int) ([]models.I if err != nil { return nil, 0, err } + hydrateInstancesDesktopStreamProfile(instances) total, err := s.instanceRepo.CountByUserID(userID) if err != nil { @@ -745,6 +762,7 @@ func (s *instanceService) GetAllInstances(offset, limit int) ([]models.Instance, if err != nil { return nil, 0, err } + hydrateInstancesDesktopStreamProfile(instances) total, err := s.instanceRepo.CountAll() if err != nil { @@ -754,6 +772,23 @@ func (s *instanceService) GetAllInstances(offset, limit int) ([]models.Instance, return instances, total, nil } +func hydrateInstancesDesktopStreamProfile(instances []models.Instance) { + for idx := range instances { + hydrateInstanceDesktopStreamProfile(&instances[idx]) + } +} + +func hydrateInstanceDesktopStreamProfile(instance *models.Instance) { + if instance == nil { + return + } + environmentOverrides, err := parseEnvironmentOverridesJSON(instance.EnvironmentOverridesJSON) + if err != nil { + return + } + instance.DesktopStreamProfile = desktopStreamProfileFromEnv(environmentOverrides) +} + // Start starts an instance func (s *instanceService) Start(instanceID int) error { ctx := context.Background() @@ -1610,6 +1645,22 @@ func (s *instanceService) Update(instanceID int, req UpdateInstanceRequest) erro if req.Description != nil { instance.Description = req.Description } + if req.DesktopStreamProfile != nil { + profile, ok := normalizeDesktopStreamProfile(*req.DesktopStreamProfile) + if !ok || profile == "" { + return fmt.Errorf("invalid desktop stream profile") + } + environmentOverrides, err := parseEnvironmentOverridesJSON(instance.EnvironmentOverridesJSON) + if err != nil { + return err + } + environmentOverrides = applyDesktopStreamProfileEnv(environmentOverrides, profile) + environmentOverridesJSON, err := marshalEnvironmentOverrides(environmentOverrides) + if err != nil { + return err + } + instance.EnvironmentOverridesJSON = environmentOverridesJSON + } instance.UpdatedAt = time.Now() diff --git a/backend/internal/services/k8s/pvc_service.go b/backend/internal/services/k8s/pvc_service.go index ca383fd..4d82496 100644 --- a/backend/internal/services/k8s/pvc_service.go +++ b/backend/internal/services/k8s/pvc_service.go @@ -119,9 +119,12 @@ func (s *PVCService) CreatePVC(ctx context.Context, userID, instanceID int, stor return nil, fmt.Errorf("failed to create PVC %s: %w", pvcName, err) } - // Wait for PVC to be bound, if not bound within timeout, create PV manually - fmt.Printf("PVC %s created, scheduling async binding monitor...\n", pvcName) - go s.monitorPVCBinding(context.Background(), namespace, pvcName, userID, instanceID, storageSizeGB, storageClass, 15*time.Second) + if usesManualHostPathFallback(storageClass) { + fmt.Printf("PVC %s created, scheduling manual hostPath binding monitor...\n", pvcName) + go s.monitorPVCBinding(context.Background(), namespace, pvcName, userID, instanceID, storageSizeGB, storageClass, 15*time.Second) + } else { + fmt.Printf("PVC %s created with dynamic storageClass %s, leaving binding to the provisioner\n", pvcName, storageClass) + } return createdPVC, nil } @@ -185,7 +188,7 @@ func (s *PVCService) CreateTeamSharedPVC(ctx context.Context, userID, teamID, st return nil, err } } - if existingPVC.Status.Phase != corev1.ClaimBound { + if existingPVC.Status.Phase != corev1.ClaimBound && usesManualHostPathFallback(storageClass) { go s.monitorTeamSharedPVCBinding(context.Background(), namespace, pvcName, userID, teamID, storageSizeGB, storageClass, 15*time.Second) } return existingPVC, nil @@ -199,10 +202,16 @@ func (s *PVCService) CreateTeamSharedPVC(ctx context.Context, userID, teamID, st return nil, err } } - go s.monitorTeamSharedPVCBinding(context.Background(), namespace, pvcName, userID, teamID, storageSizeGB, storageClass, 15*time.Second) + if usesManualHostPathFallback(storageClass) { + go s.monitorTeamSharedPVCBinding(context.Background(), namespace, pvcName, userID, teamID, storageSizeGB, storageClass, 15*time.Second) + } return createdPVC, nil } +func usesManualHostPathFallback(storageClass string) bool { + return strings.EqualFold(strings.TrimSpace(storageClass), "manual") +} + func (s *PVCService) monitorTeamSharedPVCBinding(ctx context.Context, namespace, pvcName string, userID, teamID, storageSizeGB int, storageClass string, timeout time.Duration) { if _, err := s.waitForTeamSharedPVCBinding(ctx, namespace, pvcName, userID, teamID, storageSizeGB, storageClass, timeout); err != nil { fmt.Printf("Async Team shared PVC binding monitor failed for %s: %v\n", pvcName, err) @@ -225,6 +234,10 @@ func (s *PVCService) waitForTeamSharedPVCBinding(ctx context.Context, namespace, for { select { case <-timeoutChan: + if !usesManualHostPathFallback(storageClass) { + fmt.Printf("Team shared PVC %s binding timeout with dynamic storageClass %s, leaving binding to the provisioner\n", pvcName, storageClass) + return pvc, nil + } fmt.Printf("Team shared PVC %s binding timeout, creating hostPath RWX PV manually\n", pvcName) return s.createPVForTeamSharedPVC(ctx, namespace, pvcName, userID, teamID, storageSizeGB, storageClass) case <-ticker.C: @@ -629,7 +642,11 @@ func (s *PVCService) waitForPVCBinding(ctx context.Context, namespace, pvcName s for { select { case <-timeoutChan: - // Timeout, try to create PV manually + if !usesManualHostPathFallback(storageClass) { + fmt.Printf("PVC %s binding timeout with dynamic storageClass %s, leaving binding to the provisioner\n", pvcName, storageClass) + return pvc, nil + } + // Timeout, try to create PV manually. fmt.Printf("PVC %s binding timeout, creating PV manually\n", pvcName) return s.createPVForPVC(ctx, namespace, pvcName, userID, instanceID, storageSizeGB, storageClass) case <-ticker.C: diff --git a/backend/internal/services/k8s/pvc_service_test.go b/backend/internal/services/k8s/pvc_service_test.go index 3bd6c3a..47f3140 100644 --- a/backend/internal/services/k8s/pvc_service_test.go +++ b/backend/internal/services/k8s/pvc_service_test.go @@ -6,6 +6,7 @@ import ( "path/filepath" "strings" "testing" + "time" corev1 "k8s.io/api/core/v1" storagev1 "k8s.io/api/storage/v1" @@ -456,6 +457,45 @@ func TestNodeSelectorForPVCFallsBackForNoProvisionerStorageClass(t *testing.T) { } } +func TestWaitForPVCBindingLeavesDynamicStorageClassToProvisioner(t *testing.T) { + pvc := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "clawreef-114-pvc", + Namespace: "clawmanager-user-155", + }, + Spec: corev1.PersistentVolumeClaimSpec{ + StorageClassName: stringPtr("local-path"), + }, + Status: corev1.PersistentVolumeClaimStatus{ + Phase: corev1.ClaimPending, + }, + } + service := &PVCService{ + client: &Client{ + Clientset: fake.NewSimpleClientset(pvc), + }, + } + + got, err := service.waitForPVCBinding(context.Background(), pvc.Namespace, pvc.Name, 155, 114, 100, "local-path", time.Nanosecond) + if err != nil { + t.Fatalf("waitForPVCBinding returned error: %v", err) + } + if got.Name != pvc.Name { + t.Fatalf("expected PVC %q, got %q", pvc.Name, got.Name) + } + pvs, err := service.client.Clientset.CoreV1().PersistentVolumes().List(context.Background(), metav1.ListOptions{}) + if err != nil { + t.Fatalf("failed to list PVs: %v", err) + } + if len(pvs.Items) != 0 { + t.Fatalf("expected no manual PVs for dynamic storageClass, got %#v", pvs.Items) + } +} + +func stringPtr(value string) *string { + return &value +} + func requireHostnameAffinity(t *testing.T, affinity *corev1.VolumeNodeAffinity, hostname string) { t.Helper() if affinity == nil || affinity.Required == nil || len(affinity.Required.NodeSelectorTerms) != 1 { diff --git a/backend/internal/services/leader/leader.go b/backend/internal/services/leader/leader.go new file mode 100644 index 0000000..e9050d2 --- /dev/null +++ b/backend/internal/services/leader/leader.go @@ -0,0 +1,124 @@ +// Package leader wraps Kubernetes lease-based leader election so that +// control-plane singleton background loops (the K8s sync loop, Team event +// consumers and the stale-task monitor) run on exactly one clawmanager-app +// replica at a time. The HTTP API and the in-pod nginx desktop data plane run +// on every replica; only these background loops must be gated. +package leader + +import ( + "context" + "log" + "os" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/leaderelection" + "k8s.io/client-go/tools/leaderelection/resourcelock" +) + +// Config holds the resolved leader-election parameters. +type Config struct { + Namespace string + LeaseName string + Identity string + LeaseDuration time.Duration + RenewDeadline time.Duration + RetryPeriod time.Duration +} + +// Callbacks are invoked when this replica acquires or loses leadership. +// +// OnStartedLeading receives a context that is cancelled when leadership is +// lost; callers may use it to scope leader-only work. OnStoppedLeading is +// invoked when leadership is lost (or the parent context is cancelled) and must +// stop any work started in OnStartedLeading. Both callbacks must be safe to +// call repeatedly across leadership transitions. +type Callbacks struct { + OnStartedLeading func(ctx context.Context) + OnStoppedLeading func() +} + +// Run participates in leader election and blocks until ctx is cancelled. It is +// resilient to leadership loss: after losing the lease it re-participates so a +// replica that briefly lost leadership can re-acquire it and restart the +// background loops. Run is intended to be called in its own goroutine. +func Run(ctx context.Context, clientset kubernetes.Interface, cfg Config, cb Callbacks) { + identity := cfg.Identity + if identity == "" { + if host, err := os.Hostname(); err == nil && host != "" { + identity = host + } else { + identity = "clawmanager-app" + } + } + + leaseDuration := cfg.LeaseDuration + if leaseDuration <= 0 { + leaseDuration = 15 * time.Second + } + renewDeadline := cfg.RenewDeadline + if renewDeadline <= 0 || renewDeadline >= leaseDuration { + renewDeadline = leaseDuration * 2 / 3 + } + retryPeriod := cfg.RetryPeriod + if retryPeriod <= 0 { + retryPeriod = 2 * time.Second + } + + lock := &resourcelock.LeaseLock{ + LeaseMeta: metav1.ObjectMeta{ + Name: cfg.LeaseName, + Namespace: cfg.Namespace, + }, + Client: clientset.CoordinationV1(), + LockConfig: resourcelock.ResourceLockConfig{ + Identity: identity, + }, + } + + electionCfg := leaderelection.LeaderElectionConfig{ + Lock: lock, + ReleaseOnCancel: true, + LeaseDuration: leaseDuration, + RenewDeadline: renewDeadline, + RetryPeriod: retryPeriod, + Callbacks: leaderelection.LeaderCallbacks{ + OnStartedLeading: func(c context.Context) { + log.Printf("[leader] acquired leadership (identity=%s lease=%s/%s)", identity, cfg.Namespace, cfg.LeaseName) + if cb.OnStartedLeading != nil { + cb.OnStartedLeading(c) + } + }, + OnStoppedLeading: func() { + log.Printf("[leader] lost leadership (identity=%s)", identity) + if cb.OnStoppedLeading != nil { + cb.OnStoppedLeading() + } + }, + }, + } + + for { + select { + case <-ctx.Done(): + return + default: + } + + elector, err := leaderelection.NewLeaderElector(electionCfg) + if err != nil { + log.Printf("[leader] failed to create leader elector: %v; retrying in %s", err, retryPeriod) + select { + case <-ctx.Done(): + return + case <-time.After(retryPeriod): + continue + } + } + + // Blocks until leadership is lost or ctx is cancelled. On loss we loop + // and re-participate. + elector.Run(ctx) + } +} diff --git a/backend/internal/services/sync_service.go b/backend/internal/services/sync_service.go index 937e22d..9911e4d 100644 --- a/backend/internal/services/sync_service.go +++ b/backend/internal/services/sync_service.go @@ -3,6 +3,7 @@ package services import ( "context" "fmt" + "sync" "time" "clawreef/internal/models" @@ -20,7 +21,10 @@ type SyncService struct { podService *k8s.PodService deploymentService *k8s.InstanceDeploymentService interval time.Duration - stopChan chan struct{} + + mu sync.Mutex + running bool + stopChan chan struct{} } // NewSyncService creates a new sync service @@ -31,23 +35,40 @@ func NewSyncService(instanceRepo repository.InstanceRepository, runtimeStatusSer podService: k8s.NewPodService(), deploymentService: k8s.NewInstanceDeploymentService(), interval: 5 * time.Second, // Sync every 5 seconds for more responsive status updates - stopChan: make(chan struct{}), } } -// Start starts the sync service +// Start starts the sync loop. It is safe to call repeatedly: a second call +// while already running is a no-op, and after Stop the service can be started +// again (used by leader-election re-acquisition). func (s *SyncService) Start() { + s.mu.Lock() + defer s.mu.Unlock() + if s.running { + return + } + s.stopChan = make(chan struct{}) + s.running = true fmt.Println("Starting K8s state sync service...") - go s.syncLoop() + go s.syncLoop(s.stopChan) } -// Stop stops the sync service +// Stop stops the sync loop. It is idempotent: calling Stop when not running is +// a no-op, and it never closes the same channel twice. func (s *SyncService) Stop() { + s.mu.Lock() + defer s.mu.Unlock() + if !s.running { + return + } close(s.stopChan) + s.running = false } -// syncLoop runs the synchronization loop -func (s *SyncService) syncLoop() { +// syncLoop runs the synchronization loop until its stop channel is closed. The +// channel is passed in (rather than read from the struct) so a restarted loop +// never races with a previously stopped one. +func (s *SyncService) syncLoop(stop <-chan struct{}) { fmt.Printf("[SyncService] Starting sync loop with interval %v\n", s.interval) ticker := time.NewTicker(s.interval) defer ticker.Stop() @@ -62,7 +83,7 @@ func (s *SyncService) syncLoop() { case <-ticker.C: fmt.Println("[SyncService] Tick - running scheduled sync...") s.syncAllInstances() - case <-s.stopChan: + case <-stop: fmt.Println("[SyncService] Stopping K8s state sync service...") return } diff --git a/backend/internal/services/team_service.go b/backend/internal/services/team_service.go index a0fa5aa..8e5fbbb 100644 --- a/backend/internal/services/team_service.go +++ b/backend/internal/services/team_service.go @@ -29,6 +29,7 @@ const ( defaultTeamTaskStaleTimeout = 30 * time.Minute teamTaskStaleSweepInterval = 30 * time.Second + teamConsumerScanInterval = 10 * time.Second initialLeaderTaskIntent = "team_bootstrap_introduction" teamTaskCompletionTool = "team_complete_task" @@ -42,8 +43,8 @@ var ( ) type TeamService interface { - Start() - Stop() + StartBackground(ctx context.Context) + StopBackground() CreateTeam(userID int, req CreateTeamRequest) (*TeamDetailsPayload, error) ListTeams(userID, offset, limit int) (*TeamListPayload, error) GetTeam(userID, teamID int) (*TeamDetailsPayload, error) @@ -137,6 +138,8 @@ type teamService struct { ctx context.Context cancel context.CancelFunc mu sync.Mutex + running bool + wg sync.WaitGroup consumers map[int]struct{} staleMonitorStarted bool runtimeWorkspaceRoot string @@ -188,20 +191,93 @@ func NewTeamService(repo repository.TeamRepository, instanceService InstanceServ return service } -func (s *teamService) Start() { - teams, err := s.repo.ListActiveTeams() - if err != nil { - fmt.Printf("Warning: failed to start Team event consumers: %v\n", err) +// StartBackground starts the leader-only background workers: a periodic scan +// that ensures a Redis event consumer is running for every active team, and +// the stale-task monitor. It is safe to call repeatedly (a second call while +// running is a no-op) and can be called again after StopBackground, which is +// required for leader-election re-acquisition. HTTP request handling does not +// depend on these workers, so followers can still serve the API and the in-pod +// nginx data plane while only the leader runs them. +func (s *teamService) StartBackground(parent context.Context) { + s.mu.Lock() + if s.running { + s.mu.Unlock() return } - for _, team := range teams { - s.ensureConsumer(team.ID) + if parent == nil { + parent = context.Background() } - s.ensureStaleTaskMonitor() + ctx, cancel := context.WithCancel(parent) + s.ctx = ctx + s.cancel = cancel + s.running = true + s.staleMonitorStarted = false + s.consumers = map[int]struct{}{} + s.wg.Add(1) + go s.consumerScanLoop(ctx) + s.mu.Unlock() + + fmt.Println("[TeamService] Starting leader-only background workers...") + s.ensureStaleTaskMonitor(ctx) } -func (s *teamService) Stop() { - s.cancel() +// StopBackground stops all background workers and blocks until they have fully +// exited, so a subsequent StartBackground starts from a clean state with no +// goroutines from the previous generation still touching shared maps. It is +// idempotent. +func (s *teamService) StopBackground() { + s.mu.Lock() + if !s.running { + s.mu.Unlock() + return + } + s.running = false + cancel := s.cancel + s.mu.Unlock() + + fmt.Println("[TeamService] Stopping leader-only background workers...") + if cancel != nil { + cancel() + } + s.wg.Wait() + + s.mu.Lock() + s.consumers = map[int]struct{}{} + s.staleMonitorStarted = false + s.mu.Unlock() +} + +// consumerScanLoop periodically ensures a consumer goroutine exists for every +// active team. Team creation no longer starts consumers inline (that would run +// on whichever replica served the request); the leader picks up newly active +// teams here within teamConsumerScanInterval. +func (s *teamService) consumerScanLoop(ctx context.Context) { + defer s.wg.Done() + + s.ensureConsumersForActiveTeams(ctx) + + ticker := time.NewTicker(teamConsumerScanInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.ensureConsumersForActiveTeams(ctx) + } + } +} + +func (s *teamService) ensureConsumersForActiveTeams(ctx context.Context) { + teams, err := s.repo.ListActiveTeams() + if err != nil { + fmt.Printf("Warning: failed to list active teams for consumer scan: %v\n", err) + return + } + for i := range teams { + s.ensureConsumer(ctx, teams[i].ID) + } } func (s *teamService) CreateTeam(userID int, req CreateTeamRequest) (*TeamDetailsPayload, error) { @@ -306,8 +382,10 @@ func (s *teamService) CreateTeam(userID int, req CreateTeamRequest) (*TeamDetail if err := s.repo.UpdateTeam(team); err != nil { return nil, err } - s.ensureConsumer(team.ID) - s.ensureStaleTaskMonitor() + // Background consumers / stale-task monitor are leader-only and started by + // the leader's periodic scan (consumerScanLoop). Starting them here would + // run them on whichever replica served the create request, bypassing + // leader election, so we intentionally do not call ensureConsumer here. if err := s.dispatchInitialLeaderTask(userID, team); err != nil { fmt.Printf("Warning: failed to dispatch initial Team %d leader task: %v\n", team.ID, err) if recordErr := s.recordInitialLeaderTaskDispatchFailure(team.ID, err); recordErr != nil { @@ -962,17 +1040,22 @@ func (s *teamService) requireOwnedTeam(userID, teamID int) (*models.Team, error) return team, nil } -func (s *teamService) ensureConsumer(teamID int) { +func (s *teamService) ensureConsumer(ctx context.Context, teamID int) { s.mu.Lock() defer s.mu.Unlock() + if !s.running { + return + } if _, exists := s.consumers[teamID]; exists { return } s.consumers[teamID] = struct{}{} - go s.consumeTeamEvents(teamID) + s.wg.Add(1) + go s.consumeTeamEvents(ctx, teamID) } -func (s *teamService) consumeTeamEvents(teamID int) { +func (s *teamService) consumeTeamEvents(ctx context.Context, teamID int) { + defer s.wg.Done() defer func() { s.mu.Lock() delete(s.consumers, teamID) @@ -981,7 +1064,7 @@ func (s *teamService) consumeTeamEvents(teamID int) { for { select { - case <-s.ctx.Done(): + case <-ctx.Done(): return default: } @@ -991,7 +1074,7 @@ func (s *teamService) consumeTeamEvents(teamID int) { time.Sleep(5 * time.Second) continue } - bus, err := s.redisBusForTeam(s.ctx, team) + bus, err := s.redisBusForTeam(ctx, team) if err != nil { time.Sleep(5 * time.Second) continue @@ -1000,7 +1083,7 @@ func (s *teamService) consumeTeamEvents(teamID int) { if lastID == "" { lastID = "0-0" } - messages, err := bus.XRead(s.ctx, teamEventsKey(teamID), lastID, 5*time.Second) + messages, err := bus.XRead(ctx, teamEventsKey(teamID), lastID, 5*time.Second) if err != nil { time.Sleep(2 * time.Second) continue @@ -1016,23 +1099,28 @@ func (s *teamService) consumeTeamEvents(teamID int) { } } -func (s *teamService) ensureStaleTaskMonitor() { +func (s *teamService) ensureStaleTaskMonitor(ctx context.Context) { s.mu.Lock() defer s.mu.Unlock() + if !s.running { + return + } if s.staleMonitorStarted { return } s.staleMonitorStarted = true - go s.monitorStaleTasks() + s.wg.Add(1) + go s.monitorStaleTasks(ctx) } -func (s *teamService) monitorStaleTasks() { +func (s *teamService) monitorStaleTasks(ctx context.Context) { + defer s.wg.Done() ticker := time.NewTicker(teamTaskStaleSweepInterval) defer ticker.Stop() for { select { - case <-s.ctx.Done(): + case <-ctx.Done(): return case <-ticker.C: if err := s.sweepStaleTasks(); err != nil { diff --git a/deployments/container/start.sh b/deployments/container/start.sh index a58a977..6c9645e 100644 --- a/deployments/container/start.sh +++ b/deployments/container/start.sh @@ -96,6 +96,15 @@ esac sed -i "s/client_max_body_size [0-9][0-9]*m;/client_max_body_size ${CLAWMANAGER_WORKSPACE_ARCHIVE_MAX_MIB}m;/" /etc/nginx/nginx.conf +# Resolve the cluster DNS server for nginx so the desktop location can resolve +# per-instance Service FQDNs at request time. Prefer the first nameserver from +# /etc/resolv.conf, falling back to the common in-cluster DNS ClusterIP. +DNS_RESOLVER="$(awk '/^nameserver/ {print $2; exit}' /etc/resolv.conf 2>/dev/null || true)" +if [ -z "${DNS_RESOLVER}" ]; then + DNS_RESOLVER="10.96.0.10" +fi +sed -i "s/__DNS_RESOLVER__/${DNS_RESOLVER}/g" /etc/nginx/nginx.conf + /usr/local/bin/clawreef-server & backend_pid=$! diff --git a/deployments/k3s/clawmanager.yaml b/deployments/k3s/clawmanager.yaml index 36399f1..93d2e89 100644 --- a/deployments/k3s/clawmanager.yaml +++ b/deployments/k3s/clawmanager.yaml @@ -979,6 +979,33 @@ subjects: name: clawmanager-app namespace: clawmanager-system --- +# Explicit lease permissions for control-plane leader election. The +# cluster-admin binding above already covers this; this Role documents the +# requirement and keeps leader election working if RBAC is tightened later. +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: clawmanager-app-leaderelection + namespace: clawmanager-system +rules: + - apiGroups: ["coordination.k8s.io"] + resources: ["leases"] + verbs: ["get", "list", "watch", "create", "update"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: clawmanager-app-leaderelection + namespace: clawmanager-system +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: clawmanager-app-leaderelection +subjects: + - kind: ServiceAccount + name: clawmanager-app + namespace: clawmanager-system +--- apiVersion: apps/v1 kind: Deployment metadata: @@ -1039,6 +1066,21 @@ spec: value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0" - name: TEAM_REDIS_URL value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0" + - name: CLAWMANAGER_DESKTOP_DIRECT_PROXY + value: "true" + # Leader election for control-plane singleton background loops. + # Required before scaling replicas > 1 so SyncService / Team event + # consumers / stale-task monitor run on exactly one replica. + - name: CLAWMANAGER_LEADER_ELECTION + value: "true" + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace - name: CLAWMANAGER_TEAM_REDIS_URL value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0" - name: CLAWMANAGER_TEAM_MANAGER_BASE_URL diff --git a/deployments/k8s/clawmanager.yaml b/deployments/k8s/clawmanager.yaml index 1a4f209..eec00e0 100644 --- a/deployments/k8s/clawmanager.yaml +++ b/deployments/k8s/clawmanager.yaml @@ -1116,6 +1116,33 @@ subjects: name: clawmanager-app namespace: clawmanager-system --- +# Explicit lease permissions for control-plane leader election. The +# cluster-admin binding above already covers this; this Role documents the +# requirement and keeps leader election working if RBAC is tightened later. +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: clawmanager-app-leaderelection + namespace: clawmanager-system +rules: + - apiGroups: ["coordination.k8s.io"] + resources: ["leases"] + verbs: ["get", "list", "watch", "create", "update"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: clawmanager-app-leaderelection + namespace: clawmanager-system +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: clawmanager-app-leaderelection +subjects: + - kind: ServiceAccount + name: clawmanager-app + namespace: clawmanager-system +--- apiVersion: apps/v1 kind: Deployment metadata: @@ -1255,6 +1282,21 @@ spec: value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0" - name: TEAM_REDIS_URL value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0" + - name: CLAWMANAGER_DESKTOP_DIRECT_PROXY + value: "true" + # Leader election for control-plane singleton background loops. + # Required before scaling replicas > 1 so SyncService / Team event + # consumers / stale-task monitor run on exactly one replica. + - name: CLAWMANAGER_LEADER_ELECTION + value: "true" + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace - name: CLAWMANAGER_TEAM_REDIS_URL value: "redis://clawmanager-redis.clawmanager-system.svc.cluster.local:6379/0" - name: CLAWMANAGER_TEAM_MANAGER_BASE_URL diff --git a/deployments/nginx/nginx.conf b/deployments/nginx/nginx.conf index 488d82a..ba31cf1 100644 --- a/deployments/nginx/nginx.conf +++ b/deployments/nginx/nginx.conf @@ -1,5 +1,13 @@ +load_module modules/ngx_http_js_module.so; + worker_processes auto; +# Expose the instance-access JWT secret to njs (process.env) so the edge +# gateway can verify desktop access tokens locally. Values come from the +# container environment; they are not logged. +env INSTANCE_ACCESS_TOKEN_SECRET; +env JWT_SECRET; + events { worker_connections 8192; } @@ -8,6 +16,12 @@ http { include /etc/nginx/mime.types; default_type application/octet-stream; + js_import desktop from /etc/nginx/njs/desktop_auth.js; + + # Cluster DNS resolver, templated by start.sh from /etc/resolv.conf so the + # desktop location can resolve per-instance Service FQDNs at request time. + resolver __DNS_RESOLVER__ valid=10s ipv6=off; + sendfile on; tcp_nopush on; tcp_nodelay on; @@ -34,6 +48,16 @@ http { '' close; } + map $desktop_target $desktop_proxy_mode { + "deny" "deny"; + "http://127.0.0.1:9001" "fallback"; + default "direct"; + } + + log_format desktop_proxy '$remote_addr - $status "$request_method $desktop_clean_uri $server_protocol" ' + 'mode=$desktop_proxy_mode upstream="$upstream_addr" ' + 'rt=$request_time urt=$upstream_response_time bytes=$body_bytes_sent'; + upstream clawreef_backend { server 127.0.0.1:9001; keepalive 128; @@ -104,20 +128,45 @@ http { proxy_request_buffering off; } - location ~ ^/api/v1/instances/[0-9]+/proxy/? { - proxy_pass http://clawreef_backend; + location ~ ^/api/v1/instances/(?[0-9]+)/proxy(?/.*)?$ { + # njs verifies the instance-access JWT locally and chooses the + # proxy target: the instance Service (direct), the in-process + # control-plane proxy (gray fallback), or "deny". + js_set $desktop_target desktop.resolveTarget; + # Forward the original URI to the upstream but with the JWT "token" + # query param stripped, so the access token never reaches webtop. + js_set $desktop_clean_uri desktop.cleanUri; + + access_log /var/log/nginx/access.log desktop_proxy; + add_header X-ClawManager-Desktop-Proxy $desktop_proxy_mode always; + + error_page 463 = @desktop_denied; + if ($desktop_target = "deny") { + return 463; + } + proxy_http_version 1.1; - proxy_set_header Host $host; + proxy_set_header Host $http_host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto https; + proxy_set_header X-Forwarded-Prefix /api/v1/instances/$inst_id/proxy; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection $connection_upgrade; + # Browsers may include the original iframe URL (with ?token=...) + # as Referer on subresource requests. Do not forward it to webtop. + proxy_set_header Referer ""; + + proxy_ssl_verify off; + proxy_ssl_server_name on; + proxy_connect_timeout 15s; proxy_read_timeout 3600s; proxy_send_timeout 3600s; proxy_buffering off; proxy_request_buffering off; + + proxy_pass $desktop_target$desktop_clean_uri; } location /api/ { @@ -136,5 +185,10 @@ http { location / { try_files $uri $uri/ /index.html; } + + location @desktop_denied { + add_header Content-Type text/plain always; + return 401 "Access token expired or invalid"; + } } } diff --git a/deployments/nginx/njs/desktop_auth.js b/deployments/nginx/njs/desktop_auth.js new file mode 100644 index 0000000..ec7f47e --- /dev/null +++ b/deployments/nginx/njs/desktop_auth.js @@ -0,0 +1,145 @@ +// desktop_auth.js +// +// njs helper used by the in-pod nginx to decide where a desktop proxy request +// should be sent. It locally verifies the ClawManager instance-access JWT +// (HS256, same secret as the Go control plane) and returns a proxy_pass target: +// +// "deny" -> token missing/invalid/expired/mismatched +// "https://" -> direct connection to the instance Service +// (taken from the token's "upstream" claim) +// "http://127.0.0.1:9001" -> fall back to the in-process control-plane +// proxy (gray rollout / legacy tokens without +// an "upstream" claim) +// +// The function runs in the main request context (js_set), so query args and +// cookies of the original request are available. + +var crypto = require('crypto'); + +var CONTROL_PLANE_FALLBACK = 'http://127.0.0.1:9001'; +var DENY = 'deny'; + +function secret() { + return process.env.INSTANCE_ACCESS_TOKEN_SECRET || process.env.JWT_SECRET || ''; +} + +// Normalize base64url / base64 to pad-less standard base64 for comparison. +function toStdB64NoPad(s) { + return String(s).replace(/-/g, '+').replace(/_/g, '/').replace(/=+$/, ''); +} + +function b64urlToString(s) { + var std = String(s).replace(/-/g, '+').replace(/_/g, '/'); + while (std.length % 4 !== 0) { + std += '='; + } + return Buffer.from(std, 'base64').toString(); +} + +function readToken(r) { + if (r.args && r.args.token) { + return r.args.token; + } + + var cookie = r.headersIn['Cookie']; + if (!cookie) { + return ''; + } + + var name = 'instance_access_' + r.variables.inst_id; + var parts = cookie.split(';'); + for (var i = 0; i < parts.length; i++) { + var kv = parts[i].trim(); + var eq = kv.indexOf('='); + if (eq > 0 && kv.substring(0, eq) === name) { + return kv.substring(eq + 1); + } + } + return ''; +} + +function resolveTarget(r) { + var key = secret(); + if (!key) { + r.error('desktop_auth: missing JWT secret in environment'); + return DENY; + } + + var token = readToken(r); + if (!token) { + return DENY; + } + + var segments = token.split('.'); + if (segments.length !== 3) { + return DENY; + } + + var signingInput = segments[0] + '.' + segments[1]; + var expected = crypto.createHmac('sha256', key).update(signingInput).digest('base64'); + if (toStdB64NoPad(expected) !== toStdB64NoPad(segments[2])) { + return DENY; + } + + var payload; + try { + payload = JSON.parse(b64urlToString(segments[1])); + } catch (e) { + return DENY; + } + + if (payload.token_type !== 'instance_access') { + return DENY; + } + + if (payload.exp && (Date.now() / 1000) >= Number(payload.exp)) { + return DENY; + } + + if (String(payload.instance_id) !== String(r.variables.inst_id)) { + return DENY; + } + + if (payload.upstream) { + return 'https://' + payload.upstream; + } + + return CONTROL_PLANE_FALLBACK; +} + +// cleanUri returns the original request URI (path + query) with the "token" +// query parameter stripped, so the instance-access JWT is never forwarded to +// the upstream desktop (it would otherwise land in webtop/KasmVNC access logs). +// njs validation in resolveTarget still reads the original token from the +// request, so authentication is unaffected. Subresource / WebSocket requests +// carry no token and are returned unchanged. +function cleanUri(r) { + var uri = r.variables.request_uri || r.uri || '/'; + var q = uri.indexOf('?'); + if (q < 0) { + return uri; + } + + var path = uri.substring(0, q); + var query = uri.substring(q + 1); + var parts = query.split('&'); + var kept = []; + for (var i = 0; i < parts.length; i++) { + if (parts[i] === '') { + continue; + } + var eq = parts[i].indexOf('='); + var name = eq < 0 ? parts[i] : parts[i].substring(0, eq); + if (name === 'token') { + continue; + } + kept.push(parts[i]); + } + + if (kept.length === 0) { + return path; + } + return path + '?' + kept.join('&'); +} + +export default { resolveTarget, cleanUri }; diff --git a/frontend/src/hooks/useInstanceDesktopAccess.ts b/frontend/src/hooks/useInstanceDesktopAccess.ts index 95a10e3..358484a 100644 --- a/frontend/src/hooks/useInstanceDesktopAccess.ts +++ b/frontend/src/hooks/useInstanceDesktopAccess.ts @@ -256,7 +256,7 @@ export function useInstanceDesktopAccess({ return; } - const nextEmbedUrl = resolveEmbedUrl(data.proxy_url || data.access_url); + const nextEmbedUrl = resolveEmbedUrl(data.access_url || data.proxy_url); const nextExpiresAt = new Date(data.expires_at); const previousEmbedUrl = embedUrlRef.current; diff --git a/frontend/src/lib/i18n.ts b/frontend/src/lib/i18n.ts index eab5af7..bba2f51 100644 --- a/frontend/src/lib/i18n.ts +++ b/frontend/src/lib/i18n.ts @@ -1860,6 +1860,24 @@ export const translations: Record = { presetCpuShort: "CPU", presetRamShort: "RAM", presetDiskShort: "Disk", + desktopStreamProfile: "Desktop Stream Profile", + desktopStreamSettings: "Stream Settings", + desktopStreamProfileDesc: + "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.", + desktopStreamLow: "Low", + desktopStreamLowDesc: + "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.", + desktopStreamStandard: "Standard", + desktopStreamStandardDesc: + "Balanced default with responsive motion and moderate clarity.", + desktopStreamHigh: "High", + desktopStreamHighDesc: + "Smoother motion with higher bandwidth and CPU usage.", + restartRequiredAfterChange: + "Changes take effect after the instance is restarted.", + desktopStreamSavedRestartRequired: + "Stream profile saved. Restart the instance for it to take effect.", + desktopStreamSaveFailed: "Failed to save stream profile.", environmentVariables: "Environment Variables", clawManagerBuiltIns: "ClawManager Built-ins", hideBuiltIns: "Hide Built-ins", @@ -3126,6 +3144,19 @@ export const translations: Record = { presetCpuShort: "CPU", presetRamShort: "RAM", presetDiskShort: "磁盘", + desktopStreamProfile: "桌面流档位", + desktopStreamSettings: "桌面流设置", + desktopStreamProfileDesc: + "选择 Selkies 桌面流的流畅度与带宽平衡。分辨率仍会跟随浏览器窗口动态适配。", + desktopStreamLow: "低带宽", + desktopStreamLowDesc: "适合弱网和高并发场景。保留可用帧率，降低清晰度。", + desktopStreamStandard: "标准", + desktopStreamStandardDesc: "适合日常桌面操作，兼顾跟手感和清晰度。", + desktopStreamHigh: "高清", + desktopStreamHighDesc: "画面更顺滑，但会占用更多带宽和 CPU。", + restartRequiredAfterChange: "修改会在实例重启后生效。", + desktopStreamSavedRestartRequired: "桌面流档位已保存，请重启实例后生效。", + desktopStreamSaveFailed: "保存桌面流档位失败。", environmentVariables: "环境变量", clawManagerBuiltIns: "ClawManager 内置变量", hideBuiltIns: "收起内置变量", @@ -4373,6 +4404,24 @@ export const translations: Record = { presetCpuShort: "CPU", presetRamShort: "RAM", presetDiskShort: "ディスク", + desktopStreamProfile: "Desktop Stream Profile", + desktopStreamSettings: "Stream Settings", + desktopStreamProfileDesc: + "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.", + desktopStreamLow: "Low", + desktopStreamLowDesc: + "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.", + desktopStreamStandard: "Standard", + desktopStreamStandardDesc: + "Balanced default with responsive motion and moderate clarity.", + desktopStreamHigh: "High", + desktopStreamHighDesc: + "Smoother motion with higher bandwidth and CPU usage.", + restartRequiredAfterChange: + "Changes take effect after the instance is restarted.", + desktopStreamSavedRestartRequired: + "Stream profile saved. Restart the instance for it to take effect.", + desktopStreamSaveFailed: "Failed to save stream profile.", environmentVariables: "環境変数", clawManagerBuiltIns: "ClawManager 組み込み変数", hideBuiltIns: "組み込み変数を隠す", @@ -5630,6 +5679,24 @@ export const translations: Record = { presetCpuShort: "CPU", presetRamShort: "RAM", presetDiskShort: "디스크", + desktopStreamProfile: "Desktop Stream Profile", + desktopStreamSettings: "Stream Settings", + desktopStreamProfileDesc: + "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.", + desktopStreamLow: "Low", + desktopStreamLowDesc: + "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.", + desktopStreamStandard: "Standard", + desktopStreamStandardDesc: + "Balanced default with responsive motion and moderate clarity.", + desktopStreamHigh: "High", + desktopStreamHighDesc: + "Smoother motion with higher bandwidth and CPU usage.", + restartRequiredAfterChange: + "Changes take effect after the instance is restarted.", + desktopStreamSavedRestartRequired: + "Stream profile saved. Restart the instance for it to take effect.", + desktopStreamSaveFailed: "Failed to save stream profile.", environmentVariables: "환경 변수", clawManagerBuiltIns: "ClawManager 내장 변수", hideBuiltIns: "내장 변수 숨기기", @@ -6912,6 +6979,24 @@ export const translations: Record = { presetCpuShort: "CPU", presetRamShort: "RAM", presetDiskShort: "Disk", + desktopStreamProfile: "Desktop Stream Profile", + desktopStreamSettings: "Stream Settings", + desktopStreamProfileDesc: + "Choose the Selkies desktop streaming balance. Resolution remains adaptive to the browser window.", + desktopStreamLow: "Low", + desktopStreamLowDesc: + "Best for weak networks and large concurrency. Keeps motion usable while reducing clarity.", + desktopStreamStandard: "Standard", + desktopStreamStandardDesc: + "Balanced default with responsive motion and moderate clarity.", + desktopStreamHigh: "High", + desktopStreamHighDesc: + "Smoother motion with higher bandwidth and CPU usage.", + restartRequiredAfterChange: + "Changes take effect after the instance is restarted.", + desktopStreamSavedRestartRequired: + "Stream profile saved. Restart the instance for it to take effect.", + desktopStreamSaveFailed: "Failed to save stream profile.", environmentVariables: "Umgebungsvariablen", clawManagerBuiltIns: "ClawManager Built-ins", hideBuiltIns: "Built-ins ausblenden", diff --git a/frontend/src/pages/instances/CreateInstancePage.tsx b/frontend/src/pages/instances/CreateInstancePage.tsx index aa60cf6..4616059 100644 --- a/frontend/src/pages/instances/CreateInstancePage.tsx +++ b/frontend/src/pages/instances/CreateInstancePage.tsx @@ -20,6 +20,7 @@ import { systemSettingsService, type SystemImageSetting, } from "../../services/systemSettingsService"; +import type { DesktopStreamProfile } from "../../types/instance"; type BuiltInEnvTemplate = { key: string; @@ -46,6 +47,31 @@ const CUSTOM_RESOURCE_PRESET = "custom"; const SKILLS_PER_PAGE = 6; const supportsRuntimeInjection = (type: string) => type === "openclaw" || type === "hermes"; +const DESKTOP_STREAM_PROFILES: Array<{ + id: DesktopStreamProfile; + labelKey: string; + descriptionKey: string; + details: string; +}> = [ + { + id: "low", + labelKey: "instances.desktopStreamLow", + descriptionKey: "instances.desktopStreamLowDesc", + details: "30 FPS / CRF 42", + }, + { + id: "standard", + labelKey: "instances.desktopStreamStandard", + descriptionKey: "instances.desktopStreamStandardDesc", + details: "35 FPS / CRF 34", + }, + { + id: "high", + labelKey: "instances.desktopStreamHigh", + descriptionKey: "instances.desktopStreamHighDesc", + details: "40 FPS / CRF 24", + }, +]; const runtimeWorkspaceDirectory = (type: string) => type === "hermes" ? ".hermes" : ".openclaw"; @@ -442,6 +468,8 @@ const CreateInstancePage: React.FC = () => { Record >({}); const [customEnvRows, setCustomEnvRows] = useState([]); + const [desktopStreamProfile, setDesktopStreamProfile] = + useState("standard"); const [resourcePresetMode, setResourcePresetMode] = useState< keyof typeof PRESET_CONFIGS | typeof CUSTOM_RESOURCE_PRESET >("medium"); @@ -821,6 +849,8 @@ const CreateInstancePage: React.FC = () => { : PRESET_CONFIGS.small.disk_gb, gpu_enabled: usesDedicatedResources ? formData.gpu_enabled : false, gpu_count: usesDedicatedResources ? formData.gpu_count : 0, + desktop_stream_profile: + selectedRuntimeType === "desktop" ? desktopStreamProfile : undefined, image_registry: selectedRuntimeImage?.image, image_tag: selectedRuntimeImage ? undefined : formData.image_tag, environment_overrides: overrides, @@ -1579,6 +1609,54 @@ const CreateInstancePage: React.FC = () => { )} + {selectedRuntimeType === "desktop" && ( +
+
+
+

+ {t("instances.desktopStreamProfile")} +

+

+ {t("instances.desktopStreamProfileDesc")} +

+
+ + {t("instances.restartRequiredAfterChange")} + +
+ +
+ {DESKTOP_STREAM_PROFILES.map((profile) => { + const selected = desktopStreamProfile === profile.id; + return ( + + ); + })} +
+
+ )} +
diff --git a/frontend/src/pages/instances/InstanceDetailPage.tsx b/frontend/src/pages/instances/InstanceDetailPage.tsx index 6e15dc1..0d0a0ce 100644 --- a/frontend/src/pages/instances/InstanceDetailPage.tsx +++ b/frontend/src/pages/instances/InstanceDetailPage.tsx @@ -40,11 +40,21 @@ import type { InstanceRuntimeCommand, InstanceRuntimeDetails, InstanceStatus, + DesktopStreamProfile, } from "../../types/instance"; import type { InstanceSkill, Skill } from "../../types/skill"; const META_POLL_INTERVAL_MS = 5000; const RUNTIME_POLL_INTERVAL_MS = 5000; +const DESKTOP_STREAM_PROFILES: Array<{ + id: DesktopStreamProfile; + labelKey: string; + detail: string; +}> = [ + { id: "low", labelKey: "instances.desktopStreamLow", detail: "30 FPS / CRF 42" }, + { id: "standard", labelKey: "instances.desktopStreamStandard", detail: "35 FPS / CRF 34" }, + { id: "high", labelKey: "instances.desktopStreamHigh", detail: "40 FPS / CRF 24" }, +]; function availabilityForStatus(status: string): InstanceAvailability { if (status === "running") { @@ -304,6 +314,11 @@ const InstanceDetailPage: React.FC = () => { const [selectedSkillId, setSelectedSkillId] = useState(""); const [skillLoading, setSkillLoading] = useState(false); const [skillError, setSkillError] = useState(null); + const [desktopStreamProfile, setDesktopStreamProfile] = + useState("standard"); + const [desktopStreamSavedProfile, setDesktopStreamSavedProfile] = + useState("standard"); + const [desktopStreamMessage, setDesktopStreamMessage] = useState(null); const fetchMeta = useCallback( async (targetInstanceId: number, options?: { background?: boolean }) => { @@ -379,6 +394,13 @@ const InstanceDetailPage: React.FC = () => { } }, []); + useEffect(() => { + const profile = instance?.desktop_stream_profile || "standard"; + setDesktopStreamProfile(profile); + setDesktopStreamSavedProfile(profile); + setDesktopStreamMessage(null); + }, [instance?.desktop_stream_profile, instance?.id]); + useEffect(() => { if (!instanceId || Number.isNaN(instanceId)) { setError(t("instances.instanceNotFound")); @@ -562,6 +584,31 @@ const InstanceDetailPage: React.FC = () => { } }; + const handleSaveDesktopStreamProfile = async () => { + if (!instance || desktopStreamProfile === desktopStreamSavedProfile) { + return; + } + + try { + setActionLoading("desktop-stream-profile"); + setDesktopStreamMessage(null); + await instanceService.updateInstance(instance.id, { + desktop_stream_profile: desktopStreamProfile, + }); + const updatedInstance = await instanceService.getInstance(instance.id); + setInstance(updatedInstance); + const savedProfile = updatedInstance.desktop_stream_profile || desktopStreamProfile; + setDesktopStreamSavedProfile(savedProfile); + setDesktopStreamProfile(savedProfile); + setDesktopStreamMessage(t("instances.desktopStreamSavedRestartRequired")); + } catch (streamError) { + console.error("Failed to update desktop stream profile", streamError); + setDesktopStreamMessage(t("instances.desktopStreamSaveFailed")); + } finally { + setActionLoading(null); + } + }; + const copyExternalValue = async (key: string, value: string) => { if (!value) { return; @@ -983,6 +1030,55 @@ const InstanceDetailPage: React.FC = () => { )} + {(instance.runtime_type || "desktop") === "desktop" && ( +
+
+
+

+ {t("instances.desktopStreamProfile")} +

+

+ {t("instances.desktopStreamProfileDesc")} +

+
+ +
+
+ {DESKTOP_STREAM_PROFILES.map((profile) => { + const selected = desktopStreamProfile === profile.id; + return ( + + ); + })} +
+

+ {desktopStreamMessage || t("instances.restartRequiredAfterChange")} +

+
+ )} +
diff --git a/frontend/src/services/instanceService.ts b/frontend/src/services/instanceService.ts index a00ef38..fecf2a4 100644 --- a/frontend/src/services/instanceService.ts +++ b/frontend/src/services/instanceService.ts @@ -132,6 +132,8 @@ export const instanceService = { access_url: string; proxy_url: string; expires_at: string; + desktop_proxy_mode?: "control-plane" | "fallback" | "direct"; + desktop_upstream_present?: boolean; }> => { const response = await api.post(`/instances/${id}/access`); return response.data.data; diff --git a/frontend/src/types/instance.ts b/frontend/src/types/instance.ts index 1120eff..3cb45fb 100644 --- a/frontend/src/types/instance.ts +++ b/frontend/src/types/instance.ts @@ -26,6 +26,7 @@ export interface Instance { os_version: string; image_registry?: string; image_tag?: string; + desktop_stream_profile?: DesktopStreamProfile; storage_class: string; mount_path: string; workspace_path?: string; @@ -179,6 +180,7 @@ export interface CreateInstanceRequest { mode?: InstanceMode; instance_mode?: InstanceMode; runtime_type?: "desktop" | "shell" | "gateway"; + desktop_stream_profile?: DesktopStreamProfile; cpu_cores: number; memory_gb: number; disk_gb: number; @@ -197,8 +199,11 @@ export interface CreateInstanceRequest { export interface UpdateInstanceRequest { name?: string; description?: string; + desktop_stream_profile?: DesktopStreamProfile; } +export type DesktopStreamProfile = "low" | "standard" | "high"; + export interface InstanceListResponse { instances: Instance[]; total: number;