From 6e71f82b4350c961754f5308a1a08281f32ce429 Mon Sep 17 00:00:00 2001 From: Anik Date: Wed, 16 Aug 2023 12:20:48 -0400 Subject: [PATCH] start server in goroutine and shutdown gracefully --- cmd/manager/main.go | 19 +++++++++++++------ pkg/catalogserver/server.go | 35 ++++++++++++++++++++++++----------- 2 files changed, 37 insertions(+), 17 deletions(-) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index f1bcc696..70bba894 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -19,6 +19,7 @@ package main import ( "flag" "fmt" + "net/http" "os" // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.) @@ -36,7 +37,7 @@ import ( "github.com/operator-framework/catalogd/internal/source" "github.com/operator-framework/catalogd/internal/version" - catalogserver "github.com/operator-framework/catalogd/pkg/catalogserver" + "github.com/operator-framework/catalogd/pkg/catalogserver" corecontrollers "github.com/operator-framework/catalogd/pkg/controllers/core" "github.com/operator-framework/catalogd/pkg/features" "github.com/operator-framework/catalogd/pkg/profile" @@ -68,7 +69,7 @@ func main() { catalogdVersion bool systemNamespace string storageDir string - serverPort string + catalogServerAddr string ) flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.") flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") @@ -78,8 +79,8 @@ func main() { // TODO: should we move the unpacker to some common place? Or... hear me out... should catalogd just be a rukpak provisioner? flag.StringVar(&unpackImage, "unpack-image", "quay.io/operator-framework/rukpak:v0.12.0", "The unpack image to use when unpacking catalog images") flag.StringVar(&systemNamespace, "system-namespace", "", "The namespace catalogd uses for internal state, configuration, and workloads") - flag.StringVar(&storageDir, "catalogs-storage-dir", "/var/cache", "The directory in the filesystem where unpacked catalog content will be stored and served from") - flag.StringVar(&serverPort, "catalogs-server-port", ":8083", "The port where the unpacked catalogs' content will be accessible") + flag.StringVar(&storageDir, "catalogs-storage-dir", "/var/cache/catalogs", "The directory in the filesystem where unpacked catalog content will be stored and served from") + flag.StringVar(&catalogServerAddr, "catalogs-server-addr", "127.0.0.1:8083", "The port where the unpacked catalogs' content will be accessible") flag.BoolVar(&profiling, "profiling", false, "enable profiling endpoints to allow for using pprof") flag.BoolVar(&catalogdVersion, "version", false, "print the catalogd version and exit") opts := zap.Options{ @@ -122,16 +123,22 @@ func main() { os.Exit(1) } - catalogServer := catalogserver.NewServer(storageDir, serverPort) + serverMux := &http.ServeMux{} + catalogServer := catalogserver.New(storageDir, catalogServerAddr, serverMux) if err := mgr.Add(catalogServer); err != nil { setupLog.Error(err, "unable to start catalog server") os.Exit(1) } + if _, err := os.Stat(storageDir); os.IsNotExist(err) { + if err := os.MkdirAll(storageDir, 0700); err != nil { + setupLog.Error(err, "unable to create storage directory for catalogs") + } + } if err = (&corecontrollers.CatalogReconciler{ Client: mgr.GetClient(), Unpacker: unpacker, - Storage: storage.NewStorage(storageDir, catalogServer.Mux), + Storage: storage.NewStorage(storageDir, serverMux), }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "Catalog") os.Exit(1) diff --git a/pkg/catalogserver/server.go b/pkg/catalogserver/server.go index e8809a64..755098ab 100644 --- a/pkg/catalogserver/server.go +++ b/pkg/catalogserver/server.go @@ -9,26 +9,27 @@ import ( "time" ) -// Server is a manager.Runnable Server, that serves the FBC -// content of the extension catalogs added to the cluster -type Server struct { +// Instance is a manager.Runnable catalog server instance, +// that serves the FBC content of the extension catalogs +// added to the cluster +type Instance struct { Dir string Port string Mux *http.ServeMux } -// NewServer takes directory and port number, and returns -// a Server that serves the FBC content stored in the -// directory on the given port number. -func NewServer(dir, port string) Server { - return Server{ +// New returns an Instance of a catalog server that serves +// the FBC content stored in the given directory on the given +// http address. +func New(dir, port string, mux *http.ServeMux) Instance { + return Instance{ Dir: dir, Port: port, - Mux: &http.ServeMux{}, + Mux: mux, } } -func (s Server) Start(_ context.Context) error { +func (s Instance) Start(ctx context.Context) error { s.Mux.HandleFunc("/catalogs", func(w http.ResponseWriter, r *http.Request) { files, err := os.ReadDir(s.Dir) if err != nil { @@ -46,5 +47,17 @@ func (s Server) Start(_ context.Context) error { Addr: s.Port, ReadHeaderTimeout: 3 * time.Second, } - return server.ListenAndServe() + e := make(chan error) + go func(server *http.Server, e chan error) { + err := server.ListenAndServe() + e <- err + close(e) + }(server, e) + err := <-e + if err != nil { + return err + } + + <-ctx.Done() + return server.Shutdown(context.TODO()) }