diff --git a/go.mod b/go.mod index bae38547..e08dc6b7 100644 --- a/go.mod +++ b/go.mod @@ -28,7 +28,7 @@ require ( github.com/huandu/xstrings v1.4.0 github.com/lib/pq v1.10.9 github.com/mattn/go-sqlite3 v1.14.22 - github.com/minio/minio-go/v6 v6.0.57 + github.com/minio/minio-go/v7 v7.0.70 github.com/mitchellh/copystructure v1.2.0 github.com/mitchellh/go-homedir v1.1.0 github.com/mitchellh/mapstructure v1.5.0 @@ -71,6 +71,7 @@ require ( github.com/distribution/reference v0.6.0 // indirect github.com/docker/go-connections v0.5.0 // indirect github.com/docker/go-units v0.5.0 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect github.com/emicklei/go-restful/v3 v3.12.0 // indirect github.com/emirpasic/gods v1.18.1 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect @@ -82,6 +83,7 @@ require ( github.com/go-openapi/jsonpointer v0.21.0 // indirect github.com/go-openapi/jsonreference v0.21.0 // indirect github.com/go-openapi/swag v0.23.0 // indirect + github.com/goccy/go-json v0.10.2 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect github.com/golang/protobuf v1.5.4 // indirect @@ -107,7 +109,6 @@ require ( github.com/mattn/go-colorable v0.1.13 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/minio/md5-simd v1.1.2 // indirect - github.com/minio/sha256-simd v1.0.0 // indirect github.com/mitchellh/go-wordwrap v1.0.1 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect @@ -122,6 +123,7 @@ require ( github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.0 // indirect github.com/pjbgf/sha1cd v0.3.0 // indirect + github.com/rs/xid v1.5.0 // indirect github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect github.com/shopspring/decimal v1.4.0 // indirect github.com/sirupsen/logrus v1.9.3 // indirect diff --git a/go.sum b/go.sum index d729e575..45249d90 100644 --- a/go.sum +++ b/go.sum @@ -66,7 +66,8 @@ github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= -github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/elazarl/go-bindata-assetfs v1.0.1 h1:m0kkaHRKEu7tUIUFVwhGGGYClXvyl4RE03qmvRTNfbw= github.com/elazarl/go-bindata-assetfs v1.0.1/go.mod h1:v+YaWX3bdea5J/mo8dSETolEo7R71Vk1u8bnjau5yw4= github.com/elazarl/goproxy v0.0.0-20230808193330-2592e75ae04a h1:mATvB/9r/3gvcejNsXKSkQ6lcIaNec2nyfOdlTBR2lU= @@ -115,6 +116,8 @@ github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1v github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/go-test/deep v1.0.3 h1:ZrJSEWsXzPOxaZnFteGEfooLba+ju3FYIbOrS+rQd68= github.com/go-test/deep v1.0.3/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA= +github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= +github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= github.com/gofrs/uuid v4.4.0+incompatible h1:3qXRTX8/NbyulANqlc0lchS1gqAVxRgsuW1YrTJupqA= github.com/gofrs/uuid v4.4.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= @@ -148,7 +151,6 @@ github.com/google/pprof v0.0.0-20240424215950-a892ee059fd6/go.mod h1:kf6iHlnVGwg github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= github.com/gorilla/csrf v1.7.2 h1:oTUjx0vyf2T+wkrx09Trsev1TE+/EbDAeHtSTbtC2eI= github.com/gorilla/csrf v1.7.2/go.mod h1:F1Fj3KG23WYHE6gozCmBAezKookxbIvUJT+121wTuLk= github.com/gorilla/handlers v1.5.2 h1:cLTUSsNkgcwhgRqvCNmdbRWG0A3N4F+M2nWKdScwyEE= @@ -190,10 +192,8 @@ github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOl github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= -github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= -github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs= github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8= github.com/kevinburke/ssh_config v1.2.0 h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4= @@ -202,12 +202,9 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU= github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= -github.com/klauspost/cpuid v1.2.3/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= -github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM= github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= -github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -227,14 +224,10 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU= github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= -github.com/minio/md5-simd v1.1.0/go.mod h1:XpBqgZULrMYD3R+M28PcmP0CkI7PEMzB3U77ZrKZ0Gw= github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= -github.com/minio/minio-go/v6 v6.0.57 h1:ixPkbKkyD7IhnluRgQpGSpHdpvNVaW6OD5R9IAO/9Tw= -github.com/minio/minio-go/v6 v6.0.57/go.mod h1:5+R/nM9Pwrh0vqF+HbYYDQ84wdUFPyXHkrdT4AIkifM= -github.com/minio/sha256-simd v0.1.1/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM= -github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g= -github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM= +github.com/minio/minio-go/v7 v7.0.70 h1:1u9NtMgfK1U42kUxcsl5v0yj6TEOPR497OAQxpJnn2g= +github.com/minio/minio-go/v7 v7.0.70/go.mod h1:4yBA8v80xGA30cfM3fz0DKYMXunWl/AV/6tWEs9ryzo= github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFWgYaq1OVrnRcwhnw= github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw= github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s= @@ -262,7 +255,6 @@ github.com/moby/term v0.0.0-20221205130635-1aeaba878587/go.mod h1:8FzsFHVUBGZdbD github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= -github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= @@ -289,6 +281,7 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M= github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA= github.com/rs/xid v1.3.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= +github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc= github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= github.com/rs/zerolog v1.26.1/go.mod h1:/wSSJWX7lVrsOwlbyTRSOJvqRlc+WjWlfes+CiJ+tmc= github.com/rs/zerolog v1.32.0 h1:keLypqrlIjaFsbmJOBdB/qvyF8KEtCWHwobLp5l/mQ0= @@ -303,14 +296,11 @@ github.com/sgotti/gexpect v0.0.0-20210315095146-1ec64e69809b/go.mod h1:iw90eoXMZ github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o= github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= -github.com/sirupsen/logrus v1.5.0/go.mod h1:+F7Ogzej0PZc/94MaYx/nvG9jOFMD2osvC3s+Squfpo= github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/skeema/knownhosts v1.2.2 h1:Iug2P4fLmDw9f41PB6thxUkNUkJzB5i+1/exaj40L3A= github.com/skeema/knownhosts v1.2.2/go.mod h1:xYbVRSPxqBZFrdmDyMmsOs+uX1UZC3nTN3ThzgDxUwo= -github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= -github.com/smartystreets/goconvey v0.0.0-20190330032615-68dc04aab96a/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= github.com/sorintlab/errors v0.0.0-20230328073828-75145c80e43d h1:OOO0G7VT35z8rN9pAnK0rBueL4HqZItHxEjWFP1DjlA= github.com/sorintlab/errors v0.0.0-20230328073828-75145c80e43d/go.mod h1:y6BMSfuxo7hQ7iJCppECcFn2jg5PKhkWvF5TMC6Q+3U= github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE= @@ -360,7 +350,6 @@ go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v8 go.starlark.net v0.0.0-20240517230649-3792562d0b7f h1:APah0oANPHA7m/z/1Ngcccc+BEO/dmLcEfrzHAQQY6w= go.starlark.net v0.0.0-20240517230649-3792562d0b7f/go.mod h1:YKMCv9b1WrfWmeqdV5MAuEHWsu5iC+fe6kYl2sQjdI8= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= -golang.org/x/crypto v0.0.0-20190513172903-22d7a77e9e5f/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8= @@ -379,9 +368,7 @@ golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA= golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= -golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= @@ -406,7 +393,6 @@ golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -446,7 +432,6 @@ golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk= golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= -golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= @@ -474,7 +459,6 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntN gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= -gopkg.in/ini.v1 v1.42.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA= gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/warnings.v0 v0.1.2 h1:wFXVbFY8DY5/xOe1ECiWdKCzZlxgshcYVNkBHstARME= diff --git a/internal/common/common.go b/internal/common/common.go index c003e59c..4661924a 100644 --- a/internal/common/common.go +++ b/internal/common/common.go @@ -15,6 +15,7 @@ package common import ( + "context" "io" "net/url" "os" @@ -73,11 +74,9 @@ func WriteFileAtomic(filename string, data []byte, perm os.FileMode) error { }) } -func NewObjectStorage(c *config.ObjectStorage) (*objectstorage.ObjStorage, error) { - var ( - err error - ost objectstorage.Storage - ) +func NewObjectStorage(ctx context.Context, c *config.ObjectStorage) (objectstorage.ObjStorage, error) { + var err error + var ost objectstorage.ObjStorage switch c.Type { case config.ObjectStorageTypePosix: @@ -100,11 +99,11 @@ func NewObjectStorage(c *config.ObjectStorage) (*objectstorage.ObjStorage, error return nil, errors.Errorf("wrong s3 endpoint scheme %q (must be http or https)", u.Scheme) } } - ost, err = objectstorage.NewS3(c.Bucket, c.Location, endpoint, c.AccessKey, c.SecretAccessKey, secure) + ost, err = objectstorage.NewS3(ctx, c.Bucket, c.Location, endpoint, c.AccessKey, c.SecretAccessKey, secure) if err != nil { return nil, errors.Wrapf(err, "failed to create s3 object storage") } } - return objectstorage.NewObjStorage(ost, "/"), nil + return ost, nil } diff --git a/internal/objectstorage/objectstorage.go b/internal/objectstorage/objectstorage.go index 41e5e4d6..ae29ccdf 100644 --- a/internal/objectstorage/objectstorage.go +++ b/internal/objectstorage/objectstorage.go @@ -15,6 +15,7 @@ package objectstorage import ( + "context" "io" "time" @@ -23,17 +24,17 @@ import ( "agola.io/agola/internal/util" ) -type Storage interface { - Stat(filepath string) (*ObjectInfo, error) - ReadObject(filepath string) (ReadSeekCloser, error) +type ObjStorage interface { + Stat(ctx context.Context, filepath string) (*ObjectInfo, error) + ReadObject(ctx context.Context, filepath string) (ReadSeekCloser, error) // WriteObject atomically writes an object. If size is greater or equal to // zero then only size bytes will be read from data and wrote. If size is // less than zero data will be wrote until EOF. When persist is true the // implementation must ensure that data is persisted to the underlying // storage. - WriteObject(filepath string, data io.Reader, size int64, persist bool) error - DeleteObject(filepath string) error - List(prefix, startWith, delimiter string, doneCh <-chan struct{}) <-chan ObjectInfo + WriteObject(ctx context.Context, filepath string, data io.Reader, size int64, persist bool) error + DeleteObject(ctx context.Context, filepath string) error + List(ctx context.Context, prefix, startAfter string, recursive bool) <-chan ObjectInfo } type ErrNotExist struct { @@ -64,26 +65,3 @@ type ObjectInfo struct { Err error } - -// ObjStorage wraps a Storage providing additional helper functions -type ObjStorage struct { - Storage - delimiter string -} - -func NewObjStorage(s Storage, delimiter string) *ObjStorage { - return &ObjStorage{Storage: s, delimiter: delimiter} -} - -func (s *ObjStorage) Delimiter() string { - return s.delimiter -} - -func (s *ObjStorage) List(prefix, startWith string, recursive bool, doneCh <-chan struct{}) <-chan ObjectInfo { - delimiter := s.delimiter - if recursive { - delimiter = "" - } - - return s.Storage.List(prefix, startWith, delimiter, doneCh) -} diff --git a/internal/objectstorage/objectstorage_test.go b/internal/objectstorage/objectstorage_test.go index 65de3237..e47c308b 100644 --- a/internal/objectstorage/objectstorage_test.go +++ b/internal/objectstorage/objectstorage_test.go @@ -16,6 +16,7 @@ package objectstorage import ( "bytes" + "context" "fmt" "os" "path" @@ -29,14 +30,10 @@ import ( "agola.io/agola/internal/testutil" ) -func setupPosix(t *testing.T, dir string) (*PosixStorage, error) { +func setupPosix(dir string) (*PosixStorage, error) { return NewPosix(path.Join(dir, "posix")) } -func setupPosixFlat(t *testing.T, dir string) (*PosixFlatStorage, error) { - return NewPosixFlat(path.Join(dir, "posixflat")) -} - func setupS3(t *testing.T, dir string) (*S3Storage, error) { minioEndpoint := os.Getenv("MINIO_ENDPOINT") minioAccessKey := os.Getenv("MINIO_ACCESSKEY") @@ -46,16 +43,13 @@ func setupS3(t *testing.T, dir string) (*S3Storage, error) { return nil, nil } - return NewS3(filepath.Base(dir), "", minioEndpoint, minioAccessKey, minioSecretKey, false) + return NewS3(context.Background(), filepath.Base(dir), "", minioEndpoint, minioAccessKey, minioSecretKey, false) } func TestList(t *testing.T) { dir := t.TempDir() - ps, err := setupPosix(t, dir) - testutil.NilError(t, err) - - pfs, err := setupPosixFlat(t, dir) + ps, err := setupPosix(dir) testutil.NilError(t, err) s3s, err := setupS3(t, dir) @@ -68,31 +62,23 @@ func TestList(t *testing.T) { expected []string } tests := []struct { - s map[string]Storage + s map[string]ObjStorage objects []string ops []listop }{ { - map[string]Storage{"posixflat": pfs}, + map[string]ObjStorage{}, []string{ // Minio (as of 20190201) IMHO is not real S3 since it tries to map to a // file system and not a flat namespace like S3. For this reason this test - // won't work with minio beacuse it creates a file called "path/of" and so + // won't work with minio because it creates a file called "path/of" and so // it's not possible to create a file "path/of/a" because it needs "of" to // be a directory + // The same for the posix storage since it uses a posix filesystem. - // All of the below tests will fail on Minio due to the above reason and also the multiple '/' - // so we aren't testing these with it + // All of the below tests will fail on Minio and posix due to the above + // reasons also due to the multiple '/' so we aren't testing these with them. - //"path/of", - //"path/of/a/file02", - //"path/of/a/file03", - //"path/of/a/file04", - //"path/of/a/file05", - - // These are multiple of 8 chars on purpose to test the filemarker behavior to - // distinguish between a file or a directory when the files ends at the path - // separator point "s3/is/not/a/file///system/file01", "s3/is/not/a/file///system/file02", "s3/is/not/a/file///system/file03", @@ -183,11 +169,8 @@ func TestList(t *testing.T) { }, }, { - map[string]Storage{"posix": ps, "posixflat": pfs, "minio": s3s}, + map[string]ObjStorage{"posix": ps, "minio": s3s}, []string{ - // These are multiple of 8 chars on purpose to test the filemarker behavior to - // distinguish between a file or a directory when the files ends at the path - // separator point "s3/is/not/a/file/sy/st/em/file01", "s3/is/not/a/file/sy/st/em/file02", "s3/is/not/a/file/sy/st/em/file03", @@ -260,27 +243,26 @@ func TestList(t *testing.T) { } for i, tt := range tests { - for sname, s := range tt.s { - t.Run(fmt.Sprintf("test with storage type %s", sname), func(t *testing.T) { - switch s := s.(type) { + for sname, os := range tt.s { + t.Run(fmt.Sprintf("with storage type %s", sname), func(t *testing.T) { + ctx := context.Background() + switch s := os.(type) { case *S3Storage: if s == nil { t.SkipNow() } } - os := NewObjStorage(s, "/") + // populate for _, p := range tt.objects { - if err := os.WriteObject(p, strings.NewReader(""), 0, true); err != nil { + if err := os.WriteObject(ctx, p, strings.NewReader(""), 0, true); err != nil { t.Fatalf("%s %d err: %v", sname, i, err) } } - doneCh := make(chan struct{}) - defer close(doneCh) for j, op := range tt.ops { paths := []string{} - for object := range os.List(op.prefix, op.start, op.recursive, doneCh) { + for object := range os.List(ctx, op.prefix, op.start, op.recursive) { if object.Err != nil { t.Fatalf("%s %d-%d err: %v", sname, i, j, object.Err) return @@ -299,10 +281,7 @@ func TestList(t *testing.T) { func TestWriteObject(t *testing.T) { dir := t.TempDir() - ps, err := setupPosix(t, dir) - testutil.NilError(t, err) - - pfs, err := setupPosixFlat(t, dir) + ps, err := setupPosix(dir) testutil.NilError(t, err) s3s, err := setupS3(t, dir) @@ -316,25 +295,26 @@ func TestWriteObject(t *testing.T) { return bytes.NewBuffer(testBytes) } - for sname, s := range map[string]Storage{"posix": ps, "posixflat": pfs, "minio": s3s} { - t.Run(fmt.Sprintf("test with storage type %s", sname), func(t *testing.T) { - switch s := s.(type) { + for sname, os := range map[string]ObjStorage{"posix": ps, "minio": s3s} { + t.Run(fmt.Sprintf("with storage type %s", sname), func(t *testing.T) { + ctx := context.Background() + + switch s := os.(type) { case *S3Storage: if s == nil { t.SkipNow() } } - os := NewObjStorage(s, "/") n := int64(10000) // Test write without size. Should write whole buffer. buf := newBuf(n) objName := "obj01" - err := os.WriteObject(objName, buf, -1, false) + err := os.WriteObject(ctx, objName, buf, -1, false) testutil.NilError(t, err) - oi, err := os.Stat(objName) + oi, err := os.Stat(ctx, objName) testutil.NilError(t, err) assert.Equal(t, oi.Size, n, "expected object size") @@ -342,10 +322,10 @@ func TestWriteObject(t *testing.T) { // Test write with object size equal to buf size. buf = newBuf(n) objName = "obj02" - err = os.WriteObject(objName, buf, n, false) + err = os.WriteObject(ctx, objName, buf, n, false) testutil.NilError(t, err) - oi, err = os.Stat(objName) + oi, err = os.Stat(ctx, objName) testutil.NilError(t, err) assert.Equal(t, oi.Size, n, "expected object size") @@ -354,10 +334,10 @@ func TestWriteObject(t *testing.T) { buf = newBuf(n) objName = "obj03" size := int64(800) - err = os.WriteObject(objName, buf, int64(size), false) + err = os.WriteObject(ctx, objName, buf, int64(size), false) testutil.NilError(t, err) - oi, err = os.Stat(objName) + oi, err = os.Stat(ctx, objName) testutil.NilError(t, err) assert.Equal(t, oi.Size, size, "expected object size") diff --git a/internal/objectstorage/posix.go b/internal/objectstorage/posix.go index f6da758e..75f00266 100644 --- a/internal/objectstorage/posix.go +++ b/internal/objectstorage/posix.go @@ -15,6 +15,7 @@ package objectstorage import ( + "context" "io" "os" "path" @@ -22,6 +23,8 @@ import ( "strings" "github.com/sorintlab/errors" + + "agola.io/agola/internal/util" ) const ( @@ -56,7 +59,7 @@ func (s *PosixStorage) fsPath(p string) (string, error) { return filepath.Join(s.dataDir, p), nil } -func (s *PosixStorage) Stat(p string) (*ObjectInfo, error) { +func (s *PosixStorage) Stat(ctx context.Context, p string) (*ObjectInfo, error) { fspath, err := s.fsPath(p) if err != nil { return nil, errors.WithStack(err) @@ -73,7 +76,7 @@ func (s *PosixStorage) Stat(p string) (*ObjectInfo, error) { return &ObjectInfo{Path: p, LastModified: fi.ModTime(), Size: fi.Size()}, nil } -func (s *PosixStorage) ReadObject(p string) (ReadSeekCloser, error) { +func (s *PosixStorage) ReadObject(ctx context.Context, p string) (ReadSeekCloser, error) { fspath, err := s.fsPath(p) if err != nil { return nil, errors.WithStack(err) @@ -86,7 +89,7 @@ func (s *PosixStorage) ReadObject(p string) (ReadSeekCloser, error) { return f, errors.WithStack(err) } -func (s *PosixStorage) WriteObject(p string, data io.Reader, size int64, persist bool) error { +func (s *PosixStorage) WriteObject(ctx context.Context, p string, data io.Reader, size int64, persist bool) error { fspath, err := s.fsPath(p) if err != nil { return errors.WithStack(err) @@ -106,7 +109,7 @@ func (s *PosixStorage) WriteObject(p string, data io.Reader, size int64, persist }) } -func (s *PosixStorage) DeleteObject(p string) error { +func (s *PosixStorage) DeleteObject(ctx context.Context, p string) error { fspath, err := s.fsPath(p) if err != nil { return errors.WithStack(err) @@ -148,21 +151,14 @@ func (s *PosixStorage) DeleteObject(p string) error { return nil } -func (s *PosixStorage) List(prefix, startWith, delimiter string, doneCh <-chan struct{}) <-chan ObjectInfo { +func (s *PosixStorage) List(ctx context.Context, prefix, startAfter string, recursive bool) <-chan ObjectInfo { objectCh := make(chan ObjectInfo, 1) - if len(delimiter) > 1 { - objectCh <- ObjectInfo{Err: errors.Errorf("wrong delimiter %q", delimiter)} + if startAfter != "" && !strings.Contains(startAfter, prefix) { + objectCh <- ObjectInfo{Err: errors.Errorf("wrong startAfter value %q for prefix %q", startAfter, prefix)} return objectCh } - if startWith != "" && !strings.Contains(startWith, prefix) { - objectCh <- ObjectInfo{Err: errors.Errorf("wrong startwith value %q for prefix %q", startWith, prefix)} - return objectCh - } - - recursive := delimiter == "" - // remove leading slash from prefix prefix = strings.TrimPrefix(prefix, "/") @@ -173,10 +169,18 @@ func (s *PosixStorage) List(prefix, startWith, delimiter string, doneCh <-chan s } // remove leading slash - startWith = strings.TrimPrefix(startWith, "/") + startAfter = strings.TrimPrefix(startAfter, "/") go func(objectCh chan<- ObjectInfo) { - defer close(objectCh) + defer func() { + if util.ContextCanceled(ctx) { + objectCh <- ObjectInfo{ + Err: ctx.Err(), + } + } + close(objectCh) + }() + err := filepath.Walk(root, func(ep string, info os.FileInfo, err error) error { if err != nil && !errors.Is(err, os.ErrNotExist) { return errors.WithStack(err) @@ -195,7 +199,7 @@ func (s *PosixStorage) List(prefix, startWith, delimiter string, doneCh <-chan s } if !recursive && len(p) > len(prefix) { rel := strings.TrimPrefix(p, prefix) - skip := strings.Contains(rel, delimiter) + skip := strings.Contains(rel, "/") if info.IsDir() && skip { return filepath.SkipDir @@ -209,21 +213,20 @@ func (s *PosixStorage) List(prefix, startWith, delimiter string, doneCh <-chan s return nil } - if strings.HasPrefix(p, prefix) && p > startWith { + if strings.HasPrefix(p, prefix) && p > startAfter { select { - // Send object content. case objectCh <- ObjectInfo{Path: p, LastModified: info.ModTime(), Size: info.Size()}: - // If receives done from the caller, return here. - case <-doneCh: - return io.EOF + case <-ctx.Done(): + return nil } } return nil }) - if err != nil && !errors.Is(err, io.EOF) { - objectCh <- ObjectInfo{ - Err: err, + if err != nil { + select { + case objectCh <- ObjectInfo{Err: err}: + case <-ctx.Done(): } return } diff --git a/internal/objectstorage/posix_test.go b/internal/objectstorage/posix_test.go index 2af9bdf3..6063c511 100644 --- a/internal/objectstorage/posix_test.go +++ b/internal/objectstorage/posix_test.go @@ -16,6 +16,7 @@ package objectstorage import ( "bytes" + "context" "os" "path/filepath" "testing" @@ -27,6 +28,7 @@ import ( ) func TestPosixDeleteObject(t *testing.T) { + ctx := context.Background() objects := []string{"☺☺☺☺a☺☺☺☺☺☺b☺☺☺☺", "s3/is/nota/fil.fa", "s3/is/not/a/file///system/fi%l%%e01"} dir := t.TempDir() @@ -35,10 +37,10 @@ func TestPosixDeleteObject(t *testing.T) { testutil.NilError(t, err) for _, obj := range objects { - err := ls.WriteObject(obj, bytes.NewReader([]byte{}), 0, true) + err := ls.WriteObject(ctx, obj, bytes.NewReader([]byte{}), 0, true) testutil.NilError(t, err) - err = ls.DeleteObject(obj) + err = ls.DeleteObject(ctx, obj) testutil.NilError(t, err) } diff --git a/internal/objectstorage/posixflat.go b/internal/objectstorage/posixflat.go deleted file mode 100644 index f08573bf..00000000 --- a/internal/objectstorage/posixflat.go +++ /dev/null @@ -1,430 +0,0 @@ -// Copyright 2019 Sorint.lab -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied -// See the License for the specific language governing permissions and -// limitations under the License. - -package objectstorage - -import ( - "io" - "os" - "path" - "path/filepath" - "strconv" - "strings" - "unicode/utf8" - - "github.com/sorintlab/errors" -) - -const ( - splitLength = 8 -) - -func shouldEscape(c rune) bool { - return c == '/' || c == '%' -} - -// escape does percent encoding to '/' and adds a slash every 8 (of the original -// string) chars -func escape(s string) string { - sepCount, hexCount := 0, 0 - nc := 0 - for _, c := range s { - nc++ - if shouldEscape(c) { - hexCount++ - } - if nc%splitLength == 0 { - sepCount++ - } - } - - if sepCount == 0 && hexCount == 0 { - return s - } - - hasFileMarker := nc%splitLength == 0 - l := len(s) + sepCount + 2*hexCount - // if the string length is a multiple of 8 then we have to add a file marker - // ".f" to not ovverride a possible directory in our fs representation - if hasFileMarker { - l++ - } - - t := make([]byte, l) - j := 0 - nc = 0 - for _, c := range s { - nc++ - switch { - case shouldEscape(c): - t[j] = '%' - t[j+1] = "0123456789ABCDEF"[c>>4] - t[j+2] = "0123456789ABCDEF"[c&15] - j += 3 - default: - s := string(c) - for i := 0; i < len(s); i++ { - t[j] = s[i] - j++ - } - } - if nc%splitLength == 0 { - t[j] = '/' - j++ - } - } - - // add file marker - if hasFileMarker { - t[j-1] = '.' - t[j] = 'f' - } - - return string(t) -} - -func ishex(c byte) bool { - switch { - case '0' <= c && c <= '9': - return true - case 'a' <= c && c <= 'f': - return true - case 'A' <= c && c <= 'F': - return true - } - return false -} - -func unhex(c byte) byte { - switch { - case '0' <= c && c <= '9': - return c - '0' - case 'a' <= c && c <= 'f': - return c - 'a' + 10 - case 'A' <= c && c <= 'F': - return c - 'A' + 10 - } - return 0 -} - -type EscapeError string - -func (e EscapeError) Error() string { - return "invalid URL escape " + strconv.Quote(string(e)) -} - -func unescape(s string) (string, bool, error) { - // number of percent encoded - n := 0 - // number of slashes - ns := 0 - // number of char in the unescaped string - nc := 0 - - for i := 0; i < len(s); { - r, width := utf8.DecodeRuneInString(s[i:]) - if r == utf8.RuneError { - return "", false, errors.Errorf("bad UTF-8 string") - } - switch r { - case '%': - n++ - if i+2 >= len(s) || !ishex(s[i+1]) || !ishex(s[i+2]) { - s = s[i:] - if len(s) > 3 { - s = s[:3] - } - return "", false, EscapeError(s) - } - i += 3 - nc++ - case '/': - ns++ - if nc%splitLength != 0 { - return "", false, EscapeError(s) - } - i++ - default: - i += width - nc++ - } - } - - // check and remove trailing file marker - hasFileMarker := false - if nc > splitLength && nc%splitLength == 2 && s[len(s)-2:] == ".f" { - hasFileMarker = true - s = s[:len(s)-2] - } - - if n == 0 && ns == 0 { - return s, hasFileMarker, nil - } - - // destination string is - // the length of the escaped one (with the ending file marker already removed) - number of percent * 2 - number os slashes - t := make([]byte, len(s)-n*2-ns) - j := 0 - for i := 0; i < len(s); { - r, width := utf8.DecodeRuneInString(s[i:]) - if r == utf8.RuneError { - return "", false, errors.Errorf("bad UTF-8 string") - } - switch r { - case '%': - t[j] = unhex(s[i+1])<<4 | unhex(s[i+2]) - j++ - i += 3 - case '/': - // skip "/" - i++ - default: - for k := 0; k < width; k++ { - t[j] = s[i] - j++ - i++ - } - } - } - return string(t), hasFileMarker, nil -} - -type PosixFlatStorage struct { - dataDir string - tmpDir string -} - -func NewPosixFlat(baseDir string) (*PosixFlatStorage, error) { - if err := os.MkdirAll(baseDir, 0770); err != nil { - return nil, errors.WithStack(err) - } - dataDir := filepath.Join(baseDir, dataDirName) - tmpDir := filepath.Join(baseDir, tmpDirName) - if err := os.MkdirAll(dataDir, 0770); err != nil { - return nil, errors.Wrapf(err, "failed to create data dir") - } - if err := os.MkdirAll(tmpDir, 0770); err != nil { - return nil, errors.Wrapf(err, "failed to create tmp dir") - } - return &PosixFlatStorage{ - dataDir: dataDir, - tmpDir: tmpDir, - }, nil -} - -func (s *PosixFlatStorage) fsPath(p string) (string, error) { - if p == "" { - return "", errors.Errorf("empty key name") - } - return filepath.Join(s.dataDir, escape(p)), nil -} - -func (s *PosixFlatStorage) Stat(p string) (*ObjectInfo, error) { - fspath, err := s.fsPath(p) - if err != nil { - return nil, errors.WithStack(err) - } - - fi, err := os.Stat(fspath) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - return nil, NewErrNotExist(err, "object %q doesn't exist", p) - } - return nil, errors.WithStack(err) - } - - return &ObjectInfo{Path: p, LastModified: fi.ModTime(), Size: fi.Size()}, nil -} - -func (s *PosixFlatStorage) ReadObject(p string) (ReadSeekCloser, error) { - fspath, err := s.fsPath(p) - if err != nil { - return nil, errors.WithStack(err) - } - - f, err := os.Open(fspath) - if err != nil && errors.Is(err, os.ErrNotExist) { - return nil, NewErrNotExist(err, "object %q doesn't exist", p) - } - return f, errors.WithStack(err) -} - -func (s *PosixFlatStorage) WriteObject(p string, data io.Reader, size int64, persist bool) error { - fspath, err := s.fsPath(p) - if err != nil { - return errors.WithStack(err) - } - - if err := os.MkdirAll(path.Dir(fspath), 0770); err != nil { - return errors.WithStack(err) - } - - r := data - if size >= 0 { - r = io.LimitReader(data, size) - } - return writeFileAtomicFunc(fspath, s.dataDir, s.tmpDir, 0660, persist, func(f io.Writer) error { - _, err := io.Copy(f, r) - return errors.WithStack(err) - }) -} - -func (s *PosixFlatStorage) DeleteObject(p string) error { - fspath, err := s.fsPath(p) - if err != nil { - return errors.WithStack(err) - } - - if err := os.Remove(fspath); err != nil { - if errors.Is(err, os.ErrNotExist) { - return NewErrNotExist(err, "object %q doesn't exist", p) - } - return errors.WithStack(err) - } - - // try to remove parent empty dirs - // TODO(sgotti) if this fails we ignore errors and the dirs will be left as - // empty, clean them asynchronously - pdir := filepath.Dir(fspath) - for { - if pdir == s.dataDir || !strings.HasPrefix(pdir, s.dataDir) { - break - } - f, err := os.Open(pdir) - if err != nil { - return nil - } - - _, err = f.Readdirnames(1) - if errors.Is(err, io.EOF) { - f.Close() - if err := os.Remove(pdir); err != nil { - return nil - } - } else { - f.Close() - break - } - - pdir = filepath.Dir(pdir) - } - return nil -} - -func (s *PosixFlatStorage) List(prefix, startWith, delimiter string, doneCh <-chan struct{}) <-chan ObjectInfo { - objectCh := make(chan ObjectInfo, 1) - - if len(delimiter) > 1 { - objectCh <- ObjectInfo{Err: errors.Errorf("wrong delimiter %q", delimiter)} - return objectCh - } - - if startWith != "" && !strings.Contains(startWith, prefix) { - objectCh <- ObjectInfo{Err: errors.Errorf("wrong startwith value %q for prefix %q", startWith, prefix)} - return objectCh - } - - recursive := delimiter == "" - - // remove leading slash from prefix - prefix = strings.TrimPrefix(prefix, "/") - - fprefix := filepath.Join(s.dataDir, escape(prefix)) - root := filepath.Dir(fprefix) - if len(root) < len(s.dataDir) { - root = s.dataDir - } - - // remove leading slash - startWith = strings.TrimPrefix(startWith, "/") - - go func(objectCh chan<- ObjectInfo) { - var prevp string - defer close(objectCh) - err := filepath.Walk(root, func(ep string, info os.FileInfo, err error) error { - if err != nil && !errors.Is(err, os.ErrNotExist) { - return errors.WithStack(err) - } - if errors.Is(err, os.ErrNotExist) { - return nil - } - p := ep - - // get the path with / separator - p = filepath.ToSlash(p) - - p, err = filepath.Rel(s.dataDir, p) - if err != nil { - return errors.WithStack(err) - } - p, _, err = unescape(p) - if err != nil { - return errors.WithStack(err) - } - if !recursive && len(p) > len(prefix) { - rel := strings.TrimPrefix(p, prefix) - skip := strings.Contains(rel, delimiter) - - if info.IsDir() && skip { - return filepath.SkipDir - } - if skip { - return nil - } - } - - // don't list dirs if there's not a file with the same name (with filemarker) - // it's not an issue if the file in the meantime has been removed, it won't - // just be listed - hasFile := true - _, err = os.Stat(ep + ".f") - if err != nil && !errors.Is(err, os.ErrNotExist) { - return errors.WithStack(err) - } - if errors.Is(err, os.ErrNotExist) { - hasFile = false - } - if info.IsDir() && !hasFile { - return nil - } - - if strings.HasPrefix(p, prefix) && p > startWith { - // skip keys smaller than the previously returned one. This happens when we - // receive a file with a file marker that we already returned previously - // when we received a dir with the same name - // it'not an issue if the dir has been removed since we already returned the file - if p > prevp { - select { - // Send object content. - case objectCh <- ObjectInfo{Path: p, LastModified: info.ModTime(), Size: info.Size()}: - // If receives done from the caller, return here. - case <-doneCh: - return io.EOF - } - } - prevp = p - } - - return nil - }) - if err != nil && !errors.Is(err, io.EOF) { - objectCh <- ObjectInfo{ - Err: err, - } - return - } - }(objectCh) - - return objectCh -} diff --git a/internal/objectstorage/posixflat_test.go b/internal/objectstorage/posixflat_test.go deleted file mode 100644 index 2dfd47ac..00000000 --- a/internal/objectstorage/posixflat_test.go +++ /dev/null @@ -1,108 +0,0 @@ -// Copyright 2019 Sorint.lab -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied -// See the License for the specific language governing permissions and -// limitations under the License. - -package objectstorage - -import ( - "bytes" - "os" - "path/filepath" - "testing" - - "github.com/sorintlab/errors" - "gotest.tools/assert" - - "agola.io/agola/internal/testutil" -) - -func TestEscapeUnescape(t *testing.T) { - tests := []struct { - in string - expected string - err error - }{ - {"", "", nil}, - {"/", "%2F", nil}, - {"//", "%2F%2F", nil}, - {"☺", "☺", nil}, - {"☺☺☺☺☺☺☺☺", "☺☺☺☺☺☺☺☺.f", nil}, - {"☺☺☺☺☺☺☺☺", "☺☺☺☺☺☺☺☺.f", nil}, - {"☺☺☺☺☺☺☺☺☺☺☺☺☺☺☺☺", "☺☺☺☺☺☺☺☺/☺☺☺☺☺☺☺☺.f", nil}, - {"☺☺☺☺a☺☺☺☺☺☺☺☺☺☺☺", "☺☺☺☺a☺☺☺/☺☺☺☺☺☺☺☺.f", nil}, - {"☺☺☺☺a☺☺☺☺☺☺b☺☺☺☺", "☺☺☺☺a☺☺☺/☺☺☺b☺☺☺☺.f", nil}, - {"⌘", "⌘", nil}, - {"⌘⌘⌘⌘⌘⌘⌘⌘⌘⌘⌘", "⌘⌘⌘⌘⌘⌘⌘⌘/⌘⌘⌘", nil}, - - // These are 16 chars on purpose to test the filemarker behavior - {"s3/is/not/a/file", "s3%2Fis%2Fno/t%2Fa%2Ffile.f", nil}, - {"s3/is/nota/file/", "s3%2Fis%2Fno/ta%2Ffile%2F.f", nil}, - {"s3/is/nota/files", "s3%2Fis%2Fno/ta%2Ffiles.f", nil}, - {"s3/is/nota/fil.f", "s3%2Fis%2Fno/ta%2Ffil.f.f", nil}, - - {"s3/is/nota/fil.fa", "s3%2Fis%2Fno/ta%2Ffil.f/a", nil}, - {"/s3/is/nota/fil.fa/", "%2Fs3%2Fis%2Fn/ota%2Ffil./fa%2F", nil}, - {"s3/is/not/a/file///system/fi%l%%e01", "s3%2Fis%2Fno/t%2Fa%2Ffile/%2F%2F%2Fsyste/m%2Ffi%25l%25%25/e01", nil}, - {"s3/is/not/a/file///system/file01", "s3%2Fis%2Fno/t%2Fa%2Ffile/%2F%2F%2Fsyste/m%2Ffile01.f", nil}, - {"s3/is/not/a/file///system/file01/", "s3%2Fis%2Fno/t%2Fa%2Ffile/%2F%2F%2Fsyste/m%2Ffile01/%2F", nil}, - {"s3/is/not/a/file///system/file01/a", "s3%2Fis%2Fno/t%2Fa%2Ffile/%2F%2F%2Fsyste/m%2Ffile01/%2Fa", nil}, - } - - for i, tt := range tests { - out := escape(tt.in) - if out != tt.expected { - t.Fatalf("%d: escape: expected %q got %q", i, tt.expected, out) - } - - unescaped, _, err := unescape(out) - if err != nil { - if tt.err == nil { - t.Fatalf("%d: unescape: expected no error got %v", i, err) - } else if !errors.Is(tt.err, err) { - t.Fatalf("%d: unescape: expected error %v got %v", i, tt.err, err) - } - } else { - if tt.err != nil { - t.Fatalf("%d: unescape: expected error %v got no error", i, tt.err) - } else if unescaped != tt.in { - t.Fatalf("%d: unescape: expected %q got %q", i, tt.in, unescaped) - } - } - } -} - -func TestPosixFlatDeleteObject(t *testing.T) { - objects := []string{"/", "//", "☺☺☺☺a☺☺☺☺☺☺b☺☺☺☺", "s3/is/nota/fil.fa", "s3/is/not/a/file///system/fi%l%%e01"} - - dir := t.TempDir() - - ls, err := NewPosixFlat(dir) - testutil.NilError(t, err) - - for _, obj := range objects { - err := ls.WriteObject(obj, bytes.NewReader([]byte{}), 0, true) - testutil.NilError(t, err) - - err = ls.DeleteObject(obj) - testutil.NilError(t, err) - } - - // no files and directories should be left - bd, err := os.Open(filepath.Join(dir, dataDirName)) - testutil.NilError(t, err) - - files, err := bd.Readdirnames(0) - testutil.NilError(t, err) - - assert.Equal(t, len(files), 0) -} diff --git a/internal/objectstorage/s3.go b/internal/objectstorage/s3.go index 53c0b5c1..ded6beb8 100644 --- a/internal/objectstorage/s3.go +++ b/internal/objectstorage/s3.go @@ -15,13 +15,17 @@ package objectstorage import ( + "context" "io" "net/http" "os" "strings" - minio "github.com/minio/minio-go/v6" + minio "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" "github.com/sorintlab/errors" + + "agola.io/agola/internal/util" ) type S3Storage struct { @@ -31,23 +35,23 @@ type S3Storage struct { minioCore *minio.Core } -func NewS3(bucket, location, endpoint, accessKeyID, secretAccessKey string, secure bool) (*S3Storage, error) { - minioClient, err := minio.New(endpoint, accessKeyID, secretAccessKey, secure) +func NewS3(ctx context.Context, bucket, location, endpoint, accessKeyID, secretAccessKey string, secure bool) (*S3Storage, error) { + minioClient, err := minio.New(endpoint, &minio.Options{Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""), Secure: secure}) if err != nil { return nil, errors.WithStack(err) } - minioCore, err := minio.NewCore(endpoint, accessKeyID, secretAccessKey, secure) + minioCore, err := minio.NewCore(endpoint, &minio.Options{Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""), Secure: secure}) if err != nil { return nil, errors.WithStack(err) } - exists, err := minioClient.BucketExists(bucket) + exists, err := minioClient.BucketExists(ctx, bucket) if err != nil { return nil, errors.Wrapf(err, "cannot check if bucket %q in location %q exits", bucket, location) } if !exists { - if err := minioClient.MakeBucket(bucket, location); err != nil { + if err := minioClient.MakeBucket(ctx, bucket, minio.MakeBucketOptions{Region: location}); err != nil { return nil, errors.Wrapf(err, "cannot create bucket %q in location %q", bucket, location) } } @@ -59,8 +63,8 @@ func NewS3(bucket, location, endpoint, accessKeyID, secretAccessKey string, secu }, nil } -func (s *S3Storage) Stat(p string) (*ObjectInfo, error) { - oi, err := s.minioClient.StatObject(s.bucket, p, minio.StatObjectOptions{}) +func (s *S3Storage) Stat(ctx context.Context, p string) (*ObjectInfo, error) { + oi, err := s.minioClient.StatObject(ctx, s.bucket, p, minio.StatObjectOptions{}) if err != nil { merr := minio.ToErrorResponse(err) if merr.StatusCode == http.StatusNotFound { @@ -72,8 +76,8 @@ func (s *S3Storage) Stat(p string) (*ObjectInfo, error) { return &ObjectInfo{Path: p, LastModified: oi.LastModified, Size: oi.Size}, nil } -func (s *S3Storage) ReadObject(filepath string) (ReadSeekCloser, error) { - if _, err := s.minioClient.StatObject(s.bucket, filepath, minio.StatObjectOptions{}); err != nil { +func (s *S3Storage) ReadObject(ctx context.Context, filepath string) (ReadSeekCloser, error) { + if _, err := s.minioClient.StatObject(ctx, s.bucket, filepath, minio.StatObjectOptions{}); err != nil { merr := minio.ToErrorResponse(err) if merr.StatusCode == http.StatusNotFound { return nil, NewErrNotExist(err, "object %q doesn't exist", filepath) @@ -81,19 +85,19 @@ func (s *S3Storage) ReadObject(filepath string) (ReadSeekCloser, error) { return nil, errors.WithStack(merr) } - o, err := s.minioClient.GetObject(s.bucket, filepath, minio.GetObjectOptions{}) + o, err := s.minioClient.GetObject(ctx, s.bucket, filepath, minio.GetObjectOptions{}) return o, errors.WithStack(err) } -func (s *S3Storage) WriteObject(filepath string, data io.Reader, size int64, persist bool) error { +func (s *S3Storage) WriteObject(ctx context.Context, filepath string, data io.Reader, size int64, persist bool) error { // if size is not specified, limit max object size to defaultMaxObjectSize so // minio client will not calculate a very big part size using tons of ram. // An alternative is to write the file locally so we can calculate the size and // then put it. See commented out code below. if size >= 0 { lr := io.LimitReader(data, size) - _, err := s.minioClient.PutObject(s.bucket, filepath, lr, size, minio.PutObjectOptions{ContentType: "application/octet-stream"}) + _, err := s.minioClient.PutObject(ctx, s.bucket, filepath, lr, size, minio.PutObjectOptions{ContentType: "application/octet-stream"}) return errors.WithStack(err) } @@ -112,61 +116,48 @@ func (s *S3Storage) WriteObject(filepath string, data io.Reader, size int64, per if _, err := tmpfile.Seek(0, 0); err != nil { return errors.WithStack(err) } - _, err = s.minioClient.PutObject(s.bucket, filepath, tmpfile, size, minio.PutObjectOptions{ContentType: "application/octet-stream"}) + _, err = s.minioClient.PutObject(ctx, s.bucket, filepath, tmpfile, size, minio.PutObjectOptions{ContentType: "application/octet-stream"}) return errors.WithStack(err) } -func (s *S3Storage) DeleteObject(filepath string) error { - return errors.WithStack(s.minioClient.RemoveObject(s.bucket, filepath)) +func (s *S3Storage) DeleteObject(ctx context.Context, filepath string) error { + return errors.WithStack(s.minioClient.RemoveObject(ctx, s.bucket, filepath, minio.RemoveObjectOptions{})) } -func (s *S3Storage) List(prefix, startWith, delimiter string, doneCh <-chan struct{}) <-chan ObjectInfo { +func (s *S3Storage) List(ctx context.Context, prefix, startAfter string, recursive bool) <-chan ObjectInfo { objectCh := make(chan ObjectInfo, 1) - if len(delimiter) > 1 { - objectCh <- ObjectInfo{ - Err: errors.Errorf("wrong delimiter %q", delimiter), - } - return objectCh - } - // remove leading slash prefix = strings.TrimPrefix(prefix, "/") - startWith = strings.TrimPrefix(startWith, "/") + startAfter = strings.TrimPrefix(startAfter, "/") - // Initiate list objects goroutine here. go func(objectCh chan<- ObjectInfo) { - defer close(objectCh) - // Save continuationToken for next request. - var continuationToken string - for { - // Get list of objects a maximum of 1000 per request. - result, err := s.minioCore.ListObjectsV2(s.bucket, prefix, continuationToken, false, delimiter, 1000, startWith) - if err != nil { + defer func() { + if util.ContextCanceled(ctx) { objectCh <- ObjectInfo{ - Err: err, + Err: ctx.Err(), + } + } + close(objectCh) + }() + + for object := range s.minioClient.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: recursive, StartAfter: startAfter}) { + if object.Err != nil { + select { + case objectCh <- ObjectInfo{Err: object.Err}: + case <-ctx.Done(): } return } - // If contents are available loop through and send over channel. - for _, object := range result.Contents { - select { - // Send object content. - case objectCh <- ObjectInfo{Path: object.Key, LastModified: object.LastModified, Size: object.Size}: - // If receives done from the caller, return here. - case <-doneCh: - return - } + // minioClient.ListObject also returns common prefixes as objects, but we only want objects + if strings.HasSuffix(object.Key, "/") { + continue } - // If continuation token present, save it for next request. - if result.NextContinuationToken != "" { - continuationToken = result.NextContinuationToken - } - - // Listing ends result is not truncated, return right here. - if !result.IsTruncated { + select { + case objectCh <- ObjectInfo{Path: object.Key, LastModified: object.LastModified, Size: object.Size}: + case <-ctx.Done(): return } } diff --git a/internal/services/configstore/configstore.go b/internal/services/configstore/configstore.go index cdf70954..75ae24a6 100644 --- a/internal/services/configstore/configstore.go +++ b/internal/services/configstore/configstore.go @@ -77,7 +77,7 @@ func (s *Configstore) maintenanceModeWatcher(ctx context.Context, runCtxCancel c type Configstore struct { log zerolog.Logger c *config.Configstore - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage d *db.DB lf lock.LockFactory ah *action.ActionHandler @@ -89,7 +89,7 @@ func NewConfigstore(ctx context.Context, log zerolog.Logger, c *config.Configsto log = log.Level(zerolog.DebugLevel) } - ost, err := scommon.NewObjectStorage(&c.ObjectStorage) + ost, err := scommon.NewObjectStorage(ctx, &c.ObjectStorage) if err != nil { return nil, errors.WithStack(err) } diff --git a/internal/services/gateway/gateway.go b/internal/services/gateway/gateway.go index 9325b6b6..c3ce2d8d 100644 --- a/internal/services/gateway/gateway.go +++ b/internal/services/gateway/gateway.go @@ -51,7 +51,7 @@ type Gateway struct { log zerolog.Logger c *config.Gateway - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage runserviceClient *rsclient.Client configstoreClient *csclient.Client notificationClient *nsclient.Client @@ -124,7 +124,7 @@ func NewGateway(ctx context.Context, log zerolog.Logger, gc *config.Config) (*Ga Key: c.CookieSigning.Key, }) - ost, err := icommon.NewObjectStorage(&c.ObjectStorage) + ost, err := icommon.NewObjectStorage(ctx, &c.ObjectStorage) if err != nil { return nil, errors.WithStack(err) } diff --git a/internal/services/runservice/action/action.go b/internal/services/runservice/action/action.go index 42295c36..6eb64fe3 100644 --- a/internal/services/runservice/action/action.go +++ b/internal/services/runservice/action/action.go @@ -39,13 +39,13 @@ import ( type ActionHandler struct { log zerolog.Logger d *db.DB - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage lf lock.LockFactory maintenanceMode bool maintenanceModeMutex sync.Mutex } -func NewActionHandler(log zerolog.Logger, d *db.DB, ost *objectstorage.ObjStorage, lf lock.LockFactory) *ActionHandler { +func NewActionHandler(log zerolog.Logger, d *db.DB, ost objectstorage.ObjStorage, lf lock.LockFactory) *ActionHandler { return &ActionHandler{ log: log, d: d, diff --git a/internal/services/runservice/api/api.go b/internal/services/runservice/api/api.go index 701fe090..2c3471d8 100644 --- a/internal/services/runservice/api/api.go +++ b/internal/services/runservice/api/api.go @@ -87,10 +87,10 @@ func addHasMoreHeader(w http.ResponseWriter, hasMore bool) { type LogsHandler struct { log zerolog.Logger d *db.DB - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage } -func NewLogsHandler(log zerolog.Logger, d *db.DB, ost *objectstorage.ObjStorage) *LogsHandler { +func NewLogsHandler(log zerolog.Logger, d *db.DB, ost objectstorage.ObjStorage) *LogsHandler { return &LogsHandler{ log: log, d: d, @@ -192,7 +192,7 @@ func (h *LogsHandler) readTaskLogs(ctx context.Context, runID, taskID string, se } else { logPath = store.OSTRunTaskStepLogPath(task.ID, step) } - f, err := h.ost.ReadObject(logPath) + f, err := h.ost.ReadObject(ctx, logPath) if err != nil { if objectstorage.IsNotExist(err) { return true, util.NewAPIErrorWrap(util.ErrNotExist, err) @@ -303,10 +303,10 @@ func sendLogs(w http.ResponseWriter, r io.Reader) error { type LogsDeleteHandler struct { log zerolog.Logger d *db.DB - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage } -func NewLogsDeleteHandler(log zerolog.Logger, d *db.DB, ost *objectstorage.ObjStorage) *LogsDeleteHandler { +func NewLogsDeleteHandler(log zerolog.Logger, d *db.DB, ost objectstorage.ObjStorage) *LogsDeleteHandler { return &LogsDeleteHandler{ log: log, d: d, @@ -404,7 +404,7 @@ func (h *LogsDeleteHandler) deleteTaskLogs(ctx context.Context, runID, taskID st } else { logPath = store.OSTRunTaskStepLogPath(task.ID, step) } - err := h.ost.DeleteObject(logPath) + err := h.ost.DeleteObject(ctx, logPath) if err != nil { if objectstorage.IsNotExist(err) { return util.NewAPIErrorWrap(util.ErrNotExist, err) @@ -1005,10 +1005,10 @@ func (h *RunTaskActionsHandler) do(r *http.Request) error { type RunEventsHandler struct { log zerolog.Logger d *db.DB - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage } -func NewRunEventsHandler(log zerolog.Logger, d *db.DB, ost *objectstorage.ObjStorage) *RunEventsHandler { +func NewRunEventsHandler(log zerolog.Logger, d *db.DB, ost objectstorage.ObjStorage) *RunEventsHandler { return &RunEventsHandler{ log: log, d: d, diff --git a/internal/services/runservice/api/executor.go b/internal/services/runservice/api/executor.go index 1d30b8b6..7386c8b3 100644 --- a/internal/services/runservice/api/executor.go +++ b/internal/services/runservice/api/executor.go @@ -360,10 +360,10 @@ func (h *ExecutorTasksHandler) do(r *http.Request) ([]*rsapitypes.ExecutorTask, type ArchivesHandler struct { log zerolog.Logger - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage } -func NewArchivesHandler(log zerolog.Logger, ost *objectstorage.ObjStorage) *ArchivesHandler { +func NewArchivesHandler(log zerolog.Logger, ost objectstorage.ObjStorage) *ArchivesHandler { return &ArchivesHandler{ log: log, ost: ost, @@ -380,6 +380,7 @@ func (h *ArchivesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (h *ArchivesHandler) do(w http.ResponseWriter, r *http.Request) error { // TODO(sgotti) Check authorized call from executors + ctx := r.Context() taskID := r.URL.Query().Get("taskid") if taskID == "" { @@ -396,7 +397,7 @@ func (h *ArchivesHandler) do(w http.ResponseWriter, r *http.Request) error { w.Header().Set("Cache-Control", "no-cache") - if err := h.readArchive(taskID, step, w); err != nil { + if err := h.readArchive(ctx, taskID, step, w); err != nil { switch { case util.APIErrorIs(err, util.ErrNotExist): return util.NewAPIErrorWrap(util.ErrNotExist, err) @@ -408,9 +409,9 @@ func (h *ArchivesHandler) do(w http.ResponseWriter, r *http.Request) error { return nil } -func (h *ArchivesHandler) readArchive(rtID string, step int, w io.Writer) error { +func (h *ArchivesHandler) readArchive(ctx context.Context, rtID string, step int, w io.Writer) error { archivePath := store.OSTRunTaskArchivePath(rtID, step) - f, err := h.ost.ReadObject(archivePath) + f, err := h.ost.ReadObject(ctx, archivePath) if err != nil { if objectstorage.IsNotExist(err) { return util.NewAPIErrorWrap(util.ErrNotExist, err) @@ -427,10 +428,10 @@ func (h *ArchivesHandler) readArchive(rtID string, step int, w io.Writer) error type CacheHandler struct { log zerolog.Logger - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage } -func NewCacheHandler(log zerolog.Logger, ost *objectstorage.ObjStorage) *CacheHandler { +func NewCacheHandler(log zerolog.Logger, ost objectstorage.ObjStorage) *CacheHandler { return &CacheHandler{ log: log, ost: ost, @@ -446,6 +447,7 @@ func (h *CacheHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } func (h *CacheHandler) do(w http.ResponseWriter, r *http.Request) error { + ctx := r.Context() vars := mux.Vars(r) // TODO(sgotti) Check authorized call from executors @@ -460,7 +462,7 @@ func (h *CacheHandler) do(w http.ResponseWriter, r *http.Request) error { query := r.URL.Query() _, prefix := query["prefix"] - matchedKey, err := matchCache(h.ost, key, prefix) + matchedKey, err := matchCache(ctx, h.ost, key, prefix) if err != nil { return errors.WithStack(err) } @@ -474,7 +476,7 @@ func (h *CacheHandler) do(w http.ResponseWriter, r *http.Request) error { w.Header().Set("Cache-Control", "no-cache") - if err := h.readCache(matchedKey, w); err != nil { + if err := h.readCache(ctx, matchedKey, w); err != nil { switch { case util.APIErrorIs(err, util.ErrNotExist): return util.NewAPIErrorWrap(util.ErrNotExist, err) @@ -486,16 +488,13 @@ func (h *CacheHandler) do(w http.ResponseWriter, r *http.Request) error { return nil } -func matchCache(ost *objectstorage.ObjStorage, key string, prefix bool) (string, error) { +func matchCache(ctx context.Context, ost objectstorage.ObjStorage, key string, prefix bool) (string, error) { cachePath := store.OSTCachePath(key) if prefix { - doneCh := make(chan struct{}) - defer close(doneCh) - // get the latest modified object var lastObject *objectstorage.ObjectInfo - for object := range ost.List(store.OSTCacheDir()+"/"+key, "", false, doneCh) { + for object := range ost.List(ctx, store.OSTCacheDir()+"/"+key, "", false) { if object.Err != nil { return "", errors.WithStack(object.Err) } @@ -512,7 +511,7 @@ func matchCache(ost *objectstorage.ObjStorage, key string, prefix bool) (string, return store.OSTCacheKey(lastObject.Path), nil } - _, err := ost.Stat(cachePath) + _, err := ost.Stat(ctx, cachePath) if objectstorage.IsNotExist(err) { return "", nil } @@ -522,9 +521,9 @@ func matchCache(ost *objectstorage.ObjStorage, key string, prefix bool) (string, return key, nil } -func (h *CacheHandler) readCache(key string, w io.Writer) error { +func (h *CacheHandler) readCache(ctx context.Context, key string, w io.Writer) error { cachePath := store.OSTCachePath(key) - f, err := h.ost.ReadObject(cachePath) + f, err := h.ost.ReadObject(ctx, cachePath) if err != nil { if objectstorage.IsNotExist(err) { return util.NewAPIErrorWrap(util.ErrNotExist, err) @@ -541,10 +540,10 @@ func (h *CacheHandler) readCache(key string, w io.Writer) error { type CacheCreateHandler struct { log zerolog.Logger - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage } -func NewCacheCreateHandler(log zerolog.Logger, ost *objectstorage.ObjStorage) *CacheCreateHandler { +func NewCacheCreateHandler(log zerolog.Logger, ost objectstorage.ObjStorage) *CacheCreateHandler { return &CacheCreateHandler{ log: log, ost: ost, @@ -560,6 +559,7 @@ func (h *CacheCreateHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } func (h *CacheCreateHandler) do(w http.ResponseWriter, r *http.Request) error { + ctx := r.Context() vars := mux.Vars(r) // TODO(sgotti) Check authorized call from executors @@ -574,7 +574,7 @@ func (h *CacheCreateHandler) do(w http.ResponseWriter, r *http.Request) error { w.Header().Set("Cache-Control", "no-cache") - matchedKey, err := matchCache(h.ost, key, false) + matchedKey, err := matchCache(ctx, h.ost, key, false) if err != nil { return errors.WithStack(err) } @@ -593,7 +593,7 @@ func (h *CacheCreateHandler) do(w http.ResponseWriter, r *http.Request) error { } cachePath := store.OSTCachePath(key) - if err := h.ost.WriteObject(cachePath, r.Body, size, false); err != nil { + if err := h.ost.WriteObject(ctx, cachePath, r.Body, size, false); err != nil { return errors.WithStack(err) } diff --git a/internal/services/runservice/runservice.go b/internal/services/runservice/runservice.go index dd4f6d5f..f71b2e0f 100644 --- a/internal/services/runservice/runservice.go +++ b/internal/services/runservice/runservice.go @@ -76,7 +76,7 @@ func (s *Runservice) maintenanceModeWatcher(ctx context.Context, runCtxCancel co type Runservice struct { log zerolog.Logger c *config.Runservice - ost *objectstorage.ObjStorage + ost objectstorage.ObjStorage d *db.DB lf lock.LockFactory ah *action.ActionHandler @@ -88,7 +88,7 @@ func NewRunservice(ctx context.Context, log zerolog.Logger, c *config.Runservice log = log.Level(zerolog.DebugLevel) } - ost, err := scommon.NewObjectStorage(&c.ObjectStorage) + ost, err := scommon.NewObjectStorage(ctx, &c.ObjectStorage) if err != nil { return nil, errors.WithStack(err) } diff --git a/internal/services/runservice/runservice_test.go b/internal/services/runservice/runservice_test.go index 6d87025d..2c176bd6 100644 --- a/internal/services/runservice/runservice_test.go +++ b/internal/services/runservice/runservice_test.go @@ -297,10 +297,10 @@ func TestLogleaner(t *testing.T) { body := io.NopCloser(bytes.NewBufferString("log test")) logPath := store.OSTRunTaskStepLogPath("task01", 0) - err := rs.ost.WriteObject(logPath, body, -1, false) + err := rs.ost.WriteObject(ctx, logPath, body, -1, false) testutil.NilError(t, err) - _, err = rs.ost.ReadObject(logPath) + _, err = rs.ost.ReadObject(ctx, logPath) testutil.NilError(t, err) time.Sleep(1 * time.Second) @@ -308,7 +308,7 @@ func TestLogleaner(t *testing.T) { err = rs.objectsCleaner(ctx, store.OSTLogsBaseDir(), common.LogCleanerLockKey, 1*time.Millisecond) testutil.NilError(t, err) - _, err = rs.ost.ReadObject(logPath) + _, err = rs.ost.ReadObject(ctx, logPath) assert.ErrorType(t, err, objectstorage.IsNotExist) } diff --git a/internal/services/runservice/scheduler.go b/internal/services/runservice/scheduler.go index 07969336..f30bd89c 100644 --- a/internal/services/runservice/scheduler.go +++ b/internal/services/runservice/scheduler.go @@ -1014,8 +1014,8 @@ func (s *Runservice) runTasksUpdater(ctx context.Context) error { return nil } -func (s *Runservice) OSTFileExists(path string) (bool, error) { - _, err := s.ost.Stat(path) +func (s *Runservice) OSTFileExists(ctx context.Context, path string) (bool, error) { + _, err := s.ost.Stat(ctx, path) if err != nil && !objectstorage.IsNotExist(err) { return false, errors.WithStack(err) } @@ -1063,7 +1063,7 @@ func (s *Runservice) fetchLog(ctx context.Context, runID string, rt *types.RunTa } else { logPath = store.OSTRunTaskStepLogPath(rt.ID, stepnum) } - ok, err := s.OSTFileExists(logPath) + ok, err := s.OSTFileExists(ctx, logPath) if err != nil { return errors.WithStack(err) } @@ -1108,7 +1108,7 @@ func (s *Runservice) fetchLog(ctx context.Context, runID string, rt *types.RunTa } } - return errors.WithStack(s.ost.WriteObject(logPath, resp.Body, size, false)) + return errors.WithStack(s.ost.WriteObject(ctx, logPath, resp.Body, size, false)) } func (s *Runservice) finishSetupLogPhase(ctx context.Context, runID, runTaskID string) error { @@ -1277,7 +1277,7 @@ func (s *Runservice) fetchArchive(ctx context.Context, runID string, rt *types.R } path := store.OSTRunTaskArchivePath(rt.ID, stepnum) - ok, err := s.OSTFileExists(path) + ok, err := s.OSTFileExists(ctx, path) if err != nil { return errors.WithStack(err) } @@ -1319,7 +1319,7 @@ func (s *Runservice) fetchArchive(ctx context.Context, runID string, rt *types.R } } - return errors.WithStack(s.ost.WriteObject(path, resp.Body, size, false)) + return errors.WithStack(s.ost.WriteObject(ctx, path, resp.Body, size, false)) } func (s *Runservice) fetchTaskArchives(ctx context.Context, runID string, rt *types.RunTask) { @@ -1399,22 +1399,22 @@ func (s *Runservice) taskFetcher(ctx context.Context, r *types.Run, rt *types.Ru // write related logs runID runIDPath := store.OSTRunTaskLogsRunPath(rt.ID, r.ID) - exists, err := s.OSTFileExists(runIDPath) + exists, err := s.OSTFileExists(ctx, runIDPath) if err != nil { s.log.Err(err).Send() } else if !exists { - if err := s.ost.WriteObject(runIDPath, bytes.NewReader([]byte{}), 0, false); err != nil { + if err := s.ost.WriteObject(ctx, runIDPath, bytes.NewReader([]byte{}), 0, false); err != nil { s.log.Err(err).Send() } } // write related archives runID runIDPath = store.OSTRunTaskArchivesRunPath(rt.ID, r.ID) - exists, err = s.OSTFileExists(runIDPath) + exists, err = s.OSTFileExists(ctx, runIDPath) if err != nil { s.log.Err(err).Send() } else if !exists { - if err := s.ost.WriteObject(runIDPath, bytes.NewReader([]byte{}), 0, false); err != nil { + if err := s.ost.WriteObject(ctx, runIDPath, bytes.NewReader([]byte{}), 0, false); err != nil { s.log.Err(err).Send() } } @@ -1604,14 +1604,12 @@ func (s *Runservice) cacheCleaner(ctx context.Context, cacheExpireInterval time. } defer func() { _ = l.Unlock() }() - doneCh := make(chan struct{}) - defer close(doneCh) - for object := range s.ost.List(store.OSTCacheDir()+"/", "", true, doneCh) { + for object := range s.ost.List(ctx, store.OSTCacheDir()+"/", "", true) { if object.Err != nil { return object.Err } if object.LastModified.Add(cacheExpireInterval).Before(time.Now()) { - if err := s.ost.DeleteObject(object.Path); err != nil { + if err := s.ost.DeleteObject(ctx, object.Path); err != nil { if !objectstorage.IsNotExist(err) { s.log.Warn().Err(err).Msgf("failed to delete cache object %q", object.Path) } @@ -1663,14 +1661,12 @@ func (s *Runservice) objectsCleaner(ctx context.Context, prefix string, lockKey } defer func() { _ = l.Unlock() }() - doneCh := make(chan struct{}) - defer close(doneCh) - for object := range s.ost.List(prefix+"/", "", true, doneCh) { + for object := range s.ost.List(ctx, prefix+"/", "", true) { if object.Err != nil { return object.Err } if object.LastModified.Add(objectExpireInterval).Before(time.Now()) { - if err := s.ost.DeleteObject(object.Path); err != nil { + if err := s.ost.DeleteObject(ctx, object.Path); err != nil { if !objectstorage.IsNotExist(err) { s.log.Warn().Err(err).Msgf("failed to delete object %q", object.Path) } diff --git a/internal/util/context.go b/internal/util/context.go new file mode 100644 index 00000000..ae5e87af --- /dev/null +++ b/internal/util/context.go @@ -0,0 +1,13 @@ +package util + +import "context" + +// ContextCanceled returns whether a context is canceled. +func ContextCanceled(ctx context.Context) bool { + select { + case <-ctx.Done(): + return true + default: + return false + } +}