dashboard / erock/pico / reactor(metric-drain): use caddy json format #35 rss

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

diff --git a/auth/api.go b/auth/api.go
index 9c38bdc..8840af7 100644
--- a/auth/api.go
+++ b/auth/api.go
@@ -14,6 +14,7 @@ import (
 	"log/slog"
 	"net/http"
 	"net/url"
+	"strings"
 	"time"
 
 	"github.com/gorilla/feeds"
@@ -578,6 +579,89 @@ func checkoutHandler() http.HandlerFunc {
 	}
 }
 
+type AccessLogReq struct {
+	RemoteIP   string `json:"remote_ip"`
+	RemotePort string `json:"remote_port"`
+	ClientIP   string `json:"client_ip"`
+	Method     string `json:"method"`
+	Host       string `json:"host"`
+	Uri        string `json:"uri"`
+	Headers    struct {
+		UserAgent string `json:"User-Agent"`
+		Referer   string `json:"Referer"`
+	} `json:"headers"`
+	Tls struct {
+		ServerName string `json:"server_name"`
+	} `json:"tls"`
+}
+
+type CaddyAccessLog struct {
+	Request AccessLogReq `json:"request"`
+	Status  int          `json:"status"`
+}
+
+func deserializeCaddyAccessLog(dbpool db.DB, access *CaddyAccessLog) (*db.AnalyticsVisits, error) {
+	spaceRaw := strings.SplitN(access.Request.Tls.ServerName, ".", 2)
+	space := spaceRaw[0]
+	host := access.Request.Host
+	path := access.Request.Uri
+	subdomain := ""
+
+	// grab subdomain based on host
+	if strings.HasSuffix(host, "tuns.sh") {
+		subdomain = strings.TrimSuffix(host, ".tuns.sh")
+	} else if strings.HasSuffix(host, "pgs.sh") {
+		subdomain = strings.TrimSuffix(host, ".pgs.sh")
+	} else if strings.HasSuffix(host, "prose.sh") {
+		subdomain = strings.TrimSuffix(host, ".prose.sh")
+	} else {
+		subdomain = shared.GetCustomDomain(host, space)
+	}
+
+	// get user and namespace details from subdomain
+	props, err := shared.GetProjectFromSubdomain(subdomain)
+	if err != nil {
+		return nil, err
+	}
+	// get user ID
+	user, err := dbpool.FindUserForName(props.Username)
+	if err != nil {
+		return nil, err
+	}
+
+	projectID := ""
+	postID := ""
+	if space == "pgs" { // figure out project ID
+		project, err := dbpool.FindProjectByName(user.ID, props.ProjectName)
+		if err != nil {
+			return nil, err
+		}
+		projectID = project.ID
+	} else if space == "prose" { // figure out post ID
+		if path == "" || path == "/" {
+		} else {
+			post, err := dbpool.FindPostWithSlug(path, user.ID, space)
+			if err != nil {
+				return nil, err
+			}
+			postID = post.ID
+		}
+	}
+
+	return &db.AnalyticsVisits{
+		UserID:    user.ID,
+		ProjectID: projectID,
+		PostID:    postID,
+		Namespace: space,
+		Host:      host,
+		Path:      path,
+		IpAddress: access.Request.ClientIP,
+		UserAgent: access.Request.Headers.UserAgent,
+		Referer:   access.Request.Headers.Referer, // TODO: I don't see referer in the access log
+		Status:    access.Status,
+	}, nil
+}
+
 func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secret string) {
 	drain := metrics.ReconnectReadMetrics(
 		ctx,
@@ -587,36 +671,32 @@ func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secr
 		-1,
 	)
 
-	for {
-		scanner := bufio.NewScanner(drain)
-		for scanner.Scan() {
-			line := scanner.Text()
-			visit := db.AnalyticsVisits{}
-			err := json.Unmarshal([]byte(line), &visit)
-			if err != nil {
-				logger.Error("json unmarshal", "err", err)
-				continue
-			}
-
-			user := slog.Any("userId", visit.UserID)
-
-			err = shared.AnalyticsVisitFromVisit(&visit, dbpool, secret)
-			if err != nil {
-				if !errors.Is(err, shared.ErrAnalyticsDisabled) {
-					logger.Info("could not record analytics visit", "reason", err, "visit", visit, user)
-					continue
-				}
-			}
+	scanner := bufio.NewScanner(drain)
+	for scanner.Scan() {
+		line := scanner.Text()
+		accessLog := CaddyAccessLog{}
+		err := json.Unmarshal([]byte(line), &accessLog)
+		if err != nil {
+			logger.Error("json unmarshal", "err", err)
+			continue
+		}
 
-			logger.Info("inserting visit", "visit", visit, user)
-			err = dbpool.InsertVisit(&visit)
-			if err != nil {
-				logger.Error("could not insert visit record", "err", err, "visit", visit, user)
+		visit, err := deserializeCaddyAccessLog(dbpool, &accessLog)
+		if err != nil {
+			logger.Error("cannot deserialize access log", "err", err)
+			continue
+		}
+		err = shared.AnalyticsVisitFromVisit(visit, dbpool, secret)
+		if err != nil {
+			if !errors.Is(err, shared.ErrAnalyticsDisabled) {
+				logger.Info("could not record analytics visit", "reason", err)
 			}
 		}
 
-		if scanner.Err() != nil {
-			logger.Error("scanner error", "err", scanner.Err())
+		logger.Info("inserting visit", "visit", visit)
+		err = dbpool.InsertVisit(visit)
+		if err != nil {
+			logger.Error("could not insert visit record", "err", err)
 		}
 	}
 }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

diff --git a/caddy.json b/caddy.json
new file mode 100644
index 0000000..49e80ec
--- /dev/null
+++ b/caddy.json
@@ -0,0 +1,40 @@
+{
+  "level": "info",
+  "ts": 1731644477.313701,
+  "logger": "http.log.access",
+  "msg": "handled request",
+  "request": {
+    "remote_ip": "127.0.0.1",
+    "remote_port": "40400",
+    "client_ip": "127.0.0.1",
+    "proto": "HTTP/2.0",
+    "method": "GET",
+    "host": "pgs.sh",
+    "uri": "/",
+    "headers": { "User-Agent": ["Blackbox Exporter/0.24.0"] },
+    "tls": {
+      "resumed": false,
+      "version": 772,
+      "cipher_suite": 4865,
+      "proto": "h2",
+      "server_name": "pgs.sh"
+    }
+  },
+  "bytes_read": 0,
+  "user_id": "",
+  "duration": 0.001207084,
+  "size": 3718,
+  "status": 200,
+  "resp_headers": {
+    "Referrer-Policy": ["no-referrer-when-downgrade"],
+    "Strict-Transport-Security": ["max-age=31536000;"],
+    "X-Content-Type-Options": ["nosniff"],
+    "X-Frame-Options": ["DENY"],
+    "Server": ["Caddy"],
+    "Alt-Svc": ["h3=\":443\"; ma=2592000"],
+    "Date": ["Fri, 15 Nov 2024 04:21:17 GMT"],
+    "Content-Type": ["text/html; charset=utf-8"],
+    "X-Xss-Protection": ["1; mode=block"],
+    "Permissions-Policy": ["interest-cohort=()"]
+  }
+}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

diff --git a/pgs/tunnel.go b/pgs/tunnel.go
index b635c8e..accacc5 100644
--- a/pgs/tunnel.go
+++ b/pgs/tunnel.go
@@ -51,7 +51,7 @@ func createHttpHandler(apiConfig *shared.ApiConfig) CtxHttpBridge {
 			"pubkey", pubkeyStr,
 		)
 
-		props, err := getProjectFromSubdomain(subdomain)
+		props, err := shared.GetProjectFromSubdomain(subdomain)
 		if err != nil {
 			log.Error(err.Error())
 			return http.HandlerFunc(shared.UnauthorizedHandler)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

diff --git a/pgs/web.go b/pgs/web.go
index 685f6ef..0903c1a 100644
--- a/pgs/web.go
+++ b/pgs/web.go
@@ -177,7 +177,7 @@ func (web *WebRouter) checkHandler(w http.ResponseWriter, r *http.Request) {
 
 		if !strings.Contains(hostDomain, appDomain) {
 			subdomain := shared.GetCustomDomain(hostDomain, cfg.Space)
-			props, err := getProjectFromSubdomain(subdomain)
+			props, err := shared.GetProjectFromSubdomain(subdomain)
 			if err != nil {
 				logger.Error(
 					"could not get project from subdomain",
@@ -333,7 +333,7 @@ func (web *WebRouter) ServeAsset(fname string, opts *storage.ImgProcessOpts, fro
 		"host", r.Host,
 	)
 
-	props, err := getProjectFromSubdomain(subdomain)
+	props, err := shared.GetProjectFromSubdomain(subdomain)
 	if err != nil {
 		logger.Info(
 			"could not determine project from subdomain",
@@ -450,20 +450,3 @@ func (web *WebRouter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctx = context.WithValue(ctx, shared.CtxSubdomainKey{}, subdomain)
 	router.ServeHTTP(w, r.WithContext(ctx))
 }
-
-type SubdomainProps struct {
-	ProjectName string
-	Username    string
-}
-
-func getProjectFromSubdomain(subdomain string) (*SubdomainProps, error) {
-	props := &SubdomainProps{}
-	strs := strings.SplitN(subdomain, "-", 2)
-	props.Username = strs[0]
-	if len(strs) == 2 {
-		props.ProjectName = strs[1]
-	} else {
-		props.ProjectName = props.Username
-	}
-	return props, nil
-}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

diff --git a/shared/api.go b/shared/api.go
index 96a4e68..a33ad59 100644
--- a/shared/api.go
+++ b/shared/api.go
@@ -13,6 +13,23 @@ import (
 	"github.com/picosh/utils"
 )
 
+type SubdomainProps struct {
+	ProjectName string
+	Username    string
+}
+
+func GetProjectFromSubdomain(subdomain string) (*SubdomainProps, error) {
+	props := &SubdomainProps{}
+	strs := strings.SplitN(subdomain, "-", 2)
+	props.Username = strs[0]
+	if len(strs) == 2 {
+		props.ProjectName = strs[1]
+	} else {
+		props.ProjectName = props.Username
+	}
+	return props, nil
+}
+
 func CorsHeaders(headers http.Header) {
 	headers.Add("Access-Control-Allow-Origin", "*")
 	headers.Add("Vary", "Origin")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

diff --git a/auth/api.go b/auth/api.go
index 8840af7..6d997ab 100644
--- a/auth/api.go
+++ b/auth/api.go
@@ -22,6 +22,7 @@ import (
 	"github.com/picosh/pico/db/postgres"
 	"github.com/picosh/pico/shared"
 	"github.com/picosh/utils"
+	"github.com/picosh/utils/pipe"
 	"github.com/picosh/utils/pipe/metrics"
 )
 
@@ -662,6 +663,30 @@ func deserializeCaddyAccessLog(dbpool db.DB, access *CaddyAccessLog) (*db.Analyt
 	}, nil
 }
 
+func containerDrainSub(ctx context.Context, logger *slog.Logger) {
+	drain := pipe.NewReconnectReadWriteCloser(
+		ctx,
+		logger,
+		shared.NewPicoPipeClient(),
+		"container logs",
+		"sub container-drain -k",
+		100,
+		-1,
+	)
+
+	fmt.Println("WTFFFFFF")
+	scanner := bufio.NewScanner(drain)
+	for scanner.Scan() {
+		line := scanner.Text()
+		fmt.Println("HMMMM", line)
+		if strings.Contains(line, "http.log.access") {
+			clean := strings.TrimSpace(line)
+			fmt.Println("LINE", clean)
+			// TODO: send to metric drain
+		}
+	}
+}
+
 func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secret string) {
 	drain := metrics.ReconnectReadMetrics(
 		ctx,
@@ -699,6 +724,7 @@ func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secr
 			logger.Error("could not insert visit record", "err", err)
 		}
 	}
+	fmt.Println("DROPINNGGGGGGG")
 }
 
 func authMux(apiConfig *shared.ApiConfig) *http.ServeMux {
@@ -769,6 +795,9 @@ func StartApiServer() {
 
 	// gather metrics in the auth service
 	go metricDrainSub(ctx, db, logger, cfg.Secret)
+	// convert container logs to access logs
+	// go containerDrainSub(ctx, logger)
+
 	defer ctx.Done()
 
 	apiConfig := &shared.ApiConfig{

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

diff --git a/auth/api.go b/auth/api.go
index 6d997ab..c25e2e7 100644
--- a/auth/api.go
+++ b/auth/api.go
@@ -588,8 +588,8 @@ type AccessLogReq struct {
 	Host       string `json:"host"`
 	Uri        string `json:"uri"`
 	Headers    struct {
-		UserAgent string `json:"User-Agent"`
-		Referer   string `json:"Referer"`
+		UserAgent []string `json:"User-Agent"`
+		Referer   string   `json:"Referer"`
 	} `json:"headers"`
 	Tls struct {
 		ServerName string `json:"server_name"`
@@ -657,36 +657,72 @@ func deserializeCaddyAccessLog(dbpool db.DB, access *CaddyAccessLog) (*db.Analyt
 		Host:      host,
 		Path:      path,
 		IpAddress: access.Request.ClientIP,
-		UserAgent: access.Request.Headers.UserAgent,
+		UserAgent: strings.Join(access.Request.Headers.UserAgent, " "),
 		Referer:   access.Request.Headers.Referer, // TODO: I don't see referer in the access log
 		Status:    access.Status,
 	}, nil
 }
 
-func containerDrainSub(ctx context.Context, logger *slog.Logger) {
+// this feels really stupid because i'm taking containter-drain,
+// filtering it, and then sending it to metric-drain.  The
+// metricDrainSub function listens on the metric-drain and saves it.
+// So why not just call the necessary functions to save the visit?
+// We want to be able to have pipe as a debugging tool which means we
+// can sub to `metric-drain` and have a nice clean output to look use.
+func containerDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger) {
+	info := shared.NewPicoPipeClient()
 	drain := pipe.NewReconnectReadWriteCloser(
 		ctx,
 		logger,
-		shared.NewPicoPipeClient(),
-		"container logs",
+		info,
+		"container drain",
 		"sub container-drain -k",
 		100,
 		-1,
 	)
 
-	fmt.Println("WTFFFFFF")
-	scanner := bufio.NewScanner(drain)
-	for scanner.Scan() {
-		line := scanner.Text()
-		fmt.Println("HMMMM", line)
-		if strings.Contains(line, "http.log.access") {
-			clean := strings.TrimSpace(line)
-			fmt.Println("LINE", clean)
-			// TODO: send to metric drain
+	send := pipe.NewReconnectReadWriteCloser(
+		ctx,
+		logger,
+		info,
+		"from container drain to metric drain",
+		"pub metric-drain -b=false",
+		100,
+		-1,
+	)
+
+	for {
+		scanner := bufio.NewScanner(drain)
+		for scanner.Scan() {
+			line := scanner.Text()
+			if strings.Contains(line, "http.log.access") {
+				clean := strings.TrimSpace(line)
+				visit, err := accessLogToVisit(dbpool, clean)
+				if err != nil {
+					logger.Error("could not convert access log to a visit", "err", err)
+					continue
+				}
+				jso, err := json.Marshal(visit)
+				if err != nil {
+					logger.Error("could not marshal json of a visit", "err", err)
+					continue
+				}
+				_, _ = send.Write(jso)
+			}
 		}
 	}
 }
 
+func accessLogToVisit(dbpool db.DB, line string) (*db.AnalyticsVisits, error) {
+	accessLog := CaddyAccessLog{}
+	err := json.Unmarshal([]byte(line), &accessLog)
+	if err != nil {
+		return nil, err
+	}
+
+	return deserializeCaddyAccessLog(dbpool, &accessLog)
+}
+
 func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secret string) {
 	drain := metrics.ReconnectReadMetrics(
 		ctx,
@@ -696,35 +732,29 @@ func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secr
 		-1,
 	)
 
-	scanner := bufio.NewScanner(drain)
-	for scanner.Scan() {
-		line := scanner.Text()
-		accessLog := CaddyAccessLog{}
-		err := json.Unmarshal([]byte(line), &accessLog)
-		if err != nil {
-			logger.Error("json unmarshal", "err", err)
-			continue
-		}
-
-		visit, err := deserializeCaddyAccessLog(dbpool, &accessLog)
-		if err != nil {
-			logger.Error("cannot deserialize access log", "err", err)
-			continue
-		}
-		err = shared.AnalyticsVisitFromVisit(visit, dbpool, secret)
-		if err != nil {
-			if !errors.Is(err, shared.ErrAnalyticsDisabled) {
-				logger.Info("could not record analytics visit", "reason", err)
+	for {
+		scanner := bufio.NewScanner(drain)
+		for scanner.Scan() {
+			line := scanner.Text()
+			visit, err := accessLogToVisit(dbpool, line)
+			if err != nil {
+				logger.Error("could not convert access log to a visit", "err", err)
+				continue
+			}
+			err = shared.AnalyticsVisitFromVisit(visit, dbpool, secret)
+			if err != nil {
+				if !errors.Is(err, shared.ErrAnalyticsDisabled) {
+					logger.Info("could not record analytics visit", "reason", err)
+				}
 			}
-		}
 
-		logger.Info("inserting visit", "visit", visit)
-		err = dbpool.InsertVisit(visit)
-		if err != nil {
-			logger.Error("could not insert visit record", "err", err)
+			logger.Info("inserting visit", "visit", visit)
+			err = dbpool.InsertVisit(visit)
+			if err != nil {
+				logger.Error("could not insert visit record", "err", err)
+			}
 		}
 	}
-	fmt.Println("DROPINNGGGGGGG")
 }
 
 func authMux(apiConfig *shared.ApiConfig) *http.ServeMux {
@@ -796,7 +826,7 @@ func StartApiServer() {
 	// gather metrics in the auth service
 	go metricDrainSub(ctx, db, logger, cfg.Secret)
 	// convert container logs to access logs
-	// go containerDrainSub(ctx, logger)
+	go containerDrainSub(ctx, db, logger)
 
 	defer ctx.Done()
 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

diff --git a/auth/api.go b/auth/api.go
index c25e2e7..09d7661 100644
--- a/auth/api.go
+++ b/auth/api.go
@@ -589,7 +589,7 @@ type AccessLogReq struct {
 	Uri        string `json:"uri"`
 	Headers    struct {
 		UserAgent []string `json:"User-Agent"`
-		Referer   string   `json:"Referer"`
+		Referer   []string `json:"Referer"`
 	} `json:"headers"`
 	Tls struct {
 		ServerName string `json:"server_name"`
@@ -658,7 +658,7 @@ func deserializeCaddyAccessLog(dbpool db.DB, access *CaddyAccessLog) (*db.Analyt
 		Path:      path,
 		IpAddress: access.Request.ClientIP,
 		UserAgent: strings.Join(access.Request.Headers.UserAgent, " "),
-		Referer:   access.Request.Headers.Referer, // TODO: I don't see referer in the access log
+		Referer:   strings.Join(access.Request.Headers.Referer, " "),
 		Status:    access.Status,
 	}, nil
 }
@@ -699,7 +699,7 @@ func containerDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger) {
 				clean := strings.TrimSpace(line)
 				visit, err := accessLogToVisit(dbpool, clean)
 				if err != nil {
-					logger.Error("could not convert access log to a visit", "err", err)
+					logger.Debug("could not convert access log to a visit", "err", err)
 					continue
 				}
 				jso, err := json.Marshal(visit)
@@ -736,12 +736,13 @@ func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secr
 		scanner := bufio.NewScanner(drain)
 		for scanner.Scan() {
 			line := scanner.Text()
-			visit, err := accessLogToVisit(dbpool, line)
+			visit := db.AnalyticsVisits{}
+			err := json.Unmarshal([]byte(line), &visit)
 			if err != nil {
-				logger.Error("could not convert access log to a visit", "err", err)
+				logger.Info("could not unmarshal json", "err", err, "line", line)
 				continue
 			}
-			err = shared.AnalyticsVisitFromVisit(visit, dbpool, secret)
+			err = shared.AnalyticsVisitFromVisit(&visit, dbpool, secret)
 			if err != nil {
 				if !errors.Is(err, shared.ErrAnalyticsDisabled) {
 					logger.Info("could not record analytics visit", "reason", err)
@@ -749,7 +750,7 @@ func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secr
 			}
 
 			logger.Info("inserting visit", "visit", visit)
-			err = dbpool.InsertVisit(visit)
+			err = dbpool.InsertVisit(&visit)
 			if err != nil {
 				logger.Error("could not insert visit record", "err", err)
 			}