Statistics
| Branch: | Tag: | Revision:

root / src / Ganeti / Query / Server.hs @ 178ad717

History | View | Annotate | Download (16.7 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)
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 _ _ _ op =
338
  return . Bad $
339
    GenericError ("Luxi call '" ++ strOfOp op ++ "' not implemented")
340

    
341
{-# ANN handleCall "HLint: ignore Too strict if" #-}
342

    
343
-- | Query the status of a job and return the requested fields
344
-- and the logs newer than the given log number.
345
computeJobUpdate :: ConfigData -> JobId -> [String] -> JSValue
346
                    -> IO (JSValue, JSValue)
347
computeJobUpdate cfg jid fields prev_log = do
348
  let sjid = show $ fromJobId jid
349
  logDebug $ "Inspecting fields " ++ show fields ++ " of job " ++ sjid
350
  let fromJSArray (JSArray xs) = xs
351
      fromJSArray _ = []
352
  let logFilter JSNull (JSArray _) = True
353
      logFilter (JSRational _ n) (JSArray (JSRational _ m:_)) = n < m
354
      logFilter _ _ = False
355
  let filterLogs n logs = JSArray (filter (logFilter n) (logs >>= fromJSArray))
356
  jobQuery <- handleClassicQuery cfg (Qlang.ItemTypeLuxi Qlang.QRJob)
357
                [Right . fromIntegral $ fromJobId jid] ("oplog" : fields) False
358
  let (rfields, rlogs) = case jobQuery of
359
        Ok (JSArray [JSArray (JSArray logs : answer)]) ->
360
          (answer, filterLogs prev_log logs)
361
        _ -> (map (const JSNull) fields, JSArray [])
362
  logDebug $ "Updates for job " ++ sjid ++ " are " ++ encode (rfields, rlogs)
363
  return (JSArray rfields, rlogs)
364

    
365

    
366
type LuxiConfig = (MVar (), JQStatus, ConfigReader)
367

    
368
luxiExec
369
    :: LuxiConfig
370
    -> LuxiOp
371
    -> IO (Bool, GenericResult GanetiException JSValue)
372
luxiExec (qlock, qstat, creader) args = do
373
  cfg <- creader
374
  result <- handleCallWrapper qlock qstat cfg args
375
  return (True, result)
376

    
377
luxiHandler :: LuxiConfig -> U.Handler LuxiOp JSValue
378
luxiHandler cfg = U.Handler { U.hParse         = decodeLuxiCall
379
                            , U.hInputLogShort = strOfOp
380
                            , U.hInputLogLong  = show
381
                            , U.hExec          = luxiExec cfg
382
                            }
383

    
384
-- | Type alias for prepMain results
385
type PrepResult = (Server, IORef (Result ConfigData), JQStatus)
386

    
387
-- | Check function for luxid.
388
checkMain :: CheckFn ()
389
checkMain _ = return $ Right ()
390

    
391
-- | Prepare function for luxid.
392
prepMain :: PrepFn () PrepResult
393
prepMain _ _ = do
394
  socket_path <- Path.defaultQuerySocket
395
  cleanupSocket socket_path
396
  s <- describeError "binding to the Luxi socket"
397
         Nothing (Just socket_path) $ getLuxiServer True socket_path
398
  cref <- newIORef (Bad "Configuration not yet loaded")
399
  jq <- emptyJQStatus
400
  return (s, cref, jq)
401

    
402
-- | Main function.
403
main :: MainFn () PrepResult
404
main _ _ (server, cref, jq) = do
405
  initConfigReader id cref
406
  let creader = readIORef cref
407
  initJQScheduler jq
408

    
409
  qlockFile <- jobQueueLockFile
410
  lockFile qlockFile >>= exitIfBad "Failed to obtain the job-queue lock"
411
  qlock <- newMVar ()
412

    
413
  finally
414
    (forever $ U.listener (luxiHandler (qlock, jq, creader)) server)
415
    (closeServer server)