Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ 56aa9fd5

History | View | Annotate | Download (69.1 kB)

1
#
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
"""Functions used by the node daemon"""
23

    
24

    
25
import os
26
import os.path
27
import shutil
28
import time
29
import stat
30
import errno
31
import re
32
import subprocess
33
import random
34
import logging
35
import tempfile
36

    
37
from ganeti import errors
38
from ganeti import utils
39
from ganeti import ssh
40
from ganeti import hypervisor
41
from ganeti import constants
42
from ganeti import bdev
43
from ganeti import objects
44
from ganeti import ssconf
45

    
46

    
47
def _GetConfig():
48
  """Simple wrapper to return a SimpleStore.
49

50
  @rtype: L{ssconf.SimpleStore}
51
  @return: a SimpleStore instance
52

53
  """
54
  return ssconf.SimpleStore()
55

    
56

    
57
def _GetSshRunner(cluster_name):
58
  """Simple wrapper to return an SshRunner.
59

60
  @type cluster_name: str
61
  @param cluster_name: the cluster name, which is needed
62
      by the SshRunner constructor
63
  @rtype: L{ssh.SshRunner}
64
  @return: an SshRunner instance
65

66
  """
67
  return ssh.SshRunner(cluster_name)
68

    
69

    
70
def _CleanDirectory(path, exclude=[]):
71
  """Removes all regular files in a directory.
72

73
  @type path: str
74
  @param path: the directory to clean
75
  @type exclude: list
76
  @param exclude: list of files to be excluded, defaults
77
      to the empty list
78
  @rtype: None
79

80
  """
81
  if not os.path.isdir(path):
82
    return
83

    
84
  # Normalize excluded paths
85
  exclude = [os.path.normpath(i) for i in exclude]
86

    
87
  for rel_name in utils.ListVisibleFiles(path):
88
    full_name = os.path.normpath(os.path.join(path, rel_name))
89
    if full_name in exclude:
90
      continue
91
    if os.path.isfile(full_name) and not os.path.islink(full_name):
92
      utils.RemoveFile(full_name)
93

    
94

    
95
def JobQueuePurge():
96
  """Removes job queue files and archived jobs.
97

98
  @rtype: None
99

100
  """
101
  _CleanDirectory(constants.QUEUE_DIR, exclude=[constants.JOB_QUEUE_LOCK_FILE])
102
  _CleanDirectory(constants.JOB_QUEUE_ARCHIVE_DIR)
103

    
104

    
105
def GetMasterInfo():
106
  """Returns master information.
107

108
  This is an utility function to compute master information, either
109
  for consumption here or from the node daemon.
110

111
  @rtype: tuple
112
  @return: (master_netdev, master_ip, master_name) if we have a good
113
      configuration, otherwise (None, None, None)
114

115
  """
116
  try:
117
    cfg = _GetConfig()
118
    master_netdev = cfg.GetMasterNetdev()
119
    master_ip = cfg.GetMasterIP()
120
    master_node = cfg.GetMasterNode()
121
  except errors.ConfigurationError, err:
122
    logging.exception("Cluster configuration incomplete")
123
    return (None, None, None)
124
  return (master_netdev, master_ip, master_node)
125

    
126

    
127
def StartMaster(start_daemons):
128
  """Activate local node as master node.
129

130
  The function will always try activate the IP address of the master
131
  (unless someone else has it). It will also start the master daemons,
132
  based on the start_daemons parameter.
133

134
  @type start_daemons: boolean
135
  @param start_daemons: whther to also start the master
136
      daemons (ganeti-masterd and ganeti-rapi)
137
  @rtype: None
138

139
  """
140
  ok = True
141
  master_netdev, master_ip, _ = GetMasterInfo()
142
  if not master_netdev:
143
    return False
144

    
145
  if utils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
146
    if utils.OwnIpAddress(master_ip):
147
      # we already have the ip:
148
      logging.debug("Already started")
149
    else:
150
      logging.error("Someone else has the master ip, not activating")
151
      ok = False
152
  else:
153
    result = utils.RunCmd(["ip", "address", "add", "%s/32" % master_ip,
154
                           "dev", master_netdev, "label",
155
                           "%s:0" % master_netdev])
156
    if result.failed:
157
      logging.error("Can't activate master IP: %s", result.output)
158
      ok = False
159

    
160
    result = utils.RunCmd(["arping", "-q", "-U", "-c 3", "-I", master_netdev,
161
                           "-s", master_ip, master_ip])
162
    # we'll ignore the exit code of arping
163

    
164
  # and now start the master and rapi daemons
165
  if start_daemons:
166
    for daemon in 'ganeti-masterd', 'ganeti-rapi':
167
      result = utils.RunCmd([daemon])
168
      if result.failed:
169
        logging.error("Can't start daemon %s: %s", daemon, result.output)
170
        ok = False
171
  return ok
172

    
173

    
174
def StopMaster(stop_daemons):
175
  """Deactivate this node as master.
176

177
  The function will always try to deactivate the IP address of the
178
  master. It will also stop the master daemons depending on the
179
  stop_daemons parameter.
180

181
  @type stop_daemons: boolean
182
  @param stop_daemons: whether to also stop the master daemons
183
      (ganeti-masterd and ganeti-rapi)
184
  @rtype: None
185

186
  """
187
  master_netdev, master_ip, _ = GetMasterInfo()
188
  if not master_netdev:
189
    return False
190

    
191
  result = utils.RunCmd(["ip", "address", "del", "%s/32" % master_ip,
192
                         "dev", master_netdev])
193
  if result.failed:
194
    logging.error("Can't remove the master IP, error: %s", result.output)
195
    # but otherwise ignore the failure
196

    
197
  if stop_daemons:
198
    # stop/kill the rapi and the master daemon
199
    for daemon in constants.RAPI_PID, constants.MASTERD_PID:
200
      utils.KillProcess(utils.ReadPidFile(utils.DaemonPidFileName(daemon)))
201

    
202
  return True
203

    
204

    
205
def AddNode(dsa, dsapub, rsa, rsapub, sshkey, sshpub):
206
  """Joins this node to the cluster.
207

208
  This does the following:
209
      - updates the hostkeys of the machine (rsa and dsa)
210
      - adds the ssh private key to the user
211
      - adds the ssh public key to the users' authorized_keys file
212

213
  @type dsa: str
214
  @param dsa: the DSA private key to write
215
  @type dsapub: str
216
  @param dsapub: the DSA public key to write
217
  @type rsa: str
218
  @param rsa: the RSA private key to write
219
  @type rsapub: str
220
  @param rsapub: the RSA public key to write
221
  @type sshkey: str
222
  @param sshkey: the SSH private key to write
223
  @type sshpub: str
224
  @param sshpub: the SSH public key to write
225
  @rtype: boolean
226
  @return: the success of the operation
227

228
  """
229
  sshd_keys =  [(constants.SSH_HOST_RSA_PRIV, rsa, 0600),
230
                (constants.SSH_HOST_RSA_PUB, rsapub, 0644),
231
                (constants.SSH_HOST_DSA_PRIV, dsa, 0600),
232
                (constants.SSH_HOST_DSA_PUB, dsapub, 0644)]
233
  for name, content, mode in sshd_keys:
234
    utils.WriteFile(name, data=content, mode=mode)
235

    
236
  try:
237
    priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS,
238
                                                    mkdir=True)
239
  except errors.OpExecError, err:
240
    logging.exception("Error while processing user ssh files")
241
    return False
242

    
243
  for name, content in [(priv_key, sshkey), (pub_key, sshpub)]:
244
    utils.WriteFile(name, data=content, mode=0600)
245

    
246
  utils.AddAuthorizedKey(auth_keys, sshpub)
247

    
248
  utils.RunCmd([constants.SSH_INITD_SCRIPT, "restart"])
249

    
250
  return True
251

    
252

    
253
def LeaveCluster():
254
  """Cleans up and remove the current node.
255

256
  This function cleans up and prepares the current node to be removed
257
  from the cluster.
258

259
  If processing is successful, then it raises an
260
  L{errors.GanetiQuitException} which is used as a special case to
261
  shutdown the node daemon.
262

263
  """
264
  _CleanDirectory(constants.DATA_DIR)
265
  JobQueuePurge()
266

    
267
  try:
268
    priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
269
  except errors.OpExecError:
270
    logging.exception("Error while processing ssh files")
271
    return
272

    
273
  f = open(pub_key, 'r')
274
  try:
275
    utils.RemoveAuthorizedKey(auth_keys, f.read(8192))
276
  finally:
277
    f.close()
278

    
279
  utils.RemoveFile(priv_key)
280
  utils.RemoveFile(pub_key)
281

    
282
  # Return a reassuring string to the caller, and quit
283
  raise errors.QuitGanetiException(False, 'Shutdown scheduled')
284

    
285

    
286
def GetNodeInfo(vgname, hypervisor_type):
287
  """Gives back a hash with different informations about the node.
288

289
  @type vgname: C{string}
290
  @param vgname: the name of the volume group to ask for disk space information
291
  @type hypervisor_type: C{str}
292
  @param hypervisor_type: the name of the hypervisor to ask for
293
      memory information
294
  @rtype: C{dict}
295
  @return: dictionary with the following keys:
296
      - vg_size is the size of the configured volume group in MiB
297
      - vg_free is the free size of the volume group in MiB
298
      - memory_dom0 is the memory allocated for domain0 in MiB
299
      - memory_free is the currently available (free) ram in MiB
300
      - memory_total is the total number of ram in MiB
301

302
  """
303
  outputarray = {}
304
  vginfo = _GetVGInfo(vgname)
305
  outputarray['vg_size'] = vginfo['vg_size']
306
  outputarray['vg_free'] = vginfo['vg_free']
307

    
308
  hyper = hypervisor.GetHypervisor(hypervisor_type)
309
  hyp_info = hyper.GetNodeInfo()
310
  if hyp_info is not None:
311
    outputarray.update(hyp_info)
312

    
313
  f = open("/proc/sys/kernel/random/boot_id", 'r')
314
  try:
315
    outputarray["bootid"] = f.read(128).rstrip("\n")
316
  finally:
317
    f.close()
318

    
319
  return outputarray
320

    
321

    
322
def VerifyNode(what, cluster_name):
323
  """Verify the status of the local node.
324

325
  Based on the input L{what} parameter, various checks are done on the
326
  local node.
327

328
  If the I{filelist} key is present, this list of
329
  files is checksummed and the file/checksum pairs are returned.
330

331
  If the I{nodelist} key is present, we check that we have
332
  connectivity via ssh with the target nodes (and check the hostname
333
  report).
334

335
  If the I{node-net-test} key is present, we check that we have
336
  connectivity to the given nodes via both primary IP and, if
337
  applicable, secondary IPs.
338

339
  @type what: C{dict}
340
  @param what: a dictionary of things to check:
341
      - filelist: list of files for which to compute checksums
342
      - nodelist: list of nodes we should check ssh communication with
343
      - node-net-test: list of nodes we should check node daemon port
344
        connectivity with
345
      - hypervisor: list with hypervisors to run the verify for
346
  @rtype: dict
347
  @return: a dictionary with the same keys as the input dict, and
348
      values representing the result of the checks
349

350
  """
351
  result = {}
352

    
353
  if constants.NV_HYPERVISOR in what:
354
    result[constants.NV_HYPERVISOR] = tmp = {}
355
    for hv_name in what[constants.NV_HYPERVISOR]:
356
      tmp[hv_name] = hypervisor.GetHypervisor(hv_name).Verify()
357

    
358
  if constants.NV_FILELIST in what:
359
    result[constants.NV_FILELIST] = utils.FingerprintFiles(
360
      what[constants.NV_FILELIST])
361

    
362
  if constants.NV_NODELIST in what:
363
    result[constants.NV_NODELIST] = tmp = {}
364
    random.shuffle(what[constants.NV_NODELIST])
365
    for node in what[constants.NV_NODELIST]:
366
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
367
      if not success:
368
        tmp[node] = message
369

    
370
  if constants.NV_NODENETTEST in what:
371
    result[constants.NV_NODENETTEST] = tmp = {}
372
    my_name = utils.HostInfo().name
373
    my_pip = my_sip = None
374
    for name, pip, sip in what[constants.NV_NODENETTEST]:
375
      if name == my_name:
376
        my_pip = pip
377
        my_sip = sip
378
        break
379
    if not my_pip:
380
      tmp[my_name] = ("Can't find my own primary/secondary IP"
381
                      " in the node list")
382
    else:
383
      port = utils.GetNodeDaemonPort()
384
      for name, pip, sip in what[constants.NV_NODENETTEST]:
385
        fail = []
386
        if not utils.TcpPing(pip, port, source=my_pip):
387
          fail.append("primary")
388
        if sip != pip:
389
          if not utils.TcpPing(sip, port, source=my_sip):
390
            fail.append("secondary")
391
        if fail:
392
          tmp[name] = ("failure using the %s interface(s)" %
393
                       " and ".join(fail))
394

    
395
  if constants.NV_LVLIST in what:
396
    result[constants.NV_LVLIST] = GetVolumeList(what[constants.NV_LVLIST])
397

    
398
  if constants.NV_INSTANCELIST in what:
399
    result[constants.NV_INSTANCELIST] = GetInstanceList(
400
      what[constants.NV_INSTANCELIST])
401

    
402
  if constants.NV_VGLIST in what:
403
    result[constants.NV_VGLIST] = ListVolumeGroups()
404

    
405
  if constants.NV_VERSION in what:
406
    result[constants.NV_VERSION] = constants.PROTOCOL_VERSION
407

    
408
  if constants.NV_HVINFO in what:
409
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
410
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
411

    
412
  return result
413

    
414

    
415
def GetVolumeList(vg_name):
416
  """Compute list of logical volumes and their size.
417

418
  @type vg_name: str
419
  @param vg_name: the volume group whose LVs we should list
420
  @rtype: dict
421
  @return:
422
      dictionary of all partions (key) with value being a tuple of
423
      their size (in MiB), inactive and online status::
424

425
        {'test1': ('20.06', True, True)}
426

427
      in case of errors, a string is returned with the error
428
      details.
429

430
  """
431
  lvs = {}
432
  sep = '|'
433
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
434
                         "--separator=%s" % sep,
435
                         "-olv_name,lv_size,lv_attr", vg_name])
436
  if result.failed:
437
    logging.error("Failed to list logical volumes, lvs output: %s",
438
                  result.output)
439
    return result.output
440

    
441
  valid_line_re = re.compile("^ *([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
442
  for line in result.stdout.splitlines():
443
    line = line.strip()
444
    match = valid_line_re.match(line)
445
    if not match:
446
      logging.error("Invalid line returned from lvs output: '%s'", line)
447
      continue
448
    name, size, attr = match.groups()
449
    inactive = attr[4] == '-'
450
    online = attr[5] == 'o'
451
    lvs[name] = (size, inactive, online)
452

    
453
  return lvs
454

    
455

    
456
def ListVolumeGroups():
457
  """List the volume groups and their size.
458

459
  @rtype: dict
460
  @return: dictionary with keys volume name and values the
461
      size of the volume
462

463
  """
464
  return utils.ListVolumeGroups()
465

    
466

    
467
def NodeVolumes():
468
  """List all volumes on this node.
469

470
  @rtype: list
471
  @return:
472
    A list of dictionaries, each having four keys:
473
      - name: the logical volume name,
474
      - size: the size of the logical volume
475
      - dev: the physical device on which the LV lives
476
      - vg: the volume group to which it belongs
477

478
    In case of errors, we return an empty list and log the
479
    error.
480

481
    Note that since a logical volume can live on multiple physical
482
    volumes, the resulting list might include a logical volume
483
    multiple times.
484

485
  """
486
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
487
                         "--separator=|",
488
                         "--options=lv_name,lv_size,devices,vg_name"])
489
  if result.failed:
490
    logging.error("Failed to list logical volumes, lvs output: %s",
491
                  result.output)
492
    return []
493

    
494
  def parse_dev(dev):
495
    if '(' in dev:
496
      return dev.split('(')[0]
497
    else:
498
      return dev
499

    
500
  def map_line(line):
501
    return {
502
      'name': line[0].strip(),
503
      'size': line[1].strip(),
504
      'dev': parse_dev(line[2].strip()),
505
      'vg': line[3].strip(),
506
    }
507

    
508
  return [map_line(line.split('|')) for line in result.stdout.splitlines()
509
          if line.count('|') >= 3]
510

    
511

    
512
def BridgesExist(bridges_list):
513
  """Check if a list of bridges exist on the current node.
514

515
  @rtype: boolean
516
  @return: C{True} if all of them exist, C{False} otherwise
517

518
  """
519
  for bridge in bridges_list:
520
    if not utils.BridgeExists(bridge):
521
      return False
522

    
523
  return True
524

    
525

    
526
def GetInstanceList(hypervisor_list):
527
  """Provides a list of instances.
528

529
  @type hypervisor_list: list
530
  @param hypervisor_list: the list of hypervisors to query information
531

532
  @rtype: list
533
  @return: a list of all running instances on the current node
534
    - instance1.example.com
535
    - instance2.example.com
536

537
  """
538
  results = []
539
  for hname in hypervisor_list:
540
    try:
541
      names = hypervisor.GetHypervisor(hname).ListInstances()
542
      results.extend(names)
543
    except errors.HypervisorError, err:
544
      logging.exception("Error enumerating instances for hypevisor %s", hname)
545
      # FIXME: should we somehow not propagate this to the master?
546
      raise
547

    
548
  return results
549

    
550

    
551
def GetInstanceInfo(instance, hname):
552
  """Gives back the informations about an instance as a dictionary.
553

554
  @type instance: string
555
  @param instance: the instance name
556
  @type hname: string
557
  @param hname: the hypervisor type of the instance
558

559
  @rtype: dict
560
  @return: dictionary with the following keys:
561
      - memory: memory size of instance (int)
562
      - state: xen state of instance (string)
563
      - time: cpu time of instance (float)
564

565
  """
566
  output = {}
567

    
568
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
569
  if iinfo is not None:
570
    output['memory'] = iinfo[2]
571
    output['state'] = iinfo[4]
572
    output['time'] = iinfo[5]
573

    
574
  return output
575

    
576

    
577
def GetAllInstancesInfo(hypervisor_list):
578
  """Gather data about all instances.
579

580
  This is the equivalent of L{GetInstanceInfo}, except that it
581
  computes data for all instances at once, thus being faster if one
582
  needs data about more than one instance.
583

584
  @type hypervisor_list: list
585
  @param hypervisor_list: list of hypervisors to query for instance data
586

587
  @rtype: dict
588
  @return: dictionary of instance: data, with data having the following keys:
589
      - memory: memory size of instance (int)
590
      - state: xen state of instance (string)
591
      - time: cpu time of instance (float)
592
      - vcpus: the number of vcpus
593

594
  """
595
  output = {}
596

    
597
  for hname in hypervisor_list:
598
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
599
    if iinfo:
600
      for name, inst_id, memory, vcpus, state, times in iinfo:
601
        value = {
602
          'memory': memory,
603
          'vcpus': vcpus,
604
          'state': state,
605
          'time': times,
606
          }
607
        if name in output and output[name] != value:
608
          raise errors.HypervisorError("Instance %s running duplicate"
609
                                       " with different parameters" % name)
610
        output[name] = value
611

    
612
  return output
613

    
614

    
615
def AddOSToInstance(instance):
616
  """Add an OS to an instance.
617

618
  @type instance: L{objects.Instance}
619
  @param instance: Instance whose OS is to be installed
620
  @rtype: boolean
621
  @return: the success of the operation
622

623
  """
624
  inst_os = OSFromDisk(instance.os)
625

    
626
  create_env = OSEnvironment(instance)
627

    
628
  logfile = "%s/add-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
629
                                     instance.name, int(time.time()))
630

    
631
  result = utils.RunCmd([inst_os.create_script], env=create_env,
632
                        cwd=inst_os.path, output=logfile,)
633
  if result.failed:
634
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
635
                  " output: %s", result.cmd, result.fail_reason, logfile,
636
                  result.output)
637
    return False
638

    
639
  return True
640

    
641

    
642
def RunRenameInstance(instance, old_name):
643
  """Run the OS rename script for an instance.
644

645
  @type instance: L{objects.Instance}
646
  @param instance: Instance whose OS is to be installed
647
  @type old_name: string
648
  @param old_name: previous instance name
649
  @rtype: boolean
650
  @return: the success of the operation
651

652
  """
653
  inst_os = OSFromDisk(instance.os)
654

    
655
  rename_env = OSEnvironment(instance)
656
  rename_env['OLD_INSTANCE_NAME'] = old_name
657

    
658
  logfile = "%s/rename-%s-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
659
                                           old_name,
660
                                           instance.name, int(time.time()))
661

    
662
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
663
                        cwd=inst_os.path, output=logfile)
664

    
665
  if result.failed:
666
    logging.error("os create command '%s' returned error: %s output: %s",
667
                  result.cmd, result.fail_reason, result.output)
668
    return False
669

    
670
  return True
671

    
672

    
673
def _GetVGInfo(vg_name):
674
  """Get informations about the volume group.
675

676
  @type vg_name: str
677
  @param vg_name: the volume group which we query
678
  @rtype: dict
679
  @return:
680
    A dictionary with the following keys:
681
      - C{vg_size} is the total size of the volume group in MiB
682
      - C{vg_free} is the free size of the volume group in MiB
683
      - C{pv_count} are the number of physical disks in that VG
684

685
    If an error occurs during gathering of data, we return the same dict
686
    with keys all set to None.
687

688
  """
689
  retdic = dict.fromkeys(["vg_size", "vg_free", "pv_count"])
690

    
691
  retval = utils.RunCmd(["vgs", "-ovg_size,vg_free,pv_count", "--noheadings",
692
                         "--nosuffix", "--units=m", "--separator=:", vg_name])
693

    
694
  if retval.failed:
695
    logging.error("volume group %s not present", vg_name)
696
    return retdic
697
  valarr = retval.stdout.strip().rstrip(':').split(':')
698
  if len(valarr) == 3:
699
    try:
700
      retdic = {
701
        "vg_size": int(round(float(valarr[0]), 0)),
702
        "vg_free": int(round(float(valarr[1]), 0)),
703
        "pv_count": int(valarr[2]),
704
        }
705
    except ValueError, err:
706
      logging.exception("Fail to parse vgs output")
707
  else:
708
    logging.error("vgs output has the wrong number of fields (expected"
709
                  " three): %s", str(valarr))
710
  return retdic
711

    
712

    
713
def _GatherBlockDevs(instance):
714
  """Set up an instance's block device(s).
715

716
  This is run on the primary node at instance startup. The block
717
  devices must be already assembled.
718

719
  @type instance: L{objects.Instance}
720
  @param instance: the instance whose disks we shoul assemble
721
  @rtype: list of L{bdev.BlockDev}
722
  @return: list of the block devices
723

724
  """
725
  block_devices = []
726
  for disk in instance.disks:
727
    device = _RecursiveFindBD(disk)
728
    if device is None:
729
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
730
                                    str(disk))
731
    device.Open()
732
    block_devices.append((disk, device))
733
  return block_devices
734

    
735

    
736
def StartInstance(instance, extra_args):
737
  """Start an instance.
738

739
  @type instance: L{objects.Instance}
740
  @param instance: the instance object
741
  @rtype: boolean
742
  @return: whether the startup was successful or not
743

744
  """
745
  running_instances = GetInstanceList([instance.hypervisor])
746

    
747
  if instance.name in running_instances:
748
    return True
749

    
750
  block_devices = _GatherBlockDevs(instance)
751
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
752

    
753
  try:
754
    hyper.StartInstance(instance, block_devices, extra_args)
755
  except errors.HypervisorError, err:
756
    logging.exception("Failed to start instance")
757
    return False
758

    
759
  return True
760

    
761

    
762
def ShutdownInstance(instance):
763
  """Shut an instance down.
764

765
  @note: this functions uses polling with a hardcoded timeout.
766

767
  @type instance: L{objects.Instance}
768
  @param instance: the instance object
769
  @rtype: boolean
770
  @return: whether the startup was successful or not
771

772
  """
773
  hv_name = instance.hypervisor
774
  running_instances = GetInstanceList([hv_name])
775

    
776
  if instance.name not in running_instances:
777
    return True
778

    
779
  hyper = hypervisor.GetHypervisor(hv_name)
780
  try:
781
    hyper.StopInstance(instance)
782
  except errors.HypervisorError, err:
783
    logging.error("Failed to stop instance")
784
    return False
785

    
786
  # test every 10secs for 2min
787
  shutdown_ok = False
788

    
789
  time.sleep(1)
790
  for dummy in range(11):
791
    if instance.name not in GetInstanceList([hv_name]):
792
      break
793
    time.sleep(10)
794
  else:
795
    # the shutdown did not succeed
796
    logging.error("shutdown of '%s' unsuccessful, using destroy", instance)
797

    
798
    try:
799
      hyper.StopInstance(instance, force=True)
800
    except errors.HypervisorError, err:
801
      logging.exception("Failed to stop instance")
802
      return False
803

    
804
    time.sleep(1)
805
    if instance.name in GetInstanceList([hv_name]):
806
      logging.error("could not shutdown instance '%s' even by destroy",
807
                    instance.name)
808
      return False
809

    
810
  return True
811

    
812

    
813
def RebootInstance(instance, reboot_type, extra_args):
814
  """Reboot an instance.
815

816
  @type instance: L{objects.Instance}
817
  @param instance: the instance object to reboot
818
  @type reboot_type: str
819
  @param reboot_type: the type of reboot, one the following
820
    constants:
821
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
822
        instance OS, do not recreate the VM
823
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
824
        restart the VM (at the hypervisor level)
825
      - the other reboot type (L{constants.INSTANCE_REBOOT_HARD})
826
        is not accepted here, since that mode is handled
827
        differently
828
  @rtype: boolean
829
  @return: the success of the operation
830

831
  """
832
  running_instances = GetInstanceList([instance.hypervisor])
833

    
834
  if instance.name not in running_instances:
835
    logging.error("Cannot reboot instance that is not running")
836
    return False
837

    
838
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
839
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
840
    try:
841
      hyper.RebootInstance(instance)
842
    except errors.HypervisorError, err:
843
      logging.exception("Failed to soft reboot instance")
844
      return False
845
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
846
    try:
847
      ShutdownInstance(instance)
848
      StartInstance(instance, extra_args)
849
    except errors.HypervisorError, err:
850
      logging.exception("Failed to hard reboot instance")
851
      return False
852
  else:
853
    raise errors.ParameterError("reboot_type invalid")
854

    
855
  return True
856

    
857

    
858
def MigrateInstance(instance, target, live):
859
  """Migrates an instance to another node.
860

861
  @type instance: L{objects.Instance}
862
  @param instance: the instance definition
863
  @type target: string
864
  @param target: the target node name
865
  @type live: boolean
866
  @param live: whether the migration should be done live or not (the
867
      interpretation of this parameter is left to the hypervisor)
868
  @rtype: tuple
869
  @return: a tuple of (success, msg) where:
870
      - succes is a boolean denoting the success/failure of the operation
871
      - msg is a string with details in case of failure
872

873
  """
874
  hyper = hypervisor.GetHypervisor(instance.hypervisor_name)
875

    
876
  try:
877
    hyper.MigrateInstance(instance.name, target, live)
878
  except errors.HypervisorError, err:
879
    msg = "Failed to migrate instance: %s" % str(err)
880
    logging.error(msg)
881
    return (False, msg)
882
  return (True, "Migration successfull")
883

    
884

    
885
def CreateBlockDevice(disk, size, owner, on_primary, info):
886
  """Creates a block device for an instance.
887

888
  @type disk: L{objects.Disk}
889
  @param disk: the object describing the disk we should create
890
  @type size: int
891
  @param size: the size of the physical underlying device, in MiB
892
  @type owner: str
893
  @param owner: the name of the instance for which disk is created,
894
      used for device cache data
895
  @type on_primary: boolean
896
  @param on_primary:  indicates if it is the primary node or not
897
  @type info: string
898
  @param info: string that will be sent to the physical device
899
      creation, used for example to set (LVM) tags on LVs
900

901
  @return: the new unique_id of the device (this can sometime be
902
      computed only after creation), or None. On secondary nodes,
903
      it's not required to return anything.
904

905
  """
906
  clist = []
907
  if disk.children:
908
    for child in disk.children:
909
      crdev = _RecursiveAssembleBD(child, owner, on_primary)
910
      if on_primary or disk.AssembleOnSecondary():
911
        # we need the children open in case the device itself has to
912
        # be assembled
913
        crdev.Open()
914
      clist.append(crdev)
915
  try:
916
    device = bdev.FindDevice(disk.dev_type, disk.physical_id, clist)
917
    if device is not None:
918
      logging.info("removing existing device %s", disk)
919
      device.Remove()
920
  except errors.BlockDeviceError, err:
921
    pass
922

    
923
  device = bdev.Create(disk.dev_type, disk.physical_id,
924
                       clist, size)
925
  if device is None:
926
    raise ValueError("Can't create child device for %s, %s" %
927
                     (disk, size))
928
  if on_primary or disk.AssembleOnSecondary():
929
    if not device.Assemble():
930
      errorstring = "Can't assemble device after creation"
931
      logging.error(errorstring)
932
      raise errors.BlockDeviceError("%s, very unusual event - check the node"
933
                                    " daemon logs" % errorstring)
934
    device.SetSyncSpeed(constants.SYNC_SPEED)
935
    if on_primary or disk.OpenOnSecondary():
936
      device.Open(force=True)
937
    DevCacheManager.UpdateCache(device.dev_path, owner,
938
                                on_primary, disk.iv_name)
939

    
940
  device.SetInfo(info)
941

    
942
  physical_id = device.unique_id
943
  return physical_id
944

    
945

    
946
def RemoveBlockDevice(disk):
947
  """Remove a block device.
948

949
  @note: This is intended to be called recursively.
950

951
  @type disk: L{objects.disk}
952
  @param disk: the disk object we should remove
953
  @rtype: boolean
954
  @return: the success of the operation
955

956
  """
957
  try:
958
    # since we are removing the device, allow a partial match
959
    # this allows removal of broken mirrors
960
    rdev = _RecursiveFindBD(disk, allow_partial=True)
961
  except errors.BlockDeviceError, err:
962
    # probably can't attach
963
    logging.info("Can't attach to device %s in remove", disk)
964
    rdev = None
965
  if rdev is not None:
966
    r_path = rdev.dev_path
967
    result = rdev.Remove()
968
    if result:
969
      DevCacheManager.RemoveCache(r_path)
970
  else:
971
    result = True
972
  if disk.children:
973
    for child in disk.children:
974
      result = result and RemoveBlockDevice(child)
975
  return result
976

    
977

    
978
def _RecursiveAssembleBD(disk, owner, as_primary):
979
  """Activate a block device for an instance.
980

981
  This is run on the primary and secondary nodes for an instance.
982

983
  @note: this function is called recursively.
984

985
  @type disk: L{objects.Disk}
986
  @param disk: the disk we try to assemble
987
  @type owner: str
988
  @param owner: the name of the instance which owns the disk
989
  @type as_primary: boolean
990
  @param as_primary: if we should make the block device
991
      read/write
992

993
  @return: the assembled device or None (in case no device
994
      was assembled)
995
  @raise errors.BlockDeviceError: in case there is an error
996
      during the activation of the children or the device
997
      itself
998

999
  """
1000
  children = []
1001
  if disk.children:
1002
    mcn = disk.ChildrenNeeded()
1003
    if mcn == -1:
1004
      mcn = 0 # max number of Nones allowed
1005
    else:
1006
      mcn = len(disk.children) - mcn # max number of Nones
1007
    for chld_disk in disk.children:
1008
      try:
1009
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1010
      except errors.BlockDeviceError, err:
1011
        if children.count(None) >= mcn:
1012
          raise
1013
        cdev = None
1014
        logging.debug("Error in child activation: %s", str(err))
1015
      children.append(cdev)
1016

    
1017
  if as_primary or disk.AssembleOnSecondary():
1018
    r_dev = bdev.AttachOrAssemble(disk.dev_type, disk.physical_id, children)
1019
    r_dev.SetSyncSpeed(constants.SYNC_SPEED)
1020
    result = r_dev
1021
    if as_primary or disk.OpenOnSecondary():
1022
      r_dev.Open()
1023
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1024
                                as_primary, disk.iv_name)
1025

    
1026
  else:
1027
    result = True
1028
  return result
1029

    
1030

    
1031
def AssembleBlockDevice(disk, owner, as_primary):
1032
  """Activate a block device for an instance.
1033

1034
  This is a wrapper over _RecursiveAssembleBD.
1035

1036
  @rtype: str or boolean
1037
  @return: a C{/dev/...} path for primary nodes, and
1038
      C{True} for secondary nodes
1039

1040
  """
1041
  result = _RecursiveAssembleBD(disk, owner, as_primary)
1042
  if isinstance(result, bdev.BlockDev):
1043
    result = result.dev_path
1044
  return result
1045

    
1046

    
1047
def ShutdownBlockDevice(disk):
1048
  """Shut down a block device.
1049

1050
  First, if the device is assembled (can L{Attach()}), then the device
1051
  is shutdown. Then the children of the device are shutdown.
1052

1053
  This function is called recursively. Note that we don't cache the
1054
  children or such, as oppossed to assemble, shutdown of different
1055
  devices doesn't require that the upper device was active.
1056

1057
  @type disk: L{objects.Disk}
1058
  @param disk: the description of the disk we should
1059
      shutdown
1060
  @rtype: boolean
1061
  @return: the success of the operation
1062

1063
  """
1064
  r_dev = _RecursiveFindBD(disk)
1065
  if r_dev is not None:
1066
    r_path = r_dev.dev_path
1067
    result = r_dev.Shutdown()
1068
    if result:
1069
      DevCacheManager.RemoveCache(r_path)
1070
  else:
1071
    result = True
1072
  if disk.children:
1073
    for child in disk.children:
1074
      result = result and ShutdownBlockDevice(child)
1075
  return result
1076

    
1077

    
1078
def MirrorAddChildren(parent_cdev, new_cdevs):
1079
  """Extend a mirrored block device.
1080

1081
  @type parent_cdev: L{objects.Disk}
1082
  @param parent_cdev: the disk to which we should add children
1083
  @type new_cdevs: list of L{objects.Disk}
1084
  @param new_cdevs: the list of children which we should add
1085
  @rtype: boolean
1086
  @return: the success of the operation
1087

1088
  """
1089
  parent_bdev = _RecursiveFindBD(parent_cdev, allow_partial=True)
1090
  if parent_bdev is None:
1091
    logging.error("Can't find parent device")
1092
    return False
1093
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1094
  if new_bdevs.count(None) > 0:
1095
    logging.error("Can't find new device(s) to add: %s:%s",
1096
                  new_bdevs, new_cdevs)
1097
    return False
1098
  parent_bdev.AddChildren(new_bdevs)
1099
  return True
1100

    
1101

    
1102
def MirrorRemoveChildren(parent_cdev, new_cdevs):
1103
  """Shrink a mirrored block device.
1104

1105
  @type parent_cdev: L{objects.Disk}
1106
  @param parent_cdev: the disk from which we should remove children
1107
  @type new_cdevs: list of L{objects.Disk}
1108
  @param new_cdevs: the list of children which we should remove
1109
  @rtype: boolean
1110
  @return: the success of the operation
1111

1112
  """
1113
  parent_bdev = _RecursiveFindBD(parent_cdev)
1114
  if parent_bdev is None:
1115
    logging.error("Can't find parent in remove children: %s", parent_cdev)
1116
    return False
1117
  devs = []
1118
  for disk in new_cdevs:
1119
    rpath = disk.StaticDevPath()
1120
    if rpath is None:
1121
      bd = _RecursiveFindBD(disk)
1122
      if bd is None:
1123
        logging.error("Can't find dynamic device %s while removing children",
1124
                      disk)
1125
        return False
1126
      else:
1127
        devs.append(bd.dev_path)
1128
    else:
1129
      devs.append(rpath)
1130
  parent_bdev.RemoveChildren(devs)
1131
  return True
1132

    
1133

    
1134
def GetMirrorStatus(disks):
1135
  """Get the mirroring status of a list of devices.
1136

1137
  @type disks: list of L{objects.Disk}
1138
  @param disks: the list of disks which we should query
1139
  @rtype: disk
1140
  @return:
1141
      a list of (mirror_done, estimated_time) tuples, which
1142
      are the result of L{bdev.BlockDevice.CombinedSyncStatus}
1143
  @raise errors.BlockDeviceError: if any of the disks cannot be
1144
      found
1145

1146
  """
1147
  stats = []
1148
  for dsk in disks:
1149
    rbd = _RecursiveFindBD(dsk)
1150
    if rbd is None:
1151
      raise errors.BlockDeviceError("Can't find device %s" % str(dsk))
1152
    stats.append(rbd.CombinedSyncStatus())
1153
  return stats
1154

    
1155

    
1156
def _RecursiveFindBD(disk, allow_partial=False):
1157
  """Check if a device is activated.
1158

1159
  If so, return informations about the real device.
1160

1161
  @type disk: L{objects.Disk}
1162
  @param disk: the disk object we need to find
1163
  @type allow_partial: boolean
1164
  @param allow_partial: if true, don't abort the find if a
1165
      child of the device can't be found; this is intended
1166
      to be used when repairing mirrors
1167

1168
  @return: None if the device can't be found,
1169
      otherwise the device instance
1170

1171
  """
1172
  children = []
1173
  if disk.children:
1174
    for chdisk in disk.children:
1175
      children.append(_RecursiveFindBD(chdisk))
1176

    
1177
  return bdev.FindDevice(disk.dev_type, disk.physical_id, children)
1178

    
1179

    
1180
def FindBlockDevice(disk):
1181
  """Check if a device is activated.
1182

1183
  If it is, return informations about the real device.
1184

1185
  @type disk: L{objects.Disk}
1186
  @param disk: the disk to find
1187
  @rtype: None or tuple
1188
  @return: None if the disk cannot be found, otherwise a
1189
      tuple (device_path, major, minor, sync_percent,
1190
      estimated_time, is_degraded)
1191

1192
  """
1193
  rbd = _RecursiveFindBD(disk)
1194
  if rbd is None:
1195
    return rbd
1196
  return (rbd.dev_path, rbd.major, rbd.minor) + rbd.GetSyncStatus()
1197

    
1198

    
1199
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
1200
  """Write a file to the filesystem.
1201

1202
  This allows the master to overwrite(!) a file. It will only perform
1203
  the operation if the file belongs to a list of configuration files.
1204

1205
  @type file_name: str
1206
  @param file_name: the target file name
1207
  @type data: str
1208
  @param data: the new contents of the file
1209
  @type mode: int
1210
  @param mode: the mode to give the file (can be None)
1211
  @type uid: int
1212
  @param uid: the owner of the file (can be -1 for default)
1213
  @type gid: int
1214
  @param gid: the group of the file (can be -1 for default)
1215
  @type atime: float
1216
  @param atime: the atime to set on the file (can be None)
1217
  @type mtime: float
1218
  @param mtime: the mtime to set on the file (can be None)
1219
  @rtype: boolean
1220
  @return: the success of the operation; errors are logged
1221
      in the node daemon log
1222

1223
  """
1224
  if not os.path.isabs(file_name):
1225
    logging.error("Filename passed to UploadFile is not absolute: '%s'",
1226
                  file_name)
1227
    return False
1228

    
1229
  allowed_files = [
1230
    constants.CLUSTER_CONF_FILE,
1231
    constants.ETC_HOSTS,
1232
    constants.SSH_KNOWN_HOSTS_FILE,
1233
    constants.VNC_PASSWORD_FILE,
1234
    ]
1235

    
1236
  if file_name not in allowed_files:
1237
    logging.error("Filename passed to UploadFile not in allowed"
1238
                 " upload targets: '%s'", file_name)
1239
    return False
1240

    
1241
  utils.WriteFile(file_name, data=data, mode=mode, uid=uid, gid=gid,
1242
                  atime=atime, mtime=mtime)
1243
  return True
1244

    
1245

    
1246
def WriteSsconfFiles(values):
1247
  """Update all ssconf files.
1248

1249
  Wrapper around the SimpleStore.WriteFiles.
1250

1251
  """
1252
  ssconf.SimpleStore().WriteFiles(values)
1253

    
1254

    
1255
def _ErrnoOrStr(err):
1256
  """Format an EnvironmentError exception.
1257

1258
  If the L{err} argument has an errno attribute, it will be looked up
1259
  and converted into a textual C{E...} description. Otherwise the
1260
  string representation of the error will be returned.
1261

1262
  @type err: L{EnvironmentError}
1263
  @param err: the exception to format
1264

1265
  """
1266
  if hasattr(err, 'errno'):
1267
    detail = errno.errorcode[err.errno]
1268
  else:
1269
    detail = str(err)
1270
  return detail
1271

    
1272

    
1273
def _OSOndiskVersion(name, os_dir):
1274
  """Compute and return the API version of a given OS.
1275

1276
  This function will try to read the API version of the OS given by
1277
  the 'name' parameter and residing in the 'os_dir' directory.
1278

1279
  @type name: str
1280
  @param name: the OS name we should look for
1281
  @type os_dir: str
1282
  @param os_dir: the directory inwhich we should look for the OS
1283
  @rtype: int or None
1284
  @return:
1285
      Either an integer denoting the version or None in the
1286
      case when this is not a valid OS name.
1287
  @raise errors.InvalidOS: if the OS cannot be found
1288

1289
  """
1290
  api_file = os.path.sep.join([os_dir, "ganeti_api_version"])
1291

    
1292
  try:
1293
    st = os.stat(api_file)
1294
  except EnvironmentError, err:
1295
    raise errors.InvalidOS(name, os_dir, "'ganeti_api_version' file not"
1296
                           " found (%s)" % _ErrnoOrStr(err))
1297

    
1298
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1299
    raise errors.InvalidOS(name, os_dir, "'ganeti_api_version' file is not"
1300
                           " a regular file")
1301

    
1302
  try:
1303
    f = open(api_file)
1304
    try:
1305
      api_versions = f.readlines()
1306
    finally:
1307
      f.close()
1308
  except EnvironmentError, err:
1309
    raise errors.InvalidOS(name, os_dir, "error while reading the"
1310
                           " API version (%s)" % _ErrnoOrStr(err))
1311

    
1312
  api_versions = [version.strip() for version in api_versions]
1313
  try:
1314
    api_versions = [int(version) for version in api_versions]
1315
  except (TypeError, ValueError), err:
1316
    raise errors.InvalidOS(name, os_dir,
1317
                           "API version is not integer (%s)" % str(err))
1318

    
1319
  return api_versions
1320

    
1321

    
1322
def DiagnoseOS(top_dirs=None):
1323
  """Compute the validity for all OSes.
1324

1325
  @type top_dirs: list
1326
  @param top_dirs: the list of directories in which to
1327
      search (if not given defaults to
1328
      L{constants.OS_SEARCH_PATH})
1329
  @rtype: list of L{objects.OS}
1330
  @return: an OS object for each name in all the given
1331
      directories
1332

1333
  """
1334
  if top_dirs is None:
1335
    top_dirs = constants.OS_SEARCH_PATH
1336

    
1337
  result = []
1338
  for dir_name in top_dirs:
1339
    if os.path.isdir(dir_name):
1340
      try:
1341
        f_names = utils.ListVisibleFiles(dir_name)
1342
      except EnvironmentError, err:
1343
        logging.exception("Can't list the OS directory %s", dir_name)
1344
        break
1345
      for name in f_names:
1346
        try:
1347
          os_inst = OSFromDisk(name, base_dir=dir_name)
1348
          result.append(os_inst)
1349
        except errors.InvalidOS, err:
1350
          result.append(objects.OS.FromInvalidOS(err))
1351

    
1352
  return result
1353

    
1354

    
1355
def OSFromDisk(name, base_dir=None):
1356
  """Create an OS instance from disk.
1357

1358
  This function will return an OS instance if the given name is a
1359
  valid OS name. Otherwise, it will raise an appropriate
1360
  L{errors.InvalidOS} exception, detailing why this is not a valid OS.
1361

1362
  @type base_dir: string
1363
  @keyword base_dir: Base directory containing OS installations.
1364
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
1365
  @rtype: L{objects.OS}
1366
  @return: the OS instance if we find a valid one
1367
  @raise errors.InvalidOS: if we don't find a valid OS
1368

1369
  """
1370
  if base_dir is None:
1371
    os_dir = utils.FindFile(name, constants.OS_SEARCH_PATH, os.path.isdir)
1372
    if os_dir is None:
1373
      raise errors.InvalidOS(name, None, "OS dir not found in search path")
1374
  else:
1375
    os_dir = os.path.sep.join([base_dir, name])
1376

    
1377
  api_versions = _OSOndiskVersion(name, os_dir)
1378

    
1379
  if constants.OS_API_VERSION not in api_versions:
1380
    raise errors.InvalidOS(name, os_dir, "API version mismatch"
1381
                           " (found %s want %s)"
1382
                           % (api_versions, constants.OS_API_VERSION))
1383

    
1384
  # OS Scripts dictionary, we will populate it with the actual script names
1385
  os_scripts = dict.fromkeys(constants.OS_SCRIPTS)
1386

    
1387
  for script in os_scripts:
1388
    os_scripts[script] = os.path.sep.join([os_dir, script])
1389

    
1390
    try:
1391
      st = os.stat(os_scripts[script])
1392
    except EnvironmentError, err:
1393
      raise errors.InvalidOS(name, os_dir, "'%s' script missing (%s)" %
1394
                             (script, _ErrnoOrStr(err)))
1395

    
1396
    if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
1397
      raise errors.InvalidOS(name, os_dir, "'%s' script not executable" %
1398
                             script)
1399

    
1400
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1401
      raise errors.InvalidOS(name, os_dir, "'%s' is not a regular file" %
1402
                             script)
1403

    
1404

    
1405
  return objects.OS(name=name, path=os_dir, status=constants.OS_VALID_STATUS,
1406
                    create_script=os_scripts[constants.OS_SCRIPT_CREATE],
1407
                    export_script=os_scripts[constants.OS_SCRIPT_EXPORT],
1408
                    import_script=os_scripts[constants.OS_SCRIPT_IMPORT],
1409
                    rename_script=os_scripts[constants.OS_SCRIPT_RENAME],
1410
                    api_versions=api_versions)
1411

    
1412
def OSEnvironment(instance, debug=0):
1413
  """Calculate the environment for an os script.
1414

1415
  @type instance: L{objects.Instance}
1416
  @param instance: target instance for the os script run
1417
  @type debug: integer
1418
  @param debug: debug level (0 or 1, for OS Api 10)
1419
  @rtype: dict
1420
  @return: dict of environment variables
1421
  @raise errors.BlockDeviceError: if the block device
1422
      cannot be found
1423

1424
  """
1425
  result = {}
1426
  result['OS_API_VERSION'] = '%d' % constants.OS_API_VERSION
1427
  result['INSTANCE_NAME'] = instance.name
1428
  result['HYPERVISOR'] = instance.hypervisor
1429
  result['DISK_COUNT'] = '%d' % len(instance.disks)
1430
  result['NIC_COUNT'] = '%d' % len(instance.nics)
1431
  result['DEBUG_LEVEL'] = '%d' % debug
1432
  for idx, disk in enumerate(instance.disks):
1433
    real_disk = _RecursiveFindBD(disk)
1434
    if real_disk is None:
1435
      raise errors.BlockDeviceError("Block device '%s' is not set up" %
1436
                                    str(disk))
1437
    real_disk.Open()
1438
    result['DISK_%d_PATH' % idx] = real_disk.dev_path
1439
    # FIXME: When disks will have read-only mode, populate this
1440
    result['DISK_%d_ACCESS' % idx] = 'W'
1441
    if constants.HV_DISK_TYPE in instance.hvparams:
1442
      result['DISK_%d_FRONTEND_TYPE' % idx] = \
1443
        instance.hvparams[constants.HV_DISK_TYPE]
1444
    if disk.dev_type in constants.LDS_BLOCK:
1445
      result['DISK_%d_BACKEND_TYPE' % idx] = 'block'
1446
    elif disk.dev_type == constants.LD_FILE:
1447
      result['DISK_%d_BACKEND_TYPE' % idx] = \
1448
        'file:%s' % disk.physical_id[0]
1449
  for idx, nic in enumerate(instance.nics):
1450
    result['NIC_%d_MAC' % idx] = nic.mac
1451
    if nic.ip:
1452
      result['NIC_%d_IP' % idx] = nic.ip
1453
    result['NIC_%d_BRIDGE' % idx] = nic.bridge
1454
    if constants.HV_NIC_TYPE in instance.hvparams:
1455
      result['NIC_%d_FRONTEND_TYPE' % idx] = \
1456
        instance.hvparams[constants.HV_NIC_TYPE]
1457

    
1458
  return result
1459

    
1460
def GrowBlockDevice(disk, amount):
1461
  """Grow a stack of block devices.
1462

1463
  This function is called recursively, with the childrens being the
1464
  first ones to resize.
1465

1466
  @type disk: L{objects.Disk}
1467
  @param disk: the disk to be grown
1468
  @rtype: (status, result)
1469
  @return: a tuple with the status of the operation
1470
      (True/False), and the errors message if status
1471
      is False
1472

1473
  """
1474
  r_dev = _RecursiveFindBD(disk)
1475
  if r_dev is None:
1476
    return False, "Cannot find block device %s" % (disk,)
1477

    
1478
  try:
1479
    r_dev.Grow(amount)
1480
  except errors.BlockDeviceError, err:
1481
    return False, str(err)
1482

    
1483
  return True, None
1484

    
1485

    
1486
def SnapshotBlockDevice(disk):
1487
  """Create a snapshot copy of a block device.
1488

1489
  This function is called recursively, and the snapshot is actually created
1490
  just for the leaf lvm backend device.
1491

1492
  @type disk: L{objects.Disk}
1493
  @param disk: the disk to be snapshotted
1494
  @rtype: string
1495
  @return: snapshot disk path
1496

1497
  """
1498
  if disk.children:
1499
    if len(disk.children) == 1:
1500
      # only one child, let's recurse on it
1501
      return SnapshotBlockDevice(disk.children[0])
1502
    else:
1503
      # more than one child, choose one that matches
1504
      for child in disk.children:
1505
        if child.size == disk.size:
1506
          # return implies breaking the loop
1507
          return SnapshotBlockDevice(child)
1508
  elif disk.dev_type == constants.LD_LV:
1509
    r_dev = _RecursiveFindBD(disk)
1510
    if r_dev is not None:
1511
      # let's stay on the safe side and ask for the full size, for now
1512
      return r_dev.Snapshot(disk.size)
1513
    else:
1514
      return None
1515
  else:
1516
    raise errors.ProgrammerError("Cannot snapshot non-lvm block device"
1517
                                 " '%s' of type '%s'" %
1518
                                 (disk.unique_id, disk.dev_type))
1519

    
1520

    
1521
def ExportSnapshot(disk, dest_node, instance, cluster_name, idx):
1522
  """Export a block device snapshot to a remote node.
1523

1524
  @type disk: L{objects.Disk}
1525
  @param disk: the description of the disk to export
1526
  @type dest_node: str
1527
  @param dest_node: the destination node to export to
1528
  @type instance: L{objects.Instance}
1529
  @param instance: the instance object to whom the disk belongs
1530
  @type cluster_name: str
1531
  @param cluster_name: the cluster name, needed for SSH hostalias
1532
  @type idx: int
1533
  @param idx: the index of the disk in the instance's disk list,
1534
      used to export to the OS scripts environment
1535
  @rtype: boolean
1536
  @return: the success of the operation
1537

1538
  """
1539
  export_env = OSEnvironment(instance)
1540

    
1541
  inst_os = OSFromDisk(instance.os)
1542
  export_script = inst_os.export_script
1543

    
1544
  logfile = "%s/exp-%s-%s-%s.log" % (constants.LOG_OS_DIR, inst_os.name,
1545
                                     instance.name, int(time.time()))
1546
  if not os.path.exists(constants.LOG_OS_DIR):
1547
    os.mkdir(constants.LOG_OS_DIR, 0750)
1548
  real_disk = _RecursiveFindBD(disk)
1549
  if real_disk is None:
1550
    raise errors.BlockDeviceError("Block device '%s' is not set up" %
1551
                                  str(disk))
1552
  real_disk.Open()
1553

    
1554
  export_env['EXPORT_DEVICE'] = real_disk.dev_path
1555
  export_env['EXPORT_INDEX'] = str(idx)
1556

    
1557
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
1558
  destfile = disk.physical_id[1]
1559

    
1560
  # the target command is built out of three individual commands,
1561
  # which are joined by pipes; we check each individual command for
1562
  # valid parameters
1563
  expcmd = utils.BuildShellCmd("cd %s; %s 2>%s", inst_os.path,
1564
                               export_script, logfile)
1565

    
1566
  comprcmd = "gzip"
1567

    
1568
  destcmd = utils.BuildShellCmd("mkdir -p %s && cat > %s/%s",
1569
                                destdir, destdir, destfile)
1570
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
1571
                                                   constants.GANETI_RUNAS,
1572
                                                   destcmd)
1573

    
1574
  # all commands have been checked, so we're safe to combine them
1575
  command = '|'.join([expcmd, comprcmd, utils.ShellQuoteArgs(remotecmd)])
1576

    
1577
  result = utils.RunCmd(command, env=export_env)
1578

    
1579
  if result.failed:
1580
    logging.error("os snapshot export command '%s' returned error: %s"
1581
                  " output: %s", command, result.fail_reason, result.output)
1582
    return False
1583

    
1584
  return True
1585

    
1586

    
1587
def FinalizeExport(instance, snap_disks):
1588
  """Write out the export configuration information.
1589

1590
  @type instance: L{objects.Instance}
1591
  @param instance: the instance which we export, used for
1592
      saving configuration
1593
  @type snap_disks: list of L{objects.Disk}
1594
  @param snap_disks: list of snapshot block devices, which
1595
      will be used to get the actual name of the dump file
1596

1597
  @rtype: boolean
1598
  @return: the success of the operation
1599

1600
  """
1601
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
1602
  finaldestdir = os.path.join(constants.EXPORT_DIR, instance.name)
1603

    
1604
  config = objects.SerializableConfigParser()
1605

    
1606
  config.add_section(constants.INISECT_EXP)
1607
  config.set(constants.INISECT_EXP, 'version', '0')
1608
  config.set(constants.INISECT_EXP, 'timestamp', '%d' % int(time.time()))
1609
  config.set(constants.INISECT_EXP, 'source', instance.primary_node)
1610
  config.set(constants.INISECT_EXP, 'os', instance.os)
1611
  config.set(constants.INISECT_EXP, 'compression', 'gzip')
1612

    
1613
  config.add_section(constants.INISECT_INS)
1614
  config.set(constants.INISECT_INS, 'name', instance.name)
1615
  config.set(constants.INISECT_INS, 'memory', '%d' %
1616
             instance.beparams[constants.BE_MEMORY])
1617
  config.set(constants.INISECT_INS, 'vcpus', '%d' %
1618
             instance.beparams[constants.BE_VCPUS])
1619
  config.set(constants.INISECT_INS, 'disk_template', instance.disk_template)
1620

    
1621
  nic_count = 0
1622
  for nic_count, nic in enumerate(instance.nics):
1623
    config.set(constants.INISECT_INS, 'nic%d_mac' %
1624
               nic_count, '%s' % nic.mac)
1625
    config.set(constants.INISECT_INS, 'nic%d_ip' % nic_count, '%s' % nic.ip)
1626
    config.set(constants.INISECT_INS, 'nic%d_bridge' % nic_count,
1627
               '%s' % nic.bridge)
1628
  # TODO: redundant: on load can read nics until it doesn't exist
1629
  config.set(constants.INISECT_INS, 'nic_count' , '%d' % nic_count)
1630

    
1631
  disk_total = 0
1632
  for disk_count, disk in enumerate(snap_disks):
1633
    if disk:
1634
      disk_total += 1
1635
      config.set(constants.INISECT_INS, 'disk%d_ivname' % disk_count,
1636
                 ('%s' % disk.iv_name))
1637
      config.set(constants.INISECT_INS, 'disk%d_dump' % disk_count,
1638
                 ('%s' % disk.physical_id[1]))
1639
      config.set(constants.INISECT_INS, 'disk%d_size' % disk_count,
1640
                 ('%d' % disk.size))
1641

    
1642
  config.set(constants.INISECT_INS, 'disk_count' , '%d' % disk_total)
1643

    
1644
  utils.WriteFile(os.path.join(destdir, constants.EXPORT_CONF_FILE),
1645
                  data=config.Dumps())
1646
  shutil.rmtree(finaldestdir, True)
1647
  shutil.move(destdir, finaldestdir)
1648

    
1649
  return True
1650

    
1651

    
1652
def ExportInfo(dest):
1653
  """Get export configuration information.
1654

1655
  @type dest: str
1656
  @param dest: directory containing the export
1657

1658
  @rtype: L{objects.SerializableConfigParser}
1659
  @return: a serializable config file containing the
1660
      export info
1661

1662
  """
1663
  cff = os.path.join(dest, constants.EXPORT_CONF_FILE)
1664

    
1665
  config = objects.SerializableConfigParser()
1666
  config.read(cff)
1667

    
1668
  if (not config.has_section(constants.INISECT_EXP) or
1669
      not config.has_section(constants.INISECT_INS)):
1670
    return None
1671

    
1672
  return config
1673

    
1674

    
1675
def ImportOSIntoInstance(instance, src_node, src_images, cluster_name):
1676
  """Import an os image into an instance.
1677

1678
  @type instance: L{objects.Instance}
1679
  @param instance: instance to import the disks into
1680
  @type src_node: string
1681
  @param src_node: source node for the disk images
1682
  @type src_images: list of string
1683
  @param src_images: absolute paths of the disk images
1684
  @rtype: list of boolean
1685
  @return: each boolean represent the success of importing the n-th disk
1686

1687
  """
1688
  import_env = OSEnvironment(instance)
1689
  inst_os = OSFromDisk(instance.os)
1690
  import_script = inst_os.import_script
1691

    
1692
  logfile = "%s/import-%s-%s-%s.log" % (constants.LOG_OS_DIR, instance.os,
1693
                                        instance.name, int(time.time()))
1694
  if not os.path.exists(constants.LOG_OS_DIR):
1695
    os.mkdir(constants.LOG_OS_DIR, 0750)
1696

    
1697
  comprcmd = "gunzip"
1698
  impcmd = utils.BuildShellCmd("(cd %s; %s >%s 2>&1)", inst_os.path,
1699
                               import_script, logfile)
1700

    
1701
  final_result = []
1702
  for idx, image in enumerate(src_images):
1703
    if image:
1704
      destcmd = utils.BuildShellCmd('cat %s', image)
1705
      remotecmd = _GetSshRunner(cluster_name).BuildCmd(src_node,
1706
                                                       constants.GANETI_RUNAS,
1707
                                                       destcmd)
1708
      command = '|'.join([utils.ShellQuoteArgs(remotecmd), comprcmd, impcmd])
1709
      import_env['IMPORT_DEVICE'] = import_env['DISK_%d_PATH' % idx]
1710
      import_env['IMPORT_INDEX'] = str(idx)
1711
      result = utils.RunCmd(command, env=import_env)
1712
      if result.failed:
1713
        logging.error("Disk import command '%s' returned error: %s"
1714
                      " output: %s", command, result.fail_reason,
1715
                      result.output)
1716
        final_result.append(False)
1717
      else:
1718
        final_result.append(True)
1719
    else:
1720
      final_result.append(True)
1721

    
1722
  return final_result
1723

    
1724

    
1725
def ListExports():
1726
  """Return a list of exports currently available on this machine.
1727

1728
  @rtype: list
1729
  @return: list of the exports
1730

1731
  """
1732
  if os.path.isdir(constants.EXPORT_DIR):
1733
    return utils.ListVisibleFiles(constants.EXPORT_DIR)
1734
  else:
1735
    return []
1736

    
1737

    
1738
def RemoveExport(export):
1739
  """Remove an existing export from the node.
1740

1741
  @type export: str
1742
  @param export: the name of the export to remove
1743
  @rtype: boolean
1744
  @return: the success of the operation
1745

1746
  """
1747
  target = os.path.join(constants.EXPORT_DIR, export)
1748

    
1749
  shutil.rmtree(target)
1750
  # TODO: catch some of the relevant exceptions and provide a pretty
1751
  # error message if rmtree fails.
1752

    
1753
  return True
1754

    
1755

    
1756
def RenameBlockDevices(devlist):
1757
  """Rename a list of block devices.
1758

1759
  @type devlist: list of tuples
1760
  @param devlist: list of tuples of the form  (disk,
1761
      new_logical_id, new_physical_id); disk is an
1762
      L{objects.Disk} object describing the current disk,
1763
      and new logical_id/physical_id is the name we
1764
      rename it to
1765
  @rtype: boolean
1766
  @return: True if all renames succeeded, False otherwise
1767

1768
  """
1769
  result = True
1770
  for disk, unique_id in devlist:
1771
    dev = _RecursiveFindBD(disk)
1772
    if dev is None:
1773
      result = False
1774
      continue
1775
    try:
1776
      old_rpath = dev.dev_path
1777
      dev.Rename(unique_id)
1778
      new_rpath = dev.dev_path
1779
      if old_rpath != new_rpath:
1780
        DevCacheManager.RemoveCache(old_rpath)
1781
        # FIXME: we should add the new cache information here, like:
1782
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
1783
        # but we don't have the owner here - maybe parse from existing
1784
        # cache? for now, we only lose lvm data when we rename, which
1785
        # is less critical than DRBD or MD
1786
    except errors.BlockDeviceError, err:
1787
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
1788
      result = False
1789
  return result
1790

    
1791

    
1792
def _TransformFileStorageDir(file_storage_dir):
1793
  """Checks whether given file_storage_dir is valid.
1794

1795
  Checks wheter the given file_storage_dir is within the cluster-wide
1796
  default file_storage_dir stored in SimpleStore. Only paths under that
1797
  directory are allowed.
1798

1799
  @type file_storage_dir: str
1800
  @param file_storage_dir: the path to check
1801

1802
  @return: the normalized path if valid, None otherwise
1803

1804
  """
1805
  cfg = _GetConfig()
1806
  file_storage_dir = os.path.normpath(file_storage_dir)
1807
  base_file_storage_dir = cfg.GetFileStorageDir()
1808
  if (not os.path.commonprefix([file_storage_dir, base_file_storage_dir]) ==
1809
      base_file_storage_dir):
1810
    logging.error("file storage directory '%s' is not under base file"
1811
                  " storage directory '%s'",
1812
                  file_storage_dir, base_file_storage_dir)
1813
    return None
1814
  return file_storage_dir
1815

    
1816

    
1817
def CreateFileStorageDir(file_storage_dir):
1818
  """Create file storage directory.
1819

1820
  @type file_storage_dir: str
1821
  @param file_storage_dir: directory to create
1822

1823
  @rtype: tuple
1824
  @return: tuple with first element a boolean indicating wheter dir
1825
      creation was successful or not
1826

1827
  """
1828
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
1829
  result = True,
1830
  if not file_storage_dir:
1831
    result = False,
1832
  else:
1833
    if os.path.exists(file_storage_dir):
1834
      if not os.path.isdir(file_storage_dir):
1835
        logging.error("'%s' is not a directory", file_storage_dir)
1836
        result = False,
1837
    else:
1838
      try:
1839
        os.makedirs(file_storage_dir, 0750)
1840
      except OSError, err:
1841
        logging.error("Cannot create file storage directory '%s': %s",
1842
                      file_storage_dir, err)
1843
        result = False,
1844
  return result
1845

    
1846

    
1847
def RemoveFileStorageDir(file_storage_dir):
1848
  """Remove file storage directory.
1849

1850
  Remove it only if it's empty. If not log an error and return.
1851

1852
  @type file_storage_dir: str
1853
  @param file_storage_dir: the directory we should cleanup
1854
  @rtype: tuple (success,)
1855
  @return: tuple of one element, C{success}, denoting
1856
      whether the operation was successfull
1857

1858
  """
1859
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
1860
  result = True,
1861
  if not file_storage_dir:
1862
    result = False,
1863
  else:
1864
    if os.path.exists(file_storage_dir):
1865
      if not os.path.isdir(file_storage_dir):
1866
        logging.error("'%s' is not a directory", file_storage_dir)
1867
        result = False,
1868
      # deletes dir only if empty, otherwise we want to return False
1869
      try:
1870
        os.rmdir(file_storage_dir)
1871
      except OSError, err:
1872
        logging.exception("Cannot remove file storage directory '%s'",
1873
                          file_storage_dir)
1874
        result = False,
1875
  return result
1876

    
1877

    
1878
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
1879
  """Rename the file storage directory.
1880

1881
  @type old_file_storage_dir: str
1882
  @param old_file_storage_dir: the current path
1883
  @type new_file_storage_dir: str
1884
  @param new_file_storage_dir: the name we should rename to
1885
  @rtype: tuple (success,)
1886
  @return: tuple of one element, C{success}, denoting
1887
      whether the operation was successful
1888

1889
  """
1890
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
1891
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
1892
  result = True,
1893
  if not old_file_storage_dir or not new_file_storage_dir:
1894
    result = False,
1895
  else:
1896
    if not os.path.exists(new_file_storage_dir):
1897
      if os.path.isdir(old_file_storage_dir):
1898
        try:
1899
          os.rename(old_file_storage_dir, new_file_storage_dir)
1900
        except OSError, err:
1901
          logging.exception("Cannot rename '%s' to '%s'",
1902
                            old_file_storage_dir, new_file_storage_dir)
1903
          result =  False,
1904
      else:
1905
        logging.error("'%s' is not a directory", old_file_storage_dir)
1906
        result = False,
1907
    else:
1908
      if os.path.exists(old_file_storage_dir):
1909
        logging.error("Cannot rename '%s' to '%s'. Both locations exist.",
1910
                      old_file_storage_dir, new_file_storage_dir)
1911
        result = False,
1912
  return result
1913

    
1914

    
1915
def _IsJobQueueFile(file_name):
1916
  """Checks whether the given filename is in the queue directory.
1917

1918
  @type file_name: str
1919
  @param file_name: the file name we should check
1920
  @rtype: boolean
1921
  @return: whether the file is under the queue directory
1922

1923
  """
1924
  queue_dir = os.path.normpath(constants.QUEUE_DIR)
1925
  result = (os.path.commonprefix([queue_dir, file_name]) == queue_dir)
1926

    
1927
  if not result:
1928
    logging.error("'%s' is not a file in the queue directory",
1929
                  file_name)
1930

    
1931
  return result
1932

    
1933

    
1934
def JobQueueUpdate(file_name, content):
1935
  """Updates a file in the queue directory.
1936

1937
  This is just a wrapper over L{utils.WriteFile}, with proper
1938
  checking.
1939

1940
  @type file_name: str
1941
  @param file_name: the job file name
1942
  @type content: str
1943
  @param content: the new job contents
1944
  @rtype: boolean
1945
  @return: the success of the operation
1946

1947
  """
1948
  if not _IsJobQueueFile(file_name):
1949
    return False
1950

    
1951
  # Write and replace the file atomically
1952
  utils.WriteFile(file_name, data=content)
1953

    
1954
  return True
1955

    
1956

    
1957
def JobQueueRename(old, new):
1958
  """Renames a job queue file.
1959

1960
  This is just a wrapper over L{os.rename} with proper checking.
1961

1962
  @type old: str
1963
  @param old: the old (actual) file name
1964
  @type new: str
1965
  @param new: the desired file name
1966
  @rtype: boolean
1967
  @return: the success of the operation
1968

1969
  """
1970
  if not (_IsJobQueueFile(old) and _IsJobQueueFile(new)):
1971
    return False
1972

    
1973
  os.rename(old, new)
1974

    
1975
  return True
1976

    
1977

    
1978
def JobQueueSetDrainFlag(drain_flag):
1979
  """Set the drain flag for the queue.
1980

1981
  This will set or unset the queue drain flag.
1982

1983
  @type drain_flag: boolean
1984
  @param drain_flag: if True, will set the drain flag, otherwise reset it.
1985
  @rtype: boolean
1986
  @return: always True
1987
  @warning: the function always returns True
1988

1989
  """
1990
  if drain_flag:
1991
    utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
1992
  else:
1993
    utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
1994

    
1995
  return True
1996

    
1997

    
1998
def CloseBlockDevices(disks):
1999
  """Closes the given block devices.
2000

2001
  This means they will be switched to secondary mode (in case of
2002
  DRBD).
2003

2004
  @type disks: list of L{objects.Disk}
2005
  @param disks: the list of disks to be closed
2006
  @rtype: tuple (success, message)
2007
  @return: a tuple of success and message, where success
2008
      indicates the succes of the operation, and message
2009
      which will contain the error details in case we
2010
      failed
2011

2012
  """
2013
  bdevs = []
2014
  for cf in disks:
2015
    rd = _RecursiveFindBD(cf)
2016
    if rd is None:
2017
      return (False, "Can't find device %s" % cf)
2018
    bdevs.append(rd)
2019

    
2020
  msg = []
2021
  for rd in bdevs:
2022
    try:
2023
      rd.Close()
2024
    except errors.BlockDeviceError, err:
2025
      msg.append(str(err))
2026
  if msg:
2027
    return (False, "Can't make devices secondary: %s" % ",".join(msg))
2028
  else:
2029
    return (True, "All devices secondary")
2030

    
2031

    
2032
def ValidateHVParams(hvname, hvparams):
2033
  """Validates the given hypervisor parameters.
2034

2035
  @type hvname: string
2036
  @param hvname: the hypervisor name
2037
  @type hvparams: dict
2038
  @param hvparams: the hypervisor parameters to be validated
2039
  @rtype: tuple (success, message)
2040
  @return: a tuple of success and message, where success
2041
      indicates the succes of the operation, and message
2042
      which will contain the error details in case we
2043
      failed
2044

2045
  """
2046
  try:
2047
    hv_type = hypervisor.GetHypervisor(hvname)
2048
    hv_type.ValidateParameters(hvparams)
2049
    return (True, "Validation passed")
2050
  except errors.HypervisorError, err:
2051
    return (False, str(err))
2052

    
2053

    
2054
def DemoteFromMC():
2055
  """Demotes the current node from master candidate role.
2056

2057
  """
2058
  # try to ensure we're not the master by mistake
2059
  master, myself = ssconf.GetMasterAndMyself()
2060
  if master == myself:
2061
    return (False, "ssconf status shows I'm the master node, will not demote")
2062
  pid_file = utils.DaemonPidFileName(constants.MASTERD_PID)
2063
  if utils.IsProcessAlive(utils.ReadPidFile(pid_file)):
2064
    return (False, "The master daemon is running, will not demote")
2065
  try:
2066
    utils.CreateBackup(constants.CLUSTER_CONF_FILE)
2067
  except EnvironmentError, err:
2068
    if err.errno != errno.ENOENT:
2069
      return (False, "Error while backing up cluster file: %s" % str(err))
2070
  utils.RemoveFile(constants.CLUSTER_CONF_FILE)
2071
  return (True, "Done")
2072

    
2073

    
2074
class HooksRunner(object):
2075
  """Hook runner.
2076

2077
  This class is instantiated on the node side (ganeti-noded) and not
2078
  on the master side.
2079

2080
  """
2081
  RE_MASK = re.compile("^[a-zA-Z0-9_-]+$")
2082

    
2083
  def __init__(self, hooks_base_dir=None):
2084
    """Constructor for hooks runner.
2085

2086
    @type hooks_base_dir: str or None
2087
    @param hooks_base_dir: if not None, this overrides the
2088
        L{constants.HOOKS_BASE_DIR} (useful for unittests)
2089

2090
    """
2091
    if hooks_base_dir is None:
2092
      hooks_base_dir = constants.HOOKS_BASE_DIR
2093
    self._BASE_DIR = hooks_base_dir
2094

    
2095
  @staticmethod
2096
  def ExecHook(script, env):
2097
    """Exec one hook script.
2098

2099
    @type script: str
2100
    @param script: the full path to the script
2101
    @type env: dict
2102
    @param env: the environment with which to exec the script
2103
    @rtype: tuple (success, message)
2104
    @return: a tuple of success and message, where success
2105
        indicates the succes of the operation, and message
2106
        which will contain the error details in case we
2107
        failed
2108

2109
    """
2110
    # exec the process using subprocess and log the output
2111
    fdstdin = None
2112
    try:
2113
      fdstdin = open("/dev/null", "r")
2114
      child = subprocess.Popen([script], stdin=fdstdin, stdout=subprocess.PIPE,
2115
                               stderr=subprocess.STDOUT, close_fds=True,
2116
                               shell=False, cwd="/", env=env)
2117
      output = ""
2118
      try:
2119
        output = child.stdout.read(4096)
2120
        child.stdout.close()
2121
      except EnvironmentError, err:
2122
        output += "Hook script error: %s" % str(err)
2123

    
2124
      while True:
2125
        try:
2126
          result = child.wait()
2127
          break
2128
        except EnvironmentError, err:
2129
          if err.errno == errno.EINTR:
2130
            continue
2131
          raise
2132
    finally:
2133
      # try not to leak fds
2134
      for fd in (fdstdin, ):
2135
        if fd is not None:
2136
          try:
2137
            fd.close()
2138
          except EnvironmentError, err:
2139
            # just log the error
2140
            #logging.exception("Error while closing fd %s", fd)
2141
            pass
2142

    
2143
    return result == 0, output
2144

    
2145
  def RunHooks(self, hpath, phase, env):
2146
    """Run the scripts in the hooks directory.
2147

2148
    @type hpath: str
2149
    @param hpath: the path to the hooks directory which
2150
        holds the scripts
2151
    @type phase: str
2152
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
2153
        L{constants.HOOKS_PHASE_POST}
2154
    @type env: dict
2155
    @param env: dictionary with the environment for the hook
2156
    @rtype: list
2157
    @return: list of 3-element tuples:
2158
      - script path
2159
      - script result, either L{constants.HKR_SUCCESS} or
2160
        L{constants.HKR_FAIL}
2161
      - output of the script
2162

2163
    @raise errors.ProgrammerError: for invalid input
2164
        parameters
2165

2166
    """
2167
    if phase == constants.HOOKS_PHASE_PRE:
2168
      suffix = "pre"
2169
    elif phase == constants.HOOKS_PHASE_POST:
2170
      suffix = "post"
2171
    else:
2172
      raise errors.ProgrammerError("Unknown hooks phase: '%s'" % phase)
2173
    rr = []
2174

    
2175
    subdir = "%s-%s.d" % (hpath, suffix)
2176
    dir_name = "%s/%s" % (self._BASE_DIR, subdir)
2177
    try:
2178
      dir_contents = utils.ListVisibleFiles(dir_name)
2179
    except OSError, err:
2180
      # FIXME: must log output in case of failures
2181
      return rr
2182

    
2183
    # we use the standard python sort order,
2184
    # so 00name is the recommended naming scheme
2185
    dir_contents.sort()
2186
    for relname in dir_contents:
2187
      fname = os.path.join(dir_name, relname)
2188
      if not (os.path.isfile(fname) and os.access(fname, os.X_OK) and
2189
          self.RE_MASK.match(relname) is not None):
2190
        rrval = constants.HKR_SKIP
2191
        output = ""
2192
      else:
2193
        result, output = self.ExecHook(fname, env)
2194
        if not result:
2195
          rrval = constants.HKR_FAIL
2196
        else:
2197
          rrval = constants.HKR_SUCCESS
2198
      rr.append(("%s/%s" % (subdir, relname), rrval, output))
2199

    
2200
    return rr
2201

    
2202

    
2203
class IAllocatorRunner(object):
2204
  """IAllocator runner.
2205

2206
  This class is instantiated on the node side (ganeti-noded) and not on
2207
  the master side.
2208

2209
  """
2210
  def Run(self, name, idata):
2211
    """Run an iallocator script.
2212

2213
    @type name: str
2214
    @param name: the iallocator script name
2215
    @type idata: str
2216
    @param idata: the allocator input data
2217

2218
    @rtype: tuple
2219
    @return: four element tuple of:
2220
       - run status (one of the IARUN_ constants)
2221
       - stdout
2222
       - stderr
2223
       - fail reason (as from L{utils.RunResult})
2224

2225
    """
2226
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
2227
                                  os.path.isfile)
2228
    if alloc_script is None:
2229
      return (constants.IARUN_NOTFOUND, None, None, None)
2230

    
2231
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
2232
    try:
2233
      os.write(fd, idata)
2234
      os.close(fd)
2235
      result = utils.RunCmd([alloc_script, fin_name])
2236
      if result.failed:
2237
        return (constants.IARUN_FAILURE, result.stdout, result.stderr,
2238
                result.fail_reason)
2239
    finally:
2240
      os.unlink(fin_name)
2241

    
2242
    return (constants.IARUN_SUCCESS, result.stdout, result.stderr, None)
2243

    
2244

    
2245
class DevCacheManager(object):
2246
  """Simple class for managing a cache of block device information.
2247

2248
  """
2249
  _DEV_PREFIX = "/dev/"
2250
  _ROOT_DIR = constants.BDEV_CACHE_DIR
2251

    
2252
  @classmethod
2253
  def _ConvertPath(cls, dev_path):
2254
    """Converts a /dev/name path to the cache file name.
2255

2256
    This replaces slashes with underscores and strips the /dev
2257
    prefix. It then returns the full path to the cache file.
2258

2259
    @type dev_path: str
2260
    @param dev_path: the C{/dev/} path name
2261
    @rtype: str
2262
    @return: the converted path name
2263

2264
    """
2265
    if dev_path.startswith(cls._DEV_PREFIX):
2266
      dev_path = dev_path[len(cls._DEV_PREFIX):]
2267
    dev_path = dev_path.replace("/", "_")
2268
    fpath = "%s/bdev_%s" % (cls._ROOT_DIR, dev_path)
2269
    return fpath
2270

    
2271
  @classmethod
2272
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
2273
    """Updates the cache information for a given device.
2274

2275
    @type dev_path: str
2276
    @param dev_path: the pathname of the device
2277
    @type owner: str
2278
    @param owner: the owner (instance name) of the device
2279
    @type on_primary: bool
2280
    @param on_primary: whether this is the primary
2281
        node nor not
2282
    @type iv_name: str
2283
    @param iv_name: the instance-visible name of the
2284
        device, as in L{objects.Disk.iv_name}
2285

2286
    @rtype: None
2287

2288
    """
2289
    if dev_path is None:
2290
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
2291
      return
2292
    fpath = cls._ConvertPath(dev_path)
2293
    if on_primary:
2294
      state = "primary"
2295
    else:
2296
      state = "secondary"
2297
    if iv_name is None:
2298
      iv_name = "not_visible"
2299
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
2300
    try:
2301
      utils.WriteFile(fpath, data=fdata)
2302
    except EnvironmentError, err:
2303
      logging.exception("Can't update bdev cache for %s", dev_path)
2304

    
2305
  @classmethod
2306
  def RemoveCache(cls, dev_path):
2307
    """Remove data for a dev_path.
2308

2309
    This is just a wrapper over L{utils.RemoveFile} with a converted
2310
    path name and logging.
2311

2312
    @type dev_path: str
2313
    @param dev_path: the pathname of the device
2314

2315
    @rtype: None
2316

2317
    """
2318
    if dev_path is None:
2319
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
2320
      return
2321
    fpath = cls._ConvertPath(dev_path)
2322
    try:
2323
      utils.RemoveFile(fpath)
2324
    except EnvironmentError, err:
2325
      logging.exception("Can't update bdev cache for %s", dev_path)