feat: immediate start HTTP server in DLE and provide the actual instance status (#307):

* start DLE API and Local UI instance before running data retrieval
* display a link to get support in logs in case of errors
* provide actual instance status considering subsystems
* provide data retrieval state and alerts

Author: agneum

cmd/database-lab/main.go

@@ -42,6 +42,8 @@ import (
 
 const (
 	shutdownTimeout = 30 * time.Second
+	contactSupport  = "If you have problems or questions, " +
+		"please contact Postgres.ai: https://postgres.ai/support"
 )
 
 func main() {
@@ -50,15 +52,21 @@ func main() {
 		log.Fatal(errors.WithMessage(err, "failed to parse config"))
 	}
 
-	dockerCLI, err := client.NewClientWithOpts(client.FromEnv)
+	docker, err := client.NewClientWithOpts(client.FromEnv)
 	if err != nil {
 		log.Fatal("Failed to create a Docker client:", err)
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	engProps, err := getEngineProperties(ctx, dockerCLI, cfg)
+	defer func() {
+		if err != nil {
+			log.Msg(contactSupport)
+		}
+	}()
+
+	engProps, err := getEngineProperties(ctx, docker, cfg)
 	if err != nil {
 		log.Err("failed to get Database Lab Engine properties:", err.Error())
 		return
@@ -74,18 +82,18 @@ func main() {
 	runner := runners.NewLocalRunner(cfg.Provision.UseSudo)
 
 	pm := pool.NewPoolManager(&cfg.PoolManager, runner)
-	if err := pm.ReloadPools(); err != nil {
+	if err = pm.ReloadPools(); err != nil {
 		log.Err(err.Error())
 		return
 	}
 
-	internalNetworkID, err := networks.Setup(ctx, dockerCLI, engProps.InstanceID, engProps.ContainerName)
+	internalNetworkID, err := networks.Setup(ctx, docker, engProps.InstanceID, engProps.ContainerName)
 	if err != nil {
 		log.Errf(err.Error())
 		return
 	}
 
-	defer networks.Stop(dockerCLI, internalNetworkID, engProps.ContainerName)
+	defer networks.Stop(docker, internalNetworkID, engProps.ContainerName)
 
 	// Create a platform service to make requests to Platform.
 	platformSvc, err := platform.New(ctx, cfg.Platform)
@@ -105,7 +113,7 @@ func main() {
 		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), shutdownTimeout)
 		defer shutdownCancel()
 
-		shutdownDatabaseLabEngine(shutdownCtx, dockerCLI, engProps, pm.First().Pool())
+		shutdownDatabaseLabEngine(shutdownCtx, docker, engProps, pm.First().Pool())
 	}
 
 	tm, err := telemetry.New(cfg.Global, engProps)
@@ -115,56 +123,46 @@ func main() {
 	}
 
 	// Create a new retrieval service to prepare a data directory and start snapshotting.
-	retrievalSvc := retrieval.New(cfg, engProps, dockerCLI, pm, tm, runner)
-
-	if err := retrievalSvc.Run(ctx); err != nil {
-		log.Err("Failed to run the data retrieval service:", err)
-		emergencyShutdown()
-
-		return
-	}
-
-	defer retrievalSvc.Stop()
+	retrievalSvc := retrieval.New(cfg, engProps, docker, pm, tm, runner)
 
 	// Create a cloning service to provision new clones.
-	provisionSvc, err := provision.New(ctx, &cfg.Provision, dbCfg, dockerCLI, pm, engProps.InstanceID, internalNetworkID)
+	provisioner, err := provision.New(ctx, &cfg.Provision, dbCfg, docker, pm, engProps.InstanceID, internalNetworkID)
 	if err != nil {
 		log.Errf(errors.WithMessage(err, `error in the "provision" section of the config`).Error())
 	}
 
 	observingChan := make(chan string, 1)
 
-	cloningSvc := cloning.NewBase(&cfg.Cloning, provisionSvc, tm, observingChan)
+	cloningSvc := cloning.NewBase(&cfg.Cloning, provisioner, tm, observingChan)
 	if err = cloningSvc.Run(ctx); err != nil {
 		log.Err(err)
 		emergencyShutdown()
 
 		return
 	}
 
-	obs := observer.NewObserver(dockerCLI, &cfg.Observer, pm)
+	obs := observer.NewObserver(docker, &cfg.Observer, pm)
 	est := estimator.NewEstimator(&cfg.Estimator)
 
 	go removeObservingClones(observingChan, obs)
 
 	tm.SendEvent(ctx, telemetry.EngineStartedEvent, telemetry.EngineStarted{
 		EngineVersion: version.GetVersion(),
-		DBVersion:     provisionSvc.DetectDBVersion(),
+		DBVersion:     provisioner.DetectDBVersion(),
 		Pools:         pm.CollectPoolStat(),
 		Restore:       retrievalSvc.CollectRestoreTelemetry(),
 	})
 
-	localUI := localui.New(cfg.LocalUI, engProps, runner, dockerCLI)
-	server := srv.NewServer(&cfg.Server, &cfg.Global, engProps, cloningSvc, retrievalSvc, platformSvc, dockerCLI, obs, est, pm, tm)
+	localUI := localui.New(cfg.LocalUI, engProps, runner, docker)
+	server := srv.NewServer(&cfg.Server, &cfg.Global, engProps, docker, cloningSvc, provisioner, retrievalSvc, platformSvc, obs, est, pm, tm)
 	shutdownCh := setShutdownListener()
 
-	go setReloadListener(ctx, provisionSvc, tm, retrievalSvc, pm, cloningSvc, platformSvc, est, localUI, server)
+	go setReloadListener(ctx, provisioner, tm, retrievalSvc, pm, cloningSvc, platformSvc, est, localUI, server)
 
 	server.InitHandlers()
 
 	go func() {
-		// Start the Database Lab.
-		if err = server.Run(); err != nil {
+		if err := server.Run(); err != nil {
 			log.Msg(err)
 		}
 	}()
@@ -178,6 +176,13 @@ func main() {
 		}()
 	}
 
+	if err := retrievalSvc.Run(ctx); err != nil {
+		log.Err("Failed to run the data retrieval service:", err)
+		log.Msg(contactSupport)
+	}
+
+	defer retrievalSvc.Stop()
+
 	<-shutdownCh
 	cancel()
 
@@ -188,7 +193,7 @@ func main() {
 		log.Msg(err)
 	}
 
-	shutdownDatabaseLabEngine(shutdownCtx, dockerCLI, engProps, pm.First().Pool())
+	shutdownDatabaseLabEngine(shutdownCtx, docker, engProps, pm.First().Pool())
 	cloningSvc.SaveClonesState()
 	tm.SendEvent(context.Background(), telemetry.EngineStoppedEvent, telemetry.EngineStopped{Uptime: server.Uptime()})
 }

internal/cloning/base.go

@@ -302,7 +302,10 @@ func (c *Base) DestroyClone(cloneID string) error {
 
 	if w.Session == nil {
 		c.deleteClone(cloneID)
-		c.decrementCloneNumber(w.Clone.Snapshot.ID)
+
+		if w.Clone.Snapshot != nil {
+			c.decrementCloneNumber(w.Clone.Snapshot.ID)
+		}
 
 		return nil
 	}
@@ -472,18 +475,16 @@ func (c *Base) ResetClone(cloneID string, resetOptions types.ResetCloneRequest)
 	return nil
 }
 
-// GetInstanceState returns the current state of instance.
-func (c *Base) GetInstanceState() (*models.InstanceStatus, error) {
+// GetCloningState returns the current state of instance.
+func (c *Base) GetCloningState() models.Cloning {
 	clones := c.GetClones()
-	c.instanceStatus.Cloning = models.Cloning{
+	cloning := models.Cloning{
 		ExpectedCloningTime: c.getExpectedCloningTime(),
 		Clones:              clones,
 		NumClones:           uint64(len(clones)),
 	}
-	c.instanceStatus.Pools = c.provision.GetPoolEntryList()
-	c.instanceStatus.Engine.Telemetry = pointer.ToBool(c.tm.IsEnabled())
 
-	return c.instanceStatus, nil
+	return cloning
 }
 
 // GetSnapshots returns all available snapshots.

internal/retrieval/engine/postgres/snapshot/logical.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"path"
+	"time"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
@@ -27,9 +28,9 @@ import (
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/tools/health"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/options"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/telemetry"
-
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/config/global"
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/log"
+	"gitlab.com/postgres-ai/database-lab/v3/pkg/util"
 )
 
 const (
@@ -153,11 +154,30 @@ func (s *LogicalInitial) Run(ctx context.Context) error {
 		return errors.Wrap(err, "failed to create a snapshot")
 	}
 
+	if err := s.markDatabaseData(dataStateAt); err != nil {
+		return errors.Wrap(err, "failed to mark logical data")
+	}
+
 	s.tm.SendEvent(ctx, telemetry.SnapshotCreatedEvent, telemetry.SnapshotCreated{})
 
 	return nil
 }
 
+func (s *LogicalInitial) markDatabaseData(dataStateAt string) error {
+	if dataStateAt != "" {
+		return nil
+	}
+
+	if err := s.dbMarker.CreateConfig(); err != nil {
+		return errors.Wrap(err, "failed to create a DBMarker config of the database")
+	}
+
+	return s.dbMarker.SaveConfig(&dbmarker.Config{
+		DataType:    dbmarker.LogicalDataType,
+		DataStateAt: time.Now().Format(util.DataStateAtFormat),
+	})
+}
+
 func (s *LogicalInitial) touchConfigFiles() error {
 	dataDir := s.fsPool.DataDir()
 

internal/retrieval/retrieval.go

@@ -76,6 +76,7 @@ func New(cfg *dblabCfg.Config, engineProps global.EngineProps, docker *client.Cl
 		jobSpecs:    make(map[string]config.JobSpec, len(cfg.Retrieval.Jobs)),
 		State: State{
 			Status: models.Inactive,
+			alerts: make(map[models.AlertType]models.Alert),
 		},
 	}
 
@@ -124,25 +125,32 @@ func (r *Retrieval) Run(ctx context.Context) error {
 	if err != nil {
 		var skipError *SkipRefreshingError
 		if errors.As(err, &skipError) {
+			r.State.Status = models.Finished
+
 			log.Msg("Continue without performing a full refresh:", skipError.Error())
 			r.setupScheduler(ctx)
 
 			return nil
 		}
 
-		r.tm.SendEvent(ctx, telemetry.AlertEvent, telemetry.Alert{
+		alert := telemetry.Alert{
 			Level:   models.RefreshFailed,
 			Message: "Pool to perform data refresh not found",
-		})
+		}
+		r.State.addAlert(alert)
+		r.tm.SendEvent(ctx, telemetry.AlertEvent, alert)
 
 		return fmt.Errorf("failed to choose pool to refresh: %w", err)
 	}
 
 	log.Msg("Pool to perform data retrieving: ", fsManager.Pool().Name)
 
 	if err := r.run(runCtx, fsManager); err != nil {
-		r.tm.SendEvent(ctx, telemetry.AlertEvent, telemetry.Alert{Level: models.RefreshFailed,
-			Message: fmt.Sprintf("Failed to perform initial data retrieving: %s", r.State.Mode)})
+		alert := telemetry.Alert{Level: models.RefreshFailed,
+			Message: fmt.Sprintf("Failed to perform initial data retrieving: %s", r.State.Mode)}
+		r.State.addAlert(alert)
+		r.tm.SendEvent(ctx, telemetry.AlertEvent, alert)
+
 		return err
 	}
 
@@ -186,15 +194,6 @@ func (r *Retrieval) getPoolToDataRetrieving() (pool.FSManager, error) {
 }
 
 func (r *Retrieval) run(ctx context.Context, fsm pool.FSManager) (err error) {
-	r.retrieveMutex.Lock()
-	r.State.Status = models.Refreshing
-	r.State.LastRefresh = pointer.ToTimeOrNil(time.Now().Truncate(time.Second))
-
-	defer func() {
-		r.State.Status = models.Finished
-		r.retrieveMutex.Unlock()
-	}()
-
 	if err := r.configure(fsm); err != nil {
 		return errors.Wrap(err, "failed to configure")
 	}
@@ -213,21 +212,34 @@ func (r *Retrieval) run(ctx context.Context, fsm pool.FSManager) (err error) {
 		return errors.Errorf("pool %s not found", fsm.Pool().Name)
 	}
 
-	fsm.Pool().SetStatus(resources.RefreshingPool)
+	if len(r.jobs) > 0 {
+		fsm.Pool().SetStatus(resources.RefreshingPool)
 
-	defer func() {
-		if err != nil {
-			fsm.Pool().SetStatus(resources.EmptyPool)
-		}
-	}()
+		r.retrieveMutex.Lock()
+		r.State.Status = models.Refreshing
+		r.State.LastRefresh = pointer.ToTimeOrNil(time.Now().Truncate(time.Second))
+
+		defer func() {
+			r.State.Status = models.Finished
+
+			if err != nil {
+				r.State.Status = models.Failed
 
-	for _, j := range r.jobs {
-		if err := j.Run(ctx); err != nil {
-			return err
+				fsm.Pool().SetStatus(resources.EmptyPool)
+			}
+
+			r.retrieveMutex.Unlock()
+		}()
+
+		for _, j := range r.jobs {
+			if err := j.Run(ctx); err != nil {
+				return err
+			}
 		}
 	}
 
 	r.poolManager.MakeActive(poolByName)
+	r.State.cleanAlerts()
 
 	return nil
 }
@@ -387,9 +399,10 @@ func (r *Retrieval) setupScheduler(ctx context.Context) {
 func (r *Retrieval) refreshFunc(ctx context.Context) func() {
 	return func() {
 		if err := r.fullRefresh(ctx); err != nil {
-			r.tm.SendEvent(ctx, telemetry.AlertEvent, telemetry.Alert{Level: models.RefreshFailed,
-				Message: "Failed to run full-refresh"})
-			log.Err("Failed to run full-refresh: ", err)
+			alert := telemetry.Alert{Level: models.RefreshFailed, Message: "Failed to run full-refresh"}
+			r.State.addAlert(alert)
+			r.tm.SendEvent(ctx, telemetry.AlertEvent, alert)
+			log.Err(alert.Message, err)
 		}
 	}
 }
@@ -406,11 +419,13 @@ func (r *Retrieval) fullRefresh(ctx context.Context) error {
 	elementToUpdate := r.poolManager.GetPoolToUpdate()
 
 	if elementToUpdate == nil || elementToUpdate.Value == nil {
-		r.tm.SendEvent(ctx, telemetry.AlertEvent, telemetry.Alert{
+		alert := telemetry.Alert{
 			Level:   models.RefreshSkipped,
 			Message: "Pool to perform full refresh not found. Skip refreshing",
-		})
-		log.Msg("Pool to perform full refresh not found. Skip refreshing")
+		}
+		r.State.addAlert(alert)
+		r.tm.SendEvent(ctx, telemetry.AlertEvent, alert)
+		log.Msg(alert.Message)
 
 		return nil
 	}
@@ -438,6 +453,7 @@ func (r *Retrieval) fullRefresh(ctx context.Context) error {
 	}
 
 	r.poolManager.MakeActive(elementToUpdate)
+	r.State.cleanAlerts()
 
 	return nil
 }