diff --git a/internal/machine/cluster.go b/internal/machine/cluster.go index 13093209..b82edc70 100644 --- a/internal/machine/cluster.go +++ b/internal/machine/cluster.go @@ -15,6 +15,7 @@ import ( "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/caddyconfig" "github.com/psviderski/uncloud/internal/machine/constants" + "github.com/psviderski/uncloud/internal/machine/corromigrate" "github.com/psviderski/uncloud/internal/machine/corroservice" "github.com/psviderski/uncloud/internal/machine/dns" "github.com/psviderski/uncloud/internal/machine/docker" @@ -38,6 +39,9 @@ type clusterController struct { server *grpc.Server corroService corroservice.Service + // corrosionDir is the disk path that holds the Corrosion config and data. + // TODO: remove in 0.22 assuming all pre 0.20 clusters upgraded their pre-v1 Corrosion. + corrosionDir string dockerCtrl *docker.Controller // dockerReady is signalled when Docker is configured and ready for containers. dockerReady chan<- struct{} @@ -60,6 +64,7 @@ func newClusterController( store *store.Store, server *grpc.Server, corroService corroservice.Service, + corrosionDir string, dockerService *docker.Service, dockerReady chan<- struct{}, clusterReady chan<- struct{}, @@ -82,6 +87,7 @@ func newClusterController( endpointChanges: endpointChanges, server: server, corroService: corroService, + corrosionDir: corrosionDir, dockerCtrl: docker.NewController(state.ID, dockerService, store), dockerReady: dockerReady, clusterReady: clusterReady, @@ -126,6 +132,11 @@ func (cc *clusterController) Run(ctx context.Context) error { slog.Info("Corrosion service started.") } + // Apply the seed to finish Corrosion migrations from 0.x to 2026.5.14 (upstream v1.0.0) if applicable. + if err := corromigrate.ApplySeedIfPresent(ctx, cc.corrosionDir, cc.store); err != nil { + return fmt.Errorf("apply corrosion migration seed: %w", err) + } + errGroup, ctx := errgroup.WithContext(ctx) // Start the WireGuard control loop before waiting for store sync. This ensures endpoint rotation happens diff --git a/internal/machine/corromigrate/migrate.go b/internal/machine/corromigrate/migrate.go new file mode 100644 index 00000000..6e6b2959 --- /dev/null +++ b/internal/machine/corromigrate/migrate.go @@ -0,0 +1,282 @@ +// Package corromigrate handles the one-time migration of the on-disk Corrosion store from +// systemd-managed v0.x to v1.0.0 running in a uncloudd-managed container. +// TODO: remove in 0.22 assuming all pre 0.20 clusters upgraded their pre-v1 Corrosion. +package corromigrate + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "os/exec" + "path/filepath" + "time" + + "github.com/psviderski/uncloud/internal/fs" + "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/internal/machine/store" + "google.golang.org/protobuf/encoding/protojson" + _ "modernc.org/sqlite" +) + +// legacyUnitName is the systemd unit that ran Corrosion before uncloudd took over its lifecycle. +// The unit file at /etc/systemd/system/uncloud-corrosion.service and the binary at +// /usr/local/bin/uncloud-corrosion are left in place after migration because uncloudd's sandbox +// (ProtectSystem=full) blocks their removal. Operators should delete them manually as documented +// in the release notes. +const legacyUnitName = "uncloud-corrosion.service" + +// Seed is the on-disk representation of the durable rows dumped from a v0.x Corrosion store, +// to be re-applied to the fresh v1.0.0 store. Existence of the seed file is the signal that +// migration has not yet fully completed. +type Seed struct { + Cluster []ClusterEntry `json:"cluster"` + Machines []MachineEntry `json:"machines"` +} + +type ClusterEntry struct { + Key string `json:"key"` + Value string `json:"value"` +} + +type MachineEntry struct { + ID string `json:"id"` + Info string `json:"info"` +} + +func seedPath(dir string) string { return dir + ".seed-v1.json" } + +func backupPath(dir string) string { + return dir + ".backup-" + time.Now().UTC().Format("20060102.150405") +} + +// MigrateIfNeeded stops the legacy systemd Corrosion unit and, when a v0.x store.db is detected, dumps its durable +// rows to a seed file next to the data dir, then backs up the dir so the new version can start fresh. +// Idempotent: if the seed file already exists, the dump step is skipped (a prior run already produced it). +func MigrateIfNeeded(ctx context.Context, dir, owner string) error { + stopLegacyUnit() + + seedFile := seedPath(dir) + if _, err := os.Stat(seedFile); err == nil { + slog.Info("Corrosion migrations seed file found, dump already produced by a prior run.", + "path", seedFile) + return nil + } else if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("stat seed file '%s': %w", seedFile, err) + } + + dbFile := filepath.Join(dir, "store.db") + if _, err := os.Stat(dbFile); errors.Is(err, os.ErrNotExist) { + return nil + } else if err != nil { + return fmt.Errorf("stat '%s': %w", dbFile, err) + } + + isV0, err := isV0Store(ctx, dbFile) + if err != nil { + return fmt.Errorf("detect old store version: %w", err) + } + if !isV0 { + return nil + } + + slog.Info("Migrating Corrosion store from v0.x to v1.0.0.", "db", dbFile) + + dump, err := dumpOldStore(ctx, dbFile) + if err != nil { + return fmt.Errorf("dump old store: %w", err) + } + slog.Info("Corrosion store dumped.", + "cluster_rows", len(dump.Cluster), "machine_rows", len(dump.Machines)) + + if err = writeSeedAtomic(seedFile, dump, owner); err != nil { + return fmt.Errorf("write seed file: %w", err) + } + slog.Info("Corrosion migration seed file written.", "path", seedFile) + + backup := backupPath(dir) + if err = os.Rename(dir, backup); err != nil { + return fmt.Errorf("backup corrosion dir: %w", err) + } + slog.Info("Old Corrosion dir backed up.", "from", dir, "to", backup) + + return nil +} + +// stopLegacyUnit stops and disables the legacy uncloud-corrosion systemd unit if present. +// Best-effort: errors are swallowed because the unit may not exist (fresh install) or systemctl +// may be unavailable (containerised hosts). +func stopLegacyUnit() { + for _, cmd := range []string{"stop", "disable"} { + if err := exec.Command("systemctl", cmd, legacyUnitName).Run(); err != nil { + slog.Debug("systemctl on legacy corrosion unit failed (likely absent).", + "cmd", cmd, "unit", legacyUnitName, "err", err) + } + } +} + +// isV0Store reports whether the SQLite database at dbPath is in the old v0.x Corrosion format. +// v0.x has a __corro_bookkeeping table that was dropped in v1.0.0. +func isV0Store(ctx context.Context, dbPath string) (bool, error) { + db, err := sql.Open("sqlite", "file:"+dbPath+"?mode=ro") + if err != nil { + return false, fmt.Errorf("open db '%s': %w", dbPath, err) + } + defer db.Close() + + var one int + err = db.QueryRowContext(ctx, + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='__corro_bookkeeping' LIMIT 1", + ).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + return true, nil +} + +func dumpOldStore(ctx context.Context, dbPath string) (*Seed, error) { + db, err := sql.Open("sqlite", "file:"+dbPath+"?mode=ro") + if err != nil { + return nil, fmt.Errorf("open %q: %w", dbPath, err) + } + defer db.Close() + + seed := &Seed{} + + clusterRows, err := db.QueryContext(ctx, "SELECT key, value FROM cluster") + if err != nil { + return nil, fmt.Errorf("query cluster: %w", err) + } + for clusterRows.Next() { + var key, value string + if err = clusterRows.Scan(&key, &value); err != nil { + clusterRows.Close() + return nil, fmt.Errorf("scan cluster row: %w", err) + } + seed.Cluster = append(seed.Cluster, ClusterEntry{Key: key, Value: value}) + } + if err = clusterRows.Err(); err != nil { + clusterRows.Close() + return nil, fmt.Errorf("iterate cluster rows: %w", err) + } + clusterRows.Close() + + machineRows, err := db.QueryContext(ctx, "SELECT id, info FROM machines") + if err != nil { + return nil, fmt.Errorf("query machines: %w", err) + } + defer machineRows.Close() + for machineRows.Next() { + var id, info string + if err = machineRows.Scan(&id, &info); err != nil { + return nil, fmt.Errorf("scan machine row: %w", err) + } + seed.Machines = append(seed.Machines, MachineEntry{ID: id, Info: info}) + } + if err = machineRows.Err(); err != nil { + return nil, fmt.Errorf("iterate machine rows: %w", err) + } + return seed, nil +} + +func writeSeedAtomic(path string, seed *Seed, owner string) error { + tmp := path + ".tmp" + data, err := json.MarshalIndent(seed, "", " ") + if err != nil { + return err + } + + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) + if err != nil { + return err + } + if _, err = f.Write(data); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("write seed: %w", err) + } + if err = f.Sync(); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("fsync seed: %w", err) + } + if err = f.Close(); err != nil { + os.Remove(tmp) + return fmt.Errorf("close seed: %w", err) + } + + if owner != "" { + if chErr := fs.Chown(tmp, owner, owner); chErr != nil { + os.Remove(tmp) + return chErr + } + } + if err = os.Rename(tmp, path); err != nil { + return fmt.Errorf("rename seed: %w", err) + } + + // Fsync the parent directory so the rename entry is durable. + dir, err := os.Open(filepath.Dir(path)) + if err != nil { + return fmt.Errorf("open parent dir for fsync: %w", err) + } + defer dir.Close() + if err = dir.Sync(); err != nil { + return fmt.Errorf("fsync parent dir: %w", err) + } + return nil +} + +// ApplySeedIfPresent re-applies the durable rows from the seed file into the running Corrosion store, +// then deletes the seed file as the final completion marker. Idempotent: re-running on the same seed +// is a no-op for cluster rows (INSERT OR REPLACE) and machines rows (skipped via GetMachine). +func ApplySeedIfPresent(ctx context.Context, dir string, st *store.Store) error { + seedFile := seedPath(dir) + data, err := os.ReadFile(seedFile) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("read seed file: %w", err) + } + + var seed Seed + if err = json.Unmarshal(data, &seed); err != nil { + return fmt.Errorf("parse seed file: %w", err) + } + slog.Info("Applying Corrosion migration seed.", + "cluster_rows", len(seed.Cluster), "machine_rows", len(seed.Machines)) + + for _, e := range seed.Cluster { + if err = st.Put(ctx, e.Key, e.Value); err != nil { + return fmt.Errorf("seed cluster row '%s': %w", e.Key, err) + } + } + + for _, e := range seed.Machines { + var m pb.MachineInfo + if err = protojson.Unmarshal([]byte(e.Info), &m); err != nil { + return fmt.Errorf("parse machine '%s' info: %w", e.ID, err) + } + if _, err = st.GetMachine(ctx, m.Id); err == nil { + continue + } else if !errors.Is(err, store.ErrMachineNotFound) { + return fmt.Errorf("check machine '%s': %w", m.Id, err) + } + if err = st.CreateMachine(ctx, &m); err != nil { + return fmt.Errorf("seed machine '%s': %w", m.Id, err) + } + } + + if err = os.Remove(seedFile); err != nil { + return fmt.Errorf("delete seed file: %w", err) + } + slog.Info("Corrosion migration completed.") + return nil +} diff --git a/internal/machine/corromigrate/migrate_test.go b/internal/machine/corromigrate/migrate_test.go new file mode 100644 index 00000000..e58dd6c0 --- /dev/null +++ b/internal/machine/corromigrate/migrate_test.go @@ -0,0 +1,233 @@ +package corromigrate + +import ( + "context" + "database/sql" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + _ "modernc.org/sqlite" +) + +func TestIsV0Store(t *testing.T) { + tests := []struct { + name string + setup func(*testing.T, *sql.DB) + expectV0 bool + }{ + { + name: "v0.x store has __corro_bookkeeping table", + setup: func(t *testing.T, db *sql.DB) { + _, err := db.Exec("CREATE TABLE __corro_bookkeeping (actor_id BLOB, version INTEGER)") + require.NoError(t, err) + }, + expectV0: true, + }, + { + name: "v1.0.0 store has only __corro_bookkeeping_gaps", + setup: func(t *testing.T, db *sql.DB) { + _, err := db.Exec("CREATE TABLE __corro_bookkeeping_gaps (actor_id BLOB, start INTEGER, end INTEGER)") + require.NoError(t, err) + }, + expectV0: false, + }, + { + name: "empty store has neither table", + setup: func(t *testing.T, db *sql.DB) { + // Force the SQLite file to materialize on disk so it can be reopened read-only. + _, err := db.Exec("PRAGMA user_version = 0") + require.NoError(t, err) + }, + expectV0: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "store.db") + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + tt.setup(t, db) + require.NoError(t, db.Close()) + + got, err := isV0Store(context.Background(), dbPath) + require.NoError(t, err) + assert.Equal(t, tt.expectV0, got) + }) + } +} + +func TestDumpOldStore(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "store.db") + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + defer db.Close() + + _, err = db.Exec(`CREATE TABLE cluster (key TEXT PRIMARY KEY, value ANY)`) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE machines (id TEXT PRIMARY KEY, info TEXT)`) + require.NoError(t, err) + + _, err = db.Exec(`INSERT INTO cluster (key, value) VALUES (?, ?)`, + "network", "10.210.0.0/16") + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO cluster (key, value) VALUES (?, ?)`, + "created_at", "2026-05-22T10:00:00Z") + require.NoError(t, err) + + _, err = db.Exec(`INSERT INTO machines (id, info) VALUES (?, ?)`, + "m1", `{"id":"m1","name":"alpha"}`) + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO machines (id, info) VALUES (?, ?)`, + "m2", `{"id":"m2","name":"beta"}`) + require.NoError(t, err) + + require.NoError(t, db.Close()) + + seed, err := dumpOldStore(context.Background(), dbPath) + require.NoError(t, err) + + assert.ElementsMatch(t, []ClusterEntry{ + {Key: "network", Value: "10.210.0.0/16"}, + {Key: "created_at", Value: "2026-05-22T10:00:00Z"}, + }, seed.Cluster) + + assert.ElementsMatch(t, []MachineEntry{ + {ID: "m1", Info: `{"id":"m1","name":"alpha"}`}, + {ID: "m2", Info: `{"id":"m2","name":"beta"}`}, + }, seed.Machines) +} + +func TestMigrateIfNeeded(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T, dir string) + wantSeed bool + wantBackup bool + }{ + { + name: "greenfield (no store.db) is a no-op", + setup: func(t *testing.T, dir string) { require.NoError(t, os.MkdirAll(dir, 0o700)) }, + wantSeed: false, + wantBackup: false, + }, + { + name: "v1.0.0 store (no __corro_bookkeeping) is a no-op", + setup: func(t *testing.T, dir string) { + require.NoError(t, os.MkdirAll(dir, 0o700)) + dbPath := filepath.Join(dir, "store.db") + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE __corro_bookkeeping_gaps (actor_id BLOB)`) + require.NoError(t, err) + require.NoError(t, db.Close()) + }, + wantSeed: false, + wantBackup: false, + }, + { + name: "v0.x store triggers dump, seed, and backup", + setup: func(t *testing.T, dir string) { + require.NoError(t, os.MkdirAll(dir, 0o700)) + dbPath := filepath.Join(dir, "store.db") + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE __corro_bookkeeping (actor_id BLOB, version INTEGER)`) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE cluster (key TEXT PRIMARY KEY, value ANY)`) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE machines (id TEXT PRIMARY KEY, info TEXT)`) + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO cluster VALUES (?, ?)`, "network", "10.210.0.0/16") + require.NoError(t, err) + _, err = db.Exec(`INSERT INTO machines VALUES (?, ?)`, "m1", `{"id":"m1"}`) + require.NoError(t, err) + require.NoError(t, db.Close()) + }, + wantSeed: true, + wantBackup: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + root := t.TempDir() + dir := filepath.Join(root, "corrosion") + tt.setup(t, dir) + + err := MigrateIfNeeded(context.Background(), dir, "") + require.NoError(t, err) + + seedPath := dir + ".seed-v1.json" + _, seedErr := os.Stat(seedPath) + if tt.wantSeed { + require.NoError(t, seedErr, "seed file expected") + + data, err := os.ReadFile(seedPath) + require.NoError(t, err) + var seed Seed + require.NoError(t, json.Unmarshal(data, &seed)) + assert.Len(t, seed.Cluster, 1) + assert.Len(t, seed.Machines, 1) + } else { + assert.True(t, os.IsNotExist(seedErr), "seed file unexpected") + } + + entries, err := os.ReadDir(root) + require.NoError(t, err) + var backups int + for _, e := range entries { + if e.Name() != "corrosion" && len(e.Name()) > len("corrosion.backup-") && + e.Name()[:len("corrosion.backup-")] == "corrosion.backup-" { + backups++ + } + } + if tt.wantBackup { + assert.Equal(t, 1, backups, "expected one backup dir") + _, err := os.Stat(filepath.Join(dir, "store.db")) + assert.True(t, os.IsNotExist(err), "store.db should be gone from recreated dir") + } else { + assert.Equal(t, 0, backups, "no backup expected") + } + }) + } +} + +func TestMigrateIfNeeded_SkipsDumpWhenSeedAlreadyExists(t *testing.T) { + root := t.TempDir() + dir := filepath.Join(root, "corrosion") + require.NoError(t, os.MkdirAll(dir, 0o700)) + + dbPath := filepath.Join(dir, "store.db") + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE __corro_bookkeeping (x INTEGER)`) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE cluster (key TEXT, value ANY)`) + require.NoError(t, err) + _, err = db.Exec(`CREATE TABLE machines (id TEXT, info TEXT)`) + require.NoError(t, err) + require.NoError(t, db.Close()) + + preseed := Seed{Cluster: []ClusterEntry{{Key: "preseed", Value: "v"}}} + data, err := json.Marshal(&preseed) + require.NoError(t, err) + require.NoError(t, os.WriteFile(dir+".seed-v1.json", data, 0o600)) + + require.NoError(t, MigrateIfNeeded(context.Background(), dir, "")) + + // Existing seed must be left untouched (no dump performed, no backup created). + got, err := os.ReadFile(dir + ".seed-v1.json") + require.NoError(t, err) + var afterSeed Seed + require.NoError(t, json.Unmarshal(got, &afterSeed)) + assert.Equal(t, preseed, afterSeed) + + // The corrosion dir should still be present (not backed up). + _, err = os.Stat(filepath.Join(dir, "store.db")) + assert.NoError(t, err) +} diff --git a/internal/machine/corroservice/docker.go b/internal/machine/corroservice/docker.go index 704cebaf..27259587 100644 --- a/internal/machine/corroservice/docker.go +++ b/internal/machine/corroservice/docker.go @@ -14,9 +14,10 @@ import ( "github.com/docker/docker/api/types/mount" "github.com/docker/docker/api/types/network" "github.com/docker/docker/client" + "github.com/psviderski/uncloud/pkg/api" ) -// Image is the Corrosion image pinned to the uncloudd release. +// Image is the Corrosion image pinned to the uncloudd version. const Image = "ghcr.io/unlabs-dev/corrosion:2026.5.14" type DockerService struct { @@ -28,23 +29,36 @@ type DockerService struct { } func (s *DockerService) Start(ctx context.Context) error { - _, err := s.Client.ContainerInspect(ctx, s.Name) - if err != nil { - if !errdefs.IsNotFound(err) { - return fmt.Errorf("inspect container %q: %w", s.Name, err) - } - if err = s.startNewContainer(ctx); err != nil { + c, err := s.Client.ContainerInspect(ctx, s.Name) + switch { + case errdefs.IsNotFound(err): + if err = s.createAndStart(ctx); err != nil { return err } - } else { - // Container already exists. - // TODO: recreate only if the container configuration has to be changed. - if err = s.Client.ContainerRemove(ctx, s.Name, container.RemoveOptions{Force: true}); err != nil { - return fmt.Errorf("remove container %q: %w", s.Name, err) + case err != nil: + return fmt.Errorf("inspect container '%s': %w", s.Name, err) + case c.Config.Image != s.Image: + slog.Info("Corrosion container image needs update, recreating container.", + "name", s.Name, "current_image", c.Config.Image, "new_image", s.Image) + + // Gracefully stop the container before removing it. + if err = s.Client.ContainerStop(ctx, s.Name, container.StopOptions{}); err != nil && !errdefs.IsNotFound(err) { + return fmt.Errorf("stop container '%s': %w", s.Name, err) } - if err = s.startNewContainer(ctx); err != nil { + if err = s.Client.ContainerRemove(ctx, s.Name, container.RemoveOptions{ + // Remove anonymous volumes created by the container. + RemoveVolumes: true, + }); err != nil && !errdefs.IsNotFound(err) { + return fmt.Errorf("remove container '%s': %w", s.Name, err) + } + if err = s.createAndStart(ctx); err != nil { return err } + case !c.State.Running: + slog.Debug("Starting existing Corrosion container.", "name", s.Name) + if err = s.Client.ContainerStart(ctx, s.Name, container.StartOptions{}); err != nil { + return fmt.Errorf("start container '%s': %w", s.Name, err) + } } slog.Debug("Waiting for corrosion service to be ready.") @@ -52,27 +66,25 @@ func (s *DockerService) Start(ctx context.Context) error { return err } slog.Debug("Corrosion service is ready.") - return nil } +// Stop stops the Corrosion container without removing it. The container is kept so that +// the next Start can start it instead of pulling and recreating. func (s *DockerService) Stop(ctx context.Context) error { if err := s.Client.ContainerStop(ctx, s.Name, container.StopOptions{}); err != nil { - return fmt.Errorf("stop container %q: %w", s.Name, err) + if errdefs.IsNotFound(err) { + return nil + } + return fmt.Errorf("stop container '%s': %w", s.Name, err) } - slog.Debug("Corrosion Docker container stopped.", "name", s.Name) - - if err := s.Client.ContainerRemove(ctx, s.Name, container.RemoveOptions{}); err != nil { - return fmt.Errorf("remove container %q: %w", s.Name, err) - } - slog.Debug("Corrosion Docker container removed.", "name", s.Name) - + slog.Debug("Corrosion container stopped.", "name", s.Name) return nil } func (s *DockerService) Restart(ctx context.Context) error { if err := s.Client.ContainerRestart(ctx, s.Name, container.StopOptions{}); err != nil { - return fmt.Errorf("restart container %q: %w", s.Name, err) + return fmt.Errorf("restart container '%s': %w", s.Name, err) } return nil } @@ -82,7 +94,6 @@ func (s *DockerService) Running() bool { if err != nil { return false } - return c.State.Running } @@ -91,14 +102,18 @@ func (s *DockerService) containerConfig() *container.Config { Image: s.Image, Cmd: []string{"corrosion", "agent", "-c", filepath.Join(s.DataDir, "config.toml")}, User: s.User, + Labels: map[string]string{ + api.LabelDaemonManaged: "", + }, } } func (s *DockerService) hostConfig() *container.HostConfig { return &container.HostConfig{ NetworkMode: network.NetworkHost, + // Use unless-stopped so uncloudd-initiated stops are honoured. RestartPolicy: container.RestartPolicy{ - Name: container.RestartPolicyAlways, + Name: container.RestartPolicyUnlessStopped, }, Mounts: []mount.Mount{ // Bind mount the data directory at the same path inside the container to simplify path handling. @@ -111,7 +126,7 @@ func (s *DockerService) hostConfig() *container.HostConfig { } } -func (s *DockerService) startNewContainer(ctx context.Context) error { +func (s *DockerService) createAndStart(ctx context.Context) error { _, err := s.Client.ContainerCreate(ctx, s.containerConfig(), s.hostConfig(), nil, nil, s.Name) if err != nil { if !errdefs.IsNotFound(err) { @@ -131,7 +146,6 @@ func (s *DockerService) startNewContainer(ctx context.Context) error { if _, err := io.Copy(io.Discard, respBody); err != nil { return fmt.Errorf("read pull response: %w", err) } - slog.Info("Docker image pulled.", "image", s.Image, "duration", time.Since(start).String()) // Create container again after image pull. @@ -143,6 +157,5 @@ func (s *DockerService) startNewContainer(ctx context.Context) error { if err = s.Client.ContainerStart(ctx, s.Name, container.StartOptions{}); err != nil { return fmt.Errorf("start container: %w", err) } - return nil } diff --git a/internal/machine/corroservice/systemd.go b/internal/machine/corroservice/systemd.go deleted file mode 100644 index e2dfc5ae..00000000 --- a/internal/machine/corroservice/systemd.go +++ /dev/null @@ -1,60 +0,0 @@ -package corroservice - -import ( - "context" - "fmt" - "log/slog" - "os/exec" -) - -const DefaultSystemdUnit = "uncloud-corrosion.service" - -type SystemdService struct { - DataDir string - Unit string - running bool -} - -func DefaultSystemdService(dataDir string) *SystemdService { - return &SystemdService{ - DataDir: dataDir, - Unit: DefaultSystemdUnit, - } -} - -func (s *SystemdService) Start(ctx context.Context) error { - return s.startOrRestart(ctx, "start") -} - -func (s *SystemdService) Stop(_ context.Context) error { - if _, err := exec.Command("systemctl", "stop", s.Unit).Output(); err != nil { - return fmt.Errorf("systemctl stop %s: %w", s.Unit, err) - } - slog.Info("Corrosion systemd service stopped.", "unit", s.Unit) - - return nil -} - -func (s *SystemdService) Restart(ctx context.Context) error { - return s.startOrRestart(ctx, "restart") -} - -func (s *SystemdService) startOrRestart(ctx context.Context, cmd string) error { - if _, err := exec.Command("systemctl", cmd, s.Unit).Output(); err != nil { - return fmt.Errorf("systemctl %s %s: %w", cmd, s.Unit, err) - } - slog.Debug(fmt.Sprintf("Corrosion systemd service %sed.", cmd), "unit", s.Unit) - - slog.Debug("Waiting for corrosion service to be ready.") - if err := WaitReady(ctx, s.DataDir); err != nil { - return err - } - slog.Debug("Corrosion service is ready.") - s.running = true - - return nil -} - -func (s *SystemdService) Running() bool { - return s.running -} diff --git a/internal/machine/machine.go b/internal/machine/machine.go index f8a74504..684e2841 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -29,6 +29,7 @@ import ( "github.com/psviderski/uncloud/internal/machine/caddyconfig" "github.com/psviderski/uncloud/internal/machine/cluster" "github.com/psviderski/uncloud/internal/machine/constants" + "github.com/psviderski/uncloud/internal/machine/corromigrate" "github.com/psviderski/uncloud/internal/machine/corroservice" "github.com/psviderski/uncloud/internal/machine/dns" machinedocker "github.com/psviderski/uncloud/internal/machine/docker" @@ -122,22 +123,16 @@ func (c *Config) SetDefaults() (*Config, error) { cfg.CorrosionUser = corroservice.DefaultUser } if cfg.CorrosionService == nil { - if isRunningInDocker() { - // Run corrosion in a nested Docker container if the machine is running in a container. - uid, gid, err := fs.LookupUIDGID(cfg.CorrosionUser) - if err != nil { - return nil, fmt.Errorf("lookup corrosion user %q: %w", cfg.CorrosionUser, err) - } - - cfg.CorrosionService = &corroservice.DockerService{ - Client: cfg.DockerClient, - Image: corroservice.Image, - Name: "uncloud-corrosion", - DataDir: cfg.CorrosionDir, - User: fmt.Sprintf("%d:%d", uid, gid), - } - } else { - cfg.CorrosionService = corroservice.DefaultSystemdService(cfg.CorrosionDir) + uid, gid, err := fs.LookupUIDGID(cfg.CorrosionUser) + if err != nil { + return nil, fmt.Errorf("lookup corrosion user %q: %w", cfg.CorrosionUser, err) + } + cfg.CorrosionService = &corroservice.DockerService{ + Client: cfg.DockerClient, + Image: corroservice.Image, + Name: "uncloud-corrosion", + DataDir: cfg.CorrosionDir, + User: fmt.Sprintf("%d:%d", uid, gid), } } @@ -148,12 +143,6 @@ func (c *Config) SetDefaults() (*Config, error) { return &cfg, nil } -// isRunningInDocker returns true if the current process is running in a Docker container. -func isRunningInDocker() bool { - _, err := os.Stat("/.dockerenv") - return err == nil -} - type Machine struct { pb.UnimplementedMachineServer @@ -376,6 +365,13 @@ func (m *Machine) Run(ctx context.Context) error { return fmt.Errorf("start corrosion service: %w", err) } slog.Info("Corrosion service started.") + } else { + // Migrate the on-disk Corrosion store to 2026.5.14 (v1.0.0 upstream) if a v0.x store.db is detected, + // before any Corrosion start attempt. The legacy systemd unit (if installed) is stopped here too + // so we own the data dir exclusively. + if err := corromigrate.MigrateIfNeeded(ctx, m.config.CorrosionDir, m.config.CorrosionUser); err != nil { + return fmt.Errorf("migrate corrosion store: %w", err) + } } // Use an errgroup to coordinate error handling and graceful shutdown of multiple machine components. @@ -497,6 +493,7 @@ func (m *Machine) Run(ctx context.Context) error { m.store, proxyServer, m.config.CorrosionService, + m.config.CorrosionDir, m.dockerService, m.networkReady, m.clusterReady, @@ -539,6 +536,15 @@ func (m *Machine) Run(ctx context.Context) error { m.proxyDirector.Close() slog.Info("Local API proxy server stopped.") + // Stop the corrosion container so this node stops gossiping its membership as "Up" while the + // gRPC API is gone. Use a fresh context because ctx is already cancelled here. + slog.Info("Stopping corrosion service.") + stopCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + if stopErr := m.config.CorrosionService.Stop(stopCtx); stopErr != nil { + slog.Error("Failed to stop corrosion service.", "err", stopErr) + } + cancel() + // Clean up the machine data and resources if the machine shutdown was initiated by a reset. if m.resetting { slog.Info("Cleaning up machine data and resources.") diff --git a/pkg/api/container.go b/pkg/api/container.go index b8cc136d..7fb83f09 100644 --- a/pkg/api/container.go +++ b/pkg/api/container.go @@ -19,6 +19,10 @@ const ( // in internal/machine/docker/manager.go. DockerNetworkName = "uncloud" + // LabelDaemonManaged marks system containers as managed by the Uncloud daemon. + LabelDaemonManaged = "uncloudd.managed" + + // LabelManaged marks a service container as managed by Uncloud. LabelManaged = "uncloud.managed" LabelServiceID = "uncloud.service.id" LabelServiceName = "uncloud.service.name" diff --git a/scripts/install.sh b/scripts/install.sh index 260de035..b01f42d1 100755 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -18,9 +18,6 @@ UNCLOUD_USER="uncloud" UNCLOUD_GROUP_ADD_USER=${UNCLOUD_GROUP_ADD_USER:-} UNCLOUD_DATA_DIR=${UNCLOUD_DATA_DIR:-/var/lib/uncloud} -CORROSION_GITHUB_URL="https://github.com/psviderski/corrosion" -CORROSION_VERSION=${CORROSION_VERSION:-v0.2.2} - DOCKER_ALREADY_INSTALLED=false CONTAINERD_IMAGE_STORE_ENABLED=false DOCKER_DAEMON_CONFIG_FILE=${DOCKER_DAEMON_CONFIG_FILE:-/etc/docker/daemon.json} @@ -284,76 +281,6 @@ EOF systemctl enable uncloud.service } -install_corrosion() { - local arch - arch=$(uname -m) - - local corrosion_install_path="${INSTALL_BIN_DIR}/uncloud-corrosion" - if [ -f "${corrosion_install_path}" ]; then - # TODO: Check the version of the installed corrosion binary and update if there is a newer stable version. - log "✓ uncloud-corrosion binary is already installed." - return - fi - - # Create a temporary directory for downloads. - local tmp_dir - tmp_dir=$(mktemp -d) - # Ensure the temporary directory is deleted on script exit. - # shellcheck disable=SC2064 - trap "rm -rf '$tmp_dir'" EXIT - - local corrosion_url - if [ "${CORROSION_VERSION}" == "latest" ]; then - corrosion_url="${CORROSION_GITHUB_URL}/releases/latest/download/corrosion-${arch}-unknown-linux-gnu.tar.gz" - else - corrosion_url="${CORROSION_GITHUB_URL}/releases/download/${CORROSION_VERSION}/corrosion-${arch}-unknown-linux-gnu.tar.gz" - fi - local corrosion_download_path="${tmp_dir}/corrosion.tar.gz" - - log "⏳ Downloading uncloud-corrosion binary: ${corrosion_url}" - if ! curl -fsSL -o "${corrosion_download_path}" "${corrosion_url}"; then - error "Failed to download uncloud-corrosion binary." - fi - tar -xf "${corrosion_download_path}" -C "${tmp_dir}" - if ! install "${tmp_dir}/corrosion" "${corrosion_install_path}"; then - error "Failed to install uncloud-corrosion binary to ${corrosion_install_path}" - fi - log "✓ uncloud-corrosion binary installed: ${corrosion_install_path}" -} - -install_corrosion_systemd() { - local corrosion_service_path="${INSTALL_SYSTEMD_DIR}/uncloud-corrosion.service" - cat > "${corrosion_service_path}" << EOF -[Unit] -Description=Uncloud gossip-based distributed store -PartOf=uncloud.service - -[Service] -Type=simple -ExecStart=${INSTALL_BIN_DIR}/uncloud-corrosion agent -c ${UNCLOUD_DATA_DIR}/corrosion/config.toml -ExecReload=${INSTALL_BIN_DIR}/uncloud-corrosion reload -c ${UNCLOUD_DATA_DIR}/corrosion/config.toml -Restart=always -RestartSec=2 -User=${UNCLOUD_USER} -Group=${UNCLOUD_USER} - -# Hardening options. -ProtectSystem=full -PrivateTmp=true -NoNewPrivileges=true -ProtectHome=true -ProtectControlGroups=true -ProtectKernelTunables=true -RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX -EOF - log "✓ Systemd unit file created: ${corrosion_service_path}" - - if [[ "${INSTALL_ONLY}" != "true" ]]; then - # Reload systemd to recognize the new unit file - systemctl daemon-reload - fi -} - start_uncloud() { if [[ "${INSTALL_ONLY}" == "true" ]]; then return @@ -375,8 +302,6 @@ install_docker create_uncloud_user_and_group install_uncloud_binaries install_uncloud_systemd -install_corrosion -install_corrosion_systemd start_uncloud # Show warning if Docker was already installed without containerd image store enabled.