feat(distributed): 支持单主控多节点部署和停服迁移

复用现有 k3s 调度和 Job 生命周期,增加 worker 接入与批准、受保护节点身份、归档传输、持久迁移锁及活动 PVC 切换;同步管理员 API、CLI、面板和隔离规则。分布式模式默认关闭,保持单机兼容。

验证:Go 全量测试与 vet;面板 874 个测试、lint/build;Linux VM 安装器测试、清单服务端 dry-run、网络命名空间防火墙实测。A/B/C 三机 WireGuard、Velocity 和迁移验收仍待完成。
This commit is contained in:
Lemon-miaow committed 2026-10-01 19:48:37 +08:00
1 parent 26e817f2c9
commit adf563ffe3
77 files changed
+5224 -73

No files matched your search

+10 -4
View File
@@ -29,10 +29,11 @@ import (
// API holds the dependencies shared by every handler.
type API struct {
Repo Repo
Cluster Cluster
Internal InternalAuth
External ExternalAuth
Distribution Distribution
Repo Repo
Cluster Cluster
Internal InternalAuth
External ExternalAuth
// Builder is the image build subsystem (spec §16). It is optional: when nil
// the /images routes report 503 rather than 404, so the admin boundary is
@@ -721,6 +722,11 @@ func (a *API) externalAPIRoutes() []apiRoute {
// Zero-Trust path, unlike the app-tier /me/servers. A path distinct from the
// internal velocity GET /api/v1/servers on purpose: the parity test forbids one
// {method, path} from carrying both the service and admin tiers.
{Method: "GET", Pattern: "/api/v1/nodes", Admin: true, h: a.handleNodes},
{Method: "GET", Pattern: "/api/v1/servers/{name}/migrations", Admin: true, h: a.handleMigrationStatus},
{Method: "POST", Pattern: "/api/v1/servers/{name}/migrations", Admin: true, h: a.handleMigration},
{Method: "GET", Pattern: "/api/v1/servers/{name}/migrations/{id}", Admin: true, h: a.handleMigrationStatus},
{Method: "POST", Pattern: "/api/v1/servers/{name}/migrations/{id}/retry", Admin: true, h: a.handleMigrationRetry},
{Method: "GET", Pattern: "/api/v1/fleet", Admin: true, h: a.handleFleet},
// Image build + whitelist (spec §16, §15). Every route is admin-tier: a build
// is build-time RCE against the cluster, so submission requires the admin
+2
View File
@@ -13,6 +13,7 @@ import (
// patch (spec §7 PATCH /servers/{name}) — never the business-layer fields, which
// live in Postgres (spec §22).
type ServerInfo struct {
NodeName string `json:"nodeName,omitempty"`
Name string `json:"name"`
Subdomain string `json:"subdomain"`
Phase string `json:"phase"`
@@ -64,6 +65,7 @@ type ServerInfo struct {
// no free-form YAML path — every field is a typed, validated value. A created
// server starts DesiredState=Stopped and unowned (claimed later, spec §9.3).
type CreateServerInput struct {
NodeName string
Name string
Subdomain string
DisplayName string
+122
View File
@@ -0,0 +1,122 @@
package api
import (
"context"
"errors"
"net/http"
"felis.lolicon.best/internal/distributed"
"felis.lolicon.best/internal/naming"
apierrors "k8s.io/apimachinery/pkg/api/errors"
)
type Distribution interface {
Nodes(context.Context) ([]distributed.Node, error)
ValidateNode(context.Context, string) error
BeginMigration(context.Context, string, string, string) (distributed.Operation, error)
Migration(context.Context, string, string) (distributed.Operation, error)
RetryMigration(context.Context, string, string) (distributed.Operation, error)
}
func (a *API) distributedReady(w http.ResponseWriter, r *http.Request) bool {
if a.Distribution == nil {
writeError(w, r, newError(503, "distributed_unavailable", "distributed deployment is not configured"))
return false
}
return true
}
func (a *API) handleNodes(w http.ResponseWriter, r *http.Request) {
if !a.distributedReady(w, r) {
return
}
nodes, err := a.Distribution.Nodes(r.Context())
if err != nil {
writeError(w, r, err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"nodes": nodes})
}
func (a *API) handleMigration(w http.ResponseWriter, r *http.Request) {
if !a.distributedReady(w, r) {
return
}
server := r.PathValue("name")
if err := naming.ValidateServerName(server); err != nil {
writeError(w, r, newError(400, "bad_name", "%v", err))
return
}
rec, err := a.Repo.ServerByName(r.Context(), server)
if err != nil {
a.writeLookupError(w, r, err)
return
}
if rec.Retire != nil {
writeError(w, r, errServerRetiring)
return
}
var body struct {
TargetNode string `json:"targetNode"`
}
if err := decodeJSON(w, r, &body); err != nil {
writeError(w, r, err)
return
}
if err := a.Distribution.ValidateNode(r.Context(), body.TargetNode); err != nil {
writeError(w, r, newError(400, "bad_node", "%v", err))
return
}
op, err := a.Distribution.BeginMigration(r.Context(), server, body.TargetNode, rec.OwnerID)
if err != nil {
a.writeMigrationError(w, r, err)
return
}
writeJSON(w, http.StatusAccepted, op)
}
func (a *API) handleMigrationStatus(w http.ResponseWriter, r *http.Request) {
if !a.distributedReady(w, r) {
return
}
op, err := a.Distribution.Migration(r.Context(), r.PathValue("name"), r.PathValue("id"))
if err != nil {
a.writeMigrationError(w, r, err)
return
}
writeJSON(w, http.StatusOK, op)
}
func (a *API) handleMigrationRetry(w http.ResponseWriter, r *http.Request) {
if !a.distributedReady(w, r) {
return
}
rec, err := a.Repo.ServerByName(r.Context(), r.PathValue("name"))
if err != nil {
a.writeLookupError(w, r, err)
return
}
if rec.Retire != nil {
writeError(w, r, errServerRetiring)
return
}
op, err := a.Distribution.RetryMigration(r.Context(), r.PathValue("name"), r.PathValue("id"))
if err != nil {
a.writeMigrationError(w, r, err)
return
}
writeJSON(w, http.StatusAccepted, op)
}
func (a *API) writeMigrationError(w http.ResponseWriter, r *http.Request, err error) {
switch {
case errors.Is(err, distributed.ErrBusy):
writeError(w, r, newError(409, "migration_busy", "%v", err))
case errors.Is(err, distributed.ErrNotFound), apierrors.IsNotFound(err):
writeError(w, r, newError(404, "not_found", "%v", err))
case apierrors.IsConflict(err):
writeError(w, r, newError(409, "conflict", "retry after refreshing migration status"))
default:
writeError(w, r, err)
}
}
+77
View File
@@ -0,0 +1,77 @@
package api
import (
"context"
"errors"
"testing"
"felis.lolicon.best/internal/distributed"
)
type fakeDistribution struct {
calls int
target, owner string
}
func (d *fakeDistribution) Nodes(context.Context) ([]distributed.Node, error) {
d.calls++
return []distributed.Node{{Name: "b", Ready: true, Approved: true, Addresses: []string{}}}, nil
}
func (d *fakeDistribution) ValidateNode(_ context.Context, name string) error {
if name != "b" {
return errors.New("not approved")
}
return nil
}
func (d *fakeDistribution) BeginMigration(_ context.Context, name, target, owner string) (distributed.Operation, error) {
d.calls++
d.target = target
d.owner = owner
return distributed.Operation{ID: "op", Server: name, State: "backing_up"}, nil
}
func (d *fakeDistribution) Migration(context.Context, string, string) (distributed.Operation, error) {
d.calls++
return distributed.Operation{ID: "op", State: "failed"}, nil
}
func (d *fakeDistribution) RetryMigration(context.Context, string, string) (distributed.Operation, error) {
d.calls++
return distributed.Operation{ID: "op", State: "restoring"}, nil
}
func TestDistributedRoutesAreAdministratorOnly(t *testing.T) {
for _, role := range []string{"user", "admin"} {
t.Run(role, func(t *testing.T) {
repo := newFakeRepo()
repo.byName["survival"] = &ServerRecord{Name: "survival", OwnerID: "owner"}
a := newTestAPI(repo, newFakeCluster())
a.External = staticExternal{p: &Principal{UserID: "owner", Role: role, ViaAdminAccess: role == "admin"}}
d := &fakeDistribution{}
a.Distribution = d
for _, tc := range []struct {
method, path, body string
status int
}{
{"GET", "/api/v1/nodes", "", 200},
{"POST", "/api/v1/servers/survival/migrations", `{"targetNode":"b"}`, 202},
{"GET", "/api/v1/servers/survival/migrations", "", 200},
{"GET", "/api/v1/servers/survival/migrations/op", "", 200},
{"POST", "/api/v1/servers/survival/migrations/op/retry", "", 202},
} {
w := do(a.ExternalHandler(), tc.method, tc.path, tc.body, jsonHeader)
want := tc.status
if role == "user" {
want = 403
}
if w.Code != want {
t.Fatalf("%s: %d %s", tc.path, w.Code, w.Body.String())
}
}
if role == "user" && d.calls != 0 {
t.Fatal("owner reached cluster-wide operations")
}
if role == "admin" && (d.target != "b" || d.owner != "owner") {
t.Fatal("wrong migration scope")
}
})
}
}
+11
View File
@@ -406,6 +406,7 @@ type fleetServerView struct {
// value and decodeJSON rejects unknown fields, so a caller can never smuggle
// free-form YAML or raw CRD fields through this endpoint.
type createServerRequest struct {
NodeName string `json:"nodeName,omitempty"`
Name string `json:"name"`
Subdomain string `json:"subdomain"`
DisplayName string `json:"displayName,omitempty"`
@@ -445,6 +446,15 @@ func (a *API) handleCreateServer(w http.ResponseWriter, r *http.Request) {
return
}
if a.Distribution != nil {
if err := a.Distribution.ValidateNode(r.Context(), body.NodeName); err != nil {
writeError(w, r, newError(400, "bad_node", "select an approved worker: %v", err))
return
}
} else if body.NodeName != "" {
writeError(w, r, newError(400, "bad_node", "node selection requires distributed deployment"))
return
}
// Server name and subdomain both obey the §22 portability rule and the
// reservation list.
if err := naming.ValidateServerName(body.Name); err != nil {
@@ -572,6 +582,7 @@ func (a *API) handleCreateServer(w http.ResponseWriter, r *http.Request) {
}
in := CreateServerInput{
NodeName: body.NodeName,
Name: body.Name,
Subdomain: body.Subdomain,
DisplayName: displayName,
+59 -9
View File
@@ -3,11 +3,13 @@ package api
import (
"context"
"errors"
"strings"
"time"
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/maintenance"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -32,8 +34,10 @@ import (
// the direct client c, so a write never works from a copy the watch has not caught
// up with yet.
type K8sCluster struct {
c client.Client
namespace string
distributed bool
controller string
c client.Client
namespace string
// servers serves the fleet-wide reads; nil means c.
servers client.Reader
// synced reports whether servers has its first full list; nil means no cache.
@@ -101,15 +105,28 @@ func (k *K8sCluster) GetServer(ctx context.Context, name string) (*ServerInfo, e
// and restore Jobs mount), so existence here is exactly existence at Job mount
// time. NotFound is (false, nil): the caller refuses with a specific 409.
func (k *K8sCluster) WorldVolumeExists(ctx context.Context, name string) (bool, error) {
var pvc corev1.PersistentVolumeClaim
err := k.c.Get(ctx, types.NamespacedName{Namespace: k.namespace, Name: naming.WorldPVCName(name)}, &pvc)
if apierrors.IsNotFound(err) {
return false, nil
}
if err != nil {
var ms v1alpha1.MinecraftServer
err := k.getServer(ctx, name, &ms)
claim := naming.WorldPVCName(name)
if err == nil {
claim = ms.WorldPVC()
} else if !errors.Is(err, ErrNotFound) {
return false, err
}
return true, nil
var pvc corev1.PersistentVolumeClaim
if err := k.c.Get(ctx, types.NamespacedName{Namespace: k.namespace, Name: claim}, &pvc); err == nil {
return true, nil
} else if !apierrors.IsNotFound(err) {
return false, err
}
if ms.Name == "" {
var retained corev1.PersistentVolumeClaimList
if err := k.c.List(ctx, &retained, client.InNamespace(k.namespace), client.MatchingLabels{v1alpha1.LabelServer: name}); err != nil {
return false, err
}
return len(retained.Items) > 0, nil
}
return false, nil
}
// PodImages lists the image of every container and init container of every pod
@@ -183,6 +200,7 @@ func (k *K8sCluster) CreateServer(ctx context.Context, in CreateServerInput) err
Namespace: k.namespace,
},
Spec: v1alpha1.MinecraftServerSpec{
NodeName: in.NodeName,
Subdomain: in.Subdomain,
DisplayName: in.DisplayName,
Image: in.Image,
@@ -262,6 +280,22 @@ func (k *K8sCluster) startWith(ctx context.Context, name string, retryFailed boo
if err := k.getServer(ctx, name, &ms); err != nil {
return err
}
node := ms.Spec.NodeName
if node == "" {
node = ms.Status.NodeName
}
if node == "" {
node = k.controller
}
if k.distributed && node != "" {
var n corev1.Node
if err := k.c.Get(ctx, types.NamespacedName{Name: node}, &n); err != nil {
return err
}
if !placement.Admitted(&n, k.controller) || n.Spec.Unschedulable {
return newError(503, "node_unavailable", "execution node is offline")
}
}
kind, held, err := k.maintenanceHolder(ctx, &ms)
if err != nil {
return err
@@ -335,6 +369,9 @@ func (k *K8sCluster) ReleaseMaintenance(ctx context.Context, name string) error
}
return err
}
if value := ms.Annotations[maintenance.Annotation]; strings.HasPrefix(value, maintenance.KindMigration+"@") {
return nil
}
if _, ok := ms.Annotations[maintenance.Annotation]; !ok {
return nil
}
@@ -445,8 +482,13 @@ func serverInfo(ms *v1alpha1.MinecraftServer) *ServerInfo {
memStr = limit.String()
}
node := ms.Spec.NodeName
if node == "" {
node = ms.Status.NodeName
}
return &ServerInfo{
Name: ms.Name,
NodeName: node,
Subdomain: ms.Spec.Subdomain,
Phase: string(ms.Status.Phase),
Ready: ms.Status.Ready,
@@ -483,3 +525,11 @@ func idleStopSeconds(ms *v1alpha1.MinecraftServer) int32 {
}
return ms.Spec.Idle.EmptySecondsBeforeStop
}
func (k *K8sCluster) WithDistributed(enabled bool, controller ...string) *K8sCluster {
k.distributed = enabled
if len(controller) > 0 {
k.controller = controller[0]
}
return k
}
+34
View File
@@ -7,9 +7,12 @@ import (
"sort"
"strings"
"testing"
"time"
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/maintenance"
"felis.lolicon.best/internal/naming"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -19,6 +22,37 @@ import (
"sigs.k8s.io/controller-runtime/pkg/client/fake"
)
func TestPersistentMigrationBlocksWakeAndWorldOperations(t *testing.T) {
scheme := runtime.NewScheme()
v1alpha1.AddToScheme(scheme)
corev1.AddToScheme(scheme)
batchv1.AddToScheme(scheme)
s := &v1alpha1.MinecraftServer{ObjectMeta: metav1.ObjectMeta{Name: "survival", Namespace: "minecraft", Annotations: map[string]string{maintenance.Annotation: maintenance.LockValue(maintenance.KindMigration, time.Now().Add(-24*time.Hour))}}, Spec: v1alpha1.MinecraftServerSpec{DesiredState: v1alpha1.DesiredStopped}, Status: v1alpha1.MinecraftServerStatus{Phase: v1alpha1.PhaseStopped}}
s.Spec.Storage.ClaimName = "world-survival-migrated"
c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(s, &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: s.WorldPVC(), Namespace: s.Namespace}}).Build()
k := NewK8sCluster(c, s.Namespace)
ctx := context.Background()
if exists, err := k.WorldVolumeExists(ctx, s.Name); err != nil || !exists {
t.Fatal("active volume ignored", err)
}
if err := k.SetDesiredState(ctx, s.Name, v1alpha1.DesiredRunning); !errors.Is(err, ErrMaintenanceInProgress) {
t.Fatal("migration admitted wake", err)
}
for _, kind := range []string{maintenance.KindBackup, maintenance.KindFileWrite, maintenance.KindReap} {
if err := k.AcquireMaintenance(ctx, s.Name, kind); !errors.Is(err, ErrMaintenanceInProgress) {
t.Fatal("migration admitted", kind, err)
}
}
if err := k.ReleaseMaintenance(ctx, s.Name); err != nil {
t.Fatal(err)
}
var current v1alpha1.MinecraftServer
c.Get(ctx, types.NamespacedName{Namespace: s.Namespace, Name: s.Name}, &current)
if current.Annotations[maintenance.Annotation] == "" {
t.Fatal("normal release cleared persistent migration lock")
}
}
// K8sCluster is documented as integration-tested against a live cluster rather
// than covered by the hermetic suite, and for most of it that is the right call —
// merge-patch semantics are not worth faking. This one test departs from it
+3
View File
@@ -202,6 +202,9 @@ func (k *K8sJobStatus) PendingRestoreChains(ctx context.Context) ([]RestoreChain
snapshot = ChainSnapshotFailed
}
}
if j.Labels["felis.lolicon.best/archive-pending"] == "true" && snapshot == ChainSnapshotSucceeded {
snapshot = ChainSnapshotRunning
}
out = append(out, RestoreChain{
Job: j.Name,
Server: j.Labels[jobServerLabel],
@@ -1,6 +1,7 @@
package v1alpha1
import (
"felis.lolicon.best/internal/naming"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -159,6 +160,9 @@ type MinecraftServerSpec struct {
// ReaperExempt opts this server out of the world reaper entirely (spec §18).
ReaperExempt bool `json:"reaperExempt,omitempty"`
// NodeName is an administrator-approved execution node; empty preserves legacy placement.
NodeName string `json:"nodeName,omitempty"`
// DesiredState toggles the server up or down (default Stopped).
// +kubebuilder:validation:Enum=Running;Stopped
DesiredState DesiredState `json:"desiredState,omitempty"`
@@ -245,6 +249,8 @@ type SecretKeyRef struct {
// StorageSpec configures the world PVC (spec §4 spec.storage).
type StorageSpec struct {
// ClaimName is managed internally by stopped-world migration, never by a public spec patch.
ClaimName string `json:"claimName,omitempty"`
// Size is the requested PVC capacity (e.g. "10Gi").
Size string `json:"size,omitempty"`
// StorageClassName selects the StorageClass; empty uses the default.
@@ -317,6 +323,7 @@ type IdleSpec struct {
// MinecraftServerStatus is the observed state (spec §4 status.*).
type MinecraftServerStatus struct {
NodeName string `json:"nodeName,omitempty"`
// Phase is the coarse lifecycle phase.
Phase Phase `json:"phase,omitempty"`
// Ready is true only after a successful RCON probe (loader-agnostic; a
@@ -377,3 +384,11 @@ type PlayersStatus struct {
Online int32 `json:"online"`
Max int32 `json:"max"`
}
// WorldPVC resolves the active world, including servers created before migration support.
func (s *MinecraftServer) WorldPVC() string {
if s.Spec.Storage.ClaimName != "" {
return s.Spec.Storage.ClaimName
}
return naming.WorldPVCName(s.Name)
}
+573
View File
@@ -0,0 +1,573 @@
// Package archivetransfer serves archive bytes on A. It has no Kubernetes or database access.
package archivetransfer
import (
"context"
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"os"
"path/filepath"
"regexp"
"strings"
"sync"
"syscall"
"time"
"felis.lolicon.best/internal/backup"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/worldexport"
)
const TokenEnv = "FELIS_ARCHIVE_TOKEN"
const KeyEnv = "FELIS_ARCHIVE_KEY"
const LabelPending = "felis.lolicon.best/archive-pending"
const Annotation = "felis.lolicon.best/archive-transfer"
const DefaultLimit int64 = 100 << 30
var idRE = regexp.MustCompile(`^[a-f0-9]{32}$`)
// Ticket names exactly one operation, server and archive. Only A holds the signing key.
type Ticket struct {
ID string `json:"id"`
Server string `json:"server"`
Method string `json:"method"`
Ref string `json:"ref"`
SHA256 string `json:"sha256,omitempty"`
Limit int64 `json:"limit"`
Expires time.Time `json:"expires"`
}
type Receipt struct {
Ref string `json:"ref"`
Size int64 `json:"size"`
SHA256 string `json:"sha256"`
}
type Client struct {
URL, Root, Key string
Limit int64
}
func ID() string {
var b [16]byte
if _, err := rand.Read(b[:]); err != nil {
panic(err)
}
return hex.EncodeToString(b[:])
}
func (c Client) Issue(server, method, ref, sum string, ttl time.Duration) (Ticket, string, string, error) {
if len(c.Key) < 32 || c.URL == "" || ttl <= 0 || ttl > 24*time.Hour {
return Ticket{}, "", "", errors.New("archive transport is not configured or ticket lifetime is invalid")
}
id := ID()
if method == http.MethodPut {
ref = filepath.Join(c.Root, fmt.Sprintf("%s-%d.tar.gz", server, time.Now().UnixNano()))
}
limit := c.Limit
if limit <= 0 {
limit = DefaultLimit
}
t := Ticket{ID: id, Server: server, Method: method, Ref: ref, SHA256: sum, Limit: limit, Expires: time.Now().Add(ttl)}
if err := validate(t, c.Root, limit); err != nil {
return Ticket{}, "", "", err
}
raw, _ := json.Marshal(t)
payload := base64.RawURLEncoding.EncodeToString(raw)
mac := hmac.New(sha256.New, []byte(c.Key))
mac.Write([]byte(payload))
token := payload + "." + base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
return t, strings.TrimRight(c.URL, "/") + "/transfers/" + id, token, nil
}
func validate(t Ticket, root string, limit int64) error {
if !idRE.MatchString(t.ID) || t.Limit <= 0 || t.Limit > limit {
return errors.New("invalid transfer bounds")
}
if err := naming.ValidateSystemServerName(t.Server); err != nil {
return err
}
if t.Method != http.MethodPut && t.Method != http.MethodGet {
return errors.New("invalid transfer operation")
}
if !filepath.IsAbs(root) || filepath.Dir(t.Ref) != filepath.Clean(root) || !strings.HasSuffix(t.Ref, ".tar.gz") {
return errors.New("archive must be directly inside the archive root")
}
if !strings.HasPrefix(filepath.Base(t.Ref), t.Server+"-") {
return errors.New("archive belongs to another server")
}
if t.SHA256 != "" {
b, err := hex.DecodeString(t.SHA256)
if err != nil || len(b) != sha256.Size {
return errors.New("invalid SHA-256")
}
}
return nil
}
// Server journals consumption before IO, so process restarts cannot enable replay.
// A failure consumes the ticket too: the controller issues a fresh ticket on retry.
type Server struct {
Root, Key string
Limit int64
mu sync.Mutex
freeBytes func() (int64, error)
}
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/healthz" && r.Method == http.MethodGet {
w.WriteHeader(http.StatusNoContent)
return
}
if r.URL.Path == "/archives" && r.Method == http.MethodDelete {
if len(s.Key) < 32 || !hmac.Equal([]byte(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")), []byte(s.Key)) {
http.Error(w, "unauthorized", 401)
return
}
var q struct {
Ref string `json:"ref"`
}
if json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&q) != nil || filepath.Dir(q.Ref) != filepath.Clean(s.Root) {
http.Error(w, "bad ref", 400)
return
}
if err := os.Remove(q.Ref); err != nil && !errors.Is(err, os.ErrNotExist) {
http.Error(w, "archive delete failed", 500)
return
}
if err := syncDir(s.Root); err != nil {
http.Error(w, "archive delete commit failed", 500)
return
}
w.WriteHeader(204)
return
}
if r.URL.Path == "/inspect" && r.Method == http.MethodPost {
if len(s.Key) < 32 || !hmac.Equal([]byte(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")), []byte(s.Key)) {
http.Error(w, "unauthorized", 401)
return
}
var q struct {
Ref string `json:"ref"`
}
if json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&q) != nil || filepath.Dir(q.Ref) != filepath.Clean(s.Root) {
http.Error(w, "bad archive ref", 400)
return
}
local := &backup.TarLocal{BackupRoot: s.Root}
sum, err := local.Verify(r.Context(), backup.ArchiveRef(q.Ref), "")
if err != nil {
http.Error(w, "archive is absent or corrupt", 422)
return
}
st, err := os.Stat(q.Ref)
if err != nil {
http.Error(w, "archive absent", 404)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(Receipt{Ref: q.Ref, SHA256: sum, Size: st.Size()})
return
}
if strings.HasPrefix(r.URL.Path, "/receipts/") && r.Method == http.MethodGet {
if len(s.Key) < 32 || !hmac.Equal([]byte(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")), []byte(s.Key)) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
id := strings.TrimPrefix(r.URL.Path, "/receipts/")
if !idRE.MatchString(id) {
http.NotFound(w, r)
return
}
rec, err := s.receipt(r.Context(), id)
if err != nil {
http.NotFound(w, r)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(rec)
return
}
token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
payload, signature, ok := strings.Cut(token, ".")
mac := hmac.New(sha256.New, []byte(s.Key))
mac.Write([]byte(payload))
sig, err := base64.RawURLEncoding.DecodeString(signature)
if !ok || err != nil || len(s.Key) < 32 || !hmac.Equal(sig, mac.Sum(nil)) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
raw, err := base64.RawURLEncoding.DecodeString(payload)
var t Ticket
limit := s.Limit
if limit <= 0 {
limit = DefaultLimit
}
if err != nil || json.Unmarshal(raw, &t) != nil || validate(t, s.Root, limit) != nil || time.Now().After(t.Expires) || t.Expires.After(time.Now().Add(24*time.Hour)) || t.Method != r.Method || r.URL.Path != "/transfers/"+t.ID {
http.Error(w, "invalid or expired transfer", http.StatusForbidden)
return
}
if err := s.consume(t); err != nil {
if errors.Is(err, os.ErrExist) {
http.Error(w, "transfer already consumed", http.StatusConflict)
} else {
http.Error(w, "cannot journal transfer", http.StatusInsufficientStorage)
}
return
}
if t.Method == http.MethodGet {
s.download(w, r, t)
return
}
// Serialize uploads and disk checks to preserve a free-space reserve.
s.mu.Lock()
defer s.mu.Unlock()
if err := s.upload(r, t); err != nil {
http.Error(w, "archive upload failed: "+err.Error(), http.StatusUnprocessableEntity)
return
}
w.WriteHeader(http.StatusNoContent)
}
func syncDir(path string) error {
f, err := os.Open(path)
if err != nil {
return err
}
defer f.Close()
return f.Sync()
}
func (s *Server) consume(t Ticket) error {
id := t.ID
dir := filepath.Join(s.Root, ".transfers")
if err := os.MkdirAll(dir, 0700); err != nil {
return err
}
f, err := os.OpenFile(filepath.Join(dir, id+".used"), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
if err != nil {
return err
}
raw, _ := json.Marshal(t)
_, err = f.Write(raw)
if err == nil {
err = f.Sync()
}
cerr := f.Close()
if err != nil {
return err
}
if cerr != nil {
return cerr
}
return syncDir(dir)
}
func (s *Server) upload(r *http.Request, t Ticket) error {
available, err := s.availableBytes()
if err != nil {
return err
}
bound := min(t.Limit, available)
if bound <= 0 {
return backup.ErrNoRoom
}
tmp := filepath.Join(s.Root, "."+filepath.Base(t.Ref)+".partial")
f, err := os.OpenFile(tmp, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600)
if err != nil {
return err
}
defer os.Remove(tmp)
hash := sha256.New()
n, err := io.Copy(io.MultiWriter(f, hash), io.LimitReader(r.Body, bound+1))
if err == nil && n > bound {
err = errors.New("archive exceeds size or disk limit")
}
digest := "sha-256=:" + base64.StdEncoding.EncodeToString(hash.Sum(nil)) + ":"
if err == nil && r.Trailer.Get(worldexport.DigestTrailer) != digest {
err = errors.New("SHA-256 trailer is absent or incorrect")
}
sum := hex.EncodeToString(hash.Sum(nil))
if err == nil && t.SHA256 != "" && t.SHA256 != sum {
err = errors.New("SHA-256 does not match ticket")
}
if err == nil {
err = f.Sync()
}
cerr := f.Close()
if err == nil {
err = cerr
}
if err != nil {
return err
}
local := &backup.TarLocal{BackupRoot: s.Root}
if _, err = local.Verify(r.Context(), backup.ArchiveRef(tmp), sum); err != nil {
return err
}
// Never overwrite a committed archive, including one whose receipt was interrupted.
if err = os.Link(tmp, t.Ref); err != nil {
return err
}
if err = os.Remove(tmp); err != nil {
return err
}
if err = syncDir(s.Root); err != nil {
return err
}
rec := Receipt{Ref: t.Ref, Size: n, SHA256: sum}
raw, _ := json.Marshal(rec)
journal := filepath.Join(s.Root, ".transfers", t.ID+".json")
f, err = os.OpenFile(journal, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
if err != nil {
return err
}
_, err = f.Write(raw)
if err == nil {
err = f.Sync()
}
cerr = f.Close()
if err == nil {
err = cerr
}
if err != nil {
return err
}
return syncDir(filepath.Dir(journal))
}
func (s *Server) download(w http.ResponseWriter, r *http.Request, t Ticket) {
root, err := os.OpenRoot(s.Root)
if err != nil {
http.Error(w, "archive unavailable", 503)
return
}
defer root.Close()
f, err := root.Open(filepath.Base(t.Ref))
if err != nil {
http.NotFound(w, r)
return
}
defer f.Close()
st, err := f.Stat()
if err != nil || !st.Mode().IsRegular() || st.Size() > t.Limit {
http.Error(w, "archive exceeds transfer bounds", 413)
return
}
w.Header().Set("Content-Type", "application/gzip")
w.Header().Set("Content-Length", fmt.Sprint(st.Size()))
io.Copy(w, f)
}
func (c Client) Receipt(ctx context.Context, id string) (Receipt, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(c.URL, "/")+"/receipts/"+id, nil)
if err != nil {
return Receipt{}, err
}
req.Header.Set("Authorization", "Bearer "+c.Key)
hc := &http.Client{Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
resp, err := hc.Do(req)
if err != nil {
return Receipt{}, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return Receipt{}, fmt.Errorf("archive receipt: %s", resp.Status)
}
var rec Receipt
err = json.NewDecoder(io.LimitReader(resp.Body, 4096)).Decode(&rec)
return rec, err
}
// Fetch stages only the authorized archive and verifies its hash before extraction.
func Fetch(ctx context.Context, url, token, dest, want string, limit int64) error {
if want == "" {
return errors.New("download requires recorded SHA-256")
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+token)
hc := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
resp, err := hc.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
return fmt.Errorf("archive download: %s", resp.Status)
}
f, err := os.OpenFile(dest, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
if err != nil {
return err
}
hash := sha256.New()
n, err := io.Copy(io.MultiWriter(f, hash), io.LimitReader(resp.Body, limit+1))
if err == nil && (n > limit || hex.EncodeToString(hash.Sum(nil)) != want) {
err = errors.New("archive size or SHA-256 mismatch")
}
if err == nil {
err = f.Sync()
}
cerr := f.Close()
if err == nil {
err = cerr
}
if err != nil {
os.Remove(dest)
}
return err
}
func (c Client) Inspect(ctx context.Context, ref string) (Receipt, error) {
raw, _ := json.Marshal(map[string]string{"ref": ref})
req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.URL, "/")+"/inspect", strings.NewReader(string(raw)))
if err != nil {
return Receipt{}, err
}
req.Header.Set("Authorization", "Bearer "+c.Key)
hc := &http.Client{Timeout: 30 * time.Minute, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
resp, err := hc.Do(req)
if err != nil {
return Receipt{}, err
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
return Receipt{}, fmt.Errorf("archive inspect: %s", resp.Status)
}
var rec Receipt
err = json.NewDecoder(io.LimitReader(resp.Body, 4096)).Decode(&rec)
return rec, err
}
func (c Client) Delete(ctx context.Context, ref backup.ArchiveRef) error {
raw, _ := json.Marshal(map[string]string{"ref": string(ref)})
req, err := http.NewRequestWithContext(ctx, http.MethodDelete, strings.TrimRight(c.URL, "/")+"/archives", strings.NewReader(string(raw)))
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+c.Key)
hc := &http.Client{Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
resp, err := hc.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != 204 && resp.StatusCode != 404 {
return fmt.Errorf("archive delete: %s", resp.Status)
}
return nil
}
func (s *Server) receipt(ctx context.Context, id string) (Receipt, error) {
s.mu.Lock()
defer s.mu.Unlock()
path := filepath.Join(s.Root, ".transfers", id+".json")
var rec Receipt
if raw, err := os.ReadFile(path); err == nil && json.Unmarshal(raw, &rec) == nil {
return rec, nil
}
raw, err := os.ReadFile(filepath.Join(s.Root, ".transfers", id+".used"))
if err != nil {
return rec, err
}
var t Ticket
if err = json.Unmarshal(raw, &t); err != nil {
return rec, err
}
if t.Method != http.MethodPut {
return rec, errors.New("not an upload")
}
sum, err := (&backup.TarLocal{BackupRoot: s.Root}).Verify(ctx, backup.ArchiveRef(t.Ref), t.SHA256)
if err != nil {
return rec, err
}
st, err := os.Stat(t.Ref)
if err != nil || st.Size() > t.Limit {
return rec, errors.New("invalid committed archive")
}
rec = Receipt{Ref: t.Ref, Size: st.Size(), SHA256: sum}
raw, _ = json.Marshal(rec)
tmp := path + ".partial"
f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600)
if err != nil {
return rec, err
}
_, err = f.Write(raw)
if err == nil {
err = f.Sync()
}
cerr := f.Close()
if err == nil {
err = cerr
}
if err != nil {
return rec, err
}
if err = os.Rename(tmp, path); err != nil {
return rec, err
}
return rec, syncDir(filepath.Dir(path))
}
func (s *Server) availableBytes() (int64, error) {
if s.freeBytes != nil {
return s.freeBytes()
}
var st syscall.Statfs_t
if err := syscall.Statfs(s.Root, &st); err != nil {
return 0, err
}
return int64(st.Bavail)*int64(st.Bsize) - int64(float64(st.Blocks)*float64(st.Bsize)*backup.MinFreeAfter), nil
}
// Sweep keeps receipts for committed archives across arbitrarily long A outages.
// Expired failed transfers and downloads need no replay journal: signature expiry
// still rejects them. The extra day leaves no overlap with an active upload.
func (s *Server) Sweep(now time.Time) error {
s.mu.Lock()
defer s.mu.Unlock()
dir := filepath.Join(s.Root, ".transfers")
entries, err := os.ReadDir(dir)
if errors.Is(err, os.ErrNotExist) {
return nil
}
if err != nil {
return err
}
for _, e := range entries {
if !strings.HasSuffix(e.Name(), ".used") {
continue
}
raw, err := os.ReadFile(filepath.Join(dir, e.Name()))
if err != nil {
return err
}
var ticket Ticket
if json.Unmarshal(raw, &ticket) != nil || now.Before(ticket.Expires.Add(24*time.Hour)) {
continue
}
if ticket.Method == http.MethodPut {
if _, err := os.Stat(ticket.Ref); err == nil {
continue
} else if !errors.Is(err, os.ErrNotExist) {
return err
}
}
for _, name := range []string{e.Name(), ticket.ID + ".json", ticket.ID + ".json.partial"} {
if err := os.Remove(filepath.Join(dir, name)); err != nil && !errors.Is(err, os.ErrNotExist) {
return err
}
}
}
return syncDir(dir)
}
+228
View File
@@ -0,0 +1,228 @@
package archivetransfer
import (
"archive/tar"
"bytes"
"compress/gzip"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
"felis.lolicon.best/internal/backup"
"felis.lolicon.best/internal/worldexport"
)
func archiveBytes(t *testing.T) []byte {
t.Helper()
var buf bytes.Buffer
gz := gzip.NewWriter(&buf)
tw := tar.NewWriter(gz)
if err := tw.WriteHeader(&tar.Header{Name: "world/level.dat", Mode: 0644, Size: 5}); err != nil {
t.Fatal(err)
}
if _, err := tw.Write([]byte("world")); err != nil {
t.Fatal(err)
}
if err := tw.Close(); err != nil {
t.Fatal(err)
}
if err := gz.Close(); err != nil {
t.Fatal(err)
}
return buf.Bytes()
}
func upload(t *testing.T, url, token string, body []byte, digest bool) int {
t.Helper()
r, err := http.NewRequest(http.MethodPut, url, bytes.NewReader(body))
if err != nil {
t.Fatal(err)
}
r.Header.Set("Authorization", "Bearer "+token)
if digest {
sum := sha256.Sum256(body)
r.ContentLength = -1
r.Trailer = http.Header{worldexport.DigestTrailer: {"sha-256=:" + base64.StdEncoding.EncodeToString(sum[:]) + ":"}}
}
resp, err := http.DefaultClient.Do(r)
if err != nil {
t.Fatal(err)
}
response, _ := io.ReadAll(resp.Body)
if resp.StatusCode >= 400 {
t.Log(string(response))
}
resp.Body.Close()
return resp.StatusCode
}
func TestDurableTransferAndRestore(t *testing.T) {
ctx := context.Background()
root := t.TempDir()
key := strings.Repeat("k", 32)
s := &Server{Root: root, Key: key, Limit: 1 << 20, freeBytes: func() (int64, error) { return 1 << 30, nil }}
ts := httptest.NewServer(s)
defer ts.Close()
c := Client{Root: root, Key: key, URL: ts.URL, Limit: 1 << 20}
ticket, url, token, err := c.Issue("alice", "PUT", "", "", time.Minute)
if err != nil {
t.Fatal(err)
}
body := archiveBytes(t)
if code := upload(t, url, token, body, true); code != 204 {
t.Fatalf("upload %d", code)
}
if code := upload(t, url, token, body, true); code != 409 {
t.Fatalf("replay %d", code)
}
rec, err := c.Receipt(ctx, ticket.ID)
if err != nil {
t.Fatal(err)
}
sum := sha256.Sum256(body)
if rec.Size != int64(len(body)) || rec.SHA256 != hex.EncodeToString(sum[:]) {
t.Fatalf("receipt %+v", rec)
}
// Simulate A losing the response and crashing between archive commit and receipt write.
if err := os.Remove(filepath.Join(root, ".transfers", ticket.ID+".json")); err != nil {
t.Fatal(err)
}
restarted := &Server{Root: root, Key: key, Limit: 1 << 20, freeBytes: func() (int64, error) { return 1 << 30, nil }}
if recovered, err := restarted.receipt(ctx, ticket.ID); err != nil || recovered != rec {
t.Fatalf("recovery %+v: %v", recovered, err)
}
if _, _, _, err := c.Issue("bob", "GET", rec.Ref, rec.SHA256, time.Minute); err == nil {
t.Fatal("cross-server download ticket accepted")
}
_, getURL, getToken, err := c.Issue("alice", "GET", rec.Ref, rec.SHA256, time.Minute)
if err != nil {
t.Fatal(err)
}
dest := filepath.Join(t.TempDir(), "download.tar.gz")
if err := Fetch(ctx, getURL, getToken, dest, rec.SHA256, 1<<20); err != nil {
t.Fatal(err)
}
if err := Fetch(ctx, getURL, getToken, filepath.Join(t.TempDir(), "replay"), rec.SHA256, 1<<20); err == nil {
t.Fatal("download replay accepted")
}
world := t.TempDir()
a := &backup.TarLocal{BackupRoot: filepath.Dir(dest), Resolve: func(string) (string, error) { return world, nil }}
if err := a.Restore(ctx, backup.ArchiveRef(dest), "alice"); err != nil {
t.Fatal(err)
}
if err := backup.VerifyRestored(ctx, dest, world); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(world, "world", "level.dat"), []byte("wrong"), 0644); err != nil {
t.Fatal(err)
}
if err := backup.VerifyRestored(ctx, dest, world); err == nil {
t.Fatal("read-back corruption accepted")
}
}
func TestRejectedTransfersNeverCommit(t *testing.T) {
for _, tc := range []struct {
name string
limit int64
body []byte
digest bool
}{
{"missing checksum", 1 << 20, archiveBytes(t), false},
{"invalid tar", 1 << 20, []byte("corrupt archive"), true},
{"size limit", 10, archiveBytes(t), true},
} {
t.Run(tc.name, func(t *testing.T) {
root := t.TempDir()
key := strings.Repeat("k", 32)
ts := httptest.NewServer(&Server{Root: root, Key: key, Limit: tc.limit, freeBytes: func() (int64, error) { return 1 << 30, nil }})
defer ts.Close()
c := Client{Root: root, Key: key, URL: ts.URL, Limit: tc.limit}
ticket, url, token, err := c.Issue("alice", "PUT", "", "", time.Minute)
if err != nil {
t.Fatal(err)
}
if code := upload(t, url, token, tc.body, tc.digest); code != 422 {
t.Fatalf("status %d", code)
}
if _, err := os.Stat(ticket.Ref); !os.IsNotExist(err) {
t.Fatalf("archive committed: %v", err)
}
if _, err := c.Receipt(context.Background(), ticket.ID); err == nil {
t.Fatal("failed upload has receipt")
}
if code := upload(t, url, token, tc.body, true); code != 409 {
t.Fatalf("failed upload replay %d", code)
}
})
}
}
func signTicket(t Ticket, key string) string {
raw, _ := json.Marshal(t)
payload := base64.RawURLEncoding.EncodeToString(raw)
h := hmac.New(sha256.New, []byte(key))
h.Write([]byte(payload))
return payload + "." + base64.RawURLEncoding.EncodeToString(h.Sum(nil))
}
func TestExpiredWrongMethodAndPath(t *testing.T) {
root := t.TempDir()
key := strings.Repeat("k", 32)
s := &Server{Root: root, Key: key}
c := Client{Root: root, Key: key, URL: "http://archive"}
ticket, _, token, err := c.Issue("alice", "PUT", "", "", time.Minute)
if err != nil {
t.Fatal(err)
}
for _, tc := range []struct{ method, path, token string }{
{"GET", "/transfers/" + ticket.ID, token},
{"PUT", "/transfers/" + ID(), token},
{"PUT", "/transfers/" + ticket.ID, token + "broken"},
} {
r := httptest.NewRequest(tc.method, tc.path, nil)
r.Header.Set("Authorization", "Bearer "+tc.token)
w := httptest.NewRecorder()
s.ServeHTTP(w, r)
if w.Code != 403 && w.Code != 401 {
t.Fatalf("scope accepted: %d", w.Code)
}
}
ticket.Expires = time.Now().Add(-time.Minute)
r := httptest.NewRequest("PUT", "/transfers/"+ticket.ID, nil)
r.Header.Set("Authorization", "Bearer "+signTicket(ticket, key))
w := httptest.NewRecorder()
s.ServeHTTP(w, r)
if w.Code != 403 {
t.Fatalf("expired %d", w.Code)
}
}
func TestFullDiskDoesNotCommit(t *testing.T) {
root := t.TempDir()
key := strings.Repeat("k", 32)
ts := httptest.NewServer(&Server{Root: root, Key: key, freeBytes: func() (int64, error) { return 0, nil }})
defer ts.Close()
c := Client{Root: root, Key: key, URL: ts.URL}
ticket, url, token, err := c.Issue("alice", "PUT", "", "", time.Minute)
if err != nil {
t.Fatal(err)
}
if code := upload(t, url, token, archiveBytes(t), true); code != 422 {
t.Fatal("disk full accepted", code)
}
if _, err := os.Stat(ticket.Ref); !os.IsNotExist(err) {
t.Fatal("full disk committed archive", err)
}
}
+69
View File
@@ -0,0 +1,69 @@
package backup
import (
"archive/tar"
"compress/gzip"
"context"
"crypto/sha256"
"errors"
"fmt"
"io"
"os"
)
// VerifyRestored reads target files back and compares them with every regular archive entry.
// OpenRoot keeps a malicious path or existing symlink inside the world volume.
func VerifyRestored(ctx context.Context, ref, world string) error {
root, err := os.OpenRoot(world)
if err != nil {
return err
}
defer root.Close()
f, err := os.Open(ref)
if err != nil {
return err
}
defer f.Close()
gz, err := gzip.NewReader(f)
if err != nil {
return err
}
defer gz.Close()
tr := tar.NewReader(gz)
for {
if err := ctx.Err(); err != nil {
return err
}
h, err := tr.Next()
if errors.Is(err, io.EOF) {
return nil
}
if err != nil {
return err
}
if h.Typeflag != tar.TypeReg && h.Typeflag != tar.TypeRegA {
continue
}
dst, err := root.Open(h.Name)
if err != nil {
return err
}
st, err := dst.Stat()
if err != nil || !st.Mode().IsRegular() || st.Size() != h.Size {
dst.Close()
return fmt.Errorf("restored entry %q has the wrong type or size", h.Name)
}
sourceHash, targetHash := sha256.New(), sha256.New()
_, err = io.Copy(sourceHash, tr)
if err == nil {
_, err = io.Copy(targetHash, dst)
}
dst.Close()
if err != nil {
return err
}
if string(sourceHash.Sum(nil)) != string(targetHash.Sum(nil)) {
return fmt.Errorf("restored entry %q failed SHA-256 read-back", h.Name)
}
}
}
+25 -1
View File
@@ -35,6 +35,7 @@ import (
"time"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
)
// ErrAlreadyExists is returned by a Jobs implementation when a backup Job for a
@@ -57,6 +58,7 @@ type Jobs interface {
// the caller leaves the API's Backuper nil so the endpoint reports 503 rather than
// enqueuing a Job that cannot run.
type Config struct {
ResolveWorld placement.Resolver
// Namespace is where the world PVCs live and the backup Job runs (the minecraft
// namespace), co-located with the world it snapshots.
Namespace string
@@ -178,7 +180,15 @@ type Backuper struct {
// land on different nodes the RWO attach fails one cleanly. Add single-flight-on-
// running only if a real double-tap storm ever shows up.
func (b *Backuper) Backup(ctx context.Context, serverName, formerOwner string) error {
if err := b.Jobs.CreateBackupJob(ctx, b.jobParams(serverName, formerOwner)); err != nil {
p := b.jobParams(serverName, formerOwner)
if b.Config.ResolveWorld != nil {
w, err := b.Config.ResolveWorld(ctx, serverName)
if err != nil {
return err
}
p.WorldPVC = w.Claim
}
if err := b.Jobs.CreateBackupJob(ctx, p); err != nil {
if errors.Is(err, ErrAlreadyExists) {
return nil // suffix collision — treat as enqueued
}
@@ -195,6 +205,13 @@ func (b *Backuper) Backup(ctx context.Context, serverName, formerOwner string) e
// the world volume as a restore until then (internal/maintenance).
func (b *Backuper) BackupThenRestore(ctx context.Context, serverName, formerOwner, backupID, backupRef string) error {
p := b.jobParams(serverName, formerOwner)
if b.Config.ResolveWorld != nil {
w, err := b.Config.ResolveWorld(ctx, serverName)
if err != nil {
return err
}
p.WorldPVC = w.Claim
}
p.RestoreRef, p.RestoreBackupID = backupRef, backupID
if err := b.Jobs.CreateBackupJob(ctx, p); err != nil {
if errors.Is(err, ErrAlreadyExists) {
@@ -210,6 +227,13 @@ func (b *Backuper) BackupThenRestore(ctx context.Context, serverName, formerOwne
// [archive] scheduled_keep, so the owner's own backups keep their count.
func (b *Backuper) BackupScheduled(ctx context.Context, serverName, formerOwner string) error {
p := b.jobParams(serverName, formerOwner)
if b.Config.ResolveWorld != nil {
w, err := b.Config.ResolveWorld(ctx, serverName)
if err != nil {
return err
}
p.WorldPVC = w.Claim
}
p.Scheduled = true
if err := b.Jobs.CreateBackupJob(ctx, p); err != nil {
if errors.Is(err, ErrAlreadyExists) {
+278
View File
@@ -0,0 +1,278 @@
// Package distributed extends the existing Job executors with archive transport and stopped migration.
package distributed
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"time"
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/archivetransfer"
"felis.lolicon.best/internal/backup"
"felis.lolicon.best/internal/backupjob"
"felis.lolicon.best/internal/maintenance"
"felis.lolicon.best/internal/placement"
"felis.lolicon.best/internal/restore"
"felis.lolicon.best/internal/worldexport"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
)
const LabelTransfer = "felis.lolicon.best/archive-job"
type Backup struct {
ID string `json:"id"`
Server string `json:"server"`
Owner string `json:"owner"`
Reason string `json:"reason"`
Protect string `json:"protect,omitempty"`
Receipt archivetransfer.Receipt `json:"receipt"`
}
type pendingBackup struct {
Ticket archivetransfer.Ticket
Owner, Reason, Protect string
}
type Manager struct {
Client client.Client
Namespace, Image, Controller string
Archive archivetransfer.Client
Resolve placement.Resolver
// Record is idempotent by transfer ID and runs on A after durable archive commit.
Record func(context.Context, Backup) error
}
func (m *Manager) CreateBackupJob(ctx context.Context, p backupjob.JobParams) error {
world, err := m.Resolve(ctx, p.Server)
if err != nil {
return err
}
job, t, err := m.uploadJob(p.Server, world.Claim, world.Node, p.JobName)
if err != nil {
return err
}
reason := "manual"
if p.Scheduled {
reason = backupjob.ReasonScheduled
}
if p.RestoreRef != "" {
reason = backupjob.ReasonPreRestore
}
meta := pendingBackup{Ticket: t, Owner: p.FormerOwner, Reason: reason, Protect: p.RestoreBackupID}
raw, _ := json.Marshal(meta)
job.Annotations = map[string]string{archivetransfer.Annotation: string(raw)}
job.Labels[maintenance.LabelManagedBy] = "felis-backup"
job.Labels[archivetransfer.LabelPending] = "true"
if p.RestoreRef != "" {
job.Labels[maintenance.LabelThenRestore] = maintenance.ThenRestorePending
job.Annotations[maintenance.AnnotationRestoreRef] = p.RestoreRef
job.Annotations[maintenance.AnnotationRestoreBackupID] = p.RestoreBackupID
}
job.Spec.TTLSecondsAfterFinished = nil
err = m.Client.Create(ctx, job)
if apierrors.IsAlreadyExists(err) {
return backupjob.ErrAlreadyExists
}
return err
}
func (m *Manager) uploadJob(server, claim, node, name string) (*batchv1.Job, archivetransfer.Ticket, error) {
t, url, token, err := m.Archive.Issue(server, http.MethodPut, "", "", 2*time.Hour)
if err != nil {
return nil, t, err
}
job, err := worldexport.ExportJob(worldexport.JobParams{Server: server, ID: t.ID[:16], Mode: worldexport.ModeWorld, WorldPVC: claim, TargetURL: url, Token: token, Namespace: m.Namespace, ServiceAccount: "felis-restore", Image: m.Image, WorldsRoot: "/world", Deadline: 2 * time.Hour})
if err != nil {
return nil, t, err
}
job.Name = name
job.Labels[LabelTransfer] = "true"
job.Spec.Template.Labels[LabelTransfer] = "true"
job.Spec.Template.Spec.Containers[0].Args = append(job.Spec.Template.Spec.Containers[0].Args, "--archive-raw")
m.pin(&job.Spec.Template.Spec, node)
return job, t, nil
}
func (m *Manager) pin(p *corev1.PodSpec, node string) {
if node == "" {
node = m.Controller
}
p.NodeSelector = map[string]string{placement.LabelIdentity: node}
}
func (m *Manager) restoreParams(ctx context.Context, p restore.JobParams) (restore.JobParams, error) {
rec, err := m.Archive.Inspect(ctx, p.BackupRef)
if err != nil {
return p, err
}
_, url, token, err := m.Archive.Issue(p.Server, http.MethodGet, p.BackupRef, rec.SHA256, 2*time.Hour)
if err != nil {
return p, err
}
p.SourceURL, p.Token, p.SHA256, p.MaxBytes = url, token, rec.SHA256, m.Archive.Limit
if p.MaxBytes <= 0 {
p.MaxBytes = archivetransfer.DefaultLimit
}
p.BackupPVC = ""
return p, nil
}
func (m *Manager) CreateRestoreJob(ctx context.Context, p restore.JobParams) error {
world, err := m.Resolve(ctx, p.Server)
if err != nil {
return err
}
p.WorldPVC = world.Claim
p, err = m.restoreParams(ctx, p)
if err != nil {
return err
}
// Preserve the existing deterministic-name conflict and finished-Job retry rules.
return restore.NewK8sJobs(&pinnedClient{Client: m.Client, node: world.Node, manager: m}).CreateRestoreJob(ctx, p)
}
type pinnedClient struct {
client.Client
node string
manager *Manager
}
func (c *pinnedClient) Create(ctx context.Context, o client.Object, opts ...client.CreateOption) error {
if j, ok := o.(*batchv1.Job); ok {
c.manager.pin(&j.Spec.Template.Spec, c.node)
j.Labels[LabelTransfer] = "true"
j.Spec.Template.Labels[LabelTransfer] = "true"
}
return c.Client.Create(ctx, o, opts...)
}
// SettleBackups is retried after A restarts. Jobs remain durable until the row is recorded.
func (m *Manager) SettleBackups(ctx context.Context) error {
var list batchv1.JobList
if err := m.Client.List(ctx, &list, client.InNamespace(m.Namespace), client.MatchingLabels{archivetransfer.LabelPending: "true"}); err != nil {
return err
}
var errs []error
for i := range list.Items {
j := &list.Items[i]
var p pendingBackup
if err := json.Unmarshal([]byte(j.Annotations[archivetransfer.Annotation]), &p); err != nil {
errs = append(errs, err)
continue
}
rec, err := m.Archive.Receipt(ctx, p.Ticket.ID)
if err != nil {
if !maintenance.JobFinished(j) {
continue
}
// A terminal upload failure holds nothing, but must never start its restore chain.
if jobSucceeded(j) {
errs = append(errs, fmt.Errorf("backup %s awaits durable receipt: %w", j.Name, err))
continue
}
} else {
if rec.Ref != p.Ticket.Ref || rec.SHA256 == "" || rec.Size <= 0 || rec.Size > p.Ticket.Limit {
errs = append(errs, fmt.Errorf("invalid receipt for %s", j.Name))
continue
}
if m.Record == nil {
errs = append(errs, errors.New("backup recorder unavailable"))
continue
}
if err := m.Record(ctx, Backup{ID: p.Ticket.ID, Server: p.Ticket.Server, Owner: p.Owner, Reason: p.Reason, Protect: p.Protect, Receipt: rec}); err != nil {
errs = append(errs, err)
continue
}
}
before := j.DeepCopy()
delete(j.Labels, archivetransfer.LabelPending)
ttl := int32(600)
j.Spec.TTLSecondsAfterFinished = &ttl
if err := m.Client.Patch(ctx, j, client.MergeFromWithOptions(before, client.MergeFromWithOptimisticLock{})); err != nil {
errs = append(errs, err)
}
}
return errors.Join(errs...)
}
func jobSucceeded(j *batchv1.Job) bool {
for _, c := range j.Status.Conditions {
if c.Type == batchv1.JobComplete && c.Status == corev1.ConditionTrue {
return true
}
}
return false
}
// RemoteArchiver lets the existing reaper make every decision on A and snapshot only one remote PVC.
// Local verification, retention and offsite continue to use the same tarLocal paths.
type RemoteArchiver struct {
*backup.TarLocal
Manager *Manager
}
// A failed migration may wait for operator intervention longer than normal
// backup retention. Keep its safety archive until the persistent lock releases.
func (a *RemoteArchiver) Delete(ctx context.Context, ref backup.ArchiveRef) error {
var servers v1alpha1.MinecraftServerList
if err := a.Manager.Client.List(ctx, &servers, client.InNamespace(a.Manager.Namespace)); err != nil {
return err
}
for i := range servers.Items {
op, err := readOperation(&servers.Items[i])
if err != nil && !errors.Is(err, ErrNotFound) {
return err
}
if err == nil && op.State != "succeeded" && op.Backup.Ref == string(ref) {
return fmt.Errorf("%w: migration retains its safety archive", ErrBusy)
}
}
return a.TarLocal.Delete(ctx, ref)
}
func (a *RemoteArchiver) Archive(ctx context.Context, server, pvc string) (backup.Archived, error) {
m := a.Manager
world, err := m.Resolve(ctx, server)
if err != nil {
return backup.Archived{}, err
}
if world.Claim != pvc {
return backup.Archived{}, errors.New("active PVC changed")
}
id := archivetransfer.ID()
name := "reap-" + id[:16]
j, t, err := m.uploadJob(server, pvc, world.Node, name)
if err != nil {
return backup.Archived{}, err
}
if err = m.Client.Create(ctx, j); err != nil {
return backup.Archived{}, err
}
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
for {
rec, err := m.Archive.Receipt(ctx, t.ID)
if err == nil {
return backup.Archived{Ref: backup.ArchiveRef(rec.Ref), Size: rec.Size, SHA256: rec.SHA256}, nil
}
var job batchv1.Job
if err = m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: name}, &job); err != nil {
return backup.Archived{}, err
}
if maintenance.JobFinished(&job) && !jobSucceeded(&job) {
return backup.Archived{}, errors.New("remote archive Job failed")
}
select {
case <-ctx.Done():
return backup.Archived{}, ctx.Err()
case <-ticker.C:
}
}
}
+521
View File
@@ -0,0 +1,521 @@
package distributed
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"time"
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/archivetransfer"
"felis.lolicon.best/internal/maintenance"
"felis.lolicon.best/internal/placement"
"felis.lolicon.best/internal/restore"
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/util/retry"
"sigs.k8s.io/controller-runtime/pkg/client"
)
const MigrationAnnotation = "felis.lolicon.best/migration"
// Admission errors keep HTTP policy in the API layer.
var ErrBusy = errors.New("server must be fully stopped with no maintenance operation")
var ErrNotFound = errors.New("migration not found")
type Node struct {
Name string `json:"name"`
Role string `json:"role"`
Ready bool `json:"ready"`
Approved bool `json:"approved"`
Addresses []string `json:"addresses"`
Architecture string `json:"architecture"`
}
func (m *Manager) Nodes(ctx context.Context) ([]Node, error) {
var list corev1.NodeList
if err := m.Client.List(ctx, &list); err != nil {
return nil, err
}
out := make([]Node, 0, len(list.Items))
for _, n := range list.Items {
info := Node{Name: n.Name, Role: n.Labels[placement.LabelRole], Ready: placement.Online(&n), Approved: n.Labels[placement.LabelApproved] == "true" && !n.Spec.Unschedulable, Addresses: []string{}, Architecture: n.Status.NodeInfo.Architecture}
for _, a := range n.Status.Addresses {
if a.Type == corev1.NodeInternalIP || a.Type == corev1.NodeExternalIP {
info.Addresses = append(info.Addresses, a.Address)
}
}
out = append(out, info)
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out, nil
}
func (m *Manager) ValidateNode(ctx context.Context, name string) error {
return placement.Worker(ctx, m.Client, name)
}
// Operation is persisted on the CR together with its non-expiring maintenance lock.
// SourcePVC is retained even after success, and retry never changes the committed active world.
type Operation struct {
ID string `json:"id"`
Server string `json:"server"`
State string `json:"state"`
Stage string `json:"stage"`
SourceNode string `json:"sourceNode"`
TargetNode string `json:"targetNode"`
SourcePVC string `json:"sourcePVC"`
TargetPVC string `json:"targetPVC"`
Backup archivetransfer.Receipt `json:"backup"`
Owner string `json:"-"`
Started time.Time `json:"started"`
Updated time.Time `json:"updated"`
Error string `json:"error,omitempty"`
Switched bool `json:"switched"`
Attempt int `json:"attempt"`
}
// owner travels in persistence, but never on the public operation view.
type persistedOperation struct {
Operation
OwnerID string `json:"ownerId"`
}
func readOperation(s *v1alpha1.MinecraftServer) (Operation, error) {
var stored persistedOperation
if s.Annotations[MigrationAnnotation] == "" {
return Operation{}, ErrNotFound
}
if err := json.Unmarshal([]byte(s.Annotations[MigrationAnnotation]), &stored); err != nil {
return Operation{}, err
}
stored.Operation.Owner = stored.OwnerID
return stored.Operation, nil
}
func (m *Manager) Migration(ctx context.Context, server, id string) (Operation, error) {
var s v1alpha1.MinecraftServer
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: server}, &s); err != nil {
return Operation{}, err
}
op, err := readOperation(&s)
if err == nil && id != "" && op.ID != id {
return Operation{}, ErrNotFound
}
return op, err
}
func (m *Manager) quiet(ctx context.Context, s *v1alpha1.MinecraftServer, own ...string) error {
if s.Spec.DesiredState != v1alpha1.DesiredStopped || s.Status.Phase != v1alpha1.PhaseStopped || s.Status.Ready {
return ErrBusy
}
var pods corev1.PodList
if err := m.Client.List(ctx, &pods, client.InNamespace(m.Namespace), client.MatchingLabels{v1alpha1.LabelServer: s.Name}); err != nil {
return err
}
// Even terminal maintenance Pods must have exited before a new attempt writes its PVC.
for _, p := range pods.Items {
allowed := len(own) > 0 && p.Labels[MigrationAnnotation] == own[0]
if p.Labels[v1alpha1.LabelComponent] == "server" || (!allowed && p.Status.Phase != corev1.PodSucceeded && p.Status.Phase != corev1.PodFailed) {
return ErrBusy
}
}
return nil
}
func (m *Manager) BeginMigration(ctx context.Context, server, target, owner string) (Operation, error) {
if err := m.ValidateNode(ctx, target); err != nil {
return Operation{}, err
}
var op Operation
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
var s v1alpha1.MinecraftServer
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: server}, &s); err != nil {
return err
}
if previous, err := readOperation(&s); err == nil && previous.State != "succeeded" {
if previous.TargetNode == target {
op = previous
return nil
}
return ErrBusy
} else if err != nil && !errors.Is(err, ErrNotFound) {
return err
}
if s.Spec.ReaperExempt || s.Labels[v1alpha1.LabelSystemRole] != "" {
return ErrBusy
}
if err := m.quiet(ctx, &s); err != nil {
return err
}
var jobs batchv1.JobList
if err := m.Client.List(ctx, &jobs, client.InNamespace(m.Namespace), client.MatchingLabels{maintenance.LabelServer: server}); err != nil {
return err
}
if _, held := maintenance.Holder(server, s.Annotations, jobs.Items, time.Now()); held {
return ErrBusy
}
world, err := m.Resolve(ctx, server)
if err != nil {
return err
}
source := world.Node
if source == "" {
source = m.Controller
}
if source == target {
return errors.New("source and target nodes are identical")
}
// Bound local-path worlds have a physical node; reject a guessed or mismatched source.
var pvc corev1.PersistentVolumeClaim
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: world.Claim}, &pvc); err != nil {
return err
}
if pvc.Spec.VolumeName == "" {
return errors.New("source world is not bound")
}
if err := m.volumeOnNode(ctx, pvc.Spec.VolumeName, source); err != nil {
return err
}
id := archivetransfer.ID()
now := time.Now().UTC()
op = Operation{ID: id, Server: server, State: "backing_up", Stage: "backing_up", SourceNode: source, TargetNode: target, SourcePVC: world.Claim, TargetPVC: "world-" + server + "-m" + id[:12], Owner: owner, Started: now, Updated: now}
return m.save(ctx, &s, op, true)
})
return op, err
}
func (m *Manager) volumeOnNode(ctx context.Context, volume, node string) error {
var pv corev1.PersistentVolume
if err := m.Client.Get(ctx, types.NamespacedName{Name: volume}, &pv); err != nil {
return err
}
if pv.Spec.NodeAffinity == nil || pv.Spec.NodeAffinity.Required == nil {
return errors.New("migration requires a node-local volume with node affinity")
}
var n corev1.Node
if err := m.Client.Get(ctx, types.NamespacedName{Name: node}, &n); err != nil {
return err
}
for _, term := range pv.Spec.NodeAffinity.Required.NodeSelectorTerms {
if len(term.MatchFields) > 0 {
continue
}
matched := len(term.MatchExpressions) > 0
for _, e := range term.MatchExpressions {
if e.Operator != corev1.NodeSelectorOpIn {
matched = false
break
}
found := false
for _, v := range e.Values {
if n.Labels[e.Key] == v {
found = true
}
}
if !found {
matched = false
break
}
}
if matched {
return nil
}
}
return errors.New("world volume is not on the recorded execution node")
}
func (m *Manager) save(ctx context.Context, s *v1alpha1.MinecraftServer, op Operation, lock bool) error {
before := s.DeepCopy()
op.Updated = time.Now().UTC()
raw, _ := json.Marshal(persistedOperation{Operation: op, OwnerID: op.Owner})
if s.Annotations == nil {
s.Annotations = map[string]string{}
}
s.Annotations[MigrationAnnotation] = string(raw)
if lock {
s.Annotations[maintenance.Annotation] = maintenance.LockValue(maintenance.KindMigration, op.Started)
} else {
delete(s.Annotations, maintenance.Annotation)
}
return m.Client.Patch(ctx, s, client.MergeFromWithOptions(before, client.MergeFromWithOptimisticLock{}))
}
func (m *Manager) RetryMigration(ctx context.Context, server, id string) (Operation, error) {
var op Operation
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
var s v1alpha1.MinecraftServer
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: server}, &s); err != nil {
return err
}
var err error
op, err = readOperation(&s)
if err != nil {
return err
}
if op.ID != id {
return ErrNotFound
}
if op.State != "failed" {
return ErrBusy
}
if err := m.quiet(ctx, &s); err != nil {
return err
}
if err := m.ValidateNode(ctx, op.TargetNode); err != nil {
return err
}
op.State = op.Stage
op.Error = ""
op.Attempt++
return m.save(ctx, &s, op, true)
})
return op, err
}
func (m *Manager) ReconcileMigrations(ctx context.Context) error {
var servers v1alpha1.MinecraftServerList
if err := m.Client.List(ctx, &servers, client.InNamespace(m.Namespace)); err != nil {
return err
}
var errs []error
for i := range servers.Items {
s := &servers.Items[i]
op, err := readOperation(s)
if errors.Is(err, ErrNotFound) {
continue
}
if err != nil {
errs = append(errs, err)
continue
}
if op.State == "succeeded" {
if err := m.expireMigrationJobs(ctx, op.ID); err != nil {
errs = append(errs, err)
}
continue
}
if op.State == "failed" {
continue
}
err = m.advance(ctx, s, &op)
if err != nil {
if apierrors.IsConflict(err) {
continue
}
op.Stage = op.State
op.State = "failed"
op.Error = err.Error()
if saveErr := m.save(ctx, s, op, true); saveErr != nil {
errs = append(errs, saveErr)
}
}
}
return errors.Join(errs...)
}
func (m *Manager) advance(ctx context.Context, s *v1alpha1.MinecraftServer, op *Operation) error {
if err := m.quiet(ctx, s, op.ID); err != nil {
return err
}
if s.Annotations[maintenance.Annotation] != maintenance.LockValue(maintenance.KindMigration, op.Started) {
return errors.New("persistent migration lock was changed")
}
if err := m.ValidateNode(ctx, op.TargetNode); err != nil {
return err
}
if !op.Switched && (s.WorldPVC() != op.SourcePVC || (s.Spec.NodeName != "" && s.Spec.NodeName != op.SourceNode)) {
return errors.New("source placement changed during migration")
}
var source corev1.Node
if !op.Switched {
if err := m.Client.Get(ctx, types.NamespacedName{Name: op.SourceNode}, &source); err != nil {
return err
}
if !placement.Online(&source) {
return errors.New("source node is offline")
}
}
name := fmt.Sprintf("migration-%s-%d", op.ID[:16], op.Attempt)
switch op.State {
case "backing_up":
var j batchv1.Job
err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: name + "-backup"}, &j)
if apierrors.IsNotFound(err) {
job, t, err := m.uploadJob(op.Server, op.SourcePVC, op.SourceNode, name+"-backup")
if err != nil {
return err
}
raw, _ := json.Marshal(t)
job.Annotations = map[string]string{archivetransfer.Annotation: string(raw)}
job.Labels[MigrationAnnotation] = op.ID
job.Spec.Template.Labels[MigrationAnnotation] = op.ID
job.Spec.TTLSecondsAfterFinished = nil
return m.Client.Create(ctx, job)
}
if err != nil {
return err
}
var t archivetransfer.Ticket
if err = json.Unmarshal([]byte(j.Annotations[archivetransfer.Annotation]), &t); err != nil {
return err
}
rec, err := m.Archive.Receipt(ctx, t.ID)
if err != nil {
if maintenance.JobFinished(&j) {
return fmt.Errorf("migration backup has no durable receipt: %w", err)
}
return nil
}
if rec.Ref != t.Ref || rec.SHA256 == "" {
return errors.New("migration backup receipt mismatch")
}
if m.Record == nil {
return errors.New("backup recorder unavailable")
}
if err = m.Record(ctx, Backup{ID: t.ID, Server: op.Server, Owner: op.Owner, Reason: "pre_restore", Receipt: rec}); err != nil {
return err
}
op.Backup = rec
op.State = "restoring"
op.Stage = op.State
return m.save(ctx, s, *op, true)
case "restoring":
if err := m.ensureTarget(ctx, s, op); err != nil {
return err
}
var j batchv1.Job
err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: name + "-restore"}, &j)
if apierrors.IsNotFound(err) {
p := restore.JobParams{Server: op.Server, WorldPVC: op.TargetPVC, BackupRef: op.Backup.Ref, ArchiveStore: "tarLocal", Namespace: m.Namespace, ServiceAccount: "felis-restore", Image: m.Image, WorldsRoot: "/world", BackupRoot: m.Archive.Root, Deadline: 2 * time.Hour}
p, err = m.restoreParams(ctx, p)
if err != nil {
return err
}
if p.SHA256 != op.Backup.SHA256 {
return errors.New("migration archive digest changed")
}
job, err := restore.RestoreJob(p)
if err != nil {
return err
}
job.Name = name + "-restore"
m.pin(&job.Spec.Template.Spec, op.TargetNode)
job.Labels[LabelTransfer] = "true"
job.Spec.Template.Labels[LabelTransfer] = "true"
job.Labels[MigrationAnnotation] = op.ID
job.Spec.Template.Labels[MigrationAnnotation] = op.ID
job.Spec.TTLSecondsAfterFinished = nil
return m.Client.Create(ctx, job)
}
if err != nil {
return err
}
if !maintenance.JobFinished(&j) {
return nil
}
if !jobSucceeded(&j) {
return errors.New("target restore or read-back verification failed")
}
// Job Complete alone is not enough for a RWO handoff: all its processes must be gone.
if err := m.quiet(ctx, s); err != nil {
return nil
}
var pvc corev1.PersistentVolumeClaim
if err = m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: op.TargetPVC}, &pvc); err != nil {
return err
}
if err = m.volumeOnNode(ctx, pvc.Spec.VolumeName, op.TargetNode); err != nil {
return err
}
op.State = "switching"
op.Stage = op.State
return m.save(ctx, s, *op, true)
case "switching":
var sts appsv1.StatefulSet
err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: s.Name}, &sts)
if err == nil {
if sts.Spec.Replicas != nil && *sts.Spec.Replicas != 0 {
return ErrBusy
}
if sts.Status.Replicas != 0 {
return nil
}
policy := metav1.DeletePropagationForeground
return m.Client.Delete(ctx, &sts, &client.DeleteOptions{PropagationPolicy: &policy, Preconditions: &metav1.Preconditions{UID: &sts.UID, ResourceVersion: &sts.ResourceVersion}})
}
if !apierrors.IsNotFound(err) {
return err
}
// One optimistic CR write commits node, claim and progress. No failure can roll this back.
target := s.DeepCopy()
committed := *op
target.Spec.NodeName = op.TargetNode
target.Spec.Storage.ClaimName = op.TargetPVC
committed.Switched = true
committed.State = "succeeded"
committed.Stage = "succeeded"
committed.Updated = time.Now().UTC()
raw, _ := json.Marshal(persistedOperation{Operation: committed, OwnerID: op.Owner})
target.Annotations[MigrationAnnotation] = string(raw)
delete(target.Annotations, maintenance.Annotation)
if err := m.Client.Patch(ctx, target, client.MergeFromWithOptions(s.DeepCopy(), client.MergeFromWithOptimisticLock{})); err != nil {
return err
}
*s, *op = *target, committed
return nil
default:
return fmt.Errorf("unknown migration stage %q", op.State)
}
}
func (m *Manager) ensureTarget(ctx context.Context, s *v1alpha1.MinecraftServer, op *Operation) error {
var pvc corev1.PersistentVolumeClaim
err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: op.TargetPVC}, &pvc)
if err == nil {
if pvc.Labels[MigrationAnnotation] != op.ID {
return errors.New("target PVC identity mismatch")
}
return nil
}
if !apierrors.IsNotFound(err) {
return err
}
var source corev1.PersistentVolumeClaim
if err = m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: op.SourcePVC}, &source); err != nil {
return err
}
size := source.Spec.Resources.Requests[corev1.ResourceStorage]
if size.IsZero() {
size = resource.MustParse("8Gi")
}
pvc = corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: op.TargetPVC, Namespace: m.Namespace, Labels: map[string]string{maintenance.LabelServer: s.Name, MigrationAnnotation: op.ID}}, Spec: corev1.PersistentVolumeClaimSpec{AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce}, StorageClassName: source.Spec.StorageClassName, Resources: corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: size}}}}
return m.Client.Create(ctx, &pvc)
}
func (m *Manager) expireMigrationJobs(ctx context.Context, id string) error {
var jobs batchv1.JobList
if err := m.Client.List(ctx, &jobs, client.InNamespace(m.Namespace), client.MatchingLabels{MigrationAnnotation: id}); err != nil {
return err
}
for i := range jobs.Items {
j := &jobs.Items[i]
if j.Spec.TTLSecondsAfterFinished != nil || !maintenance.JobFinished(j) {
continue
}
before := j.DeepCopy()
ttl := int32(600)
j.Spec.TTLSecondsAfterFinished = &ttl
if err := m.Client.Patch(ctx, j, client.MergeFrom(before)); err != nil {
return err
}
}
return nil
}
+352
View File
@@ -0,0 +1,352 @@
package distributed
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/archivetransfer"
"felis.lolicon.best/internal/backup"
"felis.lolicon.best/internal/backupjob"
"felis.lolicon.best/internal/maintenance"
"felis.lolicon.best/internal/placement"
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
)
func fixture(t *testing.T) (*Manager, context.Context) {
t.Helper()
scheme := runtime.NewScheme()
for _, add := range []func(*runtime.Scheme) error{corev1.AddToScheme, batchv1.AddToScheme, appsv1.AddToScheme, v1alpha1.AddToScheme} {
if err := add(scheme); err != nil {
t.Fatal(err)
}
}
node := func(name string) *corev1.Node {
return &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{placement.LabelIdentity: name, placement.LabelRole: placement.RoleWorker, placement.LabelApproved: "true", "kubernetes.io/hostname": name}}, Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}}}
}
s := &v1alpha1.MinecraftServer{ObjectMeta: metav1.ObjectMeta{Name: "alice", Namespace: "minecraft"}, Spec: v1alpha1.MinecraftServerSpec{DesiredState: v1alpha1.DesiredStopped, NodeName: "b"}, Status: v1alpha1.MinecraftServerStatus{Phase: v1alpha1.PhaseStopped}}
s.Spec.Storage.Size = "1Gi"
pvc := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: s.WorldPVC(), Namespace: "minecraft"}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "source", Resources: corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1Gi")}}}}
zero := int32(0)
cl := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(s, &batchv1.Job{}).WithObjects(node("b"), node("c"), s, pvc, localPV("source", "b"), &appsv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: "alice", Namespace: "minecraft"}, Spec: appsv1.StatefulSetSpec{Replicas: &zero}}, &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "alice", Namespace: "minecraft"}, Spec: corev1.ServiceSpec{ClusterIP: "10.43.0.80"}}).Build()
var receipt archivetransfer.Receipt
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/inspect" && r.Method == "POST" || strings.HasPrefix(r.URL.Path, "/receipts/") {
var jobs batchv1.JobList
cl.List(r.Context(), &jobs)
for _, j := range jobs.Items {
var ticket archivetransfer.Ticket
if json.Unmarshal([]byte(j.Annotations[archivetransfer.Annotation]), &ticket) == nil && ticket.Ref != "" {
receipt = archivetransfer.Receipt{Ref: ticket.Ref, SHA256: strings.Repeat("a", 64), Size: 50}
}
}
if receipt.Ref == "" {
http.NotFound(w, r)
return
}
json.NewEncoder(w).Encode(receipt)
return
}
http.NotFound(w, r)
}))
t.Cleanup(server.Close)
m := &Manager{Client: cl, Namespace: "minecraft", Controller: "a", Image: "registry.local/felis:1", Archive: archivetransfer.Client{Root: "/backups", URL: server.URL, Key: strings.Repeat("k", 32)}, Record: func(context.Context, Backup) error { return nil }}
m.Resolve = placement.Resolve(cl, "minecraft")
return m, context.Background()
}
func localPV(name, node string) *corev1.PersistentVolume {
return &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: name}, Spec: corev1.PersistentVolumeSpec{NodeAffinity: &corev1.VolumeNodeAffinity{Required: &corev1.NodeSelector{NodeSelectorTerms: []corev1.NodeSelectorTerm{{MatchExpressions: []corev1.NodeSelectorRequirement{{Key: "kubernetes.io/hostname", Operator: corev1.NodeSelectorOpIn, Values: []string{node}}}}}}}}}
}
func getServer(t *testing.T, m *Manager, ctx context.Context) *v1alpha1.MinecraftServer {
t.Helper()
var s v1alpha1.MinecraftServer
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: "alice"}, &s); err != nil {
t.Fatal(err)
}
return &s
}
func TestMigrationSurvivesRestartAndKeepsIdentity(t *testing.T) {
m, ctx := fixture(t)
op, err := m.BeginMigration(ctx, "alice", "c", "owner")
if err != nil {
t.Fatal(err)
}
if again, err := m.BeginMigration(ctx, "alice", "c", "owner"); err != nil || again.ID != op.ID {
t.Fatalf("duplicate %+v: %v", again, err)
}
s := getServer(t, m, ctx)
if kind, held := maintenance.Holder(s.Name, s.Annotations, nil, time.Now().Add(365*24*time.Hour)); !held || kind != maintenance.KindMigration {
t.Fatal("migration lock expired")
}
if _, err := m.BeginMigration(ctx, "alice", "b", "owner"); !errors.Is(err, ErrBusy) {
t.Fatalf("competing migration: %v", err)
}
// A restart reconstructs the coordinator solely from persisted CR and Job state.
restarted := *m
m = &restarted
for i := 0; i < 3; i++ {
if err := m.ReconcileMigrations(ctx); err != nil {
t.Fatal(err)
}
}
current, err := m.Migration(ctx, "alice", op.ID)
if err != nil || current.State != "restoring" {
t.Fatalf("progress %+v: %v", current, err)
}
var pvc corev1.PersistentVolumeClaim
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: op.TargetPVC}, &pvc); err != nil {
t.Fatal(err)
}
pvc.Spec.VolumeName = "target"
if err := m.Client.Update(ctx, &pvc); err != nil {
t.Fatal(err)
}
if err := m.Client.Create(ctx, localPV("target", "c")); err != nil {
t.Fatal(err)
}
var jobs batchv1.JobList
if err := m.Client.List(ctx, &jobs); err != nil {
t.Fatal(err)
}
for i := range jobs.Items {
j := &jobs.Items[i]
assertJobIsolation(t, j)
j.Status.Conditions = []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}}
if err := m.Client.Status().Update(ctx, j); err != nil {
t.Fatal(err)
}
}
for i := 0; i < 3; i++ {
if err := m.ReconcileMigrations(ctx); err != nil {
t.Fatal(err)
}
}
s = getServer(t, m, ctx)
if s.Spec.NodeName != "c" || s.WorldPVC() != op.TargetPVC || s.Spec.DesiredState != v1alpha1.DesiredStopped || s.Status.Ready {
t.Fatalf("unsafe switch %+v", s)
}
if _, held := maintenance.Holder(s.Name, s.Annotations, nil, time.Now()); held {
t.Fatal("success did not release lock")
}
var svc corev1.Service
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: "alice"}, &svc); err != nil || svc.Spec.ClusterIP != "10.43.0.80" {
t.Fatalf("service changed: %v", err)
}
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: op.SourcePVC}, &pvc); err != nil {
t.Fatal("source PVC removed", err)
}
var sts appsv1.StatefulSet
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: "alice"}, &sts); !apierrors.IsNotFound(err) {
t.Fatal("stopped immutable StatefulSet not removed", err)
}
}
func assertJobIsolation(t *testing.T, j *batchv1.Job) {
t.Helper()
p := j.Spec.Template.Spec
if p.AutomountServiceAccountToken == nil || *p.AutomountServiceAccountToken || p.HostNetwork || p.HostPID || p.HostIPC {
t.Fatal("maintenance has host credentials/namespaces")
}
claims := 0
for _, v := range p.Volumes {
if v.Secret != nil || v.HostPath != nil {
t.Fatal("maintenance carries secret or host mount")
}
if v.PersistentVolumeClaim != nil {
claims++
}
}
if claims != 1 {
t.Fatalf("cross-node job mounts %d PVCs", claims)
}
for _, c := range p.Containers {
if c.SecurityContext.AllowPrivilegeEscalation == nil || *c.SecurityContext.AllowPrivilegeEscalation {
t.Fatal("maintenance can escalate")
}
for _, e := range c.Env {
if strings.Contains(e.Name, "DATABASE") || e.Name == archivetransfer.KeyEnv {
t.Fatal("full credentials leaked")
}
}
}
}
func TestLostSourceFailsClosedAndRetryKeepsSource(t *testing.T) {
m, ctx := fixture(t)
op, err := m.BeginMigration(ctx, "alice", "c", "owner")
if err != nil {
t.Fatal(err)
}
var n corev1.Node
m.Client.Get(ctx, types.NamespacedName{Name: "b"}, &n)
n.Status.Conditions = nil
if err := m.Client.Status().Update(ctx, &n); err != nil {
t.Fatal(err)
}
if err := m.ReconcileMigrations(ctx); err != nil {
t.Fatal(err)
}
failed, err := m.Migration(ctx, "alice", op.ID)
if err != nil || failed.State != "failed" {
t.Fatalf("offline source %+v: %v", failed, err)
}
s := getServer(t, m, ctx)
if s.WorldPVC() != op.SourcePVC || s.Spec.NodeName != "b" {
t.Fatal("failed migration switched placement")
}
if _, held := maintenance.Holder(s.Name, s.Annotations, nil, time.Now().Add(time.Hour)); !held {
t.Fatal("failed migration lost lock")
}
if _, err := m.RetryMigration(ctx, "alice", op.ID); err != nil {
t.Fatal(err)
}
if err := m.ReconcileMigrations(ctx); err != nil {
t.Fatal(err)
}
if current, _ := m.Migration(ctx, "alice", op.ID); current.State != "failed" {
t.Fatal("retry started tasks on offline source")
}
}
func TestMigrationRejectsLiveProcessAndUnapprovedTarget(t *testing.T) {
m, ctx := fixture(t)
if err := m.Client.Create(ctx, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "alice-0", Namespace: m.Namespace, Labels: map[string]string{v1alpha1.LabelServer: "alice", v1alpha1.LabelComponent: "server"}}, Status: corev1.PodStatus{Phase: corev1.PodRunning}}); err != nil {
t.Fatal(err)
}
if _, err := m.BeginMigration(ctx, "alice", "c", "owner"); !errors.Is(err, ErrBusy) {
t.Fatal("live source accepted", err)
}
var n corev1.Node
m.Client.Get(ctx, types.NamespacedName{Name: "c"}, &n)
delete(n.Labels, placement.LabelApproved)
m.Client.Update(ctx, &n)
if err := m.ValidateNode(ctx, "c"); err == nil {
t.Fatal("unapproved worker accepted")
}
}
func TestMigrationFailureRetainsSourceAndCanRetry(t *testing.T) {
for _, stage := range []string{"restoring", "switching"} {
t.Run(stage, func(t *testing.T) {
m, ctx := fixture(t)
op, err := m.BeginMigration(ctx, "alice", "c", "owner")
if err != nil {
t.Fatal(err)
}
for i := 0; i < 3; i++ {
if err := m.ReconcileMigrations(ctx); err != nil {
t.Fatal(err)
}
}
var jobs batchv1.JobList
if err := m.Client.List(ctx, &jobs); err != nil {
t.Fatal(err)
}
for i := range jobs.Items {
j := &jobs.Items[i]
condition := batchv1.JobComplete
if stage == "restoring" && strings.HasSuffix(j.Name, "-restore") {
condition = batchv1.JobFailed
}
j.Status.Conditions = []batchv1.JobCondition{{Type: condition, Status: corev1.ConditionTrue}}
if err := m.Client.Status().Update(ctx, j); err != nil {
t.Fatal(err)
}
}
if stage == "switching" {
var pvc corev1.PersistentVolumeClaim
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: op.TargetPVC}, &pvc); err != nil {
t.Fatal(err)
}
pvc.Spec.VolumeName = "target"
if err := m.Client.Update(ctx, &pvc); err != nil {
t.Fatal(err)
}
if err := m.Client.Create(ctx, localPV("target", "c")); err != nil {
t.Fatal(err)
}
fail := true
m.Client = interceptor.NewClient(m.Client.(client.WithWatch), interceptor.Funcs{
Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
if s, ok := obj.(*v1alpha1.MinecraftServer); ok && s.Spec.NodeName == "c" && fail {
fail = false
return errors.New("commit interrupted")
}
return c.Patch(ctx, obj, patch, opts...)
},
})
}
for i := 0; i < 3; i++ {
if err := m.ReconcileMigrations(ctx); err != nil {
t.Fatal(err)
}
}
failed, err := m.Migration(ctx, "alice", op.ID)
if err != nil || failed.State != "failed" || failed.Stage != stage || failed.Switched || failed.Error == "" {
t.Fatalf("failure %+v: %v", failed, err)
}
s := getServer(t, m, ctx)
if s.WorldPVC() != op.SourcePVC || s.Spec.NodeName != "b" || s.Spec.DesiredState != v1alpha1.DesiredStopped {
t.Fatal("failed operation changed active source")
}
if _, held := maintenance.Holder(s.Name, s.Annotations, nil, time.Now().Add(30*24*time.Hour)); !held {
t.Fatal("failed operation released lock")
}
archiver := &RemoteArchiver{TarLocal: &backup.TarLocal{BackupRoot: t.TempDir()}, Manager: m}
if err := archiver.Delete(ctx, backup.ArchiveRef(failed.Backup.Ref)); !errors.Is(err, ErrBusy) {
t.Fatalf("failed migration lost its safety archive: %v", err)
}
if retry, err := m.RetryMigration(ctx, "alice", op.ID); err != nil || retry.State != stage || retry.Attempt != 1 {
t.Fatalf("retry %+v: %v", retry, err)
}
if err := m.ReconcileMigrations(ctx); err != nil {
t.Fatal(err)
}
if stage == "switching" {
if done, _ := m.Migration(ctx, "alice", op.ID); done.State != "succeeded" || !done.Switched {
t.Fatalf("commit retry %+v", done)
}
}
})
}
}
func TestBackupResolvesActiveClaimAndPreservesConflictContract(t *testing.T) {
m, ctx := fixture(t)
p := backupjob.JobParams{Server: "alice", WorldPVC: "stale-claim", JobName: "manual-backup"}
if err := m.CreateBackupJob(ctx, p); err != nil {
t.Fatal(err)
}
var j batchv1.Job
if err := m.Client.Get(ctx, types.NamespacedName{Namespace: m.Namespace, Name: p.JobName}, &j); err != nil {
t.Fatal(err)
}
assertJobIsolation(t, &j)
for _, v := range j.Spec.Template.Spec.Volumes {
if v.PersistentVolumeClaim != nil && v.PersistentVolumeClaim.ClaimName != getServer(t, m, ctx).WorldPVC() {
t.Fatal("backup used a stale world claim")
}
}
if err := m.CreateBackupJob(ctx, p); !errors.Is(err, backupjob.ErrAlreadyExists) {
t.Fatalf("duplicate backup: %v", err)
}
}
+17
View File
@@ -52,6 +52,7 @@ import (
"time"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
)
// Errors the Editor returns, which internal/api maps onto HTTP status codes
@@ -106,6 +107,7 @@ type Runner interface {
// deployment-specific, and when it is empty cmd/felis leaves the API's FileEditor
// nil so the endpoints report 503 rather than creating a Job that cannot run.
type Config struct {
ResolveWorld placement.Resolver
// Namespace is where the world PVCs live and the Job runs (the minecraft
// namespace), co-located with the world it edits.
Namespace string
@@ -338,6 +340,14 @@ func (e *Editor) run(ctx context.Context, server string, p JobParams) (Result, e
return Result{}, err
}
if cfg.ResolveWorld != nil {
world, err := cfg.ResolveWorld(ctx, server)
if err != nil {
return Result{}, err
}
p.WorldPVC, p.NodeName = world.Claim, world.Node
}
// Bound the wait here rather than trusting the caller's context: this is an HTTP
// handler's goroutine and the Pod it waits on may never become ready (an
// unschedulable node, an unpullable image). The Job's own activeDeadlineSeconds
@@ -438,6 +448,13 @@ func (e *Editor) start(ctx context.Context, server string, p JobParams) (OpState
if err != nil {
return OpState{}, err
}
if cfg.ResolveWorld != nil {
world, err := cfg.ResolveWorld(ctx, server)
if err != nil {
return OpState{}, err
}
p.WorldPVC, p.NodeName = world.Claim, world.Node
}
p.Async = true
p.Deadline, p.TTLAfterFinished, p.CPULimit = cfg.AsyncDeadline, cfg.AsyncTTL, cfg.AsyncCPULimit
if err := e.Runner.Start(ctx, p); err != nil {
+11 -1
View File
@@ -5,6 +5,7 @@ import (
"strconv"
"time"
"felis.lolicon.best/internal/placement"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
@@ -43,7 +44,8 @@ const (
// operation + Config by the Editor. jobspec is a pure function of them so the
// security-critical Job shape is unit-tested without a cluster.
type JobParams struct {
Server string
NodeName string
Server string
// OpID is the per-invocation identifier that both names the Job and labels its
// Pod. See Editor.run for why every invocation gets a fresh one.
OpID string
@@ -273,6 +275,7 @@ func FilesJob(p JobParams) (*batchv1.Job, error) {
ObjectMeta: metav1.ObjectMeta{Labels: filesLabels(p)},
Spec: corev1.PodSpec{
RestartPolicy: corev1.RestartPolicyNever,
NodeSelector: jobNodeSelector(p.NodeName),
ServiceAccountName: p.ServiceAccount,
AutomountServiceAccountToken: boolPtr(false),
SecurityContext: filesPodSecurityContext(p),
@@ -356,3 +359,10 @@ func filesAnnotations(p JobParams) map[string]string {
}
return map[string]string{AnnotationPath: p.Path}
}
func jobNodeSelector(name string) map[string]string {
if name == "" {
return nil
}
return map[string]string{placement.LabelIdentity: name}
}
+5 -1
View File
@@ -90,6 +90,7 @@ const (
// Kinds of holder.
const (
KindMigration = "migration"
KindRestore = "restore"
KindBackup = "backup"
KindFileWrite = "file-write"
@@ -191,6 +192,9 @@ func Holder(server string, annotations map[string]string, jobs []batchv1.Job, no
if j.Labels[LabelServer] != server {
continue
}
if j.Labels["felis.lolicon.best/archive-pending"] == "true" {
return KindBackup, true
}
if RestorePending(j) {
return KindRestore, true
}
@@ -202,7 +206,7 @@ func Holder(server string, annotations map[string]string, jobs []batchv1.Job, no
}
}
if v, ok := annotations[Annotation]; ok {
if kind, at, ok := parseLock(v); ok && now.Sub(at) < Grace && at.Sub(now) < Grace {
if kind, at, ok := parseLock(v); ok && (kind == KindMigration || (now.Sub(at) < Grace && at.Sub(now) < Grace)) {
return kind, true
}
}
+17 -2
View File
@@ -8,6 +8,7 @@ import (
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
@@ -217,7 +218,7 @@ func healthHandler(server *v1alpha1.MinecraftServer) corev1.ProbeHandler {
// graceful shutdown is terminationGracePeriodSeconds, the time the server gets to
// save on SIGTERM; the reconciler flushes the world over RCON before it scales to
// zero (saveBeforeStop).
func buildStatefulSet(server *v1alpha1.MinecraftServer, replicas int32, felisImage string) (*appsv1.StatefulSet, error) {
func buildStatefulSet(server *v1alpha1.MinecraftServer, replicas int32, felisImage string, gateProbe ...string) (*appsv1.StatefulSet, error) {
storageSize := server.Spec.Storage.Size
if storageSize == "" {
storageSize = defaultStorageSize
@@ -278,7 +279,14 @@ func buildStatefulSet(server *v1alpha1.MinecraftServer, replicas int32, felisIma
if server.Labels[v1alpha1.LabelSystemRole] == "" {
initContainers = append(initContainers, forwardingInitContainer(felisImage))
}
initContainers = append(initContainers, egressGateInitContainer(felisImage))
gate := egressGateInitContainer(felisImage)
if server.Spec.NodeName != "" || (len(gateProbe) > 0 && gateProbe[0] != "") {
gate.Command = append(gate.Command[:len(gate.Command)-1], "--positive-probe", "kube-dns.kube-system.svc:53")
if len(gateProbe) > 0 && gateProbe[0] != "" {
gate.Command = append(gate.Command, "--probe", gateProbe[0])
}
}
initContainers = append(initContainers, gate)
}
grace := graceSeconds(server)
@@ -334,6 +342,13 @@ func buildStatefulSet(server *v1alpha1.MinecraftServer, replicas int32, felisIma
},
},
}
if server.Spec.Storage.ClaimName != "" {
sts.Spec.VolumeClaimTemplates = nil
sts.Spec.Template.Spec.Volumes = append(sts.Spec.Template.Spec.Volumes, corev1.Volume{Name: dataVolumeName, VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: server.WorldPVC()}}})
}
if server.Spec.NodeName != "" {
sts.Spec.Template.Spec.NodeSelector = map[string]string{placement.LabelIdentity: server.Spec.NodeName}
}
return sts, nil
}
@@ -6,9 +6,39 @@ import (
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
corev1 "k8s.io/api/core/v1"
)
func TestMigratedWorldPlacementAndFailClosedGate(t *testing.T) {
s := &v1alpha1.MinecraftServer{}
s.Spec.NodeName = "c"
s.Spec.Storage.ClaimName = "world-alice-migrated"
sts, err := buildStatefulSet(s, 1, "felis:test", "felis-api.felis.svc:443")
if err != nil {
t.Fatal(err)
}
if len(sts.Spec.VolumeClaimTemplates) != 0 || sts.Spec.Template.Spec.NodeSelector[placement.LabelIdentity] != "c" {
t.Fatal("migration rebuilt or moved the wrong world")
}
found := false
for _, v := range sts.Spec.Template.Spec.Volumes {
if v.PersistentVolumeClaim != nil && v.PersistentVolumeClaim.ClaimName == s.Spec.Storage.ClaimName {
found = true
}
}
if !found {
t.Fatal("active PVC not mounted")
}
for _, c := range sts.Spec.Template.Spec.InitContainers {
if c.Name == "egress-gate" {
if slices.Contains(c.Command, "--fail-open") || !slices.Contains(c.Command, "--positive-probe") {
t.Fatal("distributed gate can fail open")
}
}
}
}
// findEnv returns the env var with the given name, or nil.
func findEnv(env []corev1.EnvVar, name string) *corev1.EnvVar {
for i := range env {
+49 -1
View File
@@ -18,6 +18,7 @@ import (
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/maintenance"
"felis.lolicon.best/internal/metrics"
"felis.lolicon.best/internal/placement"
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
@@ -116,6 +117,9 @@ func podTemplateStamp(tmpl *corev1.PodTemplateSpec, felisImage string) (string,
// Reconciler reconciles a MinecraftServer with its managed children.
type Reconciler struct {
Nodes client.Reader
EgressProbe string
ControllerNode string
client.Client
Scheme *runtime.Scheme
// Prober gates readiness on RCON reachability.
@@ -312,6 +316,46 @@ func (r *Reconciler) reconcileRunning(ctx context.Context, server *v1alpha1.Mine
return ctrl.Result{RequeueAfter: requeueMaintenance}, nil
}
if r.Nodes != nil {
node := server.Spec.NodeName
var pod corev1.Pod
if err := r.podReader().Get(ctx, types.NamespacedName{Namespace: server.Namespace, Name: server.Name + "-0"}, &pod); err == nil {
server.Status.NodeName = pod.Spec.NodeName
if node == "" {
node = pod.Spec.NodeName
}
} else if !apierrors.IsNotFound(err) {
return ctrl.Result{}, err
}
if node == "" {
node = r.ControllerNode
}
if node != "" {
var n corev1.Node
err := r.Nodes.Get(ctx, types.NamespacedName{Name: node}, &n)
if err != nil && !apierrors.IsNotFound(err) {
return ctrl.Result{}, err
}
if err != nil || !placement.Admitted(&n, r.ControllerNode) {
var svc corev1.Service
if err := r.Get(ctx, types.NamespacedName{Namespace: server.Namespace, Name: server.Name}, &svc); err == nil {
if svc.Spec.Selector == nil {
svc.Spec.Selector = map[string]string{}
}
svc.Spec.Selector["felis.lolicon.best/node-online"] = "true"
if err := r.Update(ctx, &svc); err != nil {
return ctrl.Result{}, err
}
} else if !apierrors.IsNotFound(err) {
return ctrl.Result{}, err
}
server.Status.Ready = false
server.Status.Endpoint = v1alpha1.EndpointStatus{Mode: v1alpha1.EndpointFallback}
r.setCondition(server, v1alpha1.ConditionReady, metav1.ConditionFalse, "NodeUnavailable", "execution node is offline; automatic relocation is disabled")
return ctrl.Result{RequeueAfter: 10 * time.Second}, r.patchStatus(ctx, server)
}
}
}
endpointAddress, err := r.ensureServices(ctx, server)
if err != nil {
return ctrl.Result{}, err
@@ -330,7 +374,11 @@ func (r *Reconciler) reconcileRunning(ctx context.Context, server *v1alpha1.Mine
return ctrl.Result{RequeueAfter: requeueSecret}, nil
}
desired, err := buildStatefulSet(server, 1, r.FelisImage)
placed := server.DeepCopy()
if placed.Spec.NodeName == "" {
placed.Spec.NodeName = r.ControllerNode
}
desired, err := buildStatefulSet(placed, 1, r.FelisImage, r.EgressProbe)
if err != nil {
// A malformed spec (e.g. bad storage quantity) is terminal until edited.
r.markFailed(server, "InvalidSpec", err.Error())
+5 -1
View File
@@ -20,6 +20,7 @@ import (
var static embed.FS
type runtimeConfig struct {
Distributed bool `json:"distributed,omitempty"`
APIBase string `json:"apiBase"`
RootDomain string `json:"rootDomain"`
PanelHostname string `json:"panelHostname,omitempty"`
@@ -112,13 +113,14 @@ func parseBuildVersion(raw string) buildInfo {
// right surface (player console vs SysAdmin console) without a rebuild. gamePort
// is the public Minecraft port ([velocity] game_port), which the SPA appends to
// the server addresses players copy; 0 or 25565 leaves them bare.
func Handler(api http.Handler, rootDomain, panelHost, adminHost string, gamePort int, version string) http.Handler {
func Handler(api http.Handler, rootDomain, panelHost, adminHost string, gamePort int, version string, distributed ...bool) http.Handler {
files, err := fs.Sub(static, "static")
if err != nil {
panic(err)
}
return &handler{
api: api,
distributed: len(distributed) > 0 && distributed[0],
rootDomain: rootDomain,
panelHostname: panelHost,
adminHostname: adminHost,
@@ -143,6 +145,7 @@ func publicGamePort(p int) int {
}
type handler struct {
distributed bool
api http.Handler
rootDomain string
panelHostname string
@@ -223,6 +226,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store")
_ = json.NewEncoder(w).Encode(runtimeConfig{
APIBase: "/api/v1",
Distributed: h.distributed,
RootDomain: h.rootDomain,
PanelHostname: h.panelHostname,
AdminHostname: h.adminHostname,
+84
View File
@@ -0,0 +1,84 @@
// Package placement resolves worlds and checks the protected node admission labels.
package placement
import (
"context"
"fmt"
"felis.lolicon.best/internal/apis/felis/v1alpha1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
)
const (
LabelRole = "felis.node-restriction.kubernetes.io/role"
LabelIdentity = "felis.node-restriction.kubernetes.io/identity"
LabelApproved = "felis.node-restriction.kubernetes.io/approved"
RoleController = "controller"
RoleWorker = "worker"
)
type World struct{ Claim, Node string }
type Resolver func(context.Context, string) (World, error)
func Online(n *corev1.Node) bool {
for _, c := range n.Status.Conditions {
if c.Type == corev1.NodeReady {
return c.Status == corev1.ConditionTrue && n.DeletionTimestamp == nil
}
}
return false
}
// Admitted accepts only A's protected controller identity or an approved worker.
func Admitted(n *corev1.Node, controller string) bool {
if !Online(n) || n.Labels[LabelIdentity] != n.Name {
return false
}
if n.Name == controller {
return n.Labels[LabelRole] == RoleController
}
return n.Labels[LabelRole] == RoleWorker && n.Labels[LabelApproved] == "true"
}
func Worker(ctx context.Context, r client.Reader, name string) error {
var n corev1.Node
if err := r.Get(ctx, types.NamespacedName{Name: name}, &n); err != nil {
return err
}
if !Admitted(&n, "") || n.Spec.Unschedulable {
return fmt.Errorf("node %q is not an online, approved worker", name)
}
return nil
}
func Resolve(r client.Reader, namespace string, controller ...string) Resolver {
return func(ctx context.Context, name string) (World, error) {
var s v1alpha1.MinecraftServer
if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, &s); err != nil {
return World{}, err
}
node := s.Spec.NodeName
if node == "" {
node = s.Status.NodeName
}
if node == "" && len(controller) > 0 {
node = controller[0]
}
if node != "" {
var n corev1.Node
if err := r.Get(ctx, types.NamespacedName{Name: node}, &n); err != nil {
return World{}, err
}
a := ""
if len(controller) > 0 {
a = controller[0]
}
if !Admitted(&n, a) {
return World{}, fmt.Errorf("node %q is offline", node)
}
}
return World{Claim: s.WorldPVC(), Node: node}, nil
}
}
+5
View File
@@ -72,6 +72,11 @@ func Objects(p Params) []Object {
objs = append(objs, rb)
}
if p.Distributed {
objs = append(objs, DistributedRBAC(p)...)
objs = append(objs, ArchiveNetworkPolicies(p)...)
}
// Weak Job SAs. They come from the build/restore packages (single source of
// truth for AutomountServiceAccountToken=false), which set ObjectMeta but not
// TypeMeta — stamp it so the YAML header is present. The restore Job runs in
+82
View File
@@ -0,0 +1,82 @@
package platform
import (
"fmt"
"felis.lolicon.best/internal/archivetransfer"
"felis.lolicon.best/internal/distributed"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
networkingv1 "k8s.io/api/networking/v1"
rbacv1 "k8s.io/api/rbac/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/intstr"
)
const ArchiveName = "felis-archive"
const ArchiveSecret = "felis-archive-key"
const ArchivePort int32 = 8090
func archiveKeyEnv() corev1.EnvVar {
return corev1.EnvVar{Name: archivetransfer.KeyEnv, ValueFrom: &corev1.EnvVarSource{SecretKeyRef: &corev1.SecretKeySelector{LocalObjectReference: corev1.LocalObjectReference{Name: ArchiveSecret}, Key: "key"}}}
}
func distributedEnv(p Params) []corev1.EnvVar {
return []corev1.EnvVar{{Name: "FELIS_DISTRIBUTED", Value: "true"}, {Name: "FELIS_CONTROLLER_NODE", Value: p.ControllerNode}, {Name: "FELIS_ARCHIVE_URL", Value: fmt.Sprintf("http://%s.%s.svc:%d", ArchiveName, p.MinecraftNamespace, ArchivePort)}, {Name: "FELIS_EGRESS_PROBE", Value: p.EgressProbe}, archiveKeyEnv()}
}
func ArchiveDeployment(p Params) *appsv1.Deployment {
labels := map[string]string{"app.kubernetes.io/name": ArchiveName}
container := corev1.Container{Name: ArchiveName, Image: p.FelisImage, Command: []string{felisBinaryPath, "archive-serve"}, Args: []string{"--root", p.ArchiveLocalPath}, Env: []corev1.EnvVar{archiveKeyEnv()}, Ports: []corev1.ContainerPort{{Name: "archive", ContainerPort: ArchivePort}}, Resources: controlPlaneResources(), SecurityContext: &corev1.SecurityContext{RunAsUser: int64Ptr(0), RunAsNonRoot: boolPtr(false), AllowPrivilegeEscalation: boolPtr(false), Privileged: boolPtr(false), ReadOnlyRootFilesystem: boolPtr(true), Capabilities: &corev1.Capabilities{Drop: []corev1.Capability{"ALL"}}}, VolumeMounts: []corev1.VolumeMount{{Name: "archives", MountPath: p.ArchiveLocalPath}}}
container.ReadinessProbe = &corev1.Probe{ProbeHandler: corev1.ProbeHandler{HTTPGet: &corev1.HTTPGetAction{Path: "/healthz", Port: intstr.FromString("archive")}}}
container.LivenessProbe = container.ReadinessProbe.DeepCopy()
return &appsv1.Deployment{TypeMeta: metav1.TypeMeta{APIVersion: "apps/v1", Kind: "Deployment"}, ObjectMeta: metav1.ObjectMeta{Name: ArchiveName, Namespace: p.MinecraftNamespace, Labels: labels}, Spec: appsv1.DeploymentSpec{Replicas: int32Ptr(1), Strategy: appsv1.DeploymentStrategy{Type: appsv1.RecreateDeploymentStrategyType}, Selector: &metav1.LabelSelector{MatchLabels: labels}, Template: corev1.PodTemplateSpec{ObjectMeta: metav1.ObjectMeta{Labels: labels}, Spec: corev1.PodSpec{NodeSelector: controllerSelector(p), AutomountServiceAccountToken: boolPtr(false), SecurityContext: &corev1.PodSecurityContext{SeccompProfile: &corev1.SeccompProfile{Type: corev1.SeccompProfileTypeRuntimeDefault}}, Containers: []corev1.Container{container}, Volumes: []corev1.Volume{{Name: "archives", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: p.BackupPVC}}}}}}}}
}
func ArchiveService(p Params) *corev1.Service {
return &corev1.Service{TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "Service"}, ObjectMeta: metav1.ObjectMeta{Name: ArchiveName, Namespace: p.MinecraftNamespace}, Spec: corev1.ServiceSpec{Selector: map[string]string{"app.kubernetes.io/name": ArchiveName}, Ports: []corev1.ServicePort{{Name: "archive", Port: ArchivePort, TargetPort: intstr.FromInt32(ArchivePort)}}}}
}
// Cluster grants are read-only and bound only to A's API/operator/reaper identities.
func DistributedRBAC(p Params) []Object {
var out []Object
for _, entry := range []struct {
name, ns string
rules []rbacv1.PolicyRule
}{
{SAAPI, p.ControlNamespace, []rbacv1.PolicyRule{rule([]string{groupCore}, []string{"nodes"}, []string{"get", "list"}), rule([]string{groupCore}, []string{"persistentvolumes"}, []string{"get"})}},
{SAOperator, p.ControlNamespace, []rbacv1.PolicyRule{rule([]string{groupCore}, []string{"nodes"}, []string{"get"})}},
{SAReaper, p.MinecraftNamespace, []rbacv1.PolicyRule{rule([]string{groupCore}, []string{"nodes"}, []string{"get"})}},
} {
name := entry.name + "-" + p.MinecraftNamespace + "-nodes"
out = append(out, &rbacv1.ClusterRole{TypeMeta: metav1.TypeMeta{APIVersion: "rbac.authorization.k8s.io/v1", Kind: "ClusterRole"}, ObjectMeta: metav1.ObjectMeta{Name: name}, Rules: entry.rules}, &rbacv1.ClusterRoleBinding{TypeMeta: metav1.TypeMeta{APIVersion: "rbac.authorization.k8s.io/v1", Kind: "ClusterRoleBinding"}, ObjectMeta: metav1.ObjectMeta{Name: name}, RoleRef: rbacv1.RoleRef{APIGroup: rbacv1.GroupName, Kind: "ClusterRole", Name: name}, Subjects: []rbacv1.Subject{{Kind: "ServiceAccount", Name: entry.name, Namespace: entry.ns}}})
}
return out
}
func ArchiveNetworkPolicies(p Params) []Object {
tcp, udp := corev1.ProtocolTCP, corev1.ProtocolUDP
archive := metav1.LabelSelector{MatchLabels: map[string]string{"app.kubernetes.io/name": ArchiveName}}
job := metav1.LabelSelector{MatchLabels: map[string]string{distributed.LabelTransfer: "true"}}
control := networkingv1.NetworkPolicyPeer{NamespaceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": p.ControlNamespace}}, PodSelector: &metav1.LabelSelector{MatchLabels: controlPlanePodLabels(ComponentAPI)}}
reaper := networkingv1.NetworkPolicyPeer{PodSelector: &metav1.LabelSelector{MatchLabels: controlPlanePodLabels(ComponentReaper)}}
peers := []networkingv1.NetworkPolicyPeer{control, reaper, {PodSelector: &job}}
ingress := &networkingv1.NetworkPolicy{TypeMeta: metav1.TypeMeta{APIVersion: "networking.k8s.io/v1", Kind: "NetworkPolicy"}, ObjectMeta: metav1.ObjectMeta{Name: "felis-archive", Namespace: p.MinecraftNamespace}, Spec: networkingv1.NetworkPolicySpec{PodSelector: archive, PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress}, Ingress: []networkingv1.NetworkPolicyIngressRule{{From: peers, Ports: []networkingv1.NetworkPolicyPort{{Protocol: &tcp, Port: portPtr(ArchivePort)}}}}}}
egress := &networkingv1.NetworkPolicy{TypeMeta: metav1.TypeMeta{APIVersion: "networking.k8s.io/v1", Kind: "NetworkPolicy"}, ObjectMeta: metav1.ObjectMeta{Name: "felis-archive-jobs", Namespace: p.MinecraftNamespace}, Spec: networkingv1.NetworkPolicySpec{PodSelector: job, PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeEgress}, Egress: []networkingv1.NetworkPolicyEgressRule{
{To: []networkingv1.NetworkPolicyPeer{{PodSelector: &archive}}, Ports: []networkingv1.NetworkPolicyPort{{Protocol: &tcp, Port: portPtr(ArchivePort)}}},
{To: []networkingv1.NetworkPolicyPeer{{NamespaceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": "kube-system"}}, PodSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"k8s-app": "kube-dns"}}}}, Ports: []networkingv1.NetworkPolicyPort{{Protocol: &udp, Port: portPtr(53)}, {Protocol: &tcp, Port: portPtr(53)}}},
}}}
// Files/export maintenance Pods may send results only to A's existing upload receiver.
// Transfer Pods are excluded so grants remain limited to the archive service.
maintenance := &networkingv1.NetworkPolicy{TypeMeta: metav1.TypeMeta{APIVersion: "networking.k8s.io/v1", Kind: "NetworkPolicy"}, ObjectMeta: metav1.ObjectMeta{Name: "felis-maintenance-egress", Namespace: p.MinecraftNamespace}, Spec: networkingv1.NetworkPolicySpec{
PodSelector: metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{
{Key: "app.kubernetes.io/managed-by", Operator: metav1.LabelSelectorOpIn, Values: []string{"felis-files", "felis-export", "felis-restore", "felis-backup"}},
{Key: distributed.LabelTransfer, Operator: metav1.LabelSelectorOpDoesNotExist},
}}, PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeEgress},
Egress: []networkingv1.NetworkPolicyEgressRule{
{To: []networkingv1.NetworkPolicyPeer{control}, Ports: []networkingv1.NetworkPolicyPort{{Protocol: &tcp, Port: portPtr(8081)}}},
egress.Spec.Egress[1],
},
}}
return []Object{ingress, egress, maintenance}
}
func portPtr(p int32) *intstr.IntOrString { v := intstr.FromInt32(p); return &v }
+61
View File
@@ -0,0 +1,61 @@
package platform
import (
"testing"
"felis.lolicon.best/internal/archivetransfer"
"felis.lolicon.best/internal/placement"
appsv1 "k8s.io/api/apps/v1"
rbacv1 "k8s.io/api/rbac/v1"
)
func TestDistributedControllerPinningAndArchivePowers(t *testing.T) {
p := testParams()
p.Distributed = true
p.ControllerNode = "a"
p.EgressProbe = "felis-api.felis.svc:443"
p.BackupPVC = "felis-backups"
p.ArchiveLocalPath = "/backups"
p.VelocityCIDRs = []string{"192.0.2.1/32"}
if err := p.Validate(); err != nil {
t.Fatal(err)
}
found := false
for _, obj := range Objects(p) {
if d, ok := obj.(*appsv1.Deployment); ok {
if d.Spec.Template.Spec.NodeSelector[placement.LabelIdentity] != "a" {
t.Fatalf("controller workload %s may run on worker", d.Name)
}
if d.Name == ArchiveName {
found = true
s := d.Spec.Template.Spec
if s.AutomountServiceAccountToken == nil || *s.AutomountServiceAccountToken || s.ServiceAccountName != "" || len(s.Volumes) != 1 || s.Volumes[0].PersistentVolumeClaim.ClaimName != p.BackupPVC {
t.Fatal("archive has extra powers")
}
for _, c := range s.Containers {
for _, e := range c.Env {
if e.Name != archivetransfer.KeyEnv {
t.Fatal("archive received controller config")
}
}
}
}
}
if r, ok := obj.(*rbacv1.ClusterRole); ok {
for _, rule := range r.Rules {
for _, verb := range rule.Verbs {
if verb != "get" && verb != "list" {
t.Fatal("distributed cluster grant writes", r.Name, verb)
}
}
}
}
}
if !found {
t.Fatal("archive service missing")
}
p.VelocityCIDRs = []string{"10.0.0.0/8"}
if p.Validate() == nil {
t.Fatal("broad Velocity source accepted")
}
}
+24
View File
@@ -2,6 +2,7 @@ package platform
import (
"fmt"
"net"
"k8s.io/apimachinery/pkg/api/resource"
)
@@ -89,6 +90,10 @@ const (
// Params parameterises the install bundle. Namespaces and the registry location
// have safe defaults; VelocityCIDRs has none — see the field comment.
type Params struct {
Distributed bool
ControllerNode string
EgressProbe string
RegistryNodeCIDRs []string
// ControlNamespace is where felis-api/operator run; their SAs live here and the
// RoleBindings' subjects reference them here, even though the Roles they bind to
// live in the minecraft (and build) namespaces. The reaper alone runs — CronJob
@@ -212,6 +217,25 @@ type Params struct {
// Validate reports a Params the renderer cannot turn into objects.
func (p Params) Validate() error {
if p.Distributed && (p.ControllerNode == "" || p.EgressProbe == "" || p.BackupPVC == "" || p.ArchiveLocalPath == "") {
return fmt.Errorf("distributed mode requires controller node, egress probe, backup PVC and archive path")
}
if p.Distributed && len(p.ServerEgressAllowCIDRs) > 0 {
return fmt.Errorf("distributed mode does not permit private game egress exceptions")
}
if p.Distributed {
for _, cidr := range append(append([]string{}, p.VelocityCIDRs...), p.RegistryNodeCIDRs...) {
_, network, err := net.ParseCIDR(cidr)
if err != nil {
return fmt.Errorf("invalid node source %q", cidr)
}
ones, bits := network.Mask.Size()
if ones != bits {
return fmt.Errorf("distributed node sources must be exact /32 or /128 addresses: %q", cidr)
}
}
}
for _, q := range []struct{ name, v string }{
{"registry storage", p.RegistryStorage},
{"uploads storage", p.UploadsStorage},
+3
View File
@@ -266,6 +266,9 @@ func RegistryIngressPolicy(p Params) *networkingv1.NetworkPolicy {
Ports: []networkingv1.NetworkPolicyPort{{Protocol: &tcp, Port: &port}},
}},
)
for _, cidr := range p.RegistryNodeCIDRs {
np.Spec.Ingress[0].From = append(np.Spec.Ingress[0].From, networkingv1.NetworkPolicyPeer{IPBlock: &networkingv1.IPBlock{CIDR: cidr}})
}
return np
}
+1
View File
@@ -206,6 +206,7 @@ func postgresDeployment(p Params) *appsv1.Deployment {
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{Labels: labels},
Spec: corev1.PodSpec{
NodeSelector: controllerSelector(p),
AutomountServiceAccountToken: boolPtr(false),
EnableServiceLinks: boolPtr(false),
PriorityClassName: controlPlanePriorityName,
+13 -5
View File
@@ -19,7 +19,7 @@ var groupFelis = v1alpha1.GroupName // "felis.lolicon.best"
// RBAC is the control-plane authorization bundle: one SA per identity and the
// namespaced Roles + RoleBindings that grant each exactly the verbs its code path
// exercises. There is deliberately no ClusterRole or ClusterRoleBinding anywhere.
// exercises. DistributedRBAC adds read-only node/PV grants when distributed mode is enabled.
type RBAC struct {
ServiceAccounts []*corev1.ServiceAccount
Roles []*rbacv1.Role
@@ -100,7 +100,7 @@ func ControlPlaneRBAC(p Params) RBAC {
// than its logs).
func APIMinecraftRole(p Params) *rbacv1.Role {
p = p.withDefaults()
return role(p.MinecraftNamespace, "felis-api", ComponentAPI, []rbacv1.PolicyRule{
rules := []rbacv1.PolicyRule{
rule([]string{groupFelis}, []string{"minecraftservers"}, []string{"get", "list", "watch", "create", "patch"}),
rule([]string{groupCore}, []string{"secrets"}, []string{"get"}),
// get-only: WorldVolumeExists does a single direct Get of the world PVC;
@@ -116,7 +116,11 @@ func APIMinecraftRole(p Params) *rbacv1.Role {
// the verbs stay tight — list on pods, get on pods/log, and nothing else.
rule([]string{groupCore}, []string{"pods"}, []string{"list"}),
rule([]string{groupCore}, []string{"pods/log"}, []string{"get"}),
})
}
if p.Distributed {
rules = append(rules, rule([]string{groupCore}, []string{"persistentvolumeclaims"}, []string{"create", "list"}), rule([]string{groupApps}, []string{"statefulsets"}, []string{"get", "delete"}))
}
return role(p.MinecraftNamespace, "felis-api", ComponentAPI, rules)
}
// APIBuildRole grants felis-api the build-Job lifecycle in the build namespace
@@ -218,12 +222,16 @@ func OperatorRole(p Params) *rbacv1.Role {
// request and never deletes a CR itself.
func ReaperRole(p Params) *rbacv1.Role {
p = p.withDefaults()
return role(p.MinecraftNamespace, "felis-reaper", ComponentReaper, []rbacv1.PolicyRule{
rules := []rbacv1.PolicyRule{
rule([]string{groupFelis}, []string{"minecraftservers"}, []string{"get", "patch", "delete"}),
rule([]string{groupCore}, []string{"persistentvolumeclaims"}, []string{"get", "delete"}),
rule([]string{groupCore}, []string{"pods"}, []string{"list"}),
rule([]string{groupBatch}, []string{"jobs"}, []string{"list"}),
})
}
if p.Distributed {
rules = append(rules, rule([]string{groupBatch}, []string{"jobs"}, []string{"create", "get"}), rule([]string{groupCore}, []string{"persistentvolumeclaims"}, []string{"list"}))
}
return role(p.MinecraftNamespace, "felis-reaper", ComponentReaper, rules)
}
// controlPlaneServiceAccount renders a control-plane SA. Unlike the weak
+32 -3
View File
@@ -7,6 +7,7 @@ import (
"time"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
@@ -273,11 +274,16 @@ func Workloads(p Params) []Object {
objs = append(objs, backupPVC(p))
}
switch {
case p.Distributed && retentionEnabled(p):
objs = append(objs, reaperCronJob(p))
case reaperEnabled(p):
objs = append(objs, worldsRootPV(p), worldsRootPVC(p), reaperCronJob(p))
case retentionEnabled(p):
objs = append(objs, reaperCronJob(p))
}
if p.Distributed {
objs = append(objs, ArchiveDeployment(p), ArchiveService(p))
}
return objs
}
@@ -315,7 +321,7 @@ const controlPlanePriorityName = "system-cluster-critical"
// together so a partial configuration fails loudly rather than silently dropping
// retention here.
func reaperEnabled(p Params) bool {
return p.WorldsHostPath != "" && retentionEnabled(p)
return (p.WorldsHostPath != "" || p.Distributed) && retentionEnabled(p)
}
// retentionEnabled reports whether the archive store can be looked after: the
@@ -716,6 +722,11 @@ func reaperCronJob(p Params) *batchv1.CronJob {
})
}
if p.Distributed {
container.Args = []string{"--config", configFilePath}
container.Env = append(container.Env, distributedEnv(p)...)
container.Env = append(container.Env, corev1.EnvVar{Name: "FELIS_IMAGE", Value: p.FelisImage})
}
return &batchv1.CronJob{
TypeMeta: metav1.TypeMeta{APIVersion: "batch/v1", Kind: "CronJob"},
// The CronJob lives in the MINECRAFT namespace: a Pod can only mount PVCs
@@ -844,10 +855,12 @@ func reaperPodSpec(p Params, container corev1.Container, volumes []corev1.Volume
Containers: []corev1.Container{container},
Volumes: volumes,
}
if p.ReaperNode != "" {
if p.ControllerNode != "" {
spec.NodeSelector = controllerSelector(p)
} else if p.ReaperNode != "" {
spec.NodeSelector = map[string]string{"kubernetes.io/hostname": p.ReaperNode}
}
if p.WorldsHostPath != "" {
if p.WorldsHostPath != "" || p.Distributed {
spec.ServiceAccountName = SAReaper
} else {
spec.ServiceAccountName = "default"
@@ -868,6 +881,13 @@ func reaperPodSpec(p Params, container corev1.Container, volumes []corev1.Volume
// Recreate guarantees the old pod is gone before the new one starts.
func controlPlaneDeployment(p Params, sa string, container corev1.Container, volumes []corev1.Volume) *appsv1.Deployment {
labels := controlPlanePodLabels(container.Name)
if p.Distributed {
if sa == SAOperator {
container.Env = append(container.Env, corev1.EnvVar{Name: "FELIS_DISTRIBUTED", Value: "true"}, corev1.EnvVar{Name: "FELIS_CONTROLLER_NODE", Value: p.ControllerNode}, corev1.EnvVar{Name: "FELIS_EGRESS_PROBE", Value: p.EgressProbe})
} else {
container.Env = append(container.Env, distributedEnv(p)...)
}
}
return &appsv1.Deployment{
TypeMeta: metav1.TypeMeta{APIVersion: "apps/v1", Kind: "Deployment"},
ObjectMeta: metav1.ObjectMeta{Name: sa, Namespace: p.ControlNamespace, Labels: labels},
@@ -878,6 +898,7 @@ func controlPlaneDeployment(p Params, sa string, container corev1.Container, vol
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{Labels: labels},
Spec: corev1.PodSpec{
NodeSelector: controllerSelector(p),
ServiceAccountName: sa,
PriorityClassName: controlPlanePriorityName,
SecurityContext: hardenedPodSecurityContext(),
@@ -1003,6 +1024,7 @@ func registryDeployment(p Params) *appsv1.Deployment {
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{Labels: labels},
Spec: corev1.PodSpec{
NodeSelector: controllerSelector(p),
AutomountServiceAccountToken: boolPtr(false),
PriorityClassName: controlPlanePriorityName,
SecurityContext: hardenedPodSecurityContext(),
@@ -1373,3 +1395,10 @@ func hardenedContainerSecurityContext() *corev1.SecurityContext {
func boolPtr(b bool) *bool { return &b }
func int32Ptr(i int32) *int32 { return &i }
func int64Ptr(i int64) *int64 { return &i }
func controllerSelector(p Params) map[string]string {
if p.ControllerNode == "" {
return nil
}
return map[string]string{placement.LabelIdentity: p.ControllerNode, placement.LabelRole: placement.RoleController}
}
+15 -2
View File
@@ -41,7 +41,8 @@ type K8sCluster struct {
now func() time.Time
// beat is how often a held lock is rewritten; maintenance.Grace/4 unless a
// test shortens it.
beat time.Duration
beat time.Duration
distributed bool
}
// NewK8sCluster builds a Cluster over c, scoped to namespace.
@@ -49,6 +50,18 @@ func NewK8sCluster(c client.Client, namespace string) *K8sCluster {
return &K8sCluster{c: c, namespace: namespace}
}
func (k *K8sCluster) WithDistributed(enabled bool) *K8sCluster { k.distributed = enabled; return k }
func (k *K8sCluster) RetainedWorlds(ctx context.Context, name string) (bool, error) {
if !k.distributed {
return false, nil
}
var pvcs corev1.PersistentVolumeClaimList
if err := k.c.List(ctx, &pvcs, client.InNamespace(k.namespace), client.MatchingLabels{maintenance.LabelServer: name}); err != nil {
return false, err
}
return len(pvcs.Items) > 0, nil
}
func (k *K8sCluster) clock() time.Time {
if k.now != nil {
return k.now()
@@ -61,7 +74,7 @@ func (k *K8sCluster) Inspect(ctx context.Context, name string) (ServerCRD, error
if err := k.get(ctx, name, &ms); err != nil {
return ServerCRD{}, err
}
return ServerCRD{Exempt: ms.Spec.ReaperExempt, PVC: WorldPVCName(name), UID: string(ms.UID)}, nil
return ServerCRD{Exempt: ms.Spec.ReaperExempt, PVC: ms.WorldPVC(), UID: string(ms.UID)}, nil
}
// HoldWorld implements Cluster. The lock is the same Annotation felis-api
+1 -1
View File
@@ -76,7 +76,7 @@ func (s *PGStore) FreshBackup(ctx context.Context, server string, since time.Tim
func (s *PGStore) InsertBackup(ctx context.Context, rec BackupRecord) error {
const q = `INSERT INTO world_backups
(id, server_name, former_owner, backup_ref, size_bytes, reason, status, created_at, expires_at, sha256, skipped_entries)
VALUES ($1, $2, NULLIF($3, ''), $4, $5, $6, 'present', now(), $7, NULLIF($8, ''), $9)`
VALUES ($1, $2, NULLIF($3, ''), $4, $5, $6, 'present', now(), $7, NULLIF($8, ''), $9) ON CONFLICT (id) DO NOTHING`
_, err := s.db.ExecContext(ctx, q,
rec.ID, rec.ServerName, rec.FormerOwner, rec.BackupRef, rec.SizeBytes, rec.Reason, rec.ExpiresAt,
rec.SHA256, rec.SkippedEntries)
+11
View File
@@ -734,6 +734,17 @@ func (r *Reaper) retire(ctx context.Context, now time.Time, c Candidate, crd Ser
// MinecraftServer the reaper cannot hold it still to archive it: an operator
// takes it from there, and the run reports the server until then.
func (r *Reaper) forgetServer(ctx context.Context, now time.Time, c Candidate, sum *Summary) error {
if cluster, ok := r.Cluster.(interface {
RetainedWorlds(context.Context, string) (bool, error)
}); ok {
retained, err := cluster.RetainedWorlds(ctx, c.Name)
if err != nil {
return err
}
if retained {
return fmt.Errorf("retained world volumes require explicit administrator cleanup")
}
}
pvc := WorldPVCName(c.Name)
exists, err := r.Cluster.WorldExists(ctx, pvc)
if err != nil {
+31 -17
View File
@@ -1,7 +1,9 @@
package restore
import (
"felis.lolicon.best/internal/archivetransfer"
"fmt"
"strconv"
"time"
batchv1 "k8s.io/api/batch/v1"
@@ -34,22 +36,24 @@ const (
// archive ref + Config by the Restorer. jobspec is a pure function of them so
// the security-critical Job shape is unit-tested without a cluster.
type JobParams struct {
Server string
WorldPVC string
BackupPVC string
BackupRef string
ArchiveStore string
Namespace string
ServiceAccount string
Image string
BackupRoot string
WorldsRoot string
Deadline time.Duration
CPULimit string
MemLimit string
RunAsUser int64
RunAsGroup int64
FSGroup int64
SourceURL, Token, SHA256 string
MaxBytes int64
Server string
WorldPVC string
BackupPVC string
BackupRef string
ArchiveStore string
Namespace string
ServiceAccount string
Image string
BackupRoot string
WorldsRoot string
Deadline time.Duration
CPULimit string
MemLimit string
RunAsUser int64
RunAsGroup int64
FSGroup int64
TTLAfterFinished time.Duration
}
@@ -92,7 +96,7 @@ func RestoreJob(p JobParams) (*batchv1.Job, error) {
if p.Image == "" {
return nil, fmt.Errorf("restore: image is empty")
}
if p.WorldPVC == "" || p.BackupPVC == "" {
if p.WorldPVC == "" || (p.BackupPVC == "" && p.SourceURL == "") {
return nil, fmt.Errorf("restore: world and backup PVC names are required")
}
limits, err := resourceLimits(p.CPULimit, p.MemLimit)
@@ -192,6 +196,16 @@ func RestoreJob(p JobParams) (*batchv1.Job, error) {
},
},
}
if p.SourceURL != "" {
if p.Token == "" || p.SHA256 == "" || p.MaxBytes <= 0 {
return nil, fmt.Errorf("restore: remote archive credentials and bounds required")
}
c := &job.Spec.Template.Spec.Containers[0]
c.Args = append(c.Args, "--source-url", p.SourceURL, "--sha256", p.SHA256, "--max-bytes", strconv.FormatInt(p.MaxBytes, 10))
c.Env = []corev1.EnvVar{{Name: archivetransfer.TokenEnv, Value: p.Token}}
c.VolumeMounts[1] = corev1.VolumeMount{Name: "tmp", MountPath: "/tmp"}
job.Spec.Template.Spec.Volumes[1] = corev1.Volume{Name: "tmp", VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}}
}
return job, nil
}
+11 -1
View File
@@ -32,6 +32,7 @@ import (
"time"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
)
// ErrAlreadyExists is returned by a Jobs implementation when a restore Job for a
@@ -75,6 +76,7 @@ type Jobs interface {
// either is empty the caller leaves the API's Restorer nil so the endpoint
// reports 503 rather than enqueuing a Job that cannot run.
type Config struct {
ResolveWorld placement.Resolver
// Namespace is where the world PVCs live and the restore Job runs (the
// minecraft namespace). The Job is intentionally co-located with the world it
// restores; it never runs in the felis control-plane namespace.
@@ -199,7 +201,15 @@ type Restorer struct {
// keeps the handler's 202 honest in both directions — not a 500 for a genuine
// duplicate, and not a false "restoring" for a retry after a failure.
func (r *Restorer) Restore(ctx context.Context, serverName, backupRef string) error {
if err := r.Jobs.CreateRestoreJob(ctx, r.jobParams(serverName, backupRef)); err != nil {
p := r.jobParams(serverName, backupRef)
if r.Config.ResolveWorld != nil {
w, err := r.Config.ResolveWorld(ctx, serverName)
if err != nil {
return err
}
p.WorldPVC = w.Claim
}
if err := r.Jobs.CreateRestoreJob(ctx, p); err != nil {
if errors.Is(err, ErrAlreadyExists) {
return nil // already enqueued — idempotent
}
+11 -1
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"time"
"felis.lolicon.best/internal/placement"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
@@ -57,7 +58,8 @@ const (
// JobParams are the rendered inputs to an export Job. ExportJob is a pure
// function of them, so the Job shape is unit-tested without a cluster.
type JobParams struct {
Server string
NodeName string
Server string
// ID names this export: it is the tail of the Job name and of the internal
// upload path, so two exports of one server never collide.
ID string
@@ -227,6 +229,7 @@ func ExportJob(p JobParams) (*batchv1.Job, error) {
ObjectMeta: metav1.ObjectMeta{Labels: exportLabels(p)},
Spec: corev1.PodSpec{
RestartPolicy: corev1.RestartPolicyNever,
NodeSelector: jobNodeSelector(p.NodeName),
ServiceAccountName: p.ServiceAccount,
AutomountServiceAccountToken: boolPtr(false),
SecurityContext: sc,
@@ -269,3 +272,10 @@ func resourceLimits(cpu, mem string) (corev1.ResourceList, error) {
func boolPtr(b bool) *bool { return &b }
func int32Ptr(i int32) *int32 { return &i }
func int64Ptr(i int64) *int64 { return &i }
func jobNodeSelector(name string) map[string]string {
if name == "" {
return nil
}
return map[string]string{placement.LabelIdentity: name}
}
+13 -1
View File
@@ -20,6 +20,7 @@ import (
"time"
"felis.lolicon.best/internal/naming"
"felis.lolicon.best/internal/placement"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
@@ -47,6 +48,7 @@ type Request struct {
// cmd/felis leaves the API's Exporter nil when either is missing, and the
// routes answer 503.
type Config struct {
ResolveWorld placement.Resolver
Namespace string
ServiceAccount string
Image string
@@ -112,9 +114,19 @@ func New(cs kubernetes.Interface, cfg Config) *Exporter {
// Start creates the export Job for r and returns its name.
func (e *Exporter) Start(ctx context.Context, r Request) (string, error) {
c := e.cfg
claim := naming.WorldPVCName(r.Server)
node := ""
if c.ResolveWorld != nil && r.Mode != ModeBackup {
world, err := c.ResolveWorld(ctx, r.Server)
if err != nil {
return "", err
}
claim, node = world.Claim, world.Node
}
job, err := ExportJob(JobParams{
Server: r.Server, ID: r.ID, Mode: r.Mode,
WorldPVC: naming.WorldPVCName(r.Server), BackupPVC: c.BackupPVC, BackupRef: r.BackupRef,
NodeName: node,
WorldPVC: claim, BackupPVC: c.BackupPVC, BackupRef: r.BackupRef,
BackupSHA256: r.BackupSHA256, Path: r.Path, Dir: r.Dir,
TargetURL: r.TargetURL, Token: r.Token,
Namespace: c.Namespace, ServiceAccount: c.ServiceAccount, Image: c.Image,