MinaProtocol / mina / 411

Committed 24 Jul 2025 03:14PM UTC coverage: 33.188% (-27.7%) from 60.871%

Build # 411

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Merge pull request #17541 from MinaProtocol/brian/merge-compatible-into-develop

Merge compatible into develop

Run Details

164 of 702 new or added lines in 96 files covered. (23.36%)

18243 existing lines in 393 files now uncovered.

23983 of 72264 relevant lines covered (33.19%)

24667.26 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.94

/src/lib/node_status_service/node_status_service.ml

open Async
open Core
open Pipe_lib

type catchup_job_states = Transition_frontier.Full_catchup_tree.job_states =
  { finished : int
  ; failed : int
  ; to_download : int
  ; to_initial_validate : int
  ; wait_for_parent : int
  ; to_verify : int
  ; to_build_breadcrumb : int
  }
[@@deriving to_yojson]

type rpc_count =
  { get_some_initial_peers : int
  ; get_staged_ledger_aux_and_pending_coinbases_at_hash : int
  ; answer_sync_ledger_query : int
  ; get_transition_chain : int
  ; get_transition_knowledge : int
  ; get_transition_chain_proof : int
  ; get_node_status : int
  ; get_ancestry : int
  ; ban_notify : int
  ; get_best_tip : int
  ; get_epoch_ledger : int
  }
[@@deriving to_yojson]

type gossip_count =
  { new_state : int; transaction_pool_diff : int; snark_pool_diff : int }
[@@deriving to_yojson]

type block =
  { hash : Mina_base.State_hash.t
  ; sender : Network_peer.Envelope.Sender.t
  ; received_at : string
  ; is_valid : bool
  ; reason_for_rejection :
      [ `Invalid_proof
      | `Invalid_delta_transition_chain_proof
      | `Too_early
      | `Too_late
      | `Invalid_genesis_protocol_state
      | `Invalid_protocol_version
      | `Mismatched_protocol_version ]
      option
  }
[@@deriving to_yojson]

type sysinfo =
  { uptime : string
  ; load : int
  ; total_ram : int64
  ; free_ram : int64
  ; total_swap : int64
  ; free_swap : int64
  ; procs : int
  }
[@@deriving to_yojson]

type node_status_data =
  { version : int
  ; block_height_at_best_tip : int
  ; max_observed_block_height : int
  ; max_observed_unvalidated_block_height : int
  ; catchup_job_states : catchup_job_states option
  ; sync_status : Sync_status.t
  ; libp2p_input_bandwidth : float
  ; libp2p_output_bandwidth : float
  ; libp2p_cpu_usage : float
  ; commit_hash : string
  ; chain_id : string
  ; peer_id : string
  ; ip_address : string
  ; timestamp : string
  ; uptime_of_node : float
  ; peer_count : int
  ; rpc_received : rpc_count
  ; rpc_sent : rpc_count
  ; pubsub_msg_received : gossip_count
  ; pubsub_msg_broadcasted : gossip_count
  ; received_blocks : block list
  ; sysinfo : sysinfo
  }
[@@deriving to_yojson]

module Simplified = struct
  type t =
    { max_observed_block_height : int
    ; commit_hash : string
    ; chain_id : string
    ; peer_id : string
    ; peer_count : int
    ; timestamp : string
    ; block_producer_public_key : string option
    }
  [@@deriving to_yojson]
end

let send_node_status_data (type data) ~logger ~url (node_status_data : data)
    (node_status_data_to_yojson : data -> Yojson.Safe.t) =
  let node_status_json = node_status_data_to_yojson node_status_data in
  let json = `Assoc [ ("data", node_status_json) ] in
  let headers =
    Cohttp.Header.of_list [ ("Content-Type", "application/json") ]
  in
  match%map
    Async.try_with ~here:[%here] (fun () ->
        Cohttp_async.Client.post ~headers
          ~body:(Yojson.Safe.to_string json |> Cohttp_async.Body.of_string)
          url )
  with
  | Ok ({ status; _ }, body) ->
      let metadata =
        [ ("data", node_status_json); ("url", `String (Uri.to_string url)) ]
      in
      if
        Cohttp.Code.(
          code_of_status status >= 200 && code_of_status status < 300)
      then [%log info] "Sent node status data to URL $url" ~metadata
      else
        let extra_metadata =
          match body with
          | `String s ->
              [ ("error", `String s) ]
          | `Strings ss ->
              [ ("error", `List (List.map ss ~f:(fun s -> `String s))) ]
          | `Empty | `Pipe _ ->
              []
        in
        [%log error] "Failed to send node status data to URL $url"
          ~metadata:(metadata @ extra_metadata)
  | Error e ->
      [%log error] "Failed to send node status data to URL $url"
        ~metadata:
          [ ("error", `String (Exn.to_string e))
          ; ("url", `String (Uri.to_string url))
          ]

let reset_gauges () =
  let gauges =
    let open Mina_metrics.Network in
    [ new_state_received
    ; new_state_broadcasted
    ; transaction_pool_diff_received
    ; transaction_pool_diff_broadcasted
    ; snark_pool_diff_received
    ; snark_pool_diff_broadcasted
    ]
    @ List.map ~f:snd
        [ get_some_initial_peers_rpcs_sent
        ; get_some_initial_peers_rpcs_received
        ; get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_sent
        ; get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_received
        ; answer_sync_ledger_query_rpcs_sent
        ; answer_sync_ledger_query_rpcs_received
        ; get_transition_chain_rpcs_sent
        ; get_transition_chain_rpcs_received
        ; get_transition_knowledge_rpcs_sent
        ; get_transition_knowledge_rpcs_received
        ; get_transition_chain_proof_rpcs_sent
        ; get_transition_chain_proof_rpcs_received
        ; get_node_status_rpcs_sent
        ; get_node_status_rpcs_received
        ; get_ancestry_rpcs_sent
        ; get_ancestry_rpcs_received
        ; ban_notify_rpcs_sent
        ; ban_notify_rpcs_received
        ; get_best_tip_rpcs_sent
        ; get_best_tip_rpcs_received
        ; get_epoch_ledger_rpcs_sent
        ; get_epoch_ledger_rpcs_received
        ]
  in
  List.iter ~f:(fun gauge -> Mina_metrics.Gauge.set gauge 0.) gauges ;
  Queue.clear Transition_frontier.validated_blocks ;
  Queue.clear Transition_frontier.rejected_blocks

let start ~commit_id ~logger ~node_status_url ~transition_frontier ~sync_status
    ~chain_id ~network ~addrs_and_ports ~start_time ~slot_duration =
  [%log info] "Starting node status service using URL $url"
    ~metadata:[ ("url", `String node_status_url) ] ;
  let five_slots = Time.Span.scale slot_duration 5. in
  reset_gauges () ;
  every ~start:(after five_slots) ~continue_on_error:true five_slots
  @@ fun () ->
  don't_wait_for
  @@ O1trace.thread "node_status_service"
  @@ fun () ->
  match Broadcast_pipe.Reader.peek transition_frontier with
  | None ->
      [%log info] "Transition frontier not available for node status service" ;
      Deferred.unit
  | Some tf -> (
      let catchup_job_states =
        match Transition_frontier.catchup_state tf with
        | Full catchup_state ->
            Some
              (Transition_frontier.Full_catchup_tree.to_node_status_report
                 catchup_state )
      in
      let sync_status =
        sync_status |> Mina_incremental.Status.Observer.value_exn
      in
      let sysinfo =
        match Linux_ext.Sysinfo.sysinfo with
        | Ok sysinfo ->
            let open Int64 in
            let info = sysinfo () in
            let mem_unit = of_int info.mem_unit in
            { uptime = Time.Span.to_string info.uptime
            ; load = info.load15
            ; total_ram = of_int info.total_ram * mem_unit
            ; free_ram = of_int info.free_ram * mem_unit
            ; total_swap = of_int info.total_swap * mem_unit
            ; free_swap = of_int info.free_swap * mem_unit
            ; procs = info.procs
            }
        | Error e ->
            [%log error] "Failed to get sysinfo: $error"
              ~metadata:[ ("error", `String (Error.to_string_hum e)) ] ;
            { uptime = ""
            ; load = 0
            ; total_ram = 0L
            ; free_ram = 0L
            ; total_swap = 0L
            ; free_swap = 0L
            ; procs = 0
            }
      in
      [%log info] "About to send bandwidth request to libp2p" ;
      match%bind Mina_networking.bandwidth_info network with
      | Ok
          ( `Input libp2p_input_bandwidth
          , `Output libp2p_output_bandwidth
          , `Cpu_usage libp2p_cpu_usage ) ->
          let%bind peers = Mina_networking.peers network in
          let node_status_data =
            { version = 1
            ; block_height_at_best_tip =
                Transition_frontier.best_tip tf
                |> Transition_frontier.Breadcrumb.consensus_state
                |> Consensus.Data.Consensus_state.blockchain_length
                |> Mina_numbers.Length.to_uint32 |> Unsigned.UInt32.to_int
            ; max_observed_block_height =
                !Mina_metrics.Transition_frontier.max_blocklength_observed
            ; max_observed_unvalidated_block_height =
                !Mina_metrics.Transition_frontier
                 .max_unvalidated_blocklength_observed
            ; catchup_job_states
            ; sync_status
            ; libp2p_input_bandwidth
            ; libp2p_output_bandwidth
            ; libp2p_cpu_usage
            ; commit_hash = commit_id
            ; chain_id
            ; peer_id =
                (Node_addrs_and_ports.to_peer_exn addrs_and_ports).peer_id
            ; ip_address =
                Node_addrs_and_ports.external_ip addrs_and_ports
                |> Core.Unix.Inet_addr.to_string
            ; timestamp = Mina_stdlib_unix.Rfc3339_time.get_rfc3339_time ()
            ; uptime_of_node =
                Time.Span.to_sec @@ Time.diff (Time.now ()) start_time
            ; peer_count = List.length peers
            ; rpc_sent =
                { get_some_initial_peers =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_some_initial_peers_rpcs_sent
                ; get_staged_ledger_aux_and_pending_coinbases_at_hash =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_sent
                ; answer_sync_ledger_query =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network.answer_sync_ledger_query_rpcs_sent
                ; get_transition_chain =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_transition_chain_proof_rpcs_sent
                ; get_transition_knowledge =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network.get_transition_knowledge_rpcs_sent
                ; get_transition_chain_proof =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_transition_chain_proof_rpcs_sent
                ; get_node_status =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_node_status_rpcs_sent
                ; get_ancestry =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_ancestry_rpcs_sent
                ; ban_notify =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.ban_notify_rpcs_sent
                ; get_best_tip =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_best_tip_rpcs_sent
                ; get_epoch_ledger =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_epoch_ledger_rpcs_sent
                }
            ; rpc_received =
                { get_some_initial_peers =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_some_initial_peers_rpcs_received
                ; get_staged_ledger_aux_and_pending_coinbases_at_hash =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_received
                ; answer_sync_ledger_query =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .answer_sync_ledger_query_rpcs_received
                ; get_transition_chain =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_transition_chain_proof_rpcs_received
                ; get_transition_knowledge =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_transition_knowledge_rpcs_received
                ; get_transition_chain_proof =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd
                         Mina_metrics.Network
                         .get_transition_chain_proof_rpcs_received
                ; get_node_status =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_node_status_rpcs_received
                ; get_ancestry =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_ancestry_rpcs_received
                ; ban_notify =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.ban_notify_rpcs_received
                ; get_best_tip =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_best_tip_rpcs_received
                ; get_epoch_ledger =
                    Float.to_int @@ Mina_metrics.Gauge.value
                    @@ snd Mina_metrics.Network.get_epoch_ledger_rpcs_received
                }
            ; pubsub_msg_received =
                { new_state =
                    Float.to_int
                    @@ Mina_metrics.Gauge.value
                         Mina_metrics.Network.new_state_received
                ; transaction_pool_diff =
                    Float.to_int
                    @@ Mina_metrics.Gauge.value
                         Mina_metrics.Network.transaction_pool_diff_received
                ; snark_pool_diff =
                    Float.to_int
                    @@ Mina_metrics.Gauge.value
                         Mina_metrics.Network.snark_pool_diff_received
                }
            ; pubsub_msg_broadcasted =
                { new_state =
                    Float.to_int
                    @@ Mina_metrics.Gauge.value
                         Mina_metrics.Network.new_state_broadcasted
                ; transaction_pool_diff =
                    Float.to_int
                    @@ Mina_metrics.Gauge.value
                         Mina_metrics.Network.transaction_pool_diff_broadcasted
                ; snark_pool_diff =
                    Float.to_int
                    @@ Mina_metrics.Gauge.value
                         Mina_metrics.Network.snark_pool_diff_broadcasted
                }
            ; received_blocks =
                List.map (Queue.to_list Transition_frontier.rejected_blocks)
                  ~f:(fun (hash, sender, received_at, reason_for_rejection) ->
                    { hash
                    ; sender
                    ; received_at =
                        Time.to_string (Block_time.to_time_exn received_at)
                    ; is_valid = false
                    ; reason_for_rejection = Some reason_for_rejection
                    } )
                @ List.map (Queue.to_list Transition_frontier.validated_blocks)
                    ~f:(fun (hash, sender, received_at) ->
                      { hash
                      ; sender
                      ; received_at =
                          Time.to_string (Block_time.to_time_exn received_at)
                      ; is_valid = true
                      ; reason_for_rejection = None
                      } )
            ; sysinfo
            }
          in
          reset_gauges () ;
          send_node_status_data ~logger
            ~url:(Uri.of_string node_status_url)
            node_status_data node_status_data_to_yojson
      | Error e ->
          [%log info]
            ~metadata:[ ("error", `String (Error.to_string_hum e)) ]
            "Failed to get bandwidth info from libp2p" ;
          Deferred.unit )

let start_simplified ~commit_id ~logger ~node_status_url ~chain_id ~network
    ~addrs_and_ports ~slot_duration ~block_producer_public_key_base58 =
  [%log info] "Starting simplified node status service using URL $url"
    ~metadata:[ ("url", `String node_status_url) ] ;
  let five_slots = Time.Span.scale slot_duration 5. in
  every ~start:(after five_slots) ~continue_on_error:true five_slots
  @@ fun () ->
  don't_wait_for
  @@ let%bind peers = Mina_networking.peers network in
     let node_status_data =
       { Simplified.max_observed_block_height =
           !Mina_metrics.Transition_frontier.max_blocklength_observed
       ; commit_hash = commit_id
       ; chain_id
       ; peer_id = (Node_addrs_and_ports.to_peer_exn addrs_and_ports).peer_id
       ; peer_count = List.length peers
       ; timestamp = Mina_stdlib_unix.Rfc3339_time.get_rfc3339_time ()
       ; block_producer_public_key = block_producer_public_key_base58
       }
     in
     send_node_status_data ~logger
       ~url:(Uri.of_string node_status_url)
       node_status_data Simplified.to_yojson

1	open Async	53✔
2	open Core
3	open Pipe_lib
4
5	type catchup_job_states = Transition_frontier.Full_catchup_tree.job_states =	×
6	{ finished : int	×
7	; failed : int	×
8	; to_download : int	×
9	; to_initial_validate : int	×
10	; wait_for_parent : int	×
11	; to_verify : int	×
12	; to_build_breadcrumb : int	×
13	}
14	[@@deriving to_yojson]
15
16	type rpc_count =	×
17	{ get_some_initial_peers : int	×
18	; get_staged_ledger_aux_and_pending_coinbases_at_hash : int	×
19	; answer_sync_ledger_query : int	×
20	; get_transition_chain : int	×
21	; get_transition_knowledge : int	×
22	; get_transition_chain_proof : int	×
23	; get_node_status : int	×
24	; get_ancestry : int	×
25	; ban_notify : int	×
26	; get_best_tip : int	×
27	; get_epoch_ledger : int	×
28	}
29	[@@deriving to_yojson]
30
31	type gossip_count =	×
32	{ new_state : int; transaction_pool_diff : int; snark_pool_diff : int }	×
33	[@@deriving to_yojson]
34
35	type block =	×
36	{ hash : Mina_base.State_hash.t	×
37	; sender : Network_peer.Envelope.Sender.t	×
38	; received_at : string	×
39	; is_valid : bool	×
40	; reason_for_rejection :
41	[ `Invalid_proof	×
42	\| `Invalid_delta_transition_chain_proof
43	\| `Too_early
44	\| `Too_late
45	\| `Invalid_genesis_protocol_state
46	\| `Invalid_protocol_version
47	\| `Mismatched_protocol_version ]
48	option	×
49	}
50	[@@deriving to_yojson]
51
52	type sysinfo =	×
53	{ uptime : string	×
54	; load : int	×
55	; total_ram : int64	×
56	; free_ram : int64	×
57	; total_swap : int64	×
58	; free_swap : int64	×
59	; procs : int	×
60	}
61	[@@deriving to_yojson]
62
63	type node_status_data =	×
64	{ version : int	×
65	; block_height_at_best_tip : int	×
66	; max_observed_block_height : int	×
67	; max_observed_unvalidated_block_height : int	×
68	; catchup_job_states : catchup_job_states option	×
69	; sync_status : Sync_status.t	×
70	; libp2p_input_bandwidth : float	×
71	; libp2p_output_bandwidth : float	×
72	; libp2p_cpu_usage : float	×
73	; commit_hash : string	×
74	; chain_id : string	×
75	; peer_id : string	×
76	; ip_address : string	×
77	; timestamp : string	×
78	; uptime_of_node : float	×
79	; peer_count : int	×
80	; rpc_received : rpc_count	×
81	; rpc_sent : rpc_count	×
82	; pubsub_msg_received : gossip_count	×
83	; pubsub_msg_broadcasted : gossip_count	×
84	; received_blocks : block list	×
85	; sysinfo : sysinfo	×
86	}
87	[@@deriving to_yojson]
88
89	module Simplified = struct
90	type t =	×
91	{ max_observed_block_height : int	×
92	; commit_hash : string	×
93	; chain_id : string	×
94	; peer_id : string	×
95	; peer_count : int	×
96	; timestamp : string	×
97	; block_producer_public_key : string option	×
98	}
99	[@@deriving to_yojson]
100	end
101
102	let send_node_status_data (type data) ~logger ~url (node_status_data : data)
103	(node_status_data_to_yojson : data -> Yojson.Safe.t) =
104	let node_status_json = node_status_data_to_yojson node_status_data in	×
105	let json = `Assoc [ ("data", node_status_json) ] in	×
106	let headers =
107	Cohttp.Header.of_list [ ("Content-Type", "application/json") ]
108	in
109	match%map
110	Async.try_with ~here:[%here] (fun () ->	×
111	Cohttp_async.Client.post ~headers	×
112	~body:(Yojson.Safe.to_string json \|> Cohttp_async.Body.of_string)	×
113	url )
114	with
115	\| Ok ({ status; _ }, body) ->	×
116	let metadata =
117	[ ("data", node_status_json); ("url", `String (Uri.to_string url)) ]	×
118	in
119	if
120	Cohttp.Code.(
121	code_of_status status >= 200 && code_of_status status < 300)	×
122	then [%log info] "Sent node status data to URL $url" ~metadata	×
123	else
124	let extra_metadata =	×
125	match body with
126	\| `String s ->	×
127	[ ("error", `String s) ]
128	\| `Strings ss ->	×
129	[ ("error", `List (List.map ss ~f:(fun s -> `String s))) ]	×
130	\| `Empty \| `Pipe _ ->	×
131	[]
132	in
133	[%log error] "Failed to send node status data to URL $url"	×
134	~metadata:(metadata @ extra_metadata)
135	\| Error e ->	×
136	[%log error] "Failed to send node status data to URL $url"	×
137	~metadata:
138	[ ("error", `String (Exn.to_string e))	×
139	; ("url", `String (Uri.to_string url))	×
140	]
141
142	let reset_gauges () =
143	let gauges =	×
144	let open Mina_metrics.Network in
145	[ new_state_received
146	; new_state_broadcasted
147	; transaction_pool_diff_received
148	; transaction_pool_diff_broadcasted
149	; snark_pool_diff_received
150	; snark_pool_diff_broadcasted
151	]
152	@ List.map ~f:snd	×
153	[ get_some_initial_peers_rpcs_sent
154	; get_some_initial_peers_rpcs_received
155	; get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_sent
156	; get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_received
157	; answer_sync_ledger_query_rpcs_sent
158	; answer_sync_ledger_query_rpcs_received
159	; get_transition_chain_rpcs_sent
160	; get_transition_chain_rpcs_received
161	; get_transition_knowledge_rpcs_sent
162	; get_transition_knowledge_rpcs_received
163	; get_transition_chain_proof_rpcs_sent
164	; get_transition_chain_proof_rpcs_received
165	; get_node_status_rpcs_sent
166	; get_node_status_rpcs_received
167	; get_ancestry_rpcs_sent
168	; get_ancestry_rpcs_received
169	; ban_notify_rpcs_sent
170	; ban_notify_rpcs_received
171	; get_best_tip_rpcs_sent
172	; get_best_tip_rpcs_received
173	; get_epoch_ledger_rpcs_sent
174	; get_epoch_ledger_rpcs_received
175	]
176	in
177	List.iter ~f:(fun gauge -> Mina_metrics.Gauge.set gauge 0.) gauges ;	×
178	Queue.clear Transition_frontier.validated_blocks ;	×
179	Queue.clear Transition_frontier.rejected_blocks	×
180
181	let start ~commit_id ~logger ~node_status_url ~transition_frontier ~sync_status
182	~chain_id ~network ~addrs_and_ports ~start_time ~slot_duration =
183	[%log info] "Starting node status service using URL $url"	×
184	~metadata:[ ("url", `String node_status_url) ] ;
185	let five_slots = Time.Span.scale slot_duration 5. in	×
186	reset_gauges () ;	×
187	every ~start:(after five_slots) ~continue_on_error:true five_slots	×
188	@@ fun () ->
189	don't_wait_for	×
190	@@ O1trace.thread "node_status_service"	×
191	@@ fun () ->
192	match Broadcast_pipe.Reader.peek transition_frontier with	×
193	\| None ->	×
194	[%log info] "Transition frontier not available for node status service" ;	×
195	Deferred.unit	×
196	\| Some tf -> (	×
197	let catchup_job_states =
198	match Transition_frontier.catchup_state tf with
199	\| Full catchup_state ->	×
200	Some
201	(Transition_frontier.Full_catchup_tree.to_node_status_report	×
202	catchup_state )
203	in
204	let sync_status =
205	sync_status \|> Mina_incremental.Status.Observer.value_exn
206	in
207	let sysinfo =	×
208	match Linux_ext.Sysinfo.sysinfo with
209	\| Ok sysinfo ->	×
210	let open Int64 in
211	let info = sysinfo () in
212	let mem_unit = of_int info.mem_unit in	×
213	{ uptime = Time.Span.to_string info.uptime	×
214	; load = info.load15
215	; total_ram = of_int info.total_ram * mem_unit	×
216	; free_ram = of_int info.free_ram * mem_unit	×
217	; total_swap = of_int info.total_swap * mem_unit	×
218	; free_swap = of_int info.free_swap * mem_unit	×
219	; procs = info.procs
220	}
221	\| Error e ->	×
222	[%log error] "Failed to get sysinfo: $error"	×
223	~metadata:[ ("error", `String (Error.to_string_hum e)) ] ;	×
224	{ uptime = ""	×
225	; load = 0
226	; total_ram = 0L
227	; free_ram = 0L
228	; total_swap = 0L
229	; free_swap = 0L
230	; procs = 0
231	}
232	in
233	[%log info] "About to send bandwidth request to libp2p" ;	×
234	match%bind Mina_networking.bandwidth_info network with	×
235	\| Ok	×
236	( `Input libp2p_input_bandwidth
237	, `Output libp2p_output_bandwidth
238	, `Cpu_usage libp2p_cpu_usage ) ->
239	let%bind peers = Mina_networking.peers network in	×
240	let node_status_data =	×
241	{ version = 1
242	; block_height_at_best_tip =
243	Transition_frontier.best_tip tf	×
244	\|> Transition_frontier.Breadcrumb.consensus_state	×
245	\|> Consensus.Data.Consensus_state.blockchain_length	×
246	\|> Mina_numbers.Length.to_uint32 \|> Unsigned.UInt32.to_int	×
247	; max_observed_block_height =
248	!Mina_metrics.Transition_frontier.max_blocklength_observed
249	; max_observed_unvalidated_block_height =
250	!Mina_metrics.Transition_frontier
251	.max_unvalidated_blocklength_observed
252	; catchup_job_states
253	; sync_status
254	; libp2p_input_bandwidth
255	; libp2p_output_bandwidth
256	; libp2p_cpu_usage
257	; commit_hash = commit_id
258	; chain_id
259	; peer_id =
260	(Node_addrs_and_ports.to_peer_exn addrs_and_ports).peer_id	×
261	; ip_address =
262	Node_addrs_and_ports.external_ip addrs_and_ports	×
263	\|> Core.Unix.Inet_addr.to_string	×
NEW 264	; timestamp = Mina_stdlib_unix.Rfc3339_time.get_rfc3339_time ()	×
265	; uptime_of_node =
266	Time.Span.to_sec @@ Time.diff (Time.now ()) start_time	×
267	; peer_count = List.length peers	×
268	; rpc_sent =
269	{ get_some_initial_peers =
270	Float.to_int @@ Mina_metrics.Gauge.value	×
271	@@ snd Mina_metrics.Network.get_some_initial_peers_rpcs_sent	×
272	; get_staged_ledger_aux_and_pending_coinbases_at_hash =
273	Float.to_int @@ Mina_metrics.Gauge.value	×
274	@@ snd	×
275	Mina_metrics.Network
276	.get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_sent
277	; answer_sync_ledger_query =
278	Float.to_int @@ Mina_metrics.Gauge.value	×
279	@@ snd	×
280	Mina_metrics.Network.answer_sync_ledger_query_rpcs_sent
281	; get_transition_chain =
282	Float.to_int @@ Mina_metrics.Gauge.value	×
283	@@ snd	×
284	Mina_metrics.Network
285	.get_transition_chain_proof_rpcs_sent
286	; get_transition_knowledge =
287	Float.to_int @@ Mina_metrics.Gauge.value	×
288	@@ snd	×
289	Mina_metrics.Network.get_transition_knowledge_rpcs_sent
290	; get_transition_chain_proof =
291	Float.to_int @@ Mina_metrics.Gauge.value	×
292	@@ snd	×
293	Mina_metrics.Network
294	.get_transition_chain_proof_rpcs_sent
295	; get_node_status =
296	Float.to_int @@ Mina_metrics.Gauge.value	×
297	@@ snd Mina_metrics.Network.get_node_status_rpcs_sent	×
298	; get_ancestry =
299	Float.to_int @@ Mina_metrics.Gauge.value	×
300	@@ snd Mina_metrics.Network.get_ancestry_rpcs_sent	×
301	; ban_notify =
302	Float.to_int @@ Mina_metrics.Gauge.value	×
303	@@ snd Mina_metrics.Network.ban_notify_rpcs_sent	×
304	; get_best_tip =
305	Float.to_int @@ Mina_metrics.Gauge.value	×
306	@@ snd Mina_metrics.Network.get_best_tip_rpcs_sent	×
307	; get_epoch_ledger =
308	Float.to_int @@ Mina_metrics.Gauge.value	×
309	@@ snd Mina_metrics.Network.get_epoch_ledger_rpcs_sent	×
310	}
311	; rpc_received =
312	{ get_some_initial_peers =
313	Float.to_int @@ Mina_metrics.Gauge.value	×
314	@@ snd	×
315	Mina_metrics.Network
316	.get_some_initial_peers_rpcs_received
317	; get_staged_ledger_aux_and_pending_coinbases_at_hash =
318	Float.to_int @@ Mina_metrics.Gauge.value	×
319	@@ snd	×
320	Mina_metrics.Network
321	.get_staged_ledger_aux_and_pending_coinbases_at_hash_rpcs_received
322	; answer_sync_ledger_query =
323	Float.to_int @@ Mina_metrics.Gauge.value	×
324	@@ snd	×
325	Mina_metrics.Network
326	.answer_sync_ledger_query_rpcs_received
327	; get_transition_chain =
328	Float.to_int @@ Mina_metrics.Gauge.value	×
329	@@ snd	×
330	Mina_metrics.Network
331	.get_transition_chain_proof_rpcs_received
332	; get_transition_knowledge =
333	Float.to_int @@ Mina_metrics.Gauge.value	×
334	@@ snd	×
335	Mina_metrics.Network
336	.get_transition_knowledge_rpcs_received
337	; get_transition_chain_proof =
338	Float.to_int @@ Mina_metrics.Gauge.value	×
339	@@ snd	×
340	Mina_metrics.Network
341	.get_transition_chain_proof_rpcs_received
342	; get_node_status =
343	Float.to_int @@ Mina_metrics.Gauge.value	×
344	@@ snd Mina_metrics.Network.get_node_status_rpcs_received	×
345	; get_ancestry =
346	Float.to_int @@ Mina_metrics.Gauge.value	×
347	@@ snd Mina_metrics.Network.get_ancestry_rpcs_received	×
348	; ban_notify =
349	Float.to_int @@ Mina_metrics.Gauge.value	×
350	@@ snd Mina_metrics.Network.ban_notify_rpcs_received	×
351	; get_best_tip =
352	Float.to_int @@ Mina_metrics.Gauge.value	×
353	@@ snd Mina_metrics.Network.get_best_tip_rpcs_received	×
354	; get_epoch_ledger =
355	Float.to_int @@ Mina_metrics.Gauge.value	×
356	@@ snd Mina_metrics.Network.get_epoch_ledger_rpcs_received	×
357	}
358	; pubsub_msg_received =
359	{ new_state =
360	Float.to_int	×
361	@@ Mina_metrics.Gauge.value	×
362	Mina_metrics.Network.new_state_received
363	; transaction_pool_diff =
364	Float.to_int	×
365	@@ Mina_metrics.Gauge.value	×
366	Mina_metrics.Network.transaction_pool_diff_received
367	; snark_pool_diff =
368	Float.to_int	×
369	@@ Mina_metrics.Gauge.value	×
370	Mina_metrics.Network.snark_pool_diff_received
371	}
372	; pubsub_msg_broadcasted =
373	{ new_state =
374	Float.to_int	×
375	@@ Mina_metrics.Gauge.value	×
376	Mina_metrics.Network.new_state_broadcasted
377	; transaction_pool_diff =
378	Float.to_int	×
379	@@ Mina_metrics.Gauge.value	×
380	Mina_metrics.Network.transaction_pool_diff_broadcasted
381	; snark_pool_diff =
382	Float.to_int	×
383	@@ Mina_metrics.Gauge.value	×
384	Mina_metrics.Network.snark_pool_diff_broadcasted
385	}
386	; received_blocks =
387	List.map (Queue.to_list Transition_frontier.rejected_blocks)	×
388	~f:(fun (hash, sender, received_at, reason_for_rejection) ->
389	{ hash	×
390	; sender
391	; received_at =
392	Time.to_string (Block_time.to_time_exn received_at)	×
393	; is_valid = false
394	; reason_for_rejection = Some reason_for_rejection
395	} )
396	@ List.map (Queue.to_list Transition_frontier.validated_blocks)	×
397	~f:(fun (hash, sender, received_at) ->
398	{ hash	×
399	; sender
400	; received_at =
401	Time.to_string (Block_time.to_time_exn received_at)	×
402	; is_valid = true
403	; reason_for_rejection = None
404	} )
405	; sysinfo
406	}
407	in
408	reset_gauges () ;
409	send_node_status_data ~logger	×
410	~url:(Uri.of_string node_status_url)	×
411	node_status_data node_status_data_to_yojson
412	\| Error e ->	×
413	[%log info]	×
414	~metadata:[ ("error", `String (Error.to_string_hum e)) ]	×
415	"Failed to get bandwidth info from libp2p" ;
416	Deferred.unit )	×
417
418	let start_simplified ~commit_id ~logger ~node_status_url ~chain_id ~network
419	~addrs_and_ports ~slot_duration ~block_producer_public_key_base58 =
420	[%log info] "Starting simplified node status service using URL $url"	×
421	~metadata:[ ("url", `String node_status_url) ] ;
422	let five_slots = Time.Span.scale slot_duration 5. in	×
423	every ~start:(after five_slots) ~continue_on_error:true five_slots	×
424	@@ fun () ->
425	don't_wait_for	×
426	@@ let%bind peers = Mina_networking.peers network in	×
427	let node_status_data =	×
428	{ Simplified.max_observed_block_height =
429	!Mina_metrics.Transition_frontier.max_blocklength_observed
430	; commit_hash = commit_id
431	; chain_id
432	; peer_id = (Node_addrs_and_ports.to_peer_exn addrs_and_ports).peer_id	×
433	; peer_count = List.length peers	×
NEW 434	; timestamp = Mina_stdlib_unix.Rfc3339_time.get_rfc3339_time ()	×
435	; block_producer_public_key = block_producer_public_key_base58
436	}
437	in
438	send_node_status_data ~logger
439	~url:(Uri.of_string node_status_url)	×
440	node_status_data Simplified.to_yojson	106✔

MinaProtocol / mina / 411

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous