Merge branch '293-transparent-dump-errors' into 'master'

fix: recognize errors from external tools and mention them in logs (#293)

Closes #293

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

Author: agneum

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

@@ -643,7 +643,7 @@ func (d *DumpJob) buildLogicalDumpCommand(dbName string, tables []string) []stri
 
 func (d *DumpJob) buildLogicalRestoreCommand(dbName string) []string {
 	restoreCmd := []string{"|", "pg_restore", "--username", d.globalCfg.Database.User(), "--dbname", defaults.DBName,
-		"--no-privileges", "--no-owner"}
+		"--no-privileges", "--no-owner", "--exit-on-error"}
 
 	if dbName != defaults.DBName {
 		// To avoid recreating of the default database.

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

@@ -709,7 +709,7 @@ func (r *RestoreJob) buildPlainTextCommand(dumpName string, definition DumpDefin
 
 func (r *RestoreJob) buildPGRestoreCommand(dumpName string, definition DumpDefinition) []string {
 	restoreCmd := []string{"pg_restore", "--username", r.globalCfg.Database.User(), "--dbname", defaults.DBName,
-		"--no-privileges", "--no-owner"}
+		"--no-privileges", "--no-owner", "--exit-on-error"}
 
 	if definition.dbName != defaults.DBName {
 		// To avoid recreating of the default database.

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

@@ -42,14 +42,14 @@ func TestRestoreCommandBuilding(t *testing.T) {
 				},
 				DumpLocation: "/tmp/db.dump",
 			},
-			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--create", "--jobs", "1", "/tmp/db.dump"},
+			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--exit-on-error", "--create", "--jobs", "1", "/tmp/db.dump"},
 		},
 		{
 			copyOptions: RestoreOptions{
 				ParallelJobs: 4,
 				ForceInit:    true,
 			},
-			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--create", "--clean", "--if-exists", "--jobs", "4", ""},
+			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--exit-on-error", "--create", "--clean", "--if-exists", "--jobs", "4", ""},
 		},
 		{
 			copyOptions: RestoreOptions{
@@ -58,7 +58,7 @@ func TestRestoreCommandBuilding(t *testing.T) {
 				Databases:    map[string]DumpDefinition{"testDB": {}},
 				DumpLocation: "/tmp/db.dump",
 			},
-			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--create", "--jobs", "2", "/tmp/db.dump/testDB"},
+			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--exit-on-error", "--create", "--jobs", "2", "/tmp/db.dump/testDB"},
 		},
 		{
 			copyOptions: RestoreOptions{
@@ -71,7 +71,7 @@ func TestRestoreCommandBuilding(t *testing.T) {
 				},
 				DumpLocation: "/tmp/db.dump",
 			},
-			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--create", "--jobs", "1", "--table", "test", "--table", "users", "/tmp/db.dump/testDB"},
+			command: []string{"pg_restore", "--username", "john", "--dbname", "postgres", "--no-privileges", "--no-owner", "--exit-on-error", "--create", "--jobs", "1", "--table", "test", "--table", "users", "/tmp/db.dump/testDB"},
 		},
 		{
 			copyOptions: RestoreOptions{

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

@@ -473,42 +473,50 @@ func ExecCommandWithOutput(ctx context.Context, dockerClient *client.Client, con
 
 	defer attachResponse.Close()
 
-	wb := new(bytes.Buffer)
+	output, err := processAttachResponse(ctx, attachResponse.Reader)
+	if err != nil {
+		return string(output), errors.Wrap(err, "failed to read response of exec command")
+	}
 
-	if err := processAttachResponse(ctx, attachResponse.Reader, wb); err != nil {
-		return "", errors.Wrap(err, "failed to read response of exec command")
+	inspection, err := dockerClient.ContainerExecInspect(ctx, execCommand.ID)
+	if err != nil {
+		return "", fmt.Errorf("failed to inspect an exec process: %w", err)
 	}
 
-	return string(bytes.TrimSpace(wb.Bytes())), nil
+	if inspection.ExitCode != 0 {
+		err = fmt.Errorf("exit code: %d", inspection.ExitCode)
+	}
+
+	return string(output), err
 }
 
 // processAttachResponse reads and processes the cmd output.
-func processAttachResponse(ctx context.Context, reader io.Reader, output io.Writer) error {
-	var errBuf bytes.Buffer
+func processAttachResponse(ctx context.Context, reader io.Reader) ([]byte, error) {
+	var outBuf, errBuf bytes.Buffer
 
 	outputDone := make(chan error)
 
 	go func() {
 		// StdCopy de-multiplexes the stream into two writers.
-		_, err := stdcopy.StdCopy(output, &errBuf, reader)
+		_, err := stdcopy.StdCopy(&outBuf, &errBuf, reader)
 		outputDone <- err
 	}()
 
 	select {
 	case err := <-outputDone:
 		if err != nil {
-			return errors.Wrap(err, "failed to copy output")
+			return nil, errors.Wrap(err, "failed to copy output")
 		}
 
 		break
 
 	case <-ctx.Done():
-		return ctx.Err()
+		return nil, ctx.Err()
 	}
 
 	if errBuf.Len() > 0 {
-		return errors.New(errBuf.String())
+		return nil, errors.New(errBuf.String())
 	}
 
-	return nil
+	return bytes.TrimSpace(outBuf.Bytes()), nil
 }