feat(engine, api): metrics through REST API (#5089)

engine/api/api_routes.go

@@ -3,6 +3,7 @@ package api
 import (
 	"net/http"
 
+	"github.com/ovh/cds/engine/api/observability"
 	"github.com/ovh/cds/engine/service"
 	"github.com/ovh/cds/sdk"
 	"github.com/ovh/cds/sdk/log"
@@ -136,6 +137,7 @@ func (api *API) InitRouter() {
 	r.Handle("/mon/db/migrate", ScopeNone(), r.GET(api.getMonDBStatusMigrateHandler, NeedAdmin(true)))
 	r.Handle("/mon/metrics", ScopeNone(), r.GET(service.GetPrometheustMetricsHandler(api), Auth(false)))
 	r.Handle("/mon/metrics/all", ScopeNone(), r.GET(service.GetMetricsHandler, Auth(false)))
+	r.HandlePrefix("/mon/metrics/detail/", ScopeNone(), r.GET(service.GetMetricHandler(observability.StatsHTTPExporter(), "/mon/metrics/detail/"), Auth(false)))
 	r.Handle("/mon/errors/{uuid}", ScopeNone(), r.GET(api.getErrorHandler, NeedAdmin(true)))
 	r.Handle("/mon/panic/{uuid}", ScopeNone(), r.GET(api.getPanicDumpHandler, Auth(false)))
 

engine/api/observability/http_exporter.go

@@ -0,0 +1,65 @@
+package observability
+
+import (
+	"reflect"
+	"time"
+
+	"go.opencensus.io/stats/view"
+)
+
+type HTTPExporter struct {
+	Views []HTTPExporterView `json:"views"`
+}
+
+type HTTPExporterView struct {
+	Name  string            `json:"name"`
+	Tags  map[string]string `json:"tags"`
+	Value float64           `json:"value"`
+	Date  time.Time         `json:"date"`
+}
+
+func (e *HTTPExporter) GetView(name string, tags map[string]string) *HTTPExporterView {
+	for i := range e.Views {
+		if e.Views[i].Name == name && reflect.DeepEqual(e.Views[i].Tags, tags) {
+			return &e.Views[i]
+		}
+	}
+	return nil
+}
+
+func (e *HTTPExporter) NewView(name string, tags map[string]string) *HTTPExporterView {
+	v := HTTPExporterView{
+		Name: name,
+		Tags: tags,
+	}
+	e.Views = append(e.Views, v)
+	return &v
+}
+
+func (e *HTTPExporter) ExportView(vd *view.Data) {
+	for _, row := range vd.Rows {
+		tags := make(map[string]string)
+		for _, t := range row.Tags {
+			tags[t.Key.Name()] = t.Value
+		}
+		view := e.GetView(vd.View.Name, tags)
+		if view == nil {
+			view = e.NewView(vd.View.Name, tags)
+		}
+		view.Record(row.Data)
+	}
+}
+
+func (v *HTTPExporterView) Record(data view.AggregationData) {
+	v.Date = time.Now()
+	switch x := data.(type) {
+	case *view.DistributionData:
+		v.Value = x.Mean
+	case *view.CountData:
+		v.Value = float64(x.Value)
+	case *view.SumData:
+		v.Value = x.Value
+	case *view.LastValueData:
+		v.Value = x.Value
+	}
+}

engine/api/observability/init.go

@@ -14,8 +14,9 @@ import (
 )
 
 var (
-	traceExporter trace.Exporter
-	statsExporter *prometheus.Exporter
+	traceExporter     trace.Exporter
+	statsExporter     *prometheus.Exporter
+	statsHTTPExporter *HTTPExporter
 )
 
 type service interface {
@@ -31,6 +32,10 @@ func StatsExporter() *prometheus.Exporter {
 	return statsExporter
 }
 
+func StatsHTTPExporter() *HTTPExporter {
+	return statsHTTPExporter
+}
+
 // Init the opencensus exporter
 func Init(ctx context.Context, cfg Configuration, s service) (context.Context, error) {
 	ctx = ContextWithTag(ctx,
@@ -70,6 +75,10 @@ func Init(ctx context.Context, cfg Configuration, s service) (context.Context, e
 		}
 		view.RegisterExporter(e)
 		statsExporter = e
+
+		he := new(HTTPExporter)
+		view.RegisterExporter(he)
+		statsHTTPExporter = he
 	}
 
 	return ctx, nil

engine/api/router.go

@@ -71,9 +71,9 @@ type HandlerConfigParam func(*service.HandlerConfig)
 type HandlerConfigFunc func(service.Handler, ...HandlerConfigParam) *service.HandlerConfig
 
 func (r *Router) pprofLabel(config map[string]*service.HandlerConfig, fn http.HandlerFunc) http.HandlerFunc {
-	return func(w http.ResponseWriter, r *http.Request) {
+	return func(w http.ResponseWriter, req *http.Request) {
 		var name = sdk.RandomString(12)
-		rc := config[r.Method]
+		rc := config[req.Method]
 		if rc != nil && rc.Handler != nil {
 			name = runtime.FuncForPC(reflect.ValueOf(rc.Handler).Pointer()).Name()
 			name = strings.Replace(name, ".func1", "", 1)
@@ -82,14 +82,14 @@ func (r *Router) pprofLabel(config map[string]*service.HandlerConfig, fn http.Ha
 		id := fmt.Sprintf("%d", sdk.GoroutineID())
 
 		labels := pprof.Labels(
-			"http-path", r.URL.Path,
+			"http-path", req.URL.Path,
 			"goroutine-id", id,
 			"goroutine-name", name+"-"+id,
 		)
-		ctx := pprof.WithLabels(r.Context(), labels)
+		ctx := pprof.WithLabels(req.Context(), labels)
 		pprof.SetGoroutineLabels(ctx)
-		r = r.WithContext(ctx)
-		fn(w, r)
+		req = req.WithContext(ctx)
+		fn(w, req)
 	}
 }
 
@@ -112,10 +112,10 @@ func (r *Router) recoverWrap(h http.HandlerFunc) http.HandlerFunc {
 				switch t := re.(type) {
 				case string:
 					err = errors.New(t)
-				case error:
-					err = re.(error)
 				case sdk.Error:
 					err = re.(sdk.Error)
+				case error:
+					err = re.(error)
 				default:
 					err = sdk.ErrUnknownError
 				}
@@ -250,6 +250,18 @@ func (r *Router) computeScopeDetails() {
 // Handle adds all handler for their specific verb in gorilla router for given uri
 func (r *Router) Handle(uri string, scope HandlerScope, handlers ...*service.HandlerConfig) {
 	uri = r.Prefix + uri
+	config, f := r.handle(uri, scope, handlers...)
+	r.Mux.Handle(uri, r.pprofLabel(config, r.compress(r.recoverWrap(f))))
+}
+
+func (r *Router) HandlePrefix(uri string, scope HandlerScope, handlers ...*service.HandlerConfig) {
+	uri = r.Prefix + uri
+	config, f := r.handle(uri, scope, handlers...)
+	r.Mux.PathPrefix(uri).HandlerFunc(r.pprofLabel(config, r.compress(r.recoverWrap(f))))
+}
+
+// Handle adds all handler for their specific verb in gorilla router for given uri
+func (r *Router) handle(uri string, scope HandlerScope, handlers ...*service.HandlerConfig) (map[string]*service.HandlerConfig, http.HandlerFunc) {
 	cfg := &service.RouterConfig{
 		Config: map[string]*service.HandlerConfig{},
 	}
@@ -385,7 +397,7 @@ func (r *Router) Handle(uri string, scope HandlerScope, handlers ...*service.Han
 				"route":         cleanURL,
 				"request_uri":   req.RequestURI,
 				"deprecated":    rc.IsDeprecated,
-			}, "[%d] | %s | END | %s [%s]", responseWriter.statusCode, req.Method, req.URL, rc.Name)
+			}, "%s | END   | %s [%s] | [%d]", req.Method, req.URL, rc.Name, responseWriter.statusCode)
 
 			observability.RecordFloat64(ctx, ServerLatency, float64(latency)/float64(time.Millisecond))
 			observability.Record(ctx, ServerRequestBytes, responseWriter.reqSize)
@@ -429,8 +441,7 @@ func (r *Router) Handle(uri string, scope HandlerScope, handlers ...*service.Han
 		deferFunc(ctx)
 	}
 
-	// The chain is http -> mux -> f -> recover -> wrap -> pprof -> opencensus -> http
-	r.Mux.Handle(uri, r.pprofLabel(cfg.Config, r.compress(r.recoverWrap(f))))
+	return cfg.Config, f
 }
 
 type asynchronousRequest struct {
@@ -627,7 +638,7 @@ func EnableTracing() HandlerConfigParam {
 
 // NotFoundHandler is called by default by Mux is any matching handler has been found
 func NotFoundHandler(w http.ResponseWriter, req *http.Request) {
-	service.WriteError(context.Background(), w, req, sdk.WithStack(sdk.ErrNotFound))
+	service.WriteError(context.Background(), w, req, sdk.NewError(sdk.ErrNotFound, fmt.Errorf("%s not found", req.URL.Path)))
 }
 
 // StatusPanic returns router status. If nbPanic > 30 -> Alert, if nbPanic > 0 -> Warn

engine/service/metrics.go

@@ -2,11 +2,13 @@ package service
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"os"
 	"runtime"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -198,3 +200,52 @@ func RegisterCommonMetricsView(ctx context.Context) {
 		})
 	})
 }
+
+func writeJSON(w http.ResponseWriter, i interface{}, statusCode int) error {
+	btes, _ := json.Marshal(i)
+	w.Header().Add("Content-Type", "application/json")
+	w.Header().Add("Content-Length", fmt.Sprintf("%d", len(btes)))
+	w.WriteHeader(statusCode)
+	_, err := w.Write(btes)
+	return sdk.WithStack(err)
+}
+
+func GetMetricHandler(exporter *observability.HTTPExporter, prefix string) func() Handler {
+	return func() Handler {
+		return func(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
+			view := strings.TrimPrefix(r.URL.Path, prefix)
+			formValues := r.URL.Query()
+			tags := make(map[string]string)
+			threshold := formValues.Get("threshold")
+			for k := range formValues {
+				if k != "threshold" {
+					tags[k] = formValues.Get(k)
+				}
+			}
+			log.Debug("GetMetricHandler> path: %s - tags: %v", view, tags)
+
+			if view == "" {
+				return writeJSON(w, exporter, http.StatusOK)
+			}
+
+			metricsView := exporter.GetView(view, tags)
+			if metricsView == nil {
+				return sdk.WithStack(sdk.ErrNotFound)
+			}
+
+			statusCode := http.StatusOK
+			if threshold != "" {
+				thresholdF, err := strconv.ParseFloat(threshold, 64)
+				if err != nil {
+					return sdk.WithStack(sdk.ErrWrongRequest)
+				}
+				if metricsView.Value >= thresholdF {
+					log.Error(context.Background(), "GetMetricHandler> %s threshold (%s) reached or exceeded : %v", metricsView.Name, threshold, metricsView.Value)
+					statusCode = 509 // Bandwidth Limit Exceeded
+				}
+			}
+
+			return writeJSON(w, metricsView, statusCode)
+		}
+	}
+}

go.sum

@@ -465,6 +465,7 @@ github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDf
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.1.0 h1:BQ53HtBmfOitExawJ6LokA4x8ov/z0SYYb0+HxJfRI8=
 github.com/prometheus/client_golang v1.1.0/go.mod h1:I1FGZT9+L76gKKOs5djB6ezCbFQP1xR9D75/vuwEF3g=
+github.com/prometheus/client_golang v1.5.1 h1:bdHYieyGlH+6OLEk2YQha8THib30KP0/yD0YH9m6xcA=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=