Merge pull request #22 from imandra-ai/wip-format

format
This commit is contained in:
Simon Cruanes 2022-05-05 12:13:52 -04:00 committed by GitHub
commit 0fb530125a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 446 additions and 385 deletions

4
dune
View file

@ -1,3 +1,3 @@
(env
(_ (flags :standard -warn-error -a+8)))
(_
(flags :standard -warn-error -a+8)))

View file

@ -1,4 +1,3 @@
(**************************************************************************)
(* *)
(* OCaml *)
@ -19,34 +18,34 @@
(** Atomic references.
*)
(** An atomic (mutable) reference to a value of type ['a]. *)
type 'a t = 'a Stdlib.Atomic.t
(** An atomic (mutable) reference to a value of type ['a]. *)
(** Create an atomic reference. *)
val make : 'a -> 'a t
(** Create an atomic reference. *)
(** Get the current value of the atomic reference. *)
val get : 'a t -> 'a
(** Get the current value of the atomic reference. *)
(** Set a new value for the atomic reference. *)
val set : 'a t -> 'a -> unit
(** Set a new value for the atomic reference. *)
(** Set a new value for the atomic reference, and return the current value. *)
val exchange : 'a t -> 'a -> 'a
(** Set a new value for the atomic reference, and return the current value. *)
val compare_and_set : 'a t -> 'a -> 'a -> bool
(** [compare_and_set r seen v] sets the new value of [r] to [v] only
if its current value is physically equal to [seen] -- the
comparison and the set occur atomically. Returns [true] if the
comparison succeeded (so the set happened) and [false]
otherwise. *)
val compare_and_set : 'a t -> 'a -> 'a -> bool
val fetch_and_add : int t -> int -> int
(** [fetch_and_add r n] atomically increments the value of [r] by [n],
and returns the current value (before the increment). *)
val fetch_and_add : int t -> int -> int
(** [incr r] atomically increments the value of [r] by [1]. *)
val incr : int t -> unit
(** [incr r] atomically increments the value of [r] by [1]. *)
(** [decr r] atomically decrements the value of [r] by [1]. *)
val decr : int t -> unit
(** [decr r] atomically decrements the value of [r] by [1]. *)

View file

@ -1,4 +1,3 @@
(**************************************************************************)
(* *)
(* OCaml *)
@ -19,34 +18,34 @@
(** Atomic references.
*)
(** An atomic (mutable) reference to a value of type ['a]. *)
type 'a t
(** An atomic (mutable) reference to a value of type ['a]. *)
(** Create an atomic reference. *)
val make : 'a -> 'a t
(** Create an atomic reference. *)
(** Get the current value of the atomic reference. *)
val get : 'a t -> 'a
(** Get the current value of the atomic reference. *)
(** Set a new value for the atomic reference. *)
val set : 'a t -> 'a -> unit
(** Set a new value for the atomic reference. *)
(** Set a new value for the atomic reference, and return the current value. *)
val exchange : 'a t -> 'a -> 'a
(** Set a new value for the atomic reference, and return the current value. *)
val compare_and_set : 'a t -> 'a -> 'a -> bool
(** [compare_and_set r seen v] sets the new value of [r] to [v] only
if its current value is physically equal to [seen] -- the
comparison and the set occur atomically. Returns [true] if the
comparison succeeded (so the set happened) and [false]
otherwise. *)
val compare_and_set : 'a t -> 'a -> 'a -> bool
val fetch_and_add : int t -> int -> int
(** [fetch_and_add r n] atomically increments the value of [r] by [n],
and returns the current value (before the increment). *)
val fetch_and_add : int t -> int -> int
(** [incr r] atomically increments the value of [r] by [1]. *)
val incr : int t -> unit
(** [incr r] atomically increments the value of [r] by [1]. *)
(** [decr r] atomically decrements the value of [r] by [1]. *)
val decr : int t -> unit
(** [decr r] atomically decrements the value of [r] by [1]. *)

View file

@ -1,15 +1,15 @@
(library
(name opentelemetry_atomic)
(synopsis "Compatibility package for the Atomic module for opentelemetry")
(public_name opentelemetry.atomic)
(modules atomic))
(name opentelemetry_atomic)
(synopsis "Compatibility package for the Atomic module for opentelemetry")
(public_name opentelemetry.atomic)
(modules atomic))
(executable
(modules gen)
(name gen))
(modules gen)
(name gen))
(rule
(targets atomic.ml atomic.mli atomic.ml)
(deps atomic.pre412.mli atomic.post412.mli)
(action (run ./gen.exe)))
(targets atomic.ml atomic.mli atomic.ml)
(deps atomic.pre412.mli atomic.post412.mli)
(action
(run ./gen.exe)))

View file

@ -1,6 +1,5 @@
let atomic_before_412 = {|
let atomic_before_412 =
{|
type 'a t = {mutable x: 'a}
let[@inline] make x = {x}
let[@inline] get {x} = x
@ -32,7 +31,9 @@ let atomic_before_412 = {|
let atomic_after_412 = {|include Stdlib.Atomic|}
let write_file file s =
let oc = open_out file in output_string oc s; close_out oc
let oc = open_out file in
output_string oc s;
close_out oc
let copy_file file1 file2 =
let oc = open_out file2 in
@ -41,14 +42,22 @@ let copy_file file1 file2 =
try
while true do
let n = input ic buf 0 (Bytes.length buf) in
if n=0 then raise End_of_file;
if n = 0 then raise End_of_file;
output oc buf 0 n
done
with End_of_file -> ()
let () =
let version = Scanf.sscanf Sys.ocaml_version "%d.%d.%s" (fun x y _ -> x,y) in
write_file "atomic.ml" (if version >= (4,12) then atomic_after_412 else atomic_before_412);
copy_file (if version >= (4,12) then "atomic.post412.mli" else "atomic.pre412.mli") "atomic.mli" ;
let version = Scanf.sscanf Sys.ocaml_version "%d.%d.%s" (fun x y _ -> x, y) in
write_file "atomic.ml"
(if version >= (4, 12) then
atomic_after_412
else
atomic_before_412);
copy_file
(if version >= (4, 12) then
"atomic.post412.mli"
else
"atomic.pre412.mli")
"atomic.mli";
()

View file

@ -1,13 +1,21 @@
module Atomic = Opentelemetry_atomic.Atomic
type 'a t = 'a list Atomic.t
let make () = Atomic.make []
let add self x =
while
let old = Atomic.get self in
let l' = x :: old in
not (Atomic.compare_and_set self old l')
do () done
do
()
done
let rec pop_all self =
let l = Atomic.get self in
if Atomic.compare_and_set self l [] then l else pop_all self
if Atomic.compare_and_set self l [] then
l
else
pop_all self

View file

@ -1,6 +1,9 @@
(** Atomic list *)
type 'a t
val make : unit -> 'a t
val add : 'a t -> 'a -> unit
val pop_all : 'a t -> 'a list

View file

@ -1,5 +1,3 @@
type 'a t = {
arr: 'a array;
mutable i: int;
@ -7,23 +5,23 @@ type 'a t = {
let create ~dummy n : _ t =
assert (n >= 1);
{ arr=Array.make n dummy;
i=0;
}
{ arr = Array.make n dummy; i = 0 }
let[@inline] size self = self.i
let[@inline] is_full self = self.i = Array.length self.arr
let push (self:_ t) x : bool =
if is_full self then false
let push (self : _ t) x : bool =
if is_full self then
false
else (
self.arr.(self.i) <- x;
self.i <- 1 + self.i;
true
)
let pop_iter_all (self: _ t) f =
for j=0 to self.i-1 do
let pop_iter_all (self : _ t) f =
for j = 0 to self.i - 1 do
f self.arr.(j)
done;
self.i <- 0

View file

@ -1,9 +1,11 @@
(** queue of fixed size *)
type 'a t
val create : dummy:'a -> int -> 'a t
val size : _ t -> int
val push : 'a t -> 'a -> bool (* true iff it could write element *)
val pop_iter_all : 'a t -> ('a -> unit) -> unit
val create : dummy:'a -> int -> 'a t
val size : _ t -> int
val push : 'a t -> 'a -> bool (* true iff it could write element *)
val pop_iter_all : 'a t -> ('a -> unit) -> unit

View file

@ -1,4 +1,3 @@
open Common_
type t = {
@ -14,28 +13,36 @@ type t = {
let pp out self =
let ppiopt = Format.pp_print_option Format.pp_print_int in
let pp_header ppf (a, b) =
Format.fprintf ppf "@[%s: @,%s@]@." a b ;
in
let pp_header ppf (a, b) = Format.fprintf ppf "@[%s: @,%s@]@." a b in
let ppheaders = Format.pp_print_list pp_header in
let {debug; url; headers; batch_traces; batch_metrics;
batch_timeout_ms; thread; ticker_thread} = self in
let {
debug;
url;
headers;
batch_traces;
batch_metrics;
batch_timeout_ms;
thread;
ticker_thread;
} =
self
in
Format.fprintf out
"{@[ debug=%B;@ url=%S;@ headers=%a;@ \
batch_traces=%a;@ batch_metrics=%a;@ \
"{@[ debug=%B;@ url=%S;@ headers=%a;@ batch_traces=%a;@ batch_metrics=%a;@ \
batch_timeout_ms=%d; thread=%B;@ ticker_thread=%B @]}"
debug url ppheaders headers ppiopt batch_traces ppiopt batch_metrics
batch_timeout_ms thread ticker_thread
let make
?(debug= !debug_)
?(url= get_url())
?(headers= get_headers ())
?(batch_traces=Some 400)
?(batch_metrics=None)
?(batch_timeout_ms=500)
?(thread=true)
?(ticker_thread=true)
() : t =
{ debug; url; headers; batch_traces; batch_metrics; batch_timeout_ms;
thread; ticker_thread; }
let make ?(debug = !debug_) ?(url = get_url ()) ?(headers = get_headers ())
?(batch_traces = Some 400) ?(batch_metrics = None) ?(batch_timeout_ms = 500)
?(thread = true) ?(ticker_thread = true) () : t =
{
debug;
url;
headers;
batch_traces;
batch_metrics;
batch_timeout_ms;
thread;
ticker_thread;
}

View file

@ -1,56 +1,49 @@
type t = {
debug: bool;
url: string;
(** Url of the endpoint. Default is "http://localhost:4318",
(** Url of the endpoint. Default is "http://localhost:4318",
or "OTEL_EXPORTER_OTLP_ENDPOINT" if set. *)
headers: (string * string) list;
(** API headers sent to the endpoint. Default is none or
(** API headers sent to the endpoint. Default is none or
"OTEL_EXPORTER_OTLP_HEADERS" if set. *)
batch_traces: int option;
(** Batch traces? If [Some i], then this produces batches of (at most)
(** Batch traces? If [Some i], then this produces batches of (at most)
[i] items. If [None], there is no batching.
Note that traces and metrics are batched separately.
Default [Some 400].
*)
batch_metrics: int option;
(** Batch metrics? If [Some i], then this produces batches of (at most)
(** Batch metrics? If [Some i], then this produces batches of (at most)
[i] items. If [None], there is no batching.
Note that traces and metrics are batched separately.
Default [None].
*)
batch_timeout_ms: int;
(** Number of milliseconds after which we will emit a batch, even
(** Number of milliseconds after which we will emit a batch, even
incomplete.
Note that the batch might take longer than that, because this is
only checked when a new event occurs. Default 500. *)
thread: bool;
(** Is there a background thread? Default [true] *)
thread: bool; (** Is there a background thread? Default [true] *)
ticker_thread: bool;
(** Is there a ticker thread? Default [true].
(** Is there a ticker thread? Default [true].
This thread will regularly call [tick()] on the backend, to make
sure it makes progress, and regularly send events to the collector.
This option is ignored if [thread=false]. *)
}
val make :
?debug:bool -> ?url:string ->
?debug:bool ->
?url:string ->
?headers:(string * string) list ->
?batch_traces:int option ->
?batch_metrics:int option ->
?batch_timeout_ms:int ->
?thread:bool ->
?ticker_thread:bool ->
unit -> t
unit ->
t
(** Make a configuration *)
val pp : Format.formatter -> t -> unit

View file

@ -1,8 +1,5 @@
(library
(name opentelemetry_client_ocurl)
(public_name opentelemetry-client-ocurl)
(libraries opentelemetry opentelemetry.atomic
curl pbrt threads
mtime mtime.clock.os))
(name opentelemetry_client_ocurl)
(public_name opentelemetry-client-ocurl)
(libraries opentelemetry opentelemetry.atomic curl pbrt threads mtime
mtime.clock.os))

View file

@ -1,4 +1,3 @@
(*
TODO: more options from
https://opentelemetry.io/docs/reference/specification/protocol/exporter/

149
src/dune
View file

@ -8,88 +8,91 @@
; ### protobuf rules ###
(rule
(alias lint)
(mode promote)
(targets status_types.ml status_types.mli
status_pb.ml status_pb.mli
status_pp.ml status_pp.mli)
(deps (:file status.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action (run ocaml-protoc %{file} -ml_out . -pp -binary)))
(alias lint)
(mode promote)
(targets status_types.ml status_types.mli status_pb.ml status_pb.mli
status_pp.ml status_pp.mli)
(deps
(:file status.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action
(run ocaml-protoc %{file} -ml_out . -pp -binary)))
(rule
(alias lint)
(mode promote)
(targets common_types.ml common_types.mli
common_pb.ml common_pb.mli
common_pp.ml common_pp.mli)
(deps
(:file %{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/common/v1/common.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action (run ocaml-protoc %{file}
-I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(alias lint)
(mode promote)
(targets common_types.ml common_types.mli common_pb.ml common_pb.mli
common_pp.ml common_pp.mli)
(deps
(:file
%{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/common/v1/common.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action
(run ocaml-protoc %{file} -I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(rule
(alias lint)
(mode promote)
(targets resource_types.ml resource_types.mli
resource_pb.ml resource_pb.mli
resource_pp.ml resource_pp.mli)
(deps
(:file %{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/resource/v1/resource.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action (run ocaml-protoc %{file}
-I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(alias lint)
(mode promote)
(targets resource_types.ml resource_types.mli resource_pb.ml resource_pb.mli
resource_pp.ml resource_pp.mli)
(deps
(:file
%{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/resource/v1/resource.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action
(run ocaml-protoc %{file} -I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(rule
(alias lint)
(mode promote)
(targets trace_types.ml trace_types.mli
trace_pb.ml trace_pb.mli
trace_pp.ml trace_pp.mli)
(deps
(:file %{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/trace/v1/trace.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action (run ocaml-protoc %{file}
-I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(alias lint)
(mode promote)
(targets trace_types.ml trace_types.mli trace_pb.ml trace_pb.mli trace_pp.ml
trace_pp.mli)
(deps
(:file
%{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/trace/v1/trace.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action
(run ocaml-protoc %{file} -I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(rule
(alias lint)
(mode promote)
(targets metrics_types.ml metrics_types.mli
metrics_pb.ml metrics_pb.mli
metrics_pp.ml metrics_pp.mli)
(deps
(:file %{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/metrics/v1/metrics.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action (run ocaml-protoc %{file}
-I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(alias lint)
(mode promote)
(targets metrics_types.ml metrics_types.mli metrics_pb.ml metrics_pb.mli
metrics_pp.ml metrics_pp.mli)
(deps
(:file
%{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/metrics/v1/metrics.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action
(run ocaml-protoc %{file} -I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(rule
(alias lint)
(mode promote)
(targets metrics_service_types.ml metrics_service_types.mli
metrics_service_pp.ml metrics_service_pp.mli
metrics_service_pb.ml metrics_service_pb.mli)
(deps (:file %{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action (run ocaml-protoc %{file}
-I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(alias lint)
(mode promote)
(targets metrics_service_types.ml metrics_service_types.mli
metrics_service_pp.ml metrics_service_pp.mli metrics_service_pb.ml
metrics_service_pb.mli)
(deps
(:file
%{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action
(run ocaml-protoc %{file} -I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(rule
(alias lint)
(mode promote)
(targets trace_service_types.ml trace_service_types.mli
trace_service_pp.ml trace_service_pp.mli
trace_service_pb.ml trace_service_pb.mli)
(deps
(:file %{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/collector/trace/v1/trace_service.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action (run ocaml-protoc %{file}
-I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))
(alias lint)
(mode promote)
(targets trace_service_types.ml trace_service_types.mli trace_service_pp.ml
trace_service_pp.mli trace_service_pb.ml trace_service_pb.mli)
(deps
(:file
%{project_root}/vendor/opentelemetry-proto/opentelemetry/proto/collector/trace/v1/trace_service.proto)
(source_tree %{project_root}/vendor/opentelemetry-proto/))
(action
(run ocaml-protoc %{file} -I %{project_root}/vendor/opentelemetry-proto/
-ml_out . -pp -binary)))

View file

@ -4,6 +4,14 @@ open Cohttp
open Cohttp_lwt
module Server : sig
val trace :
?service_name:string ->
?attrs:Otel.Span.key_value list ->
('conn -> Request.t -> 'body -> (Response.t * 'body) Lwt.t) ->
'conn ->
Request.t ->
'body ->
(Response.t * 'body) Lwt.t
(** Trace requests to a Cohttp server.
Use it like this:
@ -18,19 +26,8 @@ module Server : sig
~mode:(`TCP (`Port 8080))
(Server.make () ~callback:callback_traced)
*)
val trace :
?service_name:string ->
?attrs:Otel.Span.key_value list ->
('conn -> Request.t -> 'body -> (Response.t * 'body) Lwt.t) ->
'conn -> Request.t -> 'body -> (Response.t * 'body) Lwt.t
(** Trace a new internal span.
Identical to [Opentelemetry_lwt.Trace.with_], but fetches/stores the trace
scope in the [x-ocaml-otel-traceparent] header in the request for
convenience.
*)
val with_:
val with_ :
?trace_state:string ->
?service_name:string ->
?attrs:Otel.Span.key_value list ->
@ -40,21 +37,28 @@ module Server : sig
Request.t ->
(Request.t -> 'a Lwt.t) ->
'a Lwt.t
(** Trace a new internal span.
Identical to [Opentelemetry_lwt.Trace.with_], but fetches/stores the trace
scope in the [x-ocaml-otel-traceparent] header in the request for
convenience.
*)
val get_trace_context :
?from:[ `Internal | `External ] -> Request.t -> Otel.Trace.scope option
(** Get the tracing scope from the custom [x-ocaml-otel-traceparent] header
added by [trace] and [with_].
*)
val get_trace_context : ?from:[`Internal | `External] -> Request.t -> Otel.Trace.scope option
val set_trace_context : Otel.Trace.scope -> Request.t -> Request.t
(** Set the tracing scope in the custom [x-ocaml-otel-traceparent] header used
by [trace] and [with_].
*)
val set_trace_context : Otel.Trace.scope -> Request.t -> Request.t
val remove_trace_context : Request.t -> Request.t
(** Strip the custom [x-ocaml-otel-traceparent] header added by [trace] and
[with_].
*)
val remove_trace_context : Request.t -> Request.t
end = struct
let attrs_of_request (req : Request.t) =
let meth = req |> Request.meth |> Code.string_of_method in
@ -63,25 +67,24 @@ end = struct
let ua = Header.get (Request.headers req) "user-agent" in
let uri = Request.uri req in
List.concat
[ [ ("http.method", `String meth) ]
; (match host with None -> [] | Some h -> [ ("http.host", `String h) ])
; [ ("http.url", `String (Uri.to_string uri)) ]
; ( match ua with
| None ->
[]
| Some ua ->
[ ("http.user_agent", `String ua) ] )
; ( match referer with
| None ->
[]
| Some r ->
[ ("http.request.header.referer", `String r) ] )
[
[ "http.method", `String meth ];
(match host with
| None -> []
| Some h -> [ "http.host", `String h ]);
[ "http.url", `String (Uri.to_string uri) ];
(match ua with
| None -> []
| Some ua -> [ "http.user_agent", `String ua ]);
(match referer with
| None -> []
| Some r -> [ "http.request.header.referer", `String r ]);
]
let attrs_of_response (res : Response.t) =
let code = Response.status res in
let code = Code.code_of_status code in
[ ("http.status_code", `Int code) ]
[ "http.status_code", `Int code ]
let header_x_ocaml_otel_traceparent = "x-ocaml-otel-traceparent"
@ -89,11 +92,12 @@ end = struct
let module Traceparent = Otel.Trace_context.Traceparent in
let headers =
Header.add (Request.headers req) header_x_ocaml_otel_traceparent
(Traceparent.to_value ~trace_id:scope.trace_id ~parent_id:scope.span_id ())
(Traceparent.to_value ~trace_id:scope.trace_id ~parent_id:scope.span_id
())
in
{ req with headers }
let get_trace_context ?(from=`Internal) req =
let get_trace_context ?(from = `Internal) req =
let module Traceparent = Otel.Trace_context.Traceparent in
let name =
match from with
@ -103,125 +107,125 @@ end = struct
match Header.get (Request.headers req) name with
| None -> None
| Some v ->
(match Traceparent.of_value v with
| Ok (trace_id, parent_id) ->
(Some Otel.Trace.{ trace_id; span_id = parent_id; events = []; attrs = []})
| Error _ -> None)
(match Traceparent.of_value v with
| Ok (trace_id, parent_id) ->
Some
Otel.Trace.{ trace_id; span_id = parent_id; events = []; attrs = [] }
| Error _ -> None)
let remove_trace_context req =
let headers = Header.remove (Request.headers req) header_x_ocaml_otel_traceparent in
let headers =
Header.remove (Request.headers req) header_x_ocaml_otel_traceparent
in
{ req with headers }
let trace ?service_name ?(attrs=[]) callback =
fun conn req body ->
let trace ?service_name ?(attrs = []) callback conn req body =
let scope = get_trace_context ~from:`External req in
Otel_lwt.Trace.with_
?service_name
"request"
~kind:Span_kind_server
Otel_lwt.Trace.with_ ?service_name "request" ~kind:Span_kind_server
?trace_id:(Option.map (fun scope -> scope.Otel.Trace.trace_id) scope)
?parent:(Option.map (fun scope -> scope.Otel.Trace.span_id) scope)
~attrs:(attrs @ attrs_of_request req)
(fun scope ->
let open Lwt.Syntax in
let req = set_trace_context scope req in
let* (res, body) = callback conn req body in
Otel.Trace.add_attrs scope (fun () -> attrs_of_response res) ;
Lwt.return (res, body) )
let* res, body = callback conn req body in
Otel.Trace.add_attrs scope (fun () -> attrs_of_response res);
Lwt.return (res, body))
let with_ ?trace_state ?service_name ?attrs ?(kind=Otel.Span.Span_kind_internal) ?links name req (f : Request.t -> 'a Lwt.t) =
let with_ ?trace_state ?service_name ?attrs
?(kind = Otel.Span.Span_kind_internal) ?links name req
(f : Request.t -> 'a Lwt.t) =
let scope = get_trace_context ~from:`Internal req in
Otel_lwt.Trace.with_
?trace_state
?service_name
?attrs
~kind
Otel_lwt.Trace.with_ ?trace_state ?service_name ?attrs ~kind
?trace_id:(Option.map (fun scope -> scope.Otel.Trace.trace_id) scope)
?parent:(Option.map (fun scope -> scope.Otel.Trace.span_id) scope)
?links
name
?links name
(fun scope ->
let open Lwt.Syntax in
let req = set_trace_context scope req in
f req)
end
let client ?(scope : Otel.Trace.scope option) (module C : Cohttp_lwt.S.Client) =
let client ?(scope : Otel.Trace.scope option) (module C : Cohttp_lwt.S.Client) =
let module Traced = struct
open Lwt.Syntax
open Lwt.Syntax
let attrs_for ~uri ~meth () =
[ ("http.method", `String (Code.string_of_method `GET))
; ("http.url", `String (Uri.to_string uri))
]
let attrs_for ~uri ~meth () =
[
"http.method", `String (Code.string_of_method `GET);
"http.url", `String (Uri.to_string uri);
]
let context_for ~uri ~meth =
let trace_id = match scope with | Some scope -> Some scope.trace_id | None -> None in
let parent = match scope with | Some scope -> Some scope.span_id | None -> None in
let attrs = attrs_for ~uri ~meth () in
(trace_id, parent, attrs)
let context_for ~uri ~meth =
let trace_id =
match scope with
| Some scope -> Some scope.trace_id
| None -> None
in
let parent =
match scope with
| Some scope -> Some scope.span_id
| None -> None
in
let attrs = attrs_for ~uri ~meth () in
trace_id, parent, attrs
let add_traceparent (scope : Otel.Trace.scope) headers =
let module Traceparent = Otel.Trace_context.Traceparent in
let headers = match headers with | None -> Header.init () | Some headers -> headers in
Header.add headers Traceparent.name
(Traceparent.to_value ~trace_id:scope.trace_id ~parent_id:scope.span_id ())
let add_traceparent (scope : Otel.Trace.scope) headers =
let module Traceparent = Otel.Trace_context.Traceparent in
let headers =
match headers with
| None -> Header.init ()
| Some headers -> headers
in
Header.add headers Traceparent.name
(Traceparent.to_value ~trace_id:scope.trace_id ~parent_id:scope.span_id
())
type ctx = C.ctx
type ctx = C.ctx
let call ?ctx ?headers ?body ?chunked meth (uri : Uri.t) : (Response.t * Cohttp_lwt.Body.t) Lwt.t =
let (trace_id, parent, attrs) = context_for ~uri ~meth in
Otel_lwt.Trace.with_ "request"
~kind:Span_kind_client
?trace_id
?parent
~attrs
(fun scope ->
let headers = add_traceparent scope headers in
let* (res, body) = C.call ?ctx ~headers ?body ?chunked meth uri in
Otel.Trace.add_attrs scope (fun () ->
let code = Response.status res in
let code = Code.code_of_status code in
[ ("http.status_code", `Int code) ]) ;
Lwt.return (res, body))
let call ?ctx ?headers ?body ?chunked meth (uri : Uri.t) :
(Response.t * Cohttp_lwt.Body.t) Lwt.t =
let trace_id, parent, attrs = context_for ~uri ~meth in
Otel_lwt.Trace.with_ "request" ~kind:Span_kind_client ?trace_id ?parent
~attrs (fun scope ->
let headers = add_traceparent scope headers in
let* res, body = C.call ?ctx ~headers ?body ?chunked meth uri in
Otel.Trace.add_attrs scope (fun () ->
let code = Response.status res in
let code = Code.code_of_status code in
[ "http.status_code", `Int code ]);
Lwt.return (res, body))
let head ?ctx ?headers uri =
let open Lwt.Infix in
call ?ctx ?headers `HEAD uri >|= fst
let head ?ctx ?headers uri =
let open Lwt.Infix in
call ?ctx ?headers `HEAD uri >|= fst
let get ?ctx ?headers uri = call ?ctx ?headers `GET uri
let get ?ctx ?headers uri = call ?ctx ?headers `GET uri
let delete ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `DELETE uri
let delete ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `DELETE uri
let post ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `POST uri
let post ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `POST uri
let put ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `PUT uri
let put ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `PUT uri
let patch ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `PATCH uri
let patch ?ctx ?body ?chunked ?headers uri =
call ?ctx ?headers ?body ?chunked `PATCH uri
let post_form ?ctx ?headers ~params uri =
let (trace_id, parent, attrs) = context_for ~uri ~meth:`POST in
Otel_lwt.Trace.with_ "request"
~kind:Span_kind_client
?trace_id
?parent
~attrs
(fun scope ->
let headers = add_traceparent scope headers in
let* (res, body) =
C.post_form ?ctx ~headers ~params uri
in
Otel.Trace.add_attrs scope (fun () ->
let code = Response.status res in
let code = Code.code_of_status code in
[ ("http.status_code", `Int code) ]) ;
Lwt.return (res, body))
let post_form ?ctx ?headers ~params uri =
let trace_id, parent, attrs = context_for ~uri ~meth:`POST in
Otel_lwt.Trace.with_ "request" ~kind:Span_kind_client ?trace_id ?parent
~attrs (fun scope ->
let headers = add_traceparent scope headers in
let* res, body = C.post_form ?ctx ~headers ~params uri in
Otel.Trace.add_attrs scope (fun () ->
let code = Response.status res in
let code = Code.code_of_status code in
[ "http.status_code", `Int code ]);
Lwt.return (res, body))
let callv = C.callv (* TODO *)
end
in
let callv = C.callv (* TODO *)
end in
(module Traced : Cohttp_lwt.S.Client)

View file

@ -1,5 +1,5 @@
(library
(name opentelemetry_lwt)
(public_name opentelemetry-lwt)
(synopsis "Lwt frontend for opentelemetry")
(libraries lwt opentelemetry))
(name opentelemetry_lwt)
(public_name opentelemetry-lwt)
(synopsis "Lwt frontend for opentelemetry")
(libraries lwt opentelemetry))

View file

@ -1,6 +1,5 @@
open Opentelemetry
open Lwt.Syntax
module Span_id = Span_id
module Trace_id = Trace_id
module Event = Event
@ -15,10 +14,8 @@ module Trace = struct
include Trace
(** Sync span guard *)
let with_
?trace_state ?service_name ?(attrs=[])
?kind ?trace_id ?parent ?scope ?links
name (f:Trace.scope -> 'a Lwt.t) : 'a Lwt.t =
let with_ ?trace_state ?service_name ?(attrs = []) ?kind ?trace_id ?parent
?scope ?links name (f : Trace.scope -> 'a Lwt.t) : 'a Lwt.t =
let trace_id =
match trace_id, scope with
| Some trace_id, _ -> trace_id
@ -31,30 +28,31 @@ module Trace = struct
| None, Some scope -> Some scope.span_id
| None, None -> None
in
let start_time = Timestamp_ns.now_unix_ns() in
let span_id = Span_id.create() in
let scope = {trace_id;span_id;events=[];attrs} in
let start_time = Timestamp_ns.now_unix_ns () in
let span_id = Span_id.create () in
let scope = { trace_id; span_id; events = []; attrs } in
let finally ok =
let status = match ok with
let status =
match ok with
| Ok () -> default_status ~code:Status_code_ok ()
| Error e -> default_status ~code:Status_code_error ~message:e () in
| Error e -> default_status ~code:Status_code_error ~message:e ()
in
let span, _ =
Span.create
?kind ~trace_id ?parent ?links ~id:span_id
?trace_state ~attrs:scope.attrs ~events:scope.events
~start_time ~end_time:(Timestamp_ns.now_unix_ns())
~status
name in
emit ?service_name [span]
Span.create ?kind ~trace_id ?parent ?links ~id:span_id ?trace_state
~attrs:scope.attrs ~events:scope.events ~start_time
~end_time:(Timestamp_ns.now_unix_ns ())
~status name
in
emit ?service_name [ span ]
in
Lwt.catch
(fun () ->
let* x = f scope in
let () = finally (Ok ()) in
Lwt.return x)
let* x = f scope in
let () = finally (Ok ()) in
Lwt.return x)
(fun e ->
let () = finally (Error (Printexc.to_string e)) in
Lwt.fail e)
let () = finally (Error (Printexc.to_string e)) in
Lwt.fail e)
end
module Metrics = struct

View file

@ -1,25 +1,29 @@
module T = Opentelemetry
module Otel_lwt = Opentelemetry_lwt
let spf = Printf.sprintf
let (let@) f x = f x
let ( let@ ) f x = f x
let sleep_inner = ref 0.1
let sleep_outer = ref 2.0
let mk_client ~scope = Opentelemetry_cohttp_lwt.client ~scope (module Cohttp_lwt_unix.Client)
let mk_client ~scope =
Opentelemetry_cohttp_lwt.client ~scope (module Cohttp_lwt_unix.Client)
let run () =
Printf.printf "collector is on %S\n%!" (Opentelemetry_client_ocurl.get_url());
Printf.printf "collector is on %S\n%!" (Opentelemetry_client_ocurl.get_url ());
let open Lwt.Syntax in
let rec go () =
let@ scope =
Otel_lwt.Trace.with_
~kind:T.Span.Span_kind_producer
"loop.outer"
Otel_lwt.Trace.with_ ~kind:T.Span.Span_kind_producer "loop.outer"
in
let* () = Lwt_unix.sleep !sleep_outer in
let module C = (val mk_client ~scope) in
let* (res, body) = C.get (Uri.of_string "https://enec1hql02hz.x.pipedream.net") in
let* res, body =
C.get (Uri.of_string "https://enec1hql02hz.x.pipedream.net")
in
let* () = Cohttp_lwt.Body.drain_body body in
go ()
in
@ -34,26 +38,40 @@ let () =
let thread = ref true in
let batch_traces = ref 400 in
let batch_metrics = ref 3 in
let opts = [
"--debug", Arg.Bool ((:=) debug), " enable debug output";
"--thread", Arg.Bool ((:=) thread), " use a background thread";
"--batch-traces", Arg.Int ((:=) batch_traces), " size of traces batch";
"--batch-metrics", Arg.Int ((:=) batch_metrics), " size of metrics batch";
let opts =
[
"--debug", Arg.Bool (( := ) debug), " enable debug output";
"--thread", Arg.Bool (( := ) thread), " use a background thread";
"--batch-traces", Arg.Int (( := ) batch_traces), " size of traces batch";
( "--batch-metrics",
Arg.Int (( := ) batch_metrics),
" size of metrics batch" );
"--sleep-inner", Arg.Set_float sleep_inner, " sleep (in s) in inner loop";
"--sleep-outer", Arg.Set_float sleep_outer, " sleep (in s) in outer loop";
] |> Arg.align in
]
|> Arg.align
in
Arg.parse opts (fun _ -> ()) "emit1 [opt]*";
let some_if_nzero r = if !r > 0 then Some !r else None in
let config = Opentelemetry_client_ocurl.Config.make
~debug:!debug
~batch_traces:(some_if_nzero batch_traces)
~batch_metrics:(some_if_nzero batch_metrics)
~thread:!thread () in
let some_if_nzero r =
if !r > 0 then
Some !r
else
None
in
let config =
Opentelemetry_client_ocurl.Config.make ~debug:!debug
~batch_traces:(some_if_nzero batch_traces)
~batch_metrics:(some_if_nzero batch_metrics)
~thread:!thread ()
in
Format.printf "@[<2>sleep outer: %.3fs,@ sleep inner: %.3fs,@ config: %a@]@."
!sleep_outer !sleep_inner Opentelemetry_client_ocurl.Config.pp config;
Format.printf "Check HTTP requests at https://requestbin.com/r/enec1hql02hz/26qShWryt5vJc1JfrOwalhr5vQt@.";
Format.printf
"Check HTTP requests at \
https://requestbin.com/r/enec1hql02hz/26qShWryt5vJc1JfrOwalhr5vQt@.";
Opentelemetry_client_ocurl.with_setup ~config () (fun () -> Lwt_main.run (run ()))
Opentelemetry_client_ocurl.with_setup ~config () (fun () ->
Lwt_main.run (run ()))

View file

@ -6,4 +6,5 @@
(executable
(name cohttp_client)
(modules cohttp_client)
(libraries cohttp-lwt-unix opentelemetry opentelemetry-client-ocurl opentelemetry-cohttp-lwt))
(libraries cohttp-lwt-unix opentelemetry opentelemetry-client-ocurl
opentelemetry-cohttp-lwt))

View file

@ -1,41 +1,43 @@
module T = Opentelemetry
let spf = Printf.sprintf
let (let@) f x = f x
let ( let@ ) f x = f x
let sleep_inner = ref 0.1
let sleep_outer = ref 2.0
let num_sleep = ref 0
let run () =
Printf.printf "collector is on %S\n%!" (Opentelemetry_client_ocurl.get_url());
T.GC_metrics.basic_setup();
Printf.printf "collector is on %S\n%!" (Opentelemetry_client_ocurl.get_url ());
T.GC_metrics.basic_setup ();
T.Metrics_callbacks.register (fun () ->
T.Metrics.[ sum ~name:"num-sleep" ~is_monotonic:true [int !num_sleep] ]);
T.Metrics.[ sum ~name:"num-sleep" ~is_monotonic:true [ int !num_sleep ] ]);
let i = ref 0 in
while true do
let@ scope =
T.Trace.with_
~kind:T.Span.Span_kind_producer
"loop.outer" ~attrs:["i", `Int !i] in
T.Trace.with_ ~kind:T.Span.Span_kind_producer "loop.outer"
~attrs:[ "i", `Int !i ]
in
for j=0 to 4 do
let@ scope = T.Trace.with_ ~kind:T.Span.Span_kind_internal ~scope
~attrs:["j", `Int j]
"loop.inner" in
for j = 0 to 4 do
let@ scope =
T.Trace.with_ ~kind:T.Span.Span_kind_internal ~scope
~attrs:[ "j", `Int j ]
"loop.inner"
in
Unix.sleepf !sleep_outer;
incr num_sleep;
incr i;
(try
let@ _ =
T.Trace.with_ ~kind:T.Span.Span_kind_internal ~scope
"alloc" in
try
let@ _ = T.Trace.with_ ~kind:T.Span.Span_kind_internal ~scope "alloc" in
(* allocate some stuff *)
let _arr = Sys.opaque_identity @@ Array.make (25 * 25551) 42.0 in
ignore _arr;
@ -43,12 +45,12 @@ let run () =
Unix.sleepf !sleep_inner;
incr num_sleep;
if j=4 && !i mod 13 = 0 then failwith "oh no"; (* simulate a failure *)
if j = 4 && !i mod 13 = 0 then failwith "oh no";
T.Trace.add_event scope (fun()->T.Event.make "done with alloc");
with Failure _ ->
());
done;
(* simulate a failure *)
T.Trace.add_event scope (fun () -> T.Event.make "done with alloc")
with Failure _ -> ()
done
done
let () =
@ -60,23 +62,34 @@ let () =
let thread = ref true in
let batch_traces = ref 400 in
let batch_metrics = ref 3 in
let opts = [
"--debug", Arg.Bool ((:=) debug), " enable debug output";
"--thread", Arg.Bool ((:=) thread), " use a background thread";
"--batch-traces", Arg.Int ((:=) batch_traces), " size of traces batch";
"--batch-metrics", Arg.Int ((:=) batch_metrics), " size of metrics batch";
"--sleep-inner", Arg.Set_float sleep_inner, " sleep (in s) in inner loop";
"--sleep-outer", Arg.Set_float sleep_outer, " sleep (in s) in outer loop";
] |> Arg.align in
let opts =
[
"--debug", Arg.Bool (( := ) debug), " enable debug output";
"--thread", Arg.Bool (( := ) thread), " use a background thread";
"--batch-traces", Arg.Int (( := ) batch_traces), " size of traces batch";
( "--batch-metrics",
Arg.Int (( := ) batch_metrics),
" size of metrics batch" );
"--sleep-inner", Arg.Set_float sleep_inner, " sleep (in s) in inner loop";
"--sleep-outer", Arg.Set_float sleep_outer, " sleep (in s) in outer loop";
]
|> Arg.align
in
Arg.parse opts (fun _ -> ()) "emit1 [opt]*";
let some_if_nzero r = if !r > 0 then Some !r else None in
let config = Opentelemetry_client_ocurl.Config.make
~debug:!debug
let some_if_nzero r =
if !r > 0 then
Some !r
else
None
in
let config =
Opentelemetry_client_ocurl.Config.make ~debug:!debug
~batch_traces:(some_if_nzero batch_traces)
~batch_metrics:(some_if_nzero batch_metrics)
~thread:!thread () in
~thread:!thread ()
in
Format.printf "@[<2>sleep outer: %.3fs,@ sleep inner: %.3fs,@ config: %a@]@."
!sleep_outer !sleep_inner Opentelemetry_client_ocurl.Config.pp config;

View file

@ -2,39 +2,49 @@ open Opentelemetry
let pp_traceparent fmt (trace_id, parent_id) =
let open Format in
fprintf fmt "trace_id:%S parent_id:%S"
(Trace_id.to_hex trace_id)
fprintf fmt "trace_id:%S parent_id:%S" (Trace_id.to_hex trace_id)
(Span_id.to_hex parent_id)
let test_of_value str =
let open Format in
printf "@[<v 2>Trace_context.Traceparent.of_value %S:@ %a@]@."
str
(pp_print_result ~ok:(fun fmt (trace_id, parent_id) ->
printf "@[<v 2>Trace_context.Traceparent.of_value %S:@ %a@]@." str
(pp_print_result
~ok:(fun fmt (trace_id, parent_id) ->
fprintf fmt "Ok %a" pp_traceparent (trace_id, parent_id))
~error:(fun fmt msg -> fprintf fmt "Error %S" msg))
(Trace_context.Traceparent.of_value str)
let () = test_of_value "xx"
let () = test_of_value "00"
let () = test_of_value "00-xxxx"
let () = test_of_value "00-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
let () = test_of_value "00-0123456789abcdef0123456789abcdef"
let () = test_of_value "00-0123456789abcdef0123456789abcdef-xxxx"
let () = test_of_value "00-0123456789abcdef0123456789abcdef-xxxxxxxxxxxxxxxx"
let () = test_of_value "00-0123456789abcdef0123456789abcdef-0123456789abcdef"
let () = test_of_value "00-0123456789abcdef0123456789abcdef-0123456789abcdef-"
let () = test_of_value "00-0123456789abcdef0123456789abcdef-0123456789abcdef-00"
let () = test_of_value "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
let () = print_endline ""
let test_to_value trace_id parent_id =
let open Format in
printf "@[<v 2>Trace_context.Traceparent.to_value %a:@ %S@]@."
pp_traceparent (trace_id, parent_id)
printf "@[<v 2>Trace_context.Traceparent.to_value %a:@ %S@]@." pp_traceparent
(trace_id, parent_id)
(Trace_context.Traceparent.to_value ~trace_id ~parent_id ())
let () = test_to_value (Trace_id.of_hex "4bf92f3577b34da6a3ce929d0e0e4736") (Span_id.of_hex "00f067aa0ba902b7")
let () =
test_to_value
(Trace_id.of_hex "4bf92f3577b34da6a3ce929d0e0e4736")
(Span_id.of_hex "00f067aa0ba902b7")