From 02580cce20bfe5329438852c03a4d5d99479941a Mon Sep 17 00:00:00 2001 From: Oleh Astappiev Date: Sun, 8 Dec 2024 20:45:44 +0100 Subject: [PATCH] feat: use queue to feed events, drop if overflowed --- forwarding.go | 106 ------------------------------------- umami.go | 47 ++++++----------- umami_send.go | 71 ------------------------- umami_token.go | 6 ++- umami_utils.go | 55 ++++++++++++++++---- umami_websites.go | 9 ++-- umami_worker.go | 130 ++++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 198 insertions(+), 226 deletions(-) delete mode 100644 forwarding.go delete mode 100644 umami_send.go create mode 100644 umami_worker.go diff --git a/forwarding.go b/forwarding.go deleted file mode 100644 index e282e4e..0000000 --- a/forwarding.go +++ /dev/null @@ -1,106 +0,0 @@ -package traefik_umami_feeder - -// Copied source from -// https://github.com/kzmake/traefik-plugin-forward-request/blob/master/util.go - -import ( - "net" - "net/http" - "strings" -) - -const ( - xForwardedProto = "x-forwarded-proto" - xForwardedFor = "x-forwarded-for" - xForwardedHost = "x-forwarded-host" - xForwardedPort = "x-forwarded-port" - xForwardedURI = "x-forwarded-uri" - xForwardedMethod = "x-forwarded-method" - - connection = "connection" - keepAlive = "keep-alive" - proxyAuthenticate = "proxy-authenticate" - proxyAuthorization = "proxy-authorization" - te = "te" // canonicalized version of "TE" - trailers = "trailers" - transferEncoding = "transfer-Encoding" - upgrade = "upgrade" -) - -var hopHeaders = []string{ - connection, - keepAlive, - proxyAuthenticate, - proxyAuthorization, - te, // canonicalized version of "TE" - trailers, - transferEncoding, - upgrade, -} - -func copyHeaders(dst, src http.Header) { - for k, vv := range src { - for _, v := range vv { - dst.Add(k, v) - } - } -} - -func removeHeaders(headers http.Header, names ...string) { - for _, h := range names { - headers.Del(h) - } -} - -func writeXForwardedHeaders(dst http.Header, req *http.Request) { - if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil { - if values := req.Header.Values(xForwardedFor); len(values) > 0 { - clientIP = strings.Join(values, ", ") + ", " + clientIP - } - dst.Set(xForwardedFor, clientIP) - } - - xfm := req.Header.Get(xForwardedMethod) - switch { - case xfm != "": - dst.Set(xForwardedMethod, xfm) - case req.Method != "": - dst.Set(xForwardedMethod, req.Method) - default: - dst.Del(xForwardedMethod) - } - - xfp := req.Header.Get(xForwardedProto) - switch { - case xfp != "": - dst.Set(xForwardedProto, xfp) - case req.TLS != nil: - dst.Set(xForwardedProto, "https") - default: - dst.Set(xForwardedProto, "http") - } - - if xfp := req.Header.Get(xForwardedPort); xfp != "" { - dst.Set(xForwardedPort, xfp) - } - - xfh := req.Header.Get(xForwardedHost) - switch { - case xfh != "": - dst.Set(xForwardedHost, xfh) - case req.Host != "": - dst.Set(xForwardedHost, req.Host) - default: - dst.Del(xForwardedHost) - } - - xfu := req.Header.Get(xForwardedURI) - switch { - case xfu != "": - dst.Set(xForwardedURI, xfu) - case req.URL.RequestURI() != "": - dst.Set(xForwardedURI, req.URL.RequestURI()) - default: - dst.Del(xForwardedURI) - } -} diff --git a/umami.go b/umami.go index 939d823..227816b 100644 --- a/umami.go +++ b/umami.go @@ -19,6 +19,8 @@ type Config struct { Disabled bool `json:"disabled"` // Debug enables debug logging, be prepared for flooding. Debug bool `json:"debug"` + // QueueSize defines the size of queue, i.e. the amount of events that are waiting to be submitted to Umami. + QueueSize int // UmamiHost is the URL of the Umami instance. UmamiHost string `json:"umamiHost"` @@ -56,8 +58,9 @@ type Config struct { // CreateConfig creates the default plugin configuration. func CreateConfig() *Config { return &Config{ - Disabled: false, - Debug: false, + Disabled: false, + Debug: false, + QueueSize: 1000, UmamiHost: "", UmamiToken: "", @@ -85,6 +88,7 @@ type UmamiFeeder struct { isDebug bool isDisabled bool logHandler *log.Logger + queue chan *UmamiPayload umamiHost string umamiToken string @@ -110,6 +114,8 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h isDebug: config.Debug, isDisabled: config.Disabled, logHandler: log.New(os.Stdout, "", 0), + // Umami API does not support batching https://github.com/umami-software/umami/discussions/1473 + queue: make(chan *UmamiPayload, config.QueueSize), umamiHost: config.UmamiHost, umamiToken: config.UmamiToken, @@ -127,7 +133,7 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h } if !h.isDisabled { - err := h.connect(config) + err := h.connect(ctx, config) if err != nil { h.error(err.Error()) h.error("unable to connect to Umami, the plugin is disabled") @@ -140,18 +146,20 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h h.error("configuration error, the plugin is disabled") h.isDisabled = true } + + go h.startWorker(ctx) } return h, nil } -func (h *UmamiFeeder) connect(config *Config) error { +func (h *UmamiFeeder) connect(ctx context.Context, config *Config) error { if h.umamiHost == "" { return fmt.Errorf("`umamiHost` is not set") } if config.UmamiUsername != "" && config.UmamiPassword != "" { - token, err := getToken(h.umamiHost, config.UmamiUsername, config.UmamiPassword) + token, err := getToken(ctx, h.umamiHost, config.UmamiUsername, config.UmamiPassword) if err != nil { return fmt.Errorf("failed to get token: %w", err) } @@ -169,7 +177,7 @@ func (h *UmamiFeeder) connect(config *Config) error { } if h.umamiToken != "" { - websites, err := fetchWebsites(h.umamiHost, h.umamiToken, h.umamiTeamId) + websites, err := fetchWebsites(ctx, h.umamiHost, h.umamiToken, h.umamiTeamId) if err != nil { return fmt.Errorf("failed to fetch websites: %w", err) } @@ -219,7 +227,7 @@ func (h *UmamiFeeder) verifyConfig(config *Config) error { func (h *UmamiFeeder) ServeHTTP(rw http.ResponseWriter, req *http.Request) { if !h.isDisabled && h.shouldTrack(req) { - go h.trackRequest(req) + h.submitToFeed(req) } h.next.ServeHTTP(rw, req) @@ -310,31 +318,6 @@ func (h *UmamiFeeder) shouldTrackResource(url string) bool { return false } -func (h *UmamiFeeder) trackRequest(req *http.Request) { - hostname := parseDomainFromHost(req.Host) - websiteId, ok := h.websites[hostname] - if !ok { - website, err := createWebsite(h.umamiHost, h.umamiToken, h.umamiTeamId, hostname) - if err != nil { - h.error("failed to create website: " + err.Error()) - return - } - - h.websites[website.Domain] = website.ID - websiteId = website.ID - h.debug("created website for: %s", website.Domain) - } - - sendBody, sendHeaders := buildSendBody(req, websiteId) - h.debug("sending tracking request %s with body %v %v", req.URL, sendBody, sendHeaders) - - _, err := sendRequest(h.umamiHost+"/api/send", sendBody, sendHeaders) - if err != nil { - h.error("failed to send tracking: " + err.Error()) - return - } -} - func (h *UmamiFeeder) error(message string) { if h.logHandler != nil { now := time.Now().Format("2006-01-02T15:04:05Z") diff --git a/umami_send.go b/umami_send.go deleted file mode 100644 index 944efde..0000000 --- a/umami_send.go +++ /dev/null @@ -1,71 +0,0 @@ -package traefik_umami_feeder - -import ( - "net/http" - "regexp" - "strings" -) - -// Copied from https://github.com/1cedsoda/traefik-umami-plugin/blob/master/umami_tracking.go -// Licensed as Apache-2.0 license - -type SendPayload struct { - Website string `json:"website"` - Hostname string `json:"hostname"` - Language string `json:"language,omitempty"` - Url string `json:"url"` - Referrer string `json:"referrer,omitempty"` - Name string `json:"name,omitempty"` - Data map[string]interface{} `json:"data,omitempty"` -} - -type SendBody struct { - Payload SendPayload `json:"payload"` - Type string `json:"type"` -} - -func buildPayload(req *http.Request, websiteId string) SendPayload { - return SendPayload{ - Website: websiteId, - Hostname: parseDomainFromHost(req.Host), - Language: parseAcceptLanguage(req.Header.Get("Accept-Language")), - Url: req.URL.String(), - Referrer: req.Referer(), - Data: map[string]interface{}{}, - } -} - -// opts the port from the host. -func parseDomainFromHost(host string) string { - // check if the host has a port - if strings.Contains(host, ":") { - host = strings.Split(host, ":")[0] - } - return strings.ToLower(host) -} - -const parseAcceptLanguagePattern = `([a-zA-Z\-]+)(?:;q=\d\.\d)?(?:,\s)?` - -var parseAcceptLanguageRegexp = regexp.MustCompile(parseAcceptLanguagePattern) - -func parseAcceptLanguage(acceptLanguage string) string { - matches := parseAcceptLanguageRegexp.FindAllStringSubmatch(acceptLanguage, -1) - if len(matches) == 0 { - return "" - } - return matches[0][1] -} - -func buildSendBody(clientReq *http.Request, websiteId string) (SendBody, http.Header) { - body := SendBody{ - Payload: buildPayload(clientReq, websiteId), - Type: "event", - } - - var headers = make(http.Header) - headers.Set("Content-Type", "application/json") - copyHeaders(headers, clientReq.Header) - removeHeaders(headers, hopHeaders...) - writeXForwardedHeaders(headers, clientReq) - return body, headers -} diff --git a/umami_token.go b/umami_token.go index 910dd25..229c6e0 100644 --- a/umami_token.go +++ b/umami_token.go @@ -1,5 +1,7 @@ package traefik_umami_feeder +import "context" + type Auth struct { Username string `json:"username"` Password string `json:"password"` @@ -9,9 +11,9 @@ type AuthResponse struct { Token string `json:"token"` } -func getToken(umamiHost string, umamiUsername string, umamiPassword string) (string, error) { +func getToken(ctx context.Context, umamiHost string, umamiUsername string, umamiPassword string) (string, error) { var result AuthResponse - err := sendRequestAndParse(umamiHost+"/api/auth/login", Auth{ + err := sendRequestAndParse(ctx, umamiHost+"/api/auth/login", Auth{ Username: umamiUsername, Password: umamiPassword, }, nil, &result) diff --git a/umami_utils.go b/umami_utils.go index 1af6936..56eb762 100644 --- a/umami_utils.go +++ b/umami_utils.go @@ -7,9 +7,12 @@ import ( "fmt" "io" "net/http" + "regexp" + "strings" + "time" ) -func sendRequest(url string, body interface{}, headers http.Header) (*http.Response, error) { +func sendRequest(ctx context.Context, url string, body interface{}, headers http.Header) (*http.Response, error) { var req *http.Request var err error @@ -19,9 +22,9 @@ func sendRequest(url string, body interface{}, headers http.Header) (*http.Respo return nil, err } - req, err = http.NewRequestWithContext(context.Background(), http.MethodPost, url, bytes.NewReader(bodyJson)) + req, err = http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(bodyJson)) } else { - req, err = http.NewRequestWithContext(context.Background(), http.MethodGet, url, nil) + req, err = http.NewRequestWithContext(ctx, http.MethodGet, url, nil) } if err != nil { @@ -36,27 +39,28 @@ func sendRequest(url string, body interface{}, headers http.Header) (*http.Respo req.Header.Set("Content-Type", "application/json") } - client := &http.Client{} - response, err := client.Do(req) + client := &http.Client{Timeout: 10 * time.Second} + resp, err := client.Do(req) if err != nil { return nil, err } - status := response.StatusCode + status := resp.StatusCode if status < 200 || status >= 300 { return nil, fmt.Errorf("request failed with status %d", status) } - return response, nil + return resp, nil } -func sendRequestAndParse(url string, body interface{}, headers http.Header, value interface{}) error { - resp, err := sendRequest(url, body, headers) - +func sendRequestAndParse(ctx context.Context, url string, body interface{}, headers http.Header, value interface{}) error { + resp, err := sendRequest(ctx, url, body, headers) if err != nil { return err } - defer resp.Body.Close() + defer func() { + _ = resp.Body.Close() + }() respBody, err := io.ReadAll(resp.Body) if err != nil { @@ -70,3 +74,32 @@ func sendRequestAndParse(url string, body interface{}, headers http.Header, valu return nil } + +// opts the port from the host. +func parseDomainFromHost(host string) string { + // check if the host has a port + if strings.Contains(host, ":") { + host = strings.Split(host, ":")[0] + } + return strings.ToLower(host) +} + +const parseAcceptLanguagePattern = `([a-zA-Z\-]+)(?:;q=\d\.\d)?(?:,\s)?` + +var parseAcceptLanguageRegexp = regexp.MustCompile(parseAcceptLanguagePattern) + +func parseAcceptLanguage(acceptLanguage string) string { + matches := parseAcceptLanguageRegexp.FindAllStringSubmatch(acceptLanguage, -1) + if len(matches) == 0 { + return "" + } + return matches[0][1] +} + +func copyHeaders(dst, src http.Header, headersToCopy []string) { + for _, key := range headersToCopy { + if values := src.Values(key); len(values) > 0 { + dst[key] = values + } + } +} diff --git a/umami_websites.go b/umami_websites.go index 5373d50..6ad4891 100644 --- a/umami_websites.go +++ b/umami_websites.go @@ -1,6 +1,7 @@ package traefik_umami_feeder import ( + "context" "net/http" "time" ) @@ -21,12 +22,12 @@ type Website struct { CreatedAt time.Time `json:"createdAt,omitempty"` } -func createWebsite(umamiHost string, umamiToken string, teamId string, websiteDomain string) (*Website, error) { +func createWebsite(ctx context.Context, umamiHost string, umamiToken string, teamId string, websiteDomain string) (*Website, error) { var headers = make(http.Header) headers.Set("Authorization", "Bearer "+umamiToken) var result Website - err := sendRequestAndParse(umamiHost+"/api/websites", Website{ + err := sendRequestAndParse(ctx, umamiHost+"/api/websites", Website{ Name: websiteDomain, Domain: websiteDomain, TeamId: teamId, @@ -39,7 +40,7 @@ func createWebsite(umamiHost string, umamiToken string, teamId string, websiteDo return &result, nil } -func fetchWebsites(umamiHost string, umamiToken string, teamId string) (*[]Website, error) { +func fetchWebsites(ctx context.Context, umamiHost string, umamiToken string, teamId string) (*[]Website, error) { var headers = make(http.Header) headers.Set("Authorization", "Bearer "+umamiToken) @@ -49,7 +50,7 @@ func fetchWebsites(umamiHost string, umamiToken string, teamId string) (*[]Websi } var result WebsitesResponse - err := sendRequestAndParse(url, nil, headers, &result) + err := sendRequestAndParse(ctx, url, nil, headers, &result) if err != nil { return nil, err diff --git a/umami_worker.go b/umami_worker.go new file mode 100644 index 0000000..d7e3c4b --- /dev/null +++ b/umami_worker.go @@ -0,0 +1,130 @@ +package traefik_umami_feeder + +import ( + "context" + "fmt" + "net/http" +) + +type SendPayload struct { + Website string `json:"website"` + Hostname string `json:"hostname"` + Ip string `json:"ip"` + Language string `json:"language,omitempty"` + Referrer string `json:"referrer,omitempty"` + Url string `json:"url"` + //Data map[string]interface{} `json:"data,omitempty"` // Additional data for the event + //Name string `json:"name,omitempty"` // Event name (for custom events) + //Screen string `json:"screen,omitempty"` // Screen resolution (ex. "1920x1080") + //Tag string `json:"tag,omitempty"` + //Title string `json:"title,omitempty"` // Page title +} + +type SendBody struct { + Payload SendPayload `json:"payload"` + Type string `json:"type"` +} + +type UmamiPayload struct { + body SendBody + headers http.Header +} + +var headersToCopy = []string{ + "User-Agent", + "X-Real-Op", + "X-Forwarded-For", + "cf-ipcountry", + "cf-region-code", + "cf-ipcity", + "cf-connecting-ip", + "x-vercel-ip-country", + "x-vercel-ip-country-region", + "x-vercel-ip-city", +} + +// Copied and adapted from https://github.com/safing/plausiblefeeder/blob/master/event.go +// Licensed as MIT license + +func (h *UmamiFeeder) submitToFeed(req *http.Request) { + body := SendBody{ + Payload: SendPayload{ + Hostname: parseDomainFromHost(req.Host), + Language: parseAcceptLanguage(req.Header.Get("Accept-Language")), + Referrer: req.Referer(), + Url: req.URL.String(), + }, + Type: "event", + } + + var headers = make(http.Header) + copyHeaders(headers, req.Header, headersToCopy) + + payload := &UmamiPayload{body: body, headers: headers} + + select { + case h.queue <- payload: + default: + h.error("failed to submit event: queue full") + } +} + +func (h *UmamiFeeder) startWorker(ctx context.Context) { + for { + err := h.umamiEventFeeder(ctx) + if err != nil { + h.error("worker failed: " + err.Error()) + } else { + return + } + } +} + +func (h *UmamiFeeder) umamiEventFeeder(ctx context.Context) (err error) { + defer func() { + // Recover from panic. + panicVal := recover() + if panicVal != nil { + h.error("panic: " + fmt.Sprint(panicVal)) + } + }() + + for { + // Wait for event. + select { + case <-ctx.Done(): + h.error("worker shutting down (canceled)") + return nil + + case event := <-h.queue: + h.reportEventToUmami(ctx, event) + } + } +} + +func (h *UmamiFeeder) reportEventToUmami(ctx context.Context, event *UmamiPayload) { + hostname := event.body.Payload.Hostname + websiteId, ok := h.websites[hostname] + if !ok { + website, err := createWebsite(ctx, h.umamiHost, h.umamiToken, h.umamiTeamId, hostname) + if err != nil { + h.error("failed to create website: " + err.Error()) + return + } + + h.websites[website.Domain] = website.ID + websiteId = website.ID + h.debug("created website for: %s", website.Domain) + } + event.body.Payload.Website = websiteId + + h.debug("sending tracking request %s with body %v %v", event.body.Payload.Url, event.body, event.headers) + resp, err := sendRequest(ctx, h.umamiHost+"/api/send", event.body, event.headers) + defer func() { + _ = resp.Body.Close() + }() + if err != nil { + h.error("failed to send tracking: " + err.Error()) + return + } +}