feat/clusterdb
This commit is contained in:
Andrew Jackson (Ajax)
parent
017f20c346
commit
48ebd1330c
@ -557,6 +557,12 @@ workflows:
|
||||
- build
|
||||
suite: itest-cli
|
||||
target: "./itests/cli_test.go"
|
||||
- test:
|
||||
name: test-itest-clusterdb
|
||||
requires:
|
||||
- build
|
||||
suite: itest-clusterdb
|
||||
target: "./itests/clusterdb_test.go"
|
||||
- test:
|
||||
name: test-itest-deadlines
|
||||
requires:
|
||||
|
||||
@ -894,3 +894,43 @@
|
||||
#GCInterval = "1m0s"
|
||||
|
||||
|
||||
[ClusterDB]
|
||||
# HOSTS is a list of hostnames to nodes running YugabyteDB
|
||||
# in a cluster. Only 1 is required
|
||||
#
|
||||
# type: []string
|
||||
# env var: LOTUS_CLUSTERDB_HOSTS
|
||||
#Hosts = ["127.0.0.1"]
|
||||
|
||||
# The Yugabyte server's username with full credentials to operate on Lotus' Database. Blank for default.
|
||||
#
|
||||
# type: string
|
||||
# env var: LOTUS_CLUSTERDB_USERNAME
|
||||
#Username = "yugabyte"
|
||||
|
||||
# The password for the related username. Blank for default.
|
||||
#
|
||||
# type: string
|
||||
# env var: LOTUS_CLUSTERDB_PASSWORD
|
||||
#Password = "yugabyte"
|
||||
|
||||
# The database (logical partition) within Yugabyte. Blank for default.
|
||||
#
|
||||
# type: string
|
||||
# env var: LOTUS_CLUSTERDB_DATABASE
|
||||
#Database = "yugabyte"
|
||||
|
||||
# The port to find Yugabyte. Blank for default.
|
||||
#
|
||||
# type: string
|
||||
# env var: LOTUS_CLUSTERDB_PORT
|
||||
#Port = "5433"
|
||||
|
||||
# ITest is for optimized integration testing and not
|
||||
# for production. Blank for default production configuration.
|
||||
#
|
||||
# type: string
|
||||
# env var: LOTUS_CLUSTERDB_ITEST
|
||||
#ITest = ""
|
||||
|
||||
|
||||
|
||||
12
go.mod
12
go.mod
@ -62,6 +62,7 @@ require (
|
||||
github.com/filecoin-project/test-vectors/schema v0.0.5
|
||||
github.com/gbrlsnchs/jwt/v3 v3.0.1
|
||||
github.com/gdamore/tcell/v2 v2.2.0
|
||||
github.com/georgysavva/scany/v2 v2.0.0
|
||||
github.com/go-openapi/spec v0.19.11
|
||||
github.com/golang/mock v1.6.0
|
||||
github.com/google/uuid v1.3.0
|
||||
@ -102,8 +103,10 @@ require (
|
||||
github.com/ipld/go-ipld-selector-text-lite v0.0.1
|
||||
github.com/ipni/go-libipni v0.0.8
|
||||
github.com/ipni/index-provider v0.12.0
|
||||
github.com/jackc/pgx/v5 v5.4.1
|
||||
github.com/kelseyhightower/envconfig v1.4.0
|
||||
github.com/koalacxr/quantile v0.0.1
|
||||
github.com/kr/pretty v0.3.1
|
||||
github.com/libp2p/go-buffer-pool v0.1.0
|
||||
github.com/libp2p/go-libp2p v0.27.6
|
||||
github.com/libp2p/go-libp2p-consensus v0.0.1
|
||||
@ -127,7 +130,7 @@ require (
|
||||
github.com/multiformats/go-varint v0.0.7
|
||||
github.com/open-rpc/meta-schema v0.0.0-20201029221707-1b72ef2ea333
|
||||
github.com/polydawn/refmt v0.89.0
|
||||
github.com/prometheus/client_golang v1.14.0
|
||||
github.com/prometheus/client_golang v1.16.0
|
||||
github.com/puzpuzpuz/xsync/v2 v2.4.0
|
||||
github.com/raulk/clock v1.1.0
|
||||
github.com/raulk/go-watchdog v1.3.0
|
||||
@ -243,6 +246,9 @@ require (
|
||||
github.com/ipfs/go-verifcid v0.0.2 // indirect
|
||||
github.com/ipld/go-ipld-adl-hamt v0.0.0-20220616142416-9004dbd839e0 // indirect
|
||||
github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.0 // indirect
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
|
||||
github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c // indirect
|
||||
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
|
||||
@ -255,6 +261,7 @@ require (
|
||||
github.com/klauspost/compress v1.16.5 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.5 // indirect
|
||||
github.com/koron/go-ssdp v0.0.4 // indirect
|
||||
github.com/kr/text v0.2.0 // indirect
|
||||
github.com/libp2p/go-cidranger v1.1.0 // indirect
|
||||
github.com/libp2p/go-flow-metrics v0.1.0 // indirect
|
||||
github.com/libp2p/go-libp2p-asn-util v0.3.0 // indirect
|
||||
@ -291,7 +298,7 @@ require (
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/prometheus/client_model v0.4.0 // indirect
|
||||
github.com/prometheus/common v0.42.0 // indirect
|
||||
github.com/prometheus/procfs v0.9.0 // indirect
|
||||
github.com/prometheus/procfs v0.10.1 // indirect
|
||||
github.com/prometheus/statsd_exporter v0.22.7 // indirect
|
||||
github.com/quic-go/qpack v0.4.0 // indirect
|
||||
github.com/quic-go/qtls-go1-19 v0.3.2 // indirect
|
||||
@ -299,6 +306,7 @@ require (
|
||||
github.com/quic-go/quic-go v0.33.0 // indirect
|
||||
github.com/quic-go/webtransport-go v0.5.3 // indirect
|
||||
github.com/rivo/uniseg v0.1.0 // indirect
|
||||
github.com/rogpeppe/go-internal v1.9.0 // indirect
|
||||
github.com/rs/cors v1.7.0 // indirect
|
||||
github.com/russross/blackfriday/v2 v2.1.0 // indirect
|
||||
github.com/shirou/gopsutil v2.18.12+incompatible // indirect
|
||||
|
||||
24
go.sum
24
go.sum
@ -169,6 +169,7 @@ github.com/circonus-labs/circonusllhist v0.1.3/go.mod h1:kMXHVDlOchFAehlya5ePtbp
|
||||
github.com/clbanning/x2j v0.0.0-20191024224557-825249438eec/go.mod h1:jMjuTZXRI4dUb/I5gc9Hdhagfvm9+RyrPryS/auMzxE=
|
||||
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
|
||||
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
|
||||
github.com/cockroachdb/cockroach-go/v2 v2.2.0 h1:/5znzg5n373N/3ESjHF5SMLxiW4RKB05Ql//KWfeTFs=
|
||||
github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa/go.mod h1:zn76sxSg3SzpJ0PPJaLDCu+Bu0Lg3sKTORVIj19EIF8=
|
||||
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd/go.mod h1:sE/e/2PUdi/liOCUjSTXgM1o87ZssimdTWN964YiIeI=
|
||||
github.com/codegangsta/cli v1.20.0/go.mod h1:/qJNoX69yVSKu5o4jLyXAENLRyk1uhi7zkbQ3slBdOA=
|
||||
@ -391,6 +392,8 @@ github.com/gdamore/encoding v1.0.0 h1:+7OoQ1Bc6eTm5niUzBa0Ctsh6JbMW6Ra+YNuAtDBdk
|
||||
github.com/gdamore/encoding v1.0.0/go.mod h1:alR0ol34c49FCSBLjhosxzcPHQbf2trDkoo5dl+VrEg=
|
||||
github.com/gdamore/tcell/v2 v2.2.0 h1:vSyEgKwraXPSOkvCk7IwOSyX+Pv3V2cV9CikJMXg4U4=
|
||||
github.com/gdamore/tcell/v2 v2.2.0/go.mod h1:cTTuF84Dlj/RqmaCIV5p4w8uG1zWdk0SF6oBpwHp4fU=
|
||||
github.com/georgysavva/scany/v2 v2.0.0 h1:RGXqxDv4row7/FYoK8MRXAZXqoWF/NM+NP0q50k3DKU=
|
||||
github.com/georgysavva/scany/v2 v2.0.0/go.mod h1:sigOdh+0qb/+aOs3TVhehVT10p8qJL7K/Zhyz8vWo38=
|
||||
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
|
||||
github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
|
||||
github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
|
||||
@ -465,6 +468,7 @@ github.com/godbus/dbus/v5 v5.0.3/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x
|
||||
github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
|
||||
github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk=
|
||||
github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
|
||||
github.com/gofrs/flock v0.8.1 h1:+gYjHKf32LDeiEEFhQaotPbLuUXjY5ZqxKgXy7n59aw=
|
||||
github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
|
||||
github.com/gogo/googleapis v1.4.1 h1:1Yx4Myt7BxzvUr5ldGSbwYiZG6t9wGBZ+8/fX3Wvtq0=
|
||||
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
|
||||
@ -867,6 +871,14 @@ github.com/ipni/index-provider v0.12.0 h1:R3F6dxxKNv4XkE4GJZNLOG0bDEbBQ/S5iztXwS
|
||||
github.com/ipni/index-provider v0.12.0/go.mod h1:GhyrADJp7n06fqoc1djzkvL4buZYHzV8SoWrlxEo5F4=
|
||||
github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52 h1:QG4CGBqCeuBo6aZlGAamSkxWdgWfZGeE49eUOWJPA4c=
|
||||
github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52/go.mod h1:fdg+/X9Gg4AsAIzWpEHwnqd+QY3b7lajxyjE1m4hkq4=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
|
||||
github.com/jackc/pgx/v5 v5.4.1 h1:oKfB/FhuVtit1bBM3zNRRsZ925ZkMN3HXL+LgLUM9lE=
|
||||
github.com/jackc/pgx/v5 v5.4.1/go.mod h1:q6iHT8uDNXWiFNOlRqJzBTaSH3+2xCXkokxHZC5qWFY=
|
||||
github.com/jackc/puddle/v2 v2.2.0 h1:RdcDk92EJBuBS55nQMMYFXTxwstHug4jkhT5pq8VxPk=
|
||||
github.com/jackc/puddle/v2 v2.2.0/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
|
||||
github.com/jackpal/gateway v1.0.5/go.mod h1:lTpwd4ACLXmpyiCTRtfiNyVnUmqT9RivzCDQetPfnjA=
|
||||
github.com/jackpal/go-nat-pmp v1.0.1/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
|
||||
github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus=
|
||||
@ -950,6 +962,7 @@ github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfn
|
||||
github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
|
||||
github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
|
||||
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/pty v1.1.3/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA=
|
||||
@ -958,6 +971,7 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
||||
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
|
||||
github.com/leodido/go-urn v1.2.0 h1:hpXL4XnriNwQ/ABnpepYM/1vCLWNDfUNts8dX3xTG6Y=
|
||||
github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
|
||||
github.com/lib/pq v1.10.0 h1:Zx5DJFEYQXio93kgXnQ09fXNiUKsqv4OUEu2UtGcB1E=
|
||||
github.com/libp2p/go-addr-util v0.0.1/go.mod h1:4ac6O7n9rIAKB1dnd+s8IbbMXkt+oBpzX4/+RACcnlQ=
|
||||
github.com/libp2p/go-addr-util v0.0.2/go.mod h1:Ecd6Fb3yIuLzq4bD7VcywcVSBtefcAwnUISBM3WG15E=
|
||||
github.com/libp2p/go-buffer-pool v0.0.1/go.mod h1:xtyIz9PMobb13WaxR6Zo1Pd1zXJKYg0a8KiIvDp3TzQ=
|
||||
@ -1398,6 +1412,7 @@ github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 h1:1/WtZae0yGtPq+TI6+
|
||||
github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9/go.mod h1:x3N5drFsm2uilKKuuYo6LdyD8vZAW55sH/9w+pbo1sw=
|
||||
github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
|
||||
github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
|
||||
github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
|
||||
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
@ -1426,8 +1441,8 @@ github.com/prometheus/client_golang v1.11.0/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqr
|
||||
github.com/prometheus/client_golang v1.12.1/go.mod h1:3Z9XVyYiZYEO+YQWt3RD2R3jrbd179Rt297l4aS6nDY=
|
||||
github.com/prometheus/client_golang v1.12.2/go.mod h1:3Z9XVyYiZYEO+YQWt3RD2R3jrbd179Rt297l4aS6nDY=
|
||||
github.com/prometheus/client_golang v1.13.0/go.mod h1:vTeo+zgvILHsnnj/39Ou/1fPN5nJFOEMgftOUOmlvYQ=
|
||||
github.com/prometheus/client_golang v1.14.0 h1:nJdhIvne2eSX/XRAFV9PcvFFRbrjbcTUj0VP62TMhnw=
|
||||
github.com/prometheus/client_golang v1.14.0/go.mod h1:8vpkKitgIVNcqrRBWh1C4TIUQgYNtG/XQE4E/Zae36Y=
|
||||
github.com/prometheus/client_golang v1.16.0 h1:yk/hx9hDbrGHovbci4BY+pRMfSuuat626eFsHb7tmT8=
|
||||
github.com/prometheus/client_golang v1.16.0/go.mod h1:Zsulrv/L9oM40tJ7T815tM89lFEugiJ9HzIqaAx4LKc=
|
||||
github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
|
||||
github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
|
||||
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
|
||||
@ -1461,8 +1476,8 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
|
||||
github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
|
||||
github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
|
||||
github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
|
||||
github.com/prometheus/procfs v0.9.0 h1:wzCHvIvM5SxWqYvwgVL7yJY8Lz3PKn49KQtpgMYJfhI=
|
||||
github.com/prometheus/procfs v0.9.0/go.mod h1:+pB4zwohETzFnmlpe6yd2lSc+0/46IYZRB/chUwxUZY=
|
||||
github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+PymziUAg=
|
||||
github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
|
||||
github.com/prometheus/statsd_exporter v0.22.7 h1:7Pji/i2GuhK6Lu7DHrtTkFmNBCudCPT1pX2CziuyQR0=
|
||||
github.com/prometheus/statsd_exporter v0.22.7/go.mod h1:N/TevpjkIh9ccs6nuzY3jQn9dFqnUakOjnEuMPJJJnI=
|
||||
github.com/puzpuzpuz/xsync/v2 v2.4.0 h1:5sXAMHrtx1bg9nbRZTOn8T4MkWe5V+o8yKRH02Eznag=
|
||||
@ -1489,6 +1504,7 @@ github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR
|
||||
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
|
||||
github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
|
||||
github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
|
||||
github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
|
||||
github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
|
||||
github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
|
||||
github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
|
||||
|
||||
188
itests/clusterdb_test.go
Normal file
188
itests/clusterdb_test.go
Normal file
@ -0,0 +1,188 @@
|
||||
package itests
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/filecoin-project/lotus/itests/kit"
|
||||
"github.com/filecoin-project/lotus/lib/sturdy/clusterdb"
|
||||
"github.com/filecoin-project/lotus/node"
|
||||
"github.com/filecoin-project/lotus/node/config"
|
||||
)
|
||||
|
||||
func staticConfig() config.ClusterDB {
|
||||
return config.ClusterDB{
|
||||
Hosts: []string{"127.0.0.1"},
|
||||
Username: "yugabyte",
|
||||
Password: "yugabyte",
|
||||
Database: "yugabyte",
|
||||
Port: "5433",
|
||||
ITest: clusterdb.ITestNewID(),
|
||||
}
|
||||
}
|
||||
func withSetup(t *testing.T, f func(*kit.TestMiner)) {
|
||||
_, miner, _ := kit.EnsembleMinimal(t,
|
||||
kit.LatestActorsAt(-1),
|
||||
kit.MockProofs(),
|
||||
kit.ConstructorOpts(
|
||||
node.Override(new(config.ClusterDB), func() config.ClusterDB {
|
||||
return staticConfig()
|
||||
}),
|
||||
node.Override(new(*clusterdb.DB), clusterdb.NewFromConfig), //Why does this not work?
|
||||
),
|
||||
)
|
||||
|
||||
var err error
|
||||
miner.ClusterDB, err = clusterdb.NewFromConfig(staticConfig())
|
||||
if err != nil {
|
||||
t.Fatal("Yugabyte connection error:", err)
|
||||
}
|
||||
|
||||
defer miner.ClusterDB.ITestDeleteAll()
|
||||
f(miner)
|
||||
}
|
||||
|
||||
func TestCrud(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
withSetup(t, func(miner *kit.TestMiner) {
|
||||
err := miner.ClusterDB.Exec(ctx, `
|
||||
INSERT INTO
|
||||
itest_scratch (some_int, content)
|
||||
VALUES
|
||||
(11, 'cows'),
|
||||
(5, 'cats')
|
||||
`)
|
||||
if err != nil {
|
||||
t.Fatal("Could not insert: ", err)
|
||||
}
|
||||
var ints []struct {
|
||||
Count int `db:"some_int"`
|
||||
Animal string `db:"content"`
|
||||
Unpopulated int
|
||||
}
|
||||
err = miner.ClusterDB.Select(ctx, &ints, "SELECT content, some_int FROM itest_scratch")
|
||||
if err != nil {
|
||||
t.Fatal("Could not select: ", err)
|
||||
}
|
||||
if len(ints) != 2 {
|
||||
t.Fatal("unexpected count of returns. Want 2, Got ", len(ints))
|
||||
}
|
||||
if ints[0].Count != 11 || ints[1].Count != 5 {
|
||||
t.Fatal("expected [11,5] got ", ints)
|
||||
}
|
||||
if ints[0].Animal != "cows" || ints[1].Animal != "cats" {
|
||||
t.Fatal("expected, [cows, cats] ", ints)
|
||||
}
|
||||
fmt.Println("test completed")
|
||||
})
|
||||
}
|
||||
|
||||
func TestTransaction(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
withSetup(t, func(miner *kit.TestMiner) {
|
||||
if err := miner.ClusterDB.Exec(ctx, "INSERT INTO itest_scratch (some_int) VALUES (4), (5), (6)"); err != nil {
|
||||
t.Fatal("E0", err)
|
||||
}
|
||||
miner.ClusterDB.BeginTransaction(ctx, func(tx *clusterdb.Transaction) (commit bool) {
|
||||
if err := tx.Exec(ctx, "INSERT INTO itest_scratch (some_int) VALUES (7), (8), (9)"); err != nil {
|
||||
t.Fatal("E1", err)
|
||||
}
|
||||
|
||||
// sum1 is read from OUTSIDE the transaction so it's the old value
|
||||
var sum1 int
|
||||
if err := miner.ClusterDB.QueryRow(ctx, "SELECT SUM(some_int) FROM itest_scratch").Scan(&sum1); err != nil {
|
||||
t.Fatal("E2", err)
|
||||
}
|
||||
if sum1 != 4+5+6 {
|
||||
t.Fatal("Expected 15, got ", sum1)
|
||||
}
|
||||
|
||||
// sum2 is from INSIDE the transaction, so the updated value.
|
||||
var sum2 int
|
||||
if err := tx.QueryRow(ctx, "SELECT SUM(some_int) FROM itest_scratch").Scan(&sum2); err != nil {
|
||||
t.Fatal("E3", err)
|
||||
}
|
||||
if sum2 != 4+5+6+7+8+9 {
|
||||
t.Fatal("Expected 39, got ", sum2)
|
||||
}
|
||||
return false // rollback
|
||||
})
|
||||
|
||||
var sum2 int
|
||||
// Query() example (yes, QueryRow would be preferred here)
|
||||
q, err := miner.ClusterDB.Query(ctx, "SELECT SUM(some_int) FROM itest_scratch")
|
||||
if err != nil {
|
||||
t.Fatal("E4", err)
|
||||
}
|
||||
defer q.Close()
|
||||
var rowCt int
|
||||
for q.Next() {
|
||||
q.Scan(&sum2)
|
||||
rowCt++
|
||||
}
|
||||
if sum2 != 4+5+6 {
|
||||
t.Fatal("Expected 15, got ", sum2)
|
||||
}
|
||||
if rowCt != 1 {
|
||||
t.Fatal("unexpected count of rows")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestPartialWalk(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
withSetup(t, func(miner *kit.TestMiner) {
|
||||
if err := miner.ClusterDB.Exec(ctx, `
|
||||
INSERT INTO
|
||||
itest_scratch (content, some_int)
|
||||
VALUES
|
||||
('andy was here', 5),
|
||||
('lotus is awesome', 6),
|
||||
('hello world', 7),
|
||||
('3rd integration test', 8),
|
||||
('fiddlesticks', 9)
|
||||
`); err != nil {
|
||||
t.Fatal("e1", err)
|
||||
}
|
||||
|
||||
// TASK: FIND THE ID of the string with a specific SHA256
|
||||
needle := "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"
|
||||
q, err := miner.ClusterDB.Query(ctx, `SELECT id, content FROM itest_scratch`)
|
||||
if err != nil {
|
||||
t.Fatal("e2", err)
|
||||
}
|
||||
defer q.Close()
|
||||
|
||||
var tmp struct {
|
||||
Src string `db:"content"`
|
||||
ID int
|
||||
}
|
||||
|
||||
var done bool
|
||||
for q.Next() {
|
||||
|
||||
if err := q.StructScan(&tmp); err != nil {
|
||||
t.Fatal("")
|
||||
}
|
||||
|
||||
bSha := sha256.Sum256([]byte(tmp.Src))
|
||||
if hex.EncodeToString(bSha[:]) == needle {
|
||||
done = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !done {
|
||||
t.Fatal("We didn't find it.")
|
||||
}
|
||||
// Answer: tmp.ID
|
||||
})
|
||||
}
|
||||
@ -49,6 +49,7 @@ import (
|
||||
"github.com/filecoin-project/lotus/cmd/lotus-worker/sealworker"
|
||||
"github.com/filecoin-project/lotus/gateway"
|
||||
"github.com/filecoin-project/lotus/genesis"
|
||||
"github.com/filecoin-project/lotus/lib/sturdy/clusterdb"
|
||||
"github.com/filecoin-project/lotus/markets/idxprov"
|
||||
"github.com/filecoin-project/lotus/markets/idxprov/idxprov_test"
|
||||
lotusminer "github.com/filecoin-project/lotus/miner"
|
||||
@ -763,6 +764,10 @@ func (n *Ensemble) Start() *Ensemble {
|
||||
)
|
||||
}
|
||||
|
||||
opts = append(opts, node.Override(new(*clusterdb.DB), func(cfg config.ClusterDB) (*clusterdb.DB, error) {
|
||||
return clusterdb.NewFromConfig(cfg)
|
||||
}))
|
||||
|
||||
// start node
|
||||
stop, err := node.New(ctx, opts...)
|
||||
require.NoError(n.t, err)
|
||||
|
||||
@ -24,6 +24,7 @@ import (
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/wallet/key"
|
||||
"github.com/filecoin-project/lotus/lib/sturdy/clusterdb"
|
||||
"github.com/filecoin-project/lotus/miner"
|
||||
sealing "github.com/filecoin-project/lotus/storage/pipeline"
|
||||
"github.com/filecoin-project/lotus/storage/sealer/storiface"
|
||||
@ -87,6 +88,8 @@ type TestMiner struct {
|
||||
RemoteListener net.Listener
|
||||
|
||||
options nodeOpts
|
||||
|
||||
ClusterDB *clusterdb.DB
|
||||
}
|
||||
|
||||
func (tm *TestMiner) PledgeSectors(ctx context.Context, n, existing int, blockNotif <-chan struct{}) {
|
||||
|
||||
251
lib/sturdy/clusterdb/clusterdb.go
Normal file
251
lib/sturdy/clusterdb/clusterdb.go
Normal file
@ -0,0 +1,251 @@
|
||||
package clusterdb
|
||||
|
||||
// TODO 2 - build integration tests
|
||||
|
||||
import (
|
||||
"context"
|
||||
"embed"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
logging "github.com/ipfs/go-log/v2"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/filecoin-project/lotus/node/config"
|
||||
)
|
||||
|
||||
// ItestNewID see ITestWithID doc
|
||||
func ITestNewID() string {
|
||||
return strconv.Itoa(rand.Intn(99999))
|
||||
}
|
||||
|
||||
type DB struct {
|
||||
pgx *pgxpool.Pool
|
||||
cfg *pgxpool.Config
|
||||
schema string
|
||||
hostnames []string
|
||||
log func(string)
|
||||
}
|
||||
|
||||
var logger = logging.Logger("sturdydb")
|
||||
|
||||
// NewFromConfig is a convenience function.
|
||||
// In usage:
|
||||
//
|
||||
// db, err := NewFromConfig(config.SturdyDB) // in binary init
|
||||
func NewFromConfig(cfg config.ClusterDB) (*DB, error) {
|
||||
return New(
|
||||
cfg.Hosts,
|
||||
cfg.Username,
|
||||
cfg.Password,
|
||||
cfg.Database,
|
||||
cfg.Port,
|
||||
cfg.ITest,
|
||||
func(s string) { logger.Error(s) },
|
||||
)
|
||||
}
|
||||
|
||||
// New is to be called once per binary to establish the pool.
|
||||
// log() is for errors. It returns an upgraded database's connection.
|
||||
// This entry point serves both production and integration tests, so it's more DI.
|
||||
func New(hosts []string, username, password, database, port, itest string, log func(string)) (*DB, error) {
|
||||
connString := ""
|
||||
if len(hosts) > 0 {
|
||||
connString = "host=" + hosts[0] + " "
|
||||
}
|
||||
for k, v := range map[string]string{"user": username, "password": password, "dbname": database, "port": port} {
|
||||
if strings.TrimSpace(v) != "" {
|
||||
connString += k + "=" + v + " "
|
||||
}
|
||||
}
|
||||
|
||||
schema := "lotus"
|
||||
if itest != "" {
|
||||
schema = "itest_" + itest
|
||||
}
|
||||
|
||||
if err := ensureSchemaExists(connString, schema); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cfg, err := pgxpool.ParseConfig(connString + "search_path=" + schema)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// enable multiple fallback hosts.
|
||||
for _, h := range hosts[1:] {
|
||||
cfg.ConnConfig.Fallbacks = append(cfg.ConnConfig.Fallbacks, &pgconn.FallbackConfig{Host: h})
|
||||
}
|
||||
|
||||
cfg.ConnConfig.OnNotice = func(conn *pgconn.PgConn, n *pgconn.Notice) {
|
||||
log("database notice: " + n.Message + ": " + n.Detail)
|
||||
DBMeasures.Errors.M(1)
|
||||
}
|
||||
|
||||
db := DB{cfg: cfg, schema: schema, hostnames: hosts, log: log} // pgx populated in AddStatsAndConnect
|
||||
if err := db.addStatsAndConnect(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &db, db.upgrade()
|
||||
}
|
||||
|
||||
type tracer struct {
|
||||
}
|
||||
|
||||
type ctxkey string
|
||||
|
||||
var sqlStart = ctxkey("sqlStart")
|
||||
|
||||
func (t tracer) TraceQueryStart(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryStartData) context.Context {
|
||||
return context.WithValue(ctx, sqlStart, time.Now())
|
||||
}
|
||||
func (t tracer) TraceQueryEnd(ctx context.Context, conn *pgx.Conn, data pgx.TraceQueryEndData) {
|
||||
DBMeasures.Hits.M(1)
|
||||
ms := time.Since(ctx.Value(sqlStart).(time.Time)).Milliseconds()
|
||||
DBMeasures.TotalWait.M(ms)
|
||||
DBMeasures.Waits.Observe(float64(ms))
|
||||
if data.Err != nil {
|
||||
DBMeasures.Errors.M(1)
|
||||
}
|
||||
// Can log what type of query it is, but not what tables
|
||||
// Can log rows affected.
|
||||
}
|
||||
|
||||
// addStatsAndConnect connects a prometheus logger. Be sure to run this before using the DB.
|
||||
func (db *DB) addStatsAndConnect() error {
|
||||
|
||||
db.cfg.ConnConfig.Tracer = tracer{}
|
||||
|
||||
hostnameToIndex := map[string]float64{}
|
||||
for i, h := range db.hostnames {
|
||||
hostnameToIndex[h] = float64(i)
|
||||
}
|
||||
db.cfg.AfterConnect = func(ctx context.Context, c *pgx.Conn) error {
|
||||
s := db.pgx.Stat()
|
||||
DBMeasures.OpenConnections.M(int64(s.TotalConns()))
|
||||
DBMeasures.WhichHost.Observe(hostnameToIndex[c.Config().Host])
|
||||
|
||||
//FUTURE place for any connection seasoning
|
||||
return nil
|
||||
}
|
||||
|
||||
var err error
|
||||
db.pgx, err = pgxpool.NewWithConfig(context.Background(), db.cfg)
|
||||
if err != nil {
|
||||
db.log(fmt.Sprintf("Unable to connect to database: %v\n", err))
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ITestDeleteAll will delete everything created for "this" integration test.
|
||||
// This must be called at the end of each integration test.
|
||||
func (db *DB) ITestDeleteAll() {
|
||||
if !strings.HasPrefix(db.schema, "itest_") {
|
||||
fmt.Println("Warning: this should never be called on anything but an itest schema.")
|
||||
return
|
||||
}
|
||||
defer db.pgx.Close()
|
||||
_, err := db.pgx.Exec(context.Background(), "DROP SCHEMA "+db.schema+" CASCADE")
|
||||
if err != nil {
|
||||
fmt.Println("warning: unclean itest shutdown: cannot delete schema: " + err.Error())
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
var schemaREString = "^[A-Za-z0-9_]+$"
|
||||
var schemaRE = regexp.MustCompile(schemaREString)
|
||||
|
||||
func ensureSchemaExists(connString, schema string) error {
|
||||
// FUTURE allow using fallback DBs for start-up.
|
||||
p, err := pgx.Connect(context.Background(), connString)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer p.Close(context.Background())
|
||||
|
||||
if len(schema) < 5 || !schemaRE.MatchString(schema) {
|
||||
return errors.New("schema must be of the form " + schemaREString + "\n Got: " + schema)
|
||||
}
|
||||
_, err = p.Exec(context.Background(), "CREATE SCHEMA IF NOT EXISTS "+schema)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot create schema: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
//go:embed sql
|
||||
var fs embed.FS
|
||||
|
||||
func (db *DB) upgrade() error {
|
||||
// Does the version table exist? if not, make it.
|
||||
// NOTE: This cannot change except via the next sql file.
|
||||
err := db.Exec(context.Background(), `CREATE TABLE IF NOT EXISTS base (
|
||||
id SERIAL PRIMARY KEY,
|
||||
entry CHAR(12),
|
||||
applied TIMESTAMP DEFAULT current_timestamp
|
||||
)`)
|
||||
if err != nil {
|
||||
db.log("Upgrade failed.")
|
||||
return err
|
||||
}
|
||||
|
||||
// __Run scripts in order.__
|
||||
|
||||
landed := map[string]bool{}
|
||||
{
|
||||
var landedEntries []struct{ Entry string }
|
||||
err = db.Select(context.Background(), &landedEntries, "SELECT entry FROM base")
|
||||
if err != nil {
|
||||
db.log("Cannot read entries: " + err.Error())
|
||||
return err
|
||||
}
|
||||
for _, l := range landedEntries {
|
||||
landed[l.Entry] = true
|
||||
}
|
||||
}
|
||||
dir, err := fs.ReadDir("sql")
|
||||
if err != nil {
|
||||
db.log("Cannot read fs entries: " + err.Error())
|
||||
return err
|
||||
}
|
||||
sort.Slice(dir, func(i, j int) bool { return dir[i].Name() < dir[j].Name() })
|
||||
for _, e := range dir {
|
||||
name := e.Name()
|
||||
if landed[name] || !strings.HasSuffix(name, ".sql") {
|
||||
continue
|
||||
}
|
||||
file, err := fs.ReadFile("sql/" + name)
|
||||
if err != nil {
|
||||
db.log("weird embed file read err")
|
||||
return err
|
||||
}
|
||||
for _, s := range strings.Split(string(file), ";") { // Implement the changes.
|
||||
if len(strings.TrimSpace(s)) == 0 {
|
||||
continue
|
||||
}
|
||||
_, err = db.pgx.Exec(context.Background(), s)
|
||||
if err != nil {
|
||||
db.log(fmt.Sprintf("Could not upgrade! File %s, Query: %s, Returned: %s", name, s, err.Error()))
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Mark Completed.
|
||||
err = db.Exec(context.Background(), "INSERT INTO base (entry) VALUES ($1)", name)
|
||||
if err != nil {
|
||||
db.log("Cannot update base: " + err.Error())
|
||||
return fmt.Errorf("cannot insert into base: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
35
lib/sturdy/clusterdb/doc.go
Normal file
35
lib/sturdy/clusterdb/doc.go
Normal file
@ -0,0 +1,35 @@
|
||||
/*
|
||||
# Clusterdb provides database abstractions over SP-wide Postgres-compatible instance(s).
|
||||
|
||||
# Features
|
||||
|
||||
Rolling to secondary database servers on connection failure
|
||||
Convenience features for Go + SQL
|
||||
Prevention of SQL injection vulnerabilities
|
||||
Monitors behavior via Prometheus stats and logging of errors.
|
||||
|
||||
# Usage
|
||||
|
||||
Processes should use New() to instantiate a *DB and keep it.
|
||||
Consumers can use this *DB concurrently.
|
||||
Creating and changing tables & views should happen in ./sql/ folder.
|
||||
Name the file "today's date" in the format: YYYYMMDD.sql (ex: 20231231.sql for the year's last day)
|
||||
|
||||
a. CREATE TABLE should NOT have a schema:
|
||||
GOOD: CREATE TABLE foo ();
|
||||
BAD: CREATE TABLE me.foo ();
|
||||
b. Schema is managed for you. It provides isolation for integraton tests & multi-use.
|
||||
c. Git Merges: All run once, so old-after-new is OK when there are no deps.
|
||||
d. NEVER change shipped sql files. Have later files make corrections.
|
||||
e. Anything not ran will be ran, so an older date making it to master is OK.
|
||||
|
||||
Write SQL with context, raw strings, and args:
|
||||
|
||||
name := "Alice"
|
||||
var ID int
|
||||
err := QueryRow(ctx, "SELECT id from people where first_name=?", name).Scan(&ID)
|
||||
fmt.Println(ID)
|
||||
|
||||
Note: Scan() is column-oriented, while Select() & StructScan() is field name/tag oriented.
|
||||
*/
|
||||
package clusterdb
|
||||
70
lib/sturdy/clusterdb/metrics.go
Normal file
70
lib/sturdy/clusterdb/metrics.go
Normal file
@ -0,0 +1,70 @@
|
||||
package clusterdb
|
||||
|
||||
import (
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.opencensus.io/stats"
|
||||
"go.opencensus.io/stats/view"
|
||||
"go.opencensus.io/tag"
|
||||
|
||||
"github.com/filecoin-project/lotus/metrics"
|
||||
)
|
||||
|
||||
var (
|
||||
dbTag, _ = tag.NewKey("db_name")
|
||||
pre = "clusterdb/base/"
|
||||
waitsBuckets = []float64{0, 10, 20, 30, 50, 80, 130, 210, 340, 550, 890}
|
||||
whichHostBuckets = []float64{0, 1, 2, 3, 4, 5}
|
||||
)
|
||||
|
||||
// DBMeasures groups all db metrics.
|
||||
var DBMeasures = struct {
|
||||
Hits *stats.Int64Measure
|
||||
TotalWait *stats.Int64Measure
|
||||
Waits prometheus.Histogram
|
||||
OpenConnections *stats.Int64Measure
|
||||
Errors *stats.Int64Measure
|
||||
WhichHost prometheus.Histogram
|
||||
}{
|
||||
Hits: stats.Int64(pre+"hits", "Total number of uses.", stats.UnitDimensionless),
|
||||
TotalWait: stats.Int64(pre+"total_wait", "Total delay. A numerator over hits to get average wait.", stats.UnitMilliseconds),
|
||||
Waits: prometheus.NewHistogram(prometheus.HistogramOpts{
|
||||
Name: pre + "waits",
|
||||
Buckets: waitsBuckets,
|
||||
Help: "The histogram of waits for query completions.",
|
||||
}),
|
||||
OpenConnections: stats.Int64(pre+"open_connections", "Total connection count.", stats.UnitDimensionless),
|
||||
Errors: stats.Int64(pre+"errors", "Total error count.", stats.UnitDimensionless),
|
||||
WhichHost: prometheus.NewHistogram(prometheus.HistogramOpts{
|
||||
Name: pre + "which_host",
|
||||
Buckets: whichHostBuckets,
|
||||
Help: "The index of the hostname being used",
|
||||
}),
|
||||
}
|
||||
|
||||
// CacheViews groups all cache-related default views.
|
||||
func init() {
|
||||
metrics.RegisterViews(
|
||||
&view.View{
|
||||
Measure: DBMeasures.Hits,
|
||||
Aggregation: view.Sum(),
|
||||
TagKeys: []tag.Key{dbTag},
|
||||
},
|
||||
&view.View{
|
||||
Measure: DBMeasures.TotalWait,
|
||||
Aggregation: view.Sum(),
|
||||
TagKeys: []tag.Key{dbTag},
|
||||
},
|
||||
&view.View{
|
||||
Measure: DBMeasures.OpenConnections,
|
||||
Aggregation: view.LastValue(),
|
||||
TagKeys: []tag.Key{dbTag},
|
||||
},
|
||||
&view.View{
|
||||
Measure: DBMeasures.Errors,
|
||||
Aggregation: view.Sum(),
|
||||
TagKeys: []tag.Key{dbTag},
|
||||
},
|
||||
)
|
||||
prometheus.Register(DBMeasures.Waits)
|
||||
prometheus.Register(DBMeasures.WhichHost)
|
||||
}
|
||||
6
lib/sturdy/clusterdb/sql/20230706.sql
Normal file
6
lib/sturdy/clusterdb/sql/20230706.sql
Normal file
@ -0,0 +1,6 @@
|
||||
CREATE TABLE itest_scratch (
|
||||
id SERIAL PRIMARY KEY,
|
||||
content TEXT,
|
||||
some_int INTEGER,
|
||||
update_time TIMESTAMP DEFAULT current_timestamp
|
||||
)
|
||||
149
lib/sturdy/clusterdb/userfuncs.go
Normal file
149
lib/sturdy/clusterdb/userfuncs.go
Normal file
@ -0,0 +1,149 @@
|
||||
package clusterdb
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/georgysavva/scany/v2/pgxscan"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
type Intf interface {
|
||||
ITestDeleteAll()
|
||||
Exec(ctx context.Context, sql rawStringOnly, arguments ...any) error
|
||||
Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error)
|
||||
QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row
|
||||
Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error
|
||||
BeginTransaction(ctx context.Context, f func(t *Transaction) (commit bool)) (retErr error)
|
||||
}
|
||||
|
||||
// rawStringOnly is _intentionally_private_ to force only basic strings in SQL queries.
|
||||
// In any package, raw strings will satisfy compilation. Ex:
|
||||
//
|
||||
// sturdydb.Exec("INSERT INTO version (number) VALUES (1)")
|
||||
//
|
||||
// This prevents SQL injection attacks where the input contains query fragments.
|
||||
type rawStringOnly string
|
||||
|
||||
// Exec executes changes (INSERT, DELETE, or UPDATE).
|
||||
// Note, for CREATE & DROP please keep these permanent and express
|
||||
// them in the ./sql/ files (next number).
|
||||
func (db *DB) Exec(ctx context.Context, sql rawStringOnly, arguments ...any) error {
|
||||
_, err := db.pgx.Exec(ctx, string(sql), arguments...)
|
||||
return err
|
||||
}
|
||||
|
||||
type Qry interface {
|
||||
Next() bool
|
||||
Err() error
|
||||
Close()
|
||||
Scan(...any) error
|
||||
Values() ([]any, error)
|
||||
}
|
||||
|
||||
// Query offers Next/Err/Close/Scan/Values/StructScan
|
||||
type Query struct {
|
||||
Qry
|
||||
}
|
||||
|
||||
// Query allows iterating returned values to save memory consumption
|
||||
// with the downside of needing to `defer q.Close()`. For a simpler interface,
|
||||
// try Select()
|
||||
// Next() must be called to advance the row cursor, including the first time:
|
||||
// Ex:
|
||||
// q, err := db.Query(ctx, "SELECT id, name FROM users")
|
||||
// handleError(err)
|
||||
// defer q.Close()
|
||||
//
|
||||
// for q.Next() {
|
||||
// var id int
|
||||
// var name string
|
||||
// handleError(q.Scan(&id, &name))
|
||||
// fmt.Println(id, name)
|
||||
// }
|
||||
func (db *DB) Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error) {
|
||||
q, err := db.pgx.Query(ctx, string(sql), arguments...)
|
||||
return &Query{q}, err
|
||||
}
|
||||
func (q *Query) StructScan(s any) error {
|
||||
return pgxscan.ScanRow(s, q.Qry.(pgx.Rows))
|
||||
}
|
||||
|
||||
type Row interface {
|
||||
Scan(...any) error
|
||||
}
|
||||
|
||||
// QueryRow gets 1 row using column order matching.
|
||||
// This is a timesaver for the special case of wanting the first row returned only.
|
||||
// EX:
|
||||
//
|
||||
// var name, pet string
|
||||
// var ID = 123
|
||||
// err := db.QueryRow(ctx, "SELECT name, pet FROM users WHERE ID=?", ID).Scan(&name, &pet)
|
||||
func (db *DB) QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row {
|
||||
return db.pgx.QueryRow(ctx, string(sql), arguments...)
|
||||
}
|
||||
|
||||
/*
|
||||
Select multiple rows into a slice using name matching
|
||||
Ex:
|
||||
|
||||
type user struct {
|
||||
Name string
|
||||
ID int
|
||||
Number string `db:"tel_no"`
|
||||
}
|
||||
|
||||
var users []user
|
||||
pet := "cat"
|
||||
err := db.Select(ctx, &users, "SELECT name, id, tel_no FROM customers WHERE pet=?", pet)
|
||||
*/
|
||||
func (db *DB) Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error {
|
||||
return pgxscan.Select(ctx, db.pgx, sliceOfStructPtr, string(sql), arguments...)
|
||||
}
|
||||
|
||||
type Transaction struct {
|
||||
pgx.Tx
|
||||
}
|
||||
|
||||
// BeginTransaction is how you can access transactions using this library.
|
||||
// The entire transaction happens in the function passed in.
|
||||
// The return must be true or a rollback will occur.
|
||||
func (db *DB) BeginTransaction(ctx context.Context, f func(t *Transaction) (commit bool)) (retErr error) {
|
||||
tx, err := db.pgx.BeginTx(ctx, pgx.TxOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var commit bool
|
||||
defer func() { // Panic clean-up.
|
||||
if !commit {
|
||||
retErr = tx.Rollback(ctx)
|
||||
}
|
||||
}()
|
||||
commit = f(&Transaction{tx})
|
||||
if commit {
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Exec in a transaction.
|
||||
func (t *Transaction) Exec(ctx context.Context, sql rawStringOnly, arguments ...any) error {
|
||||
_, err := t.Tx.Exec(ctx, string(sql), arguments...)
|
||||
return err
|
||||
}
|
||||
|
||||
// Query in a transaction.
|
||||
func (t *Transaction) Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error) {
|
||||
q, err := t.Tx.Query(ctx, string(sql), arguments...)
|
||||
return &Query{q}, err
|
||||
}
|
||||
|
||||
// QueryRow in a transaction.
|
||||
func (t *Transaction) QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row {
|
||||
return t.Tx.QueryRow(ctx, string(sql), arguments...)
|
||||
}
|
||||
|
||||
// Select in a transaction.
|
||||
func (t *Transaction) Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error {
|
||||
return pgxscan.Select(ctx, t.Tx, sliceOfStructPtr, string(sql), arguments...)
|
||||
}
|
||||
@ -695,46 +695,55 @@ var (
|
||||
}
|
||||
)
|
||||
|
||||
var views = []*view.View{
|
||||
InfoView,
|
||||
PeerCountView,
|
||||
APIRequestDurationView,
|
||||
|
||||
GraphsyncReceivingPeersCountView,
|
||||
GraphsyncReceivingActiveCountView,
|
||||
GraphsyncReceivingCountCountView,
|
||||
GraphsyncReceivingTotalMemoryAllocatedView,
|
||||
GraphsyncReceivingTotalPendingAllocationsView,
|
||||
GraphsyncReceivingPeersPendingView,
|
||||
GraphsyncSendingPeersCountView,
|
||||
GraphsyncSendingActiveCountView,
|
||||
GraphsyncSendingCountCountView,
|
||||
GraphsyncSendingTotalMemoryAllocatedView,
|
||||
GraphsyncSendingTotalPendingAllocationsView,
|
||||
GraphsyncSendingPeersPendingView,
|
||||
|
||||
RcmgrAllowConnView,
|
||||
RcmgrBlockConnView,
|
||||
RcmgrAllowStreamView,
|
||||
RcmgrBlockStreamView,
|
||||
RcmgrAllowPeerView,
|
||||
RcmgrBlockPeerView,
|
||||
RcmgrAllowProtoView,
|
||||
RcmgrBlockProtoView,
|
||||
RcmgrBlockProtoPeerView,
|
||||
RcmgrAllowSvcView,
|
||||
RcmgrBlockSvcView,
|
||||
RcmgrBlockSvcPeerView,
|
||||
RcmgrAllowMemView,
|
||||
RcmgrBlockMemView,
|
||||
}
|
||||
|
||||
// DefaultViews is an array of OpenCensus views for metric gathering purposes
|
||||
var DefaultViews = func() []*view.View {
|
||||
views := []*view.View{
|
||||
InfoView,
|
||||
PeerCountView,
|
||||
APIRequestDurationView,
|
||||
|
||||
GraphsyncReceivingPeersCountView,
|
||||
GraphsyncReceivingActiveCountView,
|
||||
GraphsyncReceivingCountCountView,
|
||||
GraphsyncReceivingTotalMemoryAllocatedView,
|
||||
GraphsyncReceivingTotalPendingAllocationsView,
|
||||
GraphsyncReceivingPeersPendingView,
|
||||
GraphsyncSendingPeersCountView,
|
||||
GraphsyncSendingActiveCountView,
|
||||
GraphsyncSendingCountCountView,
|
||||
GraphsyncSendingTotalMemoryAllocatedView,
|
||||
GraphsyncSendingTotalPendingAllocationsView,
|
||||
GraphsyncSendingPeersPendingView,
|
||||
|
||||
RcmgrAllowConnView,
|
||||
RcmgrBlockConnView,
|
||||
RcmgrAllowStreamView,
|
||||
RcmgrBlockStreamView,
|
||||
RcmgrAllowPeerView,
|
||||
RcmgrBlockPeerView,
|
||||
RcmgrAllowProtoView,
|
||||
RcmgrBlockProtoView,
|
||||
RcmgrBlockProtoPeerView,
|
||||
RcmgrAllowSvcView,
|
||||
RcmgrBlockSvcView,
|
||||
RcmgrBlockSvcPeerView,
|
||||
RcmgrAllowMemView,
|
||||
RcmgrBlockMemView,
|
||||
}
|
||||
views = append(views, blockstore.DefaultViews...)
|
||||
views = append(views, rpcmetrics.DefaultViews...)
|
||||
return views
|
||||
}()
|
||||
|
||||
// RegisterViews adds views to the default list without modifying this file.
|
||||
func RegisterViews(v ...*view.View) {
|
||||
views = append(views, v...)
|
||||
}
|
||||
|
||||
func init() {
|
||||
RegisterViews(blockstore.DefaultViews...)
|
||||
RegisterViews(rpcmetrics.DefaultViews...)
|
||||
}
|
||||
|
||||
var ChainNodeViews = append([]*view.View{
|
||||
ChainNodeHeightView,
|
||||
ChainNodeHeightExpectedView,
|
||||
|
||||
@ -19,6 +19,7 @@ import (
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/gen"
|
||||
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
|
||||
"github.com/filecoin-project/lotus/lib/sturdy/clusterdb"
|
||||
"github.com/filecoin-project/lotus/markets/dagstore"
|
||||
"github.com/filecoin-project/lotus/markets/dealfilter"
|
||||
"github.com/filecoin-project/lotus/markets/idxprov"
|
||||
@ -231,6 +232,7 @@ func ConfigStorageMiner(c interface{}) Option {
|
||||
Override(new(config.SealerConfig), cfg.Storage),
|
||||
Override(new(config.ProvingConfig), cfg.Proving),
|
||||
Override(new(*ctladdr.AddressSelector), modules.AddressSelector(&cfg.Addresses)),
|
||||
Override(new(*clusterdb.DB), clusterdb.NewFromConfig),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@ -269,6 +269,13 @@ func DefaultStorageMiner() *StorageMiner {
|
||||
MaxConcurrentUnseals: 5,
|
||||
GCInterval: Duration(1 * time.Minute),
|
||||
},
|
||||
ClusterDB: ClusterDB{
|
||||
Hosts: []string{"127.0.0.1"},
|
||||
Username: "yugabyte",
|
||||
Password: "yugabyte",
|
||||
Database: "yugabyte",
|
||||
Port: "5433",
|
||||
},
|
||||
}
|
||||
|
||||
cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http"
|
||||
|
||||
@ -117,6 +117,46 @@ without existing payment channels with available funds will fail instead
|
||||
of automatically performing on-chain operations.`,
|
||||
},
|
||||
},
|
||||
"ClusterDB": []DocField{
|
||||
{
|
||||
Name: "Hosts",
|
||||
Type: "[]string",
|
||||
|
||||
Comment: `HOSTS is a list of hostnames to nodes running YugabyteDB
|
||||
in a cluster. Only 1 is required`,
|
||||
},
|
||||
{
|
||||
Name: "Username",
|
||||
Type: "string",
|
||||
|
||||
Comment: `The Yugabyte server's username with full credentials to operate on Lotus' Database. Blank for default.`,
|
||||
},
|
||||
{
|
||||
Name: "Password",
|
||||
Type: "string",
|
||||
|
||||
Comment: `The password for the related username. Blank for default.`,
|
||||
},
|
||||
{
|
||||
Name: "Database",
|
||||
Type: "string",
|
||||
|
||||
Comment: `The database (logical partition) within Yugabyte. Blank for default.`,
|
||||
},
|
||||
{
|
||||
Name: "Port",
|
||||
Type: "string",
|
||||
|
||||
Comment: `The port to find Yugabyte. Blank for default.`,
|
||||
},
|
||||
{
|
||||
Name: "ITest",
|
||||
Type: "string",
|
||||
|
||||
Comment: `ITest is for optimized integration testing and not
|
||||
for production. Blank for default production configuration.`,
|
||||
},
|
||||
},
|
||||
"Common": []DocField{
|
||||
{
|
||||
Name: "API",
|
||||
@ -1386,6 +1426,12 @@ HotstoreMaxSpaceTarget - HotstoreMaxSpaceSafetyBuffer`,
|
||||
Name: "DAGStore",
|
||||
Type: "DAGStoreConfig",
|
||||
|
||||
Comment: ``,
|
||||
},
|
||||
{
|
||||
Name: "ClusterDB",
|
||||
Type: "ClusterDB",
|
||||
|
||||
Comment: ``,
|
||||
},
|
||||
},
|
||||
|
||||
@ -61,6 +61,8 @@ type StorageMiner struct {
|
||||
Fees MinerFeeConfig
|
||||
Addresses MinerAddressConfig
|
||||
DAGStore DAGStoreConfig
|
||||
|
||||
ClusterDB ClusterDB
|
||||
}
|
||||
|
||||
type DAGStoreConfig struct {
|
||||
@ -732,3 +734,25 @@ type IndexConfig struct {
|
||||
// EnableMsgIndex enables indexing of messages on chain.
|
||||
EnableMsgIndex bool
|
||||
}
|
||||
|
||||
type ClusterDB struct {
|
||||
// HOSTS is a list of hostnames to nodes running YugabyteDB
|
||||
// in a cluster. Only 1 is required
|
||||
Hosts []string
|
||||
|
||||
// The Yugabyte server's username with full credentials to operate on Lotus' Database. Blank for default.
|
||||
Username string
|
||||
|
||||
// The password for the related username. Blank for default.
|
||||
Password string
|
||||
|
||||
// The database (logical partition) within Yugabyte. Blank for default.
|
||||
Database string
|
||||
|
||||
// The port to find Yugabyte. Blank for default.
|
||||
Port string
|
||||
|
||||
// ITest is for optimized integration testing and not
|
||||
// for production. Blank for default production configuration.
|
||||
ITest string
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user