cachix 0.8.1 → 1.0.0
raw patch · 10 files changed
+346/−142 lines, 10 filesdep +extradep +lukkodep +stm-chans
Dependencies added: extra, lukko, stm-chans
Files
- CHANGELOG.md +12/−0
- cachix-deployment/Main.hs +33/−20
- cachix.cabal +8/−3
- src/Cachix/Client/PushQueue.hs +22/−16
- src/Cachix/Client/WatchStore.hs +1/−2
- src/Cachix/Deploy/Activate.hs +142/−75
- src/Cachix/Deploy/Agent.hs +11/−10
- src/Cachix/Deploy/Lock.hs +37/−0
- src/Cachix/Deploy/OptionsParser.hs +6/−5
- src/Cachix/Deploy/Websocket.hs +74/−11
CHANGELOG.md view
@@ -7,6 +7,18 @@ ## Unreleased +# [1.0.0] - 2022-09-06++- Cachix Deploy: auto rollback if the agent can't connect to the backend service anymore++- Cachix Deploy: allow specifying `rollbackScript`: https://docs.cachix.org/deploy/reference++- Cachix Deploy: report `system` and closure size from the agent++- Cachix Deploy: lock deployments so there's only one active at the time++- Cachix Deploy: disable negative narinfo caching+ ## [0.8.1] - 2022-07-26 - Cachix Deploy: retry exceptions every 1s instead of exponentially
cachix-deployment/Main.hs view
@@ -9,10 +9,12 @@ import qualified Cachix.API.WebSocketSubprotocol as WSS import Cachix.Client.Retry import qualified Cachix.Deploy.Activate as Activate+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 Control.Concurrent.Async as Async-import qualified Control.Concurrent.STM.TQueue as TQueue+import qualified Control.Concurrent.STM.TMQueue as TMQueue import qualified Data.Aeson as Aeson import qualified Data.Conduit as Conduit import qualified Data.Conduit.Combinators as Conduit@@ -30,13 +32,20 @@ import System.IO (BufferMode (..), hSetBuffering) import qualified Wuss +-- | Activate the new deployment.+--+-- If the target profile is already locked by another deployment, exit+-- immediately and rely on the backend to reschedule. main :: IO () main = do setLocaleEncoding utf8 hSetBuffering stdout LineBuffering hSetBuffering stderr LineBuffering+ input <- escalateAs (FatalError . toS) . Aeson.eitherDecode . toS =<< getContents- CachixWebsocket.runForever (CachixWebsocket.websocketOptions input) (handleMessage input)+ let profile = CachixWebsocket.profile . Input.websocketOptions $ input+ void . Lock.withTryLock profile $+ CachixWebsocket.runForever (CachixWebsocket.websocketOptions input) (handleMessage input) handleMessage :: CachixWebsocket.Input -> ByteString -> (K.KatipContextT IO () -> IO ()) -> WS.Connection -> CachixWebsocket.AgentState -> ByteString -> K.KatipContextT IO () handleMessage input payload runKatip connection _ agentToken =@@ -44,35 +53,39 @@ where deploymentDetails = CachixWebsocket.deploymentDetails input options = CachixWebsocket.websocketOptions input+ 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 TQueue.newTQueue+ 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 $ do- Activate.activate options connection (Conduit.sinkTQueue queue) deploymentDetails agentInformation agentToken- liftIO $ Async.race_ streamingThread activateThread- throwIO ExitSuccess- runLogStreaming :: String -> RequestHeaders -> Conduit.TQueue ByteString -> UUID -> IO ()+ 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++ 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 ->- bracket_ (return ()) (WS.sendClose connection ("Closing." :: ByteString)) $- Conduit.runConduit $- Conduit.sourceTQueue queue- .| Conduit.linesUnboundedAscii- -- TODO: prepend katip-like format to each line- -- .| (if CachixWebsocket.isVerbose options then Conduit.print else mempty)- .| sendLog 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 sendLog :: WS.Connection -> Conduit.ConduitT ByteString Conduit.Void IO ()-sendLog connection = Conduit.mapM_ f- where- f = \bs -> do- now <- getCurrentTime- WS.sendTextData connection $ Aeson.encode $ WSS.Log {WSS.line = toS bs, WSS.time = now}+sendLog connection = Conduit.mapM_ $ \bs -> do+ now <- getCurrentTime+ WS.sendTextData connection $ Aeson.encode $ WSS.Log {WSS.line = toS bs, WSS.time = now}
cachix.cabal view
@@ -1,12 +1,12 @@ cabal-version: 2.2 name: cachix-version: 0.8.1+version: 1.0.0 license: Apache-2.0 license-file: LICENSE-copyright: 2018 Domen Kožar+copyright: 2018 Domen Kozar category: Nix maintainer: domen@cachix.org-author: Domen Kožar+author: Domen Kozar homepage: https://github.com/cachix/cachix#readme bug-reports: https://github.com/cachix/cachix/issues synopsis:@@ -24,6 +24,7 @@ NoImplicitPrelude DeriveAnyClass DeriveGeneric+ DerivingVia OverloadedStrings ghc-options:@@ -65,6 +66,7 @@ Cachix.Deploy.Activate Cachix.Deploy.ActivateCommand Cachix.Deploy.Agent+ Cachix.Deploy.Lock Cachix.Deploy.OptionsParser Cachix.Deploy.StdinProcess Cachix.Deploy.Websocket@@ -90,6 +92,7 @@ , dhall >=1.28.0 , directory , ed25519+ , extra , filepath , fsnotify , hercules-ci-cnix-store@@ -100,6 +103,7 @@ , http-types , inline-c-cpp , katip+ , lukko , lzma-conduit , megaparsec >=7.0.0 , memory@@ -169,6 +173,7 @@ , protolude , safe-exceptions , stm+ , stm-chans , stm-conduit , time , uuid
src/Cachix/Client/PushQueue.hs view
@@ -1,10 +1,10 @@ {- Implements a queue with the following properties: -- waits for queue to be fully pushed when exiting using ctrl-c (SIGINT)+- waits for the queue to be fully pushed when exiting using ctrl-c (SIGINT) - allows stopping the producer-- avoid duplicate pushing of the same store paths+- avoids pushing duplicate store paths -To safetly exit on demand, signal SIGINT.+Use SIGINT to safely exit on demand. -} module Cachix.Client.PushQueue ( startWorkers,@@ -16,6 +16,7 @@ import qualified Cachix.Client.Push as Push import Cachix.Client.Retry (retryAll) import Control.Concurrent.Async+import Control.Concurrent.Extra (once) import Control.Concurrent.STM (TVar, modifyTVar', newTVarIO, readTVar) import qualified Control.Concurrent.STM.Lock as Lock import qualified Control.Concurrent.STM.TBQueue as TBQueue@@ -41,14 +42,14 @@ worker :: Push.PushParams IO () -> PushWorkerState -> IO () worker pushParams workerState = forever $ do storePath <- atomically $ TBQueue.readTBQueue $ pushQueue workerState- bracket_ (inProgresModify (+ 1)) (inProgresModify (\x -> x - 1)) $+ bracket_ (inProgressModify (+ 1)) (inProgressModify (\x -> x - 1)) $ retryAll $- \retrystatus ->- void $ do- maybeStorePath <- filterInvalidStorePath (Push.pushParamsStore pushParams) storePath- for maybeStorePath $ \validatedStorePath -> Push.uploadStorePath pushParams validatedStorePath retrystatus+ \retrystatus -> do+ maybeStorePath <- filterInvalidStorePath (Push.pushParamsStore pushParams) storePath+ for_ maybeStorePath $ \validatedStorePath ->+ Push.uploadStorePath pushParams validatedStorePath retrystatus where- inProgresModify f =+ inProgressModify f = atomically $ modifyTVar' (inProgress workerState) f -- NOTE: producer is responsible for signaling SIGINT upon termination@@ -66,7 +67,9 @@ progress <- newTVarIO 0 let pushWorkerState = PushWorkerState newPushQueue progress pushWorker <- async $ replicateConcurrently_ numWorkers $ worker pushParams pushWorkerState- let signalset = Signals.CatchOnce (exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState)+ let signalset =+ Signals.CatchOnce $+ exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState void $ Signals.installHandler Signals.sigINT signalset Nothing void $ Signals.installHandler Signals.sigTERM signalset Nothing (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker]@@ -99,12 +102,15 @@ return (missingStorePathsSet, alreadyQueuedSet) queryLoop (workerState {alreadyQueued = S.union missingStorePathsSet alreadyQueuedSet}) pushqueue pushParams +-- | Stop watching the store and push all pending store paths.+-- This should only be used once per process. exitOnceQueueIsEmpty :: IO () -> Async () -> Async () -> QueryWorkerState -> PushWorkerState -> IO ()-exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState = do- putTextError "Stopped watching /nix/store and waiting for queue to empty ..."- Systemd.notifyStopping- stopProducerCallback- go+exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState =+ join . once $ do+ putTextError "Stopped watching /nix/store and waiting for queue to empty ..."+ void Systemd.notifyStopping+ stopProducerCallback+ go where go = do (isDone, inprogress, queueLength) <- atomically $ do@@ -121,7 +127,7 @@ cancelWith pushWorker StopWorker else do -- extend shutdown for another 90s- Systemd.notify False $ "EXTEND_TIMEOUT_USEC=" <> show (90 * 1000 * 1000)+ void $ Systemd.notify False $ "EXTEND_TIMEOUT_USEC=" <> show (90 * 1000 * 1000) putTextError $ "Waiting to finish: " <> show inprogress <> " pushing, " <> show queueLength <> " in queue" threadDelay (1000 * 1000) go
src/Cachix/Client/WatchStore.hs view
@@ -6,7 +6,6 @@ import Cachix.Client.Push import qualified Cachix.Client.PushQueue as PushQueue import qualified Control.Concurrent.STM.TBQueue as TBQueue-import Data.List (isSuffixOf) import Hercules.CNix.Store (Store) import qualified Hercules.CNix.Store as Store import Protolude@@ -15,7 +14,7 @@ startWorkers :: Store -> Int -> PushParams IO () -> IO () startWorkers store numWorkers pushParams = do- Systemd.notifyReady+ void Systemd.notifyReady withManager $ \mgr -> PushQueue.startWorkers numWorkers (producer store mgr) pushParams producer :: Store -> WatchManager -> PushQueue.Queue -> IO (IO ())
src/Cachix/Deploy/Activate.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DuplicateRecordFields #-} module Cachix.Deploy.Activate where@@ -8,19 +9,28 @@ import qualified Cachix.Deploy.Websocket as CachixWebsocket import qualified Cachix.Types.BinaryCache as BinaryCache import Cachix.Types.Permission (Permission (..))+import qualified Data.Aeson as Aeson+#if MIN_VERSION_aeson(2,0,0)+import qualified Data.Aeson.KeyMap as HM+#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 (doesPathExist)+import System.Directory (canonicalizePath, doesPathExist)+import System.FilePath ((</>)) import System.IO.Temp (withSystemTempDirectory) import System.Process+import System.Timeout (timeout) import Prelude (String) domain :: CachixWebsocket.Options -> WSS.Cache -> Text@@ -30,6 +40,9 @@ uri :: CachixWebsocket.Options -> WSS.Cache -> Text uri options cache = "https://" <> domain options cache +hackFlush :: K.KatipContextT IO ()+hackFlush = liftIO $ threadDelay (5 * 1000 * 1000)+ -- TODO: what if websocket gets closed while deploying? activate :: CachixWebsocket.Options ->@@ -39,70 +52,86 @@ WSS.AgentInformation -> ByteString -> K.KatipContextT IO ()-activate options connection sourceStream deploymentDetails agentInfo agentToken = do+activate options connection sourceStream deploymentDetails agentInfo agentToken = withCacheArgs options agentInfo agentToken $ \cacheArgs -> do let storePath = WSS.storePath deploymentDetails- 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)]- sigs = ["--option", "trusted-public-keys", officialCache <> " " <> toS (domain options cache) <> "-1:" <> toS (WSS.publicKey cache)]- in substituters ++ sigs- Nothing -> []- deploymentID = WSS.id (deploymentDetails :: WSS.DeploymentDetails)- deploymentFinished hasSucceeded now =- WSS.DeploymentFinished- { WSS.id = deploymentID,- WSS.time = now,- WSS.hasSucceeded = hasSucceeded- }- deploymentFailed = do- now <- liftIO getCurrentTime- K.logLocM K.InfoS $ K.ls $ "Deploying #" <> index <> " failed."- sendMessage $ deploymentFinished False now- -- hack to flush logs- liftIO $ threadDelay (5 * 1000 * 1000) K.logLocM K.InfoS $ K.ls $ "Deploying #" <> index <> ": " <> WSS.storePath deploymentDetails- -- notify the service deployment started- now <- liftIO getCurrentTime- sendMessage $- WSS.DeploymentStarted- { WSS.id = deploymentID,- WSS.time = now- }- -- TODO: don't create tmpfile for public caches+ deploymentStarted Nothing -- TODO: add GC root so it's preserved for the next command -- get the store path using caches- (downloadExitCode, _, _) <- liftIO $- withSystemTempDirectory "netrc" $- \dir -> do- let filepath = dir <> "netrc"- args <- case WSS.cache agentInfo of- Just cache -> do- -- TODO: ugh- let bc =- BinaryCache.BinaryCache- { BinaryCache.name = "",- BinaryCache.uri = toS (uri options cache),- BinaryCache.publicSigningKeys = [],- BinaryCache.isPublic = WSS.isPublic cache,- BinaryCache.githubUsername = "",- BinaryCache.permission = Read- }- NetRc.add (Token agentToken) [bc] filepath- return $ cachesArgs <> ["--option", "netrc-file", filepath]- Nothing ->- return cachesArgs- shellOut "nix-store" (["-r", toS storePath] <> args)+ (downloadExitCode, _, _) <-+ liftIO $ shellOut "nix-store" (["-r", toS storePath] <> cacheArgs) -- TODO: use exceptions to simplify this code case downloadExitCode of ExitFailure _ -> deploymentFailed ExitSuccess -> do- (profile, activationScripts) <- liftIO $ getActivationScript storePath (CachixWebsocket.profile options)- -- TODO: document what happens if wrong user is used for the agent+ -- 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) - -- set the new profile- (activateProfileExitCode, _, _) <- liftIO $ shellOut "nix-env" ["-p", toS profile, "--set", toS storePath]+ 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+ 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@@ -112,19 +141,17 @@ return activateScriptExitCode if not (all (== ExitSuccess) exitCodes) then deploymentFailed- else do- now <- liftIO getCurrentTime- sendMessage $ deploymentFinished True now- liftIO $ log "Successfully activated the deployment."- -- TODO: this is a hack to make sure the deployment is finished- liftIO $ threadDelay (5 * 1000 * 1000)- K.logLocM K.InfoS $ K.ls $ "Deployment #" <> index <> " finished"- where- -- TODO: prevent service from being restarted while deploying- -- TODO: upgrade agent-+ 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@@ -146,25 +173,65 @@ method = case command of WSS.DeploymentStarted {} -> "DeploymentStarted" WSS.DeploymentFinished {} -> "DeploymentFinished"- getActivationScript :: Text -> Text -> IO (Text, [Command])+ -- 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"+ 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, _) -> 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", [])+ (toS storePath </> "activate-user", []),+ (toS storePath </> "activate", []) ] ) (_, _) -> return ("system", [])- return ("/nix/var/nix/profiles/" <> if profile == "" then systemProfile else profile, cmds)+ return ("/nix/var/nix/profiles" </> if profile == "" then systemProfile else toS profile, cmds) type Command = (String, [String]) --- TODO: home-manager+extractClosureSize :: Aeson.Value -> Maybe Int64+extractClosureSize (Aeson.Array vector) = case Vector.toList vector of+ [Aeson.Object obj] -> case HM.lookup "closureSize" obj of+ Just (Aeson.Number num) -> Just $ floor num+ _ -> Nothing+ _ -> Nothing+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 =+ withSystemTempDirectory "netrc" $ \dir -> do+ let filepath = dir </> "netrc"+ args <- case WSS.cache agentInfo of+ Just cache -> do+ -- TODO: ugh+ let bc =+ BinaryCache.BinaryCache+ { BinaryCache.name = "",+ BinaryCache.uri = toS (uri options cache),+ BinaryCache.publicSigningKeys = [],+ BinaryCache.isPublic = WSS.isPublic cache,+ BinaryCache.githubUsername = "",+ BinaryCache.permission = Read+ }+ liftIO $ NetRc.add (Token agentToken) [bc] filepath+ return $ cachesArgs <> ["--option", "netrc-file", filepath]+ Nothing ->+ return cachesArgs+ 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 -> []
src/Cachix/Deploy/Agent.hs view
@@ -17,29 +17,32 @@ import qualified Servant.Client as Servant run :: CachixOptions.CachixOptions -> AgentOptions.AgentOptions -> IO ()-run cachixOptions agentOpts = do+run cachixOptions agentOpts = CachixWebsocket.runForever options handleMessage where host = toS $ Servant.baseUrlHost $ getBaseUrl $ CachixOptions.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 = AgentOptions.profile agentOpts,+ CachixWebsocket.profile = profile, CachixWebsocket.isVerbose = CachixOptions.verbose cachixOptions } handleMessage :: ByteString -> (K.KatipContextT IO () -> IO ()) -> WS.Connection -> CachixWebsocket.AgentState -> ByteString -> K.KatipContextT IO ()- handleMessage payload _ _ agentState _ = do+ handleMessage payload _ _ agentState _ = CachixWebsocket.parseMessage payload (handleCommand . WSS.command) where handleCommand :: WSS.BackendCommand -> K.KatipContextT IO ()- handleCommand (WSS.AgentRegistered agentInformation) = do+ handleCommand (WSS.AgentRegistered agentInformation) = CachixWebsocket.registerAgent agentState agentInformation- handleCommand (WSS.Deployment deploymentDetails) = do- -- TODO: lock to ensure one deployment at the time- let input =+ handleCommand (WSS.Deployment deploymentDetails) =+ liftIO $ do+ binDir <- toS <$> getBinDir+ readProcess (binDir <> "/.cachix-deployment") [] $+ toS . Aeson.encode $ CachixWebsocket.Input { deploymentDetails = deploymentDetails, websocketOptions =@@ -47,9 +50,7 @@ { host = host, name = name, path = "/ws-deployment",- profile = AgentOptions.profile agentOpts,+ profile = profile, isVerbose = CachixOptions.verbose cachixOptions } }- binDir <- toS <$> liftIO getBinDir- liftIO $ readProcess (binDir <> "/.cachix-deployment") [] (toS $ Aeson.encode input)
+ src/Cachix/Deploy/Lock.hs view
@@ -0,0 +1,37 @@+module Cachix.Deploy.Lock (withTryLock) where++import qualified Lukko as Lock+import Protolude hiding ((<.>))+import qualified System.Directory as Directory+import System.FilePath ((<.>), (</>))++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+ lockDirectory <- Directory.getXdgDirectory Directory.XdgCache defaultLockDirectory++ Directory.createDirectoryIfMissing True lockDirectory+ Directory.setPermissions lockDirectory $+ Directory.emptyPermissions+ & Directory.setOwnerReadable True+ & Directory.setOwnerWritable True+ & Directory.setOwnerExecutable True+ & Directory.setOwnerSearchable True++ let lockFile = lockDirectory </> toS profileName <.> "lock"++ bracket+ (Lock.fdOpen lockFile)+ (Lock.fdUnlock *> Lock.fdClose)+ $ \fd -> do+ isLocked <- Lock.fdTryLock fd Lock.ExclusiveLock+ if isLocked+ then Just <$> action+ else pure Nothing
src/Cachix/Deploy/OptionsParser.hs view
@@ -27,7 +27,7 @@ data AgentOptions = AgentOptions { name :: Text,- profile :: Text+ profile :: Maybe Text } deriving (Show) @@ -43,10 +43,11 @@ ( metavar "AGENT-NAME" <> help "Unique identifier (usually hostname)." )- <*> strArgument- ( value ""- <> metavar "NIX-PROFILE"- <> help "Nix profile to manage. Defaults to 'system' on NixOS and 'system-profiles/system' on nix-darwin."+ <*> optional+ ( strArgument+ ( metavar "NIX-PROFILE"+ <> help "Nix profile to manage. Defaults to 'system' on NixOS and 'system-profiles/system' on nix-darwin."+ ) ) parserActivateOptions :: Parser ActivateOptions
src/Cachix/Deploy/Websocket.hs view
@@ -6,6 +6,7 @@ import Cachix.Client.Retry import Cachix.Client.Version (versionNumber) 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@@ -13,9 +14,13 @@ import qualified Katip as K import Network.HTTP.Types (Header) import qualified Network.WebSockets as WS-import Protolude hiding (toS)+import Protolude hiding (catch, onException, toS) import Protolude.Conv-import System.Environment (getEnv)+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 Wuss type AgentState = IORef (Maybe WSS.AgentInformation)@@ -35,15 +40,35 @@ } deriving (Show, Generic, ToJSON, FromJSON) +system :: String+system = System.Info.arch <> "-" <> System.Info.os+ runForever :: Options -> (ByteString -> (K.KatipContextT IO () -> IO ()) -> WS.Connection -> AgentState -> ByteString -> K.KatipContextT IO ()) -> IO ()-runForever options cmd = withKatip (isVerbose options) $ \logEnv -> do- agentToken <- getEnv "CACHIX_AGENT_TOKEN"+runForever options cmd =+ runOnce options f+ where+ f runKatip connection agentState agentToken =+ liftIO $+ WSS.recieveDataConcurrently+ connection+ (\message -> runKatip (cmd message runKatip connection agentState agentToken))++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"++ checkUserOwnsHome `catchAny` \e -> do+ runKatip $ K.logLocM K.ErrorS $ K.ls (displayException e)+ exitFailure+ -- TODO: error if token is missing+ agentToken <- getEnv "CACHIX_AGENT_TOKEN" agentState <- newIORef Nothing+ pongState <- WebsocketPong.newState mainThreadID <- myThreadId- let runKatip = K.runKatipContextT logEnv () "agent"- pingHandler = do++ 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@@ -61,10 +86,7 @@ K.logLocM K.InfoS "Connected to Cachix Deploy service" liftIO $ WS.withPingThread connection pingEvery pingHandler $- do- WSS.recieveDataConcurrently- connection- (\message -> runKatip (cmd message runKatip connection agentState (toS agentToken)))+ runKatip $ cmd runKatip connection agentState (toS agentToken) where agentIdentifier = name options <> " " <> toS versionNumber pingEvery = 30@@ -74,7 +96,8 @@ headers options agentToken = [ ("Authorization", "Bearer " <> toS agentToken), ("name", toS (name options)),- ("version", toS versionNumber)+ ("version", toS versionNumber),+ ("system", toS system) ] -- TODO: log the exception@@ -111,3 +134,43 @@ registerAgent agentState agentInformation = do K.logLocM K.InfoS "Agent registered." liftIO $ atomicWriteIORef agentState (Just agentInformation)++-- | 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+ = -- | 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 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`."