rogee/creator-hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HH-744: converge Go service infrastructure (#6)

Browse Source
This commit is contained in:
rogee
2026-08-27 19:58:45 +08:00
parent f24ed999cb
commit c474455a06
8 changed files with 493 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/docker-gateway/main.go
+169 -93
View File
@@ -7,7 +7,6 @@ import (
"errors"
"fmt"
"io"
"log"
"net"
"net/http"
"net/url"
@@ -16,6 +15,11 @@ import (
"strconv"
"strings"
"time"
"github.com/gofiber/fiber/v3"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
)
const (
@@ -36,11 +40,18 @@ var networkNamePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$`)
var errInvalidRuntimeID = errors.New("invalid runtime id")
var errUnmanagedContainer = errors.New("refusing to operate on a container not owned by CreatorHub")
type config struct {
type browserConfig struct {
image string
network string
}
type serviceConfig struct {
listenAddr string
dockerSock string
network string
logLevel logrus.Level
}
type dockerClient struct {
baseURL string
client *http.Client
@@ -48,7 +59,7 @@ type dockerClient struct {
type gateway struct {
docker dockerClient
config config
config browserConfig
}
type createRequest struct {
@@ -66,55 +77,133 @@ type browser struct {
}
func main() {
transport := &http.Transport{
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, "unix", env("DOCKER_SOCKET", "/var/run/docker.sock"))
logrus.SetFormatter(&logrus.JSONFormatter{})
if err := newCommand().Execute(); err != nil {
logrus.WithField("service", "docker-gateway").WithError(err).Error("service stopped")
os.Exit(1)
}
}
func newCommand() *cobra.Command {
command := &cobra.Command{
Use: "docker-gateway",
Short: "Run the restricted CreatorHub Docker gateway",
Args: cobra.NoArgs,
SilenceErrors: true,
SilenceUsage: true,
RunE: func(command *cobra.Command, _ []string) error {
cfg, err := loadConfig()
if err != nil {
return err
}
logrus.SetLevel(cfg.logLevel)
return run(command, cfg)
},
}
return command
}
func loadConfig() (serviceConfig, error) {
v := viper.New()
v.SetDefault("listen_addr", ":8081")
v.SetDefault("docker_socket", "/var/run/docker.sock")
v.SetDefault("browser_network", "creatorhub_browser")
v.SetDefault("log_level", "info")
_ = v.BindEnv("listen_addr", "LISTEN_ADDR")
_ = v.BindEnv("docker_socket", "DOCKER_SOCKET")
_ = v.BindEnv("browser_network", "BROWSER_NETWORK")
_ = v.BindEnv("log_level", "LOG_LEVEL")
level, err := logrus.ParseLevel(v.GetString("log_level"))
if err != nil {
return serviceConfig{}, errors.New("LOG_LEVEL must be panic, fatal, error, warn, info, debug, or trace")
}
cfg := serviceConfig{
listenAddr: strings.TrimSpace(v.GetString("listen_addr")),
dockerSock: strings.TrimSpace(v.GetString("docker_socket")),
network: strings.TrimSpace(v.GetString("browser_network")),
logLevel: level,
}
if cfg.listenAddr == "" {
return serviceConfig{}, errors.New("LISTEN_ADDR must not be empty")
}
if err := validateListenAddr(cfg.listenAddr); err != nil {
return serviceConfig{}, err
}
if cfg.dockerSock == "" {
return serviceConfig{}, errors.New("DOCKER_SOCKET must not be empty")
}
if err := validateBrowserNetwork(cfg.network); err != nil {
return serviceConfig{}, err
}
return cfg, nil
}
func validateListenAddr(addr string) error {
_, port, err := net.SplitHostPort(addr)
if err != nil {
return errors.New("LISTEN_ADDR must be a host:port address")
}
number, err := strconv.Atoi(port)
if err != nil || number < 1 || number > 65535 {
return errors.New("LISTEN_ADDR port must be 1..65535")
}
return nil
}
func run(command *cobra.Command, cfg serviceConfig) error {
transport := &http.Transport{
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, "unix", cfg.dockerSock)
},
}
defer transport.CloseIdleConnections()
docker := dockerClient{
baseURL: "http://docker/v1.43",
client: &http.Client{Transport: transport, Timeout: 30 * time.Second},
}
network := env("BROWSER_NETWORK", "creatorhub_browser")
if err := docker.ensureInternalNetwork(network); err != nil {
log.Fatal(err)
if err := docker.ensureInternalNetwork(cfg.network); err != nil {
return err
}
handler := newGateway(docker, config{image: browserImage, network: network})
server := &http.Server{
Addr: env("LISTEN_ADDR", ":8081"),
Handler: handler,
ReadHeaderTimeout: 5 * time.Second,
IdleTimeout: 60 * time.Second,
}
log.Printf("restricted Docker gateway listening on %s", server.Addr)
log.Fatal(server.ListenAndServe())
}
func newGateway(client dockerClient, cfg config) http.Handler {
api := gateway{docker: client, config: cfg}
mux := http.NewServeMux()
mux.HandleFunc("GET /healthz", func(response http.ResponseWriter, _ *http.Request) {
response.WriteHeader(http.StatusNoContent)
logrus.WithFields(logrus.Fields{
"service": "docker-gateway",
"listen_addr": cfg.listenAddr,
"network": cfg.network,
}).Info("service starting")
return newGateway(docker, browserConfig{image: browserImage, network: cfg.network}).Listen(cfg.listenAddr, fiber.ListenConfig{
GracefulContext: command.Context(),
DisableStartupMessage: true,
})
mux.HandleFunc("GET /v1/browsers", api.list)
mux.HandleFunc("POST /v1/browsers", api.create)
mux.HandleFunc("POST /v1/browsers/{id}/{action}", api.changeState)
mux.HandleFunc("DELETE /v1/browsers/{id}", api.remove)
return mux
}
func (api gateway) list(response http.ResponseWriter, _ *http.Request) {
func newGateway(client dockerClient, cfg browserConfig) *fiber.App {
api := gateway{docker: client, config: cfg}
app := fiber.New(fiber.Config{
AppName: "CreatorHub Docker gateway",
BodyLimit: 1 << 20,
ReadTimeout: 5 * time.Second,
IdleTimeout: 60 * time.Second,
})
app.Get("/healthz", func(c fiber.Ctx) error {
c.Status(fiber.StatusNoContent)
return nil
})
app.Get("/v1/browsers", api.list)
app.Post("/v1/browsers", api.create)
app.Post("/v1/browsers/:id/:action", api.changeState)
app.Delete("/v1/browsers/:id", api.remove)
return app
}
func (api gateway) list(c fiber.Ctx) error {
filters, _ := json.Marshal(map[string][]string{"label": {managedLabel + "=true"}})
result, err := api.docker.request(http.MethodGet, "/containers/json?all=1&filters="+url.QueryEscape(string(filters)), nil)
if err != nil {
writeError(response, http.StatusBadGateway, err)
return
return writeError(c, http.StatusBadGateway, err)
}
defer result.Body.Close()
if result.StatusCode != http.StatusOK {
forwardDockerError(response, result)
return
return forwardDockerError(c, result)
}
var containers []struct {
@@ -124,8 +213,7 @@ func (api gateway) list(response http.ResponseWriter, _ *http.Request) {
Labels map[string]string `json:"Labels"`
}
if err := json.NewDecoder(result.Body).Decode(&containers); err != nil {
writeError(response, http.StatusBadGateway, fmt.Errorf("decode Docker response: %w", err))
return
return writeError(c, http.StatusBadGateway, fmt.Errorf("decode Docker response: %w", err))
}
browsers := make([]browser, 0, len(containers))
@@ -143,20 +231,18 @@ func (api gateway) list(response http.ResponseWriter, _ *http.Request) {
Endpoint: "http://" + namePrefix + id + ":9222",
})
}
writeJSON(response, http.StatusOK, browsers)
return writeJSON(c, http.StatusOK, browsers)
}
func (api gateway) create(response http.ResponseWriter, request *http.Request) {
func (api gateway) create(c fiber.Ctx) error {
var input createRequest
decoder := json.NewDecoder(http.MaxBytesReader(response, request.Body, 1<<20))
decoder := json.NewDecoder(bytes.NewReader(c.Body()))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&input); err != nil {
writeError(response, http.StatusBadRequest, errors.New("body must contain only name and numeric seed"))
return
return writeError(c, http.StatusBadRequest, errors.New("body must contain only name and numeric seed"))
}
if !runtimeIDPattern.MatchString(input.Name) || input.Seed < 1 || input.Seed > 2147483647 {
writeError(response, http.StatusBadRequest, errors.New("name must match [a-z0-9][a-z0-9-]{0,31} and seed must be 1..2147483647"))
return
return writeError(c, http.StatusBadRequest, errors.New("name must match [a-z0-9][a-z0-9-]{0,31} and seed must be 1..2147483647"))
}
pidsLimit := int64(512)
@@ -194,39 +280,33 @@ func (api gateway) create(response http.ResponseWriter, request *http.Request) {
}
result, err := api.docker.request(http.MethodPost, "/containers/create?name="+url.QueryEscape(namePrefix+input.Name), payload)
if err != nil {
writeError(response, http.StatusBadGateway, err)
return
return writeError(c, http.StatusBadGateway, err)
}
defer result.Body.Close()
if result.StatusCode != http.StatusCreated {
forwardDockerError(response, result)
return
return forwardDockerError(c, result)
}
var created struct {
ID string `json:"Id"`
}
if err := json.NewDecoder(result.Body).Decode(&created); err != nil || created.ID == "" {
writeError(response, http.StatusBadGateway, errors.New("Docker returned an invalid container id"))
return
return writeError(c, http.StatusBadGateway, errors.New("Docker returned an invalid container id"))
}
if err := api.docker.expect(http.MethodPost, "/containers/"+url.PathEscape(created.ID)+"/start", nil, http.StatusNoContent, http.StatusNotModified); err != nil {
cleanupErr := api.docker.expect(http.MethodDelete, "/containers/"+url.PathEscape(created.ID)+"?force=1&v=0", nil, http.StatusNoContent)
if cleanupErr != nil {
writeError(response, http.StatusBadGateway, fmt.Errorf("container did not start and cleanup failed: %w; cleanup: %v", err, cleanupErr))
return
return writeError(c, http.StatusBadGateway, fmt.Errorf("container did not start and cleanup failed: %w; cleanup: %v", err, cleanupErr))
}
writeError(response, http.StatusBadGateway, fmt.Errorf("container did not start and was removed while preserving its Profile volume: %w", err))
return
return writeError(c, http.StatusBadGateway, fmt.Errorf("container did not start and was removed while preserving its Profile volume: %w", err))
}
writeJSON(response, http.StatusCreated, map[string]string{"id": created.ID, "name": input.Name})
return writeJSON(c, http.StatusCreated, map[string]string{"id": created.ID, "name": input.Name})
}
func (api gateway) changeState(response http.ResponseWriter, request *http.Request) {
id := request.PathValue("id")
action := request.PathValue("action")
func (api gateway) changeState(c fiber.Ctx) error {
id := c.Params("id")
action := c.Params("action")
if err := api.requireManaged(id); err != nil {
writeError(response, statusFor(err), err)
return
return writeError(c, statusFor(err), err)
}
var path string
switch action {
@@ -235,28 +315,26 @@ func (api gateway) changeState(response http.ResponseWriter, request *http.Reque
case "stop":
path = "/containers/" + url.PathEscape(namePrefix+id) + "/stop?t=10"
default:
writeError(response, http.StatusNotFound, errors.New("unknown action"))
return
return writeError(c, http.StatusNotFound, errors.New("unknown action"))
}
if err := api.docker.expect(http.MethodPost, path, nil, http.StatusNoContent, http.StatusNotModified); err != nil {
writeError(response, http.StatusBadGateway, err)
return
return writeError(c, http.StatusBadGateway, err)
}
response.WriteHeader(http.StatusNoContent)
c.Status(http.StatusNoContent)
return nil
}
func (api gateway) remove(response http.ResponseWriter, request *http.Request) {
id := request.PathValue("id")
func (api gateway) remove(c fiber.Ctx) error {
id := c.Params("id")
if err := api.requireManaged(id); err != nil {
writeError(response, statusFor(err), err)
return
return writeError(c, statusFor(err), err)
}
path := "/containers/" + url.PathEscape(namePrefix+id) + "?force=1&v=0"
if err := api.docker.expect(http.MethodDelete, path, nil, http.StatusNoContent); err != nil {
writeError(response, http.StatusBadGateway, err)
return
return writeError(c, http.StatusBadGateway, err)
}
response.WriteHeader(http.StatusNoContent)
c.Status(http.StatusNoContent)
return nil
}
func (api gateway) requireManaged(id string) error {
@@ -323,11 +401,8 @@ func (docker dockerClient) expect(method, path string, payload any, allowed ...i
}
func (docker dockerClient) ensureInternalNetwork(name string) error {
if !networkNamePattern.MatchString(name) {
return errors.New("BROWSER_NETWORK is invalid")
}
if name == controlNetworkName {
return errors.New("BROWSER_NETWORK must not reuse the control network")
if err := validateBrowserNetwork(name); err != nil {
return err
}
response, err := docker.request(http.MethodGet, "/networks/"+url.PathEscape(name), nil)
if err != nil {
@@ -370,6 +445,16 @@ func (docker dockerClient) ensureInternalNetwork(name string) error {
}, http.StatusCreated)
}
func validateBrowserNetwork(name string) error {
if !networkNamePattern.MatchString(name) {
return errors.New("BROWSER_NETWORK is invalid")
}
if name == controlNetworkName {
return errors.New("BROWSER_NETWORK must not reuse the control network")
}
return nil
}
func statusFor(err error) int {
switch {
case errors.Is(err, errInvalidRuntimeID):
@@ -383,28 +468,19 @@ func statusFor(err error) int {
}
}
func forwardDockerError(response http.ResponseWriter, result *http.Response) {
func forwardDockerError(c fiber.Ctx, result *http.Response) error {
message, _ := io.ReadAll(io.LimitReader(result.Body, 4096))
status := http.StatusBadGateway
if result.StatusCode == http.StatusConflict {
status = http.StatusConflict
}
writeError(response, status, fmt.Errorf("Docker returned %s: %s", result.Status, strings.TrimSpace(string(message))))
return writeError(c, status, fmt.Errorf("Docker returned %s: %s", result.Status, strings.TrimSpace(string(message))))
}
func writeError(response http.ResponseWriter, status int, err error) {
writeJSON(response, status, map[string]string{"error": err.Error()})
func writeError(c fiber.Ctx, status int, err error) error {
return writeJSON(c, status, map[string]string{"error": err.Error()})
}
func writeJSON(response http.ResponseWriter, status int, value any) {
response.Header().Set("Content-Type", "application/json")
response.WriteHeader(status)
_ = json.NewEncoder(response).Encode(value)
}
func env(name, fallback string) string {
if value := os.Getenv(name); value != "" {
return value
}
return fallback
func writeJSON(c fiber.Ctx, status int, value any) error {
return c.Status(status).JSON(value)
}
cmd/docker-gateway/main_test.go
+31 -12
View File
@@ -12,6 +12,8 @@ import (
"strings"
"testing"
"time"
"github.com/gofiber/fiber/v3/middleware/adaptor"
)
func TestGatewayCreatesConstrainedBrowser(t *testing.T) {
@@ -35,13 +37,13 @@ func TestGatewayCreatesConstrainedBrowser(t *testing.T) {
}))
defer docker.Close()
handler := newGateway(dockerClient{baseURL: docker.URL + "/v1.43", client: docker.Client()}, config{
handler := newGateway(dockerClient{baseURL: docker.URL + "/v1.43", client: docker.Client()}, browserConfig{
image: "registry.example/browser:1.2.3",
network: "creatorhub_browser",
})
request := httptest.NewRequest(http.MethodPost, "/v1/browsers", strings.NewReader(`{"name":"account-a","seed":1000}`))
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
adaptor.FiberApp(handler).ServeHTTP(response, request)
if response.Code != http.StatusCreated {
t.Fatalf("expected 201, got %d: %s", response.Code, response.Body.String())
@@ -66,6 +68,16 @@ func TestGatewayCreatesConstrainedBrowser(t *testing.T) {
}
}
func TestGatewayRejectsOversizedCreateRequest(t *testing.T) {
handler := newGateway(dockerClient{}, browserConfig{})
request := httptest.NewRequest(http.MethodPost, "/v1/browsers", strings.NewReader(strings.Repeat("x", (1<<20)+1)))
response := httptest.NewRecorder()
adaptor.FiberApp(handler).ServeHTTP(response, request)
if response.Code != http.StatusRequestEntityTooLarge {
t.Fatalf("expected 413 for oversized body, status=%d body=%s", response.Code, response.Body.String())
}
}
func TestGatewayRemovesFailedContainerAndPreservesProfile(t *testing.T) {
removed := false
docker := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
@@ -84,10 +96,10 @@ func TestGatewayRemovesFailedContainerAndPreservesProfile(t *testing.T) {
}))
defer docker.Close()
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, config{image: browserImage, network: "creatorhub_browser"})
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, browserConfig{image: browserImage, network: "creatorhub_browser"})
request := httptest.NewRequest(http.MethodPost, "/v1/browsers", strings.NewReader(`{"name":"account-a","seed":1000}`))
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
adaptor.FiberApp(handler).ServeHTTP(response, request)
if response.Code != http.StatusBadGateway || !removed {
t.Fatalf("expected failed container cleanup with preserved volume, status=%d removed=%v body=%s", response.Code, removed, response.Body.String())
@@ -103,9 +115,9 @@ func TestGatewayListsBrowsers(t *testing.T) {
}))
defer docker.Close()
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, config{})
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, browserConfig{})
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/v1/browsers", nil))
adaptor.FiberApp(handler).ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/v1/browsers", nil))
var browsers []browser
if response.Code != http.StatusOK || json.NewDecoder(response.Body).Decode(&browsers) != nil || len(browsers) != 1 || browsers[0].Name != "account-a" {
@@ -137,9 +149,9 @@ func TestGatewayLifecycle(t *testing.T) {
}))
defer docker.Close()
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, config{})
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, browserConfig{})
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(test.method, test.path, nil))
adaptor.FiberApp(handler).ServeHTTP(response, httptest.NewRequest(test.method, test.path, nil))
if response.Code != http.StatusNoContent {
t.Fatalf("expected 204, got %d: %s", response.Code, response.Body.String())
}
@@ -153,9 +165,9 @@ func TestGatewayMapsDockerServiceFailureToBadGateway(t *testing.T) {
}))
defer docker.Close()
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, config{})
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, browserConfig{})
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodDelete, "/v1/browsers/account-a", nil))
adaptor.FiberApp(handler).ServeHTTP(response, httptest.NewRequest(http.MethodDelete, "/v1/browsers/account-a", nil))
if response.Code != http.StatusBadGateway {
t.Fatalf("expected 502 for Docker failure, got %d: %s", response.Code, response.Body.String())
}
@@ -174,10 +186,10 @@ func TestGatewayRefusesUnmanagedContainer(t *testing.T) {
}))
defer docker.Close()
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, config{})
handler := newGateway(dockerClient{baseURL: docker.URL, client: docker.Client()}, browserConfig{})
request := httptest.NewRequest(http.MethodDelete, "/v1/browsers/foreign", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
adaptor.FiberApp(handler).ServeHTTP(response, request)
if response.Code != http.StatusForbidden || deleted {
t.Fatalf("expected unmanaged container to be rejected, status=%d deleted=%v", response.Code, deleted)
@@ -288,6 +300,13 @@ func TestEnsureInternalNetworkRejectsControlNetwork(t *testing.T) {
}
}
func TestLoadConfigRejectsControlNetwork(t *testing.T) {
t.Setenv("BROWSER_NETWORK", controlNetworkName)
if _, err := loadConfig(); err == nil {
t.Fatal("expected control network configuration to be rejected")
}
}
func TestDockerIntegrationBlocksCrossContainerTraffic(t *testing.T) {
if os.Getenv("CREATORHUB_DOCKER_INTEGRATION") != "1" {
t.Skip("set CREATORHUB_DOCKER_INTEGRATION=1 to run")