Merge pull request #164 from Notifiarr/dn2_api

Split http into its own package

Author: davidnewhall

fog.conf

@@ -4,7 +4,7 @@
 
 # FW_PASSWORD: If a password is set then clients must send it as a setting.
 password = ""
-# FW_LISTEN_ADDR: IP and port to listen on.
+# FW_LISTEN_ADDR: IP and UDP port to listen on for UDP packets to write to disk.
 listen_addr = ":9000"
 # FW_FLUSH_INTERVAL: How old an item must be to flush it to disk.
 flush_interval = "16s"
@@ -19,18 +19,37 @@ log_file_mb = 0
 # FW_LOG_FILES: How many logs files to keep around.
 log_files = 0
 # FW_BUFFER_UDP: This is the UDP socket buffer in bytes.
-buffer_udp = 1048576
+buffer_udp = 4194304 # 4MB
 # FW_BUFFER_PACKET: This is the size of the read buffer per packet.
-buffer_packet = 8192
+buffer_packet = 262144 # 256KB
 # FW_BUFFER_CHAN: This is the channel buffer between the listeners and processors.
-buffer_chan = 10240
+buffer_chan = 32768 # 32 thousand
 # FW_LISTENERS: How many UDP socket listener threads to start.
 listeners = 1
 # FW_PROCESSORS: How many packet processor threads to start.
 processors = 1
 # FW_WRITERS: How many file system operations may happen in parallel. 2-5 is fine with fast disks.
 writers = 1
 # FW_BUFFER_FILE_SYS: How many file pointers can be buffered into the file system writer.
-buffer_file_sys = 10240
+buffer_file_sys = 32768 # 32 thousand
 # FW_DEBUG: Prints 1 line per packet when enabled.
 debug = false
+
+# FW_HTTP_SERVER: Configuration for the HTTP server.
+[http_server]
+# FW_HTTP_SERVER_LISTEN_ADDR: IP and TCP port to listen on. For API and metrics via HTTP.
+listen_addr = ":9000"
+# FW_HTTP_SERVER_READ_TIMEOUT: Maximum duration for reading the entire request, including the body.
+read_timeout = "10s"
+# FW_HTTP_SERVER_READ_HEADER_TIMEOUT: Maximum duration for reading the request header.
+read_header_timeout = "2s"
+# FW_HTTP_SERVER_WRITE_TIMEOUT: Maximum duration for writing the response to the client.
+write_timeout = "10s"
+# FW_HTTP_SERVER_IDLE_TIMEOUT: Maximum amount of time to wait for the next request when keep-alives are enabled.
+idle_timeout = "120s"
+# FW_HTTP_SERVER_MAX_HEADER_BYTES: Maximum number of bytes the server will read parsing the request header's keys and values, including the request line.
+max_header_bytes = 65536 # 64KB
+# FW_HTTP_SERVER_TLS_CERT_PATH: Path to the TLS certificate file. Provide this to enable HTTPS.
+tls_cert_path = ""
+# FW_HTTP_SERVER_TLS_KEY_PATH: Path to the TLS private key file for the certificate. Provide this to enable HTTPS.
+tls_key_path = ""

pkg/fog/logger.go

@@ -64,10 +64,22 @@ func (c *Config) Errorf(msg string, v ...any) {
 // printConfig logs the current configuration information.
 func (c *Config) printConfig() {
 	c.Printf("=> Fog Willow Starting, pid: %d", os.Getpid())
-	c.Printf("=> Listen Address / Password: %s / %v", c.ListenAddr, c.Password != "")
+	c.Printf("=> UDP Listen Address / Password: %s / %v", c.ListenAddr, c.Password != "")
+
+	if c.HTTPServer.TLSCertPath != "" && c.HTTPServer.TLSKeyPath != "" {
+		c.Printf("=> HTTPS Listen Address: %s (Cert=%s, Key=%s)", c.HTTPServer.ListenAddr, c.HTTPServer.TLSCertPath, c.HTTPServer.TLSKeyPath)
+	} else {
+		c.Printf("=> HTTP Listen Address: %s", c.HTTPServer.ListenAddr)
+	}
+
+	c.Printf("=> HTTP Server; Read/Header/Write/Idle: %s/%s/%s/%s, Max Header Bytes: %d",
+		c.HTTPServer.ReadTimeout, c.HTTPServer.ReadHeaderTimeout, c.HTTPServer.WriteTimeout,
+		c.HTTPServer.IdleTimeout, c.HTTPServer.MaxHeaderBytes)
+
 	c.Printf("=> Output Path: %s", c.OutputPath)
 	c.Printf("=> Intervals; Flush/Group: %s/%s", c.FlushInterval, c.GroupInterval)
 	c.Printf("=> Buffers; UDP/Packet/Chan/FS: %d/%d/%d/%d", c.BufferUDP, c.BufferPacket, c.BufferChan, c.BufferFileSys)
+	c.Printf("=> Buffer Pool (File Buffers): %v", c.BufferPool)
 	c.Printf("=> Threads; Listen/Process/Writer: %d/%d/%d", c.Listeners, c.Processors, c.Writers)
 
 	if c.LogFile != "" {

pkg/fog/start.go

@@ -3,52 +3,50 @@ package fog
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log"
 	"net"
-	"net/http"
 	"sync"
 	"time"
 
 	"github.com/Notifiarr/fogwillow/pkg/buf"
+	"github.com/Notifiarr/fogwillow/pkg/httpserver"
 	"github.com/Notifiarr/fogwillow/pkg/metrics"
 	"github.com/Notifiarr/fogwillow/pkg/willow"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"golift.io/cnfg"
 	"golift.io/cnfgfile"
 )
 
 // Some application defaults.
 const (
 	DefaultOutputPath   = "/tmp"
-	DefaultListenAddr   = ":9000"
-	DefaultUDPBuffer    = 1024 * 1024
-	DefaultPacketBuffer = 1024 * 8
-	DefaultChanBuffer   = 1024
+	DefaultUDPBuffer    = 1 << 22 // 4MB
+	DefaultPacketBuffer = 1 << 18 // 256KB
+	DefaultChanBuffer   = 1 << 15 // 32 thousand
 )
 
 // Config is the input _and_ running data.
 type Config struct {
 	*willow.Config
-	Password     string `toml:"password"      xml:"password"`
-	OutputPath   string `toml:"output_path"   xml:"output_path"`
-	ListenAddr   string `toml:"listen_addr"   xml:"listen_addr"`
-	LogFile      string `toml:"log_file"      xml:"log_file"`
-	LogFileMB    uint   `toml:"log_file_mb"   xml:"log_file_mb"`
-	LogFiles     uint   `toml:"log_files"     xml:"log_files"`
-	BufferUDP    uint   `toml:"buffer_udp"    xml:"buffer_udp"`
-	BufferPacket uint   `toml:"buffer_packet" xml:"buffer_packet"`
-	BufferChan   uint   `toml:"buffer_chan"   xml:"buffer_chan"`
-	Listeners    uint   `toml:"listeners"     xml:"listeners"`
-	Processors   uint   `toml:"processors"    xml:"processors"`
-	Debug        bool   `toml:"debug"         xml:"debug"`
+	HTTPServer   *httpserver.Config `toml:"http_server"   xml:"http_server"`
+	Password     string             `toml:"password"      xml:"password"`
+	OutputPath   string             `toml:"output_path"   xml:"output_path"`
+	ListenAddr   string             `toml:"listen_addr"   xml:"listen_addr"`
+	LogFile      string             `toml:"log_file"      xml:"log_file"`
+	LogFileMB    uint               `toml:"log_file_mb"   xml:"log_file_mb"`
+	LogFiles     uint               `toml:"log_files"     xml:"log_files"`
+	BufferUDP    uint               `toml:"buffer_udp"    xml:"buffer_udp"`
+	BufferPacket uint               `toml:"buffer_packet" xml:"buffer_packet"`
+	BufferChan   uint               `toml:"buffer_chan"   xml:"buffer_chan"`
+	Listeners    uint               `toml:"listeners"     xml:"listeners"`
+	Processors   uint               `toml:"processors"    xml:"processors"`
+	Debug        bool               `toml:"debug"         xml:"debug"`
 	log          *log.Logger
 	packets      chan *packet
 	sock         *net.UDPConn
 	willow       *willow.Willow
 	metrics      *metrics.Metrics
-	httpSrv      *http.Server
+	httpSrv      *httpserver.Server
 	newBuf       func(path string, data []byte) *buf.FileBuffer
 	bytesPool    sync.Pool
 	listenerWg   sync.WaitGroup
@@ -59,7 +57,7 @@ type Config struct {
 func LoadConfigFile(path string) (*Config, error) {
 	config := &Config{
 		OutputPath:   DefaultOutputPath,
-		ListenAddr:   DefaultListenAddr,
+		HTTPServer:   httpserver.DefaultConfig(),
 		BufferUDP:    DefaultUDPBuffer,
 		BufferPacket: DefaultPacketBuffer,
 		BufferChan:   DefaultChanBuffer,
@@ -104,20 +102,10 @@ func (c *Config) Start() error {
 		go c.packetListener(idx)
 	}
 
-	smx := http.NewServeMux()
-	smx.Handle("/metrics", promhttp.Handler())
-
-	c.httpSrv = &http.Server{
-		Handler:           smx,
-		Addr:              c.ListenAddr,
-		ReadTimeout:       time.Second,
-		ReadHeaderTimeout: time.Second,
-		WriteTimeout:      time.Second,
-		IdleTimeout:       20 * time.Second, //nolint:mnd
-	}
+	c.httpSrv = httpserver.New(c.HTTPServer, nil)
 
 	err = c.httpSrv.ListenAndServe()
-	if err != nil && !errors.Is(err, http.ErrServerClosed) {
+	if err != nil {
 		return fmt.Errorf("web server failed: %w", err)
 	}
 
@@ -148,17 +136,23 @@ func (c *Config) setup() {
 	c.Logger = c
 	c.Metrics = c.metrics
 	c.willow = willow.NeWillow(c.Config)
+
+	if c.HTTPServer == nil {
+		c.HTTPServer = httpserver.DefaultConfig()
+	}
+
+	c.HTTPServer.Setup()
 }
 
 func (c *Config) setupSocket() error {
 	addr, err := net.ResolveUDPAddr("udp", c.ListenAddr)
 	if err != nil {
-		return fmt.Errorf("invalid listen_addr provided: %w", err)
+		return fmt.Errorf("invalid udp listen_addr: %w", err)
 	}
 
 	c.sock, err = net.ListenUDP("udp", addr)
 	if err != nil {
-		return fmt.Errorf("unable to use provided listen_addr: %w", err)
+		return fmt.Errorf("unable to use udp listen_addr: %w", err)
 	}
 
 	err = c.sock.SetReadBuffer(int(c.BufferUDP)) //nolint:gosec

pkg/httpserver/config.go

@@ -0,0 +1,64 @@
+package httpserver
+
+import "time"
+
+// HTTP server defaults.
+const (
+	DefaultListenAddr        = ":9000"
+	DefaultReadTimeout       = 10 * time.Second
+	DefaultReadHeaderTimeout = 2 * time.Second
+	DefaultWriteTimeout      = 10 * time.Second
+	DefaultIdleTimeout       = 120 * time.Second
+	DefaultMaxHeaderBytes    = 1 << 16 // 64KB
+)
+
+// Config is the configuration for the HTTP server.
+type Config struct {
+	ListenAddr        string        `toml:"listen_addr"         xml:"listen_addr"`
+	ReadTimeout       time.Duration `toml:"read_timeout"        xml:"read_timeout"`
+	ReadHeaderTimeout time.Duration `toml:"read_header_timeout" xml:"read_header_timeout"`
+	WriteTimeout      time.Duration `toml:"write_timeout"       xml:"write_timeout"`
+	IdleTimeout       time.Duration `toml:"idle_timeout"        xml:"idle_timeout"`
+	MaxHeaderBytes    int           `toml:"max_header_bytes"    xml:"max_header_bytes"`
+	TLSCertPath       string        `toml:"tls_cert_path"       xml:"tls_cert_path"`
+	TLSKeyPath        string        `toml:"tls_key_path"        xml:"tls_key_path"`
+}
+
+// DefaultConfig returns the default configuration for the HTTP server.
+func DefaultConfig() *Config {
+	return &Config{
+		ListenAddr:        DefaultListenAddr,
+		ReadTimeout:       DefaultReadTimeout,
+		ReadHeaderTimeout: DefaultReadHeaderTimeout,
+		WriteTimeout:      DefaultWriteTimeout,
+		IdleTimeout:       DefaultIdleTimeout,
+		MaxHeaderBytes:    DefaultMaxHeaderBytes,
+	}
+}
+
+// Setup fills empty listen address and any zero timeouts or limits so logs and runtime match.
+func (c *Config) Setup() {
+	if c.ListenAddr == "" {
+		c.ListenAddr = DefaultListenAddr
+	}
+
+	if c.ReadTimeout <= 0 {
+		c.ReadTimeout = DefaultReadTimeout
+	}
+
+	if c.ReadHeaderTimeout <= 0 {
+		c.ReadHeaderTimeout = DefaultReadHeaderTimeout
+	}
+
+	if c.WriteTimeout <= 0 {
+		c.WriteTimeout = DefaultWriteTimeout
+	}
+
+	if c.IdleTimeout <= 0 {
+		c.IdleTimeout = DefaultIdleTimeout
+	}
+
+	if c.MaxHeaderBytes <= 0 {
+		c.MaxHeaderBytes = DefaultMaxHeaderBytes
+	}
+}

pkg/httpserver/server.go

@@ -0,0 +1,74 @@
+// Package httpserver owns the HTTP listener used for Prometheus metrics and future API routes.
+// It is separate from the UDP ingest path and from pkg/willow.
+package httpserver
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// Server wraps http.Server with a shared ServeMux for /metrics and optional handlers.
+type Server struct {
+	Mux         *http.ServeMux
+	srv         *http.Server
+	TLSCertPath string
+	TLSKeyPath  string
+}
+
+// New builds an HTTP server on addr. register is optional; use it to attach API or other routes
+// to the same mux (e.g. mux.Handle("/api/", apiHandler)). /metrics is always registered.
+func New(config *Config, register func(mux *http.ServeMux)) *Server {
+	if config == nil {
+		config = DefaultConfig()
+	}
+
+	config.Setup()
+
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+
+	if register != nil {
+		register(mux)
+	}
+
+	return &Server{
+		Mux:         mux,
+		TLSCertPath: config.TLSCertPath,
+		TLSKeyPath:  config.TLSKeyPath,
+		srv: &http.Server{
+			Handler:           mux,
+			Addr:              config.ListenAddr,
+			ReadTimeout:       config.ReadTimeout,
+			ReadHeaderTimeout: config.ReadHeaderTimeout,
+			WriteTimeout:      config.WriteTimeout,
+			IdleTimeout:       config.IdleTimeout,
+			MaxHeaderBytes:    config.MaxHeaderBytes,
+		},
+	}
+}
+
+// ListenAndServe binds and serves until Shutdown is called or the listener errors.
+func (s *Server) ListenAndServe() error {
+	var err error
+
+	if s.TLSCertPath != "" || s.TLSKeyPath != "" {
+		err = s.srv.ListenAndServeTLS(s.TLSCertPath, s.TLSKeyPath)
+	} else {
+		err = s.srv.ListenAndServe()
+	}
+
+	if err != nil && !errors.Is(err, http.ErrServerClosed) {
+		return fmt.Errorf("http listen: %w", err)
+	}
+
+	return nil
+}
+
+// Shutdown stops the server gracefully.
+func (s *Server) Shutdown(ctx context.Context) error {
+	return s.srv.Shutdown(ctx) //nolint:wrapcheck
+}

pkg/willow/memory.go

@@ -7,7 +7,6 @@ import (
 )
 
 // Len returns the number of file buffers in the map.
-// Safe to call from any goroutine.
 func (w *Willow) Len() int {
 	return int(w.memLen.Load())
 }
@@ -85,7 +84,7 @@ func (w *Willow) washer(now time.Time, force bool) {
 	}
 }
 
-// trySet is the memoryHole-internal implementation of TrySet.
+// trySet is the internal implementation of TrySet.
 // Returns nil when candidate was stored, or the existing buffer when the path was already taken.
 func (w *Willow) trySet(candidate *buf.FileBuffer) *buf.FileBuffer {
 	if existing, ok := w.memory[candidate.Path]; ok {

pkg/willow/willow.go

@@ -12,7 +12,7 @@ import (
 
 // Defaults for the module.
 const (
-	DefaultFileSysBuffer = 1024 * 10
+	DefaultFileSysBuffer = 1 << 15 // 32 thousand
 	DefaultFlushInterval = 16 * time.Second
 )