Merge pull request #1413 from samoht/tests

Add tests for the MirageSDK and start the new control plane protocol

projects/miragesdk/src/.gitignore

@@ -8,11 +8,12 @@ hash
 dhcp-client/bpf/.merlin
 dhcp-client/.merlin
 sdk/.merlin
+test/.merlin
 
 # Generated by `make dev`
 _build/
 main.native
-calf/dhcp_client
+dhcp-client/calf/dhcp_client
 src/bpf/.merlin
 
 # Generated by the mirage tool

projects/miragesdk/src/Makefile

@@ -65,11 +65,18 @@ clean::
 	(docker rmi -f $(IMAGE):pkg || echo ok)
 	(docker rmi -f $(IMAGE):dev || echo ok)
 
+#### DEV
+
+.PHONY: test
+
+test:
+	jbuilder runtest --dev
+
+dev-clean:
+	rm -rf _build dhcp-client/calf/_build
+
 dev:
-	cd calf && mirage configure && make
-	jbuilder build src/main.exe
-#	_build/default/src/main.exe -vv \
-#	  --cmd 'calf/_build/main.native -l debug --store 10 --net 12' \
-#	  --ethif eno1
+	cd dhcp-client/calf && mirage configure && make
+	jbuilder build dhcp-client/main.exe --dev
 
 .DELETE_ON_ERROR:

projects/miragesdk/src/sdk/IO.ml

@@ -0,0 +1,46 @@
+open Lwt.Infix
+
+let src = Logs.Src.create "IO" ~doc:"IO helpers"
+module Log = (val Logs.src_log src : Logs.LOG)
+
+let rec really_write fd buf off len =
+  Log.debug (fun l -> l "really_write");
+  match len with
+  | 0   -> Lwt.return_unit
+  | len ->
+    Lwt_unix.write fd buf off len >>= fun n ->
+    really_write fd buf (off+n) (len-n)
+
+let rec really_read fd buf off len =
+  Log.debug (fun l -> l "really_read");
+  match len with
+  | 0   -> Lwt.return_unit
+  | len ->
+    Lwt_unix.read fd buf off len >>= fun n ->
+    really_read fd buf (off+n) (len-n)
+
+let read_all fd =
+  Log.debug (fun l -> l "read_all");
+  let len = 16 * 1024 in
+  let buf = Bytes.create len in
+  let rec loop acc =
+    Lwt_unix.read fd buf 0 len >>= fun n ->
+    let acc = String.sub buf 0 n :: acc in
+    if n <= len then Lwt.return (List.rev acc)
+    else loop acc
+  in
+  loop [] >|= fun bufs ->
+  String.concat "" bufs
+
+let read_n fd len =
+  Log.debug (fun l -> l "read_n");
+  let buf = Bytes.create len in
+  let rec loop acc len =
+    Lwt_unix.read fd buf 0 len >>= fun n ->
+    let acc = String.sub buf 0 n :: acc in
+    match len - n with
+    | 0 -> Lwt.return (List.rev acc)
+    | r -> loop acc r
+  in
+  loop [] len >|= fun bufs ->
+  String.concat "" bufs

projects/miragesdk/src/sdk/IO.mli

@@ -0,0 +1,13 @@
+(** IO helpers *)
+
+val really_write: Lwt_unix.file_descr -> string -> int -> int -> unit Lwt.t
+(** [really_write fd buf off len] writes exactly [len] bytes to [fd]. *)
+
+val really_read: Lwt_unix.file_descr -> string -> int -> int -> unit Lwt.t
+(** [really_read fd buf off len] reads exactly [len] bytes from [fd]. *)
+
+val read_all: Lwt_unix.file_descr -> string Lwt.t
+(** [read_all fd] reads as much data as it is available in [fd]. *)
+
+val read_n: Lwt_unix.file_descr -> int -> string Lwt.t
+(** [read_n fd n] reads exactly [n] bytes from [fd]. *)

projects/miragesdk/src/sdk/ctl.ml

@@ -4,7 +4,7 @@ let src = Logs.Src.create "init" ~doc:"Init steps"
 module Log = (val Logs.src_log src : Logs.LOG)
 
 (* FIXME: to avoid linking with gmp *)
-module IO = struct
+module No_IO = struct
   type ic = unit
   type oc = unit
   type ctx = unit
@@ -17,7 +17,7 @@ module IO = struct
 end
 
 (* FIXME: we don't use Irmin_unix.Git.FS.KV to avoid linking with gmp *)
-module Store = Irmin_git.FS.KV(IO)(Inflator)(Io_fs)
+module Store = Irmin_git.FS.KV(No_IO)(Inflator)(Io_fs)
 module KV = Store(Irmin.Contents.String)
 
 let v path =
@@ -28,17 +28,84 @@ let v path =
 let () =
   Irmin.Private.Watch.set_listen_dir_hook Irmin_watcher.hook
 
+module Message = struct
+
+  [%%cenum
+    type operation =
+      | Write
+      | Read
+      | Delete
+    [@@uint8_t]
+  ]
+
+  type t = {
+    operation: operation;
+    path     : string;
+    payload  : string option;
+  }
+
+  [%%cstruct type message = {
+      operation : uint8_t; (* = type operation *)
+      path      : uint16_t;
+      payload   : uint16_t;
+    } [@@little_endian]
+  ]
+
+  (* to avoid warning 32 *)
+  let _ = hexdump_message
+  let _ = operation_to_string
+  let _ = string_to_operation
+
+  let read_message fd =
+    IO.read_n fd 4 >>= fun buf ->
+    let len =
+      Cstruct.LE.get_uint32 (Cstruct.of_string buf) 0
+      |> Int32.to_int
+    in
+    IO.read_n fd len >>= fun buf ->
+    let buf = Cstruct.of_string buf in
+    let operation = match int_to_operation (get_message_operation buf) with
+      | None   -> failwith "invalid operation"
+      | Some o -> o
+    in
+    let path_len = get_message_path buf in
+    let payload_len = get_message_payload buf in
+    IO.read_n fd path_len >>= fun path ->
+    (match payload_len with
+     | 0 -> Lwt.return None
+     | n -> IO.read_n fd n >|= fun x -> Some x)
+    >|= fun payload ->
+    { operation; path; payload }
+
+  let write_message fd msg =
+    let operation = operation_to_int msg.operation in
+    let path = String.length msg.path in
+    let payload = match msg.payload with
+      | None   -> 0
+      | Some x -> String.length x
+    in
+    let len = sizeof_message + path + payload in
+    let buf = Cstruct.create len in
+    set_message_operation buf operation;
+    set_message_path buf path;
+    set_message_payload buf path;
+    Cstruct.blit_from_bytes msg.path 0 buf sizeof_message path;
+    let () = match msg.payload with
+      | None   -> ()
+      | Some x -> Cstruct.blit_from_bytes x 0 buf (sizeof_message+path) payload
+    in
+    IO.really_write fd (Cstruct.to_string buf) 0 len
+
+end
+
 module Dispatch = struct
 
-  module Wm = struct
-    module Rd = Webmachine.Rd
-    include Webmachine.Make(Cohttp_lwt_unix.Server.IO)
-  end
+  open Message
 
-  let with_key rd f =
-    match KV.Key.of_string rd.Wm.Rd.dispatch_path with
-    | Ok x    -> f x
-    | Error _ -> Wm.respond 404 rd
+  let with_key msg f =
+    match KV.Key.of_string msg.path with
+    | Ok x           -> f x
+    | Error (`Msg e) -> Fmt.kstrf Lwt.fail_with "invalid key: %s" e
 
   let infof fmt =
     Fmt.kstrf (fun msg () ->
@@ -46,91 +113,50 @@ module Dispatch = struct
         Irmin.Info.v ~date ~author:"calf" msg
       ) fmt
 
-  let ok = "{\"status\": \"ok\"}"
-
-  class item db = object(self)
-
-    inherit [Cohttp_lwt_body.t] Wm.resource
-
-    method private of_string rd =
-      Cohttp_lwt_body.to_string rd.Wm.Rd.req_body >>= fun value ->
-      with_key rd (fun key ->
+  let dispatch db msg =
+    with_key msg (fun key ->
+        match msg.operation with
+        | Write ->
           let info = infof "Updating %a" KV.Key.pp key in
-          KV.set db ~info key value >>= fun () ->
-          let resp_body = `String ok in
-          let rd = { rd with Wm.Rd.resp_body } in
-          Wm.continue true rd
-        )
+          (match msg.payload with
+           | None   -> Fmt.kstrf Lwt.fail_with "dispatch: missing payload"
+           | Some v -> KV.set db ~info key v)
+        | _ -> failwith "TODO"
+      )
 
-    method private to_string rd =
-      with_key rd (fun key ->
-          KV.find db key >>= function
-          | Some value -> Wm.continue (`String value) rd
-          | None       -> assert false
-        )
-
-    method resource_exists rd =
-      with_key rd (fun key ->
-          KV.mem db key >>= fun mem ->
-          Wm.continue mem rd
-        )
-
-    method allowed_methods rd =
-      Wm.continue [`GET; `HEAD; `PUT; `DELETE] rd
-
-    method content_types_provided rd =
-      Wm.continue [
-        "plain", self#to_string
-      ] rd
-
-    method content_types_accepted rd =
-      Wm.continue [
-        "plain", self#of_string
-      ] rd
-
-    method delete_resource rd =
-      with_key rd (fun key ->
-          let info = infof "Deleting %a" KV.Key.pp key in
-          KV.remove db ~info key >>= fun () ->
-          let resp_body = `String ok in
-          Wm.continue true { rd with Wm.Rd.resp_body }
-        )
-  end
-
-  let v db routes =
-    let routes = List.map (fun r -> r, fun () -> new item db) routes in
-    let callback (_ch, _conn) request body =
-      let open Cohttp in
-      (Wm.dispatch' routes ~body ~request >|= function
-        | None        -> (`Not_found, Header.init (), `String "Not found", [])
-        | Some result -> result)
-      >>= fun (status, headers, body, path) ->
-      Log.info (fun l ->
-          l "%d - %s %s"
-            (Code.code_of_status status)
-            (Code.string_of_method (Request.meth request))
-            (Uri.path (Request.uri request)));
-      Log.debug (fun l -> l "path=%a" Fmt.(Dump.list string) path);
-      (* Finally, send the response to the client *)
-      Cohttp_lwt_unix.Server.respond ~headers ~body ~status ()
+  let serve fd db ~routes =
+    let msgs = Queue.create () in
+    let cond = Lwt_condition.create () in
+    let rec listen () =
+      read_message fd >>= fun msg ->
+      Queue.add msg msgs;
+      Lwt_condition.signal cond ();
+      listen ()
     in
-    (* create the server and handle requests with the function defined above *)
-    let conn_closed (_, conn) =
-      Log.info (fun l ->
-          l "connection %s closed\n%!" (Cohttp.Connection.to_string conn))
+    let rec process () =
+      Lwt_condition.wait cond >>= fun () ->
+      let msg = Queue.pop msgs in
+      (if List.mem msg.path routes then dispatch db msg
+       else (
+         Log.err (fun l -> l "%s is not an allowed path" msg.path);
+         Lwt.return_unit;
+       )) >>= fun () ->
+      process ()
     in
-    Cohttp_lwt_unix.Server.make ~callback ~conn_closed ()
+    Lwt.pick [
+      listen ();
+      process ();
+    ]
+
 end
 
 let int_of_fd (t:Lwt_unix.file_descr) =
   (Obj.magic (Lwt_unix.unix_file_descr t): int)
 
 let serve ~routes db fd =
-  let http = Dispatch.v db routes in
-  let on_exn e = Log.err (fun l -> l "ERROR: %a" Fmt.exn e) in
   Lwt_unix.blocking fd >>= fun blocking ->
   Log.debug (fun l ->
       l "Serving the control state over fd:%d (blocking=%b)"
         (int_of_fd fd) blocking
     );
-  Cohttp_lwt_unix.Server.create ~on_exn ~mode:(`Fd fd) http
+  Dispatch.serve fd db ~routes

projects/miragesdk/src/sdk/ctl.mli

@@ -3,6 +3,29 @@
 
 module KV: Irmin.KV with type contents = string
 
+module Message: sig
+
+  (** The type for operations. *)
+  type operation =
+    | Write
+    | Read
+    | Delete
+
+  (** The type for control messages. *)
+  type t = {
+    operation: operation;
+    path     : string;
+    payload  : string option;
+  }
+
+  val write_message: Lwt_unix.file_descr -> t -> unit Lwt.t
+  (** [write_message fd t] writes a control message. *)
+
+  val read_message: Lwt_unix.file_descr -> t Lwt.t
+  (** [read_message fd] reads a control message. *)
+
+end
+
 val v: string -> KV.t Lwt.t
 (** [v p] is the KV store storing the control state, located at path
     [p] in the filesystem of the privileged container. *)

projects/miragesdk/src/sdk/init.ml

@@ -72,13 +72,6 @@ module Fd = struct
       listen_socket ();
     ]
 
-  let rec really_write dst buf off len =
-    match len with
-    | 0   -> Lwt.return_unit
-    | len ->
-      Lwt_unix.write dst.fd buf off len >>= fun n ->
-      really_write dst buf (off+n) (len-n)
-
   let forward ~src ~dst =
     Log.debug (fun l -> l "forward %a => %a" pp src pp dst);
     let len = 16 * 1024 in
@@ -92,18 +85,12 @@ module Fd = struct
         Log.debug (fun l ->
             l "FORWARD[%a => %a]: %S (%d)"
               pp src pp dst (Bytes.sub buf 0 len) len);
-        really_write dst buf 0 len >>= fun () ->
+        IO.really_write dst.fd buf 0 len >>= fun () ->
         loop ()
       )
     in
     loop ()
 
-  let proxy x y =
-    Lwt.pick [
-      forward ~src:x ~dst:y;
-      forward ~src:y ~dst:x;
-    ]
-
 end
 
 module Pipe = struct
@@ -207,7 +194,7 @@ let exec_priv ~pid ~cmd ~net ~ctl ~handlers =
       Fd.forward ~src:Pipe.(priv stderr)  ~dst:Fd.stderr;
       (* TODO: Init.Fd.forward ~src:Init.Pipe.(priv metrics) ~dst:Init.Fd.metric; *)
       ctl ();
-(*      handlers (); *)
+      handlers ();
     ])
 
 let run ~net ~ctl ~handlers cmd =

projects/miragesdk/src/sdk/init.mli

@@ -14,7 +14,6 @@
        data, e.g. the IP address once a DHCP lease is obtained.}
     }*)
 
-
 module Fd: sig
 
   type t

projects/miragesdk/src/sdk/jbuild

@@ -2,8 +2,8 @@
 
 (library
  ((name sdk)
-  (libraries (logs-syslog.lwt threads cohttp.lwt cstruct.lwt
-              cmdliner fmt.cli logs.fmt logs.cli fmt.tty decompress
-              irmin irmin-git irmin-http lwt.unix rawlink tuntap
+  (libraries (threads cstruct.lwt cmdliner fmt.cli logs.fmt logs.cli fmt.tty
+              decompress irmin irmin-git lwt.unix rawlink tuntap dispatch
               irmin-watcher inotify))
+  (preprocess (per_file ((pps (cstruct.ppx)) (ctl))))
  ))

projects/miragesdk/src/test/jbuild

@@ -0,0 +1,10 @@
+(jbuild_version 1)
+
+(executables
+  ((names (test))
+   (libraries (sdk alcotest astring mtime.os))))
+
+(alias
+ ((name    runtest)
+  (deps    (test.exe))
+  (action  (run ${<}))))

projects/miragesdk/src/test/test.ml

@@ -0,0 +1,74 @@
+open Astring
+open Lwt.Infix
+open Sdk
+
+let random_string n = Bytes.create n
+
+let test_pipe pipe () =
+  let calf = Init.Fd.fd @@ Init.Pipe.(calf pipe) in
+  let priv = Init.Fd.fd @@ Init.Pipe.(priv pipe) in
+  let test str =
+    (* check the the pipe is unidirectional *)
+    IO.really_write calf str 0 (String.length str) >>= fun () ->
+    IO.read_all priv >>= fun buf ->
+    Alcotest.(check string) "stdout"
+      (String.Ascii.escape str) (String.Ascii.escape buf);
+    Lwt.catch (fun () ->
+        IO.really_write priv str 0 (String.length str) >|= fun () ->
+        Alcotest.fail "priv side is writable!"
+      ) (fun _ -> Lwt.return_unit)
+    >>= fun () ->
+    Lwt.catch (fun () ->
+        IO.read_all calf >|= fun _ ->
+        Alcotest.fail "calf sid is readable!"
+      ) (fun _ -> Lwt.return_unit)
+    >>= fun () ->
+    Lwt.return_unit
+  in
+  test (random_string 1) >>= fun () ->
+  test (random_string 100) >>= fun () ->
+  test (random_string 10241) >>= fun () ->
+
+  Lwt.return_unit
+
+let run f () =
+  try Lwt_main.run (f ())
+  with e ->
+    Fmt.epr "ERROR: %a" Fmt.exn e;
+    raise e
+
+let test_stderr () = ()
+
+let test = [
+  "stdout" , `Quick, run (test_pipe Init.Pipe.stdout);
+  "stdout" , `Quick, run (test_pipe Init.Pipe.stderr);
+  ]
+
+let reporter ?(prefix="") () =
+  let pad n x =
+    if String.length x > n then x
+    else x ^ String.v ~len:(n - String.length x) (fun _ -> ' ')
+  in
+  let report src level ~over k msgf =
+    let k _ = over (); k () in
+    let ppf = match level with Logs.App -> Fmt.stdout | _ -> Fmt.stderr in
+    let with_stamp h _tags k fmt =
+      let dt = Mtime.to_us (Mtime.elapsed ()) in
+      Fmt.kpf k ppf ("%s%+04.0fus %a %a @[" ^^ fmt ^^ "@]@.")
+        prefix
+        dt
+        Fmt.(styled `Magenta string) (pad 10 @@ Logs.Src.name src)
+        Logs_fmt.pp_header (level, h)
+    in
+    msgf @@ fun ?header ?tags fmt ->
+    with_stamp header tags k fmt
+  in
+  { Logs.report = report }
+
+let () =
+  Logs.set_level (Some Logs.Debug);
+  Logs.set_reporter (reporter ())
+
+let () = Alcotest.run "sdk" [
+    "init", test;
+  ]