ecad5985169cefa5771d700a923ff69a27fad0aa
[ganeti-github.git] / src / Ganeti / Luxi.hs
1 {-# LANGUAGE TemplateHaskell #-}
2
3 {-| Implementation of the Ganeti LUXI interface.
4
5 -}
6
7 {-
8
9 Copyright (C) 2009, 2010, 2011, 2012 Google Inc.
10
11 This program is free software; you can redistribute it and/or modify
12 it under the terms of the GNU General Public License as published by
13 the Free Software Foundation; either version 2 of the License, or
14 (at your option) any later version.
15
16 This program is distributed in the hope that it will be useful, but
17 WITHOUT ANY WARRANTY; without even the implied warranty of
18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
19 General Public License for more details.
20
21 You should have received a copy of the GNU General Public License
22 along with this program; if not, write to the Free Software
23 Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
24 02110-1301, USA.
25
26 -}
27
28 module Ganeti.Luxi
29 ( LuxiOp(..)
30 , LuxiReq(..)
31 , Client
32 , JobId
33 , fromJobId
34 , makeJobId
35 , RecvResult(..)
36 , strOfOp
37 , getClient
38 , getServer
39 , acceptClient
40 , closeClient
41 , closeServer
42 , callMethod
43 , submitManyJobs
44 , queryJobsStatus
45 , buildCall
46 , buildResponse
47 , validateCall
48 , decodeCall
49 , recvMsg
50 , recvMsgExt
51 , sendMsg
52 , allLuxiCalls
53 ) where
54
55 import Control.Exception (catch)
56 import Data.IORef
57 import qualified Data.ByteString as B
58 import qualified Data.ByteString.Lazy as BL
59 import qualified Data.ByteString.UTF8 as UTF8
60 import qualified Data.ByteString.Lazy.UTF8 as UTF8L
61 import Data.Word (Word8)
62 import Control.Monad
63 import Text.JSON (encodeStrict, decodeStrict)
64 import qualified Text.JSON as J
65 import Text.JSON.Pretty (pp_value)
66 import Text.JSON.Types
67 import System.Directory (removeFile)
68 import System.IO (hClose, hFlush, hWaitForInput, Handle, IOMode(..))
69 import System.IO.Error (isEOFError)
70 import System.Timeout
71 import qualified Network.Socket as S
72
73 import Ganeti.BasicTypes
74 import Ganeti.Constants
75 import Ganeti.Errors
76 import Ganeti.JSON
77 import Ganeti.OpParams (pTagsObject)
78 import Ganeti.OpCodes
79 import Ganeti.Runtime
80 import qualified Ganeti.Query.Language as Qlang
81 import Ganeti.THH
82 import Ganeti.Types
83 import Ganeti.Utils
84
85 -- * Utility functions
86
87 -- | Wrapper over System.Timeout.timeout that fails in the IO monad.
88 withTimeout :: Int -> String -> IO a -> IO a
89 withTimeout secs descr action = do
90 result <- timeout (secs * 1000000) action
91 case result of
92 Nothing -> fail $ "Timeout in " ++ descr
93 Just v -> return v
94
95 -- * Generic protocol functionality
96
97 -- | Result of receiving a message from the socket.
98 data RecvResult = RecvConnClosed -- ^ Connection closed
99 | RecvError String -- ^ Any other error
100 | RecvOk String -- ^ Successfull receive
101 deriving (Show, Eq)
102
103 -- | Currently supported Luxi operations and JSON serialization.
104 $(genLuxiOp "LuxiOp"
105 [ (luxiReqQuery,
106 [ simpleField "what" [t| Qlang.ItemType |]
107 , simpleField "fields" [t| [String] |]
108 , simpleField "qfilter" [t| Qlang.Filter Qlang.FilterField |]
109 ])
110 , (luxiReqQueryFields,
111 [ simpleField "what" [t| Qlang.ItemType |]
112 , simpleField "fields" [t| [String] |]
113 ])
114 , (luxiReqQueryNodes,
115 [ simpleField "names" [t| [String] |]
116 , simpleField "fields" [t| [String] |]
117 , simpleField "lock" [t| Bool |]
118 ])
119 , (luxiReqQueryGroups,
120 [ simpleField "names" [t| [String] |]
121 , simpleField "fields" [t| [String] |]
122 , simpleField "lock" [t| Bool |]
123 ])
124 , (luxiReqQueryNetworks,
125 [ simpleField "names" [t| [String] |]
126 , simpleField "fields" [t| [String] |]
127 , simpleField "lock" [t| Bool |]
128 ])
129 , (luxiReqQueryInstances,
130 [ simpleField "names" [t| [String] |]
131 , simpleField "fields" [t| [String] |]
132 , simpleField "lock" [t| Bool |]
133 ])
134 , (luxiReqQueryJobs,
135 [ simpleField "ids" [t| [JobId] |]
136 , simpleField "fields" [t| [String] |]
137 ])
138 , (luxiReqQueryExports,
139 [ simpleField "nodes" [t| [String] |]
140 , simpleField "lock" [t| Bool |]
141 ])
142 , (luxiReqQueryConfigValues,
143 [ simpleField "fields" [t| [String] |] ]
144 )
145 , (luxiReqQueryClusterInfo, [])
146 , (luxiReqQueryTags,
147 [ pTagsObject ])
148 , (luxiReqSubmitJob,
149 [ simpleField "job" [t| [MetaOpCode] |] ]
150 )
151 , (luxiReqSubmitManyJobs,
152 [ simpleField "ops" [t| [[MetaOpCode]] |] ]
153 )
154 , (luxiReqWaitForJobChange,
155 [ simpleField "job" [t| JobId |]
156 , simpleField "fields" [t| [String]|]
157 , simpleField "prev_job" [t| JSValue |]
158 , simpleField "prev_log" [t| JSValue |]
159 , simpleField "tmout" [t| Int |]
160 ])
161 , (luxiReqArchiveJob,
162 [ simpleField "job" [t| JobId |] ]
163 )
164 , (luxiReqAutoArchiveJobs,
165 [ simpleField "age" [t| Int |]
166 , simpleField "tmout" [t| Int |]
167 ])
168 , (luxiReqCancelJob,
169 [ simpleField "job" [t| JobId |] ]
170 )
171 , (luxiReqChangeJobPriority,
172 [ simpleField "job" [t| JobId |]
173 , simpleField "priority" [t| Int |] ]
174 )
175 , (luxiReqSetDrainFlag,
176 [ simpleField "flag" [t| Bool |] ]
177 )
178 , (luxiReqSetWatcherPause,
179 [ simpleField "duration" [t| Double |] ]
180 )
181 ])
182
183 $(makeJSONInstance ''LuxiReq)
184
185 -- | List of all defined Luxi calls.
186 $(genAllConstr (drop 3) ''LuxiReq "allLuxiCalls")
187
188 -- | The serialisation of LuxiOps into strings in messages.
189 $(genStrOfOp ''LuxiOp "strOfOp")
190
191 -- | Type holding the initial (unparsed) Luxi call.
192 data LuxiCall = LuxiCall LuxiReq JSValue
193
194 -- | The end-of-message separator.
195 eOM :: Word8
196 eOM = 3
197
198 -- | The end-of-message encoded as a ByteString.
199 bEOM :: B.ByteString
200 bEOM = B.singleton eOM
201
202 -- | Valid keys in the requests and responses.
203 data MsgKeys = Method
204 | Args
205 | Success
206 | Result
207
208 -- | The serialisation of MsgKeys into strings in messages.
209 $(genStrOfKey ''MsgKeys "strOfKey")
210
211 -- | Luxi client encapsulation.
212 data Client = Client { socket :: Handle -- ^ The socket of the client
213 , rbuf :: IORef B.ByteString -- ^ Already received buffer
214 }
215
216 -- | Connects to the master daemon and returns a luxi Client.
217 getClient :: String -> IO Client
218 getClient path = do
219 s <- S.socket S.AF_UNIX S.Stream S.defaultProtocol
220 withTimeout luxiDefCtmo "creating luxi connection" $
221 S.connect s (S.SockAddrUnix path)
222 rf <- newIORef B.empty
223 h <- S.socketToHandle s ReadWriteMode
224 return Client { socket=h, rbuf=rf }
225
226 -- | Creates and returns a server endpoint.
227 getServer :: Bool -> FilePath -> IO S.Socket
228 getServer setOwner path = do
229 s <- S.socket S.AF_UNIX S.Stream S.defaultProtocol
230 S.bindSocket s (S.SockAddrUnix path)
231 when setOwner . setOwnerAndGroupFromNames path GanetiLuxid $
232 ExtraGroup DaemonsGroup
233 S.listen s 5 -- 5 is the max backlog
234 return s
235
236 -- | Closes a server endpoint.
237 -- FIXME: this should be encapsulated into a nicer type.
238 closeServer :: FilePath -> S.Socket -> IO ()
239 closeServer path sock = do
240 S.sClose sock
241 removeFile path
242
243 -- | Accepts a client
244 acceptClient :: S.Socket -> IO Client
245 acceptClient s = do
246 -- second return is the address of the client, which we ignore here
247 (client_socket, _) <- S.accept s
248 new_buffer <- newIORef B.empty
249 handle <- S.socketToHandle client_socket ReadWriteMode
250 return Client { socket=handle, rbuf=new_buffer }
251
252 -- | Closes the client socket.
253 closeClient :: Client -> IO ()
254 closeClient = hClose . socket
255
256 -- | Sends a message over a luxi transport.
257 sendMsg :: Client -> String -> IO ()
258 sendMsg s buf = withTimeout luxiDefRwto "sending luxi message" $ do
259 let encoded = UTF8L.fromString buf
260 handle = socket s
261 BL.hPut handle encoded
262 B.hPut handle bEOM
263 hFlush handle
264
265 -- | Given a current buffer and the handle, it will read from the
266 -- network until we get a full message, and it will return that
267 -- message and the leftover buffer contents.
268 recvUpdate :: Handle -> B.ByteString -> IO (B.ByteString, B.ByteString)
269 recvUpdate handle obuf = do
270 nbuf <- withTimeout luxiDefRwto "reading luxi response" $ do
271 _ <- hWaitForInput handle (-1)
272 B.hGetNonBlocking handle 4096
273 let (msg, remaining) = B.break (eOM ==) nbuf
274 newbuf = B.append obuf msg
275 if B.null remaining
276 then recvUpdate handle newbuf
277 else return (newbuf, B.tail remaining)
278
279 -- | Waits for a message over a luxi transport.
280 recvMsg :: Client -> IO String
281 recvMsg s = do
282 cbuf <- readIORef $ rbuf s
283 let (imsg, ibuf) = B.break (eOM ==) cbuf
284 (msg, nbuf) <-
285 if B.null ibuf -- if old buffer didn't contain a full message
286 then recvUpdate (socket s) cbuf -- then we read from network
287 else return (imsg, B.tail ibuf) -- else we return data from our buffer
288 writeIORef (rbuf s) nbuf
289 return $ UTF8.toString msg
290
291 -- | Extended wrapper over recvMsg.
292 recvMsgExt :: Client -> IO RecvResult
293 recvMsgExt s =
294 Control.Exception.catch (liftM RecvOk (recvMsg s)) $ \e ->
295 return $ if isEOFError e
296 then RecvConnClosed
297 else RecvError (show e)
298
299 -- | Serialize a request to String.
300 buildCall :: LuxiOp -- ^ The method
301 -> String -- ^ The serialized form
302 buildCall lo =
303 let ja = [ (strOfKey Method, J.showJSON $ strOfOp lo)
304 , (strOfKey Args, opToArgs lo)
305 ]
306 jo = toJSObject ja
307 in encodeStrict jo
308
309 -- | Serialize the response to String.
310 buildResponse :: Bool -- ^ Success
311 -> JSValue -- ^ The arguments
312 -> String -- ^ The serialized form
313 buildResponse success args =
314 let ja = [ (strOfKey Success, JSBool success)
315 , (strOfKey Result, args)]
316 jo = toJSObject ja
317 in encodeStrict jo
318
319 -- | Check that luxi request contains the required keys and parse it.
320 validateCall :: String -> Result LuxiCall
321 validateCall s = do
322 arr <- fromJResult "parsing top-level luxi message" $
323 decodeStrict s::Result (JSObject JSValue)
324 let aobj = fromJSObject arr
325 call <- fromObj aobj (strOfKey Method)::Result LuxiReq
326 args <- fromObj aobj (strOfKey Args)
327 return (LuxiCall call args)
328
329 -- | Converts Luxi call arguments into a 'LuxiOp' data structure.
330 --
331 -- This is currently hand-coded until we make it more uniform so that
332 -- it can be generated using TH.
333 decodeCall :: LuxiCall -> Result LuxiOp
334 decodeCall (LuxiCall call args) =
335 case call of
336 ReqQueryJobs -> do
337 (jids, jargs) <- fromJVal args
338 jids' <- case jids of
339 JSNull -> return []
340 _ -> fromJVal jids
341 return $ QueryJobs jids' jargs
342 ReqQueryInstances -> do
343 (names, fields, locking) <- fromJVal args
344 return $ QueryInstances names fields locking
345 ReqQueryNodes -> do
346 (names, fields, locking) <- fromJVal args
347 return $ QueryNodes names fields locking
348 ReqQueryGroups -> do
349 (names, fields, locking) <- fromJVal args
350 return $ QueryGroups names fields locking
351 ReqQueryClusterInfo ->
352 return QueryClusterInfo
353 ReqQueryNetworks -> do
354 (names, fields, locking) <- fromJVal args
355 return $ QueryNetworks names fields locking
356 ReqQuery -> do
357 (what, fields, qfilter) <- fromJVal args
358 return $ Query what fields qfilter
359 ReqQueryFields -> do
360 (what, fields) <- fromJVal args
361 fields' <- case fields of
362 JSNull -> return []
363 _ -> fromJVal fields
364 return $ QueryFields what fields'
365 ReqSubmitJob -> do
366 [ops1] <- fromJVal args
367 ops2 <- mapM (fromJResult (luxiReqToRaw call) . J.readJSON) ops1
368 return $ SubmitJob ops2
369 ReqSubmitManyJobs -> do
370 [ops1] <- fromJVal args
371 ops2 <- mapM (fromJResult (luxiReqToRaw call) . J.readJSON) ops1
372 return $ SubmitManyJobs ops2
373 ReqWaitForJobChange -> do
374 (jid, fields, pinfo, pidx, wtmout) <-
375 -- No instance for 5-tuple, code copied from the
376 -- json sources and adapted
377 fromJResult "Parsing WaitForJobChange message" $
378 case args of
379 JSArray [a, b, c, d, e] ->
380 (,,,,) `fmap`
381 J.readJSON a `ap`
382 J.readJSON b `ap`
383 J.readJSON c `ap`
384 J.readJSON d `ap`
385 J.readJSON e
386 _ -> J.Error "Not enough values"
387 return $ WaitForJobChange jid fields pinfo pidx wtmout
388 ReqArchiveJob -> do
389 [jid] <- fromJVal args
390 return $ ArchiveJob jid
391 ReqAutoArchiveJobs -> do
392 (age, tmout) <- fromJVal args
393 return $ AutoArchiveJobs age tmout
394 ReqQueryExports -> do
395 (nodes, lock) <- fromJVal args
396 return $ QueryExports nodes lock
397 ReqQueryConfigValues -> do
398 [fields] <- fromJVal args
399 return $ QueryConfigValues fields
400 ReqQueryTags -> do
401 (kind, name) <- fromJVal args
402 item <- tagObjectFrom kind name
403 return $ QueryTags item
404 ReqCancelJob -> do
405 [jid] <- fromJVal args
406 return $ CancelJob jid
407 ReqChangeJobPriority -> do
408 (jid, priority) <- fromJVal args
409 return $ ChangeJobPriority jid priority
410 ReqSetDrainFlag -> do
411 [flag] <- fromJVal args
412 return $ SetDrainFlag flag
413 ReqSetWatcherPause -> do
414 [duration] <- fromJVal args
415 return $ SetWatcherPause duration
416
417 -- | Check that luxi responses contain the required keys and that the
418 -- call was successful.
419 validateResult :: String -> ErrorResult JSValue
420 validateResult s = do
421 when (UTF8.replacement_char `elem` s) $
422 fail "Failed to decode UTF-8, detected replacement char after decoding"
423 oarr <- fromJResult "Parsing LUXI response" (decodeStrict s)
424 let arr = J.fromJSObject oarr
425 status <- fromObj arr (strOfKey Success)
426 result <- fromObj arr (strOfKey Result)
427 if status
428 then return result
429 else decodeError result
430
431 -- | Try to decode an error from the server response. This function
432 -- will always fail, since it's called only on the error path (when
433 -- status is False).
434 decodeError :: JSValue -> ErrorResult JSValue
435 decodeError val =
436 case fromJVal val of
437 Ok e -> Bad e
438 Bad msg -> Bad $ GenericError msg
439
440 -- | Generic luxi method call.
441 callMethod :: LuxiOp -> Client -> IO (ErrorResult JSValue)
442 callMethod method s = do
443 sendMsg s $ buildCall method
444 result <- recvMsg s
445 let rval = validateResult result
446 return rval
447
448 -- | Parse job submission result.
449 parseSubmitJobResult :: JSValue -> ErrorResult JobId
450 parseSubmitJobResult (JSArray [JSBool True, v]) =
451 case J.readJSON v of
452 J.Error msg -> Bad $ LuxiError msg
453 J.Ok v' -> Ok v'
454 parseSubmitJobResult (JSArray [JSBool False, JSString x]) =
455 Bad . LuxiError $ fromJSString x
456 parseSubmitJobResult v =
457 Bad . LuxiError $ "Unknown result from the master daemon: " ++
458 show (pp_value v)
459
460 -- | Specialized submitManyJobs call.
461 submitManyJobs :: Client -> [[MetaOpCode]] -> IO (ErrorResult [JobId])
462 submitManyJobs s jobs = do
463 rval <- callMethod (SubmitManyJobs jobs) s
464 -- map each result (status, payload) pair into a nice Result ADT
465 return $ case rval of
466 Bad x -> Bad x
467 Ok (JSArray r) -> mapM parseSubmitJobResult r
468 x -> Bad . LuxiError $
469 "Cannot parse response from Ganeti: " ++ show x
470
471 -- | Custom queryJobs call.
472 queryJobsStatus :: Client -> [JobId] -> IO (ErrorResult [JobStatus])
473 queryJobsStatus s jids = do
474 rval <- callMethod (QueryJobs jids ["status"]) s
475 return $ case rval of
476 Bad x -> Bad x
477 Ok y -> case J.readJSON y::(J.Result [[JobStatus]]) of
478 J.Ok vals -> if any null vals
479 then Bad $
480 LuxiError "Missing job status field"
481 else Ok (map head vals)
482 J.Error x -> Bad $ LuxiError x