Statistics
| Branch: | Tag: | Revision:

root / daemons / ganeti-masterd @ 26d3fd2f

History | View | Annotate | Download (18 kB)

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

    
4
# Copyright (C) 2006, 2007, 2010 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
# pylint: disable-msg=C0103
30
# C0103: Invalid name ganeti-masterd
31

    
32
import grp
33
import os
34
import pwd
35
import sys
36
import socket
37
import time
38
import tempfile
39
import logging
40

    
41
from optparse import OptionParser
42

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

    
59

    
60
CLIENT_REQUEST_WORKERS = 16
61

    
62
EXIT_NOTMASTER = constants.EXIT_NOTMASTER
63
EXIT_NODESETUP_ERROR = constants.EXIT_NODESETUP_ERROR
64

    
65

    
66
class ClientRequestWorker(workerpool.BaseWorker):
67
  # pylint: disable-msg=W0221
68
  def RunTask(self, server, message, client):
69
    """Process the request.
70

    
71
    """
72
    client_ops = ClientOps(server)
73

    
74
    try:
75
      (method, args) = luxi.ParseRequest(message)
76
    except luxi.ProtocolError, err:
77
      logging.error("Protocol Error: %s", err)
78
      client.close_log()
79
      return
80

    
81
    success = False
82
    try:
83
      result = client_ops.handle_request(method, args)
84
      success = True
85
    except errors.GenericError, err:
86
      logging.exception("Unexpected exception")
87
      success = False
88
      result = errors.EncodeException(err)
89
    except:
90
      logging.exception("Unexpected exception")
91
      err = sys.exc_info()
92
      result = "Caught exception: %s" % str(err[1])
93

    
94
    try:
95
      reply = luxi.FormatResponse(success, result)
96
      client.send_message(reply)
97
      # awake the main thread so that it can write out the data.
98
      server.awaker.signal()
99
    except: # pylint: disable-msg=W0702
100
      logging.exception("Send error")
101
      client.close_log()
102

    
103

    
104
class MasterClientHandler(daemon.AsyncTerminatedMessageStream):
105
  """Handler for master peers.
106

    
107
  """
108
  _MAX_UNHANDLED = 1
109
  def __init__(self, server, connected_socket, client_address, family):
110
    daemon.AsyncTerminatedMessageStream.__init__(self, connected_socket,
111
                                                 client_address,
112
                                                 constants.LUXI_EOM,
113
                                                 family, self._MAX_UNHANDLED)
114
    self.server = server
115

    
116
  def handle_message(self, message, _):
117
    self.server.request_workers.AddTask((self.server, message, self))
118

    
119

    
120
class MasterServer(daemon.AsyncStreamServer):
121
  """Master Server.
122

    
123
  This is the main asynchronous master server. It handles connections to the
124
  master socket.
125

    
126
  """
127
  family = socket.AF_UNIX
128

    
129
  def __init__(self, mainloop, address, uid, gid):
130
    """MasterServer constructor
131

    
132
    @type mainloop: ganeti.daemon.Mainloop
133
    @param mainloop: Mainloop used to poll for I/O events
134
    @param address: the unix socket address to bind the MasterServer to
135
    @param uid: The uid of the owner of the socket
136
    @param gid: The gid of the owner of the socket
137

    
138
    """
139
    temp_name = tempfile.mktemp(dir=os.path.dirname(address))
140
    daemon.AsyncStreamServer.__init__(self, self.family, temp_name)
141
    os.chmod(temp_name, 0770)
142
    os.chown(temp_name, uid, gid)
143
    os.rename(temp_name, address)
144

    
145
    self.mainloop = mainloop
146
    self.awaker = daemon.AsyncAwaker()
147

    
148
    # We'll only start threads once we've forked.
149
    self.context = None
150
    self.request_workers = None
151

    
152
  def handle_connection(self, connected_socket, client_address):
153
    # TODO: add connection count and limit the number of open connections to a
154
    # maximum number to avoid breaking for lack of file descriptors or memory.
155
    MasterClientHandler(self, connected_socket, client_address, self.family)
156

    
157
  def setup_queue(self):
158
    self.context = GanetiContext()
159
    self.request_workers = workerpool.WorkerPool("ClientReq",
160
                                                 CLIENT_REQUEST_WORKERS,
161
                                                 ClientRequestWorker)
162

    
163
  def server_cleanup(self):
164
    """Cleanup the server.
165

    
166
    This involves shutting down the processor threads and the master
167
    socket.
168

    
169
    """
170
    try:
171
      self.close()
172
    finally:
173
      if self.request_workers:
174
        self.request_workers.TerminateWorkers()
175
      if self.context:
176
        self.context.jobqueue.Shutdown()
177

    
178

    
179
class ClientOps:
180
  """Class holding high-level client operations."""
181
  def __init__(self, server):
182
    self.server = server
183

    
184
  def handle_request(self, method, args): # pylint: disable-msg=R0911
185
    queue = self.server.context.jobqueue
186

    
187
    # TODO: Parameter validation
188

    
189
    # TODO: Rewrite to not exit in each 'if/elif' branch
190

    
191
    if method == luxi.REQ_SUBMIT_JOB:
192
      logging.info("Received new job")
193
      ops = [opcodes.OpCode.LoadOpCode(state) for state in args]
194
      return queue.SubmitJob(ops)
195

    
196
    if method == luxi.REQ_SUBMIT_MANY_JOBS:
197
      logging.info("Received multiple jobs")
198
      jobs = []
199
      for ops in args:
200
        jobs.append([opcodes.OpCode.LoadOpCode(state) for state in ops])
201
      return queue.SubmitManyJobs(jobs)
202

    
203
    elif method == luxi.REQ_CANCEL_JOB:
204
      job_id = args
205
      logging.info("Received job cancel request for %s", job_id)
206
      return queue.CancelJob(job_id)
207

    
208
    elif method == luxi.REQ_ARCHIVE_JOB:
209
      job_id = args
210
      logging.info("Received job archive request for %s", job_id)
211
      return queue.ArchiveJob(job_id)
212

    
213
    elif method == luxi.REQ_AUTOARCHIVE_JOBS:
214
      (age, timeout) = args
215
      logging.info("Received job autoarchive request for age %s, timeout %s",
216
                   age, timeout)
217
      return queue.AutoArchiveJobs(age, timeout)
218

    
219
    elif method == luxi.REQ_WAIT_FOR_JOB_CHANGE:
220
      (job_id, fields, prev_job_info, prev_log_serial, timeout) = args
221
      logging.info("Received job poll request for %s", job_id)
222
      return queue.WaitForJobChanges(job_id, fields, prev_job_info,
223
                                     prev_log_serial, timeout)
224

    
225
    elif method == luxi.REQ_QUERY_JOBS:
226
      (job_ids, fields) = args
227
      if isinstance(job_ids, (tuple, list)) and job_ids:
228
        msg = utils.CommaJoin(job_ids)
229
      else:
230
        msg = str(job_ids)
231
      logging.info("Received job query request for %s", msg)
232
      return queue.QueryJobs(job_ids, fields)
233

    
234
    elif method == luxi.REQ_QUERY_INSTANCES:
235
      (names, fields, use_locking) = args
236
      logging.info("Received instance query request for %s", names)
237
      if use_locking:
238
        raise errors.OpPrereqError("Sync queries are not allowed",
239
                                   errors.ECODE_INVAL)
240
      op = opcodes.OpQueryInstances(names=names, output_fields=fields,
241
                                    use_locking=use_locking)
242
      return self._Query(op)
243

    
244
    elif method == luxi.REQ_QUERY_NODES:
245
      (names, fields, use_locking) = args
246
      logging.info("Received node query request for %s", names)
247
      if use_locking:
248
        raise errors.OpPrereqError("Sync queries are not allowed",
249
                                   errors.ECODE_INVAL)
250
      op = opcodes.OpQueryNodes(names=names, output_fields=fields,
251
                                use_locking=use_locking)
252
      return self._Query(op)
253

    
254
    elif method == luxi.REQ_QUERY_EXPORTS:
255
      nodes, use_locking = args
256
      if use_locking:
257
        raise errors.OpPrereqError("Sync queries are not allowed",
258
                                   errors.ECODE_INVAL)
259
      logging.info("Received exports query request")
260
      op = opcodes.OpQueryExports(nodes=nodes, use_locking=use_locking)
261
      return self._Query(op)
262

    
263
    elif method == luxi.REQ_QUERY_CONFIG_VALUES:
264
      fields = args
265
      logging.info("Received config values query request for %s", fields)
266
      op = opcodes.OpQueryConfigValues(output_fields=fields)
267
      return self._Query(op)
268

    
269
    elif method == luxi.REQ_QUERY_CLUSTER_INFO:
270
      logging.info("Received cluster info query request")
271
      op = opcodes.OpQueryClusterInfo()
272
      return self._Query(op)
273

    
274
    elif method == luxi.REQ_QUERY_TAGS:
275
      kind, name = args
276
      logging.info("Received tags query request")
277
      op = opcodes.OpGetTags(kind=kind, name=name)
278
      return self._Query(op)
279

    
280
    elif method == luxi.REQ_QUERY_LOCKS:
281
      (fields, sync) = args
282
      logging.info("Received locks query request")
283
      return self.server.context.glm.QueryLocks(fields, sync)
284

    
285
    elif method == luxi.REQ_QUEUE_SET_DRAIN_FLAG:
286
      drain_flag = args
287
      logging.info("Received queue drain flag change request to %s",
288
                   drain_flag)
289
      return queue.SetDrainFlag(drain_flag)
290

    
291
    elif method == luxi.REQ_SET_WATCHER_PAUSE:
292
      (until, ) = args
293

    
294
      if until is None:
295
        logging.info("Received request to no longer pause the watcher")
296
      else:
297
        if not isinstance(until, (int, float)):
298
          raise TypeError("Duration must be an integer or float")
299

    
300
        if until < time.time():
301
          raise errors.GenericError("Unable to set pause end time in the past")
302

    
303
        logging.info("Received request to pause the watcher until %s", until)
304

    
305
      return _SetWatcherPause(until)
306

    
307
    else:
308
      logging.info("Received invalid request '%s'", method)
309
      raise ValueError("Invalid operation '%s'" % method)
310

    
311
  def _Query(self, op):
312
    """Runs the specified opcode and returns the result.
313

    
314
    """
315
    # Queries don't have a job id
316
    proc = mcpu.Processor(self.server.context, None)
317

    
318
    # TODO: Executing an opcode using locks will acquire them in blocking mode.
319
    # Consider using a timeout for retries.
320
    return proc.ExecOpCode(op, None)
321

    
322

    
323
class GanetiContext(object):
324
  """Context common to all ganeti threads.
325

    
326
  This class creates and holds common objects shared by all threads.
327

    
328
  """
329
  # pylint: disable-msg=W0212
330
  # we do want to ensure a singleton here
331
  _instance = None
332

    
333
  def __init__(self):
334
    """Constructs a new GanetiContext object.
335

    
336
    There should be only a GanetiContext object at any time, so this
337
    function raises an error if this is not the case.
338

    
339
    """
340
    assert self.__class__._instance is None, "double GanetiContext instance"
341

    
342
    # Create global configuration object
343
    self.cfg = config.ConfigWriter()
344

    
345
    # Locking manager
346
    self.glm = locking.GanetiLockManager(
347
                self.cfg.GetNodeList(),
348
                self.cfg.GetInstanceList())
349

    
350
    # Job queue
351
    self.jobqueue = jqueue.JobQueue(self)
352

    
353
    # setting this also locks the class against attribute modifications
354
    self.__class__._instance = self
355

    
356
  def __setattr__(self, name, value):
357
    """Setting GanetiContext attributes is forbidden after initialization.
358

    
359
    """
360
    assert self.__class__._instance is None, "Attempt to modify Ganeti Context"
361
    object.__setattr__(self, name, value)
362

    
363
  def AddNode(self, node, ec_id):
364
    """Adds a node to the configuration and lock manager.
365

    
366
    """
367
    # Add it to the configuration
368
    self.cfg.AddNode(node, ec_id)
369

    
370
    # If preseeding fails it'll not be added
371
    self.jobqueue.AddNode(node)
372

    
373
    # Add the new node to the Ganeti Lock Manager
374
    self.glm.add(locking.LEVEL_NODE, node.name)
375

    
376
  def ReaddNode(self, node):
377
    """Updates a node that's already in the configuration
378

    
379
    """
380
    # Synchronize the queue again
381
    self.jobqueue.AddNode(node)
382

    
383
  def RemoveNode(self, name):
384
    """Removes a node from the configuration and lock manager.
385

    
386
    """
387
    # Remove node from configuration
388
    self.cfg.RemoveNode(name)
389

    
390
    # Notify job queue
391
    self.jobqueue.RemoveNode(name)
392

    
393
    # Remove the node from the Ganeti Lock Manager
394
    self.glm.remove(locking.LEVEL_NODE, name)
395

    
396

    
397
def _SetWatcherPause(until):
398
  """Creates or removes the watcher pause file.
399

    
400
  @type until: None or int
401
  @param until: Unix timestamp saying until when the watcher shouldn't run
402

    
403
  """
404
  if until is None:
405
    utils.RemoveFile(constants.WATCHER_PAUSEFILE)
406
  else:
407
    utils.WriteFile(constants.WATCHER_PAUSEFILE,
408
                    data="%d\n" % (until, ))
409

    
410
  return until
411

    
412

    
413
@rpc.RunWithRPC
414
def CheckAgreement():
415
  """Check the agreement on who is the master.
416

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

    
423
  In order to account for cold-start of all nodes, we retry for up to
424
  a minute until we get a real answer as the top-voted one. If the
425
  nodes are more out-of-sync, for now manual startup of the master
426
  should be attempted.
427

    
428
  Note that for a even number of nodes cluster, we need at least half
429
  of the nodes (beside ourselves) to vote for us. This creates a
430
  problem on two-node clusters, since in this case we require the
431
  other node to be up too to confirm our status.
432

    
433
  """
434
  myself = netutils.Hostname.GetSysName()
435
  #temp instantiation of a config writer, used only to get the node list
436
  cfg = config.ConfigWriter()
437
  node_list = cfg.GetNodeList()
438
  del cfg
439
  retries = 6
440
  while retries > 0:
441
    votes = bootstrap.GatherMasterVotes(node_list)
442
    if not votes:
443
      # empty node list, this is a one node cluster
444
      return True
445
    if votes[0][0] is None:
446
      retries -= 1
447
      time.sleep(10)
448
      continue
449
    break
450
  if retries == 0:
451
    logging.critical("Cluster inconsistent, most of the nodes didn't answer"
452
                     " after multiple retries. Aborting startup")
453
    logging.critical("Use the --no-voting option if you understand what"
454
                     " effects it has on the cluster state")
455
    return False
456
  # here a real node is at the top of the list
457
  all_votes = sum(item[1] for item in votes)
458
  top_node, top_votes = votes[0]
459

    
460
  result = False
461
  if top_node != myself:
462
    logging.critical("It seems we are not the master (top-voted node"
463
                     " is %s with %d out of %d votes)", top_node, top_votes,
464
                     all_votes)
465
  elif top_votes < all_votes - top_votes:
466
    logging.critical("It seems we are not the master (%d votes for,"
467
                     " %d votes against)", top_votes, all_votes - top_votes)
468
  else:
469
    result = True
470

    
471
  return result
472

    
473

    
474
@rpc.RunWithRPC
475
def ActivateMasterIP():
476
  # activate ip
477
  master_node = ssconf.SimpleStore().GetMasterNode()
478
  result = rpc.RpcRunner.call_node_start_master(master_node, False, False)
479
  msg = result.fail_msg
480
  if msg:
481
    logging.error("Can't activate master IP address: %s", msg)
482

    
483

    
484
def CheckMasterd(options, args):
485
  """Initial checks whether to run or exit with a failure.
486

    
487
  """
488
  if args: # masterd doesn't take any arguments
489
    print >> sys.stderr, ("Usage: %s [-f] [-d]" % sys.argv[0])
490
    sys.exit(constants.EXIT_FAILURE)
491

    
492
  ssconf.CheckMaster(options.debug)
493

    
494
  try:
495
    options.uid = pwd.getpwnam(constants.MASTERD_USER).pw_uid
496
    options.gid = grp.getgrnam(constants.DAEMONS_GROUP).gr_gid
497
  except KeyError:
498
    print >> sys.stderr, ("User or group not existing on system: %s:%s" %
499
                          (constants.MASTERD_USER, constants.DAEMONS_GROUP))
500
    sys.exit(constants.EXIT_FAILURE)
501

    
502

    
503
  # If CheckMaster didn't fail we believe we are the master, but we have to
504
  # confirm with the other nodes.
505
  if options.no_voting:
506
    if options.yes_do_it:
507
      return
508

    
509
    sys.stdout.write("The 'no voting' option has been selected.\n")
510
    sys.stdout.write("This is dangerous, please confirm by"
511
                     " typing uppercase 'yes': ")
512
    sys.stdout.flush()
513

    
514
    confirmation = sys.stdin.readline().strip()
515
    if confirmation != "YES":
516
      print >> sys.stderr, "Aborting."
517
      sys.exit(constants.EXIT_FAILURE)
518

    
519
    return
520

    
521
  # CheckAgreement uses RPC and threads, hence it needs to be run in a separate
522
  # process before we call utils.Daemonize in the current process.
523
  if not utils.RunInSeparateProcess(CheckAgreement):
524
    sys.exit(constants.EXIT_FAILURE)
525

    
526
  # ActivateMasterIP also uses RPC/threads, so we run it again via a
527
  # separate process.
528

    
529
  # TODO: decide whether failure to activate the master IP is a fatal error
530
  utils.RunInSeparateProcess(ActivateMasterIP)
531

    
532

    
533
def ExecMasterd(options, args): # pylint: disable-msg=W0613
534
  """Main master daemon function, executed with the PID file held.
535

    
536
  """
537
  # This is safe to do as the pid file guarantees against
538
  # concurrent execution.
539
  utils.RemoveFile(constants.MASTER_SOCKET)
540

    
541
  mainloop = daemon.Mainloop()
542
  master = MasterServer(mainloop, constants.MASTER_SOCKET,
543
                        options.uid, options.gid)
544
  try:
545
    rpc.Init()
546
    try:
547
      master.setup_queue()
548
      try:
549
        mainloop.Run()
550
      finally:
551
        master.server_cleanup()
552
    finally:
553
      rpc.Shutdown()
554
  finally:
555
    utils.RemoveFile(constants.MASTER_SOCKET)
556

    
557

    
558
def main():
559
  """Main function"""
560
  parser = OptionParser(description="Ganeti master daemon",
561
                        usage="%prog [-f] [-d]",
562
                        version="%%prog (ganeti) %s" %
563
                        constants.RELEASE_VERSION)
564
  parser.add_option("--no-voting", dest="no_voting",
565
                    help="Do not check that the nodes agree on this node"
566
                    " being the master and start the daemon unconditionally",
567
                    default=False, action="store_true")
568
  parser.add_option("--yes-do-it", dest="yes_do_it",
569
                    help="Override interactive check for --no-voting",
570
                    default=False, action="store_true")
571
  daemon.GenericMain(constants.MASTERD, parser, CheckMasterd, ExecMasterd,
572
                     multithreaded=True)
573

    
574

    
575
if __name__ == "__main__":
576
  main()