Cluster: add nicparams, and update them on upgrade
[ganeti-local] / daemons / ganeti-noded
1 #!/usr/bin/python
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 """Ganeti node daemon"""
23
24 # functions in this module need to have a given name structure, so:
25 # pylint: disable-msg=C0103
26
27 import os
28 import sys
29 import traceback
30 import SocketServer
31 import errno
32 import logging
33 import signal
34
35 from optparse import OptionParser
36
37 from ganeti import backend
38 from ganeti import constants
39 from ganeti import objects
40 from ganeti import errors
41 from ganeti import jstore
42 from ganeti import daemon
43 from ganeti import http
44 from ganeti import utils
45
46 import ganeti.http.server
47
48
49 queue_lock = None
50
51
52 def _RequireJobQueueLock(fn):
53   """Decorator for job queue manipulating functions.
54
55   """
56   QUEUE_LOCK_TIMEOUT = 10
57
58   def wrapper(*args, **kwargs):
59     # Locking in exclusive, blocking mode because there could be several
60     # children running at the same time. Waiting up to 10 seconds.
61     queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT)
62     try:
63       return fn(*args, **kwargs)
64     finally:
65       queue_lock.Unlock()
66
67   return wrapper
68
69
70 class NodeHttpServer(http.server.HttpServer):
71   """The server implementation.
72
73   This class holds all methods exposed over the RPC interface.
74
75   """
76   def __init__(self, *args, **kwargs):
77     http.server.HttpServer.__init__(self, *args, **kwargs)
78     self.noded_pid = os.getpid()
79
80   def HandleRequest(self, req):
81     """Handle a request.
82
83     """
84     if req.request_method.upper() != http.HTTP_PUT:
85       raise http.HttpBadRequest()
86
87     path = req.request_path
88     if path.startswith("/"):
89       path = path[1:]
90
91     method = getattr(self, "perspective_%s" % path, None)
92     if method is None:
93       raise http.HttpNotFound()
94
95     try:
96       try:
97         return method(req.request_body)
98       except:
99         logging.exception("Error in RPC call")
100         raise
101     except errors.QuitGanetiException, err:
102       # Tell parent to quit
103       os.kill(self.noded_pid, signal.SIGTERM)
104
105   # the new block devices  --------------------------
106
107   @staticmethod
108   def perspective_blockdev_create(params):
109     """Create a block device.
110
111     """
112     bdev_s, size, owner, on_primary, info = params
113     bdev = objects.Disk.FromDict(bdev_s)
114     if bdev is None:
115       raise ValueError("can't unserialize data!")
116     return backend.BlockdevCreate(bdev, size, owner, on_primary, info)
117
118   @staticmethod
119   def perspective_blockdev_remove(params):
120     """Remove a block device.
121
122     """
123     bdev_s = params[0]
124     bdev = objects.Disk.FromDict(bdev_s)
125     return backend.BlockdevRemove(bdev)
126
127   @staticmethod
128   def perspective_blockdev_rename(params):
129     """Remove a block device.
130
131     """
132     devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params]
133     return backend.BlockdevRename(devlist)
134
135   @staticmethod
136   def perspective_blockdev_assemble(params):
137     """Assemble a block device.
138
139     """
140     bdev_s, owner, on_primary = params
141     bdev = objects.Disk.FromDict(bdev_s)
142     if bdev is None:
143       raise ValueError("can't unserialize data!")
144     return backend.BlockdevAssemble(bdev, owner, on_primary)
145
146   @staticmethod
147   def perspective_blockdev_shutdown(params):
148     """Shutdown a block device.
149
150     """
151     bdev_s = params[0]
152     bdev = objects.Disk.FromDict(bdev_s)
153     if bdev is None:
154       raise ValueError("can't unserialize data!")
155     return backend.BlockdevShutdown(bdev)
156
157   @staticmethod
158   def perspective_blockdev_addchildren(params):
159     """Add a child to a mirror device.
160
161     Note: this is only valid for mirror devices. It's the caller's duty
162     to send a correct disk, otherwise we raise an error.
163
164     """
165     bdev_s, ndev_s = params
166     bdev = objects.Disk.FromDict(bdev_s)
167     ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
168     if bdev is None or ndevs.count(None) > 0:
169       raise ValueError("can't unserialize data!")
170     return backend.BlockdevAddchildren(bdev, ndevs)
171
172   @staticmethod
173   def perspective_blockdev_removechildren(params):
174     """Remove a child from a mirror device.
175
176     This is only valid for mirror devices, of course. It's the callers
177     duty to send a correct disk, otherwise we raise an error.
178
179     """
180     bdev_s, ndev_s = params
181     bdev = objects.Disk.FromDict(bdev_s)
182     ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
183     if bdev is None or ndevs.count(None) > 0:
184       raise ValueError("can't unserialize data!")
185     return backend.BlockdevRemovechildren(bdev, ndevs)
186
187   @staticmethod
188   def perspective_blockdev_getmirrorstatus(params):
189     """Return the mirror status for a list of disks.
190
191     """
192     disks = [objects.Disk.FromDict(dsk_s)
193             for dsk_s in params]
194     return backend.BlockdevGetmirrorstatus(disks)
195
196   @staticmethod
197   def perspective_blockdev_find(params):
198     """Expose the FindBlockDevice functionality for a disk.
199
200     This will try to find but not activate a disk.
201
202     """
203     disk = objects.Disk.FromDict(params[0])
204     return backend.BlockdevFind(disk)
205
206   @staticmethod
207   def perspective_blockdev_snapshot(params):
208     """Create a snapshot device.
209
210     Note that this is only valid for LVM disks, if we get passed
211     something else we raise an exception. The snapshot device can be
212     remove by calling the generic block device remove call.
213
214     """
215     cfbd = objects.Disk.FromDict(params[0])
216     return backend.BlockdevSnapshot(cfbd)
217
218   @staticmethod
219   def perspective_blockdev_grow(params):
220     """Grow a stack of devices.
221
222     """
223     cfbd = objects.Disk.FromDict(params[0])
224     amount = params[1]
225     return backend.BlockdevGrow(cfbd, amount)
226
227   @staticmethod
228   def perspective_blockdev_close(params):
229     """Closes the given block devices.
230
231     """
232     disks = [objects.Disk.FromDict(cf) for cf in params[1]]
233     return backend.BlockdevClose(params[0], disks)
234
235   # blockdev/drbd specific methods ----------
236
237   @staticmethod
238   def perspective_drbd_disconnect_net(params):
239     """Disconnects the network connection of drbd disks.
240
241     Note that this is only valid for drbd disks, so the members of the
242     disk list must all be drbd devices.
243
244     """
245     nodes_ip, disks = params
246     disks = [objects.Disk.FromDict(cf) for cf in disks]
247     return backend.DrbdDisconnectNet(nodes_ip, disks)
248
249   @staticmethod
250   def perspective_drbd_attach_net(params):
251     """Attaches the network connection of drbd disks.
252
253     Note that this is only valid for drbd disks, so the members of the
254     disk list must all be drbd devices.
255
256     """
257     nodes_ip, disks, instance_name, multimaster = params
258     disks = [objects.Disk.FromDict(cf) for cf in disks]
259     return backend.DrbdAttachNet(nodes_ip, disks,
260                                      instance_name, multimaster)
261
262   @staticmethod
263   def perspective_drbd_wait_sync(params):
264     """Wait until DRBD disks are synched.
265
266     Note that this is only valid for drbd disks, so the members of the
267     disk list must all be drbd devices.
268
269     """
270     nodes_ip, disks = params
271     disks = [objects.Disk.FromDict(cf) for cf in disks]
272     return backend.DrbdWaitSync(nodes_ip, disks)
273
274   # export/import  --------------------------
275
276   @staticmethod
277   def perspective_snapshot_export(params):
278     """Export a given snapshot.
279
280     """
281     disk = objects.Disk.FromDict(params[0])
282     dest_node = params[1]
283     instance = objects.Instance.FromDict(params[2])
284     cluster_name = params[3]
285     dev_idx = params[4]
286     return backend.ExportSnapshot(disk, dest_node, instance,
287                                   cluster_name, dev_idx)
288
289   @staticmethod
290   def perspective_finalize_export(params):
291     """Expose the finalize export functionality.
292
293     """
294     instance = objects.Instance.FromDict(params[0])
295     snap_disks = [objects.Disk.FromDict(str_data)
296                   for str_data in params[1]]
297     return backend.FinalizeExport(instance, snap_disks)
298
299   @staticmethod
300   def perspective_export_info(params):
301     """Query information about an existing export on this node.
302
303     The given path may not contain an export, in which case we return
304     None.
305
306     """
307     path = params[0]
308     einfo = backend.ExportInfo(path)
309     if einfo is None:
310       return einfo
311     return einfo.Dumps()
312
313   @staticmethod
314   def perspective_export_list(params):
315     """List the available exports on this node.
316
317     Note that as opposed to export_info, which may query data about an
318     export in any path, this only queries the standard Ganeti path
319     (constants.EXPORT_DIR).
320
321     """
322     return backend.ListExports()
323
324   @staticmethod
325   def perspective_export_remove(params):
326     """Remove an export.
327
328     """
329     export = params[0]
330     return backend.RemoveExport(export)
331
332   # volume  --------------------------
333
334   @staticmethod
335   def perspective_volume_list(params):
336     """Query the list of logical volumes in a given volume group.
337
338     """
339     vgname = params[0]
340     return backend.GetVolumeList(vgname)
341
342   @staticmethod
343   def perspective_vg_list(params):
344     """Query the list of volume groups.
345
346     """
347     return backend.ListVolumeGroups()
348
349   # bridge  --------------------------
350
351   @staticmethod
352   def perspective_bridges_exist(params):
353     """Check if all bridges given exist on this node.
354
355     """
356     bridges_list = params[0]
357     return backend.BridgesExist(bridges_list)
358
359   # instance  --------------------------
360
361   @staticmethod
362   def perspective_instance_os_add(params):
363     """Install an OS on a given instance.
364
365     """
366     inst_s = params[0]
367     inst = objects.Instance.FromDict(inst_s)
368     reinstall = params[1]
369     return backend.InstanceOsAdd(inst, reinstall)
370
371   @staticmethod
372   def perspective_instance_run_rename(params):
373     """Runs the OS rename script for an instance.
374
375     """
376     inst_s, old_name = params
377     inst = objects.Instance.FromDict(inst_s)
378     return backend.RunRenameInstance(inst, old_name)
379
380   @staticmethod
381   def perspective_instance_os_import(params):
382     """Run the import function of an OS onto a given instance.
383
384     """
385     inst_s, src_node, src_images, cluster_name = params
386     inst = objects.Instance.FromDict(inst_s)
387     return backend.ImportOSIntoInstance(inst, src_node, src_images,
388                                         cluster_name)
389
390   @staticmethod
391   def perspective_instance_shutdown(params):
392     """Shutdown an instance.
393
394     """
395     instance = objects.Instance.FromDict(params[0])
396     return backend.InstanceShutdown(instance)
397
398   @staticmethod
399   def perspective_instance_start(params):
400     """Start an instance.
401
402     """
403     instance = objects.Instance.FromDict(params[0])
404     return backend.StartInstance(instance)
405
406   @staticmethod
407   def perspective_migration_info(params):
408     """Gather information about an instance to be migrated.
409
410     """
411     instance = objects.Instance.FromDict(params[0])
412     return backend.MigrationInfo(instance)
413
414   @staticmethod
415   def perspective_accept_instance(params):
416     """Prepare the node to accept an instance.
417
418     """
419     instance, info, target = params
420     instance = objects.Instance.FromDict(instance)
421     return backend.AcceptInstance(instance, info, target)
422
423   @staticmethod
424   def perspective_finalize_migration(params):
425     """Finalize the instance migration.
426
427     """
428     instance, info, success = params
429     instance = objects.Instance.FromDict(instance)
430     return backend.FinalizeMigration(instance, info, success)
431
432   @staticmethod
433   def perspective_instance_migrate(params):
434     """Migrates an instance.
435
436     """
437     instance, target, live = params
438     instance = objects.Instance.FromDict(instance)
439     return backend.MigrateInstance(instance, target, live)
440
441   @staticmethod
442   def perspective_instance_reboot(params):
443     """Reboot an instance.
444
445     """
446     instance = objects.Instance.FromDict(params[0])
447     reboot_type = params[1]
448     return backend.InstanceReboot(instance, reboot_type)
449
450   @staticmethod
451   def perspective_instance_info(params):
452     """Query instance information.
453
454     """
455     return backend.GetInstanceInfo(params[0], params[1])
456
457   @staticmethod
458   def perspective_instance_migratable(params):
459     """Query whether the specified instance can be migrated.
460
461     """
462     instance = objects.Instance.FromDict(params[0])
463     return backend.GetInstanceMigratable(instance)
464
465   @staticmethod
466   def perspective_all_instances_info(params):
467     """Query information about all instances.
468
469     """
470     return backend.GetAllInstancesInfo(params[0])
471
472   @staticmethod
473   def perspective_instance_list(params):
474     """Query the list of running instances.
475
476     """
477     return backend.GetInstanceList(params[0])
478
479   # node --------------------------
480
481   @staticmethod
482   def perspective_node_tcp_ping(params):
483     """Do a TcpPing on the remote node.
484
485     """
486     return utils.TcpPing(params[1], params[2], timeout=params[3],
487                          live_port_needed=params[4], source=params[0])
488
489   @staticmethod
490   def perspective_node_has_ip_address(params):
491     """Checks if a node has the given ip address.
492
493     """
494     return utils.OwnIpAddress(params[0])
495
496   @staticmethod
497   def perspective_node_info(params):
498     """Query node information.
499
500     """
501     vgname, hypervisor_type = params
502     return backend.GetNodeInfo(vgname, hypervisor_type)
503
504   @staticmethod
505   def perspective_node_add(params):
506     """Complete the registration of this node in the cluster.
507
508     """
509     return backend.AddNode(params[0], params[1], params[2],
510                            params[3], params[4], params[5])
511
512   @staticmethod
513   def perspective_node_verify(params):
514     """Run a verify sequence on this node.
515
516     """
517     return backend.VerifyNode(params[0], params[1])
518
519   @staticmethod
520   def perspective_node_start_master(params):
521     """Promote this node to master status.
522
523     """
524     return backend.StartMaster(params[0])
525
526   @staticmethod
527   def perspective_node_stop_master(params):
528     """Demote this node from master status.
529
530     """
531     return backend.StopMaster(params[0])
532
533   @staticmethod
534   def perspective_node_leave_cluster(params):
535     """Cleanup after leaving a cluster.
536
537     """
538     return backend.LeaveCluster()
539
540   @staticmethod
541   def perspective_node_volumes(params):
542     """Query the list of all logical volume groups.
543
544     """
545     return backend.NodeVolumes()
546
547   @staticmethod
548   def perspective_node_demote_from_mc(params):
549     """Demote a node from the master candidate role.
550
551     """
552     return backend.DemoteFromMC()
553
554
555   @staticmethod
556   def perspective_node_powercycle(params):
557     """Tries to powercycle the nod.
558
559     """
560     hypervisor_type = params[0]
561     return backend.PowercycleNode(hypervisor_type)
562
563
564   # cluster --------------------------
565
566   @staticmethod
567   def perspective_version(params):
568     """Query version information.
569
570     """
571     return constants.PROTOCOL_VERSION
572
573   @staticmethod
574   def perspective_upload_file(params):
575     """Upload a file.
576
577     Note that the backend implementation imposes strict rules on which
578     files are accepted.
579
580     """
581     return backend.UploadFile(*params)
582
583   @staticmethod
584   def perspective_master_info(params):
585     """Query master information.
586
587     """
588     return backend.GetMasterInfo()
589
590   @staticmethod
591   def perspective_write_ssconf_files(params):
592     """Write ssconf files.
593
594     """
595     (values,) = params
596     return backend.WriteSsconfFiles(values)
597
598   # os -----------------------
599
600   @staticmethod
601   def perspective_os_diagnose(params):
602     """Query detailed information about existing OSes.
603
604     """
605     return [os_obj.ToDict() for os_obj in backend.DiagnoseOS()]
606
607   @staticmethod
608   def perspective_os_get(params):
609     """Query information about a given OS.
610
611     """
612     name = params[0]
613     try:
614       os_obj = backend.OSFromDisk(name)
615     except errors.InvalidOS, err:
616       os_obj = objects.OS.FromInvalidOS(err)
617     return os_obj.ToDict()
618
619   # hooks -----------------------
620
621   @staticmethod
622   def perspective_hooks_runner(params):
623     """Run hook scripts.
624
625     """
626     hpath, phase, env = params
627     hr = backend.HooksRunner()
628     return hr.RunHooks(hpath, phase, env)
629
630   # iallocator -----------------
631
632   @staticmethod
633   def perspective_iallocator_runner(params):
634     """Run an iallocator script.
635
636     """
637     name, idata = params
638     iar = backend.IAllocatorRunner()
639     return iar.Run(name, idata)
640
641   # test -----------------------
642
643   @staticmethod
644   def perspective_test_delay(params):
645     """Run test delay.
646
647     """
648     duration = params[0]
649     return utils.TestDelay(duration)
650
651   # file storage ---------------
652
653   @staticmethod
654   def perspective_file_storage_dir_create(params):
655     """Create the file storage directory.
656
657     """
658     file_storage_dir = params[0]
659     return backend.CreateFileStorageDir(file_storage_dir)
660
661   @staticmethod
662   def perspective_file_storage_dir_remove(params):
663     """Remove the file storage directory.
664
665     """
666     file_storage_dir = params[0]
667     return backend.RemoveFileStorageDir(file_storage_dir)
668
669   @staticmethod
670   def perspective_file_storage_dir_rename(params):
671     """Rename the file storage directory.
672
673     """
674     old_file_storage_dir = params[0]
675     new_file_storage_dir = params[1]
676     return backend.RenameFileStorageDir(old_file_storage_dir,
677                                         new_file_storage_dir)
678
679   # jobs ------------------------
680
681   @staticmethod
682   @_RequireJobQueueLock
683   def perspective_jobqueue_update(params):
684     """Update job queue.
685
686     """
687     (file_name, content) = params
688     return backend.JobQueueUpdate(file_name, content)
689
690   @staticmethod
691   @_RequireJobQueueLock
692   def perspective_jobqueue_purge(params):
693     """Purge job queue.
694
695     """
696     return backend.JobQueuePurge()
697
698   @staticmethod
699   @_RequireJobQueueLock
700   def perspective_jobqueue_rename(params):
701     """Rename a job queue file.
702
703     """
704     # TODO: What if a file fails to rename?
705     return [backend.JobQueueRename(old, new) for old, new in params]
706
707   @staticmethod
708   def perspective_jobqueue_set_drain(params):
709     """Set/unset the queue drain flag.
710
711     """
712     drain_flag = params[0]
713     return backend.JobQueueSetDrainFlag(drain_flag)
714
715
716   # hypervisor ---------------
717
718   @staticmethod
719   def perspective_hypervisor_validate_params(params):
720     """Validate the hypervisor parameters.
721
722     """
723     (hvname, hvparams) = params
724     return backend.ValidateHVParams(hvname, hvparams)
725
726
727 def ParseOptions():
728   """Parse the command line options.
729
730   @return: (options, args) as from OptionParser.parse_args()
731
732   """
733   parser = OptionParser(description="Ganeti node daemon",
734                         usage="%prog [-f] [-d] [-b ADDRESS]",
735                         version="%%prog (ganeti) %s" %
736                         constants.RELEASE_VERSION)
737
738   parser.add_option("-f", "--foreground", dest="fork",
739                     help="Don't detach from the current terminal",
740                     default=True, action="store_false")
741   parser.add_option("-d", "--debug", dest="debug",
742                     help="Enable some debug messages",
743                     default=False, action="store_true")
744   parser.add_option("-b", "--bind", dest="bind_address",
745                     help="Bind address",
746                     default="", metavar="ADDRESS")
747
748   options, args = parser.parse_args()
749   return options, args
750
751
752 def main():
753   """Main function for the node daemon.
754
755   """
756   global queue_lock
757
758   options, args = ParseOptions()
759   utils.debug = options.debug
760
761   if options.fork:
762     utils.CloseFDs()
763
764   for fname in (constants.SSL_CERT_FILE,):
765     if not os.path.isfile(fname):
766       print "config %s not there, will not run." % fname
767       sys.exit(5)
768
769   try:
770     port = utils.GetNodeDaemonPort()
771   except errors.ConfigurationError, err:
772     print "Cluster configuration incomplete: '%s'" % str(err)
773     sys.exit(5)
774
775   dirs = [(val, constants.RUN_DIRS_MODE) for val in constants.SUB_RUN_DIRS]
776   dirs.append((constants.LOG_OS_DIR, 0750))
777   dirs.append((constants.LOCK_DIR, 1777))
778   utils.EnsureDirs(dirs)
779
780   # become a daemon
781   if options.fork:
782     utils.Daemonize(logfile=constants.LOG_NODESERVER)
783
784   utils.WritePidFile(constants.NODED_PID)
785   try:
786     utils.SetupLogging(logfile=constants.LOG_NODESERVER, debug=options.debug,
787                        stderr_logging=not options.fork)
788     logging.info("ganeti node daemon startup")
789
790     # Read SSL certificate
791     ssl_params = http.HttpSslParams(ssl_key_path=constants.SSL_CERT_FILE,
792                                     ssl_cert_path=constants.SSL_CERT_FILE)
793
794     # Prepare job queue
795     queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
796
797     mainloop = daemon.Mainloop()
798     server = NodeHttpServer(mainloop, options.bind_address, port,
799                             ssl_params=ssl_params, ssl_verify_peer=True)
800     server.Start()
801     try:
802       mainloop.Run()
803     finally:
804       server.Stop()
805   finally:
806     utils.RemovePidFile(constants.NODED_PID)
807
808
809 if __name__ == '__main__':
810   main()