Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ 93384844

History | View | Annotate | Download (67.7 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 'hypervisor' in what:
354
    result['hypervisor'] = my_dict = {}
355
    for hv_name in what['hypervisor']:
356
      my_dict[hv_name] = hypervisor.GetHypervisor(hv_name).Verify()
357

    
358
  if 'filelist' in what:
359
    result['filelist'] = utils.FingerprintFiles(what['filelist'])
360

    
361
  if 'nodelist' in what:
362
    result['nodelist'] = {}
363
    random.shuffle(what['nodelist'])
364
    for node in what['nodelist']:
365
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
366
      if not success:
367
        result['nodelist'][node] = message
368
  if 'node-net-test' in what:
369
    result['node-net-test'] = {}
370
    my_name = utils.HostInfo().name
371
    my_pip = my_sip = None
372
    for name, pip, sip in what['node-net-test']:
373
      if name == my_name:
374
        my_pip = pip
375
        my_sip = sip
376
        break
377
    if not my_pip:
378
      result['node-net-test'][my_name] = ("Can't find my own"
379
                                          " primary/secondary IP"
380
                                          " in the node list")
381
    else:
382
      port = utils.GetNodeDaemonPort()
383
      for name, pip, sip in what['node-net-test']:
384
        fail = []
385
        if not utils.TcpPing(pip, port, source=my_pip):
386
          fail.append("primary")
387
        if sip != pip:
388
          if not utils.TcpPing(sip, port, source=my_sip):
389
            fail.append("secondary")
390
        if fail:
391
          result['node-net-test'][name] = ("failure using the %s"
392
                                           " interface(s)" %
393
                                           " and ".join(fail))
394

    
395
  return result
396

    
397

    
398
def GetVolumeList(vg_name):
399
  """Compute list of logical volumes and their size.
400

401
  @type vg_name: str
402
  @param vg_name: the volume group whose LVs we should list
403
  @rtype: dict
404
  @return:
405
      dictionary of all partions (key) with value being a tuple of
406
      their size (in MiB), inactive and online status::
407

408
        {'test1': ('20.06', True, True)}
409

410
      in case of errors, a string is returned with the error
411
      details.
412

413
  """
414
  lvs = {}
415
  sep = '|'
416
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
417
                         "--separator=%s" % sep,
418
                         "-olv_name,lv_size,lv_attr", vg_name])
419
  if result.failed:
420
    logging.error("Failed to list logical volumes, lvs output: %s",
421
                  result.output)
422
    return result.output
423

    
424
  valid_line_re = re.compile("^ *([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
425
  for line in result.stdout.splitlines():
426
    line = line.strip()
427
    match = valid_line_re.match(line)
428
    if not match:
429
      logging.error("Invalid line returned from lvs output: '%s'", line)
430
      continue
431
    name, size, attr = match.groups()
432
    inactive = attr[4] == '-'
433
    online = attr[5] == 'o'
434
    lvs[name] = (size, inactive, online)
435

    
436
  return lvs
437

    
438

    
439
def ListVolumeGroups():
440
  """List the volume groups and their size.
441

442
  @rtype: dict
443
  @return: dictionary with keys volume name and values the
444
      size of the volume
445

446
  """
447
  return utils.ListVolumeGroups()
448

    
449

    
450
def NodeVolumes():
451
  """List all volumes on this node.
452

453
  @rtype: list
454
  @return:
455
    A list of dictionaries, each having four keys:
456
      - name: the logical volume name,
457
      - size: the size of the logical volume
458
      - dev: the physical device on which the LV lives
459
      - vg: the volume group to which it belongs
460

461
    In case of errors, we return an empty list and log the
462
    error.
463

464
    Note that since a logical volume can live on multiple physical
465
    volumes, the resulting list might include a logical volume
466
    multiple times.
467

468
  """
469
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
470
                         "--separator=|",
471
                         "--options=lv_name,lv_size,devices,vg_name"])
472
  if result.failed:
473
    logging.error("Failed to list logical volumes, lvs output: %s",
474
                  result.output)
475
    return []
476

    
477
  def parse_dev(dev):
478
    if '(' in dev:
479
      return dev.split('(')[0]
480
    else:
481
      return dev
482

    
483
  def map_line(line):
484
    return {
485
      'name': line[0].strip(),
486
      'size': line[1].strip(),
487
      'dev': parse_dev(line[2].strip()),
488
      'vg': line[3].strip(),
489
    }
490

    
491
  return [map_line(line.split('|')) for line in result.stdout.splitlines()
492
          if line.count('|') >= 3]
493

    
494

    
495
def BridgesExist(bridges_list):
496
  """Check if a list of bridges exist on the current node.
497

498
  @rtype: boolean
499
  @return: C{True} if all of them exist, C{False} otherwise
500

501
  """
502
  for bridge in bridges_list:
503
    if not utils.BridgeExists(bridge):
504
      return False
505

    
506
  return True
507

    
508

    
509
def GetInstanceList(hypervisor_list):
510
  """Provides a list of instances.
511

512
  @type hypervisor_list: list
513
  @param hypervisor_list: the list of hypervisors to query information
514

515
  @rtype: list
516
  @return: a list of all running instances on the current node
517
    - instance1.example.com
518
    - instance2.example.com
519

520
  """
521
  results = []
522
  for hname in hypervisor_list:
523
    try:
524
      names = hypervisor.GetHypervisor(hname).ListInstances()
525
      results.extend(names)
526
    except errors.HypervisorError, err:
527
      logging.exception("Error enumerating instances for hypevisor %s", hname)
528
      # FIXME: should we somehow not propagate this to the master?
529
      raise
530

    
531
  return results
532

    
533

    
534
def GetInstanceInfo(instance, hname):
535
  """Gives back the informations about an instance as a dictionary.
536

537
  @type instance: string
538
  @param instance: the instance name
539
  @type hname: string
540
  @param hname: the hypervisor type of the instance
541

542
  @rtype: dict
543
  @return: dictionary with the following keys:
544
      - memory: memory size of instance (int)
545
      - state: xen state of instance (string)
546
      - time: cpu time of instance (float)
547

548
  """
549
  output = {}
550

    
551
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
552
  if iinfo is not None:
553
    output['memory'] = iinfo[2]
554
    output['state'] = iinfo[4]
555
    output['time'] = iinfo[5]
556

    
557
  return output
558

    
559

    
560
def GetAllInstancesInfo(hypervisor_list):
561
  """Gather data about all instances.
562

563
  This is the equivalent of L{GetInstanceInfo}, except that it
564
  computes data for all instances at once, thus being faster if one
565
  needs data about more than one instance.
566

567
  @type hypervisor_list: list
568
  @param hypervisor_list: list of hypervisors to query for instance data
569

570
  @rtype: dict
571
  @return: dictionary of instance: data, with data having the following keys:
572
      - memory: memory size of instance (int)
573
      - state: xen state of instance (string)
574
      - time: cpu time of instance (float)
575
      - vcpus: the number of vcpus
576

577
  """
578
  output = {}
579

    
580
  for hname in hypervisor_list:
581
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
582
    if iinfo:
583
      for name, inst_id, memory, vcpus, state, times in iinfo:
584
        value = {
585
          'memory': memory,
586
          'vcpus': vcpus,
587
          'state': state,
588
          'time': times,
589
          }
590
        if name in output and output[name] != value:
591
          raise errors.HypervisorError("Instance %s running duplicate"
592
                                       " with different parameters" % name)
593
        output[name] = value
594

    
595
  return output
596

    
597

    
598
def AddOSToInstance(instance):
599
  """Add an OS to an instance.
600

601
  @type instance: L{objects.Instance}
602
  @param instance: Instance whose OS is to be installed
603
  @rtype: boolean
604
  @return: the success of the operation
605

606
  """
607
  inst_os = OSFromDisk(instance.os)
608

    
609
  create_env = OSEnvironment(instance)
610

    
611
  logfile = "%s/add-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
612
                                     instance.name, int(time.time()))
613

    
614
  result = utils.RunCmd([inst_os.create_script], env=create_env,
615
                        cwd=inst_os.path, output=logfile,)
616
  if result.failed:
617
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
618
                  " output: %s", result.cmd, result.fail_reason, logfile,
619
                  result.output)
620
    return False
621

    
622
  return True
623

    
624

    
625
def RunRenameInstance(instance, old_name):
626
  """Run the OS rename script for an instance.
627

628
  @type instance: L{objects.Instance}
629
  @param instance: Instance whose OS is to be installed
630
  @type old_name: string
631
  @param old_name: previous instance name
632
  @rtype: boolean
633
  @return: the success of the operation
634

635
  """
636
  inst_os = OSFromDisk(instance.os)
637

    
638
  rename_env = OSEnvironment(instance)
639
  rename_env['OLD_INSTANCE_NAME'] = old_name
640

    
641
  logfile = "%s/rename-%s-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
642
                                           old_name,
643
                                           instance.name, int(time.time()))
644

    
645
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
646
                        cwd=inst_os.path, output=logfile)
647

    
648
  if result.failed:
649
    logging.error("os create command '%s' returned error: %s output: %s",
650
                  result.cmd, result.fail_reason, result.output)
651
    return False
652

    
653
  return True
654

    
655

    
656
def _GetVGInfo(vg_name):
657
  """Get informations about the volume group.
658

659
  @type vg_name: str
660
  @param vg_name: the volume group which we query
661
  @rtype: dict
662
  @return:
663
    A dictionary with the following keys:
664
      - C{vg_size} is the total size of the volume group in MiB
665
      - C{vg_free} is the free size of the volume group in MiB
666
      - C{pv_count} are the number of physical disks in that VG
667

668
    If an error occurs during gathering of data, we return the same dict
669
    with keys all set to None.
670

671
  """
672
  retdic = dict.fromkeys(["vg_size", "vg_free", "pv_count"])
673

    
674
  retval = utils.RunCmd(["vgs", "-ovg_size,vg_free,pv_count", "--noheadings",
675
                         "--nosuffix", "--units=m", "--separator=:", vg_name])
676

    
677
  if retval.failed:
678
    logging.error("volume group %s not present", vg_name)
679
    return retdic
680
  valarr = retval.stdout.strip().rstrip(':').split(':')
681
  if len(valarr) == 3:
682
    try:
683
      retdic = {
684
        "vg_size": int(round(float(valarr[0]), 0)),
685
        "vg_free": int(round(float(valarr[1]), 0)),
686
        "pv_count": int(valarr[2]),
687
        }
688
    except ValueError, err:
689
      logging.exception("Fail to parse vgs output")
690
  else:
691
    logging.error("vgs output has the wrong number of fields (expected"
692
                  " three): %s", str(valarr))
693
  return retdic
694

    
695

    
696
def _GatherBlockDevs(instance):
697
  """Set up an instance's block device(s).
698

699
  This is run on the primary node at instance startup. The block
700
  devices must be already assembled.
701

702
  @type instance: L{objects.Instance}
703
  @param instance: the instance whose disks we shoul assemble
704
  @rtype: list of L{bdev.BlockDev}
705
  @return: list of the block devices
706

707
  """
708
  block_devices = []
709
  for disk in instance.disks:
710
    device = _RecursiveFindBD(disk)
711
    if device is None:
712
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
713
                                    str(disk))
714
    device.Open()
715
    block_devices.append((disk, device))
716
  return block_devices
717

    
718

    
719
def StartInstance(instance, extra_args):
720
  """Start an instance.
721

722
  @type instance: L{objects.Instance}
723
  @param instance: the instance object
724
  @rtype: boolean
725
  @return: whether the startup was successful or not
726

727
  """
728
  running_instances = GetInstanceList([instance.hypervisor])
729

    
730
  if instance.name in running_instances:
731
    return True
732

    
733
  block_devices = _GatherBlockDevs(instance)
734
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
735

    
736
  try:
737
    hyper.StartInstance(instance, block_devices, extra_args)
738
  except errors.HypervisorError, err:
739
    logging.exception("Failed to start instance")
740
    return False
741

    
742
  return True
743

    
744

    
745
def ShutdownInstance(instance):
746
  """Shut an instance down.
747

748
  @note: this functions uses polling with a hardcoded timeout.
749

750
  @type instance: L{objects.Instance}
751
  @param instance: the instance object
752
  @rtype: boolean
753
  @return: whether the startup was successful or not
754

755
  """
756
  hv_name = instance.hypervisor
757
  running_instances = GetInstanceList([hv_name])
758

    
759
  if instance.name not in running_instances:
760
    return True
761

    
762
  hyper = hypervisor.GetHypervisor(hv_name)
763
  try:
764
    hyper.StopInstance(instance)
765
  except errors.HypervisorError, err:
766
    logging.error("Failed to stop instance")
767
    return False
768

    
769
  # test every 10secs for 2min
770
  shutdown_ok = False
771

    
772
  time.sleep(1)
773
  for dummy in range(11):
774
    if instance.name not in GetInstanceList([hv_name]):
775
      break
776
    time.sleep(10)
777
  else:
778
    # the shutdown did not succeed
779
    logging.error("shutdown of '%s' unsuccessful, using destroy", instance)
780

    
781
    try:
782
      hyper.StopInstance(instance, force=True)
783
    except errors.HypervisorError, err:
784
      logging.exception("Failed to stop instance")
785
      return False
786

    
787
    time.sleep(1)
788
    if instance.name in GetInstanceList([hv_name]):
789
      logging.error("could not shutdown instance '%s' even by destroy",
790
                    instance.name)
791
      return False
792

    
793
  return True
794

    
795

    
796
def RebootInstance(instance, reboot_type, extra_args):
797
  """Reboot an instance.
798

799
  @type instance: L{objects.Instance}
800
  @param instance: the instance object to reboot
801
  @type reboot_type: str
802
  @param reboot_type: the type of reboot, one the following
803
    constants:
804
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
805
        instance OS, do not recreate the VM
806
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
807
        restart the VM (at the hypervisor level)
808
      - the other reboot type (L{constants.INSTANCE_REBOOT_HARD})
809
        is not accepted here, since that mode is handled
810
        differently
811
  @rtype: boolean
812
  @return: the success of the operation
813

814
  """
815
  running_instances = GetInstanceList([instance.hypervisor])
816

    
817
  if instance.name not in running_instances:
818
    logging.error("Cannot reboot instance that is not running")
819
    return False
820

    
821
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
822
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
823
    try:
824
      hyper.RebootInstance(instance)
825
    except errors.HypervisorError, err:
826
      logging.exception("Failed to soft reboot instance")
827
      return False
828
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
829
    try:
830
      ShutdownInstance(instance)
831
      StartInstance(instance, extra_args)
832
    except errors.HypervisorError, err:
833
      logging.exception("Failed to hard reboot instance")
834
      return False
835
  else:
836
    raise errors.ParameterError("reboot_type invalid")
837

    
838
  return True
839

    
840

    
841
def MigrateInstance(instance, target, live):
842
  """Migrates an instance to another node.
843

844
  @type instance: L{objects.Instance}
845
  @param instance: the instance definition
846
  @type target: string
847
  @param target: the target node name
848
  @type live: boolean
849
  @param live: whether the migration should be done live or not (the
850
      interpretation of this parameter is left to the hypervisor)
851
  @rtype: tuple
852
  @return: a tuple of (success, msg) where:
853
      - succes is a boolean denoting the success/failure of the operation
854
      - msg is a string with details in case of failure
855

856
  """
857
  hyper = hypervisor.GetHypervisor(instance.hypervisor_name)
858

    
859
  try:
860
    hyper.MigrateInstance(instance.name, target, live)
861
  except errors.HypervisorError, err:
862
    msg = "Failed to migrate instance: %s" % str(err)
863
    logging.error(msg)
864
    return (False, msg)
865
  return (True, "Migration successfull")
866

    
867

    
868
def CreateBlockDevice(disk, size, owner, on_primary, info):
869
  """Creates a block device for an instance.
870

871
  @type disk: L{objects.Disk}
872
  @param disk: the object describing the disk we should create
873
  @type size: int
874
  @param size: the size of the physical underlying device, in MiB
875
  @type owner: str
876
  @param owner: the name of the instance for which disk is created,
877
      used for device cache data
878
  @type on_primary: boolean
879
  @param on_primary:  indicates if it is the primary node or not
880
  @type info: string
881
  @param info: string that will be sent to the physical device
882
      creation, used for example to set (LVM) tags on LVs
883

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

888
  """
889
  clist = []
890
  if disk.children:
891
    for child in disk.children:
892
      crdev = _RecursiveAssembleBD(child, owner, on_primary)
893
      if on_primary or disk.AssembleOnSecondary():
894
        # we need the children open in case the device itself has to
895
        # be assembled
896
        crdev.Open()
897
      clist.append(crdev)
898
  try:
899
    device = bdev.FindDevice(disk.dev_type, disk.physical_id, clist)
900
    if device is not None:
901
      logging.info("removing existing device %s", disk)
902
      device.Remove()
903
  except errors.BlockDeviceError, err:
904
    pass
905

    
906
  device = bdev.Create(disk.dev_type, disk.physical_id,
907
                       clist, size)
908
  if device is None:
909
    raise ValueError("Can't create child device for %s, %s" %
910
                     (disk, size))
911
  if on_primary or disk.AssembleOnSecondary():
912
    if not device.Assemble():
913
      errorstring = "Can't assemble device after creation"
914
      logging.error(errorstring)
915
      raise errors.BlockDeviceError("%s, very unusual event - check the node"
916
                                    " daemon logs" % errorstring)
917
    device.SetSyncSpeed(constants.SYNC_SPEED)
918
    if on_primary or disk.OpenOnSecondary():
919
      device.Open(force=True)
920
    DevCacheManager.UpdateCache(device.dev_path, owner,
921
                                on_primary, disk.iv_name)
922

    
923
  device.SetInfo(info)
924

    
925
  physical_id = device.unique_id
926
  return physical_id
927

    
928

    
929
def RemoveBlockDevice(disk):
930
  """Remove a block device.
931

932
  @note: This is intended to be called recursively.
933

934
  @type disk: L{objects.disk}
935
  @param disk: the disk object we should remove
936
  @rtype: boolean
937
  @return: the success of the operation
938

939
  """
940
  try:
941
    # since we are removing the device, allow a partial match
942
    # this allows removal of broken mirrors
943
    rdev = _RecursiveFindBD(disk, allow_partial=True)
944
  except errors.BlockDeviceError, err:
945
    # probably can't attach
946
    logging.info("Can't attach to device %s in remove", disk)
947
    rdev = None
948
  if rdev is not None:
949
    r_path = rdev.dev_path
950
    result = rdev.Remove()
951
    if result:
952
      DevCacheManager.RemoveCache(r_path)
953
  else:
954
    result = True
955
  if disk.children:
956
    for child in disk.children:
957
      result = result and RemoveBlockDevice(child)
958
  return result
959

    
960

    
961
def _RecursiveAssembleBD(disk, owner, as_primary):
962
  """Activate a block device for an instance.
963

964
  This is run on the primary and secondary nodes for an instance.
965

966
  @note: this function is called recursively.
967

968
  @type disk: L{objects.Disk}
969
  @param disk: the disk we try to assemble
970
  @type owner: str
971
  @param owner: the name of the instance which owns the disk
972
  @type as_primary: boolean
973
  @param as_primary: if we should make the block device
974
      read/write
975

976
  @return: the assembled device or None (in case no device
977
      was assembled)
978
  @raise errors.BlockDeviceError: in case there is an error
979
      during the activation of the children or the device
980
      itself
981

982
  """
983
  children = []
984
  if disk.children:
985
    mcn = disk.ChildrenNeeded()
986
    if mcn == -1:
987
      mcn = 0 # max number of Nones allowed
988
    else:
989
      mcn = len(disk.children) - mcn # max number of Nones
990
    for chld_disk in disk.children:
991
      try:
992
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
993
      except errors.BlockDeviceError, err:
994
        if children.count(None) >= mcn:
995
          raise
996
        cdev = None
997
        logging.debug("Error in child activation: %s", str(err))
998
      children.append(cdev)
999

    
1000
  if as_primary or disk.AssembleOnSecondary():
1001
    r_dev = bdev.AttachOrAssemble(disk.dev_type, disk.physical_id, children)
1002
    r_dev.SetSyncSpeed(constants.SYNC_SPEED)
1003
    result = r_dev
1004
    if as_primary or disk.OpenOnSecondary():
1005
      r_dev.Open()
1006
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1007
                                as_primary, disk.iv_name)
1008

    
1009
  else:
1010
    result = True
1011
  return result
1012

    
1013

    
1014
def AssembleBlockDevice(disk, owner, as_primary):
1015
  """Activate a block device for an instance.
1016

1017
  This is a wrapper over _RecursiveAssembleBD.
1018

1019
  @rtype: str or boolean
1020
  @return: a C{/dev/...} path for primary nodes, and
1021
      C{True} for secondary nodes
1022

1023
  """
1024
  result = _RecursiveAssembleBD(disk, owner, as_primary)
1025
  if isinstance(result, bdev.BlockDev):
1026
    result = result.dev_path
1027
  return result
1028

    
1029

    
1030
def ShutdownBlockDevice(disk):
1031
  """Shut down a block device.
1032

1033
  First, if the device is assembled (can L{Attach()}), then the device
1034
  is shutdown. Then the children of the device are shutdown.
1035

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

1040
  @type disk: L{objects.Disk}
1041
  @param disk: the description of the disk we should
1042
      shutdown
1043
  @rtype: boolean
1044
  @return: the success of the operation
1045

1046
  """
1047
  r_dev = _RecursiveFindBD(disk)
1048
  if r_dev is not None:
1049
    r_path = r_dev.dev_path
1050
    result = r_dev.Shutdown()
1051
    if result:
1052
      DevCacheManager.RemoveCache(r_path)
1053
  else:
1054
    result = True
1055
  if disk.children:
1056
    for child in disk.children:
1057
      result = result and ShutdownBlockDevice(child)
1058
  return result
1059

    
1060

    
1061
def MirrorAddChildren(parent_cdev, new_cdevs):
1062
  """Extend a mirrored block device.
1063

1064
  @type parent_cdev: L{objects.Disk}
1065
  @param parent_cdev: the disk to which we should add children
1066
  @type new_cdevs: list of L{objects.Disk}
1067
  @param new_cdevs: the list of children which we should add
1068
  @rtype: boolean
1069
  @return: the success of the operation
1070

1071
  """
1072
  parent_bdev = _RecursiveFindBD(parent_cdev, allow_partial=True)
1073
  if parent_bdev is None:
1074
    logging.error("Can't find parent device")
1075
    return False
1076
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1077
  if new_bdevs.count(None) > 0:
1078
    logging.error("Can't find new device(s) to add: %s:%s",
1079
                  new_bdevs, new_cdevs)
1080
    return False
1081
  parent_bdev.AddChildren(new_bdevs)
1082
  return True
1083

    
1084

    
1085
def MirrorRemoveChildren(parent_cdev, new_cdevs):
1086
  """Shrink a mirrored block device.
1087

1088
  @type parent_cdev: L{objects.Disk}
1089
  @param parent_cdev: the disk from which we should remove children
1090
  @type new_cdevs: list of L{objects.Disk}
1091
  @param new_cdevs: the list of children which we should remove
1092
  @rtype: boolean
1093
  @return: the success of the operation
1094

1095
  """
1096
  parent_bdev = _RecursiveFindBD(parent_cdev)
1097
  if parent_bdev is None:
1098
    logging.error("Can't find parent in remove children: %s", parent_cdev)
1099
    return False
1100
  devs = []
1101
  for disk in new_cdevs:
1102
    rpath = disk.StaticDevPath()
1103
    if rpath is None:
1104
      bd = _RecursiveFindBD(disk)
1105
      if bd is None:
1106
        logging.error("Can't find dynamic device %s while removing children",
1107
                      disk)
1108
        return False
1109
      else:
1110
        devs.append(bd.dev_path)
1111
    else:
1112
      devs.append(rpath)
1113
  parent_bdev.RemoveChildren(devs)
1114
  return True
1115

    
1116

    
1117
def GetMirrorStatus(disks):
1118
  """Get the mirroring status of a list of devices.
1119

1120
  @type disks: list of L{objects.Disk}
1121
  @param disks: the list of disks which we should query
1122
  @rtype: disk
1123
  @return:
1124
      a list of (mirror_done, estimated_time) tuples, which
1125
      are the result of L{bdev.BlockDevice.CombinedSyncStatus}
1126
  @raise errors.BlockDeviceError: if any of the disks cannot be
1127
      found
1128

1129
  """
1130
  stats = []
1131
  for dsk in disks:
1132
    rbd = _RecursiveFindBD(dsk)
1133
    if rbd is None:
1134
      raise errors.BlockDeviceError("Can't find device %s" % str(dsk))
1135
    stats.append(rbd.CombinedSyncStatus())
1136
  return stats
1137

    
1138

    
1139
def _RecursiveFindBD(disk, allow_partial=False):
1140
  """Check if a device is activated.
1141

1142
  If so, return informations about the real device.
1143

1144
  @type disk: L{objects.Disk}
1145
  @param disk: the disk object we need to find
1146
  @type allow_partial: boolean
1147
  @param allow_partial: if true, don't abort the find if a
1148
      child of the device can't be found; this is intended
1149
      to be used when repairing mirrors
1150

1151
  @return: None if the device can't be found,
1152
      otherwise the device instance
1153

1154
  """
1155
  children = []
1156
  if disk.children:
1157
    for chdisk in disk.children:
1158
      children.append(_RecursiveFindBD(chdisk))
1159

    
1160
  return bdev.FindDevice(disk.dev_type, disk.physical_id, children)
1161

    
1162

    
1163
def FindBlockDevice(disk):
1164
  """Check if a device is activated.
1165

1166
  If it is, return informations about the real device.
1167

1168
  @type disk: L{objects.Disk}
1169
  @param disk: the disk to find
1170
  @rtype: None or tuple
1171
  @return: None if the disk cannot be found, otherwise a
1172
      tuple (device_path, major, minor, sync_percent,
1173
      estimated_time, is_degraded)
1174

1175
  """
1176
  rbd = _RecursiveFindBD(disk)
1177
  if rbd is None:
1178
    return rbd
1179
  return (rbd.dev_path, rbd.major, rbd.minor) + rbd.GetSyncStatus()
1180

    
1181

    
1182
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
1183
  """Write a file to the filesystem.
1184

1185
  This allows the master to overwrite(!) a file. It will only perform
1186
  the operation if the file belongs to a list of configuration files.
1187

1188
  @type file_name: str
1189
  @param file_name: the target file name
1190
  @type data: str
1191
  @param data: the new contents of the file
1192
  @type mode: int
1193
  @param mode: the mode to give the file (can be None)
1194
  @type uid: int
1195
  @param uid: the owner of the file (can be -1 for default)
1196
  @type gid: int
1197
  @param gid: the group of the file (can be -1 for default)
1198
  @type atime: float
1199
  @param atime: the atime to set on the file (can be None)
1200
  @type mtime: float
1201
  @param mtime: the mtime to set on the file (can be None)
1202
  @rtype: boolean
1203
  @return: the success of the operation; errors are logged
1204
      in the node daemon log
1205

1206
  """
1207
  if not os.path.isabs(file_name):
1208
    logging.error("Filename passed to UploadFile is not absolute: '%s'",
1209
                  file_name)
1210
    return False
1211

    
1212
  allowed_files = [
1213
    constants.CLUSTER_CONF_FILE,
1214
    constants.ETC_HOSTS,
1215
    constants.SSH_KNOWN_HOSTS_FILE,
1216
    constants.VNC_PASSWORD_FILE,
1217
    ]
1218

    
1219
  if file_name not in allowed_files:
1220
    logging.error("Filename passed to UploadFile not in allowed"
1221
                 " upload targets: '%s'", file_name)
1222
    return False
1223

    
1224
  utils.WriteFile(file_name, data=data, mode=mode, uid=uid, gid=gid,
1225
                  atime=atime, mtime=mtime)
1226
  return True
1227

    
1228

    
1229
def WriteSsconfFiles(values):
1230
  ssconf.WriteSsconfFiles(values)
1231

    
1232

    
1233
def _ErrnoOrStr(err):
1234
  """Format an EnvironmentError exception.
1235

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

1240
  @type err: L{EnvironmentError}
1241
  @param err: the exception to format
1242

1243
  """
1244
  if hasattr(err, 'errno'):
1245
    detail = errno.errorcode[err.errno]
1246
  else:
1247
    detail = str(err)
1248
  return detail
1249

    
1250

    
1251
def _OSOndiskVersion(name, os_dir):
1252
  """Compute and return the API version of a given OS.
1253

1254
  This function will try to read the API version of the OS given by
1255
  the 'name' parameter and residing in the 'os_dir' directory.
1256

1257
  @type name: str
1258
  @param name: the OS name we should look for
1259
  @type os_dir: str
1260
  @param os_dir: the directory inwhich we should look for the OS
1261
  @rtype: int or None
1262
  @return:
1263
      Either an integer denoting the version or None in the
1264
      case when this is not a valid OS name.
1265
  @raise errors.InvalidOS: if the OS cannot be found
1266

1267
  """
1268
  api_file = os.path.sep.join([os_dir, "ganeti_api_version"])
1269

    
1270
  try:
1271
    st = os.stat(api_file)
1272
  except EnvironmentError, err:
1273
    raise errors.InvalidOS(name, os_dir, "'ganeti_api_version' file not"
1274
                           " found (%s)" % _ErrnoOrStr(err))
1275

    
1276
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1277
    raise errors.InvalidOS(name, os_dir, "'ganeti_api_version' file is not"
1278
                           " a regular file")
1279

    
1280
  try:
1281
    f = open(api_file)
1282
    try:
1283
      api_versions = f.readlines()
1284
    finally:
1285
      f.close()
1286
  except EnvironmentError, err:
1287
    raise errors.InvalidOS(name, os_dir, "error while reading the"
1288
                           " API version (%s)" % _ErrnoOrStr(err))
1289

    
1290
  api_versions = [version.strip() for version in api_versions]
1291
  try:
1292
    api_versions = [int(version) for version in api_versions]
1293
  except (TypeError, ValueError), err:
1294
    raise errors.InvalidOS(name, os_dir,
1295
                           "API version is not integer (%s)" % str(err))
1296

    
1297
  return api_versions
1298

    
1299

    
1300
def DiagnoseOS(top_dirs=None):
1301
  """Compute the validity for all OSes.
1302

1303
  @type top_dirs: list
1304
  @param top_dirs: the list of directories in which to
1305
      search (if not given defaults to
1306
      L{constants.OS_SEARCH_PATH})
1307
  @rtype: list of L{objects.OS}
1308
  @return: an OS object for each name in all the given
1309
      directories
1310

1311
  """
1312
  if top_dirs is None:
1313
    top_dirs = constants.OS_SEARCH_PATH
1314

    
1315
  result = []
1316
  for dir_name in top_dirs:
1317
    if os.path.isdir(dir_name):
1318
      try:
1319
        f_names = utils.ListVisibleFiles(dir_name)
1320
      except EnvironmentError, err:
1321
        logging.exception("Can't list the OS directory %s", dir_name)
1322
        break
1323
      for name in f_names:
1324
        try:
1325
          os_inst = OSFromDisk(name, base_dir=dir_name)
1326
          result.append(os_inst)
1327
        except errors.InvalidOS, err:
1328
          result.append(objects.OS.FromInvalidOS(err))
1329

    
1330
  return result
1331

    
1332

    
1333
def OSFromDisk(name, base_dir=None):
1334
  """Create an OS instance from disk.
1335

1336
  This function will return an OS instance if the given name is a
1337
  valid OS name. Otherwise, it will raise an appropriate
1338
  L{errors.InvalidOS} exception, detailing why this is not a valid OS.
1339

1340
  @type base_dir: string
1341
  @keyword base_dir: Base directory containing OS installations.
1342
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
1343
  @rtype: L{objects.OS}
1344
  @return: the OS instance if we find a valid one
1345
  @raise errors.InvalidOS: if we don't find a valid OS
1346

1347
  """
1348
  if base_dir is None:
1349
    os_dir = utils.FindFile(name, constants.OS_SEARCH_PATH, os.path.isdir)
1350
    if os_dir is None:
1351
      raise errors.InvalidOS(name, None, "OS dir not found in search path")
1352
  else:
1353
    os_dir = os.path.sep.join([base_dir, name])
1354

    
1355
  api_versions = _OSOndiskVersion(name, os_dir)
1356

    
1357
  if constants.OS_API_VERSION not in api_versions:
1358
    raise errors.InvalidOS(name, os_dir, "API version mismatch"
1359
                           " (found %s want %s)"
1360
                           % (api_versions, constants.OS_API_VERSION))
1361

    
1362
  # OS Scripts dictionary, we will populate it with the actual script names
1363
  os_scripts = dict.fromkeys(constants.OS_SCRIPTS)
1364

    
1365
  for script in os_scripts:
1366
    os_scripts[script] = os.path.sep.join([os_dir, script])
1367

    
1368
    try:
1369
      st = os.stat(os_scripts[script])
1370
    except EnvironmentError, err:
1371
      raise errors.InvalidOS(name, os_dir, "'%s' script missing (%s)" %
1372
                             (script, _ErrnoOrStr(err)))
1373

    
1374
    if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
1375
      raise errors.InvalidOS(name, os_dir, "'%s' script not executable" %
1376
                             script)
1377

    
1378
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1379
      raise errors.InvalidOS(name, os_dir, "'%s' is not a regular file" %
1380
                             script)
1381

    
1382

    
1383
  return objects.OS(name=name, path=os_dir, status=constants.OS_VALID_STATUS,
1384
                    create_script=os_scripts[constants.OS_SCRIPT_CREATE],
1385
                    export_script=os_scripts[constants.OS_SCRIPT_EXPORT],
1386
                    import_script=os_scripts[constants.OS_SCRIPT_IMPORT],
1387
                    rename_script=os_scripts[constants.OS_SCRIPT_RENAME],
1388
                    api_versions=api_versions)
1389

    
1390
def OSEnvironment(instance, debug=0):
1391
  """Calculate the environment for an os script.
1392

1393
  @type instance: L{objects.Instance}
1394
  @param instance: target instance for the os script run
1395
  @type debug: integer
1396
  @param debug: debug level (0 or 1, for OS Api 10)
1397
  @rtype: dict
1398
  @return: dict of environment variables
1399
  @raise errors.BlockDeviceError: if the block device
1400
      cannot be found
1401

1402
  """
1403
  result = {}
1404
  result['OS_API_VERSION'] = '%d' % constants.OS_API_VERSION
1405
  result['INSTANCE_NAME'] = instance.name
1406
  result['HYPERVISOR'] = instance.hypervisor
1407
  result['DISK_COUNT'] = '%d' % len(instance.disks)
1408
  result['NIC_COUNT'] = '%d' % len(instance.nics)
1409
  result['DEBUG_LEVEL'] = '%d' % debug
1410
  for idx, disk in enumerate(instance.disks):
1411
    real_disk = _RecursiveFindBD(disk)
1412
    if real_disk is None:
1413
      raise errors.BlockDeviceError("Block device '%s' is not set up" %
1414
                                    str(disk))
1415
    real_disk.Open()
1416
    result['DISK_%d_PATH' % idx] = real_disk.dev_path
1417
    # FIXME: When disks will have read-only mode, populate this
1418
    result['DISK_%d_ACCESS' % idx] = 'W'
1419
    if constants.HV_DISK_TYPE in instance.hvparams:
1420
      result['DISK_%d_FRONTEND_TYPE' % idx] = \
1421
        instance.hvparams[constants.HV_DISK_TYPE]
1422
    if disk.dev_type in constants.LDS_BLOCK:
1423
      result['DISK_%d_BACKEND_TYPE' % idx] = 'block'
1424
    elif disk.dev_type == constants.LD_FILE:
1425
      result['DISK_%d_BACKEND_TYPE' % idx] = \
1426
        'file:%s' % disk.physical_id[0]
1427
  for idx, nic in enumerate(instance.nics):
1428
    result['NIC_%d_MAC' % idx] = nic.mac
1429
    if nic.ip:
1430
      result['NIC_%d_IP' % idx] = nic.ip
1431
    result['NIC_%d_BRIDGE' % idx] = nic.bridge
1432
    if constants.HV_NIC_TYPE in instance.hvparams:
1433
      result['NIC_%d_FRONTEND_TYPE' % idx] = \
1434
        instance.hvparams[constants.HV_NIC_TYPE]
1435

    
1436
  return result
1437

    
1438
def GrowBlockDevice(disk, amount):
1439
  """Grow a stack of block devices.
1440

1441
  This function is called recursively, with the childrens being the
1442
  first ones to resize.
1443

1444
  @type disk: L{objects.Disk}
1445
  @param disk: the disk to be grown
1446
  @rtype: (status, result)
1447
  @return: a tuple with the status of the operation
1448
      (True/False), and the errors message if status
1449
      is False
1450

1451
  """
1452
  r_dev = _RecursiveFindBD(disk)
1453
  if r_dev is None:
1454
    return False, "Cannot find block device %s" % (disk,)
1455

    
1456
  try:
1457
    r_dev.Grow(amount)
1458
  except errors.BlockDeviceError, err:
1459
    return False, str(err)
1460

    
1461
  return True, None
1462

    
1463

    
1464
def SnapshotBlockDevice(disk):
1465
  """Create a snapshot copy of a block device.
1466

1467
  This function is called recursively, and the snapshot is actually created
1468
  just for the leaf lvm backend device.
1469

1470
  @type disk: L{objects.Disk}
1471
  @param disk: the disk to be snapshotted
1472
  @rtype: string
1473
  @return: snapshot disk path
1474

1475
  """
1476
  if disk.children:
1477
    if len(disk.children) == 1:
1478
      # only one child, let's recurse on it
1479
      return SnapshotBlockDevice(disk.children[0])
1480
    else:
1481
      # more than one child, choose one that matches
1482
      for child in disk.children:
1483
        if child.size == disk.size:
1484
          # return implies breaking the loop
1485
          return SnapshotBlockDevice(child)
1486
  elif disk.dev_type == constants.LD_LV:
1487
    r_dev = _RecursiveFindBD(disk)
1488
    if r_dev is not None:
1489
      # let's stay on the safe side and ask for the full size, for now
1490
      return r_dev.Snapshot(disk.size)
1491
    else:
1492
      return None
1493
  else:
1494
    raise errors.ProgrammerError("Cannot snapshot non-lvm block device"
1495
                                 " '%s' of type '%s'" %
1496
                                 (disk.unique_id, disk.dev_type))
1497

    
1498

    
1499
def ExportSnapshot(disk, dest_node, instance, cluster_name, idx):
1500
  """Export a block device snapshot to a remote node.
1501

1502
  @type disk: L{objects.Disk}
1503
  @param disk: the description of the disk to export
1504
  @type dest_node: str
1505
  @param dest_node: the destination node to export to
1506
  @type instance: L{objects.Instance}
1507
  @param instance: the instance object to whom the disk belongs
1508
  @type cluster_name: str
1509
  @param cluster_name: the cluster name, needed for SSH hostalias
1510
  @type idx: int
1511
  @param idx: the index of the disk in the instance's disk list,
1512
      used to export to the OS scripts environment
1513
  @rtype: boolean
1514
  @return: the success of the operation
1515

1516
  """
1517
  export_env = OSEnvironment(instance)
1518

    
1519
  inst_os = OSFromDisk(instance.os)
1520
  export_script = inst_os.export_script
1521

    
1522
  logfile = "%s/exp-%s-%s-%s.log" % (constants.LOG_OS_DIR, inst_os.name,
1523
                                     instance.name, int(time.time()))
1524
  if not os.path.exists(constants.LOG_OS_DIR):
1525
    os.mkdir(constants.LOG_OS_DIR, 0750)
1526
  real_disk = _RecursiveFindBD(disk)
1527
  if real_disk is None:
1528
    raise errors.BlockDeviceError("Block device '%s' is not set up" %
1529
                                  str(disk))
1530
  real_disk.Open()
1531

    
1532
  export_env['EXPORT_DEVICE'] = real_disk.dev_path
1533
  export_env['EXPORT_INDEX'] = str(idx)
1534

    
1535
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
1536
  destfile = disk.physical_id[1]
1537

    
1538
  # the target command is built out of three individual commands,
1539
  # which are joined by pipes; we check each individual command for
1540
  # valid parameters
1541
  expcmd = utils.BuildShellCmd("cd %s; %s 2>%s", inst_os.path,
1542
                               export_script, logfile)
1543

    
1544
  comprcmd = "gzip"
1545

    
1546
  destcmd = utils.BuildShellCmd("mkdir -p %s && cat > %s/%s",
1547
                                destdir, destdir, destfile)
1548
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
1549
                                                   constants.GANETI_RUNAS,
1550
                                                   destcmd)
1551

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

    
1555
  result = utils.RunCmd(command, env=export_env)
1556

    
1557
  if result.failed:
1558
    logging.error("os snapshot export command '%s' returned error: %s"
1559
                  " output: %s", command, result.fail_reason, result.output)
1560
    return False
1561

    
1562
  return True
1563

    
1564

    
1565
def FinalizeExport(instance, snap_disks):
1566
  """Write out the export configuration information.
1567

1568
  @type instance: L{objects.Instance}
1569
  @param instance: the instance which we export, used for
1570
      saving configuration
1571
  @type snap_disks: list of L{objects.Disk}
1572
  @param snap_disks: list of snapshot block devices, which
1573
      will be used to get the actual name of the dump file
1574

1575
  @rtype: boolean
1576
  @return: the success of the operation
1577

1578
  """
1579
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
1580
  finaldestdir = os.path.join(constants.EXPORT_DIR, instance.name)
1581

    
1582
  config = objects.SerializableConfigParser()
1583

    
1584
  config.add_section(constants.INISECT_EXP)
1585
  config.set(constants.INISECT_EXP, 'version', '0')
1586
  config.set(constants.INISECT_EXP, 'timestamp', '%d' % int(time.time()))
1587
  config.set(constants.INISECT_EXP, 'source', instance.primary_node)
1588
  config.set(constants.INISECT_EXP, 'os', instance.os)
1589
  config.set(constants.INISECT_EXP, 'compression', 'gzip')
1590

    
1591
  config.add_section(constants.INISECT_INS)
1592
  config.set(constants.INISECT_INS, 'name', instance.name)
1593
  config.set(constants.INISECT_INS, 'memory', '%d' %
1594
             instance.beparams[constants.BE_MEMORY])
1595
  config.set(constants.INISECT_INS, 'vcpus', '%d' %
1596
             instance.beparams[constants.BE_VCPUS])
1597
  config.set(constants.INISECT_INS, 'disk_template', instance.disk_template)
1598

    
1599
  nic_count = 0
1600
  for nic_count, nic in enumerate(instance.nics):
1601
    config.set(constants.INISECT_INS, 'nic%d_mac' %
1602
               nic_count, '%s' % nic.mac)
1603
    config.set(constants.INISECT_INS, 'nic%d_ip' % nic_count, '%s' % nic.ip)
1604
    config.set(constants.INISECT_INS, 'nic%d_bridge' % nic_count,
1605
               '%s' % nic.bridge)
1606
  # TODO: redundant: on load can read nics until it doesn't exist
1607
  config.set(constants.INISECT_INS, 'nic_count' , '%d' % nic_count)
1608

    
1609
  disk_total = 0
1610
  for disk_count, disk in enumerate(snap_disks):
1611
    if disk:
1612
      disk_total += 1
1613
      config.set(constants.INISECT_INS, 'disk%d_ivname' % disk_count,
1614
                 ('%s' % disk.iv_name))
1615
      config.set(constants.INISECT_INS, 'disk%d_dump' % disk_count,
1616
                 ('%s' % disk.physical_id[1]))
1617
      config.set(constants.INISECT_INS, 'disk%d_size' % disk_count,
1618
                 ('%d' % disk.size))
1619

    
1620
  config.set(constants.INISECT_INS, 'disk_count' , '%d' % disk_total)
1621

    
1622
  utils.WriteFile(os.path.join(destdir, constants.EXPORT_CONF_FILE),
1623
                  data=config.Dumps())
1624
  shutil.rmtree(finaldestdir, True)
1625
  shutil.move(destdir, finaldestdir)
1626

    
1627
  return True
1628

    
1629

    
1630
def ExportInfo(dest):
1631
  """Get export configuration information.
1632

1633
  @type dest: str
1634
  @param dest: directory containing the export
1635

1636
  @rtype: L{objects.SerializableConfigParser}
1637
  @return: a serializable config file containing the
1638
      export info
1639

1640
  """
1641
  cff = os.path.join(dest, constants.EXPORT_CONF_FILE)
1642

    
1643
  config = objects.SerializableConfigParser()
1644
  config.read(cff)
1645

    
1646
  if (not config.has_section(constants.INISECT_EXP) or
1647
      not config.has_section(constants.INISECT_INS)):
1648
    return None
1649

    
1650
  return config
1651

    
1652

    
1653
def ImportOSIntoInstance(instance, src_node, src_images, cluster_name):
1654
  """Import an os image into an instance.
1655

1656
  @type instance: L{objects.Instance}
1657
  @param instance: instance to import the disks into
1658
  @type src_node: string
1659
  @param src_node: source node for the disk images
1660
  @type src_images: list of string
1661
  @param src_images: absolute paths of the disk images
1662
  @rtype: list of boolean
1663
  @return: each boolean represent the success of importing the n-th disk
1664

1665
  """
1666
  import_env = OSEnvironment(instance)
1667
  inst_os = OSFromDisk(instance.os)
1668
  import_script = inst_os.import_script
1669

    
1670
  logfile = "%s/import-%s-%s-%s.log" % (constants.LOG_OS_DIR, instance.os,
1671
                                        instance.name, int(time.time()))
1672
  if not os.path.exists(constants.LOG_OS_DIR):
1673
    os.mkdir(constants.LOG_OS_DIR, 0750)
1674

    
1675
  comprcmd = "gunzip"
1676
  impcmd = utils.BuildShellCmd("(cd %s; %s >%s 2>&1)", inst_os.path,
1677
                               import_script, logfile)
1678

    
1679
  final_result = []
1680
  for idx, image in enumerate(src_images):
1681
    if image:
1682
      destcmd = utils.BuildShellCmd('cat %s', image)
1683
      remotecmd = _GetSshRunner(cluster_name).BuildCmd(src_node,
1684
                                                       constants.GANETI_RUNAS,
1685
                                                       destcmd)
1686
      command = '|'.join([utils.ShellQuoteArgs(remotecmd), comprcmd, impcmd])
1687
      import_env['IMPORT_DEVICE'] = import_env['DISK_%d_PATH' % idx]
1688
      import_env['IMPORT_INDEX'] = str(idx)
1689
      result = utils.RunCmd(command, env=import_env)
1690
      if result.failed:
1691
        logging.error("Disk import command '%s' returned error: %s"
1692
                      " output: %s", command, result.fail_reason,
1693
                      result.output)
1694
        final_result.append(False)
1695
      else:
1696
        final_result.append(True)
1697
    else:
1698
      final_result.append(True)
1699

    
1700
  return final_result
1701

    
1702

    
1703
def ListExports():
1704
  """Return a list of exports currently available on this machine.
1705

1706
  @rtype: list
1707
  @return: list of the exports
1708

1709
  """
1710
  if os.path.isdir(constants.EXPORT_DIR):
1711
    return utils.ListVisibleFiles(constants.EXPORT_DIR)
1712
  else:
1713
    return []
1714

    
1715

    
1716
def RemoveExport(export):
1717
  """Remove an existing export from the node.
1718

1719
  @type export: str
1720
  @param export: the name of the export to remove
1721
  @rtype: boolean
1722
  @return: the success of the operation
1723

1724
  """
1725
  target = os.path.join(constants.EXPORT_DIR, export)
1726

    
1727
  shutil.rmtree(target)
1728
  # TODO: catch some of the relevant exceptions and provide a pretty
1729
  # error message if rmtree fails.
1730

    
1731
  return True
1732

    
1733

    
1734
def RenameBlockDevices(devlist):
1735
  """Rename a list of block devices.
1736

1737
  @type devlist: list of tuples
1738
  @param devlist: list of tuples of the form  (disk,
1739
      new_logical_id, new_physical_id); disk is an
1740
      L{objects.Disk} object describing the current disk,
1741
      and new logical_id/physical_id is the name we
1742
      rename it to
1743
  @rtype: boolean
1744
  @return: True if all renames succeeded, False otherwise
1745

1746
  """
1747
  result = True
1748
  for disk, unique_id in devlist:
1749
    dev = _RecursiveFindBD(disk)
1750
    if dev is None:
1751
      result = False
1752
      continue
1753
    try:
1754
      old_rpath = dev.dev_path
1755
      dev.Rename(unique_id)
1756
      new_rpath = dev.dev_path
1757
      if old_rpath != new_rpath:
1758
        DevCacheManager.RemoveCache(old_rpath)
1759
        # FIXME: we should add the new cache information here, like:
1760
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
1761
        # but we don't have the owner here - maybe parse from existing
1762
        # cache? for now, we only lose lvm data when we rename, which
1763
        # is less critical than DRBD or MD
1764
    except errors.BlockDeviceError, err:
1765
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
1766
      result = False
1767
  return result
1768

    
1769

    
1770
def _TransformFileStorageDir(file_storage_dir):
1771
  """Checks whether given file_storage_dir is valid.
1772

1773
  Checks wheter the given file_storage_dir is within the cluster-wide
1774
  default file_storage_dir stored in SimpleStore. Only paths under that
1775
  directory are allowed.
1776

1777
  @type file_storage_dir: str
1778
  @param file_storage_dir: the path to check
1779

1780
  @return: the normalized path if valid, None otherwise
1781

1782
  """
1783
  cfg = _GetConfig()
1784
  file_storage_dir = os.path.normpath(file_storage_dir)
1785
  base_file_storage_dir = cfg.GetFileStorageDir()
1786
  if (not os.path.commonprefix([file_storage_dir, base_file_storage_dir]) ==
1787
      base_file_storage_dir):
1788
    logging.error("file storage directory '%s' is not under base file"
1789
                  " storage directory '%s'",
1790
                  file_storage_dir, base_file_storage_dir)
1791
    return None
1792
  return file_storage_dir
1793

    
1794

    
1795
def CreateFileStorageDir(file_storage_dir):
1796
  """Create file storage directory.
1797

1798
  @type file_storage_dir: str
1799
  @param file_storage_dir: directory to create
1800

1801
  @rtype: tuple
1802
  @return: tuple with first element a boolean indicating wheter dir
1803
      creation was successful or not
1804

1805
  """
1806
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
1807
  result = True,
1808
  if not file_storage_dir:
1809
    result = False,
1810
  else:
1811
    if os.path.exists(file_storage_dir):
1812
      if not os.path.isdir(file_storage_dir):
1813
        logging.error("'%s' is not a directory", file_storage_dir)
1814
        result = False,
1815
    else:
1816
      try:
1817
        os.makedirs(file_storage_dir, 0750)
1818
      except OSError, err:
1819
        logging.error("Cannot create file storage directory '%s': %s",
1820
                      file_storage_dir, err)
1821
        result = False,
1822
  return result
1823

    
1824

    
1825
def RemoveFileStorageDir(file_storage_dir):
1826
  """Remove file storage directory.
1827

1828
  Remove it only if it's empty. If not log an error and return.
1829

1830
  @type file_storage_dir: str
1831
  @param file_storage_dir: the directory we should cleanup
1832
  @rtype: tuple (success,)
1833
  @return: tuple of one element, C{success}, denoting
1834
      whether the operation was successfull
1835

1836
  """
1837
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
1838
  result = True,
1839
  if not file_storage_dir:
1840
    result = False,
1841
  else:
1842
    if os.path.exists(file_storage_dir):
1843
      if not os.path.isdir(file_storage_dir):
1844
        logging.error("'%s' is not a directory", file_storage_dir)
1845
        result = False,
1846
      # deletes dir only if empty, otherwise we want to return False
1847
      try:
1848
        os.rmdir(file_storage_dir)
1849
      except OSError, err:
1850
        logging.exception("Cannot remove file storage directory '%s'",
1851
                          file_storage_dir)
1852
        result = False,
1853
  return result
1854

    
1855

    
1856
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
1857
  """Rename the file storage directory.
1858

1859
  @type old_file_storage_dir: str
1860
  @param old_file_storage_dir: the current path
1861
  @type new_file_storage_dir: str
1862
  @param new_file_storage_dir: the name we should rename to
1863
  @rtype: tuple (success,)
1864
  @return: tuple of one element, C{success}, denoting
1865
      whether the operation was successful
1866

1867
  """
1868
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
1869
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
1870
  result = True,
1871
  if not old_file_storage_dir or not new_file_storage_dir:
1872
    result = False,
1873
  else:
1874
    if not os.path.exists(new_file_storage_dir):
1875
      if os.path.isdir(old_file_storage_dir):
1876
        try:
1877
          os.rename(old_file_storage_dir, new_file_storage_dir)
1878
        except OSError, err:
1879
          logging.exception("Cannot rename '%s' to '%s'",
1880
                            old_file_storage_dir, new_file_storage_dir)
1881
          result =  False,
1882
      else:
1883
        logging.error("'%s' is not a directory", old_file_storage_dir)
1884
        result = False,
1885
    else:
1886
      if os.path.exists(old_file_storage_dir):
1887
        logging.error("Cannot rename '%s' to '%s'. Both locations exist.",
1888
                      old_file_storage_dir, new_file_storage_dir)
1889
        result = False,
1890
  return result
1891

    
1892

    
1893
def _IsJobQueueFile(file_name):
1894
  """Checks whether the given filename is in the queue directory.
1895

1896
  @type file_name: str
1897
  @param file_name: the file name we should check
1898
  @rtype: boolean
1899
  @return: whether the file is under the queue directory
1900

1901
  """
1902
  queue_dir = os.path.normpath(constants.QUEUE_DIR)
1903
  result = (os.path.commonprefix([queue_dir, file_name]) == queue_dir)
1904

    
1905
  if not result:
1906
    logging.error("'%s' is not a file in the queue directory",
1907
                  file_name)
1908

    
1909
  return result
1910

    
1911

    
1912
def JobQueueUpdate(file_name, content):
1913
  """Updates a file in the queue directory.
1914

1915
  This is just a wrapper over L{utils.WriteFile}, with proper
1916
  checking.
1917

1918
  @type file_name: str
1919
  @param file_name: the job file name
1920
  @type content: str
1921
  @param content: the new job contents
1922
  @rtype: boolean
1923
  @return: the success of the operation
1924

1925
  """
1926
  if not _IsJobQueueFile(file_name):
1927
    return False
1928

    
1929
  # Write and replace the file atomically
1930
  utils.WriteFile(file_name, data=content)
1931

    
1932
  return True
1933

    
1934

    
1935
def JobQueueRename(old, new):
1936
  """Renames a job queue file.
1937

1938
  This is just a wrapper over L{os.rename} with proper checking.
1939

1940
  @type old: str
1941
  @param old: the old (actual) file name
1942
  @type new: str
1943
  @param new: the desired file name
1944
  @rtype: boolean
1945
  @return: the success of the operation
1946

1947
  """
1948
  if not (_IsJobQueueFile(old) and _IsJobQueueFile(new)):
1949
    return False
1950

    
1951
  os.rename(old, new)
1952

    
1953
  return True
1954

    
1955

    
1956
def JobQueueSetDrainFlag(drain_flag):
1957
  """Set the drain flag for the queue.
1958

1959
  This will set or unset the queue drain flag.
1960

1961
  @type drain_flag: boolean
1962
  @param drain_flag: if True, will set the drain flag, otherwise reset it.
1963
  @rtype: boolean
1964
  @return: always True
1965
  @warning: the function always returns True
1966

1967
  """
1968
  if drain_flag:
1969
    utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
1970
  else:
1971
    utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
1972

    
1973
  return True
1974

    
1975

    
1976
def CloseBlockDevices(disks):
1977
  """Closes the given block devices.
1978

1979
  This means they will be switched to secondary mode (in case of
1980
  DRBD).
1981

1982
  @type disks: list of L{objects.Disk}
1983
  @param disks: the list of disks to be closed
1984
  @rtype: tuple (success, message)
1985
  @return: a tuple of success and message, where success
1986
      indicates the succes of the operation, and message
1987
      which will contain the error details in case we
1988
      failed
1989

1990
  """
1991
  bdevs = []
1992
  for cf in disks:
1993
    rd = _RecursiveFindBD(cf)
1994
    if rd is None:
1995
      return (False, "Can't find device %s" % cf)
1996
    bdevs.append(rd)
1997

    
1998
  msg = []
1999
  for rd in bdevs:
2000
    try:
2001
      rd.Close()
2002
    except errors.BlockDeviceError, err:
2003
      msg.append(str(err))
2004
  if msg:
2005
    return (False, "Can't make devices secondary: %s" % ",".join(msg))
2006
  else:
2007
    return (True, "All devices secondary")
2008

    
2009

    
2010
def ValidateHVParams(hvname, hvparams):
2011
  """Validates the given hypervisor parameters.
2012

2013
  @type hvname: string
2014
  @param hvname: the hypervisor name
2015
  @type hvparams: dict
2016
  @param hvparams: the hypervisor parameters to be validated
2017
  @rtype: tuple (success, message)
2018
  @return: a tuple of success and message, where success
2019
      indicates the succes of the operation, and message
2020
      which will contain the error details in case we
2021
      failed
2022

2023
  """
2024
  try:
2025
    hv_type = hypervisor.GetHypervisor(hvname)
2026
    hv_type.ValidateParameters(hvparams)
2027
    return (True, "Validation passed")
2028
  except errors.HypervisorError, err:
2029
    return (False, str(err))
2030

    
2031

    
2032
class HooksRunner(object):
2033
  """Hook runner.
2034

2035
  This class is instantiated on the node side (ganeti-noded) and not
2036
  on the master side.
2037

2038
  """
2039
  RE_MASK = re.compile("^[a-zA-Z0-9_-]+$")
2040

    
2041
  def __init__(self, hooks_base_dir=None):
2042
    """Constructor for hooks runner.
2043

2044
    @type hooks_base_dir: str or None
2045
    @param hooks_base_dir: if not None, this overrides the
2046
        L{constants.HOOKS_BASE_DIR} (useful for unittests)
2047

2048
    """
2049
    if hooks_base_dir is None:
2050
      hooks_base_dir = constants.HOOKS_BASE_DIR
2051
    self._BASE_DIR = hooks_base_dir
2052

    
2053
  @staticmethod
2054
  def ExecHook(script, env):
2055
    """Exec one hook script.
2056

2057
    @type script: str
2058
    @param script: the full path to the script
2059
    @type env: dict
2060
    @param env: the environment with which to exec the script
2061
    @rtype: tuple (success, message)
2062
    @return: a tuple of success and message, where success
2063
        indicates the succes of the operation, and message
2064
        which will contain the error details in case we
2065
        failed
2066

2067
    """
2068
    # exec the process using subprocess and log the output
2069
    fdstdin = None
2070
    try:
2071
      fdstdin = open("/dev/null", "r")
2072
      child = subprocess.Popen([script], stdin=fdstdin, stdout=subprocess.PIPE,
2073
                               stderr=subprocess.STDOUT, close_fds=True,
2074
                               shell=False, cwd="/", env=env)
2075
      output = ""
2076
      try:
2077
        output = child.stdout.read(4096)
2078
        child.stdout.close()
2079
      except EnvironmentError, err:
2080
        output += "Hook script error: %s" % str(err)
2081

    
2082
      while True:
2083
        try:
2084
          result = child.wait()
2085
          break
2086
        except EnvironmentError, err:
2087
          if err.errno == errno.EINTR:
2088
            continue
2089
          raise
2090
    finally:
2091
      # try not to leak fds
2092
      for fd in (fdstdin, ):
2093
        if fd is not None:
2094
          try:
2095
            fd.close()
2096
          except EnvironmentError, err:
2097
            # just log the error
2098
            #logging.exception("Error while closing fd %s", fd)
2099
            pass
2100

    
2101
    return result == 0, output
2102

    
2103
  def RunHooks(self, hpath, phase, env):
2104
    """Run the scripts in the hooks directory.
2105

2106
    @type hpath: str
2107
    @param hpath: the path to the hooks directory which
2108
        holds the scripts
2109
    @type phase: str
2110
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
2111
        L{constants.HOOKS_PHASE_POST}
2112
    @type env: dict
2113
    @param env: dictionary with the environment for the hook
2114
    @rtype: list
2115
    @return: list of 3-element tuples:
2116
      - script path
2117
      - script result, either L{constants.HKR_SUCCESS} or
2118
        L{constants.HKR_FAIL}
2119
      - output of the script
2120

2121
    @raise errors.ProgrammerError: for invalid input
2122
        parameters
2123

2124
    """
2125
    if phase == constants.HOOKS_PHASE_PRE:
2126
      suffix = "pre"
2127
    elif phase == constants.HOOKS_PHASE_POST:
2128
      suffix = "post"
2129
    else:
2130
      raise errors.ProgrammerError("Unknown hooks phase: '%s'" % phase)
2131
    rr = []
2132

    
2133
    subdir = "%s-%s.d" % (hpath, suffix)
2134
    dir_name = "%s/%s" % (self._BASE_DIR, subdir)
2135
    try:
2136
      dir_contents = utils.ListVisibleFiles(dir_name)
2137
    except OSError, err:
2138
      # FIXME: must log output in case of failures
2139
      return rr
2140

    
2141
    # we use the standard python sort order,
2142
    # so 00name is the recommended naming scheme
2143
    dir_contents.sort()
2144
    for relname in dir_contents:
2145
      fname = os.path.join(dir_name, relname)
2146
      if not (os.path.isfile(fname) and os.access(fname, os.X_OK) and
2147
          self.RE_MASK.match(relname) is not None):
2148
        rrval = constants.HKR_SKIP
2149
        output = ""
2150
      else:
2151
        result, output = self.ExecHook(fname, env)
2152
        if not result:
2153
          rrval = constants.HKR_FAIL
2154
        else:
2155
          rrval = constants.HKR_SUCCESS
2156
      rr.append(("%s/%s" % (subdir, relname), rrval, output))
2157

    
2158
    return rr
2159

    
2160

    
2161
class IAllocatorRunner(object):
2162
  """IAllocator runner.
2163

2164
  This class is instantiated on the node side (ganeti-noded) and not on
2165
  the master side.
2166

2167
  """
2168
  def Run(self, name, idata):
2169
    """Run an iallocator script.
2170

2171
    @type name: str
2172
    @param name: the iallocator script name
2173
    @type idata: str
2174
    @param idata: the allocator input data
2175

2176
    @rtype: tuple
2177
    @return: four element tuple of:
2178
       - run status (one of the IARUN_ constants)
2179
       - stdout
2180
       - stderr
2181
       - fail reason (as from L{utils.RunResult})
2182

2183
    """
2184
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
2185
                                  os.path.isfile)
2186
    if alloc_script is None:
2187
      return (constants.IARUN_NOTFOUND, None, None, None)
2188

    
2189
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
2190
    try:
2191
      os.write(fd, idata)
2192
      os.close(fd)
2193
      result = utils.RunCmd([alloc_script, fin_name])
2194
      if result.failed:
2195
        return (constants.IARUN_FAILURE, result.stdout, result.stderr,
2196
                result.fail_reason)
2197
    finally:
2198
      os.unlink(fin_name)
2199

    
2200
    return (constants.IARUN_SUCCESS, result.stdout, result.stderr, None)
2201

    
2202

    
2203
class DevCacheManager(object):
2204
  """Simple class for managing a cache of block device information.
2205

2206
  """
2207
  _DEV_PREFIX = "/dev/"
2208
  _ROOT_DIR = constants.BDEV_CACHE_DIR
2209

    
2210
  @classmethod
2211
  def _ConvertPath(cls, dev_path):
2212
    """Converts a /dev/name path to the cache file name.
2213

2214
    This replaces slashes with underscores and strips the /dev
2215
    prefix. It then returns the full path to the cache file.
2216

2217
    @type dev_path: str
2218
    @param dev_path: the C{/dev/} path name
2219
    @rtype: str
2220
    @return: the converted path name
2221

2222
    """
2223
    if dev_path.startswith(cls._DEV_PREFIX):
2224
      dev_path = dev_path[len(cls._DEV_PREFIX):]
2225
    dev_path = dev_path.replace("/", "_")
2226
    fpath = "%s/bdev_%s" % (cls._ROOT_DIR, dev_path)
2227
    return fpath
2228

    
2229
  @classmethod
2230
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
2231
    """Updates the cache information for a given device.
2232

2233
    @type dev_path: str
2234
    @param dev_path: the pathname of the device
2235
    @type owner: str
2236
    @param owner: the owner (instance name) of the device
2237
    @type on_primary: bool
2238
    @param on_primary: whether this is the primary
2239
        node nor not
2240
    @type iv_name: str
2241
    @param iv_name: the instance-visible name of the
2242
        device, as in L{objects.Disk.iv_name}
2243

2244
    @rtype: None
2245

2246
    """
2247
    if dev_path is None:
2248
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
2249
      return
2250
    fpath = cls._ConvertPath(dev_path)
2251
    if on_primary:
2252
      state = "primary"
2253
    else:
2254
      state = "secondary"
2255
    if iv_name is None:
2256
      iv_name = "not_visible"
2257
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
2258
    try:
2259
      utils.WriteFile(fpath, data=fdata)
2260
    except EnvironmentError, err:
2261
      logging.exception("Can't update bdev cache for %s", dev_path)
2262

    
2263
  @classmethod
2264
  def RemoveCache(cls, dev_path):
2265
    """Remove data for a dev_path.
2266

2267
    This is just a wrapper over L{utils.RemoveFile} with a converted
2268
    path name and logging.
2269

2270
    @type dev_path: str
2271
    @param dev_path: the pathname of the device
2272

2273
    @rtype: None
2274

2275
    """
2276
    if dev_path is None:
2277
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
2278
      return
2279
    fpath = cls._ConvertPath(dev_path)
2280
    try:
2281
      utils.RemoveFile(fpath)
2282
    except EnvironmentError, err:
2283
      logging.exception("Can't update bdev cache for %s", dev_path)