changed env GOPROXY_*_PORT to GOPROXY_*_ADDR, changed api server default to listen on localhost only, readme update

.gitignore

@@ -1,7 +1,7 @@
 compose.yml
 
-config/
-certs/
+config*/
+certs*/
 bin/
 
 templates/codemirror/
@@ -13,6 +13,6 @@ log/
 
 go.work.sum
 
-!src/config/
+!src/**/
 
 todo.md

README.md

@@ -54,7 +54,9 @@ A lightweight, easy-to-use, and [performant](docs/benchmark_result.md) reverse p
 
 2. Setup `go-proxy` [See here](docs/docker.md)
 
-3. Configure `go-proxy`
+3. Setup `docker-socket-proxy` (see [example](docs/docker_socket_proxy.md) other machine that is running docker (if any)
+
+4. Configure `go-proxy`
    - with text editor (e.g. Visual Studio Code)
    - or with web config editor via `http://gp.y.z`
 
@@ -74,13 +76,13 @@ A lightweight, easy-to-use, and [performant](docs/benchmark_result.md) reverse p
 
 ### Environment variables
 
-| Environment Variable           | Description                   | Default | Values  |
-| ------------------------------ | ----------------------------- | ------- | ------- |
-| `GOPROXY_NO_SCHEMA_VALIDATION` | disable schema validation     | `false` | boolean |
-| `GOPROXY_DEBUG`                | enable debug behaviors        | `false` | boolean |
-| `GOPROXY_HTTP_PORT`            | http server port              | `80`    | integer |
-| `GOPROXY_HTTPS_PORT`           | http server port (if enabled) | `443`   | integer |
-| `GOPROXY_API_PORT`             | api server port               | `8888`  | integer |
+| Environment Variable           | Description                                 | Default          | Values        |
+| ------------------------------ | ------------------------------------------- | ---------------- | ------------- |
+| `GOPROXY_NO_SCHEMA_VALIDATION` | disable schema validation                   | `false`          | boolean       |
+| `GOPROXY_DEBUG`                | enable debug behaviors                      | `false`          | boolean       |
+| `GOPROXY_HTTP_ADDR`            | http server listening address               | `:80`            | `[host]:port` |
+| `GOPROXY_HTTPS_ADDR`           | https server listening address (if enabled) | `:443`           | `[host]:port` |
+| `GOPROXY_API_ADDR`             | api server listening address                | `127.0.0.1:8888` | `[host]:port` |
 
 ### Use JSON Schema in VSCode
 
@@ -124,8 +126,6 @@ See [providers.example.yml](providers.example.yml) for examples
 
 ## Known issues
 
-- Cert "renewal" is actually obtaining a new cert instead of renewing the existing one
-
 - `autocert` config is not hot-reloadable
 
 [🔼Back to top](#table-of-content)

README_CHT.md

@@ -6,7 +6,7 @@
 [![Maintainability Rating](https://sonarcloud.io/api/project_badges/measure?project=yusing_go-proxy&metric=sqale_rating)](https://sonarcloud.io/summary/new_code?id=yusing_go-proxy)
 [![Vulnerabilities](https://sonarcloud.io/api/project_badges/measure?project=yusing_go-proxy&metric=vulnerabilities)](https://sonarcloud.io/summary/new_code?id=yusing_go-proxy)
 
-一個輕量化、易用且[高效](docs/benchmark_result.md)的反向代理工具
+一個輕量化、易用且[高效](docs/benchmark_result.md)的反向代理和端口轉發工具
 
 ## 目錄
 
@@ -72,10 +72,13 @@
 
 ### 環境變量
 
-| 環境變量                       | 描述             | 默認    | 值      |
-| ------------------------------ | ---------------- | ------- | ------- |
-| `GOPROXY_NO_SCHEMA_VALIDATION` | 禁用 schema 驗證 | `false` | boolean |
-| `GOPROXY_DEBUG`                | 啟用調試輸出     | `false` | boolean |
+| 環境變量                       | 描述             | 默認             | 格式          |
+| ------------------------------ | ---------------- | ---------------- | ------------- |
+| `GOPROXY_NO_SCHEMA_VALIDATION` | 禁用 schema 驗證 | `false`          | boolean       |
+| `GOPROXY_DEBUG`                | 啟用調試輸出     | `false`          | boolean       |
+| `GOPROXY_HTTP_ADDR`            | http 收聽地址    | `:80`            | `[host]:port` |
+| `GOPROXY_HTTPS_ADDR`           | https 收聽地址   | `:443`           | `[host]:port` |
+| `GOPROXY_API_ADDR`             | api 收聽地址     | `127.0.0.1:8888` | `[host]:port` |
 
 ### VSCode 中使用 JSON Schema
 
@@ -119,8 +122,6 @@ providers:
 
 ## 已知問題
 
-- 證書“更新”實際上是獲取新證書而不是更新現有證書
-
 - `autocert` 配置不能熱重載
 
 [🔼 返回頂部](#目錄)

docs/docker.md

@@ -95,7 +95,7 @@
 | `proxy.stop_timeout`     | time to wait for stop command                                         | `10s`                | `number[unit]...`                                                         |
 | `proxy.stop_signal`      | signal sent to container for `stop` and `kill` methods                | docker's default     | `SIGINT`, `SIGTERM`, `SIGHUP`, `SIGQUIT` and those without **SIG** prefix |
 | `proxy.<alias>.<field>`  | set field for specific alias                                          | N/A                  | N/A                                                                       |
-| `proxy.$<index>.<field>` | set field for specific alias at index (starting from **1**)            | N/A                  | N/A                                                                       |
+| `proxy.$<index>.<field>` | set field for specific alias at index (starting from **1**)           | N/A                  | N/A                                                                       |
 | `proxy.*.<field>`        | set field for all aliases                                             | N/A                  | N/A                                                                       |
 
 ### Fields
@@ -215,6 +215,8 @@ service_a:
 
 ## Docker compose examples
 
+More examples in [here](examples/)
+
 ```yaml
 volumes:
   adg-work:

docs/docker_socket_proxy.md

@@ -0,0 +1,59 @@
+## Docker Socket Proxy
+
+For docker client on other machine, set this up, then add `name: tcp://<machine_ip>:2375` to `config.yml` under `docker` section
+
+```yml
+# compose.yml on remote machine (e.g. server1)
+services:
+  docker-proxy:
+    container_name: docker-proxy
+    image: ghcr.io/linuxserver/socket-proxy
+    environment:
+      - ALLOW_START=1 #optional
+      - ALLOW_STOP=1 #optional
+      - ALLOW_RESTARTS=0 #optional
+      - AUTH=0 #optional
+      - BUILD=0 #optional
+      - COMMIT=0 #optional
+      - CONFIGS=0 #optional
+      - CONTAINERS=1 #optional
+      - DISABLE_IPV6=1 #optional
+      - DISTRIBUTION=0 #optional
+      - EVENTS=1 #optional
+      - EXEC=0 #optional
+      - IMAGES=0 #optional
+      - INFO=0 #optional
+      - NETWORKS=0 #optional
+      - NODES=0 #optional
+      - PING=1 #optional
+      - POST=1 #optional
+      - PLUGINS=0 #optional
+      - SECRETS=0 #optional
+      - SERVICES=0 #optional
+      - SESSION=0 #optional
+      - SWARM=0 #optional
+      - SYSTEM=0 #optional
+      - TASKS=0 #optional
+      - VERSION=1 #optional
+      - VOLUMES=0 #optional
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+    restart: always
+    tmpfs:
+      - /run
+    ports:
+      - 2375:2375
+```
+
+```yml
+# config.yml on go-proxy machine
+autocert:
+    ... # your config
+
+providers:
+    include:
+        ...
+    docker:
+        ...
+        server1: tcp://<machine_ip>:2375
+```

examples/microbin.yml

@@ -0,0 +1,16 @@
+services:
+  app:
+    container_name: microbin
+    cpu_shares: 10
+    deploy:
+      resources:
+        limits:
+          memory: 256M
+    env_file: .env
+    image: docker.i.sh/danielszabo99/microbin:latest
+    ports:
+      - 8080
+    restart: unless-stopped
+    volumes:
+      - ./data:/app/microbin_data
+# microbin.domain.tld

examples/siyuan.yml

@@ -0,0 +1,16 @@
+services:
+  main:
+    image: b3log/siyuan:v3.1.0
+    container_name: siyuan
+    command:
+      - --workspace=/siyuan/workspace/
+      - --accessAuthCode=<some password>
+    user: 1000:1000
+    volumes:
+      - ./workspace:/siyuan/workspace
+    restart: unless-stopped
+    environment:
+      - TZ=Asia/Hong_Kong
+    ports:
+      - 6806
+# siyuan.domain.tld

src/api/v1/utils/net.go

@@ -36,7 +36,7 @@ func IsStreamHealthy(scheme, address string) bool {
 }
 
 func ReloadServer() E.NestedError {
-	resp, err := HttpClient.Post(fmt.Sprintf("http://localhost%v/reload", common.APIHTTPPort), "", nil)
+	resp, err := HttpClient.Post(fmt.Sprintf("http://localhost%v/reload", common.APIHTTPAddr), "", nil)
 	if err != nil {
 		return E.From(err)
 	}

src/common/env.go

@@ -9,9 +9,9 @@ import (
 var (
 	NoSchemaValidation = getEnvBool("GOPROXY_NO_SCHEMA_VALIDATION")
 	IsDebug            = getEnvBool("GOPROXY_DEBUG")
-	ProxyHTTPPort      = ":" + getEnv("GOPROXY_HTTP_PORT", "80")
-	ProxyHTTPSPort     = ":" + getEnv("GOPROXY_HTTPS_PORT", "443")
-	APIHTTPPort        = ":" + getEnv("GOPROXY_API_PORT", "8888")
+	ProxyHTTPAddr      = getEnv("GOPROXY_HTTP_ADDR", ":80")
+	ProxyHTTPSAddr     = getEnv("GOPROXY_HTTPS_ADDR", ":443")
+	APIHTTPAddr        = getEnv("GOPROXY_API_ADDR", "127.0.0.1:8888")
 )
 
 func getEnvBool(key string) bool {

src/config/config.go

@@ -164,8 +164,8 @@ func (cfg *Config) Statistics() map[string]any {
 	}
 }
 
-func (cfg *Config) DumpEntries() map[string]*M.ProxyEntry {
-	entries := make(map[string]*M.ProxyEntry)
+func (cfg *Config) DumpEntries() map[string]*M.RawEntry {
+	entries := make(map[string]*M.RawEntry)
 	cfg.forEachRoute(func(alias string, r R.Route, p *PR.Provider) {
 		entries[alias] = r.Entry()
 	})
@@ -249,7 +249,7 @@ func (cfg *Config) loadProviders(providers *M.ProxyProviders) (res E.NestedError
 	for name, dockerHost := range providers.Docker {
 		p, err := PR.NewDockerProvider(name, dockerHost)
 		if err != nil {
-			b.Add(err.Subject(dockerHost))
+			b.Add(err.Subjectf("%s (%s)", name, dockerHost))
 			continue
 		}
 		cfg.proxyProviders.Store(p.GetName(), p)

src/docker/client.go

@@ -21,9 +21,8 @@ type Client struct {
 }
 
 func ParseDockerHostname(host string) (string, E.NestedError) {
-	if host == common.DockerHostFromEnv {
-		return host, nil
-	} else if host == "" {
+	switch host {
+	case common.DockerHostFromEnv, "":
 		return "localhost", nil
 	}
 	url, err := E.Check(client.ParseHostURL(host))

src/main.go

@@ -3,11 +3,11 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"io"
 	"log"
 	"net/http"
 	"os"
 	"os/signal"
-	"runtime"
 	"sync"
 	"syscall"
 	"time"
@@ -26,8 +26,6 @@ import (
 )
 
 func main() {
-	runtime.GOMAXPROCS(runtime.NumCPU())
-
 	args := common.GetArgs()
 	l := logrus.WithField("module", "main")
 
@@ -35,13 +33,17 @@ func main() {
 		logrus.SetLevel(logrus.DebugLevel)
 	}
 
-	logrus.SetFormatter(&logrus.TextFormatter{
-		DisableSorting:         true,
-		DisableLevelTruncation: true,
-		FullTimestamp:          true,
-		ForceColors:            true,
-		TimestampFormat:        "01-02 15:04:05",
-	})
+	if args.Command != common.CommandStart {
+		logrus.SetOutput(io.Discard)
+	} else {
+		logrus.SetFormatter(&logrus.TextFormatter{
+			DisableSorting:         true,
+			DisableLevelTruncation: true,
+			FullTimestamp:          true,
+			ForceColors:            true,
+			TimestampFormat:        "01-02 15:04:05",
+		})
+	}
 
 	if args.Command == common.CommandReload {
 		if err := apiUtils.ReloadServer(); err.HasError() {
@@ -59,15 +61,15 @@ func main() {
 			err = config.Validate(data).Error()
 		}
 		if err != nil {
-			l.Fatal("config error: ", err)
+			log.Fatal("config error: ", err)
 		}
-		l.Printf("config OK")
+		log.Print("config OK")
 		return
 	}
 
 	cfg, err := config.Load()
 	if err.IsFatal() {
-		l.Fatal(err)
+		log.Fatal(err)
 	}
 
 	if args.Command == common.CommandListConfigs {
@@ -75,8 +77,6 @@ func main() {
 		return
 	}
 
-	cfg.StartProxyProviders()
-
 	if args.Command == common.CommandListRoutes {
 		printJSON(cfg.RoutesByAlias())
 		return
@@ -87,6 +87,8 @@ func main() {
 		return
 	}
 
+	cfg.StartProxyProviders()
+
 	if err.HasError() {
 		l.Warn(err)
 	}
@@ -131,15 +133,15 @@ func main() {
 	proxyServer := server.InitProxyServer(server.Options{
 		Name:            "proxy",
 		CertProvider:    autocert,
-		HTTPPort:        common.ProxyHTTPPort,
-		HTTPSPort:       common.ProxyHTTPSPort,
+		HTTPAddr:        common.ProxyHTTPAddr,
+		HTTPSAddr:       common.ProxyHTTPSAddr,
 		Handler:         http.HandlerFunc(R.ProxyHandler),
 		RedirectToHTTPS: cfg.Value().RedirectToHTTPS,
 	})
 	apiServer := server.InitAPIServer(server.Options{
 		Name:            "api",
 		CertProvider:    autocert,
-		HTTPPort:        common.APIHTTPPort,
+		HTTPAddr:        common.APIHTTPAddr,
 		Handler:         api.NewHandler(cfg),
 		RedirectToHTTPS: cfg.Value().RedirectToHTTPS,
 	})
@@ -172,10 +174,11 @@ func main() {
 		close(done)
 	}()
 
+	timeout := time.After(time.Duration(cfg.Value().TimeoutShutdown) * time.Second)
 	select {
 	case <-done:
 		logrus.Info("shutdown complete")
-	case <-time.After(time.Duration(cfg.Value().TimeoutShutdown) * time.Second):
+	case <-timeout:
 		logrus.Info("timeout waiting for shutdown")
 	}
 }

src/models/raw_entry.go

@@ -10,7 +10,9 @@ import (
 )
 
 type (
-	ProxyEntry struct { // raw entry object before validation
+	RawEntry struct {
+		// raw entry object before validation
+		// loaded from docker labels or yaml file
 		Alias        string            `yaml:"-" json:"-"`
 		Scheme       string            `yaml:"scheme" json:"scheme"`
 		Host         string            `yaml:"host" json:"host"`
@@ -24,12 +26,12 @@ type (
 		*D.ProxyProperties `yaml:"-" json:"proxy_properties"`
 	}
 
-	ProxyEntries = F.Map[string, *ProxyEntry]
+	RawEntries = F.Map[string, *RawEntry]
 )
 
-var NewProxyEntries = F.NewMapOf[string, *ProxyEntry]
+var NewProxyEntries = F.NewMapOf[string, *RawEntry]
 
-func (e *ProxyEntry) SetDefaults() {
+func (e *RawEntry) SetDefaults() {
 	if e.ProxyProperties == nil {
 		e.ProxyProperties = &D.ProxyProperties{}
 	}

src/proxy/entry.go

@@ -43,7 +43,7 @@ func (rp *ReverseProxyEntry) UseIdleWatcher() bool {
 	return rp.IdleTimeout > 0 && rp.DockerHost != ""
 }
 
-func ValidateEntry(m *M.ProxyEntry) (any, E.NestedError) {
+func ValidateEntry(m *M.RawEntry) (any, E.NestedError) {
 	m.SetDefaults()
 	scheme, err := T.NewScheme(m.Scheme)
 	if err.HasError() {
@@ -63,7 +63,7 @@ func ValidateEntry(m *M.ProxyEntry) (any, E.NestedError) {
 	return entry, nil
 }
 
-func validateRPEntry(m *M.ProxyEntry, s T.Scheme, b E.Builder) *ReverseProxyEntry {
+func validateRPEntry(m *M.RawEntry, s T.Scheme, b E.Builder) *ReverseProxyEntry {
 	var stopTimeOut time.Duration
 
 	host, err := T.ValidateHost(m.Host)
@@ -121,7 +121,7 @@ func validateRPEntry(m *M.ProxyEntry, s T.Scheme, b E.Builder) *ReverseProxyEntr
 	}
 }
 
-func validateStreamEntry(m *M.ProxyEntry, b E.Builder) *StreamEntry {
+func validateStreamEntry(m *M.RawEntry, b E.Builder) *StreamEntry {
 	host, err := T.ValidateHost(m.Host)
 	b.Add(err)
 

src/proxy/provider/docker_provider.go

@@ -55,12 +55,12 @@ func (p *DockerProvider) LoadRoutesImpl() (routes R.Routes, err E.NestedError) {
 		// there may be some valid entries in `en`
 		dups := entries.MergeFrom(newEntries)
 		// add the duplicate proxy entries to the error
-		dups.RangeAll(func(k string, v *M.ProxyEntry) {
+		dups.RangeAll(func(k string, v *M.RawEntry) {
 			errors.Addf("duplicate alias %s", k)
 		})
 	}
 
-	entries.RangeAll(func(_ string, e *M.ProxyEntry) {
+	entries.RangeAll(func(_ string, e *M.RawEntry) {
 		e.DockerHost = p.dockerHost
 	})
 
@@ -96,7 +96,7 @@ func (p *DockerProvider) OnEvent(event W.Event, routes R.Routes) (res EventResul
 	entries, err := p.entriesFromContainerLabels(cont)
 	b.Add(err)
 
-	entries.RangeAll(func(alias string, entry *M.ProxyEntry) {
+	entries.RangeAll(func(alias string, entry *M.RawEntry) {
 		if routes.Has(alias) {
 			b.Add(E.AlreadyExist("alias", alias))
 		} else {
@@ -115,12 +115,12 @@ func (p *DockerProvider) OnEvent(event W.Event, routes R.Routes) (res EventResul
 
 // Returns a list of proxy entries for a container.
 // Always non-nil
-func (p *DockerProvider) entriesFromContainerLabels(container D.Container) (M.ProxyEntries, E.NestedError) {
+func (p *DockerProvider) entriesFromContainerLabels(container D.Container) (M.RawEntries, E.NestedError) {
 	entries := M.NewProxyEntries()
 
 	// init entries map for all aliases
 	for _, a := range container.Aliases {
-		entries.Store(a, &M.ProxyEntry{
+		entries.Store(a, &M.RawEntry{
 			Alias:           a,
 			Host:            p.hostname,
 			ProxyProperties: container.ProxyProperties,
@@ -156,7 +156,7 @@ func (p *DockerProvider) entriesFromContainerLabels(container D.Container) (M.Pr
 	return entries, errors.Build().Subject(container.ContainerName)
 }
 
-func (p *DockerProvider) applyLabel(container D.Container, entries M.ProxyEntries, key, val string) (res E.NestedError) {
+func (p *DockerProvider) applyLabel(container D.Container, entries M.RawEntries, key, val string) (res E.NestedError) {
 	b := E.NewBuilder("errors in label %s", key)
 	defer b.To(&res)
 
@@ -169,7 +169,7 @@ func (p *DockerProvider) applyLabel(container D.Container, entries M.ProxyEntrie
 	}
 	if lbl.Target == D.WildcardAlias {
 		// apply label for all aliases
-		entries.RangeAll(func(a string, e *M.ProxyEntry) {
+		entries.RangeAll(func(a string, e *M.RawEntry) {
 			if err = D.ApplyLabel(e, lbl); err.HasError() {
 				b.Add(err.Subject(lbl.Target))
 			}

src/proxy/provider/provider.go

@@ -143,10 +143,13 @@ func (p *Provider) GetRoute(alias string) (R.Route, bool) {
 }
 
 func (p *Provider) LoadRoutes() E.NestedError {
-	routes, err := p.LoadRoutesImpl()
-	p.routes = routes
-	p.l.Infof("loaded %d routes", routes.Size())
-	return err
+	var err E.NestedError
+	p.routes, err = p.LoadRoutesImpl()
+	if p.routes.Size() > 0 {
+		p.l.Infof("loaded %d routes", p.routes.Size())
+		return err
+	}
+	return E.FailWith("loading routes", err)
 }
 
 func (p *Provider) watchEvents() {

src/route/route.go

@@ -13,7 +13,7 @@ import (
 type (
 	Route interface {
 		RouteImpl
-		Entry() *M.ProxyEntry
+		Entry() *M.RawEntry
 		Type() RouteType
 		URL() *url.URL
 	}
@@ -28,7 +28,7 @@ type (
 	route struct {
 		RouteImpl
 		type_ RouteType
-		entry *M.ProxyEntry
+		entry *M.RawEntry
 	}
 )
 
@@ -40,7 +40,7 @@ const (
 // function alias
 var NewRoutes = F.NewMapOf[string, Route]
 
-func NewRoute(en *M.ProxyEntry) (Route, E.NestedError) {
+func NewRoute(en *M.RawEntry) (Route, E.NestedError) {
 	rt, err := P.ValidateEntry(en)
 	if err.HasError() {
 		return nil, err
@@ -61,7 +61,7 @@ func NewRoute(en *M.ProxyEntry) (Route, E.NestedError) {
 	return &route{RouteImpl: rt.(RouteImpl), entry: en, type_: t}, err
 }
 
-func (rt *route) Entry() *M.ProxyEntry {
+func (rt *route) Entry() *M.RawEntry {
 	return rt.entry
 }
 
@@ -74,11 +74,11 @@ func (rt *route) URL() *url.URL {
 	return url
 }
 
-func FromEntries(entries M.ProxyEntries) (Routes, E.NestedError) {
+func FromEntries(entries M.RawEntries) (Routes, E.NestedError) {
 	b := E.NewBuilder("errors in routes")
 
 	routes := NewRoutes()
-	entries.RangeAll(func(alias string, entry *M.ProxyEntry) {
+	entries.RangeAll(func(alias string, entry *M.RawEntry) {
 		entry.Alias = alias
 		r, err := NewRoute(entry)
 		if err.HasError() {

src/server/server.go

@@ -22,11 +22,9 @@ type server struct {
 }
 
 type Options struct {
-	Name string
-	// port (with leading colon)
-	HTTPPort string
-	// port (with leading colon)
-	HTTPSPort       string
+	Name            string
+	HTTPAddr        string
+	HTTPSAddr       string
 	CertProvider    *autocert.Provider
 	RedirectToHTTPS bool
 	Handler         http.Handler
@@ -55,22 +53,22 @@ func NewServer(opt Options) (s *server) {
 		certAvailable = err == nil
 	}
 
-	if certAvailable && opt.RedirectToHTTPS && opt.HTTPSPort != "" {
-		httpHandler = redirectToTLSHandler(opt.HTTPSPort)
+	if certAvailable && opt.RedirectToHTTPS && opt.HTTPSAddr != "" {
+		httpHandler = redirectToTLSHandler(opt.HTTPSAddr)
 	} else {
 		httpHandler = opt.Handler
 	}
 
-	if opt.HTTPPort != "" {
+	if opt.HTTPAddr != "" {
 		httpSer = &http.Server{
-			Addr:     opt.HTTPPort,
+			Addr:     opt.HTTPAddr,
 			Handler:  httpHandler,
 			ErrorLog: logger,
 		}
 	}
-	if certAvailable && opt.HTTPSPort != "" {
+	if certAvailable && opt.HTTPSAddr != "" {
 		httpsSer = &http.Server{
-			Addr:     opt.HTTPSPort,
+			Addr:     opt.HTTPSAddr,
 			Handler:  opt.Handler,
 			ErrorLog: logger,
 			TLSConfig: &tls.Config{

src/watcher/docker_watcher.go

@@ -2,6 +2,7 @@ package watcher
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	docker_events "github.com/docker/docker/api/types/events"
@@ -32,6 +33,8 @@ var (
 	DockerFilterUnpause   = filters.Arg("event", string(docker_events.ActionUnPause))
 
 	NewDockerFilter = filters.NewArgs
+
+	dockerWatcherRetryInterval = 3 * time.Second
 )
 
 func DockerrFilterContainerName(name string) filters.KeyValuePair {
@@ -55,27 +58,38 @@ func (w DockerWatcher) EventsWithOptions(ctx context.Context, options DockerList
 	errCh := make(chan E.NestedError)
 	started := make(chan struct{})
 
+	eventsCtx, eventsCancel := context.WithCancel(ctx)
+
 	go func() {
 		defer close(eventCh)
 		defer close(errCh)
 
+		defer func() {
+			if w.client.Connected() {
+				w.client.Close()
+			}
+		}()
+
 		if !w.client.Connected() {
 			var err E.NestedError
-			for range 3 {
+			attempts := 0
+			for {
 				w.client, err = D.ConnectClient(w.host)
 				if err != nil {
-					defer w.client.Close()
 					break
 				}
-				time.Sleep(1 * time.Second)
-			}
-			if err.HasError() {
-				errCh <- E.FailWith("docker connection", err)
-				return
+				attempts++
+				errCh <- E.FailWith(fmt.Sprintf("docker connection attempt #%d", attempts), err)
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					time.Sleep(dockerWatcherRetryInterval)
+				}
 			}
 		}
 
-		cEventCh, cErrCh := w.client.Events(ctx, options)
+		cEventCh, cErrCh := w.client.Events(eventsCtx, options)
 		started <- struct{}{}
 
 		for {
@@ -108,10 +122,10 @@ func (w DockerWatcher) EventsWithOptions(ctx context.Context, options DockerList
 				case <-ctx.Done():
 					return
 				default:
-					if D.IsErrConnectionFailed(err) {
-						time.Sleep(100 * time.Millisecond)
-						cEventCh, cErrCh = w.client.Events(ctx, options)
-					}
+					eventsCancel()
+					time.Sleep(dockerWatcherRetryInterval)
+					eventsCtx, eventsCancel = context.WithCancel(ctx)
+					cEventCh, cErrCh = w.client.Events(ctx, options)
 				}
 			}
 		}