albatross/daemon/albatross_log.ml

200 lines
6.7 KiB
OCaml
Raw Normal View History

2017-05-26 14:30:34 +00:00
(* (c) 2017 Hannes Mehnert, all rights reserved *)
(* the process responsible for event log *)
(* communication channel is a single unix domain socket shared between vmmd and
vmm_log. There are two commands from vmmd to vmm_log, history and data. *)
(* internally, a ring buffer for the last N events is preserved in memory
each new event is directly written to disk! *)
open Lwt.Infix
2018-11-12 21:11:06 +00:00
let my_version = `AV3
2017-05-26 14:30:34 +00:00
let broadcast prefix wire t =
Lwt_list.fold_left_s (fun t (id, s) ->
Vmm_lwt.write_wire s wire >|= function
| Ok () -> t
2018-09-28 20:44:38 +00:00
| Error `Exception -> Vmm_trie.remove id t)
t (Vmm_trie.collect prefix t)
let write_complete s cs =
let l = Cstruct.len cs in
let b = Cstruct.to_bytes cs in
2017-05-26 14:30:34 +00:00
let rec w off =
let len = l - off in
Lwt_unix.write s b off len >>= fun n ->
if n = len then Lwt.return_unit else w (off + n)
in
w 0
let read_from_file file =
2018-10-28 00:03:27 +00:00
Vmm_lwt.read_from_file file >|= fun data ->
let logs = Vmm_asn.logs_of_disk data in
List.rev logs
let write_to_file mvar file =
let get_fd () =
Lwt_unix.openfile file Lwt_unix.[O_APPEND;O_CREAT;O_WRONLY] 0o600
in
let rec loop ?(retry = true) ?log_entry fd =
(match log_entry with
| Some l -> Lwt.return l
| None -> Lwt_mvar.take mvar >>= function
| `Hup ->
Vmm_lwt.safe_close fd >>= fun () ->
get_fd () >>= fun fd ->
loop ~log_entry:(Ptime_clock.now (), `Hup) fd
| `Entry log_entry -> Lwt.return log_entry) >>= fun log_entry ->
let data = Vmm_asn.log_to_disk my_version log_entry in
Lwt.catch
(fun () ->
write_complete fd data >>= fun () ->
loop fd)
(fun e ->
Logs.err (fun m -> m "exception %s while writing" (Printexc.to_string e)) ;
Vmm_lwt.safe_close fd >>= fun () ->
if retry then
get_fd () >>= fun fd ->
loop ~retry:false ~log_entry fd
else begin
Logs.err (fun m -> m "retry is false, exiting") ;
Lwt.return log_entry
end)
in
get_fd () >>= fun fd ->
loop fd >|= fun _ ->
()
let send_history s ring id ts =
let elements =
match ts with
| None -> Vmm_ring.read ring
| Some since -> Vmm_ring.read_history ring since
in
2018-09-28 20:44:38 +00:00
let res =
List.fold_left (fun acc (ts, event) ->
let sub = Vmm_core.Log.name event in
if Vmm_core.Name.is_sub ~super:id ~sub
then (ts, event) :: acc
else acc)
2018-09-28 20:44:38 +00:00
[] elements
in
(* just need a wrapper in tag = Log.Data, id = reqid *)
2018-10-23 22:03:36 +00:00
Lwt_list.fold_left_s (fun r (ts, event) ->
2018-09-28 20:44:38 +00:00
match r with
2018-10-23 22:03:36 +00:00
| Ok () ->
let header = Vmm_commands.{ version = my_version ; sequence = 0L ; name = id } in
2018-10-23 22:03:36 +00:00
Vmm_lwt.write_wire s (header, `Data (`Log_data (ts, event)))
2018-09-28 20:44:38 +00:00
| Error e -> Lwt.return (Error e))
(Ok ()) (List.rev res)
2018-09-28 20:44:38 +00:00
2018-10-26 19:35:40 +00:00
let tree = ref Vmm_trie.empty
2018-10-25 22:43:37 +00:00
let handle_data s mvar ring hdr entry =
if not (Vmm_commands.version_eq hdr.Vmm_commands.version my_version) then begin
Logs.warn (fun m -> m "unsupported version") ;
Lwt.return_unit
end else begin
2018-10-25 22:43:37 +00:00
Vmm_lwt.write_wire s (hdr, `Success `Empty) >>= fun _ ->
Vmm_ring.write ring entry ;
Lwt_mvar.put mvar (`Entry entry) >>= fun () ->
let data' = (hdr, `Data (`Log_data entry)) in
broadcast hdr.Vmm_commands.name data' !tree >|= fun tree' ->
tree := tree'
end
let read_data mvar ring s =
2017-05-26 14:30:34 +00:00
let rec loop () =
Vmm_lwt.read_wire s >>= function
| Error _ ->
Logs.err (fun m -> m "error while reading") ;
Lwt.return_unit
| Ok (hdr, `Data (`Log_data entry)) ->
2018-10-25 22:43:37 +00:00
handle_data s mvar ring hdr entry >>= fun () ->
2017-05-26 14:30:34 +00:00
loop ()
| Ok wire ->
Logs.warn (fun m -> m "unexpected wire %a" Vmm_commands.pp_wire wire) ;
Lwt.return_unit
in
loop ()
2019-10-10 20:26:36 +00:00
let handle mvar ring s addr =
Logs.info (fun m -> m "handling connection from %a" Vmm_lwt.pp_sockaddr addr) ;
Vmm_lwt.read_wire s >>= begin function
| Error _ ->
Logs.err (fun m -> m "error while reading") ;
Lwt.return_unit
| Ok (hdr, `Data (`Log_data entry)) ->
2018-10-25 22:43:37 +00:00
handle_data s mvar ring hdr entry >>= fun () ->
read_data mvar ring s
2018-10-22 21:20:00 +00:00
| Ok (hdr, `Command (`Log_cmd lc)) ->
2018-10-23 22:03:36 +00:00
if not (Vmm_commands.version_eq hdr.Vmm_commands.version my_version) then begin
2018-10-22 21:20:00 +00:00
Logs.warn (fun m -> m "unsupported version") ;
Lwt.return_unit
end else begin
match lc with
| `Log_subscribe ts ->
let tree', ret = Vmm_trie.insert hdr.Vmm_commands.name s !tree in
2018-10-22 21:20:00 +00:00
tree := tree' ;
(match ret with
| None -> Lwt.return_unit
| Some s' -> Vmm_lwt.safe_close s') >>= fun () ->
let out = `Success `Empty in
Vmm_lwt.write_wire s (hdr, out) >>= function
| Error _ -> Logs.err (fun m -> m "error while sending reply for subscribe") ;
2018-10-22 21:20:00 +00:00
Lwt.return_unit
| Ok () ->
send_history s ring hdr.Vmm_commands.name ts >>= function
| Error _ -> Logs.err (fun m -> m "error while sending history") ; Lwt.return_unit
| Ok () ->
(* command processing is finished, but we leave the socket open
until read returns (either with a message we ignore or a failure from the closed connection) *)
Vmm_lwt.read_wire s >|= fun _ -> ()
2018-10-22 21:20:00 +00:00
end
2018-10-22 22:02:05 +00:00
| Ok wire ->
2018-10-23 22:03:36 +00:00
Logs.warn (fun m -> m "ignoring %a" Vmm_commands.pp_wire wire) ;
Lwt.return_unit
end >>= fun () ->
Vmm_lwt.safe_close s
2017-05-26 14:30:34 +00:00
2019-10-10 20:26:36 +00:00
let m = Albatross_cli.conn_metrics "unix"
let jump _ file influx =
2017-05-26 14:30:34 +00:00
Sys.(set_signal sigpipe Signal_ignore) ;
Lwt_main.run
2019-10-10 20:26:36 +00:00
(Albatross_cli.init_influx "albatross_log" influx;
Vmm_lwt.server_socket `Log >>= fun s ->
let ring = Vmm_ring.create `Startup () in
read_from_file file >>= fun entries ->
Logs.app (fun m -> m "read %d entries from disk" (List.length entries)) ;
List.iter (Vmm_ring.write ring) entries ;
let mvar = Lwt_mvar.create_empty () in
Sys.(set_signal sighup (Signal_handle (fun _ ->
Lwt.async (fun () -> Lwt_mvar.put mvar `Hup)))) ;
Lwt.async (fun () -> write_to_file mvar file) ;
let start = Ptime_clock.now (), `Startup in
Lwt_mvar.put mvar (`Entry start) >>= fun () ->
Vmm_ring.write ring start ;
2017-05-26 14:30:34 +00:00
let rec loop () =
Lwt_unix.accept s >>= fun (cs, addr) ->
2019-10-10 20:26:36 +00:00
m `Open;
Lwt.async (fun () -> handle mvar ring cs addr >|= fun () -> m `Close) ;
2017-05-26 14:30:34 +00:00
loop ()
in
2019-03-27 23:11:43 +00:00
loop ())
2017-05-26 14:30:34 +00:00
open Cmdliner
2019-03-27 23:11:43 +00:00
open Albatross_cli
2017-05-26 14:30:34 +00:00
let file =
let doc = "File to write the log to" in
Arg.(value & opt string "/var/log/albatross" & info [ "logfile" ] ~doc)
2017-05-26 14:30:34 +00:00
let cmd =
2019-10-10 20:26:36 +00:00
Term.(term_result (const jump $ setup_log $ file $ influx)),
2019-03-27 23:11:43 +00:00
Term.info "albatross_log" ~version:"%%VERSION_NUM%%"
2017-05-26 14:30:34 +00:00
let () = match Term.eval cmd with `Ok () -> exit 0 | _ -> exit 1