public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phlare: Support both Phlare and Pyroscope backends (#66989)

Browse Source
This commit is contained in:
Andrej Ocenas
2023-04-25 16:08:18 +02:00
committed by GitHub
parent 2f1a08511a
commit 63777ea368
26 changed files with 1249 additions and 496 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tests/api/plugins/api_plugins_test.go
+4
View File
@@ -76,6 +76,10 @@ func TestIntegrationPlugins(t *testing.T) {
})
})
//
// NOTE:
// If this test is failing due to changes in plugins just rerun with updateSnapshotFlag = true at the top.
//
t.Run("List", func(t *testing.T) {
testCases := []testCase{
{
pkg/tests/api/plugins/data/expectedListResp.json
+50 -45
View File
@@ -643,6 +643,47 @@
"signatureType": "",
"signatureOrg": ""
},
{
"name": "Grafana Pyroscope",
"type": "datasource",
"id": "phlare",
"enabled": true,
"pinned": false,
"info": {
"author": {
"name": "Grafana Labs",
"url": "https://www.grafana.com"
},
"description": "Supports Phlare and Pyroscope backends, horizontally-scalable, highly-available, multi-tenant continuous profiling aggregation systems.",
"links": [
{
"name": "GitHub Project",
"url": "https://github.com/grafana/phlare"
}
],
"logos": {
"small": "public/app/plugins/datasource/phlare/img/grafana_pyroscope_icon.svg",
"large": "public/app/plugins/datasource/phlare/img/grafana_pyroscope_icon.svg"
},
"build": {},
"screenshots": null,
"version": "",
"updated": ""
},
"dependencies": {
"grafanaDependency": "",
"grafanaVersion": "*",
"plugins": []
},
"latestVersion": "",
"hasUpdate": false,
"defaultNavUrl": "/plugins/phlare/",
"category": "profiling",
"state": "",
"signature": "internal",
"signatureType": "",
"signatureOrg": ""
},
{
"name": "Graph (old)",
"type": "panel",
@@ -1137,7 +1178,8 @@
"signature": "internal",
"signatureType": "",
"signatureOrg": ""
},{
},
{
"name": "Parca",
"type": "datasource",
"id": "parca",
@@ -1149,10 +1191,12 @@
"url": "https://www.grafana.com"
},
"description": "Continuous profiling for analysis of CPU and memory usage, down to the line number and throughout time. Saving infrastructure cost, improving performance, and increasing reliability.",
"links": [{
"name": "GitHub Project",
"url": "https://github.com/parca-dev/parca"
}],
"links": [
{
"name": "GitHub Project",
"url": "https://github.com/parca-dev/parca"
}
],
"logos": {
"small": "public/app/plugins/datasource/parca/img/logo-small.svg",
"large": "public/app/plugins/datasource/parca/img/logo-small.svg"
@@ -1176,45 +1220,6 @@
"signatureType": "",
"signatureOrg": ""
},
{
"name": "Phlare",
"type": "datasource",
"id": "phlare",
"enabled": true,
"pinned": false,
"info": {
"author": {
"name": "Grafana Labs",
"url": "https://www.grafana.com"
},
"description": "Horizontally-scalable, highly-available, multi-tenant continuous profiling aggregation system. OSS profiling solution from Grafana Labs.",
"links": [{
"name": "GitHub Project",
"url": "https://github.com/grafana/phlare"
}],
"logos": {
"small": "public/app/plugins/datasource/phlare/img/phlare_icon_color.svg",
"large": "public/app/plugins/datasource/phlare/img/phlare_icon_color.svg"
},
"build": {},
"screenshots": null,
"version": "",
"updated": ""
},
"dependencies": {
"grafanaDependency": "",
"grafanaVersion": "*",
"plugins": []
},
"latestVersion": "",
"hasUpdate": false,
"defaultNavUrl": "/plugins/phlare/",
"category": "profiling",
"state": "",
"signature": "internal",
"signatureType": "",
"signatureOrg": ""
},
{
"name": "Pie chart",
"type": "panel",
@@ -1806,4 +1811,4 @@
"signatureType": "",
"signatureOrg": ""
}
]
]
pkg/tsdb/phlare/client.go
+66
View File
@@ -0,0 +1,66 @@
package phlare
import (
"context"
"net/http"
)
type ProfilingClient interface {
ProfileTypes(context.Context) ([]*ProfileType, error)
LabelNames(ctx context.Context, query string, start int64, end int64) ([]string, error)
LabelValues(ctx context.Context, query string, label string, start int64, end int64) ([]string, error)
GetSeries(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64, groupBy []string, step float64) (*SeriesResponse, error)
GetProfile(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64) (*ProfileResponse, error)
}
type ProfileType struct {
ID string `json:"id"`
Label string `json:"label"`
}
func getClient(backendType string, httpClient *http.Client, url string) ProfilingClient {
if backendType == "pyroscope" {
return NewPyroscopeClient(httpClient, url)
}
// We treat unset value as phlare
return NewPhlareClient(httpClient, url)
}
type Flamebearer struct {
Names []string
Levels []*Level
Total int64
MaxSelf int64
}
type Level struct {
Values []int64
}
type Series struct {
Labels []*LabelPair
Points []*Point
}
type LabelPair struct {
Name string
Value string
}
type Point struct {
Value float64
// Milliseconds unix timestamp
Timestamp int64
}
type ProfileResponse struct {
Flamebearer *Flamebearer
Units string
}
type SeriesResponse struct {
Series []*Series
Units string
Label string
}
pkg/tsdb/phlare/instance.go
+131 -78
View File
@@ -3,18 +3,16 @@ package phlare
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"strings"
"strconv"
"time"
"github.com/bufbuild/connect-go"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/infra/httpclient"
querierv1 "github.com/grafana/phlare/api/gen/proto/go/querier/v1"
"github.com/grafana/phlare/api/gen/proto/go/querier/v1/querierv1connect"
typesv1 "github.com/grafana/phlare/api/gen/proto/go/types/v1"
)
var (
@@ -26,7 +24,13 @@ var (
// PhlareDatasource is a datasource for querying application performance profiles.
type PhlareDatasource struct {
client querierv1connect.QuerierServiceClient
httpClient *http.Client
client ProfilingClient
settings backend.DataSourceInstanceSettings
}
type JsonData struct {
BackendType string `json:"backendType"`
}
// NewPhlareDatasource creates a new datasource instance.
@@ -40,93 +44,152 @@ func NewPhlareDatasource(httpClientProvider httpclient.Provider, settings backen
return nil, err
}
var jsonData *JsonData
err = json.Unmarshal(settings.JSONData, &jsonData)
if err != nil {
return nil, err
}
return &PhlareDatasource{
client: querierv1connect.NewQuerierServiceClient(httpClient, settings.URL),
httpClient: httpClient,
client: getClient(jsonData.BackendType, httpClient, settings.URL),
settings: settings,
}, nil
}
func (d *PhlareDatasource) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
logger.Debug("CallResource", "Path", req.Path, "Method", req.Method, "Body", req.Body)
if req.Path == "profileTypes" {
return d.callProfileTypes(ctx, req, sender)
return d.profileTypes(ctx, req, sender)
}
if req.Path == "labelNames" {
return d.callLabelNames(ctx, req, sender)
return d.labelNames(ctx, req, sender)
}
if req.Path == "series" {
return d.callSeries(ctx, req, sender)
if req.Path == "labelValues" {
return d.labelValues(ctx, req, sender)
}
if req.Path == "backendType" {
return d.backendType(ctx, req, sender)
}
return sender.Send(&backend.CallResourceResponse{
Status: 404,
})
}
func (d *PhlareDatasource) callProfileTypes(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
res, err := d.client.ProfileTypes(ctx, connect.NewRequest(&querierv1.ProfileTypesRequest{}))
func (d *PhlareDatasource) profileTypes(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
types, err := d.client.ProfileTypes(ctx)
if err != nil {
return err
}
var data []byte
if res.Msg.ProfileTypes == nil {
// Let's make sure we send at least empty array if we don't have any types
data, err = json.Marshal([]*typesv1.ProfileType{})
bodyData, err := json.Marshal(types)
if err != nil {
return err
}
err = sender.Send(&backend.CallResourceResponse{Body: bodyData, Headers: req.Headers, Status: 200})
if err != nil {
return err
}
return nil
}
func (d *PhlareDatasource) labelNames(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
u, err := url.Parse(req.URL)
if err != nil {
return err
}
query := u.Query()
start, err := strconv.ParseInt(query["start"][0], 10, 64)
if err != nil {
return err
}
end, err := strconv.ParseInt(query["end"][0], 10, 64)
if err != nil {
return err
}
res, err := d.client.LabelNames(ctx, query["query"][0], start, end)
if err != nil {
return fmt.Errorf("error calling LabelNames: %v", err)
}
data, err := json.Marshal(res)
if err != nil {
return err
}
err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200})
if err != nil {
return err
}
return nil
}
type LabelValuesPayload struct {
Query string
Label string
Start int64
End int64
}
func (d *PhlareDatasource) labelValues(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
u, err := url.Parse(req.URL)
if err != nil {
return err
}
query := u.Query()
start, err := strconv.ParseInt(query["start"][0], 10, 64)
if err != nil {
return err
}
end, err := strconv.ParseInt(query["end"][0], 10, 64)
if err != nil {
return err
}
res, err := d.client.LabelValues(ctx, query["query"][0], query["label"][0], start, end)
if err != nil {
return fmt.Errorf("error calling LabelValues: %v", err)
}
data, err := json.Marshal(res)
if err != nil {
return err
}
err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200})
if err != nil {
return err
}
return nil
}
type BackendTypeRespBody struct {
BackendType string `json:"backendType"` // "phlare" or "pyroscope"
}
// backendType is a simplistic test to figure out if we are speaking to phlare or pyroscope backend
func (d *PhlareDatasource) backendType(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
u, err := url.Parse(req.URL)
if err != nil {
return err
}
query := u.Query()
body := &BackendTypeRespBody{BackendType: "unknown"}
pyroClient := getClient("pyroscope", d.httpClient, query["url"][0])
_, err = pyroClient.ProfileTypes(ctx)
if err == nil {
body.BackendType = "pyroscope"
} else {
data, err = json.Marshal(res.Msg.ProfileTypes)
}
if err != nil {
return err
}
err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200})
if err != nil {
return err
}
return nil
}
type SeriesRequestJson struct {
Matchers []string `json:"matchers"`
}
func (d *PhlareDatasource) callSeries(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
parsedUrl, err := url.Parse(req.URL)
if err != nil {
return err
}
matchers, ok := parsedUrl.Query()["matchers"]
if !ok {
matchers = []string{"{}"}
phlareClient := getClient("phlare", d.httpClient, d.settings.URL)
_, err := phlareClient.ProfileTypes(ctx)
if err == nil {
body.BackendType = "phlare"
}
}
res, err := d.client.Series(ctx, connect.NewRequest(&querierv1.SeriesRequest{Matchers: matchers}))
data, err := json.Marshal(body)
if err != nil {
return err
}
for _, val := range res.Msg.LabelsSet {
withoutPrivate := withoutPrivateLabels(val.Labels)
val.Labels = withoutPrivate
}
data, err := json.Marshal(res.Msg.LabelsSet)
if err != nil {
return err
}
err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200})
if err != nil {
return err
}
return nil
}
func (d *PhlareDatasource) callLabelNames(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
res, err := d.client.LabelNames(ctx, connect.NewRequest(&querierv1.LabelNamesRequest{}))
if err != nil {
return err
}
data, err := json.Marshal(res.Msg.Names)
if err != nil {
return err
}
err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200})
if err != nil {
return err
@@ -166,7 +229,7 @@ func (d *PhlareDatasource) CheckHealth(ctx context.Context, _ *backend.CheckHeal
status := backend.HealthStatusOk
message := "Data source is working"
if _, err := d.client.ProfileTypes(ctx, connect.NewRequest(&querierv1.ProfileTypesRequest{})); err != nil {
if _, err := d.client.ProfileTypes(ctx); err != nil {
status = backend.HealthStatusError
message = err.Error()
}
@@ -239,13 +302,3 @@ func (d *PhlareDatasource) PublishStream(_ context.Context, _ *backend.PublishSt
Status: backend.PublishStreamStatusPermissionDenied,
}, nil
}
func withoutPrivateLabels(labels []*typesv1.LabelPair) []*typesv1.LabelPair {
res := make([]*typesv1.LabelPair, 0, len(labels))
for _, l := range labels {
if !strings.HasPrefix(l.Name, "__") {
res = append(res, l)
}
}
return res
}
pkg/tsdb/phlare/instance_test.go
+3 -3
View File
@@ -40,9 +40,9 @@ func Test_CallResource(t *testing.T) {
context.Background(),
&backend.CallResourceRequest{
PluginContext: backend.PluginContext{},
Path: "series",
Path: "profileTypes",
Method: "GET",
URL: "series?matchers=%7B%7D",
URL: "profileTypes",
Headers: nil,
Body: nil,
},
@@ -50,7 +50,7 @@ func Test_CallResource(t *testing.T) {
)
require.NoError(t, err)
require.Equal(t, 200, sender.Resp.Status)
require.Equal(t, `[{"labels":[{"name":"instance","value":"127.0.0.1"},{"name":"job","value":"default"}]}]`, string(sender.Resp.Body))
require.Equal(t, `[{"id":"type:1","label":"cpu"},{"id":"type:2","label":"memory"}]`, string(sender.Resp.Body))
})
}
pkg/tsdb/phlare/kinds/dataquery/types_dataquery_gen.go
+2 -2
View File
@@ -16,8 +16,8 @@ const (
PhlareQueryTypeProfile PhlareQueryType = "profile"
)
// PhlareDataQuery defines model for PhlareDataQuery.
type PhlareDataQuery struct {
// GrafanaPyroscopeDataQuery defines model for GrafanaPyroscopeDataQuery.
type GrafanaPyroscopeDataQuery struct {
// For mixed data sources the selected datasource is on the query level.
// For non mixed scenarios this is undefined.
// TODO find a better way to do this ^ that's friendly to schema
pkg/tsdb/phlare/phlareClient.go
+164
View File
@@ -0,0 +1,164 @@
package phlare
import (
"context"
"fmt"
"net/http"
"strings"
"github.com/bufbuild/connect-go"
querierv1 "github.com/grafana/phlare/api/gen/proto/go/querier/v1"
"github.com/grafana/phlare/api/gen/proto/go/querier/v1/querierv1connect"
)
type PhlareClient struct {
connectClient querierv1connect.QuerierServiceClient
}
func NewPhlareClient(httpClient *http.Client, url string) *PhlareClient {
return &PhlareClient{
connectClient: querierv1connect.NewQuerierServiceClient(httpClient, url),
}
}
func (c *PhlareClient) ProfileTypes(ctx context.Context) ([]*ProfileType, error) {
res, err := c.connectClient.ProfileTypes(ctx, connect.NewRequest(&querierv1.ProfileTypesRequest{}))
if err != nil {
return nil, err
}
if res.Msg.ProfileTypes == nil {
// Let's make sure we send at least empty array if we don't have any types
return []*ProfileType{}, nil
} else {
pTypes := make([]*ProfileType, len(res.Msg.ProfileTypes))
for i, pType := range res.Msg.ProfileTypes {
pTypes[i] = &ProfileType{
ID: pType.ID,
Label: pType.Name + " - " + pType.SampleType,
}
}
return pTypes, nil
}
}
func (c *PhlareClient) GetSeries(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64, groupBy []string, step float64) (*SeriesResponse, error) {
req := connect.NewRequest(&querierv1.SelectSeriesRequest{
ProfileTypeID: profileTypeID,
LabelSelector: labelSelector,
Start: start,
End: end,
Step: step,
GroupBy: groupBy,
})
resp, err := c.connectClient.SelectSeries(ctx, req)
if err != nil {
return nil, err
}
series := make([]*Series, len(resp.Msg.Series))
for i, s := range resp.Msg.Series {
labels := make([]*LabelPair, len(s.Labels))
for i, l := range s.Labels {
labels[i] = &LabelPair{
Name: l.Name,
Value: l.Value,
}
}
points := make([]*Point, len(s.Points))
for i, p := range s.Points {
points[i] = &Point{
Value: p.Value,
Timestamp: p.Timestamp,
}
}
series[i] = &Series{
Labels: labels,
Points: points,
}
}
parts := strings.Split(profileTypeID, ":")
return &SeriesResponse{
Series: series,
Units: getUnits(profileTypeID),
Label: parts[1],
}, nil
}
func (c *PhlareClient) GetProfile(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64) (*ProfileResponse, error) {
req := &connect.Request[querierv1.SelectMergeStacktracesRequest]{
Msg: &querierv1.SelectMergeStacktracesRequest{
ProfileTypeID: profileTypeID,
LabelSelector: labelSelector,
Start: start,
End: end,
},
}
resp, err := c.connectClient.SelectMergeStacktraces(ctx, req)
if err != nil {
return nil, err
}
levels := make([]*Level, len(resp.Msg.Flamegraph.Levels))
for i, level := range resp.Msg.Flamegraph.Levels {
levels[i] = &Level{
Values: level.Values,
}
}
return &ProfileResponse{
Flamebearer: &Flamebearer{
Names: resp.Msg.Flamegraph.Names,
Levels: levels,
Total: resp.Msg.Flamegraph.Total,
MaxSelf: resp.Msg.Flamegraph.MaxSelf,
},
Units: getUnits(profileTypeID),
}, nil
}
func getUnits(profileTypeID string) string {
parts := strings.Split(profileTypeID, ":")
unit := parts[2]
if unit == "nanoseconds" {
return "ns"
}
if unit == "count" {
return "short"
}
return unit
}
func (c *PhlareClient) LabelNames(ctx context.Context, query string, start int64, end int64) ([]string, error) {
resp, err := c.connectClient.LabelNames(ctx, connect.NewRequest(&querierv1.LabelNamesRequest{}))
if err != nil {
return nil, fmt.Errorf("error seding LabelNames request %v", err)
}
var filtered []string
for _, label := range resp.Msg.Names {
if !isPrivateLabel(label) {
filtered = append(filtered, label)
}
}
return filtered, nil
}
func (c *PhlareClient) LabelValues(ctx context.Context, query string, label string, start int64, end int64) ([]string, error) {
resp, err := c.connectClient.LabelValues(ctx, connect.NewRequest(&querierv1.LabelValuesRequest{Name: label}))
if err != nil {
return nil, err
}
return resp.Msg.Names, nil
}
func isPrivateLabel(label string) bool {
return strings.HasPrefix(label, "__")
}
pkg/tsdb/phlare/phlareClient_test.go
+109
View File
@@ -0,0 +1,109 @@
package phlare
import (
"context"
"testing"
"github.com/bufbuild/connect-go"
googlev1 "github.com/grafana/phlare/api/gen/proto/go/google/v1"
querierv1 "github.com/grafana/phlare/api/gen/proto/go/querier/v1"
typesv1 "github.com/grafana/phlare/api/gen/proto/go/types/v1"
"github.com/stretchr/testify/require"
)
func Test_PhlareClient(t *testing.T) {
connectClient := &FakePhlareConnectClient{}
client := &PhlareClient{
connectClient: connectClient,
}
t.Run("GetSeries", func(t *testing.T) {
resp, err := client.GetSeries(context.Background(), "memory:alloc_objects:count:space:bytes", "{}", 0, 100, []string{}, 15)
require.Nil(t, err)
series := &SeriesResponse{
Series: []*Series{
{Labels: []*LabelPair{{Name: "foo", Value: "bar"}}, Points: []*Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
},
Units: "short",
Label: "alloc_objects",
}
require.Equal(t, series, resp)
})
t.Run("GetProfile", func(t *testing.T) {
resp, err := client.GetProfile(context.Background(), "memory:alloc_objects:count:space:bytes", "{}", 0, 100)
require.Nil(t, err)
series := &ProfileResponse{
Flamebearer: &Flamebearer{
Names: []string{"foo", "bar", "baz"},
Levels: []*Level{
{Values: []int64{0, 10, 0, 0}},
{Values: []int64{0, 9, 0, 1}},
{Values: []int64{0, 8, 8, 2}},
},
Total: 100,
MaxSelf: 56,
},
Units: "short",
}
require.Equal(t, series, resp)
})
}
type FakePhlareConnectClient struct {
Req interface{}
}
func (f *FakePhlareConnectClient) ProfileTypes(ctx context.Context, c *connect.Request[querierv1.ProfileTypesRequest]) (*connect.Response[querierv1.ProfileTypesResponse], error) {
panic("implement me")
}
func (f *FakePhlareConnectClient) LabelValues(ctx context.Context, c *connect.Request[querierv1.LabelValuesRequest]) (*connect.Response[querierv1.LabelValuesResponse], error) {
panic("implement me")
}
func (f *FakePhlareConnectClient) LabelNames(context.Context, *connect.Request[querierv1.LabelNamesRequest]) (*connect.Response[querierv1.LabelNamesResponse], error) {
panic("implement me")
}
func (f *FakePhlareConnectClient) Series(ctx context.Context, c *connect.Request[querierv1.SeriesRequest]) (*connect.Response[querierv1.SeriesResponse], error) {
panic("implement me")
}
func (f *FakePhlareConnectClient) SelectMergeStacktraces(ctx context.Context, c *connect.Request[querierv1.SelectMergeStacktracesRequest]) (*connect.Response[querierv1.SelectMergeStacktracesResponse], error) {
f.Req = c
return &connect.Response[querierv1.SelectMergeStacktracesResponse]{
Msg: &querierv1.SelectMergeStacktracesResponse{
Flamegraph: &querierv1.FlameGraph{
Names: []string{"foo", "bar", "baz"},
Levels: []*querierv1.Level{
{Values: []int64{0, 10, 0, 0}},
{Values: []int64{0, 9, 0, 1}},
{Values: []int64{0, 8, 8, 2}},
},
Total: 100,
MaxSelf: 56,
},
},
}, nil
}
func (f *FakePhlareConnectClient) SelectSeries(ctx context.Context, req *connect.Request[querierv1.SelectSeriesRequest]) (*connect.Response[querierv1.SelectSeriesResponse], error) {
f.Req = req
return &connect.Response[querierv1.SelectSeriesResponse]{
Msg: &querierv1.SelectSeriesResponse{
Series: []*typesv1.Series{
{
Labels: []*typesv1.LabelPair{{Name: "foo", Value: "bar"}},
Points: []*typesv1.Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}},
},
},
},
}, nil
}
func (f *FakePhlareConnectClient) SelectMergeProfile(ctx context.Context, c *connect.Request[querierv1.SelectMergeProfileRequest]) (*connect.Response[googlev1.Profile], error) {
panic("implement me")
}
pkg/tsdb/phlare/pyroscopeClient.go
+279
View File
@@ -0,0 +1,279 @@
package phlare
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
)
type PyroscopeClient struct {
httpClient *http.Client
URL string
}
type App struct {
Name string `json:"name"`
}
func NewPyroscopeClient(httpClient *http.Client, url string) *PyroscopeClient {
return &PyroscopeClient{
httpClient: httpClient,
URL: url,
}
}
func (c *PyroscopeClient) ProfileTypes(ctx context.Context) ([]*ProfileType, error) {
resp, err := c.httpClient.Get(c.URL + "/api/apps")
if err != nil {
return nil, err
}
defer func() {
if err := resp.Body.Close(); err != nil {
logger.Error("failed to close response body", "err", err)
}
}()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
var apps []App
err = json.Unmarshal(body, &apps)
if err != nil {
return nil, err
}
var profileTypes []*ProfileType
for _, app := range apps {
profileTypes = append(profileTypes, &ProfileType{
ID: app.Name,
Label: app.Name,
})
}
return profileTypes, nil
}
type PyroscopeProfileResponse struct {
Flamebearer *PyroFlamebearer `json:"flamebearer"`
Metadata *Metadata `json:"metadata"`
Groups map[string]*Group `json:"groups"`
}
type Metadata struct {
Units string `json:"units"`
}
type Group struct {
StartTime int64 `json:"startTime"`
Samples []int64 `json:"samples"`
DurationDelta int64 `json:"durationDelta"`
}
type PyroFlamebearer struct {
Levels [][]int64 `json:"levels"`
MaxSelf int64 `json:"maxSelf"`
NumTicks int64 `json:"numTicks"`
Names []string `json:"names"`
}
func (c *PyroscopeClient) getProfileData(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64, groupBy []string) (*PyroscopeProfileResponse, error) {
params := url.Values{}
params.Add("from", strconv.FormatInt(start, 10))
params.Add("until", strconv.FormatInt(end, 10))
params.Add("query", profileTypeID+labelSelector)
params.Add("format", "json")
if len(groupBy) > 0 {
params.Add("groupBy", groupBy[0])
}
url := c.URL + "/render?" + params.Encode()
logger.Debug("calling /render", "url", url)
resp, err := c.httpClient.Get(url)
if err != nil {
return nil, fmt.Errorf("error calling /render api: %v", err)
}
defer func() {
if err := resp.Body.Close(); err != nil {
logger.Error("failed to close response body", "err", err)
}
}()
var respData *PyroscopeProfileResponse
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("error reading response body: %v", err)
}
err = json.Unmarshal(body, &respData)
if err != nil {
logger.Debug("flamegraph data", "body", string(body))
return nil, fmt.Errorf("error decoding flamegraph data: %v", err)
}
return respData, nil
}
func (c *PyroscopeClient) GetProfile(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64) (*ProfileResponse, error) {
respData, err := c.getProfileData(ctx, profileTypeID, labelSelector, start, end, nil)
if err != nil {
return nil, err
}
mappedLevels := make([]*Level, len(respData.Flamebearer.Levels))
for i, level := range respData.Flamebearer.Levels {
mappedLevels[i] = &Level{
Values: level,
}
}
units := "short"
if respData.Metadata.Units == "bytes" {
units = "bytes"
}
if respData.Metadata.Units == "samples" {
units = "ms"
}
return &ProfileResponse{
Flamebearer: &Flamebearer{
Names: respData.Flamebearer.Names,
Levels: mappedLevels,
Total: respData.Flamebearer.NumTicks,
MaxSelf: respData.Flamebearer.MaxSelf,
},
Units: units,
}, nil
}
func (c *PyroscopeClient) GetSeries(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64, groupBy []string, step float64) (*SeriesResponse, error) {
// This is super ineffective at the moment. We need 2 different APIs one for profile one for series (timeline) data
// but Pyro returns all in single response. This currently does the simplest thing and calls the same API 2 times
// and gets the part of the response it needs.
respData, err := c.getProfileData(ctx, profileTypeID, labelSelector, start, end, groupBy)
if err != nil {
return nil, err
}
stepMillis := int64(step * 1000)
var series []*Series
if len(respData.Groups) == 1 {
series = []*Series{processGroup(respData.Groups["*"], stepMillis, nil)}
} else {
for key, val := range respData.Groups {
// If we have a group by, we don't want the * group
if key != "*" {
label := &LabelPair{
Name: groupBy[0],
Value: key,
}
series = append(series, processGroup(val, stepMillis, label))
}
}
}
return &SeriesResponse{Series: series}, nil
}
// processGroup turns group timeline data into the Series format. Pyro does not seem to have a way to define step, so we
// always get the data in specific step, and we have to aggregate a bit into s step size we need.
func processGroup(group *Group, step int64, label *LabelPair) *Series {
series := &Series{}
if label != nil {
series.Labels = []*LabelPair{label}
}
durationDeltaMillis := group.DurationDelta * 1000
timestamp := group.StartTime * 1000
value := int64(0)
for i, sample := range group.Samples {
pointsLen := int64(len(series.Points))
// Check if the timestamp of the sample is more than next timestamp in the series. If so we create a new point
// with the value we have so far.
if int64(i)*durationDeltaMillis > step*pointsLen+1 {
series.Points = append(series.Points, &Point{
Value: float64(value),
Timestamp: timestamp + step*pointsLen,
})
value = 0
}
value += sample
}
return series
}
func (c *PyroscopeClient) LabelNames(ctx context.Context, query string, start int64, end int64) ([]string, error) {
params := url.Values{}
// Seems like this should be seconds instead of millis for other endpoints
params.Add("from", strconv.FormatInt(start/1000, 10))
params.Add("until", strconv.FormatInt(end/1000, 10))
params.Add("query", query)
resp, err := c.httpClient.Get(c.URL + "/labels?" + params.Encode())
if err != nil {
return nil, err
}
defer func() {
if err := resp.Body.Close(); err != nil {
logger.Error("failed to close response body", "err", err)
}
}()
var names []string
err = json.NewDecoder(resp.Body).Decode(&names)
if err != nil {
return nil, err
}
var filtered []string
for _, label := range names {
// Using the same func from Phlare client, works but should do separate one probably
if !isPrivateLabel(label) {
filtered = append(filtered, label)
}
}
return filtered, nil
}
func (c *PyroscopeClient) LabelValues(ctx context.Context, query string, label string, start int64, end int64) ([]string, error) {
params := url.Values{}
// Seems like this should be seconds instead of millis for other endpoints
params.Add("from", strconv.FormatInt(start/1000, 10))
params.Add("until", strconv.FormatInt(end/1000, 10))
params.Add("label", label)
params.Add("query", query)
resp, err := c.httpClient.Get(c.URL + "/labels?" + params.Encode())
if err != nil {
return nil, err
}
defer func() {
if err := resp.Body.Close(); err != nil {
logger.Error("failed to close response body", "err", err)
}
}()
var values []string
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
err = json.Unmarshal(body, &values)
if err != nil {
logger.Debug("response", "body", string(body))
return nil, fmt.Errorf("error unmarshaling response %v", err)
}
return values, nil
}
pkg/tsdb/phlare/query.go
+32 -63
View File
@@ -5,22 +5,19 @@ import (
"encoding/json"
"fmt"
"math"
"strings"
"time"
"github.com/bufbuild/connect-go"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/backend/gtime"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana-plugin-sdk-go/live"
"github.com/grafana/grafana/pkg/tsdb/phlare/kinds/dataquery"
querierv1 "github.com/grafana/phlare/api/gen/proto/go/querier/v1"
"github.com/xlab/treeprint"
)
type queryModel struct {
WithStreaming bool
dataquery.PhlareDataQuery
dataquery.GrafanaPyroscopeDataQuery
}
type dsJsonModel struct {
@@ -60,36 +57,34 @@ func (d *PhlareDatasource) query(ctx context.Context, pCtx backend.PluginContext
logger.Debug("Failed to parse the MinStep using default", "MinStep", dsJson.MinStep)
}
}
req := connect.NewRequest(&querierv1.SelectSeriesRequest{
ProfileTypeID: qm.ProfileTypeId,
LabelSelector: qm.LabelSelector,
Start: query.TimeRange.From.UnixMilli(),
End: query.TimeRange.To.UnixMilli(),
Step: math.Max(query.Interval.Seconds(), parsedInterval.Seconds()),
GroupBy: qm.GroupBy,
})
logger.Debug("Sending SelectSeriesRequest", "request", req, "queryModel", qm)
seriesResp, err := d.client.SelectSeries(ctx, req)
logger.Debug("Sending SelectSeriesRequest", "queryModel", qm)
seriesResp, err := d.client.GetSeries(
ctx,
qm.ProfileTypeId,
qm.LabelSelector,
query.TimeRange.From.UnixMilli(),
query.TimeRange.To.UnixMilli(),
qm.GroupBy,
math.Max(query.Interval.Seconds(), parsedInterval.Seconds()),
)
if err != nil {
logger.Error("Querying SelectSeries()", "err", err)
response.Error = err
return response
}
// add the frames to the response.
response.Frames = append(response.Frames, seriesToDataFrames(seriesResp, qm.ProfileTypeId)...)
response.Frames = append(response.Frames, seriesToDataFrames(seriesResp)...)
}
if query.QueryType == queryTypeProfile || query.QueryType == queryTypeBoth {
req := makeRequest(qm, query)
logger.Debug("Sending SelectMergeStacktracesRequest", "request", req, "queryModel", qm)
resp, err := d.client.SelectMergeStacktraces(ctx, req)
logger.Debug("Calling GetProfile", "queryModel", qm)
prof, err := d.client.GetProfile(ctx, qm.ProfileTypeId, qm.LabelSelector, query.TimeRange.From.UnixMilli(), query.TimeRange.To.UnixMilli())
if err != nil {
logger.Error("Querying SelectMergeStacktraces()", "err", err)
logger.Error("Error GetProfile()", "err", err)
response.Error = err
return response
}
frame := responseToDataFrames(resp.Msg, qm.ProfileTypeId)
frame := responseToDataFrames(prof)
response.Frames = append(response.Frames, frame)
// If query called with streaming on then return a channel
@@ -108,23 +103,12 @@ func (d *PhlareDatasource) query(ctx context.Context, pCtx backend.PluginContext
return response
}
func makeRequest(qm queryModel, query backend.DataQuery) *connect.Request[querierv1.SelectMergeStacktracesRequest] {
return &connect.Request[querierv1.SelectMergeStacktracesRequest]{
Msg: &querierv1.SelectMergeStacktracesRequest{
ProfileTypeID: qm.ProfileTypeId,
LabelSelector: qm.LabelSelector,
Start: query.TimeRange.From.UnixMilli(),
End: query.TimeRange.To.UnixMilli(),
},
}
}
// responseToDataFrames turns Phlare response to data.Frame. We encode the data into a nested set format where we have
// [level, value, label] columns and by ordering the items in a depth first traversal order we can recreate the whole
// tree back.
func responseToDataFrames(resp *querierv1.SelectMergeStacktracesResponse, profileTypeID string) *data.Frame {
tree := levelsToTree(resp.Flamegraph.Levels, resp.Flamegraph.Names)
return treeToNestedSetDataFrame(tree, profileTypeID)
func responseToDataFrames(resp *ProfileResponse) *data.Frame {
tree := levelsToTree(resp.Flamebearer.Levels, resp.Flamebearer.Names)
return treeToNestedSetDataFrame(tree, resp.Units)
}
// START_OFFSET is offset of the bar relative to previous sibling
@@ -153,7 +137,11 @@ type ProfileTree struct {
// levelsToTree converts flamebearer format into a tree. This is needed to then convert it into nested set format
// dataframe. This should be temporary, and ideally we should get some sort of tree struct directly from Phlare API.
func levelsToTree(levels []*querierv1.Level, names []string) *ProfileTree {
func levelsToTree(levels []*Level, names []string) *ProfileTree {
if len(levels) == 0 {
return nil
}
tree := &ProfileTree{
Start: 0,
Value: levels[0].Values[VALUE_OFFSET],
@@ -278,7 +266,7 @@ type CustomMeta struct {
// where ordering the items in depth first order and knowing the level/depth of each item we can recreate the
// parent - child relationship without explicitly needing parent/child column, and we can later just iterate over the
// dataFrame to again basically walking depth first over the tree/profile.
func treeToNestedSetDataFrame(tree *ProfileTree, profileTypeID string) *data.Frame {
func treeToNestedSetDataFrame(tree *ProfileTree, unit string) *data.Frame {
frame := data.NewFrame("response")
frame.Meta = &data.FrameMeta{PreferredVisualization: "flamegraph"}
@@ -287,9 +275,8 @@ func treeToNestedSetDataFrame(tree *ProfileTree, profileTypeID string) *data.Fra
selfField := data.NewField("self", nil, []int64{})
// profileTypeID should encode the type of the profile with unit being the 3rd part
parts := strings.Split(profileTypeID, ":")
valueField.Config = &data.FieldConfig{Unit: normalizeUnit(parts[2])}
selfField.Config = &data.FieldConfig{Unit: normalizeUnit(parts[2])}
valueField.Config = &data.FieldConfig{Unit: unit}
selfField.Config = &data.FieldConfig{Unit: unit}
frame.Fields = data.Fields{levelField, valueField, selfField}
labelField := NewEnumField("label", nil)
@@ -366,10 +353,10 @@ func walkTree(tree *ProfileTree, fn func(tree *ProfileTree)) {
}
}
func seriesToDataFrames(seriesResp *connect.Response[querierv1.SelectSeriesResponse], profileTypeID string) []*data.Frame {
frames := make([]*data.Frame, 0, len(seriesResp.Msg.Series))
func seriesToDataFrames(resp *SeriesResponse) []*data.Frame {
frames := make([]*data.Frame, 0, len(resp.Series))
for _, series := range seriesResp.Msg.Series {
for _, series := range resp.Series {
// We create separate data frames as the series may not have the same length
frame := data.NewFrame("series")
frame.Meta = &data.FrameMeta{PreferredVisualization: "graph"}
@@ -378,21 +365,13 @@ func seriesToDataFrames(seriesResp *connect.Response[querierv1.SelectSeriesRespo
timeField := data.NewField("time", nil, []time.Time{})
fields = append(fields, timeField)
label := ""
unit := ""
parts := strings.Split(profileTypeID, ":")
if len(parts) == 5 {
label = parts[1] // sample type e.g. cpu, goroutine, alloc_objects
unit = normalizeUnit(parts[2])
}
labels := make(map[string]string)
for _, label := range series.Labels {
labels[label.Name] = label.Value
}
valueField := data.NewField(label, labels, []float64{})
valueField.Config = &data.FieldConfig{Unit: unit}
valueField := data.NewField(resp.Label, labels, []float64{})
valueField.Config = &data.FieldConfig{Unit: resp.Units}
for _, point := range series.Points {
timeField.Append(time.UnixMilli(point.Timestamp))
@@ -405,13 +384,3 @@ func seriesToDataFrames(seriesResp *connect.Response[querierv1.SelectSeriesRespo
}
return frames
}
func normalizeUnit(unit string) string {
if unit == "nanoseconds" {
return "ns"
}
if unit == "count" {
return "short"
}
return unit
}
pkg/tsdb/phlare/query_test.go
+78 -127
View File
@@ -5,14 +5,9 @@ import (
"testing"
"time"
"github.com/bufbuild/connect-go"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/stretchr/testify/require"
googlev1 "github.com/grafana/phlare/api/gen/proto/go/google/v1"
querierv1 "github.com/grafana/phlare/api/gen/proto/go/querier/v1"
typesv1 "github.com/grafana/phlare/api/gen/proto/go/types/v1"
)
// This is where the tests for the datasource backend live.
@@ -60,9 +55,9 @@ func Test_query(t *testing.T) {
dataQuery.QueryType = queryTypeMetrics
resp := ds.query(context.Background(), pCtx, *dataQuery)
require.Nil(t, resp.Error)
r, ok := client.Req.(*connect.Request[querierv1.SelectSeriesRequest])
step, ok := client.Args[5].(float64)
require.True(t, ok)
require.Equal(t, float64(30), r.Msg.Step)
require.Equal(t, float64(30), step)
})
t.Run("query metrics uses default min step", func(t *testing.T) {
@@ -75,9 +70,9 @@ func Test_query(t *testing.T) {
}
resp := ds.query(context.Background(), pCtxNoMinStep, *dataQuery)
require.Nil(t, resp.Error)
r, ok := client.Req.(*connect.Request[querierv1.SelectSeriesRequest])
step, ok := client.Args[5].(float64)
require.True(t, ok)
require.Equal(t, float64(15), r.Msg.Step)
require.Equal(t, float64(15), step)
})
t.Run("query metrics uses group by", func(t *testing.T) {
@@ -86,9 +81,9 @@ func Test_query(t *testing.T) {
dataQuery.JSON = []byte(`{"profileTypeId":"memory:alloc_objects:count:space:bytes","labelSelector":"{app=\\\"baz\\\"}","groupBy":["app","instance"]}`)
resp := ds.query(context.Background(), pCtx, *dataQuery)
require.Nil(t, resp.Error)
r, ok := client.Req.(*connect.Request[querierv1.SelectSeriesRequest])
groupBy, ok := client.Args[4].([]string)
require.True(t, ok)
require.Equal(t, []string{"app", "instance"}, r.Msg.GroupBy)
require.Equal(t, []string{"app", "instance"}, groupBy)
})
}
@@ -116,20 +111,19 @@ func fieldValues[T any](field *data.Field) []T {
// This is where the tests for the datasource backend live.
func Test_profileToDataFrame(t *testing.T) {
resp := &connect.Response[querierv1.SelectMergeStacktracesResponse]{
Msg: &querierv1.SelectMergeStacktracesResponse{
Flamegraph: &querierv1.FlameGraph{
Names: []string{"func1", "func2", "func3"},
Levels: []*querierv1.Level{
{Values: []int64{0, 20, 1, 2}},
{Values: []int64{0, 10, 3, 1, 4, 5, 5, 2}},
},
Total: 987,
MaxSelf: 123,
profile := &ProfileResponse{
Flamebearer: &Flamebearer{
Names: []string{"func1", "func2", "func3"},
Levels: []*Level{
{Values: []int64{0, 20, 1, 2}},
{Values: []int64{0, 10, 3, 1, 4, 5, 5, 2}},
},
Total: 987,
MaxSelf: 123,
},
Units: "short",
}
frame := responseToDataFrames(resp.Msg, "memory:alloc_objects:count:space:bytes")
frame := responseToDataFrames(profile)
require.Equal(t, 4, len(frame.Fields))
require.Equal(t, data.NewField("level", nil, []int64{0, 1, 1}), frame.Fields[0])
require.Equal(t, data.NewField("value", nil, []int64{20, 10, 5}).SetConfig(&data.FieldConfig{Unit: "short"}), frame.Fields[1])
@@ -142,7 +136,7 @@ func Test_profileToDataFrame(t *testing.T) {
// This is where the tests for the datasource backend live.
func Test_levelsToTree(t *testing.T) {
t.Run("simple", func(t *testing.T) {
levels := []*querierv1.Level{
levels := []*Level{
{Values: []int64{0, 100, 0, 0}},
{Values: []int64{0, 40, 0, 1, 0, 30, 0, 2}},
{Values: []int64{0, 15, 0, 3}},
@@ -162,7 +156,7 @@ func Test_levelsToTree(t *testing.T) {
})
t.Run("medium", func(t *testing.T) {
levels := []*querierv1.Level{
levels := []*Level{
{Values: []int64{0, 100, 0, 0}},
{Values: []int64{0, 40, 0, 1, 0, 30, 0, 2, 0, 30, 0, 3}},
{Values: []int64{0, 20, 0, 4, 50, 10, 0, 5}},
@@ -200,7 +194,7 @@ func Test_treeToNestedDataFrame(t *testing.T) {
},
}
frame := treeToNestedSetDataFrame(tree, "memory:alloc_objects:count:space:bytes")
frame := treeToNestedSetDataFrame(tree, "short")
labelConfig := &data.FieldConfig{
TypeConfig: &data.FieldTypeConfig{
@@ -219,7 +213,7 @@ func Test_treeToNestedDataFrame(t *testing.T) {
})
t.Run("nil profile tree", func(t *testing.T) {
frame := treeToNestedSetDataFrame(nil, "memory:alloc_objects:count:space:bytes")
frame := treeToNestedSetDataFrame(nil, "short")
require.Equal(t, 4, len(frame.Fields))
require.Equal(t, 0, frame.Fields[0].Len())
})
@@ -227,40 +221,41 @@ func Test_treeToNestedDataFrame(t *testing.T) {
func Test_seriesToDataFrame(t *testing.T) {
t.Run("single series", func(t *testing.T) {
resp := &connect.Response[querierv1.SelectSeriesResponse]{
Msg: &querierv1.SelectSeriesResponse{
Series: []*typesv1.Series{
{Labels: []*typesv1.LabelPair{}, Points: []*typesv1.Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
},
series := &SeriesResponse{
Series: []*Series{
{Labels: []*LabelPair{}, Points: []*Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
},
Units: "short",
Label: "samples",
}
frames := seriesToDataFrames(resp, "process_cpu:samples:count:cpu:nanoseconds")
frames := seriesToDataFrames(series)
require.Equal(t, 2, len(frames[0].Fields))
require.Equal(t, data.NewField("time", nil, []time.Time{time.UnixMilli(1000), time.UnixMilli(2000)}), frames[0].Fields[0])
require.Equal(t, data.NewField("samples", map[string]string{}, []float64{30, 10}).SetConfig(&data.FieldConfig{Unit: "short"}), frames[0].Fields[1])
// with a label pair, the value field should name itself with a label pair name and not the profile type
resp = &connect.Response[querierv1.SelectSeriesResponse]{
Msg: &querierv1.SelectSeriesResponse{
Series: []*typesv1.Series{
{Labels: []*typesv1.LabelPair{{Name: "app", Value: "bar"}}, Points: []*typesv1.Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
},
series = &SeriesResponse{
Series: []*Series{
{Labels: []*LabelPair{{Name: "app", Value: "bar"}}, Points: []*Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
},
Units: "short",
Label: "samples",
}
frames = seriesToDataFrames(resp, "process_cpu:samples:count:cpu:nanoseconds")
frames = seriesToDataFrames(series)
require.Equal(t, data.NewField("samples", map[string]string{"app": "bar"}, []float64{30, 10}).SetConfig(&data.FieldConfig{Unit: "short"}), frames[0].Fields[1])
})
t.Run("single series", func(t *testing.T) {
resp := &connect.Response[querierv1.SelectSeriesResponse]{
Msg: &querierv1.SelectSeriesResponse{
Series: []*typesv1.Series{
{Labels: []*typesv1.LabelPair{{Name: "foo", Value: "bar"}}, Points: []*typesv1.Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
{Labels: []*typesv1.LabelPair{{Name: "foo", Value: "baz"}}, Points: []*typesv1.Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
},
resp := &SeriesResponse{
Series: []*Series{
{Labels: []*LabelPair{{Name: "foo", Value: "bar"}}, Points: []*Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
{Labels: []*LabelPair{{Name: "foo", Value: "baz"}}, Points: []*Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}}},
},
Units: "short",
Label: "samples",
}
frames := seriesToDataFrames(resp, "process_cpu:samples:count:cpu:nanoseconds")
frames := seriesToDataFrames(resp)
require.Equal(t, 2, len(frames))
require.Equal(t, 2, len(frames[0].Fields))
require.Equal(t, 2, len(frames[1].Fields))
@@ -270,100 +265,56 @@ func Test_seriesToDataFrame(t *testing.T) {
}
type FakeClient struct {
Req interface{}
Args []interface{}
}
func (f *FakeClient) ProfileTypes(ctx context.Context, c *connect.Request[querierv1.ProfileTypesRequest]) (*connect.Response[querierv1.ProfileTypesResponse], error) {
panic("implement me")
}
func (f *FakeClient) LabelValues(ctx context.Context, c *connect.Request[querierv1.LabelValuesRequest]) (*connect.Response[querierv1.LabelValuesResponse], error) {
panic("implement me")
}
func (f *FakeClient) LabelNames(context.Context, *connect.Request[querierv1.LabelNamesRequest]) (*connect.Response[querierv1.LabelNamesResponse], error) {
panic("implement me")
}
func (f *FakeClient) Series(ctx context.Context, c *connect.Request[querierv1.SeriesRequest]) (*connect.Response[querierv1.SeriesResponse], error) {
return &connect.Response[querierv1.SeriesResponse]{
Msg: &querierv1.SeriesResponse{
LabelsSet: []*typesv1.Labels{{
Labels: []*typesv1.LabelPair{
{
Name: "__unit__",
Value: "cpu",
},
{
Name: "instance",
Value: "127.0.0.1",
},
{
Name: "job",
Value: "default",
},
},
}},
func (f *FakeClient) ProfileTypes(ctx context.Context) ([]*ProfileType, error) {
return []*ProfileType{
{
ID: "type:1",
Label: "cpu",
},
{
ID: "type:2",
Label: "memory",
},
}, nil
}
func (f *FakeClient) SelectMergeStacktraces(ctx context.Context, c *connect.Request[querierv1.SelectMergeStacktracesRequest]) (*connect.Response[querierv1.SelectMergeStacktracesResponse], error) {
f.Req = c
return &connect.Response[querierv1.SelectMergeStacktracesResponse]{
Msg: &querierv1.SelectMergeStacktracesResponse{
Flamegraph: &querierv1.FlameGraph{
Names: []string{"foo", "bar", "baz"},
Levels: []*querierv1.Level{
{Values: []int64{0, 10, 0, 0}},
{Values: []int64{0, 9, 0, 1}},
{Values: []int64{0, 8, 8, 2}},
},
Total: 100,
MaxSelf: 56,
func (f *FakeClient) LabelValues(ctx context.Context, query string, label string, start int64, end int64) ([]string, error) {
panic("implement me")
}
func (f *FakeClient) LabelNames(ctx context.Context, query string, start int64, end int64) ([]string, error) {
panic("implement me")
}
func (f *FakeClient) GetProfile(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64) (*ProfileResponse, error) {
return &ProfileResponse{
Flamebearer: &Flamebearer{
Names: []string{"foo", "bar", "baz"},
Levels: []*Level{
{Values: []int64{0, 10, 0, 0}},
{Values: []int64{0, 9, 0, 1}},
{Values: []int64{0, 8, 8, 2}},
},
Total: 100,
MaxSelf: 56,
},
Units: "count",
}, nil
}
func (f *FakeClient) SelectSeries(ctx context.Context, req *connect.Request[querierv1.SelectSeriesRequest]) (*connect.Response[querierv1.SelectSeriesResponse], error) {
f.Req = req
return &connect.Response[querierv1.SelectSeriesResponse]{
Msg: &querierv1.SelectSeriesResponse{
Series: []*typesv1.Series{
{
Labels: []*typesv1.LabelPair{{Name: "foo", Value: "bar"}},
Points: []*typesv1.Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}},
},
},
},
}, nil
}
func (f *FakeClient) SelectMergeProfile(ctx context.Context, c *connect.Request[querierv1.SelectMergeProfileRequest]) (*connect.Response[googlev1.Profile], error) {
f.Req = c
p := &googlev1.Profile{
SampleType: []*googlev1.ValueType{
{Type: 1, Unit: 2},
},
Sample: []*googlev1.Sample{
func (f *FakeClient) GetSeries(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64, groupBy []string, step float64) (*SeriesResponse, error) {
f.Args = []interface{}{profileTypeID, labelSelector, start, end, groupBy, step}
return &SeriesResponse{
Series: []*Series{
{
Value: []int64{1},
LocationId: []uint64{
1, 2,
},
Labels: []*LabelPair{{Name: "foo", Value: "bar"}},
Points: []*Point{{Timestamp: int64(1000), Value: 30}, {Timestamp: int64(2000), Value: 10}},
},
},
Mapping: []*googlev1.Mapping{{Id: 1}},
Location: []*googlev1.Location{
{Id: 1, MappingId: 1, Line: []*googlev1.Line{{FunctionId: 1}}},
{Id: 2, MappingId: 1, Line: []*googlev1.Line{{FunctionId: 2}}},
},
Function: []*googlev1.Function{
{Id: 1, Name: 3},
{Id: 2, Name: 4},
},
StringTable: []string{"", "cpu", "nanoseconds", "foo", "bar"},
}
return connect.NewResponse(p), nil
Units: "count",
Label: "test",
}, nil
}