diff --git a/go.mod b/go.mod index 8013e2789..cfc89c8de 100644 --- a/go.mod +++ b/go.mod @@ -17,6 +17,7 @@ require ( github.com/envoyproxy/go-control-plane v0.9.7 github.com/fsnotify/fsnotify v1.4.9 github.com/go-chi/chi v4.1.2+incompatible + github.com/go-redis/redis/v8 v8.4.0 github.com/golang/mock v1.4.4 github.com/golang/protobuf v1.4.3 github.com/gomodule/redigo v1.8.2 @@ -35,9 +36,7 @@ require ( github.com/mitchellh/hashstructure/v2 v2.0.1 github.com/natefinch/atomic v0.0.0-20200526193002-18c0533a5b09 github.com/nsf/jsondiff v0.0.0-20200515183724-f29ed568f4ce - github.com/onsi/ginkgo v1.11.0 // indirect github.com/onsi/gocleanup v0.0.0-20140331211545-c1a5478700b5 - github.com/onsi/gomega v1.8.1 // indirect github.com/open-policy-agent/opa v0.25.1 github.com/openzipkin/zipkin-go v0.2.5 github.com/ory/dockertest/v3 v3.6.2 diff --git a/go.sum b/go.sum index 63288c24b..2a1097472 100644 --- a/go.sum +++ b/go.sum @@ -148,6 +148,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no= github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ= github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec= @@ -190,6 +192,8 @@ github.com/go-kit/kit v0.10.0/go.mod h1:xUsJbQ/Fp4kEt7AFgCuvyX4a71u8h9jB8tj/ORgO github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE= github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A= +github.com/go-redis/redis/v8 v8.4.0 h1:J5NCReIgh3QgUJu398hUncxDExN4gMOHI11NVbVicGQ= +github.com/go-redis/redis/v8 v8.4.0/go.mod h1:A1tbYoHSa1fXwN+//ljcCYYJeLmVrwL9hbQN45Jdy0M= github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y= @@ -246,6 +250,7 @@ github.com/google/go-cmp v0.5.0 h1:/QaMHBdZ26BB3SSst0Iwl10Epc+xhTquomWX0oZEB6w= github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.1 h1:JFrFEBb2xKufg6XkJsJr+WbKb4FQlURi5RUcBveYu9k= github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.4 h1:L8R9j+yAqZuZjsqh/z+F1NCffTKKLShY6zXTItVIZ8M= github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-jsonnet v0.17.0 h1:/9NIEfhK1NQRKl3sP2536b2+x5HnZMdql7x3yK/l8JY= @@ -418,6 +423,8 @@ github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxzi github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/nsf/jsondiff v0.0.0-20200515183724-f29ed568f4ce h1:RPclfga2SEJmgMmz2k+Mg7cowZ8yv4Trqw9UsJby758= github.com/nsf/jsondiff v0.0.0-20200515183724-f29ed568f4ce/go.mod h1:uFMI8w+ref4v2r9jz+c9i1IfIttS/OkmLfrk1jne5hs= +github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78= +github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A= github.com/oklog/oklog v0.3.2/go.mod h1:FCV+B7mhrz4o+ueLpx+KqkyXRGMWOYEvfiXtdGtbWGs= github.com/oklog/run v1.0.0/go.mod h1:dlhp/R75TPv97u0XWUtDeV/lRKWPKSdTuV0TZvrmrQA= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= @@ -425,13 +432,16 @@ github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:v github.com/olekukonko/tablewriter v0.0.1/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo= 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.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.2 h1:8mVmC9kjFFmA8H4pKMUhcblgifdkOIXPvbhN1T36q1M= +github.com/onsi/ginkgo v1.14.2/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY= github.com/onsi/gocleanup v0.0.0-20140331211545-c1a5478700b5 h1:uuhPqmc+m7Nj7btxZEjdEUv+uFoBHNf2Tk/E7gGM+kY= github.com/onsi/gocleanup v0.0.0-20140331211545-c1a5478700b5/go.mod h1:tHaogb+iP6wJXwCqVUlmxYuJb4XDyEKxxs3E4DvMBK0= github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= -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.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY= +github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo= +github.com/onsi/gomega v1.10.3 h1:gph6h/qe9GSUw1NhH1gp+qb+h8rXD8Cy60Z32Qw3ELA= +github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc= github.com/op/go-logging v0.0.0-20160315200505-970db520ece7/go.mod h1:HzydrMdWErDVzsI23lYNej1Htcns9BCg93Dk0bBINWk= github.com/open-policy-agent/opa v0.25.1 h1:oqC/1wSrnkiddeFA7V/oKImsIeLIR7LPy/0F295AhFM= github.com/open-policy-agent/opa v0.25.1/go.mod h1:iGThTRECCfKQKICueOZkXUi0opN7BR3qiAnIrNHCmlI= @@ -615,6 +625,8 @@ go.opencensus.io v0.22.4 h1:LYy1Hy3MJdrCdMwwzxA/dRok4ejH+RwNGbuoD9fCjto= go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.opencensus.io v0.22.5 h1:dntmOdLpSpHlVqbW5Eay97DelsZHe+55D+xC6i0dDS0= go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk= +go.opentelemetry.io/otel v0.14.0 h1:YFBEfjCk9MTjaytCNSUkp9Q8lF7QJezA06T71FbQxLQ= +go.opentelemetry.io/otel v0.14.0/go.mod h1:vH5xEuwy7Rts0GNtsCW3HYQoZDY+OmBJ6t1bFGGlxgw= 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.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= @@ -709,18 +721,19 @@ golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/ golang.org/x/net v0.0.0-20200501053045-e0ff5e5a1de5/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20200506145744-7e3656a0809f/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20200513185701-a91f0712d120/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= +golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20200520182314-0ba52f642ac2/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20200927032502-5d4f70055728/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= +golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb h1:eBmm0M9fYhWpKZLjQUUKka/LtIxf46G4fxeEz5KJr9U= golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20191202225959-858c2ad4c8b6/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= -golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d h1:TzXSXBo42m9gQenoE3b9BGiEpg5IG2JkU5FkPIawgtw= golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20201207163604-931764155e3f h1:bGuVhRryQ3m1t3U3cQOa4TdSuMIXKrTrvmdJjQLbMKc= golang.org/x/oauth2 v0.0.0-20201207163604-931764155e3f/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A= @@ -731,7 +744,6 @@ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208 h1:qwRHBd0NqMbJxfbotnDhm2ByMI1Shq4Y6oRJo21SGJA= golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -754,12 +766,14 @@ golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7w 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-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/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-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191113165036-4c7a9d0fe056/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191220142924-d4481acd189f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -775,14 +789,12 @@ golang.org/x/sys v0.0.0-20200331124033-c3d80250170d/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20200501052902-10377860bb8e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200511232937-7e40ca221e25/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200515095857-1151b9dac4a9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200523222454-059865788121/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200803210538-64077c9b5642 h1:B6caxRw+hozq68X2MY7jEpZh/cr4/aHLv9xU8Kkadrw= golang.org/x/sys v0.0.0-20200803210538-64077c9b5642/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200831180312-196b9ba8737a h1:i47hUS795cOydZI4AwJQCKXOr4BvxzvikwDoDtHhP2Y= golang.org/x/sys v0.0.0-20200831180312-196b9ba8737a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201015000850-e3ed0017c211 h1:9UQO31fZ+0aKQOFldThf7BKPMJTiBfWycGh/u3UoO88= golang.org/x/sys v0.0.0-20201015000850-e3ed0017c211/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -846,15 +858,12 @@ golang.org/x/tools v0.0.0-20200512131952-2bc93b1c0c88/go.mod h1:EkVYQZoAsY45+roY golang.org/x/tools v0.0.0-20200515010526-7d3b6ebf133d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200618134242-20370b0cb4b2/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= -golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d h1:szSOL78iTCl0LF1AMjhSWJj8tIM0KixlUUnBtYXsmd8= golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= -golang.org/x/tools v0.0.0-20200825202427-b303f430e36d h1:W07d4xkoAUSNOkOzdzXCdFGxT7o2rW4q8M34tB2i//k= golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= golang.org/x/tools v0.0.0-20201009032223-96877f285f7e h1:G1acLyqfyttmexrW7XPhzsaS8m6s+P9XsW9djwh10s4= golang.org/x/tools v0.0.0-20201009032223-96877f285f7e/go.mod h1:z6u4i615ZeAfBE4XtMziQW1fSVJXACjjbWkB/mvPzlU= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -958,7 +967,6 @@ gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qS gopkg.in/cookieo9/resources-go.v2 v2.0.0-20150225115733-d27c04069d0d h1:YjTGSRV59gG1DHCq68v2B771I9dGFxvMkugf7OKglpk= gopkg.in/cookieo9/resources-go.v2 v2.0.0-20150225115733-d27c04069d0d/go.mod h1:kbUs813+JgwKQdecaTv87br/FZUaSEuPj8vbr2vq8sY= gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= -gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4= gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys= gopkg.in/gcfg.v1 v1.2.3/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o= gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= @@ -977,7 +985,6 @@ gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU= gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= diff --git a/internal/databroker/server.go b/internal/databroker/server.go index 31b02298f..a8dae60dc 100644 --- a/internal/databroker/server.go +++ b/internal/databroker/server.go @@ -193,7 +193,7 @@ func (srv *Server) GetAll(ctx context.Context, req *databroker.GetAllRequest) (* return nil, err } - all, err := db.GetAll(ctx) + all, err := db.List(ctx, "") if err != nil { return nil, err } @@ -253,7 +253,7 @@ func (srv *Server) Query(ctx context.Context, req *databroker.QueryRequest) (*da return nil, err } - all, err := db.GetAll(ctx) + all, err := db.List(ctx, "") if err != nil { return nil, err } @@ -482,7 +482,7 @@ func (srv *Server) newDB(recordType string) (db storage.Backend, err error) { case config.StorageRedisName: db, err = redis.New( srv.cfg.storageConnectionString, - recordType, + redis.WithRecordType(recordType), redis.WithTLSConfig(tlsConfig), ) if err != nil { diff --git a/internal/identity/manager/manager.go b/internal/identity/manager/manager.go index 7d5d3cb2a..f809bc5f4 100644 --- a/internal/identity/manager/manager.go +++ b/internal/identity/manager/manager.go @@ -356,7 +356,7 @@ func (mgr *Manager) mergeUsers(ctx context.Context, directoryUsers []*directory. Type: any.GetTypeUrl(), Id: id, }); err != nil { - return fmt.Errorf("failed to delete directory user: %s", id) + return fmt.Errorf("failed to delete directory user (%s): %w", id, err) } return nil }) diff --git a/pkg/storage/encrypted.go b/pkg/storage/encrypted.go index fd76af60e..54b7b5cbc 100644 --- a/pkg/storage/encrypted.go +++ b/pkg/storage/encrypted.go @@ -50,20 +50,6 @@ func (e *encryptedBackend) Get(ctx context.Context, id string) (*databroker.Reco return record, nil } -func (e *encryptedBackend) GetAll(ctx context.Context) ([]*databroker.Record, error) { - records, err := e.Backend.GetAll(ctx) - if err != nil { - return nil, err - } - for i := range records { - records[i], err = e.decryptRecord(records[i]) - if err != nil { - return nil, err - } - } - return records, nil -} - func (e *encryptedBackend) List(ctx context.Context, sinceVersion string) ([]*databroker.Record, error) { records, err := e.Backend.List(ctx, sinceVersion) if err != nil { diff --git a/pkg/storage/encrypted_test.go b/pkg/storage/encrypted_test.go index cd0feb7e8..d5d94f294 100644 --- a/pkg/storage/encrypted_test.go +++ b/pkg/storage/encrypted_test.go @@ -14,7 +14,6 @@ import ( ) func TestEncryptedBackend(t *testing.T) { - ctx := context.Background() m := map[string]*anypb.Any{} @@ -79,7 +78,7 @@ func TestEncryptedBackend(t *testing.T) { assert.Equal(t, any.Value, record.Data.Value, "value should be preserved") assert.Equal(t, any.TypeUrl, record.Type, "record type should be preserved") - records, err := e.GetAll(ctx) + records, err := e.List(ctx, "") if !assert.NoError(t, err) { return } diff --git a/pkg/storage/redis/option.go b/pkg/storage/redis/option.go index aec80df07..8be15587a 100644 --- a/pkg/storage/redis/option.go +++ b/pkg/storage/redis/option.go @@ -1,13 +1,35 @@ package redis -import "crypto/tls" +import ( + "crypto/tls" +) + +type dbConfig struct { + tls *tls.Config + recordType string +} // Option customizes a DB. -type Option func(*DB) +type Option func(*dbConfig) + +// WithRecordType sets the record type in the config. +func WithRecordType(recordType string) Option { + return func(cfg *dbConfig) { + cfg.recordType = recordType + } +} // WithTLSConfig sets the tls.Config which DB uses. func WithTLSConfig(tlsConfig *tls.Config) Option { - return func(db *DB) { - db.tlsConfig = tlsConfig + return func(cfg *dbConfig) { + cfg.tls = tlsConfig } } + +func getConfig(options ...Option) *dbConfig { + cfg := new(dbConfig) + for _, o := range options { + o(cfg) + } + return cfg +} diff --git a/pkg/storage/redis/redis.go b/pkg/storage/redis/redis.go index 52ee73799..84d123661 100644 --- a/pkg/storage/redis/redis.go +++ b/pkg/storage/redis/redis.go @@ -1,404 +1,393 @@ -// Package redis is the redis database, implements storage.Backend interface. +// Package redis implements the storage.Backend interface for redis. package redis import ( "context" - "crypto/tls" + "errors" "fmt" - "net" "strconv" "sync" "time" - "github.com/cenkalti/backoff/v4" + redis "github.com/go-redis/redis/v8" "github.com/golang/protobuf/proto" - "github.com/golang/protobuf/ptypes" - "github.com/gomodule/redigo/redis" "google.golang.org/protobuf/types/known/anypb" + "google.golang.org/protobuf/types/known/timestamppb" "github.com/pomerium/pomerium/config" "github.com/pomerium/pomerium/internal/log" + "github.com/pomerium/pomerium/internal/signal" "github.com/pomerium/pomerium/internal/telemetry/metrics" "github.com/pomerium/pomerium/internal/telemetry/trace" "github.com/pomerium/pomerium/pkg/grpc/databroker" - "github.com/pomerium/pomerium/pkg/storage" ) -// Name is the storage type name for redis backend. +// Name of the storage backend. const Name = config.StorageRedisName -var _ storage.Backend = (*DB)(nil) +const ( + maxTransactionRetries = 100 + watchPollInterval = 30 * time.Second +) -// DB wraps redis conn to interact with redis server. +// custom errors +var ( + ErrExceededMaxRetries = errors.New("redis: transaction reached maximum number of retries") +) + +// DB implements the storage.Backend on top of redis. type DB struct { - pool *redis.Pool - recordType string - lastVersionKey string - lastVersionChannelKey string - versionSet string - deletedSet string - tlsConfig *tls.Config - notifyChMu sync.Mutex + cfg *dbConfig + + client *redis.Client closeOnce sync.Once closed chan struct{} } -// New returns new DB instance. -func New(rawURL, recordType string, opts ...Option) (*DB, error) { +// New creates a new redis storage backend. +func New(rawURL string, options ...Option) (*DB, error) { db := &DB{ - recordType: recordType, - versionSet: recordType + "_version_set", - deletedSet: recordType + "_deleted_set", - lastVersionKey: recordType + "_last_version", - lastVersionChannelKey: recordType + "_last_version_ch", - closed: make(chan struct{}), + cfg: getConfig(options...), + closed: make(chan struct{}), } - - for _, o := range opts { - o(db) + opts, err := redis.ParseURL(rawURL) + if err != nil { + return nil, err } - db.pool = &redis.Pool{ - Wait: true, - Dial: func() (redis.Conn, error) { - c, err := redis.DialURL(rawURL, redis.DialTLSConfig(db.tlsConfig)) - if err != nil { - return nil, fmt.Errorf(`redis.DialURL(): %w`, err) - } - return c, nil - }, - TestOnBorrow: func(c redis.Conn, t time.Time) error { - if time.Since(t) < time.Minute { - return nil - } - _, err := c.Do("PING") - if err != nil { - return fmt.Errorf(`c.Do("PING"): %w`, err) - } - return nil - }, - MaxIdle: 64, - IdleTimeout: time.Minute, - MaxActive: 128, + // when using TLS, the TLS config will not be set to nil, in which case we replace it with our own + if opts.TLSConfig != nil { + opts.TLSConfig = db.cfg.tls } - metrics.AddRedisMetrics(db.pool.Stats) + db.client = redis.NewClient(opts) return db, nil } -// Close closes the redis db connection. +// ClearDeleted clears all the deleted records older than the cutoff time. +func (db *DB) ClearDeleted(ctx context.Context, cutoff time.Time) { + var err error + + _, span := trace.StartSpan(ctx, "databroker.redis.ClearDeleted") + defer span.End() + defer func(start time.Time) { recordOperation(ctx, start, "clear_deleted", err) }(time.Now()) + + ids, _ := db.client.SMembers(ctx, formatDeletedSetKey(db.cfg.recordType)).Result() + records, _ := redisGetRecords(ctx, db.client, db.cfg.recordType, ids) + _, err = db.client.Pipelined(ctx, func(p redis.Pipeliner) error { + for _, record := range records { + if record.GetDeletedAt().AsTime().Before(cutoff) { + p.HDel(ctx, formatRecordsKey(db.cfg.recordType), record.GetId()) + p.ZRem(ctx, formatVersionSetKey(db.cfg.recordType), record.GetId()) + p.SRem(ctx, formatDeletedSetKey(db.cfg.recordType), record.GetId()) + } + } + return nil + }) +} + +// Close closes the underlying redis connection and any watchers. func (db *DB) Close() error { + var err error db.closeOnce.Do(func() { + err = db.client.Close() close(db.closed) }) - return nil + return err } -// Put sets new record for given id with input data. -func (db *DB) Put(ctx context.Context, id string, data *anypb.Any) (err error) { - c := db.pool.Get() - _, span := trace.StartSpan(ctx, "databroker.redis.Put") +// Delete marks a record as deleted. +func (db *DB) Delete(ctx context.Context, id string) (err error) { + _, span := trace.StartSpan(ctx, "databroker.redis.Delete") defer span.End() - defer recordOperation(ctx, time.Now(), "put", err) - defer c.Close() + defer func(start time.Time) { recordOperation(ctx, start, "delete", err) }(time.Now()) - record, err := db.Get(ctx, id) - if err != nil { - record = new(databroker.Record) - record.CreatedAt = ptypes.TimestampNow() - } + var record *databroker.Record + err = db.incrementVersion(ctx, + func(tx *redis.Tx, version int64) error { + var err error + record, err = redisGetRecord(ctx, tx, db.cfg.recordType, id) + if errors.Is(err, redis.Nil) { + // nothing to do, as the record doesn't exist + return nil + } else if err != nil { + return err + } - lastVersion, err := redis.Int64(c.Do("INCR", db.lastVersionKey)) - if err != nil { - return err - } - record.Data = data - record.ModifiedAt = ptypes.TimestampNow() - record.Type = db.recordType - record.Id = id - record.Version = fmt.Sprintf("%012X", lastVersion) - b, err := proto.Marshal(record) - if err != nil { - return err - } - cmds := []map[string][]interface{}{ - {"MULTI": nil}, - {"HSET": {db.recordType, id, string(b)}}, - {"ZADD": {db.versionSet, lastVersion, id}}, - {"PUBLISH": {db.lastVersionChannelKey, lastVersion}}, - } - if err := db.tx(c, cmds); err != nil { - return err - } - return nil + // mark it as deleted + record.DeletedAt = timestamppb.Now() + + return nil + }, + func(p redis.Pipeliner, version int64) error { + err := redisSetRecord(ctx, p, db.cfg.recordType, record) + if err != nil { + return err + } + + // add it to the collection of deleted entries + p.SAdd(ctx, formatDeletedSetKey(db.cfg.recordType), record.GetId()) + return nil + }) + return err } -// Get retrieves a record from redis. -func (db *DB) Get(ctx context.Context, id string) (rec *databroker.Record, err error) { - c := db.pool.Get() +// Get gets a record. +func (db *DB) Get(ctx context.Context, id string) (record *databroker.Record, err error) { _, span := trace.StartSpan(ctx, "databroker.redis.Get") defer span.End() - defer recordOperation(ctx, time.Now(), "get", err) - defer c.Close() + defer func(start time.Time) { recordOperation(ctx, start, "get", err) }(time.Now()) - b, err := redis.Bytes(c.Do("HGET", db.recordType, id)) + record, err = redisGetRecord(ctx, db.client, db.cfg.recordType, id) + return record, err +} + +// List lists all the records changed since the sinceVersion. Records are sorted in version order. +func (db *DB) List(ctx context.Context, sinceVersion string) (records []*databroker.Record, err error) { + _, span := trace.StartSpan(ctx, "databroker.redis.List") + defer span.End() + defer func(start time.Time) { recordOperation(ctx, start, "list", err) }(time.Now()) + + var ids []string + ids, err = redisListIDsSince(ctx, db.client, db.cfg.recordType, sinceVersion) + if err != nil { + return nil, err + } + records, err = redisGetRecords(ctx, db.client, db.cfg.recordType, ids) + return records, err +} + +// Put updates a record. +func (db *DB) Put(ctx context.Context, id string, data *anypb.Any) (err error) { + _, span := trace.StartSpan(ctx, "databroker.redis.Put") + defer span.End() + defer func(start time.Time) { recordOperation(ctx, start, "put", err) }(time.Now()) + + var record *databroker.Record + err = db.incrementVersion(ctx, + func(tx *redis.Tx, version int64) error { + var err error + record, err = redisGetRecord(ctx, db.client, db.cfg.recordType, id) + if errors.Is(err, redis.Nil) { + record = new(databroker.Record) + record.CreatedAt = timestamppb.Now() + } else if err != nil { + return err + } + + record.ModifiedAt = timestamppb.Now() + record.Type = db.cfg.recordType + record.Id = id + record.Data = data + record.Version = formatVersion(version) + + return nil + }, + func(p redis.Pipeliner, version int64) error { + return redisSetRecord(ctx, p, db.cfg.recordType, record) + }) + return err +} + +// Watch returns a channel that is signaled any time the last version is incremented (ie on Put/Delete). +func (db *DB) Watch(ctx context.Context) <-chan struct{} { + s := signal.New() + ch := s.Bind() + go func() { + defer s.Unbind(ch) + defer close(ch) + + // force a check + poll := time.NewTicker(watchPollInterval) + defer poll.Stop() + + // use pub/sub for quicker notify + pubsub := db.client.Subscribe(ctx, formatLastVersionChannelKey(db.cfg.recordType)) + defer func() { _ = pubsub.Close() }() + pubsubCh := pubsub.Channel() + + var lastVersion int64 + + for { + v, err := redisGetLastVersion(ctx, db.client, db.cfg.recordType) + if err != nil { + log.Error().Err(err).Msg("redis: error retrieving last version") + } else if v != lastVersion { + // don't broadcast the first time + if lastVersion != 0 { + s.Broadcast() + } + lastVersion = v + } + + select { + case <-ctx.Done(): + return + case <-db.closed: + return + case <-poll.C: + case <-pubsubCh: + // re-check + } + } + }() + return ch +} + +// incrementVersion increments the last version key, runs the code in `query`, then attempts to commit the code in +// `commit`. If the last version changes in the interim, we will retry the transaction. +func (db *DB) incrementVersion(ctx context.Context, + query func(tx *redis.Tx, version int64) error, + commit func(p redis.Pipeliner, version int64) error, +) error { + // code is modeled on https://pkg.go.dev/github.com/go-redis/redis/v8#example-Client.Watch + txf := func(tx *redis.Tx) error { + version, err := redisGetLastVersion(ctx, tx, db.cfg.recordType) + if err != nil { + return err + } + version++ + + err = query(tx, version) + if err != nil { + return err + } + + // the `commit` code is run in a transaction so that the EXEC cmd will run for the original redis watch + _, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error { + err := commit(p, version) + if err != nil { + return err + } + p.Set(ctx, formatLastVersionKey(db.cfg.recordType), version, 0) + p.Publish(ctx, formatLastVersionChannelKey(db.cfg.recordType), version) + return nil + }) + return err + } + + for i := 0; i < maxTransactionRetries; i++ { + err := db.client.Watch(ctx, txf, formatLastVersionKey(db.cfg.recordType)) + if errors.Is(err, redis.TxFailedErr) { + continue // retry + } else if err != nil { + return err + } + + return nil // tx was successful + } + + return ErrExceededMaxRetries +} + +func redisGetLastVersion(ctx context.Context, c redis.Cmdable, recordType string) (int64, error) { + version, err := c.Get(ctx, formatLastVersionKey(recordType)).Int64() + if errors.Is(err, redis.Nil) { + version = 0 + } else if err != nil { + return 0, err + } + return version, nil +} + +func redisGetRecord(ctx context.Context, c redis.Cmdable, recordType string, id string) (*databroker.Record, error) { + records, err := redisGetRecords(ctx, c, recordType, []string{id}) + if err != nil { + return nil, err + } else if len(records) < 1 { + return nil, redis.Nil + } + return records[0], nil +} + +func redisGetRecords(ctx context.Context, c redis.Cmdable, recordType string, ids []string) ([]*databroker.Record, error) { + if len(ids) == 0 { + return nil, nil + } + + results, err := c.HMGet(ctx, formatRecordsKey(recordType), ids...).Result() if err != nil { return nil, err } - return db.toPbRecord(b) + records := make([]*databroker.Record, 0, len(results)) + for _, result := range results { + // results are returned as either nil or a string + if result == nil { + continue + } + rawstr, ok := result.(string) + if !ok { + continue + } + var record databroker.Record + err := proto.Unmarshal([]byte(rawstr), &record) + if err != nil { + continue + } + records = append(records, &record) + } + return records, nil } -// GetAll retrieves all records from redis. -func (db *DB) GetAll(ctx context.Context) (recs []*databroker.Record, err error) { - _, span := trace.StartSpan(ctx, "databroker.redis.GetAll") - defer span.End() - defer recordOperation(ctx, time.Now(), "get_all", err) - return db.getAll(ctx, func(record *databroker.Record) bool { return true }) +func redisListIDsSince(ctx context.Context, + c redis.Cmdable, recordType string, + sinceVersion string, +) ([]string, error) { + v, err := strconv.ParseInt(sinceVersion, 16, 64) + if err != nil { + v = 0 + } + rng := &redis.ZRangeBy{ + Min: fmt.Sprintf("(%d", v), + Max: "+inf", + } + return c.ZRangeByScore(ctx, formatVersionSetKey(recordType), rng).Result() } -// List retrieves all records since given version. -// -// "version" is in hex format, invalid version will be treated as 0. -func (db *DB) List(ctx context.Context, sinceVersion string) (rec []*databroker.Record, err error) { - c := db.pool.Get() - _, span := trace.StartSpan(ctx, "databroker.redis.List") - defer span.End() - defer recordOperation(ctx, time.Now(), "list", err) - defer c.Close() - - v, err := strconv.ParseUint(sinceVersion, 16, 64) +func redisSetRecord(ctx context.Context, p redis.Pipeliner, recordType string, record *databroker.Record) error { + v, err := strconv.ParseInt(record.GetVersion(), 16, 64) if err != nil { v = 0 } - ids, err := redis.Strings(c.Do("ZRANGEBYSCORE", db.versionSet, fmt.Sprintf("(%d", v), "+inf")) - if err != nil { - return nil, err - } - - pbRecords := make([]*databroker.Record, 0, len(ids)) - for _, id := range ids { - b, err := redis.Bytes(c.Do("HGET", db.recordType, id)) - if err != nil { - return nil, err - } - pbRecord, err := db.toPbRecord(b) - if err != nil { - return nil, err - } - pbRecords = append(pbRecords, pbRecord) - } - return pbRecords, nil -} - -// Delete sets a record DeletedAt field and set its TTL. -func (db *DB) Delete(ctx context.Context, id string) (err error) { - c := db.pool.Get() - _, span := trace.StartSpan(ctx, "databroker.redis.Delete") - defer span.End() - defer recordOperation(ctx, time.Now(), "delete", err) - defer c.Close() - - r, err := db.Get(ctx, id) - if err != nil { - return fmt.Errorf("failed to get record: %w", err) - } - - lastVersion, err := redis.Int64(c.Do("INCR", db.lastVersionKey)) + raw, err := proto.Marshal(record) if err != nil { return err } - r.DeletedAt = ptypes.TimestampNow() - r.Version = fmt.Sprintf("%012X", lastVersion) - b, err := proto.Marshal(r) - if err != nil { - return err - } - cmds := []map[string][]interface{}{ - {"MULTI": nil}, - {"HSET": {db.recordType, id, string(b)}}, - {"SADD": {db.deletedSet, id}}, - {"ZADD": {db.versionSet, lastVersion, id}}, - {"PUBLISH": {db.lastVersionChannelKey, lastVersion}}, - } - if err := db.tx(c, cmds); err != nil { - return err - } + // store the record in the hash + p.HSet(ctx, formatRecordsKey(recordType), record.GetId(), string(raw)) + // set its score for sorting by version + p.ZAdd(ctx, formatVersionSetKey(recordType), &redis.Z{ + Score: float64(v), + Member: record.GetId(), + }) + return nil } -// ClearDeleted clears all the currently deleted records older than the given cutoff. -func (db *DB) ClearDeleted(ctx context.Context, cutoff time.Time) { - c := db.pool.Get() - _, span := trace.StartSpan(ctx, "databroker.redis.ClearDeleted") - defer span.End() - var opErr error - defer func(startTime time.Time) { - recordOperation(ctx, startTime, "clear_deleted", opErr) - }(time.Now()) - defer c.Close() - - ids, _ := redis.Strings(c.Do("SMEMBERS", db.deletedSet)) - for _, id := range ids { - b, _ := redis.Bytes(c.Do("HGET", db.recordType, id)) - record, err := db.toPbRecord(b) - if err != nil { - continue - } - - ts, _ := ptypes.Timestamp(record.DeletedAt) - if ts.Before(cutoff) { - cmds := []map[string][]interface{}{ - {"MULTI": nil}, - {"HDEL": {db.recordType, id}}, - {"ZREM": {db.versionSet, id}}, - {"SREM": {db.deletedSet, id}}, - } - opErr = db.tx(c, cmds) - } - } +func formatDeletedSetKey(recordType string) string { + return fmt.Sprintf("%s_deleted_set", recordType) } -// doNotifyLoop receives event from redis and send signal to the channel. -func (db *DB) doNotifyLoop(ctx context.Context, ch chan struct{}) { - eb := backoff.NewExponentialBackOff() - psConn := db.pool.Get() - psc := redis.PubSubConn{Conn: psConn} - defer func(psc *redis.PubSubConn) { - psc.Conn.Close() - }(&psc) - - if err := psc.Subscribe(db.lastVersionChannelKey); err != nil { - log.Error().Err(err).Msg("failed to subscribe to version set channel") - return - } - for { - select { - case <-db.closed: - return - case <-ctx.Done(): - return - default: - } - switch v := psc.Receive().(type) { - case redis.Message: - log.Debug().Str("action", string(v.Data)).Msg("got redis message") - recordOperation(ctx, time.Now(), "sub_received", nil) - - select { - case <-db.closed: - return - case <-ctx.Done(): - log.Warn().Err(ctx.Err()).Msg("context done, stop receive from redis channel") - return - default: - db.notifyChMu.Lock() - select { - case <-db.closed: - db.notifyChMu.Unlock() - return - case <-ctx.Done(): - db.notifyChMu.Unlock() - log.Warn().Err(ctx.Err()).Msg("context done while holding notify lock, stop receive from redis channel") - return - case ch <- struct{}{}: - } - db.notifyChMu.Unlock() - } - case error: - log.Warn().Err(v).Msg("failed to receive from redis channel") - recordOperation(ctx, time.Now(), "sub_received", v) - if _, ok := v.(net.Error); ok { - return - } - time.Sleep(eb.NextBackOff()) - log.Warn().Msg("retry with new connection") - _ = psc.Conn.Close() - psc.Conn = db.pool.Get() - _ = psc.Subscribe(db.lastVersionChannelKey) - } - } +func formatLastVersionChannelKey(recordType string) string { + return fmt.Sprintf("%s_last_version_ch", recordType) } -// watch runs the doNotifyLoop. It returns when ctx was done or doNotifyLoop exits. -func (db *DB) watch(ctx context.Context, ch chan struct{}) { - defer func() { - db.notifyChMu.Lock() - close(ch) - db.notifyChMu.Unlock() - }() - done := make(chan struct{}) - go func() { - defer close(done) - db.doNotifyLoop(ctx, ch) - }() - select { - case <-db.closed: - case <-ctx.Done(): - case <-done: - } +func formatLastVersionKey(recordType string) string { + return fmt.Sprintf("%s_last_version", recordType) } -// Watch returns a channel to the caller, when there is a change to the version set, -// sending message to the channel to notify the caller. -func (db *DB) Watch(ctx context.Context) <-chan struct{} { - ch := make(chan struct{}) - go db.watch(ctx, ch) - return ch +func formatRecordsKey(recordType string) string { + return recordType } -func (db *DB) getAll(_ context.Context, filter func(record *databroker.Record) bool) ([]*databroker.Record, error) { - c := db.pool.Get() - defer c.Close() - iter := 0 - records := make([]*databroker.Record, 0) - for { - arr, err := redis.Values(c.Do("HSCAN", db.recordType, iter, "MATCH", "*")) - if err != nil { - return nil, err - } - - iter, _ = redis.Int(arr[0], nil) - pairs, _ := redis.StringMap(arr[1], nil) - - for _, v := range pairs { - record, err := db.toPbRecord([]byte(v)) - if err != nil { - return nil, err - } - if filter(record) { - records = append(records, record) - } - } - - if iter == 0 { - break - } - } - - return records, nil +func formatVersion(version int64) string { + return fmt.Sprintf("%012d", version) } -func (db *DB) toPbRecord(b []byte) (*databroker.Record, error) { - record := &databroker.Record{} - if err := proto.Unmarshal(b, record); err != nil { - return nil, err - } - return record, nil -} - -func (db *DB) tx(c redis.Conn, commands []map[string][]interface{}) error { - for _, m := range commands { - for cmd, args := range m { - if err := c.Send(cmd, args...); err != nil { - return err - } - } - } - - _, err := c.Do("EXEC") - return err +func formatVersionSetKey(recordType string) string { + return fmt.Sprintf("%s_version_set", recordType) } func recordOperation(ctx context.Context, startTime time.Time, operation string, err error) { diff --git a/pkg/storage/redis/redis_test.go b/pkg/storage/redis/redis_test.go index 1a0d19613..832cc2f7d 100644 --- a/pkg/storage/redis/redis_test.go +++ b/pkg/storage/redis/redis_test.go @@ -12,7 +12,6 @@ import ( "testing" "time" - "github.com/gomodule/redigo/redis" "github.com/ory/dockertest/v3" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -24,13 +23,8 @@ import ( var db *DB -func cleanup(c redis.Conn, db *DB, t *testing.T) { - require.NoError(t, c.Send("MULTI")) - require.NoError(t, c.Send("DEL", db.recordType)) - require.NoError(t, c.Send("DEL", db.versionSet)) - require.NoError(t, c.Send("DEL", db.deletedSet)) - _, err := c.Do("EXEC") - require.NoError(t, err) +func cleanup(ctx context.Context, db *DB, t *testing.T) { + require.NoError(t, db.client.FlushAll(ctx).Err()) } func tlsConfig(rawURL string, t *testing.T) *tls.Config { @@ -59,6 +53,7 @@ func runWithRedisDockerImage(t *testing.T, runOpts *dockertest.RunOptions, withT if err != nil { t.Fatalf("Could not start resource: %s", err) } + resource.Expire(30) defer func() { if err := pool.Purge(resource); err != nil { @@ -73,11 +68,11 @@ func runWithRedisDockerImage(t *testing.T, runOpts *dockertest.RunOptions, withT address := fmt.Sprintf(scheme+"://localhost:%s/0", resource.GetPort("6379/tcp")) if err := pool.Retry(func() error { var err error - db, err = New(address, "record_type", WithTLSConfig(tlsConfig(address, t))) + db, err = New(address, WithRecordType("record_type"), WithTLSConfig(tlsConfig(address, t))) if err != nil { return err } - _, err = db.pool.Get().Do("PING") + err = db.client.Ping(context.Background()).Err() return err }); err != nil { t.Fatalf("Could not connect to docker: %s", err) @@ -128,10 +123,6 @@ func testDB(t *testing.T) { } ids := []string{"a", "b", "c"} id := ids[0] - c := db.pool.Get() - defer c.Close() - - ch := db.Watch(ctx) t.Run("get missing record", func(t *testing.T) { record, err := db.Get(ctx, id) @@ -165,24 +156,8 @@ func testDB(t *testing.T) { assert.Error(t, err) assert.Nil(t, record) }) - t.Run("get all", func(t *testing.T) { - records, err := db.GetAll(ctx) - assert.NoError(t, err) - assert.Len(t, records, 0) - - for i, id := range ids { - data, _ := anypb.New(users[i]) - assert.NoError(t, db.Put(ctx, id, data)) - } - records, err = db.GetAll(ctx) - assert.NoError(t, err) - assert.Len(t, records, len(ids)) - for _, id := range ids { - _, _ = c.Do("DEL", id) - } - }) t.Run("list", func(t *testing.T) { - cleanup(c, db, t) + cleanup(ctx, db, t) for i := 0; i < 10; i++ { id := fmt.Sprintf("%02d", i) @@ -193,20 +168,23 @@ func testDB(t *testing.T) { records, err := db.List(ctx, "") assert.NoError(t, err) assert.Len(t, records, 10) - records, err = db.List(ctx, "00000000000A") + records, err = db.List(ctx, "000000000005") assert.NoError(t, err) assert.Len(t, records, 5) - records, err = db.List(ctx, "00000000000F") + records, err = db.List(ctx, "000000000010") assert.NoError(t, err) assert.Len(t, records, 0) }) + t.Run("watch", func(t *testing.T) { + ch := db.Watch(ctx) + time.Sleep(time.Second) - expectedNumEvents := 14 - actualNumEvents := 0 - for range ch { - actualNumEvents++ - if actualNumEvents == expectedNumEvents { - cancelFunc() + go db.Put(ctx, "WATCH", new(anypb.Any)) + + select { + case <-ch: + case <-time.After(time.Second * 10): + t.Error("expected watch signal on put") } - } + }) } diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 1d34b3e47..a2a4a5679 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -24,9 +24,6 @@ type Backend interface { // Get is used to retrieve a record. Get(ctx context.Context, id string) (*databroker.Record, error) - // GetAll is used to retrieve all the records. - GetAll(ctx context.Context) ([]*databroker.Record, error) - // List is used to retrieve all the records since a version. List(ctx context.Context, sinceVersion string) ([]*databroker.Record, error)