chaas/traefik-rybbit-feeder
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

feat: use queue to feed events, drop if overflowed

Browse source
This commit is contained in:
Oleh Astappiev 2024-12-08 20:45:44 +01:00
parent b0f0510e8e
commit 02580cce20
No known key found for this signature in database
GPG key ID: 37D7AC22589E9632
7 changed files with 198 additions and 226 deletions
Download patch file Download diff file Expand all files Collapse all files

106
forwarding.go
View file

@ -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)
}
}

47
umami.go
View file

@ -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")

71
umami_send.go
View file

@ -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
}

6
umami_token.go
View file

@ -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)

55
umami_utils.go
View file

@ -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
}
}
}

9
umami_websites.go
View file

@ -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

130
umami_worker.go Normal file
View file

@ -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
}
}