diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ ## Unreleased +## [0.7.0] - 2022-01-12 + +### Added + +- Cachix Deploy support + +### Fixed + +- #386: use /run/current-system/nixos-version to check if we're running NixOS +- Filter out invalid paths when pushing + +### Changed + +- watch-store command: use systemd notifications while shutting down + ## [0.6.1] - 2021-06-21 ### Fixed diff --git a/cachix.cabal b/cachix.cabal --- a/cachix.cabal +++ b/cachix.cabal @@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix -version: 0.6.1 +version: 0.7.0 license: Apache-2.0 license-file: LICENSE copyright: 2018 Domen Kožar @@ -42,6 +42,7 @@ import: defaults exposed-modules: Cachix.Client + Cachix.Client.CNix Cachix.Client.Commands Cachix.Client.Config Cachix.Client.Config.Orphans @@ -59,13 +60,19 @@ Cachix.Client.Secrets Cachix.Client.Servant Cachix.Client.URI + Cachix.Client.Version Cachix.Client.WatchStore + Cachix.Deploy.Activate + Cachix.Deploy.ActivateCommand + Cachix.Deploy.Agent + Cachix.Deploy.OptionsParser System.Nix.Base32 hs-source-dirs: src other-modules: Paths_cachix autogen-modules: Paths_cachix build-depends: + , aeson , async , base >=4.7 && <5 , base64-bytestring @@ -88,12 +95,15 @@ , http-client-tls , http-conduit , http-types + , inline-c-cpp + , katip , lzma-conduit , megaparsec >=7.0.0 , memory , mmorph , netrc , optparse-applicative + , pretty-terminal , process , protolude , resourcet @@ -106,11 +116,19 @@ , servant-client-core >=0.16 , servant-conduit , stm + , stm-conduit + , systemd + , temporary , text + , time , unix + , unordered-containers , uri-bytestring + , uuid , vector , versions + , websockets + , wuss pkgconfig-depends: nix-store ==2.0 || >2.0, nix-main ==2.0 || >2.0 @@ -123,9 +141,10 @@ other-modules: Paths_cachix autogen-modules: Paths_cachix build-depends: - , base >=4.7 && <5 + , base >=4.7 && <5 , cachix , cachix-api + , safe-exceptions test-suite cachix-test import: defaults diff --git a/cachix/Main.hs b/cachix/Main.hs --- a/cachix/Main.hs +++ b/cachix/Main.hs @@ -2,7 +2,7 @@ import qualified Cachix.Client as CC import Cachix.Client.Exception (CachixException) -import Control.Exception (displayException, handle) +import Control.Exception.Safe (displayException, handle) import GHC.IO.Encoding import System.Exit (exitFailure) import System.IO diff --git a/src/Cachix/Client.hs b/src/Cachix/Client.hs --- a/src/Cachix/Client.hs +++ b/src/Cachix/Client.hs @@ -4,8 +4,12 @@ where import Cachix.Client.Commands as Commands -import Cachix.Client.Env (cachixVersion, mkEnv) +import Cachix.Client.Env (mkEnv) import Cachix.Client.OptionsParser (CachixCommand (..), getOpts) +import Cachix.Client.Version (cachixVersion) +import Cachix.Deploy.ActivateCommand as ActivateCommand +import qualified Cachix.Deploy.Agent as AgentCommand +import qualified Cachix.Deploy.OptionsParser as DeployOptions import Protolude main :: IO () @@ -20,3 +24,7 @@ WatchExec pushArgs name cmd args -> Commands.watchExec env pushArgs name cmd args Use name useOptions -> Commands.use env name useOptions Version -> putText cachixVersion + DeployCommand deployCommand -> + case deployCommand of + DeployOptions.Agent opts -> AgentCommand.run cachixoptions opts + DeployOptions.Activate opts -> ActivateCommand.run cachixoptions opts diff --git a/src/Cachix/Client/CNix.hs b/src/Cachix/Client/CNix.hs new file mode 100644 --- /dev/null +++ b/src/Cachix/Client/CNix.hs @@ -0,0 +1,22 @@ +module Cachix.Client.CNix where + +import qualified Data.Text as T +import qualified Language.C.Inline.Cpp.Exceptions as C +import Protolude +import System.Console.Pretty (Color (..), color) + +predicateInvalidPath :: SomeException -> Maybe Text +predicateInvalidPath e + | Just (C.CppStdException msg) <- fromException e = + if "nix::InvalidPath" `T.isSuffixOf` toS msg + then Just $ toS msg + else Nothing +predicateInvalidPath _ = Nothing + +handleInvalidPath :: IO (Maybe a) -> IO (Maybe a) +handleInvalidPath = handleJust predicateInvalidPath handle_ + where + handle_ :: Text -> IO (Maybe a) + handle_ msg = do + hPutStrLn stderr $ color Yellow $ "Warning: " <> msg + return Nothing diff --git a/src/Cachix/Client/Commands.hs b/src/Cachix/Client/Commands.hs --- a/src/Cachix/Client/Commands.hs +++ b/src/Cachix/Client/Commands.hs @@ -15,6 +15,7 @@ import qualified Cachix.API as API import Cachix.API.Error +import Cachix.Client.CNix (handleInvalidPath) import Cachix.Client.Config ( BinaryCacheConfig (BinaryCacheConfig), Config (..), @@ -136,7 +137,7 @@ user <- InstallationMode.getUser nc <- NixConf.read NixConf.Global isTrusted <- InstallationMode.isTrustedUser $ NixConf.readLines (catMaybes [nc]) NixConf.isTrustedUsers - isNixOS <- doesFileExist "/etc/NIXOS" + isNixOS <- doesFileExist "/run/current-system/nixos-version" let nixEnv = InstallationMode.NixEnv { InstallationMode.isRoot = user == "root", @@ -160,15 +161,23 @@ -- This is somewhat like the behavior of `cat` for example. (_, paths) -> return paths pushParams <- getPushParams env opts name - normalized <- liftIO $ for inputStorePaths (followLinksToStorePath (pushParamsStore pushParams) . encodeUtf8) - void $ + normalized <- + liftIO $ + for inputStorePaths $ \path -> + handleInvalidPath $ + Just <$> followLinksToStorePath (pushParamsStore pushParams) (encodeUtf8 path) + pushedPaths <- pushClosure (mapConcurrentlyBounded (numJobs opts)) pushParams - normalized - putText "All done." -push _ _ = do - throwIO $ DeprecatedCommand "DEPRECATED: cachix watch-store has replaced cachix push --watch-store." + (catMaybes normalized) + case (length normalized, length pushedPaths) of + (0, _) -> putText "Nothing to push." + (_, 0) -> putText "Nothing to push - all store paths are already on Cachix." + _ -> putText "All done." +push _ _ = + throwIO $ + DeprecatedCommand "DEPRECATED: cachix watch-store has replaced cachix push --watch-store." watchStore :: Env -> PushOptions -> Text -> IO () watchStore env opts name = do diff --git a/src/Cachix/Client/Env.hs b/src/Cachix/Client/Env.hs --- a/src/Cachix/Client/Env.hs +++ b/src/Cachix/Client/Env.hs @@ -1,7 +1,7 @@ module Cachix.Client.Env ( Env (..), mkEnv, - cachixVersion, + createClientEnv, customManagerSettings, ) where @@ -9,7 +9,7 @@ import Cachix.Client.Config (Config, readConfig) import Cachix.Client.OptionsParser (CachixOptions (..)) import Cachix.Client.URI (getBaseUrl) -import Data.Version (showVersion) +import Cachix.Client.Version (cachixVersion) import Hercules.CNix.Store (Store, openStore) import Network.HTTP.Client ( ManagerSettings, @@ -19,10 +19,9 @@ ) import Network.HTTP.Client.TLS (newTlsManagerWith, tlsManagerSettings) import Network.HTTP.Simple (setRequestHeader) -import Paths_cachix (version) import Protolude hiding (toS) import Protolude.Conv -import Servant.Client (ClientEnv, mkClientEnv) +import Servant.Client.Streaming (ClientEnv, mkClientEnv) import System.Directory (canonicalizePath) data Env = Env @@ -39,8 +38,7 @@ canonicalConfigPath <- canonicalizePath (configPath rawcachixoptions) let cachixOptions = rawcachixoptions {configPath = canonicalConfigPath} cfg <- readConfig $ configPath cachixOptions - manager <- newTlsManagerWith customManagerSettings - let clientEnv = mkClientEnv manager $ getBaseUrl (host cachixOptions) + clientEnv <- createClientEnv cachixOptions return Env { config = cfg, @@ -57,5 +55,7 @@ managerModifyRequest = return . setRequestHeader "User-Agent" [toS cachixVersion] } -cachixVersion :: Text -cachixVersion = "cachix " <> toS (showVersion version) +createClientEnv :: CachixOptions -> IO ClientEnv +createClientEnv cachixOptions = do + manager <- newTlsManagerWith customManagerSettings + return $ mkClientEnv manager $ getBaseUrl (host cachixOptions) diff --git a/src/Cachix/Client/InstallationMode.hs b/src/Cachix/Client/InstallationMode.hs --- a/src/Cachix/Client/InstallationMode.hs +++ b/src/Cachix/Client/InstallationMode.hs @@ -245,5 +245,5 @@ getUser = do maybeUser <- lookupEnv "USER" case maybeUser of - Nothing -> throwIO $ UserEnvNotSet "$USER must be set" + Nothing -> throwIO $ UserEnvNotSet "$USER must be set. If running in a container, try setting USER=root." Just user -> return $ toS user diff --git a/src/Cachix/Client/OptionsParser.hs b/src/Cachix/Client/OptionsParser.hs --- a/src/Cachix/Client/OptionsParser.hs +++ b/src/Cachix/Client/OptionsParser.hs @@ -11,6 +11,7 @@ import qualified Cachix.Client.Config as Config import qualified Cachix.Client.InstallationMode as InstallationMode import Cachix.Client.URI (defaultCachixURI) +import qualified Cachix.Deploy.OptionsParser as DeployOptions import Options.Applicative import Protolude hiding (option, toS) import Protolude.Conv @@ -67,6 +68,7 @@ | WatchStore PushOptions Text | WatchExec PushOptions Text Text [Text] | Use BinaryCacheName InstallationMode.UseOptions + | DeployCommand DeployOptions.DeployCommand | Version deriving (Show) @@ -91,6 +93,7 @@ <> command "watch-exec" (infoH watchExec (progDesc "Run a command while it's running watch /nix/store for newly added store paths and upload them to a binary cache")) <> command "watch-store" (infoH watchStore (progDesc "Indefinitely watch /nix/store for newly added store paths and upload them to a binary cache")) <> command "use" (infoH use (progDesc "Configure a binary cache by writing nix.conf and netrc files")) + <> command "deploy" (infoH (DeployCommand <$> DeployOptions.parser) (progDesc "Cachix Deploy commands")) where nameArg = strArgument (metavar "CACHE-NAME") authtoken = AuthToken <$> (stdinFlag <|> (Just <$> authTokenArg)) diff --git a/src/Cachix/Client/Push.hs b/src/Cachix/Client/Push.hs --- a/src/Cachix/Client/Push.hs +++ b/src/Cachix/Client/Push.hs @@ -188,7 +188,7 @@ narSize <- readIORef narSizeRef narHash <- ("sha256:" <>) . System.Nix.Base32.encode <$> readIORef narHashRef narHashNix <- Store.validPathInfoNarHash32 pathinfo - when (narHash /= toS narHashNix) $ throwM $ NarHashMismatch "Nar hash mismatch between nix-store --dump and nix db. You can repair db metadata by running as root: $ nix-store --verify --repair" + when (narHash /= toS narHashNix) $ throwM $ NarHashMismatch $ toS storePathText <> ": Nar hash mismatch between nix-store --dump and nix db. You can repair db metadata by running as root: $ nix-store --verify --repair --check-contents" fileHash <- readIORef fileHashRef fileSize <- readIORef fileSizeRef deriverPath <- diff --git a/src/Cachix/Client/PushQueue.hs b/src/Cachix/Client/PushQueue.hs --- a/src/Cachix/Client/PushQueue.hs +++ b/src/Cachix/Client/PushQueue.hs @@ -12,6 +12,7 @@ ) where +import Cachix.Client.CNix (handleInvalidPath) import qualified Cachix.Client.Push as Push import Cachix.Client.Retry (retryAll) import Control.Concurrent.Async @@ -22,6 +23,7 @@ import Hercules.CNix.Store (StorePath) import Protolude import qualified System.Posix.Signals as Signals +import qualified System.Systemd.Daemon as Systemd type Queue = TBQueue.TBQueue StorePath @@ -40,8 +42,11 @@ worker pushParams workerState = forever $ do storePath <- atomically $ TBQueue.readTBQueue $ pushQueue workerState bracket_ (inProgresModify (+ 1)) (inProgresModify (\x -> x - 1)) $ - retryAll $ - Push.uploadStorePath pushParams storePath + retryAll $ \retrystatus -> + void $ + handleInvalidPath $ + Just + <$> Push.uploadStorePath pushParams storePath retrystatus where inProgresModify f = atomically $ modifyTVar' (inProgress workerState) f @@ -61,7 +66,9 @@ progress <- newTVarIO 0 let pushWorkerState = PushWorkerState newPushQueue progress pushWorker <- async $ replicateConcurrently_ numWorkers $ worker pushParams pushWorkerState - void $ Signals.installHandler Signals.sigINT (Signals.CatchOnce (exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState)) Nothing + 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] case eitherException of Left exc | fromException exc == Just StopWorker -> return () @@ -95,6 +102,7 @@ exitOnceQueueIsEmpty :: IO () -> Async () -> Async () -> QueryWorkerState -> PushWorkerState -> IO () exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState = do putText "Stopped watching /nix/store and waiting for queue to empty ..." + Systemd.notifyStopping stopProducerCallback go where @@ -112,6 +120,8 @@ cancelWith queryWorker StopWorker cancelWith pushWorker StopWorker else do + -- extend shutdown for another 90s + Systemd.notify False $ "EXTEND_TIMEOUT_USEC=" <> show (90 * 1000 * 1000) putText $ "Waiting to finish: " <> show inprogress <> " pushing, " <> show queueLength <> " in queue" threadDelay (1000 * 1000) go diff --git a/src/Cachix/Client/Retry.hs b/src/Cachix/Client/Retry.hs --- a/src/Cachix/Client/Retry.hs +++ b/src/Cachix/Client/Retry.hs @@ -1,16 +1,32 @@ module Cachix.Client.Retry ( retryAll, + retryAllWithLogging, + endlessRetryPolicy, ) where -import Control.Exception.Safe (MonadMask) -import Control.Retry (RetryPolicy, RetryStatus, exponentialBackoff, limitRetries, recoverAll) +import Control.Exception.Safe (MonadMask, isSyncException) +import Control.Retry (RetryPolicy, RetryPolicyM, RetryStatus, exponentialBackoff, limitRetries, logRetries, recoverAll, recovering, skipAsyncExceptions) import Protolude --- Catches all exceptions except skipAsyncExceptions retryAll :: (MonadIO m, MonadMask m) => (RetryStatus -> m a) -> m a -retryAll = recoverAll defaultRetryPolicy +retryAll = + recoverAll defaultRetryPolicy + +-- Catches all exceptions except async exceptions with logging support +retryAllWithLogging :: (MonadIO m, MonadMask m) => RetryPolicyM m -> (Bool -> SomeException -> RetryStatus -> m ()) -> m a -> m a +retryAllWithLogging retryPolicy logger action = recovering retryPolicy handlers $ const action where - defaultRetryPolicy :: RetryPolicy - defaultRetryPolicy = - exponentialBackoff (1000 * 1000) <> limitRetries 5 + handlers = skipAsyncExceptions ++ [loggingHandler] + + loggingHandler = logRetries exceptionPredicate logger + + exceptionPredicate = return . isSyncException + +defaultRetryPolicy :: RetryPolicy +defaultRetryPolicy = + exponentialBackoff (1000 * 1000) <> limitRetries 5 + +endlessRetryPolicy :: RetryPolicy +endlessRetryPolicy = + exponentialBackoff (1000 * 1000) diff --git a/src/Cachix/Client/Servant.hs b/src/Cachix/Client/Servant.hs --- a/src/Cachix/Client/Servant.hs +++ b/src/Cachix/Client/Servant.hs @@ -8,6 +8,7 @@ module Cachix.Client.Servant ( isErr, cachixClient, + deployClient ) where @@ -16,12 +17,13 @@ import Protolude import Servant.API.Generic import Servant.Auth.Client () +import Cachix.Types.ContentTypes () import qualified Servant.Client import Servant.Client.Generic (AsClientT) import Servant.Client.Streaming +import qualified Cachix.API.Deploy as Cachix.Deploy.API import Servant.Conduit () - isErr :: ClientError -> Status -> Bool isErr (Servant.Client.FailureResponse _ resp) status | Servant.Client.responseStatusCode resp == status = True @@ -29,3 +31,6 @@ cachixClient :: Cachix.API.BinaryCacheAPI (AsClientT ClientM) cachixClient = fromServant $ client Cachix.API.api + +deployClient :: Cachix.Deploy.API.DeployAPI (AsClientT ClientM) +deployClient = fromServant $ client Cachix.Deploy.API.api diff --git a/src/Cachix/Client/Version.hs b/src/Cachix/Client/Version.hs new file mode 100644 --- /dev/null +++ b/src/Cachix/Client/Version.hs @@ -0,0 +1,11 @@ +module Cachix.Client.Version where + +import Data.Version (showVersion) +import Paths_cachix (version) +import Protolude + +cachixVersion :: Text +cachixVersion = "cachix " <> versionNumber + +versionNumber :: Text +versionNumber = toS $ showVersion version diff --git a/src/Cachix/Client/WatchStore.hs b/src/Cachix/Client/WatchStore.hs --- a/src/Cachix/Client/WatchStore.hs +++ b/src/Cachix/Client/WatchStore.hs @@ -11,9 +11,11 @@ import qualified Hercules.CNix.Store as Store import Protolude import System.FSNotify +import qualified System.Systemd.Daemon as Systemd startWorkers :: Store -> Int -> PushParams IO () -> IO () startWorkers store numWorkers pushParams = do + Systemd.notifyReady withManager $ \mgr -> PushQueue.startWorkers numWorkers (producer store mgr) pushParams producer :: Store -> WatchManager -> PushQueue.Queue -> IO (IO ()) diff --git a/src/Cachix/Deploy/Activate.hs b/src/Cachix/Deploy/Activate.hs new file mode 100644 --- /dev/null +++ b/src/Cachix/Deploy/Activate.hs @@ -0,0 +1,190 @@ +{-# LANGUAGE DuplicateRecordFields #-} + +module Cachix.Deploy.Activate where + +import qualified Cachix.API.WebSocketSubprotocol as WSS +import qualified Cachix.Client.InstallationMode as InstallationMode +import qualified Cachix.Client.NetRc as NetRc +import qualified Cachix.Client.OptionsParser as CachixOptions +import Cachix.Client.URI (getBaseUrl) +import qualified Cachix.Deploy.OptionsParser as AgentOptions +import qualified Cachix.Types.BinaryCache as BinaryCache +import Cachix.Types.Permission (Permission (..)) +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 Katip as K +import qualified Network.WebSockets as WS +import Protolude hiding (log, toS) +import Protolude.Conv (toS) +import Servant.Auth.Client (Token (..)) +import qualified Servant.Client as Servant +import System.Directory (doesPathExist) +import System.IO.Temp (withSystemTempDirectory) +import System.Process +import Prelude (String) + +-- TODO: duplicated in Agent.hs +host :: CachixOptions.CachixOptions -> String +host cachixOptions = Servant.baseUrlHost $ getBaseUrl $ CachixOptions.host cachixOptions + +domain :: CachixOptions.CachixOptions -> WSS.Cache -> Text +domain cachixOptions cache = toS (WSS.cacheName cache) <> "." <> toS (host cachixOptions) + +-- TODO: get uri scheme +uri :: CachixOptions.CachixOptions -> WSS.Cache -> Text +uri cachixOptions cache = "https://" <> domain cachixOptions cache + +-- TODO: what if websocket gets closed while deploying? +activate :: + CachixOptions.CachixOptions -> + AgentOptions.AgentOptions -> + WS.Connection -> + Conduit.ConduitT ByteString Void IO () -> + WSS.DeploymentDetails -> + WSS.AgentInformation -> + ByteString -> + K.KatipContextT IO () +activate cachixOptions agentArgs connection sourceStream deploymentDetails agentInfo agentToken = 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 cachixOptions cache)] + sigs = ["--option", "trusted-public-keys", officialCache <> " " <> toS (domain cachixOptions 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 (1 * 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 + -- 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 cachixOptions 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) + + -- TODO: use exceptions to simplify this code + case downloadExitCode of + ExitFailure _ -> deploymentFailed + ExitSuccess -> do + (profile, activationScripts) <- liftIO $ getActivationScript storePath (AgentOptions.profile agentArgs) + -- TODO: document what happens if wrong user is used for the agent + + -- set the new profile + (activateProfileExitCode, _, _) <- liftIO $ shellOut "nix-env" ["-p", toS profile, "--set", toS storePath] + 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 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 (2 * 1000 * 1000) + + K.logLocM K.InfoS $ K.ls $ "Deployment #" <> index <> " finished" + where + -- TODO: prevent service from being restarted while deploying + -- TODO: upgrade agent + + index :: Text + index = show $ WSS.index deploymentDetails + + 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" + + getActivationScript :: Text -> Text -> IO (Text, [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 profile, cmds) + +type Command = (String, [String]) + +-- TODO: home-manager diff --git a/src/Cachix/Deploy/ActivateCommand.hs b/src/Cachix/Deploy/ActivateCommand.hs new file mode 100644 --- /dev/null +++ b/src/Cachix/Deploy/ActivateCommand.hs @@ -0,0 +1,33 @@ +{-# LANGUAGE NamedFieldPuns #-} + +module Cachix.Deploy.ActivateCommand where + +import qualified Cachix.API.Deploy as API +import Cachix.API.Error (escalate) +import qualified Cachix.Client.Env as Env +import qualified Cachix.Client.OptionsParser as CachixOptions +import Cachix.Client.Servant (deployClient) +import qualified Cachix.Deploy.OptionsParser as DeployOptions +import qualified Cachix.Types.DeployResponse as DeployResponse +import qualified Data.Aeson as Aeson +import qualified Data.HashMap.Strict as HM +import Protolude hiding (toS) +import Protolude.Conv +import Servant.Auth.Client (Token (..)) +import Servant.Client.Streaming (runClientM) +import Servant.Conduit () +import System.Environment (getEnv) + +run :: CachixOptions.CachixOptions -> DeployOptions.ActivateOptions -> IO () +run cachixOptions DeployOptions.ActivateOptions {DeployOptions.payloadPath} = do + agentToken <- getEnv "CACHIX_ACTIVATE_TOKEN" + clientEnv <- Env.createClientEnv cachixOptions + payloadEither <- Aeson.eitherDecodeFileStrict' payloadPath + case payloadEither of + Left err -> do + hPutStrLn stderr $ "Error while parsing JSON: " <> err + exitFailure + Right payload -> do + response <- escalate <=< (`runClientM` clientEnv) $ API.activate deployClient (Token $ toS agentToken) payload + for_ (HM.toList $ DeployResponse.agents response) $ + \(_, url) -> putStrLn url diff --git a/src/Cachix/Deploy/Agent.hs b/src/Cachix/Deploy/Agent.hs new file mode 100644 --- /dev/null +++ b/src/Cachix/Deploy/Agent.hs @@ -0,0 +1,119 @@ +{-# LANGUAGE DuplicateRecordFields #-} + +module Cachix.Deploy.Agent where + +import qualified Cachix.API.WebSocketSubprotocol as WSS +import qualified Cachix.Client.OptionsParser as CachixOptions +import Cachix.Client.Retry +import Cachix.Client.URI (getBaseUrl) +import Cachix.Client.Version (versionNumber) +import qualified Cachix.Deploy.Activate as Activate +import qualified Cachix.Deploy.OptionsParser as AgentOptions +import Conduit ((.|)) +import qualified Control.Concurrent.Async as Async +import qualified Control.Concurrent.STM.TQueue as TQueue +import qualified Control.Exception.Safe as Exception +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.IORef +import Data.String (String) +import Data.Time.Clock (getCurrentTime) +import Data.UUID (UUID) +import qualified Data.UUID as UUID +import Katip (KatipContextT) +import qualified Katip as K +import Network.HTTP.Simple (RequestHeaders) +import qualified Network.WebSockets as WS +import Protolude hiding (toS) +import Protolude.Conv +import qualified Servant.Client as Servant +import System.Environment (getEnv) +import qualified Wuss + +type AgentState = IORef (Maybe WSS.AgentInformation) + +run :: CachixOptions.CachixOptions -> AgentOptions.AgentOptions -> IO () +run cachixOptions agentOpts = withKatip (CachixOptions.verbose cachixOptions) $ \logEnv -> do + agentToken <- getEnv "CACHIX_AGENT_TOKEN" + -- TODO: error if token is missing + agentState <- newIORef Nothing + let runKatip = K.runKatipContextT logEnv () "agent" + headers = + [ ("Authorization", "Bearer " <> toS agentToken), + ("name", name), + ("version", toS versionNumber) + ] + host = Servant.baseUrlHost $ getBaseUrl $ CachixOptions.host cachixOptions + path = "/ws" + runKatip $ + retryAllWithLogging endlessRetryPolicy (logger runKatip) $ do + K.logLocM K.InfoS $ K.ls ("Agent " <> agentIdentifier <> " connecting to " <> toS host <> toS path) + liftIO $ + Wuss.runSecureClientWith host 443 path WS.defaultConnectionOptions headers $ \connection -> runKatip $ do + K.logLocM K.InfoS "Connected to Cachix Deploy service" + liftIO $ + WS.withPingThread connection 30 (runKatip $ K.logLocM K.DebugS "WebSocket keep-alive ping") $ do + WSS.recieveDataConcurrently + connection + (\message -> Exception.handle (handler runKatip) $ runKatip (handleMessage runKatip host headers message connection agentState (toS agentToken))) + where + name = toS (AgentOptions.name agentOpts) + agentIdentifier = name <> " " <> toS versionNumber + logger runKatip _ exception _ = runKatip $ K.logLocM K.ErrorS $ K.ls $ "Retrying due to an exception:" <> displayException exception + handleMessage :: (KatipContextT IO () -> IO ()) -> String -> RequestHeaders -> ByteString -> WS.Connection -> AgentState -> ByteString -> KatipContextT IO () + handleMessage runKatip host headers payload connection agentState agentToken = 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) -> + case WSS.command message of + WSS.AgentRegistered agentInformation -> do + K.logLocM K.InfoS $ K.ls $ "Agent " <> agentIdentifier <> " registered." + liftIO $ atomicWriteIORef agentState (Just agentInformation) + WSS.Deployment deploymentDetails -> do + maybeAgentInformation <- liftIO $ readIORef agentState + let index :: Text + index = show $ WSS.index deploymentDetails + deploymentID = WSS.id (deploymentDetails :: WSS.DeploymentDetails) + case maybeAgentInformation of + Nothing -> K.logLocM K.InfoS $ K.ls $ "Ignoring deployment #" <> index <> " as agent isn't registered yet." + Just agentInformation -> do + queue <- liftIO $ atomically TQueue.newTQueue + liftIO $ Async.race_ (runLogStreaming runKatip host headers queue deploymentID) $ runKatip $ Activate.activate cachixOptions agentOpts connection (Conduit.sinkTQueue queue) deploymentDetails agentInformation agentToken + + runLogStreaming :: (KatipContextT IO () -> IO ()) -> String -> RequestHeaders -> Conduit.TQueue ByteString -> UUID -> IO () + runLogStreaming runKatip host headers queue deploymentID = do + -- TODO: debug Conduit.print + let path = "/api/v1/deploy/log/" <> UUID.toText deploymentID + retryAllWithLogging endlessRetryPolicy (logger runKatip) $ do + liftIO $ + Wuss.runSecureClientWith host 443 (toS path) WS.defaultConnectionOptions headers $ \connection -> + bracket_ (return ()) (WS.sendClose connection ("Closing." :: ByteString)) $ + Conduit.runConduit $ + Conduit.sourceTQueue queue + .| Conduit.linesUnboundedAscii + .| websocketSend connection + +handler :: (KatipContextT IO () -> IO ()) -> Exception.SomeException -> IO () +handler runKatip e = do + runKatip $ K.logLocM K.ErrorS $ "Unexpected exception: " <> K.ls (Exception.displayException e) + +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 + +websocketSend :: WS.Connection -> Conduit.ConduitT ByteString Conduit.Void IO () +websocketSend connection = Conduit.mapM_ f + where + f = \bs -> do + now <- getCurrentTime + WS.sendTextData connection $ Aeson.encode $ WSS.Log {line = toS bs, time = now} diff --git a/src/Cachix/Deploy/OptionsParser.hs b/src/Cachix/Deploy/OptionsParser.hs new file mode 100644 --- /dev/null +++ b/src/Cachix/Deploy/OptionsParser.hs @@ -0,0 +1,58 @@ +module Cachix.Deploy.OptionsParser where + +import Options.Applicative +import Protolude + +data DeployCommand + = Activate ActivateOptions + | Agent AgentOptions + deriving (Show) + +parser :: Parser DeployCommand +parser = + subparser $ + mconcat + [ command "activate" $ + info + (helper <*> activate) + (progDesc "Deploy a new configuration to agents using CACHIX_ACTIVATE_TOKEN."), + command "agent" $ + info + (helper <*> agent) + (progDesc "Run an agent in foreground using CACHIX_AGENT_TOKEN.") + ] + where + activate = Activate <$> parserActivateOptions + agent = Agent <$> parserAgentOptions + +data AgentOptions = AgentOptions + { name :: Text, + profile :: Text + } + deriving (Show) + +data ActivateOptions = ActivateOptions + { payloadPath :: FilePath + } + deriving (Show) + +parserAgentOptions :: Parser AgentOptions +parserAgentOptions = + AgentOptions + <$> strArgument + ( 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." + ) + +parserActivateOptions :: Parser ActivateOptions +parserActivateOptions = + ActivateOptions + <$> strArgument + ( metavar "DEPLOY-SPEC.JSON" + <> help "https://docs.cachix.org/deploy/reference.html#deploy-json" + )