Statistics
| Branch: | Tag: | Revision:

root / src / Ganeti / Confd / Client.hs @ 7dc27988

History | View | Annotate | Download (4.9 kB)

1
{-| Implementation of the Ganeti Confd client functionality.
2

    
3
-}
4

    
5
{-
6

    
7
Copyright (C) 2012 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.Confd.Client
27
  ( getConfdClient
28
  , query
29
  ) where
30

    
31
import Control.Concurrent
32
import Control.Monad
33
import Data.List
34
import Data.Maybe
35
import qualified Network.Socket as S
36
import System.Posix.Time
37
import qualified Text.JSON as J
38

    
39
import Ganeti.BasicTypes
40
import Ganeti.Confd.Types
41
import Ganeti.Confd.Utils
42
import qualified Ganeti.Constants as C
43
import Ganeti.Hash
44
import Ganeti.Ssconf
45

    
46
-- | Builds a properly initialized ConfdClient.
47
-- The parameters (an IP address and the port number for the Confd client
48
-- to connect to) are mainly meant for testing purposes. If they are not
49
-- provided, the list of master candidates and the default port number will
50
-- be used.
51
getConfdClient :: Maybe String -> Maybe Int -> IO ConfdClient
52
getConfdClient addr portNum = S.withSocketsDo $ do
53
  hmac <- getClusterHmac
54
  candList <- getMasterCandidatesIps Nothing
55
  peerList <-
56
    case candList of
57
      (Ok p) -> return p
58
      (Bad msg) -> fail msg
59
  let addrList = maybe peerList (:[]) addr
60
      port = fromMaybe C.defaultConfdPort portNum
61
  return . ConfdClient hmac addrList $ fromIntegral port
62

    
63
-- | Sends a query to all the Confd servers the client is connected to.
64
-- Returns the most up-to-date result according to the serial number,
65
-- chosen between those received before the timeout.
66
query :: ConfdClient -> ConfdRequestType -> ConfdQuery -> IO (Maybe ConfdReply)
67
query client crType cQuery = do
68
  semaphore <- newMVar ()
69
  answer <- newMVar Nothing
70
  let dest = [(host, serverPort client) | host <- peers client]
71
      hmac = hmacKey client
72
      jobs = map (queryOneServer semaphore answer crType cQuery hmac) dest
73
      watchdog reqAnswers = do
74
        threadDelay $ 1000000 * C.confdClientExpireTimeout
75
        _ <- swapMVar reqAnswers 0
76
        putMVar semaphore ()
77
      waitForResult reqAnswers = do
78
        _ <- takeMVar semaphore
79
        l <- takeMVar reqAnswers
80
        unless (l == 0) $ do
81
          putMVar reqAnswers $ l - 1
82
          waitForResult reqAnswers
83
  reqAnswers <- newMVar . min C.confdDefaultReqCoverage $ length dest
84
  workers <- mapM forkIO jobs
85
  watcher <- forkIO $ watchdog reqAnswers
86
  waitForResult reqAnswers
87
  mapM_ killThread $ watcher:workers
88
  takeMVar answer
89

    
90
-- | Updates the reply to the query. As per the Confd design document,
91
-- only the reply with the highest serial number is kept.
92
updateConfdReply :: ConfdReply -> Maybe ConfdReply -> Maybe ConfdReply
93
updateConfdReply newValue Nothing = Just newValue
94
updateConfdReply newValue (Just currentValue) = Just $
95
  if confdReplyStatus newValue == ReplyStatusOk
96
      && (confdReplyStatus currentValue /= ReplyStatusOk
97
          || confdReplySerial newValue > confdReplySerial currentValue)
98
    then newValue
99
    else currentValue
100

    
101
-- | Send a query to a single server, waits for the result and stores it
102
-- in a shared variable. Then, sends a signal on another shared variable
103
-- acting as a semaphore.
104
-- This function is meant to be used as one of multiple threads querying
105
-- multiple servers in parallel.
106
queryOneServer
107
  :: MVar ()                 -- ^ The semaphore that will be signalled
108
  -> MVar (Maybe ConfdReply) -- ^ The shared variable for the result
109
  -> ConfdRequestType        -- ^ The type of the query to be sent
110
  -> ConfdQuery              -- ^ The content of the query
111
  -> HashKey                 -- ^ The hmac key to sign the message
112
  -> (String, S.PortNumber)  -- ^ The address and port of the server
113
  -> IO ()
114
queryOneServer semaphore answer crType cQuery hmac (host, port) = do
115
  request <- newConfdRequest crType cQuery
116
  timestamp <- fmap show epochTime
117
  let signedMsg =
118
        signMessage hmac timestamp (J.encodeStrict request)
119
      completeMsg = C.confdMagicFourcc ++ J.encodeStrict signedMsg
120
  s <- S.socket S.AF_INET S.Datagram S.defaultProtocol
121
  hostAddr <- S.inet_addr host
122
  _ <- S.sendTo s completeMsg $ S.SockAddrInet port hostAddr
123
  replyMsg <- S.recv s C.maxUdpDataSize
124
  parsedReply <-
125
    if C.confdMagicFourcc `isPrefixOf` replyMsg
126
      then return . parseReply hmac (drop 4 replyMsg) $ confdRqRsalt request
127
      else fail "Invalid magic code!"
128
  reply <-
129
    case parsedReply of
130
      Ok (_, r) -> return r
131
      Bad msg -> fail msg
132
  modifyMVar_ answer $! return . updateConfdReply reply
133
  putMVar semaphore ()