tao.chen
dab4cd062e
对应代码审查发现的问题（#1, #2, #4-#8, #10）： #1 CRITICAL：spark_submit 在 argv 中重复拼接 binary。此前 cmd := []string{binary} 后又把 cmd 作为 Args 传给 executor.Run， 而 executor 会再拼一次 Binary，导致 OS argv 为 [binary, binary, ...]， spark-submit 会把自身当作应用 jar。现在 cmd 从 --master 开始， executor.Run 使用 Binary + Args，argv 正确。 #2：Validate 曾接受 .meta.json 路径本身。现在显式拒绝 sidecar 路径， 要求传入数据文件路径。 #4：Sweep 对 sidecar 损坏的数据文件跳过清理。现在损坏 sidecar 会回退 到数据文件 mtime，超期即删除。 #5：upload_file 描述仍引用已移除的 args 字段，已改为引用 script_path 及结构化字段。 #6：Deps.UploadStore 改为值类型 uploads.Store，避免 nil 绕过上传校验； 移除 spark_submit/upload_file 中的 nil 检查。 #7：master/queue/executor_memory 增加空字符串校验。 #8：提取 buildSparkSubmitCommand 构建 argv，消除双写参数的结构性根因。 #10：Validate 失败时记录 slog.Warn("spark_submit.unminted_path_rejected")。 新增测试： - TestSparkSubmit_StructuredCommand：断言 argv 首行为 --master，末行 仍为 script_path。 - TestSparkSubmit_EmptyMaster：空 master 返回错误。 - TestStore_Validate_RejectsSidecarPath：拒绝 .meta.json 路径。 - TestStore_Sweep_DeletesDataWithCorruptSidecar：损坏 sidecar 的数据文件 被清理。 未在本提交处理： - #9 cluster.DefaultSubmitArgs 弃用留作后续批次。 Co-Authored-By: tao.chen <93983997+taochen-ct@users.noreply.github.com>
317 lines
8.3 KiB
Go
317 lines
8.3 KiB
Go
package uploads
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestStore_SaveAndRetrieve(t *testing.T) {
|
|
store, err := New(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
|
|
data := []byte("hello uploads")
|
|
sum := sha256.Sum256(data)
|
|
wantSHA := hex.EncodeToString(sum[:])
|
|
|
|
fileID, name, size, sha256Hex, absPath, err := store.Save(data, "hello.txt")
|
|
if err != nil {
|
|
t.Fatalf("save: %v", err)
|
|
}
|
|
|
|
if fileID == "" || !regexp.MustCompile(`^[0-9a-f]{32}$`).MatchString(fileID) {
|
|
t.Errorf("fileID=%q, want 32 lowercase hex chars", fileID)
|
|
}
|
|
if name != "hello.txt" {
|
|
t.Errorf("name=%q, want hello.txt", name)
|
|
}
|
|
if size != int64(len(data)) {
|
|
t.Errorf("size=%d, want %d", size, len(data))
|
|
}
|
|
if sha256Hex != wantSHA {
|
|
t.Errorf("sha256=%q, want %q", sha256Hex, wantSHA)
|
|
}
|
|
if !filepath.IsAbs(absPath) {
|
|
t.Errorf("absPath=%q is not absolute", absPath)
|
|
}
|
|
if filepath.Base(absPath) != fileID {
|
|
t.Errorf("absPath base=%q, want fileID %q", filepath.Base(absPath), fileID)
|
|
}
|
|
|
|
gotData, err := os.ReadFile(absPath)
|
|
if err != nil {
|
|
t.Fatalf("read data file: %v", err)
|
|
}
|
|
if string(gotData) != string(data) {
|
|
t.Errorf("data=%q, want %q", gotData, data)
|
|
}
|
|
|
|
info, err := os.Stat(absPath)
|
|
if err != nil {
|
|
t.Fatalf("stat data file: %v", err)
|
|
}
|
|
if info.Mode().Perm() != 0o640 {
|
|
t.Errorf("data mode=%o, want %o", info.Mode().Perm(), 0o640)
|
|
}
|
|
|
|
metaPath := absPath + ".meta.json"
|
|
metaBytes, err := os.ReadFile(metaPath)
|
|
if err != nil {
|
|
t.Fatalf("read sidecar: %v", err)
|
|
}
|
|
var sc sidecar
|
|
if err := json.Unmarshal(metaBytes, &sc); err != nil {
|
|
t.Fatalf("unmarshal sidecar: %v", err)
|
|
}
|
|
if sc.Name != "hello.txt" {
|
|
t.Errorf("sidecar name=%q, want hello.txt", sc.Name)
|
|
}
|
|
if sc.Size != int64(len(data)) {
|
|
t.Errorf("sidecar size=%d, want %d", sc.Size, len(data))
|
|
}
|
|
if sc.Sha256 != wantSHA {
|
|
t.Errorf("sidecar sha256=%q, want %q", sc.Sha256, wantSHA)
|
|
}
|
|
if sc.UploadedAt.IsZero() {
|
|
t.Errorf("sidecar uploaded_at is zero")
|
|
}
|
|
|
|
metaInfo, err := os.Stat(metaPath)
|
|
if err != nil {
|
|
t.Fatalf("stat sidecar: %v", err)
|
|
}
|
|
if metaInfo.Mode().Perm() != 0o600 {
|
|
t.Errorf("sidecar mode=%o, want %o", metaInfo.Mode().Perm(), 0o600)
|
|
}
|
|
|
|
validatedID, err := store.Validate(absPath)
|
|
if err != nil {
|
|
t.Fatalf("validate: %v", err)
|
|
}
|
|
if validatedID != fileID {
|
|
t.Errorf("validatedID=%q, want %q", validatedID, fileID)
|
|
}
|
|
}
|
|
|
|
func TestStore_AbsPath_RejectsMalformed(t *testing.T) {
|
|
store, err := New(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
|
|
validID := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
if _, err := store.AbsPath(validID); err != nil {
|
|
t.Errorf("valid fileID rejected: %v", err)
|
|
}
|
|
|
|
cases := []string{
|
|
"",
|
|
"abc",
|
|
"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
|
|
"0123456789abcdef0123456789abcdefg",
|
|
"0123456789abcdef0123456789abcde ",
|
|
}
|
|
for _, id := range cases {
|
|
t.Run(fmt.Sprintf("id=%q", id), func(t *testing.T) {
|
|
_, err := store.AbsPath(id)
|
|
if err == nil {
|
|
t.Errorf("AbsPath(%q) succeeded, want error", id)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStore_Validate_RejectsOutsideRoot(t *testing.T) {
|
|
if os.PathSeparator != '/' {
|
|
t.Skip("Unix-style path test")
|
|
}
|
|
|
|
store, err := New(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
|
|
fileID := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
if err := os.WriteFile(filepath.Join(store.Root, fileID), []byte("x"), 0o640); err != nil {
|
|
t.Fatalf("create data file: %v", err)
|
|
}
|
|
if err := os.WriteFile(filepath.Join(store.Root, fileID+".meta.json"), []byte(`{"uploaded_at":"`+time.Now().Format(time.RFC3339Nano)+`"}`), 0o600); err != nil {
|
|
t.Fatalf("create sidecar: %v", err)
|
|
}
|
|
|
|
cases := []struct {
|
|
name string
|
|
path string
|
|
}{
|
|
{"parent", filepath.Join(store.Root, "..", "other", fileID)},
|
|
{"dotdot", filepath.Join(store.Root, "..", "..", "tmp", fileID)},
|
|
{"different_volume", "/other/volume/" + fileID},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
_, err := store.Validate(c.path)
|
|
if err == nil {
|
|
t.Errorf("Validate(%q) succeeded, want error", c.path)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStore_Sweep_DeletesExpiredKeepsRecent(t *testing.T) {
|
|
store, err := New(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
|
|
now := time.Now()
|
|
expiredID := "00000000000000000000000000000001"
|
|
freshID := "00000000000000000000000000000002"
|
|
noMetaID := "00000000000000000000000000000003"
|
|
|
|
create := func(id string, uploadedAt *time.Time, withData bool) {
|
|
if withData {
|
|
if err := os.WriteFile(filepath.Join(store.Root, id), []byte("x"), 0o640); err != nil {
|
|
t.Fatalf("create data file %s: %v", id, err)
|
|
}
|
|
}
|
|
if uploadedAt != nil {
|
|
sc := map[string]any{
|
|
"name": "x",
|
|
"size": 1,
|
|
"sha256": "abc",
|
|
"uploaded_at": uploadedAt.Format(time.RFC3339Nano),
|
|
}
|
|
b, _ := json.Marshal(sc)
|
|
if err := os.WriteFile(filepath.Join(store.Root, id+".meta.json"), b, 0o600); err != nil {
|
|
t.Fatalf("create sidecar %s: %v", id, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
create(expiredID, timePtr(now.Add(-2*time.Hour)), true)
|
|
create(freshID, timePtr(now), true)
|
|
create(noMetaID, nil, true)
|
|
|
|
deleted, err := store.Sweep(1 * time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("sweep: %v", err)
|
|
}
|
|
if deleted != 1 {
|
|
t.Errorf("deleted=%d, want 1", deleted)
|
|
}
|
|
|
|
if _, err := os.Stat(filepath.Join(store.Root, expiredID)); !os.IsNotExist(err) {
|
|
t.Errorf("expired data file still exists")
|
|
}
|
|
if _, err := os.Stat(filepath.Join(store.Root, expiredID+".meta.json")); !os.IsNotExist(err) {
|
|
t.Errorf("expired sidecar still exists")
|
|
}
|
|
if _, err := os.Stat(filepath.Join(store.Root, freshID)); err != nil {
|
|
t.Errorf("fresh data file missing: %v", err)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(store.Root, noMetaID)); err != nil {
|
|
t.Errorf("no-sidecar data file missing: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestStore_Sweep_OrphanSidecarReaped(t *testing.T) {
|
|
store, err := New(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
|
|
id := "00000000000000000000000000000004"
|
|
sc := map[string]any{
|
|
"name": "orphan",
|
|
"size": 1,
|
|
"sha256": "abc",
|
|
"uploaded_at": time.Now().Add(-2 * time.Hour).Format(time.RFC3339Nano),
|
|
}
|
|
b, _ := json.Marshal(sc)
|
|
if err := os.WriteFile(filepath.Join(store.Root, id+".meta.json"), b, 0o600); err != nil {
|
|
t.Fatalf("create orphan sidecar: %v", err)
|
|
}
|
|
|
|
deleted, err := store.Sweep(1 * time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("sweep: %v", err)
|
|
}
|
|
if deleted != 1 {
|
|
t.Errorf("deleted=%d, want 1", deleted)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(store.Root, id+".meta.json")); !os.IsNotExist(err) {
|
|
t.Errorf("orphan sidecar still exists")
|
|
}
|
|
}
|
|
|
|
func timePtr(t time.Time) *time.Time {
|
|
return &t
|
|
}
|
|
|
|
func TestStore_Validate_RejectsSidecarPath(t *testing.T) {
|
|
store, err := New(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
|
|
fileID := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
if err := os.WriteFile(filepath.Join(store.Root, fileID), []byte("x"), 0o640); err != nil {
|
|
t.Fatalf("create data file: %v", err)
|
|
}
|
|
if err := os.WriteFile(filepath.Join(store.Root, fileID+".meta.json"), []byte(`{"uploaded_at":"`+time.Now().Format(time.RFC3339Nano)+`"}`), 0o600); err != nil {
|
|
t.Fatalf("create sidecar: %v", err)
|
|
}
|
|
|
|
_, err = store.Validate(filepath.Join(store.Root, fileID+".meta.json"))
|
|
if err == nil {
|
|
t.Errorf("Validate(sidecar) succeeded, want error")
|
|
}
|
|
}
|
|
|
|
func TestStore_Sweep_DeletesDataWithCorruptSidecar(t *testing.T) {
|
|
store, err := New(t.TempDir())
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
|
|
id := "00000000000000000000000000000005"
|
|
dataPath := filepath.Join(store.Root, id)
|
|
metaPath := dataPath + ".meta.json"
|
|
|
|
if err := os.WriteFile(dataPath, []byte("stale data"), 0o640); err != nil {
|
|
t.Fatalf("create data file: %v", err)
|
|
}
|
|
// Corrupt sidecar: not valid JSON.
|
|
if err := os.WriteFile(metaPath, []byte("not valid json"), 0o600); err != nil {
|
|
t.Fatalf("create sidecar: %v", err)
|
|
}
|
|
// Set mtime well in the past so the fallback triggers.
|
|
past := time.Now().Add(-2 * time.Hour)
|
|
if err := os.Chtimes(dataPath, past, past); err != nil {
|
|
t.Fatalf("set mtime: %v", err)
|
|
}
|
|
|
|
deleted, err := store.Sweep(1 * time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("sweep: %v", err)
|
|
}
|
|
if deleted != 1 {
|
|
t.Errorf("deleted=%d, want 1", deleted)
|
|
}
|
|
if _, err := os.Stat(dataPath); !os.IsNotExist(err) {
|
|
t.Errorf("data file with corrupt sidecar still exists")
|
|
}
|
|
if _, err := os.Stat(metaPath); !os.IsNotExist(err) {
|
|
t.Errorf("corrupt sidecar still exists")
|
|
}
|
|
}
|