package opentelemetry-client-cohttp-lwt

  1. Overview
  2. Docs

Source file opentelemetry_client_cohttp_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
128
129
130
131
132
133
(*
   https://github.com/open-telemetry/oteps/blob/main/text/0035-opentelemetry-protocol.md
   https://github.com/open-telemetry/oteps/blob/main/text/0099-otlp-http.md
 *)

module Config = Config
open Opentelemetry_client
open Opentelemetry
open Common_

type error = Export_error.t

open struct
  module IO = Opentelemetry_client_lwt.Io_lwt
end

module Httpc : Generic_http_consumer.HTTPC with module IO = IO = struct
  module IO = IO
  open Lwt.Syntax
  module Httpc = Cohttp_lwt_unix.Client

  type t = unit

  let create () : t = ()

  let cleanup _self = ()

  (* send the content to the remote endpoint/path *)
  let send (_self : t) ~attempt_descr ~url ~headers:user_headers ~decode
      (bod : string) : ('a, error) result Lwt.t =
    let uri = Uri.of_string url in

    let open Cohttp in
    let headers = Header.(add_list (init ()) user_headers) in

    let body = Cohttp_lwt.Body.of_string bod in

    let* r =
      try%lwt
        let+ r = Httpc.post ~headers ~body uri in
        Ok r
      with e -> Lwt.return @@ Error e
    in
    match r with
    | Error e ->
      let err =
        `Failure
          (spf "sending signals via http POST to %S\nfailed with:\n%s" url
             (Printexc.to_string e))
      in
      Lwt.return @@ Error err
    | Ok (resp, body) ->
      let* body = Cohttp_lwt.Body.to_string body in
      let code = Response.status resp |> Code.code_of_status in
      if not (Code.is_error code) then (
        match decode with
        | `Ret x -> Lwt.return @@ Ok x
        | `Dec f ->
          let dec = Pbrt.Decoder.of_string body in
          let r =
            try Ok (f dec)
            with e ->
              let bt = Printexc.get_backtrace () in
              Error
                (`Failure
                   (spf "decoding failed with:\n%s\n%s" (Printexc.to_string e)
                      bt))
          in
          Lwt.return r
      ) else
        Lwt.return
          (Error
             (Export_error.decode_invalid_http_response ~attempt_descr ~url
                ~code body))
end

module Consumer_impl =
  Generic_http_consumer.Make (IO) (Opentelemetry_client_lwt.Notifier_lwt)
    (Httpc)

let create_consumer ?(config = Config.make ()) () =
  Consumer_impl.consumer ~ticker_task:(Some 0.5) ~on_tick:OTEL.Sdk.tick ~config
    ()

let create_exporter ?(config = Config.make ()) () =
  let consumer = create_consumer ~config () in
  let bq =
    Opentelemetry_client_sync.Bounded_queue_sync.create
      ~measure:OTEL.Any_signal_l.length
      ~high_watermark:Bounded_queue.Defaults.high_watermark ()
  in
  Exporter_queued.create ~clock:Clock.ptime_clock ~q:bq ~consumer ()

let create_backend = create_exporter

let setup_ ~config () : unit =
  Opentelemetry_client_lwt.Util_ambient_context.setup_ambient_context ();
  let exp = create_exporter ~config () in
  Sdk.set ~traces:config.traces ~metrics:config.metrics ~logs:config.logs
    ?sampler:config.sampler exp;

  Option.iter
    (fun min_level -> Opentelemetry.Self_debug.to_stderr ~min_level ())
    config.log_level;

  Opentelemetry.Self_debug.log Opentelemetry.Self_debug.Info (fun () ->
      "opentelemetry: cohttp-lwt exporter installed");
  Opentelemetry_client.Self_trace.set_enabled config.self_trace;
  if config.self_metrics then Opentelemetry.Sdk.setup_self_metrics ();
  ()

let setup ?(config = Config.make ()) ?(enable = true) () =
  if enable && not config.sdk_disabled then setup_ ~config ()

let remove_exporter () : unit Lwt.t =
  let done_fut, done_u = Lwt.wait () in
  (* Printf.eprintf "otel.client.cohttp-lwt: removing…\n%!"; *)
  Sdk.remove
    ~on_done:(fun () ->
      (* Printf.eprintf "otel.client.cohttp-lwt: done removing\n%!"; *)
      Lwt.wakeup_later done_u ())
    ();
  done_fut

let remove_backend = remove_exporter

let with_setup ?(config = Config.make ()) ?(enable = true) () f : _ Lwt.t =
  if enable && not config.sdk_disabled then (
    setup_ ~config ();

    Lwt.finalize f remove_exporter
  ) else
    f ()