module Ganeti.Luxi
( LuxiOp(..)
, LuxiReq(..)
, Client
, JobId
, fromJobId
, makeJobId
, RecvResult(..)
, strOfOp
, getClient
, getServer
, acceptClient
, closeClient
, closeServer
, callMethod
, submitManyJobs
, queryJobsStatus
, buildCall
, buildResponse
, validateCall
, decodeCall
, recvMsg
, recvMsgExt
, sendMsg
, allLuxiCalls
) where
import Control.Exception (catch)
import Data.IORef
import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as BL
import qualified Data.ByteString.UTF8 as UTF8
import qualified Data.ByteString.Lazy.UTF8 as UTF8L
import Data.Word (Word8)
import Control.Monad
import Text.JSON (encodeStrict, decodeStrict)
import qualified Text.JSON as J
import Text.JSON.Pretty (pp_value)
import Text.JSON.Types
import System.Directory (removeFile)
import System.IO (hClose, hFlush, hWaitForInput, Handle, IOMode(..))
import System.IO.Error (isEOFError)
import System.Posix.Files
import System.Timeout
import qualified Network.Socket as S
import Ganeti.BasicTypes
import Ganeti.Constants
import Ganeti.Errors
import Ganeti.JSON
import Ganeti.OpParams (pTagsObject)
import Ganeti.OpCodes
import Ganeti.Runtime
import qualified Ganeti.Query.Language as Qlang
import Ganeti.THH
import Ganeti.Types
import Ganeti.Utils
withTimeout :: Int -> String -> IO a -> IO a
withTimeout secs descr action = do
result <- timeout (secs * 1000000) action
case result of
Nothing -> fail $ "Timeout in " ++ descr
Just v -> return v
data RecvResult = RecvConnClosed
| RecvError String
| RecvOk String
deriving (Show, Eq)
$(genLuxiOp "LuxiOp"
[ (luxiReqQuery,
[ simpleField "what" [t| Qlang.ItemType |]
, simpleField "fields" [t| [String] |]
, simpleField "qfilter" [t| Qlang.Filter Qlang.FilterField |]
])
, (luxiReqQueryFields,
[ simpleField "what" [t| Qlang.ItemType |]
, simpleField "fields" [t| [String] |]
])
, (luxiReqQueryNodes,
[ simpleField "names" [t| [String] |]
, simpleField "fields" [t| [String] |]
, simpleField "lock" [t| Bool |]
])
, (luxiReqQueryGroups,
[ simpleField "names" [t| [String] |]
, simpleField "fields" [t| [String] |]
, simpleField "lock" [t| Bool |]
])
, (luxiReqQueryNetworks,
[ simpleField "names" [t| [String] |]
, simpleField "fields" [t| [String] |]
, simpleField "lock" [t| Bool |]
])
, (luxiReqQueryInstances,
[ simpleField "names" [t| [String] |]
, simpleField "fields" [t| [String] |]
, simpleField "lock" [t| Bool |]
])
, (luxiReqQueryJobs,
[ simpleField "ids" [t| [JobId] |]
, simpleField "fields" [t| [String] |]
])
, (luxiReqQueryExports,
[ simpleField "nodes" [t| [String] |]
, simpleField "lock" [t| Bool |]
])
, (luxiReqQueryConfigValues,
[ simpleField "fields" [t| [String] |] ]
)
, (luxiReqQueryClusterInfo, [])
, (luxiReqQueryTags,
[ pTagsObject ])
, (luxiReqSubmitJob,
[ simpleField "job" [t| [MetaOpCode] |] ]
)
, (luxiReqSubmitManyJobs,
[ simpleField "ops" [t| [[MetaOpCode]] |] ]
)
, (luxiReqWaitForJobChange,
[ simpleField "job" [t| JobId |]
, simpleField "fields" [t| [String]|]
, simpleField "prev_job" [t| JSValue |]
, simpleField "prev_log" [t| JSValue |]
, simpleField "tmout" [t| Int |]
])
, (luxiReqArchiveJob,
[ simpleField "job" [t| JobId |] ]
)
, (luxiReqAutoArchiveJobs,
[ simpleField "age" [t| Int |]
, simpleField "tmout" [t| Int |]
])
, (luxiReqCancelJob,
[ simpleField "job" [t| JobId |] ]
)
, (luxiReqChangeJobPriority,
[ simpleField "job" [t| JobId |]
, simpleField "priority" [t| Int |] ]
)
, (luxiReqSetDrainFlag,
[ simpleField "flag" [t| Bool |] ]
)
, (luxiReqSetWatcherPause,
[ simpleField "duration" [t| Double |] ]
)
])
$(makeJSONInstance ''LuxiReq)
$(genAllConstr (drop 3) ''LuxiReq "allLuxiCalls")
$(genStrOfOp ''LuxiOp "strOfOp")
data LuxiCall = LuxiCall LuxiReq JSValue
eOM :: Word8
eOM = 3
bEOM :: B.ByteString
bEOM = B.singleton eOM
data MsgKeys = Method
| Args
| Success
| Result
$(genStrOfKey ''MsgKeys "strOfKey")
data Client = Client { socket :: Handle
, rbuf :: IORef B.ByteString
}
getClient :: String -> IO Client
getClient path = do
s <- S.socket S.AF_UNIX S.Stream S.defaultProtocol
withTimeout luxiDefCtmo "creating luxi connection" $
S.connect s (S.SockAddrUnix path)
rf <- newIORef B.empty
h <- S.socketToHandle s ReadWriteMode
return Client { socket=h, rbuf=rf }
getServer :: Bool -> FilePath -> IO S.Socket
getServer setOwner path = do
s <- S.socket S.AF_UNIX S.Stream S.defaultProtocol
S.bindSocket s (S.SockAddrUnix path)
when setOwner $ do
setOwnerAndGroupFromNames path GanetiLuxid $ ExtraGroup DaemonsGroup
setFileMode path $ fromIntegral luxiSocketPerms
S.listen s 5
return s
closeServer :: FilePath -> S.Socket -> IO ()
closeServer path sock = do
S.sClose sock
removeFile path
acceptClient :: S.Socket -> IO Client
acceptClient s = do
(client_socket, _) <- S.accept s
new_buffer <- newIORef B.empty
handle <- S.socketToHandle client_socket ReadWriteMode
return Client { socket=handle, rbuf=new_buffer }
closeClient :: Client -> IO ()
closeClient = hClose . socket
sendMsg :: Client -> String -> IO ()
sendMsg s buf = withTimeout luxiDefRwto "sending luxi message" $ do
let encoded = UTF8L.fromString buf
handle = socket s
BL.hPut handle encoded
B.hPut handle bEOM
hFlush handle
recvUpdate :: Handle -> B.ByteString -> IO (B.ByteString, B.ByteString)
recvUpdate handle obuf = do
nbuf <- withTimeout luxiDefRwto "reading luxi response" $ do
_ <- hWaitForInput handle (1)
B.hGetNonBlocking handle 4096
let (msg, remaining) = B.break (eOM ==) nbuf
newbuf = B.append obuf msg
if B.null remaining
then recvUpdate handle newbuf
else return (newbuf, B.tail remaining)
recvMsg :: Client -> IO String
recvMsg s = do
cbuf <- readIORef $ rbuf s
let (imsg, ibuf) = B.break (eOM ==) cbuf
(msg, nbuf) <-
if B.null ibuf
then recvUpdate (socket s) cbuf
else return (imsg, B.tail ibuf)
writeIORef (rbuf s) nbuf
return $ UTF8.toString msg
recvMsgExt :: Client -> IO RecvResult
recvMsgExt s =
Control.Exception.catch (liftM RecvOk (recvMsg s)) $ \e ->
return $ if isEOFError e
then RecvConnClosed
else RecvError (show e)
buildCall :: LuxiOp
-> String
buildCall lo =
let ja = [ (strOfKey Method, J.showJSON $ strOfOp lo)
, (strOfKey Args, opToArgs lo)
]
jo = toJSObject ja
in encodeStrict jo
buildResponse :: Bool
-> JSValue
-> String
buildResponse success args =
let ja = [ (strOfKey Success, JSBool success)
, (strOfKey Result, args)]
jo = toJSObject ja
in encodeStrict jo
validateCall :: String -> Result LuxiCall
validateCall s = do
arr <- fromJResult "parsing top-level luxi message" $
decodeStrict s::Result (JSObject JSValue)
let aobj = fromJSObject arr
call <- fromObj aobj (strOfKey Method)::Result LuxiReq
args <- fromObj aobj (strOfKey Args)
return (LuxiCall call args)
decodeCall :: LuxiCall -> Result LuxiOp
decodeCall (LuxiCall call args) =
case call of
ReqQueryJobs -> do
(jids, jargs) <- fromJVal args
jids' <- case jids of
JSNull -> return []
_ -> fromJVal jids
return $ QueryJobs jids' jargs
ReqQueryInstances -> do
(names, fields, locking) <- fromJVal args
return $ QueryInstances names fields locking
ReqQueryNodes -> do
(names, fields, locking) <- fromJVal args
return $ QueryNodes names fields locking
ReqQueryGroups -> do
(names, fields, locking) <- fromJVal args
return $ QueryGroups names fields locking
ReqQueryClusterInfo ->
return QueryClusterInfo
ReqQueryNetworks -> do
(names, fields, locking) <- fromJVal args
return $ QueryNetworks names fields locking
ReqQuery -> do
(what, fields, qfilter) <- fromJVal args
return $ Query what fields qfilter
ReqQueryFields -> do
(what, fields) <- fromJVal args
fields' <- case fields of
JSNull -> return []
_ -> fromJVal fields
return $ QueryFields what fields'
ReqSubmitJob -> do
[ops1] <- fromJVal args
ops2 <- mapM (fromJResult (luxiReqToRaw call) . J.readJSON) ops1
return $ SubmitJob ops2
ReqSubmitManyJobs -> do
[ops1] <- fromJVal args
ops2 <- mapM (fromJResult (luxiReqToRaw call) . J.readJSON) ops1
return $ SubmitManyJobs ops2
ReqWaitForJobChange -> do
(jid, fields, pinfo, pidx, wtmout) <-
fromJResult "Parsing WaitForJobChange message" $
case args of
JSArray [a, b, c, d, e] ->
(,,,,) `fmap`
J.readJSON a `ap`
J.readJSON b `ap`
J.readJSON c `ap`
J.readJSON d `ap`
J.readJSON e
_ -> J.Error "Not enough values"
return $ WaitForJobChange jid fields pinfo pidx wtmout
ReqArchiveJob -> do
[jid] <- fromJVal args
return $ ArchiveJob jid
ReqAutoArchiveJobs -> do
(age, tmout) <- fromJVal args
return $ AutoArchiveJobs age tmout
ReqQueryExports -> do
(nodes, lock) <- fromJVal args
return $ QueryExports nodes lock
ReqQueryConfigValues -> do
[fields] <- fromJVal args
return $ QueryConfigValues fields
ReqQueryTags -> do
(kind, name) <- fromJVal args
item <- tagObjectFrom kind name
return $ QueryTags item
ReqCancelJob -> do
[jid] <- fromJVal args
return $ CancelJob jid
ReqChangeJobPriority -> do
(jid, priority) <- fromJVal args
return $ ChangeJobPriority jid priority
ReqSetDrainFlag -> do
[flag] <- fromJVal args
return $ SetDrainFlag flag
ReqSetWatcherPause -> do
[duration] <- fromJVal args
return $ SetWatcherPause duration
validateResult :: String -> ErrorResult JSValue
validateResult s = do
when (UTF8.replacement_char `elem` s) $
fail "Failed to decode UTF-8, detected replacement char after decoding"
oarr <- fromJResult "Parsing LUXI response" (decodeStrict s)
let arr = J.fromJSObject oarr
status <- fromObj arr (strOfKey Success)
result <- fromObj arr (strOfKey Result)
if status
then return result
else decodeError result
decodeError :: JSValue -> ErrorResult JSValue
decodeError val =
case fromJVal val of
Ok e -> Bad e
Bad msg -> Bad $ GenericError msg
callMethod :: LuxiOp -> Client -> IO (ErrorResult JSValue)
callMethod method s = do
sendMsg s $ buildCall method
result <- recvMsg s
let rval = validateResult result
return rval
parseSubmitJobResult :: JSValue -> ErrorResult JobId
parseSubmitJobResult (JSArray [JSBool True, v]) =
case J.readJSON v of
J.Error msg -> Bad $ LuxiError msg
J.Ok v' -> Ok v'
parseSubmitJobResult (JSArray [JSBool False, JSString x]) =
Bad . LuxiError $ fromJSString x
parseSubmitJobResult v =
Bad . LuxiError $ "Unknown result from the master daemon: " ++
show (pp_value v)
submitManyJobs :: Client -> [[MetaOpCode]] -> IO (ErrorResult [JobId])
submitManyJobs s jobs = do
rval <- callMethod (SubmitManyJobs jobs) s
return $ case rval of
Bad x -> Bad x
Ok (JSArray r) -> mapM parseSubmitJobResult r
x -> Bad . LuxiError $
"Cannot parse response from Ganeti: " ++ show x
queryJobsStatus :: Client -> [JobId] -> IO (ErrorResult [JobStatus])
queryJobsStatus s jids = do
rval <- callMethod (QueryJobs jids ["status"]) s
return $ case rval of
Bad x -> Bad x
Ok y -> case J.readJSON y::(J.Result [[JobStatus]]) of
J.Ok vals -> if any null vals
then Bad $
LuxiError "Missing job status field"
else Ok (map head vals)
J.Error x -> Bad $ LuxiError x