commit b0045277f1e6544bcb88a05e7bde9700acc977d7
parent dcdf10dba5e2887040a8099caa15827a42fee7f1
Author: MTRNord <mtrnord1@gmail.com>
Date: Thu, 15 Feb 2024 21:05:38 +0100
Make updates async from the requests and fix pong logic
Diffstat:
| M | main.go | | | 37 | ++++++++++++++++++++++++++++++++++--- |
| M | ping_json.go | | | 8 | ++++---- |
| M | ping_manager.go | | | 96 | ++++++++++++++++++++++++++++++++++++++++++++----------------------------------- |
3 files changed, 92 insertions(+), 49 deletions(-)
diff --git a/main.go b/main.go
@@ -33,11 +33,42 @@ func main() {
config.RemoteHomeservers[i].Client = createMatrixClient(&config, &config.RemoteHomeservers[i])
}
+ pingCollector := &PingCollector{
+ Config: &config,
+ Mean: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+ Namespace: collector,
+ Name: "mean",
+ Help: "Mean ping time",
+ }, []string{"homeserver", "origin", "direction"}),
+ Median: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+ Namespace: collector,
+ Name: "median",
+ Help: "Median ping time",
+ }, []string{"homeserver", "origin", "direction"}),
+ GMean: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+ Namespace: collector,
+ Name: "gmean",
+ Help: "GMean ping time",
+ }, []string{"homeserver", "origin", "direction"}),
+ Failures: prometheus.NewCounterVec(prometheus.CounterOpts{
+ Namespace: collector,
+ Name: "failures",
+ Help: "Ping failures",
+ }, []string{"origin", "direction"}),
+ }
+
+ go func() {
+ pingCollector.UpdateData()
+ time.Sleep(time.Duration(config.PingRateSeconds+1) * time.Second)
+ for {
+ pingCollector.UpdateData()
+ time.Sleep(time.Duration(config.PingRateSeconds+1) * time.Second)
+ }
+ }()
+
reg := prometheus.NewRegistry()
reg.MustRegister(version.NewCollector(collector))
- reg.MustRegister(&PingCollector{
- Config: &config,
- })
+ reg.MustRegister(pingCollector)
http.Handle("/metrics", promhttp.HandlerFor(
reg,
diff --git a/ping_json.go b/ping_json.go
@@ -12,10 +12,10 @@ type Ping struct {
type Pings map[string]Ping
type Pong struct {
- Diffs []string `json:"diffs"`
- Mean float64 `json:"mean"`
- Median float64 `json:"median"`
- GMean float64 `json:"gmean"`
+ Diffs map[string]string `json:"diffs"`
+ Mean float64 `json:"mean"`
+ Median float64 `json:"median"`
+ GMean float64 `json:"gmean"`
}
type Data struct {
diff --git a/ping_manager.go b/ping_manager.go
@@ -11,28 +11,33 @@ import (
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus"
"maunium.net/go/mautrix"
+ "maunium.net/go/mautrix/id"
)
type PingCollector struct {
Config *Config
- Metrics map[string]*prometheus.Desc
+ Mean *prometheus.GaugeVec
+ Median *prometheus.GaugeVec
+ GMean *prometheus.GaugeVec
+ Failures *prometheus.CounterVec
LastCollected map[string]time.Time
- Failures map[string]int
}
func (c *PingCollector) Describe(ch chan<- *prometheus.Desc) {
- mean := prometheus.NewDesc(prometheus.BuildFQName(collector, "", "mean"), "Mean ping time", []string{"homeserver", "origin", "direction"}, nil)
- median := prometheus.NewDesc(prometheus.BuildFQName(collector, "", "median"), "Median ping time", []string{"homeserver", "origin", "direction"}, nil)
- gmean := prometheus.NewDesc(prometheus.BuildFQName(collector, "", "gmean"), "GMean ping time", []string{"homeserver", "origin", "direction"}, nil)
- failures := prometheus.NewDesc(prometheus.BuildFQName(collector, "", "failures"), "Ping failures", []string{"origin", "direction"}, nil)
- c.Metrics = make(map[string]*prometheus.Desc)
- c.Metrics["mean"] = mean
- c.Metrics["median"] = median
- c.Metrics["gmean"] = gmean
- c.Metrics["failures"] = failures
+ c.Mean.Describe(ch)
+ c.Median.Describe(ch)
+ c.GMean.Describe(ch)
+ c.Failures.Describe(ch)
log.Infof("Registered metrics")
}
+func (c *PingCollector) Collect(ch chan<- prometheus.Metric) {
+ c.Mean.Collect(ch)
+ c.Median.Collect(ch)
+ c.GMean.Collect(ch)
+ c.Failures.Collect(ch)
+}
+
// For collection of matrix metrics we first need to ping.
// To do that we do 2 things:
// 1. Write `!ping` in the ping room as a message from our own homeserver
@@ -40,18 +45,18 @@ func (c *PingCollector) Describe(ch chan<- *prometheus.Desc) {
// We then track our event_ids and make sure that at least one event is reaching our own homeserver
// And at least one event id (can be a different one) reaches the other homeservers.
// We also do react on our own `!ping` message and send a `!pong` back to the ping room based on the maubot echobot logic.
-func (c *PingCollector) Collect(ch chan<- prometheus.Metric) {
+func (c *PingCollector) UpdateData() {
log.Infoln("Starting Collecting metrics")
var wg sync.WaitGroup
wg.Add(1)
// Send ping from our own homeserver
- go c.SendPing(context.Background(), c.Config.OwnHomeserver.Client, ch, &wg)
+ go c.SendPing(context.Background(), c.Config.OwnHomeserver.Client, &wg)
// Send ping from all remote homeservers
for _, homeserver := range c.Config.RemoteHomeservers {
wg.Add(1)
- go c.SendPing(context.Background(), homeserver.Client, ch, &wg)
+ go c.SendPing(context.Background(), homeserver.Client, &wg)
}
wg.Wait()
@@ -59,7 +64,7 @@ func (c *PingCollector) Collect(ch chan<- prometheus.Metric) {
// This sends a ping into the ping room
// It takes a homeserver config as an argument to know where to send the ping
-func (c *PingCollector) SendPing(ctx context.Context, client *mautrix.Client, ch chan<- prometheus.Metric, wg *sync.WaitGroup) {
+func (c *PingCollector) SendPing(ctx context.Context, client *mautrix.Client, wg *sync.WaitGroup) {
defer wg.Done()
if c.LastCollected[client.UserID.Homeserver()].Add(time.Duration(c.Config.PingRateSeconds) * time.Second).After(time.Now()) {
log.Infof("Not sending ping as we sent one less than %d seconds ago", c.Config.PingRateSeconds)
@@ -81,6 +86,7 @@ func (c *PingCollector) SendPing(ctx context.Context, client *mautrix.Client, ch
log.Errorf("Failed to send ping: %s", err)
}
log.Infof("Sent ping as %s with event_id %s", client.UserID, resp.EventID)
+ eventID := resp.EventID
// Poll the json to check if any pongs have been received for this ping
// Otherwise timeout after PingThresholdSeconds
@@ -92,6 +98,10 @@ func (c *PingCollector) SendPing(ctx context.Context, client *mautrix.Client, ch
log.Errorf("Failed to parse homeserver url: %s", err)
return
}
+ direction := "outgoing"
+ if client.HomeserverURL.Host != u.Host {
+ direction = "incoming"
+ }
outer:
for time.Now().Before(pingTime.Add(time.Duration(c.Config.PingThresholdSeconds) * time.Second)) {
@@ -109,16 +119,17 @@ outer:
// Check if we have a pong for this ping
log.Infof("Checking for pong for ping as %s", client.UserID.Homeserver())
if ping, ok := currentData.Pings[client.UserID.Homeserver()]; ok {
- log.Infof("Got pong for ping as %s", client.UserID)
-
+ hs_parsed := id.UserID(c.Config.OwnHomeserver.Username).Homeserver()
+ if _, ok := ping.Pongs[hs_parsed].Diffs[eventID.String()]; ok {
+ log.Infof("Got pong for ping as %s", client.UserID)
+ gotKnownPong = true
+ break outer
+ }
// Check if its from a known remote homeserver in ping.Pongs
- for homeserver := range ping.Pongs {
- if homeserver == client.UserID.Homeserver() {
- continue
- }
- if _, ok := currentData.Pings[homeserver]; ok {
- log.Infof("Got pong for ping as %s from %s", client.UserID, homeserver)
- // break out of outer loop
+ for _, homeserver := range c.Config.RemoteHomeservers {
+ hs_parsed := id.UserID(homeserver.Username).Homeserver()
+ if _, ok := ping.Pongs[hs_parsed].Diffs[eventID.String()]; ok {
+ log.Infof("Got pong for ping as %s", client.UserID)
gotKnownPong = true
break outer
}
@@ -134,31 +145,32 @@ outer:
// If we have not received a pong for this ping, we should log it
if !gotKnownPong {
log.Errorf("Failed to get pong for ping as %s", client.UserID)
- if c.Failures == nil {
- c.Failures = make(map[string]int)
- }
+ c.Failures.WithLabelValues(client.UserID.Homeserver(), direction).Inc()
+ }
- if client.HomeserverURL.Host == u.Host {
- c.Failures["outgoing"]++
- ch <- prometheus.MustNewConstMetric(c.Metrics["failures"], prometheus.CounterValue, float64(c.Failures["outgoing"]), client.UserID.Homeserver(), "outgoing")
- } else {
- c.Failures["incoming"]++
- ch <- prometheus.MustNewConstMetric(c.Metrics["failures"], prometheus.CounterValue, float64(c.Failures["incoming"]), client.UserID.Homeserver(), "incoming")
- }
+ // Wait 5s before we collect the final data
+ time.Sleep(5 * time.Second)
+ pingresp, err := http.Get(c.Config.PingJsonURL)
+ if err != nil {
+ log.Errorf("Failed to get ping json: %s", err)
}
+ // Parse the json
+ // If we have a pong for this ping, we can break the loop
+ // Otherwise we sleep for 1 second and try again
+ defer pingresp.Body.Close()
+ json.NewDecoder(pingresp.Body).Decode(¤tData)
+
// Update mean, median and gmean metrics
// All of these are per homeserver we received a pong from
// They are all of type Gauge (Should they be a historgram?)
for homeserver, ping := range currentData.Pings[client.UserID.Homeserver()].Pongs {
- if client.HomeserverURL.Host == u.Host {
- ch <- prometheus.MustNewConstMetric(c.Metrics["mean"], prometheus.GaugeValue, ping.Mean, homeserver, client.UserID.Homeserver(), "outgoing")
- ch <- prometheus.MustNewConstMetric(c.Metrics["median"], prometheus.GaugeValue, ping.Median, homeserver, client.UserID.Homeserver(), "outgoing")
- ch <- prometheus.MustNewConstMetric(c.Metrics["gmean"], prometheus.GaugeValue, ping.GMean, homeserver, client.UserID.Homeserver(), "outgoing")
- } else {
- ch <- prometheus.MustNewConstMetric(c.Metrics["mean"], prometheus.GaugeValue, ping.Mean, homeserver, client.UserID.Homeserver(), "incoming")
- ch <- prometheus.MustNewConstMetric(c.Metrics["median"], prometheus.GaugeValue, ping.Median, homeserver, client.UserID.Homeserver(), "incoming")
- ch <- prometheus.MustNewConstMetric(c.Metrics["gmean"], prometheus.GaugeValue, ping.GMean, homeserver, client.UserID.Homeserver(), "incoming")
+ direction := "outgoing"
+ if client.HomeserverURL.Host != u.Host {
+ direction = "incoming"
}
+ c.Mean.WithLabelValues(homeserver, client.UserID.Homeserver(), direction).Set(ping.Mean)
+ c.Median.WithLabelValues(homeserver, client.UserID.Homeserver(), direction).Set(ping.Median)
+ c.GMean.WithLabelValues(homeserver, client.UserID.Homeserver(), direction).Set(ping.GMean)
}
}
