From 8f93b3e2639a9860bafb93190170e90a4b5aae8d Mon Sep 17 00:00:00 2001 From: Mike Sawka Date: Tue, 23 Apr 2024 17:40:14 -0700 Subject: [PATCH] activity updates for v0.7.3 (#600) * adding more activity updates, tabs, workspaces, startup/shutdown * name change, allow activity updates from FE. aichat + history opens * activity updates for non-standard renderers --- src/models/input.ts | 2 + src/models/model.ts | 9 +++ src/types/custom.d.ts | 5 ++ wavesrv/cmd/main-server.go | 32 ++++++++- wavesrv/pkg/cmdrunner/cmdrunner.go | 22 ++++-- wavesrv/pkg/remote/remote.go | 2 +- wavesrv/pkg/scpacket/scpacket.go | 30 ++++++-- wavesrv/pkg/scws/scws.go | 6 ++ wavesrv/pkg/sstore/dbops.go | 14 +++- wavesrv/pkg/telemetry/telemetry.go | 108 +++++++++++++++++++++++------ 10 files changed, 194 insertions(+), 36 deletions(-) diff --git a/src/models/input.ts b/src/models/input.ts index 0ec199c78..39657922d 100644 --- a/src/models/input.ts +++ b/src/models/input.ts @@ -283,6 +283,7 @@ class InputModel { if (this.getActiveAuxView() != appconst.InputAuxView_History) { this.dropModHistory(true); this.setActiveAuxView(appconst.InputAuxView_History); + this.globalModel.sendActivity("history-open"); } } @@ -686,6 +687,7 @@ class InputModel { openAIAssistantChat(): void { this.setActiveAuxView(appconst.InputAuxView_AIChat); + this.globalModel.sendActivity("aichat-open"); } clearAIAssistantChat(): void { diff --git a/src/models/model.ts b/src/models/model.ts index e7106e209..3ba346db3 100644 --- a/src/models/model.ts +++ b/src/models/model.ts @@ -1816,6 +1816,15 @@ class Model { getElectronApi(): ElectronApi { return getApi(); } + + sendActivity(atype: string) { + const pk: FeActivityPacketType = { + type: "feactivity", + activity: {}, + }; + pk.activity[atype] = 1; + this.ws.pushMessage(pk); + } } export { Model, getApi }; diff --git a/src/types/custom.d.ts b/src/types/custom.d.ts index 1d7410daa..34b04375b 100644 --- a/src/types/custom.d.ts +++ b/src/types/custom.d.ts @@ -217,6 +217,11 @@ declare global { winsize?: TermWinSize; }; + type FeActivityPacketType = { + type: string; + activity: Record; + }; + type RemoteInputPacketType = { type: string; remoteid: string; diff --git a/wavesrv/cmd/main-server.go b/wavesrv/cmd/main-server.go index ab3669d48..ee2059403 100644 --- a/wavesrv/cmd/main-server.go +++ b/wavesrv/cmd/main-server.go @@ -227,7 +227,9 @@ func HandleLogActiveState(w http.ResponseWriter, r *http.Request) { activity.OpenMinutes = 1 } activity.NumConns = remote.NumRemotes() - err = telemetry.UpdateCurrentActivity(r.Context(), activity) + activity.NumWorkspaces, _ = sstore.NumSessions(r.Context()) + activity.NumTabs, _ = sstore.NumScreens(r.Context()) + err = telemetry.UpdateActivity(r.Context(), activity) if err != nil { WriteJsonError(w, fmt.Errorf("error updating activity: %w", err)) return @@ -969,6 +971,7 @@ func installSignalHandlers() { func doShutdown(reason string) { shutdownOnce.Do(func() { log.Printf("[wave] local server %v, start shutdown\n", reason) + shutdownActivityUpdate() sendTelemetryWrapper() log.Printf("[wave] closing db connection\n") sstore.CloseDB() @@ -1013,6 +1016,31 @@ func configDirHandler(w http.ResponseWriter, r *http.Request) { w.Write(dirListJson) } +func startupActivityUpdate() { + activity := telemetry.ActivityUpdate{ + NumConns: remote.NumRemotes(), + Startup: 1, + } + ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second) + defer cancelFn() + activity.NumWorkspaces, _ = sstore.NumSessions(ctx) + activity.NumTabs, _ = sstore.NumScreens(ctx) + err := telemetry.UpdateActivity(ctx, activity) // set at least one record into activity (don't use go routine wrap here) + if err != nil { + log.Printf("error updating startup activity: %v\n", err) + } +} + +func shutdownActivityUpdate() { + activity := telemetry.ActivityUpdate{Shutdown: 1} + ctx, cancelFn := context.WithTimeout(context.Background(), 1*time.Second) + defer cancelFn() + err := telemetry.UpdateActivity(ctx, activity) // do NOT use the go routine wrap here (this needs to be synchronous) + if err != nil { + log.Printf("error updating shutdown activity: %v\n", err) + } +} + func main() { scbase.BuildTime = BuildTime scbase.WaveVersion = WaveVersion @@ -1089,7 +1117,7 @@ func main() { } log.Printf("PCLOUD_ENDPOINT=%s\n", pcloud.GetEndpoint()) - telemetry.UpdateActivityWrap(context.Background(), telemetry.ActivityUpdate{NumConns: remote.NumRemotes()}, "numconns") // set at least one record into activity + startupActivityUpdate() installSignalHandlers() go telemetryLoop() go stdinReadWatch() diff --git a/wavesrv/pkg/cmdrunner/cmdrunner.go b/wavesrv/pkg/cmdrunner/cmdrunner.go index eaebd5562..fd4c0ac11 100644 --- a/wavesrv/pkg/cmdrunner/cmdrunner.go +++ b/wavesrv/pkg/cmdrunner/cmdrunner.go @@ -743,7 +743,7 @@ func EvalCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.U } evalDepth := getEvalDepth(ctx) if pk.Interactive && evalDepth == 0 { - telemetry.UpdateActivityWrap(ctx, telemetry.ActivityUpdate{NumCommands: 1}, "numcommands") + telemetry.GoUpdateActivityWrap(telemetry.ActivityUpdate{NumCommands: 1}, "numcommands") } if evalDepth > MaxEvalDepth { return nil, fmt.Errorf("alias/history expansion max-depth exceeded") @@ -895,6 +895,7 @@ func ScreenOpenCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (s return nil, err } update.Merge(crUpdate) + telemetry.GoUpdateActivityWrap(telemetry.ActivityUpdate{NewTab: 1}, "screen:open") return update, nil } @@ -2955,6 +2956,7 @@ func OpenAICommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus if err != nil { return nil, fmt.Errorf("cannot add new line: %v", err) } + sendRendererActivityUpdate("openai") if resolveBool(pk.Kwargs["stream"], true) { go doOpenAIStreamCompletion(cmd, clientData.ClientId, opts, prompt) @@ -3138,6 +3140,7 @@ func addLineForCmd(ctx context.Context, metaCmd string, shouldFocus bool, ids re if err != nil { return nil, err } + sendRendererActivityUpdate(renderer) screen, err := sstore.GetScreenById(ctx, ids.ScreenId) if err != nil { // ignore error here, because the command has already run (nothing to do) @@ -3468,7 +3471,7 @@ func validateRemoteColor(color string, typeStr string) error { func SessionOpenSharedCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.UpdatePacket, error) { activity := telemetry.ActivityUpdate{ClickShared: 1} - telemetry.UpdateActivityWrap(ctx, activity, "click-shared") + telemetry.GoUpdateActivityWrap(activity, "click-shared") return nil, fmt.Errorf("shared sessions are not available in this version of prompt (stay tuned)") } @@ -4258,7 +4261,7 @@ func HistoryCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbu } show := !resolveBool(pk.Kwargs["noshow"], false) if show { - telemetry.UpdateActivityWrap(ctx, telemetry.ActivityUpdate{HistoryView: 1}, "history") + telemetry.GoUpdateActivityWrap(telemetry.ActivityUpdate{HistoryView: 1}, "history") } update := scbus.MakeUpdatePacket() update.AddUpdate(history.HistoryInfoType{ @@ -4486,6 +4489,15 @@ func focusScreenLine(ctx context.Context, screenId string, lineNum int64) (*ssto return screen, nil } +func sendRendererActivityUpdate(renderer string) { + if renderer == "" || !telemetry.IsAllowedRenderer(renderer) { + return + } + activity := telemetry.ActivityUpdate{Renderers: make(map[string]int)} + activity.Renderers[renderer] = 1 + telemetry.GoUpdateActivityWrap(activity, "renderer") +} + func LineSetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.UpdatePacket, error) { ids, err := resolveUiIds(ctx, pk, R_Session|R_Screen) if err != nil { @@ -4508,6 +4520,7 @@ func LineSetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbu if err != nil { return nil, fmt.Errorf("error changing line renderer: %v", err) } + sendRendererActivityUpdate(renderer) varsUpdated = append(varsUpdated, KwArgRenderer) } if view, found := pk.Kwargs[KwArgView]; found { @@ -4518,6 +4531,7 @@ func LineSetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbu if err != nil { return nil, fmt.Errorf("error changing line view: %v", err) } + sendRendererActivityUpdate(view) varsUpdated = append(varsUpdated, KwArgView) } if stateJson, found := pk.Kwargs[KwArgState]; found { @@ -4608,7 +4622,7 @@ func BookmarksShowCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) if err != nil { return nil, fmt.Errorf("cannot retrieve bookmarks: %v", err) } - telemetry.UpdateActivityWrap(ctx, telemetry.ActivityUpdate{BookmarksView: 1}, "bookmarks") + telemetry.GoUpdateActivityWrap(telemetry.ActivityUpdate{BookmarksView: 1}, "bookmarks") update := scbus.MakeUpdatePacket() update.AddUpdate(&MainViewUpdate{ diff --git a/wavesrv/pkg/remote/remote.go b/wavesrv/pkg/remote/remote.go index 9f96212d1..cd16910d0 100644 --- a/wavesrv/pkg/remote/remote.go +++ b/wavesrv/pkg/remote/remote.go @@ -1431,7 +1431,7 @@ func (msh *MShellProc) ReInit(ctx context.Context, ck base.CommandKey, shellType } defer func() { if rtnErr != nil { - telemetry.UpdateActivityWrap(ctx, makeReinitErrorUpdate(shellType), "reiniterror") + telemetry.GoUpdateActivityWrap(makeReinitErrorUpdate(shellType), "reiniterror") } }() startTs := time.Now() diff --git a/wavesrv/pkg/scpacket/scpacket.go b/wavesrv/pkg/scpacket/scpacket.go index df6a64239..6385e3c75 100644 --- a/wavesrv/pkg/scpacket/scpacket.go +++ b/wavesrv/pkg/scpacket/scpacket.go @@ -80,12 +80,15 @@ func (r RemotePtrType) MakeFullRemoteRef() string { return fmt.Sprintf("@%s:%s:%s", r.OwnerId, r.RemoteId, r.Name) } -const FeCommandPacketStr = "fecmd" -const WatchScreenPacketStr = "watchscreen" -const FeInputPacketStr = "feinput" -const RemoteInputPacketStr = "remoteinput" -const CmdInputTextPacketStr = "cmdinputtext" -const EphemeralCommandResponsePacketStr = "ephemeralcommandresponse" +const ( + FeCommandPacketStr = "fecmd" + WatchScreenPacketStr = "watchscreen" + FeInputPacketStr = "feinput" + RemoteInputPacketStr = "remoteinput" + CmdInputTextPacketStr = "cmdinputtext" + EphemeralCommandResponsePacketStr = "ephemeralcommandresponse" + FeActivityPacketStr = "feactivity" +) type FeCommandPacketType struct { Type string `json:"type"` @@ -175,12 +178,19 @@ type CmdInputTextPacketType struct { Text utilfn.StrWithPos `json:"text"` } +type FeActivityPacketType struct { + Type string `json:"type"` + Activity map[string]int `json:"activity"` +} + func init() { packet.RegisterPacketType(FeCommandPacketStr, reflect.TypeOf(FeCommandPacketType{})) packet.RegisterPacketType(WatchScreenPacketStr, reflect.TypeOf(WatchScreenPacketType{})) packet.RegisterPacketType(FeInputPacketStr, reflect.TypeOf(FeInputPacketType{})) packet.RegisterPacketType(RemoteInputPacketStr, reflect.TypeOf(RemoteInputPacketType{})) packet.RegisterPacketType(CmdInputTextPacketStr, reflect.TypeOf(CmdInputTextPacketType{})) + packet.RegisterPacketType(EphemeralCommandResponsePacketStr, reflect.TypeOf(EphemeralCommandResponsePacketType{})) + packet.RegisterPacketType(FeActivityPacketStr, reflect.TypeOf(FeActivityPacketType{})) } type PacketType interface { @@ -234,3 +244,11 @@ func MakeRemoteInputPacket() *RemoteInputPacketType { func (*RemoteInputPacketType) GetType() string { return RemoteInputPacketStr } + +func MakeFeActivityPacket() *FeActivityPacketType { + return &FeActivityPacketType{Type: FeActivityPacketStr} +} + +func (*FeActivityPacketType) GetType() string { + return FeActivityPacketStr +} diff --git a/wavesrv/pkg/scws/scws.go b/wavesrv/pkg/scws/scws.go index aedd54f81..2e4df20f1 100644 --- a/wavesrv/pkg/scws/scws.go +++ b/wavesrv/pkg/scws/scws.go @@ -18,6 +18,7 @@ import ( "github.com/wavetermdev/waveterm/wavesrv/pkg/scbus" "github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket" "github.com/wavetermdev/waveterm/wavesrv/pkg/sstore" + "github.com/wavetermdev/waveterm/wavesrv/pkg/telemetry" "github.com/wavetermdev/waveterm/wavesrv/pkg/userinput" "github.com/wavetermdev/waveterm/wavesrv/pkg/wsshell" ) @@ -288,6 +289,11 @@ func (ws *WSState) processMessage(msgBytes []byte) error { } return nil } + if pk.GetType() == scpacket.FeActivityPacketStr { + feActivityPk := pk.(*scpacket.FeActivityPacketType) + telemetry.UpdateFeActivityWrap(feActivityPk) + return nil + } return fmt.Errorf("got ws bad message: %v", pk.GetType()) } diff --git a/wavesrv/pkg/sstore/dbops.go b/wavesrv/pkg/sstore/dbops.go index f28a3da61..336f797c9 100644 --- a/wavesrv/pkg/sstore/dbops.go +++ b/wavesrv/pkg/sstore/dbops.go @@ -95,6 +95,17 @@ func NumSessions(ctx context.Context) (int, error) { return numSessions, txErr } +func NumScreens(ctx context.Context) (int, error) { + var numScreens int + txErr := WithTx(ctx, func(tx *TxWrap) error { + query := "SELECT count(*) FROM screen" + numScreens = tx.GetInt(query) + return nil + }) + return numScreens, txErr + +} + func GetAllRemotes(ctx context.Context) ([]*RemoteType, error) { var rtn []*RemoteType err := WithTx(ctx, func(tx *TxWrap) error { @@ -688,9 +699,6 @@ INSERT INTO cmd ( screenid, lineid, remoteownerid, remoteid, remotename, cmdstr ` tx.NamedExec(query, cmdMap) } - if isWebShare(tx, line.ScreenId) { - insertScreenLineUpdate(tx, line.ScreenId, line.LineId, UpdateType_LineNew) - } return nil }) } diff --git a/wavesrv/pkg/telemetry/telemetry.go b/wavesrv/pkg/telemetry/telemetry.go index a8418aa06..3b9ac7aa7 100644 --- a/wavesrv/pkg/telemetry/telemetry.go +++ b/wavesrv/pkg/telemetry/telemetry.go @@ -14,11 +14,24 @@ import ( "github.com/wavetermdev/waveterm/wavesrv/pkg/dbutil" "github.com/wavetermdev/waveterm/wavesrv/pkg/scbase" + "github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket" "github.com/wavetermdev/waveterm/wavesrv/pkg/sstore" ) const MaxTzNameLen = 50 +// "terminal" should not be in this list +var allowedRenderers = map[string]bool{ + "markdown": true, + "code": true, + "openai": true, + "csv": true, + "image": true, + "pdf": true, + "media": true, + "mustache": true, +} + type ActivityUpdate struct { FgMinutes int ActiveMinutes int @@ -28,10 +41,17 @@ type ActivityUpdate struct { HistoryView int BookmarksView int NumConns int - WebShareLimit int + NumWorkspaces int + NumTabs int + NewTab int ReinitBashErrors int ReinitZshErrors int + Startup int + Shutdown int + FeAIChatOpen int + FeHistoryOpen int BuildTime string + Renderers map[string]int } type ActivityType struct { @@ -48,17 +68,24 @@ type ActivityType struct { } type TelemetryData struct { - NumCommands int `json:"numcommands"` - ActiveMinutes int `json:"activeminutes"` - FgMinutes int `json:"fgminutes"` - OpenMinutes int `json:"openminutes"` - ClickShared int `json:"clickshared,omitempty"` - HistoryView int `json:"historyview,omitempty"` - BookmarksView int `json:"bookmarksview,omitempty"` - NumConns int `json:"numconns"` - WebShareLimit int `json:"websharelimit,omitempty"` - ReinitBashErrors int `json:"reinitbasherrors,omitempty"` - ReinitZshErrors int `json:"reinitzsherrors,omitempty"` + NumCommands int `json:"numcommands"` + ActiveMinutes int `json:"activeminutes"` + FgMinutes int `json:"fgminutes"` + OpenMinutes int `json:"openminutes"` + ClickShared int `json:"clickshared,omitempty"` + HistoryView int `json:"historyview,omitempty"` + BookmarksView int `json:"bookmarksview,omitempty"` + NumConns int `json:"numconns"` + NumWorkspaces int `json:"numworkspaces"` + NumTabs int `json:"numtabs"` + NewTab int `json:"newtab"` + NumStartup int `json:"numstartup,omitempty"` + NumShutdown int `json:"numshutdown,omitempty"` + NumAIChatOpen int `json:"numaichatopen,omitempty"` + NumHistoryOpen int `json:"numhistoryopen,omitempty"` + ReinitBashErrors int `json:"reinitbasherrors,omitempty"` + ReinitZshErrors int `json:"reinitzsherrors,omitempty"` + Renderers map[string]int `json:"renderers,omitempty"` } func (tdata TelemetryData) Value() (driver.Value, error) { @@ -69,13 +96,21 @@ func (tdata *TelemetryData) Scan(val interface{}) error { return dbutil.QuickScanJson(tdata, val) } -// Wraps UpdateCurrentActivity, but ignores errors -func UpdateActivityWrap(ctx context.Context, update ActivityUpdate, debugStr string) { - err := UpdateCurrentActivity(ctx, update) - if err != nil { - // ignore error, just log, since this is not critical - log.Printf("error updating current activity (%s): %v\n", debugStr, err) - } +func IsAllowedRenderer(renderer string) bool { + return allowedRenderers[renderer] +} + +// Wraps UpdateCurrentActivity, spawns goroutine, and logs errors +func GoUpdateActivityWrap(update ActivityUpdate, debugStr string) { + go func() { + ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second) + defer cancelFn() + err := UpdateActivity(ctx, update) + if err != nil { + // ignore error, just log, since this is not critical + log.Printf("error updating current activity (%s): %v\n", debugStr, err) + } + }() } func GetCurDayStr() string { @@ -174,10 +209,24 @@ func atoiNoErr(str string) int { return val } +func UpdateFeActivityWrap(feActivity *scpacket.FeActivityPacketType) { + update := ActivityUpdate{} + for key, val := range feActivity.Activity { + if key == "aichat-open" { + update.FeAIChatOpen = val + } else if key == "history-open" { + update.FeHistoryOpen = val + } else { + log.Printf("unknown feactivity key: %s\n", key) + } + } + GoUpdateActivityWrap(update, "feactivity") +} + var customDayStrRe = regexp.MustCompile(`^((?:\d{4}-\d{2}-\d{2})|today|yesterday|bom|bow)?((?:[+-]\d+[dwm])*)$`) var daystrRe = regexp.MustCompile(`^(\d{4})-(\d{2})-(\d{2})$`) -func UpdateCurrentActivity(ctx context.Context, update ActivityUpdate) error { +func UpdateActivity(ctx context.Context, update ActivityUpdate) error { now := time.Now() dayStr := GetCurDayStr() txErr := sstore.WithTx(ctx, func(tx *sstore.TxWrap) error { @@ -202,9 +251,28 @@ func UpdateCurrentActivity(ctx context.Context, update ActivityUpdate) error { tdata.BookmarksView += update.BookmarksView tdata.ReinitBashErrors += update.ReinitBashErrors tdata.ReinitZshErrors += update.ReinitZshErrors + tdata.NewTab += update.NewTab + tdata.NumStartup += update.Startup + tdata.NumShutdown += update.Shutdown + tdata.NumAIChatOpen += update.FeAIChatOpen + tdata.NumHistoryOpen += update.FeHistoryOpen if update.NumConns > 0 { tdata.NumConns = update.NumConns } + if update.NumWorkspaces > 0 { + tdata.NumWorkspaces = update.NumWorkspaces + } + if update.NumTabs > 0 { + tdata.NumTabs = update.NumTabs + } + if len(update.Renderers) > 0 { + if tdata.Renderers == nil { + tdata.Renderers = make(map[string]int) + } + for key, val := range update.Renderers { + tdata.Renderers[key] += val + } + } query = `UPDATE activity SET tdata = ?, clientversion = ?,