package docker import ( "context" "fmt" "io" "net/http" "net/netip" "os" "path/filepath" "sort" "strings" "time" "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/mount" "github.com/docker/docker/api/types/network" "github.com/docker/docker/client" "github.com/docker/docker/errdefs" "github.com/rs/zerolog/log" ) // ManagerConfig holds Docker manager settings. type ManagerConfig struct { Image string Network string BaseDomain string TrialActivationPublicKey string TrustedProxyCIDRs []string MemoryLimit int64 // bytes CPUShares int64 ContainerPort int // port inside the container (default 7655) } // Manager orchestrates Docker containers for tenant lifecycle. type Manager struct { cli *client.Client cfg ManagerConfig } // RuntimeContainerInfo is the canonical live-container snapshot used by // operator rollout and reconciliation flows. type RuntimeContainerInfo struct { ID string Name string ImageRef string ImageID string Running bool RouteHost string PublicURL string } const immutableOwnershipPathsEnv = "PULSE_IMMUTABLE_OWNERSHIP_PATHS" const ( tenantRuntimeUID = 1000 tenantRuntimeGID = 1000 ) // NewManager creates a Docker manager connected to the local daemon. func NewManager(cfg ManagerConfig) (*Manager, error) { cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) if err != nil { return nil, fmt.Errorf("create docker client: %w", err) } if cfg.ContainerPort == 0 { cfg.ContainerPort = 7655 } return &Manager{cli: cli, cfg: cfg}, nil } // Close closes the Docker client. func (m *Manager) Close() error { if m.cli != nil { return m.cli.Close() } return nil } // DesiredRuntimeRouting returns the canonical hosted route/public URL contract // for a tenant on this manager's configured base domain. func (m *Manager) DesiredRuntimeRouting(tenantID string) TenantRuntimeRoutingContract { if m == nil { return TenantRuntimeRoutingContract{} } return CanonicalTenantRuntimeRouting(tenantID, m.cfg.BaseDomain) } // IsNotFound reports whether Docker treated an identifier as missing. func IsNotFound(err error) bool { return errdefs.IsNotFound(err) } func (m *Manager) ensureDaemonReachable(ctx context.Context) error { if m == nil || m.cli == nil { return fmt.Errorf("docker client unavailable") } if _, err := m.cli.Ping(ctx); err != nil { return fmt.Errorf("ping docker daemon: %w", err) } return nil } // CreateAndStart creates and starts a tenant container. // tenantDataDir is the host path that gets bind-mounted to /etc/pulse in the container. func (m *Manager) CreateAndStart(ctx context.Context, tenantID, tenantDataDir string) (containerID string, err error) { if err := m.ensureDaemonReachable(ctx); err != nil { return "", err } if err := prepareTenantRuntimeMountSources(tenantDataDir, tenantRuntimeUID, tenantRuntimeGID); err != nil { return "", fmt.Errorf("prepare tenant runtime mounts for %s: %w", tenantID, err) } labels := TraefikLabels(tenantID, m.cfg.BaseDomain, m.cfg.ContainerPort) labels["pulse.managed"] = "true" containerName := "pulse-" + tenantID resp, err := m.cli.ContainerCreate(ctx, &container.Config{ Image: m.cfg.Image, Labels: labels, Env: tenantEnv(tenantID, m.cfg.BaseDomain, m.cfg.TrialActivationPublicKey, m.tenantTrustedProxyCIDRs(ctx)), }, &container.HostConfig{ RestartPolicy: container.RestartPolicy{Name: "unless-stopped"}, Resources: container.Resources{ Memory: m.cfg.MemoryLimit, CPUShares: m.cfg.CPUShares, }, Mounts: tenantMounts(tenantDataDir), }, &network.NetworkingConfig{ EndpointsConfig: map[string]*network.EndpointSettings{ m.cfg.Network: {}, }, }, nil, // platform containerName, ) if err != nil { return "", fmt.Errorf("create container for %s: %w", tenantID, err) } if err := m.cli.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil { return resp.ID, fmt.Errorf("start container for %s: %w", tenantID, err) } log.Info(). Str("tenant_id", tenantID). Str("container_id", resp.ID[:12]). Str("container_name", containerName). Msg("Tenant container started") return resp.ID, nil } func tenantImmutableOwnershipPaths() []string { return []string{ "/etc/pulse/secrets/handoff.key", "/etc/pulse/.cloud_handoff_key", } } func tenantRuntimeOwnershipPaths() []string { return []string{ "billing.json", filepath.Join("secrets", "handoff.key"), ".cloud_handoff_key", } } func tenantEnv(tenantID, baseDomain, trialActivationPublicKey string, trustedProxyCIDRs []string) []string { routing := CanonicalTenantRuntimeRouting(tenantID, baseDomain) tenantID = strings.TrimSpace(tenantID) env := []string{ "PULSE_DATA_DIR=/etc/pulse", "PULSE_HOSTED_MODE=true", "PULSE_TENANT_ID=" + tenantID, "PULSE_MULTI_TENANT_ENABLED=true", fmt.Sprintf("PUID=%d", tenantRuntimeUID), fmt.Sprintf("PGID=%d", tenantRuntimeGID), fmt.Sprintf("%s=%s", immutableOwnershipPathsEnv, strings.Join(tenantImmutableOwnershipPaths(), ":")), } if routing.PublicURL != "" { env = append(env, "PULSE_PUBLIC_URL="+routing.PublicURL) } if strings.TrimSpace(trialActivationPublicKey) != "" { env = append(env, "PULSE_TRIAL_ACTIVATION_PUBLIC_KEY="+strings.TrimSpace(trialActivationPublicKey)) } if len(trustedProxyCIDRs) > 0 { env = append(env, "PULSE_TRUSTED_PROXY_CIDRS="+strings.Join(trustedProxyCIDRs, ",")) } return env } func routeHostFromLabels(labels map[string]string) string { for key, value := range labels { if !strings.HasPrefix(key, "traefik.http.routers.") || !strings.HasSuffix(key, ".rule") { continue } if host := parseTraefikHostRule(value); host != "" { return host } } return "" } func parseTraefikHostRule(rule string) string { rule = strings.TrimSpace(rule) if !strings.HasPrefix(rule, "Host(`") || !strings.HasSuffix(rule, "`)") { return "" } return strings.TrimSuffix(strings.TrimPrefix(rule, "Host(`"), "`)") } func envValue(env []string, key string) string { prefix := key + "=" for _, item := range env { if !strings.HasPrefix(item, prefix) { continue } return strings.TrimPrefix(item, prefix) } return "" } func (m *Manager) tenantTrustedProxyCIDRs(ctx context.Context) []string { values := make([]string, 0, len(m.cfg.TrustedProxyCIDRs)+1) seen := make(map[string]struct{}) appendCIDR := func(raw string) { canonical := canonicalTrustedProxyCIDR(raw) if canonical == "" { return } if _, ok := seen[canonical]; ok { return } seen[canonical] = struct{}{} values = append(values, canonical) } for _, cidr := range m.cfg.TrustedProxyCIDRs { appendCIDR(cidr) } if m != nil && m.cli != nil && strings.TrimSpace(m.cfg.Network) != "" { networkName := strings.TrimSpace(m.cfg.Network) inspect, err := m.cli.NetworkInspect(ctx, networkName, network.InspectOptions{}) if err != nil { log.Warn().Err(err).Str("network", networkName).Msg("Failed to inspect tenant network for trusted proxy CIDRs") } else { for _, cfg := range inspect.IPAM.Config { appendCIDR(cfg.Subnet) } } } sort.Strings(values) return values } func canonicalTrustedProxyCIDR(raw string) string { raw = strings.TrimSpace(raw) if raw == "" { return "" } if prefix, err := netip.ParsePrefix(raw); err == nil { return prefix.Masked().String() } if addr, err := netip.ParseAddr(raw); err == nil { bits := 32 if addr.Is6() { bits = 128 } return netip.PrefixFrom(addr, bits).Masked().String() } return "" } func tenantMounts(tenantDataDir string) []mount.Mount { return []mount.Mount{ { Type: mount.TypeBind, Source: tenantDataDir, Target: "/etc/pulse", }, { Type: mount.TypeBind, Source: filepath.Join(tenantDataDir, "secrets", "handoff.key"), Target: "/etc/pulse/secrets/handoff.key", ReadOnly: true, }, { Type: mount.TypeBind, Source: filepath.Join(tenantDataDir, ".cloud_handoff_key"), Target: "/etc/pulse/.cloud_handoff_key", ReadOnly: true, }, } } func prepareTenantRuntimeMountSources(tenantDataDir string, uid, gid int) error { for _, relPath := range tenantRuntimeOwnershipPaths() { path := filepath.Join(tenantDataDir, relPath) if err := os.Chmod(path, 0o600); err != nil { return fmt.Errorf("chmod %s: %w", path, err) } if err := os.Chown(path, uid, gid); err != nil { return fmt.Errorf("chown %s to %d:%d: %w", path, uid, gid, err) } } return nil } // Stop stops a tenant container gracefully. func (m *Manager) Stop(ctx context.Context, containerID string) error { timeout := 30 return m.cli.ContainerStop(ctx, containerID, container.StopOptions{Timeout: &timeout}) } // Start starts a stopped tenant container. func (m *Manager) Start(ctx context.Context, containerID string) error { return m.cli.ContainerStart(ctx, containerID, container.StartOptions{}) } // Remove removes a stopped tenant container. func (m *Manager) Remove(ctx context.Context, containerID string) error { return m.cli.ContainerRemove(ctx, containerID, container.RemoveOptions{ Force: true, }) } // Rename renames a tenant container to the supplied Docker name. func (m *Manager) Rename(ctx context.Context, containerIDOrName, newName string) error { if err := m.ensureDaemonReachable(ctx); err != nil { return err } if err := m.cli.ContainerRename(ctx, containerIDOrName, newName); err != nil { return fmt.Errorf("rename container %s -> %s: %w", containerIDOrName, newName, err) } return nil } // Inspect returns the canonical live metadata for the supplied container // identifier. func (m *Manager) Inspect(ctx context.Context, containerIDOrName string) (*RuntimeContainerInfo, error) { if err := m.ensureDaemonReachable(ctx); err != nil { return nil, err } inspect, err := m.cli.ContainerInspect(ctx, containerIDOrName) if err != nil { return nil, fmt.Errorf("inspect container %s: %w", containerIDOrName, err) } imageRef := "" if inspect.Config != nil { imageRef = strings.TrimSpace(inspect.Config.Image) } running := inspect.State != nil && inspect.State.Running routeHost := "" publicURL := "" if inspect.Config != nil { routeHost = routeHostFromLabels(inspect.Config.Labels) publicURL = envValue(inspect.Config.Env, "PULSE_PUBLIC_URL") } return &RuntimeContainerInfo{ ID: inspect.ID, Name: strings.TrimPrefix(strings.TrimSpace(inspect.Name), "/"), ImageRef: imageRef, ImageID: strings.TrimSpace(inspect.Image), Running: running, RouteHost: routeHost, PublicURL: publicURL, }, nil } // StopAndRemove stops then removes a tenant container. func (m *Manager) StopAndRemove(ctx context.Context, containerID string) error { if err := m.Stop(ctx, containerID); err != nil { log.Warn().Err(err).Str("container_id", containerID).Msg("Failed to stop container, forcing remove") } return m.Remove(ctx, containerID) } // HealthCheck performs an HTTP health check against a running container. // It connects to the container's published port via the Docker network. func (m *Manager) HealthCheck(ctx context.Context, containerID string) (bool, error) { inspect, err := m.cli.ContainerInspect(ctx, containerID) if err != nil { return false, fmt.Errorf("inspect container: %w", err) } if !inspect.State.Running { return false, nil } // Find the container's IP on our network netSettings, ok := inspect.NetworkSettings.Networks[m.cfg.Network] if !ok || netSettings.IPAddress == "" { return false, fmt.Errorf("container not connected to network %s", m.cfg.Network) } healthURL := fmt.Sprintf("http://%s:%d/api/health", netSettings.IPAddress, m.cfg.ContainerPort) httpClient := &http.Client{Timeout: 5 * time.Second} resp, err := httpClient.Get(healthURL) if err != nil { return false, nil // unreachable, not an error condition } defer func() { _, _ = io.Copy(io.Discard, resp.Body); resp.Body.Close() }() return resp.StatusCode == http.StatusOK, nil }