From 7611cd963c3af3722f1317427eeec0f9421fb7b2 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 11 May 2017 14:02:06 +0000 Subject: [PATCH] Don't return OK from status until all parts are ready --- api/weaveapi.go | 8 +++++--- common/wait.go | 25 +++++++++++++++++++++++++ plugin/plugin.go | 7 ++++--- prog/weaver/http.go | 11 +++++++++-- prog/weaver/main.go | 15 +++++++++------ proxy/proxy.go | 6 +++++- 6 files changed, 57 insertions(+), 15 deletions(-) create mode 100644 common/wait.go diff --git a/api/weaveapi.go b/api/weaveapi.go index 19a567a2..452ad0b5 100644 --- a/api/weaveapi.go +++ b/api/weaveapi.go @@ -78,10 +78,12 @@ func (client *Client) Connect(remote string) error { } // IsReady returns true if the API server is up and running +// (note it returns StatusServiceUnavailable until all parts are ready, +// but callers of this function are expected to be ok with a partial service) func (client *Client) IsReady() bool { - _, err := client.httpVerb("GET", "/status", nil) - - return err == nil + resp, err := http.Get(client.baseURL + "/status") + return err == nil && + (resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusServiceUnavailable) } // WaitAPIServer waits until the API server is ready to serve. diff --git a/common/wait.go b/common/wait.go new file mode 100644 index 00000000..1d625b7c --- /dev/null +++ b/common/wait.go @@ -0,0 +1,25 @@ +package common + +import ( + "sync" + "sync/atomic" +) + +type WaitGroup struct { + wg sync.WaitGroup + count int32 +} + +// Increase the count of things waiting and return a closure which calls Done +func (w *WaitGroup) Add() func() { + atomic.AddInt32(&w.count, 1) + w.wg.Add(1) + return func() { + w.wg.Done() + atomic.AddInt32(&w.count, -1) + } +} + +func (w *WaitGroup) IsDone() bool { + return atomic.LoadInt32(&w.count) == 0 +} diff --git a/plugin/plugin.go b/plugin/plugin.go index 85bd4fc4..ddbc7327 100644 --- a/plugin/plugin.go +++ b/plugin/plugin.go @@ -23,19 +23,19 @@ const ( var Log = common.Log -func Start(weaveAPIAddr string, dockerClient *docker.Client, address string, meshAddress string, dns bool, isPluginV2, forceMulticast bool, defaultSubnet string) { +func Start(weaveAPIAddr string, dockerClient *docker.Client, address string, meshAddress string, dns bool, isPluginV2, forceMulticast bool, defaultSubnet string, ready func()) { weave := weaveapi.NewClient(weaveAPIAddr, Log) Log.Info("Waiting for Weave API Server...") weave.WaitAPIServer(30) Log.Info("Finished waiting for Weave API Server") - if err := run(dockerClient, weave, address, meshAddress, dns, isPluginV2, forceMulticast, defaultSubnet); err != nil { + if err := run(dockerClient, weave, address, meshAddress, dns, isPluginV2, forceMulticast, defaultSubnet, ready); err != nil { Log.Fatal(err) } } -func run(dockerClient *docker.Client, weave *weaveapi.Client, address, meshAddress string, dns, isPluginV2, forceMulticast bool, defaultSubnet string) error { +func run(dockerClient *docker.Client, weave *weaveapi.Client, address, meshAddress string, dns, isPluginV2, forceMulticast bool, defaultSubnet string, ready func()) error { endChan := make(chan error, 1) if address != "" { @@ -60,6 +60,7 @@ func run(dockerClient *docker.Client, weave *weaveapi.Client, address, meshAddre // TODO: the driver name should be extracted from pluginMeshSocket dockerClient.EnsureNetwork("weave", "weavemesh", defaultSubnet, options) } + ready() return <-endChan } diff --git a/prog/weaver/http.go b/prog/weaver/http.go index ccaae4af..f18ac745 100644 --- a/prog/weaver/http.go +++ b/prog/weaver/http.go @@ -13,6 +13,7 @@ import ( "github.com/weaveworks/mesh" "github.com/weaveworks/go-checkpoint" + "github.com/weaveworks/weave/common" "github.com/weaveworks/weave/ipam" "github.com/weaveworks/weave/nameserver" "github.com/weaveworks/weave/net/address" @@ -273,6 +274,7 @@ func (v *VersionCheck) String() string { } type WeaveStatus struct { + Ready bool Version string VersionCheck *VersionCheck `json:"VersionCheck,omitempty"` Router *weave.NetworkRouterStatus `json:"Router,omitempty"` @@ -281,9 +283,10 @@ type WeaveStatus struct { } // Read-only functions, suitable for exposing on an unprotected socket -func HandleHTTP(muxRouter *mux.Router, version string, router *weave.NetworkRouter, allocator *ipam.Allocator, defaultSubnet address.CIDR, ns *nameserver.Nameserver, dnsserver *nameserver.DNSServer) { +func HandleHTTP(muxRouter *mux.Router, version string, router *weave.NetworkRouter, allocator *ipam.Allocator, defaultSubnet address.CIDR, ns *nameserver.Nameserver, dnsserver *nameserver.DNSServer, waitReady *common.WaitGroup) { status := func() WeaveStatus { return WeaveStatus{ + waitReady.IsDone(), version, versionCheck(), weave.NewNetworkRouterStatus(router), @@ -324,7 +327,11 @@ func HandleHTTP(muxRouter *mux.Router, version string, router *weave.NetworkRout defHandler := func(path string, template *template.Template) { muxRouter.Methods("GET").Path(path).HandlerFunc( func(w http.ResponseWriter, r *http.Request) { - if err := template.Execute(w, status()); err != nil { + s := status() + if !s.Ready { + w.WriteHeader(http.StatusServiceUnavailable) + } + if err := template.Execute(w, s); err != nil { http.Error(w, "error during template execution", http.StatusInternalServerError) Log.Error(err) } diff --git a/prog/weaver/main.go b/prog/weaver/main.go index be787916..ed0fb5a9 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -255,6 +255,8 @@ func main() { } config.ProtocolMinVersion = byte(protocolMinVersion) + var waitReady common.WaitGroup + if proxyConfig.Enabled { // Start Weave Proxy: proxy, err := weaveproxy.NewProxy(*proxyConfig) @@ -264,7 +266,7 @@ func main() { defer proxy.Stop() listeners := proxy.Listen() proxy.AttachExistingContainers() - go proxy.Serve(listeners) + go proxy.Serve(listeners, waitReady.Add()) go proxy.ListenAndServeStatus("/home/weave/status.sock") } @@ -432,7 +434,7 @@ func main() { ns.HandleHTTP(muxRouter, dockerCli) } router.HandleHTTP(muxRouter) - HandleHTTP(muxRouter, version, router, allocator, defaultSubnet, ns, dnsserver) + HandleHTTP(muxRouter, version, router, allocator, defaultSubnet, ns, dnsserver, &waitReady) HandleHTTPPeer(muxRouter, allocator, discoveryEndpoint, token, name.String()) muxRouter.Methods("GET").Path("/metrics").Handler(metricsHandler(router, allocator, ns, dnsserver)) http.Handle("/", common.LoggingHTTPHandler(muxRouter)) @@ -442,7 +444,7 @@ func main() { if statusAddr != "" { muxRouter := mux.NewRouter() - HandleHTTP(muxRouter, version, router, allocator, defaultSubnet, ns, dnsserver) + HandleHTTP(muxRouter, version, router, allocator, defaultSubnet, ns, dnsserver, &waitReady) muxRouter.Methods("GET").Path("/metrics").Handler(metricsHandler(router, allocator, ns, dnsserver)) statusMux := http.NewServeMux() statusMux.Handle("/", muxRouter) @@ -451,26 +453,27 @@ func main() { } if enablePlugin || enablePluginV2 { - go plugin.Start(httpAddr, dockerCli, pluginSocket, pluginMeshSocket, !noDNS, enablePluginV2, enablePluginV2Multicast, defaultSubnet.String()) + go plugin.Start(httpAddr, dockerCli, pluginSocket, pluginMeshSocket, !noDNS, enablePluginV2, enablePluginV2Multicast, defaultSubnet.String(), waitReady.Add()) } if bridgeConfig.AWSVPC { // Run this on its own goroutine because the allocator can block // We remove the default route installed by the kernel, // because awsvpc has installed it as well - go expose(allocator, defaultSubnet, bridgeConfig.WeaveBridgeName, bridgeConfig.AWSVPC) + go expose(allocator, defaultSubnet, bridgeConfig.WeaveBridgeName, bridgeConfig.AWSVPC, waitReady.Add()) } signals.SignalHandlerLoop(common.Log, router) } -func expose(alloc *ipam.Allocator, subnet address.CIDR, bridgeName string, removeDefaultRoute bool) { +func expose(alloc *ipam.Allocator, subnet address.CIDR, bridgeName string, removeDefaultRoute bool, ready func()) { addr, err := alloc.Allocate("weave:expose", subnet, false, func() bool { return false }) checkFatal(err) cidr := address.MakeCIDR(subnet, addr) err = weavenet.AddBridgeAddr(bridgeName, cidr.IPNet(), removeDefaultRoute) checkFatal(err) Log.Printf("Bridge %q exposed on address %v", bridgeName, cidr) + ready() } func options() map[string]string { diff --git a/proxy/proxy.go b/proxy/proxy.go index 53d770ee..846a60d4 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -280,13 +280,17 @@ func (proxy *Proxy) Listen() []net.Listener { return listeners } -func (proxy *Proxy) Serve(listeners []net.Listener) { +func (proxy *Proxy) Serve(listeners []net.Listener, ready func()) { errs := make(chan error) for _, listener := range listeners { go func(listener net.Listener) { errs <- (&http.Server{Handler: proxy}).Serve(listener) }(listener) } + // It would be better if we could delay calling Done() until all + // the listeners are ready, but it doesn't seem to be possible to + // hook the right point in http.Server + ready() for range listeners { err := <-errs if err != nil {