pico
created pr with
32.1
added 32.2
1: 7ec3569 = 1: 7ec3569 feat(auth): subscribe to pico's metric-drain pipe
2: 8a197f0 = 2: 8a197f0 chore: update pubsub
3: d4bda15 = 3: d4bda15 refactor: use pipe for analytics
-: ------- > 4: 2b7c358 chore: prep for release
cmds
checkout latest patchset:
ssh pr.pico.sh print 32 | git am -3checkout any patchset in a patch request:
ssh pr.pico.sh print 32.[rev] | git am -3add changes to patch request:
git format-patch main --stdout | ssh pr.pico.sh pr add 32
Patchset
32.2
feat(auth): subscribe to pico's metric-drain pipe
Eric Bower
2024-11-11T02:41:27ZWe have a lot of services that need to record site usage analytics so we need a distributed way to receive these events. This overloads the auth service since it's now serving as a destination for our metric-drain.
Semantic diff summary
2 added,
8 modified,
0 signature changed,
0 removed
across 6 analyzed files
+33
-1
auth/api.go
#
| ... | ... | @@ -639,6 +641,31 @@ func handler(routes []shared.Route, client *Client) http.HandlerFunc { | |
| 639 | 641 | } | |
| 640 | 642 | } | |
| 641 | 643 | ||
| 644 | + | func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger) { | |
| 645 | + | conn := shared.NewPicoPipeClient() | |
| 646 | + | stdoutPipe, err := pubsub.RemoteSub("sub metric-drain -k", ctx, conn) | |
| 647 | + | ||
| 648 | + | if err != nil { | |
| 649 | + | logger.Error("could not sub to metric-drain", "err", err) | |
| 650 | + | return | |
| 651 | + | } | |
| 652 | + | ||
| 653 | + | scanner := bufio.NewScanner(stdoutPipe) | |
| 654 | + | for scanner.Scan() { | |
| 655 | + | line := scanner.Text() | |
| 656 | + | view := db.AnalyticsVisits{} | |
| 657 | + | err := json.Unmarshal([]byte(line), &view) | |
| 658 | + | if err != nil { | |
| 659 | + | logger.Error("json unmarshal", "err", err) | |
| 660 | + | continue | |
| 661 | + | } | |
| 662 | + | err = dbpool.InsertVisit(&view) | |
| 663 | + | if err != nil { | |
| 664 | + | logger.Error("could not insert view record", "err", err) | |
| 665 | + | } | |
| 666 | + | } | |
| 667 | + | } | |
| 668 | + | ||
| 642 | 669 | type AuthCfg struct { | |
| 643 | 670 | Debug bool | |
| 644 | 671 | Port string |
| ... | ... | @@ -679,6 +711,6 @@ func StartApiServer() { | |
| 679 | 711 | client.Logger.Info("starting server on port", "port", cfg.Port) | |
| 680 | 712 | err := http.ListenAndServe(portStr, router) | |
| 681 | 713 | if err != nil { | |
| 682 | - | client.Logger.Info(err.Error()) | |
| 714 | + | client.Logger.Info("http-serve", "err", err.Error()) | |
| 683 | 715 | } | |
| 684 | 716 | } |
+10
-10
db/db.go
#
| ... | ... | @@ -161,16 +161,16 @@ type PostAnalytics struct { | |
| 161 | 161 | } | |
| 162 | 162 | ||
| 163 | 163 | type AnalyticsVisits struct { | |
| 164 | - | ID string | |
| 165 | - | UserID string | |
| 166 | - | ProjectID string | |
| 167 | - | PostID string | |
| 168 | - | Host string | |
| 169 | - | Path string | |
| 170 | - | IpAddress string | |
| 171 | - | UserAgent string | |
| 172 | - | Referer string | |
| 173 | - | Status int | |
| 164 | + | ID string `json:"id"` | |
| 165 | + | UserID string `json:"user_id"` | |
| 166 | + | ProjectID string `json:"project_id"` | |
| 167 | + | PostID string `json:"post_id"` | |
| 168 | + | Host string `json:"host"` | |
| 169 | + | Path string `json:"path"` | |
| 170 | + | IpAddress string `json:"ip_adress"` | |
| 171 | + | UserAgent string `json:"user_agent"` | |
| 172 | + | Referer string `json:"referer"` | |
| 173 | + | Status int `json:"status"` | |
| 174 | 174 | } | |
| 175 | 175 | ||
| 176 | 176 | type VisitInterval struct { |
+2
-7
pico/cli.go
#
| ... | ... | @@ -70,13 +70,8 @@ func (c *Cmd) notifications() error { | |
| 70 | 70 | } | |
| 71 | 71 | ||
| 72 | 72 | func (c *Cmd) logs(ctx context.Context) error { | |
| 73 | - | stdoutPipe, err := pipeLogger.ConnectToLogs(ctx, &pipeLogger.PubSubConnectionInfo{ | |
| 74 | - | RemoteHost: utils.GetEnv("PICO_PIPE_ENDPOINT", "pipe.pico.sh:22"), | |
| 75 | - | KeyLocation: utils.GetEnv("PICO_PIPE_KEY", "ssh_data/term_info_ed25519"), | |
| 76 | - | KeyPassphrase: utils.GetEnv("PICO_PIPE_PASSPHRASE", ""), | |
| 77 | - | RemoteHostname: utils.GetEnv("PICO_PIPE_REMOTE_HOST", "pipe.pico.sh"), | |
| 78 | - | RemoteUser: utils.GetEnv("PICO_PIPE_USER", "pico"), | |
| 79 | - | }) | |
| 73 | + | conn := shared.NewPicoPipeClient() | |
| 74 | + | stdoutPipe, err := pipeLogger.ConnectToLogs(ctx, conn) | |
| 80 | 75 | ||
| 81 | 76 | if err != nil { | |
| 82 | 77 | return err |
+3
-8
tui/logs/logs.go
#
| ... | ... | @@ -11,6 +11,7 @@ import ( | |
| 11 | 11 | "github.com/charmbracelet/bubbles/viewport" | |
| 12 | 12 | tea "github.com/charmbracelet/bubbletea" | |
| 13 | 13 | "github.com/charmbracelet/lipgloss" | |
| 14 | + | "github.com/picosh/pico/shared" | |
| 14 | 15 | "github.com/picosh/pico/tui/common" | |
| 15 | 16 | "github.com/picosh/pico/tui/pages" | |
| 16 | 17 | "github.com/picosh/utils" |
| ... | ... | @@ -170,14 +171,8 @@ func (m Model) waitForActivity(sub chan map[string]any) tea.Cmd { | |
| 170 | 171 | ||
| 171 | 172 | func (m Model) connectLogs(sub chan map[string]any) tea.Cmd { | |
| 172 | 173 | return func() tea.Msg { | |
| 173 | - | stdoutPipe, err := pipeLogger.ConnectToLogs(m.ctx, &pipeLogger.PubSubConnectionInfo{ | |
| 174 | - | RemoteHost: utils.GetEnv("PICO_PIPE_ENDPOINT", "pipe.pico.sh:22"), | |
| 175 | - | KeyLocation: utils.GetEnv("PICO_PIPE_KEY", "ssh_data/term_info_ed25519"), | |
| 176 | - | KeyPassphrase: utils.GetEnv("PICO_PIPE_PASSPHRASE", ""), | |
| 177 | - | RemoteHostname: utils.GetEnv("PICO_PIPE_REMOTE_HOST", "pipe.pico.sh"), | |
| 178 | - | RemoteUser: utils.GetEnv("PICO_PIPE_USER", "pico"), | |
| 179 | - | }) | |
| 180 | - | ||
| 174 | + | conn := shared.NewPicoPipeClient() | |
| 175 | + | stdoutPipe, err := pipeLogger.ConnectToLogs(m.ctx, conn) | |
| 181 | 176 | if err != nil { | |
| 182 | 177 | return errMsg(err) | |
| 183 | 178 | } |