diff --git a/apps/nsqd/options.go b/apps/nsqd/options.go index 731681ea4..701256667 100644 --- a/apps/nsqd/options.go +++ b/apps/nsqd/options.go @@ -128,6 +128,7 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet { flagSet.Int("broadcast-http-port", opts.BroadcastHTTPPort, "HTTP port that will be registered with lookupd (defaults to the HTTP port that this nsqd is listening on)") lookupdTCPAddrs := app.StringArray{} flagSet.Var(&lookupdTCPAddrs, "lookupd-tcp-address", "lookupd TCP address (may be given multiple times)") + flagSet.Duration("intial-grace-period", opts.InitialGracePeriod, "time to retry startup nsqlookupd sync before opening for traffic (max 300s)") flagSet.Duration("http-client-connect-timeout", opts.HTTPClientConnectTimeout, "timeout for HTTP connect") flagSet.Duration("http-client-request-timeout", opts.HTTPClientRequestTimeout, "timeout for HTTP request") diff --git a/contrib/nsqd.cfg.example b/contrib/nsqd.cfg.example index 15033268e..f1e6d1b9e 100644 --- a/contrib/nsqd.cfg.example +++ b/contrib/nsqd.cfg.example @@ -21,6 +21,9 @@ nsqlookupd_tcp_addresses = [ "127.0.0.1:4160" ] +## time to retry startup nsqlookupd sync before opening for traffic (max 300s) +intial_grace_period = "0s" + ## duration to wait before HTTP client connection timeout http_client_connect_timeout = "2s" diff --git a/nsqd/README.md b/nsqd/README.md index a2becf5a2..4318db767 100644 --- a/nsqd/README.md +++ b/nsqd/README.md @@ -2,4 +2,6 @@ `nsqd` is the daemon that receives, queues, and delivers messages to clients. +When `nsqd` is configured with `nsqlookupd_tcp_addresses`, `intial-grace-period` controls how long startup will wait for the initial nsqlookupd sync and topic/channel pre-creation before opening for message traffic. The default is `0s`, which means `nsqd` will open immediately if lookupd is unavailable; the maximum supported value is `300s`. + Read the [docs](https://nsq.io/components/nsqd.html) diff --git a/nsqd/lookup.go b/nsqd/lookup.go index 953a450eb..83d2d5318 100644 --- a/nsqd/lookup.go +++ b/nsqd/lookup.go @@ -12,6 +12,8 @@ import ( "github.com/nsqio/nsq/internal/version" ) +const lookupdBootstrapRetryInterval = 100 * time.Millisecond + func connectCallback(n *NSQD, hostname string) func(*lookupPeer) { return func(lp *lookupPeer) { ci := make(map[string]interface{}) @@ -75,17 +77,75 @@ func connectCallback(n *NSQD, hostname string) func(*lookupPeer) { } } -func (n *NSQD) lookupLoop() { +func (n *NSQD) lookupLoop(startupSyncCh chan struct{}) { var lookupPeers []*lookupPeer var lookupAddrs []string - connect := true + connect := false + signalStartup := func() { + if startupSyncCh == nil { + return + } + close(startupSyncCh) + startupSyncCh = nil + } hostname, err := os.Hostname() if err != nil { + signalStartup() n.logf(LOG_FATAL, "failed to get hostname - %s", err) os.Exit(1) } + if len(n.getOpts().NSQLookupdTCPAddresses) > 0 { + deadline := time.Now().Add(n.getOpts().InitialGracePeriod) + for { + lookupPeers = nil + lookupAddrs = nil + for _, host := range n.getOpts().NSQLookupdTCPAddresses { + n.logf(LOG_INFO, "LOOKUP(%s): adding peer", host) + lookupPeer := newLookupPeer(host, n.getOpts().MaxBodySize, n.logf, + connectCallback(n, hostname)) + if _, err := lookupPeer.Command(nil); err != nil { + n.logf(LOG_ERROR, "LOOKUP(%s): failed to connect - %s", host, err) + lookupPeer.Close() + continue + } + lookupPeers = append(lookupPeers, lookupPeer) + lookupAddrs = append(lookupAddrs, host) + } + + if len(lookupPeers) > 0 { + n.lookupPeers.Store(lookupPeers) + if err := n.preCreateTopicsFromLookupd(); err == nil { + break + } else { + n.logf(LOG_WARN, "startup lookupd sync failed - %s", err) + } + } + + if !time.Now().Before(deadline) { + if len(lookupPeers) == 0 { + n.logf(LOG_WARN, "startup lookupd sync grace period expired; opening for traffic before connecting to nsqlookupd") + } else { + n.logf(LOG_WARN, "startup lookupd sync grace period expired; opening for traffic before topic/channel pre-creation completed") + } + connect = len(lookupPeers) == 0 + break + } + + select { + case <-time.After(lookupdBootstrapRetryInterval): + case <-n.exitChan: + signalStartup() + return + } + } + if len(lookupPeers) > 0 { + n.lookupPeers.Store(lookupPeers) + } + } + signalStartup() + // for announcements, lookupd determines the host automatically ticker := time.Tick(15 * time.Second) for { diff --git a/nsqd/nsqd.go b/nsqd/nsqd.go index d918b7856..4dc66ae50 100644 --- a/nsqd/nsqd.go +++ b/nsqd/nsqd.go @@ -116,6 +116,10 @@ func New(opts *Options) (*NSQD, error) { return nil, errors.New("--node-id must be [0,1024)") } + if opts.InitialGracePeriod < 0 || opts.InitialGracePeriod > 300*time.Second { + return nil, errors.New("--intial-grace-period must be [0,300s]") + } + if opts.TLSClientAuthPolicy != "" && opts.TLSRequired == TLSNotRequired { opts.TLSRequired = TLSRequired } @@ -239,6 +243,12 @@ func (n *NSQD) Main() error { }) } + lookupSyncCh := make(chan struct{}) + n.waitGroup.Wrap(func() { + n.lookupLoop(lookupSyncCh) + }) + <-lookupSyncCh + n.waitGroup.Wrap(func() { exitFunc(protocol.TCPServer(n.tcpListener, n.tcpServer, n.logf)) }) @@ -255,7 +265,6 @@ func (n *NSQD) Main() error { } n.waitGroup.Wrap(n.queueScanLoop) - n.waitGroup.Wrap(n.lookupLoop) if n.getOpts().StatsdAddress != "" { n.waitGroup.Wrap(n.statsdLoop) } @@ -479,25 +488,70 @@ func (n *NSQD) GetTopic(topicName string) *Topic { // if using lookupd, make a blocking call to get channels and immediately create them // to ensure that all channels receive published messages + err := n.syncLookupdTopicChannels(t, false) + if err != nil { + n.logf(LOG_WARN, "%s", err) + } + + // now that all channels are added, start topic messagePump + t.Start() + return t +} + +func (n *NSQD) syncLookupdTopicChannels(t *Topic, failOnError bool) error { lookupdHTTPAddrs := n.lookupdHTTPAddrs() if len(lookupdHTTPAddrs) > 0 { channelNames, err := n.ci.GetLookupdTopicChannels(t.name, lookupdHTTPAddrs) if err != nil { - n.logf(LOG_WARN, "failed to query nsqlookupd for channels to pre-create for topic %s - %s", t.name, err) + return fmt.Errorf("failed to query nsqlookupd for channels to pre-create for topic %s - %s", t.name, err) } for _, channelName := range channelNames { if strings.HasSuffix(channelName, "#ephemeral") { - continue // do not create ephemeral channel with no consumer client + continue } t.GetChannel(channelName) } - } else if len(n.getOpts().NSQLookupdTCPAddresses) > 0 { - n.logf(LOG_ERROR, "no available nsqlookupd to query for channels to pre-create for topic %s", t.name) + return nil } - // now that all channels are added, start topic messagePump - t.Start() - return t + if failOnError && len(n.getOpts().NSQLookupdTCPAddresses) > 0 { + return fmt.Errorf("no available nsqlookupd to query for channels to pre-create for topic %s", t.name) + } + if len(n.getOpts().NSQLookupdTCPAddresses) > 0 { + return fmt.Errorf("no available nsqlookupd to query for channels to pre-create for topic %s", t.name) + } + return nil +} + +func (n *NSQD) preCreateTopicsFromLookupd() error { + lookupdHTTPAddrs := n.lookupdHTTPAddrs() + if len(lookupdHTTPAddrs) == 0 { + if len(n.getOpts().NSQLookupdTCPAddresses) == 0 { + return nil + } + return errors.New("no available nsqlookupd to query for startup topic/channel pre-creation") + } + + topicNames, err := n.ci.GetLookupdTopics(lookupdHTTPAddrs) + if err != nil { + return fmt.Errorf("failed to query nsqlookupd for topics to pre-create - %s", err) + } + + for _, topicName := range topicNames { + if strings.HasSuffix(topicName, "#ephemeral") { + continue + } + if !protocol.IsValidTopicName(topicName) { + n.logf(LOG_WARN, "skipping creation of invalid topic %s", topicName) + continue + } + topic := n.GetTopic(topicName) + if err := n.syncLookupdTopicChannels(topic, true); err != nil { + return err + } + } + + return nil } // GetExistingTopic gets a topic only if it exists @@ -577,8 +631,7 @@ func (n *NSQD) channels() []*Channel { // resizePool adjusts the size of the pool of queueScanWorker goroutines // -// 1 <= pool <= min(num * 0.25, QueueScanWorkerPoolMax) -// +// 1 <= pool <= min(num * 0.25, QueueScanWorkerPoolMax) func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int) { idealPoolSize := int(float64(num) * 0.25) if idealPoolSize < 1 { diff --git a/nsqd/nsqd_test.go b/nsqd/nsqd_test.go index b39df44e8..fc8045a44 100644 --- a/nsqd/nsqd_test.go +++ b/nsqd/nsqd_test.go @@ -415,6 +415,94 @@ func TestCluster(t *testing.T) { test.Equal(t, 0, len(dd["channel:"+topicName+":ch"])) } +func TestStartupPreCreatesTopicsAndChannelsFromLookupd(t *testing.T) { + lopts := nsqlookupd.NewOptions() + lopts.Logger = test.NewTestLogger(t) + lopts.BroadcastAddress = "127.0.0.1" + _, _, lookupd := mustStartNSQLookupd(lopts) + defer lookupd.Exit() + + seedOpts := NewOptions() + seedOpts.Logger = test.NewTestLogger(t) + seedOpts.NSQLookupdTCPAddresses = []string{lookupd.RealTCPAddr().String()} + seedOpts.BroadcastAddress = "127.0.0.1" + _, _, seedNSQD := mustStartNSQD(seedOpts) + defer os.RemoveAll(seedOpts.DataPath) + defer seedNSQD.Exit() + + topicName := "startup_lookupd_precreate_" + strconv.Itoa(int(time.Now().UnixNano())) + seedTopic := seedNSQD.GetTopic(topicName) + seedTopic.GetChannel("ch") + + time.Sleep(350 * time.Millisecond) + + targetOpts := NewOptions() + targetOpts.Logger = test.NewTestLogger(t) + targetOpts.NSQLookupdTCPAddresses = []string{lookupd.RealTCPAddr().String()} + targetOpts.BroadcastAddress = "127.0.0.1" + _, _, targetNSQD := mustStartNSQD(targetOpts) + defer os.RemoveAll(targetOpts.DataPath) + defer targetNSQD.Exit() + + var targetTopic *Topic + var targetChannel *Channel + for i := 0; i < 100; i++ { + topic, err := targetNSQD.GetExistingTopic(topicName) + if err == nil { + channel, channelErr := topic.GetExistingChannel("ch") + if channelErr == nil { + targetTopic = topic + targetChannel = channel + break + } + } + time.Sleep(10 * time.Millisecond) + } + + test.NotNil(t, targetTopic) + test.NotNil(t, targetChannel) + + body := []byte("bootstrapped") + msg := NewMessage(targetTopic.GenerateID(), body) + err := targetTopic.PutMessage(msg) + test.Nil(t, err) + + var received *Message + select { + case received = <-targetChannel.memoryMsgChan: + case b := <-targetChannel.backend.ReadChan(): + received, _ = decodeMessage(b) + case <-time.After(time.Second): + t.Fatalf("timed out waiting for bootstrapped channel to receive a message") + } + + test.Equal(t, body, received.Body) +} + +func TestStartupLookupdGracePeriodExpires(t *testing.T) { + opts := NewOptions() + opts.Logger = test.NewTestLogger(t) + opts.NSQLookupdTCPAddresses = []string{"127.0.0.1:1"} + opts.InitialGracePeriod = 50 * time.Millisecond + _, httpAddr, nsqd := mustStartNSQD(opts) + defer os.RemoveAll(opts.DataPath) + defer nsqd.Exit() + + var info struct { + TCPPort int `json:"tcp_port"` + } + url := fmt.Sprintf("http://%s/info", httpAddr) + for i := 0; i < 50; i++ { + err := http_api.NewClient(nil, ConnectTimeout, RequestTimeout).GETV1(url, &info) + if err == nil && info.TCPPort > 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + + test.Equal(t, nsqd.RealTCPAddr().Port, info.TCPPort) +} + func TestSetHealth(t *testing.T) { opts := NewOptions() opts.Logger = test.NewTestLogger(t) diff --git a/nsqd/options.go b/nsqd/options.go index 30fe5ea1e..ee0c6364c 100644 --- a/nsqd/options.go +++ b/nsqd/options.go @@ -26,6 +26,7 @@ type Options struct { BroadcastTCPPort int `flag:"broadcast-tcp-port"` BroadcastHTTPPort int `flag:"broadcast-http-port"` NSQLookupdTCPAddresses []string `flag:"lookupd-tcp-address" cfg:"nsqlookupd_tcp_addresses"` + InitialGracePeriod time.Duration `flag:"intial-grace-period" cfg:"intial_grace_period"` AuthHTTPAddresses []string `flag:"auth-http-address" cfg:"auth_http_addresses"` HTTPClientConnectTimeout time.Duration `flag:"http-client-connect-timeout" cfg:"http_client_connect_timeout"` HTTPClientRequestTimeout time.Duration `flag:"http-client-request-timeout" cfg:"http_client_request_timeout"` @@ -109,6 +110,7 @@ func NewOptions() *Options { BroadcastHTTPPort: 0, NSQLookupdTCPAddresses: make([]string, 0), + InitialGracePeriod: 0, AuthHTTPAddresses: make([]string, 0), HTTPClientConnectTimeout: 2 * time.Second,