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 <build@mattermost.com>
Этот коммит содержится в:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
Ссылка в новой задаче
Block a user