81 lines
4.3 KiB
Go
81 lines
4.3 KiB
Go
package metrics
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"runtime"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
var startedAt = time.Now()
|
|
var submitAccepted atomic.Uint64
|
|
var submitFailed atomic.Uint64
|
|
var submitDurationNanoseconds atomic.Uint64
|
|
|
|
type Snapshot struct {
|
|
UpstreamDesired int
|
|
UpstreamConnected int
|
|
DownstreamConnected int
|
|
SubmitWorkerUp bool
|
|
QueueAvailable bool
|
|
QueuePending int64
|
|
QueueLag int64
|
|
QueueOldestAgeSeconds float64
|
|
}
|
|
|
|
type SnapshotFunc func(context.Context) Snapshot
|
|
|
|
func ObserveSubmit(accepted bool, duration time.Duration) {
|
|
if accepted {
|
|
submitAccepted.Add(1)
|
|
} else {
|
|
submitFailed.Add(1)
|
|
}
|
|
submitDurationNanoseconds.Add(uint64(max(duration, 0)))
|
|
}
|
|
|
|
func Handler(load SnapshotFunc) http.Handler {
|
|
return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
|
|
if request.Method != http.MethodGet || request.URL.Path != "/metrics" {
|
|
response.WriteHeader(http.StatusNotFound)
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(request.Context(), time.Second)
|
|
defer cancel()
|
|
snapshot := Snapshot{}
|
|
if load != nil {
|
|
snapshot = load(ctx)
|
|
}
|
|
var memory runtime.MemStats
|
|
runtime.ReadMemStats(&memory)
|
|
accepted := submitAccepted.Load()
|
|
failed := submitFailed.Load()
|
|
count := accepted + failed
|
|
response.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
|
response.Header().Set("Cache-Control", "no-store")
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_process_uptime_seconds Gateway process uptime.\n# TYPE cmpp_gateway_process_uptime_seconds gauge\ncmpp_gateway_process_uptime_seconds %f\n", time.Since(startedAt).Seconds())
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_go_goroutines Current goroutine count.\n# TYPE cmpp_gateway_go_goroutines gauge\ncmpp_gateway_go_goroutines %d\n", runtime.NumGoroutine())
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_go_heap_alloc_bytes Current Go heap allocation.\n# TYPE cmpp_gateway_go_heap_alloc_bytes gauge\ncmpp_gateway_go_heap_alloc_bytes %d\n", memory.HeapAlloc)
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_submit_total Upstream submit attempts by bounded result.\n# TYPE cmpp_gateway_submit_total counter\ncmpp_gateway_submit_total{result=\"accepted\"} %d\ncmpp_gateway_submit_total{result=\"failed\"} %d\n", accepted, failed)
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_sum Total upstream submit duration.\n# TYPE cmpp_gateway_submit_duration_seconds_sum counter\ncmpp_gateway_submit_duration_seconds_sum %f\n", float64(submitDurationNanoseconds.Load())/float64(time.Second))
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_count Total measured upstream submits.\n# TYPE cmpp_gateway_submit_duration_seconds_count counter\ncmpp_gateway_submit_duration_seconds_count %d\n", count)
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_upstream_connections Desired and live supplier connections.\n# TYPE cmpp_gateway_upstream_connections gauge\ncmpp_gateway_upstream_connections{state=\"desired\"} %d\ncmpp_gateway_upstream_connections{state=\"connected\"} %d\n", snapshot.UpstreamDesired, snapshot.UpstreamConnected)
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_downstream_connections Authenticated client connections.\n# TYPE cmpp_gateway_downstream_connections gauge\ncmpp_gateway_downstream_connections %d\n", snapshot.DownstreamConnected)
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_up Whether the submit worker was initialized.\n# TYPE cmpp_gateway_submit_worker_up gauge\ncmpp_gateway_submit_worker_up %d\n", boolNumber(snapshot.SubmitWorkerUp))
|
|
if snapshot.QueueAvailable {
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_pending Pending entries owned by the consumer group.\n# TYPE cmpp_gateway_submit_queue_pending gauge\ncmpp_gateway_submit_queue_pending %d\n", snapshot.QueuePending)
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_lag Undelivered entries for the consumer group.\n# TYPE cmpp_gateway_submit_queue_lag gauge\ncmpp_gateway_submit_queue_lag %d\n", snapshot.QueueLag)
|
|
fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_oldest_pending_age_seconds Age of the oldest pending entry.\n# TYPE cmpp_gateway_submit_queue_oldest_pending_age_seconds gauge\ncmpp_gateway_submit_queue_oldest_pending_age_seconds %f\n", snapshot.QueueOldestAgeSeconds)
|
|
}
|
|
})
|
|
}
|
|
|
|
func boolNumber(value bool) int {
|
|
if value {
|
|
return 1
|
|
}
|
|
return 0
|
|
}
|