27143603019

Committed 08 Jun 2026 02:08PM UTC coverage: 58.544% (-0.2%) from 58.772%

Build # 27143603019

Build Type

Pull #501

github

Committed by

rsafonseca

Commit Message

feat(metrics): instrument handler/RPC channel buffers and worker pools

Several buffered channels and goroutine pools that sit on the hot path
had no instrumentation, so there was no way to tell from metrics when a
server was hitting its buffer (BUFFER_*) or concurrency (CONCURRENCY_*)
limits — the channels would just fill and apply backpressure silently.

Add capacity/utilization metrics for the previously-unmonitored ones:

Channel capacity (reuses the channel_capacity histogram, new `channel` tags):
  - handler_chlocalprocess   (buffer.handler.localprocess)
  - handler_chremoteprocess  (buffer.handler.remoteprocess)
  - rpc_server_userkickchan  (shares the nats messages buffer)

Worker-pool utilization (new gauges pitaya_worker_pool_busy_workers and
pitaya_worker_pool_total_workers, tagged by `pool`):
  - handler_dispatch    (concurrency.handler.dispatch)
  - rpc_server_remote   (cluster.rpc.server.nats.services)

Implementation:
  - Handler dispatch channels and pool are sampled periodically from
    App.periodicMetrics (no extra work on the per-message path).
  - userKickCh and the remote pool are reported from the existing nats
    reportMetrics, alongside the channels already covered there.
  - Add metrics.ReportChannelCapacity / ReportWorkerPoolUsage helpers and
    refactor nats reportMetrics to use them; each call builds its own
    label map so prometheus's label mutation can't leak across reporters.

Busy-worker counts are tracked with atomic int32 counters (two adds per
message, dwarfed by the actual handler/RPC work). Purely additive — no
existing metric, behaviour, or config is changed; works on both the
prometheus and statsd reporters.

Pull Request Pull Request #501: feat(metrics): instrument handler/RPC channel buffers and worker pools

Coverage Stats

26 of 81 new or added lines in 5 files covered. (32.1%)

5 existing lines in 3 files now uncovered.

5105 of 8720 relevant lines covered (58.54%)

0.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.91

/pkg/modules/binary.go

//go:build linux || darwin
// Copyright (c) TFG Co. All Rights Reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.

package modules

import (
        "bufio"
        "os/exec"
        "syscall"
        "time"

        "github.com/topfreegames/pitaya/v3/pkg/constants"
        "github.com/topfreegames/pitaya/v3/pkg/logger"
)

// Binary is a pitaya module that starts a binary as a child process and
// pipes its stdout
type Binary struct {
        Base
        binPath                  string
        args                     []string
        gracefulShutdownInterval time.Duration
        cmd                      *exec.Cmd
        exitCh                   chan struct{}
}

// NewBinary creates a new binary module with the given path
func NewBinary(binPath string, args []string, gracefulShutdownInterval ...time.Duration) *Binary {
        gracefulTime := 15 * time.Second
        if len(gracefulShutdownInterval) > 0 {
                gracefulTime = gracefulShutdownInterval[0]
        }
        return &Binary{
                binPath:                  binPath,
                args:                     args,
                gracefulShutdownInterval: gracefulTime,
                exitCh:                   make(chan struct{}),
        }
}

// GetExitChannel gets a channel that is closed when the binary dies
func (b *Binary) GetExitChannel() chan struct{} {
        return b.exitCh
}

// Init initializes the binary
func (b *Binary) Init() error {
        b.cmd = exec.Command(b.binPath, b.args...)
        stdout, _ := b.cmd.StdoutPipe()
        stdOutScanner := bufio.NewScanner(stdout)
        stderr, _ := b.cmd.StderrPipe()
        stdErrScanner := bufio.NewScanner(stderr)
        go func() {
                for stdOutScanner.Scan() {
                        logger.Log.Info(stdOutScanner.Text())
                }
        }()
        go func() {
                for stdErrScanner.Scan() {
                        logger.Log.Error(stdErrScanner.Text())
                }
        }()
        err := b.cmd.Start()
        go func() {
                b.cmd.Wait()
                close(b.exitCh)
        }()
        return err
}

// Shutdown shutdowns the binary module
func (b *Binary) Shutdown() error {
        err := b.cmd.Process.Signal(syscall.SIGTERM)
        if err != nil {
                return err
        }
        timeout := time.After(b.gracefulShutdownInterval)
        select {
        case <-b.exitCh:
                return nil
        case <-timeout:
                b.cmd.Process.Kill()
                return constants.ErrTimeoutTerminatingBinaryModule
        }
}

1	//go:build linux \|\| darwin
2	// Copyright (c) TFG Co. All Rights Reserved.
3	//
4	// Permission is hereby granted, free of charge, to any person obtaining a copy
5	// of this software and associated documentation files (the "Software"), to deal
6	// in the Software without restriction, including without limitation the rights
7	// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8	// copies of the Software, and to permit persons to whom the Software is
9	// furnished to do so, subject to the following conditions:
10	//
11	// The above copyright notice and this permission notice shall be included in all
12	// copies or substantial portions of the Software.
13	//
14	// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15	// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16	// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17	// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18	// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19	// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
20	// SOFTWARE.
21
22	package modules
23
24	import (
25	"bufio"
26	"os/exec"
27	"syscall"
28	"time"
29
30	"github.com/topfreegames/pitaya/v3/pkg/constants"
31	"github.com/topfreegames/pitaya/v3/pkg/logger"
32	)
33
34	// Binary is a pitaya module that starts a binary as a child process and
35	// pipes its stdout
36	type Binary struct {
37	Base
38	binPath string
39	args []string
40	gracefulShutdownInterval time.Duration
41	cmd *exec.Cmd
42	exitCh chan struct{}
43	}
44
45	// NewBinary creates a new binary module with the given path
46	func NewBinary(binPath string, args []string, gracefulShutdownInterval ...time.Duration) *Binary {	1✔
47	gracefulTime := 15 * time.Second	1✔
48	if len(gracefulShutdownInterval) > 0 {	2✔
49	gracefulTime = gracefulShutdownInterval[0]	1✔
50	}	1✔
51	return &Binary{	1✔
52	binPath: binPath,	1✔
53	args: args,	1✔
54	gracefulShutdownInterval: gracefulTime,	1✔
55	exitCh: make(chan struct{}),	1✔
56	}	1✔
57	}
58
59	// GetExitChannel gets a channel that is closed when the binary dies
60	func (b *Binary) GetExitChannel() chan struct{} {	×
61	return b.exitCh	×
62	}	×
63
64	// Init initializes the binary
65	func (b *Binary) Init() error {	1✔
66	b.cmd = exec.Command(b.binPath, b.args...)	1✔
67	stdout, _ := b.cmd.StdoutPipe()	1✔
68	stdOutScanner := bufio.NewScanner(stdout)	1✔
69	stderr, _ := b.cmd.StderrPipe()	1✔
70	stdErrScanner := bufio.NewScanner(stderr)	1✔
71	go func() {	2✔
72	for stdOutScanner.Scan() {	1✔
UNCOV 73	logger.Log.Info(stdOutScanner.Text())	×
UNCOV 74	}	×
75	}()
76	go func() {	2✔
77	for stdErrScanner.Scan() {	1✔
78	logger.Log.Error(stdErrScanner.Text())	×
79	}	×
80	}()
81	err := b.cmd.Start()	1✔
82	go func() {	2✔
83	b.cmd.Wait()	1✔
84	close(b.exitCh)	1✔
85	}()	1✔
86	return err	1✔
87	}
88
89	// Shutdown shutdowns the binary module
90	func (b *Binary) Shutdown() error {	1✔
91	err := b.cmd.Process.Signal(syscall.SIGTERM)	1✔
92	if err != nil {	1✔
93	return err	×
94	}	×
95	timeout := time.After(b.gracefulShutdownInterval)	1✔
96	select {	1✔
97	case <-b.exitCh:	1✔
98	return nil	1✔
99	case <-timeout:	×
100	b.cmd.Process.Kill()	×
101	return constants.ErrTimeoutTerminatingBinaryModule	×
102	}
103	}

topfreegames / pitaya / 27143603019

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous