CP-309305: Split Spans.since into chunks for exporting

Http exporting appears to get overwhelmed when too many spans are
exported at the same time. This adds the option to export a smaller
chunk of spans at a time. This also reduces the size of the file exports
as we only check for max file size after exporting all of the finished
spans.

Signed-off-by: Steven Woods <[email protected]>

Author: snwoods

ocaml/libs/tracing/tracing.ml

@@ -351,7 +351,7 @@ module Span = struct
     let span_id = Span_id.make () in
     let extra_context_with_depth =
       TraceContext.(
-        with_added_baggage depth_key (string_of_int depth) extra_context
+        update_with_baggage depth_key (string_of_int depth) extra_context
       )
     in
     let context : SpanContext.t =
@@ -363,7 +363,7 @@ module Span = struct
       match trace_context with
       | Some tc ->
           let tc_with_depth =
-            TraceContext.(with_added_baggage depth_key (string_of_int depth) tc)
+            TraceContext.(update_with_baggage depth_key (string_of_int depth) tc)
           in
           SpanContext.with_trace_context tc_with_depth context
       | None ->
@@ -751,7 +751,7 @@ module Tracer = struct
     let tc = SpanContext.context_of_span_context context in
     let new_depth = TraceContext.baggage_depth_of tc in
     let new_tc =
-      TraceContext.(with_added_baggage depth_key (string_of_int new_depth) tc)
+      TraceContext.(update_with_baggage depth_key (string_of_int new_depth) tc)
     in
     let context = SpanContext.with_trace_context new_tc context in
     {
@@ -784,8 +784,17 @@ module Tracer = struct
     if not t.enabled then
       ok_none (* Do not start span if the max depth has been reached *)
     else if parent_depth >= Atomic.get Spans.max_depth then (
-      let parent_trace_id = Option.fold ~none:"None" ~some:(fun p -> p.Span.context |> SpanContext.span_id_of_span_context |> Span_id.to_string) parent in
-      debug "Max_span_depth limit reached, not creating span %s (parent %s)" name parent_trace_id ;
+      let parent_trace_id =
+        Option.fold ~none:"None"
+          ~some:(fun p ->
+            p.Span.context
+            |> SpanContext.span_id_of_span_context
+            |> Span_id.to_string
+          )
+          parent
+      in
+      debug "Max_span_depth limit reached, not creating span %s (parent %s)"
+        name parent_trace_id ;
       ok_none
     ) else
       let attributes = Attributes.merge_into t.attributes attributes in
@@ -811,7 +820,7 @@ module Tracer = struct
                  let new_context : SpanContext.t =
                    let trace_context =
                      TraceContext.(
-                       with_added_baggage depth_key (string_of_int new_depth)
+                       update_with_baggage depth_key (string_of_int new_depth)
                          span.Span.context.trace_context
                      )
                    in
@@ -993,7 +1002,7 @@ module Propagator = struct
             in
             let trace_context'' =
               TraceContext.(
-                with_added_baggage depth_key new_depth trace_context'
+                update_with_baggage depth_key new_depth trace_context'
               )
             in
             let carrier' = P.inject_into trace_context'' carrier in

ocaml/libs/tracing/tracing_export.ml

@@ -24,6 +24,10 @@ let export_interval = ref 30.
 
 let set_export_interval t = export_interval := t
 
+let export_chunk_size = Atomic.make 10000
+
+let set_export_chunk_size x = Atomic.set export_chunk_size x
+
 let host_id = ref "localhost"
 
 let set_host_id id = host_id := id
@@ -289,17 +293,40 @@ module Destination = struct
     with exn ->
       debug "Tracing: unable to export span : %s" (Printexc.to_string exn)
 
+  let rec span_info_chunks span_info batch_size =
+    let rec list_to_chunks_inner l n curr chunks =
+      if n = 0 then
+        if List.length l > 0 then
+          list_to_chunks_inner l batch_size [] ((curr, batch_size) :: chunks)
+        else
+          (curr, batch_size) :: chunks
+      else
+        match l with
+        | [] ->
+            (curr, List.length curr) :: chunks
+        | h :: t ->
+            list_to_chunks_inner t (n - 1) (h :: curr) chunks
+    in
+    list_to_chunks_inner (fst span_info) batch_size [] []
+
   let flush_spans () =
     let ((_span_list, span_count) as span_info) = Spans.since () in
     let attributes = [("export.traces.count", string_of_int span_count)] in
     let@ parent =
       with_tracing ~span_kind:Server ~trace_context:TraceContext.empty
         ~parent:None ~attributes ~name:"Tracing.flush_spans"
     in
-    TracerProvider.get_tracer_providers ()
-    |> List.filter TracerProvider.get_enabled
-    |> List.concat_map TracerProvider.get_endpoints
-    |> List.iter (export_to_endpoint parent span_info)
+    let endpoints =
+      TracerProvider.get_tracer_providers ()
+      |> List.filter TracerProvider.get_enabled
+      |> List.concat_map TracerProvider.get_endpoints
+    in
+    let span_info_chunks =
+      span_info_chunks span_info (Atomic.get export_chunk_size)
+    in
+    List.iter
+      (fun s_i -> List.iter (export_to_endpoint parent s_i) endpoints)
+      span_info_chunks
 
   let delay = Delay.make ()
 

ocaml/libs/tracing/tracing_export.mli

@@ -23,6 +23,13 @@ val set_export_interval : float -> unit
     Default is every [30.] seconds.
   *)
 
+val set_export_chunk_size : int -> unit
+(** [set_export_chunk_size size] sets the maximum number of finished spans that
+    can be exported in one chunk to [size].
+
+    Default is 10000 spans.
+  *)
+
 val set_host_id : string -> unit
 (** [set_host_id id] sets the id of the host to [id]. 
 

ocaml/xapi/xapi_globs.ml

@@ -1055,6 +1055,8 @@ let trace_log_dir = ref "/var/log/dt/zipkinv2/json"
 
 let export_interval = ref 30.
 
+let export_chunk_size = ref 10000
+
 let max_spans = ref 10000
 
 let max_traces = ref 10000
@@ -1678,6 +1680,11 @@ let other_options =
     , (fun () -> string_of_float !export_interval)
     , "The interval for exports in Tracing"
     )
+  ; ( "export-chunk-size"
+    , Arg.Set_int export_chunk_size
+    , (fun () -> string_of_int !export_chunk_size)
+    , "The span chunk size for exports in Tracing"
+    )
   ; ( "max-spans"
     , Arg.Set_int max_spans
     , (fun () -> string_of_int !max_spans)

ocaml/xapi/xapi_observer.ml

@@ -600,6 +600,7 @@ let initialise_observer ~__context component =
 
 let initialise ~__context =
   Tracing.Spans.set_max_depth !Xapi_globs.max_span_depth ;
+  Tracing_export.set_export_chunk_size !Xapi_globs.export_chunk_size ;
   List.iter (initialise_observer_meta ~__context) (startup_components ()) ;
   Db.Observer.get_all ~__context
   |> List.iter (fun self ->