-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Rory Z <[email protected]>
- Loading branch information
Showing
8 changed files
with
153 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| package prober | ||
|
|
||
| import ( | ||
| "net/http" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/promhttp" | ||
| ) | ||
|
|
||
| func Handler(w http.ResponseWriter, r *http.Request, logger log.Logger) { | ||
| probeSuccessGauge := prometheus.NewGauge(prometheus.GaugeOpts{ | ||
| Namespace: "emqx", | ||
| Subsystem: "mqtt", | ||
| Name: "probe_success", | ||
| Help: "Displays whether or not the probe was a success", | ||
| }) | ||
| probeDurationGauge := prometheus.NewGauge(prometheus.GaugeOpts{ | ||
| Namespace: "emqx", | ||
| Subsystem: "mqtt", | ||
| Name: "probe_duration_seconds", | ||
| Help: "Returns how long the probe took to complete in seconds", | ||
| }) | ||
|
|
||
| registry := prometheus.NewRegistry() | ||
| registry.MustRegister(probeSuccessGauge) | ||
| registry.MustRegister(probeDurationGauge) | ||
|
|
||
| start := time.Now() | ||
| if ProbeMQTT(logger) { | ||
| probeSuccessGauge.Set(1) | ||
| } else { | ||
| probeSuccessGauge.Set(0) | ||
| } | ||
| probeDurationGauge.Set(time.Since(start).Seconds()) | ||
|
|
||
| h := promhttp.HandlerFor(registry, promhttp.HandlerOpts{}) | ||
| h.ServeHTTP(w, r) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| package prober | ||
|
|
||
| import ( | ||
| "time" | ||
|
|
||
| mqtt "github.com/eclipse/paho.mqtt.golang" | ||
| "github.com/go-kit/log" | ||
| "github.com/go-kit/log/level" | ||
| ) | ||
|
|
||
| type MQTTProbe struct { | ||
| Client mqtt.Client | ||
| MsgChan <-chan mqtt.Message | ||
| } | ||
|
|
||
| var mqttProbe *MQTTProbe | ||
|
|
||
| func initMQTTProbe(logger log.Logger) *MQTTProbe { | ||
| opt := mqtt.NewClientOptions().AddBroker("tcp://broker.emqx.io:1883").SetClientID("emqx-exporter") | ||
| opt.SetOnConnectHandler(func(c mqtt.Client) { | ||
| level.Info(logger).Log("msg", "Connected to MQTT broker") | ||
| }) | ||
| opt.SetConnectionLostHandler(func(c mqtt.Client, err error) { | ||
| level.Error(logger).Log("msg", "Lost connection to MQTT broker", "err", err) | ||
| }) | ||
| c := mqtt.NewClient(opt) | ||
| if token := c.Connect(); token.Wait() && token.Error() != nil { | ||
| level.Error(logger).Log("msg", "Failed to connect to MQTT broker", "err", token.Error()) | ||
| panic(token.Error()) | ||
| } | ||
|
|
||
| var msgChan = make(chan mqtt.Message) | ||
| if token := c.Subscribe("emqx-exporter", 1, func(c mqtt.Client, m mqtt.Message) { | ||
| msgChan <- m | ||
| }); token.Wait() && token.Error() != nil { | ||
| level.Error(logger).Log("msg", "Failed to subscribe to MQTT topic", "err", token.Error()) | ||
| panic(token.Error()) | ||
| } | ||
|
|
||
| return &MQTTProbe{ | ||
| Client: c, | ||
| MsgChan: msgChan, | ||
| } | ||
| } | ||
|
|
||
| func ProbeMQTT(logger log.Logger) bool { | ||
| if mqttProbe == nil { | ||
| mqttProbe = initMQTTProbe(logger) | ||
| } | ||
|
|
||
| if !mqttProbe.Client.IsConnected() { | ||
| return false | ||
| } | ||
|
|
||
| if token := mqttProbe.Client.Publish("emqx-exporter", 1, false, "hello world"); token.Wait() && token.Error() != nil { | ||
| return false | ||
| } | ||
|
|
||
| select { | ||
| case msg := <-mqttProbe.MsgChan: | ||
| if msg == nil { | ||
| return false | ||
| } | ||
| case <-time.After(5 * time.Second): | ||
| return false | ||
| } | ||
|
|
||
| return true | ||
| } |