Skip to content

feat(usage): add durable receipt collection and ACK clients #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dr-hoseyn wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(usage): add durable receipt collection and ACK clients #22

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions PasarGuardNodeBridge/abstract_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,16 @@ async def get_stats(
) -> service.StatResponse | None:
raise NotImplementedError

async def collect_usage(self, stat_type: service.StatType, timeout: int | None = None) -> service.UsageReceipt:
"""Read a durable receipt without resetting counters; safe to repeat."""
raise NotImplementedError

async def acknowledge_usage(
self, stat_type: service.StatType, receipt_id: str, timeout: int | None = None
) -> service.Empty:
"""Release a receipt only after the caller durably stores it; safe to repeat."""
raise NotImplementedError

@abstractmethod
async def get_outbounds_latency(self, name: str = "", timeout: int | None = None) -> service.LatencyResponse | None:
raise NotImplementedError
Expand Down
19 changes: 19 additions & 0 deletions PasarGuardNodeBridge/common/service.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,23 @@ message StatRequest {
StatType type = 3;
}

// Accounting reads never reset core counters. Receipts survive until the panel
// acknowledges them after committing its own durable transaction.
message UsageRequest {
StatType type = 1;
}

message UsageReceipt {
string receipt_id = 1;
int64 collected_at = 2; // Unix milliseconds
repeated Stat stats = 3;
}

message UsageAck {
StatType type = 1;
string receipt_id = 2;
}

message OnlineStatResponse {
string name = 1;
int64 value = 2;
Expand Down Expand Up @@ -220,6 +237,8 @@ service NodeService {
rpc GetBackendStats(Empty) returns (BackendStatsResponse) {}

rpc GetStats(StatRequest) returns (StatResponse) {}
rpc CollectUsage(UsageRequest) returns (UsageReceipt) {}
rpc AcknowledgeUsage(UsageAck) returns (Empty) {}
rpc GetOutboundsLatency(LatencyRequest) returns (LatencyResponse) {}

rpc GetUserOnlineStats(StatRequest) returns (OnlineStatResponse) {}
Expand Down
32 changes: 32 additions & 0 deletions PasarGuardNodeBridge/common/service_grpc.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,14 @@ async def GetBackendStats(self, stream: 'grpclib.server.Stream[PasarGuardNodeBri
async def GetStats(self, stream: 'grpclib.server.Stream[PasarGuardNodeBridge.common.service_pb2.StatRequest, PasarGuardNodeBridge.common.service_pb2.StatResponse]') -> None:
pass

@abc.abstractmethod
async def CollectUsage(self, stream: 'grpclib.server.Stream[PasarGuardNodeBridge.common.service_pb2.UsageRequest, PasarGuardNodeBridge.common.service_pb2.UsageReceipt]') -> None:
pass

@abc.abstractmethod
async def AcknowledgeUsage(self, stream: 'grpclib.server.Stream[PasarGuardNodeBridge.common.service_pb2.UsageAck, PasarGuardNodeBridge.common.service_pb2.Empty]') -> None:
pass

@abc.abstractmethod
async def GetOutboundsLatency(self, stream: 'grpclib.server.Stream[PasarGuardNodeBridge.common.service_pb2.LatencyRequest, PasarGuardNodeBridge.common.service_pb2.LatencyResponse]') -> None:
pass
Expand Down Expand Up @@ -134,6 +142,18 @@ def __mapping__(self) -> typing.Dict[str, grpclib.const.Handler]:
PasarGuardNodeBridge.common.service_pb2.StatRequest,
PasarGuardNodeBridge.common.service_pb2.StatResponse,
),
'/service.NodeService/CollectUsage': grpclib.const.Handler(
self.CollectUsage,
grpclib.const.Cardinality.UNARY_UNARY,
PasarGuardNodeBridge.common.service_pb2.UsageRequest,
PasarGuardNodeBridge.common.service_pb2.UsageReceipt,
),
'/service.NodeService/AcknowledgeUsage': grpclib.const.Handler(
self.AcknowledgeUsage,
grpclib.const.Cardinality.UNARY_UNARY,
PasarGuardNodeBridge.common.service_pb2.UsageAck,
PasarGuardNodeBridge.common.service_pb2.Empty,
),
'/service.NodeService/GetOutboundsLatency': grpclib.const.Handler(
self.GetOutboundsLatency,
grpclib.const.Cardinality.UNARY_UNARY,
Expand Down Expand Up @@ -254,6 +274,18 @@ def __init__(self, channel: grpclib.client.Channel) -> None:
PasarGuardNodeBridge.common.service_pb2.StatRequest,
PasarGuardNodeBridge.common.service_pb2.StatResponse,
)
self.CollectUsage = grpclib.client.UnaryUnaryMethod(
channel,
'/service.NodeService/CollectUsage',
PasarGuardNodeBridge.common.service_pb2.UsageRequest,
PasarGuardNodeBridge.common.service_pb2.UsageReceipt,
)
self.AcknowledgeUsage = grpclib.client.UnaryUnaryMethod(
channel,
'/service.NodeService/AcknowledgeUsage',
PasarGuardNodeBridge.common.service_pb2.UsageAck,
PasarGuardNodeBridge.common.service_pb2.Empty,
)
self.GetOutboundsLatency = grpclib.client.UnaryUnaryMethod(
channel,
'/service.NodeService/GetOutboundsLatency',
Expand Down
Loading