pico

reactor(metric-drain): use caddy json format open rss
created pr with 35.1 on 2024-11-15T15:03:31Z · by c8ef7d19
added 35.2 on 2024-11-23T03:34:44Z · by c8ef7d19
1: 77aaa29 ! 1: 8d56535 reactor(metric-drain): use caddy json format
-: ------- > 2: a336041 wip
-: ------- > 3: 7ae45b3 chore: wrap
-: ------- > 4: bfa5c4f done
added 35.3 on 2024-11-27T20:17:20Z · by c8ef7d19
1: 8d56535 < -: ------- reactor(metric-drain): use caddy json format
-: ------- > 1: c7eeb12 reactor(metric-drain): use caddy access logs
2: a336041 < -: ------- wip
3: 7ae45b3 < -: ------- chore: wrap
4: bfa5c4f < -: ------- done
added 35.4 on 2024-11-27T20:18:31Z · by c8ef7d19
1: c7eeb12 ! 1: 4e0839a reactor(metric-drain): use caddy access logs
changed status to accepted on 2024-11-28T03:03:53Z · by c8ef7d19
cmds
checkout latest patchset: 
ssh pr.pico.sh print 35 | git am -3
checkout any patchset in a patch request: 
ssh pr.pico.sh print 35.[rev] | git am -3
add changes to patch request: 
git format-patch main --stdout | ssh pr.pico.sh pr add 35
set PR to open (enables RSS notifications): 
ssh pr.pico.sh pr open 35
set PR to draft (stops RSS notifications): 
ssh pr.pico.sh pr draft 35

Patchset 35.2 on 2024-11-23T03:34:44Z · commit 7ae45b3

reactor(metric-drain): use caddy json format
Eric Bower 2024-11-15T15:02:24Z
 wip
Eric Bower 2024-11-23T02:06:47Z
 → chore: wrap
Eric Bower 2024-11-23T02:41:15Z
 done
Eric Bower 2024-11-23T03:32:31Z
+71 -41 auth/api.go #
......@@ -588,8 +588,8 @@ type AccessLogReq struct {
588588 Host string `json:"host"`
589589 Uri string `json:"uri"`
590590 Headers struct {
591- UserAgent string `json:"User-Agent"`
592- Referer string `json:"Referer"`
591+ UserAgent []string `json:"User-Agent"`
592+ Referer string `json:"Referer"`
593593 } `json:"headers"`
594594 Tls struct {
595595 ServerName string `json:"server_name"`
......@@ -657,36 +657,72 @@ func deserializeCaddyAccessLog(dbpool db.DB, access *CaddyAccessLog) (*db.Analyt
657657 Host: host,
658658 Path: path,
659659 IpAddress: access.Request.ClientIP,
660- UserAgent: access.Request.Headers.UserAgent,
660+ UserAgent: strings.Join(access.Request.Headers.UserAgent, " "),
661661 Referer: access.Request.Headers.Referer, // TODO: I don't see referer in the access log
662662 Status: access.Status,
663663 }, nil
664664 }
665665
666-func containerDrainSub(ctx context.Context, logger *slog.Logger) {
666+// this feels really stupid because i'm taking containter-drain,
667+// filtering it, and then sending it to metric-drain. The
668+// metricDrainSub function listens on the metric-drain and saves it.
669+// So why not just call the necessary functions to save the visit?
670+// We want to be able to have pipe as a debugging tool which means we
671+// can sub to `metric-drain` and have a nice clean output to look use.
672+func containerDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger) {
673+ info := shared.NewPicoPipeClient()
667674 drain := pipe.NewReconnectReadWriteCloser(
668675 ctx,
669676 logger,
670- shared.NewPicoPipeClient(),
671- "container logs",
677+ info,
678+ "container drain",
672679 "sub container-drain -k",
673680 100,
674681 -1,
675682 )
676683
677- fmt.Println("WTFFFFFF")
678- scanner := bufio.NewScanner(drain)
679- for scanner.Scan() {
680- line := scanner.Text()
681- fmt.Println("HMMMM", line)
682- if strings.Contains(line, "http.log.access") {
683- clean := strings.TrimSpace(line)
684- fmt.Println("LINE", clean)
685- // TODO: send to metric drain
684+ send := pipe.NewReconnectReadWriteCloser(
685+ ctx,
686+ logger,
687+ info,
688+ "from container drain to metric drain",
689+ "pub metric-drain -b=false",
690+ 100,
691+ -1,
692+ )
693+
694+ for {
695+ scanner := bufio.NewScanner(drain)
696+ for scanner.Scan() {
697+ line := scanner.Text()
698+ if strings.Contains(line, "http.log.access") {
699+ clean := strings.TrimSpace(line)
700+ visit, err := accessLogToVisit(dbpool, clean)
701+ if err != nil {
702+ logger.Error("could not convert access log to a visit", "err", err)
703+ continue
704+ }
705+ jso, err := json.Marshal(visit)
706+ if err != nil {
707+ logger.Error("could not marshal json of a visit", "err", err)
708+ continue
709+ }
710+ _, _ = send.Write(jso)
711+ }
686712 }
687713 }
688714 }
689715
716+func accessLogToVisit(dbpool db.DB, line string) (*db.AnalyticsVisits, error) {
717+ accessLog := CaddyAccessLog{}
718+ err := json.Unmarshal([]byte(line), &accessLog)
719+ if err != nil {
720+ return nil, err
721+ }
722+
723+ return deserializeCaddyAccessLog(dbpool, &accessLog)
724+}
725+
690726 func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secret string) {
691727 drain := metrics.ReconnectReadMetrics(
692728 ctx,
......@@ -696,35 +732,29 @@ func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secr
696732 -1,
697733 )
698734
699- scanner := bufio.NewScanner(drain)
700- for scanner.Scan() {
701- line := scanner.Text()
702- accessLog := CaddyAccessLog{}
703- err := json.Unmarshal([]byte(line), &accessLog)
704- if err != nil {
705- logger.Error("json unmarshal", "err", err)
706- continue
707- }
708-
709- visit, err := deserializeCaddyAccessLog(dbpool, &accessLog)
710- if err != nil {
711- logger.Error("cannot deserialize access log", "err", err)
712- continue
713- }
714- err = shared.AnalyticsVisitFromVisit(visit, dbpool, secret)
715- if err != nil {
716- if !errors.Is(err, shared.ErrAnalyticsDisabled) {
717- logger.Info("could not record analytics visit", "reason", err)
735+ for {
736+ scanner := bufio.NewScanner(drain)
737+ for scanner.Scan() {
738+ line := scanner.Text()
739+ visit, err := accessLogToVisit(dbpool, line)
740+ if err != nil {
741+ logger.Error("could not convert access log to a visit", "err", err)
742+ continue
743+ }
744+ err = shared.AnalyticsVisitFromVisit(visit, dbpool, secret)
745+ if err != nil {
746+ if !errors.Is(err, shared.ErrAnalyticsDisabled) {
747+ logger.Info("could not record analytics visit", "reason", err)
748+ }
718749 }
719- }
720750
721- logger.Info("inserting visit", "visit", visit)
722- err = dbpool.InsertVisit(visit)
723- if err != nil {
724- logger.Error("could not insert visit record", "err", err)
751+ logger.Info("inserting visit", "visit", visit)
752+ err = dbpool.InsertVisit(visit)
753+ if err != nil {
754+ logger.Error("could not insert visit record", "err", err)
755+ }
725756 }
726757 }
727- fmt.Println("DROPINNGGGGGGG")
728758 }
729759
730760 func authMux(apiConfig *shared.ApiConfig) *http.ServeMux {
......@@ -796,7 +826,7 @@ func StartApiServer() {
796826 // gather metrics in the auth service
797827 go metricDrainSub(ctx, db, logger, cfg.Secret)
798828 // convert container logs to access logs
799- // go containerDrainSub(ctx, logger)
829+ go containerDrainSub(ctx, db, logger)
800830
801831 defer ctx.Done()
802832
Back to top