cachix 1.0.1 → 1.1
raw patch · 22 files changed
+1494/−545 lines, 22 filesdep +conduit-zstddep ~dhalldep ~servant-client-core
Dependencies added: conduit-zstd
Dependency ranges changed: dhall, servant-client-core
Files
- CHANGELOG.md +14/−0
- cachix-deployment/Main.hs +205/−52
- cachix.cabal +17/−6
- src/Cachix/Client/Commands.hs +22/−6
- src/Cachix/Client/Config.hs +6/−7
- src/Cachix/Client/Config/Orphans.hs +1/−33
- src/Cachix/Client/InstallationMode.hs +9/−7
- src/Cachix/Client/OptionsParser.hs +29/−10
- src/Cachix/Client/Push.hs +25/−9
- src/Cachix/Client/PushQueue.hs +1/−1
- src/Cachix/Client/URI.hs +132/−21
- src/Cachix/Deploy/Activate.hs +163/−176
- src/Cachix/Deploy/ActivateCommand.hs +5/−4
- src/Cachix/Deploy/Agent.hs +263/−42
- src/Cachix/Deploy/Deployment.hs +21/−0
- src/Cachix/Deploy/Lock.hs +48/−10
- src/Cachix/Deploy/Log.hs +84/−0
- src/Cachix/Deploy/OptionsParser.hs +17/−6
- src/Cachix/Deploy/StdinProcess.hs +6/−7
- src/Cachix/Deploy/Websocket.hs +328/−138
- src/Cachix/Deploy/WebsocketPong.hs +11/−10
- test/URISpec.hs +87/−0
CHANGELOG.md view
@@ -7,6 +7,20 @@ ## Unreleased +# [1.1] - 2022-12-16++### Added ++- Use ZSTD compresion method by default and allow overriding it via `--compression-method` back to XZ. You can also change the default permanently on your binary cache settings page.++- Cachix Deploy got a complete rewrite with correctness in mind and reliablity.++- Cachix Deploy agent now supports --bootstrap that awaits a new agent to spawn and then shuts down.++- Cachix Deploy now supports [home-manager](https://github.com/nix-community/home-manager/pull/3380)++- Generated NixOS module now uses the naming of Nix settings as introduced in NixOS 22.05.+ # [1.0.1] - 2022-09-24 ### Added
cachix-deployment/Main.hs view
@@ -7,31 +7,31 @@ import Cachix.API.Error (escalateAs) import qualified Cachix.API.WebSocketSubprotocol as WSS-import Cachix.Client.Retry+import qualified Cachix.Client.URI as URI import qualified Cachix.Deploy.Activate as Activate+import qualified Cachix.Deploy.Agent as Agent+import Cachix.Deploy.Deployment (Deployment (..)) import qualified Cachix.Deploy.Lock as Lock-import qualified Cachix.Deploy.Websocket as CachixWebsocket-import qualified Cachix.Deploy.Websocket as Input-import Conduit ((.|))+import qualified Cachix.Deploy.Log as Log+import qualified Cachix.Deploy.Websocket as WebSocket import qualified Control.Concurrent.Async as Async import qualified Control.Concurrent.STM.TMQueue as TMQueue+import qualified Control.Exception.Safe as Safe import qualified Data.Aeson as Aeson-import qualified Data.Conduit as Conduit-import qualified Data.Conduit.Combinators as Conduit import qualified Data.Conduit.TQueue as Conduit-import Data.String (String) import Data.Time.Clock (getCurrentTime)-import Data.UUID (UUID) import qualified Data.UUID as UUID+import qualified Data.UUID.V4 as UUID import GHC.IO.Encoding import qualified Katip as K-import Network.HTTP.Simple (RequestHeaders) import qualified Network.WebSockets as WS import Protolude hiding (toS) import Protolude.Conv import System.IO (BufferMode (..), hSetBuffering)-import qualified Wuss +lockFilename :: Text -> FilePath+lockFilename agentName = "deployment-" <> toS agentName+ -- | Activate the new deployment. -- -- If the target profile is already locked by another deployment, exit@@ -42,50 +42,203 @@ hSetBuffering stdout LineBuffering hSetBuffering stderr LineBuffering - input <- escalateAs (FatalError . toS) . Aeson.eitherDecode . toS =<< getContents- let profile = CachixWebsocket.profile . Input.websocketOptions $ input- void . Lock.withTryLock profile $- CachixWebsocket.runForever (CachixWebsocket.websocketOptions input) (handleMessage input)+ deployment@Deployment+ { agentName,+ agentToken,+ host,+ logOptions,+ deploymentDetails+ } <-+ escalateAs (FatalError . toS) . Aeson.eitherDecode . toS =<< getContents -handleMessage :: CachixWebsocket.Input -> ByteString -> (K.KatipContextT IO () -> IO ()) -> WS.Connection -> CachixWebsocket.AgentState -> ByteString -> K.KatipContextT IO ()-handleMessage input payload runKatip connection _ agentToken =- CachixWebsocket.parseMessage payload (handleCommand . WSS.command)+ let deploymentID = WSS.id (deploymentDetails :: WSS.DeploymentDetails)+ let headers = WebSocket.createHeaders agentName agentToken+ let port = fromMaybe (URI.Port 80) (URI.getPortFor (URI.getScheme host))+ let logWebsocketOptions =+ WebSocket.Options+ { WebSocket.host = URI.getHostname host,+ WebSocket.port = port,+ WebSocket.path = "/api/v1/deploy/log/" <> UUID.toText deploymentID,+ WebSocket.useSSL = URI.requiresSSL (URI.getScheme host),+ WebSocket.headers = headers,+ WebSocket.identifier = Agent.agentIdentifier agentName+ }+ let serviceWebsocketOptions =+ WebSocket.Options+ { WebSocket.host = URI.getHostname host,+ WebSocket.port = port,+ WebSocket.path = "/ws-deployment",+ WebSocket.useSSL = URI.requiresSSL (URI.getScheme host),+ WebSocket.headers = headers,+ WebSocket.identifier = Agent.agentIdentifier agentName+ }++ Log.withLog logOptions $ \withLog ->+ void . Lock.withTryLock (lockFilename agentName) $ do+ -- Open a connection to logging stream+ (logQueue, loggingThread) <- runLogStream withLog logWebsocketOptions++ -- Open a connection to Cachix and block until it's ready.+ service <- WebSocket.new withLog serviceWebsocketOptions+ shutdownService <- Agent.connectToService service++ deploy withLog deployment service (Conduit.sinkTMQueue logQueue)+ `finally` do+ withLog $ K.logLocM K.DebugS "Cleaning up websocket connections"+ atomically $ TMQueue.closeTMQueue logQueue+ shutdownService+ Async.wait loggingThread++-- | Run the deployment commands+deploy ::+ -- | Logging context+ Log.WithLog ->+ -- | Deployment information passed from the agent+ Deployment ->+ Agent.ServiceWebSocket ->+ -- | Logging Websocket connection+ Log.LogStream ->+ IO ()+deploy withLog Deployment {..} service logStream = do+ withLog $ K.logLocM K.InfoS $ K.ls $ "Deploying #" <> deploymentIndex <> ": " <> storePath++ activationStatus <- handleAsActivationStatus $+ Activate.withCacheArgs host agentInformation agentToken $ \cacheArgs -> do+ startDeployment Nothing++ Activate.downloadStorePaths logStream deploymentDetails cacheArgs++ -- Read the closure size and report+ --+ -- TODO: query the remote store to get the size before downloading (and+ -- possibly running out of disk space)+ closureSize <- fromRight Nothing <$> Activate.getClosureSize cacheArgs storePath+ when (isJust closureSize) $ startDeployment closureSize++ rollbackAction <- Activate.activate logStream profileName (toS storePath)++ -- Run tests on the new deployment+ testResults <- handleAsFailureReason $ do+ -- Run a basic network test against the backend+ pong <- WebSocket.waitForPong 10 service+ when (isNothing pong) $ throwIO Activate.NetworkTestFailure++ -- Run the optional rollback script+ for (WSS.rollbackScript deploymentDetails) $ \rollbackScript -> do+ Log.streamLine logStream "Running rollback script."+ rollbackScriptResult <- Safe.tryIO $ Activate.runShellWithExitCode logStream (toS rollbackScript) []+ case rollbackScriptResult of+ Right ExitSuccess -> pure ()+ Right (ExitFailure _) -> throwIO Activate.RollbackScriptExitFailure+ Left e -> throwIO (Activate.RollbackScriptUnexpectedError e)++ -- Roll back if any of the tests have failed+ case testResults of+ Right _ -> pure Activate.Success+ Left testErrors ->+ case rollbackAction of+ Just rollback -> do+ Log.streamLine logStream "Deployment failed, rolling back ..."+ rollback+ throwIO (Activate.Rollback testErrors)+ Nothing -> do+ Log.streamLine logStream "Skipping rollback as this is the first deployment."+ throwIO (Activate.Failure testErrors)++ case activationStatus of+ Activate.Failure e -> logDeploymentFailed e+ Activate.Rollback e -> logDeploymentFailed e+ Activate.Success -> do+ Log.streamLine logStream "Successfully activated the deployment."+ withLog $ K.logLocM K.InfoS $ K.ls $ "Deployment #" <> deploymentIndex <> " finished"++ endDeployment activationStatus where- deploymentDetails = CachixWebsocket.deploymentDetails input- options = CachixWebsocket.websocketOptions input+ storePath = WSS.storePath deploymentDetails+ deploymentID = WSS.id (deploymentDetails :: WSS.DeploymentDetails)+ deploymentIndex = show (WSS.index deploymentDetails) - handleCommand :: WSS.BackendCommand -> K.KatipContextT IO ()- handleCommand (WSS.Deployment _) =- K.logLocM K.ErrorS "cachix-deployment should have never gotten a deployment command directly."- handleCommand (WSS.AgentRegistered agentInformation) = do- queue <- liftIO $ atomically TMQueue.newTMQueue- let deploymentID = WSS.id (deploymentDetails :: WSS.DeploymentDetails)- streamingThread = runLogStreaming (toS $ CachixWebsocket.host options) (CachixWebsocket.headers options agentToken) queue deploymentID- activateThread =- runKatip- (Activate.activate options connection (Conduit.sinkTMQueue queue) deploymentDetails agentInformation agentToken)- `finally` atomically (TMQueue.closeTMQueue queue)- liftIO $ do- Async.concurrently_ streamingThread activateThread- -- TODO: move this into a `finally` after refactoring- WS.sendClose connection ("Closing." :: ByteString)- throwIO ExitSuccess+ handleAsActivationStatus :: IO Activate.Status -> IO Activate.Status+ handleAsActivationStatus action =+ action+ `Safe.catches` [ Safe.Handler (\(e :: Activate.Status) -> pure e),+ Safe.Handler (\(e :: SomeException) -> pure $ Activate.Failure (Activate.UnexpectedError e))+ ] - runLogStreaming :: String -> RequestHeaders -> TMQueue.TMQueue ByteString -> UUID -> IO ()- runLogStreaming host headers queue deploymentID = do- let path = "/api/v1/deploy/log/" <> UUID.toText deploymentID- retryAllWithLogging endlessRetryPolicy (CachixWebsocket.logger runKatip) $ do- liftIO $- Wuss.runSecureClientWith host 443 (toS path) WS.defaultConnectionOptions headers $- \conn ->- Conduit.runConduit $- Conduit.sourceTMQueue queue- .| Conduit.linesUnboundedAscii- -- TODO: prepend katip-like format to each line- -- .| (if CachixWebsocket.isVerbose options then Conduit.print else mempty)- .| sendLog conn+ handleAsFailureReason :: IO a -> IO (Either Activate.FailureReason a)+ handleAsFailureReason action =+ fmap Right action+ `Safe.catches` [ Safe.Handler (\(e :: Activate.FailureReason) -> pure (Left e)),+ Safe.Handler (\(e :: SomeException) -> pure $ Left (Activate.UnexpectedError e))+ ] -sendLog :: WS.Connection -> Conduit.ConduitT ByteString Conduit.Void IO ()-sendLog connection = Conduit.mapM_ $ \bs -> do- now <- getCurrentTime- WS.sendTextData connection $ Aeson.encode $ WSS.Log {WSS.line = toS bs, WSS.time = now}+ logDeploymentFailed e = do+ Log.streamLine logStream $+ toS $+ unwords+ [ "Failed to activate the deployment.",+ toS $ displayException e+ ]+ withLog $ K.logLocM K.InfoS $ K.ls $ "Deploying #" <> deploymentIndex <> " failed."++ startDeployment :: Maybe Int64 -> IO ()+ startDeployment closureSize = do+ now <- getCurrentTime+ msg <-+ createMessage $+ WSS.DeploymentStarted+ { WSS.id = deploymentID,+ WSS.time = now,+ WSS.closureSize = closureSize+ }+ WebSocket.send service (WebSocket.DataMessage msg)++ endDeployment :: Activate.Status -> IO ()+ endDeployment status = do+ let hasSucceeded =+ case status of+ Activate.Success -> True+ _ -> False+ now <- getCurrentTime+ msg <-+ createMessage $+ WSS.DeploymentFinished+ { WSS.id = deploymentID,+ WSS.time = now,+ WSS.hasSucceeded = hasSucceeded+ }+ WebSocket.send service (WebSocket.DataMessage msg)++ createMessage :: WSS.AgentCommand -> IO (WSS.Message WSS.AgentCommand)+ createMessage command = do+ uuid <- UUID.nextRandom+ return $+ WSS.Message+ { WSS.method = method,+ WSS.command = command,+ WSS.id = uuid,+ WSS.agent = Just $ WSS.id (agentInformation :: WSS.AgentInformation)+ }+ where+ -- TODO: move to WSS+ method = case command of+ WSS.DeploymentStarted {} -> "DeploymentStarted"+ WSS.DeploymentFinished {} -> "DeploymentFinished"++-- Log++-- TODO: prepend katip-like format to each line+-- TODO: Re-use the WebSocket module here (without ping?)+runLogStream ::+ Log.WithLog ->+ WebSocket.Options ->+ -- | Returns a queue for writing messages and the thread handle+ IO (TMQueue.TMQueue ByteString, Async.Async ())+runLogStream withLog options = do+ queue <- TMQueue.newTMQueueIO+ thread <- Async.async $+ WebSocket.reconnectWithLog withLog $+ WebSocket.runClientWith options WS.defaultConnectionOptions $ \connection ->+ Log.streamLog withLog connection queue+ `finally` WebSocket.waitForGracefulShutdown connection+ return (queue, thread)
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix-version: 1.0.1+version: 1.1 license: Apache-2.0 license-file: LICENSE copyright: 2018 Domen Kozar@@ -25,8 +25,11 @@ DeriveAnyClass DeriveGeneric DerivingVia+ LambdaCase NamedFieldPuns OverloadedStrings+ RecordWildCards+ ScopedTypeVariables ghc-options: -Wall -Wcompat -Wincomplete-record-updates@@ -67,7 +70,9 @@ Cachix.Deploy.Activate Cachix.Deploy.ActivateCommand Cachix.Deploy.Agent+ Cachix.Deploy.Deployment Cachix.Deploy.Lock+ Cachix.Deploy.Log Cachix.Deploy.OptionsParser Cachix.Deploy.StdinProcess Cachix.Deploy.Websocket@@ -87,6 +92,7 @@ , concurrent-extra , conduit >=1.3.0 , conduit-extra+ , conduit-zstd , containers , cookie , cryptonite@@ -126,6 +132,7 @@ , servant-client-core >=0.16 , servant-conduit , stm+ , stm-chans , stm-conduit , systemd , temporary@@ -141,14 +148,12 @@ , wuss -- These versions should match those supported by hercules-ci-cnix-store- pkgconfig-depends:- nix-main (==2.4 || >2.4) && <2.11,- nix-store (==2.4 || >2.4) && <2.11+ pkgconfig-depends: nix-main >=2.4 && <2.11, nix-store >=2.4 && <2.11 executable cachix import: defaults main-is: Main.hs- build-tool-depends: hspec-discover:hspec-discover -any+ build-tool-depends: hspec-discover:hspec-discover ghc-options: -threaded -rtsopts -with-rtsopts=-maxN8 hs-source-dirs: cachix other-modules: Paths_cachix@@ -162,7 +167,7 @@ executable .cachix-deployment import: defaults main-is: Main.hs- build-tool-depends: hspec-discover:hspec-discover -any+ build-tool-depends: hspec-discover:hspec-discover ghc-options: -g -threaded -rtsopts -with-rtsopts=-maxN8 hs-source-dirs: cachix-deployment other-modules: Paths_cachix@@ -198,14 +203,20 @@ NixConfSpec NixVersionSpec Spec+ URISpec build-depends:+ , aeson , base >=4.7 && <5+ , bytestring , cachix , cachix-api+ , dhall , directory+ , extra , here , hspec , protolude , servant-auth-client+ , servant-client-core , temporary
src/Cachix/Client/Commands.hs view
@@ -35,6 +35,7 @@ ) import Cachix.Client.Servant import qualified Cachix.Client.WatchStore as WatchStore+import qualified Cachix.Types.BinaryCache as BinaryCache import qualified Cachix.Types.SigningKeyCreate as SigningKeyCreate import Control.Exception.Safe (throwM) import Control.Retry (RetryStatus (rsIterNumber))@@ -122,7 +123,7 @@ -- 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."+ | isErr err status404 -> throwM $ BinaryCacheNotFound $ "Binary cache " <> name <> " does not exist." | otherwise -> throwM err Right binaryCache -> do () <- escalateAs UnsupportedNixVersion =<< assertNixVersion@@ -201,8 +202,8 @@ then "" else "(retry #" <> show (rsIterNumber retrystatus) <> ") " -pushStrategy :: Store -> Maybe Token -> PushOptions -> Text -> StorePath -> PushStrategy IO ()-pushStrategy store authToken opts name storePath =+pushStrategy :: Store -> Maybe Token -> PushOptions -> Text -> BinaryCache.CompressionMethod -> StorePath -> PushStrategy IO ()+pushStrategy store authToken opts name compressionMethod storePath = PushStrategy { onAlreadyPresent = pass, on401 =@@ -213,9 +214,10 @@ onAttempt = \retrystatus size -> do path <- decodeUtf8With lenientDecode <$> storePathToPath store storePath -- we append newline instead of putStrLn due to https://github.com/haskell/text/issues/242- putStr $ retryText retrystatus <> "compressing and pushing " <> path <> " (" <> humanSize (fromIntegral size) <> ")\n",+ putStr $ retryText retrystatus <> "compressing using " <> T.toLower (show compressionMethod) <> " and pushing " <> path <> " (" <> humanSize (fromIntegral size) <> ")\n", onDone = pass,- withXzipCompressor = defaultWithXzipCompressorWithLevel (compressionLevel opts),+ Cachix.Client.Push.compressionMethod = compressionMethod,+ Cachix.Client.Push.compressionLevel = Cachix.Client.OptionsParser.compressionLevel opts, Cachix.Client.Push.omitDeriver = Cachix.Client.OptionsParser.omitDeriver opts } @@ -224,11 +226,25 @@ pushSecret <- findPushSecret (config env) name store <- wait (storeAsync env) authToken <- Config.getAuthTokenMaybe (config env)+ compressionMethodBackend <- case pushSecret of+ PushSigningKey {} -> pure Nothing+ PushToken {} -> do+ 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+ Right binaryCache -> pure (Just $ BinaryCache.preferredCompressionMethod binaryCache)+ let compressionMethod = fromMaybe BinaryCache.ZSTD (head $ catMaybes [Cachix.Client.OptionsParser.compressionMethod pushOpts, compressionMethodBackend]) return $ PushParams { pushParamsName = name, pushParamsSecret = pushSecret, pushParamsClientEnv = clientenv env,- pushParamsStrategy = pushStrategy store authToken pushOpts name,+ pushParamsStrategy = pushStrategy store authToken pushOpts name compressionMethod, pushParamsStore = store }
src/Cachix/Client/Config.hs view
@@ -51,10 +51,9 @@ setFileMode, unionFileModes, )-import qualified URI.ByteString as URI data CachixOptions = CachixOptions- { host :: URI.URIRef URI.Absolute,+ { host :: URI, configPath :: ConfigPath, verbose :: Bool }@@ -62,7 +61,7 @@ data Config = Config { authToken :: Token,- hostname :: URI.URIRef URI.Absolute,+ hostname :: URI, binaryCaches :: [BinaryCacheConfig] } deriving (Show, Generic, Dhall.ToDhall, Dhall.FromDhall)@@ -180,7 +179,7 @@ deriving (Show) data SetCommand- = SetHostname (URI.URIRef URI.Absolute)+ = SetHostname URI deriving (Show) run :: CachixOptions -> Command -> IO ()@@ -191,7 +190,7 @@ getConfigOption configPath cmd = do config <- getConfig configPath case cmd of- GetHostname -> putStrLn $ URI.serializeURIRef' (hostname config)+ GetHostname -> putStrLn (URI.serialize (hostname config) :: Text) setConfigOption :: ConfigPath -> SetCommand -> IO () setConfigOption configPath (SetHostname hostname) = do@@ -235,6 +234,6 @@ SetHostname <$> Opt.argument uriOption (Opt.metavar "HOSTNAME") -uriOption :: Opt.ReadM (URI.URIRef URI.Absolute)+uriOption :: Opt.ReadM URI uriOption = Opt.eitherReader $ \s ->- first show $ URI.parseURI URI.strictURIParserOptions (toS s)+ first show $ URI.parseURI (toS s)
src/Cachix/Client/Config/Orphans.hs view
@@ -1,18 +1,12 @@ {-# OPTIONS_GHC -Wno-orphans #-}-{-# LANGUAGE FlexibleInstances #-} -module Cachix.Client.Config.Orphans- (- )-where+module Cachix.Client.Config.Orphans where import qualified Dhall import qualified Dhall.Core import Protolude hiding (toS) import Protolude.Conv import Servant.Auth.Client-import qualified URI.ByteString as URI-import Data.Either.Validation ( Validation(Failure, Success) ) instance Dhall.FromDhall Token where autoWith _ = Dhall.strictText {Dhall.extract = ex}@@ -25,29 +19,3 @@ { Dhall.embed = Dhall.Core.TextLit . Dhall.Core.Chunks [] . toS . getToken, Dhall.declared = Dhall.Core.Text }--instance Dhall.FromDhall (URI.URIRef URI.Absolute) where- autoWith opts =- Dhall.Decoder extract expected- where- textDecoder :: Dhall.Decoder Text- textDecoder = Dhall.autoWith opts-- extract expression =- case Dhall.extract textDecoder expression of- Success x -> case URI.parseURI URI.strictURIParserOptions (toS x) of- Left exception -> Dhall.extractError (show exception)- Right path -> Success path- Failure e -> Failure e-- expected = Dhall.expected textDecoder--instance Dhall.ToDhall (URI.URIRef URI.Absolute) where- injectWith opts = Dhall.Encoder embed declared- where- textEncoder :: Dhall.Encoder Text- textEncoder = Dhall.injectWith opts-- embed uri = Dhall.embed textEncoder $ toS (URI.serializeURIRef' uri)-- declared = Dhall.Core.Text
src/Cachix/Client/InstallationMode.hs view
@@ -194,7 +194,7 @@ imports = lib.mapAttrsToList toImport (lib.filterAttrs filterCaches (builtins.readDir folder)); in { inherit imports;- nix.binaryCaches = ["https://cache.nixos.org/"];+ nix.settings.substituters = ["https://cache.nixos.org/"]; } |] cacheModule :: Text@@ -202,12 +202,14 @@ [i| { nix = {- binaryCaches = [- "${BinaryCache.uri bc}"- ];- binaryCachePublicKeys = [- ${T.intercalate " " (map (\s -> "\"" <> s <> "\"") (BinaryCache.publicSigningKeys bc))}- ];+ settings = {+ substituters = [+ "${BinaryCache.uri bc}"+ ];+ trusted-public-keys = [+ ${T.intercalate " " (map (\s -> "\"" <> s <> "\"") (BinaryCache.publicSigningKeys bc))}+ ];+ }; }; } |]
src/Cachix/Client/OptionsParser.hs view
@@ -12,16 +12,19 @@ import qualified Cachix.Client.Config as Config import qualified Cachix.Client.InstallationMode as InstallationMode+import Cachix.Client.URI (URI) import qualified Cachix.Client.URI as URI import qualified Cachix.Deploy.OptionsParser as DeployOptions+import qualified Cachix.Types.BinaryCache as BinaryCache+import qualified Data.Text as T import Options.Applicative-import Protolude hiding (toS)+import Protolude hiding (option, toS) import Protolude.Conv-import qualified URI.ByteString as URI+import qualified Prelude data Flags = Flags { configPath :: Config.ConfigPath,- hostname :: Maybe (URI.URIRef URI.Absolute),+ hostname :: Maybe URI, verbose :: Bool } @@ -61,11 +64,11 @@ pure Flags {hostname, configPath, verbose} where- defaultHostname = toS (URI.serializeURIRef' URI.defaultCachixURI)+ defaultHostname = URI.serialize URI.defaultCachixURI -uriOption :: ReadM (URI.URIRef URI.Absolute)+uriOption :: ReadM URI uriOption = eitherReader $ \s ->- first show $ URI.parseURI URI.strictURIParserOptions $ toS s+ first show $ URI.parseURI (toS s) type BinaryCacheName = Text @@ -88,6 +91,7 @@ data PushOptions = PushOptions { compressionLevel :: Int,+ compressionMethod :: Maybe BinaryCache.CompressionMethod, numJobs :: Int, omitDeriver :: Bool }@@ -112,7 +116,14 @@ authTokenArg = strArgument (metavar "AUTH-TOKEN") generateKeypair = GenerateKeypair <$> nameArg validatedLevel l =- l <$ unless (l `elem` [0 .. 9]) (readerError $ "value " <> show l <> " not in expected range: [0..9]")+ l <$ unless (l `elem` [0 .. 16]) (readerError $ "value " <> show l <> " not in expected range: [0..16]")+ validatedMethod :: Prelude.String -> Either Prelude.String (Maybe BinaryCache.CompressionMethod)+ validatedMethod method =+ if method `elem` ["xz", "zstd"]+ then case readEither (T.toUpper (toS method)) of+ Right a -> Right $ Just a+ Left b -> Left $ toS b+ else Left $ "Compression method " <> show method <> " not expected. Use xz or zstd." pushOptions :: Parser PushOptions pushOptions = PushOptions@@ -120,12 +131,20 @@ (auto >>= validatedLevel) ( long "compression-level" <> short 'c'- <> metavar "[0..9]"+ <> metavar "[0..16]" <> help- "The compression level for XZ compression.\- \ Take compressor *and* decompressor memory usage into account before using [7..9]!"+ "The compression level for XZ compression between 0-9 and ZSTD 0-16." <> showDefault <> value 2+ )+ <*> option+ (eitherReader validatedMethod)+ ( long "compression-method"+ <> short 'm'+ <> metavar "xz | zstd"+ <> help+ "The compression method, either xz or zstd. Defaults to zstd."+ <> value Nothing ) <*> option auto
src/Cachix/Client/Push.hs view
@@ -14,6 +14,8 @@ PushStrategy (..), defaultWithXzipCompressor, defaultWithXzipCompressorWithLevel,+ defaultWithZstdCompressor,+ defaultWithZstdCompressorWithLevel, findPushSecret, -- * Pushing a closure of store paths@@ -31,6 +33,7 @@ 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.NarInfoCreate as Api import qualified Cachix.Types.NarInfoHash as NarInfoHash@@ -43,8 +46,9 @@ import qualified Data.ByteString.Base64 as B64 import Data.Coerce (coerce) import Data.Conduit-import Data.Conduit.Lzma (compress)+import qualified Data.Conduit.Lzma as Lzma (compress) import Data.Conduit.Process hiding (env)+import qualified Data.Conduit.Zstd as Zstd (compress) import Data.IORef import qualified Data.Set as Set import Data.String.Here@@ -85,16 +89,23 @@ on401 :: m r, onError :: ClientError -> m r, onDone :: m r,- withXzipCompressor :: forall a. (ConduitM ByteString ByteString (ResourceT IO) () -> m a) -> m a,+ compressionMethod :: BinaryCache.CompressionMethod,+ compressionLevel :: Int, omitDeriver :: Bool } defaultWithXzipCompressor :: forall m a. (ConduitM ByteString ByteString (ResourceT IO) () -> m a) -> m a-defaultWithXzipCompressor = ($ compress (Just 2))+defaultWithXzipCompressor = ($ Lzma.compress (Just 2)) defaultWithXzipCompressorWithLevel :: Int -> forall m a. (ConduitM ByteString ByteString (ResourceT IO) () -> m a) -> m a-defaultWithXzipCompressorWithLevel l = ($ compress (Just l))+defaultWithXzipCompressorWithLevel l = ($ Lzma.compress (Just l)) +defaultWithZstdCompressor :: forall m a. (ConduitM ByteString ByteString (ResourceT IO) () -> m a) -> m a+defaultWithZstdCompressor = ($ Zstd.compress 8)++defaultWithZstdCompressorWithLevel :: Int -> forall m a. (ConduitM ByteString ByteString (ResourceT IO) () -> m a) -> m a+defaultWithZstdCompressorWithLevel l = ($ Zstd.compress l)+ pushSingleStorePath :: (MonadMask m, MonadIO m) => -- | details for pushing to cache@@ -143,6 +154,9 @@ name = pushParamsName cache clientEnv = pushParamsClientEnv cache strategy = pushParamsStrategy cache storePath+ withCompressor = case compressionMethod strategy of+ BinaryCache.XZ -> defaultWithXzipCompressorWithLevel (compressionLevel strategy)+ BinaryCache.ZSTD -> defaultWithZstdCompressorWithLevel (compressionLevel strategy) narSizeRef <- liftIO $ newIORef 0 fileSizeRef <- liftIO $ newIORef 0 narHashRef <- liftIO $ newIORef ("" :: ByteString)@@ -150,7 +164,7 @@ -- 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+ -- stream store path as compressed nar file let cmd = proc "nix-store" ["--dump", toS storePathText] storePathSize :: Int64 storePathSize = Store.validPathInfoNarSize pathinfo@@ -158,12 +172,12 @@ -- create_group makes subprocess ignore signals such as ctrl-c that we handle in haskell main thread -- see https://github.com/haskell/process/issues/198 (ClosedStream, stdoutStream, Inherited, cph) <- liftIO $ streamingProcess (cmd {create_group = True})- withXzipCompressor strategy $ \xzCompressor -> do+ withCompressor $ \compressor -> do let stream' = stdoutStream .| passthroughSizeSink narSizeRef .| passthroughHashSink narHashRef- .| xzCompressor+ .| compressor .| passthroughSizeSink fileSizeRef .| passthroughHashSinkB16 fileHashRef let subdomain =@@ -178,7 +192,7 @@ (_ :: NoContent) <- liftIO $ (`withClientM` newClientEnv)- (API.createNar cachixClient (getCacheAuthToken (pushParamsSecret cache)) name (mapOutput coerce stream'))+ (API.createNar cachixClient (getCacheAuthToken (pushParamsSecret cache)) name (Just $ compressionMethod strategy) (mapOutput coerce stream')) $ escalate >=> \NoContent -> do exitcode <- waitForStreamingProcess cph@@ -297,7 +311,9 @@ Nothing -> throwIO $ NoSigningKey msg where -- we reverse list of caches to prioritize keys added as last- getBinaryCache c = filter (\bc -> Config.name bc == name) $ reverse $ Config.binaryCaches c+ getBinaryCache c =+ reverse $+ filter (\bc -> Config.name bc == name) (Config.binaryCaches c) msg :: Text msg = [iTrim|
src/Cachix/Client/PushQueue.hs view
@@ -127,7 +127,7 @@ cancelWith pushWorker StopWorker else do -- extend shutdown for another 90s- void $ Systemd.notify False $ "EXTEND_TIMEOUT_USEC=" <> show (90 * 1000 * 1000)+ void $ Systemd.notify False $ "EXTEND_TIMEOUT_USEC=" <> show (90 * 1000 * 1000 :: Int) putTextError $ "Waiting to finish: " <> show inprogress <> " pushing, " <> show queueLength <> " in queue" threadDelay (1000 * 1000) go
src/Cachix/Client/URI.hs view
@@ -1,48 +1,159 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE QuasiQuotes #-} module Cachix.Client.URI- ( getBaseUrl,+ ( URI,+ fromURIRef,+ getScheme,+ getHostname,+ appendSubdomain,+ getPortFor,+ getPath,+ requiresSSL,+ parseURI,+ serialize,+ getBaseUrl, defaultCachixURI, defaultCachixBaseUrl,+ UBS.Host (..),+ UBS.Scheme (..),+ UBS.Port (..), ) where +import Control.Monad (fail)+import qualified Data.Aeson as Aeson+import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as Char8+import Data.Either.Validation (Validation (Failure, Success))+import qualified Data.Map.Strict as Map+import Data.Maybe (fromJust)+import qualified Dhall+import qualified Dhall.Core import Protolude hiding (toS) import Protolude.Conv import Servant.Client-import URI.ByteString hiding (Scheme) import qualified URI.ByteString as UBS-import URI.ByteString.QQ+import qualified URI.ByteString.QQ as UBS --- TODO: make getBaseUrl internal+-- Default URIs +defaultCachixURI :: URI+defaultCachixURI = fromURIRef [UBS.uri|https://cachix.org|]++defaultCachixBaseUrl :: BaseUrl+defaultCachixBaseUrl = getBaseUrl defaultCachixURI++newtype URI = URI {getUri :: UBS.URIRef UBS.Absolute}+ deriving stock (Eq, Show)++fromURIRef :: UBS.URIRef UBS.Absolute -> URI+fromURIRef = URI++getScheme :: URI -> UBS.Scheme+getScheme = UBS.uriScheme . getUri++getHostname :: URI -> UBS.Host+getHostname = UBS.authorityHost . fromJust . UBS.uriAuthority . getUri++-- TODO: lenses?+appendSubdomain :: Text -> URI -> URI+appendSubdomain domain uri =+ let UBS.URI uScheme uAuthority uPath uQuery uFragment = getUri uri+ UBS.Authority aUserInfo aHost aPort = fromJust uAuthority+ newHost = UBS.Host $ toS domain <> "." <> UBS.hostBS aHost+ in URI $+ UBS.URI+ uScheme+ (Just (UBS.Authority aUserInfo newHost aPort))+ uPath+ uQuery+ uFragment++getPortFor :: UBS.Scheme -> Maybe UBS.Port+getPortFor scheme = Map.lookup scheme UBS.httpDefaultPorts++getPath :: URI -> ByteString+getPath = UBS.uriPath . getUri++requiresSSL :: UBS.Scheme -> Bool+requiresSSL (UBS.Scheme "https") = True+requiresSSL _ = False++parseURI :: ByteString -> Either UBS.URIParseError URI+parseURI bs = fromURIRef <$> UBS.parseURI UBS.strictURIParserOptions bs++serialize :: StringConv BS.ByteString s => URI -> s+serialize = toS . UBS.serializeURIRef' . getUri++instance Aeson.ToJSON URI where+ toJSON (URI uri) = Aeson.String . toS . UBS.serializeURIRef' $ uri++instance Aeson.FromJSON URI where+ parseJSON = Aeson.withText "URI" $ \text ->+ either (fail . show) (return . URI) $+ UBS.parseURI UBS.strictURIParserOptions (toS text)++instance Dhall.FromDhall URI where+ autoWith opts =+ Dhall.Decoder extract expected+ where+ textDecoder :: Dhall.Decoder Text+ textDecoder = Dhall.autoWith opts++ extract expression =+ case Dhall.extract textDecoder expression of+ Success x -> case UBS.parseURI UBS.strictURIParserOptions (toS x) of+ Left exception -> Dhall.extractError (show exception)+ Right path -> Success (fromURIRef path)+ Failure e -> Failure e++ expected = Dhall.expected textDecoder++instance Dhall.ToDhall URI where+ injectWith opts = Dhall.Encoder embed declared+ where+ textEncoder :: Dhall.Encoder Text+ textEncoder = Dhall.injectWith opts++ embed (URI uri) = Dhall.embed textEncoder $ toS (UBS.serializeURIRef' uri)++ declared = Dhall.Core.Text+ -- | Partial function from URI to BaseUrl-getBaseUrl :: URIRef Absolute -> BaseUrl-getBaseUrl uriref =- case uriAuthority uriref of+--+-- TODO: We should error out during the parsing stage with a nice error.+-- TODO: make getBaseUrl internal+getBaseUrl :: URI -> BaseUrl+getBaseUrl (URI uriref) =+ case UBS.uriAuthority uriref of Nothing -> panic "missing host in url" Just authority ->- BaseUrl- getScheme- (toS (hostBS (authorityHost authority)))- getPort- (toS (uriPath uriref))+ BaseUrl scheme hostname port path where- getScheme :: Scheme- getScheme = case uriScheme uriref of+ scheme :: Scheme+ scheme = case UBS.uriScheme uriref of UBS.Scheme "http" -> Http UBS.Scheme "https" -> Https _ -> panic "uri can only be http/https"- getPort :: Int- getPort = maybe defaultPort portNumber $ authorityPort authority++ hostname = toS $ UBS.hostBS (UBS.authorityHost authority)++ port :: Int+ port = maybe defaultPort UBS.portNumber $ UBS.authorityPort authority+ defaultPort :: Int- defaultPort = case getScheme of+ defaultPort = case scheme of Http -> 80 Https -> 443 -defaultCachixURI :: URIRef Absolute-defaultCachixURI = [uri|https://cachix.org|]+ path = toS $ removeTrailingSlash (UBS.uriPath uriref) -defaultCachixBaseUrl :: BaseUrl-defaultCachixBaseUrl = getBaseUrl defaultCachixURI+ -- Servant expects the trailing slash to be removed+ -- https://hackage.haskell.org/package/servant-client-core-0.19/docs/Servant-Client-Core.html#v:parseBaseUrl+ removeTrailingSlash :: ByteString -> ByteString+ removeTrailingSlash "" = ""+ removeTrailingSlash str = case Char8.last str of+ '/' -> Char8.init str+ _ -> str
src/Cachix/Deploy/Activate.hs view
@@ -6,7 +6,9 @@ import qualified Cachix.API.WebSocketSubprotocol as WSS import qualified Cachix.Client.InstallationMode as InstallationMode import qualified Cachix.Client.NetRc as NetRc-import qualified Cachix.Deploy.Websocket as CachixWebsocket+import Cachix.Client.URI (URI)+import qualified Cachix.Client.URI as URI+import qualified Cachix.Deploy.Log as Log import qualified Cachix.Types.BinaryCache as BinaryCache import Cachix.Types.Permission (Permission (..)) import qualified Data.Aeson as Aeson@@ -15,186 +17,155 @@ #else import qualified Data.HashMap.Strict as HM #endif-import qualified Data.Conduit as Conduit import qualified Data.Conduit.Combinators as Conduit import qualified Data.Conduit.Process as Conduit-import Data.Time.Clock (getCurrentTime)-import qualified Data.UUID.V4 as UUID import qualified Data.Vector as Vector-import qualified Katip as K-import qualified Network.WebSockets as WS import Protolude hiding (log, toS) import Protolude.Conv (toS) import Servant.Auth.Client (Token (..))-import System.Directory (canonicalizePath, doesPathExist)+import qualified System.Directory as Directory import System.FilePath ((</>)) import System.IO.Temp (withSystemTempDirectory) import System.Process-import System.Timeout (timeout) import Prelude (String) -domain :: CachixWebsocket.Options -> WSS.Cache -> Text-domain options cache = toS (WSS.cacheName cache) <> "." <> toS (CachixWebsocket.host options)+data Status+ = Success+ | Failure FailureReason+ | Rollback FailureReason+ deriving (Show, Exception) --- TODO: get uri scheme-uri :: CachixWebsocket.Options -> WSS.Cache -> Text-uri options cache = "https://" <> domain options cache+data FailureReason+ = NetworkTestFailure+ | RollbackScriptExitFailure+ | RollbackScriptUnexpectedError IOException+ | ShellCommandFailure {command :: String, exitCode :: Int}+ | UnexpectedError SomeException+ deriving (Show) -hackFlush :: K.KatipContextT IO ()-hackFlush = liftIO $ threadDelay (5 * 1000 * 1000)+instance Exception FailureReason where+ displayException = \case+ NetworkTestFailure -> "Cannot connect back to Cachix Deploy after activating the new deployment"+ RollbackScriptExitFailure -> "The rollback script returned a non-zero exit code"+ RollbackScriptUnexpectedError e -> "Cannot run rollback script: " <> displayException e+ ShellCommandFailure {command, exitCode} ->+ toS $+ unwords+ [ "Failed to run " <> toS command,+ show exitCode+ ]+ UnexpectedError e ->+ toS $+ unwords+ [ "The deployment failed with an unexpected error:",+ toS (displayException e)+ ] --- TODO: what if websocket gets closed while deploying?-activate ::- CachixWebsocket.Options ->- WS.Connection ->- Conduit.ConduitT ByteString Void IO () ->+downloadStorePaths ::+ -- | Logging context+ Log.LogStream ->+ -- | Deployment details WSS.DeploymentDetails ->- WSS.AgentInformation ->- ByteString ->- K.KatipContextT IO ()-activate options connection sourceStream deploymentDetails agentInfo agentToken = withCacheArgs options agentInfo agentToken $ \cacheArgs -> do- let storePath = WSS.storePath deploymentDetails- K.logLocM K.InfoS $ K.ls $ "Deploying #" <> index <> ": " <> WSS.storePath deploymentDetails- deploymentStarted Nothing+ -- | Binary cache args+ [String] ->+ IO ()+downloadStorePaths logStream deploymentDetails cacheArgs = do+ -- Download the store path from the binary cache -- TODO: add GC root so it's preserved for the next command- -- get the store path using caches- (downloadExitCode, _, _) <-- liftIO $ shellOut "nix-store" (["-r", toS storePath] <> cacheArgs)- -- TODO: use exceptions to simplify this code- case downloadExitCode of- ExitFailure _ -> deploymentFailed- ExitSuccess -> do- -- TODO: query the remote store to get the size before downloading (and possibly running out of disk space)- (_, pathInfoJSON, _) <- liftIO $ shellNoStream "nix" (cacheArgs <> ["--extra-experimental-features", "nix-command", "path-info", "-S", "--json", toS storePath])- deploymentStarted $ extractClosureSize =<< Aeson.decode (toS pathInfoJSON)+ runShell logStream "nix-store" (["-r", toS storePath] <> cacheArgs) - activateProfile storePath $ \oldStorePath -> do- -- connect the the backend to see if we broke networking- websocketSucceeded <- liftIO $ timeout (10 * 1000 * 1000) $ CachixWebsocket.runOnce options $ \_ _ _ _ -> return ()- case websocketSucceeded of- Nothing -> rollback oldStorePath- Just () -> do- case WSS.rollbackScript deploymentDetails of- Just script -> do- (downloadRollbackExitCode, _, _) <- liftIO $ shellOut "nix-store" (["-r", toS script] <> cacheArgs)- case downloadRollbackExitCode of- ExitFailure _ -> deploymentFailed- ExitSuccess -> do- (rollbackExitCode, _, _) <- liftIO $ shellOut (toS script) []- case rollbackExitCode of- ExitFailure _ -> rollback oldStorePath- ExitSuccess -> deloymentSucceeded- Nothing -> deloymentSucceeded+ -- Download the rollback script, if provided+ for_ (WSS.rollbackScript deploymentDetails) $ \script ->+ runShell logStream "nix-store" (["-r", toS script] <> cacheArgs) where- deploymentID = WSS.id (deploymentDetails :: WSS.DeploymentDetails)- deloymentSucceeded = do- now <- liftIO getCurrentTime- sendMessage $ deploymentFinished True now- liftIO $ log "Successfully activated the deployment."- K.logLocM K.InfoS $ K.ls $ "Deployment #" <> index <> " finished"- hackFlush- deploymentStarted closureSize = do- now <- liftIO getCurrentTime- sendMessage $- WSS.DeploymentStarted- { WSS.id = deploymentID,- WSS.time = now,- WSS.closureSize = closureSize- }- deploymentFinished hasSucceeded now =- WSS.DeploymentFinished- { WSS.id = deploymentID,- WSS.time = now,- WSS.hasSucceeded = hasSucceeded- }- deploymentFailed = do- now <- liftIO getCurrentTime- liftIO $ log "Failed to activate the deployment."- K.logLocM K.InfoS $ K.ls $ "Deploying #" <> index <> " failed."- sendMessage $ deploymentFinished False now- hackFlush- rollback :: Maybe Text -> K.KatipContextT IO ()- rollback (Just path) = do- liftIO $ log "Deployment failed, rolling back ..."- activateProfile path $ const deploymentFailed- rollback Nothing = do- liftIO $ log "Skipping rollback as this is the first deployment."- deploymentFailed- activateProfile :: Text -> (Maybe Text -> K.KatipContextT IO ()) -> K.KatipContextT IO ()- activateProfile path action = do- (profilePath, activationScripts) <- liftIO $ getActivationScript path (CachixWebsocket.profile options)- profileExists <- liftIO $ doesPathExist profilePath- -- in case we're doing the first deployment the profile won't exist so we can't rollback- previousStorePath <-- if profileExists- then do- oldStorePath <- liftIO $ canonicalizePath profilePath- return $ Just $ toS oldStorePath- else return Nothing- (activateProfileExitCode, _, _) <- liftIO $ shellOut "nix-env" ["-p", toS profilePath, "--set", toS path]- case activateProfileExitCode of- ExitFailure _ -> deploymentFailed- ExitSuccess -> do- -- activate configuration- exitCodes <- for activationScripts $ \(cmd, args) -> do- (activateScriptExitCode, _, _) <- liftIO $ shellOut cmd args- return activateScriptExitCode- if not (all (== ExitSuccess) exitCodes)- then deploymentFailed- else action previousStorePath- index :: Text- index = show $ WSS.index deploymentDetails- -- TODO: it would be better to stream and also return the text- shellNoStream :: FilePath -> [String] -> IO (ExitCode, String, String)- shellNoStream cmd args = do- log $ "$ " <> toS cmd <> " " <> toS (unwords $ fmap toS args)- (exitCode, procstdout, procstderr) <- readProcessWithExitCode cmd args ""- log $ toS procstderr- return (exitCode, toS procstdout, toS procstderr)- shellOut :: FilePath -> [String] -> IO (ExitCode, (), ())- shellOut cmd args = do- log $ "$ " <> toS cmd <> " " <> toS (unwords $ fmap toS args)- Conduit.sourceProcessWithStreams (proc cmd args) Conduit.sinkNull sourceStream sourceStream- log :: ByteString -> IO ()- log msg = Conduit.connect (Conduit.yieldMany ["\n" <> msg <> "\n"]) sourceStream- sendMessage cmd = liftIO $ do- command <- createMessage cmd- WSS.sendMessage connection command- createMessage command = do- uuid <- UUID.nextRandom- return $- WSS.Message- { WSS.method = method,- WSS.command = command,- WSS.id = uuid,- WSS.agent = Just $ WSS.id (agentInfo :: WSS.AgentInformation)- }- where- method = case command of- WSS.DeploymentStarted {} -> "DeploymentStarted"- WSS.DeploymentFinished {} -> "DeploymentFinished"- -- TODO: home-manager- getActivationScript :: Text -> Text -> IO (FilePath, [Command])- getActivationScript storePath profile = do- isNixOS <- doesPathExist $ toS storePath </> "nixos-version"- isNixDarwin <- doesPathExist $ toS storePath </> "darwin-version"- user <- InstallationMode.getUser- (systemProfile, cmds) <- case (isNixOS, isNixDarwin) of- (True, _) -> return ("system", [(toS storePath </> "bin/switch-to-configuration", ["switch"])])- (_, True) ->- -- https://github.com/LnL7/nix-darwin/blob/master/pkgs/nix-tools/darwin-rebuild.sh- return- ( "system-profiles/system",- [ ("mkdir", ["-p", "-m", "0755", "/nix/var/nix/profiles/system-profiles"]),- (toS storePath </> "activate-user", []),- (toS storePath </> "activate", [])- ]- )- (_, _) -> return ("system", [])- return ("/nix/var/nix/profiles" </> if profile == "" then systemProfile else toS profile, cmds)+ storePath = WSS.storePath deploymentDetails +type RollbackAction = IO ()++-- | Activate the profile and return a rollback action if available.+activate ::+ -- | Logging context+ Log.LogStream ->+ -- | Profile name+ Text ->+ -- | Store path to activate+ FilePath ->+ -- | Returns a rollback action if available+ IO (Maybe RollbackAction)+activate logStream profileName storePath = do+ (profilePath, activationScripts) <- getActivationScript profileName storePath+ previousProfilePath <- toStorePath profilePath++ -- Activate the configuration+ -- TODO: Check with Domen whether we can exit early here+ forM_ activationScripts $ uncurry (runShell logStream)++ pure $ Just rollback <*> previousProfilePath+ where+ toStorePath profilePath = do+ profileExists <- Directory.doesPathExist profilePath+ if profileExists+ then Just <$> Directory.canonicalizePath profilePath+ else pure Nothing++ -- We can't use '--rollback' because it just selects the next generation+ -- down from our deployment, which is not necessarily the generation that+ -- was previously active.+ rollback = void . activate logStream profileName+ type Command = (String, [String]) +getActivationScript :: Text -> FilePath -> IO (FilePath, [Command])+getActivationScript profile storePath = do+ let checkPath p = Directory.doesPathExist $ toS storePath </> p+ isNixOS <- checkPath "nixos-version"+ isNixDarwin <- checkPath "darwin-version"+ isHomeManager <- checkPath "hm-version"+ user <- InstallationMode.getUser+ let systemProfileDir = "/nix/var/nix/profiles"+ let perUserProfileDir = systemProfileDir </> "per-user" </> toS user+ let mkProfilePath profileBaseDir defaultProfile =+ profileBaseDir </> if profile == "" then defaultProfile else toS profile+ -- Sets the new profile. This is needed for NixOS and nix-darwin. The Home+ -- Manager activation script does the profile setting by itself.+ -- TODO: document what happens if the wrong user is used for the agent+ let setNewProfile profilePath =+ ("nix-env", ["-p", profilePath, "--set", storePath])+ return $ case (isNixOS, isNixDarwin, isHomeManager) of+ (True, _, _) ->+ let profilePath = mkProfilePath systemProfileDir "system"+ in ( profilePath,+ [ setNewProfile profilePath,+ (toS storePath </> "bin/switch-to-configuration", ["switch"])+ ]+ )+ (_, True, _) ->+ -- https://github.com/LnL7/nix-darwin/blob/master/pkgs/nix-tools/darwin-rebuild.sh+ let profilePath = mkProfilePath systemProfileDir "system-profiles/system"+ in ( profilePath,+ [ ("mkdir", ["-p", "-m", "0755", "/nix/var/nix/profiles/system-profiles"]),+ setNewProfile profilePath,+ (toS storePath </> "activate-user", []),+ (toS storePath </> "activate", [])+ ]+ )+ (_, _, True) ->+ ( mkProfilePath perUserProfileDir "home-manager",+ [(toS storePath </> "activate", [])]+ )+ (_, _, _) ->+ let profilePath = mkProfilePath systemProfileDir "system"+ in (profilePath, [setNewProfile profilePath])++-- TODO: send errors as well+-- TODO: fix either/maybe types+getClosureSize :: [String] -> Text -> IO (Either Text (Maybe Int64))+getClosureSize cacheArgs storePath = do+ (exitCode, pathInfoJSON, nixError) <- readProcessWithExitCode "nix" (cacheArgs <> ["--extra-experimental-features", "nix-command", "path-info", "-S", "--json", toS storePath]) ""+ case exitCode of+ ExitFailure _ -> pure $ Left (toS nixError)+ ExitSuccess -> pure $ Right $ Aeson.decode (toS pathInfoJSON) >>= extractClosureSize+ extractClosureSize :: Aeson.Value -> Maybe Int64 extractClosureSize (Aeson.Array vector) = case Vector.toList vector of [Aeson.Object obj] -> case HM.lookup "closureSize" obj of@@ -204,34 +175,50 @@ extractClosureSize _ = Nothing -- TODO: don't create tmpfile for public caches-withCacheArgs :: CachixWebsocket.Options -> WSS.AgentInformation -> ByteString -> ([String] -> K.KatipContextT IO ()) -> K.KatipContextT IO ()-withCacheArgs options agentInfo agentToken m =+withCacheArgs :: URI -> WSS.AgentInformation -> Text -> ([String] -> IO a) -> IO a+withCacheArgs host agentInfo agentToken m = withSystemTempDirectory "netrc" $ \dir -> do let filepath = dir </> "netrc" args <- case WSS.cache agentInfo of Just cache -> do+ let cacheName = WSS.cacheName cache+ let cacheURI = URI.appendSubdomain cacheName host -- TODO: ugh let bc = BinaryCache.BinaryCache { BinaryCache.name = "",- BinaryCache.uri = toS (uri options cache),+ BinaryCache.uri = URI.serialize cacheURI, BinaryCache.publicSigningKeys = [], BinaryCache.isPublic = WSS.isPublic cache, BinaryCache.githubUsername = "",- BinaryCache.permission = Read+ BinaryCache.permission = Read,+ BinaryCache.preferredCompressionMethod = BinaryCache.XZ }- liftIO $ NetRc.add (Token agentToken) [bc] filepath- return $ cachesArgs <> ["--option", "netrc-file", filepath]+ NetRc.add (Token (toS agentToken)) [bc] filepath+ return $ cachesArgs cache <> ["--option", "netrc-file", filepath] Nothing ->- return cachesArgs+ return [] m args where- cachesArgs :: [String]- cachesArgs = case WSS.cache agentInfo of- Just cache ->- let officialCache = "cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY="- substituters = ["--option", "extra-substituters", toS (uri options cache)]- noNegativeCaching = ["--option", "narinfo-cache-negative-ttl", "0"]- sigs = ["--option", "trusted-public-keys", officialCache <> " " <> toS (domain options cache) <> "-1:" <> toS (WSS.publicKey cache)]- in substituters ++ sigs ++ noNegativeCaching- Nothing -> []+ cachesArgs :: WSS.Cache -> [String]+ cachesArgs cache =+ let cacheName = WSS.cacheName cache+ cacheURI = URI.appendSubdomain cacheName host+ hostname = (URI.hostBS . URI.getHostname) cacheURI+ officialCache = "cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY="+ substituters = ["--option", "extra-substituters", URI.serialize cacheURI]+ noNegativeCaching = ["--option", "narinfo-cache-negative-ttl", "0"]+ sigs = ["--option", "trusted-public-keys", officialCache <> " " <> toS hostname <> "-1:" <> toS (WSS.publicKey cache)]+ in substituters ++ sigs ++ noNegativeCaching++runShell :: Log.LogStream -> FilePath -> [String] -> IO ()+runShell logStream cmd args = runShellWithExitCode logStream cmd args >>= handleError+ where+ handleError (ExitFailure exitCode) = throwIO $ ShellCommandFailure {command = cmd, exitCode}+ handleError ExitSuccess = pure ()++runShellWithExitCode :: Log.LogStream -> FilePath -> [String] -> IO ExitCode+runShellWithExitCode logStream cmd args = do+ Log.streamLine logStream $ "$ " <> toS cmd <> " " <> toS (unwords $ fmap toS args)+ (exitCode, _, _) <- Conduit.sourceProcessWithStreams (proc cmd args) Conduit.sinkNull logStream logStream+ pure exitCode
src/Cachix/Deploy/ActivateCommand.hs view
@@ -1,5 +1,3 @@-{-# LANGUAGE NamedFieldPuns #-}- module Cachix.Deploy.ActivateCommand where import qualified Cachix.API.Deploy as API@@ -8,8 +6,10 @@ import qualified Cachix.Client.Env as Env import Cachix.Client.Servant (deployClient) import qualified Cachix.Deploy.OptionsParser as DeployOptions+import qualified Cachix.Types.Deploy as Types import qualified Cachix.Types.DeployResponse as DeployResponse import qualified Data.Aeson as Aeson+import Data.HashMap.Strict (filterWithKey) import qualified Data.HashMap.Strict as HM import Protolude hiding (toS) import Protolude.Conv@@ -19,7 +19,7 @@ import System.Environment (getEnv) run :: Config.CachixOptions -> DeployOptions.ActivateOptions -> IO ()-run cachixOptions DeployOptions.ActivateOptions {DeployOptions.payloadPath} = do+run cachixOptions DeployOptions.ActivateOptions {DeployOptions.payloadPath, DeployOptions.agents} = do agentToken <- getEnv "CACHIX_ACTIVATE_TOKEN" clientEnv <- Env.createClientEnv cachixOptions payloadEither <- Aeson.eitherDecodeFileStrict' payloadPath@@ -28,6 +28,7 @@ hPutStrLn stderr $ "Error while parsing JSON: " <> err exitFailure Right payload -> do- response <- escalate <=< (`runClientM` clientEnv) $ API.activate deployClient (Token $ toS agentToken) payload+ let deploy = payload {Types.agents = filterWithKey (\k _ -> k `elem` agents) (Types.agents payload)}+ response <- escalate <=< (`runClientM` clientEnv) $ API.activate deployClient (Token $ toS agentToken) deploy for_ (HM.toList $ DeployResponse.agents response) $ \(_, url) -> putStrLn url
src/Cachix/Deploy/Agent.hs view
@@ -1,56 +1,277 @@+{-# LANGUAGE DataKinds #-} {-# LANGUAGE DuplicateRecordFields #-} module Cachix.Deploy.Agent where import qualified Cachix.API.WebSocketSubprotocol as WSS import qualified Cachix.Client.Config as Config-import Cachix.Client.URI (getBaseUrl)-import qualified Cachix.Deploy.OptionsParser as AgentOptions-import Cachix.Deploy.StdinProcess (readProcess)-import qualified Cachix.Deploy.Websocket as CachixWebsocket+import Cachix.Client.URI (URI)+import qualified Cachix.Client.URI as URI+import Cachix.Client.Version (versionNumber)+import qualified Cachix.Deploy.Deployment as Deployment+import qualified Cachix.Deploy.Lock as Lock+import qualified Cachix.Deploy.Log as Log+import qualified Cachix.Deploy.OptionsParser as CLI+import qualified Cachix.Deploy.StdinProcess as StdinProcess+import qualified Cachix.Deploy.Websocket as WebSocket+import qualified Control.Concurrent.Async as Async+import Control.Concurrent.Extra (once)+import qualified Control.Concurrent.MVar as MVar+import Control.Exception.Safe (onException, throwString, withException)+import qualified Control.Exception.Safe as Safe+import qualified Control.Retry as Retry import qualified Data.Aeson as Aeson+import Data.IORef+import Data.String (String) import qualified Katip as K-import qualified Network.WebSockets as WS import Paths_cachix (getBinDir)-import Protolude hiding (toS)+import Protolude hiding (onException, toS) import Protolude.Conv-import qualified Servant.Client as Servant+import qualified System.Directory as Directory+import System.Environment (getEnv, lookupEnv)+import qualified System.Posix.Files as Posix.Files+import qualified System.Posix.Process as Posix+import qualified System.Posix.Signals as Signals+import qualified System.Posix.Types as Posix+import qualified System.Posix.User as Posix.User -run :: Config.CachixOptions -> AgentOptions.AgentOptions -> IO ()-run cachixOptions agentOpts =- CachixWebsocket.runForever options handleMessage+type ServiceWebSocket = WebSocket.WebSocket (WSS.Message WSS.AgentCommand) (WSS.Message WSS.BackendCommand)++data Agent = Agent+ { name :: Text,+ token :: Text,+ profileName :: Text,+ agentState :: IORef (Maybe WSS.AgentInformation),+ pid :: Posix.CPid,+ bootstrap :: Bool,+ host :: URI,+ logOptions :: Log.Options,+ withLog :: Log.WithLog,+ websocket :: ServiceWebSocket+ }++agentIdentifier :: Text -> Text+agentIdentifier agentName = unwords [agentName, toS versionNumber]++run :: Config.CachixOptions -> CLI.AgentOptions -> IO ()+run cachixOptions agentOptions =+ Log.withLog logOptions $ \withLog ->+ logAndExitWithFailure withLog $+ withAgentLock agentOptions $ do+ checkUserOwnsHome++ -- TODO: show a more helpful error if the token is missing+ -- TODO: show a more helpful error when the token isn't valid+ -- TODO: wrap the token in a newtype or use servant's Token+ agentToken <- toS <$> getEnv "CACHIX_AGENT_TOKEN"+ agentState <- newIORef Nothing++ pid <- Posix.getProcessID++ let port = fromMaybe (URI.Port 80) $ (URI.getPortFor . URI.getScheme) host+ let websocketOptions =+ WebSocket.Options+ { WebSocket.host = basename,+ WebSocket.port = port,+ WebSocket.path = "/ws",+ WebSocket.useSSL = URI.requiresSSL (URI.getScheme host),+ WebSocket.headers = WebSocket.createHeaders agentName agentToken,+ WebSocket.identifier = agentIdentifier agentName+ }++ websocket <- WebSocket.new withLog websocketOptions+ 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++ let agent =+ Agent+ { name = agentName,+ token = agentToken,+ profileName = profileName,+ agentState = agentState,+ pid = pid,+ bootstrap = CLI.bootstrap agentOptions,+ host = host,+ logOptions = logOptions,+ withLog = withLog,+ websocket = websocket+ }++ WebSocket.readDataMessages channel $ \message ->+ handleCommand agent (WSS.command message) where- host = toS $ Servant.baseUrlHost $ getBaseUrl $ Config.host cachixOptions- name = AgentOptions.name agentOpts- profile = fromMaybe "system" (AgentOptions.profile agentOpts)- options =- CachixWebsocket.Options- { CachixWebsocket.host = host,- CachixWebsocket.name = name,- CachixWebsocket.path = "/ws",- CachixWebsocket.profile = profile,- CachixWebsocket.isVerbose = Config.verbose cachixOptions+ host = Config.host cachixOptions+ basename = URI.getHostname host+ 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+ else Log.Normal++ logOptions =+ Log.Options+ { verbosity = verbosity,+ namespace = "agent",+ environment = "production" }- handleMessage :: ByteString -> (K.KatipContextT IO () -> IO ()) -> WS.Connection -> CachixWebsocket.AgentState -> ByteString -> K.KatipContextT IO ()- handleMessage payload _ _ agentState _ =- CachixWebsocket.parseMessage payload (handleCommand . WSS.command)++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)++registerAgent :: Agent -> WSS.AgentInformation -> IO ()+registerAgent Agent {agentState, withLog} agentInformation = do+ withLog $ K.logLocM K.InfoS "Agent registered."+ atomicWriteIORef agentState (Just agentInformation)++launchDeployment :: Agent -> WSS.DeploymentDetails -> IO ()+launchDeployment agent@Agent {..} deploymentDetails = do+ agentRegistered <- readIORef agentState++ case agentRegistered of+ -- TODO: the agent should either not exist before we register or+ -- we should re-register here as a precaution.+ Nothing -> pure ()+ Just agentInformation -> do+ binDir <- toS <$> getBinDir+ exitCode <-+ StdinProcess.readProcess (binDir <> "/.cachix-deployment") [] $+ toS . Aeson.encode $+ Deployment.Deployment+ { Deployment.agentName = name,+ Deployment.agentToken = token,+ Deployment.profileName = profileName,+ Deployment.host = host,+ Deployment.deploymentDetails = deploymentDetails,+ Deployment.agentInformation = agentInformation,+ Deployment.logOptions = logOptions+ }++ when+ (bootstrap && exitCode == ExitSuccess)+ (verifyBootstrapSuccess agent)++verifyBootstrapSuccess :: Agent -> IO ()+verifyBootstrapSuccess Agent {name, withLog} = do+ withLog . K.logLocM K.InfoS . K.ls $+ unwords ["Waiting for another agent to take over..."]++ eAgentPid <-+ Safe.tryIO $+ Retry.recoverAll+ (Retry.limitRetries 20 <> Retry.constantDelay 1000)+ (const waitForAgent)++ case eAgentPid of+ Right pid -> do+ withLog . K.logLocM K.InfoS . K.ls $+ unwords ["Found an active agent for", name, "with PID " <> show pid <> ".", "Exiting."]+ exitSuccess+ _ -> do+ withLog . K.logLocM K.InfoS . K.ls $+ unwords ["Cannot find an active agent for", name <> ".", "Waiting for more deployments."]+ where+ lockfile = lockFilename name++ -- The PID might be stale in rare cases. Only use this for diagnostics.+ waitForAgent :: IO Posix.CPid+ waitForAgent = do+ lock <- Lock.withTryLock lockfile (pure ())+ mpid <- Lock.readPidFile lockfile+ case (lock, mpid) of+ (Nothing, Just pid) -> pure pid+ _ -> throwString "No active agent found"++handleCommand :: Agent -> WSS.BackendCommand -> IO ()+handleCommand agent command =+ case command of+ WSS.AgentRegistered agentInformation -> registerAgent agent agentInformation+ WSS.Deployment deploymentDetails -> launchDeployment agent deploymentDetails++-- | Asynchronously open and maintain a websocket connection to the backend for+-- sending deployment progress updates.+connectToService :: ServiceWebSocket -> IO (IO ())+connectToService websocket = do+ initialConnection <- MVar.newEmptyMVar+ close <- MVar.newEmptyMVar++ thread <- Async.async $+ WebSocket.runConnection websocket $ do+ MVar.putMVar initialConnection ()+ WebSocket.handleJSONMessages websocket (MVar.readMVar close)++ -- Block until the connection has been established+ void $ MVar.takeMVar initialConnection++ once $ MVar.putMVar close () >> Async.wait thread++-- | Fetch the home directory and verify that the owner matches the current user.+-- Throws either 'NoHomeFound' or 'UserDoesNotOwnHome'.+checkUserOwnsHome :: IO ()+checkUserOwnsHome = do+ home <- Directory.getHomeDirectory `onException` throwIO NoHomeFound+ stat <- Posix.Files.getFileStatus home+ userId <- Posix.User.getEffectiveUserID++ when (userId /= Posix.Files.fileOwner stat) $ do+ userName <- Posix.User.userName <$> Posix.User.getUserEntryForID userId+ sudoUser <- lookupEnv "SUDO_USER"+ throwIO $+ UserDoesNotOwnHome+ { userName = userName,+ sudoUser = sudoUser,+ home = home+ }++data Error+ = -- | An agent with the same name is already running.+ AgentAlreadyRunning Text+ | -- | No home directory.+ NoHomeFound+ | -- | Safeguard against creating root-owned files in user directories.+ -- This is an issue on macOS, where, by default, sudo does not reset $HOME.+ UserDoesNotOwnHome+ { userName :: String,+ sudoUser :: Maybe String,+ home :: FilePath+ }+ deriving (Show)++instance Exception Error where+ displayException = \case+ AgentAlreadyRunning agentName -> toS $ unwords ["The agent", agentName, "is already running."]+ NoHomeFound -> "Could not find the user’s home directory. Make sure to set the $HOME variable."+ UserDoesNotOwnHome {userName = userName, sudoUser = sudoUser, home = home} ->+ if isJust sudoUser+ then toS $ unlines [warningMessage, suggestSudoFlagH]+ else toS warningMessage where- handleCommand :: WSS.BackendCommand -> K.KatipContextT IO ()- handleCommand (WSS.AgentRegistered agentInformation) =- CachixWebsocket.registerAgent agentState agentInformation- handleCommand (WSS.Deployment deploymentDetails) =- liftIO $ do- binDir <- toS <$> getBinDir- readProcess (binDir <> "/.cachix-deployment") [] $- toS . Aeson.encode $- CachixWebsocket.Input- { deploymentDetails = deploymentDetails,- websocketOptions =- CachixWebsocket.Options- { host = host,- name = name,- path = "/ws-deployment",- profile = profile,- isVerbose = Config.verbose cachixOptions- }- }+ warningMessage = "The current user (" <> toS userName <> ") does not own the home directory (" <> toS home <> ")"+ suggestSudoFlagH = "Try running the agent with `sudo -H`."
+ src/Cachix/Deploy/Deployment.hs view
@@ -0,0 +1,21 @@+module Cachix.Deploy.Deployment where++import qualified Cachix.API.WebSocketSubprotocol as WSS+import Cachix.Client.URI as Cachix+import qualified Cachix.Deploy.Log as Log+import qualified Data.Aeson as Aeson+import Protolude++-- | Everything required for the standalone deployment binary to complete a+-- deployment.+data Deployment = Deployment+ { agentName :: Text,+ agentToken :: Text,+ profileName :: Text,+ agentInformation :: WSS.AgentInformation,+ deploymentDetails :: WSS.DeploymentDetails,+ host :: Cachix.URI,+ logOptions :: Log.Options+ }+ deriving stock (Show, Generic)+ deriving anyclass (Aeson.ToJSON, Aeson.FromJSON)
src/Cachix/Deploy/Lock.hs view
@@ -1,20 +1,30 @@-module Cachix.Deploy.Lock (withTryLock) where+module Cachix.Deploy.Lock+ ( defaultLockDirectory,+ getLockDirectory,+ readPidFile,+ withTryLock,+ withTryLockAndPid,+ )+where import qualified Lukko as Lock import Protolude hiding ((<.>)) import qualified System.Directory as Directory import System.FilePath ((<.>), (</>))+import System.Posix (getProcessID)+import System.Posix.Types (CPid (..)) +lockExtension :: FilePath+lockExtension = "lock"++pidExtension :: FilePath+pidExtension = "pid"+ defaultLockDirectory :: FilePath defaultLockDirectory = "cachix" </> "deploy" </> "locks" --- | Run an IO action with an acquired profile lock. Returns immediately if the profile is already locked.------ Lock files are not deleted after use.------ macOS: if using sudo, make sure to use `-H` to reset the home directory.-withTryLock :: Text -> IO a -> IO (Maybe a)-withTryLock profileName action = do+getLockDirectory :: IO FilePath+getLockDirectory = do lockDirectory <- Directory.getXdgDirectory Directory.XdgCache defaultLockDirectory Directory.createDirectoryIfMissing True lockDirectory@@ -25,13 +35,41 @@ & Directory.setOwnerExecutable True & Directory.setOwnerSearchable True - let lockFile = lockDirectory </> toS profileName <.> "lock"+ pure lockDirectory +readPidFile :: FilePath -> IO (Maybe CPid)+readPidFile pidFilename = do+ lockDirectory <- getLockDirectory+ pidContents <- readFile (lockDirectory </> pidFilename <.> pidExtension)+ pure (readMaybe pidContents)++-- | Run an IO action with an acquired profile lock. Returns immediately if the profile is already locked.+--+-- Lock files are not deleted after use.+--+-- macOS: if using sudo, make sure to use `-H` to reset the home directory.+withTryLock :: FilePath -> IO a -> IO (Maybe a)+withTryLock lockFilename action = do+ lockDirectory <- getLockDirectory++ let lockFile = lockDirectory </> lockFilename <.> lockExtension+ bracket (Lock.fdOpen lockFile) (Lock.fdUnlock *> Lock.fdClose) $ \fd -> do isLocked <- Lock.fdTryLock fd Lock.ExclusiveLock if isLocked- then Just <$> action+ then fmap Just action else pure Nothing++withTryLockAndPid :: FilePath -> IO a -> IO (Maybe a)+withTryLockAndPid lockFilename action = do+ lockDirectory <- getLockDirectory++ let pidFile = lockDirectory </> lockFilename <.> pidExtension++ withTryLock lockFilename $ do+ CPid pid <- getProcessID+ writeFile pidFile (show pid)+ action
+ src/Cachix/Deploy/Log.hs view
@@ -0,0 +1,84 @@+{-# LANGUAGE DerivingStrategies #-}++module Cachix.Deploy.Log where++import qualified Cachix.API.WebSocketSubprotocol as WSS+import qualified Control.Concurrent.STM.TMQueue as TMQueue+import Control.Exception.Safe (MonadMask, bracket)+import qualified Data.Aeson as Aeson+import Data.Conduit ((.|))+import qualified Data.Conduit as Conduit+import qualified Data.Conduit.Combinators as Conduit+import qualified Data.Conduit.TQueue as Conduit+import Data.Time.Clock (getCurrentTime)+import qualified Katip+import qualified Network.WebSockets as WS+import Protolude hiding (bracket, toS)+import Protolude.Conv (toS)++-- A temporary crutch while we fix up the types+type WithLog = Katip.KatipContextT IO () -> IO ()++data Options = Options+ { -- | The minimum verbosity level+ verbosity :: Verbosity,+ -- | The logging namespace, e.g. agent+ namespace :: Katip.Namespace,+ -- | The logging environment, e.g. production+ environment :: Katip.Environment+ }+ deriving stock (Show, Generic)+ deriving anyclass (Aeson.ToJSON, Aeson.FromJSON)++data Verbosity+ = Normal+ | Verbose+ deriving stock (Show, Generic)+ deriving anyclass (Aeson.ToJSON, Aeson.FromJSON)++withLog ::+ (MonadIO m, MonadMask m) =>+ Options ->+ (WithLog -> m a) ->+ m a+withLog Options {..} inner =+ let permit = case verbosity of+ Verbose -> Katip.DebugS+ Normal -> Katip.InfoS++ createLogEnv = liftIO $ do+ logEnv <- Katip.initLogEnv namespace environment+ stdoutScribe <- Katip.mkHandleScribe Katip.ColorIfTerminal stdout (Katip.permitItem permit) Katip.V2+ Katip.registerScribe "stdout" stdoutScribe Katip.defaultScribeSettings logEnv++ createContext logEnv = Katip.runKatipContextT logEnv () namespace++ closeLog = liftIO . Katip.closeScribes+ in bracket createLogEnv closeLog $ inner . createContext++-- Streaming log++type LogStream = Conduit.ConduitT ByteString Conduit.Void IO ()++-- TODO: prepend katip-like format to each line+streamLog :: WithLog -> WS.Connection -> TMQueue.TMQueue ByteString -> IO ()+streamLog logger connection queue = do+ Conduit.runConduit $+ Conduit.sourceTMQueue queue+ .| Conduit.linesUnboundedAscii+ .| logLocal logger+ .| sendLog connection++streamLine :: LogStream -> ByteString -> IO ()+streamLine logStream msg = Conduit.connect (Conduit.yield $ "\n" <> msg <> "\n") logStream++logLocal :: WithLog -> Conduit.ConduitT ByteString ByteString IO ()+logLocal logger =+ Conduit.mapM $ \bs -> do+ logger . Katip.logLocM Katip.DebugS . Katip.ls $ bs+ return bs++sendLog :: WS.Connection -> LogStream+sendLog connection = Conduit.mapM_ $ \bs -> do+ now <- getCurrentTime+ WS.sendTextData connection $ Aeson.encode $ WSS.Log {WSS.line = toS bs, WSS.time = now}
src/Cachix/Deploy/OptionsParser.hs view
@@ -15,11 +15,11 @@ [ command "activate" $ info (helper <*> activate)- (progDesc "Deploy a new configuration to agents using CACHIX_ACTIVATE_TOKEN."),+ (progDesc "Deploy a new configuration to agents using CACHIX_ACTIVATE_TOKEN. See https://docs.cachix.org/deploy/deploying-to-agents"), command "agent" $ info (helper <*> agent)- (progDesc "Run an agent in foreground using CACHIX_AGENT_TOKEN.")+ (progDesc "Run an agent in foreground using CACHIX_AGENT_TOKEN. See https://docs.cachix.org/deploy/running-an-agent/") ] where activate = Activate <$> parserActivateOptions@@ -27,12 +27,14 @@ data AgentOptions = AgentOptions { name :: Text,- profile :: Maybe Text+ profile :: Maybe Text,+ bootstrap :: Bool } deriving (Show) data ActivateOptions = ActivateOptions- { payloadPath :: FilePath+ { payloadPath :: FilePath,+ agents :: [Text] } deriving (Show) @@ -41,14 +43,15 @@ AgentOptions <$> strArgument ( metavar "AGENT-NAME"- <> help "Unique identifier (usually hostname)."+ <> help "Unique agent identifier (usually hostname)." ) <*> optional ( strArgument ( metavar "NIX-PROFILE"- <> help "Nix profile to manage. Defaults to 'system' on NixOS and 'system-profiles/system' on nix-darwin."+ <> help "Nix profile to manage. Defaults to 'system' on NixOS, 'system-profiles/system' (nix-darwin) on macOS, and 'home-manager' for Home Manager." ) )+ <*> switch (long "bootstrap" <> help "Exit once the system agent takes over.") parserActivateOptions :: Parser ActivateOptions parserActivateOptions =@@ -56,4 +59,12 @@ <$> strArgument ( metavar "DEPLOY-SPEC.JSON" <> help "https://docs.cachix.org/deploy/reference.html#deploy-json"+ )+ <*> many+ ( strOption+ ( long "agent"+ <> short 'a'+ <> metavar "AGENT-NAME"+ <> help "Deploy only specific agent(s)."+ ) )
src/Cachix/Deploy/StdinProcess.hs view
@@ -3,12 +3,12 @@ import Protolude hiding (stdin) import System.IO (hClose) import System.Process-import Prelude (String, userError)+import Prelude (String) -- | Run a process with only stdin as an input-readProcess :: FilePath -> [String] -> String -> IO ()+readProcess :: FilePath -> [String] -> String -> IO ExitCode readProcess cmd args input = do- (Just stdin, _, _, ph) <-+ (Just stdin, _, _, pHandle) <- createProcess (proc cmd args) { std_in = CreatePipe,@@ -17,9 +17,8 @@ -- Same, but with posix api new_session = True }+ hPutStr stdin input hClose stdin- exitcode <- waitForProcess ph- case exitcode of- ExitSuccess -> return ()- ExitFailure code -> ioError $ userError $ "Process " <> show cmd <> " failed with exit code " <> show code++ waitForProcess pHandle
src/Cachix/Deploy/Websocket.hs view
@@ -1,176 +1,366 @@--- high level interface for websocket clients+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | A high-level, multiple reader, single writer interface for Websocket clients. module Cachix.Deploy.Websocket where -import Cachix.API.WebSocketSubprotocol (AgentInformation)-import qualified Cachix.API.WebSocketSubprotocol as WSS-import Cachix.Client.Retry+import qualified Cachix.Client.Retry as Retry+import qualified Cachix.Client.URI as URI import Cachix.Client.Version (versionNumber)+import qualified Cachix.Deploy.Log as Log import qualified Cachix.Deploy.WebsocketPong as WebsocketPong-import Control.Exception.Safe (catchAny, onException)-import Control.Retry (RetryStatus (..))-import Data.Aeson (FromJSON, ToJSON)-import Data.IORef+import qualified Control.Concurrent.Async as Async+import qualified Control.Concurrent.MVar as MVar+import qualified Control.Concurrent.STM.TBMQueue as TBMQueue+import qualified Control.Concurrent.STM.TMChan as TMChan+import Control.Exception.Safe (Handler (..), MonadMask, isSyncException)+import qualified Control.Exception.Safe as Safe+import qualified Control.Retry as Retry+import qualified Data.Aeson as Aeson+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BL+import Data.Conduit ((.|))+import qualified Data.Conduit as Conduit+import qualified Data.Conduit.Combinators as Conduit+import qualified Data.Conduit.TQueue as Conduit+import qualified Data.IORef as IORef import Data.String (String)+import Data.String.Here (iTrim)+import qualified Data.Text as Text+import qualified Data.Time.Clock as Time import qualified Katip as K-import Network.HTTP.Types (Header)+import qualified Network.HTTP.Simple as HTTP import qualified Network.WebSockets as WS-import Protolude hiding (catch, onException, toS)+import qualified Network.WebSockets.Connection as WS.Connection+import Protolude hiding (Handler, toS) import Protolude.Conv-import qualified System.Directory as Directory-import System.Environment (getEnv, lookupEnv) import qualified System.Info-import qualified System.Posix.Files as Posix.Files-import qualified System.Posix.User as Posix.User+import qualified System.Timeout as Timeout import qualified Wuss -type AgentState = IORef (Maybe WSS.AgentInformation)+-- | A reliable WebSocket connection that can be run ergonomically in a+-- separate thread.+--+-- Maintains the connection by periodically sending pings.+data WebSocket tx rx = WebSocket+ { -- | The active WebSocket connection, if available+ connection :: MVar.MVar WS.Connection,+ -- | The connection options+ options :: Options,+ -- | A timestamp of the last pong message received+ lastPong :: WebsocketPong.LastPongState,+ -- | See 'Transmit'+ tx :: Transmit tx,+ -- | See 'Receive'+ rx :: Receive rx,+ withLog :: Log.WithLog+ } data Options = Options- { host :: Text,+ { host :: URI.Host,+ port :: URI.Port, path :: Text,- name :: Text,- isVerbose :: Bool,- profile :: Text+ useSSL :: Bool,+ headers :: HTTP.RequestHeaders,+ -- | The identifier used when logging. Usually a combination of the agent+ -- name and the CLI version.+ identifier :: Text }- deriving (Show, Generic, ToJSON, FromJSON)+ deriving (Show) -data Input = Input- { deploymentDetails :: WSS.DeploymentDetails,- websocketOptions :: Options- }- deriving (Show, Generic, ToJSON, FromJSON)+-- | A more ergonomic version of the Websocket 'Message' data type+data Message msg+ = ControlMessage WS.ControlMessage+ | DataMessage msg -system :: String-system = System.Info.arch <> "-" <> System.Info.os+-- | A bounded queue of outbound messages.+type Transmit msg = TBMQueue.TBMQueue (Message msg) -runForever :: Options -> (ByteString -> (K.KatipContextT IO () -> IO ()) -> WS.Connection -> AgentState -> ByteString -> K.KatipContextT IO ()) -> IO ()-runForever options cmd =- runOnce options f+-- | A broadcast channel for incoming messages.+type Receive msg = TMChan.TMChan (Message msg)++-- | Send messages over the socket.+send :: WebSocket tx rx -> Message tx -> IO ()+send WebSocket {tx} = atomically . TBMQueue.writeTBMQueue tx++-- | Open a new receiving channel.+receive :: WebSocket tx rx -> IO (Receive rx)+receive WebSocket {rx} = atomically $ TMChan.dupTMChan rx++-- | Read incoming messages on a channel opened with 'receive'.+read :: Receive rx -> IO (Maybe (Message rx))+read = atomically . TMChan.readTMChan++-- | Read incoming data messages, ignoring everything else.+--+-- Stops reading when either a close request is received or the channel is+-- closed.+readDataMessages :: Receive rx -> (rx -> IO ()) -> IO ()+readDataMessages channel action = loop where- f runKatip connection agentState agentToken =- liftIO $- WSS.recieveDataConcurrently- connection- (\message -> runKatip (cmd message runKatip connection agentState agentToken))+ loop =+ read channel >>= \case+ Just (DataMessage message) -> action message *> loop+ Just (ControlMessage (WS.Close _ _)) -> pure ()+ Just (ControlMessage _) -> loop+ Nothing -> pure () -runOnce :: Options -> ((K.KatipContextT IO () -> IO ()) -> WS.Connection -> AgentState -> ByteString -> K.KatipContextT IO ()) -> IO ()-runOnce options cmd = withKatip (isVerbose options) $ \logEnv -> do- let runKatip = K.runKatipContextT logEnv () "agent"+-- | Close the outgoing queue.+drainQueue :: WebSocket tx rx -> IO ()+drainQueue WebSocket {tx} = atomically $ TBMQueue.closeTBMQueue tx - checkUserOwnsHome `catchAny` \e -> do- runKatip $ K.logLocM K.ErrorS $ K.ls (displayException e)- exitFailure+shutdownNow :: WebSocket tx rx -> Word16 -> BL.ByteString -> IO ()+shutdownNow websocket@WebSocket {rx} code msg = do+ -- Close the outgoing queue+ drainQueue websocket+ -- Signal to all receivers that the socket is closed+ atomically (TMChan.closeTMChan rx)+ throwIO $ WS.CloseRequest code msg - -- TODO: error if token is missing- agentToken <- getEnv "CACHIX_AGENT_TOKEN"- agentState <- newIORef Nothing+-- | Run an app inside a new WebSocket connection.+withConnection :: Log.WithLog -> Options -> (WebSocket tx rx -> IO ()) -> IO ()+withConnection withLog options app = do+ websocket <- new withLog options+ runConnection websocket (app websocket) - pongState <- WebsocketPong.newState- mainThreadID <- myThreadId+-- | Set up state for a new WebSocket connection. Use 'runConnection' to then+-- open the connection.+--+-- This is useful for setting up message processing using the tx/rx channels+-- before the connection is established. For example, you might want to use+-- 'receive' to open a receiving channel and capture incoming messages.+new :: Log.WithLog -> Options -> IO (WebSocket tx rx)+new withLog options = do+ connection <- MVar.newEmptyMVar+ tx <- TBMQueue.newTBMQueueIO 100+ rx <- TMChan.newBroadcastTMChanIO+ lastPong <- WebsocketPong.newState+ pure $ WebSocket {connection, options, tx, rx, lastPong, withLog} - let pingHandler = do- last <- WebsocketPong.secondsSinceLastPong pongState- runKatip $ K.logLocM K.DebugS $ K.ls $ "Sending WebSocket keep-alive ping, last pong was " <> (show last :: Text) <> " seconds ago"- WebsocketPong.pingHandler pongState mainThreadID pongTimeout- connectionOptions = WebsocketPong.installPongHandler pongState WS.defaultConnectionOptions- runKatip $- -- TODO: use exponential retry with reset: https://github.com/Soostone/retry/issues/25- retryAllWithLogging endlessConstantRetryPolicy (logger runKatip) $- do- K.logLocM K.InfoS $ K.ls ("Agent " <> agentIdentifier <> " connecting to " <> toS (host options) <> toS (path options))- liftIO $ do- -- refresh pong state in case we're reconnecting- WebsocketPong.pongHandler pongState- -- TODO: https://github.com/jaspervdj/websockets/issues/229- Wuss.runSecureClientWith (toS $ host options) 443 (toS $ path options) connectionOptions (headers options (toS agentToken)) $ \connection -> runKatip $ do- K.logLocM K.InfoS "Connected to Cachix Deploy service"- liftIO $- WS.withPingThread connection pingEvery pingHandler $- runKatip $ cmd runKatip connection agentState (toS agentToken)+runConnection :: WebSocket tx rx -> IO () -> IO ()+runConnection websocket@WebSocket {connection, options, tx, rx, withLog, lastPong} app = do+ threadId <- myThreadId++ -- TODO: store this in the WebSocket record+ let pingEvery = 30+ let pongTimeout = pingEvery * 2+ let onPing = do+ last <- WebsocketPong.secondsSinceLastPong lastPong+ withLog $ K.logLocM K.DebugS $ K.ls $ "Sending WebSocket keep-alive ping, last pong was " <> (show last :: Text) <> " seconds ago"+ WebsocketPong.pingHandler lastPong threadId pongTimeout+ let connectionOptions = WebsocketPong.installPongHandler lastPong WS.defaultConnectionOptions++ let dropConnection = void $ MVar.tryTakeMVar connection+ let closeChannels = atomically $ do+ TBMQueue.closeTBMQueue tx+ TMChan.closeTMChan rx++ flip Safe.finally closeChannels $+ reconnectWithLog withLog $ do+ withLog $ K.logLocM K.InfoS $ K.ls (logOnMessage options)++ -- TODO: https://github.com/jaspervdj/websockets/issues/229+ runClientWith options connectionOptions $+ \newConnection -> flip Safe.finally dropConnection $ do+ withLog $ K.logLocM K.InfoS "Connected to Cachix Deploy service"++ -- Reset the pong state in case we're reconnecting+ WebsocketPong.pongHandler lastPong++ -- Update the connection+ MVar.putMVar connection newConnection++ Async.concurrently_ (sendPingEvery pingEvery onPing websocket) app++runClientWith :: Options -> WS.Connection.ConnectionOptions -> WS.ClientApp a -> IO a+runClientWith Options {host, port, path, headers, useSSL} connectionOptions app =+ if useSSL+ then Wuss.runSecureClientWith hostS (fromIntegral (URI.portNumber port)) (toS path) connectionOptions headers app+ else WS.runClientWith hostS (URI.portNumber port) (toS path) connectionOptions headers app where- agentIdentifier = name options <> " " <> toS versionNumber- pingEvery = 30- pongTimeout = pingEvery * 2+ hostS = toS (URI.hostBS host) -headers :: Options -> ByteString -> [Header]-headers options agentToken =- [ ("Authorization", "Bearer " <> toS agentToken),- ("name", toS (name options)),- ("version", toS versionNumber),- ("system", toS system)- ]+-- Handle JSON messages --- TODO: log the exception-logger runKatip _ exception retryStatus =- runKatip $- K.logLocM K.ErrorS $ K.ls $ "Retrying in " <> delay (rsPreviousDelay retryStatus) <> " due to an exception: " <> displayException exception+-- | Start processing incoming and outgoing JSON messages.+--+-- Make sure to open an incoming channel with [receive] beforehand to avoid+-- dropping messages.+handleJSONMessages :: (Aeson.ToJSON tx, Aeson.FromJSON rx) => WebSocket tx rx -> IO () -> IO ()+handleJSONMessages websocket app =+ Async.withAsync (handleIncomingJSON websocket) $ \incomingThread ->+ Async.withAsync (handleOutgoingJSON websocket) $ \outgoingThread ->+ Async.withAsync+ ( do+ app+ drainQueue websocket+ Async.wait outgoingThread+ closeGracefully incomingThread+ )+ (\appThread -> mapM_ wait [appThread, outgoingThread, incomingThread]) where+ closeGracefully incomingThread = do+ repsonseToCloseRequest <- startGracePeriod $ do+ MVar.tryReadMVar (connection websocket) >>= \case+ Just activeConnection -> do+ WS.sendClose activeConnection ("Peer initiated a close request" :: ByteString)+ Async.wait incomingThread+ Nothing -> pure ()++ when (isNothing repsonseToCloseRequest) throwNoResponseToCloseRequest++handleIncomingJSON :: (Aeson.FromJSON rx) => WebSocket tx rx -> IO ()+handleIncomingJSON websocket@WebSocket {connection, rx, withLog} = do+ activeConnection <- MVar.readMVar connection+ let broadcast = atomically . TMChan.writeTMChan rx++ forever $ do+ msg <- WS.receive activeConnection+ case msg of+ WS.DataMessage _ _ _ am ->+ case Aeson.eitherDecodeStrict' (WS.fromDataMessage am :: ByteString) of+ Left e -> withLog $ K.logLocM K.DebugS . K.ls $ "Cannot parse websocket payload: " <> e+ Right pMsg -> broadcast (DataMessage pMsg)+ WS.ControlMessage controlMsg -> do+ case controlMsg of+ WS.Ping pl -> send websocket (ControlMessage (WS.Pong pl))+ WS.Pong _ -> WS.connectionOnPong (WS.Connection.connectionOptions activeConnection)+ WS.Close code closeMsg -> do+ hasSentClose <- IORef.readIORef $ WS.Connection.connectionSentClose activeConnection+ unless hasSentClose $ WS.send activeConnection msg+ shutdownNow websocket code closeMsg++ broadcast (ControlMessage controlMsg)++handleOutgoingJSON :: forall tx rx. Aeson.ToJSON tx => WebSocket tx rx -> IO ()+handleOutgoingJSON WebSocket {connection, tx} = do+ activeConnection <- MVar.readMVar connection+ Conduit.runConduit $+ Conduit.sourceTBMQueue tx+ .| Conduit.mapM_ (sendJSONMessage activeConnection)+ where+ sendJSONMessage :: WS.Connection -> Message tx -> IO ()+ sendJSONMessage conn (ControlMessage msg) = WS.send conn (WS.ControlMessage msg)+ sendJSONMessage conn (DataMessage msg) = WS.sendTextData conn (Aeson.encode msg)++-- | Log exceptions and retry, specialized for reconnecting WebSockets.+--+-- Close requests should be retried unless the status code is 1000, which+-- indicates that both the client and server have acknowledged the close+-- request and are ready to terminate the connection.+--+-- Other status codes typically indicate some sort of error. For example,+-- Cloudflare periodically restarts WebSockets and sends a pre-defined+-- status code in the 1xxx range.+--+-- Defined status codes:+-- https://www.rfc-editor.org/rfc/rfc6455.html#section-7.4.1+--+-- TODO: use exponential retry with reset: https://github.com/Soostone/retry/issues/25+-- TODO: clients should be able to decide which errors to recover from.+reconnectWithLog :: (MonadMask m, MonadIO m) => Log.WithLog -> m () -> m ()+reconnectWithLog withLog app =+ Safe.handle closeRequest $+ Retry.recovering Retry.endlessConstantRetryPolicy handlers (const app)+ where+ handlers = Retry.skipAsyncExceptions ++ [exitOnSuccess, exitOnCloseRequest, logSyncExceptions]++ closeRequest (WS.CloseRequest 1000 _) = return ()+ closeRequest e = Safe.throwM e++ exitOnSuccess _ = Handler $ \(_ :: ExitCode) -> return False++ exitOnCloseRequest _ = Handler $ \(e :: WS.ConnectionException) ->+ case e of+ WS.CloseRequest code msg -> do+ liftIO . withLog $+ K.logLocM K.DebugS . K.ls $+ "Received close request from peer (code: " <> show code <> ", message: " <> msg <> ")"++ -- Retry on any code other than 1000+ pure (code /= 1000)+ _ -> return True++ logSyncExceptions = Retry.logRetries (return . isSyncException) logRetries++ logRetries :: (MonadIO m) => Bool -> SomeException -> Retry.RetryStatus -> m ()+ logRetries _ exception retryStatus =+ liftIO . withLog $+ K.logLocM K.ErrorS . K.ls $+ "Retrying in " <> delay (Retry.rsPreviousDelay retryStatus) <> " due to an exception: " <> displayException exception+ delay :: Maybe Int -> String delay Nothing = "0 seconds"- delay (Just s) = show (floor (fromIntegral s / 1000 / 1000)) <> " seconds"+ delay (Just t) = show (toSeconds t) <> " seconds" -withKatip :: Bool -> (K.LogEnv -> IO a) -> IO a-withKatip isVerbose =- bracket createLogEnv K.closeScribes- where- permit = if isVerbose then K.DebugS else K.InfoS- createLogEnv = do- logEnv <- K.initLogEnv "agent" "production"- stdoutScribe <- K.mkHandleScribe K.ColorIfTerminal stdout (K.permitItem permit) K.V2- K.registerScribe "stdout" stdoutScribe K.defaultScribeSettings logEnv+ toSeconds :: Int -> Int+ toSeconds t =+ floor $ (fromIntegral t :: Double) / 1000 / 1000 -parseMessage :: FromJSON cmd => ByteString -> (WSS.Message cmd -> K.KatipContextT IO ()) -> K.KatipContextT IO ()-parseMessage payload m = do- case WSS.parseMessage payload of- (Left err) ->- -- TODO: show the bytestring?- K.logLocM K.ErrorS $ K.ls $ "Failed to parse websocket payload: " <> err- (Right message) ->- m message+waitForPong :: Int -> WebSocket tx rx -> IO (Maybe Time.UTCTime)+waitForPong seconds websocket = do+ channel <- receive websocket+ Async.withAsync (sendPingEvery 1 pass websocket) $ \_ ->+ Timeout.timeout (seconds * 1000 * 1000) $+ fix $ \waitForNextMsg -> do+ read channel >>= \case+ Just (ControlMessage (WS.Pong _)) -> Time.getCurrentTime+ _ -> waitForNextMsg --- commands+sendPingEvery :: Int -> IO () -> WebSocket tx rx -> IO ()+sendPingEvery seconds onPing WebSocket {connection} = forever $ do+ onPing+ activeConnection <- MVar.readMVar connection+ WS.sendPing activeConnection BS.empty+ threadDelay (seconds * 1000 * 1000) -registerAgent :: AgentState -> AgentInformation -> K.KatipContextT IO ()-registerAgent agentState agentInformation = do- K.logLocM K.InfoS "Agent registered."- liftIO $ atomicWriteIORef agentState (Just agentInformation)+startGracePeriod :: IO a -> IO (Maybe a)+startGracePeriod = Timeout.timeout (5 * 1000 * 1000) --- | Fetch the home directory and verify that the owner matches the current user.--- Throws either 'NoHomeFound' or 'UserDoesNotOwnHome'.-checkUserOwnsHome :: IO ()-checkUserOwnsHome = do- home <- Directory.getHomeDirectory `onException` throwIO NoHomeFound- stat <- Posix.Files.getFileStatus home- userId <- Posix.User.getEffectiveUserID+-- | Try to gracefully close the WebSocket.+--+-- Do not run with asynchronous exceptions masked, ie. Control.Exception.Safe.finally.+--+-- We send a close request to the peer and continue processing+-- any incoming messages until the server replies with its own+-- close control message.+waitForGracefulShutdown :: WS.Connection -> IO ()+waitForGracefulShutdown connection = do+ WS.sendClose connection ("Closing." :: ByteString) - when (userId /= Posix.Files.fileOwner stat) $ do- userName <- Posix.User.userName <$> Posix.User.getUserEntryForID userId- sudoUser <- lookupEnv "SUDO_USER"- throwIO $- UserDoesNotOwnHome- { userName = userName,- sudoUser = sudoUser,- home = home- }+ -- Grace period+ response <- startGracePeriod $ forever (WS.receiveDataMessage connection) -data Error- = -- | No home directory.- NoHomeFound- | -- | Safeguard against creating root-owned files in user directories.- -- This is an issue on macOS, where, by default, sudo does not reset $HOME.- UserDoesNotOwnHome- { userName :: String,- sudoUser :: Maybe String,- home :: FilePath- }- deriving (Show)+ when (isNothing response) throwNoResponseToCloseRequest -instance Exception Error where- displayException NoHomeFound = "Could not find the user’s home directory. Make sure to set the $HOME variable."- displayException UserDoesNotOwnHome {userName = userName, sudoUser = sudoUser, home = home} =- if isJust sudoUser- then toS $ unlines [warningMessage, suggestSudoFlagH]- else toS warningMessage- where- warningMessage = "The current user (" <> toS userName <> ") does not own the home directory (" <> toS home <> ")"- suggestSudoFlagH = "Try running the agent with `sudo -H`."+throwNoResponseToCloseRequest :: IO a+throwNoResponseToCloseRequest = throwIO $ WS.CloseRequest 1000 "No response to close request"++-- Authorization headers for Cachix Deploy++system :: String+system = System.Info.arch <> "-" <> System.Info.os++createHeaders ::+ -- | Agent name+ Text ->+ -- | Agent Token+ Text ->+ HTTP.RequestHeaders+createHeaders agentName agentToken =+ [ ("Authorization", "Bearer " <> toS agentToken),+ ("name", toS agentName),+ ("version", toS versionNumber),+ ("system", toS system)+ ]++logOnMessage :: Options -> Text+logOnMessage Options {host, identifier, path, useSSL} =+ [iTrim|+ ${Text.toTitle identifier} connecting to ${uri} over ${protocol}+ |]+ where+ uri = decodeUtf8 (URI.hostBS host) <> path++ protocol :: Text+ protocol = if useSSL then "HTTPS" else "HTTP"
src/Cachix/Deploy/WebsocketPong.hs view
@@ -2,7 +2,8 @@ -- TODO: upstream to https://github.com/jaspervdj/websockets/issues/159 module Cachix.Deploy.WebsocketPong where -import Data.IORef+import Data.IORef (IORef)+import qualified Data.IORef as IORef import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDiffTimeToSeconds) import qualified Network.WebSockets as WS import Protolude@@ -18,25 +19,25 @@ newState :: IO LastPongState newState = do now <- getCurrentTime- newIORef now+ IORef.newIORef now -- everytime we send a ping we check if we also got a pong back pingHandler :: LastPongState -> ThreadId -> Int -> IO ()-pingHandler state threadID maxLastPing = do- last <- secondsSinceLastPong state- when (last > maxLastPing) $ do+pingHandler pong threadID maxLastPing = do+ last <- secondsSinceLastPong pong+ when (last > maxLastPing) $ throwTo threadID WebsocketPongTimeout secondsSinceLastPong :: LastPongState -> IO Int-secondsSinceLastPong state = do+secondsSinceLastPong pong = do+ last <- IORef.readIORef pong now <- getCurrentTime- last <- readIORef state return $ ceiling $ nominalDiffTimeToSeconds $ diffUTCTime now last pongHandler :: LastPongState -> IO ()-pongHandler state = do+pongHandler pong = do now <- getCurrentTime- writeIORef state now+ void $ IORef.atomicWriteIORef pong now installPongHandler :: LastPongState -> WS.ConnectionOptions -> WS.ConnectionOptions-installPongHandler state opts = opts {WS.connectionOnPong = pongHandler state}+installPongHandler pong opts = opts {WS.connectionOnPong = pongHandler pong}
+ test/URISpec.hs view
@@ -0,0 +1,87 @@+module URISpec where++import qualified Cachix.Client.URI as URI+import qualified Data.Aeson as Aeson+import qualified Data.ByteString.Char8 as Char8+import qualified Data.ByteString.Lazy as BL+import Data.Either.Extra+import qualified Dhall+import qualified Dhall.Core+import Protolude+import qualified Servant.Client.Core as Servant+import Test.Hspec++secureScheme :: ByteString+secureScheme = "https"++unsecureScheme :: ByteString+unsecureScheme = "http"++host :: ByteString+host = "cachix.org"++subdomain :: Text+subdomain = "foo"++secureUri :: ByteString+secureUri = secureScheme <> "://" <> host++unsecureUri :: ByteString+unsecureUri = unsecureScheme <> "://" <> host++secureUriWithSubdomain :: ByteString+secureUriWithSubdomain = secureScheme <> "://" <> encodeUtf8 subdomain <> "." <> host++-- A helper that throws an error when parsing fails.+parseURI' :: ByteString -> URI.URI+parseURI' = fromRight' . URI.parseURI++spec :: Spec+spec =+ describe "URI" $ do+ it "parses a URI" $+ URI.parseURI secureUri `shouldSatisfy` isRight++ it "re-serializes a URI" $+ URI.serialize <$> URI.parseURI secureUri `shouldBe` Right secureUri++ it "appends a subdomain" $+ let parsedURI = parseURI' secureUri+ newURI = URI.appendSubdomain subdomain parsedURI+ in URI.serialize newURI `shouldBe` secureUriWithSubdomain++ it "returns the hostname" $+ URI.getHostname (parseURI' secureUri) `shouldBe` URI.Host host++ it "returns port 80 for HTTP URIs" $+ let scheme = URI.getScheme (parseURI' unsecureUri)+ in URI.getPortFor scheme `shouldBe` Just (URI.Port 80)++ it "returns port 443 for HTTPS URIs" $+ let scheme = URI.getScheme (parseURI' secureUri)+ in URI.getPortFor scheme `shouldBe` Just (URI.Port 443)++ it "detects if SSL is required" $+ let getScheme = URI.getScheme . parseURI'+ in do+ URI.requiresSSL (getScheme secureUri) `shouldBe` True+ URI.requiresSSL (getScheme unsecureUri) `shouldBe` False++ it "converts to JSON" $+ Aeson.encode (parseURI' secureUri) `shouldBe` "\"" <> BL.fromStrict secureUri <> "\""++ it "converts from JSON" $+ Aeson.decodeStrict' ("\"" <> secureUri <> "\"") `shouldBe` Just (parseURI' secureUri)++ it "converts to Dhall" $+ let asDhall = Dhall.embed Dhall.inject (parseURI' secureUri)+ in Dhall.Core.pretty asDhall `shouldBe` "\"" <> decodeUtf8 secureUri <> "\""++ it "converts from Dhall" $+ Dhall.input Dhall.auto ("\"" <> decodeUtf8 secureUri <> "\"") `shouldReturn` parseURI' secureUri++ -- https://github.com/cachix/cachix/issues/462+ it "converts to a Servant BaseUrl without trailing slashes" $+ let uriWithTrailingSlash = secureUri <> "/"+ uri = parseURI' uriWithTrailingSlash+ in Servant.baseUrlPath (URI.getBaseUrl uri) `shouldBe` ""