Skip to content

Commit d81cad3

Browse files
ethanyzhangethanyzhang
committed
feat: flux statistics profiler
1 parent a0df39b commit d81cad3

File tree

9 files changed

+160
-6
lines changed

9 files changed

+160
-6
lines changed

‎cmd/influxd/launcher/query_test.go

+123
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"github.com/influxdata/flux/execute/executetest"
2222
"github.com/influxdata/flux/execute/table"
2323
"github.com/influxdata/flux/lang"
24+
"github.com/influxdata/flux/memory"
2425
"github.com/influxdata/flux/runtime"
2526
"github.com/influxdata/flux/values"
2627
"github.com/influxdata/influxdb/v2"
@@ -752,6 +753,128 @@ from(bucket: "%s")
752753
}
753754
}
754755

756+
type FluxStatisticsTestProfiler struct{}
757+
758+
func (s FluxStatisticsTestProfiler) Name() string {
759+
return "TestFluxStatistics"
760+
}
761+
762+
func (s FluxStatisticsTestProfiler) GetResult(q flux.Query, alloc *memory.Allocator) (flux.Table, error) {
763+
groupKey := execute.NewGroupKey(
764+
[]flux.ColMeta{
765+
{
766+
Label: "_measurement",
767+
Type: flux.TString,
768+
},
769+
},
770+
[]values.Value{
771+
values.NewString("profiler/FluxStatistics"),
772+
},
773+
)
774+
b := execute.NewColListTableBuilder(groupKey, alloc)
775+
for _, colName := range []string{"_measurement", "_field", execute.DefaultValueColLabel} {
776+
if _, err := b.AddCol(flux.ColMeta{
777+
Label: colName,
778+
Type: flux.TString,
779+
}); err != nil {
780+
return nil, err
781+
}
782+
}
783+
q.Statistics().Range(func(key string, value string) {
784+
b.AppendString(0, "profiler/FluxStatistics")
785+
b.AppendString(1, key)
786+
b.AppendString(2, key)
787+
})
788+
b.Sort([]string{"_field"}, false)
789+
tbl, err := b.Table()
790+
if err != nil {
791+
return nil, err
792+
}
793+
return tbl, nil
794+
}
795+
796+
func TestFluxProfiler(t *testing.T) {
797+
testcases := []struct {
798+
name string
799+
data []string
800+
query string
801+
want string
802+
}{
803+
{
804+
name: "range last single point start time",
805+
data: []string{
806+
"m,tag=a f=1i 1",
807+
},
808+
query: `
809+
option profiler.enabledProfilers = ["TestFluxStatistics", "TestFluxStatistics", "NonExistent"]
810+
from(bucket: v.bucket)
811+
|> range(start: 1970-01-01T00:00:00.000000001Z, stop: 1970-01-01T01:00:00Z)
812+
|> last()
813+
`,
814+
want: `
815+
#datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,dateTime:RFC3339,long,string,string,string
816+
#group,false,false,true,true,false,false,true,true,true
817+
#default,_result,,,,,,,,
818+
,result,table,_start,_stop,_time,_value,_field,_measurement,tag
819+
,,0,1970-01-01T00:00:00.000000001Z,1970-01-01T01:00:00Z,1970-01-01T00:00:00.000000001Z,1,f,m,a
820+
821+
#datatype,string,long,string,string,string
822+
#group,false,false,true,false,false
823+
#default,_profiler,,,,
824+
,result,table,_measurement,_field,_value
825+
,,0,profiler/FluxStatistics,CompileDuration,CompileDuration
826+
,,0,profiler/FluxStatistics,Concurrency,Concurrency
827+
,,0,profiler/FluxStatistics,ExecuteDuration,ExecuteDuration
828+
,,0,profiler/FluxStatistics,MaxAllocated,MaxAllocated
829+
,,0,profiler/FluxStatistics,PlanDuration,PlanDuration
830+
,,0,profiler/FluxStatistics,QueueDuration,QueueDuration
831+
,,0,profiler/FluxStatistics,RequeueDuration,RequeueDuration
832+
,,0,profiler/FluxStatistics,RuntimeErrors,RuntimeErrors
833+
,,0,profiler/FluxStatistics,TotalAllocated,TotalAllocated
834+
,,0,profiler/FluxStatistics,TotalDuration,TotalDuration
835+
,,0,profiler/FluxStatistics,flux/query-plan,flux/query-plan
836+
,,0,profiler/FluxStatistics,influxdb/scanned-bytes,influxdb/scanned-bytes
837+
,,0,profiler/FluxStatistics,influxdb/scanned-values,influxdb/scanned-values
838+
`,
839+
},
840+
}
841+
execute.RegisterProfilers(&FluxStatisticsTestProfiler{})
842+
for _, tc := range testcases {
843+
tc := tc
844+
t.Run(tc.name, func(t *testing.T) {
845+
l := launcher.RunTestLauncherOrFail(t, ctx, nil)
846+
847+
l.SetupOrFail(t)
848+
defer l.ShutdownOrFail(t, ctx)
849+
850+
l.WritePointsOrFail(t, strings.Join(tc.data, "\n"))
851+
852+
queryStr := "import \"profiler\"\nv = {bucket: " + "\"" + l.Bucket.Name + "\"" + "}\n" + tc.query
853+
req := &query.Request{
854+
Authorization: l.Auth,
855+
OrganizationID: l.Org.ID,
856+
Compiler: lang.FluxCompiler{
857+
Query: queryStr,
858+
},
859+
}
860+
if got, err := l.FluxQueryService().Query(ctx, req); err != nil {
861+
t.Error(err)
862+
} else {
863+
dec := csv.NewMultiResultDecoder(csv.ResultDecoderConfig{})
864+
want, err := dec.Decode(ioutil.NopCloser(strings.NewReader(tc.want)))
865+
if err != nil {
866+
t.Fatal(err)
867+
}
868+
defer want.Release()
869+
870+
if err := executetest.EqualResultIterators(want, got); err != nil {
871+
t.Fatal(err)
872+
}
873+
}
874+
})
875+
}
876+
}
877+
755878
func TestQueryPushDowns(t *testing.T) {
756879
testcases := []struct {
757880
name string

‎go.mod

+1-1
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ require (
4747
github.com/hashicorp/vault/api v1.0.2
4848
github.com/imdario/mergo v0.3.9 // indirect
4949
github.com/influxdata/cron v0.0.0-20191203200038-ded12750aac6
50-
github.com/influxdata/flux v0.81.0
50+
github.com/influxdata/flux v0.81.1-0.20200818120340-b9cbe85476cf
5151
github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69
5252
github.com/influxdata/influxql v0.0.0-20180925231337-1cbfca8e56b6
5353
github.com/influxdata/pkg-config v0.2.3

‎go.sum

+2-2
Original file line numberDiff line numberDiff line change
@@ -350,8 +350,8 @@ github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NH
350350
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
351351
github.com/influxdata/cron v0.0.0-20191203200038-ded12750aac6 h1:OtjKkeWDjUbyMi82C7XXy7Tvm2LXMwiBBXyFIGNPaGA=
352352
github.com/influxdata/cron v0.0.0-20191203200038-ded12750aac6/go.mod h1:XabtPPW2qsCg0tl+kjaPU+cFS+CjQXEXbT1VJvHT4og=
353-
github.com/influxdata/flux v0.81.0 h1:O1ApTwk4DoeWqCRbihYRMsRxulrkStXjh/tnNl07a6w=
354-
github.com/influxdata/flux v0.81.0/go.mod h1:sAAIEgQTlTpsXCUQ49ymoRsKqraPzIb7F3paT72/lE0=
353+
github.com/influxdata/flux v0.81.1-0.20200818120340-b9cbe85476cf h1:Mm8jXY8O3zAp/SuDU5BCteMoj8Sh2zeMFZAF8nqQZ94=
354+
github.com/influxdata/flux v0.81.1-0.20200818120340-b9cbe85476cf/go.mod h1:sAAIEgQTlTpsXCUQ49ymoRsKqraPzIb7F3paT72/lE0=
355355
github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69 h1:WQsmW0fXO4ZE/lFGIE84G6rIV5SJN3P3sjIXAP1a8eU=
356356
github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69/go.mod h1:pwymjR6SrP3gD3pRj9RJwdl1j5s3doEEV8gS4X9qSzA=
357357
github.com/influxdata/influxql v0.0.0-20180925231337-1cbfca8e56b6 h1:CFx+pP90q/qg3spoiZjf8donE4WpAdjeJfPOcoNqkWo=

‎query/bridges.go

+9
Original file line numberDiff line numberDiff line change
@@ -149,6 +149,15 @@ func (b ProxyQueryServiceAsyncBridge) Query(ctx context.Context, w io.Writer, re
149149
if err != nil {
150150
return stats, tracing.LogError(span, err)
151151
}
152+
153+
if results, err := q.ProfilerResults(); err != nil {
154+
return stats, tracing.LogError(span, err)
155+
} else if results != nil {
156+
_, err = encoder.Encode(w, results)
157+
if err != nil {
158+
return stats, tracing.LogError(span, err)
159+
}
160+
}
152161
return stats, nil
153162
}
154163

‎query/control/controller.go

+15
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626

2727
"github.com/influxdata/flux"
2828
"github.com/influxdata/flux/codes"
29+
"github.com/influxdata/flux/execute/table"
2930
"github.com/influxdata/flux/lang"
3031
"github.com/influxdata/flux/memory"
3132
"github.com/influxdata/flux/runtime"
@@ -548,6 +549,20 @@ type Query struct {
548549
alloc *memory.Allocator
549550
}
550551

552+
func (q *Query) ProfilerResults() (flux.ResultIterator, error) {
553+
p := q.program.(*lang.AstProgram)
554+
tables := make([]flux.Table, 0)
555+
for _, profiler := range p.Profilers {
556+
if result, err := profiler.GetResult(q, q.alloc); err != nil {
557+
return nil, err
558+
} else {
559+
tables = append(tables, result)
560+
}
561+
}
562+
res := table.NewProfilerResult(tables...)
563+
return flux.NewSliceResultIterator([]flux.Result{&res}), nil
564+
}
565+
551566
// ID reports an ephemeral unique ID for the query.
552567
func (q *Query) ID() QueryID {
553568
return q.id

‎query/mock/service.go

+4
Original file line numberDiff line numberDiff line change
@@ -122,3 +122,7 @@ func (q *Query) Statistics() flux.Statistics {
122122
Metadata: q.Metadata,
123123
}
124124
}
125+
126+
func (q *Query) ProfilerResults() (flux.ResultIterator, error) {
127+
return nil, nil
128+
}

‎query/promql/internal/promqltests/go.mod

+1-1
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ require (
1010
github.com/go-kit/kit v0.10.0 // indirect
1111
github.com/google/go-cmp v0.5.0
1212
github.com/hashicorp/go-rootcerts v1.0.2 // indirect
13-
github.com/influxdata/flux v0.81.0
13+
github.com/influxdata/flux v0.81.1-0.20200818120340-b9cbe85476cf
1414
github.com/influxdata/influxdb/v2 v2.0.0-00010101000000-000000000000
1515
github.com/influxdata/influxql v1.0.1 // indirect
1616
github.com/influxdata/promql/v2 v2.12.0

‎query/promql/internal/promqltests/go.sum

+2-2
Original file line numberDiff line numberDiff line change
@@ -389,8 +389,8 @@ github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NH
389389
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
390390
github.com/influxdata/cron v0.0.0-20191203200038-ded12750aac6 h1:OtjKkeWDjUbyMi82C7XXy7Tvm2LXMwiBBXyFIGNPaGA=
391391
github.com/influxdata/cron v0.0.0-20191203200038-ded12750aac6/go.mod h1:XabtPPW2qsCg0tl+kjaPU+cFS+CjQXEXbT1VJvHT4og=
392-
github.com/influxdata/flux v0.81.0 h1:O1ApTwk4DoeWqCRbihYRMsRxulrkStXjh/tnNl07a6w=
393-
github.com/influxdata/flux v0.81.0/go.mod h1:sAAIEgQTlTpsXCUQ49ymoRsKqraPzIb7F3paT72/lE0=
392+
github.com/influxdata/flux v0.81.1-0.20200818120340-b9cbe85476cf h1:Mm8jXY8O3zAp/SuDU5BCteMoj8Sh2zeMFZAF8nqQZ94=
393+
github.com/influxdata/flux v0.81.1-0.20200818120340-b9cbe85476cf/go.mod h1:sAAIEgQTlTpsXCUQ49ymoRsKqraPzIb7F3paT72/lE0=
394394
github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69 h1:WQsmW0fXO4ZE/lFGIE84G6rIV5SJN3P3sjIXAP1a8eU=
395395
github.com/influxdata/httprouter v1.3.1-0.20191122104820-ee83e2772f69/go.mod h1:pwymjR6SrP3gD3pRj9RJwdl1j5s3doEEV8gS4X9qSzA=
396396
github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=

‎task/backend/executor/support_test.go

+3
Original file line numberDiff line numberDiff line change
@@ -175,6 +175,9 @@ func (q *fakeQuery) Done() {}
175175
func (q *fakeQuery) Cancel() { close(q.results) }
176176
func (q *fakeQuery) Statistics() flux.Statistics { return flux.Statistics{} }
177177
func (q *fakeQuery) Results() <-chan flux.Result { return q.results }
178+
func (q *fakeQuery) ProfilerResults() (flux.ResultIterator, error) {
179+
return nil, nil
180+
}
178181

179182
func (q *fakeQuery) Err() error {
180183
if q.ctxErr != nil {

0 commit comments

Comments
 (0)