feat: reorganizing clone datasets (#586)

Author: agneum

README.md

@@ -63,7 +63,7 @@ Try it yourself right now:
 - Visit [Postgres.ai Console](https://console.postgres.ai/), set up your first organization and provision a DBLab Standard Edition (DBLab SE) to any cloud or on-prem
     - [Pricing](https://postgres.ai/pricing) (starting at $62/month)
     - [Doc: How to install DBLab SE](https://postgres.ai/docs/how-to-guides/administration/install-dle-from-postgres-ai)
-- Demo: https://demo.aws.postgres.ai (use the token `demo-token` to access)
+- Demo: https://demo.dblab.dev (use the token `demo-token` to access)
 - Looking for a free version? Install DBLab Community Edition by [following this tutorial](https://postgres.ai/docs/tutorials/database-lab-tutorial)
 
 ## How it works
@@ -200,7 +200,7 @@ Reach out to the Postgres.ai team if you want a trial or commercial license that
 
 Many thanks to our amazing contributors!
 
-<a href = "https://github.com/postgresml/pgcat/graphs/contributors">
+<a href = "https://github.com/postgres-ai/database-lab/graphs/contributors">
   <img src = "https://contrib.rocks/image?repo=postgres-ai/database-lab"/>
 </a>
 

engine/.gitlab-ci.yml

@@ -420,6 +420,8 @@ build-image-swagger-release:
   artifacts:
     paths:
       - engine/bin
+  before_script:
+    - bash engine/test/_cleanup.sh
   script:
     - bash engine/test/1.synthetic.sh
     - bash engine/test/2.logical_generic.sh
@@ -482,7 +484,7 @@ integration-test:
   stage: integration-test
   variables:
     # Instruct Testcontainers to use the daemon of DinD.
-    DOCKER_HOST: "tcp://docker:2375"
+#    DOCKER_HOST: "tcp://docker:2375"
     # Instruct Docker not to start over TLS.
     DOCKER_TLS_CERTDIR: ""
     # Improve performance with overlayfs.

engine/api/swagger-spec/dblab_openapi.yaml

@@ -186,6 +186,11 @@ paths:
         required: true
         schema:
           type: string
+      - name: branch
+        in: query
+        required: false
+        schema:
+          type: string
       responses:
         200:
           description: Returned a list of snapshots
@@ -1800,4 +1805,4 @@ components:
           type: string
           format: date-time
         message:
-          type: string
+          type: string

engine/api/swagger-spec/dblab_server_swagger.yaml

@@ -21,8 +21,8 @@ externalDocs:
   url: https://gitlab.com/postgres-ai/docs/tree/master/docs/database-lab
 
 servers:
-  - url: "https://demo.aws.postgres.ai/api"
-    description: "DBLab 3.x demo server; token: 'demo-token'"
+  - url: "https://demo.dblab.dev/api"
+    description: "DBLab demo server; token: 'demo-token'"
     x-examples:
       Verification-Token: "demo-token"
   - url: "{scheme}://{host}:{port}/{basePath}"

engine/internal/cloning/base.go

@@ -171,6 +171,10 @@ func (c *Base) CreateClone(cloneRequest *types.CloneCreateRequest) (*models.Clon
 		}
 	}
 
+	if cloneRequest.Branch == "" {
+		cloneRequest.Branch = snapshot.Branch
+	}
+
 	clone := &models.Clone{
 		ID:        cloneRequest.ID,
 		Snapshot:  snapshot,
@@ -386,7 +390,7 @@ func (c *Base) refreshCloneMetadata(w *CloneWrapper) {
 		return
 	}
 
-	sessionState, err := c.provision.GetSessionState(w.Session, w.Clone.ID)
+	sessionState, err := c.provision.GetSessionState(w.Session, w.Clone.Branch, w.Clone.ID)
 	if err != nil {
 		// Session not ready yet.
 		log.Err(fmt.Errorf("failed to get a session state: %w", err))
@@ -482,6 +486,18 @@ func (c *Base) ResetClone(cloneID string, resetOptions types.ResetCloneRequest)
 		return errors.Wrap(err, "failed to update clone status")
 	}
 
+	if c.hasDependentSnapshots(w) {
+		log.Warn("clone has dependent snapshots", cloneID)
+		c.cloneMutex.Lock()
+		w.Clone.Revision++
+		w.Clone.HasDependent = true
+		c.cloneMutex.Unlock()
+	} else {
+		c.cloneMutex.Lock()
+		w.Clone.HasDependent = false
+		c.cloneMutex.Unlock()
+	}
+
 	go func() {
 		var originalSnapshotID string
 
@@ -719,9 +735,9 @@ func (c *Base) isIdleClone(wrapper *CloneWrapper) (bool, error) {
 		return false, errors.New("failed to get clone session")
 	}
 
-	if _, err := c.provision.LastSessionActivity(session, wrapper.Clone.ID, minimumTime); err != nil {
+	if _, err := c.provision.LastSessionActivity(session, wrapper.Clone.Branch, wrapper.Clone.ID, minimumTime); err != nil {
 		if err == pglog.ErrNotFound {
-			log.Dbg(fmt.Sprintf("Not found recent activity for the session: %q. Clone name: %q",
+			log.Dbg(fmt.Sprintf("Not found recent activity for session: %q. Clone name: %q",
 				session.ID, wrapper.Clone.ID))
 
 			return hasNotQueryActivity(session)

engine/internal/cloning/snapshots.go

@@ -13,7 +13,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/branching"
 )
 
 // SnapshotBox contains instance snapshots.
@@ -204,7 +204,7 @@ func (c *Base) hasDependentSnapshots(w *CloneWrapper) bool {
 	c.snapshotBox.snapshotMutex.RLock()
 	defer c.snapshotBox.snapshotMutex.RUnlock()
 
-	poolName := util.GetPoolName(w.Clone.Snapshot.Pool, w.Clone.ID)
+	poolName := branching.CloneName(w.Clone.Snapshot.Pool, w.Clone.Branch, w.Clone.ID, w.Clone.Revision)
 
 	for name := range c.snapshotBox.items {
 		if strings.HasPrefix(name, poolName) {

engine/internal/observer/observer.go

@@ -80,7 +80,7 @@ func NewObserver(dockerClient *client.Client, cfg *Config, pm *pool.Manager) *Ob
 // GetCloneLog gets clone logs.
 // TODO (akartasov): Split log to chunks.
 func (o *Observer) GetCloneLog(ctx context.Context, obsClone *ObservingClone) ([]byte, error) {
-	fileSelector := pglog.NewSelector(obsClone.pool.ClonePath(obsClone.cloneID))
+	fileSelector := pglog.NewSelector(obsClone.pool.ClonePath(obsClone.branch, obsClone.cloneID))
 	fileSelector.SetMinimumTime(obsClone.session.StartedAt)
 
 	if err := fileSelector.DiscoverLogDir(); err != nil {
@@ -187,11 +187,12 @@ func (o *Observer) maskLogs(entry []string, maskedFieldIndexes []int) {
 }
 
 // AddObservingClone adds a new observing session to storage.
-func (o *Observer) AddObservingClone(cloneID string, port uint, session *ObservingClone) {
+func (o *Observer) AddObservingClone(cloneID, branch string, port uint, session *ObservingClone) {
 	o.sessionMu.Lock()
 	defer o.sessionMu.Unlock()
 	session.pool = o.pm.First().Pool()
 	session.cloneID = cloneID
+	session.branch = branch
 	session.port = port
 
 	o.storage[cloneID] = session

engine/internal/observer/observing_clone.go

@@ -43,6 +43,7 @@ var maskedFields = map[string]struct{}{
 type ObservingClone struct {
 	pool        *resources.Pool
 	cloneID     string
+	branch      string
 	port        uint
 	superUserDB *pgx.Conn
 
@@ -479,7 +480,7 @@ func (c *ObservingClone) currentArtifactsSessionPath() string {
 }
 
 func (c *ObservingClone) artifactsSessionPath(sessionID uint64) string {
-	return path.Join(c.pool.ObserverDir(c.cloneID), c.cloneID, strconv.FormatUint(sessionID, 10))
+	return path.Join(c.pool.ObserverDir(c.branch, c.cloneID), c.cloneID, strconv.FormatUint(sessionID, 10))
 }
 
 // CheckPerformanceRequirements checks monitoring data and returns an error if any of performance requires was not satisfied.

engine/internal/provision/databases/postgres/postgres.go

@@ -99,7 +99,7 @@ func Start(r runners.Runner, c *resources.AppConfig) error {
 
 				_, err = pgctlPromote(r, c)
 				if err != nil {
-					if runnerError := Stop(r, c.Pool, c.CloneName); runnerError != nil {
+					if runnerError := Stop(r, c.Pool, c.CloneName, strconv.FormatUint(uint64(c.Port), 10)); runnerError != nil {
 						log.Err(runnerError)
 					}
 
@@ -115,7 +115,7 @@ func Start(r runners.Runner, c *resources.AppConfig) error {
 		if cnt > waitPostgresTimeout {
 			collectDiagnostics(c)
 
-			if runnerErr := Stop(r, c.Pool, c.CloneName); runnerErr != nil {
+			if runnerErr := Stop(r, c.Pool, c.CloneName, strconv.FormatUint(uint64(c.Port), 10)); runnerErr != nil {
 				log.Err(runnerErr)
 			}
 
@@ -138,7 +138,7 @@ func collectDiagnostics(c *resources.AppConfig) {
 }
 
 // Stop stops Postgres instance.
-func Stop(r runners.Runner, p *resources.Pool, name string) error {
+func Stop(r runners.Runner, p *resources.Pool, name, port string) error {
 	log.Dbg("Stopping Postgres container...")
 
 	if _, err := docker.RemoveContainer(r, name); err != nil {
@@ -151,8 +151,8 @@ func Stop(r runners.Runner, p *resources.Pool, name string) error {
 		log.Msg("docker container was not found, ignore", err)
 	}
 
-	if _, err := r.Run("rm -rf " + p.SocketCloneDir(name) + "/*"); err != nil {
-		return errors.Wrap(err, "failed to clean unix socket directory")
+	if _, err := r.Run("rm -rf " + p.SocketCloneDir(name) + "/.*" + port); err != nil {
+		return errors.Wrap(err, "failed to clean Unix socket directory")
 	}
 
 	return nil

engine/internal/provision/databases/postgres/postgres_test.go

@@ -67,7 +67,7 @@ func TestRemoveContainers(t *testing.T) {
 					})).
 			Return("", nil)
 
-		err := Stop(runner, p, "test_clone")
+		err := Stop(runner, p, "test_clone", "6200")
 
 		assert.Equal(t, tc.err, errors.Cause(err))
 	}