493 lines
16 KiB
Go
493 lines
16 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.ipao.vip/rogee/creator-hub/internal/hub"
|
|
"github.com/gofiber/fiber/v3"
|
|
)
|
|
|
|
// hubStore 是控制面编排所需的存储能力；生产实现为 *hub.Store，测试使用内存桩。
|
|
type hubStore interface {
|
|
CreateGateway(ctx context.Context, name, endpoint, token string) (hub.Gateway, error)
|
|
ListGateways(ctx context.Context) ([]hub.Gateway, error)
|
|
GetGateway(ctx context.Context, name string) (hub.Gateway, error)
|
|
DeleteGateway(ctx context.Context, name string) error
|
|
CreateImage(ctx context.Context, image hub.Image) error
|
|
UpdateImage(ctx context.Context, image hub.Image) error
|
|
ListImages(ctx context.Context, enabledOnly bool) ([]hub.Image, error)
|
|
DeleteImage(ctx context.Context, version string) error
|
|
ImageRef(ctx context.Context, version string) (string, error)
|
|
CreateEnv(ctx context.Context, env hub.Env) error
|
|
ListEnvs(ctx context.Context) ([]hub.Env, error)
|
|
GetEnv(ctx context.Context, alias string) (hub.Env, error)
|
|
UpgradeEnv(ctx context.Context, alias, version string) error
|
|
DeleteEnv(ctx context.Context, alias string) error
|
|
}
|
|
|
|
const (
|
|
gatewayLongTimeout = 11 * time.Minute // 覆盖网关侧最长 10 分钟的镜像拉取
|
|
gatewayReconcileDelay = 100 * time.Millisecond
|
|
gatewayReconcileAttempts = 10
|
|
)
|
|
|
|
// gatewayCall 调用某个网关的 /v1 路由；ok 为 false 时 status/body 携带网关错误。
|
|
func gatewayCall(ctx context.Context, target hub.Gateway, method, path string, body any, timeout time.Duration) (status int, responseBody []byte, err error) {
|
|
callCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
var payload io.Reader
|
|
if body != nil {
|
|
encoded, marshalErr := json.Marshal(body)
|
|
if marshalErr != nil {
|
|
return 0, nil, marshalErr
|
|
}
|
|
payload = bytes.NewReader(encoded)
|
|
}
|
|
request, requestErr := http.NewRequestWithContext(callCtx, method, target.Endpoint+path, payload)
|
|
if requestErr != nil {
|
|
return 0, nil, requestErr
|
|
}
|
|
request.Header.Set("Authorization", "Bearer "+target.Token)
|
|
if body != nil {
|
|
request.Header.Set("Content-Type", "application/json")
|
|
}
|
|
response, requestErr := http.DefaultClient.Do(request)
|
|
if requestErr != nil {
|
|
return 0, nil, requestErr
|
|
}
|
|
defer response.Body.Close()
|
|
responseBody, _ = io.ReadAll(io.LimitReader(response.Body, 1<<20))
|
|
return response.StatusCode, responseBody, nil
|
|
}
|
|
|
|
func gatewayCreatePayload(env hub.Env, imageRef string) map[string]any {
|
|
cmd := append(env.Fingerprint.Args(), "about:blank")
|
|
return map[string]any{
|
|
"alias": env.Alias,
|
|
"name": env.Name,
|
|
"image": imageRef,
|
|
"cmd": cmd,
|
|
"volume": "creatorhub-profile-" + env.Alias,
|
|
}
|
|
}
|
|
|
|
// gatewayFailure 将网关错误转换为对调用方可读的失败；4xx 保留原状态，其余一律 502。
|
|
type gatewayFailure struct {
|
|
status int
|
|
message string
|
|
}
|
|
|
|
func (e gatewayFailure) Error() string { return e.message }
|
|
|
|
func gatewayRejected(status int, body []byte) error {
|
|
if status >= 400 && status < 500 {
|
|
return gatewayFailure{status: status, message: "gateway rejected: " + errorFromBody(body, status)}
|
|
}
|
|
return gatewayFailure{status: http.StatusBadGateway, message: fmt.Sprintf("gateway call failed with status %d", status)}
|
|
}
|
|
|
|
func gatewayUnreachable(err error) error {
|
|
return gatewayFailure{status: http.StatusBadGateway, message: fmt.Sprintf("gateway unreachable: %v", err)}
|
|
}
|
|
|
|
func reconcileGatewayBrowser(ctx context.Context, target hub.Gateway, alias string) (bool, error) {
|
|
var lastErr error
|
|
for attempt := 0; attempt < gatewayReconcileAttempts; attempt++ {
|
|
if attempt > 0 {
|
|
timer := time.NewTimer(gatewayReconcileDelay)
|
|
select {
|
|
case <-ctx.Done():
|
|
timer.Stop()
|
|
return false, ctx.Err()
|
|
case <-timer.C:
|
|
}
|
|
}
|
|
status, body, err := gatewayCall(ctx, target, http.MethodGet, "/v1/browsers", nil, 30*time.Second)
|
|
if err != nil {
|
|
lastErr = err
|
|
continue
|
|
}
|
|
if status != http.StatusOK {
|
|
lastErr = gatewayRejected(status, body)
|
|
continue
|
|
}
|
|
lastErr = nil
|
|
var browsers []containerStatus
|
|
if err := json.Unmarshal(body, &browsers); err != nil {
|
|
lastErr = fmt.Errorf("decode gateway browser list: %w", err)
|
|
continue
|
|
}
|
|
for _, browser := range browsers {
|
|
if browser.Alias == alias {
|
|
return true, nil
|
|
}
|
|
}
|
|
}
|
|
return false, lastErr
|
|
}
|
|
|
|
func errorFromBody(body []byte, status int) string {
|
|
var envelope struct {
|
|
Error string `json:"error"`
|
|
}
|
|
if json.Unmarshal(body, &envelope) == nil && envelope.Error != "" {
|
|
return envelope.Error
|
|
}
|
|
return http.StatusText(status)
|
|
}
|
|
|
|
type containerStatus struct {
|
|
ID string `json:"id"`
|
|
Alias string `json:"alias"`
|
|
Name string `json:"name"`
|
|
State string `json:"state"`
|
|
Status string `json:"status"`
|
|
Endpoint string `json:"endpoint"`
|
|
}
|
|
|
|
type envView struct {
|
|
hub.Env
|
|
State string `json:"state"`
|
|
Status string `json:"status"`
|
|
ContainerID string `json:"container_id"`
|
|
Endpoint string `json:"endpoint"`
|
|
}
|
|
|
|
func registerHub(app *fiber.App, store hubStore) {
|
|
var operations sync.Mutex
|
|
// ponytail: one control-plane instance is serialized globally; use keyed/distributed locks if replicas or throughput require it.
|
|
serialized := func(handler fiber.Handler) fiber.Handler {
|
|
return func(c fiber.Ctx) error {
|
|
operations.Lock()
|
|
defer operations.Unlock()
|
|
return handler(c)
|
|
}
|
|
}
|
|
app.Get("/api/browsers", listBrowsers(store))
|
|
app.Post("/api/browsers", serialized(createBrowser(store)))
|
|
app.Post("/api/browsers/:alias/:action", serialized(browserAction(store)))
|
|
app.Delete("/api/browsers/:alias", serialized(deleteBrowser(store)))
|
|
|
|
app.Get("/api/browser-images", func(c fiber.Ctx) error {
|
|
images, err := store.ListImages(c.Context(), false)
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.JSON(images)
|
|
})
|
|
app.Post("/api/browser-images", func(c fiber.Ctx) error {
|
|
input := struct {
|
|
Version string `json:"version"`
|
|
ImageRef string `json:"image_ref"`
|
|
Note string `json:"note"`
|
|
Enabled *bool `json:"enabled"`
|
|
}{}
|
|
if err := decodeHubJSON(c, &input); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
enabled := true
|
|
if input.Enabled != nil {
|
|
enabled = *input.Enabled
|
|
}
|
|
if err := store.CreateImage(c.Context(), hub.Image{Version: input.Version, ImageRef: input.ImageRef, Note: input.Note, Enabled: enabled}); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.Status(fiber.StatusCreated).JSON(map[string]any{
|
|
"version": input.Version, "image_ref": input.ImageRef, "note": input.Note, "enabled": enabled,
|
|
})
|
|
})
|
|
app.Put("/api/browser-images/:version", serialized(func(c fiber.Ctx) error {
|
|
input := struct {
|
|
ImageRef string `json:"image_ref"`
|
|
Note string `json:"note"`
|
|
Enabled *bool `json:"enabled"`
|
|
}{}
|
|
if err := decodeHubJSON(c, &input); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
enabled := true
|
|
if input.Enabled != nil {
|
|
enabled = *input.Enabled
|
|
}
|
|
if err := store.UpdateImage(c.Context(), hub.Image{Version: c.Params("version"), ImageRef: input.ImageRef, Note: input.Note, Enabled: enabled}); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.SendStatus(fiber.StatusNoContent)
|
|
}))
|
|
app.Delete("/api/browser-images/:version", serialized(func(c fiber.Ctx) error {
|
|
if err := store.DeleteImage(c.Context(), c.Params("version")); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.SendStatus(fiber.StatusNoContent)
|
|
}))
|
|
|
|
app.Get("/api/gateways", func(c fiber.Ctx) error {
|
|
gateways, err := store.ListGateways(c.Context())
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.JSON(gateways)
|
|
})
|
|
app.Post("/api/gateways", func(c fiber.Ctx) error {
|
|
input := struct {
|
|
Name string `json:"name"`
|
|
Endpoint string `json:"endpoint"`
|
|
Token string `json:"token"`
|
|
}{}
|
|
if err := decodeHubJSON(c, &input); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
gateway, err := store.CreateGateway(c.Context(), input.Name, input.Endpoint, input.Token)
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.Status(fiber.StatusCreated).JSON(gateway)
|
|
})
|
|
app.Delete("/api/gateways/:name", func(c fiber.Ctx) error {
|
|
if err := store.DeleteGateway(c.Context(), c.Params("name")); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.SendStatus(fiber.StatusNoContent)
|
|
})
|
|
}
|
|
|
|
func listBrowsers(store hubStore) fiber.Handler {
|
|
return func(c fiber.Ctx) error {
|
|
envs, err := store.ListEnvs(c.Context())
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
gateways := map[string]hub.Gateway{}
|
|
containers := map[string]map[string]containerStatus{}
|
|
for _, env := range envs {
|
|
if _, ok := gateways[env.Gateway]; ok {
|
|
continue
|
|
}
|
|
gateway, err := store.GetGateway(c.Context(), env.Gateway)
|
|
if err != nil {
|
|
continue // 环境引用的网关被删除属于异常状态，列表仍要展示该环境
|
|
}
|
|
gateways[env.Gateway] = gateway
|
|
status, body, callErr := gatewayCall(c.Context(), gateway, http.MethodGet, "/v1/browsers", nil, 30*time.Second)
|
|
if callErr != nil || status != http.StatusOK {
|
|
continue
|
|
}
|
|
var list []containerStatus
|
|
if json.Unmarshal(body, &list) == nil {
|
|
byAlias := map[string]containerStatus{}
|
|
for _, container := range list {
|
|
byAlias[container.Alias] = container
|
|
}
|
|
containers[env.Gateway] = byAlias
|
|
}
|
|
}
|
|
views := make([]envView, 0, len(envs))
|
|
for _, env := range envs {
|
|
view := envView{Env: env, State: "missing", Status: "网关上不存在容器"}
|
|
if container, ok := containers[env.Gateway][env.Alias]; ok {
|
|
view.State, view.Status = container.State, container.Status
|
|
view.ContainerID, view.Endpoint = container.ID, container.Endpoint
|
|
}
|
|
views = append(views, view)
|
|
}
|
|
return c.JSON(views)
|
|
}
|
|
}
|
|
|
|
func createBrowser(store hubStore) fiber.Handler {
|
|
return func(c fiber.Ctx) error {
|
|
input := struct {
|
|
Alias string `json:"alias"`
|
|
Name string `json:"name"`
|
|
Gateway string `json:"gateway"`
|
|
ImageVersion string `json:"image_version"`
|
|
Fingerprint hub.Fingerprint `json:"fingerprint"`
|
|
}{}
|
|
if err := decodeHubJSON(c, &input); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
env := hub.Env{Alias: input.Alias, Name: input.Name, Gateway: input.Gateway, ImageVersion: input.ImageVersion, Fingerprint: input.Fingerprint}
|
|
if err := env.Fingerprint.Validate(); err != nil {
|
|
return c.Status(fiber.StatusBadRequest).JSON(map[string]string{"error": err.Error()})
|
|
}
|
|
gateway, err := store.GetGateway(c.Context(), env.Gateway)
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
imageRef, err := store.ImageRef(c.Context(), env.ImageVersion)
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
if err := store.CreateEnv(c.Context(), env); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
status, body, callErr := gatewayCall(c.Context(), gateway, http.MethodPost, "/v1/browsers", gatewayCreatePayload(env, imageRef), gatewayLongTimeout)
|
|
if callErr != nil || status >= http.StatusInternalServerError {
|
|
createErr := gatewayRejected(status, body)
|
|
if callErr != nil {
|
|
createErr = gatewayUnreachable(callErr)
|
|
}
|
|
exists, reconcileErr := reconcileGatewayBrowser(c.Context(), gateway, env.Alias)
|
|
if reconcileErr == nil && exists {
|
|
return c.Status(fiber.StatusCreated).JSON(map[string]string{"alias": env.Alias})
|
|
}
|
|
if reconcileErr != nil {
|
|
return hubError(c, gatewayFailure{status: http.StatusBadGateway, message: fmt.Sprintf(
|
|
"gateway create result unknown; environment retained for reconciliation: %v", createErr)})
|
|
}
|
|
_ = store.DeleteEnv(c.Context(), env.Alias)
|
|
return hubError(c, createErr)
|
|
}
|
|
if status != http.StatusCreated {
|
|
_ = store.DeleteEnv(c.Context(), env.Alias)
|
|
return hubError(c, gatewayRejected(status, body))
|
|
}
|
|
return c.Status(fiber.StatusCreated).JSON(map[string]string{"alias": env.Alias})
|
|
}
|
|
}
|
|
|
|
func browserAction(store hubStore) fiber.Handler {
|
|
return func(c fiber.Ctx) error {
|
|
switch c.Params("action") {
|
|
case "start", "stop":
|
|
return lifecycleAction(store, c)
|
|
case "upgrade":
|
|
return upgradeBrowser(store, c)
|
|
default:
|
|
return hubError(c, hub.ErrInvalid)
|
|
}
|
|
}
|
|
}
|
|
|
|
func lifecycleAction(store hubStore, c fiber.Ctx) error {
|
|
env, gateway, err := envWithGateway(c, store, c.Params("alias"))
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
status, body, callErr := gatewayCall(c.Context(), gateway, http.MethodPost, "/v1/browsers/"+env.Alias+"/"+c.Params("action"), nil, 30*time.Second)
|
|
if callErr != nil {
|
|
return hubError(c, gatewayUnreachable(callErr))
|
|
}
|
|
if status != http.StatusNoContent && status != http.StatusNotModified {
|
|
return hubError(c, gatewayRejected(status, body))
|
|
}
|
|
return c.SendStatus(fiber.StatusNoContent)
|
|
}
|
|
|
|
func upgradeBrowser(store hubStore, c fiber.Ctx) error {
|
|
var input struct {
|
|
Version string `json:"version"`
|
|
}
|
|
if err := decodeHubJSON(c, &input); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
env, gateway, err := envWithGateway(c, store, c.Params("alias"))
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
imageRef, err := store.ImageRef(c.Context(), input.Version)
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
// 先删容器（保留卷）；404 视为已删除，保证升级可重试。
|
|
status, body, callErr := gatewayCall(c.Context(), gateway, http.MethodDelete, "/v1/browsers/"+env.Alias, nil, 30*time.Second)
|
|
if callErr != nil {
|
|
return hubError(c, gatewayUnreachable(callErr))
|
|
}
|
|
if status != http.StatusNoContent && status != http.StatusNotFound {
|
|
return hubError(c, gatewayRejected(status, body))
|
|
}
|
|
env.ImageVersion = input.Version
|
|
status, body, callErr = gatewayCall(c.Context(), gateway, http.MethodPost, "/v1/browsers", gatewayCreatePayload(env, imageRef), gatewayLongTimeout)
|
|
if callErr != nil {
|
|
return hubError(c, gatewayUnreachable(callErr))
|
|
}
|
|
if status != http.StatusCreated {
|
|
return hubError(c, gatewayRejected(status, body))
|
|
}
|
|
if err := store.UpgradeEnv(c.Context(), env.Alias, input.Version); err != nil {
|
|
cleanupStatus, cleanupBody, cleanupErr := gatewayCall(c.Context(), gateway, http.MethodDelete, "/v1/browsers/"+env.Alias, nil, 30*time.Second)
|
|
if cleanupErr != nil || (cleanupStatus != http.StatusNoContent && cleanupStatus != http.StatusNotFound) {
|
|
if cleanupErr == nil {
|
|
cleanupErr = gatewayRejected(cleanupStatus, cleanupBody)
|
|
}
|
|
return hubError(c, gatewayFailure{status: http.StatusBadGateway, message: fmt.Sprintf(
|
|
"persist upgrade: %v; remove uncommitted container: %v", err, cleanupErr)})
|
|
}
|
|
return hubError(c, err)
|
|
}
|
|
return c.SendStatus(fiber.StatusNoContent)
|
|
}
|
|
|
|
func deleteBrowser(store hubStore) fiber.Handler {
|
|
return func(c fiber.Ctx) error {
|
|
env, gateway, err := envWithGateway(c, store, c.Params("alias"))
|
|
if err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
status, _, callErr := gatewayCall(c.Context(), gateway, http.MethodDelete, "/v1/browsers/"+env.Alias, nil, 30*time.Second)
|
|
if callErr != nil {
|
|
return hubError(c, gatewayUnreachable(callErr))
|
|
}
|
|
if status != http.StatusNoContent && status != http.StatusNotFound {
|
|
return hubError(c, errors.New("gateway refused container removal"))
|
|
}
|
|
if err := store.DeleteEnv(c.Context(), env.Alias); err != nil {
|
|
return hubError(c, err)
|
|
}
|
|
return c.SendStatus(fiber.StatusNoContent)
|
|
}
|
|
}
|
|
|
|
func envWithGateway(c fiber.Ctx, store hubStore, alias string) (hub.Env, hub.Gateway, error) {
|
|
env, err := store.GetEnv(c.Context(), alias)
|
|
if err != nil {
|
|
return hub.Env{}, hub.Gateway{}, err
|
|
}
|
|
gateway, err := store.GetGateway(c.Context(), env.Gateway)
|
|
if err != nil {
|
|
return hub.Env{}, hub.Gateway{}, err
|
|
}
|
|
return env, gateway, nil
|
|
}
|
|
|
|
func decodeHubJSON(c fiber.Ctx, destination any) error {
|
|
decoder := json.NewDecoder(bytes.NewReader(c.Body()))
|
|
decoder.DisallowUnknownFields()
|
|
if err := decoder.Decode(destination); err != nil {
|
|
return hub.ErrInvalid
|
|
}
|
|
if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
|
|
return hub.ErrInvalid
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func hubError(c fiber.Ctx, err error) error {
|
|
var failure gatewayFailure
|
|
if errors.As(err, &failure) {
|
|
return c.Status(failure.status).JSON(map[string]string{"error": failure.message})
|
|
}
|
|
status := fiber.StatusInternalServerError
|
|
message := "hub operation failed"
|
|
switch {
|
|
case errors.Is(err, hub.ErrInvalid):
|
|
status, message = fiber.StatusBadRequest, hub.ErrInvalid.Error()
|
|
case errors.Is(err, hub.ErrConflict):
|
|
status, message = fiber.StatusConflict, hub.ErrConflict.Error()
|
|
case errors.Is(err, hub.ErrNotFound):
|
|
status, message = fiber.StatusNotFound, hub.ErrNotFound.Error()
|
|
default:
|
|
return c.Status(status).JSON(map[string]string{"error": err.Error()})
|
|
}
|
|
return c.Status(status).JSON(map[string]string{"error": message})
|
|
}
|