Synnefo » snf-ganeti » ganeti-local

	htools/Ganeti/Query/Job.hs \

import Ganeti.Query.Filter

import qualified Ganeti.Query.Job as Job

-- ** Node queries

-- ** Group queries

-- ** Job queries

-- | Tests that querying any existing fields, via either query or

-- queryFields, will not return unknown fields. This uses 'undefined'

-- for config, as job queries shouldn't use the configuration, and an

-- explicit filter as otherwise non-live queries wouldn't return any

-- result rows.

prop_queryJob_noUnknown :: Property

prop_queryJob_noUnknown =

  forAll (listOf (arbitrary::Gen (Positive Integer))) $ \ids ->

  forAll (elements (Map.keys Job.fieldsMap)) $ \field -> monadicIO $ do

  let qtype = ItemTypeLuxi QRJob

      flt = makeSimpleFilter (nameField qtype) $

            map (\(Positive i) -> Right i) ids

  QueryResult fdefs fdata <-

    run (query undefined False (Query qtype [field] flt)) >>= resultProp

  QueryFieldsResult fdefs' <-

    resultProp $ queryFields (QueryFields qtype [field])

  stop $ conjoin

         [ printTestCase ("Got unknown fields via query (" ++

                          show fdefs ++ ")") (hasUnknownFields fdefs)

         , printTestCase ("Got unknown result status via query (" ++

                          show fdata ++ ")")

           (all (all ((/= RSUnknown) . rentryStatus)) fdata)

         , printTestCase ("Got unknown fields via query fields (" ++

                          show fdefs'++ ")") (hasUnknownFields fdefs')

         ]

-- | Tests that an unknown field is returned as such.

prop_queryJob_Unknown :: Property

prop_queryJob_Unknown =

  forAll (listOf (arbitrary::Gen (Positive Integer))) $ \ids ->

  forAll (arbitrary `suchThat` (`notElem` Map.keys Job.fieldsMap))

    $ \field -> monadicIO $ do

  let qtype = ItemTypeLuxi QRJob

      flt = makeSimpleFilter (nameField qtype) $

            map (\(Positive i) -> Right i) ids

  QueryResult fdefs fdata <-

    run (query undefined False (Query qtype [field] flt)) >>= resultProp

  QueryFieldsResult fdefs' <-

    resultProp $ queryFields (QueryFields qtype [field])

  stop $ conjoin

         [ printTestCase ("Got known fields via query (" ++ show fdefs ++ ")")

           (not $ hasUnknownFields fdefs)

         , printTestCase ("Got /= ResultUnknown result status via query (" ++

                          show fdata ++ ")")

           (all (all ((== RSUnknown) . rentryStatus)) fdata)

         , printTestCase ("Got a Just in a result value (" ++

                          show fdata ++ ")")

           (all (all (isNothing . rentryValue)) fdata)

         , printTestCase ("Got known fields via query fields (" ++ show fdefs'

                          ++ ")") (not $ hasUnknownFields fdefs')

         ]

-- ** Misc other tests

  , 'prop_queryJob_noUnknown

  , 'prop_queryJob_Unknown

{-| Implementation of the Ganeti Query2 job queries.

 -}

{-

Copyright (C) 2012 Google Inc.

This program is free software; you can redistribute it and/or modify

it under the terms of the GNU General Public License as published by

the Free Software Foundation; either version 2 of the License, or

(at your option) any later version.

This program is distributed in the hope that it will be useful, but

WITHOUT ANY WARRANTY; without even the implied warranty of

MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

General Public License for more details.

You should have received a copy of the GNU General Public License

along with this program; if not, write to the Free Software

Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

02110-1301, USA.

-}

module Ganeti.Query.Job

  ( RuntimeData

  , fieldsMap

  , loadRuntimeData

  , wantArchived

  ) where

import qualified Data.Map as Map

import qualified Text.JSON as J

import Ganeti.BasicTypes

import qualified Ganeti.Constants as C

import Ganeti.JQueue

import Ganeti.OpCodes (opSummary, metaOpCode)

import Ganeti.Path

import Ganeti.Query.Common

import Ganeti.Query.Language

import Ganeti.Query.Types

import Ganeti.Types

-- | The runtime data for a job.

type RuntimeData = Result (QueuedJob, Bool)

-- | Job priority explanation.

jobPrioDoc :: String

jobPrioDoc = "Current job priority (" ++ show C.opPrioLowest ++ " to " ++

             show C.opPrioHighest ++ ")"

-- | Timestamp doc.

tsDoc :: String -> String

tsDoc = (++ " (tuple containing seconds and microseconds)")

-- | Wrapper for unavailable job.

maybeJob :: (J.JSON a) =>

            (QueuedJob -> a) -> RuntimeData -> JobId -> ResultEntry

maybeJob _ (Bad _) _      = rsUnavail

maybeJob f (Ok (v, _))  _ = rsNormal $ f v

-- | Simple helper for a job getter.

jobGetter :: (J.JSON a) => (QueuedJob -> a) -> FieldGetter JobId RuntimeData

jobGetter = FieldRuntime . maybeJob

-- | Simple helper for a per-opcode getter.

opsGetter :: (J.JSON a) => (QueuedOpCode -> a) -> FieldGetter JobId RuntimeData

opsGetter f = FieldRuntime $ maybeJob (map f . qjOps)

-- | Archived field name.

archivedField :: String

archivedField = "archived"

-- | Check whether we should look at archived jobs as well.

wantArchived :: [FilterField] -> Bool

wantArchived = (archivedField `elem`)

-- | List of all node fields. FIXME: QFF_JOB_ID on the id field.

jobFields :: FieldList JobId RuntimeData

jobFields =

  [ (FieldDefinition "id" "ID" QFTNumber "Job ID", FieldSimple rsNormal,

     QffNormal)

  , (FieldDefinition "status" "Status" QFTText "Job status",

     jobGetter calcJobStatus, QffNormal)

  , (FieldDefinition "priority" "Priority" QFTNumber jobPrioDoc,

     jobGetter calcJobPriority, QffNormal)

  , (FieldDefinition archivedField "Archived" QFTBool

       "Whether job is archived",

     FieldRuntime (\jinfo _ -> case jinfo of

                                 Ok (_, archive) -> rsNormal archive

                                 _ -> rsUnavail), QffNormal)

  , (FieldDefinition "ops" "OpCodes" QFTOther "List of all opcodes",

     opsGetter qoInput, QffNormal)

  , (FieldDefinition "opresult" "OpCode_result" QFTOther

       "List of opcodes results", opsGetter qoResult, QffNormal)

  , (FieldDefinition "opstatus" "OpCode_status" QFTOther

       "List of opcodes status", opsGetter qoStatus, QffNormal)

  , (FieldDefinition "oplog" "OpCode_log" QFTOther

       "List of opcode output logs", opsGetter qoLog, QffNormal)

  , (FieldDefinition "opstart" "OpCode_start" QFTOther

       "List of opcode start timestamps (before acquiring locks)",

     opsGetter qoStartTimestamp, QffNormal)

  , (FieldDefinition "opexec" "OpCode_exec" QFTOther

       "List of opcode execution start timestamps (after acquiring locks)",

     opsGetter qoExecTimestamp, QffNormal)

  , (FieldDefinition "opend" "OpCode_end" QFTOther

       "List of opcode execution end timestamps",

     opsGetter qoEndTimestamp, QffNormal)

  , (FieldDefinition "oppriority" "OpCode_prio" QFTOther

       "List of opcode priorities", opsGetter qoPriority, QffNormal)

  , (FieldDefinition "summary" "Summary" QFTOther

       "List of per-opcode summaries",

     opsGetter (opSummary . metaOpCode . qoInput), QffNormal)

  , (FieldDefinition "received_ts" "Received" QFTOther

       (tsDoc "Timestamp of when job was received"),

     jobGetter qjReceivedTimestamp, QffTimestamp)

  , (FieldDefinition "start_ts" "Start" QFTOther

       (tsDoc "Timestamp of job start"),

     jobGetter qjStartTimestamp, QffTimestamp)

  , (FieldDefinition "end_ts" "End" QFTOther

       (tsDoc "Timestamp of job end"),

     jobGetter qjEndTimestamp, QffTimestamp)

  ]

-- | The node fields map.

fieldsMap :: FieldMap JobId RuntimeData

fieldsMap =

  Map.fromList $ map (\v@(f, _, _) -> (fdefName f, v)) jobFields

-- | Load the given jobs from disk.

loadRuntimeData :: [JobId] -> Bool -> IO [RuntimeData]

loadRuntimeData ids archived = do

  qdir <- queueDir

  mapM (loadJobFromDisk qdir archived) ids

import Control.DeepSeq

import Control.Monad (filterM, liftM, foldM)

import Ganeti.Errors

import Ganeti.JQueue

import Ganeti.Objects

import qualified Ganeti.Query.Job as Query.Job

import Ganeti.Query.Language

import Ganeti.Query.Node

import Ganeti.Query.Types

import Ganeti.Path

import Ganeti.Types

-- | Compute the requested job IDs. This is custom since we need to

-- handle both strings and integers.

getRequestedJobIDs :: Filter FilterField -> Result [JobId]

getRequestedJobIDs qfilter =

  case requestedNames (nameField (ItemTypeLuxi QRJob)) qfilter of

    Nothing -> Ok []

    Just [] -> Ok []

    Just vals ->

      mapM (\e -> case e of

                    QuotedString s -> makeJobIdS s

                    NumericValue i -> makeJobId $ fromIntegral i

           ) vals

query cfg live (Query (ItemTypeLuxi QRJob) fields qfilter) =

  queryJobs cfg live fields qfilter

-- | Query jobs specific query function, needed as we need to accept

-- both 'QuotedString' and 'NumericValue' as wanted names.

queryJobs :: ConfigData                   -- ^ The current configuration

          -> Bool                         -- ^ Whether to collect live data

          -> [FilterField]                -- ^ Item

          -> Filter FilterField           -- ^ Filter

          -> IO (ErrorResult QueryResult) -- ^ Result

queryJobs cfg live fields qfilter =

  runResultT $ do

  rootdir <- lift queueDir

  let wanted_names = getRequestedJobIDs qfilter

      want_arch = Query.Job.wantArchived fields

  rjids <- case wanted_names of

             Bad msg -> resultT . Bad $ GenericError msg

             Ok [] -> if live

                        -- we can check the filesystem for actual jobs

                        then lift $ liftM sortJobIDs

                             (determineJobDirectories rootdir want_arch >>=

                              getJobIDs)

                        -- else we shouldn't look at the filesystem...

                        else return []

             Ok v -> resultT $ Ok v

  cfilter <- resultT $ compileFilter Query.Job.fieldsMap qfilter

  let selected = getSelectedFields Query.Job.fieldsMap fields

      (fdefs, fgetters, _) = unzip3 selected

      live' = live && needsLiveData fgetters

      disabled_data = Bad "live data disabled"

  -- runs first pass of the filter, without a runtime context; this

  -- will limit the jobs that we'll load from disk

  jids <- resultT $

          filterM (\jid -> evaluateFilter cfg Nothing jid cfilter) rjids

  -- here we run the runtime data gathering, filtering and evaluation,

  -- all in the same step, so that we don't keep jobs in memory longer

  -- than we need; we can't be fully lazy due to the multiple monad

  -- wrapping across different steps

  qdir <- lift queueDir

  fdata <- foldM

           -- big lambda, but we use many variables from outside it...

           (\lst jid -> do

              job <- lift $ if live'

                              then loadJobFromDisk qdir want_arch jid

                              else return disabled_data

              pass <- resultT $ evaluateFilter cfg (Just job) jid cfilter

              let nlst = if pass

                           then let row = map (execGetter cfg job jid) fgetters

                                in rnf row `seq` row:lst

                           else lst

              -- evaluate nlst (to WHNF), otherwise we're too lazy

              return $! nlst

           ) [] jids

  return QueryResult { qresFields = fdefs, qresData = reverse fdata }

queryFields (QueryFields (ItemTypeLuxi QRJob) fields) =

  Ok $ fieldsExtractor Query.Job.fieldsMap fields

handleCall cfg (QueryJobs names fields) =

  handleClassicQuery cfg (Qlang.ItemTypeLuxi Qlang.QRJob)

    (map (Right . fromIntegral . fromJobId) names)  fields False