Repository: nats-io/nats-server
Pull request #3774 (2023-01-10 15:45:01)

Add `profilez` server endpoint
This PR adds a `profilez` endpoint to the system account, which accepts two parameters: `profile` for the name of the profile (i.e. `allocs`, `block`, `goroutine`, `heap`, `mutex`, `threadcreate`) and `debug` for the return format (defaults to `0`, which is usually gzipped for `go tool pprof`).

It will only be possible to query this endpoint if `PROF_PORT` is already enabled, otherwise it will return a `Profiling is not enabled` error.

Signed-off-by: Neil Twigg <neil@nats.io>

/cc @nats-io/core


ripienaar: We also have configuration option in the server to enable it at runtime and when embedded. This kind of thing I think should be an opt-in as it can have runtime performance impact on the servers
neilalexander: > This kind of thing I think should be an opt-in as it can have runtime performance impact on the servers

So far the only thing that `StartProfiler` does that has any overhead is to call `runtime.SetBlockProfileRate`, which `profilez` does not do admittedly, so maybe we can just guard off the `block` profile unless `StartProfiler` has been called.

There's no runtime overhead on things like `allocs`, `goroutine`, `heap`, `threadcreate` as they just are internally extracting statistics out of the `runtime` package.
ripienaar: There's certainly some cost associated with gathering a profile though - if its same as when you hit the normal pprof end points the go team aims to keep the impact aroun 5%, its still an impact though and a potential vector for abuse that someone might want to turn off.
neilalexander: I should have probably noted in the original PR that it isn't actually possible to collect a CPU profile via `profilez` right now (like you would with `/debug/pprof/profile`) because the `pprof` package registers that straight onto `DefaultServeMux` and not into the `profiles` struct, so the only accessible profiles [are these](https://cs.opensource.google/go/go/+/refs/tags/go1.19.4:src/runtime/pprof/pprof.go;l=182-195;drc=0202ad0b3a2bfddf9f3eafb94e19d5a0fa3d1f31). 

I can make it opt-in, although it doesn't look as though the monitoring port actually has a configuration section of its own, so I guess I could do one of the following:
1. Add a `http_profiling` option, to control whether or not it should be exposed on both the monitoring port and the system account;
2. Only expose it on the system account only (and not the HTTP port) when `PROF_PORT` is already enabled?
ripienaar: 2nd options sounds good - the current `/debug/pprof` stuff is nice cos its just exactly what `go tool pprof` wants for real time debugging etc. so for HTTP I think its great as is, then just additionally expose on system account once opted in is great
derekcollison: Please do a rebase onto dev vs merge.

Patch:
diff --git a/server/events.go b/server/events.go
index dce08086..8a3391ce 100644
--- a/server/events.go
+++ b/server/events.go
@@ -949,6 +949,10 @@ func (s *Server) initEventTracking() {
 			optz := &HealthzEventOptions{}
 			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.healthz(&optz.HealthzOptions), nil })
 		},
+		"PROFILEZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
+			optz := &ProfilezEventOptions{}
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.profilez(&optz.ProfilezOptions), nil })
+		},
 	}
 	for name, req := range monSrvc {
 		subject = fmt.Sprintf(serverDirectReqSubj, s.info.ID, name)
@@ -1574,6 +1578,12 @@ type HealthzEventOptions struct {
 	EventFilterOptions
 }
 
+// In the context of system events, ProfilezEventOptions are options passed to Profilez
+type ProfilezEventOptions struct {
+	ProfilezOptions
+	EventFilterOptions
+}
+
 // returns true if the request does NOT apply to this server and can be ignored.
 // DO NOT hold the server lock when
 func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 8a49ecad..f4b703ad 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -3744,12 +3744,11 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
 		} else {
 			streamName = streamNameFromSubject(subject)
 			consumerName = consumerNameFromSubject(subject)
-			// New has optional filtered subject as part of main subject..
-			if n > 6 {
-
-				tokens := strings.Split(subject, tsep)
-				filteredSubject = strings.Join(tokens[6:], tsep)
-			}
+		}
+		// New has optional filtered subject as part of main subject..
+		if n > 7 {
+			tokens := strings.Split(subject, tsep)
+			filteredSubject = strings.Join(tokens[6:], tsep)
 		}
 	}
 
diff --git a/server/monitor.go b/server/monitor.go
index 62beb564..4fb32c6e 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -14,6 +14,7 @@
 package server
 
 import (
+	"bytes"
 	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
@@ -26,6 +27,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"runtime/pprof"
 	"sort"
 	"strconv"
 	"strings"
@@ -2639,6 +2641,12 @@ type HealthzOptions struct {
 	JSServerOnly  bool `json:"js-server-only,omitempty"`
 }
 
+// ProfilezOptions are options passed to Profilez
+type ProfilezOptions struct {
+	Name  string `json:"name"`
+	Debug int    `json:"debug"`
+}
+
 type StreamDetail struct {
 	Name     string              `json:"name"`
 	Created  time.Time           `json:"created"`
@@ -3123,3 +3131,36 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 	// Success.
 	return health
 }
+
+type ProfilezStatus struct {
+	Profile []byte `json:"profile"`
+	Error   string `json:"error"`
+}
+
+func (s *Server) profilez(opts *ProfilezOptions) *ProfilezStatus {
+	if s.profiler == nil {
+		return &ProfilezStatus{
+			Error: "Profiling is not enabled",
+		}
+	}
+	if opts.Name == _EMPTY_ {
+		return &ProfilezStatus{
+			Error: "Profile name not specified",
+		}
+	}
+	profile := pprof.Lookup(opts.Name)
+	if profile == nil {
+		return &ProfilezStatus{
+			Error: fmt.Sprintf("Profile %q not found", opts.Name),
+		}
+	}
+	var buffer bytes.Buffer
+	if err := profile.WriteTo(&buffer, opts.Debug); err != nil {
+		return &ProfilezStatus{
+			Error: fmt.Sprintf("Profile %q error: %s", opts.Name, err),
+		}
+	}
+	return &ProfilezStatus{
+		Profile: buffer.Bytes(),
+	}
+}


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index c6fc99d7..05eeb1fd 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1661,7 +1661,7 @@ func TestSystemAccountWithGateways(t *testing.T) {
 
 	// If this tests fails with wrong number after 10 seconds we may have
 	// added a new inititial subscription for the eventing system.
-	checkExpectedSubs(t, 50, sa)
+	checkExpectedSubs(t, 52, sa)
 
 	// Create a client on B and see if we receive the event
 	urlb := fmt.Sprintf("nats://%s:%d", ob.Host, ob.Port)
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 844a944f..65381f1d 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -19216,45 +19216,6 @@ func TestJetStreamMetaDataFailOnKernelFault(t *testing.T) {
 	require_True(t, si.State.Msgs == 10)
 }
 
-func TestJetstreamConsumerSingleTokenSubject(t *testing.T) {
-
-	s := RunBasicJetStreamServer(t)
-	defer s.Shutdown()
-
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	filterSubject := "foo"
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{filterSubject},
-	})
-	require_NoError(t, err)
-
-	req, err := json.Marshal(&CreateConsumerRequest{Stream: "TEST", Config: ConsumerConfig{
-		FilterSubject: filterSubject,
-		Name:          "name",
-	}})
-
-	if err != nil {
-		t.Fatalf("failed to marshal consumer create request: %v", err)
-	}
-
-	resp, err := nc.Request(fmt.Sprintf("$JS.API.CONSUMER.CREATE.%s.%s.%s", "TEST", "name", "not_filter_subject"), req, time.Second*10)
-
-	var apiResp ApiResponse
-	json.Unmarshal(resp.Data, &apiResp)
-	if err != nil {
-		t.Fatalf("failed to unmarshal response: %v", err)
-	}
-	if apiResp.Error == nil {
-		t.Fatalf("expected error, got nil")
-	}
-	if apiResp.Error.ErrCode != 10131 {
-		t.Fatalf("expected error 10131, got %v", apiResp.Error)
-	}
-}
-
 // https://github.com/nats-io/nats-server/issues/3734
 func TestJetStreamMsgBlkFailOnKernelFault(t *testing.T) {
 	s := RunBasicJetStreamServer(t)
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 89b8f1ff..cebd90c4 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -3925,7 +3925,7 @@ func TestMonitorAccountz(t *testing.T) {
 	body = string(readBody(t, fmt.Sprintf("http://127.0.0.1:%d%s?acc=$SYS", s.MonitorAddr().Port, AccountzPath)))
 	require_Contains(t, body, `"account_detail": {`)
 	require_Contains(t, body, `"account_name": "$SYS",`)
-	require_Contains(t, body, `"subscriptions": 44,`)
+	require_Contains(t, body, `"subscriptions": 46,`)
 	require_Contains(t, body, `"is_system": true,`)
 	require_Contains(t, body, `"system_account": "$SYS"`)
 
@@ -4585,3 +4585,38 @@ func TestServerIDZRequest(t *testing.T) {
 	require_True(t, sid.Name == "TEST22")
 	require_True(t, strings.HasPrefix(sid.ID, "N"))
 }
+
+func TestMonitorProfilez(t *testing.T) {
+	s := RunServer(DefaultOptions())
+	defer s.Shutdown()
+
+	// First of all, check that the profiles aren't accessible
+	// when profiling hasn't been started in the usual way.
+	if ps := s.profilez(&ProfilezOptions{
+		Name: "allocs", Debug: 0,
+	}); ps.Error == "" {
+		t.Fatal("Profile should not be accessible when profiling not started")
+	}
+
+	// Then start profiling.
+	s.StartProfiler()
+
+	// Now check that all of the profiles that we expect are
+	// returning instead of erroring.
+	for _, try := range []*ProfilezOptions{
+		{Name: "allocs", Debug: 0},
+		{Name: "allocs", Debug: 1},
+		{Name: "block", Debug: 0},
+		{Name: "goroutine", Debug: 0},
+		{Name: "goroutine", Debug: 1},
+		{Name: "goroutine", Debug: 2},
+		{Name: "heap", Debug: 0},
+		{Name: "heap", Debug: 1},
+		{Name: "mutex", Debug: 0},
+		{Name: "threadcreate", Debug: 0},
+	} {
+		if ps := s.profilez(try); ps.Error != _EMPTY_ {
+			t.Fatalf("Unexpected error on %v: %s", try, ps.Error)
+		}
+	}
+}