diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,23 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [1.3] - 2023-03-06 + +### Added + +- Upload nars using multiple parts, improving bandwidth speed and parallelization. + +### Changed + +- Bump defaults jobs to 8. +- Improve 401 erros by showing the body of backend response. +- Deploy: improve agent startup and shutdown. + +### Fixed + +- Unblock interrupt signal on darwin (possibly also Linux). +- Deploy: wait for the logs to finish when activating. + ## [1.2] - 2023-01-06 ### Added @@ -15,11 +32,11 @@ - `cachix deploy activate` now by default waits for the agents to be deployed, displays the logs and exists if any deployments fail. If you'd like to keep the old behaviour pass `--async` flag. -## Changed +### Changed - We no longer pin Nix to speed up version bumps of Nix -## Fixed +### Fixed - A number of improvements to stability of the websocket connection used in cachix deploy. diff --git a/cachix-deployment/Main.hs b/cachix-deployment/Main.hs --- a/cachix-deployment/Main.hs +++ b/cachix-deployment/Main.hs @@ -149,8 +149,9 @@ Activate.Failure e -> logDeploymentFailed e Activate.Rollback e -> logDeploymentFailed e Activate.Success -> do + -- NOTE: the activate command uses this message to detect the end of the log Log.streamLine logStream "Successfully activated the deployment." - withLog $ K.logLocM K.InfoS $ K.ls $ "Deployment #" <> deploymentIndex <> " finished" + withLog $ K.logLocM K.InfoS $ K.ls $ "Deployment #" <> deploymentIndex <> " finished." endDeployment activationStatus where @@ -176,6 +177,7 @@ Log.streamLine logStream $ toS $ unwords + -- NOTE: the activate command uses this message to detect the end of the log [ "Failed to activate the deployment.", toS $ displayException e ] diff --git a/cachix.cabal b/cachix.cabal --- a/cachix.cabal +++ b/cachix.cabal @@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix -version: 1.2 +version: 1.3 license: Apache-2.0 license-file: LICENSE copyright: 2018 Domen Kozar @@ -22,6 +22,7 @@ common defaults default-extensions: NoImplicitPrelude + NoImportQualifiedPost DeriveAnyClass DeriveGeneric DerivingVia @@ -60,6 +61,7 @@ Cachix.Client.NixVersion Cachix.Client.OptionsParser Cachix.Client.Push + Cachix.Client.Push.S3 Cachix.Client.PushQueue Cachix.Client.Retry Cachix.Client.Secrets @@ -77,6 +79,7 @@ Cachix.Deploy.StdinProcess Cachix.Deploy.Websocket Cachix.Deploy.WebsocketPong + Data.Conduit.ByteString System.Nix.Base32 hs-source-dirs: src @@ -91,11 +94,13 @@ , cachix-api , concurrent-extra , conduit >=1.3.0 + , conduit-concurrent-map , conduit-extra , conduit-zstd , containers , cookie , cryptonite + , deepseq , dhall >=1.28.0 , directory , ed25519 @@ -118,6 +123,7 @@ , memory , mmorph , netrc + , network-uri , optparse-applicative , pretty-terminal , prettyprinter diff --git a/src/Cachix/Client/Commands.hs b/src/Cachix/Client/Commands.hs --- a/src/Cachix/Client/Commands.hs +++ b/src/Cachix/Client/Commands.hs @@ -37,6 +37,7 @@ import qualified Cachix.Client.WatchStore as WatchStore import qualified Cachix.Types.BinaryCache as BinaryCache import qualified Cachix.Types.SigningKeyCreate as SigningKeyCreate +import qualified Control.Concurrent.Async as Async import Control.Exception.Safe (throwM) import Control.Retry (RetryStatus (rsIterNumber)) import Crypto.Sign.Ed25519 (PublicKey (PublicKey), createKeypair) @@ -55,7 +56,6 @@ import Servant.Conduit () import System.Directory (doesFileExist) import System.IO (hIsTerminalDevice) -import qualified System.Posix.Signals as Signals import qualified System.Process -- TODO: check that token actually authenticates! @@ -106,11 +106,14 @@ notAuthenticatedBinaryCache :: Text -> CachixException notAuthenticatedBinaryCache name = AccessDeniedBinaryCache $ - "Binary cache " <> name <> " doesn't exist or it's private. " <> Config.noAuthTokenError + "Binary cache " <> name <> " doesn't exist or it's private and you need a token: " <> Config.noAuthTokenError -accessDeniedBinaryCache :: Text -> CachixException -accessDeniedBinaryCache name = - AccessDeniedBinaryCache $ "Binary cache " <> name <> " doesn't exist or it's private and you don't have access it" +accessDeniedBinaryCache :: Text -> Maybe ByteString -> CachixException +accessDeniedBinaryCache name maybeBody = + AccessDeniedBinaryCache $ "Binary cache " <> name <> " doesn't exist or you don't have access." <> context maybeBody + where + context Nothing = "" + context (Just body) = " Error: " <> toS body use :: Env -> Text -> InstallationMode.UseOptions -> IO () use env name useOptions = do @@ -119,12 +122,7 @@ -- 1. get cache public key res <- retryAll $ \_ -> (`runClientM` clientenv env) $ API.getCache cachixClient token name case res of - Left err - -- TODO: is checking for the existence of the config file the right thing to do here? - | isErr err status401 && isJust optionalAuthToken -> throwM $ accessDeniedBinaryCache name - | isErr err status401 -> throwM $ notAuthenticatedBinaryCache name - | isErr err status404 -> throwM $ BinaryCacheNotFound $ "Binary cache " <> name <> " does not exist." - | otherwise -> throwM err + Left err -> handleCacheResponse name optionalAuthToken err Right binaryCache -> do () <- escalateAs UnsupportedNixVersion =<< assertNixVersion user <- InstallationMode.getUser @@ -140,6 +138,17 @@ InstallationMode.addBinaryCache (config env) binaryCache useOptions $ InstallationMode.getInstallationMode nixEnv useOptions +handleCacheResponse :: Text -> Maybe Token -> ClientError -> IO a +handleCacheResponse name optionalAuthToken err + | isErr err status401 && isJust optionalAuthToken = throwM $ accessDeniedBinaryCache name (failureResponseBody err) + | isErr err status401 = throwM $ notAuthenticatedBinaryCache name + | isErr err status404 = throwM $ BinaryCacheNotFound $ "Binary cache " <> name <> " does not exist." + | otherwise = throwM err + +failureResponseBody :: ClientError -> Maybe ByteString +failureResponseBody (FailureResponse _ response) = Just $ toS $ responseBody response +failureResponseBody _ = Nothing + push :: Env -> PushArguments -> IO () push env (PushPaths opts name cliPaths) = do hasStdin <- not <$> hIsTerminalDevice stdin @@ -184,17 +193,32 @@ watchExec :: Env -> PushOptions -> Text -> Text -> [Text] -> IO () watchExec env pushOpts name cmd args = withPushParams env pushOpts name $ \pushParams -> do stdoutOriginal <- hDuplicate stdout - let process = (System.Process.proc (toS cmd) (toS <$> args)) {System.Process.std_out = System.Process.UseHandle stdoutOriginal} + let process = + (System.Process.proc (toS cmd) (toS <$> args)) + { System.Process.std_out = System.Process.UseHandle stdoutOriginal + } watch = do hDuplicateTo stderr stdout -- redirect all stdout to stderr WatchStore.startWorkers (pushParamsStore pushParams) (numJobs pushOpts) pushParams - (_, exitCode) <- concurrently watch $ do - (_, _, _, processHandle) <- System.Process.createProcess process - exitCode <- System.Process.waitForProcess processHandle - Signals.raiseSignal Signals.sigINT - return exitCode - exitWith exitCode + Async.withAsync watch $ \watchThread -> + bracketOnError + (getProcessHandle <$> System.Process.createProcess process) + ( \processHandle -> do + -- Terminate the process + uninterruptibleMask_ (System.Process.terminateProcess processHandle) + -- Wait for the process to clean up and exit + _ <- System.Process.waitForProcess processHandle + -- Stop watching the store and wait for all paths to be pushed + Async.cancel watchThread + ) + $ \processHandle -> do + exitCode <- System.Process.waitForProcess processHandle + Async.cancel watchThread + exitWith exitCode + where + getProcessHandle (_, _, _, processHandle) = processHandle + retryText :: RetryStatus -> Text retryText retrystatus = if rsIterNumber retrystatus == 0 @@ -205,10 +229,7 @@ pushStrategy store authToken opts name compressionMethod storePath = PushStrategy { onAlreadyPresent = pass, - on401 = - if isJust authToken - then throwM (accessDeniedBinaryCache name) - else throwM (notAuthenticatedBinaryCache name), + on401 = handleCacheResponse name authToken, onError = throwM, onAttempt = \retrystatus size -> do path <- decodeUtf8With lenientDecode <$> storePathToPath store storePath @@ -230,12 +251,7 @@ let token = fromMaybe (Token "") authToken res <- retryAll $ \_ -> (`runClientM` clientenv env) $ API.getCache cachixClient token name case res of - Left err - -- TODO: is checking for the existence of the config file the right thing to do here? - | isErr err status401 && isJust authToken -> throwM $ accessDeniedBinaryCache name - | isErr err status401 -> throwM $ notAuthenticatedBinaryCache name - | isErr err status404 -> throwM $ BinaryCacheNotFound $ "Binary cache " <> name <> " does not exist." - | otherwise -> throwM err + Left err -> handleCacheResponse name authToken err Right binaryCache -> pure (Just $ BinaryCache.preferredCompressionMethod binaryCache) let compressionMethod = fromMaybe BinaryCache.ZSTD (head $ catMaybes [Cachix.Client.OptionsParser.compressionMethod pushOpts, compressionMethodBackend]) withStore $ \store -> diff --git a/src/Cachix/Client/Env.hs b/src/Cachix/Client/Env.hs --- a/src/Cachix/Client/Env.hs +++ b/src/Cachix/Client/Env.hs @@ -12,7 +12,7 @@ import Cachix.Client.URI (getBaseUrl) import Cachix.Client.Version (cachixVersion) import qualified Hercules.CNix as CNix -import Hercules.CNix.Store (Store, openStore) +import qualified Hercules.CNix.Util as CNix.Util import Network.HTTP.Client ( ManagerSettings, managerModifyRequest, @@ -25,6 +25,7 @@ import Protolude.Conv import Servant.Client.Streaming (ClientEnv, mkClientEnv) import System.Directory (canonicalizePath) +import System.Posix.Signals (getSignalMask, setSignalMask) data Env = Env { cachixoptions :: Config.CachixOptions, @@ -34,7 +35,17 @@ mkEnv :: Options.Flags -> IO Env mkEnv flags = do + signalset <- getSignalMask + -- Initialize the Nix library CNix.init + + -- darwin: restore the signal mask modified by Nix + -- https://github.com/cachix/cachix/issues/501 + setSignalMask signalset + + -- Interrupt Nix before throwing UserInterrupt + CNix.Util.installDefaultSigINTHandler + -- make sure path to the config is passed as absolute to dhall logic canonicalConfigPath <- canonicalizePath (Options.configPath flags) cfg <- Config.getConfig canonicalConfigPath diff --git a/src/Cachix/Client/InstallationMode.hs b/src/Cachix/Client/InstallationMode.hs --- a/src/Cachix/Client/InstallationMode.hs +++ b/src/Cachix/Client/InstallationMode.hs @@ -90,7 +90,7 @@ b) Add the following to your configuration.nix to add your user as trusted and then try again: - nix.trustedUsers = [ "root" "${user}" ]; + nix.settings.trusted-users = [ "root" "${user}" ]; |] addBinaryCache _ _ _ UntrustedRequiresSudo = do diff --git a/src/Cachix/Client/OptionsParser.hs b/src/Cachix/Client/OptionsParser.hs --- a/src/Cachix/Client/OptionsParser.hs +++ b/src/Cachix/Client/OptionsParser.hs @@ -152,7 +152,7 @@ <> short 'j' <> help "Number of threads used for pushing store paths." <> showDefault - <> value 4 + <> value 8 ) <*> switch (long "omit-deriver" <> help "Do not publish which derivations built the store paths.") push = (\opts cache f -> Push $ f opts cache) <$> pushOptions <*> nameArg <*> (pushPaths <|> pushWatchStore) diff --git a/src/Cachix/Client/Push.hs b/src/Cachix/Client/Push.hs --- a/src/Cachix/Client/Push.hs +++ b/src/Cachix/Client/Push.hs @@ -30,11 +30,12 @@ import Cachix.API.Signing (fingerprint, passthroughHashSink, passthroughHashSinkB16, passthroughSizeSink) import qualified Cachix.Client.Config as Config import Cachix.Client.Exception (CachixException (..)) +import qualified Cachix.Client.Push.S3 as Push.S3 import Cachix.Client.Retry (retryAll) import Cachix.Client.Secrets import Cachix.Client.Servant import qualified Cachix.Types.BinaryCache as BinaryCache -import qualified Cachix.Types.ByteStringStreaming +import qualified Cachix.Types.MultipartUpload as Multipart import qualified Cachix.Types.NarInfoCreate as Api import qualified Cachix.Types.NarInfoHash as NarInfoHash import Control.Concurrent.Async (mapConcurrently) @@ -44,7 +45,6 @@ import Control.Retry (RetryStatus) import Crypto.Sign.Ed25519 import qualified Data.ByteString.Base64 as B64 -import Data.Coerce (coerce) import Data.Conduit import qualified Data.Conduit.Lzma as Lzma (compress) import qualified Data.Conduit.Zstd as Zstd (compress) @@ -86,7 +86,7 @@ { -- | Called when a path is already in the cache. onAlreadyPresent :: m r, onAttempt :: RetryStatus -> Int64 -> m (), - on401 :: m r, + on401 :: ClientError -> m r, onError :: ClientError -> m r, onDone :: m r, compressionMethod :: BinaryCache.CompressionMethod, @@ -131,7 +131,7 @@ Right NoContent -> onAlreadyPresent strategy -- we're done as store path is already in the cache Left err | isErr err status404 -> uploadStorePath cache storePath retrystatus - | isErr err status401 -> on401 strategy + | isErr err status401 -> on401 strategy err | otherwise -> onError strategy err getCacheAuthToken :: PushSecret -> Token @@ -139,7 +139,7 @@ getCacheAuthToken (PushSigningKey token _) = token uploadStorePath :: - (MonadMask m, MonadIO m) => + (MonadIO m) => -- | details for pushing to cache PushParams m r -> StorePath -> @@ -151,85 +151,85 @@ -- TODO: storePathText is redundant. Use storePath directly. storePathText <- liftIO $ Store.storePathToPath store storePath let (storeHash, storeSuffix) = splitStorePath $ toS storePathText - name = pushParamsName cache + cacheName = pushParamsName cache + authToken = getCacheAuthToken (pushParamsSecret cache) clientEnv = pushParamsClientEnv cache strategy = pushParamsStrategy cache storePath withCompressor = case compressionMethod strategy of BinaryCache.XZ -> defaultWithXzipCompressorWithLevel (compressionLevel strategy) BinaryCache.ZSTD -> defaultWithZstdCompressorWithLevel (compressionLevel strategy) + cacheClientEnv = + clientEnv + { baseUrl = (baseUrl clientEnv) {baseUrlHost = toS cacheName <> "." <> baseUrlHost (baseUrl clientEnv)} + } + narSizeRef <- liftIO $ newIORef 0 fileSizeRef <- liftIO $ newIORef 0 narHashRef <- liftIO $ newIORef ("" :: ByteString) fileHashRef <- liftIO $ newIORef ("" :: ByteString) + -- This should be a noop because storePathText came from a StorePath normalized <- liftIO $ Store.followLinksToStorePath store $ toS storePathText pathinfo <- liftIO $ Store.queryPathInfo store normalized - -- stream store path as xz compressed nar file - let storePathSize :: Int64 - storePathSize = Store.validPathInfoNarSize pathinfo + let storePathSize = Store.validPathInfoNarSize pathinfo onAttempt strategy retrystatus storePathSize - withCompressor $ \compressor -> do - let stream' = - streamNarIO narEffectsIO (toS storePathText) Data.Conduit.yield - .| passthroughSizeSink narSizeRef - .| passthroughHashSink narHashRef - .| compressor - .| passthroughSizeSink fileSizeRef - .| passthroughHashSinkB16 fileHashRef - let subdomain = - -- TODO: multipart - if (fromIntegral storePathSize / (1024 * 1024) :: Double) > 100 - then "api" - else toS name - newClientEnv = - clientEnv - { baseUrl = (baseUrl clientEnv) {baseUrlHost = subdomain <> "." <> baseUrlHost (baseUrl clientEnv)} - } - (_ :: NoContent) <- liftIO $ do - (`withClientM` newClientEnv) - (API.createNar cachixClient (getCacheAuthToken (pushParamsSecret cache)) name (Just $ compressionMethod strategy) (mapOutput coerce stream')) - escalate - (_ :: NoContent) <- liftIO $ do - narSize <- readIORef narSizeRef - narHash <- ("sha256:" <>) . System.Nix.Base32.encode <$> readIORef narHashRef - narHashNix <- Store.validPathInfoNarHash32 pathinfo - when (narHash /= toS narHashNix) $ throwM $ NarHashMismatch $ toS storePathText <> ": Nar hash mismatch between nix-store --dump and nix db. You can repair db metadata by running as root: $ nix-store --verify --repair --check-contents" - fileHash <- readIORef fileHashRef - fileSize <- readIORef fileSizeRef - deriverPath <- - if omitDeriver strategy - then pure Nothing - else Store.validPathInfoDeriver store pathinfo - deriver <- for deriverPath Store.getStorePathBaseName - referencesPathSet <- Store.validPathInfoReferences store pathinfo - referencesPaths <- sort . fmap toS <$> for referencesPathSet (Store.storePathToPath store) - references <- sort . fmap toS <$> for referencesPathSet Store.getStorePathBaseName - let fp = fingerprint (decodeUtf8With lenientDecode storePathText) narHash narSize referencesPaths - (sig, authToken) = case pushParamsSecret cache of - PushToken token -> (Nothing, token) - PushSigningKey token signKey -> (Just $ toS $ B64.encode $ unSignature $ dsign (signingSecretKey signKey) fp, token) - nic = - Api.NarInfoCreate - { Api.cStoreHash = storeHash, - Api.cStoreSuffix = storeSuffix, - Api.cNarHash = narHash, - Api.cNarSize = narSize, - Api.cFileSize = fileSize, - Api.cFileHash = toS fileHash, - Api.cReferences = references, - Api.cDeriver = maybe "unknown-deriver" (decodeUtf8With lenientDecode) deriver, - Api.cSig = sig - } - escalate $ Api.isNarInfoCreateValid nic - -- Upload narinfo with signature - escalate <=< (`runClientM` clientEnv) $ - API.createNarinfo - cachixClient - authToken - name - (NarInfoHash.NarInfoHash storeHash) - nic - onDone strategy + + withCompressor $ \compressor -> liftIO $ do + uploadResult <- + runConduitRes $ + streamNarIO narEffectsIO (toS storePathText) Data.Conduit.yield + .| passthroughSizeSink narSizeRef + .| passthroughHashSink narHashRef + .| compressor + .| passthroughSizeSink fileSizeRef + .| passthroughHashSinkB16 fileHashRef + .| Push.S3.streamUpload cacheClientEnv authToken cacheName (compressionMethod strategy) + + case uploadResult of + Left err -> throwIO err + Right (narId, uploadId, parts) -> do + narSize <- readIORef narSizeRef + narHash <- ("sha256:" <>) . System.Nix.Base32.encode <$> readIORef narHashRef + narHashNix <- Store.validPathInfoNarHash32 pathinfo + when (narHash /= toS narHashNix) $ throwM $ NarHashMismatch $ toS storePathText <> ": Nar hash mismatch between nix-store --dump and nix db. You can repair db metadata by running as root: $ nix-store --verify --repair --check-contents" + fileHash <- readIORef fileHashRef + fileSize <- readIORef fileSizeRef + deriverPath <- + if omitDeriver strategy + then pure Nothing + else Store.validPathInfoDeriver store pathinfo + deriver <- for deriverPath Store.getStorePathBaseName + referencesPathSet <- Store.validPathInfoReferences store pathinfo + referencesPaths <- sort . fmap toS <$> for referencesPathSet (Store.storePathToPath store) + references <- sort . fmap toS <$> for referencesPathSet Store.getStorePathBaseName + let fp = fingerprint (decodeUtf8With lenientDecode storePathText) narHash narSize referencesPaths + sig = case pushParamsSecret cache of + PushToken _ -> Nothing + PushSigningKey _ signKey -> Just $ toS $ B64.encode $ unSignature $ dsign (signingSecretKey signKey) fp + nic = + Api.NarInfoCreate + { Api.cStoreHash = storeHash, + Api.cStoreSuffix = storeSuffix, + Api.cNarHash = narHash, + Api.cNarSize = narSize, + Api.cFileSize = fileSize, + Api.cFileHash = toS fileHash, + Api.cReferences = references, + Api.cDeriver = maybe "unknown-deriver" (decodeUtf8With lenientDecode) deriver, + Api.cSig = sig + } + escalate $ Api.isNarInfoCreateValid nic + + -- Complete the multipart upload and upload the narinfo + let completeMultipartUploadRequest = + API.completeNarUpload cachixClient authToken cacheName narId uploadId $ + Multipart.CompletedMultipartUpload + { Multipart.parts = parts, + Multipart.narInfoCreate = nic + } + void $ withClientM completeMultipartUploadRequest cacheClientEnv escalate + + onDone strategy -- | Push an entire closure -- diff --git a/src/Cachix/Client/Push/S3.hs b/src/Cachix/Client/Push/S3.hs new file mode 100644 --- /dev/null +++ b/src/Cachix/Client/Push/S3.hs @@ -0,0 +1,116 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE Rank2Types #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Cachix.Client.Push.S3 where + +import qualified Cachix.API as API +import Cachix.API.Error +import Cachix.Client.Retry (retryAll) +import Cachix.Client.Servant (cachixClient) +import Cachix.Types.BinaryCache +import qualified Cachix.Types.MultipartUpload as Multipart +import Conduit (MonadResource, MonadUnliftIO) +import Control.DeepSeq (rwhnf) +import Crypto.Hash (Digest, MD5) +import qualified Crypto.Hash as Crypto +import Data.ByteArray.Encoding (Base (..), convertToBase) +import Data.Conduit (ConduitT, handleC, (.|)) +import Data.Conduit.ByteString (ChunkSize, chunkStream) +import qualified Data.Conduit.Combinators as CC +import Data.Conduit.ConcurrentMap (concurrentMapM_) +import Data.List (lookup) +import qualified Data.List.NonEmpty as NonEmpty +import Data.UUID (UUID) +import Network.HTTP.Client as HTTP +import qualified Network.HTTP.Types.Header as HTTP +import Protolude +import Servant.Auth () +import Servant.Auth.Client +import qualified Servant.Client as Client +import Servant.Client.Streaming +import Servant.Conduit () + +-- | The size of each uploaded part. +-- +-- Common values for S3 are 8MB and 16MB. The minimum is 5MB. +-- +-- Lower values will increase HTTP overhead. Some cloud services impose request body limits. +-- For example, Amazon API Gateway caps out at 10MB. +chunkSize :: ChunkSize +chunkSize = 8 * 1024 * 1024 + +-- | The number of parts to upload concurrently. +-- Speeds up the upload of very large files. +concurrentParts :: Int +concurrentParts = 8 + +-- | The size of the temporary output buffer. +-- +-- Keep this value high to avoid stalling smaller uploads while waiting for a large upload to complete. +-- Each completed upload response is very lightweight. +outputBufferSize :: Int +outputBufferSize = 100 + +streamUpload :: + forall m. + (MonadUnliftIO m, MonadResource m) => + ClientEnv -> + Token -> + Text -> + CompressionMethod -> + ConduitT + ByteString + Void + m + (Either SomeException (UUID, Text, Maybe (NonEmpty Multipart.CompletedPart))) +streamUpload env authToken cacheName compressionMethod = do + Multipart.CreateMultipartUploadResponse {narId, uploadId} <- createMultipartUpload + + handleC (abortMultipartUpload narId uploadId) $ + chunkStream (Just chunkSize) + .| concurrentMapM_ concurrentParts outputBufferSize (uploadPart narId uploadId) + .| completeMultipartUpload narId uploadId + where + manager = Client.manager env + + createMultipartUpload :: ConduitT ByteString Void m Multipart.CreateMultipartUploadResponse + createMultipartUpload = + liftIO $ withClientM createNarRequest env escalate + where + createNarRequest = API.createNar cachixClient authToken cacheName (Just compressionMethod) + + uploadPart :: UUID -> Text -> (Int, ByteString) -> m (Maybe Multipart.CompletedPart) + uploadPart narId uploadId (partNumber, !part) = do + let partHashMD5 :: Digest MD5 = Crypto.hash part + contentMD5 :: ByteString = convertToBase Base64 partHashMD5 + + let uploadNarPartRequest = API.uploadNarPart cachixClient authToken cacheName narId uploadId partNumber (Multipart.SigningData (decodeUtf8 contentMD5)) + Multipart.UploadPartResponse {uploadUrl} <- liftIO $ withClientM uploadNarPartRequest env escalate + + initialRequest <- liftIO $ HTTP.parseUrlThrow (toS uploadUrl) + let request = + initialRequest + { HTTP.method = "PUT", + HTTP.requestBody = HTTP.RequestBodyBS part, + HTTP.requestHeaders = + [ ("Content-Type", "application/octet-stream"), + ("Content-MD5", contentMD5) + ] + } + + response <- liftIO $ retryAll $ \_ -> HTTP.httpNoBody request manager + let eTag = decodeUtf8 <$> lookup HTTP.hETag (HTTP.responseHeaders response) + -- Strictly evaluate each eTag after uploading each part + let !_ = rwhnf eTag + return $ Multipart.CompletedPart partNumber <$> eTag + + completeMultipartUpload narId uploadId = do + parts <- CC.sinkList + return $ Right (narId, uploadId, sequenceA $ NonEmpty.fromList parts) + + abortMultipartUpload narId uploadId err = do + let abortMultipartUploadRequest = API.abortMultipartUpload cachixClient authToken cacheName narId uploadId + _ <- liftIO $ withClientM abortMultipartUploadRequest env escalate + return $ Left err diff --git a/src/Cachix/Client/PushQueue.hs b/src/Cachix/Client/PushQueue.hs --- a/src/Cachix/Client/PushQueue.hs +++ b/src/Cachix/Client/PushQueue.hs @@ -16,14 +16,12 @@ import qualified Cachix.Client.Push as Push import Cachix.Client.Retry (retryAll) import Control.Concurrent.Async -import Control.Concurrent.Extra (once) import Control.Concurrent.STM (TVar, modifyTVar', newTVarIO, readTVar) import qualified Control.Concurrent.STM.Lock as Lock import qualified Control.Concurrent.STM.TBQueue as TBQueue import qualified Data.Set as S import Hercules.CNix.Store (StorePath) import Protolude -import qualified System.Posix.Signals as Signals import qualified System.Systemd.Daemon as Systemd type Queue = TBQueue.TBQueue StorePath @@ -67,12 +65,10 @@ progress <- newTVarIO 0 let pushWorkerState = PushWorkerState newPushQueue progress pushWorker <- async $ replicateConcurrently_ numWorkers $ worker pushParams pushWorkerState - let signalset = - Signals.CatchOnce $ - exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState - void $ Signals.installHandler Signals.sigINT signalset Nothing - void $ Signals.installHandler Signals.sigTERM signalset Nothing - (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker] + let drainQueue = + exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState + + (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker] `finally` drainQueue case eitherException of Left exc | fromException exc == Just StopWorker -> return () Left exc -> throwIO exc @@ -103,15 +99,22 @@ queryLoop (workerState {alreadyQueued = S.union missingStorePathsSet alreadyQueuedSet}) pushqueue pushParams -- | Stop watching the store and push all pending store paths. --- This should only be used once per process. exitOnceQueueIsEmpty :: IO () -> Async () -> Async () -> QueryWorkerState -> PushWorkerState -> IO () -exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState = - join . once $ do - putTextError "Stopped watching /nix/store and waiting for queue to empty ..." - void Systemd.notifyStopping - stopProducerCallback - go +exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState = do + putTextError "Stopped watching /nix/store and waiting for queue to empty ..." + + -- Skip uploading the remaining paths when run in an interruptible mask. + getMaskingState >>= \case + MaskedUninterruptible -> stopWorkers + _ -> do + void Systemd.notifyStopping + stopProducerCallback + go where + stopWorkers = do + cancelWith queryWorker StopWorker + cancelWith pushWorker StopWorker + go = do (isDone, inprogress, queueLength) <- atomically $ do pushQueueLength <- TBQueue.lengthTBQueue $ pushQueue pushWorkerState @@ -122,9 +125,8 @@ return (isDone, inprogress, pushQueueLength) if isDone then do + stopWorkers putTextError "Done." - cancelWith queryWorker StopWorker - cancelWith pushWorker StopWorker else do -- extend shutdown for another 90s void $ Systemd.notify False $ "EXTEND_TIMEOUT_USEC=" <> show (90 * 1000 * 1000 :: Int) diff --git a/src/Cachix/Deploy/ActivateCommand.hs b/src/Cachix/Deploy/ActivateCommand.hs --- a/src/Cachix/Deploy/ActivateCommand.hs +++ b/src/Cachix/Deploy/ActivateCommand.hs @@ -31,6 +31,8 @@ import Servant.Client.Streaming (ClientEnv, runClientM) import Servant.Conduit () import System.Environment (getEnv) +import qualified Text.Megaparsec as Parse +import qualified Text.Megaparsec.Char as Parse run :: Env.Env -> DeployOptions.ActivateOptions -> IO () run env DeployOptions.ActivateOptions {DeployOptions.payloadPath, DeployOptions.agents, DeployOptions.deployAsync} = do @@ -60,6 +62,7 @@ Text.putStr (renderOverview agents) + -- Skip streaming the logs when run with the --async flag when deployAsync exitSuccess Text.putStr "\n\n" @@ -93,10 +96,16 @@ WebSocket.identifier = identifier } - deployment <- - Async.withAsync (printLogsToTerminal options agentName) $ \_ -> - pollDeploymentStatus clientenv (Token agentToken) deploymentID + deployment <- Async.withAsync (printLogsToTerminal options agentName) $ \logThread -> do + deployment <- pollDeploymentStatus clientenv (Token agentToken) deploymentID + -- Wait for all the logs to arrive + let status = Deployment.status deployment + when (status == Deployment.Failed || status == Deployment.Succeeded) $ + void (Async.waitCatch logThread) + + pure deployment + pure (agentName, deployment) pollDeploymentStatus :: ClientEnv -> Token -> UUID -> IO Deployment.Deployment @@ -116,14 +125,28 @@ threadDelay (2 * 1000 * 1000) loop -printLogsToTerminal :: WebSocket.Options -> Text -> IO a +printLogsToTerminal :: WebSocket.Options -> Text -> IO () printLogsToTerminal options agentName = WebSocket.runClientWith options WS.defaultConnectionOptions $ \connection -> - forever $ do + fix $ \loop -> do message <- WS.receiveData connection case Aeson.eitherDecodeStrict' message of - Left error -> Text.putStrLn $ "Error parsing the log message: " <> show error - Right msg -> Text.putStrLn $ unwords [inBrackets agentName, WSS.line msg] + Left error -> do + Text.putStrLn $ "Error parsing the log message: " <> show error + loop + Right msg -> do + putStrLn $ unwords [inBrackets agentName, WSS.line msg] + unless (isDeploymentDone (WSS.line msg)) loop + where + -- Parse each log line looking for the success/failure messages. + -- TODO: figure out a way to avoid this. How can we tell when the log is done? + isDeploymentDone :: Text -> Bool + isDeploymentDone = isRight . Parse.parse logEndMessageParser "" + + logEndMessageParser :: Parse.Parsec Void Text Text + logEndMessageParser = + Parse.string "Successfully activated the deployment" + <|> Parse.string "Failed to activate the deployment" renderOverview :: [(Text, DeployResponse.Details)] -> Text renderOverview agents = diff --git a/src/Cachix/Deploy/Agent.hs b/src/Cachix/Deploy/Agent.hs --- a/src/Cachix/Deploy/Agent.hs +++ b/src/Cachix/Deploy/Agent.hs @@ -34,6 +34,7 @@ import qualified System.Posix.Signals as Signals import qualified System.Posix.Types as Posix import qualified System.Posix.User as Posix.User +import qualified System.Timeout as Timeout type ServiceWebSocket = WebSocket.WebSocket (WSS.Message WSS.AgentCommand) (WSS.Message WSS.BackendCommand) @@ -56,7 +57,7 @@ run :: Config.CachixOptions -> CLI.AgentOptions -> IO () run cachixOptions agentOptions = Log.withLog logOptions $ \withLog -> - logAndExitWithFailure withLog $ + logExceptions withLog $ withAgentLock agentOptions $ do checkUserOwnsHome @@ -83,9 +84,7 @@ channel <- WebSocket.receive websocket shutdownWebsocket <- connectToService websocket - let signalSet = Signals.CatchOnce shutdownWebsocket - void $ Signals.installHandler Signals.sigINT signalSet Nothing - void $ Signals.installHandler Signals.sigTERM signalSet Nothing + installSignalHandlers shutdownWebsocket let agent = Agent @@ -109,20 +108,6 @@ agentName = CLI.name agentOptions profileName = fromMaybe "system" (CLI.profile agentOptions) - logAndExitWithFailure withLog action = withException action logException - where - logException :: SomeException -> IO () - logException someE = - case fromException someE of - Nothing -> pure () - Just ExitSuccess -> pure () - Just e -> - withLog . K.logLocM K.ErrorS . K.ls $ - unlines - [ "The agent encountered an exception:", - toS (displayException e) - ] - verbosity = if Config.verbose cachixOptions then Log.Verbose @@ -135,17 +120,44 @@ environment = "production" } +logExceptions :: Log.WithLog -> IO a -> IO a +logExceptions withLog action = withException action $ \someE -> do + case fromException someE of + Just ExitSuccess -> exitSuccess + Just e -> do + withLog . K.logLocM K.ErrorS . K.ls $ + unlines + [ "The agent encountered an exception:", + toS (displayException e) + ] + exitFailure + Nothing -> exitFailure + lockFilename :: Text -> FilePath lockFilename agentName = "agent-" <> toS agentName -- | Acquire a lock for this agent. Skip this step if we're bootstrapping the agent. withAgentLock :: CLI.AgentOptions -> IO () -> IO () withAgentLock CLI.AgentOptions {bootstrap = True} action = action -withAgentLock CLI.AgentOptions {name} action = do - lock <- Lock.withTryLockAndPid (lockFilename name) action - when (isNothing lock) $ - throwIO (AgentAlreadyRunning name) +withAgentLock CLI.AgentOptions {name} action = tryToAcquireLock 0 + where + tryToAcquireLock :: Int -> IO () + tryToAcquireLock attempts = do + lock <- Lock.withTryLockAndPid (lockFilename name) action + when (isNothing lock) $ + if attempts >= 5 + then throwIO (AgentAlreadyRunning name) + else do + threadDelay (3 * 1000 * 1000) + tryToAcquireLock (attempts + 1) +installSignalHandlers :: IO () -> IO () +installSignalHandlers shutdown = + for_ [Signals.sigINT, Signals.sigTERM] $ \signal -> + Signals.installHandler signal handler Nothing + where + handler = Signals.CatchOnce shutdown + registerAgent :: Agent -> WSS.AgentInformation -> IO () registerAgent Agent {agentState, withLog} agentInformation = do withLog $ K.logLocM K.InfoS "Agent registered." @@ -228,7 +240,9 @@ -- Block until the initial connection is established void $ MVar.readMVar (WebSocket.connection websocket) - once $ MVar.putMVar close () >> Async.wait thread + once $ do + void $ MVar.tryPutMVar close () + void $ Timeout.timeout (5 * 1000 * 1000) (Async.wait thread) -- | Fetch the home directory and verify that the owner matches the current user. -- Throws either 'NoHomeFound' or 'UserDoesNotOwnHome'. diff --git a/src/Cachix/Deploy/Websocket.hs b/src/Cachix/Deploy/Websocket.hs --- a/src/Cachix/Deploy/Websocket.hs +++ b/src/Cachix/Deploy/Websocket.hs @@ -316,7 +316,7 @@ threadDelay (seconds * 1000 * 1000) startGracePeriod :: IO a -> IO (Maybe a) -startGracePeriod = Timeout.timeout (5 * 1000 * 1000) +startGracePeriod = Timeout.timeout (3 * 1000 * 1000) -- | Try to gracefully close the WebSocket. -- diff --git a/src/Data/Conduit/ByteString.hs b/src/Data/Conduit/ByteString.hs new file mode 100644 --- /dev/null +++ b/src/Data/Conduit/ByteString.hs @@ -0,0 +1,71 @@ +module Data.Conduit.ByteString where + +import Conduit (MonadUnliftIO) +import qualified Data.ByteString as BS +import Data.ByteString.Internal (ByteString (PS), mallocByteString) +import Data.ByteString.Unsafe (unsafeIndex) +import Data.Conduit (ConduitT, await, yield, (.|)) +import Foreign.ForeignPtr (ForeignPtr) +import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr) +import Foreign.Storable (pokeByteOff) +import Protolude hiding (hash, yield) + +type ChunkSize = Int + +minimumChunkSize :: ChunkSize +minimumChunkSize = 5 * 1024 * 1024 + +chunkStream :: (MonadUnliftIO m) => Maybe ChunkSize -> ConduitT ByteString (Int, ByteString) m () +chunkStream mChunkSize = do + processAndChunkOutputRaw chunkSize + .| enumerateConduit + where + chunkSize :: ChunkSize + chunkSize = maybe minimumChunkSize (max minimumChunkSize) mChunkSize + + -- count from 1 + enumerateConduit :: (Monad m) => ConduitT a (Int, a) m () + enumerateConduit = loop 1 + where + loop i = await >>= maybe (return ()) (go i) + go i x = do + yield (i, x) + loop (i + 1) + {-# INLINE enumerateConduit #-} + +data S = S (ForeignPtr Word8) (Ptr Word8) {-# UNPACK #-} !Int + +newS :: ChunkSize -> IO S +newS chunkSize = do + fptr <- mallocByteString chunkSize + return (S fptr (unsafeForeignPtrToPtr fptr) 0) + +processChunk :: ChunkSize -> ByteString -> S -> IO ([ByteString], S) +processChunk chunkSize input = + loop identity 0 + where + loop front idxIn s@(S fptr ptr idxOut) + | idxIn >= BS.length input = return (front [], s) + | otherwise = do + pokeByteOff ptr idxOut (unsafeIndex input idxIn) + let idxOut' = idxOut + 1 + idxIn' = idxIn + 1 + if idxOut' >= chunkSize + then do + let bs = PS fptr 0 idxOut' + s' <- newS chunkSize + loop (front . (bs :)) idxIn' s' + else loop front idxIn' (S fptr ptr idxOut') + +processAndChunkOutputRaw :: MonadIO m => ChunkSize -> ConduitT ByteString ByteString m () +processAndChunkOutputRaw chunkSize = + liftIO (newS chunkSize) >>= loop + where + loop s@(S fptr _ len) = do + mbs <- await + case mbs of + Nothing -> yield $ PS fptr 0 len + Just bs -> do + (bss, s') <- liftIO $ processChunk chunkSize bs s + mapM_ yield bss + loop s'