Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ 030b218a

History | View | Annotate | Download (80.6 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
import zlib
37
import base64
38

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

    
48

    
49
def _GetConfig():
50
  """Simple wrapper to return a SimpleStore.
51

52
  @rtype: L{ssconf.SimpleStore}
53
  @return: a SimpleStore instance
54

55
  """
56
  return ssconf.SimpleStore()
57

    
58

    
59
def _GetSshRunner(cluster_name):
60
  """Simple wrapper to return an SshRunner.
61

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

68
  """
69
  return ssh.SshRunner(cluster_name)
70

    
71

    
72
def _Decompress(data):
73
  """Unpacks data compressed by the RPC client.
74

75
  @type data: list or tuple
76
  @param data: Data sent by RPC client
77
  @rtype: str
78
  @return: Decompressed data
79

80
  """
81
  assert isinstance(data, (list, tuple))
82
  assert len(data) == 2
83
  (encoding, content) = data
84
  if encoding == constants.RPC_ENCODING_NONE:
85
    return content
86
  elif encoding == constants.RPC_ENCODING_ZLIB_BASE64:
87
    return zlib.decompress(base64.b64decode(content))
88
  else:
89
    raise AssertionError("Unknown data encoding")
90

    
91

    
92
def _CleanDirectory(path, exclude=None):
93
  """Removes all regular files in a directory.
94

95
  @type path: str
96
  @param path: the directory to clean
97
  @type exclude: list
98
  @param exclude: list of files to be excluded, defaults
99
      to the empty list
100

101
  """
102
  if not os.path.isdir(path):
103
    return
104
  if exclude is None:
105
    exclude = []
106
  else:
107
    # Normalize excluded paths
108
    exclude = [os.path.normpath(i) for i in exclude]
109

    
110
  for rel_name in utils.ListVisibleFiles(path):
111
    full_name = os.path.normpath(os.path.join(path, rel_name))
112
    if full_name in exclude:
113
      continue
114
    if os.path.isfile(full_name) and not os.path.islink(full_name):
115
      utils.RemoveFile(full_name)
116

    
117

    
118
def JobQueuePurge():
119
  """Removes job queue files and archived jobs.
120

121
  @rtype: None
122

123
  """
124
  _CleanDirectory(constants.QUEUE_DIR, exclude=[constants.JOB_QUEUE_LOCK_FILE])
125
  _CleanDirectory(constants.JOB_QUEUE_ARCHIVE_DIR)
126

    
127

    
128
def GetMasterInfo():
129
  """Returns master information.
130

131
  This is an utility function to compute master information, either
132
  for consumption here or from the node daemon.
133

134
  @rtype: tuple
135
  @return: (master_netdev, master_ip, master_name) if we have a good
136
      configuration, otherwise (None, None, None)
137

138
  """
139
  try:
140
    cfg = _GetConfig()
141
    master_netdev = cfg.GetMasterNetdev()
142
    master_ip = cfg.GetMasterIP()
143
    master_node = cfg.GetMasterNode()
144
  except errors.ConfigurationError, err:
145
    logging.exception("Cluster configuration incomplete")
146
    return (None, None, None)
147
  return (master_netdev, master_ip, master_node)
148

    
149

    
150
def StartMaster(start_daemons):
151
  """Activate local node as master node.
152

153
  The function will always try activate the IP address of the master
154
  (unless someone else has it). It will also start the master daemons,
155
  based on the start_daemons parameter.
156

157
  @type start_daemons: boolean
158
  @param start_daemons: whther to also start the master
159
      daemons (ganeti-masterd and ganeti-rapi)
160
  @rtype: None
161

162
  """
163
  ok = True
164
  master_netdev, master_ip, _ = GetMasterInfo()
165
  if not master_netdev:
166
    return False
167

    
168
  if utils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
169
    if utils.OwnIpAddress(master_ip):
170
      # we already have the ip:
171
      logging.debug("Already started")
172
    else:
173
      logging.error("Someone else has the master ip, not activating")
174
      ok = False
175
  else:
176
    result = utils.RunCmd(["ip", "address", "add", "%s/32" % master_ip,
177
                           "dev", master_netdev, "label",
178
                           "%s:0" % master_netdev])
179
    if result.failed:
180
      logging.error("Can't activate master IP: %s", result.output)
181
      ok = False
182

    
183
    result = utils.RunCmd(["arping", "-q", "-U", "-c 3", "-I", master_netdev,
184
                           "-s", master_ip, master_ip])
185
    # we'll ignore the exit code of arping
186

    
187
  # and now start the master and rapi daemons
188
  if start_daemons:
189
    for daemon in 'ganeti-masterd', 'ganeti-rapi':
190
      result = utils.RunCmd([daemon])
191
      if result.failed:
192
        logging.error("Can't start daemon %s: %s", daemon, result.output)
193
        ok = False
194
  return ok
195

    
196

    
197
def StopMaster(stop_daemons):
198
  """Deactivate this node as master.
199

200
  The function will always try to deactivate the IP address of the
201
  master. It will also stop the master daemons depending on the
202
  stop_daemons parameter.
203

204
  @type stop_daemons: boolean
205
  @param stop_daemons: whether to also stop the master daemons
206
      (ganeti-masterd and ganeti-rapi)
207
  @rtype: None
208

209
  """
210
  master_netdev, master_ip, _ = GetMasterInfo()
211
  if not master_netdev:
212
    return False
213

    
214
  result = utils.RunCmd(["ip", "address", "del", "%s/32" % master_ip,
215
                         "dev", master_netdev])
216
  if result.failed:
217
    logging.error("Can't remove the master IP, error: %s", result.output)
218
    # but otherwise ignore the failure
219

    
220
  if stop_daemons:
221
    # stop/kill the rapi and the master daemon
222
    for daemon in constants.RAPI_PID, constants.MASTERD_PID:
223
      utils.KillProcess(utils.ReadPidFile(utils.DaemonPidFileName(daemon)))
224

    
225
  return True
226

    
227

    
228
def AddNode(dsa, dsapub, rsa, rsapub, sshkey, sshpub):
229
  """Joins this node to the cluster.
230

231
  This does the following:
232
      - updates the hostkeys of the machine (rsa and dsa)
233
      - adds the ssh private key to the user
234
      - adds the ssh public key to the users' authorized_keys file
235

236
  @type dsa: str
237
  @param dsa: the DSA private key to write
238
  @type dsapub: str
239
  @param dsapub: the DSA public key to write
240
  @type rsa: str
241
  @param rsa: the RSA private key to write
242
  @type rsapub: str
243
  @param rsapub: the RSA public key to write
244
  @type sshkey: str
245
  @param sshkey: the SSH private key to write
246
  @type sshpub: str
247
  @param sshpub: the SSH public key to write
248
  @rtype: boolean
249
  @return: the success of the operation
250

251
  """
252
  sshd_keys =  [(constants.SSH_HOST_RSA_PRIV, rsa, 0600),
253
                (constants.SSH_HOST_RSA_PUB, rsapub, 0644),
254
                (constants.SSH_HOST_DSA_PRIV, dsa, 0600),
255
                (constants.SSH_HOST_DSA_PUB, dsapub, 0644)]
256
  for name, content, mode in sshd_keys:
257
    utils.WriteFile(name, data=content, mode=mode)
258

    
259
  try:
260
    priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS,
261
                                                    mkdir=True)
262
  except errors.OpExecError, err:
263
    msg = "Error while processing user ssh files"
264
    logging.exception(msg)
265
    return (False, "%s: %s" % (msg, err))
266

    
267
  for name, content in [(priv_key, sshkey), (pub_key, sshpub)]:
268
    utils.WriteFile(name, data=content, mode=0600)
269

    
270
  utils.AddAuthorizedKey(auth_keys, sshpub)
271

    
272
  utils.RunCmd([constants.SSH_INITD_SCRIPT, "restart"])
273

    
274
  return (True, "Node added successfully")
275

    
276

    
277
def LeaveCluster():
278
  """Cleans up and remove the current node.
279

280
  This function cleans up and prepares the current node to be removed
281
  from the cluster.
282

283
  If processing is successful, then it raises an
284
  L{errors.QuitGanetiException} which is used as a special case to
285
  shutdown the node daemon.
286

287
  """
288
  _CleanDirectory(constants.DATA_DIR)
289
  JobQueuePurge()
290

    
291
  try:
292
    priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
293
  except errors.OpExecError:
294
    logging.exception("Error while processing ssh files")
295
    return
296

    
297
  f = open(pub_key, 'r')
298
  try:
299
    utils.RemoveAuthorizedKey(auth_keys, f.read(8192))
300
  finally:
301
    f.close()
302

    
303
  utils.RemoveFile(priv_key)
304
  utils.RemoveFile(pub_key)
305

    
306
  # Return a reassuring string to the caller, and quit
307
  raise errors.QuitGanetiException(False, 'Shutdown scheduled')
308

    
309

    
310
def GetNodeInfo(vgname, hypervisor_type):
311
  """Gives back a hash with different informations about the node.
312

313
  @type vgname: C{string}
314
  @param vgname: the name of the volume group to ask for disk space information
315
  @type hypervisor_type: C{str}
316
  @param hypervisor_type: the name of the hypervisor to ask for
317
      memory information
318
  @rtype: C{dict}
319
  @return: dictionary with the following keys:
320
      - vg_size is the size of the configured volume group in MiB
321
      - vg_free is the free size of the volume group in MiB
322
      - memory_dom0 is the memory allocated for domain0 in MiB
323
      - memory_free is the currently available (free) ram in MiB
324
      - memory_total is the total number of ram in MiB
325

326
  """
327
  outputarray = {}
328
  vginfo = _GetVGInfo(vgname)
329
  outputarray['vg_size'] = vginfo['vg_size']
330
  outputarray['vg_free'] = vginfo['vg_free']
331

    
332
  hyper = hypervisor.GetHypervisor(hypervisor_type)
333
  hyp_info = hyper.GetNodeInfo()
334
  if hyp_info is not None:
335
    outputarray.update(hyp_info)
336

    
337
  f = open("/proc/sys/kernel/random/boot_id", 'r')
338
  try:
339
    outputarray["bootid"] = f.read(128).rstrip("\n")
340
  finally:
341
    f.close()
342

    
343
  return outputarray
344

    
345

    
346
def VerifyNode(what, cluster_name):
347
  """Verify the status of the local node.
348

349
  Based on the input L{what} parameter, various checks are done on the
350
  local node.
351

352
  If the I{filelist} key is present, this list of
353
  files is checksummed and the file/checksum pairs are returned.
354

355
  If the I{nodelist} key is present, we check that we have
356
  connectivity via ssh with the target nodes (and check the hostname
357
  report).
358

359
  If the I{node-net-test} key is present, we check that we have
360
  connectivity to the given nodes via both primary IP and, if
361
  applicable, secondary IPs.
362

363
  @type what: C{dict}
364
  @param what: a dictionary of things to check:
365
      - filelist: list of files for which to compute checksums
366
      - nodelist: list of nodes we should check ssh communication with
367
      - node-net-test: list of nodes we should check node daemon port
368
        connectivity with
369
      - hypervisor: list with hypervisors to run the verify for
370
  @rtype: dict
371
  @return: a dictionary with the same keys as the input dict, and
372
      values representing the result of the checks
373

374
  """
375
  result = {}
376

    
377
  if constants.NV_HYPERVISOR in what:
378
    result[constants.NV_HYPERVISOR] = tmp = {}
379
    for hv_name in what[constants.NV_HYPERVISOR]:
380
      tmp[hv_name] = hypervisor.GetHypervisor(hv_name).Verify()
381

    
382
  if constants.NV_FILELIST in what:
383
    result[constants.NV_FILELIST] = utils.FingerprintFiles(
384
      what[constants.NV_FILELIST])
385

    
386
  if constants.NV_NODELIST in what:
387
    result[constants.NV_NODELIST] = tmp = {}
388
    random.shuffle(what[constants.NV_NODELIST])
389
    for node in what[constants.NV_NODELIST]:
390
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
391
      if not success:
392
        tmp[node] = message
393

    
394
  if constants.NV_NODENETTEST in what:
395
    result[constants.NV_NODENETTEST] = tmp = {}
396
    my_name = utils.HostInfo().name
397
    my_pip = my_sip = None
398
    for name, pip, sip in what[constants.NV_NODENETTEST]:
399
      if name == my_name:
400
        my_pip = pip
401
        my_sip = sip
402
        break
403
    if not my_pip:
404
      tmp[my_name] = ("Can't find my own primary/secondary IP"
405
                      " in the node list")
406
    else:
407
      port = utils.GetNodeDaemonPort()
408
      for name, pip, sip in what[constants.NV_NODENETTEST]:
409
        fail = []
410
        if not utils.TcpPing(pip, port, source=my_pip):
411
          fail.append("primary")
412
        if sip != pip:
413
          if not utils.TcpPing(sip, port, source=my_sip):
414
            fail.append("secondary")
415
        if fail:
416
          tmp[name] = ("failure using the %s interface(s)" %
417
                       " and ".join(fail))
418

    
419
  if constants.NV_LVLIST in what:
420
    result[constants.NV_LVLIST] = GetVolumeList(what[constants.NV_LVLIST])
421

    
422
  if constants.NV_INSTANCELIST in what:
423
    result[constants.NV_INSTANCELIST] = GetInstanceList(
424
      what[constants.NV_INSTANCELIST])
425

    
426
  if constants.NV_VGLIST in what:
427
    result[constants.NV_VGLIST] = ListVolumeGroups()
428

    
429
  if constants.NV_VERSION in what:
430
    result[constants.NV_VERSION] = (constants.PROTOCOL_VERSION,
431
                                    constants.RELEASE_VERSION)
432

    
433
  if constants.NV_HVINFO in what:
434
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
435
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
436

    
437
  if constants.NV_DRBDLIST in what:
438
    try:
439
      used_minors = bdev.DRBD8.GetUsedDevs().keys()
440
    except errors.BlockDeviceError, err:
441
      logging.warning("Can't get used minors list", exc_info=True)
442
      used_minors = str(err)
443
    result[constants.NV_DRBDLIST] = used_minors
444

    
445
  return result
446

    
447

    
448
def GetVolumeList(vg_name):
449
  """Compute list of logical volumes and their size.
450

451
  @type vg_name: str
452
  @param vg_name: the volume group whose LVs we should list
453
  @rtype: dict
454
  @return:
455
      dictionary of all partions (key) with value being a tuple of
456
      their size (in MiB), inactive and online status::
457

458
        {'test1': ('20.06', True, True)}
459

460
      in case of errors, a string is returned with the error
461
      details.
462

463
  """
464
  lvs = {}
465
  sep = '|'
466
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
467
                         "--separator=%s" % sep,
468
                         "-olv_name,lv_size,lv_attr", vg_name])
469
  if result.failed:
470
    logging.error("Failed to list logical volumes, lvs output: %s",
471
                  result.output)
472
    return result.output
473

    
474
  valid_line_re = re.compile("^ *([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
475
  for line in result.stdout.splitlines():
476
    line = line.strip()
477
    match = valid_line_re.match(line)
478
    if not match:
479
      logging.error("Invalid line returned from lvs output: '%s'", line)
480
      continue
481
    name, size, attr = match.groups()
482
    inactive = attr[4] == '-'
483
    online = attr[5] == 'o'
484
    lvs[name] = (size, inactive, online)
485

    
486
  return lvs
487

    
488

    
489
def ListVolumeGroups():
490
  """List the volume groups and their size.
491

492
  @rtype: dict
493
  @return: dictionary with keys volume name and values the
494
      size of the volume
495

496
  """
497
  return utils.ListVolumeGroups()
498

    
499

    
500
def NodeVolumes():
501
  """List all volumes on this node.
502

503
  @rtype: list
504
  @return:
505
    A list of dictionaries, each having four keys:
506
      - name: the logical volume name,
507
      - size: the size of the logical volume
508
      - dev: the physical device on which the LV lives
509
      - vg: the volume group to which it belongs
510

511
    In case of errors, we return an empty list and log the
512
    error.
513

514
    Note that since a logical volume can live on multiple physical
515
    volumes, the resulting list might include a logical volume
516
    multiple times.
517

518
  """
519
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
520
                         "--separator=|",
521
                         "--options=lv_name,lv_size,devices,vg_name"])
522
  if result.failed:
523
    logging.error("Failed to list logical volumes, lvs output: %s",
524
                  result.output)
525
    return []
526

    
527
  def parse_dev(dev):
528
    if '(' in dev:
529
      return dev.split('(')[0]
530
    else:
531
      return dev
532

    
533
  def map_line(line):
534
    return {
535
      'name': line[0].strip(),
536
      'size': line[1].strip(),
537
      'dev': parse_dev(line[2].strip()),
538
      'vg': line[3].strip(),
539
    }
540

    
541
  return [map_line(line.split('|')) for line in result.stdout.splitlines()
542
          if line.count('|') >= 3]
543

    
544

    
545
def BridgesExist(bridges_list):
546
  """Check if a list of bridges exist on the current node.
547

548
  @rtype: boolean
549
  @return: C{True} if all of them exist, C{False} otherwise
550

551
  """
552
  for bridge in bridges_list:
553
    if not utils.BridgeExists(bridge):
554
      return False
555

    
556
  return True
557

    
558

    
559
def GetInstanceList(hypervisor_list):
560
  """Provides a list of instances.
561

562
  @type hypervisor_list: list
563
  @param hypervisor_list: the list of hypervisors to query information
564

565
  @rtype: list
566
  @return: a list of all running instances on the current node
567
    - instance1.example.com
568
    - instance2.example.com
569

570
  """
571
  results = []
572
  for hname in hypervisor_list:
573
    try:
574
      names = hypervisor.GetHypervisor(hname).ListInstances()
575
      results.extend(names)
576
    except errors.HypervisorError, err:
577
      logging.exception("Error enumerating instances for hypevisor %s", hname)
578
      raise
579

    
580
  return results
581

    
582

    
583
def GetInstanceInfo(instance, hname):
584
  """Gives back the informations about an instance as a dictionary.
585

586
  @type instance: string
587
  @param instance: the instance name
588
  @type hname: string
589
  @param hname: the hypervisor type of the instance
590

591
  @rtype: dict
592
  @return: dictionary with the following keys:
593
      - memory: memory size of instance (int)
594
      - state: xen state of instance (string)
595
      - time: cpu time of instance (float)
596

597
  """
598
  output = {}
599

    
600
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
601
  if iinfo is not None:
602
    output['memory'] = iinfo[2]
603
    output['state'] = iinfo[4]
604
    output['time'] = iinfo[5]
605

    
606
  return output
607

    
608

    
609
def GetInstanceMigratable(instance):
610
  """Gives whether an instance can be migrated.
611

612
  @type instance: L{objects.Instance}
613
  @param instance: object representing the instance to be checked.
614

615
  @rtype: tuple
616
  @return: tuple of (result, description) where:
617
      - result: whether the instance can be migrated or not
618
      - description: a description of the issue, if relevant
619

620
  """
621
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
622
  if instance.name not in hyper.ListInstances():
623
    return (False, 'not running')
624

    
625
  for idx in range(len(instance.disks)):
626
    link_name = _GetBlockDevSymlinkPath(instance.name, idx)
627
    if not os.path.islink(link_name):
628
      return (False, 'not restarted since ganeti 1.2.5')
629

    
630
  return (True, '')
631

    
632

    
633
def GetAllInstancesInfo(hypervisor_list):
634
  """Gather data about all instances.
635

636
  This is the equivalent of L{GetInstanceInfo}, except that it
637
  computes data for all instances at once, thus being faster if one
638
  needs data about more than one instance.
639

640
  @type hypervisor_list: list
641
  @param hypervisor_list: list of hypervisors to query for instance data
642

643
  @rtype: dict
644
  @return: dictionary of instance: data, with data having the following keys:
645
      - memory: memory size of instance (int)
646
      - state: xen state of instance (string)
647
      - time: cpu time of instance (float)
648
      - vcpus: the number of vcpus
649

650
  """
651
  output = {}
652

    
653
  for hname in hypervisor_list:
654
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
655
    if iinfo:
656
      for name, inst_id, memory, vcpus, state, times in iinfo:
657
        value = {
658
          'memory': memory,
659
          'vcpus': vcpus,
660
          'state': state,
661
          'time': times,
662
          }
663
        if name in output:
664
          # we only check static parameters, like memory and vcpus,
665
          # and not state and time which can change between the
666
          # invocations of the different hypervisors
667
          for key in 'memory', 'vcpus':
668
            if value[key] != output[name][key]:
669
              raise errors.HypervisorError("Instance %s is running twice"
670
                                           " with different parameters" % name)
671
        output[name] = value
672

    
673
  return output
674

    
675

    
676
def InstanceOsAdd(instance):
677
  """Add an OS to an instance.
678

679
  @type instance: L{objects.Instance}
680
  @param instance: Instance whose OS is to be installed
681
  @rtype: boolean
682
  @return: the success of the operation
683

684
  """
685
  try:
686
    inst_os = OSFromDisk(instance.os)
687
  except errors.InvalidOS, err:
688
    os_name, os_dir, os_err = err.args
689
    if os_dir is None:
690
      return (False, "Can't find OS '%s': %s" % (os_name, os_err))
691
    else:
692
      return (False, "Error parsing OS '%s' in directory %s: %s" %
693
              (os_name, os_dir, os_err))
694

    
695
  create_env = OSEnvironment(instance)
696

    
697
  logfile = "%s/add-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
698
                                     instance.name, int(time.time()))
699

    
700
  result = utils.RunCmd([inst_os.create_script], env=create_env,
701
                        cwd=inst_os.path, output=logfile,)
702
  if result.failed:
703
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
704
                  " output: %s", result.cmd, result.fail_reason, logfile,
705
                  result.output)
706
    lines = [utils.SafeEncode(val)
707
             for val in utils.TailFile(logfile, lines=20)]
708
    return (False, "OS create script failed (%s), last lines in the"
709
            " log file:\n%s" % (result.fail_reason, "\n".join(lines)))
710

    
711
  return (True, "Successfully installed")
712

    
713

    
714
def RunRenameInstance(instance, old_name):
715
  """Run the OS rename script for an instance.
716

717
  @type instance: L{objects.Instance}
718
  @param instance: Instance whose OS is to be installed
719
  @type old_name: string
720
  @param old_name: previous instance name
721
  @rtype: boolean
722
  @return: the success of the operation
723

724
  """
725
  inst_os = OSFromDisk(instance.os)
726

    
727
  rename_env = OSEnvironment(instance)
728
  rename_env['OLD_INSTANCE_NAME'] = old_name
729

    
730
  logfile = "%s/rename-%s-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
731
                                           old_name,
732
                                           instance.name, int(time.time()))
733

    
734
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
735
                        cwd=inst_os.path, output=logfile)
736

    
737
  if result.failed:
738
    logging.error("os create command '%s' returned error: %s output: %s",
739
                  result.cmd, result.fail_reason, result.output)
740
    lines = [utils.SafeEncode(val)
741
             for val in utils.TailFile(logfile, lines=20)]
742
    return (False, "OS rename script failed (%s), last lines in the"
743
            " log file:\n%s" % (result.fail_reason, "\n".join(lines)))
744

    
745
  return (True, "Rename successful")
746

    
747

    
748
def _GetVGInfo(vg_name):
749
  """Get informations about the volume group.
750

751
  @type vg_name: str
752
  @param vg_name: the volume group which we query
753
  @rtype: dict
754
  @return:
755
    A dictionary with the following keys:
756
      - C{vg_size} is the total size of the volume group in MiB
757
      - C{vg_free} is the free size of the volume group in MiB
758
      - C{pv_count} are the number of physical disks in that VG
759

760
    If an error occurs during gathering of data, we return the same dict
761
    with keys all set to None.
762

763
  """
764
  retdic = dict.fromkeys(["vg_size", "vg_free", "pv_count"])
765

    
766
  retval = utils.RunCmd(["vgs", "-ovg_size,vg_free,pv_count", "--noheadings",
767
                         "--nosuffix", "--units=m", "--separator=:", vg_name])
768

    
769
  if retval.failed:
770
    logging.error("volume group %s not present", vg_name)
771
    return retdic
772
  valarr = retval.stdout.strip().rstrip(':').split(':')
773
  if len(valarr) == 3:
774
    try:
775
      retdic = {
776
        "vg_size": int(round(float(valarr[0]), 0)),
777
        "vg_free": int(round(float(valarr[1]), 0)),
778
        "pv_count": int(valarr[2]),
779
        }
780
    except ValueError, err:
781
      logging.exception("Fail to parse vgs output")
782
  else:
783
    logging.error("vgs output has the wrong number of fields (expected"
784
                  " three): %s", str(valarr))
785
  return retdic
786

    
787

    
788
def _GetBlockDevSymlinkPath(instance_name, idx):
789
  return os.path.join(constants.DISK_LINKS_DIR,
790
                      "%s:%d" % (instance_name, idx))
791

    
792

    
793
def _SymlinkBlockDev(instance_name, device_path, idx):
794
  """Set up symlinks to a instance's block device.
795

796
  This is an auxiliary function run when an instance is start (on the primary
797
  node) or when an instance is migrated (on the target node).
798

799

800
  @param instance_name: the name of the target instance
801
  @param device_path: path of the physical block device, on the node
802
  @param idx: the disk index
803
  @return: absolute path to the disk's symlink
804

805
  """
806
  link_name = _GetBlockDevSymlinkPath(instance_name, idx)
807
  try:
808
    os.symlink(device_path, link_name)
809
  except OSError, err:
810
    if err.errno == errno.EEXIST:
811
      if (not os.path.islink(link_name) or
812
          os.readlink(link_name) != device_path):
813
        os.remove(link_name)
814
        os.symlink(device_path, link_name)
815
    else:
816
      raise
817

    
818
  return link_name
819

    
820

    
821
def _RemoveBlockDevLinks(instance_name, disks):
822
  """Remove the block device symlinks belonging to the given instance.
823

824
  """
825
  for idx, disk in enumerate(disks):
826
    link_name = _GetBlockDevSymlinkPath(instance_name, idx)
827
    if os.path.islink(link_name):
828
      try:
829
        os.remove(link_name)
830
      except OSError:
831
        logging.exception("Can't remove symlink '%s'", link_name)
832

    
833

    
834
def _GatherAndLinkBlockDevs(instance):
835
  """Set up an instance's block device(s).
836

837
  This is run on the primary node at instance startup. The block
838
  devices must be already assembled.
839

840
  @type instance: L{objects.Instance}
841
  @param instance: the instance whose disks we shoul assemble
842
  @rtype: list
843
  @return: list of (disk_object, device_path)
844

845
  """
846
  block_devices = []
847
  for idx, disk in enumerate(instance.disks):
848
    device = _RecursiveFindBD(disk)
849
    if device is None:
850
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
851
                                    str(disk))
852
    device.Open()
853
    try:
854
      link_name = _SymlinkBlockDev(instance.name, device.dev_path, idx)
855
    except OSError, e:
856
      raise errors.BlockDeviceError("Cannot create block device symlink: %s" %
857
                                    e.strerror)
858

    
859
    block_devices.append((disk, link_name))
860

    
861
  return block_devices
862

    
863

    
864
def StartInstance(instance):
865
  """Start an instance.
866

867
  @type instance: L{objects.Instance}
868
  @param instance: the instance object
869
  @rtype: boolean
870
  @return: whether the startup was successful or not
871

872
  """
873
  running_instances = GetInstanceList([instance.hypervisor])
874

    
875
  if instance.name in running_instances:
876
    return (True, "Already running")
877

    
878
  try:
879
    block_devices = _GatherAndLinkBlockDevs(instance)
880
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
881
    hyper.StartInstance(instance, block_devices)
882
  except errors.BlockDeviceError, err:
883
    logging.exception("Failed to start instance")
884
    return (False, "Block device error: %s" % str(err))
885
  except errors.HypervisorError, err:
886
    logging.exception("Failed to start instance")
887
    _RemoveBlockDevLinks(instance.name, instance.disks)
888
    return (False, "Hypervisor error: %s" % str(err))
889

    
890
  return (True, "Instance started successfully")
891

    
892

    
893
def InstanceShutdown(instance):
894
  """Shut an instance down.
895

896
  @note: this functions uses polling with a hardcoded timeout.
897

898
  @type instance: L{objects.Instance}
899
  @param instance: the instance object
900
  @rtype: boolean
901
  @return: whether the startup was successful or not
902

903
  """
904
  hv_name = instance.hypervisor
905
  running_instances = GetInstanceList([hv_name])
906

    
907
  if instance.name not in running_instances:
908
    return (True, "Instance already stopped")
909

    
910
  hyper = hypervisor.GetHypervisor(hv_name)
911
  try:
912
    hyper.StopInstance(instance)
913
  except errors.HypervisorError, err:
914
    msg = "Failed to stop instance %s: %s" % (instance.name, err)
915
    logging.error(msg)
916
    return (False, msg)
917

    
918
  # test every 10secs for 2min
919

    
920
  time.sleep(1)
921
  for dummy in range(11):
922
    if instance.name not in GetInstanceList([hv_name]):
923
      break
924
    time.sleep(10)
925
  else:
926
    # the shutdown did not succeed
927
    logging.error("Shutdown of '%s' unsuccessful, using destroy",
928
                  instance.name)
929

    
930
    try:
931
      hyper.StopInstance(instance, force=True)
932
    except errors.HypervisorError, err:
933
      msg = "Failed to force stop instance %s: %s" % (instance.name, err)
934
      logging.error(msg)
935
      return (False, msg)
936

    
937
    time.sleep(1)
938
    if instance.name in GetInstanceList([hv_name]):
939
      msg = ("Could not shutdown instance %s even by destroy" %
940
             instance.name)
941
      logging.error(msg)
942
      return (False, msg)
943

    
944
  _RemoveBlockDevLinks(instance.name, instance.disks)
945

    
946
  return (True, "Instance has been shutdown successfully")
947

    
948

    
949
def InstanceReboot(instance, reboot_type):
950
  """Reboot an instance.
951

952
  @type instance: L{objects.Instance}
953
  @param instance: the instance object to reboot
954
  @type reboot_type: str
955
  @param reboot_type: the type of reboot, one the following
956
    constants:
957
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
958
        instance OS, do not recreate the VM
959
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
960
        restart the VM (at the hypervisor level)
961
      - the other reboot type (L{constants.INSTANCE_REBOOT_HARD})
962
        is not accepted here, since that mode is handled
963
        differently
964
  @rtype: boolean
965
  @return: the success of the operation
966

967
  """
968
  running_instances = GetInstanceList([instance.hypervisor])
969

    
970
  if instance.name not in running_instances:
971
    msg = "Cannot reboot instance %s that is not running" % instance.name
972
    logging.error(msg)
973
    return (False, msg)
974

    
975
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
976
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
977
    try:
978
      hyper.RebootInstance(instance)
979
    except errors.HypervisorError, err:
980
      msg = "Failed to soft reboot instance %s: %s" % (instance.name, err)
981
      logging.error(msg)
982
      return (False, msg)
983
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
984
    try:
985
      stop_result = InstanceShutdown(instance)
986
      if not stop_result[0]:
987
        return stop_result
988
      return StartInstance(instance)
989
    except errors.HypervisorError, err:
990
      msg = "Failed to hard reboot instance %s: %s" % (instance.name, err)
991
      logging.error(msg)
992
      return (False, msg)
993
  else:
994
    return (False, "Invalid reboot_type received: %s" % (reboot_type,))
995

    
996
  return (True, "Reboot successful")
997

    
998

    
999
def MigrationInfo(instance):
1000
  """Gather information about an instance to be migrated.
1001

1002
  @type instance: L{objects.Instance}
1003
  @param instance: the instance definition
1004

1005
  """
1006
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1007
  try:
1008
    info = hyper.MigrationInfo(instance)
1009
  except errors.HypervisorError, err:
1010
    msg = "Failed to fetch migration information"
1011
    logging.exception(msg)
1012
    return (False, '%s: %s' % (msg, err))
1013
  return (True, info)
1014

    
1015

    
1016
def AcceptInstance(instance, info, target):
1017
  """Prepare the node to accept an instance.
1018

1019
  @type instance: L{objects.Instance}
1020
  @param instance: the instance definition
1021
  @type info: string/data (opaque)
1022
  @param info: migration information, from the source node
1023
  @type target: string
1024
  @param target: target host (usually ip), on this node
1025

1026
  """
1027
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1028
  try:
1029
    hyper.AcceptInstance(instance, info, target)
1030
  except errors.HypervisorError, err:
1031
    msg = "Failed to accept instance"
1032
    logging.exception(msg)
1033
    return (False, '%s: %s' % (msg, err))
1034
  return (True, "Accept successfull")
1035

    
1036

    
1037
def FinalizeMigration(instance, info, success):
1038
  """Finalize any preparation to accept an instance.
1039

1040
  @type instance: L{objects.Instance}
1041
  @param instance: the instance definition
1042
  @type info: string/data (opaque)
1043
  @param info: migration information, from the source node
1044
  @type success: boolean
1045
  @param success: whether the migration was a success or a failure
1046

1047
  """
1048
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1049
  try:
1050
    hyper.FinalizeMigration(instance, info, success)
1051
  except errors.HypervisorError, err:
1052
    msg = "Failed to finalize migration"
1053
    logging.exception(msg)
1054
    return (False, '%s: %s' % (msg, err))
1055
  return (True, "Migration Finalized")
1056

    
1057

    
1058
def MigrateInstance(instance, target, live):
1059
  """Migrates an instance to another node.
1060

1061
  @type instance: L{objects.Instance}
1062
  @param instance: the instance definition
1063
  @type target: string
1064
  @param target: the target node name
1065
  @type live: boolean
1066
  @param live: whether the migration should be done live or not (the
1067
      interpretation of this parameter is left to the hypervisor)
1068
  @rtype: tuple
1069
  @return: a tuple of (success, msg) where:
1070
      - succes is a boolean denoting the success/failure of the operation
1071
      - msg is a string with details in case of failure
1072

1073
  """
1074
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1075

    
1076
  try:
1077
    hyper.MigrateInstance(instance.name, target, live)
1078
  except errors.HypervisorError, err:
1079
    msg = "Failed to migrate instance"
1080
    logging.exception(msg)
1081
    return (False, "%s: %s" % (msg, err))
1082
  return (True, "Migration successfull")
1083

    
1084

    
1085
def BlockdevCreate(disk, size, owner, on_primary, info):
1086
  """Creates a block device for an instance.
1087

1088
  @type disk: L{objects.Disk}
1089
  @param disk: the object describing the disk we should create
1090
  @type size: int
1091
  @param size: the size of the physical underlying device, in MiB
1092
  @type owner: str
1093
  @param owner: the name of the instance for which disk is created,
1094
      used for device cache data
1095
  @type on_primary: boolean
1096
  @param on_primary:  indicates if it is the primary node or not
1097
  @type info: string
1098
  @param info: string that will be sent to the physical device
1099
      creation, used for example to set (LVM) tags on LVs
1100

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

1105
  """
1106
  clist = []
1107
  if disk.children:
1108
    for child in disk.children:
1109
      try:
1110
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1111
      except errors.BlockDeviceError, err:
1112
        errmsg = "Can't assemble device %s: %s" % (child, err)
1113
        logging.error(errmsg)
1114
        return False, errmsg
1115
      if on_primary or disk.AssembleOnSecondary():
1116
        # we need the children open in case the device itself has to
1117
        # be assembled
1118
        try:
1119
          crdev.Open()
1120
        except errors.BlockDeviceError, err:
1121
          errmsg = "Can't make child '%s' read-write: %s" % (child, err)
1122
          logging.error(errmsg)
1123
          return False, errmsg
1124
      clist.append(crdev)
1125

    
1126
  try:
1127
    device = bdev.Create(disk.dev_type, disk.physical_id, clist, size)
1128
  except errors.BlockDeviceError, err:
1129
    return False, "Can't create block device: %s" % str(err)
1130

    
1131
  if on_primary or disk.AssembleOnSecondary():
1132
    try:
1133
      device.Assemble()
1134
    except errors.BlockDeviceError, err:
1135
      errmsg = ("Can't assemble device after creation, very"
1136
                " unusual event: %s" % str(err))
1137
      logging.error(errmsg)
1138
      return False, errmsg
1139
    device.SetSyncSpeed(constants.SYNC_SPEED)
1140
    if on_primary or disk.OpenOnSecondary():
1141
      try:
1142
        device.Open(force=True)
1143
      except errors.BlockDeviceError, err:
1144
        errmsg = ("Can't make device r/w after creation, very"
1145
                  " unusual event: %s" % str(err))
1146
        logging.error(errmsg)
1147
        return False, errmsg
1148
    DevCacheManager.UpdateCache(device.dev_path, owner,
1149
                                on_primary, disk.iv_name)
1150

    
1151
  device.SetInfo(info)
1152

    
1153
  physical_id = device.unique_id
1154
  return True, physical_id
1155

    
1156

    
1157
def BlockdevRemove(disk):
1158
  """Remove a block device.
1159

1160
  @note: This is intended to be called recursively.
1161

1162
  @type disk: L{objects.Disk}
1163
  @param disk: the disk object we should remove
1164
  @rtype: boolean
1165
  @return: the success of the operation
1166

1167
  """
1168
  msgs = []
1169
  result = True
1170
  try:
1171
    rdev = _RecursiveFindBD(disk)
1172
  except errors.BlockDeviceError, err:
1173
    # probably can't attach
1174
    logging.info("Can't attach to device %s in remove", disk)
1175
    rdev = None
1176
  if rdev is not None:
1177
    r_path = rdev.dev_path
1178
    try:
1179
      rdev.Remove()
1180
    except errors.BlockDeviceError, err:
1181
      msgs.append(str(err))
1182
      result = False
1183
    if result:
1184
      DevCacheManager.RemoveCache(r_path)
1185

    
1186
  if disk.children:
1187
    for child in disk.children:
1188
      c_status, c_msg = BlockdevRemove(child)
1189
      result = result and c_status
1190
      if c_msg: # not an empty message
1191
        msgs.append(c_msg)
1192

    
1193
  return (result, "; ".join(msgs))
1194

    
1195

    
1196
def _RecursiveAssembleBD(disk, owner, as_primary):
1197
  """Activate a block device for an instance.
1198

1199
  This is run on the primary and secondary nodes for an instance.
1200

1201
  @note: this function is called recursively.
1202

1203
  @type disk: L{objects.Disk}
1204
  @param disk: the disk we try to assemble
1205
  @type owner: str
1206
  @param owner: the name of the instance which owns the disk
1207
  @type as_primary: boolean
1208
  @param as_primary: if we should make the block device
1209
      read/write
1210

1211
  @return: the assembled device or None (in case no device
1212
      was assembled)
1213
  @raise errors.BlockDeviceError: in case there is an error
1214
      during the activation of the children or the device
1215
      itself
1216

1217
  """
1218
  children = []
1219
  if disk.children:
1220
    mcn = disk.ChildrenNeeded()
1221
    if mcn == -1:
1222
      mcn = 0 # max number of Nones allowed
1223
    else:
1224
      mcn = len(disk.children) - mcn # max number of Nones
1225
    for chld_disk in disk.children:
1226
      try:
1227
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1228
      except errors.BlockDeviceError, err:
1229
        if children.count(None) >= mcn:
1230
          raise
1231
        cdev = None
1232
        logging.error("Error in child activation (but continuing): %s",
1233
                      str(err))
1234
      children.append(cdev)
1235

    
1236
  if as_primary or disk.AssembleOnSecondary():
1237
    r_dev = bdev.Assemble(disk.dev_type, disk.physical_id, children)
1238
    r_dev.SetSyncSpeed(constants.SYNC_SPEED)
1239
    result = r_dev
1240
    if as_primary or disk.OpenOnSecondary():
1241
      r_dev.Open()
1242
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1243
                                as_primary, disk.iv_name)
1244

    
1245
  else:
1246
    result = True
1247
  return result
1248

    
1249

    
1250
def BlockdevAssemble(disk, owner, as_primary):
1251
  """Activate a block device for an instance.
1252

1253
  This is a wrapper over _RecursiveAssembleBD.
1254

1255
  @rtype: str or boolean
1256
  @return: a C{/dev/...} path for primary nodes, and
1257
      C{True} for secondary nodes
1258

1259
  """
1260
  status = True
1261
  result = "no error information"
1262
  try:
1263
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1264
    if isinstance(result, bdev.BlockDev):
1265
      result = result.dev_path
1266
  except errors.BlockDeviceError, err:
1267
    result = "Error while assembling disk: %s" % str(err)
1268
    status = False
1269
  return (status, result)
1270

    
1271

    
1272
def BlockdevShutdown(disk):
1273
  """Shut down a block device.
1274

1275
  First, if the device is assembled (Attach() is successfull), then
1276
  the device is shutdown. Then the children of the device are
1277
  shutdown.
1278

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

1283
  @type disk: L{objects.Disk}
1284
  @param disk: the description of the disk we should
1285
      shutdown
1286
  @rtype: boolean
1287
  @return: the success of the operation
1288

1289
  """
1290
  msgs = []
1291
  result = True
1292
  r_dev = _RecursiveFindBD(disk)
1293
  if r_dev is not None:
1294
    r_path = r_dev.dev_path
1295
    try:
1296
      r_dev.Shutdown()
1297
      DevCacheManager.RemoveCache(r_path)
1298
    except errors.BlockDeviceError, err:
1299
      msgs.append(str(err))
1300
      result = False
1301

    
1302
  if disk.children:
1303
    for child in disk.children:
1304
      c_status, c_msg = BlockdevShutdown(child)
1305
      result = result and c_status
1306
      if c_msg: # not an empty message
1307
        msgs.append(c_msg)
1308

    
1309
  return (result, "; ".join(msgs))
1310

    
1311

    
1312
def BlockdevAddchildren(parent_cdev, new_cdevs):
1313
  """Extend a mirrored block device.
1314

1315
  @type parent_cdev: L{objects.Disk}
1316
  @param parent_cdev: the disk to which we should add children
1317
  @type new_cdevs: list of L{objects.Disk}
1318
  @param new_cdevs: the list of children which we should add
1319
  @rtype: boolean
1320
  @return: the success of the operation
1321

1322
  """
1323
  parent_bdev = _RecursiveFindBD(parent_cdev)
1324
  if parent_bdev is None:
1325
    logging.error("Can't find parent device")
1326
    return False
1327
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1328
  if new_bdevs.count(None) > 0:
1329
    logging.error("Can't find new device(s) to add: %s:%s",
1330
                  new_bdevs, new_cdevs)
1331
    return False
1332
  parent_bdev.AddChildren(new_bdevs)
1333
  return True
1334

    
1335

    
1336
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1337
  """Shrink a mirrored block device.
1338

1339
  @type parent_cdev: L{objects.Disk}
1340
  @param parent_cdev: the disk from which we should remove children
1341
  @type new_cdevs: list of L{objects.Disk}
1342
  @param new_cdevs: the list of children which we should remove
1343
  @rtype: boolean
1344
  @return: the success of the operation
1345

1346
  """
1347
  parent_bdev = _RecursiveFindBD(parent_cdev)
1348
  if parent_bdev is None:
1349
    logging.error("Can't find parent in remove children: %s", parent_cdev)
1350
    return False
1351
  devs = []
1352
  for disk in new_cdevs:
1353
    rpath = disk.StaticDevPath()
1354
    if rpath is None:
1355
      bd = _RecursiveFindBD(disk)
1356
      if bd is None:
1357
        logging.error("Can't find dynamic device %s while removing children",
1358
                      disk)
1359
        return False
1360
      else:
1361
        devs.append(bd.dev_path)
1362
    else:
1363
      devs.append(rpath)
1364
  parent_bdev.RemoveChildren(devs)
1365
  return True
1366

    
1367

    
1368
def BlockdevGetmirrorstatus(disks):
1369
  """Get the mirroring status of a list of devices.
1370

1371
  @type disks: list of L{objects.Disk}
1372
  @param disks: the list of disks which we should query
1373
  @rtype: disk
1374
  @return:
1375
      a list of (mirror_done, estimated_time) tuples, which
1376
      are the result of L{bdev.BlockDev.CombinedSyncStatus}
1377
  @raise errors.BlockDeviceError: if any of the disks cannot be
1378
      found
1379

1380
  """
1381
  stats = []
1382
  for dsk in disks:
1383
    rbd = _RecursiveFindBD(dsk)
1384
    if rbd is None:
1385
      raise errors.BlockDeviceError("Can't find device %s" % str(dsk))
1386
    stats.append(rbd.CombinedSyncStatus())
1387
  return stats
1388

    
1389

    
1390
def _RecursiveFindBD(disk):
1391
  """Check if a device is activated.
1392

1393
  If so, return informations about the real device.
1394

1395
  @type disk: L{objects.Disk}
1396
  @param disk: the disk object we need to find
1397

1398
  @return: None if the device can't be found,
1399
      otherwise the device instance
1400

1401
  """
1402
  children = []
1403
  if disk.children:
1404
    for chdisk in disk.children:
1405
      children.append(_RecursiveFindBD(chdisk))
1406

    
1407
  return bdev.FindDevice(disk.dev_type, disk.physical_id, children)
1408

    
1409

    
1410
def BlockdevFind(disk):
1411
  """Check if a device is activated.
1412

1413
  If it is, return informations about the real device.
1414

1415
  @type disk: L{objects.Disk}
1416
  @param disk: the disk to find
1417
  @rtype: None or tuple
1418
  @return: None if the disk cannot be found, otherwise a
1419
      tuple (device_path, major, minor, sync_percent,
1420
      estimated_time, is_degraded)
1421

1422
  """
1423
  try:
1424
    rbd = _RecursiveFindBD(disk)
1425
  except errors.BlockDeviceError, err:
1426
    return (False, str(err))
1427
  if rbd is None:
1428
    return (True, None)
1429
  return (True, (rbd.dev_path, rbd.major, rbd.minor) + rbd.GetSyncStatus())
1430

    
1431

    
1432
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
1433
  """Write a file to the filesystem.
1434

1435
  This allows the master to overwrite(!) a file. It will only perform
1436
  the operation if the file belongs to a list of configuration files.
1437

1438
  @type file_name: str
1439
  @param file_name: the target file name
1440
  @type data: str
1441
  @param data: the new contents of the file
1442
  @type mode: int
1443
  @param mode: the mode to give the file (can be None)
1444
  @type uid: int
1445
  @param uid: the owner of the file (can be -1 for default)
1446
  @type gid: int
1447
  @param gid: the group of the file (can be -1 for default)
1448
  @type atime: float
1449
  @param atime: the atime to set on the file (can be None)
1450
  @type mtime: float
1451
  @param mtime: the mtime to set on the file (can be None)
1452
  @rtype: boolean
1453
  @return: the success of the operation; errors are logged
1454
      in the node daemon log
1455

1456
  """
1457
  if not os.path.isabs(file_name):
1458
    logging.error("Filename passed to UploadFile is not absolute: '%s'",
1459
                  file_name)
1460
    return False
1461

    
1462
  allowed_files = [
1463
    constants.CLUSTER_CONF_FILE,
1464
    constants.ETC_HOSTS,
1465
    constants.SSH_KNOWN_HOSTS_FILE,
1466
    constants.VNC_PASSWORD_FILE,
1467
    ]
1468

    
1469
  if file_name not in allowed_files:
1470
    logging.error("Filename passed to UploadFile not in allowed"
1471
                 " upload targets: '%s'", file_name)
1472
    return False
1473

    
1474
  raw_data = _Decompress(data)
1475

    
1476
  utils.WriteFile(file_name, data=raw_data, mode=mode, uid=uid, gid=gid,
1477
                  atime=atime, mtime=mtime)
1478
  return True
1479

    
1480

    
1481
def WriteSsconfFiles(values):
1482
  """Update all ssconf files.
1483

1484
  Wrapper around the SimpleStore.WriteFiles.
1485

1486
  """
1487
  ssconf.SimpleStore().WriteFiles(values)
1488

    
1489

    
1490
def _ErrnoOrStr(err):
1491
  """Format an EnvironmentError exception.
1492

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

1497
  @type err: L{EnvironmentError}
1498
  @param err: the exception to format
1499

1500
  """
1501
  if hasattr(err, 'errno'):
1502
    detail = errno.errorcode[err.errno]
1503
  else:
1504
    detail = str(err)
1505
  return detail
1506

    
1507

    
1508
def _OSOndiskVersion(name, os_dir):
1509
  """Compute and return the API version of a given OS.
1510

1511
  This function will try to read the API version of the OS given by
1512
  the 'name' parameter and residing in the 'os_dir' directory.
1513

1514
  @type name: str
1515
  @param name: the OS name we should look for
1516
  @type os_dir: str
1517
  @param os_dir: the directory inwhich we should look for the OS
1518
  @rtype: int or None
1519
  @return:
1520
      Either an integer denoting the version or None in the
1521
      case when this is not a valid OS name.
1522
  @raise errors.InvalidOS: if the OS cannot be found
1523

1524
  """
1525
  api_file = os.path.sep.join([os_dir, "ganeti_api_version"])
1526

    
1527
  try:
1528
    st = os.stat(api_file)
1529
  except EnvironmentError, err:
1530
    raise errors.InvalidOS(name, os_dir, "'ganeti_api_version' file not"
1531
                           " found (%s)" % _ErrnoOrStr(err))
1532

    
1533
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1534
    raise errors.InvalidOS(name, os_dir, "'ganeti_api_version' file is not"
1535
                           " a regular file")
1536

    
1537
  try:
1538
    f = open(api_file)
1539
    try:
1540
      api_versions = f.readlines()
1541
    finally:
1542
      f.close()
1543
  except EnvironmentError, err:
1544
    raise errors.InvalidOS(name, os_dir, "error while reading the"
1545
                           " API version (%s)" % _ErrnoOrStr(err))
1546

    
1547
  api_versions = [version.strip() for version in api_versions]
1548
  try:
1549
    api_versions = [int(version) for version in api_versions]
1550
  except (TypeError, ValueError), err:
1551
    raise errors.InvalidOS(name, os_dir,
1552
                           "API version is not integer (%s)" % str(err))
1553

    
1554
  return api_versions
1555

    
1556

    
1557
def DiagnoseOS(top_dirs=None):
1558
  """Compute the validity for all OSes.
1559

1560
  @type top_dirs: list
1561
  @param top_dirs: the list of directories in which to
1562
      search (if not given defaults to
1563
      L{constants.OS_SEARCH_PATH})
1564
  @rtype: list of L{objects.OS}
1565
  @return: an OS object for each name in all the given
1566
      directories
1567

1568
  """
1569
  if top_dirs is None:
1570
    top_dirs = constants.OS_SEARCH_PATH
1571

    
1572
  result = []
1573
  for dir_name in top_dirs:
1574
    if os.path.isdir(dir_name):
1575
      try:
1576
        f_names = utils.ListVisibleFiles(dir_name)
1577
      except EnvironmentError, err:
1578
        logging.exception("Can't list the OS directory %s", dir_name)
1579
        break
1580
      for name in f_names:
1581
        try:
1582
          os_inst = OSFromDisk(name, base_dir=dir_name)
1583
          result.append(os_inst)
1584
        except errors.InvalidOS, err:
1585
          result.append(objects.OS.FromInvalidOS(err))
1586

    
1587
  return result
1588

    
1589

    
1590
def OSFromDisk(name, base_dir=None):
1591
  """Create an OS instance from disk.
1592

1593
  This function will return an OS instance if the given name is a
1594
  valid OS name. Otherwise, it will raise an appropriate
1595
  L{errors.InvalidOS} exception, detailing why this is not a valid OS.
1596

1597
  @type base_dir: string
1598
  @keyword base_dir: Base directory containing OS installations.
1599
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
1600
  @rtype: L{objects.OS}
1601
  @return: the OS instance if we find a valid one
1602
  @raise errors.InvalidOS: if we don't find a valid OS
1603

1604
  """
1605
  if base_dir is None:
1606
    os_dir = utils.FindFile(name, constants.OS_SEARCH_PATH, os.path.isdir)
1607
    if os_dir is None:
1608
      raise errors.InvalidOS(name, None, "OS dir not found in search path")
1609
  else:
1610
    os_dir = os.path.sep.join([base_dir, name])
1611

    
1612
  api_versions = _OSOndiskVersion(name, os_dir)
1613

    
1614
  if constants.OS_API_VERSION not in api_versions:
1615
    raise errors.InvalidOS(name, os_dir, "API version mismatch"
1616
                           " (found %s want %s)"
1617
                           % (api_versions, constants.OS_API_VERSION))
1618

    
1619
  # OS Scripts dictionary, we will populate it with the actual script names
1620
  os_scripts = dict.fromkeys(constants.OS_SCRIPTS)
1621

    
1622
  for script in os_scripts:
1623
    os_scripts[script] = os.path.sep.join([os_dir, script])
1624

    
1625
    try:
1626
      st = os.stat(os_scripts[script])
1627
    except EnvironmentError, err:
1628
      raise errors.InvalidOS(name, os_dir, "'%s' script missing (%s)" %
1629
                             (script, _ErrnoOrStr(err)))
1630

    
1631
    if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
1632
      raise errors.InvalidOS(name, os_dir, "'%s' script not executable" %
1633
                             script)
1634

    
1635
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1636
      raise errors.InvalidOS(name, os_dir, "'%s' is not a regular file" %
1637
                             script)
1638

    
1639

    
1640
  return objects.OS(name=name, path=os_dir, status=constants.OS_VALID_STATUS,
1641
                    create_script=os_scripts[constants.OS_SCRIPT_CREATE],
1642
                    export_script=os_scripts[constants.OS_SCRIPT_EXPORT],
1643
                    import_script=os_scripts[constants.OS_SCRIPT_IMPORT],
1644
                    rename_script=os_scripts[constants.OS_SCRIPT_RENAME],
1645
                    api_versions=api_versions)
1646

    
1647
def OSEnvironment(instance, debug=0):
1648
  """Calculate the environment for an os script.
1649

1650
  @type instance: L{objects.Instance}
1651
  @param instance: target instance for the os script run
1652
  @type debug: integer
1653
  @param debug: debug level (0 or 1, for OS Api 10)
1654
  @rtype: dict
1655
  @return: dict of environment variables
1656
  @raise errors.BlockDeviceError: if the block device
1657
      cannot be found
1658

1659
  """
1660
  result = {}
1661
  result['OS_API_VERSION'] = '%d' % constants.OS_API_VERSION
1662
  result['INSTANCE_NAME'] = instance.name
1663
  result['INSTANCE_OS'] = instance.os
1664
  result['HYPERVISOR'] = instance.hypervisor
1665
  result['DISK_COUNT'] = '%d' % len(instance.disks)
1666
  result['NIC_COUNT'] = '%d' % len(instance.nics)
1667
  result['DEBUG_LEVEL'] = '%d' % debug
1668
  for idx, disk in enumerate(instance.disks):
1669
    real_disk = _RecursiveFindBD(disk)
1670
    if real_disk is None:
1671
      raise errors.BlockDeviceError("Block device '%s' is not set up" %
1672
                                    str(disk))
1673
    real_disk.Open()
1674
    result['DISK_%d_PATH' % idx] = real_disk.dev_path
1675
    result['DISK_%d_ACCESS' % idx] = disk.mode
1676
    if constants.HV_DISK_TYPE in instance.hvparams:
1677
      result['DISK_%d_FRONTEND_TYPE' % idx] = \
1678
        instance.hvparams[constants.HV_DISK_TYPE]
1679
    if disk.dev_type in constants.LDS_BLOCK:
1680
      result['DISK_%d_BACKEND_TYPE' % idx] = 'block'
1681
    elif disk.dev_type == constants.LD_FILE:
1682
      result['DISK_%d_BACKEND_TYPE' % idx] = \
1683
        'file:%s' % disk.physical_id[0]
1684
  for idx, nic in enumerate(instance.nics):
1685
    result['NIC_%d_MAC' % idx] = nic.mac
1686
    if nic.ip:
1687
      result['NIC_%d_IP' % idx] = nic.ip
1688
    result['NIC_%d_BRIDGE' % idx] = nic.bridge
1689
    if constants.HV_NIC_TYPE in instance.hvparams:
1690
      result['NIC_%d_FRONTEND_TYPE' % idx] = \
1691
        instance.hvparams[constants.HV_NIC_TYPE]
1692

    
1693
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
1694
    for key, value in source.items():
1695
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
1696

    
1697
  return result
1698

    
1699
def BlockdevGrow(disk, amount):
1700
  """Grow a stack of block devices.
1701

1702
  This function is called recursively, with the childrens being the
1703
  first ones to resize.
1704

1705
  @type disk: L{objects.Disk}
1706
  @param disk: the disk to be grown
1707
  @rtype: (status, result)
1708
  @return: a tuple with the status of the operation
1709
      (True/False), and the errors message if status
1710
      is False
1711

1712
  """
1713
  r_dev = _RecursiveFindBD(disk)
1714
  if r_dev is None:
1715
    return False, "Cannot find block device %s" % (disk,)
1716

    
1717
  try:
1718
    r_dev.Grow(amount)
1719
  except errors.BlockDeviceError, err:
1720
    return False, str(err)
1721

    
1722
  return True, None
1723

    
1724

    
1725
def BlockdevSnapshot(disk):
1726
  """Create a snapshot copy of a block device.
1727

1728
  This function is called recursively, and the snapshot is actually created
1729
  just for the leaf lvm backend device.
1730

1731
  @type disk: L{objects.Disk}
1732
  @param disk: the disk to be snapshotted
1733
  @rtype: string
1734
  @return: snapshot disk path
1735

1736
  """
1737
  if disk.children:
1738
    if len(disk.children) == 1:
1739
      # only one child, let's recurse on it
1740
      return BlockdevSnapshot(disk.children[0])
1741
    else:
1742
      # more than one child, choose one that matches
1743
      for child in disk.children:
1744
        if child.size == disk.size:
1745
          # return implies breaking the loop
1746
          return BlockdevSnapshot(child)
1747
  elif disk.dev_type == constants.LD_LV:
1748
    r_dev = _RecursiveFindBD(disk)
1749
    if r_dev is not None:
1750
      # let's stay on the safe side and ask for the full size, for now
1751
      return r_dev.Snapshot(disk.size)
1752
    else:
1753
      return None
1754
  else:
1755
    raise errors.ProgrammerError("Cannot snapshot non-lvm block device"
1756
                                 " '%s' of type '%s'" %
1757
                                 (disk.unique_id, disk.dev_type))
1758

    
1759

    
1760
def ExportSnapshot(disk, dest_node, instance, cluster_name, idx):
1761
  """Export a block device snapshot to a remote node.
1762

1763
  @type disk: L{objects.Disk}
1764
  @param disk: the description of the disk to export
1765
  @type dest_node: str
1766
  @param dest_node: the destination node to export to
1767
  @type instance: L{objects.Instance}
1768
  @param instance: the instance object to whom the disk belongs
1769
  @type cluster_name: str
1770
  @param cluster_name: the cluster name, needed for SSH hostalias
1771
  @type idx: int
1772
  @param idx: the index of the disk in the instance's disk list,
1773
      used to export to the OS scripts environment
1774
  @rtype: boolean
1775
  @return: the success of the operation
1776

1777
  """
1778
  export_env = OSEnvironment(instance)
1779

    
1780
  inst_os = OSFromDisk(instance.os)
1781
  export_script = inst_os.export_script
1782

    
1783
  logfile = "%s/exp-%s-%s-%s.log" % (constants.LOG_OS_DIR, inst_os.name,
1784
                                     instance.name, int(time.time()))
1785
  if not os.path.exists(constants.LOG_OS_DIR):
1786
    os.mkdir(constants.LOG_OS_DIR, 0750)
1787
  real_disk = _RecursiveFindBD(disk)
1788
  if real_disk is None:
1789
    raise errors.BlockDeviceError("Block device '%s' is not set up" %
1790
                                  str(disk))
1791
  real_disk.Open()
1792

    
1793
  export_env['EXPORT_DEVICE'] = real_disk.dev_path
1794
  export_env['EXPORT_INDEX'] = str(idx)
1795

    
1796
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
1797
  destfile = disk.physical_id[1]
1798

    
1799
  # the target command is built out of three individual commands,
1800
  # which are joined by pipes; we check each individual command for
1801
  # valid parameters
1802
  expcmd = utils.BuildShellCmd("cd %s; %s 2>%s", inst_os.path,
1803
                               export_script, logfile)
1804

    
1805
  comprcmd = "gzip"
1806

    
1807
  destcmd = utils.BuildShellCmd("mkdir -p %s && cat > %s/%s",
1808
                                destdir, destdir, destfile)
1809
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
1810
                                                   constants.GANETI_RUNAS,
1811
                                                   destcmd)
1812

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

    
1816
  result = utils.RunCmd(command, env=export_env)
1817

    
1818
  if result.failed:
1819
    logging.error("os snapshot export command '%s' returned error: %s"
1820
                  " output: %s", command, result.fail_reason, result.output)
1821
    return False
1822

    
1823
  return True
1824

    
1825

    
1826
def FinalizeExport(instance, snap_disks):
1827
  """Write out the export configuration information.
1828

1829
  @type instance: L{objects.Instance}
1830
  @param instance: the instance which we export, used for
1831
      saving configuration
1832
  @type snap_disks: list of L{objects.Disk}
1833
  @param snap_disks: list of snapshot block devices, which
1834
      will be used to get the actual name of the dump file
1835

1836
  @rtype: boolean
1837
  @return: the success of the operation
1838

1839
  """
1840
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
1841
  finaldestdir = os.path.join(constants.EXPORT_DIR, instance.name)
1842

    
1843
  config = objects.SerializableConfigParser()
1844

    
1845
  config.add_section(constants.INISECT_EXP)
1846
  config.set(constants.INISECT_EXP, 'version', '0')
1847
  config.set(constants.INISECT_EXP, 'timestamp', '%d' % int(time.time()))
1848
  config.set(constants.INISECT_EXP, 'source', instance.primary_node)
1849
  config.set(constants.INISECT_EXP, 'os', instance.os)
1850
  config.set(constants.INISECT_EXP, 'compression', 'gzip')
1851

    
1852
  config.add_section(constants.INISECT_INS)
1853
  config.set(constants.INISECT_INS, 'name', instance.name)
1854
  config.set(constants.INISECT_INS, 'memory', '%d' %
1855
             instance.beparams[constants.BE_MEMORY])
1856
  config.set(constants.INISECT_INS, 'vcpus', '%d' %
1857
             instance.beparams[constants.BE_VCPUS])
1858
  config.set(constants.INISECT_INS, 'disk_template', instance.disk_template)
1859

    
1860
  nic_total = 0
1861
  for nic_count, nic in enumerate(instance.nics):
1862
    nic_total += 1
1863
    config.set(constants.INISECT_INS, 'nic%d_mac' %
1864
               nic_count, '%s' % nic.mac)
1865
    config.set(constants.INISECT_INS, 'nic%d_ip' % nic_count, '%s' % nic.ip)
1866
    config.set(constants.INISECT_INS, 'nic%d_bridge' % nic_count,
1867
               '%s' % nic.bridge)
1868
  # TODO: redundant: on load can read nics until it doesn't exist
1869
  config.set(constants.INISECT_INS, 'nic_count' , '%d' % nic_total)
1870

    
1871
  disk_total = 0
1872
  for disk_count, disk in enumerate(snap_disks):
1873
    if disk:
1874
      disk_total += 1
1875
      config.set(constants.INISECT_INS, 'disk%d_ivname' % disk_count,
1876
                 ('%s' % disk.iv_name))
1877
      config.set(constants.INISECT_INS, 'disk%d_dump' % disk_count,
1878
                 ('%s' % disk.physical_id[1]))
1879
      config.set(constants.INISECT_INS, 'disk%d_size' % disk_count,
1880
                 ('%d' % disk.size))
1881

    
1882
  config.set(constants.INISECT_INS, 'disk_count' , '%d' % disk_total)
1883

    
1884
  utils.WriteFile(os.path.join(destdir, constants.EXPORT_CONF_FILE),
1885
                  data=config.Dumps())
1886
  shutil.rmtree(finaldestdir, True)
1887
  shutil.move(destdir, finaldestdir)
1888

    
1889
  return True
1890

    
1891

    
1892
def ExportInfo(dest):
1893
  """Get export configuration information.
1894

1895
  @type dest: str
1896
  @param dest: directory containing the export
1897

1898
  @rtype: L{objects.SerializableConfigParser}
1899
  @return: a serializable config file containing the
1900
      export info
1901

1902
  """
1903
  cff = os.path.join(dest, constants.EXPORT_CONF_FILE)
1904

    
1905
  config = objects.SerializableConfigParser()
1906
  config.read(cff)
1907

    
1908
  if (not config.has_section(constants.INISECT_EXP) or
1909
      not config.has_section(constants.INISECT_INS)):
1910
    return None
1911

    
1912
  return config
1913

    
1914

    
1915
def ImportOSIntoInstance(instance, src_node, src_images, cluster_name):
1916
  """Import an os image into an instance.
1917

1918
  @type instance: L{objects.Instance}
1919
  @param instance: instance to import the disks into
1920
  @type src_node: string
1921
  @param src_node: source node for the disk images
1922
  @type src_images: list of string
1923
  @param src_images: absolute paths of the disk images
1924
  @rtype: list of boolean
1925
  @return: each boolean represent the success of importing the n-th disk
1926

1927
  """
1928
  import_env = OSEnvironment(instance)
1929
  inst_os = OSFromDisk(instance.os)
1930
  import_script = inst_os.import_script
1931

    
1932
  logfile = "%s/import-%s-%s-%s.log" % (constants.LOG_OS_DIR, instance.os,
1933
                                        instance.name, int(time.time()))
1934
  if not os.path.exists(constants.LOG_OS_DIR):
1935
    os.mkdir(constants.LOG_OS_DIR, 0750)
1936

    
1937
  comprcmd = "gunzip"
1938
  impcmd = utils.BuildShellCmd("(cd %s; %s >%s 2>&1)", inst_os.path,
1939
                               import_script, logfile)
1940

    
1941
  final_result = []
1942
  for idx, image in enumerate(src_images):
1943
    if image:
1944
      destcmd = utils.BuildShellCmd('cat %s', image)
1945
      remotecmd = _GetSshRunner(cluster_name).BuildCmd(src_node,
1946
                                                       constants.GANETI_RUNAS,
1947
                                                       destcmd)
1948
      command = '|'.join([utils.ShellQuoteArgs(remotecmd), comprcmd, impcmd])
1949
      import_env['IMPORT_DEVICE'] = import_env['DISK_%d_PATH' % idx]
1950
      import_env['IMPORT_INDEX'] = str(idx)
1951
      result = utils.RunCmd(command, env=import_env)
1952
      if result.failed:
1953
        logging.error("Disk import command '%s' returned error: %s"
1954
                      " output: %s", command, result.fail_reason,
1955
                      result.output)
1956
        final_result.append(False)
1957
      else:
1958
        final_result.append(True)
1959
    else:
1960
      final_result.append(True)
1961

    
1962
  return final_result
1963

    
1964

    
1965
def ListExports():
1966
  """Return a list of exports currently available on this machine.
1967

1968
  @rtype: list
1969
  @return: list of the exports
1970

1971
  """
1972
  if os.path.isdir(constants.EXPORT_DIR):
1973
    return utils.ListVisibleFiles(constants.EXPORT_DIR)
1974
  else:
1975
    return []
1976

    
1977

    
1978
def RemoveExport(export):
1979
  """Remove an existing export from the node.
1980

1981
  @type export: str
1982
  @param export: the name of the export to remove
1983
  @rtype: boolean
1984
  @return: the success of the operation
1985

1986
  """
1987
  target = os.path.join(constants.EXPORT_DIR, export)
1988

    
1989
  shutil.rmtree(target)
1990
  # TODO: catch some of the relevant exceptions and provide a pretty
1991
  # error message if rmtree fails.
1992

    
1993
  return True
1994

    
1995

    
1996
def BlockdevRename(devlist):
1997
  """Rename a list of block devices.
1998

1999
  @type devlist: list of tuples
2000
  @param devlist: list of tuples of the form  (disk,
2001
      new_logical_id, new_physical_id); disk is an
2002
      L{objects.Disk} object describing the current disk,
2003
      and new logical_id/physical_id is the name we
2004
      rename it to
2005
  @rtype: boolean
2006
  @return: True if all renames succeeded, False otherwise
2007

2008
  """
2009
  result = True
2010
  for disk, unique_id in devlist:
2011
    dev = _RecursiveFindBD(disk)
2012
    if dev is None:
2013
      result = False
2014
      continue
2015
    try:
2016
      old_rpath = dev.dev_path
2017
      dev.Rename(unique_id)
2018
      new_rpath = dev.dev_path
2019
      if old_rpath != new_rpath:
2020
        DevCacheManager.RemoveCache(old_rpath)
2021
        # FIXME: we should add the new cache information here, like:
2022
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2023
        # but we don't have the owner here - maybe parse from existing
2024
        # cache? for now, we only lose lvm data when we rename, which
2025
        # is less critical than DRBD or MD
2026
    except errors.BlockDeviceError, err:
2027
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2028
      result = False
2029
  return result
2030

    
2031

    
2032
def _TransformFileStorageDir(file_storage_dir):
2033
  """Checks whether given file_storage_dir is valid.
2034

2035
  Checks wheter the given file_storage_dir is within the cluster-wide
2036
  default file_storage_dir stored in SimpleStore. Only paths under that
2037
  directory are allowed.
2038

2039
  @type file_storage_dir: str
2040
  @param file_storage_dir: the path to check
2041

2042
  @return: the normalized path if valid, None otherwise
2043

2044
  """
2045
  cfg = _GetConfig()
2046
  file_storage_dir = os.path.normpath(file_storage_dir)
2047
  base_file_storage_dir = cfg.GetFileStorageDir()
2048
  if (not os.path.commonprefix([file_storage_dir, base_file_storage_dir]) ==
2049
      base_file_storage_dir):
2050
    logging.error("file storage directory '%s' is not under base file"
2051
                  " storage directory '%s'",
2052
                  file_storage_dir, base_file_storage_dir)
2053
    return None
2054
  return file_storage_dir
2055

    
2056

    
2057
def CreateFileStorageDir(file_storage_dir):
2058
  """Create file storage directory.
2059

2060
  @type file_storage_dir: str
2061
  @param file_storage_dir: directory to create
2062

2063
  @rtype: tuple
2064
  @return: tuple with first element a boolean indicating wheter dir
2065
      creation was successful or not
2066

2067
  """
2068
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2069
  result = True,
2070
  if not file_storage_dir:
2071
    result = False,
2072
  else:
2073
    if os.path.exists(file_storage_dir):
2074
      if not os.path.isdir(file_storage_dir):
2075
        logging.error("'%s' is not a directory", file_storage_dir)
2076
        result = False,
2077
    else:
2078
      try:
2079
        os.makedirs(file_storage_dir, 0750)
2080
      except OSError, err:
2081
        logging.error("Cannot create file storage directory '%s': %s",
2082
                      file_storage_dir, err)
2083
        result = False,
2084
  return result
2085

    
2086

    
2087
def RemoveFileStorageDir(file_storage_dir):
2088
  """Remove file storage directory.
2089

2090
  Remove it only if it's empty. If not log an error and return.
2091

2092
  @type file_storage_dir: str
2093
  @param file_storage_dir: the directory we should cleanup
2094
  @rtype: tuple (success,)
2095
  @return: tuple of one element, C{success}, denoting
2096
      whether the operation was successfull
2097

2098
  """
2099
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2100
  result = True,
2101
  if not file_storage_dir:
2102
    result = False,
2103
  else:
2104
    if os.path.exists(file_storage_dir):
2105
      if not os.path.isdir(file_storage_dir):
2106
        logging.error("'%s' is not a directory", file_storage_dir)
2107
        result = False,
2108
      # deletes dir only if empty, otherwise we want to return False
2109
      try:
2110
        os.rmdir(file_storage_dir)
2111
      except OSError, err:
2112
        logging.exception("Cannot remove file storage directory '%s'",
2113
                          file_storage_dir)
2114
        result = False,
2115
  return result
2116

    
2117

    
2118
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2119
  """Rename the file storage directory.
2120

2121
  @type old_file_storage_dir: str
2122
  @param old_file_storage_dir: the current path
2123
  @type new_file_storage_dir: str
2124
  @param new_file_storage_dir: the name we should rename to
2125
  @rtype: tuple (success,)
2126
  @return: tuple of one element, C{success}, denoting
2127
      whether the operation was successful
2128

2129
  """
2130
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2131
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2132
  result = True,
2133
  if not old_file_storage_dir or not new_file_storage_dir:
2134
    result = False,
2135
  else:
2136
    if not os.path.exists(new_file_storage_dir):
2137
      if os.path.isdir(old_file_storage_dir):
2138
        try:
2139
          os.rename(old_file_storage_dir, new_file_storage_dir)
2140
        except OSError, err:
2141
          logging.exception("Cannot rename '%s' to '%s'",
2142
                            old_file_storage_dir, new_file_storage_dir)
2143
          result =  False,
2144
      else:
2145
        logging.error("'%s' is not a directory", old_file_storage_dir)
2146
        result = False,
2147
    else:
2148
      if os.path.exists(old_file_storage_dir):
2149
        logging.error("Cannot rename '%s' to '%s'. Both locations exist.",
2150
                      old_file_storage_dir, new_file_storage_dir)
2151
        result = False,
2152
  return result
2153

    
2154

    
2155
def _IsJobQueueFile(file_name):
2156
  """Checks whether the given filename is in the queue directory.
2157

2158
  @type file_name: str
2159
  @param file_name: the file name we should check
2160
  @rtype: boolean
2161
  @return: whether the file is under the queue directory
2162

2163
  """
2164
  queue_dir = os.path.normpath(constants.QUEUE_DIR)
2165
  result = (os.path.commonprefix([queue_dir, file_name]) == queue_dir)
2166

    
2167
  if not result:
2168
    logging.error("'%s' is not a file in the queue directory",
2169
                  file_name)
2170

    
2171
  return result
2172

    
2173

    
2174
def JobQueueUpdate(file_name, content):
2175
  """Updates a file in the queue directory.
2176

2177
  This is just a wrapper over L{utils.WriteFile}, with proper
2178
  checking.
2179

2180
  @type file_name: str
2181
  @param file_name: the job file name
2182
  @type content: str
2183
  @param content: the new job contents
2184
  @rtype: boolean
2185
  @return: the success of the operation
2186

2187
  """
2188
  if not _IsJobQueueFile(file_name):
2189
    return False
2190

    
2191
  # Write and replace the file atomically
2192
  utils.WriteFile(file_name, data=_Decompress(content))
2193

    
2194
  return True
2195

    
2196

    
2197
def JobQueueRename(old, new):
2198
  """Renames a job queue file.
2199

2200
  This is just a wrapper over os.rename with proper checking.
2201

2202
  @type old: str
2203
  @param old: the old (actual) file name
2204
  @type new: str
2205
  @param new: the desired file name
2206
  @rtype: boolean
2207
  @return: the success of the operation
2208

2209
  """
2210
  if not (_IsJobQueueFile(old) and _IsJobQueueFile(new)):
2211
    return False
2212

    
2213
  utils.RenameFile(old, new, mkdir=True)
2214

    
2215
  return True
2216

    
2217

    
2218
def JobQueueSetDrainFlag(drain_flag):
2219
  """Set the drain flag for the queue.
2220

2221
  This will set or unset the queue drain flag.
2222

2223
  @type drain_flag: boolean
2224
  @param drain_flag: if True, will set the drain flag, otherwise reset it.
2225
  @rtype: boolean
2226
  @return: always True
2227
  @warning: the function always returns True
2228

2229
  """
2230
  if drain_flag:
2231
    utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
2232
  else:
2233
    utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
2234

    
2235
  return True
2236

    
2237

    
2238
def BlockdevClose(instance_name, disks):
2239
  """Closes the given block devices.
2240

2241
  This means they will be switched to secondary mode (in case of
2242
  DRBD).
2243

2244
  @param instance_name: if the argument is not empty, the symlinks
2245
      of this instance will be removed
2246
  @type disks: list of L{objects.Disk}
2247
  @param disks: the list of disks to be closed
2248
  @rtype: tuple (success, message)
2249
  @return: a tuple of success and message, where success
2250
      indicates the succes of the operation, and message
2251
      which will contain the error details in case we
2252
      failed
2253

2254
  """
2255
  bdevs = []
2256
  for cf in disks:
2257
    rd = _RecursiveFindBD(cf)
2258
    if rd is None:
2259
      return (False, "Can't find device %s" % cf)
2260
    bdevs.append(rd)
2261

    
2262
  msg = []
2263
  for rd in bdevs:
2264
    try:
2265
      rd.Close()
2266
    except errors.BlockDeviceError, err:
2267
      msg.append(str(err))
2268
  if msg:
2269
    return (False, "Can't make devices secondary: %s" % ",".join(msg))
2270
  else:
2271
    if instance_name:
2272
      _RemoveBlockDevLinks(instance_name, disks)
2273
    return (True, "All devices secondary")
2274

    
2275

    
2276
def ValidateHVParams(hvname, hvparams):
2277
  """Validates the given hypervisor parameters.
2278

2279
  @type hvname: string
2280
  @param hvname: the hypervisor name
2281
  @type hvparams: dict
2282
  @param hvparams: the hypervisor parameters to be validated
2283
  @rtype: tuple (success, message)
2284
  @return: a tuple of success and message, where success
2285
      indicates the succes of the operation, and message
2286
      which will contain the error details in case we
2287
      failed
2288

2289
  """
2290
  try:
2291
    hv_type = hypervisor.GetHypervisor(hvname)
2292
    hv_type.ValidateParameters(hvparams)
2293
    return (True, "Validation passed")
2294
  except errors.HypervisorError, err:
2295
    return (False, str(err))
2296

    
2297

    
2298
def DemoteFromMC():
2299
  """Demotes the current node from master candidate role.
2300

2301
  """
2302
  # try to ensure we're not the master by mistake
2303
  master, myself = ssconf.GetMasterAndMyself()
2304
  if master == myself:
2305
    return (False, "ssconf status shows I'm the master node, will not demote")
2306
  pid_file = utils.DaemonPidFileName(constants.MASTERD_PID)
2307
  if utils.IsProcessAlive(utils.ReadPidFile(pid_file)):
2308
    return (False, "The master daemon is running, will not demote")
2309
  try:
2310
    utils.CreateBackup(constants.CLUSTER_CONF_FILE)
2311
  except EnvironmentError, err:
2312
    if err.errno != errno.ENOENT:
2313
      return (False, "Error while backing up cluster file: %s" % str(err))
2314
  utils.RemoveFile(constants.CLUSTER_CONF_FILE)
2315
  return (True, "Done")
2316

    
2317

    
2318
def _FindDisks(nodes_ip, disks):
2319
  """Sets the physical ID on disks and returns the block devices.
2320

2321
  """
2322
  # set the correct physical ID
2323
  my_name = utils.HostInfo().name
2324
  for cf in disks:
2325
    cf.SetPhysicalID(my_name, nodes_ip)
2326

    
2327
  bdevs = []
2328

    
2329
  for cf in disks:
2330
    rd = _RecursiveFindBD(cf)
2331
    if rd is None:
2332
      return (False, "Can't find device %s" % cf)
2333
    bdevs.append(rd)
2334
  return (True, bdevs)
2335

    
2336

    
2337
def DrbdDisconnectNet(nodes_ip, disks):
2338
  """Disconnects the network on a list of drbd devices.
2339

2340
  """
2341
  status, bdevs = _FindDisks(nodes_ip, disks)
2342
  if not status:
2343
    return status, bdevs
2344

    
2345
  # disconnect disks
2346
  for rd in bdevs:
2347
    try:
2348
      rd.DisconnectNet()
2349
    except errors.BlockDeviceError, err:
2350
      logging.exception("Failed to go into standalone mode")
2351
      return (False, "Can't change network configuration: %s" % str(err))
2352
  return (True, "All disks are now disconnected")
2353

    
2354

    
2355
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
2356
  """Attaches the network on a list of drbd devices.
2357

2358
  """
2359
  status, bdevs = _FindDisks(nodes_ip, disks)
2360
  if not status:
2361
    return status, bdevs
2362

    
2363
  if multimaster:
2364
    for idx, rd in enumerate(bdevs):
2365
      try:
2366
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
2367
      except EnvironmentError, err:
2368
        return (False, "Can't create symlink: %s" % str(err))
2369
  # reconnect disks, switch to new master configuration and if
2370
  # needed primary mode
2371
  for rd in bdevs:
2372
    try:
2373
      rd.AttachNet(multimaster)
2374
    except errors.BlockDeviceError, err:
2375
      return (False, "Can't change network configuration: %s" % str(err))
2376
  # wait until the disks are connected; we need to retry the re-attach
2377
  # if the device becomes standalone, as this might happen if the one
2378
  # node disconnects and reconnects in a different mode before the
2379
  # other node reconnects; in this case, one or both of the nodes will
2380
  # decide it has wrong configuration and switch to standalone
2381
  RECONNECT_TIMEOUT = 2 * 60
2382
  sleep_time = 0.100 # start with 100 miliseconds
2383
  timeout_limit = time.time() + RECONNECT_TIMEOUT
2384
  while time.time() < timeout_limit:
2385
    all_connected = True
2386
    for rd in bdevs:
2387
      stats = rd.GetProcStatus()
2388
      if not (stats.is_connected or stats.is_in_resync):
2389
        all_connected = False
2390
      if stats.is_standalone:
2391
        # peer had different config info and this node became
2392
        # standalone, even though this should not happen with the
2393
        # new staged way of changing disk configs
2394
        try:
2395
          rd.ReAttachNet(multimaster)
2396
        except errors.BlockDeviceError, err:
2397
          return (False, "Can't change network configuration: %s" % str(err))
2398
    if all_connected:
2399
      break
2400
    time.sleep(sleep_time)
2401
    sleep_time = min(5, sleep_time * 1.5)
2402
  if not all_connected:
2403
    return (False, "Timeout in disk reconnecting")
2404
  if multimaster:
2405
    # change to primary mode
2406
    for rd in bdevs:
2407
      try:
2408
        rd.Open()
2409
      except errors.BlockDeviceError, err:
2410
        return (False, "Can't change to primary mode: %s" % str(err))
2411
  if multimaster:
2412
    msg = "multi-master and primary"
2413
  else:
2414
    msg = "single-master"
2415
  return (True, "Disks are now configured as %s" % msg)
2416

    
2417

    
2418
def DrbdWaitSync(nodes_ip, disks):
2419
  """Wait until DRBDs have synchronized.
2420

2421
  """
2422
  status, bdevs = _FindDisks(nodes_ip, disks)
2423
  if not status:
2424
    return status, bdevs
2425

    
2426
  min_resync = 100
2427
  alldone = True
2428
  failure = False
2429
  for rd in bdevs:
2430
    stats = rd.GetProcStatus()
2431
    if not (stats.is_connected or stats.is_in_resync):
2432
      failure = True
2433
      break
2434
    alldone = alldone and (not stats.is_in_resync)
2435
    if stats.sync_percent is not None:
2436
      min_resync = min(min_resync, stats.sync_percent)
2437
  return (not failure, (alldone, min_resync))
2438

    
2439

    
2440
class HooksRunner(object):
2441
  """Hook runner.
2442

2443
  This class is instantiated on the node side (ganeti-noded) and not
2444
  on the master side.
2445

2446
  """
2447
  RE_MASK = re.compile("^[a-zA-Z0-9_-]+$")
2448

    
2449
  def __init__(self, hooks_base_dir=None):
2450
    """Constructor for hooks runner.
2451

2452
    @type hooks_base_dir: str or None
2453
    @param hooks_base_dir: if not None, this overrides the
2454
        L{constants.HOOKS_BASE_DIR} (useful for unittests)
2455

2456
    """
2457
    if hooks_base_dir is None:
2458
      hooks_base_dir = constants.HOOKS_BASE_DIR
2459
    self._BASE_DIR = hooks_base_dir
2460

    
2461
  @staticmethod
2462
  def ExecHook(script, env):
2463
    """Exec one hook script.
2464

2465
    @type script: str
2466
    @param script: the full path to the script
2467
    @type env: dict
2468
    @param env: the environment with which to exec the script
2469
    @rtype: tuple (success, message)
2470
    @return: a tuple of success and message, where success
2471
        indicates the succes of the operation, and message
2472
        which will contain the error details in case we
2473
        failed
2474

2475
    """
2476
    # exec the process using subprocess and log the output
2477
    fdstdin = None
2478
    try:
2479
      fdstdin = open("/dev/null", "r")
2480
      child = subprocess.Popen([script], stdin=fdstdin, stdout=subprocess.PIPE,
2481
                               stderr=subprocess.STDOUT, close_fds=True,
2482
                               shell=False, cwd="/", env=env)
2483
      output = ""
2484
      try:
2485
        output = child.stdout.read(4096)
2486
        child.stdout.close()
2487
      except EnvironmentError, err:
2488
        output += "Hook script error: %s" % str(err)
2489

    
2490
      while True:
2491
        try:
2492
          result = child.wait()
2493
          break
2494
        except EnvironmentError, err:
2495
          if err.errno == errno.EINTR:
2496
            continue
2497
          raise
2498
    finally:
2499
      # try not to leak fds
2500
      for fd in (fdstdin, ):
2501
        if fd is not None:
2502
          try:
2503
            fd.close()
2504
          except EnvironmentError, err:
2505
            # just log the error
2506
            #logging.exception("Error while closing fd %s", fd)
2507
            pass
2508

    
2509
    return result == 0, utils.SafeEncode(output.strip())
2510

    
2511
  def RunHooks(self, hpath, phase, env):
2512
    """Run the scripts in the hooks directory.
2513

2514
    @type hpath: str
2515
    @param hpath: the path to the hooks directory which
2516
        holds the scripts
2517
    @type phase: str
2518
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
2519
        L{constants.HOOKS_PHASE_POST}
2520
    @type env: dict
2521
    @param env: dictionary with the environment for the hook
2522
    @rtype: list
2523
    @return: list of 3-element tuples:
2524
      - script path
2525
      - script result, either L{constants.HKR_SUCCESS} or
2526
        L{constants.HKR_FAIL}
2527
      - output of the script
2528

2529
    @raise errors.ProgrammerError: for invalid input
2530
        parameters
2531

2532
    """
2533
    if phase == constants.HOOKS_PHASE_PRE:
2534
      suffix = "pre"
2535
    elif phase == constants.HOOKS_PHASE_POST:
2536
      suffix = "post"
2537
    else:
2538
      raise errors.ProgrammerError("Unknown hooks phase: '%s'" % phase)
2539
    rr = []
2540

    
2541
    subdir = "%s-%s.d" % (hpath, suffix)
2542
    dir_name = "%s/%s" % (self._BASE_DIR, subdir)
2543
    try:
2544
      dir_contents = utils.ListVisibleFiles(dir_name)
2545
    except OSError, err:
2546
      # FIXME: must log output in case of failures
2547
      return rr
2548

    
2549
    # we use the standard python sort order,
2550
    # so 00name is the recommended naming scheme
2551
    dir_contents.sort()
2552
    for relname in dir_contents:
2553
      fname = os.path.join(dir_name, relname)
2554
      if not (os.path.isfile(fname) and os.access(fname, os.X_OK) and
2555
          self.RE_MASK.match(relname) is not None):
2556
        rrval = constants.HKR_SKIP
2557
        output = ""
2558
      else:
2559
        result, output = self.ExecHook(fname, env)
2560
        if not result:
2561
          rrval = constants.HKR_FAIL
2562
        else:
2563
          rrval = constants.HKR_SUCCESS
2564
      rr.append(("%s/%s" % (subdir, relname), rrval, output))
2565

    
2566
    return rr
2567

    
2568

    
2569
class IAllocatorRunner(object):
2570
  """IAllocator runner.
2571

2572
  This class is instantiated on the node side (ganeti-noded) and not on
2573
  the master side.
2574

2575
  """
2576
  def Run(self, name, idata):
2577
    """Run an iallocator script.
2578

2579
    @type name: str
2580
    @param name: the iallocator script name
2581
    @type idata: str
2582
    @param idata: the allocator input data
2583

2584
    @rtype: tuple
2585
    @return: four element tuple of:
2586
       - run status (one of the IARUN_ constants)
2587
       - stdout
2588
       - stderr
2589
       - fail reason (as from L{utils.RunResult})
2590

2591
    """
2592
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
2593
                                  os.path.isfile)
2594
    if alloc_script is None:
2595
      return (constants.IARUN_NOTFOUND, None, None, None)
2596

    
2597
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
2598
    try:
2599
      os.write(fd, idata)
2600
      os.close(fd)
2601
      result = utils.RunCmd([alloc_script, fin_name])
2602
      if result.failed:
2603
        return (constants.IARUN_FAILURE, result.stdout, result.stderr,
2604
                result.fail_reason)
2605
    finally:
2606
      os.unlink(fin_name)
2607

    
2608
    return (constants.IARUN_SUCCESS, result.stdout, result.stderr, None)
2609

    
2610

    
2611
class DevCacheManager(object):
2612
  """Simple class for managing a cache of block device information.
2613

2614
  """
2615
  _DEV_PREFIX = "/dev/"
2616
  _ROOT_DIR = constants.BDEV_CACHE_DIR
2617

    
2618
  @classmethod
2619
  def _ConvertPath(cls, dev_path):
2620
    """Converts a /dev/name path to the cache file name.
2621

2622
    This replaces slashes with underscores and strips the /dev
2623
    prefix. It then returns the full path to the cache file.
2624

2625
    @type dev_path: str
2626
    @param dev_path: the C{/dev/} path name
2627
    @rtype: str
2628
    @return: the converted path name
2629

2630
    """
2631
    if dev_path.startswith(cls._DEV_PREFIX):
2632
      dev_path = dev_path[len(cls._DEV_PREFIX):]
2633
    dev_path = dev_path.replace("/", "_")
2634
    fpath = "%s/bdev_%s" % (cls._ROOT_DIR, dev_path)
2635
    return fpath
2636

    
2637
  @classmethod
2638
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
2639
    """Updates the cache information for a given device.
2640

2641
    @type dev_path: str
2642
    @param dev_path: the pathname of the device
2643
    @type owner: str
2644
    @param owner: the owner (instance name) of the device
2645
    @type on_primary: bool
2646
    @param on_primary: whether this is the primary
2647
        node nor not
2648
    @type iv_name: str
2649
    @param iv_name: the instance-visible name of the
2650
        device, as in objects.Disk.iv_name
2651

2652
    @rtype: None
2653

2654
    """
2655
    if dev_path is None:
2656
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
2657
      return
2658
    fpath = cls._ConvertPath(dev_path)
2659
    if on_primary:
2660
      state = "primary"
2661
    else:
2662
      state = "secondary"
2663
    if iv_name is None:
2664
      iv_name = "not_visible"
2665
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
2666
    try:
2667
      utils.WriteFile(fpath, data=fdata)
2668
    except EnvironmentError, err:
2669
      logging.exception("Can't update bdev cache for %s", dev_path)
2670

    
2671
  @classmethod
2672
  def RemoveCache(cls, dev_path):
2673
    """Remove data for a dev_path.
2674

2675
    This is just a wrapper over L{utils.RemoveFile} with a converted
2676
    path name and logging.
2677

2678
    @type dev_path: str
2679
    @param dev_path: the pathname of the device
2680

2681
    @rtype: None
2682

2683
    """
2684
    if dev_path is None:
2685
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
2686
      return
2687
    fpath = cls._ConvertPath(dev_path)
2688
    try:
2689
      utils.RemoveFile(fpath)
2690
    except EnvironmentError, err:
2691
      logging.exception("Can't update bdev cache for %s", dev_path)