Compare commits

...

23 Commits
v0.2.0 ... main

Author SHA1 Message Date
97b73fb442 remove binary from git 2025-02-03 11:35:06 -05:00
42b8afc9da Remove log from test 2025-01-24 22:10:58 -05:00
74ade60009 Refactor: Add GetListener and TLS versions 2025-01-24 19:01:12 -05:00
27d7f247c3 return ctx wrapping all values 2025-01-24 17:01:39 -05:00
f5fdf698ef Fix bad example for boolean options 2024-12-21 16:34:12 -05:00
94c737a02b Merge pull request 'Change address syntax to be a URL' (#3) from addr-url into main
Reviewed-on: #3
2024-12-19 10:06:02 -05:00
74efe7c808 update README 2024-12-19 09:46:36 -05:00
06e51af628 do dupe keys check only once 2024-12-18 15:45:36 -05:00
b14072106a Fix golint issues and add test 2024-12-18 15:38:52 -05:00
7cab173c7d fix ParseAddress 2024-12-17 19:36:31 -05:00
47ee5fb4f4 remove channel close 2024-12-17 18:16:38 -05:00
79256b0440 update to new syntax 2024-12-17 13:21:37 -05:00
e789d40258 Add helper function to parse address 2024-12-17 12:07:42 -05:00
917f340870 remove trailing space 2024-05-09 12:10:33 -04:00
0693a56697 add doc for idle server auto shutdown 2023-09-11 21:01:20 -04:00
e95d2d0a96 Merge pull request 'Add global idler and other helpers' (#2) from globalIdler into main
Reviewed-on: #2

Fixes #1
2023-09-11 20:45:51 -04:00
d4304636a2 add basic test 2023-09-11 20:44:52 -04:00
5c43a1a4fc un http Server async and return error via channel 2023-09-11 20:28:04 -04:00
ee28a78eda Add WrapHandler to automatically call Tick on every request 2023-09-11 20:24:10 -04:00
8ca82581fe WIP: Enter/Exit won't work :( 2023-09-08 19:35:58 -04:00
b5588989de Add function to return address type 2023-09-08 19:10:36 -04:00
ea04fe735b Add example for idle 2023-05-09 22:45:52 -04:00
710e8b66e6 Add idle package 2023-05-08 22:37:44 -04:00
7 changed files with 652 additions and 53 deletions

View File

@ -1,4 +1,4 @@
Create http server listening on unix sockets or systemd socket activated fds
Create http server listening on unix sockets and systemd socket activated fds
## Quick Usage
@ -17,36 +17,52 @@ Just replace `http.ListenAndServe` with `anyhttp.ListenAndServe`.
Syntax
unix/<path to socket>
unix?path=<socket_path>&mode=<socket file mode>&remove_existing=<true|false>
Examples
unix/relative/path.sock
unix//var/run/app/absolutepath.sock
unix?path=relative/path.sock
unix?path=/var/run/app/absolutepath.sock
unix?path=/run/app.sock&mode=600&remove_existing=false
| option | description | default |
|-----------------|------------------------------------------------|----------|
| path | path to unix socket | Required |
| mode | socket file mode | 666 |
| remove_existing | Whether to remove existing socket file or fail | true |
### Systemd Socket activated fd:
Syntax
sysd/fdidx/<fd index starting at 0>
sysd/fdname/<fd name set using FileDescriptorName socket setting >
sysd?idx=<fd index>&name=<fd name>&check_pid=<true|false>&unset_env=<true|false>&idle_timeout=<duration>
Only one of `idx` or `name` has to be set
Examples:
# First (or only) socket fd passed to app
sysd/fdidx/0
sysd?idx=0
# Socket with FileDescriptorName
sysd/fdname/myapp
sysd?name=myapp
# Using default name
sysd/fdname/myapp.socket
# Using default name and auto shutdown if no requests received in last 30 minutes
sysd?name=myapp.socket&idle_timeout=30m
### TCP port
| option | description | default |
|--------------|--------------------------------------------------------------------------------------------|------------------|
| name | Name configured via FileDescriptorName or socket file name | Required |
| idx | FD Index. Actual fd num will be 3 + idx | Required |
| idle_timeout | time to wait before shutdown. [syntax][0] | no auto shutdown |
| check_pid | Check process PID matches LISTEN_PID | true |
| unset_env | Unsets the LISTEN\* environment variables, so they don't get passed to any child processes | true |
If the address is a number less than 65536, it is assumed as a port and passed as `http.ListenAndServe(":<port>",...)`
### TCP
Anything else is directly passed to `http.ListenAndServe` as well. Below examples should work
If the address is not one of above, it is assumed to be tcp and passed to `http.ListenAndServe`.
Examples:
:http
:8888
@ -59,4 +75,6 @@ https://pkg.go.dev/go.balki.me/anyhttp
### Related links
* https://gist.github.com/teknoraver/5ffacb8757330715bcbcc90e6d46ac74#file-unixhttpd-go
* https://github.com/coreos/go-systemd/tree/main/activation
* https://github.com/coreos/go-systemd/tree/main/activation
[0]: https://pkg.go.dev/time#ParseDuration

View File

@ -2,16 +2,35 @@
package anyhttp
import (
"context"
"errors"
"fmt"
"io/fs"
"net"
"net/http"
"net/url"
"os"
"strconv"
"strings"
"sync"
"syscall"
"time"
"go.balki.me/anyhttp/idle"
)
// AddressType of the address passed
type AddressType string
var (
// UnixSocket - address is a unix socket, e.g. unix?path=/run/foo.sock
UnixSocket AddressType = "UnixSocket"
// SystemdFD - address is a systemd fd, e.g. sysd?name=myapp.socket
SystemdFD AddressType = "SystemdFD"
// TCP - address is a TCP address, e.g. :1234
TCP AddressType = "TCP"
// Unknown - address is not recognized
Unknown AddressType = "Unknown"
)
// UnixSocketConfig has the configuration for Unix socket
@ -83,6 +102,8 @@ type SysdConfig struct {
CheckPID bool
// Unsets the LISTEN* environment variables, so they don't get passed to any child processes
UnsetEnv bool
// Shutdown http server if no requests received for below timeout
IdleTimeout *time.Duration
}
// DefaultSysdConfig has the default values for SysdConfig
@ -182,59 +203,86 @@ func (s *SysdConfig) GetListener() (net.Listener, error) {
return nil, errors.New("neither FDIndex nor FDName set")
}
// UnknownAddress Error is returned when address does not match any known syntax
type UnknownAddress struct{}
// GetListener is low level function for use with non-http servers. e.g. tcp, smtp
// Caller should handle idle timeout if needed
func GetListener(addr string) (net.Listener, AddressType, any /* cfg */, error) {
func (u UnknownAddress) Error() string {
return "unknown address"
addrType, unixSocketConfig, sysdConfig, perr := parseAddress(addr)
if perr != nil {
return nil, Unknown, nil, perr
}
if unixSocketConfig != nil {
listener, err := unixSocketConfig.GetListener()
if err != nil {
return nil, Unknown, nil, err
}
return listener, addrType, unixSocketConfig, nil
} else if sysdConfig != nil {
listener, err := sysdConfig.GetListener()
if err != nil {
return nil, Unknown, nil, err
}
return listener, addrType, sysdConfig, nil
}
if addr == "" {
addr = ":http"
}
listener, err := net.Listen("tcp", addr)
return listener, TCP, nil, err
}
// GetListener gets a unix or systemd socket listener
func GetListener(addr string) (net.Listener, error) {
if strings.HasPrefix(addr, "unix/") {
usc := NewUnixSocketConfig(strings.TrimPrefix(addr, "unix/"))
return usc.GetListener()
}
type ServerCtx struct {
AddressType AddressType
Listener net.Listener
Server *http.Server
Idler idle.Idler
Done <-chan error
UnixSocketConfig *UnixSocketConfig
SysdConfig *SysdConfig
}
if strings.HasPrefix(addr, "sysd/fdidx/") {
idx, err := strconv.Atoi(strings.TrimPrefix(addr, "sysd/fdidx/"))
if err != nil {
return nil, fmt.Errorf("invalid fdidx, addr:%q err: %w", addr, err)
}
sysdc := NewSysDConfigWithFDIdx(idx)
return sysdc.GetListener()
}
func (s *ServerCtx) Wait() error {
return <-s.Done
}
if strings.HasPrefix(addr, "sysd/fdname/") {
sysdc := NewSysDConfigWithFDName(strings.TrimPrefix(addr, "sysd/fdname/"))
return sysdc.GetListener()
}
func (s *ServerCtx) Addr() net.Addr {
return s.Listener.Addr()
}
return nil, UnknownAddress{}
func (s *ServerCtx) Shutdown(ctx context.Context) error {
err := s.Server.Shutdown(ctx)
if err != nil {
return err
}
return <-s.Done
}
// ServeTLS creates and serves a HTTPS server.
func ServeTLS(addr string, h http.Handler, certFile string, keyFile string) (*ServerCtx, error) {
return serve(addr, h, certFile, keyFile)
}
// Serve creates and serves a HTTP server.
func Serve(addr string, h http.Handler) (*ServerCtx, error) {
return serve(addr, h, "", "")
}
// ListenAndServe is the drop-in replacement for `http.ListenAndServe`.
// Supports unix and systemd sockets in addition
func ListenAndServe(addr string, h http.Handler) error {
listener, err := GetListener(addr)
if _, isUnknown := err.(UnknownAddress); err != nil && !isUnknown {
ctx, err := Serve(addr, h)
if err != nil {
return err
}
return ctx.Wait()
}
if listener != nil {
return http.Serve(listener, h)
func ListenAndServeTLS(addr string, certFile string, keyFile string, h http.Handler) error {
ctx, err := ServeTLS(addr, h, certFile, keyFile)
if err != nil {
return err
}
if port, err := strconv.Atoi(addr); err == nil {
if port > 0 && port < 65536 {
return http.ListenAndServe(fmt.Sprintf(":%v", port), h)
}
return fmt.Errorf("invalid port: %v", port)
}
return http.ListenAndServe(addr, h)
return ctx.Wait()
}
// UnsetSystemdListenVars unsets the LISTEN* environment variables so they are not passed to any child processes
@ -243,3 +291,150 @@ func UnsetSystemdListenVars() {
_ = os.Unsetenv("LISTEN_FDS")
_ = os.Unsetenv("LISTEN_FDNAMES")
}
func parseAddress(addr string) (addrType AddressType, usc *UnixSocketConfig, sysc *SysdConfig, err error) {
usc = nil
sysc = nil
err = nil
u, err := url.Parse(addr)
if err != nil {
return TCP, nil, nil, nil
}
if u.Path == "unix" {
duc := DefaultUnixSocketConfig
usc = &duc
addrType = UnixSocket
for key, val := range u.Query() {
if len(val) != 1 {
err = fmt.Errorf("unix socket address error. Multiple %v found: %v", key, val)
return
}
if key == "path" {
usc.SocketPath = val[0]
} else if key == "mode" {
if _, serr := fmt.Sscanf(val[0], "%o", &usc.SocketMode); serr != nil {
err = fmt.Errorf("unix socket address error. Bad mode: %v, err: %w", val, serr)
return
}
} else if key == "remove_existing" {
if removeExisting, berr := strconv.ParseBool(val[0]); berr == nil {
usc.RemoveExisting = removeExisting
} else {
err = fmt.Errorf("unix socket address error. Bad remove_existing: %v, err: %w", val, berr)
return
}
} else {
err = fmt.Errorf("unix socket address error. Bad option; key: %v, val: %v", key, val)
return
}
}
if usc.SocketPath == "" {
err = fmt.Errorf("unix socket address error. Missing path; addr: %v", addr)
return
}
} else if u.Path == "sysd" {
dsc := DefaultSysdConfig
sysc = &dsc
addrType = SystemdFD
for key, val := range u.Query() {
if len(val) != 1 {
err = fmt.Errorf("systemd socket fd address error. Multiple %v found: %v", key, val)
return
}
if key == "name" {
sysc.FDName = &val[0]
} else if key == "idx" {
if idx, ierr := strconv.Atoi(val[0]); ierr == nil {
sysc.FDIndex = &idx
} else {
err = fmt.Errorf("systemd socket fd address error. Bad idx: %v, err: %w", val, ierr)
return
}
} else if key == "check_pid" {
if checkPID, berr := strconv.ParseBool(val[0]); berr == nil {
sysc.CheckPID = checkPID
} else {
err = fmt.Errorf("systemd socket fd address error. Bad check_pid: %v, err: %w", val, berr)
return
}
} else if key == "unset_env" {
if unsetEnv, berr := strconv.ParseBool(val[0]); berr == nil {
sysc.UnsetEnv = unsetEnv
} else {
err = fmt.Errorf("systemd socket fd address error. Bad unset_env: %v, err: %w", val, berr)
return
}
} else if key == "idle_timeout" {
if timeout, terr := time.ParseDuration(val[0]); terr == nil {
sysc.IdleTimeout = &timeout
} else {
err = fmt.Errorf("systemd socket fd address error. Bad idle_timeout: %v, err: %w", val, terr)
return
}
} else {
err = fmt.Errorf("systemd socket fd address error. Bad option; key: %v, val: %v", key, val)
return
}
}
if (sysc.FDIndex == nil) == (sysc.FDName == nil) {
err = fmt.Errorf("systemd socket fd address error. Exactly only one of name and idx has to be set. name: %v, idx: %v", sysc.FDName, sysc.FDIndex)
return
}
} else {
// Just assume as TCP address
return TCP, nil, nil, nil
}
return
}
func serve(addr string, h http.Handler, certFile string, keyFile string) (*ServerCtx, error) {
serveFn := func() func(ctx *ServerCtx) error {
if certFile != "" {
return func(ctx *ServerCtx) error {
return ctx.Server.ServeTLS(ctx.Listener, certFile, keyFile)
}
}
return func(ctx *ServerCtx) error {
return ctx.Server.Serve(ctx.Listener)
}
}()
var ctx ServerCtx
var err error
var cfg any
ctx.Listener, ctx.AddressType, cfg, err = GetListener(addr)
if err != nil {
return nil, err
}
switch ctx.AddressType {
case UnixSocket:
ctx.UnixSocketConfig = cfg.(*UnixSocketConfig)
case SystemdFD:
ctx.SysdConfig = cfg.(*SysdConfig)
}
errChan := make(chan error)
ctx.Done = errChan
if ctx.AddressType == SystemdFD && ctx.SysdConfig.IdleTimeout != nil {
ctx.Idler = idle.CreateIdler(*ctx.SysdConfig.IdleTimeout)
ctx.Server = &http.Server{Handler: idle.WrapIdlerHandler(ctx.Idler, h)}
waitErrChan := make(chan error)
go func() {
waitErrChan <- serveFn(&ctx)
}()
go func() {
select {
case err := <-waitErrChan:
errChan <- err
case <-ctx.Idler.Chan():
errChan <- ctx.Server.Shutdown(context.TODO())
}
}()
} else {
ctx.Server = &http.Server{Handler: h}
go func() {
errChan <- serveFn(&ctx)
}()
}
return &ctx, nil
}

158
anyhttp_test.go Normal file
View File

@ -0,0 +1,158 @@
package anyhttp
import (
"context"
"encoding/json"
"testing"
"time"
)
func Test_parseAddress(t *testing.T) {
tests := []struct {
name string // description of this test case
// Named input parameters for target function.
addr string
wantAddrType AddressType
wantUsc *UnixSocketConfig
wantSysc *SysdConfig
wantErr bool
}{
{
name: "tcp port",
addr: ":8080",
wantAddrType: TCP,
wantUsc: nil,
wantSysc: nil,
wantErr: false,
},
{
name: "unix address",
addr: "unix?path=/run/foo.sock&mode=660",
wantAddrType: UnixSocket,
wantUsc: &UnixSocketConfig{
SocketPath: "/run/foo.sock",
SocketMode: 0660,
RemoveExisting: true,
},
wantSysc: nil,
wantErr: false,
},
{
name: "systemd address",
addr: "sysd?name=foo.socket",
wantAddrType: SystemdFD,
wantUsc: nil,
wantSysc: &SysdConfig{
FDIndex: nil,
FDName: ptr("foo.socket"),
CheckPID: true,
UnsetEnv: true,
IdleTimeout: nil,
},
wantErr: false,
},
{
name: "systemd address with index",
addr: "sysd?idx=0&idle_timeout=30m",
wantAddrType: SystemdFD,
wantUsc: nil,
wantSysc: &SysdConfig{
FDIndex: ptr(0),
FDName: nil,
CheckPID: true,
UnsetEnv: true,
IdleTimeout: ptr(30 * time.Minute),
},
wantErr: false,
},
{
name: "systemd address. Bad example",
addr: "sysd?idx=0&idle_timeout=30m&name=foo",
wantAddrType: SystemdFD,
wantUsc: nil,
wantSysc: nil,
wantErr: true,
},
{
name: "systemd address with check_pid and unset_env",
addr: "sysd?idx=0&check_pid=false&unset_env=f",
wantAddrType: SystemdFD,
wantUsc: nil,
wantSysc: &SysdConfig{
FDIndex: ptr(0),
FDName: nil,
CheckPID: false,
UnsetEnv: false,
IdleTimeout: nil,
},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
gotAddrType, gotUsc, gotSysc, gotErr := parseAddress(tt.addr)
if gotErr != nil {
if !tt.wantErr {
t.Errorf("parseAddress() failed: %v", gotErr)
}
return
}
if tt.wantErr {
t.Fatal("parseAddress() succeeded unexpectedly")
}
if gotAddrType != tt.wantAddrType {
t.Errorf("parseAddress() addrType = %v, want %v", gotAddrType, tt.wantAddrType)
}
if !check(gotUsc, tt.wantUsc) {
t.Errorf("parseAddress() Usc = %v, want %v", gotUsc, tt.wantUsc)
}
if !check(gotSysc, tt.wantSysc) {
if (gotSysc == nil || tt.wantSysc == nil) ||
!(check(gotSysc.FDIndex, tt.wantSysc.FDIndex) &&
check(gotSysc.FDName, tt.wantSysc.FDName) &&
check(gotSysc.IdleTimeout, tt.wantSysc.IdleTimeout)) {
t.Errorf("parseAddress() Sysc = %v, want %v", asJSON(gotSysc), asJSON(tt.wantSysc))
}
}
})
}
}
func TestServe(t *testing.T) {
ctx, err := Serve("unix?path=/tmp/foo.sock", nil)
if err != nil {
t.Fatal()
}
if ctx.AddressType != UnixSocket {
t.Errorf("Serve() ServerCtx = %v, want %v", ctx.AddressType, UnixSocket)
}
ctx.Shutdown(context.TODO())
}
// Helpers
// print value instead of pointer
func asJSON[T any](val T) string {
op, err := json.Marshal(val)
if err != nil {
return err.Error()
}
return string(op)
}
func ptr[T any](val T) *T {
return &val
}
// nil safe equal check
func check[T comparable](got, want *T) bool {
if (got == nil) != (want == nil) {
return false
}
if got == nil {
return true
}
return *got == *want
}

View File

@ -0,0 +1,35 @@
package main
import (
"context"
"net/http"
"time"
"go.balki.me/anyhttp/idle"
)
func main() {
idler := idle.CreateIdler(10 * time.Second)
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
idler.Tick()
w.Write([]byte("Hey there!\n"))
})
http.HandleFunc("/job", func(w http.ResponseWriter, r *http.Request) {
go func() {
idler.Enter()
defer idler.Exit()
time.Sleep(15 * time.Second)
}()
w.Write([]byte("Job scheduled\n"))
})
server := http.Server{
Addr: ":8888",
}
go server.ListenAndServe()
idler.Wait()
server.Shutdown(context.TODO())
}

32
examples/simple/main.go Normal file
View File

@ -0,0 +1,32 @@
package main
import (
"context"
"log"
"net/http"
"os"
"time"
"go.balki.me/anyhttp"
)
func main() {
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte("hello\n"))
})
//log.Println("Got error: ", anyhttp.ListenAndServe(os.Args[1], nil))
ctx, err := anyhttp.Serve(os.Args[1], nil)
log.Printf("Got ctx: %v\n, err: %v", ctx, err)
log.Println(ctx.Addr())
if err != nil {
return
}
select {
case doneErr := <-ctx.Done:
log.Println(doneErr)
case <-time.After(1 * time.Minute):
log.Println("Awake")
ctx.Shutdown(context.TODO())
}
}

127
idle/idle.go Normal file
View File

@ -0,0 +1,127 @@
// Package idle helps to gracefully shutdown idle (typically http) servers
package idle
import (
"fmt"
"net/http"
"sync/atomic"
"time"
)
var (
// For simple servers without backgroud jobs, global singleton for simpler API
// Enter/Exit worn't work for global idler as Enter may be called before Wait, use CreateIdler in those cases
gIdler atomic.Pointer[idler]
)
// Wait waits till the server is idle and returns. i.e. no Ticks in last <timeout> duration
func Wait(timeout time.Duration) error {
i := CreateIdler(timeout).(*idler)
ok := gIdler.CompareAndSwap(nil, i)
if !ok {
return fmt.Errorf("idler already waiting")
}
i.Wait()
return nil
}
// Tick records the current time. This will make the server not idle until next Tick or timeout
func Tick() {
i := gIdler.Load()
if i != nil {
i.Tick()
}
}
// WrapHandler calls Tick() before processing passing request to http.Handler
func WrapHandler(h http.Handler) http.Handler {
if h == nil {
h = http.DefaultServeMux
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
Tick()
h.ServeHTTP(w, r)
})
}
// WrapIdlerHandler calls idler.Tick() before processing passing request to http.Handler
func WrapIdlerHandler(i Idler, h http.Handler) http.Handler {
if h == nil {
h = http.DefaultServeMux
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
i.Tick()
h.ServeHTTP(w, r)
})
}
// Idler helps manage idle servers
type Idler interface {
// Tick records the current time. This will make the server not idle until next Tick or timeout
Tick()
// Wait waits till the server is idle and returns. i.e. no Ticks in last <timeout> duration
Wait()
// For long running background jobs, use Enter to record start time. Wait will not return while there are active jobs running
Enter()
// Exit records end of a background job
Exit()
// Get the channel to wait yourself
Chan() <-chan struct{}
}
type idler struct {
lastTick atomic.Pointer[time.Time]
c chan struct{}
active atomic.Int64
}
func (i *idler) Enter() {
i.active.Add(1)
}
func (i *idler) Exit() {
i.Tick()
i.active.Add(-1)
}
// CreateIdler creates an Idler with given timeout
func CreateIdler(timeout time.Duration) Idler {
i := &idler{}
i.c = make(chan struct{})
i.Tick()
go func() {
for {
if i.active.Load() != 0 {
time.Sleep(timeout)
continue
}
t := *i.lastTick.Load()
now := time.Now()
dur := t.Add(timeout).Sub(now)
if dur == dur.Abs() {
time.Sleep(dur)
continue
}
break
}
close(i.c)
}()
return i
}
func (i *idler) Tick() {
now := time.Now()
i.lastTick.Store(&now)
}
func (i *idler) Wait() {
<-i.c
}
func (i *idler) Chan() <-chan struct{} {
return i.c
}

34
idle/idle_test.go Normal file
View File

@ -0,0 +1,34 @@
package idle
import (
"testing"
"time"
)
func TestIdlerChan(_ *testing.T) {
i := CreateIdler(10 * time.Millisecond)
<-i.Chan()
}
func TestGlobalIdler(t *testing.T) {
err := Wait(10 * time.Millisecond)
if err != nil {
t.Fatalf("idle.Wait failed, %v", err)
}
err = Wait(10 * time.Millisecond)
if err == nil {
t.Fatal("idle.Wait should fail when called second time")
}
}
func TestIdlerEnterExit(t *testing.T) {
i := CreateIdler(10 * time.Millisecond).(*idler)
i.Enter()
if i.active.Load() != 1 {
t.FailNow()
}
i.Exit()
if i.active.Load() != 0 {
t.FailNow()
}
}