mirror of
https://github.com/OpenStaticFish/MicroServices.git
synced 2026-10-13 06:35:12 +00:00
Add production observability manifests
This commit is contained in:
@@ -5,6 +5,7 @@ resources:
|
||||
- webshare-api-externalsecret.yaml
|
||||
- site-analyzer-deployment.yaml
|
||||
- site-analyzer-service.yaml
|
||||
- site-analyzer-servicemonitor.yaml
|
||||
- site-analyzer-hpa.yaml
|
||||
- lightpanda-cdp-deployment.yaml
|
||||
- lightpanda-cdp-service.yaml
|
||||
@@ -14,7 +15,10 @@ resources:
|
||||
- lightpanda-mcp-hpa.yaml
|
||||
- scraper-deployment.yaml
|
||||
- scraper-service.yaml
|
||||
- scraper-servicemonitor.yaml
|
||||
- scraper-hpa.yaml
|
||||
- microservices-workload-prometheusrule.yaml
|
||||
- microservices-http-prometheusrule.yaml
|
||||
- traefik-middleware-strip-prefix.yaml
|
||||
- ingressroute-microservices.yaml
|
||||
images:
|
||||
|
||||
@@ -3,6 +3,8 @@ kind: Service
|
||||
metadata:
|
||||
name: lightpanda-cdp
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: lightpanda-cdp
|
||||
spec:
|
||||
type: ClusterIP
|
||||
selector:
|
||||
|
||||
@@ -3,6 +3,8 @@ kind: Service
|
||||
metadata:
|
||||
name: lightpanda-mcp
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: lightpanda-mcp
|
||||
spec:
|
||||
type: ClusterIP
|
||||
selector:
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
apiVersion: monitoring.coreos.com/v1
|
||||
kind: PrometheusRule
|
||||
metadata:
|
||||
name: microservices-http
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: microservices
|
||||
spec:
|
||||
groups:
|
||||
- name: microservices.http
|
||||
rules:
|
||||
- alert: MicroserviceHigh5xxRate
|
||||
expr: sum by (service) (rate(http_requests_total{service=~"scraper|site-analyzer",status=~"5.."}[5m])) / clamp_min(sum by (service) (rate(http_requests_total{service=~"scraper|site-analyzer"}[5m])), 0.001) > 0.05
|
||||
for: 10m
|
||||
labels:
|
||||
severity: warning
|
||||
annotations:
|
||||
summary: "Microservice has elevated 5xx errors"
|
||||
description: "{{ $labels.service }} has more than 5% 5xx responses over 10 minutes."
|
||||
- alert: MicroserviceHighP95Latency
|
||||
expr: histogram_quantile(0.95, sum by (service, le) (rate(http_request_duration_seconds_bucket{service=~"scraper|site-analyzer",path!~"/health|/ready"}[5m]))) > 2
|
||||
for: 10m
|
||||
labels:
|
||||
severity: warning
|
||||
annotations:
|
||||
summary: "Microservice has high p95 latency"
|
||||
description: "{{ $labels.service }} p95 HTTP latency is above 2 seconds over 10 minutes."
|
||||
@@ -0,0 +1,27 @@
|
||||
apiVersion: monitoring.coreos.com/v1
|
||||
kind: PrometheusRule
|
||||
metadata:
|
||||
name: microservices-workloads
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: microservices
|
||||
spec:
|
||||
groups:
|
||||
- name: microservices.workloads
|
||||
rules:
|
||||
- alert: MicroserviceDeploymentUnavailable
|
||||
expr: kube_deployment_status_replicas_unavailable{namespace="microservices",deployment=~"scraper|site-analyzer|lightpanda-cdp|lightpanda-mcp"} > 0
|
||||
for: 5m
|
||||
labels:
|
||||
severity: warning
|
||||
annotations:
|
||||
summary: "Microservice deployment has unavailable replicas"
|
||||
description: "{{ $labels.deployment }} has {{ $value }} unavailable replica(s) in namespace {{ $labels.namespace }}."
|
||||
- alert: MicroservicePodRestarting
|
||||
expr: increase(kube_pod_container_status_restarts_total{namespace="microservices",container=~"scraper|site-analyzer|lightpanda-cdp|lightpanda-mcp"}[15m]) > 2
|
||||
for: 5m
|
||||
labels:
|
||||
severity: warning
|
||||
annotations:
|
||||
summary: "Microservice pod is restarting"
|
||||
description: "{{ $labels.pod }}/{{ $labels.container }} restarted {{ $value }} times in the last 15 minutes."
|
||||
@@ -56,7 +56,7 @@ spec:
|
||||
periodSeconds: 10
|
||||
readinessProbe:
|
||||
httpGet:
|
||||
path: /health
|
||||
path: /ready
|
||||
port: 8080
|
||||
initialDelaySeconds: 2
|
||||
periodSeconds: 5
|
||||
|
||||
@@ -3,6 +3,8 @@ kind: Service
|
||||
metadata:
|
||||
name: scraper
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: scraper
|
||||
spec:
|
||||
type: ClusterIP
|
||||
selector:
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
apiVersion: monitoring.coreos.com/v1
|
||||
kind: ServiceMonitor
|
||||
metadata:
|
||||
name: scraper
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: scraper
|
||||
spec:
|
||||
selector:
|
||||
matchLabels:
|
||||
app: scraper
|
||||
endpoints:
|
||||
- port: http
|
||||
path: /metrics
|
||||
interval: 30s
|
||||
scrapeTimeout: 10s
|
||||
@@ -3,6 +3,8 @@ kind: Service
|
||||
metadata:
|
||||
name: site-analyzer
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: site-analyzer
|
||||
spec:
|
||||
type: ClusterIP
|
||||
selector:
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
apiVersion: monitoring.coreos.com/v1
|
||||
kind: ServiceMonitor
|
||||
metadata:
|
||||
name: site-analyzer
|
||||
namespace: microservices
|
||||
labels:
|
||||
app: site-analyzer
|
||||
spec:
|
||||
selector:
|
||||
matchLabels:
|
||||
app: site-analyzer
|
||||
endpoints:
|
||||
- port: http
|
||||
path: /metrics
|
||||
interval: 30s
|
||||
scrapeTimeout: 10s
|
||||
+201
-5
@@ -8,9 +8,11 @@ import (
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/PuerkitoBio/goquery"
|
||||
@@ -46,6 +48,8 @@ type ErrorResponse struct {
|
||||
|
||||
const webshareProxyCacheTTL = 5 * time.Minute
|
||||
|
||||
var metricsState = newMetricsState()
|
||||
|
||||
type webshareProxy struct {
|
||||
Username string `json:"username"`
|
||||
Password string `json:"password"`
|
||||
@@ -75,31 +79,218 @@ func main() {
|
||||
port = "8080"
|
||||
}
|
||||
|
||||
http.HandleFunc("/scrape", handleScrape)
|
||||
http.HandleFunc("/health", handleHealth)
|
||||
http.Handle("/scrape", instrumentHandler("/scrape", http.HandlerFunc(handleScrape)))
|
||||
http.Handle("/health", instrumentHandler("/health", http.HandlerFunc(handleHealth)))
|
||||
http.HandleFunc("/ready", handleHealth)
|
||||
http.HandleFunc("/metrics", handleMetrics)
|
||||
|
||||
log.Printf("scraper listening on :%s", port)
|
||||
log.Fatal(http.ListenAndServe(":"+port, nil))
|
||||
}
|
||||
|
||||
type metricsStateData struct {
|
||||
startedAt time.Time
|
||||
inFlight atomic.Int64
|
||||
proxyErrorsTotal atomic.Int64
|
||||
mu sync.Mutex
|
||||
httpRequests map[string]int64
|
||||
httpErrors map[string]int64
|
||||
httpDuration map[string]*histogram
|
||||
scraperJobs map[string]int64
|
||||
scraperDuration map[string]*histogram
|
||||
}
|
||||
|
||||
type histogram struct {
|
||||
buckets []float64
|
||||
counts []int64
|
||||
sum float64
|
||||
count int64
|
||||
}
|
||||
|
||||
type statusRecorder struct {
|
||||
http.ResponseWriter
|
||||
status int
|
||||
}
|
||||
|
||||
func newMetricsState() *metricsStateData {
|
||||
return &metricsStateData{
|
||||
startedAt: time.Now(),
|
||||
httpRequests: map[string]int64{},
|
||||
httpErrors: map[string]int64{},
|
||||
httpDuration: map[string]*histogram{},
|
||||
scraperJobs: map[string]int64{},
|
||||
scraperDuration: map[string]*histogram{},
|
||||
}
|
||||
}
|
||||
|
||||
func newHistogram() *histogram {
|
||||
buckets := []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}
|
||||
return &histogram{buckets: buckets, counts: make([]int64, len(buckets))}
|
||||
}
|
||||
|
||||
func (r *statusRecorder) WriteHeader(status int) {
|
||||
r.status = status
|
||||
r.ResponseWriter.WriteHeader(status)
|
||||
}
|
||||
|
||||
func instrumentHandler(path string, next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
metricsState.inFlight.Add(1)
|
||||
defer metricsState.inFlight.Add(-1)
|
||||
|
||||
recorder := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
|
||||
next.ServeHTTP(recorder, r)
|
||||
metricsState.recordHTTP(r.Method, path, recorder.status, time.Since(start).Seconds())
|
||||
})
|
||||
}
|
||||
|
||||
func (m *metricsStateData) recordHTTP(method string, path string, status int, seconds float64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
statusCode := fmt.Sprintf("%d", status)
|
||||
key := method + "|" + path + "|" + statusCode
|
||||
m.httpRequests[key]++
|
||||
if status >= 500 {
|
||||
m.httpErrors[key]++
|
||||
}
|
||||
histogramKey := method + "|" + path
|
||||
m.histogram(m.httpDuration, histogramKey).observe(seconds)
|
||||
}
|
||||
|
||||
func (m *metricsStateData) recordScrape(result string, seconds float64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.scraperJobs[result]++
|
||||
m.histogram(m.scraperDuration, result).observe(seconds)
|
||||
}
|
||||
|
||||
func (m *metricsStateData) histogram(histograms map[string]*histogram, key string) *histogram {
|
||||
h, ok := histograms[key]
|
||||
if !ok {
|
||||
h = newHistogram()
|
||||
histograms[key] = h
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func (h *histogram) observe(seconds float64) {
|
||||
h.count++
|
||||
h.sum += seconds
|
||||
for i, bucket := range h.buckets {
|
||||
if seconds <= bucket {
|
||||
h.counts[i]++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func handleHealth(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
fmt.Fprint(w, "ok")
|
||||
}
|
||||
|
||||
func handleMetrics(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
||||
metricsState.writePrometheus(w)
|
||||
}
|
||||
|
||||
func (m *metricsStateData) writePrometheus(w io.Writer) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
var runtimeStats runtime.MemStats
|
||||
runtime.ReadMemStats(&runtimeStats)
|
||||
|
||||
fmt.Fprintf(w, "# HELP http_requests_total Total HTTP requests by method, path, and status.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_requests_total counter\n")
|
||||
for key, count := range m.httpRequests {
|
||||
method, path, status := splitMetricKey(key)
|
||||
fmt.Fprintf(w, "http_requests_total{service=\"scraper\",method=\"%s\",path=\"%s\",status=\"%s\"} %d\n", method, path, status, count)
|
||||
}
|
||||
|
||||
fmt.Fprintf(w, "# HELP http_errors_total Total HTTP 5xx responses by method, path, and status.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_errors_total counter\n")
|
||||
for key, count := range m.httpErrors {
|
||||
method, path, status := splitMetricKey(key)
|
||||
fmt.Fprintf(w, "http_errors_total{service=\"scraper\",method=\"%s\",path=\"%s\",status=\"%s\"} %d\n", method, path, status, count)
|
||||
}
|
||||
|
||||
fmt.Fprintf(w, "# HELP http_requests_in_flight Current in-flight HTTP requests.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_requests_in_flight gauge\n")
|
||||
fmt.Fprintf(w, "http_requests_in_flight{service=\"scraper\"} %d\n", m.inFlight.Load())
|
||||
|
||||
fmt.Fprintf(w, "# HELP http_request_duration_seconds HTTP request duration by method and path.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_request_duration_seconds histogram\n")
|
||||
for key, histogram := range m.httpDuration {
|
||||
method, path, _ := splitMetricKey(key)
|
||||
writeHistogram(w, "http_request_duration_seconds", fmt.Sprintf("service=\"scraper\",method=\"%s\",path=\"%s\"", method, path), histogram)
|
||||
}
|
||||
|
||||
fmt.Fprintf(w, "# HELP scraper_jobs_total Total scrape jobs by result.\n")
|
||||
fmt.Fprintf(w, "# TYPE scraper_jobs_total counter\n")
|
||||
for result, count := range m.scraperJobs {
|
||||
fmt.Fprintf(w, "scraper_jobs_total{result=\"%s\"} %d\n", result, count)
|
||||
}
|
||||
|
||||
fmt.Fprintf(w, "# HELP scraper_job_duration_seconds Scrape job duration by result.\n")
|
||||
fmt.Fprintf(w, "# TYPE scraper_job_duration_seconds histogram\n")
|
||||
for result, histogram := range m.scraperDuration {
|
||||
writeHistogram(w, "scraper_job_duration_seconds", fmt.Sprintf("result=\"%s\"", result), histogram)
|
||||
}
|
||||
|
||||
fmt.Fprintf(w, "# HELP scraper_proxy_errors_total Total Webshare proxy selection/configuration errors.\n")
|
||||
fmt.Fprintf(w, "# TYPE scraper_proxy_errors_total counter\n")
|
||||
fmt.Fprintf(w, "scraper_proxy_errors_total %d\n", m.proxyErrorsTotal.Load())
|
||||
fmt.Fprintf(w, "# HELP process_start_time_seconds Start time of the process since unix epoch in seconds.\n")
|
||||
fmt.Fprintf(w, "# TYPE process_start_time_seconds gauge\n")
|
||||
fmt.Fprintf(w, "process_start_time_seconds %.0f\n", float64(m.startedAt.Unix()))
|
||||
fmt.Fprintf(w, "# HELP go_goroutines Number of goroutines that currently exist.\n")
|
||||
fmt.Fprintf(w, "# TYPE go_goroutines gauge\n")
|
||||
fmt.Fprintf(w, "go_goroutines %d\n", runtime.NumGoroutine())
|
||||
fmt.Fprintf(w, "# HELP go_memstats_alloc_bytes Number of bytes allocated and still in use.\n")
|
||||
fmt.Fprintf(w, "# TYPE go_memstats_alloc_bytes gauge\n")
|
||||
fmt.Fprintf(w, "go_memstats_alloc_bytes %d\n", runtimeStats.Alloc)
|
||||
}
|
||||
|
||||
func writeHistogram(w io.Writer, name string, labels string, histogram *histogram) {
|
||||
for i, bucket := range histogram.buckets {
|
||||
fmt.Fprintf(w, "%s_bucket{%s,le=\"%g\"} %d\n", name, labels, bucket, histogram.counts[i])
|
||||
}
|
||||
fmt.Fprintf(w, "%s_bucket{%s,le=\"+Inf\"} %d\n", name, labels, histogram.count)
|
||||
fmt.Fprintf(w, "%s_sum{%s} %.6f\n", name, labels, histogram.sum)
|
||||
fmt.Fprintf(w, "%s_count{%s} %d\n", name, labels, histogram.count)
|
||||
}
|
||||
|
||||
func splitMetricKey(key string) (string, string, string) {
|
||||
parts := strings.Split(key, "|")
|
||||
for len(parts) < 3 {
|
||||
parts = append(parts, "")
|
||||
}
|
||||
return parts[0], parts[1], parts[2]
|
||||
}
|
||||
|
||||
func handleScrape(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
result := "success"
|
||||
defer func() {
|
||||
metricsState.recordScrape(result, time.Since(start).Seconds())
|
||||
}()
|
||||
|
||||
if r.Method != http.MethodPost {
|
||||
result = "error"
|
||||
respondWithError(w, http.StatusMethodNotAllowed, "Method not allowed", "Only POST requests are supported")
|
||||
return
|
||||
}
|
||||
|
||||
var req ScrapeRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
result = "error"
|
||||
respondWithError(w, http.StatusBadRequest, "Invalid JSON", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if req.URL == "" {
|
||||
result = "error"
|
||||
respondWithError(w, http.StatusBadRequest, "Missing URL", "url field is required")
|
||||
return
|
||||
}
|
||||
@@ -109,29 +300,32 @@ func handleScrape(w http.ResponseWriter, r *http.Request) {
|
||||
format = "json"
|
||||
}
|
||||
if format != "md" && format != "json" {
|
||||
result = "error"
|
||||
respondWithError(w, http.StatusBadRequest, "Invalid format", "format must be 'md' or 'json'")
|
||||
return
|
||||
}
|
||||
|
||||
country := normalizeCountryCode(req.Country)
|
||||
if err := validateCountry(country); err != nil {
|
||||
result = "error"
|
||||
respondWithError(w, http.StatusBadRequest, "Invalid country", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
req.URL = normalizeURL(req.URL)
|
||||
result, err := scrapeURL(req.URL, format, country)
|
||||
scrapeResult, err := scrapeURL(req.URL, format, country)
|
||||
if err != nil {
|
||||
result = "error"
|
||||
respondWithError(w, http.StatusInternalServerError, "Scraping failed", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if format == "md" {
|
||||
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
|
||||
fmt.Fprint(w, result.Content)
|
||||
fmt.Fprint(w, scrapeResult.Content)
|
||||
} else {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(result)
|
||||
json.NewEncoder(w).Encode(scrapeResult)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -139,10 +333,12 @@ func scrapeURL(url, format string, country string) (*ScrapeResponse, error) {
|
||||
if country != "" {
|
||||
proxy, err := selectWebshareProxy(country)
|
||||
if err != nil {
|
||||
metricsState.proxyErrorsTotal.Add(1)
|
||||
return nil, err
|
||||
}
|
||||
transport, err := webshareProxyTransport(proxy)
|
||||
if err != nil {
|
||||
metricsState.proxyErrorsTotal.Add(1)
|
||||
return nil, err
|
||||
}
|
||||
return scrapeURLWithTransport(url, format, country, transport)
|
||||
|
||||
+171
-14
@@ -12,9 +12,11 @@ import (
|
||||
"net/url"
|
||||
"os"
|
||||
"os/signal"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
@@ -114,6 +116,7 @@ type app struct {
|
||||
}
|
||||
|
||||
type metrics struct {
|
||||
startedAt time.Time
|
||||
requestsTotal atomic.Int64
|
||||
analyzeRequestsTotal atomic.Int64
|
||||
analyzeRejectedTotal atomic.Int64
|
||||
@@ -121,6 +124,23 @@ type metrics struct {
|
||||
analyzeDurationCount atomic.Int64
|
||||
analyzeDurationNanos atomic.Int64
|
||||
activeAnalyses atomic.Int64
|
||||
mu sync.Mutex
|
||||
httpRequests map[string]int64
|
||||
httpErrors map[string]int64
|
||||
httpDuration map[string]*histogram
|
||||
analyzeDuration *histogram
|
||||
}
|
||||
|
||||
type histogram struct {
|
||||
buckets []float64
|
||||
counts []int64
|
||||
sum float64
|
||||
count int64
|
||||
}
|
||||
|
||||
type statusRecorder struct {
|
||||
http.ResponseWriter
|
||||
status int
|
||||
}
|
||||
|
||||
var errUnsafeTarget = errors.New("target resolves to a private or otherwise unsafe address")
|
||||
@@ -130,9 +150,9 @@ func main() {
|
||||
application := newApp(cfg)
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/analyze", application.handleAnalyze)
|
||||
mux.HandleFunc("/health", application.handleHealth)
|
||||
mux.HandleFunc("/ready", application.handleReady)
|
||||
mux.Handle("/analyze", application.instrumentHandler("/analyze", http.HandlerFunc(application.handleAnalyze)))
|
||||
mux.Handle("/health", application.instrumentHandler("/health", http.HandlerFunc(application.handleHealth)))
|
||||
mux.Handle("/ready", application.instrumentHandler("/ready", http.HandlerFunc(application.handleReady)))
|
||||
mux.HandleFunc("/metrics", application.handleMetrics)
|
||||
|
||||
server := &http.Server{
|
||||
@@ -176,7 +196,7 @@ func newApp(cfg config) *app {
|
||||
application := &app{
|
||||
config: cfg,
|
||||
semaphore: make(chan struct{}, cfg.maxConcurrentAnalyses),
|
||||
metrics: &metrics{},
|
||||
metrics: newMetrics(),
|
||||
}
|
||||
|
||||
transport := &http.Transport{
|
||||
@@ -203,6 +223,73 @@ func newApp(cfg config) *app {
|
||||
return application
|
||||
}
|
||||
|
||||
func newMetrics() *metrics {
|
||||
return &metrics{
|
||||
startedAt: time.Now(),
|
||||
httpRequests: map[string]int64{},
|
||||
httpErrors: map[string]int64{},
|
||||
httpDuration: map[string]*histogram{},
|
||||
analyzeDuration: newHistogram(),
|
||||
}
|
||||
}
|
||||
|
||||
func newHistogram() *histogram {
|
||||
buckets := []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}
|
||||
return &histogram{buckets: buckets, counts: make([]int64, len(buckets))}
|
||||
}
|
||||
|
||||
func (r *statusRecorder) WriteHeader(status int) {
|
||||
r.status = status
|
||||
r.ResponseWriter.WriteHeader(status)
|
||||
}
|
||||
|
||||
func (a *app) instrumentHandler(path string, next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
recorder := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
|
||||
next.ServeHTTP(recorder, r)
|
||||
a.metrics.recordHTTP(r.Method, path, recorder.status, time.Since(start).Seconds())
|
||||
})
|
||||
}
|
||||
|
||||
func (m *metrics) recordHTTP(method string, path string, status int, seconds float64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
statusCode := fmt.Sprintf("%d", status)
|
||||
key := method + "|" + path + "|" + statusCode
|
||||
m.httpRequests[key]++
|
||||
if status >= 500 {
|
||||
m.httpErrors[key]++
|
||||
}
|
||||
histogramKey := method + "|" + path
|
||||
m.histogram(m.httpDuration, histogramKey).observe(seconds)
|
||||
}
|
||||
|
||||
func (m *metrics) recordAnalyzeDuration(seconds float64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.analyzeDuration.observe(seconds)
|
||||
}
|
||||
|
||||
func (m *metrics) histogram(histograms map[string]*histogram, key string) *histogram {
|
||||
h, ok := histograms[key]
|
||||
if !ok {
|
||||
h = newHistogram()
|
||||
histograms[key] = h
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func (h *histogram) observe(seconds float64) {
|
||||
h.count++
|
||||
h.sum += seconds
|
||||
for i, bucket := range h.buckets {
|
||||
if seconds <= bucket {
|
||||
h.counts[i]++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func loadConfig() config {
|
||||
analysisTimeout := durationEnv("ANALYSIS_TIMEOUT", 15*time.Second)
|
||||
fetchTimeout := durationEnv("FETCH_TIMEOUT", 10*time.Second)
|
||||
@@ -285,32 +372,100 @@ func (a *app) handleReady(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func (a *app) handleMetrics(w http.ResponseWriter, r *http.Request) {
|
||||
a.metrics.requestsTotal.Add(1)
|
||||
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
||||
a.metrics.writePrometheus(w, a.config)
|
||||
}
|
||||
|
||||
durationSeconds := float64(a.metrics.analyzeDurationNanos.Load()) / float64(time.Second)
|
||||
func (m *metrics) writePrometheus(w io.Writer, cfg config) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
var runtimeStats runtime.MemStats
|
||||
runtime.ReadMemStats(&runtimeStats)
|
||||
|
||||
durationSeconds := float64(m.analyzeDurationNanos.Load()) / float64(time.Second)
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_requests_total Total HTTP requests received.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_requests_total counter\n")
|
||||
fmt.Fprintf(w, "site_analyzer_requests_total %d\n", a.metrics.requestsTotal.Load())
|
||||
fmt.Fprintf(w, "site_analyzer_requests_total %d\n", m.requestsTotal.Load())
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_analyze_requests_total Total analyze requests received.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_analyze_requests_total counter\n")
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_requests_total %d\n", a.metrics.analyzeRequestsTotal.Load())
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_requests_total %d\n", m.analyzeRequestsTotal.Load())
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_analyze_rejected_total Analyze requests rejected because the pod is saturated.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_analyze_rejected_total counter\n")
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_rejected_total %d\n", a.metrics.analyzeRejectedTotal.Load())
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_rejected_total %d\n", m.analyzeRejectedTotal.Load())
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_analyze_errors_total Analyze requests that failed before producing a response.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_analyze_errors_total counter\n")
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_errors_total %d\n", a.metrics.analyzeErrorsTotal.Load())
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_errors_total %d\n", m.analyzeErrorsTotal.Load())
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_active_analyses Current number of in-flight analyses.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_active_analyses gauge\n")
|
||||
fmt.Fprintf(w, "site_analyzer_active_analyses %d\n", a.metrics.activeAnalyses.Load())
|
||||
fmt.Fprintf(w, "site_analyzer_active_analyses %d\n", m.activeAnalyses.Load())
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_max_concurrent_analyses Configured maximum in-flight analyses per pod.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_max_concurrent_analyses gauge\n")
|
||||
fmt.Fprintf(w, "site_analyzer_max_concurrent_analyses %d\n", a.config.maxConcurrentAnalyses)
|
||||
fmt.Fprintf(w, "site_analyzer_max_concurrent_analyses %d\n", cfg.maxConcurrentAnalyses)
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_analyze_duration_seconds Total analyze request duration.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_analyze_duration_seconds summary\n")
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_duration_seconds_sum %.6f\n", durationSeconds)
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_duration_seconds_count %d\n", a.metrics.analyzeDurationCount.Load())
|
||||
fmt.Fprintf(w, "site_analyzer_analyze_duration_seconds_count %d\n", m.analyzeDurationCount.Load())
|
||||
|
||||
fmt.Fprintf(w, "# HELP http_requests_total Total HTTP requests by method, path, and status.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_requests_total counter\n")
|
||||
for key, count := range m.httpRequests {
|
||||
method, path, status := splitMetricKey(key)
|
||||
fmt.Fprintf(w, "http_requests_total{service=\"site-analyzer\",method=\"%s\",path=\"%s\",status=\"%s\"} %d\n", method, path, status, count)
|
||||
}
|
||||
fmt.Fprintf(w, "# HELP http_errors_total Total HTTP 5xx responses by method, path, and status.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_errors_total counter\n")
|
||||
for key, count := range m.httpErrors {
|
||||
method, path, status := splitMetricKey(key)
|
||||
fmt.Fprintf(w, "http_errors_total{service=\"site-analyzer\",method=\"%s\",path=\"%s\",status=\"%s\"} %d\n", method, path, status, count)
|
||||
}
|
||||
fmt.Fprintf(w, "# HELP http_requests_in_flight Current in-flight HTTP analyze requests.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_requests_in_flight gauge\n")
|
||||
fmt.Fprintf(w, "http_requests_in_flight{service=\"site-analyzer\"} %d\n", m.activeAnalyses.Load())
|
||||
fmt.Fprintf(w, "# HELP http_request_duration_seconds HTTP request duration by method and path.\n")
|
||||
fmt.Fprintf(w, "# TYPE http_request_duration_seconds histogram\n")
|
||||
for key, histogram := range m.httpDuration {
|
||||
method, path, _ := splitMetricKey(key)
|
||||
writeHistogram(w, "http_request_duration_seconds", fmt.Sprintf("service=\"site-analyzer\",method=\"%s\",path=\"%s\"", method, path), histogram)
|
||||
}
|
||||
fmt.Fprintf(w, "# HELP site_analyzer_duration_seconds Analyze request duration.\n")
|
||||
fmt.Fprintf(w, "# TYPE site_analyzer_duration_seconds histogram\n")
|
||||
writeHistogram(w, "site_analyzer_duration_seconds", "", m.analyzeDuration)
|
||||
fmt.Fprintf(w, "# HELP process_start_time_seconds Start time of the process since unix epoch in seconds.\n")
|
||||
fmt.Fprintf(w, "# TYPE process_start_time_seconds gauge\n")
|
||||
fmt.Fprintf(w, "process_start_time_seconds %.0f\n", float64(m.startedAt.Unix()))
|
||||
fmt.Fprintf(w, "# HELP go_goroutines Number of goroutines that currently exist.\n")
|
||||
fmt.Fprintf(w, "# TYPE go_goroutines gauge\n")
|
||||
fmt.Fprintf(w, "go_goroutines %d\n", runtime.NumGoroutine())
|
||||
fmt.Fprintf(w, "# HELP go_memstats_alloc_bytes Number of bytes allocated and still in use.\n")
|
||||
fmt.Fprintf(w, "# TYPE go_memstats_alloc_bytes gauge\n")
|
||||
fmt.Fprintf(w, "go_memstats_alloc_bytes %d\n", runtimeStats.Alloc)
|
||||
}
|
||||
|
||||
func writeHistogram(w io.Writer, name string, labels string, histogram *histogram) {
|
||||
labelPrefix := ""
|
||||
if labels != "" {
|
||||
labelPrefix = labels + ","
|
||||
}
|
||||
for i, bucket := range histogram.buckets {
|
||||
fmt.Fprintf(w, "%s_bucket{%sle=\"%g\"} %d\n", name, labelPrefix, bucket, histogram.counts[i])
|
||||
}
|
||||
fmt.Fprintf(w, "%s_bucket{%sle=\"+Inf\"} %d\n", name, labelPrefix, histogram.count)
|
||||
if labels == "" {
|
||||
fmt.Fprintf(w, "%s_sum %.6f\n", name, histogram.sum)
|
||||
fmt.Fprintf(w, "%s_count %d\n", name, histogram.count)
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(w, "%s_sum{%s} %.6f\n", name, labels, histogram.sum)
|
||||
fmt.Fprintf(w, "%s_count{%s} %d\n", name, labels, histogram.count)
|
||||
}
|
||||
|
||||
func splitMetricKey(key string) (string, string, string) {
|
||||
parts := strings.Split(key, "|")
|
||||
for len(parts) < 3 {
|
||||
parts = append(parts, "")
|
||||
}
|
||||
return parts[0], parts[1], parts[2]
|
||||
}
|
||||
|
||||
func (a *app) handleAnalyze(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -318,8 +473,10 @@ func (a *app) handleAnalyze(w http.ResponseWriter, r *http.Request) {
|
||||
a.metrics.requestsTotal.Add(1)
|
||||
a.metrics.analyzeRequestsTotal.Add(1)
|
||||
defer func() {
|
||||
duration := time.Since(start)
|
||||
a.metrics.analyzeDurationCount.Add(1)
|
||||
a.metrics.analyzeDurationNanos.Add(time.Since(start).Nanoseconds())
|
||||
a.metrics.analyzeDurationNanos.Add(duration.Nanoseconds())
|
||||
a.metrics.recordAnalyzeDuration(duration.Seconds())
|
||||
}()
|
||||
|
||||
if r.Method != http.MethodPost {
|
||||
|
||||
Reference in New Issue
Block a user