6830c8879f
* feat(server): moved an http server object into a pkg subdir to be reusable for http, spots, and integrations * feat(web): isolated web module (server, router, middleware, utils) used in spots and new integrations * feat(web): removed possible panic * feat(web): split all handlers from http service into different packages for better management. * feat(web): changed router's method signature * feat(web): added missing handlers interface * feat(web): added health middleware to remove unnecessary checks * feat(web): customizable middleware set for web servers * feat(web): simplified the handler's structure * feat(web): created an unified server.Run method for all web services (http, spot, integrations) * feat(web): fixed a json size limit issue * feat(web): removed Keys and PG connection from router * feat(web): simplified integration's main file * feat(web): simplified spot's main file * feat(web): simplified http's main file (builder) * feat(web): refactored audit trail functionality * feat(web): added ee version of audit trail * feat(web): added ee version of conditions module * feat(web): moved ee version of some web session structs * feat(web): new format of web metrics * feat(web): added new web metrics to all handlers * feat(web): added justExpired feature to web ingest handler * feat(web): added small integrations improvements
106 lines
2.3 KiB
Go
106 lines
2.3 KiB
Go
package tracer
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/http"
|
|
|
|
"openreplay/backend/pkg/db/postgres"
|
|
db "openreplay/backend/pkg/db/postgres/pool"
|
|
"openreplay/backend/pkg/logger"
|
|
"openreplay/backend/pkg/pool"
|
|
"openreplay/backend/pkg/server/user"
|
|
)
|
|
|
|
type Tracer interface {
|
|
Middleware(next http.Handler) http.Handler
|
|
Close() error
|
|
}
|
|
|
|
type tracerImpl struct {
|
|
log logger.Logger
|
|
conn db.Pool
|
|
traces postgres.Bulk
|
|
saver pool.WorkerPool
|
|
}
|
|
|
|
func NewTracer(log logger.Logger, conn db.Pool) (Tracer, error) {
|
|
switch {
|
|
case log == nil:
|
|
return nil, errors.New("logger is required")
|
|
case conn == nil:
|
|
return nil, errors.New("connection is required")
|
|
}
|
|
tracer := &tracerImpl{
|
|
log: log,
|
|
conn: conn,
|
|
}
|
|
if err := tracer.initBulk(); err != nil {
|
|
return nil, err
|
|
}
|
|
tracer.saver = pool.NewPool(1, 200, tracer.sendTraces)
|
|
return tracer, nil
|
|
}
|
|
|
|
func (t *tracerImpl) initBulk() (err error) {
|
|
t.traces, err = postgres.NewBulk(t.conn,
|
|
"traces",
|
|
"(user_id, tenant_id, auth, action, method, path_format, endpoint, payload, parameters, status)",
|
|
"($%d, $%d, $%d, $%d, $%d, $%d, $%d, $%d, $%d, $%d)",
|
|
10, 50)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type Task struct {
|
|
UserID *uint64
|
|
TenantID uint64
|
|
Auth *string
|
|
Data *RequestData
|
|
}
|
|
|
|
func (t *tracerImpl) sendTraces(payload interface{}) {
|
|
rec := payload.(*Task)
|
|
t.log.Debug(context.Background(), "Sending traces, %v", rec)
|
|
if err := t.traces.Append(rec.UserID, rec.TenantID, rec.Auth, rec.Data.Action, rec.Data.Method, rec.Data.PathFormat,
|
|
rec.Data.Endpoint, rec.Data.Payload, rec.Data.Parameters, rec.Data.Status); err != nil {
|
|
t.log.Error(context.Background(), "can't append trace: %s", err)
|
|
}
|
|
}
|
|
|
|
type RequestData struct {
|
|
Action string
|
|
Method string
|
|
PathFormat string
|
|
Endpoint string
|
|
Payload []byte
|
|
Parameters []byte
|
|
Status int
|
|
}
|
|
|
|
func (t *tracerImpl) trace(user *user.User, data *RequestData) error {
|
|
switch {
|
|
case user == nil:
|
|
return errors.New("user is required")
|
|
case data == nil:
|
|
return errors.New("request is required")
|
|
}
|
|
trace := &Task{
|
|
UserID: &user.ID,
|
|
TenantID: user.TenantID,
|
|
Auth: &user.AuthMethod,
|
|
Data: data,
|
|
}
|
|
t.saver.Submit(trace)
|
|
return nil
|
|
}
|
|
|
|
func (t *tracerImpl) Close() error {
|
|
t.saver.Stop()
|
|
if err := t.traces.Send(); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|