DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Added profiler to apigw

Browse Source
This commit is contained in:
Peter Grlica
2022-02-17 16:05:23 +01:00
parent 046855cb30
commit 476084a3bf
32 changed files with 1865 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/boot_levels.go
+6 -1
View File
@@ -478,7 +478,7 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) {
// will most likely be merged in the future
err = cmpService.Initialize(ctx, app.Log, app.Store, cmpService.Config{
ActionLog: app.Opt.ActionLog,
Discovery: app.Opt.Discovery,
Discovery: app.Opt.Discovery,
Storage: app.Opt.ObjStore,
UserFinder: sysService.DefaultUser,
})
@@ -602,6 +602,7 @@ func (app *CortezaApp) Activate(ctx context.Context) (err error) {
updateFederationSettings(app.Opt.Federation, sysService.CurrentSettings)
updateAuthSettings(app.AuthService, sysService.CurrentSettings)
updatePasswdSettings(app.Opt.Auth, sysService.CurrentSettings)
updateApigwSettings(app.Opt.Apigw, sysService.CurrentSettings)
sysService.DefaultSettings.Register("auth.", func(ctx context.Context, current interface{}, set types.SettingValueSet) {
appSettings, is := current.(*types.AppSettings)
if !is {
@@ -720,6 +721,10 @@ func updatePasswdSettings(opt options.AuthOpt, current *types.AppSettings) {
current.Auth.Internal.PasswordConstraints.PasswordSecurity = opt.PasswordSecurity
}
func updateApigwSettings(opt options.ApigwOpt, current *types.AppSettings) {
current.Apigw.ProfilerEnabled = opt.ProfilerEnabled
}
// Checks if discovery is enabled in the options
func updateDiscoverySettings(opt options.DiscoveryOpt, current *types.AppSettings) {
current.Discovery.Enabled = opt.Enabled
app/options/apigw.cue
+4
View File
@@ -25,6 +25,10 @@ apigw: schema.#optionsGroup & {
type: "bool"
description: "Enable extra logging"
}
profiler_enabled: {
type: "bool"
description: "Enable profiler"
}
log_request_body: {
type: "bool"
description: "Enable incoming request body output in logs"
automation/automation/http_request_handler.go
+15 -1
View File
@@ -9,6 +9,7 @@ import (
"net/http"
"net/url"
"strings"
"time"
"github.com/cortezaproject/corteza-server/pkg/version"
)
@@ -34,6 +35,8 @@ func (h httpRequestHandler) send(ctx context.Context, args *httpRequestSendArgs)
rsp *http.Response
)
// spew.Dump("TIMEOUT", args.Timeout)
r = &httpRequestSendResults{}
req, err = h.makeRequest(ctx, args)
@@ -41,7 +44,16 @@ func (h httpRequestHandler) send(ctx context.Context, args *httpRequestSendArgs)
return nil, err
}
rsp, err = http.DefaultClient.Do(req)
// spew.Dump("ERR, REQ", err)
// spew.Dump(httputil.DumpRequestOut(req, false))
client := http.DefaultClient
client.Timeout = 5 * time.Minute
rsp, err = client.Do(req)
// spew.Dump("ERR", err)
// spew.Dump("RESP")
// spew.Dump(httputil.DumpResponse(rsp, true))
if err != nil {
return
}
@@ -129,6 +141,8 @@ func (h httpRequestHandler) makeRequest(ctx context.Context, args *httpRequestSe
args.Url = purl.String()
}
// spew.Dump("CTX", ctx)
req, err = http.NewRequestWithContext(ctx, args.Method, args.Url, args.bodyStream)
if err != nil {
return nil, err
pkg/apigw/ctx/ctx.go
+15
View File
@@ -9,6 +9,7 @@ import (
type ContextKey string
const ContextKeyScope ContextKey = "scope"
const ContextKeyProfiler ContextKey = "profiler"
func ScopeToContext(ctx context.Context, s *types.Scp) context.Context {
return context.WithValue(ctx, ContextKeyScope, s)
@@ -23,3 +24,17 @@ func ScopeFromContext(ctx context.Context) (ss *types.Scp) {
return s.(*types.Scp)
}
func ProfilerToContext(ctx context.Context, h interface{}) context.Context {
return context.WithValue(ctx, ContextKeyProfiler, h)
}
func ProfilerFromContext(ctx context.Context) (h interface{}) {
hh := ctx.Value(ContextKeyProfiler)
if hh == nil {
return nil
}
return hh
}
pkg/apigw/filter/prefilter.go
+64
View File
@@ -6,7 +6,10 @@ import (
"errors"
"fmt"
"net/http"
"time"
agctx "github.com/cortezaproject/corteza-server/pkg/apigw/ctx"
prf "github.com/cortezaproject/corteza-server/pkg/apigw/profiler"
"github.com/cortezaproject/corteza-server/pkg/apigw/types"
pe "github.com/cortezaproject/corteza-server/pkg/errors"
"github.com/cortezaproject/corteza-server/pkg/expr"
@@ -36,6 +39,10 @@ type (
Expr string `json:"expr"`
}
}
profiler struct {
types.FilterMeta
}
)
func NewHeader() (v *header) {
@@ -205,3 +212,60 @@ func (qp *queryParam) Handler() types.HandlerFunc {
return nil
}
}
func NewProfiler() (pp *profiler) {
pp = &profiler{}
pp.Name = "profiler"
pp.Label = "Profiler"
pp.Kind = types.PreFilter
return
}
func (pr profiler) New() types.Handler {
return NewProfiler()
}
func (pr profiler) String() string {
return fmt.Sprintf("apigw filter %s (%s)", pr.Name, pr.Label)
}
func (pr profiler) Meta() types.FilterMeta {
return pr.FilterMeta
}
func (pr *profiler) Merge(params []byte) (types.Handler, error) {
return pr, nil
}
func (pr *profiler) Handler() types.HandlerFunc {
return func(rw http.ResponseWriter, r *http.Request) (err error) {
var (
ctx = r.Context()
scope = agctx.ScopeFromContext(ctx)
)
if scope.Opts().ProfilerEnabled {
// profiler enabled overrides any profiling prefilter
// the hit is registered on lower level
return
}
var (
n = time.Now()
hit = agctx.ProfilerFromContext(r.Context())
)
if hit == nil {
return
}
hit.(*prf.Hit).Ts = &n
hit.(*prf.Hit).R = scope.Request()
r = r.WithContext(agctx.ProfilerToContext(r.Context(), hit))
return
}
}
pkg/apigw/filter/processer.go
-1
View File
@@ -97,7 +97,6 @@ func (h workflow) Handler() types.HandlerFunc {
ctx = r.Context()
scope = agctx.ScopeFromContext(ctx)
)
// cleanup scope for wf
scp := filterScope(scope, "opts")
pkg/apigw/helpers.go
+37 -1
View File
@@ -3,6 +3,8 @@ package apigw
import (
"net/http"
"github.com/cortezaproject/corteza-server/pkg/apigw/profiler"
h "github.com/cortezaproject/corteza-server/pkg/http"
"github.com/cortezaproject/corteza-server/pkg/options"
)
@@ -10,8 +12,10 @@ const (
devHelperResponseBody string = `Hey developer!`
)
func helperDefaultResponse(opt *options.ApigwOpt) http.HandlerFunc {
func helperDefaultResponse(opt *options.ApigwOpt, pr *profiler.Profiler) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
addToProfiler(opt, pr, r, http.StatusNotFound)
if opt.LogEnabled {
// Say something friendly when logging is enabled
http.Error(w, devHelperResponseBody, http.StatusTeapot)
@@ -21,3 +25,35 @@ func helperDefaultResponse(opt *options.ApigwOpt) http.HandlerFunc {
}
}
}
func helperMethodNotAllowed(opt *options.ApigwOpt, pr *profiler.Profiler) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
addToProfiler(opt, pr, r, http.StatusMethodNotAllowed)
if opt.LogEnabled {
// Say something friendly when logging is enabled
http.Error(w, devHelperResponseBody, http.StatusTeapot)
} else {
// Default 405 response
http.Error(w, "", http.StatusMethodNotAllowed)
}
}
}
func addToProfiler(opt *options.ApigwOpt, pr *profiler.Profiler, r *http.Request, status int) {
if !opt.ProfilerEnabled {
return
}
// add to profiler
ar, err := h.NewRequest(r)
if err != nil {
panic(err)
}
h := pr.Hit(ar)
h.Status = status
pr.Push(h)
}
pkg/apigw/pipeline/chain.go
-19
View File
@@ -1,19 +0,0 @@
package pipeline
import (
"net/http"
"github.com/go-chi/chi/v5"
)
type (
chiHandlerChain struct {
chain []func(http.Handler) http.Handler
}
)
func (cc chiHandlerChain) Handler() (h http.Handler) {
return chi.
Chain(cc.chain...).
HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {})
}
pkg/apigw/pipeline/chain/default.go
+38
View File
@@ -0,0 +1,38 @@
package chain
import (
"net/http"
"github.com/go-chi/chi/v5"
)
type (
middleware []func(http.Handler) http.Handler
ChainHandler interface {
Handler() http.Handler
Chain(middleware)
}
chiHandlerChainDefault struct {
chain middleware
}
)
func NewDefault() *chiHandlerChainDefault {
return &chiHandlerChainDefault{}
}
func (cc *chiHandlerChainDefault) Chain(m middleware) {
cc.chain = m
}
func (cc *chiHandlerChainDefault) Handler() http.Handler {
return chiHandler(cc.chain)
}
func chiHandler(cc middleware) http.Handler {
return chi.
Chain(cc...).
HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {})
}
pkg/apigw/pipeline/chain/profiler.go
+29
View File
@@ -0,0 +1,29 @@
package chain
import (
"net/http"
"github.com/cortezaproject/corteza-server/pkg/apigw/profiler"
)
type (
chiHandlerChainProfiler struct {
chain middleware
}
)
func NewProfiler() *chiHandlerChainProfiler {
return &chiHandlerChainProfiler{}
}
func (cc *chiHandlerChainProfiler) Chain(m middleware) {
cc.chain = m
}
func (cc *chiHandlerChainProfiler) Handler() http.Handler {
// wrap handlers around the chain so we get some profiling info
cc.chain = append([]func(http.Handler) http.Handler{profiler.StartHandler}, cc.chain...)
cc.chain = append(cc.chain, profiler.FinishHandler)
return chiHandler(cc.chain)
}
pkg/apigw/pipeline/pipeline.go
+6 -6
View File
@@ -7,6 +7,7 @@ import (
"time"
actx "github.com/cortezaproject/corteza-server/pkg/apigw/ctx"
"github.com/cortezaproject/corteza-server/pkg/apigw/pipeline/chain"
"github.com/cortezaproject/corteza-server/pkg/apigw/types"
"github.com/cortezaproject/corteza-server/pkg/auth"
"go.uber.org/zap"
@@ -25,18 +26,20 @@ type (
Pl struct {
workers workerSet
ch chain.ChainHandler
err types.ErrorHandlerFunc
log *zap.Logger
async bool
}
)
func NewPipeline(log *zap.Logger) *Pl {
func NewPipeline(log *zap.Logger, ch chain.ChainHandler) *Pl {
var (
defaultErrorHandler = types.NewDefaultErrorHandler(log)
)
return &Pl{
ch: ch,
log: log,
err: defaultErrorHandler.Handler(),
}
@@ -65,11 +68,8 @@ func (pp *Pl) Add(w *Worker) {
// that handles filter groups
func (pp *Pl) Handler() http.Handler {
// use the chi implementation of chains
chiChain := chiHandlerChain{
chain: pp.makeMiddleware(pp.workers...),
}
return chiChain.Handler()
pp.ch.Chain(pp.makeMiddleware(pp.workers...))
return pp.ch.Handler()
}
// makeMiddleware creates a list of handlers from workers
pkg/apigw/pipeline/pipeline_test.go
+2 -1
View File
@@ -6,6 +6,7 @@ import (
"net/http/httptest"
"testing"
"github.com/cortezaproject/corteza-server/pkg/apigw/pipeline/chain"
"github.com/cortezaproject/corteza-server/pkg/apigw/types"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
@@ -16,7 +17,7 @@ var (
)
func NewPl() *Pl {
return NewPipeline(zap.NewNop())
return NewPipeline(zap.NewNop(), chain.NewDefault())
}
func Test_pipelineAdd(t *testing.T) {
pkg/apigw/profiler/profiler.go
+301
View File
@@ -0,0 +1,301 @@
package profiler
import (
"encoding/base64"
"fmt"
"net/http"
"sort"
"time"
h "github.com/cortezaproject/corteza-server/pkg/http"
"github.com/cortezaproject/corteza-server/system/types"
)
const (
// default fallback on amount of items
FILTER_NUM_ITEMS = 20
// default fallback on amount of aggregated items
FILTER_NUM_AGG_ITEMS = 10
)
var sortAggFields = []string{"path", "count", "size_min", "size_max", "size_avg", "time_min", "time_max", "time_avg"}
var sortRouteFields = []string{"time_start", "time_finish", "time_duration"}
type (
Hits map[string][]*Hit
Profiler struct {
l Hits
}
Hit struct {
ID string
Status int
Route uint64
R *h.Request
Ts *time.Time
Tf *time.Time
D *time.Duration
}
CtxHit []*Stage
Stage struct {
Name string
Ts *time.Time
Tf *time.Time
}
Sort struct {
Hit string
Path string
Size uint64
Before string
}
)
func New() *Profiler {
return &Profiler{make(Hits)}
}
func (p *Profiler) Hit(r *h.Request) (h *Hit) {
var (
n = time.Now()
)
h = &Hit{"", http.StatusOK, 0, r, &n, nil, nil}
h.generateID()
return
}
func (p *Profiler) Push(h *Hit) (id string) {
if h.Tf == nil {
n := time.Now()
d := n.Sub(*h.Ts)
h.Tf = &n
h.D = &d
}
h.generateID()
id = p.id(h.R)
p.l[id] = append(p.l[id], h)
return
}
func (p *Profiler) Dump(s Sort) Hits {
ll := p.l.Filter(func(k string, v *Hit) bool {
var b bool = true
if s.Path != "" && v.R.URL.Path != s.Path {
b = false
}
if s.Hit != "" && v.ID != s.Hit {
b = false
}
return b
})
return ll
}
func (p *Profiler) id(r *h.Request) string {
return r.URL.Path
}
func (h *Hit) generateID() {
h.ID = base64.URLEncoding.EncodeToString([]byte(fmt.Sprintf("%s_%d", h.R.URL.Path, h.Ts.UnixNano())))
}
func (s Hits) Filter(fn func(k string, v *Hit) bool) Hits {
ss := Hits{}
for k, v := range s {
for _, vv := range v {
if !fn(k, vv) {
continue
}
ss[k] = append(ss[k], vv)
}
}
return ss
}
func StartHandler(next http.Handler) http.Handler {
return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
// add some info to context
next.ServeHTTP(rw, r)
})
}
func FinishHandler(next http.Handler) http.Handler {
return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
next.ServeHTTP(rw, r)
})
}
func (h Hits) Len() int {
return h.Len()
}
func (h Hits) Less(i, j int) bool {
return false
}
func (h Hits) Swap(i, j int) {
return
}
func SortAggregation(list *types.ApigwProfilerAggregationSet, filter *types.ApigwProfilerFilter) {
for _, ff := range sortAggFields {
fe := filter.Sort.Get(ff)
if fe == nil {
continue
}
if filter.Sort.Get(ff).Descending {
sort.Sort(sort.Reverse(getSortType(ff, list)))
break
}
sort.Sort(getSortType(ff, list))
break
}
}
func FilterAggregation(list *types.ApigwProfilerAggregationSet, filter *types.ApigwProfilerFilter) {
var (
dec string = ""
i uint = 0
b = filter.Before == ""
)
if filter.Limit == 0 {
filter.Limit = FILTER_NUM_AGG_ITEMS
}
dec, _ = decodeRoutePath(filter.Before)
*list, _ = list.Filter(func(apa *types.ApigwProfilerAggregation) (bool, error) {
// after a specific hit and inside the limits
if b && i < filter.Limit {
i++
filter.Next = encodeRoutePath(apa.Path)
return true, nil
}
// after the specific hit check
if dec != "" && b == false {
b = apa.Path == dec
}
return false, nil
})
return
}
func FilterHits(list *types.ApigwProfilerHitSet, filter *types.ApigwProfilerFilter) {
var (
i uint = 0
b = filter.Before == ""
)
if filter.Limit == 0 {
filter.Limit = FILTER_NUM_ITEMS
}
*list, _ = list.Filter(func(aph *types.ApigwProfilerHit) (bool, error) {
// after a specific hit and inside the limits
if b && i < filter.Limit {
i++
filter.Next = aph.ID
return true, nil
}
// after the specific hit check
if filter.Before != "" && b == false {
b = aph.ID == filter.Before
}
return false, nil
})
return
}
func SortHits(list *types.ApigwProfilerHitSet, filter *types.ApigwProfilerFilter) {
for _, ff := range sortRouteFields {
fe := filter.Sort.Get(ff)
if fe == nil {
continue
}
if filter.Sort.Get(ff).Descending {
sort.Sort(sort.Reverse(getSortTypeHit(ff, list)))
break
}
sort.Sort(getSortTypeHit(ff, list))
break
}
}
func getSortType(s string, list *types.ApigwProfilerAggregationSet) sort.Interface {
switch s {
case "path":
return types.ByPath(*list)
case "count":
return types.ByCount(*list)
case "size_min":
return types.BySizeMin(*list)
case "size_max":
return types.BySizeMax(*list)
case "size_avg":
return types.BySizeAvg(*list)
case "time_min":
return types.ByTimeMin(*list)
case "time_max":
return types.ByTimeMax(*list)
case "time_avg":
return types.ByTimeAvg(*list)
default:
return types.ByCount(*list)
}
}
func getSortTypeHit(s string, list *types.ApigwProfilerHitSet) sort.Interface {
switch s {
case "time_start":
return types.BySTime(*list)
case "time_finish":
return types.ByFTime(*list)
case "time_duration":
return types.ByDuration(*list)
default:
return types.BySTime(*list)
}
}
func encodeRoutePath(p string) string {
return base64.URLEncoding.EncodeToString([]byte(p))
}
func decodeRoutePath(p string) (s string, err error) {
b, err := base64.URLEncoding.DecodeString(p)
s = string(b)
return
}
pkg/apigw/profiler/profiler_test.go
+157
View File
@@ -0,0 +1,157 @@
package profiler
// import (
// "net/http/httptest"
// "strings"
// "testing"
// "time"
// "github.com/cortezaproject/corteza-server/pkg/filter"
// h "github.com/cortezaproject/corteza-server/pkg/http"
// "github.com/cortezaproject/corteza-server/system/types"
// "github.com/davecgh/go-spew/spew"
// )
// func Test_Profiler(t *testing.T) {
// // var (
// // p = Profiler{
// // l: make(map[string]*Hit),
// // }
// // req = require.New(t)
// // )
// // // in goes the h.Request
// // rr, err := http.NewRequest("POST", "/foo", strings.NewReader(`foo`))
// // req.NoError(err)
// // hh, err := h.NewRequest(rr)
// // req.NoError(err)
// // // need to create an internal profiling struct to hold the request?
// // p.Push(hh)
// // spew.Dump(p.l)
// t.Fail()
// }
// func Test_Profiler2(t *testing.T) {
// // types:
// // + list of hits, aggregated by endpoint (ie /parse/js)
// // - list of hits for a specific endpoint
// // - list of hits for a specific registered route
// now := time.Now()
// then := time.Date(2022, time.March, 1, 1, 1, 1, 0, time.UTC)
// later := time.Date(2022, time.March, 1, 1, 1, 1, 30, time.UTC)
// pp := New()
// hr, _ := h.NewRequest(httptest.NewRequest("POST", "/foo", strings.NewReader(`foo`)))
// hr2, _ := h.NewRequest(httptest.NewRequest("GET", "/sometotherpath", strings.NewReader(`foo`)))
// pp.Push(&Hit{R: hr, Ts: &then})
// pp.Push(&Hit{R: hr, Ts: &then})
// pp.Push(&Hit{R: hr, Ts: &now})
// pp.Push(&Hit{R: hr, Ts: &now})
// pp.Push(&Hit{R: hr2, Ts: &now})
// pp.Push(&Hit{R: hr2, Ts: &later})
// pp.Push(&Hit{R: hr2, Ts: &later})
// pp.Push(&Hit{R: hr2, Ts: &later})
// pp.Push(&Hit{R: hr2, Ts: &later})
// pp.Push(&Hit{R: hr2, Ts: &later})
// pp.Push(&Hit{R: hr2, Ts: &later})
// pp.Push(&Hit{R: hr2, Ts: &later})
// var err error
// f := types.ApigwProfilerFilter{}
// if f.Sorting, err = filter.NewSorting("count DESC"); err != nil {
// spew.Dump(err)
// }
// list := pp.Dump(Sort{})
// // list of aggregations
// // - keep showing, just refresh
// // list := pp.Dump(Sort{Before: &later, Size: 3})
// // spew.Dump(list)
// // list = list.Filter(func(k string, v *Hit) bool {
// // if v.R.URL.Path == "/sometotherpath" {
// // return true
// // }
// // return false
// // })
// // spew.Dump("LIST", list)
// // for p, v := range list {
// // for _, vv := range v {
// // spew.Dump(fmt.Sprintf("Path: %s, S: %s, F: %s", p, vv.Ts, vv.Ts))
// // }
// // }
// var (
// r = make(types.ApigwProfilerAggregationSet, 0)
// tsum, tmin, tmax time.Duration
// ssum, smin, smax int64
// i uint64 = 1
// )
// for p, v := range list {
// tmin, tmax, tsum = time.Hour, 0, 0
// smin, smax, ssum = 0, 0, 0
// i = 0
// for _, vv := range v {
// var (
// d = vv.Tf.Sub(*vv.Ts)
// s = vv.R.ContentLength
// )
// if d < tmin {
// tmin = d
// }
// if d > tmax {
// tmax = d
// }
// if s < smin {
// smin = s
// }
// if s > smax {
// smax = s
// }
// tsum += d
// ssum += s
// i++
// }
// spew.Dump("TSUM", tsum.Seconds())
// r = append(r, &types.ApigwProfilerAggregation{
// Path: p,
// Count: i,
// Tmin: tmin,
// Tmax: tmax,
// Tavg: time.Duration(int64(tsum.Seconds()/float64(i))) * time.Second,
// Smin: smin,
// Smax: smax,
// Savg: float64(ssum) / float64(i),
// })
// }
// spew.Dump(r)
// SortAggregation(&r, &f)
// spew.Dump(r)
// t.Fail()
// }
pkg/apigw/registry/registry.go
+1
View File
@@ -59,6 +59,7 @@ func (r *Registry) Preload() {
// prefilters
r.Add("queryParam", filter.NewQueryParam())
r.Add("header", filter.NewHeader())
r.Add("profiler", filter.NewProfiler())
// processers
r.Add("workflow", filter.NewWorkflow(NewWorkflow()))
pkg/apigw/route.go
+23 -1
View File
@@ -6,6 +6,7 @@ import (
"time"
actx "github.com/cortezaproject/corteza-server/pkg/apigw/ctx"
"github.com/cortezaproject/corteza-server/pkg/apigw/profiler"
"github.com/cortezaproject/corteza-server/pkg/apigw/types"
"github.com/cortezaproject/corteza-server/pkg/auth"
h "github.com/cortezaproject/corteza-server/pkg/http"
@@ -22,6 +23,7 @@ type (
opts *options.ApigwOpt
log *zap.Logger
pr *profiler.Profiler
handler http.Handler
errHandler types.ErrorHandlerFunc
@@ -36,8 +38,9 @@ type (
func (r route) ServeHTTP(w http.ResponseWriter, req *http.Request) {
var (
ctx = auth.SetIdentityToContext(req.Context(), auth.ServiceUser())
scope = types.Scp{}
start = time.Now()
scope = types.Scp{}
hit = &profiler.Hit{}
)
r.log.Debug("started serving route")
@@ -51,10 +54,29 @@ func (r route) ServeHTTP(w http.ResponseWriter, req *http.Request) {
scope.Set("opts", r.opts)
scope.Set("request", ar)
// use profiler, override any profiling prefilter
if r.opts.ProfilerEnabled {
// add request to profiler
hit = r.pr.Hit(ar)
hit.Route = r.ID
}
req = req.WithContext(actx.ScopeToContext(ctx, &scope))
req = req.WithContext(actx.ProfilerToContext(req.Context(), hit))
r.handler.ServeHTTP(w, req)
if r.opts.ProfilerEnabled {
r.pr.Push(hit)
} else {
if hit = actx.ProfilerFromContext(req.Context()).(*profiler.Hit); hit != nil && hit.R != nil {
// updated hit from a possible prefilter
// we need to push route ID even if the profiler is disabled
hit.Route = r.ID
r.pr.Push(hit)
}
}
r.log.Debug("finished serving route",
zap.Duration("duration", time.Since(start)),
)
pkg/apigw/route_test.go
+2 -1
View File
@@ -7,6 +7,7 @@ import (
"testing"
"github.com/cortezaproject/corteza-server/pkg/apigw/pipeline"
"github.com/cortezaproject/corteza-server/pkg/apigw/pipeline/chain"
"github.com/cortezaproject/corteza-server/pkg/apigw/types"
"github.com/cortezaproject/corteza-server/pkg/options"
"github.com/stretchr/testify/require"
@@ -77,7 +78,7 @@ func Test_pl(t *testing.T) {
var (
req = require.New(t)
rr = httptest.NewRecorder()
pipe = pipeline.NewPipeline(zap.NewNop())
pipe = pipeline.NewPipeline(zap.NewNop(), chain.NewDefault())
)
r := httptest.NewRequest("POST", "/foo", http.NoBody)
pkg/apigw/service.go
+29 -6
View File
@@ -10,6 +10,8 @@ import (
"github.com/cortezaproject/corteza-server/pkg/apigw/filter"
"github.com/cortezaproject/corteza-server/pkg/apigw/filter/proxy"
"github.com/cortezaproject/corteza-server/pkg/apigw/pipeline"
"github.com/cortezaproject/corteza-server/pkg/apigw/pipeline/chain"
"github.com/cortezaproject/corteza-server/pkg/apigw/profiler"
"github.com/cortezaproject/corteza-server/pkg/apigw/registry"
"github.com/cortezaproject/corteza-server/pkg/apigw/types"
f "github.com/cortezaproject/corteza-server/pkg/filter"
@@ -31,6 +33,7 @@ type (
reg *registry.Registry
routes []*route
mx *chi.Mux
pr *profiler.Profiler
storer storer
}
)
@@ -54,7 +57,11 @@ func Setup(opts *options.ApigwOpt, log *zap.Logger, storer storer) {
}
func New(opts *options.ApigwOpt, logger *zap.Logger, storer storer) *apigw {
reg := registry.NewRegistry()
var (
pr = profiler.New()
reg = registry.NewRegistry()
)
reg.Preload()
return &apigw{
@@ -62,6 +69,7 @@ func New(opts *options.ApigwOpt, logger *zap.Logger, storer storer) *apigw {
log: logger,
storer: storer,
reg: reg,
pr: pr,
}
}
@@ -71,12 +79,12 @@ func New(opts *options.ApigwOpt, logger *zap.Logger, storer storer) *apigw {
// When reloading routes, make sure to replace the original mux
func (s *apigw) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if s.mx == nil {
http.Error(w, "API Gateway not initialized", http.StatusInternalServerError)
http.Error(w, "Integration Gateway not initialized", http.StatusInternalServerError)
return
}
if len(s.routes) == 0 {
helperDefaultResponse(s.opts)(w, r)
helperDefaultResponse(s.opts, s.pr)(w, r)
return
}
@@ -113,8 +121,18 @@ func (s *apigw) Reload(ctx context.Context) (err error) {
s.mx.Method(r.method, r.endpoint, r)
}
// API GW 404 handler
s.mx.NotFound(helperDefaultResponse(s.opts))
// handling missed hits
// profiler gets the missed hit info also
{
var (
defaultMethodResponse = helperMethodNotAllowed(s.opts, s.pr)
defaultResponse = helperDefaultResponse(s.opts, s.pr)
// profiler = helperProfiler(s.opts, s.pr)
)
s.mx.NotFound(defaultResponse)
s.mx.MethodNotAllowed(defaultMethodResponse)
}
return nil
}
@@ -138,7 +156,7 @@ func (s *apigw) Init(ctx context.Context, routes ...*route) {
for _, r := range s.routes {
var (
log = s.log.With(zap.String("route", r.String()))
pipe = pipeline.NewPipeline(log)
pipe = pipeline.NewPipeline(log, chain.NewDefault())
)
// pipeline needs to know how to handle
@@ -147,6 +165,7 @@ func (s *apigw) Init(ctx context.Context, routes ...*route) {
r.opts = s.opts
r.log = log
r.pr = s.pr
regFilters, err := s.loadFilters(ctx, r.ID)
@@ -282,3 +301,7 @@ func (s *apigw) loadFilters(ctx context.Context, route uint64) (ff []*st.ApigwFi
return
}
func (s *apigw) Profiler() *profiler.Profiler {
return s.pr
}
pkg/apigw/types/scope.go
+9 -8
View File
@@ -4,7 +4,7 @@ import (
"fmt"
"net/http"
"github.com/cortezaproject/corteza-server/pkg/expr"
h "github.com/cortezaproject/corteza-server/pkg/http"
"github.com/cortezaproject/corteza-server/pkg/options"
)
@@ -20,15 +20,16 @@ func (s Scp) Keys() (kk []string) {
return
}
func (s Scp) Request() *http.Request {
func (s Scp) Request() *h.Request {
// todo - fix with expr.Request
if _, ok := s["request"]; ok {
switch s["request"].(type) {
case *http.Request:
return s["request"].(*http.Request)
case *expr.Any:
return s["request"].(*expr.Any).Get().(*http.Request)
}
// switch s["request"].(type) {
// case *http.Request:
// return s["request"].(*http.Request)
// case *expr.Any:
// return s["request"].(*expr.Any).Get().(*http.Request)
// }
return s["request"].(*h.Request)
}
return nil
pkg/options/options.gen.go
Generated
+1
View File
@@ -85,6 +85,7 @@ type (
Enabled bool `env:"APIGW_ENABLED"`
Debug bool `env:"APIGW_DEBUG"`
LogEnabled bool `env:"APIGW_LOG_ENABLED"`
ProfilerEnabled bool `env:"APIGW_PROFILER_ENABLED"`
LogRequestBody bool `env:"APIGW_LOG_REQUEST_BODY"`
ProxyEnableDebugLog bool `env:"APIGW_PROXY_ENABLE_DEBUG_LOG"`
ProxyFollowRedirects bool `env:"APIGW_PROXY_FOLLOW_REDIRECTS"`
pkg/options/options.go
+2
View File
@@ -28,6 +28,7 @@ type (
Limit LimitOpt
Plugins PluginsOpt
Discovery DiscoveryOpt
Apigw ApigwOpt
}
)
@@ -59,5 +60,6 @@ func Init() *Options {
Limit: *Limit(),
Plugins: *Plugins(),
Discovery: *Discovery(),
Apigw: *Apigw(),
}
}
system/rest.yaml
+33
View File
@@ -1822,6 +1822,39 @@ endpoints:
title: Proxy auth definitions
path: "/proxy_auth/def"
- title: API Gateway profiler
path: "/apigw/profiler"
entrypoint: apigwProfiler
authentication: []
apis:
- name: aggregation
method: GET
title: List aggregated list of routes
path: "/"
parameters:
get:
- { name: path, type: "string", title: "Filter by request path" }
- { name: before, type: "string", title: "Entries before specified route" }
- { name: sort, type: "string", title: "Sort items" }
- { name: limit, type: "uint", title: "Limit" }
- name: route
method: GET
title: List hits per route
path: "/route/{routeID}"
parameters:
path:
- { name: routeID, type: "string", title: "Route ID", required: true }
get:
- { name: path, type: "string", title: "Filter by request path" }
- { name: before, type: "string", title: "Entries before specified hit ID" }
- { name: sort, type: "string", title: "Sort items" }
- { name: limit, type: "uint", title: "Limit" }
- name: hit
method: GET
title: Hit details
path: "/hit/{hitID}"
parameters: { path: [ { name: hitID, type: string, required: true, title: "Hit ID" } ] }
- title: Locale
entrypoint: locale
path: "/locale"
system/rest/apigw_profiler.go
+158
View File
@@ -0,0 +1,158 @@
package rest
import (
"context"
"github.com/cortezaproject/corteza-server/pkg/filter"
"github.com/cortezaproject/corteza-server/system/rest/request"
"github.com/cortezaproject/corteza-server/system/service"
"github.com/cortezaproject/corteza-server/system/types"
)
type (
profilerService interface {
Hits(context.Context, types.ApigwProfilerFilter) (types.ApigwProfilerHitSet, types.ApigwProfilerFilter, error)
HitsAggregated(context.Context, types.ApigwProfilerFilter) (types.ApigwProfilerAggregationSet, types.ApigwProfilerFilter, error)
}
ApigwProfiler struct {
svc profilerService
ac templateAccessController
}
profilerRoutePayload struct {
*types.ApigwProfilerHit
}
profilerHitPayload struct {
*types.ApigwProfilerAggregation
}
profilerRouteSetPayload struct {
Filter types.ApigwProfilerFilter `json:"filter"`
Set []*profilerRoutePayload `json:"set"`
}
profilerHitSetPayload struct {
Filter types.ApigwProfilerFilter `json:"filter"`
Set []*profilerHitPayload `json:"set"`
}
)
func (ApigwProfiler) New() *ApigwProfiler {
return &ApigwProfiler{
svc: service.DefaultApigwRoute,
ac: service.DefaultAccessControl,
}
}
// List displays the the aggregated list of routes
func (ctrl *ApigwProfiler) Aggregation(ctx context.Context, r *request.ApigwProfilerAggregation) (interface{}, error) {
var (
err error
f = types.ApigwProfilerFilter{
Path: r.Path,
Before: r.Before,
}
)
if f.Sorting, err = filter.NewSorting(r.Sort); err != nil {
return nil, err
}
if f.Paging, err = filter.NewPaging(r.Limit, ""); err != nil {
return nil, err
}
set, f, err := ctrl.svc.HitsAggregated(ctx, f)
return ctrl.makeFilterPayload(ctx, set, f, err)
}
// Route displays the list of hits per-route
func (ctrl *ApigwProfiler) Route(ctx context.Context, r *request.ApigwProfilerRoute) (interface{}, error) {
var (
err error
f = types.ApigwProfilerFilter{
Path: r.RouteID,
Before: r.Before,
}
)
if f.Sorting, err = filter.NewSorting(r.Sort); err != nil {
return nil, err
}
if f.Paging, err = filter.NewPaging(r.Limit, ""); err != nil {
return nil, err
}
set, f, err := ctrl.svc.Hits(ctx, f)
return ctrl.makeRouteFilterPayload(ctx, set, f, err)
}
// Hit displays the details of a certain hit on a route
func (ctrl *ApigwProfiler) Hit(ctx context.Context, r *request.ApigwProfilerHit) (interface{}, error) {
var (
f = types.ApigwProfilerFilter{
Hit: r.HitID,
}
)
set, f, err := ctrl.svc.Hits(ctx, f)
if len(set) != 1 {
return nil, nil
}
return ctrl.makeRoutePayload(ctx, set[0], err)
}
func (ctrl *ApigwProfiler) makePayload(ctx context.Context, q *types.ApigwProfilerAggregation, err error) (*profilerHitPayload, error) {
if err != nil || q == nil {
return nil, err
}
qq := &profilerHitPayload{
ApigwProfilerAggregation: q,
}
return qq, nil
}
func (ctrl *ApigwProfiler) makeFilterPayload(ctx context.Context, nn types.ApigwProfilerAggregationSet, f types.ApigwProfilerFilter, err error) (*profilerHitSetPayload, error) {
if err != nil {
return nil, err
}
msp := &profilerHitSetPayload{Filter: f, Set: make([]*profilerHitPayload, len(nn))}
for i := range nn {
msp.Set[i], _ = ctrl.makePayload(ctx, nn[i], nil)
}
return msp, nil
}
func (ctrl *ApigwProfiler) makeRoutePayload(ctx context.Context, q *types.ApigwProfilerHit, err error) (*profilerRoutePayload, error) {
if err != nil || q == nil {
return nil, err
}
return &profilerRoutePayload{q}, nil
}
func (ctrl *ApigwProfiler) makeRouteFilterPayload(ctx context.Context, nn types.ApigwProfilerHitSet, f types.ApigwProfilerFilter, err error) (*profilerRouteSetPayload, error) {
if err != nil {
return nil, err
}
msp := &profilerRouteSetPayload{Filter: f, Set: make([]*profilerRoutePayload, len(nn))}
for i := range nn {
msp.Set[i], _ = ctrl.makeRoutePayload(ctx, nn[i], nil)
}
return msp, nil
}
system/rest/handlers/apigwProfiler.go
+95
View File
@@ -0,0 +1,95 @@
package handlers
// This file is auto-generated.
//
// Changes to this file may cause incorrect behavior and will be lost if
// the code is regenerated.
//
// Definitions file that controls how this file is generated:
//
import (
"context"
"github.com/cortezaproject/corteza-server/pkg/api"
"github.com/cortezaproject/corteza-server/system/rest/request"
"github.com/go-chi/chi/v5"
"net/http"
)
type (
// Internal API interface
ApigwProfilerAPI interface {
Aggregation(context.Context, *request.ApigwProfilerAggregation) (interface{}, error)
Route(context.Context, *request.ApigwProfilerRoute) (interface{}, error)
Hit(context.Context, *request.ApigwProfilerHit) (interface{}, error)
}
// HTTP API interface
ApigwProfiler struct {
Aggregation func(http.ResponseWriter, *http.Request)
Route func(http.ResponseWriter, *http.Request)
Hit func(http.ResponseWriter, *http.Request)
}
)
func NewApigwProfiler(h ApigwProfilerAPI) *ApigwProfiler {
return &ApigwProfiler{
Aggregation: func(w http.ResponseWriter, r *http.Request) {
defer r.Body.Close()
params := request.NewApigwProfilerAggregation()
if err := params.Fill(r); err != nil {
api.Send(w, r, err)
return
}
value, err := h.Aggregation(r.Context(), params)
if err != nil {
api.Send(w, r, err)
return
}
api.Send(w, r, value)
},
Route: func(w http.ResponseWriter, r *http.Request) {
defer r.Body.Close()
params := request.NewApigwProfilerRoute()
if err := params.Fill(r); err != nil {
api.Send(w, r, err)
return
}
value, err := h.Route(r.Context(), params)
if err != nil {
api.Send(w, r, err)
return
}
api.Send(w, r, value)
},
Hit: func(w http.ResponseWriter, r *http.Request) {
defer r.Body.Close()
params := request.NewApigwProfilerHit()
if err := params.Fill(r); err != nil {
api.Send(w, r, err)
return
}
value, err := h.Hit(r.Context(), params)
if err != nil {
api.Send(w, r, err)
return
}
api.Send(w, r, value)
},
}
}
func (h ApigwProfiler) MountRoutes(r chi.Router, middlewares ...func(http.Handler) http.Handler) {
r.Group(func(r chi.Router) {
r.Use(middlewares...)
r.Get("/apigw/profiler/", h.Aggregation)
r.Get("/apigw/profiler/route/{routeID}", h.Route)
r.Get("/apigw/profiler/hit/{hitID}", h.Hit)
})
}
system/rest/request/apigwProfiler.go
+286
View File
@@ -0,0 +1,286 @@
package request
// This file is auto-generated.
//
// Changes to this file may cause incorrect behavior and will be lost if
// the code is regenerated.
//
// Definitions file that controls how this file is generated:
//
import (
"encoding/json"
"fmt"
"github.com/cortezaproject/corteza-server/pkg/payload"
"github.com/go-chi/chi/v5"
"io"
"mime/multipart"
"net/http"
"strings"
)
// dummy vars to prevent
// unused imports complain
var (
_ = chi.URLParam
_ = multipart.ErrMessageTooLarge
_ = payload.ParseUint64s
_ = strings.ToLower
_ = io.EOF
_ = fmt.Errorf
_ = json.NewEncoder
)
type (
// Internal API interface
ApigwProfilerAggregation struct {
// Path GET parameter
//
// Filter by request path
Path string
// Before GET parameter
//
// Entries before specified route
Before string
// Sort GET parameter
//
// Sort items
Sort string
// Limit GET parameter
//
// Limit
Limit uint
}
ApigwProfilerRoute struct {
// RouteID PATH parameter
//
// Route ID
RouteID string
// Path GET parameter
//
// Filter by request path
Path string
// Before GET parameter
//
// Entries before specified hit ID
Before string
// Sort GET parameter
//
// Sort items
Sort string
// Limit GET parameter
//
// Limit
Limit uint
}
ApigwProfilerHit struct {
// HitID PATH parameter
//
// Hit ID
HitID string
}
)
// NewApigwProfilerAggregation request
func NewApigwProfilerAggregation() *ApigwProfilerAggregation {
return &ApigwProfilerAggregation{}
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerAggregation) Auditable() map[string]interface{} {
return map[string]interface{}{
"path": r.Path,
"before": r.Before,
"sort": r.Sort,
"limit": r.Limit,
}
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerAggregation) GetPath() string {
return r.Path
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerAggregation) GetBefore() string {
return r.Before
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerAggregation) GetSort() string {
return r.Sort
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerAggregation) GetLimit() uint {
return r.Limit
}
// Fill processes request and fills internal variables
func (r *ApigwProfilerAggregation) Fill(req *http.Request) (err error) {
{
// GET params
tmp := req.URL.Query()
if val, ok := tmp["path"]; ok && len(val) > 0 {
r.Path, err = val[0], nil
if err != nil {
return err
}
}
if val, ok := tmp["before"]; ok && len(val) > 0 {
r.Before, err = val[0], nil
if err != nil {
return err
}
}
if val, ok := tmp["sort"]; ok && len(val) > 0 {
r.Sort, err = val[0], nil
if err != nil {
return err
}
}
if val, ok := tmp["limit"]; ok && len(val) > 0 {
r.Limit, err = payload.ParseUint(val[0]), nil
if err != nil {
return err
}
}
}
return err
}
// NewApigwProfilerRoute request
func NewApigwProfilerRoute() *ApigwProfilerRoute {
return &ApigwProfilerRoute{}
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerRoute) Auditable() map[string]interface{} {
return map[string]interface{}{
"routeID": r.RouteID,
"path": r.Path,
"before": r.Before,
"sort": r.Sort,
"limit": r.Limit,
}
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerRoute) GetRouteID() string {
return r.RouteID
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerRoute) GetPath() string {
return r.Path
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerRoute) GetBefore() string {
return r.Before
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerRoute) GetSort() string {
return r.Sort
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerRoute) GetLimit() uint {
return r.Limit
}
// Fill processes request and fills internal variables
func (r *ApigwProfilerRoute) Fill(req *http.Request) (err error) {
{
// GET params
tmp := req.URL.Query()
if val, ok := tmp["path"]; ok && len(val) > 0 {
r.Path, err = val[0], nil
if err != nil {
return err
}
}
if val, ok := tmp["before"]; ok && len(val) > 0 {
r.Before, err = val[0], nil
if err != nil {
return err
}
}
if val, ok := tmp["sort"]; ok && len(val) > 0 {
r.Sort, err = val[0], nil
if err != nil {
return err
}
}
if val, ok := tmp["limit"]; ok && len(val) > 0 {
r.Limit, err = payload.ParseUint(val[0]), nil
if err != nil {
return err
}
}
}
{
var val string
// path params
val = chi.URLParam(req, "routeID")
r.RouteID, err = val, nil
if err != nil {
return err
}
}
return err
}
// NewApigwProfilerHit request
func NewApigwProfilerHit() *ApigwProfilerHit {
return &ApigwProfilerHit{}
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerHit) Auditable() map[string]interface{} {
return map[string]interface{}{
"hitID": r.HitID,
}
}
// Auditable returns all auditable/loggable parameters
func (r ApigwProfilerHit) GetHitID() string {
return r.HitID
}
// Fill processes request and fills internal variables
func (r *ApigwProfilerHit) Fill(req *http.Request) (err error) {
{
var val string
// path params
val = chi.URLParam(req, "hitID")
r.HitID, err = val, nil
if err != nil {
return err
}
}
return err
}
system/rest/router.go
+1
View File
@@ -43,6 +43,7 @@ func MountRoutes() func(r chi.Router) {
handlers.NewQueues(Queue{}.New()).MountRoutes(r)
handlers.NewApigwRoute(ApigwRoute{}.New()).MountRoutes(r)
handlers.NewApigwFilter(ApigwFilter{}.New()).MountRoutes(r)
handlers.NewApigwProfiler(ApigwProfiler{}.New()).MountRoutes(r)
})
}
}
system/service/apigw_route.go
+163
View File
@@ -2,9 +2,15 @@ package service
import (
"context"
"encoding/base64"
"errors"
"io/ioutil"
"math"
"time"
"github.com/cortezaproject/corteza-server/pkg/actionlog"
"github.com/cortezaproject/corteza-server/pkg/apigw"
"github.com/cortezaproject/corteza-server/pkg/apigw/profiler"
a "github.com/cortezaproject/corteza-server/pkg/auth"
"github.com/cortezaproject/corteza-server/store"
@@ -253,3 +259,160 @@ func (svc *apigwRoute) Search(ctx context.Context, filter types.ApigwRouteFilter
return r, f, svc.recordAction(ctx, aProps, ApigwRouteActionSearch, err)
}
// HitsAggregated fetches a list of hits from integration gateway profiler
func (svc *apigwRoute) Hits(ctx context.Context, filter types.ApigwProfilerFilter) (r types.ApigwProfilerHitSet, f types.ApigwProfilerFilter, err error) {
f = filter
r = make(types.ApigwProfilerHitSet, 0)
uDec, err := base64.URLEncoding.DecodeString(filter.Path)
if err != nil {
return
}
filter.Path = string(uDec)
if filter.Path == "" && filter.Hit == "" {
err = errors.New("fetching all hits (no route and hit specified) not supported")
return
}
var sorting = profiler.Sort{
Hit: filter.Hit,
Path: filter.Path,
Before: filter.Before,
}
var (
list = apigw.Service().Profiler().Hits(sorting)
)
var pp = ""
for k, _ := range list {
if filter.Hit != "" {
pp = k
break
}
if filter.Path != "" && k == filter.Path {
pp = k
break
}
}
if pp == "" {
return
}
for _, h := range list[pp] {
hh := &types.ApigwProfilerHit{
ID: h.ID,
Route: h.Route,
Status: h.Status,
Request: *h.R,
Ts: h.Ts,
Tf: h.Tf,
D: h.D,
Dr: float64(h.D.Microseconds()) / 1000,
}
// fetch body only on hit details
if filter.Hit != "" {
hh.Body, _ = ioutil.ReadAll(hh.Request.Body)
}
r = append(r, hh)
}
// sort
profiler.SortHits(&r, &f)
// filter sorted
profiler.FilterHits(&r, &f)
return
}
// HitsAggregated fetches a list of hits from integration gateway profiler
// and aggregates them with assigned filters
func (svc *apigwRoute) HitsAggregated(ctx context.Context, filter types.ApigwProfilerFilter) (r types.ApigwProfilerAggregationSet, f types.ApigwProfilerFilter, err error) {
f = filter
r = make(types.ApigwProfilerAggregationSet, 0)
uDec, err := base64.URLEncoding.DecodeString(filter.Path)
if err != nil {
return
}
filter.Path = string(uDec)
var (
list = apigw.Service().Profiler().Hits(profiler.Sort{
Path: filter.Path,
Before: filter.Before,
})
tsum, tmin, tmax time.Duration
ssum, smin, smax int64
i uint64 = 1
)
for p, v := range list {
tmin, tmax, tsum = time.Hour, 0, 0
smin, smax, ssum = math.MaxInt64, 0, 0
i = 0
for _, vv := range v {
var (
d = *vv.D
s = vv.R.ContentLength
)
if d < tmin {
tmin = d
}
if d > tmax {
tmax = d
}
if s < smin {
smin = s
}
if s > smax {
smax = s
}
tsum += d
ssum += s
i++
}
r = append(r, &types.ApigwProfilerAggregation{
Path: p,
Count: i,
Tmin: float64(tmin.Microseconds()) / 1000,
Tmax: float64(tmax.Microseconds()) / 1000,
Tavg: float64(tsum.Microseconds()) / float64(i) / 1000,
Smin: smin,
Smax: smax,
Savg: float64(ssum) / float64(i),
})
}
// sort
profiler.SortAggregation(&r, &f)
// filter
profiler.FilterAggregation(&r, &f)
return
}
system/types/apigw_route.go
+195
View File
@@ -6,6 +6,7 @@ import (
"time"
"github.com/cortezaproject/corteza-server/pkg/filter"
h "github.com/cortezaproject/corteza-server/pkg/http"
"github.com/pkg/errors"
)
@@ -26,6 +27,31 @@ type (
DeletedBy uint64 `json:"deletedBy,string,omitempty" `
}
ApigwProfilerHit struct {
ID string `json:"ID"`
Body []byte `json:"body"`
Request h.Request `json:"request"`
Route uint64 `json:"route,string"`
Status int `json:"http_status_code,string"`
Ts *time.Time `json:"time_start"`
Tf *time.Time `json:"time_finish"`
D *time.Duration `json:"-"`
Dr float64 `json:"time_duration"`
}
ApigwProfilerAggregation struct {
Path string `json:"path"`
Count uint64 `json:"count"`
Smin int64 `json:"size_min"`
Smax int64 `json:"size_max"`
Savg float64 `json:"size_avg"`
Tmin float64 `json:"time_min"`
Tmax float64 `json:"time_max"`
Tavg float64 `json:"time_avg"`
}
ApigwRouteMeta struct {
Debug bool `json:"debug"`
Async bool `json:"async"`
@@ -47,6 +73,16 @@ type (
filter.Sorting
filter.Paging
}
ApigwProfilerFilter struct {
Hit string `json:"hit,omitempty"`
Path string `json:"path,omitempty"`
Before string `json:"before,omitempty"`
Next string `json:"next,omitempty"`
filter.Sorting
filter.Paging
}
)
func (cc *ApigwRouteMeta) Scan(value interface{}) error {
@@ -67,3 +103,162 @@ func (cc *ApigwRouteMeta) Scan(value interface{}) error {
func (cc ApigwRouteMeta) Value() (driver.Value, error) {
return json.Marshal(cc)
}
// sorting methods
type (
ByPath ApigwProfilerAggregationSet
ByCount ApigwProfilerAggregationSet
BySizeMin ApigwProfilerAggregationSet
BySizeMax ApigwProfilerAggregationSet
BySizeAvg ApigwProfilerAggregationSet
ByTimeMin ApigwProfilerAggregationSet
ByTimeMax ApigwProfilerAggregationSet
ByTimeAvg ApigwProfilerAggregationSet
BySTime ApigwProfilerHitSet
ByFTime ApigwProfilerHitSet
ByDuration ApigwProfilerHitSet
)
func (h ByPath) Len() int {
return len(h)
}
func (h ByPath) Less(i, j int) bool {
return h[i].Path < h[j].Path
}
func (h ByPath) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h ByCount) Len() int {
return len(h)
}
func (h ByCount) Less(i, j int) bool {
return h[i].Count < h[j].Count
}
func (h ByCount) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h BySizeMin) Len() int {
return len(h)
}
func (h BySizeMin) Less(i, j int) bool {
return h[i].Smin < h[j].Smin
}
func (h BySizeMin) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h BySizeMax) Len() int {
return len(h)
}
func (h BySizeMax) Less(i, j int) bool {
return h[i].Smax < h[j].Smax
}
func (h BySizeMax) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h BySizeAvg) Len() int {
return len(h)
}
func (h BySizeAvg) Less(i, j int) bool {
return h[i].Savg < h[j].Savg
}
func (h BySizeAvg) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h ByTimeAvg) Len() int {
return len(h)
}
func (h ByTimeAvg) Less(i, j int) bool {
return h[i].Tavg < h[j].Tavg
}
func (h ByTimeAvg) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h ByTimeMax) Len() int {
return len(h)
}
func (h ByTimeMax) Less(i, j int) bool {
return h[i].Tmax < h[j].Tmax
}
func (h ByTimeMax) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h ByTimeMin) Len() int {
return len(h)
}
func (h ByTimeMin) Less(i, j int) bool {
return h[i].Tmin < h[j].Tmin
}
func (h ByTimeMin) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h BySTime) Len() int {
return len(h)
}
func (h BySTime) Less(i, j int) bool {
return h[j].Ts.After(*h[i].Ts)
}
func (h BySTime) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h ByFTime) Len() int {
return len(h)
}
func (h ByFTime) Less(i, j int) bool {
return h[j].Tf.After(*h[i].Tf)
}
func (h ByFTime) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
func (h ByDuration) Len() int {
return len(h)
}
func (h ByDuration) Less(i, j int) bool {
return h[i].D.Microseconds() > h[j].D.Microseconds()
}
func (h ByDuration) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
return
}
system/types/app_settings.go
+7
View File
@@ -207,6 +207,13 @@ type (
Enabled bool `kv:"-" json:"enabled"`
} `kv:"federation" json:"federation"`
// Federation settings
Apigw struct {
// This only holds the value of APIGW_PROFILER_ENABLED for now
//
ProfilerEnabled bool `kv:"-" json:"profilerEnabled"`
} `kv:"apigw" json:"apigw"`
// UserInterface settings
UI struct {
MainLogo string `kv:"main-logo" json:"mainLogo"`
system/types/type_set.gen.go
Generated
+70
View File
@@ -15,6 +15,16 @@ type (
// This type is auto-generated.
ApigwFilterSet []*ApigwFilter
// ApigwProfilerAggregationSet slice of ApigwProfilerAggregation
//
// This type is auto-generated.
ApigwProfilerAggregationSet []*ApigwProfilerAggregation
// ApigwProfilerHitSet slice of ApigwProfilerHit
//
// This type is auto-generated.
ApigwProfilerHitSet []*ApigwProfilerHit
// ApigwRouteSet slice of ApigwRoute
//
// This type is auto-generated.
@@ -162,6 +172,66 @@ func (set ApigwFilterSet) IDs() (IDs []uint64) {
return
}
// Walk iterates through every slice item and calls w(ApigwProfilerAggregation) err
//
// This function is auto-generated.
func (set ApigwProfilerAggregationSet) Walk(w func(*ApigwProfilerAggregation) error) (err error) {
for i := range set {
if err = w(set[i]); err != nil {
return
}
}
return
}
// Filter iterates through every slice item, calls f(ApigwProfilerAggregation) (bool, err) and return filtered slice
//
// This function is auto-generated.
func (set ApigwProfilerAggregationSet) Filter(f func(*ApigwProfilerAggregation) (bool, error)) (out ApigwProfilerAggregationSet, err error) {
var ok bool
out = ApigwProfilerAggregationSet{}
for i := range set {
if ok, err = f(set[i]); err != nil {
return
} else if ok {
out = append(out, set[i])
}
}
return
}
// Walk iterates through every slice item and calls w(ApigwProfilerHit) err
//
// This function is auto-generated.
func (set ApigwProfilerHitSet) Walk(w func(*ApigwProfilerHit) error) (err error) {
for i := range set {
if err = w(set[i]); err != nil {
return
}
}
return
}
// Filter iterates through every slice item, calls f(ApigwProfilerHit) (bool, err) and return filtered slice
//
// This function is auto-generated.
func (set ApigwProfilerHitSet) Filter(f func(*ApigwProfilerHit) (bool, error)) (out ApigwProfilerHitSet, err error) {
var ok bool
out = ApigwProfilerHitSet{}
for i := range set {
if ok, err = f(set[i]); err != nil {
return
} else if ok {
out = append(out, set[i])
}
}
return
}
// Walk iterates through every slice item and calls w(ApigwRoute) err
//
// This function is auto-generated.
system/types/type_set.gen_test.go
Generated
+112
View File
@@ -104,6 +104,118 @@ func TestApigwFilterSetIDs(t *testing.T) {
}
}
func TestApigwProfilerAggregationSetWalk(t *testing.T) {
var (
value = make(ApigwProfilerAggregationSet, 3)
req = require.New(t)
)
// check walk with no errors
{
err := value.Walk(func(*ApigwProfilerAggregation) error {
return nil
})
req.NoError(err)
}
// check walk with error
req.Error(value.Walk(func(*ApigwProfilerAggregation) error { return fmt.Errorf("walk error") }))
}
func TestApigwProfilerAggregationSetFilter(t *testing.T) {
var (
value = make(ApigwProfilerAggregationSet, 3)
req = require.New(t)
)
// filter nothing
{
set, err := value.Filter(func(*ApigwProfilerAggregation) (bool, error) {
return true, nil
})
req.NoError(err)
req.Equal(len(set), len(value))
}
// filter one item
{
found := false
set, err := value.Filter(func(*ApigwProfilerAggregation) (bool, error) {
if !found {
found = true
return found, nil
}
return false, nil
})
req.NoError(err)
req.Len(set, 1)
}
// filter error
{
_, err := value.Filter(func(*ApigwProfilerAggregation) (bool, error) {
return false, fmt.Errorf("filter error")
})
req.Error(err)
}
}
func TestApigwProfilerHitSetWalk(t *testing.T) {
var (
value = make(ApigwProfilerHitSet, 3)
req = require.New(t)
)
// check walk with no errors
{
err := value.Walk(func(*ApigwProfilerHit) error {
return nil
})
req.NoError(err)
}
// check walk with error
req.Error(value.Walk(func(*ApigwProfilerHit) error { return fmt.Errorf("walk error") }))
}
func TestApigwProfilerHitSetFilter(t *testing.T) {
var (
value = make(ApigwProfilerHitSet, 3)
req = require.New(t)
)
// filter nothing
{
set, err := value.Filter(func(*ApigwProfilerHit) (bool, error) {
return true, nil
})
req.NoError(err)
req.Equal(len(set), len(value))
}
// filter one item
{
found := false
set, err := value.Filter(func(*ApigwProfilerHit) (bool, error) {
if !found {
found = true
return found, nil
}
return false, nil
})
req.NoError(err)
req.Len(set, 1)
}
// filter error
{
_, err := value.Filter(func(*ApigwProfilerHit) (bool, error) {
return false, fmt.Errorf("filter error")
})
req.Error(err)
}
}
func TestApigwRouteSetWalk(t *testing.T) {
var (
value = make(ApigwRouteSet, 3)
system/types/types.yaml
+4
View File
@@ -23,6 +23,10 @@ types:
labelResourceType: template
ApigwRoute: {}
ApigwFilter: {}
ApigwProfilerHit:
noIdField: true
ApigwProfilerAggregation:
noIdField: true
Report:
labelResourceType: report
ResourceTranslation: {}