|
@@ -1,7 +1,6 @@
|
|
|
package mtproto
|
|
package mtproto
|
|
|
|
|
|
|
|
import (
|
|
import (
|
|
|
- "bufio"
|
|
|
|
|
"context"
|
|
"context"
|
|
|
"encoding/json"
|
|
"encoding/json"
|
|
|
"fmt"
|
|
"fmt"
|
|
@@ -17,13 +16,23 @@ import (
|
|
|
"github.com/mhsanaei/3x-ui/v3/internal/logger"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/logger"
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
-// Instance is the desired runtime configuration of one mtproto inbound.
|
|
|
|
|
-type Instance struct {
|
|
|
|
|
- Id int
|
|
|
|
|
- Tag string
|
|
|
|
|
- Listen string
|
|
|
|
|
- Port int
|
|
|
|
|
|
|
+// SecretEntry is one named FakeTLS secret served by an mtg-multi process. Name is
|
|
|
|
|
+// the client email, used both as the [secrets] key and as the per-user key in the
|
|
|
|
|
+// /stats API so traffic can be attributed back to the client.
|
|
|
|
|
+type SecretEntry struct {
|
|
|
|
|
+ Name string
|
|
|
Secret string
|
|
Secret string
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// Instance is the desired runtime configuration of one mtproto inbound. A single
|
|
|
|
|
+// mtg-multi process serves every active client's secret through the [secrets]
|
|
|
|
|
+// section, so one inbound maps to one process with many named secrets.
|
|
|
|
|
+type Instance struct {
|
|
|
|
|
+ Id int
|
|
|
|
|
+ Tag string
|
|
|
|
|
+ Listen string
|
|
|
|
|
+ Port int
|
|
|
|
|
+ Secrets []SecretEntry
|
|
|
|
|
|
|
|
// Optional mtg tuning; each is omitted from the generated TOML when
|
|
// Optional mtg tuning; each is omitted from the generated TOML when
|
|
|
// zero-valued so mtg falls back to its own defaults.
|
|
// zero-valued so mtg falls back to its own defaults.
|
|
@@ -34,6 +43,10 @@ type Instance struct {
|
|
|
FrontingPort int
|
|
FrontingPort int
|
|
|
FrontingProxyProtocol bool
|
|
FrontingProxyProtocol bool
|
|
|
|
|
|
|
|
|
|
+ // ThrottleMaxConnections caps concurrent connections across all users with a
|
|
|
|
|
+ // fair-share algorithm; zero disables throttling.
|
|
|
|
|
+ ThrottleMaxConnections int
|
|
|
|
|
+
|
|
|
// When RouteThroughXray is set, mtg dials Telegram through the loopback
|
|
// When RouteThroughXray is set, mtg dials Telegram through the loopback
|
|
|
// SOCKS bridge the panel injects into the Xray config at XrayRoutePort, so
|
|
// SOCKS bridge the panel injects into the Xray config at XrayRoutePort, so
|
|
|
// the egress obeys the core's routing rules instead of going out directly.
|
|
// the egress obeys the core's routing rules instead of going out directly.
|
|
@@ -50,37 +63,47 @@ func (inst Instance) bindTo() string {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// fingerprint changes whenever any value that ends up in the generated TOML
|
|
// fingerprint changes whenever any value that ends up in the generated TOML
|
|
|
-// changes, so ensureLocked restarts mtg when the operator edits a setting.
|
|
|
|
|
|
|
+// changes, so ensureLocked restarts mtg when the operator edits a setting or a
|
|
|
|
|
+// client is added, removed, disabled, or re-keyed.
|
|
|
func (inst Instance) fingerprint() string {
|
|
func (inst Instance) fingerprint() string {
|
|
|
- return strings.Join([]string{
|
|
|
|
|
|
|
+ parts := []string{
|
|
|
inst.bindTo(),
|
|
inst.bindTo(),
|
|
|
- inst.Secret,
|
|
|
|
|
strconv.FormatBool(inst.Debug),
|
|
strconv.FormatBool(inst.Debug),
|
|
|
strconv.FormatBool(inst.ProxyProtocolListener),
|
|
strconv.FormatBool(inst.ProxyProtocolListener),
|
|
|
inst.PreferIP,
|
|
inst.PreferIP,
|
|
|
inst.FrontingIP,
|
|
inst.FrontingIP,
|
|
|
strconv.Itoa(inst.FrontingPort),
|
|
strconv.Itoa(inst.FrontingPort),
|
|
|
strconv.FormatBool(inst.FrontingProxyProtocol),
|
|
strconv.FormatBool(inst.FrontingProxyProtocol),
|
|
|
|
|
+ strconv.Itoa(inst.ThrottleMaxConnections),
|
|
|
strconv.FormatBool(inst.RouteThroughXray),
|
|
strconv.FormatBool(inst.RouteThroughXray),
|
|
|
strconv.Itoa(inst.XrayRoutePort),
|
|
strconv.Itoa(inst.XrayRoutePort),
|
|
|
- }, "|")
|
|
|
|
|
|
|
+ }
|
|
|
|
|
+ for _, e := range inst.Secrets {
|
|
|
|
|
+ parts = append(parts, e.Name+"="+e.Secret)
|
|
|
|
|
+ }
|
|
|
|
|
+ return strings.Join(parts, "|")
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-// Traffic is a per-inbound traffic delta scraped from an mtg metrics endpoint.
|
|
|
|
|
|
|
+// Traffic is a per-client traffic delta scraped from an mtg /stats endpoint. Tag
|
|
|
|
|
+// is the owning inbound's tag and Email is the client the bytes belong to.
|
|
|
type Traffic struct {
|
|
type Traffic struct {
|
|
|
- Tag string
|
|
|
|
|
- Up int64
|
|
|
|
|
- Down int64
|
|
|
|
|
|
|
+ Tag string
|
|
|
|
|
+ Email string
|
|
|
|
|
+ Up int64
|
|
|
|
|
+ Down int64
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+type clientCounters struct {
|
|
|
|
|
+ up int64
|
|
|
|
|
+ down int64
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
type managed struct {
|
|
type managed struct {
|
|
|
proc *Process
|
|
proc *Process
|
|
|
tag string
|
|
tag string
|
|
|
fingerprint string
|
|
fingerprint string
|
|
|
- metricsPort int
|
|
|
|
|
- lastUp int64
|
|
|
|
|
- lastDown int64
|
|
|
|
|
- haveLast bool
|
|
|
|
|
|
|
+ apiPort int
|
|
|
|
|
+ last map[string]clientCounters
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// Manager owns the set of running mtg processes keyed by inbound id.
|
|
// Manager owns the set of running mtg processes keyed by inbound id.
|
|
@@ -106,49 +129,61 @@ func GetManager() *Manager {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// InstanceFromInbound derives a desired Instance from an mtproto inbound,
|
|
// InstanceFromInbound derives a desired Instance from an mtproto inbound,
|
|
|
-// healing the FakeTLS secret so it always matches the configured domain.
|
|
|
|
|
-// Returns false when the inbound is not a usable mtproto inbound.
|
|
|
|
|
|
|
+// building one named secret per active client. Secrets are healed on save (see
|
|
|
|
|
+// normalizeMtprotoSecret) and by the migration, so they are read as-is here to
|
|
|
|
|
+// keep the fingerprint stable across reconciles. Returns false when the inbound
|
|
|
|
|
+// is not a usable mtproto inbound or has no active client secret to serve.
|
|
|
func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
|
|
func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
|
|
|
if ib == nil || ib.Protocol != model.MTProto {
|
|
if ib == nil || ib.Protocol != model.MTProto {
|
|
|
return Instance{}, false
|
|
return Instance{}, false
|
|
|
}
|
|
}
|
|
|
settings := ib.Settings
|
|
settings := ib.Settings
|
|
|
- if healed, ok := model.HealMtprotoSecret(settings); ok {
|
|
|
|
|
- settings = healed
|
|
|
|
|
- }
|
|
|
|
|
var parsed struct {
|
|
var parsed struct {
|
|
|
- Secret string `json:"secret"`
|
|
|
|
|
- Debug bool `json:"debug"`
|
|
|
|
|
- ProxyProtocolListener bool `json:"proxyProtocolListener"`
|
|
|
|
|
- PreferIP string `json:"preferIp"`
|
|
|
|
|
|
|
+ ProxyProtocolListener bool `json:"proxyProtocolListener"`
|
|
|
|
|
+ Debug bool `json:"debug"`
|
|
|
DomainFronting struct {
|
|
DomainFronting struct {
|
|
|
IP string `json:"ip"`
|
|
IP string `json:"ip"`
|
|
|
Port int `json:"port"`
|
|
Port int `json:"port"`
|
|
|
ProxyProtocol bool `json:"proxyProtocol"`
|
|
ProxyProtocol bool `json:"proxyProtocol"`
|
|
|
} `json:"domainFronting"`
|
|
} `json:"domainFronting"`
|
|
|
- RouteThroughXray bool `json:"routeThroughXray"`
|
|
|
|
|
- RouteXrayPort int `json:"routeXrayPort"`
|
|
|
|
|
|
|
+ PreferIP string `json:"preferIp"`
|
|
|
|
|
+ ThrottleMaxConnections int `json:"throttleMaxConnections"`
|
|
|
|
|
+ RouteThroughXray bool `json:"routeThroughXray"`
|
|
|
|
|
+ RouteXrayPort int `json:"routeXrayPort"`
|
|
|
|
|
+ Clients []struct {
|
|
|
|
|
+ Email string `json:"email"`
|
|
|
|
|
+ Secret string `json:"secret"`
|
|
|
|
|
+ Enable bool `json:"enable"`
|
|
|
|
|
+ } `json:"clients"`
|
|
|
}
|
|
}
|
|
|
if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
|
|
if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
|
|
|
return Instance{}, false
|
|
return Instance{}, false
|
|
|
}
|
|
}
|
|
|
- if parsed.Secret == "" {
|
|
|
|
|
|
|
+ secrets := make([]SecretEntry, 0, len(parsed.Clients))
|
|
|
|
|
+ for _, c := range parsed.Clients {
|
|
|
|
|
+ if !c.Enable || c.Secret == "" || c.Email == "" {
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ secrets = append(secrets, SecretEntry{Name: c.Email, Secret: c.Secret})
|
|
|
|
|
+ }
|
|
|
|
|
+ if len(secrets) == 0 {
|
|
|
return Instance{}, false
|
|
return Instance{}, false
|
|
|
}
|
|
}
|
|
|
return Instance{
|
|
return Instance{
|
|
|
- Id: ib.Id,
|
|
|
|
|
- Tag: ib.Tag,
|
|
|
|
|
- Listen: ib.Listen,
|
|
|
|
|
- Port: ib.Port,
|
|
|
|
|
- Secret: parsed.Secret,
|
|
|
|
|
- Debug: parsed.Debug,
|
|
|
|
|
- ProxyProtocolListener: parsed.ProxyProtocolListener,
|
|
|
|
|
- PreferIP: parsed.PreferIP,
|
|
|
|
|
- FrontingIP: parsed.DomainFronting.IP,
|
|
|
|
|
- FrontingPort: parsed.DomainFronting.Port,
|
|
|
|
|
- FrontingProxyProtocol: parsed.DomainFronting.ProxyProtocol,
|
|
|
|
|
- RouteThroughXray: parsed.RouteThroughXray,
|
|
|
|
|
- XrayRoutePort: parsed.RouteXrayPort,
|
|
|
|
|
|
|
+ Id: ib.Id,
|
|
|
|
|
+ Tag: ib.Tag,
|
|
|
|
|
+ Listen: ib.Listen,
|
|
|
|
|
+ Port: ib.Port,
|
|
|
|
|
+ Secrets: secrets,
|
|
|
|
|
+ Debug: parsed.Debug,
|
|
|
|
|
+ ProxyProtocolListener: parsed.ProxyProtocolListener,
|
|
|
|
|
+ PreferIP: parsed.PreferIP,
|
|
|
|
|
+ FrontingIP: parsed.DomainFronting.IP,
|
|
|
|
|
+ FrontingPort: parsed.DomainFronting.Port,
|
|
|
|
|
+ FrontingProxyProtocol: parsed.DomainFronting.ProxyProtocol,
|
|
|
|
|
+ ThrottleMaxConnections: parsed.ThrottleMaxConnections,
|
|
|
|
|
+ RouteThroughXray: parsed.RouteThroughXray,
|
|
|
|
|
+ XrayRoutePort: parsed.RouteXrayPort,
|
|
|
}, true
|
|
}, true
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -185,12 +220,12 @@ func (m *Manager) ensureLocked(inst Instance) error {
|
|
|
_ = cur.proc.Stop()
|
|
_ = cur.proc.Stop()
|
|
|
delete(m.procs, inst.Id)
|
|
delete(m.procs, inst.Id)
|
|
|
}
|
|
}
|
|
|
- metricsPort, err := FreeLocalPort()
|
|
|
|
|
|
|
+ apiPort, err := FreeLocalPort()
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
return err
|
|
return err
|
|
|
}
|
|
}
|
|
|
cfgPath := configPathForID(inst.Id)
|
|
cfgPath := configPathForID(inst.Id)
|
|
|
- if err := writeConfig(cfgPath, inst, metricsPort); err != nil {
|
|
|
|
|
|
|
+ if err := writeConfig(cfgPath, inst, apiPort); err != nil {
|
|
|
return err
|
|
return err
|
|
|
}
|
|
}
|
|
|
proc := newProcess(cfgPath, fmt.Sprintf("inbound %d", inst.Id))
|
|
proc := newProcess(cfgPath, fmt.Sprintf("inbound %d", inst.Id))
|
|
@@ -201,7 +236,8 @@ func (m *Manager) ensureLocked(inst Instance) error {
|
|
|
proc: proc,
|
|
proc: proc,
|
|
|
tag: inst.Tag,
|
|
tag: inst.Tag,
|
|
|
fingerprint: fp,
|
|
fingerprint: fp,
|
|
|
- metricsPort: metricsPort,
|
|
|
|
|
|
|
+ apiPort: apiPort,
|
|
|
|
|
+ last: map[string]clientCounters{},
|
|
|
}
|
|
}
|
|
|
logger.Infof("mtproto: started mtg for inbound %d on %s", inst.Id, inst.bindTo())
|
|
logger.Infof("mtproto: started mtg for inbound %d on %s", inst.Id, inst.bindTo())
|
|
|
return nil
|
|
return nil
|
|
@@ -255,18 +291,15 @@ func (m *Manager) StopAll() {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-// CollectTraffic scrapes each running mtg metrics endpoint and returns the
|
|
|
|
|
-// per-inbound byte deltas since the previous scrape.
|
|
|
|
|
-func (m *Manager) CollectTraffic() []Traffic {
|
|
|
|
|
- // Snapshot the state we need under the lock, then release before doing
|
|
|
|
|
- // network I/O so that Ensure/Reconcile/Remove are not blocked.
|
|
|
|
|
|
|
+// CollectTraffic scrapes each running mtg /stats endpoint and returns the
|
|
|
|
|
+// per-client byte deltas since the previous scrape, plus the emails of clients
|
|
|
|
|
+// with at least one live connection.
|
|
|
|
|
+func (m *Manager) CollectTraffic() ([]Traffic, []string) {
|
|
|
type snap struct {
|
|
type snap struct {
|
|
|
- id int
|
|
|
|
|
- metricsPort int
|
|
|
|
|
- tag string
|
|
|
|
|
- haveLast bool
|
|
|
|
|
- lastUp int64
|
|
|
|
|
- lastDown int64
|
|
|
|
|
|
|
+ id int
|
|
|
|
|
+ apiPort int
|
|
|
|
|
+ tag string
|
|
|
|
|
+ last map[string]clientCounters
|
|
|
}
|
|
}
|
|
|
m.mu.Lock()
|
|
m.mu.Lock()
|
|
|
snaps := make([]snap, 0, len(m.procs))
|
|
snaps := make([]snap, 0, len(m.procs))
|
|
@@ -274,54 +307,57 @@ func (m *Manager) CollectTraffic() []Traffic {
|
|
|
if cur.proc == nil || !cur.proc.IsRunning() {
|
|
if cur.proc == nil || !cur.proc.IsRunning() {
|
|
|
continue
|
|
continue
|
|
|
}
|
|
}
|
|
|
- snaps = append(snaps, snap{
|
|
|
|
|
- id: id,
|
|
|
|
|
- metricsPort: cur.metricsPort,
|
|
|
|
|
- tag: cur.tag,
|
|
|
|
|
- haveLast: cur.haveLast,
|
|
|
|
|
- lastUp: cur.lastUp,
|
|
|
|
|
- lastDown: cur.lastDown,
|
|
|
|
|
- })
|
|
|
|
|
|
|
+ lastCopy := make(map[string]clientCounters, len(cur.last))
|
|
|
|
|
+ for k, v := range cur.last {
|
|
|
|
|
+ lastCopy[k] = v
|
|
|
|
|
+ }
|
|
|
|
|
+ snaps = append(snaps, snap{id: id, apiPort: cur.apiPort, tag: cur.tag, last: lastCopy})
|
|
|
}
|
|
}
|
|
|
m.mu.Unlock()
|
|
m.mu.Unlock()
|
|
|
|
|
|
|
|
- out := make([]Traffic, 0, len(snaps))
|
|
|
|
|
|
|
+ var out []Traffic
|
|
|
|
|
+ var online []string
|
|
|
for _, s := range snaps {
|
|
for _, s := range snaps {
|
|
|
- up, down, ok := scrapeTraffic(s.metricsPort)
|
|
|
|
|
|
|
+ users, ok := scrapeStats(s.apiPort)
|
|
|
if !ok {
|
|
if !ok {
|
|
|
continue
|
|
continue
|
|
|
}
|
|
}
|
|
|
- var du, dd int64
|
|
|
|
|
- if s.haveLast {
|
|
|
|
|
- du = up - s.lastUp
|
|
|
|
|
- dd = down - s.lastDown
|
|
|
|
|
|
|
+ newLast := make(map[string]clientCounters, len(users))
|
|
|
|
|
+ for email, u := range users {
|
|
|
|
|
+ up := u.BytesIn
|
|
|
|
|
+ down := u.BytesOut
|
|
|
|
|
+ newLast[email] = clientCounters{up: up, down: down}
|
|
|
|
|
+ if u.Connections > 0 {
|
|
|
|
|
+ online = append(online, email)
|
|
|
|
|
+ }
|
|
|
|
|
+ prev, had := s.last[email]
|
|
|
|
|
+ if !had {
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ du := up - prev.up
|
|
|
|
|
+ dd := down - prev.down
|
|
|
if du < 0 {
|
|
if du < 0 {
|
|
|
du = 0
|
|
du = 0
|
|
|
}
|
|
}
|
|
|
if dd < 0 {
|
|
if dd < 0 {
|
|
|
dd = 0
|
|
dd = 0
|
|
|
}
|
|
}
|
|
|
|
|
+ if du > 0 || dd > 0 {
|
|
|
|
|
+ out = append(out, Traffic{Tag: s.tag, Email: email, Up: du, Down: dd})
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- // Re-acquire lock to persist the new baseline, but only if the entry
|
|
|
|
|
- // still exists (it may have been removed during the scrape).
|
|
|
|
|
m.mu.Lock()
|
|
m.mu.Lock()
|
|
|
if cur, ok := m.procs[s.id]; ok {
|
|
if cur, ok := m.procs[s.id]; ok {
|
|
|
- cur.lastUp = up
|
|
|
|
|
- cur.lastDown = down
|
|
|
|
|
- cur.haveLast = true
|
|
|
|
|
|
|
+ cur.last = newLast
|
|
|
}
|
|
}
|
|
|
m.mu.Unlock()
|
|
m.mu.Unlock()
|
|
|
-
|
|
|
|
|
- if s.haveLast && (du > 0 || dd > 0) {
|
|
|
|
|
- out = append(out, Traffic{Tag: s.tag, Up: du, Down: dd})
|
|
|
|
|
- }
|
|
|
|
|
}
|
|
}
|
|
|
- return out
|
|
|
|
|
|
|
+ return out, online
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// FreeLocalPort asks the OS for an unused loopback TCP port. It is used both
|
|
// FreeLocalPort asks the OS for an unused loopback TCP port. It is used both
|
|
|
-// for mtg's metrics endpoint and to allocate the per-inbound SOCKS egress
|
|
|
|
|
|
|
+// for mtg's /stats API endpoint and to allocate the per-inbound SOCKS egress
|
|
|
// bridge port persisted into mtproto inbound settings.
|
|
// bridge port persisted into mtproto inbound settings.
|
|
|
func FreeLocalPort() (int, error) {
|
|
func FreeLocalPort() (int, error) {
|
|
|
l, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
|
|
l, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
|
|
@@ -332,13 +368,13 @@ func FreeLocalPort() (int, error) {
|
|
|
return l.Addr().(*net.TCPAddr).Port, nil
|
|
return l.Addr().(*net.TCPAddr).Port, nil
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-// renderConfig builds the mtg TOML for an instance. Top-level keys must precede
|
|
|
|
|
-// any [section] header in TOML, so the layout is: required keys, then the
|
|
|
|
|
-// optional scalar tuning, then [domain-fronting], and finally [stats.prometheus]
|
|
|
|
|
-// — which x-ui always emits and scrapes for traffic (see scrapeTraffic).
|
|
|
|
|
-func renderConfig(inst Instance, metricsPort int) string {
|
|
|
|
|
|
|
+// renderConfig builds the mtg-multi TOML for an instance. Top-level keys must
|
|
|
|
|
+// precede any [section] header in TOML, and [secrets] must be the final section
|
|
|
|
|
+// so trailing keys are not swallowed by another table. The layout is therefore:
|
|
|
|
|
+// top-level scalars (incl. api-bind-to), then [domain-fronting], [network] and
|
|
|
|
|
+// [throttle], and finally [secrets] with one named secret per active client.
|
|
|
|
|
+func renderConfig(inst Instance, apiPort int) string {
|
|
|
var b strings.Builder
|
|
var b strings.Builder
|
|
|
- fmt.Fprintf(&b, "secret = %q\n", inst.Secret)
|
|
|
|
|
fmt.Fprintf(&b, "bind-to = %q\n", inst.bindTo())
|
|
fmt.Fprintf(&b, "bind-to = %q\n", inst.bindTo())
|
|
|
if inst.Debug {
|
|
if inst.Debug {
|
|
|
b.WriteString("debug = true\n")
|
|
b.WriteString("debug = true\n")
|
|
@@ -349,10 +385,11 @@ func renderConfig(inst Instance, metricsPort int) string {
|
|
|
if inst.PreferIP != "" {
|
|
if inst.PreferIP != "" {
|
|
|
fmt.Fprintf(&b, "prefer-ip = %q\n", inst.PreferIP)
|
|
fmt.Fprintf(&b, "prefer-ip = %q\n", inst.PreferIP)
|
|
|
}
|
|
}
|
|
|
|
|
+ fmt.Fprintf(&b, "api-bind-to = \"127.0.0.1:%d\"\n", apiPort)
|
|
|
if inst.FrontingIP != "" || inst.FrontingPort > 0 || inst.FrontingProxyProtocol {
|
|
if inst.FrontingIP != "" || inst.FrontingPort > 0 || inst.FrontingProxyProtocol {
|
|
|
b.WriteString("\n[domain-fronting]\n")
|
|
b.WriteString("\n[domain-fronting]\n")
|
|
|
if inst.FrontingIP != "" {
|
|
if inst.FrontingIP != "" {
|
|
|
- fmt.Fprintf(&b, "ip = %q\n", inst.FrontingIP)
|
|
|
|
|
|
|
+ fmt.Fprintf(&b, "host = %q\n", inst.FrontingIP)
|
|
|
}
|
|
}
|
|
|
if inst.FrontingPort > 0 {
|
|
if inst.FrontingPort > 0 {
|
|
|
fmt.Fprintf(&b, "port = %d\n", inst.FrontingPort)
|
|
fmt.Fprintf(&b, "port = %d\n", inst.FrontingPort)
|
|
@@ -367,92 +404,51 @@ func renderConfig(inst Instance, metricsPort int) string {
|
|
|
if inst.RouteThroughXray && inst.XrayRoutePort > 0 {
|
|
if inst.RouteThroughXray && inst.XrayRoutePort > 0 {
|
|
|
fmt.Fprintf(&b, "\n[network]\nproxies = [\"socks5://127.0.0.1:%d\"]\n", inst.XrayRoutePort)
|
|
fmt.Fprintf(&b, "\n[network]\nproxies = [\"socks5://127.0.0.1:%d\"]\n", inst.XrayRoutePort)
|
|
|
}
|
|
}
|
|
|
- fmt.Fprintf(&b, "\n[stats.prometheus]\nenabled = true\nbind-to = \"127.0.0.1:%d\"\nhttp-path = \"/metrics\"\nmetric-prefix = \"mtg\"\n", metricsPort)
|
|
|
|
|
|
|
+ if inst.ThrottleMaxConnections > 0 {
|
|
|
|
|
+ fmt.Fprintf(&b, "\n[throttle]\nmax-connections = %d\n", inst.ThrottleMaxConnections)
|
|
|
|
|
+ }
|
|
|
|
|
+ b.WriteString("\n[secrets]\n")
|
|
|
|
|
+ for _, e := range inst.Secrets {
|
|
|
|
|
+ fmt.Fprintf(&b, "%q = %q\n", e.Name, e.Secret)
|
|
|
|
|
+ }
|
|
|
return b.String()
|
|
return b.String()
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-func writeConfig(path string, inst Instance, metricsPort int) error {
|
|
|
|
|
|
|
+func writeConfig(path string, inst Instance, apiPort int) error {
|
|
|
if err := os.MkdirAll(configDir(), 0o750); err != nil {
|
|
if err := os.MkdirAll(configDir(), 0o750); err != nil {
|
|
|
return err
|
|
return err
|
|
|
}
|
|
}
|
|
|
- return os.WriteFile(path, []byte(renderConfig(inst, metricsPort)), 0o640)
|
|
|
|
|
|
|
+ return os.WriteFile(path, []byte(renderConfig(inst, apiPort)), 0o640)
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// statsUser is one entry of the mtg-multi /stats users map. bytes_in is traffic
|
|
|
|
|
+// the client sent to the proxy (upload) and bytes_out is what the proxy returned
|
|
|
|
|
+// (download).
|
|
|
|
|
+type statsUser struct {
|
|
|
|
|
+ Connections int64 `json:"connections"`
|
|
|
|
|
+ BytesIn int64 `json:"bytes_in"`
|
|
|
|
|
+ BytesOut int64 `json:"bytes_out"`
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-// scrapeTraffic reads the mtg Prometheus metrics endpoint and sums byte
|
|
|
|
|
-// counters by direction. mtg exposes a traffic counter labelled with a
|
|
|
|
|
-// direction; "to_telegram" is treated as upload and "to_client" as download.
|
|
|
|
|
-// Best-effort: an unreachable endpoint or unrecognised format yields ok=false.
|
|
|
|
|
-func scrapeTraffic(port int) (up int64, down int64, ok bool) {
|
|
|
|
|
|
|
+// scrapeStats reads the mtg-multi /stats JSON API and returns the per-user
|
|
|
|
|
+// cumulative counters. Best-effort: an unreachable endpoint or unparseable body
|
|
|
|
|
+// yields ok=false.
|
|
|
|
|
+func scrapeStats(port int) (map[string]statsUser, bool) {
|
|
|
client := http.Client{Timeout: 3 * time.Second}
|
|
client := http.Client{Timeout: 3 * time.Second}
|
|
|
- req, reqErr := http.NewRequestWithContext(context.Background(), http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/metrics", port), nil)
|
|
|
|
|
- if reqErr != nil {
|
|
|
|
|
- return 0, 0, false
|
|
|
|
|
|
|
+ req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/stats", port), nil)
|
|
|
|
|
+ if err != nil {
|
|
|
|
|
+ return nil, false
|
|
|
}
|
|
}
|
|
|
resp, err := client.Do(req)
|
|
resp, err := client.Do(req)
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
- return 0, 0, false
|
|
|
|
|
|
|
+ return nil, false
|
|
|
}
|
|
}
|
|
|
defer resp.Body.Close()
|
|
defer resp.Body.Close()
|
|
|
- scanner := bufio.NewScanner(resp.Body)
|
|
|
|
|
- scanner.Buffer(make([]byte, 64*1024), 1024*1024)
|
|
|
|
|
- found := false
|
|
|
|
|
- for scanner.Scan() {
|
|
|
|
|
- line := strings.TrimSpace(scanner.Text())
|
|
|
|
|
- if line == "" || line[0] == '#' || !strings.Contains(line, "traffic") {
|
|
|
|
|
- continue
|
|
|
|
|
- }
|
|
|
|
|
- name, labels, value, perr := parseMetricLine(line)
|
|
|
|
|
- if perr != nil || !strings.HasPrefix(name, "mtg") {
|
|
|
|
|
- continue
|
|
|
|
|
- }
|
|
|
|
|
- switch labels["direction"] {
|
|
|
|
|
- case "to_telegram", "egress", "up":
|
|
|
|
|
- up += int64(value)
|
|
|
|
|
- case "to_client", "ingress", "down":
|
|
|
|
|
- down += int64(value)
|
|
|
|
|
- default:
|
|
|
|
|
- down += int64(value)
|
|
|
|
|
- }
|
|
|
|
|
- found = true
|
|
|
|
|
- }
|
|
|
|
|
- if err := scanner.Err(); err != nil {
|
|
|
|
|
- logger.Debug("mtproto: metrics scan error:", err)
|
|
|
|
|
- }
|
|
|
|
|
- return up, down, found
|
|
|
|
|
-}
|
|
|
|
|
-
|
|
|
|
|
-func parseMetricLine(line string) (name string, labels map[string]string, value float64, err error) {
|
|
|
|
|
- labels = map[string]string{}
|
|
|
|
|
- var rest string
|
|
|
|
|
- if brace := strings.IndexByte(line, '{'); brace >= 0 {
|
|
|
|
|
- name = line[:brace]
|
|
|
|
|
- end := strings.IndexByte(line, '}')
|
|
|
|
|
- if end < brace {
|
|
|
|
|
- return "", nil, 0, fmt.Errorf("malformed metric line")
|
|
|
|
|
- }
|
|
|
|
|
- for kv := range strings.SplitSeq(line[brace+1:end], ",") {
|
|
|
|
|
- before, after, ok := strings.Cut(kv, "=")
|
|
|
|
|
- if !ok {
|
|
|
|
|
- continue
|
|
|
|
|
- }
|
|
|
|
|
- labels[strings.TrimSpace(before)] = strings.Trim(strings.TrimSpace(after), `"`)
|
|
|
|
|
- }
|
|
|
|
|
- rest = strings.TrimSpace(line[end+1:])
|
|
|
|
|
- } else {
|
|
|
|
|
- fields := strings.Fields(line)
|
|
|
|
|
- if len(fields) < 2 {
|
|
|
|
|
- return "", nil, 0, fmt.Errorf("malformed metric line")
|
|
|
|
|
- }
|
|
|
|
|
- name = fields[0]
|
|
|
|
|
- rest = fields[1]
|
|
|
|
|
- }
|
|
|
|
|
- valFields := strings.Fields(rest)
|
|
|
|
|
- if len(valFields) == 0 {
|
|
|
|
|
- return "", nil, 0, fmt.Errorf("missing metric value")
|
|
|
|
|
|
|
+ var parsed struct {
|
|
|
|
|
+ Users map[string]statsUser `json:"users"`
|
|
|
}
|
|
}
|
|
|
- value, err = strconv.ParseFloat(valFields[0], 64)
|
|
|
|
|
- if err != nil {
|
|
|
|
|
- return "", nil, 0, err
|
|
|
|
|
|
|
+ if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
|
|
|
|
|
+ return nil, false
|
|
|
}
|
|
}
|
|
|
- return name, labels, value, nil
|
|
|
|
|
|
|
+ return parsed.Users, true
|
|
|
}
|
|
}
|