405 lines
9.5 KiB
Go
405 lines
9.5 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
pp "net/http/pprof"
|
|
"os"
|
|
"os/signal"
|
|
"runtime"
|
|
"runtime/debug"
|
|
"sync"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/pkg/profile"
|
|
"lol.mleku.dev/chk"
|
|
"lol.mleku.dev/log"
|
|
"next.orly.dev/app"
|
|
"next.orly.dev/app/config"
|
|
"next.orly.dev/pkg/acl"
|
|
"next.orly.dev/pkg/crypto/keys"
|
|
"next.orly.dev/pkg/database"
|
|
_ "next.orly.dev/pkg/dgraph" // Import to register dgraph factory
|
|
_ "next.orly.dev/pkg/neo4j" // Import to register neo4j factory
|
|
"next.orly.dev/pkg/encoders/hex"
|
|
"next.orly.dev/pkg/utils/interrupt"
|
|
"next.orly.dev/pkg/version"
|
|
)
|
|
|
|
func main() {
|
|
runtime.GOMAXPROCS(128)
|
|
debug.SetGCPercent(10)
|
|
var err error
|
|
var cfg *config.C
|
|
if cfg, err = config.New(); chk.T(err) {
|
|
}
|
|
log.I.F("starting %s %s", cfg.AppName, version.V)
|
|
|
|
// Handle 'identity' subcommand: print relay identity secret and pubkey and exit
|
|
if config.IdentityRequested() {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
var db database.Database
|
|
if db, err = database.NewDatabase(
|
|
ctx, cancel, cfg.DBType, cfg.DataDir, cfg.DBLogLevel,
|
|
); chk.E(err) {
|
|
os.Exit(1)
|
|
}
|
|
defer db.Close()
|
|
skb, err := db.GetOrCreateRelayIdentitySecret()
|
|
if chk.E(err) {
|
|
os.Exit(1)
|
|
}
|
|
pk, err := keys.SecretBytesToPubKeyHex(skb)
|
|
if chk.E(err) {
|
|
os.Exit(1)
|
|
}
|
|
fmt.Printf(
|
|
"identity secret: %s\nidentity pubkey: %s\n", hex.Enc(skb), pk,
|
|
)
|
|
os.Exit(0)
|
|
}
|
|
|
|
// Ensure profiling is stopped on interrupts (SIGINT/SIGTERM) as well as on normal exit
|
|
var profileStopOnce sync.Once
|
|
profileStop := func() {}
|
|
switch cfg.Pprof {
|
|
case "cpu":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.CPUProfile, profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("cpu profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.CPUProfile)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("cpu profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
case "memory":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.MemProfile, profile.MemProfileRate(32),
|
|
profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("memory profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.MemProfile)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("memory profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
case "allocation":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.MemProfileAllocs, profile.MemProfileRate(32),
|
|
profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("allocation profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.MemProfileAllocs)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("allocation profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
case "heap":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.MemProfileHeap, profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("heap profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.MemProfileHeap)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("heap profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
case "mutex":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.MutexProfile, profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("mutex profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.MutexProfile)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("mutex profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
case "threadcreate":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.ThreadcreationProfile,
|
|
profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("threadcreate profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.ThreadcreationProfile)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("threadcreate profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
case "goroutine":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.GoroutineProfile, profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("goroutine profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.GoroutineProfile)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("goroutine profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
case "block":
|
|
if cfg.PprofPath != "" {
|
|
prof := profile.Start(
|
|
profile.BlockProfile, profile.ProfilePath(cfg.PprofPath),
|
|
)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("block profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
} else {
|
|
prof := profile.Start(profile.BlockProfile)
|
|
profileStop = func() {
|
|
profileStopOnce.Do(
|
|
func() {
|
|
prof.Stop()
|
|
log.I.F("block profiling stopped and flushed")
|
|
},
|
|
)
|
|
}
|
|
defer profileStop()
|
|
}
|
|
|
|
}
|
|
|
|
// Register a handler so profiling is stopped when an interrupt is received
|
|
interrupt.AddHandler(
|
|
func() {
|
|
log.I.F("interrupt received: stopping profiling")
|
|
profileStop()
|
|
},
|
|
)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
var db database.Database
|
|
log.I.F("initializing %s database at %s", cfg.DBType, cfg.DataDir)
|
|
if db, err = database.NewDatabase(
|
|
ctx, cancel, cfg.DBType, cfg.DataDir, cfg.DBLogLevel,
|
|
); chk.E(err) {
|
|
os.Exit(1)
|
|
}
|
|
log.I.F("%s database initialized successfully", cfg.DBType)
|
|
acl.Registry.Active.Store(cfg.ACLMode)
|
|
if err = acl.Registry.Configure(cfg, db, ctx); chk.E(err) {
|
|
os.Exit(1)
|
|
}
|
|
acl.Registry.Syncer()
|
|
|
|
// Start HTTP pprof server if enabled
|
|
if cfg.PprofHTTP {
|
|
pprofAddr := fmt.Sprintf("%s:%d", cfg.Listen, 6060)
|
|
pprofMux := http.NewServeMux()
|
|
pprofMux.HandleFunc("/debug/pprof/", pp.Index)
|
|
pprofMux.HandleFunc("/debug/pprof/cmdline", pp.Cmdline)
|
|
pprofMux.HandleFunc("/debug/pprof/profile", pp.Profile)
|
|
pprofMux.HandleFunc("/debug/pprof/symbol", pp.Symbol)
|
|
pprofMux.HandleFunc("/debug/pprof/trace", pp.Trace)
|
|
for _, p := range []string{
|
|
"allocs", "block", "goroutine", "heap", "mutex", "threadcreate",
|
|
} {
|
|
pprofMux.Handle("/debug/pprof/"+p, pp.Handler(p))
|
|
}
|
|
ppSrv := &http.Server{Addr: pprofAddr, Handler: pprofMux}
|
|
go func() {
|
|
log.I.F("pprof server listening on %s", pprofAddr)
|
|
if err := ppSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
|
log.E.F("pprof server error: %v", err)
|
|
}
|
|
}()
|
|
go func() {
|
|
<-ctx.Done()
|
|
shutdownCtx, cancelShutdown := context.WithTimeout(
|
|
context.Background(), 2*time.Second,
|
|
)
|
|
defer cancelShutdown()
|
|
_ = ppSrv.Shutdown(shutdownCtx)
|
|
}()
|
|
}
|
|
|
|
// Start health check HTTP server if configured
|
|
var healthSrv *http.Server
|
|
if cfg.HealthPort > 0 {
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc(
|
|
"/healthz", func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(http.StatusOK)
|
|
_, _ = w.Write([]byte("ok"))
|
|
log.I.F("health check ok")
|
|
},
|
|
)
|
|
// Optional shutdown endpoint to gracefully stop the process so profiling defers run
|
|
if cfg.EnableShutdown {
|
|
mux.HandleFunc(
|
|
"/shutdown", func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(http.StatusOK)
|
|
_, _ = w.Write([]byte("shutting down"))
|
|
log.I.F("shutdown requested via /shutdown; sending SIGINT to self")
|
|
go func() {
|
|
p, _ := os.FindProcess(os.Getpid())
|
|
_ = p.Signal(os.Interrupt)
|
|
}()
|
|
},
|
|
)
|
|
}
|
|
healthSrv = &http.Server{
|
|
Addr: fmt.Sprintf(
|
|
"%s:%d", cfg.Listen, cfg.HealthPort,
|
|
), Handler: mux,
|
|
}
|
|
go func() {
|
|
log.I.F("health check server listening on %s", healthSrv.Addr)
|
|
if err := healthSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
|
log.E.F("health server error: %v", err)
|
|
}
|
|
}()
|
|
go func() {
|
|
<-ctx.Done()
|
|
shutdownCtx, cancelShutdown := context.WithTimeout(
|
|
context.Background(), 2*time.Second,
|
|
)
|
|
defer cancelShutdown()
|
|
_ = healthSrv.Shutdown(shutdownCtx)
|
|
}()
|
|
}
|
|
|
|
quit := app.Run(ctx, cfg, db)
|
|
sigs := make(chan os.Signal, 1)
|
|
signal.Notify(sigs, os.Interrupt, syscall.SIGTERM)
|
|
for {
|
|
select {
|
|
case <-sigs:
|
|
fmt.Printf("\r")
|
|
log.I.F("received shutdown signal, starting graceful shutdown")
|
|
cancel() // This will trigger HTTP server shutdown
|
|
<-quit // Wait for HTTP server to shut down
|
|
chk.E(db.Close())
|
|
log.I.F("exiting")
|
|
return
|
|
case <-quit:
|
|
log.I.F("application quit signal received")
|
|
cancel()
|
|
chk.E(db.Close())
|
|
log.I.F("exiting")
|
|
return
|
|
}
|
|
}
|
|
// log.I.F("exiting")
|
|
}
|