Fixed a few issues:

- Incorrect name being shown on dashboard "Proxies page"
- Apps being shown when homepage.show is false
- Load balanced routes are shown on homepage instead of the load balancer
- Route with idlewatcher will now be removed on container destroy
- Idlewatcher panic
- Performance improvement
- Idlewatcher infinitely loading
- Reload stucked / not working properly
- Streams stuck on shutdown / reload
- etc...
Added:
- support idlewatcher for loadbalanced routes
- partial implementation for stream type idlewatcher
Issues:
- graceful shutdown

internal/watcher/docker_watcher.go

@@ -15,8 +15,9 @@ import (
 
 type (
 	DockerWatcher struct {
-		host   string
-		client D.Client
+		host        string
+		client      D.Client
+		clientOwned bool
 		logrus.FieldLogger
 	}
 	DockerListOptions = docker_events.ListOptions
@@ -44,10 +45,11 @@ func DockerrFilterContainer(nameOrID string) filters.KeyValuePair {
 
 func NewDockerWatcher(host string) DockerWatcher {
 	return DockerWatcher{
+		host:        host,
+		clientOwned: true,
 		FieldLogger: (logrus.
 			WithField("module", "docker_watcher").
 			WithField("host", host)),
-		host: host,
 	}
 }
 
@@ -72,7 +74,7 @@ func (w DockerWatcher) EventsWithOptions(ctx context.Context, options DockerList
 		defer close(errCh)
 
 		defer func() {
-			if w.client.Connected() {
+			if w.clientOwned && w.client.Connected() {
 				w.client.Close()
 			}
 		}()

internal/watcher/events/event_queue.go

@@ -0,0 +1,91 @@
+package events
+
+import (
+	"time"
+
+	E "github.com/yusing/go-proxy/internal/error"
+	"github.com/yusing/go-proxy/internal/task"
+)
+
+type (
+	EventQueue struct {
+		task    task.Task
+		queue   []Event
+		ticker  *time.Ticker
+		onFlush OnFlushFunc
+		onError OnErrorFunc
+	}
+	OnFlushFunc = func(flushTask task.Task, events []Event)
+	OnErrorFunc = func(err E.NestedError)
+)
+
+const eventQueueCapacity = 10
+
+// NewEventQueue returns a new EventQueue with the given
+// queueTask, flushInterval, onFlush and onError.
+//
+// The returned EventQueue will start a goroutine to flush events in the queue
+// when the flushInterval is reached.
+//
+// The onFlush function is called when the flushInterval is reached and the queue is not empty,
+//
+// The onError function is called when an error received from the errCh,
+// or panic occurs in the onFlush function. Panic will cause a E.ErrPanicRecv error.
+//
+// flushTask.Finish must be called after the flush is done,
+// but the onFlush function can return earlier (e.g. run in another goroutine).
+//
+// If task is cancelled before the flushInterval is reached, the events in queue will be discarded.
+func NewEventQueue(parent task.Task, flushInterval time.Duration, onFlush OnFlushFunc, onError OnErrorFunc) *EventQueue {
+	return &EventQueue{
+		task:    parent.Subtask("event queue"),
+		queue:   make([]Event, 0, eventQueueCapacity),
+		ticker:  time.NewTicker(flushInterval),
+		onFlush: onFlush,
+		onError: onError,
+	}
+}
+
+func (e *EventQueue) Start(eventCh <-chan Event, errCh <-chan E.NestedError) {
+	go func() {
+		defer e.ticker.Stop()
+		for {
+			select {
+			case <-e.task.Context().Done():
+				e.task.Finish(e.task.FinishCause().Error())
+				return
+			case <-e.ticker.C:
+				if len(e.queue) > 0 {
+					flushTask := e.task.Subtask("flush events")
+					queue := e.queue
+					e.queue = make([]Event, 0, eventQueueCapacity)
+					go func() {
+						defer func() {
+							if err := recover(); err != nil {
+								e.onError(E.PanicRecv("panic in onFlush %s", err))
+							}
+						}()
+						e.onFlush(flushTask, queue)
+					}()
+					flushTask.Wait()
+				}
+			case event, ok := <-eventCh:
+				e.queue = append(e.queue, event)
+				if !ok {
+					return
+				}
+			case err := <-errCh:
+				if err != nil {
+					e.onError(err)
+				}
+			}
+		}
+	}()
+}
+
+// Wait waits for all events to be flushed and the task to finish.
+//
+// It is safe to call this method multiple times.
+func (e *EventQueue) Wait() {
+	e.task.Wait()
+}

internal/watcher/events/events.go

@@ -74,7 +74,7 @@ var actionNameMap = func() (m map[Action]string) {
 }()
 
 func (e Event) String() string {
-	return fmt.Sprintf("%s %s", e.ActorName, e.Action)
+	return fmt.Sprintf("%s %s", e.Action, e.ActorName)
 }
 
 func (a Action) String() string {

internal/watcher/health/http.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"net/http"
 
-	"github.com/yusing/go-proxy/internal/common"
 	"github.com/yusing/go-proxy/internal/net/types"
 )
 
@@ -15,10 +14,10 @@ type HTTPHealthMonitor struct {
 	pinger *http.Client
 }
 
-func NewHTTPHealthMonitor(task common.Task, url types.URL, config *HealthCheckConfig) HealthMonitor {
+func NewHTTPHealthMonitor(url types.URL, config *HealthCheckConfig, transport http.RoundTripper) *HTTPHealthMonitor {
 	mon := new(HTTPHealthMonitor)
-	mon.monitor = newMonitor(task, url, config, mon.checkHealth)
-	mon.pinger = &http.Client{Timeout: config.Timeout}
+	mon.monitor = newMonitor(url, config, mon.CheckHealth)
+	mon.pinger = &http.Client{Timeout: config.Timeout, Transport: transport}
 	if config.UseGet {
 		mon.method = http.MethodGet
 	} else {
@@ -27,19 +26,26 @@ func NewHTTPHealthMonitor(task common.Task, url types.URL, config *HealthCheckCo
 	return mon
 }
 
-func (mon *HTTPHealthMonitor) checkHealth() (healthy bool, detail string, err error) {
+func NewHTTPHealthChecker(url types.URL, config *HealthCheckConfig, transport http.RoundTripper) HealthChecker {
+	return NewHTTPHealthMonitor(url, config, transport)
+}
+
+func (mon *HTTPHealthMonitor) CheckHealth() (healthy bool, detail string, err error) {
+	ctx, cancel := mon.ContextWithTimeout("ping request timed out")
+	defer cancel()
+
 	req, reqErr := http.NewRequestWithContext(
-		mon.task.Context(),
+		ctx,
 		mon.method,
-		mon.url.JoinPath(mon.config.Path).String(),
+		mon.url.Load().JoinPath(mon.config.Path).String(),
 		nil,
 	)
 	if reqErr != nil {
 		err = reqErr
 		return
 	}
-	req.Header.Set("Connection", "close")
 
+	req.Header.Set("Connection", "close")
 	resp, respErr := mon.pinger.Do(req)
 	if respErr == nil {
 		resp.Body.Close()

internal/watcher/health/monitor.go

@@ -2,78 +2,93 @@ package health
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
-	"sync"
+	"fmt"
 	"time"
 
-	"github.com/yusing/go-proxy/internal/common"
+	E "github.com/yusing/go-proxy/internal/error"
 	"github.com/yusing/go-proxy/internal/net/types"
+	"github.com/yusing/go-proxy/internal/task"
 	U "github.com/yusing/go-proxy/internal/utils"
 	F "github.com/yusing/go-proxy/internal/utils/functional"
 )
 
 type (
 	HealthMonitor interface {
-		Start()
-		Stop()
+		task.TaskStarter
+		task.TaskFinisher
+		fmt.Stringer
+		json.Marshaler
 		Status() Status
 		Uptime() time.Duration
 		Name() string
-		String() string
-		MarshalJSON() ([]byte, error)
+	}
+	HealthChecker interface {
+		CheckHealth() (healthy bool, detail string, err error)
+		URL() types.URL
+		Config() *HealthCheckConfig
+		UpdateURL(url types.URL)
 	}
 	HealthCheckFunc func() (healthy bool, detail string, err error)
 	monitor         struct {
 		service string
 		config  *HealthCheckConfig
-		url     types.URL
+		url     U.AtomicValue[types.URL]
 
 		status      U.AtomicValue[Status]
 		checkHealth HealthCheckFunc
 		startTime   time.Time
 
-		task   common.Task
-		cancel context.CancelFunc
-		done   chan struct{}
-
-		mu sync.Mutex
+		task task.Task
 	}
 )
 
 var monMap = F.NewMapOf[string, HealthMonitor]()
 
-func newMonitor(task common.Task, url types.URL, config *HealthCheckConfig, healthCheckFunc HealthCheckFunc) *monitor {
-	service := task.Name()
-	task, cancel := task.SubtaskWithCancel("Health monitor for %s", service)
+func newMonitor(url types.URL, config *HealthCheckConfig, healthCheckFunc HealthCheckFunc) *monitor {
 	mon := &monitor{
-		service:     service,
 		config:      config,
-		url:         url,
 		checkHealth: healthCheckFunc,
 		startTime:   time.Now(),
-		task:        task,
-		cancel:      cancel,
-		done:        make(chan struct{}),
+		task:        task.DummyTask(),
 	}
+	mon.url.Store(url)
 	mon.status.Store(StatusHealthy)
 	return mon
 }
 
-func Inspect(name string) (HealthMonitor, bool) {
-	return monMap.Load(name)
+func Inspect(service string) (HealthMonitor, bool) {
+	return monMap.Load(service)
 }
 
-func (mon *monitor) Start() {
-	defer monMap.Store(mon.task.Name(), mon)
+func (mon *monitor) ContextWithTimeout(cause string) (ctx context.Context, cancel context.CancelFunc) {
+	if mon.task != nil {
+		return context.WithTimeoutCause(mon.task.Context(), mon.config.Timeout, errors.New(cause))
+	} else {
+		return context.WithTimeoutCause(context.Background(), mon.config.Timeout, errors.New(cause))
+	}
+}
 
+// Start implements task.TaskStarter.
+func (mon *monitor) Start(routeSubtask task.Task) E.NestedError {
+	mon.service = routeSubtask.Parent().Name()
+	mon.task = routeSubtask
+
+	if err := mon.checkUpdateHealth(); err != nil {
+		mon.task.Finish(fmt.Sprintf("healthchecker %s failure: %s", mon.service, err))
+		return err
+	}
 	go func() {
-		defer close(mon.done)
-		defer mon.task.Finished()
+		defer func() {
+			monMap.Delete(mon.task.Name())
+			if mon.status.Load() != StatusError {
+				mon.status.Store(StatusUnknown)
+			}
+			mon.task.Finish(mon.task.FinishCause().Error())
+		}()
 
-		ok := mon.checkUpdateHealth()
-		if !ok {
-			return
-		}
+		monMap.Store(mon.service, mon)
 
 		ticker := time.NewTicker(mon.config.Interval)
 		defer ticker.Stop()
@@ -83,48 +98,61 @@ func (mon *monitor) Start() {
 			case <-mon.task.Context().Done():
 				return
 			case <-ticker.C:
-				ok = mon.checkUpdateHealth()
-				if !ok {
+				err := mon.checkUpdateHealth()
+				if err != nil {
+					logger.Errorf("healthchecker %s failure: %s", mon.service, err)
 					return
 				}
 			}
 		}
 	}()
+	return nil
 }
 
-func (mon *monitor) Stop() {
-	monMap.Delete(mon.task.Name())
-
-	mon.mu.Lock()
-	defer mon.mu.Unlock()
-
-	if mon.cancel == nil {
-		return
-	}
-
-	mon.cancel()
-	<-mon.done
-
-	mon.cancel = nil
-	mon.status.Store(StatusUnknown)
+// Finish implements task.TaskFinisher.
+func (mon *monitor) Finish(reason string) {
+	mon.task.Finish(reason)
 }
 
+// UpdateURL implements HealthChecker.
+func (mon *monitor) UpdateURL(url types.URL) {
+	mon.url.Store(url)
+}
+
+// URL implements HealthChecker.
+func (mon *monitor) URL() types.URL {
+	return mon.url.Load()
+}
+
+// Config implements HealthChecker.
+func (mon *monitor) Config() *HealthCheckConfig {
+	return mon.config
+}
+
+// Status implements HealthMonitor.
 func (mon *monitor) Status() Status {
 	return mon.status.Load()
 }
 
+// Uptime implements HealthMonitor.
 func (mon *monitor) Uptime() time.Duration {
 	return time.Since(mon.startTime)
 }
 
+// Name implements HealthMonitor.
 func (mon *monitor) Name() string {
+	if mon.task == nil {
+		return ""
+	}
 	return mon.task.Name()
 }
 
+// String implements fmt.Stringer of HealthMonitor.
 func (mon *monitor) String() string {
 	return mon.Name()
 }
 
+// MarshalJSON implements json.Marshaler of HealthMonitor.
 func (mon *monitor) MarshalJSON() ([]byte, error) {
 	return (&JSONRepresentation{
 		Name:    mon.service,
@@ -132,19 +160,19 @@ func (mon *monitor) MarshalJSON() ([]byte, error) {
 		Status:  mon.status.Load(),
 		Started: mon.startTime,
 		Uptime:  mon.Uptime(),
-		URL:     mon.url,
+		URL:     mon.url.Load(),
 	}).MarshalJSON()
 }
 
-func (mon *monitor) checkUpdateHealth() (hasError bool) {
+func (mon *monitor) checkUpdateHealth() E.NestedError {
 	healthy, detail, err := mon.checkHealth()
 	if err != nil {
+		defer mon.task.Finish(err.Error())
 		mon.status.Store(StatusError)
 		if !errors.Is(err, context.Canceled) {
-			logger.Errorf("%s failed to check health: %s", mon.service, err)
+			return E.Failure("check health").With(err)
 		}
-		mon.Stop()
-		return false
+		return nil
 	}
 	var status Status
 	if healthy {
@@ -160,5 +188,5 @@ func (mon *monitor) checkUpdateHealth() (hasError bool) {
 		}
 	}
 
-	return true
+	return nil
 }

internal/watcher/health/raw.go

@@ -3,7 +3,6 @@ package health
 import (
 	"net"
 
-	"github.com/yusing/go-proxy/internal/common"
 	"github.com/yusing/go-proxy/internal/net/types"
 )
 
@@ -14,9 +13,9 @@ type (
 	}
 )
 
-func NewRawHealthMonitor(task common.Task, url types.URL, config *HealthCheckConfig) HealthMonitor {
+func NewRawHealthMonitor(url types.URL, config *HealthCheckConfig) *RawHealthMonitor {
 	mon := new(RawHealthMonitor)
-	mon.monitor = newMonitor(task, url, config, mon.checkAvail)
+	mon.monitor = newMonitor(url, config, mon.CheckHealth)
 	mon.dialer = &net.Dialer{
 		Timeout:       config.Timeout,
 		FallbackDelay: -1,
@@ -24,14 +23,22 @@ func NewRawHealthMonitor(task common.Task, url types.URL, config *HealthCheckCon
 	return mon
 }
 
-func (mon *RawHealthMonitor) checkAvail() (avail bool, detail string, err error) {
-	conn, dialErr := mon.dialer.DialContext(mon.task.Context(), mon.url.Scheme, mon.url.Host)
+func NewRawHealthChecker(url types.URL, config *HealthCheckConfig) HealthChecker {
+	return NewRawHealthMonitor(url, config)
+}
+
+func (mon *RawHealthMonitor) CheckHealth() (healthy bool, detail string, err error) {
+	ctx, cancel := mon.ContextWithTimeout("ping request timed out")
+	defer cancel()
+
+	url := mon.url.Load()
+	conn, dialErr := mon.dialer.DialContext(ctx, url.Scheme, url.Host)
 	if dialErr != nil {
 		detail = dialErr.Error()
 		/* trunk-ignore(golangci-lint/nilerr) */
 		return
 	}
 	conn.Close()
-	avail = true
+	healthy = true
 	return
 }