From 241e8edc2e81c7662ab7c7901ade482fe0b8c05c Mon Sep 17 00:00:00 2001 From: Doug Lauder Date: Tue, 9 Jan 2024 06:35:05 -0500 Subject: [PATCH] Ping new shared channels remotes immediately (#25850) * option for auto inviting plugin to all shared channels. * auto-invite remotes to shared channels when flag set * fix unit test * immediately ping new remotes; fix unique siteurl bug * make i18n-extract * fix translations * fix merge conflicts * make modules-tidy * revert accidental go.mod change * revert accidental go.sum changes --------- Co-authored-by: Mattermost Build --- server/channels/api4/remote_cluster.go | 13 +++--- server/channels/app/remote_cluster.go | 1 + .../app/slashcommands/command_remote.go | 11 ++--- server/i18n/en.json | 4 ++ .../services/remotecluster/invitation.go | 2 +- .../platform/services/remotecluster/ping.go | 42 ++++++++++--------- .../platform/services/remotecluster/recv.go | 25 +++++++++++ .../services/remotecluster/service.go | 2 + .../sharedchannel/sync_send_remote.go | 2 +- server/public/model/remote_cluster.go | 29 +++++++++++++ 10 files changed, 98 insertions(+), 33 deletions(-) diff --git a/server/channels/api4/remote_cluster.go b/server/channels/api4/remote_cluster.go index 277100d69d..729aeb9e55 100644 --- a/server/channels/api4/remote_cluster.go +++ b/server/channels/api4/remote_cluster.go @@ -122,7 +122,8 @@ func remoteClusterAcceptMessage(c *Context, w http.ResponseWriter, r *http.Reque func remoteClusterConfirmInvite(c *Context, w http.ResponseWriter, r *http.Request) { // make sure remote cluster service is running. - if _, appErr := c.App.GetRemoteClusterService(); appErr != nil { + rcs, appErr := c.App.GetRemoteClusterService() + if appErr != nil { c.Err = appErr return } @@ -155,6 +156,7 @@ func remoteClusterConfirmInvite(c *Context, w http.ResponseWriter, r *http.Reque } audit.AddEventParameterAuditable(auditRec, "remote_cluster", rc) + // check if the invitation has expired if time.Since(model.GetTimeForMillis(rc.CreateAt)) > remotecluster.InviteExpiresAfter { c.Err = model.NewAppError("remoteClusterAcceptMessage", "api.context.invitation_expired.error", nil, "", http.StatusBadRequest) return @@ -166,12 +168,9 @@ func remoteClusterConfirmInvite(c *Context, w http.ResponseWriter, r *http.Reque return } - rc.RemoteTeamId = confirm.RemoteTeamId - rc.SiteURL = confirm.SiteURL - rc.RemoteToken = confirm.Token - - if _, err := c.App.UpdateRemoteCluster(rc); err != nil { - c.Err = err + if _, rcsErr := rcs.ReceiveInviteConfirmation(confirm); rcsErr != nil { + c.Err = model.NewAppError("remoteClusterConfirmInvite", "api.command_remote.confirm_invitation.error", + map[string]any{"Error": rcsErr.Error()}, "", http.StatusInternalServerError) return } diff --git a/server/channels/app/remote_cluster.go b/server/channels/app/remote_cluster.go index ec1fc75956..77ff95cf4d 100644 --- a/server/channels/app/remote_cluster.go +++ b/server/channels/app/remote_cluster.go @@ -39,6 +39,7 @@ func (a *App) RegisterPluginForSharedChannels(opts model.RegisterPluginOpts) (re rc = &model.RemoteCluster{ Name: opts.Displayname, DisplayName: opts.Displayname, + SiteURL: model.SiteURLPlugin + opts.PluginID, // require a unique siteurl Token: model.NewId(), CreatorId: opts.CreatorID, PluginID: opts.PluginID, diff --git a/server/channels/app/slashcommands/command_remote.go b/server/channels/app/slashcommands/command_remote.go index bffa1cddda..cca5adc571 100644 --- a/server/channels/app/slashcommands/command_remote.go +++ b/server/channels/app/slashcommands/command_remote.go @@ -133,6 +133,7 @@ func (rp *RemoteProvider) doCreate(a *app.App, args *model.CommandArgs, margs ma rc := &model.RemoteCluster{ Name: name, DisplayName: displayname, + SiteURL: model.SiteURLPending + model.NewId(), // require a unique siteurl Token: model.NewId(), CreatorId: args.UserId, } @@ -207,7 +208,7 @@ func (rp *RemoteProvider) doAccept(a *app.App, args *model.CommandArgs, margs ma return responsef(args.T("api.command_remote.accept_invitation.error", map[string]any{"Error": err.Error()})) } - return responsef("##### " + args.T("api.command_remote.accept_invitation", map[string]any{"SiteURL": rc.SiteURL})) + return responsef("##### " + args.T("api.command_remote.accept_invitation", map[string]any{"SiteURL": rc.GetSiteURL()})) } // doRemove removes a remote cluster from the database, effectively revoking the trust relationship. @@ -246,11 +247,11 @@ func (rp *RemoteProvider) doStatus(a *app.App, args *model.CommandArgs, _ map[st fmt.Fprintf(&sb, "| :---- | :---- | :---- | :---- | :---- | :---- | :---- | \n") for _, rc := range list { - accepted := formatBool(args.T, rc.SiteURL != "") + accepted := formatBool(args.T, rc.IsConfirmed()) online := formatBool(args.T, isOnline(rc.LastPingAt)) lastPing := formatTimestamp(rc.LastPingAt) - fmt.Fprintf(&sb, "| %s | %s | %s | %s | %s | %s | %s |\n", rc.Name, rc.DisplayName, rc.RemoteId, rc.SiteURL, accepted, online, lastPing) + fmt.Fprintf(&sb, "| %s | %s | %s | %s | %s | %s | %s |\n", rc.Name, rc.DisplayName, rc.RemoteId, rc.GetSiteURL(), accepted, online, lastPing) } return responsef(sb.String()) } @@ -273,7 +274,7 @@ func getRemoteClusterAutocompleteListItems(a *app.App, includeOffline bool) ([]m for _, rc := range clusters { item := model.AutocompleteListItem{ Item: rc.RemoteId, - HelpText: fmt.Sprintf("%s (%s)", rc.DisplayName, rc.SiteURL)} + HelpText: fmt.Sprintf("%s (%s)", rc.DisplayName, rc.GetSiteURL())} list = append(list, item) } return list, nil @@ -294,7 +295,7 @@ func getRemoteClusterAutocompleteListItemsNotInChannel(a *app.App, channelID str for _, rc := range all { item := model.AutocompleteListItem{ Item: rc.RemoteId, - HelpText: fmt.Sprintf("%s (%s)", rc.DisplayName, rc.SiteURL)} + HelpText: fmt.Sprintf("%s (%s)", rc.DisplayName, rc.GetSiteURL())} list = append(list, item) } return list, nil diff --git a/server/i18n/en.json b/server/i18n/en.json index d3e5a6f1c4..4223c82002 100644 --- a/server/i18n/en.json +++ b/server/i18n/en.json @@ -1185,6 +1185,10 @@ "id": "api.command_remote.cluster_removed", "translation": "Secure connection {{.RemoteId}} {{.Result}}." }, + { + "id": "api.command_remote.confirm_invitation.error", + "translation": "Could not confirm invitation: {{.Error}}" + }, { "id": "api.command_remote.decode_invitation.error", "translation": "Could not decode invitation: {{.Error}}" diff --git a/server/platform/services/remotecluster/invitation.go b/server/platform/services/remotecluster/invitation.go index 76f1eeb13e..aec044b522 100644 --- a/server/platform/services/remotecluster/invitation.go +++ b/server/platform/services/remotecluster/invitation.go @@ -56,7 +56,7 @@ func (rcs *Service) AcceptInvitation(invite *model.RemoteClusterInvite, name str } // issue the first ping right away. The goroutine will exit when ping completes or PingTimeout exceeded. - go rcs.pingRemote(rcSaved) + go rcs.PingNow(rcSaved) return rcSaved, nil } diff --git a/server/platform/services/remotecluster/ping.go b/server/platform/services/remotecluster/ping.go index a03645c50c..642f7dd544 100644 --- a/server/platform/services/remotecluster/ping.go +++ b/server/platform/services/remotecluster/ping.go @@ -13,6 +13,27 @@ import ( "github.com/mattermost/mattermost/server/public/shared/mlog" ) +// PingNow emits a ping immediately without waiting for next ping loop. +func (rcs *Service) PingNow(rc *model.RemoteCluster) { + online := rc.IsOnline() + + if err := rcs.pingRemote(rc); err != nil { + rcs.server.Log().Log(mlog.LvlRemoteClusterServiceWarn, "Remote cluster ping failed", + mlog.String("remote", rc.DisplayName), + mlog.String("remoteId", rc.RemoteId), + mlog.String("pluginId", rc.PluginID), + mlog.Err(err), + ) + } + + if online != rc.IsOnline() { + if metrics := rcs.server.GetMetrics(); metrics != nil { + metrics.IncrementRemoteClusterConnStateChangeCounter(rc.RemoteId, rc.IsOnline()) + } + rcs.fireConnectionStateChgEvent(rc) + } +} + // pingLoop periodically sends a ping to all remote clusters. func (rcs *Service) pingLoop(done <-chan struct{}) { pingChan := make(chan *model.RemoteCluster, MaxConcurrentSends*2) @@ -72,24 +93,7 @@ func (rcs *Service) pingEmitter(pingChan <-chan *model.RemoteCluster, done <-cha if rc == nil { return } - - online := rc.IsOnline() - - if err := rcs.pingRemote(rc); err != nil { - rcs.server.Log().Log(mlog.LvlRemoteClusterServiceWarn, "Remote cluster ping failed", - mlog.String("remote", rc.DisplayName), - mlog.String("remoteId", rc.RemoteId), - mlog.String("pluginId", rc.PluginID), - mlog.Err(err), - ) - } - - if online != rc.IsOnline() { - if metrics := rcs.server.GetMetrics(); metrics != nil { - metrics.IncrementRemoteClusterConnStateChangeCounter(rc.RemoteId, rc.IsOnline()) - } - rcs.fireConnectionStateChgEvent(rc) - } + rcs.PingNow(rc) case <-done: return } @@ -103,7 +107,7 @@ var ErrPluginPingFail = errors.New("plugin ping failed") func (rcs *Service) pingRemote(rc *model.RemoteCluster) error { ping := model.RemoteClusterPing{} - if rc.PluginID != "" { + if rc.IsPlugin() { ping.SentAt = model.GetMillis() if ok := rcs.app.OnSharedChannelsPing(rc); !ok { return ErrPluginPingFail diff --git a/server/platform/services/remotecluster/recv.go b/server/platform/services/remotecluster/recv.go index a501a8cb09..77eef686fc 100644 --- a/server/platform/services/remotecluster/recv.go +++ b/server/platform/services/remotecluster/recv.go @@ -51,3 +51,28 @@ func callback(listener TopicListener, msg model.RemoteClusterMsg, rc *model.Remo err = listener(msg, rc, resp) return } + +// ReceiveInviteConfirmation is called by the Rest API layer when a Remote Cluster accepts an invitation from this +// local cluster. +func (rcs *Service) ReceiveInviteConfirmation(confirm model.RemoteClusterInvite) (*model.RemoteCluster, error) { + store := rcs.server.GetStore().RemoteCluster() + + rc, err := store.Get(confirm.RemoteId) + if err != nil { + return nil, fmt.Errorf("cannot accept invite confirmation for remote %s: %w", confirm.RemoteId, err) + } + + rc.RemoteTeamId = confirm.RemoteTeamId + rc.SiteURL = confirm.SiteURL + rc.RemoteToken = confirm.Token + + rcUpdated, err := store.Update(rc) + if err != nil { + return nil, fmt.Errorf("cannot apply invite confirmation for remote %s: %w", confirm.RemoteId, err) + } + + // issue the first ping right away. The goroutine will exit when ping completes or PingTimeout exceeded. + go rcs.PingNow(rcUpdated) + + return rcUpdated, nil +} diff --git a/server/platform/services/remotecluster/service.go b/server/platform/services/remotecluster/service.go index 88773b3179..1df9dd0efd 100644 --- a/server/platform/services/remotecluster/service.go +++ b/server/platform/services/remotecluster/service.go @@ -70,6 +70,8 @@ type RemoteClusterServiceIFace interface { SendProfileImage(ctx context.Context, userID string, rc *model.RemoteCluster, provider ProfileImageProvider, f SendProfileImageResultFunc) error AcceptInvitation(invite *model.RemoteClusterInvite, name string, displayName string, creatorId string, teamId string, siteURL string) (*model.RemoteCluster, error) ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response + ReceiveInviteConfirmation(invite model.RemoteClusterInvite) (*model.RemoteCluster, error) + PingNow(rc *model.RemoteCluster) } // TopicListener is a callback signature used to listen for incoming messages for diff --git a/server/platform/services/sharedchannel/sync_send_remote.go b/server/platform/services/sharedchannel/sync_send_remote.go index 0c031378ab..3a78a09f8d 100644 --- a/server/platform/services/sharedchannel/sync_send_remote.go +++ b/server/platform/services/sharedchannel/sync_send_remote.go @@ -549,7 +549,7 @@ func (scs *Service) sendSyncMsgToRemote(msg *model.SyncMsg, rc *model.RemoteClus return fmt.Errorf("cannot update remote cluster %s for channel id %s; Remote Cluster Service not enabled", rc.Name, msg.ChannelId) } - if rc.PluginID != "" { + if rc.IsPlugin() { return scs.sendSyncMsgToPlugin(msg, rc, f) } diff --git a/server/public/model/remote_cluster.go b/server/public/model/remote_cluster.go index 758a564ed3..632a70cfb3 100644 --- a/server/public/model/remote_cluster.go +++ b/server/public/model/remote_cluster.go @@ -22,6 +22,9 @@ const ( RemoteNameMinLength = 1 RemoteNameMaxLength = 64 + SiteURLPending = "pending_" + SiteURLPlugin = "plugin_" + BitflagOptionAutoShareDMs Bitmask = 1 << iota // Any new DM/GM is automatically shared BitflagOptionAutoInvited // Remote is automatically invited to all shared channels ) @@ -151,6 +154,32 @@ func (rc *RemoteCluster) IsOnline() bool { return rc.LastPingAt > GetMillis()-RemoteOfflineAfterMillis } +func (rc *RemoteCluster) IsConfirmed() bool { + if rc.IsPlugin() { + return true // local plugins are automatically confirmed + } + + if rc.SiteURL != "" && !strings.HasPrefix(rc.SiteURL, SiteURLPending) { + return true // empty or pending siteurl are not confirmed + } + return false +} + +func (rc *RemoteCluster) IsPlugin() bool { + if rc.PluginID != "" || strings.HasPrefix(rc.SiteURL, SiteURLPlugin) { + return true // local plugins are automatically confirmed + } + return false +} + +func (rc *RemoteCluster) GetSiteURL() string { + siteURL := rc.SiteURL + if strings.HasPrefix(siteURL, SiteURLPending) || strings.HasPrefix(siteURL, SiteURLPlugin) { + siteURL = "" + } + return siteURL +} + // fixTopics ensures all topics are separated by one, and only one, space. func (rc *RemoteCluster) fixTopics() { trimmed := strings.TrimSpace(rc.Topics)