ensure only single origin Get request is in flight per hash. protects against thundering herd. #45
14 changed files with 209 additions and 44 deletions
|
|
@ -94,13 +94,17 @@ func reflectorCmd(cmd *cobra.Command, args []string) {
|
|||
|
||||
blobStore = store.NewDBBackedStore(blobStore, db)
|
||||
|
||||
reflectorServer = reflector.NewServer(blobStore)
|
||||
reflectorServer.Timeout = 3 * time.Minute
|
||||
reflectorServer.EnableBlocklist = !disableBlocklist
|
||||
//this shouldn't go here but the blocklist logic requires the db backed store to be the outer-most store for it to work....
|
||||
//having this here prevents uploaded blobs from being stored in the disk cache
|
||||
if !disableUploads {
|
||||
reflectorServer = reflector.NewServer(blobStore)
|
||||
reflectorServer.Timeout = 3 * time.Minute
|
||||
reflectorServer.EnableBlocklist = !disableBlocklist
|
||||
|
||||
err = reflectorServer.Start(":" + strconv.Itoa(receiverPort))
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
err = reflectorServer.Start(":" + strconv.Itoa(receiverPort))
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
6
go.mod
6
go.mod
|
|
@ -20,19 +20,21 @@ require (
|
|||
github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf
|
||||
github.com/johntdyer/slackrus v0.0.0-20180518184837-f7aae3243a07
|
||||
github.com/lbryio/chainquery v1.9.0
|
||||
github.com/lbryio/lbry.go v1.1.2
|
||||
github.com/lbryio/lbry.go v1.1.2 // indirect
|
||||
github.com/lbryio/lbry.go/v2 v2.6.1-0.20200901175808-73382bb02128
|
||||
github.com/lbryio/types v0.0.0-20191228214437-05a22073b4ec
|
||||
github.com/lucas-clemente/quic-go v0.18.0
|
||||
github.com/lucas-clemente/quic-go v0.18.1
|
||||
github.com/phayes/freeport v0.0.0-20171002185219-e27662a4a9d6
|
||||
github.com/prometheus/client_golang v0.9.2
|
||||
github.com/sirupsen/logrus v1.4.2
|
||||
github.com/spf13/cast v1.3.0
|
||||
github.com/spf13/cobra v0.0.3
|
||||
github.com/spf13/pflag v1.0.3 // indirect
|
||||
github.com/volatiletech/null v8.0.0+incompatible
|
||||
go.uber.org/atomic v1.5.1
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
|
||||
golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f // indirect
|
||||
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e
|
||||
golang.org/x/tools v0.0.0-20191227053925-7b8e75db28f4 // indirect
|
||||
google.golang.org/appengine v1.6.2 // indirect
|
||||
)
|
||||
|
|
|
|||
32
go.sum
32
go.sum
|
|
@ -26,10 +26,6 @@ github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLM
|
|||
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
|
||||
github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
|
||||
github.com/bradfitz/go-smtpd v0.0.0-20170404230938-deb6d6237625/go.mod h1:HYsPBTaaSFSlLx/70C2HPIMNZpVV8+vt/A+FMnYP11g=
|
||||
github.com/btcsuite/btcd v0.0.0-20190213025234-306aecffea32 h1:qkOC5Gd33k54tobS36cXdAzJbeHaduLtnLQQwNoIi78=
|
||||
github.com/btcsuite/btcd v0.0.0-20190213025234-306aecffea32/go.mod h1:DrZx5ec/dmnfpw9KyYoQyYo7d0KEvTkk/5M/vbZjAr8=
|
||||
github.com/btcsuite/btcd v0.0.0-20190824003749-130ea5bddde3 h1:A/EVblehb75cUgXA5njHPn0kLAsykn6mJGz7rnmW5W0=
|
||||
github.com/btcsuite/btcd v0.0.0-20190824003749-130ea5bddde3/go.mod h1:3J08xEfcugPacsc34/LKRU2yO7YmuT8yt28J8k2+rrI=
|
||||
github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f h1:bAs4lUbRJpnnkd9VhRV3jjAVU7DJVjMaK+IsvSeZvFo=
|
||||
github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f/go.mod h1:TdznJufoqS23FtqVCzL0ZqgP5MqXbb4fg/WgDys70nA=
|
||||
github.com/btcsuite/btcutil v0.0.0-20190207003914-4c204d697803/go.mod h1:+5NJ2+qvTyV9exUAL/rxXi3DcLg2Ts+ymUAY5y4NvMg=
|
||||
|
|
@ -76,6 +72,7 @@ github.com/go-ozzo/ozzo-validation v3.5.0+incompatible/go.mod h1:gsEKFIVnabGBt6m
|
|||
github.com/go-ozzo/ozzo-validation v3.6.0+incompatible/go.mod h1:gsEKFIVnabGBt6mXmxK0MoFy+cZoTJY6mu5Ll3LVLBU=
|
||||
github.com/go-sql-driver/mysql v1.4.1 h1:g24URVg0OFbNUTx9qqY1IRZ9D9z3iPyi5zKhQZpNwpA=
|
||||
github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
|
||||
github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4+jfEaewE=
|
||||
github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
|
||||
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
|
||||
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
|
||||
|
|
@ -86,6 +83,7 @@ github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfb
|
|||
github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
|
||||
github.com/golang/mock v1.4.0 h1:Rd1kQnQu0Hq3qvJppYSG0HtP+f5LPPUiDswTLiEegLg=
|
||||
github.com/golang/mock v1.4.0/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
|
||||
github.com/golang/mock v1.4.4 h1:l75CXGRSwbaYNpl/Z2X1XIIAMSCquvXgpVZDhwEIJsc=
|
||||
github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
|
||||
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
|
||||
github.com/golang/protobuf v1.3.0/go.mod h1:Qd/q+1AKNOZr9uGQzbzCmRO6sUih6GTPZv6a1/R87v0=
|
||||
|
|
@ -203,8 +201,6 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
|||
github.com/lbryio/chainquery v1.9.0 h1:NfBZ3eKYwD3PqXU/vt+2tF3ox3WUWoW4J5YdEQ0rxw0=
|
||||
github.com/lbryio/chainquery v1.9.0/go.mod h1:7G8l7jNtANS1I7fQOvtzbiHsv6qKVmN4codXHc3C4kk=
|
||||
github.com/lbryio/errors.go v0.0.0-20180223142025-ad03d3cc6a5c/go.mod h1:muH7wpUqE8hRA3OrYYosw9+Sl681BF9cwcjzE+OCNK8=
|
||||
github.com/lbryio/go_mediainfo v0.0.0-20200109212001-4c7318fd92ad h1:sfsYsbQXjA1uE51yFNtv4k79yMBzeGgAusK6KrLrcYs=
|
||||
github.com/lbryio/go_mediainfo v0.0.0-20200109212001-4c7318fd92ad/go.mod h1:xJtOuuGAo32jLOmrdBWRCURfahqz6OvK/hdSLhmYA38=
|
||||
github.com/lbryio/lbry.go v1.1.1-0.20190825202001-8fa28d3d656f h1:ovd2wPXzkT80vdP/FX5xcQeXu0i9RAo80SQ6qIsrAjM=
|
||||
github.com/lbryio/lbry.go v1.1.1-0.20190825202001-8fa28d3d656f/go.mod h1:JtyI30bU51rm0LZ/po3mQuzf++14OWb6kR/6mMRAmKU=
|
||||
github.com/lbryio/lbry.go v1.1.2 h1:Dyxc+glT/rVWJwHfIf7vjlPYYbjzrQz5ARmJd5Hp69c=
|
||||
|
|
@ -221,10 +217,8 @@ github.com/lbryio/types v0.0.0-20191009145016-1bb8107e04f8/go.mod h1:CG3wsDv5BiV
|
|||
github.com/lbryio/types v0.0.0-20191228214437-05a22073b4ec h1:2xk/qg4VTOCJ8RzV/ED5AKqDcJ00zVb08ltf9V+sr3c=
|
||||
github.com/lbryio/types v0.0.0-20191228214437-05a22073b4ec/go.mod h1:CG3wsDv5BiVYQd5i1Jp7wGsaVyjZTJshqXeWMVKsISE=
|
||||
github.com/lib/pq v1.1.1/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
|
||||
github.com/lucas-clemente/quic-go v0.17.2 h1:4iQInIuNQkPNZmsy9rCnwuOzpH0qGnDo4jn0QfI/qE4=
|
||||
github.com/lucas-clemente/quic-go v0.17.2/go.mod h1:I0+fcNTdb9eS1ZcjQZbDVPGchJ86chcIxPALn9lEJqE=
|
||||
github.com/lucas-clemente/quic-go v0.18.0 h1:JhQDdqxdwdmGdKsKgXi1+coHRoGhvU6z0rNzOJqZ/4o=
|
||||
github.com/lucas-clemente/quic-go v0.18.0/go.mod h1:yXttHsSNxQi8AWijC/vLP+OJczXqzHSOcJrM5ITUlCg=
|
||||
github.com/lucas-clemente/quic-go v0.18.1 h1:DMR7guC0NtVS8zNZR3IO7NARZvZygkSC56GGtC6cyys=
|
||||
github.com/lucas-clemente/quic-go v0.18.1/go.mod h1:yXttHsSNxQi8AWijC/vLP+OJczXqzHSOcJrM5ITUlCg=
|
||||
github.com/lunixbochs/vtclean v1.0.0/go.mod h1:pHhQNgMf3btfWnGBVipUOjRYhoOsdGqdm/+2c2E2WMI=
|
||||
github.com/lusis/go-slackbot v0.0.0-20180109053408-401027ccfef5/go.mod h1:c2mYKRyMb1BPkO5St0c/ps62L4S0W2NAkaTXj9qEI+0=
|
||||
github.com/lusis/slack-test v0.0.0-20180109053238-3c758769bfa6/go.mod h1:sFlOUpQL1YcjhFVXhg1CG8ZASEs/Mf1oVb6H75JL/zg=
|
||||
|
|
@ -233,12 +227,8 @@ github.com/lyoshenka/bencode v0.0.0-20180323155644-b7abd7672df5/go.mod h1:H0aPCW
|
|||
github.com/magiconair/properties v1.8.0 h1:LLgXmsheXeRoUOBOjtwPQCWIYqM/LU1ayDtDePerRcY=
|
||||
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
|
||||
github.com/mailru/easyjson v0.0.0-20190312143242-1de009706dbe/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
|
||||
github.com/marten-seemann/qpack v0.1.0 h1:/0M7lkda/6mus9B8u34Asqm8ZhHAAt9Ho0vniNuVSVg=
|
||||
github.com/marten-seemann/qpack v0.1.0/go.mod h1:LFt1NU/Ptjip0C2CPkhimBz5CGE3WGDAUWqna+CNTrI=
|
||||
github.com/marten-seemann/qpack v0.2.0 h1:/r1rhZoOmgxVKBqPNnYilZBDEyw+6OUHCbBzA5jc2y0=
|
||||
github.com/marten-seemann/qpack v0.2.0/go.mod h1:F7Gl5L1jIgN1D11ucXefiuJS9UMVP2opoCp2jDKb7wc=
|
||||
github.com/marten-seemann/qtls v0.9.1 h1:O0YKQxNVPaiFgMng0suWEOY2Sb4LT2sRn9Qimq3Z1IQ=
|
||||
github.com/marten-seemann/qtls v0.9.1/go.mod h1:T1MmAdDPyISzxlK6kjRr0pcZFBVd1OZbBb/j3cvzHhk=
|
||||
github.com/marten-seemann/qtls v0.10.0 h1:ECsuYUKalRL240rRD4Ri33ISb7kAQ3qGDlrrl55b2pc=
|
||||
github.com/marten-seemann/qtls v0.10.0/go.mod h1:UvMd1oaYDACI99/oZUYLzMCkBXQVT0aGm99sJhbT8hs=
|
||||
github.com/marten-seemann/qtls-go1-15 v0.1.0 h1:i/YPXVxz8q9umso/5y474CNcHmTpA+5DH+mFPjx6PZg=
|
||||
|
|
@ -265,19 +255,18 @@ github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a
|
|||
github.com/nlopes/slack v0.5.0/go.mod h1:jVI4BBK3lSktibKahxBF74txcK2vyvkza1z/+rRnVAM=
|
||||
github.com/nlopes/slack v0.6.0 h1:jt0jxVQGhssx1Ib7naAOZEZcGdtIhTzkP0nopK0AsRA=
|
||||
github.com/nlopes/slack v0.6.0/go.mod h1:JzQ9m3PMAqcpeCam7UaHSuBuupz7CmpjehYMayT6YOk=
|
||||
github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
|
||||
github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
|
||||
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
|
||||
github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
|
||||
github.com/onsi/ginkgo v1.10.2/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
|
||||
github.com/onsi/ginkgo v1.11.0 h1:JAKSXpt1YjtLA7YpPiqO9ss6sNXEsPfSGdwN0UHqzrw=
|
||||
github.com/onsi/ginkgo v1.11.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
|
||||
github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
|
||||
github.com/onsi/ginkgo v1.14.0 h1:2mOpI4JVVPBN+WQRa0WKH2eXR+Ey+uK4n7Zj0aYpIQA=
|
||||
github.com/onsi/ginkgo v1.14.0/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
|
||||
github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
|
||||
github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
|
||||
github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
|
||||
github.com/onsi/gomega v1.8.1 h1:C5Dqfs/LeauYDX0jJXIe2SWmwCbGzx9yF8C8xy3Lh34=
|
||||
github.com/onsi/gomega v1.8.1/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA=
|
||||
github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
|
||||
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
|
||||
github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
|
||||
|
|
@ -380,8 +369,11 @@ github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqri
|
|||
github.com/uber-go/atomic v1.3.2/go.mod h1:/Ct5t2lcmbJ4OSe/waGBoaVvVqtO0bmtfVNex1PFV8g=
|
||||
github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49uaYMPRU=
|
||||
github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMILuUhzM=
|
||||
github.com/volatiletech/inflect v0.0.0-20170731032912-e7201282ae8d h1:gI4/tqP6lCY5k6Sg+4k9qSoBXmPwG+xXgMpK7jivD4M=
|
||||
github.com/volatiletech/inflect v0.0.0-20170731032912-e7201282ae8d/go.mod h1:jspfvgf53t5NLUT4o9L1IX0kIBNKamGq1tWc/MgWK9Q=
|
||||
github.com/volatiletech/null v8.0.0+incompatible h1:7wP8m5d/gZ6kW/9GnrLtMCRre2dlEnaQ9Km5OXlK4zg=
|
||||
github.com/volatiletech/null v8.0.0+incompatible/go.mod h1:0wD98JzdqB+rLyZ70fN05VDbXbafIb0KU0MdVhCzmOQ=
|
||||
github.com/volatiletech/sqlboiler v3.4.0+incompatible h1:saQ6WxZ9wEJp33q3w/DHs7an7SYi1H7Yzf4/moxCbJU=
|
||||
github.com/volatiletech/sqlboiler v3.4.0+incompatible/go.mod h1:jLfDkkHWPbS2cWRLkyC20vQWaIQsASEY7gM7zSo11Yw=
|
||||
github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg=
|
||||
github.com/ybbus/jsonrpc v0.0.0-20180411222309-2a548b7d822d/go.mod h1:XJrh1eMSzdIYFbM08flv0wp5G35eRniyeGut1z+LSiE=
|
||||
|
|
@ -405,8 +397,6 @@ golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8U
|
|||
golang.org/x/crypto v0.0.0-20191002192127-34f69633bfdc/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200221231518-2aa609cf4a9d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.0.0-20200423211502-4bdfaf469ed5 h1:Q7tZBpemrlsc2I7IyODzhtallWRSm4Q0d09pL6XbQtU=
|
||||
golang.org/x/crypto v0.0.0-20200423211502-4bdfaf469ed5/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
|
||||
|
|
@ -428,7 +418,6 @@ golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73r
|
|||
golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20190228165749-92fc7df08ae7/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190313220215-9f648a60d977/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
|
|
@ -451,6 +440,7 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
|
|||
golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
|
||||
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sys v0.0.0-20171017063910-8dbc5d05d6ed/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
|
|
|
|||
|
|
@ -117,6 +117,17 @@ var (
|
|||
Name: "cache_miss_total",
|
||||
Help: "Total number of blobs retrieved from origin rather than cache storage",
|
||||
})
|
||||
CacheOriginRequestsCount = promauto.NewGauge(prometheus.GaugeOpts{
|
||||
Namespace: ns,
|
||||
Name: "cache_origin_requests_total",
|
||||
Help: "How many Get requests are in flight from the cache to the origin",
|
||||
})
|
||||
// during thundering-herd situations, the metric below should be a lot smaller than the metric above
|
||||
CacheWaitingRequestsCount = promauto.NewGauge(prometheus.GaugeOpts{
|
||||
Namespace: ns,
|
||||
Name: "cache_waiting_requests_total",
|
||||
Help: "How many cache requests are waiting for an in-flight origin request",
|
||||
})
|
||||
BlobUploadCount = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "blob_upload_total",
|
||||
|
|
@ -137,6 +148,41 @@ var (
|
|||
Name: "error_total",
|
||||
Help: "Total number of errors",
|
||||
}, []string{labelDirection, labelErrorType})
|
||||
MtrInBytesTcp = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "tcp_in_bytes",
|
||||
Help: "Total number of bytes downloaded through TCP",
|
||||
})
|
||||
MtrOutBytesTcp = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "tcp_out_bytes",
|
||||
Help: "Total number of bytes streamed out through TCP",
|
||||
})
|
||||
MtrInBytesUdp = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "udp_in_bytes",
|
||||
Help: "Total number of bytes downloaded through UDP",
|
||||
})
|
||||
MtrOutBytesUdp = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "udp_out_bytes",
|
||||
Help: "Total number of bytes streamed out through UDP",
|
||||
})
|
||||
MtrInBytesReflector = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "reflector_in_bytes",
|
||||
Help: "Total number of incoming bytes (from users)",
|
||||
})
|
||||
MtrOutBytesReflector = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "s3_out_bytes",
|
||||
Help: "Total number of outgoing bytes (to S3)",
|
||||
})
|
||||
MtrInBytesS3 = promauto.NewCounter(prometheus.CounterOpts{
|
||||
Namespace: ns,
|
||||
Name: "s3_in_bytes",
|
||||
Help: "Total number of incoming bytes (from S3-CF)",
|
||||
})
|
||||
)
|
||||
|
||||
func TrackError(direction string, e error) (shouldLog bool) { // shouldLog is a hack, but whatever
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import (
|
|||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/lbryio/reflector.go/internal/metrics"
|
||||
"github.com/lbryio/reflector.go/store"
|
||||
|
||||
"github.com/lbryio/lbry.go/v2/extras/errors"
|
||||
|
|
@ -152,7 +153,7 @@ func (c *Client) GetBlob(hash string) (stream.Blob, error) {
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
metrics.MtrInBytesTcp.Add(float64(len(blob)))
|
||||
return blob, nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import (
|
|||
|
||||
"github.com/lbryio/lbry.go/v2/extras/errors"
|
||||
"github.com/lbryio/lbry.go/v2/stream"
|
||||
"github.com/lbryio/reflector.go/internal/metrics"
|
||||
"github.com/lbryio/reflector.go/store"
|
||||
|
||||
"github.com/lucas-clemente/quic-go/http3"
|
||||
|
|
@ -91,5 +92,6 @@ func (c *Client) GetBlob(hash string) (stream.Blob, error) {
|
|||
if err != nil {
|
||||
return nil, errors.Err(err)
|
||||
}
|
||||
metrics.MtrInBytesUdp.Add(float64(len(body.Bytes())))
|
||||
return body.Bytes(), nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ type availabilityResponse struct {
|
|||
func (s *Server) Start(address string) error {
|
||||
log.Println("HTTP3 peer listening on " + address)
|
||||
quicConf := &quic.Config{
|
||||
HandshakeTimeout: 3 * time.Second,
|
||||
HandshakeTimeout: 4 * time.Second,
|
||||
MaxIdleTimeout: 5 * time.Second,
|
||||
}
|
||||
r := mux.NewRouter()
|
||||
|
|
@ -87,6 +87,7 @@ func (s *Server) Start(address string) error {
|
|||
if err != nil {
|
||||
s.logError(err)
|
||||
}
|
||||
metrics.MtrOutBytesUdp.Add(float64(len(blob)))
|
||||
metrics.BlobDownloadCount.Inc()
|
||||
metrics.Http3DownloadCount.Inc()
|
||||
})
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ func NewStore(opts StoreOpts) *Store {
|
|||
|
||||
func (p *Store) getClient() (*Client, error) {
|
||||
var qconf quic.Config
|
||||
qconf.HandshakeTimeout = 3 * time.Second
|
||||
qconf.HandshakeTimeout = 4 * time.Second
|
||||
qconf.MaxIdleTimeout = 5 * time.Second
|
||||
pool, err := x509.SystemCertPool()
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -272,6 +272,7 @@ func (s *Server) handleCompositeRequest(data []byte) ([]byte, error) {
|
|||
BlobHash: reflector.BlobHash(blob),
|
||||
Length: len(blob),
|
||||
}
|
||||
metrics.MtrOutBytesTcp.Add(float64(len(blob)))
|
||||
metrics.BlobDownloadCount.Inc()
|
||||
metrics.PeerDownloadCount.Inc()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -256,7 +256,7 @@ func (s *Server) receiveBlob(conn net.Conn) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
metrics.MtrInBytesReflector.Add(float64(len(blob)))
|
||||
metrics.BlobUploadCount.Inc()
|
||||
if isSdBlob {
|
||||
metrics.SDBlobUploadCount.Inc()
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ import (
|
|||
"github.com/lbryio/lbry.go/v2/stream"
|
||||
|
||||
"github.com/lbryio/reflector.go/internal/metrics"
|
||||
|
||||
"golang.org/x/sync/singleflight"
|
||||
)
|
||||
|
||||
// CachingBlobStore combines two stores, typically a local and a remote store, to improve performance.
|
||||
|
|
@ -14,11 +16,13 @@ import (
|
|||
// are retrieved from the origin and cached. Puts are cached and also forwarded to the origin.
|
||||
type CachingBlobStore struct {
|
||||
origin, cache BlobStore
|
||||
|
||||
sf *singleflight.Group
|
||||
}
|
||||
|
||||
// NewCachingBlobStore makes a new caching disk store and returns a pointer to it.
|
||||
func NewCachingBlobStore(origin, cache BlobStore) *CachingBlobStore {
|
||||
return &CachingBlobStore{origin: origin, cache: cache}
|
||||
return &CachingBlobStore{origin: origin, cache: cache, sf: new(singleflight.Group)}
|
||||
}
|
||||
|
||||
// Has checks the cache and then the origin for a hash. It returns true if either store has it.
|
||||
|
|
@ -35,25 +39,42 @@ func (c *CachingBlobStore) Has(hash string) (bool, error) {
|
|||
func (c *CachingBlobStore) Get(hash string) (stream.Blob, error) {
|
||||
start := time.Now()
|
||||
blob, err := c.cache.Get(hash)
|
||||
retrievalTime := time.Since(start)
|
||||
if err == nil || !errors.Is(err, ErrBlobNotFound) {
|
||||
metrics.CacheHitCount.Inc()
|
||||
rate := float64(len(blob)) / 1024 / 1024 / retrievalTime.Seconds()
|
||||
rate := float64(len(blob)) / 1024 / 1024 / time.Since(start).Seconds()
|
||||
metrics.RetrieverSpeed.With(map[string]string{metrics.MtrLabelSource: "cache"}).Set(rate)
|
||||
return blob, err
|
||||
}
|
||||
|
||||
start = time.Now()
|
||||
blob, err = c.origin.Get(hash)
|
||||
metrics.CacheMissCount.Inc()
|
||||
return c.getFromOrigin(hash)
|
||||
}
|
||||
|
||||
// getFromOrigin ensures that only one Get per hash is sent to the origin at a time,
|
||||
// thereby protecting against https://en.wikipedia.org/wiki/Thundering_herd_problem
|
||||
func (c *CachingBlobStore) getFromOrigin(hash string) (stream.Blob, error) {
|
||||
metrics.CacheWaitingRequestsCount.Inc()
|
||||
defer metrics.CacheWaitingRequestsCount.Dec()
|
||||
originBlob, err, _ := c.sf.Do(hash, func() (interface{}, error) {
|
||||
metrics.CacheOriginRequestsCount.Inc()
|
||||
defer metrics.CacheOriginRequestsCount.Dec()
|
||||
|
||||
start := time.Now()
|
||||
blob, err := c.origin.Get(hash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rate := float64(len(blob)) / 1024 / 1024 / time.Since(start).Seconds()
|
||||
metrics.RetrieverSpeed.With(map[string]string{metrics.MtrLabelSource: "origin"}).Set(rate)
|
||||
|
||||
err = c.cache.Put(hash, blob)
|
||||
return blob, err
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
retrievalTime = time.Since(start)
|
||||
err = c.cache.Put(hash, blob)
|
||||
rate := float64(len(blob)) / 1024 / 1024 / retrievalTime.Seconds()
|
||||
metrics.RetrieverSpeed.With(map[string]string{metrics.MtrLabelSource: "origin"}).Set(rate)
|
||||
metrics.CacheMissCount.Inc()
|
||||
return blob, err
|
||||
return originBlob.(stream.Blob), nil
|
||||
}
|
||||
|
||||
// Put stores the blob in the origin and the cache
|
||||
|
|
|
|||
|
|
@ -2,7 +2,11 @@ package store
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/lbryio/lbry.go/v2/stream"
|
||||
)
|
||||
|
||||
func TestCachingBlobStore_Put(t *testing.T) {
|
||||
|
|
@ -71,3 +75,92 @@ func TestCachingBlobStore_CacheMiss(t *testing.T) {
|
|||
t.Errorf("expected cached Get() to return %s, got %s", string(b), string(res))
|
||||
}
|
||||
}
|
||||
|
||||
func TestCachingBlobStore_ThunderingHerd(t *testing.T) {
|
||||
storeDelay := 100 * time.Millisecond
|
||||
origin := NewSlowBlobStore(storeDelay)
|
||||
cache := NewMemoryBlobStore()
|
||||
s := NewCachingBlobStore(origin, cache)
|
||||
|
||||
b := []byte("this is a blob of stuff")
|
||||
hash := "hash"
|
||||
err := origin.Put(hash, b)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
wg := &sync.WaitGroup{}
|
||||
|
||||
getNoErr := func() {
|
||||
res, err := s.Get(hash)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(b, res) {
|
||||
t.Errorf("expected Get() to return %s, got %s", string(b), string(res))
|
||||
}
|
||||
wg.Done()
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
|
||||
wg.Add(4)
|
||||
go func() {
|
||||
go getNoErr()
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
go getNoErr()
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
go getNoErr()
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
go getNoErr()
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
duration := time.Since(start)
|
||||
|
||||
// only the first getNoErr() should hit the origin. the rest should wait for the first request to return
|
||||
// once the first returns, the others should return immediately
|
||||
// therefore, if the delay much longer than 100ms, it means subsequent requests also went to the origin
|
||||
expectedMaxDelay := storeDelay + 5*time.Millisecond // a bit of extra time to let requests finish
|
||||
if duration > expectedMaxDelay {
|
||||
t.Errorf("Expected delay of at most %s, got %s", expectedMaxDelay, duration)
|
||||
}
|
||||
}
|
||||
|
||||
// SlowBlobStore adds a delay to each request
|
||||
type SlowBlobStore struct {
|
||||
mem *MemoryBlobStore
|
||||
delay time.Duration
|
||||
}
|
||||
|
||||
func NewSlowBlobStore(delay time.Duration) *SlowBlobStore {
|
||||
return &SlowBlobStore{
|
||||
mem: NewMemoryBlobStore(),
|
||||
delay: delay,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SlowBlobStore) Has(hash string) (bool, error) {
|
||||
time.Sleep(s.delay)
|
||||
return s.mem.Has(hash)
|
||||
}
|
||||
|
||||
func (s *SlowBlobStore) Get(hash string) (stream.Blob, error) {
|
||||
time.Sleep(s.delay)
|
||||
return s.mem.Get(hash)
|
||||
}
|
||||
|
||||
func (s *SlowBlobStore) Put(hash string, blob stream.Blob) error {
|
||||
time.Sleep(s.delay)
|
||||
return s.mem.Put(hash, blob)
|
||||
}
|
||||
|
||||
func (s *SlowBlobStore) PutSD(hash string, blob stream.Blob) error {
|
||||
time.Sleep(s.delay)
|
||||
return s.mem.PutSD(hash, blob)
|
||||
}
|
||||
|
||||
func (s *SlowBlobStore) Delete(hash string) error {
|
||||
time.Sleep(s.delay)
|
||||
return s.mem.Delete(hash)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import (
|
|||
|
||||
"github.com/lbryio/lbry.go/v2/extras/errors"
|
||||
"github.com/lbryio/lbry.go/v2/stream"
|
||||
"github.com/lbryio/reflector.go/internal/metrics"
|
||||
"github.com/lbryio/reflector.go/meta"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
|
@ -77,6 +78,7 @@ func (s *CloudFrontBlobStore) Get(hash string) (stream.Blob, error) {
|
|||
if err != nil {
|
||||
return nil, errors.Err(err)
|
||||
}
|
||||
metrics.MtrInBytesS3.Add(float64(len(b)))
|
||||
return b, nil
|
||||
default:
|
||||
return nil, errors.Err(res.Status)
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import (
|
|||
|
||||
"github.com/lbryio/lbry.go/v2/extras/errors"
|
||||
"github.com/lbryio/lbry.go/v2/stream"
|
||||
"github.com/lbryio/reflector.go/internal/metrics"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/awserr"
|
||||
|
|
@ -126,6 +127,7 @@ func (s *S3BlobStore) Put(hash string, blob stream.Blob) error {
|
|||
Body: bytes.NewBuffer(blob),
|
||||
StorageClass: aws.String(s3.StorageClassIntelligentTiering),
|
||||
})
|
||||
metrics.MtrOutBytesReflector.Add(float64(blob.Size()))
|
||||
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue