Merge branch '314-sync-status' into 'master'

feat(engine): show continuous PGDATA synchronization details in API+CLI (#314)

Closes #314

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

Author: agneum

engine/api/swagger-spec/dblab_server_swagger.yaml

@@ -610,6 +610,8 @@ definitions:
         $ref: "#/definitions/Retrieving"
       provisioner:
         $ref: "#/definitions/Provisioner"
+      synchronization:
+        $ref: "#/definitions/Synchronization"
 
   Status:
     type: "object"
@@ -717,6 +719,24 @@ definitions:
       containerConfig:
         type: "object"
 
+  Synchronization:
+    type: "object"
+    properties:
+      status:
+        $ref: "#/definitions/Status"
+      startedAt:
+        type: "string"
+        format: "date-time"
+      lastReplayedLsn:
+        type: "string"
+      lastReplayedLsnAt:
+        type: "string"
+        format: "date-time"
+      replicationLag:
+        type: "string"
+      replicationUptime:
+        type: "integer"
+
   Snapshot:
     type: "object"
     properties:

engine/internal/retrieval/engine/postgres/physical/physical.go

@@ -11,12 +11,14 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
+	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/client"
 
 	"github.com/pkg/errors"
@@ -207,7 +209,12 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
 		return errors.Wrap(err, "failed to generate PostgreSQL password")
 	}
 
-	contID, err := r.startContainer(ctx, r.restoreContainerName(), r.buildContainerConfig(cont.DBLabRestoreLabel, pwd))
+	hostConfig, err := cont.BuildHostConfig(ctx, r.dockerClient, r.fsPool.DataDir(), r.CopyOptions.ContainerConfig)
+	if err != nil {
+		return errors.Wrap(err, "failed to build container host config")
+	}
+
+	contID, err := r.startContainer(ctx, r.restoreContainerName(), r.buildContainerConfig(cont.DBLabRestoreLabel, pwd), hostConfig)
 	if err != nil {
 		return err
 	}
@@ -288,12 +295,8 @@ func (r *RestoreJob) Run(ctx context.Context) (err error) {
 	return nil
 }
 
-func (r *RestoreJob) startContainer(ctx context.Context, containerName string, containerConfig *container.Config) (string, error) {
-	hostConfig, err := cont.BuildHostConfig(ctx, r.dockerClient, r.fsPool.DataDir(), r.CopyOptions.ContainerConfig)
-	if err != nil {
-		return "", errors.Wrap(err, "failed to build container host config")
-	}
-
+func (r *RestoreJob) startContainer(ctx context.Context, containerName string, containerConfig *container.Config,
+	hostConfig *container.HostConfig) (string, error) {
 	if err := tools.PullImage(ctx, r.dockerClient, r.CopyOptions.DockerImage); err != nil {
 		return "", err
 	}
@@ -354,7 +357,24 @@ func (r *RestoreJob) runSyncInstance(ctx context.Context) (err error) {
 
 	log.Msg("Starting sync instance: ", r.syncInstanceName())
 
-	syncInstanceID, err := r.startContainer(ctx, r.syncInstanceName(), syncInstanceConfig)
+	hostConfig, err := cont.BuildHostConfig(ctx, r.dockerClient, r.fsPool.DataDir(), r.CopyOptions.ContainerConfig)
+	if err != nil {
+		return fmt.Errorf("failed to build container host config: %w", err)
+	}
+
+	path := filepath.Join(r.fsPool.SocketDir(), r.syncInstanceName())
+	if err := os.MkdirAll(path, 0755); err != nil {
+		return fmt.Errorf("failed to make socket directory: %w", err)
+	}
+
+	hostConfig.Mounts = append(hostConfig.Mounts, mount.Mount{
+		Type:     mount.TypeBind,
+		Source:   path,
+		Target:   cont.DefaultPostgresSocket,
+		ReadOnly: false,
+	})
+
+	syncInstanceID, err := r.startContainer(ctx, r.syncInstanceName(), syncInstanceConfig, hostConfig)
 	if err != nil {
 		return err
 	}

engine/internal/retrieval/engine/postgres/tools/cont/container.go

@@ -62,6 +62,9 @@ const (
 
 	// DBLabRunner defines a label to mark runner containers.
 	DBLabRunner = "dblab_runner"
+
+	// DefaultPostgresSocket defines default path for Postgres socket
+	DefaultPostgresSocket = "/var/run/postgresql"
 )
 
 // TODO(akartasov): Control container manager.

engine/internal/retrieval/retrieval.go

@@ -9,9 +9,12 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strings"
 	"time"
 
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/client"
 	"github.com/pkg/errors"
 	"github.com/robfig/cron/v3"
@@ -26,9 +29,11 @@ import (
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/logical"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/physical"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/snapshot"
+	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/tools"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/tools/cont"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/engine/postgres/tools/db"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/options"
+	"gitlab.com/postgres-ai/database-lab/v3/internal/retrieval/status"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/telemetry"
 	"gitlab.com/postgres-ai/database-lab/v3/pkg/util"
 
@@ -690,3 +695,85 @@ func (r *Retrieval) GetStageSpec(stage string) (config.JobSpec, error) {
 
 	return stageSpec, nil
 }
+
+// ReportSyncStatus return status of sync containers.
+func (r *Retrieval) ReportSyncStatus(ctx context.Context) (*models.Sync, error) {
+	if r.State.Mode != models.Physical {
+		return &models.Sync{
+			Status: models.Status{Code: models.SyncStatusNotAvailable},
+		}, nil
+	}
+
+	filterArgs := filters.NewArgs(
+		filters.KeyValuePair{Key: "label",
+			Value: fmt.Sprintf("%s=%s", cont.DBLabControlLabel, cont.DBLabSyncLabel)})
+
+	filterArgs.Add("label", fmt.Sprintf("%s=%s", cont.DBLabInstanceIDLabel, r.engineProps.InstanceID))
+
+	ids, err := tools.ListContainersByLabel(ctx, r.docker, filterArgs)
+	if err != nil {
+		return &models.Sync{
+			Status: models.Status{Code: models.SyncStatusError, Message: err.Error()},
+		}, fmt.Errorf("failed to list containers by label %w", err)
+	}
+
+	if len(ids) != 1 {
+		return &models.Sync{
+			Status: models.Status{Code: models.SyncStatusError},
+		}, fmt.Errorf("failed to match sync container")
+	}
+
+	id := ids[0]
+
+	sync, err := r.reportContainerSyncStatus(ctx, id)
+
+	return sync, err
+}
+
+func (r *Retrieval) reportContainerSyncStatus(ctx context.Context, containerID string) (*models.Sync, error) {
+	resp, err := r.docker.ContainerInspect(ctx, containerID)
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to inspect container %w", err)
+	}
+
+	if resp.State == nil {
+		return nil, fmt.Errorf("failed to read container state")
+	}
+
+	if resp.State.Health != nil && resp.State.Health.Status == types.Unhealthy {
+		// in case of Unhealthy state, add health check output to status
+		var healthCheckOutput = ""
+
+		if healthCheckLength := len(resp.State.Health.Log); healthCheckLength > 0 {
+			if lastHealthCheck := resp.State.Health.Log[healthCheckLength-1]; lastHealthCheck.ExitCode > 1 {
+				healthCheckOutput = lastHealthCheck.Output
+			}
+		}
+
+		return &models.Sync{
+			Status: models.Status{
+				Code:    models.SyncStatusDown,
+				Message: healthCheckOutput,
+			},
+		}, nil
+	}
+
+	socketPath := filepath.Join(r.poolManager.First().Pool().SocketDir(), resp.Name)
+	value, err := status.FetchSyncMetrics(ctx, r.global, socketPath)
+
+	if err != nil {
+		log.Warn("Failed to fetch synchronization metrics", err)
+
+		return &models.Sync{
+			Status: models.Status{
+				Code:    models.SyncStatusError,
+				Message: err.Error(),
+			},
+		}, nil
+	}
+
+	value.StartedAt = resp.State.StartedAt
+
+	return value, nil
+}

engine/internal/retrieval/retrieval_test.go

@@ -1,6 +1,7 @@
 package retrieval
 
 import (
+	"context"
 	"os"
 	"path"
 	"testing"
@@ -117,3 +118,15 @@ func TestPendingMarker(t *testing.T) {
 		assert.Equal(t, models.Finished, r.State.Status)
 	})
 }
+
+func TestSyncStatusNotReportedForLogicalMode(t *testing.T) {
+	var r = Retrieval{
+		State: State{
+			Mode: models.Logical,
+		},
+	}
+	status, err := r.ReportSyncStatus(context.TODO())
+	assert.NoError(t, err)
+	assert.NotNil(t, status)
+	assert.Equal(t, models.SyncStatusNotAvailable, status.Status.Code)
+}