Statistics
| Branch: | Tag: | Revision:

root / src / Ganeti / Query / Server.hs @ 857cbfb1

History | View | Annotate | Download (17.9 kB)

1
{-| Implementation of the Ganeti Query2 server.
2

    
3
-}
4

    
5
{-
6

    
7
Copyright (C) 2012, 2013 Google Inc.
8

    
9
This program is free software; you can redistribute it and/or modify
10
it under the terms of the GNU General Public License as published by
11
the Free Software Foundation; either version 2 of the License, or
12
(at your option) any later version.
13

    
14
This program is distributed in the hope that it will be useful, but
15
WITHOUT ANY WARRANTY; without even the implied warranty of
16
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
17
General Public License for more details.
18

    
19
You should have received a copy of the GNU General Public License
20
along with this program; if not, write to the Free Software
21
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22
02110-1301, USA.
23

    
24
-}
25

    
26
module Ganeti.Query.Server
27
  ( main
28
  , checkMain
29
  , prepMain
30
  ) where
31

    
32
import Control.Applicative
33
import Control.Concurrent
34
import Control.Exception
35
import Control.Monad (forever, when, zipWithM, liftM)
36
import Data.Bits (bitSize)
37
import qualified Data.Set as Set (toList)
38
import Data.IORef
39
import Data.Maybe (fromMaybe)
40
import qualified Text.JSON as J
41
import Text.JSON (encode, showJSON, JSValue(..))
42
import System.Info (arch)
43
import System.Directory
44

    
45
import qualified Ganeti.Constants as C
46
import qualified Ganeti.ConstantUtils as ConstantUtils (unFrozenSet)
47
import Ganeti.Errors
48
import qualified Ganeti.Path as Path
49
import Ganeti.Daemon
50
import Ganeti.Objects
51
import qualified Ganeti.Config as Config
52
import Ganeti.ConfigReader
53
import Ganeti.BasicTypes
54
import Ganeti.JQueue
55
import Ganeti.JQScheduler
56
import Ganeti.Logging
57
import Ganeti.Luxi
58
import qualified Ganeti.Query.Language as Qlang
59
import qualified Ganeti.Query.Cluster as QCluster
60
import Ganeti.Path (queueDir, jobQueueLockFile, jobQueueDrainFile)
61
import Ganeti.Rpc
62
import Ganeti.Query.Query
63
import Ganeti.Query.Filter (makeSimpleFilter)
64
import Ganeti.Types
65
import qualified Ganeti.UDSServer as U (Handler(..), listener)
66
import Ganeti.Utils (lockFile, exitIfBad, watchFile, safeRenameFile)
67
import qualified Ganeti.Version as Version
68

    
69
-- | Helper for classic queries.
70
handleClassicQuery :: ConfigData      -- ^ Cluster config
71
                   -> Qlang.ItemType  -- ^ Query type
72
                   -> [Either String Integer] -- ^ Requested names
73
                                              -- (empty means all)
74
                   -> [String]        -- ^ Requested fields
75
                   -> Bool            -- ^ Whether to do sync queries or not
76
                   -> IO (GenericResult GanetiException JSValue)
77
handleClassicQuery _ _ _ _ True =
78
  return . Bad $ OpPrereqError "Sync queries are not allowed" ECodeInval
79
handleClassicQuery cfg qkind names fields _ = do
80
  let flt = makeSimpleFilter (nameField qkind) names
81
  qr <- query cfg True (Qlang.Query qkind fields flt)
82
  return $ showJSON <$> (qr >>= queryCompat)
83

    
84
-- | Minimal wrapper to handle the missing config case.
85
handleCallWrapper :: MVar () -> JQStatus ->  Result ConfigData
86
                     -> LuxiOp -> IO (ErrorResult JSValue)
87
handleCallWrapper _ _ (Bad msg) _ =
88
  return . Bad . ConfigurationError $
89
           "I do not have access to a valid configuration, cannot\
90
           \ process queries: " ++ msg
91
handleCallWrapper qlock qstat (Ok config) op = handleCall qlock qstat config op
92

    
93
-- | Actual luxi operation handler.
94
handleCall :: MVar () -> JQStatus
95
              -> ConfigData -> LuxiOp -> IO (ErrorResult JSValue)
96
handleCall _ _ cdata QueryClusterInfo =
97
  let cluster = configCluster cdata
98
      master = QCluster.clusterMasterNodeName cdata
99
      hypervisors = clusterEnabledHypervisors cluster
100
      diskTemplates = clusterEnabledDiskTemplates cluster
101
      def_hv = case hypervisors of
102
                 x:_ -> showJSON x
103
                 [] -> JSNull
104
      bits = show (bitSize (0::Int)) ++ "bits"
105
      arch_tuple = [bits, arch]
106
      obj = [ ("software_version", showJSON C.releaseVersion)
107
            , ("protocol_version", showJSON C.protocolVersion)
108
            , ("config_version", showJSON C.configVersion)
109
            , ("os_api_version", showJSON . maximum .
110
                                 Set.toList . ConstantUtils.unFrozenSet $
111
                                 C.osApiVersions)
112
            , ("export_version", showJSON C.exportVersion)
113
            , ("vcs_version", showJSON Version.version)
114
            , ("architecture", showJSON arch_tuple)
115
            , ("name", showJSON $ clusterClusterName cluster)
116
            , ("master", showJSON (case master of
117
                                     Ok name -> name
118
                                     _ -> undefined))
119
            , ("default_hypervisor", def_hv)
120
            , ("enabled_hypervisors", showJSON hypervisors)
121
            , ("hvparams", showJSON $ clusterHvparams cluster)
122
            , ("os_hvp", showJSON $ clusterOsHvp cluster)
123
            , ("beparams", showJSON $ clusterBeparams cluster)
124
            , ("osparams", showJSON $ clusterOsparams cluster)
125
            , ("ipolicy", showJSON $ clusterIpolicy cluster)
126
            , ("nicparams", showJSON $ clusterNicparams cluster)
127
            , ("ndparams", showJSON $ clusterNdparams cluster)
128
            , ("diskparams", showJSON $ clusterDiskparams cluster)
129
            , ("candidate_pool_size",
130
               showJSON $ clusterCandidatePoolSize cluster)
131
            , ("max_running_jobs",
132
               showJSON $ clusterMaxRunningJobs cluster)
133
            , ("master_netdev",  showJSON $ clusterMasterNetdev cluster)
134
            , ("master_netmask", showJSON $ clusterMasterNetmask cluster)
135
            , ("use_external_mip_script",
136
               showJSON $ clusterUseExternalMipScript cluster)
137
            , ("volume_group_name",
138
               maybe JSNull showJSON (clusterVolumeGroupName cluster))
139
            , ("drbd_usermode_helper",
140
               maybe JSNull showJSON (clusterDrbdUsermodeHelper cluster))
141
            , ("file_storage_dir", showJSON $ clusterFileStorageDir cluster)
142
            , ("shared_file_storage_dir",
143
               showJSON $ clusterSharedFileStorageDir cluster)
144
            , ("gluster_storage_dir",
145
               showJSON $ clusterGlusterStorageDir cluster)
146
            , ("maintain_node_health",
147
               showJSON $ clusterMaintainNodeHealth cluster)
148
            , ("ctime", showJSON $ clusterCtime cluster)
149
            , ("mtime", showJSON $ clusterMtime cluster)
150
            , ("uuid", showJSON $ clusterUuid cluster)
151
            , ("tags", showJSON $ clusterTags cluster)
152
            , ("uid_pool", showJSON $ clusterUidPool cluster)
153
            , ("default_iallocator",
154
               showJSON $ clusterDefaultIallocator cluster)
155
            , ("default_iallocator_params",
156
              showJSON $ clusterDefaultIallocatorParams cluster)
157
            , ("reserved_lvs", showJSON $ clusterReservedLvs cluster)
158
            , ("primary_ip_version",
159
               showJSON . ipFamilyToVersion $ clusterPrimaryIpFamily cluster)
160
            , ("prealloc_wipe_disks",
161
               showJSON $ clusterPreallocWipeDisks cluster)
162
            , ("hidden_os", showJSON $ clusterHiddenOs cluster)
163
            , ("blacklisted_os", showJSON $ clusterBlacklistedOs cluster)
164
            , ("enabled_disk_templates", showJSON diskTemplates)
165
            ]
166

    
167
  in case master of
168
    Ok _ -> return . Ok . J.makeObj $ obj
169
    Bad ex -> return $ Bad ex
170

    
171
handleCall _ _ cfg (QueryTags kind name) = do
172
  let tags = case kind of
173
               TagKindCluster  -> Ok . clusterTags $ configCluster cfg
174
               TagKindGroup    -> groupTags <$> Config.getGroup    cfg name
175
               TagKindNode     -> nodeTags  <$> Config.getNode     cfg name
176
               TagKindInstance -> instTags  <$> Config.getInstance cfg name
177
               TagKindNetwork  -> Bad $ OpPrereqError
178
                                        "Network tag is not allowed"
179
                                        ECodeInval
180
  return (J.showJSON <$> tags)
181

    
182
handleCall _ _ cfg (Query qkind qfields qfilter) = do
183
  result <- query cfg True (Qlang.Query qkind qfields qfilter)
184
  return $ J.showJSON <$> result
185

    
186
handleCall _ _ _ (QueryFields qkind qfields) = do
187
  let result = queryFields (Qlang.QueryFields qkind qfields)
188
  return $ J.showJSON <$> result
189

    
190
handleCall _ _ cfg (QueryNodes names fields lock) =
191
  handleClassicQuery cfg (Qlang.ItemTypeOpCode Qlang.QRNode)
192
    (map Left names) fields lock
193

    
194
handleCall _ _ cfg (QueryInstances names fields lock) =
195
  handleClassicQuery cfg (Qlang.ItemTypeOpCode Qlang.QRInstance)
196
    (map Left names) fields lock
197

    
198
handleCall _ _ cfg (QueryGroups names fields lock) =
199
  handleClassicQuery cfg (Qlang.ItemTypeOpCode Qlang.QRGroup)
200
    (map Left names) fields lock
201

    
202
handleCall _ _ cfg (QueryJobs names fields) =
203
  handleClassicQuery cfg (Qlang.ItemTypeLuxi Qlang.QRJob)
204
    (map (Right . fromIntegral . fromJobId) names)  fields False
205

    
206
handleCall _ _ cfg (QueryNetworks names fields lock) =
207
  handleClassicQuery cfg (Qlang.ItemTypeOpCode Qlang.QRNetwork)
208
    (map Left names) fields lock
209

    
210
handleCall _ _ cfg (QueryConfigValues fields) = do
211
  let params = [ ("cluster_name", return . showJSON . clusterClusterName
212
                                    . configCluster $ cfg)
213
               , ("watcher_pause", liftM (maybe JSNull showJSON)
214
                                     QCluster.isWatcherPaused)
215
               , ("master_node", return . genericResult (const JSNull) showJSON
216
                                   $ QCluster.clusterMasterNodeName cfg)
217
               , ("drain_flag", liftM (showJSON . not) isQueueOpen)
218
               ] :: [(String, IO JSValue)]
219
  let answer = map (fromMaybe (return JSNull) . flip lookup params) fields
220
  answerEval <- sequence answer
221
  return . Ok . showJSON $ answerEval
222

    
223
handleCall qlock qstat cfg (SubmitJobToDrainedQueue ops) =
224
  do
225
    let mcs = Config.getMasterCandidates cfg
226
    jobid <- allocateJobId mcs qlock
227
    case jobid of
228
      Bad s -> return . Bad . GenericError $ s
229
      Ok jid -> do
230
        ts <- currentTimestamp
231
        job <- liftM (setReceivedTimestamp ts)
232
                 $ queuedJobFromOpCodes jid ops
233
        qDir <- queueDir
234
        write_result <- writeJobToDisk qDir job
235
        case write_result of
236
          Bad s -> return . Bad . GenericError $ s
237
          Ok () -> do
238
            _ <- replicateManyJobs qDir mcs [job]
239
            _ <- forkIO $ enqueueNewJobs qstat [job]
240
            return . Ok . showJSON . fromJobId $ jid
241

    
242
handleCall qlock qstat cfg (SubmitJob ops) =
243
  do
244
    open <- isQueueOpen
245
    if not open
246
       then return . Bad . GenericError $ "Queue drained"
247
       else handleCall qlock qstat cfg (SubmitJobToDrainedQueue ops)
248

    
249
handleCall qlock qstat cfg (SubmitManyJobs lops) =
250
  do
251
    open <- isQueueOpen
252
    if not open
253
      then return . Bad . GenericError $ "Queue drained"
254
      else do
255
        let mcs = Config.getMasterCandidates cfg
256
        result_jobids <- allocateJobIds mcs qlock (length lops)
257
        case result_jobids of
258
          Bad s -> return . Bad . GenericError $ s
259
          Ok jids -> do
260
            ts <- currentTimestamp
261
            jobs <- liftM (map $ setReceivedTimestamp ts)
262
                      $ zipWithM queuedJobFromOpCodes jids lops
263
            qDir <- queueDir
264
            write_results <- mapM (writeJobToDisk qDir) jobs
265
            let annotated_results = zip write_results jobs
266
                succeeded = map snd $ filter (isOk . fst) annotated_results
267
            when (any isBad write_results) . logWarning
268
              $ "Writing some jobs failed " ++ show annotated_results
269
            replicateManyJobs qDir mcs succeeded
270
            _ <- forkIO $ enqueueNewJobs qstat succeeded
271
            return . Ok . JSArray
272
              . map (\(res, job) ->
273
                      if isOk res
274
                        then showJSON (True, fromJobId $ qjId job)
275
                        else showJSON (False, genericResult id (const "") res))
276
              $ annotated_results
277

    
278
handleCall _ _ cfg (WaitForJobChange jid fields prev_job prev_log tmout) = do
279
  let compute_fn = computeJobUpdate cfg jid fields prev_log
280
  qDir <- queueDir
281
  -- verify if the job is finalized, and return immediately in this case
282
  jobresult <- loadJobFromDisk qDir False jid
283
  case jobresult of
284
    Ok (job, _) | not (jobFinalized job) -> do
285
      let jobfile = liveJobFile qDir jid
286
      answer <- watchFile jobfile (min tmout C.luxiWfjcTimeout)
287
                  (prev_job, JSArray []) compute_fn
288
      return . Ok $ showJSON answer
289
    _ -> liftM (Ok . showJSON) compute_fn
290

    
291
handleCall _ _ cfg (SetWatcherPause time) = do
292
  let mcs = Config.getMasterCandidates cfg
293
      masters = genericResult (const []) return
294
                  . Config.getNode cfg . clusterMasterNode
295
                  $ configCluster cfg
296
  _ <- executeRpcCall (masters ++ mcs) $ RpcCallSetWatcherPause time
297
  return . Ok . maybe JSNull showJSON $ time
298

    
299
handleCall _ _ cfg (SetDrainFlag value) = do
300
  let mcs = Config.getMasterCandidates cfg
301
  fpath <- jobQueueDrainFile
302
  if value
303
     then writeFile fpath ""
304
     else removeFile fpath
305
  _ <- executeRpcCall mcs $ RpcCallSetDrainFlag value
306
  return . Ok . showJSON $ True
307

    
308
handleCall _ qstat  cfg (CancelJob jid) = do
309
  let jName = (++) "job " . show $ fromJobId jid
310
  dequeueResult <- dequeueJob qstat jid
311
  case dequeueResult of
312
    Ok True -> do
313
      logDebug $ jName ++ " dequeued, marking as canceled"
314
      qDir <- queueDir
315
      readResult <- loadJobFromDisk qDir True jid
316
      let jobFileFailed = return . Ok . showJSON . (,) False
317
                            . (++) ("Dequeued " ++ jName
318
                                    ++ ", but failed to mark as cancelled: ")
319
                          :: String -> IO (ErrorResult JSValue)
320
      case readResult of
321
        Bad s -> jobFileFailed s
322
        Ok (job, _) -> do
323
          now <- currentTimestamp
324
          let job' = cancelQueuedJob now job
325
              mcs = Config.getMasterCandidates cfg
326
          write_result <- writeJobToDisk qDir job'
327
          case write_result of
328
            Bad s -> jobFileFailed s
329
            Ok () -> do
330
              replicateManyJobs qDir mcs [job']
331
              return . Ok . showJSON $ (True, "Dequeued " ++ jName)
332
    Ok False -> do
333
      logDebug $ jName ++ " not queued; trying to cancel directly"
334
      cancelJob jid
335
    Bad s -> return . Ok . showJSON $ (False, s)
336

    
337
handleCall qlock _ cfg (ArchiveJob jid) = do
338
  let archiveFailed = putMVar qlock  () >> (return . Ok $ showJSON False)
339
                      :: IO (ErrorResult JSValue)
340
  qDir <- queueDir
341
  takeMVar qlock
342
  result <- loadJobFromDisk qDir False jid
343
  case result of
344
    Bad _ -> archiveFailed
345
    Ok (job, _) -> if jobFinalized job
346
                     then do
347
                       let mcs = Config.getMasterCandidates cfg
348
                           live = liveJobFile qDir jid
349
                           archive = archivedJobFile qDir jid
350
                       renameResult <- safeRenameFile live archive
351
                       putMVar qlock ()
352
                       case renameResult of
353
                         Bad s -> return . Bad . JobQueueError
354
                                    $ "Archiving failed in an unexpected way: "
355
                                        ++ s
356
                         Ok () -> do
357
                           _ <- executeRpcCall mcs
358
                                  $ RpcCallJobqueueRename [(live, archive)]
359
                           return . Ok $ showJSON True
360
                     else archiveFailed
361

    
362
handleCall _ _ _ op =
363
  return . Bad $
364
    GenericError ("Luxi call '" ++ strOfOp op ++ "' not implemented")
365

    
366
{-# ANN handleCall "HLint: ignore Too strict if" #-}
367

    
368
-- | Query the status of a job and return the requested fields
369
-- and the logs newer than the given log number.
370
computeJobUpdate :: ConfigData -> JobId -> [String] -> JSValue
371
                    -> IO (JSValue, JSValue)
372
computeJobUpdate cfg jid fields prev_log = do
373
  let sjid = show $ fromJobId jid
374
  logDebug $ "Inspecting fields " ++ show fields ++ " of job " ++ sjid
375
  let fromJSArray (JSArray xs) = xs
376
      fromJSArray _ = []
377
  let logFilter JSNull (JSArray _) = True
378
      logFilter (JSRational _ n) (JSArray (JSRational _ m:_)) = n < m
379
      logFilter _ _ = False
380
  let filterLogs n logs = JSArray (filter (logFilter n) (logs >>= fromJSArray))
381
  jobQuery <- handleClassicQuery cfg (Qlang.ItemTypeLuxi Qlang.QRJob)
382
                [Right . fromIntegral $ fromJobId jid] ("oplog" : fields) False
383
  let (rfields, rlogs) = case jobQuery of
384
        Ok (JSArray [JSArray (JSArray logs : answer)]) ->
385
          (answer, filterLogs prev_log logs)
386
        _ -> (map (const JSNull) fields, JSArray [])
387
  logDebug $ "Updates for job " ++ sjid ++ " are " ++ encode (rfields, rlogs)
388
  return (JSArray rfields, rlogs)
389

    
390

    
391
type LuxiConfig = (MVar (), JQStatus, ConfigReader)
392

    
393
luxiExec
394
    :: LuxiConfig
395
    -> LuxiOp
396
    -> IO (Bool, GenericResult GanetiException JSValue)
397
luxiExec (qlock, qstat, creader) args = do
398
  cfg <- creader
399
  result <- handleCallWrapper qlock qstat cfg args
400
  return (True, result)
401

    
402
luxiHandler :: LuxiConfig -> U.Handler LuxiOp JSValue
403
luxiHandler cfg = U.Handler { U.hParse         = decodeLuxiCall
404
                            , U.hInputLogShort = strOfOp
405
                            , U.hInputLogLong  = show
406
                            , U.hExec          = luxiExec cfg
407
                            }
408

    
409
-- | Type alias for prepMain results
410
type PrepResult = (Server, IORef (Result ConfigData), JQStatus)
411

    
412
-- | Check function for luxid.
413
checkMain :: CheckFn ()
414
checkMain _ = return $ Right ()
415

    
416
-- | Prepare function for luxid.
417
prepMain :: PrepFn () PrepResult
418
prepMain _ _ = do
419
  socket_path <- Path.defaultQuerySocket
420
  cleanupSocket socket_path
421
  s <- describeError "binding to the Luxi socket"
422
         Nothing (Just socket_path) $ getLuxiServer True socket_path
423
  cref <- newIORef (Bad "Configuration not yet loaded")
424
  jq <- emptyJQStatus cref
425
  return (s, cref, jq)
426

    
427
-- | Main function.
428
main :: MainFn () PrepResult
429
main _ _ (server, cref, jq) = do
430
  initConfigReader id cref
431
  let creader = readIORef cref
432
  initJQScheduler jq
433

    
434
  qlockFile <- jobQueueLockFile
435
  lockFile qlockFile >>= exitIfBad "Failed to obtain the job-queue lock"
436
  qlock <- newMVar ()
437

    
438
  finally
439
    (forever $ U.listener (luxiHandler (qlock, jq, creader)) server)
440
    (closeServer server)