Merge branch 'next'

Author: iBug

.github/workflows/go.yml

@@ -4,7 +4,7 @@ on:
   pull_request:
     types: [opened, synchronize, reopened]
   push:
-    branches: [master]
+    branches: ['*']
     tags: ['*']
 
 jobs:

cmd/cmd.go

@@ -73,19 +73,19 @@ func New() *cobra.Command {
 	c := &cobra.Command{
 		Use: "rsync-proxy",
 		RunE: func(cmd *cobra.Command, args []string) error {
-			switch {
-			case reload:
-				return SendReloadRequest(s.WebListenAddr, cmd.OutOrStdout(), cmd.ErrOrStderr())
-			case version:
+			if version {
 				return printVersion(cmd.OutOrStdout())
 			}
 
 			log.SetOutput(cmd.OutOrStdout(), cmd.ErrOrStderr())
 
-			err := s.LoadConfigFromFile()
+			err := s.ReadConfigFromFile()
 			if err != nil {
 				return fmt.Errorf("load config: %w", err)
 			}
+			if reload {
+				return SendReloadRequest(s.HTTPListenAddr, cmd.OutOrStdout(), cmd.ErrOrStderr())
+			}
 
 			s.WriteTimeout = time.Minute
 			s.ReadTimeout = time.Minute
@@ -98,9 +98,7 @@ func New() *cobra.Command {
 		},
 	}
 	flags := c.Flags()
-	flags.StringVar(&s.ListenAddr, "listen-addr", "0.0.0.0:9527", "Address to listen on for reverse proxy")
-	flags.StringVar(&s.WebListenAddr, "web.listen-addr", "127.0.0.1:9528", "Address to listen on for API")
-	flags.StringVar(&s.ConfigPath, "config", "/etc/rsync-proxy/config.toml", "Path to config file")
+	flags.StringVarP(&s.ConfigPath, "config", "c", "/etc/rsync-proxy/config.toml", "Path to config file")
 	flags.BoolVar(&reload, "reload", false, "Inform server to reload config")
 	flags.BoolVarP(&version, "version", "V", false, "Print version and exit")
 

dist/config.example.toml

@@ -1,19 +1,18 @@
 # Example config
 
 [proxy]
+listen = "0.0.0.0:873"
+listen_http = "127.0.0.1:9528"
 motd = "Served by rsync-proxy (https://github.com/ustclug/rsync-proxy)"
 
 [upstreams.u1]
-host = "127.0.0.1"
-port = 1234
+address = "127.0.0.1:1234"
 modules = ["foo"]
 
 [upstreams.u2]
-host = "192.168.0.10"
-port = 1235
+address = "192.168.0.10:1235"
 modules = ["bar"]
 
 [upstreams.u3]
-host = "rsync.example.internal"
-port = 1235
+address = "rsync.example.internal:1235"
 modules = ["baz"]

dist/rsync-proxy.service

@@ -6,8 +6,8 @@ Restart=on-failure
 StartLimitInterval=10s
 
 ExecStartPre=mkdir -p /var/log/rsync-proxy/
-ExecStart=/usr/local/bin/rsync-proxy -v=3 --config=/etc/rsync-proxy/config.toml --listen-addr=0.0.0.0:873 --web.listen-addr=127.0.0.1:9528
-ExecReload=/usr/local/bin/rsync-proxy --web.listen-addr=127.0.0.1:9528 --reload
+ExecStart=/usr/local/bin/rsync-proxy -v=3 --config=/etc/rsync-proxy/config.toml
+ExecReload=/usr/local/bin/rsync-proxy --config=/etc/rsync-proxy/config.toml --reload
 StandardOutput=append:/var/log/rsync-proxy/stdout.log
 StandardError=append:/var/log/rsync-proxy/stderr.log
 

pkg/server/config.go

@@ -10,21 +10,22 @@ import (
 )
 
 type Upstream struct {
-	Host    string   `toml:"host"`
-	Port    int      `toml:"port"`
+	Address string   `toml:"address"`
 	Modules []string `toml:"modules"`
 }
 
 type ProxySettings struct {
-	Motd string `toml:"motd"`
+	Listen     string `toml:"listen"`
+	ListenHTTP string `toml:"listen_http"`
+	Motd       string `toml:"motd"`
 }
 
 type Config struct {
 	Proxy     ProxySettings        `toml:"proxy"`
 	Upstreams map[string]*Upstream `toml:"upstreams"`
 }
 
-func (s *Server) LoadConfig(r io.Reader) error {
+func (s *Server) ReadConfig(r io.Reader) error {
 	log.V(3).Infof("[INFO] loading config")
 
 	dec := toml.NewDecoder(r)
@@ -33,17 +34,14 @@ func (s *Server) LoadConfig(r io.Reader) error {
 	if err != nil {
 		return err
 	}
-
-	s.Upstreams = c.Upstreams
-	s.Motd = c.Proxy.Motd
-	return s.complete()
+	return s.loadConfig(&c)
 }
 
-func (s *Server) LoadConfigFromFile() error {
+func (s *Server) ReadConfigFromFile() error {
 	f, err := os.Open(s.ConfigPath)
 	if err != nil {
 		return err
 	}
 	defer f.Close()
-	return s.LoadConfig(f)
+	return s.ReadConfig(f)
 }

pkg/server/config_test.go

@@ -6,25 +6,22 @@ import (
 	"testing"
 )
 
-func TestLoadConfig(t *testing.T) {
+func TestReadConfig(t *testing.T) {
 	s := New()
 	configContent := `
 [upstreams.u1]
-host = "127.0.0.1"
-port = 1234
+address = "127.0.0.1:1234"
 modules = ["foo1", "foo2"]
 
 [upstreams.u2]
-host = "127.0.0.1"
-port = 1235
+address = "127.0.0.1:1235"
 modules = ["bar1"]
 
 [upstreams.u3]
-host = "example.com"
-port = 1235
+address = "example.com:1235"
 modules = ["bar2"]
 `
-	err := s.LoadConfig(strings.NewReader(configContent))
+	err := s.ReadConfig(strings.NewReader(configContent))
 	if err != nil {
 		t.Fatalf("Load config: %s", err)
 	}
@@ -43,20 +40,18 @@ func TestDuplicatedModulesInConfig(t *testing.T) {
 	s := New()
 	configContent := `
 [upstreams.u1]
-host = "127.0.0.1"
-port = 1234
+address = "127.0.0.1:1234"
 modules = ["foo1", "foo2"]
 
 [upstreams.u2]
-host = "127.0.0.1"
-port = 1235
+address = "127.0.0.1:1235"
 modules = ["foo1"]
 `
-	err := s.LoadConfig(strings.NewReader(configContent))
+	err := s.ReadConfig(strings.NewReader(configContent))
 	if err == nil {
 		t.Fatalf("Unexpected success")
 	}
-	if !strings.Contains(err.Error(), "duplicated module name") {
+	if !strings.Contains(err.Error(), "duplicate module name") {
 		t.Errorf("Unexpected error. Got: %s", err)
 	}
 }
@@ -68,11 +63,10 @@ func TestLoadMotdInConfig(t *testing.T) {
 motd = "Proudly served by rsync-proxy\ntest newline"
 
 [upstreams.u1]
-host = "127.0.0.1"
-port = 1234
+address = "127.0.0.1:1234"
 modules = ["foo1", "foo2"]
 `
-	err := s.LoadConfig(strings.NewReader(configContent))
+	err := s.ReadConfig(strings.NewReader(configContent))
 	if err != nil {
 		t.Fatalf("Load config: %s", err)
 	}

pkg/server/server.go

@@ -9,8 +9,8 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"os"
 	"sort"
-	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -33,11 +33,10 @@ const lineFeed = '\n'
 type Server struct {
 	// --- Options section
 	// Listen Address
-	ListenAddr    string
-	WebListenAddr string
-	ConfigPath    string
-	// name -> upstream
-	Upstreams    map[string]*Upstream
+	ListenAddr     string
+	HTTPListenAddr string
+	ConfigPath     string
+
 	ReadTimeout  time.Duration
 	WriteTimeout time.Duration
 	// motd
@@ -53,14 +52,14 @@ type Server struct {
 	activeConnCount atomic.Int64
 	connIndex       atomic.Uint64
 
-	TCPListener  net.Listener
-	HTTPListener net.Listener
+	TCPListener  *net.TCPListener
+	HTTPListener *net.TCPListener
 }
 
 func New() *Server {
 	return &Server{
 		bufPool: sync.Pool{
-			New: func() interface{} {
+			New: func() any {
 				buf := make([]byte, TCPBufferSize)
 				return &buf
 			},
@@ -69,33 +68,33 @@ func New() *Server {
 	}
 }
 
-func (s *Server) complete() error {
-	if len(s.Upstreams) == 0 {
+func (s *Server) loadConfig(c *Config) error {
+	if len(c.Upstreams) == 0 {
 		return fmt.Errorf("no upstream found")
 	}
 
 	modules := map[string]string{}
-	for upstreamName, v := range s.Upstreams {
-		addr := net.JoinHostPort(v.Host, strconv.Itoa(v.Port))
+	for upstreamName, v := range c.Upstreams {
+		addr := v.Address
 		_, err := net.ResolveTCPAddr("tcp", addr)
 		if err != nil {
 			return fmt.Errorf("resolve address: %w, upstream=%s, address=%s", err, upstreamName, addr)
 		}
 		for _, moduleName := range v.Modules {
 			if _, ok := modules[moduleName]; ok {
-				return fmt.Errorf("duplicated module name: %s, upstream=%s", moduleName, upstreamName)
+				return fmt.Errorf("duplicate module name: %s, upstream=%s", moduleName, upstreamName)
 			}
 			modules[moduleName] = addr
 		}
 	}
 
 	s.reloadLock.Lock()
+	// s.ListenAddr = c.Proxy.Listen
+	// s.HTTPListenAddr = c.Proxy.ListenHTTP
+	s.Motd = c.Proxy.Motd
 	s.modules = modules
 	s.reloadLock.Unlock()
 
-	// .Upstreams is no longer used, reclaims the memory
-	s.Upstreams = nil
-
 	return nil
 }
 
@@ -255,12 +254,12 @@ func (s *Server) relay(ctx context.Context, downConn *net.TCPConn) error {
 	return nil
 }
 
-func (s *Server) handleConn(ctx context.Context, conn net.Conn) {
+func (s *Server) handleConn(ctx context.Context, conn *net.TCPConn) {
 	s.activeConnCount.Add(1)
 	defer s.activeConnCount.Add(-1)
 	_ = s.connIndex.Add(1)
-	downConn := conn.(*net.TCPConn)
-	err := s.relay(ctx, downConn)
+
+	err := s.relay(ctx, conn)
 	if err != nil {
 		log.V(2).Errorf("[WARN] handleConn: %s", err)
 	}
@@ -278,7 +277,7 @@ func (s *Server) runHTTPServer() error {
 			Message string `json:"message"`
 		}
 
-		err := s.LoadConfigFromFile()
+		err := s.ReadConfigFromFile()
 		if err != nil {
 			log.Errorf("[ERROR] Load config: %s", err)
 			w.WriteHeader(http.StatusInternalServerError)
@@ -290,17 +289,20 @@ func (s *Server) runHTTPServer() error {
 		_ = json.NewEncoder(w).Encode(&resp)
 	})
 
-	mux.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc("/telegraf", func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodGet {
 			w.WriteHeader(http.StatusMethodNotAllowed)
 			return
 		}
 
-		var resp struct {
-			Count int64 `json:"count"`
+		timestamp := time.Now().Truncate(time.Second).UnixNano()
+		count := s.GetActiveConnectionCount()
+		hostname, err := os.Hostname()
+		if err != nil {
+			hostname = "(unknown)"
 		}
-		resp.Count = s.GetActiveConnectionCount()
-		_ = json.NewEncoder(w).Encode(&resp)
+		// https://docs.influxdata.com/influxdb/latest/reference/syntax/line-protocol/
+		_, _ = fmt.Fprintf(w, "rsync-proxy,host=%q count=%d %d\n", hostname, count, timestamp)
 	})
 
 	return http.Serve(s.HTTPListener, &mux)
@@ -314,15 +316,15 @@ func (s *Server) Listen() error {
 	s.ListenAddr = l1.Addr().String()
 	log.V(3).Infof("[INFO] Rsync proxy listening on %s", s.ListenAddr)
 
-	l2, err := net.Listen("tcp", s.WebListenAddr)
+	l2, err := net.Listen("tcp", s.HTTPListenAddr)
 	if err != nil {
 		return fmt.Errorf("create http listener: %w", err)
 	}
-	s.WebListenAddr = l2.Addr().String()
-	log.V(3).Infof("[INFO] HTTP server listening on %s", s.WebListenAddr)
+	s.HTTPListenAddr = l2.Addr().String()
+	log.V(3).Infof("[INFO] HTTP server listening on %s", s.HTTPListenAddr)
 
-	s.TCPListener = l1
-	s.HTTPListener = l2
+	s.TCPListener = l1.(*net.TCPListener)
+	s.HTTPListener = l2.(*net.TCPListener)
 	return nil
 }
 
@@ -336,14 +338,14 @@ func (s *Server) Close() {
 }
 
 func (s *Server) Run() error {
-	errCh := make(chan error)
+	errC := make(chan error)
 	go func() {
 		err := s.runHTTPServer()
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
 				return
 			}
-			errCh <- fmt.Errorf("run http server: %w", err)
+			errC <- fmt.Errorf("run http server: %w", err)
 		}
 	}()
 
@@ -352,12 +354,12 @@ func (s *Server) Run() error {
 
 	for {
 		select {
-		case err := <-errCh:
+		case err := <-errC:
 			return err
 		default:
 		}
 
-		conn, err := s.TCPListener.Accept()
+		conn, err := s.TCPListener.AcceptTCP()
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
 				return nil

pkg/server/server_test.go

@@ -20,7 +20,7 @@ func startServer(t *testing.T) *Server {
 		addr    = "127.0.0.1:0"
 		timeout = time.Second
 	)
-	srv.WebListenAddr = addr
+	srv.HTTPListenAddr = addr
 	srv.ListenAddr = addr
 	srv.ReadTimeout = timeout
 	srv.WriteTimeout = timeout