Merge pull request #1065 from nats-io/fix_1064

[FIXED] NATS debug/trace may not be honored after reload+reopen signals

logger/logger.go

@@ -73,13 +73,26 @@ func (s *StanLogger) SetFileSizeLimit(limit int64) {
 func (s *StanLogger) SetLoggerWithOpts(log Logger, nOpts *natsd.Options, debug, trace bool) {
 	s.mu.Lock()
 	s.log = log
-	s.ltime = nOpts.Logtime
 	s.debug = debug
 	s.trace = trace
+	s.updateNATSOptions(nOpts)
+	s.mu.Unlock()
+}
+
+func (s *StanLogger) updateNATSOptions(nOpts *natsd.Options) {
+	s.ltime = nOpts.Logtime
 	s.lfile = nOpts.LogFile
 	s.fszl = nOpts.LogSizeLimit
 	s.ndbg = nOpts.Debug
 	s.ntrc = nOpts.Trace
+}
+
+// UpdateNATSOptions refreshes the NATS related options, for instance after a
+// configuration reload, so that if ReopenLogFile() is called, the logger new
+// options are applied.
+func (s *StanLogger) UpdateNATSOptions(nOpts *natsd.Options) {
+	s.mu.Lock()
+	s.updateNATSOptions(nOpts)
 	s.mu.Unlock()
 }
 

server/signal.go

@@ -53,10 +53,17 @@ func (s *StanServer) handleSignals() {
 				case syscall.SIGHUP:
 					s.mu.Lock()
 					ns := s.natsServer
+					nobr := s.natsOpts
 					s.mu.Unlock()
 					if ns != nil {
 						if err := ns.Reload(); err != nil {
 							s.log.Errorf("Reload: %v", err)
+						} else if fileOpts, err := natsd.ProcessConfigFile(nobr.ConfigFile); err == nil {
+							newOpts := natsd.MergeOptions(fileOpts, nobr)
+							s.mu.Lock()
+							s.natsOpts = newOpts.Clone()
+							s.log.UpdateNATSOptions(s.natsOpts)
+							s.mu.Unlock()
 						}
 					} else {
 						s.log.Warnf("Reload supported only for embedded NATS Server's configuration")

server/signal_test.go

@@ -208,3 +208,63 @@ func TestSignalReload(t *testing.T) {
 		t.Fatalf("Did not get the Reload trace (make sure you did a `go install` prior to running the test")
 	}
 }
+
+func TestNATSServerConfigReloadFollowedByReopen(t *testing.T) {
+	lfile := "nss.log"
+	ctemp := `
+	logfile: "%s"
+	debug: false
+	trace: %s
+	streaming {
+		sd: false
+		sv: false
+	}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(ctemp, lfile, "false")))
+	defer os.Remove(conf)
+	defer os.Remove(lfile)
+
+	// This test requires that the server be installed.
+	cmd := exec.Command("nats-streaming-server", "-c", conf)
+	cmd.Start()
+	// Wait for it to print some startup trace
+	waitFor(t, 2*time.Second, 50*time.Millisecond, func() error {
+		content, err := ioutil.ReadFile(lfile)
+		if err != nil {
+			return err
+		}
+		if bytes.Contains(content, []byte(streamingReadyLog)) {
+			return nil
+		}
+		return fmt.Errorf("process not started yet, make sure you `go install` first!")
+	})
+	changeCurrentConfigContentWithNewContent(t, conf, []byte(fmt.Sprintf(ctemp, lfile, "true")))
+	syscall.Kill(cmd.Process.Pid, syscall.SIGHUP)
+	time.Sleep(500 * time.Millisecond)
+	c := NewDefaultConnection(t)
+	c.Publish("foo", []byte("hello"))
+	c.Close()
+	syscall.Kill(cmd.Process.Pid, syscall.SIGUSR1)
+	time.Sleep(500 * time.Millisecond)
+	c = NewDefaultConnection(t)
+	c.Publish("bar", []byte("hello"))
+	c.Close()
+	syscall.Kill(cmd.Process.Pid, syscall.SIGINT)
+	cmd.Wait()
+	content, err := ioutil.ReadFile(lfile)
+	if err != nil {
+		t.Fatalf("Error reading log file: %v", err)
+	}
+
+	if n := bytes.Count(content, []byte("CONNECT {")); n != 2 {
+		t.Fatalf("Expected 2 connects, got %v\n%s\n", n, content)
+	}
+	idx := bytes.LastIndex(content, []byte("CONNECT {"))
+	backward := content[idx-150 : idx]
+	start := bytes.LastIndexByte(backward, '\n')
+	line := backward[start:]
+	// If we lost notion of logtime, we would have [pid] [TRC] directly..
+	if bytes.Contains(line, []byte("] [TRC] ")) {
+		t.Fatalf("Logtime was lost during reload: %q", line)
+	}
+}