Merge 1b9e89b478 into faffd15d13

Add support for specifying --host via environment variable

changelog/unreleased/issue-4733

@@ -0,0 +1,10 @@
+Enhancement: Allow specifying `--host` via environment variable
+
+Restic commands that operate on snapshots, such as `restic backup` and
+`restic snapshots`, support the `--host` flag to specify the hostname for
+grouoping snapshots. They now permit selecting the hostname via the
+environment variable `RESTIC_HOST`. `--host` still takes precedence over the
+environment variable.
+
+https://github.com/restic/restic/issues/4733
+https://github.com/restic/restic/pull/4734

cmd/restic/cmd_backup.go

@@ -114,7 +114,7 @@ func init() {
 	f.BoolVar(&backupOptions.StdinCommand, "stdin-from-command", false, "interpret arguments as command to execute and store its stdout")
 	f.Var(&backupOptions.Tags, "tag", "add `tags` for the new snapshot in the format `tag[,tag,...]` (can be specified multiple times)")
 	f.UintVar(&backupOptions.ReadConcurrency, "read-concurrency", 0, "read `n` files concurrently (default: $RESTIC_READ_CONCURRENCY or 2)")
-	f.StringVarP(&backupOptions.Host, "host", "H", "", "set the `hostname` for the snapshot manually. To prevent an expensive rescan use the \"parent\" flag")
+	f.StringVarP(&backupOptions.Host, "host", "H", "", "set the `hostname` for the snapshot manually (default: $RESTIC_HOST). To prevent an expensive rescan use the \"parent\" flag")
 	f.StringVar(&backupOptions.Host, "hostname", "", "set the `hostname` for the snapshot manually")
 	err := f.MarkDeprecated("hostname", "use --host")
 	if err != nil {
@@ -137,6 +137,11 @@ func init() {
 	// parse read concurrency from env, on error the default value will be used
 	readConcurrency, _ := strconv.ParseUint(os.Getenv("RESTIC_READ_CONCURRENCY"), 10, 32)
 	backupOptions.ReadConcurrency = uint(readConcurrency)
+
+	// parse host from env, if not exists or empty the default value will be used
+	if host := os.Getenv("RESTIC_HOST"); host != "" {
+		backupOptions.Host = host
+	}
 }
 
 // filterExisting returns a slice of all existing items, or an error if no

cmd/restic/find.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"os"
 
 	"github.com/restic/restic/internal/restic"
 	"github.com/spf13/pflag"
@@ -14,17 +15,27 @@ func initMultiSnapshotFilter(flags *pflag.FlagSet, filt *restic.SnapshotFilter,
 	if !addHostShorthand {
 		hostShorthand = ""
 	}
-	flags.StringArrayVarP(&filt.Hosts, "host", hostShorthand, nil, "only consider snapshots for this `host` (can be specified multiple times)")
+	flags.StringArrayVarP(&filt.Hosts, "host", hostShorthand, nil, "only consider snapshots for this `host` (can be specified multiple times) (default: $RESTIC_HOST)")
 	flags.Var(&filt.Tags, "tag", "only consider snapshots including `tag[,tag,...]` (can be specified multiple times)")
 	flags.StringArrayVar(&filt.Paths, "path", nil, "only consider snapshots including this (absolute) `path` (can be specified multiple times)")
+
+	// set default based on env if set
+	if host := os.Getenv("RESTIC_HOST"); host != "" {
+		filt.Hosts = []string{host}
+	}
 }
 
 // initSingleSnapshotFilter is used for commands that work on a single snapshot
 // MUST be combined with restic.FindFilteredSnapshot
 func initSingleSnapshotFilter(flags *pflag.FlagSet, filt *restic.SnapshotFilter) {
-	flags.StringArrayVarP(&filt.Hosts, "host", "H", nil, "only consider snapshots for this `host`, when snapshot ID \"latest\" is given (can be specified multiple times)")
+	flags.StringArrayVarP(&filt.Hosts, "host", "H", nil, "only consider snapshots for this `host`, when snapshot ID \"latest\" is given (can be specified multiple times) (default: $RESTIC_HOST)")
 	flags.Var(&filt.Tags, "tag", "only consider snapshots including `tag[,tag,...]`, when snapshot ID \"latest\" is given (can be specified multiple times)")
 	flags.StringArrayVar(&filt.Paths, "path", nil, "only consider snapshots including this (absolute) `path`, when snapshot ID \"latest\" is given (can be specified multiple times)")
+
+	// set default based on env if set
+	if host := os.Getenv("RESTIC_HOST"); host != "" {
+		filt.Hosts = []string{host}
+	}
 }
 
 // FindFilteredSnapshots yields Snapshots, either given explicitly by `snapshotIDs` or filtered from the list of all snapshots.

cmd/restic/find_test.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/restic/restic/internal/restic"
+	rtest "github.com/restic/restic/internal/test"
+	"github.com/spf13/pflag"
+)
+
+func TestSnapshotFilter(t *testing.T) {
+	for _, test := range []struct {
+		name     string
+		args     []string
+		expected []string
+		env      string
+	}{
+		{
+			"no value",
+			[]string{},
+			nil,
+			"",
+		},
+		{
+			"args only",
+			[]string{"--host", "abc"},
+			[]string{"abc"},
+			"",
+		},
+		{
+			"env default",
+			[]string{},
+			[]string{"def"},
+			"def",
+		},
+		{
+			"both",
+			[]string{"--host", "abc"},
+			[]string{"abc"},
+			"def",
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			t.Setenv("RESTIC_HOST", test.env)
+
+			for _, mode := range []bool{false, true} {
+				set := pflag.NewFlagSet("test", pflag.PanicOnError)
+				flt := &restic.SnapshotFilter{}
+				if mode {
+					initMultiSnapshotFilter(set, flt, false)
+				} else {
+					initSingleSnapshotFilter(set, flt)
+				}
+				err := set.Parse(test.args)
+				rtest.OK(t, err)
+
+				rtest.Equals(t, test.expected, flt.Hosts, "unexpected hosts")
+			}
+		})
+	}
+}

internal/backend/http_transport.go

@@ -1,6 +1,7 @@
 package backend
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/pem"
@@ -13,6 +14,7 @@ import (
 	"github.com/peterbourgon/unixtransport"
 	"github.com/restic/restic/internal/debug"
 	"github.com/restic/restic/internal/errors"
+	"github.com/restic/restic/internal/feature"
 )
 
 // TransportOptions collects various options which can be set for an HTTP based
@@ -66,14 +68,28 @@ func readPEMCertKey(filename string) (certs []byte, key []byte, err error) {
 // a custom rootCertFilename is non-empty, it must point to a valid PEM file,
 // otherwise the function will return an error.
 func Transport(opts TransportOptions) (http.RoundTripper, error) {
+	dial := (&net.Dialer{
+		Timeout:   30 * time.Second,
+		KeepAlive: 30 * time.Second,
+		DualStack: true,
+	}).DialContext
+	dialTimeout := dial
+
+	if feature.Flag.Enabled(feature.HTTPTimeouts) {
+		// inject timeoutConn to enforce progress
+		dialTimeout = func(ctx context.Context, network, addr string) (net.Conn, error) {
+			conn, err := dial(ctx, network, addr)
+			if err != nil {
+				return conn, err
+			}
+			return newTimeoutConn(conn, 5*time.Minute)
+		}
+	}
+
 	// copied from net/http
 	tr := &http.Transport{
-		Proxy: http.ProxyFromEnvironment,
-		DialContext: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
-			DualStack: true,
-		}).DialContext,
+		Proxy:                 http.ProxyFromEnvironment,
+		DialContext:           dialTimeout,
 		ForceAttemptHTTP2:     true,
 		MaxIdleConns:          100,
 		MaxIdleConnsPerHost:   100,

internal/backend/timeout_transport.go

@@ -0,0 +1,172 @@
+package backend
+
+import (
+	"net"
+	"os"
+	"sync"
+	"time"
+)
+
+// timeoutConn will timeout if no read or write progress is made for progressTimeout.
+// This ensures that stuck network connections are interrupted after some time.
+// By using a timeoutConn within a http transport (via DialContext), sending / receing
+// the request / response body is guarded with a timeout. The read progress part also
+// limits the time until a response header must be received.
+//
+// The progressTimeout must be larger than the IdleConnTimeout of the http transport.
+//
+// The http2.Transport offers a similar functionality via WriteByteTimeout & ReadIdleTimeout.
+// However, those are not available for HTTP/1 connections. Thus, there's no builtin way to
+// enforce progress for sending the request body or reading the response body.
+// See https://github.com/restic/restic/issues/4193#issuecomment-2067988727 for details.
+type timeoutConn struct {
+	conn net.Conn
+	// timeout within which a read/write must make progress, otherwise a connection is considered broken
+	// if no read/write is pending, then the timeout is inactive
+	progressTimeout time.Duration
+
+	// all access to fields below must hold m
+	m sync.Mutex
+
+	// user defined read/write deadline
+	readDeadline  time.Time
+	writeDeadline time.Time
+	// timestamp of last successful write (at least one byte)
+	lastWrite time.Time
+}
+
+var _ net.Conn = &timeoutConn{}
+
+func newTimeoutConn(conn net.Conn, progressTimeout time.Duration) (*timeoutConn, error) {
+	// reset timeouts to ensure a consistent state
+	err := conn.SetDeadline(time.Time{})
+	if err != nil {
+		return nil, err
+	}
+
+	return &timeoutConn{
+		conn:            conn,
+		progressTimeout: progressTimeout,
+	}, nil
+}
+
+func (t *timeoutConn) Write(p []byte) (n int, err error) {
+	t.m.Lock()
+	timeout := t.writeDeadline
+	t.m.Unlock()
+	var zero time.Time
+	if timeout != zero {
+		// fall back to standard behavior if a timeout was set explicitly
+		n, err := t.conn.Write(p)
+		if n > 0 {
+			t.m.Lock()
+			t.lastWrite = time.Now()
+			t.m.Unlock()
+		}
+		return n, err
+	}
+
+	// based on http2stickyErrWriter.Write from go/src/net/http/h2_bundle.go
+	for {
+		_ = t.conn.SetWriteDeadline(time.Now().Add(t.progressTimeout))
+
+		nn, err := t.conn.Write(p[n:])
+		n += nn
+		if nn > 0 {
+			// track write progress
+			t.m.Lock()
+			t.lastWrite = time.Now()
+			t.m.Unlock()
+		}
+
+		if n < len(p) && nn > 0 && err == os.ErrDeadlineExceeded {
+			// some data is still left to send, keep going as long as there is some progress
+			continue
+		}
+
+		t.m.Lock()
+		// restore configured deadline
+		_ = t.conn.SetWriteDeadline(t.writeDeadline)
+		t.m.Unlock()
+		return n, err
+	}
+}
+
+func (t *timeoutConn) Read(b []byte) (n int, err error) {
+	t.m.Lock()
+	timeout := t.readDeadline
+	t.m.Unlock()
+	var zero time.Time
+	if timeout != zero {
+		// fall back to standard behavior if a timeout was set explicitly
+		return t.conn.Read(b)
+	}
+
+	var start = time.Now()
+
+	for {
+		_ = t.conn.SetReadDeadline(start.Add(t.progressTimeout))
+
+		nn, err := t.conn.Read(b)
+		t.m.Lock()
+		lastWrite := t.lastWrite
+		t.m.Unlock()
+		if nn == 0 && err == os.ErrDeadlineExceeded && lastWrite.After(start) {
+			// deadline exceeded, but write made some progress in the meantime
+			start = lastWrite
+			continue
+		}
+
+		t.m.Lock()
+		// restore configured deadline
+		_ = t.conn.SetReadDeadline(t.readDeadline)
+		t.m.Unlock()
+		return nn, err
+	}
+}
+
+func (t *timeoutConn) Close() error {
+	return t.conn.Close()
+}
+
+func (t *timeoutConn) LocalAddr() net.Addr {
+	return t.conn.LocalAddr()
+}
+
+func (t *timeoutConn) RemoteAddr() net.Addr {
+	return t.conn.RemoteAddr()
+}
+
+func (t *timeoutConn) SetDeadline(d time.Time) error {
+	err := t.SetReadDeadline(d)
+	err2 := t.SetWriteDeadline(d)
+	if err != nil {
+		return err
+	}
+	return err2
+}
+
+func (t *timeoutConn) SetReadDeadline(d time.Time) error {
+	t.m.Lock()
+	defer t.m.Unlock()
+
+	// track timeout modifications, as the current timeout cannot be queried
+	err := t.conn.SetReadDeadline(d)
+	if err != nil {
+		return err
+	}
+	t.readDeadline = d
+	return nil
+}
+
+func (t *timeoutConn) SetWriteDeadline(d time.Time) error {
+	t.m.Lock()
+	defer t.m.Unlock()
+
+	err := t.conn.SetWriteDeadline(d)
+	if err != nil {
+		return err
+	}
+	t.writeDeadline = d
+	return nil
+}

internal/feature/registry.go

@@ -8,6 +8,7 @@ const (
 	DeprecateLegacyIndex    FlagName = "deprecate-legacy-index"
 	DeprecateS3LegacyLayout FlagName = "deprecate-s3-legacy-layout"
 	DeviceIDForHardlinks    FlagName = "device-id-for-hardlinks"
+	HTTPTimeouts            FlagName = "http-timeouts"
 )
 
 func init() {
@@ -15,5 +16,6 @@ func init() {
 		DeprecateLegacyIndex:    {Type: Beta, Description: "disable support for index format used by restic 0.1.0. Use `restic repair index` to update the index if necessary."},
 		DeprecateS3LegacyLayout: {Type: Beta, Description: "disable support for S3 legacy layout used up to restic 0.7.0. Use `RESTIC_FEATURES=deprecate-s3-legacy-layout=false restic migrate s3_layout` to migrate your S3 repository if necessary."},
 		DeviceIDForHardlinks:    {Type: Alpha, Description: "store deviceID only for hardlinks to reduce metadata changes for example when using btrfs subvolumes. Will be removed in a future restic version after repository format 3 is available"},
+		HTTPTimeouts:            {Type: Beta, Description: "improve handling of stuck HTTP connections using timeouts."},
 	})
 }