[perf] sqltracestore use new schema.
- Update the version of CockroachDB used on the bots.
- Moved to using the pgx/v4 client library.
-
Change-Id: Iea9c24201b2599fc24c6f8bba26b7863256badab
Reviewed-on: https://skia-review.googlesource.com/c/buildbot/+/306733
Reviewed-by: Joe Gregorio <jcgregorio@google.com>
Commit-Queue: Joe Gregorio <jcgregorio@google.com>
diff --git a/go.mod b/go.mod
index f11cbe9..333dc4e 100644
--- a/go.mod
+++ b/go.mod
@@ -91,6 +91,8 @@
github.com/hashicorp/golang-lru v0.5.4
github.com/huandu/xstrings v1.3.2 // indirect
github.com/imdario/mergo v0.3.10 // indirect
+ github.com/jackc/fake v0.0.0-20150926172116-812a484cc733 // indirect
+ github.com/jackc/pgx/v4 v4.8.1
github.com/jcgregorio/logger v0.1.2
github.com/jcgregorio/slog v0.0.0-20190423190439-e6f2d537f900
github.com/julienschmidt/httprouter v1.3.0 // indirect
diff --git a/go.sum b/go.sum
index edc684e..38a7756 100644
--- a/go.sum
+++ b/go.sum
@@ -119,6 +119,8 @@
github.com/Jeffail/gabs/v2 v2.5.0/go.mod h1:xCn81vdHKxFUuWWAaD5jCTQDNPBMh5pPs9IJ+NcziBI=
github.com/Jeffail/gabs/v2 v2.5.1 h1:ANfZYjpMlfTTKebycu4X1AgkVWumFVDYQl7JwOr4mDk=
github.com/Jeffail/gabs/v2 v2.5.1/go.mod h1:xCn81vdHKxFUuWWAaD5jCTQDNPBMh5pPs9IJ+NcziBI=
+github.com/Kount/pq-timeouts v1.0.0 h1:6a23dhwmQ2PukftCWm56T4RPJ4zc2iE9y5E42TMAl6E=
+github.com/Kount/pq-timeouts v1.0.0/go.mod h1:Y7rNVWI9KiI3xj1QxBmOSB12Eyv9g5Gjego8KFpV5PY=
github.com/Masterminds/goutils v1.1.0 h1:zukEsf/1JZwCMgHiK3GZftabmxiCw4apj3a28RPBiVg=
github.com/Masterminds/goutils v1.1.0/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
@@ -250,6 +252,7 @@
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58/go.mod h1:EOBUe0h4xcZ5GoxqC5SDxFQ8gwyZPKQoEzownBlhI80=
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
+github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I=
github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
github.com/cockroachdb/cockroach-go v0.0.0-20190925194419-606b3d062051 h1:eApuUG8W2EtBVwxqLlY2wgoqDYOg3WvIHGvW4fUbbow=
github.com/cockroachdb/cockroach-go v0.0.0-20190925194419-606b3d062051/go.mod h1:XGLbWH/ujMcbPbhZq52Nv6UrCghb1yGn//133kEsvDk=
@@ -353,6 +356,9 @@
github.com/gocql/gocql v0.0.0-20190301043612-f6df8288f9b4/go.mod h1:4Fw1eo5iaEhDUs8XyuhSVCVy52Jq3L+/3GJgYkwc+/0=
github.com/godbus/dbus v0.0.0-20181101234600-2ff6f7ffd60f h1:zlOR3rOlPAVvtfuxGKoghCmop5B0TRyu/ZieziZuGiM=
github.com/godbus/dbus v0.0.0-20181101234600-2ff6f7ffd60f/go.mod h1:/YcGZj5zSblfDWMMoOzV4fas9FZnQYTkDnsGvmh2Grw=
+github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
+github.com/gofrs/uuid v3.3.0+incompatible h1:8K4tyRfvU1CYPgJsveYFQMhpFd/wXNM7iK6rR7UHz84=
+github.com/gofrs/uuid v3.3.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
@@ -551,11 +557,18 @@
github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8=
github.com/jackc/chunkreader/v2 v2.0.1/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
+github.com/jackc/fake v0.0.0-20150926172116-812a484cc733 h1:vr3AYkKovP8uR8AvSGGUK1IDqRa5lAAvEkZG1LKaCRc=
+github.com/jackc/fake v0.0.0-20150926172116-812a484cc733/go.mod h1:WrMFNQdiFJ80sQsxDoMokWK1W5TQtxBFNpzWTD84ibQ=
github.com/jackc/pgconn v0.0.0-20190420214824-7e0022ef6ba3/go.mod h1:jkELnwuX+w9qN5YIfX0fl88Ehu4XC3keFuOJJk9pcnA=
github.com/jackc/pgconn v0.0.0-20190824142844-760dd75542eb/go.mod h1:lLjNuW/+OfW9/pnVKPazfWOgNfH2aPem8YQ7ilXGvJE=
github.com/jackc/pgconn v0.0.0-20190831204454-2fabfa3c18b7/go.mod h1:ZJKsE/KZfsUgOEh9hBm+xYTstcNHg7UPMVJqRfQxq4s=
github.com/jackc/pgconn v1.3.2 h1:9UIGICxEAW70RQDGilGwsCG63NCcm5amjuBQCFzrmsw=
github.com/jackc/pgconn v1.3.2/go.mod h1:LvCquS3HbBKwgl7KbX9KyqEIumJAbm1UMcTvGaIf3bM=
+github.com/jackc/pgconn v1.4.0/go.mod h1:Y2O3ZDF0q4mMacyWV3AstPJpeHXWGEetiFttmq5lahk=
+github.com/jackc/pgconn v1.5.0/go.mod h1:QeD3lBfpTFe8WUnPZWN5KY/mB8FGMIYRdd8P8Jr0fAI=
+github.com/jackc/pgconn v1.5.1-0.20200601181101-fa742c524853/go.mod h1:QeD3lBfpTFe8WUnPZWN5KY/mB8FGMIYRdd8P8Jr0fAI=
+github.com/jackc/pgconn v1.6.4 h1:S7T6cx5o2OqmxdHaXLH1ZeD1SbI8jBznyYE9Ec0RCQ8=
+github.com/jackc/pgconn v1.6.4/go.mod h1:w2pne1C2tZgP+TvjqLpOigGzNqjBgQW9dUw/4Chex78=
github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=
github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8=
github.com/jackc/pgmock v0.0.0-20190831213851-13a1b77aafa2 h1:JVX6jT/XfzNqIjye4717ITLaNwV9mWbJx0dLCpcRzdA=
@@ -570,14 +583,36 @@
github.com/jackc/pgproto3/v2 v2.0.0-rc3.0.20190831210041-4c03ce451f29/go.mod h1:ryONWYqW6dqSg1Lw6vXNMXoBJhpzvWKnT95C46ckYeM=
github.com/jackc/pgproto3/v2 v2.0.1 h1:Rdjp4NFjwHnEslx2b66FfCI2S0LhO4itac3hXz6WX9M=
github.com/jackc/pgproto3/v2 v2.0.1/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
+github.com/jackc/pgproto3/v2 v2.0.2 h1:q1Hsy66zh4vuNsajBUF2PNqfAMMfxU5mk594lPE9vjY=
+github.com/jackc/pgproto3/v2 v2.0.2/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA=
+github.com/jackc/pgservicefile v0.0.0-20200307190119-3430c5407db8/go.mod h1:vsD4gTJCa9TptPL8sPkXrLZ+hDuNrZCnj29CQpr4X1E=
+github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b h1:C8S2+VttkHFdOOCXJe+YGfa4vHYwlt4Zx+IVXQ97jYg=
+github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b/go.mod h1:vsD4gTJCa9TptPL8sPkXrLZ+hDuNrZCnj29CQpr4X1E=
github.com/jackc/pgtype v0.0.0-20190421001408-4ed0de4755e0/go.mod h1:hdSHsc1V01CGwFsrv11mJRHWJ6aifDLfdV3aVjFF0zg=
github.com/jackc/pgtype v0.0.0-20190824184912-ab885b375b90/go.mod h1:KcahbBH1nCMSo2DXpzsoWOAfFkdEtEJpPbVLq8eE+mc=
+github.com/jackc/pgtype v0.0.0-20190828014616-a8802b16cc59 h1:xOamcCJ9MFJTxR5bvw3ZXmiP8evQMohdt2VJ57C0W8Q=
github.com/jackc/pgtype v0.0.0-20190828014616-a8802b16cc59/go.mod h1:MWlu30kVJrUS8lot6TQqcg7mtthZ9T0EoIBFiJcmcyw=
+github.com/jackc/pgtype v1.2.0/go.mod h1:5m2OfMh1wTK7x+Fk952IDmI4nw3nPrvtQdM0ZT4WpC0=
+github.com/jackc/pgtype v1.3.1-0.20200510190516-8cd94a14c75a/go.mod h1:vaogEUkALtxZMCH411K+tKzNpwzCKU+AnPzBKZ+I+Po=
+github.com/jackc/pgtype v1.3.1-0.20200606141011-f6355165a91c/go.mod h1:cvk9Bgu/VzJ9/lxTO5R5sf80p0DiucVtN7ZxvaC4GmQ=
+github.com/jackc/pgtype v1.4.2 h1:t+6LWm5eWPLX1H5Se702JSBcirq6uWa4jiG4wV1rAWY=
+github.com/jackc/pgtype v1.4.2/go.mod h1:JCULISAZBFGrHaOXIIFiyfzW5VY0GRitRr8NeJsrdig=
+github.com/jackc/pgx v3.6.2+incompatible h1:2zP5OD7kiyR3xzRYMhOcXVvkDZsImVXfj+yIyTQf3/o=
+github.com/jackc/pgx v3.6.2+incompatible/go.mod h1:0ZGrqGqkRlliWnWB4zKnWtjbSWbGkVEFm4TeybAXq+I=
github.com/jackc/pgx/v4 v4.0.0-20190420224344-cc3461e65d96/go.mod h1:mdxmSJJuR08CZQyj1PVQBHy9XOp5p8/SHH6a0psbY9Y=
github.com/jackc/pgx/v4 v4.0.0-20190421002000-1b8f0016e912/go.mod h1:no/Y67Jkk/9WuGR0JG/JseM9irFbnEPbuWV2EELPNuM=
+github.com/jackc/pgx/v4 v4.0.0-pre1.0.20190824185557-6972a5742186 h1:ZQM8qLT/E/CGD6XX0E6q9FAwxJYmWpJufzmLMaFuzgQ=
github.com/jackc/pgx/v4 v4.0.0-pre1.0.20190824185557-6972a5742186/go.mod h1:X+GQnOEnf1dqHGpw7JmHqHc1NxDoalibchSk9/RWuDc=
+github.com/jackc/pgx/v4 v4.5.0/go.mod h1:EpAKPLdnTorwmPUUsqrPxy5fphV18j9q3wrfRXgo+kA=
+github.com/jackc/pgx/v4 v4.6.1-0.20200510190926-94ba730bb1e9/go.mod h1:t3/cdRQl6fOLDxqtlyhe9UWgfIi9R8+8v8GKV5TRA/o=
+github.com/jackc/pgx/v4 v4.6.1-0.20200606145419-4e5062306904/go.mod h1:ZDaNWkt9sW1JMiNn0kdYBaLelIhw7Pg4qd+Vk6tw7Hg=
+github.com/jackc/pgx/v4 v4.8.1 h1:SUbCLP2pXvf/Sr/25KsuI4aTxiFYIvpfk4l6aTSdyCw=
+github.com/jackc/pgx/v4 v4.8.1/go.mod h1:4HOLxrl8wToZJReD04/yB20GDwf4KBYETvlHciCnwW0=
github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
+github.com/jackc/puddle v1.1.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
+github.com/jackc/puddle v1.1.1 h1:PJAw7H/9hoWC4Kf3J8iNmL1SwA6E8vfsLqBiL+F6CtI=
+github.com/jackc/puddle v1.1.1/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
github.com/jcgregorio/logger v0.1.2 h1:kHiF857oOObzlUer5ANZ95U08A7k2INjivnss4IyMCg=
@@ -656,9 +691,14 @@
github.com/maruel/ut v1.0.1/go.mod h1:RV8PwPD9dd2KFlnlCc/DB2JVvkXmyaalfc5xvmSrRSs=
github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
+github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
+github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
github.com/mattn/go-runewidth v0.0.3 h1:a+kO+98RDGEfo6asOGMmpodZq4FNtnGP54yps8BzLR4=
github.com/mattn/go-runewidth v0.0.3/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
@@ -839,13 +879,16 @@
github.com/russross/blackfriday/v2 v2.0.1 h1:lPqVAte+HuHNfhJ/0LC98ESWRz8afy9tM/0RK8m9o+Q=
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
+github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
github.com/sergi/go-diff v1.0.0 h1:Kpca3qRNrduNnOQeazBd0ysaKrUJiIuISHxogkT9RPQ=
github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
+github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24 h1:pntxY8Ary0t43dCZ5dqY4YTJCObLY1kIXl0uzMv+7DE=
github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
+github.com/shopspring/decimal v0.0.0-20200227202807-02e2044944cc/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
github.com/shurcooL/sanitized_anchor_name v1.0.0 h1:PdmoCO6wvbs+7yrJyMORt4/BmY5IYyJwS/kOiWx8mHo=
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
@@ -1048,7 +1091,10 @@
go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
+go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
+go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU=
+go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -1064,6 +1110,7 @@
golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190911031432-227b76d455e7/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550 h1:ObdrDkeb4kJdCP557AjRjq69pTHfNouLtWZG7j9rPN8=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20191117063200-497ca9f6d64f/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -1071,6 +1118,7 @@
golang.org/x/crypto v0.0.0-20200220183623-bac4c82f6975/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073 h1:xMPOj6Pz6UipU1wXLkrtqpHbR0AVFnyPEQq/wRWz9lM=
golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200420201142-3c4aac89819a h1:y6sBfNd1b9Wy08a6K1Z1DZc4aXABUN5TKjkYhz7UKmo=
golang.org/x/crypto v0.0.0-20200420201142-3c4aac89819a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200429183012-4b2356b1ed79 h1:IaQbIIB2X/Mp/DKctl6ROxz1KyMlKp4uyvL6+kQ7C88=
@@ -1241,6 +1289,7 @@
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191022100944-742c48ecaeb7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1251,6 +1300,7 @@
golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200113162924-86b910548bc1 h1:gZpLHxUX5BdYLA08Lj4YCJNN/jk7KtquiArPoeX0WvA=
golang.org/x/sys v0.0.0-20200113162924-86b910548bc1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200122134326-e047566fdf82 h1:ywK/j/KkyTHcdyYSZNXGjMwgmDSfjglYZ3vStQ/gSCU=
golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200124204421-9fbb57f87de9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1345,6 +1395,8 @@
golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191010075000-0337d82405ff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191113191852-77e3bb0ad9e7/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191115202509-3a792d9c32b2/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
diff --git a/go/cipd/asset_versions_gen.go b/go/cipd/asset_versions_gen.go
index 969c18a..2df0aa6 100644
--- a/go/cipd/asset_versions_gen.go
+++ b/go/cipd/asset_versions_gen.go
@@ -86,7 +86,7 @@
"skia/bots/cockroachdb": {
Path: "cockroachdb",
Name: "skia/bots/cockroachdb",
- Version: "version:2",
+ Version: "version:3",
},
"skia/bots/gcloud_linux": {
Path: "gcloud_linux",
diff --git a/infra/bots/assets/cockroachdb/VERSION b/infra/bots/assets/cockroachdb/VERSION
index 0cfbf08..00750ed 100644
--- a/infra/bots/assets/cockroachdb/VERSION
+++ b/infra/bots/assets/cockroachdb/VERSION
@@ -1 +1 @@
-2
+3
diff --git a/infra/bots/tasks.json b/infra/bots/tasks.json
index f03d0ca..997d52e 100755
--- a/infra/bots/tasks.json
+++ b/infra/bots/tasks.json
@@ -940,7 +940,7 @@
{
"name": "skia/bots/cockroachdb",
"path": "cockroachdb",
- "version": "version:2"
+ "version": "version:3"
},
{
"name": "skia/bots/gcloud_linux",
@@ -1367,7 +1367,7 @@
{
"name": "skia/bots/cockroachdb",
"path": "cockroachdb",
- "version": "version:2"
+ "version": "version:3"
},
{
"name": "skia/bots/gcloud_linux",
diff --git a/perf/COCKROACHDB.md b/perf/COCKROACHDB.md
index 0234b69..6e8cb3a 100644
--- a/perf/COCKROACHDB.md
+++ b/perf/COCKROACHDB.md
@@ -52,7 +52,7 @@
Now port-forwarding the database port for the cockroachdb instance, for example:
- kubectl port-forward perf-skia-cockroachdb-0 26257
+ kubectl port-forward perf-cockroachdb-0 26257
Then run the migrations:
diff --git a/perf/go/alerts/sqlalertstore/sqlalertstore.go b/perf/go/alerts/sqlalertstore/sqlalertstore.go
index 2f256cd..6b96941 100644
--- a/perf/go/alerts/sqlalertstore/sqlalertstore.go
+++ b/perf/go/alerts/sqlalertstore/sqlalertstore.go
@@ -5,11 +5,11 @@
import (
"context"
- "database/sql"
"encoding/json"
"fmt"
"time"
+ "github.com/jackc/pgx/v4/pgxpool"
"go.skia.org/infra/go/skerr"
"go.skia.org/infra/perf/go/alerts"
perfsql "go.skia.org/infra/perf/go/sql"
@@ -27,25 +27,21 @@
listAllAlerts
)
-// statements allows looking up raw SQL statements by their statement id.
-type statements map[statement]string
-
-// statementsByDialect holds all the raw SQL statemens used per Dialect of SQL.
-var statementsByDialect = map[perfsql.Dialect]statements{
- perfsql.CockroachDBDialect: {
- insertAlert: `
+// statements holds all the raw SQL statements used.
+var statements = map[statement]string{
+ insertAlert: `
INSERT INTO
Alerts (alert, last_modified)
VALUES
($1, $2)
`,
- updateAlert: `
+ updateAlert: `
UPSERT INTO
Alerts (id, alert, config_state, last_modified)
VALUES
($1, $2, $3, $4)
`,
- deleteAlert: `
+ deleteAlert: `
UPDATE
Alerts
SET
@@ -54,7 +50,7 @@
WHERE
id=$2
`,
- listActiveAlerts: `
+ listActiveAlerts: `
SELECT
id, alert
FROM
@@ -62,37 +58,28 @@
WHERE
config_state=0 -- alerts.ACTIVE
`,
- listAllAlerts: `
+ listAllAlerts: `
SELECT
id, alert
FROM
ALERTS
`,
- },
}
// SQLAlertStore implements the alerts.Store interface.
type SQLAlertStore struct {
- // preparedStatements are all the prepared SQL statements.
- preparedStatements map[statement]*sql.Stmt
+ // db is the database interface.
+ db *pgxpool.Pool
}
// New returns a new *SQLAlertStore.
//
// We presume all migrations have been run against db before this function is
// called.
-func New(db *sql.DB, dialect perfsql.Dialect) (*SQLAlertStore, error) {
- preparedStatements := map[statement]*sql.Stmt{}
- for key, statement := range statementsByDialect[dialect] {
- prepared, err := db.Prepare(statement)
- if err != nil {
- return nil, skerr.Wrapf(err, "Failed to prepare statment %v %q", key, statement)
- }
- preparedStatements[key] = prepared
- }
+func New(db *pgxpool.Pool, dialect perfsql.Dialect) (*SQLAlertStore, error) {
return &SQLAlertStore{
- preparedStatements: preparedStatements,
+ db: db,
}, nil
}
@@ -106,11 +93,11 @@
now := time.Now().Unix()
if cfg.ID == alerts.BadAlertID {
// Not a valid ID, so this should be an insert, not an update.
- if _, err := s.preparedStatements[insertAlert].ExecContext(ctx, string(b), now); err != nil {
+ if _, err := s.db.Exec(ctx, statements[insertAlert], string(b), now); err != nil {
return skerr.Wrapf(err, "Failed to insert alert")
}
} else {
- if _, err := s.preparedStatements[updateAlert].ExecContext(ctx, cfg.ID, string(b), cfg.StateToInt(), now); err != nil {
+ if _, err := s.db.Exec(ctx, statements[updateAlert], cfg.ID, string(b), cfg.StateToInt(), now); err != nil {
return skerr.Wrapf(err, "Failed to update Alert with ID=%d", cfg.ID)
}
}
@@ -120,7 +107,7 @@
// Delete implements the alerts.Store interface.
func (s *SQLAlertStore) Delete(ctx context.Context, id int) error {
now := time.Now().Unix()
- if _, err := s.preparedStatements[deleteAlert].ExecContext(ctx, now, id); err != nil {
+ if _, err := s.db.Exec(ctx, statements[deleteAlert], now, id); err != nil {
return skerr.Wrapf(err, "Failed to mark Alert as deleted with ID=%d", id)
}
return nil
@@ -132,7 +119,7 @@
if includeDeleted {
stmt = listAllAlerts
}
- rows, err := s.preparedStatements[stmt].QueryContext(ctx)
+ rows, err := s.db.Query(ctx, statements[stmt])
if err != nil {
return nil, err
}
diff --git a/perf/go/builders/builders.go b/perf/go/builders/builders.go
index f0e06c8..bb2c5fc 100644
--- a/perf/go/builders/builders.go
+++ b/perf/go/builders/builders.go
@@ -11,6 +11,8 @@
"cloud.google.com/go/bigtable"
"cloud.google.com/go/datastore"
+ "github.com/jackc/pgx/v4/pgxpool"
+ _ "github.com/jackc/pgx/v4/stdlib" // pgx Go sql
"go.skia.org/infra/go/auth"
"go.skia.org/infra/go/ds"
"go.skia.org/infra/go/skerr"
@@ -41,7 +43,7 @@
// newCockroachDBFromConfig opens an existing CockroachDB database with all
// migrations applied.
-func newCockroachDBFromConfig(instanceConfig *config.InstanceConfig) (*sql.DB, error) {
+func newCockroachDBFromConfig(ctx context.Context, instanceConfig *config.InstanceConfig) (*pgxpool.Pool, error) {
// Note that the migrationsConnection is different from the sql.Open
// connection string since migrations know about CockroachDB, but we use the
// Postgres driver for the database/sql connection since there's no native
@@ -50,7 +52,7 @@
// uses.
migrationsConnection := strings.Replace(instanceConfig.DataStoreConfig.ConnectionString, "postgresql://", "cockroach://", 1)
- db, err := sql.Open("postgres", instanceConfig.DataStoreConfig.ConnectionString)
+ db, err := sql.Open("pgx", instanceConfig.DataStoreConfig.ConnectionString)
if err != nil {
return nil, skerr.Wrap(err)
}
@@ -62,8 +64,12 @@
if err != nil {
return nil, skerr.Wrap(err)
}
+ if err := db.Close(); err != nil {
+ return nil, skerr.Wrap(err)
+ }
sklog.Infof("Finished applying migrations.")
- return db, nil
+
+ return pgxpool.Connect(ctx, instanceConfig.DataStoreConfig.ConnectionString)
}
// NewPerfGitFromConfig return a new perfgit.Git for the given instanceConfig.
@@ -96,7 +102,7 @@
sklog.Infof("Constructing perfgit with dialect: %q and connection_string: %q", dialect, instanceConfig.DataStoreConfig.ConnectionString)
// Now create the appropriate db.
- db, err := newCockroachDBFromConfig(instanceConfig)
+ db, err := newCockroachDBFromConfig(ctx, instanceConfig)
if err != nil {
return nil, skerr.Wrap(err)
}
@@ -124,11 +130,11 @@
}
return traceStore, nil
case config.CockroachDBDataStoreType:
- db, err := newCockroachDBFromConfig(instanceConfig)
+ db, err := newCockroachDBFromConfig(ctx, instanceConfig)
if err != nil {
return nil, skerr.Wrap(err)
}
- return sqltracestore.New(db, perfsql.CockroachDBDialect, instanceConfig.DataStoreConfig)
+ return sqltracestore.New(db, instanceConfig.DataStoreConfig)
}
return nil, skerr.Fmt("Unknown datastore type: %q", instanceConfig.DataStoreConfig.DataStoreType)
}
@@ -161,7 +167,7 @@
}
return dsalertstore.New(), nil
case config.CockroachDBDataStoreType:
- db, err := newCockroachDBFromConfig(instanceConfig)
+ db, err := newCockroachDBFromConfig(ctx, instanceConfig)
if err != nil {
return nil, skerr.Wrap(err)
}
@@ -190,11 +196,11 @@
}
return dsregressionstore.NewRegressionStoreDS(lookup), nil
case config.CockroachDBDataStoreType:
- db, err := newCockroachDBFromConfig(instanceConfig)
+ db, err := newCockroachDBFromConfig(ctx, instanceConfig)
if err != nil {
return nil, skerr.Wrap(err)
}
- return sqlregressionstore.New(db, perfsql.CockroachDBDialect)
+ return sqlregressionstore.New(db)
}
return nil, skerr.Fmt("Unknown datastore type: %q", instanceConfig.DataStoreConfig.DataStoreType)
}
@@ -210,7 +216,7 @@
return dsshortcutstore.New(), nil
case config.CockroachDBDataStoreType:
- db, err := newCockroachDBFromConfig(instanceConfig)
+ db, err := newCockroachDBFromConfig(ctx, instanceConfig)
if err != nil {
return nil, skerr.Wrap(err)
}
diff --git a/perf/go/builders/builders_test.go b/perf/go/builders/builders_test.go
index c3ceac7..f718f7d 100644
--- a/perf/go/builders/builders_test.go
+++ b/perf/go/builders/builders_test.go
@@ -220,7 +220,7 @@
func TestNewPerfGitFromConfig_ErrIfConnectionStringNotSet(t *testing.T) {
unittest.LargeTest(t)
- ctx, _, _, _, _, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, _, _, _, _, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
instanceConfig.DataStoreConfig.DataStoreType = config.GCPDataStoreType
@@ -232,7 +232,7 @@
func TestNewPerfGitFromConfig_GCP_ErrorsOnNonPostgresConnectionString(t *testing.T) {
unittest.LargeTest(t)
- ctx, _, _, _, _, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, _, _, _, _, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
instanceConfig.DataStoreConfig.DataStoreType = config.GCPDataStoreType
@@ -252,11 +252,11 @@
func TestNewPerfGitFromConfig_GCP_SuccessIfConnectionStringIsCockroachDB(t *testing.T) {
unittest.LargeTest(t)
- ctx, _, _, hashes, _, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, _, _, hashes, _, instanceConfig, dbName, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
instanceConfig.DataStoreConfig.DataStoreType = config.GCPDataStoreType
- instanceConfig.DataStoreConfig.ConnectionString = fmt.Sprintf("postgresql://root@%s/%s?sslmode=disable", perfsql.GetCockroachDBEmulatorHost(), gittest.CockroachDatabaseName)
+ instanceConfig.DataStoreConfig.ConnectionString = fmt.Sprintf("postgresql://root@%s/%s?sslmode=disable", perfsql.GetCockroachDBEmulatorHost(), dbName)
g, err := NewPerfGitFromConfig(ctx, false, instanceConfig)
require.NoError(t, err)
@@ -268,7 +268,7 @@
func TestNewPerfGitFromConfig_CockroachDB_Success(t *testing.T) {
unittest.LargeTest(t)
- ctx, _, _, hashes, _, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, _, _, hashes, _, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
instanceConfig.DataStoreConfig.DataStoreType = config.CockroachDBDataStoreType
diff --git a/perf/go/cid/cid_test.go b/perf/go/cid/cid_test.go
index 0fa0b81..4ed62a1 100644
--- a/perf/go/cid/cid_test.go
+++ b/perf/go/cid/cid_test.go
@@ -121,7 +121,7 @@
func TestCommitIDLookup_Success(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, hashes, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, hashes, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
@@ -161,7 +161,7 @@
func TestCommitIDLookup_ErrOnBadCommit(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
diff --git a/perf/go/config/config.go b/perf/go/config/config.go
index eabead4..7c9a936 100644
--- a/perf/go/config/config.go
+++ b/perf/go/config/config.go
@@ -36,24 +36,6 @@
CockroachDBDataStoreType DataStoreType = "cockroachdb"
)
-// Cache is the configuration for the LRU cache used by the storage client.
-type Cache struct {
- // MemcachedServers is a list of memcached server names and addresses, e.g.
- // ['memcached-0:11211', 'memcached-1:11211', 'memcached-2:11211']
- //
- // If this list is empty then the server will fall back to using an
- // in-memory LRU cache per instance.
- MemcachedServers []string `json:"memcached_servers"`
-
- // Namespace is the string to add to each key to avoid conflicts with more
- // than one application or application instance using the same memcached
- // server.
- Namespace string `json:"namespace"`
-
- // Size is the size of the LRU cache to use if memcached isn't being used.
- Size int `json:"size"`
-}
-
// DataStoreConfig is the configuration for how Perf stores data.
type DataStoreConfig struct {
// DataStoreType determines what type of datastore to build. This value will
@@ -97,9 +79,6 @@
// regressions, and shortcuts should use. This value is only used for 'gcp'
// datastore types.
Namespace string `json:"namespace"`
-
- // Cache is the configuration for the LRU cache used by the storage client.
- Cache Cache `json:"cache"`
}
// SourceType determines what type of file.Source to build from a SourceConfig.
@@ -271,9 +250,10 @@
// IngestFlags are the command-line flags for the ingestion process.
type IngestFlags struct {
- InstanceConfigFile string
- PromPort string
- Local bool
+ InstanceConfigFile string
+ PromPort string
+ Local bool
+ NumParallelIngesters int
}
// Register the flags in the given FlagSet.
@@ -281,6 +261,7 @@
fs.StringVar(&flags.InstanceConfigFile, "config_filename", "", "Instance config file. Must be supplied.")
fs.StringVar(&flags.PromPort, "prom_port", ":20000", "Metrics service address (e.g., ':20000')")
fs.BoolVar(&flags.Local, "local", false, "True if running locally and not in production.")
+ fs.IntVar(&flags.NumParallelIngesters, "num_parallel_ingesters", 10, "The number of parallel Go routines to have ingesting.")
}
// InstanceConfig contains all the info needed by btts.BigTableTraceStore.
diff --git a/perf/go/dataframe/async_test.go b/perf/go/dataframe/async_test.go
index 7c8b98f..87aff52 100644
--- a/perf/go/dataframe/async_test.go
+++ b/perf/go/dataframe/async_test.go
@@ -148,7 +148,7 @@
func TestGetSkps_Success(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
@@ -170,7 +170,7 @@
func TestGetSkps_SuccessIfFileChangeMarkerNotSet(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
@@ -192,7 +192,7 @@
func TestGetSkps_ErrOnBadCommitNumber(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
diff --git a/perf/go/dataframe/dataframe_test.go b/perf/go/dataframe/dataframe_test.go
index 27f4139..0b63e13 100644
--- a/perf/go/dataframe/dataframe_test.go
+++ b/perf/go/dataframe/dataframe_test.go
@@ -134,7 +134,7 @@
func TestFromTimeRange_Success(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
@@ -156,7 +156,7 @@
func TestFromTimeRange_EmptySlicesIfNothingInTimeRange(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
diff --git a/perf/go/dfbuilder/dfbuilder_test.go b/perf/go/dfbuilder/dfbuilder_test.go
index 625ad46..64aa91d 100644
--- a/perf/go/dfbuilder/dfbuilder_test.go
+++ b/perf/go/dfbuilder/dfbuilder_test.go
@@ -2,6 +2,8 @@
import (
"context"
+ "fmt"
+ "math/rand"
"net/url"
"testing"
"time"
@@ -34,15 +36,14 @@
}
)
-const CockroachDatabaseName = "dfbuilder"
-
func TestBuildTraceMapper(t *testing.T) {
unittest.LargeTest(t)
- db, cleanup := sqltest.NewCockroachDBForTests(t, CockroachDatabaseName, sqltest.ApplyMigrations)
+ dbName := fmt.Sprintf("dfbuilder%d", rand.Uint32())
+ db, cleanup := sqltest.NewCockroachDBForTests(t, dbName, sqltest.ApplyMigrations)
defer cleanup()
- store, err := sqltracestore.New(db, perfsql.CockroachDBDialect, cfg.DataStoreConfig)
+ store, err := sqltracestore.New(db, cfg.DataStoreConfig)
require.NoError(t, err)
tileMap := buildTileMapOffsetToIndex([]types.CommitNumber{0, 1, 255, 256, 257}, store)
@@ -75,14 +76,14 @@
unittest.LargeTest(t)
ctx := context.Background()
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
instanceConfig.DataStoreConfig.TileSize = 6
- store, err := sqltracestore.New(db, perfsql.CockroachDBDialect, instanceConfig.DataStoreConfig)
+ store, err := sqltracestore.New(db, instanceConfig.DataStoreConfig)
require.NoError(t, err)
builder := NewDataFrameBuilderFromTraceStore(g, store)
@@ -113,7 +114,7 @@
now := gittest.StartTime.Add(7 * time.Minute)
df, err := builder.NewFromQueryAndRange(ctx, now.Add(-7*time.Minute), now.Add(time.Second), q, false, nil)
- assert.NoError(t, err)
+ require.NoError(t, err)
assert.Len(t, df.TraceSet, 2)
assert.Len(t, df.Header, 8)
assert.Len(t, df.TraceSet[",arch=x86,config=8888,"], 8)
@@ -206,7 +207,7 @@
func TestFromIndexRange_Success(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
@@ -232,7 +233,7 @@
func TestFromIndexRange_EmptySliceOnBadCommitNumber(t *testing.T) {
unittest.LargeTest(t)
- ctx, db, _, _, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
defer cleanup()
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
diff --git a/perf/go/git/git.go b/perf/go/git/git.go
index 96764aa..6c3767d 100644
--- a/perf/go/git/git.go
+++ b/perf/go/git/git.go
@@ -8,7 +8,6 @@
import (
"bufio"
"context"
- "database/sql"
"io"
"os"
"os/exec"
@@ -17,6 +16,8 @@
"strings"
"time"
+ "github.com/jackc/pgx/v4"
+ "github.com/jackc/pgx/v4/pgxpool"
"go.skia.org/infra/go/auth"
"go.skia.org/infra/go/gitauth"
"go.skia.org/infra/go/metrics2"
@@ -49,13 +50,10 @@
}
)
-// statements allows looking up raw SQL statements by their statement id.
-type statements map[statement]string
+// statements holds all the raw SQL statemens used per Dialect of SQL.
+var statements = map[statement]string{
-// statementsByDialect holds all the raw SQL statemens used per Dialect of SQL.
-var statementsByDialect = map[perfsql.Dialect]statements{
- perfsql.CockroachDBDialect: {
- getMostRecentGitHashAndCommitNumber: `
+ getMostRecentGitHashAndCommitNumber: `
SELECT
git_hash, commit_number
FROM
@@ -65,7 +63,7 @@
LIMIT
1
`,
- insert: `
+ insert: `
INSERT INTO
Commits (commit_number, git_hash, commit_time, author, subject)
VALUES
@@ -73,14 +71,14 @@
ON CONFLICT
DO NOTHING
`,
- getCommitNumberFromGitHash: `
+ getCommitNumberFromGitHash: `
SELECT
commit_number
FROM
Commits
WHERE
git_hash=$1`,
- getCommitNumberFromTime: `
+ getCommitNumberFromTime: `
SELECT
commit_number
FROM
@@ -92,7 +90,7 @@
LIMIT
1
`,
- getCommitsFromTimeRange: `
+ getCommitsFromTimeRange: `
SELECT
commit_number, git_hash, commit_time, author, subject
FROM
@@ -103,7 +101,7 @@
ORDER BY
commit_number ASC
`,
- getCommitsFromCommitNumberRange: `
+ getCommitsFromCommitNumberRange: `
SELECT
commit_number, git_hash, commit_time, author, subject
FROM
@@ -114,7 +112,7 @@
ORDER BY
commit_number ASC
`,
- getCommitFromCommitNumber: `
+ getCommitFromCommitNumber: `
SELECT
commit_number, git_hash, commit_time, author, subject
FROM
@@ -122,7 +120,7 @@
WHERE
commit_number = $1
`,
- getHashFromCommitNumber: `
+ getHashFromCommitNumber: `
SELECT
git_hash
FROM
@@ -130,7 +128,6 @@
WHERE
commit_number=$1
`,
- },
}
// Git implements the minimal functionality Perf needs to interface to Git.
@@ -146,8 +143,7 @@
instanceConfig *config.InstanceConfig
- // preparedStatements are all the prepared SQL statements.
- preparedStatements map[statement]*sql.Stmt
+ db *pgxpool.Pool
// Metrics
updateCalled metrics2.Counter
@@ -164,7 +160,7 @@
//
// The instance created does not poll by default, callers need to call
// StartBackgroundPolling().
-func New(ctx context.Context, local bool, db *sql.DB, dialect perfsql.Dialect, instanceConfig *config.InstanceConfig) (*Git, error) {
+func New(ctx context.Context, local bool, db *pgxpool.Pool, dialect perfsql.Dialect, instanceConfig *config.InstanceConfig) (*Git, error) {
// Do git authentication if required.
if instanceConfig.GitRepoConfig.GitAuthType == config.GitAuthGerrit {
sklog.Info("Authenticating to Gerrit.")
@@ -199,28 +195,17 @@
}
}
- // Prepare all our SQL statements.
- sklog.Infof("Preparing statements.")
- preparedStatements := map[statement]*sql.Stmt{}
- for key, statement := range statementsByDialect[dialect] {
- prepared, err := db.Prepare(statement)
- if err != nil {
- return nil, skerr.Wrapf(err, "Failed to prepare statment %v %q", key, statement)
- }
- preparedStatements[key] = prepared
- }
-
ret := &Git{
- gitFullPath: gitFullPath,
- preparedStatements: preparedStatements,
- instanceConfig: instanceConfig,
- updateCalled: metrics2.GetCounter("perf_git_update_called"),
- commitNumberFromGitHashCalled: metrics2.GetCounter("perf_git_commit_number_from_githash_called"),
- commitNumberFromTimeCalled: metrics2.GetCounter("perf_git_commit_number_from_time_called"),
- commitSliceFromTimeRangeCalled: metrics2.GetCounter("perf_git_commits_slice_from_time_range_called"),
- commitSliceFromCommitNumberRangeCalled: metrics2.GetCounter("perf_git_commits_slice_from_commit_number_range_called"),
- commitFromCommitNumberCalled: metrics2.GetCounter("perf_git_commit_from_commit_number_called"),
- gitHashFromCommitNumberCalled: metrics2.GetCounter("perf_git_githash_from_commit_number_called"),
+ gitFullPath: gitFullPath,
+ db: db,
+ instanceConfig: instanceConfig,
+ updateCalled: metrics2.GetCounter("perf_git_update_called"),
+ commitNumberFromGitHashCalled: metrics2.GetCounter("perf_git_commit_number_from_githash_called"),
+ commitNumberFromTimeCalled: metrics2.GetCounter("perf_git_commit_number_from_time_called"),
+ commitSliceFromTimeRangeCalled: metrics2.GetCounter("perf_git_commits_slice_from_time_range_called"),
+ commitSliceFromCommitNumberRangeCalled: metrics2.GetCounter("perf_git_commits_slice_from_commit_number_range_called"),
+ commitFromCommitNumberCalled: metrics2.GetCounter("perf_git_commit_from_commit_number_called"),
+ gitHashFromCommitNumberCalled: metrics2.GetCounter("perf_git_githash_from_commit_number_called"),
commitNumbersWhenFileChangesInCommitNumberRangeCalled: metrics2.GetCounter("perf_git_commit_numbers_when_file_changes_in_commit_number_range_called"),
}
@@ -362,7 +347,7 @@
if err != nil {
// If the Commits table is empty then start populating it from the very
// first commit to the repo.
- if err == sql.ErrNoRows {
+ if err == pgx.ErrNoRows {
cmd = exec.CommandContext(ctx, g.gitFullPath, "rev-list", "HEAD", `--pretty=%aN <%aE>%n%s%n%ct`, "--reverse")
nextCommitNumber = types.CommitNumber(0)
} else {
@@ -386,7 +371,7 @@
total := 0
err = parseGitRevLogStream(stdout, func(p Commit) error {
// Add p to the database starting at nextCommitNumber.
- _, err := g.preparedStatements[insert].ExecContext(ctx, nextCommitNumber, p.GitHash, p.Timestamp, p.Author, p.Subject)
+ _, err := g.db.Exec(ctx, statements[insert], nextCommitNumber, p.GitHash, p.Timestamp, p.Author, p.Subject)
if err != nil {
return skerr.Wrapf(err, "Failed to insert commit %q into database.", p.GitHash)
}
@@ -415,7 +400,7 @@
func (g *Git) getMostRecentCommit(ctx context.Context) (string, types.CommitNumber, error) {
var gitHash string
var commitNumber types.CommitNumber
- if err := g.preparedStatements[getMostRecentGitHashAndCommitNumber].QueryRowContext(ctx).Scan(&gitHash, &commitNumber); err != nil {
+ if err := g.db.QueryRow(ctx, statements[getMostRecentGitHashAndCommitNumber]).Scan(&gitHash, &commitNumber); err != nil {
// Don't wrap the err, we need to see if it's sql.ErrNoRows.
return "", types.BadCommitNumber, err
}
@@ -426,7 +411,7 @@
func (g *Git) CommitNumberFromGitHash(ctx context.Context, githash string) (types.CommitNumber, error) {
g.commitNumberFromGitHashCalled.Inc(1)
ret := types.BadCommitNumber
- if err := g.preparedStatements[getCommitNumberFromGitHash].QueryRowContext(ctx, githash).Scan(&ret); err != nil {
+ if err := g.db.QueryRow(ctx, statements[getCommitNumberFromGitHash], githash).Scan(&ret); err != nil {
return ret, skerr.Wrapf(err, "Failed get for hash: %q", githash)
}
return ret, nil
@@ -445,7 +430,7 @@
_, mostRecentCommitNumber, err := g.getMostRecentCommit(ctx)
return mostRecentCommitNumber, err
}
- if err := g.preparedStatements[getCommitNumberFromTime].QueryRowContext(ctx, t.Unix()).Scan(&ret); err != nil {
+ if err := g.db.QueryRow(ctx, statements[getCommitNumberFromTime], t.Unix()).Scan(&ret); err != nil {
return ret, skerr.Wrapf(err, "Failed get for time: %q", t)
}
return ret, nil
@@ -455,7 +440,7 @@
// [begin, end), i.e inclusive of begin and exclusive of end.
func (g *Git) CommitSliceFromTimeRange(ctx context.Context, begin, end time.Time) ([]Commit, error) {
g.commitSliceFromTimeRangeCalled.Inc(1)
- rows, err := g.preparedStatements[getCommitsFromTimeRange].QueryContext(ctx, begin.Unix(), end.Unix())
+ rows, err := g.db.Query(ctx, statements[getCommitsFromTimeRange], begin.Unix(), end.Unix())
if err != nil {
return nil, skerr.Wrapf(err, "Failed to query for commit slice in range %s-%s", begin, end)
}
@@ -474,7 +459,7 @@
// [begin, end], i.e inclusive of both begin and end.
func (g *Git) CommitSliceFromCommitNumberRange(ctx context.Context, begin, end types.CommitNumber) ([]Commit, error) {
g.commitSliceFromCommitNumberRangeCalled.Inc(1)
- rows, err := g.preparedStatements[getCommitsFromCommitNumberRange].QueryContext(ctx, begin, end)
+ rows, err := g.db.Query(ctx, statements[getCommitsFromCommitNumberRange], begin, end)
if err != nil {
return nil, skerr.Wrapf(err, "Failed to query for commit slice in range %v-%v", begin, end)
}
@@ -493,7 +478,7 @@
func (g *Git) CommitFromCommitNumber(ctx context.Context, commitNumber types.CommitNumber) (Commit, error) {
g.commitFromCommitNumberCalled.Inc(1)
var c Commit
- if err := g.preparedStatements[getCommitFromCommitNumber].QueryRowContext(ctx, commitNumber).Scan(&c.CommitNumber, &c.GitHash, &c.Timestamp, &c.Author, &c.Subject); err != nil {
+ if err := g.db.QueryRow(ctx, statements[getCommitFromCommitNumber], commitNumber).Scan(&c.CommitNumber, &c.GitHash, &c.Timestamp, &c.Author, &c.Subject); err != nil {
return Commit{}, skerr.Wrapf(err, "Failed to read row at %v", commitNumber)
}
return c, nil
@@ -503,7 +488,7 @@
func (g *Git) GitHashFromCommitNumber(ctx context.Context, commitNumber types.CommitNumber) (string, error) {
g.gitHashFromCommitNumberCalled.Inc(1)
var ret string
- if err := g.preparedStatements[getHashFromCommitNumber].QueryRowContext(ctx, commitNumber).Scan(&ret); err != nil {
+ if err := g.db.QueryRow(ctx, statements[getHashFromCommitNumber], commitNumber).Scan(&ret); err != nil {
return "", skerr.Wrapf(err, "Failed to find git hash for commit number: %v", commitNumber)
}
return ret, nil
diff --git a/perf/go/git/git_test.go b/perf/go/git/git_test.go
index 007b7a3..cdf2693 100644
--- a/perf/go/git/git_test.go
+++ b/perf/go/git/git_test.go
@@ -24,7 +24,7 @@
for name, subTest := range subTests {
t.Run(name, func(t *testing.T) {
- ctx, db, gb, hashes, dialect, instanceConfig, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, gb, hashes, dialect, instanceConfig, _, cleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
g, err := New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
diff --git a/perf/go/git/gittest/gittest.go b/perf/go/git/gittest/gittest.go
index 4555abe..57a652f 100644
--- a/perf/go/git/gittest/gittest.go
+++ b/perf/go/git/gittest/gittest.go
@@ -3,13 +3,15 @@
import (
"context"
- "database/sql"
+ "fmt"
"io/ioutil"
+ "math/rand"
"os"
"path/filepath"
"testing"
"time"
+ "github.com/jackc/pgx/v4/pgxpool"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.skia.org/infra/go/git/testutils"
@@ -37,7 +39,7 @@
// The repo is populated with 8 commits, one minute apart, starting at StartTime.
//
// The hashes for each commit are going to be random and so are returned also.
-func NewForTest(t *testing.T, dialect perfsql.Dialect) (context.Context, *sql.DB, *testutils.GitBuilder, []string, perfsql.Dialect, *config.InstanceConfig, CleanupFunc) {
+func NewForTest(t *testing.T, dialect perfsql.Dialect) (context.Context, *pgxpool.Pool, *testutils.GitBuilder, []string, perfsql.Dialect, *config.InstanceConfig, string, CleanupFunc) {
ctx, cancel := context.WithCancel(context.Background())
// Create a git repo for testing purposes.
@@ -52,8 +54,9 @@
hashes = append(hashes, gb.CommitGenAt(ctx, "bar.txt", StartTime.Add(6*time.Minute)))
hashes = append(hashes, gb.CommitGenAt(ctx, "foo.txt", StartTime.Add(7*time.Minute)))
+ dbName := fmt.Sprintf("git%d", rand.Uint64())
// Init our sql database.
- db, sqlCleanup := sqltest.NewCockroachDBForTests(t, CockroachDatabaseName, sqltest.ApplyMigrations)
+ db, sqlCleanup := sqltest.NewCockroachDBForTests(t, dbName, sqltest.ApplyMigrations)
// Get tmp dir to use for repo checkout.
tmpDir, err := ioutil.TempDir("", "git")
@@ -74,5 +77,5 @@
Dir: filepath.Join(tmpDir, "checkout"),
},
}
- return ctx, db, gb, hashes, dialect, instanceConfig, clean
+ return ctx, db, gb, hashes, dialect, instanceConfig, dbName, clean
}
diff --git a/perf/go/ingest/process/process.go b/perf/go/ingest/process/process.go
index 0085a5d..ec5f3c6 100644
--- a/perf/go/ingest/process/process.go
+++ b/perf/go/ingest/process/process.go
@@ -3,6 +3,7 @@
import (
"context"
+ "sync"
"time"
"cloud.google.com/go/pubsub"
@@ -12,16 +13,18 @@
"go.skia.org/infra/go/query"
"go.skia.org/infra/go/skerr"
"go.skia.org/infra/go/sklog"
+
"go.skia.org/infra/perf/go/builders"
"go.skia.org/infra/perf/go/config"
+ "go.skia.org/infra/perf/go/file"
+ "go.skia.org/infra/perf/go/git"
"go.skia.org/infra/perf/go/ingest/parser"
"go.skia.org/infra/perf/go/ingestevents"
+ "go.skia.org/infra/perf/go/tracestore"
"google.golang.org/api/option"
)
-const (
- gitRefreshDuration = time.Minute
-)
+const writeRetries = 10
// sendPubSubEvent sends the unencoded params and paramset found in a single
// ingested file to the PubSub topic specified in the selected Perf instances
@@ -55,12 +58,8 @@
return skerr.Wrap(err)
}
-// Start a single go routine to process incoming ingestion files and write
-// the data they contain to a trace store.
-//
-// Except for file.Sources of type "dir" this function should never return
-// except on error.
-func Start(ctx context.Context, local bool, instanceConfig *config.InstanceConfig) error {
+// worker ingests files that arrive on the given 'ch' channel.
+func worker(ctx context.Context, wg *sync.WaitGroup, g *git.Git, store tracestore.TraceStore, ch <-chan file.File, pubSubClient *pubsub.Client, instanceConfig *config.InstanceConfig) {
// Metrics.
filesReceived := metrics2.GetCounter("perfserver_ingest_files_received")
failedToParse := metrics2.GetCounter("perfserver_ingest_failed_to_parse")
@@ -70,47 +69,13 @@
successfulWrite := metrics2.GetCounter("perfserver_ingest_successful_write")
successfulWriteCount := metrics2.GetCounter("perfserver_ingest_num_points_written")
- var pubSubClient *pubsub.Client
- if instanceConfig.IngestionConfig.FileIngestionTopicName != "" {
- ts, err := auth.NewDefaultTokenSource(local, pubsub.ScopePubSub)
- if err != nil {
- sklog.Fatalf("Failed to create TokenSource: %s", err)
- }
-
- pubSubClient, err = pubsub.NewClient(ctx, instanceConfig.IngestionConfig.SourceConfig.Project, option.WithTokenSource(ts))
- if err != nil {
- sklog.Fatal(err)
- }
- }
-
- // New file.Source.
- source, err := builders.NewSourceFromConfig(ctx, instanceConfig, local)
- ch, err := source.Start(ctx)
- if err != nil {
- return skerr.Wrap(err)
- }
-
// New Parser.
p := parser.New(instanceConfig)
- // New TraceStore.
- store, err := builders.NewTraceStoreFromConfig(ctx, local, instanceConfig)
- if err != nil {
- return skerr.Wrap(err)
- }
-
- // New perfgit.Git.
- sklog.Infof("Cloning repo %q into %q", instanceConfig.GitRepoConfig.URL, instanceConfig.GitRepoConfig.Dir)
- g, err := builders.NewPerfGitFromConfig(ctx, local, instanceConfig)
- if err != nil {
- return skerr.Wrap(err)
- }
- g.StartBackgroundPolling(ctx, gitRefreshDuration)
-
- sklog.Info("Waiting on files to process.")
for f := range ch {
if err := ctx.Err(); err != nil {
- return skerr.Wrap(err)
+ sklog.Error(err)
+ break
}
sklog.Infof("Ingest received: %v", f)
filesReceived.Inc(1)
@@ -152,10 +117,20 @@
}
sklog.Info("WriteTraces")
- // Write data to the trace store.
- if err := store.WriteTraces(commitNumber, params, values, ps, f.Name, time.Now()); err != nil {
- failedToWrite.Inc(1)
- sklog.Error("Failed to write %v: %s", f, err)
+ const retries = writeRetries
+ i := 0
+ for {
+ // Write data to the trace store.
+ err := store.WriteTraces(commitNumber, params, values, ps, f.Name, time.Now())
+ if err == nil {
+ break
+ }
+ i++
+ if i > retries {
+ failedToWrite.Inc(1)
+ sklog.Errorf("Failed to write after %d retries %q: %s", retries, f.Name, err)
+ continue
+ }
}
successfulWrite.Inc(1)
successfulWriteCount.Inc(int64(len(params)))
@@ -166,6 +141,61 @@
sklog.Info("FileIngestionTopicName pubsub message sent.")
}
}
+ wg.Done()
+}
+
+// Start a single go routine to process incoming ingestion files and write
+// the data they contain to a trace store.
+//
+// Except for file.Sources of type "dir" this function should never return
+// except on error.
+func Start(ctx context.Context, local bool, numParallelIngesters int, instanceConfig *config.InstanceConfig) error {
+
+ var pubSubClient *pubsub.Client
+ if instanceConfig.IngestionConfig.FileIngestionTopicName != "" {
+ ts, err := auth.NewDefaultTokenSource(local, pubsub.ScopePubSub)
+ if err != nil {
+ sklog.Fatalf("Failed to create TokenSource: %s", err)
+ }
+
+ pubSubClient, err = pubsub.NewClient(ctx, instanceConfig.IngestionConfig.SourceConfig.Project, option.WithTokenSource(ts))
+ if err != nil {
+ sklog.Fatal(err)
+ }
+ }
+
+ // New file.Source.
+ source, err := builders.NewSourceFromConfig(ctx, instanceConfig, local)
+ ch, err := source.Start(ctx)
+ if err != nil {
+ return skerr.Wrap(err)
+ }
+
+ // New TraceStore.
+ store, err := builders.NewTraceStoreFromConfig(ctx, local, instanceConfig)
+ if err != nil {
+ return skerr.Wrap(err)
+ }
+
+ // New perfgit.Git.
+ sklog.Infof("Cloning repo %q into %q", instanceConfig.GitRepoConfig.URL, instanceConfig.GitRepoConfig.Dir)
+ g, err := builders.NewPerfGitFromConfig(ctx, local, instanceConfig)
+ if err != nil {
+ return skerr.Wrap(err)
+ }
+ // Polling isn't needed because we call update on the repo if we find a git hash we don't recognize.
+ // g.StartBackgroundPolling(ctx, gitRefreshDuration)
+
+ sklog.Info("Waiting on files to process.")
+
+ var wg sync.WaitGroup
+
+ for i := 0; i < numParallelIngesters; i++ {
+ wg.Add(1)
+ go worker(ctx, &wg, g, store, ch, pubSubClient, instanceConfig)
+ }
+ wg.Wait()
+
sklog.Infof("Exited while waiting on files. Should only happen on source_type=dir.")
return nil
}
diff --git a/perf/go/ingest/process/process_test.go b/perf/go/ingest/process/process_test.go
index 07c9144..545ff0f 100644
--- a/perf/go/ingest/process/process_test.go
+++ b/perf/go/ingest/process/process_test.go
@@ -87,7 +87,7 @@
},
}
- err = Start(context.Background(), true, &instanceConfig)
+ err = Start(context.Background(), true, 1, &instanceConfig)
require.NoError(t, err)
// The integration data set has 9 good files, 1 file with a bad commit, and
// 1 malformed JSON file.
diff --git a/perf/go/perfserver/cmd/ingest.go b/perf/go/perfserver/cmd/ingest.go
index 66929d4..1149f4c 100644
--- a/perf/go/perfserver/cmd/ingest.go
+++ b/perf/go/perfserver/cmd/ingest.go
@@ -43,7 +43,7 @@
sklog.Infof("Flags: --%s=%v", f.Name, f.Value)
})
- return process.Start(context.Background(), ingestFlags.Local, instanceConfig)
+ return process.Start(context.Background(), ingestFlags.Local, ingestFlags.NumParallelIngesters, instanceConfig)
},
}
diff --git a/perf/go/regression/dfiter_test.go b/perf/go/regression/dfiter_test.go
index 2fa2e9a..4802152 100644
--- a/perf/go/regression/dfiter_test.go
+++ b/perf/go/regression/dfiter_test.go
@@ -56,7 +56,7 @@
Shards: 8,
}
- store, err := sqltracestore.New(db, perfsql.CockroachDBDialect, cfg)
+ store, err := sqltracestore.New(db, cfg)
require.NoError(t, err)
// Add some points to the first and second tile.
@@ -79,7 +79,7 @@
}, "gs://foo.json", time.Now()) // Time is irrelevent.
assert.NoError(t, err)
- ctx, db, _, _, dialect, instanceConfig, gitCleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
+ ctx, db, _, _, dialect, instanceConfig, _, gitCleanup := gittest.NewForTest(t, perfsql.CockroachDBDialect)
instanceConfig.DataStoreConfig.TileSize = testTileSize
g, err := perfgit.New(ctx, true, db, dialect, instanceConfig)
require.NoError(t, err)
diff --git a/perf/go/regression/sqlregressionstore/sqlregressionstore.go b/perf/go/regression/sqlregressionstore/sqlregressionstore.go
index 48be0fa..1820686 100644
--- a/perf/go/regression/sqlregressionstore/sqlregressionstore.go
+++ b/perf/go/regression/sqlregressionstore/sqlregressionstore.go
@@ -6,9 +6,9 @@
import (
"context"
- "database/sql"
"encoding/json"
+ "github.com/jackc/pgx/v4/pgxpool"
"go.skia.org/infra/go/skerr"
"go.skia.org/infra/go/sklog"
"go.skia.org/infra/perf/go/alerts"
@@ -16,7 +16,6 @@
"go.skia.org/infra/perf/go/clustering2"
"go.skia.org/infra/perf/go/dataframe"
"go.skia.org/infra/perf/go/regression"
- perfsql "go.skia.org/infra/perf/go/sql"
"go.skia.org/infra/perf/go/types"
)
@@ -30,19 +29,15 @@
readRange
)
-// statements allows looking up raw SQL statements by their statement id.
-type statements map[statement]string
-
// statementsByDialect holds all the raw SQL statemens used per Dialect of SQL.
-var statementsByDialect = map[perfsql.Dialect]statements{
- perfsql.CockroachDBDialect: {
- write: `
+var statements = map[statement]string{
+ write: `
UPSERT INTO
Regressions (commit_number, alert_id, regression)
VALUES
($1, $2, $3)
`,
- read: `
+ read: `
SELECT
regression
FROM
@@ -50,7 +45,7 @@
WHERE
commit_number=$1 AND
alert_id=$2`,
- readRange: `
+ readRange: `
SELECT
commit_number, alert_id, regression
FROM
@@ -59,42 +54,28 @@
commit_number >= $1
AND commit_number <= $2
`,
- },
}
// SQLRegressionStore implements the regression.Store interface.
type SQLRegressionStore struct {
// db is the underlying database.
- db *sql.DB
-
- // preparedStatements are all the prepared SQL statements.
- preparedStatements map[statement]*sql.Stmt
+ db *pgxpool.Pool
}
// New returns a new *SQLRegressionStore.
//
// We presume all migrations have been run against db before this function is
// called.
-func New(db *sql.DB, dialect perfsql.Dialect) (*SQLRegressionStore, error) {
- preparedStatements := map[statement]*sql.Stmt{}
- for key, statement := range statementsByDialect[dialect] {
- prepared, err := db.Prepare(statement)
- if err != nil {
- return nil, skerr.Wrapf(err, "Failed to prepare statment %v %q", key, statement)
- }
- preparedStatements[key] = prepared
- }
-
+func New(db *pgxpool.Pool) (*SQLRegressionStore, error) {
return &SQLRegressionStore{
- db: db,
- preparedStatements: preparedStatements,
+ db: db,
}, nil
}
// Range implements the regression.Store interface.
func (s *SQLRegressionStore) Range(ctx context.Context, begin, end types.CommitNumber) (map[types.CommitNumber]*regression.AllRegressionsForCommit, error) {
ret := map[types.CommitNumber]*regression.AllRegressionsForCommit{}
- rows, err := s.preparedStatements[readRange].QueryContext(ctx, begin, end)
+ rows, err := s.db.Query(ctx, statements[readRange], begin, end)
if err != nil {
return nil, skerr.Wrapf(err, "Failed to read regressions in range: %d %d", begin, end)
}
@@ -190,7 +171,7 @@
if err != nil {
return skerr.Wrapf(err, "Failed to serialize regression for alertID: %d commitNumber=%d", alertID, commitNumber)
}
- if _, err := s.preparedStatements[write].ExecContext(ctx, commitNumber, alertID, string(b)); err != nil {
+ if _, err := s.db.Exec(ctx, statements[write], commitNumber, alertID, string(b)); err != nil {
return skerr.Wrapf(err, "Failed to write regression for alertID: %d commitNumber=%d", alertID, commitNumber)
}
return nil
@@ -204,7 +185,7 @@
return nil, skerr.Fmt("Failed to convert alertIDString %q to an int.", alertIDString)
}
var jsonString string
- if err := s.preparedStatements[read].QueryRowContext(ctx, commitNumber, alertID).Scan(&jsonString); err != nil {
+ if err := s.db.QueryRow(ctx, statements[read], commitNumber, alertID).Scan(&jsonString); err != nil {
return nil, skerr.Wrapf(err, "Failed to read regression for alertID: %d commitNumber=%d", alertID, commitNumber)
}
r := regression.NewRegression()
@@ -228,7 +209,7 @@
}
// Do everything in a transaction so we don't have any lost updates.
- tx, err := s.db.BeginTx(ctx, nil)
+ tx, err := s.db.Begin(ctx)
if err != nil {
return skerr.Wrapf(err, "Can't start transaction")
}
@@ -238,13 +219,13 @@
// Read the regression from the database. If any part of that fails then
// just use the default regression we've already constructed.
var jsonString string
- if err := tx.StmtContext(ctx, s.preparedStatements[read]).QueryRowContext(ctx, commitNumber, alertID).Scan(&jsonString); err == nil {
+ if err := tx.QueryRow(ctx, statements[read], commitNumber, alertID).Scan(&jsonString); err == nil {
if err := json.Unmarshal([]byte(jsonString), r); err != nil {
sklog.Warningf("Failed to deserialize the JSON Regression: %s", err)
}
} else {
if mustExist {
- if err := tx.Rollback(); err != nil {
+ if err := tx.Rollback(ctx); err != nil {
sklog.Errorf("Failed on rollback: %s", err)
}
return skerr.Wrapf(err, "Regression doesn't exist.")
@@ -255,19 +236,19 @@
b, err := json.Marshal(r)
if err != nil {
- if err := tx.Rollback(); err != nil {
+ if err := tx.Rollback(ctx); err != nil {
sklog.Errorf("Failed on rollback: %s", err)
}
return skerr.Wrapf(err, "Failed to serialize regression for alertID: %d commitNumber=%d", alertID, commitNumber)
}
- if _, err := tx.StmtContext(ctx, s.preparedStatements[write]).ExecContext(ctx, commitNumber, alertID, string(b)); err != nil {
- if err := tx.Rollback(); err != nil {
+ if _, err := tx.Exec(ctx, statements[write], commitNumber, alertID, string(b)); err != nil {
+ if err := tx.Rollback(ctx); err != nil {
sklog.Errorf("Failed on rollback: %s", err)
}
return skerr.Wrapf(err, "Failed to write regression for alertID: %d commitNumber=%d", alertID, commitNumber)
}
- return tx.Commit()
+ return tx.Commit(ctx)
}
// Confirm that SQLRegressionStore implements regression.Store.
diff --git a/perf/go/regression/sqlregressionstore/sqlregressionstore_test.go b/perf/go/regression/sqlregressionstore/sqlregressionstore_test.go
index 73b4f6d..1246441 100644
--- a/perf/go/regression/sqlregressionstore/sqlregressionstore_test.go
+++ b/perf/go/regression/sqlregressionstore/sqlregressionstore_test.go
@@ -9,7 +9,6 @@
"go.skia.org/infra/go/testutils/unittest"
"go.skia.org/infra/perf/go/regression"
"go.skia.org/infra/perf/go/regression/regressiontest"
- perfsql "go.skia.org/infra/perf/go/sql"
"go.skia.org/infra/perf/go/sql/sqltest"
)
@@ -24,7 +23,7 @@
// actual errors.
defer cleanup()
- store, err := New(db, perfsql.CockroachDBDialect)
+ store, err := New(db)
require.NoError(t, err)
subTest(t, store)
})
@@ -38,7 +37,7 @@
// actual errors.
defer cleanup()
- store, err := New(db, perfsql.CockroachDBDialect)
+ store, err := New(db)
require.NoError(t, err)
subTest(t, store)
})
diff --git a/perf/go/shortcut/sqlshortcutstore/sqlshortcutstore.go b/perf/go/shortcut/sqlshortcutstore/sqlshortcutstore.go
index 333e428..7b536a3 100644
--- a/perf/go/shortcut/sqlshortcutstore/sqlshortcutstore.go
+++ b/perf/go/shortcut/sqlshortcutstore/sqlshortcutstore.go
@@ -5,10 +5,10 @@
import (
"context"
- "database/sql"
"encoding/json"
"io"
+ "github.com/jackc/pgx/v4/pgxpool"
"go.skia.org/infra/go/query"
"go.skia.org/infra/go/skerr"
"go.skia.org/infra/go/sklog"
@@ -26,20 +26,16 @@
getAllShortcuts
)
-// statements allows looking up raw SQL statements by their statement id.
-type statements map[statement]string
-
-// statementsByDialect holds all the raw SQL statemens used per Dialect of SQL.
-var statementsByDialect = map[perfsql.Dialect]statements{
- perfsql.CockroachDBDialect: {
- insertShortcut: `
+// statements holds all the raw SQL statemens.
+var statements = map[statement]string{
+ insertShortcut: `
INSERT INTO
Shortcuts (id, trace_ids)
VALUES
($1, $2)
ON CONFLICT
DO NOTHING`,
- getShortcut: `
+ getShortcut: `
SELECT
(trace_ids)
FROM
@@ -47,38 +43,27 @@
WHERE
id=$1
`,
- getAllShortcuts: `
+ getAllShortcuts: `
SELECT
(trace_ids)
FROM
Shortcuts
`,
- },
}
// SQLShortcutStore implements the shortcut.Store interface using an SQL
// database.
type SQLShortcutStore struct {
- // preparedStatements are all the prepared SQL statements.
- preparedStatements map[statement]*sql.Stmt
+ db *pgxpool.Pool
}
// New returns a new *SQLShortcutStore.
//
// We presume all migrations have been run against db before this function is
// called.
-func New(db *sql.DB, dialect perfsql.Dialect) (*SQLShortcutStore, error) {
- preparedStatements := map[statement]*sql.Stmt{}
- for key, statement := range statementsByDialect[dialect] {
- prepared, err := db.Prepare(statement)
- if err != nil {
- return nil, skerr.Wrapf(err, "Failed to prepare statment %v %q", key, statement)
- }
- preparedStatements[key] = prepared
- }
-
+func New(db *pgxpool.Pool, dialect perfsql.Dialect) (*SQLShortcutStore, error) {
return &SQLShortcutStore{
- preparedStatements: preparedStatements,
+ db: db,
}, nil
}
@@ -103,7 +88,7 @@
if err != nil {
return "", err
}
- if _, err := s.preparedStatements[insertShortcut].ExecContext(ctx, id, string(b)); err != nil {
+ if _, err := s.db.Exec(ctx, statements[insertShortcut], id, string(b)); err != nil {
return "", skerr.Wrap(err)
}
return id, nil
@@ -112,7 +97,7 @@
// Get implements the shortcut.Store interface.
func (s *SQLShortcutStore) Get(ctx context.Context, id string) (*shortcut.Shortcut, error) {
var encoded string
- if err := s.preparedStatements[getShortcut].QueryRowContext(ctx, id).Scan(&encoded); err != nil {
+ if err := s.db.QueryRow(ctx, statements[getShortcut], id).Scan(&encoded); err != nil {
return nil, skerr.Wrapf(err, "Failed to load shortcuts.")
}
var sc shortcut.Shortcut
@@ -127,7 +112,7 @@
func (s *SQLShortcutStore) GetAll(ctx context.Context) (<-chan *shortcut.Shortcut, error) {
ret := make(chan *shortcut.Shortcut)
- rows, err := s.preparedStatements[getAllShortcuts].QueryContext(ctx)
+ rows, err := s.db.Query(ctx, statements[getAllShortcuts])
if err != nil {
return ret, skerr.Wrapf(err, "Failed to query for all shortcuts.")
}
@@ -151,5 +136,4 @@
}()
return ret, nil
-
}
diff --git a/perf/go/sql/sqltest/sqltest.go b/perf/go/sql/sqltest/sqltest.go
index 50602ba..46db2a5 100644
--- a/perf/go/sql/sqltest/sqltest.go
+++ b/perf/go/sql/sqltest/sqltest.go
@@ -1,10 +1,12 @@
package sqltest
import (
+ "context"
"database/sql"
"fmt"
"testing"
+ "github.com/jackc/pgx/v4/pgxpool"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
perfsql "go.skia.org/infra/perf/go/sql"
@@ -37,7 +39,7 @@
// test.
//
// If migrations to are be applied then set applyMigrations to true.
-func NewCockroachDBForTests(t *testing.T, databaseName string, applyMigrations ApplyMigrationsOption) (*sql.DB, Cleanup) {
+func NewCockroachDBForTests(t *testing.T, databaseName string, applyMigrations ApplyMigrationsOption) (*pgxpool.Pool, Cleanup) {
// Note that the migrationsConnection is different from the sql.Open
// connection string since migrations know about CockroachDB, but we use the
// Postgres driver for the database/sql connection since there's no native
@@ -63,12 +65,17 @@
require.NoError(t, err)
}
+ ctx := context.Background()
+ conn, err := pgxpool.Connect(ctx, connectionString)
+ require.NoError(t, err)
+
cleanup := func() {
// Don't bother applying migrations.Down since we aren't testing
// migrations here and it just slows down the tests.
_, err = db.Exec(fmt.Sprintf("DROP DATABASE %s CASCADE;", databaseName))
assert.NoError(t, err)
+ require.NoError(t, db.Close())
+ conn.Close()
}
-
- return db, cleanup
+ return conn, cleanup
}
diff --git a/perf/go/tracestore/sqltracestore/sqltracestore.go b/perf/go/tracestore/sqltracestore/sqltracestore.go
index b7b3325..39452c9 100644
--- a/perf/go/tracestore/sqltracestore/sqltracestore.go
+++ b/perf/go/tracestore/sqltracestore/sqltracestore.go
@@ -5,119 +5,89 @@
traces into an SQL database.
Each trace name, which is a structured key (See /infra/go/query) of the form
-,key1=value1,key2=value2,..., is stored in the TraceIds table so we can use the
-much shorter 64 bit trace_id in other tables.
+,key1=value1,key2=value2,..., is stored in the TraceNames table so we can use the
+much shorter 128 bit md5 hash in trace_id in other tables. The value of the
+trace name is parsed into a paramtools.Params and stored in the 'params' column
+with an inverted index, which enables all the queries that Perf supports.
- TraceIDs (
- trace_id INTEGER PRIMARY KEY,
- trace_name TEXT UNIQUE NOT NULL
- )
+ CREATE TABLE IF NOT EXISTS TraceNames (
+ -- md5(trace_name)
+ trace_id BYTES PRIMARY KEY,
+ -- The params that make up the trace_id, {"arch=x86", "config=8888"}.
+ params JSONB NOT NULL,
+ INVERTED INDEX (params)
+ );
Similarly we store the name of every source file that has been ingested in the
SourceFiles table so we can use the shorter 64 bit source_file_id in other
tables.
- SourceFiles (
- source_file_id INTEGER PRIMARY KEY,
- source_file TEXT UNIQUE NOT NULL
- )
+ SourceFiles (
+ source_file_id INTEGER PRIMARY KEY,
+ source_file TEXT UNIQUE NOT NULL
+ )
+ CREATE TABLE IF NOT EXISTS SourceFiles (
+ source_file_id INT PRIMARY KEY DEFAULT unique_rowid(),
+ source_file STRING UNIQUE NOT NULL
+ );
-We store the values of each trace in the TraceValues table, and use the trace_id
+We store the values of each trace in the TraceValues2 table, and use the trace_id
and the commit_number as the primary key. We also store not only the value but
the id of the source file that the value came from.
- TraceValues (
- trace_id INTEGER,
- commit_number INTEGER,
- val REAL,
- source_file_id INTEGER,
- PRIMARY KEY (trace_id, commit_number)
- )
+ CREATE TABLE IF NOT EXISTS TraceValues2 (
+ -- md5(trace_name) from TraceNames.
+ trace_id BYTES,
+ -- A types.CommitNumber.
+ commit_number INT,
+ -- The floating point measurement.
+ val REAL,
+ -- Id of the source filename, from SourceFiles.
+ source_file_id INT,
+ PRIMARY KEY (trace_id, commit_number)
+ );
Just using this table we can construct some useful queries. For example
we can count the number of traces in a single tile, in this case the
0th tile in a system with a tileSize of 256:
- SELECT
- COUNT(DISTINCT trace_id)
- FROM
- TraceValues
- WHERE
- commit_number >= 0 AND commit_number < 256;
+ SELECT
+ COUNT(DISTINCT trace_id)
+ FROM
+ TraceValues2
+ WHERE
+ commit_number >= 0 AND commit_number < 256;
-The Postings table is our inverted index for looking up which trace ids
-contain which key=value pairs. For a good introduction to postings and search
-https://www.tbray.org/ongoing/When/200x/2003/06/18/HowSearchWorks is a good
-resource.
+The JSONB serialized Params in the TraceNames table allows
+building ParamSets for a range of commits:
-Remember that each trace name is a structured key of the form
-,arch=x86,config=8888,..., and that over time traces may come and go, i.e. we
-may stop running a test, or start running new tests, so if we want to make
-searching for traces efficient we need to be aware of how those trace ids change
-over time. The answer is to break our store in Tiles, i.e. blocks of commits of
-tileSize length, and then for each Tile we keep an inverted index of the trace
-ids. This allows us to not only construct fast queries, but to also do things
-like build ParamSets, a collection of all the keys and all their values ever
-seen for a particular Tile.
+ SELECT
+ DISTINCT TraceNames.params
+ FROM
+ TraceNames
+ INNER JOIN TraceValues2 ON TraceNames.trace_id = TraceValues2.trace_id
+ WHERE
+ TraceValues2.commit_number >= 0
+ AND TraceValues2.commit_number < 512;
-In the table below we store a key_value which is the literal "key=value" part of
-a trace name, along with the tile_number and the 64 bit trace id. Note that
-tile_number is just int(commitNumber/tileSize).
-
- Postings (
- tile_number INTEGER,
- key_value text NOT NULL,
- trace_id INTEGER,
- PRIMARY KEY (tile_number, key_value, trace_id)
- )
-
-So for example to build a ParamSet from Postings:
-
- SELECT DISTINCT
- key_value
- FROM
- Postings
- WHERE
- tile_number=0;
-
-To find the most recent tile:
-
- SELECT
- tile_number
- FROM
- Postings
- ORDER BY
- tile_number DESC LIMIT 1;
And finally, to retrieve all the trace values that
-would match a query, we first start with sub-queries for
-each of the common keys in the query, which produce the
-trace_ids, which are then ANDed across all the distinct
-keys in the query. Finally that list is inner joined to the
-TraceValues table to load up all the values.
+would match a query:
- SELECT
- TraceIDs.trace_name, TraceValues.commit_number, TraceValues.val
- FROM
- TraceIDs
- INNER JOIN
- TraceValues
- ON
- TraceValues.trace_id = TraceIDs.trace_id
- WHERE
- TraceValues.trace_id IN (
- SELECT trace_id FROM Postings
- WHERE key_value IN ("arch=x86", "arch=arm")
- AND tile_number=0
- )
- AND
- TraceValues.trace_id IN (
- SELECT trace_id FROM Postings
- WHERE key_value IN ("config=8888")
- AND tile_number=0
- );
+ SELECT
+ TraceNames.params,
+ TraceValues2.commit_number,
+ TraceValues2.val
+ FROM
+ TraceNames
+ INNER JOIN TraceValues2 ON TraceValues2.trace_id = TraceNames.trace_id
+ WHERE
+ TraceNames.params ->> 'arch' IN ('x86')
+ AND TraceNames.params ->> 'config' IN ('565', '8888')
+ AND TraceValues2.commit_number >= 0
+ AND TraceValues2.commit_number < 255;
-Look in migrations/test.sql for more example of raw queries using
+Look in migrations/cdb.sql for more example of raw queries using
a simple example dataset.
*/
package sqltracestore
@@ -126,15 +96,14 @@
"bytes"
"context"
"crypto/md5"
- "database/sql"
- "encoding/hex"
+ "encoding/json"
"fmt"
"sort"
- "strconv"
- "strings"
"text/template"
"time"
+ "github.com/jackc/pgx/v4"
+ "github.com/jackc/pgx/v4/pgxpool"
"go.skia.org/infra/go/metrics2"
"go.skia.org/infra/go/paramtools"
"go.skia.org/infra/go/query"
@@ -145,27 +114,31 @@
"go.skia.org/infra/go/vec32"
"go.skia.org/infra/perf/go/cache"
"go.skia.org/infra/perf/go/cache/local"
- "go.skia.org/infra/perf/go/cache/memcached"
"go.skia.org/infra/perf/go/config"
- perfsql "go.skia.org/infra/perf/go/sql"
"go.skia.org/infra/perf/go/tracestore"
"go.skia.org/infra/perf/go/tracestore/sqltracestore/engine"
"go.skia.org/infra/perf/go/types"
)
+const writeTracesChunkSize = 200
+
// defaultCacheSize is the size of the in-memory LRU cache if no size was
// specified in the config file.
const defaultCacheSize = 20 * 1000 * 1000
-// Ingest instances have around 8 cores and many ingested files have ~10K
-// values, so pick a batch size that allows roughly one request per core.
-const traceValuesInsertBatchSize = 1000
+// traceIDForSQL is the type of the IDs that are used in the SQL queries,
+// they are hex encoded md5 hashes of a trace name, e.g. "\x00112233...".
+// Note the \x prefix which tells CockroachDB that this is hex encoded.
+type traceIDForSQL string
-// traceIDFromSQL is the type of the IDs that are used in the SQL database for
-// traces.
-type traceIDFromSQL int64
+var badTraceIDFromSQL traceIDForSQL = ""
-const badTraceIDFromSQL traceIDFromSQL = -1
+// Calculates the traceIDForSQL for the given trace name, e.g. "\x00112233...".
+// Note the \x prefix which tells CockroachDB that this is hex encoded.
+func traceIDForSQLFromTraceName(traceName string) traceIDForSQL {
+ b := md5.Sum([]byte(traceName))
+ return traceIDForSQL(fmt.Sprintf("\\x%x", b))
+}
// sourceFileIDFromSQL is the type of the IDs that are used in the SQL database
// for source files.
@@ -176,189 +149,231 @@
// statement is an SQL statement or fragment of an SQL statement.
type statement int
-// All the different statements we need.
+// All the different statements we need. Each statement will appear either in
+// templatesByDialect or statementsByDialect.
const (
insertIntoSourceFiles statement = iota
getSourceFileID
- insertIntoTraceIDs
+ replaceTraceNames
getTraceID
- insertIntoPostings
replaceTraceValues
- countIndices
- getLatestTile
+ getLatestCommit
paramSetForTile
getSource
traceCount
+ queryTraces
+ queryTracesIDOnly
+ readTraces
)
-type statements map[statement]string
-
-type templates map[statement]string
-
-var templatesByDialect = map[perfsql.Dialect]templates{
- perfsql.CockroachDBDialect: {
- insertIntoPostings: `
- UPSERT INTO
- Postings (tile_number, key_value, trace_id)
- VALUES
- {{ range $index, $element := . -}}
- {{ if $index }},{{end}}({{ $element.TileNumber }}, '{{ $element.KeyValue }}', {{ $element.TraceID }})
- {{ end }}`,
- replaceTraceValues: `
- UPSERT INTO
- TraceValues (trace_id, commit_number, val, source_file_id)
- VALUES
- {{ range $index, $element := . -}}
- {{ if $index }},{{end}}({{ $element.TraceID }}, {{ $element.CommitNumber }}, {{ $element.Val }}, {{ $element.SourceFileID }})
- {{ end }}`,
- },
+var templates = map[statement]string{
+ replaceTraceValues: `INSERT INTO
+ TraceValues2 (trace_id, commit_number, val, source_file_id)
+ VALUES
+ {{ range $index, $element := . -}}
+ {{ if $index }},{{end}}
+ (
+ '{{ $element.MD5HexTraceID }}', {{ $element.CommitNumber }}, {{ $element.Val }}, {{ $element.SourceFileID }}
+ )
+ {{ end }}
+ ON CONFLICT
+ DO NOTHING
+ `,
+ replaceTraceNames: `INSERT INTO
+ TraceNames (trace_id, params)
+ VALUES
+ {{ range $index, $element := . -}}
+ {{ if $index }},{{end}}
+ (
+ '{{ $element.MD5HexTraceID }}', '{{ $element.JSONParams }}'
+ )
+ {{ end }}
+ ON CONFLICT
+ DO NOTHING
+ `,
+ queryTraces: `
+ SELECT
+ TraceNames.params,
+ TraceValues2.commit_number,
+ TraceValues2.val
+ FROM
+ TraceNames
+ INNER JOIN TraceValues2 ON TraceValues2.trace_id = TraceNames.trace_id
+ WHERE
+ TraceValues2.commit_number >= {{ .BeginCommitNumber }}
+ AND TraceValues2.commit_number < {{ .EndCommitNumber }}
+ {{ range $key, $values := .QueryPlan }}
+ AND TraceNames.params ->> '{{ $key }}' IN
+ (
+ {{ range $index, $value := $values -}}
+ {{ if $index }},{{end}} '{{ $value }}'
+ {{ end }}
+ )
+ {{ end }}
+ `,
+ queryTracesIDOnly: `
+ SELECT
+ TraceNames.params
+ FROM
+ TraceNames
+ INNER JOIN TraceValues2 ON TraceValues2.trace_id = TraceNames.trace_id
+ WHERE
+ TraceValues2.commit_number >= {{ .BeginCommitNumber }}
+ AND TraceValues2.commit_number < {{ .EndCommitNumber }}
+ {{ range $key, $values := .QueryPlan }}
+ AND TraceNames.params ->> '{{ $key }}' IN
+ (
+ {{ range $index, $value := $values -}}
+ {{ if $index }},{{end}} '{{ $value }}'
+ {{ end }}
+ )
+ {{ end }}
+ `,
+ readTraces: `
+ SELECT
+ TraceNames.params,
+ TraceValues2.commit_number,
+ TraceValues2.val
+ FROM
+ TraceNames
+ INNER JOIN TraceValues2 ON TraceValues2.trace_id = TraceNames.trace_id
+ WHERE
+ TraceValues2.commit_number >= {{ .BeginCommitNumber }}
+ AND TraceValues2.commit_number < {{ .EndCommitNumber }}
+ AND TraceValues2.trace_id IN
+ (
+ {{ range $index, $trace_id := .TraceIDs -}}
+ {{ if $index }},{{end}} '{{ $trace_id }}'
+ {{ end }}
+ )
+ `,
+ getSource: `
+ SELECT
+ SourceFiles.source_file
+ FROM
+ TraceNames
+ INNER JOIN TraceValues2 ON TraceValues2.trace_id = TraceNames.trace_id
+ INNER JOIN SourceFiles ON SourceFiles.source_file_id = TraceValues2.source_file_id
+ WHERE
+ TraceNames.trace_id = '{{ .MD5HexTraceID }}'
+ AND TraceValues2.commit_number = {{ .CommitNumber }}`,
}
-// insertIntoPostingsValue represents the values used in the insertIntoPostings template.
-type insertIntoPostingsValue struct {
- TileNumber types.TileNumber
- KeyValue string
- TraceID traceIDFromSQL
-}
+// replaceTraceValuesContext is the context for the replaceTraceValues template.
+type replaceTraceValuesContext struct {
+ // The MD5 sum of the trace name as a hex string, i.e.
+ // "\xfe385b159ff55dca481069805e5ff050". Note the leading \x which
+ // CockroachDB will use to know the string is in hex.
+ MD5HexTraceID traceIDForSQL
-// replaceTraceValue is used in the replaceTraceValues template.
-type replaceTraceValue struct {
- TraceID traceIDFromSQL
CommitNumber types.CommitNumber
Val float32
SourceFileID sourceFileIDFromSQL
}
-var statementsByDialect = map[perfsql.Dialect]statements{
- perfsql.CockroachDBDialect: {
- insertIntoSourceFiles: `
- INSERT INTO
- SourceFiles (source_file)
- VALUES
- ($1)
- ON CONFLICT
- DO NOTHING`,
- getSourceFileID: `
- SELECT
- source_file_id
- FROM
- SourceFiles
- WHERE
- source_file=$1`,
- insertIntoTraceIDs: `
- INSERT INTO
- TraceIDs (trace_name)
- VALUES
- ($1)
- ON CONFLICT
- DO NOTHING`,
- getTraceID: `
- SELECT
- trace_id
- FROM
- TraceIDs
- WHERE
- trace_name=$1`,
- insertIntoPostings: `
- INSERT INTO
- Postings (tile_number, key_value, trace_id)
- VALUES
- ($1, $2, $3)
- ON CONFLICT
- DO NOTHING`,
- countIndices: `
- SELECT
- COUNT(*)
- FROM
- Postings
- WHERE
- tile_number=$1`,
- getLatestTile: `
- SELECT
- tile_number
- FROM
- Postings
- ORDER BY
- tile_number DESC
- LIMIT 1`,
- paramSetForTile: `
- SELECT DISTINCT
- key_value
- FROM
- Postings
- WHERE
- tile_number=$1`,
- getSource: `
- SELECT
- SourceFiles.source_file
- FROM
- TraceIDs
- INNER JOIN
- TraceValues ON TraceValues.trace_id = TraceIDs.trace_id
- INNER JOIN
- SourceFiles ON SourceFiles.source_file_id = TraceValues.source_file_id
- WHERE
- TraceIDs.trace_name=$1 AND TraceValues.commit_number=$2`,
- traceCount: `
- SELECT
- COUNT(DISTINCT trace_id)
- FROM
- TraceValues
- WHERE
- commit_number >= $1 AND commit_number <= $2`,
- },
+// replaceTraceNamesContext is the context for the replaceTraceNames template.
+type replaceTraceNamesContext struct {
+ // The trace's Params serialize as JSON.
+ JSONParams string
+
+ // The MD5 sum of the trace name as a hex string, i.e.
+ // "\xfe385b159ff55dca481069805e5ff050". Note the leading \x which
+ // CockroachDB will use to know the string is in hex.
+ MD5HexTraceID traceIDForSQL
+}
+
+// queryTracesContext is the context for the queryTraces template.
+type queryTracesContext struct {
+ BeginCommitNumber types.CommitNumber
+ EndCommitNumber types.CommitNumber
+ QueryPlan paramtools.ParamSet
+}
+
+// readTracesContext is the context for the readTraces template.
+type readTracesContext struct {
+ BeginCommitNumber types.CommitNumber
+ EndCommitNumber types.CommitNumber
+ TraceIDs []traceIDForSQL
+}
+
+// getSourceContext is the context for the getSource template.
+type getSourceContext struct {
+ CommitNumber types.CommitNumber
+
+ // The MD5 sum of the trace name as a hex string, i.e.
+ // "\xfe385b159ff55dca481069805e5ff050". Note the leading \x which
+ // CockroachDB will use to know the string is in hex.
+ MD5HexTraceID traceIDForSQL
+}
+
+var statements = map[statement]string{
+ insertIntoSourceFiles: `
+ INSERT INTO
+ SourceFiles (source_file)
+ VALUES
+ ($1)
+ ON CONFLICT
+ DO NOTHING`,
+ getSourceFileID: `
+ SELECT
+ source_file_id
+ FROM
+ SourceFiles
+ WHERE
+ source_file=$1`,
+ getLatestCommit: `
+ SELECT
+ commit_number
+ FROM
+ TraceValues2
+ ORDER BY
+ commit_number DESC
+ LIMIT
+ 1;`,
+ paramSetForTile: `
+ SELECT
+ DISTINCT TraceNames.params
+ FROM
+ TraceNames
+ INNER JOIN TraceValues2 ON TraceNames.trace_id = TraceValues2.trace_id
+ WHERE
+ TraceValues2.commit_number >= $1
+ AND TraceValues2.commit_number < $2;`,
+ traceCount: `
+ SELECT
+ COUNT(DISTINCT trace_id)
+ FROM
+ TraceValues2
+ WHERE
+ commit_number >= $1 AND commit_number <= $2`,
}
// SQLTraceStore implements tracestore.TraceStore backed onto an SQL database.
type SQLTraceStore struct {
// db is the SQL database instance.
- db *sql.DB
-
- // preparedStatements are all the prepared SQL statements.
- preparedStatements map[statement]*sql.Stmt
+ db *pgxpool.Pool
// unpreparedStatements are parsed templates that can be used to construct SQL statements.
unpreparedStatements map[statement]*template.Template
- // cache is an LRU cache used to store the ids (traceIDFromSQL) of hashed trace names (string):
- // "md5(traceName)" -> traceIDFromSQL
- // and if a traces postings have been written for a tile (bool).
- // "traceID-tileNumber" -> bool
- //
- // For traceNames the md5 sum is used to keep the cache keys shorter, thus preserving
- // memory and also allowing the use of memcached which restricts keys to 250 chars.
cache cache.Cache
// tileSize is the number of commits per Tile.
tileSize int32
// metrics
- writeTracesMetric metrics2.Float64SummaryMetric
- writeTracesMetricSQL metrics2.Float64SummaryMetric
- updateTraceValuesMetric metrics2.Float64SummaryMetric
- buildTracesContextMetric metrics2.Float64SummaryMetric
- writeTraceIDAndPostingsCacheHitMetric metrics2.Counter
- writeTraceIDAndPostingsCacheMissMetric metrics2.Counter
- writeTraceIDAndPostingsPostingsCacheHitMetric metrics2.Counter
- writeTraceIDAndPostingsPostingsCacheMissMetric metrics2.Counter
+ writeTracesMetric metrics2.Float64SummaryMetric
+ writeTracesMetricSQL metrics2.Float64SummaryMetric
+ buildTracesContextsMetric metrics2.Float64SummaryMetric
}
// New returns a new *SQLTraceStore.
//
// We presume all migrations have been run against db before this function is
// called.
-func New(db *sql.DB, dialect perfsql.Dialect, datastoreConfig config.DataStoreConfig) (*SQLTraceStore, error) {
- preparedStatements := map[statement]*sql.Stmt{}
- for key, statement := range statementsByDialect[dialect] {
- prepared, err := db.Prepare(statement)
- if err != nil {
- return nil, skerr.Wrapf(err, "preparing statement %v, %q", key, statement)
- }
- preparedStatements[key] = prepared
- }
-
+func New(db *pgxpool.Pool, datastoreConfig config.DataStoreConfig) (*SQLTraceStore, error) {
unpreparedStatements := map[statement]*template.Template{}
- for key, tmpl := range templatesByDialect[dialect] {
+ for key, tmpl := range templates {
t, err := template.New("").Parse(tmpl)
if err != nil {
return nil, skerr.Wrapf(err, "parsing template %v, %q", key, tmpl)
@@ -366,44 +381,19 @@
unpreparedStatements[key] = t
}
- var memoryCacheSize = datastoreConfig.Cache.Size
- if memoryCacheSize == 0 {
- memoryCacheSize = defaultCacheSize
- }
-
- var err error
- var cache cache.Cache
- if len(datastoreConfig.Cache.MemcachedServers) > 0 && datastoreConfig.Cache.Namespace != "" {
- cache, err = memcached.New(datastoreConfig.Cache.MemcachedServers, datastoreConfig.Cache.Namespace)
- if err != nil {
- // Fall back to in-memory cache.
- cache, err = local.New(memoryCacheSize)
- if err != nil {
- return nil, skerr.Wrap(err)
- }
-
- }
- } else {
- cache, err = local.New(memoryCacheSize)
- if err != nil {
- return nil, skerr.Wrap(err)
- }
+ cache, err := local.New(defaultCacheSize)
+ if err != nil {
+ return nil, skerr.Wrap(err)
}
return &SQLTraceStore{
- db: db,
- preparedStatements: preparedStatements,
- unpreparedStatements: unpreparedStatements,
- cache: cache,
- tileSize: datastoreConfig.TileSize,
- writeTracesMetric: metrics2.GetFloat64SummaryMetric("perfserver_sqltracestore_writeTraces"),
- writeTracesMetricSQL: metrics2.GetFloat64SummaryMetric("perfserver_sqltracestore_writeTracesSQL"),
- updateTraceValuesMetric: metrics2.GetFloat64SummaryMetric("perfserver_sqltracestore_updateTraceValues"),
- buildTracesContextMetric: metrics2.GetFloat64SummaryMetric("perfserver_sqltracestore_buildTracesContext"),
- writeTraceIDAndPostingsCacheHitMetric: metrics2.GetCounter("perfserver_sqltracestore_writeTraceIDAndPostings_cache_hit"),
- writeTraceIDAndPostingsCacheMissMetric: metrics2.GetCounter("perfserver_sqltracestore_writeTraceIDAndPostings_cache_miss"),
- writeTraceIDAndPostingsPostingsCacheHitMetric: metrics2.GetCounter("perfserver_sqltracestore_writeTraceIDAndPostings_postings_cache_hit"),
- writeTraceIDAndPostingsPostingsCacheMissMetric: metrics2.GetCounter("perfserver_sqltracestore_writeTraceIDAndPostings_postings_cache_miss"),
+ db: db,
+ unpreparedStatements: unpreparedStatements,
+ tileSize: datastoreConfig.TileSize,
+ cache: cache,
+ writeTracesMetric: metrics2.GetFloat64SummaryMetric("perfserver_sqltracestore_writeTraces"),
+ writeTracesMetricSQL: metrics2.GetFloat64SummaryMetric("perfserver_sqltracestore_writeTracesSQL"),
+ buildTracesContextsMetric: metrics2.GetFloat64SummaryMetric("perfserver_sqltracestore_buildTracesContext"),
}, nil
}
@@ -416,41 +406,47 @@
// CountIndices implements the tracestore.TraceStore interface.
func (s *SQLTraceStore) CountIndices(ctx context.Context, tileNumber types.TileNumber) (int64, error) {
- var ret int64
- if err := s.preparedStatements[countIndices].QueryRowContext(ctx, tileNumber).Scan(&ret); err != nil {
- return 0, skerr.Wrap(err)
- }
- return ret, nil
+
+ // This doesn't make any sense for the SQL implementation of TraceStore.
+ return 0, nil
}
// GetLatestTile implements the tracestore.TraceStore interface.
func (s *SQLTraceStore) GetLatestTile() (types.TileNumber, error) {
- tileNumber := types.BadTileNumber
- if err := s.preparedStatements[getLatestTile].QueryRowContext(context.TODO()).Scan(&tileNumber); err != nil {
- return tileNumber, skerr.Wrap(err)
+ mostRecentCommit := types.BadCommitNumber
+ if err := s.db.QueryRow(context.TODO(), statements[getLatestCommit]).Scan(&mostRecentCommit); err != nil {
+ return types.BadTileNumber, skerr.Wrap(err)
}
- return tileNumber, nil
+ return types.TileNumberFromCommitNumber(mostRecentCommit, s.tileSize), nil
}
func (s *SQLTraceStore) paramSetForTile(tileNumber types.TileNumber) (paramtools.ParamSet, error) {
- rows, err := s.preparedStatements[paramSetForTile].QueryContext(context.TODO(), tileNumber)
+ // Convert the tile number into a range of commits, since we don't store data by
+ // tile anymore.
+ beginCommit, endCommit := types.TileCommitRangeForTileNumber(tileNumber, s.tileSize)
+
+ rows, err := s.db.Query(context.TODO(), statements[paramSetForTile], beginCommit, endCommit)
if err != nil {
- return nil, skerr.Wrapf(err, "tileNumer=%d", tileNumber)
+ fmt.Printf("%q %d %d", statements[paramSetForTile], beginCommit, endCommit)
+ return nil, skerr.Wrapf(err, "Failed querying - tileNumber=%d", tileNumber)
}
ret := paramtools.NewParamSet()
for rows.Next() {
- var keyValue string
- if err := rows.Scan(&keyValue); err != nil {
- return nil, skerr.Wrapf(err, "tileNumer=%d", tileNumber)
+ var jsonParams string
+ if err := rows.Scan(&jsonParams); err != nil {
+ return nil, skerr.Wrapf(err, "Failed scanning row - tileNumber=%d", tileNumber)
}
- parts := strings.Split(keyValue, "=")
- if len(parts) != 2 {
- return nil, skerr.Fmt("Invalid key=value form: %q", keyValue)
+ var ps paramtools.Params
+ if err := json.Unmarshal([]byte(jsonParams), &ps); err != nil {
+ return nil, skerr.Wrapf(err, "Failed unmarshal - tileNumber=%d", tileNumber)
}
- ret.AddParams(paramtools.Params{parts[0]: parts[1]})
+ ret.AddParams(ps)
+ }
+ if err == pgx.ErrNoRows {
+ return ret, nil
}
if err := rows.Err(); err != nil {
- return nil, skerr.Wrapf(err, "tileNumer=%d", tileNumber)
+ return nil, skerr.Wrapf(err, "Other failure - tileNumber=%d", tileNumber)
}
ret.Normalize()
return ret, nil
@@ -471,8 +467,21 @@
// GetSource implements the tracestore.TraceStore interface.
func (s *SQLTraceStore) GetSource(ctx context.Context, commitNumber types.CommitNumber, traceName string) (string, error) {
var filename string
- if err := s.preparedStatements[getSource].QueryRowContext(ctx, traceName, commitNumber).Scan(&filename); err != nil {
- return "", skerr.Wrapf(err, "commitNumber=%d traceName=%q", commitNumber, traceName)
+ traceID := traceIDForSQLFromTraceName(traceName)
+
+ sourceContext := getSourceContext{
+ MD5HexTraceID: traceID,
+ CommitNumber: commitNumber,
+ }
+
+ var b bytes.Buffer
+ if err := s.unpreparedStatements[getSource].Execute(&b, sourceContext); err != nil {
+ return "", skerr.Wrapf(err, "failed to expand get source template")
+ }
+ sql := b.String()
+
+ if err := s.db.QueryRow(ctx, sql).Scan(&filename); err != nil {
+ return "", skerr.Wrapf(err, "commitNumber=%d traceName=%q traceID=%q", commitNumber, traceName, traceID)
}
return filename, nil
}
@@ -507,64 +516,56 @@
if err := query.ValidateParamSet(plan); err != nil {
return nil, skerr.Wrapf(err, "invalid query %#v", *q)
}
- tileNumberString := strconv.FormatInt(int64(tileNumber), 10)
+
+ // Prepare the template context.
beginCommit, endCommit := types.TileCommitRangeForTileNumber(tileNumber, s.tileSize)
-
- stmt := fmt.Sprintf(`
-SELECT
- TraceIDs.trace_name, TraceValues.commit_number, TraceValues.val
-FROM
- TraceIDs
-INNER JOIN
- TraceValues
-ON
- TraceValues.trace_id = TraceIDs.trace_id
-WHERE
- TraceValues.commit_number >= %d
- AND TraceValues.commit_number <= %d
- AND `, beginCommit, endCommit)
-
- // TODO(jcgregorio) Break out into own function.
- whereClauses := make([]string, len(plan))
- clauseIndex := 0
- for key, values := range plan {
- inClause := make([]string, len(values))
- for i, value := range values {
- inClause[i] = fmt.Sprintf("'%s=%s'", key, value)
- }
- whereClauses[clauseIndex] = `
- TraceValues.trace_id IN (
- SELECT trace_id FROM Postings WHERE key_value IN (` +
- strings.Join(inClause, ",") +
- `) AND tile_number=` + tileNumberString + `
- )`
- clauseIndex++
+ context := queryTracesContext{
+ BeginCommitNumber: beginCommit,
+ EndCommitNumber: endCommit,
+ QueryPlan: plan,
}
- fullStatement := stmt + strings.Join(whereClauses, " AND ")
- rows, err := s.db.QueryContext(context.TODO(), fullStatement)
+ // Expand the template for the SQL.
+ var b bytes.Buffer
+ if err := s.unpreparedStatements[queryTraces].Execute(&b, context); err != nil {
+ return nil, skerr.Wrapf(err, "failed to expand trace names template")
+ }
+
+ sql := b.String()
+ // Execute the query.
+ rows, err := s.db.Query(ctx, sql)
if err != nil {
- sklog.Debugf("QueryTracesByIndex: fullStatement: %q", fullStatement)
return nil, skerr.Wrap(err)
}
ret := types.TraceSet{}
-
for rows.Next() {
- var traceName string
+ var jsonParams string
var commitNumber types.CommitNumber
var val float64
- if err := rows.Scan(&traceName, &commitNumber, &val); err != nil {
+ if err := rows.Scan(&jsonParams, &commitNumber, &val); err != nil {
return nil, skerr.Wrap(err)
}
+
+ p := paramtools.Params{}
+ if err := json.Unmarshal([]byte(jsonParams), &p); err != nil {
+ sklog.Warningf("Invalid JSON params found in query response: %s", err)
+ continue
+ }
+ traceName, err := query.MakeKey(p)
+ if err != nil {
+ sklog.Warningf("Invalid trace name found in query response: %s", err)
+ continue
+ }
+ offset := s.OffsetFromCommitNumber(commitNumber)
if _, ok := ret[traceName]; ok {
- ret[traceName][s.OffsetFromCommitNumber(commitNumber)] = float32(val)
+ ret[traceName][offset] = float32(val)
} else {
// TODO(jcgregorio) Replace this vec32.New() with a
// https://golang.org/pkg/sync/#Pool since this is our most used/reused
// type of memory.
ret[traceName] = vec32.New(int(s.tileSize))
- ret[traceName][s.OffsetFromCommitNumber(commitNumber)] = float32(val)
+ ret[traceName][offset] = float32(val)
}
}
if err := rows.Err(); err != nil {
@@ -604,74 +605,52 @@
close(outParams)
return outParams, nil
}
+
// Sanitize our inputs.
if err := query.ValidateParamSet(plan); err != nil {
- close(outParams)
- return outParams, skerr.Wrapf(err, "invalid query %#v", *q)
+ return nil, skerr.Wrapf(err, "invalid query %#v", *q)
}
- tileNumberString := strconv.FormatInt(int64(tileNumber), 10)
+ // Prepare the template context.
beginCommit, endCommit := types.TileCommitRangeForTileNumber(tileNumber, s.tileSize)
-
- stmt := fmt.Sprintf(`
-SELECT DISTINCT
- TraceIDs.trace_name
-FROM
- TraceIDs
-INNER JOIN
- TraceValues
-ON
- TraceValues.trace_id = TraceIDs.trace_id
-WHERE
- TraceValues.commit_number >= %d
- AND TraceValues.commit_number <= %d
- AND `, beginCommit, endCommit)
-
- // TODO(jcgregorio) Break out into own function.
- // TODO(jcgregorio) Convert to use text/templates.
- whereClauses := make([]string, len(plan))
- clauseIndex := 0
- for key, values := range plan {
- inClause := make([]string, len(values))
- for i, value := range values {
- inClause[i] = fmt.Sprintf("'%s=%s'", key, value)
- }
- whereClauses[clauseIndex] = `
- TraceValues.trace_id IN (
- SELECT trace_id FROM Postings WHERE key_value IN (` +
- strings.Join(inClause, ",") +
- `) AND tile_number=` + tileNumberString + `
- )`
- clauseIndex++
+ context := queryTracesContext{
+ BeginCommitNumber: beginCommit,
+ EndCommitNumber: endCommit,
+ QueryPlan: plan,
}
- fullStatement := stmt + strings.Join(whereClauses, " AND ")
- rows, err := s.db.QueryContext(context.TODO(), fullStatement)
+ // Expand the template for the SQL.
+ var b bytes.Buffer
+ if err := s.unpreparedStatements[queryTracesIDOnly].Execute(&b, context); err != nil {
+ return nil, skerr.Wrapf(err, "failed to expand trace names template")
+ }
+
+ // Execute the query.
+ rows, err := s.db.Query(ctx, b.String())
if err != nil {
- sklog.Debugf("QueryTracesIDOnlyByIndex: fullStatement: %q", fullStatement)
- close(outParams)
- return outParams, skerr.Wrap(err)
+ return nil, skerr.Wrap(err)
}
go func() {
defer close(outParams)
for rows.Next() {
- var traceName string
- if err := rows.Scan(&traceName); err != nil {
+ var jsonParams string
+ if err := rows.Scan(&jsonParams); err != nil {
sklog.Errorf("Failed to scan traceName: %s", skerr.Wrap(err))
return
}
- p, err := query.ParseKey(traceName)
- if err != nil {
+
+ p := paramtools.Params{}
+ if err := json.Unmarshal([]byte(jsonParams), &p); err != nil {
sklog.Errorf("Failed to parse traceName: %s", skerr.Wrap(err))
- return
+ continue
}
outParams <- p
+
}
if err := rows.Err(); err != nil {
sklog.Errorf("Failed while reading traceNames: %s", skerr.Wrap(err))
- return
}
}()
@@ -696,39 +675,51 @@
// Get the traceIDs for the given keys.
beginCommit, endCommit := types.TileCommitRangeForTileNumber(tileNumber, s.tileSize)
- stmt := fmt.Sprintf(`
-SELECT
- TraceIDs.trace_name, TraceValues.commit_number, TraceValues.val
-FROM
- TraceIDs
-INNER JOIN
- TraceValues ON TraceValues.trace_id = TraceIDs.trace_id
-WHERE
- TraceValues.commit_number >= %d
- AND TraceValues.commit_number <= %d
- AND TraceIDs.trace_name IN (`, beginCommit, endCommit)
- inClause := make([]string, len(keys))
- for i, key := range keys {
- singleQuoted := "'" + key + "'"
- inClause[i] = singleQuoted
+
+ // Populate the context for the SQL template.
+ readTracesContext := readTracesContext{
+ BeginCommitNumber: beginCommit,
+ EndCommitNumber: endCommit,
+ TraceIDs: make([]traceIDForSQL, 0, len(keys)),
}
- fullStatement := stmt + strings.Join(inClause, ",") + ")"
- rows, err := s.db.QueryContext(context.TODO(), fullStatement)
+
+ for _, traceName := range keys {
+ readTracesContext.TraceIDs = append(readTracesContext.TraceIDs, traceIDForSQLFromTraceName(traceName))
+ }
+ // Expand the template for the SQL.
+ var b bytes.Buffer
+ if err := s.unpreparedStatements[readTraces].Execute(&b, readTracesContext); err != nil {
+ return nil, skerr.Wrapf(err, "failed to expand read traces template")
+ }
+
+ // Execute the query.
+ rows, err := s.db.Query(context.TODO(), b.String())
if err != nil {
- sklog.Debugf("ReadTraces: fullStatement: %q", fullStatement)
return nil, skerr.Wrap(err)
}
+
for rows.Next() {
- var traceName string
+ var jsonParams string
var commitNumber types.CommitNumber
var val float64
- if err := rows.Scan(&traceName, &commitNumber, &val); err != nil {
+ if err := rows.Scan(&jsonParams, &commitNumber, &val); err != nil {
return nil, skerr.Wrap(err)
}
+
+ p := paramtools.Params{}
+ if err := json.Unmarshal([]byte(jsonParams), &p); err != nil {
+ sklog.Warningf("Invalid JSON params found in query response: %s", err)
+ continue
+ }
+ traceName, err := query.MakeKey(p)
+ if err != nil {
+ sklog.Warningf("Invalid trace name found in query response: %s", err)
+ continue
+ }
ret[traceName][s.OffsetFromCommitNumber(commitNumber)] = float32(val)
}
if err := rows.Err(); err != nil {
- return nil, skerr.Wrapf(err, "tileNumber=%d", tileNumber)
+ return nil, skerr.Wrap(err)
}
return ret, nil
@@ -748,7 +739,7 @@
func (s *SQLTraceStore) TraceCount(ctx context.Context, tileNumber types.TileNumber) (int64, error) {
var ret int64
beginCommit, endCommit := types.TileCommitRangeForTileNumber(tileNumber, s.tileSize)
- err := s.preparedStatements[traceCount].QueryRowContext(context.TODO(), beginCommit, endCommit).Scan(&ret)
+ err := s.db.QueryRow(context.TODO(), statements[traceCount], beginCommit, endCommit).Scan(&ret)
return ret, skerr.Wrap(err)
}
@@ -762,11 +753,11 @@
// the sourceFileIDFromSQL of that filename.
func (s *SQLTraceStore) updateSourceFile(filename string) (sourceFileIDFromSQL, error) {
ret := badSourceFileIDFromSQL
- _, err := s.preparedStatements[insertIntoSourceFiles].ExecContext(context.TODO(), filename)
+ _, err := s.db.Exec(context.TODO(), statements[insertIntoSourceFiles], filename)
if err != nil {
return ret, skerr.Wrap(err)
}
- err = s.preparedStatements[getSourceFileID].QueryRowContext(context.TODO(), filename).Scan(&ret)
+ err = s.db.QueryRow(context.TODO(), statements[getSourceFileID], filename).Scan(&ret)
if err != nil {
return ret, skerr.Wrap(err)
}
@@ -774,155 +765,99 @@
return ret, nil
}
-// updatePostings writes all the entries into our inverted index in Postings for
-// the given traceID and tileNumber. The Params given are from the parse trace
-// name.
-func (s *SQLTraceStore) updatePostings(p paramtools.Params, tileNumber types.TileNumber, traceID traceIDFromSQL) error {
- // Clean the data to avoid SQL injection attacks.
- p = query.ForceValid(p)
-
- // Prepare the data for the SQL template.
- values := make([]insertIntoPostingsValue, 0, len(p))
- for k, v := range p {
- keyValue := fmt.Sprintf("%s=%s", k, v)
- values = append(values, insertIntoPostingsValue{
- TileNumber: tileNumber,
- KeyValue: keyValue,
- TraceID: traceID,
- })
- }
-
- // Expand the template.
- var b bytes.Buffer
- if err := s.unpreparedStatements[insertIntoPostings].Execute(&b, values); err != nil {
- return skerr.Wrapf(err, "Failed to expand insertIntoPostings template with: %v", values)
- }
-
- _, err := s.db.ExecContext(context.TODO(), b.String())
- if err != nil {
- return skerr.Wrap(err)
- }
-
- return nil
-}
-
-func getPostingsCacheEntryKey(traceID traceIDFromSQL, tileNumber types.TileNumber) string {
- return fmt.Sprintf("%d-%d", traceID, tileNumber)
-}
-
-func getHashedTraceName(traceName string) string {
- sum := md5.Sum([]byte(traceName))
- return hex.EncodeToString(sum[:])
-}
-
-func (s *SQLTraceStore) getTraceIDFromCache(hashedTraceName string) (traceIDFromSQL, bool) {
- if traceIDString, ok := s.cache.Get(hashedTraceName); ok {
- traceIDInt64, err := strconv.ParseInt(traceIDString, 10, 64)
- if err != nil {
- return badTraceIDFromSQL, false
- }
- return traceIDFromSQL(traceIDInt64), true
- }
- return badTraceIDFromSQL, false
-}
-
-// postingsCacheValue is what we write to the cache when postings exist for the
-// given key. The actual value doesn't matter since we only check for existence
-// when reading the cache.
-const postingsCacheValue = "1"
-
-// writeTraceIDAndPostings writes the trace name into the TraceIDs table and
-// returns the traceIDFromSQL of that trace name. This operation will happen
-// repeatedly as data is ingested so we cache the results in the LRU cache.
-func (s *SQLTraceStore) writeTraceIDAndPostings(traceNameAsParams paramtools.Params, tileNumber types.TileNumber) (traceIDFromSQL, error) {
- traceID := badTraceIDFromSQL
-
- traceName, err := query.MakeKey(traceNameAsParams)
- if err != nil {
- return traceID, skerr.Wrap(err)
- }
- hashedTraceName := getHashedTraceName(traceName)
-
- // Get a traceIDFromSQL for the traceName.
- var ok bool
- if traceID, ok = s.getTraceIDFromCache(hashedTraceName); ok {
- s.writeTraceIDAndPostingsCacheHitMetric.Inc(1)
- } else {
- s.writeTraceIDAndPostingsCacheMissMetric.Inc(1)
- _, err := s.preparedStatements[insertIntoTraceIDs].ExecContext(context.TODO(), traceName)
- if err != nil {
- return traceID, skerr.Wrapf(err, "traceName=%q", traceName)
- }
- err = s.preparedStatements[getTraceID].QueryRowContext(context.TODO(), traceName).Scan(&traceID)
- if err != nil {
- return traceID, skerr.Wrapf(err, "traceName=%q", traceName)
- }
- s.cache.Add(hashedTraceName, fmt.Sprintf("%d", traceID))
- }
- postingsCacheEntryKey := getPostingsCacheEntryKey(traceID, tileNumber)
- if !s.cache.Exists(postingsCacheEntryKey) {
- s.writeTraceIDAndPostingsPostingsCacheMissMetric.Inc(1)
- // TODO(jcgregorio) Do a query on the database to see if the postings are there.
-
- // Update postings.
- if err := s.updatePostings(traceNameAsParams, tileNumber, traceID); err != nil {
- return traceID, skerr.Wrapf(err, "traceName=%q tileNumber=%d traceID=%d", traceName, tileNumber, traceID)
- }
- s.cache.Add(postingsCacheEntryKey, postingsCacheValue)
- } else {
- s.writeTraceIDAndPostingsPostingsCacheHitMetric.Inc(1)
- }
-
- return traceID, nil
-}
-
-// updateTraceValues writes the given slice of replaceTraceValues into the store.
-func (s *SQLTraceStore) updateTraceValues(templateContext []replaceTraceValue) error {
- defer timer.NewWithSummary("perfserver_sqltracestore_updateTraceValues", s.updateTraceValuesMetric).Stop()
- var b bytes.Buffer
- if err := s.unpreparedStatements[replaceTraceValues].Execute(&b, templateContext); err != nil {
- return skerr.Wrapf(err, "failed to expand template")
- }
- _, err := s.db.ExecContext(context.TODO(), b.String())
- return skerr.Wrap(err)
-}
-
// WriteTraces implements the tracestore.TraceStore interface.
func (s *SQLTraceStore) WriteTraces(commitNumber types.CommitNumber, params []paramtools.Params, values []float32, _ paramtools.ParamSet, source string, _ time.Time) error {
defer timer.NewWithSummary("perfserver_sqltracestore_writeTraces", s.writeTracesMetric).Stop()
- tileNumber := types.TileNumberFromCommitNumber(commitNumber, s.tileSize)
// Get the row id for the source file.
sourceID, err := s.updateSourceFile(source)
if err != nil {
return skerr.Wrap(err)
}
- t := timer.NewWithSummary("perfserver_sqltracestore_buildTracesContext", s.buildTracesContextMetric)
+ t := timer.NewWithSummary("perfserver_sqltracestore_buildTracesContexts", s.buildTracesContextsMetric)
// Build the 'context' which will be used to populate the SQL template.
- templateContext := make([]replaceTraceValue, 0, len(params))
+ namesTemplateContext := make([]replaceTraceNamesContext, 0, len(params))
+ valuesTemplateContext := make([]replaceTraceValuesContext, 0, len(params))
+
for i, p := range params {
- traceID, err := s.writeTraceIDAndPostings(p, tileNumber)
+ traceName, err := query.MakeKey(p)
if err != nil {
- t.Stop()
- return skerr.Wrap(err)
+ continue
}
- templateContext = append(templateContext, replaceTraceValue{
- TraceID: traceID,
- CommitNumber: commitNumber,
- Val: values[i],
- SourceFileID: sourceID,
+ traceID := traceIDForSQLFromTraceName(traceName)
+ jsonParams, err := json.Marshal(p)
+ if err != nil {
+ continue
+ }
+ valuesTemplateContext = append(valuesTemplateContext, replaceTraceValuesContext{
+ MD5HexTraceID: traceID,
+ CommitNumber: commitNumber,
+ Val: values[i],
+ SourceFileID: sourceID,
})
+
+ if !s.cache.Exists(string(traceID)) {
+ namesTemplateContext = append(namesTemplateContext, replaceTraceNamesContext{
+ MD5HexTraceID: traceID,
+ JSONParams: string(jsonParams),
+ })
+ }
}
t.Stop()
defer timer.NewWithSummary("perfserver_sqltracestore_writeTraces_sql", s.writeTracesMetricSQL).Stop()
- err = util.ChunkIterParallel(context.TODO(), len(templateContext), traceValuesInsertBatchSize, func(ctx context.Context, startIdx int, endIdx int) error {
- if err := s.updateTraceValues(templateContext[startIdx:endIdx]); err != nil {
- return skerr.Wrapf(err, "failed inserting subSlice: [%d:%d]", startIdx, endIdx)
+ sklog.Infof("About to format %d trace names", len(params))
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+ defer cancel()
+
+ if len(namesTemplateContext) > 0 {
+
+ err = util.ChunkIter(len(namesTemplateContext), 100, func(startIdx int, endIdx int) error {
+ var b bytes.Buffer
+ if err := s.unpreparedStatements[replaceTraceNames].Execute(&b, namesTemplateContext[startIdx:endIdx]); err != nil {
+ return skerr.Wrapf(err, "failed to expand trace names template on slice [%d, %d]", startIdx, endIdx)
+ }
+ sql := b.String()
+
+ sklog.Infof("About to write %d trace names with sql of length %d", len(params), len(sql))
+ if _, err := s.db.Exec(ctx, sql); err != nil {
+ return skerr.Wrapf(err, "Executing: %q", b.String())
+ }
+ return nil
+ })
+
+ if err != nil {
+ return err
+ }
+
+ for _, entry := range namesTemplateContext {
+ s.cache.Add(string(entry.MD5HexTraceID), "1")
+ }
+ }
+
+ sklog.Infof("About to format %d trace values", len(params))
+
+ err = util.ChunkIter(len(valuesTemplateContext), writeTracesChunkSize, func(startIdx int, endIdx int) error {
+ var b bytes.Buffer
+ if err := s.unpreparedStatements[replaceTraceValues].Execute(&b, valuesTemplateContext[startIdx:endIdx]); err != nil {
+ return skerr.Wrapf(err, "failed to expand trace values template")
+ }
+
+ sql := b.String()
+ sklog.Infof("About to write %d trace values with sql of length %d", len(params), len(sql))
+ if _, err := s.db.Exec(ctx, sql); err != nil {
+ return skerr.Wrapf(err, "Executing: %q", b.String())
}
return nil
})
+ if err != nil {
+ return err
+ }
+
+ sklog.Info("Finished writing trace values.")
+
return err
}
diff --git a/perf/go/tracestore/sqltracestore/sqltracestore_test.go b/perf/go/tracestore/sqltracestore/sqltracestore_test.go
index 9d404bc..c4604ad 100644
--- a/perf/go/tracestore/sqltracestore/sqltracestore_test.go
+++ b/perf/go/tracestore/sqltracestore/sqltracestore_test.go
@@ -3,6 +3,7 @@
import (
"context"
"fmt"
+ "math/rand"
"testing"
"time"
@@ -13,7 +14,6 @@
"go.skia.org/infra/go/testutils/unittest"
"go.skia.org/infra/go/vec32"
"go.skia.org/infra/perf/go/config"
- perfsql "go.skia.org/infra/perf/go/sql"
"go.skia.org/infra/perf/go/sql/sqltest"
"go.skia.org/infra/perf/go/types"
)
@@ -23,36 +23,36 @@
e = vec32.MissingDataSentinel
// testTileSize is the size of tiles we use for tests.
- testTileSize = 8
+ testTileSize = int32(8)
)
-func TestCockroachDB(t *testing.T) {
- unittest.LargeTest(t)
-
- cfg := config.DataStoreConfig{
- TileSize: testTileSize,
- Project: "test",
- Instance: "test",
- Table: "test",
- Shards: 8,
- }
-
- for name, subTest := range subTests {
- t.Run(name, func(t *testing.T) {
- db, cleanup := sqltest.NewCockroachDBForTests(t, "tracestore", sqltest.ApplyMigrations)
- // Commenting out the defer cleanup() can sometimes make failures
- // easier to understand.
- defer cleanup()
-
- store, err := New(db, perfsql.CockroachDBDialect, cfg)
- require.NoError(t, err)
-
- subTest(t, store)
- })
- }
+var cfg = config.DataStoreConfig{
+ TileSize: testTileSize,
+ Project: "test",
+ Instance: "test",
+ Table: "test",
+ Shards: 8,
}
-func testUpdateSourceFile(t *testing.T, s *SQLTraceStore) {
+func commonTestSetup(t *testing.T, populateTraces bool) (context.Context, *SQLTraceStore, sqltest.Cleanup) {
+ unittest.LargeTest(t)
+ ctx := context.Background()
+ db, cleanup := sqltest.NewCockroachDBForTests(t, fmt.Sprintf("tracestore%d", rand.Int63()), sqltest.ApplyMigrations)
+
+ store, err := New(db, cfg)
+ require.NoError(t, err)
+
+ if populateTraces {
+ populatedTestDB(t, store)
+ }
+
+ return ctx, store, cleanup
+}
+
+func TestUpdateSourceFile(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
+
// Do each update twice to ensure the IDs don't change.
id, err := s.updateSourceFile("foo.txt")
assert.NoError(t, err)
@@ -69,90 +69,9 @@
assert.Equal(t, id, id2)
}
-func testWriteTraceIDAndPostings(t *testing.T, s *SQLTraceStore) {
- const traceName = ",arch=x86,config=8888,"
- p := paramtools.NewParams(traceName)
- const tileNumber types.TileNumber = 1
-
- // Do each update twice to ensure the IDs don't change.
- traceID, err := s.writeTraceIDAndPostings(p, tileNumber)
- assert.NoError(t, err)
-
- traceID2, err := s.writeTraceIDAndPostings(p, tileNumber)
- assert.NoError(t, err)
- assert.Equal(t, traceID, traceID2)
-
- // Confirm the cache entries exist.
- got, ok := s.getTraceIDFromCache(getHashedTraceName(traceName))
- require.True(t, ok)
- assert.Equal(t, traceID, got)
- assert.True(t, s.cache.Exists(getPostingsCacheEntryKey(traceID, tileNumber)))
-
- const traceName2 = ",arch=arm,config=8888,"
- p2 := paramtools.NewParams(traceName2)
-
- traceID, err = s.writeTraceIDAndPostings(p2, tileNumber)
- assert.NoError(t, err)
- assert.NotEqual(t, traceID, traceID2)
-
- traceID2, err = s.writeTraceIDAndPostings(p2, tileNumber)
- assert.NoError(t, err)
- assert.Equal(t, traceID, traceID2)
-}
-
-func testWriteTraces_MultipleBatches_Success(t *testing.T, s *SQLTraceStore) {
- ctx := context.Background()
-
- const commitNumber = types.CommitNumber(1)
-
- // Add enough values to force it to be done in batches.
- const testLength = 2*traceValuesInsertBatchSize + 1
-
- const tileNumber = types.TileNumber(0)
-
- traceNames := make([]paramtools.Params, 0, testLength)
- values := make([]float32, 0, testLength)
-
- for i := 0; i < testLength; i++ {
- traceNames = append(traceNames, paramtools.Params{
- "traceid": fmt.Sprintf("%d", i),
- "config": "8888",
- })
- values = append(values, float32(i))
- }
- err := s.WriteTraces(
- commitNumber,
- traceNames,
- values,
- paramtools.ParamSet{}, // ParamSet is empty because WriteTraces doesn't use it in this impl.
- "gs://not-tested-as-part-of-this-test.json",
- time.Time{}) // time is unused in this impl of TraceStore.
- require.NoError(t, err)
-
- // Confirm all traces were written.
- q, err := query.NewFromString("config=8888")
- require.NoError(t, err)
- ts, err := s.QueryTracesByIndex(ctx, tileNumber, q)
- assert.NoError(t, err)
- assert.Len(t, ts, testLength)
-
- // Spot test some values.
- q, err = query.NewFromString("config=8888&traceid=0")
- require.NoError(t, err)
- ts, err = s.QueryTracesByIndex(ctx, tileNumber, q)
- assert.NoError(t, err)
-
- assert.Equal(t, float32(0), ts[",config=8888,traceid=0,"][s.OffsetFromCommitNumber(commitNumber)])
-
- q, err = query.NewFromString(fmt.Sprintf("config=8888&traceid=%d", testLength-1))
- require.NoError(t, err)
- ts, err = s.QueryTracesByIndex(ctx, tileNumber, q)
- assert.NoError(t, err)
- assert.Equal(t, float32(testLength-1), ts[fmt.Sprintf(",config=8888,traceid=%d,", testLength-1)][s.OffsetFromCommitNumber(commitNumber)])
-}
-
-func testReadTraces(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
+func TestReadTraces(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
keys := []string{
",arch=x86,config=8888,",
@@ -174,8 +93,9 @@
}, ts)
}
-func testReadTraces_InvalidKey(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
+func TestReadTraces_InvalidKey(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
keys := []string{
",arch=x86,config='); DROP TABLE TraceValues,",
@@ -186,8 +106,9 @@
require.Error(t, err)
}
-func testReadTraces_NoResults(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
+func TestReadTraces_NoResults(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
keys := []string{
",arch=unknown,",
@@ -200,8 +121,9 @@
})
}
-func testReadTraces_EmptyTileReturnsNoData(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
+func TestReadTraces_EmptyTileReturnsNoData(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
keys := []string{
",arch=x86,config=8888,",
@@ -217,9 +139,9 @@
})
}
-func testQueryTracesIDOnlyByIndex_EmptyQueryReturnsError(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesIDOnlyByIndex_EmptyQueryReturnsError(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that matches one trace.
q, err := query.NewFromString("")
@@ -240,9 +162,9 @@
return ret
}
-func testQueryTracesIDOnlyByIndex_EmptyTileReturnsEmptyParamset(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesIDOnlyByIndex_EmptyTileReturnsEmptyParamset(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that matches one trace.
q, err := query.NewFromString("config=565")
@@ -252,9 +174,9 @@
assert.Empty(t, paramSetFromParamsChan(ch))
}
-func testQueryTracesIDOnlyByIndex_MatchesOneTrace(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesIDOnlyByIndex_MatchesOneTrace(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that matches one trace.
q, err := query.NewFromString("config=565")
@@ -268,9 +190,9 @@
assert.Equal(t, expected, paramSetFromParamsChan(ch))
}
-func testQueryTracesIDOnlyByIndex_MatchesTwoTraces(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesIDOnlyByIndex_MatchesTwoTraces(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that matches two traces.
q, err := query.NewFromString("arch=x86")
@@ -284,9 +206,9 @@
assert.Equal(t, expected, paramSetFromParamsChan(ch))
}
-func testQueryTracesByIndex_MatchesOneTrace(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesByIndex_MatchesOneTrace(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that matches one trace.
q, err := query.NewFromString("config=565")
@@ -298,9 +220,9 @@
})
}
-func testQueryTracesByIndex_MatchesOneTraceInTheSecondTile(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesByIndex_MatchesOneTraceInTheSecondTile(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that matches one trace second tile.
q, err := query.NewFromString("config=565")
@@ -312,9 +234,9 @@
})
}
-func testQueryTracesByIndex_MatchesTwoTraces(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesByIndex_MatchesTwoTraces(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that matches two traces.
q, err := query.NewFromString("arch=x86")
@@ -327,9 +249,9 @@
})
}
-func testQueryTracesByIndex_QueryHasUnknownParamReturnsNoError(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestQueryTracesByIndex_QueryHasUnknownParamReturnsNoError(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Query that has no matching params in the given tile.
q, err := query.NewFromString("arch=unknown")
@@ -339,8 +261,9 @@
assert.Nil(t, ts)
}
-func testQueryTracesByIndex_QueryAgainstTileWithNoDataReturnsNoError(t *testing.T, s *SQLTraceStore) {
- ctx := context.Background()
+func TestQueryTracesByIndex_QueryAgainstTileWithNoDataReturnsNoError(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
// Query that has no Postings for the given tile.
q, err := query.NewFromString("arch=unknown")
@@ -350,9 +273,9 @@
assert.Nil(t, ts)
}
-func testTraceCount(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestTraceCount(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
count, err := s.TraceCount(ctx, 0)
assert.NoError(t, err)
@@ -367,79 +290,65 @@
assert.Equal(t, int64(0), count)
}
-func testParamSetForTile(t *testing.T, s *SQLTraceStore) {
- const tileNumber types.TileNumber = 1
- _, err := s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=x86,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=565,arch=arm,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=arm64,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=gpu,arch=x86_64,"), tileNumber)
- assert.NoError(t, err)
+func TestParamSetForTile(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
ps, err := s.paramSetForTile(1)
assert.NoError(t, err)
expected := paramtools.ParamSet{
- "arch": []string{"arm", "arm64", "x86", "x86_64"},
- "config": []string{"565", "8888", "gpu"},
+ "arch": []string{"x86"},
+ "config": []string{"565", "8888"},
}
assert.Equal(t, expected, ps)
}
-func testParamSetForTile_Empty(t *testing.T, s *SQLTraceStore) {
+func TestParamSetForTile_Empty(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
+
// Test the empty case where there is no data in the store.
ps, err := s.paramSetForTile(1)
assert.NoError(t, err)
assert.Equal(t, paramtools.ParamSet{}, ps)
}
-func testGetLatestTile(t *testing.T, s *SQLTraceStore) {
- _, err := s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=x86,"), types.TileNumber(1))
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=arm64,"), types.TileNumber(5))
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=gpu,arch=x86_64,"), types.TileNumber(7))
- assert.NoError(t, err)
+func TestGetLatestTile(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
tileNumber, err := s.GetLatestTile()
assert.NoError(t, err)
- assert.Equal(t, types.TileNumber(7), tileNumber)
+ assert.Equal(t, types.TileNumber(1), tileNumber)
}
-func testGetLatestTile_Empty(t *testing.T, s *SQLTraceStore) {
+func TestGetLatestTile_Empty(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
+
// Test the empty case where there is no data in datastore.
tileNumber, err := s.GetLatestTile()
assert.Error(t, err)
assert.Equal(t, types.BadTileNumber, tileNumber)
}
-func testGetOrderedParamSet(t *testing.T, s *SQLTraceStore) {
- ctx := context.Background()
-
- const tileNumber types.TileNumber = 1
- // Now add some trace ids.
- _, err := s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=x86,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=565,arch=arm,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=arm64,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=gpu,arch=x86_64,"), tileNumber)
- assert.NoError(t, err)
+func TestGetOrderedParamSet(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
ops, err := s.GetOrderedParamSet(ctx, 1)
assert.NoError(t, err)
expected := paramtools.ParamSet{
- "arch": []string{"arm", "arm64", "x86", "x86_64"},
- "config": []string{"565", "8888", "gpu"},
+ "arch": []string{"x86"},
+ "config": []string{"565", "8888"},
}
assert.Equal(t, expected, ops.ParamSet)
assert.Equal(t, []string{"arch", "config"}, ops.KeyOrder)
}
-func testGetOrderedParamSet_Empty(t *testing.T, s *SQLTraceStore) {
- ctx := context.Background()
+func TestGetOrderedParamSet_Empty(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
// Test the empty case where there is no data in datastore.
ops, err := s.GetOrderedParamSet(ctx, 1)
@@ -447,27 +356,20 @@
assert.Equal(t, paramtools.ParamSet{}, ops.ParamSet)
}
-func testCountIndices(t *testing.T, s *SQLTraceStore) {
- ctx := context.Background()
-
- const tileNumber types.TileNumber = 1
- // Now add some trace ids.
- _, err := s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=x86,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=565,arch=arm,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=8888,arch=arm64,"), tileNumber)
- assert.NoError(t, err)
- _, err = s.writeTraceIDAndPostings(paramtools.NewParams(",config=gpu,arch=x86_64,"), tileNumber)
- assert.NoError(t, err)
+func TestCountIndices(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
count, err := s.CountIndices(ctx, 1)
assert.NoError(t, err)
- assert.Equal(t, int64(8), count)
+
+ // The always returns 0.
+ assert.Equal(t, int64(0), count)
}
-func testCountIndices_Empty(t *testing.T, s *SQLTraceStore) {
- ctx := context.Background()
+func TestCountIndices_Empty(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
// Test the empty case where there is no data in datastore.
count, err := s.CountIndices(ctx, 1)
@@ -475,17 +377,18 @@
assert.Equal(t, int64(0), count)
}
-func testGetSource(t *testing.T, s *SQLTraceStore) {
- populatedTestDB(t, s)
- ctx := context.Background()
+func TestGetSource(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
filename, err := s.GetSource(ctx, types.CommitNumber(2), ",arch=x86,config=8888,")
- assert.NoError(t, err)
+ require.NoError(t, err)
assert.Equal(t, "gs://perf-bucket/2020/02/08/12/testdata.json", filename)
}
-func testGetSource_Empty(t *testing.T, s *SQLTraceStore) {
- ctx := context.Background()
+func TestGetSource_Empty(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, true)
+ defer cleanup()
// Confirm the call works with an empty tracestore.
filename, err := s.GetSource(ctx, types.CommitNumber(5), ",arch=x86,config=8888,")
@@ -493,6 +396,29 @@
assert.Equal(t, "", filename)
}
+func TestSQLTraceStore_TileNumber(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
+
+ assert.Equal(t, types.TileNumber(0), s.TileNumber(types.CommitNumber(1)))
+ assert.Equal(t, types.TileNumber(1), s.TileNumber(types.CommitNumber(9)))
+}
+
+func TestSQLTraceStore_TileSize(t *testing.T) {
+ _, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
+
+ assert.Equal(t, testTileSize, s.TileSize())
+}
+
+func TestSQLTraceStore_WriteIndices_Success(t *testing.T) {
+ ctx, s, cleanup := commonTestSetup(t, false)
+ defer cleanup()
+
+ // WriteIndices is a no-op in the SQL backed trace store.
+ assert.NoError(t, s.WriteIndices(ctx, types.TileNumber(1)))
+}
+
func TestCommitNumberOfTileStart(t *testing.T) {
unittest.SmallTest(t)
s := &SQLTraceStore{
@@ -530,36 +456,16 @@
require.NoError(t, err)
}
-// subTestFunction is a func we will call to test one aspect of *SQLTraceStore.
-type subTestFunction func(t *testing.T, s *SQLTraceStore)
-
-// subTests are all the tests we have for *SQLTraceStore.
-var subTests = map[string]subTestFunction{
- "testUpdateSourceFile": testUpdateSourceFile,
- "testWriteTraceIDAndPostings": testWriteTraceIDAndPostings,
- "testParamSetForTile": testParamSetForTile,
- "testParamSetForTile_Empty": testParamSetForTile_Empty,
- "testGetLatestTile": testGetLatestTile,
- "testGetLatestTile_Empty": testGetLatestTile_Empty,
- "testGetOrderedParamSet": testGetOrderedParamSet,
- "testGetOrderedParamSet_Empty": testGetOrderedParamSet_Empty,
- "testCountIndices": testCountIndices,
- "testCountIndices_Empty": testCountIndices_Empty,
- "testGetSource_Empty": testGetSource_Empty,
- "testReadTraces": testReadTraces,
- "testWriteTraces_MultipleBatches_Success": testWriteTraces_MultipleBatches_Success,
- "testReadTraces_InvalidKey": testReadTraces_InvalidKey,
- "testReadTraces_NoResults": testReadTraces_NoResults,
- "testReadTraces_EmptyTileReturnsNoData": testReadTraces_EmptyTileReturnsNoData,
- "testQueryTracesIDOnlyByIndex_EmptyQueryReturnsError": testQueryTracesIDOnlyByIndex_EmptyQueryReturnsError,
- "testQueryTracesIDOnlyByIndex_EmptyTileReturnsEmptyParamset": testQueryTracesIDOnlyByIndex_EmptyTileReturnsEmptyParamset,
- "testQueryTracesIDOnlyByIndex_MatchesOneTrace": testQueryTracesIDOnlyByIndex_MatchesOneTrace,
- "testQueryTracesIDOnlyByIndex_MatchesTwoTraces": testQueryTracesIDOnlyByIndex_MatchesTwoTraces,
- "testQueryTracesByIndex_MatchesOneTrace": testQueryTracesByIndex_MatchesOneTrace,
- "testQueryTracesByIndex_MatchesOneTraceInTheSecondTile": testQueryTracesByIndex_MatchesOneTraceInTheSecondTile,
- "testQueryTracesByIndex_MatchesTwoTraces": testQueryTracesByIndex_MatchesTwoTraces,
- "testQueryTracesByIndex_QueryHasUnknownParamReturnsNoError": testQueryTracesByIndex_QueryHasUnknownParamReturnsNoError,
- "testQueryTracesByIndex_QueryAgainstTileWithNoDataReturnsNoError": testQueryTracesByIndex_QueryAgainstTileWithNoDataReturnsNoError,
- "testTraceCount": testTraceCount,
- "testGetSource": testGetSource,
+func Test_traceIDForSQLFromTraceName_Success(t *testing.T) {
+ unittest.SmallTest(t)
+ /*
+ $ python3
+ Python 3.7.3 (default, Dec 20 2019, 18:57:59)
+ [GCC 8.3.0] on linux
+ Type "help", "copyright", "credits" or "license" for more information.
+ >>> import hashlib
+ >>> hashlib.md5(b",arch=x86,config=8888,").hexdigest()
+ 'fe385b159ff55dca481069805e5ff050'
+ */
+ assert.Equal(t, traceIDForSQL(`\xfe385b159ff55dca481069805e5ff050`), traceIDForSQLFromTraceName(",arch=x86,config=8888,"))
}
diff --git a/perf/go/tracestore/tracestore.go b/perf/go/tracestore/tracestore.go
index b3df325..4e082b1 100644
--- a/perf/go/tracestore/tracestore.go
+++ b/perf/go/tracestore/tracestore.go
@@ -40,6 +40,7 @@
// QueryTracesIDOnlyByIndex returns a stream of ParamSets that match the
// given query.
+ // TODO(jcgregorio) Change to just return count and ParamSet.
QueryTracesIDOnlyByIndex(ctx context.Context, tileNumber types.TileNumber, q *query.Query) (<-chan paramtools.Params, error)
// ReadTraces loads the traces for the given trace keys.
diff --git a/perf/migrations/cdb.sql b/perf/migrations/cdb.sql
index cfd8b76..abbe312 100644
--- a/perf/migrations/cdb.sql
+++ b/perf/migrations/cdb.sql
@@ -146,7 +146,7 @@
TraceNames
INNER JOIN TraceValues2 ON TraceValues2.trace_id = TraceNames.trace_id
WHERE
- TraceNames.params ->> 'arch' IN ('x86')
- AND TraceNames.params ->> 'config' IN ('565', '8888')
- AND TraceValues2.commit_number >= 0
- AND TraceValues2.commit_number < 255;
\ No newline at end of file
+ TraceValues2.commit_number >= 0
+ AND TraceValues2.commit_number < 255
+ AND TraceNames.params ->> 'arch' IN ('x86')
+ AND TraceNames.params ->> 'config' IN ('565', '8888');
\ No newline at end of file
diff --git a/perf/migrations/cockroachdb/0002_create_initial_tables.up.sql b/perf/migrations/cockroachdb/0002_create_initial_tables.up.sql
index 1614a69..30452d6 100644
--- a/perf/migrations/cockroachdb/0002_create_initial_tables.up.sql
+++ b/perf/migrations/cockroachdb/0002_create_initial_tables.up.sql
@@ -17,5 +17,5 @@
val REAL,
-- Id of the source filename, from SourceFiles.
source_file_id INT,
- PRIMARY KEY (trace_id, commit_number)
+ PRIMARY KEY (commit_number, trace_id)
);
\ No newline at end of file
