Add bandwidth limit to filesystem operations, rename fields

docs/docs.go

@@ -3003,8 +3003,9 @@ const docTemplate = `{
                 "operation"
             ],
             "properties": {
-                "from": {
-                    "type": "string"
+                "bandwidth_limit_kbit": {
+                    "description": "kbit/s",
+                    "type": "integer"
                 },
                 "operation": {
                     "type": "string",
@@ -3013,7 +3014,10 @@ const docTemplate = `{
                         "move"
                     ]
                 },
-                "to": {
+                "source": {
+                    "type": "string"
+                },
+                "target": {
                     "type": "string"
                 }
             }

docs/swagger.json

@@ -2996,8 +2996,9 @@
                 "operation"
             ],
             "properties": {
-                "from": {
-                    "type": "string"
+                "bandwidth_limit_kbit": {
+                    "description": "kbit/s",
+                    "type": "integer"
                 },
                 "operation": {
                     "type": "string",
@@ -3006,7 +3007,10 @@
                         "move"
                     ]
                 },
-                "to": {
+                "source": {
+                    "type": "string"
+                },
+                "target": {
                     "type": "string"
                 }
             }

docs/swagger.yaml

@@ -493,14 +493,17 @@ definitions:
     type: object
   api.FilesystemOperation:
     properties:
-      from:
-        type: string
+      bandwidth_limit_kbit:
+        description: kbit/s
+        type: integer
       operation:
         enum:
         - copy
         - move
         type: string
-      to:
+      source:
+        type: string
+      target:
         type: string
     required:
     - operation

go.mod

@@ -5,10 +5,12 @@ go 1.18
 require (
 	github.com/99designs/gqlgen v0.17.20
 	github.com/Masterminds/semver/v3 v3.1.1
+	github.com/adhocore/gronx v1.1.2
 	github.com/atrox/haikunatorgo/v2 v2.0.1
 	github.com/caddyserver/certmagic v0.17.2
 	github.com/datarhei/gosrt v0.3.1
 	github.com/datarhei/joy4 v0.0.0-20220914170649-23c70d207759
+	github.com/fujiwara/shapeio v1.0.0
 	github.com/go-playground/validator/v10 v10.11.1
 	github.com/gobwas/glob v0.2.3
 	github.com/golang-jwt/jwt/v4 v4.4.3
@@ -35,7 +37,6 @@ require (
 
 require (
 	github.com/KyleBanks/depth v1.2.1 // indirect
-	github.com/adhocore/gronx v1.1.2 // indirect
 	github.com/agnivade/levenshtein v1.1.1 // indirect
 	github.com/benburkert/openpgp v0.0.0-20160410205803-c2471f86866c // indirect
 	github.com/beorn7/perks v1.0.1 // indirect

go.sum

@@ -43,8 +43,11 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48 h1:fRzb/w+pyskVMQ+UbP35JkH8yB7MYb4q/qhBarqZE6g=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48/go.mod h1:if7Fbed8SFyPtHLHbg49SI7NAdJiC5WIA09pe59rfAA=
+github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/fujiwara/shapeio v1.0.0 h1:xG5D9oNqCSUUbryZ/jQV3cqe1v2suEjwPIcEg1gKM8M=
+github.com/fujiwara/shapeio v1.0.0/go.mod h1:LmEmu6L/8jetyj1oewewFb7bZCNRwE7wLCUNzDLaLVA=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
@@ -351,6 +354,7 @@ golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/time v0.0.0-20201208040808-7e3f01d25324/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
 golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

http/api/filesystems.go

@@ -17,6 +17,7 @@ type FilesystemInfo struct {
 // FilesystemOperation represents a file operation on one or more filesystems
 type FilesystemOperation struct {
 	Operation string `json:"operation" validate:"required" enums:"copy,move" jsonschema:"enum=copy,enum=move"`
-	From      string `json:"from"`
-	To        string `json:"to"`
+	Source    string `json:"source"`
+	Target    string `json:"target"`
+	RateLimit uint64 `json:"bandwidth_limit_kbit"` // kbit/s
 }

http/handler/api/filesystems.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"io"
 	"net/http"
 	"regexp"
 
@@ -8,6 +9,7 @@ import (
 	"github.com/datarhei/core/v16/http/handler"
 	"github.com/datarhei/core/v16/http/handler/util"
 
+	"github.com/fujiwara/shapeio"
 	"github.com/labstack/echo/v4"
 )
 
@@ -207,29 +209,29 @@ func (h *FSHandler) FileOperation(c echo.Context) error {
 
 	rePrefix := regexp.MustCompile(`^(.+):`)
 
-	matches := rePrefix.FindStringSubmatch(operation.From)
+	matches := rePrefix.FindStringSubmatch(operation.Source)
 	if matches == nil {
 		return api.Err(http.StatusBadRequest, "Missing source filesystem prefix")
 	}
 
 	fromFSName := matches[1]
-	fromPath := rePrefix.ReplaceAllString(operation.From, "")
+	fromPath := rePrefix.ReplaceAllString(operation.Source, "")
 	fromFS, ok := h.filesystems[fromFSName]
 	if !ok {
 		return api.Err(http.StatusBadRequest, "Source filesystem not found", "%s", fromFSName)
 	}
 
-	if operation.From == operation.To {
+	if operation.Source == operation.Target {
 		return c.JSON(http.StatusOK, "OK")
 	}
 
-	matches = rePrefix.FindStringSubmatch(operation.To)
+	matches = rePrefix.FindStringSubmatch(operation.Target)
 	if matches == nil {
 		return api.Err(http.StatusBadRequest, "Missing target filesystem prefix")
 	}
 
 	toFSName := matches[1]
-	toPath := rePrefix.ReplaceAllString(operation.To, "")
+	toPath := rePrefix.ReplaceAllString(operation.Target, "")
 	toFS, ok := h.filesystems[toFSName]
 	if !ok {
 		return api.Err(http.StatusBadRequest, "Target filesystem not found", "%s", toFSName)
@@ -242,7 +244,17 @@ func (h *FSHandler) FileOperation(c echo.Context) error {
 
 	defer fromFile.Close()
 
-	_, _, err := toFS.Handler.FS.Filesystem.WriteFileReader(toPath, fromFile)
+	var reader io.Reader = fromFile
+
+	if operation.RateLimit != 0 {
+		ratelimit := float64(operation.RateLimit) * 1024 / 8 // Calculate kbit to bytes
+		shapedReader := shapeio.NewReader(reader)
+		shapedReader.SetRateLimit(ratelimit)
+
+		reader = shapedReader
+	}
+
+	_, _, err := toFS.Handler.FS.Filesystem.WriteFileReader(toPath, reader)
 	if err != nil {
 		toFS.Handler.FS.Filesystem.Remove(toPath)
 		return api.Err(http.StatusBadRequest, "Writing target file failed", "%s", err)

http/handler/api/filesystems_test.go

@@ -445,7 +445,7 @@ func TestFileOperation(t *testing.T) {
 
 	op = api.FilesystemOperation{
 		Operation: "copy",
-		From:      "foo:/elif",
+		Source:    "foo:/elif",
 	}
 
 	jsondata, err = json.Marshal(op)
@@ -455,8 +455,8 @@ func TestFileOperation(t *testing.T) {
 
 	op = api.FilesystemOperation{
 		Operation: "copy",
-		From:      "foo:/elif",
-		To:        "/bar",
+		Source:    "foo:/elif",
+		Target:    "/bar",
 	}
 
 	jsondata, err = json.Marshal(op)
@@ -466,8 +466,8 @@ func TestFileOperation(t *testing.T) {
 
 	op = api.FilesystemOperation{
 		Operation: "copy",
-		From:      "foo:/file",
-		To:        "/bar",
+		Source:    "foo:/file",
+		Target:    "/bar",
 	}
 
 	jsondata, err = json.Marshal(op)
@@ -477,8 +477,8 @@ func TestFileOperation(t *testing.T) {
 
 	op = api.FilesystemOperation{
 		Operation: "copy",
-		From:      "foo:file",
-		To:        "bar:/file",
+		Source:    "foo:file",
+		Target:    "bar:/file",
 	}
 
 	jsondata, err = json.Marshal(op)
@@ -496,8 +496,8 @@ func TestFileOperation(t *testing.T) {
 
 	op = api.FilesystemOperation{
 		Operation: "move",
-		From:      "foo:file",
-		To:        "bar:/file",
+		Source:    "foo:file",
+		Target:    "bar:/file",
 	}
 
 	jsondata, err = json.Marshal(op)

vendor/github.com/fujiwara/shapeio/.gitignore

@@ -0,0 +1,24 @@
+# Compiled Object files, Static and Dynamic libs (Shared Objects)
+*.o
+*.a
+*.so
+
+# Folders
+_obj
+_test
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+*.exe
+*.test
+*.prof

vendor/github.com/fujiwara/shapeio/LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2016 FUJIWARA Shunichiro
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

vendor/github.com/fujiwara/shapeio/README.md

@@ -0,0 +1,93 @@
+# shapeio
+
+Traffic shaper for Golang io.Reader and io.Writer
+
+```go
+import "github.com/fujiwara/shapeio"
+
+func ExampleReader() {
+	// example for downloading http body with rate limit.
+	resp, _ := http.Get("http://example.com")
+	defer resp.Body.Close()
+
+	reader := shapeio.NewReader(resp.Body)
+	reader.SetRateLimit(1024 * 10) // 10KB/sec
+	io.Copy(ioutil.Discard, reader)
+}
+
+func ExampleWriter() {
+	// example for writing file with rate limit.
+	src := bytes.NewReader(bytes.Repeat([]byte{0}, 32*1024)) // 32KB
+	f, _ := os.Create("/tmp/foo")
+	writer := shapeio.NewWriter(f)
+	writer.SetRateLimit(1024 * 10) // 10KB/sec
+	io.Copy(writer, src)
+	f.Close()
+}
+```
+
+## Usage
+
+#### type Reader
+
+```go
+type Reader struct {
+}
+```
+
+
+#### func  NewReader
+
+```go
+func NewReader(r io.Reader) *Reader
+```
+NewReader returns a reader that implements io.Reader with rate limiting.
+
+#### func (*Reader) Read
+
+```go
+func (s *Reader) Read(p []byte) (int, error)
+```
+Read reads bytes into p.
+
+#### func (*Reader) SetRateLimit
+
+```go
+func (s *Reader) SetRateLimit(l float64)
+```
+SetRateLimit sets rate limit (bytes/sec) to the reader.
+
+#### type Writer
+
+```go
+type Writer struct {
+}
+```
+
+
+#### func  NewWriter
+
+```go
+func NewWriter(w io.Writer) *Writer
+```
+NewWriter returns a writer that implements io.Writer with rate limiting.
+
+#### func (*Writer) SetRateLimit
+
+```go
+func (s *Writer) SetRateLimit(l float64)
+```
+SetRateLimit sets rate limit (bytes/sec) to the writer.
+
+#### func (*Writer) Write
+
+```go
+func (s *Writer) Write(p []byte) (int, error)
+```
+Write writes bytes from p.
+
+##  License
+
+The MIT License (MIT)
+
+Copyright (c) 2016 FUJIWARA Shunichiro

vendor/github.com/fujiwara/shapeio/shapeio.go

@@ -0,0 +1,97 @@
+package shapeio
+
+import (
+	"context"
+	"io"
+	"time"
+
+	"golang.org/x/time/rate"
+)
+
+const burstLimit = 1000 * 1000 * 1000
+
+type Reader struct {
+	r       io.Reader
+	limiter *rate.Limiter
+	ctx     context.Context
+}
+
+type Writer struct {
+	w       io.Writer
+	limiter *rate.Limiter
+	ctx     context.Context
+}
+
+// NewReader returns a reader that implements io.Reader with rate limiting.
+func NewReader(r io.Reader) *Reader {
+	return &Reader{
+		r:   r,
+		ctx: context.Background(),
+	}
+}
+
+// NewReaderWithContext returns a reader that implements io.Reader with rate limiting.
+func NewReaderWithContext(r io.Reader, ctx context.Context) *Reader {
+	return &Reader{
+		r:   r,
+		ctx: ctx,
+	}
+}
+
+// NewWriter returns a writer that implements io.Writer with rate limiting.
+func NewWriter(w io.Writer) *Writer {
+	return &Writer{
+		w:   w,
+		ctx: context.Background(),
+	}
+}
+
+// NewWriterWithContext returns a writer that implements io.Writer with rate limiting.
+func NewWriterWithContext(w io.Writer, ctx context.Context) *Writer {
+	return &Writer{
+		w:   w,
+		ctx: ctx,
+	}
+}
+
+// SetRateLimit sets rate limit (bytes/sec) to the reader.
+func (s *Reader) SetRateLimit(bytesPerSec float64) {
+	s.limiter = rate.NewLimiter(rate.Limit(bytesPerSec), burstLimit)
+	s.limiter.AllowN(time.Now(), burstLimit) // spend initial burst
+}
+
+// Read reads bytes into p.
+func (s *Reader) Read(p []byte) (int, error) {
+	if s.limiter == nil {
+		return s.r.Read(p)
+	}
+	n, err := s.r.Read(p)
+	if err != nil {
+		return n, err
+	}
+	if err := s.limiter.WaitN(s.ctx, n); err != nil {
+		return n, err
+	}
+	return n, nil
+}
+
+// SetRateLimit sets rate limit (bytes/sec) to the writer.
+func (s *Writer) SetRateLimit(bytesPerSec float64) {
+	s.limiter = rate.NewLimiter(rate.Limit(bytesPerSec), burstLimit)
+	s.limiter.AllowN(time.Now(), burstLimit) // spend initial burst
+}
+
+// Write writes bytes from p.
+func (s *Writer) Write(p []byte) (int, error) {
+	if s.limiter == nil {
+		return s.w.Write(p)
+	}
+	n, err := s.w.Write(p)
+	if err != nil {
+		return n, err
+	}
+	if err := s.limiter.WaitN(s.ctx, n); err != nil {
+		return n, err
+	}
+	return n, err
+}

vendor/modules.txt

@@ -91,6 +91,9 @@ github.com/davecgh/go-spew/spew
 # github.com/dustin/go-humanize v1.0.1
 ## explicit; go 1.16
 github.com/dustin/go-humanize
+# github.com/fujiwara/shapeio v1.0.0
+## explicit; go 1.16
+github.com/fujiwara/shapeio
 # github.com/go-ole/go-ole v1.2.6
 ## explicit; go 1.12
 github.com/go-ole/go-ole