feat: PRT - provider load forwarded to provider optimizer

protocol/chainlib/common.go

@@ -30,11 +30,13 @@ const (
 	relayMsgLogMaxChars        = 200
 	RPCProviderNodeAddressHash = "Lava-Provider-Node-Address-Hash"
 	RPCProviderNodeExtension   = "Lava-Provider-Node-Extension"
+	RpcProviderLoadRateHeader  = "Lava-Provider-Load-Rate"
 	RpcProviderUniqueIdHeader  = "Lava-Provider-Unique-Id"
 	WebSocketExtension         = "websocket"
 )
 
 var (
+	TrailersToAddToHeaderResponse      = []string{RPCProviderNodeExtension, RpcProviderLoadRateHeader}
 	InvalidResponses                   = []string{"null", "", "nil", "undefined"}
 	FailedSendingSubscriptionToClients = sdkerrors.New("failed Sending Subscription To Clients", 1015, "Failed Sending Subscription To Clients connection might have been closed by the user")
 	NoActiveSubscriptionFound          = sdkerrors.New("failed finding an active subscription on provider side", 1016, "no active subscriptions for hashed params.")

protocol/chainlib/grpc.go

@@ -128,7 +128,7 @@ func (apip *GrpcChainParser) ParseMsg(url string, data []byte, connectionType st
 	// Check API is supported and save it in nodeMsg.
 	apiCont, err := apip.getSupportedApi(url, connectionType)
 	if err != nil {
-		return nil, utils.LavaFormatError("failed to getSupportedApi gRPC", err)
+		return nil, utils.LavaFormatError("failed to getSupportedApi gRPC", err, utils.LogAttr("url", url))
 	}
 
 	apiCollection, err := apip.getApiCollection(connectionType, apiCont.collectionKey.InternalPath, apiCont.collectionKey.Addon)

protocol/common/cobra_common.go

@@ -46,6 +46,7 @@ const (
 	// websocket flags
 	RateLimitWebSocketFlag                       = "rate-limit-websocket-requests-per-connection"
 	BanDurationForWebsocketRateLimitExceededFlag = "ban-duration-for-websocket-rate-limit-exceeded"
+	RateLimitRequestPerSecondFlag                = "rate-limit-requests-per-second"
 )
 
 const (

protocol/integration/protocol_test.go

@@ -344,7 +344,7 @@ func createRpcProvider(t *testing.T, ctx context.Context, rpcProviderOptions rpc
 	chainTracker.StartAndServe(ctx)
 	reliabilityManager := reliabilitymanager.NewReliabilityManager(chainTracker, &mockProviderStateTracker, rpcProviderOptions.account.Addr.String(), chainRouter, chainParser)
 	mockReliabilityManager := NewMockReliabilityManager(reliabilityManager)
-	rpcProviderServer.ServeRPCRequests(ctx, rpcProviderEndpoint, chainParser, rws, providerSessionManager, mockReliabilityManager, rpcProviderOptions.account.SK, cache, chainRouter, &mockProviderStateTracker, rpcProviderOptions.account.Addr, rpcProviderOptions.lavaChainID, rpcprovider.DEFAULT_ALLOWED_MISSING_CU, nil, nil, nil, false)
+	rpcProviderServer.ServeRPCRequests(ctx, rpcProviderEndpoint, chainParser, rws, providerSessionManager, mockReliabilityManager, rpcProviderOptions.account.SK, cache, chainRouter, &mockProviderStateTracker, rpcProviderOptions.account.Addr, rpcProviderOptions.lavaChainID, rpcprovider.DEFAULT_ALLOWED_MISSING_CU, nil, nil, nil, false, nil)
 	listener := rpcprovider.NewProviderListener(ctx, rpcProviderEndpoint.NetworkAddress, "/health")
 	err = listener.RegisterReceiver(rpcProviderServer, rpcProviderEndpoint)
 	require.NoError(t, err)

protocol/lavasession/consumer_session_manager.go

@@ -959,7 +959,8 @@ func (csm *ConsumerSessionManager) OnSessionFailure(consumerSession *SingleConsu
 	}
 	cuToDecrease := consumerSession.LatestRelayCu
 	// latency, isHangingApi, syncScore aren't updated when there is a failure
-	go csm.providerOptimizer.AppendRelayFailure(consumerSession.Parent.PublicLavaAddress)
+
+	go csm.providerOptimizer.AppendRelayFailure(consumerSession.Parent.PublicLavaAddress, consumerSession.GetProviderLoad())
 	consumerSession.LatestRelayCu = 0 // making sure no one uses it in a wrong way
 	consecutiveErrors := uint64(len(consumerSession.ConsecutiveErrors))
 	parentConsumerSessionsWithProvider := consumerSession.Parent // must read this pointer before unlocking
@@ -1042,7 +1043,8 @@ func (csm *ConsumerSessionManager) OnSessionDone(
 	consumerSession.LatestBlock = latestServicedBlock // update latest serviced block
 	// calculate QoS
 	consumerSession.CalculateQoS(currentLatency, expectedLatency, expectedBH-latestServicedBlock, numOfProviders, int64(providersCount))
-	go csm.providerOptimizer.AppendRelayData(consumerSession.Parent.PublicLavaAddress, currentLatency, isHangingApi, specComputeUnits, uint64(latestServicedBlock))
+
+	go csm.providerOptimizer.AppendRelayData(consumerSession.Parent.PublicLavaAddress, currentLatency, isHangingApi, specComputeUnits, uint64(latestServicedBlock), consumerSession.GetProviderLoad())
 	csm.updateMetricsManager(consumerSession, currentLatency, !isHangingApi) // apply latency only for non hanging apis
 	return nil
 }

protocol/lavasession/consumer_types.go

@@ -71,8 +71,8 @@ type ConsumerSessionsMap map[string]*SessionInfo
 
 type ProviderOptimizer interface {
 	AppendProbeRelayData(providerAddress string, latency time.Duration, success bool)
-	AppendRelayFailure(providerAddress string)
-	AppendRelayData(providerAddress string, latency time.Duration, isHangingApi bool, cu, syncBlock uint64)
+	AppendRelayFailure(providerAddress string, providerLoad *provideroptimizer.ProviderLoadReport)
+	AppendRelayData(providerAddress string, latency time.Duration, isHangingApi bool, cu, syncBlock uint64, providerLoad *provideroptimizer.ProviderLoadReport)
 	ChooseProvider(allAddresses []string, ignoredProviders map[string]struct{}, cu uint64, requestedBlock int64) (addresses []string, tier int)
 	GetExcellenceQoSReportForProvider(string) (*pairingtypes.QualityOfServiceReport, *pairingtypes.QualityOfServiceReport)
 	Strategy() provideroptimizer.Strategy

protocol/lavasession/single_consumer_session.go

@@ -7,6 +7,7 @@
 	"time"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/lavanet/lava/v4/protocol/provideroptimizer"
 	"github.com/lavanet/lava/v4/utils"
 	pairingtypes "github.com/lavanet/lava/v4/x/pairing/types"
 )
@@ -21,13 +22,49 @@
 	RelayNum      uint64
 	LatestBlock   int64
 	// Each session will holds a pointer to a connection, if the connection is lost, this session will be banned (wont be picked)
-	EndpointConnection *EndpointConnection
-	BlockListed        bool // if session lost sync we blacklist it.
-	ConsecutiveErrors  []error
-	errorsCount        uint64
-	relayProcessor     UsedProvidersInf
-	providerUniqueId   string
-	StaticProvider     bool
+	EndpointConnection    *EndpointConnection
+	BlockListed           bool // if session lost sync we blacklist it.
+	ConsecutiveErrors     []error
+	errorsCount           uint64
+	relayProcessor        UsedProvidersInf
+	providerUniqueId      string
+	StaticProvider        bool
+	latestKnownLoadReport *provideroptimizer.ProviderLoadReport
+}
+
+// should only be called when locked, returning a copy of the object
+func (cs *SingleConsumerSession) GetProviderLoad() *provideroptimizer.ProviderLoadReport {
+	// create new provider load pointer so we can read it later without locks
+	var providerLoadReport *provideroptimizer.ProviderLoadReport
+	if cs.latestKnownLoadReport != nil {
+		providerLoadReport = &provideroptimizer.ProviderLoadReport{
+			ProviderLoad: cs.latestKnownLoadReport.ProviderLoad,
+			TimeStamp:    cs.latestKnownLoadReport.TimeStamp,
+		}
+	}
+	return providerLoadReport
+}
+
+// should only be called when locked.
+func (cs *SingleConsumerSession) SetLoadReport(loadReport []string) {
+	if len(loadReport) <= 0 {
+		// no load report
+		return
+	}
+	load := loadReport[0]
+	floatLoad, err := strconv.ParseFloat(load, 64)
+	if err != nil {
+		utils.LavaFormatWarning("Failed parsing load report from provider", err, utils.LogAttr("load_reported", loadReport))
+		return
+	}
+	if floatLoad == 0 {
+		// Provider did not set his max load options or has 0 load.
+		return
+	}
+	cs.latestKnownLoadReport = &provideroptimizer.ProviderLoadReport{
+		TimeStamp:    time.Now(),
+		ProviderLoad: floatLoad,
+	}
 }
 
 // returns the expected latency to a threshold.

protocol/metrics/provider_metrics.go

@@ -22,6 +22,7 @@ type ProviderMetrics struct {
 	totalRelaysServicedMetric *prometheus.CounterVec
 	totalErroredMetric        *prometheus.CounterVec
 	consumerQoSMetric         *prometheus.GaugeVec
+	loadRateMetric            *prometheus.GaugeVec
 }
 
 func (pm *ProviderMetrics) AddRelay(consumerAddress string, cu uint64, qos *pairingtypes.QualityOfServiceReport) {
@@ -49,6 +50,13 @@ func (pm *ProviderMetrics) AddRelay(consumerAddress string, cu uint64, qos *pair
 	}
 }
 
+func (pm *ProviderMetrics) SetLoadRate(loadRate float64) {
+	if pm == nil {
+		return
+	}
+	pm.loadRateMetric.WithLabelValues(pm.specID).Set(loadRate)
+}
+
 func (pm *ProviderMetrics) AddPayment(cu uint64) {
 	if pm == nil {
 		return
@@ -72,6 +80,7 @@ func NewProviderMetrics(specID, apiInterface string, totalCUServicedMetric *prom
 	totalRelaysServicedMetric *prometheus.CounterVec,
 	totalErroredMetric *prometheus.CounterVec,
 	consumerQoSMetric *prometheus.GaugeVec,
+	loadRateMetric *prometheus.GaugeVec,
 ) *ProviderMetrics {
 	pm := &ProviderMetrics{
 		specID:                    specID,
@@ -82,6 +91,7 @@ func NewProviderMetrics(specID, apiInterface string, totalCUServicedMetric *prom
 		totalRelaysServicedMetric: totalRelaysServicedMetric,
 		totalErroredMetric:        totalErroredMetric,
 		consumerQoSMetric:         consumerQoSMetric,
+		loadRateMetric:            loadRateMetric,
 	}
 	return pm
 }

protocol/metrics/provider_metrics_manager.go

@@ -41,6 +41,7 @@ type ProviderMetricsManager struct {
 	endpointsHealthChecksOk       uint64
 	relaysMonitors                map[string]*RelaysMonitor
 	relaysMonitorsLock            sync.RWMutex
+	loadRateMetric                *prometheus.GaugeVec
 }
 
 func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
@@ -107,6 +108,11 @@ func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
 		Help: "The total number of get latest block queries that succeeded by chainfetcher",
 	}, []string{"spec"})
 
+	loadRateMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "lava_provider_load_rate",
+		Help: "The load rate according to the load rate limit - Given Y simultaneous relay calls, a value of X  and will measure Y/X load rate.",
+	}, []string{"spec"})
+
 	fetchBlockSuccessMetric := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "lava_provider_fetch_block_success",
 		Help: "The total number of get specific block queries that succeeded by chainfetcher",
@@ -141,6 +147,7 @@ func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
 	prometheus.MustRegister(virtualEpochMetric)
 	prometheus.MustRegister(endpointsHealthChecksOkMetric)
 	prometheus.MustRegister(protocolVersionMetric)
+	prometheus.MustRegister(loadRateMetric)
 
 	providerMetricsManager := &ProviderMetricsManager{
 		providerMetrics:               map[string]*ProviderMetrics{},
@@ -161,6 +168,7 @@ func NewProviderMetricsManager(networkAddress string) *ProviderMetricsManager {
 		endpointsHealthChecksOk:       1,
 		protocolVersionMetric:         protocolVersionMetric,
 		relaysMonitors:                map[string]*RelaysMonitor{},
+		loadRateMetric:                loadRateMetric,
 	}
 
 	http.Handle("/metrics", promhttp.Handler())
@@ -209,7 +217,7 @@ func (pme *ProviderMetricsManager) AddProviderMetrics(specID, apiInterface strin
 	}
 
 	if pme.getProviderMetric(specID, apiInterface) == nil {
-		providerMetric := NewProviderMetrics(specID, apiInterface, pme.totalCUServicedMetric, pme.totalCUPaidMetric, pme.totalRelaysServicedMetric, pme.totalErroredMetric, pme.consumerQoSMetric)
+		providerMetric := NewProviderMetrics(specID, apiInterface, pme.totalCUServicedMetric, pme.totalCUPaidMetric, pme.totalRelaysServicedMetric, pme.totalErroredMetric, pme.consumerQoSMetric, pme.loadRateMetric)
 		pme.setProviderMetric(providerMetric)
 
 		endpoint := fmt.Sprintf("/metrics/%s/%s/health", specID, apiInterface)

protocol/provideroptimizer/provider_optimizer.go

@@ -71,13 +71,19 @@ type Exploration struct {
 	time    time.Time
 }
 
+type ProviderLoadReport struct {
+	ProviderLoad float64 // float describing load set by the provider can go above 1.0.
+	TimeStamp    time.Time
+}
+
 type ProviderData struct {
 	Availability score.ScoreStore // will be used to calculate the probability of error
 	Latency      score.ScoreStore // will be used to calculate the latency score
 	Sync         score.ScoreStore // will be used to calculate the sync score for spectypes.LATEST_BLOCK/spectypes.NOT_APPLICABLE requests
 	SyncBlock    uint64           // will be used to calculate the probability of block error
 	LatencyRaw   score.ScoreStore // will be used when reporting reputation to the node (Latency = LatencyRaw / baseLatency)
 	SyncRaw      score.ScoreStore // will be used when reporting reputation to the node (Sync = SyncRaw / baseSync)
+	ProviderLoad *ProviderLoadReport
 }
 
 type Strategy int
@@ -101,17 +107,22 @@ func (po *ProviderOptimizer) UpdateWeights(weights map[string]int64, epoch uint6
 	}
 }
 
-func (po *ProviderOptimizer) AppendRelayFailure(providerAddress string) {
-	po.appendRelayData(providerAddress, 0, false, false, 0, 0, time.Now())
+// TODO forward load also on relay failure
+func (po *ProviderOptimizer) AppendRelayFailure(providerAddress string, providerLoad *ProviderLoadReport) {
+	po.appendRelayData(providerAddress, 0, false, false, 0, 0, time.Now(), providerLoad)
 }
 
-func (po *ProviderOptimizer) AppendRelayData(providerAddress string, latency time.Duration, isHangingApi bool, cu, syncBlock uint64) {
-	po.appendRelayData(providerAddress, latency, isHangingApi, true, cu, syncBlock, time.Now())
+func (po *ProviderOptimizer) AppendRelayData(providerAddress string, latency time.Duration, isHangingApi bool, cu, syncBlock uint64, providerLoad *ProviderLoadReport) {
+	po.appendRelayData(providerAddress, latency, isHangingApi, true, cu, syncBlock, time.Now(), providerLoad)
 }
 
-func (po *ProviderOptimizer) appendRelayData(providerAddress string, latency time.Duration, isHangingApi, success bool, cu, syncBlock uint64, sampleTime time.Time) {
+func (po *ProviderOptimizer) appendRelayData(providerAddress string, latency time.Duration, isHangingApi, success bool, cu, syncBlock uint64, sampleTime time.Time, providerLoad *ProviderLoadReport) {
 	latestSync, timeSync := po.updateLatestSyncData(syncBlock, sampleTime)
 	providerData, _ := po.getProviderData(providerAddress)
+	// set current provider load only if incoming data is more fresh than previous stored data
+	if providerLoad != nil && (providerData.ProviderLoad == nil || providerLoad.TimeStamp.After(providerData.ProviderLoad.TimeStamp)) {
+		providerData.ProviderLoad = providerLoad
+	}
 	halfTime := po.calculateHalfTime(providerAddress, sampleTime)
 	providerData = po.updateProbeEntryAvailability(providerData, success, RELAY_UPDATE_WEIGHT, halfTime, sampleTime)
 	if success {