Merge branch '310-recreate-clones' into 'master'

feat: keep clone containers after VM reboot (#310)

Closes #310

See merge request postgres-ai/database-lab!389

Author: agneum

cmd/database-lab/main.go

@@ -127,7 +127,7 @@ func main() {
 	defer retrievalSvc.Stop()
 
 	// Create a cloning service to provision new clones.
-	provisionSvc, err := provision.New(ctx, &cfg.Provision, dbCfg, dockerCLI, pm, internalNetworkID)
+	provisionSvc, err := provision.New(ctx, &cfg.Provision, dbCfg, dockerCLI, pm, engProps.InstanceID, internalNetworkID)
 	if err != nil {
 		log.Errf(errors.WithMessage(err, `error in the "provision" section of the config`).Error())
 	}

internal/cloning/base.go

@@ -105,6 +105,8 @@ func (c *Base) Run(ctx context.Context) error {
 		log.Err("Failed to load stored sessions:", err)
 	}
 
+	c.restartCloneContainers(ctx)
+
 	c.filterRunningClones(ctx)
 
 	if err := c.cleanupInvalidClones(); err != nil {

internal/cloning/storage.go

@@ -46,6 +46,31 @@ func (c *Base) loadSessionState(sessionsPath string) error {
 
 	return json.Unmarshal(data, &c.clones)
 }
+func (c *Base) restartCloneContainers(ctx context.Context) {
+	c.cloneMutex.Lock()
+	defer c.cloneMutex.Unlock()
+
+	for _, wrapper := range c.clones {
+		if wrapper.Clone == nil || wrapper.Session == nil {
+			continue
+		}
+
+		cloneName := util.GetCloneName(wrapper.Session.Port)
+		if c.provision.IsCloneRunning(ctx, cloneName) {
+			continue
+		}
+
+		if err := c.provision.ReconnectClone(ctx, cloneName); err != nil {
+			log.Err(fmt.Sprintf("Clone container %s cannot be reconnected to the internal network: %s", cloneName, err))
+		}
+
+		if err := c.provision.StartCloneContainer(ctx, cloneName); err != nil {
+			log.Err(fmt.Sprintf("Clone container %s cannot start: %s", cloneName, err))
+		}
+
+		log.Dbg(fmt.Sprintf("Clone container %s is running", cloneName))
+	}
+}
 
 func (c *Base) filterRunningClones(ctx context.Context) {
 	c.cloneMutex.Lock()

internal/cloning/storage_test.go

@@ -83,7 +83,7 @@ func newProvisioner() (*provision.Provisioner, error) {
 			From: 1,
 			To:   5,
 		},
-	}, nil, nil, nil, "nwID")
+	}, nil, nil, nil, "instID", "nwID")
 }
 
 func TestLoadingSessionState(t *testing.T) {

internal/provision/docker/docker.go

@@ -261,8 +261,8 @@ func PullImage(r runners.Runner, dockerImage string) error {
 	return nil
 }
 
-// IsContainerExist checks the existence of Docker container.
-func IsContainerExist(ctx context.Context, docker *client.Client, containerName string) (bool, error) {
+// IsContainerRunning checks if specified container is running.
+func IsContainerRunning(ctx context.Context, docker *client.Client, containerName string) (bool, error) {
 	inspection, err := docker.ContainerInspect(ctx, containerName)
 	if err != nil {
 		return false, fmt.Errorf("failed to inpect container: %w", err)

internal/provision/mode_local.go

@@ -18,6 +18,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/client"
 	"github.com/pkg/errors"
 
@@ -31,6 +32,7 @@ import (
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/log"
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/models"
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/util"
+	"gitlab.com/postgres-ai/database-lab/v3/pkg/util/networks"
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/util/pglog"
 )
 
@@ -67,11 +69,12 @@ type Provisioner struct {
 	portChecker    portChecker
 	pm             *pool.Manager
 	networkID      string
+	instanceID     string
 }
 
 // New creates a new Provisioner instance.
 func New(ctx context.Context, cfg *Config, dbCfg *resources.DB, docker *client.Client, pm *pool.Manager,
-	networkID string) (*Provisioner, error) {
+	instanceID, networkID string) (*Provisioner, error) {
 	if err := IsValidConfig(*cfg); err != nil {
 		return nil, errors.Wrap(err, "configuration is not valid")
 	}
@@ -86,6 +89,7 @@ func New(ctx context.Context, cfg *Config, dbCfg *resources.DB, docker *client.C
 		portChecker:  &localPortChecker{},
 		pm:           pm,
 		networkID:    networkID,
+		instanceID:   instanceID,
 		ports:        make([]bool, cfg.PortPool.To-cfg.PortPool.From),
 	}
 
@@ -699,14 +703,24 @@ func (p *Provisioner) prepareDB(pgConf *resources.AppConfig, user resources.Ephe
 
 // IsCloneRunning checks if clone is running.
 func (p *Provisioner) IsCloneRunning(ctx context.Context, cloneName string) bool {
-	isRunning, err := docker.IsContainerExist(ctx, p.dockerClient, cloneName)
+	isRunning, err := docker.IsContainerRunning(ctx, p.dockerClient, cloneName)
 	if err != nil {
 		log.Err(err)
 	}
 
 	return isRunning
 }
 
+// ReconnectClone disconnects clone from the old instance network and connect to the actual one.
+func (p *Provisioner) ReconnectClone(ctx context.Context, cloneName string) error {
+	return networks.Reconnect(ctx, p.dockerClient, p.instanceID, cloneName)
+}
+
+// StartCloneContainer starts clone container.
+func (p *Provisioner) StartCloneContainer(ctx context.Context, containerName string) error {
+	return p.dockerClient.ContainerStart(ctx, containerName, types.ContainerStartOptions{})
+}
+
 // DetectDBVersion detects version of the database.
 func (p *Provisioner) DetectDBVersion() string {
 	pgVersion, err := tools.DetectPGVersion(p.pm.First().Pool().DataDir())

internal/provision/mode_local_test.go

@@ -24,7 +24,7 @@ func TestPortAllocation(t *testing.T) {
 		},
 	}
 
-	p, err := New(context.Background(), cfg, &resources.DB{}, &client.Client{}, &pool.Manager{}, "networkID")
+	p, err := New(context.Background(), cfg, &resources.DB{}, &client.Client{}, &pool.Manager{}, "instanceID", "networkID")
 	require.NoError(t, err)
 
 	// Allocate a new port.

pkg/util/networks/networks.go

@@ -117,6 +117,36 @@ func Connect(ctx context.Context, dockerCLI *client.Client, instanceID, containe
 	return nil
 }
 
+// Reconnect connects a container to internal Docker network.
+func Reconnect(ctx context.Context, dockerCLI *client.Client, instanceID, containerID string) error {
+	log.Dbg(fmt.Sprintf("Reconnect container %s to internal network", containerID))
+
+	networkName := getNetworkName(instanceID)
+
+	log.Dbg("Discovering internal network:", networkName)
+
+	networkResource, err := dockerCLI.NetworkInspect(ctx, networkName, types.NetworkInspectOptions{})
+	if err != nil {
+		return fmt.Errorf("internal network not found: %w", err)
+	}
+
+	log.Dbg(fmt.Sprintf("Disconnecting container %s from internal network", containerID))
+
+	if err := dockerCLI.NetworkDisconnect(context.Background(), networkName, containerID, true); err != nil {
+		return err
+	}
+
+	log.Dbg(fmt.Sprintf("Container %s has been disconnected from internal network", containerID))
+
+	if err := dockerCLI.NetworkConnect(ctx, networkResource.ID, containerID, &network.EndpointSettings{}); err != nil {
+		return err
+	}
+
+	log.Dbg(fmt.Sprintf("Container %s has been connected to %s", instanceID, networkName))
+
+	return nil
+}
+
 func getNetworkName(instanceID string) string {
 	return networkPrefix + instanceID
 }