Add get/put/rm to storage tools (wal-g#1069)

.github/workflows/dockertests.yml

@@ -121,6 +121,7 @@ jobs:
                    'make MYSQL_TEST=mysql_delete_tests mysql_integration_test',
                    'make MYSQL_TEST=mysql_copy_tests mysql_integration_test',
                    'make gp_test',
+                   'make st_test',
       ]
       # do not cancel all tests if one failed
       fail-fast: false

Makefile

@@ -179,6 +179,12 @@ gp_integration_test: load_docker_common
 	docker-compose build gp gp_tests
 	docker-compose up --exit-code-from gp_tests gp_tests
 
+st_test: deps pg_build unlink_brotli st_integration_test
+
+st_integration_test: load_docker_common
+	docker-compose build st_tests
+	docker-compose up --exit-code-from st_tests st_tests
+
 unittest:
 	go list ./... | grep -Ev 'vendor|submodules|tmp' | xargs go vet
 	go test -mod vendor -v $(TEST_MODIFIER) -tags "$(BUILD_TAGS)" ./internal/

cmd/st/delete_object.go

@@ -0,0 +1,27 @@
+package st
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/wal-g/tracelog"
+	"github.com/wal-g/wal-g/internal"
+	"github.com/wal-g/wal-g/internal/storagetools"
+)
+
+const deleteObjectShortDescription = "Delete the specified storage object"
+
+// deleteObjectCmd represents the deleteObject command
+var deleteObjectCmd = &cobra.Command{
+	Use:   "rm relative_object_path",
+	Short: deleteObjectShortDescription,
+	Args:  cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		folder, err := internal.ConfigureFolder()
+		tracelog.ErrorLogger.FatalOnError(err)
+
+		storagetools.HandleDeleteObject(args[0], folder)
+	},
+}
+
+func init() {
+	StorageToolsCmd.AddCommand(deleteObjectCmd)
+}

cmd/st/get_object.go

@@ -0,0 +1,40 @@
+package st
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/wal-g/tracelog"
+	"github.com/wal-g/wal-g/internal"
+	"github.com/wal-g/wal-g/internal/storagetools"
+)
+
+const (
+	getObjectShortDescription = "Download the specified storage object"
+
+	noDecryptFlag    = "no-decrypt"
+	noDecompressFlag = "no-decompress"
+)
+
+// getObjectCmd represents the getObject command
+var getObjectCmd = &cobra.Command{
+	Use:   "get relative_object_path destination_path",
+	Short: getObjectShortDescription,
+	Args:  cobra.ExactArgs(2),
+	Run: func(cmd *cobra.Command, args []string) {
+		objectPath := args[0]
+		dstPath := args[1]
+
+		folder, err := internal.ConfigureFolder()
+		tracelog.ErrorLogger.FatalOnError(err)
+
+		storagetools.HandleGetObject(objectPath, dstPath, folder, !noDecrypt, !noDecompress)
+	},
+}
+
+var noDecrypt bool
+var noDecompress bool
+
+func init() {
+	StorageToolsCmd.AddCommand(getObjectCmd)
+	getObjectCmd.Flags().BoolVar(&noDecrypt, noDecryptFlag, false, "Do not decrypt the object")
+	getObjectCmd.Flags().BoolVar(&noDecompress, noDecompressFlag, false, "Do not decompress the object")
+}

cmd/st/put_object.go

@@ -0,0 +1,45 @@
+package st
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/wal-g/tracelog"
+	"github.com/wal-g/wal-g/internal"
+	"github.com/wal-g/wal-g/internal/storagetools"
+)
+
+const (
+	putObjectShortDescription = "Upload the specified file to the storage"
+
+	noEncryptFlag      = "no-encrypt"
+	noCompressFlag     = "no-compress"
+	overwriteFlag      = "force"
+	overwriteShorthand = "f"
+)
+
+// putObjectCmd represents the putObject command
+var putObjectCmd = &cobra.Command{
+	Use:   "put local_path destination_path",
+	Short: putObjectShortDescription,
+	Args:  cobra.ExactArgs(2),
+	Run: func(cmd *cobra.Command, args []string) {
+		uploader, err := internal.ConfigureUploader()
+		tracelog.ErrorLogger.FatalOnError(err)
+
+		localPath := args[0]
+		dstPath := args[1]
+
+		storagetools.HandlePutObject(localPath, dstPath, uploader, overwrite, !noEncrypt, !noCompress)
+	},
+}
+
+var noEncrypt bool
+var noCompress bool
+var overwrite bool
+
+func init() {
+	StorageToolsCmd.AddCommand(putObjectCmd)
+	putObjectCmd.Flags().BoolVar(&noEncrypt, noEncryptFlag, false, "Do not encrypt the object")
+	putObjectCmd.Flags().BoolVar(&noCompress, noCompressFlag, false, "Do not compress the object")
+	putObjectCmd.Flags().BoolVarP(&overwrite, overwriteFlag, overwriteShorthand,
+		false, "Overwrite the existing object")
+}

docker-compose.yml

@@ -73,6 +73,7 @@ services:
       &&  mkdir -p /export/mysqldeletebinlogs
       &&  mkdir -p /export/archivereadyrename
       &&  mkdir -p /export/gpfullbucket
+      &&  mkdir -p /export/storagetoolsbucket
       &&  /usr/bin/minio server /export'
 
   s3-another:
@@ -635,3 +636,16 @@ services:
       - s3
     links:
       - s3
+
+  st_tests:
+    build:
+      dockerfile: docker/st_tests/Dockerfile
+      context: .
+    image: wal-g/st_tests
+    container_name: wal-g_st_tests
+    env_file:
+      - docker/common/common_walg.env
+    depends_on:
+      - s3
+    links:
+      - s3

docker/st_tests/Dockerfile

@@ -0,0 +1,34 @@
+FROM wal-g/golang:latest as build
+
+WORKDIR /go/src/github.com/wal-g/wal-g
+
+RUN apt-get update && \
+    apt-get install --yes --no-install-recommends --no-install-suggests \
+    liblzo2-dev
+
+RUN ls
+
+COPY go.mod go.mod
+COPY vendor/ vendor/
+COPY internal/ internal/
+COPY pkg/ pkg/
+COPY cmd/ cmd/
+COPY main/ main/
+COPY utility/ utility/
+
+RUN sed -i 's|#cgo LDFLAGS: -lbrotli.*|&-static -lbrotlicommon-static -lm|' \
+        vendor/github.com/google/brotli/go/cbrotli/cgo.go && \
+    sed -i 's|\(#cgo LDFLAGS:\) .*|\1 -Wl,-Bstatic -llzo2 -Wl,-Bdynamic|' \
+        vendor/github.com/cyberdelia/lzo/lzo.go && \
+    cd main/pg && \
+    go build -mod vendor -race -o wal-g -tags "brotli lzo" -ldflags "-s -w -X main.buildDate=`date -u +%Y.%m.%d_%H:%M:%S`"
+
+FROM wal-g/ubuntu:latest
+
+RUN apt-get update && apt-get install --yes --no-install-recommends --no-install-suggests brotli
+
+COPY --from=build /go/src/github.com/wal-g/wal-g/main/pg/wal-g /usr/bin
+
+COPY docker/st_tests/scripts/ /tmp
+
+CMD /tmp/tests/storage_tool_tests.sh

docker/st_tests/scripts/tests/storage_tool_tests.sh

@@ -0,0 +1,54 @@
+#!/bin/sh
+set -e -x
+
+export WALE_S3_PREFIX=s3://storagetoolsbucket
+
+# Empty list on empty storage
+test "1" -eq "$(wal-g st ls | wc -l)"
+
+# Generate and upload some file to storage
+head -c 100M </dev/urandom >testfile
+wal-g st put testfile testfolder/testfile
+
+# Should not upload the duplicate file by default
+wal-g st put testfile testfolder/testfile && EXIT_STATUS=$? || EXIT_STATUS=$?
+
+if [ "$EXIT_STATUS" -eq 0 ] ; then
+    echo "Error: Duplicate object was uploaded without the -f flag"
+    exit 1
+fi
+
+# Should upload the duplicate file if -f flag is present
+wal-g st put testfile testfolder/testfile -f
+
+wal-g st ls
+# WAL-G should show the uploaded file in the wal-g st ls output
+test "2" -eq "$(wal-g st ls | wc -l)"
+
+# WAL-G should be able to download the uploaded file
+wal-g st get testfolder/testfile.br fetched_testfile
+
+# Downloaded file should be identical to the original one
+diff testfile fetched_testfile
+rm fetched_testfile
+
+# WAL-G should be able to download the uploaded file without decompression
+wal-g st get testfolder/testfile.br uncompressed_testfile.br --no-decompress
+
+brotli --decompress uncompressed_testfile.br
+diff testfile uncompressed_testfile
+rm uncompressed_testfile
+
+# WAL-G should be able to delete the uploaded file
+wal-g st rm testfolder/testfile.br
+
+# Should get empty storage after file removal
+test "1" -eq "$(wal-g st ls | wc -l)"
+
+# Should upload the file uncompressed without error
+wal-g st put testfile testfolder/testfile --no-compress
+
+# Should download the file uncompressed without error
+wal-g st get testfolder/testfile uncompressed_file --no-decompress
+
+diff testfile uncompressed_file

docs/README.md

@@ -158,19 +158,48 @@ If `FIND_FULL` is specified, WAL-G will calculate minimum backup needed to keep
 
 ``target FIND_FULL base_0000000100000000000000C9_D_0000000100000000000000C4`` delete delta backup and all delta backups with the same base backup
 
+**More commands are available for the chosen database engine. See it in [Databases](#databases)**
+
 ## Storage tools (danger zone)
-Storage tools allow interacting with the configured storage. Be aware that these commands can do potentially harmful operations and make sure that you know what you're doing.
+`wal-g st` command series allows interacting with the configured storage. Be aware that these commands can do potentially harmful operations and make sure that you know what you're doing.
 
 ### ``ls``
 Prints listing of the objects in the provided storage folder.
 
+``wal-g st ls`` get listing with all objects in the configured storage.
+
+``wal-g st ls some_folder/some_subfolder`` get listing with all objects in the provided storage path.
+
+### ``get``
+Download the specified storage object. By default, the command will try to apply the decompression and decryption (if configured).
+
+Flags:
+1. Add `--no-decompress` to download the remote object without decompression
+2. Add `--no-decrypt` to download the remote object without decryption
+
 Examples:
 
-``wal-g dh ls`` get listing with all objects in the configured storage.
+``wal-g st get path/to/remote_file path/to/local_file`` download the file from storage.
 
-``wal-g dh ls some_folder/some_subfolder`` get listing with all objects in the provided storage path.
+``wal-g st get path/to/remote_file path/to/local_file --no-decrypt`` download the file from storage without decryption.
 
-**More commands are available for the chosen database engine. See it in [Databases](#databases)**
+### ``rm``
+Remove the specified storage object.
+
+Example:
+
+``wal-g st rm path/to/remote_file`` remove the file from storage.
+
+### ``put``
+Upload the specified file to the storage. By default, the command will try to apply the compression and encryption (if configured).
+
+Flags:
+1. Add `--no-compress` to upload the object without compression
+2. Add `--no-encrypt` to upload the object without encryption
+
+Example:
+
+``wal-g st put path/to/local_file path/to/remote_file`` upload the local file to storage.
 
 Databases
 -----------

internal/compression/compression.go

@@ -19,6 +19,11 @@ func GetDecompressorByCompressor(compressor Compressor) Decompressor {
 }
 
 func FindDecompressor(fileExtension string) Decompressor {
+	// cut the leading '.' (e.g. ".lz4" => "lz4")
+	if len(fileExtension) > 0 && fileExtension[0] == '.' {
+		fileExtension = fileExtension[1:]
+	}
+
 	for _, decompressor := range Decompressors {
 		if decompressor.FileExtension() == fileExtension {
 			return decompressor

internal/fetch_helper.go

@@ -9,10 +9,11 @@ import (
 	"os/user"
 	"path/filepath"
 
+	"github.com/wal-g/wal-g/internal/ioextensions"
+
 	"github.com/pkg/errors"
 	"github.com/wal-g/tracelog"
 	"github.com/wal-g/wal-g/internal/compression"
-	"github.com/wal-g/wal-g/internal/ioextensions"
 	"github.com/wal-g/wal-g/pkg/storages/storage"
 	"github.com/wal-g/wal-g/utility"
 )
@@ -66,29 +67,39 @@ func TryDownloadFile(folder storage.Folder, path string) (walFileReader io.ReadC
 
 // TODO : unit tests
 func DecompressDecryptBytes(dst io.Writer, archiveReader io.ReadCloser, decompressor compression.Decompressor) error {
-	crypter := ConfigureCrypter()
-	if crypter != nil {
-		tracelog.DebugLogger.Printf("Selected crypter: %s", crypter.Name())
-
-		reader, err := crypter.Decrypt(archiveReader)
-		if err != nil {
-			return fmt.Errorf("failed to init decrypt reader: %w", err)
-		}
-		archiveReader = ioextensions.ReadCascadeCloser{
-			Reader: reader,
-			Closer: archiveReader,
-		}
-	} else {
-		tracelog.DebugLogger.Printf("No crypter has been selected")
+	decryptReadCloser, err := DecryptBytes(archiveReader)
+	if err != nil {
+		return err
 	}
 
-	err := decompressor.Decompress(dst, archiveReader)
+	err = decompressor.Decompress(dst, decryptReadCloser)
 	if err != nil {
 		return fmt.Errorf("failed to decompress archive reader: %w", err)
 	}
 	return nil
 }
 
+func DecryptBytes(archiveReader io.ReadCloser) (io.ReadCloser, error) {
+	crypter := ConfigureCrypter()
+	if crypter == nil {
+		tracelog.DebugLogger.Printf("No crypter has been selected")
+		return archiveReader, nil
+	}
+
+	tracelog.DebugLogger.Printf("Selected crypter: %s", crypter.Name())
+
+	decryptReader, err := crypter.Decrypt(archiveReader)
+	if err != nil {
+		return nil, fmt.Errorf("failed to init decrypt reader: %w", err)
+	}
+	decryptReadCloser := ioextensions.ReadCascadeCloser{
+		Reader: decryptReader,
+		Closer: archiveReader,
+	}
+
+	return decryptReadCloser, nil
+}
+
 // CachedDecompressor is the file extension describing decompressor
 type CachedDecompressor struct {
 	FileExtension string