Statistics
| Branch: | Tag: | Revision:

root / lib / confd / client.py @ 23057d29

History | View | Annotate | Download (12.7 kB)

1 e4ccf6cd Guido Trotter
#
2 e4ccf6cd Guido Trotter
#
3 e4ccf6cd Guido Trotter
4 e4ccf6cd Guido Trotter
# Copyright (C) 2009 Google Inc.
5 e4ccf6cd Guido Trotter
#
6 e4ccf6cd Guido Trotter
# This program is free software; you can redistribute it and/or modify
7 e4ccf6cd Guido Trotter
# it under the terms of the GNU General Public License as published by
8 e4ccf6cd Guido Trotter
# the Free Software Foundation; either version 2 of the License, or
9 e4ccf6cd Guido Trotter
# (at your option) any later version.
10 e4ccf6cd Guido Trotter
#
11 e4ccf6cd Guido Trotter
# This program is distributed in the hope that it will be useful, but
12 e4ccf6cd Guido Trotter
# WITHOUT ANY WARRANTY; without even the implied warranty of
13 e4ccf6cd Guido Trotter
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 e4ccf6cd Guido Trotter
# General Public License for more details.
15 e4ccf6cd Guido Trotter
#
16 e4ccf6cd Guido Trotter
# You should have received a copy of the GNU General Public License
17 e4ccf6cd Guido Trotter
# along with this program; if not, write to the Free Software
18 e4ccf6cd Guido Trotter
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19 e4ccf6cd Guido Trotter
# 02110-1301, USA.
20 e4ccf6cd Guido Trotter
21 e4ccf6cd Guido Trotter
22 e4ccf6cd Guido Trotter
"""Ganeti confd client
23 e4ccf6cd Guido Trotter

24 cf7b0cc4 Guido Trotter
Clients can use the confd client library to send requests to a group of master
25 cf7b0cc4 Guido Trotter
candidates running confd. The expected usage is through the asyncore framework,
26 cf7b0cc4 Guido Trotter
by sending queries, and asynchronously receiving replies through a callback.
27 cf7b0cc4 Guido Trotter

28 cf7b0cc4 Guido Trotter
This way the client library doesn't ever need to "wait" on a particular answer,
29 cf7b0cc4 Guido Trotter
and can proceed even if some udp packets are lost. It's up to the user to
30 cf7b0cc4 Guido Trotter
reschedule queries if they haven't received responses and they need them.
31 cf7b0cc4 Guido Trotter

32 69b99987 Michael Hanselmann
Example usage::
33 69b99987 Michael Hanselmann

34 cf7b0cc4 Guido Trotter
  client = ConfdClient(...) # includes callback specification
35 cf7b0cc4 Guido Trotter
  req = confd_client.ConfdClientRequest(type=constants.CONFD_REQ_PING)
36 cf7b0cc4 Guido Trotter
  client.SendRequest(req)
37 cf7b0cc4 Guido Trotter
  # then make sure your client calls asyncore.loop() or daemon.Mainloop.Run()
38 cf7b0cc4 Guido Trotter
  # ... wait ...
39 cf7b0cc4 Guido Trotter
  # And your callback will be called by asyncore, when your query gets a
40 cf7b0cc4 Guido Trotter
  # response, or when it expires.
41 cf7b0cc4 Guido Trotter

42 392ca296 Guido Trotter
You can use the provided ConfdFilterCallback to act as a filter, only passing
43 392ca296 Guido Trotter
"newer" answer to your callback, and filtering out outdated ones, or ones
44 392ca296 Guido Trotter
confirming what you already got.
45 392ca296 Guido Trotter

46 e4ccf6cd Guido Trotter
"""
47 69b99987 Michael Hanselmann
48 e4ccf6cd Guido Trotter
import socket
49 e4ccf6cd Guido Trotter
import time
50 e4ccf6cd Guido Trotter
import random
51 e4ccf6cd Guido Trotter
52 e4ccf6cd Guido Trotter
from ganeti import utils
53 e4ccf6cd Guido Trotter
from ganeti import constants
54 e4ccf6cd Guido Trotter
from ganeti import objects
55 e4ccf6cd Guido Trotter
from ganeti import serializer
56 e4ccf6cd Guido Trotter
from ganeti import daemon # contains AsyncUDPSocket
57 e4ccf6cd Guido Trotter
from ganeti import errors
58 e4ccf6cd Guido Trotter
from ganeti import confd
59 e4ccf6cd Guido Trotter
60 e4ccf6cd Guido Trotter
61 e4ccf6cd Guido Trotter
class ConfdAsyncUDPClient(daemon.AsyncUDPSocket):
62 e4ccf6cd Guido Trotter
  """Confd udp asyncore client
63 e4ccf6cd Guido Trotter

64 e4ccf6cd Guido Trotter
  This is kept separate from the main ConfdClient to make sure it's easy to
65 e4ccf6cd Guido Trotter
  implement a non-asyncore based client library.
66 e4ccf6cd Guido Trotter

67 e4ccf6cd Guido Trotter
  """
68 e4ccf6cd Guido Trotter
  def __init__(self, client):
69 e4ccf6cd Guido Trotter
    """Constructor for ConfdAsyncUDPClient
70 e4ccf6cd Guido Trotter

71 e4ccf6cd Guido Trotter
    @type client: L{ConfdClient}
72 e4ccf6cd Guido Trotter
    @param client: client library, to pass the datagrams to
73 e4ccf6cd Guido Trotter

74 e4ccf6cd Guido Trotter
    """
75 e4ccf6cd Guido Trotter
    daemon.AsyncUDPSocket.__init__(self)
76 e4ccf6cd Guido Trotter
    self.client = client
77 e4ccf6cd Guido Trotter
78 e4ccf6cd Guido Trotter
  # this method is overriding a daemon.AsyncUDPSocket method
79 e4ccf6cd Guido Trotter
  def handle_datagram(self, payload, ip, port):
80 e4ccf6cd Guido Trotter
    self.client.HandleResponse(payload, ip, port)
81 e4ccf6cd Guido Trotter
82 e4ccf6cd Guido Trotter
83 e4ccf6cd Guido Trotter
class ConfdClient:
84 e4ccf6cd Guido Trotter
  """Send queries to confd, and get back answers.
85 e4ccf6cd Guido Trotter

86 e4ccf6cd Guido Trotter
  Since the confd model works by querying multiple master candidates, and
87 e4ccf6cd Guido Trotter
  getting back answers, this is an asynchronous library. It can either work
88 e4ccf6cd Guido Trotter
  through asyncore or with your own handling.
89 e4ccf6cd Guido Trotter

90 e4ccf6cd Guido Trotter
  """
91 a3db74e4 Guido Trotter
  def __init__(self, hmac_key, peers, callback, port=None, logger=None):
92 e4ccf6cd Guido Trotter
    """Constructor for ConfdClient
93 e4ccf6cd Guido Trotter

94 e4ccf6cd Guido Trotter
    @type hmac_key: string
95 e4ccf6cd Guido Trotter
    @param hmac_key: hmac key to talk to confd
96 e4ccf6cd Guido Trotter
    @type peers: list
97 e4ccf6cd Guido Trotter
    @param peers: list of peer nodes
98 96e03b0b Guido Trotter
    @type callback: f(L{ConfdUpcallPayload})
99 96e03b0b Guido Trotter
    @param callback: function to call when getting answers
100 7d20c647 Guido Trotter
    @type port: integer
101 7d20c647 Guido Trotter
    @keyword port: confd port (default: use GetDaemonPort)
102 69b99987 Michael Hanselmann
    @type logger: logging.Logger
103 a3db74e4 Guido Trotter
    @keyword logger: optional logger for internal conditions
104 e4ccf6cd Guido Trotter

105 e4ccf6cd Guido Trotter
    """
106 96e03b0b Guido Trotter
    if not callable(callback):
107 96e03b0b Guido Trotter
      raise errors.ProgrammerError("callback must be callable")
108 e4ccf6cd Guido Trotter
109 a5229439 Guido Trotter
    self.UpdatePeerList(peers)
110 e4ccf6cd Guido Trotter
    self._hmac_key = hmac_key
111 e4ccf6cd Guido Trotter
    self._socket = ConfdAsyncUDPClient(self)
112 96e03b0b Guido Trotter
    self._callback = callback
113 7d20c647 Guido Trotter
    self._confd_port = port
114 a3db74e4 Guido Trotter
    self._logger = logger
115 96e03b0b Guido Trotter
    self._requests = {}
116 96e03b0b Guido Trotter
    self._expire_requests = []
117 7d20c647 Guido Trotter
118 7d20c647 Guido Trotter
    if self._confd_port is None:
119 7d20c647 Guido Trotter
      self._confd_port = utils.GetDaemonPort(constants.CONFD)
120 e4ccf6cd Guido Trotter
121 a5229439 Guido Trotter
  def UpdatePeerList(self, peers):
122 a5229439 Guido Trotter
    """Update the list of peers
123 a5229439 Guido Trotter

124 a5229439 Guido Trotter
    @type peers: list
125 a5229439 Guido Trotter
    @param peers: list of peer nodes
126 a5229439 Guido Trotter

127 a5229439 Guido Trotter
    """
128 a5229439 Guido Trotter
    if not isinstance(peers, list):
129 a5229439 Guido Trotter
      raise errors.ProgrammerError("peers must be a list")
130 a5229439 Guido Trotter
    self._peers = peers
131 a5229439 Guido Trotter
132 e4ccf6cd Guido Trotter
  def _PackRequest(self, request, now=None):
133 e4ccf6cd Guido Trotter
    """Prepare a request to be sent on the wire.
134 e4ccf6cd Guido Trotter

135 e4ccf6cd Guido Trotter
    This function puts a proper salt in a confd request, puts the proper salt,
136 e4ccf6cd Guido Trotter
    and adds the correct magic number.
137 e4ccf6cd Guido Trotter

138 e4ccf6cd Guido Trotter
    """
139 e4ccf6cd Guido Trotter
    if now is None:
140 e4ccf6cd Guido Trotter
      now = time.time()
141 e4ccf6cd Guido Trotter
    tstamp = '%d' % now
142 e4ccf6cd Guido Trotter
    req = serializer.DumpSignedJson(request.ToDict(), self._hmac_key, tstamp)
143 e4ccf6cd Guido Trotter
    return confd.PackMagic(req)
144 e4ccf6cd Guido Trotter
145 e4ccf6cd Guido Trotter
  def _UnpackReply(self, payload):
146 e4ccf6cd Guido Trotter
    in_payload = confd.UnpackMagic(payload)
147 c103d7ae Guido Trotter
    (dict_answer, salt) = serializer.LoadSignedJson(in_payload, self._hmac_key)
148 c103d7ae Guido Trotter
    answer = objects.ConfdReply.FromDict(dict_answer)
149 e4ccf6cd Guido Trotter
    return answer, salt
150 e4ccf6cd Guido Trotter
151 96e03b0b Guido Trotter
  def ExpireRequests(self):
152 96e03b0b Guido Trotter
    """Delete all the expired requests.
153 e4ccf6cd Guido Trotter

154 e4ccf6cd Guido Trotter
    """
155 e4ccf6cd Guido Trotter
    now = time.time()
156 96e03b0b Guido Trotter
    while self._expire_requests:
157 96e03b0b Guido Trotter
      expire_time, rsalt = self._expire_requests[0]
158 e4ccf6cd Guido Trotter
      if now >= expire_time:
159 96e03b0b Guido Trotter
        self._expire_requests.pop(0)
160 96e03b0b Guido Trotter
        (request, args) = self._requests[rsalt]
161 96e03b0b Guido Trotter
        del self._requests[rsalt]
162 96e03b0b Guido Trotter
        client_reply = ConfdUpcallPayload(salt=rsalt,
163 96e03b0b Guido Trotter
                                          type=UPCALL_EXPIRE,
164 96e03b0b Guido Trotter
                                          orig_request=request,
165 5f6f260a Guido Trotter
                                          extra_args=args,
166 5f6f260a Guido Trotter
                                          client=self,
167 5f6f260a Guido Trotter
                                          )
168 96e03b0b Guido Trotter
        self._callback(client_reply)
169 e4ccf6cd Guido Trotter
      else:
170 e4ccf6cd Guido Trotter
        break
171 e4ccf6cd Guido Trotter
172 96e03b0b Guido Trotter
  def SendRequest(self, request, args=None, coverage=None):
173 e4ccf6cd Guido Trotter
    """Send a confd request to some MCs
174 e4ccf6cd Guido Trotter

175 e4ccf6cd Guido Trotter
    @type request: L{objects.ConfdRequest}
176 e4ccf6cd Guido Trotter
    @param request: the request to send
177 e4ccf6cd Guido Trotter
    @type args: tuple
178 90469357 Guido Trotter
    @keyword args: additional callback arguments
179 e4ccf6cd Guido Trotter
    @type coverage: integer
180 e4ccf6cd Guido Trotter
    @keyword coverage: number of remote nodes to contact
181 e4ccf6cd Guido Trotter

182 e4ccf6cd Guido Trotter
    """
183 e4ccf6cd Guido Trotter
    if coverage is None:
184 e4ccf6cd Guido Trotter
      coverage = min(len(self._peers), constants.CONFD_DEFAULT_REQ_COVERAGE)
185 e4ccf6cd Guido Trotter
186 e4ccf6cd Guido Trotter
    if coverage > len(self._peers):
187 e4ccf6cd Guido Trotter
      raise errors.ConfdClientError("Not enough MCs known to provide the"
188 e4ccf6cd Guido Trotter
                                    " desired coverage")
189 e4ccf6cd Guido Trotter
190 e4ccf6cd Guido Trotter
    if not request.rsalt:
191 e4ccf6cd Guido Trotter
      raise errors.ConfdClientError("Missing request rsalt")
192 e4ccf6cd Guido Trotter
193 96e03b0b Guido Trotter
    self.ExpireRequests()
194 96e03b0b Guido Trotter
    if request.rsalt in self._requests:
195 e4ccf6cd Guido Trotter
      raise errors.ConfdClientError("Duplicate request rsalt")
196 e4ccf6cd Guido Trotter
197 e4ccf6cd Guido Trotter
    if request.type not in constants.CONFD_REQS:
198 e4ccf6cd Guido Trotter
      raise errors.ConfdClientError("Invalid request type")
199 e4ccf6cd Guido Trotter
200 e4ccf6cd Guido Trotter
    random.shuffle(self._peers)
201 e4ccf6cd Guido Trotter
    targets = self._peers[:coverage]
202 e4ccf6cd Guido Trotter
203 e4ccf6cd Guido Trotter
    now = time.time()
204 e4ccf6cd Guido Trotter
    payload = self._PackRequest(request, now=now)
205 e4ccf6cd Guido Trotter
206 e4ccf6cd Guido Trotter
    for target in targets:
207 e4ccf6cd Guido Trotter
      try:
208 e4ccf6cd Guido Trotter
        self._socket.enqueue_send(target, self._confd_port, payload)
209 e4ccf6cd Guido Trotter
      except errors.UdpDataSizeError:
210 e4ccf6cd Guido Trotter
        raise errors.ConfdClientError("Request too big")
211 e4ccf6cd Guido Trotter
212 96e03b0b Guido Trotter
    self._requests[request.rsalt] = (request, args)
213 e4ccf6cd Guido Trotter
    expire_time = now + constants.CONFD_CLIENT_EXPIRE_TIMEOUT
214 96e03b0b Guido Trotter
    self._expire_requests.append((expire_time, request.rsalt))
215 e4ccf6cd Guido Trotter
216 e4ccf6cd Guido Trotter
  def HandleResponse(self, payload, ip, port):
217 e4ccf6cd Guido Trotter
    """Asynchronous handler for a confd reply
218 e4ccf6cd Guido Trotter

219 e4ccf6cd Guido Trotter
    Call the relevant callback associated to the current request.
220 e4ccf6cd Guido Trotter

221 e4ccf6cd Guido Trotter
    """
222 e4ccf6cd Guido Trotter
    try:
223 e4ccf6cd Guido Trotter
      try:
224 e4ccf6cd Guido Trotter
        answer, salt = self._UnpackReply(payload)
225 a3db74e4 Guido Trotter
      except (errors.SignatureError, errors.ConfdMagicError), err:
226 a3db74e4 Guido Trotter
        if self._logger:
227 a3db74e4 Guido Trotter
          self._logger.debug("Discarding broken package: %s" % err)
228 e4ccf6cd Guido Trotter
        return
229 e4ccf6cd Guido Trotter
230 e4ccf6cd Guido Trotter
      try:
231 96e03b0b Guido Trotter
        (request, args) = self._requests[salt]
232 e4ccf6cd Guido Trotter
      except KeyError:
233 a3db74e4 Guido Trotter
        if self._logger:
234 a3db74e4 Guido Trotter
          self._logger.debug("Discarding unknown (expired?) reply: %s" % err)
235 96e03b0b Guido Trotter
        return
236 96e03b0b Guido Trotter
237 96e03b0b Guido Trotter
      client_reply = ConfdUpcallPayload(salt=salt,
238 96e03b0b Guido Trotter
                                        type=UPCALL_REPLY,
239 96e03b0b Guido Trotter
                                        server_reply=answer,
240 96e03b0b Guido Trotter
                                        orig_request=request,
241 96e03b0b Guido Trotter
                                        server_ip=ip,
242 96e03b0b Guido Trotter
                                        server_port=port,
243 5f6f260a Guido Trotter
                                        extra_args=args,
244 5f6f260a Guido Trotter
                                        client=self,
245 5f6f260a Guido Trotter
                                       )
246 96e03b0b Guido Trotter
      self._callback(client_reply)
247 e4ccf6cd Guido Trotter
248 e4ccf6cd Guido Trotter
    finally:
249 96e03b0b Guido Trotter
      self.ExpireRequests()
250 96e03b0b Guido Trotter
251 96e03b0b Guido Trotter
252 96e03b0b Guido Trotter
# UPCALL_REPLY: server reply upcall
253 96e03b0b Guido Trotter
# has all ConfdUpcallPayload fields populated
254 96e03b0b Guido Trotter
UPCALL_REPLY = 1
255 96e03b0b Guido Trotter
# UPCALL_EXPIRE: internal library request expire
256 96e03b0b Guido Trotter
# has only salt, type, orig_request and extra_args
257 96e03b0b Guido Trotter
UPCALL_EXPIRE = 2
258 96e03b0b Guido Trotter
CONFD_UPCALL_TYPES = frozenset([
259 96e03b0b Guido Trotter
  UPCALL_REPLY,
260 96e03b0b Guido Trotter
  UPCALL_EXPIRE,
261 96e03b0b Guido Trotter
  ])
262 96e03b0b Guido Trotter
263 96e03b0b Guido Trotter
264 96e03b0b Guido Trotter
class ConfdUpcallPayload(objects.ConfigObject):
265 96e03b0b Guido Trotter
  """Callback argument for confd replies
266 96e03b0b Guido Trotter

267 96e03b0b Guido Trotter
  @type salt: string
268 96e03b0b Guido Trotter
  @ivar salt: salt associated with the query
269 96e03b0b Guido Trotter
  @type type: one of confd.client.CONFD_UPCALL_TYPES
270 96e03b0b Guido Trotter
  @ivar type: upcall type (server reply, expired request, ...)
271 96e03b0b Guido Trotter
  @type orig_request: L{objects.ConfdRequest}
272 96e03b0b Guido Trotter
  @ivar orig_request: original request
273 96e03b0b Guido Trotter
  @type server_reply: L{objects.ConfdReply}
274 96e03b0b Guido Trotter
  @ivar server_reply: server reply
275 96e03b0b Guido Trotter
  @type server_ip: string
276 96e03b0b Guido Trotter
  @ivar server_ip: answering server ip address
277 96e03b0b Guido Trotter
  @type server_port: int
278 96e03b0b Guido Trotter
  @ivar server_port: answering server port
279 96e03b0b Guido Trotter
  @type extra_args: any
280 96e03b0b Guido Trotter
  @ivar extra_args: 'args' argument of the SendRequest function
281 5f6f260a Guido Trotter
  @type client: L{ConfdClient}
282 5f6f260a Guido Trotter
  @ivar client: current confd client instance
283 96e03b0b Guido Trotter

284 96e03b0b Guido Trotter
  """
285 96e03b0b Guido Trotter
  __slots__ = [
286 96e03b0b Guido Trotter
    "salt",
287 96e03b0b Guido Trotter
    "type",
288 96e03b0b Guido Trotter
    "orig_request",
289 96e03b0b Guido Trotter
    "server_reply",
290 96e03b0b Guido Trotter
    "server_ip",
291 96e03b0b Guido Trotter
    "server_port",
292 96e03b0b Guido Trotter
    "extra_args",
293 5f6f260a Guido Trotter
    "client",
294 96e03b0b Guido Trotter
    ]
295 e4ccf6cd Guido Trotter
296 e4ccf6cd Guido Trotter
297 e4ccf6cd Guido Trotter
class ConfdClientRequest(objects.ConfdRequest):
298 e4ccf6cd Guido Trotter
  """This is the client-side version of ConfdRequest.
299 e4ccf6cd Guido Trotter

300 e4ccf6cd Guido Trotter
  This version of the class helps creating requests, on the client side, by
301 e4ccf6cd Guido Trotter
  filling in some default values.
302 e4ccf6cd Guido Trotter

303 e4ccf6cd Guido Trotter
  """
304 e4ccf6cd Guido Trotter
  def __init__(self, **kwargs):
305 e4ccf6cd Guido Trotter
    objects.ConfdRequest.__init__(self, **kwargs)
306 e4ccf6cd Guido Trotter
    if not self.rsalt:
307 e4ccf6cd Guido Trotter
      self.rsalt = utils.NewUUID()
308 e4ccf6cd Guido Trotter
    if not self.protocol:
309 e4ccf6cd Guido Trotter
      self.protocol = constants.CONFD_PROTOCOL_VERSION
310 e4ccf6cd Guido Trotter
    if self.type not in constants.CONFD_REQS:
311 e4ccf6cd Guido Trotter
      raise errors.ConfdClientError("Invalid request type")
312 e4ccf6cd Guido Trotter
313 392ca296 Guido Trotter
314 392ca296 Guido Trotter
class ConfdFilterCallback:
315 392ca296 Guido Trotter
  """Callback that calls another callback, but filters duplicate results.
316 392ca296 Guido Trotter

317 392ca296 Guido Trotter
  """
318 392ca296 Guido Trotter
  def __init__(self, callback, logger=None):
319 392ca296 Guido Trotter
    """Constructor for ConfdFilterCallback
320 392ca296 Guido Trotter

321 392ca296 Guido Trotter
    @type callback: f(L{ConfdUpcallPayload})
322 392ca296 Guido Trotter
    @param callback: function to call when getting answers
323 69b99987 Michael Hanselmann
    @type logger: logging.Logger
324 392ca296 Guido Trotter
    @keyword logger: optional logger for internal conditions
325 392ca296 Guido Trotter

326 392ca296 Guido Trotter
    """
327 392ca296 Guido Trotter
    if not callable(callback):
328 392ca296 Guido Trotter
      raise errors.ProgrammerError("callback must be callable")
329 392ca296 Guido Trotter
330 392ca296 Guido Trotter
    self._callback = callback
331 392ca296 Guido Trotter
    self._logger = logger
332 392ca296 Guido Trotter
    # answers contains a dict of salt -> answer
333 392ca296 Guido Trotter
    self._answers = {}
334 392ca296 Guido Trotter
335 392ca296 Guido Trotter
  def _LogFilter(self, salt, new_reply, old_reply):
336 392ca296 Guido Trotter
    if not self._logger:
337 392ca296 Guido Trotter
      return
338 392ca296 Guido Trotter
339 392ca296 Guido Trotter
    if new_reply.serial > old_reply.serial:
340 392ca296 Guido Trotter
      self._logger.debug("Filtering confirming answer, with newer"
341 392ca296 Guido Trotter
                         " serial for query %s" % salt)
342 392ca296 Guido Trotter
    elif new_reply.serial == old_reply.serial:
343 392ca296 Guido Trotter
      if new_reply.answer != old_reply.answer:
344 392ca296 Guido Trotter
        self._logger.warning("Got incoherent answers for query %s"
345 392ca296 Guido Trotter
                             " (serial: %s)" % (salt, new_reply.serial))
346 392ca296 Guido Trotter
      else:
347 392ca296 Guido Trotter
        self._logger.debug("Filtering confirming answer, with same"
348 392ca296 Guido Trotter
                           " serial for query %s" % salt)
349 392ca296 Guido Trotter
    else:
350 392ca296 Guido Trotter
      self._logger.debug("Filtering outdated answer for query %s"
351 392ca296 Guido Trotter
                         " serial: (%d < %d)" % (salt, old_reply.serial,
352 392ca296 Guido Trotter
                                                 new_reply.serial))
353 392ca296 Guido Trotter
354 392ca296 Guido Trotter
  def _HandleExpire(self, up):
355 392ca296 Guido Trotter
    # if we have no answer we have received none, before the expiration.
356 a9613def Guido Trotter
    if up.salt in self._answers:
357 a9613def Guido Trotter
      del self._answers[up.salt]
358 392ca296 Guido Trotter
359 392ca296 Guido Trotter
  def _HandleReply(self, up):
360 392ca296 Guido Trotter
    """Handle a single confd reply, and decide whether to filter it.
361 392ca296 Guido Trotter

362 392ca296 Guido Trotter
    @rtype: boolean
363 392ca296 Guido Trotter
    @return: True if the reply should be filtered, False if it should be passed
364 392ca296 Guido Trotter
             on to the up-callback
365 392ca296 Guido Trotter

366 392ca296 Guido Trotter
    """
367 392ca296 Guido Trotter
    filter_upcall = False
368 392ca296 Guido Trotter
    salt = up.salt
369 392ca296 Guido Trotter
    if salt not in self._answers:
370 392ca296 Guido Trotter
      # first answer for a query (don't filter, and record)
371 392ca296 Guido Trotter
      self._answers[salt] = up.server_reply
372 392ca296 Guido Trotter
    elif up.server_reply.serial > self._answers[salt].serial:
373 392ca296 Guido Trotter
      # newer answer (record, and compare contents)
374 392ca296 Guido Trotter
      old_answer = self._answers[salt]
375 392ca296 Guido Trotter
      self._answers[salt] = up.server_reply
376 392ca296 Guido Trotter
      if up.server_reply.answer == old_answer.answer:
377 392ca296 Guido Trotter
        # same content (filter) (version upgrade was unrelated)
378 392ca296 Guido Trotter
        filter_upcall = True
379 392ca296 Guido Trotter
        self._LogFilter(salt, up.server_reply, old_answer)
380 392ca296 Guido Trotter
      # else: different content, pass up a second answer
381 392ca296 Guido Trotter
    else:
382 392ca296 Guido Trotter
      # older or same-version answer (duplicate or outdated, filter)
383 392ca296 Guido Trotter
      filter_upcall = True
384 392ca296 Guido Trotter
      self._LogFilter(salt, up.server_reply, self._answers[salt])
385 392ca296 Guido Trotter
386 392ca296 Guido Trotter
    return filter_upcall
387 392ca296 Guido Trotter
388 392ca296 Guido Trotter
  def __call__(self, up):
389 392ca296 Guido Trotter
    """Filtering callback
390 392ca296 Guido Trotter

391 392ca296 Guido Trotter
    @type up: L{ConfdUpcallPayload}
392 392ca296 Guido Trotter
    @param up: upper callback
393 392ca296 Guido Trotter

394 392ca296 Guido Trotter
    """
395 392ca296 Guido Trotter
    filter_upcall = False
396 392ca296 Guido Trotter
    if up.type == UPCALL_REPLY:
397 392ca296 Guido Trotter
      filter_upcall = self._HandleReply(up)
398 392ca296 Guido Trotter
    elif up.type == UPCALL_EXPIRE:
399 392ca296 Guido Trotter
      self._HandleExpire(up)
400 392ca296 Guido Trotter
401 392ca296 Guido Trotter
    if not filter_upcall:
402 392ca296 Guido Trotter
      self._callback(up)