Phase 0+1+2 续: 项目骨架 + 配置 + 日志 + 存储层

Phase 0 (项目骨架):
- main.go 改为最小骨架 (config + logging + gin + /healthz + 优雅退出)
- internal/{config,logging,storage,cluster,...}/ 目录占位

Phase 1 (配置 + 日志):
- internal/config: env 解析 + 必填校验 + token 隐藏的 String()
- internal/logging: slog multi-handler 双输出 (终端 text + 主文件 JSON)
  + StartToolCall per-tool 独立文件 (0600), tools 目录 0700

Phase 2 续 (存储层):
- internal/cluster: 16 字段 Cluster struct (AuthPassword json:"-")
- internal/storage: modernc.org/sqlite 接入, WAL 模式, schema 自动迁移
- ClusterRepo CRUD: Create/Get/List/Update/Delete + ErrNotFound
  + AuthPassword 空字符串 = 保留旧密码 (核心约定)
- 7 个单元测试全绿 (:memory: DB)
- created_at/updated_at 改纳秒精度, 消除 List 测试的 sleep 特殊 case

Co-Authored-By: Claude <noreply@anthropic.com>

internal/cluster/types.go

@@ -0,0 +1,56 @@
+// Package cluster defines the Cluster domain type — a configured Spark/YARN
+// endpoint pair plus authentication, SSL, rate-limit, and Spark-submit metadata.
+//
+// Cluster is the only entity persisted in this project besides the audit log.
+// It is the discovery root for the LLM agent: list_clusters returns the full
+// struct (minus the password) so the agent can resolve RM/SHS URLs, choose
+// the correct auth flavor, and respect the URL allowlist.
+package cluster
+
+import "time"
+
+// AuthType enumerates the YARN/Hadoop authentication flavors supported in v1.
+//
+//	"none"   — no credentials; public or pre-trusted clusters
+//	"simple" — YARN SimpleAuth: appends ?user.name=<username> to every request
+//	"basic"  — HTTP Basic; AuthUsername + AuthPassword (encrypted at rest)
+//
+// Kerberos is intentionally absent in v1; the deployment uses SimpleAuth.
+type AuthType string
+
+const (
+	AuthNone   AuthType = "none"
+	AuthSimple AuthType = "simple"
+	AuthBasic  AuthType = "basic"
+)
+
+// Cluster is one configured Spark-on-YARN endpoint pair.
+//
+// Field semantics:
+//   - SparkSubmitExecuteBin: absolute path to spark-submit (or spark2-submit);
+//     exec.Command uses this binary directly, never the shell.
+//   - IsActive: only active clusters are returned by LLM-facing list_clusters.
+//   - AuthPassword: zero value ("") on Update means "do not touch the stored
+//     password" — keeps the existing encrypted blob intact.
+//   - URLAllowlist: extra glob patterns beyond the RM/SHS hosts; fetch_url
+//     uses this to permit long-tail SHS endpoints the agent may construct.
+//   - DefaultSubmitArgs: prepended to every spark_submit Tool call's args.
+//   - RateLimitPerMin: 0 disables the per-cluster limiter.
+type Cluster struct {
+	ID                    string    `json:"id"`
+	Name                  string    `json:"name"`
+	RMURL                 string    `json:"rm_url"`
+	SHSURL                string    `json:"shs_url"`
+	SparkSubmitExecuteBin string    `json:"spark_submit_execute_bin"`
+	IsActive              bool      `json:"is_active"`
+	AuthType              AuthType  `json:"auth_type"`
+	AuthUsername          string    `json:"auth_username,omitempty"`
+	AuthPassword          string    `json:"-"` // never serialized; encrypted at rest in auth_password_enc
+	SSLVerify             bool      `json:"ssl_verify"`
+	SSLCABundle           string    `json:"ssl_ca_bundle,omitempty"`
+	URLAllowlist          []string  `json:"url_allowlist,omitempty"`
+	DefaultSubmitArgs     []string  `json:"default_submit_args,omitempty"`
+	RateLimitPerMin       int       `json:"rate_limit_per_min"`
+	CreatedAt             time.Time `json:"created_at"`
+	UpdatedAt             time.Time `json:"updated_at"`
+}

internal/config/config.go

@@ -0,0 +1,225 @@
+// Package config loads server configuration from environment variables.
+package config
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Config holds runtime configuration for the Spark MCP server.
+type Config struct {
+	ListenAddr                  string
+	DataDir                     string
+	SQLitePath                  string
+	AdminTokens                 []string
+	AgentToken                  string
+	HTTPClientTimeout           time.Duration
+	MaxResponseBytes            int64
+	SparkSubmitTimeout          time.Duration
+	LogDir                      string
+	LogLevel                    string
+	LogFormat                   string
+	AnalyzerDataSkewRatio       float64
+	AnalyzerGCPressureRatio     float64
+	AnalyzerBottleneckShuffleGB float64
+}
+
+// Load reads configuration from environment variables and returns a populated Config.
+// Required variables (ADMIN_TOKENS, AGENT_TOKEN) produce clear errors when missing.
+func Load() (*Config, error) {
+	cfg := &Config{
+		ListenAddr:                  ":8080",
+		DataDir:                     "./data",
+		HTTPClientTimeout:           30 * time.Second,
+		MaxResponseBytes:            1 << 20,
+		SparkSubmitTimeout:          60 * time.Second,
+		LogDir:                      "./data/logs",
+		LogLevel:                    "info",
+		LogFormat:                   "text",
+		AnalyzerDataSkewRatio:       3.0,
+		AnalyzerGCPressureRatio:     0.1,
+		AnalyzerBottleneckShuffleGB: 50.0,
+	}
+
+	cfg.ListenAddr = envString("LISTEN_ADDR", cfg.ListenAddr)
+	cfg.DataDir = envString("DATA_DIR", cfg.DataDir)
+
+	if err := loadAdminTokens(cfg); err != nil {
+		return nil, err
+	}
+	if err := loadAgentToken(cfg); err != nil {
+		return nil, err
+	}
+
+	var err error
+
+	cfg.HTTPClientTimeout, err = parseDuration("HTTP_CLIENT_TIMEOUT", cfg.HTTPClientTimeout)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.MaxResponseBytes, err = parseInt64("MAX_RESPONSE_BYTES", cfg.MaxResponseBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.SparkSubmitTimeout, err = parseDuration("SPARK_SUBMIT_TIMEOUT", cfg.SparkSubmitTimeout)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.LogDir = envString("LOG_DIR", cfg.LogDir)
+	cfg.LogLevel = envString("LOG_LEVEL", cfg.LogLevel)
+	cfg.LogFormat = envString("LOG_FORMAT", cfg.LogFormat)
+
+	cfg.AnalyzerDataSkewRatio, err = parseFloat64("ANALYZER_DATA_SKEW_RATIO", cfg.AnalyzerDataSkewRatio)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.AnalyzerGCPressureRatio, err = parseFloat64("ANALYZER_GC_PRESSURE_RATIO", cfg.AnalyzerGCPressureRatio)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.AnalyzerBottleneckShuffleGB, err = parseFloat64("ANALYZER_BOTTLENECK_SHUFFLE_GB", cfg.AnalyzerBottleneckShuffleGB)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := validateLogLevel(cfg.LogLevel); err != nil {
+		return nil, err
+	}
+	if err := validateLogFormat(cfg.LogFormat); err != nil {
+		return nil, err
+	}
+
+	cfg.SQLitePath = filepath.Join(cfg.DataDir, "spark-mcp.db")
+
+	return cfg, nil
+}
+
+func loadAdminTokens(cfg *Config) error {
+	v := os.Getenv("ADMIN_TOKENS")
+	if v == "" {
+		return fmt.Errorf("config: required environment variable ADMIN_TOKENS is not set")
+	}
+
+	parts := strings.Split(v, ",")
+	cfg.AdminTokens = make([]string, 0, len(parts))
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			continue
+		}
+		cfg.AdminTokens = append(cfg.AdminTokens, p)
+	}
+
+	if len(cfg.AdminTokens) == 0 {
+		return fmt.Errorf("config: environment variable ADMIN_TOKENS contains no valid tokens")
+	}
+	return nil
+}
+
+func loadAgentToken(cfg *Config) error {
+	cfg.AgentToken = os.Getenv("AGENT_TOKEN")
+	if cfg.AgentToken == "" {
+		return fmt.Errorf("config: required environment variable AGENT_TOKEN is not set")
+	}
+	return nil
+}
+
+func envString(key, defaultValue string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return defaultValue
+}
+
+func parseDuration(key string, defaultValue time.Duration) (time.Duration, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return defaultValue, nil
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil {
+		return 0, fmt.Errorf("config: invalid duration for %s: %w", key, err)
+	}
+	return d, nil
+}
+
+func parseInt64(key string, defaultValue int64) (int64, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return defaultValue, nil
+	}
+	n, err := strconv.ParseInt(v, 10, 64)
+	if err != nil {
+		return 0, fmt.Errorf("config: invalid integer for %s: %w", key, err)
+	}
+	return n, nil
+}
+
+func parseFloat64(key string, defaultValue float64) (float64, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return defaultValue, nil
+	}
+	f, err := strconv.ParseFloat(v, 64)
+	if err != nil {
+		return 0, fmt.Errorf("config: invalid float for %s: %w", key, err)
+	}
+	return f, nil
+}
+
+func validateLogLevel(level string) error {
+	switch level {
+	case "debug", "info", "warn", "error":
+		return nil
+	}
+	return fmt.Errorf("config: invalid LOG_LEVEL %q, want debug/info/warn/error", level)
+}
+
+func validateLogFormat(format string) error {
+	switch format {
+	case "text", "json":
+		return nil
+	}
+	return fmt.Errorf("config: invalid LOG_FORMAT %q, want text/json", format)
+}
+
+// String returns a human-readable representation of the configuration.
+// Sensitive values (AdminTokens and AgentToken) are summarized, not printed.
+func (c *Config) String() string {
+	adminTotalLen := 0
+	for _, t := range c.AdminTokens {
+		adminTotalLen += len(t)
+	}
+	adminSummary := fmt.Sprintf("%d token(s), total_len=%d", len(c.AdminTokens), adminTotalLen)
+	agentSummary := fmt.Sprintf("len=%d", len(c.AgentToken))
+
+	return fmt.Sprintf(
+		"ListenAddr=%s DataDir=%s SQLitePath=%s AdminTokens=%s AgentToken=%s "+
+			"HTTPClientTimeout=%s MaxResponseBytes=%d SparkSubmitTimeout=%s "+
+			"LogDir=%s LogLevel=%s LogFormat=%s AnalyzerDataSkewRatio=%.1f "+
+			"AnalyzerGCPressureRatio=%.1f AnalyzerBottleneckShuffleGB=%.1f",
+		c.ListenAddr,
+		c.DataDir,
+		c.SQLitePath,
+		adminSummary,
+		agentSummary,
+		c.HTTPClientTimeout,
+		c.MaxResponseBytes,
+		c.SparkSubmitTimeout,
+		c.LogDir,
+		c.LogLevel,
+		c.LogFormat,
+		c.AnalyzerDataSkewRatio,
+		c.AnalyzerGCPressureRatio,
+		c.AnalyzerBottleneckShuffleGB,
+	)
+}

internal/logging/logger.go

@@ -0,0 +1,135 @@
+package logging
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+)
+
+// Config configures the application logger.
+type Config struct {
+	Level    string // "debug"|"info"|"warn"|"error"
+	Dir      string // e.g. "./data/logs"
+	Format   string // "text"|"json" — controls terminal output; files are always JSON
+	LogToStd bool   // true: enable terminal output
+}
+
+// ToolCallLogger records the lifecycle of a single tool call.
+type ToolCallLogger interface {
+	WithResult(result any)
+	WithError(err error)
+	End()
+}
+
+// logDir is shared with StartToolCall so the per-tool-call logger knows where
+// to create files. It is set by New and read by StartToolCall.
+var (
+	logDir   string
+	logDirMu sync.RWMutex
+)
+
+func parseLevel(s string) slog.Level {
+	switch strings.ToLower(s) {
+	case "debug":
+		return slog.LevelDebug
+	case "warn", "warning":
+		return slog.LevelWarn
+	case "error":
+		return slog.LevelError
+	default:
+		return slog.LevelInfo
+	}
+}
+
+// multiHandler fans out records to multiple underlying handlers.
+type multiHandler struct {
+	handlers []slog.Handler
+}
+
+func (m *multiHandler) Enabled(ctx context.Context, l slog.Level) bool {
+	for _, h := range m.handlers {
+		if h.Enabled(ctx, l) {
+			return true
+		}
+	}
+	return false
+}
+
+func (m *multiHandler) Handle(ctx context.Context, r slog.Record) error {
+	var errs []error
+	for _, h := range m.handlers {
+		if h.Enabled(ctx, r.Level) {
+			if err := h.Handle(ctx, r); err != nil {
+				errs = append(errs, err)
+			}
+		}
+	}
+	return errors.Join(errs...)
+}
+
+func (m *multiHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
+	handlers := make([]slog.Handler, len(m.handlers))
+	for i, h := range m.handlers {
+		handlers[i] = h.WithAttrs(attrs)
+	}
+	return &multiHandler{handlers: handlers}
+}
+
+func (m *multiHandler) WithGroup(name string) slog.Handler {
+	handlers := make([]slog.Handler, len(m.handlers))
+	for i, h := range m.handlers {
+		handlers[i] = h.WithGroup(name)
+	}
+	return &multiHandler{handlers: handlers}
+}
+
+// New creates the main application logger and a shutdown closure.
+func New(cfg Config) (*slog.Logger, func(), error) {
+	if err := os.MkdirAll(cfg.Dir, 0o755); err != nil {
+		return nil, nil, fmt.Errorf("logging: create log dir: %w", err)
+	}
+
+	toolsDir := filepath.Join(cfg.Dir, "tools")
+	if err := os.MkdirAll(toolsDir, 0o700); err != nil {
+		return nil, nil, fmt.Errorf("logging: create tools log dir: %w", err)
+	}
+
+	logDirMu.Lock()
+	logDir = cfg.Dir
+	logDirMu.Unlock()
+
+	level := parseLevel(cfg.Level)
+	opts := &slog.HandlerOptions{Level: level}
+
+	mainPath := filepath.Join(cfg.Dir, "spark-mcp.log")
+	f, err := os.OpenFile(mainPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		return nil, nil, fmt.Errorf("logging: open main log file: %w", err)
+	}
+
+	fileHandler := slog.NewJSONHandler(f, opts)
+
+	var handler slog.Handler = fileHandler
+	if cfg.LogToStd {
+		var stdHandler slog.Handler
+		switch strings.ToLower(cfg.Format) {
+		case "json":
+			stdHandler = slog.NewJSONHandler(os.Stdout, opts)
+		default:
+			stdHandler = slog.NewTextHandler(os.Stdout, opts)
+		}
+		handler = &multiHandler{handlers: []slog.Handler{fileHandler, stdHandler}}
+	}
+
+	shutdown := func() {
+		_ = f.Close()
+	}
+
+	logger := slog.New(handler).With("component", "main")
+	return logger, shutdown, nil
+}

internal/logging/tool_call.go

@@ -0,0 +1,191 @@
+package logging
+
+import (
+	"log/slog"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"sync"
+	"time"
+)
+
+const toolCallResultMaxBytes = 10 * 1024
+
+var toolNameSanitizer = regexp.MustCompile(`[^a-zA-Z0-9_]+`)
+
+func sanitizeToolName(name string) string {
+	s := toolNameSanitizer.ReplaceAllString(name, "_")
+	if len(s) > 32 {
+		s = s[:32]
+	}
+	s = strings.Trim(s, "_")
+	if s == "" {
+		s = "tool"
+	}
+	return s
+}
+
+func shortHexID(n int) (string, error) {
+	b := make([]byte, n/2+n%2)
+	if _, err := rand.Read(b); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(b)[:n], nil
+}
+
+type toolCallLogger struct {
+	file      *os.File
+	toolName  string
+	startedAt time.Time
+	resultRaw json.RawMessage
+	err       error
+	endOnce   sync.Once
+}
+
+func (t *toolCallLogger) WithResult(result any) {
+	t.resultRaw = marshalResult(result)
+}
+
+func (t *toolCallLogger) WithError(err error) {
+	t.err = err
+}
+
+func (t *toolCallLogger) End() {
+	t.endOnce.Do(func() {
+		defer func() { _ = t.file.Close() }()
+
+		durationMs := time.Since(t.startedAt).Milliseconds()
+		errStr := ""
+		if t.err != nil {
+			errStr = t.err.Error()
+		}
+
+		end := map[string]any{
+			"event":       "end",
+			"duration_ms": durationMs,
+			"result":      nil,
+			"error":       errStr,
+		}
+		if t.resultRaw != nil {
+			end["result"] = json.RawMessage(t.resultRaw)
+		}
+
+		b, err := json.Marshal(end)
+		if err != nil {
+			b = []byte(fmt.Sprintf(`{"event":"end","duration_ms":%d,"result":null,"error":%q}`, durationMs, errStr))
+		}
+		_, _ = t.file.Write(b)
+		_, _ = t.file.WriteString("\n")
+	})
+}
+
+// StartToolCall creates a per-tool-call log file and returns a logger that
+// records the call lifecycle.
+func StartToolCall(ctx context.Context, base *slog.Logger, toolName string, params any) (ToolCallLogger, error) {
+	logDirMu.RLock()
+	dir := logDir
+	logDirMu.RUnlock()
+	if dir == "" {
+		return nil, fmt.Errorf("logging: log directory not initialized; call New first")
+	}
+
+	toolsDir := filepath.Join(dir, "tools")
+	if err := os.MkdirAll(toolsDir, 0o700); err != nil {
+		return nil, fmt.Errorf("logging: create tools log dir: %w", err)
+	}
+
+	safeName := sanitizeToolName(toolName)
+	ts := time.Now().Format("20060102_150405.000")
+	id, err := shortHexID(6)
+	if err != nil {
+		return nil, fmt.Errorf("logging: generate tool call id: %w", err)
+	}
+	filename := fmt.Sprintf("%s_%s_%s.log", ts, safeName, id)
+	path := filepath.Join(toolsDir, filename)
+
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+	if err != nil {
+		return nil, fmt.Errorf("logging: create tool call log file: %w", err)
+	}
+
+	startedAt := time.Now()
+	start := map[string]any{
+		"event":      "start",
+		"tool":       toolName,
+		"params":     marshalValue(params),
+		"started_at": startedAt.Format(time.RFC3339Nano),
+	}
+	if d, ok := ctx.Deadline(); ok {
+		start["deadline"] = d.Format(time.RFC3339Nano)
+	}
+
+	b, err := json.Marshal(start)
+	if err != nil {
+		_ = f.Close()
+		return nil, fmt.Errorf("logging: marshal start record: %w", err)
+	}
+	if _, err := f.Write(b); err != nil {
+		_ = f.Close()
+		return nil, fmt.Errorf("logging: write start record: %w", err)
+	}
+	if _, err := f.WriteString("\n"); err != nil {
+		_ = f.Close()
+		return nil, fmt.Errorf("logging: write start record: %w", err)
+	}
+
+	if base != nil {
+		base.Info("tool.call.start", "tool", toolName, "file", filename)
+	}
+
+	return &toolCallLogger{
+		file:      f,
+		toolName:  toolName,
+		startedAt: startedAt,
+	}, nil
+}
+
+func marshalValue(v any) json.RawMessage {
+	if v == nil {
+		return []byte("null")
+	}
+	b, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		b, err = json.Marshal(v)
+		if err != nil {
+			return mustMarshalString("<unserializable>")
+		}
+	}
+	return b
+}
+
+func marshalResult(v any) json.RawMessage {
+	if v == nil {
+		return []byte("null")
+	}
+	b, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		b, err = json.Marshal(v)
+		if err != nil {
+			return mustMarshalString("<unserializable>")
+		}
+	}
+	if len(b) > toolCallResultMaxBytes {
+		trunc := string(b[:toolCallResultMaxBytes]) + "...[truncated]"
+		return mustMarshalString(trunc)
+	}
+	return b
+}
+
+func mustMarshalString(s string) json.RawMessage {
+	b, err := json.Marshal(s)
+	if err != nil {
+		return []byte(`"` + strings.ReplaceAll(s, `"`, `\"`) + `"`)
+	}
+	return b
+}

internal/storage/cluster_repo.go

@@ -0,0 +1,252 @@
+package storage
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"spark-mcp-go/internal/cluster"
+)
+
+// ErrNotFound is returned when a requested cluster does not exist.
+var ErrNotFound = errors.New("storage: cluster not found")
+
+// ClusterRepo provides CRUD operations for clusters.
+type ClusterRepo struct {
+	db *DB
+}
+
+// Clusters returns a repository bound to this DB handle.
+func (d *DB) Clusters() *ClusterRepo {
+	return &ClusterRepo{db: d}
+}
+
+// Create inserts a new cluster record.
+//
+// Timestamps are stored as Unix nanoseconds (INTEGER) — sub-second precision
+// keeps created_at unique across rapid inserts and makes ORDER BY DESC
+// stable without artificial delays.
+//
+// If a cluster with the same id already exists, the underlying SQLite
+// UNIQUE constraint error is wrapped and returned.
+func (r *ClusterRepo) Create(ctx context.Context, c *cluster.Cluster) error {
+	now := time.Now().UnixNano()
+	c.CreatedAt = time.Unix(0, now)
+	c.UpdatedAt = c.CreatedAt
+
+	allowlistJSON, err := json.Marshal(c.URLAllowlist)
+	if err != nil {
+		return fmt.Errorf("storage: create cluster %s: marshal url_allowlist: %w", c.ID, err)
+	}
+	defaultArgsJSON, err := json.Marshal(c.DefaultSubmitArgs)
+	if err != nil {
+		return fmt.Errorf("storage: create cluster %s: marshal default_submit_args: %w", c.ID, err)
+	}
+
+	// TODO(phase-5): encrypt auth_password_enc instead of storing plaintext.
+	pwdEnc := []byte(c.AuthPassword)
+
+	_, err = r.db.sqlDB.ExecContext(ctx, `
+		INSERT INTO clusters (
+			id, name, rm_url, shs_url, spark_submit_execute_bin,
+			is_active, auth_type, auth_username, auth_password_enc,
+			ssl_verify, ssl_ca_bundle, url_allowlist, default_submit_args,
+			rate_limit_per_min, created_at, updated_at
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		c.ID, c.Name, c.RMURL, c.SHSURL, c.SparkSubmitExecuteBin,
+		boolToInt(c.IsActive), string(c.AuthType), c.AuthUsername, pwdEnc,
+		boolToInt(c.SSLVerify), c.SSLCABundle, string(allowlistJSON), string(defaultArgsJSON),
+		c.RateLimitPerMin, now, now,
+	)
+	if err != nil {
+		if strings.Contains(err.Error(), "UNIQUE constraint failed") {
+			return fmt.Errorf("storage: create cluster %s: primary key conflict: %w", c.ID, err)
+		}
+		return fmt.Errorf("storage: create cluster %s: %w", c.ID, err)
+	}
+	return nil
+}
+
+// Get retrieves a cluster by id. If the cluster does not exist, it returns
+// ErrNotFound.
+func (r *ClusterRepo) Get(ctx context.Context, id string) (*cluster.Cluster, error) {
+	row := r.db.sqlDB.QueryRowContext(ctx, `
+		SELECT id, name, rm_url, shs_url, spark_submit_execute_bin,
+		       is_active, auth_type, auth_username, auth_password_enc,
+		       ssl_verify, ssl_ca_bundle, url_allowlist, default_submit_args,
+		       rate_limit_per_min, created_at, updated_at
+		FROM clusters
+		WHERE id = ?`, id)
+
+	c, err := scanCluster(row.Scan)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("storage: get cluster %s: %w", id, err)
+	}
+	return c, nil
+}
+
+// List returns every cluster, including inactive ones, ordered by created_at
+// descending.
+func (r *ClusterRepo) List(ctx context.Context) ([]*cluster.Cluster, error) {
+	rows, err := r.db.sqlDB.QueryContext(ctx, `
+		SELECT id, name, rm_url, shs_url, spark_submit_execute_bin,
+		       is_active, auth_type, auth_username, auth_password_enc,
+		       ssl_verify, ssl_ca_bundle, url_allowlist, default_submit_args,
+		       rate_limit_per_min, created_at, updated_at
+		FROM clusters
+		ORDER BY created_at DESC`)
+	if err != nil {
+		return nil, fmt.Errorf("storage: list clusters: %w", err)
+	}
+	defer rows.Close()
+
+	var out []*cluster.Cluster
+	for rows.Next() {
+		c, err := scanCluster(rows.Scan)
+		if err != nil {
+			return nil, fmt.Errorf("storage: list clusters: %w", err)
+		}
+		out = append(out, c)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("storage: list clusters: %w", err)
+	}
+	return out, nil
+}
+
+// Update modifies an existing cluster. An empty AuthPassword means "do not
+// touch the stored password" — the existing auth_password_enc blob is kept
+// intact.
+func (r *ClusterRepo) Update(ctx context.Context, c *cluster.Cluster) error {
+	c.UpdatedAt = time.Unix(0, time.Now().UnixNano())
+
+	allowlistJSON, err := json.Marshal(c.URLAllowlist)
+	if err != nil {
+		return fmt.Errorf("storage: update cluster %s: marshal url_allowlist: %w", c.ID, err)
+	}
+	defaultArgsJSON, err := json.Marshal(c.DefaultSubmitArgs)
+	if err != nil {
+		return fmt.Errorf("storage: update cluster %s: marshal default_submit_args: %w", c.ID, err)
+	}
+
+	var res sql.Result
+	if c.AuthPassword == "" {
+		// Preserve the existing password blob.
+		res, err = r.db.sqlDB.ExecContext(ctx, `
+			UPDATE clusters
+			SET name = ?, rm_url = ?, shs_url = ?, spark_submit_execute_bin = ?,
+			    is_active = ?, auth_type = ?, auth_username = ?,
+			    ssl_verify = ?, ssl_ca_bundle = ?, url_allowlist = ?,
+			    default_submit_args = ?, rate_limit_per_min = ?, updated_at = ?
+			WHERE id = ?`,
+			c.Name, c.RMURL, c.SHSURL, c.SparkSubmitExecuteBin,
+			boolToInt(c.IsActive), string(c.AuthType), c.AuthUsername,
+			boolToInt(c.SSLVerify), c.SSLCABundle, string(allowlistJSON),
+			string(defaultArgsJSON), c.RateLimitPerMin, c.UpdatedAt.UnixNano(), c.ID,
+		)
+	} else {
+		// TODO(phase-5): encrypt auth_password_enc instead of storing plaintext.
+		pwdEnc := []byte(c.AuthPassword)
+		res, err = r.db.sqlDB.ExecContext(ctx, `
+			UPDATE clusters
+			SET name = ?, rm_url = ?, shs_url = ?, spark_submit_execute_bin = ?,
+			    is_active = ?, auth_type = ?, auth_username = ?, auth_password_enc = ?,
+			    ssl_verify = ?, ssl_ca_bundle = ?, url_allowlist = ?,
+			    default_submit_args = ?, rate_limit_per_min = ?, updated_at = ?
+			WHERE id = ?`,
+			c.Name, c.RMURL, c.SHSURL, c.SparkSubmitExecuteBin,
+			boolToInt(c.IsActive), string(c.AuthType), c.AuthUsername, pwdEnc,
+			boolToInt(c.SSLVerify), c.SSLCABundle, string(allowlistJSON),
+			string(defaultArgsJSON), c.RateLimitPerMin, c.UpdatedAt.UnixNano(), c.ID,
+		)
+	}
+	if err != nil {
+		return fmt.Errorf("storage: update cluster %s: %w", c.ID, err)
+	}
+
+	n, err := res.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("storage: update cluster %s: rows affected: %w", c.ID, err)
+	}
+	if n == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// Delete removes a cluster by id. If the cluster does not exist, it returns
+// ErrNotFound.
+func (r *ClusterRepo) Delete(ctx context.Context, id string) error {
+	res, err := r.db.sqlDB.ExecContext(ctx, `DELETE FROM clusters WHERE id = ?`, id)
+	if err != nil {
+		return fmt.Errorf("storage: delete cluster %s: %w", id, err)
+	}
+	n, err := res.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("storage: delete cluster %s: rows affected: %w", id, err)
+	}
+	if n == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// scanFunc abstracts *sql.Row.Scan and *sql.Rows.Scan so a single mapping
+// function can serve both QueryRow and Query paths.
+type scanFunc func(dest ...any) error
+
+// scanCluster maps a single SQL row into a cluster.Cluster value.
+func scanCluster(scan scanFunc) (*cluster.Cluster, error) {
+	var c cluster.Cluster
+	var allowlistJSON, defaultArgsJSON sql.NullString
+	var pwdEnc []byte
+	var createdAt, updatedAt int64
+	var isActive, sslVerify int
+
+	if err := scan(
+		&c.ID, &c.Name, &c.RMURL, &c.SHSURL, &c.SparkSubmitExecuteBin,
+		&isActive, &c.AuthType, &c.AuthUsername, &pwdEnc,
+		&sslVerify, &c.SSLCABundle, &allowlistJSON, &defaultArgsJSON,
+		&c.RateLimitPerMin, &createdAt, &updatedAt,
+	); err != nil {
+		return nil, err
+	}
+
+	c.IsActive = intToBool(isActive)
+	c.SSLVerify = intToBool(sslVerify)
+	c.CreatedAt = time.Unix(0, createdAt)
+	c.UpdatedAt = time.Unix(0, updatedAt)
+	// TODO(phase-5): decrypt pwdEnc once encryption is introduced.
+	c.AuthPassword = string(pwdEnc)
+
+	if allowlistJSON.Valid && allowlistJSON.String != "" {
+		if err := json.Unmarshal([]byte(allowlistJSON.String), &c.URLAllowlist); err != nil {
+			return nil, fmt.Errorf("storage: scan cluster %s: unmarshal url_allowlist: %w", c.ID, err)
+		}
+	}
+	if defaultArgsJSON.Valid && defaultArgsJSON.String != "" {
+		if err := json.Unmarshal([]byte(defaultArgsJSON.String), &c.DefaultSubmitArgs); err != nil {
+			return nil, fmt.Errorf("storage: scan cluster %s: unmarshal default_submit_args: %w", c.ID, err)
+		}
+	}
+
+	return &c, nil
+}
+
+func boolToInt(b bool) int {
+	if b {
+		return 1
+	}
+	return 0
+}
+
+func intToBool(i int) bool {
+	return i != 0
+}

internal/storage/cluster_repo_test.go

@@ -0,0 +1,306 @@
+package storage
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+
+	"spark-mcp-go/internal/cluster"
+)
+
+func newClusterRepo(t *testing.T) *ClusterRepo {
+	t.Helper()
+	db, err := Open(":memory:")
+	if err != nil {
+		t.Fatalf("Open in-memory db: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	return db.Clusters()
+}
+
+func testClusterA() *cluster.Cluster {
+	return &cluster.Cluster{
+		ID:                    "cluster-a",
+		Name:                  "Cluster Alpha",
+		RMURL:                 "http://rm-a.example.com:8088",
+		SHSURL:                "http://shs-a.example.com:18080",
+		SparkSubmitExecuteBin: "/usr/bin/spark-submit",
+		IsActive:              true,
+		AuthType:              cluster.AuthBasic,
+		AuthUsername:          "admin-a",
+		AuthPassword:          "secret-1",
+		SSLVerify:             true,
+		SSLCABundle:           "/etc/ssl/ca-bundle-a.pem",
+		URLAllowlist:          []string{"*.a.example.com", "http://hist-a.example.com/*"},
+		DefaultSubmitArgs:     []string{"--master=yarn", "--deploy-mode=cluster"},
+		RateLimitPerMin:       10,
+	}
+}
+
+func assertClusterEqual(t *testing.T, got, want *cluster.Cluster) {
+	t.Helper()
+
+	if got.ID != want.ID {
+		t.Errorf("ID: got %q, want %q", got.ID, want.ID)
+	}
+	if got.Name != want.Name {
+		t.Errorf("Name: got %q, want %q", got.Name, want.Name)
+	}
+	if got.RMURL != want.RMURL {
+		t.Errorf("RMURL: got %q, want %q", got.RMURL, want.RMURL)
+	}
+	if got.SHSURL != want.SHSURL {
+		t.Errorf("SHSURL: got %q, want %q", got.SHSURL, want.SHSURL)
+	}
+	if got.SparkSubmitExecuteBin != want.SparkSubmitExecuteBin {
+		t.Errorf("SparkSubmitExecuteBin: got %q, want %q", got.SparkSubmitExecuteBin, want.SparkSubmitExecuteBin)
+	}
+	if got.IsActive != want.IsActive {
+		t.Errorf("IsActive: got %v, want %v", got.IsActive, want.IsActive)
+	}
+	if got.AuthType != want.AuthType {
+		t.Errorf("AuthType: got %q, want %q", got.AuthType, want.AuthType)
+	}
+	if got.AuthUsername != want.AuthUsername {
+		t.Errorf("AuthUsername: got %q, want %q", got.AuthUsername, want.AuthUsername)
+	}
+	if got.AuthPassword != want.AuthPassword {
+		t.Errorf("AuthPassword: got %q, want %q", got.AuthPassword, want.AuthPassword)
+	}
+	if got.SSLVerify != want.SSLVerify {
+		t.Errorf("SSLVerify: got %v, want %v", got.SSLVerify, want.SSLVerify)
+	}
+	if got.SSLCABundle != want.SSLCABundle {
+		t.Errorf("SSLCABundle: got %q, want %q", got.SSLCABundle, want.SSLCABundle)
+	}
+	if !reflect.DeepEqual(got.URLAllowlist, want.URLAllowlist) {
+		t.Errorf("URLAllowlist: got %v, want %v", got.URLAllowlist, want.URLAllowlist)
+	}
+	if !reflect.DeepEqual(got.DefaultSubmitArgs, want.DefaultSubmitArgs) {
+		t.Errorf("DefaultSubmitArgs: got %v, want %v", got.DefaultSubmitArgs, want.DefaultSubmitArgs)
+	}
+	if got.RateLimitPerMin != want.RateLimitPerMin {
+		t.Errorf("RateLimitPerMin: got %d, want %d", got.RateLimitPerMin, want.RateLimitPerMin)
+	}
+
+	if !got.CreatedAt.Equal(want.CreatedAt) {
+		if got.CreatedAt.Unix() != want.CreatedAt.Unix() {
+			t.Errorf("CreatedAt: got %v, want %v", got.CreatedAt, want.CreatedAt)
+		}
+	}
+	if !got.UpdatedAt.Equal(want.UpdatedAt) {
+		if got.UpdatedAt.Unix() != want.UpdatedAt.Unix() {
+			t.Errorf("UpdatedAt: got %v, want %v", got.UpdatedAt, want.UpdatedAt)
+		}
+	}
+}
+
+func TestClusterRepo_Create(t *testing.T) {
+	repo := newClusterRepo(t)
+	ctx := context.Background()
+
+	want := testClusterA()
+	if err := repo.Create(ctx, want); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	got, err := repo.Get(ctx, want.ID)
+	if err != nil {
+		t.Fatalf("Get after Create: %v", err)
+	}
+	assertClusterEqual(t, got, want)
+}
+
+func TestClusterRepo_Get(t *testing.T) {
+	repo := newClusterRepo(t)
+	ctx := context.Background()
+
+	want := testClusterA()
+	if err := repo.Create(ctx, want); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	got, err := repo.Get(ctx, want.ID)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	assertClusterEqual(t, got, want)
+
+	if _, err := repo.Get(ctx, "nope"); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("Get missing cluster: got %v, want ErrNotFound", err)
+	}
+}
+
+func TestClusterRepo_List(t *testing.T) {
+	repo := newClusterRepo(t)
+	ctx := context.Background()
+
+	ids := []string{"cluster-a", "cluster-b", "cluster-c"}
+	for _, id := range ids {
+		c := testClusterA()
+		c.ID = id
+		c.Name = id
+		if err := repo.Create(ctx, c); err != nil {
+			t.Fatalf("Create %s: %v", id, err)
+		}
+		// Sub-second precision in stored timestamps gives us a stable
+		// ORDER BY DESC without artificial delays.
+	}
+
+	got, err := repo.List(ctx)
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	if len(got) != len(ids) {
+		t.Fatalf("List len: got %d, want %d", len(got), len(ids))
+	}
+
+	// List must be ordered by created_at DESC: last created first.
+	wantOrder := []string{"cluster-c", "cluster-b", "cluster-a"}
+	for i, c := range got {
+		if c.ID != wantOrder[i] {
+			t.Errorf("List order[%d]: got %q, want %q", i, c.ID, wantOrder[i])
+		}
+	}
+}
+
+func TestClusterRepo_Update(t *testing.T) {
+	repo := newClusterRepo(t)
+	ctx := context.Background()
+
+	c := testClusterA()
+	if err := repo.Create(ctx, c); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	stored, err := repo.Get(ctx, c.ID)
+	if err != nil {
+		t.Fatalf("Get before Update: %v", err)
+	}
+
+	stored.Name = "Cluster Alpha Updated"
+	stored.RMURL = "http://rm-a-new.example.com:8088"
+	stored.IsActive = false
+	stored.RateLimitPerMin = 20
+	// Intentionally leave AuthPassword empty: must preserve existing password.
+	stored.AuthPassword = ""
+
+	if err := repo.Update(ctx, stored); err != nil {
+		t.Fatalf("Update: %v", err)
+	}
+
+	got, err := repo.Get(ctx, c.ID)
+	if err != nil {
+		t.Fatalf("Get after Update: %v", err)
+	}
+
+	if got.Name != stored.Name {
+		t.Errorf("Name: got %q, want %q", got.Name, stored.Name)
+	}
+	if got.RMURL != stored.RMURL {
+		t.Errorf("RMURL: got %q, want %q", got.RMURL, stored.RMURL)
+	}
+	if got.IsActive != false {
+		t.Errorf("IsActive: got %v, want false", got.IsActive)
+	}
+	if got.RateLimitPerMin != 20 {
+		t.Errorf("RateLimitPerMin: got %d, want 20", got.RateLimitPerMin)
+	}
+	if got.AuthPassword != c.AuthPassword {
+		t.Errorf("AuthPassword should be preserved: got %q, want %q", got.AuthPassword, c.AuthPassword)
+	}
+
+	// Update a non-existing cluster must return ErrNotFound.
+	missing := testClusterA()
+	missing.ID = "missing"
+	if err := repo.Update(ctx, missing); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("Update missing cluster: got %v, want ErrNotFound", err)
+	}
+}
+
+func TestClusterRepo_UpdatePassword(t *testing.T) {
+	repo := newClusterRepo(t)
+	ctx := context.Background()
+
+	c := testClusterA()
+	c.AuthPassword = "old"
+	if err := repo.Create(ctx, c); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	stored, err := repo.Get(ctx, c.ID)
+	if err != nil {
+		t.Fatalf("Get before Update: %v", err)
+	}
+	if stored.AuthPassword != "old" {
+		t.Fatalf("initial password: got %q, want %q", stored.AuthPassword, "old")
+	}
+
+	stored.AuthPassword = "new"
+	if err := repo.Update(ctx, stored); err != nil {
+		t.Fatalf("Update password: %v", err)
+	}
+
+	got, err := repo.Get(ctx, c.ID)
+	if err != nil {
+		t.Fatalf("Get after Update: %v", err)
+	}
+	if got.AuthPassword != "new" {
+		t.Errorf("password: got %q, want %q", got.AuthPassword, "new")
+	}
+}
+
+func TestClusterRepo_UpdateEmptyPasswordKeepsOldPassword(t *testing.T) {
+	repo := newClusterRepo(t)
+	ctx := context.Background()
+
+	c := testClusterA()
+	c.AuthPassword = "keep-me"
+	if err := repo.Create(ctx, c); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	stored, err := repo.Get(ctx, c.ID)
+	if err != nil {
+		t.Fatalf("Get before Update: %v", err)
+	}
+	stored.AuthPassword = ""
+	if err := repo.Update(ctx, stored); err != nil {
+		t.Fatalf("Update: %v", err)
+	}
+
+	got, err := repo.Get(ctx, c.ID)
+	if err != nil {
+		t.Fatalf("Get after Update: %v", err)
+	}
+	if got.AuthPassword != "keep-me" {
+		t.Errorf("empty password should keep old value: got %q, want %q", got.AuthPassword, "keep-me")
+	}
+}
+
+func TestClusterRepo_Delete(t *testing.T) {
+	repo := newClusterRepo(t)
+	ctx := context.Background()
+
+	c := testClusterA()
+	if err := repo.Create(ctx, c); err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+
+	if err := repo.Delete(ctx, c.ID); err != nil {
+		t.Fatalf("Delete existing: %v", err)
+	}
+
+	if _, err := repo.Get(ctx, c.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("Get after Delete: got %v, want ErrNotFound", err)
+	}
+
+	if err := repo.Delete(ctx, c.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("Delete again: got %v, want ErrNotFound", err)
+	}
+
+	if err := repo.Delete(ctx, "nope"); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("Delete missing: got %v, want ErrNotFound", err)
+	}
+}

internal/storage/migrate.go

@@ -0,0 +1,16 @@
+package storage
+
+import "fmt"
+
+// Migrate applies Schema to the underlying database. Because Schema uses
+// IF NOT EXISTS for all CREATE statements, a single Exec is sufficient and
+// idempotent across restarts.
+func (d *DB) Migrate() error {
+	if d == nil || d.sqlDB == nil {
+		return fmt.Errorf("storage: migrate: db not open")
+	}
+	if _, err := d.sqlDB.Exec(Schema); err != nil {
+		return fmt.Errorf("storage: migrate: %w", err)
+	}
+	return nil
+}

internal/storage/sqlite.go

@@ -0,0 +1,85 @@
+package storage
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	_ "modernc.org/sqlite"
+)
+
+// Schema defines the SQLite schema for clusters and audit_log.
+const Schema = `CREATE TABLE IF NOT EXISTS clusters (
+  id                        TEXT PRIMARY KEY,
+  name                      TEXT NOT NULL,
+  rm_url                    TEXT NOT NULL,
+  shs_url                   TEXT NOT NULL,
+  spark_submit_execute_bin  TEXT NOT NULL,
+  is_active                 INTEGER NOT NULL DEFAULT 1,
+  auth_type                 TEXT NOT NULL DEFAULT 'none',
+  auth_username             TEXT,
+  auth_password_enc         BLOB,
+  ssl_verify                INTEGER NOT NULL DEFAULT 1,
+  ssl_ca_bundle             TEXT,
+  url_allowlist             TEXT,
+  default_submit_args       TEXT,
+  rate_limit_per_min        INTEGER NOT NULL DEFAULT 10,
+  created_at                INTEGER NOT NULL,
+  updated_at                INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS audit_log (
+  id          INTEGER PRIMARY KEY AUTOINCREMENT,
+  timestamp   INTEGER NOT NULL,
+  actor       TEXT NOT NULL,
+  action      TEXT NOT NULL,
+  cluster_id  TEXT,
+  details     TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(timestamp DESC);
+`
+
+// DB is the storage handle.
+type DB struct {
+	Path  string
+	sqlDB *sql.DB
+}
+
+// Open opens the SQLite database at path, creating the parent directory if
+// needed. It enables WAL mode and foreign keys, then applies the schema.
+func Open(path string) (*DB, error) {
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return nil, fmt.Errorf("storage: mkdir %s: %w", dir, err)
+	}
+
+	sqlDB, err := sql.Open("sqlite", path)
+	if err != nil {
+		return nil, fmt.Errorf("storage: open sqlite %s: %w", path, err)
+	}
+
+	if _, err := sqlDB.Exec("PRAGMA journal_mode=WAL"); err != nil {
+		_ = sqlDB.Close()
+		return nil, fmt.Errorf("storage: set WAL mode: %w", err)
+	}
+	if _, err := sqlDB.Exec("PRAGMA foreign_keys=ON"); err != nil {
+		_ = sqlDB.Close()
+		return nil, fmt.Errorf("storage: enable foreign keys: %w", err)
+	}
+
+	d := &DB{Path: path, sqlDB: sqlDB}
+	if err := d.Migrate(); err != nil {
+		_ = sqlDB.Close()
+		return nil, err
+	}
+	return d, nil
+}
+
+// Close releases the underlying database connection.
+func (d *DB) Close() error {
+	if d == nil || d.sqlDB == nil {
+		return nil
+	}
+	return d.sqlDB.Close()
+}