Additional metrics (#11271)

* Remove _set_state_changed_callback from harvester_api.py as its never actually called

In rpc/rpc_server.py:318, what is actually called is rpc_server.rpc_api.service._set_state_changed_callback, so the version of this in harvester.py is what is actually used

* Remove duplicate initial value for state_changed_callback in init

* _state_changed -> state_changed so it can be called from harvester_api as well

* Add farming info state_changed event from harvester

* Add time to farming_info event

* Add [pre_]validation_time to block event

* Remove unused set_state_changed_callback on full_node_api. This is actually called on full_node_rpc_api.service

* Remove unused set_state_changed_callback on crawler_api. This is actually called on crawler_rpc_api.service (crawler)

* Remove unused set_state_changed_callback on farmer_api. This is actually called on farmer_rpc_api.service (farmer), not the api itself

* Add state changed event for submitting a partial

* Lint fixes

chia/farmer/farmer_api.py

@@ -1,6 +1,6 @@
 import json
 import time
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 import aiohttp
 from blspy import AugSchemeMPL, G2Element, PrivateKey
@@ -50,9 +50,6 @@ class FarmerAPI:
     def __init__(self, farmer) -> None:
         self.farmer = farmer
 
-    def _set_state_changed_callback(self, callback: Callable):
-        self.farmer.state_changed_callback = callback
-
     @api_request
     @peer_required
     async def new_proof_of_space(
@@ -272,6 +269,17 @@ class FarmerAPI:
                     self.farmer.log.error(f"Error connecting to pool: {e}")
                     return
 
+                self.farmer.state_changed(
+                    "submitted_partial",
+                    {
+                        "launcher_id": post_partial_request.payload.launcher_id.hex(),
+                        "pool_url": pool_url,
+                        "current_difficulty": pool_state_dict["current_difficulty"],
+                        "points_acknowledged_since_start": pool_state_dict["points_acknowledged_since_start"],
+                        "points_acknowledged_24h": pool_state_dict["points_acknowledged_24h"],
+                    },
+                )
+
                 return
 
     @api_request

chia/full_node/full_node.py

@@ -1629,6 +1629,8 @@ class FullNode:
             "k_size": block.reward_chain_block.proof_of_space.size,
             "header_hash": block.header_hash,
             "height": block.height,
+            "validation_time": validation_time,
+            "pre_validation_time": pre_validation_time,
         }
 
         if block.transactions_info is not None:

chia/full_node/full_node_api.py

@@ -3,7 +3,7 @@ import dataclasses
 import time
 import traceback
 from secrets import token_bytes
-from typing import Callable, Dict, List, Optional, Tuple, Set
+from typing import Dict, List, Optional, Tuple, Set
 
 from blspy import AugSchemeMPL, G2Element
 from chiabip158 import PyBIP158
@@ -54,9 +54,6 @@ class FullNodeAPI:
     def __init__(self, full_node) -> None:
         self.full_node = full_node
 
-    def _set_state_changed_callback(self, callback: Callable):
-        self.full_node.state_changed_callback = callback
-
     @property
     def server(self):
         return self.full_node.server

chia/harvester/harvester.py

@@ -4,7 +4,7 @@ import dataclasses
 import logging
 from concurrent.futures.thread import ThreadPoolExecutor
 from pathlib import Path
-from typing import Callable, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 import chia.server.ws_connection as ws  # lgtm [py/import-and-import-from]
 from chia.consensus.constants import ConsensusConstants
@@ -58,7 +58,6 @@ class Harvester:
         self.plot_sync_sender = Sender(self.plot_manager)
         self._is_shutdown = False
         self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=config["num_threads"])
-        self.state_changed_callback = None
         self.server = None
         self.constants = constants
         self.cached_challenges = []
@@ -82,9 +81,9 @@ class Harvester:
     def _set_state_changed_callback(self, callback: Callable):
         self.state_changed_callback = callback
 
-    def _state_changed(self, change: str):
+    def state_changed(self, change: str, change_data: Dict[str, Any] = None):
         if self.state_changed_callback is not None:
-            self.state_changed_callback(change)
+            self.state_changed_callback(change, change_data)
 
     def _plot_refresh_callback(self, event: PlotRefreshEvents, update_result: PlotRefreshResult):
         log_function = self.log.debug if event != PlotRefreshEvents.done else self.log.info
@@ -104,7 +103,7 @@ class Harvester:
 
     def on_disconnect(self, connection: ws.WSChiaConnection):
         self.log.info(f"peer disconnected {connection.get_peer_logging()}")
-        self._state_changed("close_connection")
+        self.state_changed("close_connection")
         self.plot_manager.stop_refreshing()
         self.plot_sync_sender.stop()
 
@@ -140,7 +139,7 @@ class Harvester:
     def delete_plot(self, str_path: str):
         remove_plot(Path(str_path))
         self.plot_manager.trigger_refresh()
-        self._state_changed("plots")
+        self.state_changed("plots")
         return True
 
     async def add_plot_directory(self, str_path: str) -> bool:

chia/harvester/harvester_api.py

@@ -1,7 +1,7 @@
 import asyncio
 import time
 from pathlib import Path
-from typing import Callable, List, Tuple
+from typing import List, Tuple
 
 from blspy import AugSchemeMPL, G1Element, G2Element
 
@@ -27,9 +27,6 @@ class HarvesterAPI:
     def __init__(self, harvester: Harvester):
         self.harvester = harvester
 
-    def _set_state_changed_callback(self, callback: Callable):
-        self.harvester.state_changed_callback = callback
-
     @peer_required
     @api_request
     async def harvester_handshake(
@@ -218,11 +215,22 @@ class HarvesterAPI:
         )
         pass_msg = make_msg(ProtocolMessageTypes.farming_info, farming_info)
         await peer.send_message(pass_msg)
+        found_time = time.time() - start
         self.harvester.log.info(
             f"{len(awaitables)} plots were eligible for farming {new_challenge.challenge_hash.hex()[:10]}..."
-            f" Found {total_proofs_found} proofs. Time: {time.time() - start:.5f} s. "
+            f" Found {total_proofs_found} proofs. Time: {found_time:.5f} s. "
             f"Total {self.harvester.plot_manager.plot_count()} plots"
         )
+        self.harvester.state_changed(
+            "farming_info",
+            {
+                "challenge_hash": new_challenge.challenge_hash.hex(),
+                "total_plots": self.harvester.plot_manager.plot_count(),
+                "found_proofs": total_proofs_found,
+                "eligible_plots": len(awaitables),
+                "time": found_time,
+            },
+        )
 
     @api_request
     async def request_signatures(self, request: harvester_protocol.RequestSignatures):

chia/rpc/farmer_rpc_api.py

@@ -24,37 +24,48 @@ class FarmerRpcApi:
         }
 
     async def _state_changed(self, change: str, change_data: Dict) -> List[WsRpcMessage]:
+        payloads = []
+
         if change == "new_signage_point":
             sp_hash = change_data["sp_hash"]
             data = await self.get_signage_point({"sp_hash": sp_hash.hex()})
-            return [
+            payloads.append(
                 create_payload_dict(
                     "new_signage_point",
                     data,
                     self.service_name,
                     "wallet_ui",
                 )
-            ]
+            )
         elif change == "new_farming_info":
-            return [
+            payloads.append(
                 create_payload_dict(
                     "new_farming_info",
                     change_data,
                     self.service_name,
                     "wallet_ui",
                 )
-            ]
+            )
         elif change == "new_plots":
-            return [
+            payloads.append(
                 create_payload_dict(
                     "get_harvesters",
                     change_data,
                     self.service_name,
                     "wallet_ui",
                 )
-            ]
+            )
+        elif change == "submitted_partial":
+            payloads.append(
+                create_payload_dict(
+                    "submitted_partial",
+                    change_data,
+                    self.service_name,
+                    "metrics",
+                )
+            )
 
-        return []
+        return payloads
 
     async def get_signage_point(self, request: Dict) -> Dict:
         sp_hash = hexstr_to_bytes(request["sp_hash"])

chia/rpc/harvester_rpc_api.py

@@ -1,4 +1,4 @@
-from typing import Callable, Dict, List
+from typing import Any, Callable, Dict, List
 
 from chia.harvester.harvester import Harvester
 from chia.util.ws_message import WsRpcMessage, create_payload_dict
@@ -19,12 +19,21 @@ class HarvesterRpcApi:
             "/remove_plot_directory": self.remove_plot_directory,
         }
 
-    async def _state_changed(self, change: str) -> List[WsRpcMessage]:
+    async def _state_changed(self, change: str, change_data: Dict[str, Any] = None) -> List[WsRpcMessage]:
+        if change_data is None:
+            change_data = {}
+
+        payloads = []
+
         if change == "plots":
             data = await self.get_plots({})
             payload = create_payload_dict("get_plots", data, self.service_name, "wallet_ui")
-            return [payload]
-        return []
+            payloads.append(payload)
+
+        if change == "farming_info":
+            payloads.append(create_payload_dict("farming_info", change_data, self.service_name, "metrics"))
+
+        return payloads
 
     async def get_plots(self, request: Dict) -> Dict:
         plots, failed_to_open, not_found = self.service.get_plots()

chia/seeder/crawler_api.py

@@ -1,4 +1,4 @@
-from typing import Callable, Optional
+from typing import Optional
 
 import chia.server.ws_connection as ws
 from chia.full_node.full_node import full_node_protocol, wallet_protocol
@@ -13,9 +13,6 @@ class CrawlerAPI:
     def __init__(self, crawler):
         self.crawler = crawler
 
-    def _set_state_changed_callback(self, callback: Callable):
-        self.crawler.state_changed_callback = callback
-
     def __getattr__(self, attr_name: str):
         async def invoke(*args, **kwargs):
             pass