X-Git-Url: http://git.annexia.org/?a=blobdiff_plain;f=slave.ml;h=01105ef8700c143eadb89f83c684a3ccadb0f8f4;hb=f09bb82de01019f24411cac2916d9567b5e9a235;hp=1017dd89624a0ae1c856441575cc0ee7080bde10;hpb=bbfe03c47f1d7f03c3e6c0cab9e4f500f588c80a;p=guestfs-browser.git diff --git a/slave.ml b/slave.ml index 1017dd8..01105ef 100644 --- a/slave.ml +++ b/slave.ml @@ -16,8 +16,15 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. *) +open ExtList +open ExtString +open CamomileLibrary +open Default.Camomile + open Utils +open Printf + module C = Libvirt.Connect module Cond = Condition module D = Libvirt.Domain @@ -30,8 +37,14 @@ type 'a callback = 'a -> unit (* The commands. *) type command = | Exit_thread - | Connect of string option * unit callback - | Get_domains of domain list callback + | Connect of string option * domain list callback + | Disk_usage of source * string * int64 callback + | Download_dir_find0 of source * string * string * unit callback + | Download_dir_tarball of source * string * download_dir_tarball_format * string * unit callback + | Download_file of source * string * string * unit callback + | Open_domain of string * inspection_data callback + | Open_images of (string * string option) list * inspection_data callback + | Read_directory of source * string * direntry list callback and domain = { dom_id : int; @@ -39,15 +52,96 @@ and domain = { dom_state : D.state; } +and inspection_data = { + insp_all_filesystems : (string * string) list; + insp_oses : inspection_os list; +} + +and inspection_os = { + insp_root : string; + insp_arch : string; + insp_distro : string; + insp_filesystems : string array; + insp_hostname : string; + insp_major_version : int; + insp_minor_version : int; + insp_mountpoints : (string * string) list; + insp_package_format : string; + insp_package_management : string; + insp_product_name : string; + insp_type : string; + insp_windows_systemroot : string option; + insp_winreg_DEFAULT : string option; + insp_winreg_SAM : string option; + insp_winreg_SECURITY : string option; + insp_winreg_SOFTWARE : string option; + insp_winreg_SYSTEM : string option; +} + +and source = OS of inspection_os | Volume of string + +and direntry = { + dent_name : string; + dent_stat : G.stat; + dent_link : string; +} + +and download_dir_tarball_format = Tar | TGZ | TXZ + +let rec string_of_command = function + | Exit_thread -> "Exit_thread" + | Connect (Some name, _) -> sprintf "Connect %s" name + | Connect (None, _) -> "Connect NULL" + | Disk_usage (src, remotedir, _) -> + sprintf "Disk_usage (%s, %s)" (string_of_source src) remotedir + | Download_dir_find0 (src, remotedir, localfile, _) -> + sprintf "Download_dir_find0 (%s, %s, %s)" + (string_of_source src) remotedir localfile + | Download_dir_tarball (src, remotedir, format, localfile, _) -> + sprintf "Download_dir_tarball (%s, %s, %s, %s)" + (string_of_source src) remotedir + (string_of_download_dir_tarball_format format) localfile + | Download_file (src, remotefile, localfile, _) -> + sprintf "Download_file (%s, %s, %s)" + (string_of_source src) remotefile localfile + | Open_domain (name, _) -> sprintf "Open_domain %s" name + | Open_images (images, _) -> + sprintf "Open_images %s" (string_of_images images) + | Read_directory (src, dir, _) -> + sprintf "Read_directory (%s, %s)" (string_of_source src) dir + +and string_of_images images = + "[" ^ + String.concat "; " + (List.map (function + | fn, None -> fn + | fn, Some format -> sprintf "%s (%s)" fn format) + images) ^ "]" + +and string_of_source = function + | OS { insp_root = root } -> + sprintf "OS %s" root + | Volume dev -> + sprintf "Volume %s" dev + +and string_of_download_dir_tarball_format = function + | Tar -> "Tar" + | TGZ -> "TGZ" + | TXZ -> "TXZ" + let no_callback _ = () let failure_hook = ref (fun _ -> ()) let busy_hook = ref (fun _ -> ()) let idle_hook = ref (fun _ -> ()) +let status_hook = ref (fun _ -> ()) +let progress_hook = ref (fun _ -> ()) let set_failure_hook cb = failure_hook := cb let set_busy_hook cb = busy_hook := cb let set_idle_hook cb = idle_hook := cb +let set_status_hook cb = status_hook := cb +let set_progress_hook cb = progress_hook := cb (* Execute a function, while holding a mutex. If the function * fails, ensure we release the mutex before rethrowing the @@ -63,90 +157,577 @@ let with_lock m f = (* The queue of commands, and a lock and condition to protect it. *) let q = Q.create () +let q_discard = ref false let q_lock = M.create () let q_cond = Cond.create () (* Send a command message to the slave thread. *) -let send_to_slave cmd = - debug "sending message %s to slave thread ..." (string_of_command cmd) +let send_to_slave ?fail cmd = + debug "sending message %s to slave thread ..." (string_of_command cmd); with_lock q_lock ( fun () -> - Q.push cmd q; + Q.push (fail, cmd) q; Cond.signal q_cond ) -let discard_command_queue () = with_lock q_lock (fun () -> Q.clear q) +let discard_command_queue () = + with_lock q_lock ( + fun () -> + Q.clear q; + (* Discard the currently running command. *) + q_discard := true + ) -let connect uri cb = send_to_slave (Connect (uri, cb)) -let get_domains cb = send_to_slave (Get_domains cb) +let connect ?fail uri cb = send_to_slave ?fail (Connect (uri, cb)) +let disk_usage ?fail src remotedir cb = + send_to_slave ?fail (Disk_usage (src, remotedir, cb)) +let download_dir_find0 ?fail src remotedir localfile cb = + send_to_slave ?fail (Download_dir_find0 (src, remotedir, localfile, cb)) +let download_dir_tarball ?fail src remotedir format localfile cb = + send_to_slave ?fail + (Download_dir_tarball (src, remotedir, format, localfile, cb)) +let download_file ?fail src remotefile localfile cb = + send_to_slave ?fail (Download_file (src, remotefile, localfile, cb)) +let open_domain ?fail name cb = send_to_slave ?fail (Open_domain (name, cb)) +let open_images ?fail images cb = send_to_slave ?fail (Open_images (images, cb)) +let read_directory ?fail src path cb = + send_to_slave ?fail (Read_directory (src, path, cb)) (*----- Slave thread starts here -----*) (* Set this to true to exit the thread. *) let quit = ref false +(* Handles. These are not protected by locks because only the slave + * thread has access to them. + *) +let conn = ref None +let g = ref None + +(* Run the callback unless someone set the q_discard flag while + * we were running the command. + *) +let callback_if_not_discarded (cb : 'a callback) (arg : 'a) = + let discard = with_lock q_lock (fun () -> !q_discard) in + if not discard then + GtkThread.async cb arg + +(* Call 'f ()' with source mounted read-only. Ensure that everything + * is unmounted even if an exception is thrown. + *) +let with_mount_ro g src (f : unit -> 'a) : 'a = + Std.finally (fun () -> g#umount_all ()) ( + fun () -> + (* Do the mount - could be OS or single volume. *) + (match src with + | Volume dev -> g#mount_ro dev "/"; + | OS { insp_mountpoints = mps } -> + (* Sort the mountpoint keys by length, shortest first. *) + let cmp (a,_) (b,_) = compare (String.length a) (String.length b) in + let mps = List.sort ~cmp mps in + (* Mount the filesystems. *) + List.iter ( + fun (mp, dev) -> g#mount_ro dev mp + ) mps + ); + f () + ) () + +(* Update the status bar. *) +let status fs = + let f str = GtkThread.async !status_hook str in + ksprintf f fs + let rec loop () = + debug "top of slave loop"; + (* Get the next command. *) - let cmd = + let fail, cmd = with_lock q_lock ( fun () -> - while Q.is_empty q do - Cond.wait q_cond q_lock - done; + while Q.is_empty q do Cond.wait q_cond q_lock done; + q_discard := false; Q.pop q ) in - debug "slave thread processing command %s ..." (string_of_command cmd); + debug "slave processing command %s ..." (string_of_command cmd); (try - call_callback !busy_hook (); - execute_command cmd; - call_callback !idle_hook (); + GtkThread.async !busy_hook (); + execute_command cmd with exn -> - (* If a command fails, clear the command queue and run the - * failure hook in the main thread. + (* If the user provided an override ?fail parameter to the + * original call, call that, else call the global hook. *) - call_callback !idle_hook (); - discard_command_queue (); - call_callback !failure_hook exn + match fail with + | Some cb -> GtkThread.async cb exn + | None -> GtkThread.async !failure_hook exn ); + (* If there are no more commands in the queue, run the idle hook. *) + let empty = with_lock q_lock (fun () -> Q.is_empty q) in + if empty then GtkThread.async !idle_hook (); + if !quit then Thread.exit (); loop () and execute_command = function | Exit_thread -> quit := true; - disconnect_all () + close_all () + + | Connect (name, cb) -> + let printable_name = + match name with None -> "default hypervisor" | Some uri -> uri in + status "Connecting to %s ..." printable_name; - | Connect (uri, cb) -> - disconnect_all (); - conn := Some (C.connect_readonly ?uri ()); - call_callback cb () + close_all (); + conn := Some (C.connect_readonly ?name ()); - | Get_domains cb -> let conn = get_conn () in let doms = D.get_domains conn [D.ListAll] in let doms = List.map ( fun d -> - D.get_id d, D.get_name d, (D.get_info d).D.state + { dom_id = D.get_id d; + dom_name = D.get_name d; + dom_state = (D.get_info d).D.state } ) doms in - call_callback cb doms + let cmp { dom_name = n1 } { dom_name = n2 } = UTF8.compare n1 n2 in + let doms = List.sort ~cmp doms in + + status "Connected to %s" printable_name; + callback_if_not_discarded cb doms + + | Disk_usage (src, remotedir, cb) -> + status "Calculating disk usage of %s ..." remotedir; + + let g = get_g () in + let r = + with_mount_ro g src ( + fun () -> + g#du remotedir + ) in + + status "Finished calculating disk usage of %s" remotedir; + callback_if_not_discarded cb r + + | Download_dir_find0 (src, remotedir, localfile, cb) -> + status "Downloading %s filenames to %s ..." remotedir localfile; + + let g = get_g () in + with_mount_ro g src ( + fun () -> + g#find0 remotedir localfile + ); + + status "Finished downloading %s" localfile; + callback_if_not_discarded cb () + + | Download_dir_tarball (src, remotedir, format, localfile, cb) -> + status "Downloading %s to %s ..." remotedir localfile; + + let g = get_g () in + let f = match format with + | Tar -> g#tar_out + | TGZ -> g#tgz_out + | TXZ -> g#txz_out + in + with_mount_ro g src ( + fun () -> + f remotedir localfile + ); + + status "Finished downloading %s" localfile; + callback_if_not_discarded cb () + + | Download_file (src, remotefile, localfile, cb) -> + status "Downloading %s to %s ..." remotefile localfile; + + let g = get_g () in + with_mount_ro g src ( + fun () -> + g#download remotefile localfile + ); + + status "Finished downloading %s" localfile; + callback_if_not_discarded cb () + + | Open_domain (name, cb) -> + status "Opening %s ..." name; + + let conn = get_conn () in + let dom = D.lookup_by_name conn name in + let xml = D.get_xml_desc dom in + let images = get_disk_images_from_xml xml in + open_disk_images images cb -(* Call a callback function or hook in the main thread. *) -and call_callback cb arg = - GtkThread.async cb arg + | Open_images (images, cb) -> + status "Opening disk images ..."; + + open_disk_images images cb + + | Read_directory (src, dir, cb) -> + status "Reading directory %s ..." dir; + + let g = get_g () in + let names, stats, links = + with_mount_ro g src ( + fun () -> + let names = g#ls dir in (* sorted and without . and .. *) + let names = Array.to_list names in + let stats = lstatlist_wrapper g dir names in + let links = readlink_wrapper g dir names stats in + names, stats, links + ) in + assert ( + let n = List.length names in + n = List.length stats && n = List.length links + ); + let entries = List.combine (List.combine names stats) links in + let entries = List.map ( + fun ((name, stat), link) -> + { dent_name = name; dent_stat = stat; dent_link = link } + ) entries in + + status "Finished reading directory %s" dir; + callback_if_not_discarded cb entries (* Expect to be connected, and return the current libvirt connection. *) -let get_conn () = +and get_conn () = match !conn with | Some conn -> conn | None -> failwith "not connected to libvirt" +and get_g () = + match !g with + | Some g -> g + | None -> failwith "no domain or disk image is open" + (* Close all libvirt and libguestfs handles. *) -and disconnect_all () = +and close_all () = (match !conn with Some conn -> C.close conn | None -> ()); - conn := None + conn := None; + close_g () + +and close_g () = + (match !g with Some g -> g#close () | None -> ()); + g := None + +and get_disk_images_from_xml xml = + let xml = Xml.parse_string xml in + + (* Return the device nodes. *) + let devices = + match xml with + | Xml.Element ("domain", _, children) -> + let devices = + List.filter_map ( + function + | Xml.Element ("devices", _, devices) -> Some devices + | _ -> None + ) children in + List.concat devices + | _ -> + failwith "get_xml_desc didn't return " in + + (* Look for and return attr_val. *) + let rec source_of attr_name = function + | [] -> None + | Xml.Element ("source", attrs, _) :: rest -> + (try Some (List.assoc attr_name attrs) + with Not_found -> source_of attr_name rest) + | _ :: rest -> source_of attr_name rest + in + + (* Look for and return attr_val. *) + let rec format_of = function + | [] -> None + | Xml.Element ("driver", attrs, _) :: rest -> + (try Some (List.assoc "type" attrs) + with Not_found -> format_of rest) + | _ :: rest -> format_of rest + in + + (* Look for nodes and return the sources (block devices) of those. *) + let blkdevs = + List.filter_map ( + function + | Xml.Element ("disk", attrs, disks) -> + let filename = + try + let typ = List.assoc "type" attrs in + if typ = "file" then source_of "file" disks + else if typ = "block" then source_of "dev" disks + else None + with + Not_found -> None in + (match filename with + | None -> None + | Some filename -> + let format = format_of disks in + Some (filename, format) + ); + | _ -> None + ) devices in + blkdevs + +(* The common code for Open_domain and Open_images which opens the + * libguestfs handle, adds the disks, and launches the appliance. + *) +and open_disk_images images cb = + debug "opening disk image %s" (string_of_images images); + + close_g (); + let g' = new G.guestfs () in + g := Some g'; + let g = g' in + + g#set_trace (trace ()); + + (* Uncomment the next line to pass the verbose flag from the command + * line through to libguestfs. This is not generally necessary since + * we are not so interested in debugging libguestfs problems at this + * level, and the user can always set LIBGUESTFS_DEBUG=1 if they need + * to. + *) + (* g#set_verbose (verbose ());*) + + (* Attach progress bar callback. *) + g#set_progress_callback ( + fun proc_nr serial position total -> + debug "progress callback proc_nr=%d serial=%d posn=%Ld total=%Ld" + proc_nr serial position total; + GtkThread.async !progress_hook (position, total) + ); + + List.iter ( + function + | filename, None -> + g#add_drive_opts ~readonly:true filename + | filename, Some format -> + g#add_drive_opts ~readonly:true ~format filename + ) images; + + g#launch (); + + status "Listing filesystems ..."; + + (* Get list of filesystems. *) + let fses = g#list_filesystems () in + + status "Looking for operating systems ..."; + + (* Perform inspection. This can fail, ignore errors. *) + let roots = + try Array.to_list (g#inspect_os ()) + with + G.Error msg -> + debug "inspection failed (error ignored): %s" msg; + [] in + + let oses = List.map ( + fun root -> + let typ = g#inspect_get_type root in + let windows_systemroot = + if typ <> "windows" then None + else ( + try Some (g#inspect_get_windows_systemroot root) + with G.Error _ -> None + ) in + + (* Create most of the OS object that we're going to return. We + * have to pass this to with_mount_ro below which is why we need + * to partially create it here. + *) + let os = { + insp_root = root; + insp_arch = g#inspect_get_arch root; + insp_distro = g#inspect_get_distro root; + insp_filesystems = g#inspect_get_filesystems root; + insp_hostname = g#inspect_get_hostname root; + insp_major_version = g#inspect_get_major_version root; + insp_minor_version = g#inspect_get_minor_version root; + insp_mountpoints = g#inspect_get_mountpoints root; + insp_package_format = g#inspect_get_package_format root; + insp_package_management = g#inspect_get_package_management root; + insp_product_name = g#inspect_get_product_name root; + insp_type = typ; + insp_windows_systemroot = windows_systemroot; + insp_winreg_DEFAULT = None; (* incomplete, see below *) + insp_winreg_SAM = None; + insp_winreg_SECURITY = None; + insp_winreg_SOFTWARE = None; + insp_winreg_SYSTEM = None; + } in + + (* We need to mount the root in order to look for Registry hives. *) + let winreg_DEFAULT, winreg_SAM, winreg_SECURITY, winreg_SOFTWARE, + winreg_SYSTEM = + match windows_systemroot with + | None -> None, None, None, None, None + | Some sysroot -> + with_mount_ro g (OS os) ( + fun () -> + let check_for_hive filename = + let path = + sprintf "%s/system32/config/%s" sysroot filename in + try Some (g#case_sensitive_path path) + with G.Error _ -> None + in + check_for_hive "default", + check_for_hive "sam", + check_for_hive "security", + check_for_hive "software", + check_for_hive "system" + ) in + + (* Fill in the remaining struct fields. *) + let os = { os with + insp_winreg_DEFAULT = winreg_DEFAULT; + insp_winreg_SAM = winreg_SAM; + insp_winreg_SECURITY = winreg_SECURITY; + insp_winreg_SOFTWARE = winreg_SOFTWARE; + insp_winreg_SYSTEM = winreg_SYSTEM + } in + os + ) roots in + + let data = { + insp_all_filesystems = fses; + insp_oses = oses; + } in + + status "Finished opening disk"; + callback_if_not_discarded cb data + +(* guestfs_lstatlist has a "hidden" limit of the protocol message size. + * Call this function, but split the list of names into chunks. + *) +and lstatlist_wrapper g dir = function + | [] -> [] + | names -> + let names', names = List.take 1000 names, List.drop 1000 names in + let xs = g#lstatlist dir (Array.of_list names') in + let xs = Array.to_list xs in + xs @ lstatlist_wrapper g dir names + +(* For each entry which is a symlink, read the destination of the + * symlink. This is non-trivial because on Windows we cannot use + * readlink but need to instead parse the reparse data from NTFS. + *) +and readlink_wrapper g dir names stats = + (* Is the directory on an NTFS filesystem? *) + let dev = get_mounted_device g dir in + if g#vfs_type dev <> "ntfs" then ( + (* Not NTFS, use the fast g#readlinklist method. *) + let rec readlinklist_wrapper g dir = function + | [] -> [] + | names -> + let names', names = List.take 1000 names, List.drop 1000 names in + let xs = g#readlinklist dir (Array.of_list names') in + let xs = Array.to_list xs in + xs @ readlinklist_wrapper g dir names + in + readlinklist_wrapper g dir names + ) + else ( + (* NTFS: look up each symlink individually. *) + List.map ( + fun (name, stat) -> + if not (is_symlink stat.G.mode) then "" + else + let path = if dir = "/" then dir ^ name else dir ^ "/" ^ name in + try + let _, display = get_ntfs_reparse_data g path in + display + with exn -> + debug "get_ntfs_reparse_data: %s: failed: %s" + path (Printexc.to_string exn); + "?" + ) (List.combine names stats) + ) + +(* See: + * https://bugzilla.redhat.com/show_bug.cgi?id=663407 + * http://git.annexia.org/?p=libguestfs.git;a=commit;h=3a3836b933b80c4f9f2c767fda4f8b459f998db2 + * http://www.tuxera.com/community/ntfs-3g-advanced/junction-points-and-symbolic-links/ + * http://www.tuxera.com/community/ntfs-3g-advanced/extended-attributes/ + * http://www.codeproject.com/KB/winsdk/junctionpoints.aspx + *) +and get_ntfs_reparse_data g path = + let data = g#lgetxattr path "system.ntfs_reparse_data" in + let link, display = + bitmatch Bitstring.bitstring_of_string data with + (* IO_REPARSE_TAG_MOUNT_POINT *) + | { 0xa0000003_l : 32 : littleendian; + _ : 16 : littleendian; (* data length - ignore it *) + _ : 16 : littleendian; (* reserved *) + link_offset : 16 : littleendian; + link_len : 16 : littleendian; + display_offset : 16 : littleendian; + display_len : 16 : littleendian; + link : link_len * 8 : + string, offset (8 * (link_offset + 0x10)); + display : display_len * 8 : + string, offset (8 * (display_offset + 0x10)) } -> + (* These strings should always be valid UTF16LE, but the caller + * is prepared to catch any exception if this fails. + *) + let link = windows_string_to_utf8 link in + let display = windows_string_to_utf8 display in + link, display + | { 0xa0000003_l : 32 : littleendian } -> + invalid_arg ( + sprintf "%s: could not parse IO_REPARSE_TAG_MOUNT_POINT data" path + ) + + (* IO_REPARSE_TAG_SYMLINK *) + | { 0xa000000c_l : 32 : littleendian; + _ : 16 : littleendian; (* data length - ignore it *) + _ : 16 : littleendian; (* reserved *) + link_offset : 16 : littleendian; + link_len : 16 : littleendian; + display_offset : 16 : littleendian; + display_len : 16 : littleendian; + link : link_len * 8 : + string, offset (8 * (link_offset + 0x14)); + display : display_len * 8 : + string, offset (8 * (display_offset + 0x14)) } -> + let link = windows_string_to_utf8 link in + let display = windows_string_to_utf8 display in + link, display + | { 0xa000000c_l : 32 : littleendian } -> + invalid_arg ( + sprintf "%s: could not parse IO_REPARSE_TAG_SYMLINK data" path + ) + + | { i : 32 : littleendian } -> + invalid_arg ( + sprintf "%s: reparse data of type 0x%lx is not supported" path i + ) + | { _ } -> + invalid_arg (sprintf "%s: reparse data is too short" path) in + + link, display + +(* Given a path which is located somewhere on a mountpoint, return the + * device name. This works by using g#mountpoints and then looking for + * the mount path with the longest match. + *) +and get_mounted_device g path = + let mps = g#mountpoints () in + let mps = List.map ( + fun (dev, mp) -> + if String.starts_with path mp then dev, String.length mp else dev, 0 + ) mps in + let cmp (_,n1) (_,n2) = compare n2 n1 in + let mps = List.sort ~cmp mps in + match mps with + | [] -> + invalid_arg (sprintf "%s: not mounted" path) + | (_,0) :: _ -> + invalid_arg (sprintf "%s: not found on any filesystem" path) + | (dev,_) :: _ -> dev (* Start up one slave thread. *) let slave_thread = Thread.create loop () @@ -154,5 +735,5 @@ let slave_thread = Thread.create loop () (* Note the following function is called from the main thread. *) let exit_thread () = discard_command_queue (); - send_to_slave Exit_thread; + ignore (send_to_slave Exit_thread); Thread.join slave_thread