Source file logs_syslog_lwt.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
open Lwt.Infix
open Logs_syslog_lwt_common
open Logs_syslog

let udp_reporter ?hostname ip ?(port = 514) ?(truncate = 65535) ?facility () =
  let sa = Lwt_unix.ADDR_INET (ip, port) in
  let s = Lwt_unix.(socket PF_INET SOCK_DGRAM 0) in
  let send msg =
    Lwt.catch (fun () ->
        let b = Bytes.of_string msg in
        Lwt_unix.sendto s b 0 (String.length msg) [] sa >|= fun _ -> ())
      (function
        | Unix.Unix_error (e, f, _) ->
          Printf.eprintf "error in %s %s while sending to %s:%d\n%s %s\n"
            f (Unix.error_message e) (Unix.string_of_inet_addr ip) port
            (Ptime.to_rfc3339 (Ptime_clock.now ()))
            msg ;
          Lwt.return_unit
        | exn -> Lwt.fail exn)
  in
  (match hostname with
   | Some x -> Lwt.return x
   | None -> Lwt_unix.gethostname ()) >|= fun host ->
  syslog_report_common facility host truncate Ptime_clock.now send
                       Syslog_message.encode

let conn_reporter sd st sa truncate frame encode hostname facility =
  let s = ref None in
  let m = Lwt_mutex.create () in
  let connect () =
    let sock = Lwt_unix.(socket sd st 0) in
    Lwt_unix.(setsockopt sock SO_REUSEADDR true) ;
    Lwt_unix.(setsockopt sock SO_KEEPALIVE true) ;
    Lwt.catch
      (fun () -> Lwt_unix.connect sock sa >|= fun () -> s := Some sock ; Ok ())
      (function Unix.Unix_error (e, f, _) ->
         let endpoint =
           match sa with
           | Unix.ADDR_UNIX socket -> socket
           | Unix.ADDR_INET (ip, port) ->
               Printf.sprintf "%s:%d" (Unix.string_of_inet_addr ip) port
         in
         let err =
           Printf.sprintf "error %s in function %s while connecting to %s"
             (Unix.error_message e) f endpoint
         in
         Lwt.return (Error err)
       | exn -> Lwt.fail exn)
  in
  let reconnect k msg =
    Lwt_mutex.lock m >>= fun () ->
    (match !s with
     | None -> connect ()
     | Some _ -> Lwt.return (Ok ())) >>= function
    | Ok () -> Lwt_mutex.unlock m ; k msg
    | Error e ->
      Printf.eprintf "%s while sending syslog message\n%s %s\n"
        e (Ptime.to_rfc3339 (Ptime_clock.now ())) msg ;
      Lwt_mutex.unlock m ;
      Lwt.return_unit
  in
  connect () >>= function
  | Error e -> Lwt.return (Error e)
  | Ok () ->
    let transmit =
      if st = Unix.SOCK_DGRAM then
        fun sock b len () -> Lwt_unix.send sock b 0 len [] >|= fun _ -> ()
      else
        fun sock b len () ->
          let rec aux idx =
            let should = len - idx in
            Lwt_unix.send sock b idx (len - idx) [] >>= fun n ->
              if n = should then
                Lwt.return_unit
              else
                aux (idx + n)
          in
          aux 0
    in
    let rec send omsg =
      match !s with
      | None -> reconnect send omsg
      | Some sock ->
        let msg = frame omsg |> Bytes.of_string in
        (Lwt.catch (transmit sock msg (Bytes.length msg))
           (function
             | Unix.Unix_error (e, f, _) ->
               s := None ;
               let err = Unix.error_message e in
               Printf.eprintf "error %s in function %s, reconnecting\n" err f ;
               Lwt.catch
                 (fun () -> Lwt_unix.close sock)
                 (function
                   | Unix.Unix_error _ -> Lwt.return_unit
                   | exn -> Lwt.fail exn) >>= fun () ->
               reconnect send omsg
             | exn -> Lwt.fail exn))
    in
    at_exit (fun () -> match !s with
        | None -> ()
        | Some x -> Lwt.async (fun () -> Lwt_unix.close x)) ;
    Lwt.return (Ok (syslog_report_common facility hostname truncate
                                         Ptime_clock.now send encode))

let tcp_reporter ?hostname ip ?(port = 514) ?(truncate = 0) ?(framing = `Null) ?facility () =
  let sa = Lwt_unix.ADDR_INET (ip, port) in
  let frame msg = frame_message msg framing in
  let encode = Syslog_message.encode in
  (match hostname with
   | Some x -> Lwt.return x
   | None -> Lwt_unix.gethostname ()) >>= fun host ->
  conn_reporter Unix.PF_INET Unix.SOCK_STREAM sa truncate frame encode host facility

let unix_reporter ?(socket = "/dev/log") ?truncate ?framing ?facility () =
  let truncate =
    match truncate with
    | Some truncate -> truncate
    | None -> if framing = None then 65536 else 0
  in
  let frame, socket_type =
    match framing with
    | Some framing -> (fun msg -> frame_message msg framing), Unix.SOCK_STREAM
    | None -> (fun msg -> msg), Unix.SOCK_DGRAM
  in
  let sa = Unix.ADDR_UNIX socket in
  let encode = Syslog_message.encode_local in
  conn_reporter Unix.PF_UNIX socket_type sa truncate frame encode "localhost" facility