diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,10 @@ +## Version 0.0.2 (2018-12-27) + +- Fix missing `Real` from `mismi-p` +- Include `mismi-s3` + +## Version 0.0.1 (2018-12-16) + +- Replace `p` submodule with `mismi-p` +- Upgrade tests from QuickCheck to hedgehog +- Remove `amazonka`, `x`, `disorder` and `twine` submodules diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright 2017, Nick Hibberd , All Rights Reserved. + +Copyright 2017, Ambiata, All Rights Reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + 1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + 2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. + + 3. Neither the name of the copyright holder nor the names of + its contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/mismi-s3.cabal b/mismi-s3.cabal new file mode 100644 --- /dev/null +++ b/mismi-s3.cabal @@ -0,0 +1,265 @@ +version: 0.0.2 + +name: + mismi-s3 +author: + Nick Hibberd +maintainer: + Nick Hibberd +homepage: + https://github.com/nhibberd/mismi +bug-reports: + https://github.com/nhibberd/mismi/issues +synopsis: + AWS Library +description: + mismi-s3 is a library that provides a set of common and useful + operations on top of AWS S3. +category: + AWS +license: + BSD3 +license-file: + LICENSE +cabal-version: + >= 1.8 +build-type: + Simple +tested-with: + GHC == 8.2.2 + , GHC == 8.4.3 +extra-source-files: + CHANGELOG.md + + +library + build-depends: + base >= 3 && < 5 + , mismi-core == 0.0.2.* + , mismi-s3-core == 0.0.2.* + , mismi-p == 0.0.2.* + , template-haskell + , amazonka >= 1.5 && < 1.7 + , amazonka-core >= 1.5 && < 1.7 + , amazonka-s3 >= 1.5 && < 1.7 + , bytestring >= 0.10 && < 0.11 + , bifunctors >= 4.2 && < 6 + , conduit >= 1.2 && < 1.4 + , conduit-extra >= 1.1.13 && < 1.5 + , directory >= 1.2 && < 1.4 + , exceptions >= 0.7 && < 0.11 + , extra == 1.6.* + , filepath >= 1.3 && < 1.5 + , http-client >= 0.4.18 && < 0.6 + , http-types >= 0.8 && < 1 + , lifted-async >= 0.7 && < 0.11 + , mtl >= 2.1 && < 2.3 + , monad-loops == 0.4.* + , process >= 1.2 && < 1.7 + , resourcet >= 1.1 && < 1.3 + , semigroups >= 0.16 && < 0.19 + , text == 1.2.* + , transformers >= 0.3.1 && < 0.6 + , transformers-bifunctors >= 0.1 && < 1 + , attoparsec >= 0.12 && < 0.14 + , unix == 2.7.* + , async >= 2.0 && < 2.3 + , retry == 0.7.* + , lens >= 4.8 && < 4.18 + , SafeSemaphore == 0.10.* + , stm == 2.4.* + , unix-bytestring == 0.3.7.* + , time >= 1.4 && < 1.10 + , unordered-containers >= 0.2.5 && < 0.3 + , uuid == 1.3.* + , unliftio + , mmorph >= 1.0 && < 1.2 + , lifted-base == 0.2.* + + ghc-options: + -Wall + + hs-source-dirs: + src + + + exposed-modules: + Mismi.S3 + Mismi.S3.Amazonka + Mismi.S3.Commands + Mismi.S3.Stream + Mismi.S3.Patch.Network + Mismi.S3.Patch.PutObjectACL + Mismi.S3.Data + Mismi.S3.Unsafe + + Mismi.S3.Internal + Mismi.S3.Internal.Async + Mismi.S3.Internal.Binary + Mismi.S3.Internal.Parallel + Mismi.S3.Internal.Queue + +test-suite test + type: + exitcode-stdio-1.0 + + main-is: + test.hs + + ghc-options: + -Wall -threaded -O2 + + hs-source-dirs: + test + + other-modules: + Test.Mismi.S3 + Test.Mismi.S3.Commands + Test.Mismi.S3.Internal + + build-depends: + base + , amazonka >= 1.5 && < 1.7 + , amazonka-core >= 1.5 && < 1.7 + , amazonka-s3 >= 1.5 && < 1.7 + , hedgehog + , mismi-core == 0.0.2.* + , mismi-core-test == 0.0.2.* + , mismi-s3 + , mismi-s3-core == 0.0.2.* + , mismi-s3-core-test == 0.0.2.* + , mismi-p == 0.0.2.* + , conduit >= 1.2 && < 1.4 + , containers == 0.5.* + , directory >= 1.2 && < 1.4 + , exceptions >= 0.7 && < 0.11 + , filepath >= 1.3 && < 1.5 + , http-client >= 0.4.18 && < 0.6 + , lens >= 4.8 && < 4.18 + , mtl >= 2.1 && < 2.3 + , resourcet >= 1.1 && < 1.3 + , temporary >= 1.2 && < 1.4 + , text == 1.2.* + , time >= 1.4 && < 1.10 + , transformers >= 0.3.1 && < 0.6 + , transformers-bifunctors >= 0.1 && < 1 + , unix == 2.7.* + , uuid == 1.3.* + +test-suite test-io + type: + exitcode-stdio-1.0 + + main-is: + test-io.hs + + ghc-options: + -Wall -threaded -O2 + + hs-source-dirs: + test + + other-modules: + Test.Mismi.Amazonka + Test.Mismi.S3 + Test.IO.Mismi.S3.Commands + Test.IO.Mismi.S3.Control + Test.IO.Mismi.S3.Internal + + build-depends: + base + , mismi-core == 0.0.2.* + , mismi-core-test == 0.0.2.* + , mismi-s3 + , mismi-s3-core == 0.0.2.* + , mismi-s3-core-test == 0.0.2.* + , mismi-p == 0.0.2.* + , bytestring >= 0.10 && < 0.11 + , cryptohash + , directory >= 1.2 && < 1.4 + , exceptions >= 0.7 && < 0.11 + , filepath >= 1.3 && < 1.5 + , hedgehog + , lens >= 4.8 && < 4.18 + , mtl >= 2.1 && < 2.3 + , resourcet >= 1.1 && < 1.3 + , temporary >= 1.2 && < 1.4 + , text == 1.2.* + , time >= 1.4 && < 1.10 + , transformers >= 0.3.1 && < 0.6 + , unix == 2.7.* + , uuid == 1.3.* + + +test-suite test-reliability + type: + exitcode-stdio-1.0 + + main-is: + test-reliability.hs + + ghc-options: + -Wall -threaded -O2 + + hs-source-dirs: + test + + other-modules: + Test.Mismi.S3 + Test.Reliability.Reliability + Test.Reliability.Mismi.S3.Commands + + build-depends: + base + , mismi-core == 0.0.2.* + , mismi-core-test == 0.0.2.* + , mismi-s3 + , mismi-p == 0.0.2.* + , directory >= 1.2 && < 1.4 + , exceptions >= 0.7 && < 0.11 + , filepath >= 1.3 && < 1.5 + , hedgehog + , mtl >= 2.1 && < 2.3 + , resourcet >= 1.1 && < 1.3 + , text == 1.2.* + , time >= 1.4 && < 1.10 + , transformers >= 0.3.1 && < 0.6 + , unix == 2.7.* + , uuid == 1.3.* + +benchmark bench + type: + exitcode-stdio-1.0 + + main-is: + bench.hs + + ghc-options: + -Wall -threaded -O2 + + hs-source-dirs: + test + + other-modules: + Test.Mismi.S3 + + build-depends: + base + , mismi-core == 0.0.2.* + , mismi-core-test == 0.0.2.* + , mismi-s3 + , mismi-p == 0.0.2.* + , criterion >= 1.3 && < 1.5 + , conduit-extra >= 1.1.13 && < 1.5 + , directory >= 1.2 && < 1.4 + , exceptions >= 0.7 && < 0.11 + , filepath >= 1.3 && < 1.5 + , hedgehog + , mtl >= 2.1 && < 2.3 + , random + , resourcet >= 1.1 && < 1.3 + , temporary >= 1.2 && < 1.4 + , text == 1.2.* + , transformers >= 0.3.1 && < 0.6 + , unix == 2.7.* + , uuid == 1.3.* diff --git a/src/Mismi/S3.hs b/src/Mismi/S3.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3.hs @@ -0,0 +1,8 @@ +module Mismi.S3 ( + module X + ) where + +import Mismi.Control as X +import Mismi.S3.Commands as X +import Mismi.S3.Core.Data as X +import Mismi.S3.Data as X diff --git a/src/Mismi/S3/Amazonka.hs b/src/Mismi/S3/Amazonka.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Amazonka.hs @@ -0,0 +1,8 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Mismi.S3.Amazonka ( + module AWS + ) where + +import Mismi.Amazonka as AWS hiding (Document) + +import Network.AWS.S3 as AWS diff --git a/src/Mismi/S3/Commands.hs b/src/Mismi/S3/Commands.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Commands.hs @@ -0,0 +1,738 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE PackageImports #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Mismi.S3.Commands ( + headObject + , exists + , existsPrefix + , getSize + , size + , sizeRecursively + , delete + , read + , concatMultipart + , copy + , copyWithMode + , copyMultipart + , move + , upload + , uploadWithMode + , uploadRecursive + , uploadRecursiveWithMode + , multipartUpload + , uploadSingle + , write + , writeWithMode + , getObjects + , getObjectsRecursively + , listObjects + , list + , download + , downloadWithMode + , downloadSingle + , downloadWithRange + , downloadRecursive + , downloadRecursiveWithMode + , multipartDownload + , listMultipartParts + , listMultiparts + , listOldMultiparts + , listOldMultiparts' + , abortMultipart + , abortMultipart' + , filterOld + , filterNDays + , listRecursively + , sync + , syncWithMode + , createMultipartUpload + , grantReadAccess + , chunkFilesBySize + ) where + +import Control.Arrow ((***)) +import Control.Exception (ioError) +import qualified Control.Exception as CE +import Control.Lens ((.~), (^.), to, view) +import Control.Monad.Catch (Handler(..), throwM, onException) +import Control.Monad.Extra (concatMapM) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Reader (ask) +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Except (ExceptT (..), runExceptT, throwE) +import Control.Monad.Trans.Bifunctor (firstT) +import Control.Monad.Trans.Resource (allocate, runResourceT) +import qualified Control.Retry as Retry + +import Data.Conduit (runConduit, (.|)) +import qualified Data.Conduit as Conduit +import qualified Data.Conduit.Binary as Conduit +import qualified Data.Conduit.List as DC + +import Data.List (filter) +import qualified Data.List as L +import qualified Data.List.NonEmpty as NEL +import Data.Maybe (maybeToList, catMaybes, isJust) +import qualified Data.Text as T +import qualified Data.Text.Encoding as T +import qualified Data.Text.Lazy as TL +import qualified Data.Text.Lazy.Encoding as TL +import Data.Time.Clock (UTCTime, NominalDiffTime, getCurrentTime, addUTCTime) + +import Mismi.Amazonka (Env, send, paginate) +import Mismi.Control +import Mismi.S3.Core.Data +import Mismi.S3.Data +import Mismi.S3.Internal +import qualified Mismi.S3.Patch.Network as N +import qualified Mismi.S3.Patch.PutObjectACL as P +import qualified Mismi.S3.Internal.Binary as XB +import Mismi.S3.Internal.Queue (writeQueue) +import Mismi.S3.Internal.Parallel (consume) +import qualified Mismi.S3.Stream as Stream + +import Network.AWS.Data.Body (ChunkedBody (..), ChunkSize (..)) +import Network.AWS.Data.Body (RqBody (..), RsBody (..), toBody) +import Network.AWS.Data.Text (toText) +import Network.AWS.S3 (BucketName (..)) +import Network.AWS.S3 (GetObjectResponse, HeadObjectResponse) +import Network.AWS.S3 (ListObjects) +import Network.AWS.S3 (MetadataDirective (..)) +import Network.AWS.S3 (MultipartUpload, Part) +import Network.AWS.S3 (Object, ObjectKey (..)) +import qualified Network.AWS.S3 as A + +import P +import Prelude (toInteger) + +import System.IO (IO, IOMode (..), SeekMode (..)) +import System.IO (hFileSize, hSetFileSize, withFile) +import System.IO.Error (IOError) +import System.Directory (createDirectoryIfMissing, doesFileExist, getDirectoryContents) +import System.FilePath (FilePath, (), takeDirectory) +import System.Posix.IO (OpenMode(..), openFd, closeFd, fdSeek, defaultFileFlags) +import System.Posix.Files (fileSize, getFileStatus, isDirectory, isRegularFile) +import qualified "unix-bytestring" System.Posix.IO.ByteString as UBS + +import qualified UnliftIO.Async as UnliftIO +import qualified UnliftIO.Timeout as UnliftIO +import System.IO.Error (userError) + +-- | Retrieves the 'HeadObjectResponse'. Handles any 404 response by converting to Maybe. +-- +headObject :: Address -> AWS (Maybe HeadObjectResponse) +headObject a = + handle404 . send . f' A.headObject $ a + +-- | Checks for the existence of 'Address'. +-- +exists :: Address -> AWS Bool +exists a = + headObject a >>= pure . isJust + +existsPrefix :: Address -> AWS Bool +existsPrefix (Address (Bucket b) (Key k)) = + fmap (\r -> length (view A.lorsContents r) == 1 || length (view A.lorsCommonPrefixes r) == 1) . send $ A.listObjects (BucketName b) + & A.loPrefix .~ Just ((+/) k) + & A.loDelimiter .~ Just '/' + & A.loMaxKeys .~ Just 1 + +getSize :: Address -> AWS (Maybe Int) +getSize a = + size a >>= pure . fmap fromIntegral +{-# DEPRECATED getSize "Use Mismi.S3.Commands.size instead" #-} + +size :: Address -> AWS (Maybe Bytes) +size a = + headObject a >>= pure . fmap (Bytes . fromIntegral) . maybe Nothing (^. A.horsContentLength) + +sizeRecursively :: Address -> AWS [Sized Address] +sizeRecursively prefix = + runConduit $ Stream.sizeRecursively prefix .| DC.consume + +-- | Delete 'Address' +-- +delete :: Address -> AWS () +delete = + void . send . f' A.deleteObject + +-- | Retrieve the object at 'Address'. Handles any 404 response by converting to Maybe. +getObject' :: Address -> AWS (Maybe GetObjectResponse) +getObject' = + handle404 . send . f' A.getObject + +-- | Read contents of 'Address'. +-- +read :: Address -> AWS (Maybe Text) +read a = withRetries 5 $ do + r <- Stream.read a + z <- liftIO . sequence $ (\x -> runResourceT . runConduit $ x .| Conduit.sinkLbs) <$> r + pure $ fmap (T.concat . TL.toChunks . TL.decodeUtf8) z + +concatMultipart :: WriteMode -> Int -> [Address] -> Address -> ExceptT ConcatError AWS () +concatMultipart mode fork inputs dest = do + when (mode == Fail) . + whenM (lift $ exists dest) . + throwE $ ConcatDestinationExists $ dest + + when (null inputs) $ + throwE NoInputFiles + + things <- fmap (join . catMaybes) . forM inputs $ \input -> do + r <- lift $ size input + case r of + Nothing -> + throwE $ ConcatSourceMissing input + Just x -> + let + s = fromIntegral $ unBytes x + minChunk = 5 * 1024 * 1024 -- 5 MiB + chunk = 1024 * 1024 * 1024 -- 1 gb + big = 5 * 1024 * 1024 -- 5 gb + in + case s == 0 of + True -> + pure Nothing + False -> + case s < minChunk of + True -> + throwE $ ConcatSourceTooSmall input s + False -> + case s < big of + True -> + pure $ Just [(input, 0, s)] + False -> + let + chunks = calculateChunksCapped s chunk 4096 + in + pure . Just $ (\(a, b, _) -> (input, a, b)) <$> chunks + + when (null things) $ + throwE NoInputFilesWithData + + e <- ask + mpu <- lift $ createMultipartUpload dest + + let + (is, bs, ls) = L.unzip3 things + chunks = L.zip4 is bs ls [1..] + + rs <- liftIO $ + consume (forM_ chunks . writeQueue) fork $ multipartCopyWorker e mpu dest + + let + abort = + lift $ abortMultipart' dest mpu + + case rs of + Left f -> + abort >> + (throwE $ ConcatCopyError f) + + Right prts -> + flip onException abort $ + void . send $ f' A.completeMultipartUpload dest mpu & + A.cMultipartUpload .~ pure (A.completedMultipartUpload & A.cmuParts .~ sortPartResponse (snd prts)) + +copy :: Address -> Address -> ExceptT CopyError AWS () +copy s d = + copyWithMode Overwrite s d + +copyWithMode :: WriteMode -> Address -> Address -> ExceptT CopyError AWS () +copyWithMode mode s d = do + unlessM (lift $ exists s) . throwE $ CopySourceMissing s + when (mode == Fail) . whenM (lift $ exists d) . throwE $ CopyDestinationExists $ d + sz' <- lift $ getSize s + sz <- fromMaybeM (throwE $ CopySourceSize s) sz' + let + chunk = 100 * 1024 * 1024 -- 100 mb + big = 1024 * 1024 * 1024 -- 1 gb + case sz < big of + True -> + lift $ copySingle s d + False -> + copyMultipart s d sz chunk 100 + +copySingle :: Address -> Address -> AWS () +copySingle (Address (Bucket sb) (Key sk)) (Address (Bucket b) (Key dk)) = + void . send $ A.copyObject (BucketName b) (sb <> "/" <> sk) (ObjectKey dk) + & A.coServerSideEncryption .~ Just sse & A.coMetadataDirective .~ Just MDCopy + +copyMultipart :: Address -> Address -> Int -> Int -> Int -> ExceptT CopyError AWS () +copyMultipart source dest sz chunk fork = do + e <- ask + mpu <- lift $ createMultipartUpload dest -- target + + let + chunks = calculateChunksCapped sz chunk 4096 + things = (\(o, c, i) -> (source, o, c, i)) <$> chunks + + r <- liftIO $ + consume (forM_ things . writeQueue) fork $ multipartCopyWorker e mpu dest + + let abort = + lift $ abortMultipart' dest mpu + + case r of + Left f -> + abort >> + (throwE $ MultipartCopyError f) + + Right prts -> + flip onException abort $ + void . send $ f' A.completeMultipartUpload dest mpu & + A.cMultipartUpload .~ pure (A.completedMultipartUpload & A.cmuParts .~ sortPartResponse (snd prts)) + +-- Sort is required here because the completeMultipartUpload api expects an +-- ascending list of part id's +sortPartResponse :: [PartResponse] -> Maybe (NEL.NonEmpty A.CompletedPart) +sortPartResponse prts = + let + z = L.sortOn (\(PartResponse i _) -> i) prts + l = (\(PartResponse i etag) -> A.completedPart i etag) <$> z + in + NEL.nonEmpty l + +multipartCopyWorker :: Env -> Text -> Address -> (Address, Int, Int, Int) -> IO (Either Error PartResponse) +multipartCopyWorker e mpu dest (source, o, c, i) = do + let + sb = unBucket $ bucket source + sk = unKey $ key source + db = unBucket $ bucket dest + dk = unKey $ key dest + req = + A.uploadPartCopy (BucketName db) (sb <> "/" <> sk) (ObjectKey dk) i mpu + & A.upcCopySourceRange .~ (Just $ bytesRange o (o + c - 1)) + + Retry.recovering (Retry.fullJitterBackoff 500000) [s3Condition] $ \_ -> do + r <- runExceptT . runAWS e $ send req + case r of + Left z -> + pure $! Left z + + Right z -> do + pr <- fromMaybeM (throwM . Invariant $ "upcrsCopyPartResult") $ z ^. A.upcrsCopyPartResult + m <- fromMaybeM (throwM . Invariant $ "cprETag") $ pr ^. A.cprETag + pure $! Right $! PartResponse i m + +createMultipartUpload :: Address -> AWS Text +createMultipartUpload a = do + mpu <- send $ f' A.createMultipartUpload a & A.cmuServerSideEncryption .~ Just sse + maybe (throwM . Invariant $ "MultipartUpload: missing 'UploadId'") pure (mpu ^. A.cmursUploadId) + +move :: Address -> Address -> ExceptT CopyError AWS () +move source destination' = + copy source destination' >> + lift (delete source) + +upload :: FilePath -> Address -> ExceptT UploadError AWS () +upload = + uploadWithMode Fail + +uploadRecursive :: FilePath -> Address -> Int -> ExceptT UploadError AWS () +uploadRecursive = + uploadRecursiveWithMode Fail + +uploadWithMode :: WriteMode -> FilePath -> Address -> ExceptT UploadError AWS () +uploadWithMode m f a = do + when (m == Fail) . whenM (lift $ exists a) . throwE $ UploadDestinationExists a + unlessM (liftIO $ doesFileExist f) . throwE $ UploadSourceMissing f + s <- liftIO $ withFile f ReadMode $ \h -> + hFileSize h + case s < bigChunkSize of + True -> + lift $ uploadSingle f a + False -> + -- Originally had a concurrency of 100 (instead of 20). + -- + -- Based on the reasoning behind downloadWithMode which resulted in a 5 + -- as it's concurrency default. Testing showed that for upload 20 was a + -- better default. + case s > 1024 * 1024 * 1024 of + True -> + multipartUpload f a s (2 * bigChunkSize) 20 + False -> + multipartUpload f a s bigChunkSize 20 + + + +bigChunkSize :: Integer +bigChunkSize = 100 * 1024 * 1024 + +uploadSingle :: FilePath -> Address -> AWS () +uploadSingle file a = do + rq <- N.chunkedFile (ChunkSize $ 1024 * 1024) file + void . send $ f' A.putObject a rq & A.poServerSideEncryption .~ pure sse + +multipartUpload :: FilePath -> Address -> Integer -> Integer -> Int -> ExceptT UploadError AWS () +multipartUpload file a fSize chunk fork = do + e <- ask + mpu <- lift $ createMultipartUpload a + + let chunks = calculateChunksCapped (fromInteger fSize) (fromInteger chunk) 4096 -- max 4096 prts returned + + r <- liftIO $ + consume (forM_ chunks . writeQueue) fork $ multipartUploadWorker e mpu file a + + let abort = lift $ abortMultipart' a mpu + + case r of + Left f -> + abort >> + (throwE $ MultipartUploadError f) + + Right prts -> + flip onException abort $ + void . send $ f' A.completeMultipartUpload a mpu & + A.cMultipartUpload .~ pure (A.completedMultipartUpload & A.cmuParts .~ sortPartResponse (snd prts)) + + +multipartUploadWorker :: Env -> Text -> FilePath -> Address -> (Int, Int, Int) -> IO (Either Error PartResponse) +multipartUploadWorker e mpu file a (o, c, i) = + withFile file ReadMode $ \h -> + let + cs = (1024 * 1024) -- 1 mb + cl = toInteger c + b = XB.slurpHandle h (toInteger o) (Just $ toInteger c) + cb = ChunkedBody cs cl b + req' = f' A.uploadPart a i mpu $ Chunked cb + in + Retry.recovering (Retry.fullJitterBackoff 500000) [s3Condition] $ \_ -> do + r <- runExceptT . runAWS e $ send req' + case r of + Left z -> + pure $! Left z + Right z -> do + m <- fromMaybeM (throwM MissingETag) $ z ^. A.uprsETag + pure $! Right $! PartResponse i m + +s3Condition :: Applicative a => Retry.RetryStatus -> Handler a Bool +s3Condition s = + Handler $ \(ex :: S3Error) -> + pure $ case ex of + MissingETag -> + Retry.rsIterNumber s < 5 + _ -> + False + +uploadRecursiveWithMode :: WriteMode -> FilePath -> Address -> Int -> ExceptT UploadError AWS () +uploadRecursiveWithMode mode src (Address buck ky) fork = do + es <- tryIO $ getFileStatus src + case es of + Left _ -> throwE $ UploadSourceMissing src + Right st -> unless (isDirectory st) . throwE $ UploadSourceNotDirectory src + files <- liftIO (listRecursivelyLocal src) + mapM_ uploadFiles $ chunkFilesBySize fork (fromIntegral bigChunkSize) files + where + uploadFiles :: [(FilePath, Int64)] -> ExceptT UploadError AWS () + uploadFiles [] = pure () + uploadFiles [(f,s)] + | fromIntegral s < bigChunkSize = lift . uploadSingle f $ uploadAddress f + | otherwise = uploadWithMode mode f $ uploadAddress f + uploadFiles xs = + lift $ UnliftIO.mapConcurrently_ (\ (f, _) -> uploadSingle f $ uploadAddress f) xs + + + prefixLen = L.length (src "a") - 1 + + uploadAddress :: FilePath -> Address + uploadAddress fp = + Address buck (ky // Key (T.pack $ L.drop prefixLen fp)) + +-- Take a list of files and their sizes, and convert it to a list of tests +-- where the total size of the files in the sub list is less than `maxSize` +-- and the length of the sub lists is <= `maxCount`. +chunkFilesBySize :: Int -> Int64 -> [(FilePath, Int64)] -> [[(FilePath, Int64)]] +chunkFilesBySize maxCount maxSize = + takeFiles 0 [] . L.sortOn snd + where + takeFiles :: Int64 -> [(FilePath, Int64)] -> [(FilePath, Int64)] -> [[(FilePath, Int64)]] + takeFiles _ acc [] = [acc] + takeFiles current acc ((x, s):xs) = + if current + s < maxSize && L.length acc < maxCount + then takeFiles (current + s) ((x, s):acc) xs + else acc : takeFiles s [(x, s)] xs + +-- | Like `listRecursively` but for the local filesystem. +-- Also returns +listRecursivelyLocal :: MonadIO m => FilePath -> m [(FilePath, Int64)] +listRecursivelyLocal topdir = do + entries <- liftIO $ listDirectory topdir + (dirs, files) <- liftIO . partitionDirsFilesWithSizes $ fmap (topdir ) entries + others <- concatMapM listRecursivelyLocal dirs + pure $ files <> others + + +-- Not available with ghc 7.10 so copy it here. +listDirectory :: FilePath -> IO [FilePath] +listDirectory path = + filter f <$> getDirectoryContents path + where + f filename = + filename /= "." && filename /= ".." + +partitionDirsFilesWithSizes :: MonadIO m => [FilePath] -> m ([FilePath], [(FilePath, Int64)]) +partitionDirsFilesWithSizes = + pworker ([], []) + where + pworker (dirs, files) [] = pure (dirs, files) + pworker (dirs, files) (x:xs) = do + xstat <- liftIO $ getFileStatus x + let xsize = fromIntegral $ fileSize xstat + newDirs = if isDirectory xstat then x : dirs else dirs + newFiles = if isRegularFile xstat then (x, xsize) : files else files + pworker (newDirs, newFiles) xs + +write :: Address -> Text -> AWS WriteResult +write = + writeWithMode Fail + +writeWithMode :: WriteMode -> Address -> Text -> AWS WriteResult +writeWithMode w a t = do + result <- runExceptT $ do + case w of + Fail -> + whenM (lift $ exists a) $ + throwE (WriteDestinationExists a) + Overwrite -> + return () + void . lift . send $ + f' A.putObject a (toBody . T.encodeUtf8 $ t) + & A.poServerSideEncryption .~ Just sse + + either pure (const $ pure WriteOk) result + +-- pair of prefixs and keys +getObjects :: Address -> AWS ([Key], [Key]) +getObjects (Address (Bucket buck) (Key ky)) = + ((Key <$>) *** ((\(ObjectKey t) -> Key t) <$>)) <$> ff (A.listObjects (BucketName buck) & A.loPrefix .~ Just ((+/) ky) & A.loDelimiter .~ Just '/' ) + where + ff :: ListObjects -> AWS ([T.Text], [ObjectKey]) + ff b = do + r <- send b + if r ^. A.lorsIsTruncated == Just True + then + do + let d = (maybeToList =<< fmap (^. A.cpPrefix) (r ^. A.lorsCommonPrefixes), fmap (^. A.oKey) (r ^. A.lorsContents)) + n <- ff $ b & A.loMarker .~ (r ^. A.lorsNextMarker) + pure $ d <> n + else + pure (maybeToList =<< fmap (^. A.cpPrefix) (r ^. A.lorsCommonPrefixes), fmap (^. A.oKey) (r ^. A.lorsContents)) + +getObjectsRecursively :: Address -> AWS [Object] +getObjectsRecursively (Address (Bucket b) (Key ky)) = + getObjects' $ A.listObjects (BucketName b) & A.loPrefix .~ Just ((+/) ky) + where + -- Hoping this will have ok performance in cases where the results are large, it shouldnt + -- affect correctness since we search through the list for it anyway + go x ks = (NEL.toList ks <>) <$> getObjects' (x & A.loMarker .~ Just (toText $ NEL.last ks ^. A.oKey)) + getObjects' :: ListObjects -> AWS [Object] + getObjects' x = do + resp <- send x + if resp ^. A.lorsIsTruncated == Just True + then + maybe + (throwM . Invariant $ "Truncated response with empty contents list.") + (go x) + (NEL.nonEmpty $ resp ^. A.lorsContents) + else + pure $ resp ^. A.lorsContents + +-- | Return a tuple of the prefixes and keys at the provided S3 Address. +listObjects :: Address -> AWS ([Address], [Address]) +listObjects a = + (\(p, k) -> (Address (bucket a) <$> p, Address (bucket a) <$> k)) <$> getObjects a + +list :: Address -> AWS [Address] +list a = + runConduit $ Stream.list a .| DC.consume + +download :: Address -> FilePath -> ExceptT DownloadError AWS () +download = + downloadWithMode Fail + +downloadWithMode :: WriteMode -> Address -> FilePath -> ExceptT DownloadError AWS () +downloadWithMode mode a f = do + when (mode == Fail) . whenM (liftIO $ doesFileExist f) . throwE $ DownloadDestinationExists f + liftIO $ createDirectoryIfMissing True (takeDirectory f) + + sz' <- lift $ getSize a + sz <- maybe (throwE $ DownloadSourceMissing a) pure sz' + + if (sz > 200 * 1024 * 1024) + then -- Originally had a concurrecy of 100 (instead of 5). Tested a number of + -- values between 2 and 100 and found empirically that 5 gave the fastest + -- downloads (less than 10% better), but significantly reduced the + -- likelihood of triggering the S3 rate limiter (by a factor of 20) + -- which in turn reduces the liklehood of `IOExceptions` and hung + -- threads. + multipartDownload a f sz 100 5 + else downloadSingle a f + +downloadSingle :: Address -> FilePath -> ExceptT DownloadError AWS () +downloadSingle a f = do + r <- maybe (throwE $ DownloadSourceMissing a) pure =<< lift (getObject' a) + liftIO . withRetries 5 . withFileSafe f $ \p -> + runResourceT . runConduit $ + (r ^. A.gorsBody ^. to _streamBody) .| Conduit.sinkFile p + +multipartDownload :: Address -> FilePath -> Int -> Integer -> Int -> ExceptT DownloadError AWS () +multipartDownload source destination sz chunk fork = + firstT MultipartError $ do + e <- ask + let chunks = calculateChunks sz (fromInteger $ chunk * 1024 * 1024) + void . withFileSafe destination $ \f -> do + liftIO $ withFile f WriteMode $ \h -> + hSetFileSize h (toInteger sz) + + ExceptT . liftIO . + consume (\q -> mapM (writeQueue q) chunks) fork $ \(o, c, _) -> + runExceptT . runAWS e $ downloadWithRange source o (o + c) f + +downloadWithRange :: Address -> Int -> Int -> FilePath -> AWS () +downloadWithRange a start end dest = withRetries 5 $ do + -- Use a timeout of ten minutes. Arrivied at empirically. With a timeout of 5 + -- minutes this was triggering too often. Want this to be the last resort. + res <- UnliftIO.timeout (10 * 60 * 1000 * 1000) $ do + r <- send $ f' A.getObject a & + A.goRange .~ (Just $ bytesRange start end) + + -- write to file + liftIO . runResourceT $ do + fd <- snd <$> allocate (openFd dest WriteOnly Nothing defaultFileFlags) closeFd + void . liftIO $ fdSeek fd AbsoluteSeek (fromInteger . toInteger $ start) + let + source = r ^. A.gorsBody ^. to _streamBody + sink = Conduit.awaitForever $ liftIO . UBS.fdWrite fd + runConduit $ source .| sink + + case res of + Just () -> + pure () + Nothing -> + liftIO $ ioError (userError "downloadWithRange timeout") + +downloadRecursiveWithMode :: WriteMode -> Address -> FilePath -> ExceptT DownloadError AWS () +downloadRecursiveWithMode mode src dest = do + -- Check if the destination already exists and is not a directory. + es <- tryIO $ getFileStatus dest + case es of + Left _ -> pure () + Right st -> unless (isDirectory st) . throwE $ DownloadDestinationNotDirectory dest + -- Real business starts here. + addrs <- lift $ listRecursively src + mapM_ drWorker addrs + where + drWorker :: Address -> ExceptT DownloadError AWS () + drWorker addr = do + fpdest <- maybe (throwE $ DownloadInvariant addr src) pure $ + (() dest) . T.unpack . unKey <$> removeCommonPrefix src addr + downloadWithMode mode addr fpdest + +downloadRecursive :: Address -> FilePath -> ExceptT DownloadError AWS () +downloadRecursive = + downloadRecursiveWithMode Fail + +listMultipartParts :: Address -> Text -> AWS [Part] +listMultipartParts a uploadId = do + let req = f' A.listParts a uploadId + runConduit $ paginate req .| DC.foldMap (^. A.lprsParts) + +listMultiparts :: Bucket -> AWS [MultipartUpload] +listMultiparts (Bucket bn) = do + let req = A.listMultipartUploads $ BucketName bn + runConduit $ paginate req .| DC.foldMap (^. A.lmursUploads) + +listOldMultiparts :: Bucket -> AWS [MultipartUpload] +listOldMultiparts b = do + mus <- listMultiparts b + now <- liftIO getCurrentTime + pure $ filter (filterOld now) mus + +listOldMultiparts' :: Bucket -> Int -> AWS [MultipartUpload] +listOldMultiparts' b i = do + mus <- listMultiparts b + now <- liftIO getCurrentTime + pure $ filter (filterNDays i now) mus + +filterOld :: UTCTime -> MultipartUpload -> Bool +filterOld = filterNDays 7 + +filterNDays :: Int -> UTCTime -> MultipartUpload -> Bool +filterNDays n now m = case m ^. A.muInitiated of + Nothing -> False + Just x -> nDaysOld n now x + +nDaysOld :: Int -> UTCTime -> UTCTime -> Bool +nDaysOld n now utc = do + let n' = fromInteger $ toInteger n + let diff = -1 * 60 * 60 * 24 * n' :: NominalDiffTime + let boundary = addUTCTime diff now + boundary > utc + +abortMultipart :: Bucket -> MultipartUpload -> AWS () +abortMultipart (Bucket b) mu = do + (ObjectKey k) <- maybe (throwM $ Invariant "Multipart key missing") pure (mu ^. A.muKey) + i <- maybe (throwM $ Invariant "Multipart uploadId missing") pure (mu ^. A.muUploadId) + abortMultipart' (Address (Bucket b) (Key k)) i + +abortMultipart' :: Address -> Text -> AWS () +abortMultipart' a i = + void . send $ f' A.abortMultipartUpload a i + +listRecursively :: Address -> AWS [Address] +listRecursively a = + runConduit $ Stream.listRecursively a .| DC.consume + +grantReadAccess :: Address -> ReadGrant -> AWS () +grantReadAccess a g = + void . send $ (f' P.putObjectACL a & P.poaGrantRead .~ Just (readGrant g)) + +sync :: Address -> Address -> Int -> ExceptT SyncError AWS () +sync = + syncWithMode FailSync + +syncWithMode :: SyncMode -> Address -> Address -> Int -> ExceptT SyncError AWS () +syncWithMode mode source dest fork = do + e <- ask + void . firstT SyncError . ExceptT . liftIO $ + (consume (sinkQueue e (Stream.listRecursively source)) fork (worker source dest mode e)) + +worker :: Address -> Address -> SyncMode -> Env -> Address -> IO (Either SyncWorkerError ()) +worker input output mode env f = runExceptT . runAWST env SyncAws $ do + n <- maybe (throwE $ SyncInvariant input f) pure $ removeCommonPrefix input f + let out = withKey (// n) output + liftCopy = firstT SyncCopyError + cp = liftCopy $ copy f out + foldSyncMode + (ifM (lift $ exists out) (throwE $ OutputExists out) cp) + (liftCopy $ copyWithMode Overwrite f out) + (ifM (lift $ exists out) (pure ()) cp) + mode + +tryIO :: MonadIO m => IO a -> m (Either IOError a) +tryIO = liftIO . CE.try + + + +-- compat + +ifM :: Monad m => m Bool -> m a -> m a -> m a +ifM p x y = + p >>= \b -> if b then x else y + +whenM :: Monad m => m Bool -> m () -> m () +whenM p m = + p >>= flip when m + +unlessM :: Monad m => m Bool -> m () -> m () +unlessM p m = + p >>= flip unless m + + +fromMaybeM :: Applicative f => f a -> Maybe a -> f a +fromMaybeM = + flip maybe pure diff --git a/src/Mismi/S3/Data.hs b/src/Mismi/S3/Data.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Data.hs @@ -0,0 +1,256 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE DeriveDataTypeable #-} +module Mismi.S3.Data ( + PartResponse (..) + , WriteMode (..) + , SyncMode (..) + , Bucket (..) + , Address (..) + , Key (..) + , ReadGrant (..) + , Upload (..) + , S3Error (..) + , ErrorType (..) + , DownloadError (..) + , ConcatError (..) + , CopyError (..) + , UploadError (..) + , SyncError (..) + , SyncWorkerError (..) + , WriteResult (..) + , (//) + , combineKey + , dirname + , foldWriteMode + , foldSyncMode + , basename + , addressFromText + , addressToText + , removeCommonPrefix + , withKey + , s3Parser + , s3ErrorRender + , renderDownloadError + , renderConcatError + , renderCopyError + , renderUploadError + , renderSyncError + , renderSyncWorkerError + , sse + ) where + +import Control.Exception (Exception) + +import qualified Data.Text as T +import Data.Typeable (Typeable) + +import P + +import Mismi (Error, renderError) + -- Just for compatibility, would be good to not do + -- this at some point but for now we import everything + -- and keep current export list. +import Mismi.S3.Core.Data (Address (..), Bucket (..), Key (..), ReadGrant (..) + , SyncMode (..), WriteMode (..), WriteResult (..), (//) + , addressFromText, addressToText, basename, combineKey + , dirname, foldSyncMode, foldWriteMode, removeCommonPrefix + , s3Parser, withKey) +import Mismi.S3.Internal.Parallel (RunError (..), renderRunError) +import Network.AWS.S3 (ETag, ServerSideEncryption (..)) + +import System.FilePath (FilePath) + + + +data PartResponse = + PartResponse !Int !ETag + deriving (Eq, Show) + +data S3Error = + SourceMissing ErrorType Address + | SourceFileMissing FilePath + | DestinationAlreadyExists Address + | DestinationDoesNotExist Address + | DestinationFileExists FilePath + | DestinationNotDirectory FilePath + | DestinationMissing FilePath + | SourceNotDirectory FilePath + | AccessDenied Address + | Invariant Text + | Target Address Address + | MissingETag + deriving (Eq, Typeable) + +instance Exception S3Error + +instance Show S3Error where + show = T.unpack . s3ErrorRender + +s3ErrorRender :: S3Error -> Text +s3ErrorRender s3err = "[Mismi internal error] - " <> case s3err of + SourceMissing e a -> + "Can not " <> renderErrorType e <> " when the source object does not exist [" <> addressToText a <> "]" + SourceFileMissing f -> + "Can not upload when the source file does not exist [" <> T.pack f <> "]" + DestinationAlreadyExists a -> + "Can not upload to an address that already exists [" <> addressToText a <> "]" + DestinationFileExists f -> + "Can not download to a target that already exists [" <> T.pack f <> "]" + DestinationNotDirectory f -> + "Expecting destination " <> T.pack f <> " to be a directory." + DestinationMissing f -> + "Download destination directory " <> T.pack f <> " does not exist." + SourceNotDirectory f -> + "Recursive upload source " <> T.pack f <> " must be a directory." + DestinationDoesNotExist a -> + "This address does not exist [" <> addressToText a <> "]" + AccessDenied a -> + "The access to this address is denied [" <> addressToText a <> "]" + Invariant e -> + e + Target a o -> + "Can not copy [" <> addressToText a <> "] to [" <> addressToText o <> "]. Target file exists" + MissingETag -> + "missing ETag" + +data ErrorType = + DownloadError + | CopyError' + deriving (Eq, Show) + +renderErrorType :: ErrorType -> Text +renderErrorType e = case e of + DownloadError -> + "download" + CopyError' -> + "copy" + +data DownloadError = + DownloadSourceMissing Address + | DownloadDestinationExists FilePath + | DownloadDestinationNotDirectory FilePath + | DownloadInvariant Address Address + | MultipartError (RunError Error) + deriving Show + +renderDownloadError :: DownloadError -> Text +renderDownloadError d = + case d of + DownloadSourceMissing a -> + "Can not download when the source object does not exist [" <> addressToText a <> "]" + DownloadDestinationExists f -> + "Can not download to a target that already exists [" <> T.pack f <> "]" + DownloadDestinationNotDirectory f -> + "Destination for a recursive download, " <> T.pack f <> " is not a directory." + DownloadInvariant a b -> + "Remove common prefix invariant: " <> + "[" <> addressToText b <> "] is not a common prefix of " <> + "[" <> addressToText a <> "]" + MultipartError r -> + "Multipart download error: " <> renderRunError r renderError + +data ConcatError = + ConcatSourceMissing Address + | ConcatDestinationExists Address + | ConcatCopyError (RunError Error) + | NoInputFiles + | NoInputFilesWithData + | ConcatSourceTooSmall Address Int + deriving Show + +renderConcatError :: ConcatError -> Text +renderConcatError e = + case e of + ConcatSourceMissing a -> + "Can not concat objects when the source object does not exist [" <> addressToText a <> "]" + ConcatDestinationExists a -> + "Can not concat objects when the destination object already exists [" <> addressToText a <> "]" + ConcatCopyError a -> + renderRunError a ((<>) "Multipart concat failed on a worker: " . renderError) + NoInputFiles -> + "Can not concat with no input keys." + NoInputFilesWithData -> + "Can not concat with no input keys with data." + ConcatSourceTooSmall a s -> + T.intercalate " " [ + "Source file" + , addressToText a + , "(" <> T.pack (show s) <> ") bytes" + , "is too small to use as part of a multipart upload." + ] + +data CopyError = + CopySourceMissing Address + | CopyDestinationExists Address + | CopySourceSize Address + | MultipartCopyError (RunError Error) + +renderCopyError :: CopyError -> Text +renderCopyError e = + case e of + CopySourceMissing a -> + "Can not copy an object when the source object does not exist [" <> addressToText a <> "]" + CopyDestinationExists a -> + "Can not copy an object when the destination object already exists [" <> addressToText a <> "]" + CopySourceSize a -> + "Can not calculate the size of the source object [" <> addressToText a <> "]" + MultipartCopyError a -> + renderRunError a ((<>) "Multipart copy failed on a worker: " . renderError) + +data UploadError = + UploadSourceMissing FilePath + | UploadDestinationExists Address + | UploadSourceNotDirectory FilePath + | MultipartUploadError (RunError Error) + deriving Show + +renderUploadError :: UploadError -> Text +renderUploadError e = + case e of + UploadSourceMissing f -> + "Can not upload when the source file does not exist [" <> T.pack f <> "]" + UploadDestinationExists a -> + "Can not upload when the destination object already exists [" <> addressToText a <> "]" + UploadSourceNotDirectory f -> + "Recursive upload source " <> T.pack f <> " must be a directory." + MultipartUploadError a -> + renderRunError a ((<>) "Multipart upload failed on a worker: " . renderError) + +newtype SyncError = + SyncError (RunError SyncWorkerError) + +renderSyncError :: SyncError -> Text +renderSyncError (SyncError r) = + renderRunError r renderSyncWorkerError + +data SyncWorkerError = + SyncInvariant Address Address + | OutputExists Address + | SyncAws Error + | SyncCopyError CopyError + +renderSyncWorkerError :: SyncWorkerError -> Text +renderSyncWorkerError w = + case w of + SyncInvariant a b -> + "Remove common prefix invariant: " <> + "[" <> addressToText b <> "] is not a common prefix of " <> + "[" <> addressToText a <> "]" + OutputExists a -> + "Can not copy to an address that already exists [" <> addressToText a <> "]" + SyncAws e -> + "AWS failure during 'sync': " <> renderError e + SyncCopyError c -> + "Copy failure during 'sync': " <> renderCopyError c + + +data Upload = + UploadSingle + | UploadMultipart Integer Integer + deriving (Eq, Show) + +sse :: ServerSideEncryption +sse = + AES256 diff --git a/src/Mismi/S3/Internal.hs b/src/Mismi/S3/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Internal.hs @@ -0,0 +1,122 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE BangPatterns #-} +module Mismi.S3.Internal ( + f' + , (+/) + , calculateChunks + , calculateChunksCapped + , bytesRange + , sinkChan + , sinkChanWithDelay + , sinkQueue + , waitForNResults + , withFileSafe + ) where + +import Control.Concurrent (Chan, readChan, threadDelay, writeChan) + +import Control.Monad.Catch (MonadCatch, onException) +import Control.Monad.IO.Class (MonadIO, liftIO) + +import Data.Conduit (ConduitT, runConduit, (.|)) +import qualified Data.Conduit.List as DC + +import qualified Data.Text as T +import Data.UUID (toString) +import Data.UUID.V4 (nextRandom) + +import P +import Prelude (ceiling, (/)) + +import Mismi (AWS, rawRunAWS) +import Mismi.Amazonka (Env) +import Mismi.S3.Data +import Network.AWS.S3 (BucketName (..), ObjectKey (..)) +import Mismi.S3.Internal.Queue (Queue, writeQueue) + +import System.Directory (renameFile, removeFile) +import System.IO (IO) +import System.FilePath (FilePath, takeDirectory, takeFileName) + + +f' :: (BucketName -> ObjectKey -> a) -> Address -> a +f' f (Address (Bucket b) k) = + BucketName b `f` ObjectKey (unKey k) + +-- | add a "/" at the end of some text if missing and if the text is not empty +(+/) :: Text -> Text +(+/) k + | T.null k = "" + | T.isSuffixOf "/" k = k + | otherwise = k <> "/" + + +calculateChunksCapped :: Int -> Int -> Int -> [(Int, Int, Int)] +calculateChunksCapped size chunk' capped = + calculateChunks size cappedChunk + where + minChunk = ceiling $ size' / capped' + + cappedChunk = max chunk' minChunk + + size' :: Double + size' = fromIntegral size + + capped' :: Double + capped' = fromIntegral capped + +-- filesize -> Chunk -> [(offset, chunk, index)] +calculateChunks :: Int -> Int -> [(Int, Int, Int)] +calculateChunks size chunk' = + let chunk = max 1 chunk' + go :: Int -> Int -> [(Int, Int, Int)] + go !index offset = + let !offset' = (offset + chunk) in + if (offset' < size) + then + (offset, chunk, index) : go (index + 1) offset' + else + let !c' = (size - offset) in -- last chunk + [(offset, c', index)] + in + go 1 0 + +-- http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.35 +-- https://github.com/aws/aws-sdk-java/blob/master/aws-java-sdk-s3/src/main/java/com/amazonaws/services/s3/AmazonS3Client.java#L1135 +bytesRange :: Int -> Int -> Text +bytesRange start end = + T.pack $ "bytes=" <> show start <> "-" <> show end + +sinkChan :: MonadIO m => ConduitT () a m () -> Chan a -> m Int +sinkChan source c = + sinkChanWithDelay 0 source c + +sinkChanWithDelay :: MonadIO m => Int -> ConduitT () a m () -> Chan a -> m Int +sinkChanWithDelay delay source c = + runConduit $ source .| DC.foldM (\i v -> liftIO $ threadDelay delay >> writeChan c v >> pure (i + 1)) 0 + +sinkQueue :: Env -> ConduitT () a AWS () -> Queue a -> IO () +sinkQueue e source q = + rawRunAWS e (runConduit $ source .| DC.mapM_ (liftIO . writeQueue q)) + + +waitForNResults :: Int -> Chan a -> IO [a] +waitForNResults i c = do + let waitForDone acc = + if (length acc == i) + then pure acc + else do + r <- readChan c + waitForDone (r : acc) + waitForDone [] + +-- | Create a temporary file location that can be used safely, and on a successful operation, do an (atomic) rename +-- NOTE: This function requires that the `FilePath` provided in the callback exists, otherwise throws an exception +withFileSafe :: (MonadCatch m, MonadIO m) => FilePath -> (FilePath -> m a) -> m a +withFileSafe f1 run = do + uuid <- liftIO nextRandom >>= return . toString + let f2 = takeDirectory f1 <> "/" <> "." <> takeFileName f1 <> "." <> uuid + onException + (run f2 >>= \a -> liftIO (renameFile f2 f1) >> return a) + (liftIO $ removeFile f2) diff --git a/src/Mismi/S3/Internal/Async.hs b/src/Mismi/S3/Internal/Async.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Internal/Async.hs @@ -0,0 +1,25 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Mismi.S3.Internal.Async ( + waitEitherBoth + ) where + + +import Control.Concurrent.Async (Async, waitSTM) +import Control.Concurrent.STM (atomically, orElse, retry) + +import P + +import System.IO (IO) + +waitEitherBoth :: Async a -> Async b -> Async c -> IO (Either a (b, c)) +waitEitherBoth a b c = + atomically $ do + let + l = waitSTM a + r = do + bb <- waitSTM b `orElse` (waitSTM c >> retry) + cc <- waitSTM c + return (bb, cc) + fmap Left l `orElse` fmap Right r diff --git a/src/Mismi/S3/Internal/Binary.hs b/src/Mismi/S3/Internal/Binary.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Internal/Binary.hs @@ -0,0 +1,60 @@ +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE DeriveDataTypeable #-} +module Mismi.S3.Internal.Binary ( + slurpHandle + , slurpHandleWithBuffer + ) where + +import Control.Monad.IO.Class (MonadIO, liftIO) + +import qualified Data.ByteString as BS +import Data.Conduit (ConduitT, yield) + +import System.IO (Handle, hSeek, SeekMode (..)) + +-- This is based on `conduit-extra` version 1.1.7.3 +-- Data.Conduit.Binary +-- sourceFileRange +-- souceHandleRange +-- +-- Ideally this will be replaced by conduit-extra in a future release +-- based on (https://github.com/snoyberg/conduit/pull/213) +-- + +slurpHandle :: MonadIO m + => Handle + -> Integer -- ^ Offset + -> Maybe Integer -- ^ Maximum count + -> ConduitT () BS.ByteString m () +slurpHandle handle offset count' = + slurpHandleWithBuffer handle offset count' (1024 * 1024) + +slurpHandleWithBuffer :: MonadIO m + => Handle + -> Integer -- ^ Offset + -> Maybe Integer -- ^ Maximum count + -> Int -- ^ Buffer size + -> ConduitT () BS.ByteString m () +slurpHandleWithBuffer handle offset count' buffer = do + liftIO $ hSeek handle AbsoluteSeek offset + case count' of + Nothing -> pullUnlimited + Just c -> pullLimited (fromInteger c) + where + pullUnlimited = do + bs <- liftIO $ BS.hGetSome handle buffer + if BS.null bs + then return () + else do + yield bs + pullUnlimited + + pullLimited c = do + bs <- liftIO $ BS.hGetSome handle (min c $ buffer) + let c' = c - BS.length bs + if BS.null bs + then return () + else do + yield bs + pullLimited c' diff --git a/src/Mismi/S3/Internal/Parallel.hs b/src/Mismi/S3/Internal/Parallel.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Internal/Parallel.hs @@ -0,0 +1,216 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ExplicitForAll #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Mismi.S3.Internal.Parallel ( + -- * Data + Workers + , Result + , newResult + , emptyResult + , addResult + , getResult + , emptyWorkers + , failWorkers + , getWorkers + , addWorker + , waitForWorkers + , waitForWorkers' + + -- * Operations + , RunError (..) + , renderRunError + , consume_ + , consume + , waitEitherBoth + ) where + +import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (Async, async, cancel, poll) +import Control.Concurrent.Async (waitBoth, wait, waitEither) +import Control.Concurrent.MSem (new, signal) +import qualified Control.Concurrent.MSem as M +import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar, modifyMVar_) +import Control.Monad.Catch (Exception, SomeException, catch, catchAll, finally, throwM) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Loops (untilM_) +import Control.Monad.Trans.Except (ExceptT (..), runExceptT) +import Control.Monad.Trans.Bifunctor (firstT) + +import qualified Data.Text as T +import Data.Typeable (Typeable) + +import P + +import Mismi.S3.Internal.Async (waitEitherBoth) +import Mismi.S3.Internal.Queue + +newtype Result a = + Result (MVar a) + +newResult :: a -> IO (Result a) +newResult a = + Result <$> newMVar a + +-- fix pair with takeMVar +emptyResult :: IO (Result a) +emptyResult = + Result <$> newEmptyMVar + +addResult :: Monad m => Result (m a) -> m a -> IO (m a) +addResult (Result r) w = + modifyMVar r (\x -> pure $ ((x >> w), x)) + +getResult :: Result a -> IO a +getResult (Result r) = + readMVar r + +newtype Workers a = + Workers (MVar [Async a]) + +emptyWorkers :: IO (Workers a) +emptyWorkers = + Workers <$> newMVar [] + +failWorkers :: Workers a -> IO () +failWorkers (Workers w) = + readMVar w >>= + mapM_ cancel + +getWorkers :: Workers a -> IO [Async a] +getWorkers (Workers w) = + readMVar w + +addWorker :: (Workers a) -> Async a -> IO () +addWorker (Workers w) r = + modifyMVar_ w $ pure . (:) r + +waitForWorkers :: (Workers a) -> IO () +waitForWorkers (Workers w) = + readMVar w >>= mapM_ wait + +waitForWorkers' :: (Workers a) -> IO [a] +waitForWorkers' (Workers w) = + readMVar w >>= mapM wait + +-- | Provide a producer and an action to be run across the result +-- of that producer in parallel. +-- +-- +-- Common usage: +-- @ +-- let producer :: Address -> Queue Address -> IO () +-- producer prefix q = +-- list' prefix $$ writeQueue q +-- +-- consume producer 100 (\(a :: Address) -> doThis) +-- @ +-- +consume_ :: MonadIO m => (Queue b -> IO a) -> Int -> (b -> ExceptT e IO ()) -> ExceptT (RunError e) m a +consume_ pro fork action = ExceptT . liftIO $ do + q <- newQueue fork + + producer <- async $ pro q + + workers <- emptyWorkers + result <- newResult $ Right () + sem <- new fork + + let spawn = do + m <- tryReadQueue q + flip (maybe $ return ()) m $ \a -> do + w <- do + M.wait sem + async $ (runExceptT (action a) >>= void . addResult result . first WorkerError) `finally` signal sem + addWorker workers w + + let check = do + threadDelay 1000 {-- 1 ms --} + p <- poll producer + e <- isQueueEmpty q + pure $ (isJust p) && e + + submitter <- async $ untilM_ spawn check + + -- early termination + void . async . forever $ do + threadDelay 1000000 {-- 1 second --} + getResult result >>= + either (const $ cancel producer >> cancel submitter) pure + + let waiter = do + (i, _) <- waitBoth producer submitter + waitForWorkers workers + pure i + + (waiter >>= \i -> getResult result >>= pure . second (const $ i)) + `catchAll` (\z -> + failWorkers workers >> + getResult result >>= \w -> + pure (w >> Left (BlowUpError z))) + +data RunError a = + WorkerError a + | BlowUpError SomeException + deriving Show + +renderRunError :: RunError a -> (a -> Text) -> Text +renderRunError r render = + case r of + WorkerError a -> + "Worker failed: " <> render a + BlowUpError e -> + "An unknown exception was caught: " <> T.pack (show e) + +data EarlyTermination = + EarlyTermination deriving (Eq, Show, Typeable) + +instance Exception EarlyTermination + +consume :: forall a b c e . (Queue a -> IO b) -> Int -> (a -> IO (Either e c)) -> IO (Either (RunError e) (b, [c])) +consume pro fork action = flip catchAll (pure . Left . BlowUpError) $ do + q <- newQueue fork -- not fork + producer <- async $ pro q + workers <- (emptyWorkers :: IO (Workers c)) + sem <- new fork + early <- newEmptyMVar + + terminator <- async $ takeMVar early + + let spawn :: IO () + spawn = do + m <- tryReadQueue q + flip (maybe $ return ()) m $ \a -> do + w <- do + M.wait sem + async $ flip finally (signal sem) $ do + r <- action a + case r of + Left e -> + putMVar early e >> + throwM EarlyTermination + Right c -> + pure $! c + addWorker workers w + + let check = do + threadDelay 1000 {-- 1 ms --} + p <- poll producer + e <- isQueueEmpty q + pure $ (isJust p) && e + + submitter <- async $ untilM_ spawn check + + let waiter = runExceptT $ do + (i, _) <- firstT WorkerError . ExceptT $ + waitEitherBoth terminator producer submitter + + ws <- liftIO $ getWorkers workers + ii <- mapM (firstT WorkerError . ExceptT . waitEither terminator) ws + pure $ (i, ii) + + waiter `catch` (\(_ :: EarlyTermination) -> + failWorkers workers >> + (Left . WorkerError) <$> wait terminator) diff --git a/src/Mismi/S3/Internal/Queue.hs b/src/Mismi/S3/Internal/Queue.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Internal/Queue.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE NoImplicitPrelude #-} +module Mismi.S3.Internal.Queue ( + Queue + , newQueue + , readQueue + , tryReadQueue + , writeQueue + , isQueueEmpty + ) where + +import Control.Concurrent.STM.TBQueue (TBQueue, newTBQueue, tryReadTBQueue, readTBQueue, writeTBQueue, isEmptyTBQueue) + +import GHC.Conc (atomically) + +import P + +newtype Queue a = + Queue { + queue :: TBQueue a + } + +newQueue :: Int -> IO (Queue a) +newQueue i = + atomically $ Queue <$> newTBQueue i + +readQueue :: Queue a -> IO a +readQueue = + atomically . readTBQueue . queue + +tryReadQueue :: Queue a -> IO (Maybe a) +tryReadQueue = + atomically . tryReadTBQueue . queue + +writeQueue :: Queue a -> a -> IO () +writeQueue q = + atomically . writeTBQueue (queue q) + +isQueueEmpty :: Queue a -> IO Bool +isQueueEmpty = + atomically . isEmptyTBQueue . queue diff --git a/src/Mismi/S3/Patch/Network.hs b/src/Mismi/S3/Patch/Network.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Patch/Network.hs @@ -0,0 +1,47 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Mismi.S3.Patch.Network ( + chunkedFile + ) where + +import Control.Monad +import Control.Monad.Trans.Resource +import Control.Monad.IO.Class + +import Data.Conduit (ConduitT, yield, bracketP) + +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as LBS + +import Mismi.Amazonka (getFileSize, unsafeChunkedBody) + +import Network.AWS.Data.Body + +import P +import Prelude (toInteger) + +import System.IO + +chunkedFile :: MonadIO m => ChunkSize -> FilePath -> m RqBody +chunkedFile c f = do + n <- getFileSize f + if n > toInteger c + then + return $ unsafeChunkedBody c n (sourceFileChunks c f) + else + liftIO $ (=<<) (return . toBody) $ LBS.readFile f + +-- Uses hGet with a specific buffer size, instead of hGetSome. +sourceFileChunks :: MonadResource m + => ChunkSize + -> FilePath + -> ConduitT () BS.ByteString m () +sourceFileChunks (ChunkSize sz) f = + bracketP (openBinaryFile f ReadMode) hClose go + where + go h = do + bs <- liftIO (BS.hGet h sz) + unless (BS.null bs) $ do + yield bs + go h diff --git a/src/Mismi/S3/Patch/PutObjectACL.hs b/src/Mismi/S3/Patch/PutObjectACL.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Patch/PutObjectACL.hs @@ -0,0 +1,226 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeFamilies #-} + +{-# OPTIONS_GHC -fno-warn-unused-imports #-} +{-# OPTIONS_GHC -fno-warn-unused-binds #-} +{-# OPTIONS_GHC -fno-warn-unused-matches #-} + +-- Derived from AWS service descriptions, licensed under Apache 2.0. + +-- | +-- Module : Network.AWS.S3.PutObjectACL +-- Copyright : (c) 2013-2015 Brendan Hay +-- License : Mozilla Public License, v. 2.0. +-- Maintainer : Brendan Hay +-- Stability : auto-generated +-- Portability : non-portable (GHC extensions) +-- +-- uses the acl subresource to set the access control list (ACL) +-- permissions for an object that already exists in a bucket +-- +-- /See:/ for PutObjectACL. +module Mismi.S3.Patch.PutObjectACL + ( + -- * Creating a Request + putObjectACL + , PutObjectACL + -- * Request Lenses + , poaGrantReadACP + , poaRequestPayer + , poaGrantWriteACP + , poaGrantRead + , poaGrantFullControl + , poaContentMD5 + , poaAccessControlPolicy + , poaGrantWrite + , poaACL + , poaBucket + , poaKey + + -- * Destructuring the Response + , putObjectACLResponse + , PutObjectACLResponse + -- * Response Lenses + , poarsRequestCharged + , poarsResponseStatus + ) where + +import Control.Lens (Lens', lens) +import Network.AWS.Prelude +import Network.AWS.Request +import Network.AWS.Response +import Network.AWS.S3.Types + +-- | /See:/ 'putObjectACL' smart constructor. +data PutObjectACL = PutObjectACL' + { _poaGrantReadACP :: !(Maybe Text) + , _poaRequestPayer :: !(Maybe RequestPayer) + , _poaGrantWriteACP :: !(Maybe Text) + , _poaGrantRead :: !(Maybe Text) + , _poaGrantFullControl :: !(Maybe Text) + , _poaContentMD5 :: !(Maybe Text) + , _poaAccessControlPolicy :: !(Maybe AccessControlPolicy) + , _poaGrantWrite :: !(Maybe Text) + , _poaACL :: !(Maybe ObjectCannedACL) + , _poaBucket :: !BucketName + , _poaKey :: !ObjectKey + } deriving (Eq,Read,Show,Data,Typeable,Generic) + +-- | Creates a value of 'PutObjectACL' with the minimum fields required to make a request. +-- +-- Use one of the following lenses to modify other fields as desired: +-- +-- * 'poaGrantReadACP' +-- +-- * 'poaRequestPayer' +-- +-- * 'poaGrantWriteACP' +-- +-- * 'poaGrantRead' +-- +-- * 'poaGrantFullControl' +-- +-- * 'poaContentMD5' +-- +-- * 'poaAccessControlPolicy' +-- +-- * 'poaGrantWrite' +-- +-- * 'poaACL' +-- +-- * 'poaBucket' +-- +-- * 'poaKey' +putObjectACL + :: BucketName -- ^ 'poaBucket' + -> ObjectKey -- ^ 'poaKey' + -> PutObjectACL +putObjectACL pBucket_ pKey_ = + PutObjectACL' + { _poaGrantReadACP = Nothing + , _poaRequestPayer = Nothing + , _poaGrantWriteACP = Nothing + , _poaGrantRead = Nothing + , _poaGrantFullControl = Nothing + , _poaContentMD5 = Nothing + , _poaAccessControlPolicy = Nothing + , _poaGrantWrite = Nothing + , _poaACL = Nothing + , _poaBucket = pBucket_ + , _poaKey = pKey_ + } + +-- | Allows grantee to read the bucket ACL. +poaGrantReadACP :: Lens' PutObjectACL (Maybe Text) +poaGrantReadACP = lens _poaGrantReadACP (\ s a -> s{_poaGrantReadACP = a}); + +-- | Undocumented member. +poaRequestPayer :: Lens' PutObjectACL (Maybe RequestPayer) +poaRequestPayer = lens _poaRequestPayer (\ s a -> s{_poaRequestPayer = a}); + +-- | Allows grantee to write the ACL for the applicable bucket. +poaGrantWriteACP :: Lens' PutObjectACL (Maybe Text) +poaGrantWriteACP = lens _poaGrantWriteACP (\ s a -> s{_poaGrantWriteACP = a}); + +-- | Allows grantee to list the objects in the bucket. +poaGrantRead :: Lens' PutObjectACL (Maybe Text) +poaGrantRead = lens _poaGrantRead (\ s a -> s{_poaGrantRead = a}); + +-- | Allows grantee the read, write, read ACP, and write ACP permissions on +-- the bucket. +poaGrantFullControl :: Lens' PutObjectACL (Maybe Text) +poaGrantFullControl = lens _poaGrantFullControl (\ s a -> s{_poaGrantFullControl = a}); + +-- | Undocumented member. +poaContentMD5 :: Lens' PutObjectACL (Maybe Text) +poaContentMD5 = lens _poaContentMD5 (\ s a -> s{_poaContentMD5 = a}); + +-- | Undocumented member. +poaAccessControlPolicy :: Lens' PutObjectACL (Maybe AccessControlPolicy) +poaAccessControlPolicy = lens _poaAccessControlPolicy (\ s a -> s{_poaAccessControlPolicy = a}); + +-- | Allows grantee to create, overwrite, and delete any object in the +-- bucket. +poaGrantWrite :: Lens' PutObjectACL (Maybe Text) +poaGrantWrite = lens _poaGrantWrite (\ s a -> s{_poaGrantWrite = a}); + +-- | The canned ACL to apply to the object. +poaACL :: Lens' PutObjectACL (Maybe ObjectCannedACL) +poaACL = lens _poaACL (\ s a -> s{_poaACL = a}); + +-- | Undocumented member. +poaBucket :: Lens' PutObjectACL BucketName +poaBucket = lens _poaBucket (\ s a -> s{_poaBucket = a}); + +-- | Undocumented member. +poaKey :: Lens' PutObjectACL ObjectKey +poaKey = lens _poaKey (\ s a -> s{_poaKey = a}); + +instance AWSRequest PutObjectACL where + type Rs PutObjectACL = PutObjectACLResponse + request = put s3 + response + = receiveEmpty + (\ s h x -> + PutObjectACLResponse' <$> + (h .#? "x-amz-request-charged") <*> + (pure (fromEnum s))) + +instance ToElement PutObjectACL where + toElement + = mkElement + "{http://s3.amazonaws.com/doc/2006-03-01/}AccessControlPolicy" + . + _poaAccessControlPolicy + +instance ToHeaders PutObjectACL where + toHeaders PutObjectACL'{..} + = mconcat + ["x-amz-grant-read-acp" =# _poaGrantReadACP, + "x-amz-request-payer" =# _poaRequestPayer, + "x-amz-grant-write-acp" =# _poaGrantWriteACP, + "x-amz-grant-read" =# _poaGrantRead, + "x-amz-grant-full-control" =# _poaGrantFullControl, + "Content-MD5" =# _poaContentMD5, + "x-amz-grant-write" =# _poaGrantWrite, + "x-amz-acl" =# _poaACL] + +instance ToPath PutObjectACL where + toPath PutObjectACL'{..} + = mconcat ["/", toBS _poaBucket, "/", toBS _poaKey] + +instance ToQuery PutObjectACL where + toQuery = const (mconcat ["acl"]) + +-- | /See:/ 'putObjectACLResponse' smart constructor. +data PutObjectACLResponse = PutObjectACLResponse' + { _poarsRequestCharged :: !(Maybe RequestCharged) + , _poarsResponseStatus :: !Int + } deriving (Eq,Read,Show,Data,Typeable,Generic) + +-- | Creates a value of 'PutObjectACLResponse' with the minimum fields required to make a request. +-- +-- Use one of the following lenses to modify other fields as desired: +-- +-- * 'poarsRequestCharged' +-- +-- * 'poarsResponseStatus' +putObjectACLResponse + :: Int -- ^ 'poarsResponseStatus' + -> PutObjectACLResponse +putObjectACLResponse pResponseStatus_ = + PutObjectACLResponse' + { _poarsRequestCharged = Nothing + , _poarsResponseStatus = pResponseStatus_ + } + +-- | Undocumented member. +poarsRequestCharged :: Lens' PutObjectACLResponse (Maybe RequestCharged) +poarsRequestCharged = lens _poarsRequestCharged (\ s a -> s{_poarsRequestCharged = a}); + +-- | The response status code. +poarsResponseStatus :: Lens' PutObjectACLResponse Int +poarsResponseStatus = lens _poarsResponseStatus (\ s a -> s{_poarsResponseStatus = a}); diff --git a/src/Mismi/S3/Stream.hs b/src/Mismi/S3/Stream.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Stream.hs @@ -0,0 +1,184 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Mismi.S3.Stream ( + sizeRecursively + , read + , list + , liftAddressAndPrefix + , listRecursively + , liftAddress + ) where + +import qualified Control.Exception as CE +import Control.Lens ((.~), (^.), to) +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Reader (ask) +import Control.Monad.Trans.Resource (ResourceT) +import qualified Control.Retry as Retry + +import qualified Data.ByteString as BS +import Data.Conduit (ConduitT, (.|)) +import qualified Data.Conduit as Conduit +import qualified Data.Conduit.List as DC +import Data.IORef (IORef) +import qualified Data.IORef as IORef +import Data.Maybe (maybeToList) + +import Mismi.Amazonka (Env, send, paginate) +import Mismi.Control (AWS, throwOrRetry, handle404) +import Mismi.S3.Core.Data +import Mismi.S3.Internal (f', (+/), bytesRange) + +import qualified Network.AWS as A +import Network.AWS.Data.Body (RsBody (..)) +import Network.AWS.S3 (BucketName (..)) +import Network.AWS.S3 (ListObjectsResponse) +import Network.AWS.S3 (ObjectKey (..)) +import qualified Network.AWS.S3 as A + +import P + + +takeObjectSizes :: Bucket -> ListObjectsResponse -> [Sized Address] +takeObjectSizes b lors = + with (lors ^. A.lorsContents) $ \o -> + let + ObjectKey k = + o ^. A.oKey + + bytes = + Bytes $ fromIntegral (o ^. A.oSize) + -- We shouldn't need this fromIntegral but amazonka incorrectly uses + -- an Int instead of Int64 for sizes, we don't want to propagate this + -- mistake. + -- + -- See https://github.com/brendanhay/amazonka/issues/320 + in + Sized bytes $ Address b (Key k) + +sizeRecursively :: Address -> ConduitT () (Sized Address) AWS () +sizeRecursively (Address b (Key k)) = + let + cmd = + A.listObjects (BucketName $ unBucket b) + & A.loPrefix .~ Just k + in + paginate cmd .| + DC.mapFoldable (takeObjectSizes b) + +countBytes :: + IORef Int64 + -> ConduitT () BS.ByteString (ResourceT IO) () + -> ConduitT () BS.ByteString (ResourceT IO) () +countBytes ref src = + let + loop = do + mbs <- Conduit.await + case mbs of + Nothing -> + pure () + Just bs -> do + liftIO $ IORef.modifyIORef' ref (+ fromIntegral (BS.length bs)) + Conduit.yield bs + loop + in + src .| loop + +readRange :: + Int + -> Int + -> Address +-- -> AWS (ConduitT BS.ByteString (ResourceT IO) (), ResourceT IO ()) + -> AWS (ConduitT () BS.ByteString (ResourceT IO) ()) +readRange start end a = do + result <- + send $ + f' A.getObject a + & A.goRange .~ Just (bytesRange start end) + +-- liftResourceT $ + pure $ + result ^. A.gorsBody . to _streamBody + +readRetry :: + Env + -> Retry.RetryStatus +-- -> IORef (ResourceT IO ()) + -> IORef Int64 + -> Int + -> Address + -> ConduitT () BS.ByteString (ResourceT IO) () +readRetry env status0 startRef end a = do + start <- fmap fromIntegral . liftIO $ IORef.readIORef startRef + + source <- A.runAWS env $ readRange start end a +-- liftIO $ IORef.writeIORef finalizerRef finalizer + + Conduit.catchC (countBytes startRef source) $ \(err :: CE.SomeException) -> do + status <- liftIO $ throwOrRetry 5 err status0 + readRetry env status startRef end a + +--newResumableSource :: Source m a -> m () -> ResumableSource m a +--newResumableSource (Conduit.ConduitM source) final = +-- Conduit.ResumableSource (source Conduit.Done) final + + +-- | WARNING : The returned @ResumableResource@ must be consumed within the +-- @AWS@ monad. Failure to do so can result in run time errors (recv on a bad +-- file descriptor) when the @MonadResouce@ cleans up the socket. +read :: Address -> AWS (Maybe (ConduitT () BS.ByteString (ResourceT IO) ())) +read a = do + env <- ask + startRef <- liftIO $ IORef.newIORef 0 + mend <- + (handle404 . send . f' A.headObject $ a) >>= pure . fmap (Bytes . fromIntegral) . maybe Nothing (^. A.horsContentLength) >>= pure . fmap fromIntegral + + case mend of + Nothing -> + pure Nothing + + Just 0 -> + pure $ Just mempty +-- pure . Just $ newResumableSource mempty (pure ()) + + Just end -> do +-- finalizerRef <- liftIO $ IORef.newIORef (pure ()) + + let + source = + readRetry env Retry.defaultRetryStatus startRef end a + +-- final = do +-- finalizer <- liftIO $ IORef.readIORef finalizerRef +-- finalizer + + pure . Just $ source +-- newResumableSource source final + +list :: Address -> ConduitT () Address AWS () +list a@(Address (Bucket b) (Key k)) = + let + run s = s .| liftAddressAndPrefix a + in + run . paginate $ + A.listObjects (BucketName b) + & A.loPrefix .~ Just ((+/) k) + & A.loDelimiter .~ Just '/' + +liftAddressAndPrefix :: Address -> ConduitT ListObjectsResponse Address AWS () +liftAddressAndPrefix a = + DC.mapFoldable (\r -> + fmap (\o -> + let ObjectKey t = o ^. A.oKey + in a { key = Key t }) (r ^. A.lorsContents) + <> join (forM (r ^. A.lorsCommonPrefixes) $ \cp -> + maybeToList . fmap (\cp' -> a { key = Key cp' }) $ cp ^. A.cpPrefix)) + +listRecursively :: Address -> ConduitT () Address AWS () +listRecursively a@(Address (Bucket bn) (Key k)) = + paginate (A.listObjects (BucketName bn) & A.loPrefix .~ Just k) .| liftAddress a + +liftAddress :: Address -> ConduitT ListObjectsResponse Address AWS () +liftAddress a = + DC.mapFoldable (\r -> (\o -> a { key = Key (let ObjectKey t = o ^. A.oKey in t) }) <$> (r ^. A.lorsContents) ) diff --git a/src/Mismi/S3/Unsafe.hs b/src/Mismi/S3/Unsafe.hs new file mode 100644 --- /dev/null +++ b/src/Mismi/S3/Unsafe.hs @@ -0,0 +1,119 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +module Mismi.S3.Unsafe ( + -- * Upload + hoistUploadError + , upload + , uploadWithMode + , uploadRecursive + , uploadRecursiveWithMode + + -- * Write + , write + , writeWithMode + , liftWriteResult + + -- * Download + , hoistDownloadError + , download + , downloadWithMode + , downloadRecursive + , downloadRecursiveWithMode + ) where + +import Control.Monad.Catch (throwM) +import Control.Monad.Trans.Except (runExceptT) + +import Mismi.Control +import qualified Mismi.S3.Commands as Commands +import Mismi.S3.Core.Data +import Mismi.S3.Data +import Mismi.S3.Internal.Parallel (RunError (..)) + +import P + +import System.FilePath (FilePath) + + +hoistUploadError :: UploadError -> AWS () +hoistUploadError e = + case e of + UploadSourceMissing f -> + throwM $ SourceFileMissing f + UploadDestinationExists a -> + throwM $ DestinationAlreadyExists a + UploadSourceNotDirectory f -> + throwM $ SourceNotDirectory f + MultipartUploadError (WorkerError a) -> + throwM $ a + MultipartUploadError (BlowUpError a) -> + throwM $ a + +upload :: FilePath -> Address -> AWS () +upload f a = + either hoistUploadError pure =<< runExceptT (Commands.upload f a) + +uploadWithMode :: WriteMode -> FilePath -> Address -> AWS () +uploadWithMode w f a = + either hoistUploadError pure =<< runExceptT (Commands.uploadWithMode w f a) + +uploadRecursive :: FilePath -> Address -> Int -> AWS () +uploadRecursive f a i = + either hoistUploadError pure =<< runExceptT (Commands.uploadRecursive f a i) + +uploadRecursiveWithMode :: WriteMode -> FilePath -> Address -> Int -> AWS () +uploadRecursiveWithMode w f a i = + either hoistUploadError pure =<< runExceptT (Commands.uploadRecursiveWithMode w f a i) + + + +liftWriteResult :: WriteResult -> AWS () +liftWriteResult = \case + WriteOk -> + pure () + WriteDestinationExists a -> + throwM $ DestinationAlreadyExists a + +write :: Address -> Text -> AWS () +write a t = + Commands.write a t >>= liftWriteResult + +writeWithMode :: WriteMode -> Address -> Text -> AWS () +writeWithMode m a t = + Commands.writeWithMode m a t >>= liftWriteResult + + +hoistDownloadError :: DownloadError -> AWS () +hoistDownloadError e = + case e of + DownloadSourceMissing a -> + throwM $ SourceMissing DownloadError a + DownloadDestinationExists f -> + throwM $ DestinationFileExists f + DownloadDestinationNotDirectory f -> + throwM $ DestinationNotDirectory f + DownloadInvariant a b -> + throwM $ Invariant (renderDownloadError $ DownloadInvariant a b) + MultipartError (WorkerError a) -> + throwM a + MultipartError (BlowUpError a) -> + throwM a + +download :: Address -> FilePath -> AWS () +download a f = + either hoistDownloadError pure =<< runExceptT (Commands.download a f) + +downloadWithMode :: WriteMode -> Address -> FilePath -> AWS () +downloadWithMode m a f = + either hoistDownloadError pure =<< runExceptT (Commands.downloadWithMode m a f) + + +downloadRecursive :: Address -> FilePath -> AWS () +downloadRecursive a f = + either hoistDownloadError pure =<< runExceptT (Commands.downloadRecursive a f) + +downloadRecursiveWithMode :: WriteMode -> Address -> FilePath -> AWS () +downloadRecursiveWithMode m a f = + either hoistDownloadError pure =<< runExceptT (Commands.downloadRecursiveWithMode m a f) diff --git a/test/Test/IO/Mismi/S3/Commands.hs b/test/Test/IO/Mismi/S3/Commands.hs new file mode 100644 --- /dev/null +++ b/test/Test/IO/Mismi/S3/Commands.hs @@ -0,0 +1,774 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE PackageImports #-} +module Test.IO.Mismi.S3.Commands where + +import Control.Concurrent (threadDelay) +import Control.Monad.Catch (catchAll, throwM) +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Except (runExceptT) + +import "cryptohash" Crypto.Hash + +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as LBS +import Data.Either (isRight, isLeft) +import qualified Data.List as L +import qualified Data.Text as T +import qualified Data.Text.Encoding as T +import qualified Data.Text.IO as T + +import Control.Lens ((^.), to) +import Control.Monad (replicateM_) + +import Hedgehog +import qualified Hedgehog.Gen as Gen +import qualified Hedgehog.Range as Range + +import Mismi.S3 +import qualified Mismi.S3.Amazonka as A +import qualified Mismi.S3.Unsafe as Unsafe + +import P + +import qualified System.Directory as D +import System.FilePath (()) +import qualified System.FilePath as F +import System.IO (withFile, IOMode (..), hFileSize, putStrLn) +import System.IO.Error (userError) + +import Test.Mismi.Amazonka (sendMultipart, newMultipart) +import Test.Mismi.S3 +import qualified Test.Mismi.S3.Core.Gen as Gen + +import Mismi.S3.Internal.Parallel (RunError (..)) + +prop_exists :: Property +prop_exists = + withTests 2 . property . liftAWS $ do + a <- newAddress + lift $ Unsafe.write a "" + result <- lift $ exists a + result === True + +prop_exists_empty :: Property +prop_exists_empty = + withTests 2 . property . liftAWS $ do + a <- newAddress + result <- lift $ exists a + result === False + +prop_exists_failure :: Property +prop_exists_failure = + withTests 2 . property . liftAWS $ do + a <- newAddress + e <- lift $ exists a + e === False + +prop_exists_prefix :: Property +prop_exists_prefix = + withTests 2 . property . liftAWS $ do + k <- forAll $ Gen.genKey + a <- newAddress + lift $ Unsafe.write (withKey (// k) a) "" + e <- lift $ existsPrefix a + e === True + +prop_exists_prefix_missing :: Property +prop_exists_prefix_missing = + withTests 2 . property . liftAWS $ do + a <- newAddress + lift $ Unsafe.write a "" + e <- lift $ existsPrefix a + e === False + +prop_exists_prefix_key :: Property +prop_exists_prefix_key = + withTests 2 . property . liftAWS $ do + a <- newAddress + e <- lift $ existsPrefix a + e === False + +prop_headObject :: Property +prop_headObject = + withTests 2 . property . liftAWS $ do + a <- newAddress + h <- lift $ headObject a + h === Nothing + +prop_getObjects_empty :: Property +prop_getObjects_empty = + withTests 2 . property . liftAWS $ do + a <- newAddress + objs <- lift $ getObjectsRecursively $ a + objs === [] + +prop_getObjectsR :: Property +prop_getObjectsR = -- d p1 p2 = p1 /= p2 ==> testAWS $ do + withTests 2 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.constant 5 15) Gen.alphaNum + items <- fmap toList . forAll $ Gen.set (Range.constant 2 2) Gen.genKey + (p1, p2) <- case items of + p1 : p2 : [] -> + pure (p1, p2) + _ -> + annotate "Invariant generator." >> failure + + root <- newAddress + let + keys = [p1, p2 // p1, p2 // p2] + lift . forM_ keys $ \k -> + Unsafe.write (withKey (// k) root) d + objs <- lift $ getObjectsRecursively root + on (===) L.sort ((^. A.oKey . to A.toText) <$> objs) (unKey . (//) (key root) <$> keys) + + +-- TODO This is incredible slow +prop_pagination_list :: Property +prop_pagination_list = + -- TODO + withTests 0 . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 10 15) Gen.alphaNum + n <- forAll $ Gen.int (Range.linear 1000 1500) + a <- newAddress + lift . forM_ [1..n] $ \n' -> + Unsafe.write (withKey(// Key (m <> T.pack (show n'))) a) "" + r' <- lift $ list a + length r' === n + +prop_size :: Property +prop_size = + withTests 10 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + lift $ Unsafe.write a d + i <- lift $ size a + i === (Just . fromIntegral . BS.length $ T.encodeUtf8 d) + +prop_size_failure :: Property +prop_size_failure = + withTests 2 . property . liftAWS $ do + a <- newAddress + i <- lift $ size a + i === Nothing + +prop_size_recursively :: Property +prop_size_recursively = + withTests 2 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + lift $ Unsafe.write a d + r <- lift $ sizeRecursively (a { key = dirname $ key a }) + r === [Sized (fromIntegral . BS.length $ T.encodeUtf8 d) a] + +-- TODO This is incredible slow +prop_concat :: Property +prop_concat = + withTests 1 . property . liftAWS $ do + a <- newAddress + b <- newAddress + c <- newAddress + f <- newFilePath + let + s = f T.unpack "fred" + d = f T.unpack "down" + bs10k = BS.concat $ L.replicate 10000 "fred" + liftIO $ withFile s WriteMode $ \h -> + replicateM_ 1000 (BS.hPut h bs10k) + lift $ Unsafe.upload s a + lift $ Unsafe.upload s b + + r <- lift . runExceptT $ concatMultipart Fail 1 [a, b] c + () <- either (fail . show . renderConcatError) pure r + + lift $ Unsafe.download c d + s' <- liftIO $ LBS.readFile s + d' <- liftIO $ LBS.readFile d + sha1 (LBS.concat [s', s']) === sha1 d' + +prop_concat_empty_input :: Property +prop_concat_empty_input = + withTests 2 . property . liftAWS $ do + a <- newAddress + r <- lift . runExceptT $ concatMultipart Fail 1 [] a + case r of + Left NoInputFiles -> + success + _ -> + annotate "concat didn't fail correctly" >> failure + +prop_concat_empty_input_files :: Property +prop_concat_empty_input_files = + withTests 2 . property . liftAWS $ do + a <- newAddress + b <- newAddress + lift $ Unsafe.write a "" + r <- lift . runExceptT $ concatMultipart Fail 1 [a] b + case r of + Left NoInputFilesWithData -> + success + _ -> + annotate "concat didn't fail correctly" >> failure + +prop_copy :: Property +prop_copy = + withTests 2 . property . liftAWS $ do + t <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + b <- newAddress + lift $ Unsafe.write a t + lift $ either (fail . T.unpack . renderCopyError) pure =<< runExceptT (copy a b) + a' <- lift $ read a + b' <- lift $ read b + a' === b' + +prop_copy_missing :: Property +prop_copy_missing = + withTests 2 . property . liftAWS $ do + a <- newAddress + r <- lift . runExceptT $ copy a a + case r of + Left (CopySourceMissing b) -> + a === b + _ -> + annotate "Copy didn't fail correctly" >> failure + +prop_copy_overwrite :: Property +prop_copy_overwrite = + withTests 2 . property . liftAWS $ do + t <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + t' <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + b <- newAddress + lift $ Unsafe.write a t + lift $ Unsafe.write b t' + lift $ either (fail . T.unpack . renderCopyError) pure =<< runExceptT (copyWithMode Overwrite a b) + b' <- lift $ read b + b' === Just t + +prop_copy_fail :: Property +prop_copy_fail = + withTests 2 . property . liftAWS $ do + t <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + b <- newAddress + lift $ Unsafe.write a t + lift $ Unsafe.write b t + r <- lift . runExceptT $ copyWithMode Fail a b + case r of + Left (CopyDestinationExists z) -> + b === z + _ -> + annotate "Copy didn't failure correctly" >> failure + +prop_copy_multipart :: Property +prop_copy_multipart = + withTests 1 . property . liftAWS $ do + c <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + m <- forAll $ Gen.text (Range.linear 20 30) Gen.alphaNum + bs <- forAll $ Gen.utf8 (Range.linear 1 100) Gen.unicodeAll + + f <- newFilePath + a' <- newAddress + let + a = withKey (// Key c) a' + b = withKey (// Key m) a' + s = f T.unpack c + d = f T.unpack m + -- create large file to copy + liftIO $ D.createDirectoryIfMissing True f + liftIO $ withFile s WriteMode $ \h -> + replicateM_ 1000 (LBS.hPut h (LBS.fromChunks . return $ (BS.concat . L.replicate 10000 $ bs))) + liftIO . putStrLn $ "Generated file" + + lift $ Unsafe.upload s a + liftIO . putStrLn $ "Uploaded file" + + liftIO . putStrLn $ "Running copy ..." + lift $ either (fail . T.unpack . renderCopyError) pure =<< runExceptT (copy a b) + + liftIO . putStrLn $ "Done copy" + -- compare + lift $ either (fail . show) pure =<< runExceptT (download b d) + liftIO . putStrLn $ "Done download" + + s' <- liftIO $ LBS.readFile s + d' <- liftIO $ LBS.readFile d + sha1 s' === sha1 d' + + + +prop_move :: Property +prop_move = + withTests 2 . property . liftAWS $ do + t <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + s <- newAddress + d <- newAddress + lift $ Unsafe.write s t + lift $ either (fail . T.unpack . renderCopyError) pure =<< runExceptT (move s d) + es <- lift $ exists s + ed <- lift $ exists d + (es, ed) === (False, True) + +prop_upload_mode :: Property +prop_upload_mode = + withTests 2 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + l <- forAll $ genLocalPath + m <- forAll $ Gen.genWriteMode + + p <- newFilePath + a <- newAddress + let + t = p localPath l + liftIO . D.createDirectoryIfMissing True $ F.takeDirectory t + liftIO $ T.writeFile t d + lift $ Unsafe.uploadWithMode m t a + r <- lift $ read a + r === Just d + +prop_upload_overwrite :: Property +prop_upload_overwrite = + withTests 2 . property . liftAWS $ do + d1 <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + d2 <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + l <- forAll $ genLocalPath + + p <- newFilePath + a <- newAddress + let t = p localPath l + liftIO . D.createDirectoryIfMissing True $ F.takeDirectory t + liftIO $ T.writeFile t d1 + lift $ Unsafe.uploadWithMode Fail t a + liftIO $ T.writeFile t d2 + lift $ Unsafe.uploadWithMode Overwrite t a + r <- lift $ read a + r === Just d2 + +prop_upload_fail :: Property +prop_upload_fail = + withTests 2 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + l <- forAll $ genLocalPath + + p <- newFilePath + a <- newAddress + let t = p localPath l + liftIO . D.createDirectoryIfMissing True $ F.takeDirectory t + liftIO $ T.writeFile t d + lift $ Unsafe.uploadWithMode Fail t a + r <- lift . runExceptT $ uploadWithMode Fail t a + case r of + Left (UploadDestinationExists _) -> + success + _ -> + annotate "Upload succeded but should have failed" >> failure + +prop_upload :: Property +prop_upload = + withTests 2 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + l <- forAll $ genLocalPath + p <- newFilePath + a <- newAddress + let t = p localPath l + liftIO . D.createDirectoryIfMissing True $ F.takeDirectory t + liftIO $ T.writeFile t d + lift $ Unsafe.upload t a + r <- lift $ read a + r === Just d + +prop_upload_multipart :: Property +prop_upload_multipart = + withTests 2 . property . liftAWS $ do + bs <- forAll $ Gen.utf8 (Range.linear 0 100) Gen.unicodeAll + l <- forAll $ genLocalPath + + p <- newFilePath + a <- newAddress + let t = p localPath l + liftIO . D.createDirectoryIfMissing True $ F.takeDirectory t + liftIO $ withFile t WriteMode $ \h -> + replicateM_ 1000 (LBS.hPut h (LBS.fromChunks . return $ (BS.concat . L.replicate 10000 $ bs))) + lift $ Unsafe.upload t a + result <- lift $ exists a + assert result + +prop_abort_multipart :: Property +prop_abort_multipart = + withTests 2 . property . liftAWS $ do + + (a, i) <- newMultipart + lift $ sendMultipart "" a 1 i + l <- lift $ listMultiparts (bucket a) + let + abortCheck :: Text -> Bucket -> Int -> A.MultipartUpload -> AWS () + abortCheck x b n u = do + abortMultipart b u + r <- listMultiparts b + unless (n <= (0 :: Int) || L.null (findMultiparts x r)) $ do + liftIO $ threadDelay 500000 + abortCheck x b (n-1) u + + lift . forM_ (findMultiparts i l) $ + abortCheck i (bucket a) 3 + + r <- lift $ listMultiparts (bucket a) + + (L.filter (== Just i) . fmap (^. A.muUploadId) $ l) === [Just i] + + findMultiparts i r === [] + + + +prop_list_multipart :: Property +prop_list_multipart = + withTests 2 . property . liftAWS $ do + (a, i) <- newMultipart + lift $ sendMultipart "" a 1 i + l <- lift $ listMultiparts (bucket a) + multipartExists i l + +prop_list_parts :: Property +prop_list_parts = + withTests 2 . property . liftAWS $ do + (a, i) <- newMultipart + lift $ sendMultipart "" a 1 i + l2 <- lift $ listMultipartParts a i + length l2 === 1 + + + +multipartExists :: Monad m => Text -> [A.MultipartUpload] -> PropertyT m () +multipartExists uploadId multiparts = + L.length (L.filter (findMultipart uploadId) multiparts) === 1 + +findMultiparts :: Text -> [A.MultipartUpload] -> [A.MultipartUpload] +findMultiparts uploadId = + L.filter (findMultipart uploadId) + +findMultipart :: Text -> A.MultipartUpload -> Bool +findMultipart uploadId m = + m ^. A.muUploadId == Just uploadId + + +prop_list :: Property +prop_list = + withTests 2 . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + s <- forAll $ Gen.text (Range.linear 20 30) Gen.alphaNum + a <- newAddress + lift $ Unsafe.write (withKey (// Key m) a) "" + lift $ Unsafe.write (withKey (// (Key s // Key m)) a) "" + r' <- lift $ list a + (Just . Key <$> [m, s <> "/"]) === (removeCommonPrefix a <$> r') + +prop_listObjects :: Property +prop_listObjects = + withTests 2 . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + s <- forAll $ Gen.text (Range.linear 20 30) Gen.alphaNum + a <- newAddress + lift $ Unsafe.write (withKey (// Key m) a) "" + lift $ Unsafe.write (withKey (// (Key s // Key m)) a) "" + (p, k) <- lift $ listObjects a + ([Just . Key $ s <> "/"], [Just $ Key m]) === (removeCommonPrefix a <$> p, removeCommonPrefix a <$> k) + +prop_list_recursively :: Property +prop_list_recursively = + withTests 2 . property . liftAWS $ do + a <- newAddress + lift $ Unsafe.write a "" + r' <- lift $ listRecursively (a { key = dirname $ key a }) + assert $ a `elem` r' + +prop_list_forbidden_bucket :: Property +prop_list_forbidden_bucket = + withTests 1 . property . liftAWS $ do + _ <- lift $ write (Address (Bucket "ambiata-dev-view") (Key "")) "" + success + +prop_download :: Property +prop_download = + withTests 1 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + l <- forAll $ genLocalPath + p <- newFilePath + a <- newAddress + let t = p localPath l + lift $ Unsafe.write a d + r <- lift . runExceptT $ download a t + res <- liftIO $ T.readFile t + + assert $ isRight r + res === d + +prop_download_multipart :: Property +prop_download_multipart = + withTests 1 . property . liftAWS $ do + c <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + m <- forAll $ Gen.text (Range.linear 20 30) Gen.alphaNum + bs <- forAll $ Gen.utf8 (Range.linear 1 100) Gen.unicodeAll + + p <- newFilePath + a <- newAddress + let t = p T.unpack c + let o = p T.unpack m + liftIO . D.createDirectoryIfMissing True $ F.takeDirectory t + liftIO . D.createDirectoryIfMissing True $ F.takeDirectory o + liftIO $ withFile t WriteMode $ \h -> + replicateM_ 1000 (LBS.hPut h (LBS.fromChunks . return $ (BS.concat . L.replicate 10000 $ bs))) + sz <- liftIO . withFile t ReadMode $ hFileSize + lift $ Unsafe.upload t a + + let ten :: Integer = 10 + + r <- lift . runExceptT $ multipartDownload a o (fromInteger sz) ten 100 + b <- liftIO $ LBS.readFile t + + let b' = sha1 b + o' <- liftIO $ LBS.readFile o + let o'' = sha1 o' + + assert $ isRight r + b' === o'' + +prop_write_download_overwrite :: Property +prop_write_download_overwrite = + withTests 2 . property . liftAWS $ do + old <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + new <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + l <- forAll $ genLocalPath + p <- newFilePath + a <- newAddress + let t = p localPath l + lift $ Unsafe.write a old + x <- lift . runExceptT $ downloadWithMode Fail a t + lift $ Unsafe.writeWithMode Overwrite a new + y <- lift . runExceptT $ downloadWithMode Overwrite a t + r <- liftIO $ T.readFile t + + assert $ isRight x + assert $ isRight y + r === new + + +prop_write_download_fail :: Property +prop_write_download_fail = + withTests 2 . property . liftAWS $ do + old <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + new <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + l <- forAll $ genLocalPath + p <- newFilePath + a <- newAddress + let t = p localPath l + lift $ Unsafe.write a old + x <- lift . runExceptT $ downloadWithMode Fail a t + lift $ Unsafe.writeWithMode Overwrite a new + y <- lift . runExceptT $ downloadWithMode Fail a t + + assert $ isRight x + assert $ isLeft y + +prop_delete :: Property +prop_delete = + withTests 2 . property . liftAWS $ do + w <- forAll Gen.genWriteMode + a <- newAddress + lift $ Unsafe.writeWithMode w a "" + x <- lift $ exists a + lift $ delete a + y <- lift $ exists a + assert x + assert $ not y + +prop_delete_empty :: Property +prop_delete_empty = + withTests 2 . property . liftAWS $ do + a <- newAddress + result <- (True <$ lift (delete a)) `catchAll` (const . pure $ False) + assert result + +prop_read_write :: Property +prop_read_write = + withTests 2 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + lift $ Unsafe.write a d + r <- lift $ read a + r === Just d + +prop_write_failure :: Property +prop_write_failure = + withTests 2 . property . liftAWS $ do + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + lift $ Unsafe.write a d + r <- lift $ write a d + r === WriteDestinationExists a + +prop_write_overwrite :: Property +prop_write_overwrite = + withTests 2 . property . liftAWS $ do + let + gen = Gen.text (Range.constant 5 15) Gen.alphaNum + items <- fmap toList . forAll $ Gen.set (Range.constant 2 2) gen + (x, y) <- case items of + x : y : [] -> + pure (x, y) + _ -> + annotate "Invariant generator." >> failure + + a <- newAddress + lift $ Unsafe.writeWithMode Fail a x + lift $ Unsafe.writeWithMode Overwrite a y + r <- lift $ read a + r === Just y + +prop_sync_overwrite :: Property +prop_sync_overwrite = + withTests 2 . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 5 20) Gen.alphaNum + a <- newAddress + b <- newAddress + createSmallFiles a m 10 + x <- lift . runExceptT $ syncWithMode OverwriteSync a b 1 + y <- lift . runExceptT $ syncWithMode OverwriteSync a b 1 + lift . forM_ (files b m 10) $ \e -> + exists e >>= \e' -> + when (e' == False) $ + (throwM . userError $ "Output files do not exist") + + assert $ isRight x + assert $ isRight y + + +prop_sync_fail :: Property +prop_sync_fail = + withTests 2 . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 5 20) Gen.alphaNum + a <- newAddress + b <- newAddress + createSmallFiles a m 1 + x <- lift . runExceptT $ syncWithMode FailSync a b 1 + y <- lift . runExceptT $ syncWithMode FailSync a b 1 + case y of + (Left (SyncError (WorkerError (OutputExists q)))) -> + q === withKey (// Key (m <> "-1")) b + _ -> + failure + assert $ isRight x + + +-- | If the object does not exist, then the behaviour should be invariant with the WriteMode +prop_write_nonexisting :: Property +prop_write_nonexisting = + withTests 2 . property . liftAWS $ do + w <- forAll Gen.genWriteMode + d <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + lift $ Unsafe.writeWithMode w a d + r <- lift $ read a + r === Just d + +prop_write_grant :: Property +prop_write_grant = + withTests 2 . property . liftAWS $ do + t <- forAll $ Gen.text (Range.linear 0 100) Gen.alphaNum + a <- newAddress + lift $ Unsafe.write a t + lift . grantReadAccess a $ ReadGrant "id=e3abd0cceaecbd471c3eaaa47bb722bf199296c5e41c9ee4222877cc91b536fc" + r <- lift $ read a + r === Just t + +prop_read_empty :: Property +prop_read_empty = + withTests 1 . property $ do + k <- forAll Gen.genKey + bucket' <- liftIO testBucket + t <- liftIO . runAWSDefaultRegion . read $ Address bucket' k + t === Nothing + +prop_download_recursive :: Property +prop_download_recursive = + withTests 1 . property . liftAWS $ do + let name1 = "first name" + name2 = "second name" + name3 = "third name" + tmpdir <- newFilePath + addr <- withKey (// Key "top") <$> newAddress + lift $ do + Unsafe.write (withKey (// Key "a") addr) name1 + Unsafe.write (withKey (// Key "b/c") addr) name2 + Unsafe.write (withKey (// Key "c/d/e") addr) name3 + + lift $ either (fail . show) pure =<< runExceptT (downloadRecursive addr tmpdir) + + a <- liftIO $ T.readFile (tmpdir "a") + c <- liftIO $ T.readFile (tmpdir "b" "c") + e <- liftIO $ T.readFile (tmpdir "c" "d" "e") + + a === name1 + c === name2 + e === name3 + +prop_upload_recursive :: Property +prop_upload_recursive = + withTests 1 . property . liftAWS $ do + let name1 = "first name" + name2 = "second name" + name3 = "third name" + tmpdir <- newFilePath + liftIO $ do + D.createDirectoryIfMissing True (tmpdir "b") + D.createDirectoryIfMissing True (tmpdir "c" "d") + + T.writeFile (tmpdir "a") name1 + T.writeFile (tmpdir "b" "c") name2 + T.writeFile (tmpdir "c" "d" "e") name3 + + addr <- withKey (// Key "top") <$> newAddress + + lift $ either (fail . show) pure =<< runExceptT (uploadRecursive tmpdir addr 2) + + a <- lift $ read (withKey (// Key "a") addr) + c <- lift $ read (withKey (// Key "b/c") addr) + e <- lift $ read (withKey (// Key "c/d/e") addr) + + a === Just name1 + c === Just name2 + e === Just name3 + +prop_on_status_ok :: Property +prop_on_status_ok = + withTests 2 . property . liftAWS $ do + let + handler _ = Just 2 + r <- lift $ onStatus_ (1 :: Int) handler (void (exists (Address (Bucket "ambiata-dev-view") (Key "")))) + r === 1 + +prop_on_status_ko :: Property +prop_on_status_ko = + withTests 2 . property . liftAWS $ do + let + handler _ = Just 2 + r <- lift $ onStatus_ (1 :: Int) handler (void (write missingAddress "text")) + r === 2 + +---------- +-- HELPERS +---------- +missingAddress :: Address +missingAddress = + Address (Bucket "ambiata-missing") (Key "m") + +sha1 :: LBS.ByteString -> Digest SHA1 +sha1 = + hashlazy + +tests :: IO Bool +tests = + checkParallel $$(discover) diff --git a/test/Test/IO/Mismi/S3/Control.hs b/test/Test/IO/Mismi/S3/Control.hs new file mode 100644 --- /dev/null +++ b/test/Test/IO/Mismi/S3/Control.hs @@ -0,0 +1,34 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell #-} +module Test.IO.Mismi.S3.Control where + +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Trans.Class (lift) + +import Hedgehog + +import Mismi.S3.Commands +import qualified Mismi.S3.Unsafe as Unsafe + +import P + +import Test.Mismi (liftAWS, runAWSDefaultRegion) +import Test.Mismi.S3 (newAddress) + +prop_finalizer :: Property +prop_finalizer = + withTests 10 . property . liftAWS $ do + a <- newAddress + + r <- liftIO . runAWSDefaultRegion $ do + Unsafe.write a "" + pure $ a + + e <- lift $ exists r + + e === False + +tests :: IO Bool +tests = + checkSequential $$(discover) diff --git a/test/Test/IO/Mismi/S3/Internal.hs b/test/Test/IO/Mismi/S3/Internal.hs new file mode 100644 --- /dev/null +++ b/test/Test/IO/Mismi/S3/Internal.hs @@ -0,0 +1,71 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Test.IO.Mismi.S3.Internal where + +import Control.Monad.Catch (catchIOError) +import Control.Monad.IO.Class (liftIO) + +import Data.Text.IO as T + +import Hedgehog +import qualified Hedgehog.Gen as Gen +import qualified Hedgehog.Range as Range + +import Mismi.S3.Internal + +import P + +import System.Directory (doesFileExist) +import System.IO.Temp (withSystemTempDirectory) + +prop_withFileSafe :: Property +prop_withFileSafe = + property $ do + t <- forAll $ Gen.text (Range.constant 5 15) Gen.alphaNum + (f2e, t2) <- liftIO . withSystemTempPath $ \f -> do + f2 <- withFileSafe f $ \f2 -> do + T.writeFile f2 t + pure f2 + t2 <- T.readFile f + f2e <- doesFileExist f2 + pure (f2e, t2) + (t, f2e) === (t2, False) + + +prop_withFileSafe_empty :: Property +prop_withFileSafe_empty = + property $ do + f2 <- liftIO . withSystemTempPath $ \f -> + flip catchIOError (\_ -> pure Nothing) $ withFileSafe f (pure . Just) + f2 === Nothing + +prop_withFileSafe_error :: Property +prop_withFileSafe_error = + property $ do + result <- liftIO . withSystemTempPath $ \f -> do + flip catchIOError (\_ -> pure ()) . withFileSafe f $ \_ -> fail "" + fmap not $ doesFileExist f + assert result + +prop_withFileSafe_error_exists :: Property +prop_withFileSafe_error_exists = + property $ do + t <- forAll $ Gen.text (Range.constant 5 15) Gen.alphaNum + result <- liftIO . withSystemTempPath $ \f -> do + flip catchIOError (\_ -> pure ()) . withFileSafe f $ \f2 -> T.writeFile f2 t >> fail "" + fmap not $ doesFileExist f + assert result + +withSystemTempPath :: (FilePath -> IO t) -> IO t +withSystemTempPath run = + withSystemTempDirectory "mismi-s3-file-safe" $ \dir -> + run $ dir <> "/file" + + +tests :: IO Bool +tests = + checkSequential $$(discover) +-- TODO +--tests = $forAllProperties $ quickCheckWithResult (stdArgs { maxSuccess = 20 }) diff --git a/test/Test/Mismi/Amazonka.hs b/test/Test/Mismi/Amazonka.hs new file mode 100644 --- /dev/null +++ b/test/Test/Mismi/Amazonka.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +module Test.Mismi.Amazonka ( + sendMultipart + , newMultipart + ) where + +import Control.Monad.Catch +import Control.Monad.Reader (ask) +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Resource +import Control.Monad.Trans.Except (runExceptT) + +import Data.Text as T +import Data.Text.Encoding (encodeUtf8) + +import Hedgehog + +import Mismi.S3 +import qualified Mismi.S3.Amazonka as A +import Mismi.S3.Internal + +import P + +import Test.Mismi.S3 + + +sendMultipart :: Text -> Address -> Int -> Text -> AWS () +sendMultipart t a i ui = do + let req = f' A.uploadPart a i ui (A.toBody $ encodeUtf8 t) + void $ A.send req + +newMultipart :: PropertyT AWS (Address, Text) +newMultipart = do + a <- newAddress + result <- lift $ do + r <- createMultipartUpload a + e <- ask + void $ register (either throwM pure =<< runExceptT (runAWS e $ abortMultipart' a r)) + void $ register (either throwM pure =<< runExceptT (runAWS e $ listRecursively a >>= mapM_ delete >> delete a)) + pure (a, r) + pure result diff --git a/test/Test/Mismi/S3.hs b/test/Test/Mismi/S3.hs new file mode 100644 --- /dev/null +++ b/test/Test/Mismi/S3.hs @@ -0,0 +1,155 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +module Test.Mismi.S3 ( + module X + + , Token (..) + , genToken + + , LocalPath (..) + , genLocalPath + + , testBucket + , createSmallFiles + , files + , newAddress + , newFilePath + , addCleanupFinalizer + , addPrintFinalizer + , addLocalCleanupFinalizer + , addLocalPrintFinalizer + ) where + + +import Control.Monad.Catch +import Control.Monad.Reader (ask) +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Except +import Control.Monad.Trans.Resource +import Control.Monad.IO.Class (MonadIO, liftIO) + +import qualified Data.List as L +import Data.Text as T +import qualified Data.Text.IO as T +import Data.UUID as U +import Data.UUID.V4 as U + +import Hedgehog +import qualified Hedgehog.Gen as Gen +import qualified Hedgehog.Range as Range + +import System.Environment (lookupEnv) +import System.Posix.Env +import System.FilePath +import System.Directory + +import Mismi.Control +import Mismi.S3 + +import P + +import Test.Mismi as X + +data Token = + Token { + unToken :: Text + } deriving (Eq, Show) + +genToken :: Gen Token +genToken = do + let + genA = Gen.text (Range.linear 10 15) Gen.alphaNum + genB = Gen.text (Range.linear 5 10) Gen.alphaNum + + n <- T.pack . show <$> (Gen.int $ Range.linear 0 10000) + c <- genA + m <- genB + sep <- Gen.element ["-", "=", "."] + pure . Token . T.intercalate sep $ [c, m, n] + +data LocalPath = + LocalPath { + localPath :: FilePath + } deriving (Eq, Show) + +genLocalPath :: Gen LocalPath +genLocalPath = do + let + gen = Gen.text (Range.linear 5 10) Gen.alphaNum + + x <- gen + xs <- Gen.list (Range.linear 0 5) gen + pure . LocalPath $ L.intercalate "/" (T.unpack <$> x : xs) + +testBucket :: IO Bucket +testBucket = + Bucket . T.pack . fromMaybe "ambiata-dev-view" <$> getEnv "AWS_TEST_BUCKET" + +createSmallFiles :: Address -> Text -> Int -> PropertyT AWS () +createSmallFiles prefix name n = do + lift . mapM_ (flip write "data") $ files prefix name n + +files :: Address -> Text -> Int -> [Address] +files prefix name n = + fmap (\i -> withKey (// Key (name <> "-" <> (T.pack $ show i))) prefix) [1..n] + +newAddress :: PropertyT AWS Address +newAddress = do + a <- liftIO $ do + t <- Gen.sample genToken + b <- testBucket + u <- T.pack . U.toString <$> U.nextRandom + pure $ Address b (Key . T.intercalate "/" $ ["mismi", u, unToken t]) + lift $ addCleanupFinalizer a + lift $ addPrintFinalizer a + pure $ a + +newFilePath :: PropertyT AWS FilePath +newFilePath = do + p <- liftIO $ do + t <- Gen.sample genToken + d <- getTemporaryDirectory + u <- liftIO $ U.toString <$> U.nextRandom + let p = d <> "/mismi/" <> u <> "-" <> (T.unpack . unToken $ t) + createDirectoryIfMissing True p + pure p + lift$ addLocalCleanupFinalizer p + lift $ addLocalPrintFinalizer p + pure p + +vk :: MonadIO m => Text -> m Bool +vk k = do + m <- liftIO $ lookupEnv (T.unpack k) + return $ maybe False (\v -> v == "1" || v == "true") m + +addCleanupFinalizer :: Address -> AWS () +addCleanupFinalizer a = do + e <- ask + r <- vk "TEST_SKIP_CLEANUP_RESOURCES" + p <- vk "TEST_PRINT_CLEANUP_RESOURCES" + unless r . + void $ register (either throwM pure =<< runExceptT (runAWS e $ + listRecursively a >>= mapM_ delete >> delete a)) + when p . + void $ register (T.putStrLn $ "Cleaning up [" <> addressToText a <> "]") + +addPrintFinalizer :: Address -> AWS () +addPrintFinalizer a = do + r <- vk "TEST_PRINT_PATHS" + when r . + void $ register (T.putStrLn $ "Temporary s3 address [" <> addressToText a <> "]") + +addLocalCleanupFinalizer :: FilePath -> AWS () +addLocalCleanupFinalizer a = do + r <- vk "TEST_SKIP_CLEANUP_RESOURCES" + p <- vk "TEST_PRINT_CLEANUP_RESOURCES" + unless r . + void $ register (removeDirectoryRecursive a) + when p . + void $ register (T.putStrLn $ "Cleaning up [" <> T.pack a <> "]") + +addLocalPrintFinalizer :: FilePath -> AWS () +addLocalPrintFinalizer a = do + r <- vk "TEST_PRINT_PATHS" + when r . + void $ register (T.putStrLn $ "Temporary local filepath [" <> T.pack a <> "]") diff --git a/test/Test/Mismi/S3/Commands.hs b/test/Test/Mismi/S3/Commands.hs new file mode 100644 --- /dev/null +++ b/test/Test/Mismi/S3/Commands.hs @@ -0,0 +1,68 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Test.Mismi.S3.Commands where + +import Control.Lens ((.~)) +import Control.Monad.IO.Class (liftIO) + +import qualified Data.List as DL +import Data.Maybe (mapMaybe) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as DM +import Data.Time.Clock (NominalDiffTime, addUTCTime, getCurrentTime) + +import Hedgehog +import qualified Hedgehog.Gen as Gen +import qualified Hedgehog.Range as Range + +import Mismi.S3.Commands +import qualified Mismi.S3.Amazonka as A + +import P + +import qualified Test.Mismi.S3.Core.Gen as Gen + +-- Positive NominalDiffTime +prop_filter_old :: Property +prop_filter_old = + property $ do + i <- fromInteger <$> forAll (Gen.integral $ Range.linear 1 10000) + n <- liftIO getCurrentTime + let + t = addUTCTime ((-1 * ((60 * 60 * 24 * 7) + i)) :: NominalDiffTime) n + r = filterOld n $ A.multipartUpload & A.muInitiated .~ Just t + r === True + +prop_filter_failure :: Property +prop_filter_failure = + property $ do + n <- liftIO getCurrentTime + let + r = filterOld n $ A.multipartUpload & A.muInitiated .~ Just n + r === False + +prop_chunk_files_by_size :: Property +prop_chunk_files_by_size = + property $ do + maxFilesPerChunk <- forAll (Gen.int $ Range.linear 2 10) + fileCount <- forAll (Gen.int $ Range.linear 10 100) + maxChunkSize <- forAll (Gen.int64 $ Range.linear 1000 10000) + pairs <- forAll (Gen.fileNameSizePairs fileCount) + let + chunks = chunkFilesBySize maxFilesPerChunk maxChunkSize pairs + chunkSizes = DL.map (multiChunkSum (DM.fromList pairs) . DL.map fst) chunks + + DL.filter (> maxChunkSize) chunkSizes === [] + + where + multiChunkSum :: Map FilePath Int64 -> [FilePath] -> Int64 + multiChunkSum _ [] = 0 + multiChunkSum _ [_] = 0 -- Don't care about size of single file chunk. + multiChunkSum sizes xs = + sum $ mapMaybe (\ x -> DM.lookup x sizes) xs + +tests :: IO Bool +tests = + checkSequential $$(discover) diff --git a/test/Test/Mismi/S3/Internal.hs b/test/Test/Mismi/S3/Internal.hs new file mode 100644 --- /dev/null +++ b/test/Test/Mismi/S3/Internal.hs @@ -0,0 +1,35 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Test.Mismi.S3.Internal where + +import Hedgehog +import qualified Hedgehog.Gen as Gen +import qualified Hedgehog.Range as Range + +import Mismi.S3.Internal + +import P + +prop_chunks :: Property +prop_chunks = + property $ do + size <- forAll $ Gen.int (Range.linear 1 10000) + chunk <- forAll $ Gen.int (Range.linear 1 size) + foldl' (+) 0 (snd' <$> calculateChunks size chunk) === size + +prop_chunks_capped :: Property +prop_chunks_capped = + property $ do + size <- forAll $ Gen.int (Range.linear 10 10000) + chunk <- forAll $ Gen.int (Range.linear 1 size) + cap <- forAll $ Gen.int (Range.constant 5 15) + assert $ length (calculateChunksCapped size chunk cap) <= cap + +snd' :: (Int, Int, Int) -> Int +snd' (_, b, _) = b + +tests :: IO Bool +tests = + checkSequential $$(discover) diff --git a/test/Test/Reliability/Mismi/S3/Commands.hs b/test/Test/Reliability/Mismi/S3/Commands.hs new file mode 100644 --- /dev/null +++ b/test/Test/Reliability/Mismi/S3/Commands.hs @@ -0,0 +1,77 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Test.Reliability.Mismi.S3.Commands where + +import Control.Monad (replicateM_) +import Control.Monad.Catch (throwM) +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Except (runExceptT) + +import Data.Either (isRight) +import qualified Data.Text.IO as T + +import Hedgehog +import Hedgehog.Internal.Property (TestLimit (..)) +import qualified Hedgehog.Gen as Gen +import qualified Hedgehog.Range as Range + +import Mismi.S3 +import qualified Mismi.S3.Unsafe as Unsafe + +import P + +import System.FilePath (()) +import System.IO.Error (userError) +import qualified System.IO.Unsafe as Unsafe + +import Test.Mismi.S3 +import qualified Test.Reliability.Reliability as Reliability + +testSize :: TestLimit +testSize = + TestLimit $ + Unsafe.unsafePerformIO Reliability.getMaxSuccess + +prop_sync :: Property +prop_sync = + withTests testSize . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + a <- newAddress + b <- newAddress + i <- Reliability.testSize + createSmallFiles a m i + r <- lift . runExceptT $ syncWithMode OverwriteSync a b 10 + lift . forM_ (files a m i) $ \e -> + exists e >>= \e' -> + when (e' == False) (throwM $ userError "Output files do not exist") + assert $ isRight r + +prop_list :: Property +prop_list = + withTests testSize . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + a <- newAddress + i <- Reliability.testSize + createSmallFiles a m i + lift $ replicateM_ 100 (list a >>= \z -> when (length z /= i) (throwM $ userError "List is not the same as original response")) + + +prop_upload_single :: Property +prop_upload_single = + withTests testSize . property . liftAWS $ do + m <- forAll $ Gen.text (Range.linear 10 20) Gen.alphaNum + a <- newAddress + i <- Reliability.testSize + p <- newFilePath + l <- forAll $ genLocalPath + let f = p localPath l + liftIO $ T.writeFile f "data" + lift . mapM_ (Unsafe.upload f) $ files a m i + + +tests :: IO Bool +tests = + checkSequential $$(discover) diff --git a/test/Test/Reliability/Reliability.hs b/test/Test/Reliability/Reliability.hs new file mode 100644 --- /dev/null +++ b/test/Test/Reliability/Reliability.hs @@ -0,0 +1,29 @@ +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +module Test.Reliability.Reliability ( + testSize + , getMaxSuccess + ) where + +import Control.Monad.IO.Class (MonadIO, liftIO) + +import Hedgehog + +import P +import qualified Prelude as P + +import System.Environment + +testSize :: MonadIO m => PropertyT m Int +testSize = do + view <- liftIO $ lookupEnv "TEST_RELIABILITY_SIZE" + let x = maybe 10 P.read view + pure x + +getMaxSuccess :: IO Int +getMaxSuccess = do + view <- lookupEnv "TEST_RELIABILITY_SUCCESS" + let x = maybe 5 P.read view + pure x diff --git a/test/bench.hs b/test/bench.hs new file mode 100644 --- /dev/null +++ b/test/bench.hs @@ -0,0 +1,85 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE NoImplicitPrelude #-} +{-# LANGUAGE TemplateHaskell #-} + +import Control.Monad.Catch +import Control.Monad.IO.Class (liftIO) +import Control.Monad.Trans.Except (ExceptT, runExceptT, mapExceptT) + +import Criterion.Main + +import qualified Data.Text as T + +import Mismi.S3 +import qualified Mismi.S3.Unsafe as Unsafe + +import P + +import System.IO +import System.IO.Error +import System.IO.Temp + +import Test.Mismi + +import qualified Test.Mismi.S3 as S3 + +createFiles :: Address -> Int -> AWS () +createFiles prefix n = do + mapM_ (flip (Unsafe.writeWithMode Overwrite) "data") $ files prefix n + +createLargeFiles :: Address -> Int -> AWS () +createLargeFiles prefix n = do + withSystemTempFile "file" $ \f h -> do + liftIO $ hSetFileSize h (100 * 1024 * 1024 :: Integer) {-- 100 mb --} + liftIO $ hClose h + mapM_ (Unsafe.uploadWithMode Overwrite f) $ files prefix n + +files :: Address -> Int -> [Address] +files prefix n = + fmap (\i -> withKey (// Key ("file-" <> (T.pack $ show i))) prefix) [1..n] + +run :: AWS a -> IO a +run = + runAWSDefaultRegion + +runE :: ExceptT SyncError AWS a -> IO a +runE t = + either (throwM . userError . T.unpack . renderSyncError) pure =<< runExceptT (mapExceptT runAWSDefaultRegion t) + +main :: IO () +main = do + buck <- S3.testBucket + let a = Address buck (Key "s3-benchmarks/small-foo") + b = Address buck (Key "s3-benchmarks/small-bar") + c = Address buck (Key "s3-benchmarks/large-foo") + d = Address buck (Key "s3-benchmarks/large-bar") + o = Address buck (Key "s3-benchmarks/output") + run $ do + createFiles a 100 + createFiles b 1000 + createLargeFiles c 100 + createLargeFiles d 1000 + runE $ syncWithMode OverwriteSync b o 100 + defaultMain [ + bgroup "sync-small-files" [ + bench "sem-100-100" (nfIO . runE $ syncWithMode OverwriteSync a o 100) + , bench "sem-100-50" (nfIO . runE $ syncWithMode OverwriteSync a o 50) + , bench "sem-100-20" (nfIO . runE $ syncWithMode OverwriteSync a o 20) + , bench "sem-100-10" (nfIO . runE $ syncWithMode OverwriteSync a o 10) + , bench "sem-100-1" (nfIO . runE $ syncWithMode OverwriteSync a o 1) + , bench "sem-1000-1000" (nfIO . runE $ syncWithMode OverwriteSync b o 1000) + , bench "sem-1000-100" (nfIO . runE $ syncWithMode OverwriteSync b o 100) + , bench "sem-1000-10" (nfIO . runE $ syncWithMode OverwriteSync b o 10) + ] + , bgroup "sync-large-files" [ + bench "sem-100-100" (nfIO . runE $ syncWithMode OverwriteSync c o 100) + , bench "sem-100-50" (nfIO . runE $ syncWithMode OverwriteSync c o 50) + , bench "sem-100-20" (nfIO . runE $ syncWithMode OverwriteSync c o 20) + , bench "sem-100-10" (nfIO . runE $ syncWithMode OverwriteSync c o 10) + , bench "sem-100-1" (nfIO . runE $ syncWithMode OverwriteSync c o 1) + , bench "sem-1000-1000" (nfIO . runE $ syncWithMode OverwriteSync d o 1000) + , bench "sem-1000-100" (nfIO . runE $ syncWithMode OverwriteSync d o 100) + , bench "sem-1000-10" (nfIO . runE $ syncWithMode OverwriteSync d o 10) + ] + + ] diff --git a/test/test-io.hs b/test/test-io.hs new file mode 100644 --- /dev/null +++ b/test/test-io.hs @@ -0,0 +1,25 @@ +import Control.Monad (unless) + +import System.Exit (exitFailure) +import System.IO (BufferMode(..), hSetBuffering, stdout, stderr) + +import qualified Test.IO.Mismi.S3.Commands +import qualified Test.IO.Mismi.S3.Control +import qualified Test.IO.Mismi.S3.Internal +import qualified Test.Mismi as Mismi + +main :: IO () +main = do + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering + + results <- sequence =<< Mismi.enableTests "AWS_TEST_S3" [ + Test.IO.Mismi.S3.Internal.tests + ] [ + Test.IO.Mismi.S3.Internal.tests + , Test.IO.Mismi.S3.Commands.tests + , Test.IO.Mismi.S3.Control.tests + ] + + + unless (and results) exitFailure diff --git a/test/test-reliability.hs b/test/test-reliability.hs new file mode 100644 --- /dev/null +++ b/test/test-reliability.hs @@ -0,0 +1,20 @@ +import Control.Monad (unless) + +import System.Exit (exitFailure) +import System.IO (BufferMode(..), hSetBuffering, stdout, stderr) + +import qualified Test.Reliability.Mismi.S3.Commands + +import qualified Test.Mismi as Mismi + +main :: IO () +main = do + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering + + results <- sequence =<< Mismi.enableTests "AWS_TEST_S3" [] [ + Test.Reliability.Mismi.S3.Commands.tests + ] + + + unless (and results) exitFailure diff --git a/test/test.hs b/test/test.hs new file mode 100644 --- /dev/null +++ b/test/test.hs @@ -0,0 +1,19 @@ +import Control.Monad (unless) + +import System.Exit (exitFailure) +import System.IO (BufferMode(..), hSetBuffering, stdout, stderr) + +import qualified Test.Mismi.S3.Commands +import qualified Test.Mismi.S3.Internal + +main :: IO () +main = do + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering + + results <- sequence [ + Test.Mismi.S3.Commands.tests + , Test.Mismi.S3.Internal.tests + ] + + unless (and results) exitFailure