Merge branch '605-idle-clone-detection' into 'dle-4-0'

fix: logs location for idle clone detection (#605)

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

Author: bogdan-ts

engine/internal/cloning/base.go

@@ -767,7 +767,8 @@ 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.Branch, wrapper.Clone.ID, minimumTime); err != nil {
+	if _, err := c.provision.LastSessionActivity(session, wrapper.Clone.Branch, wrapper.Clone.ID, wrapper.Clone.Revision,
+		minimumTime); err != nil {
 		if err == pglog.ErrNotFound {
 			log.Dbg(fmt.Sprintf("Not found recent activity for session: %q. Clone name: %q",
 				session.ID, wrapper.Clone.ID))

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.branch, obsClone.cloneID))
+	fileSelector := pglog.NewSelector(obsClone.pool.ClonePath(obsClone.branch, obsClone.cloneID, obsClone.revision))
 	fileSelector.SetMinimumTime(obsClone.session.StartedAt)
 
 	if err := fileSelector.DiscoverLogDir(); err != nil {
@@ -187,12 +187,13 @@ func (o *Observer) maskLogs(entry []string, maskedFieldIndexes []int) {
 }
 
 // AddObservingClone adds a new observing session to storage.
-func (o *Observer) AddObservingClone(cloneID, branch string, port uint, session *ObservingClone) {
+func (o *Observer) AddObservingClone(cloneID, branch string, revision int, port uint, session *ObservingClone) {
 	o.sessionMu.Lock()
 	defer o.sessionMu.Unlock()
 	session.pool = o.pm.First().Pool()
 	session.cloneID = cloneID
 	session.branch = branch
+	session.revision = revision
 	session.port = port
 
 	o.storage[cloneID] = session

engine/internal/observer/observing_clone.go

@@ -44,6 +44,7 @@ type ObservingClone struct {
 	pool        *resources.Pool
 	cloneID     string
 	branch      string
+	revision    int
 	port        uint
 	superUserDB *pgx.Conn
 
@@ -480,7 +481,7 @@ func (c *ObservingClone) currentArtifactsSessionPath() string {
 }
 
 func (c *ObservingClone) artifactsSessionPath(sessionID uint64) string {
-	return path.Join(c.pool.ObserverDir(c.branch, c.cloneID), c.cloneID, strconv.FormatUint(sessionID, 10))
+	return path.Join(c.pool.ObserverDir(c.branch, c.cloneID, c.revision), 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/mode_local.go

@@ -660,7 +660,8 @@ func (p *Provisioner) getProvisionHosts() string {
 }
 
 // LastSessionActivity returns the time of the last session activity.
-func (p *Provisioner) LastSessionActivity(session *resources.Session, branch, cloneID string, minimumTime time.Time) (*time.Time, error) {
+func (p *Provisioner) LastSessionActivity(session *resources.Session, branch, cloneID string, revision int,
+	minimumTime time.Time) (*time.Time, error) {
 	fsm, err := p.pm.GetFSManager(session.Pool)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to find filesystem manager")
@@ -669,7 +670,7 @@ func (p *Provisioner) LastSessionActivity(session *resources.Session, branch, cl
 	ctx, cancel := context.WithCancel(p.ctx)
 	defer cancel()
 
-	clonePath := fsm.Pool().ClonePath(branch, cloneID)
+	clonePath := fsm.Pool().ClonePath(branch, cloneID, revision)
 	fileSelector := pglog.NewSelector(clonePath)
 
 	if err := fileSelector.DiscoverLogDir(); err != nil {

engine/internal/provision/resources/pool.go

@@ -65,8 +65,8 @@ func (p *Pool) SocketDir() string {
 }
 
 // ObserverDir returns a path to the observer directory of the storage pool.
-func (p *Pool) ObserverDir(branch, name string) string {
-	return path.Join(p.ClonePath(branch, name), p.ObserverSubDir)
+func (p *Pool) ObserverDir(branch, name string, revision int) string {
+	return path.Join(p.ClonePath(branch, name, revision), p.ObserverSubDir)
 }
 
 // ClonesDir returns a path to the clones directory of the storage pool.
@@ -75,8 +75,8 @@ func (p *Pool) ClonesDir(branch string) string {
 }
 
 // ClonePath returns a path to the data clone directory.
-func (p *Pool) ClonePath(branchName, name string) string {
-	return path.Join(p.MountDir, p.PoolDirName, branching.BranchDir, branchName, name, p.DataSubDir)
+func (p *Pool) ClonePath(branchName, name string, revision int) string {
+	return path.Join(p.MountDir, p.PoolDirName, branching.BranchDir, branchName, name, branching.RevisionSegment(revision), p.DataSubDir)
 }
 
 // CloneLocation returns a path to the initialized clone directory.

engine/internal/srv/routes.go

@@ -743,7 +743,7 @@ func (s *Server) startObservation(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	s.Observer.AddObservingClone(clone.ID, clone.Branch, uint(port), observingClone)
+	s.Observer.AddObservingClone(clone.ID, clone.Branch, clone.Revision, uint(port), observingClone)
 
 	// Start session on the Platform.
 	platformRequest := platform.StartObservationRequest{