mirror of
https://github.com/hanzoai/team.git
synced 2026-09-30 01:57:35 +00:00
Doug LauderandGitHub
f1f9c3be8c
* Use multi-level logging for shared channel and remote cluster service errors Service-specific log levels (LvlRemoteClusterServiceError, LvlSharedChannelServiceError) were hidden from the main log by default, requiring explicit configuration to see them. Switch all call sites to use LogM with multi-level combos so each log line is attributed to both the standard level (error/warn) and the service-specific level. This surfaces errors in the main log while preserving the ability to isolate them into dedicated files. Each instance was reviewed and either kept as error (DB failures, security issues, config errors, exhausted retries on critical data) or downgraded to warn (transient network failures, remote-side reported errors with retry logic, non-critical data like profile images, reactions, acknowledgements, and status).
94 lines
3.1 KiB
Go
94 lines
3.1 KiB
Go
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
|
// See LICENSE.txt for license information.
|
|
|
|
package remotecluster
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/mattermost/mattermost/server/public/model"
|
|
"github.com/mattermost/mattermost/server/public/shared/mlog"
|
|
)
|
|
|
|
// ReceiveIncomingMsg is called by the Rest API layer, or websocket layer (future), when a Remote Cluster
|
|
// message is received. Here we route the message to any topic listeners.
|
|
// `rc` and `msg` cannot be nil.
|
|
func (rcs *Service) ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response {
|
|
rcs.mux.RLock()
|
|
defer rcs.mux.RUnlock()
|
|
|
|
if metrics := rcs.server.GetMetrics(); metrics != nil {
|
|
metrics.IncrementRemoteClusterMsgReceivedCounter(rc.RemoteId)
|
|
}
|
|
|
|
rcSanitized := *rc
|
|
rcSanitized.Token = ""
|
|
rcSanitized.RemoteToken = ""
|
|
|
|
var response Response
|
|
response.Status = ResponseStatusOK
|
|
|
|
listeners := rcs.getTopicListeners(msg.Topic)
|
|
|
|
for _, l := range listeners {
|
|
if err := callback(l, msg, &rcSanitized, &response); err != nil {
|
|
rcs.server.Log().LogM(mlog.MlvlRemoteClusterServiceError, "Error from remote cluster message listener",
|
|
mlog.String("msgId", msg.Id), mlog.String("topic", msg.Topic), mlog.String("remote", rc.DisplayName), mlog.Err(err))
|
|
|
|
response.Status = ResponseStatusFail
|
|
response.Err = err.Error()
|
|
}
|
|
}
|
|
return response
|
|
}
|
|
|
|
func callback(listener TopicListener, msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) (err error) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
err = fmt.Errorf("%v", r)
|
|
}
|
|
}()
|
|
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, false)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("cannot accept invite confirmation for remote %s: %w", confirm.RemoteId, err)
|
|
}
|
|
|
|
if rc.IsConfirmed() {
|
|
return nil, fmt.Errorf("cannot accept invite confirmation for remote %s: %w", confirm.RemoteId, RemoteClusterAlreadyConfirmedError)
|
|
}
|
|
|
|
rc.SiteURL = confirm.SiteURL
|
|
rc.RemoteToken = confirm.Token
|
|
|
|
// If the accepting cluster sent a RefreshedToken (its RemoteToken), set it as our Token
|
|
if confirm.Version >= 2 && confirm.RefreshedToken != "" {
|
|
if confirm.RefreshedToken == rc.Token {
|
|
return nil, fmt.Errorf("cannot accept invite confirmation for remote %s: RefreshedToken must be different from the original invite token", confirm.RemoteId)
|
|
}
|
|
rc.Token = confirm.RefreshedToken
|
|
} else {
|
|
// For older versions or if no RefreshedToken was provided, generate a new token
|
|
// to invalidate the original invite token and prevent reuse
|
|
rc.Token = model.NewId()
|
|
}
|
|
|
|
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
|
|
}
|