nsqd: pre-create topics/channels or wait until after init grace period

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")
 

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"
 

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)

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 {

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 {

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)

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,