feat: traefik-selectel-dns — A-записи Selectel по роутам Traefik + CI сборка образа

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

internal/dockerevents/watch.go

@@ -0,0 +1,134 @@
+// Package dockerevents подписывается на события контейнеров Docker (raw HTTP,
+// без зависимости от docker SDK) и вызывает trigger как сигнал к немедленному reconcile.
+// События — только триггер; источник истины — Traefik API.
+package dockerevents
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+const eventsFilters = `{"type":["container"],"event":["start","die","destroy"]}`
+
+// Watcher читает поток /events.
+type Watcher struct {
+	host    string
+	log     *slog.Logger
+	trigger func()
+
+	// для тестов
+	minBackoff, maxBackoff time.Duration
+}
+
+// New создаёт Watcher. host: unix:///path/to.sock или tcp://host:port.
+func New(host string, log *slog.Logger, trigger func()) *Watcher {
+	return &Watcher{host: host, log: log, trigger: trigger, minBackoff: time.Second, maxBackoff: time.Minute}
+}
+
+func (w *Watcher) client() (*http.Client, string, error) {
+	u, err := url.Parse(w.host)
+	if err != nil {
+		return nil, "", fmt.Errorf("docker.host: %w", err)
+	}
+	dialer := &net.Dialer{Timeout: 5 * time.Second}
+	tr := &http.Transport{ResponseHeaderTimeout: 10 * time.Second}
+	switch u.Scheme {
+	case "unix":
+		path := u.Path
+		if path == "" {
+			return nil, "", fmt.Errorf("docker.host: пустой путь сокета")
+		}
+		tr.DialContext = func(ctx context.Context, _, _ string) (net.Conn, error) {
+			return dialer.DialContext(ctx, "unix", path)
+		}
+		return &http.Client{Transport: tr}, "http://docker", nil
+	case "tcp", "http":
+		tr.DialContext = dialer.DialContext
+		return &http.Client{Transport: tr}, "http://" + u.Host, nil
+	default:
+		return nil, "", fmt.Errorf("docker.host: неподдерживаемая схема %q (unix|tcp)", u.Scheme)
+	}
+}
+
+// Run блокируется до отмены ctx, переподключаясь при обрывах. Ошибки не фатальны.
+func (w *Watcher) Run(ctx context.Context) {
+	hc, base, err := w.client()
+	if err != nil {
+		w.log.Error("Docker events отключены", "error", err)
+		return
+	}
+	backoff := w.minBackoff
+	failing := false
+	for ctx.Err() == nil {
+		connected, err := w.stream(ctx, hc, base)
+		if ctx.Err() != nil {
+			return
+		}
+		if connected {
+			backoff = w.minBackoff
+			failing = false
+		}
+		if !failing {
+			w.log.Warn("Docker events недоступны, работаем по опросу; повторные попытки в фоне", "error", err)
+			failing = true
+		} else {
+			w.log.Debug("Docker events: повторная попытка не удалась", "error", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(backoff):
+		}
+		if backoff *= 2; backoff > w.maxBackoff {
+			backoff = w.maxBackoff
+		}
+	}
+}
+
+// stream возвращает connected=true, если соединение было установлено (ответ 200).
+func (w *Watcher) stream(ctx context.Context, hc *http.Client, base string) (bool, error) {
+	q := url.Values{"filters": {eventsFilters}}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/events?"+q.Encode(), nil)
+	if err != nil {
+		return false, err
+	}
+	resp, err := hc.Do(req)
+	if err != nil {
+		return false, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return false, fmt.Errorf("docker events: статус %d", resp.StatusCode)
+	}
+	w.log.Info("подписка на Docker events установлена")
+	sc := bufio.NewScanner(resp.Body)
+	sc.Buffer(make([]byte, 64*1024), 1<<20)
+	for sc.Scan() {
+		line := strings.TrimSpace(sc.Text())
+		if line == "" {
+			continue
+		}
+		var ev struct {
+			Type   string `json:"Type"`
+			Action string `json:"Action"`
+		}
+		if err := json.Unmarshal([]byte(line), &ev); err != nil {
+			w.log.Debug("Docker events: не удалось разобрать событие", "error", err)
+			continue
+		}
+		w.log.Debug("событие Docker", "type", ev.Type, "action", ev.Action)
+		w.trigger()
+	}
+	if err := sc.Err(); err != nil {
+		return true, err
+	}
+	return true, fmt.Errorf("docker events: поток закрыт")
+}

internal/dockerevents/watch_test.go

@@ -0,0 +1,105 @@
+package dockerevents
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func quiet() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }
+
+func waitFor(t *testing.T, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("условие не выполнено за отведённое время")
+}
+
+func TestTriggerOnEventsTCP(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/events" || !strings.Contains(r.URL.Query().Get("filters"), "container") {
+			t.Errorf("url = %s", r.URL)
+		}
+		_, _ = w.Write([]byte(`{"Type":"container","Action":"start"}` + "\n" + `{"Type":"container","Action":"die"}` + "\n"))
+		w.(http.Flusher).Flush()
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	var n int32
+	w := New("tcp://"+strings.TrimPrefix(srv.URL, "http://"), quiet(), func() { atomic.AddInt32(&n, 1) })
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { w.Run(ctx); close(done) }()
+	waitFor(t, func() bool { return atomic.LoadInt32(&n) >= 2 })
+	cancel()
+	<-done
+}
+
+func TestUnixSocketAndReconnect(t *testing.T) {
+	sock := filepath.Join(t.TempDir(), "d.sock")
+	l, err := net.Listen("unix", sock)
+	if err != nil {
+		t.Skipf("unix сокеты недоступны: %v", err)
+	}
+	var conns int32
+	srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt32(&conns, 1)
+		_, _ = w.Write([]byte(`{"Type":"container","Action":"start"}` + "\n"))
+		// поток закрывается сразу — ждём переподключения
+	})}
+	go srv.Serve(l)
+	defer srv.Close()
+
+	var n int32
+	w := New("unix://"+sock, quiet(), func() { atomic.AddInt32(&n, 1) })
+	w.minBackoff, w.maxBackoff = 10*time.Millisecond, 20*time.Millisecond
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { w.Run(ctx); close(done) }()
+	waitFor(t, func() bool { return atomic.LoadInt32(&conns) >= 2 && atomic.LoadInt32(&n) >= 2 })
+	cancel()
+	<-done
+}
+
+func TestUnavailableSocketDoesNotPanicOrExit(t *testing.T) {
+	w := New("unix:///nonexistent/docker.sock", quiet(), func() { t.Error("trigger не ожидался") })
+	w.minBackoff, w.maxBackoff = 5*time.Millisecond, 10*time.Millisecond
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	done := make(chan struct{})
+	go func() { w.Run(ctx); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run не завершился после отмены контекста")
+	}
+}
+
+func TestInvalidHostScheme(t *testing.T) {
+	w := New("ftp://x", quiet(), func() {})
+	if _, _, err := w.client(); err == nil {
+		t.Fatal("ожидалась ошибка")
+	}
+	// Run на невалидном host возвращается сразу
+	done := make(chan struct{})
+	go func() { w.Run(context.Background()); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Run должен вернуться при невалидном host")
+	}
+}