feat: implement Prometheus pusher

go.mod

@@ -5,6 +5,15 @@ go 1.22
 require (
 	github.com/google/uuid v1.6.0
 	github.com/jessevdk/go-flags v1.5.0
+	github.com/prometheus/client_golang v1.19.0
 )
 
-require golang.org/x/sys v0.17.0 // indirect
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/prometheus/client_model v0.5.0 // indirect
+	github.com/prometheus/common v0.48.0 // indirect
+	github.com/prometheus/procfs v0.12.0 // indirect
+	golang.org/x/sys v0.17.0 // indirect
+	google.golang.org/protobuf v1.32.0 // indirect
+)

go.sum

@@ -1,7 +1,23 @@
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/jessevdk/go-flags v1.5.0 h1:1jKYvbxEjfUl0fmqTCOfonvskHHXMjBySTLW4y9LFvc=
 github.com/jessevdk/go-flags v1.5.0/go.mod h1:Fw0T6WPc1dYxT4mKEZRfG5kJhaTDP9pj1c2EWnYs/m4=
+github.com/prometheus/client_golang v1.19.0 h1:ygXvpU1AoN1MhdzckN+PyD9QJOSD4x7kmXYlnfbA6JU=
+github.com/prometheus/client_golang v1.19.0/go.mod h1:ZRM9uEAypZakd+q/x7+gmsvXdURP+DABIEIjnmDdp+k=
+github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw=
+github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
+github.com/prometheus/common v0.48.0 h1:QO8U2CdOzSn1BBsmXJXduaaW+dY/5QLjfB8svtSzKKE=
+github.com/prometheus/common v0.48.0/go.mod h1:0/KsvlIEfPQCQ5I2iNSAWKPZziNCvRs5EC6ILDTlAPc=
+github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
+github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
 golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
+google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=

prometheus.go

@@ -0,0 +1,77 @@
+package gojob
+
+import (
+	"fmt"
+	"log/slog"
+	"strings"
+	"sync"
+
+	"github.com/WangYihang/gojob/pkg/runner"
+	"github.com/WangYihang/gojob/pkg/version"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/push"
+)
+
+var (
+	numTotal = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "num_total",
+			Help: "Total number of processed events",
+		},
+		[]string{"version", "runner_ip", "runner_country", "runner_region", "runner_city"},
+	)
+	numFailed = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "num_failed",
+			Help: "Total number of failed events",
+		},
+		[]string{"version", "runner_ip", "runner_country", "runner_region", "runner_city"},
+	)
+	numSucceed = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "num_succeed",
+			Help: "Total number of succeeded events",
+		},
+		[]string{"version", "runner_ip", "runner_country", "runner_region", "runner_city"},
+	)
+	numFinished = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "num_finished",
+			Help: "Total number of finished events",
+		},
+		[]string{"version", "runner_ip", "runner_country", "runner_region", "runner_city"},
+	)
+)
+
+func prometheusPusher(url, job string, statusChan <-chan Status, wg *sync.WaitGroup) {
+	registry := prometheus.NewRegistry()
+	registry.MustRegister(numTotal, numFailed, numSucceed, numFinished)
+	instance := fmt.Sprintf(
+		"gojob-%s-%s-%s",
+		version.Version,
+		strings.ToLower(runner.Runner.Country),
+		runner.Runner.IP,
+	)
+	labels := prometheus.Labels{
+		"version":        version.Version,
+		"runner_ip":      runner.Runner.IP,
+		"runner_country": runner.Runner.Country,
+		"runner_region":  runner.Runner.Region,
+		"runner_city":    runner.Runner.City,
+	}
+	go func() {
+		for status := range statusChan {
+			slog.Info("promehteus pusher", slog.Any("status", status))
+			numTotal.With(labels).Set(float64(status.NumTotal))
+			numFailed.With(labels).Set(float64(status.NumFailed))
+			numSucceed.With(labels).Set(float64(status.NumSucceed))
+			numFinished.With(labels).Set(float64(status.NumFinished))
+			if err := push.New(url, job).Grouping(
+				"instance", instance,
+			).Gatherer(registry).Push(); err != nil {
+				slog.Error("error occured while pushing to prometheus", slog.String("error", err.Error()))
+			}
+		}
+		wg.Done()
+	}()
+}