From f1b762f908d8dab75731ca9c41aaa4402527a5d1 Mon Sep 17 00:00:00 2001 From: jakenichols2719 Date: Sat, 3 Dec 2022 14:44:03 -0700 Subject: [PATCH 1/7] MySQL parsing implemented, plus minor fixes in some related files Signed-off-by: jakenichols2719 --- pkg/backends/mysql/errors.go | 35 ++ pkg/backends/mysql/handler_proxy.go | 31 ++ pkg/backends/mysql/handler_proxy_test.go | 66 +++ pkg/backends/mysql/handler_query.go | 61 +++ pkg/backends/mysql/handler_query_test.go | 113 +++++ pkg/backends/mysql/health.go | 38 ++ pkg/backends/mysql/health_test.go | 39 ++ pkg/backends/mysql/lex.go | 15 + pkg/backends/mysql/model/datatypes.go | 57 +++ pkg/backends/mysql/model/model.go | 614 +++++++++++++++++++++++ pkg/backends/mysql/model/model_test.go | 307 ++++++++++++ pkg/backends/mysql/mysql.go | 106 ++++ pkg/backends/mysql/mysql_test.go | 137 +++++ pkg/backends/mysql/parsing.go | 279 ++++++++++ pkg/backends/mysql/parsing_test.go | 50 ++ pkg/backends/mysql/routes.go | 53 ++ pkg/backends/mysql/routes_test.go | 68 +++ pkg/backends/mysql/stubs.go | 43 ++ pkg/backends/mysql/stubs_test.go | 53 ++ pkg/parsing/run_state.go | 3 +- 20 files changed, 2166 insertions(+), 2 deletions(-) create mode 100644 pkg/backends/mysql/errors.go create mode 100644 pkg/backends/mysql/handler_proxy.go create mode 100644 pkg/backends/mysql/handler_proxy_test.go create mode 100644 pkg/backends/mysql/handler_query.go create mode 100644 pkg/backends/mysql/handler_query_test.go create mode 100644 pkg/backends/mysql/health.go create mode 100644 pkg/backends/mysql/health_test.go create mode 100644 pkg/backends/mysql/lex.go create mode 100644 pkg/backends/mysql/model/datatypes.go create mode 100644 pkg/backends/mysql/model/model.go create mode 100644 pkg/backends/mysql/model/model_test.go create mode 100644 pkg/backends/mysql/mysql.go create mode 100644 pkg/backends/mysql/mysql_test.go create mode 100644 pkg/backends/mysql/parsing.go create mode 100644 pkg/backends/mysql/parsing_test.go create mode 100644 pkg/backends/mysql/routes.go create mode 100644 pkg/backends/mysql/routes_test.go create mode 100644 pkg/backends/mysql/stubs.go create mode 100644 pkg/backends/mysql/stubs_test.go diff --git a/pkg/backends/mysql/errors.go b/pkg/backends/mysql/errors.go new file mode 100644 index 000000000..81bdb7202 --- /dev/null +++ b/pkg/backends/mysql/errors.go @@ -0,0 +1,35 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import "errors" + +// ErrLimitUnsupported indicates the input a LIMIT keyword, which is currently unsupported +// in the caching layer +var ErrLimitUnsupported = errors.New("limit queries are not supported") + +// ErrUnsupportedOutputFormat indicates the FORMAT value for the query is not supported +var ErrUnsupportedOutputFormat = errors.New("unsupported output format requested") + +// ErrInvalidWithClause indicates the WITH clause of the query is not properly formatted +var ErrInvalidWithClause = errors.New("invalid WITH expression list") + +// ErrUnsupportedToStartOfFunc indicates the ToStartOf func used in the query is not supported by Trickster +var ErrUnsupportedToStartOfFunc = errors.New("unsupported ToStartOf* func") + +// ErrNotAtPreWhere indicates AtPreWhere was called but the current token is not of type tokenPreWhere +var ErrNotAtPreWhere = errors.New("not at PREWHERE") diff --git a/pkg/backends/mysql/handler_proxy.go b/pkg/backends/mysql/handler_proxy.go new file mode 100644 index 000000000..d149083ee --- /dev/null +++ b/pkg/backends/mysql/handler_proxy.go @@ -0,0 +1,31 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "net/http" + + "github.com/trickstercache/trickster/v2/pkg/proxy/engines" + "github.com/trickstercache/trickster/v2/pkg/proxy/urls" +) + +// ProxyHandler sends a request through the basic reverse proxy to the origin, +// and services non-cacheable calls +func (c *Client) ProxyHandler(w http.ResponseWriter, r *http.Request) { + r.URL = urls.BuildUpstreamURL(r, c.BaseUpstreamURL()) + engines.DoProxy(w, r, true) +} diff --git a/pkg/backends/mysql/handler_proxy_test.go b/pkg/backends/mysql/handler_proxy_test.go new file mode 100644 index 000000000..3288d38c7 --- /dev/null +++ b/pkg/backends/mysql/handler_proxy_test.go @@ -0,0 +1,66 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "io" + "testing" + + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + tu "github.com/trickstercache/trickster/v2/pkg/testutil" +) + +func TestProxyHandler(t *testing.T) { + + backendClient, err := NewClient("test", nil, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + ts, w, r, _, err := tu.NewTestInstance("", + backendClient.DefaultPathConfigs, 200, "test", nil, "mysql", "/health", "debug") + if err != nil { + t.Error(err) + } else { + defer ts.Close() + } + rsc := request.GetResources(r) + backendClient, err = NewClient("test", rsc.BackendOptions, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + client := backendClient.(*Client) + rsc.BackendClient = client + rsc.BackendOptions.HTTPClient = backendClient.HTTPClient() + + client.ProxyHandler(w, r) + resp := w.Result() + + // it should return 200 OK + if resp.StatusCode != 200 { + t.Errorf("expected 200 got %d.", resp.StatusCode) + } + + bodyBytes, err := io.ReadAll(resp.Body) + if err != nil { + t.Error(err) + } + + if string(bodyBytes) != "test" { + t.Errorf("expected 'test' got %s.", bodyBytes) + } + +} diff --git a/pkg/backends/mysql/handler_query.go b/pkg/backends/mysql/handler_query.go new file mode 100644 index 000000000..e514088a3 --- /dev/null +++ b/pkg/backends/mysql/handler_query.go @@ -0,0 +1,61 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "io" + "net/http" + "strings" + + "github.com/trickstercache/trickster/v2/pkg/proxy/engines" + "github.com/trickstercache/trickster/v2/pkg/proxy/handlers" + "github.com/trickstercache/trickster/v2/pkg/proxy/methods" + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + "github.com/trickstercache/trickster/v2/pkg/proxy/urls" +) + +// QueryHandler for MySQL client +func (c *Client) QueryHandler(w http.ResponseWriter, r *http.Request) { + + q := r.URL.Query() + sqlQuery := q.Get("query") + if methods.HasBody(r.Method) { + b, err := io.ReadAll(r.Body) + if err != nil { + handlers.HandleBadRequestResponse(w, r) + return + } + var body []byte + if sqlQuery != "" { + body = make([]byte, 0, len(sqlQuery)+len(b)) + body = append([]byte(sqlQuery), b...) + q.Del("query") + r.URL.RawQuery = q.Encode() + } else { + body = b + } + sqlQuery = string(body) + r = request.SetBody(r, body) + } + sqlQuery = strings.ToLower(sqlQuery) + if (!strings.HasPrefix(sqlQuery, "select ")) && (!(strings.Index(sqlQuery, " select ") > 0)) { + c.ProxyHandler(w, r) + return + } + r.URL = urls.BuildUpstreamURL(r, c.BaseUpstreamURL()) + engines.DeltaProxyCacheRequest(w, r, c.Modeler()) +} diff --git a/pkg/backends/mysql/handler_query_test.go b/pkg/backends/mysql/handler_query_test.go new file mode 100644 index 000000000..715786eca --- /dev/null +++ b/pkg/backends/mysql/handler_query_test.go @@ -0,0 +1,113 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "io" + "net/http" + "net/http/httptest" + "net/url" + "testing" + + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + tu "github.com/trickstercache/trickster/v2/pkg/testutil" +) + +func testRawQuery() string { + return url.Values(map[string][]string{"query": { + `SELECT (intDiv(toUInt32(time_column), 60) * 60) * 1000 AS t, countMerge(some_count) AS cnt, field1, field2 ` + + `FROM testdb.test_table WHERE time_column BETWEEN toDateTime(1516665600) AND toDateTime(1516687200) ` + + `AND date_column >= toDate(1516665600) AND toDate(1516687200) ` + + `AND field1 > 0 AND field2 = 'some_value' GROUP BY t, field1, field2 ORDER BY t, field1 FORMAT JSON`}}). + Encode() +} + +func testNonSelectQuery() string { + return url.Values(map[string][]string{"enable_http_compression": {"1"}}).Encode() + // not a real query, just something to trigger a non-select proxy-only request +} + +func TestQueryHandler(t *testing.T) { + + backendClient, err := NewClient("test", nil, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + ts, w, r, _, err := tu.NewTestInstance("", backendClient.DefaultPathConfigs, + 200, "{}", nil, "mysql", "/?"+testRawQuery(), "debug") + ctx := r.Context() + if err != nil { + t.Error(err) + } else { + defer ts.Close() + } + rsc := request.GetResources(r) + backendClient, err = NewClient("test", rsc.BackendOptions, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + client := backendClient.(*Client) + rsc.BackendClient = client + rsc.BackendOptions.HTTPClient = backendClient.HTTPClient() + + _, ok := client.Configuration().Paths["/"] + if !ok { + t.Errorf("could not find path config named %s", "/") + } + + client.QueryHandler(w, r) + + resp := w.Result() + + // it should return 200 OK + if resp.StatusCode != 200 { + t.Errorf("expected 200 got %d.", resp.StatusCode) + } + + bodyBytes, err := io.ReadAll(resp.Body) + if err != nil { + t.Error(err) + } + + if string(bodyBytes) != "{}" { + t.Errorf("expected '{}' got %s.", bodyBytes) + } + + r, _ = http.NewRequest(http.MethodGet, ts.URL+"/?"+testNonSelectQuery(), nil) + w = httptest.NewRecorder() + + r = r.WithContext(ctx) + + client.QueryHandler(w, r) + + resp = w.Result() + + // it should return 200 OK + if resp.StatusCode != 200 { + t.Errorf("expected 200 got %d.", resp.StatusCode) + } + + bodyBytes, err = io.ReadAll(resp.Body) + if err != nil { + t.Error(err) + } + + if string(bodyBytes) != "{}" { + t.Errorf("expected '{}' got %s.", bodyBytes) + } + +} diff --git a/pkg/backends/mysql/health.go b/pkg/backends/mysql/health.go new file mode 100644 index 000000000..9d966cea5 --- /dev/null +++ b/pkg/backends/mysql/health.go @@ -0,0 +1,38 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "net/url" + + ho "github.com/trickstercache/trickster/v2/pkg/backends/healthcheck/options" +) + +const ( + healthQuery = "SELECT 1 FORMAT JSON" +) + +// DefaultHealthCheckConfig returns the default HealthCheck Config for this backend provider +func (c *Client) DefaultHealthCheckConfig() *ho.Options { + o := ho.New() + u := c.BaseUpstreamURL() + o.Scheme = u.Scheme + o.Host = u.Host + o.Path = u.Path + o.Query = url.Values{"query": {healthQuery}}.Encode() + return o +} diff --git a/pkg/backends/mysql/health_test.go b/pkg/backends/mysql/health_test.go new file mode 100644 index 000000000..fa3662b3d --- /dev/null +++ b/pkg/backends/mysql/health_test.go @@ -0,0 +1,39 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "strings" + "testing" + + bo "github.com/trickstercache/trickster/v2/pkg/backends/options" +) + +func TestDefaultHealthCheckConfig(t *testing.T) { + + c, _ := NewClient("test", bo.New(), nil, nil, nil, nil) + + dho := c.DefaultHealthCheckConfig() + if dho == nil { + t.Error("expected non-nil result") + } + + if !strings.HasPrefix(dho.Query, "query=SELECT") { + t.Error("expected SELECT-based query string") + } + +} diff --git a/pkg/backends/mysql/lex.go b/pkg/backends/mysql/lex.go new file mode 100644 index 000000000..b09f62c98 --- /dev/null +++ b/pkg/backends/mysql/lex.go @@ -0,0 +1,15 @@ +package mysql + +import ( + "github.com/trickstercache/trickster/v2/pkg/parsing/lex" + "github.com/trickstercache/trickster/v2/pkg/parsing/lex/sql" +) + +var lopts = LexerOptions() +var lexer = sql.NewLexer(lopts) + +func LexerOptions() *lex.Options { + return &lex.Options{ + SpacedKeywordHints: sql.SpacedKeywords(), + } +} diff --git a/pkg/backends/mysql/model/datatypes.go b/pkg/backends/mysql/model/datatypes.go new file mode 100644 index 000000000..3cd690733 --- /dev/null +++ b/pkg/backends/mysql/model/datatypes.go @@ -0,0 +1,57 @@ +package model + +import ( + "github.com/trickstercache/trickster/v2/pkg/parsing/token" + "github.com/trickstercache/trickster/v2/pkg/util/strings" +) + +type DataType int + +const ( + INTEGER = DataType(iota) + INT + SMALLINT + TINYINT + MEDIUMINT + BIGINT + + DECIMAL + NUMERIC + + FLOAT + DOUBLE + + BIT + + DATE + DATETIME + TIMESTAMP + TIME + YEAR + + CHAR + VARCHAR + BINARY + VARBINARY + BLOB + TEXT + ENUM + SET +) + +var dts = []string{ + "integer", "smallint", "tinyint", "mediumint", "bigint", + "decimal", "numberic", + "float", "double", + "bit", + "date", "datetime", "timestamp", "time", "year", + "char", "varchar", "binary", "varbinary", "blob", "text", "enum", "set", +} + +func IsDataType(tk *token.Token) (string, bool) { + return tk.Val, tk.Typ == token.Identifier && strings.IndexInSlice(dts, tk.Val) != -1 +} + +func (dt DataType) String() string { + return dts[dt] +} diff --git a/pkg/backends/mysql/model/model.go b/pkg/backends/mysql/model/model.go new file mode 100644 index 000000000..1b09580b2 --- /dev/null +++ b/pkg/backends/mysql/model/model.go @@ -0,0 +1,614 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package model + +import ( + "bufio" + "bytes" + "io" + "net/http" + "sort" + "strconv" + "strings" + "sync" + + "github.com/trickstercache/trickster/v2/pkg/proxy/headers" + "github.com/trickstercache/trickster/v2/pkg/timeseries" + "github.com/trickstercache/trickster/v2/pkg/timeseries/dataset" + "github.com/trickstercache/trickster/v2/pkg/timeseries/sqlparser" +) + +var marshalers = map[byte]dataset.Marshaler{ + 0: marshalTimeseriesJSON, + 1: marshalTimeseriesCSV, + 2: marshalTimeseriesCSVWithNames, + 3: marshalTimeseriesTSV, + 4: marshalTimeseriesTSVWithNames, + 5: marshalTimeseriesTSVWithNamesAndTypes, +} + +type tsvWriter struct { + io.Writer + writeNames bool + writeTypes bool + separator string +} + +type col struct { + name string + val string + quote string +} + +func (c col) String() string { + return ` "` + c.name + `": ` + c.quote + c.val + c.quote +} + +type cols []col + +func (cs cols) String() string { + var sb = strings.Builder{} + sb.WriteString(" {\n") + j := len(cs) - 1 + for i, c := range cs { + sb.WriteString(c.String()) + if i < j { + sb.WriteString(",") + } + sb.WriteString("\n") + } + sb.WriteString(" }") + return sb.String() +} + +// NewModeler returns a collection of modeling functions for clickhouse interoperability +func NewModeler() *timeseries.Modeler { + return ×eries.Modeler{ + WireUnmarshalerReader: UnmarshalTimeseriesReader, + WireMarshaler: MarshalTimeseries, + WireMarshalWriter: MarshalTimeseriesWriter, + WireUnmarshaler: UnmarshalTimeseries, + CacheMarshaler: dataset.MarshalDataSet, + CacheUnmarshaler: dataset.UnmarshalDataSet, + } +} + +func marshalTimeseriesJSON(ds *dataset.DataSet, rlo *timeseries.RequestOptions, + status int, w io.Writer) error { + + type md struct { + name string + typ string + quote bool + } + + trq := ds.TimeRangeQuery + if trq == nil { + return timeseries.ErrNoTimerangeQuery + } + + if rw, ok := w.(http.ResponseWriter); ok { + h := rw.Header() + h.Set(headers.NameContentType, headers.ValueApplicationJSON+"; charset=UTF-8") + h.Set("X-Clickhouse-Format", "JSON") + rw.WriteHeader(status) + } + + w.Write([]byte(`{ + "meta": + [`, + )) + + fieldCount := len(trq.TagFieldDefintions) + len(trq.ValueFieldDefinitions) + mds := make([]md, fieldCount) + + fd := trq.TimestampDefinition + + mds[fd.OutputPosition] = md{ + name: fd.Name, + typ: fd.SDataType, + quote: shouldQuote(fd.SDataType), + } + + for _, fd = range trq.TagFieldDefintions { + if fd.Name == trq.TimestampDefinition.Name { + continue + } + if fd.OutputPosition > fieldCount { + continue + } + mds[fd.OutputPosition] = md{ + name: fd.Name, + typ: fd.SDataType, + quote: shouldQuote(fd.SDataType), + } + + } + for _, fd = range trq.ValueFieldDefinitions { + if fd.OutputPosition > fieldCount { + continue + } + mds[fd.OutputPosition] = md{ + name: fd.Name, + typ: fd.SDataType, + quote: shouldQuote(fd.SDataType), + } + } + l := len(mds) - 1 + for i, m := range mds { + w.Write([]byte(` + { + "name": "` + m.name + `", + "type": "` + m.typ + `" + }`), + ) + if i < l { + w.Write([]byte(",")) + } + } + + w.Write([]byte(` + ], + + "data": + [ +`, + )) + + var j int64 + var ending = "" + for _, s := range ds.Results[0].SeriesList { + for _, p := range s.Points { + c := make(cols, fieldCount) + fd := trq.TimestampDefinition + if fd.OutputPosition > fieldCount { + continue + } + c[fd.OutputPosition] = col{ + name: mds[fd.OutputPosition].name, + val: sqlparser.FormatOutputTime(p.Epoch, byte(fd.DataType)), + } + if mds[fd.OutputPosition].quote { + c[fd.OutputPosition].quote = `"` + } + + var i int + for _, fd = range trq.TagFieldDefintions { + if fd.Name == trq.TimestampDefinition.Name { + continue + } + if fd.OutputPosition > fieldCount { + continue + } + + c[fd.OutputPosition] = col{ + name: mds[fd.OutputPosition].name, + val: s.Header.Tags[fd.Name], + } + if mds[fd.OutputPosition].quote { + c[fd.OutputPosition].quote = `"` + } + } + + for i, fd = range trq.ValueFieldDefinitions { + if fd.Name == trq.TimestampDefinition.Name { + continue + } + if fd.OutputPosition > fieldCount { + continue + } + if i >= len(p.Values) { + continue + } + c[fd.OutputPosition] = col{ + name: mds[fd.OutputPosition].name, + val: p.Values[i].(string), + } + if mds[fd.OutputPosition].quote { + c[fd.OutputPosition].quote = `"` + } + } + + w.Write([]byte(ending)) + w.Write([]byte(c.String())) + j++ + ending = ",\n" + } + } + w.Write([]byte( + ` + ], + + "rows": `)) + w.Write([]byte(strconv.FormatInt(j, 10))) + w.Write([]byte("\n}\n")) + return nil +} + +func shouldQuote(in string) bool { + if in == "String" || in == "UInt64" || in == "FixedString" { + return true + } + return false +} + +func marshalTimeseriesCSV(ds *dataset.DataSet, rlo *timeseries.RequestOptions, + status int, w io.Writer) error { + return marshalTimeseriesXSV(ds, rlo, status, + &tsvWriter{Writer: w, separator: ","}) +} + +func marshalTimeseriesCSVWithNames(ds *dataset.DataSet, rlo *timeseries.RequestOptions, + status int, w io.Writer) error { + return marshalTimeseriesXSV(ds, rlo, status, + &tsvWriter{Writer: w, writeNames: true, separator: ","}) +} + +func marshalTimeseriesXSV(ds *dataset.DataSet, rlo *timeseries.RequestOptions, + status int, tw *tsvWriter) error { + trq := ds.TimeRangeQuery + if trq == nil { + return timeseries.ErrNoTimerangeQuery + } + + if rw, ok := tw.Writer.(http.ResponseWriter); ok { + h := rw.Header() + + if tw.separator == "" { + tw.separator = "," + } + + var ctPart, fmtPart string + switch tw.separator { + case "\t": + ctPart = "tab" + fmtPart = "TSV" + default: + ctPart = "comma" + fmtPart = "CSV" + tw.separator = "," + } + + h.Set(headers.NameContentType, "text/"+ctPart+"-separated-values; charset=UTF-8") + if tw.writeTypes { + h.Set("X-Clickhouse-Format", fmtPart+"WithNamesAndTypes") + } else if tw.writeNames { + h.Set("X-Clickhouse-Format", fmtPart+"WithNames") + } else { + h.Set("X-Clickhouse-Format", fmtPart) + } + rw.WriteHeader(status) + } + + if trq == nil || (len(trq.TagFieldDefintions) == 0 && + len(trq.ValueFieldDefinitions) == 0) { + return timeseries.ErrNoTimerangeQuery + } + + if len(ds.Results) == 0 { + return nil + } + + fieldCount := len(trq.TagFieldDefintions) + len(trq.ValueFieldDefinitions) + if trq.TimestampDefinition.OutputPosition > fieldCount { + return timeseries.ErrTableHeader + } + + lookup := make(map[string]timeseries.FieldDefinition) + for _, fd := range trq.ValueFieldDefinitions { + lookup[fd.Name] = fd + } + + var i int + if tw.writeNames || tw.writeTypes { + rowVals := make([]string, fieldCount) + fd := trq.TimestampDefinition + rowVals[fd.OutputPosition] = fd.Name + for _, fd = range trq.TagFieldDefintions { + if fd.Name == trq.TimestampDefinition.Name { + continue + } + if fd.OutputPosition > fieldCount { + continue + } + rowVals[fd.OutputPosition] = fd.Name + } + for i, fd = range trq.ValueFieldDefinitions { + rowVals[fd.OutputPosition] = fd.Name + } + tw.Write([]byte(strings.Join(rowVals, tw.separator) + "\n")) + } + + if tw.writeTypes { + rowVals := make([]string, fieldCount) + fd := trq.TimestampDefinition + rowVals[fd.OutputPosition] = fd.SDataType + for _, fd = range trq.TagFieldDefintions { + if fd.Name == trq.TimestampDefinition.Name { + continue + } + if fd.OutputPosition > fieldCount { + continue + } + rowVals[fd.OutputPosition] = fd.SDataType + } + for i, fd = range trq.ValueFieldDefinitions { + rowVals[fd.OutputPosition] = fd.SDataType + } + tw.Write([]byte(strings.Join(rowVals, tw.separator) + "\n")) + } + + for _, s := range ds.Results[0].SeriesList { + for _, p := range s.Points { + rowVals := make([]string, fieldCount) + fd := trq.TimestampDefinition + rowVals[fd.OutputPosition] = sqlparser.FormatOutputTime(p.Epoch, byte(fd.DataType)) + for _, fd = range trq.TagFieldDefintions { + if fd.Name == trq.TimestampDefinition.Name { + continue + } + if fd.OutputPosition > fieldCount { + continue + } + rowVals[fd.OutputPosition] = wrapCSVCell(s.Header.Tags[fd.Name], tw.separator) + } + for i, fd = range trq.ValueFieldDefinitions { + if i >= len(p.Values) { + continue + } + rowVals[fd.OutputPosition] = wrapCSVCell(p.Values[i].(string), tw.separator) + } + tw.Write([]byte(strings.Join(rowVals, tw.separator) + "\n")) + } + } + return nil +} + +func wrapCSVCell(in, separator string) string { + if strings.Contains(in, separator) || + (separator == "," && strings.Contains(in, " ")) { + return `"` + in + `"` + } + return in +} + +func marshalTimeseriesTSV(ds *dataset.DataSet, rlo *timeseries.RequestOptions, + status int, w io.Writer) error { + return marshalTimeseriesXSV(ds, rlo, status, + &tsvWriter{Writer: w, separator: "\t"}) +} + +func marshalTimeseriesTSVWithNames(ds *dataset.DataSet, rlo *timeseries.RequestOptions, + status int, w io.Writer) error { + return marshalTimeseriesXSV(ds, rlo, status, + &tsvWriter{Writer: w, writeNames: true, separator: "\t"}) +} + +func marshalTimeseriesTSVWithNamesAndTypes(ds *dataset.DataSet, rlo *timeseries.RequestOptions, + status int, w io.Writer) error { + return marshalTimeseriesXSV(ds, rlo, status, + &tsvWriter{Writer: w, writeNames: true, writeTypes: true, separator: "\t"}) +} + +// MarshalTimeseries converts a Timeseries into a JSON blob +func MarshalTimeseries(ts timeseries.Timeseries, rlo *timeseries.RequestOptions, status int) ([]byte, error) { + buf := bytes.NewBuffer(nil) + err := MarshalTimeseriesWriter(ts, rlo, status, buf) + return buf.Bytes(), err +} + +// MarshalTimeseriesWriter converts a Timeseries into a JSON blob via an io.Writer +func MarshalTimeseriesWriter(ts timeseries.Timeseries, rlo *timeseries.RequestOptions, + status int, w io.Writer) error { + if ts == nil { + return timeseries.ErrUnknownFormat + } + ds, ok := ts.(*dataset.DataSet) + if !ok { + return timeseries.ErrUnknownFormat + } + var of byte + if rlo != nil { + of = rlo.OutputFormat + } + marshaler, ok := marshalers[of] + if !ok { + return timeseries.ErrUnknownFormat + } + return marshaler(ds, rlo, status, w) +} + +// UnmarshalTimeseries converts a TSV blob into a Timeseries +func UnmarshalTimeseries(data []byte, trq *timeseries.TimeRangeQuery) (timeseries.Timeseries, error) { + buf := bytes.NewReader(data) + return UnmarshalTimeseriesReader(buf, trq) +} + +// UnmarshalTimeseriesReader converts a TSV blob into a Timeseries via io.Reader +func UnmarshalTimeseriesReader(reader io.Reader, trq *timeseries.TimeRangeQuery) (timeseries.Timeseries, error) { + + if trq == nil { + return nil, timeseries.ErrNoTimerangeQuery + } + + // ClickHouse doesn't support multi-statements, so there is always exactly 1 result element + r := &dataset.Result{ + SeriesList: make([]*dataset.Series, 0, 8), + } + ds := &dataset.DataSet{ + TimeRangeQuery: trq, + ExtentList: timeseries.ExtentList{trq.Extent}, + Results: []*dataset.Result{r}, + } + sl := dataset.SeriesLookup{} + var s *dataset.Series + var key dataset.SeriesLookupKey + var i, fieldCount, valueCount, tsi int + var ok bool + var err error + + tfi := make(map[int]int) + vfi := make(map[int]int) + + br := bufio.NewReader(reader) + + for { + line, err := br.ReadString('\n') + if len(line) > 0 { + line = line[:len(line)-1] + } + if err != nil && (err != io.EOF || line == "") { + break + } + + parts := strings.Split(line, "\t") + lp := len(parts) + + p := dataset.Point{} + var ps, m int + var b, c byte + sh := dataset.SeriesHeader{ + QueryStatement: trq.Statement, + Tags: make(dataset.Tags), + } + + // ensure the correct number of fields are present on this line + if i != 0 && lp != fieldCount { + if i == 1 { + return nil, timeseries.ErrTableHeader + } + goto nextIteration + } + + // special handling for the 2 header rows (field names and types) + if i < 2 { + if i == 0 { + // line 0 is the Field Names line, so its column count is authoritative + fieldCount = len(parts) + + // iterate the line parts to get value vs tag fields defined + for j, name := range parts { + if name == "" { + goto nextPart + } + if trq.TimestampDefinition.Name == name { + trq.TimestampDefinition.OutputPosition = j + goto nextPart + } + for l := range trq.TagFieldDefintions { + if trq.TagFieldDefintions[l].Name == name { + trq.TagFieldDefintions[l].OutputPosition = j + tfi[j] = l + goto nextPart + } + } + + if trq.ValueFieldDefinitions == nil { + trq.ValueFieldDefinitions = make([]timeseries.FieldDefinition, 0, 8) + } + + vfi[j] = len(trq.ValueFieldDefinitions) + + trq.ValueFieldDefinitions = append(trq.ValueFieldDefinitions, + timeseries.FieldDefinition{ + OutputPosition: j, + Name: name, + }, + ) + valueCount++ + goto nextPart + + nextPart: + } + goto nextIteration + } + // line 1 is the Field Types line + for j, dt := range parts { + if tsi == j { + trq.TimestampDefinition.SDataType = dt + continue + } + if l, ok := vfi[j]; ok { + trq.ValueFieldDefinitions[l].SDataType = dt + continue + } + if l, ok := tfi[j]; ok { + trq.TagFieldDefintions[l].SDataType = dt + } + } + goto nextIteration + } + + // it's a data row. this sort the fields into tags and values, + m = 0 + p.Values = make([]interface{}, valueCount) + for j, val := range parts { + if tsi == j { + p.Epoch, c, _ = sqlparser.ParseEpoch(val) + if c > 0 && b == 0 { + b = c + } + continue + } + if l, ok := tfi[j]; ok { + sh.Tags[trq.TagFieldDefintions[l].Name] = val + continue + } + p.Values[m] = val + ps += len(val) + m++ + } + if trq.TimestampDefinition.DataType == 0 && b > 0 { + trq.TimestampDefinition.DataType = timeseries.FieldDataType(b) + } + + // this determines if the series is already defined and creates if it is not + sh.CalculateSize() + key.Hash = sh.CalculateHash() + if s, ok = sl[key]; !ok { + s = &dataset.Series{ + Header: sh, + Points: make(dataset.Points, 0, 512), + PointSize: int64(ps), + } + sl[key] = s + r.SeriesList = append(r.SeriesList, s) + } + s.Points = append(s.Points, p) + + nextIteration: + if err == io.EOF { + break + } + i++ + } + if err != nil && err != io.EOF { + return nil, err + } + + var wg sync.WaitGroup + for _, s = range r.SeriesList { + wg.Add(1) + go func(gs *dataset.Series) { + sort.Sort(gs.Points) + wg.Done() + }(s) + } + wg.Wait() + + return ds, nil +} diff --git a/pkg/backends/mysql/model/model_test.go b/pkg/backends/mysql/model/model_test.go new file mode 100644 index 000000000..eeaa4a6a2 --- /dev/null +++ b/pkg/backends/mysql/model/model_test.go @@ -0,0 +1,307 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package model + +import ( + "io" + "net/http/httptest" + "testing" + "time" + + "github.com/trickstercache/trickster/v2/pkg/timeseries" + "github.com/trickstercache/trickster/v2/pkg/timeseries/dataset" +) + +func TestTestColString(t *testing.T) { + c := col{ + name: "test", + val: "yes", + quote: "`", + } + expected := " \"test\": `yes`" + got := c.String() + + if expected != got { + t.Errorf("expected: %s\ngot: %s", expected, got) + } +} + +func TestTestColsString(t *testing.T) { + c := cols{ + col{ + name: "test", + val: "yes", + quote: "`", + }, + col{ + name: "test2", + val: "no", + quote: "|", + }, + } + + expected := " {\n \"test\": `yes`,\n \"test2\": |no|\n }" + got := c.String() + + if expected != got { + t.Errorf("expected: %s\ngot: %s", expected, got) + } +} + +func TestNewModeler(t *testing.T) { + m := NewModeler() + if m == nil || m.CacheMarshaler == nil { + t.Error("failed to get valid modeler") + } +} + +const testStatement = `SELECT (intDiv(toUInt32(dt), 60) * 60) * 1000 AS t, ` + + `hostname, avg(query) avg_query, avg(global_thread) avg_global_thread ` + + `FROM trickster.metrics_history WHERE <$RANGE$> ` + + `GROUP BY t, hostname ORDER BY t, hostname FORMAT <$FORMAT$>` + +const testDataTSV = `1577836800000 localhost 1 54 +1577836860000 localhost 1 27 +1577836920000 localhost 1 39 +` + +const testDataTSVWithNames = `t hostname avg_query avg_global_thread +1577836800000 localhost 1 54 +1577836860000 localhost 1 27 +1577836920000 localhost 1 39 +` + +const testDataTSVWithNamesAndTypes = `t hostname avg_query avg_global_thread +UInt64 String Float64 Float64 +1577836800000 localhost 1 54 +1577836860000 localhost 1 27 +1577836920000 localhost 1 39 +` + +const testDataCSV = `1577836800000,localhost,1,54 +1577836860000,localhost,1,27 +1577836920000,localhost,1,39 +` + +const testDataCSVWithNames = `t,hostname,avg_query,avg_global_thread +1577836800000,localhost,1,54 +1577836860000,localhost,1,27 +1577836920000,localhost,1,39 +` + +const testDataJSON = `{ + "meta": + [ + { + "name": "t", + "type": "UInt64" + }, + { + "name": "hostname", + "type": "String" + }, + { + "name": "avg_query", + "type": "Float64" + }, + { + "name": "avg_global_thread", + "type": "Float64" + } + ], + + "data": + [ + { + "t": "1577836800000", + "hostname": "localhost", + "avg_query": 1, + "avg_global_thread": 54 + }, + { + "t": "1577836860000", + "hostname": "localhost", + "avg_query": 1, + "avg_global_thread": 27 + }, + { + "t": "1577836920000", + "hostname": "localhost", + "avg_query": 1, + "avg_global_thread": 39 + } + ], + + "rows": 3 +} +` + +var testTRQ = ×eries.TimeRangeQuery{ + Statement: testStatement, + Extent: timeseries.Extent{ + Start: time.Unix(1577836800, 0), + End: time.Unix(1577836920, 0), + }, + Step: time.Second * 60, + StepNS: (time.Second * 60).Nanoseconds(), + TimestampDefinition: timeseries.FieldDefinition{ + Name: "t", + DataType: 1, + ProviderData1: 1, + SDataType: "UInt64", + }, + TagFieldDefintions: []timeseries.FieldDefinition{ + { + Name: "t", + }, + { + Name: "hostname", + OutputPosition: 1, + SDataType: "String", + }, + }, + ValueFieldDefinitions: []timeseries.FieldDefinition{ + { + Name: "avg_query", + OutputPosition: 2, + SDataType: "Float64", + }, + { + Name: "avg_global_thread", + OutputPosition: 3, + SDataType: "Float64", + }, + }, +} + +var testDataset = &dataset.DataSet{ + TimeRangeQuery: testTRQ, + ExtentList: timeseries.ExtentList{testTRQ.Extent}, + Results: []*dataset.Result{ + { + SeriesList: []*dataset.Series{ + { + Header: dataset.SeriesHeader{ + QueryStatement: testTRQ.Statement, + Tags: dataset.Tags{ + "hostname": "localhost", + }, + }, + Points: []dataset.Point{ + { + Epoch: 1577836800000000000, + Values: []interface{}{"1", "54"}, + }, + { + Epoch: 1577836860000000000, + Values: []interface{}{"1", "27"}, + }, + { + Epoch: 1577836920000000000, + Values: []interface{}{"1", "39"}, + }, + }, + }, + }, + }, + }, +} + +func TestMarshalJSON(t *testing.T) { + w := httptest.NewRecorder() + marshalTimeseriesJSON(testDataset, ×eries.RequestOptions{}, 200, w) + b, _ := io.ReadAll(w.Result().Body) + if string(b) != testDataJSON { + t.Error() + } +} + +func TestMarshalCSV(t *testing.T) { + w := httptest.NewRecorder() + marshalTimeseriesCSV(testDataset, ×eries.RequestOptions{OutputFormat: 1}, 200, w) + b, _ := io.ReadAll(w.Result().Body) + if string(b) != testDataCSV { + t.Error() + } +} + +func TestMarshalCSVWithNames(t *testing.T) { + w := httptest.NewRecorder() + marshalTimeseriesCSVWithNames(testDataset, ×eries.RequestOptions{OutputFormat: 2}, 200, w) + b, _ := io.ReadAll(w.Result().Body) + if string(b) != testDataCSVWithNames { + t.Error() + } +} + +func TestMarshalTSV(t *testing.T) { + w := httptest.NewRecorder() + marshalTimeseriesTSV(testDataset, ×eries.RequestOptions{OutputFormat: 3}, 200, w) + b, _ := io.ReadAll(w.Result().Body) + if string(b) != testDataTSV { + t.Error() + } +} + +func TestMarshalTSVWithNames(t *testing.T) { + w := httptest.NewRecorder() + marshalTimeseriesTSVWithNames(testDataset, ×eries.RequestOptions{OutputFormat: 4}, 200, w) + b, _ := io.ReadAll(w.Result().Body) + if string(b) != testDataTSVWithNames { + t.Error() + } +} + +func TestMarshalTSVWithNamesAndTypes(t *testing.T) { + w := httptest.NewRecorder() + marshalTimeseriesTSVWithNamesAndTypes(testDataset, ×eries.RequestOptions{OutputFormat: 5}, 200, w) + b, _ := io.ReadAll(w.Result().Body) + if string(b) != testDataTSVWithNamesAndTypes { + t.Error() + } +} + +func TestUnmarshalTimeseries(t *testing.T) { + + ts, err := UnmarshalTimeseries([]byte(testDataTSVWithNamesAndTypes), testTRQ.Clone()) + if err != nil { + t.Error(err) + } + + ds, ok := ts.(*dataset.DataSet) + if !ok || ds == nil { + t.Error("expected non-nil dataset") + return + } + + if len(ds.ExtentList) != 1 || !ds.ExtentList[0].Start.Equal(testDataset.ExtentList[0].Start) || + !ds.ExtentList[0].End.Equal(testDataset.ExtentList[0].End) { + t.Error("unexpected extents: ", ds.ExtentList) + } + +} + +func TestMarshalTimeseries(t *testing.T) { + b, err := MarshalTimeseries(testDataset, ×eries.RequestOptions{OutputFormat: 5}, 200) + if err != nil { + t.Error(err) + } + if string(b) != testDataTSVWithNamesAndTypes { + t.Errorf("unexpected output:\n%s", string(b)) + } + +} diff --git a/pkg/backends/mysql/mysql.go b/pkg/backends/mysql/mysql.go new file mode 100644 index 000000000..cf72255a2 --- /dev/null +++ b/pkg/backends/mysql/mysql.go @@ -0,0 +1,106 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// package mysql provides the ClickHouse backend provider +package mysql + +import ( + "net/http" + "net/url" + "time" + + "github.com/trickstercache/trickster/v2/pkg/backends" + modelch "github.com/trickstercache/trickster/v2/pkg/backends/clickhouse/model" + bo "github.com/trickstercache/trickster/v2/pkg/backends/options" + "github.com/trickstercache/trickster/v2/pkg/backends/providers/registration/types" + "github.com/trickstercache/trickster/v2/pkg/cache" + "github.com/trickstercache/trickster/v2/pkg/proxy/errors" + "github.com/trickstercache/trickster/v2/pkg/proxy/methods" + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + "github.com/trickstercache/trickster/v2/pkg/proxy/urls" + "github.com/trickstercache/trickster/v2/pkg/timeseries" +) + +var _ backends.TimeseriesBackend = (*Client)(nil) + +// Client Implements the Proxy Client Interface +type Client struct { + backends.TimeseriesBackend +} + +var _ types.NewBackendClientFunc = NewClient + +// NewClient returns a new Client Instance +func NewClient(name string, o *bo.Options, router http.Handler, + cache cache.Cache, _ backends.Backends, + _ types.Lookup) (backends.Backend, error) { + if o != nil { + o.FastForwardDisable = true + } + c := &Client{} + b, err := backends.NewTimeseriesBackend(name, o, c.RegisterHandlers, router, cache, modelch.NewModeler()) + c.TimeseriesBackend = b + return c, err +} + +// ParseTimeRangeQuery parses the key parts of a TimeRangeQuery from the inbound HTTP Request +func (c *Client) ParseTimeRangeQuery(r *http.Request) (*timeseries.TimeRangeQuery, *timeseries.RequestOptions, bool, error) { + + var sqlQuery string + var qi url.Values + isBody := methods.HasBody(r.Method) + if isBody { + sqlQuery = string(request.GetBody(r)) + } else { + qi = r.URL.Query() + if p, ok := qi["query"]; ok { + sqlQuery = p[0] + } else { + return nil, nil, false, errors.MissingURLParam("query") + } + } + + trq, ro, canOPC, err := parse(sqlQuery) + if err != nil { + return nil, nil, canOPC, err + } + + var bf time.Duration + res := request.GetResources(r) + if res == nil { + // 60-second default backfill tolerance for ClickHouse + bf = 60 * time.Second + } else { + bf = res.BackendOptions.BackfillTolerance + } + + if trq.BackfillTolerance == 0 { + trq.BackfillTolerance = bf + } + trq.BackfillToleranceNS = bf.Nanoseconds() + + trq.TemplateURL = urls.Clone(r.URL) + + if isBody { + r = request.SetBody(r, []byte(trq.Statement)) + } else { + // Swap in the Tokenized Query in the Url Params + qi.Set("query", trq.Statement) + trq.TemplateURL.RawQuery = qi.Encode() + } + + return trq, ro, canOPC, nil +} diff --git a/pkg/backends/mysql/mysql_test.go b/pkg/backends/mysql/mysql_test.go new file mode 100644 index 000000000..8ebfaeaf6 --- /dev/null +++ b/pkg/backends/mysql/mysql_test.go @@ -0,0 +1,137 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "net/http" + "net/url" + "testing" + + "github.com/trickstercache/trickster/v2/cmd/trickster/config" + "github.com/trickstercache/trickster/v2/pkg/backends" + "github.com/trickstercache/trickster/v2/pkg/backends/clickhouse/model" + bo "github.com/trickstercache/trickster/v2/pkg/backends/options" + cr "github.com/trickstercache/trickster/v2/pkg/cache/registration" + tl "github.com/trickstercache/trickster/v2/pkg/observability/logging" +) + +var testModeler = model.NewModeler() + +func TestClickhouseClientInterfacing(t *testing.T) { + + // this test ensures the client will properly conform to the + // Client and TimeseriesBackend interfaces + + c, err := backends.NewTimeseriesBackend("test", nil, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + + var oc backends.Backend = c + var tc backends.TimeseriesBackend = c + + if oc.Name() != "test" { + t.Errorf("expected %s got %s", "test", oc.Name()) + } + + if tc.Name() != "test" { + t.Errorf("expected %s got %s", "test", tc.Name()) + } +} + +func TestNewClient(t *testing.T) { + + conf, _, err := config.Load("trickster", "test", []string{"-provider", "clickhouse", "-origin-url", "http://1"}) + if err != nil { + t.Fatalf("Could not load configuration: %s", err.Error()) + } + + caches := cr.LoadCachesFromConfig(conf, tl.ConsoleLogger("error")) + defer cr.CloseCaches(caches) + cache, ok := caches["default"] + if !ok { + t.Errorf("Could not find default configuration") + } + + o := &bo.Options{Provider: "TEST_CLIENT"} + c, err := NewClient("default", o, nil, cache, nil, nil) + if err != nil { + t.Error(err) + } + + if c.Name() != "default" { + t.Errorf("expected %s got %s", "default", c.Name()) + } + + if c.Cache().Configuration().Provider != "memory" { + t.Errorf("expected %s got %s", "memory", c.Cache().Configuration().Provider) + } + + if c.Configuration().Provider != "TEST_CLIENT" { + t.Errorf("expected %s got %s", "TEST_CLIENT", c.Configuration().Provider) + } +} + +func TestParseTimeRangeQuery(t *testing.T) { + req := &http.Request{URL: &url.URL{ + Scheme: "https", + Host: "blah.com", + Path: "/", + RawQuery: testRawQuery(), + }, + Header: http.Header{}, + } + client := &Client{} + res, _, _, err := client.ParseTimeRangeQuery(req) + if err != nil { + t.Error(err) + } else { + if res.Step.Seconds() != 60 { + t.Errorf("expected 60 got %f", res.Step.Seconds()) + } + if res.Extent.End.Sub(res.Extent.Start).Hours() != 6 { + t.Errorf("expected 6 got %f", res.Extent.End.Sub(res.Extent.Start).Hours()) + } + } + + req.URL.RawQuery = "" + _, _, _, err = client.ParseTimeRangeQuery(req) + if err == nil { + t.Errorf("expected error for: %s", "missing URL parameter: [query]") + } + + req.URL.RawQuery = url.Values(map[string][]string{"query": { + `SELECT (intDiv(toUInt32(abc), 6z0) * 6z0) * 1000 AS t, countMerge(some_count) AS cnt, field1, field2 ` + + `FROM testdb.test_table WHERE abc BETWEEN toDateTime(1516665600) AND toDateTime(1516687200) ` + + `AND date_column >= toDate(1516665600) AND toDate(1516687200) ` + + `AND field1 > 0 AND field2 = 'some_value' GROUP BY t, field1, field2 ORDER BY t, field1 FORMAT JSON`}}).Encode() + _, _, _, err = client.ParseTimeRangeQuery(req) + if err == nil { + t.Errorf("expected error for: %s", "not a time range query") + } + + req.URL.RawQuery = url.Values(map[string][]string{"query": { + `SELECT (intDiv(toUInt32(0^^^), 60) * 60) * 1000 AS t, countMerge(some_count) AS cnt, field1, field2 ` + + `FROM testdb.test_table WHERE 0^^^ BETWEEN toDateTime(1516665600) AND toDateTime(1516687200) ` + + `AND date_column >= toDate(1516665600) AND toDate(1516687200) ` + + `AND field1 > 0 AND field2 = 'some_value' GROUP BY t, field1, field2 ORDER BY t, field1 FORMAT JSON`}}).Encode() + _, _, _, err = client.ParseTimeRangeQuery(req) + if err == nil { + t.Errorf("expected error for: %s", "not a time range query") + } + +} diff --git a/pkg/backends/mysql/parsing.go b/pkg/backends/mysql/parsing.go new file mode 100644 index 000000000..97eb7ce91 --- /dev/null +++ b/pkg/backends/mysql/parsing.go @@ -0,0 +1,279 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "time" + + "github.com/trickstercache/trickster/v2/pkg/backends/mysql/model" + "github.com/trickstercache/trickster/v2/pkg/parsing" + lsql "github.com/trickstercache/trickster/v2/pkg/parsing/lex/sql" + "github.com/trickstercache/trickster/v2/pkg/parsing/sql" + "github.com/trickstercache/trickster/v2/pkg/parsing/token" + "github.com/trickstercache/trickster/v2/pkg/timeseries" + "github.com/trickstercache/trickster/v2/pkg/timeseries/sqlparser" +) + +var parser = sqlparser.New(parsing.New(nil, lexer, lopts)) + +// parse parses the Time Range Query +func parse(statement string) (*timeseries.TimeRangeQuery, *timeseries.RequestOptions, bool, error) { + trq := ×eries.TimeRangeQuery{Statement: statement} + ro := ×eries.RequestOptions{} + rs, err := parser.Run(sqlparser.NewRunContext(trq, ro), parser, trq.Statement) + + results := rs.Results() + verb, ok := rs.GetResultsCollection("verb") + var canObjectCache bool // indicates the query, while not a time series can be cached by the Object Proxy Cache + if !ok { + return nil, nil, false, sqlparser.ErrNotTimeRangeQuery + } + if vs, ok := verb.(string); ok { + canObjectCache = vs == lsql.TokenValSelect + } + if err != nil { + return nil, nil, canObjectCache, parsing.ParserError(err, rs.Current()) + } + if sql.HasLimitClause(results) { + return nil, nil, canObjectCache, ErrLimitUnsupported + } + err = parseTSColumn(rs, trq, ro) + if err != nil { + return nil, nil, false, err + } + err = parseFromTokens(rs, trq, ro) + if err != nil { + return nil, nil, false, err + } + err, canCache := parseWhereTokens(rs, trq, ro) + // We only want to allow object caching if something is a select statement and otherwise + // follows caching rules + canObjectCache = canObjectCache && canCache + if err != nil { + return nil, nil, false, err + } + err = parseGroupByTokens(rs, trq, ro) + if err != nil { + return nil, nil, false, err + } + //var t *token.Token + /* + if t, err = parseGroupByTokens(results, trq, ro); err != nil { + return nil, nil, canObjectCache, parsing.ParserError(err, t) + } + if t, err = parseSelectTokens(results, trq, ro); err != nil { + return nil, nil, canObjectCache, parsing.ParserError(err, t) + } + if t, err = parseWhereTokens(results, trq, ro); err != nil { + return nil, nil, canObjectCache, parsing.ParserError(err, t) + } + */ + return trq, ro, canObjectCache, nil +} + +// Parse the timeseries column out of the runstate. +// Returns an error if unable to find a timeseries column; this indicates a poorly constructed query. +func parseTSColumn(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *timeseries.RequestOptions) error { + selectTokens, ok := rs.GetResultsCollection("selectTokens") + if !ok { + return sqlparser.ErrMissingTimeseries + } + tokens, ok := selectTokens.([]token.Tokens) + if !ok { + return sqlparser.ErrMissingTimeseries + } + // Parse the first "statement" in the select portion. Need to check for datatype, column, alias + stmnt := tokens[0] + var col, alias string + var foundCol, foundAlias, useAlias bool + for i := 0; i < len(stmnt); i++ { + t := stmnt[i] + if t.Typ == token.Typ(66111) { + useAlias = true + continue + } + if t.Typ != token.Identifier { + continue + } + if _, isDT := model.IsDataType(t); isDT { + continue + } + if !foundCol { + col = t.Val + foundCol = true + } + if foundCol && useAlias { + alias = t.Val + foundAlias = true + } + } + // Set in trq/ro + if foundCol { + trq.TimestampDefinition.Name = col + } + if foundAlias { + ro.BaseTimestampFieldName = alias + } else { + ro.BaseTimestampFieldName = col + } + return nil +} + +// Parse tokens related to the "from" part of the query. +// This should really just not fail. +func parseFromTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *timeseries.RequestOptions) error { + fromTokens, ok := rs.GetResultsCollection("fromTokens") + if !ok { + return sqlparser.ErrMissingTimeseries + } + _, ok = fromTokens.([]token.Tokens) + if !ok { + return sqlparser.ErrMissingTimeseries + } + return nil +} + +// Parse tokens related to the "where" portion of the query. +// Queries MUST include a start clause with the tscol, and may OPTIONALLY include an end clause. Function +// returns an error where a timeseries range query could not be parsed from the WHERE portion. Function also +// returns a boolean value representing if the WHERE portion indicates a cacheable query, which is false if +// the query could not be parsed properly OR if there's some WHERE statement including the tscol that is unrelated +// to the timerange being requested. +func parseWhereTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *timeseries.RequestOptions) (error, bool) { + whereTokens, ok := rs.GetResultsCollection("whereTokens") + if !ok { + return sqlparser.ErrMissingTimeseries, false + } + tokens, ok := whereTokens.([]token.Tokens) + if !ok { + return sqlparser.ErrMissingTimeseries, false + } + tscol := ro.BaseTimestampFieldName + var start, end time.Time + var startInclusive, endInclusive bool + var hasStart, hasEnd bool + var canCache bool = true + var err error + // Run through statements to parse out start/end + for idxStatement := 0; idxStatement < len(tokens); idxStatement++ { + statement := tokens[idxStatement] + if !hasStart && isTimeseriesStartStatement(statement, tscol) { + start, _, err = lsql.TokenToTime(statement[2]) + if err != nil { + return sqlparser.ErrNoLowerBound, false + } + if statement[2].Typ.IsOrEquals() { + startInclusive = true + } + hasStart = true + } else if !hasStart && isTimeseriesBetweenStatement(statement, tscol) { + start, _, err = lsql.TokenToTime(statement[2]) + if err != nil { + return sqlparser.ErrNoLowerBound, false + } + end, _, err = lsql.TokenToTime(statement[4]) + if err != nil { + return sqlparser.ErrNoUpperBound, false + } + hasStart = true + hasEnd = true + break + } else if hasStart && isTimeseriesEndStatement(statement, tscol) { + end, _, err = lsql.TokenToTime(statement[2]) + if err != nil { + return sqlparser.ErrNoUpperBound, false + } + if statement[2].Typ.IsOrEquals() { + endInclusive = true + } + hasEnd = true + break + } else if statementContainsColumn(statement, tscol) { + canCache = false + } + } + // Use parsed info to update TRQ/RO + // Check for error conditions first + if !hasStart { + return sqlparser.ErrNoLowerBound, false + } + if !hasEnd { + end = time.Now() + } + // Account for inclusivity; if not OrEqualTo, we need to adjust by the TRQ step. + if !startInclusive { + start = start.Add(trq.Step) + } + if !endInclusive { + end = end.Add(trq.Step * -1) + } + trq.Extent = timeseries.Extent{ + Start: start, + End: end, + } + return nil, canCache +} + +func parseGroupByTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *timeseries.RequestOptions) error { + gbTokens, ok := rs.GetResultsCollection("groupByTokens") + if !ok { + return lsql.ErrInvalidGroupByClause + } + tokens, ok := gbTokens.(token.Tokens) + if !ok || len(tokens) == 0 { + return lsql.ErrInvalidGroupByClause + } + trq.TagFieldDefintions = make([]timeseries.FieldDefinition, len(tokens)) + for i, v := range tokens { + trq.TagFieldDefintions[i].Name = v.Val + } + return nil +} + +// TIMESERIES STUFF + +func isTimeseriesStartStatement(statement token.Tokens, forColumn string) bool { + return len(statement) == 3 && + (statement[0].Typ == token.Identifier && statement[0].Val == forColumn) && + (statement[1].Typ.IsGTorGTE()) && + (statement[2].Typ == token.Number) +} + +func isTimeseriesEndStatement(statement token.Tokens, forColumn string) bool { + return len(statement) == 3 && + (statement[0].Typ == token.Identifier && statement[0].Val == forColumn) && + (statement[1].Typ.IsLTorLTE()) && + (statement[2].Typ == token.Number) +} + +func isTimeseriesBetweenStatement(statement token.Tokens, forColumn string) bool { + return len(statement) == 5 && + (statement[0].Typ == token.Identifier && statement[0].Val == forColumn) && + (statement[1].Val == "between") && + (statement[2].Typ == token.Number) && + (statement[3].Typ == token.LogicalAnd) && + (statement[4].Typ == token.Number) +} + +func statementContainsColumn(statement token.Tokens, forColumn string) bool { + for _, t := range statement { + if t.Val == forColumn { + return true + } + } + return false +} diff --git a/pkg/backends/mysql/parsing_test.go b/pkg/backends/mysql/parsing_test.go new file mode 100644 index 000000000..37d5422c0 --- /dev/null +++ b/pkg/backends/mysql/parsing_test.go @@ -0,0 +1,50 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "strconv" + "testing" +) + +const tq00 = `SELECT col1, col2 FROM table WHERE col1 >= 1589904000 AND col1 < 1589997600 GROUP BY col1 ORDER BY col1` +const tq01 = `SELECT col1, col2 FROM table WHERE col1 BETWEEN 1589904000 AND 1589997600 GROUP BY col1 ORDER BY col1` +const tq02 = `SELECT col1 as ts, col2 FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts ORDER BY col1` +const tq03 = `SELECT DATETIME(col1) as ts, col2, COUNT() as ct FROM table WHERE ts BETWEEN 1589904000 AND 1589997600 GROUP BY ts ORDER BY ts` +const tq04 = `SELECT col1 as ts, col2 FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts ORDER BY col1 LIMIT 10` + +func TestParseRawQuery(t *testing.T) { + tests := []struct { + query string + err error + }{ + {tq00, nil}, + {tq01, nil}, + {tq02, nil}, + {tq03, nil}, + {tq04, ErrLimitUnsupported}, + } + for i, test := range tests { + t.Run(strconv.Itoa(i), func(t *testing.T) { + trq, _, _, err := parse(test.query) + if err != test.err { + t.Errorf("got '%v' expected '%v'", err, test.err) + } + t.Logf("%+v\n", trq) + }) + } +} diff --git a/pkg/backends/mysql/routes.go b/pkg/backends/mysql/routes.go new file mode 100644 index 000000000..046669530 --- /dev/null +++ b/pkg/backends/mysql/routes.go @@ -0,0 +1,53 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "net/http" + + bo "github.com/trickstercache/trickster/v2/pkg/backends/options" + "github.com/trickstercache/trickster/v2/pkg/proxy/paths/matching" + po "github.com/trickstercache/trickster/v2/pkg/proxy/paths/options" +) + +func (c *Client) RegisterHandlers(map[string]http.Handler) { + + c.TimeseriesBackend.RegisterHandlers( + map[string]http.Handler{ + // This is the registry of handlers that Trickster supports for ClickHouse, + // and are able to be referenced by name (map key) in Config Files + "health": http.HandlerFunc(c.HealthHandler), + "query": http.HandlerFunc(c.QueryHandler), + "proxy": http.HandlerFunc(c.ProxyHandler), + }, + ) +} + +// DefaultPathConfigs returns the default PathConfigs for the given Provider +func (c *Client) DefaultPathConfigs(o *bo.Options) map[string]*po.Options { + paths := map[string]*po.Options{ + "/": { + Path: "/", + HandlerName: "query", + Methods: []string{http.MethodGet, http.MethodPost}, + MatchType: matching.PathMatchTypePrefix, + MatchTypeName: "prefix", + CacheKeyParams: []string{"query", "database"}, + }, + } + return paths +} diff --git a/pkg/backends/mysql/routes_test.go b/pkg/backends/mysql/routes_test.go new file mode 100644 index 000000000..a65c207db --- /dev/null +++ b/pkg/backends/mysql/routes_test.go @@ -0,0 +1,68 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "testing" + + "github.com/trickstercache/trickster/v2/pkg/proxy/request" + tu "github.com/trickstercache/trickster/v2/pkg/testutil" +) + +func TestRegisterHandlers(t *testing.T) { + c, err := NewClient("test", nil, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + c.RegisterHandlers(nil) + if _, ok := c.Handlers()["query"]; !ok { + t.Errorf("expected to find handler named: %s", "query") + } +} + +func TestDefaultPathConfigs(t *testing.T) { + + backendClient, err := NewClient("test", nil, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + ts, _, r, _, err := tu.NewTestInstance("", backendClient.DefaultPathConfigs, 204, "", + nil, "clickhouse", "/", "debug") + if err != nil { + t.Error(err) + } else { + defer ts.Close() + } + rsc := request.GetResources(r) + backendClient, err = NewClient("test", rsc.BackendOptions, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + client := backendClient.(*Client) + rsc.BackendClient = client + rsc.BackendOptions.HTTPClient = backendClient.HTTPClient() + + if _, ok := backendClient.Configuration().Paths["/"]; !ok { + t.Errorf("expected to find path named: %s", "/") + } + + const expectedLen = 1 + if len(backendClient.Configuration().Paths) != expectedLen { + t.Errorf("expected %d got %d", expectedLen, len(backendClient.Configuration().Paths)) + } + +} diff --git a/pkg/backends/mysql/stubs.go b/pkg/backends/mysql/stubs.go new file mode 100644 index 000000000..43403495c --- /dev/null +++ b/pkg/backends/mysql/stubs.go @@ -0,0 +1,43 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "net/http" + + "github.com/trickstercache/trickster/v2/pkg/timeseries" +) + +// This file holds funcs required by the Proxy Client or Timeseries interfaces, +// but are (currently) unused by the ClickHouse implementation. + +// Series (timeseries.Timeseries Interface) stub funcs + +// FastForwardRequest is not used for ClickHouse and is here to conform to the Proxy Client interface +func (c *Client) FastForwardRequest(r *http.Request) (*http.Request, error) { + return nil, nil +} + +// ClickHouse Client (proxy.Client Interface) stub funcs + +// UnmarshalInstantaneous is not used for ClickHouse and is here to conform to the Proxy Client interface +func (c *Client) UnmarshalInstantaneous(data []byte) (timeseries.Timeseries, error) { + return nil, nil +} + +// QueryRangeHandler is not used for ClickHouse and is here to conform to the Proxy Client interface +func (c *Client) QueryRangeHandler(w http.ResponseWriter, r *http.Request) {} diff --git a/pkg/backends/mysql/stubs_test.go b/pkg/backends/mysql/stubs_test.go new file mode 100644 index 000000000..d0a82224a --- /dev/null +++ b/pkg/backends/mysql/stubs_test.go @@ -0,0 +1,53 @@ +/* + * Copyright 2018 The Trickster Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package mysql + +import ( + "testing" +) + +func TestFastForwardURL(t *testing.T) { + + client := &Client{} + r, err := client.FastForwardRequest(nil) + if r != nil { + t.Errorf("Expected nil url, got %v", r) + } + if err != nil { + t.Errorf("Expected nil err, got %s", err) + } +} + +func TestUnmarshalInstantaneous(t *testing.T) { + + client := &Client{} + tr, err := client.UnmarshalInstantaneous(nil) + + if tr != nil { + t.Errorf("Expected nil timeseries, got %s", tr) + } + + if err != nil { + t.Errorf("Expected nil err, got %s", err) + } + +} + +func TestQueryRangeHandler(t *testing.T) { + client := &Client{} + client.QueryRangeHandler(nil, nil) +} diff --git a/pkg/parsing/run_state.go b/pkg/parsing/run_state.go index 98acf11b5..725689a3b 100644 --- a/pkg/parsing/run_state.go +++ b/pkg/parsing/run_state.go @@ -29,7 +29,6 @@ type RunState struct { err error ctx context.Context nextOverride StateFn - isPeeked bool results map[string]interface{} } @@ -54,7 +53,7 @@ func (rs *RunState) GetResultsCollection(collectionName string) (interface{}, bo return v, ok } -// Results returns the results objecxt from the RunState +// Results returns the results object from the RunState func (rs *RunState) Results() map[string]interface{} { return rs.results } From d71a611a21f456c9f9c8c64ccb6e144b1909d4bb Mon Sep 17 00:00:00 2001 From: James Ranson Date: Sun, 22 Jan 2023 11:11:32 -0700 Subject: [PATCH 2/7] fix codespell typo Signed-off-by: James Ranson --- pkg/backends/mysql/model/datatypes.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/backends/mysql/model/datatypes.go b/pkg/backends/mysql/model/datatypes.go index 3cd690733..41ea4ebb1 100644 --- a/pkg/backends/mysql/model/datatypes.go +++ b/pkg/backends/mysql/model/datatypes.go @@ -41,7 +41,7 @@ const ( var dts = []string{ "integer", "smallint", "tinyint", "mediumint", "bigint", - "decimal", "numberic", + "decimal", "numeric", "float", "double", "bit", "date", "datetime", "timestamp", "time", "year", From 17c49675200981848ecfad0d5ff1de54a5bb4b9d Mon Sep 17 00:00:00 2001 From: James Ranson Date: Sun, 22 Jan 2023 12:26:03 -0700 Subject: [PATCH 3/7] fix failing tests Signed-off-by: James Ranson --- cmd/trickster/listeners.go | 16 ++++++++++++---- cmd/trickster/main.go | 4 +++- cmd/trickster/main_test.go | 1 + 3 files changed, 16 insertions(+), 5 deletions(-) diff --git a/cmd/trickster/listeners.go b/cmd/trickster/listeners.go index 3f4de54cf..f35e67f28 100644 --- a/cmd/trickster/listeners.go +++ b/cmd/trickster/listeners.go @@ -87,7 +87,9 @@ func applyListenerConfigs(conf, oldConf *config.Config, if err != nil { tl.Error(log, "unable to start tls listener due to certificate error", tl.Pairs{"detail": err}) } else { - wg.Add(1) + if wg != nil { + wg.Add(1) + } tracerFlusherSet = true go lg.StartListener("tlsListener", conf.Frontend.TLSListenAddress, conf.Frontend.TLSListenPort, @@ -118,7 +120,9 @@ func applyListenerConfigs(conf, oldConf *config.Config, (oldConf.Frontend.ListenAddress != conf.Frontend.ListenAddress || oldConf.Frontend.ListenPort != conf.Frontend.ListenPort)) { lg.DrainAndClose("httpListener", drainTimeout) - wg.Add(1) + if wg != nil { + wg.Add(1) + } var t2 tracing.Tracers if !tracerFlusherSet { t2 = tracers @@ -138,7 +142,9 @@ func applyListenerConfigs(conf, oldConf *config.Config, if conf.Main.PprofServer == "both" || conf.Main.PprofServer == "metrics" { routing.RegisterPprofRoutes("metrics", metricsRouter, log) } - wg.Add(1) + if wg != nil { + wg.Add(1) + } go lg.StartListener("metricsListener", conf.Metrics.ListenAddress, conf.Metrics.ListenPort, conf.Frontend.ConnectionsLimit, nil, metricsRouter, wg, nil, exitFunc, 0, log) @@ -154,7 +160,9 @@ func applyListenerConfigs(conf, oldConf *config.Config, if conf.ReloadConfig != nil && conf.ReloadConfig.ListenPort > 0 && (!hasOldRC || (conf.ReloadConfig.ListenAddress != oldConf.ReloadConfig.ListenAddress || conf.ReloadConfig.ListenPort != oldConf.ReloadConfig.ListenPort)) { - wg.Add(1) + if wg != nil { + wg.Add(1) + } lg.DrainAndClose("reloadListener", time.Millisecond*500) rr.HandleFunc(conf.Main.ConfigHandlerPath, ph.ConfigHandleFunc(conf)) rr.Handle(conf.ReloadConfig.HandlerPath, reloadHandler) diff --git a/cmd/trickster/main.go b/cmd/trickster/main.go index d778f4394..a99dc3486 100644 --- a/cmd/trickster/main.go +++ b/cmd/trickster/main.go @@ -42,7 +42,9 @@ func main() { runtime.ApplicationName = applicationName runtime.ApplicationVersion = applicationVersion runConfig(nil, wg, nil, nil, os.Args[1:], exitFunc) - wg.Wait() + if wg != nil { + wg.Wait() + } } func exitFatal() { diff --git a/cmd/trickster/main_test.go b/cmd/trickster/main_test.go index 94fa4f2a6..470b8179e 100644 --- a/cmd/trickster/main_test.go +++ b/cmd/trickster/main_test.go @@ -23,6 +23,7 @@ import ( func TestMain(t *testing.T) { exitFunc = nil + wg = nil main() // Successful test criteria is that the call to main returns without timing out on wg.Wait() } From 183e24f1cc7e189478458f2b8063e7f63fc558e2 Mon Sep 17 00:00:00 2001 From: James Ranson Date: Sun, 22 Jan 2023 12:26:28 -0700 Subject: [PATCH 4/7] s/clickhouse/MySQL/gi Signed-off-by: James Ranson --- pkg/backends/mysql/handler_query_test.go | 6 +----- pkg/backends/mysql/model/model.go | 16 +++------------- pkg/backends/mysql/mysql.go | 8 ++++---- pkg/backends/mysql/mysql_test.go | 6 +++--- pkg/backends/mysql/routes.go | 2 +- pkg/backends/mysql/routes_test.go | 2 +- pkg/backends/mysql/stubs.go | 10 +++++----- 7 files changed, 18 insertions(+), 32 deletions(-) diff --git a/pkg/backends/mysql/handler_query_test.go b/pkg/backends/mysql/handler_query_test.go index 715786eca..c6693c7fb 100644 --- a/pkg/backends/mysql/handler_query_test.go +++ b/pkg/backends/mysql/handler_query_test.go @@ -28,11 +28,7 @@ import ( ) func testRawQuery() string { - return url.Values(map[string][]string{"query": { - `SELECT (intDiv(toUInt32(time_column), 60) * 60) * 1000 AS t, countMerge(some_count) AS cnt, field1, field2 ` + - `FROM testdb.test_table WHERE time_column BETWEEN toDateTime(1516665600) AND toDateTime(1516687200) ` + - `AND date_column >= toDate(1516665600) AND toDate(1516687200) ` + - `AND field1 > 0 AND field2 = 'some_value' GROUP BY t, field1, field2 ORDER BY t, field1 FORMAT JSON`}}). + return url.Values(map[string][]string{"query": {tq00}}). Encode() } diff --git a/pkg/backends/mysql/model/model.go b/pkg/backends/mysql/model/model.go index 1b09580b2..93b6b9258 100644 --- a/pkg/backends/mysql/model/model.go +++ b/pkg/backends/mysql/model/model.go @@ -75,7 +75,7 @@ func (cs cols) String() string { return sb.String() } -// NewModeler returns a collection of modeling functions for clickhouse interoperability +// NewModeler returns a collection of modeling functions for mysql interoperability func NewModeler() *timeseries.Modeler { return ×eries.Modeler{ WireUnmarshalerReader: UnmarshalTimeseriesReader, @@ -104,7 +104,6 @@ func marshalTimeseriesJSON(ds *dataset.DataSet, rlo *timeseries.RequestOptions, if rw, ok := w.(http.ResponseWriter); ok { h := rw.Header() h.Set(headers.NameContentType, headers.ValueApplicationJSON+"; charset=UTF-8") - h.Set("X-Clickhouse-Format", "JSON") rw.WriteHeader(status) } @@ -272,25 +271,16 @@ func marshalTimeseriesXSV(ds *dataset.DataSet, rlo *timeseries.RequestOptions, tw.separator = "," } - var ctPart, fmtPart string + var ctPart string switch tw.separator { case "\t": ctPart = "tab" - fmtPart = "TSV" default: ctPart = "comma" - fmtPart = "CSV" tw.separator = "," } h.Set(headers.NameContentType, "text/"+ctPart+"-separated-values; charset=UTF-8") - if tw.writeTypes { - h.Set("X-Clickhouse-Format", fmtPart+"WithNamesAndTypes") - } else if tw.writeNames { - h.Set("X-Clickhouse-Format", fmtPart+"WithNames") - } else { - h.Set("X-Clickhouse-Format", fmtPart) - } rw.WriteHeader(status) } @@ -445,7 +435,7 @@ func UnmarshalTimeseriesReader(reader io.Reader, trq *timeseries.TimeRangeQuery) return nil, timeseries.ErrNoTimerangeQuery } - // ClickHouse doesn't support multi-statements, so there is always exactly 1 result element + // don't support multi-statements, so there is always exactly 1 result element r := &dataset.Result{ SeriesList: make([]*dataset.Series, 0, 8), } diff --git a/pkg/backends/mysql/mysql.go b/pkg/backends/mysql/mysql.go index cf72255a2..6fc1dcfc3 100644 --- a/pkg/backends/mysql/mysql.go +++ b/pkg/backends/mysql/mysql.go @@ -14,7 +14,7 @@ * limitations under the License. */ -// package mysql provides the ClickHouse backend provider +// package mysql provides the MySQL backend provider package mysql import ( @@ -23,7 +23,7 @@ import ( "time" "github.com/trickstercache/trickster/v2/pkg/backends" - modelch "github.com/trickstercache/trickster/v2/pkg/backends/clickhouse/model" + "github.com/trickstercache/trickster/v2/pkg/backends/mysql/model" bo "github.com/trickstercache/trickster/v2/pkg/backends/options" "github.com/trickstercache/trickster/v2/pkg/backends/providers/registration/types" "github.com/trickstercache/trickster/v2/pkg/cache" @@ -51,7 +51,7 @@ func NewClient(name string, o *bo.Options, router http.Handler, o.FastForwardDisable = true } c := &Client{} - b, err := backends.NewTimeseriesBackend(name, o, c.RegisterHandlers, router, cache, modelch.NewModeler()) + b, err := backends.NewTimeseriesBackend(name, o, c.RegisterHandlers, router, cache, model.NewModeler()) c.TimeseriesBackend = b return c, err } @@ -81,7 +81,7 @@ func (c *Client) ParseTimeRangeQuery(r *http.Request) (*timeseries.TimeRangeQuer var bf time.Duration res := request.GetResources(r) if res == nil { - // 60-second default backfill tolerance for ClickHouse + // 60-second default backfill tolerance for MySQL bf = 60 * time.Second } else { bf = res.BackendOptions.BackfillTolerance diff --git a/pkg/backends/mysql/mysql_test.go b/pkg/backends/mysql/mysql_test.go index 8ebfaeaf6..6cfffa1cc 100644 --- a/pkg/backends/mysql/mysql_test.go +++ b/pkg/backends/mysql/mysql_test.go @@ -23,7 +23,7 @@ import ( "github.com/trickstercache/trickster/v2/cmd/trickster/config" "github.com/trickstercache/trickster/v2/pkg/backends" - "github.com/trickstercache/trickster/v2/pkg/backends/clickhouse/model" + "github.com/trickstercache/trickster/v2/pkg/backends/mysql/model" bo "github.com/trickstercache/trickster/v2/pkg/backends/options" cr "github.com/trickstercache/trickster/v2/pkg/cache/registration" tl "github.com/trickstercache/trickster/v2/pkg/observability/logging" @@ -31,7 +31,7 @@ import ( var testModeler = model.NewModeler() -func TestClickhouseClientInterfacing(t *testing.T) { +func TestClientInterfacing(t *testing.T) { // this test ensures the client will properly conform to the // Client and TimeseriesBackend interfaces @@ -55,7 +55,7 @@ func TestClickhouseClientInterfacing(t *testing.T) { func TestNewClient(t *testing.T) { - conf, _, err := config.Load("trickster", "test", []string{"-provider", "clickhouse", "-origin-url", "http://1"}) + conf, _, err := config.Load("trickster", "test", []string{"-provider", "mysql", "-origin-url", "http://1"}) if err != nil { t.Fatalf("Could not load configuration: %s", err.Error()) } diff --git a/pkg/backends/mysql/routes.go b/pkg/backends/mysql/routes.go index 046669530..7a51c7799 100644 --- a/pkg/backends/mysql/routes.go +++ b/pkg/backends/mysql/routes.go @@ -28,7 +28,7 @@ func (c *Client) RegisterHandlers(map[string]http.Handler) { c.TimeseriesBackend.RegisterHandlers( map[string]http.Handler{ - // This is the registry of handlers that Trickster supports for ClickHouse, + // This is the registry of handlers that Trickster supports for MySQL, // and are able to be referenced by name (map key) in Config Files "health": http.HandlerFunc(c.HealthHandler), "query": http.HandlerFunc(c.QueryHandler), diff --git a/pkg/backends/mysql/routes_test.go b/pkg/backends/mysql/routes_test.go index a65c207db..cb4161636 100644 --- a/pkg/backends/mysql/routes_test.go +++ b/pkg/backends/mysql/routes_test.go @@ -41,7 +41,7 @@ func TestDefaultPathConfigs(t *testing.T) { t.Error(err) } ts, _, r, _, err := tu.NewTestInstance("", backendClient.DefaultPathConfigs, 204, "", - nil, "clickhouse", "/", "debug") + nil, "mysql", "/", "debug") if err != nil { t.Error(err) } else { diff --git a/pkg/backends/mysql/stubs.go b/pkg/backends/mysql/stubs.go index 43403495c..e66ddb212 100644 --- a/pkg/backends/mysql/stubs.go +++ b/pkg/backends/mysql/stubs.go @@ -23,21 +23,21 @@ import ( ) // This file holds funcs required by the Proxy Client or Timeseries interfaces, -// but are (currently) unused by the ClickHouse implementation. +// but are (currently) unused by the MySQL implementation. // Series (timeseries.Timeseries Interface) stub funcs -// FastForwardRequest is not used for ClickHouse and is here to conform to the Proxy Client interface +// FastForwardRequest is not used for MySQL and is here to conform to the Proxy Client interface func (c *Client) FastForwardRequest(r *http.Request) (*http.Request, error) { return nil, nil } -// ClickHouse Client (proxy.Client Interface) stub funcs +// MySQL Client (proxy.Client Interface) stub funcs -// UnmarshalInstantaneous is not used for ClickHouse and is here to conform to the Proxy Client interface +// UnmarshalInstantaneous is not used for MySQL and is here to conform to the Proxy Client interface func (c *Client) UnmarshalInstantaneous(data []byte) (timeseries.Timeseries, error) { return nil, nil } -// QueryRangeHandler is not used for ClickHouse and is here to conform to the Proxy Client interface +// QueryRangeHandler is not used for MySQL and is here to conform to the Proxy Client interface func (c *Client) QueryRangeHandler(w http.ResponseWriter, r *http.Request) {} From ea0bdb29af2c85959e4fed08483a5256124fc7fd Mon Sep 17 00:00:00 2001 From: jakenichols2719 Date: Tue, 30 May 2023 12:52:50 -0600 Subject: [PATCH 5/7] Intepret timeseries column division as step, test more queries Signed-off-by: jakenichols2719 --- pkg/backends/mysql/handler_query_test.go | 4 ++ pkg/backends/mysql/mysql_test.go | 87 ++++++++++++------------ pkg/backends/mysql/parsing.go | 20 ++++++ pkg/backends/mysql/parsing_test.go | 11 +-- 4 files changed, 73 insertions(+), 49 deletions(-) diff --git a/pkg/backends/mysql/handler_query_test.go b/pkg/backends/mysql/handler_query_test.go index c6693c7fb..c4dbd0dfc 100644 --- a/pkg/backends/mysql/handler_query_test.go +++ b/pkg/backends/mysql/handler_query_test.go @@ -27,6 +27,10 @@ import ( tu "github.com/trickstercache/trickster/v2/pkg/testutil" ) +func testQuery(q string) string { + return url.Values(map[string][]string{"query": {q}}).Encode() +} + func testRawQuery() string { return url.Values(map[string][]string{"query": {tq00}}). Encode() diff --git a/pkg/backends/mysql/mysql_test.go b/pkg/backends/mysql/mysql_test.go index 6cfffa1cc..383a57c80 100644 --- a/pkg/backends/mysql/mysql_test.go +++ b/pkg/backends/mysql/mysql_test.go @@ -20,6 +20,7 @@ import ( "net/http" "net/url" "testing" + "time" "github.com/trickstercache/trickster/v2/cmd/trickster/config" "github.com/trickstercache/trickster/v2/pkg/backends" @@ -87,51 +88,49 @@ func TestNewClient(t *testing.T) { } func TestParseTimeRangeQuery(t *testing.T) { - req := &http.Request{URL: &url.URL{ - Scheme: "https", - Host: "blah.com", - Path: "/", - RawQuery: testRawQuery(), - }, - Header: http.Header{}, + type test_query struct { + name string + raw string + pass bool + step time.Duration + over time.Duration } - client := &Client{} - res, _, _, err := client.ParseTimeRangeQuery(req) - if err != nil { - t.Error(err) - } else { - if res.Step.Seconds() != 60 { - t.Errorf("expected 60 got %f", res.Step.Seconds()) - } - if res.Extent.End.Sub(res.Extent.Start).Hours() != 6 { - t.Errorf("expected 6 got %f", res.Extent.End.Sub(res.Extent.Start).Hours()) - } - } - - req.URL.RawQuery = "" - _, _, _, err = client.ParseTimeRangeQuery(req) - if err == nil { - t.Errorf("expected error for: %s", "missing URL parameter: [query]") + tests := []test_query{ + {"basic", tq00, true, 60 * time.Second, 26 * time.Hour}, + {"between", tq01, true, 60 * time.Second, 26 * time.Hour}, + {"order-by", tq02, true, 60 * time.Second, 26 * time.Hour}, + {"count", tq03, true, 30 * time.Second, 26 * time.Hour}, + {"limit", tq04, false, 0, 0}, + {"nostep", tq05, false, 0, 0}, + {"empty", "", false, 0, 0}, } - - req.URL.RawQuery = url.Values(map[string][]string{"query": { - `SELECT (intDiv(toUInt32(abc), 6z0) * 6z0) * 1000 AS t, countMerge(some_count) AS cnt, field1, field2 ` + - `FROM testdb.test_table WHERE abc BETWEEN toDateTime(1516665600) AND toDateTime(1516687200) ` + - `AND date_column >= toDate(1516665600) AND toDate(1516687200) ` + - `AND field1 > 0 AND field2 = 'some_value' GROUP BY t, field1, field2 ORDER BY t, field1 FORMAT JSON`}}).Encode() - _, _, _, err = client.ParseTimeRangeQuery(req) - if err == nil { - t.Errorf("expected error for: %s", "not a time range query") - } - - req.URL.RawQuery = url.Values(map[string][]string{"query": { - `SELECT (intDiv(toUInt32(0^^^), 60) * 60) * 1000 AS t, countMerge(some_count) AS cnt, field1, field2 ` + - `FROM testdb.test_table WHERE 0^^^ BETWEEN toDateTime(1516665600) AND toDateTime(1516687200) ` + - `AND date_column >= toDate(1516665600) AND toDate(1516687200) ` + - `AND field1 > 0 AND field2 = 'some_value' GROUP BY t, field1, field2 ORDER BY t, field1 FORMAT JSON`}}).Encode() - _, _, _, err = client.ParseTimeRangeQuery(req) - if err == nil { - t.Errorf("expected error for: %s", "not a time range query") + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + req := &http.Request{URL: &url.URL{ + Scheme: "https", + Host: "blah.com", + Path: "/", + RawQuery: testQuery(test.raw), + }, + Header: http.Header{}, + } + client := &Client{} + res, _, _, err := client.ParseTimeRangeQuery(req) + if err != nil { + if test.pass { + t.Error(err) + } + } else { + if !test.pass { + t.Error("expected parsing to fail") + } + if res.Step != test.step { + t.Errorf("expected step %v, got %v", test.step, res.Step) + } + if res.Extent.End.Sub(res.Extent.Start) != test.over { + t.Errorf("expected query over duration %v, got %v", test.over, res.Extent.End.Sub(res.Extent.Start)) + } + } + }) } - } diff --git a/pkg/backends/mysql/parsing.go b/pkg/backends/mysql/parsing.go index 97eb7ce91..daca3fcdb 100644 --- a/pkg/backends/mysql/parsing.go +++ b/pkg/backends/mysql/parsing.go @@ -17,6 +17,9 @@ package mysql import ( + "fmt" + "strconv" + "strings" "time" "github.com/trickstercache/trickster/v2/pkg/backends/mysql/model" @@ -237,9 +240,26 @@ func parseGroupByTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro if !ok || len(tokens) == 0 { return lsql.ErrInvalidGroupByClause } + var nextStep, hasStep bool trq.TagFieldDefintions = make([]timeseries.FieldDefinition, len(tokens)) for i, v := range tokens { trq.TagFieldDefintions[i].Name = v.Val + if strings.ToLower(v.Val) == "div" { + nextStep = true + continue + } + if nextStep { + step, err := strconv.ParseInt(v.Val, 10, 64) + if err != nil { + return fmt.Errorf("failed to parse step from query: %s", err) + } + trq.Step = time.Duration(step) * time.Second + hasStep = true + nextStep = false + } + } + if !hasStep { + return fmt.Errorf("queries require a step; use GROUP BY tscol DIV (step in seconds)") } return nil } diff --git a/pkg/backends/mysql/parsing_test.go b/pkg/backends/mysql/parsing_test.go index 37d5422c0..f51aa030e 100644 --- a/pkg/backends/mysql/parsing_test.go +++ b/pkg/backends/mysql/parsing_test.go @@ -21,11 +21,12 @@ import ( "testing" ) -const tq00 = `SELECT col1, col2 FROM table WHERE col1 >= 1589904000 AND col1 < 1589997600 GROUP BY col1 ORDER BY col1` -const tq01 = `SELECT col1, col2 FROM table WHERE col1 BETWEEN 1589904000 AND 1589997600 GROUP BY col1 ORDER BY col1` -const tq02 = `SELECT col1 as ts, col2 FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts ORDER BY col1` -const tq03 = `SELECT DATETIME(col1) as ts, col2, COUNT() as ct FROM table WHERE ts BETWEEN 1589904000 AND 1589997600 GROUP BY ts ORDER BY ts` -const tq04 = `SELECT col1 as ts, col2 FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts ORDER BY col1 LIMIT 10` +const tq00 = `SELECT col1, AVG(col2) FROM table WHERE col1 >= 1589904000 AND col1 < 1589997600 GROUP BY col1 div 60 ORDER BY col1` +const tq01 = `SELECT col1, AVG(col2) FROM table WHERE col1 BETWEEN 1589904000 AND 1589997600 GROUP BY col1 div 60 ORDER BY col1` +const tq02 = `SELECT col1 as ts, SUM(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts div 60 ORDER BY col1` +const tq03 = `SELECT DATETIME(col1) as ts, SUM(col2), COUNT() as ct FROM table WHERE ts BETWEEN 1589904000 AND 1589997600 GROUP BY ts div 30 ORDER BY ts` +const tq04 = `SELECT col1 as ts, AVG(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts div 60 ORDER BY col1 LIMIT 10` +const tq05 = `SELECT col1 as ts, AVG(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts ORDER BY ts` func TestParseRawQuery(t *testing.T) { tests := []struct { From cd46f4755fd47ae88b39237ba638190a346dadb3 Mon Sep 17 00:00:00 2001 From: jakenichols2719 Date: Wed, 31 May 2023 11:54:00 -0600 Subject: [PATCH 6/7] Handle aggregate functions on tscol Signed-off-by: jakenichols2719 --- pkg/backends/mysql/model/datatypes.go | 8 ++++++++ pkg/backends/mysql/parsing.go | 4 ++++ pkg/backends/mysql/parsing_test.go | 12 ++++++------ 3 files changed, 18 insertions(+), 6 deletions(-) diff --git a/pkg/backends/mysql/model/datatypes.go b/pkg/backends/mysql/model/datatypes.go index 41ea4ebb1..f5ac62d60 100644 --- a/pkg/backends/mysql/model/datatypes.go +++ b/pkg/backends/mysql/model/datatypes.go @@ -55,3 +55,11 @@ func IsDataType(tk *token.Token) (string, bool) { func (dt DataType) String() string { return dts[dt] } + +var afs = []string{ + "min", "max", "avg", +} + +func IsAggregateFunction(tk *token.Token) (string, bool) { + return tk.Val, tk.Typ == token.Identifier && strings.IndexInSlice(afs, tk.Val) != -1 +} diff --git a/pkg/backends/mysql/parsing.go b/pkg/backends/mysql/parsing.go index daca3fcdb..fce9e26b5 100644 --- a/pkg/backends/mysql/parsing.go +++ b/pkg/backends/mysql/parsing.go @@ -99,6 +99,7 @@ func parseTSColumn(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *tim if !ok { return sqlparser.ErrMissingTimeseries } + fmt.Println(tokens) // Parse the first "statement" in the select portion. Need to check for datatype, column, alias stmnt := tokens[0] var col, alias string @@ -115,6 +116,9 @@ func parseTSColumn(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *tim if _, isDT := model.IsDataType(t); isDT { continue } + if _, isAF := model.IsAggregateFunction(t); isAF { + continue + } if !foundCol { col = t.Val foundCol = true diff --git a/pkg/backends/mysql/parsing_test.go b/pkg/backends/mysql/parsing_test.go index f51aa030e..d9e3c4181 100644 --- a/pkg/backends/mysql/parsing_test.go +++ b/pkg/backends/mysql/parsing_test.go @@ -21,12 +21,12 @@ import ( "testing" ) -const tq00 = `SELECT col1, AVG(col2) FROM table WHERE col1 >= 1589904000 AND col1 < 1589997600 GROUP BY col1 div 60 ORDER BY col1` -const tq01 = `SELECT col1, AVG(col2) FROM table WHERE col1 BETWEEN 1589904000 AND 1589997600 GROUP BY col1 div 60 ORDER BY col1` -const tq02 = `SELECT col1 as ts, SUM(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts div 60 ORDER BY col1` -const tq03 = `SELECT DATETIME(col1) as ts, SUM(col2), COUNT() as ct FROM table WHERE ts BETWEEN 1589904000 AND 1589997600 GROUP BY ts div 30 ORDER BY ts` -const tq04 = `SELECT col1 as ts, AVG(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts div 60 ORDER BY col1 LIMIT 10` -const tq05 = `SELECT col1 as ts, AVG(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts ORDER BY ts` +const tq00 = `SELECT MIN(col1), AVG(col2) FROM table WHERE col1 >= 1589904000 AND col1 < 1589997600 GROUP BY col1 div 60 ORDER BY col1` +const tq01 = `SELECT MIN(col1), AVG(col2) FROM table WHERE col1 BETWEEN 1589904000 AND 1589997600 GROUP BY col1 div 60 ORDER BY col1` +const tq02 = `SELECT MIN(col1) as ts, SUM(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts div 60 ORDER BY col1` +const tq03 = `SELECT DATETIME(MIN(col1)) as ts, SUM(col2), COUNT() as ct FROM table WHERE ts BETWEEN 1589904000 AND 1589997600 GROUP BY ts div 30 ORDER BY ts` +const tq04 = `SELECT MIN(col1) as ts, AVG(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts div 60 ORDER BY col1 LIMIT 10` +const tq05 = `SELECT MIN as ts, AVG(col2) FROM table WHERE ts > 1589904000 AND ts <= 1589997600 GROUP BY ts ORDER BY ts` func TestParseRawQuery(t *testing.T) { tests := []struct { From 8eff7198262215c96647e458d67d8c305310331e Mon Sep 17 00:00:00 2001 From: jakenichols2719 Date: Wed, 31 May 2023 12:55:19 -0600 Subject: [PATCH 7/7] Add SetExtent for mysql client Signed-off-by: jakenichols2719 --- pkg/backends/mysql/parsing.go | 28 ++++++++++++++-- pkg/backends/mysql/parsing_test.go | 3 +- pkg/backends/mysql/url.go | 25 ++++++++++++++ pkg/backends/mysql/url_test.go | 53 ++++++++++++++++++++++++++++++ 4 files changed, 105 insertions(+), 4 deletions(-) create mode 100644 pkg/backends/mysql/url.go create mode 100644 pkg/backends/mysql/url_test.go diff --git a/pkg/backends/mysql/parsing.go b/pkg/backends/mysql/parsing.go index fce9e26b5..e372e10b1 100644 --- a/pkg/backends/mysql/parsing.go +++ b/pkg/backends/mysql/parsing.go @@ -31,11 +31,24 @@ import ( "github.com/trickstercache/trickster/v2/pkg/timeseries/sqlparser" ) +type sqlquery struct { + tscol string + rangeStart int + rangeEnd int + extent *timeseries.Extent + query string +} + +func (sq *sqlquery) String() string { + r := fmt.Sprintf("%s >= %d AND %s < %d", sq.tscol, sq.extent.Start.Unix(), sq.tscol, sq.extent.End.Unix()) + return fmt.Sprintf(sq.query, r) +} + var parser = sqlparser.New(parsing.New(nil, lexer, lopts)) // parse parses the Time Range Query func parse(statement string) (*timeseries.TimeRangeQuery, *timeseries.RequestOptions, bool, error) { - trq := ×eries.TimeRangeQuery{Statement: statement} + trq := ×eries.TimeRangeQuery{Statement: statement, ParsedQuery: &sqlquery{}} ro := ×eries.RequestOptions{} rs, err := parser.Run(sqlparser.NewRunContext(trq, ro), parser, trq.Statement) @@ -73,6 +86,12 @@ func parse(statement string) (*timeseries.TimeRangeQuery, *timeseries.RequestOpt if err != nil { return nil, nil, false, err } + // Set modifiable parsedquery + pq := trq.ParsedQuery.(*sqlquery) + pq.query = trq.Statement[:pq.rangeStart] + "%s" + trq.Statement[pq.rangeEnd:] + pq.extent = &trq.Extent + // Modify upstream request with new values + trq.Statement = pq.String() //var t *token.Token /* if t, err = parseGroupByTokens(results, trq, ro); err != nil { @@ -99,7 +118,6 @@ func parseTSColumn(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *tim if !ok { return sqlparser.ErrMissingTimeseries } - fmt.Println(tokens) // Parse the first "statement" in the select portion. Need to check for datatype, column, alias stmnt := tokens[0] var col, alias string @@ -131,6 +149,7 @@ func parseTSColumn(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *tim // Set in trq/ro if foundCol { trq.TimestampDefinition.Name = col + trq.ParsedQuery.(*sqlquery).tscol = col } if foundAlias { ro.BaseTimestampFieldName = alias @@ -161,6 +180,7 @@ func parseFromTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *t // the query could not be parsed properly OR if there's some WHERE statement including the tscol that is unrelated // to the timerange being requested. func parseWhereTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro *timeseries.RequestOptions) (error, bool) { + pq := trq.ParsedQuery.(*sqlquery) whereTokens, ok := rs.GetResultsCollection("whereTokens") if !ok { return sqlparser.ErrMissingTimeseries, false @@ -179,6 +199,7 @@ func parseWhereTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro * for idxStatement := 0; idxStatement < len(tokens); idxStatement++ { statement := tokens[idxStatement] if !hasStart && isTimeseriesStartStatement(statement, tscol) { + pq.rangeStart = statement[0].Pos start, _, err = lsql.TokenToTime(statement[2]) if err != nil { return sqlparser.ErrNoLowerBound, false @@ -188,10 +209,12 @@ func parseWhereTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro * } hasStart = true } else if !hasStart && isTimeseriesBetweenStatement(statement, tscol) { + pq.rangeStart = statement[0].Pos start, _, err = lsql.TokenToTime(statement[2]) if err != nil { return sqlparser.ErrNoLowerBound, false } + pq.rangeEnd = statement[4].Pos + len(statement[4].Val) end, _, err = lsql.TokenToTime(statement[4]) if err != nil { return sqlparser.ErrNoUpperBound, false @@ -200,6 +223,7 @@ func parseWhereTokens(rs *parsing.RunState, trq *timeseries.TimeRangeQuery, ro * hasEnd = true break } else if hasStart && isTimeseriesEndStatement(statement, tscol) { + pq.rangeEnd = statement[2].Pos + len(statement[2].Val) end, _, err = lsql.TokenToTime(statement[2]) if err != nil { return sqlparser.ErrNoUpperBound, false diff --git a/pkg/backends/mysql/parsing_test.go b/pkg/backends/mysql/parsing_test.go index d9e3c4181..63f3b1b1f 100644 --- a/pkg/backends/mysql/parsing_test.go +++ b/pkg/backends/mysql/parsing_test.go @@ -41,11 +41,10 @@ func TestParseRawQuery(t *testing.T) { } for i, test := range tests { t.Run(strconv.Itoa(i), func(t *testing.T) { - trq, _, _, err := parse(test.query) + _, _, _, err := parse(test.query) if err != test.err { t.Errorf("got '%v' expected '%v'", err, test.err) } - t.Logf("%+v\n", trq) }) } } diff --git a/pkg/backends/mysql/url.go b/pkg/backends/mysql/url.go new file mode 100644 index 000000000..889e4fc19 --- /dev/null +++ b/pkg/backends/mysql/url.go @@ -0,0 +1,25 @@ +package mysql + +import ( + "net/http" + + "github.com/trickstercache/trickster/v2/pkg/proxy/params" + "github.com/trickstercache/trickster/v2/pkg/timeseries" +) + +func (c *Client) SetExtent(r *http.Request, trq *timeseries.TimeRangeQuery, extent *timeseries.Extent) { + pq, ok := trq.ParsedQuery.(*sqlquery) + if !ok { + return + } + pq.extent = extent + + trq.Extent = *pq.extent + trq.Statement = pq.String() + qi := r.URL.Query() + qi.Set("query", trq.Statement) + trq.TemplateURL.RawQuery = qi.Encode() + v, _, _ := params.GetRequestValues(r) + v.Set("query", pq.String()) + params.SetRequestValues(r, v) +} diff --git a/pkg/backends/mysql/url_test.go b/pkg/backends/mysql/url_test.go new file mode 100644 index 000000000..fc54ae796 --- /dev/null +++ b/pkg/backends/mysql/url_test.go @@ -0,0 +1,53 @@ +package mysql + +import ( + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/trickstercache/trickster/v2/cmd/trickster/config" + "github.com/trickstercache/trickster/v2/pkg/timeseries" +) + +func TestSetExtent(t *testing.T) { + conf, _, err := config.Load("trickster", "test", + []string{"-origin-url", "none:9090", + "-provider", "mysql", + "-log-level", "debug"}) + if err != nil { + t.Fatalf("Could not load configuration: %s", err.Error()) + } + + o := conf.Backends["default"] + backendClient, err := NewClient("test", o, nil, nil, nil, nil) + if err != nil { + t.Error(err) + } + client := backendClient.(*Client) + r := httptest.NewRequest(http.MethodGet, "/", nil) + q := url.Values{"query": {tq03}} + r.URL.RawQuery = q.Encode() + + trq, _, _, err := client.ParseTimeRangeQuery(r) + if err != nil { + t.Fatal(err) + } + orig := trq.Statement + + start := time.Now().Add(time.Duration(-6) * time.Hour) + end := time.Now() + e := timeseries.Extent{Start: start, End: end} + + client.SetExtent(r, trq, &e) + now := trq.Statement + + // Check everything to make sure values change appropriately + if trq.Extent != e { + t.Error(trq.Extent, e) + } + if orig == now { + t.Error(orig, now) + } +}