[Data][Solana] Report Solana batch requests to data pipeline

Tiltfile

@@ -1,3 +1,6 @@
+# TODO_TECHDEBT(@adshmh): Include a simple data pipeline in Local development mode.
+# Example: fluentd with logging to stdout.
+#
 # Load necessary Tilt extensions
 load("ext://restart_process", "docker_build_with_restart")
 load("ext://helm_resource", "helm_resource", "helm_repo")
@@ -310,4 +313,4 @@ local_resource(
 #     ''',
 #     labels=["k8s_logs"],
 #     resource_deps=["path-stack"]
-# )
+# )

data/legacy_qos.go

@@ -35,9 +35,7 @@ func setLegacyFieldsFromQoSObservations(
 
 	// Use Solana observations to update the legacy record's fields.
 	if solanaObservations := observations.GetSolana(); solanaObservations != nil {
-		populatedRecord := setLegacyFieldsFromQoSSolanaObservations(logger, baseLegacyRecord, solanaObservations)
-		// Solana does not support batch requests so expect a single record.
-		return []*legacyRecord{populatedRecord}
+		return setLegacyFieldsFromQoSSolanaObservations(logger, baseLegacyRecord, solanaObservations)
 	}
 
 	// Use Cosmos SDK observations to update the legacy record's fields.
@@ -146,13 +144,13 @@ func populateEVMErrorFields(legacyRecord *legacyRecord, evmInterpreter *qosobser
 // Returns: the populated legacy record
 func setLegacyFieldsFromQoSSolanaObservations(
 	logger polylog.Logger,
-	legacyRecord *legacyRecord,
+	record *legacyRecord,
 	observations *qosobservation.SolanaRequestObservations,
-) *legacyRecord {
+) []*legacyRecord {
 	logger = logger.With("method", "setLegacyFieldsFromQoSSolanaObservations")
 
 	// In bytes: the length of the request: float64 type is for compatibility with the legacy data pipeline.
-	legacyRecord.RequestDataSize = float64(observations.RequestPayloadLength)
+	record.RequestDataSize = float64(observations.RequestPayloadLength)
 
 	// Initialize the Solana observations interpreter.
 	// Used to extract required fields from the observations.
@@ -162,24 +160,49 @@ func setLegacyFieldsFromQoSSolanaObservations(
 	}
 
 	// Extract the JSONRPC request's method.
-	legacyRecord.ChainMethod = solanaInterpreter.GetRequestMethod()
+	record.ChainMethod = solanaInterpreter.GetRequestMethod()
 
 	// ErrorType is already set at gateway or protocol level.
 	// Skip updating the error fields to preserve the original error.
-	if legacyRecord.ErrorType != "" {
-		return legacyRecord
+	if record.ErrorType != "" {
+		return []*legacyRecord{record}
+	}
+
+	// TODO_UPNEXT(@adshmh): Track and report the `method` field of each JSONRPC request in a batch.
+	// - This requires updating the gateway.QoSRequestContext interface to guarantee a 1:1 map between requests of a batch and responses.
+	//
+	// TODO_TECHDEBT(@adshmh): Track each request of a batch JSONRPC request separately in proto messages.
+	// TODO_TECHDEBT(@adshmh): Include a num_requests fields for batch JSONRPC requests once data pipeline is refactored.
+	endpointObservations := observations.GetEndpointObservations()
+	// 0 or 1 endpoint observations: not a batch JSONRPC request.
+	if len(endpointObservations) <= 1 {
+		return []*legacyRecord{record}
 	}
 
-	errType := solanaInterpreter.GetRequestErrorType()
-	legacyRecord.ErrorType = errType
-	legacyRecord.ErrorMessage = errType
+	// TODO_UPNEXT(@adshmh): Track and report errors on each request of a JSONRPC batch request.
+	//
+	// Create a separate legacy record for each method
+	var legacyRecords []*legacyRecord
+	for index := range endpointObservations {
+		// Create a copy of the base record
+		recordCopy := *record
+
+		// Track the index of the request to ensure correctness of records.
+		recordCopy.ChainMethod = fmt.Sprintf("batch_request_index:%d", index)
+
+		legacyRecords = append(legacyRecords, &recordCopy)
+	}
 
-	return legacyRecord
+	return legacyRecords
 }
 
 // qosCosmosErrorTypeStr defines the prefix for Cosmos QoS error types in legacy records
 const qosCosmosErrorTypeStr = "QOS_COSMOS"
 
+// TODO_TECHDEBT(@adshmh): Refactor the data reporting logic:
+// - QoS logic should simply return a list of legacy records (to eliminate the need for copying the original record)
+// - Protocol and Gateway logic sets the fields related to their perspective on the QoS returned set of records.
+//
 // setLegacyFieldsFromQoSCosmosObservations populates legacy records with Cosmos SDK-specific QoS data.
 // It captures:
 // - Request payload size (aggregated across all request profiles)
@@ -217,6 +240,8 @@ func setLegacyFieldsFromQoSCosmosObservations(
 		return []*legacyRecord{baseLegacyRecord}
 	}
 
+	// TODO_TECHDEBT(@adshmh): Refactor to loop over the endpoint observations instead.
+	//
 	// Create a separate legacy record for each method
 	// This enables the data pipeline to track metrics per individual method
 	// Similar to EVM batch request handling

proto/path/qos/cosmos.proto

@@ -9,6 +9,9 @@ import "path/qos/cosmos_response.proto";
 import "path/qos/request_origin.proto";
 import "path/qos/request_error.proto";
 
+// TODO_TECHDEBT(@adshmh): Reorganize the messages to be consistent with both single and batch JSONRPC requests:
+// - Directly associate each request of a batch with the corresponding endpoint observation(s).
+//
 // CosmosRequestObservations captures all observations made while serving a single Cosmos blockchain service request.
 message CosmosRequestObservations {
     // Next free index: 10

proto/path/qos/evm.proto

@@ -164,6 +164,9 @@ message EVMRequestUnmarshalingFailure {
   optional string error_details = 3;
 }
 
+// TODO_TECHDEBT(@adshmh): Enhance the endpoint observation to include the corresponding request's details (e.g. method field of JSONRPC)
+// This will enable tracking each request of a batch of JSONRPC request alongside the endpoint's response.
+//
 // EVMEndpointObservation stores a single observation from an endpoint servicing the protocol response.
 // Example: A Pocket node on Shannon backed by an Ethereum data node servicing an `eth_getBlockNumber` request.
 message EVMEndpointObservation {

proto/path/qos/solana.proto

@@ -29,6 +29,10 @@ message SolanaRequestObservations {
   // Tracks request errors, if any.
   optional RequestError request_error = 5;
 
+  // TODO_TECHDEBT(@adshmh): refactor this proto struct to add separate entries for batch JSONRPC requests.
+  // - Introduce a batch JSONRPC request message.
+  // - Each batch contains one or more JSONRPC requests, each with their separate endpoint observations.
+  //
   // JSON-RPC request to the Solana blockchain service.
   // Only set if the HTTP request payload was successfully parsed into JSONRPC.
   // TODO_TECHDEBT: This assumes all SolanaVM blockchains only (and always) support JSON-RPC.

qos/evm/context.go

@@ -285,6 +285,9 @@ func (rc requestContext) createNoResponseObservations() []*qosobservations.EVMRe
 func (rc requestContext) createResponseObservations() []*qosobservations.EVMRequestObservation {
 	var observations []*qosobservations.EVMRequestObservation
 
+	// TODO_TECHDEBT(@adshmh): Simplify this code to use the order of payloads in the slice to map them to the requests in a batch of JSONRPC requests.
+	// This requires gateway package's RequestQoSContext interface to be updated to accept a slice of responses.
+	//
 	for _, endpointResp := range rc.endpointResponses {
 		var jsonrpcResponse jsonrpc.Response
 		err := json.Unmarshal(endpointResp.GetHTTPResponse().GetPayload(), &jsonrpcResponse)

qos/solana/context_batch.go

@@ -51,7 +51,7 @@ type batchJSONRPCRequestContext struct {
 	// - QoS: requests built by the QoS service to get additional data points on endpoints.
 	requestOrigin qosobservations.RequestOrigin
 
-	// endpointResponses is the set of responses received from one or
+	// endpointJSONRPCResponses is the set of responses received from one or
 	// more endpoints as part of handling this service request.
 	endpointJSONRPCResponses []endpointJSONRPCResponse
 }
@@ -130,6 +130,11 @@ func (brc batchJSONRPCRequestContext) GetHTTPResponse() pathhttp.HTTPResponse {
 	}
 }
 
+// TODO_IMPROVE(@adshmh): Track the method field of each request in a JSONRPC batch request:
+// - Update proto/path/qos/solana.proto to include request details in each endpoint observation.
+// - Map each request in a batch to its corresponding response: needs gateway.QoSRequestContext interface update to handle slice of response.
+// - Update the endpoint observation building code below to include details of the corresponding request.
+//
 // GetObservations returns all the observations contained in the request context.
 // Implements the gateway.RequestQoSContext interface.
 func (rc batchJSONRPCRequestContext) GetObservations() qosobservations.Observations {
@@ -154,8 +159,32 @@ func (rc batchJSONRPCRequestContext) GetObservations() qosobservations.Observati
 		}
 	}
 
-	// TODO_UPNEXT(@adshmh): Report batch JSONRPC requests endpoint observations via metrics.
-	//
+	// Add one endpoint observation per request in the JSONRPC batch request.
+	endpointObservations := make([]*qosobservations.SolanaEndpointObservation, len(rc.endpointJSONRPCResponses))
+	for index, endpointResp := range rc.endpointJSONRPCResponses {
+		// TODO_TECHDEBT(@adshmh): Support method-specific JSONRPC responses on batch requests.
+		// This requires mapping each endpoint response to its corresponding request in the batch.
+		//
+		endpointObs := &qosobservations.SolanaEndpointObservation{
+			// TODO_DOCUMENT(@adshmh): Add a reference for the choice of HTTP status code on batch requests.
+			//
+			// HTTP status code 200 for batch requests.
+			HttpStatusCode: int32(http.StatusOK),
+			// Track response as an unrecognized response, since QoS does not currently use batch requests to evaluate endpoints.
+			ResponseObservation: &qosobservations.SolanaEndpointObservation_UnrecognizedResponse{
+				UnrecognizedResponse: &qosobservations.SolanaUnrecognizedResponse{
+					// Track details of the JSONRPC response: e.g. ID and a preview of result.
+					JsonrpcResponse: endpointResp.GetObservation(),
+				},
+			},
+		}
+
+		// Store in the list of endpoint observations.
+		endpointObservations[index] = endpointObs
+	}
+
+	observations.EndpointObservations = endpointObservations
+
 	return qosobservations.Observations{
 		ServiceObservations: &qosobservations.Observations_Solana{
 			Solana: observations,