Merge pull request #4256 from aduffeck/cleanup-uploads

Add a cli for listing and cleaning up expired uploads

services/storage-users/pkg/command/root.go

@@ -18,6 +18,7 @@ func GetCommands(cfg *config.Config) cli.Commands {
 		Server(cfg),
 
 		// interaction with this service
+		Uploads(cfg),
 
 		// infos about this service
 		Health(cfg),

services/storage-users/pkg/command/uploads.go

@@ -0,0 +1,150 @@
+package command
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+
+	tusd "github.com/tus/tusd/pkg/handler"
+	"github.com/urfave/cli/v2"
+
+	"github.com/cs3org/reva/v2/pkg/storage"
+	"github.com/cs3org/reva/v2/pkg/storage/fs/registry"
+	"github.com/owncloud/ocis/v2/services/storage-users/pkg/config"
+	"github.com/owncloud/ocis/v2/services/storage-users/pkg/config/parser"
+	"github.com/owncloud/ocis/v2/services/storage-users/pkg/revaconfig"
+)
+
+func Uploads(cfg *config.Config) *cli.Command {
+	return &cli.Command{
+		Name:  "uploads",
+		Usage: "manage uploads",
+		Before: func(c *cli.Context) error {
+			if err := parser.ParseConfig(cfg); err != nil {
+				fmt.Printf("%v", err)
+				return err
+			}
+			return nil
+		},
+		Subcommands: []*cli.Command{
+			ListUploads(cfg),
+			PurgeExpiredUploads(cfg),
+		},
+	}
+}
+
+// ListUploads prints a list of all incomplete uploads
+func ListUploads(cfg *config.Config) *cli.Command {
+	return &cli.Command{
+		Name:     "list",
+		Usage:    fmt.Sprintf("Print a list of all incomplete uploads"),
+		Category: "services",
+		Before: func(c *cli.Context) error {
+			err := parser.ParseConfig(cfg)
+			if err != nil {
+				fmt.Printf("%v", err)
+				os.Exit(1)
+			}
+			return err
+		},
+		Action: func(c *cli.Context) error {
+			f, ok := registry.NewFuncs[cfg.Driver]
+			if !ok {
+				fmt.Fprintf(os.Stderr, "Unknown filesystem driver '%s'\n", cfg.Driver)
+				os.Exit(1)
+			}
+			drivers := revaconfig.UserDrivers(cfg)
+			fs, err := f(drivers[cfg.Driver].(map[string]interface{}))
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Failed to initialize filesystem driver '%s'\n", cfg.Driver)
+				return err
+			}
+
+			managingFS, ok := fs.(storage.UploadsManager)
+			if !ok {
+				fmt.Fprintf(os.Stderr, "'%s' storage does not support listing expired uploads\n", cfg.Driver)
+				os.Exit(1)
+			}
+
+			uploads, err := managingFS.ListUploads()
+			if err != nil {
+				return err
+			}
+
+			fmt.Println("Incomplete uploads:")
+			for _, u := range uploads {
+				fmt.Printf(" - %s (%s, Size: %d, Expires: %s)\n", u.ID, u.MetaData["filename"], u.Size, expiredString(u.MetaData["expires"]))
+			}
+			return nil
+		},
+	}
+}
+
+// PurgeExpiredUploads is the entry point for the server command.
+func PurgeExpiredUploads(cfg *config.Config) *cli.Command {
+	return &cli.Command{
+		Name:     "purge",
+		Usage:    fmt.Sprintf("Let %s extension clean up leftovers from expired downloads", cfg.Service.Name),
+		Category: "services",
+		Before: func(c *cli.Context) error {
+			err := parser.ParseConfig(cfg)
+			if err != nil {
+				fmt.Printf("%v", err)
+				os.Exit(1)
+			}
+			return err
+		},
+		Action: func(c *cli.Context) error {
+			f, ok := registry.NewFuncs[cfg.Driver]
+			if !ok {
+				fmt.Fprintf(os.Stderr, "Unknown filesystem driver '%s'\n", cfg.Driver)
+				os.Exit(1)
+			}
+			drivers := revaconfig.UserDrivers(cfg)
+			fs, err := f(drivers[cfg.Driver].(map[string]interface{}))
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Failed to initialize filesystem driver '%s'\n", cfg.Driver)
+				return err
+			}
+
+			managingFS, ok := fs.(storage.UploadsManager)
+			if !ok {
+				fmt.Fprintf(os.Stderr, "'%s' storage does not support purging expired uploads\n", cfg.Driver)
+				os.Exit(1)
+			}
+
+			wg := sync.WaitGroup{}
+			wg.Add(1)
+			purgedChannel := make(chan tusd.FileInfo)
+
+			go func() {
+				for purged := range purgedChannel {
+					fmt.Printf("Purging %s (Filename: %s, Size: %d, Expires: %s)\n",
+						purged.ID, purged.MetaData["filename"], purged.Size, expiredString(purged.MetaData["expires"]))
+
+				}
+				wg.Done()
+			}()
+
+			err = managingFS.PurgeExpiredUploads(purgedChannel)
+			close(purgedChannel)
+			wg.Wait()
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Failed to purge expired uploads '%s'\n", err)
+				return err
+			}
+			return nil
+		},
+	}
+}
+
+func expiredString(e string) string {
+	expired := "N/A"
+	iExpires, err := strconv.Atoi(e)
+	if err == nil {
+		expired = time.Unix(int64(iExpires), 0).Format(time.RFC3339)
+	}
+	return expired
+}

services/storage-users/pkg/config/config.go

@@ -28,6 +28,7 @@ type Config struct {
 	MountID          string  `yaml:"mount_id" env:"STORAGE_USERS_MOUNT_ID" desc:"Mount ID of this storage."`
 	ExposeDataServer bool    `yaml:"expose_data_server" env:"STORAGE_USERS_EXPOSE_DATA_SERVER" desc:"Exposes the data server directly to users and bypasses the data gateway. Ensure that the data server address is reachable by users."`
 	ReadOnly         bool    `yaml:"readonly" env:"STORAGE_USERS_READ_ONLY" desc:"Set this storage to be read-only."`
+	UploadExpiration int64   `yaml:"upload_expiration" env:"STORAGE_USERS_UPLOAD_EXPIRATION" desc:"Duration after which uploads will expire."`
 
 	Supervised bool            `yaml:"-"`
 	Context    context.Context `yaml:"-"`

services/storage-users/pkg/config/defaults/defaultconfig.go

@@ -39,9 +39,10 @@ func DefaultConfig() *config.Config {
 		Reva: &config.Reva{
 			Address: "127.0.0.1:9142",
 		},
-		DataServerURL: "http://localhost:9158/data",
-		MountID:       "1284d238-aa92-42ce-bdc4-0b0000009157",
-		Driver:        "ocis",
+		DataServerURL:    "http://localhost:9158/data",
+		MountID:          "1284d238-aa92-42ce-bdc4-0b0000009157",
+		UploadExpiration: 24 * 60 * 60,
+		Driver:           "ocis",
 		Drivers: config.Drivers{
 			OwnCloudSQL: config.OwnCloudSQLDriver{
 				Root:                  filepath.Join(defaults.BaseDataPath(), "storage", "owncloud"),

services/storage-users/pkg/revaconfig/config.go

@@ -29,6 +29,7 @@ func StorageUsersConfigFromStruct(cfg *config.Config) map[string]interface{} {
 					"mount_id":           cfg.MountID,
 					"expose_data_server": cfg.ExposeDataServer,
 					"data_server_url":    cfg.DataServerURL,
+					"upload_expiration":  cfg.UploadExpiration,
 				},
 			},
 			"interceptors": map[string]interface{}{