Statistics
| Branch: | Tag: | Revision:

root / daemons / ganeti-masterd @ 6797ec29

History | View | Annotate | Download (12.8 kB)

1
#!/usr/bin/python -u
2
#
3

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

    
21

    
22
"""Master daemon program.
23

    
24
Some classes deviates from the standard style guide since the
25
inheritance from parent classes requires it.
26

    
27
"""
28

    
29

    
30
import sys
31
import SocketServer
32
import time
33
import collections
34
import Queue
35
import random
36
import signal
37
import simplejson
38
import logging
39

    
40
from cStringIO import StringIO
41
from optparse import OptionParser
42

    
43
from ganeti import config
44
from ganeti import constants
45
from ganeti import mcpu
46
from ganeti import opcodes
47
from ganeti import jqueue
48
from ganeti import locking
49
from ganeti import luxi
50
from ganeti import utils
51
from ganeti import errors
52
from ganeti import ssconf
53
from ganeti import logger
54
from ganeti import workerpool
55
from ganeti import rpc
56

    
57

    
58
CLIENT_REQUEST_WORKERS = 16
59

    
60
EXIT_NOTMASTER = constants.EXIT_NOTMASTER
61
EXIT_NODESETUP_ERROR = constants.EXIT_NODESETUP_ERROR
62

    
63

    
64
class ClientRequestWorker(workerpool.BaseWorker):
65
  def RunTask(self, server, request, client_address):
66
    """Process the request.
67

    
68
    This is copied from the code in ThreadingMixIn.
69

    
70
    """
71
    try:
72
      server.finish_request(request, client_address)
73
      server.close_request(request)
74
    except:
75
      server.handle_error(request, client_address)
76
      server.close_request(request)
77

    
78

    
79
class IOServer(SocketServer.UnixStreamServer):
80
  """IO thread class.
81

    
82
  This class takes care of initializing the other threads, setting
83
  signal handlers (which are processed only in this thread), and doing
84
  cleanup at shutdown.
85

    
86
  """
87
  def __init__(self, address, rqhandler):
88
    """IOServer constructor
89

    
90
    Args:
91
      address: the address to bind this IOServer to
92
      rqhandler: RequestHandler type object
93

    
94
    """
95
    SocketServer.UnixStreamServer.__init__(self, address, rqhandler)
96

    
97
    # We'll only start threads once we've forked.
98
    self.context = None
99
    self.request_workers = None
100

    
101
  def setup_queue(self):
102
    self.context = GanetiContext()
103
    self.request_workers = workerpool.WorkerPool(CLIENT_REQUEST_WORKERS,
104
                                                 ClientRequestWorker)
105

    
106
  def process_request(self, request, client_address):
107
    """Add task to workerpool to process request.
108

    
109
    """
110
    self.request_workers.AddTask(self, request, client_address)
111

    
112
  def serve_forever(self):
113
    """Handle one request at a time until told to quit."""
114
    sighandler = utils.SignalHandler([signal.SIGINT, signal.SIGTERM])
115
    try:
116
      while not sighandler.called:
117
        self.handle_request()
118
    finally:
119
      sighandler.Reset()
120

    
121
  def server_cleanup(self):
122
    """Cleanup the server.
123

    
124
    This involves shutting down the processor threads and the master
125
    socket.
126

    
127
    """
128
    try:
129
      self.server_close()
130
    finally:
131
      if self.request_workers:
132
        self.request_workers.TerminateWorkers()
133
      if self.context:
134
        self.context.jobqueue.Shutdown()
135

    
136

    
137
class ClientRqHandler(SocketServer.BaseRequestHandler):
138
  """Client handler"""
139
  EOM = '\3'
140
  READ_SIZE = 4096
141

    
142
  def setup(self):
143
    self._buffer = ""
144
    self._msgs = collections.deque()
145
    self._ops = ClientOps(self.server)
146

    
147
  def handle(self):
148
    while True:
149
      msg = self.read_message()
150
      if msg is None:
151
        logging.info("client closed connection")
152
        break
153

    
154
      request = simplejson.loads(msg)
155
      logging.debug("request: %s", request)
156
      if not isinstance(request, dict):
157
        logging.error("wrong request received: %s", msg)
158
        break
159

    
160
      method = request.get(luxi.KEY_METHOD, None)
161
      args = request.get(luxi.KEY_ARGS, None)
162
      if method is None or args is None:
163
        logging.error("no method or args in request")
164
        break
165

    
166
      success = False
167
      try:
168
        result = self._ops.handle_request(method, args)
169
        success = True
170
      except errors.GenericError, err:
171
        success = False
172
        result = (err.__class__.__name__, err.args)
173
      except:
174
        logging.error("Unexpected exception", exc_info=True)
175
        err = sys.exc_info()
176
        result = "Caught exception: %s" % str(err[1])
177

    
178
      response = {
179
        luxi.KEY_SUCCESS: success,
180
        luxi.KEY_RESULT: result,
181
        }
182
      logging.debug("response: %s", response)
183
      self.send_message(simplejson.dumps(response))
184

    
185
  def read_message(self):
186
    while not self._msgs:
187
      data = self.request.recv(self.READ_SIZE)
188
      if not data:
189
        return None
190
      new_msgs = (self._buffer + data).split(self.EOM)
191
      self._buffer = new_msgs.pop()
192
      self._msgs.extend(new_msgs)
193
    return self._msgs.popleft()
194

    
195
  def send_message(self, msg):
196
    #print "sending", msg
197
    self.request.sendall(msg + self.EOM)
198

    
199

    
200
class ClientOps:
201
  """Class holding high-level client operations."""
202
  def __init__(self, server):
203
    self.server = server
204

    
205
  def handle_request(self, method, args):
206
    queue = self.server.context.jobqueue
207

    
208
    # TODO: Parameter validation
209

    
210
    if method == luxi.REQ_SUBMIT_JOB:
211
      ops = [opcodes.OpCode.LoadOpCode(state) for state in args]
212
      return queue.SubmitJob(ops)
213

    
214
    elif method == luxi.REQ_CANCEL_JOB:
215
      job_id = args
216
      return queue.CancelJob(job_id)
217

    
218
    elif method == luxi.REQ_ARCHIVE_JOB:
219
      job_id = args
220
      return queue.ArchiveJob(job_id)
221

    
222
    elif method == luxi.REQ_AUTOARCHIVE_JOBS:
223
      age = args
224
      return queue.AutoArchiveJobs(age)
225

    
226
    elif method == luxi.REQ_WAIT_FOR_JOB_CHANGE:
227
      (job_id, fields, prev_job_info, prev_log_serial, timeout) = args
228
      return queue.WaitForJobChanges(job_id, fields, prev_job_info,
229
                                     prev_log_serial, timeout)
230

    
231
    elif method == luxi.REQ_QUERY_JOBS:
232
      (job_ids, fields) = args
233
      return queue.QueryJobs(job_ids, fields)
234

    
235
    elif method == luxi.REQ_QUERY_INSTANCES:
236
      (names, fields) = args
237
      op = opcodes.OpQueryInstances(names=names, output_fields=fields)
238
      return self._Query(op)
239

    
240
    elif method == luxi.REQ_QUERY_NODES:
241
      (names, fields) = args
242
      op = opcodes.OpQueryNodes(names=names, output_fields=fields)
243
      return self._Query(op)
244

    
245
    elif method == luxi.REQ_QUERY_EXPORTS:
246
      nodes = args
247
      op = opcodes.OpQueryExports(nodes=nodes)
248
      return self._Query(op)
249

    
250
    elif method == luxi.REQ_QUERY_CONFIG_VALUES:
251
      fields = args
252
      op = opcodes.OpQueryConfigValues(output_fields=fields)
253
      return self._Query(op)
254

    
255
    else:
256
      raise ValueError("Invalid operation")
257

    
258
  def _DummyLog(self, *args):
259
    pass
260

    
261
  def _Query(self, op):
262
    """Runs the specified opcode and returns the result.
263

    
264
    """
265
    proc = mcpu.Processor(self.server.context)
266
    # TODO: Where should log messages go?
267
    return proc.ExecOpCode(op, self._DummyLog, None)
268

    
269

    
270
class GanetiContext(object):
271
  """Context common to all ganeti threads.
272

    
273
  This class creates and holds common objects shared by all threads.
274

    
275
  """
276
  _instance = None
277

    
278
  def __init__(self):
279
    """Constructs a new GanetiContext object.
280

    
281
    There should be only a GanetiContext object at any time, so this
282
    function raises an error if this is not the case.
283

    
284
    """
285
    assert self.__class__._instance is None, "double GanetiContext instance"
286

    
287
    # Create global configuration object
288
    self.cfg = config.ConfigWriter()
289

    
290
    # Locking manager
291
    self.glm = locking.GanetiLockManager(
292
                self.cfg.GetNodeList(),
293
                self.cfg.GetInstanceList())
294

    
295
    # Job queue
296
    self.jobqueue = jqueue.JobQueue(self)
297

    
298
    # setting this also locks the class against attribute modifications
299
    self.__class__._instance = self
300

    
301
  def __setattr__(self, name, value):
302
    """Setting GanetiContext attributes is forbidden after initialization.
303

    
304
    """
305
    assert self.__class__._instance is None, "Attempt to modify Ganeti Context"
306
    object.__setattr__(self, name, value)
307

    
308
  def AddNode(self, node):
309
    """Adds a node to the configuration and lock manager.
310

    
311
    """
312
    # Add it to the configuration
313
    self.cfg.AddNode(node)
314

    
315
    # If preseeding fails it'll not be added
316
    self.jobqueue.AddNode(node.name)
317

    
318
    # Add the new node to the Ganeti Lock Manager
319
    self.glm.add(locking.LEVEL_NODE, node.name)
320

    
321
  def ReaddNode(self, node):
322
    """Updates a node that's already in the configuration
323

    
324
    """
325
    # Synchronize the queue again
326
    self.jobqueue.AddNode(node.name)
327

    
328
  def RemoveNode(self, name):
329
    """Removes a node from the configuration and lock manager.
330

    
331
    """
332
    # Remove node from configuration
333
    self.cfg.RemoveNode(name)
334

    
335
    # Notify job queue
336
    self.jobqueue.RemoveNode(name)
337

    
338
    # Remove the node from the Ganeti Lock Manager
339
    self.glm.remove(locking.LEVEL_NODE, name)
340

    
341

    
342
def ParseOptions():
343
  """Parse the command line options.
344

    
345
  Returns:
346
    (options, args) as from OptionParser.parse_args()
347

    
348
  """
349
  parser = OptionParser(description="Ganeti master daemon",
350
                        usage="%prog [-f] [-d]",
351
                        version="%%prog (ganeti) %s" %
352
                        constants.RELEASE_VERSION)
353

    
354
  parser.add_option("-f", "--foreground", dest="fork",
355
                    help="Don't detach from the current terminal",
356
                    default=True, action="store_false")
357
  parser.add_option("-d", "--debug", dest="debug",
358
                    help="Enable some debug messages",
359
                    default=False, action="store_true")
360
  options, args = parser.parse_args()
361
  return options, args
362

    
363

    
364
def CheckAgreement():
365
  """Check the agreement on who is the master.
366

    
367
  The function uses a very simple algorithm: we must get more positive
368
  than negative answers. Since in most of the cases we are the master,
369
  we'll use our own config file for getting the node list. In the
370
  future we could collect the current node list from our (possibly
371
  obsolete) known nodes.
372

    
373
  """
374
  myself = utils.HostInfo().name
375
  #temp instantiation of a config writer, used only to get the node list
376
  cfg = config.ConfigWriter()
377
  node_list = cfg.GetNodeList()
378
  del cfg
379
  try:
380
    node_list.remove(myself)
381
  except KeyError:
382
    pass
383
  if not node_list:
384
    # either single node cluster, or a misconfiguration, but I won't
385
    # break any other node, so I can proceed
386
    return True
387
  results = rpc.RpcRunner.call_master_info(node_list)
388
  if not isinstance(results, dict):
389
    # this should not happen (unless internal error in rpc)
390
    logging.critical("Can't complete rpc call, aborting master startup")
391
    return False
392
  positive = negative = 0
393
  other_masters = {}
394
  for node in results:
395
    if not isinstance(results[node], (tuple, list)) or len(results[node]) < 3:
396
      logging.warning("Can't contact node %s", node)
397
      continue
398
    master_node = results[node][2]
399
    if master_node == myself:
400
      positive += 1
401
    else:
402
      negative += 1
403
      if not master_node in other_masters:
404
        other_masters[master_node] = 0
405
      other_masters[master_node] += 1
406
  if positive <= negative:
407
    # bad!
408
    logging.critical("It seems we are not the master (%d votes for,"
409
                     " %d votes against)", positive, negative)
410
    if len(other_masters) > 1:
411
      logging.critical("The other nodes do not agree on a single master")
412
    elif other_masters:
413
      # TODO: resync my files from the master
414
      logging.critical("It seems the real master is %s",
415
                       other_masters.keys()[0])
416
    else:
417
      logging.critical("Can't contact any node for data, aborting startup")
418
    return False
419
  return True
420

    
421

    
422
def main():
423
  """Main function"""
424

    
425
  options, args = ParseOptions()
426
  utils.debug = options.debug
427
  utils.no_fork = True
428

    
429
  ssconf.CheckMaster(options.debug)
430

    
431
  # we believe we are the master, let's ask the other nodes...
432
  if not CheckAgreement():
433
    return
434

    
435
  master = IOServer(constants.MASTER_SOCKET, ClientRqHandler)
436

    
437
  # become a daemon
438
  if options.fork:
439
    utils.Daemonize(logfile=constants.LOG_MASTERDAEMON,
440
                    noclose_fds=[master.fileno()])
441

    
442
  utils.WritePidFile(constants.MASTERD_PID)
443

    
444
  logger.SetupLogging(constants.LOG_MASTERDAEMON, debug=options.debug,
445
                      stderr_logging=not options.fork)
446

    
447
  logging.info("ganeti master daemon startup")
448

    
449
  # activate ip
450
  master_node = ssconf.SimpleConfigReader().GetMasterNode()
451
  if not rpc.RpcRunner.call_node_start_master(master_node, False):
452
    logging.error("Can't activate master IP address")
453

    
454
  master.setup_queue()
455
  try:
456
    master.serve_forever()
457
  finally:
458
    master.server_cleanup()
459
    utils.RemovePidFile(constants.MASTERD_PID)
460

    
461

    
462
if __name__ == "__main__":
463
  main()