CA-383867: Add startup procedure to xapi-guard

Because timestamps depend on a monotonic timestamp that depends on boot, files
need to be renamed to ensure future writes have higher timestamps to be
considered newer and be uploaded to xapi.

On top of this, allows to report about remnant temporary files, delete invalid
files and remove empty directories.

Signed-off-by: Pau Ruiz Safont <[email protected]>

Author: psafont

doc/content/xapi-guard/_index.md

@@ -35,6 +35,9 @@ This situation usually happens when xapi is being restarted as part of an update
 SWTPM, the vTPM daemon, reads the contents of the TPM from xapi-guard on startup, suspend, and resume.
 During normal operation SWTPM does not send read requests from xapi-guard.
 
+Structure
+---------
+
 The cache module consists of two Lwt threads, one that writes to disk, and another one that reads from disk.
 The writer is triggered when a VM writes to the vTPM.
 It never blocks if xapi is unreachable, but responds as soon as the data has been stored either by xapi or on the local disk, such that the VM receives a timely response to the write request.
@@ -82,3 +85,14 @@ stateDiagram-v2
     Engaged --> Engaged : Writer receives TPM, queue is not full
     Engaged --> Disengaged : Writer receives TPM, queue is full
 ```
+
+Startup
+------
+
+At startup, there's a dedicated routine to transform the existing contents of the cache.
+This is currently done because the timestamp reference change on each boot.
+This means that the existing contents might have timestamps considered more recent than timestamps of writes coming from running events, leading to missing content updates.
+This must be avoided and instead the updates with offending timestamps are renamed to a timestamp taken from the current timestamp, ensuring a consistent
+ordering.
+The routine is also used to keep a minimal file tree: unrecognised files are deleted, temporary files created to ensure atomic writes are left untouched, and empty directories are deleted.
+This mechanism can be changed in the future to migrate to other formats.

ocaml/xapi-guard/lib/disk_cache.ml

@@ -179,7 +179,6 @@ type channel = {
    IDEA: carryover: read contents of cache and "convert it" to the current run
 
    TODO:
-     - Add startup step to convert existing content to new time
      - Exponential backoff on xapi push error
      - Limit error logging on xapi push error: once per downtime is enough
  *)
@@ -449,8 +448,89 @@ end = struct
     loop
 end
 
+(** Module use to change the cache contents before the reader and writer start
+    running *)
+module Setup : sig
+  val retime_cache_contents : Types.Service.t -> unit Lwt.t
+end = struct
+  type file_action =
+    | Keep of file
+    | Delete of string
+    | Move of {from: string; into: string}
+
+  let get_fs_action root now = function
+    | Latest ((uuid, timestamp, key), from) as latest ->
+        if Mtime.is_later ~than:now timestamp then
+          let timestamp = now in
+          let into = path_of_key root (uuid, timestamp, key) in
+          Move {from; into}
+        else
+          Keep latest
+    | Temporary _ as temp ->
+        Keep temp
+    | Invalid p | Outdated (_, p) ->
+        Delete p
+
+  let commit __FUN = function
+    | Keep (Temporary p) ->
+        Logs_lwt.warn (fun m ->
+            m "%s: Found temporary file, ignoring '%s'" __FUN p
+        )
+    | Keep _ ->
+        Lwt.return_unit
+    | Delete p ->
+        let* () = Logs_lwt.info (fun m -> m "%s: Deleting '%s'" __FUN p) in
+        Lwt_unix.unlink p
+    | Move {from; into} ->
+        let* () =
+          Logs_lwt.info (fun m -> m "%s: Moving '%s' to '%s'" __FUN from into)
+        in
+        Lwt_unix.rename from into
+
+  let rec delete_empty_dirs ~delete_root root =
+    (* Delete subdirectories, then *)
+    let* files = files_in root ~otherwise:(fun _ -> Lwt.return []) in
+    let* () =
+      Lwt_list.iter_p
+        (fun path ->
+          let* {st_kind; _} = Lwt_unix.stat path in
+          match st_kind with
+          | S_DIR ->
+              delete_empty_dirs ~delete_root:true path
+          | _ ->
+              Lwt.return_unit
+        )
+        files
+    in
+    if not delete_root then
+      Lwt.return_unit
+    else
+      let* files = files_in root ~otherwise:(fun _ -> Lwt.return []) in
+      Lwt.catch
+        (fun () ->
+          if files = [] then
+            Lwt_unix.rmdir root
+          else
+            Lwt.return_unit
+        )
+        (fun _ -> Lwt.return_unit)
+
+  (* The code assumes it's the only with access to the disk cache while running *)
+  let retime_cache_contents typ =
+    let now = Mtime_clock.now () in
+    let root = cache_of typ in
+    let* contents = get_all_contents root in
+    let* () =
+      contents
+      |> List.map (get_fs_action root now)
+      |> Lwt_list.iter_p (commit __FUNCTION__)
+    in
+    delete_empty_dirs ~delete_root:false root
+end
+
 let setup typ direct =
+  let* () = Setup.retime_cache_contents typ in
   let queue, push = Lwt_bounded_stream.create 4098 in
   let lock = Lwt_mutex.create () in
   let q = {queue; push; lock; state= Disengaged} in
-  (Writer.with_cache ~direct typ q, Watcher.watch ~direct typ q)
+  Lwt.return (Writer.with_cache ~direct typ q, Watcher.watch ~direct typ q)

ocaml/xapi-guard/lib/disk_cache.mli

@@ -18,8 +18,10 @@ type t = Uuidm.t * Mtime.t * Types.Tpm.key
 val setup :
      Types.Service.t
   -> (t -> string -> (unit, exn) Lwt_result.t)
-  -> (((t -> string -> unit Lwt.t) -> 'a Lwt.t) -> 'a Lwt.t)
+  -> ( (((t -> string -> unit Lwt.t) -> 'a Lwt.t) -> 'a Lwt.t)
      * (unit -> unit Lwt.t)
+     )
+     Lwt.t
 (** [setup service push_callback] Returns a local disk buffer for [service]
     which will use [push_callback] to push the elements to their final
     destination *)

ocaml/xapi-guard/src/main.ml

@@ -269,7 +269,7 @@ let retry_forever fname f =
 let cache_reader with_watcher = retry_forever "cache watcher" with_watcher
 
 let make_message_switch_server () =
-  let with_swtpm_push, with_watch =
+  let* with_swtpm_push, with_watch =
     Xapi_guard.Disk_cache.(setup Swtpm (Server_interface.push_vtpm ~cache))
   in
   let open Message_switch_lwt.Protocol_lwt in

ocaml/xapi-guard/test/cache_test.ml

@@ -128,7 +128,7 @@ let to_cache with_writer =
 let from_cache with_watcher = retry_forever "watcher" with_watcher
 
 let main () =
-  let with_writer, with_watcher = Xapi_guard.Disk_cache.(setup Swtpm log) in
+  let* with_writer, with_watcher = Xapi_guard.Disk_cache.(setup Swtpm log) in
   let reader = from_cache with_watcher in
   let writers = to_cache with_writer in
   let* _ = Lwt.all (reader :: writers) in