Merge pull request #2 from hikhvar/improve-client-resilience

Improved client resilience

go.sum

@@ -76,6 +76,7 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190613194153-d28f0bde5980 h1:dfGZHvZk057jK2MCeWus/TowKpJ8y4AmooUzdBSR9GU=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

main.go

@@ -32,6 +32,8 @@ func main() {
 	mc := collector.NewMultiCollector(sInfo)
 
 	prometheus.MustRegister(mc)
+
+	prometheus.MustRegister(mc, collector.NewClient(c))
 	// The Handler function provides a default handler to expose metrics
 	// via an HTTP server. "/metrics" is the usual endpoint for that.
 	http.Handle("/metrics", promhttp.Handler())

pkg/collector/client.go

@@ -0,0 +1,39 @@
+package collector
+
+import (
+	"github.com/hikhvar/ts3exporter/pkg/serverquery"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const clientSubsystem = "client"
+
+// InstrumentedClient provides metrics from a serverquery client
+type InstrumentedClient interface {
+	Metrics() serverquery.ClientMetrics
+}
+
+// Client is a collector providing metrics of the internal serverquery client
+type Client struct {
+	source InstrumentedClient
+
+	Failed  *prometheus.Desc
+	Success *prometheus.Desc
+}
+
+func NewClient(c InstrumentedClient) *Client {
+	return &Client{
+		source:  c,
+		Failed:  prometheus.NewDesc(fqdn(clientSubsystem, "commands_failed_total"), "total failed server query command", nil, nil),
+		Success: prometheus.NewDesc(fqdn(clientSubsystem, "commands_successful_total"), "total successful server query command", nil, nil),
+	}
+}
+
+func (c *Client) Describe(desc chan<- *prometheus.Desc) {
+	prometheus.DescribeByCollect(c, desc)
+}
+
+func (c *Client) Collect(met chan<- prometheus.Metric) {
+	m := c.source.Metrics()
+	met <- prometheus.MustNewConstMetric(c.Success, prometheus.CounterValue, float64(m.Success()))
+	met <- prometheus.MustNewConstMetric(c.Failed, prometheus.CounterValue, float64(m.Failed()))
+}

pkg/serverquery/client.go

@@ -1,16 +1,23 @@
 package serverquery
 
 import (
+	"errors"
 	"fmt"
+	"syscall"
+	"time"
 
 	"github.com/multiplay/go-ts3"
 )
 
 type Client struct {
-	ts3Client       *ts3.Client
-	user            string
-	password        string
-	virtualServerID int
+	ts3Client          *ts3.Client
+	limiter            *time.Ticker
+	user               string
+	password           string
+	remote             string
+	serverQueryOptions []options
+	virtualServerID    int
+	metrics            ClientMetrics
 }
 
 type options func() func(client *ts3.Client) error
@@ -19,11 +26,22 @@ type options func() func(client *ts3.Client) error
 // logged in.
 func NewClient(remote, user, password string, serverQueryOptions ...options) (*Client, error) {
 	c := &Client{
-		user:     user,
-		password: password,
+		user:               user,
+		password:           password,
+		remote:             remote,
+		serverQueryOptions: serverQueryOptions,
 	}
-	err := c.login(remote, serverQueryOptions...)
-	return c, err
+	return c, c.reconnect()
+}
+
+func (c *Client) reconnect() error {
+	if err := c.login(c.remote, c.serverQueryOptions...); err != nil {
+		return fmt.Errorf("failed to login: %w", err)
+	}
+	if err := c.setupLimiter(); err != nil {
+		return fmt.Errorf("failed to setup limiter: %w", err)
+	}
+	return nil
 }
 
 // login connects to the given remote with the given options.
@@ -45,18 +63,64 @@ func (c *Client) login(remote string, serverQueryOptions ...options) error {
 	return nil
 }
 
+func (c *Client) setupLimiter() error {
+	type instanceInfo struct {
+		TimeWindow float64 `sq:"serverinstance_serverquery_flood_time"`
+		Commands   float64 `sq:"serverinstance_serverquery_flood_commands"`
+	}
+	res, err := c.Exec("instanceinfo")
+	if err != nil {
+		return fmt.Errorf("failed to execute instanceinfo command: %w", err)
+	}
+	if len(res) != 1 {
+		return fmt.Errorf("expected exactly one response got: %d", len(res))
+	}
+	if len(res[0].Items) != 1 {
+		return fmt.Errorf("expected exactly one item got %d", len(res[0].Items))
+	}
+	var ii instanceInfo
+	if err := res[0].Items[0].ReadInto(&ii); err != nil {
+		return fmt.Errorf("failed to parse answer into a instance info")
+	}
+	// add 10 % buffer to not run into the flood limit
+	// multiply by 1000 to get milliseconds. Most probably the flood limit interval is several hundred milliseconds
+	interval := time.Duration((ii.TimeWindow/ii.Commands)*1100) * time.Millisecond
+	c.limiter = time.NewTicker(interval)
+	return nil
+}
+
+func (c *Client) Metrics() ClientMetrics {
+	return c.metrics
+}
+
 func (c *Client) Exec(cmd string) ([]Result, error) {
+	if c.limiter != nil {
+		<-c.limiter.C
+	}
+	if c.ts3Client == nil {
+		err := c.reconnect()
+		if err != nil {
+			return nil, fmt.Errorf("failed to reconnect: %w", err)
+		}
+	}
 	raw, err := c.ts3Client.Exec(cmd)
 	if err != nil {
+		c.metrics.CountFailure()
+		// If pipe broke, reconnect on next execution
+		if errors.Is(err, syscall.EPIPE) {
+			c.ts3Client = nil
+		}
 		return nil, fmt.Errorf("failed to execute command: %w", err)
 	}
 	ret := make([]Result, 0, len(raw))
 	for _, r := range raw {
 		p, err := Parse(r)
 		if err != nil {
+			c.metrics.CountFailure()
 			return nil, fmt.Errorf("failed to parse answer: %w", err)
 		}
 		ret = append(ret, p)
 	}
+	c.metrics.CountSuccess()
 	return ret, nil
 }

pkg/serverquery/clientmetrics.go

@@ -0,0 +1,22 @@
+package serverquery
+
+type ClientMetrics struct {
+	successful int
+	failed     int
+}
+
+func (c *ClientMetrics) CountFailure() {
+	c.failed = c.failed + 1
+}
+
+func (c *ClientMetrics) CountSuccess() {
+	c.successful = c.successful + 1
+}
+
+func (c *ClientMetrics) Success() int {
+	return c.successful
+}
+
+func (c *ClientMetrics) Failed() int {
+	return c.failed
+}