mirror of
https://github.com/ocaml-tracing/ocaml-opentelemetry.git
synced 2026-03-07 18:37:56 -05:00
port changes to cohttp client
This commit is contained in:
parent
1b7b8edbe0
commit
3daa0d8762
8 changed files with 127 additions and 65 deletions
|
|
@ -14,12 +14,30 @@ let debug_ =
|
||||||
|
|
||||||
let default_url = "http://localhost:4318"
|
let default_url = "http://localhost:4318"
|
||||||
|
|
||||||
let url =
|
let make_get_from_env env_name =
|
||||||
ref (try Sys.getenv "OTEL_EXPORTER_OTLP_ENDPOINT" with _ -> default_url)
|
let value = ref None in
|
||||||
|
fun () ->
|
||||||
|
match !value with
|
||||||
|
| None ->
|
||||||
|
value := Sys.getenv_opt env_name;
|
||||||
|
!value
|
||||||
|
| Some value -> Some value
|
||||||
|
|
||||||
let get_url () = !url
|
let get_url_from_env = make_get_from_env "OTEL_EXPORTER_OTLP_ENDPOINT"
|
||||||
|
|
||||||
let set_url s = url := s
|
let get_url_traces_from_env =
|
||||||
|
make_get_from_env "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
|
||||||
|
|
||||||
|
let get_url_metrics_from_env =
|
||||||
|
make_get_from_env "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"
|
||||||
|
|
||||||
|
let get_url_logs_from_env = make_get_from_env "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT"
|
||||||
|
|
||||||
|
let remove_trailing_slash url =
|
||||||
|
if url <> "" && String.get url (String.length url - 1) = '/' then
|
||||||
|
String.sub url 0 (String.length url - 1)
|
||||||
|
else
|
||||||
|
url
|
||||||
|
|
||||||
let parse_headers s =
|
let parse_headers s =
|
||||||
let parse_header s =
|
let parse_header s =
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,9 @@ open Common_
|
||||||
|
|
||||||
type t = {
|
type t = {
|
||||||
debug: bool;
|
debug: bool;
|
||||||
url: string;
|
url_traces: string;
|
||||||
|
url_metrics: string;
|
||||||
|
url_logs: string;
|
||||||
headers: (string * string) list;
|
headers: (string * string) list;
|
||||||
batch_traces: int option;
|
batch_traces: int option;
|
||||||
batch_metrics: int option;
|
batch_metrics: int option;
|
||||||
|
|
@ -16,7 +18,9 @@ let pp out self : unit =
|
||||||
let ppheaders = Format.pp_print_list pp_header in
|
let ppheaders = Format.pp_print_list pp_header in
|
||||||
let {
|
let {
|
||||||
debug;
|
debug;
|
||||||
url;
|
url_traces;
|
||||||
|
url_metrics;
|
||||||
|
url_logs;
|
||||||
headers;
|
headers;
|
||||||
batch_traces;
|
batch_traces;
|
||||||
batch_metrics;
|
batch_metrics;
|
||||||
|
|
@ -26,17 +30,52 @@ let pp out self : unit =
|
||||||
self
|
self
|
||||||
in
|
in
|
||||||
Format.fprintf out
|
Format.fprintf out
|
||||||
"{@[ debug=%B;@ url=%S;@ headers=%a;@ batch_traces=%a;@ batch_metrics=%a;@ \
|
"{@[ debug=%B;@ url_traces=%S;@ url_metrics=%S;@ url_logs=%S;@ \
|
||||||
batch_logs=%a;@ batch_timeout_ms=%d; @]}"
|
headers=%a;@ batch_traces=%a;@ batch_metrics=%a;@ batch_logs=%a;@ \
|
||||||
debug url ppheaders headers ppiopt batch_traces ppiopt batch_metrics ppiopt
|
batch_timeout_ms=%d; @]}"
|
||||||
batch_logs batch_timeout_ms
|
debug url_traces url_metrics url_logs ppheaders headers ppiopt batch_traces
|
||||||
|
ppiopt batch_metrics ppiopt batch_logs batch_timeout_ms
|
||||||
|
|
||||||
|
let make ?(debug = !debug_) ?url ?url_traces ?url_metrics ?url_logs
|
||||||
|
?(headers = get_headers ()) ?(batch_traces = Some 400)
|
||||||
|
?(batch_metrics = Some 20) ?(batch_logs = Some 400)
|
||||||
|
?(batch_timeout_ms = 500) () : t =
|
||||||
|
let url_traces, url_metrics, url_logs =
|
||||||
|
let base_url =
|
||||||
|
match url with
|
||||||
|
| None -> Option.value (get_url_from_env ()) ~default:default_url
|
||||||
|
| Some url -> remove_trailing_slash url
|
||||||
|
in
|
||||||
|
let url_traces =
|
||||||
|
match url_traces with
|
||||||
|
| None ->
|
||||||
|
Option.value
|
||||||
|
(get_url_traces_from_env ())
|
||||||
|
~default:(base_url ^ "/v1/traces")
|
||||||
|
| Some url -> url
|
||||||
|
in
|
||||||
|
let url_metrics =
|
||||||
|
match url_metrics with
|
||||||
|
| None ->
|
||||||
|
Option.value
|
||||||
|
(get_url_metrics_from_env ())
|
||||||
|
~default:(base_url ^ "/v1/metrics")
|
||||||
|
| Some url -> url
|
||||||
|
in
|
||||||
|
let url_logs =
|
||||||
|
match url_logs with
|
||||||
|
| None ->
|
||||||
|
Option.value (get_url_logs_from_env ()) ~default:(base_url ^ "/v1/logs")
|
||||||
|
| Some url -> url
|
||||||
|
in
|
||||||
|
url_traces, url_metrics, url_logs
|
||||||
|
in
|
||||||
|
|
||||||
let make ?(debug = !debug_) ?(url = get_url ()) ?(headers = get_headers ())
|
|
||||||
?(batch_traces = Some 400) ?(batch_metrics = Some 20)
|
|
||||||
?(batch_logs = Some 400) ?(batch_timeout_ms = 500) () : t =
|
|
||||||
{
|
{
|
||||||
debug;
|
debug;
|
||||||
url;
|
url_traces;
|
||||||
|
url_metrics;
|
||||||
|
url_logs;
|
||||||
headers;
|
headers;
|
||||||
batch_traces;
|
batch_traces;
|
||||||
batch_metrics;
|
batch_metrics;
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,8 @@
|
||||||
type t = private {
|
type t = private {
|
||||||
debug: bool;
|
debug: bool;
|
||||||
url: string;
|
url_traces: string; (** Url to send traces *)
|
||||||
(** Url of the endpoint. Default is "http://localhost:4318",
|
url_metrics: string; (** Url to send metrics*)
|
||||||
or "OTEL_EXPORTER_OTLP_ENDPOINT" if set. *)
|
url_logs: string; (** Url to send logs *)
|
||||||
headers: (string * string) list;
|
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. *)
|
"OTEL_EXPORTER_OTLP_HEADERS" if set. *)
|
||||||
|
|
@ -37,6 +37,9 @@ type t = private {
|
||||||
val make :
|
val make :
|
||||||
?debug:bool ->
|
?debug:bool ->
|
||||||
?url:string ->
|
?url:string ->
|
||||||
|
?url_traces:string ->
|
||||||
|
?url_metrics:string ->
|
||||||
|
?url_logs:string ->
|
||||||
?headers:(string * string) list ->
|
?headers:(string * string) list ->
|
||||||
?batch_traces:int option ->
|
?batch_traces:int option ->
|
||||||
?batch_metrics:int option ->
|
?batch_metrics:int option ->
|
||||||
|
|
@ -49,6 +52,25 @@ val make :
|
||||||
@param thread if true and [bg_threads] is not provided, we will pick a number
|
@param thread if true and [bg_threads] is not provided, we will pick a number
|
||||||
of bg threads. Otherwise the number of [bg_threads] superseeds this option.
|
of bg threads. Otherwise the number of [bg_threads] superseeds this option.
|
||||||
|
|
||||||
|
@param url base url used to construct per-signal urls. Per-signal url options take precedence over this base url.
|
||||||
|
Default is "http://localhost:4318", or "OTEL_EXPORTER_OTLP_ENDPOINT" if set.
|
||||||
|
|
||||||
|
Example of constructed per-signal urls with the base url http://localhost:4318
|
||||||
|
- Traces: http://localhost:4318/v1/traces
|
||||||
|
- Metrics: http://localhost:4318/v1/metrics
|
||||||
|
- Logs: http://localhost:4318/v1/logs
|
||||||
|
|
||||||
|
Use per-signal url options if different urls are needed for each signal type.
|
||||||
|
|
||||||
|
@param url_traces url to send traces, or "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" if set.
|
||||||
|
The url is used as-is without any modification.
|
||||||
|
|
||||||
|
@param url_metrics url to send metrics, or "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" if set.
|
||||||
|
The url is used as-is without any modification.
|
||||||
|
|
||||||
|
@param url_logs url to send logs, or "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT" if set.
|
||||||
|
The url is used as-is without any modification.
|
||||||
|
|
||||||
*)
|
*)
|
||||||
|
|
||||||
val pp : Format.formatter -> t -> unit
|
val pp : Format.formatter -> t -> unit
|
||||||
|
|
|
||||||
|
|
@ -73,7 +73,7 @@ module Httpc : sig
|
||||||
val send :
|
val send :
|
||||||
t ->
|
t ->
|
||||||
config:Config.t ->
|
config:Config.t ->
|
||||||
path:string ->
|
url:string ->
|
||||||
decode:[ `Dec of Pbrt.Decoder.t -> 'a | `Ret of 'a ] ->
|
decode:[ `Dec of Pbrt.Decoder.t -> 'a | `Ret of 'a ] ->
|
||||||
string ->
|
string ->
|
||||||
('a, error) result Lwt.t
|
('a, error) result Lwt.t
|
||||||
|
|
@ -91,17 +91,9 @@ end = struct
|
||||||
let cleanup _self = ()
|
let cleanup _self = ()
|
||||||
|
|
||||||
(* send the content to the remote endpoint/path *)
|
(* send the content to the remote endpoint/path *)
|
||||||
let send (_self : t) ~(config : Config.t) ~path ~decode (bod : string) :
|
let send (_self : t) ~(config : Config.t) ~url ~decode (bod : string) :
|
||||||
('a, error) result Lwt.t =
|
('a, error) result Lwt.t =
|
||||||
let url =
|
let uri = Uri.of_string url in
|
||||||
let url = config.url in
|
|
||||||
if url <> "" && String.get url (String.length url - 1) = '/' then
|
|
||||||
String.sub url 0 (String.length url - 1)
|
|
||||||
else
|
|
||||||
url
|
|
||||||
in
|
|
||||||
let full_url = url ^ path in
|
|
||||||
let uri = Uri.of_string full_url in
|
|
||||||
|
|
||||||
let open Cohttp in
|
let open Cohttp in
|
||||||
let headers = Header.(add_list (init ()) !headers) in
|
let headers = Header.(add_list (init ()) !headers) in
|
||||||
|
|
@ -121,7 +113,7 @@ end = struct
|
||||||
| Error e ->
|
| Error e ->
|
||||||
let err =
|
let err =
|
||||||
`Failure
|
`Failure
|
||||||
(spf "sending signals via http POST to %S\nfailed with:\n%s" full_url
|
(spf "sending signals via http POST to %S\nfailed with:\n%s" url
|
||||||
(Printexc.to_string e))
|
(Printexc.to_string e))
|
||||||
in
|
in
|
||||||
Lwt.return @@ Error err
|
Lwt.return @@ Error err
|
||||||
|
|
@ -158,7 +150,7 @@ end = struct
|
||||||
%s\n\
|
%s\n\
|
||||||
status: %S\n\
|
status: %S\n\
|
||||||
%s"
|
%s"
|
||||||
full_url code (Printexc.to_string e) body bt))
|
url code (Printexc.to_string e) body bt))
|
||||||
in
|
in
|
||||||
Lwt.return r
|
Lwt.return r
|
||||||
)
|
)
|
||||||
|
|
@ -292,11 +284,11 @@ let mk_emitter ~stop ~(config : Config.t) () : (module EMITTER) =
|
||||||
|
|
||||||
let set_on_tick_callbacks = Atomic.set on_tick_cbs_
|
let set_on_tick_callbacks = Atomic.set on_tick_cbs_
|
||||||
|
|
||||||
let send_http_ (httpc : Httpc.t) encoder ~path ~encode x : unit Lwt.t =
|
let send_http_ (httpc : Httpc.t) encoder ~url ~encode x : unit Lwt.t =
|
||||||
Pbrt.Encoder.reset encoder;
|
Pbrt.Encoder.reset encoder;
|
||||||
encode x encoder;
|
encode x encoder;
|
||||||
let data = Pbrt.Encoder.to_string encoder in
|
let data = Pbrt.Encoder.to_string encoder in
|
||||||
let* r = Httpc.send httpc ~config ~path ~decode:(`Ret ()) data in
|
let* r = Httpc.send httpc ~config ~url ~decode:(`Ret ()) data in
|
||||||
match r with
|
match r with
|
||||||
| Ok () -> Lwt.return ()
|
| Ok () -> Lwt.return ()
|
||||||
| Error `Sysbreak ->
|
| Error `Sysbreak ->
|
||||||
|
|
@ -317,7 +309,8 @@ let mk_emitter ~stop ~(config : Config.t) () : (module EMITTER) =
|
||||||
Metrics_service.default_export_metrics_service_request
|
Metrics_service.default_export_metrics_service_request
|
||||||
~resource_metrics:l ()
|
~resource_metrics:l ()
|
||||||
in
|
in
|
||||||
send_http_ curl encoder ~path:"/v1/metrics"
|
let url = config.Config.url_metrics in
|
||||||
|
send_http_ curl encoder ~url
|
||||||
~encode:Metrics_service.encode_pb_export_metrics_service_request x
|
~encode:Metrics_service.encode_pb_export_metrics_service_request x
|
||||||
|
|
||||||
let send_traces_http curl encoder (l : Trace.resource_spans list list) =
|
let send_traces_http curl encoder (l : Trace.resource_spans list list) =
|
||||||
|
|
@ -325,7 +318,8 @@ let mk_emitter ~stop ~(config : Config.t) () : (module EMITTER) =
|
||||||
let x =
|
let x =
|
||||||
Trace_service.default_export_trace_service_request ~resource_spans:l ()
|
Trace_service.default_export_trace_service_request ~resource_spans:l ()
|
||||||
in
|
in
|
||||||
send_http_ curl encoder ~path:"/v1/traces"
|
let url = config.Config.url_traces in
|
||||||
|
send_http_ curl encoder ~url
|
||||||
~encode:Trace_service.encode_pb_export_trace_service_request x
|
~encode:Trace_service.encode_pb_export_trace_service_request x
|
||||||
|
|
||||||
let send_logs_http curl encoder (l : Logs.resource_logs list list) =
|
let send_logs_http curl encoder (l : Logs.resource_logs list list) =
|
||||||
|
|
@ -333,7 +327,8 @@ let mk_emitter ~stop ~(config : Config.t) () : (module EMITTER) =
|
||||||
let x =
|
let x =
|
||||||
Logs_service.default_export_logs_service_request ~resource_logs:l ()
|
Logs_service.default_export_logs_service_request ~resource_logs:l ()
|
||||||
in
|
in
|
||||||
send_http_ curl encoder ~path:"/v1/logs"
|
let url = config.Config.url_logs in
|
||||||
|
send_http_ curl encoder ~url
|
||||||
~encode:Logs_service.encode_pb_export_logs_service_request x
|
~encode:Logs_service.encode_pb_export_logs_service_request x
|
||||||
|
|
||||||
(* emit metrics, if the batch is full or timeout lapsed *)
|
(* emit metrics, if the batch is full or timeout lapsed *)
|
||||||
|
|
@ -459,12 +454,13 @@ let mk_emitter ~stop ~(config : Config.t) () : (module EMITTER) =
|
||||||
end in
|
end in
|
||||||
(module M)
|
(module M)
|
||||||
|
|
||||||
module Backend (Arg : sig
|
module Backend
|
||||||
val stop : bool Atomic.t
|
(Arg : sig
|
||||||
|
val stop : bool Atomic.t
|
||||||
|
|
||||||
val config : Config.t
|
val config : Config.t
|
||||||
end)
|
end)
|
||||||
() : Opentelemetry.Collector.BACKEND = struct
|
() : Opentelemetry.Collector.BACKEND = struct
|
||||||
include (val mk_emitter ~stop:Arg.stop ~config:Arg.config ())
|
include (val mk_emitter ~stop:Arg.stop ~config:Arg.config ())
|
||||||
|
|
||||||
open Opentelemetry.Proto
|
open Opentelemetry.Proto
|
||||||
|
|
@ -475,10 +471,10 @@ end)
|
||||||
send =
|
send =
|
||||||
(fun l ~ret ->
|
(fun l ~ret ->
|
||||||
(if !debug_ then
|
(if !debug_ then
|
||||||
let@ () = Lock.with_lock in
|
let@ () = Lock.with_lock in
|
||||||
Format.eprintf "send spans %a@."
|
Format.eprintf "send spans %a@."
|
||||||
(Format.pp_print_list Trace.pp_resource_spans)
|
(Format.pp_print_list Trace.pp_resource_spans)
|
||||||
l);
|
l);
|
||||||
push_trace l;
|
push_trace l;
|
||||||
ret ());
|
ret ());
|
||||||
}
|
}
|
||||||
|
|
@ -532,10 +528,10 @@ end)
|
||||||
send =
|
send =
|
||||||
(fun m ~ret ->
|
(fun m ~ret ->
|
||||||
(if !debug_ then
|
(if !debug_ then
|
||||||
let@ () = Lock.with_lock in
|
let@ () = Lock.with_lock in
|
||||||
Format.eprintf "send metrics %a@."
|
Format.eprintf "send metrics %a@."
|
||||||
(Format.pp_print_list Metrics.pp_resource_metrics)
|
(Format.pp_print_list Metrics.pp_resource_metrics)
|
||||||
m);
|
m);
|
||||||
|
|
||||||
let m = List.rev_append (additional_metrics ()) m in
|
let m = List.rev_append (additional_metrics ()) m in
|
||||||
push_metrics m;
|
push_metrics m;
|
||||||
|
|
@ -547,10 +543,10 @@ end)
|
||||||
send =
|
send =
|
||||||
(fun m ~ret ->
|
(fun m ~ret ->
|
||||||
(if !debug_ then
|
(if !debug_ then
|
||||||
let@ () = Lock.with_lock in
|
let@ () = Lock.with_lock in
|
||||||
Format.eprintf "send logs %a@."
|
Format.eprintf "send logs %a@."
|
||||||
(Format.pp_print_list Logs.pp_resource_logs)
|
(Format.pp_print_list Logs.pp_resource_logs)
|
||||||
m);
|
m);
|
||||||
|
|
||||||
push_logs m;
|
push_logs m;
|
||||||
ret ());
|
ret ());
|
||||||
|
|
@ -560,8 +556,6 @@ end
|
||||||
let create_backend ?(stop = Atomic.make false) ?(config = Config.make ()) () =
|
let create_backend ?(stop = Atomic.make false) ?(config = Config.make ()) () =
|
||||||
debug_ := config.debug;
|
debug_ := config.debug;
|
||||||
|
|
||||||
if config.url <> get_url () then set_url config.url;
|
|
||||||
|
|
||||||
let module B =
|
let module B =
|
||||||
Backend
|
Backend
|
||||||
(struct
|
(struct
|
||||||
|
|
|
||||||
|
|
@ -5,12 +5,6 @@
|
||||||
|
|
||||||
open Common_
|
open Common_
|
||||||
|
|
||||||
val get_url : unit -> string
|
|
||||||
|
|
||||||
val set_url : string -> unit
|
|
||||||
(** Url of the endpoint. Default is "http://localhost:4318",
|
|
||||||
or "OTEL_EXPORTER_OTLP_ENDPOINT" if set. *)
|
|
||||||
|
|
||||||
val get_headers : unit -> (string * string) list
|
val get_headers : unit -> (string * string) list
|
||||||
|
|
||||||
val set_headers : (string * string) list -> unit
|
val set_headers : (string * string) list -> unit
|
||||||
|
|
|
||||||
|
|
@ -13,8 +13,6 @@ let mk_client ~scope =
|
||||||
Opentelemetry_cohttp_lwt.client ~scope (module Cohttp_lwt_unix.Client)
|
Opentelemetry_cohttp_lwt.client ~scope (module Cohttp_lwt_unix.Client)
|
||||||
|
|
||||||
let run () =
|
let run () =
|
||||||
Printf.printf "collector is on %S\n%!"
|
|
||||||
(Opentelemetry_client_cohttp_lwt.get_url ());
|
|
||||||
let open Lwt.Syntax in
|
let open Lwt.Syntax in
|
||||||
let rec go () =
|
let rec go () =
|
||||||
let@ scope =
|
let@ scope =
|
||||||
|
|
|
||||||
|
|
@ -74,8 +74,6 @@ let run_job () : unit Lwt.t =
|
||||||
done
|
done
|
||||||
|
|
||||||
let run () : unit Lwt.t =
|
let run () : unit Lwt.t =
|
||||||
Printf.printf "collector is on %S\n%!"
|
|
||||||
(Opentelemetry_client_cohttp_lwt.get_url ());
|
|
||||||
T.GC_metrics.basic_setup ();
|
T.GC_metrics.basic_setup ();
|
||||||
|
|
||||||
T.Metrics_callbacks.register (fun () ->
|
T.Metrics_callbacks.register (fun () ->
|
||||||
|
|
|
||||||
|
|
@ -3,7 +3,6 @@ let url = "http://localhost:3000"
|
||||||
let cohttp () =
|
let cohttp () =
|
||||||
let config = Opentelemetry_client_cohttp_lwt.Config.make ~url () in
|
let config = Opentelemetry_client_cohttp_lwt.Config.make ~url () in
|
||||||
Opentelemetry_client_cohttp_lwt.with_setup ~config () @@ fun () ->
|
Opentelemetry_client_cohttp_lwt.with_setup ~config () @@ fun () ->
|
||||||
let url = Opentelemetry_client_cohttp_lwt.get_url () in
|
|
||||||
print_endline @@ Printf.sprintf "cohttp url = %s" url
|
print_endline @@ Printf.sprintf "cohttp url = %s" url
|
||||||
|
|
||||||
let () = cohttp ()
|
let () = cohttp ()
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue