diff --git a/backend/wireguard/stats.go b/backend/wireguard/stats.go index 4c3469d8..4bcf4a60 100644 --- a/backend/wireguard/stats.go +++ b/backend/wireguard/stats.go @@ -14,6 +14,23 @@ import ( const onlineActivityThreshold = 45 * time.Second +// The tracker preserves cumulative deltas across peer changes and soft restarts. +// A new backend instance receives a new epoch when its trackers are recreated. +func (wg *WireGuard) UsageSnapshot(ctx context.Context, kind common.StatType) (string, *common.StatResponse, error) { + wg.syncMu.Lock() + defer wg.syncMu.Unlock() + if kind == common.StatType_Outbounds { + rx, tx, err := wg.getInterfaceCounters() + if err != nil { + return "", nil, err + } + rx, tx = wg.interfaceStats.Cumulative(rx, tx) + return wg.usageEpoch, &common.StatResponse{Stats: stats.BuildInterfaceStats(wg.config.InterfaceName, "interface", rx, tx)}, nil + } + stats, err := wg.GetStats(ctx, &common.StatRequest{Type: kind, Reset_: false}) + return wg.usageEpoch, stats, err +} + func (wg *WireGuard) getInterfaceCounters() (int64, int64, error) { wg.mu.RLock() mgr := wg.manager diff --git a/backend/wireguard/wireguard.go b/backend/wireguard/wireguard.go index 403cf9c8..d8c08269 100644 --- a/backend/wireguard/wireguard.go +++ b/backend/wireguard/wireguard.go @@ -11,6 +11,8 @@ import ( "sync" "time" + "github.com/google/uuid" + "github.com/pasarguard/node/common" "github.com/pasarguard/node/config" "github.com/pasarguard/node/pkg/stats" @@ -65,6 +67,7 @@ type WireGuard struct { lastStatsErrAt time.Time newManager newManagerFunc hostRouting func() + usageEpoch string } // getWireGuardVersion fetches the wireguard-tools version @@ -128,6 +131,7 @@ func newWithManagerFactory(cfg *config.Config, wgConfig *Config, users []*common version := getWireGuardVersion() wg := &WireGuard{ + usageEpoch: uuid.NewString(), cancelFunc: wgCancel, cfg: cfg, statsTracker: stats.New(), diff --git a/backend/xray/stats.go b/backend/xray/stats.go index 0638e2e8..e5a65049 100644 --- a/backend/xray/stats.go +++ b/backend/xray/stats.go @@ -41,3 +41,12 @@ func (x *Xray) GetStats(ctx context.Context, request *common.StatRequest) (*comm return nil, errors.New("not implemented stat type") } } + +// UsageSnapshot holds the lifecycle lock across the read, binding cumulative +// counters to precisely one core generation, even during a health restart. +func (x *Xray) UsageSnapshot(ctx context.Context, kind common.StatType) (string, *common.StatResponse, error) { + x.mu.RLock() + defer x.mu.RUnlock() + stats, err := x.GetStats(ctx, &common.StatRequest{Type: kind, Reset_: false}) + return x.usageEpoch, stats, err +} diff --git a/backend/xray/xray.go b/backend/xray/xray.go index 205a611a..4e49f0bb 100644 --- a/backend/xray/xray.go +++ b/backend/xray/xray.go @@ -7,6 +7,8 @@ import ( "sync" "time" + "github.com/google/uuid" + "github.com/pasarguard/node/backend/xray/api" "github.com/pasarguard/node/common" "github.com/pasarguard/node/config" @@ -21,6 +23,7 @@ type Xray struct { cancelFunc context.CancelFunc mu sync.RWMutex syncMu sync.Mutex + usageEpoch string } func New(ctx context.Context, xrayConfig *Config, users []*common.User, apiPort, metricPort int, cfg *config.Config) (*Xray, error) { @@ -42,6 +45,7 @@ func New(ctx context.Context, xrayConfig *Config, users []*common.User, apiPort, xCtx, xCancel := context.WithCancel(context.Background()) xray := &Xray{ + usageEpoch: uuid.NewString(), cancelFunc: xCancel, cfg: cfg, metricPort: metricPort, @@ -132,6 +136,7 @@ func (x *Xray) restartCoreWithConfig(config *Config) error { if err := x.core.Restart(config, x.cfg.Debug); err != nil { return err } + x.usageEpoch = uuid.NewString() return nil } diff --git a/common/service.pb.go b/common/service.pb.go index a4b9a8d7..880062f5 100644 --- a/common/service.pb.go +++ b/common/service.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 -// protoc v7.34.1 +// protoc-gen-go v1.36.12 +// protoc v7.35.1 // source: common/service.proto package common @@ -516,6 +516,164 @@ func (x *StatRequest) GetType() StatType { return StatType_Outbounds } +// Accounting reads never reset core counters. Receipts survive until the panel +// acknowledges them after committing its own durable transaction. +type UsageRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Type StatType `protobuf:"varint,1,opt,name=type,proto3,enum=service.StatType" json:"type,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *UsageRequest) Reset() { + *x = UsageRequest{} + mi := &file_common_service_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *UsageRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*UsageRequest) ProtoMessage() {} + +func (x *UsageRequest) ProtoReflect() protoreflect.Message { + mi := &file_common_service_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use UsageRequest.ProtoReflect.Descriptor instead. +func (*UsageRequest) Descriptor() ([]byte, []int) { + return file_common_service_proto_rawDescGZIP(), []int{7} +} + +func (x *UsageRequest) GetType() StatType { + if x != nil { + return x.Type + } + return StatType_Outbounds +} + +type UsageReceipt struct { + state protoimpl.MessageState `protogen:"open.v1"` + ReceiptId string `protobuf:"bytes,1,opt,name=receipt_id,json=receiptId,proto3" json:"receipt_id,omitempty"` + CollectedAt int64 `protobuf:"varint,2,opt,name=collected_at,json=collectedAt,proto3" json:"collected_at,omitempty"` // Unix milliseconds + Stats []*Stat `protobuf:"bytes,3,rep,name=stats,proto3" json:"stats,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *UsageReceipt) Reset() { + *x = UsageReceipt{} + mi := &file_common_service_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *UsageReceipt) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*UsageReceipt) ProtoMessage() {} + +func (x *UsageReceipt) ProtoReflect() protoreflect.Message { + mi := &file_common_service_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use UsageReceipt.ProtoReflect.Descriptor instead. +func (*UsageReceipt) Descriptor() ([]byte, []int) { + return file_common_service_proto_rawDescGZIP(), []int{8} +} + +func (x *UsageReceipt) GetReceiptId() string { + if x != nil { + return x.ReceiptId + } + return "" +} + +func (x *UsageReceipt) GetCollectedAt() int64 { + if x != nil { + return x.CollectedAt + } + return 0 +} + +func (x *UsageReceipt) GetStats() []*Stat { + if x != nil { + return x.Stats + } + return nil +} + +type UsageAck struct { + state protoimpl.MessageState `protogen:"open.v1"` + Type StatType `protobuf:"varint,1,opt,name=type,proto3,enum=service.StatType" json:"type,omitempty"` + ReceiptId string `protobuf:"bytes,2,opt,name=receipt_id,json=receiptId,proto3" json:"receipt_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *UsageAck) Reset() { + *x = UsageAck{} + mi := &file_common_service_proto_msgTypes[9] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *UsageAck) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*UsageAck) ProtoMessage() {} + +func (x *UsageAck) ProtoReflect() protoreflect.Message { + mi := &file_common_service_proto_msgTypes[9] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use UsageAck.ProtoReflect.Descriptor instead. +func (*UsageAck) Descriptor() ([]byte, []int) { + return file_common_service_proto_rawDescGZIP(), []int{9} +} + +func (x *UsageAck) GetType() StatType { + if x != nil { + return x.Type + } + return StatType_Outbounds +} + +func (x *UsageAck) GetReceiptId() string { + if x != nil { + return x.ReceiptId + } + return "" +} + type OnlineStatResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` @@ -526,7 +684,7 @@ type OnlineStatResponse struct { func (x *OnlineStatResponse) Reset() { *x = OnlineStatResponse{} - mi := &file_common_service_proto_msgTypes[7] + mi := &file_common_service_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -538,7 +696,7 @@ func (x *OnlineStatResponse) String() string { func (*OnlineStatResponse) ProtoMessage() {} func (x *OnlineStatResponse) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[7] + mi := &file_common_service_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -551,7 +709,7 @@ func (x *OnlineStatResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use OnlineStatResponse.ProtoReflect.Descriptor instead. func (*OnlineStatResponse) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{7} + return file_common_service_proto_rawDescGZIP(), []int{10} } func (x *OnlineStatResponse) GetName() string { @@ -578,7 +736,7 @@ type StatsOnlineIpListResponse struct { func (x *StatsOnlineIpListResponse) Reset() { *x = StatsOnlineIpListResponse{} - mi := &file_common_service_proto_msgTypes[8] + mi := &file_common_service_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -590,7 +748,7 @@ func (x *StatsOnlineIpListResponse) String() string { func (*StatsOnlineIpListResponse) ProtoMessage() {} func (x *StatsOnlineIpListResponse) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[8] + mi := &file_common_service_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -603,7 +761,7 @@ func (x *StatsOnlineIpListResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use StatsOnlineIpListResponse.ProtoReflect.Descriptor instead. func (*StatsOnlineIpListResponse) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{8} + return file_common_service_proto_rawDescGZIP(), []int{11} } func (x *StatsOnlineIpListResponse) GetName() string { @@ -635,7 +793,7 @@ type Latency struct { func (x *Latency) Reset() { *x = Latency{} - mi := &file_common_service_proto_msgTypes[9] + mi := &file_common_service_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -647,7 +805,7 @@ func (x *Latency) String() string { func (*Latency) ProtoMessage() {} func (x *Latency) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[9] + mi := &file_common_service_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -660,7 +818,7 @@ func (x *Latency) ProtoReflect() protoreflect.Message { // Deprecated: Use Latency.ProtoReflect.Descriptor instead. func (*Latency) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{9} + return file_common_service_proto_rawDescGZIP(), []int{12} } func (x *Latency) GetName() string { @@ -721,7 +879,7 @@ type LatencyRequest struct { func (x *LatencyRequest) Reset() { *x = LatencyRequest{} - mi := &file_common_service_proto_msgTypes[10] + mi := &file_common_service_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -733,7 +891,7 @@ func (x *LatencyRequest) String() string { func (*LatencyRequest) ProtoMessage() {} func (x *LatencyRequest) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[10] + mi := &file_common_service_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -746,7 +904,7 @@ func (x *LatencyRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use LatencyRequest.ProtoReflect.Descriptor instead. func (*LatencyRequest) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{10} + return file_common_service_proto_rawDescGZIP(), []int{13} } func (x *LatencyRequest) GetName() string { @@ -765,7 +923,7 @@ type LatencyResponse struct { func (x *LatencyResponse) Reset() { *x = LatencyResponse{} - mi := &file_common_service_proto_msgTypes[11] + mi := &file_common_service_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -777,7 +935,7 @@ func (x *LatencyResponse) String() string { func (*LatencyResponse) ProtoMessage() {} func (x *LatencyResponse) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[11] + mi := &file_common_service_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -790,7 +948,7 @@ func (x *LatencyResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use LatencyResponse.ProtoReflect.Descriptor instead. func (*LatencyResponse) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{11} + return file_common_service_proto_rawDescGZIP(), []int{14} } func (x *LatencyResponse) GetLatencies() []*Latency { @@ -818,7 +976,7 @@ type BackendStatsResponse struct { func (x *BackendStatsResponse) Reset() { *x = BackendStatsResponse{} - mi := &file_common_service_proto_msgTypes[12] + mi := &file_common_service_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -830,7 +988,7 @@ func (x *BackendStatsResponse) String() string { func (*BackendStatsResponse) ProtoMessage() {} func (x *BackendStatsResponse) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[12] + mi := &file_common_service_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -843,7 +1001,7 @@ func (x *BackendStatsResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use BackendStatsResponse.ProtoReflect.Descriptor instead. func (*BackendStatsResponse) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{12} + return file_common_service_proto_rawDescGZIP(), []int{15} } func (x *BackendStatsResponse) GetNumGoroutine() uint32 { @@ -931,7 +1089,7 @@ type SystemStatsResponse struct { func (x *SystemStatsResponse) Reset() { *x = SystemStatsResponse{} - mi := &file_common_service_proto_msgTypes[13] + mi := &file_common_service_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -943,7 +1101,7 @@ func (x *SystemStatsResponse) String() string { func (*SystemStatsResponse) ProtoMessage() {} func (x *SystemStatsResponse) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[13] + mi := &file_common_service_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -956,7 +1114,7 @@ func (x *SystemStatsResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use SystemStatsResponse.ProtoReflect.Descriptor instead. func (*SystemStatsResponse) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{13} + return file_common_service_proto_rawDescGZIP(), []int{16} } func (x *SystemStatsResponse) GetMemTotal() uint64 { @@ -1018,7 +1176,7 @@ type Vmess struct { func (x *Vmess) Reset() { *x = Vmess{} - mi := &file_common_service_proto_msgTypes[14] + mi := &file_common_service_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1030,7 +1188,7 @@ func (x *Vmess) String() string { func (*Vmess) ProtoMessage() {} func (x *Vmess) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[14] + mi := &file_common_service_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1043,7 +1201,7 @@ func (x *Vmess) ProtoReflect() protoreflect.Message { // Deprecated: Use Vmess.ProtoReflect.Descriptor instead. func (*Vmess) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{14} + return file_common_service_proto_rawDescGZIP(), []int{17} } func (x *Vmess) GetId() string { @@ -1063,7 +1221,7 @@ type Vless struct { func (x *Vless) Reset() { *x = Vless{} - mi := &file_common_service_proto_msgTypes[15] + mi := &file_common_service_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1075,7 +1233,7 @@ func (x *Vless) String() string { func (*Vless) ProtoMessage() {} func (x *Vless) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[15] + mi := &file_common_service_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1088,7 +1246,7 @@ func (x *Vless) ProtoReflect() protoreflect.Message { // Deprecated: Use Vless.ProtoReflect.Descriptor instead. func (*Vless) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{15} + return file_common_service_proto_rawDescGZIP(), []int{18} } func (x *Vless) GetId() string { @@ -1114,7 +1272,7 @@ type Trojan struct { func (x *Trojan) Reset() { *x = Trojan{} - mi := &file_common_service_proto_msgTypes[16] + mi := &file_common_service_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1126,7 +1284,7 @@ func (x *Trojan) String() string { func (*Trojan) ProtoMessage() {} func (x *Trojan) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[16] + mi := &file_common_service_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1139,7 +1297,7 @@ func (x *Trojan) ProtoReflect() protoreflect.Message { // Deprecated: Use Trojan.ProtoReflect.Descriptor instead. func (*Trojan) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{16} + return file_common_service_proto_rawDescGZIP(), []int{19} } func (x *Trojan) GetPassword() string { @@ -1159,7 +1317,7 @@ type Shadowsocks struct { func (x *Shadowsocks) Reset() { *x = Shadowsocks{} - mi := &file_common_service_proto_msgTypes[17] + mi := &file_common_service_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1171,7 +1329,7 @@ func (x *Shadowsocks) String() string { func (*Shadowsocks) ProtoMessage() {} func (x *Shadowsocks) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[17] + mi := &file_common_service_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1184,7 +1342,7 @@ func (x *Shadowsocks) ProtoReflect() protoreflect.Message { // Deprecated: Use Shadowsocks.ProtoReflect.Descriptor instead. func (*Shadowsocks) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{17} + return file_common_service_proto_rawDescGZIP(), []int{20} } func (x *Shadowsocks) GetPassword() string { @@ -1211,7 +1369,7 @@ type Wireguard struct { func (x *Wireguard) Reset() { *x = Wireguard{} - mi := &file_common_service_proto_msgTypes[18] + mi := &file_common_service_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1223,7 +1381,7 @@ func (x *Wireguard) String() string { func (*Wireguard) ProtoMessage() {} func (x *Wireguard) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[18] + mi := &file_common_service_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1236,7 +1394,7 @@ func (x *Wireguard) ProtoReflect() protoreflect.Message { // Deprecated: Use Wireguard.ProtoReflect.Descriptor instead. func (*Wireguard) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{18} + return file_common_service_proto_rawDescGZIP(), []int{21} } func (x *Wireguard) GetPublicKey() string { @@ -1262,7 +1420,7 @@ type Hysteria struct { func (x *Hysteria) Reset() { *x = Hysteria{} - mi := &file_common_service_proto_msgTypes[19] + mi := &file_common_service_proto_msgTypes[22] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1274,7 +1432,7 @@ func (x *Hysteria) String() string { func (*Hysteria) ProtoMessage() {} func (x *Hysteria) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[19] + mi := &file_common_service_proto_msgTypes[22] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1287,7 +1445,7 @@ func (x *Hysteria) ProtoReflect() protoreflect.Message { // Deprecated: Use Hysteria.ProtoReflect.Descriptor instead. func (*Hysteria) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{19} + return file_common_service_proto_rawDescGZIP(), []int{22} } func (x *Hysteria) GetAuth() string { @@ -1311,7 +1469,7 @@ type Proxy struct { func (x *Proxy) Reset() { *x = Proxy{} - mi := &file_common_service_proto_msgTypes[20] + mi := &file_common_service_proto_msgTypes[23] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1323,7 +1481,7 @@ func (x *Proxy) String() string { func (*Proxy) ProtoMessage() {} func (x *Proxy) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[20] + mi := &file_common_service_proto_msgTypes[23] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1336,7 +1494,7 @@ func (x *Proxy) ProtoReflect() protoreflect.Message { // Deprecated: Use Proxy.ProtoReflect.Descriptor instead. func (*Proxy) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{20} + return file_common_service_proto_rawDescGZIP(), []int{23} } func (x *Proxy) GetVmess() *Vmess { @@ -1392,7 +1550,7 @@ type User struct { func (x *User) Reset() { *x = User{} - mi := &file_common_service_proto_msgTypes[21] + mi := &file_common_service_proto_msgTypes[24] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1404,7 +1562,7 @@ func (x *User) String() string { func (*User) ProtoMessage() {} func (x *User) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[21] + mi := &file_common_service_proto_msgTypes[24] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1417,7 +1575,7 @@ func (x *User) ProtoReflect() protoreflect.Message { // Deprecated: Use User.ProtoReflect.Descriptor instead. func (*User) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{21} + return file_common_service_proto_rawDescGZIP(), []int{24} } func (x *User) GetEmail() string { @@ -1450,7 +1608,7 @@ type Users struct { func (x *Users) Reset() { *x = Users{} - mi := &file_common_service_proto_msgTypes[22] + mi := &file_common_service_proto_msgTypes[25] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1462,7 +1620,7 @@ func (x *Users) String() string { func (*Users) ProtoMessage() {} func (x *Users) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[22] + mi := &file_common_service_proto_msgTypes[25] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1475,7 +1633,7 @@ func (x *Users) ProtoReflect() protoreflect.Message { // Deprecated: Use Users.ProtoReflect.Descriptor instead. func (*Users) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{22} + return file_common_service_proto_rawDescGZIP(), []int{25} } func (x *Users) GetUsers() []*User { @@ -1496,7 +1654,7 @@ type UsersChunk struct { func (x *UsersChunk) Reset() { *x = UsersChunk{} - mi := &file_common_service_proto_msgTypes[23] + mi := &file_common_service_proto_msgTypes[26] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1508,7 +1666,7 @@ func (x *UsersChunk) String() string { func (*UsersChunk) ProtoMessage() {} func (x *UsersChunk) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[23] + mi := &file_common_service_proto_msgTypes[26] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1521,7 +1679,7 @@ func (x *UsersChunk) ProtoReflect() protoreflect.Message { // Deprecated: Use UsersChunk.ProtoReflect.Descriptor instead. func (*UsersChunk) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{23} + return file_common_service_proto_rawDescGZIP(), []int{26} } func (x *UsersChunk) GetUsers() []*User { @@ -1556,7 +1714,7 @@ type RoutingRule struct { func (x *RoutingRule) Reset() { *x = RoutingRule{} - mi := &file_common_service_proto_msgTypes[24] + mi := &file_common_service_proto_msgTypes[27] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1568,7 +1726,7 @@ func (x *RoutingRule) String() string { func (*RoutingRule) ProtoMessage() {} func (x *RoutingRule) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[24] + mi := &file_common_service_proto_msgTypes[27] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1581,7 +1739,7 @@ func (x *RoutingRule) ProtoReflect() protoreflect.Message { // Deprecated: Use RoutingRule.ProtoReflect.Descriptor instead. func (*RoutingRule) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{24} + return file_common_service_proto_rawDescGZIP(), []int{27} } func (x *RoutingRule) GetOutboundTag() string { @@ -1607,7 +1765,7 @@ type RoutingRulesResponse struct { func (x *RoutingRulesResponse) Reset() { *x = RoutingRulesResponse{} - mi := &file_common_service_proto_msgTypes[25] + mi := &file_common_service_proto_msgTypes[28] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1619,7 +1777,7 @@ func (x *RoutingRulesResponse) String() string { func (*RoutingRulesResponse) ProtoMessage() {} func (x *RoutingRulesResponse) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[25] + mi := &file_common_service_proto_msgTypes[28] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1632,7 +1790,7 @@ func (x *RoutingRulesResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RoutingRulesResponse.ProtoReflect.Descriptor instead. func (*RoutingRulesResponse) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{25} + return file_common_service_proto_rawDescGZIP(), []int{28} } func (x *RoutingRulesResponse) GetRules() []*RoutingRule { @@ -1651,7 +1809,7 @@ type BalancerInfoRequest struct { func (x *BalancerInfoRequest) Reset() { *x = BalancerInfoRequest{} - mi := &file_common_service_proto_msgTypes[26] + mi := &file_common_service_proto_msgTypes[29] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1663,7 +1821,7 @@ func (x *BalancerInfoRequest) String() string { func (*BalancerInfoRequest) ProtoMessage() {} func (x *BalancerInfoRequest) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[26] + mi := &file_common_service_proto_msgTypes[29] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1676,7 +1834,7 @@ func (x *BalancerInfoRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use BalancerInfoRequest.ProtoReflect.Descriptor instead. func (*BalancerInfoRequest) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{26} + return file_common_service_proto_rawDescGZIP(), []int{29} } func (x *BalancerInfoRequest) GetTag() string { @@ -1696,7 +1854,7 @@ type BalancerInfoResponse struct { func (x *BalancerInfoResponse) Reset() { *x = BalancerInfoResponse{} - mi := &file_common_service_proto_msgTypes[27] + mi := &file_common_service_proto_msgTypes[30] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1708,7 +1866,7 @@ func (x *BalancerInfoResponse) String() string { func (*BalancerInfoResponse) ProtoMessage() {} func (x *BalancerInfoResponse) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[27] + mi := &file_common_service_proto_msgTypes[30] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1721,7 +1879,7 @@ func (x *BalancerInfoResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use BalancerInfoResponse.ProtoReflect.Descriptor instead. func (*BalancerInfoResponse) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{27} + return file_common_service_proto_rawDescGZIP(), []int{30} } func (x *BalancerInfoResponse) GetOverrideTarget() string { @@ -1756,7 +1914,7 @@ type TestRouteRequest struct { func (x *TestRouteRequest) Reset() { *x = TestRouteRequest{} - mi := &file_common_service_proto_msgTypes[28] + mi := &file_common_service_proto_msgTypes[31] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1768,7 +1926,7 @@ func (x *TestRouteRequest) String() string { func (*TestRouteRequest) ProtoMessage() {} func (x *TestRouteRequest) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[28] + mi := &file_common_service_proto_msgTypes[31] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1781,7 +1939,7 @@ func (x *TestRouteRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use TestRouteRequest.ProtoReflect.Descriptor instead. func (*TestRouteRequest) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{28} + return file_common_service_proto_rawDescGZIP(), []int{31} } func (x *TestRouteRequest) GetInboundTag() string { @@ -1867,7 +2025,7 @@ type RouteResult struct { func (x *RouteResult) Reset() { *x = RouteResult{} - mi := &file_common_service_proto_msgTypes[29] + mi := &file_common_service_proto_msgTypes[32] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1879,7 +2037,7 @@ func (x *RouteResult) String() string { func (*RouteResult) ProtoMessage() {} func (x *RouteResult) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[29] + mi := &file_common_service_proto_msgTypes[32] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1892,7 +2050,7 @@ func (x *RouteResult) ProtoReflect() protoreflect.Message { // Deprecated: Use RouteResult.ProtoReflect.Descriptor instead. func (*RouteResult) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{29} + return file_common_service_proto_rawDescGZIP(), []int{32} } func (x *RouteResult) GetOutboundTag() string { @@ -1943,7 +2101,7 @@ type AddRoutingRuleRequest struct { func (x *AddRoutingRuleRequest) Reset() { *x = AddRoutingRuleRequest{} - mi := &file_common_service_proto_msgTypes[30] + mi := &file_common_service_proto_msgTypes[33] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1955,7 +2113,7 @@ func (x *AddRoutingRuleRequest) String() string { func (*AddRoutingRuleRequest) ProtoMessage() {} func (x *AddRoutingRuleRequest) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[30] + mi := &file_common_service_proto_msgTypes[33] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1968,7 +2126,7 @@ func (x *AddRoutingRuleRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use AddRoutingRuleRequest.ProtoReflect.Descriptor instead. func (*AddRoutingRuleRequest) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{30} + return file_common_service_proto_rawDescGZIP(), []int{33} } func (x *AddRoutingRuleRequest) GetRule() string { @@ -1994,7 +2152,7 @@ type RemoveRoutingRuleRequest struct { func (x *RemoveRoutingRuleRequest) Reset() { *x = RemoveRoutingRuleRequest{} - mi := &file_common_service_proto_msgTypes[31] + mi := &file_common_service_proto_msgTypes[34] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2006,7 +2164,7 @@ func (x *RemoveRoutingRuleRequest) String() string { func (*RemoveRoutingRuleRequest) ProtoMessage() {} func (x *RemoveRoutingRuleRequest) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[31] + mi := &file_common_service_proto_msgTypes[34] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2019,7 +2177,7 @@ func (x *RemoveRoutingRuleRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RemoveRoutingRuleRequest.ProtoReflect.Descriptor instead. func (*RemoveRoutingRuleRequest) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{31} + return file_common_service_proto_rawDescGZIP(), []int{34} } func (x *RemoveRoutingRuleRequest) GetRuleTag() string { @@ -2039,7 +2197,7 @@ type OverrideBalancerTargetRequest struct { func (x *OverrideBalancerTargetRequest) Reset() { *x = OverrideBalancerTargetRequest{} - mi := &file_common_service_proto_msgTypes[32] + mi := &file_common_service_proto_msgTypes[35] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2051,7 +2209,7 @@ func (x *OverrideBalancerTargetRequest) String() string { func (*OverrideBalancerTargetRequest) ProtoMessage() {} func (x *OverrideBalancerTargetRequest) ProtoReflect() protoreflect.Message { - mi := &file_common_service_proto_msgTypes[32] + mi := &file_common_service_proto_msgTypes[35] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2064,7 +2222,7 @@ func (x *OverrideBalancerTargetRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use OverrideBalancerTargetRequest.ProtoReflect.Descriptor instead. func (*OverrideBalancerTargetRequest) Descriptor() ([]byte, []int) { - return file_common_service_proto_rawDescGZIP(), []int{32} + return file_common_service_proto_rawDescGZIP(), []int{35} } func (x *OverrideBalancerTargetRequest) GetBalancerTag() string { @@ -2110,7 +2268,18 @@ const file_common_service_proto_rawDesc = "" + "\vStatRequest\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x14\n" + "\x05reset\x18\x02 \x01(\bR\x05reset\x12%\n" + - "\x04type\x18\x03 \x01(\x0e2\x11.service.StatTypeR\x04type\">\n" + + "\x04type\x18\x03 \x01(\x0e2\x11.service.StatTypeR\x04type\"5\n" + + "\fUsageRequest\x12%\n" + + "\x04type\x18\x01 \x01(\x0e2\x11.service.StatTypeR\x04type\"u\n" + + "\fUsageReceipt\x12\x1d\n" + + "\n" + + "receipt_id\x18\x01 \x01(\tR\treceiptId\x12!\n" + + "\fcollected_at\x18\x02 \x01(\x03R\vcollectedAt\x12#\n" + + "\x05stats\x18\x03 \x03(\v2\r.service.StatR\x05stats\"P\n" + + "\bUsageAck\x12%\n" + + "\x04type\x18\x01 \x01(\x0e2\x11.service.StatTypeR\x04type\x12\x1d\n" + + "\n" + + "receipt_id\x18\x02 \x01(\tR\treceiptId\">\n" + "\x12OnlineStatResponse\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x14\n" + "\x05value\x18\x02 \x01(\x03R\x05value\"\xa6\x01\n" + @@ -2240,7 +2409,8 @@ const file_common_service_proto_rawDesc = "" + "\bInbounds\x10\x02\x12\v\n" + "\aInbound\x10\x03\x12\r\n" + "\tUsersStat\x10\x04\x12\f\n" + - "\bUserStat\x10\x052\xdc\t\n" + + "\bUserStat\x10\x052\xd5\n" + + "\n" + "\vNodeService\x126\n" + "\x05Start\x12\x10.service.Backend\x1a\x19.service.BaseInfoResponse\"\x00\x12(\n" + "\x04Stop\x12\x0e.service.Empty\x1a\x0e.service.Empty\"\x00\x12:\n" + @@ -2248,7 +2418,9 @@ const file_common_service_proto_rawDesc = "" + "\aGetLogs\x12\x0e.service.Empty\x1a\f.service.Log\"\x000\x01\x12@\n" + "\x0eGetSystemStats\x12\x0e.service.Empty\x1a\x1c.service.SystemStatsResponse\"\x00\x12B\n" + "\x0fGetBackendStats\x12\x0e.service.Empty\x1a\x1d.service.BackendStatsResponse\"\x00\x129\n" + - "\bGetStats\x12\x14.service.StatRequest\x1a\x15.service.StatResponse\"\x00\x12J\n" + + "\bGetStats\x12\x14.service.StatRequest\x1a\x15.service.StatResponse\"\x00\x12>\n" + + "\fCollectUsage\x12\x15.service.UsageRequest\x1a\x15.service.UsageReceipt\"\x00\x127\n" + + "\x10AcknowledgeUsage\x12\x11.service.UsageAck\x1a\x0e.service.Empty\"\x00\x12J\n" + "\x13GetOutboundsLatency\x12\x17.service.LatencyRequest\x1a\x18.service.LatencyResponse\"\x00\x12I\n" + "\x12GetUserOnlineStats\x12\x14.service.StatRequest\x1a\x1b.service.OnlineStatResponse\"\x00\x12V\n" + "\x18GetUserOnlineIpListStats\x12\x14.service.StatRequest\x1a\".service.StatsOnlineIpListResponse\"\x00\x12-\n" + @@ -2275,7 +2447,7 @@ func file_common_service_proto_rawDescGZIP() []byte { } var file_common_service_proto_enumTypes = make([]protoimpl.EnumInfo, 2) -var file_common_service_proto_msgTypes = make([]protoimpl.MessageInfo, 35) +var file_common_service_proto_msgTypes = make([]protoimpl.MessageInfo, 38) var file_common_service_proto_goTypes = []any{ (BackendType)(0), // 0: service.BackendType (StatType)(0), // 1: service.StatType @@ -2286,96 +2458,106 @@ var file_common_service_proto_goTypes = []any{ (*Stat)(nil), // 6: service.Stat (*StatResponse)(nil), // 7: service.StatResponse (*StatRequest)(nil), // 8: service.StatRequest - (*OnlineStatResponse)(nil), // 9: service.OnlineStatResponse - (*StatsOnlineIpListResponse)(nil), // 10: service.StatsOnlineIpListResponse - (*Latency)(nil), // 11: service.Latency - (*LatencyRequest)(nil), // 12: service.LatencyRequest - (*LatencyResponse)(nil), // 13: service.LatencyResponse - (*BackendStatsResponse)(nil), // 14: service.BackendStatsResponse - (*SystemStatsResponse)(nil), // 15: service.SystemStatsResponse - (*Vmess)(nil), // 16: service.Vmess - (*Vless)(nil), // 17: service.Vless - (*Trojan)(nil), // 18: service.Trojan - (*Shadowsocks)(nil), // 19: service.Shadowsocks - (*Wireguard)(nil), // 20: service.Wireguard - (*Hysteria)(nil), // 21: service.Hysteria - (*Proxy)(nil), // 22: service.Proxy - (*User)(nil), // 23: service.User - (*Users)(nil), // 24: service.Users - (*UsersChunk)(nil), // 25: service.UsersChunk - (*RoutingRule)(nil), // 26: service.RoutingRule - (*RoutingRulesResponse)(nil), // 27: service.RoutingRulesResponse - (*BalancerInfoRequest)(nil), // 28: service.BalancerInfoRequest - (*BalancerInfoResponse)(nil), // 29: service.BalancerInfoResponse - (*TestRouteRequest)(nil), // 30: service.TestRouteRequest - (*RouteResult)(nil), // 31: service.RouteResult - (*AddRoutingRuleRequest)(nil), // 32: service.AddRoutingRuleRequest - (*RemoveRoutingRuleRequest)(nil), // 33: service.RemoveRoutingRuleRequest - (*OverrideBalancerTargetRequest)(nil), // 34: service.OverrideBalancerTargetRequest - nil, // 35: service.StatsOnlineIpListResponse.IpsEntry - nil, // 36: service.TestRouteRequest.AttributesEntry + (*UsageRequest)(nil), // 9: service.UsageRequest + (*UsageReceipt)(nil), // 10: service.UsageReceipt + (*UsageAck)(nil), // 11: service.UsageAck + (*OnlineStatResponse)(nil), // 12: service.OnlineStatResponse + (*StatsOnlineIpListResponse)(nil), // 13: service.StatsOnlineIpListResponse + (*Latency)(nil), // 14: service.Latency + (*LatencyRequest)(nil), // 15: service.LatencyRequest + (*LatencyResponse)(nil), // 16: service.LatencyResponse + (*BackendStatsResponse)(nil), // 17: service.BackendStatsResponse + (*SystemStatsResponse)(nil), // 18: service.SystemStatsResponse + (*Vmess)(nil), // 19: service.Vmess + (*Vless)(nil), // 20: service.Vless + (*Trojan)(nil), // 21: service.Trojan + (*Shadowsocks)(nil), // 22: service.Shadowsocks + (*Wireguard)(nil), // 23: service.Wireguard + (*Hysteria)(nil), // 24: service.Hysteria + (*Proxy)(nil), // 25: service.Proxy + (*User)(nil), // 26: service.User + (*Users)(nil), // 27: service.Users + (*UsersChunk)(nil), // 28: service.UsersChunk + (*RoutingRule)(nil), // 29: service.RoutingRule + (*RoutingRulesResponse)(nil), // 30: service.RoutingRulesResponse + (*BalancerInfoRequest)(nil), // 31: service.BalancerInfoRequest + (*BalancerInfoResponse)(nil), // 32: service.BalancerInfoResponse + (*TestRouteRequest)(nil), // 33: service.TestRouteRequest + (*RouteResult)(nil), // 34: service.RouteResult + (*AddRoutingRuleRequest)(nil), // 35: service.AddRoutingRuleRequest + (*RemoveRoutingRuleRequest)(nil), // 36: service.RemoveRoutingRuleRequest + (*OverrideBalancerTargetRequest)(nil), // 37: service.OverrideBalancerTargetRequest + nil, // 38: service.StatsOnlineIpListResponse.IpsEntry + nil, // 39: service.TestRouteRequest.AttributesEntry } var file_common_service_proto_depIdxs = []int32{ 0, // 0: service.Backend.type:type_name -> service.BackendType - 23, // 1: service.Backend.users:type_name -> service.User + 26, // 1: service.Backend.users:type_name -> service.User 6, // 2: service.StatResponse.stats:type_name -> service.Stat 1, // 3: service.StatRequest.type:type_name -> service.StatType - 35, // 4: service.StatsOnlineIpListResponse.ips:type_name -> service.StatsOnlineIpListResponse.IpsEntry - 11, // 5: service.LatencyResponse.latencies:type_name -> service.Latency - 16, // 6: service.Proxy.vmess:type_name -> service.Vmess - 17, // 7: service.Proxy.vless:type_name -> service.Vless - 18, // 8: service.Proxy.trojan:type_name -> service.Trojan - 19, // 9: service.Proxy.shadowsocks:type_name -> service.Shadowsocks - 20, // 10: service.Proxy.wireguard:type_name -> service.Wireguard - 21, // 11: service.Proxy.hysteria:type_name -> service.Hysteria - 22, // 12: service.User.proxies:type_name -> service.Proxy - 23, // 13: service.Users.users:type_name -> service.User - 23, // 14: service.UsersChunk.users:type_name -> service.User - 26, // 15: service.RoutingRulesResponse.rules:type_name -> service.RoutingRule - 36, // 16: service.TestRouteRequest.attributes:type_name -> service.TestRouteRequest.AttributesEntry - 4, // 17: service.NodeService.Start:input_type -> service.Backend - 2, // 18: service.NodeService.Stop:input_type -> service.Empty - 2, // 19: service.NodeService.GetBaseInfo:input_type -> service.Empty - 2, // 20: service.NodeService.GetLogs:input_type -> service.Empty - 2, // 21: service.NodeService.GetSystemStats:input_type -> service.Empty - 2, // 22: service.NodeService.GetBackendStats:input_type -> service.Empty - 8, // 23: service.NodeService.GetStats:input_type -> service.StatRequest - 12, // 24: service.NodeService.GetOutboundsLatency:input_type -> service.LatencyRequest - 8, // 25: service.NodeService.GetUserOnlineStats:input_type -> service.StatRequest - 8, // 26: service.NodeService.GetUserOnlineIpListStats:input_type -> service.StatRequest - 23, // 27: service.NodeService.SyncUser:input_type -> service.User - 24, // 28: service.NodeService.SyncUsers:input_type -> service.Users - 25, // 29: service.NodeService.SyncUsersChunked:input_type -> service.UsersChunk - 2, // 30: service.NodeService.ListRoutingRules:input_type -> service.Empty - 28, // 31: service.NodeService.GetBalancerInfo:input_type -> service.BalancerInfoRequest - 30, // 32: service.NodeService.TestRoute:input_type -> service.TestRouteRequest - 32, // 33: service.NodeService.AddRoutingRule:input_type -> service.AddRoutingRuleRequest - 33, // 34: service.NodeService.RemoveRoutingRule:input_type -> service.RemoveRoutingRuleRequest - 34, // 35: service.NodeService.OverrideBalancerTarget:input_type -> service.OverrideBalancerTargetRequest - 3, // 36: service.NodeService.Start:output_type -> service.BaseInfoResponse - 2, // 37: service.NodeService.Stop:output_type -> service.Empty - 3, // 38: service.NodeService.GetBaseInfo:output_type -> service.BaseInfoResponse - 5, // 39: service.NodeService.GetLogs:output_type -> service.Log - 15, // 40: service.NodeService.GetSystemStats:output_type -> service.SystemStatsResponse - 14, // 41: service.NodeService.GetBackendStats:output_type -> service.BackendStatsResponse - 7, // 42: service.NodeService.GetStats:output_type -> service.StatResponse - 13, // 43: service.NodeService.GetOutboundsLatency:output_type -> service.LatencyResponse - 9, // 44: service.NodeService.GetUserOnlineStats:output_type -> service.OnlineStatResponse - 10, // 45: service.NodeService.GetUserOnlineIpListStats:output_type -> service.StatsOnlineIpListResponse - 2, // 46: service.NodeService.SyncUser:output_type -> service.Empty - 2, // 47: service.NodeService.SyncUsers:output_type -> service.Empty - 2, // 48: service.NodeService.SyncUsersChunked:output_type -> service.Empty - 27, // 49: service.NodeService.ListRoutingRules:output_type -> service.RoutingRulesResponse - 29, // 50: service.NodeService.GetBalancerInfo:output_type -> service.BalancerInfoResponse - 31, // 51: service.NodeService.TestRoute:output_type -> service.RouteResult - 2, // 52: service.NodeService.AddRoutingRule:output_type -> service.Empty - 2, // 53: service.NodeService.RemoveRoutingRule:output_type -> service.Empty - 2, // 54: service.NodeService.OverrideBalancerTarget:output_type -> service.Empty - 36, // [36:55] is the sub-list for method output_type - 17, // [17:36] is the sub-list for method input_type - 17, // [17:17] is the sub-list for extension type_name - 17, // [17:17] is the sub-list for extension extendee - 0, // [0:17] is the sub-list for field type_name + 1, // 4: service.UsageRequest.type:type_name -> service.StatType + 6, // 5: service.UsageReceipt.stats:type_name -> service.Stat + 1, // 6: service.UsageAck.type:type_name -> service.StatType + 38, // 7: service.StatsOnlineIpListResponse.ips:type_name -> service.StatsOnlineIpListResponse.IpsEntry + 14, // 8: service.LatencyResponse.latencies:type_name -> service.Latency + 19, // 9: service.Proxy.vmess:type_name -> service.Vmess + 20, // 10: service.Proxy.vless:type_name -> service.Vless + 21, // 11: service.Proxy.trojan:type_name -> service.Trojan + 22, // 12: service.Proxy.shadowsocks:type_name -> service.Shadowsocks + 23, // 13: service.Proxy.wireguard:type_name -> service.Wireguard + 24, // 14: service.Proxy.hysteria:type_name -> service.Hysteria + 25, // 15: service.User.proxies:type_name -> service.Proxy + 26, // 16: service.Users.users:type_name -> service.User + 26, // 17: service.UsersChunk.users:type_name -> service.User + 29, // 18: service.RoutingRulesResponse.rules:type_name -> service.RoutingRule + 39, // 19: service.TestRouteRequest.attributes:type_name -> service.TestRouteRequest.AttributesEntry + 4, // 20: service.NodeService.Start:input_type -> service.Backend + 2, // 21: service.NodeService.Stop:input_type -> service.Empty + 2, // 22: service.NodeService.GetBaseInfo:input_type -> service.Empty + 2, // 23: service.NodeService.GetLogs:input_type -> service.Empty + 2, // 24: service.NodeService.GetSystemStats:input_type -> service.Empty + 2, // 25: service.NodeService.GetBackendStats:input_type -> service.Empty + 8, // 26: service.NodeService.GetStats:input_type -> service.StatRequest + 9, // 27: service.NodeService.CollectUsage:input_type -> service.UsageRequest + 11, // 28: service.NodeService.AcknowledgeUsage:input_type -> service.UsageAck + 15, // 29: service.NodeService.GetOutboundsLatency:input_type -> service.LatencyRequest + 8, // 30: service.NodeService.GetUserOnlineStats:input_type -> service.StatRequest + 8, // 31: service.NodeService.GetUserOnlineIpListStats:input_type -> service.StatRequest + 26, // 32: service.NodeService.SyncUser:input_type -> service.User + 27, // 33: service.NodeService.SyncUsers:input_type -> service.Users + 28, // 34: service.NodeService.SyncUsersChunked:input_type -> service.UsersChunk + 2, // 35: service.NodeService.ListRoutingRules:input_type -> service.Empty + 31, // 36: service.NodeService.GetBalancerInfo:input_type -> service.BalancerInfoRequest + 33, // 37: service.NodeService.TestRoute:input_type -> service.TestRouteRequest + 35, // 38: service.NodeService.AddRoutingRule:input_type -> service.AddRoutingRuleRequest + 36, // 39: service.NodeService.RemoveRoutingRule:input_type -> service.RemoveRoutingRuleRequest + 37, // 40: service.NodeService.OverrideBalancerTarget:input_type -> service.OverrideBalancerTargetRequest + 3, // 41: service.NodeService.Start:output_type -> service.BaseInfoResponse + 2, // 42: service.NodeService.Stop:output_type -> service.Empty + 3, // 43: service.NodeService.GetBaseInfo:output_type -> service.BaseInfoResponse + 5, // 44: service.NodeService.GetLogs:output_type -> service.Log + 18, // 45: service.NodeService.GetSystemStats:output_type -> service.SystemStatsResponse + 17, // 46: service.NodeService.GetBackendStats:output_type -> service.BackendStatsResponse + 7, // 47: service.NodeService.GetStats:output_type -> service.StatResponse + 10, // 48: service.NodeService.CollectUsage:output_type -> service.UsageReceipt + 2, // 49: service.NodeService.AcknowledgeUsage:output_type -> service.Empty + 16, // 50: service.NodeService.GetOutboundsLatency:output_type -> service.LatencyResponse + 12, // 51: service.NodeService.GetUserOnlineStats:output_type -> service.OnlineStatResponse + 13, // 52: service.NodeService.GetUserOnlineIpListStats:output_type -> service.StatsOnlineIpListResponse + 2, // 53: service.NodeService.SyncUser:output_type -> service.Empty + 2, // 54: service.NodeService.SyncUsers:output_type -> service.Empty + 2, // 55: service.NodeService.SyncUsersChunked:output_type -> service.Empty + 30, // 56: service.NodeService.ListRoutingRules:output_type -> service.RoutingRulesResponse + 32, // 57: service.NodeService.GetBalancerInfo:output_type -> service.BalancerInfoResponse + 34, // 58: service.NodeService.TestRoute:output_type -> service.RouteResult + 2, // 59: service.NodeService.AddRoutingRule:output_type -> service.Empty + 2, // 60: service.NodeService.RemoveRoutingRule:output_type -> service.Empty + 2, // 61: service.NodeService.OverrideBalancerTarget:output_type -> service.Empty + 41, // [41:62] is the sub-list for method output_type + 20, // [20:41] is the sub-list for method input_type + 20, // [20:20] is the sub-list for extension type_name + 20, // [20:20] is the sub-list for extension extendee + 0, // [0:20] is the sub-list for field type_name } func init() { file_common_service_proto_init() } @@ -2389,7 +2571,7 @@ func file_common_service_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_common_service_proto_rawDesc), len(file_common_service_proto_rawDesc)), NumEnums: 2, - NumMessages: 35, + NumMessages: 38, NumExtensions: 0, NumServices: 1, }, diff --git a/common/service.proto b/common/service.proto index 486396e0..8dd3cfe5 100644 --- a/common/service.proto +++ b/common/service.proto @@ -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; @@ -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) {} diff --git a/common/service_grpc.pb.go b/common/service_grpc.pb.go index 86ea1c57..138e34c6 100644 --- a/common/service_grpc.pb.go +++ b/common/service_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.1 -// - protoc v7.34.1 +// - protoc v7.35.1 // source: common/service.proto package common @@ -26,6 +26,8 @@ const ( NodeService_GetSystemStats_FullMethodName = "/service.NodeService/GetSystemStats" NodeService_GetBackendStats_FullMethodName = "/service.NodeService/GetBackendStats" NodeService_GetStats_FullMethodName = "/service.NodeService/GetStats" + NodeService_CollectUsage_FullMethodName = "/service.NodeService/CollectUsage" + NodeService_AcknowledgeUsage_FullMethodName = "/service.NodeService/AcknowledgeUsage" NodeService_GetOutboundsLatency_FullMethodName = "/service.NodeService/GetOutboundsLatency" NodeService_GetUserOnlineStats_FullMethodName = "/service.NodeService/GetUserOnlineStats" NodeService_GetUserOnlineIpListStats_FullMethodName = "/service.NodeService/GetUserOnlineIpListStats" @@ -53,6 +55,8 @@ type NodeServiceClient interface { GetSystemStats(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*SystemStatsResponse, error) GetBackendStats(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*BackendStatsResponse, error) GetStats(ctx context.Context, in *StatRequest, opts ...grpc.CallOption) (*StatResponse, error) + CollectUsage(ctx context.Context, in *UsageRequest, opts ...grpc.CallOption) (*UsageReceipt, error) + AcknowledgeUsage(ctx context.Context, in *UsageAck, opts ...grpc.CallOption) (*Empty, error) GetOutboundsLatency(ctx context.Context, in *LatencyRequest, opts ...grpc.CallOption) (*LatencyResponse, error) GetUserOnlineStats(ctx context.Context, in *StatRequest, opts ...grpc.CallOption) (*OnlineStatResponse, error) GetUserOnlineIpListStats(ctx context.Context, in *StatRequest, opts ...grpc.CallOption) (*StatsOnlineIpListResponse, error) @@ -154,6 +158,26 @@ func (c *nodeServiceClient) GetStats(ctx context.Context, in *StatRequest, opts return out, nil } +func (c *nodeServiceClient) CollectUsage(ctx context.Context, in *UsageRequest, opts ...grpc.CallOption) (*UsageReceipt, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(UsageReceipt) + err := c.cc.Invoke(ctx, NodeService_CollectUsage_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *nodeServiceClient) AcknowledgeUsage(ctx context.Context, in *UsageAck, opts ...grpc.CallOption) (*Empty, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(Empty) + err := c.cc.Invoke(ctx, NodeService_AcknowledgeUsage_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *nodeServiceClient) GetOutboundsLatency(ctx context.Context, in *LatencyRequest, opts ...grpc.CallOption) (*LatencyResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(LatencyResponse) @@ -293,6 +317,8 @@ type NodeServiceServer interface { GetSystemStats(context.Context, *Empty) (*SystemStatsResponse, error) GetBackendStats(context.Context, *Empty) (*BackendStatsResponse, error) GetStats(context.Context, *StatRequest) (*StatResponse, error) + CollectUsage(context.Context, *UsageRequest) (*UsageReceipt, error) + AcknowledgeUsage(context.Context, *UsageAck) (*Empty, error) GetOutboundsLatency(context.Context, *LatencyRequest) (*LatencyResponse, error) GetUserOnlineStats(context.Context, *StatRequest) (*OnlineStatResponse, error) GetUserOnlineIpListStats(context.Context, *StatRequest) (*StatsOnlineIpListResponse, error) @@ -336,6 +362,12 @@ func (UnimplementedNodeServiceServer) GetBackendStats(context.Context, *Empty) ( func (UnimplementedNodeServiceServer) GetStats(context.Context, *StatRequest) (*StatResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetStats not implemented") } +func (UnimplementedNodeServiceServer) CollectUsage(context.Context, *UsageRequest) (*UsageReceipt, error) { + return nil, status.Error(codes.Unimplemented, "method CollectUsage not implemented") +} +func (UnimplementedNodeServiceServer) AcknowledgeUsage(context.Context, *UsageAck) (*Empty, error) { + return nil, status.Error(codes.Unimplemented, "method AcknowledgeUsage not implemented") +} func (UnimplementedNodeServiceServer) GetOutboundsLatency(context.Context, *LatencyRequest) (*LatencyResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetOutboundsLatency not implemented") } @@ -512,6 +544,42 @@ func _NodeService_GetStats_Handler(srv interface{}, ctx context.Context, dec fun return interceptor(ctx, in, info, handler) } +func _NodeService_CollectUsage_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(UsageRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(NodeServiceServer).CollectUsage(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: NodeService_CollectUsage_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(NodeServiceServer).CollectUsage(ctx, req.(*UsageRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _NodeService_AcknowledgeUsage_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(UsageAck) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(NodeServiceServer).AcknowledgeUsage(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: NodeService_AcknowledgeUsage_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(NodeServiceServer).AcknowledgeUsage(ctx, req.(*UsageAck)) + } + return interceptor(ctx, in, info, handler) +} + func _NodeService_GetOutboundsLatency_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(LatencyRequest) if err := dec(in); err != nil { @@ -737,6 +805,14 @@ var NodeService_ServiceDesc = grpc.ServiceDesc{ MethodName: "GetStats", Handler: _NodeService_GetStats_Handler, }, + { + MethodName: "CollectUsage", + Handler: _NodeService_CollectUsage_Handler, + }, + { + MethodName: "AcknowledgeUsage", + Handler: _NodeService_AcknowledgeUsage_Handler, + }, { MethodName: "GetOutboundsLatency", Handler: _NodeService_GetOutboundsLatency_Handler, diff --git a/config/config.go b/config/config.go index 6dd5be84..a30f4221 100644 --- a/config/config.go +++ b/config/config.go @@ -3,6 +3,7 @@ package config import ( "log" "os" + "path/filepath" "regexp" "strconv" @@ -21,6 +22,7 @@ type Config struct { ServiceProtocol string Debug bool GeneratedConfigPath string + UsageJournalPath string LogBufferSize int StartupLogTailSize int StatsUpdateIntervalSeconds int @@ -48,6 +50,7 @@ func Load() (*Config, error) { SslCertFile: GetEnv("SSL_CERT_FILE", "/var/lib/pg-node/certs/ssl_cert.pem"), SslKeyFile: GetEnv("SSL_KEY_FILE", "/var/lib/pg-node/certs/ssl_key.pem"), GeneratedConfigPath: GetEnv("GENERATED_CONFIG_PATH", "/var/lib/pg-node/generated/"), + UsageJournalPath: GetEnv("USAGE_JOURNAL_PATH", "/var/lib/pg-node/usage/receipts.db"), ServiceProtocol: GetEnv("SERVICE_PROTOCOL", "grpc"), Debug: GetEnvAsBool("DEBUG", false), LogBufferSize: GetEnvAsInt("LOG_BUFFER_SIZE", 10000), @@ -91,6 +94,7 @@ func Load() (*Config, error) { func NewTestConfig(generatedConfigPath string, key uuid.UUID) *Config { cfg, _ := Load() cfg.GeneratedConfigPath = generatedConfigPath + cfg.UsageJournalPath = filepath.Join(generatedConfigPath, "usage-test.db") cfg.ApiKey = key return cfg } diff --git a/controller/controller.go b/controller/controller.go index 904b36af..7a55279e 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -17,6 +17,7 @@ import ( "github.com/pasarguard/node/config" "github.com/pasarguard/node/pkg/netutil" "github.com/pasarguard/node/pkg/sysstats" + "github.com/pasarguard/node/pkg/usage" ) const NodeVersion = "0.5.4" @@ -40,6 +41,7 @@ type Controller struct { cancelFunc context.CancelFunc mu sync.RWMutex controlMu sync.Mutex + usageStore *usage.Store } func New(cfg *config.Config) *Controller { @@ -49,6 +51,7 @@ func New(cfg *config.Config) *Controller { apiPort: netutil.FindFreePort(), metricPort: netutil.FindFreePort(), cancelFunc: cancel, + usageStore: &usage.Store{Path: cfg.UsageJournalPath}, } } diff --git a/controller/rest/service.go b/controller/rest/service.go index fc9aa6db..d10556a9 100644 --- a/controller/rest/service.go +++ b/controller/rest/service.go @@ -33,6 +33,9 @@ func (s *Service) setRouter() { router.Post("/start", s.Start) router.Get("/info", s.Base) + // Receipt replay and ACK remain available while the backend is stopped. + router.Post("/usage/collect", s.CollectUsage) + router.Post("/usage/ack", s.AcknowledgeUsage) router.Group(func(private chi.Router) { private.Use(s.checkBackendMiddleware) diff --git a/controller/rest/stats.go b/controller/rest/stats.go index 916b9afd..e4880db2 100644 --- a/controller/rest/stats.go +++ b/controller/rest/stats.go @@ -14,7 +14,7 @@ func (s *Service) GetStats(w http.ResponseWriter, r *http.Request) { return } - stats, err := s.Backend().GetStats(r.Context(), &request) + stats, err := s.Controller.GetStats(r.Context(), &request) if err != nil { err = common.InterceptNotFound(err) st, _ := status.FromError(err) diff --git a/controller/rest/usage.go b/controller/rest/usage.go new file mode 100644 index 00000000..d9c421ce --- /dev/null +++ b/controller/rest/usage.go @@ -0,0 +1,38 @@ +package rest + +import ( + "net/http" + + "github.com/pasarguard/node/common" + "google.golang.org/grpc/status" +) + +func (s *Service) CollectUsage(w http.ResponseWriter, r *http.Request) { + var request common.UsageRequest + r.Body = http.MaxBytesReader(w, r.Body, 1024) + if err := common.ReadProtoBody(r.Body, &request); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + receipt, err := s.Controller.CollectUsage(r.Context(), &request) + if err != nil { + http.Error(w, err.Error(), common.GrpcCodeToHTTP(status.Code(err))) + return + } + common.SendProtoResponse(w, receipt) +} + +func (s *Service) AcknowledgeUsage(w http.ResponseWriter, r *http.Request) { + var request common.UsageAck + r.Body = http.MaxBytesReader(w, r.Body, 1024) + if err := common.ReadProtoBody(r.Body, &request); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + response, err := s.Controller.AcknowledgeUsage(r.Context(), &request) + if err != nil { + http.Error(w, err.Error(), common.GrpcCodeToHTTP(status.Code(err))) + return + } + common.SendProtoResponse(w, response) +} diff --git a/controller/rpc/stats.go b/controller/rpc/stats.go index 082cd8a1..b84f1e3b 100644 --- a/controller/rpc/stats.go +++ b/controller/rpc/stats.go @@ -18,12 +18,7 @@ func (s *Service) backend() (backend.Backend, error) { } func (s *Service) GetStats(ctx context.Context, request *common.StatRequest) (*common.StatResponse, error) { - backend, err := s.backend() - if err != nil { - return nil, err - } - - stats, err := backend.GetStats(ctx, request) + stats, err := s.Controller.GetStats(ctx, request) if err != nil { err = common.InterceptNotFound(err) return nil, err @@ -31,6 +26,14 @@ func (s *Service) GetStats(ctx context.Context, request *common.StatRequest) (*c return stats, nil } +func (s *Service) CollectUsage(ctx context.Context, request *common.UsageRequest) (*common.UsageReceipt, error) { + return s.Controller.CollectUsage(ctx, request) +} + +func (s *Service) AcknowledgeUsage(ctx context.Context, request *common.UsageAck) (*common.Empty, error) { + return s.Controller.AcknowledgeUsage(ctx, request) +} + func (s *Service) GetUserOnlineStats(ctx context.Context, request *common.StatRequest) (*common.OnlineStatResponse, error) { backend, err := s.backend() if err != nil { diff --git a/controller/usage.go b/controller/usage.go new file mode 100644 index 00000000..ff8e2742 --- /dev/null +++ b/controller/usage.go @@ -0,0 +1,67 @@ +package controller + +import ( + "context" + + "github.com/pasarguard/node/common" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type usageBackend interface { + UsageSnapshot(context.Context, common.StatType) (string, *common.StatResponse, error) +} + +func (c *Controller) CollectUsage(ctx context.Context, request *common.UsageRequest) (*common.UsageReceipt, error) { + if c.usageStore == nil { + return nil, status.Error(codes.FailedPrecondition, "usage journal unavailable") + } + // Replay does not require a running core. Only a new snapshot enters this + // callback. A durable pending receipt remains retrievable after core failure. + return c.usageStore.Collect(ctx, request.GetType(), func(ctx context.Context) (string, *common.StatResponse, error) { + c.controlMu.Lock() + defer c.controlMu.Unlock() + back, ok := c.Backend().(usageBackend) + if !ok { + return "", nil, status.Error(codes.Unavailable, "backend does not support cumulative usage snapshots") + } + return back.UsageSnapshot(ctx, request.GetType()) + }) +} + +func (c *Controller) AcknowledgeUsage(ctx context.Context, request *common.UsageAck) (*common.Empty, error) { + if c.usageStore == nil { + return nil, status.Error(codes.FailedPrecondition, "usage journal unavailable") + } + if err := c.usageStore.Acknowledge(ctx, request.GetType(), request.GetReceiptId()); err != nil { + return nil, err + } + return &common.Empty{}, nil +} + +// GetStats preserves the legacy API until receipt accounting is activated. +func (c *Controller) GetStats(ctx context.Context, request *common.StatRequest) (*common.StatResponse, error) { + read := func() (*common.StatResponse, error) { + back := c.Backend() + if back == nil { + return nil, status.Error(codes.Unavailable, "backend not initialized") + } + return back.GetStats(ctx, request) + } + if !request.GetReset_() { + return read() + } + kind := request.GetType() + switch kind { + case common.StatType_UserStat: + kind = common.StatType_UsersStat + case common.StatType_Outbound: + kind = common.StatType_Outbounds + case common.StatType_Inbound, common.StatType_Inbounds: + return read() + } + if c.usageStore == nil { + return nil, status.Error(codes.FailedPrecondition, "usage journal unavailable") + } + return c.usageStore.LegacyReset(kind, read) +} diff --git a/go.mod b/go.mod index a208fada..a688204b 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/shirou/gopsutil/v4 v4.26.8 github.com/vishvananda/netlink v1.3.1 github.com/xtls/xray-core v1.260327.0 + go.etcd.io/bbolt v1.4.3 golang.org/x/sys v0.48.0 golang.zx2c4.com/wireguard/wgctrl v0.0.0-20241231184526-a9ab2273dd10 google.golang.org/grpc v1.84.0 diff --git a/go.sum b/go.sum index 7b74c057..b597efbb 100644 --- a/go.sum +++ b/go.sum @@ -131,6 +131,8 @@ github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.etcd.io/bbolt v1.4.3 h1:dEadXpI6G79deX5prL3QRNP6JB8UxVkqo4UPnHaNXJo= +go.etcd.io/bbolt v1.4.3/go.mod h1:tKQlpPaYCVFctUIgFKFnAlvbmB3tpy1vkTnDWohtc0E= go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A= go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko= diff --git a/pkg/stats/interface_counters.go b/pkg/stats/interface_counters.go index 4bdadb8f..b37b0ddb 100644 --- a/pkg/stats/interface_counters.go +++ b/pkg/stats/interface_counters.go @@ -13,6 +13,40 @@ type InterfaceCountersTracker struct { baseRx int64 baseTx int64 baseSet bool + + usageSet bool + usageLastRx int64 + usageLastTx int64 + usageTotalRx int64 + usageTotalTx int64 +} + +// Cumulative preserves observed traffic across interface counter resets without +// changing the legacy reset baseline. Activation includes only traffic since +// the last legacy reset, avoiding double billing during protocol migration. +func (t *InterfaceCountersTracker) Cumulative(rx, tx int64) (int64, int64) { + t.mu.Lock() + defer t.mu.Unlock() + if !t.usageSet { + t.usageSet = true + if t.baseSet { + t.usageTotalRx = max(0, rx-t.baseRx) + t.usageTotalTx = max(0, tx-t.baseTx) + } + } else { + if rx >= t.usageLastRx { + t.usageTotalRx += rx - t.usageLastRx + } else { + t.usageTotalRx += max(0, rx) + } + if tx >= t.usageLastTx { + t.usageTotalTx += tx - t.usageLastTx + } else { + t.usageTotalTx += max(0, tx) + } + } + t.usageLastRx, t.usageLastTx = rx, tx + return t.usageTotalRx, t.usageTotalTx } func NewInterfaceCountersTracker() *InterfaceCountersTracker { diff --git a/pkg/usage/store.go b/pkg/usage/store.go new file mode 100644 index 00000000..883f433f --- /dev/null +++ b/pkg/usage/store.go @@ -0,0 +1,259 @@ +// Package usage persists non-destructive counter snapshots and replayable receipts. +package usage + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "math" + "os" + "path/filepath" + "runtime" + "time" + + "github.com/google/uuid" + "github.com/pasarguard/node/common" + bolt "go.etcd.io/bbolt" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +// Store opens the database per operation, so locks are released on shutdown or +// process death. bbolt serializes even independent processes using the same file. +// Its default synchronous commits must never be replaced by NoSync writes. +type Store struct{ Path string } + +// Read-only replays must not rewrite bbolt metadata or fsync the journal. +var errUnchanged = errors.New("usage journal unchanged") + +type Snapshot func(context.Context) (epoch string, stats *common.StatResponse, err error) + +type streamState struct { + Version int `json:"version"` + Epoch string `json:"epoch"` + Baseline map[string]int64 `json:"baseline"` + Pending []byte `json:"pending,omitempty"` +} + +func streamKey(kind common.StatType) ([]byte, error) { + switch kind { + case common.StatType_UsersStat: + return []byte("users"), nil + case common.StatType_Outbounds: + return []byte("outbounds"), nil + default: + return nil, status.Error(codes.InvalidArgument, "usage requires UsersStat or Outbounds") + } +} + +func (s *Store) update(fn func(*bolt.Tx) error) error { + if s.Path == "" { + return status.Error(codes.FailedPrecondition, "usage storage path is not configured") + } + if err := os.MkdirAll(filepath.Dir(s.Path), 0700); err != nil { + return fmt.Errorf("create usage directory: %w", err) + } + db, err := bolt.Open(s.Path, 0600, &bolt.Options{Timeout: time.Second}) + if err != nil { + return fmt.Errorf("open usage journal: %w", err) + } + defer db.Close() + err = db.Update(func(tx *bolt.Tx) error { + // Persist the new journal's directory entry before publishing any receipt. + // bbolt synchronizes file contents; directory creation needs its own sync. + // Retry a failed directory sync: existence of the file alone is not proof + // that a previous operation reached the durable activation commit. + if tx.Bucket([]byte("usage-v1")) == nil && runtime.GOOS != "windows" { + for dir := filepath.Dir(s.Path); ; dir = filepath.Dir(dir) { + f, err := os.Open(dir) + if err != nil { + return err + } + err = f.Sync() + f.Close() + if err != nil { + return err + } + if filepath.Dir(dir) == dir { + break + } + } + } + return fn(tx) + }) + if errors.Is(err, errUnchanged) { + return nil + } + return err +} + +func load(bucket *bolt.Bucket, key []byte) (*streamState, error) { + state := &streamState{Version: 1, Baseline: make(map[string]int64)} + if data := bucket.Get(key); data != nil { + if err := json.Unmarshal(data, state); err != nil { + return nil, fmt.Errorf("decode usage journal: %w", err) + } + if state.Version != 1 || state.Baseline == nil { + return nil, status.Error(codes.FailedPrecondition, "unsupported usage journal") + } + } + return state, nil +} + +func save(bucket *bolt.Bucket, key []byte, state *streamState) error { + data, err := json.Marshal(state) + if err != nil { + return err + } + if bytes.Equal(bucket.Get(key), data) { + return errUnchanged + } + return bucket.Put(key, data) +} + +// Collect publishes a receipt only after its payload and cumulative baseline +// commit together. A failed read/write leaves counters untouched. Until ACK, +// all callers receive exactly the same receipt, including after a node restart. +func (s *Store) Collect(ctx context.Context, kind common.StatType, snapshot Snapshot) (*common.UsageReceipt, error) { + key, err := streamKey(kind) + if err != nil { + return nil, err + } + var receipt *common.UsageReceipt + err = s.update(func(tx *bolt.Tx) error { + if err := ctx.Err(); err != nil { + return err + } + bucket, err := tx.CreateBucketIfNotExists([]byte("usage-v1")) + if err != nil { + return err + } + state, err := load(bucket, key) + if err != nil { + return err + } + if len(state.Pending) != 0 { + receipt = new(common.UsageReceipt) + if err := proto.Unmarshal(state.Pending, receipt); err != nil { + return err + } + return errUnchanged + } + epoch, counters, err := snapshot(ctx) + if err != nil { + return err + } + if epoch == "" || counters == nil { + return status.Error(codes.FailedPrecondition, "backend has no stable counter epoch") + } + if state.Epoch != epoch { + state.Epoch = epoch + state.Baseline = make(map[string]int64) + } + receipt = &common.UsageReceipt{ReceiptId: uuid.NewString(), CollectedAt: time.Now().UTC().UnixMilli()} + // Preserve absent counters in the baseline: a temporarily omitted counter + // must not be charged from zero if it reappears. Epochs fence core resets. + totals := make(map[string]int64) + rows := make(map[string]*common.Stat) + for _, counter := range counters.GetStats() { + if counter == nil || (counter.GetType() != "uplink" && counter.GetType() != "downlink") { + continue + } + encoded, _ := json.Marshal([3]string{counter.GetName(), counter.GetType(), counter.GetLink()}) + name := string(encoded) + value := counter.GetValue() + if value < 0 || totals[name] > math.MaxInt64-value { + return status.Error(codes.FailedPrecondition, "invalid cumulative usage counter") + } + totals[name] += value + rows[name] = counter + } + for name, value := range totals { + previous := state.Baseline[name] + if value < previous { + return status.Error(codes.FailedPrecondition, "usage counter decreased without a new epoch") + } + if value > previous { + counter := proto.Clone(rows[name]).(*common.Stat) + counter.Value = value - previous + receipt.Stats = append(receipt.Stats, counter) + } + state.Baseline[name] = value + } + if len(receipt.Stats) == 0 { + // Empty polls still persist activation/baselines but need no ACK and + // must not create an ever-growing panel ledger of zero-usage rows. + receipt.ReceiptId = "" + } else { + state.Pending, err = proto.Marshal(receipt) + if err != nil { + return err + } + } + return save(bucket, key, state) + }) + if err != nil { + return nil, err + } + return receipt, nil +} + +// Acknowledge is idempotent. An ACK for an older receipt can never remove the +// current receipt. Baselines and activation survive ACK and process restarts. +func (s *Store) Acknowledge(ctx context.Context, kind common.StatType, id string) error { + key, err := streamKey(kind) + if err != nil { + return err + } + if _, err := uuid.Parse(id); err != nil { + return status.Error(codes.InvalidArgument, "invalid usage receipt ID") + } + return s.update(func(tx *bolt.Tx) error { + if err := ctx.Err(); err != nil { + return err + } + bucket := tx.Bucket([]byte("usage-v1")) + if bucket == nil || bucket.Get(key) == nil { + return status.Error(codes.FailedPrecondition, "usage stream not initialized") + } + state, err := load(bucket, key) + if err != nil { + return err + } + if len(state.Pending) == 0 { + return errUnchanged + } + receipt := new(common.UsageReceipt) + if err := proto.Unmarshal(state.Pending, receipt); err != nil { + return err + } + if receipt.GetReceiptId() != id { + return errUnchanged + } + state.Pending = nil + return save(bucket, key, state) + }) +} + +// LegacyReset serializes activation with old destructive reads. Once a stream +// is activated, no old client may reset its counters, including after restart. +func (s *Store) LegacyReset(kind common.StatType, read func() (*common.StatResponse, error)) (*common.StatResponse, error) { + key, err := streamKey(kind) + if err != nil { + return nil, err + } + var result *common.StatResponse + err = s.update(func(tx *bolt.Tx) error { + bucket := tx.Bucket([]byte("usage-v1")) + if bucket != nil && bucket.Get(key) != nil { + return status.Error(codes.FailedPrecondition, "usage receipts enabled; destructive stats reads are disabled") + } + var err error + result, err = read() + return err + }) + return result, err +}