albatross/app/vmm_prometheus_stats.ml
2018-05-02 15:36:53 +02:00

357 lines
13 KiB
OCaml

(* (c) 2017 Hannes Mehnert, all rights reserved *)
open Lwt.Infix
open Astring
open Vmm_core
let my_version = `WV0
let command = ref 1
let t : (Lwt_unix.file_descr * Lwt_unix.sockaddr * string) IM.t ref = ref IM.empty
module S = struct
type t = Lwt_unix.sockaddr
let compare : Lwt_unix.sockaddr -> Lwt_unix.sockaddr -> int = compare
end
module SM = Map.Make(S)
let count : int SM.t ref = ref SM.empty
let dec s =
match SM.find s !count with
| exception Not_found -> `Not_found
| 1 -> count := SM.remove s !count ; `Close
| x -> count := SM.add s (pred x) !count ; `Continue
let known_vms : string list ref = ref []
module P = struct
let p vm ?(typ = `Counter) name help value =
let t_s = function `Counter -> "counter" | `Gauge -> "gauge" in
let name = vm ^ "_" ^ name in
let p a v = String.concat ~sep:" " [ "#" ; a ; name ; v ] in
String.concat ~sep:"\n"
[ p "HELP" help ; p "TYPE" (t_s typ) ; name ^ " " ^ value ]
let tv (sec, usec) = Printf.sprintf "%Lu.%06d" sec usec
let i64 i = Printf.sprintf "%Lu" i
let encode_ru vm ru =
let p = p vm in
String.concat ~sep:"\n"
[ p "utime" "user time used" (tv ru.utime) ;
p "stime" "system time used" (tv ru.stime) ;
p "maxrss" "maximum resident set" (i64 ru.maxrss) ;
p ~typ:`Gauge "ixrss" "shared memory" (i64 ru.ixrss) ;
p ~typ:`Gauge "idrss" "unshared data" (i64 ru.idrss) ;
p ~typ:`Gauge "isrss" "unshared stack" (i64 ru.isrss) ;
p "minflt" "page reclaims" (i64 ru.minflt) ;
p "maxflt" "page faults" (i64 ru.majflt) ;
p "nswap" "swaps" (i64 ru.nswap) ;
p "inblock" "block input ops" (i64 ru.inblock) ;
p "outblock" "block output ops" (i64 ru.outblock) ;
p "msgsnd" "messages send" (i64 ru.msgsnd) ;
p "msgrcv" "messages received" (i64 ru.msgrcv) ;
p "nsignals" "signals received" (i64 ru.nsignals) ;
p "nvcsw" "voluntary context switches" (i64 ru.nvcsw) ;
p "nivcsw" "involuntary context switches" (i64 ru.nivcsw)
]
let encode_vmm vm xs =
let p = p vm in
let massage s =
let cutted = match String.cut ~sep:"umber of " s with
| Some (_, r) -> r
| None -> s
in
let cutted = match String.cut ~sep:"[" cutted with
| None -> cutted
| Some (l, r) -> match String.cut ~sep:"]" r with
| None -> cutted
| Some (l', r) when r = "" -> l ^ "_" ^ l'
| Some (l', r') -> l ^ "_" ^ l' ^ "_" ^ r'
in
let cutted =
List.fold_left (fun str sep ->
match String.cut ~sep str with
| None -> str
| Some (l, r) -> l ^ r)
cutted [ "%" ; "/" ; "-" ]
in
String.concat ~sep:"_" (String.cuts ~sep:" " cutted)
in
String.concat ~sep:"\n"
(List.map (fun (k, v) -> p (massage k) k (i64 v)) xs)
let i32 i = Printf.sprintf "%lu" i
let encode_if vm ifd =
let p = p (vm ^ "_" ^ ifd.name) in
String.concat ~sep:"\n"
(* TODO: flags *)
[ p ~typ:`Gauge "send_length" "length of send queue" (i32 ifd.send_length) ;
p "max_send_length" "maximum length of send queue" (i32 ifd.max_send_length) ;
p "send_drops" "drops in send queue" (i32 ifd.send_drops) ;
p ~typ:`Gauge "mtu" "maximum transmission unit" (i32 ifd.mtu) ;
p ~typ:`Gauge "baudrate" "linespeed" (i64 ifd.baudrate) ;
p "vm_to_host_packets" "packets from vm" (i64 ifd.input_packets) ;
p "vm_to_host_errors" "packet errors from vm" (i64 ifd.input_errors) ;
p "vm_to_host_bytes" "bytes from vm" (i64 ifd.input_bytes) ;
p "vm_to_host_mcast" "packets from vm via multicast" (i64 ifd.input_mcast) ;
p "vm_to_host_dropped" "packets dropped from vm" (i64 ifd.input_dropped) ;
p "collisions" "collisions on csma interface" (i64 ifd.collisions) ;
p "host_to_vm_packets" "packets to vm" (i64 ifd.output_packets) ;
p "host_to_vm_errors" "packet errors to vm" (i64 ifd.output_errors) ;
p "host_to_vm_bytes" "bytes to vm" (i64 ifd.output_bytes) ;
p "host_to_vm_mcast" "packets to vm via multicast" (i64 ifd.output_mcast) ;
p "host_to_vm_dropped" "packets dropped to vm" (i64 ifd.output_dropped)
]
end
(* just a reminder whether we already sent the initial "info" or not *)
let f_done = ref false
let process db tls hdr data =
let open Vmm_wire in
let open Rresult.R.Infix in
if not (version_eq hdr.version my_version) then begin
Logs.err (fun m -> m "unknown wire protocol version") ; Lwt.return_unit
end else
match hdr.tag with
| x when x = Client.log_msg_tag && not !f_done ->
f_done := true ;
(* issue initial "info" to get all the vm names *)
let out = Vmm_wire.Client.cmd Info !command my_version in
command := succ !command ;
Logs.debug (fun m -> m "writing %a over TLS" Cstruct.hexdump_pp (Cstruct.of_string out)) ;
(Vmm_tls.write_tls tls out >|= function
| Ok () -> ()
| Error _ -> Logs.err (fun m -> m "error while writing") ; ())
| _ ->
let r =
match hdr.tag with
| x when x = Client.log_msg_tag ->
Client.decode_log data >>= fun (hdr, event) ->
let nam = translate_serial db hdr.Vmm_core.Log.name in
begin match event with
| `VM_start _ -> known_vms := nam :: !known_vms
| `VM_stop _ -> known_vms := List.filter (fun m -> m <> nam) !known_vms
| _ -> ()
end ;
Ok `None
| x when x = Client.info_msg_tag ->
Client.decode_info data >>= fun vms ->
let vms = List.map (fun (name, _, _, _) -> translate_serial db name) vms in
known_vms := vms ;
Ok `None
| x when x = Client.stat_msg_tag ->
Client.decode_stat data >>= fun (ru, vmm, ifd) ->
begin match IM.find hdr.id !t with
| exception Not_found -> Logs.err (fun m -> m "unexpected reply") ; Ok `None
| (fd, s, vm) ->
t := IM.remove hdr.id !t ;
let out = String.concat ~sep:"\n" (P.encode_ru vm ru :: P.encode_vmm vm vmm :: List.map (P.encode_if vm) ifd @ [""]) in
Ok (`Stat (fd, s, out))
end
| x when x = fail_tag ->
let res =
match IM.find hdr.id !t with
| exception Not_found -> `None
| (fd, s, _) -> `Sockaddr (fd, s)
in
t := IM.remove hdr.id !t ;
decode_str data >>= fun (msg, _) ->
Logs.err (fun m -> m "failed %s" msg) ;
Ok res
| x -> Rresult.R.error_msgf "ignoring header tag %02X" x
in
let d (fd, s) = match dec s with
| `Continue -> Lwt.return_unit
| `Close -> Lwt_unix.close fd
| `Not_found -> Logs.err (fun m -> m "sockaddr not found") ; Lwt.return_unit
in
let open Lwt.Infix in
match r with
| Ok `None -> Lwt.return_unit
| Ok (`Sockaddr s) -> d s
| Ok (`Stat (fd, s, out)) ->
(Vmm_lwt.write_raw fd out >>= function
| Ok () -> d (fd, s)
| Error _ -> Logs.err (fun m -> m "exception while writing") ; Lwt.return_unit)
| Error (`Msg msg) -> Logs.err (fun m -> m "error while processing: %s" msg) ; Lwt.return_unit
let rec tls_listener db tls =
(Vmm_tls.read_tls tls >>= function
| Error (`Msg msg) ->
Logs.err (fun m -> m "error while reading %s" msg) ;
Lwt.return (Ok ())
| Error _ ->
Logs.err (fun m -> m "received exception in read_tls") ;
Lwt.return (Error ())
| Ok (hdr, data) ->
process db tls hdr data >>= fun () ->
Lwt.return (Ok ())) >>= function
| Ok () -> tls_listener db tls
| Error () -> Lwt.return_unit
let hdr =
String.concat ~sep:"\r\n"
[ "HTTP/1.1 200 OK" ;
"Content-Type: text/plain; version=0.0.4" ;
"\r\n" ]
(* wait for TCP connection, once received request stats from vmmd, and loop *)
let rec tcp_listener db tcp tls =
Lwt_unix.accept tcp >>= fun (cs, sockaddr) ->
Vmm_lwt.write_raw cs hdr >>= function
| Error _ -> Logs.err (fun m -> m "exception while accepting") ; Lwt.return_unit
| Ok () ->
let l = List.length !known_vms in
let ip, port = match sockaddr with Lwt_unix.ADDR_INET (ip, port) -> ip, port | _ -> invalid_arg "unexpected" in
Logs.info (fun m -> m "connection from %s:%d with %d known" (Unix.string_of_inet_addr ip) port l) ;
(if l = 0 then
Lwt_unix.close cs >|= fun () -> Error ()
else begin
count := SM.add sockaddr (List.length !known_vms) !count ;
Lwt_list.fold_left_s
(fun r vm ->
match r with
| Error () -> Lwt.return (Error ())
| Ok () ->
let vm_id = translate_name db vm in
let out = Vmm_wire.Client.cmd Statistics ~arg:vm_id !command my_version in
t := IM.add !command (cs, sockaddr, vm) !t ;
command := succ !command ;
Vmm_tls.write_tls tls out >|= function
| Ok () -> Ok ()
| Error _ -> Logs.err (fun m -> m "exception while writing") ; Error ())
(Ok ()) !known_vms
end) >>= function
| Ok () -> tcp_listener db tcp tls
| Error () -> Lwt.return_unit
let client cas host port cert priv_key db listen_ip listen_port =
Nocrypto_entropy_lwt.initialize () >>= fun () ->
let auth = if Sys.is_directory cas then `Ca_dir cas else `Ca_file cas in
X509_lwt.authenticator auth >>= fun authenticator ->
Lwt.catch (fun () ->
(* start TCP listening socket *)
let tcp = Lwt_unix.(socket PF_INET SOCK_STREAM 0) in
Lwt_unix.(setsockopt tcp SO_REUSEADDR true) ;
let addr = Lwt_unix.ADDR_INET (Ipaddr_unix.V4.to_inet_addr listen_ip, listen_port) in
Lwt_unix.bind tcp addr >>= fun () ->
Lwt_unix.listen tcp 1 ;
(* setup remote connection to VMMD *)
Lwt_unix.gethostbyname host >>= fun host_entry ->
let host_inet_addr = Array.get host_entry.Lwt_unix.h_addr_list 0 in
let fd = Lwt_unix.socket host_entry.Lwt_unix.h_addrtype Lwt_unix.SOCK_STREAM 0 in
Lwt_unix.connect fd (Lwt_unix.ADDR_INET (host_inet_addr, port)) >>= fun _ ->
X509_lwt.private_of_pems ~cert ~priv_key >>= fun cert ->
let certificates = `Single cert in
let client = Tls.Config.client ~reneg:true ~certificates ~authenticator () in
Tls_lwt.Unix.client_of_fd client (* ~host *) fd >>= fun tls ->
(* loop on both tcp and tls connections *)
Lwt.join [ tcp_listener db tcp tls ; tls_listener db tls ])
(fun exn ->
Logs.err (fun m -> m "failed to establish TLS connection: %s"
(Printexc.to_string exn)) ;
Lwt.return_unit)
let run_client _ cas cert key (host, port) db listen_ip listen_port =
Printexc.register_printer (function
| Tls_lwt.Tls_alert x -> Some ("TLS alert: " ^ Tls.Packet.alert_type_to_string x)
| Tls_lwt.Tls_failure f -> Some ("TLS failure: " ^ Tls.Engine.string_of_failure f)
| _ -> None) ;
Sys.(set_signal sigpipe Signal_ignore) ;
let db =
let open Rresult.R.Infix in
match db with
| None -> []
| Some db ->
match Bos.OS.File.read_lines (Fpath.v db) >>= parse_db with
| Ok db -> db
| Error (`Msg m) -> Logs.warn (fun f -> f "couldn't parse database %s" m) ; []
in
Lwt_main.run (client cas host port cert key db listen_ip listen_port)
let setup_log style_renderer level =
Fmt_tty.setup_std_outputs ?style_renderer ();
Logs.set_level level;
Logs.set_reporter (Logs_fmt.reporter ~dst:Format.std_formatter ())
open Cmdliner
let setup_log =
Term.(const setup_log
$ Fmt_cli.style_renderer ()
$ Logs_cli.level ())
let host_port : (string * int) Arg.converter =
let parse s =
match String.cut ~sep:":" s with
| None -> `Error "broken: no port specified"
| Some (hostname, port) ->
try
`Ok (hostname, int_of_string port)
with
Not_found -> `Error "failed to parse port"
in
parse, fun ppf (h, p) -> Format.fprintf ppf "%s:%d" h p
let cas =
let doc = "The full path to PEM encoded certificate authorities. Can either be a FILE or a DIRECTORY." in
Arg.(required & pos 0 (some string) None & info [] ~docv:"FILE" ~doc)
let client_cert =
let doc = "Use a client certificate chain" in
Arg.(required & pos 1 (some file) None & info [] ~doc)
let client_key =
let doc = "Use a client key" in
Arg.(required & pos 2 (some file) None & info [] ~doc)
let destination =
Arg.(required & pos 3 (some host_port) None & info [] ~docv:"destination"
~doc:"the destination hostname:port to connect to")
let ip : Ipaddr.V4.t Arg.converter =
let parse s =
try
`Ok (Ipaddr.V4.of_string_exn s)
with
Not_found -> `Error "broken"
in
parse, Ipaddr.V4.pp_hum
let address =
let doc = "Address to listen on" in
Arg.(value & opt ip (Ipaddr.V4.of_string_exn "127.0.0.1") & info [ "address" ] ~doc)
let port =
let doc = "TCP port to listen on" in
Arg.(value & opt int 9080 & info [ "port" ] ~doc)
let db =
let doc = "Certificate database" in
Arg.(value & opt (some file) None & info [ "db" ] ~doc)
let cmd =
let doc = "VMM Prometheus connector" in
let man = [
`S "DESCRIPTION" ;
`P "$(tname) connects to a VMMD to gather statistics and serves them for Prometheus via HTTP" ]
in
Term.(pure run_client $ setup_log $ cas $ client_cert $ client_key $ destination $ db $ address $ port),
Term.info "vmm_prometheus_stats" ~version:"%%VERSION_NUM%%" ~doc ~man
let () =
match Term.eval cmd
with `Error _ -> exit 1 | _ -> exit 0