Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add query subcommand to profilecli for downloading pprof from phlare (g…
…rafana/phlare#475) * Add query subcommand to download pprof from phlare Usage: ``` $ profilecli query merge --from now-5m --output pprof=my.pprof $ profilecli query merge --from now-5m --profile-type memory:alloc_space:bytes:space:bytes &googlev1.Profile{ SampleType: []*googlev1.ValueType{ &googlev1.ValueType{ Type: 1, Unit: 2, }, }, [...] DropFrames: 0, KeepFrames: 0, TimeNanos: 1673453999823000000, DurationNanos: 300000000000, PeriodType: &googlev1.ValueType{ Type: 434, Unit: 2, }, Period: 524288, Comment: []int64(nil), DefaultSampleType: 1, } ``` * Use parsed profile output for console by default Previous output is available with --output=raw * Implement gzip compression for pprof file This will also change the behaviour, when create a pprof file, it will fail when the file already exists.
- Loading branch information
1 parent
45ef1bc
commit 7b1bbe2
Showing
4 changed files
with
206 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,188 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "io" | ||
| "net/http" | ||
| "os" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/bufbuild/connect-go" | ||
| "github.com/go-kit/log/level" | ||
| gprofile "github.com/google/pprof/profile" | ||
| "github.com/grafana/dskit/runutil" | ||
| "github.com/k0kubun/pp/v3" | ||
| "github.com/klauspost/compress/gzip" | ||
| "github.com/mattn/go-isatty" | ||
| "github.com/pkg/errors" | ||
| "github.com/prometheus/common/model" | ||
| "gopkg.in/alecthomas/kingpin.v2" | ||
|
|
||
| querierv1 "github.com/grafana/phlare/api/gen/proto/go/querier/v1" | ||
| "github.com/grafana/phlare/api/gen/proto/go/querier/v1/querierv1connect" | ||
| ) | ||
|
|
||
| const ( | ||
| outputConsole = "console" | ||
| outputRaw = "raw" | ||
| outputPprof = "pprof=" | ||
| ) | ||
|
|
||
| func parseTime(s string) (time.Time, error) { | ||
| if s == "" { | ||
| return time.Time{}, fmt.Errorf("empty time") | ||
| } | ||
| t, err := time.Parse(time.RFC3339, s) | ||
| if err == nil { | ||
| return t, nil | ||
| } | ||
|
|
||
| // try if it is a relative time | ||
| d, rerr := parseRelativeTime(s) | ||
| if rerr == nil { | ||
| return time.Now().Add(-d), nil | ||
| } | ||
|
|
||
| // if not return first error | ||
| return time.Time{}, err | ||
|
|
||
| } | ||
|
|
||
| func parseRelativeTime(s string) (time.Duration, error) { | ||
| s = strings.TrimSpace(s) | ||
| if s == "now" { | ||
| return 0, nil | ||
| } | ||
| s = strings.TrimPrefix(s, "now-") | ||
|
|
||
| d, err := model.ParseDuration(s) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| return time.Duration(d), nil | ||
| } | ||
|
|
||
| type queryParams struct { | ||
| URL string | ||
| From string | ||
| To string | ||
| ProfileType string | ||
| Query string | ||
| } | ||
|
|
||
| func (p *queryParams) parseFromTo() (from time.Time, to time.Time, err error) { | ||
| from, err = parseTime(p.From) | ||
| if err != nil { | ||
| return time.Time{}, time.Time{}, errors.Wrap(err, "failed to parse from") | ||
| } | ||
| to, err = parseTime(p.To) | ||
| if err != nil { | ||
| return time.Time{}, time.Time{}, errors.Wrap(err, "failed to parse to") | ||
| } | ||
|
|
||
| if to.Before(from) { | ||
| return time.Time{}, time.Time{}, errors.Wrap(err, "from cannot be after") | ||
| } | ||
|
|
||
| return from, to, nil | ||
| } | ||
|
|
||
| func (p *queryParams) client() querierv1connect.QuerierServiceClient { | ||
| return querierv1connect.NewQuerierServiceClient( | ||
| http.DefaultClient, | ||
| p.URL, | ||
| ) | ||
| } | ||
|
|
||
| type flagger interface { | ||
| Flag(name, help string) *kingpin.FlagClause | ||
| } | ||
|
|
||
| func addQueryParams(queryCmd flagger) *queryParams { | ||
| params := &queryParams{} | ||
| queryCmd.Flag("url", "URL of the profile store.").Default("http://localhost:4100").StringVar(¶ms.URL) | ||
| queryCmd.Flag("from", "Beginning of the query.").Default("now-1h").StringVar(¶ms.From) | ||
| queryCmd.Flag("to", "End of the query.").Default("now").StringVar(¶ms.To) | ||
| queryCmd.Flag("profile-type", "Profile type to query.").Default("process_cpu:cpu:nanoseconds:cpu:nanoseconds").StringVar(¶ms.ProfileType) | ||
| queryCmd.Flag("query", "Label selector to query.").Default("{}").StringVar(¶ms.Query) | ||
| return params | ||
| } | ||
|
|
||
| func queryMerge(ctx context.Context, params *queryParams, outputFlag string) (err error) { | ||
| from, to, err := params.parseFromTo() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| level.Info(logger).Log("msg", "query aggregated profile from profile store", "url", params.URL, "from", from, "to", to, "query", params.Query, "type", params.ProfileType) | ||
|
|
||
| qc := params.client() | ||
|
|
||
| resp, err := qc.SelectMergeProfile(ctx, connect.NewRequest(&querierv1.SelectMergeProfileRequest{ | ||
| ProfileTypeID: params.ProfileType, | ||
| Start: from.UnixMilli(), | ||
| End: to.UnixMilli(), | ||
| LabelSelector: params.Query, | ||
| })) | ||
|
|
||
| if err != nil { | ||
| return errors.Wrap(err, "failed to query") | ||
| } | ||
|
|
||
| mypp := pp.New() | ||
| mypp.SetColoringEnabled(isatty.IsTerminal(os.Stdout.Fd())) | ||
| mypp.SetExportedOnly(true) | ||
|
|
||
| if outputFlag == outputConsole { | ||
| buf, err := resp.Msg.MarshalVT() | ||
| if err != nil { | ||
| return errors.Wrap(err, "failed to marshal protobuf") | ||
| } | ||
|
|
||
| p, err := gprofile.Parse(bytes.NewReader(buf)) | ||
| if err != nil { | ||
| return errors.Wrap(err, "failed to parse profile") | ||
| } | ||
|
|
||
| fmt.Fprintln(output(ctx), p.String()) | ||
| return nil | ||
|
|
||
| } | ||
|
|
||
| if outputFlag == outputRaw { | ||
| mypp.Print(resp.Msg) | ||
| return nil | ||
| } | ||
|
|
||
| if strings.HasPrefix(outputFlag, outputPprof) { | ||
| filePath := strings.TrimPrefix(outputFlag, outputPprof) | ||
| if filePath == "" { | ||
| return errors.New("no file path specified after pprof=") | ||
| } | ||
| buf, err := resp.Msg.MarshalVT() | ||
| if err != nil { | ||
| return errors.Wrap(err, "failed to marshal protobuf") | ||
| } | ||
|
|
||
| // open new file, fail when the file already exists | ||
| f, err := os.OpenFile(filePath, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0644) | ||
| if err != nil { | ||
| return errors.Wrap(err, "failed to create pprof file") | ||
| } | ||
| defer runutil.CloseWithErrCapture(&err, f, "failed to close pprof file") | ||
|
|
||
| gzipWriter := gzip.NewWriter(f) | ||
| defer runutil.CloseWithErrCapture(&err, gzipWriter, "failed to close pprof gzip writer") | ||
|
|
||
| if _, err := io.Copy(gzipWriter, bytes.NewReader(buf)); err != nil { | ||
| return errors.Wrap(err, "failed to write pprof") | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| return errors.Errorf("unknown output %s", outputFlag) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters