feat: storage writes via storage-agent, accepts_uploads and write_template (M5.5)
ci / api / api (push) Successful in 33s
ci / contract / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 43s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m11s
ci / ts / ts (packages + web) (push) Successful in 1m10s
ci / docker / docker (образы собираются — их сервер собирает сам при обновлении) (push) Successful in 51s
ci / api / api (push) Successful in 33s
ci / contract / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 43s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m11s
ci / ts / ts (packages + web) (push) Successful in 1m10s
ci / docker / docker (образы собираются — их сервер собирает сам при обновлении) (push) Successful in 51s
- storage-agent: POST /internal/files copies a staged file into a source
(temp file, fsync, SHA-256 check, hard link so nothing is overwritten,
" (2)" suffix on conflict, owner PIPU_FILE_UID/GID) and
POST /internal/files/test. NFS sources that accept uploads are mounted
rw in an agent-only /library-rw and bind-mounted read-only for others;
local sources are written through /library-host-rw.
- meta: write templates ({track:02}, {disc:1-}, [optional]) with name
sanitising and length limits, suggestion from the source's regexes
and a reversibility check against 50 catalogue tracks.
- API and admin UI: accepts_uploads / write_template on sources,
write-template suggest and check endpoints, write test result.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
9b50038bb4
commit
f66be2ba0a
27 files changed
+2422
-213
No files matched your search
@@ -0,0 +1,289 @@
|
||||
package storageagent
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Запись в хранилище (docs/library-and-storage.md#запись-в-хранилище): только импорт одобренных загрузок,
|
||||
// только создание новых файлов. Перезаписи и удаления файлов библиотеки в коде агента нет.
|
||||
|
||||
// Paths — каталоги агента.
|
||||
type Paths struct {
|
||||
Library string // ro-точки NFS: <Library>/<source_id> (видны api, nginx, analyzer)
|
||||
LibraryRW string // rw-точки NFS-источников с accepts_uploads; видны только агенту. Пусто — запись в NFS выключена
|
||||
HostRW string // LIBRARY_HOST на запись (local-источники); пусто — запись в local выключена
|
||||
Uploads string // staging очереди на добавление (только чтение)
|
||||
FileUID int // владелец новых файлов и каталогов; < 0 — не менять
|
||||
FileGID int
|
||||
}
|
||||
|
||||
// WriteRequest — тело POST /internal/files.
|
||||
type WriteRequest struct {
|
||||
SourceID string `json:"source_id"`
|
||||
LocalPath *string `json:"local_path,omitempty"` // для local-источника: папка внутри LIBRARY_HOST
|
||||
RelPath string `json:"rel_path"` // желаемый путь в хранилище
|
||||
StagingPath string `json:"staging_path"` // путь файла относительно каталога staging
|
||||
SHA256 string `json:"sha256"` // ожидаемый хэш (hex), копия сверяется с ним
|
||||
}
|
||||
|
||||
// WriteResult — где файл оказался (при конфликте имени — с суффиксом « (2)»).
|
||||
type WriteResult struct {
|
||||
RelPath string `json:"rel_path"`
|
||||
Size int64 `json:"size"`
|
||||
Renamed bool `json:"renamed"`
|
||||
}
|
||||
|
||||
// WriteTestRequest — тело POST /internal/files/test.
|
||||
type WriteTestRequest struct {
|
||||
SourceID string `json:"source_id"`
|
||||
LocalPath *string `json:"local_path,omitempty"`
|
||||
}
|
||||
|
||||
// WriteTestResult — можно ли писать в хранилище.
|
||||
type WriteTestResult struct {
|
||||
OK bool `json:"ok"`
|
||||
Message string `json:"message,omitempty"`
|
||||
}
|
||||
|
||||
const maxConflicts = 100
|
||||
|
||||
// errBadPath — путь вне корня, абсолютный или с «..».
|
||||
var errBadPath = errors.New("path must be relative, without .. and hidden segments")
|
||||
|
||||
// cleanRel проверяет относительный путь с `/` и возвращает его в каноническом виде.
|
||||
func cleanRel(rel string) (string, error) {
|
||||
if rel == "" || strings.HasPrefix(rel, "/") || strings.ContainsAny(rel, "\\\x00") {
|
||||
return "", errBadPath
|
||||
}
|
||||
for _, seg := range strings.Split(rel, "/") {
|
||||
if seg == "" || seg == "." || seg == ".." || strings.HasPrefix(seg, ".") {
|
||||
return "", errBadPath
|
||||
}
|
||||
}
|
||||
return path.Clean(rel), nil
|
||||
}
|
||||
|
||||
// writeRoot — каталог источника на запись.
|
||||
func (a *agent) writeRoot(sourceID string, localPath *string) (string, error) {
|
||||
if localPath != nil {
|
||||
if a.paths.HostRW == "" {
|
||||
return "", errors.New("writing to local sources is disabled")
|
||||
}
|
||||
root := a.paths.HostRW
|
||||
if *localPath != "" {
|
||||
rel, err := cleanRel(strings.Trim(*localPath, "/"))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
root = filepath.Join(root, filepath.FromSlash(rel))
|
||||
}
|
||||
if info, err := os.Stat(root); err != nil || !info.IsDir() {
|
||||
return "", errors.New("source folder not found")
|
||||
}
|
||||
return root, nil
|
||||
}
|
||||
if a.paths.LibraryRW == "" {
|
||||
return "", errors.New("writing to NFS sources is disabled")
|
||||
}
|
||||
root, err := mountPoint(a.paths.LibraryRW, sourceID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, err := os.Stat(root); err != nil {
|
||||
return "", errors.New("source is not mounted for writing (accepts_uploads is off?)")
|
||||
}
|
||||
return root, nil
|
||||
}
|
||||
|
||||
func (a *agent) writeFile(w http.ResponseWriter, r *http.Request) {
|
||||
var req WriteRequest
|
||||
if !a.decode(w, r, &req) {
|
||||
return
|
||||
}
|
||||
root, err := a.writeRoot(req.SourceID, req.LocalPath)
|
||||
if err != nil {
|
||||
writeValidation(w, &ValidationError{Field: "source_id", Reason: err.Error()})
|
||||
return
|
||||
}
|
||||
rel, err := cleanRel(req.RelPath)
|
||||
if err != nil {
|
||||
writeValidation(w, &ValidationError{Field: "rel_path", Reason: err.Error()})
|
||||
return
|
||||
}
|
||||
staging, err := cleanRel(req.StagingPath)
|
||||
if err != nil || a.paths.Uploads == "" {
|
||||
writeValidation(w, &ValidationError{Field: "staging_path", Reason: "invalid staging path"})
|
||||
return
|
||||
}
|
||||
want, err := hex.DecodeString(req.SHA256)
|
||||
if err != nil || len(want) != sha256.Size {
|
||||
writeValidation(w, &ValidationError{Field: "sha256", Reason: "must be 64 hex characters"})
|
||||
return
|
||||
}
|
||||
res, err := a.write(root, rel, filepath.Join(a.paths.Uploads, filepath.FromSlash(staging)), want)
|
||||
if err != nil {
|
||||
a.log.ErrorContext(r.Context(), "write file", "source_id", req.SourceID, "rel_path", rel, "err", err)
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]any{
|
||||
"status": http.StatusInternalServerError, "code": "storage.write_failed", "detail": err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
a.log.InfoContext(r.Context(), "file written", "source_id", req.SourceID, "rel_path", res.RelPath, "size", res.Size)
|
||||
writeJSON(w, http.StatusOK, res)
|
||||
}
|
||||
|
||||
// write копирует staging во временный файл рядом с целью, fsync, сверяет SHA-256 и создаёт итоговое имя
|
||||
// жёсткой ссылкой (не перезаписывает существующий файл, в отличие от rename), затем убирает временный.
|
||||
func (a *agent) write(root, rel, staging string, want []byte) (WriteResult, error) {
|
||||
target := filepath.Join(root, filepath.FromSlash(rel))
|
||||
dir := filepath.Dir(target)
|
||||
if err := a.mkdirs(root, dir); err != nil {
|
||||
return WriteResult{}, err
|
||||
}
|
||||
tmp := filepath.Join(dir, "."+filepath.Base(target)+".pipu-tmp")
|
||||
size, err := a.copyTemp(staging, tmp)
|
||||
if err != nil {
|
||||
_ = os.Remove(tmp)
|
||||
return WriteResult{}, err
|
||||
}
|
||||
defer func() { _ = os.Remove(tmp) }()
|
||||
got, err := fileSHA256(tmp)
|
||||
if err != nil {
|
||||
return WriteResult{}, err
|
||||
}
|
||||
if !bytes.Equal(got, want) {
|
||||
return WriteResult{}, fmt.Errorf("checksum mismatch: copy %x, expected %x", got, want)
|
||||
}
|
||||
|
||||
ext := path.Ext(rel)
|
||||
base := strings.TrimSuffix(rel, ext)
|
||||
for n := 1; n <= maxConflicts; n++ {
|
||||
name := rel
|
||||
if n > 1 {
|
||||
name = fmt.Sprintf("%s (%d)%s", base, n, ext)
|
||||
}
|
||||
err := os.Link(tmp, filepath.Join(root, filepath.FromSlash(name)))
|
||||
if err == nil {
|
||||
syncDir(dir)
|
||||
return WriteResult{RelPath: name, Size: size, Renamed: n > 1}, nil
|
||||
}
|
||||
if !os.IsExist(err) {
|
||||
return WriteResult{}, fmt.Errorf("link: %w", err)
|
||||
}
|
||||
}
|
||||
return WriteResult{}, errors.New("too many files with the same name")
|
||||
}
|
||||
|
||||
// mkdirs создаёт недостающие каталоги от root до dir (0755, владелец — FileUID/FileGID).
|
||||
func (a *agent) mkdirs(root, dir string) error {
|
||||
rel, err := filepath.Rel(root, dir)
|
||||
if err != nil || strings.HasPrefix(rel, "..") {
|
||||
return errBadPath
|
||||
}
|
||||
cur := root
|
||||
for _, seg := range strings.Split(rel, string(filepath.Separator)) {
|
||||
if seg == "." || seg == "" {
|
||||
continue
|
||||
}
|
||||
cur = filepath.Join(cur, seg)
|
||||
if err := os.Mkdir(cur, 0o755); err != nil {
|
||||
if os.IsExist(err) {
|
||||
continue
|
||||
}
|
||||
return fmt.Errorf("mkdir: %w", err)
|
||||
}
|
||||
a.chown(cur)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *agent) copyTemp(staging, tmp string) (int64, error) {
|
||||
src, err := os.Open(staging)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("staging: %w", err)
|
||||
}
|
||||
defer func() { _ = src.Close() }()
|
||||
_ = os.Remove(tmp) // остаток прерванной попытки
|
||||
dst, err := os.OpenFile(tmp, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("create temp: %w", err)
|
||||
}
|
||||
n, err := io.Copy(dst, src)
|
||||
if err == nil {
|
||||
err = dst.Sync()
|
||||
}
|
||||
if cerr := dst.Close(); err == nil {
|
||||
err = cerr
|
||||
}
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("copy: %w", err)
|
||||
}
|
||||
a.chown(tmp)
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (a *agent) chown(p string) {
|
||||
if a.paths.FileUID >= 0 {
|
||||
if err := os.Lchown(p, a.paths.FileUID, a.paths.FileGID); err != nil {
|
||||
a.log.Warn("chown failed", "path", p, "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// testWrite создаёт и удаляет .pipu-write-test в корне источника на запись.
|
||||
func (a *agent) testWrite(w http.ResponseWriter, r *http.Request) {
|
||||
var req WriteTestRequest
|
||||
if !a.decode(w, r, &req) {
|
||||
return
|
||||
}
|
||||
root, err := a.writeRoot(req.SourceID, req.LocalPath)
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusOK, WriteTestResult{Message: err.Error()})
|
||||
return
|
||||
}
|
||||
p := filepath.Join(root, ".pipu-write-test")
|
||||
f, err := os.OpenFile(p, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
|
||||
if err == nil {
|
||||
_, err = f.WriteString("pipu")
|
||||
if cerr := f.Close(); err == nil {
|
||||
err = cerr
|
||||
}
|
||||
}
|
||||
_ = os.Remove(p)
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusOK, WriteTestResult{Message: "cannot write: " + err.Error()})
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, WriteTestResult{OK: true, Message: "write access ok"})
|
||||
}
|
||||
|
||||
func fileSHA256(p string) ([]byte, error) {
|
||||
f, err := os.Open(p)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = f.Close() }()
|
||||
h := sha256.New()
|
||||
if _, err := io.Copy(h, f); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return h.Sum(nil), nil
|
||||
}
|
||||
|
||||
// syncDir — fsync каталога, чтобы новое имя пережило сбой питания (на Windows не поддерживается — молча).
|
||||
func syncDir(dir string) {
|
||||
if d, err := os.Open(dir); err == nil {
|
||||
_ = d.Sync()
|
||||
_ = d.Close()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package storageagent
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const srcID = "0191d3a4-7c1e-7a51-9a3b-2f6f5c3e9b10"
|
||||
|
||||
type writeEnv struct {
|
||||
h http.Handler
|
||||
paths Paths
|
||||
mounter *fakeMounter
|
||||
}
|
||||
|
||||
func newWriteEnv(t *testing.T) writeEnv {
|
||||
base := t.TempDir()
|
||||
p := Paths{
|
||||
Library: filepath.Join(base, "library"), LibraryRW: filepath.Join(base, "library-rw"),
|
||||
HostRW: filepath.Join(base, "host"), Uploads: filepath.Join(base, "uploads"), FileUID: -1, FileGID: -1,
|
||||
}
|
||||
for _, d := range []string{p.Library, p.LibraryRW, p.HostRW, filepath.Join(p.LibraryRW, srcID), filepath.Join(p.Uploads, "u1"), filepath.Join(p.HostRW, "music")} {
|
||||
require.NoError(t, os.MkdirAll(d, 0o755))
|
||||
}
|
||||
m := &fakeMounter{mounted: map[string]NFSSpec{}}
|
||||
return writeEnv{h: NewRouterWithPaths(nil, "secret", p, m), paths: p, mounter: m}
|
||||
}
|
||||
|
||||
func (e writeEnv) post(path string, body any) *httptest.ResponseRecorder {
|
||||
raw, _ := json.Marshal(body)
|
||||
req := httptest.NewRequest(http.MethodPost, path, bytes.NewReader(raw))
|
||||
req.Header.Set("Authorization", "Bearer secret")
|
||||
rec := httptest.NewRecorder()
|
||||
e.h.ServeHTTP(rec, req)
|
||||
return rec
|
||||
}
|
||||
|
||||
func (e writeEnv) staging(t *testing.T, name, content string) string {
|
||||
require.NoError(t, os.WriteFile(filepath.Join(e.paths.Uploads, "u1", name), []byte(content), 0o644))
|
||||
sum := sha256.Sum256([]byte(content))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func TestWriteFile(t *testing.T) {
|
||||
e := newWriteEnv(t)
|
||||
sum := e.staging(t, "f1", "audio-bytes")
|
||||
req := WriteRequest{SourceID: srcID, RelPath: "Кино/1988 - Группа крови/01 - Группа крови.flac", StagingPath: "u1/f1", SHA256: sum}
|
||||
|
||||
rec := e.post("/internal/files", req)
|
||||
require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
|
||||
var res WriteResult
|
||||
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &res))
|
||||
assert.Equal(t, WriteResult{RelPath: req.RelPath, Size: 11}, res)
|
||||
dir := filepath.Join(e.paths.LibraryRW, srcID, "Кино", "1988 - Группа крови")
|
||||
got, err := os.ReadFile(filepath.Join(dir, "01 - Группа крови.flac"))
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "audio-bytes", string(got))
|
||||
|
||||
// Существующий файл не перезаписывается — новое имя с суффиксом; временных файлов не остаётся.
|
||||
rec = e.post("/internal/files", req)
|
||||
require.Equal(t, http.StatusOK, rec.Code)
|
||||
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &res))
|
||||
assert.Equal(t, "Кино/1988 - Группа крови/01 - Группа крови (2).flac", res.RelPath)
|
||||
assert.True(t, res.Renamed)
|
||||
entries, _ := os.ReadDir(dir)
|
||||
assert.Len(t, entries, 2)
|
||||
|
||||
// Local-источник пишется в LIBRARY_HOST.
|
||||
local := "music"
|
||||
rec = e.post("/internal/files", WriteRequest{SourceID: srcID, LocalPath: &local, RelPath: "A/t.mp3", StagingPath: "u1/f1", SHA256: sum})
|
||||
require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
|
||||
assert.FileExists(t, filepath.Join(e.paths.HostRW, "music", "A", "t.mp3"))
|
||||
}
|
||||
|
||||
func TestWriteFileRejects(t *testing.T) {
|
||||
e := newWriteEnv(t)
|
||||
sum := e.staging(t, "f1", "audio-bytes")
|
||||
cases := map[string]WriteRequest{
|
||||
"traversal": {SourceID: srcID, RelPath: "../escape.flac", StagingPath: "u1/f1", SHA256: sum},
|
||||
"absolute": {SourceID: srcID, RelPath: "/etc/x.flac", StagingPath: "u1/f1", SHA256: sum},
|
||||
"hidden": {SourceID: srcID, RelPath: "A/.x.flac", StagingPath: "u1/f1", SHA256: sum},
|
||||
"staging escape": {SourceID: srcID, RelPath: "A/x.flac", StagingPath: "../library/x", SHA256: sum},
|
||||
"not writable": {SourceID: "0191d3a4-7c1e-7a51-9a3b-2f6f5c3e9b11", RelPath: "A/x.flac", StagingPath: "u1/f1", SHA256: sum},
|
||||
"bad sha256": {SourceID: srcID, RelPath: "A/x.flac", StagingPath: "u1/f1", SHA256: "zz"},
|
||||
"bad local path": {SourceID: srcID, LocalPath: ptr("../x"), RelPath: "A/x.flac", StagingPath: "u1/f1", SHA256: sum},
|
||||
}
|
||||
for name, req := range cases {
|
||||
rec := e.post("/internal/files", req)
|
||||
assert.Equal(t, http.StatusUnprocessableEntity, rec.Code, name)
|
||||
}
|
||||
|
||||
// Копия не совпала с ожидаемым хэшем — ошибка, ни итогового, ни временного файла.
|
||||
other := sha256.Sum256([]byte("other"))
|
||||
rec := e.post("/internal/files", WriteRequest{SourceID: srcID, RelPath: "A/x.flac", StagingPath: "u1/f1", SHA256: hex.EncodeToString(other[:])})
|
||||
assert.Equal(t, http.StatusInternalServerError, rec.Code)
|
||||
entries, _ := os.ReadDir(filepath.Join(e.paths.LibraryRW, srcID, "A"))
|
||||
assert.Empty(t, entries)
|
||||
}
|
||||
|
||||
func TestWritableMount(t *testing.T) {
|
||||
e := newWriteEnv(t)
|
||||
rec := e.post("/internal/mounts", MountRequest{SourceID: srcID, NFSSpec: NFSSpec{Server: "nas", Export: "/music"}, Writable: true})
|
||||
require.Equal(t, http.StatusOK, rec.Code)
|
||||
rw := filepath.Join(e.paths.LibraryRW, srcID)
|
||||
ro := filepath.Join(e.paths.Library, srcID)
|
||||
assert.Contains(t, e.mounter.mounted[rw].Options, "rw")
|
||||
assert.Equal(t, "bind:"+rw, e.mounter.mounted[ro].Export)
|
||||
|
||||
rec = e.post("/internal/files/test", WriteTestRequest{SourceID: srcID})
|
||||
var res WriteTestResult
|
||||
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &res))
|
||||
assert.True(t, res.OK, res.Message)
|
||||
assert.NoFileExists(t, filepath.Join(rw, ".pipu-write-test"))
|
||||
|
||||
// Без accepts_uploads rw-точка снимается.
|
||||
e.post("/internal/mounts", MountRequest{SourceID: srcID, NFSSpec: NFSSpec{Server: "nas", Export: "/music"}})
|
||||
assert.NotContains(t, e.mounter.mounted, rw)
|
||||
assert.Contains(t, e.mounter.mounted[ro].Options, "ro")
|
||||
}
|
||||
|
||||
func ptr(s string) *string { return &s }
|
||||
@@ -101,6 +101,8 @@ func (s NFSSpec) Normalize(writable bool) (NFSSpec, error) {
|
||||
type Mounter interface {
|
||||
Mount(ctx context.Context, spec NFSSpec, target string) error
|
||||
Unmount(ctx context.Context, target string) error
|
||||
// BindReadOnly показывает каталог src в target только на чтение.
|
||||
BindReadOnly(ctx context.Context, src, target string) error
|
||||
}
|
||||
|
||||
// ExecMounter вызывает системные mount / umount.
|
||||
@@ -119,6 +121,21 @@ func (ExecMounter) Mount(ctx context.Context, spec NFSSpec, target string) error
|
||||
return nil
|
||||
}
|
||||
|
||||
// BindReadOnly реализует Mounter: bind, затем remount ro (флаг ro у bind ставится только перемонтированием).
|
||||
func (ExecMounter) BindReadOnly(ctx context.Context, src, target string) error {
|
||||
if err := os.MkdirAll(target, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
if out, err := exec.CommandContext(ctx, "mount", "--bind", src, target).CombinedOutput(); err != nil {
|
||||
return fmt.Errorf("mount --bind: %s", strings.TrimSpace(string(out)))
|
||||
}
|
||||
if out, err := exec.CommandContext(ctx, "mount", "-o", "remount,bind,ro", target).CombinedOutput(); err != nil {
|
||||
_, _ = exec.CommandContext(ctx, "umount", target).CombinedOutput()
|
||||
return fmt.Errorf("remount ro: %s", strings.TrimSpace(string(out)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Unmount реализует Mounter. Ленивое отключение: зависший NFS-сервер не блокирует агента.
|
||||
func (ExecMounter) Unmount(ctx context.Context, target string) error {
|
||||
out, err := exec.CommandContext(ctx, "umount", "-l", target).CombinedOutput()
|
||||
|
||||
@@ -68,6 +68,11 @@ func (f *fakeMounter) Mount(_ context.Context, spec NFSSpec, target string) erro
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeMounter) BindReadOnly(_ context.Context, src, target string) error {
|
||||
f.mounted[target] = NFSSpec{Export: "bind:" + src}
|
||||
return os.MkdirAll(target, 0o755)
|
||||
}
|
||||
|
||||
func (f *fakeMounter) Unmount(_ context.Context, target string) error {
|
||||
delete(f.mounted, target)
|
||||
return nil
|
||||
|
||||
@@ -33,6 +33,12 @@ func NewRouter(log *slog.Logger, token, libraryRoot string) http.Handler {
|
||||
|
||||
// NewRouterWithMounter — NewRouter с заданным способом монтирования (для тестов).
|
||||
func NewRouterWithMounter(log *slog.Logger, token, libraryRoot string, mounter Mounter) http.Handler {
|
||||
return NewRouterWithPaths(log, token, Paths{Library: libraryRoot, FileUID: -1, FileGID: -1}, mounter)
|
||||
}
|
||||
|
||||
// NewRouterWithPaths — роутер с каталогами записи (импорт загрузок).
|
||||
func NewRouterWithPaths(log *slog.Logger, token string, paths Paths, mounter Mounter) http.Handler {
|
||||
libraryRoot := paths.Library
|
||||
if log == nil {
|
||||
log = slog.New(slog.DiscardHandler)
|
||||
}
|
||||
@@ -52,10 +58,12 @@ func NewRouterWithMounter(log *slog.Logger, token, libraryRoot string, mounter M
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"mounts": mounts})
|
||||
})
|
||||
a := &agent{log: log, root: libraryRoot, mounter: mounter}
|
||||
a := &agent{log: log, root: libraryRoot, paths: paths, mounter: mounter}
|
||||
r.Post("/mounts", a.mount)
|
||||
r.Post("/mounts/test", a.test)
|
||||
r.Delete("/mounts/{sourceID}", a.unmount)
|
||||
r.Post("/files", a.writeFile)
|
||||
r.Post("/files/test", a.testWrite)
|
||||
})
|
||||
return r
|
||||
}
|
||||
@@ -63,6 +71,7 @@ func NewRouterWithMounter(log *slog.Logger, token, libraryRoot string, mounter M
|
||||
type agent struct {
|
||||
log *slog.Logger
|
||||
root string
|
||||
paths Paths
|
||||
mounter Mounter
|
||||
}
|
||||
|
||||
@@ -119,7 +128,40 @@ func (a *agent) mount(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), probeTimeout)
|
||||
defer cancel()
|
||||
_ = a.mounter.Unmount(ctx, target)
|
||||
writeJSON(w, http.StatusOK, a.mountAndCount(ctx, spec, target))
|
||||
rwTarget := a.rwPoint(req.SourceID)
|
||||
if rwTarget != "" {
|
||||
_ = a.mounter.Unmount(ctx, rwTarget)
|
||||
_ = os.Remove(rwTarget)
|
||||
}
|
||||
if !req.Writable {
|
||||
writeJSON(w, http.StatusOK, a.mountAndCount(ctx, spec, target))
|
||||
return
|
||||
}
|
||||
// Источник принимает загрузки: rw-точка видна только агенту, всем остальным — bind-копия только на чтение.
|
||||
if rwTarget == "" {
|
||||
writeJSON(w, http.StatusOK, MountResult{Status: "error", Message: "writable mounts are disabled (LIBRARY_RW_ROOT)"})
|
||||
return
|
||||
}
|
||||
res := a.mountAndCount(ctx, spec, rwTarget)
|
||||
if res.Status == "mounted" {
|
||||
if err := a.mounter.BindReadOnly(ctx, rwTarget, target); err != nil {
|
||||
_ = a.mounter.Unmount(ctx, rwTarget)
|
||||
res = MountResult{Status: "error", Message: err.Error()}
|
||||
}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, res)
|
||||
}
|
||||
|
||||
// rwPoint — rw-точка источника ("" — запись в NFS выключена или id неверный).
|
||||
func (a *agent) rwPoint(sourceID string) string {
|
||||
if a.paths.LibraryRW == "" {
|
||||
return ""
|
||||
}
|
||||
p, err := mountPoint(a.paths.LibraryRW, sourceID)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// test монтирует во временную точку, считает файлы и отключает.
|
||||
@@ -167,6 +209,10 @@ func (a *agent) unmount(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
_ = os.Remove(target)
|
||||
if rw := a.rwPoint(chi.URLParam(r, "sourceID")); rw != "" {
|
||||
_ = a.mounter.Unmount(r.Context(), rw)
|
||||
_ = os.Remove(rw)
|
||||
}
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user